From f1632cc3511fd614879dcdfdf1f349ae392b746d Mon Sep 17 00:00:00 2001 From: MiaoRuidx <965742329@qq.com> Date: Thu, 23 Jul 2026 21:33:57 +0800 Subject: [PATCH] fix(run): add run event stream contract (#4342) * docs: document run event stream contract * fix(run): address event stream review feedback --------- Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com> Co-authored-by: Willem Jiang --- backend/AGENTS.md | 3 +- backend/docs/README.md | 2 + backend/docs/RUN_EVENT_STREAM.md | 179 ++++++ .../safety_finish_reason_middleware.py | 5 +- .../skill_activation_middleware.py | 12 +- .../packages/harness/deerflow/constants.py | 11 + .../harness/deerflow/guardrails/middleware.py | 5 +- .../deerflow/persistence/models/run_event.py | 7 +- .../deerflow/runtime/events/catalog.py | 113 ++++ .../deerflow/runtime/events/store/base.py | 6 +- .../harness/deerflow/runtime/journal.py | 65 ++- .../harness/deerflow/subagents/step_events.py | 37 +- .../deerflow/workspace_changes/recorder.py | 3 +- .../deerflow/workspace_changes/types.py | 6 +- backend/tests/test_guardrail_middleware.py | 6 +- .../tests/test_run_event_stream_contract.py | 549 ++++++++++++++++++ .../test_safety_finish_reason_middleware.py | 7 +- .../test_skill_request_scoped_secrets.py | 20 + backend/tests/test_slash_skills.py | 6 +- contracts/run_event_stream_contract.json | 455 +++++++++++++++ 20 files changed, 1444 insertions(+), 53 deletions(-) create mode 100644 backend/docs/RUN_EVENT_STREAM.md create mode 100644 backend/packages/harness/deerflow/runtime/events/catalog.py create mode 100644 backend/tests/test_run_event_stream_contract.py create mode 100644 contracts/run_event_stream_contract.json diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 22d66d271..0f84caec7 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -444,6 +444,7 @@ metadata only. - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. +- Run event stream changes must keep producer code, `deerflow/constants.py`, `runtime/events/catalog.py`, `contracts/run_event_stream_contract.json`, `backend/docs/RUN_EVENT_STREAM.md`, and `tests/test_run_event_stream_contract.py` in sync. The dependency-free constants module owns the persisted envelope limits (`event_type` 32 characters, `category` 16) and cross-layer workspace event identity; the catalog owns validated runtime definitions and categories. Dynamic middleware tags are limited to 21 characters after the `middleware:` prefix. The JSON contract owns payload schemas, backend-specific storage semantics, legacy aliases, and compatibility rules; conformance tests require both views and all producer groups to agree. `run.end.content` remains opaque and may retain nested Python values in memory while JSONL/database stores stringify non-JSON nested values, so consumers must not assume backend-identical nested output representations. Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runtime, all other `/api/*` → Gateway REST APIs. @@ -509,7 +510,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti **Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol. **Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.) **Context compaction (#3875 Phase 3, #4039)**: subagents inherit `DeerFlowSummarizationMiddleware` via `build_subagent_runtime_middlewares`, gated on the **same** `summarization.enabled` switch the lead reads (one config covers both chains; trigger/keep/model/prompt come from the shared `summarization` config so they cannot drift). The subagent builder attaches `DurableContextMiddleware` immediately before summarization, using the same skills path/read-tool settings as the lead chain. Compaction stores the generated summary in `ThreadState.summary_text` rather than as a `messages` item; the durable-context wrapper therefore projects it into the next model request as guarded hidden human data. This is required when a message-count keep policy preserves only an assistant tool-call plus its tool results: without the injected summary the next request begins with assistant/tool history and strict OpenAI-compatible providers can reject it. Because `DurableContextMiddleware` inserts a second `SystemMessage(authority_contract)` after the subagent's leading system prompt, the builder also appends `SystemMessageCoalescingMiddleware` innermost (mirroring the lead chain, appended after the optional summarization middleware so it is unconditionally last) to merge every `SystemMessage` into one leading `system_message` — otherwise the durable fix would trade #4039's assistant-first HTTP 400 for a duplicate-system 400 on the same strict backends (#4040). The factory is called with `skip_memory_flush=True` on the subagent path: the lead's `memory_flush_hook` (attached when `memory.enabled`) flushes pre-compaction messages into durable memory keyed by `thread_id`, and subagents share the parent's `thread_id`, so without skipping the hook a subagent's internal turns would pollute the **parent** thread's durable memory. Placement differs from the lead chain (lead appends summarization *before* the guard trio; subagent appends it *after*) — benign because the middleware implements only `before_model` (compaction) with no `after_model`/`consume_stop_reason`, so it cannot disturb the Phase 2 guard-cap stop-reason channel. Compaction rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, which shrinks `len(messages)` below the step-capture cursor mid-run; `capture_new_step_messages` (see Step capture below) resets the cursor to the new tail on contraction so steps appended after the compaction point are not silently dropped. -**Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **History contraction (#3875 Phase 3)**: `capture_new_step_messages` assumes append-only growth, but `DeerFlowSummarizationMiddleware` rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, shrinking `len(messages)` below the cursor mid-run. On contraction (`total < processed_count`) the cursor resets to the new tail; `capture_step_message`'s id/content dedup prevents re-emitting pre-compaction steps, so steps appended after the compaction point are still captured instead of being dropped until `total` overtakes the stale cursor. +**Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `subagent_run_event` rejects malformed chunks that lack a non-empty `task_id`; running chunks additionally require a non-negative integer `message_index` and a message object, so persisted records always satisfy the required lifecycle envelope. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **History contraction (#3875 Phase 3)**: `capture_new_step_messages` assumes append-only growth, but `DeerFlowSummarizationMiddleware` rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, shrinking `len(messages)` below the cursor mid-run. On contraction (`total < processed_count`) the cursor resets to the new tail; `capture_step_message`'s id/content dedup prevents re-emitting pre-compaction steps, so steps appended after the compaction point are still captured instead of being dropped until `total` overtakes the stale cursor. **Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), appends the `tool_search` tool, injects the `` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `McpRoutingMiddleware` (when PR1 routing metadata matches deferred tools) before `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(...)`. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run **Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume. **Checkpoint lineage / stream isolation**: `_aexecute` deliberately omits checkpoint-coordinate keys (`thread_id`, `checkpoint_ns`, `checkpoint_id`, `checkpoint_map`) from the child `RunnableConfig`. LangGraph must inherit those coordinates from the copied parent ContextVar so the delegated graph retains a non-root subgraph namespace; explicitly re-supplying even the same parent `thread_id` starts a new root lineage on LangGraph 1.2.6+ and can route child AI/tool frames into the parent `messages` stream. DeerFlow business components still receive the parent `thread_id` through `runtime.context`, which is the preferred lookup path for sandbox, middleware, and attribution code. Regression coverage in `tests/test_subagent_executor.py::TestSubagentCheckpointLineage` keeps the invocation-contract assertion active on every supported version and version-gates the production-shaped parent-stream test to LangGraph 1.2.6+, where the leak exists. diff --git a/backend/docs/README.md b/backend/docs/README.md index 36b3b573a..c3c334d5e 100644 --- a/backend/docs/README.md +++ b/backend/docs/README.md @@ -19,6 +19,7 @@ This directory contains detailed documentation for the DeerFlow backend. | Document | Description | |----------|-------------| | [STREAMING.md](STREAMING.md) | Token-level streaming design: Gateway vs DeerFlowClient paths, `stream_mode` semantics, per-id dedup | +| [RUN_EVENT_STREAM.md](RUN_EVENT_STREAM.md) | Persisted run event stream contract: envelope, producers, consumers, and known gaps | | [FILE_UPLOAD.md](FILE_UPLOAD.md) | File upload functionality | | [PATH_EXAMPLES.md](PATH_EXAMPLES.md) | Path types and usage examples | | [SANDBOX_MEMORY_PROFILING.md](SANDBOX_MEMORY_PROFILING.md) | Sandbox memory baseline and runtime comparison guide | @@ -54,6 +55,7 @@ docs/ ├── summarization.md # Summarization feature ├── plan_mode_usage.md # Plan mode feature ├── STREAMING.md # Token-level streaming design +├── RUN_EVENT_STREAM.md # Persisted run event stream contract ├── AUTO_TITLE_GENERATION.md # Title generation ├── TITLE_GENERATION_IMPLEMENTATION.md # Title implementation details └── TODO.md # Roadmap and issues diff --git a/backend/docs/RUN_EVENT_STREAM.md b/backend/docs/RUN_EVENT_STREAM.md new file mode 100644 index 000000000..5a1897b05 --- /dev/null +++ b/backend/docs/RUN_EVENT_STREAM.md @@ -0,0 +1,179 @@ +# Run Event Stream + +The run event stream is DeerFlow's append-only record of what happened during +an agent run. Producers write through `RunEventStore`; history, debug, subtask, +memory-audit, and workspace-review consumers read projections of the same rows. + +The machine-readable contract is +`contracts/run_event_stream_contract.json`. Canonical event names and +categories live in `deerflow.runtime.events.catalog`; conformance tests require +the runtime catalog and JSON contract to match exactly. + +## Record Envelope + +Every persisted event has these required fields: + +| Field | Meaning | +| --- | --- | +| `thread_id` | Thread that owns the event. | +| `run_id` | Run that produced the event. | +| `seq` | Store-assigned sequence, strictly increasing within a thread. | +| `event_type` | Fixed event name or documented dynamic pattern. | +| `category` | Consumer-routing bucket. | +| `content` | Event payload, normally a string or JSON object. | +| `metadata` | Filterable or audit metadata. | +| `created_at` | Timezone-aware ISO-8601 timestamp. | + +Backends may return additional fields. `DbRunEventStore`, for example, returns +`user_id` and may add serialization markers such as `content_is_json` to +metadata. Consumers must ignore unknown envelope and metadata fields. + +`event_type` is limited to 32 characters and `category` to 16 characters by the +database schema. Catalog-backed definitions enforce the same limits before +writing so they cannot emit values that only the memory or JSONL store accepts. + +`seq` is thread-global, not run-local. Memory and database stores assign it +monotonically for their supported deployment modes. JSONL only provides this +guarantee within one process; shared multi-process deployments must use the +database store. + +## Categories + +`category="message"` means an event is eligible for a message projection; it +does not guarantee that the row is visible in the UI. Thread-history APIs also +filter middleware model calls and superseded regenerate runs, and the frontend +honors message-level visibility markers such as `hide_from_ui`. + +All other categories are excluded from message projections and are available +through run-event or specialized APIs: + +| Category | Purpose | +| --- | --- | +| `trace` | Execution evidence. | +| `outputs` | Root graph completion output. | +| `error` | Callback-observed failure evidence. | +| `middleware` | Middleware state-change audit evidence. | +| `context` | Effective hidden-context identity. | +| `subagent` | Subagent lifecycle and step history. | +| `workspace` | Workspace/output file-change evidence. | + +## Producers + +`RunJournal` emits callback-derived events: + +| Event type | Category | Producer | +| --- | --- | --- | +| `run.start` | `trace` | Root `on_chain_start()` | +| `run.end` | `outputs` | Root `on_chain_end()` | +| `run.error` | `error` | `on_chain_error()` | +| `llm.human.input` | `message` | First persisted lead-agent human input | +| `llm.ai.response` | `message` | `on_llm_end()` | +| `llm.tool.result` | `message` | `on_tool_end()` | +| `llm.error` | `trace` | `on_llm_error()` | +| `context:memory` | `context` | `record_memory_context()` | +| `middleware:{tag}` | `middleware` | `record_middleware()` | + +Current middleware tags are `guardrail`, `safety_termination`, +`skill_activation`, and `skill_secrets`. 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. + +### Opaque Run Outputs + +`run.end.content` is the root graph output and is intentionally opaque. Its +nested representation is not currently identical across storage backends: + +- `MemoryRunEventStore` retains the original Python container and nested + values. +- `JsonlRunEventStore` and `DbRunEventStore` serialize through + `json.dumps(default=str)`, so nested values that are not directly JSON + serializable are read back as strings. + +Consumers may use `run.end` as completion evidence, but must not depend on +backend-identical nested output values. Normalizing those values would be a +separate runtime compatibility change rather than part of this current-state +contract. + +`subagents/step_events.py::subagent_run_event()` maps streamed `task_*` chunks +to persisted events. The worker batches them through `put_batch()`: + +| Event type | Source chunk | Required content | +| --- | --- | --- | +| `subagent.start` | `task_started` | `task_id`, `description` | +| `subagent.step` | `task_running` | `task_id`, `message_index`, `kind`, `text`, `truncated`; AI steps add `tool_calls`, tool steps add `tool_name` | +| `subagent.end` | terminal `task_*` | `task_id`, `status`; optional model, usage, result/error, and truncation fields | + +Terminal subagent status is one of `completed`, `failed`, `cancelled`, or +`timed_out`. + +Malformed lifecycle chunks are not persisted. Every chunk requires a non-empty +string `task_id`; `task_running` additionally requires a non-negative integer +`message_index` and a message object. + +`workspace_changes.record_workspace_changes()` writes `workspace_changes` in +category `workspace` when a run changed files. Its string content is a summary; +the structured versioned summary, file list, and limits live in +`metadata.workspace_changes`. + +The JSON contract defines required and optional payload fields using JSON +Schema. It is the authoritative field-level reference. + +## Consumers + +| Consumer | Read path and behavior | +| --- | --- | +| Frontend thread history | `GET /api/threads/{thread_id}/messages/page` scans `list_messages()`, removes middleware rows and superseded regenerate runs, then applies frontend message visibility rules. | +| Per-run message clients | Thread-scoped and stateless run message endpoints call `list_messages_by_run()`. | +| Run debug/audit | `GET /api/threads/{thread_id}/runs/{run_id}/events` calls `list_events()` and supports `event_types`, `task_id`, `limit`, and `after_seq`. | +| Historical subtask cards | Fetch `subagent.step` through the run-events endpoint, filtered and paginated by `task_id`. | +| Memory audit | Filters run events to `context:memory` and compares `content_sha256`; full memory text is not duplicated into the event store. | +| Workspace review | `GET /api/threads/{thread_id}/runs/{run_id}/workspace-changes` projects the latest `workspace_changes` payload. | + +Token and cost summaries are not reconstructed by reading event rows. +`RunJournal` accumulates usage while callbacks fire, and the worker writes the +aggregates to `RunRow`. + +External Langfuse/LangSmith tracing is a parallel callback pipeline, not a +`RunEventStore` consumer. It is correlated through trace metadata rather than +being derived from these rows. + +Evaluation consumers discussed in #4243 are planned rather than present in +this tree. They should read evidence through `list_events()` and treat the +compatibility and terminal-state limits below as part of that integration. + +## Compatibility + +The existing mixture of dot-separated, colon-separated, and bare-word names is +frozen. This contract documents current behavior; it does not normalize names. +A rename, removal, category change, required-field removal, or required-field +type change is breaking and needs an explicit versioned migration or dual-write +period. + +Adding a new event type or optional field is additive. Consumers must ignore +unknown event types and unknown optional fields. Producers must add a catalog +entry, update the JSON contract and this document, and extend the conformance +tests in the same change. + +`ai_message` is a read-only legacy alias for `llm.ai.response`. Current +producers never emit it. Category-based message projections and store queries +for the last visible AI message recognize previously persisted alias rows, so +the `/messages/page` endpoint also attaches feedback correctly. The legacy +`/messages` endpoint still returns those rows but only enriches feedback for the +canonical name. Legacy aliases live outside the canonical catalog and must not +be used by new producers. + +## Known Gaps + +- Tool-call intent is embedded in `llm.ai.response.content.tool_calls`; it is + not a first-class event. A missing or timed-out result may have no dedicated + outcome event. +- `run.end.metadata.status` is only a root graph completion marker and is + always `success`. `RunRow.status` remains authoritative for lifecycle state, + and worker loss may leave no terminal event. +- 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. +- Loop detection and deferred-tool promotion do not currently emit middleware + events. +- Journal attribution, token accounting, and external tracing metadata still + depend on manual instrumentation at several LLM call sites. diff --git a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py index e3a3d00ef..4a57588ea 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py @@ -61,6 +61,7 @@ from deerflow.agents.middlewares.safety_termination_detectors import ( default_detectors, ) from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls +from deerflow.runtime.events.catalog import MIDDLEWARE_SAFETY_TERMINATION_TAG from deerflow.utils.messages import message_content_to_text if TYPE_CHECKING: @@ -273,7 +274,7 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]): try: journal.record_middleware( - tag="safety_termination", + tag=MIDDLEWARE_SAFETY_TERMINATION_TAG, name=type(self).__name__, hook="after_model", action="suppress_tool_calls", @@ -281,7 +282,7 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]): ) except Exception: # noqa: BLE001 # Audit-event persistence must never break agent execution. - logger.debug("Failed to record middleware:safety_termination event", exc_info=True) + logger.warning("Failed to record middleware:safety_termination event", exc_info=True) # ----- main apply ------------------------------------------------------ diff --git a/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py index 57fa7459b..57cdcaeb6 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py @@ -17,6 +17,10 @@ from langchain.agents.middleware import AgentMiddleware from langchain.agents.middleware.types import ModelRequest, ModelResponse from langchain_core.messages import AIMessage, HumanMessage +from deerflow.runtime.events.catalog import ( + MIDDLEWARE_SKILL_ACTIVATION_TAG, + MIDDLEWARE_SKILL_SECRETS_TAG, +) from deerflow.runtime.secret_context import ( _SECRETS_BINDING_AUDIT_KEY, _SLASH_SKILL_ACTIVATION_RUN_KEY, @@ -294,7 +298,7 @@ Follow this skill before choosing a general workflow. Load supporting resources return try: journal.record_middleware( - "skill_activation", + MIDDLEWARE_SKILL_ACTIVATION_TAG, name="SkillActivationMiddleware", hook=hook, action="activate", @@ -306,7 +310,7 @@ Follow this skill before choosing a general workflow. Load supporting resources }, ) except Exception: - logger.debug("Failed to record slash skill activation audit event", exc_info=True) + logger.warning("Failed to record slash skill activation audit event", exc_info=True) def _prepare_model_request(self, request: ModelRequest, *, hook: str) -> tuple[ModelRequest | AIMessage | None, _Activation | None]: run_context = self._run_context(request) @@ -533,14 +537,14 @@ Follow this skill before choosing a general workflow. Load supporting resources return try: journal.record_middleware( - "skill_secrets", + MIDDLEWARE_SKILL_SECRETS_TAG, name="SkillActivationMiddleware", hook=hook, action="bind_secrets", changes=audit_state, ) except Exception: - logger.debug("Failed to record skill secret binding audit event", exc_info=True) + logger.warning("Failed to record skill secret binding audit event", exc_info=True) @staticmethod def _make_activation_message(target: HumanMessage, activation_content: str) -> HumanMessage: diff --git a/backend/packages/harness/deerflow/constants.py b/backend/packages/harness/deerflow/constants.py index c97abaa0f..b8974d6e8 100644 --- a/backend/packages/harness/deerflow/constants.py +++ b/backend/packages/harness/deerflow/constants.py @@ -8,3 +8,14 @@ DEFAULT_SKILLS_CONTAINER_PATH = "/mnt/skills" # the browser tools (which write here) and the scanner (which ignores it) import # this single source of truth so the name cannot drift between them. BROWSER_FRAMES_DIRNAME = ".browser-frames" + +# Persisted run-event envelope limits. Runtime definitions and the ORM both +# import these from this dependency-free module so lower layers never need to +# initialize deerflow.runtime just to validate storage constraints. +RUN_EVENT_TYPE_MAX_LENGTH = 32 +RUN_EVENT_CATEGORY_MAX_LENGTH = 16 + +# Workspace changes are produced below the runtime layer, so their persisted +# event identity also lives here rather than in the runtime event catalog. +WORKSPACE_CHANGES_EVENT_TYPE = "workspace_changes" +WORKSPACE_CHANGES_EVENT_CATEGORY = "workspace" diff --git a/backend/packages/harness/deerflow/guardrails/middleware.py b/backend/packages/harness/deerflow/guardrails/middleware.py index 636042251..a8b650695 100644 --- a/backend/packages/harness/deerflow/guardrails/middleware.py +++ b/backend/packages/harness/deerflow/guardrails/middleware.py @@ -14,6 +14,7 @@ from langgraph.types import Command from deerflow.authz.principal import normalize_authz_attributes from deerflow.guardrails.provider import GuardrailDecision, GuardrailProvider, GuardrailReason, GuardrailRequest +from deerflow.runtime.events.catalog import MIDDLEWARE_GUARDRAIL_TAG logger = logging.getLogger(__name__) @@ -112,14 +113,14 @@ class GuardrailMiddleware(AgentMiddleware[AgentState]): try: journal.record_middleware( - tag="guardrail", + tag=MIDDLEWARE_GUARDRAIL_TAG, name=type(self).__name__, hook="wrap_tool_call", action=action, changes=changes, ) except Exception: # noqa: BLE001 - logger.debug("Failed to record middleware:guardrail event", exc_info=True) + logger.warning("Failed to record middleware:guardrail event", exc_info=True) @override def wrap_tool_call( diff --git a/backend/packages/harness/deerflow/persistence/models/run_event.py b/backend/packages/harness/deerflow/persistence/models/run_event.py index 4f22b4616..e0cc51974 100644 --- a/backend/packages/harness/deerflow/persistence/models/run_event.py +++ b/backend/packages/harness/deerflow/persistence/models/run_event.py @@ -7,6 +7,7 @@ from datetime import UTC, datetime from sqlalchemy import JSON, DateTime, Index, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column +from deerflow.constants import RUN_EVENT_CATEGORY_MAX_LENGTH, RUN_EVENT_TYPE_MAX_LENGTH from deerflow.persistence.base import Base @@ -20,9 +21,9 @@ class RunEventRow(Base): # created before auth was introduced; populated by auth middleware on # new writes and by the boot-time orphan migration on existing rows. user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) - event_type: Mapped[str] = mapped_column(String(32), nullable=False) - category: Mapped[str] = mapped_column(String(16), nullable=False) - # "message" | "trace" | "lifecycle" + event_type: Mapped[str] = mapped_column(String(RUN_EVENT_TYPE_MAX_LENGTH), nullable=False) + category: Mapped[str] = mapped_column(String(RUN_EVENT_CATEGORY_MAX_LENGTH), nullable=False) + # Category values and semantics are defined by runtime/events/catalog.py content: Mapped[str] = mapped_column(Text, default="") event_metadata: Mapped[dict] = mapped_column(JSON, default=dict) seq: Mapped[int] = mapped_column(nullable=False) diff --git a/backend/packages/harness/deerflow/runtime/events/catalog.py b/backend/packages/harness/deerflow/runtime/events/catalog.py new file mode 100644 index 000000000..06c6ba1a9 --- /dev/null +++ b/backend/packages/harness/deerflow/runtime/events/catalog.py @@ -0,0 +1,113 @@ +"""Canonical names and categories for persisted run events. + +Producers import these definitions instead of repeating event-name/category +pairs. The public JSON contract is checked against this catalog in backend +tests, so either side changing without the other fails CI. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from deerflow.constants import ( + RUN_EVENT_CATEGORY_MAX_LENGTH, + RUN_EVENT_TYPE_MAX_LENGTH, + WORKSPACE_CHANGES_EVENT_CATEGORY, + WORKSPACE_CHANGES_EVENT_TYPE, +) + + +def _validate_category(category: str) -> None: + if not category: + raise ValueError("Run event category must not be empty") + if len(category) > RUN_EVENT_CATEGORY_MAX_LENGTH: + raise ValueError(f"Run event category must not exceed {RUN_EVENT_CATEGORY_MAX_LENGTH} characters: {category!r}") + + +@dataclass(frozen=True, slots=True) +class RunEventDefinition: + event_type: str + category: str + + def __post_init__(self) -> None: + if not self.event_type: + raise ValueError("Run event type must not be empty") + if len(self.event_type) > RUN_EVENT_TYPE_MAX_LENGTH: + raise ValueError(f"Run event type must not exceed {RUN_EVENT_TYPE_MAX_LENGTH} characters: {self.event_type!r}") + _validate_category(self.category) + + +@dataclass(frozen=True, slots=True) +class RunEventPattern: + pattern: str + prefix: str + category: str + + def __post_init__(self) -> None: + _validate_category(self.category) + + def event_type(self, suffix: str) -> str: + if not suffix: + raise ValueError("Run event suffix must not be empty") + max_suffix_length = RUN_EVENT_TYPE_MAX_LENGTH - len(self.prefix) + if len(suffix) > max_suffix_length: + raise ValueError(f"Run event suffix for {self.pattern!r} must not exceed {max_suffix_length} characters: {suffix!r}") + return f"{self.prefix}{suffix}" + + +RUN_START_EVENT = RunEventDefinition("run.start", "trace") +RUN_END_EVENT = RunEventDefinition("run.end", "outputs") +RUN_ERROR_EVENT = RunEventDefinition("run.error", "error") +LLM_HUMAN_INPUT_EVENT = RunEventDefinition("llm.human.input", "message") +LLM_AI_RESPONSE_EVENT = RunEventDefinition("llm.ai.response", "message") +LLM_TOOL_RESULT_EVENT = RunEventDefinition("llm.tool.result", "message") +LLM_ERROR_EVENT = RunEventDefinition("llm.error", "trace") +MEMORY_CONTEXT_EVENT = RunEventDefinition("context:memory", "context") + +SUBAGENT_START_EVENT = RunEventDefinition("subagent.start", "subagent") +SUBAGENT_STEP_EVENT = RunEventDefinition("subagent.step", "subagent") +SUBAGENT_END_EVENT = RunEventDefinition("subagent.end", "subagent") + +WORKSPACE_CHANGES_EVENT = RunEventDefinition(WORKSPACE_CHANGES_EVENT_TYPE, WORKSPACE_CHANGES_EVENT_CATEGORY) + +MIDDLEWARE_EVENT_PATTERN = RunEventPattern( + pattern="middleware:{tag}", + prefix="middleware:", + category="middleware", +) +MIDDLEWARE_EVENT_TAG_MAX_LENGTH = RUN_EVENT_TYPE_MAX_LENGTH - len(MIDDLEWARE_EVENT_PATTERN.prefix) +MIDDLEWARE_GUARDRAIL_TAG = "guardrail" +MIDDLEWARE_SAFETY_TERMINATION_TAG = "safety_termination" +MIDDLEWARE_SKILL_ACTIVATION_TAG = "skill_activation" +MIDDLEWARE_SKILL_SECRETS_TAG = "skill_secrets" +MIDDLEWARE_EVENT_TAGS = ( + MIDDLEWARE_GUARDRAIL_TAG, + MIDDLEWARE_SAFETY_TERMINATION_TAG, + MIDDLEWARE_SKILL_ACTIVATION_TAG, + MIDDLEWARE_SKILL_SECRETS_TAG, +) + +JOURNAL_RUN_EVENT_DEFINITIONS = ( + RUN_START_EVENT, + RUN_END_EVENT, + RUN_ERROR_EVENT, + LLM_HUMAN_INPUT_EVENT, + LLM_AI_RESPONSE_EVENT, + LLM_TOOL_RESULT_EVENT, + LLM_ERROR_EVENT, + MEMORY_CONTEXT_EVENT, +) + +SUBAGENT_RUN_EVENT_DEFINITIONS = ( + SUBAGENT_START_EVENT, + SUBAGENT_STEP_EVENT, + SUBAGENT_END_EVENT, +) + +WORKSPACE_RUN_EVENT_DEFINITIONS = (WORKSPACE_CHANGES_EVENT,) + +FIXED_RUN_EVENT_DEFINITIONS = ( + *JOURNAL_RUN_EVENT_DEFINITIONS, + *SUBAGENT_RUN_EVENT_DEFINITIONS, + *WORKSPACE_RUN_EVENT_DEFINITIONS, +) diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index 72757d33a..af4858310 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -6,7 +6,8 @@ through the same interface, distinguished by the ``category`` field. Implementations: - MemoryRunEventStore: in-memory dict (development, tests) -- Future: DB-backed store (SQLAlchemy ORM), JSONL file store +- DbRunEventStore: SQLAlchemy ORM-backed persistence +- JsonlRunEventStore: JSONL file persistence for local/debug use """ from __future__ import annotations @@ -24,7 +25,8 @@ class RunEventStore(abc.ABC): 2. seq is strictly increasing within the same thread 3. list_messages() only returns category="message" events 4. list_events() returns all events for the specified run - 5. Returned dicts match the RunEvent field structure + 5. Returned dicts contain the required RunEvent envelope fields; backends + may add documented fields such as DbRunEventStore.user_id """ @abc.abstractmethod diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 9c0cdb57a..bf5706a33 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -6,11 +6,11 @@ handles token usage accumulation. Key design decisions: - on_llm_new_token is NOT implemented -- only complete messages via on_llm_end -- on_chat_model_start captures structured prompts as llm_request (OpenAI format) and +- on_chat_model_start captures the first user-visible prompt as llm.human.input and extracts the first human message for run.input, because it is more reliable than on_chain_start (fires on every node) — messages here are fully structured. - on_chain_start with parent_run_id=None emits a run.start trace marking root invocation. -- on_llm_end emits llm_response in OpenAI Chat Completions format +- on_llm_end emits llm.ai.response in checkpoint-aligned AIMessage.model_dump() format - Token usage accumulated in memory, written to RunRow on run completion - Caller identification via tags injection (lead_agent / subagent:{name} / middleware:{name}) """ @@ -30,6 +30,17 @@ from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMes from langgraph.types import Command from deerflow.agents.human_input import read_human_input_response +from deerflow.runtime.events.catalog import ( + LLM_AI_RESPONSE_EVENT, + LLM_ERROR_EVENT, + LLM_HUMAN_INPUT_EVENT, + LLM_TOOL_RESULT_EVENT, + MEMORY_CONTEXT_EVENT, + MIDDLEWARE_EVENT_PATTERN, + RUN_END_EVENT, + RUN_ERROR_EVENT, + RUN_START_EVENT, +) from deerflow.utils.messages import message_to_text, restore_original_human_message if TYPE_CHECKING: @@ -252,8 +263,8 @@ class RunJournal(BaseCallbackHandler): # Root graph invocation — emit a single trace event for the run start. chain_name = (serialized or {}).get("name", "unknown") self._put( - event_type="run.start", - category="trace", + event_type=RUN_START_EVENT.event_type, + category=RUN_START_EVENT.category, content={"chain": chain_name}, metadata={"caller": caller, **(metadata or {})}, ) @@ -271,13 +282,18 @@ class RunJournal(BaseCallbackHandler): if parent_run_id is not None: return self._reconcile_final_tool_messages(outputs) - self._put(event_type="run.end", category="outputs", content=outputs, metadata={"status": "success"}) + self._put( + event_type=RUN_END_EVENT.event_type, + category=RUN_END_EVENT.category, + content=outputs, + metadata={"status": "success"}, + ) self._flush_sync() def on_chain_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: self._put( - event_type="run.error", - category="error", + event_type=RUN_ERROR_EVENT.event_type, + category=RUN_ERROR_EVENT.category, content=str(error), metadata={"error_type": type(error).__name__}, ) @@ -294,7 +310,7 @@ class RunJournal(BaseCallbackHandler): tags: list[str] | None = None, **kwargs: Any, ) -> None: - """Capture structured prompt messages for llm_request event. + """Capture the first user-visible prompt as llm.human.input. This is also the canonical place to extract the first human message: messages are fully structured here, it fires only on real LLM calls, @@ -322,8 +338,8 @@ class RunJournal(BaseCallbackHandler): persisted_message = restore_original_human_message(m) self.set_first_human_message(self._message_text(persisted_message)) self._put( - event_type="llm.human.input", - category="message", + event_type=LLM_HUMAN_INPUT_EVENT.event_type, + category=LLM_HUMAN_INPUT_EVENT.category, content=persisted_message.model_dump(), metadata={"caller": caller}, ) @@ -387,10 +403,10 @@ class RunJournal(BaseCallbackHandler): call_index = self._llm_call_index self._seen_llm_starts.add(rid) - # Trace event: llm_response (OpenAI completion format) + # Message event: checkpoint-aligned llm.ai.response payload. self._put( - event_type="llm.ai.response", - category="message", + event_type=LLM_AI_RESPONSE_EVENT.event_type, + category=LLM_AI_RESPONSE_EVENT.category, content=message.model_dump(), metadata={ "caller": caller, @@ -438,7 +454,11 @@ class RunJournal(BaseCallbackHandler): def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: self._llm_start_times.pop(str(run_id), None) - self._put(event_type="llm.error", category="trace", content=str(error)) + self._put( + event_type=LLM_ERROR_EVENT.event_type, + category=LLM_ERROR_EVENT.category, + content=str(error), + ) def on_tool_start(self, serialized, input_str, *, run_id, parent_run_id=None, tags=None, metadata=None, inputs=None, **kwargs): """Handle tool start event, cache tool call ID for later correlation""" @@ -499,7 +519,11 @@ class RunJournal(BaseCallbackHandler): self._current_run_tool_call_names[tool_call_id] = str(name or "") def _persist_tool_result_message(self, message: BaseMessage) -> None: - self._put(event_type="llm.tool.result", category="message", content=message.model_dump()) + self._put( + event_type=LLM_TOOL_RESULT_EVENT.event_type, + category=LLM_TOOL_RESULT_EVENT.category, + content=message.model_dump(), + ) identity = self._message_identity(message) if identity: self._persisted_tool_message_identities.add(identity) @@ -715,15 +739,16 @@ class RunJournal(BaseCallbackHandler): Args: tag: Short identifier for the middleware (e.g., "title", "summarize", - "guardrail"). Used to form event_type="middleware:{tag}". + "guardrail"). Used to form event_type="middleware:{tag}" and + limited by the persisted event-type column width. name: Full middleware class name. hook: Lifecycle hook that triggered the action (e.g., "after_model"). action: Specific action performed (e.g., "generate_title"). changes: Dict describing the state changes made. """ self._put( - event_type=f"middleware:{tag}", - category="middleware", + event_type=MIDDLEWARE_EVENT_PATTERN.event_type(tag), + category=MIDDLEWARE_EVENT_PATTERN.category, content={"name": name, "hook": hook, "action": action, "changes": changes}, ) @@ -738,8 +763,8 @@ class RunJournal(BaseCallbackHandler): if self._memory_context_recorded: return self._put( - event_type="context:memory", - category="context", + event_type=MEMORY_CONTEXT_EVENT.event_type, + category=MEMORY_CONTEXT_EVENT.category, content={"content_sha256": content_sha256}, ) self._memory_context_recorded = True diff --git a/backend/packages/harness/deerflow/subagents/step_events.py b/backend/packages/harness/deerflow/subagents/step_events.py index 9bc9f16ec..fcce512e1 100644 --- a/backend/packages/harness/deerflow/subagents/step_events.py +++ b/backend/packages/harness/deerflow/subagents/step_events.py @@ -23,6 +23,11 @@ from typing import Any from langchain_core.messages import AIMessage, BaseMessage, ToolMessage +from deerflow.runtime.events.catalog import ( + SUBAGENT_END_EVENT, + SUBAGENT_START_EVENT, + SUBAGENT_STEP_EVENT, +) from deerflow.utils.messages import message_content_to_text from .status_contract import normalize_token_usage @@ -36,7 +41,7 @@ SUBAGENT_STEP_MAX_CHARS = 8192 #: ``RunEvent.category`` for persisted subagent steps. A dedicated category (not #: ``"message"``) keeps these events out of ``list_messages`` (the thread message #: feed) while still being returned by ``list_events`` for fetch-on-expand (#3779). -SUBAGENT_EVENT_CATEGORY = "subagent" +SUBAGENT_EVENT_CATEGORY = SUBAGENT_START_EVENT.category #: Map of ``task_*`` terminal custom-event types to their persisted status. _TERMINAL_EVENT_STATUS: dict[str, str] = { @@ -193,8 +198,8 @@ def subagent_run_event(chunk: Any) -> dict[str, Any] | None: Returns the ``event_type`` / ``category`` / ``content`` / ``metadata`` for a persistable subagent lifecycle event, or ``None`` for any chunk that is not a - subagent event (so the worker only persists what it recognizes). ``thread_id`` - / ``run_id`` are filled in by the caller. + valid subagent event (so the worker only persists what it recognizes). + ``thread_id`` / ``run_id`` are filled in by the caller. """ if not isinstance(chunk, dict): return None @@ -204,21 +209,31 @@ def subagent_run_event(chunk: Any) -> dict[str, Any] | None: return None task_id = chunk.get("task_id") + if not isinstance(task_id, str) or not task_id: + return None if event == "task_started": + description = chunk.get("description") + if description is not None and not isinstance(description, str): + return None return { - "event_type": "subagent.start", - "category": SUBAGENT_EVENT_CATEGORY, - "content": {"task_id": task_id, "description": chunk.get("description")}, + "event_type": SUBAGENT_START_EVENT.event_type, + "category": SUBAGENT_START_EVENT.category, + "content": {"task_id": task_id, "description": description}, "metadata": {"task_id": task_id}, } if event == "task_running": message_index = chunk.get("message_index") + message = chunk.get("message") + if isinstance(message_index, bool) or not isinstance(message_index, int) or message_index < 0: + return None + if not isinstance(message, dict): + return None return { - "event_type": "subagent.step", - "category": SUBAGENT_EVENT_CATEGORY, - "content": build_subagent_step(chunk.get("message") or {}, task_id=task_id, message_index=message_index), + "event_type": SUBAGENT_STEP_EVENT.event_type, + "category": SUBAGENT_STEP_EVENT.category, + "content": build_subagent_step(message, task_id=task_id, message_index=message_index), "metadata": {"task_id": task_id, "message_index": message_index}, } @@ -245,8 +260,8 @@ def subagent_run_event(chunk: Any) -> dict[str, Any] | None: if error_truncated: content["error_truncated"] = True return { - "event_type": "subagent.end", - "category": SUBAGENT_EVENT_CATEGORY, + "event_type": SUBAGENT_END_EVENT.event_type, + "category": SUBAGENT_END_EVENT.category, "content": content, "metadata": {"task_id": task_id}, } diff --git a/backend/packages/harness/deerflow/workspace_changes/recorder.py b/backend/packages/harness/deerflow/workspace_changes/recorder.py index 2607197b0..c6502e8b5 100644 --- a/backend/packages/harness/deerflow/workspace_changes/recorder.py +++ b/backend/packages/harness/deerflow/workspace_changes/recorder.py @@ -12,6 +12,7 @@ from deerflow.config import get_paths from .diff import compare_snapshots, get_changed_paths from .scanner import scan_workspace_roots from .types import ( + WORKSPACE_CHANGES_EVENT_CATEGORY, WORKSPACE_CHANGES_EVENT_TYPE, WORKSPACE_CHANGES_METADATA_KEY, WorkspaceChangeLimits, @@ -153,7 +154,7 @@ async def record_workspace_changes( thread_id=thread_id, run_id=run_id, event_type=WORKSPACE_CHANGES_EVENT_TYPE, - category="workspace", + category=WORKSPACE_CHANGES_EVENT_CATEGORY, content=content, metadata={WORKSPACE_CHANGES_METADATA_KEY: payload}, ) diff --git a/backend/packages/harness/deerflow/workspace_changes/types.py b/backend/packages/harness/deerflow/workspace_changes/types.py index 4fd25d221..fa88d23b2 100644 --- a/backend/packages/harness/deerflow/workspace_changes/types.py +++ b/backend/packages/harness/deerflow/workspace_changes/types.py @@ -4,7 +4,11 @@ from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Literal -WORKSPACE_CHANGES_EVENT_TYPE = "workspace_changes" +from deerflow.constants import ( + WORKSPACE_CHANGES_EVENT_CATEGORY as WORKSPACE_CHANGES_EVENT_CATEGORY, +) +from deerflow.constants import WORKSPACE_CHANGES_EVENT_TYPE as WORKSPACE_CHANGES_EVENT_TYPE + WORKSPACE_CHANGES_METADATA_KEY = "workspace_changes" WorkspaceChangeStatus = Literal["created", "modified", "deleted", "symlink_created"] diff --git a/backend/tests/test_guardrail_middleware.py b/backend/tests/test_guardrail_middleware.py index 5a60bef51..c38c3a45b 100644 --- a/backend/tests/test_guardrail_middleware.py +++ b/backend/tests/test_guardrail_middleware.py @@ -431,17 +431,19 @@ class TestGuardrailMiddleware: assert journal.calls == [] # Journal: a recording failure must not alter the guardrail denial outcome. - def test_guardrail_event_recording_failure_does_not_change_denial(self): + def test_guardrail_event_recording_failure_warns_without_changing_denial(self, caplog): journal = _FakeJournal(fail=True) mw = GuardrailMiddleware(_DenyAllProvider()) req = _make_tool_call_request("bash", context={"__run_journal": journal}) handler = MagicMock() - result = mw.wrap_tool_call(req, handler) + with caplog.at_level("WARNING"): + result = mw.wrap_tool_call(req, handler) handler.assert_not_called() assert result.status == "error" assert "oap.denied" in result.content + assert "Failed to record middleware:guardrail event" in caplog.text # Journal: the async denial path records the same guardrail audit event. def test_async_denied_tool_records_guardrail_event(self): diff --git a/backend/tests/test_run_event_stream_contract.py b/backend/tests/test_run_event_stream_contract.py new file mode 100644 index 000000000..70cd2c9c5 --- /dev/null +++ b/backend/tests/test_run_event_stream_contract.py @@ -0,0 +1,549 @@ +"""Conformance tests for the documented run event stream contract.""" + +from __future__ import annotations + +import ast +import json +from pathlib import Path +from uuid import uuid4 + +import pytest +from jsonschema import Draft202012Validator, FormatChecker +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langchain_core.outputs import ChatGeneration, LLMResult + +from deerflow.runtime.events.catalog import ( + FIXED_RUN_EVENT_DEFINITIONS, + JOURNAL_RUN_EVENT_DEFINITIONS, + MIDDLEWARE_EVENT_PATTERN, + MIDDLEWARE_EVENT_TAG_MAX_LENGTH, + MIDDLEWARE_EVENT_TAGS, + RUN_EVENT_CATEGORY_MAX_LENGTH, + RUN_EVENT_TYPE_MAX_LENGTH, + SUBAGENT_RUN_EVENT_DEFINITIONS, + WORKSPACE_RUN_EVENT_DEFINITIONS, + RunEventDefinition, + RunEventPattern, +) +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import RunJournal +from deerflow.subagents.step_events import SUBAGENT_STEP_MAX_CHARS, capture_step_message, subagent_run_event + +REPO_ROOT = Path(__file__).resolve().parents[2] +CONTRACT_PATH = REPO_ROOT / "contracts" / "run_event_stream_contract.json" + + +def _load_contract() -> dict: + return json.loads(CONTRACT_PATH.read_text(encoding="utf-8")) + + +def _contract_events() -> dict[str, dict]: + return {event["event_type"]: event for event in _load_contract()["events"]} + + +def _assert_schema_valid(schema: dict | bool, instance: object) -> None: + Draft202012Validator.check_schema(schema) + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + errors = sorted(validator.iter_errors(instance), key=lambda error: list(error.path)) + assert not errors, "; ".join(error.message for error in errors) + + +def _assert_fixed_event_valid(event: dict, *, persisted: bool = False) -> None: + contract_event = _contract_events()[event["event_type"]] + assert event["category"] == contract_event["category"] + _assert_schema_valid(contract_event["content_schema"], event["content"]) + _assert_schema_valid(contract_event["metadata_schema"], event.get("metadata", {})) + if persisted: + _assert_schema_valid(_load_contract()["record_schema"], event) + + +def _make_llm_response(content: str = "answer", usage: dict | None = None) -> LLMResult: + message = AIMessage( + content=content, + id=f"msg-{uuid4()}", + response_metadata={"model_name": "test-model"}, + usage_metadata=usage, + ) + return LLMResult(generations=[[ChatGeneration(message=message)]]) + + +def _subagent_batch() -> list[dict]: + chunks = [ + {"type": "task_started", "task_id": "call-batch", "description": "research"}, + { + "type": "task_running", + "task_id": "call-batch", + "message": { + "type": "ai", + "content": "searching", + "tool_calls": [{"name": "web_search", "args": {"query": "deerflow"}}], + }, + "message_index": 1, + }, + {"type": "task_completed", "task_id": "call-batch", "result": "done"}, + ] + events = [subagent_run_event(chunk) for chunk in chunks] + assert all(event is not None for event in events) + return [{"thread_id": "thread-batch", "run_id": "run-batch", **event} for event in events if event is not None] + + +async def _persist_subagent_batch(store) -> list[dict]: + await store.put_batch(_subagent_batch()) + return await store.list_events("thread-batch", "run-batch") + + +async def _record_run_end(store) -> dict: + journal = RunJournal("run-output", "thread-output", store, flush_threshold=100) + journal.on_chain_end( + {"messages": [AIMessage(content="final answer", id="final-message")]}, + run_id=uuid4(), + parent_run_id=None, + ) + await journal.flush() + events = await store.list_events("thread-output", "run-output", event_types=["run.end"]) + assert len(events) == 1 + return events[0] + + +def test_contract_and_runtime_catalog_have_the_same_fixed_events(): + contract = _load_contract() + contract_types = [event["event_type"] for event in contract["events"]] + runtime_types = [definition.event_type for definition in FIXED_RUN_EVENT_DEFINITIONS] + contract_pairs = {(event["event_type"], event["category"]) for event in contract["events"]} + runtime_pairs = {(definition.event_type, definition.category) for definition in FIXED_RUN_EVENT_DEFINITIONS} + + assert len(set(contract_types)) == len(contract_types) + assert len(set(runtime_types)) == len(runtime_types) + assert contract_pairs == runtime_pairs + assert set(contract["categories"]) == {definition.category for definition in FIXED_RUN_EVENT_DEFINITIONS} | {MIDDLEWARE_EVENT_PATTERN.category} + + event_type_schema = contract["record_schema"]["properties"]["event_type"] + category_schema = contract["record_schema"]["properties"]["category"] + middleware_pattern = contract["dynamic_event_patterns"][0] + assert event_type_schema["maxLength"] == RUN_EVENT_TYPE_MAX_LENGTH + assert category_schema["maxLength"] == RUN_EVENT_CATEGORY_MAX_LENGTH + assert middleware_pattern["event_type_schema"]["maxLength"] == RUN_EVENT_TYPE_MAX_LENGTH + assert middleware_pattern["tag_schema"]["maxLength"] == MIDDLEWARE_EVENT_TAG_MAX_LENGTH + + from deerflow.persistence.models.run_event import RunEventRow + + assert RunEventRow.__table__.c.event_type.type.length == RUN_EVENT_TYPE_MAX_LENGTH + assert RunEventRow.__table__.c.category.type.length == RUN_EVENT_CATEGORY_MAX_LENGTH + + +@pytest.mark.parametrize( + ("definition_type", "kwargs"), + [ + (RunEventDefinition, {"event_type": "test.event"}), + (RunEventPattern, {"pattern": "test:{tag}", "prefix": "test:"}), + ], +) +def test_runtime_catalog_rejects_categories_that_do_not_fit_persistence(definition_type, kwargs): + assert definition_type(category="x" * RUN_EVENT_CATEGORY_MAX_LENGTH, **kwargs).category + + for invalid_category in ("", "x" * (RUN_EVENT_CATEGORY_MAX_LENGTH + 1)): + with pytest.raises(ValueError, match="category"): + definition_type(category=invalid_category, **kwargs) + + +@pytest.mark.parametrize( + "relative_path", + [ + "persistence/models/run_event.py", + "workspace_changes/types.py", + ], +) +def test_lower_level_run_event_modules_do_not_import_runtime(relative_path): + module_path = REPO_ROOT / "backend" / "packages" / "harness" / "deerflow" / relative_path + tree = ast.parse(module_path.read_text(encoding="utf-8"), filename=str(module_path)) + imports = [node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module is not None] + imports.extend(alias.name for node in ast.walk(tree) if isinstance(node, ast.Import) for alias in node.names) + + assert not [module for module in imports if module == "deerflow.runtime" or module.startswith("deerflow.runtime.")] + + +def test_legacy_aliases_are_read_only_and_outside_the_current_catalog(): + contract = _load_contract() + aliases = {alias["event_type"]: alias for alias in contract["legacy_event_aliases"]} + current_types = {definition.event_type for definition in FIXED_RUN_EVENT_DEFINITIONS} + + assert set(aliases) == {"ai_message"} + assert aliases["ai_message"]["canonical_event_type"] == "llm.ai.response" + assert aliases["ai_message"]["produced_by_current_runtime"] is False + assert "/messages/page" in aliases["ai_message"]["compatibility_scope"] + assert "legacy /messages endpoint" in aliases["ai_message"]["known_limitations"] + assert set(aliases).isdisjoint(current_types) + + +def test_record_envelope_accepts_every_json_content_type(): + schema = _load_contract()["record_schema"] + envelope = { + "thread_id": "thread-1", + "run_id": "run-1", + "seq": 1, + "event_type": "run.end", + "category": "outputs", + "metadata": {}, + "created_at": "2026-07-21T00:00:00+00:00", + } + + for content in ("text", {"key": "value"}, ["value"], 1, 1.5, True, None): + _assert_schema_valid(schema, {**envelope, "content": content}) + + +def test_contract_schemas_are_valid_json_schema(): + contract = _load_contract() + Draft202012Validator.check_schema(contract["record_schema"]) + for event in contract["events"]: + Draft202012Validator.check_schema(event["content_schema"]) + Draft202012Validator.check_schema(event["metadata_schema"]) + for pattern in contract["dynamic_event_patterns"]: + Draft202012Validator.check_schema(pattern["event_type_schema"]) + Draft202012Validator.check_schema(pattern["tag_schema"]) + Draft202012Validator.check_schema(pattern["content_schema"]) + Draft202012Validator.check_schema(pattern["metadata_schema"]) + + +@pytest.mark.anyio +@pytest.mark.parametrize("backend", ["memory", "jsonl"]) +async def test_non_database_stores_return_contract_records(backend, tmp_path): + if backend == "memory": + store = MemoryRunEventStore() + else: + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + store = JsonlRunEventStore(base_dir=tmp_path / "events") + + record = await store.put( + thread_id="thread-1", + run_id="run-1", + event_type="context:memory", + category="context", + content={"content_sha256": "a" * 64}, + metadata={}, + ) + + _assert_fixed_event_valid(record, persisted=True) + assert record["seq"] == 1 + + +@pytest.mark.anyio +async def test_database_store_returns_contract_record_with_backend_fields(tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'events.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + store = DbRunEventStore(get_session_factory()) + record = await store.put( + thread_id="thread-1", + run_id="run-1", + event_type="context:memory", + category="context", + content={"content_sha256": "a" * 64}, + metadata={}, + ) + + _assert_fixed_event_valid(record, persisted=True) + assert "user_id" in record + assert record["metadata"]["content_is_json"] is True + finally: + await close_engine() + + +@pytest.mark.anyio +async def test_run_end_backend_storage_semantics_match_contract(tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + contract_event = _contract_events()["run.end"] + assert set(contract_event["storage_semantics"]) == {"memory", "jsonl", "database"} + + memory_event = await _record_run_end(MemoryRunEventStore()) + jsonl_event = await _record_run_end(JsonlRunEventStore(base_dir=tmp_path / "events")) + + url = f"sqlite+aiosqlite:///{tmp_path / 'run-output.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + database_event = await _record_run_end(DbRunEventStore(get_session_factory())) + finally: + await close_engine() + + for event in (memory_event, jsonl_event, database_event): + _assert_fixed_event_valid(event, persisted=True) + + assert isinstance(memory_event["content"]["messages"][0], AIMessage) + assert isinstance(jsonl_event["content"]["messages"][0], str) + assert isinstance(database_event["content"]["messages"][0], str) + assert "final answer" in jsonl_event["content"]["messages"][0] + assert "final answer" in database_event["content"]["messages"][0] + + +@pytest.mark.anyio +async def test_run_journal_observed_events_exactly_match_its_catalog(): + store = MemoryRunEventStore() + journal = RunJournal("run-1", "thread-1", store, flush_threshold=100) + + root_run_id = uuid4() + llm_run_id = uuid4() + journal.on_chain_start( + {"name": "root"}, + {}, + run_id=root_run_id, + parent_run_id=None, + tags=["lead_agent"], + metadata={"langgraph_step": 1}, + ) + journal.on_chat_model_start( + {}, + [[HumanMessage(content="question", id="human-1")]], + run_id=llm_run_id, + tags=["lead_agent"], + ) + journal.on_llm_end( + _make_llm_response("answer", usage={"input_tokens": 3, "output_tokens": 4, "total_tokens": 7}), + run_id=llm_run_id, + parent_run_id=None, + tags=["lead_agent"], + ) + journal.on_tool_end( + ToolMessage(content="tool result", tool_call_id="call-1", name="web_search", id="tool-1"), + run_id=uuid4(), + ) + journal.on_llm_error(RuntimeError("model failed"), run_id=uuid4()) + journal.on_chain_error(ValueError("run failed"), run_id=uuid4()) + journal.on_chain_end({"messages": []}, run_id=root_run_id, parent_run_id=None) + journal.record_memory_context(content_sha256="a" * 64) + await journal.flush() + + events = await store.list_events("thread-1", "run-1") + expected_types = {definition.event_type for definition in JOURNAL_RUN_EVENT_DEFINITIONS} + + assert {event["event_type"] for event in events} == expected_types + for event in events: + _assert_fixed_event_valid(event, persisted=True) + + +@pytest.mark.anyio +@pytest.mark.parametrize("tag", MIDDLEWARE_EVENT_TAGS) +async def test_dynamic_middleware_event_matches_pattern_contract(tag): + store = MemoryRunEventStore() + journal = RunJournal("run-1", "thread-1", store, flush_threshold=100) + journal.record_middleware( + tag, + name="GuardrailMiddleware", + hook="wrap_tool_call", + action="deny", + changes={"reason": "policy"}, + ) + await journal.flush() + + event = (await store.list_events("thread-1", "run-1"))[0] + pattern = _load_contract()["dynamic_event_patterns"][0] + + assert pattern["pattern"] == MIDDLEWARE_EVENT_PATTERN.pattern + assert set(pattern["known_tags"]) == set(MIDDLEWARE_EVENT_TAGS) + assert event["event_type"] == MIDDLEWARE_EVENT_PATTERN.event_type(tag) + assert event["category"] == MIDDLEWARE_EVENT_PATTERN.category == pattern["category"] + _assert_schema_valid(pattern["event_type_schema"], event["event_type"]) + _assert_schema_valid(pattern["tag_schema"], tag) + _assert_schema_valid(pattern["content_schema"], event["content"]) + _assert_schema_valid(pattern["metadata_schema"], event["metadata"]) + + +@pytest.mark.parametrize("tag", ["", "x" * (MIDDLEWARE_EVENT_TAG_MAX_LENGTH + 1)]) +def test_dynamic_middleware_event_rejects_tags_that_do_not_fit_persistence(tag): + journal = RunJournal("run-1", "thread-1", MemoryRunEventStore(), flush_threshold=100) + + with pytest.raises(ValueError): + journal.record_middleware( + tag, + name="CustomMiddleware", + hook="after_model", + action="record", + changes={}, + ) + + +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) + cases = [ + {"type": "task_started", "task_id": "call-1", "description": "research"}, + { + "type": "task_running", + "task_id": "call-1", + "message": { + "type": "ai", + "content": "searching", + "tool_calls": [{"name": "web_search", "args": {"query": "deerflow"}}], + }, + "message_index": 1, + }, + { + "type": "task_running", + "task_id": "call-1", + "message": {"type": "tool", "name": "web_search", "content": "result"}, + "message_index": 2, + }, + { + "type": "task_completed", + "task_id": "call-1", + "result": "done", + "model_name": "test-model", + "usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7}, + }, + {"type": "task_failed", "task_id": "call-2", "error": "boom"}, + {"type": "task_cancelled", "task_id": "call-3"}, + {"type": "task_timed_out", "task_id": "call-4", "error": "timed out"}, + {"type": "task_completed", "task_id": "call-5", "result": long_result}, + {"type": "task_failed", "task_id": "call-6", "error": long_error}, + ] + + records = [subagent_run_event(case) for case in cases] + assert all(record is not None for record in records) + typed_records = [record for record in records if record is not None] + expected_types = {definition.event_type for definition in SUBAGENT_RUN_EVENT_DEFINITIONS} + + assert {record["event_type"] for record in typed_records} == expected_types + for record in typed_records: + _assert_fixed_event_valid(record) + + ai_step, tool_step = typed_records[1]["content"], typed_records[2]["content"] + completed, failed = typed_records[3]["content"], typed_records[4]["content"] + timed_out = typed_records[6]["content"] + truncated_result, truncated_error = typed_records[7]["content"], typed_records[8]["content"] + assert ai_step["tool_calls"][0]["name"] == "web_search" + assert tool_step["tool_name"] == "web_search" + assert completed["result"] == "done" + assert completed["model_name"] == "test-model" + assert completed["usage"]["total_tokens"] == 7 + assert failed["error"] == "boom" + assert timed_out["error"] == "timed out" + assert len(truncated_result["result"]) == SUBAGENT_STEP_MAX_CHARS + assert truncated_result["result_truncated"] is True + assert len(truncated_error["error"]) == SUBAGENT_STEP_MAX_CHARS + assert truncated_error["error_truncated"] is True + assert {record["content"]["status"] for record in typed_records[3:]} == { + "completed", + "failed", + "cancelled", + "timed_out", + } + + +def test_captured_subagent_message_survives_task_running_conversion(): + captured: list[dict] = [] + assert capture_step_message( + AIMessage( + content="searching", + id="ai-step-1", + tool_calls=[{"id": "call-1", "name": "web_search", "args": {"query": "deerflow"}}], + ), + captured, + set(), + ) + + event = subagent_run_event( + { + "type": "task_running", + "task_id": "task-1", + "message": captured[0], + "message_index": 0, + } + ) + + assert event is not None + assert event["event_type"] == "subagent.step" + assert event["content"]["task_id"] == "task-1" + assert event["content"]["message_index"] == 0 + assert event["content"]["text"] == "searching" + assert event["content"]["tool_calls"] == [{"name": "web_search", "args": {"query": "deerflow"}}] + _assert_fixed_event_valid(event) + + +@pytest.mark.parametrize( + "chunk", + [ + {"type": "task_started", "description": "missing task id"}, + {"type": "task_started", "task_id": "", "description": "empty task id"}, + {"type": "task_started", "task_id": "call-1", "description": 42}, + {"type": "task_running", "task_id": "call-1", "message": {"type": "ai"}}, + {"type": "task_running", "task_id": "call-1", "message": {"type": "ai"}, "message_index": -1}, + {"type": "task_running", "task_id": "call-1", "message": {"type": "ai"}, "message_index": True}, + {"type": "task_running", "task_id": "call-1", "message": "not-an-object", "message_index": 0}, + {"type": "task_completed"}, + ], +) +def test_subagent_producer_rejects_chunks_missing_contract_fields(chunk): + assert subagent_run_event(chunk) is None + + +@pytest.mark.anyio +@pytest.mark.parametrize("backend", ["memory", "jsonl"]) +async def test_subagent_batch_round_trip_matches_contract_for_non_database_stores(backend, tmp_path): + if backend == "memory": + store = MemoryRunEventStore() + else: + from deerflow.runtime.events.store.jsonl import JsonlRunEventStore + + store = JsonlRunEventStore(base_dir=tmp_path / "subagent-events") + + records = await _persist_subagent_batch(store) + + assert [record["event_type"] for record in records] == ["subagent.start", "subagent.step", "subagent.end"] + for record in records: + _assert_fixed_event_valid(record, persisted=True) + + +@pytest.mark.anyio +async def test_subagent_batch_round_trip_matches_contract_for_database_store(tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + from deerflow.runtime.events.store.db import DbRunEventStore + + url = f"sqlite+aiosqlite:///{tmp_path / 'subagent-events.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + records = await _persist_subagent_batch(DbRunEventStore(get_session_factory())) + finally: + await close_engine() + + assert [record["event_type"] for record in records] == ["subagent.start", "subagent.step", "subagent.end"] + for record in records: + _assert_fixed_event_valid(record, persisted=True) + + +@pytest.mark.anyio +async def test_workspace_change_producer_matches_catalog_and_payload(monkeypatch, tmp_path): + from deerflow.workspace_changes import WorkspaceRoot, scan_workspace_roots + from deerflow.workspace_changes import recorder as recorder_module + + workspace = tmp_path / "workspace" + outputs = tmp_path / "outputs" + workspace.mkdir() + outputs.mkdir() + roots = [ + WorkspaceRoot("workspace", workspace, "/mnt/user-data/workspace"), + WorkspaceRoot("outputs", outputs, "/mnt/user-data/outputs"), + ] + before = scan_workspace_roots(roots) + (workspace / "report.md").write_text("# Report\n", encoding="utf-8") + monkeypatch.setattr(recorder_module, "build_thread_workspace_roots", lambda *_args, **_kwargs: roots) + + store = MemoryRunEventStore() + record = await recorder_module.record_workspace_changes(store, "thread-1", "run-1", before) + + assert record is not None + assert {record["event_type"]} == {definition.event_type for definition in WORKSPACE_RUN_EVENT_DEFINITIONS} + _assert_fixed_event_valid(record, persisted=True) + + +def test_known_gaps_do_not_reclassify_current_events_as_missing(): + contract = _load_contract() + gap_ids = {gap["id"] for gap in contract["known_gaps"]} + current_types = {definition.event_type for definition in FIXED_RUN_EVENT_DEFINITIONS} + + assert {"tool-call-intent", "terminal-run-status"}.issubset(gap_ids) + assert all(gap.get("event_type") not in current_types for gap in contract["known_gaps"]) diff --git a/backend/tests/test_safety_finish_reason_middleware.py b/backend/tests/test_safety_finish_reason_middleware.py index cbffccca2..5cf4af7b1 100644 --- a/backend/tests/test_safety_finish_reason_middleware.py +++ b/backend/tests/test_safety_finish_reason_middleware.py @@ -649,7 +649,7 @@ class TestAuditEvent: assert result is not None assert result["messages"][0].tool_calls == [] - def test_journal_record_exception_does_not_break_run(self): + def test_journal_record_exception_warns_without_breaking_run(self, caplog): """Buggy journal must never propagate an exception into the agent loop.""" journal = MagicMock() journal.record_middleware.side_effect = RuntimeError("db down") @@ -663,9 +663,12 @@ class TestAuditEvent: ] } # Must not raise. - result = mw._apply(state, self._runtime_with_journal(journal)) + with caplog.at_level("WARNING"): + result = mw._apply(state, self._runtime_with_journal(journal)) + assert result is not None assert result["messages"][0].tool_calls == [] + assert "Failed to record middleware:safety_termination event" in caplog.text def test_no_record_when_passthrough(self): """When the middleware does NOT intervene, no audit event is written.""" diff --git a/backend/tests/test_skill_request_scoped_secrets.py b/backend/tests/test_skill_request_scoped_secrets.py index d863f955b..d43492dcc 100644 --- a/backend/tests/test_skill_request_scoped_secrets.py +++ b/backend/tests/test_skill_request_scoped_secrets.py @@ -898,6 +898,26 @@ class TestInContextBindsSecrets: # Values must never reach the audit journal. assert "tok-secret-value" not in str(bind_calls[0]) + def test_binding_audit_failure_warns_without_breaking_binding(self, tmp_path, monkeypatch, caplog): + from deerflow.runtime.secret_context import read_active_secrets + + skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")]) + journal = MagicMock() + journal.record_middleware.side_effect = RuntimeError("db down") + context = {"secrets": {"ERP_TOKEN": "tok-123"}, "__run_journal": journal} + + with caplog.at_level("WARNING"): + self._run_call( + tmp_path, + monkeypatch, + [skill], + context=context, + skill_context=[_skill_context_entry(skill)], + ) + + assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"} + assert "Failed to record skill secret binding audit event" in caplog.text + def test_slash_binding_persists_across_model_calls_in_same_run(self, tmp_path, monkeypatch): """#3861 semantics preserved under per-call recompute: after the single activation call, the tool loop issues more model calls without a fresh diff --git a/backend/tests/test_slash_skills.py b/backend/tests/test_slash_skills.py index e9291bcd6..b033f3dcf 100644 --- a/backend/tests/test_slash_skills.py +++ b/backend/tests/test_slash_skills.py @@ -505,7 +505,7 @@ def test_skill_activation_middleware_async_records_activation_audit_event(monkey assert kwargs["changes"]["content_hash"] == hashlib.sha256(b"# Data Analysis\nUse pandas.").hexdigest() -def test_skill_activation_middleware_ignores_activation_audit_errors(monkeypatch, tmp_path): +def test_skill_activation_middleware_warns_and_ignores_activation_audit_errors(monkeypatch, tmp_path, caplog): skill = _make_skill(tmp_path, "data-analysis", content="# Data Analysis\nUse pandas.") monkeypatch.setattr(middleware_module, "get_or_new_skill_storage", lambda **kwargs: _make_storage(tmp_path, [skill])) @@ -517,10 +517,12 @@ def test_skill_activation_middleware_ignores_activation_audit_errors(monkeypatch def handler(model_request: ModelRequest): return AIMessage(content="ok") - result = middleware.wrap_model_call(_make_model_request([original], runtime=runtime), handler) + with caplog.at_level("WARNING"): + result = middleware.wrap_model_call(_make_model_request([original], runtime=runtime), handler) assert isinstance(result, AIMessage) assert result.content == "ok" + assert "Failed to record slash skill activation audit event" in caplog.text def test_skill_activation_middleware_activates_only_latest_real_user_message(monkeypatch, tmp_path): diff --git a/contracts/run_event_stream_contract.json b/contracts/run_event_stream_contract.json new file mode 100644 index 000000000..0b2826235 --- /dev/null +++ b/contracts/run_event_stream_contract.json @@ -0,0 +1,455 @@ +{ + "version": 1, + "schema_dialect": "https://json-schema.org/draft/2020-12/schema", + "description": "The current DeerFlow run event stream contract. It freezes existing names and categories, describes producer payloads, and records compatibility rules without changing runtime behavior.", + "compatibility": { + "current_event_names": "frozen", + "consumer_rule": "Consumers must ignore unknown event types, unknown envelope fields, and unknown optional payload or metadata fields.", + "additive_changes": [ + "add_event_type", + "add_optional_payload_field", + "add_optional_metadata_field", + "add_envelope_field" + ], + "breaking_changes": [ + "remove_event_type", + "rename_event_type", + "change_event_category", + "remove_required_field", + "change_required_field_type" + ] + }, + "legacy_event_aliases": [ + { + "event_type": "ai_message", + "canonical_event_type": "llm.ai.response", + "status": "read-only compatibility", + "produced_by_current_runtime": false, + "compatibility_scope": "Category-based message projections and last-visible-AI store queries, including the /messages/page endpoint, recognize this historical name.", + "known_limitations": "The legacy /messages endpoint returns category=message rows but only enriches feedback for llm.ai.response.", + "notes": "New producers must use llm.ai.response." + } + ], + "record_schema": { + "type": "object", + "required": [ + "thread_id", + "run_id", + "seq", + "event_type", + "category", + "content", + "metadata", + "created_at" + ], + "properties": { + "thread_id": {"type": "string"}, + "run_id": {"type": "string"}, + "seq": {"type": "integer", "minimum": 1}, + "event_type": {"type": "string", "minLength": 1, "maxLength": 32}, + "category": {"type": "string", "minLength": 1, "maxLength": 16}, + "content": {"type": ["string", "object", "array", "number", "boolean", "null"]}, + "metadata": {"type": "object"}, + "created_at": {"type": "string", "format": "date-time"}, + "user_id": {"type": ["string", "null"]} + }, + "additionalProperties": true + }, + "sequence": { + "scope": "thread_id", + "guarantee": "strictly increasing within a thread", + "jsonl_multi_process_limit": "JsonlRunEventStore only guarantees monotonic assignment within one process; use DbRunEventStore for shared multi-process writes." + }, + "categories": { + "trace": "Execution evidence that is excluded from message projections.", + "message": "A candidate message projection. Server and frontend visibility filters still apply.", + "outputs": "Root graph completion output; not an authoritative run lifecycle status.", + "error": "Callback-observed run or chain failure evidence.", + "middleware": "Middleware state-change audit evidence.", + "context": "Identity-only evidence about effective hidden context.", + "subagent": "Subagent lifecycle and persisted step evidence.", + "workspace": "Run-scoped workspace and output file-change evidence." + }, + "events": [ + { + "event_type": "run.start", + "category": "trace", + "producer": "RunJournal.on_chain_start(parent_run_id=None)", + "content_schema": { + "type": "object", + "required": ["chain"], + "properties": {"chain": {"type": ["string", "null"]}}, + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["caller"], + "properties": {"caller": {"type": "string"}}, + "additionalProperties": true + } + }, + { + "event_type": "run.end", + "category": "outputs", + "producer": "RunJournal.on_chain_end(parent_run_id=None)", + "content_schema": true, + "content_description": "Opaque root graph outputs as supplied by LangGraph. Values nested inside the output may have backend-dependent representations when they are not directly JSON serializable.", + "storage_semantics": { + "memory": "Retains the original Python container and nested values.", + "jsonl": "Persists JSON with json.dumps(default=str), so nested non-JSON values are restored as strings.", + "database": "Persists the top-level payload as JSON text with json.dumps(default=str), then restores the JSON container on read; nested non-JSON values remain strings." + }, + "metadata_schema": { + "type": "object", + "required": ["status"], + "properties": {"status": {"const": "success"}}, + "additionalProperties": true + } + }, + { + "event_type": "run.error", + "category": "error", + "producer": "RunJournal.on_chain_error()", + "content_schema": {"type": "string"}, + "metadata_schema": { + "type": "object", + "required": ["error_type"], + "properties": {"error_type": {"type": "string"}}, + "additionalProperties": true + } + }, + { + "event_type": "llm.human.input", + "category": "message", + "producer": "RunJournal.on_chat_model_start() for the first persisted lead-agent HumanMessage", + "content_schema": { + "type": "object", + "required": ["type", "content"], + "properties": { + "type": {"const": "human"}, + "content": true, + "additional_kwargs": {"type": "object"} + }, + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["caller"], + "properties": {"caller": {"const": "lead_agent"}}, + "additionalProperties": true + } + }, + { + "event_type": "llm.ai.response", + "category": "message", + "producer": "RunJournal.on_llm_end()", + "content_schema": { + "type": "object", + "required": ["type", "content", "tool_calls"], + "properties": { + "type": {"const": "ai"}, + "content": true, + "tool_calls": {"type": "array"}, + "additional_kwargs": {"type": "object"}, + "response_metadata": {"type": "object"} + }, + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["caller", "usage", "latency_ms", "llm_call_index"], + "properties": { + "caller": {"type": "string"}, + "usage": {"type": "object"}, + "latency_ms": {"type": ["integer", "null"], "minimum": 0}, + "llm_call_index": {"type": "integer", "minimum": 1} + }, + "additionalProperties": true + } + }, + { + "event_type": "llm.tool.result", + "category": "message", + "producer": "RunJournal.on_tool_end() for ToolMessage or Command(update.messages[])", + "content_schema": { + "type": "object", + "required": ["type", "content"], + "properties": { + "type": {"type": "string"}, + "content": true, + "tool_call_id": {"type": ["string", "null"]} + }, + "additionalProperties": true + }, + "metadata_schema": {"type": "object", "additionalProperties": true} + }, + { + "event_type": "llm.error", + "category": "trace", + "producer": "RunJournal.on_llm_error()", + "content_schema": {"type": "string"}, + "metadata_schema": {"type": "object", "additionalProperties": true} + }, + { + "event_type": "context:memory", + "category": "context", + "producer": "RunJournal.record_memory_context() from DynamicContextMiddleware", + "content_schema": { + "type": "object", + "required": ["content_sha256"], + "properties": { + "content_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + }, + "additionalProperties": true + }, + "metadata_schema": {"type": "object", "additionalProperties": true} + }, + { + "event_type": "subagent.start", + "category": "subagent", + "producer": "subagent_run_event(task_started)", + "content_schema": { + "type": "object", + "required": ["task_id", "description"], + "properties": { + "task_id": {"type": "string", "minLength": 1}, + "description": {"type": ["string", "null"]} + }, + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["task_id"], + "properties": {"task_id": {"type": "string", "minLength": 1}}, + "additionalProperties": true + } + }, + { + "event_type": "subagent.step", + "category": "subagent", + "producer": "subagent_run_event(task_running)", + "content_schema": { + "type": "object", + "required": ["task_id", "message_index", "kind", "text", "truncated"], + "properties": { + "task_id": {"type": "string", "minLength": 1}, + "message_index": {"type": "integer", "minimum": 0}, + "kind": {"enum": ["ai", "tool"]}, + "text": {"type": "string"}, + "truncated": {"type": "boolean"}, + "tool_calls": { + "type": "array", + "items": { + "type": "object", + "required": ["name", "args"], + "properties": { + "name": {"type": ["string", "null"]}, + "args": true, + "args_truncated": {"type": "boolean"} + }, + "additionalProperties": true + } + }, + "tool_name": {"type": ["string", "null"]} + }, + "allOf": [ + { + "if": {"properties": {"kind": {"const": "ai"}}}, + "then": {"required": ["tool_calls"]} + }, + { + "if": {"properties": {"kind": {"const": "tool"}}}, + "then": {"required": ["tool_name"]} + } + ], + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["task_id", "message_index"], + "properties": { + "task_id": {"type": "string", "minLength": 1}, + "message_index": {"type": "integer", "minimum": 0} + }, + "additionalProperties": true + } + }, + { + "event_type": "subagent.end", + "category": "subagent", + "producer": "subagent_run_event(task_completed|task_failed|task_cancelled|task_timed_out)", + "content_schema": { + "type": "object", + "required": ["task_id", "status"], + "properties": { + "task_id": {"type": "string", "minLength": 1}, + "status": {"enum": ["completed", "failed", "cancelled", "timed_out"]}, + "model_name": {"type": "string"}, + "usage": {"type": "object"}, + "result": {"type": "string"}, + "result_truncated": {"type": "boolean"}, + "error": {"type": "string"}, + "error_truncated": {"type": "boolean"} + }, + "additionalProperties": true + }, + "metadata_schema": { + "type": "object", + "required": ["task_id"], + "properties": {"task_id": {"type": "string", "minLength": 1}}, + "additionalProperties": true + } + }, + { + "event_type": "workspace_changes", + "category": "workspace", + "producer": "workspace_changes.record_workspace_changes()", + "content_schema": {"type": "string"}, + "metadata_schema": { + "type": "object", + "required": ["workspace_changes"], + "properties": { + "workspace_changes": { + "type": "object", + "required": ["version", "summary", "files", "limits"], + "properties": { + "version": {"const": 1}, + "summary": { + "type": "object", + "required": ["created", "modified", "deleted", "symlink_created", "additions", "deletions", "truncated"], + "properties": { + "created": {"type": "integer", "minimum": 0}, + "modified": {"type": "integer", "minimum": 0}, + "deleted": {"type": "integer", "minimum": 0}, + "symlink_created": {"type": "integer", "minimum": 0}, + "additions": {"type": "integer", "minimum": 0}, + "deletions": {"type": "integer", "minimum": 0}, + "truncated": {"type": "boolean"} + }, + "additionalProperties": true + }, + "files": { + "type": "array", + "items": { + "type": "object", + "required": [ + "path", + "root", + "status", + "binary", + "sensitive", + "size_before", + "size_after", + "sha256_before", + "sha256_after", + "diff", + "diff_truncated", + "diff_unavailable_reason", + "additions", + "deletions", + "symlink", + "symlink_target_before", + "symlink_target_after" + ], + "properties": { + "path": {"type": "string"}, + "root": {"type": "string"}, + "status": {"enum": ["created", "modified", "deleted", "symlink_created"]}, + "binary": {"type": "boolean"}, + "sensitive": {"type": "boolean"}, + "size_before": {"type": ["integer", "null"], "minimum": 0}, + "size_after": {"type": ["integer", "null"], "minimum": 0}, + "sha256_before": {"type": ["string", "null"]}, + "sha256_after": {"type": ["string", "null"]}, + "diff": {"type": "string"}, + "diff_truncated": {"type": "boolean"}, + "diff_unavailable_reason": {"type": ["string", "null"]}, + "additions": {"type": "integer", "minimum": 0}, + "deletions": {"type": "integer", "minimum": 0}, + "symlink": {"type": "boolean"}, + "symlink_target_before": {"type": ["string", "null"]}, + "symlink_target_after": {"type": ["string", "null"]} + }, + "additionalProperties": true + } + }, + "limits": { + "type": "object", + "required": ["max_files", "max_scanned_files", "max_file_bytes_for_diff", "max_total_diff_bytes"], + "properties": { + "max_files": {"type": "integer", "minimum": 0}, + "max_scanned_files": {"type": "integer", "minimum": 0}, + "max_file_bytes_for_diff": {"type": "integer", "minimum": 0}, + "max_total_diff_bytes": {"type": "integer", "minimum": 0} + }, + "additionalProperties": true + } + }, + "additionalProperties": true + } + }, + "additionalProperties": true + } + } + ], + "dynamic_event_patterns": [ + { + "pattern": "middleware:{tag}", + "category": "middleware", + "producer": "RunJournal.record_middleware(tag, ...)", + "known_tags": ["guardrail", "safety_termination", "skill_activation", "skill_secrets"], + "event_type_schema": { + "type": "string", + "pattern": "^middleware:", + "minLength": 12, + "maxLength": 32 + }, + "tag_schema": {"type": "string", "minLength": 1, "maxLength": 21}, + "content_schema": { + "type": "object", + "required": ["name", "hook", "action", "changes"], + "properties": { + "name": {"type": "string"}, + "hook": {"type": "string"}, + "action": {"type": "string"}, + "changes": {"type": "object"} + }, + "additionalProperties": true + }, + "metadata_schema": {"type": "object", "additionalProperties": true} + } + ], + "known_gaps": [ + { + "id": "mixed-event-name-separators", + "status": "frozen for compatibility", + "notes": "Current dot, colon, and bare-word event names remain unchanged. Any future normalization requires a versioned migration or dual-write period." + }, + { + "id": "tool-call-intent", + "status": "not first-class", + "notes": "Tool call requests are embedded in llm.ai.response content.tool_calls. llm.tool.result records returned messages, and a missing or timed-out result may leave no dedicated outcome event." + }, + { + "id": "terminal-run-status", + "status": "split source of truth", + "notes": "run.end is only a root graph completion marker and always says success. RunRow.status is authoritative for success, error, interrupted, and timeout; worker loss may leave no terminal run event." + }, + { + "id": "run-end-backend-serialization", + "status": "backend-dependent opaque payload", + "affected_event_types": ["run.end"], + "notes": "Memory retains nested Python values in root graph outputs, while JSONL and database persistence stringify nested values that are not directly JSON serializable. Consumers must not rely on backend-identical nested run.end content." + }, + { + "id": "middleware-coverage", + "status": "partial", + "notes": "Loop detection and deferred-tool promotion do not currently emit middleware events." + }, + { + "id": "run-scoped-observation-context", + "status": "manual wiring", + "notes": "Journal attribution, token accounting, and external tracing metadata are still attached manually at several LLM call sites." + } + ] +}