From e6a3f83f2f7b98892d7256bc63fe0c3f48677843 Mon Sep 17 00:00:00 2001 From: Totoro Date: Mon, 28 Sep 2026 14:44:45 +0800 Subject: [PATCH] fix(goal): preserve active goals when artifact delivery fails (#5980) * fix(goal): preserve active goals when artifact delivery fails * chore(goal): explain completion guards and log skipped clears * docs(goal): clarify required artifact delivery --------- Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com> --- README.md | 2 +- backend/docs/runtime-guidance-details.md | 11 + .../harness/deerflow/runtime/AGENTS.md | 2 +- .../harness/deerflow/runtime/runs/worker.py | 125 +++-- .../tests/test_goal_delivery_consistency.py | 443 ++++++++++++++++++ backend/tests/test_goal_worker.py | 16 +- 6 files changed, 559 insertions(+), 40 deletions(-) create mode 100644 backend/tests/test_goal_delivery_consistency.py diff --git a/README.md b/README.md index 7dd454e685..c30d6a78bb 100644 --- a/README.md +++ b/README.md @@ -1574,7 +1574,7 @@ Supported commands: /goal clear # clear it ``` -After each Gateway-backed run, DeerFlow evaluates the visible conversation against the active goal with a non-thinking evaluator model. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the goal is satisfied, DeerFlow clears it automatically and publishes the updated thread state. +After each Gateway-backed run, DeerFlow evaluates the visible conversation against the active goal with a non-thinking evaluator model. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the evaluator considers the goal satisfied, DeerFlow clears it after the run successfully finalizes and publishes the updated thread state. If required artifact delivery or its receipt persistence fails, the run reports an error and the goal remains active for a later retry; this does not start another hidden continuation. The Web UI shows the active goal above the composer. The same command is available from the TUI and supported IM channels. In the Web UI and supported IM channels, setting `/goal ` also starts a run with the condition as the task; status and clear commands only manage goal state. Setting or clearing a goal is rejected while that thread has a run in flight, including a run owned by another Gateway worker, so the goal checkpoint cannot branch away from an active run's checkpoint lineage. diff --git a/backend/docs/runtime-guidance-details.md b/backend/docs/runtime-guidance-details.md index a75ec8ee0a..be563e6b6a 100644 --- a/backend/docs/runtime-guidance-details.md +++ b/backend/docs/runtime-guidance-details.md @@ -83,6 +83,17 @@ one accumulated receipt across multiple goal-continuation `_stream_once` calls; journal tests drive LangChain's real async callback dispatcher against a single journal to pin serialized, deduplicated parallel tool callbacks. +A satisfied goal evaluation remains a run-local completion candidate until +delivery, receipt persistence and terminal cancellation arbitration finish. +The worker confirms successful status persistence before clearing the goal; +delivery failure leaves goal state and continuation counters unchanged. Final +clearing holds the goal lock and the existing durable `checkpoint_write` +reservation, rechecks the complete goal snapshot and visible conversation, +then guards the write with the latest checkpoint ID. If another run has already +been admitted, the old completion leaves the goal alone. This allows duration +bookkeeping while preserving newer goal or conversation changes. Delivery +failure does not itself queue a hidden continuation. + **Extension changed-run discovery** (`runtime/runs/store/` and `extensions/run_evidence.py`) orders public run-record changes by the backend-owned `(change_seq, run_id)` key rather than timestamps or per-thread diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index edc8fa794c..3a13d27f10 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -53,7 +53,7 @@ from `on_llm_end` before inspecting the response or touching any run state. **Skill history:** `record_skill_usage` saves lead-run snapshots on terminal answers for paginated history. See `docs/skill-usage-ui.md`. -**Run delivery receipts:** Journal artifact evidence and terminal status must finalize in order. Details: `backend/docs/runtime-guidance-details.md`. +**Run delivery receipts:** Journal artifact evidence and terminal status must finalize before a satisfied goal is cleared. Goal cleanup uses a durable checkpoint-write reservation; delivery failure retains the goal without another continuation. Details: `backend/docs/runtime-guidance-details.md`. **Deferred-tool promotion event deduplication** (`runtime/journal.py`): one `RunJournal` owns the lead graph's run-scoped atomic promotion claim. Parallel diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 500a7a4fe0..d608762429 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -98,9 +98,9 @@ from deerflow.utils.messages import message_to_text from deerflow.workspace_changes import capture_workspace_snapshot, get_changed_output_paths, record_workspace_changes from deerflow.workspace_changes.types import WorkspaceSnapshot -from .manager import RunManager, RunRecord, RunStartOutcome +from .manager import ConflictError, RunManager, RunRecord, RunStartOutcome from .naming import resolve_root_run_name -from .schemas import RunStatus +from .schemas import RunStatus, ThreadOperationKind logger = logging.getLogger(__name__) _THREAD_INCARNATION_UNSET = object() @@ -933,6 +933,7 @@ async def run_agent( runnable_configs: list[dict[str, Any]] = [config] goal_evaluator_model: Any | None = None delivery_content: dict[str, Any] | None = None + goal_completion: _GoalCompletionCandidate | None = None produced_output_paths: list[str] | None = None # Journal construction moved ahead of preflight so every terminal run can # emit a receipt. Completion persistence keeps its prior boundary: before @@ -1460,6 +1461,9 @@ async def run_agent( extensions=extensions, run_stop_reason=runtime.context.get("stop_reason") if isinstance(runtime.context, dict) else None, ) + if isinstance(continuation_input, _GoalCompletionCandidate): + goal_completion = continuation_input + break if continuation_input is None or record.abort_event.is_set(): break await _stream_once(continuation_input, _continuation_runnable_config()) @@ -1688,6 +1692,24 @@ async def run_agent( except Exception: logger.warning("Failed to persist run completion for %s (non-fatal)", run_id, exc_info=True) + # A satisfied evaluator is only a candidate until artifact delivery, + # receipt persistence and durable cancellation arbitration have ended. + # Status writes are best-effort in single-worker mode: confirm the + # existing outcome before deleting recoverable goal state. + if goal_completion is not None and record.status == RunStatus.success and not record.abort_event.is_set() and not record.ownership_lost: + try: + if await run_manager.persist_current_status(run_id): + await _clear_completed_goal( + candidate=goal_completion, + record=record, + run_manager=run_manager, + bridge=bridge, + accessor=accessor, + checkpointer=checkpointer, + ) + except Exception: + logger.warning("Could not finalize satisfied goal for thread %s", thread_id, exc_info=True) + if started and not record.ownership_lost and checkpointer is not None and record.status == RunStatus.interrupted and not _is_edit_replay_run(record): try: await run_manager.wait_for_prior_finalizing(thread_id, run_id) @@ -1819,6 +1841,7 @@ async def run_agent( rollback_point = None subagent_events = None goal_evaluator_model = None + goal_completion = None task_store = None task_info = None pre_run_workspace_snapshot = None @@ -1845,6 +1868,73 @@ async def run_agent( # --------------------------------------------------------------------------- +@dataclass(frozen=True) +class _GoalCompletionCandidate: + goal: GoalState + conversation_signature: str + + +async def _clear_completed_goal( + *, + candidate: _GoalCompletionCandidate, + record: RunRecord, + run_manager: RunManager, + bridge: StreamBridge, + accessor: CheckpointStateAccessor, + checkpointer: Any, +) -> None: + """Clear only the evaluated goal after the run has successfully finalized.""" + try: + # Terminal status has released the run's admission. Use the same durable + # reservation as other out-of-run checkpoint writers; a peer that already + # admitted a new run wins, even before it has written any checkpoint. + async with ( + goal_thread_lock(record.thread_id), + run_manager.reserve_thread_operation(record.thread_id, kind=ThreadOperationKind.checkpoint_write, user_id=record.user_id), + ): + checkpoint_tuple = await _call_checkpointer_method( + checkpointer, + "aget_tuple", + "get_tuple", + {"configurable": {"thread_id": record.thread_id, "checkpoint_ns": ""}}, + ) + if checkpoint_tuple is None: + logger.debug("Skipping goal completion for run %s: checkpoint is unavailable", record.run_id) + return + # Full equality is intentional: even same-instance goal updates must win over clearing. + if _read_checkpoint_goal(checkpoint_tuple) != candidate.goal: + logger.debug("Skipping goal completion for run %s: goal snapshot changed", record.run_id) + return + messages = await _materialized_checkpoint_messages(accessor, record.thread_id) + if visible_conversation_signature(messages) != candidate.conversation_signature: + logger.debug("Skipping goal completion for run %s: visible conversation changed", record.run_id) + return + if record.status != RunStatus.success or record.abort_event.is_set() or record.ownership_lost: + logger.debug( + "Skipping goal completion for run %s: status=%s, aborted=%s, ownership_lost=%s", + record.run_id, + record.status.value, + record.abort_event.is_set(), + record.ownership_lost, + ) + return + # Duration bookkeeping may advance the checkpoint after evaluation. + # Compare goal/conversation above, then guard against stale writes. + values = await write_thread_goal( + checkpointer, + record.thread_id, + None, + as_node="goal_evaluator", + expected_checkpoint_id=_checkpoint_id(checkpoint_tuple), + ) + await bridge.publish(record.run_id, "values", serialize(values, mode="values")) + except GoalWriteConflict: + logger.debug("Skipping goal completion for run %s: checkpoint changed before the goal write", record.run_id) + return + except ConflictError: + return + + def _checkpoint_id(checkpoint_tuple: Any) -> str | None: config = getattr(checkpoint_tuple, "config", {}) or {} configurable = config.get("configurable", {}) if isinstance(config, dict) else {} @@ -2023,8 +2113,8 @@ async def _prepare_goal_continuation_input( task_store: Any | None = None, extensions: Any | None = None, run_stop_reason: str | None = None, -) -> dict[str, Any] | None: - """Evaluate the active goal and return a hidden continuation input if needed. +) -> dict[str, Any] | _GoalCompletionCandidate | None: + """Return a continuation input or a completion candidate for finalization. NOTE: The re-reads below catch a racing user message or ``/goal clear`` before we queue a continuation. Goal writes then serialize per thread and @@ -2145,32 +2235,7 @@ async def _prepare_goal_continuation_input( return None if evaluation["satisfied"]: - try: - async with goal_thread_lock(thread_id): - latest_checkpoint_tuple = await _call_checkpointer_method( - checkpointer, - "aget_tuple", - "get_tuple", - {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}, - ) - if latest_checkpoint_tuple is None: - return None - latest_goal = _read_checkpoint_goal(latest_checkpoint_tuple) - if latest_goal is None or not _goal_instance_matches(goal, latest_goal): - return None - values = await write_thread_goal( - checkpointer, - thread_id, - None, - as_node="goal_evaluator", - expected_checkpoint_id=_checkpoint_id(latest_checkpoint_tuple), - ) - await bridge.publish(run_id, "values", serialize(values, mode="values")) - except GoalWriteConflict: - return None - except Exception: - logger.warning("Could not clear satisfied goal for thread %s", thread_id, exc_info=True) - return None + return _GoalCompletionCandidate(copy.deepcopy(current_goal), conversation_signature_before) stand_down_reason = _stand_down_reason(goal, evaluation, no_progress_count) if stand_down_reason is None and run_stop_reason == "token_capped": diff --git a/backend/tests/test_goal_delivery_consistency.py b/backend/tests/test_goal_delivery_consistency.py new file mode 100644 index 0000000000..300cf7f381 --- /dev/null +++ b/backend/tests/test_goal_delivery_consistency.py @@ -0,0 +1,443 @@ +"""Exercise goal completion and artifact delivery through the real worker.""" + +import importlib +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from langchain_core.messages import AIMessage, HumanMessage +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.graph import END, START, StateGraph + +from deerflow.agents.thread_state import get_thread_state_schema +from deerflow.config.paths import Paths +from deerflow.config.run_ownership_config import RunOwnershipConfig +from deerflow.runtime.checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.goal import build_goal_state, read_thread_goal, write_thread_goal +from deerflow.runtime.runs import worker +from deerflow.runtime.runs.manager import ConflictError, RunManager +from deerflow.runtime.runs.schemas import RunStatus +from deerflow.runtime.runs.store.memory import MemoryRunStore +from deerflow.runtime.user_context import get_effective_user_id +from deerflow.tools.builtins.present_file_tool import present_file_tool + +OUTPUT_PATH = "/mnt/user-data/outputs/report.md" + + +async def _run_goal_delivery( + tmp_path, + monkeypatch, + *, + produce=True, + present_on_turn=None, + verdicts=(True,), + fail_receipt=False, + during_evaluation=None, + during_receipt=None, + stop_reason=None, + heartbeat_enabled=False, + fail_terminal_status=None, + completion_recovers=False, + checkpoint_mode="full", + after_terminal_completion=None, +): + """Keep graphs, checkpoints, output scanning and delivery verification real. + + The graph writes a real file and calls the real present_files body. Its + deterministic node emits the tool callbacks normally supplied by the agent; + only the evaluator model response is fixed, without provider/network access. + """ + checkpointer = InMemorySaver() + failed_terminal_writes = [] + + class RunStore(MemoryRunStore): + async def update_status(self, run_id, status, **kwargs): + if status == "success" and fail_terminal_status is not None: + failed_terminal_writes.append(status) + if fail_terminal_status == "raise": + raise RuntimeError("terminal run store unavailable") + return False + return await super().update_status(run_id, status, **kwargs) + + async def update_run_completion(self, run_id, *, status, **kwargs): + if status == "success" and fail_terminal_status is not None and not completion_recovers: + if fail_terminal_status == "raise": + raise RuntimeError("terminal run store unavailable") + return False + updated = await super().update_run_completion(run_id, status=status, **kwargs) + if after_terminal_completion is not None and status == "success": + await after_terminal_completion(self, record) + return updated + + run_store = RunStore() + manager = RunManager(store=run_store, run_ownership_config=RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled)) + record = await manager.create("goal-delivery-thread") + thread_id = record.thread_id + paths = Paths(base_dir=tmp_path) + output_dir = paths.sandbox_outputs_dir(thread_id, user_id=get_effective_user_id()) + output_dir.mkdir(parents=True) + monkeypatch.setattr("deerflow.workspace_changes.recorder.get_paths", lambda: paths) + monkeypatch.setattr(importlib.import_module("deerflow.tools.builtins.present_file_tool"), "get_paths", lambda: paths) + monkeypatch.setenv("LANGCHAIN_TRACING_V2", "false") + monkeypatch.setenv("LANGSMITH_TRACING", "false") + + turns = [] + model_calls = [] + goals_at_receipt = [] + + class EventStore(MemoryRunEventStore): + async def put_if_absent(self, **event): + if event["event_type"] == "run.delivery": + goals_at_receipt.append(await read_thread_goal(checkpointer, thread_id)) + if during_receipt is not None: + await during_receipt(checkpointer, manager, record) + if fail_receipt: + raise RuntimeError("delivery receipt store unavailable") + return await super().put_if_absent(**event) + + event_store = EventStore() + bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()) + + class EvaluatorModel: + async def ainvoke(self, messages, config=None): + satisfied = verdicts[min(len(model_calls), len(verdicts) - 1)] + model_calls.append(satisfied) + if during_evaluation is not None: + await during_evaluation(checkpointer, manager, record) + return AIMessage( + content=json.dumps( + { + "satisfied": satisfied, + "blocker": "none" if satisfied else "goal_not_met_yet", + "reason": "Fixed evaluator response for the worker regression.", + "evidence_summary": "The assistant reports that the report is complete.", + } + ) + ) + + monkeypatch.setattr(worker, "create_goal_evaluator_model", lambda **kwargs: EvaluatorModel()) + + async def agent_node(state, config): + turn = len(turns) + 1 + turns.append(turn) + if produce: + (output_dir / "report.md").write_text("# Finished report\n", encoding="utf-8") + context = config["configurable"]["__pregel_runtime"].context + updates = {} + messages = [] + if present_on_turn == turn: + tool_id = f"present-{turn}" + ai = AIMessage(content="", tool_calls=[{"id": tool_id, "name": "present_files", "args": {"filepaths": [OUTPUT_PATH]}}]) + runtime = SimpleNamespace(state={**state, "thread_data": {"outputs_path": str(output_dir)}}, context=context, config=config) + command = present_file_tool.func(runtime=runtime, filepaths=[OUTPUT_PATH], tool_call_id=tool_id) + journal = context["__run_journal"] + journal._remember_current_run_tool_calls(ai, caller="lead_agent") + journal.on_tool_end(command, run_id=uuid4()) + messages = [ai, *command.update["messages"]] + updates["artifacts"] = command.update.get("artifacts", []) + if stop_reason is not None: + context["stop_reason"] = stop_reason + updates["messages"] = [*messages, AIMessage(content="The report is complete and ready for delivery.")] + return updates + + graph = StateGraph(get_thread_state_schema(checkpoint_mode)) + graph.add_node("report", agent_node) + graph.add_edge(START, "report") + graph.add_edge("report", END) + compiled = graph.compile(checkpointer=checkpointer) + accessor = CheckpointStateAccessor.bind(compiled, checkpointer, mode=checkpoint_mode) + await accessor.aupdate( + {"configurable": {"thread_id": thread_id}}, + {"messages": [HumanMessage(content="Create and present report.md.")], "title": "Report"}, + as_node="report", + ) + goal = build_goal_state("Create and present report.md.", max_continuations=2) + await write_thread_goal(checkpointer, thread_id, goal) + + await worker.run_agent( + bridge, + manager, + record, + ctx=worker.RunContext(checkpointer=checkpointer, event_store=event_store, checkpoint_channel_mode=checkpoint_mode), + agent_factory=lambda config: compiled, + graph_input={"messages": [HumanMessage(content="Finish the report now.")]}, + config={"configurable": {"thread_id": thread_id}}, + stream_modes=["values"], + ) + + events = await event_store.list_events(thread_id, record.run_id) + return SimpleNamespace( + record=record, + durable_run=await run_store.get(record.run_id), + original_goal=goal, + goal=await read_thread_goal(checkpointer, thread_id), + model_calls=model_calls, + turns=turns, + delivery=[event["content"] for event in events if event["event_type"] == "run.delivery"], + goals_at_receipt=goals_at_receipt, + failed_terminal_writes=failed_terminal_writes, + bridge=bridge, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stop_reason", [None, "token_capped"]) +async def test_delivery_failure_preserves_satisfied_goal_without_continuing(tmp_path, monkeypatch, stop_reason): + result = await _run_goal_delivery(tmp_path, monkeypatch, stop_reason=stop_reason) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.error + assert result.durable_run["status"] == "error" + assert result.record.error == worker._DELIVERY_INCOMPLETE_ERROR + assert result.delivery[0]["produced_paths"] == [OUTPUT_PATH] + assert result.delivery[0]["satisfied"] is False + assert result.goal is not None + assert result.goal["objective"] == result.original_goal["objective"] + assert result.goal["continuation_count"] == 0 + assert all(call.args[2].get("goal") is not None for call in result.bridge.publish.call_args_list if call.args[1] == "values" and "goal" in call.args[2]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("produce,present_on_turn", [(True, 1), (False, None)]) +async def test_successful_delivery_clears_satisfied_goal(tmp_path, monkeypatch, produce, present_on_turn): + result = await _run_goal_delivery(tmp_path, monkeypatch, produce=produce, present_on_turn=present_on_turn) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.success + assert result.durable_run["status"] == "success" + assert result.goal is None + assert len(result.goals_at_receipt) == 1 + assert result.goals_at_receipt[0] is not None + assert result.goals_at_receipt[0]["objective"] == result.original_goal["objective"] + assert len(result.delivery) == 1 + if produce: + assert result.delivery[0]["satisfied"] is True + assert result.delivery[0]["matched_paths"] == [OUTPUT_PATH] + else: + assert result.delivery[0] == {"presented": 0, "paths": [], "by_tool": {}} + + +@pytest.mark.asyncio +async def test_goal_completion_waits_for_durable_delivery_receipt(tmp_path, monkeypatch): + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, fail_receipt=True) + + assert result.record.status == RunStatus.error + assert result.durable_run["status"] == "error" + assert result.record.error == worker._DELIVERY_RECEIPT_FAILED_ERROR + assert result.model_calls == [True] + assert result.turns == [1] + assert result.delivery == [] + assert result.goal is not None + assert result.goal["objective"] == result.original_goal["objective"] + + +@pytest.mark.asyncio +async def test_unmet_goal_can_still_continue_and_deliver(tmp_path, monkeypatch): + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=2, verdicts=(False, True)) + + assert result.model_calls == [False, True] + assert result.turns == [1, 2] + assert result.record.status == RunStatus.success + assert result.durable_run["status"] == "success" + assert result.delivery[0]["satisfied"] is True + assert result.goal is None + assert result.goals_at_receipt[0]["continuation_count"] == 1 + + +@pytest.mark.asyncio +async def test_cancellation_during_evaluation_preserves_goal(tmp_path, monkeypatch): + async def cancel(_checkpointer, manager, record): + await manager.cancel(record.run_id) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, during_evaluation=cancel) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.interrupted + assert result.durable_run["status"] == "interrupted" + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_new_goal_during_delivery_receipt_is_not_cleared(tmp_path, monkeypatch): + replacement = build_goal_state("Prepare the next report.") + + async def replace_goal(checkpointer, _manager, record): + await write_thread_goal(checkpointer, record.thread_id, replacement) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, during_receipt=replace_goal) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.success + assert result.goal == replacement + + +@pytest.mark.asyncio +async def test_ownership_loss_during_delivery_receipt_preserves_goal(tmp_path, monkeypatch): + async def lose_ownership(_checkpointer, manager, record): + await manager._mark_ownership_lost(record, reason="Test lease takeover", require_active=False) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, during_receipt=lose_ownership) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.ownership_lost is True + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_durable_cancellation_during_delivery_receipt_preserves_goal(tmp_path, monkeypatch): + async def cancel(_checkpointer, manager, record): + assert await manager._store.request_cancel(record.run_id, action="interrupt") == "interrupt" + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, heartbeat_enabled=True, during_receipt=cancel) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.interrupted + assert result.durable_run["status"] == "interrupted" + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_changed_goal_evaluation_during_delivery_receipt_is_preserved(tmp_path, monkeypatch): + changed_goals = [] + + async def update_goal(checkpointer, _manager, record): + goal = await read_thread_goal(checkpointer, record.thread_id) + assert goal is not None + changed = { + **goal, + "continuation_count": 1, + "last_evaluation": {"satisfied": False, "blocker": "needs_user_input", "reason": "New evidence needs confirmation.", "evidence_summary": "Waiting for the user."}, + } + changed_goals.append(changed) + await write_thread_goal(checkpointer, record.thread_id, changed) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, during_receipt=update_goal) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.success + assert len(changed_goals) == 1 + assert result.goal == changed_goals[0] + + +@pytest.mark.asyncio +async def test_new_user_message_during_delivery_receipt_preserves_goal(tmp_path, monkeypatch): + async def add_message(checkpointer, _manager, record): + graph = build_state_mutation_graph("new_user", "full") + accessor = CheckpointStateAccessor.bind(graph, checkpointer, mode="full") + await accessor.aupdate( + {"configurable": {"thread_id": record.thread_id}}, + {"messages": [HumanMessage(content="Also include the appendix.")]}, + as_node="new_user", + ) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, during_receipt=add_message) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.success + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_user_clear_during_delivery_receipt_is_not_restored(tmp_path, monkeypatch): + async def clear_goal(checkpointer, _manager, record): + await write_thread_goal(checkpointer, record.thread_id, None) + + result = await _run_goal_delivery(tmp_path, monkeypatch, during_receipt=clear_goal) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.record.status == RunStatus.error + assert result.goal is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["return_false", "raise"]) +async def test_unconfirmed_terminal_status_preserves_goal(tmp_path, monkeypatch, failure): + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, fail_terminal_status=failure) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.delivery[0]["satisfied"] is True + assert result.failed_terminal_writes + assert result.durable_run["status"] == "running" + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_recovered_terminal_status_can_clear_satisfied_goal(tmp_path, monkeypatch): + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, fail_terminal_status="return_false", completion_recovers=True) + + assert result.model_calls == [True] + assert result.turns == [1] + assert result.failed_terminal_writes + assert result.delivery[0]["satisfied"] is True + assert result.durable_run["status"] == "success" + assert result.goal is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("present_on_turn", [None, 1]) +async def test_delta_checkpoint_goal_follows_delivery_outcome(tmp_path, monkeypatch, present_on_turn): + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=present_on_turn, checkpoint_mode="delta") + + assert result.model_calls == [True] + assert result.turns == [1] + if present_on_turn is None: + assert result.record.status == RunStatus.error + assert result.goal == result.original_goal + else: + assert result.record.status == RunStatus.success + assert result.goal is None + + +@pytest.mark.asyncio +async def test_peer_admitted_before_goal_clear_preserves_goal(tmp_path, monkeypatch): + peer_runs = [] + + async def admit_peer(run_store, record): + peer = RunManager(store=run_store, worker_id="peer") + peer_runs.append(await peer.create_or_reject(record.thread_id)) + + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, after_terminal_completion=admit_peer) + + assert len(peer_runs) == 1 + assert result.record.status == RunStatus.success + assert result.durable_run["status"] == "success" + assert result.goal == result.original_goal + + +@pytest.mark.asyncio +async def test_goal_clear_reservation_rejects_peer_admission(tmp_path, monkeypatch): + peers = [] + blocked_admissions = [] + original_write_goal = worker.write_thread_goal + + async def capture_peer(run_store, _record): + peers.append(RunManager(store=run_store, worker_id="peer")) + + async def write_goal(checkpointer, thread_id, goal, **kwargs): + if goal is None: + assert peers, "A goal must not be cleared before terminal completion" + with pytest.raises(ConflictError): + await peers[0].create_or_reject(thread_id) + blocked_admissions.append(thread_id) + return await original_write_goal(checkpointer, thread_id, goal, **kwargs) + + monkeypatch.setattr(worker, "write_thread_goal", write_goal) + result = await _run_goal_delivery(tmp_path, monkeypatch, present_on_turn=1, after_terminal_completion=capture_peer) + + assert blocked_admissions == [result.record.thread_id] + assert result.record.status == RunStatus.success + assert result.durable_run["status"] == "success" + assert result.goal is None diff --git a/backend/tests/test_goal_worker.py b/backend/tests/test_goal_worker.py index 803c2aa31d..8f1d774b04 100644 --- a/backend/tests/test_goal_worker.py +++ b/backend/tests/test_goal_worker.py @@ -196,7 +196,7 @@ async def test_goal_worker_returns_hidden_continuation_when_goal_is_unmet(monkey @pytest.mark.asyncio -async def test_goal_worker_clears_goal_when_evaluator_is_satisfied(monkeypatch): +async def test_goal_worker_defers_satisfied_goal_completion_until_run_finalization(monkeypatch): checkpointer = InMemorySaver() thread_id = "done-goal-thread" await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="Finish all tests") @@ -222,9 +222,9 @@ async def test_goal_worker_clears_goal_when_evaluator_is_satisfied(monkeypatch): app_config=None, ) - assert continuation is None - assert await read_thread_goal(checkpointer, thread_id) is None - assert bridge.events[0][0] == "values" + assert isinstance(continuation, worker._GoalCompletionCandidate) + assert await read_thread_goal(checkpointer, thread_id) == continuation.goal + assert bridge.events == [] @pytest.mark.asyncio @@ -457,7 +457,7 @@ async def test_goal_worker_resumes_after_the_user_answers_clarification(monkeypa @pytest.mark.asyncio -async def test_goal_worker_clears_a_satisfied_goal_even_after_the_run_hit_its_token_budget(monkeypatch): +async def test_goal_worker_defers_satisfied_goal_completion_even_after_token_cap(monkeypatch): checkpointer = InMemorySaver() thread_id = "token-capped-done-goal-thread" await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="Finish all tests") @@ -484,9 +484,9 @@ async def test_goal_worker_clears_a_satisfied_goal_even_after_the_run_hit_its_to run_stop_reason="token_capped", ) - # The satisfied branch runs before the token-cap stand-down: the goal is cleared, not stood down. - assert continuation is None - assert await read_thread_goal(checkpointer, thread_id) is None + # A token cap does not veto completion, but delivery must still succeed. + assert isinstance(continuation, worker._GoalCompletionCandidate) + assert await read_thread_goal(checkpointer, thread_id) == continuation.goal @pytest.mark.asyncio