Merge remote-tracking branch 'origin/main' into mandolin

# Conflicts:
#	crates/maestro-daemon/src/local_api.rs
#	crates/maestro-daemon/src/session.rs

crates/maestro-daemon/src/local_api.rs

@@ -482,7 +482,7 @@ async fn structured_prompt(
     Path(id): Path<Uuid>,
     Json(body): Json<StructuredPromptRequest>,
 ) -> Response {
-    match s.manager.queue_acp_prompt(id, body.prompt).await {
+    match s.manager.queue_acp_prompt(id, body.prompt, None).await {
         Ok(QueueAcpPrompt::Dispatched) => StatusCode::NO_CONTENT.into_response(),
         Ok(QueueAcpPrompt::Queued) => StatusCode::ACCEPTED.into_response(),
         Ok(QueueAcpPrompt::Full) => error_response(

crates/maestro-daemon/src/session.rs

@@ -757,6 +757,7 @@ impl SessionManager {
         &self,
         id: Uuid,
         text: String,
+        row_metadata: Option<serde_json::Value>,
     ) -> Result<QueueAcpPrompt, SessionError> {
         let (acp, stream) = {
             let sessions = self.inner.sessions.lock().await;
@@ -833,12 +834,19 @@ impl SessionManager {
                         self.persist_prompt_queue(id).await;
                         return Ok(QueueAcpPrompt::Queued);
                     }
+                    let mut data = serde_json::json!({
+                        "content": {"type": "text", "text": text},
+                    });
+                    if let Some(extra) = &row_metadata {
+                        if let (Some(base), Some(extra)) = (data.as_object_mut(), extra.as_object())
+                        {
+                            base.extend(extra.iter().map(|(k, v)| (k.clone(), v.clone())));
+                        }
+                    }
                     let row = TranscriptRow {
                         id: format!("user:{}", Uuid::new_v4()),
                         kind: TranscriptRowKind::User,
-                        data: serde_json::json!({
-                            "content": {"type": "text", "text": text},
-                        }),
+                        data,
                     };
                     let event = PersistedEvent::User { row: row.clone() };
                     let seq = match stream.log.append(&event) {
@@ -1631,7 +1639,7 @@ impl SessionManager {
                             let manager = SessionManager {
                                 inner: inner.clone(),
                             };
-                            let _ = manager.queue_acp_prompt(id, text).await;
+                            let _ = manager.queue_acp_prompt(id, text, None).await;
                         }
                     }
                     Err(e) => {
@@ -2022,7 +2030,9 @@ impl SessionManager {
                 if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                     record.acp_event_pump = Some(done);
                 }
-                // Drain any prompts that were queued before the daemon restarted.
+                // Drain any prompts that were queued before the daemon
+                // restarted. If none were queued, send a synthetic continue
+                // prompt so the agent resumes autonomously.
                 let queued_text = {
                     let stream = acp_events
                         .as_ref()
@@ -2037,7 +2047,15 @@ impl SessionManager {
                 };
                 if let Some(text) = queued_text {
                     self.persist_prompt_queue(id).await;
-                    let _ = self.queue_acp_prompt(id, text).await;
+                    let _ = self.queue_acp_prompt(id, text, None).await;
+                } else {
+                    let _ = self
+                        .queue_acp_prompt(
+                            id,
+                            "The daemon restarted. Continue where you left off.".into(),
+                            Some(serde_json::json!({"maestro": {"type": "restart_continue"}})),
+                        )
+                        .await;
                 }
             }
             let _ = self
@@ -3017,7 +3035,7 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 fn drain_prompt_queue(manager: SessionManager, id: Uuid, text: String) {
     let fut: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> =
         Box::pin(async move {
-            let _ = manager.queue_acp_prompt(id, text).await;
+            let _ = manager.queue_acp_prompt(id, text, None).await;
         });
     tokio::spawn(fut);
 }

crates/maestro-daemon/tests/acp.rs

@@ -84,15 +84,27 @@ impl Fake {
 }
 
 fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
-    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
-        panic!("unexpected daemon event")
-    };
-    match &notification.update {
-        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
-            ContentBlock::Text(text) => &text.text,
-            other => panic!("unexpected content {other:?}"),
-        },
-        other => panic!("unexpected update {other:?}"),
+    match &event.event {
+        maestro_daemon::sidecar::PersistedEvent::Notification(notification) => {
+            match &notification.update {
+                SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+                    ContentBlock::Text(text) => &text.text,
+                    other => panic!("unexpected content {other:?}"),
+                },
+                other => panic!("unexpected update {other:?}"),
+            }
+        }
+        maestro_daemon::sidecar::PersistedEvent::User { row } => row
+            .data
+            .pointer("/content/text")
+            .and_then(serde_json::Value::as_str)
+            .expect("user row text"),
+        maestro_daemon::sidecar::PersistedEvent::Turn { row } => row
+            .data
+            .pointer("/maestroTurnMeta/outcome")
+            .and_then(serde_json::Value::as_str)
+            .unwrap_or("[turn]"),
+        maestro_daemon::sidecar::PersistedEvent::Approval { .. } => "[approval]",
     }
 }
 
@@ -259,7 +271,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "fail".into())
+            .queue_acp_prompt(session.id, "fail".into(), None)
             .await
             .unwrap(),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -286,7 +298,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "succeed".into())
+            .queue_acp_prompt(session.id, "succeed".into(), None)
             .await
             .unwrap(),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -426,7 +438,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "show my prompt".into())
+            .queue_acp_prompt(session.id, "show my prompt".into(), None)
             .await
             .expect("queue prompt"),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -516,7 +528,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "first".into())
