feat(api): add durable exec launch admission

architecture/gateway.md

@@ -91,7 +91,20 @@ in-memory extension. Replay reauthorizes original and current effective scopes,
 requires the same effective payload, and reruns current interceptor validation.
 Interceptors cannot mutate the request UUID. Server-marked replay suppresses
 post-commit observation, which remains best-effort rather than an outbox.
-Credential capabilities and streaming execution require separate contracts.
+Credential capabilities require separate contracts.
+
+Both exec RPCs share keyed admission but defer completion to the SSH producer.
+The initial interactive Start uses the exec request schema under a distinct RPC
+namespace; later stdin and resize frames are not replayable inputs. Admission
+precedes relay opening and hands its CAS-only finalizer to the owned producer,
+releasing the shared admission-worker permit after handoff. Only a confirmed
+remote exit records a terminal marker and starts 24-hour retention. Synthetic
+timeouts, disconnects without exit confirmation, and persistence failures leave
+permanent unresolved claims. Duplicates never launch, attach, or replay output:
+pending records report uncertainty, and terminal records report stream
+unavailability. Existing transport cancellation behavior remains unchanged.
+Exec timeouts retain the public duration's precision and presence: an absent
+timeout is unbounded, while an explicit zero is a finite timeout.
 
 The gateway listens on one service port and multiplexes gRPC and HTTP traffic.
 The default local single-user deployment mode is mTLS user authentication:

crates/openshell-cli/src/run.rs

