From 224f8de2cf60e094647ff35af43c8184b45c3640 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Sun, 12 Jul 2026 23:33:52 +0800 Subject: [PATCH] fix(goal): prevent continuation_count regression from racing continuations (#4088) * fix(goal): prevent continuation_count regression from racing continuations * fix(goal): prevent continuation_count regression from racing continuations --- .../harness/deerflow/runtime/runs/worker.py | 6 ++ backend/tests/test_goal_worker.py | 84 +++++++++++++++++++ 2 files changed, 90 insertions(+) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 437616e16..6b0c50711 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -735,6 +735,12 @@ async def _persist_goal_evaluation( current_goal = _read_checkpoint_goal(checkpoint_tuple) if current_goal is None or not _goal_instance_matches(goal, current_goal): return None + # Defensive: compute continuation_count from the fresh current_goal + # inside the lock. The caller computed it from a possibly-stale goal + # snapshot; a racing continuation may have already bumped the count. + if continuation_count is not None: + current_count = int(current_goal.get("continuation_count", 0)) + continuation_count = max(continuation_count, current_count + 1) expected_checkpoint_id = _checkpoint_id(checkpoint_tuple) updated_goal = attach_goal_evaluation( current_goal, diff --git a/backend/tests/test_goal_worker.py b/backend/tests/test_goal_worker.py index cf35d036f..30db79285 100644 --- a/backend/tests/test_goal_worker.py +++ b/backend/tests/test_goal_worker.py @@ -596,6 +596,90 @@ async def test_run_agent_reuses_goal_evaluator_model_for_goal_loop(monkeypatch): assert record.status == RunStatus.success +@pytest.mark.asyncio +async def test_persist_goal_evaluation_does_not_regress_continuation_count_on_race(): + """A racing continuation must not overwrite a higher count with a lower one. + + Scenario: two goal continuations run concurrently. Continuation A reads + continuation_count=1, computes next=2. Continuation B reads the same + count=1, computes next=2, but acquires the lock first and writes count=2. + When A acquires the lock, the current_goal already has count=2. Without + the defensive guard, A would write count=2 again (stale computation), + effectively losing one continuation event. The guard must compute + ``max(stale_next, current_count + 1)`` so A writes count=3. + """ + checkpointer = InMemorySaver() + thread_id = "race-count-thread" + await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="Race test") + # Simulate a racing continuation: bump the persisted continuation_count to 2 + # before calling _persist_goal_evaluation with a next_count computed from + # stale state (count=1 → next=2). + existing_goal = await read_thread_goal(checkpointer, thread_id) + assert existing_goal is not None + bumped_goal = attach_goal_evaluation( + existing_goal, + GoalEvaluation(satisfied=False, blocker="goal_not_met_yet", reason="racing", evidence_summary=""), + run_id="racing-run", + continuation_count=2, # racing continuation already bumped to 2 + ) + await write_thread_goal(checkpointer, thread_id, bumped_goal) + + # Now call _persist_goal_evaluation with continuation_count=2 computed from + # stale state (old count was 1). The guard should detect current_count=2 + # and write max(2, 2+1) = 3. + bridge = _CollectingBridge() + result = await worker._persist_goal_evaluation( + bridge=bridge, + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-late", + goal=existing_goal, # stale goal with continuation_count=1 + evaluation=GoalEvaluation( + satisfied=False, + blocker="goal_not_met_yet", + reason="More work remains.", + evidence_summary="Work remains.", + ), + no_progress_count=1, + continuation_count=2, # computed from stale state: stale_count(1) + 1 + ) + + assert result is not None + # Without the guard this would be 2 (stale computation wins). With the + # guard it must be 3 (current_count + 1 taken inside the lock). + assert result["continuation_count"] == 3 + + +@pytest.mark.asyncio +async def test_persist_goal_evaluation_no_race_uses_caller_count(): + """When no racing continuation exists, the caller's continuation_count is used.""" + checkpointer = InMemorySaver() + thread_id = "no-race-thread" + await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="No race test") + existing_goal = await read_thread_goal(checkpointer, thread_id) + assert existing_goal is not None + + bridge = _CollectingBridge() + result = await worker._persist_goal_evaluation( + bridge=bridge, + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-normal", + goal=existing_goal, + evaluation=GoalEvaluation( + satisfied=False, + blocker="goal_not_met_yet", + reason="More work.", + evidence_summary="Work.", + ), + no_progress_count=1, + continuation_count=1, # 0 + 1 = 1 + ) + + assert result is not None + assert result["continuation_count"] == 1 + + @pytest.mark.asyncio async def test_run_agent_strips_branch_checkpoint_for_goal_continuation(monkeypatch): class FakeAgent: