mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
fix(ci): keep runtime guidance within the chain budget (#7522)
Move checkpoint modes, replay, module maps and terminal-ownership details intact into the existing runtime reference. Keep concise invariants and explicit links in AGENTS.md, reducing the effective chain from 99,004 to 80,817 bytes. The existing guidance regression now passes without changing budget thresholds.
This commit is contained in:
Willem Jiang
GitHub
parent
01815467aa
commit
3d5107dc0d
@ -291,3 +291,100 @@ PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_production.py \
|
||||
PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py \
|
||||
/tmp/production-bench.jsonl
|
||||
```
|
||||
|
||||
## Checkpoint channel modes and replay
|
||||
|
||||
Reject missing selected/head tuples or pending Sends before delta rewrites.
|
||||
`Command(resume=...)` also rejects pending nodes except `__start__` on both.
|
||||
Use `CheckpointResumeRejected` to skip pre-write rollback; resume at head.
|
||||
|
||||
Checkpointer storage runs in one of two channel modes, selected by `checkpoint_channel_mode` in `config.yaml` (default `full`). `delta` mode adopts LangGraph 1.2's `DeltaChannel` for `messages`: checkpoints store a sentinel + per-step writes instead of the full message list, so storage/serde grows O(N) instead of O(N²) in turns. All checkpointer backends (memory/sqlite/postgres) serve both modes unchanged — the semantics live in the compiled graph's channel table, not in the saver.
|
||||
|
||||
**Mode is process-frozen and restart-required.** `make_lead_agent` and the embedded `DeerFlowClient` freeze the resolved mode (`runtime/checkpoint_mode.py::freeze_checkpoint_channel_mode`) before compiling the graph with the mode-matched schema (`agents/thread_state.py::get_thread_state_schema`, plus `adapt_state_schema_for_mode` / `normalize_middleware_state_schemas` for middleware state). Adapted middleware schemas are cached by schema, mode, and resolved snapshot frequency so a pre-freeze ephemeral graph cannot leave a stale default-frequency schema behind. A second, different mode or frequency in the same process raises `CheckpointModeReconfigurationError`. To switch: edit config, restart.
|
||||
|
||||
**Delta snapshot cadence is configurable but frozen with the mode.** `database.checkpoint_delta.snapshot_frequency` (default `10`) sets the `DeltaChannel` snapshot cadence. It is frozen alongside the mode (`freeze_checkpoint_snapshot_frequency`; non-positive direct inputs raise `ValueError`, while a frozen-value mismatch raises `CheckpointModeReconfigurationError`), restart-required, and must match across every process sharing one checkpoint database — the cadence lives in each compiled graph's channel table and is deliberately NOT stamped into checkpoint metadata, so the mode-compatibility marker and full -> delta migration semantics are unchanged. Schema helpers resolve it explicit-arg -> frozen -> default, and every schema/graph cache (`_delta_thread_state_schema`, `_adapt_state_schema_for_delta`, the client agent-config key, the gateway accessor-graph cache) keys on the resolved value.
|
||||
|
||||
**Compiled-graph cache cap is configurable and hot-reloadable.** `database.checkpoint_graph_cache.accessor_graph_max` (default `64`) bounds the gateway accessor-graph cache, which clears wholesale at the cap. The cap is re-read on every eviction check (`resolve_checkpoint_graph_cache_max`), so a config.yaml reload takes effect without a restart — a size change never affects graph semantics, only eviction timing.
|
||||
|
||||
**Compatibility is asymmetric and fail-closed.** Every checkpoint written in delta mode carries metadata marker `deerflow_checkpoint_channel_mode: "delta"` (injected via `inject_checkpoint_mode`; absence of marker = full, so pre-feature checkpoints need no migration). Before any state read/write, `ensure_checkpoint_mode_compatible` rejects a full-mode process opening a delta thread with `CheckpointModeMismatchError` (surfaced as HTTP 409 with the cause and thread id by the threads router; `CheckpointModeReconfigurationError` maps to 503) — a full-mode raw read of a delta blob would silently return empty/partial `messages`. The reverse direction is allowed: delta-mode processes read full checkpoints transparently (old full checkpoints seed the delta channel), so full → delta is the smooth migration path; delta → full requires materializing/converting the data first. Detection also honors upstream's `counters_since_delta_snapshot.messages` metadata, and an explicit config marker takes precedence over any ambient context value.
|
||||
|
||||
**Never bypass `CheckpointStateAccessor` (`runtime/checkpoint_state.py`) for thread-state access.** It is the single choke point binding graph + checkpointer + mode: it injects the mode marker into configs, runs the compatibility check before every `get`/`update`/`history`, and returns materialized state (delta checkpoints lack `channel_values.messages` — raw `get_tuple` reads see a sentinel). An async unpinned delta read verifies that its resolved checkpoint still exists after materialization and retries once from the new head when retention removed that target mid-read; explicit historical selections never silently move. For metadata-only reads, use `get_metadata` / `aget_metadata` to retain the mode gate without materialization. Thread-owned reads must use `build_thread_checkpoint_state_accessor` so the recorded assistant schema materializes every channel. Sync checkpoint mutations must drain off-thread commits before cancellation propagates; sync `get_tuple` reads remain cancellable. `history(limit)` semantics: `0` means zero items (explicit empty), `None` means unlimited — do not pass `limit=0` through to `graph.get_state_history`. Assistant metadata lookup is fail-closed for mutation accessors so a store outage cannot silently select the default schema and discard extension channels. In `full` mode the read path degrades to a raw checkpointer read (`_RawCheckpointReadAccessor`) when the agent factory cannot build the graph (bad model config, MCP outage) — full checkpoints carry complete `channel_values`, so reads don't need the graph; degraded snapshots take `created_at` from the standard checkpoint `ts` field, falling back to metadata only for compatibility. The delta gate still applies on the degraded path; `next`/`tasks` degrade to empty and thread status falls back to the stored status because task presence is not derivable, while delta mode has no fallback (materialization needs the channel table).
|
||||
|
||||
**Replay checkpoint lookup prefers lineage and degrades only for an explicitly missing legacy parent link.** Branch and regenerate paths first walk `parent_config`, which prevents a global chronological scan from selecting a sibling created by regeneration. `CheckpointParentMissingError` alone enables the bounded newest-first history fallback in `app/gateway/checkpoint_lineage.py`; cycles, dangling/non-addressable parents, target mismatches, and depth exhaustion raise `CheckpointLineageIntegrityError` and fail closed instead of selecting a sibling. The compatibility scans request 400 raw checkpoints so up to 200 duration-only entries do not consume the effective branch-history budget; the fallback scans oldest-to-newest internally, skips duration-only checkpoints, and accepts only checkpoints with an addressable id as the replay base. A source history with no discoverable pre-user checkpoint preserves the historical single-checkpoint branch behavior instead of rejecting the branch; regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are not mutated by regenerate preparation, and no raw checkpoint tuple is copied across threads because delta state depends on ancestry and pending writes. Regenerate source-run lookup uses the current thread's exact event, then the server-stamped `run_id` on the copied human message, then verified RunManager content matching; it does not read parent-thread events. When an interrupted response was streamed but never checkpointed, regeneration accepts only the latest visible human message's server-stamped `run_id` after verifying that it belongs to the same thread and still has `interrupted` status. Storage or checkpoint-mode failures are not treated as a missing base and still fail closed.
|
||||
|
||||
**A delta-mode run cannot fork; `runtime/runs/worker.py` linearizes the resume instead.** An older-checkpoint resume creates siblings whose pending writes can replay into the fork, retaining a superseded answer. Write-to-child ownership belongs to the upstream delta contract, so DeerFlow does not reimplement the walk: `_linearize_delta_checkpoint_resume` materializes the requested checkpoint's complete state and writes every channel onto the **current head** (which has no siblings) through the state mutation graph, using `Overwrite` for reducer channels and resetting newer head-only channels to their schema default (or `None` when no constructible default exists); it then drops the `checkpoint_id` selector and lets the run proceed linearly, while the abandoned turn stays in history as the rewritten head's ancestry. The worker holds `_checkpoint_thread_lock` across `_capture_rollback_point` and the optional linear rewrite, making the rollback snapshot and rewrite atomic with graph streaming and the preceding run's duration-metadata checkpoint write. Capture preserves the complete real pre-run state; cancel-with-rollback then linearly replaces the current delta head with that captured state rather than forking the now-shared pre-run checkpoint, so the abandoned turn is restored without replaying the resume sibling's writes. The worker also recomputes the current-run message boundary from the rewritten state and fails closed (an unreadable resume checkpoint raises rather than falling back to the corrupt fork). `full` mode keeps forking — its checkpoints carry complete `channel_values` and need no replay — so LangGraph branching semantics are unchanged there. Root namespace only; subgraph namespaces are left alone.
|
||||
|
||||
**Wholesale state replacement uses a state-only mutation graph + `Overwrite`.** `update_state` values pass through channel reducers (`add_messages` merge in full, append in delta), so replacing reducer values requires `Overwrite` rather than an ordinary update. Full-mode rollback and context compaction replace `messages`; delta resume and delta rollback replace every materialized channel and reset current-head-only channels to their schema default (or `None`). These writes go through `build_state_mutation_graph(as_node, mode, state_schema)`, and `state_schema` MUST be the thread's effective schema (`graph_state_schema(assistant_graph)`), because the base-ThreadState fallback silently discards written channels contributed by custom `AgentMiddleware.state_schema`. Channels absent from a full-mode fork write inherit the parent's channel blobs, so middleware channels survive rollback/compaction (locked by `test_rollback_preserves_middleware_contributed_channels` and `test_compact_thread_context_preserves_middleware_contributed_channels`). The compiled mutation graph has one no-op node (entry = finish) whose checkpoint machinery (channels/versions/metadata) is identical to the agent graph's but schedules no pending tasks, so the restored/compacted head stays idle instead of re-triggering the agent. Never hand-write checkpoints via `checkpointer.aput` for this; raw writers elsewhere must preserve checkpoint parentage — severed ancestry breaks delta replay (see `runtime/runs/worker.py` writer parenting and `checkpoint_patches.py`).
|
||||
|
||||
**Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure, including cancellation delivered before or during the capture await, disables rollback (fail-closed), never restores partial state, and never deletes the thread; only a capture that completes and finds no checkpoint may reset the thread to empty. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Resume and rollback state rewrites copy only the selected/captured checkpoint's server-authored agent binding; missing or malformed bindings remain unbound. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement.
|
||||
|
||||
### Checkpoint implementation map
|
||||
|
||||
**Where things live**:
|
||||
- `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
|
||||
- `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`
|
||||
- `checkpoint_patches.py` (package root) — the one remaining patch: `BinaryOperatorAggregate` unwrapping an `Overwrite` first write into an empty channel (#4380; probe-guarded). The `InMemorySaver` delta-history patch is gone: `langgraph-checkpoint` 4.2.0 fixed that write loss upstream (#8526) while keeping its override, so the dependency floor plus the full → delta migration contract test are the gate — never re-add a version-guarded saver patch.
|
||||
- `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `delta_messages_field` / `DELTA_MESSAGES_FIELD` (`DeltaChannel` at the configured `snapshot_frequency`, default 10), schema adaptation helpers
|
||||
- `runtime/context_compaction.py` — compaction via accessor + mutation graph (reference consumer). Runs stamp their effective agent into server-owned checkpoint metadata; manual compaction uses that binding—not request `agent_name`—for memory policy and bucket. Missing/invalid legacy bindings and unreadable agent configs fail closed by skipping the optional flush while compaction may continue with the default model; a missing pre-binding checkpoint emits a warning so the skipped write is observable.
|
||||
- `runtime/checkpoint_cache/` + `runtime/checkpointer/cached_saver.py` — delta-mode checkpoint history cache; checkpoint state reads MUST go through `CheckpointStateAccessor`, and the checkpointer may be a `CachedHistorySaver` wrapper — never rely on concrete saver types
|
||||
- `runtime/checkpointer/thread_spans.py` — thread enumeration for `DeerFlowClient.list_threads`, ordered by `checkpoint_id` (goal writes keep the old `ts`). SQLite/Postgres savers (unwrapped from `CachedHistorySaver`) use the primary-key index; any other saver walks `list(None)`. Saver type selects speed, never correctness
|
||||
- `runtime/goal.py` `_call_checkpointer_method` — tries the async saver method and falls back to the sync one only on `NotImplementedError`: every saver defines the async methods, but the sync TUI/embedded `SqliteSaver`/`PostgresSaver` (also behind `CachedHistorySaver`) raise from them. Sync writes stay in `await_drained`
|
||||
- Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py`, `tests/test_cached_history_saver.py` + `tests/test_cached_history_saver_integration.py` (history cache)
|
||||
|
||||
## Deferred terminal commits and lease ownership
|
||||
|
||||
**Deferred terminal commit:** With an event store, the worker stages its terminal
|
||||
status locally and commits it only after finalization's receipt and duration
|
||||
writes. `RunRecord.terminal_commit_pending` keeps `_renew_leases()` renewing that
|
||||
still-active row until its own terminal write is acknowledged or ownership is
|
||||
lost. Manager-owned attempts register separate futures before Store I/O and
|
||||
publish an immutable outcome proof only on explicit successful commit results;
|
||||
matching row fields and legacy `None` results cannot confirm a commit. Atomic
|
||||
replacement registers attempts before admission I/O and acknowledges only the
|
||||
returned claimed rows, using their committed outcome fields. It does not repeat
|
||||
the interrupted update. An interrupted ACK is intermediate while a live Worker
|
||||
still owes rollback: keep a separate rollback barrier and explicitly renew that
|
||||
interrupted row under the same live-owner/lease predicates until the final error
|
||||
ACK. An active-only renewal that was already in flight across atomic rollback
|
||||
admission is retried once with `allow_interrupted=True` only after this Worker's
|
||||
interrupted ACK is available, and only while that same per-run decision remains
|
||||
inside its original deadline. An in-flight admission ACK is awaited; row state is
|
||||
never reread as attribution. Peer takeover, an expired lease, a changed/final
|
||||
outcome, or a second rejection still fences. Even without an event store, defer
|
||||
rollback's error write until restoration has finished. `tests/test_stale_active_renewal.py`
|
||||
covers the Manager transition (including an ACK still in flight), and
|
||||
`tests/test_atomic_admission_worker.py` covers the real Worker restoration/error
|
||||
tail with and without an event Store. A fenced Worker must not begin a rollback checkpoint restoration,
|
||||
including when fencing itself delivered cancellation. Admission compensation never retries an unowned write in heartbeat
|
||||
mode. A local deadline guard remains armed after fast renewal errors and after
|
||||
heartbeat stop; it rechecks the latest lease/proof and fences without waiting for
|
||||
the admission lock. Record removal and final ACK cancel its timer. Bounded orphan
|
||||
recovery drain requests cancellation, retains/observes unfinished I/O, and returns
|
||||
without joining a cancellation-resistant Store. Neither that return nor heartbeat
|
||||
stop authorizes closing still-in-use resources. A rejected
|
||||
CAS never falls back to another terminal writer. In heartbeat mode completion
|
||||
updates require that proof and an atomic matching terminal status/owner in Store;
|
||||
they only enrich metadata. Single-worker completion retains its recovery fallback.
|
||||
Never select runs for renewal by local status alone. Each run has a single-flight
|
||||
renewal check and supervised I/O, bounded by its last confirmed durable lease and
|
||||
a monotonic wait budget. Heartbeat ticks do not join slow checks; cancellation
|
||||
signals are delivered per run as results arrive. Timeout requests cancellation
|
||||
without joining cancellation-resistant I/O, whose late results are observed but
|
||||
cannot restore local authority. Shutdown drains within its shared budget and
|
||||
reports outstanding I/O rather than claiming it stopped. SQL/Memory reject expired
|
||||
renewals. `stop_heartbeat()` is not a Store-resource idle barrier: a driver that
|
||||
suppresses cancellation can retain its connection until the supervised I/O settles
|
||||
or the process exits; closing that resource cannot be claimed safe from this return.
|
||||
Owned terminal writes require a live owner lease (Postgres checks DB time
|
||||
after its row lock). These guards do not fence every RunStore mutation or provide
|
||||
a persistent attempt identity. Permanently missing ACKs may conservatively fence
|
||||
a committed run. Shutdown's drain treats the
|
||||
same state as active, and also awaits committed terminal workers still executing
|
||||
hooks, END or resource cleanup. The registered worker shields and drains one owned
|
||||
finalizer across repeated host cancellation, preserving the first interruption even
|
||||
if a step fails. There is no deadline that cancels or detaches writes; backend I/O
|
||||
must be bounded by its backend. Only the optional completion hook receives host
|
||||
cancellation, and its child task is drained before the terminal tail continues.
|
||||
A drain timeout leaves staged statuses and commit barriers intact without persisting `interrupted`. A finalizer that raises is
|
||||
logged with its run ID, staged status and exception while the other runs continue
|
||||
draining. Coverage: `tests/test_gateway_run_drain_shutdown.py`.
|
||||
|
||||
@ -53,29 +53,17 @@ with real ASGI requests and LangGraph SDK decoding.
|
||||
|
||||
### Checkpoint Channel Modes (`full` / `delta`)
|
||||
|
||||
Reject missing selected/head tuples or pending Sends before delta rewrites.
|
||||
`Command(resume=...)` also rejects pending nodes except `__start__` on both.
|
||||
Use `CheckpointResumeRejected` to skip pre-write rollback; resume at head.
|
||||
Keep mode and delta snapshot cadence process-frozen; only the accessor graph
|
||||
cache cap is hot-reloadable. Full-mode readers reject delta checkpoints.
|
||||
Use `CheckpointStateAccessor` with the thread's effective schema for state reads
|
||||
and mutations; preserve mode gates, parentage and pending writes. State replacement
|
||||
uses `Overwrite`. Delta resume/rollback rewrites the current head rather than
|
||||
forking; reject missing selected/head tuples and pending Sends before writing.
|
||||
`Command(resume=...)` also rejects pending nodes except `__start__` on both;
|
||||
`CheckpointResumeRejected` skips pre-write rollback. Resume at head.
|
||||
|
||||
Checkpointer storage runs in one of two channel modes, selected by `checkpoint_channel_mode` in `config.yaml` (default `full`). `delta` mode adopts LangGraph 1.2's `DeltaChannel` for `messages`: checkpoints store a sentinel + per-step writes instead of the full message list, so storage/serde grows O(N) instead of O(N²) in turns. All checkpointer backends (memory/sqlite/postgres) serve both modes unchanged — the semantics live in the compiled graph's channel table, not in the saver.
|
||||
|
||||
**Mode is process-frozen and restart-required.** `make_lead_agent` and the embedded `DeerFlowClient` freeze the resolved mode (`runtime/checkpoint_mode.py::freeze_checkpoint_channel_mode`) before compiling the graph with the mode-matched schema (`agents/thread_state.py::get_thread_state_schema`, plus `adapt_state_schema_for_mode` / `normalize_middleware_state_schemas` for middleware state). Adapted middleware schemas are cached by schema, mode, and resolved snapshot frequency so a pre-freeze ephemeral graph cannot leave a stale default-frequency schema behind. A second, different mode or frequency in the same process raises `CheckpointModeReconfigurationError`. To switch: edit config, restart.
|
||||
|
||||
**Delta snapshot cadence is configurable but frozen with the mode.** `database.checkpoint_delta.snapshot_frequency` (default `10`) sets the `DeltaChannel` snapshot cadence. It is frozen alongside the mode (`freeze_checkpoint_snapshot_frequency`; non-positive direct inputs raise `ValueError`, while a frozen-value mismatch raises `CheckpointModeReconfigurationError`), restart-required, and must match across every process sharing one checkpoint database — the cadence lives in each compiled graph's channel table and is deliberately NOT stamped into checkpoint metadata, so the mode-compatibility marker and full -> delta migration semantics are unchanged. Schema helpers resolve it explicit-arg -> frozen -> default, and every schema/graph cache (`_delta_thread_state_schema`, `_adapt_state_schema_for_delta`, the client agent-config key, the gateway accessor-graph cache) keys on the resolved value.
|
||||
|
||||
**Compiled-graph cache cap is configurable and hot-reloadable.** `database.checkpoint_graph_cache.accessor_graph_max` (default `64`) bounds the gateway accessor-graph cache, which clears wholesale at the cap. The cap is re-read on every eviction check (`resolve_checkpoint_graph_cache_max`), so a config.yaml reload takes effect without a restart — a size change never affects graph semantics, only eviction timing.
|
||||
|
||||
**Compatibility is asymmetric and fail-closed.** Every checkpoint written in delta mode carries metadata marker `deerflow_checkpoint_channel_mode: "delta"` (injected via `inject_checkpoint_mode`; absence of marker = full, so pre-feature checkpoints need no migration). Before any state read/write, `ensure_checkpoint_mode_compatible` rejects a full-mode process opening a delta thread with `CheckpointModeMismatchError` (surfaced as HTTP 409 with the cause and thread id by the threads router; `CheckpointModeReconfigurationError` maps to 503) — a full-mode raw read of a delta blob would silently return empty/partial `messages`. The reverse direction is allowed: delta-mode processes read full checkpoints transparently (old full checkpoints seed the delta channel), so full → delta is the smooth migration path; delta → full requires materializing/converting the data first. Detection also honors upstream's `counters_since_delta_snapshot.messages` metadata, and an explicit config marker takes precedence over any ambient context value.
|
||||
|
||||
**Never bypass `CheckpointStateAccessor` (`runtime/checkpoint_state.py`) for thread-state access.** It is the single choke point binding graph + checkpointer + mode: it injects the mode marker into configs, runs the compatibility check before every `get`/`update`/`history`, and returns materialized state (delta checkpoints lack `channel_values.messages` — raw `get_tuple` reads see a sentinel). An async unpinned delta read verifies that its resolved checkpoint still exists after materialization and retries once from the new head when retention removed that target mid-read; explicit historical selections never silently move. For metadata-only reads, use `get_metadata` / `aget_metadata` to retain the mode gate without materialization. Thread-owned reads must use `build_thread_checkpoint_state_accessor` so the recorded assistant schema materializes every channel. Sync checkpoint mutations must drain off-thread commits before cancellation propagates; sync `get_tuple` reads remain cancellable. `history(limit)` semantics: `0` means zero items (explicit empty), `None` means unlimited — do not pass `limit=0` through to `graph.get_state_history`. Assistant metadata lookup is fail-closed for mutation accessors so a store outage cannot silently select the default schema and discard extension channels. In `full` mode the read path degrades to a raw checkpointer read (`_RawCheckpointReadAccessor`) when the agent factory cannot build the graph (bad model config, MCP outage) — full checkpoints carry complete `channel_values`, so reads don't need the graph; degraded snapshots take `created_at` from the standard checkpoint `ts` field, falling back to metadata only for compatibility. The delta gate still applies on the degraded path; `next`/`tasks` degrade to empty and thread status falls back to the stored status because task presence is not derivable, while delta mode has no fallback (materialization needs the channel table).
|
||||
|
||||
**Replay checkpoint lookup prefers lineage and degrades only for an explicitly missing legacy parent link.** Branch and regenerate paths first walk `parent_config`, which prevents a global chronological scan from selecting a sibling created by regeneration. `CheckpointParentMissingError` alone enables the bounded newest-first history fallback in `app/gateway/checkpoint_lineage.py`; cycles, dangling/non-addressable parents, target mismatches, and depth exhaustion raise `CheckpointLineageIntegrityError` and fail closed instead of selecting a sibling. The compatibility scans request 400 raw checkpoints so up to 200 duration-only entries do not consume the effective branch-history budget; the fallback scans oldest-to-newest internally, skips duration-only checkpoints, and accepts only checkpoints with an addressable id as the replay base. A source history with no discoverable pre-user checkpoint preserves the historical single-checkpoint branch behavior instead of rejecting the branch; regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are not mutated by regenerate preparation, and no raw checkpoint tuple is copied across threads because delta state depends on ancestry and pending writes. Regenerate source-run lookup uses the current thread's exact event, then the server-stamped `run_id` on the copied human message, then verified RunManager content matching; it does not read parent-thread events. When an interrupted response was streamed but never checkpointed, regeneration accepts only the latest visible human message's server-stamped `run_id` after verifying that it belongs to the same thread and still has `interrupted` status. Storage or checkpoint-mode failures are not treated as a missing base and still fail closed.
|
||||
|
||||
**A delta-mode run cannot fork; `runtime/runs/worker.py` linearizes the resume instead.** An older-checkpoint resume creates siblings whose pending writes can replay into the fork, retaining a superseded answer. Write-to-child ownership belongs to the upstream delta contract, so DeerFlow does not reimplement the walk: `_linearize_delta_checkpoint_resume` materializes the requested checkpoint's complete state and writes every channel onto the **current head** (which has no siblings) through the state mutation graph, using `Overwrite` for reducer channels and resetting newer head-only channels to their schema default (or `None` when no constructible default exists); it then drops the `checkpoint_id` selector and lets the run proceed linearly, while the abandoned turn stays in history as the rewritten head's ancestry. The worker holds `_checkpoint_thread_lock` across `_capture_rollback_point` and the optional linear rewrite, making the rollback snapshot and rewrite atomic with graph streaming and the preceding run's duration-metadata checkpoint write. Capture preserves the complete real pre-run state; cancel-with-rollback then linearly replaces the current delta head with that captured state rather than forking the now-shared pre-run checkpoint, so the abandoned turn is restored without replaying the resume sibling's writes. The worker also recomputes the current-run message boundary from the rewritten state and fails closed (an unreadable resume checkpoint raises rather than falling back to the corrupt fork). `full` mode keeps forking — its checkpoints carry complete `channel_values` and need no replay — so LangGraph branching semantics are unchanged there. Root namespace only; subgraph namespaces are left alone.
|
||||
|
||||
**Wholesale state replacement uses a state-only mutation graph + `Overwrite`.** `update_state` values pass through channel reducers (`add_messages` merge in full, append in delta), so replacing reducer values requires `Overwrite` rather than an ordinary update. Full-mode rollback and context compaction replace `messages`; delta resume and delta rollback replace every materialized channel and reset current-head-only channels to their schema default (or `None`). These writes go through `build_state_mutation_graph(as_node, mode, state_schema)`, and `state_schema` MUST be the thread's effective schema (`graph_state_schema(assistant_graph)`), because the base-ThreadState fallback silently discards written channels contributed by custom `AgentMiddleware.state_schema`. Channels absent from a full-mode fork write inherit the parent's channel blobs, so middleware channels survive rollback/compaction (locked by `test_rollback_preserves_middleware_contributed_channels` and `test_compact_thread_context_preserves_middleware_contributed_channels`). The compiled mutation graph has one no-op node (entry = finish) whose checkpoint machinery (channels/versions/metadata) is identical to the agent graph's but schedules no pending tasks, so the restored/compacted head stays idle instead of re-triggering the agent. Never hand-write checkpoints via `checkpointer.aput` for this; raw writers elsewhere must preserve checkpoint parentage — severed ancestry breaks delta replay (see `runtime/runs/worker.py` writer parenting and `checkpoint_patches.py`).
|
||||
|
||||
**Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure, including cancellation delivered before or during the capture await, disables rollback (fail-closed), never restores partial state, and never deletes the thread; only a capture that completes and finds no checkpoint may reset the thread to empty. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Resume and rollback state rewrites copy only the selected/captured checkpoint's server-authored agent binding; missing or malformed bindings remain unbound. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement.
|
||||
Before changing checkpoint modes, caches, replay lookup, state replacement or
|
||||
rollback, read the [checkpoint contracts](../../../../docs/runtime-guidance-details.md#checkpoint-channel-modes-and-replay).
|
||||
|
||||
**Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`.
|
||||
|
||||
@ -135,61 +123,11 @@ answers for paginated history. See `docs/skill-usage-ui.md`.
|
||||
|
||||
Unmet goals stand down on run-level token/loop hard stops and retain the original stop reason; hidden continuations must not re-enter that run's shared safety state. Operation-based no-progress counting: `backend/docs/runtime-guidance-details.md`.
|
||||
|
||||
**Deferred terminal commit:** With an event store, the worker stages its terminal
|
||||
status locally and commits it only after finalization's receipt and duration
|
||||
writes. `RunRecord.terminal_commit_pending` keeps `_renew_leases()` renewing that
|
||||
still-active row until its own terminal write is acknowledged or ownership is
|
||||
lost. Manager-owned attempts register separate futures before Store I/O and
|
||||
publish an immutable outcome proof only on explicit successful commit results;
|
||||
matching row fields and legacy `None` results cannot confirm a commit. Atomic
|
||||
replacement registers attempts before admission I/O and acknowledges only the
|
||||
returned claimed rows, using their committed outcome fields. It does not repeat
|
||||
the interrupted update. An interrupted ACK is intermediate while a live Worker
|
||||
still owes rollback: keep a separate rollback barrier and explicitly renew that
|
||||
interrupted row under the same live-owner/lease predicates until the final error
|
||||
ACK. An active-only renewal that was already in flight across atomic rollback
|
||||
admission is retried once with `allow_interrupted=True` only after this Worker's
|
||||
interrupted ACK is available, and only while that same per-run decision remains
|
||||
inside its original deadline. An in-flight admission ACK is awaited; row state is
|
||||
never reread as attribution. Peer takeover, an expired lease, a changed/final
|
||||
outcome, or a second rejection still fences. Even without an event store, defer
|
||||
rollback's error write until restoration has finished. `tests/test_stale_active_renewal.py`
|
||||
covers the Manager transition (including an ACK still in flight), and
|
||||
`tests/test_atomic_admission_worker.py` covers the real Worker restoration/error
|
||||
tail with and without an event Store. A fenced Worker must not begin a rollback checkpoint restoration,
|
||||
including when fencing itself delivered cancellation. Admission compensation never retries an unowned write in heartbeat
|
||||
mode. A local deadline guard remains armed after fast renewal errors and after
|
||||
heartbeat stop; it rechecks the latest lease/proof and fences without waiting for
|
||||
the admission lock. Record removal and final ACK cancel its timer. Bounded orphan
|
||||
recovery drain requests cancellation, retains/observes unfinished I/O, and returns
|
||||
without joining a cancellation-resistant Store. Neither that return nor heartbeat
|
||||
stop authorizes closing still-in-use resources. A rejected
|
||||
CAS never falls back to another terminal writer. In heartbeat mode completion
|
||||
updates require that proof and an atomic matching terminal status/owner in Store;
|
||||
they only enrich metadata. Single-worker completion retains its recovery fallback.
|
||||
Never select runs for renewal by local status alone. Each run has a single-flight
|
||||
renewal check and supervised I/O, bounded by its last confirmed durable lease and
|
||||
a monotonic wait budget. Heartbeat ticks do not join slow checks; cancellation
|
||||
signals are delivered per run as results arrive. Timeout requests cancellation
|
||||
without joining cancellation-resistant I/O, whose late results are observed but
|
||||
cannot restore local authority. Shutdown drains within its shared budget and
|
||||
reports outstanding I/O rather than claiming it stopped. SQL/Memory reject expired
|
||||
renewals. `stop_heartbeat()` is not a Store-resource idle barrier: a driver that
|
||||
suppresses cancellation can retain its connection until the supervised I/O settles
|
||||
or the process exits; closing that resource cannot be claimed safe from this return.
|
||||
Owned terminal writes require a live owner lease (Postgres checks DB time
|
||||
after its row lock). These guards do not fence every RunStore mutation or provide
|
||||
a persistent attempt identity. Permanently missing ACKs may conservatively fence
|
||||
a committed run. Shutdown's drain treats the
|
||||
same state as active, and also awaits committed terminal workers still executing
|
||||
hooks, END or resource cleanup. The registered worker shields and drains one owned
|
||||
finalizer across repeated host cancellation, preserving the first interruption even
|
||||
if a step fails. There is no deadline that cancels or detaches writes; backend I/O
|
||||
must be bounded by its backend. Only the optional completion hook receives host
|
||||
cancellation, and its child task is drained before the terminal tail continues.
|
||||
A drain timeout leaves staged statuses and commit barriers intact without persisting `interrupted`. A finalizer that raises is
|
||||
logged with its run ID, staged status and exception while the other runs continue
|
||||
draining. Coverage: `tests/test_gateway_run_drain_shutdown.py`.
|
||||
**Deferred terminal commit:** Keep terminal commit authority on the owning
|
||||
Worker's confirmed ACK; row reads never grant it. Renew leases through receipt,
|
||||
rollback and finalization barriers; lost or expired authority fences writes.
|
||||
Drain owned finalizers across cancellation. Read the [terminal ownership contract](../../../../docs/runtime-guidance-details.md#deferred-terminal-commits-and-lease-ownership)
|
||||
before changing terminal commits, renewal, rollback or shutdown.
|
||||
|
||||
**Deferred-tool promotion event deduplication** (`runtime/journal.py`): one
|
||||
`RunJournal` owns the lead graph's run-scoped atomic promotion claim. Parallel
|
||||
@ -304,16 +242,7 @@ the DEBUG checkpoint-inspection failure regression in `test_run_worker_rollback.
|
||||
|
||||
**A `RunRecord` owner matches its durable row.** HTTP admissions omit `user_id` and the SQL store stamps the ambient user, so `create()` and `_admit_thread_operation()` resolve an omitted owner from the contextvar, keeping `None` without one — never `get_effective_user_id()`'s `default` bucket. A `None` record made SQL keyed retries on a peer or after `cleanup()` return 500, hid HTTP runs from owner-scoped history, and skipped their MCP `background_tasks` projection. Pinned by `test_run_record_owner_matches_row_stamped_from_context`, `test_keyed_retry_without_explicit_user_*`, the `sql` case of `test_start_run_peer_idempotent_reuse_*`, `test_run_manager_*_admitted_without_explicit_user`, and `test_run_manager_keeps_omitted_owner_unset_without_user_context`.
|
||||
|
||||
**Where things live**:
|
||||
- `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
|
||||
- `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`
|
||||
- `checkpoint_patches.py` (package root) — the one remaining patch: `BinaryOperatorAggregate` unwrapping an `Overwrite` first write into an empty channel (#4380; probe-guarded). The `InMemorySaver` delta-history patch is gone: `langgraph-checkpoint` 4.2.0 fixed that write loss upstream (#8526) while keeping its override, so the dependency floor plus the full → delta migration contract test are the gate — never re-add a version-guarded saver patch.
|
||||
- `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `delta_messages_field` / `DELTA_MESSAGES_FIELD` (`DeltaChannel` at the configured `snapshot_frequency`, default 10), schema adaptation helpers
|
||||
- `runtime/context_compaction.py` — compaction via accessor + mutation graph (reference consumer). Runs stamp their effective agent into server-owned checkpoint metadata; manual compaction uses that binding—not request `agent_name`—for memory policy and bucket. Missing/invalid legacy bindings and unreadable agent configs fail closed by skipping the optional flush while compaction may continue with the default model; a missing pre-binding checkpoint emits a warning so the skipped write is observable.
|
||||
- `runtime/checkpoint_cache/` + `runtime/checkpointer/cached_saver.py` — delta-mode checkpoint history cache; checkpoint state reads MUST go through `CheckpointStateAccessor`, and the checkpointer may be a `CachedHistorySaver` wrapper — never rely on concrete saver types
|
||||
- `runtime/checkpointer/thread_spans.py` — thread enumeration for `DeerFlowClient.list_threads`, ordered by `checkpoint_id` (goal writes keep the old `ts`). SQLite/Postgres savers (unwrapped from `CachedHistorySaver`) use the primary-key index; any other saver walks `list(None)`. Saver type selects speed, never correctness
|
||||
- `runtime/goal.py` `_call_checkpointer_method` — tries the async saver method and falls back to the sync one only on `NotImplementedError`: every saver defines the async methods, but the sync TUI/embedded `SqliteSaver`/`PostgresSaver` (also behind `CachedHistorySaver`) raise from them. Sync writes stay in `await_drained`
|
||||
- Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py`, `tests/test_cached_history_saver.py` + `tests/test_cached_history_saver_integration.py` (history cache)
|
||||
**Checkpoint implementation map:** Module ownership and regression suites are in the [checkpoint reference](../../../../docs/runtime-guidance-details.md#checkpoint-implementation-map).
|
||||
|
||||
**Checkpoint benchmarks:** Paired full/delta cases and production-shaped runs are documented in `backend/docs/runtime-guidance-details.md`.
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user