From 9815af01b12eaf370f9b2d24deab0dfba18f1a86 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sun, 21 Jun 2026 19:03:19 +0200 Subject: [PATCH] =?UTF-8?q?feat(domain):=20live-state=20agent=20=E2=80=94?= =?UTF-8?q?=20mod=C3=A8le=20+=20port=20(LS1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Domaine pur du live-state des agents : `LiveState`/`LiveEntry`/`WorkStatus`. - Fusion en keyed last-writer-wins : une entrée par clé, la plus récente écrase l'ancienne (ordonnancement déterministe). - Invariants de bornes anti-dump : bornes douces (troncature) + rejet dur au-delà des limites, pour empêcher un agent de noyer le live-state. - `prune` pour borner la rétention. - Port `LiveStateStore` (lecture/écriture), sans dépendance d'infra. 9 nouveaux tests live_state (cargo test -p domain : 208 passed / 0 failed). Co-Authored-By: Claude Opus 4.8 --- crates/domain/src/lib.rs | 11 +- crates/domain/src/live_state.rs | 334 ++++++++++++++++++++++++++++++++ crates/domain/src/ports.rs | 29 +++ 3 files changed, 370 insertions(+), 4 deletions(-) create mode 100644 crates/domain/src/live_state.rs diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 740cc56..2f604e2 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -40,6 +40,7 @@ pub mod git; pub mod ids; pub mod input; pub mod layout; +pub mod live_state; pub mod mailbox; pub mod markdown; pub mod memory; @@ -95,6 +96,8 @@ pub use conversation::{ pub use input::{AgentBusyState, AgentLiveness, InputMediator, InputSource}; +pub use live_state::{LiveEntry, LiveState, WorkStatus, FIELD_MAX_BYTES, FIELD_PREVIEW_MAX_CHARS}; + pub use readiness::{ReadinessPolicy, ReadinessSignal}; pub use session_limit::{plan_resume, RateLimitSource, ResumePlan, SessionLimit}; @@ -152,8 +155,8 @@ pub use ports::{ EmbedderEnvInspector, EmbedderEnvReport, EmbedderError, EmbedderProfileStore, EmbedderPromptDismissal, EmbedderPromptStore, EventBus, EventStream, ExitStatus, FileSystem, FsError, GitCommitInfo, GitError, GitFileStatus, GitPort, GraphCommit, IdGenerator, - MemoryError, MemoryQuery, MemoryRecall, MemoryStore, Output, OutputStream, PermissionStore, - PreparedContext, ProcessError, ProcessSpawner, ProfileStore, ProjectStore, PtyError, PtyHandle, - PtyPort, RemoteError, RemoteHost, RemotePath, RuntimeError, ScheduledTask, Scheduler, - SpawnSpec, StoreError, TemplateStore, + LiveStateStore, MemoryError, MemoryQuery, MemoryRecall, MemoryStore, Output, OutputStream, + PermissionStore, PreparedContext, ProcessError, ProcessSpawner, ProfileStore, ProjectStore, + PtyError, PtyHandle, PtyPort, RemoteError, RemoteHost, RemotePath, RuntimeError, ScheduledTask, + Scheduler, SpawnSpec, StoreError, TemplateStore, }; diff --git a/crates/domain/src/live_state.rs b/crates/domain/src/live_state.rs new file mode 100644 index 0000000..40d84ed --- /dev/null +++ b/crates/domain/src/live_state.rs @@ -0,0 +1,334 @@ +//! 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; + +/// 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, +} + +/// 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); + } + } + + /// 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 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 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"); + } +} diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 5cd37ae..087b828 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -816,6 +816,35 @@ pub trait SkillStore: Send + Sync { ) -> Result<(), StoreError>; } +/// Persistence for the agent **live-state** snapshot (programme live-state). +/// +/// Mirrors the other stores ([`SkillStore`], [`MemoryStore`]): a driven port the +/// infrastructure implements. The contract is **keyed last-writer-wins**, never +/// append — [`upsert`](LiveStateStore::upsert) replaces the row for an agent and +/// [`prune`](LiveStateStore::prune) applies the TTL + cardinality bound, matching +/// [`LiveState`](crate::live_state::LiveState)'s pure semantics. +#[async_trait] +pub trait LiveStateStore: Send + Sync { + /// Loads the current live-state snapshot. + /// + /// # Errors + /// [`StoreError`] on failure. + async fn load(&self) -> Result; + + /// Inserts or replaces (keyed by `agent_id`) a single live-state row. + /// + /// # Errors + /// [`StoreError`] on failure. + async fn upsert(&self, entry: crate::live_state::LiveEntry) -> Result<(), StoreError>; + + /// Drops rows older than `ttl_ms` relative to `now_ms`, then bounds the set + /// to `max_n`, keeping the most recently updated rows. + /// + /// # Errors + /// [`StoreError`] on failure. + async fn prune(&self, now_ms: u64, ttl_ms: u64, max_n: usize) -> Result<(), StoreError>; +} + /// CRUD for project [`Memory`] notes — the `.md` knowledge base under /// `.ideai/memory/` (LOT A, étage 1). Notes are the single source of truth; the /// aggregated `MEMORY.md` index is derived and kept in sync on every write.