+            .queue_acp_prompt(session.id, "first".into(), None)
             .await
             .unwrap(),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -524,7 +536,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
     assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
     let second = fx
         .manager
-        .queue_acp_prompt(session.id, "second".into())
+        .queue_acp_prompt(session.id, "second".into(), None)
         .await
         .unwrap();
     assert!(
@@ -555,7 +567,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "first".into())
+            .queue_acp_prompt(session.id, "first".into(), None)
             .await
             .unwrap(),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -563,7 +575,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
     assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "second".into())
+            .queue_acp_prompt(session.id, "second".into(), None)
             .await
             .unwrap(),
         maestro_daemon::session::QueueAcpPrompt::Dispatched
@@ -655,25 +667,18 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         revived.status(session.id).await,
         Some(SessionStatus::Running)
     );
-    assert_eq!(
-        revived
-            .snapshot()
-            .await
-            .into_iter()
-            .find(|candidate| candidate.id == session.id)
-            .unwrap()
-            .activity,
-        ActivityState::Idle
-    );
     let log = fake.log();
     assert!(
         log.contains(&format!("recv session/load sessionId={acp_id}")),
         "{log}"
     );
+    // Recovery adds: a continue-prompt User row, the adapter's post-load
+    // notification, and a turn-completed footer — 5 events total before the
+    // next manual tick.
     wait_for_async(Duration::from_secs(5), || async {
         revived
             .read_acp_events(session.id, EventCursor::Since(0), 10)
-            .is_ok_and(|events| events.len() == 3)
+            .is_ok_and(|events| events.len() == 5)
     })
     .await;
 
@@ -682,7 +687,7 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         .await
         .expect("replay read")
         .expect("recovered event subscription");
-    assert_eq!(replay.len(), 3);
+    assert_eq!(replay.len(), 5);
     assert!(live.try_recv().is_err(), "replay duplicated onto live");
     let new_pid = fake.pid();
     assert_ne!(new_pid, old_pid);
@@ -693,16 +698,23 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         .await
         .expect("live event after recovery")
         .expect("broadcast open");
-    assert_eq!(event.seq, 4);
+    assert_eq!(event.seq, 6);
     assert_eq!(event_text(&event), "tick 1");
 
     let stored = revived
         .read_acp_events(session.id, EventCursor::Since(0), 10)
         .expect("recovered log");
-    assert_eq!(stored.len(), 4, "session/load replay must not be appended");
+    assert_eq!(stored.len(), 6, "session/load replay must not be appended");
     assert_eq!(
         stored.iter().map(event_text).collect::<Vec<_>>(),
-        ["hello from fake", "tick 1", "hello after load", "tick 1"]
+        [
+            "hello from fake",
+            "tick 1",
+            "The daemon restarted. Continue where you left off.",
+            "hello after load",
+            "completed",
+            "tick 1",
+        ]
     );
 
     revived.kill(session.id, false).await.expect("kill");