From fa419b3e4c3849fcd25b15dd65aef7c6066f682b Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Wed, 15 Jul 2026 05:21:04 -0700 Subject: [PATCH] fix(goal): stop continuation_count double-bump in thread_changed_before_continuation stand-down (#4199) _prepare_goal_continuation_input calls the same _persist closure twice with the identical next_count in one evaluation cycle: once to commit the real continuation, and again to record a thread_changed_before_continuation stand-down if a race is detected right after that commit. The second call re-passed continuation_count=next_count, so #4088's defensive max(continuation_count, current_count + 1) guard saw the first call's own write as a "current_count" bump and added another +1 on top of it - silently consuming 2 units of the continuation budget for a cycle that delivered zero actual continuations. #4088's guard is correct for the independent-concurrent-continuations race it targets; this is a separate call site incorrectly re-triggering that same guard against itself. The second _persist call no longer passes continuation_count, matching every other stand-down call site in this function - the count was already correctly committed by the first call. Adds a regression test mirroring the existing thread_changed_after_evaluation race test's checkpointer-wrapper technique, since this sibling branch had zero prior coverage. --- .../harness/deerflow/runtime/runs/worker.py | 9 +- backend/tests/test_goal_worker.py | 95 +++++++++++++++++++ 2 files changed, 103 insertions(+), 1 deletion(-) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 7509976cf..71f3ebfb6 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -993,11 +993,18 @@ async def _prepare_goal_continuation_input( if not _goal_instance_matches(updated_goal, latest_goal) or latest_checkpoint_tuple is None: return None if visible_conversation_signature(_read_checkpoint_messages(latest_checkpoint_tuple)) != conversation_signature_before: + # Do not pass continuation_count here: the persist above already + # committed it (as next_count). Re-passing next_count would make + # _persist_goal_evaluation's race guard (#4088) see that same write as + # a "current_count" bump and add another +1 on top of it, silently + # double-counting this single continuation attempt against the + # continuation budget even though it is being stood down, not + # delivered. Omitting it leaves the already-committed count untouched, + # matching every other stand-down call site in this function. await _persist( latest_goal, evaluation, no_progress_count, - continuation_count=next_count, stand_down_reason="thread_changed_before_continuation", ) return None diff --git a/backend/tests/test_goal_worker.py b/backend/tests/test_goal_worker.py index 30db79285..6ade2962c 100644 --- a/backend/tests/test_goal_worker.py +++ b/backend/tests/test_goal_worker.py @@ -48,6 +48,47 @@ class _ClearBeforeSecondGoalReadCheckpointer: return await self.inner.aput(*args, **kwargs) +class _RaceAfterFirstContinuationCommitCheckpointer: + """Wrap a saver and inject a racing user message right after the first + goal-continuation commit lands. + + ``_prepare_goal_continuation_input``'s real continuation commit (the + ``_persist(..., continuation_count=next_count)`` call that records the + evaluator's decision to continue) performs this scenario's first + ``aput``. Injecting a racing visible message immediately after that write + lands lets the worker's trailing visible-conversation-signature re-check + observe a thread change that happened *after* the continuation was + committed but *before* that re-check runs -- modelling the + ``thread_changed_before_continuation`` race. + """ + + def __init__(self, inner: InMemorySaver, thread_id: str) -> None: + self.inner = inner + self.thread_id = thread_id + self.put_count = 0 + + def get_next_version(self, current, channel): + return self.inner.get_next_version(current, channel) + + async def aget_tuple(self, config): + return await self.inner.aget_tuple(config) + + async def aput(self, *args, **kwargs): + result = await self.inner.aput(*args, **kwargs) + self.put_count += 1 + if self.put_count == 1: + checkpoint_tuple = await self.inner.aget_tuple({"configurable": {"thread_id": self.thread_id, "checkpoint_ns": ""}}) + checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} + channel_values = checkpoint.get("channel_values", {}) or {} + current_messages = channel_values.get("messages", []) or [] + await _write_messages( + self.inner, + thread_id=self.thread_id, + messages=[*current_messages, HumanMessage(content="Actually, stop and wait.")], + ) + return result + + async def _seed_goal_thread( checkpointer: InMemorySaver, *, @@ -384,6 +425,60 @@ async def test_goal_worker_stands_down_when_thread_changes_after_evaluation(monk assert latest_goal["last_evaluation"]["stand_down_reason"] == "thread_changed_after_evaluation" +@pytest.mark.asyncio +async def test_goal_worker_stands_down_when_thread_changes_before_continuation(monkeypatch): + """A user message racing in right after the continuation commits must not + double-bump continuation_count. + + Sibling scenario to ``..._after_evaluation`` above, but the race lands + later: after the evaluator runs and after _prepare_goal_continuation_input + commits the real continuation (``_persist(..., continuation_count=next_count)``), + a racing visible message arrives before the function's trailing re-check. + That re-check detects the changed thread and stands down via a second + ``_persist(..., continuation_count=next_count, stand_down_reason=...)`` + call using the *same* next_count as the first, already-successful call. + + Without the fix, that second call re-triggers PR #4088's + max(continuation_count, current_count + 1) guard against its own sibling + call's prior write (current_count is already next_count from the first + call), bumping continuation_count to next_count + 1 a second time -- + consuming 2 units of the continuation budget for a cycle that delivered + zero actual continuations. The fix must leave it at next_count (1). + """ + inner = InMemorySaver() + thread_id = "race-before-continuation-thread" + await _seed_goal_thread(inner, thread_id=thread_id, goal_text="Finish all tests") + checkpointer = _RaceAfterFirstContinuationCommitCheckpointer(inner, thread_id) + bridge = _CollectingBridge() + + async def fake_evaluate_goal_completion(_goal, _messages, **_kwargs): + return GoalEvaluation( + satisfied=False, + blocker="goal_not_met_yet", + reason="More work remains.", + evidence_summary="Work remains.", + ) + + monkeypatch.setattr(worker, "evaluate_goal_completion", fake_evaluate_goal_completion) + + continuation = await worker._prepare_goal_continuation_input( + bridge=bridge, + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-race-before-continuation", + model_name="test-model", + app_config=None, + ) + + assert continuation is None + latest_goal = await read_thread_goal(inner, thread_id) + assert latest_goal is not None + # Without the fix this is 2 (double-bumped). It must be 1: one real + # continuation attempt was committed and then stood down, not two. + assert latest_goal["continuation_count"] == 1 + assert latest_goal["last_evaluation"]["stand_down_reason"] == "thread_changed_before_continuation" + + @pytest.mark.asyncio async def test_goal_worker_stands_down_without_durable_assistant_receipt(): checkpointer = InMemorySaver()