docs(maestro): fix disabled-unstable-variant claim in STRUCTURED-VIEW.md

A SessionUpdate variant gated behind an unstable ACP feature this build
doesn't enable is compiled out of the enum, not decoded into a Notice.
It fails deserialization the same as a genuinely unknown future tag.

crates/maestro-daemon/src/projection.rs

@@ -438,6 +438,16 @@ mod tests {
         .unwrap()
     }
 
+    #[test]
+    fn a_session_update_tag_the_acp_dependency_does_not_know_fails_to_deserialize() {
+        // `SessionUpdate` has no unknown-variant fallback, so this never reaches `Projection::apply`.
+        let result: Result<SessionNotification, _> = serde_json::from_value(serde_json::json!({
+            "sessionId": "acp-1",
+            "update": {"sessionUpdate": "some_future_update_variant"},
+        }));
+        assert!(result.is_err());
+    }
+
     #[test]
     fn folds_content_tool_diff_plan_and_notice_rows() {
         let mut projection = Projection::default();

crates/maestro-daemon/src/session.rs

@@ -2003,6 +2003,41 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
 }
 
 async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
+    let events = {
+        let sessions = inner.sessions.lock().await;
+        let Some(record) = sessions.get(&id) else {
+            return;
+        };
+        if record.session.view != SessionView::Structured {
+            return;
+        }
+        record.acp_events.clone()
+    };
+
+    match events {
+        Some(events) => sync_structured_activity_from_stream(inner, id, events).await,
+        None => apply_structured_activity(inner, id, None, ActivityState::Error).await,
+    }
+}
+
+async fn sync_structured_activity_from_stream(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    events: Arc<Mutex<AcpEventStream>>,
+) {
+    let event_stream = events.lock().await;
+    let raw = event_stream.projection.activity();
+    // Stream state must stay fixed until its catalog transition is applied;
+    // every other path releases `sessions` before awaiting this stream lock.
+    apply_structured_activity(inner, id, Some(&events), raw).await;
+}
+
+async fn apply_structured_activity(
+    inner: &Arc<Inner>,
+    id: Uuid,
+    expected_events: Option<&Arc<Mutex<AcpEventStream>>>,
+    raw: ActivityState,
+) {
     let updated = {
         let mut sessions = inner.sessions.lock().await;
         let Some(record) = sessions.get_mut(&id) else {
@@ -2011,12 +2046,19 @@ async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
         if record.session.view != SessionView::Structured {
             return;
         }
+        let current_events = record.acp_events.as_ref();
+        let stream_matches = match (expected_events, current_events) {
+            (Some(expected), Some(current)) => Arc::ptr_eq(expected, current),
+            (None, None) => true,
+            _ => false,
+        };
+        if !stream_matches {
+            return;
+        }
         let raw = if record.session.status == SessionStatus::Dead {
             ActivityState::Error
-        } else if let Some(events) = &record.acp_events {
-            events.lock().await.projection.activity()
         } else {
-            ActivityState::Error
+            raw
         };
         let activity = crate::activity::apply_focus_suppression(
             raw,
@@ -2702,20 +2744,28 @@ async fn cleanup_record(
 #[cfg(test)]
 mod tests {
     use std::fs::File;
+    use std::sync::Arc;
+    use std::task::Poll;
+    use std::time::Duration;
 
     use maestro_acp::schema::{
         PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
         ToolCallUpdate, ToolCallUpdateFields,
     };
+    use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView};
 
     use super::{
-        cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState,
-        AcpEventStream, ApprovalDecision, RecordOutcome,
+        cancel_pending_approvals, permission_outcome, record_permission_request,
+        run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState,
+        AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager,
     };
+    use crate::activity::FOCUS_CLEAR_HOLD;
+    use crate::panes::FakePanes;
     use crate::projection::Projection;
     use crate::sidecar::{
         EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
     };
+    use crate::worktree::WorktreeManager;
 
     fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
         PermissionOption::new(id.to_owned(), id, kind)
@@ -2743,6 +2793,144 @@ mod tests {
         }
     }
 
+    fn terminal_request(cwd: &std::path::Path) -> CreateSessionRequest {
+        CreateSessionRequest {
+            view: SessionView::Terminal,
+            cwd: cwd.to_path_buf(),
+            command: "sh".into(),
+            env: std::collections::HashMap::new(),
+            metadata: std::collections::HashMap::new(),
+            worktree: None,
+        }
+    }
+
+    async fn manager_with_structured_stream() -> (
+        tempfile::TempDir,
+        SessionManager,
+        uuid::Uuid,
+        Arc<tokio::sync::Mutex<AcpEventStream>>,
+        uuid::Uuid,
+    ) {
+        let state = tempfile::tempdir().unwrap();
+        let manager = SessionManager::new(
+            Arc::new(FakePanes::new()),
+            SidecarStore::new(state.path().join("sessions")),
+            WorktreeManager::new(None),
+            state.path().join("fifos"),
+        );
+        let structured = manager
+            .create(terminal_request(state.path()))
+            .await
+            .expect("create structured fixture");
+        let terminal = manager
+            .create(terminal_request(state.path()))
+            .await
+            .expect("create unrelated terminal");
+        let events = Arc::new(tokio::sync::Mutex::new(stream(
+            manager
+                .inner
+                .sidecar
+                .open_events(structured.id)
+                .expect("open event log"),
+        )));
+        {
+            let mut event_stream = events.lock().await;
+            event_stream.projection.begin_turn();
+            event_stream
+                .projection
+                .request_approval("nonce".into(), serde_json::json!({}));
+        }
+        {
+            let mut sessions = manager.inner.sessions.lock().await;
+            let record = sessions.get_mut(&structured.id).unwrap();
+            record.session.view = SessionView::Structured;
+            record.session.activity = ActivityState::AwaitingInput;
+            record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
+            record.acp_events = Some(events.clone());
+        }
+        (state, manager, structured.id, events, terminal.id)
+    }
+
+    #[tokio::test]
+    async fn blocked_structured_activity_read_does_not_block_the_catalog() {
+        let (state, manager, structured_id, events, terminal_id) =
+            manager_with_structured_stream().await;
+        let held_stream = events.lock().await;
+        let activity_pass = run_structured_activity_pass(&manager.inner);
+        tokio::pin!(activity_pass);
+        assert!(matches!(
+            futures_util::poll!(activity_pass.as_mut()),
+            Poll::Pending
+        ));
+
+        tokio::time::timeout(Duration::from_secs(1), manager.snapshot())
+            .await
+            .expect("catalog snapshot blocked behind another session's event stream");
+        tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.send_keys(terminal_id, b"responsive"),
+        )
+        .await
+        .expect("terminal input blocked behind another session's event stream")
+        .expect("terminal input");
+        let created = tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.create(terminal_request(state.path())),
+        )
+        .await
+        .expect("session create blocked behind another session's event stream")
+        .expect("create unrelated session");
+        tokio::time::timeout(Duration::from_secs(1), manager.kill(created.id, false))
+            .await
+            .expect("session kill blocked behind another session's event stream")
+            .expect("kill unrelated session");
+
+        drop(held_stream);
+        tokio::time::timeout(Duration::from_secs(1), activity_pass)
+            .await
+            .expect("activity pass did not resume");
+        let structured = manager
+            .snapshot()
+            .await
+            .into_iter()
+            .find(|session| session.id == structured_id)
+            .unwrap();
+        assert_eq!(structured.activity, ActivityState::Idle);
+    }
+
+    #[tokio::test]
+    async fn delayed_activity_from_a_replaced_stream_is_discarded() {
+        let (_state, manager, structured_id, old_events, _terminal_id) =
+            manager_with_structured_stream().await;
+        let replacement = Arc::new(tokio::sync::Mutex::new(stream(
+            manager
+                .inner
+                .sidecar
+                .open_events(uuid::Uuid::new_v4())
+                .expect("open replacement event log"),
+        )));
+        let mut sessions = manager.inner.sessions.lock().await;
+        let sync =
+            sync_structured_activity_from_stream(&manager.inner, structured_id, old_events.clone());
+        tokio::pin!(sync);
+        assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending));
+        let record = sessions.get_mut(&structured_id).unwrap();
+        record.acp_events = Some(replacement.clone());
+        record.session.activity = ActivityState::Error;
+        drop(sessions);
+
+        tokio::time::timeout(Duration::from_secs(1), sync)
+            .await
+            .expect("stale activity sync did not finish");
+        let sessions = manager.inner.sessions.lock().await;
+        let record = sessions.get(&structured_id).unwrap();
+        assert_eq!(record.session.activity, ActivityState::Error);
+        assert!(Arc::ptr_eq(
+            record.acp_events.as_ref().unwrap(),
+            &replacement
+        ));
+    }
+
     /// A handle that can never append *or* roll back: opened read-only, so
     /// both `write_all` and the rollback `truncate` fail with the OS's own
     /// error (`ftruncate` requires a writable fd too) instead of a

