diff --git a/strix/core/execution.py b/strix/core/execution.py index 87a2bf02..fc37c18b 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -199,7 +199,14 @@ async def _salvage_stream_to_session( stream: Any, agent_id: str, ) -> None: - """Persist a crashed run's full history so a revived agent loses no context.""" + """Persist a crashed run's history so a revived agent loses no context. + + ``stream.to_input_list()`` reconstructs the conversation from this run cycle + as ``original_input + new_items``. After ``seed_initial_input``, the session + (pre_run_items) already contains the initial input on the first cycle, so + blindly concatenating duplicates it. We detect the overlap and append only + the delta. + """ if stream is None: return try: @@ -207,8 +214,12 @@ async def _salvage_stream_to_session( except Exception: logger.exception("could not build salvage history for %s", agent_id) return - desired = list(pre_run_items) + replay - if len(desired) <= len(pre_run_items): + pre_len = len(pre_run_items) + if pre_len > 0 and len(replay) >= pre_len and replay[:pre_len] == pre_run_items: + desired = replay + else: + desired = list(pre_run_items) + replay + if len(desired) <= pre_len: return try: await replace_session_items(session, desired) diff --git a/tests/test_execution.py b/tests/test_execution.py index 0daf0da2..a8d0f4ab 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -829,11 +829,22 @@ async def test_salvage_stream_to_session_preserves_full_history(tmp_path: Any) - "recon turn 2", ] + # A replayed full history starts with the already-persisted session items. + replayed_history = [ + *stored, + {"role": "assistant", "content": "recovered turn"}, + ] + await execution._salvage_stream_to_session( + session, stored, _SalvageStream(replayed_history), "child" + ) + stored = [cast("dict[str, Any]", i) for i in await session.get_items()] + assert stored == replayed_history + # A crash with nothing new to salvage leaves the session untouched. await execution._salvage_stream_to_session( session, list(await session.get_items()), _SalvageStream([]), "child" ) - assert len(await session.get_items()) == 3 + assert len(await session.get_items()) == len(replayed_history) session.close()