fix(sdk): preserve lifecycle outcomes and output

docs/gjc-compat/gap-matrix.md

@@ -161,52 +161,52 @@ GJC sessions have **three entry points**:
 
 ### claw-code Status
 
-**Current** (as of PR #3330): Partial. SDK transport merged (see SDK Wire Protocol). Local session lifecycle exists in `runtime::Session`. Remaining: operation handlers, session router with durable lookup, invocation reconciliation store, terminal scope tracking.
+**Current**: The SDK now exposes `sdk::session::SessionRouter` with durable `turn.create`, `turn.prompt`, `turn.steer`, `turn.abort_and_prompt`, and paginated `turn.list` handling. It bridges `runtime::Session`/`ConversationRuntime`, streams ordered lifecycle events, persists invocation outcomes for reconciliation, and stores terminal session scope by correlation key. The WebSocket transport remains a client transport; a remotely hosted broker and automatic recovery of an unknown post-crash outcome remain outside this slice.
 
 **Files**:
 - `rust/crates/runtime/src/session.rs` (1,961 LOC)
-- `rust/crates/runtime/src/conversation.rs` (1,878 LOC)
+- `rust/crates/runtime/src/conversation.rs` (conversation loop and lifecycle event bridge)
 - `rust/crates/runtime/src/session_control.rs` (file ops, migration)
 - `rust/crates/runtime/src/lib.rs` (re-exports `Session`, `SessionStore`, `ConversationRuntime`)
+- `rust/crates/sdk/src/session/` (router, event stream, reconciliation, terminal scope)
 
 **Lines**: ~3,900 LOC (session + conversation)
 
 **Reality**:
-- `Session` struct holds in-memory state (messages, usage, context).
-- `SessionStore` reads/writes `.claw/sessions/{id}/` directory.
-- `ConversationRuntime` drives the local agent loop (sync).
-- No WebSocket or remote access.
-- No invocation reconciliation or uncertain outcome recovery.
-- No terminal scope tracking (per-session-per-correlation).
+- `Session` holds persisted conversation state and `SessionStore` resolves session IDs in the workspace namespace.
+- `ConversationRuntime` drives the local agent loop and emits lifecycle events to an optional callback.
+- `sdk::session::SessionRouter` dispatches the five turn operations and exposes a per-invocation event stream.
+- Invocation terminal outcomes and correlation-key terminal scopes are durable; startup recovery marks unfinished invocations uncertain for later reconciliation.
+- A remote WebSocket server/broker and inference of an outcome that was never durably recorded are still absent.
 
 ### Gap Assessment
 
 | Component | GJC | claw-code | Gap | Complexity |
 |-----------|-----|-----------|-----|------------|
-| Remote Session Router | 2,500 LOC | — | 100% | High |
-| Invocation Reconciliation | 1,200 LOC | — | 100% | High |
-| Terminal Scope Tracking | 600 LOC | — | 100% | Medium |
-| Async Event Streaming | 800 LOC | Partial | 60% | Medium |
-| Uncertain Outcome Recovery | 1,000 LOC | — | 100% | High |
-| **Total** | **6,100 LOC** | **3,900** | **2,200 LOC** | — |
+| Remote Session Router | 2,500 LOC | Local router/dispatch | Partial | High |
+| Invocation Reconciliation | 1,200 LOC | Durable invocation records | Partial | High |
+| Terminal Scope Tracking | 600 LOC | Implemented | — | Medium |
+| Async Event Streaming | 800 LOC | Local per-invocation stream | Partial | Medium |
+| Uncertain Outcome Recovery | 1,000 LOC | Unfinished records recovered as uncertain | Partial | High |
+| **Total** | **6,100 LOC** | **Session lifecycle slice implemented** | **Remote broker remains** | — |
 
 ### Proposed Implementation
 
 **Owner Crate**: `sdk` (new, or extend `runtime`)
 
-**Module Structure** (add to `sdk/session/`):
+**Module Structure** (`sdk/session/`):
 ```
 sdk/session/
-├── router.rs            (session routing, durable lookup)
+├── mod.rs                (session routing and operation dispatch)
 ├── reconciliation.rs    (uncertain → terminal recovery)
 ├── terminal_scope.rs    (per-correlation state retention)
-└── events.rs            (event publishing, reconciliation)
+└── events.rs            (ordered lifecycle event types)
 ```
 
 **Integration Points**:
-- Bridge `runtime::Session` into `sdk::Session` wrapper
-- Add `runtime::SessionStore` extensions for reconciliation data
-- Extend `ConversationRuntime` with event emission
+- Bridge `runtime::Session` and `ConversationRuntime` into SDK lifecycle operations
+- Persist reconciliation and terminal-scope data alongside the session namespace
+- Emit lifecycle callbacks from `ConversationRuntime`
 
 **Size Estimate**: 2,500 LOC (Rust, new + modifications)
 

rust/Cargo.lock

@@ -2337,8 +2337,10 @@ name = "sdk"
 version = "0.1.4"
 dependencies = [
  "futures-util",
+ "runtime",
  "serde",
  "serde_json",
+ "tempfile",
  "tokio",
  "tokio-tungstenite",
  "tracing",

rust/crates/runtime/src/lib.rs

@@ -84,8 +84,8 @@ pub use config_validate::{
 };
 pub use conversation::{
     auto_compaction_threshold_from_env, ApiClient, ApiRequest, AssistantEvent, AutoCompactionEvent,
-    ConversationRuntime, PromptCacheEvent, RuntimeError, StaticToolExecutor, ToolError,
-    ToolExecutor, TurnInterruptSignal, TurnSummary,
+    ConversationEvent, ConversationRuntime, PromptCacheEvent, RuntimeError, StaticToolExecutor,
+    ToolError, ToolExecutor, TurnControlSignal, TurnInterruptSignal, TurnSummary,
 };
 pub use file_ops::{
     edit_file, edit_file_in_workspace, glob_search, glob_search_in_workspace, grep_search,

rust/crates/sdk/Cargo.toml

@@ -13,3 +13,10 @@ serde = { version = "1", features = ["derive"] }
 uuid = { version = "1", features = ["v4", "serde"] }
 tracing = "0.1"
 url = "2"
+runtime = { path = "../runtime" }
+
+[dev-dependencies]
+tempfile = "3"
+
+[lints]
+workspace = true

rust/crates/sdk/src/error.rs

@@ -15,6 +15,8 @@ pub enum SdkErrorCode {
     Unauthorized,
     EndpointCredentialForbidden,
     Unavailable,
+    /// Requested session state conflicts with an active invocation
+    Conflict,
     ConnectionError,
     ConnectionClosed,
     Timeout,
@@ -37,6 +39,7 @@ impl SdkErrorCode {
             Self::Unauthorized => "unauthorized",
             Self::EndpointCredentialForbidden => "endpoint_credential_forbidden",
             Self::Unavailable => "unavailable",
+            Self::Conflict => "conflict",
             Self::ConnectionError => "connection_error",
             Self::ConnectionClosed => "connection_closed",
             Self::Timeout => "timeout",
@@ -59,6 +62,7 @@ impl SdkErrorCode {
             "unauthorized" => Self::Unauthorized,
             "endpoint_credential_forbidden" => Self::EndpointCredentialForbidden,
             "unavailable" => Self::Unavailable,
+            "conflict" => Self::Conflict,
             "connection_error" => Self::ConnectionError,
             "connection_closed" => Self::ConnectionClosed,
             "timeout" => Self::Timeout,
@@ -132,6 +136,14 @@ impl SdkError {
         Self::new(SdkErrorCode::Unavailable, message)
     }
 
+    pub fn not_found(message: impl Into<String>) -> Self {
+        Self::new(SdkErrorCode::NotFound, message)
+    }
+
+    pub fn conflict(message: impl Into<String>) -> Self {
+        Self::new(SdkErrorCode::Conflict, message)
+    }
+
     pub fn connection_error(message: impl Into<String>) -> Self {
         Self::new(SdkErrorCode::ConnectionError, message)
     }
@@ -209,6 +221,13 @@ mod tests {
         );
     }
 
+    #[test]
+    fn conflict_error_code_round_trips_on_the_wire() {
+        let error = SdkError::conflict("session is active");
+        assert_eq!(error.code.to_string(), "conflict");
+        assert_eq!(SdkErrorCode::from_wire("conflict"), SdkErrorCode::Conflict);
+    }
+
     #[test]
     fn unknown_remote_code_is_preserved_with_typed_internal_code() {
         let error = SdkError::from_remote("future_error", "failed");

rust/crates/sdk/src/lib.rs

@@ -14,9 +14,16 @@
 pub mod error;
 pub mod frame;
 pub mod registry;
+pub mod session;
 pub mod transport;
 
 pub use error::{SdkError, SdkErrorCode, SdkResult, SdkSentRecord};
 pub use frame::{FrameType, SdkFrame};
 pub use registry::{OperationMetadata, OperationRegistry, Schema};
+pub use session::{
+    InvocationOutcome, InvocationReconciliationStore, InvocationRecord, InvocationStatus,
+    RuntimeSessionRunner, SessionDispatch, SessionEvent, SessionEventStream, SessionEventType,
+    SessionInvocation, SessionPage, SessionRouter, SessionRunResult, SessionSummary,
+    SessionTurnRunner, TerminalScopeRecord, TerminalScopeStore,
+};
 pub use transport::{SdkClient, SdkClientConfig};

rust/crates/sdk/src/registry.rs

@@ -263,6 +263,89 @@ impl OperationRegistry {
                 )),
         );
 
+        // SDK session lifecycle operations.
+        for (name, description, required) in [
+            (
+                "turn.create",
+                "Create a durable SDK session",
+                Vec::<&str>::new(),
+            ),
+            (
+                "turn.prompt",
+                "Prompt a session and stream lifecycle events",
+                Vec::<&str>::new(),
+            ),
+            (
+                "turn.steer",
+                "Steer an active session invocation",
+                vec!["invocationId"],
+            ),
+            (
+                "turn.abort_and_prompt",
+                "Abort the active invocation and start a replacement prompt",
+                Vec::<&str>::new(),
+            ),
+            (
+                "turn.list",
+                "List persisted SDK sessions",
+                Vec::<&str>::new(),
+            ),
+        ] {
+            let mut properties = serde_json::Map::new();
+            properties.insert("sessionId".to_string(), json!({ "type": "string" }));
+            properties.insert("correlationKey".to_string(), json!({ "type": "string" }));
+            properties.insert("invocationId".to_string(), json!({ "type": "string" }));
+            properties.insert("prompt".to_string(), json!({ "type": "string" }));
+            properties.insert(
+                "override".to_string(),
+                json!({ "type": ["string", "object", "array"] }),
+            );
+            properties.insert(
+                "request".to_string(),
+                json!({ "type": ["string", "object", "array"] }),
+            );
+            properties.insert(
+                "input".to_string(),
+                json!({ "type": ["string", "object", "array"] }),
+            );
+            properties.insert("tools".to_string(), json!({ "type": "array" }));
+            properties.insert("settings".to_string(), json!({ "type": "object" }));
+            properties.insert("model".to_string(), json!({ "type": "string" }));
+            properties.insert("cursor".to_string(), json!({ "type": "string" }));
+            properties.insert(
+                "limit".to_string(),
+                json!({ "type": "integer", "minimum": 1 }),
+            );
+            let mut input_schema = json!({
+                "type": "object",
+                "properties": properties,
+                "required": required,
+            });
+            let prompt_alternatives = json!([
+                { "required": ["prompt"] },
+                { "required": ["request"] },
+                { "required": ["input"] },
+            ]);
+            match name {
+                "turn.prompt" | "turn.abort_and_prompt" => {
+                    input_schema["anyOf"] = prompt_alternatives;
+                }
+                "turn.steer" => {
+                    input_schema["anyOf"] = json!([
+                        { "required": ["override"] },
+                        { "required": ["prompt"] },
+                        { "required": ["request"] },
+                        { "required": ["input"] },
+                    ]);
+                }
+                _ => {}
+            }
+            registry.register(
+                OperationMetadata::new(name, description)
+                    .with_input_schema(Schema::new(format!("Input for {name}"), input_schema)),
+            );
+        }
+
         registry
     }
 
@@ -397,6 +480,20 @@ mod tests {
         assert!(registry.has_operation("events.replay"));
     }
 
+    #[test]
+    fn standard_registry_has_session_lifecycle_operations() {
+        let registry = OperationRegistry::with_standard_operations();
+        for operation in [
+            "turn.create",
+            "turn.prompt",
+            "turn.steer",
+            "turn.abort_and_prompt",
+            "turn.list",
+        ] {
+            assert!(registry.has_operation(operation), "missing {operation}");
+        }
+    }
+
     #[test]
     fn registry_validation_catches_missing_required_fields() {
         let registry = OperationRegistry::with_standard_operations();

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

@@ -0,0 +1,31 @@
+//! Event stream types for session lifecycle operations.
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+/// Lifecycle events published for one SDK turn invocation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SessionEventType {
+    TurnAccepted,
+    AgentBegin,
+    ToolUse,
+    ToolResult,
+    AgentEnd,
+    TurnSteered,
+    TurnEnded,
+}
+
+/// One ordered session lifecycle event.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionEvent {
+    pub event_id: String,
+    pub event_type: SessionEventType,
+    pub session_id: String,
+    pub invocation_id: String,
+    pub sequence: u64,
+    pub timestamp_ms: u64,
+    pub durable: bool,
+    pub payload: Value,
+}

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

@@ -0,0 +1,673 @@
+//! Durable invocation state for recovering SDK sessions across process restarts.
+
+use std::{
+    fs::{self, OpenOptions},
+    io::Write,
+    path::{Path, PathBuf},
+    sync::Mutex,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+#[cfg(unix)]
+use std::fs::File;
+
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::error::{SdkError, SdkResult};
+
+static RECONCILIATION_WRITE_LOCK: Mutex<()> = Mutex::new(());
+
+/// Lifecycle state persisted for an invocation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum InvocationStatus {
+    Accepted,
+    Running,
+    Uncertain,
+    Terminal,
+}
+
+/// Result captured when an invocation reaches a terminal state.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum InvocationOutcome {
+    Completed {
+        message_count: usize,
+    },
+    Aborted {
+        message_count: usize,
+    },
+    Failed {
+        message: String,
+        message_count: usize,
+    },
+}
+
+/// Persisted lifecycle and reconciliation data for one invocation.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct InvocationRecord {
+    pub invocation_id: String,
+    pub session_id: String,
+    pub correlation_key: Option<String>,
+    pub status: InvocationStatus,
+    pub accepted_at_ms: u64,
+    pub updated_at_ms: u64,
+    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.
+#[derive(Debug, Clone)]
+pub struct InvocationReconciliationStore {
+    root: PathBuf,
+}
+
+impl InvocationReconciliationStore {
+    /// Creates a store handle without creating its root directory.
+    #[must_use]
+    pub fn new(root: impl Into<PathBuf>) -> Self {
+        Self { root: root.into() }
+    }
+
+    /// Inserts a new invocation record. Existing IDs are not overwritten.
+    pub fn insert(&self, record: InvocationRecord) -> SdkResult<()> {
+        let _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let path = self.record_path(&record.invocation_id)?;
+        if self.get(&record.invocation_id)?.is_some() {
+            return Err(SdkError::invalid_input(format!(
+                "invocation '{}' already exists",
+                record.invocation_id
+            )));
+        }
+        atomic_write_json(&path, &record)
+    }
+
+    /// Marks a non-terminal invocation as running.
+    pub fn mark_running(&self, invocation_id: &str) -> SdkResult<()> {
+        let _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut record = self.required_record(invocation_id)?;
+        if record.status != InvocationStatus::Accepted {
+            return Err(SdkError::conflict(format!(
+                "invocation '{}' cannot start from state {:?}",
+                invocation_id, record.status
+            )));
+        }
+        record.status = InvocationStatus::Running;
+        record.uncertainty_reason = None;
+        record.updated_at_ms = updated_at_ms(record.updated_at_ms);
+        self.write_record(&record)
+    }
+
+    /// Marks a non-terminal invocation as uncertain with a recovery reason.
+    pub fn mark_uncertain(&self, invocation_id: &str, reason: impl Into<String>) -> SdkResult<()> {
+        let reason = reason.into();
+        self.update_nonterminal(invocation_id, move |record| {
+            record.status = InvocationStatus::Uncertain;
+            record.uncertainty_reason = Some(reason);
+        })
+    }
+
+    /// Records a terminal result. Once terminal, a record cannot be changed again.
+    pub fn record_terminal(
+        &self,
+        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 _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut record = self.required_record(invocation_id)?;
+        if record.status == InvocationStatus::Terminal {
+            return Err(SdkError::invalid_input(format!(
+                "invocation '{}' is already terminal",
+                invocation_id
+            )));
+        }
+        if record
+            .outcome
+            .as_ref()
+            .is_some_and(|pending| pending != &outcome)
+            || record
+                .terminal_event
+                .as_ref()
+                .is_some_and(|pending| Some(pending) != terminal_event.as_ref())
+        {
+            return Err(SdkError::conflict(format!(
+                "invocation '{}' has a different pending terminal result",
+                invocation_id
+            )));
+        }
+        record.status = InvocationStatus::Terminal;
+        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)
+    }
+
+    /// Persists a known turn result before its correlation scope is updated.
+    ///
+    /// The record remains non-terminal until the scope update succeeds, allowing
+    /// startup or a later reconciliation read to safely finish the transition.
+    pub fn record_pending_terminal(
+        &self,
+        invocation_id: &str,
+        outcome: InvocationOutcome,
+        terminal_event: serde_json::Value,
+    ) -> SdkResult<()> {
+        let _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut record = self.required_record(invocation_id)?;
+        if record.status == InvocationStatus::Terminal {
+            return Err(SdkError::invalid_input(format!(
+                "invocation '{}' is already terminal",
+                invocation_id
+            )));
+        }
+        if record.outcome.is_some() || record.terminal_event.is_some() {
+            return Err(SdkError::conflict(format!(
+                "invocation '{}' already has a pending terminal result",
+                invocation_id
+            )));
+        }
+        record.outcome = Some(outcome);
+        record.terminal_event = Some(terminal_event);
+        record.terminal_scope_revision = None;
+        record.updated_at_ms = updated_at_ms(record.updated_at_ms);
+        self.write_record(&record)
+    }
+
+    /// Loads one invocation record, if present.
+    pub fn get(&self, invocation_id: &str) -> SdkResult<Option<InvocationRecord>> {
+        let path = self.record_path(invocation_id)?;
+        read_record(&path, invocation_id)
+    }
+
+    /// Loads all invocation records in stable invocation-ID order.
+    pub fn list(&self) -> SdkResult<Vec<InvocationRecord>> {
+        let entries = match fs::read_dir(&self.root) {
+            Ok(entries) => entries,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(error) => {
+                return Err(io_error(
+                    "read invocation store directory",
+                    &self.root,
+                    error,
+                ))
+            }
+        };
+
+        let mut records = Vec::new();
+        for entry in entries {
+            let entry = entry
+                .map_err(|error| io_error("read invocation store entry", &self.root, error))?;
+            if !entry
+                .file_type()
+                .map_err(|error| io_error("inspect invocation store entry", &entry.path(), error))?
+                .is_file()
+            {
+                continue;
+            }
+
+            let path = entry.path();
+            if path.extension() != Some(std::ffi::OsStr::new("json")) {
+                continue;
+            }
+            let invocation_id = path
+                .file_stem()
+                .and_then(|stem| stem.to_str())
+                .ok_or_else(|| SdkError::invalid_input("invocation filename is not valid UTF-8"))?;
+            validate_invocation_id(invocation_id)?;
+            if let Some(record) = read_record(&path, invocation_id)? {
+                records.push(record);
+            }
+        }
+        records.sort_by(|left, right| left.invocation_id.cmp(&right.invocation_id));
+        Ok(records)
+    }
+
+    /// Loads records associated with the exact correlation key.
+    pub fn list_by_correlation_key(&self, key: &str) -> SdkResult<Vec<InvocationRecord>> {
+        Ok(self
+            .list()?
+            .into_iter()
+            .filter(|record| record.correlation_key.as_deref() == Some(key))
+            .collect())
+    }
+
+    /// Durably marks accepted and running records uncertain after a restart.
+    ///
+    /// Returns the records changed by this recovery pass; already uncertain and
+    /// terminal records are left untouched and are not included in the result.
+    pub fn reconcile_unfinished(&self) -> SdkResult<Vec<InvocationRecord>> {
+        let _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut recovered = Vec::new();
+        for mut record in self.list()? {
+            if matches!(
+                record.status,
+                InvocationStatus::Accepted | InvocationStatus::Running
+            ) {
+                record.status = InvocationStatus::Uncertain;
+                record.uncertainty_reason =
+                    Some("recovered unfinished invocation after restart".into());
+                record.updated_at_ms = updated_at_ms(record.updated_at_ms);
+                self.write_record(&record)?;
+                recovered.push(record);
+            }
+        }
+        Ok(recovered)
+    }
+
+    fn record_path(&self, invocation_id: &str) -> SdkResult<PathBuf> {
+        validate_invocation_id(invocation_id)?;
+        Ok(self.root.join(format!("{invocation_id}.json")))
+    }
+
+    fn required_record(&self, invocation_id: &str) -> SdkResult<InvocationRecord> {
+        self.get(invocation_id)?.ok_or_else(|| {
+            SdkError::invalid_input(format!("invocation '{invocation_id}' does not exist"))
+        })
+    }
+
+    fn update_nonterminal(
+        &self,
+        invocation_id: &str,
+        update: impl FnOnce(&mut InvocationRecord),
+    ) -> SdkResult<()> {
+        let _guard = RECONCILIATION_WRITE_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut record = self.required_record(invocation_id)?;
+        if record.status == InvocationStatus::Terminal {
+            return Err(SdkError::invalid_input(format!(
+                "invocation '{}' is already terminal",
+                invocation_id
+            )));
+        }
+        update(&mut record);
+        record.updated_at_ms = updated_at_ms(record.updated_at_ms);
+        self.write_record(&record)
+    }
+
+    fn write_record(&self, record: &InvocationRecord) -> SdkResult<()> {
+        let path = self.record_path(&record.invocation_id)?;
+        atomic_write_json(&path, record)
+    }
+}
+
+/// Atomically serializes a value beside its target and renames it into place.
+pub(super) fn atomic_write_json<T: Serialize + ?Sized>(path: &Path, value: &T) -> SdkResult<()> {
+    let bytes = serde_json::to_vec(value)
+        .map_err(|error| SdkError::internal(format!("serialize {}: {error}", path.display())))?;
+    let parent = path
+        .parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
+        .unwrap_or_else(|| Path::new("."));
+    fs::create_dir_all(parent)
+        .map_err(|error| io_error("create store directory", parent, error))?;
+    let filename = path
+        .file_name()
+        .ok_or_else(|| SdkError::invalid_input("store path has no filename"))?;
+    let temporary_path = parent.join(format!(
+        ".{}.{}.tmp",
+        filename.to_string_lossy(),
+        Uuid::new_v4()
+    ));
+
+    let write_result = (|| -> std::io::Result<()> {
+        let mut temporary = OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&temporary_path)?;
+        temporary.write_all(&bytes)?;
+        temporary.sync_all()?;
+        drop(temporary);
+        fs::rename(&temporary_path, path)?;
+        #[cfg(unix)]
+        File::open(parent)?.sync_all()?;
+        Ok(())
+    })();
+
+    match write_result {
+        Ok(()) => Ok(()),
+        Err(error) => {
+            let _ = fs::remove_file(&temporary_path);
+            Err(io_error("atomically write JSON", path, error))
+        }
+    }
+}
+
+fn read_record(path: &Path, expected_id: &str) -> SdkResult<Option<InvocationRecord>> {
+    let bytes = match fs::read(path) {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(io_error("read invocation record", path, error)),
+    };
+    let record: InvocationRecord = serde_json::from_slice(&bytes).map_err(|error| {
+        SdkError::internal(format!(
+            "parse invocation record {}: {error}",
+            path.display()
+        ))
+    })?;
+    validate_invocation_id(&record.invocation_id)?;
+    if record.invocation_id != expected_id {
+        return Err(SdkError::internal(format!(
+            "invocation record ID '{}' does not match filename ID '{expected_id}'",
+            record.invocation_id
+        )));
+    }
+    Ok(Some(record))
+}
+
+fn validate_invocation_id(invocation_id: &str) -> SdkResult<()> {
+    if invocation_id.is_empty()
+        || invocation_id.len() > 200
+        || !invocation_id
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
+    {
+        return Err(SdkError::invalid_input(format!(
+            "unsafe invocation ID '{invocation_id}'"
+        )));
+    }
+    Ok(())
+}
+
+fn updated_at_ms(previous: u64) -> u64 {
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis()
+        .min(u128::from(u64::MAX)) as u64;
+    now.max(previous.saturating_add(1))
+}
+
+fn io_error(action: &str, path: &Path, error: std::io::Error) -> SdkError {
+    SdkError::internal(format!("{action} '{}': {error}", path.display()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct TestDirectory(PathBuf);
+
+    impl TestDirectory {
+        fn new() -> Self {
+            let path = std::env::temp_dir().join(format!("sdk-session-{}", Uuid::new_v4()));
+            fs::create_dir(&path).expect("create test directory");
+            Self(path)
+        }
+    }
+
+    impl Drop for TestDirectory {
+        fn drop(&mut self) {
+            let _ = fs::remove_dir_all(&self.0);
+        }
+    }
+
+    fn accepted_record(invocation_id: &str) -> InvocationRecord {
+        InvocationRecord {
+            invocation_id: invocation_id.into(),
+            session_id: "session-a".into(),
+            correlation_key: Some("request-a".into()),
+            status: InvocationStatus::Accepted,
+            accepted_at_ms: 10,
+            updated_at_ms: 10,
+            outcome: None,
+            uncertainty_reason: None,
+            terminal_event: None,
+            terminal_scope_revision: None,
+        }
+    }
+
+    #[test]
+    fn constructor_has_no_filesystem_side_effects() {
+        let root = std::env::temp_dir().join(format!("sdk-session-{}", Uuid::new_v4()));
+        let _store = InvocationReconciliationStore::new(&root);
+        assert!(!root.exists());
+    }
+
+    #[test]
+    fn invocation_records_survive_store_restart() {
+        let root = TestDirectory::new();
+        let record = accepted_record("invocation-a");
+        InvocationReconciliationStore::new(&root.0)
+            .insert(record.clone())
+            .expect("insert record");
+
+        let loaded = InvocationReconciliationStore::new(&root.0)
+            .get("invocation-a")
+            .expect("load record")
+            .expect("record exists");
+        assert_eq!(loaded, record);
+    }
+
+    #[test]
+    fn running_transition_only_accepts_an_accepted_invocation() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        store
+            .insert(accepted_record("invocation-a"))
+            .expect("insert");
+        store.mark_running("invocation-a").expect("mark running");
+        assert_eq!(
+            store.get("invocation-a").unwrap().unwrap().status,
+            InvocationStatus::Running
+        );
+        assert!(store.mark_running("invocation-a").is_err());
+        store
+            .mark_uncertain("invocation-a", "recovered")
+            .expect("mark uncertain");
+        assert!(store.mark_running("invocation-a").is_err());
+        assert_eq!(
+            store.get("invocation-a").unwrap().unwrap().status,
+            InvocationStatus::Uncertain
+        );
+    }
+
+    #[test]
+    fn uncertain_invocation_can_become_terminal_only_once() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        store
+            .insert(accepted_record("invocation-a"))
+            .expect("insert");
+        store
+            .mark_uncertain("invocation-a", "runtime disconnected")
+            .expect("mark uncertain");
+        store
+            .record_terminal(
+                "invocation-a",
+                InvocationOutcome::Completed { message_count: 2 },
+                Some(serde_json::json!({"type": "completed"})),
+            )
+            .expect("record terminal");
+
+        let terminal = store.get("invocation-a").expect("load").expect("exists");
+        assert_eq!(terminal.status, InvocationStatus::Terminal);
+        assert_eq!(terminal.uncertainty_reason, None);
+        assert_eq!(
+            terminal.outcome,
+            Some(InvocationOutcome::Completed { message_count: 2 })
+        );
+        assert!(store
+            .record_terminal(
+                "invocation-a",
+                InvocationOutcome::Aborted { message_count: 0 },
+                None,
+            )
+            .is_err());
+        assert_eq!(store.get("invocation-a").expect("reload"), Some(terminal));
+    }
+
+    #[test]
+    fn recovery_marks_only_accepted_and_running_records_uncertain() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        store
+            .insert(accepted_record("accepted"))
+            .expect("insert accepted");
+        let mut running = accepted_record("running");
+        running.status = InvocationStatus::Running;
+        store.insert(running).expect("insert running");
+        let mut terminal = accepted_record("terminal");
+        terminal.status = InvocationStatus::Terminal;
+        terminal.outcome = Some(InvocationOutcome::Aborted { message_count: 0 });
+        store.insert(terminal.clone()).expect("insert terminal");
+
+        let recovered = store.reconcile_unfinished().expect("reconcile");
+        assert_eq!(recovered.len(), 2);
+        assert!(recovered
+            .iter()
+            .all(|record| record.status == InvocationStatus::Uncertain));
+        assert_eq!(
+            store.get("terminal").expect("load terminal"),
+            Some(terminal)
+        );
+        assert!(store
+            .reconcile_unfinished()
+            .expect("repeat recovery")
+            .is_empty());
+    }
+
+    #[test]
+    fn pending_terminal_result_survives_uncertainty_and_can_be_finalized() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        store
+            .insert(accepted_record("pending"))
+            .expect("insert accepted record");
+        store.mark_running("pending").expect("mark running");
+        store
+            .record_pending_terminal(
+                "pending",
+                InvocationOutcome::Completed { message_count: 3 },
+                serde_json::json!({"eventType": "turn_ended"}),
+            )
+            .expect("record pending result");
+        store
+            .mark_uncertain("pending", "terminal scope store was unavailable")
+            .expect("mark uncertain");
+
+        let pending = store.get("pending").unwrap().unwrap();
+        assert_eq!(pending.status, InvocationStatus::Uncertain);
+        assert_eq!(
+            pending.outcome,
+            Some(InvocationOutcome::Completed { message_count: 3 })
+        );
+        store
+            .record_terminal_with_scope_revision(
+                "pending",
+                pending.outcome.unwrap(),
+                pending.terminal_event,
+                Some(1),
+            )
+            .expect("finalize after recovery");
+        assert_eq!(
+            store.get("pending").unwrap().unwrap().status,
+            InvocationStatus::Terminal
+        );
+    }
+
+    #[test]
+    fn concurrent_terminal_and_uncertain_transitions_preserve_terminal_result() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        store
+            .insert(accepted_record("concurrent"))
+            .expect("insert accepted record");
+        store.mark_running("concurrent").expect("mark running");
+
+        let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
+        let terminal_store = store.clone();
+        let terminal_barrier = std::sync::Arc::clone(&barrier);
+        let terminal = std::thread::spawn(move || {
+            terminal_barrier.wait();
+            terminal_store.record_terminal(
+                "concurrent",
+                InvocationOutcome::Completed { message_count: 1 },
+                None,
+            )
+        });
+        let uncertain_store = store.clone();
+        let uncertain_barrier = std::sync::Arc::clone(&barrier);
+        let uncertain = std::thread::spawn(move || {
+            uncertain_barrier.wait();
+            uncertain_store.mark_uncertain("concurrent", "raced with terminal write")
+        });
+        barrier.wait();
+        let _ = terminal.join().expect("terminal writer thread");
+        let _ = uncertain.join().expect("uncertain writer thread");
+
+        let record = store.get("concurrent").unwrap().unwrap();
+        assert_eq!(record.status, InvocationStatus::Terminal);
+        assert_eq!(
+            record.outcome,
+            Some(InvocationOutcome::Completed { message_count: 1 })
+        );
+    }
+
+    #[test]
+    fn concurrent_duplicate_inserts_have_only_one_winner() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
+        let first_store = store.clone();
+        let first_barrier = std::sync::Arc::clone(&barrier);
+        let first = std::thread::spawn(move || {
+            first_barrier.wait();
+            first_store.insert(accepted_record("duplicate"))
+        });
+        let second_store = store.clone();
+        let second_barrier = std::sync::Arc::clone(&barrier);
+        let second = std::thread::spawn(move || {
+            second_barrier.wait();
+            second_store.insert(accepted_record("duplicate"))
+        });
+        barrier.wait();
+        let first = first.join().expect("first insert thread");
+        let second = second.join().expect("second insert thread");
+
+        assert_ne!(first.is_ok(), second.is_ok());
+        assert_eq!(
+            store.get("duplicate").unwrap().unwrap().invocation_id,
+            "duplicate"
+        );
+    }
+
+    #[test]
+    fn malformed_ids_and_json_are_rejected() {
+        let root = TestDirectory::new();
+        let store = InvocationReconciliationStore::new(&root.0);
+        assert!(store.get("../escape").is_err());
+        assert!(store.insert(accepted_record("bad/id")).is_err());
+
+        fs::write(root.0.join("malformed.json"), "{").expect("write malformed record");
+        assert!(store.get("malformed").is_err());
+        assert!(store.list().is_err());
+    }
+}

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