docs/STRUCTURED-VIEW.md

@@ -113,13 +113,25 @@ crate.
 | `ToolDiff` | The same tool-call machinery, when the tool's content includes a `Diff` block. | Same `tool:<tool_call_id>` identity as `ToolCall`; the *kind* flips between the two as the tool's own content changes, not the row. |
 | `Approval` | A daemon-originated permission request (`RequestPermissionRequest`), serialized without its nonce. | `approval:<uuid>` — minted fresh per request, unrelated to the ACP request's own id. |
 | `Plan` | A `plan` update — the agent's current step list. | The fixed id `"plan"`. Each new plan **replaces** the row wholesale; entries are not diffed against the previous plan. |
-| `Notice` | The raw JSON of any `session/update` variant the projection does not special-case. | `notice:<seq>`. |
+| `Notice` | The raw JSON of any `session/update` variant the ACP dependency already decodes but the projection does not special-case. | `notice:<seq>`. |
 
-`Notice` is the projection's forward-compatibility seam: ACP adds
-notification variants over time, and an unrecognised one becomes a generic
-row instead of being silently dropped or making the daemon reject the whole
-stream. A client that does not know what to do with a `Notice` can render it
-as an opaque, collapsed line — that is a reasonable floor, not a client bug.
+`Notice` covers known-but-unspecialized `SessionUpdate` variants: ones the
+`agent-client-protocol` dependency's schema already deserializes (it
+currently folds `available_commands_update` and `session_info_update`
+there) but the projection's `match` doesn't fold into one of the other
+seven kinds. It is not a wire-level forward-compatibility seam:
+`SessionUpdate` is an internally-tagged enum with no unknown-variant
+fallback, so a `session/update` whose `sessionUpdate` tag this build's
+dependency has never heard of — including a variant gated behind an
+unstable ACP feature this build doesn't enable, which is compiled out of
+the enum entirely — fails to deserialize before the projection ever runs,
+rather than becoming a `Notice`. A genuinely new ACP variant needs the
+`agent-client-protocol` dependency upgraded (and, typically, the projection's
+`match` taught about it) before it can appear on this transcript at all — see
+`projection.rs`'s test asserting that an unrecognised `sessionUpdate` tag is a
+deserialization error, not a `Notice`. A client that does not know what to do
+with a `Notice` can still render it as an opaque, collapsed line — that
+remains a reasonable floor, not a client bug.
 
 Rendering is entirely a client concern. The projection's job stops at
 folding ACP into these eight shapes; markdown formatting, diff rendering,
@@ -139,8 +151,11 @@ choices, not daemon ones.
 - Approval nonce minting, secrecy, and the single-active-approval rule (see
   below) — a client never sees an ACP permission request before the daemon
   has already turned it into an `Approval` row and a nonce.
-- What happens to an ACP update variant nobody has taught the daemon about
-  yet (falls back to `Notice` rather than erroring).
+- What happens to an ACP update variant the dependency decodes but nobody has
+  taught the projection to special-case (falls back to `Notice` rather than
+  erroring). A variant the dependency itself can't decode is a deserialization
+  error, not a `Notice` — see [The eight transcript row
+  kinds](#the-eight-transcript-row-kinds).
 - Persistence and replay: what is written to the event log, in what order,
   and how a restart resumes it.