diff --git a/README.md b/README.md index 6d96b6278d..fec4f6e355 100644 --- a/README.md +++ b/README.md @@ -1679,6 +1679,8 @@ losing that turn's per-run delegation count. The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions — when delegation has clear net benefit from real parallel latency, specialist capability, or context isolation. It keeps interdependent scopes and overlapping side effects out of parallel dispatch; a bounded sequential chain can still run in one sub-agent when specialist or context-isolation benefit clearly wins. The lead uses the fewest useful sub-agents and re-evaluates later batches instead of fanning out solely because a task is large or multi-step. Sub-agents report back structured results, and the lead agent verifies and synthesizes them into a coherent output. Deterministic tool receipts cover both direct tool messages and state-updating `Command` results such as delegated `task` responses; when the receipt ledger reaches its context budget, it retains the newest actions and their original receipt IDs. Operators can disable this provenance layer with `verification.receipts_enabled: false`. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Reloaded thread history enforces the same boundary: callback-captured sub-agent AI responses remain available in run-event diagnostics but are excluded from the parent transcript, while the parent `task` result remains attached to its subtask card. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Their system instructions, including the role and report contract, survive compaction; if only those instructions and the current request would be summarized, compaction is skipped. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Concurrent parent runs also receive independent server-side sub-agent execution IDs, so a provider that reuses a tool-call ID cannot make one run poll, cancel, or clean up another run's background task. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step from that run's terminal tool-message metadata rather than a process-global provider-ID cache. +Cancelled or timed-out background sub-agent executions retain provider-reported token usage from completed model calls, including responses received before their next progress update. Final usage delivery to the parent run does not count earlier progress snapshots twice. + For file acceptance criteria, an empty regular file in the shared workspace can satisfy `file: exists` and `file_written:`, including on remote sandboxes. It fails `file: non-empty` with a deterministic empty-file result. To request JSON syntax validation, explicitly set a `task` or `batch_task` item's diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index b7e95f1b66..b73247415b 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -19,6 +19,11 @@ citation references, and opaque base64/URL/image sources. **Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock until service shutdown finishes. Drain without a timeout so owned work cannot detach; propagate the first caller cancellation, chaining any service failure. Cover terminal outcomes and repeated cancellation in `tests/test_subagent_runtime.py`. +**Interrupted usage**: After stream cleanup, publish the collector's final +snapshot on cancellation/timeout; callbacks can precede values chunks. +`execute_async()` owns terminalization; `SubagentResult` guards first write. +Parent journals deduplicate by `source_run_id`; batch writes need a live lease. + **Durable batch acceptance**: Normalize optional `batch_task` criteria before persistence (empty → null; 20 items × 500 neutralized characters) with shared `normalize_acceptance_criteria`. Completed items use `acceptance_checks` through @@ -31,7 +36,7 @@ does not change retry policy. **Built-in Agents**: `general-purpose` (all tools except `task`) and `bash` (command specialist) **Registry and managed definitions**: Runtime resolution is built-in → `config.yaml custom_agents` → enabled administrator-managed definitions, followed by explicit `subagents.agents.` overrides. Managed definitions are deployment-wide, persist through the same `agent_storage.backend` selection as Custom Agent definitions, and remain stored but are excluded from runtime when a built-in or later-added config definition owns the same name. The default Lead Agent sees the whole enabled catalog. A Custom Agent's `allowed_subagents` is snapshotted into run metadata (`None` = all, `[]` = hard deny, list = allowlist) and must filter both prompt discovery and `task` execution; never reload caller policy from mutable agent config inside the tool. -**Benefit-based routing policy**: Enabling subagents exposes delegation as an optimization, not a default response to complexity. The lead prompt defaults to direct execution and permits `task` only when parallel latency, specialist capability, or context-isolation benefit clearly exceeds startup, duplicate-discovery, synthesis, state-conflict, and side-effect costs. Inter-agent output dependencies and overlapping mutable state are hard vetoes for parallel dispatch, while duplicate discovery and a cheap direct path remain costs rather than categorical vetoes; a bounded sequential chain may run in one subagent when specialist or context-isolation benefit clearly wins. Parallel scopes must be independent and non-overlapping, the lead uses the fewest useful subagents, and every later batch is re-evaluated while retaining any within-batch parallel benefit. When the enforced per-response limit is 1, the rendered prompt removes parallel and multi-batch benefit guidance and permits delegation only for material specialist or context-isolation benefit. Keep this policy aligned across `lead_agent/prompt.py`, the `task` tool description, and both built-in role descriptions; routing regressions are pinned in `tests/test_subagent_routing_prompt.py`, `tests/test_subagent_prompt_security.py`, and `tests/test_lead_agent_prompt.py`. +**Benefit-based routing policy**: Direct execution is the default. Permit `task` when parallel latency, specialist skill, or context isolation clearly outweighs startup, duplicate discovery, synthesis, state conflicts, and side effects. Parallel scopes need independent outputs and disjoint mutable state; use the fewest useful agents and reassess each batch. A bounded sequential subagent is allowed for strong specialist/context benefit. With a per-response limit of 1, prompts omit parallel and multi-batch benefits and allow only material specialist/context benefit. Keep `lead_agent/prompt.py`, the `task` description, and built-in roles aligned; see `tests/test_subagent_routing_prompt.py`, `tests/test_subagent_prompt_security.py`, and `tests/test_lead_agent_prompt.py`. **User-scoped Skills**: Subagents resolve their configured skills through `get_or_new_user_skill_storage(user_id)` using the parent runtime identity, with `DEFAULT_USER_ID` only when no identity is available. This keeps custom-skill shadowing and visibility aligned with the lead agent instead of reading the global-only catalog. **Upload-state boundary**: Ordinary `task` delegation snapshots a valid parent `ThreadState.uploaded_files` list at dispatch, deep-copies it across the isolated-loop boundary, seeds it into the child's fresh state, and only then makes `list_uploaded_files` eligible for normal tool-policy filtering. An explicit empty list is valid and must be preserved because it means every upload in the thread is historical for this run. Missing or malformed state fails closed with the tool disabled. Durable `batch_task` execution intentionally keeps the tool disabled: delayed and recovered items have no valid parent-run upload boundary, and supporting that case requires a separate persisted-state contract. **Knowledge-scope boundary**: `task` and `batch_task` inherit the parent's canonical execution scope, never display data. Batch specs persist it across lease recovery; malformed scope fails. Model input cannot set or broaden it, and shared middleware/retrieval enforce `disabled`. @@ -39,7 +44,7 @@ does not change retry policy. **Execution**: Ordinary and durable-batch native subagents submit coroutines directly to one persistent isolated event loop. Gateway/embedded startup installs one process-wide async FIFO admission controller (default 3 running, bounded queue). Direct `create_deerflow_agent` callers can instead pass a caller-owned `SubagentRuntime`; reuse the same instance across graphs so its bound `task`, optional batch tools/service, middleware limits, and `SubagentExecutor` all share one controller without reading global YAML. An owned batch service must be started before graph construction and stopped at application shutdown. Waiters hold no scheduler thread, and cancellation/timeout release queue/slot ownership. **Shared sandbox execution lifecycle** (#5128): every admitted subagent run carries a stable task-derived `sandbox_lease_owner_id` and matching `sandbox_command_scope_id` in its runtime context. Sandbox middleware retains that execution against the lead thread's active provider client, so one child finishing cannot close the sandbox while siblings still run; the final holder performs any pending provider release. A rollback/fork-restored child reusing the parent's live client binds a non-releasing holder: it fences parent cleanup and owns its command scope without requesting a park itself; a parent's earlier park request waits for the child, while a missing inherited client falls through to a normal fresh acquire. On AIO, the command scope selects one explicit persistent shell session per subagent, allowing independent scopes to run concurrently while preserving in-order shell state within one child. Sync sandbox tool bodies offloaded with `asyncio.to_thread` are shielded and drained across repeated cancellation before the outer execution can clean its holder; a cancelled worker can therefore neither re-admit an already-released owner nor run after subagent terminalization. `SubagentExecutor` invokes active LangGraph stream cleanup before releasing its sandbox lease or notifying task stop. Slow cooperative cleanup keeps the result non-terminal and retains its sandbox lease and capacity slot; its warning is diagnostic, not a safe hard-timeout boundary. Middleware performs the normal release, and `SubagentExecutor` repeats it idempotently in `finally` so exceptions, cooperative cancellation, and timeout unwind paths cannot leak a lease or scoped session. **Concurrency and total delegation cap**: Ordinary `task` concurrency is resolved once as the minimum of the per-run request, the startup-frozen `subagent_runtime.max_running`, and the schema safety ceiling (1-64), then shared by the lead prompt and `SubagentLimitMiddleware`. Hot reloads must not make either layer advertise more capacity than the already-created process controller; a changed startup-only value takes effect only after restart. The same middleware separately enforces `subagents.max_total_per_run` (default 6, config schema 1-50, runtime override `max_total_subagents` clamped to the same range) against current-run entries in the durable delegation ledger, so a long lead-agent run cannot bypass concurrency limits by launching repeated legal-sized batches at each planning checkpoint, but historical delegations from previous runs in the same thread do not consume the new run's budget. Explicit `batch_task` work does not consume or relax that ordinary-run ledger: its persisted total/live/running limits live under `subagent_batches`. Gateway `run_agent()` and embedded `DeerFlowClient.stream()` both provide a per-invocation `run_id` in runtime context; `DeerFlowClient.stream()` also tags its input `HumanMessage` with that same id so durable-context capture can identify the current request boundary. Gateway resume paths may not append a new `HumanMessage`, so the worker also exposes the pre-run checkpoint's message ids in runtime context; durable-context capture uses that as the current-run boundary and never re-tags older task calls as the resumed run. When no delegation slots remain, task calls are stripped, provider raw tool-call metadata is synced, `finish_reason` is forced to `stop`, and a visible "subagent delegation limit" note is appended so the agent can synthesize already-collected results. Default subagent timeout `subagents.timeout_seconds=1800` (30 min) and built-in `general-purpose` `max_turns=150`. -**Flow**: Ordinary `task()` → `SubagentExecutor` → shared process slot → result polling/SSE. Explicit `batch_task()` → durable batch/item rows → lease-based batch service (`subagents/batch_service.py`, started by Gateway or an explicit direct runtime) → the same `SubagentExecutor`/process slots → bounded stored result and owner-scoped API/JSONL export. Batch mode is selected only by the explicit tool, never inferred from prompt size. Executor queue rejection/timeout occurs before model execution and therefore releases the durable lease without consuming an item attempt; real execution failure and expired leases still consume the retry budget. User cancellation terminalizes every nonterminal item immediately and clears its lease, fencing any stale worker completion. Background cancellation resolves the result/future under `_background_tasks_lock` but calls `Future.cancel()` only after releasing it, because cancellation may synchronously invoke the completion callback that reacquires the registry lock. Direct runtimes provide the tools and worker but not Gateway's HTTP/UI surface. `task_started` carries the resolved effective model name. The per-subagent `SubagentTokenCollector` publishes a cumulative usage snapshot to the shared `SubagentResult` after every completed LLM response; the next `task_running` event carries that snapshot, so collapsed workspace cards can update without re-accounting parent-run totals. Terminal ToolMessage metadata (`subagent_model_name`, `subagent_token_usage`) and the persisted `subagent.end` event retain the model/usage after reload; absent provider usage stays absent rather than being estimated as zero. The executor caches one resolved `AppConfig` snapshot (explicit or `get_app_config()` fallback) for agent assembly, deferred setup, and receipt harvesting, so `verification.receipts_enabled=false` remains authoritative on both construction paths. Terminal tool receipts are harvested before `try_set_terminal` and committed with the other payload fields under the same state lock, so status polling cannot observe a terminal result before its receipt metadata is available. Each yielded values chunk becomes the latest terminal-harvest state and immediately publishes its harvested receipts to the shared result before cooperative cancellation is checked. Tool-ended cancellation/failure evidence uses the current ToolMessage scan, but a completed result always uses the bounded ledger snapshot attached to the assistant text being returned—even when a max-turn partial ends on a later tool chunk—so omitted receipts cannot validate its citations; a missing/malformed completed snapshot fails closed with no receipts. Therefore direct task cancellation and both execution/polling timeouts retain the latest execution evidence even when cancellation interrupts before another stream boundary. +**Flow**: `task()` → `SubagentExecutor` → shared slot → polling/SSE. Explicit `batch_task()` → durable rows → lease-based batch service (`subagents/batch_service.py`, Gateway or direct runtime) → same executor/slots → bounded result and owner-scoped API/JSONL. Batch mode is explicit, never inferred from prompt size. Queue rejection/timeout releases the lease without an item attempt; execution failure/expired leases consume one. User cancellation terminalizes nonterminal items and clears leases, fencing stale workers. Resolve background cancellation under `_background_tasks_lock`, then call `Future.cancel()` outside it because callbacks may reacquire the lock. `task_started` includes the effective model. `SubagentTokenCollector` publishes cumulative usage after each LLM response; `task_running` updates cards without double-counting. Terminal ToolMessage metadata (`subagent_model_name`, `subagent_token_usage`) and `subagent.end` preserve model/usage after reload; absent usage stays absent. Resolve one `AppConfig` snapshot (explicit or `get_app_config()`) for assembly, deferred setup, and receipt harvesting; `verification.receipts_enabled=false` holds on both construction paths. Harvest terminal receipts before `try_set_terminal` under the result lock so polling cannot see terminal status without metadata. Each values chunk updates the latest harvest state and publishes receipts before cooperative cancellation. Cancellation/failure scans current ToolMessages; completion uses the bounded ledger snapshot attached to returned assistant text, including max-turn partials. Missing/malformed completed snapshots yield no receipts, preventing omitted receipts from validating citations. Cancellation/timeouts retain the latest streamed evidence. **Report contract (RFC #4651 PR3)**: `report_contract.py` owns the prompt-layer text that makes Layer 1 receipt verification non-inert. `SubagentExecutor._build_initial_state` appends `build_report_contract_section(receipts_enabled=...)` to every subagent's consolidated `SystemMessage` — built-in and custom alike — requiring `[rN tool_name]` citations (from the Tool receipts ledger) for action claims, verifiable handles (absolute path, URL, ID, HTTP status) for deliverables, and explicit reporting of failures; the citation clause follows `verification.receipts_enabled`, and the citation example derives from the single-owner `format_citation`/`receipt_id` so prompt text cannot drift from the verifier. The `task` tool hands lead-supplied `acceptance_criteria` to the `SubagentExecutor` constructor, which appends them via `render_acceptance_criteria_block(...)` to the task `HumanMessage` (stripped, capped at 20 items × 500 chars, each entry neutralized) — the untrusted channel `InputSanitizationMiddleware` escapes and boundary-frames, matching their model-supplied provenance. The subagent's `SystemMessage` never carries criterion text; it gets only the framework-owned `build_acceptance_criteria_system_note(...)` pointer naming the list's location and authority, so natural-language injection inside a criterion cannot gain system-channel priority over framework instructions. Deterministic leaf checking is a separate layer. **Acceptance checklist (RFC #4651 PR4)**: `acceptance_checks.py` checks lead-supplied `acceptance_criteria` in code on the `task` tool's completed branch (offloaded via `asyncio.to_thread`, failure-isolated). Decidable leaves: `file: exists|non-empty` and `file_written:` read through `read_current_file_content` scoped to the shared thread workspace (`workspace_path`/`outputs_path`; virtual `/mnt/user-data/...` prefixes and workspace-relative spellings normalized first) — the read itself uses the sandbox-native virtual form because the local read validator and provider mount tables resolve virtual paths, not host paths; a remote provider's `"Error: ..."` return string (remote providers return instead of raise for missing files) is normalized to a failed check, never evaluated as content — provider-typed via `is_local_sandbox`, so a genuine `Error:`-prefixed file on the local sandbox stays valid content; a UnicodeDecodeError on a text read marks a binary deliverable (PDF/image) as existing and non-empty instead of dropping the verdict; reads are byte-bounded — the size is established first (`os.stat` on the validated host path locally, so the host-bash-disabled configuration needs no shell; a metadata-only `stat`/`realpath` probe in a fresh `env -i` shell on remote providers — absolute-path utilities and a marker env that routes AIO off its persistent shell, so a completed subagent's poisoned session state cannot steer it; `stat` never opens content, so a FIFO cannot block the parent for the provider's idle timeout; the file must stat as a regular non-symlink file, and containment is canonicalized — the file's realpath must stay under the mount root's realpath, which is exactly what the provider's own read path resolves (e2b and Tenki realize `/mnt/user-data` as a symlink to the home dir by default), so a final-component symlink is rejected outright and an intermediate dir-link escape under a sane root still lands outside the canonical root), leaves above `_FILE_CONTENT_READ_CAP_BYTES` answer from the size alone — `file_written` only with an added bounded one-byte open probe (a mode-000 file stats fine while any open raises EACCES), and smaller files run the full read, and an unestablishable size degrades to UNVERIFIED rather than an unbounded fallback read; out-of-scope paths degrade to UNVERIFIED, never misjudge — on the local sandbox the scope decision canonicalizes with realpath, so a workspace symlink into uploads cannot satisfy a scoped leaf with upload content. `tests_passed:` anchors to a matching bash execution (newest match wins) with `status=success` and a test-summary shape in its bounded output tail — each harvested execution carries a `shell_persistent` provenance stamp — the producing sandbox's `persistent_shell_sessions` flag (AIO's legacy exec path) resolved from the state that carried the evidence, never the parent runtime, which has no `sandbox` key when the parent delegated before touching one — and a persistent stamp (or an unidentifiable or undeclared one — a custom provider that never declared its session semantics is unknown, not fresh-shell; unknown provenance fails closed) degrades the leaf to UNVERIFIED instead, because any earlier call in the shared session could have mutated the state the clean-looking run executed in and a fresh controlled session (RFC §6 verifier) would be needed to prove otherwise, harvested by the executor (`_harvest_bash_executions`, only when criteria were delegated) from the same stamped `ToolMessage`s the receipt layer reads, accumulated per streamed chunk (merged by `tool_call_id`, newest-capped) so subagent summarization compacting earlier messages cannot erase a recorded execution, with over-cap commands carrying `command_truncated` so the matcher degrades to UNVERIFIED instead of proving a match on a suffix-less prefix; the recorded status is the actual shell exit status parsed from the output's `Exit Code: N` / `Command exited with code N` marker (a nonzero bash exit returns ordinary text that `deerflow_tool_meta` still reports as success; every provider appends a marker on nonzero exit — local/e2b/opensandbox/tenki/boxlite textually, aio via the SDK's structured exit_code — and local timeouts append `Exit Code: 124` while signal kills parse as signed markers (`Exit Code: -9` records error); `_truncate_bash_output` preserves a trailing exit marker inside its budget, with a 32-char floor for smaller configured limits and the remote `Command exited with code N` form accepted only as the whole trimmed output; the matched marker text travels on the entry as `status_marker`), falling back to the meta status only when no marker exists; matching is shell-structure aware (operator-separated segments — a physical newline separates with `;` semantics — or with the continuation operator the next line opens with (`cmd1\n&& cmd2` is `&&`; `cmd1\n|| cmd2` is `||`, which after a successful first command skips the rest while exiting 0, so flattening it to `;` would record a run that never happened) —, so a multi-line script's trailing `echo`/`seq` lines are never merged into the matched segment's arguments, status, or output attribution —, executable identity — directional: a bare criterion executable accepts any path spelling of the name, while an explicitly path-spelled criterion requires a path-spelled execution of the same normalized executable path (spelling judged on the raw token: `./pytest` names the project-local file and normpath collapsing `./` must not demote it to a PATH lookup; a `..` component on either side is unprovable outright), so `/tmp/fake/pytest` cannot certify `/opt/project/.venv/bin/pytest` —, ordered argument subsequence whose env-assignment prefix must equal the criterion's exactly as an effective name → final-value mapping — extra, missing, or differently-valued assignments degrade the match, as does any reordering of a repeated name (`CI=0 CI=1` vs `CI=1 CI=0` are last-wins opposite environments), since no variable is provably inert across repositories, and any assignment or argumented `export`/`unset` in a preceding segment is state pollution, and any span token carrying a runtime expansion (`$VAR`/`$( )`/backticks) or an extra token carrying glob metacharacters (crafted option-looking filenames narrow invisibly) is likewise unprovable), so a command that merely mentions the criterion string (`echo '12 passed'; # pytest x.py`) cannot anchor the leaf, and control flow is preserved — the matching span must end at the command's last segment with provable execution (`&&` needs recorded success, `||` needs recorded failure, pipelines inside the span and backgrounding are never provable), and the criterion's own connectors are preserved — an expected `&&` executed as `;` (`cd missing; pytest x` for `cd missing && pytest x`) lets a failed preceding step be bypassed, so it degrades; only the stricter direction (criterion `;` executed as `&&` with recorded success) survives), so a short-circuited segment (`false && pytest x; echo '3 passed'`) degrades to UNVERIFIED instead of a false hold; the summary shape is evaluated only when the output is attributable to the matched segment (every preceding segment provably silent by invocation form — only shape-free `cd dir` and pure assignments qualify; `pushd`/`umask`/`ulimit`, any `export`/`unset` (including an invalid identifier, whose bash error text is subagent-chosen), and any `source`/`.`, whose `*/bin/activate` path shape says nothing about what a crafted script prints, are not; the `cd` print channel is closed too — CDPATH makes `cd` print the subagent-chosen resolved path and the pass shapes match as substrings, so a `cd` argument carrying a summary shape or runtime expansion is non-silent, and any `CDPATH=` assignment degrades the match as state pollution), pass shapes require a nonzero passed count (`0 passed`, go `[no test files]`, `Ran 0 tests` veto), argument matching treats tokens consumed by negating options (`--ignore`/`--deselect`/`--exclude`/`--skip` families) as ineligible evidence (`pytest --ignore tests/security tests` cannot affirm the excluded tests) and a target negated anywhere in the command degrades the match (`pytest tests/security tests/unit --ignore tests/security` cannot certify the excluded target), and a bare criterion with no positional selection target (bare `pytest`, `make test`) treats ANY negating option as a different selection, and accepts extra executed flags only when provably selection-preserving (verbosity/output/xdist/coverage forms; `-k`/`-m`/`--lf`/`--collect-only`/`-c`/`-p` and friends degrade to UNVERIFIED — `pytest -k smoke tests/security` cannot certify the full selection; extra positionals widen only after a path-scoped criterion — scoping reads positional targets by option arity, so a path embedded in an option (`--basetemp=/tmp/p`) never counts as a target, and an option of unknown arity (absent from the value-taking table, no glued `=`) followed by a path-like token (`--rootdir /tmp/project`) fails closed, so `python -m unittest pkg.OneTest` cannot certify bare `python -m unittest`), and criterion text is neutralized with `neutralize_untrusted_tags` before storage/rendering so a blocked tag in a criterion can never reach the lead-visible checklist section. Any other criterion is UNVERIFIED, never silently passed. The verdict (`subagent_acceptance_verdict` in `additional_kwargs`, validated on read) feeds the delegation ledger's `acceptance:` segment and the model-visible checklist section appended to the result text; the gateway strips caller-forged verdicts from both ledger entries and message metadata like the citation verdict. Leaf read errors catch only `OSError`/`SandboxError` — unexpected errors propagate to the task-tool-level isolation so the blocking-IO gate keeps teeth (`tests/blocking_io/test_task_tool_acceptance_checklist.py`). Known accepted boundaries (pinned by `TestKnownBoundaries` — conscious, do not re-raise): a bare criterion executable trusts PATH/filesystem spelling (a subagent-crafted same-name script is textually indistinguishable; spelling an absolute criterion path restricts evidence to that normalized path); relative criterion targets resolve in the wrapper's cwd, so `cd` prefixes must stay in scope (relative without `..`, or absolute under the thread data roots / virtual prefix) while a symlink inside an allowed root pointing out is a filesystem-layer concern; runner semantics are trusted (a Makefile swallowing failures, a runner exiting 0 on failure) — Layer 2 is execution evidence only, claim correctness belongs to the PR5 judge / RFC §6 re-execution; evidence is bounded (500-char command, 1000-char tail) and truncation degrades to UNVERIFIED rather than proving. **Acceptance checklist path portability**: Paths are host-independent and raw `..` fails closed. Drive/UNC absolutes retain their class across `ntpath` normalization and cannot cross the root; drive-relative, shell-dependent provider/PSDrive, and POSIX-rooted `cd` on Windows are unprovable. Selection overlap separates pytest node IDs, normalizes safe `.`/duplicate separators, rejects `..`, and applies Windows casing to drive, UNC, PSDrive, or Windows-context paths; provider-qualified drive/UNC roots remain paths until a later node-ID split. POSIX paths and all node IDs keep their required case sensitivity. Ambiguous trailing-dot/space or 8.3 components, different volume IDs, cross-family pairs, and absolute/relative PSDrive pairs fail closed without execution/filesystem provenance; only same-volume, same-root-form paths compare lexically. Without shell provenance, raw commands reject backslashes; cmd `%VAR%`/`!VAR!`, `^`, `#`; Bash braces/tilde; and PowerShell splatting, typographic quotes, or unquoted parentheses. A lone `%` remains eligible. POSIX-only markers must short-circuit on Windows because pytest evaluates them at import. diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 9b8649b3e8..eecb1a4410 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -1801,6 +1801,13 @@ class SubagentExecutor: ) finally: + # Cancellation (including wait_for's execution timeout) bypasses + # the terminal branches above. A completed model response can be + # in the collector before its graph node publishes a values chunk. + # Stream teardown has drained at this point; publish its final + # usage before the outer wrapper chooses CANCELLED or TIMED_OUT. + if collector is not None: + result.update_token_usage_records(collector.snapshot_records()) if execution_context is not None and execution_context.get("sandbox_id") is not None: try: from deerflow.sandbox import get_sandbox_provider diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 8deed6c7c1..646384fc7d 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -32,6 +32,7 @@ import pytest from packaging.version import Version from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware +from deerflow.runtime.journal import RunJournal from deerflow.sandbox.lease import SandboxLeaseManager from deerflow.skills.types import Skill from deerflow.subagents.capacity import SubagentCapacityRejected @@ -5137,6 +5138,198 @@ class TestSubagentGuardrailAttribution: ) +class TestInterruptedTokenUsage: + @pytest.mark.parametrize("outcome", ["cancelled", "timed_out", "completed"]) + def test_retains_usage_before_model_node_publishes_state(self, classes, monkeypatch, outcome): + """A completed model call costs tokens even if its wrapper is interrupted.""" + from langchain.agents import create_agent + from langchain.agents.middleware import AgentMiddleware + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + from langchain_core.tools import tool + + from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig + from deerflow.subagents.capacity import SubagentExecutionCapacity + from deerflow.tools.builtins.task_tool import _report_usage_records, _summarize_usage, _task_result_command + + executor_module = importlib.import_module("deerflow.subagents.executor") + model_returned = threading.Event() + terminal = threading.Event() + gate = {} + + class Model(FakeMessagesListChatModel): + def bind_tools(self, tools, **kwargs): + return self + + class PauseAfterModel(AgentMiddleware): + async def awrap_model_call(self, request, handler): + response = await handler(request) + # The first model node has already published its 15 tokens. + # Pause the second *after* on_llm_end but *before* this model + # node can yield a values chunk to SubagentExecutor. + if not response.result[0].tool_calls: + gate["loop"] = asyncio.get_running_loop() + gate["release"] = asyncio.Event() + model_returned.set() + await gate["release"].wait() + return response + + @tool + def echo(value: str) -> str: + """Return the input.""" + return value + + ai = classes["AIMessage"] + model = Model( + responses=[ + ai( + content="", + tool_calls=[{"name": "echo", "args": {"value": "ok"}, "id": "echo-1", "type": "tool_call"}], + usage_metadata={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + response_metadata={"model_name": "usage-test-model"}, + ), + ai( + content="Done", + usage_metadata={"input_tokens": 20, "output_tokens": 10, "total_tokens": 30}, + response_metadata={"model_name": "usage-test-model"}, + ), + ] + ) + graph = create_agent(model, tools=[echo], middleware=[PauseAfterModel()], checkpointer=False) + executor = classes["SubagentExecutor"]( + config=classes["SubagentConfig"]( + name="usage-test", + description="Test interrupted usage accounting", + model="usage-test-model", + timeout_seconds=5 if outcome == "timed_out" else 10, + ), + tools=[], + extensions=SimpleNamespace(needs_task_store=False, has_task_lifecycle=False), + execution_capacity=SubagentExecutionCapacity(SubagentRuntimeConfig()), + ) + monkeypatch.setattr(executor, "_build_initial_state", AsyncMock(return_value=({"messages": [classes["HumanMessage"](content="Echo ok")]}, [], None))) + monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: graph) + monkeypatch.setattr(executor_module, "build_tracing_callbacks", lambda: []) + monkeypatch.setattr(executor_module, "inject_langfuse_metadata", lambda *args, **kwargs: None) + original_terminal = classes["SubagentResult"].try_set_terminal + + def signal_terminal(holder, *args, **kwargs): + changed = original_terminal(holder, *args, **kwargs) + if changed: + terminal.set() + return changed + + monkeypatch.setattr(classes["SubagentResult"], "try_set_terminal", signal_terminal) + execution_id = executor.execute_async("Echo ok") + try: + assert model_returned.wait(5), "second model response was not received" + result = executor_module.get_background_task_result(execution_id) + assert _summarize_usage(result.token_usage_records) == {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15} + journal = RunJournal(run_id="parent-run", thread_id="parent-thread", event_store=MagicMock()) + _report_usage_records(journal, result) + if outcome == "cancelled": + executor_module.request_cancel_background_task(execution_id) + elif outcome == "completed": + gate["loop"].call_soon_threadsafe(gate["release"].set) + assert terminal.wait(10), "subagent did not reach a terminal status" + assert result.status.value == outcome + records = result.token_usage_records + assert len(records) == 2 + assert len({record["source_run_id"] for record in records}) == 2 + assert {record["model_name"] for record in records} == {"usage-test-model"} + expected_usage = {"input_tokens": 30, "output_tokens": 15, "total_tokens": 45} + assert _summarize_usage(records) == expected_usage + command = _task_result_command(tool_call_id="parent-call", status=outcome, usage=_summarize_usage(records)) + assert command.update["messages"][0].additional_kwargs["subagent_token_usage"] == expected_usage + # An interrupted poller may have delivered the earlier snapshot. + # The final delivery must add only the unseen response, once. + _report_usage_records(journal, result, final=True) + _report_usage_records(journal, result, final=True) + assert journal.get_completion_data()["total_tokens"] == 45 + assert journal.get_completion_data()["subagent_tokens"] == 45 + assert journal.get_completion_data()["token_usage_by_model"]["usage-test-model"] == expected_usage + finally: + if "release" in gate: + gate["loop"].call_soon_threadsafe(gate["release"].set) + executor_module.request_cancel_background_task(execution_id) + assert terminal.wait(5) + # Drain the production loop before stopping it and restoring the + # module fixture; no background task or admission slot may leak. + executor_module.run_on_isolated_subagent_loop(asyncio.sleep(0)).result(timeout=5) + executor_module.cleanup_background_task(execution_id) + executor_module._shutdown_isolated_subagent_loop() + + @pytest.mark.anyio + async def test_usage_snapshot_waits_for_stream_cleanup_despite_repeated_cancellation(self, classes, base_config, monkeypatch): + from langchain_core.outputs import ChatGeneration, LLMResult + + waiting = asyncio.Event() + closing = asyncio.Event() + release = asyncio.Event() + + class Stream: + def __init__(self, collector): + self.collector = collector + + def __aiter__(self): + return self + + async def __anext__(self): + waiting.set() + await asyncio.Event().wait() + + async def aclose(self): + closing.set() + await release.wait() + self.collector.on_llm_end( + LLMResult(generations=[[ChatGeneration(message=classes["AIMessage"](content="Done", usage_metadata={"input_tokens": 20, "output_tokens": 10, "total_tokens": 30}))]]), + run_id="response-during-cleanup", + ) + + graph = SimpleNamespace(astream=lambda state, config, **kwargs: Stream(config["callbacks"][0])) + executor = classes["SubagentExecutor"](config=base_config, tools=[]) + monkeypatch.setattr(executor, "_build_initial_state", AsyncMock(return_value=({}, [], None))) + monkeypatch.setattr(executor, "_create_agent", lambda *args, **kwargs: graph) + holder = classes["SubagentResult"](task_id="cleanup-usage", trace_id="trace", status=classes["SubagentStatus"].RUNNING) + running = asyncio.create_task(executor._aexecute_admitted("Task", holder)) + try: + await asyncio.wait_for(waiting.wait(), timeout=5) + running.cancel() + await asyncio.wait_for(closing.wait(), timeout=5) + running.cancel() + await asyncio.sleep(0) + assert not running.done() + assert not holder.token_usage_records + finally: + release.set() + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + + assert len(holder.token_usage_records) == 1 + assert holder.token_usage_records[0]["source_run_id"] == "response-during-cleanup" + assert holder.token_usage_records[0]["total_tokens"] == 30 + + @pytest.mark.anyio + async def test_cancellation_before_collector_creation_keeps_usage_absent(self, classes, base_config, monkeypatch): + building = asyncio.Event() + + async def build_state(task): + building.set() + await asyncio.Event().wait() + + executor = classes["SubagentExecutor"](config=base_config, tools=[]) + monkeypatch.setattr(executor, "_build_initial_state", build_state) + holder = classes["SubagentResult"](task_id="setup-cancelled", trace_id="trace", status=classes["SubagentStatus"].RUNNING) + running = asyncio.create_task(executor._aexecute_admitted("Task", holder)) + try: + await asyncio.wait_for(building.wait(), timeout=5) + finally: + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert holder.token_usage_records == [] + + class TestToolReceiptHarvest: """RFC #4651 PR2: the executor harvests the child's receipts at terminal."""