@@ -0,0 +1,310 @@
+//! Durable mapping from SDK correlation keys to their active terminal session.
+
+use std::{
+    fs,
+    path::PathBuf,
+    sync::Mutex,
+    time::{SystemTime, UNIX_EPOCH},
+};
+
+use serde::{Deserialize, Serialize};
+
+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,
+}
+
+/// File-backed mapping of correlation keys to terminal sessions.
+#[derive(Debug, Clone)]
+pub struct TerminalScopeStore {
+    root: PathBuf,
+}
+
+impl TerminalScopeStore {
+    /// Creates a store handle without creating its root directory.
+    #[must_use]
+    pub fn new(root: impl Into<PathBuf>) -> Self {
+        Self { root: root.into() }
+    }
+
+    /// Sets the session associated with a correlation key.
+    pub fn set(
+        &self,
+        correlation_key: impl Into<String>,
+        session_id: impl Into<String>,
+    ) -> SdkResult<TerminalScopeRecord> {
+        let correlation_key = correlation_key.into();
+        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)?;
+        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.clone(),
+            session_id: session_id.into(),
+            revision,
+            updated_at_ms: current_timestamp_ms().max(previous_updated_at.saturating_add(1)),
+        };
+        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)
+                    .saturating_add(1),
+            ),
+        };
+        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,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(error) => {
+                return Err(SdkError::internal(format!(
+                    "read terminal scope '{}': {error}",
+                    path.display()
+                )));
+            }
+        };
+        let record: TerminalScopeRecord = serde_json::from_slice(&bytes).map_err(|error| {
+            SdkError::internal(format!(
+                "parse terminal scope '{}': {error}",
+                path.display()
+            ))
+        })?;
+        if record.correlation_key != correlation_key {
+            return Err(SdkError::internal(format!(
+                "terminal scope key does not match requested key for '{}'",
+                path.display()
+            )));
+        }
+        Ok(Some(record))
+    }
+
+    fn record_path(&self, correlation_key: &str) -> PathBuf {
+        const MAX_COMPONENT_BYTES: usize = 180;
+        let encoded = encode_key_as_hex(correlation_key);
+        let mut path = self.root.clone();
+        let mut remaining = encoded.as_str();
+        while remaining.len() > MAX_COMPONENT_BYTES {
+            let (component, tail) = remaining.split_at(MAX_COMPONENT_BYTES);
+            path.push(component);
+            remaining = tail;
+        }
+        path.push(format!("{remaining}.json"));
+        path
+    }
+}
+
+fn validate_correlation_key(correlation_key: &str) -> SdkResult<()> {
+    if correlation_key.len() <= MAX_CORRELATION_KEY_BYTES {
+        Ok(())
+    } else {
+        Err(SdkError::invalid_input("correlation key exceeds 512 bytes"))
+    }
+}
+
+fn encode_key_as_hex(key: &str) -> String {
+    const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef";
+    let mut encoded = String::with_capacity(key.len() * 2);
+    for byte in key.as_bytes() {
+        encoded.push(char::from(HEX_DIGITS[usize::from(*byte >> 4)]));
+        encoded.push(char::from(HEX_DIGITS[usize::from(*byte & 0x0f)]));
+    }
+    encoded
+}
+
+fn current_timestamp_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis()
+        .min(u128::from(u64::MAX)) as u64
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs;
+    use uuid::Uuid;
+
+    struct TestDirectory(PathBuf);
+
+    impl TestDirectory {
+        fn new() -> Self {
+            let path = std::env::temp_dir().join(format!("sdk-terminal-scope-{}", Uuid::new_v4()));
+            fs::create_dir(&path).expect("create test directory");
+            Self(path)
+        }
+    }
+
+    impl Drop for TestDirectory {
+        fn drop(&mut self) {
+            let _ = fs::remove_dir_all(&self.0);
+        }
+    }
+
+    #[test]
+    fn terminal_scope_survives_store_restart() {
+        let root = TestDirectory::new();
+        TerminalScopeStore::new(&root.0)
+            .set("request-1", "session-1")
+            .expect("set terminal scope");
+
+        let loaded = TerminalScopeStore::new(&root.0)
+            .get("request-1")
+            .expect("load terminal scope")
+            .expect("scope exists");
+        assert_eq!(loaded.correlation_key, "request-1");
+        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);
+        assert!(second.updated_at_ms > first.updated_at_ms);
+
+        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();
+        let store = TerminalScopeStore::new(&root.0);
+        store.set("same", "session-a").expect("set first scope");
+        store
+            .set("../same", "session-b")
+            .expect("set path-like key");
+        store
+            .set("../sämé", "session-c")
+            .expect("set Unicode path-like key");
+
+        assert_eq!(
+            store
+                .get("same")
+                .expect("get first")
+                .expect("exists")
+                .session_id,
+            "session-a"
+        );
+        assert_eq!(
+            store
+                .get("../same")
+                .expect("get path-like key")
+                .expect("exists")
+                .session_id,
+            "session-b"
+        );
+        assert_eq!(
+            store
+                .get("../sämé")
+                .expect("get Unicode path-like key")
+                .expect("exists")
+                .session_id,
+            "session-c"
+        );
+        assert_eq!(fs::read_dir(&root.0).expect("read root").count(), 3);
+    }
+
+    #[test]
+    fn malformed_terminal_scope_json_is_rejected() {
+        let root = TestDirectory::new();
+        let store = TerminalScopeStore::new(&root.0);
+        let path = root
+            .0
+            .join(format!("{}.json", encode_key_as_hex("request")));
+        fs::write(path, "{").expect("write malformed scope");
+        assert!(store.get("request").is_err());
+    }
+
+    #[test]
+    fn long_hex_keys_use_safe_path_components() {
+        let root = TestDirectory::new();
+        let store = TerminalScopeStore::new(&root.0);
+        let key = "k".repeat(512);
+        store
+            .set(&key, "session-long")
+            .expect("long key should persist");
+        assert_eq!(
+            store.get(&key).expect("get long key").unwrap().session_id,
+            "session-long"
+        );
+    }
+
+    #[test]
+    fn oversized_correlation_keys_are_rejected() {
+        let root = TestDirectory::new();
+        let store = TerminalScopeStore::new(&root.0);
+        let key = "k".repeat(513);
+        assert!(store.set(&key, "session-long").is_err());
+        assert!(store.get(&key).is_err());
+        assert!(!root.0.exists() || fs::read_dir(&root.0).unwrap().next().is_none());
+    }
+}