feat(workstate): snapshot des délégations en file par agent (Lot B backend)
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<dyn …>`). - 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 <noreply@anthropic.com>
This commit is contained in:
@ -9,8 +9,9 @@
|
|||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use application::{
|
use application::{
|
||||||
AppError, CreateProjectInput, CreateProjectOutput, GitGraphOutput, HealthInput, HealthReport,
|
AgentTicketState, AppError, CreateProjectInput, CreateProjectOutput, GitGraphOutput,
|
||||||
LayoutKind, ListProjectsOutput, LiveSessionKind, OpenProjectOutput, ProjectWorkState,
|
HealthInput, HealthReport, LayoutKind, ListProjectsOutput, LiveSessionKind, OpenProjectOutput,
|
||||||
|
ProjectWorkState, TicketWorkSource, TicketWorkStatus,
|
||||||
};
|
};
|
||||||
use domain::{AgentBusyState, Project, ProjectId};
|
use domain::{AgentBusyState, Project, ProjectId};
|
||||||
|
|
||||||
@ -1503,6 +1504,87 @@ pub struct LiveWorkSessionDto {
|
|||||||
pub kind: LiveWorkSessionKindDto,
|
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<TicketWorkStatus> 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<TicketWorkSource> 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<AgentTicketState> 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.
|
/// One manifest agent's current live/busy state.
|
||||||
#[derive(Debug, Clone, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
#[serde(rename_all = "camelCase")]
|
#[serde(rename_all = "camelCase")]
|
||||||
@ -1517,6 +1599,8 @@ pub struct AgentWorkStateDto {
|
|||||||
pub live: Option<LiveWorkSessionDto>,
|
pub live: Option<LiveWorkSessionDto>,
|
||||||
/// Current mediated-input busy state.
|
/// Current mediated-input busy state.
|
||||||
pub busy: AgentBusyState,
|
pub busy: AgentBusyState,
|
||||||
|
/// Pending/in-progress delegation tickets, in FIFO order.
|
||||||
|
pub tickets: Vec<AgentTicketStateDto>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Project-level read model for conversation/delegation UX.
|
/// Project-level read model for conversation/delegation UX.
|
||||||
@ -1543,6 +1627,11 @@ impl From<ProjectWorkState> for ProjectWorkStateDto {
|
|||||||
kind: live.kind.into(),
|
kind: live.kind.into(),
|
||||||
}),
|
}),
|
||||||
busy: agent.busy,
|
busy: agent.busy,
|
||||||
|
tickets: agent
|
||||||
|
.tickets
|
||||||
|
.into_iter()
|
||||||
|
.map(AgentTicketStateDto::from)
|
||||||
|
.collect(),
|
||||||
})
|
})
|
||||||
.collect(),
|
.collect(),
|
||||||
}
|
}
|
||||||
|
|||||||
@ -993,6 +993,10 @@ impl AppState {
|
|||||||
// partagé pour `resolve`/`resolve_ticket`/`cancel_head` côté orchestrateur.
|
// partagé pour `resolve`/`resolve_ticket`/`cancel_head` côté orchestrateur.
|
||||||
let inmemory_mailbox = Arc::new(InMemoryMailbox::new());
|
let inmemory_mailbox = Arc::new(InMemoryMailbox::new());
|
||||||
let mailbox = Arc::clone(&inmemory_mailbox) as Arc<dyn domain::mailbox::AgentMailbox>;
|
let mailbox = Arc::clone(&inmemory_mailbox) as Arc<dyn domain::mailbox::AgentMailbox>;
|
||||||
|
// 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<dyn domain::mailbox::AgentQueueSnapshot>;
|
||||||
let mediated_inbox = Arc::new(
|
let mediated_inbox = Arc::new(
|
||||||
MediatedInbox::with_pty(
|
MediatedInbox::with_pty(
|
||||||
Arc::clone(&inmemory_mailbox),
|
Arc::clone(&inmemory_mailbox),
|
||||||
@ -1031,6 +1035,7 @@ impl AppState {
|
|||||||
Arc::clone(&contexts_port),
|
Arc::clone(&contexts_port),
|
||||||
Arc::clone(&live_sessions),
|
Arc::clone(&live_sessions),
|
||||||
Arc::clone(&input_mediator),
|
Arc::clone(&input_mediator),
|
||||||
|
queue_snapshot,
|
||||||
));
|
));
|
||||||
|
|
||||||
// --- Limites de session des agents (ARCHITECTURE §21, LS7) ---
|
// --- Limites de session des agents (ARCHITECTURE §21, LS7) ---
|
||||||
|
|||||||
@ -9,8 +9,9 @@ use app_tauri_lib::dto::{
|
|||||||
};
|
};
|
||||||
use application::AppError;
|
use application::AppError;
|
||||||
use application::{
|
use application::{
|
||||||
AgentWorkState, CreateAgentOutput, InspectConversationOutput, LaunchAgentOutput,
|
AgentTicketState, AgentWorkState, CreateAgentOutput, InspectConversationOutput,
|
||||||
ListAgentsOutput, LiveSessionKind, LiveWorkSession, ProjectWorkState,
|
LaunchAgentOutput, ListAgentsOutput, LiveSessionKind, LiveWorkSession, ProjectWorkState,
|
||||||
|
TicketWorkSource, TicketWorkStatus,
|
||||||
};
|
};
|
||||||
use domain::ids::{AgentId, NodeId, ProfileId, SessionId};
|
use domain::ids::{AgentId, NodeId, ProfileId, SessionId};
|
||||||
use domain::ports::ConversationDetails;
|
use domain::ports::ConversationDetails;
|
||||||
@ -219,6 +220,7 @@ fn project_work_state_dto_serialises_live_and_busy_camelcase() {
|
|||||||
ticket,
|
ticket,
|
||||||
since_ms: 1_234,
|
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"]["state"], "busy");
|
||||||
assert_eq!(v["agents"][0]["busy"]["ticket"], ticket.to_string());
|
assert_eq!(v["agents"][0]["busy"]["ticket"], ticket.to_string());
|
||||||
assert_eq!(v["agents"][0]["busy"]["sinceMs"], 1_234);
|
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].get("agent_id").is_none());
|
||||||
assert!(v["agents"][0]["live"].get("session_id").is_none());
|
assert!(v["agents"][0]["live"].get("session_id").is_none());
|
||||||
assert!(v["agents"][0]["busy"].get("since_ms").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]
|
#[test]
|
||||||
fn launch_agent_request_carries_conversation_id_for_resume() {
|
fn launch_agent_request_carries_conversation_id_for_resume() {
|
||||||
let raw = json!({
|
let raw = json!({
|
||||||
|
|||||||
@ -119,6 +119,6 @@ pub use terminal::{
|
|||||||
};
|
};
|
||||||
pub use window::{MoveTabToNewWindow, MoveTabToNewWindowInput, MoveTabToNewWindowOutput};
|
pub use window::{MoveTabToNewWindow, MoveTabToNewWindowInput, MoveTabToNewWindowOutput};
|
||||||
pub use workstate::{
|
pub use workstate::{
|
||||||
AgentWorkState, GetProjectWorkState, GetProjectWorkStateInput, LiveWorkSession,
|
AgentTicketState, AgentWorkState, GetProjectWorkState, GetProjectWorkStateInput,
|
||||||
ProjectWorkState,
|
LiveWorkSession, ProjectWorkState, TicketWorkSource, TicketWorkStatus,
|
||||||
};
|
};
|
||||||
|
|||||||
@ -8,13 +8,22 @@
|
|||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use domain::input::AgentBusyState;
|
use domain::input::{AgentBusyState, InputSource};
|
||||||
use domain::ports::AgentContextStore;
|
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::error::AppError;
|
||||||
use crate::terminal::{LiveSessionKind, LiveSessionSnapshot, LiveSessions};
|
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`].
|
/// Input for [`GetProjectWorkState::execute`].
|
||||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
pub struct GetProjectWorkStateInput {
|
pub struct GetProjectWorkStateInput {
|
||||||
@ -42,6 +51,64 @@ pub struct AgentWorkState {
|
|||||||
pub live: Option<LiveWorkSession>,
|
pub live: Option<LiveWorkSession>,
|
||||||
/// Current FIFO/busy state.
|
/// Current FIFO/busy state.
|
||||||
pub busy: AgentBusyState,
|
pub busy: AgentBusyState,
|
||||||
|
/// Pending/in-progress delegation tickets, in FIFO order.
|
||||||
|
pub tickets: Vec<AgentTicketState>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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<InputSource> 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.
|
/// Live session coordinates exposed by the work-state read model.
|
||||||
@ -60,6 +127,7 @@ pub struct GetProjectWorkState {
|
|||||||
contexts: Arc<dyn AgentContextStore>,
|
contexts: Arc<dyn AgentContextStore>,
|
||||||
live: Arc<LiveSessions>,
|
live: Arc<LiveSessions>,
|
||||||
input: Arc<dyn InputMediator>,
|
input: Arc<dyn InputMediator>,
|
||||||
|
queue: Arc<dyn AgentQueueSnapshot>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl GetProjectWorkState {
|
impl GetProjectWorkState {
|
||||||
@ -69,11 +137,13 @@ impl GetProjectWorkState {
|
|||||||
contexts: Arc<dyn AgentContextStore>,
|
contexts: Arc<dyn AgentContextStore>,
|
||||||
live: Arc<LiveSessions>,
|
live: Arc<LiveSessions>,
|
||||||
input: Arc<dyn InputMediator>,
|
input: Arc<dyn InputMediator>,
|
||||||
|
queue: Arc<dyn AgentQueueSnapshot>,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
contexts,
|
contexts,
|
||||||
live,
|
live,
|
||||||
input,
|
input,
|
||||||
|
queue,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -102,12 +172,23 @@ impl GetProjectWorkState {
|
|||||||
session_id: snapshot.session_id,
|
session_id: snapshot.session_id,
|
||||||
kind: snapshot.kind,
|
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 {
|
Ok(AgentWorkState {
|
||||||
agent_id: agent.id,
|
agent_id: agent.id,
|
||||||
name: agent.name,
|
name: agent.name,
|
||||||
profile_id: agent.profile_id,
|
profile_id: agent.profile_id,
|
||||||
live,
|
live,
|
||||||
busy: self.input.busy_state(agent.id),
|
busy: self.input.busy_state(agent.id),
|
||||||
|
tickets,
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.collect::<Result<Vec<_>, AppError>>()?;
|
.collect::<Result<Vec<_>, 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<TicketId>) -> 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::<Vec<_>>().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<LiveSessionSnapshot>) -> HashMap<AgentId, LiveSessionSnapshot> {
|
fn live_by_agent(snapshots: Vec<LiveSessionSnapshot>) -> HashMap<AgentId, LiveSessionSnapshot> {
|
||||||
let mut out = HashMap::new();
|
let mut out = HashMap::new();
|
||||||
for snapshot in snapshots {
|
for snapshot in snapshots {
|
||||||
|
|||||||
@ -9,16 +9,18 @@ use async_trait::async_trait;
|
|||||||
|
|
||||||
use application::{
|
use application::{
|
||||||
GetProjectWorkState, GetProjectWorkStateInput, LiveSessionKind, LiveSessions,
|
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::{
|
use domain::ports::{
|
||||||
AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError,
|
AgentContextStore, AgentSession, AgentSessionError, PtyHandle, ReplyStream, StoreError,
|
||||||
};
|
};
|
||||||
use domain::{
|
use domain::{
|
||||||
Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, InputMediator, ManifestEntry,
|
Agent, AgentBusyState, AgentId, AgentManifest, AgentOrigin, ConversationId, InputMediator,
|
||||||
MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath, PtySize, RemoteRef, SessionId,
|
InputSource, ManifestEntry, MarkdownDoc, NodeId, ProfileId, Project, ProjectId, ProjectPath,
|
||||||
SessionKind, TerminalSession,
|
PtySize, RemoteRef, SessionId, SessionKind, TerminalSession, TicketId,
|
||||||
};
|
};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@ -138,6 +140,45 @@ impl InputMediator for FakeInput {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct FakeQueue {
|
||||||
|
queues: Mutex<HashMap<AgentId, Vec<QueuedTicketSnapshot>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl FakeQueue {
|
||||||
|
fn set(&self, agent: AgentId, tickets: Vec<QueuedTicketSnapshot>) {
|
||||||
|
self.queues.lock().unwrap().insert(agent, tickets);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AgentQueueSnapshot for FakeQueue {
|
||||||
|
fn queue_for(&self, agent: AgentId) -> Vec<QueuedTicketSnapshot> {
|
||||||
|
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 {
|
struct FakeSession {
|
||||||
id: SessionId,
|
id: SessionId,
|
||||||
}
|
}
|
||||||
@ -188,6 +229,7 @@ struct Fixture {
|
|||||||
pty: Arc<TerminalSessions>,
|
pty: Arc<TerminalSessions>,
|
||||||
structured: Arc<StructuredSessions>,
|
structured: Arc<StructuredSessions>,
|
||||||
input: Arc<FakeInput>,
|
input: Arc<FakeInput>,
|
||||||
|
queue: Arc<FakeQueue>,
|
||||||
project: Project,
|
project: Project,
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -197,18 +239,22 @@ fn fixture(agents: &[Agent]) -> Fixture {
|
|||||||
let live = Arc::new(LiveSessions::new(Arc::clone(&pty), Arc::clone(&structured)));
|
let live = Arc::new(LiveSessions::new(Arc::clone(&pty), Arc::clone(&structured)));
|
||||||
let input = Arc::new(FakeInput::default());
|
let input = Arc::new(FakeInput::default());
|
||||||
let input_port = Arc::clone(&input) as Arc<dyn InputMediator>;
|
let input_port = Arc::clone(&input) as Arc<dyn InputMediator>;
|
||||||
|
let queue = Arc::new(FakeQueue::default());
|
||||||
|
let queue_port = Arc::clone(&queue) as Arc<dyn AgentQueueSnapshot>;
|
||||||
let usecase = GetProjectWorkState::new(
|
let usecase = GetProjectWorkState::new(
|
||||||
Arc::new(FakeContexts {
|
Arc::new(FakeContexts {
|
||||||
manifest: manifest(agents),
|
manifest: manifest(agents),
|
||||||
}),
|
}),
|
||||||
live,
|
live,
|
||||||
input_port,
|
input_port,
|
||||||
|
queue_port,
|
||||||
);
|
);
|
||||||
Fixture {
|
Fixture {
|
||||||
usecase,
|
usecase,
|
||||||
pty,
|
pty,
|
||||||
structured,
|
structured,
|
||||||
input,
|
input,
|
||||||
|
queue,
|
||||||
project: project(),
|
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].agent_id, a.id);
|
||||||
assert_eq!(out.agents[0].live, None);
|
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");
|
||||||
|
}
|
||||||
|
|||||||
@ -82,7 +82,10 @@ pub use profile::{
|
|||||||
McpServerWiring, RateLimitPattern, SessionStrategy,
|
McpServerWiring, RateLimitPattern, SessionStrategy,
|
||||||
};
|
};
|
||||||
|
|
||||||
pub use mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId};
|
pub use mailbox::{
|
||||||
|
AgentMailbox, AgentQueueSnapshot, MailboxError, PendingReply, QueuedTicketSnapshot, Ticket,
|
||||||
|
TicketId,
|
||||||
|
};
|
||||||
|
|
||||||
pub use conversation::{
|
pub use conversation::{
|
||||||
Conversation, ConversationError, ConversationId, ConversationParty, ConversationRegistry,
|
Conversation, ConversationError, ConversationId, ConversationParty, ConversationRegistry,
|
||||||
|
|||||||
@ -33,6 +33,47 @@ use crate::conversation::ConversationId;
|
|||||||
use crate::ids::AgentId;
|
use crate::ids::AgentId;
|
||||||
use crate::input::InputSource;
|
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<dyn AgentQueueSnapshot>`; 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<QueuedTicketSnapshot>;
|
||||||
|
}
|
||||||
|
|
||||||
/// Identifies one queued [`Ticket`] within a target agent's mailbox.
|
/// Identifies one queued [`Ticket`] within a target agent's mailbox.
|
||||||
///
|
///
|
||||||
/// Newtype around [`uuid::Uuid`]; minted by the adapter on `enqueue`. It is **never
|
/// 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));
|
let a = AgentId::from_uuid(uuid::Uuid::from_u128(1));
|
||||||
assert_ne!(MailboxError::NoPendingRequest(a), MailboxError::Cancelled);
|
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");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -25,7 +25,10 @@ use std::sync::Mutex;
|
|||||||
use tokio::sync::oneshot;
|
use tokio::sync::oneshot;
|
||||||
|
|
||||||
use domain::ids::AgentId;
|
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`].
|
/// One queued request plus the sender that resolves its awaiting [`PendingReply`].
|
||||||
struct Slot {
|
struct Slot {
|
||||||
@ -160,6 +163,35 @@ impl AgentMailbox for InMemoryMailbox {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl AgentQueueSnapshot for InMemoryMailbox {
|
||||||
|
fn queue_for(&self, agent: AgentId) -> Vec<QueuedTicketSnapshot> {
|
||||||
|
// 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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@ -270,4 +302,98 @@ mod tests {
|
|||||||
Some(TicketId::from_uuid(uuid::Uuid::from_u128(10)))
|
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");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user