feat(domain): live-state agent — modèle + port (LS1)

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 <noreply@anthropic.com>
This commit is contained in:
2026-06-21 19:03:19 +02:00
parent 827d4774cd
commit 9815af01b1
3 changed files with 370 additions and 4 deletions

View File

@ -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,
};

View File

@ -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<TicketId>,
/// 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<String>,
/// The ticket of the most recent delegation this agent issued, if any.
pub last_delegation: Option<TicketId>,
/// 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<TicketId>,
intent: impl Into<String>,
status: WorkStatus,
progress: Option<String>,
last_delegation: Option<TicketId>,
updated_at_ms: u64,
) -> Result<Self, DomainError> {
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<String, DomainError> {
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<LiveEntry>,
}
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<u64> = 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");
}
}

View File

@ -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<crate::live_state::LiveState, StoreError>;
/// 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.