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>

Author: mrunalp

architecture/gateway.md

@@ -95,7 +95,22 @@ 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
+resolves the public sandbox name and workspace, durably binds both original and
+effective sandbox UUIDs, and rejects same-name replacements. The owner checks
+the effective UUID again before relay opening, then 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

@@ -1900,6 +1900,7 @@ pub async fn sandbox_exec_grpc(
     // Make the streaming gRPC call.
     let mut stream = client
         .exec_sandbox(ExecSandboxRequest {
+            request_id: String::new(),
             sandbox: name.to_string(),
             workspace_scope: Some(openshell_core::proto::workspace_selector(
                 workspace.to_string(),
@@ -2303,6 +2304,7 @@ async fn sandbox_exec_interactive_grpc(
     input_tx
         .send(ExecSandboxInput {
             payload: Some(exec_sandbox_input::Payload::Start(ExecSandboxRequest {
+                request_id: String::new(),
                 sandbox: sandbox.object_name().to_string(),
                 workspace_scope: Some(openshell_core::proto::workspace_selector(
                     (sandbox.object_workspace()).to_string(),

crates/openshell-sdk/src/client.rs

@@ -607,6 +607,7 @@ impl OpenShellClient {
     /// call `exec_sandbox` directly.
     pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result<ExecResult> {
         let request = proto::ExecSandboxRequest {
+            request_id: String::new(),
             sandbox: name.to_string(),
             workspace_scope: Some(proto::workspace_selector("default")),
             command: cmd.to_vec(),
@@ -1067,6 +1068,7 @@ impl WorkspaceScopedClient {
     /// Run a command inside a sandbox and buffer stdout/stderr.
     pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result<ExecResult> {
         let request = proto::ExecSandboxRequest {
+            request_id: String::new(),
             sandbox: name.to_string(),
             workspace_scope: Some(proto::workspace_selector(&self.workspace)),
             command: cmd.to_vec(),

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

@@ -430,7 +430,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

@@ -56,6 +56,8 @@ struct Admission {
     #[serde(default)]
     protection: Option<Protection>,
     workspace_id: Option<String>,
+    #[serde(default)]
+    target_id: Option<String>,
     success: Option<Success>,
     completed_at_ms: Option<i64>,
 }
@@ -65,6 +67,8 @@ struct Protection {
     key_id: String,
     effective_payload_hash: String,
     effective_workspace_id: Option<String>,
+    #[serde(default)]
+    effective_target_id: Option<String>,
 }
 
 /// Gateway-owned, in-memory input before hydration and interceptor modification.
@@ -78,19 +82,27 @@ pub(super) enum Success {
     Resource { id: String, version: u64 },
     Deletion { outcome: i32 },
     Ordinary(ordinary::Outcome),
+    StreamTerminal,
 }
 
 pub(super) struct Scope {
     name: String,
     workspace_id: Option<String>,
+    // Name-addressed exec requests must remain bound to one sandbox instance.
+    target_id: Option<String>,
 }
 
 #[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>,
@@ -141,6 +153,7 @@ pub(super) async fn run<M: Mutation>(
                 key_id: key.id(),
                 effective_payload_hash: key.fingerprint(request.get_ref())?,
                 effective_workspace_id: effective_scope.workspace_id,
+                effective_target_id: effective_scope.target_id,
             }),
         )
     } else {
@@ -207,11 +220,12 @@ 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,
         workspace_id: scope.workspace_id,
+        target_id: scope.target_id,
         success: None,
         completed_at_ms: None,
     };
@@ -256,6 +270,7 @@ async fn execute_owned<M: Mutation>(
                 ));
             }
             if previous.workspace_id != admission.workspace_id
+                || previous.target_id != admission.target_id
                 || previous.protection != admission.protection
             {
                 return Err(replay_unavailable());
@@ -302,32 +317,85 @@ 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 {
+    /// Reject replacement between admission authorization and relay creation.
+    pub(super) fn ensure_target(&self, target_id: &str) -> Result<(), Status> {
+        let admission: Admission =
+            serde_json::from_slice(&self.admission).map_err(|_| uncertain())?;
+        if admission
+            .protection
+            .as_ref()
+            .and_then(|protection| protection.effective_target_id.as_deref())
+            != Some(target_id)
+        {
+            return Err(uncertain());
+        }
+        Ok(())
+    }
+
+    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 +445,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"))?;
@@ -492,6 +564,7 @@ async fn named_scope(state: &ServerState, name: &str) -> Result<Scope, Status> {
     Ok(Scope {
         name: name.into(),
         workspace_id: Some(resource.object_id().into()),
+        target_id: None,
     })
 }
 
@@ -500,6 +573,7 @@ fn global_scope(state: &ServerState, principal: &Principal) -> Result<Scope, Sta
     Ok(Scope {
         name: String::new(),
         workspace_id: None,
+        target_id: None,
     })
 }
 
@@ -719,3 +793,4 @@ deletion_mutation!(
 mod tests;
 
 pub(super) mod ordinary;
+pub(super) mod streaming;