mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-19 11:06:18 +00:00
feat(runtime): persist tool-progress phase transitions (#5214)
* feat(runtime): persist tool-progress phase transitions Record bounded warn, block, and recover decisions for lead and task subagent runs while preserving event-loop isolation, fail-open behavior, and concurrent transition order. * fix(runtime): trust server-owned tool progress attribution * fix(runtime): centralize trusted audit attribution * fix(runtime): preserve complete tool progress audit state * docs: trim tool progress guidance to pass size check * fix(runtime): fence subagent audit recorder loop * docs(readme): sync tool-progress event coverage Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> --------- Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: 嗜鵼 <hy2010hy2010@qq.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
PeaceMaker-best
GitHub
parent
4ad55f598f
commit
14c9d44440
@ -814,8 +814,10 @@ finalization) keep the existing completion-data behavior: they receive the
|
||||
zero-delivery receipt but do not overwrite RunStore completion fields with an
|
||||
empty snapshot.
|
||||
|
||||
The same run event history records loop-detection decisions and deferred MCP
|
||||
tool promotions for both the lead agent and ordinary task subagents. Promotion
|
||||
When `tool_progress.enabled` is true, the same run event history also records
|
||||
result-quality guard phase changes. It records loop-detection decisions and
|
||||
deferred MCP tool promotions for both the lead agent and ordinary task
|
||||
subagents. Promotion
|
||||
events identify newly promoted deferred-tool names and whether routing metadata or
|
||||
`tool_search` selected them, without copying the search query, routing keywords,
|
||||
schemas, arguments, results, or catalog hash into the promotion event itself.
|
||||
|
||||
@ -562,7 +562,7 @@ logging:
|
||||
|
||||
Gateway 的运行历史还会为每次运行记录一条终止时的 `run.delivery` 回执，包括零产出与崩溃恢复的运行。正常执行时，该回执会在持久化终止运行状态之前写入。孤儿恢复会先原子地认领过期租约，再幂等地回填回执，因此过期的恢复扫描不会覆盖仍在运行的详细交付事实。在事件存储中断期间，回执持久化保持尽力而为。对 checkpoint 预检失败（或在等待前序 finalization 时被取消）的运行，保持既有的完成数据行为：它们会收到零交付回执，但不会用空快照覆盖 RunStore 的完成字段。
|
||||
|
||||
同一份运行事件历史还会为 lead agent 与普通 task subagent 记录 loop-detection 判定和延迟 MCP 工具晋升。晋升事件会标识新晋升的延迟工具名称，以及是路由元数据还是 `tool_search` 选中了它们，但不会把搜索查询、路由关键词、schema、参数、结果或目录哈希复制进晋升事件本身。
|
||||
当 `tool_progress.enabled` 为 true 时，同一份运行事件历史还会记录结果质量防护器的阶段变化。它也会为 lead agent 与普通 task subagent 记录 loop-detection 判定和延迟 MCP 工具晋升。晋升事件会标识新晋升的延迟工具名称，以及是路由元数据还是 `tool_search` 选中了它们，但不会把搜索查询、路由关键词、schema、参数、结果或目录哈希复制进晋升事件本身。
|
||||
|
||||
#### LangSmith 链路追踪
|
||||
|
||||
|
||||
@ -533,6 +533,11 @@ _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: frozenset[str] = (
|
||||
"is_internal",
|
||||
"authz_attributes",
|
||||
"channel_user_id",
|
||||
"is_subagent",
|
||||
"agent_id",
|
||||
"__run_loop_detection_recorder",
|
||||
"__run_tool_promotion_recorder",
|
||||
"__run_tool_progress_recorder",
|
||||
"langgraph_auth_user",
|
||||
"langgraph_auth_user_id",
|
||||
}
|
||||
@ -575,7 +580,7 @@ _CONTEXT_RUNTIME_ONLY_KEYS: frozenset[str] = frozenset({"github_token", "disable
|
||||
# credential that ``bash`` exports as ``GH_TOKEN``/``GITHUB_TOKEN``, and a copy
|
||||
# smuggled through ``body.config['configurable']`` would be written to the
|
||||
# checkpoint store.
|
||||
_INTERNAL_ONLY_CONTEXT_KEYS: frozenset[str] = _CONTEXT_INTERNAL_CALLER_KEYS | _CONTEXT_RUNTIME_ONLY_KEYS
|
||||
_INTERNAL_ONLY_CONTEXT_KEYS: frozenset[str] = _CONTEXT_INTERNAL_CALLER_KEYS | _CONTEXT_RUNTIME_ONLY_KEYS | _SERVER_OWNED_RUNTIME_CONTEXT_KEYS
|
||||
|
||||
|
||||
def strip_internal_context_keys(config: dict[str, Any]) -> None:
|
||||
|
||||
@ -77,9 +77,9 @@ through run-event or specialized APIs:
|
||||
| `middleware:{tag}` | `middleware` | `record_middleware()` |
|
||||
|
||||
Current middleware tags are `guardrail`, `loop_detection`,
|
||||
`safety_termination`, `skill_activation`, `skill_secrets`, and
|
||||
`tool_promotion`. The pattern is intentionally open so new middleware tags are
|
||||
additive. Because the full event type is limited to 32 characters and
|
||||
`safety_termination`, `skill_activation`, `skill_secrets`, `tool_promotion`,
|
||||
and `tool_progress`. The pattern is intentionally open so new middleware tags
|
||||
are additive. Because the full event type is limited to 32 characters and
|
||||
`middleware:` uses 11, a tag must contain 1-21 characters.
|
||||
|
||||
`middleware:loop_detection` records transitions into the warned state (first
|
||||
@ -95,6 +95,25 @@ subagent, and its agent id when applicable. Tool arguments, prompts, message
|
||||
content, tool results, and argument-derived hashes are not persisted in this
|
||||
event.
|
||||
|
||||
`middleware:tool_progress` records effective result-quality state transitions
|
||||
from `ToolProgressMiddleware`: `warn` for ACTIVE → WARNED, `block` for a
|
||||
transition to BLOCKED, `recover` for WARNED → ACTIVE after a useful result,
|
||||
and `reset` when a later agent invocation (for example, goal continuation)
|
||||
deliberately clears a prior WARNED/BLOCKED phase.
|
||||
The bounded `changes` object contains the tool name, source and destination
|
||||
phases, consecutive-problem count, normalized result status, error
|
||||
classification and recovery advice, effective count threshold (null when no
|
||||
count threshold fired), and
|
||||
subagent attribution. Tool
|
||||
arguments, prompts, message content, tool results, and content-derived hashes
|
||||
are never copied into this event. Producer-supplied metadata is projected onto
|
||||
fixed error/action vocabularies and a strict boolean-or-null before persistence,
|
||||
so a custom tool cannot smuggle arbitrary values through its result stamp.
|
||||
Recorder calls happen after the state lock is released, so a slow custom
|
||||
recorder cannot stall unrelated tool-state updates. Recorder failures are
|
||||
fail-open and do not change guard behavior. These events are emitted only when
|
||||
`tool_progress.enabled` is true (the default is false).
|
||||
|
||||
`middleware:tool_promotion` records deferred MCP schemas newly promoted for
|
||||
the active catalog. `changes.source` distinguishes automatic `routing_hint`
|
||||
promotion from an explicit `tool_search`; the remaining fields contain sorted
|
||||
@ -107,9 +126,10 @@ no names emit no event. Queries, routing keywords, catalog hashes, tool schemas
|
||||
and descriptions, arguments, and results are not copied into this middleware
|
||||
event. Other event types retain their existing payload contracts.
|
||||
|
||||
Ordinary task-tool subagents forward both loop-detection and tool-promotion
|
||||
appends to the parent run loop through dedicated recorder context keys. The
|
||||
loop-bound `RunJournal` itself never enters the isolated subagent loop.
|
||||
Ordinary task-tool subagents forward loop-detection, tool-promotion, and
|
||||
tool-progress appends to the parent run loop through dedicated recorder context
|
||||
keys. The loop-bound `RunJournal` itself never enters the isolated subagent
|
||||
loop.
|
||||
|
||||
### Opaque Run Outputs
|
||||
|
||||
@ -206,7 +226,8 @@ be used by new producers.
|
||||
- Nested non-JSON values in `run.end.content` have backend-dependent
|
||||
representations: memory retains Python values, while JSONL and database
|
||||
stores read them back as strings.
|
||||
- Durable batch subagent loop detection and deferred-tool promotion do not
|
||||
emit middleware events because those runs have no parent run journal.
|
||||
- Durable batch subagent loop detection, deferred-tool promotion, and
|
||||
tool-progress transitions do not emit middleware events because those runs
|
||||
have no parent run journal.
|
||||
- Journal attribution, token accounting, and external tracing metadata still
|
||||
depend on manual instrumentation at several LLM call sites.
|
||||
|
||||
@ -73,7 +73,7 @@ ordinary compaction or a model call.
|
||||
contract module.
|
||||
10. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations before tool execution; command classification is **defense-in-depth and audit, not a security boundary** (the sandbox is the isolation boundary). Command substitution is judged by *position*, not the presence of `$(`: **command position** (`$(curl url)`, `` `curl url` ``, the word after `|`/`&&`/`;`, an `eval`/`source` argument) executes fetched content and is blocked; **value position** (`x=$(curl url)`, `echo $(curl url)`, an argument, a `for` word list) only captures output and passes (#4611). So `_HIGH_RISK_COMMAND_POSITION_PATTERNS` is matched anchored against each sub-command from `_split_compound_command(split_pipes=True)`, never the whole string; pipe-spanning rules (`| sh`, `base64 -d | ...`) still use `_classify_command`'s whole-command Pass 1. `_COMMAND_POSITION_PREFIX` extends the anchor over leading assignments and exec wrappers (`FOO=1 $(curl url)`, `env`/`command`/`builtin`/`exec`/`nohup`/`time`/`sudo`/`doas`); its assignment branch requires whitespace before the substitution, which keeps `x=$(curl url)` in value position. Two contexts are deliberately **position-blind** (matched whole-command in Pass 1, since they execute their input anywhere, e.g. `xargs sh -c "$(curl url)"`): an `eval`/`source` argument, and an interpreter **code-string flag** — `-c` (shells, `python`), `-e` (`perl`/`ruby`/`node`), `-p` (`perl`/`node`), `-r` (`php`) — plus the here-string (`<<<`) reaching the same place via stdin. All three substitution spellings (`$(`, `<(`, `` ` ``) share one `_RISKY_SUBSTITUTION` opener. An unquoted newline splits like `;` (else `echo hi\n$(curl url)` evades the anchored rules). Heredoc bodies are data: `_split_compound_command` records headers (`<<EOF`, `<<-EOF`, `<<'EOF'`) and consumes their bodies verbatim, so a body line starting `$(curl url)` isn't promoted to a command position; `<<<` (here-string, needs look-ahead + look-behind) and a `<<` inside `$(( ))`/`(( ))` (bit shift, arithmetic depth tracked with the quote flags) must not open one. This is a heuristic, not shell parsing — an unterminated body consumes the rest of the string, an unclosed `((` only disables heredoc detection, and the failure direction is always toward *more* command positions, not fewer. Known gaps: process substitution outside `eval`/`source` (`. <(curl u)`) is undetected, and two-step forms (`x=$(curl u); eval "$x"`) need dataflow analysis. No config gate — appended unconditionally in `_build_runtime_middlewares`, for both lead and subagents.
|
||||
11. **ReadBeforeWriteMiddleware** - *(optional, `read_before_write.enabled`, default on)* Outermost write gate (#3857): `read_file` stamps a content hash on its ToolMessage; `write_file` (existing file, incl. append) and `str_replace` are blocked unless the newest mark for the path matches its current hash. Sits outside ToolProgress/ToolErrorHandling (a block consumes no ToolProgress slot); blocked results self-stamp `deerflow_tool_meta` and carry `deerflow_write_block` (`{path, tool}`). Marks live on messages, so summarization dropping the read invalidates the gate; writes never refresh marks. Gate check + execution are serialized per (thread, path); `"Error: ..."`-string sandboxes (AIO/E2B) fail open. It owns the composed call's sandbox authorization scope; `SandboxAuthorizationError` becomes an error ToolMessage. Its `wrap_model_call` swaps blocked calls' dead payload (`content`, `old_str`/`new_str`) for a deterministic placeholder in the model-bound request only (`elide_blocked_payloads`, `elide_min_chars`); state, receipts, and the journal keep the originals. Policy stays in the gate; the shared `tool_call_args` helper rewrites every arg surface together and every model-bound arg rewrite must use it.
|
||||
12. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its `wrap_tool_call` receives results already stamped with `deerflow_tool_meta`. Tracks per-(thread, tool) consecutive "no-new-info" calls across three error categories: (a) `recoverable_by_model=True` (no_results, not_found, permission, Jaccard-duplicate success): ACTIVE → WARNED (terminal — hint re-injected on each subsequent problem); (b) `recoverable_by_model=False, action≠stop` (rate_limited, transient): ACTIVE → WARNED → BLOCKED after `warn_escalation_count` more problems; (c) `recoverable_by_model=False, action=stop` (auth, config, internal): immediately BLOCKED on first occurrence. **Division of labor with LoopDetectionMiddleware:** ToolProgressMiddleware is a result-quality guard — fires after tool execution and blocks specific tools that stop producing new information; LoopDetectionMiddleware is a call-pattern guard — fires after the model responds and hard-stops the whole turn when the model repeatedly issues identical tool_calls. Both can inject HumanMessage hints in the same model call without conflict; neither reads the other's internal state.
|
||||
12. **ToolProgressMiddleware** - *(optional, if `tool_progress.enabled`)* State-machine-based stagnation guard (RFC #3177). Outer wrapper around ToolErrorHandlingMiddleware so its tool wrapper receives results already stamped with `deerflow_tool_meta`. Recoverable problems stay WARNED with hints; retryable non-recoverable problems escalate to BLOCKED; stop-category failures block immediately. It is independent of LoopDetection's turn-level call-pattern guard. Effective phase changes emit `middleware:tool_progress`, including `reset` when a later agent invocation (for example, goal continuation) clears WARNED/BLOCKED state. The state machine supplies the action and count threshold (`null` for category rules, recovery, and reset); recorder calls happen after releasing the state lock. Events include the actual sync/async hook plus bounded status, error, and recovery fields. Server-owned recorder keys determine subagent attribution. Arguments, content, prompts, and hashes are never persisted; recorder failures are fail-open. See [event semantics](../../../../../docs/RUN_EVENT_STREAM.md).
|
||||
13. **ToolReceiptMiddleware + ToolErrorHandlingMiddleware** - `ToolReceiptMiddleware` is *(optional, if `verification.receipts_enabled`, default on)*. It is the **outermost `wrap_tool_call` layer** — registered ahead of entries 9-12 — because Guardrail/SandboxAudit/ReadBeforeWrite/ToolProgress can short-circuit a call with their own ToolMessage (and SandboxAudit rebuilds medium-risk results); an inner receipt layer would silently gap the ledger on those results (ordering constraints in `deerflow.extensions.ordering`). Normal results still carry the `deerflow_tool_meta` status ToolErrorHandlingMiddleware stamps on the inner return path; short-circuit messages self-stamp meta or fall back to `message.status`. It stamps deterministic provenance (tool name, status, args/output hashes, byte count, timestamp) onto direct `ToolMessage` results and every matching `ToolMessage` carried in `Command.update.messages`, including delegated `task`, `present_file`, `view_image`, and `tool_search` results; before model calls it derives a hidden receipt ledger (display ids r1..rN) from message state, and when the 2,000-character budget is exceeded the newest receipts are retained in chronological order with their original ids plus an older-receipts omission marker. Rendering returns both the text and its retained receipt subset; every response that received a ledger carries only that exact server-owned subset, never omitted receipts. Snapshot validation accepts a strictly consecutive positive original-id range (for example `r24`–`r30`) rather than requiring `r1`, so subagent terminal citation verification resolves ids against evidence present in the citing turn even when later summarization drops and renumbers tool messages. Model-generated citation IDs are digit-bounded before integer conversion; oversized IDs are ignored as malformed input rather than raising through task write-back. Citation parsing deduplicates exact `(id, anchor)` pairs, not IDs alone, so repeated identical references stay compact while every distinct anchor claim is verified. Gateway strips delegated receipts/verdicts from external messages. `ToolErrorHandlingMiddleware` receives `AppConfig`, converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting, stamps every result with `deerflow_tool_meta` (status / error_type / recoverable_by_model / recommended_next_action / source) via `tool_result_meta.normalize_tool_result`, stamps structured metadata for task exception wrappers, and stamps skill-read metadata for downstream durable-context capture. Task tool result text is generated from the same status/result/error inputs as the structured metadata so callers do not hand-write a second protocol string.
|
||||
|
||||
Authorization identity is independent of enforcement. Gateway strips client identity overrides: only the server auth source sets `is_internal`, and only authenticated IM `body.context` supplies `channel_user_id` (never `body.config`). `build_principal_from_context` applies role defaults, strict provenance, and copied attributes; RBAC rejects unknown defaults. Delegation and `GuardrailMiddleware` share this identity. Layer 1 precedes deferred assembly across agent paths and its provider is reused for Layer 2; framework skill/memory ordering stays stable. Trusted `DeerFlowClient.stream()` accepts identity overrides. Its graph key always includes effective storage `user_id` and, when enforced, the full Principal; nested attributes are copied so mutation cannot hide stale cache state.
|
||||
@ -99,7 +99,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
|
||||
22. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses); captures the runtime-resolved user so standalone LangGraph Server reads and writes stay in the same bucket
|
||||
23. **ViewImageMiddleware** - *(optional, if the model supports vision)* Appends a hidden HumanMessage with base64 image data, identified by a reserved ID prefix plus a server-owned metadata marker, to `ModelRequest.messages` in `wrap_model_call` / `awrap_model_call`. The payload lives only in that request and is never returned as a state update, so no checkpoint carries it and an interrupted run cannot strand it in history; state keeps only the lightweight `viewed_images` metadata. It owns that context and rebuilds it per call: its own message is swept out of the request first — a thread checkpointed by the earlier `before_model`/`after_model` pair (which wrote the payload into state and took it back out with `RemoveMessage`) can carry one that reached state but was never removed, and leaving it in would resend that base64 in every later request for the life of the thread — then a freshly built one is appended when warranted. The sweep requires both the reserved ID prefix and the server-owned marker, so a client cannot get its own message dropped; unmarked leftovers predating the marker are left in place and merely not duplicated
|
||||
24. **McpRoutingMiddleware** - Auto-promotes deferred schemas matching the latest real user message before `DeferredToolFilterMiddleware`, without executing tools. New names emit `middleware:tool_promotion` with `source=routing_hint`; repeat passes emit nothing
|
||||
25. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, returns the original `Command`, and omits private payloads
|
||||
25. **DeferredToolPromotionAuditMiddleware** - Observes final `tool_search` `Command`s; keep it outer of `SkillToolPolicyMiddleware` so denied names are excluded. It atomically claims new names per lead run or subagent execution to dedupe parallel searches, derives subagent attribution from the server-installed recorder, returns the original `Command`, and omits private payloads
|
||||
26. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` or `McpRoutingMiddleware` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)
|
||||
27. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives. The subagent builder places its date-only context middleware immediately before this coalescer, so the built-in subagent prompt and hidden date reminder still reach providers as one leading system block
|
||||
28. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess ordinary `task` tool calls to enforce both the per-response concurrency limit (`max_concurrent_subagents`, resolved against startup `subagent_runtime.max_running` and the 1-64 safety range before construction) and the per-run total delegation cap (`max_total_subagents` runtime override or `subagents.max_total_per_run`, default 6, clamped to 1-50). The total cap counts current-run entries in the durable delegation ledger (entries are tagged with `run_id` when captured), so repeated planning checkpoints in one run cannot keep launching legal-sized batches indefinitely, while later user turns in the same thread get a fresh run budget. Explicit durable `batch_task` calls are a separate mode with persisted total/live/running limits and are not rewritten into ordinary ledger entries. If the ordinary cap is exhausted, the middleware strips remaining `task` calls, forces `finish_reason="stop"`, and appends a visible limit note so the run can synthesize existing results instead of ending with an empty tool-call response.
|
||||
|
||||
@ -1,4 +1,29 @@
|
||||
"""Private runtime-context keys for narrowly scoped middleware audit recorders."""
|
||||
"""Private runtime-context helpers for narrowly scoped audit recorders."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
LOOP_DETECTION_RECORDER_CONTEXT_KEY = "__run_loop_detection_recorder"
|
||||
TOOL_PROMOTION_RECORDER_CONTEXT_KEY = "__run_tool_promotion_recorder"
|
||||
TOOL_PROGRESS_RECORDER_CONTEXT_KEY = "__run_tool_progress_recorder"
|
||||
|
||||
|
||||
def resolve_audit_recorder(
|
||||
context: object,
|
||||
*,
|
||||
recorder_key: str,
|
||||
) -> tuple[Any | None, bool, str | None]:
|
||||
"""Resolve a recorder and trusted lead/subagent attribution.
|
||||
|
||||
Ordinary lead runs own ``__run_journal``. Task-tool subagents receive only
|
||||
a server-installed narrow recorder, so its presence is the authority for
|
||||
subagent attribution; caller-supplied ``is_subagent`` is never consulted.
|
||||
"""
|
||||
if not isinstance(context, dict):
|
||||
return None, False, None
|
||||
|
||||
recorder = context.get(recorder_key)
|
||||
if recorder is not None:
|
||||
agent_id = context.get("agent_id")
|
||||
return recorder, True, agent_id if isinstance(agent_id, str) else None
|
||||
|
||||
return context.get("__run_journal"), False, None
|
||||
|
||||
@ -79,7 +79,10 @@ from langchain_core.messages import HumanMessage
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
from deerflow.agents.middlewares._bounded_dict import BoundedDict
|
||||
from deerflow.agents.middlewares.audit_context import LOOP_DETECTION_RECORDER_CONTEXT_KEY
|
||||
from deerflow.agents.middlewares.audit_context import (
|
||||
LOOP_DETECTION_RECORDER_CONTEXT_KEY,
|
||||
resolve_audit_recorder,
|
||||
)
|
||||
from deerflow.runtime.events.catalog import MIDDLEWARE_LOOP_DETECTION_TAG
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -731,13 +734,10 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
|
||||
runtime: Runtime,
|
||||
) -> None:
|
||||
"""Persist a loop-detection transition without sensitive tool data."""
|
||||
context = getattr(runtime, "context", None)
|
||||
is_subagent = isinstance(context, dict) and context.get("is_subagent") is True
|
||||
recorder = context.get(LOOP_DETECTION_RECORDER_CONTEXT_KEY) if isinstance(context, dict) else None
|
||||
if recorder is None and isinstance(context, dict):
|
||||
# Lead-agent runs expose the ordinary RunJournal. Native task-tool
|
||||
# subagents receive only the narrow, loop-safe recorder key above.
|
||||
recorder = context.get("__run_journal")
|
||||
recorder, is_subagent, agent_id = resolve_audit_recorder(
|
||||
getattr(runtime, "context", None),
|
||||
recorder_key=LOOP_DETECTION_RECORDER_CONTEXT_KEY,
|
||||
)
|
||||
if recorder is None:
|
||||
return
|
||||
|
||||
@ -749,7 +749,7 @@ class LoopDetectionMiddleware(AgentMiddleware[AgentState]):
|
||||
action=decision.action,
|
||||
changes={
|
||||
"is_subagent": is_subagent,
|
||||
"agent_id": context.get("agent_id") if is_subagent else None,
|
||||
"agent_id": agent_id,
|
||||
"detection_layer": decision.detection_layer,
|
||||
"tool_names": list(decision.tool_names),
|
||||
"count": decision.count,
|
||||
|
||||
@ -64,7 +64,19 @@ from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY, ToolResultMeta
|
||||
from deerflow.agents.middlewares.audit_context import (
|
||||
TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
|
||||
resolve_audit_recorder,
|
||||
)
|
||||
from deerflow.agents.middlewares.tool_result_meta import (
|
||||
PROGRESS_GUARD_ERROR_TYPE,
|
||||
TOOL_META_KEY,
|
||||
TOOL_RESULT_ERROR_TYPES,
|
||||
TOOL_RESULT_NEXT_ACTIONS,
|
||||
TOOL_RESULT_STATUSES,
|
||||
ToolResultMeta,
|
||||
)
|
||||
from deerflow.runtime.events.catalog import MIDDLEWARE_TOOL_PROGRESS_TAG
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from deerflow.config.tool_progress_config import ToolProgressConfig
|
||||
@ -76,6 +88,28 @@ _MAX_PENDING_PER_RUN = 3
|
||||
_MAX_CONTENT_FOR_WORDSET = 8192
|
||||
|
||||
|
||||
def _audit_error_type(value: object) -> str | None:
|
||||
"""Project an untrusted tool stamp onto the bounded audit vocabulary."""
|
||||
if value is None:
|
||||
return None
|
||||
return value if type(value) is str and value in TOOL_RESULT_ERROR_TYPES else "unknown"
|
||||
|
||||
|
||||
def _audit_next_action(value: object) -> str:
|
||||
"""Return only framework-defined recovery actions to persistence."""
|
||||
return value if type(value) is str and value in TOOL_RESULT_NEXT_ACTIONS else "unknown"
|
||||
|
||||
|
||||
def _audit_status(value: object) -> str:
|
||||
"""Project the producer status onto the canonical result vocabulary."""
|
||||
return value if type(value) is str and value in TOOL_RESULT_STATUSES else "unknown"
|
||||
|
||||
|
||||
def _audit_recoverable(value: object) -> bool | None:
|
||||
"""Reject truthy non-booleans from producer-supplied tool metadata."""
|
||||
return value if type(value) is bool else None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# State data structures
|
||||
|
||||
@ -93,6 +127,14 @@ class ToolPhaseState:
|
||||
recent_word_sets: tuple[frozenset[str], ...] = field(default_factory=tuple)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ToolPhaseTransition:
|
||||
"""The exact state-machine rule that produced a durable phase change."""
|
||||
|
||||
action: Literal["warn", "block", "recover", "reset"]
|
||||
threshold: int | None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Content helpers
|
||||
|
||||
@ -231,11 +273,10 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
self._exempt_tools: set[str] = exempt_tools if exempt_tools is not None else {"ask_clarification", "write_todos", "present_files", "task"}
|
||||
self._max_tracked_threads = max_tracked_threads
|
||||
|
||||
# threading.Lock (not asyncio.Lock): critical sections are short in-memory dict
|
||||
# ops with no I/O, so event-loop stall risk is negligible. asyncio.Lock would
|
||||
# not protect the sync wrap_tool_call path used by subagent executor thread
|
||||
# pools — two separate locks would be required instead. This matches the
|
||||
# existing LoopDetectionMiddleware pattern; see module docstring for details.
|
||||
# threading.Lock (not asyncio.Lock) also protects embedded callers that use the
|
||||
# synchronous wrapper from multiple threads. Recorder callbacks are deliberately
|
||||
# invoked after this state lock is released; observability must not stall tool
|
||||
# state updates. This matches LoopDetectionMiddleware's publication convention.
|
||||
self._lock = threading.Lock()
|
||||
# LRU-evicting store: thread_id → {tool_name → ToolPhaseState}
|
||||
self._phase_states: OrderedDict[str, dict[str, ToolPhaseState]] = OrderedDict()
|
||||
@ -270,6 +311,52 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
def _pending_key(self, runtime: Runtime) -> tuple[str, str]:
|
||||
return self._thread_id(runtime), self._run_id(runtime)
|
||||
|
||||
def _record_phase_transition(
|
||||
self,
|
||||
*,
|
||||
runtime: Runtime,
|
||||
tool_name: str,
|
||||
state: ToolPhaseState,
|
||||
new_state: ToolPhaseState,
|
||||
meta: ToolResultMeta | None,
|
||||
hook: Literal["wrap_tool_call", "awrap_tool_call", "before_agent", "abefore_agent"],
|
||||
transition: ToolPhaseTransition,
|
||||
) -> None:
|
||||
"""Persist one effective transition without copying tool content."""
|
||||
recorder, is_subagent, agent_id = resolve_audit_recorder(
|
||||
getattr(runtime, "context", None),
|
||||
recorder_key=TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
|
||||
)
|
||||
if recorder is None:
|
||||
return
|
||||
|
||||
try:
|
||||
recorder.record_middleware(
|
||||
tag=MIDDLEWARE_TOOL_PROGRESS_TAG,
|
||||
name=type(self).__name__,
|
||||
hook=hook,
|
||||
action=transition.action,
|
||||
changes={
|
||||
"is_subagent": is_subagent,
|
||||
"agent_id": agent_id,
|
||||
"tool_name": tool_name,
|
||||
"from_phase": state.phase,
|
||||
"to_phase": new_state.phase,
|
||||
"consecutive_problems": new_state.consecutive_problems,
|
||||
"status": _audit_status(meta.status) if meta is not None else None,
|
||||
"error_type": _audit_error_type(meta.error_type) if meta is not None else None,
|
||||
"recoverable_by_model": _audit_recoverable(meta.recoverable_by_model) if meta is not None else None,
|
||||
"recommended_next_action": _audit_next_action(meta.recommended_next_action) if meta is not None else None,
|
||||
"threshold": transition.threshold,
|
||||
},
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
# Observability must never alter the progress guard or agent run.
|
||||
logger.warning(
|
||||
"Failed to record middleware:tool_progress event",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# State store (caller holds lock)
|
||||
|
||||
@ -308,7 +395,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
additional_kwargs={
|
||||
TOOL_META_KEY: {
|
||||
"status": "error",
|
||||
"error_type": "blocked_by_progress_guard",
|
||||
"error_type": PROGRESS_GUARD_ERROR_TYPE,
|
||||
"recoverable_by_model": True,
|
||||
"recommended_next_action": "summarize",
|
||||
"source": "progress_middleware",
|
||||
@ -322,6 +409,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
tool_name: str,
|
||||
runtime: Runtime,
|
||||
tool_call_id: str,
|
||||
*,
|
||||
hook: Literal["wrap_tool_call", "awrap_tool_call"],
|
||||
) -> ToolMessage | Command:
|
||||
"""Update the state machine from a tool result; queue hints if warranted."""
|
||||
message = _result_tool_message(result, tool_call_id)
|
||||
@ -337,10 +426,24 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
return result
|
||||
content = _message_content_str(message)
|
||||
thread_id = self._thread_id(runtime)
|
||||
phase_change: tuple[ToolPhaseState, ToolPhaseState, ToolPhaseTransition] | None = None
|
||||
with self._lock:
|
||||
state = self._get_state(thread_id, tool_name)
|
||||
new_state, hint = self._assess_and_transition(state, meta, content)
|
||||
new_state, hint, transition = self._assess_and_transition(state, meta, content)
|
||||
self._set_state(thread_id, tool_name, new_state)
|
||||
if transition is not None:
|
||||
phase_change = (state, new_state, transition)
|
||||
if phase_change is not None:
|
||||
old_state, changed_state, phase_transition = phase_change
|
||||
self._record_phase_transition(
|
||||
runtime=runtime,
|
||||
tool_name=tool_name,
|
||||
state=old_state,
|
||||
new_state=changed_state,
|
||||
meta=meta,
|
||||
hook=hook,
|
||||
transition=phase_transition,
|
||||
)
|
||||
if new_state.phase != state.phase:
|
||||
if new_state.phase == "blocked":
|
||||
logger.warning(
|
||||
@ -374,8 +477,8 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
state: ToolPhaseState,
|
||||
meta: ToolResultMeta,
|
||||
content: str,
|
||||
) -> tuple[ToolPhaseState, str | None]:
|
||||
"""Return (new_state, hint_text_or_None).
|
||||
) -> tuple[ToolPhaseState, str | None, ToolPhaseTransition | None]:
|
||||
"""Return the new state, optional hint, and rule that changed phase.
|
||||
|
||||
The outer wrap_tool_call gate intercepts already-blocked states before
|
||||
the handler is called, so this function is normally reached only for
|
||||
@ -389,7 +492,7 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
# to make concurrent-race semantics well-defined and prevent a
|
||||
# recoverable-error result from silently demoting the phase back to warned.)
|
||||
if state.phase == "blocked":
|
||||
return state, None
|
||||
return state, None, None
|
||||
|
||||
# Count this call as a problem before branching so all exit paths leave
|
||||
# consecutive_problems in a consistent state (never 0 when the tool has failed).
|
||||
@ -397,12 +500,16 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
|
||||
# Immediately block on unrecoverable stop signals (auth, config, internal).
|
||||
if not meta.recoverable_by_model and meta.recommended_next_action == "stop":
|
||||
return replace(
|
||||
state,
|
||||
phase="blocked",
|
||||
consecutive_problems=new_count,
|
||||
block_reason=_block_reason(meta),
|
||||
), None
|
||||
return (
|
||||
replace(
|
||||
state,
|
||||
phase="blocked",
|
||||
consecutive_problems=new_count,
|
||||
block_reason=_block_reason(meta),
|
||||
),
|
||||
None,
|
||||
ToolPhaseTransition(action="block", threshold=None),
|
||||
)
|
||||
|
||||
# Compute word_set only for success results: error/partial_success are problems by
|
||||
# definition and never reach the Jaccard check, so the O(n) regex is wasted on them.
|
||||
@ -412,9 +519,11 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
if not is_problem:
|
||||
# Good result: reset consecutive count, return to active.
|
||||
new_recent = (*state.recent_word_sets, ws)[-3:]
|
||||
return replace(state, consecutive_problems=0, phase="active", recent_word_sets=new_recent), None
|
||||
transition = ToolPhaseTransition(action="recover", threshold=None) if state.phase == "warned" else None
|
||||
return replace(state, consecutive_problems=0, phase="active", recent_word_sets=new_recent), None, transition
|
||||
|
||||
hint: str | None = None
|
||||
transition: ToolPhaseTransition | None = None
|
||||
|
||||
if new_count >= self._stagnation_threshold + self._warn_escalation:
|
||||
if meta.recoverable_by_model:
|
||||
@ -422,17 +531,28 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
# BLOCKED would prevent a legitimate retry with different parameters.
|
||||
hint = _format_hint(meta)
|
||||
new_state = replace(state, consecutive_problems=new_count, phase="warned")
|
||||
if state.phase != "warned":
|
||||
transition = ToolPhaseTransition(
|
||||
action="warn",
|
||||
threshold=self._stagnation_threshold + self._warn_escalation,
|
||||
)
|
||||
else:
|
||||
# Model cannot fix this by retrying — block the tool.
|
||||
reason = _block_reason(meta)
|
||||
new_state = replace(state, consecutive_problems=new_count, phase="blocked", block_reason=reason)
|
||||
transition = ToolPhaseTransition(
|
||||
action="block",
|
||||
threshold=self._stagnation_threshold + self._warn_escalation,
|
||||
)
|
||||
elif new_count >= self._stagnation_threshold:
|
||||
hint = _format_hint(meta)
|
||||
new_state = replace(state, consecutive_problems=new_count, phase="warned")
|
||||
if state.phase != "warned":
|
||||
transition = ToolPhaseTransition(action="warn", threshold=self._stagnation_threshold)
|
||||
else:
|
||||
new_state = replace(state, consecutive_problems=new_count)
|
||||
|
||||
return new_state, hint
|
||||
return new_state, hint, transition
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Pending queue helpers
|
||||
@ -462,7 +582,12 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
if key[0] == thread_id and key[1] != current_run:
|
||||
del self._pending[key]
|
||||
|
||||
def _reset_run_states(self, runtime: Runtime) -> None:
|
||||
def _reset_run_states(
|
||||
self,
|
||||
runtime: Runtime,
|
||||
*,
|
||||
hook: Literal["before_agent", "abefore_agent"],
|
||||
) -> None:
|
||||
"""Reset all per-run tool state for the thread at the start of a new agent run.
|
||||
|
||||
Every tool's consecutive_problems counter and recent_word_sets Jaccard window are
|
||||
@ -483,18 +608,33 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
middlewares therefore guard different failure modes and use different lifetimes.
|
||||
"""
|
||||
thread_id = self._thread_id(runtime)
|
||||
transitions: list[tuple[str, ToolPhaseState, ToolPhaseState]] = []
|
||||
with self._lock:
|
||||
thread_tools = self._phase_states.get(thread_id)
|
||||
if thread_tools is None:
|
||||
return
|
||||
for tool_name, tool_state in list(thread_tools.items()):
|
||||
thread_tools[tool_name] = replace(
|
||||
new_state = replace(
|
||||
tool_state,
|
||||
phase="active",
|
||||
consecutive_problems=0,
|
||||
block_reason=None,
|
||||
recent_word_sets=(),
|
||||
)
|
||||
thread_tools[tool_name] = new_state
|
||||
if tool_state.phase != new_state.phase:
|
||||
transitions.append((tool_name, tool_state, new_state))
|
||||
|
||||
for tool_name, tool_state, new_state in transitions:
|
||||
self._record_phase_transition(
|
||||
runtime=runtime,
|
||||
tool_name=tool_name,
|
||||
state=tool_state,
|
||||
new_state=new_state,
|
||||
meta=None,
|
||||
hook=hook,
|
||||
transition=ToolPhaseTransition(action="reset", threshold=None),
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# wrap_tool_call
|
||||
@ -520,7 +660,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
block_reason,
|
||||
)
|
||||
return self._make_blocked_message(request, tool_name, block_reason)
|
||||
return self._update_state_from_result(handler(request), tool_name, runtime, str(request.tool_call.get("id") or ""))
|
||||
return self._update_state_from_result(
|
||||
handler(request),
|
||||
tool_name,
|
||||
runtime,
|
||||
str(request.tool_call.get("id") or ""),
|
||||
hook="wrap_tool_call",
|
||||
)
|
||||
|
||||
@override
|
||||
async def awrap_tool_call(
|
||||
@ -543,7 +689,13 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
block_reason,
|
||||
)
|
||||
return self._make_blocked_message(request, tool_name, block_reason)
|
||||
return self._update_state_from_result(await handler(request), tool_name, runtime, str(request.tool_call.get("id") or ""))
|
||||
return self._update_state_from_result(
|
||||
await handler(request),
|
||||
tool_name,
|
||||
runtime,
|
||||
str(request.tool_call.get("id") or ""),
|
||||
hook="awrap_tool_call",
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# wrap_model_call: drain pending hints and inject before model sees messages
|
||||
@ -586,11 +738,11 @@ class ToolProgressMiddleware(AgentMiddleware[AgentState]):
|
||||
@override
|
||||
def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
|
||||
self._clear_stale_pending(runtime)
|
||||
self._reset_run_states(runtime)
|
||||
self._reset_run_states(runtime, hook="before_agent")
|
||||
return None
|
||||
|
||||
@override
|
||||
async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict | None:
|
||||
self._clear_stale_pending(runtime)
|
||||
self._reset_run_states(runtime)
|
||||
self._reset_run_states(runtime, hook="abefore_agent")
|
||||
return None
|
||||
|
||||
@ -13,7 +13,10 @@ from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.runtime import Runtime
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.middlewares.audit_context import TOOL_PROMOTION_RECORDER_CONTEXT_KEY
|
||||
from deerflow.agents.middlewares.audit_context import (
|
||||
TOOL_PROMOTION_RECORDER_CONTEXT_KEY,
|
||||
resolve_audit_recorder,
|
||||
)
|
||||
from deerflow.runtime.events.catalog import MIDDLEWARE_TOOL_PROMOTION_TAG
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -34,15 +37,10 @@ def record_tool_promotion(
|
||||
if not names:
|
||||
return
|
||||
|
||||
context = getattr(runtime, "context", None)
|
||||
if not isinstance(context, dict):
|
||||
return
|
||||
is_subagent = context.get("is_subagent") is True
|
||||
recorder = context.get(TOOL_PROMOTION_RECORDER_CONTEXT_KEY)
|
||||
if recorder is None:
|
||||
# Lead runs own a RunJournal. Ordinary task-tool subagents receive only
|
||||
# the narrow, loop-safe recorder key above.
|
||||
recorder = context.get("__run_journal")
|
||||
recorder, is_subagent, agent_id = resolve_audit_recorder(
|
||||
getattr(runtime, "context", None),
|
||||
recorder_key=TOOL_PROMOTION_RECORDER_CONTEXT_KEY,
|
||||
)
|
||||
if recorder is None:
|
||||
return
|
||||
|
||||
@ -62,7 +60,7 @@ def record_tool_promotion(
|
||||
"tool_names": names,
|
||||
"count": len(names),
|
||||
"is_subagent": is_subagent,
|
||||
"agent_id": context.get("agent_id") if is_subagent else None,
|
||||
"agent_id": agent_id,
|
||||
},
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
|
||||
@ -10,12 +10,18 @@ from __future__ import annotations
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
from typing import Literal, get_args
|
||||
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langgraph.types import Command
|
||||
|
||||
TOOL_META_KEY = "deerflow_tool_meta"
|
||||
PROGRESS_GUARD_ERROR_TYPE = "blocked_by_progress_guard"
|
||||
|
||||
ToolResultStatus = Literal["success", "error", "partial_success"]
|
||||
RecommendedNextAction = Literal["continue", "rewrite_query", "try_alternative", "summarize", "stop"]
|
||||
TOOL_RESULT_STATUSES = frozenset(get_args(ToolResultStatus)) | {"unknown"}
|
||||
TOOL_RESULT_NEXT_ACTIONS = frozenset(get_args(RecommendedNextAction)) | {"unknown"}
|
||||
|
||||
_ERROR_PREFIX = "Error:"
|
||||
_PARTIAL_MARKERS = (
|
||||
@ -33,10 +39,10 @@ _PARTIAL_MARKERS = (
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ToolResultMeta:
|
||||
status: Literal["success", "error", "partial_success"]
|
||||
status: ToolResultStatus
|
||||
error_type: str | None
|
||||
recoverable_by_model: bool
|
||||
recommended_next_action: Literal["continue", "rewrite_query", "try_alternative", "summarize", "stop"]
|
||||
recommended_next_action: RecommendedNextAction
|
||||
source: Literal["exception", "tool_return", "content_analysis", "progress_middleware"]
|
||||
|
||||
|
||||
@ -97,6 +103,7 @@ _PAGE_CONTENT_TOOL_NAMES: frozenset[str] = frozenset({"web_fetch"})
|
||||
# _ERROR_RULES already declares. Derived rather than duplicated so a shell can
|
||||
# never drift from the recoverable/next-action contract of its own category.
|
||||
_ATTRS_BY_ERROR_TYPE: dict[str, dict[str, object]] = {str(attrs["error_type"]): attrs for _keywords, attrs in _ERROR_RULES}
|
||||
TOOL_RESULT_ERROR_TYPES = frozenset(_ATTRS_BY_ERROR_TYPE) | {"unknown", PROGRESS_GUARD_ERROR_TYPE}
|
||||
|
||||
# Reason phrases (RFC 9110 §15 plus the wording real servers ship) mapped onto the
|
||||
# error_type they already have in _ERROR_RULES. Restricted to the statuses a fetch
|
||||
|
||||
@ -135,6 +135,24 @@ their per-execution parent-loop proxy, preserving separate events when two
|
||||
different delegated agents promote the same tool. The active catalog is fixed
|
||||
for one graph execution, so the claim needs no persisted catalog hash.
|
||||
|
||||
**Tool-progress phase events** (`agents/middlewares/tool_progress_middleware.py`):
|
||||
effective ACTIVE → WARNED, WARNED/ACTIVE → BLOCKED, WARNED → ACTIVE recovery,
|
||||
and later-invocation WARNED/BLOCKED → ACTIVE resets append
|
||||
`middleware:tool_progress` through `RunJournal`. Recorder calls happen after
|
||||
the middleware releases its state lock, matching LoopDetectionMiddleware, so a
|
||||
slow recorder cannot stall tool-state updates. Cross-thread middleware
|
||||
producers (currently slash-skill activation via `asyncio.to_thread`) schedule
|
||||
journal mutation directly onto its owning event loop; they never mutate or
|
||||
flush `RunJournal._buffer` from the worker thread. The task-tool subagent proxy
|
||||
rejects a loop that differs from the journal owner, so its close fence always
|
||||
drains the only scheduling hop.
|
||||
The persisted projection accepts
|
||||
only framework-defined error/action values and strict booleans (using null for
|
||||
invalid values) from the producer-supplied tool stamp; tool content, args,
|
||||
prompts, and derived hashes do not enter the event. Ordinary task-tool subagents use the narrow parent-loop
|
||||
recorder proxy, never the journal itself; durable batch runs have no parent
|
||||
journal and emit no such event. Recorder failures are fail-open.
|
||||
|
||||
**JSONL record boundaries** (`runtime/events/store/jsonl.py`): thread reads,
|
||||
run reads, and sequence recovery split on physical newlines. Do not use
|
||||
`str.splitlines()`: U+0085/U+2028/U+2029 inside valid JSON strings must remain
|
||||
|
||||
@ -82,6 +82,7 @@ MIDDLEWARE_SAFETY_TERMINATION_TAG = "safety_termination"
|
||||
MIDDLEWARE_SKILL_ACTIVATION_TAG = "skill_activation"
|
||||
MIDDLEWARE_SKILL_SECRETS_TAG = "skill_secrets"
|
||||
MIDDLEWARE_TOOL_PROMOTION_TAG = "tool_promotion"
|
||||
MIDDLEWARE_TOOL_PROGRESS_TAG = "tool_progress"
|
||||
MIDDLEWARE_EVENT_TAGS = (
|
||||
MIDDLEWARE_GUARDRAIL_TAG,
|
||||
MIDDLEWARE_LOOP_DETECTION_TAG,
|
||||
@ -89,6 +90,7 @@ MIDDLEWARE_EVENT_TAGS = (
|
||||
MIDDLEWARE_SKILL_ACTIVATION_TAG,
|
||||
MIDDLEWARE_SKILL_SECRETS_TAG,
|
||||
MIDDLEWARE_TOOL_PROMOTION_TAG,
|
||||
MIDDLEWARE_TOOL_PROGRESS_TAG,
|
||||
)
|
||||
|
||||
JOURNAL_RUN_EVENT_DEFINITIONS = (
|
||||
|
||||
@ -25,6 +25,7 @@ from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from functools import partial
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
from uuid import UUID
|
||||
|
||||
@ -250,11 +251,16 @@ class RunJournal(BaseCallbackHandler):
|
||||
self._flush_threshold = flush_threshold
|
||||
self._progress_reporter = progress_reporter
|
||||
self._progress_flush_interval = progress_flush_interval
|
||||
try:
|
||||
self._owner_loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
self._owner_loop = None
|
||||
|
||||
# Write buffer
|
||||
self._buffer: list[dict] = []
|
||||
self._pending_llm_response: _PendingLlmResponse | None = None
|
||||
self._pending_flush_tasks: set[asyncio.Task[None]] = set()
|
||||
self._explicit_flush_in_progress = False
|
||||
self._pending_progress_task: asyncio.Task[None] | None = None
|
||||
self._pending_progress_delayed = False
|
||||
self._progress_dirty = False
|
||||
@ -829,7 +835,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
return
|
||||
# Skip if a flush is already in flight — avoids concurrent writes
|
||||
# to the same SQLite file from multiple fire-and-forget tasks.
|
||||
if self._pending_flush_tasks:
|
||||
if self._pending_flush_tasks or self._explicit_flush_in_progress:
|
||||
return
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
@ -995,8 +1001,31 @@ class RunJournal(BaseCallbackHandler):
|
||||
action: Specific action performed (e.g., "generate_title").
|
||||
changes: Dict describing the state changes made.
|
||||
"""
|
||||
event_type = MIDDLEWARE_EVENT_PATTERN.event_type(tag)
|
||||
owner_loop = self._owner_loop
|
||||
try:
|
||||
running_loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
running_loop = None
|
||||
if owner_loop is not None and running_loop is not owner_loop:
|
||||
if owner_loop.is_closed() or not owner_loop.is_running():
|
||||
logger.warning("Dropping cross-thread middleware event after run loop shutdown")
|
||||
return
|
||||
try:
|
||||
owner_loop.call_soon_threadsafe(
|
||||
partial(
|
||||
self._put,
|
||||
event_type=event_type,
|
||||
category=MIDDLEWARE_EVENT_PATTERN.category,
|
||||
content={"name": name, "hook": hook, "action": action, "changes": dict(changes)},
|
||||
)
|
||||
)
|
||||
except RuntimeError:
|
||||
logger.warning("Dropping cross-thread middleware event after run loop shutdown")
|
||||
return
|
||||
|
||||
self._put(
|
||||
event_type=MIDDLEWARE_EVENT_PATTERN.event_type(tag),
|
||||
event_type=event_type,
|
||||
category=MIDDLEWARE_EVENT_PATTERN.category,
|
||||
content={"name": name, "hook": hook, "action": action, "changes": changes},
|
||||
)
|
||||
@ -1068,39 +1097,43 @@ class RunJournal(BaseCallbackHandler):
|
||||
"""Force flush remaining buffer. Called in worker's finally block."""
|
||||
if self._closed:
|
||||
return
|
||||
self._commit_pending_llm_response()
|
||||
if self._pending_flush_tasks:
|
||||
await asyncio.gather(*tuple(self._pending_flush_tasks), return_exceptions=True)
|
||||
while self._pending_progress_task is not None:
|
||||
pending_progress_task = self._pending_progress_task
|
||||
if pending_progress_task.done():
|
||||
if self._pending_progress_task is pending_progress_task:
|
||||
self._pending_progress_task = None
|
||||
break
|
||||
if self._pending_progress_delayed:
|
||||
pending_progress_task.cancel()
|
||||
self._explicit_flush_in_progress = True
|
||||
try:
|
||||
self._commit_pending_llm_response()
|
||||
if self._pending_flush_tasks:
|
||||
await asyncio.gather(*tuple(self._pending_flush_tasks), return_exceptions=True)
|
||||
while self._pending_progress_task is not None:
|
||||
pending_progress_task = self._pending_progress_task
|
||||
if pending_progress_task.done():
|
||||
if self._pending_progress_task is pending_progress_task:
|
||||
self._pending_progress_task = None
|
||||
break
|
||||
if self._pending_progress_delayed:
|
||||
pending_progress_task.cancel()
|
||||
await asyncio.gather(pending_progress_task, return_exceptions=True)
|
||||
if self._pending_progress_task is pending_progress_task:
|
||||
self._pending_progress_task = None
|
||||
self._progress_dirty = False
|
||||
self._pending_progress_delayed = False
|
||||
break
|
||||
await asyncio.gather(pending_progress_task, return_exceptions=True)
|
||||
if self._pending_progress_task is pending_progress_task:
|
||||
self._pending_progress_task = None
|
||||
self._progress_dirty = False
|
||||
self._pending_progress_delayed = False
|
||||
break
|
||||
await asyncio.gather(pending_progress_task, return_exceptions=True)
|
||||
if self._pending_progress_task is pending_progress_task:
|
||||
self._pending_progress_task = None
|
||||
|
||||
while self._buffer:
|
||||
batch = self._buffer[: self._flush_threshold]
|
||||
del self._buffer[: self._flush_threshold]
|
||||
try:
|
||||
store = self._store
|
||||
if store is None:
|
||||
return
|
||||
await store.put_batch(batch)
|
||||
self._feed_generation += 1
|
||||
except Exception:
|
||||
self._buffer = batch + self._buffer
|
||||
raise
|
||||
while self._buffer:
|
||||
batch = self._buffer[: self._flush_threshold]
|
||||
del self._buffer[: self._flush_threshold]
|
||||
try:
|
||||
store = self._store
|
||||
if store is None:
|
||||
return
|
||||
await store.put_batch(batch)
|
||||
self._feed_generation += 1
|
||||
except Exception:
|
||||
self._buffer = batch + self._buffer
|
||||
raise
|
||||
finally:
|
||||
self._explicit_flush_in_progress = False
|
||||
|
||||
def _detach_runtime_dependencies(self) -> None:
|
||||
"""Drop every external or potentially cyclic run-scoped reference."""
|
||||
@ -1110,6 +1143,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
self._buffer.clear()
|
||||
self._pending_llm_response = None
|
||||
self._pending_flush_tasks.clear()
|
||||
self._explicit_flush_in_progress = False
|
||||
self._pending_progress_task = None
|
||||
self._pending_progress_delayed = False
|
||||
self._progress_dirty = False
|
||||
|
||||
@ -521,6 +521,11 @@ _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: Final[frozenset[str]] = (
|
||||
CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY,
|
||||
DEERFLOW_TRACE_METADATA_KEY,
|
||||
CONVERSATION_READER_CONTEXT_KEY,
|
||||
"is_subagent",
|
||||
"agent_id",
|
||||
"__run_loop_detection_recorder",
|
||||
"__run_tool_promotion_recorder",
|
||||
"__run_tool_progress_recorder",
|
||||
}
|
||||
)
|
||||
| SANDBOX_SERVER_OWNED_CONTEXT_KEYS
|
||||
|
||||
@ -38,6 +38,6 @@ executions are not checked, and acceptance never changes automatic retry policy.
|
||||
|
||||
**Isolated-loop callback boundary**: sync delegation from an active event loop and `execute_async()` copy the ambient ContextVars into the persistent subagent loop so checkpoint lineage, user identity, tracing context, tags, metadata, and LangGraph's namespaced message-stream handler survive. Before submission, `_copy_isolated_subagent_context()` copies the callback manager/list and removes only handlers marked `deerflow_loop_bound`; `RunJournal` carries that marker because it owns parent-loop tasks and a SQL store/pool. LangGraph merges inherited callbacks with the child run's explicit `SubagentTokenCollector`/tracing callbacks, so letting `RunJournal` cross loops causes duplicate accounting and `Future attached to a different loop` failures, while dropping the whole callback chain silently removes child token frames. Do not replace the boundary with a blank `Context`; the inherited checkpoint namespace and framework stream callback are required by the stream-isolation contract above.
|
||||
|
||||
Loop-detection and deferred-tool-promotion audit records cross this boundary through a deliberately narrow exception: `task_tool` captures the parent loop and passes one `_ParentLoopMiddlewareRecorderProxy` to `SubagentExecutor`, which installs it under separate loop-detection and tool-promotion context keys and schedules the real journal append back onto its owner loop with `call_soon_threadsafe`. The proxy exposes only `record_middleware` plus an execution-local atomic promotion claim; that claim deduplicates overlapping parallel `tool_search` calls that read the same pre-step state. Other subagent middleware consumers still do not see `__run_journal`. The task tool closes the shared proxy once before returning: close fences later child events and yields once on the owner loop so every previously accepted append reaches the journal before the parent run captures completion data. The proxy, not `RunJournal`, crosses into the isolated loop; do not broaden it into a generic journal facade or call the event store from the subagent loop. Durable batch subagents have no parent run journal and therefore do not use this bridge.
|
||||
Loop-detection, deferred-tool-promotion, and tool-progress audit records cross this boundary through one narrow exception. `task_tool` captures the parent loop and passes a `_ParentLoopMiddlewareRecorderProxy` to `SubagentExecutor`, which installs separate context keys and schedules `record_middleware` on the journal's owner loop. The proxy also owns the execution-local atomic promotion claim that deduplicates parallel `tool_search` calls. Other middleware consumers never see `__run_journal`. Close the shared proxy once before `task` returns to fence late events and drain accepted appends before completion capture. Only the proxy crosses loops; do not broaden it or call the event store from a subagent loop. Durable batch subagents have no parent journal and do not use this bridge.
|
||||
|
||||
**Reverse direction of the loop boundary — deferred cleanup & final usage delivery (#5069)**: when a task-tool poller exits unexpectedly, the registry cleanup is pinned **to** the persistent subagent loop via the public `run_on_isolated_subagent_loop()` (executor) so it survives caller-loop teardown — `asyncio.run()` cancels caller-loop tasks on exit, so a caller-loop `asyncio.create_task` would be cancelled before running. The final usage report crosses the boundary the **other way**: `_schedule_deferred_subagent_cleanup` captures the parent run's loop at unwind time (alive in every path that continues the run — the polling-timeout branch returns normally, and a generic poller error becomes an error `ToolMessage`), and `_deliver_final_usage_report` hands the report back onto that loop with `call_soon_threadsafe`. `record_external_llm_usage_records` must never be invoked from the persistent loop or a worker thread (`to_thread`): the journal's accumulators are unlocked read-modify-write fields and `get_completion_data()` iterates `_tokens_by_model`, so a cross-thread write silently loses token updates or breaks iteration mid-run — calling `_report_subagent_usage` directly inside `_deferred_cleanup_subagent_task` would reintroduce exactly this race. The deferred cleaner captures only the resolved usage recorder (plus ids and the captured report loop) — never the whole `runtime`: the strongly-referenced cleanup task lives for up to the full poll budget, and through `runtime` it would pin the parent run's journal and event store for that entire window. When the captured parent loop is already closed (synchronous `asyncio.run` teardown), the report is dropped on purpose — the run has persisted its completion data and nothing reads the counters back — and logged at info with the execution id and unaccounted record count, because the registry entry is removed right after and those records exist nowhere else.
|
||||
|
||||
@ -26,7 +26,11 @@ from langchain_core.runnables import RunnableConfig
|
||||
from langchain_core.runnables.config import var_child_runnable_config
|
||||
from langgraph.errors import GraphRecursionError
|
||||
|
||||
from deerflow.agents.middlewares.audit_context import LOOP_DETECTION_RECORDER_CONTEXT_KEY, TOOL_PROMOTION_RECORDER_CONTEXT_KEY
|
||||
from deerflow.agents.middlewares.audit_context import (
|
||||
LOOP_DETECTION_RECORDER_CONTEXT_KEY,
|
||||
TOOL_PROGRESS_RECORDER_CONTEXT_KEY,
|
||||
TOOL_PROMOTION_RECORDER_CONTEXT_KEY,
|
||||
)
|
||||
from deerflow.agents.thread_state import SandboxState, ThreadDataState, ThreadState
|
||||
from deerflow.authz.principal import normalize_authz_attributes
|
||||
from deerflow.config import get_app_config
|
||||
@ -791,6 +795,7 @@ class SubagentExecutor:
|
||||
acceptance_criteria: list[str] | None = None,
|
||||
loop_detection_recorder: Any | None = None,
|
||||
tool_promotion_recorder: Any | None = None,
|
||||
tool_progress_recorder: Any | None = None,
|
||||
context_snapshot: ParentContextSnapshot | None = None,
|
||||
):
|
||||
"""Initialize the executor.
|
||||
@ -841,6 +846,8 @@ class SubagentExecutor:
|
||||
``RunJournal`` itself.
|
||||
tool_promotion_recorder: Optional loop-safe recorder for deferred-tool
|
||||
promotion events. It follows the same isolated-loop boundary.
|
||||
tool_progress_recorder: Optional loop-safe recorder for tool-progress
|
||||
phase transitions. It follows the same isolated-loop boundary.
|
||||
context_snapshot: Optional immutable parent history captured by the
|
||||
ordinary task tool at dispatch. Rendered as background data,
|
||||
never as child execution evidence or inherited system authority.
|
||||
@ -894,6 +901,7 @@ class SubagentExecutor:
|
||||
self.acceptance_criteria = acceptance_criteria
|
||||
self.loop_detection_recorder = loop_detection_recorder
|
||||
self.tool_promotion_recorder = tool_promotion_recorder
|
||||
self.tool_progress_recorder = tool_progress_recorder
|
||||
|
||||
self._base_tools = _filter_tools(
|
||||
tools,
|
||||
@ -1511,6 +1519,8 @@ class SubagentExecutor:
|
||||
context[LOOP_DETECTION_RECORDER_CONTEXT_KEY] = self.loop_detection_recorder
|
||||
if self.tool_promotion_recorder is not None:
|
||||
context[TOOL_PROMOTION_RECORDER_CONTEXT_KEY] = self.tool_promotion_recorder
|
||||
if self.tool_progress_recorder is not None:
|
||||
context[TOOL_PROGRESS_RECORDER_CONTEXT_KEY] = self.tool_progress_recorder
|
||||
|
||||
logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} starting async execution with max_turns={self.config.max_turns}")
|
||||
|
||||
|
||||
@ -26,7 +26,7 @@ the same host reader serves; keep reading guidance separate from permission enfo
|
||||
- `batch_task`, `batch_status`, `cancel_batch` - Explicit durable batch submission/progress/cancellation. Added only while the startup SQL-backed batch submitter is installed; large results stay in the owner-scoped API/JSONL export rather than the lead context. Items accept optional `acceptance_criteria`; item queries and exports expose the separate `acceptance_verdict`. Progress counts describe execution, not acceptance; unmet and UNVERIFIED conditions never trigger automatic retries.
|
||||
- Direct `create_deerflow_agent` integrations receive cloned tools bound to their explicit `SubagentRuntime`. The bound `task` forwards that runtime's exact execution controller and optional caller-owned `AppConfig` into registry/model/tool resolution and `SubagentExecutor`; bound batch tools use the same config snapshot and resolve only that runtime's submitter before falling back to no other application's active worker. Keep the original tool name/schema unchanged so model contracts and user-tool deduplication remain stable.
|
||||
|
||||
The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection and tool-promotion keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge.
|
||||
The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection, tool-promotion, and tool-progress keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge.
|
||||
|
||||
Scheduled-task runtime note:
|
||||
- Scheduled background runs set `context.non_interactive=true` and therefore exclude `ask_clarification` from the lead-agent tool list. This keeps scheduler-triggered runs from stalling on human confirmation mid-execution. `non_interactive` is an internal-only context key: it is merged from `body.context` only when the request authenticated as the process-internal user (the scheduler path), never from arbitrary HTTP/IM clients.
|
||||
|
||||
@ -96,6 +96,9 @@ class _ParentLoopMiddlewareRecorderProxy:
|
||||
"""
|
||||
|
||||
def __init__(self, journal: Any, loop: asyncio.AbstractEventLoop) -> None:
|
||||
journal_owner_loop = getattr(journal, "_owner_loop", None)
|
||||
if isinstance(journal_owner_loop, asyncio.AbstractEventLoop) and journal_owner_loop is not loop:
|
||||
raise ValueError("subagent middleware recorder loop must match the RunJournal owner loop")
|
||||
self._journal = journal
|
||||
self._loop = loop
|
||||
self._state_lock = threading.Lock()
|
||||
@ -941,6 +944,7 @@ async def task_tool(
|
||||
)
|
||||
executor_kwargs["loop_detection_recorder"] = middleware_recorder
|
||||
executor_kwargs["tool_promotion_recorder"] = middleware_recorder
|
||||
executor_kwargs["tool_progress_recorder"] = middleware_recorder
|
||||
if resolved_app_config is not None:
|
||||
executor_kwargs["app_config"] = resolved_app_config
|
||||
if run_extensions is not None:
|
||||
|
||||
@ -3276,6 +3276,28 @@ def test_strip_internal_context_keys_scrubs_config_smuggled_context_only_keys():
|
||||
assert "disable_clarification" not in via_configurable["configurable"]
|
||||
|
||||
|
||||
def test_strip_internal_context_keys_scrubs_audit_attribution_and_recorders():
|
||||
from app.gateway.services import build_run_config, strip_internal_context_keys
|
||||
|
||||
server_owned = {
|
||||
"is_subagent": True,
|
||||
"agent_id": "forged-agent",
|
||||
"__run_loop_detection_recorder": "forged",
|
||||
"__run_tool_promotion_recorder": "forged",
|
||||
"__run_tool_progress_recorder": "forged",
|
||||
}
|
||||
config = build_run_config(
|
||||
"thread-1",
|
||||
{"context": dict(server_owned), "configurable": dict(server_owned)},
|
||||
None,
|
||||
)
|
||||
|
||||
strip_internal_context_keys(config)
|
||||
|
||||
for section in ("context", "configurable"):
|
||||
assert not server_owned.keys() & config[section].keys()
|
||||
|
||||
|
||||
def test_start_run_sequence_drops_context_only_keys_for_session_caller():
|
||||
"""Replay the real ``start_run`` assembly order for a session-authenticated caller
|
||||
that pushes the keys through *both* smuggling surfaces at once."""
|
||||
|
||||
@ -941,7 +941,6 @@ class TestLoopDetectionRunEvents:
|
||||
recorder = MagicMock()
|
||||
runtime = _make_runtime()
|
||||
runtime.context["__run_loop_detection_recorder"] = recorder
|
||||
runtime.context["is_subagent"] = True
|
||||
runtime.context["agent_id"] = "general-purpose"
|
||||
assert "__run_journal" not in runtime.context
|
||||
mw = LoopDetectionMiddleware(
|
||||
@ -960,6 +959,26 @@ class TestLoopDetectionRunEvents:
|
||||
assert recorder.record_middleware.call_args.kwargs["changes"]["is_subagent"] is True
|
||||
assert recorder.record_middleware.call_args.kwargs["changes"]["agent_id"] == "general-purpose"
|
||||
|
||||
def test_lead_attribution_ignores_caller_supplied_subagent_fields(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
runtime.context["is_subagent"] = True
|
||||
runtime.context["agent_id"] = "forged-agent"
|
||||
mw = LoopDetectionMiddleware(
|
||||
warn_threshold=2,
|
||||
hard_limit=10,
|
||||
tool_freq_warn=100,
|
||||
tool_freq_hard_limit=200,
|
||||
)
|
||||
call = [_bash_call("ls")]
|
||||
|
||||
assert mw._apply(_make_state(tool_calls=call), runtime) is None
|
||||
assert mw._apply(_make_state(tool_calls=call), runtime) is None
|
||||
|
||||
changes = journal.record_middleware.call_args.kwargs["changes"]
|
||||
assert changes["is_subagent"] is False
|
||||
assert changes["agent_id"] is None
|
||||
|
||||
def test_identical_call_hard_stop_records_event(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
|
||||
@ -18,6 +18,7 @@ from deerflow.runtime.events.catalog import (
|
||||
MIDDLEWARE_EVENT_PATTERN,
|
||||
MIDDLEWARE_EVENT_TAG_MAX_LENGTH,
|
||||
MIDDLEWARE_EVENT_TAGS,
|
||||
MIDDLEWARE_TOOL_PROGRESS_TAG,
|
||||
MIDDLEWARE_TOOL_PROMOTION_TAG,
|
||||
RUN_EVENT_CATEGORY_MAX_LENGTH,
|
||||
RUN_EVENT_TYPE_MAX_LENGTH,
|
||||
@ -377,6 +378,15 @@ def test_tool_promotion_tag_is_declared_and_fits_the_persisted_event_type():
|
||||
assert len(MIDDLEWARE_EVENT_PATTERN.event_type(MIDDLEWARE_TOOL_PROMOTION_TAG)) <= RUN_EVENT_TYPE_MAX_LENGTH
|
||||
|
||||
|
||||
def test_tool_progress_tag_is_declared_and_fits_the_persisted_event_type():
|
||||
pattern = _load_contract()["dynamic_event_patterns"][0]
|
||||
|
||||
assert MIDDLEWARE_TOOL_PROGRESS_TAG == "tool_progress"
|
||||
assert MIDDLEWARE_TOOL_PROGRESS_TAG in MIDDLEWARE_EVENT_TAGS
|
||||
assert MIDDLEWARE_TOOL_PROGRESS_TAG in pattern["known_tags"]
|
||||
assert len(MIDDLEWARE_EVENT_PATTERN.event_type(MIDDLEWARE_TOOL_PROGRESS_TAG)) <= RUN_EVENT_TYPE_MAX_LENGTH
|
||||
|
||||
|
||||
def test_subagent_observed_events_exactly_match_its_catalog_and_payloads():
|
||||
long_result = "r" * (SUBAGENT_STEP_MAX_CHARS + 1)
|
||||
long_error = "e" * (SUBAGENT_STEP_MAX_CHARS + 1)
|
||||
|
||||
@ -4,6 +4,7 @@ Uses MemoryRunEventStore as the backend for direct event inspection.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
import weakref
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from threading import Barrier
|
||||
@ -37,6 +38,137 @@ def test_tool_promotion_claim_is_atomic_across_parallel_sync_wrappers():
|
||||
assert sum((result for result in results), []) == ["mcp_a"]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_thread_middleware_events_are_serialized_on_owner_loop():
|
||||
store = MemoryRunEventStore()
|
||||
journal = RunJournal("r-thread", "t-thread", store, flush_threshold=1)
|
||||
owner_thread_id = threading.get_ident()
|
||||
put_thread_ids: list[int] = []
|
||||
original_put = journal._put
|
||||
|
||||
def tracked_put(**kwargs) -> None:
|
||||
put_thread_ids.append(threading.get_ident())
|
||||
original_put(**kwargs)
|
||||
|
||||
journal._put = tracked_put
|
||||
|
||||
def record_from_tool_worker() -> None:
|
||||
journal.record_middleware(
|
||||
"tool_progress",
|
||||
name="ToolProgressMiddleware",
|
||||
hook="wrap_tool_call",
|
||||
action="warn",
|
||||
changes={"from_phase": "active", "to_phase": "warned"},
|
||||
)
|
||||
|
||||
await asyncio.to_thread(record_from_tool_worker)
|
||||
await journal.flush()
|
||||
|
||||
assert put_thread_ids == [owner_thread_id]
|
||||
events = await store.list_events("t-thread", "r-thread")
|
||||
assert [event["event_type"] for event in events] == ["middleware:tool_progress"]
|
||||
assert events[0]["content"]["changes"]["to_phase"] == "warned"
|
||||
|
||||
|
||||
def test_middleware_event_without_owner_loop_keeps_cross_thread_append():
|
||||
store = MemoryRunEventStore()
|
||||
journal = RunJournal("r-sync", "t-sync", store, flush_threshold=100)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=1) as pool:
|
||||
pool.submit(
|
||||
journal.record_middleware,
|
||||
"tool_progress",
|
||||
name="ToolProgressMiddleware",
|
||||
hook="wrap_tool_call",
|
||||
action="warn",
|
||||
changes={"from_phase": "active", "to_phase": "warned"},
|
||||
).result(timeout=5)
|
||||
|
||||
asyncio.run(journal.flush())
|
||||
events = asyncio.run(store.list_events("t-sync", "r-sync"))
|
||||
assert [event["event_type"] for event in events] == ["middleware:tool_progress"]
|
||||
|
||||
|
||||
def test_middleware_event_uses_owner_loop_identity_after_loop_moves_threads():
|
||||
loop = asyncio.new_event_loop()
|
||||
|
||||
async def build_journal():
|
||||
return RunJournal("r-moved", "t-moved", MemoryRunEventStore(), flush_threshold=100)
|
||||
|
||||
journal = loop.run_until_complete(build_journal())
|
||||
|
||||
async def record_on_current_loop() -> int:
|
||||
journal.record_middleware(
|
||||
"tool_progress",
|
||||
name="ToolProgressMiddleware",
|
||||
hook="wrap_tool_call",
|
||||
action="warn",
|
||||
changes={},
|
||||
)
|
||||
return len(journal._buffer)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=1) as pool:
|
||||
buffered = pool.submit(loop.run_until_complete, record_on_current_loop()).result(timeout=5)
|
||||
|
||||
assert buffered == 1
|
||||
loop.run_until_complete(journal.flush())
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_cross_thread_append_during_explicit_flush_is_not_flushed_concurrently():
|
||||
class BlockingStore(MemoryRunEventStore):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.started = asyncio.Event()
|
||||
self.release = asyncio.Event()
|
||||
self.active_writes = 0
|
||||
self.max_active_writes = 0
|
||||
|
||||
async def put_batch(self, events):
|
||||
self.active_writes += 1
|
||||
self.max_active_writes = max(self.max_active_writes, self.active_writes)
|
||||
if not self.started.is_set():
|
||||
self.started.set()
|
||||
await self.release.wait()
|
||||
try:
|
||||
await super().put_batch(events)
|
||||
finally:
|
||||
self.active_writes -= 1
|
||||
|
||||
store = BlockingStore()
|
||||
journal = RunJournal("r-flush", "t-flush", store, flush_threshold=1)
|
||||
journal._buffer.append(
|
||||
journal._make_event(
|
||||
event_type="middleware:test",
|
||||
category="middleware",
|
||||
content={},
|
||||
)
|
||||
)
|
||||
flush_task = asyncio.create_task(journal.flush())
|
||||
await store.started.wait()
|
||||
|
||||
await asyncio.to_thread(
|
||||
journal.record_middleware,
|
||||
"tool_progress",
|
||||
name="ToolProgressMiddleware",
|
||||
hook="wrap_tool_call",
|
||||
action="warn",
|
||||
changes={},
|
||||
)
|
||||
await asyncio.sleep(0)
|
||||
assert store.max_active_writes == 1
|
||||
|
||||
store.release.set()
|
||||
await flush_task
|
||||
assert store.max_active_writes == 1
|
||||
events = await store.list_events("t-flush", "r-flush")
|
||||
assert [event["event_type"] for event in events] == [
|
||||
"middleware:test",
|
||||
"middleware:tool_progress",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_close_flushes_and_detaches_runtime_dependencies():
|
||||
class ProgressReporter:
|
||||
|
||||
@ -2532,6 +2532,20 @@ def test_build_runtime_context_ignores_caller_sandbox_execution_identities():
|
||||
assert SANDBOX_COMMAND_SCOPE_CONTEXT_KEY not in ctx
|
||||
|
||||
|
||||
def test_build_runtime_context_ignores_caller_audit_attribution_and_recorders():
|
||||
caller_context = {
|
||||
"is_subagent": True,
|
||||
"agent_id": "forged-agent",
|
||||
"__run_loop_detection_recorder": object(),
|
||||
"__run_tool_promotion_recorder": object(),
|
||||
"__run_tool_progress_recorder": object(),
|
||||
}
|
||||
|
||||
ctx = _build_runtime_context("thread-1", "run-1", caller_context)
|
||||
|
||||
assert set(caller_context).isdisjoint(ctx)
|
||||
|
||||
|
||||
def test_build_runtime_context_ignores_non_dict_caller_context():
|
||||
ctx = _build_runtime_context("thread-1", "run-1", "not-a-dict")
|
||||
assert ctx == {"thread_id": "thread-1", "run_id": "run-1"}
|
||||
|
||||
@ -3987,6 +3987,7 @@ class TestSubagentGuardrailAttribution:
|
||||
run_id=None,
|
||||
loop_detection_recorder=None,
|
||||
tool_promotion_recorder=None,
|
||||
tool_progress_recorder=None,
|
||||
name="general-purpose",
|
||||
parent_model="test-model",
|
||||
):
|
||||
@ -3999,6 +4000,12 @@ class TestSubagentGuardrailAttribution:
|
||||
max_turns=5,
|
||||
timeout_seconds=30,
|
||||
)
|
||||
recorder_kwargs = {}
|
||||
if tool_progress_recorder is not None:
|
||||
# Kept conditional so the pre-feature attribution cases exercise
|
||||
# the existing constructor surface; only the new propagation case
|
||||
# requires the additive recorder argument.
|
||||
recorder_kwargs["tool_progress_recorder"] = tool_progress_recorder
|
||||
return SubagentExecutor(
|
||||
config=config,
|
||||
tools=[],
|
||||
@ -4012,6 +4019,7 @@ class TestSubagentGuardrailAttribution:
|
||||
run_id=run_id,
|
||||
loop_detection_recorder=loop_detection_recorder,
|
||||
tool_promotion_recorder=tool_promotion_recorder,
|
||||
**recorder_kwargs,
|
||||
)
|
||||
|
||||
@pytest.mark.anyio
|
||||
@ -4104,6 +4112,32 @@ class TestSubagentGuardrailAttribution:
|
||||
assert "__run_journal" not in context
|
||||
assert context.get("agent_id") == "general-purpose"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_propagates_narrow_tool_progress_recorder(
|
||||
self,
|
||||
classes,
|
||||
executor_module,
|
||||
monkeypatch,
|
||||
):
|
||||
"""Progress audit crosses the child-loop boundary without the raw journal."""
|
||||
recorder = object()
|
||||
executor = self._make_executor(
|
||||
classes,
|
||||
run_id="run-42",
|
||||
tool_progress_recorder=recorder,
|
||||
)
|
||||
fake_agent = _FakeStreamAgent()
|
||||
monkeypatch.setattr(executor, "_build_initial_state", self._noop_build_initial_state)
|
||||
monkeypatch.setattr(executor, "_create_agent", lambda *a, **kw: fake_agent)
|
||||
|
||||
await executor._aexecute("do something")
|
||||
|
||||
context = fake_agent.captured_context
|
||||
assert context is not None
|
||||
assert context.get("__run_tool_progress_recorder") is recorder
|
||||
assert "__run_journal" not in context
|
||||
assert context.get("agent_id") == "general-purpose"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_propagates_channel_user_id_to_subagent_context(
|
||||
self,
|
||||
|
||||
@ -36,6 +36,18 @@ from deerflow.subagents.status_contract import (
|
||||
task_tool_module = importlib.import_module("deerflow.tools.builtins.task_tool")
|
||||
|
||||
|
||||
def test_parent_loop_middleware_recorder_requires_the_journal_owner_loop():
|
||||
owner_loop = asyncio.new_event_loop()
|
||||
other_loop = asyncio.new_event_loop()
|
||||
journal = SimpleNamespace(_owner_loop=owner_loop)
|
||||
try:
|
||||
with pytest.raises(ValueError, match="must match"):
|
||||
task_tool_module._ParentLoopMiddlewareRecorderProxy(journal, other_loop)
|
||||
finally:
|
||||
owner_loop.close()
|
||||
other_loop.close()
|
||||
|
||||
|
||||
def test_parent_loop_middleware_recorder_proxy_delivers_on_owner_loop():
|
||||
"""Subagent middleware events must never call RunJournal from the child loop."""
|
||||
calls: list[tuple[object, dict]] = []
|
||||
@ -462,6 +474,7 @@ def test_task_tool_installs_and_closes_narrow_middleware_recorder(monkeypatch):
|
||||
kwargs = captured["executor_kwargs"]
|
||||
proxy = kwargs["loop_detection_recorder"]
|
||||
assert kwargs["tool_promotion_recorder"] is proxy
|
||||
assert kwargs["tool_progress_recorder"] is proxy
|
||||
assert proxy.is_closed is True
|
||||
proxy.record_middleware(tag="loop_detection", name="LoopDetectionMiddleware", hook="after_model", action="warn", changes={})
|
||||
journal.record_middleware.assert_not_called()
|
||||
|
||||
@ -3,6 +3,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
@ -16,6 +18,8 @@ from deerflow.agents.middlewares.tool_progress_middleware import (
|
||||
word_set,
|
||||
)
|
||||
from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
|
||||
from deerflow.runtime.events.store.memory import MemoryRunEventStore
|
||||
from deerflow.runtime.journal import RunJournal
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
@ -593,6 +597,8 @@ def test_before_agent_resets_blocked_states_for_new_run():
|
||||
mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
|
||||
rt_run1 = _make_runtime(thread_id="t1", run_id="run-1")
|
||||
rt_run2 = _make_runtime(thread_id="t1", run_id="run-2")
|
||||
journal = MagicMock()
|
||||
rt_run2.context["__run_journal"] = journal
|
||||
req = _make_tool_request(runtime=rt_run1)
|
||||
|
||||
# Drive the tool to BLOCKED via auth error (immediate block, no WARN stage)
|
||||
@ -622,6 +628,23 @@ def test_before_agent_resets_blocked_states_for_new_run():
|
||||
assert tool_state.consecutive_problems == 0
|
||||
assert tool_state.block_reason is None
|
||||
assert tool_state.recent_word_sets == ()
|
||||
journal.record_middleware.assert_called_once()
|
||||
reset = journal.record_middleware.call_args.kwargs
|
||||
assert reset["hook"] == "before_agent"
|
||||
assert reset["action"] == "reset"
|
||||
assert reset["changes"] == {
|
||||
"is_subagent": False,
|
||||
"agent_id": None,
|
||||
"tool_name": "web_search",
|
||||
"from_phase": "blocked",
|
||||
"to_phase": "active",
|
||||
"consecutive_problems": 0,
|
||||
"status": None,
|
||||
"error_type": None,
|
||||
"recoverable_by_model": None,
|
||||
"recommended_next_action": None,
|
||||
"threshold": None,
|
||||
}
|
||||
|
||||
|
||||
def test_before_agent_resets_warned_states_for_new_run():
|
||||
@ -634,6 +657,8 @@ def test_before_agent_resets_warned_states_for_new_run():
|
||||
mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5)
|
||||
rt_run1 = _make_runtime(thread_id="t1", run_id="run-1")
|
||||
rt_run2 = _make_runtime(thread_id="t1", run_id="run-2")
|
||||
journal = MagicMock()
|
||||
rt_run2.context["__run_journal"] = journal
|
||||
req = _make_tool_request(runtime=rt_run1)
|
||||
error_msg = _make_error_message()
|
||||
|
||||
@ -651,6 +676,53 @@ def test_before_agent_resets_warned_states_for_new_run():
|
||||
assert tool_state.phase == "active"
|
||||
assert tool_state.consecutive_problems == 0
|
||||
assert tool_state.recent_word_sets == ()
|
||||
assert journal.record_middleware.call_args.kwargs["action"] == "reset"
|
||||
assert journal.record_middleware.call_args.kwargs["changes"]["from_phase"] == "warned"
|
||||
|
||||
|
||||
def test_before_agent_reset_recorder_does_not_hold_the_state_lock():
|
||||
"""A slow reset recorder must not stall concurrent state access."""
|
||||
|
||||
class DelayingResetRecorder:
|
||||
def __init__(self):
|
||||
self.reset_started = threading.Event()
|
||||
self.release_reset = threading.Event()
|
||||
|
||||
def record_middleware(self, **kwargs):
|
||||
if kwargs["action"] == "reset":
|
||||
self.reset_started.set()
|
||||
self.release_reset.wait(timeout=30)
|
||||
|
||||
mw = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
|
||||
run_one = _make_runtime(thread_id="t1", run_id="run-1")
|
||||
request = _make_tool_request(runtime=run_one)
|
||||
blocked = ToolMessage(
|
||||
content="Error: invalid api key",
|
||||
tool_call_id="tc-web_search",
|
||||
name="web_search",
|
||||
status="error",
|
||||
additional_kwargs=_meta_kwargs(
|
||||
status="error",
|
||||
error_type="auth",
|
||||
recoverable_by_model=False,
|
||||
recommended_next_action="stop",
|
||||
),
|
||||
)
|
||||
mw.wrap_tool_call(request, lambda _request: blocked)
|
||||
|
||||
recorder = DelayingResetRecorder()
|
||||
run_two = _make_runtime(thread_id="t1", run_id="run-2")
|
||||
run_two.context["__run_journal"] = recorder
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
reset_future = pool.submit(mw.before_agent, MagicMock(), run_two)
|
||||
assert recorder.reset_started.wait(timeout=5)
|
||||
state_future = pool.submit(mw._get_block_reason, run_two, "web_search")
|
||||
try:
|
||||
assert state_future.result(timeout=5) is None
|
||||
finally:
|
||||
recorder.release_reset.set()
|
||||
reset_future.result(timeout=5)
|
||||
|
||||
|
||||
def test_before_agent_resets_active_state_consecutive_problems_and_word_sets():
|
||||
@ -954,11 +1026,12 @@ def test_assess_and_transition_blocked_state_immediate_stop_is_idempotent():
|
||||
|
||||
auth_meta = ToolResultMeta(**auth_meta_kwargs)
|
||||
|
||||
new_state, hint = mw._assess_and_transition(blocked_state, auth_meta, "")
|
||||
new_state, hint, transition = mw._assess_and_transition(blocked_state, auth_meta, "")
|
||||
|
||||
assert new_state.phase == "blocked"
|
||||
assert new_state.block_reason is not None
|
||||
assert hint is None # no hint on immediate block path
|
||||
assert transition is None
|
||||
|
||||
|
||||
def test_assess_and_transition_blocked_state_non_stop_increments_count():
|
||||
@ -986,12 +1059,13 @@ def test_assess_and_transition_blocked_state_non_stop_increments_count():
|
||||
|
||||
rate_meta = ToolResultMeta(**rate_meta_kwargs)
|
||||
|
||||
new_state, _hint = mw._assess_and_transition(blocked_state, rate_meta, "")
|
||||
new_state, _hint, transition = mw._assess_and_transition(blocked_state, rate_meta, "")
|
||||
|
||||
# Must stay blocked (not regress to warned or active).
|
||||
assert new_state.phase == "blocked"
|
||||
# Counter must NOT be incremented: blocked is terminal, state returned unchanged.
|
||||
assert new_state.consecutive_problems == 3
|
||||
assert transition is None
|
||||
|
||||
|
||||
def test_assess_and_transition_blocked_recoverable_does_not_regress_to_warned():
|
||||
@ -1021,11 +1095,12 @@ def test_assess_and_transition_blocked_recoverable_does_not_regress_to_warned():
|
||||
|
||||
no_results_meta = ToolResultMeta(**no_results_meta_kwargs)
|
||||
|
||||
new_state, hint = mw._assess_and_transition(blocked_state, no_results_meta, "")
|
||||
new_state, hint, transition = mw._assess_and_transition(blocked_state, no_results_meta, "")
|
||||
|
||||
assert new_state.phase == "blocked", "blocked must not regress to warned even when the new error is recoverable"
|
||||
assert hint is None
|
||||
assert new_state is blocked_state # exact same object returned (no copy)
|
||||
assert transition is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -1382,6 +1457,290 @@ def test_log_warned_to_active_reset_emits_info(caplog):
|
||||
assert "web_search" in reset_records[0].message
|
||||
|
||||
|
||||
class TestToolProgressRunEvents:
|
||||
"""Durable audit coverage for state-machine interventions.
|
||||
|
||||
The tool result feed records what a tool returned, but it cannot prove that
|
||||
ToolProgressMiddleware crossed a phase boundary and changed later runtime
|
||||
behavior. Persist exactly those phase transitions, without copying tool
|
||||
arguments or result content into the middleware event.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _runtime_with_journal(journal):
|
||||
runtime = _make_runtime()
|
||||
runtime.context["__run_journal"] = journal
|
||||
return runtime
|
||||
|
||||
def test_lead_warn_and_recover_transitions_are_recorded_without_result_content(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=2, warn_escalation_count=2)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
secret = "SENSITIVE_TOOL_RESULT_MUST_NOT_BE_PERSISTED"
|
||||
no_results = _make_error_message(content=f"Error: no results found {secret}")
|
||||
|
||||
# The first problem remains ACTIVE; only the phase-changing second call
|
||||
# is a durable intervention.
|
||||
assert middleware.wrap_tool_call(request, lambda _request: no_results) is no_results
|
||||
assert middleware.wrap_tool_call(request, lambda _request: no_results) is no_results
|
||||
|
||||
journal.record_middleware.assert_called_once()
|
||||
warned = journal.record_middleware.call_args
|
||||
assert warned.kwargs["tag"] == "tool_progress"
|
||||
assert warned.kwargs["name"] == "ToolProgressMiddleware"
|
||||
assert warned.kwargs["hook"] == "wrap_tool_call"
|
||||
assert warned.kwargs["action"] == "warn"
|
||||
assert warned.kwargs["changes"] == {
|
||||
"is_subagent": False,
|
||||
"agent_id": None,
|
||||
"tool_name": "web_search",
|
||||
"from_phase": "active",
|
||||
"to_phase": "warned",
|
||||
"consecutive_problems": 2,
|
||||
"status": "error",
|
||||
"error_type": "no_results",
|
||||
"recoverable_by_model": True,
|
||||
"recommended_next_action": "rewrite_query",
|
||||
"threshold": 2,
|
||||
}
|
||||
assert secret not in repr(warned)
|
||||
assert "content" not in warned.kwargs["changes"]
|
||||
assert "args" not in warned.kwargs["changes"]
|
||||
|
||||
recovered_result = _make_tool_message(
|
||||
"fresh evidence with enough distinct words to remain a useful result",
|
||||
)
|
||||
assert middleware.wrap_tool_call(request, lambda _request: recovered_result) is recovered_result
|
||||
|
||||
assert journal.record_middleware.call_count == 2
|
||||
recovered = journal.record_middleware.call_args_list[-1]
|
||||
assert recovered.kwargs["action"] == "recover"
|
||||
assert recovered.kwargs["changes"] == {
|
||||
"is_subagent": False,
|
||||
"agent_id": None,
|
||||
"tool_name": "web_search",
|
||||
"from_phase": "warned",
|
||||
"to_phase": "active",
|
||||
"consecutive_problems": 0,
|
||||
"status": "success",
|
||||
"error_type": None,
|
||||
"recoverable_by_model": True,
|
||||
"recommended_next_action": "continue",
|
||||
"threshold": None,
|
||||
}
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_warn_transition_round_trips_through_run_journal(self):
|
||||
store = MemoryRunEventStore()
|
||||
journal = RunJournal("r1", "t1", store, flush_threshold=100)
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
|
||||
middleware.wrap_tool_call(request, lambda _request: _make_error_message())
|
||||
await journal.flush()
|
||||
|
||||
events = await store.list_events("t1", "r1")
|
||||
assert len(events) == 1
|
||||
assert events[0]["event_type"] == "middleware:tool_progress"
|
||||
assert events[0]["category"] == "middleware"
|
||||
assert events[0]["content"]["action"] == "warn"
|
||||
assert events[0]["content"]["changes"]["to_phase"] == "warned"
|
||||
|
||||
def test_lead_immediate_block_transition_is_recorded(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=5)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
auth_error = _make_error_message(
|
||||
content="Error: invalid API key",
|
||||
error_type="auth",
|
||||
recoverable_by_model=False,
|
||||
recommended_next_action="stop",
|
||||
)
|
||||
|
||||
assert middleware.wrap_tool_call(request, lambda _request: auth_error) is auth_error
|
||||
|
||||
journal.record_middleware.assert_called_once()
|
||||
blocked = journal.record_middleware.call_args
|
||||
assert blocked.kwargs["tag"] == "tool_progress"
|
||||
assert blocked.kwargs["action"] == "block"
|
||||
assert blocked.kwargs["changes"] == {
|
||||
"is_subagent": False,
|
||||
"agent_id": None,
|
||||
"tool_name": "web_search",
|
||||
"from_phase": "active",
|
||||
"to_phase": "blocked",
|
||||
"consecutive_problems": 1,
|
||||
"status": "error",
|
||||
"error_type": "auth",
|
||||
"recoverable_by_model": False,
|
||||
"recommended_next_action": "stop",
|
||||
"threshold": None,
|
||||
}
|
||||
|
||||
def test_zero_warn_escalation_still_records_active_to_warned(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1, warn_escalation_count=0)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
|
||||
result = _make_error_message()
|
||||
assert middleware.wrap_tool_call(request, lambda _request: result) is result
|
||||
|
||||
journal.record_middleware.assert_called_once()
|
||||
recorded = journal.record_middleware.call_args.kwargs
|
||||
assert recorded["action"] == "warn"
|
||||
assert recorded["changes"]["from_phase"] == "active"
|
||||
assert recorded["changes"]["to_phase"] == "warned"
|
||||
assert recorded["changes"]["threshold"] == 1
|
||||
|
||||
def test_near_duplicate_warn_records_success_status(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
content = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda"
|
||||
result = _make_tool_message(content)
|
||||
|
||||
middleware.wrap_tool_call(request, lambda _request: result)
|
||||
middleware.wrap_tool_call(request, lambda _request: result)
|
||||
|
||||
recorded = journal.record_middleware.call_args.kwargs
|
||||
assert recorded["action"] == "warn"
|
||||
assert recorded["changes"]["status"] == "success"
|
||||
assert recorded["changes"]["error_type"] is None
|
||||
assert recorded["changes"]["recommended_next_action"] == "continue"
|
||||
|
||||
def test_producer_supplied_meta_is_projected_onto_bounded_audit_values(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
secret = "PRIVATE_EXTENSION_VALUE_" * 50
|
||||
result = _make_tool_message(
|
||||
"ordinary result content",
|
||||
meta_kwargs=_meta_kwargs(
|
||||
status="error",
|
||||
error_type=secret,
|
||||
recoverable_by_model=secret,
|
||||
recommended_next_action=secret,
|
||||
source=secret,
|
||||
),
|
||||
)
|
||||
|
||||
assert middleware.wrap_tool_call(request, lambda _request: result) is result
|
||||
|
||||
recorded = journal.record_middleware.call_args
|
||||
assert recorded.kwargs["changes"]["error_type"] == "unknown"
|
||||
assert recorded.kwargs["changes"]["recoverable_by_model"] is None
|
||||
assert recorded.kwargs["changes"]["recommended_next_action"] == "unknown"
|
||||
assert recorded.kwargs["changes"]["status"] == "error"
|
||||
assert secret not in repr(recorded)
|
||||
|
||||
def test_recorder_failure_is_fail_open(self, caplog):
|
||||
journal = MagicMock()
|
||||
journal.record_middleware.side_effect = RuntimeError("event store unavailable")
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
no_results = _make_error_message()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger=_MW_LOGGER):
|
||||
result = middleware.wrap_tool_call(request, lambda _request: no_results)
|
||||
|
||||
assert result is no_results
|
||||
assert middleware._phase_states["t1"]["web_search"].phase == "warned"
|
||||
assert middleware._pending[("t1", "r1")]
|
||||
assert "Failed to record middleware:tool_progress event" in caplog.text
|
||||
|
||||
def test_narrow_subagent_recorder_records_without_crossing_raw_journal(self):
|
||||
recorder = MagicMock()
|
||||
runtime = _make_runtime()
|
||||
runtime.context["__run_tool_progress_recorder"] = recorder
|
||||
runtime.context["agent_id"] = "general-purpose"
|
||||
assert "__run_journal" not in runtime.context
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
|
||||
middleware.wrap_tool_call(request, lambda _request: _make_error_message())
|
||||
|
||||
recorder.record_middleware.assert_called_once()
|
||||
recorded = recorder.record_middleware.call_args
|
||||
assert recorded.kwargs["tag"] == "tool_progress"
|
||||
assert recorded.kwargs["action"] == "warn"
|
||||
assert recorded.kwargs["changes"]["is_subagent"] is True
|
||||
assert recorded.kwargs["changes"]["agent_id"] == "general-purpose"
|
||||
|
||||
def test_lead_attribution_ignores_caller_supplied_subagent_fields(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
runtime.context["is_subagent"] = True
|
||||
runtime.context["agent_id"] = "forged-agent"
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
|
||||
middleware.wrap_tool_call(request, lambda _request: _make_error_message())
|
||||
|
||||
recorded = journal.record_middleware.call_args
|
||||
assert recorded.kwargs["changes"]["is_subagent"] is False
|
||||
assert recorded.kwargs["changes"]["agent_id"] is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_async_transition_records_actual_hook(self):
|
||||
journal = MagicMock()
|
||||
runtime = self._runtime_with_journal(journal)
|
||||
middleware = _make_mw(stagnation_threshold=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
|
||||
result = _make_error_message()
|
||||
assert await middleware.awrap_tool_call(request, AsyncMock(return_value=result)) is result
|
||||
|
||||
journal.record_middleware.assert_called_once()
|
||||
assert journal.record_middleware.call_args.kwargs["hook"] == "awrap_tool_call"
|
||||
|
||||
def test_slow_recorder_does_not_hold_the_state_lock(self):
|
||||
"""A custom recorder cannot stall a second state-machine transition."""
|
||||
|
||||
class DelayingRecorder:
|
||||
def __init__(self):
|
||||
self.warn_started = threading.Event()
|
||||
self.release_warn = threading.Event()
|
||||
self.block_seen = threading.Event()
|
||||
self.warn_timed_out = False
|
||||
|
||||
def record_middleware(self, **kwargs):
|
||||
action = kwargs["action"]
|
||||
if action == "warn":
|
||||
self.warn_started.set()
|
||||
self.warn_timed_out = not self.release_warn.wait(timeout=5)
|
||||
elif action == "block":
|
||||
self.block_seen.set()
|
||||
|
||||
recorder = DelayingRecorder()
|
||||
runtime = self._runtime_with_journal(recorder)
|
||||
middleware = _make_mw(stagnation_threshold=1, warn_escalation_count=1)
|
||||
request = _make_tool_request(runtime=runtime)
|
||||
result = _make_non_recoverable_error_message()
|
||||
handlers_ready = threading.Barrier(2)
|
||||
|
||||
def complete_tool(_request):
|
||||
handlers_ready.wait(timeout=5)
|
||||
return result
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||||
futures = [pool.submit(middleware.wrap_tool_call, request, complete_tool) for _ in range(2)]
|
||||
assert recorder.warn_started.wait(timeout=5)
|
||||
assert recorder.block_seen.wait(timeout=5)
|
||||
recorder.release_warn.set()
|
||||
results = [future.result(timeout=5) for future in futures]
|
||||
|
||||
assert results == [result, result]
|
||||
assert recorder.warn_timed_out is False
|
||||
assert middleware._phase_states["t1"]["web_search"].phase == "blocked"
|
||||
|
||||
|
||||
def test_log_hint_injection_emits_debug(caplog):
|
||||
mw = _make_mw(stagnation_threshold=2, warn_escalation_count=5)
|
||||
rt = _make_runtime()
|
||||
|
||||
@ -74,6 +74,39 @@ def test_records_only_new_names_from_the_final_current_catalog_command():
|
||||
assert "h1" not in persisted
|
||||
|
||||
|
||||
def test_narrow_recorder_identifies_subagent_without_caller_flag():
|
||||
recorder = _Recorder()
|
||||
request = _ToolRequest(
|
||||
context={
|
||||
"__run_tool_promotion_recorder": recorder,
|
||||
"agent_id": "general-purpose",
|
||||
}
|
||||
)
|
||||
result = Command(update={"promoted": {"catalog_hash": "h1", "names": ["mcp_a"]}})
|
||||
|
||||
assert _middleware().wrap_tool_call(request, lambda _: result) is result
|
||||
|
||||
assert recorder.calls[0]["changes"]["is_subagent"] is True
|
||||
assert recorder.calls[0]["changes"]["agent_id"] == "general-purpose"
|
||||
|
||||
|
||||
def test_lead_attribution_ignores_caller_supplied_subagent_fields():
|
||||
recorder = _Recorder()
|
||||
request = _ToolRequest(
|
||||
context={
|
||||
"__run_journal": recorder,
|
||||
"is_subagent": True,
|
||||
"agent_id": "forged-agent",
|
||||
}
|
||||
)
|
||||
result = Command(update={"promoted": {"catalog_hash": "h1", "names": ["mcp_a"]}})
|
||||
|
||||
assert _middleware().wrap_tool_call(request, lambda _: result) is result
|
||||
|
||||
assert recorder.calls[0]["changes"]["is_subagent"] is False
|
||||
assert recorder.calls[0]["changes"]["agent_id"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_records_promotion_but_repeated_stale_and_non_search_results_do_not():
|
||||
recorder = _Recorder()
|
||||
|
||||
@ -10,6 +10,7 @@ from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.middlewares.tool_result_meta import (
|
||||
TOOL_META_KEY,
|
||||
TOOL_RESULT_ERROR_TYPES,
|
||||
ToolResultMeta,
|
||||
normalize_tool_message,
|
||||
normalize_tool_result,
|
||||
@ -66,6 +67,7 @@ def test_error_prefix_classification(snippet: str, expected_type: str):
|
||||
m = _meta(result)
|
||||
assert m["status"] == "error"
|
||||
assert m["error_type"] == expected_type
|
||||
assert m["error_type"] in TOOL_RESULT_ERROR_TYPES
|
||||
assert m["source"] == "tool_return"
|
||||
|
||||
|
||||
|
||||
@ -397,7 +397,7 @@
|
||||
"pattern": "middleware:{tag}",
|
||||
"category": "middleware",
|
||||
"producer": "RunJournal.record_middleware(tag, ...)",
|
||||
"known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion"],
|
||||
"known_tags": ["guardrail", "loop_detection", "safety_termination", "skill_activation", "skill_secrets", "tool_promotion", "tool_progress"],
|
||||
"event_type_schema": {
|
||||
"type": "string",
|
||||
"pattern": "^middleware:",
|
||||
@ -444,7 +444,7 @@
|
||||
{
|
||||
"id": "middleware-coverage",
|
||||
"status": "partial",
|
||||
"notes": "Loop-detection and deferred-tool promotion events cover lead-agent and ordinary task-tool subagent runs. Durable batch subagents have no parent run journal and do not currently emit either event."
|
||||
"notes": "Loop-detection, deferred-tool promotion, and tool-progress events cover lead-agent and ordinary task-tool subagent runs. Durable batch subagents have no parent run journal and do not currently emit these events."
|
||||
},
|
||||
{
|
||||
"id": "run-scoped-observation-context",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user