mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-25 15:37:53 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
MiaoRuidx
GitHub
parent
62dd8d2b67
commit
f1632cc351
@ -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 `<available-deferred-tools>` 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.
|
||||
|
||||
@ -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
|
||||
|
||||
179
backend/docs/RUN_EVENT_STREAM.md
Normal file
179
backend/docs/RUN_EVENT_STREAM.md
Normal file
@ -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.
|
||||
@ -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 ------------------------------------------------------
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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)
|
||||
|
||||
113
backend/packages/harness/deerflow/runtime/events/catalog.py
Normal file
113
backend/packages/harness/deerflow/runtime/events/catalog.py
Normal file
@ -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,
|
||||
)
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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},
|
||||
}
|
||||
|
||||
@ -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},
|
||||
)
|
||||
|
||||
@ -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"]
|
||||
|
||||
@ -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):
|
||||
|
||||
549
backend/tests/test_run_event_stream_contract.py
Normal file
549
backend/tests/test_run_event_stream_contract.py
Normal file
@ -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"])
|
||||
@ -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."""
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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):
|
||||
|
||||
455
contracts/run_event_stream_contract.json
Normal file
455
contracts/run_event_stream_contract.json
Normal file
@ -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."
|
||||
}
|
||||
]
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user