diff --git a/crates/lash-core/src/runtime/lifecycle.rs b/crates/lash-core/src/runtime/lifecycle.rs
index 4362197a1..0b25a292d 100644
--- a/crates/lash-core/src/runtime/lifecycle.rs
+++ b/crates/lash-core/src/runtime/lifecycle.rs
@@ -178,6 +178,7 @@ impl LashRuntime {
             process_sync_needed: Arc::new(AtomicBool::new(false)),
             turn_phase_probe: None,
             last_committed_lease_continuity: None,
+            last_committed_observation_turn: None,
             graph_loaded_from_store: false,
             residency: Residency::default(),
         })
diff --git a/crates/lash-core/src/runtime/mod.rs b/crates/lash-core/src/runtime/mod.rs
index 0a03587f0..073d8a8f2 100644
--- a/crates/lash-core/src/runtime/mod.rs
+++ b/crates/lash-core/src/runtime/mod.rs
@@ -905,6 +905,17 @@ pub trait TurnActivitySink: Send + Sync {
     }
 
     async fn emit(&self, activity: TurnActivity);
+
+    /// Emit activity with the identity of the physical turn that produced it.
+    ///
+    /// Sinks that only consume turn-local activity can keep implementing
+    /// [`emit`](Self::emit). Observation sinks override this method to carry
+    /// turn identity on their enclosing event without adding it to
+    /// [`TurnActivity`].
+    async fn emit_for_turn(&self, turn_id: &str, activity: TurnActivity) {
+        let _ = turn_id;
+        self.emit(activity).await;
+    }
 }
 
 pub struct NoopTurnActivitySink;
