Blomios
803e9de093
Cancel transitif, idempotent et ownership-aware sur le graphe parent→enfants des background tasks: - domain/background_task + ports: modèle parent/enfant (parent_task_id, rejet du self-parent), index children_by_parent, traversal des descendants en largeur. - application/background + infrastructure/store: cascade cancel (annule les descendants running/open, laisse les terminaux), idempotente face aux double-cancel et ordres de terminaison non déterministes; respect de l'ownership (parent_task_link). - orchestrator/service: stop agent → cascade cancel des tasks lancées par cet agent (relatesTo #196); cancel task parent → cascade de ses sous-tasks (dependsOn #199). Refactor unifiant le chemin de cancel, sans sémantique #196/#199 regreffée (tests hérités verts). - frontend: projection parentTaskId (workstate normalization/domain/services). Tests QA verts: domain background_task, application background_task_cancel (cascades_background_task_subtree, cascades_to_running_child, cascades_to_open_descendants_and_leaves_terminal_child, parent_task_link_roundtrips_and_rejects_self_parent), orchestrator_service, workstate_actions, workstate.test.tsx.
874 lines
32 KiB
Rust
874 lines
32 KiB
Rust
//! Read-only project work-state read model.
|
|
//!
|
|
//! This module composes existing runtime state only: the project agent manifest,
|
|
//! live session registries and the input mediator busy state. It performs no I/O
|
|
//! beyond loading the manifest through the existing context store and creates no
|
|
//! durable projection.
|
|
|
|
mod actions;
|
|
mod live;
|
|
mod reconcile;
|
|
|
|
pub use actions::{
|
|
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopAgentExecutionCascade,
|
|
StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
|
|
};
|
|
pub use live::{
|
|
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
|
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
|
};
|
|
pub use reconcile::{ReconcileLiveState, ReconcileLiveStateInput};
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::sync::Arc;
|
|
|
|
use domain::input::{AgentBusyState, InputSource};
|
|
use domain::ports::{AgentContextStore, BackgroundTaskPortError, BackgroundTaskStore};
|
|
use domain::{
|
|
AgentId, AgentQueueSnapshot, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult,
|
|
BackgroundTaskState, ConversationId, ConversationLog, ConversationTurn, Handoff, HandoffStore,
|
|
InputMediator, NodeId, ProfileId, Project, ProjectPath, QueuedTicketSnapshot, SessionId,
|
|
TaskId, TicketId, TurnId, TurnRole,
|
|
};
|
|
|
|
use crate::error::AppError;
|
|
use crate::terminal::{LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, LiveSessions};
|
|
use crate::HandoffProvider;
|
|
|
|
/// 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;
|
|
|
|
/// Maximum length (in characters) of a conversation [`ConversationWorkSummary::summary_preview`].
|
|
const HANDOFF_PREVIEW_MAX_CHARS: usize = 480;
|
|
|
|
/// Maximum length (in characters) of a conversation [`ConversationWorkSummary::objective_preview`].
|
|
const OBJECTIVE_PREVIEW_MAX_CHARS: usize = 160;
|
|
|
|
/// Maximum number of recent turns surfaced in a [`ConversationWorkSummary::recent_turns`] fallback.
|
|
const RECENT_TURNS_MAX: usize = 3;
|
|
|
|
/// Maximum length (in characters) of a [`ConversationTurnWorkPreview::text_preview`].
|
|
const TURN_PREVIEW_MAX_CHARS: usize = 220;
|
|
|
|
/// Input for [`GetProjectWorkState::execute`].
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct GetProjectWorkStateInput {
|
|
/// Project whose manifest provides the agent order and boundary.
|
|
pub project: Project,
|
|
}
|
|
|
|
/// Read model for a project's current agent work state.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ProjectWorkState {
|
|
/// Agents in manifest order.
|
|
pub agents: Vec<AgentWorkState>,
|
|
/// Best-effort summaries of the conversations visible through the agents'
|
|
/// tickets, joined frontend-side via `tickets[].conversation_id`. Built from
|
|
/// the [`HandoffStore`] (primary) with a bounded [`ConversationLog`] fallback;
|
|
/// a preview failure never blocks `agents`.
|
|
pub conversations: Vec<ConversationWorkSummary>,
|
|
}
|
|
|
|
/// Input for [`GetAppExitWorkGuardState::execute`].
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct GetAppExitWorkGuardStateInput {
|
|
/// Projects currently open in the application.
|
|
pub projects: Vec<Project>,
|
|
}
|
|
|
|
/// Application-wide shutdown guard summary for the close-confirmation UX.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct AppExitWorkGuardState {
|
|
/// Whether at least one active work item would be interrupted by app exit.
|
|
pub has_work_in_progress: bool,
|
|
/// Number of busy agents across all open projects.
|
|
pub busy_agent_count: usize,
|
|
/// Number of non-terminal background tasks across all open projects.
|
|
pub active_background_task_count: usize,
|
|
/// Best-effort detail for compact UX display.
|
|
pub details: Vec<AppExitWorkGuardDetail>,
|
|
}
|
|
|
|
/// One work item contributing to [`AppExitWorkGuardState`].
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum AppExitWorkGuardDetail {
|
|
/// A manifest agent is currently processing a turn.
|
|
BusyAgent {
|
|
/// Owning project id.
|
|
project_id: domain::ProjectId,
|
|
/// Owning project display name.
|
|
project_name: String,
|
|
/// Agent id.
|
|
agent_id: AgentId,
|
|
/// Agent display name.
|
|
agent_name: String,
|
|
/// Busy ticket, if carried by the mediator state.
|
|
ticket_id: Option<TicketId>,
|
|
},
|
|
/// A first-class background task is queued, running or waiting.
|
|
ActiveBackgroundTask {
|
|
/// Owning project id.
|
|
project_id: domain::ProjectId,
|
|
/// Owning project display name.
|
|
project_name: String,
|
|
/// Owning agent id.
|
|
agent_id: AgentId,
|
|
/// Owning agent display name.
|
|
agent_name: String,
|
|
/// Stable task id.
|
|
task_id: TaskId,
|
|
/// Lifecycle state.
|
|
state: BackgroundTaskState,
|
|
/// Kind discriminant.
|
|
kind: BackgroundTaskKindLabel,
|
|
},
|
|
}
|
|
|
|
/// Best-effort, read-only summary of one conversation visible through the tickets.
|
|
///
|
|
/// Derived live from the [`HandoffStore`] (primary source) with a bounded
|
|
/// [`ConversationLog`] fallback — never a durable projection, never a repair of the
|
|
/// underlying log/handoff. Joined frontend-side to tickets via
|
|
/// [`ConversationWorkSummary::conversation_id`].
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ConversationWorkSummary {
|
|
/// The conversation (pair) this summary describes.
|
|
pub conversation_id: ConversationId,
|
|
/// How much could be derived (see [`ConversationPreviewStatus`]).
|
|
pub status: ConversationPreviewStatus,
|
|
/// Bounded excerpt of the handoff objective, when present.
|
|
pub objective_preview: Option<String>,
|
|
/// Bounded excerpt of the handoff summary, when a handoff is readable.
|
|
pub summary_preview: Option<String>,
|
|
/// Character length of the **original** (un-truncated) handoff summary (`0` when none).
|
|
pub summary_len: usize,
|
|
/// Cursor (last [`TurnId`]) covered by the handoff summary, when present.
|
|
pub up_to: Option<TurnId>,
|
|
/// Bounded, recent turns from the log fallback (empty when a handoff is `Ready`).
|
|
pub recent_turns: Vec<ConversationTurnWorkPreview>,
|
|
}
|
|
|
|
/// How much of a [`ConversationWorkSummary`] could be derived, best-effort.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum ConversationPreviewStatus {
|
|
/// A handoff was present and usable: summary/objective/cursor are filled.
|
|
Ready,
|
|
/// No handoff yet; `recent_turns` carries the log fallback if any.
|
|
Missing,
|
|
/// The handoff was unreadable but the log fallback was readable.
|
|
Partial,
|
|
/// Neither the handoff nor the log could be read.
|
|
Unavailable,
|
|
}
|
|
|
|
/// One recent turn, projected for the conversation-summary fallback.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ConversationTurnWorkPreview {
|
|
/// Nature of the turn (prompt, response, tool activity).
|
|
pub role: TurnRole,
|
|
/// Origin of the turn (human operator or a delegating agent).
|
|
pub source: TicketWorkSource,
|
|
/// Timestamp (epoch milliseconds) of the turn.
|
|
pub at_ms: u64,
|
|
/// Bounded, whitespace-normalised excerpt of the turn text.
|
|
pub text_preview: String,
|
|
/// Character length of the **original** (un-truncated) turn text.
|
|
pub text_len: usize,
|
|
}
|
|
|
|
/// Builds the [`ConversationLog`] bound to a given project `root`, best-effort.
|
|
///
|
|
/// Twin of [`HandoffProvider`]: the work-state read model is a singleton (the project
|
|
/// root arrives per call via [`GetProjectWorkStateInput`]) while the `Fs*` adapters fix
|
|
/// their root at construction. This port materialises the log targeting the **right**
|
|
/// folder on each call; `None` ⇒ no log fallback (zero regression for call
|
|
/// sites/tests that do not wire it). Implemented in `app-tauri` (sole owner of `Fs*`).
|
|
pub trait ConversationLogProvider: Send + Sync {
|
|
/// Builds the [`ConversationLog`] whose persistence targets `root`, or `None`.
|
|
fn conversation_log_for(&self, root: &ProjectPath) -> Option<Arc<dyn ConversationLog>>;
|
|
}
|
|
|
|
/// Read model for one manifest agent.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct AgentWorkState {
|
|
/// Agent id.
|
|
pub agent_id: AgentId,
|
|
/// Agent display name.
|
|
pub name: String,
|
|
/// Runtime profile id currently assigned to the agent.
|
|
pub profile_id: ProfileId,
|
|
/// Current live session, if the agent is live.
|
|
pub live: Option<LiveWorkSession>,
|
|
/// Last known runtime lifecycle state for the agent, distinct from `live`.
|
|
pub runtime_state: Option<LiveRuntimeState>,
|
|
/// Current FIFO/busy state.
|
|
pub busy: AgentBusyState,
|
|
/// Pending/in-progress delegation tickets, in FIFO order.
|
|
pub tickets: Vec<AgentTicketState>,
|
|
/// Best-effort first-class background tasks owned by this agent.
|
|
///
|
|
/// V1 bound: terminal tasks already delivered to the owner are no longer
|
|
/// enumerable by the background-task store, so they disappear from Work;
|
|
/// retry is available only while the terminal completion is still undelivered.
|
|
pub background_tasks: Vec<AgentBackgroundTaskState>,
|
|
}
|
|
|
|
/// Background task projected into an agent's work state.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct AgentBackgroundTaskState {
|
|
/// Stable task id.
|
|
pub task_id: TaskId,
|
|
/// Optional parent background task id.
|
|
pub parent_task_id: Option<TaskId>,
|
|
/// Kind discriminant.
|
|
pub kind: BackgroundTaskKindLabel,
|
|
/// Current lifecycle state.
|
|
pub state: BackgroundTaskState,
|
|
/// Process exit code, when the terminal result carries one.
|
|
pub exit_code: Option<i32>,
|
|
/// Human-readable summary / error / reason of the terminal result.
|
|
pub summary: Option<String>,
|
|
/// Bounded stdout tail.
|
|
pub stdout_tail: Option<String>,
|
|
/// Bounded stderr tail.
|
|
pub stderr_tail: Option<String>,
|
|
/// Agent that requested a headless rendezvous, when this task is one.
|
|
pub requester_agent_id: Option<AgentId>,
|
|
/// Target agent for a headless rendezvous, when this task is one.
|
|
pub target_agent_id: Option<AgentId>,
|
|
/// Conversation opened by a headless rendezvous, when this task is one.
|
|
pub conversation_id: Option<domain::ConversationId>,
|
|
/// Creation timestamp, epoch milliseconds.
|
|
pub created_at_ms: u64,
|
|
/// Last update timestamp, epoch milliseconds.
|
|
pub updated_at_ms: u64,
|
|
}
|
|
|
|
/// Background task kind label used by the work-state read model.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum BackgroundTaskKindLabel {
|
|
/// A non-interactive command.
|
|
Command,
|
|
/// A headless inter-agent rendezvous.
|
|
HeadlessRendezvous,
|
|
/// Resume of a structured agent session.
|
|
SessionResume,
|
|
/// Internal maintenance work.
|
|
Maintenance,
|
|
}
|
|
|
|
/// 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.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct LiveWorkSession {
|
|
/// The layout node currently hosting the session view.
|
|
pub node_id: NodeId,
|
|
/// The live session id.
|
|
pub session_id: SessionId,
|
|
/// Runtime family that owns the session.
|
|
pub kind: LiveSessionKind,
|
|
}
|
|
|
|
/// Read-only use case aggregating manifest agents with live and busy state.
|
|
pub struct GetProjectWorkState {
|
|
contexts: Arc<dyn AgentContextStore>,
|
|
live: Arc<LiveSessions>,
|
|
input: Arc<dyn InputMediator>,
|
|
queue: Arc<dyn AgentQueueSnapshot>,
|
|
/// Per-root handoff source (primary), wired best-effort; `None` ⇒ no summaries.
|
|
handoffs: Option<Arc<dyn HandoffProvider>>,
|
|
/// Per-root log source (fallback), wired best-effort; `None` ⇒ no fallback turns.
|
|
logs: Option<Arc<dyn ConversationLogProvider>>,
|
|
/// Background-task read model source, wired best-effort; `None` ⇒ no tasks.
|
|
background_tasks: Option<Arc<dyn BackgroundTaskStore>>,
|
|
}
|
|
|
|
/// Read-only use case aggregating app-wide work that should guard application exit.
|
|
pub struct GetAppExitWorkGuardState {
|
|
work_state: Arc<GetProjectWorkState>,
|
|
}
|
|
|
|
impl GetAppExitWorkGuardState {
|
|
/// Builds the app-exit guard from the existing per-project work-state read model.
|
|
#[must_use]
|
|
pub fn new(work_state: Arc<GetProjectWorkState>) -> Self {
|
|
Self { work_state }
|
|
}
|
|
|
|
/// Executes the guard aggregation across all currently open projects.
|
|
///
|
|
/// # Errors
|
|
/// Propagates the per-project work-state read errors.
|
|
pub async fn execute(
|
|
&self,
|
|
input: GetAppExitWorkGuardStateInput,
|
|
) -> Result<AppExitWorkGuardState, AppError> {
|
|
let mut busy_agent_count = 0;
|
|
let mut active_background_task_count = 0;
|
|
let mut details = Vec::new();
|
|
|
|
for project in input.projects {
|
|
let project_id = project.id;
|
|
let project_name = project.name.clone();
|
|
let state = self
|
|
.work_state
|
|
.execute(GetProjectWorkStateInput {
|
|
project: project.clone(),
|
|
})
|
|
.await?;
|
|
|
|
for agent in state.agents {
|
|
if agent.busy.is_busy() {
|
|
busy_agent_count += 1;
|
|
details.push(AppExitWorkGuardDetail::BusyAgent {
|
|
project_id,
|
|
project_name: project_name.clone(),
|
|
agent_id: agent.agent_id,
|
|
agent_name: agent.name.clone(),
|
|
ticket_id: agent.busy.ticket(),
|
|
});
|
|
}
|
|
|
|
for task in agent
|
|
.background_tasks
|
|
.into_iter()
|
|
.filter(|task| !task.state.is_terminal())
|
|
{
|
|
active_background_task_count += 1;
|
|
details.push(AppExitWorkGuardDetail::ActiveBackgroundTask {
|
|
project_id,
|
|
project_name: project_name.clone(),
|
|
agent_id: agent.agent_id,
|
|
agent_name: agent.name.clone(),
|
|
task_id: task.task_id,
|
|
state: task.state,
|
|
kind: task.kind,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(AppExitWorkGuardState {
|
|
has_work_in_progress: busy_agent_count > 0 || active_background_task_count > 0,
|
|
busy_agent_count,
|
|
active_background_task_count,
|
|
details,
|
|
})
|
|
}
|
|
}
|
|
|
|
impl GetProjectWorkState {
|
|
/// Builds the read-model use case from existing stores/registries.
|
|
///
|
|
/// Conversation summaries are off by default; wire them with
|
|
/// [`Self::with_conversation_sources`].
|
|
#[must_use]
|
|
pub fn new(
|
|
contexts: Arc<dyn AgentContextStore>,
|
|
live: Arc<LiveSessions>,
|
|
input: Arc<dyn InputMediator>,
|
|
queue: Arc<dyn AgentQueueSnapshot>,
|
|
) -> Self {
|
|
Self {
|
|
contexts,
|
|
live,
|
|
input,
|
|
queue,
|
|
handoffs: None,
|
|
logs: None,
|
|
background_tasks: None,
|
|
}
|
|
}
|
|
|
|
/// Wires the best-effort conversation-summary sources (Lot C): the handoff
|
|
/// store (primary) and the conversation log (bounded fallback), both resolved
|
|
/// per project root. Builder so existing call sites/tests stay unchanged.
|
|
#[must_use]
|
|
pub fn with_conversation_sources(
|
|
mut self,
|
|
handoffs: Arc<dyn HandoffProvider>,
|
|
logs: Arc<dyn ConversationLogProvider>,
|
|
) -> Self {
|
|
self.handoffs = Some(handoffs);
|
|
self.logs = Some(logs);
|
|
self
|
|
}
|
|
|
|
/// Wires the best-effort background-task source for the Work panel. Builder so
|
|
/// existing call sites/tests stay unchanged; absent or failing store ⇒ empty
|
|
/// `background_tasks` without affecting live/busy/tickets.
|
|
#[must_use]
|
|
pub fn with_background_tasks(mut self, store: Arc<dyn BackgroundTaskStore>) -> Self {
|
|
self.background_tasks = Some(store);
|
|
self
|
|
}
|
|
|
|
/// Executes the read-only aggregation.
|
|
///
|
|
/// # Errors
|
|
/// - [`AppError::Store`] when the manifest cannot be loaded,
|
|
/// - [`AppError::Invalid`] if a manifest entry violates agent invariants.
|
|
pub async fn execute(
|
|
&self,
|
|
input: GetProjectWorkStateInput,
|
|
) -> Result<ProjectWorkState, AppError> {
|
|
let manifest = self.contexts.load_manifest(&input.project).await?;
|
|
let live_by_agent = live_by_agent(
|
|
self.live
|
|
.live_agent_snapshots()
|
|
.into_iter()
|
|
.filter(|snapshot| snapshot.project_id == input.project.id),
|
|
);
|
|
let undelivered_completions = match &self.background_tasks {
|
|
Some(store) => Some(store.list_undelivered_completions().await),
|
|
None => None,
|
|
};
|
|
|
|
let mut agents = Vec::with_capacity(manifest.entries.len());
|
|
for entry in &manifest.entries {
|
|
let agent = entry
|
|
.to_agent()
|
|
.map_err(|err| AppError::Invalid(err.to_string()))?;
|
|
let live = live_by_agent
|
|
.get(&agent.id)
|
|
.map(|snapshot| LiveWorkSession {
|
|
node_id: snapshot.node_id,
|
|
session_id: snapshot.session_id,
|
|
kind: snapshot.kind,
|
|
});
|
|
let runtime_state = self
|
|
.live
|
|
.runtime_status_for(input.project.id, &agent.id)
|
|
.map(|status| status.state);
|
|
// 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 runtime_key = domain::RuntimeAgentKey::new(input.project.id, agent.id);
|
|
let busy = self.input.busy_state(runtime_key);
|
|
let busy_ticket = busy.ticket();
|
|
let tickets = self
|
|
.queue
|
|
.queue_for(runtime_key)
|
|
.into_iter()
|
|
.map(|snapshot| ticket_state(snapshot, busy_ticket))
|
|
.collect();
|
|
let background_tasks = self
|
|
.background_tasks_for_agent(&input.project, agent.id, &undelivered_completions)
|
|
.await;
|
|
agents.push(AgentWorkState {
|
|
agent_id: agent.id,
|
|
name: agent.name,
|
|
profile_id: agent.profile_id,
|
|
live,
|
|
runtime_state,
|
|
busy,
|
|
tickets,
|
|
background_tasks,
|
|
});
|
|
}
|
|
|
|
// Lot C — best-effort conversation summaries. Only the conversations
|
|
// visible through the projected tickets are summarised, deduplicated in
|
|
// first-seen order. A preview failure degrades the per-conversation status
|
|
// (never an `AppError`): `agents`/live/busy/tickets stay intact above.
|
|
let conversations = self
|
|
.summarise_conversations(&input.project.root, &agents)
|
|
.await;
|
|
|
|
Ok(ProjectWorkState {
|
|
agents,
|
|
conversations,
|
|
})
|
|
}
|
|
|
|
/// Derives the best-effort summaries of the conversations referenced by the
|
|
/// projected tickets. Resolves the per-root handoff/log stores once, then folds
|
|
/// each distinct conversation id through [`conversation_summary`].
|
|
async fn summarise_conversations(
|
|
&self,
|
|
root: &ProjectPath,
|
|
agents: &[AgentWorkState],
|
|
) -> Vec<ConversationWorkSummary> {
|
|
let conversation_ids = distinct_conversation_ids(agents);
|
|
if conversation_ids.is_empty() {
|
|
return Vec::new();
|
|
}
|
|
let handoff_store = self
|
|
.handoffs
|
|
.as_ref()
|
|
.and_then(|provider| provider.handoff_store_for(root));
|
|
let log_store = self
|
|
.logs
|
|
.as_ref()
|
|
.and_then(|provider| provider.conversation_log_for(root));
|
|
|
|
let mut summaries = Vec::with_capacity(conversation_ids.len());
|
|
for conversation in conversation_ids {
|
|
summaries.push(conversation_summary(conversation, &handoff_store, &log_store).await);
|
|
}
|
|
summaries
|
|
}
|
|
|
|
async fn background_tasks_for_agent(
|
|
&self,
|
|
project: &Project,
|
|
agent_id: AgentId,
|
|
undelivered_completions: &Option<Result<Vec<BackgroundTask>, BackgroundTaskPortError>>,
|
|
) -> Vec<AgentBackgroundTaskState> {
|
|
let Some(store) = &self.background_tasks else {
|
|
return Vec::new();
|
|
};
|
|
let Some(Ok(undelivered_completions)) = undelivered_completions.as_ref() else {
|
|
return Vec::new();
|
|
};
|
|
let Ok(open) = store.list_open_for_agent(agent_id).await else {
|
|
return Vec::new();
|
|
};
|
|
|
|
let mut by_id: HashMap<TaskId, BackgroundTask> = HashMap::new();
|
|
for task in open {
|
|
by_id.insert(task.id, task);
|
|
}
|
|
for task in undelivered_completions
|
|
.iter()
|
|
.filter(|task| task.owner_agent_id == agent_id)
|
|
{
|
|
by_id.entry(task.id).or_insert_with(|| task.clone());
|
|
}
|
|
|
|
let mut tasks: Vec<_> = by_id
|
|
.into_values()
|
|
.filter(|task| task.project_id == project.id)
|
|
.map(AgentBackgroundTaskState::from)
|
|
.collect();
|
|
tasks.sort_by_key(|task| (task.created_at_ms, task.task_id));
|
|
tasks
|
|
}
|
|
}
|
|
|
|
impl From<BackgroundTask> for AgentBackgroundTaskState {
|
|
fn from(task: BackgroundTask) -> Self {
|
|
let (exit_code, summary, stdout_tail, stderr_tail) = flatten_background_result(&task);
|
|
let (requester_agent_id, target_agent_id, conversation_id) =
|
|
flatten_background_rendezvous_context(&task.kind);
|
|
Self {
|
|
task_id: task.id,
|
|
parent_task_id: task.parent_task_id,
|
|
kind: BackgroundTaskKindLabel::from(&task.kind),
|
|
state: task.state,
|
|
exit_code,
|
|
summary,
|
|
stdout_tail,
|
|
stderr_tail,
|
|
requester_agent_id,
|
|
target_agent_id,
|
|
conversation_id,
|
|
created_at_ms: task.created_at_ms,
|
|
updated_at_ms: task.updated_at_ms,
|
|
}
|
|
}
|
|
}
|
|
|
|
fn flatten_background_rendezvous_context(
|
|
kind: &BackgroundTaskKind,
|
|
) -> (
|
|
Option<AgentId>,
|
|
Option<AgentId>,
|
|
Option<domain::ConversationId>,
|
|
) {
|
|
match kind {
|
|
BackgroundTaskKind::HeadlessRendezvous {
|
|
requester_agent_id,
|
|
target_agent_id,
|
|
conversation_id,
|
|
..
|
|
} => (
|
|
*requester_agent_id,
|
|
Some(*target_agent_id),
|
|
Some(*conversation_id),
|
|
),
|
|
_ => (None, None, None),
|
|
}
|
|
}
|
|
|
|
impl From<&BackgroundTaskKind> for BackgroundTaskKindLabel {
|
|
fn from(kind: &BackgroundTaskKind) -> Self {
|
|
match kind {
|
|
BackgroundTaskKind::Command { .. } => Self::Command,
|
|
BackgroundTaskKind::HeadlessRendezvous { .. } => Self::HeadlessRendezvous,
|
|
BackgroundTaskKind::SessionResume { .. } => Self::SessionResume,
|
|
BackgroundTaskKind::Maintenance { .. } => Self::Maintenance,
|
|
}
|
|
}
|
|
}
|
|
|
|
fn flatten_background_result(
|
|
task: &BackgroundTask,
|
|
) -> (Option<i32>, Option<String>, Option<String>, Option<String>) {
|
|
match &task.result {
|
|
Some(BackgroundTaskResult::Success {
|
|
exit_code,
|
|
summary,
|
|
stdout_tail,
|
|
stderr_tail,
|
|
..
|
|
}) => (
|
|
*exit_code,
|
|
Some(summary.clone()),
|
|
stdout_tail.clone(),
|
|
stderr_tail.clone(),
|
|
),
|
|
Some(BackgroundTaskResult::Failure {
|
|
exit_code,
|
|
error,
|
|
stdout_tail,
|
|
stderr_tail,
|
|
..
|
|
}) => (
|
|
*exit_code,
|
|
Some(error.clone()),
|
|
stdout_tail.clone(),
|
|
stderr_tail.clone(),
|
|
),
|
|
Some(BackgroundTaskResult::Cancelled { reason, .. })
|
|
| Some(BackgroundTaskResult::Expired { reason, .. }) => {
|
|
(None, Some(reason.clone()), None, None)
|
|
}
|
|
Some(BackgroundTaskResult::CancelFailed { error, .. }) => {
|
|
(None, Some(error.clone()), None, None)
|
|
}
|
|
None => (None, None, None, None),
|
|
}
|
|
}
|
|
|
|
/// Collects the conversation ids referenced by the agents' tickets, deduplicated in
|
|
/// first-seen (FIFO) order so the summary list mirrors the panel's ordering.
|
|
fn distinct_conversation_ids(agents: &[AgentWorkState]) -> Vec<ConversationId> {
|
|
let mut seen = HashSet::new();
|
|
let mut ids = Vec::new();
|
|
for agent in agents {
|
|
for ticket in &agent.tickets {
|
|
if seen.insert(ticket.conversation_id) {
|
|
ids.push(ticket.conversation_id);
|
|
}
|
|
}
|
|
}
|
|
ids
|
|
}
|
|
|
|
/// Derives one conversation summary, best-effort, following the Lot C algorithm:
|
|
/// handoff present ⇒ `Ready`; absent ⇒ `Missing` (+ log fallback); unreadable but
|
|
/// log readable ⇒ `Partial`; both KO ⇒ `Unavailable`. Never returns an error.
|
|
async fn conversation_summary(
|
|
conversation: ConversationId,
|
|
handoff_store: &Option<Arc<dyn HandoffStore>>,
|
|
log_store: &Option<Arc<dyn ConversationLog>>,
|
|
) -> ConversationWorkSummary {
|
|
// A handoff store absent from the wiring is treated as "no handoff yet"
|
|
// (`Missing` path), never as an error: the read model stays best-effort.
|
|
let handoff = match handoff_store {
|
|
Some(store) => Some(store.load(conversation).await),
|
|
None => None,
|
|
};
|
|
match handoff {
|
|
Some(Ok(Some(handoff))) => ready_summary(conversation, &handoff),
|
|
Some(Ok(None)) | None => {
|
|
// Handoff legitimately absent: status stays Missing even if the log
|
|
// fallback is itself unreadable (then simply no recent turns).
|
|
let recent_turns = recent_turns(log_store, conversation)
|
|
.await
|
|
.unwrap_or_default();
|
|
ConversationWorkSummary {
|
|
conversation_id: conversation,
|
|
status: ConversationPreviewStatus::Missing,
|
|
objective_preview: None,
|
|
summary_preview: None,
|
|
summary_len: 0,
|
|
up_to: None,
|
|
recent_turns,
|
|
}
|
|
}
|
|
Some(Err(_)) => match recent_turns(log_store, conversation).await {
|
|
// Handoff unreadable but the log answered: partial view from the turns.
|
|
Some(recent_turns) => ConversationWorkSummary {
|
|
conversation_id: conversation,
|
|
status: ConversationPreviewStatus::Partial,
|
|
objective_preview: None,
|
|
summary_preview: None,
|
|
summary_len: 0,
|
|
up_to: None,
|
|
recent_turns,
|
|
},
|
|
// Both sources failed: nothing usable to show.
|
|
None => ConversationWorkSummary {
|
|
conversation_id: conversation,
|
|
status: ConversationPreviewStatus::Unavailable,
|
|
objective_preview: None,
|
|
summary_preview: None,
|
|
summary_len: 0,
|
|
up_to: None,
|
|
recent_turns: Vec::new(),
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Projects a readable [`Handoff`] into a `Ready` summary with bounded previews.
|
|
fn ready_summary(conversation: ConversationId, handoff: &Handoff) -> ConversationWorkSummary {
|
|
ConversationWorkSummary {
|
|
conversation_id: conversation,
|
|
status: ConversationPreviewStatus::Ready,
|
|
objective_preview: handoff
|
|
.objective
|
|
.as_deref()
|
|
.map(|objective| preview(objective, OBJECTIVE_PREVIEW_MAX_CHARS)),
|
|
summary_preview: Some(preview(&handoff.summary_md, HANDOFF_PREVIEW_MAX_CHARS)),
|
|
summary_len: handoff.summary_md.chars().count(),
|
|
up_to: Some(handoff.up_to),
|
|
recent_turns: Vec::new(),
|
|
}
|
|
}
|
|
|
|
/// Reads the bounded recent-turns fallback for `conversation`.
|
|
///
|
|
/// Returns `Some(turns)` (possibly empty) when a log store is wired and answers,
|
|
/// `None` when there is no log store or the read failed — the caller uses that
|
|
/// distinction to separate `Partial` from `Unavailable`. Never reads the full thread
|
|
/// (only [`ConversationLog::last`]) and re-caps defensively to [`RECENT_TURNS_MAX`].
|
|
async fn recent_turns(
|
|
log_store: &Option<Arc<dyn ConversationLog>>,
|
|
conversation: ConversationId,
|
|
) -> Option<Vec<ConversationTurnWorkPreview>> {
|
|
let store = log_store.as_ref()?;
|
|
match store.last(conversation, RECENT_TURNS_MAX).await {
|
|
Ok(turns) => Some(
|
|
turns
|
|
.into_iter()
|
|
.take(RECENT_TURNS_MAX)
|
|
.map(turn_preview)
|
|
.collect(),
|
|
),
|
|
Err(_) => None,
|
|
}
|
|
}
|
|
|
|
/// Projects one [`ConversationTurn`] into its bounded read-model preview.
|
|
fn turn_preview(turn: ConversationTurn) -> ConversationTurnWorkPreview {
|
|
ConversationTurnWorkPreview {
|
|
role: turn.role,
|
|
source: turn.source.into(),
|
|
at_ms: turn.at_ms,
|
|
text_preview: preview(&turn.text, TURN_PREVIEW_MAX_CHARS),
|
|
text_len: turn.text.chars().count(),
|
|
}
|
|
}
|
|
|
|
/// 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 the bounded task excerpt for the UI panel (see [`preview`]).
|
|
///
|
|
/// The original length is reported separately as [`AgentTicketState::task_len`].
|
|
fn task_preview(task: &str) -> String {
|
|
preview(task, TASK_PREVIEW_MAX_CHARS)
|
|
}
|
|
|
|
/// Builds a bounded, whitespace-normalised excerpt of `text` for the UI.
|
|
///
|
|
/// Trims, collapses any run of whitespace to a single space, then truncates to
|
|
/// `max_chars` characters (never mid-codepoint). The original length is reported
|
|
/// separately by each caller's `*_len` field.
|
|
fn preview(text: &str, max_chars: usize) -> String {
|
|
let normalised = text.split_whitespace().collect::<Vec<_>>().join(" ");
|
|
if normalised.chars().count() <= max_chars {
|
|
normalised
|
|
} else {
|
|
normalised.chars().take(max_chars).collect()
|
|
}
|
|
}
|
|
|
|
fn live_by_agent(
|
|
snapshots: impl IntoIterator<Item = LiveSessionSnapshot>,
|
|
) -> HashMap<AgentId, LiveSessionSnapshot> {
|
|
let mut out = HashMap::new();
|
|
for snapshot in snapshots {
|
|
out.entry(snapshot.agent_id).or_insert(snapshot);
|
|
}
|
|
out
|
|
}
|