543 lines
20 KiB
Rust
543 lines
20 KiB
Rust
//! 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<Self> {
|
|
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<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);
|
|
}
|
|
}
|
|
|
|
/// 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<LiveEntry> {
|
|
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<u64> = 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<AgentId> = 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");
|
|
}
|
|
}
|