fix: harden structured replay

crates/maestro-daemon/src/session.rs

@@ -15,7 +15,7 @@ use chrono::Utc;
 use maestro_acp::{AcpSession, SpawnConfig};
 use maestro_protocol::{
     ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
-    SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
+    SessionStatus, SessionView, StructuredReplayEvent, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
 };
 use tokio::sync::{broadcast, watch, Mutex};
 use uuid::Uuid;
@@ -357,6 +357,60 @@ impl SessionManager {
             .map_err(SessionError::Sidecar)
     }
 
+    /// Replay durable structured-session data as client-neutral projections.
+    ///
+    /// The complete log is folded before selecting the requested cursor range,
+    /// so each returned state is exactly the state at its cursor.
+    pub async fn read_acp_projection_replay(
+        &self,
+        id: Uuid,
+        cursor: EventCursor,
+        limit: usize,
+    ) -> Result<Option<Vec<StructuredReplayEvent>>, SessionError> {
+        let stream = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if record.session.view != SessionView::Structured {
+                return Ok(None);
+            }
+            record.acp_events.clone()
+        };
+        let Some(stream) = stream else {
+            return Ok(None);
+        };
+        let stream = stream.lock().await;
+        let events = self
+            .inner
+            .sidecar
+            .read_events(id, EventCursor::Since(0), usize::MAX)?;
+        let mut projection = Projection::default();
+        let mut replay = Vec::new();
+        for event in events {
+            let update = projection.apply(event.seq, &event.notification);
+            let include = match cursor {
+                EventCursor::Since(since) => event.seq > since,
+                EventCursor::Before(Some(before)) => event.seq < before,
+                EventCursor::Before(None) => true,
+            };
+            if include {
+                replay.push(StructuredReplayEvent {
+                    cursor: event.seq,
+                    transcript: update.transcript,
+                    state: update.state,
+                });
+            }
+        }
+        drop(stream);
+        match cursor {
+            EventCursor::Since(_) => replay.truncate(limit),
+            EventCursor::Before(_) if replay.len() > limit => {
+                replay.drain(..replay.len() - limit);
+            }
+            _ => {}
+        }
+        Ok(Some(replay))
+    }
+
     /// Subscribe to the two projections of a structured session's ACP log.
     ///
     /// The transcript snapshot and whole reduced state are folded once by the
@@ -402,6 +456,7 @@ impl SessionManager {
         &self,
         id: Uuid,
         lifecycle: TurnLifecycle,
+        turn: u64,
     ) -> Result<bool, SessionError> {
         let stream = {
             let sessions = self.inner.sessions.lock().await;
@@ -412,14 +467,19 @@ impl SessionManager {
             return Ok(false);
         };
         let mut stream = stream.lock().await;
-        let projected = stream.projection.turn_lifecycle(lifecycle);
+        let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
+            return Ok(false);
+        };
         let _ = stream.projected_live.send(projected);
         Ok(true)
     }
 
     /// Begin an ACP turn only when the previous one has ended. The projection
     /// lock makes the rejection and lifecycle update one indivisible action.
-    pub async fn begin_acp_turn(&self, id: Uuid) -> Result<Option<Arc<AcpSession>>, SessionError> {
+    pub async fn begin_acp_turn(
+        &self,
+        id: Uuid,
+    ) -> Result<Option<(Arc<AcpSession>, u64)>, SessionError> {
         let (acp, stream) = {
             let sessions = self.inner.sessions.lock().await;
             let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
@@ -432,24 +492,32 @@ impl SessionManager {
         if stream.projection.turn_active() {
             return Ok(None);
         }
-        let update = stream.projection.turn_lifecycle(TurnLifecycle::Begin);
+        let (turn, update) = stream.projection.begin_turn();
         let _ = stream.projected_live.send(update);
-        Ok(Some(acp))
+        Ok(Some((acp, turn)))
     }
 
     /// Cancel an ACP turn and immediately clear its client-facing lifecycle.
     pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
-        let acp = {
+        let (acp, stream) = {
             let sessions = self.inner.sessions.lock().await;
             let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
-            record.acp.clone()
+            (record.acp.clone(), record.acp_events.clone())
         };
-        let Some(acp) = acp else {
+        let (Some(acp), Some(stream)) = (acp, stream) else {
             return Ok(false);
         };
+        let mut stream = stream.lock().await;
+        let Some(turn) = stream.projection.active_turn() else {
+            return Ok(true);
+        };
         acp.cancel();
-        self.apply_acp_turn_lifecycle(id, TurnLifecycle::Cancel)
-            .await
+        let projected = stream
+            .projection
+            .turn_lifecycle(TurnLifecycle::Cancel, turn)
+            .expect("active turn remains current while stream is locked");
+        let _ = stream.projected_live.send(projected);
+        Ok(true)
     }
 
     /// End every adapter child before the daemon exits (#134): each gets