From 046c110460565a564bfce81b0b0f7648bb62696c Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 5 Sep 2026 01:13:39 +0100 Subject: [PATCH] feat(maestro): project ACP transcript state --- Cargo.toml | 2 +- crates/maestro-daemon/src/lib.rs | 1 + crates/maestro-daemon/src/projection.rs | 359 ++++++++++++++++++++++++ crates/maestro-daemon/src/session.rs | 97 ++++++- crates/maestro-daemon/tests/acp.rs | 53 +++- crates/maestro-protocol/src/lib.rs | 106 +++++++ 6 files changed, 614 insertions(+), 4 deletions(-) create mode 100644 crates/maestro-daemon/src/projection.rs diff --git a/Cargo.toml b/Cargo.toml index ebdb7ba..a49af96 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ repository = "https://git.naps.pt/naps62/maestro" publish = false [workspace.dependencies] -agent-client-protocol = "2.1" +agent-client-protocol = { version = "2.1", features = ["unstable_tool_call_name"] } axum = { version = "0.7", features = ["ws"] } chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] } clap = { version = "4.5", features = ["derive", "env"] } diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 6d67cea..db818a4 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -21,6 +21,7 @@ pub mod hooks; pub mod local_api; pub mod panes; pub mod proc; +pub mod projection; pub mod projects; pub mod pty; pub mod release; diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs new file mode 100644 index 0000000..1d330ca --- /dev/null +++ b/crates/maestro-daemon/src/projection.rs @@ -0,0 +1,359 @@ +//! Fold persisted ACP notifications into client-neutral transcript and state. + +use std::collections::HashMap; + +use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall}; +use maestro_protocol::{ + 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, + pub state: ReducedState, +} + +/// 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, + state: ReducedState, +} + +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.state.turn.responding = true; + self.state.turn.thinking = false; + self.content(seq, chunk, TranscriptRowKind::AgentText, "agent") + } + SessionUpdate::AgentThoughtChunk(chunk) => { + self.state.turn.responding = true; + self.state.turn.thinking = true; + self.content(seq, chunk, TranscriptRowKind::AgentThought, "thought") + } + SessionUpdate::ToolCall(tool) => self.tool_call(tool), + SessionUpdate::ToolCallUpdate(update) => self.tool_update(update), + SessionUpdate::Plan(plan) => { + let data = to_value(plan); + self.state.plan = Some(data.clone()); + Some(self.append(format!("plan:{seq}"), 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()); + 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. + _ => 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() + } + + fn content( + &mut self, + seq: u64, + chunk: &ContentChunk, + kind: TranscriptRowKind, + prefix: &str, + ) -> Option { + let id = chunk + .message_id + .as_ref() + .map(|id| format!("{prefix}:{id}")) + .unwrap_or_else(|| format!("{prefix}:{seq}")); + let data = to_value(chunk); + 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 { + let id = format!("tool:{}", tool.tool_call_id); + let kind = if tool_has_diff(tool) { + TranscriptRowKind::ToolDiff + } else { + TranscriptRowKind::ToolCall + }; + self.tools + .insert(tool.tool_call_id.to_string(), tool.clone()); + self.sync_active_tools(); + Some(self.append(id, kind, to_value(tool))) + } + + fn tool_update( + &mut self, + update: &maestro_acp::schema::ToolCallUpdate, + ) -> 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), + )); + }; + tool.update(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_tools(); + 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 sync_active_tools(&mut self) { + self.state.turn.active_tool_calls = self + .tools + .iter() + .filter(|(_, tool)| { + matches!( + tool.status, + maestro_acp::schema::ToolCallStatus::Pending + | maestro_acp::schema::ToolCallStatus::InProgress + ) + }) + .map(|(id, _)| id.clone()) + .collect(); + self.state.turn.active_tool_calls.sort(); + } +} + +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 { + 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, TranscriptRowKind}; + + use super::Projection; + + fn event(value: serde_json::Value) -> SessionNotification { + serde_json::from_value(serde_json::json!({ + "sessionId": "acp-1", + "update": value, + })) + .unwrap() + } + + #[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 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 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.plan.as_ref().unwrap()["entries"], + serde_json::json!([]) + ); + } +} diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 2d087fc..6f93e94 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -21,6 +21,7 @@ use tokio::sync::{broadcast, watch, Mutex}; use uuid::Uuid; use crate::panes::Panes; +use crate::projection::{Projection, ProjectionUpdate}; use crate::pty::{pipe_pane_command, PtyBridge}; use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore}; use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord}; @@ -106,6 +107,8 @@ pub struct SessionRecord { struct AcpEventStream { log: EventLog, live: broadcast::Sender, + projection: Projection, + projected_live: broadcast::Sender, } #[derive(Debug, thiserror::Error)] @@ -354,6 +357,45 @@ impl SessionManager { .map_err(SessionError::Sidecar) } + /// Subscribe to the two projections of a structured session's ACP log. + /// + /// The transcript snapshot and whole reduced state are folded once by the + /// event pump. Holding the stream lock while subscribing makes the live + /// continuation start strictly after that snapshot, with no replay/live + /// overlap. A cursor is accepted for the future transport layer; a + /// snapshot intentionally remains complete so a tool patch always has its + /// original row available, while reduced state is always full-history. + pub async fn subscribe_acp_projection( + &self, + id: Uuid, + _cursor: EventCursor, + ) -> Result< + Option<( + maestro_protocol::TranscriptSnapshot, + maestro_protocol::ReducedState, + broadcast::Receiver, + )>, + SessionError, + > { + let stream = { + let sessions = self.inner.sessions.lock().await; + let Some(record) = sessions.get(&id) else { + return Ok(None); + }; + let Some(stream) = &record.acp_events else { + return Ok(None); + }; + stream.clone() + }; + let stream = stream.lock().await; + let live = stream.projected_live.subscribe(); + Ok(Some(( + stream.projection.snapshot(stream.log.last_seq()), + stream.projection.state(), + live, + ))) + } + /// 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 @@ -607,10 +649,35 @@ impl SessionManager { }; let acp_events = if let Some(acp) = &acp { + let persisted = + match self + .inner + .sidecar + .read_events(id, EventCursor::Since(0), usize::MAX) + { + Ok(events) => events, + Err(error) => { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + let _ = std::fs::remove_file(&fifo_path); + self.rollback_worktree(&worktree_base_repo, &worktree_path); + return Err(SessionError::Sidecar(error)); + } + }; match self.inner.sidecar.open_events(id) { Ok(log) => { let (tx, _) = broadcast::channel(EVENTS_CAPACITY); - Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx }))) + let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY); + let mut projection = Projection::default(); + for event in &persisted { + projection.apply(event.seq, &event.notification); + } + Some(Arc::new(Mutex::new(AcpEventStream { + log, + live: tx, + projection, + projected_live: projected_tx, + }))) } Err(e) => { acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) @@ -1025,10 +1092,34 @@ impl SessionManager { } let acp_events = if let Some(acp) = &acp { + let persisted = + match self + .inner + .sidecar + .read_events(id, EventCursor::Since(0), usize::MAX) + { + Ok(events) => events, + Err(error) => { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + let _ = std::fs::remove_file(&fifo_path); + return Err(SessionError::Sidecar(error)); + } + }; match self.inner.sidecar.open_events(id) { Ok(log) => { let (tx, _) = broadcast::channel(EVENTS_CAPACITY); - Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx }))) + let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY); + let mut projection = Projection::default(); + for event in &persisted { + projection.apply(event.seq, &event.notification); + } + Some(Arc::new(Mutex::new(AcpEventStream { + log, + live: tx, + projection, + projected_live: projected_tx, + }))) } Err(error) => { acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) @@ -1435,7 +1526,9 @@ fn start_acp_event_pump( let mut events = events.lock().await; match events.log.append(¬ification) { Ok(seq) => { + let projected = events.projection.apply(seq, ¬ification); let _ = events.live.send(LoggedEvent { seq, notification }); + let _ = events.projected_live.send(projected); None } Err(error) => Some(error), diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index 47314c8..53a0d98 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -16,7 +16,8 @@ use maestro_daemon::sidecar::{EventCursor, SidecarStore}; use maestro_daemon::worktree::WorktreeManager; use maestro_daemon::SessionManager; use maestro_protocol::{ - ActivityState, CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec, + ActivityState, CreateSessionRequest, SessionStatus, SessionView, TranscriptDelta, + TranscriptRowKind, WorktreeSpec, }; use tempfile::TempDir; @@ -199,6 +200,56 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { .exists()); } +#[tokio::test] +async fn projection_snapshot_and_live_stream_share_one_fold() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + let session = fx + .manager + .create(structured(cwd.path(), fake.env())) + .await + .expect("create"); + + wait_for_async(Duration::from_secs(5), || async { + fx.manager + .read_acp_events(session.id, EventCursor::Since(0), 10) + .is_ok_and(|events| events.len() == 1) + }) + .await; + + // The cursor does not truncate either projection: a reconnect needs the + // source row for future patches and whole reduced state from the session. + let (snapshot, state, mut live) = fx + .manager + .subscribe_acp_projection(session.id, EventCursor::Since(1)) + .await + .expect("projection subscription") + .expect("structured session"); + assert_eq!(snapshot.cursor, 1); + assert_eq!(snapshot.rows.len(), 1); + assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText); + assert!(state.turn.responding); + + unsafe { + libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1); + } + let live = tokio::time::timeout(Duration::from_secs(5), live.recv()) + .await + .expect("live projection") + .expect("projection broadcast"); + match live.transcript { + Some(TranscriptDelta::Append { row }) => { + assert_eq!(row.kind, TranscriptRowKind::AgentText); + assert_eq!(row.data["content"]["text"], "tick 1"); + } + other => panic!("expected one appended transcript row, got {other:?}"), + } + assert!(live.state.turn.responding); + + fx.manager.kill(session.id, false).await.expect("kill"); +} + #[tokio::test] async fn recovery_loads_the_acp_session_without_duplicating_history() { let fx = DaemonFixture::new(); diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index ad38b97..40bc66e 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -755,6 +755,87 @@ pub enum CatalogEvent { }, } +// --------------------------------------------------------------------------- +// Structured-session projections +// --------------------------------------------------------------------------- + +/// One client-neutral item in a structured session transcript. +/// +/// `data` preserves the ACP payload without committing clients to an ACP +/// dependency. Rendering choices, including markdown and tool-card treatment, +/// stay with the client. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct TranscriptRow { + /// Stable within one session, so later deltas can address this row. + pub id: String, + pub kind: TranscriptRowKind, + pub data: Value, +} + +/// The semantic shape of a [`TranscriptRow`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TranscriptRowKind { + User, + AgentText, + AgentThought, + ToolCall, + ToolDiff, + Approval, + Plan, + Notice, +} + +/// A complete transcript at a durable event-log position. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct TranscriptSnapshot { + pub cursor: u64, + #[serde(default)] + pub rows: Vec, +} + +/// A change to a transcript row. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum TranscriptDelta { + Append { row: TranscriptRow }, + Patch { row: TranscriptRow }, + Remove { id: String }, +} + +/// Per-turn indicators derived from the ACP stream. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct TurnState { + #[serde(default)] + pub responding: bool, + #[serde(default)] + pub thinking: bool, + #[serde(default)] + pub active_tool_calls: Vec, +} + +/// A whole-session view derived from ACP notifications. +/// +/// Values whose shape ACP owns are carried as JSON so this wire protocol stays +/// forward-compatible when ACP adds fields or variants. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReducedState { + #[serde(default)] + pub turn: TurnState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pending_approval: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plan: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub context: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub mode: Option, +} + // --------------------------------------------------------------------------- // Client ↔ Daemon control messages (text frames on the attach WS) // --------------------------------------------------------------------------- @@ -1144,6 +1225,31 @@ mod tests { assert_eq!(structured, round_trip(&structured)); } + #[test] + fn structured_projection_types_round_trip_and_default() { + let row = TranscriptRow { + id: "tool:read-1".into(), + kind: TranscriptRowKind::ToolCall, + data: serde_json::json!({"name": "vendor.read", "title": "Read"}), + }; + let snapshot = TranscriptSnapshot { + cursor: 7, + rows: vec![row.clone()], + }; + assert_eq!(snapshot, round_trip(&snapshot)); + assert_eq!( + TranscriptDelta::Patch { row: row.clone() }, + round_trip(&TranscriptDelta::Patch { row }) + ); + + let state: ReducedState = serde_json::from_value(serde_json::json!({})).unwrap(); + assert_eq!(state, ReducedState::default()); + assert_eq!( + serde_json::to_value(TranscriptRowKind::AgentThought).unwrap(), + serde_json::json!("agent_thought") + ); + } + #[test] fn acp_session_id_defaults_to_none_and_round_trips() { // A payload from before #134 carries no `acp_session_id`; it must