fix(runtime): close capability lifecycle durably

Persist canonical ToolResult through DurableEventCommitter::commit_step
for production commit_result and interrupt, retain ExecutionLease from
prepare until commit, recover open tokens on stop/shutdown/drop, reject
same-run retry of unresolved calls, and add authorize() for future
cap::* effects.

Author: fffonion

src/capabilities/host.rs

@@ -45,6 +45,9 @@ fn error_message(error: &LifecycleError) -> String {
             "registry identity does not match frozen snapshot".to_string()
         }
         LifecycleError::InvalidMetadata(message) => message.clone(),
+        LifecycleError::UnresolvedCall => {
+            "an unresolved execution token already exists for this call".to_string()
+        }
     }
 }
 

src/capabilities/lifecycle.rs

@@ -310,6 +310,14 @@ impl CapabilityLifecycle {
         )? {
             return Ok(PrepareOutcome::Replay { result });
         }
+        {
+            let unresolved = self.inner.token_states.lock().values().any(|state| {
+                matches!(state, TokenState::Open(claims) if claims.call_id == metadata.call_id)
+            });
+            if unresolved {
+                return Err(LifecycleError::UnresolvedCall);
+            }
+        }
         if self.inner.clock.now_ms() >= self.inner.deadline_ms {
             return Err(LifecycleError::DeadlineElapsed);
         }
@@ -439,6 +447,51 @@ impl CapabilityLifecycle {
         }
     }
 
+    /// Lookup and authorize an open execution token before a future `cap::*` effect.
+    pub fn authorize(
+        &self,
+        owner: &CapabilityOwner,
+        token: &str,
+        requested: CapabilityRisk,
+    ) -> Result<TokenClaims, LifecycleError> {
+        if owner != &self.inner.owner {
+            return Err(LifecycleError::OwnerMismatch {
+                expected: self.inner.owner.key(),
+                actual: owner.key(),
+            });
+        }
+        if self.inner.cancellation.is_cancelled() {
+            return Err(LifecycleError::Cancelled);
+        }
+        let states = self.inner.token_states.lock();
+        let claims = match states.get(token) {
+            Some(TokenState::Open(claims)) => claims.as_ref().clone(),
+            Some(TokenState::Committed) => return Err(LifecycleError::DuplicateClose),
+            Some(TokenState::Interrupted) => return Err(LifecycleError::Interrupted),
+            None => return Err(LifecycleError::TokenUnknown),
+        };
+        drop(states);
+        if &claims.owner != owner {
+            return Err(LifecycleError::OwnerMismatch {
+                expected: claims.owner.key(),
+                actual: owner.key(),
+            });
+        }
+        if self.inner.clock.now_ms() >= claims.deadline_ms {
+            return Err(LifecycleError::DeadlineElapsed);
+        }
+        if claims.generation != self.inner.generation.load(Ordering::SeqCst) {
+            return Err(LifecycleError::Interrupted);
+        }
+        if requested > claims.risk_ceiling {
+            return Err(LifecycleError::ApprovalCeiling {
+                requested,
+                ceiling: claims.risk_ceiling,
+            });
+        }
+        Ok(claims)
+    }
+
     pub fn recover_open_tokens(&self) -> Result<Vec<String>, LifecycleError> {
         let mut states = self.inner.token_states.lock();
         let open: Vec<(String, String)> = states
@@ -485,6 +538,11 @@ impl ExecutionLease {
     pub fn token(&self) -> &str {
         &self.token
     }
+
+    /// Disarm the lease after a successful commit so Drop does not interrupt.
+    pub fn disarm(&mut self) {
+        self.closed = true;
+    }
 }
 
 impl Drop for ExecutionLease {

src/capabilities/types.rs

@@ -168,6 +168,7 @@ pub enum LifecycleError {
     Interrupted,
     RegistryMismatch,
     InvalidMetadata(String),
+    UnresolvedCall,
 }
 
 impl LifecycleError {
@@ -189,6 +190,7 @@ impl LifecycleError {
             Self::Interrupted => "interrupted",
             Self::RegistryMismatch => "registry_mismatch",
             Self::InvalidMetadata(_) => "invalid_metadata",
+            Self::UnresolvedCall => "unresolved_call",
         }
     }
 }

src/runtime/agent_host.rs

@@ -4,7 +4,7 @@
 //! through these host functions. Provider adapters stay in RSS; this module
 //! does not add an OpenAI-compatible inference path.
 
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
@@ -18,8 +18,8 @@ use serde_json::{Value as JsonValue, json};
 
 use super::rss_runner::RunCancellation;
 use crate::capabilities::{
-    CapabilityLifecycle, CapabilityOwner, LifecycleError, parse_prepare_metadata, tool_commit,
-    tool_prepare,
+    CapabilityLifecycle, CapabilityOwner, ExecutionLease, LifecycleError, parse_prepare_metadata,
+    tool_commit, tool_prepare,
 };
 use crate::domain::{ToolCall, json_to_vm_value, vm_value_to_json};
 use crate::metrics::Metrics;
@@ -141,6 +141,7 @@ pub struct AgentHostState {
     pub metrics: Option<Arc<Metrics>>,
     pub lifecycle: Option<Arc<CapabilityLifecycle>>,
     pub capability_owner: Option<CapabilityOwner>,
+    pub(crate) leases: Arc<Mutex<HashMap<String, ExecutionLease>>>,
 }
 
 impl AgentHostState {
@@ -172,10 +173,21 @@ impl AgentHostState {
                 "capability owner is not installed".to_string(),
             ));
         };
