fix(daemon): persist structured user prompts

This commit is contained in:
Miguel Palhas
2026-09-06 14:19:45 +01:00
parent 30f91176ae
commit ccdfec8257
5 changed files with 116 additions and 12 deletions
+34 -3
View File
@@ -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<String> {
#[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() })
+26 -3
View File
@@ -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())
}
}
}
+3
View File
@@ -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,
},
+45
View File
@@ -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();
+8 -6
View File
@@ -106,7 +106,7 @@ crate.
| Kind | Carries | Row identity |
| --- | --- | --- |
| `User` | An echoed `user_message_chunk` — the prompt text as ACP saw it. | `user:<message_id>`, or `user:<seq>` 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:<uuid>` for a daemon-recorded prompt; `user:<message_id>` or `user:<seq>` for an ACP chunk. |
| `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:<message_id>`, or `agent:<seq>` 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:<message_id>` / first anonymous `thought:<seq>`, coalesced the same way as `AgentText`. |
| `ToolCall` | A `tool_call` / `tool_call_update` pair whose content carries no diff. | `tool:<tool_call_id>` — 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 (`<id>.json`), plus two next to it:
- **`<id>.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