From ccdfec8257bbaec0ecfdaf6ee26d28e3ac5495fc Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sun, 6 Sep 2026 14:19:45 +0100 Subject: [PATCH] fix(daemon): persist structured user prompts --- crates/maestro-daemon/src/projection.rs | 37 ++++++++++++++++++-- crates/maestro-daemon/src/session.rs | 29 ++++++++++++++-- crates/maestro-daemon/src/sidecar.rs | 3 ++ crates/maestro-daemon/tests/acp.rs | 45 +++++++++++++++++++++++++ docs/STRUCTURED-VIEW.md | 14 ++++---- 5 files changed, 116 insertions(+), 12 deletions(-) diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 0209d4d..576764f 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -192,7 +192,8 @@ impl Projection { } } - pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate { + 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()); @@ -454,7 +455,7 @@ fn model_from_options(value: &Value) -> Option { #[cfg(test)] mod tests { use maestro_acp::schema::SessionNotification; - use maestro_protocol::{TranscriptDelta, TranscriptRowKind}; + use maestro_protocol::{TranscriptDelta, TranscriptRow, TranscriptRowKind}; use super::{Projection, TurnLifecycle}; @@ -572,6 +573,36 @@ mod tests { 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(); @@ -905,7 +936,7 @@ mod tests { panic!("approval row") }; let mut recovered = Projection::default(); - let update = recovered.restore_approval(row.clone()); + let update = recovered.restore_row(row.clone()); assert_eq!( update.transcript, Some(TranscriptDelta::Append { row: row.clone() }) diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index fdadef5..8227b10 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -18,8 +18,8 @@ use chrono::Utc; use maestro_acp::{AcpSession, PermissionRequest, SpawnConfig}; use maestro_protocol::{ ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, - SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, WorktreeSpec, NAME_KEY, - NAME_SOURCE_KEY, + SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow, + TranscriptRowKind, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, }; use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify}; use uuid::Uuid; @@ -719,6 +719,27 @@ impl SessionManager { if stream.projection.turn_active() { return Ok(QueueAcpPrompt::InFlight); } + let row = TranscriptRow { + id: format!("user:{}", Uuid::new_v4()), + kind: TranscriptRowKind::User, + data: serde_json::json!({ + "content": {"type": "text", "text": text}, + }), + }; + let event = PersistedEvent::User { row: row.clone() }; + let seq = match stream.log.append(&event) { + Ok(seq) => seq, + Err(error) => { + if matches!(error, AppendError::Fatal { .. }) { + stream.dispatch.failed = true; + } + return Ok(QueueAcpPrompt::Failed(error.to_string())); + } + }; + let logged = LoggedEvent { seq, event }; + let projected = stream.projection.restore_row(row); + let _ = stream.live.send(logged); + let _ = stream.projected_live.send(projected); let (turn, update) = stream.projection.begin_turn(); let _ = stream.projected_live.send(update); match acp.prompt(text.clone(), turn) { @@ -2118,7 +2139,9 @@ fn start_acp_event_pump( fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> ProjectionUpdate { match &event.event { PersistedEvent::Notification(notification) => projection.apply(event.seq, notification), - PersistedEvent::Approval { row } => projection.restore_approval(row.clone()), + PersistedEvent::User { row } | PersistedEvent::Approval { row } => { + projection.restore_row(row.clone()) + } } } diff --git a/crates/maestro-daemon/src/sidecar.rs b/crates/maestro-daemon/src/sidecar.rs index 8e2762d..fb8c3e8 100644 --- a/crates/maestro-daemon/src/sidecar.rs +++ b/crates/maestro-daemon/src/sidecar.rs @@ -263,6 +263,9 @@ pub struct LoggedEvent { #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] #[serde(tag = "maestroEvent", rename_all = "snake_case")] pub enum PersistedEvent { + User { + row: TranscriptRow, + }, Approval { row: TranscriptRow, }, diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index d8461ff..8e7b595 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -355,6 +355,51 @@ async fn projection_snapshot_and_live_stream_share_one_fold() { fx.manager.kill(session.id, false).await.expect("kill"); } +#[tokio::test] +async fn queued_prompt_is_a_durable_user_transcript_row() { + 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"); + + assert!(matches!( + fx.manager + .queue_acp_prompt(session.id, "show my prompt".into()) + .await + .expect("queue prompt"), + maestro_daemon::session::QueueAcpPrompt::Queued + )); + + let (snapshot, _, _) = fx + .manager + .subscribe_acp_projection(session.id, EventCursor::Since(0)) + .await + .expect("projection subscription") + .expect("structured session"); + let user = snapshot + .rows + .iter() + .find(|row| row.kind == TranscriptRowKind::User) + .expect("user prompt row"); + assert_eq!(user.data["content"]["text"], "show my prompt"); + + let persisted = fx + .manager + .read_acp_events(session.id, EventCursor::Since(0), 100) + .expect("event replay"); + assert!(persisted.iter().any(|event| matches!( + &event.event, + maestro_daemon::sidecar::PersistedEvent::User { row } + if row.data["content"]["text"] == "show my prompt" + ))); + + fx.manager.kill(session.id, false).await.unwrap(); +} + #[tokio::test] async fn stale_turn_completion_cannot_end_the_next_turn() { let fx = DaemonFixture::new(); diff --git a/docs/STRUCTURED-VIEW.md b/docs/STRUCTURED-VIEW.md index 69f0876..2be8347 100644 --- a/docs/STRUCTURED-VIEW.md +++ b/docs/STRUCTURED-VIEW.md @@ -106,7 +106,7 @@ crate. | Kind | Carries | Row identity | | --- | --- | --- | -| `User` | An echoed `user_message_chunk` — the prompt text as ACP saw it. | `user:`, or `user:` when the chunk carries no id. | +| `User` | The prompt accepted by Maestro's structured prompt endpoint, persisted before ACP dispatch. An adapter `user_message_chunk` is also supported for imported ACP history. | `user:` for a daemon-recorded prompt; `user:` or `user:` for an ACP chunk. | | `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:`, or `agent:` from the first chunk when no id is supplied. Later chunks for the same id—or consecutive anonymous text chunks—are appended into `data.content.text` in place. An empty anonymous chunk starts a fresh row, matching the turn marker emitted by adapters that omit message ids. | | `AgentThought` | An `agent_thought_chunk` — the model's reasoning stream. | `thought:` / first anonymous `thought:`, coalesced the same way as `AgentText`. | | `ToolCall` | A `tool_call` / `tool_call_update` pair whose content carries no diff. | `tool:` — one row for the call's whole life; updates patch it in place, they never create a second row. | @@ -197,7 +197,7 @@ every row, once appended, is only ever patched or left alone. The event log's own line position is the cursor: 1-based, shared by ACP notifications and daemon-generated rows alike (`sidecar.rs`'s -`PersistedEvent` envelope is what lets an `Approval` row share a cursor +`PersistedEvent` envelope is what lets `User` and `Approval` rows share a cursor sequence with raw ACP JSON without pretending to be ACP traffic). Two ways to read it: @@ -247,7 +247,7 @@ pending decision, entirely on the daemon side: - **Secrecy from the agent and from durable history.** The adapter itself never learns the nonce — it is a daemon-to-client secret, never sent back over the ACP connection. It is likewise never written to the event log: - restoring an `Approval` row from disk (`Projection::restore_approval`) + restoring an `Approval` row from disk (`Projection::restore_row`) produces a historical row with no `pending_approval` in state, because the nonce that would make it "pending" never made it to disk in the first place. A restart therefore cannot resume a pending approval — see @@ -289,7 +289,9 @@ thing a callback can be attributed to, so at most one turn may ever be in flight, and a cancelled turn must fully settle before the next prompt is allowed to claim the daemon's idea of "the active turn." -- A prompt claims the turn and dispatches it to ACP as one operation. While a +- A prompt is appended as a durable `User` row before it claims the turn and + dispatches to ACP, so the transcript never depends on an adapter echoing the + operator's input. If that append fails, the prompt is not dispatched. While a turn is active, a second prompt gets `409 turn_in_flight` rather than queuing — one active turn per session, never a hidden queue of prompts. - Cancel sends `session/cancel`, clears any pending approval as `Cancelled`, @@ -315,8 +317,8 @@ directory: the sidecar itself (`.json`), plus two next to it: - **`.events.jsonl`** — the unified event log. Each line is either a raw ACP `SessionNotification` (kept in its original JSON so old logs stay - readable) or a `{"maestroEvent": "approval", "row": …}` envelope for a - daemon-generated row. Both share the same 1-based line-position cursor. A + readable) or a `{"maestroEvent": "user"|"approval", "row": …}` envelope + for a daemon-generated row. Both share the same 1-based line-position cursor. A line a hard kill left half-written is terminated before the next append, so it keeps its position and is skipped as malformed rather than corrupting what follows. An append that fails *while the daemon is running* — a