-        match parse_prepare_metadata(metadata) {
+        let envelope = match parse_prepare_metadata(metadata) {
             Ok(metadata) => tool_prepare(lifecycle, owner, metadata),
-            Err(error) => crate::capabilities::host::error_envelope(&error),
+            Err(error) => return crate::capabilities::host::error_envelope(&error),
+        };
+        if envelope.get("ok") == Some(&JsonValue::Bool(true))
+            && envelope.get("kind") == Some(&JsonValue::String("execute".to_string()))
+            && let Some(token) = envelope.get("execution_token").and_then(JsonValue::as_str)
+            && let Ok(lease) = lifecycle.lease(token)
+        {
+            self.leases
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .insert(token.to_string(), lease);
         }
+        envelope
     }
 
     fn capability_commit(&self, token: &str, result: &JsonValue) -> JsonValue {
@@ -189,7 +201,17 @@ impl AgentHostState {
                 "capability owner is not installed".to_string(),
             ));
         };
-        tool_commit(lifecycle, owner, token, result.clone())
+        let envelope = tool_commit(lifecycle, owner, token, result.clone());
+        if envelope.get("ok") == Some(&JsonValue::Bool(true))
+            && let Some(mut lease) = self
+                .leases
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .remove(token)
+        {
+            lease.disarm();
+        }
+        envelope
     }
 
     fn tool_dispatch(&self, call: &JsonValue) -> JsonValue {

src/runtime/rss_runner.rs

@@ -624,6 +624,7 @@ impl AgentRunner {
             metrics: self.host.metrics.clone(),
             lifecycle: self.host.lifecycle.clone(),
             capability_owner: self.host.capability_owner.clone(),
+            leases: Arc::new(Mutex::new(HashMap::new())),
         });
         if let Some(cancellation) = cancellation {
             vm.set_epoch_check_interval(RUN_EPOCH_CHECK_INTERVAL)

src/service.rs

@@ -303,6 +303,7 @@ impl NativeDispatchState {
         if let Some(observer) = &self.shutdown_entered {
             observer();
         }
+        let _ = self.lifecycle.recover_open_tokens();
         self.dispatcher.close();
         let quiesced = self.dispatcher.try_quiesce(grace);
         let owner = self.owner();
@@ -357,6 +358,19 @@ impl RunHandle {
 
     fn cancel_native_tools(&self) {
         self.tool_cancel.cancel();
+        let lifecycle = {
+            let phase = self
+                .native_dispatch
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            match &*phase {
+                NativeDispatchPhase::Ready(state) => Some(Arc::clone(&state.lifecycle)),
+                _ => None,
+            }
+        };
+        if let Some(lifecycle) = lifecycle {
+            let _ = lifecycle.recover_open_tokens();
+        }
     }
 
     fn native_dispatch_closed(&self) -> bool {
@@ -987,6 +1001,26 @@ impl AgentService {
         .commit_step(event_type, data, result)
     }
 
+    /// Run-scoped capability engine used by `agent_runtime::tool_prepare`
+    /// and `agent_runtime::tool_commit`. Initializes native dispatch if needed.
+    pub fn capability_lifecycle(
+        &self,
+        run_id: &str,
+    ) -> Result<(Arc<CapabilityLifecycle>, CapabilityOwner), RunContextError> {
+        let handle = self
+            .handle(run_id)
+            .ok_or_else(|| RunContextError::Missing {
+                run_id: run_id.to_string(),
+            })?;
+        match self.native_dispatch_state(run_id, &handle)? {
+            Some(state) => Ok((Arc::clone(&state.lifecycle), state.capability_owner.clone())),
+            None => Err(RunContextError::InvalidMetadata {
+                run_id: run_id.to_string(),
+                reason: "native dispatch is closed".to_string(),
+            }),
+        }
+    }
+
     /// Serial, validated native dispatch against the admitted registry snapshot.
     ///
     /// The live registry is not consulted. Durable event append uses the same
@@ -2952,6 +2986,7 @@ impl AgentService {
             // observing the cancellation commits exactly this reason.
             *handle.cancel_reason.lock().expect("cancel reason lock") = Some("requested");
             handle.cancel.request(CancellationReason::Requested);
+            drop(store);
             handle.cancel_native_tools();
             tracing::debug!(
                 run_id,
@@ -4222,14 +4257,21 @@ impl DurableToolLifecycle for ServiceDurableLifecycle {
         call_id: &str,
         result: &serde_json::Value,
     ) -> Result<serde_json::Value, LifecycleError> {
+        let tool_result = canonical_tool_result(result)?;
+        let event_type = if tool_result.ok {
+            "tool.completed"
+        } else {
+            "tool.failed"
+        };
+        let mut data = json!({
+            "tool_call_id": call_id,
+            "ok": tool_result.ok,
+        });
+        if let Some(error) = &tool_result.error {
+            data["error_code"] = json!(error.code);
+        }
         self.events
-            .commit(
-                "tool.completed",
-                json!({
-                    "tool_call_id": call_id,
-                    "result": result,
-                }),
-            )
+            .commit_step(event_type, data, Some(&tool_result))
             .map_err(|error| match error {
                 EventCommitError::PersistFailed(message) => {
                     LifecycleError::ResultCommitFailed(message)
@@ -4240,16 +4282,17 @@ impl DurableToolLifecycle for ServiceDurableLifecycle {
     }
 
     fn interrupt(&self, call_id: &str) -> Result<(), LifecycleError> {
+        let tool_result =
+            ToolResult::failure("interrupted_effect", "effect interrupted by restart");
         self.events
-            .commit(
+            .commit_step(
                 "tool.failed",
                 json!({
                     "tool_call_id": call_id,
-                    "error": {
-                        "code": "interrupted",
-                        "message": "execution was interrupted",
-                    }
+                    "error_code": "interrupted_effect",
+                    "ok": false,
                 }),
+                Some(&tool_result),
             )
             .map_err(map_event_commit_error)
     }
@@ -4265,6 +4308,46 @@ fn map_event_commit_error(error: EventCommitError) -> LifecycleError {
     }
 }
 
+fn canonical_tool_result(result: &JsonValue) -> Result<ToolResult, LifecycleError> {
+    let ok = result
+        .get("ok")
+        .and_then(JsonValue::as_bool)
+        .unwrap_or(false);
+    if ok {
+        let mut tool_result = ToolResult::success(
+            result
+                .get("content")
+                .and_then(JsonValue::as_str)
+                .unwrap_or("")
+                .to_string(),
+            result.get("data").cloned().unwrap_or_else(|| json!({})),
+        );
+        tool_result.truncated = result
+            .get("truncated")
+            .and_then(JsonValue::as_bool)
+            .unwrap_or(false);
+        if let Some(artifacts) = result.get("artifacts").and_then(JsonValue::as_array) {
+            tool_result.artifacts = artifacts
+                .iter()
+                .filter_map(JsonValue::as_str)
+                .map(str::to_string)
+                .collect();
+        }
+        Ok(tool_result)
+    } else {
+        let error = result.get("error");
+        let code = error
+            .and_then(|value| value.get("code"))
+            .and_then(JsonValue::as_str)
+            .unwrap_or("tool_failed");
+        let message = error
+            .and_then(|value| value.get("message"))
+            .and_then(JsonValue::as_str)
+            .unwrap_or("tool failed");
+        Ok(ToolResult::failure(code, message))
+    }
+}
+
 struct ServiceEventCommitter {
     store: Arc<RwLock<GatewayStore>>,
     persistence: Option<Arc<GatewayPersistence>>,