fix(sdk): serialize session lifecycle recovery

rust/crates/sdk/src/session/mod.rs

@@ -13,7 +13,7 @@ pub use terminal_scope::{TerminalScopeRecord, TerminalScopeStore};
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::{Debug, Formatter};
 use std::panic::{catch_unwind, AssertUnwindSafe};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, MutexGuard};
 use std::thread;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -209,6 +209,7 @@ struct SessionEventPublisher {
     session_id: String,
     invocation_id: String,
     sequence: std::sync::atomic::AtomicU64,
+    publication_lock: Mutex<()>,
     sender: mpsc::UnboundedSender<SessionEvent>,
 }
 
@@ -222,11 +223,23 @@ impl SessionEventPublisher {
             session_id,
             invocation_id,
             sequence: std::sync::atomic::AtomicU64::new(0),
+            publication_lock: Mutex::new(()),
             sender,
         }
     }
 
-    fn build(&self, event_type: SessionEventType, payload: Value, durable: bool) -> SessionEvent {
+    fn lock(&self) -> MutexGuard<'_, ()> {
+        self.publication_lock
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    fn build_locked(
+        &self,
+        event_type: SessionEventType,
+        payload: Value,
+        durable: bool,
+    ) -> SessionEvent {
         let sequence = self
             .sequence
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
@@ -244,11 +257,34 @@ impl SessionEventPublisher {
     }
 
     fn publish(&self, event_type: SessionEventType, payload: Value, durable: bool) -> SessionEvent {
-        let event = self.build(event_type, payload, durable);
+        let _guard = self.lock();
+        self.publish_locked(event_type, payload, durable)
+    }
+
+    fn publish_locked(
+        &self,
+        event_type: SessionEventType,
+        payload: Value,
+        durable: bool,
+    ) -> SessionEvent {
+        let event = self.build_locked(event_type, payload, durable);
         let _ = self.sender.send(event.clone());
         event
     }
 
+    fn publish_steer(&self, control: &TurnControlSignal, prompt: &str) -> bool {
+        let _guard = self.lock();
+        if !control.steer(prompt.to_string()) {
+            return false;
+        }
+        self.publish_locked(
+            SessionEventType::TurnSteered,
+            json!({ "prompt": prompt }),
+            false,
+        );
+        true
+    }
+
     fn publish_runtime_event(&self, event: ConversationEvent) {
         let (event_type, payload) = match event {
             ConversationEvent::AgentBegin { session_id: _ } => {
@@ -361,11 +397,38 @@ impl SessionRouter {
     /// Resolve the most recently recorded session for a correlation key.
     pub fn session_for_correlation_key(&self, correlation_key: &str) -> SdkResult<Option<String>> {
         validate_correlation_key(correlation_key)?;
-        Ok(self
+        {
+            let active = self
+                .inner
+                .active
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            let mut matches = active.iter().filter(|(_, invocation)| {
+                invocation.correlation_key.as_deref() == Some(correlation_key)
+            });
+            if let Some((session_id, _)) = matches.next() {
+                if matches.next().is_some() {
+                    return Err(SdkError::conflict(format!(
+                        "correlation key '{correlation_key}' has multiple active sessions"
+                    )));
+                }
+                return Ok(Some(session_id.clone()));
+            }
+        }
+
+        let has_unfinished_invocation = self
             .inner
-            .terminal_scopes
-            .get(correlation_key)?
-            .map(|scope| scope.session_id))
+            .reconciliation
+            .list_by_correlation_key(correlation_key)?
+            .iter()
+            .any(|record| record.status != InvocationStatus::Terminal);
+        if has_unfinished_invocation {
+            return Err(SdkError::unavailable(format!(
+                "correlation key '{correlation_key}' has an invocation awaiting reconciliation"
+            )));
+        }
+        let scope = self.inner.terminal_scopes.get(correlation_key)?;
+        Ok(scope.map(|scope| scope.session_id))
     }
 
     /// Start a prompt and return immediately with an ordered event receiver.
@@ -404,6 +467,7 @@ impl SessionRouter {
             outcome: None,
             uncertainty_reason: None,
             terminal_event: None,
+            terminal_scope_revision: None,
         };
         self.inner.reconciliation.insert(record)?;
 
@@ -447,8 +511,9 @@ impl SessionRouter {
                 if let Err(error) = inner.reconciliation.mark_running(&worker_invocation_id) {
                     finish_uncertain(
                         &inner,
-                        &worker_session_id,
                         &worker_invocation_id,
+                        &worker_control,
+                        &worker_publisher,
                         &error.to_string(),
                     );
                     remove_active(&inner, &worker_session_id, &worker_invocation_id);
@@ -469,6 +534,7 @@ impl SessionRouter {
                         &worker_session_id,
                         &worker_invocation_id,
                         worker_correlation_key.as_deref(),
+                        &worker_control,
                         &worker_publisher,
                         run_result,
                     ),
@@ -476,8 +542,9 @@ impl SessionRouter {
                         let reason = panic_message(panic);
                         finish_uncertain(
                             &inner,
-                            &worker_session_id,
                             &worker_invocation_id,
+                            &worker_control,
+                            &worker_publisher,
                             &reason,
                         );
                     }
@@ -493,6 +560,7 @@ impl SessionRouter {
                 session_id,
                 &invocation_id,
                 correlation_key,
+                &control,
                 &publisher,
                 SessionRunResult::Failed { session, message },
             );
@@ -518,27 +586,28 @@ impl SessionRouter {
         validate_invocation_id(invocation_id)?;
         let prompt = override_prompt.into();
         validate_prompt(&prompt)?;
-        let active = self
-            .inner
-            .active
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let invocation = active
-            .get(session_id)
-            .ok_or_else(|| SdkError::not_found("session has no active invocation"))?;
-        if invocation.invocation_id != invocation_id {
-            return Err(SdkError::not_found(format!(
-                "invocation `{invocation_id}` is not active for session `{session_id}`"
-            )));
-        }
-        if !invocation.control.steer(prompt.clone()) {
+        let (control, publisher) = {
+            let active = self
+                .inner
+                .active
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            let invocation = active
+                .get(session_id)
+                .ok_or_else(|| SdkError::not_found("session has no active invocation"))?;
+            if invocation.invocation_id != invocation_id {
+                return Err(SdkError::not_found(format!(
+                    "invocation `{invocation_id}` is not active for session `{session_id}`"
+                )));
+            }
+            (
+                invocation.control.clone(),
+                Arc::clone(&invocation.event_publisher),
+            )
+        };
+        if !publisher.publish_steer(&control, &prompt) {
             return Err(SdkError::conflict("invocation no longer accepts steering"));
         }
-        invocation.event_publisher.publish(
-            SessionEventType::TurnSteered,
-            json!({ "prompt": prompt }),
-            false,
-        );
         Ok(())
     }
 
@@ -554,7 +623,7 @@ impl SessionRouter {
         validate_prompt(&prompt)?;
         validate_optional_correlation_key(correlation_key)?;
 
-        let (mut done, inherited_correlation) = {
+        let (mut done, inherited_correlation, previous_invocation_id) = {
             let active = self
                 .inner
                 .active
@@ -566,9 +635,10 @@ impl SessionRouter {
                     (
                         Some(invocation.done.clone()),
                         invocation.correlation_key.clone(),
+                        Some(invocation.invocation_id.clone()),
                     )
                 }
-                None => (None, None),
+                None => (None, None, None),
             }
         };
         if let Some(receiver) = done.as_mut() {
@@ -579,6 +649,19 @@ impl SessionRouter {
             }
         }
 
+        if let Some(invocation_id) = previous_invocation_id {
+            let record = self
+                .inner
+                .reconciliation
+                .get(&invocation_id)?
+                .ok_or_else(|| SdkError::not_found("aborted invocation record was not found"))?;
+            if record.status != InvocationStatus::Terminal {
+                return Err(SdkError::unavailable(format!(
+                    "invocation `{invocation_id}` has no durable terminal outcome"
+                )));
+            }
+        }
+
         let effective_correlation = correlation_key
             .map(str::to_string)
             .or(inherited_correlation);
@@ -795,9 +878,13 @@ fn finish_invocation(
     session_id: &str,
     invocation_id: &str,
     correlation_key: Option<&str>,
+    control: &TurnControlSignal,
     publisher: &SessionEventPublisher,
     result: SessionRunResult,
 ) {
+    let _publication_guard = publisher.lock();
+    control.close();
+
     let (outcome, message_count, status, error_message) = match result {
         SessionRunResult::Completed(session) => (
             InvocationOutcome::Completed {
@@ -829,7 +916,19 @@ fn finish_invocation(
     if let Some(error) = error_message {
         payload["error"] = Value::String(error);
     }
-    let terminal_event = publisher.build(SessionEventType::TurnEnded, payload, true);
+    let terminal_scope_revision = match correlation_key {
+        Some(key) => match inner.terminal_scopes.set(key, session_id) {
+            Ok(scope) => Some(scope.revision),
+            Err(error) => {
+                let _ = inner
+                    .reconciliation
+                    .mark_uncertain(invocation_id, error.to_string());
+                return;
+            }
+        },
+        None => None,
+    };
+    let terminal_event = publisher.build_locked(SessionEventType::TurnEnded, payload, true);
     let stored_event = match serde_json::to_value(&terminal_event) {
         Ok(event) => event,
         Err(error) => {
@@ -839,14 +938,13 @@ fn finish_invocation(
             return;
         }
     };
-    match inner
-        .reconciliation
-        .record_terminal(invocation_id, outcome, Some(stored_event))
-    {
+    match inner.reconciliation.record_terminal_with_scope_revision(
+        invocation_id,
+        outcome,
+        Some(stored_event),
+        terminal_scope_revision,
+    ) {
         Ok(()) => {
-            if let Some(key) = correlation_key {
-                let _ = inner.terminal_scopes.set(key, session_id);
-            }
             let _ = publisher.sender.send(terminal_event);
         }
         Err(error) => {
@@ -861,7 +959,7 @@ fn recover_terminal_scopes(
     reconciliation: &InvocationReconciliationStore,
     terminal_scopes: &TerminalScopeStore,
 ) -> SdkResult<()> {
-    let mut latest_by_key = BTreeMap::<String, InvocationRecord>::new();
+    let mut latest_by_key = BTreeMap::<String, (u64, String)>::new();
     for record in reconciliation.list()? {
         if record.status != InvocationStatus::Terminal {
             continue;
@@ -869,29 +967,39 @@ fn recover_terminal_scopes(
         let Some(key) = record.correlation_key.as_ref() else {
             continue;
         };
-        if !latest_by_key
-            .get(key)
-            .is_some_and(|current| current.updated_at_ms >= record.updated_at_ms)
-        {
-            latest_by_key.insert(key.clone(), record);
+        let Some(revision) = record.terminal_scope_revision else {
+            continue;
+        };
+        match latest_by_key.get(key) {
+            Some((current_revision, current_session)) if *current_revision == revision => {
+                if current_session != &record.session_id {
+                    return Err(SdkError::internal(format!(
+                        "terminal scope revision {revision} conflicts for key '{key}'"
+                    )));
+                }
+            }
+            Some((current_revision, _)) if *current_revision > revision => {}
+            _ => {
+                latest_by_key.insert(key.clone(), (revision, record.session_id));
+            }
         }
     }
-    for (key, record) in latest_by_key {
+    for (key, (revision, session_id)) in latest_by_key {
         validate_correlation_key(&key)?;
-        let existing = terminal_scopes.get(&key)?;
-        if existing.is_none_or(|scope| scope.updated_at_ms < record.updated_at_ms) {
-            terminal_scopes.set(key, record.session_id)?;
-        }
+        terminal_scopes.restore_if_newer(&key, &session_id, revision)?;
     }
     Ok(())
 }
 
 fn finish_uncertain(
     inner: &SessionRouterInner,
-    _session_id: &str,
     invocation_id: &str,
+    control: &TurnControlSignal,
+    publisher: &SessionEventPublisher,
     reason: &str,
 ) {
+    let _publication_guard = publisher.lock();
+    control.close();
     let _ = inner
         .reconciliation
         .mark_uncertain(invocation_id, reason.to_string());

rust/crates/sdk/src/session/reconciliation.rs

@@ -53,6 +53,7 @@ pub struct InvocationRecord {
     pub outcome: Option<InvocationOutcome>,
     pub uncertainty_reason: Option<String>,
     pub terminal_event: Option<serde_json::Value>,
+    pub terminal_scope_revision: Option<u64>,
 }
 
 /// File-backed store for invocation lifecycle reconciliation.
@@ -110,6 +111,17 @@ impl InvocationReconciliationStore {
         invocation_id: &str,
         outcome: InvocationOutcome,
         terminal_event: Option<serde_json::Value>,
+    ) -> SdkResult<()> {
+        self.record_terminal_with_scope_revision(invocation_id, outcome, terminal_event, None)
+    }
+
+    /// Records a terminal result together with its durable correlation-scope revision.
+    pub fn record_terminal_with_scope_revision(
+        &self,
+        invocation_id: &str,
+        outcome: InvocationOutcome,
+        terminal_event: Option<serde_json::Value>,
+        terminal_scope_revision: Option<u64>,
     ) -> SdkResult<()> {
         let mut record = self.required_record(invocation_id)?;
         if record.status == InvocationStatus::Terminal {
@@ -122,6 +134,7 @@ impl InvocationReconciliationStore {
         record.outcome = Some(outcome);
         record.uncertainty_reason = None;
         record.terminal_event = terminal_event;
+        record.terminal_scope_revision = terminal_scope_revision;
         record.updated_at_ms = updated_at_ms(record.updated_at_ms);
         self.write_record(&record)
     }
@@ -362,6 +375,7 @@ mod tests {
             outcome: None,
             uncertainty_reason: None,
             terminal_event: None,
+            terminal_scope_revision: None,
         }
     }
 

rust/crates/sdk/src/session/terminal_scope.rs

@@ -3,6 +3,7 @@
 use std::{
     fs,
     path::PathBuf,
+    sync::Mutex,
     time::{SystemTime, UNIX_EPOCH},
 };
 
@@ -13,12 +14,14 @@ use crate::error::{SdkError, SdkResult};
 use super::reconciliation::atomic_write_json;
 
 const MAX_CORRELATION_KEY_BYTES: usize = 512;
+static TERMINAL_SCOPE_WRITE_LOCK: Mutex<()> = Mutex::new(());
 
 /// Persisted terminal-session scope for one correlation key.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct TerminalScopeRecord {
     pub correlation_key: String,
     pub session_id: String,
+    pub revision: u64,
     pub updated_at_ms: u64,
 }
 
@@ -40,24 +43,69 @@ impl TerminalScopeStore {
         &self,
         correlation_key: impl Into<String>,
         session_id: impl Into<String>,
-    ) -> SdkResult<()> {
+    ) -> SdkResult<TerminalScopeRecord> {
         let correlation_key = correlation_key.into();
         validate_correlation_key(&correlation_key)?;
-        let path = self.record_path(&correlation_key);
-        let previous_updated_at = self
-            .get(&correlation_key)?
-            .map_or(0, |record| record.updated_at_ms);
+        let _guard = TERMINAL_SCOPE_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let previous = self.read_record(&correlation_key)?;
+        let revision = previous
+            .as_ref()
+            .map_or(Some(1), |record| record.revision.checked_add(1))
+            .ok_or_else(|| SdkError::internal("terminal scope revision is exhausted"))?;
+        let previous_updated_at = previous.map_or(0, |record| record.updated_at_ms);
         let record = TerminalScopeRecord {
-            correlation_key,
+            correlation_key: correlation_key.clone(),
             session_id: session_id.into(),
+            revision,
             updated_at_ms: current_timestamp_ms().max(previous_updated_at),
         };
-        atomic_write_json(&path, &record)
+        atomic_write_json(&self.record_path(&correlation_key), &record)?;
+        Ok(record)
+    }
+
+    /// Rebuilds a mapping from durable invocation history without moving it backward.
+    pub fn restore_if_newer(
+        &self,
+        correlation_key: &str,
+        session_id: &str,
+        revision: u64,
+    ) -> SdkResult<()> {
+        validate_correlation_key(correlation_key)?;
+        let _guard = TERMINAL_SCOPE_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let previous = self.read_record(correlation_key)?;
+        match previous.as_ref() {
+            Some(record) if record.revision > revision => return Ok(()),
+            Some(record) if record.revision == revision && record.session_id == session_id => {
+                return Ok(())
+            }
+            Some(record) if record.revision == revision => {
+                return Err(SdkError::internal(format!(
+                    "terminal scope revision {revision} conflicts for key '{correlation_key}'"
+                )));
+            }
+            _ => {}
+        }
+        let record = TerminalScopeRecord {
+            correlation_key: correlation_key.to_string(),
+            session_id: session_id.to_string(),
+            revision,
+            updated_at_ms: current_timestamp_ms()
+                .max(previous.map_or(0, |record| record.updated_at_ms)),
+        };
+        atomic_write_json(&self.record_path(correlation_key), &record)
     }
 
     /// Loads the session associated with a correlation key, if present.
     pub fn get(&self, correlation_key: &str) -> SdkResult<Option<TerminalScopeRecord>> {
         validate_correlation_key(correlation_key)?;
+        self.read_record(correlation_key)
+    }
+
+    fn read_record(&self, correlation_key: &str) -> SdkResult<Option<TerminalScopeRecord>> {
         let path = self.record_path(correlation_key);
         let bytes = match fs::read(&path) {
             Ok(bytes) => bytes,
@@ -162,6 +210,26 @@ mod tests {
         assert_eq!(loaded.session_id, "session-1");
     }
 
+    #[test]
+    fn scope_revisions_increase_and_recovery_never_moves_backward() {
+        let root = TestDirectory::new();
+        let store = TerminalScopeStore::new(&root.0);
+        let first = store.set("request-1", "session-1").unwrap();
+        let second = store.set("request-1", "session-2").unwrap();
+        assert_eq!(second.revision, first.revision + 1);
+
+        store
+            .restore_if_newer("request-1", "session-1", first.revision)
+            .expect("stale scope restoration should be ignored");
+        assert_eq!(
+            store.get("request-1").unwrap().unwrap().session_id,
+            "session-2"
+        );
+        assert!(store
+            .restore_if_newer("request-1", "other-session", second.revision)
+            .is_err());
+    }
+
     #[test]
     fn encoded_keys_are_isolated_and_cannot_traverse_paths() {
         let root = TestDirectory::new();

rust/crates/sdk/tests/session_lifecycle.rs

@@ -1,5 +1,5 @@
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{mpsc, Arc, Mutex};
+use std::sync::{mpsc, Arc, Barrier, Mutex};
 use std::thread;
 use std::time::Duration;
 
@@ -897,7 +897,7 @@ fn reconciliation_returns_not_found_for_unknown_invocation() {
 fn startup_recovers_running_invocations_as_uncertain() {
     let (_temp, store, router) = setup();
     let session = router
-        .create_session(None, None)
+        .create_session(None, Some("recover-me"))
         .expect("session should persist");
     let invocation_id = Uuid::new_v4().to_string();
     let reconciliation = sdk::InvocationReconciliationStore::new(
@@ -914,11 +914,16 @@ fn startup_recovers_running_invocations_as_uncertain() {
             outcome: None,
             uncertainty_reason: None,
             terminal_event: None,
+            terminal_scope_revision: None,
         })
         .expect("running record");
     let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).expect("new router");
     let record = restarted.reconcile(&invocation_id).unwrap();
     assert_eq!(record.status, InvocationStatus::Uncertain);
+    let error = restarted
+        .session_for_correlation_key("recover-me")
+        .unwrap_err();
+    assert_eq!(error.code, ErrorCode::Unavailable);
 }
 
 #[test]
@@ -938,6 +943,41 @@ fn terminal_scope_survives_router_restart() {
     assert!(temp.path().exists());
 }
 
+#[tokio::test]
+async fn terminal_scope_write_failure_keeps_outcome_uncertain_and_blocks_stale_routing() {
+    let (started_tx, started_rx) = mpsc::sync_channel(1);
+    let (release_tx, release_rx) = mpsc::sync_channel(1);
+    let runner = Arc::new(GateRunner {
+        started: Mutex::new(Some(started_tx)),
+        release: Mutex::new(release_rx),
+    });
+    let (_temp, store, router) = setup_with(runner);
+    let stale_session = router.create_session(None, Some("repair-key")).unwrap();
+    let current_session = router.create_session(None, None).unwrap();
+    let invocation = router
+        .prompt(&current_session.session_id, "complete", Some("repair-key"))
+        .unwrap();
+    let invocation_id = invocation.invocation_id.clone();
+    started_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+
+    let scope_root = store.sessions_dir().join("sdk").join("terminal-scopes");
+    std::fs::remove_dir_all(&scope_root).unwrap();
+    std::fs::write(&scope_root, "block the scope store").unwrap();
+    release_tx.send(()).unwrap();
+
+    let events = collect_events(invocation).await;
+    assert_eq!(
+        events.last().unwrap().event_type,
+        SessionEventType::AgentEnd
+    );
+    assert_eq!(
+        router.reconcile(&invocation_id).unwrap().status,
+        InvocationStatus::Uncertain
+    );
+    assert!(router.session_for_correlation_key("repair-key").is_err());
+    assert_ne!(stale_session.session_id, current_session.session_id);
+}
+
 #[test]
 fn terminal_scope_keys_do_not_collide() {
     let (_temp, _store, router) = setup();
@@ -958,6 +998,38 @@ fn terminal_scope_keys_do_not_collide() {
     );
 }
 
+#[tokio::test]
+async fn correlation_lookup_rejects_multiple_active_sessions() {
+    let (started_tx, started_rx) = mpsc::sync_channel(1);
+    let (release_tx, release_rx) = mpsc::sync_channel(2);
+    let runner = Arc::new(GateRunner {
+        started: Mutex::new(Some(started_tx)),
+        release: Mutex::new(release_rx),
+    });
+    let (_temp, _store, router) = setup_with(runner);
+    let first = router.create_session(None, Some("active-key")).unwrap();
+    let second = router.create_session(None, Some("active-key")).unwrap();
+    let first_invocation = router
+        .prompt(&first.session_id, "first", Some("active-key"))
+        .unwrap();
+    let second_invocation = router
+        .prompt(&second.session_id, "second", Some("active-key"))
+        .unwrap();
+    started_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+
+    assert_eq!(
+        router
+            .session_for_correlation_key("active-key")
+            .unwrap_err()
+            .code,
+        ErrorCode::Conflict
+    );
+    release_tx.send(()).unwrap();
+    release_tx.send(()).unwrap();
+    collect_events(first_invocation).await;
+    collect_events(second_invocation).await;
+}
+
 #[tokio::test]
 async fn session_router_clones_share_active_invocations() {
     let (started_tx, started_rx) = mpsc::sync_channel(1);
@@ -1274,6 +1346,7 @@ fn new_router_recovers_only_unfinished_invocations() {
             outcome: Some(InvocationOutcome::Completed { message_count: 0 }),
             uncertainty_reason: None,
             terminal_event: Some(json!({ "type": "turn_ended" })),
+            terminal_scope_revision: Some(1),
         })
         .unwrap();
     let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).unwrap();
@@ -1289,6 +1362,43 @@ fn new_router_recovers_only_unfinished_invocations() {
     );
 }
 
+#[test]
+fn terminal_scope_recovery_uses_revision_when_timestamps_tie() {
+    let (_temp, store, router) = setup();
+    let first = router.create_session(None, Some("revision-tie")).unwrap();
+    let latest = router.create_session(None, None).unwrap();
+    let reconciliation = sdk::InvocationReconciliationStore::new(
+        store.sessions_dir().join("sdk").join("reconciliation"),
+    );
+    for (invocation_id, session_id, revision) in [
+        ("invocation-a", first.session_id.as_str(), 1),
+        ("invocation-z", latest.session_id.as_str(), 2),
+    ] {
+        reconciliation
+            .insert(InvocationRecord {
+                invocation_id: invocation_id.to_string(),
+                session_id: session_id.to_string(),
+                correlation_key: Some("revision-tie".to_string()),
+                status: InvocationStatus::Terminal,
+                accepted_at_ms: 10,
+                updated_at_ms: 20,
+                outcome: Some(InvocationOutcome::Completed { message_count: 1 }),
+                uncertainty_reason: None,
+                terminal_event: Some(json!({ "type": "turn_ended" })),
+                terminal_scope_revision: Some(revision),
+            })
+            .unwrap();
+    }
+
+    let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).unwrap();
+    assert_eq!(
+        restarted
+            .session_for_correlation_key("revision-tie")
+            .unwrap(),
+        Some(latest.session_id)
+    );
+}
+
 #[test]
 fn runtime_session_runner_type_is_available_for_runtime_integration() {
     struct EmptyApi;
@@ -1659,6 +1769,59 @@ async fn turn_steer_event_is_sequenced_after_acceptance() {
     assert!(steer_index > 0);
 }
 
+#[tokio::test]
+async fn concurrent_steer_and_completion_preserve_sequence_and_terminal_order() {
+    for _ in 0..16 {
+        let (started_tx, started_rx) = mpsc::sync_channel(1);
+        let (release_tx, release_rx) = mpsc::sync_channel(1);
+        let runner = Arc::new(GateRunner {
+            started: Mutex::new(Some(started_tx)),
+            release: Mutex::new(release_rx),
+        });
+        let (_temp, _store, router) = setup_with(runner);
+        let session = router.create_session(None, None).unwrap();
+        let invocation = router.prompt(&session.session_id, "initial", None).unwrap();
+        let invocation_id = invocation.invocation_id.clone();
+        started_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+
+        let barrier = Arc::new(Barrier::new(3));
+        let steer_router = router.clone();
+        let steer_session_id = session.session_id.clone();
+        let steer_invocation_id = invocation_id.clone();
+        let steer_barrier = Arc::clone(&barrier);
+        let steer_thread = thread::spawn(move || {
+            steer_barrier.wait();
+            steer_router
+                .steer(&steer_session_id, &steer_invocation_id, "racing steer")
+                .is_ok()
+        });
+        let finish_barrier = Arc::clone(&barrier);
+        let finish_thread = thread::spawn(move || {
+            finish_barrier.wait();
+            release_tx.send(()).unwrap();
+        });
+        barrier.wait();
+
+        let steer_accepted = steer_thread.join().unwrap();
+        finish_thread.join().unwrap();
+        let events = collect_events(invocation).await;
+        assert!(events
+            .windows(2)
+            .all(|pair| pair[0].sequence < pair[1].sequence));
+        assert_eq!(
+            events.last().unwrap().event_type,
+            SessionEventType::TurnEnded
+        );
+        let steer_index = events
+            .iter()
+            .position(|event| event.event_type == SessionEventType::TurnSteered);
+        assert_eq!(steer_accepted, steer_index.is_some());
+        if let Some(index) = steer_index {
+            assert!(index < events.len() - 1);
+        }
+    }
+}
+
 #[test]
 fn session_page_types_round_trip_as_json() {
     let page = sdk::SessionPage {