feat: show structured turn activity

crates/maestro-daemon/src/session.rs

@@ -657,6 +657,7 @@ impl SessionManager {
         let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
             return Ok(false);
         };
+        let projected = persist_turn_update(&mut stream, projected);
         let _ = stream.projected_live.send(projected);
         drop(stream);
         sync_structured_activity(&self.inner, id).await;
@@ -774,6 +775,7 @@ impl SessionManager {
                                 .expect(
                                     "newly claimed turn remains current while stream is locked",
                                 );
+                            let projected = persist_turn_update(&mut stream, projected);
                             let _ = stream.projected_live.send(projected);
                             drop(stream);
                             sync_structured_activity(&self.inner, id).await;
@@ -819,6 +821,7 @@ impl SessionManager {
         }
         cancel_pending_approvals(&mut stream);
         if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
+            let projected = persist_turn_update(&mut stream, projected);
             let _ = stream.projected_live.send(projected);
         }
         drop(stream);
@@ -935,6 +938,7 @@ impl SessionManager {
             .projection
             .turn_lifecycle(TurnLifecycle::Cancel, turn)
             .expect("active turn remains current while stream is locked");
+        let projected = persist_turn_update(&mut stream, projected);
         let _ = stream.projected_live.send(projected);
         drop(stream);
         sync_structured_activity(&self.inner, id).await;
@@ -2190,10 +2194,31 @@ 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::User { row } | PersistedEvent::Approval { row } => {
-            projection.restore_row(row.clone())
+        PersistedEvent::User { row }
+        | PersistedEvent::Approval { row }
+        | PersistedEvent::Turn { row } => projection.restore_row(row.clone()),
+    }
+}
+
+fn persist_turn_update(
+    stream: &mut AcpEventStream,
+    mut update: ProjectionUpdate,
+) -> ProjectionUpdate {
+    let Some(TranscriptDelta::Append { row }) = update.transcript.as_ref() else {
+        return update;
+    };
+    let event = PersistedEvent::Turn { row: row.clone() };
+    match stream.log.append(&event) {
+        Ok(seq) => {
+            let _ = stream.live.send(LoggedEvent { seq, event });
+        }
+        Err(error) => {
+            tracing::error!(%error, "turn metadata history append failed");
+            stream.projection.rollback_row(&row.id);
+            update.transcript = None;
         }
     }
+    update
 }
 
 async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {