1107 lines
40 KiB
Rust
1107 lines
40 KiB
Rust
//! Fold persisted ACP notifications into client-neutral transcript and state.
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall};
|
|
use maestro_protocol::{
|
|
ActivityState, ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind,
|
|
TranscriptSnapshot,
|
|
};
|
|
use serde_json::Value;
|
|
|
|
/// A transcript delta paired with the state after the same ACP notification.
|
|
#[derive(Debug, Clone, PartialEq)]
|
|
pub struct ProjectionUpdate {
|
|
pub transcript: Option<TranscriptDelta>,
|
|
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,
|
|
active_turn: Option<u64>,
|
|
active_turn_started_at: Option<DateTime<Utc>>,
|
|
active_turn_context_tokens: Option<u64>,
|
|
turn_failed: bool,
|
|
next_turn: u64,
|
|
turn_tool_calls: HashSet<String>,
|
|
anonymous_content_row: Option<(String, String)>,
|
|
}
|
|
|
|
impl Projection {
|
|
/// Fold one durable event and return its client-visible changes.
|
|
pub fn apply(&mut self, seq: u64, notification: &SessionNotification) -> ProjectionUpdate {
|
|
let transcript = match ¬ification.update {
|
|
SessionUpdate::UserMessageChunk(chunk) => {
|
|
self.content(seq, chunk, TranscriptRowKind::User, "user")
|
|
}
|
|
SessionUpdate::AgentMessageChunk(chunk) => {
|
|
self.agent_text();
|
|
self.content(seq, chunk, TranscriptRowKind::AgentText, "agent")
|
|
}
|
|
SessionUpdate::AgentThoughtChunk(chunk) => {
|
|
self.agent_thought();
|
|
self.content(seq, chunk, TranscriptRowKind::AgentThought, "thought")
|
|
}
|
|
SessionUpdate::ToolCall(tool) => self.tool_call(tool),
|
|
SessionUpdate::ToolCallUpdate(update) => self.tool_update(update),
|
|
SessionUpdate::Plan(plan) => {
|
|
self.anonymous_content_row = None;
|
|
let data = to_value(plan);
|
|
self.state.plan = Some(data.clone());
|
|
Some(self.replace("plan".into(), TranscriptRowKind::Plan, data))
|
|
}
|
|
SessionUpdate::UsageUpdate(usage) => {
|
|
let value = to_value(usage);
|
|
self.state.usage = Some(value.clone());
|
|
self.state.context = Some(value);
|
|
None
|
|
}
|
|
SessionUpdate::CurrentModeUpdate(mode) => {
|
|
self.state.mode = Some(mode.current_mode_id.to_string());
|
|
None
|
|
}
|
|
SessionUpdate::ConfigOptionUpdate(options) => {
|
|
let value = to_value(options);
|
|
self.state.model = model_from_options(&value).or_else(|| self.state.model.clone());
|
|
self.state.config_options = Some(value);
|
|
None
|
|
}
|
|
// ACP deliberately adds notification variants over time. Known but
|
|
// non-transcript updates become a generic notice instead of being
|
|
// dropped or forcing clients to understand ACP's enum.
|
|
_ => {
|
|
self.anonymous_content_row = None;
|
|
Some(self.append(
|
|
format!("notice:{seq}"),
|
|
TranscriptRowKind::Notice,
|
|
to_value(¬ification.update),
|
|
))
|
|
}
|
|
};
|
|
ProjectionUpdate {
|
|
transcript,
|
|
state: self.state.clone(),
|
|
}
|
|
}
|
|
|
|
/// The current transcript at `cursor`.
|
|
pub fn snapshot(&self, cursor: u64) -> TranscriptSnapshot {
|
|
TranscriptSnapshot {
|
|
cursor,
|
|
rows: self.rows.clone(),
|
|
}
|
|
}
|
|
|
|
/// The state folded over every event seen so far.
|
|
pub fn state(&self) -> ReducedState {
|
|
self.state.clone()
|
|
}
|
|
|
|
pub fn turn_active(&self) -> bool {
|
|
self.active_turn.is_some()
|
|
}
|
|
|
|
/// Sidebar activity derived from the live turn projection.
|
|
pub fn activity(&self) -> ActivityState {
|
|
crate::activity::decide_structured(
|
|
self.active_turn.is_some(),
|
|
self.state.pending_approval.is_some(),
|
|
self.turn_failed,
|
|
)
|
|
}
|
|
|
|
/// Start a prompt turn and return its monotonically increasing token.
|
|
pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) {
|
|
self.anonymous_content_row = None;
|
|
self.next_turn += 1;
|
|
let turn = self.next_turn;
|
|
self.active_turn = Some(turn);
|
|
let started_at = Utc::now();
|
|
let started_context_tokens = usage_tokens(self.state.usage.as_ref());
|
|
self.active_turn_started_at = Some(started_at);
|
|
self.active_turn_context_tokens = started_context_tokens;
|
|
self.turn_failed = false;
|
|
self.state.turn.responding = true;
|
|
self.state.turn.thinking = true;
|
|
self.state.turn.started_at = Some(started_at);
|
|
self.state.turn.started_context_tokens = started_context_tokens;
|
|
self.turn_tool_calls.clear();
|
|
self.state.turn.active_tool_calls.clear();
|
|
(
|
|
turn,
|
|
ProjectionUpdate {
|
|
transcript: None,
|
|
state: self.state(),
|
|
},
|
|
)
|
|
}
|
|
|
|
/// End `turn` only when it remains the active prompt turn.
|
|
pub fn turn_lifecycle(
|
|
&mut self,
|
|
lifecycle: TurnLifecycle,
|
|
turn: u64,
|
|
) -> Option<ProjectionUpdate> {
|
|
if !matches!(
|
|
lifecycle,
|
|
TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure
|
|
) || self.active_turn != Some(turn)
|
|
{
|
|
return None;
|
|
}
|
|
self.active_turn = None;
|
|
self.turn_failed = lifecycle == TurnLifecycle::Failure;
|
|
self.state.turn.responding = false;
|
|
self.state.turn.thinking = false;
|
|
let finished_at = Utc::now();
|
|
let started_at = self.active_turn_started_at.take();
|
|
let started_context_tokens = self.active_turn_context_tokens.take();
|
|
self.state.turn.started_at = None;
|
|
self.state.turn.started_context_tokens = None;
|
|
self.turn_tool_calls.clear();
|
|
self.state.turn.active_tool_calls.clear();
|
|
let context_tokens = usage_tokens(self.state.usage.as_ref());
|
|
let context_delta = context_tokens
|
|
.zip(started_context_tokens)
|
|
.map(|(end, start)| i128::from(end) - i128::from(start));
|
|
let elapsed_ms = started_at.map_or(0, |started| {
|
|
finished_at
|
|
.signed_duration_since(started)
|
|
.num_milliseconds()
|
|
.max(0)
|
|
});
|
|
let outcome = match lifecycle {
|
|
TurnLifecycle::Finish => "completed",
|
|
TurnLifecycle::Cancel => "cancelled",
|
|
TurnLifecycle::Failure => "failed",
|
|
TurnLifecycle::Begin => unreachable!(),
|
|
};
|
|
let row = self.append(
|
|
format!("turn_meta:{}", uuid::Uuid::new_v4()),
|
|
TranscriptRowKind::Notice,
|
|
serde_json::json!({
|
|
"maestroTurnMeta": {
|
|
"elapsedMs": elapsed_ms,
|
|
"contextTokens": context_tokens,
|
|
"contextDelta": context_delta,
|
|
"outcome": outcome,
|
|
"model": self.state.model,
|
|
}
|
|
}),
|
|
);
|
|
Some(ProjectionUpdate {
|
|
transcript: Some(row),
|
|
state: self.state(),
|
|
})
|
|
}
|
|
|
|
pub fn active_turn(&self) -> Option<u64> {
|
|
self.active_turn
|
|
}
|
|
|
|
/// Add a daemon-originated permission row. The nonce belongs only in the
|
|
/// reduced state, which is the client control surface, never in ACP data.
|
|
pub fn request_approval(&mut self, nonce: String, request: Value) -> ProjectionUpdate {
|
|
let row = self.append(
|
|
format!("approval:{}", uuid::Uuid::new_v4()),
|
|
TranscriptRowKind::Approval,
|
|
request.clone(),
|
|
);
|
|
self.state.pending_approval = Some(serde_json::json!({
|
|
"nonce": nonce,
|
|
"request": request,
|
|
}));
|
|
ProjectionUpdate {
|
|
transcript: Some(row),
|
|
state: self.state(),
|
|
}
|
|
}
|
|
|
|
pub fn restore_row(&mut self, row: TranscriptRow) -> ProjectionUpdate {
|
|
self.anonymous_content_row = None;
|
|
let transcript = if !self.row_index.contains_key(&row.id) {
|
|
self.row_index.insert(row.id.clone(), self.rows.len());
|
|
self.rows.push(row.clone());
|
|
Some(TranscriptDelta::Append { row })
|
|
} else {
|
|
None
|
|
};
|
|
ProjectionUpdate {
|
|
transcript,
|
|
state: self.state(),
|
|
}
|
|
}
|
|
|
|
/// Clear an approval only when it is still the one the client answered.
|
|
pub fn resolve_approval(&mut self, nonce: &str) -> Option<ProjectionUpdate> {
|
|
let pending = self.state.pending_approval.as_ref()?;
|
|
if pending.get("nonce").and_then(Value::as_str) != Some(nonce) {
|
|
return None;
|
|
}
|
|
self.state.pending_approval = None;
|
|
Some(ProjectionUpdate {
|
|
transcript: None,
|
|
state: self.state(),
|
|
})
|
|
}
|
|
|
|
/// Terminal turn paths must not leave a recovered snapshot permanently
|
|
/// blocked on an ACP request whose transport no longer exists.
|
|
pub fn clear_approval(&mut self) -> Option<ProjectionUpdate> {
|
|
self.state.pending_approval.take()?;
|
|
Some(ProjectionUpdate {
|
|
transcript: None,
|
|
state: self.state(),
|
|
})
|
|
}
|
|
|
|
/// Undo an `append`ed row that a caller decides never happened — used
|
|
/// when persisting the row's own event fails (#184), so a snapshot taken
|
|
/// after the failure cannot show a row durable replay would never
|
|
/// reproduce. Only ever called for a row that was just appended and
|
|
/// never patched, so removing it by id fully reverses `append`.
|
|
pub fn rollback_row(&mut self, id: &str) {
|
|
let Some(index) = self.row_index.remove(id) else {
|
|
return;
|
|
};
|
|
self.rows.remove(index);
|
|
for (shifted_index, row) in self.rows.iter().enumerate().skip(index) {
|
|
self.row_index.insert(row.id.clone(), shifted_index);
|
|
}
|
|
}
|
|
|
|
fn content(
|
|
&mut self,
|
|
seq: u64,
|
|
chunk: &ContentChunk,
|
|
kind: TranscriptRowKind,
|
|
prefix: &str,
|
|
) -> Option<TranscriptDelta> {
|
|
let data = to_value(chunk);
|
|
let id = if let Some(message_id) = chunk.message_id.as_ref() {
|
|
self.anonymous_content_row = None;
|
|
format!("{prefix}:{message_id}")
|
|
} else {
|
|
let text = data.pointer("/content/text").and_then(Value::as_str);
|
|
let previous = self
|
|
.anonymous_content_row
|
|
.as_ref()
|
|
.filter(|(previous_prefix, _)| previous_prefix == prefix)
|
|
.map(|(_, id)| id.clone());
|
|
if text.is_some_and(|text| !text.is_empty()) {
|
|
if let Some(id) = previous {
|
|
let index = *self
|
|
.row_index
|
|
.get(&id)
|
|
.expect("anonymous content row remains addressable");
|
|
let row = &mut self.rows[index];
|
|
append_text(&mut row.data, &data);
|
|
return Some(TranscriptDelta::Patch { row: row.clone() });
|
|
}
|
|
}
|
|
let id = format!("{prefix}:{seq}");
|
|
self.anonymous_content_row = text.map(|_| (prefix.to_owned(), id.clone()));
|
|
id
|
|
};
|
|
if let Some(index) = self.row_index.get(&id).copied() {
|
|
let row = &mut self.rows[index];
|
|
append_text(&mut row.data, &data);
|
|
return Some(TranscriptDelta::Patch { row: row.clone() });
|
|
}
|
|
Some(self.append(id, kind, data))
|
|
}
|
|
|
|
fn tool_call(&mut self, tool: &ToolCall) -> Option<TranscriptDelta> {
|
|
self.anonymous_content_row = None;
|
|
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_id, canonical.clone());
|
|
self.track_tool_call(&canonical);
|
|
Some(self.replace(
|
|
id,
|
|
if tool_has_diff(&canonical) {
|
|
TranscriptRowKind::ToolDiff
|
|
} else {
|
|
kind
|
|
},
|
|
to_value(&canonical),
|
|
))
|
|
}
|
|
|
|
fn tool_update(
|
|
&mut self,
|
|
update: &maestro_acp::schema::ToolCallUpdate,
|
|
) -> Option<TranscriptDelta> {
|
|
self.anonymous_content_row = None;
|
|
let tool_id = update.tool_call_id.to_string();
|
|
let Some(tool) = self.tools.get_mut(&tool_id) else {
|
|
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
|
|
.get(&row_id)
|
|
.expect("a tracked tool call always has a transcript row");
|
|
let row = {
|
|
let row = &mut self.rows[index];
|
|
row.kind = if tool_has_diff(tool) {
|
|
TranscriptRowKind::ToolDiff
|
|
} else {
|
|
TranscriptRowKind::ToolCall
|
|
};
|
|
row.data = to_value(tool);
|
|
row.clone()
|
|
};
|
|
self.sync_active_tool(&tool_id);
|
|
Some(TranscriptDelta::Patch { row })
|
|
}
|
|
|
|
fn append(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta {
|
|
let row = TranscriptRow { id, kind, data };
|
|
self.row_index.insert(row.id.clone(), self.rows.len());
|
|
self.rows.push(row.clone());
|
|
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 track_tool_call(&mut self, tool: &ToolCall) {
|
|
if self.turn_active() {
|
|
self.turn_tool_calls.insert(tool.tool_call_id.to_string());
|
|
self.sync_active_tool(&tool.tool_call_id.to_string());
|
|
}
|
|
}
|
|
|
|
fn sync_active_tool(&mut self, tool_id: &str) {
|
|
if !self.turn_active() || !self.turn_tool_calls.contains(tool_id) {
|
|
return;
|
|
}
|
|
let active = self.tools.get(tool_id).is_some_and(|tool| {
|
|
matches!(
|
|
tool.status,
|
|
maestro_acp::schema::ToolCallStatus::Pending
|
|
| maestro_acp::schema::ToolCallStatus::InProgress
|
|
)
|
|
});
|
|
let active_tool_calls = &mut self.state.turn.active_tool_calls;
|
|
if active {
|
|
if !active_tool_calls.iter().any(|id| id == tool_id) {
|
|
active_tool_calls.push(tool_id.to_owned());
|
|
active_tool_calls.sort();
|
|
}
|
|
} else {
|
|
active_tool_calls.retain(|id| id != tool_id);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn usage_tokens(usage: Option<&Value>) -> Option<u64> {
|
|
usage?.get("used")?.as_u64()
|
|
}
|
|
|
|
fn to_value(value: &impl serde::Serialize) -> Value {
|
|
serde_json::to_value(value).expect("ACP schema values serialize")
|
|
}
|
|
|
|
fn tool_has_diff(tool: &ToolCall) -> bool {
|
|
tool.content
|
|
.iter()
|
|
.any(|content| matches!(content, maestro_acp::schema::ToolCallContent::Diff(_)))
|
|
}
|
|
|
|
fn append_text(existing: &mut Value, next: &Value) {
|
|
let Some(existing_text) = existing.pointer_mut("/content/text") else {
|
|
*existing = next.clone();
|
|
return;
|
|
};
|
|
let Some(next_text) = next.pointer("/content/text").and_then(Value::as_str) else {
|
|
*existing = next.clone();
|
|
return;
|
|
};
|
|
if let Some(text) = existing_text.as_str() {
|
|
*existing_text = Value::String(format!("{text}{next_text}"));
|
|
} else {
|
|
*existing = next.clone();
|
|
}
|
|
}
|
|
|
|
fn model_from_options(value: &Value) -> Option<String> {
|
|
value
|
|
.get("configOptions")?
|
|
.as_array()?
|
|
.iter()
|
|
.find(|option| {
|
|
option.get("id").and_then(Value::as_str) == Some("model")
|
|
|| option.get("category").and_then(Value::as_str) == Some("model")
|
|
})?
|
|
.get("currentValue")?
|
|
.as_str()
|
|
.map(str::to_owned)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use maestro_acp::schema::SessionNotification;
|
|
use maestro_protocol::{TranscriptDelta, TranscriptRow, TranscriptRowKind};
|
|
|
|
use super::{Projection, TurnLifecycle};
|
|
|
|
fn event(value: serde_json::Value) -> SessionNotification {
|
|
serde_json::from_value(serde_json::json!({
|
|
"sessionId": "acp-1",
|
|
"update": value,
|
|
}))
|
|
.unwrap()
|
|
}
|
|
|
|
#[test]
|
|
fn a_session_update_tag_the_acp_dependency_does_not_know_fails_to_deserialize() {
|
|
// `SessionUpdate` has no unknown-variant fallback, so this never reaches `Projection::apply`.
|
|
let result: Result<SessionNotification, _> = serde_json::from_value(serde_json::json!({
|
|
"sessionId": "acp-1",
|
|
"update": {"sessionUpdate": "some_future_update_variant"},
|
|
}));
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn folds_content_tool_diff_plan_and_notice_rows() {
|
|
let mut projection = Projection::default();
|
|
let events = [
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"hi"}}),
|
|
),
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}),
|
|
),
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}}),
|
|
),
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"read-1","title":"Read","kind":"read"}),
|
|
),
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"edit-1","title":"Edit","content":[{"type":"diff","path":"x.rs","oldText":"a","newText":"b"}]}),
|
|
),
|
|
event(serde_json::json!({"sessionUpdate":"plan","entries":[]})),
|
|
event(
|
|
serde_json::json!({"sessionUpdate":"available_commands_update","availableCommands":[]}),
|
|
),
|
|
];
|
|
|
|
let kinds: Vec<_> = events
|
|
.iter()
|
|
.enumerate()
|
|
.filter_map(|(i, event)| projection.apply((i + 1) as u64, event).transcript)
|
|
.map(|delta| match delta {
|
|
TranscriptDelta::Append { row } => row.kind,
|
|
other => panic!("expected append, got {other:?}"),
|
|
})
|
|
.collect();
|
|
assert_eq!(
|
|
kinds,
|
|
[
|
|
TranscriptRowKind::User,
|
|
TranscriptRowKind::AgentText,
|
|
TranscriptRowKind::AgentThought,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::ToolDiff,
|
|
TranscriptRowKind::Plan,
|
|
TranscriptRowKind::Notice,
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn projects_the_codex_adapter_fixture_without_provider_specific_rows() {
|
|
let fixture = include_str!("../tests/fixtures/codex-acp-session.jsonl");
|
|
let events = fixture
|
|
.lines()
|
|
.map(serde_json::from_str::<SessionNotification>)
|
|
.collect::<Result<Vec<_>, _>>()
|
|
.expect("Codex ACP fixture follows the supported schema");
|
|
let mut projection = Projection::default();
|
|
|
|
for (seq, event) in events.iter().enumerate() {
|
|
projection.apply((seq + 1) as u64, event);
|
|
}
|
|
|
|
let rows = projection.snapshot(events.len() as u64).rows;
|
|
assert_eq!(
|
|
rows.iter().map(|row| row.kind).collect::<Vec<_>>(),
|
|
[
|
|
TranscriptRowKind::User,
|
|
TranscriptRowKind::AgentThought,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::ToolDiff,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::ToolCall,
|
|
TranscriptRowKind::Plan,
|
|
TranscriptRowKind::Notice,
|
|
TranscriptRowKind::Notice,
|
|
TranscriptRowKind::AgentText,
|
|
]
|
|
);
|
|
assert_eq!(rows[2].data["rawInput"]["command"], "rg TODO");
|
|
assert_eq!(rows[2].data["rawOutput"]["exit_code"], 0);
|
|
assert_eq!(rows[4].data["_meta"]["is_mcp_tool_call"], true);
|
|
assert_eq!(rows[5].data["rawInput"]["query"], "Agent Client Protocol");
|
|
assert_eq!(
|
|
rows[7].data["content"][0]["content"]["mimeType"],
|
|
"image/png"
|
|
);
|
|
|
|
let state = projection.state();
|
|
assert_eq!(state.model.as_deref(), Some("gpt-5.3-codex"));
|
|
assert_eq!(state.mode.as_deref(), Some("agent"));
|
|
assert_eq!(state.usage.as_ref().unwrap()["used"], 1200);
|
|
assert_eq!(state.context.as_ref().unwrap()["size"], 258400);
|
|
assert_eq!(
|
|
state.config_options.as_ref().unwrap()["configOptions"][2]["currentValue"],
|
|
"high"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn coalesces_anonymous_stream_chunks_without_joining_turns() {
|
|
let mut projection = Projection::default();
|
|
let (first_turn, _) = projection.begin_turn();
|
|
for (seq, text) in [(1, ""), (2, "p"), (3, "ong")] {
|
|
projection.apply(
|
|
seq,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": text},
|
|
})),
|
|
);
|
|
}
|
|
projection.turn_lifecycle(TurnLifecycle::Finish, first_turn);
|
|
projection.begin_turn();
|
|
projection.apply(
|
|
4,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "next"},
|
|
})),
|
|
);
|
|
|
|
let rows = projection.snapshot(4).rows;
|
|
assert_eq!(rows.len(), 3);
|
|
assert_eq!(rows[0].data.pointer("/content/text").unwrap(), "pong");
|
|
assert!(rows[1].data.get("maestroTurnMeta").is_some());
|
|
assert_eq!(rows[2].data.pointer("/content/text").unwrap(), "next");
|
|
}
|
|
|
|
#[test]
|
|
fn anonymous_empty_chunk_delimits_turns_during_replay() {
|
|
let mut projection = Projection::default();
|
|
for (seq, text) in [(1, ""), (2, "visible"), (3, ""), (4, "p"), (5, "ong")] {
|
|
projection.apply(
|
|
seq,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": text},
|
|
})),
|
|
);
|
|
}
|
|
|
|
let rows = projection.snapshot(5).rows;
|
|
assert_eq!(rows.len(), 2);
|
|
assert_eq!(rows[0].data.pointer("/content/text").unwrap(), "visible");
|
|
assert_eq!(rows[1].data.pointer("/content/text").unwrap(), "pong");
|
|
}
|
|
|
|
#[test]
|
|
fn restored_user_prompt_delimits_anonymous_agent_rows() {
|
|
let mut projection = Projection::default();
|
|
projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "before"},
|
|
})),
|
|
);
|
|
projection.restore_row(TranscriptRow {
|
|
id: "user:prompt".into(),
|
|
kind: TranscriptRowKind::User,
|
|
data: serde_json::json!({"content": {"type": "text", "text": "next"}}),
|
|
});
|
|
projection.apply(
|
|
2,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "after"},
|
|
})),
|
|
);
|
|
|
|
let rows = projection.snapshot(2).rows;
|
|
assert_eq!(rows.len(), 3);
|
|
assert_eq!(rows[0].data.pointer("/content/text").unwrap(), "before");
|
|
assert_eq!(rows[1].kind, TranscriptRowKind::User);
|
|
assert_eq!(rows[2].data.pointer("/content/text").unwrap(), "after");
|
|
}
|
|
|
|
#[test]
|
|
fn tool_update_patches_the_original_stable_row() {
|
|
let mut projection = Projection::default();
|
|
let first = projection.apply(
|
|
4,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run"
|
|
})),
|
|
);
|
|
let updated = projection.apply(
|
|
5,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed"
|
|
})),
|
|
);
|
|
let id = match first.transcript.unwrap() {
|
|
TranscriptDelta::Append { row } => row.id,
|
|
other => panic!("expected append, got {other:?}"),
|
|
};
|
|
match updated.transcript.unwrap() {
|
|
TranscriptDelta::Patch { row } => {
|
|
assert_eq!(row.id, id);
|
|
assert_eq!(row.data["status"], "completed");
|
|
}
|
|
other => panic!("expected patch, got {other:?}"),
|
|
}
|
|
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();
|
|
let update = projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"tool_call", "toolCallId":"custom-1", "title":"Custom",
|
|
"name":"vendor.custom", "kind":"future_kind"
|
|
})),
|
|
);
|
|
match update.transcript.unwrap() {
|
|
TranscriptDelta::Append { row } => {
|
|
assert_eq!(row.kind, TranscriptRowKind::ToolCall);
|
|
assert_eq!(row.data["name"], "vendor.custom");
|
|
}
|
|
other => panic!("expected generic tool row, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn folds_usage_mode_model_and_plan_into_whole_state() {
|
|
let mut projection = Projection::default();
|
|
for (seq, update) in [
|
|
serde_json::json!({"sessionUpdate":"usage_update","used":12,"size":200}),
|
|
serde_json::json!({"sessionUpdate":"current_mode_update","currentModeId":"plan"}),
|
|
serde_json::json!({"sessionUpdate":"config_option_update","configOptions":[{"id":"model","name":"Model","category":"model","type":"select","currentValue":"opus","options":[]}]}),
|
|
serde_json::json!({"sessionUpdate":"plan","entries":[]}),
|
|
]
|
|
.into_iter()
|
|
.enumerate()
|
|
{
|
|
projection.apply((seq + 1) as u64, &event(update));
|
|
}
|
|
let state = projection.state();
|
|
assert_eq!(state.usage.as_ref().unwrap()["used"], 12);
|
|
assert_eq!(state.context.as_ref().unwrap()["size"], 200);
|
|
assert_eq!(state.mode.as_deref(), Some("plan"));
|
|
assert_eq!(state.model.as_deref(), Some("opus"));
|
|
assert_eq!(
|
|
state.config_options.as_ref().unwrap()["configOptions"][0]["currentValue"],
|
|
"opus"
|
|
);
|
|
assert_eq!(
|
|
state.plan.as_ref().unwrap()["entries"],
|
|
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());
|
|
|
|
let (_, _) = projection.begin_turn();
|
|
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,
|
|
] {
|
|
let (turn, _) = projection.begin_turn();
|
|
projection.turn_lifecycle(terminal, turn);
|
|
assert!(!projection.state().turn.responding);
|
|
assert!(!projection.state().turn.thinking);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn completed_turn_keeps_duration_and_context_delta_in_transcript() {
|
|
let mut projection = Projection::default();
|
|
projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"usage_update", "used":10_000, "size":200_000
|
|
})),
|
|
);
|
|
let (turn, begun) = projection.begin_turn();
|
|
assert!(begun.state.turn.started_at.is_some());
|
|
assert_eq!(begun.state.turn.started_context_tokens, Some(10_000));
|
|
projection.apply(
|
|
2,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"usage_update", "used":12_500, "size":200_000
|
|
})),
|
|
);
|
|
|
|
let finished = projection
|
|
.turn_lifecycle(TurnLifecycle::Finish, turn)
|
|
.expect("active turn finishes");
|
|
let TranscriptDelta::Append { row } = finished.transcript.unwrap() else {
|
|
panic!("turn footer");
|
|
};
|
|
assert_eq!(row.kind, TranscriptRowKind::Notice);
|
|
assert_eq!(row.data["maestroTurnMeta"]["contextDelta"], 2_500);
|
|
assert_eq!(row.data["maestroTurnMeta"]["contextTokens"], 12_500);
|
|
assert_eq!(row.data["maestroTurnMeta"]["outcome"], "completed");
|
|
assert!(finished.state.turn.started_at.is_none());
|
|
}
|
|
|
|
fn terminal_turn_drops_unfinished_tools(
|
|
terminal: TurnLifecycle,
|
|
unfinished_id: &str,
|
|
next_id: &str,
|
|
) {
|
|
let mut projection = Projection::default();
|
|
let (turn, _) = projection.begin_turn();
|
|
projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"tool_call", "toolCallId":unfinished_id,
|
|
"title":"Unfinished", "status":"in_progress"
|
|
})),
|
|
);
|
|
assert_eq!(projection.state().turn.active_tool_calls, [unfinished_id]);
|
|
|
|
projection.turn_lifecycle(terminal, turn);
|
|
assert!(projection.state().turn.active_tool_calls.is_empty());
|
|
let rows = projection.snapshot(1).rows;
|
|
assert_eq!(rows.len(), 2);
|
|
assert!(rows[1].data.get("maestroTurnMeta").is_some());
|
|
|
|
projection.begin_turn();
|
|
projection.apply(
|
|
2,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate":"tool_call", "toolCallId":next_id,
|
|
"title":"Next", "status":"pending"
|
|
})),
|
|
);
|
|
assert_eq!(projection.state().turn.active_tool_calls, [next_id]);
|
|
}
|
|
|
|
#[test]
|
|
fn cancelled_turn_does_not_reactivate_unfinished_tools() {
|
|
terminal_turn_drops_unfinished_tools(TurnLifecycle::Cancel, "cancelled-1", "next-1");
|
|
}
|
|
|
|
#[test]
|
|
fn failed_turn_does_not_reactivate_unfinished_tools() {
|
|
terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
|
|
}
|
|
|
|
#[test]
|
|
fn approval_history_survives_pending_cleanup() {
|
|
let mut projection = Projection::default();
|
|
let update =
|
|
projection.request_approval("secret".into(), serde_json::json!({"tool":"bash"}));
|
|
let TranscriptDelta::Append { row } = update.transcript.unwrap() else {
|
|
panic!("approval row")
|
|
};
|
|
assert_eq!(row.kind, TranscriptRowKind::Approval);
|
|
assert!(row.data.get("nonce").is_none());
|
|
assert!(!row.id.contains("secret"));
|
|
let persisted =
|
|
serde_json::to_string(&crate::sidecar::PersistedEvent::Approval { row: row.clone() })
|
|
.unwrap();
|
|
assert!(!persisted.contains("secret"));
|
|
assert_eq!(
|
|
projection.state().pending_approval.unwrap()["nonce"],
|
|
"secret"
|
|
);
|
|
projection.clear_approval();
|
|
assert!(projection.state().pending_approval.is_none());
|
|
assert_eq!(projection.snapshot(0).rows, vec![row]);
|
|
}
|
|
|
|
#[test]
|
|
fn rollback_row_removes_only_that_row_and_keeps_others_addressable() {
|
|
let mut projection = Projection::default();
|
|
projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "user_message_chunk",
|
|
"messageId": "m1",
|
|
"content": {"type": "text", "text": "hi"},
|
|
})),
|
|
);
|
|
let TranscriptDelta::Append { row } = projection
|
|
.request_approval("nonce".into(), serde_json::json!({"tool": "bash"}))
|
|
.transcript
|
|
.unwrap()
|
|
else {
|
|
panic!("approval row")
|
|
};
|
|
// A row after the one being rolled back, so removing the approval
|
|
// (index 1) shifts this one from index 2 down to index 1 — proving
|
|
// `row_index` gets reindexed, not just `rows`.
|
|
projection.apply(
|
|
2,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "user_message_chunk",
|
|
"messageId": "m2",
|
|
"content": {"type": "text", "text": "hey"},
|
|
})),
|
|
);
|
|
assert_eq!(projection.snapshot(0).rows.len(), 3);
|
|
|
|
projection.rollback_row(&row.id);
|
|
|
|
assert_eq!(projection.snapshot(0).rows.len(), 2);
|
|
assert_eq!(projection.snapshot(0).rows[0].kind, TranscriptRowKind::User);
|
|
assert_eq!(projection.snapshot(0).rows[1].kind, TranscriptRowKind::User);
|
|
|
|
// The shifted row must still be independently patchable at its new
|
|
// index.
|
|
let update = projection.apply(
|
|
3,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "user_message_chunk",
|
|
"messageId": "m2",
|
|
"content": {"type": "text", "text": " there"},
|
|
})),
|
|
);
|
|
let TranscriptDelta::Patch { row } = update.transcript.unwrap() else {
|
|
panic!("patched row")
|
|
};
|
|
assert_eq!(row.data.pointer("/content/text").unwrap(), "hey there");
|
|
assert_eq!(projection.snapshot(0).rows.len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn rollback_row_is_a_no_op_for_an_unknown_id() {
|
|
let mut projection = Projection::default();
|
|
projection.apply(
|
|
1,
|
|
&event(serde_json::json!({
|
|
"sessionUpdate": "user_message_chunk",
|
|
"content": {"type": "text", "text": "hi"},
|
|
})),
|
|
);
|
|
projection.rollback_row("approval:does-not-exist");
|
|
assert_eq!(projection.snapshot(0).rows.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn recovered_approval_is_historical_not_pending() {
|
|
let mut live = Projection::default();
|
|
let TranscriptDelta::Append { row } = live
|
|
.request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"}))
|
|
.transcript
|
|
.unwrap()
|
|
else {
|
|
panic!("approval row")
|
|
};
|
|
let mut recovered = Projection::default();
|
|
let update = recovered.restore_row(row.clone());
|
|
assert_eq!(
|
|
update.transcript,
|
|
Some(TranscriptDelta::Append { row: row.clone() })
|
|
);
|
|
assert!(update.state.pending_approval.is_none());
|
|
assert_eq!(recovered.snapshot(0).rows, vec![row]);
|
|
let replay = serde_json::to_string(&recovered.snapshot(1).rows).unwrap();
|
|
assert!(!replay.contains("daemon-nonce"));
|
|
}
|
|
|
|
#[test]
|
|
fn stale_terminal_turn_cannot_clear_newer_approval() {
|
|
let mut projection = Projection::default();
|
|
let (first, _) = projection.begin_turn();
|
|
projection.turn_lifecycle(TurnLifecycle::Cancel, first);
|
|
let (second, _) = projection.begin_turn();
|
|
projection.request_approval("second-nonce".into(), serde_json::json!({"tool":"bash"}));
|
|
|
|
assert!(projection
|
|
.turn_lifecycle(TurnLifecycle::Finish, first)
|
|
.is_none());
|
|
assert_eq!(projection.active_turn(), Some(second));
|
|
assert_eq!(
|
|
projection.state().pending_approval.unwrap()["nonce"],
|
|
"second-nonce"
|
|
);
|
|
}
|
|
}
|