diff --git a/CHANGELOG.md b/CHANGELOG.md index 750d64d..b4f2095 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,24 @@ 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). **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**. + - **Multi-repo orchestration skeleton (roadmap-v0.6 F.3).** Added `crustcore_daemon::multirepo`: `RepoId`, `RepoBinding` (id/path/verify/base/keywords, from config/CLI), and a pure `classify_repo(intent, repos) → Option` that routes @@ -342,6 +360,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/F.3 | `multirepo::classify_repo` (explicit-hint → sole-repo default → ambiguous asks) + `RepoBinding`; intent matches operator keywords only, never supplies a path | `claude/v06-f3-multirepo` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 7, 11; repo paths from config/CLI, shared global cap | | 2026-06-28 | v0.6/F.2 | Admin socket protocol: parse/frame(bounded)/nonce-auth + `dispatch_admin` (status/detail/cancel/kill) feeding the same owner-scoped path as Telegram; live listener `#[ignore]`d | `claude/v06-f2-adminsock` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 5, 11, 12; operator-only, owner-scoped cancel/kill | | 2026-06-28 | v0.6/E.4 | `TokenRedactor` streaming-redaction prototype (buffer-to-boundary + dangling-prefix retention) + `docs/cot-streaming.md` feasibility (feasible, behind `reveal_reasoning`) | `claude/v06-e4-cotstream` | Claude (Implementer) | 0 kB (secrets/docs) | Enforces 2, 3, 11; no unredacted secret reaches the user mid-stream | diff --git a/crates/crustcore-daemon/src/registry.rs b/crates/crustcore-daemon/src/registry.rs index a6804d2..8ac3c4c 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,344 @@ 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, +} + +// --- 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)] @@ -541,6 +882,148 @@ 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) + ); + } + + #[test] + 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 a kill-and-restart of the daemon. + 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 279ed97..c4a2d07 100644 --- a/docs/live-socket-validation.md +++ b/docs/live-socket-validation.md @@ -83,6 +83,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) | | `multi_repo_live_smoke` | F | — | [F.5](#f5) | `classify_repo` routing core ✓ | medium (multiple repos) | | `daemon_admin_live_socket_smoke` | F | — | [F.4](#f4) | admin parse/frame/auth/dispatch cores ✓ | medium (bound socket) | @@ -451,6 +452,20 @@ 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.** ### F.5 — `multi_repo_live_smoke` — multi-repo orchestration (F.3) - **Test:** `crustcore-daemon/src/multirepo.rs::tests::multi_repo_live_smoke`. Seam tag `TODO(P10-multi-repo-live)`.