diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 1d330ca..c8415f2 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -15,13 +15,27 @@ pub struct ProjectionUpdate { pub state: ReducedState, } +/// A prompt lifecycle transition supplied by the prompt transport (#138). +/// +/// ACP notifications do not delimit turns, so replaying them must not infer +/// an in-flight prompt from an agent chunk. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TurnLifecycle { + Begin, + Finish, + Cancel, + Failure, +} + /// The one-pass ACP fold used for both persisted replay and live events. #[derive(Debug, Default)] pub struct Projection { rows: Vec, row_index: HashMap, tools: HashMap, + pending_tool_updates: HashMap>, state: ReducedState, + turn_active: bool, } impl Projection { @@ -32,13 +46,11 @@ impl Projection { self.content(seq, chunk, TranscriptRowKind::User, "user") } SessionUpdate::AgentMessageChunk(chunk) => { - self.state.turn.responding = true; - self.state.turn.thinking = false; + self.agent_text(); self.content(seq, chunk, TranscriptRowKind::AgentText, "agent") } SessionUpdate::AgentThoughtChunk(chunk) => { - self.state.turn.responding = true; - self.state.turn.thinking = true; + self.agent_thought(); self.content(seq, chunk, TranscriptRowKind::AgentThought, "thought") } SessionUpdate::ToolCall(tool) => self.tool_call(tool), @@ -46,7 +58,7 @@ impl Projection { SessionUpdate::Plan(plan) => { let data = to_value(plan); self.state.plan = Some(data.clone()); - Some(self.append(format!("plan:{seq}"), TranscriptRowKind::Plan, data)) + Some(self.replace("plan".into(), TranscriptRowKind::Plan, data)) } SessionUpdate::UsageUpdate(usage) => { let value = to_value(usage); @@ -91,6 +103,26 @@ impl Projection { self.state.clone() } + /// Apply a transport-owned prompt lifecycle transition. + pub fn turn_lifecycle(&mut self, lifecycle: TurnLifecycle) -> ProjectionUpdate { + match lifecycle { + TurnLifecycle::Begin => { + self.turn_active = true; + self.state.turn.responding = true; + self.state.turn.thinking = true; + } + TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure => { + self.turn_active = false; + self.state.turn.responding = false; + self.state.turn.thinking = false; + } + } + ProjectionUpdate { + transcript: None, + state: self.state(), + } + } + fn content( &mut self, seq: u64, @@ -114,15 +146,29 @@ impl Projection { fn tool_call(&mut self, tool: &ToolCall) -> Option { let id = format!("tool:{}", tool.tool_call_id); + let tool_id = tool.tool_call_id.to_string(); + let mut canonical = tool.clone(); + if let Some(updates) = self.pending_tool_updates.get(&tool_id) { + for fields in updates { + canonical.update(fields.clone()); + } + } let kind = if tool_has_diff(tool) { TranscriptRowKind::ToolDiff } else { TranscriptRowKind::ToolCall }; - self.tools - .insert(tool.tool_call_id.to_string(), tool.clone()); + self.tools.insert(tool_id, canonical.clone()); self.sync_active_tools(); - Some(self.append(id, kind, to_value(tool))) + Some(self.replace( + id, + if tool_has_diff(&canonical) { + TranscriptRowKind::ToolDiff + } else { + kind + }, + to_value(&canonical), + )) } fn tool_update( @@ -131,13 +177,17 @@ impl Projection { ) -> Option { let tool_id = update.tool_call_id.to_string(); let Some(tool) = self.tools.get_mut(&tool_id) else { - return Some(self.append( - format!("notice:tool-update:{tool_id}"), - TranscriptRowKind::Notice, - to_value(update), - )); + self.pending_tool_updates + .entry(tool_id) + .or_default() + .push(update.fields.clone()); + return None; }; tool.update(update.fields.clone()); + self.pending_tool_updates + .entry(tool_id.clone()) + .or_default() + .push(update.fields.clone()); let row_id = format!("tool:{tool_id}"); let index = *self .row_index @@ -164,6 +214,30 @@ impl Projection { TranscriptDelta::Append { row } } + fn replace(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta { + let Some(index) = self.row_index.get(&id).copied() else { + return self.append(id, kind, data); + }; + let row = &mut self.rows[index]; + row.kind = kind; + row.data = data; + TranscriptDelta::Patch { row: row.clone() } + } + + fn agent_text(&mut self) { + if self.turn_active { + self.state.turn.responding = true; + self.state.turn.thinking = false; + } + } + + fn agent_thought(&mut self) { + if self.turn_active { + self.state.turn.responding = true; + self.state.turn.thinking = true; + } + } + fn sync_active_tools(&mut self) { self.state.turn.active_tool_calls = self .tools @@ -226,7 +300,7 @@ mod tests { use maestro_acp::schema::SessionNotification; use maestro_protocol::{TranscriptDelta, TranscriptRowKind}; - use super::Projection; + use super::{Projection, TurnLifecycle}; fn event(value: serde_json::Value) -> SessionNotification { serde_json::from_value(serde_json::json!({ @@ -313,6 +387,57 @@ mod tests { assert_eq!(projection.snapshot(5).rows.len(), 1); } + #[test] + fn duplicate_tool_calls_patch_one_stable_row() { + let mut projection = Projection::default(); + projection.apply( + 1, + &event(serde_json::json!({ + "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run" + })), + ); + let duplicate = projection.apply( + 2, + &event(serde_json::json!({ + "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run again" + })), + ); + match duplicate.transcript.unwrap() { + TranscriptDelta::Patch { row } => { + assert_eq!(row.id, "tool:bash-7"); + assert_eq!(row.data["title"], "Run again"); + } + other => panic!("expected patch, got {other:?}"), + } + assert_eq!(projection.snapshot(2).rows.len(), 1); + } + + #[test] + fn early_tool_update_is_reconciled_when_the_call_arrives() { + let mut projection = Projection::default(); + let early = projection.apply( + 1, + &event(serde_json::json!({ + "sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed" + })), + ); + assert_eq!(early.transcript, None); + let call = projection.apply( + 2, + &event(serde_json::json!({ + "sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run", "status":"pending" + })), + ); + match call.transcript.unwrap() { + TranscriptDelta::Append { row } => { + assert_eq!(row.id, "tool:bash-7"); + assert_eq!(row.data["status"], "completed"); + } + other => panic!("expected append, got {other:?}"), + } + assert!(projection.state().turn.active_tool_calls.is_empty()); + } + #[test] fn unknown_tool_stays_generic_and_keeps_its_raw_name() { let mut projection = Projection::default(); @@ -356,4 +481,69 @@ mod tests { serde_json::json!([]) ); } + + #[test] + fn plans_replace_one_stable_row_and_state() { + let mut projection = Projection::default(); + let first = projection.apply( + 1, + &event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"first","priority":"high","status":"pending"}]})), + ); + assert!(matches!( + first.transcript, + Some(TranscriptDelta::Append { .. }) + )); + let replacement = projection.apply( + 2, + &event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"second","priority":"high","status":"completed"}]})), + ); + match replacement.transcript.unwrap() { + TranscriptDelta::Patch { row } => { + assert_eq!(row.id, "plan"); + assert_eq!(row.data["entries"][0]["content"], "second"); + } + other => panic!("expected patch, got {other:?}"), + } + assert_eq!(projection.snapshot(2).rows.len(), 1); + assert_eq!( + projection.state().plan.unwrap()["entries"][0]["content"], + "second" + ); + } + + #[test] + fn turn_lifecycle_gates_chunks_and_normalizes_terminal_states() { + let mut projection = Projection::default(); + projection.apply( + 1, + &event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"replay"}})), + ); + assert_eq!(projection.state().turn, Default::default()); + + projection.turn_lifecycle(TurnLifecycle::Begin); + assert!(projection.state().turn.responding); + assert!(projection.state().turn.thinking); + projection.apply( + 2, + &event(serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}})), + ); + assert!(projection.state().turn.thinking); + projection.apply( + 3, + &event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"answer"}})), + ); + assert!(projection.state().turn.responding); + assert!(!projection.state().turn.thinking); + + for terminal in [ + TurnLifecycle::Finish, + TurnLifecycle::Cancel, + TurnLifecycle::Failure, + ] { + projection.turn_lifecycle(TurnLifecycle::Begin); + projection.turn_lifecycle(terminal); + assert!(!projection.state().turn.responding); + assert!(!projection.state().turn.thinking); + } + } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 6f93e94..bb6f2e3 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -21,7 +21,7 @@ use tokio::sync::{broadcast, watch, Mutex}; use uuid::Uuid; use crate::panes::Panes; -use crate::projection::{Projection, ProjectionUpdate}; +use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle}; use crate::pty::{pipe_pane_command, PtyBridge}; use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore}; use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord}; @@ -396,6 +396,27 @@ impl SessionManager { ))) } + /// Publish a prompt lifecycle transition without adding a synthetic ACP + /// event. The prompt route will call this seam in #138. + pub async fn apply_acp_turn_lifecycle( + &self, + id: Uuid, + lifecycle: TurnLifecycle, + ) -> Result { + let stream = { + let sessions = self.inner.sessions.lock().await; + let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; + record.acp_events.clone() + }; + let Some(stream) = stream else { + return Ok(false); + }; + let mut stream = stream.lock().await; + let projected = stream.projection.turn_lifecycle(lifecycle); + let _ = stream.projected_live.send(projected); + Ok(true) + } + /// End every adapter child before the daemon exits (#134): each gets /// `session/cancel`, then SIGTERM, concurrently. The sessions stay in /// the map, marked Dead and written through to their sidecars, so what