feat(maestro): project ACP transcript state

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<LoggedEvent>,
+    projection: Projection,
+    projected_live: broadcast::Sender<ProjectionUpdate>,
 }
 
 #[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<ProjectionUpdate>,
+        )>,
+        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(&notification) {
                     Ok(seq) => {
+                        let projected = events.projection.apply(seq, &notification);
                         let _ = events.live.send(LoggedEvent { seq, notification });
+                        let _ = events.projected_live.send(projected);
                         None
                     }
                     Err(error) => Some(error),