From 2149909924a56916f41d5b2032c1bb84fd70fca0 Mon Sep 17 00:00:00 2001 From: Wang Runyuan <281843989+9s5bz2jvd2-lang@users.noreply.github.com> Date: Thu, 16 Jul 2026 03:29:58 -0700 Subject: [PATCH 1/2] feat(daemon): add semantic checkpoints --- src/lingtai/tools/daemon/ANATOMY.md | 12 +- src/lingtai/tools/daemon/CONTRACT.md | 40 ++- src/lingtai/tools/daemon/__init__.py | 122 ++++++- src/lingtai/tools/daemon/run_dir.py | 205 ++++++++++- tests/test_daemon_checkpoint.py | 512 +++++++++++++++++++++++++++ 5 files changed, 868 insertions(+), 23 deletions(-) create mode 100644 tests/test_daemon_checkpoint.py diff --git a/src/lingtai/tools/daemon/ANATOMY.md b/src/lingtai/tools/daemon/ANATOMY.md index c77e01df2..dee3f41b7 100644 --- a/src/lingtai/tools/daemon/ANATOMY.md +++ b/src/lingtai/tools/daemon/ANATOMY.md @@ -24,6 +24,7 @@ related_files: - tests/test_daemon.py - tests/test_apriori_summary_executor.py - tests/test_daemon_run_dir.py + - tests/test_daemon_checkpoint.py - tests/test_daemon_codex_usage.py - tests/test_codex_standalone_compaction.py - tests/test_daemon_detached_supervisor.py @@ -69,7 +70,7 @@ contract lives in `CONTRACT.md`. MCP-capable backends also get the built-in `daemon_common` MCP (`src/lingtai/mcp_servers/daemon_common/server.py:1-151`); its `finish` tool writes `daemon_completion.json`, and only a validated `finish(status="done")` allows terminal `done`. Parent MCP tools are not -auto-inherited. The daemon-eligible `email` intrinsic is available only when explicitly requested in a task's `tools` list, and daemon tool +auto-inherited. The daemon-eligible `email` intrinsic is available only when explicitly requested in a task's `tools` list, and each LingTai-backend run receives a private `checkpoint` tool only after its exact `DaemonRunDir` exists (`daemon/__init__.py:2529-2607`); `checkpoint` is not part of `_daemon_intrinsic_surface()` or any external CLI backend surface. Daemon tool calls still pass through the kernel `ToolExecutor`/`ToolCallGuard` path before any handler runs. Each `daemon.emanate` batch gets a stable `group_id` shared by every run in that batch, while each daemon still keeps its own `run_id`. Each @@ -102,7 +103,7 @@ polling. ## Components -- `daemon/__init__.py` — public capability surface. `get_description`, `get_schema`, and `setup`; the core class is `DaemonManager`, which manages the full emanation lifecycle and parent-stop cleanup. Key internals: `_ToolCollector` (`daemon/__init__.py:743`) intercepts `add_tool` calls during preset-driven capability setup to build a sandboxed tool surface without mutating the parent's registry. `EMANATION_BLACKLIST` (`daemon/__init__.py:65`) prevents recursion by blocking `daemon`, `avatar`, `psyche`, `skills`, and `knowledge`; `_parent_host_tool_floor()` (`daemon/__init__.py:76`) derives the NARROW set of always-on host tools a preset emanation may borrow from the parent — exactly `shell` + the file group (read/write/edit/glob/grep), derived from `CORE_DEFAULTS − EMANATION_BLACKLIST − {mcp}`; optional/provider parent tools (`vision`, `web_search`) are deliberately excluded so a preset that omits/fails a provider cap does not silently fall back to the parent; `_reconcile_terminal_notifications()` (`daemon/__init__.py:1116-1151`) retries terminal run dirs whose published receipt is missing, after stale-parent reaping; `_write_kimicode_mcp_config()` (`daemon/__init__.py:309-337`) writes the run-private Kimi `mcp.json` for stdio and HTTP registrations; `_daemon_intrinsic_surface()` (`daemon/__init__.py:1207`) is the narrow daemon intrinsic bridge for explicitly requested `email` plus the automatic LingTai `compact` schema; `_task_mcp_registrations()` (`daemon/__init__.py:1296`) validates full per-task MCP registrations and renders prompt-safe YAML; `_daemon_common_mcp_registration()` / `_read_daemon_completion()` (`daemon/__init__.py:1390` / `daemon/__init__.py:1460`) add and validate the built-in completion MCP; `_connect_task_mcp_registrations()` (`daemon/__init__.py:1550`) starts task-scoped MCP clients for the LingTai backend; `_daemon_provider_defaults()` (`daemon/__init__.py:1695`) preserves the parent/preset provider defaults for every provider and adds the per-run `daemon.json` cache anchor only for Codex; `_build_tool_surface()` (`daemon/__init__.py:1786`) automatically includes `compact`, includes only task-scoped MCP tools rather than auto-inheriting parent MCP tools, and — for preset-driven emanations — preserves the parent's narrow host tool floor (`_parent_host_tool_floor()`: shell + file primitives) that saved presets omit from `manifest.capabilities`, so requested host tools valid in the parent set are not rejected as unknown (resolution is preset-first, then intrinsics/MCP, then parent host fill-in); `_DaemonMetaState` (`daemon/__init__.py:177-333`) owns the per-emanation runtime/token/context snapshot, deterministic nine-round >=90% countdown, visible self-contained warning guidance, and expiry latch after the final warning's ordinary response opportunity. `_run_emanation()` (`daemon/__init__.py:2655`) is the LingTai-backend worker loop that builds a fresh daemon-scoped service for every no-preset run instead of reusing the parent service, sends LingTai `prompt` (or its exact default) as the first ordinary user message, projects daemon-local runtime/token/context state through `meta_block.attach_daemon_agent_meta` without parent notifications, performs sole-call provider-independent `compact` resets, mechanically compacts at expiry while retaining the latest assistant/tool-result pair and sending explicit recovery instructions, and enforces `finish(done)` before `mark_done`. +- `daemon/__init__.py` — public capability surface. `get_description`, `get_schema`, and `setup`; the core class is `DaemonManager`, which manages the full emanation lifecycle and parent-stop cleanup. Key internals: `_ToolCollector` (`daemon/__init__.py:743`) intercepts `add_tool` calls during preset-driven capability setup to build a sandboxed tool surface without mutating the parent's registry. `EMANATION_BLACKLIST` (`daemon/__init__.py:65`) prevents recursion by blocking `daemon`, `avatar`, `psyche`, `skills`, and `knowledge`; `_parent_host_tool_floor()` (`daemon/__init__.py:76`) derives the NARROW set of always-on host tools a preset emanation may borrow from the parent — exactly `shell` + the file group (read/write/edit/glob/grep), derived from `CORE_DEFAULTS − EMANATION_BLACKLIST − {mcp}`; optional/provider parent tools (`vision`, `web_search`) are deliberately excluded so a preset that omits/fails a provider cap does not silently fall back to the parent; `_reconcile_terminal_notifications()` (`daemon/__init__.py:1116-1151`) retries terminal run dirs whose published receipt is missing, after stale-parent reaping; `_write_kimicode_mcp_config()` (`daemon/__init__.py:309-337`) writes the run-private Kimi `mcp.json` for stdio and HTTP registrations; `_daemon_intrinsic_surface()` (`daemon/__init__.py:1207`) is the narrow daemon intrinsic bridge for explicitly requested `email` plus the automatic LingTai `compact` schema; `_task_mcp_registrations()` (`daemon/__init__.py:1296`) validates full per-task MCP registrations and renders prompt-safe YAML; `_daemon_common_mcp_registration()` / `_read_daemon_completion()` (`daemon/__init__.py:1390` / `daemon/__init__.py:1460`) add and validate the built-in completion MCP; `_connect_task_mcp_registrations()` (`daemon/__init__.py:1550`) starts task-scoped MCP clients for the LingTai backend; `_daemon_provider_defaults()` (`daemon/__init__.py:1695`) preserves the parent/preset provider defaults for every provider and adds the per-run `daemon.json` cache anchor only for Codex; `_build_tool_surface()` (`daemon/__init__.py:1786`) automatically includes `compact`, includes only task-scoped MCP tools rather than auto-inheriting parent MCP tools, and — for preset-driven emanations — preserves the parent's narrow host tool floor (`_parent_host_tool_floor()`: shell + file primitives) that saved presets omit from `manifest.capabilities`, so requested host tools valid in the parent set are not rejected as unknown (resolution is preset-first, then intrinsics/MCP, then parent host fill-in); `_DaemonMetaState` (`daemon/__init__.py:177-333`) owns the per-emanation runtime/token/context snapshot, deterministic nine-round >=90% countdown, visible self-contained warning guidance, and expiry latch after the final warning's ordinary response opportunity. `_run_emanation()` (`daemon/__init__.py:2655`) is the LingTai-backend worker loop that builds a fresh daemon-scoped service for every no-preset run instead of reusing the parent service, binds the run-scoped internal `checkpoint` tool after `run_dir` is known (`daemon/__init__.py:2529-2607`), sends LingTai `prompt` (or its exact default) as the first ordinary user message, projects daemon-local runtime/token/context state through `meta_block.attach_daemon_agent_meta` without parent notifications, performs sole-call provider-independent `compact` resets, mechanically compacts at expiry while retaining the latest assistant/tool-result pair and sending explicit recovery instructions, and enforces `finish(done)` before `mark_done`. - `daemon/system_prompt.py` — package-owned LingTai worker-prompt variant (`daemon/system_prompt.py:1-66`). It keeps the daemon operating contract short, names rather than re-describes host tools, teaches manual-before-first-use and precise `summary=true` use, maps parent `system.summarize` to daemon `compact`, preserves oneshot context plus the complete task, and rejects any final rendered prompt above 20,000 characters instead of truncating constraints. Its sibling architecture is the main-agent system-prompt graph in `src/lingtai/prompts/ANATOMY.md`; the daemon deliberately does not render that full resident section stack. @@ -110,7 +111,7 @@ polling. - `daemon/claude_interactive.py` — interactive Claude Code daemon backend. **Hidden / not user-selectable:** `claude` / `claude-interactive` were removed from the `get_schema()` `backend` enum and description (`daemon/__init__.py:889-904`); the code path stays only so older callers and stored daemon entries with `backend="claude"` still resolve through `_normalize_backend`/dispatch. New work uses print-mode `claude-p`. This retained PTY bridge is POSIX-only; Windows headless composition never selects it, and native interactive support remains deferred until ConPTY has its own accepted adapter. `ClaudeInteractiveBridge` (`daemon/claude_interactive.py:103`) accepts only the persistent Port injected by runtime composition; it never constructs a private adapter and rejects missing injection before managed-workspace, harness, or spawn work. It runs normal interactive `claude` under a PTY from a LingTai-managed workspace, writes the managed system prompt (`daemon/claude_interactive.py:80-96`), prepares empty or explicit-source detached worktrees (`daemon/claude_interactive.py:250-309`), answers terminal probes, injects `SessionStart`/`Stop` hooks via inline `--settings`, relays hook payloads through a FIFO, auto-selects workspace trust only inside the verified managed root (`daemon/claude_interactive.py:535-559`), and parses Claude transcript JSONL into daemon progress/result state. -- `daemon/run_dir.py` — per-emanation filesystem run directory. `DaemonRunDir` owns every filesystem effect for one run: folder layout, `daemon.json` atomic writes, batch `group_id` metadata (`DaemonRunDir.new_group_id()`), JSONL appends, CLI progress events, heartbeat touches, `result.txt`, versioned `daemon.json` data (`data_version`), visible `call_parameters`, terminal state markers, and backend resume-id persistence (`DaemonRunDir.set_session_id(key, value, *, overwrite)` — empty/duplicate values are no-ops returning `False`; `overwrite=False` keeps the first id for OpenCode-family backends; write failures propagate). The `DaemonManager` calls into a `DaemonRunDir` at every lifecycle hook without itself touching the filesystem. +- `daemon/run_dir.py` — per-emanation filesystem run directory. `DaemonRunDir` owns every filesystem effect for one run: folder layout, `daemon.json` atomic writes, batch `group_id` metadata (`DaemonRunDir.new_group_id()`), JSONL appends, CLI progress events, heartbeat touches, `result.txt`, versioned `daemon.json` data (`data_version`), visible `call_parameters`, terminal state markers, bounded semantic checkpoint constants (`daemon/run_dir.py:52-61`), `record_checkpoint()` (`daemon/run_dir.py:622-663`) for atomic `latest_checkpoint` sequence persistence plus best-effort checkpoint evidence/heartbeat writes, and backend resume-id persistence (`DaemonRunDir.set_session_id(key, value, *, overwrite)` — empty/duplicate values are no-ops returning `False`; `overwrite=False` keeps the first id for OpenCode-family backends; write failures propagate). The `DaemonManager` calls into a `DaemonRunDir` at every lifecycle hook without itself touching the filesystem. - `adapters/posix/daemon_execution_child_entrypoint.py` — fresh-interpreter execution boundary launched by the supervisor before its watcher. It registers exact execution PID/PGID/start identity and composes the existing production host; it does not duplicate backend parsers. `daemon_resume_owner_entrypoint.py` provides the bounded detached owner for one terminal supported-CLI resume generation. Durable `resume-claims/` records enforce one writer, and `followups/` plus `daemon.json` expose follow-up truth to `daemon(check)`. - `adapters/posix/process_identity.py` — shared POSIX process-incarnation helper. Linux identities combine boot ID and `/proc//stat` start ticks; Darwin/BSD identities use bounded `ps` start time plus PPID. It returns `None` when observation is unavailable, and all ownership-sensitive signal paths refuse unknown or mismatched identities. - `daemon/runtime.py` — stateless daemon backend runtime primitives shared by the LingTai in-process loop and transitional CLI runners. Port-owned Codex, Cursor, OpenCode-family, Qwen, and Kimi runners use the daemon-local process Port for stderr draining and stdout iteration; only ask workers use a local deadline, while initial streams remain watchdog-owned. The historical private names remain for unmigrated paths and compatibility tests. @@ -160,7 +161,7 @@ The `daemon` tool exposes five actions: | `emanate` | Spawn one or more subagents with specified task + tools + optional preset | | `list` | List running/completed/failed emanations with status and elapsed time | | `ask` | Send a follow-up message to a running emanation | -| `check` | Read-only progress tail: `daemon.json` state + last N events from `events.jsonl` + a compact `artifacts` block (the run's artifact manifest — relative path/size/mtime/role per important file, plus run-level state/result_path/error_path). On in-memory registry miss (e.g. after refresh/molt) falls back to the durable `daemons/*/` run dirs, resolving by full `run_id` (exact) or short `handle` (most-recent, with ambiguity flagged) | +| `check` | Read-only progress tail: `daemon.json` state + last N events from `events.jsonl` + a compact `artifacts` block (the run's artifact manifest — relative path/size/mtime/role per important file, plus run-level state/result_path/error_path) + `checkpoint` when `daemon.json.latest_checkpoint` exists (`daemon/__init__.py:7205-7207`). On in-memory registry miss (e.g. after refresh/molt) falls back to the durable `daemons/*/` run dirs, resolving by full `run_id` (exact) or short `handle` (most-recent, with ambiguity flagged) | | `list` | Progressive-disclosure index: active registry + historical run dirs; lazily rebuilds missing/invalid/stale-version `daemon.json` and returns prompt/result previews with search filters | | `reclaim` | Cancel all running emanations, shut down CLI process groups/thread pools through the same runtime-shutdown helper used by agent stop, reset ID counter | @@ -182,7 +183,7 @@ daemon/__init__.py ├── _build_tool_surface() — auto-includes `compact`, filters other requested tools against blacklist, expands groups, and merges preset/MCP/email surfaces; preset emanations also keep the NARROW parent host floor (`_parent_host_tool_floor()`: shell + file primitives only) so saved presets that omit core caps don't make requested host tools unknown — optional/provider parent tools (vision/web_search) are NOT borrowed and stay unknown unless the preset supplies them ├── _instantiate_preset_capabilities() — sets up preset tool surface in a sandbox ├── _build_emanation_prompt() — delegates to `system_prompt.py` for the bounded operating contract, compact host-tool names, complete task, and selected oneshot context - ├── _run_emanation() — LingTai worker loop: sends `prompt`/default first, runs ToolExecutor/ToolCallGuard, and performs same-run sole-call compact resets while retaining the exact call/result pair + ├── _run_emanation() — LingTai worker loop: adds the run-scoped `checkpoint` schema/handler, sends `prompt`/default first, runs ToolExecutor/ToolCallGuard, and performs same-run sole-call compact resets while retaining the exact call/result pair ├── _run_claude_interactive_emanation() — `claude` / `claude-interactive` backend; delegates to `run_claude_interactive()` (`daemon/claude_interactive.py:771`) to create the managed workspace, drive the interactive Claude TUI through PTY + hooks + transcript parsing, and persist managed-workspace state. ├── _run_claude_code_emanation() — `claude-p` / compatibility `claude-code` backend; parses `--output-format stream-json --verbose` print-mode events in real time so `claude_session_id`, per-turn text, and tool_use/tool_result land in DaemonRunDir during the run (vs. post-hoc). Claude Code's own `usage` fields are deliberately NOT forwarded to append_tokens (external billing path; semantics don't match the kernel's adapter accounting); `_normalize_claude_usage` requires the primary `input_tokens` and `output_tokens` fields, defaults the optional cache-read/cache-creation fields to zero, and rejects bool/negative/malformed consumed values (matching Codex/Cursor). The first valid terminal `result` event's usage is buffered — not persisted — until after cancellation/timeout, exit code, `is_error`, and `_require_done_completion` (the daemon_common `finish()` MCP contract — itself a terminal acceptance gate: when the run loaded `daemon_common`, a missing/bad `finish()` call fails the run here, before any usage is persisted or `mark_done` runs) are all classified/passed, then persisted exactly once UI-only to `daemon.json.cli_tokens` via `run_dir.record_cli_tokens` (`cached = cache_read + cache_creation` input tokens). Cancellation/timeout observed before the final acceptance check, a later failure classification, or a missing/bad completion sentinel arriving after the terminal line therefore cannot leave false accepted usage; cancellation published after that acceptance point retains the existing best-effort terminal semantics rather than introducing cross-backend atomic arbitration in this path. `_run_ask_claude_code_stream` (the `daemon(ask)` resume follow-up) applies the same buffer-then-classify ordering against its deadline/exit-code/`is_error` gates — resume has no `_require_done_completion` gate — so the TUI `/daemons` view can show accepted usage without touching either token ledger. ├── _run_codex_emanation() — codex backend; parses `codex exec --json` JSONL events (thread.started → codex_session_id, item.completed → agent_message text, turn.completed → terminal). Symmetric with the claude-code backend. A valid terminal `usage` object is normalized to disjoint input (`input_tokens - cached_input_tokens`, clamped at zero), cached input, and output in UI-only `daemon.json.cli_tokens`; the raw object is retained in a `cli_usage` event, with no token-ledger row and at most one terminal account per stream. Codex receives native config overrides for `daemon_common` plus parent-provided stdio MCP registrations, and `finish(done)` is required before success when daemon_common is loaded. @@ -220,6 +221,7 @@ daemon/run_dir.py ├── build_manifest() / write_manifest() — `build_manifest(run_path)` is a classmethod pure read over a run dir: lists path/size/mtime/inferred-role for well-known artifacts (priority order) + extra work-product files (sorted, `.tmp` skipped), capped at `_MANIFEST_MAX_ENTRIES` (records `artifacts_total`/`truncated`), plus run-level `state`/`result_path`/`error_path` (the last set only for failed/timeout/cancelled runs with a result.txt). Used both by `write_manifest()` (instance, called from every terminal marker after daemon.json/result.txt are final, best-effort via `_safe`, persists `artifacts.json`) and by `_artifacts_summary`'s fallback for old/running runs that have no `artifacts.json`. `MANIFEST_VERSION` mirrors `data_version` ├── record_user_send() — appends user-role entry to chat_history.jsonl ├── bump_turn() — marks end of LLM round (daemon.json + chat_history + heartbeat) + ├── record_checkpoint() — validates a bounded semantic checkpoint, assigns a monotonic run-local sequence under `_state_transaction`, atomically writes authoritative `daemon.json.latest_checkpoint`, then best-effort appends evidence/touches heartbeat ├── set_current_tool() — marks tool dispatch starting (daemon.json + events + heartbeat) ├── clear_current_tool() — marks tool dispatch finished ├── record_cli_output() — records CLI backend stdout/stderr as cli_output events diff --git a/src/lingtai/tools/daemon/CONTRACT.md b/src/lingtai/tools/daemon/CONTRACT.md index 4602a3239..bd267044c 100644 --- a/src/lingtai/tools/daemon/CONTRACT.md +++ b/src/lingtai/tools/daemon/CONTRACT.md @@ -40,6 +40,7 @@ related_files: - tests/test_daemon_cursor_backend.py - tests/test_daemon_claude_interactive_backend.py - tests/test_daemon_run_dir.py + - tests/test_daemon_checkpoint.py - tests/test_daemon_codex_usage.py - tests/test_codex_standalone_compaction.py - tests/test_daemon_windows_lock.py @@ -61,6 +62,7 @@ review_triggers: - tests/test_daemon_cursor_backend.py - tests/test_daemon_claude_interactive_backend.py - tests/test_daemon_run_dir.py + - tests/test_daemon_checkpoint.py - tests/test_daemon_codex_usage.py - tests/test_codex_standalone_compaction.py maintenance: | @@ -170,12 +172,33 @@ differs by provider:** a standalone-compaction failure is non-fatal for Codex HARD failure for native `mimo` (propagates to the caller; never silently continues on full history and never falls back to a different wire). +LingTai-backend runs also receive a private run-scoped internal `checkpoint` +tool after their exact `DaemonRunDir` exists. It is not part of the public +`daemon` action schema, not requestable through `tasks[].tools`, not inherited +from `_daemon_intrinsic_surface`, and not exposed to external CLI backends. A +daemon calls it only at meaningful semantic phase boundaries. Accepted calls +persist schema `lingtai.daemon.checkpoint.v1` in +`daemon.json.latest_checkpoint`, then best-effort append a bounded `checkpoint` +event and touch the heartbeat. The atomic state commit is the sole authority; +a secondary event/heartbeat write failure does not turn an already-committed +checkpoint into a retryable tool error. Each accepted call increments a +monotonic run-local sequence under the run-dir state transaction. Validated +fields are `phase_id` (1-96 UTF-8 bytes), `phase_status` (`complete`, `blocked`, +or `failed`), at most 8 completed rows (`id` <= 128 bytes, `outcome` <= 256 +bytes), optional advisory `next_action` (<= 512 bytes), and at most 8 evidence +refs (`event` narrow identifier or non-escaping run-relative `artifact` path, +ref <= 256 bytes). Model-supplied checkpoint text is passed through the kernel +trajectory redactor before persistence. A secret-bearing evidence ref is +rejected rather than persisted as either raw text or a broken redaction +placeholder; validation remains the authority for shape, bounds, and path +safety. + | Action | Required inputs | Optional inputs | Success output | Error shapes | |---|---|---|---|---| | `emanate` | `tasks[]` (each `task`+`tools`) | `backend`, `max_turns`, `timeout`, per-task `prompt` (LingTai only), `skills`/`mcp`/`preset`/`backend_options`/`context_token_limit` | `{status: "dispatched", count, ids: [...], group_id, handoff}`; `handoff` tells the model it may go idle or call `system(action='sleep')` while waiting for the terminal notification, and conditionally says that if Telegram is connected and a Task Card is available for the current turn, the model should use it to report progress via `telegram(action='manual')` and that manual's `Programmable Task Card` section; read `daemon-manual` and `notification-manual` for details | `{status: "error", message}` — obsolete `system_prompt` migration, CLI `prompt`, bad limits, or tool-surface/preset failure | | `list` | — | `contains`, `status`, `include_done` (default true), `last` | `{...}` list blob of matching emanations (running + persisted history) | `{status: "error", message}` | | `ask` | `id`, `message` | — | `{status: "sent", id, output}` (CLI ask returns immediately; `{status: "sent", id, async: true, ...}`) | `{status: "error", id, message}` — unknown/absent id, backend `ask` unsupported, or busy | -| `check` | `id` | `last` (default 20), `truncate` (default 500) | `{id, run_id, state, backend, path, turn, current_tool, elapsed_s, finished_at, tokens, result_preview, result_path, last_output, error, events: [...]}` | `{status: "error", message}` — unknown id, no run_dir, invalid `last`/`truncate`, or read failure | +| `check` | `id` | `last` (default 20), `truncate` (default 500) | `{id, run_id, state, backend, path, turn, current_tool, elapsed_s, finished_at, tokens, result_preview, result_path, last_output, error, events: [...], checkpoint?}` where `checkpoint` is present only when `daemon.json.latest_checkpoint` exists | `{status: "error", message}` — unknown id, no run_dir, invalid `last`/`truncate`, or read failure | | `reclaim` | — | — | `{status: "reclaimed", cancelled: }` (or `{status: "shutdown", ...}` on lifecycle shutdown) | — | `emanate` returns immediately after dispatch; terminal state (`done` / @@ -430,7 +453,7 @@ All paths are relative to the parent agent working directory (`/`): .heartbeat # mtime-touched on activity history/chat_history.jsonl # session transcript logs/token_ledger.jsonl # per-call tokens, daemon-scoped (source="daemon") - logs/events.jsonl # tool_call / tool_result / cli_output / cli_usage / daemon_* + logs/events.jsonl # tool_call / tool_result / checkpoint / cli_output / cli_usage / daemon_* result.txt # full terminal result when available /logs/token_ledger.jsonl # ALSO receives each daemon token row, tagged @@ -444,6 +467,12 @@ excludes daemon spend while `scope="all"` includes it. On daemon-manager startup stale `running`/`active` `daemon.json` records whose `parent_pid` is dead are reaped to `failed`. +`daemon.json.latest_checkpoint` is optional for backward compatibility. Old and +ordinary runs without it remain readable and `daemon(action="check")` omits the +`checkpoint` key. For live and historical runs that contain it, `check` projects +the stored checkpoint from disk; `events.jsonl` is supporting evidence, not a +second persistence authority. + ## Process and Terminal Boundaries ### External CLI process boundary @@ -622,6 +651,10 @@ Re-check this contract when touching: daemon_common completion, OpenCode-family routing, Qwen settings, Claude print MCP config, run-dir artifacts, prompt redaction, selected-skill catalog preservation, prompt mapping, or compact context reset behavior. +- `tests/test_daemon_checkpoint.py` coverage that proves the LingTai-only + run-scoped checkpoint surface, validation bounds, redaction, atomic sequence + persistence, old-run compatibility, `check` projection, and compact-reset + reconstruction. - `tests/test_codex_standalone_compaction.py` for per-task `context_token_limit` wiring/pre-flight validation. @@ -639,6 +672,7 @@ Re-check this contract when touching: | LingTai daemon tool results carry daemon-local `_meta.agent_meta`, omit parent notifications/guidance, and carry the exact warning only while current usage is >=90% | `src/lingtai/tools/daemon/__init__.py`, `src/lingtai/kernel/meta_block.py` | `tests/test_daemon.py::test_daemon_agent_meta_is_local_and_warning_tracks_current_usage` | | `compact.action` is required; `manual` is read-only, omission is refused, and explicit `run` resets with fresh post-compact metadata | `src/lingtai/tools/daemon/__init__.py` | `tests/test_daemon.py::test_compact_schema_requires_explicit_run_or_manual_action`, `::test_compact_missing_action_is_refused_without_reset`, `::test_compact_success_prunes_to_system_call_and_result` | | `reclaim` cancels running emanations; agent stop shuts the daemon down first | `src/lingtai/tools/daemon/__init__.py` | `tests/test_lifecycle_daemon_shutdown.py::test_agent_stop_shuts_down_daemon_before_heartbeat_and_lock` | +| LingTai daemon checkpoints are run-scoped, bounded, persisted atomically, and projected by `check` | `src/lingtai/tools/daemon/__init__.py`, `src/lingtai/tools/daemon/run_dir.py` | `tests/test_daemon_checkpoint.py` | ## Verification Matrix @@ -655,7 +689,7 @@ Re-check this contract when touching: Run before merging daemon tool-surface changes: ```bash -python -m pytest tests/test_daemon.py tests/test_daemon_check.py tests/test_daemon_backend_options.py tests/test_daemon_run_dir.py tests/test_lifecycle_daemon_shutdown.py tests/test_codex_standalone_compaction.py tests/test_mimo_responses_compaction.py -q +python -m pytest tests/test_daemon.py tests/test_daemon_check.py tests/test_daemon_backend_options.py tests/test_daemon_run_dir.py tests/test_daemon_checkpoint.py tests/test_lifecycle_daemon_shutdown.py tests/test_codex_standalone_compaction.py tests/test_mimo_responses_compaction.py -q ``` ## Schema and Glossary Ownership diff --git a/src/lingtai/tools/daemon/__init__.py b/src/lingtai/tools/daemon/__init__.py index d1ea1b9b5..1517937b7 100644 --- a/src/lingtai/tools/daemon/__init__.py +++ b/src/lingtai/tools/daemon/__init__.py @@ -2742,9 +2742,89 @@ def _run_emanation(self, em_id: str, run_dir, schemas, dispatch, service_kwargs["context_window"] = effective_preset_llm["context_window"] service = LLMService(**service_kwargs) + checkpoint_schema = FunctionSchema( + name="checkpoint", + description=( + "Record a bounded semantic checkpoint for this exact daemon " + "run after a meaningful phase boundary. Use this only when a " + "phase is complete, blocked, or failed; ordinary turns should " + "not call it. Persist only concise phase status, completed " + "outcomes, optional advisory next action, and run-local " + "evidence references." + ), + parameters={ + "type": "object", + "properties": { + "phase_id": { + "type": "string", + "description": "Non-empty phase identifier, max 96 UTF-8 bytes.", + }, + "phase_status": { + "type": "string", + "enum": ["complete", "blocked", "failed"], + }, + "completed": { + "type": "array", + "maxItems": 8, + "items": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Completed item id, max 128 UTF-8 bytes.", + }, + "outcome": { + "type": "string", + "description": "Bounded outcome, max 256 UTF-8 bytes.", + }, + }, + "required": ["id", "outcome"], + "additionalProperties": False, + }, + }, + "next_action": { + "type": "string", + "description": "Optional advisory text only, max 512 UTF-8 bytes.", + }, + "evidence_refs": { + "type": "array", + "maxItems": 8, + "items": { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": ["event", "artifact"], + }, + "ref": { + "type": "string", + "description": "Run-relative artifact path or narrow event id, max 256 UTF-8 bytes.", + }, + }, + "required": ["kind", "ref"], + "additionalProperties": False, + }, + }, + }, + "required": ["phase_id", "phase_status"], + "additionalProperties": False, + }, + ) + # ``checkpoint`` is a private, run-scoped daemon tool. Reserve the + # exact name here so an externally supplied schema cannot be paired + # with the internal handler below. + run_schemas = [ + schema + for schema in list(schemas or []) + if schema.name != checkpoint_schema.name + ] + run_schemas.append(checkpoint_schema) + run_dispatch = dict(dispatch) + run_dispatch["checkpoint"] = run_dir.record_checkpoint + session = service.create_session( system_prompt=run_dir.prompt_path.read_text(encoding="utf-8"), - tools=schemas or None, + tools=run_schemas or None, model=effective_model, thinking="default", tracked=False, @@ -2781,8 +2861,7 @@ def _run_emanation(self, em_id: str, run_dir, schemas, dispatch, service, run_dir, provider=provider, model=effective_model, endpoint=endpoint, ) - if "compact" in {schema.name for schema in schemas}: - dispatch = dict(dispatch) + if "compact" in {schema.name for schema in run_schemas}: def _compact_daemon_context(_args: dict) -> dict: nonlocal compact_reset_accepted @@ -2797,10 +2876,16 @@ def _compact_daemon_context(_args: dict) -> dict: if action != "run": return {"status": "error", "message": "action is required and must be 'run' or 'manual'; context was not reset"} if not compact_batch_allowed: - return {"status": "error", "message": "compact must be the only tool call in its batch"} + return { + "status": "error", + "message": "compact must be the only tool call in its batch", + } reason = _args.get("_reason") if not isinstance(reason, str) or not reason.strip(): - return {"status": "error", "message": "_reason must be a non-empty string; context was not reset"} + return { + "status": "error", + "message": "_reason must be a non-empty string; context was not reset", + } compact_reset_accepted = True return { "status": "success", @@ -2813,14 +2898,15 @@ def _compact_daemon_context(_args: dict) -> dict: }, } - dispatch["compact"] = _compact_daemon_context + run_dispatch["compact"] = _compact_daemon_context intrinsic_tool_names = set(self._daemon_intrinsic_surface()[1]) def _dispatch_daemon_tool(tc): - handler = dispatch.get(tc.name) + handler = run_dispatch.get(tc.name) if handler is None: from lingtai.kernel.types import UnknownToolError + raise UnknownToolError(tc.name) args = dict(tc.args or {}) if tc.name in intrinsic_tool_names: @@ -2828,17 +2914,21 @@ def _dispatch_daemon_tool(tc): return handler(args) def _daemon_tool_logger(event_type: str, **fields) -> None: - tool_name = fields.get("tool_name") or fields.get("tool") + safe_fields = dict(fields) + tool_name = safe_fields.get("tool_name") or safe_fields.get("tool") + if tool_name == "checkpoint" and "tool_args" in safe_fields: + safe_fields["tool_args"] = {} if event_type == "tool_call_normalized" and tool_name: - run_dir.set_current_tool(tool_name, fields.get("tool_args") or {}) + tool_args = safe_fields.get("tool_args") or {} + run_dir.set_current_tool(tool_name, tool_args) elif event_type == "tool_result" and tool_name: - status = "error" if fields.get("status") == "error" else "ok" + status = "error" if safe_fields.get("status") == "error" else "ok" run_dir.clear_current_tool(result_status=status) self._log( f"daemon_{event_type}", em_id=em_id, run_id=getattr(run_dir, "run_id", None), - **fields, + **safe_fields, ) executor = ToolExecutor( @@ -2847,7 +2937,7 @@ def _daemon_tool_logger(event_type: str, **fields) -> None: name, result, **kw ), guard=LoopGuard(), - known_tools=set(dispatch), + known_tools=set(run_dispatch), parallel_safe_tools=set(), logger_fn=_daemon_tool_logger, # Daemons receive a daemon-local agent_meta projection plus universal @@ -3013,7 +3103,7 @@ def _mechanically_compact(tool_results=None): ) session = service.create_session( system_prompt=system_prompt, - tools=schemas or None, + tools=run_schemas or None, model=effective_model, thinking="default", tracked=False, @@ -7446,7 +7536,7 @@ def _check_snapshot_from_paths( for k, v in ev.items()} events.append(ev) - return { + out = { "id": em_id, "run_id": state.get("run_id"), "state": state.get("state"), @@ -7472,6 +7562,10 @@ def _check_snapshot_from_paths( "events_total": events_total, "events_returned": len(events), } + checkpoint = state.get("latest_checkpoint") + if isinstance(checkpoint, dict): + out["checkpoint"] = checkpoint + return out def _artifacts_summary(self, run_path: Path) -> dict: """Compact artifact-manifest block for a `check` response. diff --git a/src/lingtai/tools/daemon/run_dir.py b/src/lingtai/tools/daemon/run_dir.py index 3b8a27229..ab7c93512 100644 --- a/src/lingtai/tools/daemon/run_dir.py +++ b/src/lingtai/tools/daemon/run_dir.py @@ -102,13 +102,23 @@ class DaemonRunDir: .heartbeat # mtime-touched on activity history/chat_history.jsonl # session transcript logs/token_ledger.jsonl # per-call tokens, daemon-scoped - logs/events.jsonl # tool_call, tool_result, cli_output, cli_usage, daemon_* + logs/events.jsonl # tool_call, tool_result, checkpoint, cli_output, cli_usage, daemon_* result.txt # full terminal result when available """ DATA_VERSION = 1 PENDING_LAUNCH_LEASE_S = 5.0 _TERMINAL_STATES = frozenset({"done", "failed", "cancelled", "timeout"}) + CHECKPOINT_SCHEMA = "lingtai.daemon.checkpoint.v1" + CHECKPOINT_PHASE_ID_MAX_BYTES = 96 + CHECKPOINT_COMPLETED_MAX_ROWS = 8 + CHECKPOINT_EVIDENCE_MAX_ROWS = 8 + CHECKPOINT_COMPLETED_ID_MAX_BYTES = 128 + CHECKPOINT_COMPLETED_OUTCOME_MAX_BYTES = 256 + CHECKPOINT_NEXT_ACTION_MAX_BYTES = 512 + CHECKPOINT_EVIDENCE_REF_MAX_BYTES = 256 + _CHECKPOINT_STATUSES = frozenset({"complete", "blocked", "failed"}) + _CHECKPOINT_EVIDENCE_KINDS = frozenset({"event", "artifact"}) def __init__( self, @@ -654,6 +664,199 @@ def set_session_id(self, key: str, value: str, *, overwrite: bool = True) -> boo self._atomic_write_json(self.daemon_json_path, self._state) return True + def record_checkpoint(self, payload: dict) -> dict: + """Validate and persist one run-scoped semantic checkpoint.""" + checkpoint = self._validate_checkpoint_payload(payload) + with self._state_transaction(): + current = self._state.get("latest_checkpoint") + current_sequence = 0 + if isinstance(current, dict): + raw_sequence = current.get("sequence") + if isinstance(raw_sequence, int) and not isinstance(raw_sequence, bool): + current_sequence = raw_sequence + sequence = current_sequence + 1 + checkpoint = { + "schema": self.CHECKPOINT_SCHEMA, + "run_id": self._run_id, + "checkpoint_id": f"{self._run_id}:{sequence}", + "sequence": sequence, + **checkpoint, + "created_at": self._now_iso(), + } + next_state = dict(self._state) + next_state["latest_checkpoint"] = checkpoint + self._atomic_write_json(self.daemon_json_path, next_state) + self._state = next_state + # daemon.json is the sole checkpoint authority. Evidence and the + # heartbeat are best-effort after that atomic commit so a transient + # append/touch failure cannot make the model retry an already-saved + # checkpoint and create an ambiguous extra sequence. + self._safe( + "checkpoint_event", + lambda: self._append_jsonl( + self.events_path, + { + "event": "checkpoint", + "checkpoint": checkpoint, + "turn": self._state.get("turn"), + "ts": checkpoint["created_at"], + }, + ), + ) + self._safe("checkpoint_heartbeat", self.heartbeat_path.touch) + return dict(checkpoint) + + def _validate_checkpoint_payload(self, payload: dict) -> dict: + from lingtai.kernel.trace_redaction import redact_for_trajectory + + if not isinstance(payload, dict): + raise ValueError("checkpoint payload must be an object") + allowed = { + "phase_id", + "phase_status", + "completed", + "next_action", + "evidence_refs", + } + unknown = set(payload) - allowed + if unknown: + raise ValueError(f"unknown checkpoint fields: {sorted(unknown)}") + + phase_id = self._checkpoint_text( + payload.get("phase_id"), + field="phase_id", + max_bytes=self.CHECKPOINT_PHASE_ID_MAX_BYTES, + ) + phase_status = payload.get("phase_status") + if ( + not isinstance(phase_status, str) + or phase_status not in self._CHECKPOINT_STATUSES + ): + raise ValueError("phase_status must be complete, blocked, or failed") + + completed = payload.get("completed", []) + if not isinstance(completed, list): + raise ValueError("completed must be a list") + if len(completed) > self.CHECKPOINT_COMPLETED_MAX_ROWS: + raise ValueError("completed has too many rows") + clean_completed = [] + for row in completed: + if not isinstance(row, dict) or set(row) != {"id", "outcome"}: + raise ValueError("completed rows must contain id and outcome") + clean_completed.append( + { + "id": self._checkpoint_text( + redact_for_trajectory(row.get("id")), + field="completed.id", + max_bytes=self.CHECKPOINT_COMPLETED_ID_MAX_BYTES, + ), + "outcome": self._checkpoint_text( + redact_for_trajectory(row.get("outcome")), + field="completed.outcome", + max_bytes=self.CHECKPOINT_COMPLETED_OUTCOME_MAX_BYTES, + ), + } + ) + + record = { + "phase_id": self._checkpoint_text( + redact_for_trajectory(phase_id), + field="phase_id", + max_bytes=self.CHECKPOINT_PHASE_ID_MAX_BYTES, + ), + "phase_status": phase_status, + "completed": clean_completed, + "evidence_refs": self._validate_checkpoint_evidence_refs( + payload.get("evidence_refs", []) + ), + } + + if "next_action" in payload: + record["next_action"] = self._checkpoint_text( + redact_for_trajectory(payload.get("next_action")), + field="next_action", + max_bytes=self.CHECKPOINT_NEXT_ACTION_MAX_BYTES, + allow_empty=True, + ) + return record + + @staticmethod + def _checkpoint_text( + value, *, field: str, max_bytes: int, allow_empty: bool = False + ) -> str: + if not isinstance(value, str): + raise ValueError(f"{field} must be a string") + if "\x00" in value: + raise ValueError(f"{field} must not contain NUL") + if not allow_empty and not value: + raise ValueError(f"{field} must not be empty") + if len(value.encode("utf-8")) > max_bytes: + raise ValueError(f"{field} exceeds {max_bytes} UTF-8 bytes") + return value + + def _validate_checkpoint_evidence_refs(self, refs) -> list[dict]: + from lingtai.kernel.trace_redaction import redact_for_trajectory + + if not isinstance(refs, list): + raise ValueError("evidence_refs must be a list") + if len(refs) > self.CHECKPOINT_EVIDENCE_MAX_ROWS: + raise ValueError("evidence_refs has too many rows") + clean = [] + for row in refs: + if not isinstance(row, dict) or set(row) != {"kind", "ref"}: + raise ValueError("evidence_refs rows must contain kind and ref") + kind = row.get("kind") + if not isinstance(kind, str) or kind not in self._CHECKPOINT_EVIDENCE_KINDS: + raise ValueError("evidence_refs.kind must be event or artifact") + raw_ref = self._checkpoint_text( + row.get("ref"), + field="evidence_refs.ref", + max_bytes=self.CHECKPOINT_EVIDENCE_REF_MAX_BYTES, + ) + ref = redact_for_trajectory(raw_ref) + if ref != raw_ref: + # Redacting a path/id would destroy its referential meaning. A + # secret-bearing evidence reference is therefore rejected rather + # than persisted either raw or as a non-resolving placeholder. + raise ValueError("evidence_refs.ref must not contain secret material") + ref = self._checkpoint_text( + ref, + field="evidence_refs.ref", + max_bytes=self.CHECKPOINT_EVIDENCE_REF_MAX_BYTES, + ) + if kind == "event": + self._validate_checkpoint_event_ref(ref) + else: + self._validate_checkpoint_artifact_ref(ref) + clean.append({"kind": kind, "ref": ref}) + return clean + + @staticmethod + def _validate_checkpoint_event_ref(ref: str) -> None: + allowed = set( + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.:-" + ) + if any(ch not in allowed for ch in ref): + raise ValueError("event evidence ref must be a narrow event identifier") + + def _validate_checkpoint_artifact_ref(self, ref: str) -> None: + candidate = Path(ref) + if candidate.is_absolute(): + raise ValueError("artifact evidence ref must be run-relative") + if any(part in ("", ".", "..") for part in candidate.parts): + raise ValueError("artifact evidence ref must not traverse") + target = self.path / candidate + try: + resolved_target = target.resolve(strict=target.exists()) + resolved_root = self.path.resolve(strict=True) + except OSError as exc: + raise ValueError(f"artifact evidence ref is invalid: {exc}") from exc + if ( + resolved_target != resolved_root + and resolved_root not in resolved_target.parents + ): + raise ValueError("artifact evidence ref escapes the daemon run directory") + # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ diff --git a/tests/test_daemon_checkpoint.py b/tests/test_daemon_checkpoint.py new file mode 100644 index 000000000..7e26fcbd0 --- /dev/null +++ b/tests/test_daemon_checkpoint.py @@ -0,0 +1,512 @@ +"""Run-scoped semantic checkpoint tests for LingTai daemon runs.""" + +import json +import threading +from pathlib import Path + +import pytest + +from lingtai.kernel.llm.base import FunctionSchema, LLMResponse, ToolCall +from lingtai.kernel.llm.interface import ( + ChatInterface, + TextBlock, + ToolCallBlock, + ToolResultBlock, +) +from lingtai.tools.daemon.run_dir import DaemonRunDir +from tests._daemon_helpers import ( + make_daemon_agent, + make_daemon_run_dir, + register_daemon_entry, +) + + +class _FakeSession: + def __init__( + self, responses, *, system_prompt: str, interface: ChatInterface | None = None + ): + self.interface = interface or ChatInterface() + if interface is None: + self.interface.add_system(system_prompt) + self._responses = list(responses) + self.request_snapshots = [] + + def send(self, message): + if isinstance(message, str): + self.interface.add_user_message(message) + elif isinstance(message, list): + self.interface.add_tool_results(message) + else: + raise TypeError(type(message)) + self.request_snapshots.append(self.interface.to_dict()) + response = self._responses.pop(0) + blocks = [] + if response.text: + blocks.append(TextBlock(response.text)) + for tc in response.tool_calls or []: + blocks.append( + ToolCallBlock(id=tc.id or "", name=tc.name, args=dict(tc.args or {})) + ) + if blocks: + self.interface.add_assistant_message(blocks) + return response + + +class _FakeService: + provider = "mock" + model = "mock-model" + api_key = "fake" + _base_url = None + _provider_defaults = {} + + def __init__(self, session_responses): + self._session_responses = [list(batch) for batch in session_responses] + self.sessions = [] + self.create_session_kwargs = [] + + def create_session(self, *, system_prompt, interface=None, **kwargs): + self.create_session_kwargs.append( + {"system_prompt": system_prompt, "interface": interface, **kwargs} + ) + session = _FakeSession( + self._session_responses.pop(0), + system_prompt=system_prompt, + interface=interface, + ) + self.sessions.append(session) + return session + + def make_tool_result(self, tool_name, result, *, tool_call_id=None, provider=None): + return ToolResultBlock(id=tool_call_id or "", name=tool_name, content=result) + + +def _resp(text="", tool_calls=None): + return LLMResponse(text=text, tool_calls=list(tool_calls or []), usage=None) + + +def _run_with_service( + tmp_path, + monkeypatch, + service, + *, + schemas=None, + dispatch=None, + parent_events=None, +): + import lingtai.llm.service as service_mod + + monkeypatch.setattr(service_mod, "LLMService", lambda **_kwargs: service) + agent = make_daemon_agent(tmp_path, ["daemon"]) + mgr = agent.get_capability("daemon") + if parent_events is not None: + monkeypatch.setattr( + mgr, + "_log", + lambda event_type, **fields: parent_events.append( + {"event": event_type, **fields} + ), + ) + em_id = "em-test" + run_dir = make_daemon_run_dir(agent, em_id=em_id) + mgr._emanations[em_id] = { + "followup_buffer": "", + "followup_lock": threading.Lock(), + "run_dir": run_dir, + } + if schemas is None or dispatch is None: + schemas, dispatch = mgr._build_tool_surface([]) + result = mgr._run_emanation( + em_id, + run_dir, + schemas, + dispatch, + "task", + threading.Event(), + ) + return mgr, run_dir, result + + +def test_lingtai_run_scoped_checkpoint_persists_and_projects(tmp_path, monkeypatch): + parent_events = [] + call = ToolCall( + name="checkpoint", + id="checkpoint-1", + args={ + "phase_id": "inspect-complete", + "phase_status": "complete", + "completed": [{"id": "inspect", "outcome": "looked at daemon state"}], + "next_action": "token=ghp_abcdefghijklmnopqrstuvwxyzABCDE", + "evidence_refs": [{"kind": "event", "ref": "daemon_start"}], + }, + ) + service = _FakeService([[_resp(tool_calls=[call]), _resp("done")]]) + + mgr, run_dir, result = _run_with_service( + tmp_path, + monkeypatch, + service, + parent_events=parent_events, + ) + + assert result == "done" + first_tools = service.create_session_kwargs[0]["tools"] + assert "checkpoint" in {schema.name for schema in first_tools} + state = DaemonRunDir.read_state_from_disk(run_dir.path) + checkpoint = state["latest_checkpoint"] + assert checkpoint["schema"] == "lingtai.daemon.checkpoint.v1" + assert checkpoint["run_id"] == run_dir.run_id + assert checkpoint["checkpoint_id"] == f"{run_dir.run_id}:1" + assert checkpoint["sequence"] == 1 + assert checkpoint["phase_id"] == "inspect-complete" + check = mgr._check_snapshot_from_paths( + "em-test", + run_path=run_dir.path, + daemon_json_path=run_dir.daemon_json_path, + events_path=run_dir.events_path, + last=20, + truncate=500, + ) + assert check["checkpoint"]["phase_id"] == "inspect-complete" + events = [json.loads(line) for line in run_dir.events_path.read_text().splitlines()] + tool_call_events = [ + event + for event in events + if event.get("event") == "tool_call" and event.get("name") == "checkpoint" + ] + assert tool_call_events[-1]["args_preview"] == "{}" + checkpoint_events = [ + event for event in events if event.get("event") == "checkpoint" + ] + assert ( + checkpoint_events[-1]["checkpoint"]["checkpoint_id"] + == checkpoint["checkpoint_id"] + ) + normalized = [ + event + for event in parent_events + if event.get("event") == "daemon_tool_call_normalized" + and event.get("tool_name") == "checkpoint" + ] + assert normalized[-1]["tool_args"] == {} + assert "ghp_" not in json.dumps(parent_events) + + +def test_checkpoint_reserved_schema_replaces_external_collision(tmp_path, monkeypatch): + external_calls = [] + external_schema = FunctionSchema( + name="checkpoint", + description="external collision", + parameters={"type": "object", "properties": {"wrong": {"type": "string"}}}, + ) + call = ToolCall( + name="checkpoint", + id="checkpoint-reserved", + args={"phase_id": "internal", "phase_status": "complete"}, + ) + service = _FakeService([[_resp(tool_calls=[call]), _resp("done")]]) + + _mgr, run_dir, result = _run_with_service( + tmp_path, + monkeypatch, + service, + schemas=[external_schema], + dispatch={"checkpoint": lambda args: external_calls.append(args)}, + ) + + assert result == "done" + schemas = [ + schema + for schema in service.create_session_kwargs[0]["tools"] + if schema.name == "checkpoint" + ] + assert len(schemas) == 1 + assert schemas[0].description != "external collision" + assert external_calls == [] + state = DaemonRunDir.read_state_from_disk(run_dir.path) + assert state["latest_checkpoint"]["phase_id"] == "internal" + + +def test_ordinary_turn_does_not_create_or_advance_checkpoint(tmp_path, monkeypatch): + service = _FakeService([[_resp("plain done")]]) + + mgr, run_dir, _result = _run_with_service(tmp_path, monkeypatch, service) + + state = DaemonRunDir.read_state_from_disk(run_dir.path) + assert "latest_checkpoint" not in state + check = mgr._check_snapshot_from_paths( + "em-test", + run_path=run_dir.path, + daemon_json_path=run_dir.daemon_json_path, + events_path=run_dir.events_path, + last=20, + truncate=500, + ) + assert "checkpoint" not in check + + +def test_checkpoint_sequences_are_monotonic_unique(tmp_path): + rd = make_daemon_run_dir(parent_working_dir=tmp_path / "agent", em_id="em-seq") + + first = rd.record_checkpoint({"phase_id": "one", "phase_status": "complete"}) + second = rd.record_checkpoint({"phase_id": "two", "phase_status": "blocked"}) + + assert [first["sequence"], second["sequence"]] == [1, 2] + assert first["checkpoint_id"] != second["checkpoint_id"] + assert ( + DaemonRunDir.read_state_from_disk(rd.path)["latest_checkpoint"]["sequence"] == 2 + ) + + +def test_checkpoint_concurrent_sequences_are_unique(tmp_path): + rd = make_daemon_run_dir( + parent_working_dir=tmp_path / "agent", em_id="em-concurrent" + ) + results = [] + lock = threading.Lock() + + def worker(index): + saved = rd.record_checkpoint( + {"phase_id": f"phase-{index}", "phase_status": "complete"} + ) + with lock: + results.append(saved) + + threads = [threading.Thread(target=worker, args=(index,)) for index in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + sequences = sorted(item["sequence"] for item in results) + checkpoint_ids = {item["checkpoint_id"] for item in results} + assert sequences == list(range(1, 9)) + assert len(checkpoint_ids) == 8 + assert ( + DaemonRunDir.read_state_from_disk(rd.path)["latest_checkpoint"]["sequence"] == 8 + ) + + +@pytest.mark.parametrize( + "payload", + [ + {"phase_id": "", "phase_status": "complete"}, + {"phase_id": "x" * 97, "phase_status": "complete"}, + {"phase_id": "ok", "phase_status": "waiting"}, + {"phase_id": "ok", "phase_status": []}, + {"phase_id": "ok", "phase_status": "complete", "completed": [{}]}, + { + "phase_id": "ok", + "phase_status": "complete", + "completed": [{"id": "x", "outcome": "y"}] * 9, + }, + { + "phase_id": "ok", + "phase_status": "complete", + "completed": [{"id": "x" * 129, "outcome": "y"}], + }, + {"phase_id": "ok", "phase_status": "complete", "next_action": "x" * 513}, + {"phase_id": "ok", "phase_status": "complete", "evidence_refs": [{}]}, + { + "phase_id": "ok", + "phase_status": "complete", + "evidence_refs": [{"kind": "log", "ref": "x"}], + }, + { + "phase_id": "ok", + "phase_status": "complete", + "evidence_refs": [{"kind": [], "ref": "x"}], + }, + { + "phase_id": "ok", + "phase_status": "complete", + "evidence_refs": [{"kind": "event", "ref": "../x"}], + }, + { + "phase_id": "ok", + "phase_status": "complete", + "evidence_refs": [{"kind": "artifact", "ref": "/tmp/x"}], + }, + {"phase_id": "ok", "phase_status": "complete", "extra": "nope"}, + {"phase_id": "ok", "phase_status": "complete", "sequence": True}, + ], +) +def test_checkpoint_rejects_invalid_shapes_and_bounds(tmp_path, payload): + rd = make_daemon_run_dir(parent_working_dir=tmp_path / "agent", em_id="em-invalid") + + with pytest.raises(ValueError): + rd.record_checkpoint(payload) + assert "latest_checkpoint" not in DaemonRunDir.read_state_from_disk(rd.path) + + +def test_checkpoint_rejects_symlink_artifact_escape_and_redacts_text(tmp_path): + rd = make_daemon_run_dir(parent_working_dir=tmp_path / "agent", em_id="em-redact") + outside = tmp_path / "outside.txt" + outside.write_text("outside", encoding="utf-8") + (rd.path / "escape").symlink_to(outside) + + with pytest.raises(ValueError): + rd.record_checkpoint( + { + "phase_id": "bad-link", + "phase_status": "failed", + "evidence_refs": [{"kind": "artifact", "ref": "escape"}], + } + ) + + saved = rd.record_checkpoint( + { + "phase_id": "redact", + "phase_status": "complete", + "completed": [ + {"id": "api_key=sk-abcdefghijklmnopqrstuvwxyz", "outcome": "kept"} + ], + "next_action": "token=ghp_abcdefghijklmnopqrstuvwxyzABCDE", + } + ) + assert "sk-" not in saved["completed"][0]["id"] + assert "ghp_" not in saved["next_action"] + + +def test_checkpoint_rejects_secret_bearing_evidence_refs(tmp_path): + rd = make_daemon_run_dir( + parent_working_dir=tmp_path / "agent", em_id="em-secret-ref" + ) + secret = "ghp_abcdefghijklmnopqrstuvwxyzABCDE" + + for kind, ref in ( + ("event", f"event-{secret}"), + ("artifact", f"artifacts/token={secret}/report.txt"), + ): + with pytest.raises(ValueError, match="must not contain secret material"): + rd.record_checkpoint( + { + "phase_id": "secret-ref", + "phase_status": "complete", + "evidence_refs": [{"kind": kind, "ref": ref}], + } + ) + + assert "latest_checkpoint" not in DaemonRunDir.read_state_from_disk(rd.path) + + +def test_checkpoint_utf8_byte_limits_are_not_character_limits(tmp_path): + rd = make_daemon_run_dir(parent_working_dir=tmp_path / "agent", em_id="em-utf8") + + saved = rd.record_checkpoint( + { + "phase_id": "界" * 32, + "phase_status": "complete", + "completed": [{"id": "界" * 42, "outcome": "界" * 85}], + "next_action": "界" * 170, + } + ) + assert saved["phase_id"] == "界" * 32 + + for payload in ( + {"phase_id": "界" * 33, "phase_status": "complete"}, + { + "phase_id": "ok", + "phase_status": "complete", + "completed": [{"id": "界" * 43, "outcome": "ok"}], + }, + { + "phase_id": "ok", + "phase_status": "complete", + "completed": [{"id": "ok", "outcome": "界" * 86}], + }, + {"phase_id": "ok", "phase_status": "complete", "next_action": "界" * 171}, + ): + with pytest.raises(ValueError): + rd.record_checkpoint(payload) + + +def test_checkpoint_write_failure_leaves_previous_valid_state(tmp_path, monkeypatch): + rd = make_daemon_run_dir(parent_working_dir=tmp_path / "agent", em_id="em-atomic") + first = rd.record_checkpoint({"phase_id": "first", "phase_status": "complete"}) + original = rd._atomic_write_json + + def fail_daemon_json(path: Path, data: dict): + if path == rd.daemon_json_path: + raise OSError("simulated write failure") + return original(path, data) + + monkeypatch.setattr(rd, "_atomic_write_json", fail_daemon_json) + with pytest.raises(OSError): + rd.record_checkpoint({"phase_id": "second", "phase_status": "complete"}) + state = DaemonRunDir.read_state_from_disk(rd.path) + assert state["latest_checkpoint"] == first + + +def test_checkpoint_event_failure_does_not_fail_committed_authority( + tmp_path, monkeypatch +): + rd = make_daemon_run_dir( + parent_working_dir=tmp_path / "agent", em_id="em-event-fail" + ) + original = rd._append_jsonl + + def fail_checkpoint_event(path: Path, entry: dict): + if path == rd.events_path and entry.get("event") == "checkpoint": + raise OSError("simulated event append failure") + return original(path, entry) + + monkeypatch.setattr(rd, "_append_jsonl", fail_checkpoint_event) + saved = rd.record_checkpoint({"phase_id": "committed", "phase_status": "complete"}) + + state = DaemonRunDir.read_state_from_disk(rd.path) + assert state["latest_checkpoint"] == saved + events = [json.loads(line) for line in rd.events_path.read_text().splitlines()] + assert not any(event.get("event") == "checkpoint" for event in events) + + +def test_old_runs_without_checkpoint_omit_projection(tmp_path): + agent = make_daemon_agent(tmp_path) + mgr = agent.get_capability("daemon") + rd = make_daemon_run_dir(agent, em_id="em-old") + register_daemon_entry(mgr, "em-old", rd) + + out = mgr.handle({"action": "check", "id": "em-old"}) + + assert "checkpoint" not in out + + +def test_fresh_manager_reads_historical_checkpoint_from_disk(tmp_path): + first_agent = make_daemon_agent(tmp_path) + rd = make_daemon_run_dir(first_agent, em_id="em-historical") + rd.record_checkpoint({"phase_id": "persisted", "phase_status": "complete"}) + first_agent._workdir_lease.release() + + fresh_agent = make_daemon_agent(tmp_path) + fresh_mgr = fresh_agent.get_capability("daemon") + out = fresh_mgr.handle({"action": "check", "id": "em-historical"}) + + assert out["checkpoint"]["phase_id"] == "persisted" + assert out["checkpoint"]["sequence"] == 1 + + +def test_checkpoint_schema_survives_compact_reconstruction(tmp_path, monkeypatch): + compact = ToolCall(name="compact", args={"_reason": "handoff"}, id="compact-1") + checkpoint = ToolCall( + name="checkpoint", + args={"phase_id": "after-compact", "phase_status": "complete"}, + id="checkpoint-1", + ) + service = _FakeService( + [ + [_resp(tool_calls=[compact])], + [_resp(tool_calls=[checkpoint]), _resp("done")], + ] + ) + + _mgr, run_dir, result = _run_with_service(tmp_path, monkeypatch, service) + + assert result == "done" + assert len(service.create_session_kwargs) == 2 + assert all( + "checkpoint" in {schema.name for schema in kwargs["tools"]} + for kwargs in service.create_session_kwargs + ) + assert ( + DaemonRunDir.read_state_from_disk(run_dir.path)["latest_checkpoint"]["phase_id"] + == "after-compact" + ) From 399f041428321efeefe1853cbc617f3d40f457c0 Mon Sep 17 00:00:00 2001 From: Wang Runyuan <281843989+9s5bz2jvd2-lang@users.noreply.github.com> Date: Thu, 16 Jul 2026 04:33:08 -0700 Subject: [PATCH 2/2] test(daemon): preserve finish authority after checkpoint --- tests/test_daemon_checkpoint.py | 170 +++++++++++++++++++++++++++++++- 1 file changed, 169 insertions(+), 1 deletion(-) diff --git a/tests/test_daemon_checkpoint.py b/tests/test_daemon_checkpoint.py index 7e26fcbd0..ec60e37c6 100644 --- a/tests/test_daemon_checkpoint.py +++ b/tests/test_daemon_checkpoint.py @@ -191,6 +191,174 @@ def test_lingtai_run_scoped_checkpoint_persists_and_projects(tmp_path, monkeypat assert "ghp_" not in json.dumps(parent_events) +def test_checkpoint_complete_does_not_mark_run_done_or_bypass_finish( + tmp_path, monkeypatch +): + from lingtai.kernel._fsutil import atomic_write_json + from lingtai.mcp_servers.daemon_common.server import ( + DESCRIPTION as FINISH_DESCRIPTION, + FINISH_SCHEMA, + _validate_finish, + ) + import lingtai.llm.service as service_mod + + checkpoint_call = ToolCall( + name="checkpoint", + id="checkpoint-complete", + args={ + "phase_id": "phase-one", + "phase_status": "complete", + "next_action": "Create auto-next-action.txt with content EXECUTED", + }, + ) + finish_call = ToolCall( + name="finish", + id="finish-done", + args={"status": "done", "summary": "phase two completed explicitly"}, + ) + observations = [] + run_dir_holder = {} + + class _InspectingSession(_FakeSession): + def send(self, message): + if isinstance(message, list): + run_dir = run_dir_holder["run_dir"] + state = DaemonRunDir.read_state_from_disk(run_dir.path) + events = [ + json.loads(line) + for line in run_dir.events_path.read_text().splitlines() + ] + observations.append( + { + "state": state["state"], + "finished_at": state["finished_at"], + "phase_status": state.get("latest_checkpoint", {}).get( + "phase_status" + ), + "completion_exists": ( + run_dir.path / "daemon_completion.json" + ).exists(), + "next_action_executed": ( + run_dir.path / "auto-next-action.txt" + ).exists(), + "daemon_done_seen": any( + event.get("event") == "daemon_done" for event in events + ), + } + ) + return super().send(message) + + class _InspectingService(_FakeService): + def create_session(self, *, system_prompt, interface=None, **kwargs): + self.create_session_kwargs.append( + {"system_prompt": system_prompt, "interface": interface, **kwargs} + ) + session = _InspectingSession( + self._session_responses.pop(0), + system_prompt=system_prompt, + interface=interface, + ) + self.sessions.append(session) + return session + + service = _InspectingService( + [ + [ + _resp(tool_calls=[checkpoint_call]), + _resp(tool_calls=[finish_call]), + _resp("done"), + ] + ] + ) + monkeypatch.setattr(service_mod, "LLMService", lambda **_kwargs: service) + agent = make_daemon_agent(tmp_path, ["daemon"]) + mgr = agent.get_capability("daemon") + em_id = "em-finish-authority" + run_dir = make_daemon_run_dir( + agent, + em_id=em_id, + call_parameters={"mcp": [{"name": "daemon_common", "transport": "stdio"}]}, + ) + run_dir_holder["run_dir"] = run_dir + mgr._emanations[em_id] = { + "followup_buffer": "", + "followup_lock": threading.Lock(), + "run_dir": run_dir, + } + + finish_schema = FunctionSchema( + name="finish", + description=FINISH_DESCRIPTION, + parameters=FINISH_SCHEMA, + ) + + def finish(args): + monkeypatch.setenv("LINGTAI_DAEMON_RUN_ID", run_dir.run_id) + payload = _validate_finish(args) + atomic_write_json( + run_dir.path / "daemon_completion.json", + payload, + ensure_ascii=False, + indent=2, + ) + return { + "status": "ok", + "completion_status": payload["status"], + "message": "daemon completion recorded", + } + + result = mgr._run_emanation( + em_id, + run_dir, + [finish_schema], + {"finish": finish}, + "complete two phases", + threading.Event(), + ) + + assert result == "done" + assert observations == [ + { + "state": "running", + "finished_at": None, + "phase_status": "complete", + "completion_exists": False, + "next_action_executed": False, + "daemon_done_seen": False, + }, + { + "state": "running", + "finished_at": None, + "phase_status": "complete", + "completion_exists": True, + "next_action_executed": False, + "daemon_done_seen": False, + }, + ] + final_state = DaemonRunDir.read_state_from_disk(run_dir.path) + assert final_state["state"] == "done" + assert final_state["finished_at"] is not None + assert not (run_dir.path / "auto-next-action.txt").exists() + + events = [json.loads(line) for line in run_dir.events_path.read_text().splitlines()] + checkpoint_index = next( + index + for index, event in enumerate(events) + if event.get("event") == "checkpoint" + ) + finish_index = next( + index + for index, event in enumerate(events) + if event.get("event") == "tool_call" and event.get("name") == "finish" + ) + done_index = next( + index + for index, event in enumerate(events) + if event.get("event") == "daemon_done" + ) + assert checkpoint_index < finish_index < done_index + + def test_checkpoint_reserved_schema_replaces_external_collision(tmp_path, monkeypatch): external_calls = [] external_schema = FunctionSchema( @@ -485,7 +653,7 @@ def test_fresh_manager_reads_historical_checkpoint_from_disk(tmp_path): def test_checkpoint_schema_survives_compact_reconstruction(tmp_path, monkeypatch): - compact = ToolCall(name="compact", args={"_reason": "handoff"}, id="compact-1") + compact = ToolCall(name="compact", args={"action": "run", "_reason": "handoff"}, id="compact-1") checkpoint = ToolCall( name="checkpoint", args={"phase_id": "after-compact", "phase_status": "complete"},