diff --git a/README.md b/README.md index e3921cea60..c00c3a5222 100644 --- a/README.md +++ b/README.md @@ -540,6 +540,13 @@ different cache settings. Redis thread cleanup treats a configured `key_prefix` as literal text, including glob metacharacters, so deletion stays within that namespace and thread. +Gateway checkpoint cleanup runs when the next run wins durable same-thread +admission, before its first checkpoint read. It checks only the root head and +its parent, removes at most one metadata-only duration leaf without pending +writes, and preserves explicit checkpoint selections and delta replay state. +See [checkpoint operations](https://deerflow.tech/docs/harness/checkpoints/operations) +for the retention constraints. + > [!TIP] > On Linux, if Docker-based commands fail with `permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock`, add your user to the `docker` group and re-login before retrying. See [CONTRIBUTING.md](CONTRIBUTING.md#linux-docker-daemon-permission-denied) for the full fix. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index e461dc10b8..cf907a5bb1 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -4,24 +4,19 @@ Reject external `sandbox`, `thread_data`, `viewed_images`, `goal_outcome`. Non-n Studio retains sanitized creation metadata. -Capability Center's `business` adapter validates bundled credentials and -creates an MCP connection. The MCP API permits only the isolated launcher and -credential environment keys generated by `deerflow.capabilities.business`; -reject arbitrary Python commands and manifest policy bypasses. Keep credentials -out of tool schemas; preserve admin checks, masked edits, atomic writes, and -MCP cache reloads. +Capability Center's `business` adapter validates bundled credentials and creates +an MCP connection. The MCP API permits only its isolated launcher and credential +environment keys; reject arbitrary commands and manifest-policy bypasses. Keep +credentials out of tool schemas and preserve admin checks, masked atomic edits, +and MCP cache reloads. -Memory shutdown resolves hot-reloaded config and the backend, flushes, then -closes as one `await_drained` operation. Keep config resolution inside the -best-effort error handler and off the event loop so malformed config edits do -not abort runtime teardown. Cancellation waits for the owned workers before -propagating; backend `close()` overrides must be quick or internally bounded -because close has no host timeout. Budget resolution and close in the pod grace -period in addition to the configured flush timeout and other shutdown hooks. -Extension-service teardown follows the same ownership rule: `stop_services()` -is drained across host cancellation before later runtime resources unwind; each -service `stop()` remains bounded by its 30-second `asyncio.timeout`, so include -that bound in pod grace-period budgeting. +Memory shutdown resolves hot-reloaded config/backend, flushes, then closes as +one `await_drained` operation. Resolve config inside the best-effort handler and +off-loop so malformed edits cannot abort teardown. Cancellation waits for owned +workers; `close()` must be quick or internally bounded because the host sets no +timeout. Budget it, the flush timeout, and other hooks in the pod grace period. +Likewise drain extension `stop_services()` across cancellation before later +resources unwind; each service `stop()` has a 30-second `asyncio.timeout`. Routers with persistent writes drain through `app.gateway.persistent_writes.run_drained_write`. @@ -31,6 +26,11 @@ its run drain and extension-service teardown can still make MCP calls during earlier shutdown hooks. Keep the pool close bounded and failure-isolated, but do not move it ahead of those session producers. +Retention holds durable thread admission and the checkpoint lock before +checkpoint reads. Inspect root head/parent and leaf writes only; protect selectors, +require equal channel versions and invalidate caches before deletion. No scans, +stats or blob GC; failures stay non-fatal. + `conversation_access.py` grants the opt-in read-only tool only for explicit `conversation_references` with effective `runs:read`; never infer grants from messages or checkpoints. Pass the callback through `RunContext`, not serialized diff --git a/backend/app/gateway/checkpoint_retention.py b/backend/app/gateway/checkpoint_retention.py index ef72ed3f93..5ad6440afd 100644 --- a/backend/app/gateway/checkpoint_retention.py +++ b/backend/app/gateway/checkpoint_retention.py @@ -23,24 +23,26 @@ by a deleted checkpoint in the same step (contract "deletion mechanics"). Postgres ``checkpoint_blobs`` rows are keyed by ``version`` = the id of the checkpoint that wrote the blob, so the same join keys clean them. -The service ships **without a production trigger**: where retention is -invoked from (post-run hook vs scheduler vs explicit admin action) is a -maintainer decision that lands with the contract itself. Measurement-first: -reports carry before/after per-thread stats in the same normalized shape as +The Gateway invokes the conservative production policy after the next run has +won durable same-thread admission but before that worker reads a checkpoint. +It removes at most the trailing duration/history-cache leaf; the sibling-branch +shape remains opt-in. The admission path reads only the root head and parent, +checks that head for writes, and leaves shared blobs in place. The exhaustive +maintenance policy alone collects before/after per-thread stats in the same +normalized shape as ``scripts/benchmark/checkpoint/bench_channels.py``. History fast-path interaction: the trailing duration-only leaf is also the carrier of the run-history metadata cache (``run_durations`` / -``run_message_ids``) that ``get_thread_history`` reads from the latest -checkpoint, and the parent it clones does not carry that map. Deleting the -leaf makes the next history read fall back to store scans and re-persist a -fresh leaf, so the wiring PR must sequence retention away from history reads -or adopt a policy that spares cache-carrying leaves — see the contract doc, -"History fast-path interaction (wiring requirement)". +``run_message_ids``). Retaining it until the next admission lets repeated +history reads keep their fast path; pruning immediately before checkpoint +selection prevents the next run from making a regenerated cache leaf a +permanent ancestor. """ from __future__ import annotations +import logging from collections import defaultdict from contextlib import AbstractAsyncContextManager, nullcontext from dataclasses import dataclass, field @@ -55,7 +57,15 @@ from app.gateway.checkpoint_lineage import ( is_duration_only_checkpoint, ) -__all__ = ["RetentionPolicy", "RetentionReport", "enforce_thread_retention"] +__all__ = [ + "RetentionPolicy", + "RetentionReport", + "enforce_completed_run_retention", + "enforce_thread_retention", + "enforce_trailing_duration_retention", +] + +logger = logging.getLogger(__name__) @dataclass(frozen=True) @@ -150,7 +160,13 @@ def _mark_duration_leaves_without_the_marker(nodes: dict[tuple[str, str], _Node] node.duration_only = True -def _ensure_supported_saver(saver: BaseCheckpointSaver) -> None: +def _storage_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver: + """Resolve a wrapper through its public source-saver contract.""" + storage = getattr(saver, "source_saver", saver) + return storage if isinstance(storage, BaseCheckpointSaver) else saver + + +def _ensure_supported_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver: """Fail fast before any row is read or written on an untested saver. The per-backend helpers below model exactly three storage layouts @@ -163,18 +179,12 @@ def _ensure_supported_saver(saver: BaseCheckpointSaver) -> None: ``NotImplementedError`` beats a partial deletion and a confusing traceback. """ - if isinstance(saver, (InMemorySaver, AsyncSqliteSaver)): - return - async_postgres_saver: type | None = None - try: - from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver - - async_postgres_saver = AsyncPostgresSaver - except Exception: - pass - if async_postgres_saver is not None and isinstance(saver, async_postgres_saver): - return - raise NotImplementedError(f"checkpoint retention supports InMemorySaver, AsyncSqliteSaver and AsyncPostgresSaver, got {type(saver).__module__}.{type(saver).__name__}") + storage = _storage_saver(saver) + if isinstance(storage, (InMemorySaver, AsyncSqliteSaver)): + return storage + if callable(getattr(storage, "_cursor", None)): + return storage + raise NotImplementedError(f"checkpoint retention requires a supported storage saver, got {type(storage).__module__}.{type(storage).__name__}") async def _thread_storage_stats(saver: Any, thread_id: str) -> dict[str, int]: @@ -242,6 +252,25 @@ async def _thread_storage_stats(saver: Any, thread_id: str) -> dict[str, int]: return stats +async def _checkpoint_has_writes(saver: Any, key: tuple[str, str], thread_id: str) -> bool: + """Check one checkpoint's pending writes without scanning its thread.""" + ns, cp_id = key + if isinstance(saver, InMemorySaver): + return bool(saver.writes.get((thread_id, ns, cp_id))) + if isinstance(saver, AsyncSqliteSaver): + async with saver.conn.execute( + "SELECT 1 FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? LIMIT 1", + (thread_id, ns, cp_id), + ) as cursor: + return await cursor.fetchone() is not None + async with saver._cursor() as cursor: + await cursor.execute( + "SELECT 1 FROM checkpoint_writes WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s LIMIT 1", + (thread_id, ns, cp_id), + ) + return await cursor.fetchone() is not None + + async def _checkpoint_ids_with_writes(saver: Any, thread_id: str) -> set[tuple[str, str]]: """(checkpoint_ns, checkpoint_id) pairs that still own writes rows. @@ -378,7 +407,7 @@ async def enforce_thread_retention( before writing, so a post-run-hook call site must invoke retention *after* releasing it, not from inside the held section. """ - _ensure_supported_saver(saver) + storage_saver = _ensure_supported_saver(saver) effective = policy or RetentionPolicy() if effective.max_delete_per_run is not None and effective.max_delete_per_run < 0: # Fail closed *before* any store read. ``deletable[:cap]`` with a @@ -390,10 +419,10 @@ async def enforce_thread_retention( lock: AbstractAsyncContextManager[None] = thread_lock if thread_lock is not None else nullcontext() async with lock: if collect_stats: - report.stats_before = await _thread_storage_stats(saver, thread_id) + report.stats_before = await _thread_storage_stats(storage_saver, thread_id) nodes: dict[tuple[str, str], _Node] = {} - async for tuple_ in saver.alist(_thread_config(thread_id), limit=None): + async for tuple_ in storage_saver.alist(_thread_config(thread_id), limit=None): configurable = checkpoint_configurable(tuple_) cp_id = configurable.get("checkpoint_id") if not cp_id: @@ -476,7 +505,7 @@ async def enforce_thread_retention( guarded: set[tuple[str, str]] = set() if effective.strict_pending_write_guard: - guarded = await _checkpoint_ids_with_writes(saver, thread_id) + guarded = await _checkpoint_ids_with_writes(storage_saver, thread_id) deletable: list[tuple[str, str]] = [] for key, node in nodes.items(): @@ -511,13 +540,129 @@ async def enforce_thread_retention( if key not in deleted_keys: survivor_versions.update(node.versions) + # Invalidate before touching source-of-truth rows. If cache deletion + # fails, fail closed with every checkpoint intact; doing this after the + # first DELETE could leave a stale wrapper entry for a missing node. + invalidate_cache = getattr(saver, "ainvalidate_history_cache", None) + if deletable and callable(invalidate_cache): + await invalidate_cache(thread_id) + for key in deletable: - await _delete_checkpoint_rows(saver, thread_id, key) + await _delete_checkpoint_rows(storage_saver, thread_id, key) report.deleted_checkpoint_ids.append(key[1]) if deletable: - await _delete_unreachable_blobs(saver, thread_id, survivor_versions) + await _delete_unreachable_blobs(storage_saver, thread_id, survivor_versions) if collect_stats: - report.stats_after = await _thread_storage_stats(saver, thread_id) + report.stats_after = await _thread_storage_stats(storage_saver, thread_id) return report + + +async def enforce_trailing_duration_retention( + saver: BaseCheckpointSaver, + thread_id: str, + *, + protect_checkpoint_ids: frozenset[str] = frozenset(), + thread_lock: AbstractAsyncContextManager[None] | None = None, +) -> RetentionReport: + """Prune at most the current root duration leaf with bounded storage work. + + Read the head and its parent directly, then probe only the head's pending + writes. Identical channel-version maps prove that every blob referenced by + the leaf remains reachable through its parent, so no history scan, thread + statistics or blob GC is needed. Explicit admission selectors are protected. + Invalidate wrapper caches before deleting the source row; invalidation + failure must leave the row intact for delta reads. + + As with the exhaustive service, callers must exclude same-thread mutation + for this entire operation. Gateway supplies durable admission and the local + checkpoint lock before the worker's first checkpoint read. + """ + storage = _ensure_supported_saver(saver) + report = RetentionReport(thread_id=thread_id) + lock: AbstractAsyncContextManager[None] = thread_lock if thread_lock is not None else nullcontext() + async with lock: + latest = await storage.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}) + if latest is None: + return report + latest_config = checkpoint_configurable(latest) + latest_id = latest_config.get("checkpoint_id") + if not latest_id or latest_config.get("thread_id") != thread_id or latest_config.get("checkpoint_ns") or latest_id in protect_checkpoint_ids: + return report + parent_config = getattr(latest, "parent_config", None) + parent_values = parent_config.get("configurable") if isinstance(parent_config, dict) else None + if not isinstance(parent_values, dict): + return report + parent_id = parent_values.get("checkpoint_id") + if not parent_id or parent_id == latest_id or parent_values.get("thread_id") != thread_id or parent_values.get("checkpoint_ns"): + return report + parent = await storage.aget_tuple(parent_config) + if parent is None or checkpoint_configurable(parent).get("checkpoint_id") != parent_id: + return report + + latest_versions = latest.checkpoint.get("channel_versions") + parent_versions = parent.checkpoint.get("channel_versions") + if not isinstance(latest_versions, dict) or latest_versions != parent_versions: + return report + metadata = latest.metadata or {} + run_durations = metadata.get("run_durations") + latest_node = _Node( + ns="", + cp_id=latest_id, + parent_ns="", + parent_id=parent_id, + duration_only=is_duration_only_checkpoint(latest), + metadata_source=metadata.get("source"), + carries_run_durations=isinstance(run_durations, dict) and bool(run_durations), + versions=frozenset(latest_versions.values()), + ) + parent_node = _Node(ns="", cp_id=parent_id, parent_ns=None, parent_id=None, duration_only=False, versions=latest_node.versions) + _mark_duration_leaves_without_the_marker({("", latest_id): latest_node, ("", parent_id): parent_node}) + if not latest_node.duration_only or await _checkpoint_has_writes(storage, ("", latest_id), thread_id): + return report + + invalidate_cache = getattr(saver, "ainvalidate_history_cache", None) + if callable(invalidate_cache): + await invalidate_cache(thread_id) + await _delete_checkpoint_rows(storage, thread_id, ("", latest_id)) + report.protected_head_ids = {"": parent_id} + report.deleted_checkpoint_ids.append(latest_id) + return report + + +async def enforce_completed_run_retention( + *, + checkpointer: BaseCheckpointSaver, + thread_id: str, + protect_checkpoint_ids: frozenset[str] = frozenset(), +) -> RetentionReport: + """Run the conservative policy inside the next run's durable admission. + + Successful runs append one metadata-only duration checkpoint. If it is + left in place, the next run forks from it and turns that otherwise + disposable leaf into a protected ancestor forever. The run worker calls + this hook only after its active durable row excludes all same-thread run, + branch and checkpoint-write admissions, and before it reads a checkpoint. + The process-local checkpoint lock additionally serializes the local saver. + + Only the contract-proven trailing duration leaf is eligible. Sibling + branches and resumable checkpoints remain protected. Storage work is bounded + to the root head, its parent and one pending-write probe, independently of + thread history length. + """ + from deerflow.runtime.runs.worker import _checkpoint_thread_lock + + report = await enforce_trailing_duration_retention( + checkpointer, + thread_id, + protect_checkpoint_ids=protect_checkpoint_ids, + thread_lock=_checkpoint_thread_lock(thread_id), + ) + if report.deleted_checkpoint_ids: + logger.info( + "Checkpoint retention removed trailing metadata checkpoints for thread %s: %s", + thread_id, + report.deleted_checkpoint_ids, + ) + return report diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index e945c48f19..39466c4686 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -1185,6 +1185,53 @@ def get_subagent_batch_service(request: Request): return val +def _run_admission_hook(*, checkpointer: Any): + """Prune metadata leaves inside the run's durable admission fence.""" + + async def handle(record: Any, config: dict[str, Any]) -> None: + try: + from app.gateway.checkpoint_retention import enforce_completed_run_retention + + await enforce_completed_run_retention( + checkpointer=checkpointer, + thread_id=record.thread_id, + protect_checkpoint_ids=_selected_checkpoint_ids(config), + ) + except Exception: + logger.warning( + "Checkpoint retention failed for thread %s before run %s (non-fatal)", + record.thread_id, + record.run_id, + exc_info=True, + ) + + return handle + + +def _selected_checkpoint_ids(config: dict[str, Any]) -> frozenset[str]: + """Collect checkpoint selectors already validated for an admitted run.""" + configurable = config.get("configurable", {}) + selected: set[str] = set() + checkpoint_id = configurable.get("checkpoint_id") + if isinstance(checkpoint_id, str) and checkpoint_id: + selected.add(checkpoint_id) + checkpoint_map = configurable.get("checkpoint_map") + if isinstance(checkpoint_map, dict): + selected.update(value for value in checkpoint_map.values() if isinstance(value, str) and value) + return frozenset(selected) + + +def _run_completion_hook(*, scheduled_task_service: Any | None): + """Notify the scheduler after ordinary run finalization.""" + + async def handle(record: Any) -> None: + + if scheduled_task_service is not None: + await scheduled_task_service.handle_run_completion(record) + + return handle + + def get_run_context(request: Request) -> RunContext: """Build a :class:`RunContext` from ``app.state`` singletons. @@ -1198,9 +1245,11 @@ def get_run_context(request: Request) -> RunContext: host = getattr(request.app.state, "agent_runs_host", None) # Internal/channel and PAT runs deliberately receive no retained delegation. agent_runs = host.bind(request) if host is not None else None + checkpointer = get_checkpointer(request) + scheduled_task_service = getattr(request.app.state, "scheduled_task_service", None) return RunContext( agent_runs=agent_runs, - checkpointer=get_checkpointer(request), + checkpointer=checkpointer, store=get_store(request), event_store=get_run_event_store(request), run_events_config=getattr(request.app.state, "run_events_config", None), @@ -1210,7 +1259,10 @@ def get_run_context(request: Request) -> RunContext: mcp_task_repo=getattr(request.app.state, "mcp_task_repo", None), app_config=get_config(), extensions=getattr(request.app.state, "extensions", None), - on_run_completed=getattr(request.app.state, "scheduled_task_service", None).handle_run_completion if getattr(request.app.state, "scheduled_task_service", None) is not None else None, + on_run_admitted=_run_admission_hook(checkpointer=checkpointer), + on_run_completed=_run_completion_hook( + scheduled_task_service=scheduled_task_service, + ), ) diff --git a/backend/docs/checkpoint-retention-contract.md b/backend/docs/checkpoint-retention-contract.md index adcee94d08..2d1a2b1fd9 100644 --- a/backend/docs/checkpoint-retention-contract.md +++ b/backend/docs/checkpoint-retention-contract.md @@ -73,9 +73,9 @@ resume, (c) branch from an older visible turn, and (d) orphan row counts. ## Deletion mechanics -- Deletion must cover the backend's tables jointly and account for orphans, and - blob reachability must be computed from the **surviving checkpoints in a - whole-thread pass**: after deleting a checkpoint row, a `checkpoint_blobs` / +- Deletion must cover the backend's tables jointly and account for orphans. + General retention computes blob reachability from the **surviving checkpoints + in a whole-thread pass**: after deleting a checkpoint row, a `checkpoint_blobs` / `checkpoint_writes` row is an orphan only if *no surviving checkpoint* references it. The shared-version case is not hypothetical — the real duration-only checkpoint is a copy of the head checkpoint dict @@ -89,6 +89,11 @@ resume, (c) branch from an older visible turn, and (d) orphan row counts. payload-free: it materializes the parent's values under `{"writes": {"runtime_run_duration": {...}}, "source": "update", "step": ...}` metadata, which is what makes reclaiming it worthwhile.) +- Bounded admission cleanup instead proves that the root duration leaf and + its addressable parent have identical `channel_versions` maps. It deletes + only that leaf's checkpoint/writes rows and leaves all blobs intact: the + parent still references every version the leaf did. No whole-thread blob + scan or GC is needed for this shape. - Failure semantics: if a proposed deletion cannot be proven safe against the protected set, it must not ship. Partial deletion that leaves a dangling `parent_config` converts a cleanup into a thread-level outage (branch and @@ -116,21 +121,26 @@ duration-only leaf — which the next retention pass deletes again. Net effect without sequencing: the reclaimed row comes straight back, plus recurring store scans and an extra write per read. -The wiring PR that introduces the production trigger must therefore either: +Gateway runs retention only when the next run has won durable same-thread +admission, before that worker's first checkpoint read, and under the local +checkpoint lock. The cache leaf remains available while the thread is idle; +cleanup does not run from the completion callback. The admitted run's +`checkpoint_id` and every `checkpoint_map` selector are explicitly protected. -1. **Sequence retention away from history reads** — e.g. run retention on a - schedule whose next pass re-reclaims the re-created leaf, or run it when - the thread is not being read; or -2. **Adopt a policy that spares cache-carrying leaves** — e.g. a - `RetentionPolicy` flag that keeps any trailing duration-only leaf whose - metadata still carries `run_durations` / `run_message_ids` (same spirit - as the strict pending-writes guard), at the cost of not reclaiming that - leaf's rows. +The production `enforce_completed_run_retention` entrypoint delegates to +`enforce_trailing_duration_retention`: at most two checkpoint lookups (the +root head and its parent), one checkpoint-scoped pending-write existence +probe, and one leaf deletion. It performs no `alist`, whole-thread writes +audit, statistics or blob GC; work does not grow with retained history length. +The exhaustive `enforce_thread_retention` engine remains separate for +maintenance and contract measurements. A missing parent, changed channel +versions, or writes on the candidate makes admission cleanup a no-op. -Without either, E1 pruning and history reads churn against each other. This -decision belongs to the wiring PR, not to the storage-level service: the -service cannot tell a cache-carrying leaf from a payload-free one on the -alist path without re-implementing the writer's merge semantics. +Wrapper histories are invalidated through the public cache lifecycle API +before source deletion; an invalidation failure leaves the checkpoint intact. +Cleanup failures are non-fatal to admission. Unpinned delta state reads retry +once if their resolved target disappeared during hydration; explicit +historical reads never silently move to a different checkpoint. ## Item 4 note (large tool results) diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 0398a6ba3a..f53de1527c 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -50,7 +50,7 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **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). 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). +**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. diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_state.py b/backend/packages/harness/deerflow/runtime/checkpoint_state.py index 8867858f02..ba36931763 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_state.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_state.py @@ -148,6 +148,20 @@ class CheckpointStateAccessor: prepared = self._prepare_config(config) snapshot = await self.graph.aget_state(prepared) raise_if_snapshot_incompatible(snapshot, self.mode) + # Delta materialization may perform a second saver read after LangGraph + # has resolved the latest checkpoint. Retention can remove a trailing + # metadata-only leaf in that interval. CachedHistorySaver then cannot + # walk the vanished target's parent chain and the first materialization + # can contain an empty messages channel. For an unpinned latest-state + # read, verify that the resolved target still exists and retry once from + # the new head when it does not. Explicit historical selections remain + # fail-closed instead of silently moving to another checkpoint. + if self.mode == "delta" and not prepared["configurable"].get("checkpoint_id"): + snapshot_config = getattr(snapshot, "config", None) + checkpoint_id = (snapshot_config or {}).get("configurable", {}).get("checkpoint_id") + if checkpoint_id and await self.checkpointer.aget_tuple(snapshot_config) is None: + snapshot = await self.graph.aget_state(prepared) + raise_if_snapshot_incompatible(snapshot, self.mode) return snapshot def get_tuple(self, config: dict[str, Any]) -> Any | None: diff --git a/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py b/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py index 3408c55913..ca66f5084d 100644 --- a/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py +++ b/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py @@ -81,6 +81,11 @@ class CachedHistorySaver(BaseCheckpointSaver): self._compose_hits = 0 self._full_walks = 0 + @property + def source_saver(self) -> BaseCheckpointSaver: + """Return the saver that owns the authoritative checkpoint rows.""" + return self._inner + def __getattr__(self, name: str) -> Any: # Safety net for saver-specific extras (e.g. AsyncSqliteSaver.setup). # Base-class methods are explicitly delegated below, so this only @@ -336,6 +341,16 @@ class CachedHistorySaver(BaseCheckpointSaver): if delete is not None: await delete(self._key_prefix, thread_id) + async def ainvalidate_history_cache(self, thread_id: str) -> None: + """Drop cached delta histories after an external chain mutation. + + Storage-level retention deliberately operates on ``_inner`` because the + wrapper has no row-deletion API. Keeping invalidation on the wrapper + preserves its lifecycle contract without exposing cache internals to + the Gateway. + """ + await self._apurge_thread(thread_id) + async def acopy_thread(self, source_thread_id: str, target_thread_id: str) -> None: await self._inner.acopy_thread(source_thread_id, target_thread_id) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 4569e959da..345795a05d 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -675,6 +675,9 @@ class RunContext: # Delta snapshot cadence frozen at startup; ``None`` means "not frozen in # this process" (embedded/tests) and resolves to the config default. checkpoint_snapshot_frequency: int | None = None + # Called after durable same-thread admission is active and before the first + # checkpoint read. Failures are isolated from the run outcome. + on_run_admitted: Any | None = field(default=None) on_run_completed: Any | None = field(default=None) # The host binds this capability to one run's authenticated reader and references. conversation_reader: Any | None = field(default=None) @@ -1150,6 +1153,12 @@ async def run_agent( return started = True + if ctx.on_run_admitted is not None: + try: + await ctx.on_run_admitted(record, config) + except Exception: + logger.warning("Run admission hook failed for %s (non-fatal)", run_id, exc_info=True) + task_id = lead_task_id(run_id) if extensions.needs_task_store: task_store = ExtensionData(task_id) diff --git a/backend/tests/test_checkpoint_retention_service.py b/backend/tests/test_checkpoint_retention_service.py index 753845171f..dbaada4ab2 100644 --- a/backend/tests/test_checkpoint_retention_service.py +++ b/backend/tests/test_checkpoint_retention_service.py @@ -14,14 +14,17 @@ from __future__ import annotations import asyncio import copy +import logging import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager +from types import SimpleNamespace from typing import Annotated, Any, TypedDict from uuid import uuid4 import pytest from langchain_core.messages import AnyMessage, HumanMessage +from langgraph.channels import DeltaChannel from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver from langgraph.graph import END, START, StateGraph @@ -32,15 +35,28 @@ from app.gateway.checkpoint_lineage import find_checkpoint_before_message from app.gateway.checkpoint_retention import ( RetentionPolicy, _row_field, + enforce_completed_run_retention, enforce_thread_retention, ) -from deerflow.runtime.runs.worker import _new_checkpoint_marker, persist_run_durations +from deerflow.agents.thread_state import merge_message_writes +from deerflow.runtime.checkpoint_cache.base import make_history_key +from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache +from deerflow.runtime.checkpoint_state import CheckpointStateAccessor +from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver +from deerflow.runtime.runs.worker import _new_checkpoint_marker, persist_run_durations, persist_run_history_metadata class FullState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] +class DeltaState(TypedDict): + messages: Annotated[ + list[AnyMessage], + DeltaChannel(merge_message_writes, snapshot_frequency=2), + ] + + def _thread_id() -> str: return f"retention-service-{uuid4().hex}" @@ -604,6 +620,451 @@ async def test_empty_thread_reports_the_same_before_and_after_stats(saver_env: _ assert without_stats.stats_after == {} +@pytest.mark.anyio +@pytest.mark.parametrize("state_schema", [FullState, DeltaState], ids=["full", "delta"]) +async def test_completed_run_retention_removes_only_the_trailing_duration_leaf( + saver_env: _SaverEnv, + state_schema: Any, + caplog: pytest.LogCaptureFixture, +) -> None: + """The production hook removes the metadata leaf before a later run can + make it a permanent ancestor, without touching resumable checkpoints.""" + graph = _build_graph(state_schema, saver_env.saver) + thread_id = _thread_id() + await graph.ainvoke({"messages": [HumanMessage(content="hello")]}, _config(thread_id)) + resumable_head = await saver_env.saver.aget_tuple(_config(thread_id)) + assert resumable_head is not None + resumable_head_id = resumable_head.checkpoint["id"] + + assert await persist_run_durations( + checkpointer=saver_env.saver, + thread_id=thread_id, + durations={"run-1": 3}, + ) + duration_head = await saver_env.saver.aget_tuple(_config(thread_id)) + assert duration_head is not None + duration_head_id = duration_head.checkpoint["id"] + assert duration_head_id != resumable_head_id + + with caplog.at_level(logging.INFO, logger="app.gateway.checkpoint_retention"): + report = await enforce_completed_run_retention( + checkpointer=saver_env.saver, + thread_id=thread_id, + ) + + assert report.deleted_checkpoint_ids == [duration_head_id] + assert report.stats_before == {} + assert report.stats_after == {} + resumed = await saver_env.saver.aget_tuple(_config(thread_id)) + assert resumed is not None + assert resumed.checkpoint["id"] == resumable_head_id + materialized = await graph.aget_state(_config(thread_id)) + assert materialized.values["messages"][-1].content == "hello" + info_records = [record for record in caplog.records if record.levelno == logging.INFO] + assert len(info_records) == 1 + assert thread_id in info_records[0].message + assert duration_head_id in info_records[0].message + + await enforce_completed_run_retention(checkpointer=saver_env.saver, thread_id=thread_id) + assert len([record for record in caplog.records if record.levelno == logging.INFO]) == 1 + + +@pytest.mark.anyio +async def test_completed_run_retention_supports_cached_history_saver_and_invalidates_cache( + saver_env: _SaverEnv, +) -> None: + """The delta wrapper prunes through its source-of-truth saver and purges + cached histories whose checkpoint keys may have just been removed.""" + cache = MemoryCheckpointHistoryCache(max_entries=16) + wrapped = CachedHistorySaver(saver_env.saver, cache, key_prefix="retention-test") + graph = _build_graph(DeltaState, wrapped) + thread_id = _thread_id() + await graph.ainvoke({"messages": [HumanMessage(content="hello")]}, _config(thread_id)) + assert await persist_run_history_metadata( + checkpointer=wrapped, + thread_id=thread_id, + durations={"run-1": 3}, + message_run_ids={"message-1": "run-1"}, + ) + duration_head = await wrapped.aget_tuple(_config(thread_id)) + assert duration_head is not None + duration_head_id = duration_head.checkpoint["id"] + cache_key = make_history_key("retention-test", thread_id, "", duration_head_id, "messages") + await cache.aset_many({cache_key: {"writes": []}}) + assert cache.stats().entries == 1 + + report = await enforce_completed_run_retention( + checkpointer=wrapped, + thread_id=thread_id, + ) + + assert report.deleted_checkpoint_ids == [duration_head_id] + assert cache.stats().entries == 0 + assert await saver_env.saver.aget_tuple(_config_thread(thread_id, duration_head_id)) is None + + +@pytest.mark.anyio +async def test_sqlite_delta_state_read_retries_when_retention_deletes_resolved_leaf( + tmp_path: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A latest-state read must not lose history when retention removes the + duration leaf after LangGraph resolves it but before delta hydration.""" + async with _open_sqlite(tmp_path / "retention-read-race.sqlite") as inner: + wrapped = CachedHistorySaver( + inner, + MemoryCheckpointHistoryCache(max_entries=16), + key_prefix="retention-read-race", + ) + graph = _build_graph(DeltaState, wrapped) + accessor = CheckpointStateAccessor.bind(graph, wrapped, mode="delta") + thread_id = _thread_id() + expected = [HumanMessage(content="history survives", id="message-1")] + await graph.ainvoke({"messages": expected}, _config(thread_id)) + assert await persist_run_history_metadata( + checkpointer=wrapped, + thread_id=thread_id, + durations={"run-1": 3}, + message_run_ids={"message-1": "run-1"}, + ) + duration_head = await inner.aget_tuple(_config(thread_id)) + assert duration_head is not None + duration_head_id = duration_head.checkpoint["id"] + + target_resolved = asyncio.Event() + allow_hydration = asyncio.Event() + original_aget_tuple = inner.aget_tuple + paused = False + + async def pause_before_hydrating_target(config: dict[str, Any]) -> Any: + nonlocal paused + checkpoint_id = config.get("configurable", {}).get("checkpoint_id") + if checkpoint_id == duration_head_id and not paused: + paused = True + target_resolved.set() + await allow_hydration.wait() + return await original_aget_tuple(config) + + monkeypatch.setattr(inner, "aget_tuple", pause_before_hydrating_target) + read_task = asyncio.create_task(accessor.aget(_config(thread_id))) + await asyncio.wait_for(target_resolved.wait(), timeout=2) + + report = await enforce_completed_run_retention( + checkpointer=wrapped, + thread_id=thread_id, + ) + assert report.deleted_checkpoint_ids == [duration_head_id] + allow_hydration.set() + + snapshot = await asyncio.wait_for(read_task, timeout=2) + messages = snapshot.values["messages"] + assert [(message.id, message.content) for message in messages] == [ + ("message-1", "history survives"), + ] + assert snapshot.config["configurable"]["checkpoint_id"] != duration_head_id + + +@pytest.mark.anyio +async def test_completed_run_retention_preserves_selected_duration_leaf(saver_env: _SaverEnv) -> None: + """An admitted run may explicitly resume the metadata leaf it exposed.""" + graph = _build_graph(FullState, saver_env.saver) + thread_id = _thread_id() + await graph.ainvoke({"messages": [HumanMessage(content="hello")]}, _config(thread_id)) + assert await persist_run_durations( + checkpointer=saver_env.saver, + thread_id=thread_id, + durations={"run-1": 3}, + ) + selected = await saver_env.saver.aget_tuple(_config(thread_id)) + assert selected is not None + selected_id = selected.checkpoint["id"] + + report = await enforce_completed_run_retention( + checkpointer=saver_env.saver, + thread_id=thread_id, + protect_checkpoint_ids=frozenset({selected_id}), + ) + + assert report.deleted_checkpoint_ids == [] + assert await saver_env.saver.aget_tuple(_config_thread(thread_id, selected_id)) is not None + + +@pytest.mark.anyio +@pytest.mark.parametrize("steps", [1, 40], ids=["short", "long"]) +@pytest.mark.parametrize("duration_tail", [False, True], ids=["normal-head", "duration-head"]) +async def test_admission_retention_bounds_storage_work_independently_of_history( + saver_env: _SaverEnv, + monkeypatch: pytest.MonkeyPatch, + steps: int, + duration_tail: bool, +) -> None: + thread_id, checkpoint_ids, _ = await _write_turns(saver_env, steps) + duration_id = await _append_duration_checkpoint(saver_env, thread_id) if duration_tail else None + original_get = saver_env.saver.aget_tuple + reads: list[dict[str, Any]] = [] + + async def count_read(config: dict[str, Any]) -> Any: + reads.append(config) + assert (config["configurable"].get("checkpoint_ns") or "") == "" + return await original_get(config) + + def forbid_scan(*_args: Any, **_kwargs: Any) -> Any: + raise AssertionError("admission retention must not scan history, writes, stats or blobs") + + monkeypatch.setattr(saver_env.saver, "aget_tuple", count_read) + monkeypatch.setattr(saver_env.saver, "alist", forbid_scan) + for helper in ("_thread_storage_stats", "_checkpoint_ids_with_writes", "_delete_unreachable_blobs"): + monkeypatch.setattr(checkpoint_retention, helper, forbid_scan) + + report = await enforce_completed_run_retention(checkpointer=saver_env.saver, thread_id=thread_id) + + assert len(reads) <= 2 + assert report.deleted_checkpoint_ids == ([duration_id] if duration_tail else []) + assert report.stats_before == report.stats_after == {} + head = await original_get(_config(thread_id)) + assert head is not None + assert head.checkpoint["id"] == checkpoint_ids[-1] + # The parent retains all shared blobs, even though no blob GC ran. + materialized = await _build_graph(FullState, saver_env.saver).aget_state(_config(thread_id)) + assert len(materialized.values["messages"]) == steps + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "shape", + ["pending-writes", "missing-parent", "no-parent-link", "changed-versions", "swapped-channels", "markerless", "empty-durations", "empty-write-entry"], +) +async def test_admission_retention_checks_only_a_proven_metadata_leaf( + saver_env: _SaverEnv, + monkeypatch: pytest.MonkeyPatch, + shape: str, +) -> None: + thread_id, _, _ = await _write_turns(saver_env, 2) + duration_id = await _append_duration_checkpoint(saver_env, thread_id) + original_get = saver_env.saver.aget_tuple + head = await original_get(_config(thread_id)) + assert head is not None + parent = await original_get(head.parent_config) + assert parent is not None + parent_id = parent.checkpoint["id"] + if shape == "empty-write-entry" and saver_env.kind == "memory": + saver_env.saver.writes[(thread_id, "", duration_id)] = {} + elif shape == "pending-writes": + await saver_env.saver.aput_writes(head.config, [("messages", [HumanMessage(content="pending")])], task_id="pending") + elif shape in {"markerless", "empty-durations"}: + metadata = copy.deepcopy(head.metadata) + metadata.pop("writes", None) + if shape == "empty-durations": + metadata["run_durations"] = {} + await saver_env.saver.aput(head.parent_config, head.checkpoint, metadata, {}) + elif shape in {"changed-versions", "swapped-channels"}: + checkpoint = copy.deepcopy(head.checkpoint) + versions = checkpoint["channel_versions"] + if shape == "changed-versions": + versions["messages"] = "different-version" + else: + # Equal sets of version values do not prove identical channel maps. + versions["renamed-messages"] = versions.pop("messages") + await saver_env.saver.aput(head.parent_config, checkpoint, head.metadata, {}) + elif shape in {"missing-parent", "no-parent-link"}: + + async def incomplete_chain(config: dict[str, Any]) -> Any: + result = await original_get(config) + if result is None: + return None + if shape == "missing-parent" and result.checkpoint["id"] == parent_id: + return None + if shape == "no-parent-link" and result.checkpoint["id"] == duration_id: + return result._replace(parent_config=None) + return result + + monkeypatch.setattr(saver_env.saver, "aget_tuple", incomplete_chain) + + report = await enforce_completed_run_retention(checkpointer=saver_env.saver, thread_id=thread_id) + + assert report.deleted_checkpoint_ids == ([duration_id] if shape in {"markerless", "empty-write-entry"} else []) + retained = await original_get(_config_thread(thread_id, duration_id)) + assert (retained is None) == (shape in {"markerless", "empty-write-entry"}) + + +@pytest.mark.anyio +@pytest.mark.parametrize("invalidation_fails", [False, True], ids=["invalidate", "fail-closed"]) +async def test_admission_retention_invalidates_delta_cache_before_deleting_source( + saver_env: _SaverEnv, + monkeypatch: pytest.MonkeyPatch, + invalidation_fails: bool, +) -> None: + cache = MemoryCheckpointHistoryCache(max_entries=16) + wrapped = CachedHistorySaver(saver_env.saver, cache, key_prefix="retention-order") + graph = _build_graph(DeltaState, wrapped) + thread_id = _thread_id() + await graph.ainvoke({"messages": [HumanMessage(content="preserved", id="message-1")]}, _config(thread_id)) + assert await persist_run_history_metadata(checkpointer=wrapped, thread_id=thread_id, durations={"run-1": 3}) + head = await wrapped.aget_tuple(_config(thread_id)) + assert head is not None + duration_id = head.checkpoint["id"] + key = make_history_key("retention-order", thread_id, "", duration_id, "messages") + await cache.aset_many({key: {"writes": []}}) + original_invalidate = wrapped.ainvalidate_history_cache + invalidated: list[str] = [] + + async def invalidate_before_delete(selected_thread: str) -> None: + assert await wrapped.source_saver.aget_tuple(_config_thread(thread_id, duration_id)) is not None + invalidated.append(selected_thread) + if invalidation_fails: + raise RuntimeError("cache unavailable") + await original_invalidate(selected_thread) + + monkeypatch.setattr(wrapped, "ainvalidate_history_cache", invalidate_before_delete) + if invalidation_fails: + with pytest.raises(RuntimeError, match="cache unavailable"): + await enforce_completed_run_retention(checkpointer=wrapped, thread_id=thread_id) + assert await saver_env.saver.aget_tuple(_config_thread(thread_id, duration_id)) is not None + assert cache.stats().entries == 1 + else: + report = await enforce_completed_run_retention(checkpointer=wrapped, thread_id=thread_id) + assert report.deleted_checkpoint_ids == [duration_id] + assert cache.stats().entries == 0 + assert (await graph.aget_state(_config(thread_id))).values["messages"][-1].content == "preserved" + assert invalidated == [thread_id] + + +@pytest.mark.anyio +async def test_admission_retention_leaves_subgraph_and_other_thread_checkpoints_untouched(saver_env: _SaverEnv) -> None: + graph = _build_nested_graph(saver_env.saver) + thread_id = _thread_id() + await graph.ainvoke({"value": 10}, _config(thread_id)) + before = {tuple_.checkpoint["id"] async for tuple_ in saver_env.saver.alist(_config(thread_id), limit=None)} + other_thread, _, _ = await _write_turns(saver_env, 1) + other_leaf = await _append_duration_checkpoint(saver_env, other_thread) + duration_id = await _append_duration_checkpoint(saver_env, thread_id) + + report = await enforce_completed_run_retention(checkpointer=saver_env.saver, thread_id=thread_id) + + assert report.deleted_checkpoint_ids == [duration_id] + assert await _listed_checkpoint_ids(saver_env, thread_id) == before + assert await saver_env.saver.aget_tuple(_config_thread(other_thread, other_leaf)) is not None + assert (await graph.aget_state(_config(thread_id))).values["value"] == 11 + + +@pytest.mark.anyio +@pytest.mark.parametrize("empty_thread", [False, True], ids=["selected-head", "empty-thread"]) +async def test_admission_retention_skips_selected_or_missing_head_without_parent_reads( + saver_env: _SaverEnv, + monkeypatch: pytest.MonkeyPatch, + empty_thread: bool, +) -> None: + thread_id = _thread_id() + protected = frozenset() + if not empty_thread: + thread_id, _, _ = await _write_turns(saver_env, 1) + protected = frozenset({await _append_duration_checkpoint(saver_env, thread_id)}) + original_get = saver_env.saver.aget_tuple + reads = 0 + + async def read_head(config: dict[str, Any]) -> Any: + nonlocal reads + reads += 1 + assert "checkpoint_id" not in config["configurable"] + return await original_get(config) + + def forbid_scan(*_args: Any, **_kwargs: Any) -> Any: + raise AssertionError("selected or missing heads need no history scan") + + monkeypatch.setattr(saver_env.saver, "aget_tuple", read_head) + monkeypatch.setattr(saver_env.saver, "alist", forbid_scan) + report = await enforce_completed_run_retention(checkpointer=saver_env.saver, thread_id=thread_id, protect_checkpoint_ids=protected) + assert reads == 1 + assert report.deleted_checkpoint_ids == [] + + +@pytest.mark.anyio +async def test_gateway_context_keeps_agent_runs_binding_and_retention_hooks(monkeypatch: pytest.MonkeyPatch) -> None: + """Main's delegation binding and admission cleanup must share one context.""" + from app.gateway import deps + + saver = InMemorySaver() + wrapped = CachedHistorySaver(saver, MemoryCheckpointHistoryCache(max_entries=16), key_prefix="context-retention") + thread_id = _thread_id() + await _build_graph(DeltaState, wrapped).ainvoke({"messages": [HumanMessage(content="preserved")]}, _config(thread_id)) + assert await persist_run_durations(checkpointer=wrapped, thread_id=thread_id, durations={"prior": 1}) + leaf = await saver.aget_tuple(_config(thread_id)) + assert leaf is not None + leaf_id = leaf.checkpoint["id"] + delegation = object() + bound_requests: list[Any] = [] + completed: list[Any] = [] + + class Host: + def bind(self, request: Any) -> Any: + bound_requests.append(request) + return delegation + + class Scheduler: + async def handle_run_completion(self, record: Any) -> None: + completed.append(record) + + request = SimpleNamespace( + app=SimpleNamespace( + state=SimpleNamespace( + agent_runs_host=Host(), + scheduled_task_service=Scheduler(), + checkpointer=wrapped, + store=object(), + run_event_store=object(), + thread_store=object(), + ) + ) + ) + monkeypatch.setattr(deps, "get_config", lambda: SimpleNamespace()) + context = deps.get_run_context(request) + + assert context.agent_runs is delegation + assert bound_requests == [request] + assert context.checkpointer is wrapped + record = SimpleNamespace(thread_id=thread_id, run_id="next") + await context.on_run_admitted(record, {"configurable": {"checkpoint_map": {"": leaf_id}}}) + assert await saver.aget_tuple(_config_thread(thread_id, leaf_id)) is not None + await context.on_run_admitted(record, _config(thread_id)) + assert await saver.aget_tuple(_config_thread(thread_id, leaf_id)) is None + await context.on_run_completed(record) + assert completed == [record] + + +@pytest.mark.anyio +async def test_gateway_admission_retention_failure_isolated_from_scheduler_completion(monkeypatch: pytest.MonkeyPatch) -> None: + """Retention remains best-effort and scheduled completion stays separate.""" + from app.gateway import checkpoint_retention as retention_module + from app.gateway.deps import _run_admission_hook, _run_completion_hook + + calls: list[tuple[str, Any]] = [] + + async def fail_retention(*, checkpointer: Any, thread_id: str, protect_checkpoint_ids: frozenset[str]): + calls.append(("retention", (checkpointer, thread_id, protect_checkpoint_ids))) + raise RuntimeError("checkpoint backend unavailable") + + class _Scheduler: + async def handle_run_completion(self, record: Any) -> None: + calls.append(("scheduler", record)) + + monkeypatch.setattr(retention_module, "enforce_completed_run_retention", fail_retention) + checkpointer = object() + record = SimpleNamespace(thread_id="thread-1", run_id="run-1") + + await _run_admission_hook(checkpointer=checkpointer)( + record, + {"configurable": {"checkpoint_id": "selected", "checkpoint_map": {"child": "mapped"}}}, + ) + await _run_completion_hook( + scheduled_task_service=_Scheduler(), + )(record) + + assert calls == [ + ("retention", (checkpointer, "thread-1", frozenset({"selected", "mapped"}))), + ("scheduler", record), + ] + + # --------------------------------------------------------------------------- # Namespaced (persistent subgraph) histories # --------------------------------------------------------------------------- diff --git a/backend/tests/test_run_worker_rollback.py b/backend/tests/test_run_worker_rollback.py index 62d7c719f5..d40fbe0dbb 100644 --- a/backend/tests/test_run_worker_rollback.py +++ b/backend/tests/test_run_worker_rollback.py @@ -44,6 +44,7 @@ from deerflow.runtime.runs.worker import ( _LargeFileToolChunkBatcher, _rollback_to_pre_run_checkpoint, _try_extract_from_message, + persist_run_durations, run_agent, ) from deerflow.sandbox.lease import ( @@ -70,6 +71,97 @@ def _lease_test_bridge(): ) +@pytest.mark.anyio +async def test_admission_hook_holds_durable_thread_exclusivity_until_checkpoint_preflight() -> None: + """A peer cannot admit a run while retention is paused in the pre-read hook.""" + store = MemoryRunStore() + owner = RunManager(store=store, worker_id="owner") + peer = RunManager(store=store, worker_id="peer") + record = await owner.create_or_reject("retention-admission-race") + hook_entered = asyncio.Event() + release_hook = asyncio.Event() + + async def hold_admission(_record, _config) -> None: + hook_entered.set() + await release_hook.wait() + + class EmptyAgent: + async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False): + del graph_input, config, stream_mode, subgraphs + yield {"messages": []} + + task = asyncio.create_task( + run_agent( + _lease_test_bridge(), + owner, + record, + ctx=RunContext(checkpointer=None, on_run_admitted=hold_admission), + agent_factory=lambda **_kwargs: EmptyAgent(), + graph_input={}, + config={}, + ) + ) + await asyncio.wait_for(hook_entered.wait(), timeout=1) + with pytest.raises(ConflictError, match="active run"): + await peer.create_or_reject("retention-admission-race") + + release_hook.set() + await asyncio.wait_for(task, timeout=1) + assert record.status == RunStatus.success + + +@pytest.mark.anyio +async def test_admission_cleanup_finishes_before_first_checkpoint_read(monkeypatch: pytest.MonkeyPatch) -> None: + """A real saver is pruned while admission is exclusive, before preflight reads.""" + from app.gateway.checkpoint_retention import enforce_completed_run_retention + + checkpointer = InMemorySaver() + thread_id = "retention-real-checkpointer" + config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + checkpoint = empty_checkpoint() + checkpoint["id"] = "00000000-0000-0000-0000-000000000001" + await checkpointer.aput(config, checkpoint, {"source": "input", "step": -1, "parents": {}}, {}) + assert await persist_run_durations(checkpointer=checkpointer, thread_id=thread_id, durations={"prior": 1}) + duration_head = await checkpointer.aget_tuple(config) + assert duration_head is not None + duration_id = duration_head.checkpoint["id"] + + store = MemoryRunStore() + manager = RunManager(store=store, worker_id="owner") + peer = RunManager(store=store, worker_id="peer") + record = await manager.create_or_reject(thread_id) + + async def admission_hook(_record, _config) -> None: + await enforce_completed_run_retention(checkpointer=checkpointer, thread_id=thread_id) + + async def observe_preflight(*_args, **_kwargs): + head = await checkpointer.aget_tuple(config) + assert head is not None + assert head.checkpoint["id"] != duration_id + with pytest.raises(ConflictError, match="active run"): + await peer.create_or_reject(thread_id) + return None + + monkeypatch.setattr("deerflow.runtime.runs.worker._capture_rollback_point", observe_preflight) + + class EmptyAgent: + async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False): + del graph_input, config, stream_mode, subgraphs + yield {"messages": []} + + await run_agent( + _lease_test_bridge(), + manager, + record, + ctx=RunContext(checkpointer=checkpointer, on_run_admitted=admission_hook), + agent_factory=lambda **_kwargs: EmptyAgent(), + graph_input={}, + config=config, + ) + + assert record.status == RunStatus.success + + @pytest.mark.anyio async def test_run_agent_releases_execution_lease_when_graph_raises(): provider = MagicMock() diff --git a/frontend/src/content/en/harness/checkpoints/operations.mdx b/frontend/src/content/en/harness/checkpoints/operations.mdx index 8b9e1f5d35..5351931775 100644 --- a/frontend/src/content/en/harness/checkpoints/operations.mdx +++ b/frontend/src/content/en/harness/checkpoints/operations.mdx @@ -77,10 +77,10 @@ Exactly two shapes are proven safe, and the retention service implements only th Two mechanical constraints come with deletion: -- **Joint table deletion, orphan-first.** Deleting a checkpoint row requires deciding which writes and blob rows are orphaned; a row is an orphan only if no *surviving* checkpoint references it, computed in a whole-thread pass. A real duration-only leaf is not payload-free — it materializes the parent's values — and on Postgres its blob rows are the same rows backing its parent. +- **Preserve shared storage.** General retention computes blob reachability in a whole-thread pass. Admission cleanup instead requires the root duration leaf and its parent to have identical channel-version maps, deletes only the leaf's checkpoint/writes rows, and leaves all shared blobs intact. - **All-or-nothing.** A partial deletion that leaves a dangling parent link converts a cleanup into a thread-level outage: branch and regenerate fail loudly for every later turn. -There is also a history fast-path interaction to sequence around: the trailing duration-only leaf carries the `run_durations` and `run_message_ids` maps that `get_thread_history` reads from the latest checkpoint. Deleting the leaf makes the next history read fall back to store scans and re-persist a fresh leaf, so any wiring must sequence retention away from history reads or spare cache-carrying leaves. +There is also a history fast-path interaction: the trailing duration-only leaf carries the `run_durations` and `run_message_ids` maps that `get_thread_history` reads from the latest checkpoint. The Gateway keeps that leaf available for history reads until the next run wins durable same-thread admission, then deletes it under the checkpoint lock before the run's first checkpoint read. Cleanup reads at most the root head and its parent and probes only the candidate's pending writes; it performs no history scan, statistics pass or blob GC. Explicit `checkpoint_id` and `checkpoint_map` selections are protected. Delta-history caches are invalidated before deletion, and a cache failure preserves the source checkpoint. ### Policy defaults @@ -92,11 +92,11 @@ There is also a history fast-path interaction to sequence around: the trailing d | `strict_pending_write_guard` | `True` | Treat any checkpoint that still owns writes rows as protected | | `max_delete_per_run` | `None` | Optional cap on deletions per run | -### There is no production trigger yet +### Production retention trigger -The retention service (`backend/app/gateway/checkpoint_retention.py`) ships **without a production trigger**: where retention is invoked from — post-run hook, scheduler, or explicit admin action — is a maintainer decision that lands with the contract itself. Nothing in DeerFlow calls it, and there is no production caller of `prune` or `aprune` either: in the pinned environment only the base implementation defines them, and it raises `NotImplementedError`. +After a run wins durable same-thread admission, the Gateway calls the conservative retention policy before its first checkpoint read. The admission fence and per-thread checkpoint lock exclude concurrent runs and writers; explicitly selected checkpoint IDs are protected. The hook deletes at most one contract-proven trailing metadata-only leaf and treats failures as non-fatal. It does not prune resumable ancestors or sibling branches. -Treat this as unfinished work rather than a safe default. The executable contract proves the branch-ancestor, explicit-resume-target, pending-writes, leaf-duration, and leaf-sibling scenarios against the **full-mode** state schema; only the growth baseline is delta-shaped. Delta-specific prune safety is neither proved by the suite nor implemented. +The production trailing-duration-leaf deletion has regressions against both **full** and **delta** state schemas. The broader branch-ancestor, explicit-resume-target, pending-writes, and sibling-branch contract is still primarily full-shaped; production does not attempt those deletions without a future rebase/TTL contract. ## The one remaining patch diff --git a/frontend/src/content/en/harness/checkpoints/reference.mdx b/frontend/src/content/en/harness/checkpoints/reference.mdx index db1629b0db..f12ff25128 100644 --- a/frontend/src/content/en/harness/checkpoints/reference.mdx +++ b/frontend/src/content/en/harness/checkpoints/reference.mdx @@ -361,7 +361,7 @@ State at the pinned set (`langgraph 1.2.9`, `langgraph-checkpoint 4.2.0`, `langg | `#8448` | upstream langgraph issue; **open**; fix PR `#8556` unmerged | `PostgresSaver.get_delta_channel_history` permanently poisons the walk cursor when the target checkpoint is not in the first pagination page, so old checkpoints hydrate empty; only Postgres pages its stage-1 scan | Assertion stays live on memory/sqlite (differential against an `InMemorySaver` oracle); Postgres trigger-skips; `test_delta_channel_checkpointers.py::test_long_chain_history_survives_pagination` | `langgraph-checkpoint-postgres 3.1.2` | `test_delta_channel_checkpointers.py:459` | | `#8548` | upstream langgraph PR; **open**; fixes upstream `#8443`, with related issue `#8551` still waiting on it | Does not replay an abandoned branch into a `DeltaChannel` fork | While unmerged the local linearization cannot be removed: `_linearize_delta_checkpoint_resume` is retained | `langgraph 1.2.9` | `worker:2267-2279`, `test_run_worker_delta_resume.py::test_linearizes_a_delta_resume_onto_the_head` | | `#4458` | DeerFlow issue; the user-visible symptom of the fork bug | Regenerating in a branched thread resurrected the old assistant message beside the new one after a reload; reproduced on postgres, sqlite, and the in-memory saver, and it also broke branch history when seeded rows shared one run id | Linearization plus per-turn seeded run ids; `test_run_worker_delta_resume.py::test_regenerating_in_a_branched_thread_does_not_resurrect_the_old_answer` | `langgraph 1.2.9` | `worker:2267-2289`, `test_run_worker_delta_resume.py:204` | -| `#4189` item 3 | DeerFlow design discussion | Retention and deletion of checkpoint rows; what may be deleted | `backend/docs/checkpoint-retention-contract.md`, `backend/tests/test_checkpoint_retention_contract.py`, `backend/app/gateway/checkpoint_retention.py`; no production trigger is wired | — | `checkpoint_retention.py:1-40`, `backend/docs/checkpoint-retention-contract.md:31-77` | +| `#4189` item 3 | DeerFlow design discussion | Retention and deletion of checkpoint rows; what may be deleted | `backend/docs/checkpoint-retention-contract.md`, `backend/tests/test_checkpoint_retention_contract.py`, `backend/app/gateway/checkpoint_retention.py`; bounded cleanup runs at durable admission before checkpoint reads, protects explicit selections, and removes only the root trailing duration leaf | — | `checkpoint_retention.py`, `backend/docs/checkpoint-retention-contract.md:31-77` | | `#3859` (review) | DeerFlow review reference | An interrupted-title write must bump `channel_versions["title"]` and declare it in `new_versions`, or database savers drop the blob | `backend/tests/test_run_worker_rollback.py:3501-3504` pins the shape | — | `backend/tests/test_run_worker_rollback.py:3501-3504` | See [Operating Checkpoints](/docs/harness/checkpoints/operations) for what the remaining patch does and how the dependency floor is enforced, and [Troubleshooting](/docs/harness/checkpoints/troubleshooting) for the symptoms each defect produces for a user. diff --git a/frontend/src/content/zh/harness/checkpoints/operations.mdx b/frontend/src/content/zh/harness/checkpoints/operations.mdx index 970833313d..68034c7375 100644 --- a/frontend/src/content/zh/harness/checkpoints/operations.mdx +++ b/frontend/src/content/zh/harness/checkpoints/operations.mdx @@ -76,10 +76,10 @@ import { Callout } from "nextra/components"; 删除还带两条机械约束： -- **联合表删除、先算孤儿。** 删除检查点行时必须判定哪些 writes 与 blob 行成了孤儿；只有当**幸存**检查点都不再引用某行时它才算孤儿，且要在整线程范围内计算。真实的仅耗时叶子并非没有负载——它会物化父节点的值——在 Postgres 上它的 blob 行与支撑父节点的正是同一批行。 +- **保留共享存储。** 通用保留策略需要在整线程范围内计算 blob 可达性。Admission 清理则要求根命名空间的仅耗时叶子与父节点具有完全相同的通道版本映射，只删除该叶子的 checkpoint/writes 行，保留全部共享 blob。 - **要么全做，要么不做。** 留下悬空父链接的部分删除会把一次清理变成线程级故障：此后每一轮的分支与重新生成都会响亮失败。 -还有一条历史快路径需要排期：尾部的仅耗时叶子同时承载 `run_durations` 与 `run_message_ids` 映射，`get_thread_history` 会从最新检查点读取它们。删掉该叶子会让下一次历史读取退化为扫描事件存储并重新落一份新叶子，因此接线时必须把保留动作与历史读错开，或采用会保留「带缓存的叶子」的策略。 +这里还存在一条历史快路径交互：尾部的仅耗时叶子同时承载 `run_durations` 与 `run_message_ids` 映射，`get_thread_history` 会从最新检查点读取它们。Gateway 会保留该叶子供历史读取，直到下一次 run 获得持久化的同线程 admission；随后在首次 checkpoint 读取前、持有 checkpoint 锁时删除它。清理最多读取根命名空间的头节点与父节点，并只探测候选叶子的 pending writes；不扫描历史、不收集统计、不做 blob GC。显式 `checkpoint_id` 与 `checkpoint_map` 选择受到保护。删除前通过公共生命周期接口失效增量历史缓存；缓存失效失败时保留源检查点。 ### 策略默认值 @@ -91,11 +91,11 @@ import { Callout } from "nextra/components"; | `strict_pending_write_guard` | `True` | 仍持有 writes 行的检查点一律视为受保护 | | `max_delete_per_run` | `None` | 单次 run 删除数量的可选上限 | -### 目前没有生产触发点 +### 生产保留触发点 -保留服务（`backend/app/gateway/checkpoint_retention.py`）**没有生产触发点**：保留动作从哪里调用——run 结束后钩子、调度器还是显式管理操作——是维护者随契约一起决定的。DeerFlow 中没有任何地方调用它；`prune` 与 `aprune` 也没有生产调用方：在锁定环境中只有基类实现定义了它们，而基类实现直接抛 `NotImplementedError`。 +run 获得持久化的同线程 admission 后，Gateway 会在首次 checkpoint 读取前调用保守保留策略。admission fence 与逐线程 checkpoint 锁会排除并发 run 和写入；显式选择的 checkpoint ID 会加入保护集。该钩子每次最多删除一个已由契约证明安全的末尾纯元数据叶子，失败不会改变 run 结果，也不会删除可恢复祖先或兄弟分支。 -请把它当作未完成的工作，而不是安全默认值。可执行契约只在 **full 模式**的状态 schema 上证明了分支祖先、显式恢复目标、待处理写入、叶子耗时、叶子兄弟这几类场景；只有增长基线是增量形态的。增量专属的裁剪安全性既未被测试套件证明，也未被实现。 +生产触发的末尾耗时叶子删除在 **full** 和 **delta** schema 上都有回归测试。更广泛的分支祖先、显式恢复目标、待处理写入和兄弟分支契约仍以 full schema 为主；在没有 rebase/TTL 契约之前，生产策略不会尝试删除这些历史。 ## 仅存的一个补丁