//! Agent **live-state** — the volatile, "what is each agent doing right now" //! projection (programme live-state, lot LS1). //! //! Unlike the conversation log (an append-only journal) this is a **keyed, //! last-writer-wins** snapshot: there is exactly **one** [`LiveEntry`] per //! [`AgentId`], replaced in place on every update. There is deliberately **no //! append API** — stacking duplicate rows per agent is an architectural //! anti-pattern here (it would turn a live snapshot back into a journal). The //! only mutations are [`LiveState::upsert`] (keyed replace-or-insert) and //! [`LiveState::prune`] (TTL + cardinality bound). //! //! The type is **pure** (zero I/O). Persistence is the job of the //! [`crate::ports::LiveStateStore`] port, implemented by infrastructure in a //! later lot. use serde::{Deserialize, Serialize}; use crate::error::DomainError; use crate::ids::AgentId; use crate::mailbox::TicketId; /// Soft bound (in characters) applied to free-text fields (`intent`, /// `progress`): values longer than this are **truncated**, not rejected. Aligned /// with the workstate UI task preview (`TASK_PREVIEW_MAX_CHARS`, ≈160 chars) so /// the live state and its rendering agree on excerpt length. pub const FIELD_PREVIEW_MAX_CHARS: usize = 160; /// Hard anti-dump threshold (in **bytes**) for any single free-text field. A /// value above this is **rejected** with [`DomainError::Invariant`] rather than /// silently truncated: the soft bound trims ordinary over-long lines, while this /// guards against a caller dumping long content (file bodies, logs) into the /// live state. Distinct semantics: soft = truncate, hard = reject. pub const FIELD_MAX_BYTES: usize = 2 * 1024; /// `progress` marker stamped on an orphan row reconciled at restart (see /// [`LiveState::reconcile_orphans`]): it records *why* a `Working`/`Waiting`/ /// `Blocked` row was downgraded to `Idle` — its session was not running when the /// project reopened. Well under both field bounds. pub const STALE_AT_RESTART_MARKER: &str = "(stale — session not running at restart)"; /// What an agent is doing right now, at a glance. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum WorkStatus { /// No active task — available. Idle, /// Actively working on its current `intent`. Working, /// Stuck / cannot proceed (e.g. needs a decision or input). Blocked, /// Suspended awaiting an external event (e.g. a delegation reply). Waiting, /// Current task finished. Done, } impl WorkStatus { /// Parses a status label (case-insensitive) into a [`WorkStatus`]. /// /// Accepts exactly the five canonical labels (`idle`, `working`, `blocked`, /// `waiting`, `done`), ignoring surrounding whitespace and ASCII case. Returns /// `None` for anything else, so the caller can raise a typed error rather than /// guess a default. #[must_use] pub fn parse(raw: &str) -> Option { match raw.trim().to_ascii_lowercase().as_str() { "idle" => Some(Self::Idle), "working" => Some(Self::Working), "blocked" => Some(Self::Blocked), "waiting" => Some(Self::Waiting), "done" => Some(Self::Done), _ => None, } } /// The canonical, human-readable label of this status (the inverse of /// [`WorkStatus::parse`]): `idle`, `working`, `blocked`, `waiting` or `done`. #[must_use] pub fn label(self) -> &'static str { match self { Self::Idle => "idle", Self::Working => "working", Self::Blocked => "blocked", Self::Waiting => "waiting", Self::Done => "done", } } } /// A single agent's live-state row. One per [`AgentId`] in a [`LiveState`]. /// /// Construct via [`LiveEntry::new`], which enforces the field invariants /// (soft-truncation of `intent`/`progress`, hard rejection of oversize input). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LiveEntry { /// The agent this row describes (the keyed identity for last-writer-wins). pub agent_id: AgentId, /// The ticket the agent is currently handling, if any. pub ticket: Option, /// Short human-readable description of the current intent (soft-bounded). pub intent: String, /// Coarse status at a glance. pub status: WorkStatus, /// Optional finer-grained progress note (soft-bounded). pub progress: Option, /// The ticket of the most recent delegation this agent issued, if any. pub last_delegation: Option, /// Wall-clock update time in epoch milliseconds (drives `prune` ordering). pub updated_at_ms: u64, } impl LiveEntry { /// Builds a validated live-state entry. /// /// `intent` and `progress` are bounded: each is **rejected** if it exceeds /// the hard [`FIELD_MAX_BYTES`] anti-dump threshold, otherwise **truncated** /// to [`FIELD_PREVIEW_MAX_CHARS`] characters (never mid-codepoint). /// /// # Errors /// [`DomainError::Invariant`] if `intent` or `progress` exceeds /// [`FIELD_MAX_BYTES`] bytes. pub fn new( agent_id: AgentId, ticket: Option, intent: impl Into, status: WorkStatus, progress: Option, last_delegation: Option, updated_at_ms: u64, ) -> Result { let intent = bound_text("intent", intent.into())?; let progress = progress.map(|p| bound_text("progress", p)).transpose()?; Ok(Self { agent_id, ticket, intent, status, progress, last_delegation, updated_at_ms, }) } } /// Enforces the field bounds: reject above the hard byte threshold, otherwise /// truncate to the soft character bound (char-boundary safe). fn bound_text(field: &'static str, value: String) -> Result { if value.len() > FIELD_MAX_BYTES { return Err(DomainError::Invariant(format!( "live-state field `{field}` exceeds the max size of {FIELD_MAX_BYTES} bytes ({} bytes)", value.len() ))); } if value.chars().count() <= FIELD_PREVIEW_MAX_CHARS { Ok(value) } else { Ok(value.chars().take(FIELD_PREVIEW_MAX_CHARS).collect()) } } /// The live-state snapshot: one [`LiveEntry`] per agent, keyed last-writer-wins. /// /// There is **no append API by design** (anti-journal invariant): the only /// mutations are [`LiveState::upsert`] and [`LiveState::prune`]. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LiveState { /// The current rows, at most one per [`AgentId`]. pub entries: Vec, } impl LiveState { /// Inserts or **replaces** the row for `entry.agent_id` (last-writer-wins). /// /// Never produces a duplicate row for an agent: if an entry with the same /// `agent_id` already exists it is overwritten in place; otherwise the entry /// is appended as a new key. (This is *keyed* upsert, not an append of /// duplicates — see the type-level anti-journal invariant.) pub fn upsert(&mut self, entry: LiveEntry) { if let Some(slot) = self .entries .iter_mut() .find(|e| e.agent_id == entry.agent_id) { *slot = entry; } else { self.entries.push(entry); } } /// Computes the **reconciliation** of orphan rows at restart: rows whose /// status implies a live session (`Working`/`Waiting`/`Blocked`) but whose /// agent is **not** currently live according to `is_live`. /// /// At reboot a crash never runs `close_project`, so `live-state.json` can keep /// "ghost" rows for agents whose sessions are dead. This pure transform returns /// the **rewritten** rows to upsert (it does **not** mutate `self`): each orphan /// is downgraded to [`WorkStatus::Idle`], keeping its `intent` (trace), clearing /// the now-dead `ticket` and `last_delegation`, stamping a `progress` stale /// marker and a fresh `updated_at_ms = now_ms` (last-writer-wins). /// /// Rows that are already `Idle`/`Done`, or whose agent is still live, are /// **never** touched and are absent from the result. The caller persists each /// returned entry through the keyed [`upsert`](LiveState::upsert) (best-effort). #[must_use] pub fn reconcile_orphans( &self, is_live: impl Fn(&AgentId) -> bool, now_ms: u64, ) -> Vec { self.entries .iter() .filter(|e| { matches!( e.status, WorkStatus::Working | WorkStatus::Waiting | WorkStatus::Blocked ) && !is_live(&e.agent_id) }) .map(|e| LiveEntry { agent_id: e.agent_id, ticket: None, intent: e.intent.clone(), status: WorkStatus::Idle, progress: Some(STALE_AT_RESTART_MARKER.to_owned()), last_delegation: None, updated_at_ms: now_ms, }) .collect() } /// Drops entries older than `ttl_ms` relative to `now_ms`, then bounds the /// result to `max_n`, keeping the most recently updated rows. /// /// Age is `now_ms.saturating_sub(updated_at_ms)`; a row is kept while its /// age is `<= ttl_ms`. After the TTL sweep, if more than `max_n` rows remain /// they are ordered by `updated_at_ms` (most recent first, stable on ties) /// and truncated to `max_n`. pub fn prune(&mut self, now_ms: u64, ttl_ms: u64, max_n: usize) { self.entries .retain(|e| now_ms.saturating_sub(e.updated_at_ms) <= ttl_ms); if self.entries.len() > max_n { self.entries .sort_by(|a, b| b.updated_at_ms.cmp(&a.updated_at_ms)); self.entries.truncate(max_n); } } } #[cfg(test)] mod tests { use super::*; fn aid(n: u128) -> AgentId { AgentId::from_uuid(uuid::Uuid::from_u128(n)) } fn tid(n: u128) -> TicketId { TicketId::from_uuid(uuid::Uuid::from_u128(n)) } #[test] fn work_status_parse_is_case_insensitive_and_rejects_unknown() { assert_eq!(WorkStatus::parse("idle"), Some(WorkStatus::Idle)); assert_eq!(WorkStatus::parse(" Working "), Some(WorkStatus::Working)); assert_eq!(WorkStatus::parse("BLOCKED"), Some(WorkStatus::Blocked)); assert_eq!(WorkStatus::parse("waiting"), Some(WorkStatus::Waiting)); assert_eq!(WorkStatus::parse("Done"), Some(WorkStatus::Done)); assert_eq!(WorkStatus::parse("busy"), None); assert_eq!(WorkStatus::parse(""), None); } #[test] fn work_status_label_round_trips_through_parse() { for status in [ WorkStatus::Idle, WorkStatus::Working, WorkStatus::Blocked, WorkStatus::Waiting, WorkStatus::Done, ] { assert_eq!(WorkStatus::parse(status.label()), Some(status)); } } #[test] fn new_truncates_intent_and_progress_at_soft_bound() { let long = "x".repeat(FIELD_PREVIEW_MAX_CHARS + 50); let entry = LiveEntry::new( aid(1), None, long.clone(), WorkStatus::Working, Some(long.clone()), None, 10, ) .expect("under the hard threshold ⇒ accepted, just truncated"); assert_eq!(entry.intent.chars().count(), FIELD_PREVIEW_MAX_CHARS); assert_eq!( entry.progress.as_deref().map(|p| p.chars().count()), Some(FIELD_PREVIEW_MAX_CHARS) ); } #[test] fn new_keeps_short_text_verbatim() { let entry = LiveEntry::new(aid(1), None, "ship it", WorkStatus::Done, None, None, 5).unwrap(); assert_eq!(entry.intent, "ship it"); assert!(entry.progress.is_none()); } #[test] fn new_rejects_intent_over_hard_byte_threshold() { let huge = "x".repeat(FIELD_MAX_BYTES + 1); let err = LiveEntry::new(aid(1), None, huge, WorkStatus::Working, None, None, 0) .expect_err("oversize intent must be rejected, not truncated"); assert!(matches!(err, DomainError::Invariant(_))); } #[test] fn new_rejects_progress_over_hard_byte_threshold() { let huge = "y".repeat(FIELD_MAX_BYTES + 1); let err = LiveEntry::new(aid(1), None, "ok", WorkStatus::Working, Some(huge), None, 0) .expect_err("oversize progress must be rejected, not truncated"); assert!(matches!(err, DomainError::Invariant(_))); } #[test] fn upsert_is_last_writer_wins_per_agent_no_duplicates() { let mut state = LiveState::default(); state.upsert( LiveEntry::new(aid(1), None, "first", WorkStatus::Working, None, None, 1).unwrap(), ); state.upsert( LiveEntry::new( aid(1), Some(tid(9)), "second", WorkStatus::Blocked, None, None, 2, ) .unwrap(), ); // A second upsert for the same agent replaces, never appends. assert_eq!(state.entries.len(), 1, "same agent ⇒ exactly one row"); let row = &state.entries[0]; assert_eq!(row.intent, "second"); assert_eq!(row.status, WorkStatus::Blocked); assert_eq!(row.ticket, Some(tid(9))); } #[test] fn upsert_distinct_agents_coexist() { let mut state = LiveState::default(); state .upsert(LiveEntry::new(aid(1), None, "a", WorkStatus::Working, None, None, 1).unwrap()); state.upsert(LiveEntry::new(aid(2), None, "b", WorkStatus::Idle, None, None, 1).unwrap()); assert_eq!(state.entries.len(), 2, "distinct agents ⇒ distinct rows"); } #[test] fn prune_drops_entries_older_than_ttl() { let mut state = LiveState::default(); state.upsert( LiveEntry::new(aid(1), None, "stale", WorkStatus::Idle, None, None, 100).unwrap(), ); state.upsert( LiveEntry::new(aid(2), None, "fresh", WorkStatus::Working, None, None, 900).unwrap(), ); // now=1000, ttl=200 ⇒ age(aid1)=900 > 200 dropped; age(aid2)=100 kept. state.prune(1000, 200, 100); assert_eq!(state.entries.len(), 1); assert_eq!(state.entries[0].agent_id, aid(2)); } #[test] fn prune_bounds_to_max_n_keeping_most_recent() { let mut state = LiveState::default(); for n in 1..=5u128 { state.upsert( LiveEntry::new( aid(n), None, "x", WorkStatus::Working, None, None, n as u64 * 10, ) .unwrap(), ); } // All within TTL; bound to 2 ⇒ keep the two highest updated_at_ms (40,50). state.prune(60, 1_000, 2); assert_eq!(state.entries.len(), 2); let kept: Vec = state.entries.iter().map(|e| e.updated_at_ms).collect(); assert_eq!(kept, vec![50, 40], "most recent first, oldest dropped"); } #[test] fn reconcile_orphans_downgrades_dead_active_rows_and_rewrites_fields() { let mut state = LiveState::default(); // An orphan: Working but its agent is not live. state.upsert( LiveEntry::new( aid(1), Some(tid(7)), "implementing lot 2", WorkStatus::Working, Some("half done".to_owned()), Some(tid(8)), 100, ) .unwrap(), ); // Nothing is live ⇒ aid(1) is an orphan. let rewritten = state.reconcile_orphans(|_| false, 999); assert_eq!(rewritten.len(), 1, "the single active-but-dead row is an orphan"); let row = &rewritten[0]; assert_eq!(row.agent_id, aid(1)); assert_eq!(row.status, WorkStatus::Idle, "downgraded to idle"); assert_eq!(row.intent, "implementing lot 2", "intent kept as trace"); assert_eq!(row.ticket, None, "dead ticket cleared"); assert_eq!(row.last_delegation, None, "dead delegation cleared"); assert_eq!(row.progress.as_deref(), Some(STALE_AT_RESTART_MARKER)); assert_eq!(row.updated_at_ms, 999, "stamped with the fresh now"); } #[test] fn reconcile_orphans_covers_working_waiting_blocked_only() { let mut state = LiveState::default(); state.upsert( LiveEntry::new(aid(1), None, "w", WorkStatus::Working, None, None, 1).unwrap(), ); state.upsert( LiveEntry::new(aid(2), None, "a", WorkStatus::Waiting, None, None, 1).unwrap(), ); state.upsert( LiveEntry::new(aid(3), None, "b", WorkStatus::Blocked, None, None, 1).unwrap(), ); // idle/done are never orphans, even when the agent is dead. state.upsert(LiveEntry::new(aid(4), None, "i", WorkStatus::Idle, None, None, 1).unwrap()); state.upsert(LiveEntry::new(aid(5), None, "d", WorkStatus::Done, None, None, 1).unwrap()); let rewritten = state.reconcile_orphans(|_| false, 50); let mut ids: Vec = rewritten.iter().map(|e| e.agent_id).collect(); ids.sort(); assert_eq!( ids, vec![aid(1), aid(2), aid(3)], "only working/waiting/blocked rows are reconciled; idle/done untouched" ); } #[test] fn reconcile_orphans_spares_live_agents() { let mut state = LiveState::default(); state.upsert( LiveEntry::new(aid(1), None, "alive", WorkStatus::Working, None, None, 1).unwrap(), ); state.upsert( LiveEntry::new(aid(2), None, "dead", WorkStatus::Working, None, None, 1).unwrap(), ); // aid(1) is still live ⇒ spared; only aid(2) is reconciled. let rewritten = state.reconcile_orphans(|a| *a == aid(1), 50); assert_eq!(rewritten.len(), 1); assert_eq!(rewritten[0].agent_id, aid(2), "only the dead agent is reconciled"); } #[test] fn reconcile_orphans_does_not_mutate_self() { let mut state = LiveState::default(); state.upsert( LiveEntry::new(aid(1), Some(tid(1)), "x", WorkStatus::Working, None, None, 1).unwrap(), ); let before = state.clone(); let _ = state.reconcile_orphans(|_| false, 999); assert_eq!(state, before, "reconcile_orphans is a pure read (returns rows to upsert)"); } #[test] fn serde_round_trip_is_camel_case() { let mut state = LiveState::default(); state.upsert( LiveEntry::new( aid(7), Some(tid(3)), "review PR", WorkStatus::Waiting, Some("waiting on QA".to_owned()), Some(tid(4)), 1234, ) .unwrap(), ); let json = serde_json::to_string(&state).unwrap(); // Field names are camelCase… assert!(json.contains("\"agentId\"")); assert!(json.contains("\"lastDelegation\"")); assert!(json.contains("\"updatedAtMs\"")); // …and the enum variant is camelCase too. assert!(json.contains("\"waiting\"")); let back: LiveState = serde_json::from_str(&json).unwrap(); assert_eq!(back, state, "round-trip is lossless"); } }