From 6a35a53253542f4dd7748bd9607f115380c08ba3 Mon Sep 17 00:00:00 2001 From: RNT56 Date: Mon, 29 Jun 2026 11:24:57 +0200 Subject: [PATCH 1/2] feat(daemon): cross-process task lease recovery (roadmap-v0.6 F.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds TaskRegistry::snapshot_all + adopt_from_snapshot (TaskSnapshot / AdoptError) — the recovery half of invariant 12. A restarting daemon re-adopts its running tasks with stable ids, re-leasing under the new LeaseOwner and marking each Pending so a fresh worker resumes from the log (an mpsc channel can't survive a restart). Carried usage is preserved so budgets re-charge, not reset (invariant 11); an already-over-budget task is adopted terminal (Done(BudgetExhausted)); an absent worktree -> WorktreeGone; a duplicate id -> Duplicate. A re-adopted task still completes only on a VerifiedPatch (invariant 13 — adoption restores supervision, never completion). Pure state-machine steps; the dump/load + SIGTERM + restart cycle are the #[ignore]d daemon_recover_xproc_live_smoke (TODO(daemon-recover-xproc-live)), in runbook F.6. Also derives PartialEq/Eq on AgentBudget. Completes Phase F. 4 new tests; daemon-only; zero nano impact. `cargo xtask verify` green. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 16 ++ crates/crustcore-daemon/src/registry.rs | 207 +++++++++++++++++++++- crates/crustcore-daemon/src/supervisor.rs | 2 +- docs/live-socket-validation.md | 16 ++ 4 files changed, 235 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a7fdf1..76984e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,21 @@ agent/PR/role/size/invariant audit trail. ### Added +- **Cross-process task lease recovery (roadmap-v0.6 F.1).** Added + `TaskRegistry::snapshot_all` + `adopt_from_snapshot` (with `TaskSnapshot` / `AdoptError`) + — the recovery half of invariant 12. A restarting daemon re-adopts its running tasks + with **stable ids**, re-leasing under the new `LeaseOwner` and marking each **`Pending`** + so a fresh worker resumes from the log (an `mpsc` channel can't survive a restart). + Carried `usage` is preserved so budgets **re-charge, not reset** (invariant 11); an + already-over-budget task is adopted **terminal** (`Done(BudgetExhausted)`); an absent + worktree → `WorktreeGone`; a duplicate id → `Duplicate`. A re-adopted task still + completes only on a `VerifiedPatch` (invariant 13 — adoption restores supervision, never + completion). Pure state-machine steps; the dump/load file I/O + SIGTERM hook + + kill-and-restart cycle are the `#[ignore]`d `daemon_recover_xproc_live_smoke` + (`TODO(daemon-recover-xproc-live)`), in runbook §F.6. Also derives `PartialEq`/`Eq` on + `AgentBudget`. **This completes Phase F (daemon hardening).** 4 new tests; daemon-only; + **zero nano impact**. + - **Evidence bundle rendering (roadmap-v0.6 C.3).** Added `EvidenceBundle::to_markdown()` and `to_json()` to `crustcore_daemon::product`. `to_markdown` is the **bounded** canonical PR-body/cockpit renderer: it opens with @@ -249,6 +264,7 @@ agent/PR/role/size/invariant audit trail. | Date | Phase/Task | Change | PR / Branch | Agent / Role | Nano Δ | Invariants | | --- | --- | --- | --- | --- | --- | --- | +| 2026-06-28 | v0.6/F.1 | `snapshot_all`/`adopt_from_snapshot` cross-process recovery: stable ids, re-leased under new owner, Pending-resume-from-log, carried-usage re-charge, over-budget→terminal. Completes Phase F | `claude/v06-f1-recovery` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 11, 12, 13; recovery restores supervision, never completion | | 2026-06-28 | v0.6/C.3 | `EvidenceBundle::to_markdown` (bounded PR-body/cockpit render, 🔴 review notice, per-list overflow) + `to_json` (schema v1); `draft_pr_body` delegates | `claude/v06-c3-evidence` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 2, 10, 11; bounded redacted evidence, every receipt included | | 2026-06-28 | v0.6/D.1 | Task-loop wiring `plan_task`/`finalize_task` composing routing (C.1) + advisory gate (C.2) into a terminal `TaskOutcome`; sandboxed run `#[ignore]`d | `claude/v06-d1-executor-wire` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 4, 5, 6, 13; verifier-owned completion, advisory only gates | | 2026-06-28 | v0.6/A.3 | `pr_intent_to_create_request`: PrIntent→CreatePrRequest for the live draft-PR POST; evidence body verbatim, draft=true; real POST `#[ignore]`d | `claude/v06-a3-draftpr` | Claude (Implementer) | 0 kB (daemon/live-only) | Enforces 6, 13, 14; body is evidence not a model claim | diff --git a/crates/crustcore-daemon/src/registry.rs b/crates/crustcore-daemon/src/registry.rs index a6804d2..831e446 100644 --- a/crates/crustcore-daemon/src/registry.rs +++ b/crates/crustcore-daemon/src/registry.rs @@ -36,11 +36,14 @@ //! talk to the user: the registry emits [`RegistryAction::SendLine`] as **data**, and only //! the runtime loop renders it through the redactor and sends it (invariants 2, 5). //! -//! `TODO(daemon-recover-xproc)`: cross-process / persistent lease recovery (surviving a -//! daemon restart) is out of scope here — the registry is in-process; the [`LeaseOwner`] -//! field is carried precisely so a future snapshot/steal protocol slots in without changing -//! this state machine. `TODO(daemon-admin)`: a remote admin socket; control stays via the -//! existing chat verbs. +//! **Cross-process recovery (roadmap-v0.6 F.1):** [`TaskRegistry::snapshot_all`] + +//! [`TaskRegistry::adopt_from_snapshot`] now realize the recovery half of invariant 12 — +//! a restarting daemon re-adopts its running tasks (stable ids), re-leasing under the new +//! [`LeaseOwner`] and marking each **`Pending`** so a fresh worker resumes from the log +//! (an `mpsc` channel cannot survive a restart). These are **pure** state-machine steps; +//! the actual dump/load file I/O + the SIGTERM hook + a kill-and-restart cycle are the +//! `TODO(daemon-recover-xproc-live)` seam. A remote operator [`admin`](crate::admin) socket +//! (`TODO(daemon-admin)`) lands the control plane beyond the chat verbs. //! //! The module is **pure** (no `TaskHandle`/socket/thread), so it is always compiled and //! CI-tested in every build; only the live loop that drives it is `live`-gated. @@ -527,6 +530,114 @@ impl TaskRegistry { } found } + + // --- cross-process recovery (roadmap-v0.6 F.1) --------------------------- + + /// Snapshots every **non-terminal** task for a cross-process dump (F.1). Pure — no + /// I/O; the caller persists the returned `Vec` (the live dump/load is the seam). A + /// terminal task has nothing to recover and is skipped. The live channel is **not** + /// captured: an `mpsc` pair cannot survive a restart, so a re-adopted task resumes + /// from its log, not by reconnecting a channel. + #[must_use] + pub fn snapshot_all(&self, now: Timestamp) -> Vec { + self.slots + .values() + .filter(|s| !s.phase.is_terminal()) + .map(|s| TaskSnapshot { + id: s.id, + chat: s.chat, + phase: s.phase, + lease_remaining_ms: s + .lease + .expires_at + .as_millis() + .saturating_sub(now.as_millis()), + budget: s.budget, + usage: s.usage, + worktree_path: None, + }) + .collect() + } + + /// Re-adopts a task from a [`TaskSnapshot`] after a daemon restart (F.1; closes the + /// recovery half of invariant 12). Re-leases under **this** instance's `LeaseOwner` + /// and marks the task **`Pending`** so the loop spawns a *fresh* worker that resumes + /// from the log. Carried `usage` is preserved so budgets are re-charged, not reset + /// (invariant 11); a re-adopted task still completes only on a `VerifiedPatch` + /// (invariant 13 — adoption restores supervision, never completion). + /// + /// - **Absent worktree** (`worktree_present == false`) → [`AdoptError::WorktreeGone`]. + /// - **Already over budget** (carried usage breaches the budget) → adopted **terminal** + /// `Done(BudgetExhausted)` so `tick` finalizes it (never resumed). + /// - **Duplicate id** already live → [`AdoptError::Duplicate`]. + /// + /// # Errors + /// [`AdoptError`] for a gone worktree or a duplicate id. + pub fn adopt_from_snapshot( + &mut self, + snap: &TaskSnapshot, + worktree_present: bool, + now: Timestamp, + ) -> Result { + if self.slots.contains_key(&snap.id) { + return Err(AdoptError::Duplicate); + } + if !worktree_present { + return Err(AdoptError::WorktreeGone); + } + // A task whose carried usage already breaches its budget is adopted terminal. + let phase = match snap.usage.charge(AgentUsage::default(), &snap.budget) { + Err(axis) => TaskPhase::Done(TaskDone::BudgetExhausted(axis)), + Ok(_) => TaskPhase::Pending, + }; + self.slots.insert( + snap.id, + TaskSlot { + id: snap.id, + chat: snap.chat, + phase, + lease: Lease::granted(self.owner, now), // re-leased under THIS instance + budget: snap.budget, + usage: snap.usage, // carried — budgets re-charge, not reset + started_at: now, + last_charge_at: now, + }, + ); + if self.next_id <= snap.id.0 { + self.next_id = snap.id.0.wrapping_add(1); // never re-issue an adopted id + } + Ok(snap.id) + } +} + +/// A serializable snapshot of one task for cross-process recovery (F.1). Carries what is +/// needed to re-adopt the task after a restart — **not** the live channel (an `mpsc` +/// pair cannot survive a restart; a re-adopted task resumes from its log). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TaskSnapshot { + /// The task id (stable across the restart). + pub id: TaskId, + /// The chat that launched it. + pub chat: ChatId, + /// The phase at dump time. + pub phase: TaskPhase, + /// Lease time remaining at dump (informational; adoption re-leases fresh). + pub lease_remaining_ms: u64, + /// The per-task budget. + pub budget: AgentBudget, + /// Carried usage (so budgets re-charge, not reset). + pub usage: AgentUsage, + /// The worktree path (set by the dumper; checked for existence on adopt). + pub worktree_path: Option, +} + +/// Why a task could not be re-adopted from a snapshot. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdoptError { + /// The task's worktree no longer exists — the work is unrecoverable. + WorktreeGone, + /// A task with that id is already live in this instance. + Duplicate, } #[cfg(test)] @@ -541,6 +652,92 @@ mod tests { Timestamp::from_millis(ms) } + // --- cross-process recovery (F.1) --- + + #[test] + fn snapshot_and_adopt_round_trip_preserves_id_and_usage() { + let mut old = reg(); + let id = old + .admit(ChatId(7), default_task_budget(), at(1000)) + .unwrap(); + let snaps = old.snapshot_all(at(1000)); + assert_eq!(snaps.len(), 1); + assert_eq!(snaps[0].id, id); + + // A fresh instance (new owner) adopts the snapshot. + let mut fresh = TaskRegistry::new(2, LeaseOwner(99)); + let adopted = fresh + .adopt_from_snapshot(&snaps[0], true, at(5000)) + .unwrap(); + assert_eq!(adopted, id, "id is stable across the restart"); + let snap2 = fresh.snapshot(at(5000)); + let row = snap2.get(id).unwrap(); + // Re-adopted as Pending (a fresh worker resumes from the log), re-leased fresh. + assert_eq!(row.phase, TaskPhase::Pending); + assert_eq!(row.lease_ttl_ms, LEASE_TTL_MS); + // A re-issued id can never collide with the adopted one. + let next = fresh + .admit(ChatId(7), default_task_budget(), at(5000)) + .unwrap(); + assert_ne!(next, id); + } + + #[test] + fn an_absent_worktree_cannot_be_adopted() { + let mut old = reg(); + old.admit(ChatId(7), default_task_budget(), at(0)).unwrap(); + let snap = old.snapshot_all(at(0)).remove(0); + let mut fresh = TaskRegistry::new(2, LeaseOwner(2)); + assert_eq!( + fresh.adopt_from_snapshot(&snap, false, at(1)), + Err(AdoptError::WorktreeGone) + ); + } + + #[test] + fn an_over_budget_task_is_adopted_terminal() { + let mut old = reg(); + let id = old.admit(ChatId(7), default_task_budget(), at(0)).unwrap(); + let mut snap = old.snapshot_all(at(0)).remove(0); + // Carry usage past the wall budget. + snap.usage.wall_ms = snap.budget.max_wall_ms + 1; + + let mut fresh = TaskRegistry::new(2, LeaseOwner(2)); + fresh.adopt_from_snapshot(&snap, true, at(1)).unwrap(); + let row = fresh.snapshot(at(1)).get(id).unwrap().phase; + assert!( + matches!(row, TaskPhase::Done(TaskDone::BudgetExhausted(_))), + "an over-budget task must adopt terminal, got {row:?}" + ); + } + + #[test] + fn a_duplicate_id_is_rejected() { + let mut r = reg(); + let id = r.admit(ChatId(7), default_task_budget(), at(0)).unwrap(); + let snap = TaskSnapshot { + id, + chat: ChatId(7), + phase: TaskPhase::Pending, + lease_remaining_ms: LEASE_TTL_MS, + budget: default_task_budget(), + usage: AgentUsage::default(), + worktree_path: None, + }; + assert_eq!( + r.adopt_from_snapshot(&snap, true, at(1)), + Err(AdoptError::Duplicate) + ); + } + + // Live seam: the real dump/load file I/O, the SIGTERM hook, and a kill-and-restart smoke. + #[test] + #[ignore = "live: dump on SIGTERM, reload + re-adopt after a process restart (TODO(daemon-recover-xproc-live))"] + fn daemon_recover_xproc_live_smoke() { + // See docs/live-socket-validation.md §F.6. Needs file I/O + a process restart. + panic!("live seam: run manually with a kill-and-restart cycle (see runbook §F.6)"); + } + #[test] fn admit_reserves_slot_and_grants_lease() { let mut r = reg(); diff --git a/crates/crustcore-daemon/src/supervisor.rs b/crates/crustcore-daemon/src/supervisor.rs index 611b6c8..3a41bc7 100644 --- a/crates/crustcore-daemon/src/supervisor.rs +++ b/crates/crustcore-daemon/src/supervisor.rs @@ -147,7 +147,7 @@ impl AgentRegistry { /// A subagent's resource budget (invariant 11; `docs/maintainer-agent.md` §4.3). /// Runaway fan-out / unbounded work is a threat, so every axis is capped. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct AgentBudget { /// Wall-clock cap in milliseconds. pub max_wall_ms: u64, diff --git a/docs/live-socket-validation.md b/docs/live-socket-validation.md index a92468b..5db6286 100644 --- a/docs/live-socket-validation.md +++ b/docs/live-socket-validation.md @@ -81,6 +81,7 @@ cargo test --workspace -- --list --ignored | `live_get_updates_smoke` | F | `live` | [F.1](#f1) | `RestTelegram` shaping + redaction ✓ | easy (bot token) | | `live_telegram_round_trip_smoke` | F | `live` | [F.2](#f2) | runtime-channel decision logic ✓ | easy (bot token) | | `live_ws_sse_emits_a_snapshot` | F | — | [F.3](#f3) | snapshot serialize + `ws_stream` ✓ | easy (loopback port) | +| `daemon_recover_xproc_live_smoke` | F | — | [F.6](#f6) | `snapshot_all`/`adopt_from_snapshot` cores ✓ | medium (restart) | --- @@ -418,6 +419,21 @@ cargo test --workspace -- --list --ignored > their live inches are covered by [F.2](#f2)/[A.2](#a2) (model + channel) and > [B.4](#b4) (the draft-PR POST) respectively. + +### F.6 — `daemon_recover_xproc_live_smoke` — cross-process recovery (F.1) +- **Test:** `crustcore-daemon/src/registry.rs::tests::daemon_recover_xproc_live_smoke`. Seam tag `TODO(daemon-recover-xproc-live)`. +- **Socket:** the real dump/load file I/O, the SIGTERM hook, and a kill-and-restart cycle. +- **CI core (passing):** `snapshot_all` (non-terminal tasks only) + `adopt_from_snapshot` + — re-leases under the new instance's owner, marks **Pending** (a fresh worker resumes + from the log; the `mpsc` channel cannot survive a restart), carries usage so budgets + **re-charge not reset** (invariant 11), adopts an over-budget task **terminal**, rejects + an absent worktree + a duplicate id. A re-adopted task still completes only on a + `VerifiedPatch` (invariant 13); recovery restores supervision (invariant 12). +- **Prereq:** a cache dir for the dump + a kill-and-restart of the daemon. +- **Run:** `cargo test -p crustcore-daemon registry::tests::daemon_recover_xproc_live_smoke -- --ignored --nocapture` +- **Success:** after a restart the daemon reloads the dump and re-adopts the running + tasks (stable ids), resuming supervision. **Difficulty: medium.** + --- ## Seam tags without a dedicated smoke test From dabb1835557e48d0ee4b3ff1ea4d3edf9ad1e877 Mon Sep 17 00:00:00 2001 From: RNT56 Date: Mon, 29 Jun 2026 12:12:07 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat(daemon):=20real=20F.1=20persistence=20?= =?UTF-8?q?=E2=80=94=20CCTS=20snapshot=20dump/load=20(end-to-end)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns F.1's recovery from snapshot/adopt-in-memory into real persistence: - encode_snapshots/decode_snapshots: a dependency-free, versioned, bounded CCTS frame (magic + version + count + per-snapshot id/chat/phase(tagged)/ lease/budget/usage/worktree-path), panic-free + fail-closed (bad magic -> BadFormat, bad version -> BadVersion, truncated/over-cap -> BadContents). - TaskRegistry::dump_snapshots(path) + load_snapshots(path): actual file I/O (std::fs only), for the SIGTERM-dump / startup-reload cycle. Now the only #[ignore]d inch is the SIGTERM hook + a real OS process restart (daemon_recover_xproc_live_smoke). 3 new tests: encode/decode round-trip (incl. worktree path + BudgetExhausted axis), bad-magic/version rejection, and a dump->load->re-adopt simulated restart. Daemon-only; zero nano impact. `cargo xtask verify` green. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 7 +- crates/crustcore-daemon/src/registry.rs | 292 +++++++++++++++++++++++- 2 files changed, 294 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 76984e6..87bbe0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,8 +39,11 @@ agent/PR/role/size/invariant audit trail. already-over-budget task is adopted **terminal** (`Done(BudgetExhausted)`); an absent worktree → `WorktreeGone`; a duplicate id → `Duplicate`. A re-adopted task still completes only on a `VerifiedPatch` (invariant 13 — adoption restores supervision, never - completion). Pure state-machine steps; the dump/load file I/O + SIGTERM hook + - kill-and-restart cycle are the `#[ignore]`d `daemon_recover_xproc_live_smoke` + completion). **Persistence is real:** `encode_snapshots`/`decode_snapshots` (a dependency-free, + versioned, bounded `CCTS` frame — panic-free, fail-closed), `TaskRegistry::dump_snapshots`, + and `load_snapshots` actually write + reload the dump (CI-tested incl. a dump→load→re-adopt + simulated restart). Only the SIGTERM hook + a real OS process restart remain the `#[ignore]`d + `daemon_recover_xproc_live_smoke` (`TODO(daemon-recover-xproc-live)`), in runbook §F.6. Also derives `PartialEq`/`Eq` on `AgentBudget`. **This completes Phase F (daemon hardening).** 4 new tests; daemon-only; **zero nano impact**. diff --git a/crates/crustcore-daemon/src/registry.rs b/crates/crustcore-daemon/src/registry.rs index 831e446..8ac3c4c 100644 --- a/crates/crustcore-daemon/src/registry.rs +++ b/crates/crustcore-daemon/src/registry.rs @@ -640,6 +640,236 @@ pub enum AdoptError { Duplicate, } +// --- F.1 persistence: a dependency-free, versioned, bounded snapshot dump --- + +/// Magic for a task-snapshot dump (`CrustCore Task Snapshots`). Mirrors the event-log / +/// memory-store frame formats — no serde, no crypto. +pub const SNAPSHOT_MAGIC: [u8; 4] = *b"CCTS"; +/// Dump format version (an old reader rejects a newer file rather than misreading it). +pub const SNAPSHOT_VERSION: u8 = 1; +/// Cap on snapshots restored from a dump (bounded — a corrupt/hostile file cannot blow up +/// memory; invariant 11). +pub const MAX_DUMP_SNAPSHOTS: usize = 4096; +/// Cap on a restored worktree-path field. +pub const MAX_WORKTREE_PATH: usize = 4096; + +/// Why a snapshot dump could not be written or read. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SnapshotIoError { + /// An I/O error (bounded message). + Io(String), + /// Not a snapshot dump (bad magic / truncated header). + BadFormat, + /// An unsupported dump version. + BadVersion(u8), + /// Malformed or over-bound contents (corrupt / hostile file). + BadContents, +} + +impl core::fmt::Display for SnapshotIoError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SnapshotIoError::Io(e) => write!(f, "snapshot io: {e}"), + SnapshotIoError::BadFormat => write!(f, "not a task-snapshot dump"), + SnapshotIoError::BadVersion(v) => write!(f, "unsupported snapshot dump version {v}"), + SnapshotIoError::BadContents => write!(f, "malformed task-snapshot dump"), + } + } +} + +impl std::error::Error for SnapshotIoError {} + +fn encode_phase(buf: &mut Vec, phase: TaskPhase) { + match phase { + TaskPhase::Pending => buf.push(0), + TaskPhase::Running => buf.push(1), + TaskPhase::Cancelling => buf.push(2), + TaskPhase::Done(done) => { + buf.push(3); + match done { + TaskDone::Completed => buf.push(0), + TaskDone::Cancelled => buf.push(1), + TaskDone::Killed => buf.push(2), + TaskDone::Expired => buf.push(3), + TaskDone::BudgetExhausted(axis) => { + buf.push(4); + buf.push(match axis { + BudgetError::Wall => 0, + BudgetError::Output => 1, + BudgetError::Tokens => 2, + }); + } + } + } + } +} + +/// Encodes snapshots into the bounded `CCTS` frame. Pure (no I/O) — testable in memory. +#[must_use] +pub fn encode_snapshots(snaps: &[TaskSnapshot]) -> Vec { + let mut buf = Vec::new(); + buf.extend_from_slice(&SNAPSHOT_MAGIC); + buf.push(SNAPSHOT_VERSION); + let count = u32::try_from(snaps.len()).unwrap_or(u32::MAX); + buf.extend_from_slice(&count.to_le_bytes()); + for s in snaps { + buf.extend_from_slice(&s.id.0.to_le_bytes()); + buf.extend_from_slice(&s.chat.0.to_le_bytes()); + encode_phase(&mut buf, s.phase); + buf.extend_from_slice(&s.lease_remaining_ms.to_le_bytes()); + buf.extend_from_slice(&s.budget.max_wall_ms.to_le_bytes()); + buf.extend_from_slice(&s.budget.max_output_bytes.to_le_bytes()); + buf.extend_from_slice(&s.budget.max_tokens.to_le_bytes()); + buf.extend_from_slice(&s.usage.wall_ms.to_le_bytes()); + buf.extend_from_slice(&s.usage.output_bytes.to_le_bytes()); + buf.extend_from_slice(&s.usage.tokens.to_le_bytes()); + let path = s.worktree_path.as_deref().unwrap_or(""); + let plen = path.len().min(MAX_WORKTREE_PATH); + buf.extend_from_slice(&(plen as u32).to_le_bytes()); + buf.extend_from_slice(&path.as_bytes()[..plen]); + } + buf +} + +/// A bounded byte cursor for decoding (no panics — every read is checked). +struct SnapCursor<'a> { + bytes: &'a [u8], + pos: usize, +} + +impl<'a> SnapCursor<'a> { + fn take(&mut self, n: usize) -> Option<&'a [u8]> { + let end = self.pos.checked_add(n)?; + if end > self.bytes.len() { + return None; + } + let s = &self.bytes[self.pos..end]; + self.pos = end; + Some(s) + } + fn u8(&mut self) -> Option { + self.take(1).map(|s| s[0]) + } + fn u32(&mut self) -> Option { + self.take(4) + .map(|s| u32::from_le_bytes(s.try_into().unwrap())) + } + fn u64(&mut self) -> Option { + self.take(8) + .map(|s| u64::from_le_bytes(s.try_into().unwrap())) + } + fn i64(&mut self) -> Option { + self.take(8) + .map(|s| i64::from_le_bytes(s.try_into().unwrap())) + } +} + +fn decode_phase(c: &mut SnapCursor) -> Option { + Some(match c.u8()? { + 0 => TaskPhase::Pending, + 1 => TaskPhase::Running, + 2 => TaskPhase::Cancelling, + 3 => TaskPhase::Done(match c.u8()? { + 0 => TaskDone::Completed, + 1 => TaskDone::Cancelled, + 2 => TaskDone::Killed, + 3 => TaskDone::Expired, + 4 => TaskDone::BudgetExhausted(match c.u8()? { + 0 => BudgetError::Wall, + 1 => BudgetError::Output, + _ => BudgetError::Tokens, + }), + _ => return None, + }), + _ => return None, + }) +} + +/// Decodes a `CCTS` dump. **Fails closed** on bad magic/version and is **panic-free + +/// bounded**: a corrupt/hostile file yields a [`SnapshotIoError`], never a panic or an +/// unbounded allocation. +/// +/// # Errors +/// [`SnapshotIoError`] on a bad header or malformed/over-cap contents. +pub fn decode_snapshots(bytes: &[u8]) -> Result, SnapshotIoError> { + let mut c = SnapCursor { bytes, pos: 0 }; + if c.take(4) != Some(&SNAPSHOT_MAGIC) { + return Err(SnapshotIoError::BadFormat); + } + let version = c.u8().ok_or(SnapshotIoError::BadFormat)?; + if version != SNAPSHOT_VERSION { + return Err(SnapshotIoError::BadVersion(version)); + } + let count = c.u32().ok_or(SnapshotIoError::BadContents)? as usize; + if count > MAX_DUMP_SNAPSHOTS { + return Err(SnapshotIoError::BadContents); + } + let mut out = Vec::with_capacity(count.min(256)); + for _ in 0..count { + let id = TaskId(c.u64().ok_or(SnapshotIoError::BadContents)?); + let chat = ChatId(c.i64().ok_or(SnapshotIoError::BadContents)?); + let phase = decode_phase(&mut c).ok_or(SnapshotIoError::BadContents)?; + let lease_remaining_ms = c.u64().ok_or(SnapshotIoError::BadContents)?; + let budget = AgentBudget { + max_wall_ms: c.u64().ok_or(SnapshotIoError::BadContents)?, + max_output_bytes: c.u64().ok_or(SnapshotIoError::BadContents)?, + max_tokens: c.u64().ok_or(SnapshotIoError::BadContents)?, + }; + let usage = AgentUsage { + wall_ms: c.u64().ok_or(SnapshotIoError::BadContents)?, + output_bytes: c.u64().ok_or(SnapshotIoError::BadContents)?, + tokens: c.u64().ok_or(SnapshotIoError::BadContents)?, + }; + let plen = c.u32().ok_or(SnapshotIoError::BadContents)? as usize; + if plen > MAX_WORKTREE_PATH { + return Err(SnapshotIoError::BadContents); + } + let path_bytes = c.take(plen).ok_or(SnapshotIoError::BadContents)?; + let worktree_path = if plen == 0 { + None + } else { + Some(String::from_utf8_lossy(path_bytes).into_owned()) + }; + out.push(TaskSnapshot { + id, + chat, + phase, + lease_remaining_ms, + budget, + usage, + worktree_path, + }); + } + Ok(out) +} + +/// Loads a snapshot dump from `path` (F.1). The daemon calls this on startup to recover +/// running tasks via [`TaskRegistry::adopt_from_snapshot`]. +/// +/// # Errors +/// [`SnapshotIoError`] on an I/O failure or a malformed dump. +pub fn load_snapshots(path: &std::path::Path) -> Result, SnapshotIoError> { + let bytes = std::fs::read(path).map_err(|e| SnapshotIoError::Io(e.to_string()))?; + decode_snapshots(&bytes) +} + +impl TaskRegistry { + /// Dumps the live (non-terminal) tasks to `path` as a `CCTS` frame (F.1). Call this on + /// a SIGTERM hook so a restart can [`load_snapshots`] + re-adopt. Pure framing + one + /// `write`. + /// + /// # Errors + /// [`SnapshotIoError::Io`] if the file cannot be written. + pub fn dump_snapshots( + &self, + path: &std::path::Path, + now: Timestamp, + ) -> Result<(), SnapshotIoError> { + let bytes = encode_snapshots(&self.snapshot_all(now)); + std::fs::write(path, &bytes).map_err(|e| SnapshotIoError::Io(e.to_string())) + } +} + #[cfg(test)] mod tests { use super::*; @@ -730,11 +960,67 @@ mod tests { ); } - // Live seam: the real dump/load file I/O, the SIGTERM hook, and a kill-and-restart smoke. #[test] - #[ignore = "live: dump on SIGTERM, reload + re-adopt after a process restart (TODO(daemon-recover-xproc-live))"] + fn snapshot_dump_encodes_and_decodes_round_trip() { + let mut old = reg(); + old.admit(ChatId(7), default_task_budget(), at(1000)) + .unwrap(); + old.admit(ChatId(-3), default_task_budget(), at(1000)) + .unwrap(); + let mut snaps = old.snapshot_all(at(1000)); + // Exercise a worktree path + a terminal-with-axis phase through the codec. + snaps[0].worktree_path = Some("/tmp/crustcore/wt-1".to_string()); + snaps[1].phase = TaskPhase::Done(TaskDone::BudgetExhausted(BudgetError::Output)); + + let bytes = encode_snapshots(&snaps); + let decoded = decode_snapshots(&bytes).unwrap(); + assert_eq!(decoded, snaps, "the CCTS frame must round-trip exactly"); + } + + #[test] + fn a_bad_magic_or_version_is_rejected_not_panicked() { + assert_eq!(decode_snapshots(b"XXXX"), Err(SnapshotIoError::BadFormat)); + let mut bytes = encode_snapshots(&[]); + bytes[4] = 99; // bump the version byte + assert_eq!( + decode_snapshots(&bytes), + Err(SnapshotIoError::BadVersion(99)) + ); + // A truncated frame fails closed, never panics. + assert!(decode_snapshots(&[b'C', b'C', b'T', b'S', 1, 9]).is_err()); + } + + #[test] + fn dump_then_load_then_readopt_survives_a_simulated_restart() { + let mut old = reg(); + let id = old + .admit(ChatId(7), default_task_budget(), at(1000)) + .unwrap(); + let mut path = std::env::temp_dir(); + path.push("cc_registry_dump_test.ccts"); + old.dump_snapshots(&path, at(1000)).unwrap(); + + // A fresh instance loads the dump and re-adopts (the worktree still exists). + let loaded = load_snapshots(&path).unwrap(); + assert_eq!(loaded.len(), 1); + let mut fresh = TaskRegistry::new(2, LeaseOwner(42)); + let adopted = fresh + .adopt_from_snapshot(&loaded[0], true, at(5000)) + .unwrap(); + assert_eq!(adopted, id); + assert_eq!( + fresh.snapshot(at(5000)).get(id).unwrap().phase, + TaskPhase::Pending + ); + let _ = std::fs::remove_file(&path); + } + + // Live seam: the SIGTERM hook + an actual kill-and-restart of the OS process (the dump + // encode/decode + file I/O above are CI-tested; this is the process-lifecycle inch). + #[test] + #[ignore = "live: dump on SIGTERM, reload + re-adopt after a real process restart (TODO(daemon-recover-xproc-live))"] fn daemon_recover_xproc_live_smoke() { - // See docs/live-socket-validation.md §F.6. Needs file I/O + a process restart. + // See docs/live-socket-validation.md §F.6. Needs a kill-and-restart of the daemon. panic!("live seam: run manually with a kill-and-restart cycle (see runbook §F.6)"); }