naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(daemon): reconcile transcript projections

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 01:26:12 +01:00
parent 046c110460
commit fd3ea64949
2 changed files with 226 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/projection.rs
+204 -14
View File
@@ -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<TranscriptRow>,
row_index: HashMap<String, usize>,
tools: HashMap<String, ToolCall>,
pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
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<TranscriptDelta> {
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<TranscriptDelta> {
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);
}
}
}
crates/maestro-daemon/src/session.rs
+22 -1
View File
@@ -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<bool, SessionError> {
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