@@ -1092,6 +1103,9 @@ pub struct LashRuntime {
     /// next physical turn on this handle.
     pub(in crate::runtime) last_committed_lease_continuity:
         Option<session_execution_lease::SessionExecutionLeaseContinuity>,
+    /// Most recent physical turn committed by this runtime, paired with the
+    /// resulting session revision for observation-envelope attribution.
+    pub(in crate::runtime) last_committed_observation_turn: Option<(u64, String)>,
     /// Set only after this handle itself has attempted a durable graph load.
     pub(in crate::runtime) graph_loaded_from_store: bool,
     /// Resident-graph policy chosen by the host. Controls whether
diff --git a/crates/lash-core/src/runtime/observation.rs b/crates/lash-core/src/runtime/observation.rs
index 6319e44a5..e4f07c327 100644
--- a/crates/lash-core/src/runtime/observation.rs
+++ b/crates/lash-core/src/runtime/observation.rs
@@ -290,12 +290,16 @@ impl RuntimeHandle {
     pub fn publish_from(&self, runtime: &LashRuntime) {
         let revision = SessionRevision::from_runtime(runtime);
         let previous = self.observation.load_full();
+        let turn_id = (previous.revision != revision)
+            .then(|| runtime.last_committed_turn_id_for_revision(revision))
+            .flatten();
         let (state, read_view, usage_report) = export_observation_state(runtime);
         if previous.persisted_state.current_agent_frame_id != state.current_agent_frame_id
             && !state.current_agent_frame_id.is_empty()
             && let Err(err) = self.live_replay_store.append(
                 runtime.session_id(),
                 revision,
+                None,
                 SessionObservationEventPayload::AgentFrameSwitched {
                     frame_id: state.current_agent_frame_id.clone(),
                 },
@@ -310,6 +314,7 @@ impl RuntimeHandle {
         let cursor = match self.live_replay_store.append(
             runtime.session_id(),
             revision,
+            turn_id,
             SessionObservationEventPayload::Committed {
                 read_view: read_view.clone(),
             },
@@ -337,11 +342,12 @@ impl RuntimeHandle {
             )));
     }
 
-    pub fn record_turn_activity(&self, activity: crate::TurnActivity) {
+    pub fn record_turn_activity(&self, turn_id: Option<&str>, activity: crate::TurnActivity) {
         let observation = self.observe();
         if let Err(err) = self.live_replay_store.append(
             observation.session_id(),
             observation.session_revision(),
+            turn_id,
             SessionObservationEventPayload::TurnActivity(activity),
         ) {
             tracing::warn!(
@@ -357,6 +363,7 @@ impl RuntimeHandle {
         if let Err(err) = self.live_replay_store.append(
             observation.session_id(),
             observation.session_revision(),
+            None,
             SessionObservationEventPayload::QueueChanged { kind, batch_ids },
         ) {
             tracing::warn!(
@@ -372,6 +379,7 @@ impl RuntimeHandle {
         if let Err(err) = self.live_replay_store.append(
             observation.session_id(),
             observation.session_revision(),
+            None,
             SessionObservationEventPayload::ProcessChanged { kind, process_ids },
         ) {
             tracing::warn!(
@@ -671,6 +679,15 @@ impl RuntimeHandle {
     }
 }
 
+impl LashRuntime {
+    fn last_committed_turn_id_for_revision(&self, revision: SessionRevision) -> Option<&str> {
+        self.last_committed_observation_turn
+            .as_ref()
+            .filter(|(committed_revision, _)| *committed_revision == revision.as_u64())
+            .map(|(_, turn_id)| turn_id.as_str())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -683,6 +700,7 @@ mod tests {
             &self,
             _session_id: &str,
             _revision: SessionRevision,
+            _turn_id: Option<&str>,
             _payload: SessionObservationEventPayload,
         ) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError> {
             panic!("append should not be called by cursor rejection tests")
@@ -825,10 +843,12 @@ mod tests {
             events[0].payload,
             SessionObservationEventPayload::AgentFrameSwitched { .. }
         ));
+        assert_eq!(events[0].turn_id, None);
         assert!(matches!(
             events[1].payload,
             SessionObservationEventPayload::Committed { .. }
         ));
+        assert_eq!(events[1].turn_id, None);
     }
 
     #[tokio::test]
diff --git a/crates/lash-core/src/runtime/observation/replay.rs b/crates/lash-core/src/runtime/observation/replay.rs
index d2160fcdf..859942fd9 100644
--- a/crates/lash-core/src/runtime/observation/replay.rs
+++ b/crates/lash-core/src/runtime/observation/replay.rs
@@ -166,6 +166,8 @@ pub struct SessionObservation {
 #[derive(Clone, Debug)]
 pub struct SessionObservationEvent {
     pub session_id: String,
+    pub replay_incarnation_id: String,
+    pub turn_id: Option<String>,
     pub revision: SessionRevision,
     pub cursor: SessionCursor,
     pub payload: SessionObservationEventPayload,
@@ -358,6 +360,7 @@ pub trait LiveReplayStore: Send + Sync {
         &self,
         session_id: &str,
         revision: SessionRevision,
+        turn_id: Option<&str>,
         payload: SessionObservationEventPayload,
     ) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError>;
 
@@ -405,6 +408,7 @@ impl Default for InMemoryLiveReplayStoreConfig {
 
 #[derive(Debug)]
 pub struct InMemoryLiveReplayStore {
+    replay_incarnation_id: String,
     config: InMemoryLiveReplayStoreConfig,
     clock: Arc<dyn crate::Clock>,
     sessions: StdMutex<HashMap<String, LiveReplaySessionBuffer>>,
@@ -417,6 +421,7 @@ impl InMemoryLiveReplayStore {
 
     pub fn with_clock(config: InMemoryLiveReplayStoreConfig, clock: Arc<dyn crate::Clock>) -> Self {
         Self {
+            replay_incarnation_id: uuid::Uuid::new_v4().to_string(),
             config,
             clock,
             sessions: StdMutex::new(HashMap::new()),
@@ -529,6 +534,7 @@ impl LiveReplayStore for InMemoryLiveReplayStore {
         &self,
         session_id: &str,
         revision: SessionRevision,
+        turn_id: Option<&str>,
         payload: SessionObservationEventPayload,
     ) -> Result<Arc<SessionObservationEvent>, LiveReplayStoreError> {
         let now = self.clock.now();
@@ -543,6 +549,8 @@ impl LiveReplayStore for InMemoryLiveReplayStore {
         let cursor = SessionCursor::new(session_id, revision, buffer.tail_position);
         let event = Arc::new(SessionObservationEvent {
             session_id: session_id.to_string(),
+            replay_incarnation_id: self.replay_incarnation_id.clone(),
+            turn_id: turn_id.map(str::to_string),
             revision,
             cursor,
             payload,
@@ -708,10 +716,10 @@ mod tests {
         let store = InMemoryLiveReplayStore::default();
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         store
-            .append("s", SessionRevision(0), activity("b"))
+            .append("s", SessionRevision(0), None, activity("b"))
             .expect("append b");
         let LiveReplayResult::Replayed(events) = store.replay_after_cursor(&start).expect("replay")
         else {
@@ -732,10 +740,10 @@ mod tests {
         let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         store
-            .append("s", SessionRevision(0), activity("b"))
+            .append("s", SessionRevision(0), None, activity("b"))
             .expect("append b");
         assert!(matches!(
             store.replay_after_cursor(&start).expect("gap"),
@@ -748,7 +756,7 @@ mod tests {
         let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         std::thread::sleep(Duration::from_millis(5));
         assert!(matches!(
@@ -772,7 +780,7 @@ mod tests {
         let store = InMemoryLiveReplayStore::default();
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         let LiveReplaySubscribeResult::Subscribed(mut subscription) =
             store.subscribe_after_cursor(&start).expect("subscribe")
@@ -782,7 +790,7 @@ mod tests {
         let first = subscription.next_event().await.expect("replay");
         assert_eq!(first.session_id, "s");
         store
-            .append("s", SessionRevision(0), activity("b"))
+            .append("s", SessionRevision(0), None, activity("b"))
             .expect("append b");
         let second = subscription.next_event().await.expect("live");
         match &second.payload {
@@ -815,6 +823,7 @@ mod tests {
                 .append(
                     "perf-session",
                     SessionRevision(7),
+                    None,
                     activity(&format!("token-{ordinal}")),
                 )
                 .expect("append token event");
@@ -845,7 +854,7 @@ mod tests {
         let store = InMemoryLiveReplayStore::default();
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         {
             let sessions = store.sessions.lock().expect("sessions");
@@ -862,7 +871,7 @@ mod tests {
         }
         drop(subscription);
         store
-            .append("s", SessionRevision(0), activity("b"))
+            .append("s", SessionRevision(0), None, activity("b"))
             .expect("append b");
         let sessions = store.sessions.lock().expect("sessions");
         assert!(sessions.get("s").expect("buffer").sender.is_none());
@@ -873,10 +882,10 @@ mod tests {
         let store = InMemoryLiveReplayStore::with_bounds(1, Duration::from_secs(120));
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         store
-            .append("s", SessionRevision(0), activity("b"))
+            .append("s", SessionRevision(0), None, activity("b"))
             .expect("append b");
         assert!(matches!(
             store.subscribe_after_cursor(&start).expect("subscribe"),
@@ -889,7 +898,7 @@ mod tests {
         let store = InMemoryLiveReplayStore::with_bounds(16, Duration::from_millis(1));
         let start = store.current_cursor("s", SessionRevision(0));
         store
-            .append("s", SessionRevision(0), activity("a"))
+            .append("s", SessionRevision(0), None, activity("a"))
             .expect("append a");
         std::thread::sleep(Duration::from_millis(5));
         assert!(matches!(
diff --git a/crates/lash-core/src/runtime/turn_loop.rs b/crates/lash-core/src/runtime/turn_loop.rs
index 6b0c59506..ffb588fee 100644
--- a/crates/lash-core/src/runtime/turn_loop.rs
+++ b/crates/lash-core/src/runtime/turn_loop.rs
@@ -194,12 +194,14 @@ pub(in crate::runtime) fn turn_input_completion_trace_payload(
 
 async fn emit_queued_work_started_to_sink(
     events: &dyn TurnActivitySink,
+    turn_id: &str,
     boundary: crate::QueuedWorkClaimBoundary,
     claim: &crate::QueuedWorkClaim,
     causes: Vec<crate::TurnCause>,
 ) {
-    emit_turn_activity_to_sink(
+    emit_turn_activity_to_sink_for_turn(
         events,
+        turn_id,
         TurnActivity::independent(TurnEvent::QueuedWorkStarted {
             boundary,
             batch_ids: queued_work_batch_ids(claim),
@@ -452,6 +454,12 @@ impl LashRuntime {
 
         let Some(session) = self.session.as_ref() else {
             self.state.apply_snapshot(&assembled.state);
+            self.last_committed_observation_turn = Some((
+                self.state
+                    .head_revision
+                    .unwrap_or(self.state.turn_index as u64),
+                trace_turn_id.clone(),
+            ));
             self.emit_completed_turn_trace(&assembled.state, &assembled.outcome, &trace_turn_id);
             publish_terminal_after_commit(
                 turn_control,
@@ -576,6 +584,12 @@ impl LashRuntime {
 
         emit_session_events_to_sink(events, finalized.events).await;
         self.state = turn_pipeline.into_final_state();
+        self.last_committed_observation_turn = Some((
+            self.state
+                .head_revision
+                .unwrap_or(self.state.turn_index as u64),
+            trace_turn_id.clone(),
+        ));
         publish_terminal_after_commit(
             turn_control,
             turn_control_resolver,
@@ -768,8 +782,9 @@ impl LashRuntime {
             }),
         };
         assembler.push(&error_event);
-        emit_turn_activity_to_sink(
+        emit_turn_activity_to_sink_for_turn(
             turn_events,
+            &trace_turn_id,
             TurnActivity::independent(TurnEvent::Error {
                 message: message.clone(),
             }),
@@ -1057,6 +1072,7 @@ impl LashRuntime {
         let causes = work.turn_causes.clone();
         emit_queued_work_started_to_sink(
             opts.turn_events_or_noop(),
+            &turn_id,
             crate::QueuedWorkClaimBoundary::Idle,
             &claim,
             causes.clone(),
@@ -1384,8 +1400,13 @@ impl LashRuntime {
                     }),
                 };
                 assembler.push(&error_event);
-                emit_turn_activity_to_sink(
+                let trace_turn_id = input
+                    .trace_turn_id
+                    .clone()
+                    .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
+                emit_turn_activity_to_sink_for_turn(
                     turn_events,
+                    &trace_turn_id,
                     TurnActivity::independent(TurnEvent::Error { message: e }),
                 )
                 .await;
@@ -1398,10 +1419,6 @@ impl LashRuntime {
                 assembler.push(&SessionStreamEvent::Done);
                 emit_session_event_to_sink(events, SessionStreamEvent::Done).await;
                 let turn_index = self.state.turn_index + 1;
-                let trace_turn_id = input
-                    .trace_turn_id
-                    .clone()
-                    .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
                 let turn_control_host = Arc::clone(&self.host.core.control.effect_host);
                 let turn_control_resolver =
                     turn_control_resolver(turn_control_host.as_ref(), &scoped_effect_controller);
@@ -1550,8 +1567,9 @@ impl LashRuntime {
             initial_turn_input_applications.extend(claim.applications.clone());
         }
         if !initial_turn_input_applications.is_empty() {
-            emit_turn_activity_to_sink(
+            emit_turn_activity_to_sink_for_turn(
                 turn_events,
+                &trace_turn_id,
                 TurnActivity::independent(TurnEvent::QueuedInputAccepted {
                     applications: initial_turn_input_applications,
                 }),
@@ -1888,6 +1906,11 @@ impl LashRuntime {
         session_execution_lease: Option<&SessionExecutionLeaseGuard>,
         session_execution_lease_release_policy: SessionExecutionLeaseReleasePolicy,
     ) -> Result<PhysicalTurnExecution, RuntimeError> {
+        let scoped_turn_events = TurnScopedActivitySink {
+            turn_id: trace_turn_id.clone(),
+            inner: turn_events,
+        };
+        let turn_events: &dyn TurnActivitySink = &scoped_turn_events;
         let turn_control_host = Arc::clone(&self.host.core.control.effect_host);
         let turn_control_resolver =
             turn_control_resolver(turn_control_host.as_ref(), &scoped_effect_controller);
@@ -2212,6 +2235,32 @@ async fn emit_turn_activity_to_sink(events: &dyn TurnActivitySink, activity: Tur
     }
 }
 
+async fn emit_turn_activity_to_sink_for_turn(
+    events: &dyn TurnActivitySink,
+    turn_id: &str,
+    activity: TurnActivity,
+) {
+    if !events.is_noop() {
+        events.emit_for_turn(turn_id, activity).await;
+    }
+}
+
+struct TurnScopedActivitySink<'a> {
+    turn_id: String,
+    inner: &'a dyn TurnActivitySink,
+}
+
+#[async_trait::async_trait]
+impl TurnActivitySink for TurnScopedActivitySink<'_> {
+    fn is_noop(&self) -> bool {
+        self.inner.is_noop()
+    }
+
+    async fn emit(&self, activity: TurnActivity) {
+        self.inner.emit_for_turn(&self.turn_id, activity).await;
+    }
+}
+
 async fn publish_terminal_after_commit(
     turn_control: &ActiveTurnControl,
     resolver: &dyn AwaitEventResolver,
diff --git a/crates/lash-core/src/testing/conformance/live_replay.rs b/crates/lash-core/src/testing/conformance/live_replay.rs
index d8e9b1fa5..5250b0d4d 100644
--- a/crates/lash-core/src/testing/conformance/live_replay.rs
+++ b/crates/lash-core/src/testing/conformance/live_replay.rs
@@ -37,6 +37,7 @@ where
         .append(
             "capacity-session",
             revision,
+            Some("capacity-turn"),
             live_replay_text_payload("capacity one"),
         )
         .expect("append first capacity event");
@@ -44,6 +45,7 @@ where
         .append(
             "capacity-session",
             revision,
+            Some("capacity-turn"),
             live_replay_text_payload("capacity two"),
         )
         .expect("append second capacity event");
@@ -81,6 +83,7 @@ where
         .append(
             "ttl-session",
             revision,
+            Some("ttl-turn"),
             live_replay_text_payload("ttl expired"),
         )
         .expect("append ttl event");
@@ -120,12 +123,18 @@ async fn live_replay_store_appends_replays_and_isolates_sessions(store: Arc<dyn
     assert!(empty.is_empty(), "initial cursor must replay no events");
 
     let first_a = store
-        .append("session-a", revision, live_replay_text_payload("alpha one"))
+        .append(
+            "session-a",
+            revision,
+            Some("alpha-turn"),
+            live_replay_text_payload("alpha one"),
+        )
         .expect("append first session-a event");
     let first_b = store
         .append(
             "session-b",
             revision,
+            None,
             SessionObservationEventPayload::ProcessChanged {
                 kind: SessionProcessEventKind::Started,
                 process_ids: vec!["proc-b".to_string()],
@@ -136,6 +145,7 @@ async fn live_replay_store_appends_replays_and_isolates_sessions(store: Arc<dyn
         .append(
             "session-a",
             SessionRevision::new(8),
+            None,
             SessionObservationEventPayload::QueueChanged {
                 kind: SessionQueueEventKind::Enqueued,
                 batch_ids: vec!["batch-a".to_string()],
@@ -145,7 +155,18 @@ async fn live_replay_store_appends_replays_and_isolates_sessions(store: Arc<dyn
 
     assert_eq!(first_a.session_id, "session-a");
     assert_eq!(first_a.revision, revision);
+    assert_eq!(first_a.turn_id.as_deref(), Some("alpha-turn"));
     assert_eq!(second_a.revision, SessionRevision::new(8));
+    assert_eq!(first_b.turn_id, None);
+    assert_eq!(second_a.turn_id, None);
+    assert_eq!(
+        first_a.replay_incarnation_id, second_a.replay_incarnation_id,
+        "one store construction must stamp one stable replay incarnation"
+    );
+    assert_eq!(
+        first_a.replay_incarnation_id, first_b.replay_incarnation_id,
+        "the replay incarnation is store-scoped rather than session-scoped"
+    );
     assert_ne!(
         first_a.cursor.as_str(),
         second_a.cursor.as_str(),
@@ -187,6 +208,7 @@ async fn live_replay_store_subscribe_replays_then_yields_live_events(
         .append(
             "subscribe-session",
             revision,
+            Some("subscribe-turn"),
             live_replay_text_payload("buffered one"),
         )
         .expect("append first buffered event");
@@ -194,6 +216,7 @@ async fn live_replay_store_subscribe_replays_then_yields_live_events(
         .append(
             "subscribe-session",
             revision,
+            Some("subscribe-turn"),
             live_replay_text_payload("buffered two"),
         )
         .expect("append second buffered event");
@@ -213,6 +236,7 @@ async fn live_replay_store_subscribe_replays_then_yields_live_events(
         .append(
             "subscribe-session",
             revision,
+            Some("subscribe-turn"),
             live_replay_text_payload("live three"),
         )
         .expect("append live event");
@@ -252,6 +276,7 @@ async fn live_replay_store_reports_unavailable_for_cursor_ahead_of_tail(
         .append(
             "ahead-session",
             revision,
+            Some("ahead-turn"),
             live_replay_text_payload("existing"),
         )
         .expect("append existing event");
diff --git a/crates/lash-perf/src/runtime_perf/measurement/live_replay.rs b/crates/lash-perf/src/runtime_perf/measurement/live_replay.rs
index 6c4061aef..fa2b82bc2 100644
--- a/crates/lash-perf/src/runtime_perf/measurement/live_replay.rs
+++ b/crates/lash-perf/src/runtime_perf/measurement/live_replay.rs
@@ -39,6 +39,7 @@ async fn run_once_live_replay_pressure(chat_turns: usize) -> anyhow::Result<Runt
         let mut phase_profile = BTreeMap::new();
         let session_id = format!("runtime-perf-live-replay-{turn_index}");
         let revision = SessionRevision::new(turn_index as u64 + 1);
+        let turn_id = format!("turn-{turn_index}");
         let start_cursor = store.current_cursor(&session_id, revision);
 
         let (first_cursor, append_phase) =
@@ -48,6 +49,7 @@ async fn run_once_live_replay_pressure(chat_turns: usize) -> anyhow::Result<Runt
                     let event = store.append(
                         &session_id,
                         revision,
+                        Some(&turn_id),
                         live_replay_text_payload(format!("turn-{turn_index}-event-{event_index}")),
                     )?;
                     if first_cursor.is_none() {
@@ -111,6 +113,7 @@ async fn run_once_live_replay_pressure(chat_turns: usize) -> anyhow::Result<Runt
                 store.append(
                     &session_id,
                     revision,
+                    Some(&turn_id),
                     live_replay_text_payload(format!("turn-{turn_index}-live-event")),
                 )?;
                 tokio::time::timeout(Duration::from_secs(1), subscription.next_event())
@@ -130,11 +133,13 @@ async fn run_once_live_replay_pressure(chat_turns: usize) -> anyhow::Result<Runt
                     Duration::from_secs(120),
                 );
                 let trim_session_id = format!("runtime-perf-live-replay-trim-{turn_index}");
+                let trim_turn_id = format!("trim-turn-{turn_index}");
                 let trim_start = trim_store.current_cursor(&trim_session_id, revision);
                 for event_index in 0..(LIVE_REPLAY_TRIM_CAPACITY * 3) {
                     trim_store.append(
                         &trim_session_id,
                         revision,
+                        Some(&trim_turn_id),
                         live_replay_text_payload(format!("trim-{turn_index}-{event_index}")),
                     )?;
                 }
diff --git a/crates/lash-remote-protocol/src/core_conversions/observations.rs b/crates/lash-remote-protocol/src/core_conversions/observations.rs
index 2696847d5..08b2c8c2a 100644
--- a/crates/lash-remote-protocol/src/core_conversions/observations.rs
+++ b/crates/lash-remote-protocol/src/core_conversions/observations.rs
@@ -103,6 +103,8 @@ impl RemoteSessionObservationEvent {
     pub fn from_core(sequence: u64, event: Arc<lash_core::SessionObservationEvent>) -> Self {
         let lash_core::SessionObservationEvent {
             session_id,
+            replay_incarnation_id,
+            turn_id,
             revision,
             cursor,
             payload,
@@ -139,6 +141,8 @@ impl RemoteSessionObservationEvent {
         Self {
             protocol_version: REMOTE_PROTOCOL_VERSION,
             session_id: session_id.clone(),
+            replay_incarnation_id: replay_incarnation_id.clone(),
+            turn_id: turn_id.clone(),
             revision: revision.as_u64(),
             cursor: cursor.to_string(),
             event: payload,
diff --git a/crates/lash-remote-protocol/src/core_conversions_tests.rs b/crates/lash-remote-protocol/src/core_conversions_tests.rs
index 764ef3e2f..b25553a13 100644
--- a/crates/lash-remote-protocol/src/core_conversions_tests.rs
+++ b/crates/lash-remote-protocol/src/core_conversions_tests.rs
@@ -1016,6 +1016,7 @@ fn remote_session_observation_from_core_maps_snapshot_metadata() {
         &store,
         "session",
         lash_core::SessionRevision::new(4),
+        None,
         lash_core::SessionObservationEventPayload::QueueChanged {
             kind: lash_core::SessionQueueEventKind::Enqueued,
             batch_ids: vec!["batch-1".to_string()],
@@ -1055,6 +1056,7 @@ fn remote_session_observation_from_core_maps_snapshot_metadata() {
 #[test]
 fn remote_session_observation_from_core_maps_all_payload_variants() {
     fn event(
+        turn_id: Option<&str>,
         payload: lash_core::SessionObservationEventPayload,
     ) -> Arc<lash_core::SessionObservationEvent> {
         let store = lash_core::InMemoryLiveReplayStore::default();
@@ -1062,6 +1064,7 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
             &store,
             "session",
             lash_core::SessionRevision::new(4),
+            turn_id,
             payload,
         )
         .expect("append observation event")
@@ -1073,11 +1076,18 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
         });
     let remote = RemoteSessionObservationEvent::from_core(
         7,
-        event(lash_core::SessionObservationEventPayload::TurnActivity(
-            activity,
-        )),
+        event(
+            Some("activity-turn"),
+            lash_core::SessionObservationEventPayload::TurnActivity(activity),
+        ),
     );
-    match remote.event {
+    assert_eq!(remote.turn_id.as_deref(), Some("activity-turn"));
+    assert!(!remote.replay_incarnation_id.is_empty());
+    let encoded = serde_json::to_value(&remote).expect("serialize activity envelope");
+    let decoded: RemoteSessionObservationEvent =
+        serde_json::from_value(encoded).expect("deserialize activity envelope");
+    assert_eq!(decoded, remote);
+    match &remote.event {
         RemoteSessionObservationEventPayload::TurnActivity { activity } => {
             assert_eq!(activity.sequence, 7);
         }
@@ -1088,8 +1098,12 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
         lash_core::SessionReadView::from_snapshot(&lash_core::SessionSnapshot::default());
     let remote = RemoteSessionObservationEvent::from_core(
         8,
-        event(lash_core::SessionObservationEventPayload::Committed { read_view }),
+        event(
+            Some("committed-turn"),
+            lash_core::SessionObservationEventPayload::Committed { read_view },
+        ),
     );
+    assert_eq!(remote.turn_id.as_deref(), Some("committed-turn"));
     assert!(matches!(
         remote.event,
         RemoteSessionObservationEventPayload::Committed
@@ -1098,11 +1112,21 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
     let remote = RemoteSessionObservationEvent::from_core(
         9,
         event(
+            None,
             lash_core::SessionObservationEventPayload::AgentFrameSwitched {
                 frame_id: "frame-1".to_string(),
             },
         ),
     );
+    assert_eq!(remote.turn_id, None);
+    let encoded = serde_json::to_value(&remote).expect("serialize frame envelope");
+    assert!(
+        encoded.get("turn_id").is_none(),
+        "absent turn identity must be omitted: {encoded}"
+    );
+    let decoded: RemoteSessionObservationEvent =
+        serde_json::from_value(encoded).expect("deserialize frame envelope");
+    assert_eq!(decoded, remote);
     assert!(matches!(
         remote.event,
         RemoteSessionObservationEventPayload::AgentFrameSwitched { frame_id }
@@ -1111,11 +1135,15 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
 
     let remote = RemoteSessionObservationEvent::from_core(
         10,
-        event(lash_core::SessionObservationEventPayload::QueueChanged {
-            kind: lash_core::SessionQueueEventKind::Cancelled,
-            batch_ids: vec!["batch-1".to_string()],
-        }),
+        event(
+            None,
+            lash_core::SessionObservationEventPayload::QueueChanged {
+                kind: lash_core::SessionQueueEventKind::Cancelled,
+                batch_ids: vec!["batch-1".to_string()],
+            },
+        ),
     );
+    assert_eq!(remote.turn_id, None);
     assert!(matches!(
         remote.event,
         RemoteSessionObservationEventPayload::QueueChanged { kind, batch_ids }
@@ -1125,11 +1153,15 @@ fn remote_session_observation_from_core_maps_all_payload_variants() {
 
     let remote = RemoteSessionObservationEvent::from_core(
         11,
-        event(lash_core::SessionObservationEventPayload::ProcessChanged {
-            kind: lash_core::SessionProcessEventKind::Started,
-            process_ids: vec!["process-1".to_string()],
-        }),
+        event(
+            None,
+            lash_core::SessionObservationEventPayload::ProcessChanged {
+                kind: lash_core::SessionProcessEventKind::Started,
+                process_ids: vec!["process-1".to_string()],
+            },
+        ),
     );
+    assert_eq!(remote.turn_id, None);
     assert!(matches!(
         remote.event,
         RemoteSessionObservationEventPayload::ProcessChanged { kind, process_ids }
diff --git a/crates/lash-remote-protocol/src/lib.rs b/crates/lash-remote-protocol/src/lib.rs
index 89c7e66ba..bdd526d2c 100644
--- a/crates/lash-remote-protocol/src/lib.rs
+++ b/crates/lash-remote-protocol/src/lib.rs
@@ -32,9 +32,9 @@ pub use turn_input::*;
 pub use turn_result::*;
 pub use usage_activity::*;
 
-// Bumped to 16: queued turn-input application is a typed observation with
-// stable admission, turn, and committed-message identity.
-pub const REMOTE_PROTOCOL_VERSION: u32 = 16;
+// Bumped to 17: session observation envelopes carry replay-store incarnation
+// and optional turn identity.
+pub const REMOTE_PROTOCOL_VERSION: u32 = 17;
 
 pub fn ensure_protocol_version(actual: u32) -> Result<(), RemoteProtocolError> {
     if actual == REMOTE_PROTOCOL_VERSION {
diff --git a/crates/lash-remote-protocol/src/observations.rs b/crates/lash-remote-protocol/src/observations.rs
index 1e4f44a07..eb2312510 100644
--- a/crates/lash-remote-protocol/src/observations.rs
+++ b/crates/lash-remote-protocol/src/observations.rs
@@ -84,6 +84,9 @@ impl RemoteSessionObservation {
 pub struct RemoteSessionObservationEvent {
     pub protocol_version: u32,
     pub session_id: String,
+    pub replay_incarnation_id: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub turn_id: Option<String>,
     pub revision: u64,
     pub cursor: String,
     #[serde(flatten)]
@@ -98,6 +101,14 @@ impl RemoteSessionObservationEvent {
             "session_id",
             &self.session_id,
         )?;
+        require_non_empty(
+            "RemoteSessionObservationEvent",
+            "replay_incarnation_id",
+            &self.replay_incarnation_id,
+        )?;
+        if let Some(turn_id) = self.turn_id.as_deref() {
+            require_non_empty("RemoteSessionObservationEvent", "turn_id", turn_id)?;
+        }
         require_non_empty("RemoteSessionObservationEvent", "cursor", &self.cursor)?;
         if let RemoteSessionObservationEventPayload::TurnActivity { activity } = &self.event {
             activity.validate()?;
diff --git a/crates/lash-remote-protocol/src/tests.rs b/crates/lash-remote-protocol/src/tests.rs
index 94eceb8a0..809e89da9 100644
--- a/crates/lash-remote-protocol/src/tests.rs
+++ b/crates/lash-remote-protocol/src/tests.rs
@@ -474,6 +474,8 @@ fn remote_session_observation_dtos_json_round_trip_typed_kinds() {
     let event = RemoteSessionObservationEvent {
         protocol_version: REMOTE_PROTOCOL_VERSION,
         session_id: "session".to_string(),
+        replay_incarnation_id: "replay-incarnation".to_string(),
+        turn_id: None,
         revision: 3,
         cursor: "lashsc1:3:7:session".to_string(),
         event: RemoteSessionObservationEventPayload::QueueChanged {
@@ -843,8 +845,8 @@ fn wrong_protocol_versions_are_rejected() {
     assert!(matches!(
         request.validate(),
         Err(RemoteProtocolError::UnsupportedProtocolVersion {
-            actual: 15,
-            expected: 16,
+            actual: 16,
+            expected: 17,
         })
     ));
 
@@ -926,7 +928,7 @@ fn nested_protocol_versions_must_match_envelope() {
 
 #[test]
 fn remote_process_env_ref_is_validated_but_serializes_as_string() {
-    assert_eq!(REMOTE_PROTOCOL_VERSION, 16);
+    assert_eq!(REMOTE_PROTOCOL_VERSION, 17);
     let env_ref: RemoteProcessExecutionEnvRef =
         canonical_env_ref().parse().expect("canonical env ref");
     assert_eq!(env_ref.as_str(), canonical_env_ref());
diff --git a/crates/lash/src/admin.rs b/crates/lash/src/admin.rs
index a3a04a6fb..2ffb61a26 100644
--- a/crates/lash/src/admin.rs
+++ b/crates/lash/src/admin.rs
@@ -527,13 +527,13 @@ impl SessionAdmin {
         pending_turn_inputs: &[lash_core::PendingTurnInput],
     ) {
         for owned in events {
-            self.runtime
-                .record_turn_activity(lash_core::TurnActivity::independent(
-                    lash_core::TurnEvent::PluginRuntime {
-                        plugin_id: owned.plugin_id.clone(),
-                        event: owned.value.clone(),
-                    },
-                ));
+            self.runtime.record_turn_activity(
+                None,
+                lash_core::TurnActivity::independent(lash_core::TurnEvent::PluginRuntime {
+                    plugin_id: owned.plugin_id.clone(),
+                    event: owned.value.clone(),
+                }),
+            );
         }
         if !pending_turn_inputs.is_empty() {
             self.runtime.record_queue_changed(
diff --git a/crates/lash/src/recoverable_chat.rs b/crates/lash/src/recoverable_chat.rs
index b255a6a66..0be4e21a9 100644
--- a/crates/lash/src/recoverable_chat.rs
+++ b/crates/lash/src/recoverable_chat.rs
@@ -4,7 +4,7 @@
 //! terminal-replacement contract. Hosts own authorization, product events,
 //! transcript presentation, and cancellation controls.
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, VecDeque};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -17,18 +17,18 @@ use lash_core::{
 use crate::Result;
 use crate::session::{ObservableSession, SessionObservationStream, SessionObservationStreamItem};
 
-/// Subscription-scoped at-least-once delivery identity for one Lash
-/// observation event.
+/// At-least-once delivery identity for one Lash observation event.
 ///
-/// Hosts retain this identity only while the current live-replay incarnation is
-/// known to be continuous. Re-delivery of the same identity must not create
-/// another row. A [`RecoverableChatUpdate::ReplayGap`] ends that continuity:
-/// the replacement snapshot is authoritative and cursor identities retained
-/// from before the gap must be discarded because an in-memory replay store can
-/// reuse them after restart.
+/// The replay-store incarnation makes this identity safe to persist across
+/// process restarts: a newly constructed store may reuse a cursor, but it
+/// cannot reproduce the old identity. Re-delivery of the same identity must
+/// not create another row. A [`RecoverableChatUpdate::ReplayGap`] still makes
+/// the replacement snapshot authoritative and clears the subscription's
+/// bounded applied-identity window.
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct RecoverableChatEventId {
     pub session_id: String,
+    pub replay_incarnation_id: String,
     pub cursor: String,
 }
 
@@ -36,11 +36,46 @@ impl RecoverableChatEventId {
     fn from_event(event: &SessionObservationEvent) -> Self {
         Self {
             session_id: event.session_id.clone(),
+            replay_incarnation_id: event.replay_incarnation_id.clone(),
             cursor: event.cursor.to_string(),
         }
     }
 }
 
+const MAX_APPLIED_EVENT_IDS: usize = 4096;
+
+#[derive(Default)]
+struct AppliedEventIds {
+    ids: BTreeSet<RecoverableChatEventId>,
+    order: VecDeque<RecoverableChatEventId>,
+}
+
+impl AppliedEventIds {
+    fn insert(&mut self, id: RecoverableChatEventId) -> bool {
+        if !self.ids.insert(id.clone()) {
+            return false;
+        }
+        self.order.push_back(id);
+        while self.order.len() > MAX_APPLIED_EVENT_IDS {
+            if let Some(expired) = self.order.pop_front() {
+                self.ids.remove(&expired);
+            }
+        }
+        true
+    }
+
+    fn clear(&mut self) {
+        self.ids.clear();
+        self.order.clear();
+    }
+
+    fn extend(&mut self, ids: impl IntoIterator<Item = RecoverableChatEventId>) {
+        for id in ids {
+            self.insert(id);
+        }
+    }
+}
+
 /// One authoritative materialization paired with the cursor at which it was
 /// captured.
 #[derive(Clone, Debug)]
@@ -91,25 +126,26 @@ pub enum RecoverableChatUpdate {
 /// [`TurnWorkDriver`](crate::TurnWorkDriver).
 pub struct RecoverableChatSubscription {
     inner: SessionObservationStream,
-    applied: BTreeSet<RecoverableChatEventId>,
+    applied: AppliedEventIds,
 }
 
 impl RecoverableChatSubscription {
     pub(crate) fn new(inner: SessionObservationStream) -> Self {
         Self {
             inner,
-            applied: BTreeSet::new(),
+            applied: AppliedEventIds::default(),
         }
     }
 
-    /// Seed identities already applied by the host projection in this replay
-    /// store incarnation.
+    /// Seed identities already applied by the host projection.
     ///
-    /// This makes same-incarnation reconnect redelivery idempotent even when
-    /// the projection cursor intentionally trails individual applied events.
-    /// Do not persist these identities across a replay gap or process restart;
-    /// the subscription clears them when it emits
-    /// [`RecoverableChatUpdate::ReplayGap`].
+    /// This makes reconnect redelivery idempotent even when the projection
+    /// cursor intentionally trails individual applied events. Persisting the
+    /// identities across a process restart is safe because the replay-store
+    /// incarnation distinguishes newly emitted events from old ones. The
+    /// subscription retains only a bounded recent window and still clears it
+    /// when it emits [`RecoverableChatUpdate::ReplayGap`], whose replacement
+    /// snapshot is authoritative.
     pub fn with_applied_event_ids(
         mut self,
         ids: impl IntoIterator<Item = RecoverableChatEventId>,
diff --git a/crates/lash/src/tests/turn_streaming.rs b/crates/lash/src/tests/turn_streaming.rs
index febf36ab8..4ff4cea3c 100644
--- a/crates/lash/src/tests/turn_streaming.rs
+++ b/crates/lash/src/tests/turn_streaming.rs
@@ -1172,6 +1172,55 @@ fn model_attempt_resets(events: &[Arc<lash_core::SessionObservationEvent>]) -> u
         .count()
 }
 
+#[tokio::test]
+async fn session_observation_envelopes_scope_activity_and_commit_to_the_turn() -> Result<()> {
+    let core = standard_core();
+    let session = core
+        .session("session-observation-turn-identity")
+        .open()
+        .await?;
+    let cursor = session.observe().current_observation().cursor;
+
+    session
+        .turn(TurnInput::text("identify this turn"))
+        .turn_id("observation-turn")
+        .run()
+        .await?;
+
+    let lash_core::SessionResume::Replayed { events } =
+        session.observe().resume_from_cursor(&cursor)?
+    else {
+        panic!("fresh turn observation cursor should remain replayable");
+    };
+    let turn_activity = events
+        .iter()
+        .filter(|event| {
+            matches!(
+                event.payload,
+                lash_core::SessionObservationEventPayload::TurnActivity(_)
+            )
+        })
+        .collect::<Vec<_>>();
+    assert!(!turn_activity.is_empty(), "turn emitted no activity");
+    assert!(
+        turn_activity
+            .iter()
+            .all(|event| event.turn_id.as_deref() == Some("observation-turn")),
+        "every turn activity must carry its producing turn identity"
+    );
+    let committed = events
+        .iter()
+        .find(|event| {
+            matches!(
+                event.payload,
+                lash_core::SessionObservationEventPayload::Committed { .. }
+            )
+        })
+        .expect("turn commit observation");
+    assert_eq!(committed.turn_id.as_deref(), Some("observation-turn"));
+    Ok(())
+}
+
 #[tokio::test]
 async fn session_observation_retracts_two_retried_visible_attempts_live_and_on_replay() -> Result<()>
 {
@@ -1553,6 +1602,7 @@ impl lash_core::LiveReplayStore for PausedCommitReplayStore {
         &self,
         session_id: &str,
         revision: lash_core::SessionRevision,
+        turn_id: Option<&str>,
         payload: lash_core::SessionObservationEventPayload,
     ) -> std::result::Result<Arc<lash_core::SessionObservationEvent>, lash_core::LiveReplayStoreError>
     {
@@ -1560,7 +1610,7 @@ impl lash_core::LiveReplayStore for PausedCommitReplayStore {
             payload,
             lash_core::SessionObservationEventPayload::Committed { .. }
         );
-        let event = self.inner.append(session_id, revision, payload)?;
+        let event = self.inner.append(session_id, revision, turn_id, payload)?;
         if pause
             && !self
                 .commit_appended
@@ -1713,7 +1763,7 @@ async fn recoverable_chat_conformance_deduplicates_redelivery_identity() -> Resu
 }
 
 #[tokio::test]
-async fn recoverable_chat_gap_allows_new_event_that_reuses_pre_restart_cursor() -> Result<()> {
+async fn recoverable_chat_restart_identity_does_not_depend_on_gap_clearing() -> Result<()> {
     let session_id = "recoverable-chat-restart-cursor";
     let store_factory = Arc::new(lash_core::InMemorySessionStoreFactory::new());
     let bootstrap_core = explicit_ephemeral_facets(LashCore::standard_builder())
@@ -1737,12 +1787,12 @@ async fn recoverable_chat_gap_allows_new_event_that_reuses_pre_restart_cursor()
         .build()?;
     let first_session = first_core.session(session_id).open().await?;
     let initial_cursor = first_session.observe().recoverable_chat_snapshot().cursor;
-    first_session
-        .observe()
-        .runtime
-        .record_turn_activity(TurnActivity::independent(TurnEvent::AssistantProseDelta {
+    first_session.observe().runtime.record_turn_activity(
+        Some("before-restart-turn"),
+        TurnActivity::independent(TurnEvent::AssistantProseDelta {
             text: "before replay-store restart".into(),
-        }));
+        }),
+    );
     let mut first_stream = first_session
         .observe()
         .subscribe_recoverable_chat(initial_cursor);
@@ -1762,6 +1812,11 @@ async fn recoverable_chat_gap_allows_new_event_that_reuses_pre_restart_cursor()
         .live_replay_store(Arc::new(lash_core::InMemoryLiveReplayStore::default()))
         .build()?;
     let second_session = second_core.session(session_id).open().await?;
+    let restarted_at = second_session.observe().recoverable_chat_snapshot().cursor;
+    let mut retained_applied_ids = second_session
+        .observe()
+        .subscribe_recoverable_chat(restarted_at)
+        .with_applied_event_ids([old_id.clone()]);
     let mut recovered = second_session
         .observe()
         .subscribe_recoverable_chat(old_cursor)
@@ -1778,23 +1833,47 @@ async fn recoverable_chat_gap_allows_new_event_that_reuses_pre_restart_cursor()
         }
     ));
 
-    second_session
-        .observe()
-        .runtime
-        .record_turn_activity(TurnActivity::independent(TurnEvent::AssistantProseDelta {
+    second_session.observe().runtime.record_turn_activity(
+        Some("after-restart-turn"),
+        TurnActivity::independent(TurnEvent::AssistantProseDelta {
             text: "after replay-store restart".into(),
-        }));
-    let update = tokio::time::timeout(std::time::Duration::from_millis(500), recovered.next())
-        .await
-        .expect("new event at a reused cursor was incorrectly suppressed")
-        .expect("recovered stream remains open")?;
+        }),
+    );
+    let gap_continuation =
+        tokio::time::timeout(std::time::Duration::from_millis(500), recovered.next())
+            .await
+            .expect("gap stream did not continue with the new event")
+            .expect("recovered stream remains open")?;
+    let crate::recoverable_chat::RecoverableChatUpdate::Event {
+        id: gap_continuation_id,
+        event: gap_continuation_event,
+    } = gap_continuation
+    else {
+        panic!("expected post-gap provisional event");
+    };
+    let update = tokio::time::timeout(
+        std::time::Duration::from_millis(500),
+        retained_applied_ids.next(),
+    )
+    .await
+    .expect("retained pre-restart identity incorrectly suppressed the new event")
+    .expect("recovered stream remains open")?;
     let crate::recoverable_chat::RecoverableChatUpdate::Event { id, event } = update else {
         panic!("expected post-restart provisional event");
     };
     assert_eq!(
-        id, old_id,
+        id.cursor, old_id.cursor,
         "the in-memory replay store deliberately reuses the pre-restart cursor"
     );
+    assert_ne!(
+        id, old_id,
+        "a fresh replay-store incarnation must distinguish a reused cursor without relying on gap clearing"
+    );
+    assert_eq!(gap_continuation_id, id);
+    assert_eq!(
+        observation_assistant_delta(&gap_continuation_event).as_deref(),
+        Some("after replay-store restart")
+    );
     assert_eq!(
         observation_assistant_delta(&event).as_deref(),
         Some("after replay-store restart")
diff --git a/crates/lash/src/turn.rs b/crates/lash/src/turn.rs
index baec8337f..7d9a9b3ed 100644
--- a/crates/lash/src/turn.rs
+++ b/crates/lash/src/turn.rs
@@ -841,11 +841,19 @@ impl TurnActivitySink for SessionObservationTurnActivitySink<'_> {
     }
 
     async fn emit(&self, activity: TurnActivity) {
-        self.runtime.record_turn_activity(activity.clone());
+        self.runtime.record_turn_activity(None, activity.clone());
         if let Some(live) = self.live {
             live.emit(activity).await;
         }
     }
+
+    async fn emit_for_turn(&self, turn_id: &str, activity: TurnActivity) {
+        self.runtime
+            .record_turn_activity(Some(turn_id), activity.clone());
+        if let Some(live) = self.live {
+            live.emit_for_turn(turn_id, activity).await;
+        }
+    }
 }
 
 struct ChannelTurnActivitySink {
diff --git a/docs/example-agent-workbench.html b/docs/example-agent-workbench.html
index 4a1342218..8b9a16c72 100644
--- a/docs/example-agent-workbench.html
+++ b/docs/example-agent-workbench.html
@@ -83,7 +83,7 @@ <h2>Browser Stream And Reconnect</h2>
         </div>
         <div class="listing__row">
           <div class="listing__label">stream rows</div>
-          <div class="listing__body"><p>The product response carries identified <code>message</code>, <code>turn_input</code>, and <code>done</code> events or an authoritative <code>resync</code>. It has no raw-error variant. The Lash response carries <code>observation</code>, <code>replay_gap</code>, and <code>terminal_replacement</code>.</p></div>
+          <div class="listing__body"><p>The product response carries identified <code>message</code>, <code>turn_input</code>, and turn-scoped <code>done</code> events or an authoritative <code>resync</code>. A completed turn retracts only its own provisional rows, preserving output from newer turns. It has no raw-error variant. The Lash response carries <code>observation</code>, <code>replay_gap</code>, and <code>terminal_replacement</code>.</p></div>
         </div>
         <div class="listing__row">
           <div class="listing__label">gap recovery</div>
diff --git a/docs/remote-protocol.html b/docs/remote-protocol.html
index 4b6564e8f..a13ed5299 100644
--- a/docs/remote-protocol.html
+++ b/docs/remote-protocol.html
@@ -32,7 +32,7 @@ <h2>What This Is</h2>
     </div>
     <aside class="contract">
       <strong>Contract</strong>
-      <p>Every top-level remote envelope carries <code>protocol_version</code>. Current code uses <code>REMOTE_PROTOCOL_VERSION == 16</code>; call <code>validate()</code> or <code>ensure_protocol_version(...)</code> before executing a received envelope. Version 16 adds the typed <code>turn_input_applied</code> activity with stable admission, turn, committed-message, and optional checkpoint identity; hosts must not infer application from pending-input snapshots or parse a runtime diagnostic. Version 15 added trigger delivery provenance, and version 14 replaced image-only attachment DTOs and blob maps with the generic MIME-tagged <code>RemoteAttachmentSource</code> union. Remote consumers use exact-version negotiation, so each semantic wire expansion receives its own protocol version.</p>
+      <p>Every top-level remote envelope carries <code>protocol_version</code>. Current code uses <code>REMOTE_PROTOCOL_VERSION == 17</code>; call <code>validate()</code> or <code>ensure_protocol_version(...)</code> before executing a received envelope. Version 17 adds replay-store incarnation identity and optional turn identity to session observation event envelopes. Version 16 added the typed <code>turn_input_applied</code> activity with stable admission, turn, committed-message, and optional checkpoint identity. Remote consumers use exact-version negotiation, so a version 16 peer is rejected instead of silently decoding the expanded envelope.</p>
     </aside>
   </div>
 
@@ -156,7 +156,7 @@ <h2>Contract Map</h2>
       </div>
       <div class="listing__row">
         <div class="listing__label">session observation</div>
-        <div class="listing__body"><p><code>RemoteSessionCursor</code>, <code>RemoteSessionObservation</code>, <code>RemoteSessionObservationEvent</code>, and <code>RemoteLiveReplayGap</code> are the remote reconnect surface. They carry opaque cursor strings, session revisions, bounded replay events, explicit remote observation snapshots, and gap recovery without serializing a full <code>SessionReadView</code>. See <a href="./streaming.html#session-reconnect">Streaming and reconnect</a> for the host/frontend folding recipe.</p></div>
+        <div class="listing__body"><p><code>RemoteSessionCursor</code>, <code>RemoteSessionObservation</code>, <code>RemoteSessionObservationEvent</code>, and <code>RemoteLiveReplayGap</code> are the remote reconnect surface. They carry opaque cursor strings, session revisions, bounded replay events, explicit remote observation snapshots, and gap recovery without serializing a full <code>SessionReadView</code>. Every event carries the replay-store incarnation that makes redelivery identity restart-safe. Its optional <code>turn_id</code> is present for turn activity and commits, and absent for session-wide queue, process, and Agent Frame changes. See <a href="./streaming.html#session-reconnect">Streaming and reconnect</a> for the host/frontend folding recipe.</p></div>
       </div>
       <div class="listing__row">
         <div class="listing__label">prompt layer</div>
diff --git a/docs/streaming.html b/docs/streaming.html
index 751f5b4c1..33ce86f90 100644
--- a/docs/streaming.html
+++ b/docs/streaming.html
@@ -130,7 +130,7 @@ <h2>Session Reconnect</h2>
     </aside>
     <aside class="contract">
       <strong>Recoverable chat</strong>
-      <p><code>subscribe_recoverable_chat</code> assigns every event the subscription-scoped <code>RecoverableChatEventId { session_id, cursor }</code>, forwards a trimmed or unavailable window as <code>ReplayGap</code>, and upgrades a commit into <code>TerminalReplacement</code> carrying the authoritative read view. Seed already-applied identities only for a reconnect within the same replay-store incarnation to suppress at-least-once redelivery. A replay gap clears those identities because an in-memory replay store can reuse cursor values after restart; the gap snapshot becomes authoritative. Dropping this stream disconnects observation only; cancellation is a separate <code>TurnWorkDriver</code> request.</p>
+      <p><code>subscribe_recoverable_chat</code> assigns every event the restart-safe <code>RecoverableChatEventId { session_id, replay_incarnation_id, cursor }</code>, forwards a trimmed or unavailable window as <code>ReplayGap</code>, and upgrades a commit into <code>TerminalReplacement</code> carrying the authoritative read view. Hosts may persist already-applied identities to suppress at-least-once redelivery: every replay-store construction receives a fresh incarnation, so a cursor reused after restart cannot collide with an older identity. The subscription keeps its applied-identity window bounded. A replay gap still clears that window because the gap snapshot becomes authoritative. Dropping this stream disconnects observation only; cancellation is a separate <code>TurnWorkDriver</code> request.</p>
     </aside>
   </div>
 
diff --git a/examples/agent-workbench/README.md b/examples/agent-workbench/README.md
index 576b5884b..4601ac731 100644
--- a/examples/agent-workbench/README.md
+++ b/examples/agent-workbench/README.md
@@ -173,9 +173,10 @@ lanes after those cursors:
   `ObservableSession::subscribe_recoverable_chat` directly, then encodes its
   updates for HTTP. It forwards provisional turn activity,
   `RemoteLiveReplayGap`, and terminal replacement. Event identity is
-  `(session_id, cursor)` within one replay-store incarnation; a replay gap
-  clears pre-gap identities before the stream continues from its authoritative
-  snapshot cursor.
+  `(session_id, replay_incarnation_id, cursor)`, so a consumer may safely retain
+  its bounded identity cache when the server restarts and reuses cursor values.
+  A replay gap still clears pre-gap identities before the stream continues from
+  its authoritative snapshot cursor.
 - `/api/events` is the product lane. `SessionEventRegistry` first appends every
   event to `.agent-workbench/product-events.json` with a monotonic per-session
   sequence and stable event id, then broadcasts it as a freshness hint. A
@@ -188,16 +189,18 @@ serialization failures are traced server-side; no raw error string is a
 product-stream variant. The UI renders stable safe failure copy.
 
 Provisional prose and reasoning are keyed by Lash correlation id. A
-`model_attempt_reset` retracts only the superseded chunks. A replay gap,
-product lag, cancellation settlement, or terminal commit replaces all
-provisional state from `/api/state`. Recovery fetches are generation-fenced:
-an out-of-order response or a response overtaken by a newer product event is
-discarded. Authoritative replacement rebuilds both dedup sets and assigns,
-rather than monotonically preserves, the snapshot cursor. Cursor and dedup
-state are scoped to the returned session id, so reset cannot carry positions
-into the new session. The state endpoint merges product-only rows onto the
-complete canonical Lash transcript by stable id; it never replaces canonical
-history with a partial product log. Canonical assistant rows use
+`model_attempt_reset` retracts only the superseded chunks. Provisional rows
+also retain their producing turn id; a `done` event retracts only rows from its
+own turn, so late settlement cannot erase a newer turn's output. A replay gap,
+product lag, cancellation settlement, or terminal replacement rebuilds state
+from `/api/state`. Recovery fetches are generation-fenced: an out-of-order
+response or a response overtaken by a newer product event is discarded.
+Authoritative replacement rebuilds both dedup sets and assigns, rather than
+monotonically preserves, the snapshot cursor. Cursor and dedup state are scoped
+to the returned session id, so reset cannot carry positions into the new
+session. The state endpoint merges product-only rows onto the complete
+canonical Lash transcript by stable id; it never replaces canonical history
+with a partial product log. Canonical assistant rows use
 `workbench-assistant:<turn_id>` in both the live product event and durable
 session transcript, so a live/canonical pair is one row, never two.
 
@@ -511,10 +514,11 @@ The ownership split is intentional:
   the real `/api/events` resync response after forced broadcast lag,
   generation-fenced browser recovery, session-scoped reset cursors, canonical
   history merged with a partial product log, the real turn-output
-  live/canonical identity across reload, distinct cancel settlement events,
-  typed turn-input application, and fixed public copy from a real provider
-  failure. The browser cases execute the production JavaScript reducer under
-  Node; the workspace test shards install Node explicitly.
+  live/canonical identity across reload, interleaved turn settlement that keeps
+  newer provisional output, distinct cancel settlement events, typed turn-input
+  application, and fixed public copy from a real provider failure. The browser
+  cases execute the production JavaScript reducer under Node; the workspace
+  test shards install Node explicitly.
 
 Each gate asserts the violated invariant directly: duplicate identity changes a
 row count, a swallowed gap prevents recovery, a missing terminal replacement
diff --git a/examples/agent-workbench/assets/index.html b/examples/agent-workbench/assets/index.html
index 9563a8c0c..f5a21dd5c 100644
--- a/examples/agent-workbench/assets/index.html
+++ b/examples/agent-workbench/assets/index.html
@@ -2161,6 +2161,7 @@ <h2>what's what</h2>
     const renderedIngressInputs = new Set();
     const appliedTurnInputs = new Map();
     let assistantDraft = null;
+    let assistantDraftTurnId = null;
     let assistantDraftText = "";
     let assistantDraftChunks = [];
     let reasoning = null;
@@ -2351,6 +2352,7 @@ <h2>what's what</h2>
         const parent = assistantDraft.closest(".message");
         if (parent) parent.remove();
         assistantDraft = null;
+        assistantDraftTurnId = null;
         assistantDraftText = "";
         assistantDraftChunks = [];
       }
@@ -2444,8 +2446,8 @@ <h2>what's what</h2>
       }
     }
 
-    function ensureAssistantDraft() {
-      if (assistantDraft) return assistantDraft;
+    function ensureAssistantDraft(turnId) {
+      if (assistantDraft && assistantDraftTurnId === turnId) return assistantDraft;
       clearEmpty();
       const node = document.createElement("div");
       node.className = "message assistant";
@@ -2453,6 +2455,7 @@ <h2>what's what</h2>
       role.className = "msg-role";
       role.textContent = "agent";
       assistantDraft = document.createElement("div");
+      assistantDraftTurnId = turnId || null;
       assistantDraft.className = "msg-body";
       assistantDraftText = "";
       assistantDraftChunks = [];
@@ -2540,10 +2543,10 @@ <h2>what's what</h2>
       return "```json\n" + JSON.stringify(value, null, 2) + "\n```";
     }
 
-    function appendAssistantText(delta, correlationId = null) {
+    function appendAssistantText(delta, correlationId = null, turnId = null) {
       if (!delta) return;
-      const draft = ensureAssistantDraft();
-      assistantDraftChunks.push({ correlationId, text: delta });
+      const draft = ensureAssistantDraft(turnId);
+      assistantDraftChunks.push({ turnId, correlationId, text: delta });
       assistantDraftText = assistantDraftChunks.map(chunk => chunk.text).join("");
       draft.innerHTML = renderMarkdownBlocks(assistantDraftText);
       scrollToEnd();
@@ -2559,6 +2562,7 @@ <h2>what's what</h2>
       if (!assistantDraftText && assistantDraft) {
         assistantDraft.closest(".message")?.remove();
         assistantDraft = null;
+        assistantDraftTurnId = null;
       } else if (assistantDraft) {
         assistantDraft.innerHTML = renderMarkdownBlocks(assistantDraftText);
       }
@@ -2681,11 +2685,11 @@ <h2>what's what</h2>
       return el;
     }
 
-    function startCodeBlock(event) {
+    function startCodeBlock(event, turnId) {
       clearEmpty();
       const runningEvent = { ...event, phase: "running" };
       const el = appendCodeBlock(runningEvent);
-      pendingCodeBlock = { event: runningEvent, el };
+      pendingCodeBlock = { event: runningEvent, el, turnId };
       if (event.graph_key) refreshWork();
     }
 
@@ -2697,27 +2701,29 @@ <h2>what's what</h2>
       for (const tool of linkedTools) appendTool(tool, el);
     }
 
-    function appendCompletedTool(event) {
+    function appendCompletedTool(event, turnId) {
       if (pendingCodeBlock) {
-        pendingTools.push(event);
+        pendingTools.push({ ...event, observationTurnId: turnId });
       } else {
         appendTool(event);
       }
     }
 
-    function completeCodeBlock(event) {
+    function completeCodeBlock(event, turnId) {
       const linkedIds = new Set(event.tool_call_ids || []);
-      const linkedTools = pendingTools.filter(tool => tool.call_id && linkedIds.has(tool.call_id));
-      const unlinkedTools = pendingTools.filter(tool => !tool.call_id || !linkedIds.has(tool.call_id));
-      const completedEvent = { ...event, code: event.code || pendingCodeBlock?.event?.code || "" };
-      if (pendingCodeBlock?.el) {
-        updateCodeBlock(pendingCodeBlock.el, completedEvent, linkedTools);
+      const sameTurnTools = pendingTools.filter(tool => tool.observationTurnId === turnId);
+      const linkedTools = sameTurnTools.filter(tool => tool.call_id && linkedIds.has(tool.call_id));
+      const unlinkedTools = sameTurnTools.filter(tool => !tool.call_id || !linkedIds.has(tool.call_id));
+      const sameTurnBlock = pendingCodeBlock?.turnId === turnId ? pendingCodeBlock : null;
+      const completedEvent = { ...event, code: event.code || sameTurnBlock?.event?.code || "" };
+      if (sameTurnBlock?.el) {
+        updateCodeBlock(sameTurnBlock.el, completedEvent, linkedTools);
       } else {
         appendCodeBlock(completedEvent, linkedTools);
       }
-      pendingCodeBlock = null;
+      if (sameTurnBlock) pendingCodeBlock = null;
       for (const tool of unlinkedTools) appendTool(tool);
-      pendingTools = [];
+      pendingTools = pendingTools.filter(tool => tool.observationTurnId !== turnId);
     }
 
     function thinkingPanel(label) {
@@ -2733,13 +2739,13 @@ <h2>what's what</h2>
       return node && node.parentNode === timeline && timeline.lastElementChild === node;
     }
 
-    function appendReasoning(delta, correlationId = null) {
+    function appendReasoning(delta, correlationId = null, turnId = null) {
       if (!isCurrentTimelineHead(reasoning)) {
         clearEmpty();
         reasoning = thinkingPanel("thinking");
         timeline.appendChild(reasoning);
       }
-      reasoningChunks.push({ correlationId, text: delta, node: reasoning });
+      reasoningChunks.push({ turnId, correlationId, text: delta, node: reasoning });
       reasoning.querySelector("pre").textContent += delta;
       scrollToEnd();
     }
@@ -2798,31 +2804,48 @@ <h2>what's what</h2>
     }
 
     // BEGIN WORKBENCH_TRANSIENT_SETTLEMENT
-    function finishTransientRows() {
-      for (const tool of pendingTools) appendTool(tool);
-      assistantDraft?.closest(".message")?.remove();
-      assistantDraft = null;
-      assistantDraftText = "";
-      assistantDraftChunks = [];
-      for (const chunk of reasoningChunks) chunk.node?.remove();
-      pendingCodeBlock = null;
-      pendingTools = [];
-      reasoning = null;
-      reasoningChunks = [];
+    function finishTransientRows(turnId) {
+      if (!turnId) return;
+      const finishedTools = pendingTools.filter(tool => tool.observationTurnId === turnId);
+      for (const tool of finishedTools) appendTool(tool);
+      pendingTools = pendingTools.filter(tool => tool.observationTurnId !== turnId);
+      if (assistantDraftTurnId === turnId) {
+        assistantDraft?.closest(".message")?.remove();
+        assistantDraft = null;
+        assistantDraftTurnId = null;
+        assistantDraftText = "";
+        assistantDraftChunks = [];
+      }
+      const affectedReasoning = new Set(
+        reasoningChunks
+          .filter(chunk => chunk.turnId === turnId)
+          .map(chunk => chunk.node)
+      );
+      reasoningChunks = reasoningChunks.filter(chunk => chunk.turnId !== turnId);
+      for (const node of affectedReasoning) {
+        const text = reasoningChunks
+          .filter(chunk => chunk.node === node)
+          .map(chunk => chunk.text)
+          .join("");
+        if (text) node.querySelector("pre").textContent = text;
+        else node.remove();
+      }
+      if (pendingCodeBlock?.turnId === turnId) pendingCodeBlock = null;
+      if (!reasoning?.isConnected) reasoning = null;
     }
     // END WORKBENCH_TRANSIENT_SETTLEMENT
 
-    function handleTurnEvent(event) {
+    function handleTurnEvent(event, turnId) {
       if (event.type === "queued_work_started") renderQueuedWorkStarted(event);
       if (event.type === "turn_input_applied") recordTurnInputApplications(event.applications);
-      if (event.type === "assistant_prose_delta") appendAssistantText(event.text, event.correlation_id);
-      if (event.type === "reasoning_delta") appendReasoning(event.text, event.correlation_id);
+      if (event.type === "assistant_prose_delta") appendAssistantText(event.text, event.correlation_id, turnId);
+      if (event.type === "reasoning_delta") appendReasoning(event.text, event.correlation_id, turnId);
       if (event.type === "model_attempt_reset") resetModelAttempt(event);
-      if (event.type === "code_block_started") startCodeBlock(event);
-      if (event.type === "code_block_completed") completeCodeBlock(event);
-      if (event.type === "tool_call_completed") appendCompletedTool(event);
-      if (event.type === "final_value") appendAssistantText(renderTerminalValue(event.value));
-      if (event.type === "tool_value") appendAssistantText(renderTerminalValue(event.value));
+      if (event.type === "code_block_started") startCodeBlock(event, turnId);
+      if (event.type === "code_block_completed") completeCodeBlock(event, turnId);
+      if (event.type === "tool_call_completed") appendCompletedTool(event, turnId);
+      if (event.type === "final_value") appendAssistantText(renderTerminalValue(event.value), null, turnId);
+      if (event.type === "tool_value") appendAssistantText(renderTerminalValue(event.value), null, turnId);
       if (event.type === "error") renderError("turn could not be completed", { retry: true });
     }
 
@@ -2835,6 +2858,7 @@ <h2>what's what</h2>
       const commandController = new AbortController();
       controller = commandController;
       assistantDraft = null;
+      assistantDraftTurnId = null;
       assistantDraftText = "";
       assistantDraftChunks = [];
       reasoning = null;
@@ -2972,7 +2996,7 @@ <h2>what's what</h2>
       if (event.type === "message") renderMessage(event.message);
       if (event.type === "turn_input") renderIngressReceipt(event.receipt);
       if (event.type === "done") {
-        finishTransientRows();
+        finishTransientRows(event.turn_id);
         setBusy(false, "ready");
         refreshUsage();
       }
@@ -2988,17 +3012,18 @@ <h2>what's what</h2>
       if (item.type === "terminal_replacement") {
         projectionState.observationCursor =
           item.cursor || item.event?.cursor || projectionState.observationCursor;
+        finishTransientRows(item.event?.turn_id);
         recoverTerminalReplacement();
       }
     }
 
     function handleObservation(event) {
-      const identity = `${event.session_id || ""}:${event.cursor || ""}`;
+      const identity = `${event.session_id || ""}:${event.replay_incarnation_id || ""}:${event.cursor || ""}`;
       if (appliedObservationEvents.has(identity)) return;
       appliedObservationEvents.add(identity);
       projectionState.observationCursor =
         event.cursor || projectionState.observationCursor;
-      if (event.type === "turn_activity") handleTurnEvent(event.activity);
+      if (event.type === "turn_activity") handleTurnEvent(event.activity, event.turn_id);
     }
 
     async function recoverFromState(failureMessage) {
@@ -3069,6 +3094,7 @@ <h2>what's what</h2>
       timeline.innerHTML = "";
       renderedMessages.clear();
       assistantDraft = null;
+      assistantDraftTurnId = null;
       assistantDraftText = "";
       assistantDraftChunks = [];
       reasoning = null;
diff --git a/examples/agent-workbench/src/main_sections/app_state.rs b/examples/agent-workbench/src/main_sections/app_state.rs
index dd2396837..c43d7f1dc 100644
--- a/examples/agent-workbench/src/main_sections/app_state.rs
+++ b/examples/agent-workbench/src/main_sections/app_state.rs
@@ -43,16 +43,6 @@ impl AppState {
         );
     }
 
-    #[cfg(test)]
-    fn publish(&self, item: StreamItem) {
-        self.publish_for_session(&self.current_session_id(), item);
-    }
-
-    #[cfg(test)]
-    fn publish_for_session(&self, session_id: &str, item: StreamItem) {
-        self.event_tx.publish(session_id, item);
-    }
-
     fn publish_for_session_identified(
         &self,
         session_id: &str,
@@ -68,7 +58,9 @@ impl AppState {
         self.publish_for_session_identified(
             session_id,
             format!("turn:{turn_id}:done"),
-            StreamItem::Done,
+            StreamItem::Done {
+                turn_id: Some(turn_id.to_string()),
+            },
         );
     }
 
@@ -77,7 +69,7 @@ impl AppState {
             self.publish_for_session_identified(
                 session_id,
                 format!("operation:{operation_id}:done"),
-                StreamItem::Done,
+                StreamItem::Done { turn_id: None },
             );
         }
     }
@@ -187,7 +179,7 @@ impl AppState {
             self.publish_for_session_identified(
                 session_id,
                 format!("turn-cancel:{}:done", operation_ids.join(",")),
-                StreamItem::Done,
+                StreamItem::Done { turn_id: None },
             );
         }
         Ok(receipts)
diff --git a/examples/agent-workbench/src/main_sections/routes.rs b/examples/agent-workbench/src/main_sections/routes.rs
index 3e145adf2..be88f5394 100644
--- a/examples/agent-workbench/src/main_sections/routes.rs
+++ b/examples/agent-workbench/src/main_sections/routes.rs
@@ -35,7 +35,7 @@ async fn app_state(
         .iter()
         .filter_map(|event| match &event.item {
             StreamItem::Message { message } => Some(message.clone()),
-            StreamItem::TurnInput { .. } | StreamItem::Done => None,
+            StreamItem::TurnInput { .. } | StreamItem::Done { .. } => None,
         })
         .collect::<Vec<_>>();
     let mut message_ids = messages
diff --git a/examples/agent-workbench/src/main_sections/state.rs b/examples/agent-workbench/src/main_sections/state.rs
index faa491556..cbc71617e 100644
--- a/examples/agent-workbench/src/main_sections/state.rs
+++ b/examples/agent-workbench/src/main_sections/state.rs
@@ -254,7 +254,10 @@ enum StreamItem {
     TurnInput {
         receipt: TurnInputReceipt,
     },
-    Done,
+    Done {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        turn_id: Option<String>,
+    },
 }
 
 const PUBLIC_TURN_FAILURE_MESSAGE: &str = "turn could not be completed";
diff --git a/examples/agent-workbench/src/main_sections/tests.rs b/examples/agent-workbench/src/main_sections/tests.rs
index 446009f6a..3fee50bd9 100644
--- a/examples/agent-workbench/src/main_sections/tests.rs
+++ b/examples/agent-workbench/src/main_sections/tests.rs
@@ -453,16 +453,19 @@ mod tests {
             active_turns: ActiveTurns::default(),
             authorization: WorkbenchAuthorization::allow_all(),
         };
-        let mut events = state.event_tx.subscribe(&state.current_session_id());
+        let session_id = state.current_session_id();
+        let mut events = state.event_tx.subscribe(&session_id);
 
-        state.publish(StreamItem::Done);
+        state.publish_turn_done(&session_id, "transient-turn");
 
         assert!(matches!(
             events.try_recv(),
             Ok(ProductEvent {
-                item: StreamItem::Done,
+                item: StreamItem::Done {
+                    turn_id: Some(turn_id),
+                },
                 ..
-            })
+            }) if turn_id == "transient-turn"
         ));
         let _ = std::fs::remove_dir_all(data_dir);
     }
@@ -545,7 +548,7 @@ mod tests {
         assert!(matches!(
             events.try_recv(),
             Ok(ProductEvent {
-                item: StreamItem::Done,
+                item: StreamItem::Done { turn_id: None },
                 ..
             })
         ));
@@ -847,7 +850,7 @@ finish "gap source"
         assert!(matches!(
             events.try_recv(),
             Ok(ProductEvent {
-                item: StreamItem::Done,
+                item: StreamItem::Done { .. },
                 ..
             })
         ));
diff --git a/examples/agent-workbench/src/main_sections/tests/recoverable_chat.rs b/examples/agent-workbench/src/main_sections/tests/recoverable_chat.rs
index 68ed0ee64..33f72b216 100644
--- a/examples/agent-workbench/src/main_sections/tests/recoverable_chat.rs
+++ b/examples/agent-workbench/src/main_sections/tests/recoverable_chat.rs
@@ -95,11 +95,11 @@ fn session_event_registry_isolates_channels_and_recreates_after_removal() {
     let mut session_a = registry.subscribe("session-a");
     let mut session_b = registry.subscribe("session-b");
 
-    registry.publish("session-a", StreamItem::Done);
+    registry.publish("session-a", StreamItem::Done { turn_id: None });
     assert!(matches!(
         session_a.try_recv(),
         Ok(ProductEvent {
-            item: StreamItem::Done,
+            item: StreamItem::Done { .. },
             ..
         })
     ));
@@ -111,11 +111,11 @@ fn session_event_registry_isolates_channels_and_recreates_after_removal() {
     registry.remove("session-a");
     assert!(!registry.contains("session-a"));
     let mut replacement_a = registry.subscribe("session-a");
-    registry.publish("session-a", StreamItem::Done);
+    registry.publish("session-a", StreamItem::Done { turn_id: None });
     assert!(matches!(
         replacement_a.try_recv(),
         Ok(ProductEvent {
-            item: StreamItem::Done,
+            item: StreamItem::Done { .. },
             ..
         })
     ));
@@ -287,7 +287,9 @@ async fn workbench_sequential_settled_turn_cancels_each_emit_done() {
         .snapshot(&session_id)
         .events
         .into_iter()
-        .filter_map(|event| matches!(event.item, StreamItem::Done).then_some(event.event_id))
+        .filter_map(|event| {
+            matches!(event.item, StreamItem::Done { .. }).then_some(event.event_id)
+        })
         .collect::<Vec<_>>();
     assert_eq!(
         done_ids.len(),
diff --git a/examples/agent-workbench/src/main_sections/tests/restate_recovery.rs b/examples/agent-workbench/src/main_sections/tests/restate_recovery.rs
index 9e4728b54..5c5106a90 100644
--- a/examples/agent-workbench/src/main_sections/tests/restate_recovery.rs
+++ b/examples/agent-workbench/src/main_sections/tests/restate_recovery.rs
@@ -329,7 +329,7 @@ async fn live_restate_provider_auth_failure_terminalizes_and_session_recovers_in
             StreamItem::Message { message } if message.role == "event" => {
                 rendered_failure = Some(message.text)
             }
-            StreamItem::Done => saw_done = true,
+            StreamItem::Done { .. } => saw_done = true,
             _ => {}
         }
     }
@@ -1317,7 +1317,7 @@ async fn live_restate_ingress_owner_restart_for_store(backend: &'static str) {
         let done_count = product_events
             .events
             .iter()
-            .filter(|event| matches!(&event.item, StreamItem::Done))
+            .filter(|event| matches!(&event.item, StreamItem::Done { .. }))
             .count();
         if done_count > 0 {
             break;
@@ -1334,7 +1334,7 @@ async fn live_restate_ingress_owner_restart_for_store(backend: &'static str) {
         .snapshot(&session_id)
         .events
         .iter()
-        .filter(|event| matches!(&event.item, StreamItem::Done))
+        .filter(|event| matches!(&event.item, StreamItem::Done { .. }))
         .count();
     assert_eq!(
         done_count, 1,
diff --git a/examples/agent-workbench/tests/browser_projection.mjs b/examples/agent-workbench/tests/browser_projection.mjs
index 5a2e1a946..72f4282fe 100644
--- a/examples/agent-workbench/tests/browser_projection.mjs
+++ b/examples/agent-workbench/tests/browser_projection.mjs
@@ -136,8 +136,11 @@ test("a Done product event behaviorally retracts the provisional draft", () => {
         };
       },
     },
+    assistantDraftTurnId: "cancel-turn",
     assistantDraftText: "provisional text",
-    assistantDraftChunks: [{ text: "provisional text" }],
+    assistantDraftChunks: [
+      { turnId: "cancel-turn", text: "provisional text" },
+    ],
     pendingTools: [],
     appendTool() {},
     reasoningChunks: [],
@@ -153,14 +156,89 @@ test("a Done product event behaviorally retracts the provisional draft", () => {
   vm.runInNewContext(
     `${markedSource("WORKBENCH_TRANSIENT_SETTLEMENT", "WORKBENCH_TRANSIENT_SETTLEMENT")}
      ${markedSource("WORKBENCH_PRODUCT_EVENT_REDUCER", "WORKBENCH_PRODUCT_EVENT_REDUCER")}
-     applyProductEvent({ event_id: "cancel-done", sequence: 1, type: "done" });
-     this.result = { assistantDraft, assistantDraftText, assistantDraftChunks };`,
+     applyProductEvent({
+       event_id: "cancel-done",
+       sequence: 1,
+       type: "done",
+       turn_id: "cancel-turn"
+     });
+     this.result = {
+       assistantDraft,
+       assistantDraftTurnId,
+       assistantDraftText,
+       assistantDraftChunks
+     };`,
     reducerContext,
   );
 
   assert.equal(draftRemoved, true);
   assert.equal(busy, false);
   assert.equal(reducerContext.result.assistantDraft, null);
+  assert.equal(reducerContext.result.assistantDraftTurnId, null);
   assert.equal(reducerContext.result.assistantDraftText, "");
   assert.deepEqual([...reducerContext.result.assistantDraftChunks], []);
 });
+
+test("turn A Done does not retract turn B provisional prose", () => {
+  let draftRemoved = false;
+  const reducerContext = {
+    Set,
+    projectionState: createWorkbenchProjectionState(),
+    renderedProductEvents: new Set(),
+    assistantDraft: {
+      closest() {
+        return {
+          remove() {
+            draftRemoved = true;
+          },
+        };
+      },
+    },
+    assistantDraftTurnId: "turn-b",
+    assistantDraftText: "turn B provisional text",
+    assistantDraftChunks: [
+      {
+        turnId: "turn-b",
+        correlationId: "turn-b-prose",
+        text: "turn B provisional text",
+      },
+    ],
+    pendingTools: [],
+    appendTool() {},
+    reasoningChunks: [],
+    pendingCodeBlock: null,
+    reasoning: null,
+    renderMessage() {},
+    renderIngressReceipt() {},
+    setBusy() {},
+    refreshUsage() {},
+  };
+  vm.runInNewContext(
+    `${markedSource("WORKBENCH_TRANSIENT_SETTLEMENT", "WORKBENCH_TRANSIENT_SETTLEMENT")}
+     ${markedSource("WORKBENCH_PRODUCT_EVENT_REDUCER", "WORKBENCH_PRODUCT_EVENT_REDUCER")}
+     applyProductEvent({
+       event_id: "turn-a-done",
+       sequence: 1,
+       type: "done",
+       turn_id: "turn-a"
+     });
+     this.result = {
+       assistantDraft,
+       assistantDraftTurnId,
+       assistantDraftText,
+       assistantDraftChunks
+     };`,
+    reducerContext,
+  );
+
+  assert.equal(draftRemoved, false);
+  assert.equal(reducerContext.result.assistantDraftTurnId, "turn-b");
+  assert.equal(reducerContext.result.assistantDraftText, "turn B provisional text");
+  assert.deepEqual(
+    [...reducerContext.result.assistantDraftChunks].map((chunk) => ({
+      turnId: chunk.turnId,
+      text: chunk.text,
+    })),
+    [{ turnId: "turn-b", text: "turn B provisional text" }],
+  );
+});