feat(api): add durable exec launch admission

Fence duplicate exec launches with keyed durable admission and producer-owned terminal completion. Keep uncertain launches unresolved and never replay output or interactive input.

Part of #3051 (phase 4a).

Signed-off-by: Mrunal Patel <mrunalp@gmail.com>

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))
+    }
+}

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

@@ -0,0 +1,445 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+use crate::grpc::mutation_replay::tests::{reason, state_for};
+use crate::grpc::mutation_replay::{
+    Admission, Completion, OBJECT_TYPE, SUCCESS_TTL_MS, fingerprint, run,
+};
+use crate::grpc::test_support::authed_request;
+use crate::persistence::{ObjectType, WriteCondition, current_time_ms};
+use openshell_core::ObjectId;
+use openshell_core::proto::datamodel::v1::ObjectMeta;
+use openshell_core::proto::{
+    GatewayMessage, Sandbox, SandboxPhase, SandboxStatus, gateway_message,
+};
+use tokio::sync::{mpsc, oneshot};
+use tokio_stream::StreamExt;
+
+async fn setup(url: &str, directory: &tempfile::TempDir) -> Arc<ServerState> {
+    let mut state = state_for(Store::connect(url).await.unwrap()).await;
+    let key = directory.path().join("key");
+    std::fs::write(&key, b"exec-test-only-key").unwrap();
+    let state_mut = Arc::get_mut(&mut state).unwrap();
+    state_mut.admin_role = "openshell-admin".into();
+    state_mut.config.gateway_jwt = Some(openshell_core::config::GatewayJwtConfig {
+        signing_key_path: key,
+        public_key_path: directory.path().join("public"),
+        kid_path: directory.path().join("kid"),
+        gateway_id: "test".into(),
+        ttl_secs: None,
+    });
+    state
+}
+
+async fn request(state: &ServerState) -> ExecSandboxRequest {
+    let sandbox = Sandbox {
+        metadata: Some(ObjectMeta {
+            id: uuid::Uuid::new_v4().to_string(),
+            name: "exec-admission".into(),
+            workspace: "default".into(),
+            ..Default::default()
+        }),
+        status: Some(SandboxStatus {
+            phase: SandboxPhase::Ready.into(),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    state.store.put_message(&sandbox).await.unwrap();
+    ExecSandboxRequest {
+        sandbox_id: sandbox.object_id().into(),
+        request_id: uuid::Uuid::new_v4().to_string(),
+        command: vec!["echo".into(), "secret-command-argument".into()],
+        stdin: b"secret-stdin".to_vec(),
+        environment: [("PRIVATE".into(), "secret-environment".into())].into(),
+        ..Default::default()
+    }
+}
+
+fn register(state: &ServerState, id: &str) -> mpsc::Receiver<GatewayMessage> {
+    let (tx, rx) = mpsc::channel(64);
+    let (shutdown, _) = oneshot::channel();
+    state
+        .supervisor_sessions
+        .register(id.into(), "session".into(), tx, shutdown);
+    rx
+}
+
+async fn receive_relay(state: &ServerState, messages: &mut mpsc::Receiver<GatewayMessage>) {
+    let message = tokio::time::timeout(std::time::Duration::from_secs(2), messages.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    let Some(gateway_message::Payload::RelayOpen(open)) = message.payload else {
+        panic!("expected relay open")
+    };
+    // No SSH transport in these admission tests. Release the detached producer.
+    state
+        .supervisor_sessions
+        .fail_pending_relay(&open.channel_id, "test relay ended".into());
+}
+
+async fn completion(state: &ServerState) -> Completion {
+    let rows = state
+        .store
+        .list_by_type_after(OBJECT_TYPE, None, 100)
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    let row = rows.into_iter().next().unwrap();
+    Completion {
+        store: state.store.clone(),
+        claim_id: row.id,
+        key: row.name,
+        bucket: row.workspace,
+        version: row.resource_version,
+        admission: row.payload,
+    }
+}
+
+#[test]
+fn interactive_fingerprint_uses_start_schema_and_ignores_request_id() {
+    let mut req = ExecSandboxRequest {
+        request_id: uuid::Uuid::new_v4().to_string(),
+        command: vec!["echo".into()],
+        ..Default::default()
+    };
+    let first = fingerprint(&req).unwrap();
+    req.request_id = uuid::Uuid::new_v4().to_string();
+    let input = ExecSandboxInput {
+        payload: Some(exec_sandbox_input::Payload::Start(req.clone())),
+    };
+    assert_eq!(first, fingerprint(&input).unwrap());
+    assert_ne!(ExecSandboxRequest::METHOD, ExecSandboxInput::METHOD);
+    req.stdin.push(42);
+    assert_ne!(first, fingerprint(&req).unwrap());
+    assert_eq!(
+        ExecSandboxInput::default()
+            .canonical_message()
+            .unwrap_err()
+            .code(),
+        tonic::Code::InvalidArgument
+    );
+}
+
+#[tokio::test]
+async fn concurrent_exec_claims_open_one_relay_and_store_no_secrets() {
+    let directory = tempfile::tempdir().unwrap();
+    let state = setup("sqlite::memory:", &directory).await;
+    let req = request(&state).await;
+    let mut messages = register(&state, &req.sandbox_id);
+    let mut tasks = Vec::new();
+    for _ in 0..16 {
+        let state = state.clone();
+        let req = req.clone();
+        tasks.push(tokio::spawn(async move {
+            run(&state, authed_request(req)).await
+        }));
+    }
+    let mut admitted = 0;
+    for task in tasks {
+        match task.await.unwrap() {
+            Ok(_) => admitted += 1,
+            Err(status) => assert_eq!(reason(&status), "REQUEST_OUTCOME_UNCERTAIN"),
+        }
+    }
+    assert_eq!(admitted, 1);
+    receive_relay(&state, &mut messages).await;
+    assert!(messages.try_recv().is_err());
+    let pending = completion(&state).await;
+    let text = String::from_utf8(pending.admission).unwrap();
+    for secret in [
+        "secret-command-argument",
+        "secret-stdin",
+        "secret-environment",
+        &fingerprint(&req).unwrap(),
+    ] {
+        assert!(!text.contains(secret));
+    }
+    let mut changed = req.clone();
+    changed.command.push("different".into());
+    assert_eq!(
+        reason(&run(&state, authed_request(changed)).await.unwrap_err()),
+        "REQUEST_ID_PAYLOAD_MISMATCH"
+    );
+}
+
+#[tokio::test]
+async fn confirmed_exit_including_nonzero_and_124_finalizes_before_output_delivery() {
+    for code in [0, 7, 124] {
+        let directory = tempfile::tempdir().unwrap();
+        let state = setup("sqlite::memory:", &directory).await;
+        let req = request(&state).await;
+        let mut messages = register(&state, &req.sandbox_id);
+        drop(run(&state, authed_request(req.clone())).await.unwrap());
+        receive_relay(&state, &mut messages).await;
+        let owner = completion(&state).await;
+        assert_eq!(
+            sandbox::wait_for_exec_terminal(async { Ok(code) }, None, Some(owner.clone()))
+                .await
+                .unwrap(),
+            Some(code)
+        );
+        assert_eq!(
+            reason(&run(&state, authed_request(req)).await.unwrap_err()),
+            "REQUEST_STREAM_UNAVAILABLE"
+        );
+        assert!(
+            owner.stream_terminal().await.is_err(),
+            "CAS rejects a second finalization"
+        );
+        let row = completion(&state).await;
+        let admission: Admission = serde_json::from_slice(&row.admission).unwrap();
+        assert!(matches!(admission.success, Some(Success::StreamTerminal)));
+        assert!(admission.completed_at_ms.is_some());
+    }
+}
+
+#[tokio::test]
+async fn timeout_and_lost_exit_status_never_finalize_or_expire_pending_launches() {
+    let directory = tempfile::tempdir().unwrap();
+    let state = setup("sqlite::memory:", &directory).await;
+    let req = request(&state).await;
+    let mut messages = register(&state, &req.sandbox_id);
+    drop(run(&state, authed_request(req.clone())).await.unwrap());
+    receive_relay(&state, &mut messages).await;
+    let owner = completion(&state).await;
+    for execution_timeout in [
+        std::time::Duration::ZERO,
+        std::time::Duration::from_nanos(1),
+        std::time::Duration::from_secs(1),
+    ] {
+        assert_eq!(
+            tokio::time::timeout(
+                std::time::Duration::from_secs(5),
+                sandbox::wait_for_exec_terminal(
+                    std::future::pending(),
+                    Some(execution_timeout),
+                    Some(owner.clone()),
+                ),
+            )
+            .await
+            .expect("a present timeout must not wait indefinitely")
+            .unwrap(),
+            None
+        );
+    }
+    assert!(
+        sandbox::wait_for_exec_terminal(
+            async { Err(Status::unavailable("lost exit")) },
+            None,
+            Some(owner.clone())
+        )
+        .await
+        .is_err()
+    );
+    let mut admission: Admission = serde_json::from_slice(&owner.admission).unwrap();
+    assert!(admission.success.is_none());
+    admission.completed_at_ms = Some(current_time_ms() - SUCCESS_TTL_MS - 1);
+    state
+        .store
+        .put_if(
+            OBJECT_TYPE,
+            &owner.claim_id,
+            &owner.key,
+            &owner.bucket,
+            &serde_json::to_vec(&admission).unwrap(),
+            None,
+            WriteCondition::MatchResourceVersion(owner.version),
+        )
+        .await
+        .unwrap();
+    // A real exit cannot be exposed as confirmed if the finalizer loses its CAS.
+    let status = sandbox::wait_for_exec_terminal(async { Ok(0) }, None, Some(owner))
+        .await
+        .unwrap_err();
+    assert_eq!(reason(&status), "REQUEST_OUTCOME_UNCERTAIN");
+    assert_eq!(
+        reason(&run(&state, authed_request(req)).await.unwrap_err()),
+        "REQUEST_OUTCOME_UNCERTAIN"
+    );
+    assert!(messages.try_recv().is_err());
+}
+
+#[tokio::test]
+async fn cancellation_before_admission_does_nothing_and_after_admission_keeps_owner() {
+    let directory = tempfile::tempdir().unwrap();
+    let state = setup("sqlite::memory:", &directory).await;
+    let req = request(&state).await;
+    drop(run(&state, authed_request(req.clone())));
+    assert!(
+        state
+            .store
+            .list_by_type_after(OBJECT_TYPE, None, 100)
+            .await
+            .unwrap()
+            .is_empty()
+    );
+    let task_state = state.clone();
+    let task_req = req.clone();
+    let task = tokio::spawn(async move { run(&task_state, authed_request(task_req)).await });
+    tokio::time::timeout(std::time::Duration::from_secs(2), async {
+        while state
+            .store
+            .list_by_type_after(OBJECT_TYPE, None, 100)
+            .await
+            .unwrap()
+            .is_empty()
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    task.abort();
+    let mut messages = register(&state, &req.sandbox_id);
+    receive_relay(&state, &mut messages).await;
+    assert_eq!(
+        reason(&run(&state, authed_request(req)).await.unwrap_err()),
+        "REQUEST_OUTCOME_UNCERTAIN"
+    );
+    assert!(messages.try_recv().is_err());
+}
+
+#[tokio::test]
+async fn exec_restart_preserves_pending_and_terminal_fences_and_reauthorizes() {
+    let directory = tempfile::tempdir().unwrap();
+    let url = format!(
+        "sqlite://{}?mode=rwc",
+        directory.path().join("store.db").display()
+    );
+    let state = setup(&url, &directory).await;
+    let req = request(&state).await;
+    let mut messages = register(&state, &req.sandbox_id);
+    let mut output = run(&state, authed_request(req.clone()))
+        .await
+        .unwrap()
+        .into_inner();
+    receive_relay(&state, &mut messages).await;
+    assert!(output.next().await.unwrap().is_err());
+    assert!(output.next().await.is_none());
+    drop(output);
+    drop(messages);
+    drop(state);
+    let state = setup(&url, &directory).await;
+    assert_eq!(
+        reason(&run(&state, authed_request(req.clone())).await.unwrap_err()),
+        "REQUEST_OUTCOME_UNCERTAIN"
+    );
+    completion(&state).await.stream_terminal().await.unwrap();
+    drop(state);
+    let state = setup(&url, &directory).await;
+    let mut sandbox: Sandbox = state
+        .store
+        .get_message(&req.sandbox_id)
+        .await
+        .unwrap()
+        .unwrap();
+    sandbox.status.as_mut().unwrap().phase = SandboxPhase::Stopped.into();
+    state.store.put_message(&sandbox).await.unwrap();
+    assert_eq!(
+        reason(&run(&state, authed_request(req.clone())).await.unwrap_err()),
+        "REQUEST_STREAM_UNAVAILABLE"
+    );
+    let mut unauthorized = authed_request(req.clone());
+    if let Principal::User(user) = unauthorized
+        .extensions_mut()
+        .get_mut::<Principal>()
+        .unwrap()
+    {
+        user.identity.roles.clear();
+    }
+    assert_eq!(
+        run(&state, unauthorized).await.unwrap_err().code(),
+        tonic::Code::NotFound
+    );
+    state
+        .store
+        .delete(Sandbox::object_type(), &req.sandbox_id)
+        .await
+        .unwrap();
+    assert_eq!(
+        run(&state, authed_request(req)).await.unwrap_err().code(),
+        tonic::Code::NotFound
+    );
+}
+
+#[tokio::test]
+async fn interactive_wire_start_is_admitted_under_its_own_method_namespace() {
+    use crate::grpc::OpenShellService;
+    use openshell_core::proto::{
+        open_shell_client::OpenShellClient, open_shell_server::OpenShellServer,
+    };
+    use tokio_stream::wrappers::TcpListenerStream;
+    let directory = tempfile::tempdir().unwrap();
+    let state = setup("sqlite::memory:", &directory).await;
+    let req = request(&state).await;
+    let mut messages = register(&state, &req.sandbox_id);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let principal = authed_request(())
+        .extensions()
+        .get::<Principal>()
+        .unwrap()
+        .clone();
+    let service = OpenShellServer::with_interceptor(
+        OpenShellService::new(state.clone()),
+        move |mut request: Request<()>| {
+            request.extensions_mut().insert(principal.clone());
+            Ok(request)
+        },
+    );
+    let server = tokio::spawn(async move {
+        tonic::transport::Server::builder()
+            .add_service(service)
+            .serve_with_incoming(TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+    let mut client = OpenShellClient::connect(format!("http://{address}"))
+        .await
+        .unwrap();
+    let start = ExecSandboxInput {
+        payload: Some(exec_sandbox_input::Payload::Start(req.clone())),
+    };
+    let output = client
+        .exec_sandbox_interactive(tokio_stream::iter([start.clone()]))
+        .await
+        .unwrap();
+    receive_relay(&state, &mut messages).await;
+    let owner = completion(&state).await;
+    assert_eq!(
+        reason(
+            &client
+                .exec_sandbox_interactive(tokio_stream::iter([start.clone()]))
+                .await
+                .unwrap_err()
+        ),
+        "REQUEST_OUTCOME_UNCERTAIN"
+    );
+    owner.stream_terminal().await.unwrap();
+    assert_eq!(
+        reason(
+            &client
+                .exec_sandbox_interactive(tokio_stream::iter([start]))
+                .await
+                .unwrap_err()
+        ),
+        "REQUEST_STREAM_UNAVAILABLE"
+    );
+    // The same UUID belongs to a different RPC namespace, not another stdin stream.
+    drop(client.exec_sandbox(req).await.unwrap());
+    receive_relay(&state, &mut messages).await;
+    assert_eq!(
+        state
+            .store
+            .list_by_type_after(OBJECT_TYPE, None, 100)
+            .await
+            .unwrap()
+            .len(),
+        2
+    );
+    drop(output);
+    server.abort();
+}

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

@@ -128,7 +128,7 @@ impl Drop for WatchSandboxStream {
 /// `NOT_FOUND` for both missing and unauthorized sandboxes so that callers
 /// cannot distinguish the two cases (CWE-203).
 pub(super) async fn fetch_and_authorize_sandbox(
-    state: &Arc<ServerState>,
+    state: &ServerState,
     principal: &crate::auth::principal::Principal,
     sandbox_id: &str,
 ) -> Result<Sandbox, Status> {
@@ -1874,19 +1874,12 @@ pub(super) async fn handle_exec_sandbox(
     use openshell_core::ObjectId;
 
     let principal = super::extract_principal(&request)?;
+    let completion = request
+        .extensions()
+        .get::<super::mutation_replay::Completion>()
+        .cloned();
     let req = request.into_inner();
-    if req.sandbox_id.is_empty() {
-        return Err(Status::invalid_argument("sandbox_id is required"));
-    }
-    if req.command.is_empty() {
-        return Err(Status::invalid_argument("command is required"));
-    }
-    if req.environment.keys().any(|key| !is_valid_env_key(key)) {
-        return Err(Status::invalid_argument(
-            "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$",
-        ));
-    }
-    validate_exec_request_fields(&req)?;
+    validate_exec_start(&req)?;
 
     let sandbox = fetch_and_authorize_sandbox(state, &principal, &req.sandbox_id).await?;
 
@@ -1940,6 +1933,7 @@ pub(super) async fn handle_exec_sandbox(
             no_login_shell,
             cols,
             rows,
+            completion,
         )
         .await
         {
@@ -2290,6 +2284,29 @@ async fn bridge_forward_tcp_stream(
 // Interactive exec handler (bidirectional stdin streaming)
 // ---------------------------------------------------------------------------
 
+pub(super) fn validate_exec_start(req: &ExecSandboxRequest) -> Result<(), Status> {
+    use openshell_core::rpc_error;
+    if req.sandbox_id.is_empty() {
+        return Err(rpc_error::invalid_argument(
+            "sandbox_id",
+            "sandbox_id is required",
+        ));
+    }
+    if req.command.is_empty() {
+        return Err(rpc_error::invalid_argument(
+            "command",
+            "command is required",
+        ));
+    }
+    if req.environment.keys().any(|key| !is_valid_env_key(key)) {
+        return Err(rpc_error::invalid_argument(
+            "environment",
+            "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$",
+        ));
+    }
+    validate_exec_request_fields(req)
+}
+
 fn validate_interactive_exec_start(
     msg: Option<ExecSandboxInput>,
 ) -> Result<ExecSandboxRequest, Status> {
@@ -2304,18 +2321,7 @@ fn validate_interactive_exec_start(
         ));
     };
 
-    if req.sandbox_id.is_empty() {
-        return Err(Status::invalid_argument("sandbox_id is required"));
-    }
-    if req.command.is_empty() {
-        return Err(Status::invalid_argument("command is required"));
-    }
-    if req.environment.keys().any(|key| !is_valid_env_key(key)) {
-        return Err(Status::invalid_argument(
-            "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$",
-        ));
-    }
-    validate_exec_request_fields(&req)?;
+    validate_exec_start(&req)?;
 
     Ok(req)
 }
@@ -2324,10 +2330,7 @@ pub(super) async fn handle_exec_sandbox_interactive(
     state: &Arc<ServerState>,
     request: Request<tonic::Streaming<ExecSandboxInput>>,
 ) -> Result<Response<ReceiverStream<Result<ExecSandboxEvent, Status>>>, Status> {
-    use openshell_core::ObjectId;
-
-    let principal = super::extract_principal(&request)?;
-    let mut input_stream = request.into_inner();
+    let (metadata, extensions, mut input_stream) = request.into_parts();
 
     let first_msg = input_stream
         .message()
@@ -2336,6 +2339,46 @@ pub(super) async fn handle_exec_sandbox_interactive(
 
     let req = validate_interactive_exec_start(first_msg)?;
 
+    let mut request = Request::from_parts(
+        metadata,
+        extensions,
+        ExecSandboxInput {
+            payload: Some(openshell_core::proto::exec_sandbox_input::Payload::Start(
+                req,
+            )),
+        },
+    );
+    request
+        .extensions_mut()
+        .insert(super::mutation_replay::streaming::InteractiveInput(
+            Arc::new(tokio::sync::Mutex::new(Some(input_stream))),
+        ));
+    super::mutation_replay::run(state, request).await
+}
+
+pub(super) async fn handle_exec_sandbox_interactive_start(
+    state: &Arc<ServerState>,
+    request: Request<ExecSandboxInput>,
+) -> Result<Response<ReceiverStream<Result<ExecSandboxEvent, Status>>>, Status> {
+    let principal = super::extract_principal(&request)?;
+    let completion = request
+        .extensions()
+        .get::<super::mutation_replay::Completion>()
+        .cloned();
+    let input = request
+        .extensions()
+        .get::<super::mutation_replay::streaming::InteractiveInput>()
+        .ok_or_else(|| Status::internal("interactive input handoff missing"))?
+        .clone();
+    let input_stream = input
+        .0
+        .lock()
+        .await
+        .take()
+        .ok_or_else(|| Status::internal("interactive input already claimed"))?;
+    let req = super::mutation_replay::streaming::start(request.get_ref())?;
+    validate_exec_start(req)?;
+
     let sandbox = fetch_and_authorize_sandbox(state, &principal, &req.sandbox_id).await?;
 
     if SandboxPhase::try_from(sandbox.phase()).ok() != Some(SandboxPhase::Ready) {
@@ -2348,7 +2391,7 @@ pub(super) async fn handle_exec_sandbox_interactive(
         .await
         .map_err(|e| Status::unavailable(format!("supervisor relay failed: {e}")))?;
 
-    let command_str = build_remote_exec_command(&req)
+    let command_str = build_remote_exec_command(req)
         .map_err(|e| Status::invalid_argument(format!("command construction failed: {e}")))?;
     let request_tty = req.tty;
     let no_login_shell = req.no_login_shell;
@@ -2388,6 +2431,7 @@ pub(super) async fn handle_exec_sandbox_interactive(
             execution_timeout,
             cols,
             rows,
+            completion,
         )
         .await
         {
@@ -2695,6 +2739,7 @@ async fn stream_exec_over_relay(
     no_login_shell: bool,
     cols: u32,
     rows: u32,
+    completion: Option<super::mutation_replay::Completion>,
 ) -> Result<(), Status> {
     let command_preview: String = command
         .chars()
@@ -2724,26 +2769,22 @@ async fn stream_exec_over_relay(
         tx.clone(),
     );
 
-    let exec_result = if let Some(execution_timeout) = execution_timeout {
-        if let Ok(result) = tokio::time::timeout(execution_timeout, exec).await {
-            result
-        } else {
-            let _ = tx
-                .send(Ok(ExecSandboxEvent {
-                    payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit(
-                        ExecSandboxExit { exit_code: 124 },
-                    )),
-                }))
-                .await;
-            let _ = proxy_task.await;
-            return Ok(());
-        }
-    } else {
-        exec.await
-    };
+    let exec_result = wait_for_exec_terminal(exec, execution_timeout, completion).await;
+    if matches!(exec_result, Ok(None)) {
+        let _ = tx
+            .send(Ok(ExecSandboxEvent {
+                payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit(
+                    ExecSandboxExit { exit_code: 124 },
+                )),
+            }))
+            .await;
+        let _ = proxy_task.await;
+        return Ok(());
+    }
 
     let exit_code = match exec_result {
-        Ok(code) => code,
+        Ok(Some(code)) => code,
+        Ok(None) => unreachable!("timeout returned above"),
         Err(status) => {
             let _ = proxy_task.await;
             return Err(status);
@@ -2776,6 +2817,7 @@ async fn stream_interactive_exec_over_relay(
     execution_timeout: Option<std::time::Duration>,
     cols: u32,
     rows: u32,
+    completion: Option<super::mutation_replay::Completion>,
 ) -> Result<(), Status> {
     let command_preview: String = command
         .chars()
@@ -2805,26 +2847,22 @@ async fn stream_interactive_exec_over_relay(
         tx.clone(),
     );
 
-    let exec_result = if let Some(execution_timeout) = execution_timeout {
-        if let Ok(result) = tokio::time::timeout(execution_timeout, exec).await {
-            result
-        } else {
-            let _ = tx
-                .send(Ok(ExecSandboxEvent {
-                    payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit(
-                        ExecSandboxExit { exit_code: 124 },
-                    )),
-                }))
-                .await;
-            let _ = proxy_task.await;
-            return Ok(());
-        }
-    } else {
-        exec.await
-    };
+    let exec_result = wait_for_exec_terminal(exec, execution_timeout, completion).await;
+    if matches!(exec_result, Ok(None)) {
+        let _ = tx
+            .send(Ok(ExecSandboxEvent {
+                payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit(
+                    ExecSandboxExit { exit_code: 124 },
+                )),
+            }))
+            .await;
+        let _ = proxy_task.await;
+        return Ok(());
+    }
 
     let exit_code = match exec_result {
-        Ok(code) => code,
+        Ok(Some(code)) => code,
+        Ok(None) => unreachable!("timeout returned above"),
         Err(status) => {
             let _ = proxy_task.await;
             return Err(status);
@@ -2844,6 +2882,29 @@ async fn stream_interactive_exec_over_relay(
     Ok(())
 }
 
+/// `Ok(Some(code))` requires a real SSH `ExitStatus`, including nonzero codes.
+/// Synthetic gateway timeouts, lost status, and dropped futures leave the claim
+/// unresolved. Finalization precedes terminal delivery to the client.
+pub(super) async fn wait_for_exec_terminal(
+    exec: impl Future<Output = Result<i32, Status>>,
+    execution_timeout: Option<std::time::Duration>,
+    completion: Option<super::mutation_replay::Completion>,
+) -> Result<Option<i32>, Status> {
+    let result = if let Some(execution_timeout) = execution_timeout {
+        match tokio::time::timeout(execution_timeout, exec).await {
+            Ok(result) => result,
+            Err(_) => return Ok(None),
+        }
+    } else {
+        exec.await
+    };
+    let exit_code = result?;
+    if let Some(completion) = completion {
+        completion.stream_terminal().await?;
+    }
+    Ok(Some(exit_code))
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn run_interactive_exec_with_russh(
     local_proxy_port: u16,

crates/openshell-server/src/storage_proto.rs

@@ -118,7 +118,7 @@ mod tests {
     const STORAGE_V1_SCHEMA_SHA256: &str =
         "574bf5fcff731bd6e3fd84ed3f124161035bd236ef0fb7e32b4d8a8c55ceba5e";
     const PUBLIC_RPC_SCHEMA_SHA256: &str =
-        "1c72f65167a5ceb00cd17043079b92324ca55d19b547c86827b51bd7376cb23c";
+        "9df91739f47f743a271536847486b7be457f66bb37370e5537ad4e7e69b91fdc";
     const DURABLE_SCHEMA_SHA256: &str =
         "65066c0b0eef57a4c708f20fcbbb8e8f47376da9f4bf73dfc3bca0b3df174ba8";
     const PUBLIC_DURABLE_OVERLAP_SHA256: &str =

docs/reference/api-errors.mdx

@@ -34,6 +34,7 @@ Recognized gateway reasons include the following.
 | `REQUEST_ID_PAYLOAD_MISMATCH` | `FAILED_PRECONDITION` | Keep the original payload for that request ID. Inspect the original operation before submitting a different one. |
 | `REQUEST_OUTCOME_UNCERTAIN` | `FAILED_PRECONDITION` | An attempt is admitted but has no confirmed replayable success. Observe resource state and reconcile effects. Do not switch to a new ID to bypass the admission. |
 | `REQUEST_REPLAY_UNAVAILABLE` | `FAILED_PRECONDITION` | The original scope, resource, interceptor transformation, or fingerprint key is no longer replayable. Reconcile effects; the gateway does not execute the request again. Missing private-key material can also reject admission before work starts. |
+| `REQUEST_STREAM_UNAVAILABLE` | `FAILED_PRECONDITION` | The exec attempt terminated, but its output was not stored. Reconcile command effects; the gateway does not launch it again or recreate its stream. |
 
 ## Status and retry guidance
 
@@ -183,12 +184,48 @@ The gateway also limits detached admission workers to 64 per process and these
 request messages to 4 MiB.
 
 Supply IDs through generated/raw RPC clients. Curated SDK request-ID helpers and
-automatic retry policies are not part of this contract yet. Exec, SSH sessions,
+automatic retry policies are not part of this contract yet. SSH sessions,
 rootfs staging, and supervisor-authenticated mutations do not gain deduplication
 from this feature. Sandbox principals cannot opt into `UpdateConfig` admission.
 Check gateway compatibility
 before relying on IDs; older protobuf servers can silently ignore unknown fields.
 
+## Exec launch admission
+
+`ExecSandbox` and `ExecSandboxInteractive` accept an optional `request_id` in
+`ExecSandboxRequest`. For interactive exec, supply it in the initial `start`
+message. The UUID rules, keyed fingerprints, current authorization checks,
+per-caller quota, and request-size limit above also apply to exec.
+
+The gateway admits one owner before opening the supervisor relay. Both RPCs
+fingerprint the command, sandbox UUID, environment, initial stdin, and other
+start fields, excluding the request ID. They use separate method namespaces.
+Later interactive stdin and resize messages are not part of the fingerprint and
+cannot be submitted through a duplicate stream.
+
+| Admitted attempt | Duplicate response |
+|---|---|
+| Still running, interrupted, timed out, or missing confirmed exit status | `REQUEST_OUTCOME_UNCERTAIN`. The record does not expire or permit another launch. |
+| Confirmed remote exit, including a nonzero exit code | `REQUEST_STREAM_UNAVAILABLE` for 24 hours after durable completion. No output, exit code, stream attachment, or stdin is replayed. |
+| Changed start payload | `REQUEST_ID_PAYLOAD_MISMATCH`. The changed command does not run. |
+
+The terminal receipt contains only a marker and completion time, not commands,
+environment values, stdin, stdout, or stderr. Receiving a stream handle is not
+completion. A gateway-generated timeout exit code of `124` does not prove the
+remote process stopped, so it leaves the admission unresolved. A confirmed remote
+exit code of `124` is a terminal result.
+
+Cancellation does not undo admission. Existing transport behavior still applies
+to the original execution, including interactive SSH closure on disconnect; the
+gateway does not promise that a disconnected command continues. When it loses
+exit confirmation, the record remains unresolved even after restart. Current
+authorization and the original sandbox UUID are checked before duplicate lookup.
+
+After confirmed terminal retention expires, the same ID can start a new command.
+Do not reuse an old ID without reconciling its effects. Empty IDs keep the existing
+streaming behavior without launch deduplication. Use generated/raw RPC clients
+for this opt-in contract; curated SDK helpers and automatic retries are separate.
+
 ## Deletion outcomes
 
 Delete, membership-removal, and SSH-revocation RPCs return a typed outcome.

e2e/python/test_exec_admission.py

@@ -0,0 +1,154 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import threading
+import time
+import uuid
+from typing import TYPE_CHECKING
+
+import grpc
+import pytest
+from google.protobuf import duration_pb2
+
+from openshell._proto import openshell_pb2
+from openshell.errors import from_grpc_error
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from openshell import Sandbox, SandboxClient
+
+
+def assert_blocked(stream, reason: str) -> None:
+    with pytest.raises(grpc.RpcError) as caught:
+        list(stream)
+    error = from_grpc_error(caught.value)
+    assert error.code() == grpc.StatusCode.FAILED_PRECONDITION
+    assert error.error_info is not None
+    assert error.error_info.reason == reason
+
+
+@pytest.mark.parametrize("interactive", [False, True])
+def test_exec_request_id_never_relaunches_or_replays_output(
+    sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool
+) -> None:
+    with sandbox(delete_on_exit=True) as sb:
+        path = f"/sandbox/exec-admission-{uuid.uuid4().hex}"
+        request = openshell_pb2.ExecSandboxRequest(
+            sandbox_id=sb.id,
+            command=[
+                "/bin/sh",
+                "-c",
+                f"printf x >> {path}; printf started; "
+                f"while [ ! -e {path}.release ]; do sleep 0.1; done; "
+                "printf finished; exit 7",
+            ],
+            request_id=str(uuid.uuid4()),
+        )
+        done = threading.Event()
+
+        def invoke(value):
+            if not interactive:
+                return sandbox_client._stub.ExecSandbox(value, timeout=30)
+
+            def inputs():
+                yield openshell_pb2.ExecSandboxInput(start=value)
+                done.wait(timeout=30)
+
+            return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30)
+
+        stream = invoke(request)
+        try:
+            first = next(stream)
+            assert first.WhichOneof("payload") == "stdout"
+            assert b"started" in first.stdout.data
+            assert_blocked(invoke(request), "REQUEST_OUTCOME_UNCERTAIN")
+            changed = openshell_pb2.ExecSandboxRequest()
+            changed.CopyFrom(request)
+            changed.command.append("changed-payload")
+            assert_blocked(invoke(changed), "REQUEST_ID_PAYLOAD_MISMATCH")
+            assert sb.exec(["touch", f"{path}.release"]).exit_code == 0
+            events = list(stream)
+            assert events[-1].WhichOneof("payload") == "exit"
+            assert events[-1].exit.exit_code == 7
+            assert_blocked(invoke(request), "REQUEST_STREAM_UNAVAILABLE")
+            result = sb.exec(["cat", path])
+            assert result.exit_code == 0
+            assert result.stdout == "x"
+        finally:
+            done.set()
+            stream.cancel()
+
+
+@pytest.mark.parametrize("interactive", [False, True])
+def test_exec_request_timeout_keeps_launch_unresolved(
+    sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool
+) -> None:
+    with sandbox(delete_on_exit=True) as sb:
+        path = f"/sandbox/exec-timeout-{uuid.uuid4().hex}"
+        request = openshell_pb2.ExecSandboxRequest(
+            sandbox_id=sb.id,
+            command=["/bin/sh", "-c", f"printf x >> {path}; sleep 5"],
+            execution_timeout=duration_pb2.Duration(seconds=1),
+            request_id=str(uuid.uuid4()),
+        )
+        done = threading.Event()
+
+        def invoke():
+            if not interactive:
+                return sandbox_client._stub.ExecSandbox(request, timeout=30)
+
+            def inputs():
+                yield openshell_pb2.ExecSandboxInput(start=request)
+                done.wait(timeout=30)
+
+            return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30)
+
+        stream = invoke()
+        try:
+            events = []
+            for event in stream:
+                events.append(event)
+                if event.WhichOneof("payload") == "exit":
+                    # A synthetic timeout leaves the interactive input task
+                    # alive until the client closes its side of the stream.
+                    done.set()
+            assert events[-1].exit.exit_code == 124
+            assert_blocked(invoke(), "REQUEST_OUTCOME_UNCERTAIN")
+            assert sb.exec(["cat", path]).stdout == "x"
+        finally:
+            done.set()
+            stream.cancel()
+
+
+def test_exec_client_cancellation_does_not_clear_launch_admission(
+    sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient
+) -> None:
+    with sandbox(delete_on_exit=True) as sb:
+        path = f"/sandbox/exec-cancel-{uuid.uuid4().hex}"
+        request = openshell_pb2.ExecSandboxRequest(
+            sandbox_id=sb.id,
+            command=[
+                "/bin/sh",
+                "-c",
+                f"printf x >> {path}; printf started; sleep 1; printf finished",
+            ],
+            request_id=str(uuid.uuid4()),
+        )
+        stream = sandbox_client._stub.ExecSandbox(request, timeout=30)
+        assert b"started" in next(stream).stdout.data
+        stream.cancel()
+        deadline = time.monotonic() + 15
+        while True:
+            with pytest.raises(grpc.RpcError) as caught:
+                list(sandbox_client._stub.ExecSandbox(request, timeout=30))
+            error = from_grpc_error(caught.value)
+            assert error.error_info is not None
+            if error.error_info.reason == "REQUEST_STREAM_UNAVAILABLE":
+                break
+            assert error.error_info.reason == "REQUEST_OUTCOME_UNCERTAIN"
+            assert time.monotonic() < deadline
+            time.sleep(0.1)
+        assert sb.exec(["cat", path]).stdout == "x"

proto/openshell.proto

@@ -1613,6 +1613,12 @@ message ExecSandboxRequest {
   // sourced by them) are applied. When true, the command runs without those
   // files (`bash -c`), for automation that needs predictable startup behavior.
   bool no_login_shell = 10;
+
+  // Optional nonzero UUID for durable launch admission. Also applies to the
+  // initial ExecSandboxInteractive start message. Duplicates never relaunch,
+  // reattach, or replay output/stdin. Unconfirmed executions remain fenced;
+  // confirmed terminal executions retain the fence for 24 hours.
+  string request_id = 11;
 }
 
 // One stdout chunk from a sandbox exec.

sdk/go/proto/openshellv1/openshell.pb.go

@@ -5014,7 +5014,12 @@ type ExecSandboxRequest struct {
 	// (`bash -lc`) so user startup files (.bash_profile/.profile, and .bashrc if
 	// sourced by them) are applied. When true, the command runs without those
 	// files (`bash -c`), for automation that needs predictable startup behavior.
-	NoLoginShell  bool `protobuf:"varint,10,opt,name=no_login_shell,json=noLoginShell,proto3" json:"no_login_shell,omitempty"`
+	NoLoginShell bool `protobuf:"varint,10,opt,name=no_login_shell,json=noLoginShell,proto3" json:"no_login_shell,omitempty"`
+	// Optional nonzero UUID for durable launch admission. Also applies to the
+	// initial ExecSandboxInteractive start message. Duplicates never relaunch,
+	// reattach, or replay output/stdin. Unconfirmed executions remain fenced;
+	// confirmed terminal executions retain the fence for 24 hours.
+	RequestId     string `protobuf:"bytes,11,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -5119,6 +5124,13 @@ func (x *ExecSandboxRequest) GetNoLoginShell() bool {
 	return false
 }
 
+func (x *ExecSandboxRequest) GetRequestId() string {
+	if x != nil {
+		return x.RequestId
+	}
+	return ""
+}
+
 // One stdout chunk from a sandbox exec.
 type ExecSandboxStdout struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
@@ -15506,7 +15518,7 @@ const file_openshell_proto_rawDesc = "" +
 	"\x05token\x18\x01 \x01(\tB\x04\x88\xb5\x18\x01R\x05token\x12#\n" +
 	"\rallow_missing\x18\x02 \x01(\bR\fallowMissing\"b\n" +
 	"\x18RevokeSshSessionResponse\x127\n" +
-	"\aoutcome\x18\x02 \x01(\x0e2\x1d.openshell.v1.DeletionOutcomeR\aoutcomeJ\x04\b\x01\x10\x02R\arevoked\"\xd1\x03\n" +
+	"\aoutcome\x18\x02 \x01(\x0e2\x1d.openshell.v1.DeletionOutcomeR\aoutcomeJ\x04\b\x01\x10\x02R\arevoked\"\xf0\x03\n" +
 	"\x12ExecSandboxRequest\x12\x1d\n" +
 	"\n" +
 	"sandbox_id\x18\x01 \x01(\tR\tsandboxId\x12\x18\n" +
@@ -15519,7 +15531,9 @@ const file_openshell_proto_rawDesc = "" +
 	"\x04cols\x18\b \x01(\rR\x04cols\x12\x12\n" +
 	"\x04rows\x18\t \x01(\rR\x04rows\x12$\n" +
 	"\x0eno_login_shell\x18\n" +
-	" \x01(\bR\fnoLoginShell\x1a>\n" +
+	" \x01(\bR\fnoLoginShell\x12\x1d\n" +
+	"\n" +
+	"request_id\x18\v \x01(\tR\trequestId\x1a>\n" +
 	"\x10EnvironmentEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01J\x04\b\x05\x10\x06R\x0ftimeout_seconds\"'\n" +