diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index d10eaf4b9..4041d4185 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -151,7 +151,7 @@ startup gate rejects process-local memory and JSONL event stores when - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. - Thread-scoped run creation accepts an optional `Idempotency-Key` header on create, stream, and wait. Gateway hashes the caller key with the authenticated owner and `thread_id` before passing it to `RunManager`, whose persistence index is process-wide; never pass an unscoped external key to that index. The same scoped key is shared across `/runs`, `/runs/stream`, and `/runs/wait`; a reused admission whose stored `input` or `assistant_id` differs from the retry returns 409. `/wait` must not treat `task is None` as completion: `store_only` records without a cross-process bridge return durable `status`/`error` instead of serializing the current checkpoint; otherwise wait on the bridge. An idempotent reuse must not serialize the latest thread checkpoint as this run's result — a later run on the same thread may have advanced the head — so reused `/wait` returns durable `status`/`error`. Capture that reuse decision before awaiting completion; `idempotency_reused` is sticky on the shared cached record and an overlapping retry must not suppress the original creating request's checkpoint. After observing completion, refresh store-backed `status`/`error` before returning them — a hydrated peer record still holds admission-time fields. A creating-endpoint retry of a terminal record whose stream is gone emits SSE `gap`/`stream_replay_gap` with `recovery: reload_durable_state` rather than a bare `end`; observer joins of that same record still emit `end`. That gap is opt-in via `sse_consumer(..., emit_gap_on_missing_stream=True)` from thread-scoped `/runs/stream` on this request's reuse — do not key it off `apply_on_disconnect` or the sticky `idempotency_reused` flag. Default `sse_consumer` callers, including stateless `/api/runs/stream`, still emit `end`. A reused still-running `store_only` record on a process-local bridge returns 409 from `/stream` with no `Retry-After`, matching `join`. Missing headers preserve ordinary non-idempotent admission. Stateless `/api/runs/*` stays outside this contract because a request without an explicit thread creates a fresh temporary thread before admission. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. In `delta` checkpoint mode the worker rewrites that fork into a linear head write before the graph starts (see "A delta-mode run cannot fork" under Checkpoint Channel Modes), because delta state for a fork replays the abandoned sibling's writes. -- Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. +- Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the run has not hit its token-budget hard stop (`stop_reason == "token_capped"`; the goal stands down with `stand_down_reason: "token_capped"`), the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. - Run event stream changes must keep producer code, `deerflow/constants.py`, `runtime/events/catalog.py`, `contracts/run_event_stream_contract.json`, `backend/docs/RUN_EVENT_STREAM.md`, and `tests/test_run_event_stream_contract.py` in sync. The dependency-free constants module owns the persisted envelope limits (`event_type` 32 characters, `category` 16) and cross-layer workspace event identity; the catalog owns validated runtime definitions and categories. Dynamic middleware tags are limited to 21 characters after the `middleware:` prefix. The JSON contract owns payload schemas, backend-specific storage semantics, legacy aliases, and compatibility rules; conformance tests require both views and all producer groups to agree. `run.end.content` remains opaque and may retain nested Python values in memory while JSONL/database stores stringify non-JSON nested values, so consumers must not assume backend-identical nested output representations. Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runtime, all other `/api/*` → Gateway REST APIs. diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index c21634892..0111876a4 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -1325,6 +1325,7 @@ async def run_agent( deerflow_trace_id=deerflow_trace_id, task_store=task_store, extensions=extensions, + run_stop_reason=runtime.context.get("stop_reason") if isinstance(runtime.context, dict) else None, ) if continuation_input is None or record.abort_event.is_set(): break @@ -1870,6 +1871,7 @@ async def _prepare_goal_continuation_input( deerflow_trace_id: str | None = None, 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. @@ -2007,6 +2009,11 @@ async def _prepare_goal_continuation_input( return None stand_down_reason = _stand_down_reason(goal, evaluation, no_progress_count) + if stand_down_reason is None and run_stop_reason == "token_capped": + # The run already used up its token budget, and continuations share that + # budget, so another hidden turn would spend one more model call only to + # have its tool calls stripped. + stand_down_reason = "token_capped" if stand_down_reason is not None or not should_continue_goal(goal, evaluation, no_progress_count=no_progress_count): await _persist(goal, evaluation, no_progress_count, stand_down_reason=stand_down_reason) return None diff --git a/backend/tests/test_goal_worker.py b/backend/tests/test_goal_worker.py index 2c910d678..6bae94f53 100644 --- a/backend/tests/test_goal_worker.py +++ b/backend/tests/test_goal_worker.py @@ -309,6 +309,75 @@ async def test_goal_worker_stands_down_for_non_continuable_blocker(monkeypatch): assert latest_goal["last_evaluation"]["stand_down_reason"] == "blocked:missing_evidence" +@pytest.mark.asyncio +async def test_goal_worker_stands_down_after_the_run_hit_its_token_budget(monkeypatch): + checkpointer = InMemorySaver() + thread_id = "token-capped-goal-thread" + await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="Finish all tests") + bridge = _CollectingBridge() + + async def fake_evaluate_goal_completion(_goal, _messages, **_kwargs): + return GoalEvaluation( + satisfied=False, + blocker="goal_not_met_yet", + reason="Tests have not passed yet.", + evidence_summary="Implementation is incomplete.", + ) + + monkeypatch.setattr(worker, "evaluate_goal_completion", fake_evaluate_goal_completion) + + continuation = await worker._prepare_goal_continuation_input( + accessor=_full_accessor(checkpointer), + bridge=bridge, + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-capped", + model_name="test-model", + app_config=None, + run_stop_reason="token_capped", + ) + + assert continuation is None + latest_goal = await read_thread_goal(checkpointer, thread_id) + assert latest_goal is not None + assert latest_goal["continuation_count"] == 0 + assert latest_goal["last_evaluation"]["blocker"] == "goal_not_met_yet" + assert latest_goal["last_evaluation"]["stand_down_reason"] == "token_capped" + + +@pytest.mark.asyncio +async def test_goal_worker_clears_a_satisfied_goal_even_after_the_run_hit_its_token_budget(monkeypatch): + checkpointer = InMemorySaver() + thread_id = "token-capped-done-goal-thread" + await _seed_goal_thread(checkpointer, thread_id=thread_id, goal_text="Finish all tests") + bridge = _CollectingBridge() + + async def fake_evaluate_goal_completion(_goal, _messages, **_kwargs): + return GoalEvaluation( + satisfied=True, + blocker="none", + reason="The visible conversation says the task is done.", + evidence_summary="Done.", + ) + + monkeypatch.setattr(worker, "evaluate_goal_completion", fake_evaluate_goal_completion) + + continuation = await worker._prepare_goal_continuation_input( + accessor=_full_accessor(checkpointer), + bridge=bridge, + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-capped-done", + model_name="test-model", + app_config=None, + 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 + + @pytest.mark.asyncio async def test_goal_worker_stands_down_when_no_progress_repeats(monkeypatch): checkpointer = InMemorySaver() @@ -682,6 +751,97 @@ async def test_run_agent_does_not_stream_continuation_after_abort(monkeypatch): assert record.status == RunStatus.interrupted +@pytest.mark.asyncio +async def test_run_agent_passes_the_run_stop_reason_to_the_goal_loop(monkeypatch): + class FakeAgent: + def __init__(self) -> None: + self.inputs = [] + self.metadata = {} + self.checkpointer = None + self.store = None + self.interrupt_before_nodes = [] + self.interrupt_after_nodes = [] + + def astream(self, input_payload, **kwargs): + self.inputs.append(input_payload) + # TokenBudgetMiddleware stamps the cap into the run's runtime context. + kwargs["config"]["configurable"]["__pregel_runtime"].context["stop_reason"] = "token_capped" + + async def _gen(): + yield {"messages": []} + + return _gen() + + class FakeRunManager: + async def try_start(self, _run_id): + record.status = RunStatus.running + return RunStartOutcome.started + + async def set_status(self, _run_id, status, **_kwargs): + record.status = status + + async def set_status_if_not_cancelled(self, _run_id, status, **kwargs): + await self.set_status(_run_id, status, **kwargs) + return None + + async def update_model_name(self, *_args, **_kwargs): + return None + + async def update_run_completion(self, *_args, **_kwargs): + return None + + async def wait_for_prior_finalizing(self, *_args, **_kwargs): + return None + + async def set_finalizing(self, _run_id, finalizing): + record.finalizing = finalizing + + async def cleanup(self, *_args, **_kwargs): + return None + + class FakeBridge: + async def publish(self, *_args, **_kwargs): + return None + + async def publish_end(self, *_args, **_kwargs): + return None + + async def cleanup(self, *_args, **_kwargs): + return None + + stop_reasons = [] + + async def fake_prepare(**kwargs): + stop_reasons.append(kwargs.get("run_stop_reason")) + return None + + monkeypatch.setattr(worker, "_prepare_goal_continuation_input", fake_prepare) + + fake_agent = FakeAgent() + record = RunRecord( + run_id="run-token-capped", + thread_id="thread-token-capped", + assistant_id="lead-agent", + status=RunStatus.pending, + on_disconnect=DisconnectMode.cancel, + model_name="test-model", + ) + record.abort_event = asyncio.Event() + + await worker.run_agent( + FakeBridge(), + FakeRunManager(), + record, + ctx=worker.RunContext(checkpointer=None), + agent_factory=lambda config: fake_agent, + graph_input={"messages": [HumanMessage(content="start")]}, + config={"configurable": {"thread_id": "thread-token-capped"}}, + ) + + assert stop_reasons == ["token_capped"] + assert len(fake_agent.inputs) == 1 + + @pytest.mark.asyncio async def test_run_agent_reuses_goal_evaluator_model_for_goal_loop(monkeypatch): class FakeAgent: