From cc7d99a63ebfe27116b0cb32bda9f888b647ce23 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sat, 20 Jun 2026 18:52:16 +0200 Subject: [PATCH] =?UTF-8?q?feat(workstate):=20snapshot=20des=20d=C3=A9l?= =?UTF-8?q?=C3=A9gations=20en=20file=20par=20agent=20(Lot=20B=20backend)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ajoute un port de lecture ségrégué `AgentQueueSnapshot` (ISP) distinct du `AgentMailbox` mutant : il expose `queue_for(agent)` qui renvoie des `QueuedTicketSnapshot` clonés, ordonnés FIFO, avec position recalculée (0 = tête). Observer la file ne la mute jamais ; le one-shot reply sender reste dans l'adaptateur. - domain : value object `QueuedTicketSnapshot` + trait `AgentQueueSnapshot` (object-safe, partagé en `Arc`). - infrastructure : `InMemoryMailbox` implémente la vue lecture en plus de la vue mutation ; positions recalculées à chaque appel. - application : le read-model work-state liste les délégations en attente via ce port, troncature de l'aperçu de tâche en conservant la longueur d'origine. - app-tauri : DTO `camelCase` des tickets en file câblé dans l'état. Tests verts : domain mailbox (6), infrastructure mailbox (13), application workstate (12), app-tauri dto_agents (20). Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/dto.rs | 93 +++++++++- crates/app-tauri/src/state.rs | 5 + crates/app-tauri/tests/dto_agents.rs | 72 +++++++- crates/application/src/lib.rs | 4 +- crates/application/src/workstate/mod.rs | 119 +++++++++++- crates/application/tests/workstate.rs | 222 ++++++++++++++++++++++- crates/domain/src/lib.rs | 5 +- crates/domain/src/mailbox.rs | 59 ++++++ crates/infrastructure/src/mailbox/mod.rs | 128 ++++++++++++- 9 files changed, 692 insertions(+), 15 deletions(-) diff --git a/crates/app-tauri/src/dto.rs b/crates/app-tauri/src/dto.rs index db026a2..1bf8c52 100644 --- a/crates/app-tauri/src/dto.rs +++ b/crates/app-tauri/src/dto.rs @@ -9,8 +9,9 @@ use serde::{Deserialize, Serialize}; use application::{ - AppError, CreateProjectInput, CreateProjectOutput, GitGraphOutput, HealthInput, HealthReport, - LayoutKind, ListProjectsOutput, LiveSessionKind, OpenProjectOutput, ProjectWorkState, + AgentTicketState, AppError, CreateProjectInput, CreateProjectOutput, GitGraphOutput, + HealthInput, HealthReport, LayoutKind, ListProjectsOutput, LiveSessionKind, OpenProjectOutput, + ProjectWorkState, TicketWorkSource, TicketWorkStatus, }; use domain::{AgentBusyState, Project, ProjectId}; @@ -1503,6 +1504,87 @@ pub struct LiveWorkSessionDto { pub kind: LiveWorkSessionKindDto, } +/// Derived processing status of a queued ticket. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum TicketWorkStatusDto { + /// The agent's current busy turn is running this ticket. + InProgress, + /// Waiting behind the head / the agent is idle. + Queued, +} + +impl From for TicketWorkStatusDto { + fn from(status: TicketWorkStatus) -> Self { + match status { + TicketWorkStatus::InProgress => Self::InProgress, + TicketWorkStatus::Queued => Self::Queued, + } + } +} + +/// Origin of a queued ticket (human operator or a delegating agent). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum TicketWorkSourceDto { + /// The human operator. + Human, + /// Another agent delegating via `idea_ask_agent`. + #[serde(rename_all = "camelCase")] + Agent { + /// The delegating agent id. + agent_id: String, + }, +} + +impl From for TicketWorkSourceDto { + fn from(source: TicketWorkSource) -> Self { + match source { + TicketWorkSource::Human => Self::Human, + TicketWorkSource::Agent { agent_id } => Self::Agent { + agent_id: agent_id.to_string(), + }, + } + } +} + +/// One queued/in-progress delegation ticket for an agent. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentTicketStateDto { + /// Stable id of the queued ticket. + pub ticket_id: String, + /// Conversation thread this task enters. + pub conversation_id: String, + /// FIFO position at snapshot time (`0` = head). + pub position: u32, + /// Derived status (in-progress vs queued). + pub status: TicketWorkStatusDto, + /// Origin of the ticket. + pub source: TicketWorkSourceDto, + /// Display label of the requester. + pub requester_label: String, + /// Bounded excerpt of the task. + pub task_preview: String, + /// Character length of the original (un-truncated) task. + pub task_len: usize, +} + +impl From for AgentTicketStateDto { + fn from(ticket: AgentTicketState) -> Self { + Self { + ticket_id: ticket.ticket_id.to_string(), + conversation_id: ticket.conversation_id.to_string(), + position: ticket.position, + status: ticket.status.into(), + source: ticket.source.into(), + requester_label: ticket.requester_label, + task_preview: ticket.task_preview, + task_len: ticket.task_len, + } + } +} + /// One manifest agent's current live/busy state. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -1517,6 +1599,8 @@ pub struct AgentWorkStateDto { pub live: Option, /// Current mediated-input busy state. pub busy: AgentBusyState, + /// Pending/in-progress delegation tickets, in FIFO order. + pub tickets: Vec, } /// Project-level read model for conversation/delegation UX. @@ -1543,6 +1627,11 @@ impl From for ProjectWorkStateDto { kind: live.kind.into(), }), busy: agent.busy, + tickets: agent + .tickets + .into_iter() + .map(AgentTicketStateDto::from) + .collect(), }) .collect(), } diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 59625d9..1c89bbe 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -993,6 +993,10 @@ impl AppState { // partagé pour `resolve`/`resolve_ticket`/`cancel_head` côté orchestrateur. let inmemory_mailbox = Arc::new(InMemoryMailbox::new()); let mailbox = Arc::clone(&inmemory_mailbox) as Arc; + // Same concrete mailbox, second (read-only) port view for the work-state + // read model: lists pending tickets without touching the mutating surface. + let queue_snapshot = + Arc::clone(&inmemory_mailbox) as Arc; let mediated_inbox = Arc::new( MediatedInbox::with_pty( Arc::clone(&inmemory_mailbox), @@ -1031,6 +1035,7 @@ impl AppState { Arc::clone(&contexts_port), Arc::clone(&live_sessions), Arc::clone(&input_mediator), + queue_snapshot, )); // --- Limites de session des agents (ARCHITECTURE §21, LS7) --- diff --git a/crates/app-tauri/tests/dto_agents.rs b/crates/app-tauri/tests/dto_agents.rs index 71d9d39..80b6c12 100644 --- a/crates/app-tauri/tests/dto_agents.rs +++ b/crates/app-tauri/tests/dto_agents.rs @@ -9,8 +9,9 @@ use app_tauri_lib::dto::{ }; use application::AppError; use application::{ - AgentWorkState, CreateAgentOutput, InspectConversationOutput, LaunchAgentOutput, - ListAgentsOutput, LiveSessionKind, LiveWorkSession, ProjectWorkState, + AgentTicketState, AgentWorkState, CreateAgentOutput, InspectConversationOutput, + LaunchAgentOutput, ListAgentsOutput, LiveSessionKind, LiveWorkSession, ProjectWorkState, + TicketWorkSource, TicketWorkStatus, }; use domain::ids::{AgentId, NodeId, ProfileId, SessionId}; use domain::ports::ConversationDetails; @@ -219,6 +220,7 @@ fn project_work_state_dto_serialises_live_and_busy_camelcase() { ticket, since_ms: 1_234, }, + tickets: vec![], }], }); @@ -231,11 +233,77 @@ fn project_work_state_dto_serialises_live_and_busy_camelcase() { assert_eq!(v["agents"][0]["busy"]["state"], "busy"); assert_eq!(v["agents"][0]["busy"]["ticket"], ticket.to_string()); assert_eq!(v["agents"][0]["busy"]["sinceMs"], 1_234); + assert_eq!(v["agents"][0]["tickets"], json!([])); assert!(v["agents"][0].get("agent_id").is_none()); assert!(v["agents"][0]["live"].get("session_id").is_none()); assert!(v["agents"][0]["busy"].get("since_ms").is_none()); } +#[test] +fn project_work_state_dto_serialises_tickets_camelcase() { + let agent = AgentId::from_uuid(Uuid::from_u128(11)); + let profile = ProfileId::from_uuid(Uuid::from_u128(12)); + let requester = AgentId::from_uuid(Uuid::from_u128(20)); + let head_ticket = domain::TicketId::from_uuid(Uuid::from_u128(30)); + let next_ticket = domain::TicketId::from_uuid(Uuid::from_u128(31)); + let conversation = domain::ConversationId::from_uuid(Uuid::from_u128(40)); + let dto = app_tauri_lib::dto::ProjectWorkStateDto::from(ProjectWorkState { + agents: vec![AgentWorkState { + agent_id: agent, + name: "Worker".to_owned(), + profile_id: profile, + live: None, + busy: domain::AgentBusyState::Idle, + tickets: vec![ + AgentTicketState { + ticket_id: head_ticket, + conversation_id: conversation, + position: 0, + status: TicketWorkStatus::InProgress, + source: TicketWorkSource::Agent { + agent_id: requester, + }, + requester_label: "Main".to_owned(), + task_preview: "Analyser le module".to_owned(), + task_len: 842, + }, + AgentTicketState { + ticket_id: next_ticket, + conversation_id: conversation, + position: 1, + status: TicketWorkStatus::Queued, + source: TicketWorkSource::Human, + requester_label: "User".to_owned(), + task_preview: "Vérifier".to_owned(), + task_len: 8, + }, + ], + }], + }); + + let v = serde_json::to_value(&dto).unwrap(); + let head = &v["agents"][0]["tickets"][0]; + assert_eq!(head["ticketId"], head_ticket.to_string()); + assert_eq!(head["conversationId"], conversation.to_string()); + assert_eq!(head["position"], 0); + assert_eq!(head["status"], "inProgress"); + assert_eq!(head["source"]["kind"], "agent"); + assert_eq!(head["source"]["agentId"], requester.to_string()); + assert_eq!(head["requesterLabel"], "Main"); + assert_eq!(head["taskPreview"], "Analyser le module"); + assert_eq!(head["taskLen"], 842); + // snake_case must not leak. + assert!(head.get("ticket_id").is_none()); + assert!(head.get("task_preview").is_none()); + assert!(head["source"].get("agent_id").is_none()); + + let next = &v["agents"][0]["tickets"][1]; + assert_eq!(next["status"], "queued"); + assert_eq!(next["source"]["kind"], "human"); + assert!(next["source"].get("agentId").is_none()); + assert_eq!(next["requesterLabel"], "User"); +} + #[test] fn launch_agent_request_carries_conversation_id_for_resume() { let raw = json!({ diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 9a50497..50e0cb6 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -119,6 +119,6 @@ pub use terminal::{ }; pub use window::{MoveTabToNewWindow, MoveTabToNewWindowInput, MoveTabToNewWindowOutput}; pub use workstate::{ - AgentWorkState, GetProjectWorkState, GetProjectWorkStateInput, LiveWorkSession, - ProjectWorkState, + AgentTicketState, AgentWorkState, GetProjectWorkState, GetProjectWorkStateInput, + LiveWorkSession, ProjectWorkState, TicketWorkSource, TicketWorkStatus, }; diff --git a/crates/application/src/workstate/mod.rs b/crates/application/src/workstate/mod.rs index 897c0a8..b1eebb9 100644 --- a/crates/application/src/workstate/mod.rs +++ b/crates/application/src/workstate/mod.rs @@ -8,13 +8,22 @@ use std::collections::HashMap; use std::sync::Arc; -use domain::input::AgentBusyState; +use domain::input::{AgentBusyState, InputSource}; use domain::ports::AgentContextStore; -use domain::{AgentId, InputMediator, NodeId, ProfileId, Project, SessionId}; +use domain::{ + AgentId, AgentQueueSnapshot, ConversationId, InputMediator, NodeId, ProfileId, Project, + QueuedTicketSnapshot, SessionId, TicketId, +}; use crate::error::AppError; use crate::terminal::{LiveSessionKind, LiveSessionSnapshot, LiveSessions}; +/// Maximum length (in characters) of a derived [`AgentTicketState::task_preview`]. +/// +/// The full task can be a very long prompt; the panel only needs a scannable +/// excerpt, with [`AgentTicketState::task_len`] signalling there is more. +const TASK_PREVIEW_MAX_CHARS: usize = 160; + /// Input for [`GetProjectWorkState::execute`]. #[derive(Debug, Clone, PartialEq, Eq)] pub struct GetProjectWorkStateInput { @@ -42,6 +51,64 @@ pub struct AgentWorkState { pub live: Option, /// Current FIFO/busy state. pub busy: AgentBusyState, + /// Pending/in-progress delegation tickets, in FIFO order. + pub tickets: Vec, +} + +/// One queued delegation ticket, projected for the work-state read model. +/// +/// A read-only, presentation-oriented view of a [`QueuedTicketSnapshot`]: the +/// status is **derived** by crossing the FIFO snapshot with the agent's busy state +/// (never stored), and `task_preview` is a bounded excerpt of the full task (the +/// full text is never sent to the panel; `task_len` signals there is more). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AgentTicketState { + /// Stable id of the queued ticket. + pub ticket_id: TicketId, + /// Conversation thread this task enters. + pub conversation_id: ConversationId, + /// FIFO position at snapshot time (`0` = head). + pub position: u32, + /// Derived status (in-progress = the agent's current busy ticket). + pub status: TicketWorkStatus, + /// Origin of the ticket (human operator or a delegating agent). + pub source: TicketWorkSource, + /// Display label of the requester. + pub requester_label: String, + /// Bounded, whitespace-normalised excerpt of the task. + pub task_preview: String, + /// Character length of the **original** (un-truncated) task. + pub task_len: usize, +} + +/// Derived processing status of a queued ticket. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TicketWorkStatus { + /// The agent's current busy turn is running this ticket. + InProgress, + /// Waiting behind the head / the agent is idle. + Queued, +} + +/// Origin of a queued ticket, mirroring [`InputSource`] for the read model. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TicketWorkSource { + /// The human operator. + Human, + /// Another agent delegating via `idea_ask_agent`. + Agent { + /// The delegating agent. + agent_id: AgentId, + }, +} + +impl From for TicketWorkSource { + fn from(source: InputSource) -> Self { + match source { + InputSource::Human => Self::Human, + InputSource::Agent { agent_id } => Self::Agent { agent_id }, + } + } } /// Live session coordinates exposed by the work-state read model. @@ -60,6 +127,7 @@ pub struct GetProjectWorkState { contexts: Arc, live: Arc, input: Arc, + queue: Arc, } impl GetProjectWorkState { @@ -69,11 +137,13 @@ impl GetProjectWorkState { contexts: Arc, live: Arc, input: Arc, + queue: Arc, ) -> Self { Self { contexts, live, input, + queue, } } @@ -102,12 +172,23 @@ impl GetProjectWorkState { session_id: snapshot.session_id, kind: snapshot.kind, }); + // Tickets are crossed with the busy state: only an agent absent from + // the manifest is dropped (this loop only visits manifest entries), so + // the manifest boundary is naturally preserved. + let busy_ticket = self.input.busy_state(agent.id).ticket(); + let tickets = self + .queue + .queue_for(agent.id) + .into_iter() + .map(|snapshot| ticket_state(snapshot, busy_ticket)) + .collect(); Ok(AgentWorkState { agent_id: agent.id, name: agent.name, profile_id: agent.profile_id, live, busy: self.input.busy_state(agent.id), + tickets, }) }) .collect::, AppError>>()?; @@ -115,6 +196,40 @@ impl GetProjectWorkState { } } +/// Projects one FIFO snapshot into a read-model ticket, deriving status from the +/// agent's current busy ticket (the one running its turn = `InProgress`). +fn ticket_state(snapshot: QueuedTicketSnapshot, busy_ticket: Option) -> AgentTicketState { + let status = if busy_ticket == Some(snapshot.id) { + TicketWorkStatus::InProgress + } else { + TicketWorkStatus::Queued + }; + AgentTicketState { + ticket_id: snapshot.id, + conversation_id: snapshot.conversation, + position: snapshot.position, + status, + source: snapshot.source.into(), + requester_label: snapshot.requester, + task_preview: task_preview(&snapshot.task), + task_len: snapshot.task.chars().count(), + } +} + +/// Builds a bounded, whitespace-normalised excerpt of a task for the UI panel. +/// +/// Trims, collapses any run of whitespace to a single space, then truncates to +/// [`TASK_PREVIEW_MAX_CHARS`] characters (never mid-codepoint). The original length +/// is reported separately as [`AgentTicketState::task_len`]. +fn task_preview(task: &str) -> String { + let normalised = task.split_whitespace().collect::>().join(" "); + if normalised.chars().count() <= TASK_PREVIEW_MAX_CHARS { + normalised + } else { + normalised.chars().take(TASK_PREVIEW_MAX_CHARS).collect() + } +} + fn live_by_agent(snapshots: Vec) -> HashMap { let mut out = HashMap::new(); for snapshot in snapshots { diff --git a/crates/application/tests/workstate.rs b/crates/application/tests/workstate.rs index 0be9502..cd8c5da 100644 --- a/crates/application/tests/workstate.rs +++ b/crates/application/tests/workstate.rs @@ -9,16 +9,18 @@ use async_trait::async_trait; use application::{ GetProjectWorkState, GetProjectWorkStateInput, LiveSessionKind, LiveSessions, - StructuredSessions, TerminalSessions, + StructuredSessions, TerminalSessions, TicketWorkSource, TicketWorkStatus, +}; +use domain::mailbox::{ + AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, }; -use domain::mailbox::{MailboxError, PendingReply, Ticket}; use domain::ports::{ AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError, }; use domain::{ - Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, InputMediator, ManifestEntry, - MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath, PtySize, RemoteRef, SessionId, - SessionKind, TerminalSession, + Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, ConversationId, InputMediator, + InputSource, ManifestEntry, MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath, + PtySize, RemoteRef, SessionId, SessionKind, TerminalSession, TicketId, }; use uuid::Uuid; @@ -138,6 +140,45 @@ impl InputMediator for FakeInput { } } +#[derive(Default)] +struct FakeQueue { + queues: Mutex>>, +} + +impl FakeQueue { + fn set(&self, agent: AgentId, tickets: Vec) { + self.queues.lock().unwrap().insert(agent, tickets); + } +} + +impl AgentQueueSnapshot for FakeQueue { + fn queue_for(&self, agent: AgentId) -> Vec { + self.queues + .lock() + .unwrap() + .get(&agent) + .cloned() + .unwrap_or_default() + } +} + +fn snapshot( + id: u128, + position: u32, + source: InputSource, + requester: &str, + task: &str, +) -> QueuedTicketSnapshot { + QueuedTicketSnapshot { + id: TicketId::from_uuid(Uuid::from_u128(id)), + source, + conversation: ConversationId::from_uuid(Uuid::from_u128(id + 1000)), + requester: requester.to_owned(), + task: task.to_owned(), + position, + } +} + struct FakeSession { id: SessionId, } @@ -188,6 +229,7 @@ struct Fixture { pty: Arc, structured: Arc, input: Arc, + queue: Arc, project: Project, } @@ -197,18 +239,22 @@ fn fixture(agents: &[Agent]) -> Fixture { let live = Arc::new(LiveSessions::new(Arc::clone(&pty), Arc::clone(&structured))); let input = Arc::new(FakeInput::default()); let input_port = Arc::clone(&input) as Arc; + let queue = Arc::new(FakeQueue::default()); + let queue_port = Arc::clone(&queue) as Arc; let usecase = GetProjectWorkState::new( Arc::new(FakeContexts { manifest: manifest(agents), }), live, input_port, + queue_port, ); Fixture { usecase, pty, structured, input, + queue, project: project(), } } @@ -305,3 +351,169 @@ async fn workstate_ignores_live_agents_absent_from_manifest() { assert_eq!(out.agents[0].agent_id, a.id); assert_eq!(out.agents[0].live, None); } + +#[tokio::test] +async fn workstate_agent_without_queue_has_no_tickets() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + assert!(out.agents[0].tickets.is_empty()); +} + +#[tokio::test] +async fn workstate_lists_two_tickets_in_fifo_order() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + let from = aid(20); + f.queue.set( + a.id, + vec![ + snapshot(1, 0, InputSource::agent(from), "Main", "first"), + snapshot(2, 1, InputSource::agent(from), "Main", "second"), + ], + ); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + let tickets = &out.agents[0].tickets; + assert_eq!(tickets.len(), 2); + assert_eq!(tickets[0].ticket_id, ticket_id(1)); + assert_eq!(tickets[0].position, 0); + assert_eq!(tickets[1].ticket_id, ticket_id(2)); + assert_eq!(tickets[1].position, 1); +} + +#[tokio::test] +async fn workstate_marks_busy_head_in_progress_and_rest_queued() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + let from = aid(20); + f.queue.set( + a.id, + vec![ + snapshot(1, 0, InputSource::agent(from), "Main", "first"), + snapshot(2, 1, InputSource::agent(from), "Main", "second"), + ], + ); + f.input.set_busy( + a.id, + AgentBusyState::Busy { + ticket: ticket_id(1), + since_ms: 5, + }, + ); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + let tickets = &out.agents[0].tickets; + assert_eq!(tickets[0].status, TicketWorkStatus::InProgress); + assert_eq!(tickets[1].status, TicketWorkStatus::Queued); +} + +#[tokio::test] +async fn workstate_marks_all_queued_when_agent_idle() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + let from = aid(20); + f.queue.set( + a.id, + vec![snapshot(1, 0, InputSource::agent(from), "Main", "first")], + ); + // Agent left Idle (default): even the head is only queued, not in-progress. + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + assert_eq!(out.agents[0].tickets[0].status, TicketWorkStatus::Queued); +} + +#[tokio::test] +async fn workstate_ignores_queue_for_agent_absent_from_manifest() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + // A queue exists for an agent that is not in the manifest: it must not surface. + f.queue.set( + aid(999), + vec![snapshot(1, 0, InputSource::Human, "User", "ghost")], + ); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + assert_eq!(out.agents.len(), 1); + assert!(out.agents[0].tickets.is_empty()); +} + +#[tokio::test] +async fn workstate_truncates_task_preview_and_keeps_original_len() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + // 400 chars with runs of whitespace to normalise; well over the 160 cap. + let long_task = format!("start{}end", " x ".repeat(80)); + let original_len = long_task.chars().count(); + f.queue.set( + a.id, + vec![snapshot(1, 0, InputSource::Human, "User", &long_task)], + ); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + let ticket = &out.agents[0].tickets[0]; + assert_eq!(ticket.task_preview.chars().count(), 160); + assert!(!ticket.task_preview.contains(" "), "whitespace normalised"); + assert_eq!(ticket.task_len, original_len); + assert!(ticket.task_len > 160); +} + +#[tokio::test] +async fn workstate_maps_human_and_agent_ticket_sources() { + let a = agent(10, "alpha"); + let f = fixture(std::slice::from_ref(&a)); + let from = aid(20); + f.queue.set( + a.id, + vec![ + snapshot(1, 0, InputSource::Human, "User", "from human"), + snapshot(2, 1, InputSource::agent(from), "Main", "from agent"), + ], + ); + + let out = f + .usecase + .execute(GetProjectWorkStateInput { project: f.project }) + .await + .unwrap(); + + let tickets = &out.agents[0].tickets; + assert_eq!(tickets[0].source, TicketWorkSource::Human); + assert_eq!(tickets[0].requester_label, "User"); + assert_eq!( + tickets[1].source, + TicketWorkSource::Agent { agent_id: from } + ); + assert_eq!(tickets[1].requester_label, "Main"); +} diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 4cbc8db..996a63f 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -82,7 +82,10 @@ pub use profile::{ McpServerWiring, RateLimitPattern, SessionStrategy, }; -pub use mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId}; +pub use mailbox::{ + AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, + TicketId, +}; pub use conversation::{ Conversation, ConversationError, ConversationId, ConversationParty, ConversationRegistry, diff --git a/crates/domain/src/mailbox.rs b/crates/domain/src/mailbox.rs index 4eeb70a..92f813f 100644 --- a/crates/domain/src/mailbox.rs +++ b/crates/domain/src/mailbox.rs @@ -33,6 +33,47 @@ use crate::conversation::ConversationId; use crate::ids::AgentId; use crate::input::InputSource; +/// A read-only, cloned view of one queued [`Ticket`] in a target agent's FIFO. +/// +/// Carries **only the data** of a ticket (never the one-shot reply sender, which +/// stays inside the adapter), plus its `position` in the FIFO at snapshot time +/// (`0` = head). Pure value object used by the [`AgentQueueSnapshot`] read port so +/// the work-state read model can list pending delegations without touching the +/// mutating [`AgentMailbox`] surface (ISP). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct QueuedTicketSnapshot { + /// Stable id of the queued ticket. + pub id: TicketId, + /// The origin of the input (Human or a delegating Agent). + pub source: InputSource, + /// The conversation thread this task enters. + pub conversation: ConversationId, + /// Display name (or id) of the requester. + pub requester: String, + /// The full task/message text (preview/truncation is the caller's concern). + pub task: String, + /// Position in the FIFO at snapshot time (`0` = head, recomputed each call). + pub position: u32, +} + +/// Read-only inspection port over the per-agent delegation FIFO. +/// +/// Segregated from the mutating [`AgentMailbox`] (Interface Segregation): the +/// work-state read model depends on this port to *observe* the queue, never on +/// `enqueue`/`resolve`/`cancel`. Implementations return **cloned** snapshots in +/// FIFO order with recomputed positions; observing the queue must never mutate it. +/// +/// Object-safe (`&self`) so the application layer holds it as +/// `Arc`; one concrete `InMemoryMailbox` can be shared as +/// both an [`AgentMailbox`] (mutation) and an [`AgentQueueSnapshot`] (read) view. +pub trait AgentQueueSnapshot: Send + Sync { + /// Returns a cloned, FIFO-ordered snapshot of `agent`'s queued tickets. + /// + /// An agent with no queue yields an empty `Vec`. Positions are recomputed from + /// the current order (`0` = head). Pure read: the queue is left untouched. + fn queue_for(&self, agent: AgentId) -> Vec; +} + /// Identifies one queued [`Ticket`] within a target agent's mailbox. /// /// Newtype around [`uuid::Uuid`]; minted by the adapter on `enqueue`. It is **never @@ -310,4 +351,22 @@ mod tests { let a = AgentId::from_uuid(uuid::Uuid::from_u128(1)); assert_ne!(MailboxError::NoPendingRequest(a), MailboxError::Cancelled); } + + #[test] + fn queued_ticket_snapshot_carries_ticket_data_and_position() { + let conv = ConversationId::from_uuid(uuid::Uuid::from_u128(3)); + let from = AgentId::from_uuid(uuid::Uuid::from_u128(4)); + let snap = QueuedTicketSnapshot { + id: TicketId::from_uuid(uuid::Uuid::from_u128(8)), + source: InputSource::agent(from), + conversation: conv, + requester: "Main".to_owned(), + task: "delegate".to_owned(), + position: 2, + }; + assert_eq!(snap.source.as_agent(), Some(from)); + assert_eq!(snap.conversation, conv); + assert_eq!(snap.position, 2); + assert_eq!(snap.task, "delegate"); + } } diff --git a/crates/infrastructure/src/mailbox/mod.rs b/crates/infrastructure/src/mailbox/mod.rs index dba9095..93b262b 100644 --- a/crates/infrastructure/src/mailbox/mod.rs +++ b/crates/infrastructure/src/mailbox/mod.rs @@ -25,7 +25,10 @@ use std::sync::Mutex; use tokio::sync::oneshot; use domain::ids::AgentId; -use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId}; +use domain::mailbox::{ + AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket, + TicketId, +}; /// One queued request plus the sender that resolves its awaiting [`PendingReply`]. struct Slot { @@ -160,6 +163,35 @@ impl AgentMailbox for InMemoryMailbox { } } +impl AgentQueueSnapshot for InMemoryMailbox { + fn queue_for(&self, agent: AgentId) -> Vec { + // Pure read: clone each ticket's *data* (never the `oneshot::Sender`) under + // the lock, recomputing the FIFO position from the current order (0 = head). + // The queue is observed, not mutated. + let queues = self + .queues + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + queues + .get(&agent) + .map(|queue| { + queue + .iter() + .enumerate() + .map(|(idx, slot)| QueuedTicketSnapshot { + id: slot.ticket.id, + source: slot.ticket.source, + conversation: slot.ticket.conversation, + requester: slot.ticket.requester.clone(), + task: slot.ticket.task.clone(), + position: u32::try_from(idx).unwrap_or(u32::MAX), + }) + .collect() + }) + .unwrap_or_default() + } +} + #[cfg(test)] mod tests { use super::*; @@ -270,4 +302,98 @@ mod tests { Some(TicketId::from_uuid(uuid::Uuid::from_u128(10))) ); } + + fn tid(n: u128) -> TicketId { + TicketId::from_uuid(uuid::Uuid::from_u128(n)) + } + + #[test] + fn snapshot_of_empty_queue_is_empty() { + let mb = InMemoryMailbox::new(); + assert!(mb.queue_for(agent(1)).is_empty()); + } + + #[test] + fn snapshot_preserves_fifo_order_and_positions() { + let mb = InMemoryMailbox::new(); + let a = agent(1); + let _p1 = mb.enqueue(a, ticket(10, "first")); + let _p2 = mb.enqueue(a, ticket(11, "second")); + + let snap = mb.queue_for(a); + assert_eq!(snap.len(), 2); + assert_eq!(snap[0].id, tid(10)); + assert_eq!(snap[0].position, 0); + assert_eq!(snap[0].task, "first"); + assert_eq!(snap[1].id, tid(11)); + assert_eq!(snap[1].position, 1); + assert_eq!(snap[1].task, "second"); + } + + #[test] + fn snapshot_carries_ticket_metadata() { + use domain::conversation::ConversationId; + use domain::input::InputSource; + + let mb = InMemoryMailbox::new(); + let a = agent(1); + let from = agent(2); + let conv = ConversationId::from_uuid(uuid::Uuid::from_u128(42)); + let _p = mb.enqueue( + a, + Ticket::from_agent(tid(10), from, conv, "Main", "delegate task"), + ); + + let snap = mb.queue_for(a); + assert_eq!(snap.len(), 1); + assert_eq!(snap[0].source, InputSource::agent(from)); + assert_eq!(snap[0].conversation, conv); + assert_eq!(snap[0].requester, "Main"); + assert_eq!(snap[0].task, "delegate task"); + } + + #[test] + fn snapshot_is_read_only() { + let mb = InMemoryMailbox::new(); + let a = agent(1); + let _p1 = mb.enqueue(a, ticket(10, "first")); + let _p2 = mb.enqueue(a, ticket(11, "second")); + + let _ = mb.queue_for(a); + let _ = mb.queue_for(a); + + // Observing the queue did not mutate it. + assert_eq!(mb.pending(&a), 2); + assert_eq!(mb.head_ticket(&a), Some(tid(10))); + } + + #[test] + fn snapshot_updates_after_resolve_ticket() { + let mb = InMemoryMailbox::new(); + let a = agent(1); + let _p1 = mb.enqueue(a, ticket(10, "first")); + let _p2 = mb.enqueue(a, ticket(11, "second")); + + mb.resolve_ticket(a, tid(10), "done".to_owned()).unwrap(); + + let snap = mb.queue_for(a); + assert_eq!(snap.len(), 1); + assert_eq!(snap[0].id, tid(11)); + assert_eq!(snap[0].position, 0, "remaining ticket becomes head"); + } + + #[test] + fn snapshot_updates_after_cancel_head() { + let mb = InMemoryMailbox::new(); + let a = agent(1); + let _p1 = mb.enqueue(a, ticket(10, "head")); + let _p2 = mb.enqueue(a, ticket(11, "next")); + + mb.cancel_head(a, tid(10)); + + let snap = mb.queue_for(a); + assert_eq!(snap.len(), 1); + assert_eq!(snap[0].id, tid(11)); + assert_eq!(snap[0].position, 0, "next ticket recomputed to position 0"); + } }