@@ -1805,6 +1805,7 @@ pub async fn sandbox_exec_grpc(
     // Make the streaming gRPC call.
     let mut stream = client
         .exec_sandbox(ExecSandboxRequest {
+            request_id: String::new(),
             sandbox_id: sandbox.object_id().to_string(),
             command: command.to_vec(),
             workdir: workdir.unwrap_or_default().to_string(),
@@ -2189,6 +2190,7 @@ async fn sandbox_exec_interactive_grpc(
     input_tx
         .send(ExecSandboxInput {
             payload: Some(exec_sandbox_input::Payload::Start(ExecSandboxRequest {
+                request_id: String::new(),
                 sandbox_id: sandbox.object_id().to_string(),
                 command: command.to_vec(),
                 workdir: workdir.unwrap_or_default().to_string(),

crates/openshell-sdk/src/client.rs

@@ -608,6 +608,7 @@ impl OpenShellClient {
     pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result<ExecResult> {
         let sandbox = self.get_sandbox(name).await?;
         let request = proto::ExecSandboxRequest {
+            request_id: String::new(),
             sandbox_id: sandbox.id,
             command: cmd.to_vec(),
             workdir: opts.workdir.unwrap_or_default(),
@@ -1068,6 +1069,7 @@ impl WorkspaceScopedClient {
     pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result<ExecResult> {
         let sandbox = self.get_sandbox(name).await?;
         let request = proto::ExecSandboxRequest {
+            request_id: String::new(),
             sandbox_id: sandbox.id,
             command: cmd.to_vec(),
             workdir: opts.workdir.unwrap_or_default(),

crates/openshell-server/src/grpc/mod.rs

@@ -378,7 +378,7 @@ impl OpenShell for OpenShellService {
         &self,
         request: Request<ExecSandboxRequest>,
     ) -> Result<Response<Self::ExecSandboxStream>, Status> {
-        sandbox::handle_exec_sandbox(&self.state, request).await
+        mutation_replay::run(&self.state, request).await
     }
 
     type ForwardTcpStream =

crates/openshell-server/src/grpc/mutation_replay.rs

@@ -78,6 +78,7 @@ pub(super) enum Success {
     Resource { id: String, version: u64 },
     Deletion { outcome: i32 },
     Ordinary(ordinary::Outcome),
+    StreamTerminal,
 }
 
 pub(super) struct Scope {
@@ -87,10 +88,15 @@ pub(super) struct Scope {
 
 #[tonic::async_trait]
 pub(super) trait Mutation: Message + Default + Send + Sync + 'static {
-    type Output: Message + Default + Send + 'static;
+    type Output: Send + 'static;
     const METHOD: &'static str;
     const PROTECTED: bool = false;
+    // Streaming producers, not returned stream handles, confirm completion.
+    const DEFERRED: bool = false;
     fn request_id(&self) -> &str;
+    fn canonical_message(&self) -> Result<DynamicMessage, Status> {
+        decode_request_message(&format!("{}Request", Self::METHOD), self)
+    }
     async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result<Scope, Status>;
     async fn execute(
         state: &Arc<ServerState>,
@@ -207,7 +213,7 @@ async fn execute_owned<M: Mutation>(
     protection: Option<Protection>,
     scope: Scope,
 ) -> Result<Response<M::Output>, Status> {
-    let mut admission = Admission {
+    let admission = Admission {
         format_version: 1,
         payload_hash,
         protection,
@@ -302,32 +308,70 @@ async fn execute_owned<M: Mutation>(
         };
         // Any error or panic from here leaves the claim unresolved. Status codes
         // do not establish that a handler performed no effects.
+        let completion = Completion {
+            store: state.store.clone(),
+            claim_id,
+            key: key.into(),
+            bucket: bucket.into(),
+            version,
+            admission: payload,
+        };
+        if M::DEFERRED {
+            request.extensions_mut().insert(completion.clone());
+        }
         let facts = ordinary::Facts::default();
         request.extensions_mut().insert(facts.clone());
         let mut response = M::execute(state, request).await?;
+        if M::DEFERRED {
+            return Ok(response);
+        }
         response.extensions_mut().insert(facts);
-        admission.success = Some(M::capture(&response)?);
+        completion.complete(M::capture(&response)?).await?;
+        return Ok(response);
+    }
+    Err(uncertain())
+}
+
+/// Only the admitted owner can finalize this exact incarnation of a claim.
+/// A dropped finalizer leaves a permanent unresolved admission, never a lease.
+#[derive(Clone)]
+pub(super) struct Completion {
+    store: Arc<Store>,
+    claim_id: String,
+    key: String,
+    bucket: String,
+    version: u64,
+    admission: Vec<u8>,
+}
+
+impl Completion {
+    pub(super) async fn stream_terminal(&self) -> Result<(), Status> {
+        self.complete(Success::StreamTerminal).await
+    }
+
+    async fn complete(&self, success: Success) -> Result<(), Status> {
+        let mut admission: Admission =
+            serde_json::from_slice(&self.admission).map_err(|_| uncertain())?;
+        admission.success = Some(success);
         admission.completed_at_ms = Some(current_time_ms());
         let payload = serde_json::to_vec(&admission).map_err(|_| uncertain())?;
         if payload.len() > 64 * 1024 {
             return Err(uncertain());
         }
-        state
-            .store
+        self.store
             .put_if(
                 OBJECT_TYPE,
-                &claim_id,
-                key,
-                bucket,
+                &self.claim_id,
+                &self.key,
+                &self.bucket,
                 &payload,
                 None,
-                WriteCondition::MatchResourceVersion(version),
+                WriteCondition::MatchResourceVersion(self.version),
             )
             .await
             .map_err(|_| uncertain())?;
-        return Ok(response);
+        Ok(())
     }
-    Err(uncertain())
 }
 
 async fn prune_expired(store: &Store, bucket: &str) -> Result<bool, Status> {
@@ -377,12 +421,16 @@ fn validate_request_id(value: &str) -> Result<String, Status> {
     Ok(id.hyphenated().to_string())
 }
 
-fn fingerprint<M: Mutation>(request: &M) -> Result<String, Status> {
+fn decode_request_message(name: &str, request: &impl Message) -> Result<DynamicMessage, Status> {
     let descriptor = DESCRIPTORS
-        .get_message_by_name(&format!("openshell.v1.{}Request", M::METHOD))
+        .get_message_by_name(&format!("openshell.v1.{name}"))
         .ok_or_else(|| Status::internal("mutation request descriptor missing"))?;
-    let mut message = DynamicMessage::decode(descriptor, request.encode_to_vec().as_slice())
-        .map_err(|_| Status::internal("decode mutation request"))?;
+    DynamicMessage::decode(descriptor, request.encode_to_vec().as_slice())
+        .map_err(|_| Status::internal("decode mutation request"))
+}
+
+fn fingerprint<M: Mutation>(request: &M) -> Result<String, Status> {
+    let mut message = request.canonical_message()?;
     message.clear_field_by_name("request_id");
     let value = serde_json::to_value(message)
         .map_err(|_| Status::internal("canonicalize mutation request"))?;
@@ -706,3 +754,4 @@ deletion_mutation!(
 mod tests;
 
 pub(super) mod ordinary;
+pub(super) mod streaming;

crates/openshell-server/src/grpc/mutation_replay/streaming.rs

@@ -0,0 +1,130 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Launch admission only: output and interactive input are never replayed.
+
+#[cfg(test)]
+mod tests;
+
+use std::sync::Arc;
+
+use openshell_core::ObjectWorkspace;
+use openshell_core::proto::{
+    ExecSandboxEvent, ExecSandboxInput, ExecSandboxRequest, exec_sandbox_input,
+};
+use openshell_core::rpc_error;
+use prost_reflect::DynamicMessage;
+use tokio::sync::Mutex;
+use tokio_stream::wrappers::ReceiverStream;
+use tonic::{Request, Response, Status};
+
+use super::{Mutation, Scope, Success, decode_request_message, named_scope, uncertain};
+use crate::ServerState;
+use crate::auth::principal::Principal;
+use crate::grpc::sandbox;
+use crate::persistence::Store;
+
+pub(in crate::grpc) type ExecStream = ReceiverStream<Result<ExecSandboxEvent, Status>>;
+
+/// Private, one-owner transport handoff. The fingerprint includes only Start.
+#[derive(Clone)]
+pub(in crate::grpc) struct InteractiveInput(
+    pub Arc<Mutex<Option<tonic::Streaming<ExecSandboxInput>>>>,
+);
+
+pub(in crate::grpc) fn start(input: &ExecSandboxInput) -> Result<&ExecSandboxRequest, Status> {
+    match input.payload.as_ref() {
+        Some(exec_sandbox_input::Payload::Start(request)) => Ok(request),
+        _ => Err(rpc_error::invalid_argument(
+            "start",
+            "first message must be a start payload",
+        )),
+    }
+}
+
+async fn authorize(
+    state: &ServerState,
+    principal: &Principal,
+    request: &ExecSandboxRequest,
+) -> Result<Scope, Status> {
+    sandbox::validate_exec_start(request)?;
+    let sandbox =
+        sandbox::fetch_and_authorize_sandbox(state, principal, &request.sandbox_id).await?;
+    named_scope(state, sandbox.object_workspace()).await
+}
+
+fn stream_unavailable(success: Success) -> Status {
+    if !matches!(success, Success::StreamTerminal) {
+        return super::replay_unavailable();
+    }
+    rpc_error::failed_precondition(
+        "REQUEST_STREAM_UNAVAILABLE",
+        "execution terminated, but its output stream is not stored; this request was not launched again",
+    )
+}
+
+#[tonic::async_trait]
+impl Mutation for ExecSandboxRequest {
+    type Output = ExecStream;
+    const METHOD: &'static str = "ExecSandbox";
+    const PROTECTED: bool = true;
+    const DEFERRED: bool = true;
+
+    fn request_id(&self) -> &str {
+        &self.request_id
+    }
+
+    async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result<Scope, Status> {
+        authorize(state, principal, self).await
+    }
+
+    async fn execute(
+        state: &Arc<ServerState>,
+        request: Request<Self>,
+    ) -> Result<Response<Self::Output>, Status> {
+        sandbox::handle_exec_sandbox(state, request).await
+    }
+
+    fn capture(_: &Response<Self::Output>) -> Result<Success, Status> {
+        Err(uncertain())
+    }
+
+    async fn restore(_: &Store, success: Success) -> Result<Self::Output, Status> {
+        Err(stream_unavailable(success))
+    }
+}
+
+#[tonic::async_trait]
+impl Mutation for ExecSandboxInput {
+    type Output = ExecStream;
+    const METHOD: &'static str = "ExecSandboxInteractive";
+    const PROTECTED: bool = true;
+    const DEFERRED: bool = true;
+
+    fn request_id(&self) -> &str {
+        start(self).map_or("", |request| request.request_id.as_str())
+    }
+
+    fn canonical_message(&self) -> Result<DynamicMessage, Status> {
+        decode_request_message("ExecSandboxRequest", start(self)?)
+    }
+
+    async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result<Scope, Status> {
+        authorize(state, principal, start(self)?).await
+    }
+
+    async fn execute(
+        state: &Arc<ServerState>,
+        request: Request<Self>,
+    ) -> Result<Response<Self::Output>, Status> {
+        sandbox::handle_exec_sandbox_interactive_start(state, request).await
+    }
+
+    fn capture(_: &Response<Self::Output>) -> Result<Success, Status> {
+        Err(uncertain())
+    }
+
+    async fn restore(_: &Store, success: Success) -> Result<Self::Output, Status> {
+        Err(stream_unavailable(success))
+    }
+}