From 29eaa86cda52f519a362b291e4435e571721a24c Mon Sep 17 00:00:00 2001 From: dajiaohuang Date: Wed, 7 Oct 2026 14:48:02 +0800 Subject: [PATCH 1/2] Fix duplicate session history on failed-run recovery (fixes #1310) _salvage_stream_to_session previously did pre_run_items + replay, which duplicated items when stream.to_input_list() already included the pre-cycle input (the seeded initial_input is in both). Detect the overlap and avoid duplication by checking if replay suffix matches pre_run_items. --- strix/core/execution.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/strix/core/execution.py b/strix/core/execution.py index 91d94498..aad336ea 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -156,7 +156,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: @@ -164,8 +171,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) From 78b537cdf73a8ac70e06203b4ee4f480c179fd96 Mon Sep 17 00:00:00 2001 From: dajiaohuang Date: Thu, 15 Oct 2026 11:34:41 +0800 Subject: [PATCH 2/2] fix: recognize salvaged histories by prefix --- strix/core/execution.py | 2 +- tests/test_execution.py | 13 ++++++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/strix/core/execution.py b/strix/core/execution.py index aad336ea..aa7fde0c 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -172,7 +172,7 @@ async def _salvage_stream_to_session( logger.exception("could not build salvage history for %s", agent_id) return pre_len = len(pre_run_items) - if pre_len > 0 and len(replay) >= pre_len and replay[-pre_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 diff --git a/tests/test_execution.py b/tests/test_execution.py index 83d3829b..2225d697 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -806,11 +806,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()