From daaace0d95115f276ff349dbbf0f98eb6bf0dfa8 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Mon, 19 Oct 2026 14:28:21 -0400 Subject: [PATCH] feat(desktop): add Tauri commands to end and recover remote thread sessions Add the backend half of explicit thread-session lifecycle control for thread-sandbox providers. Providers may advertise `ops` in `info`; `provider_stop` sends the secret-bearing `stop` request only after the same staged bytes negotiated protocol v1 and advertised `stop`. New commands `end_remote_agent_session`, `recover_remote_agent_session` and `remote_agent_session_supports_lifecycle` share the deploy path's per-agent lock, post-lock payload rebuild, tenant scope assertion, and session binding. The recovery mode rides one invocation's provider config and is never persisted on the agent record. Expose typed TypeScript bindings for the three commands. Signed-off-by: Tee8z --- crates/buzz-backend-kubernetes/src/main.rs | 5 +- crates/buzz-backend-kubernetes/src/wire.rs | 15 +- .../tests/wire_fixtures.rs | 9 +- .../src-tauri/src/commands/agent_providers.rs | 1 + desktop/src-tauri/src/commands/agents.rs | 97 ++--- .../src/commands/agents/provider_access.rs | 52 +-- .../src/commands/agents/provider_deploy.rs | 398 ++++++++++++++---- .../src/commands/agents/remote_session.rs | 202 +++++++++ .../commands/agents/remote_session_tests.rs | 174 ++++++++ .../src-tauri/src/commands/agents_tests.rs | 16 +- .../src/commands/personas/inbound.rs | 53 +-- desktop/src-tauri/src/lib.rs | 3 + .../src-tauri/src/managed_agents/backend.rs | 177 +++++++- .../src/managed_agents/backend_tests.rs | 129 +++++- desktop/src/shared/api/tauriManagedAgents.ts | 58 +++ docs/remote-agents.md | 17 +- 16 files changed, 1147 insertions(+), 259 deletions(-) create mode 100644 desktop/src-tauri/src/commands/agents/remote_session.rs create mode 100644 desktop/src-tauri/src/commands/agents/remote_session_tests.rs diff --git a/crates/buzz-backend-kubernetes/src/main.rs b/crates/buzz-backend-kubernetes/src/main.rs index 5e81eaa3bed..c48af0b5fd1 100644 --- a/crates/buzz-backend-kubernetes/src/main.rs +++ b/crates/buzz-backend-kubernetes/src/main.rs @@ -253,7 +253,10 @@ mod tests { assert_eq!(json["ok"], true); assert_eq!(json["protocol_version"], wire::PROTOCOL_VERSION); assert!(json["config_schema"]["properties"]["namespace"].is_object()); - assert!(json.get("ops").is_none()); + assert_eq!( + json["ops"], + serde_json::json!(["info", "deploy", "stop", "recover"]) + ); } /// `stop` is gated like a preflight: no approved identity policy, no diff --git a/crates/buzz-backend-kubernetes/src/wire.rs b/crates/buzz-backend-kubernetes/src/wire.rs index d31a46a1431..71b90b3917f 100644 --- a/crates/buzz-backend-kubernetes/src/wire.rs +++ b/crates/buzz-backend-kubernetes/src/wire.rs @@ -12,6 +12,12 @@ use std::collections::BTreeMap; /// The wire-contract version this provider speaks (spec §Info). pub const PROTOCOL_VERSION: u32 = 1; +/// What this provider supports, advertised in `info.ops`. Additive within +/// protocol version 1. Each entry is a request `op`, except `recover`: it +/// advertises that `deploy` accepts `provider_config.sandbox.recovery`. A +/// desktop gates its session end/recover actions on `recover`. +pub const OPS: &[&str] = &["info", "deploy", "stop", "recover"]; + /// Request envelope. `op` discriminates; unknown ops are an in-band error. /// /// `request_id` is deliberately absent from every variant. The desktop sends @@ -108,6 +114,7 @@ pub struct InfoResponse { pub name: &'static str, pub version: &'static str, pub protocol_version: u32, + pub ops: &'static [&'static str], pub description: &'static str, pub config_schema: serde_json::Value, } @@ -267,6 +274,7 @@ impl Response { name: "kubernetes", version: env!("CARGO_PKG_VERSION"), protocol_version: PROTOCOL_VERSION, + ops: OPS, description: "Runs agents as pods in a Kubernetes cluster", config_schema: crate::config::config_schema(), }) @@ -335,9 +343,12 @@ mod tests { } #[test] - fn info_remains_compatible_until_the_desktop_accepts_ops() { + fn info_advertises_stop_and_recover() { let v = serde_json::to_value(Response::info()).unwrap(); - assert!(v.get("ops").is_none()); + assert_eq!( + v["ops"], + serde_json::json!(["info", "deploy", "stop", "recover"]) + ); } #[test] diff --git a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs index 962f8d6b97b..6a211a0eb81 100644 --- a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs +++ b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs @@ -149,9 +149,12 @@ fn info_response_carries_the_contract_fields() { assert_eq!(info["ok"], true); assert_eq!(info["protocol_version"], 1); assert_eq!(info["name"], "kubernetes"); - // `ops` is not advertised yet: a desktop that predates it must keep - // accepting `info`. A later change adds it, additive within version 1. - assert!(info.get("ops").is_none()); + // Additive within protocol version 1; the desktop gates its session + // end/recover actions on `recover`. + assert_eq!( + info["ops"], + serde_json::json!(["info", "deploy", "stop", "recover"]) + ); let schema = &info["config_schema"]; assert_eq!( schema["required"], diff --git a/desktop/src-tauri/src/commands/agent_providers.rs b/desktop/src-tauri/src/commands/agent_providers.rs index 0eaca0192ca..69e51a020a6 100644 --- a/desktop/src-tauri/src/commands/agent_providers.rs +++ b/desktop/src-tauri/src/commands/agent_providers.rs @@ -58,6 +58,7 @@ pub async fn probe_backend_provider(binary_path: String) -> Result( && input.backend != BackendKind::Local && !input.backend.is_thread_sandbox() { - if let BackendKind::Provider { ref id, ref config } = input.backend { - let agent_json = { + if let BackendKind::Provider { .. } = input.backend { + { + // Fail create on a payload that cannot be built; deploy + // rebuilds it after its lock. let _g = state .managed_agents_store_lock .lock() @@ -840,15 +844,11 @@ async fn create_managed_agent_in( .iter() .find(|r| r.pubkey == pubkey) .ok_or_else(|| "agent disappeared".to_string())?; - build_deploy_payload(&app, state, rec)? - }; - match deploy_to_provider( - &app, state, &pubkey, id, config, agent_json, None, None, None, None, None, - ) - .await - { + build_deploy_payload(&app, state, rec)?; + } + match deploy_to_provider(&app, state, &pubkey, &Default::default(), None).await { Ok(()) => spawn_error, - Err(e) => Some(e), + Err(e) => Some(e.message), } } else { spawn_error @@ -895,7 +895,7 @@ pub async fn start_managed_agent( session_scope: Option, app: AppHandle, state: State<'_, AppState>, -) -> Result { +) -> Result { // Snapshot the workspace owner pubkey for the legacy auth_tag fallback. // Read outside the records lock to keep lock ordering simple. let owner_hex = workspace_owner_hex(&state)?; @@ -926,11 +926,7 @@ pub async fn start_managed_agent( )?; enum StartTarget { Local, - Provider { - backend: BackendKind, - cached_binary_path: Option, - agent_json: serde_json::Value, - }, + Provider { backend: BackendKind }, } // Collect backend info under lock; async preflight/spawn happens below. @@ -973,10 +969,11 @@ pub async fn start_managed_agent( let target = if record.backend == BackendKind::Local { StartTarget::Local } else { + // Fail before the deploy lock on a payload that cannot be + // built; deploy rebuilds it after the lock. + build_deploy_payload(&app, &state, record)?; StartTarget::Provider { backend: record.backend.clone(), - cached_binary_path: record.provider_binary_path.clone(), - agent_json: build_deploy_payload(&app, &state, record)?, } }; @@ -984,22 +981,19 @@ pub async fn start_managed_agent( }; let result = match target { - StartTarget::Local => { - start_local_agent_with_preflight( - &app, - &state, - &pubkey, - false, - expected_relay_url.as_deref(), - expected_signer_pubkey.as_deref(), - replay_floor_unix, - ) - .await - } + StartTarget::Local => start_local_agent_with_preflight( + &app, + &state, + &pubkey, + false, + expected_relay_url.as_deref(), + expected_signer_pubkey.as_deref(), + replay_floor_unix, + ) + .await + .map_err(ProviderFailure::from), StartTarget::Provider { - backend: BackendKind::Provider { id, config }, - cached_binary_path, - agent_json, + backend: BackendKind::Provider { .. }, } => { // The caller's captured scope is asserted INSIDE deploy_to_provider // against the payload rebuilt after the deploy lock — the exact @@ -1008,20 +1002,13 @@ pub async fn start_managed_agent( // The replay floor rides along so a publish-first mention send's // remote harness replays past the already-published message, same // as the local spawn path. - deploy_to_provider( - &app, - &state, - &pubkey, - &id, - &config, - agent_json, - cached_binary_path.as_deref(), - expected_relay_url.as_deref(), - expected_signer_pubkey.as_deref(), - replay_floor_unix, - session_scope.as_ref(), - ) - .await?; + let scope = InvocationScope { + expected_relay_url: expected_relay_url.as_deref(), + expected_signer_pubkey: expected_signer_pubkey.as_deref(), + session: session_scope.as_ref(), + action: ProviderAction::Deploy, + }; + deploy_to_provider(&app, &state, &pubkey, &scope, replay_floor_unix).await?; // Return updated summary. let _store_guard = state @@ -1037,11 +1024,11 @@ pub async fn start_managed_agent( .iter() .find(|r| r.pubkey == pubkey) .ok_or_else(|| format!("agent {pubkey} not found"))?; - summarize_from_disk(&app, record, &runtimes) + Ok(summarize_from_disk(&app, record, &runtimes)?) + } + StartTarget::Provider { backend } => { + Err(format!("agent {pubkey} has unsupported backend kind: {backend:?}").into()) } - StartTarget::Provider { backend, .. } => Err(format!( - "agent {pubkey} has unsupported backend kind: {backend:?}" - )), }; // ── Profile reconciliation (fire-and-forget) ──────────────────────────── @@ -1229,11 +1216,15 @@ pub async fn delete_managed_agent( mod deploy; pub(super) mod provider_access; mod provider_deploy; +mod remote_session; pub(super) use deploy::build_deploy_payload; #[cfg(test)] use deploy::{deploy_payload_json, DeployProjections}; #[cfg(test)] use deploy::{ensure_remote_provider_supported, resolve_deploy_model_provider}; +pub use remote_session::{ + end_remote_agent_session, recover_remote_agent_session, remote_agent_session_supports_lifecycle, +}; #[path = "agents_profile.rs"] mod profile; diff --git a/desktop/src-tauri/src/commands/agents/provider_access.rs b/desktop/src-tauri/src/commands/agents/provider_access.rs index caa139bc4e1..bfdac9a5499 100644 --- a/desktop/src-tauri/src/commands/agents/provider_access.rs +++ b/desktop/src-tauri/src/commands/agents/provider_access.rs @@ -26,9 +26,6 @@ pub(super) fn needs_reconciliation_with_policy( #[derive(Debug)] struct ProviderAccessTarget { pubkey: String, - provider_id: String, - config: serde_json::Value, - cached_binary_path: Option, agent_json: Result, } @@ -40,13 +37,10 @@ fn collect_targets_with( records .into_iter() .filter(|record| needs_reconciliation_with_policy(record, owner_only_access)) - .map(|record| match record.backend.clone() { - BackendKind::Provider { id, config } => ProviderAccessTarget { + .map(|record| match record.backend { + BackendKind::Provider { .. } => ProviderAccessTarget { agent_json: build_payload(&record), pubkey: record.pubkey, - provider_id: id, - config, - cached_binary_path: record.provider_binary_path, }, BackendKind::Local => { unreachable!("provider access reconciliation selected a local agent") @@ -77,36 +71,15 @@ pub(crate) async fn reconcile_on_workspace_apply( }; for target in targets { - let ProviderAccessTarget { - pubkey, - provider_id, - config, - cached_binary_path, - agent_json, - } = target; - let agent_json = match agent_json { - Ok(agent_json) => agent_json, - Err(error) => { - persist_failure(app, state, &pubkey, &error)?; - return Err(format!( - "provider access reconciliation failed for agent {pubkey}: {error}" - )); - } - }; - if let Err(error) = super::deploy_to_provider( - app, - state, - &pubkey, - &provider_id, - &config, - agent_json, - cached_binary_path.as_deref(), - None, - None, - None, - None, - ) - .await + let ProviderAccessTarget { pubkey, agent_json } = target; + if let Err(error) = agent_json { + persist_failure(app, state, &pubkey, &error)?; + return Err(format!( + "provider access reconciliation failed for agent {pubkey}: {error}" + )); + } + if let Err(error) = + super::deploy_to_provider(app, state, &pubkey, &Default::default(), None).await { return Err(format!( "provider access reconciliation failed for agent {pubkey}: {error}" @@ -198,8 +171,6 @@ mod tests { assert_eq!(targets.len(), 1); assert_eq!(targets[0].pubkey, "agent"); - assert_eq!(targets[0].provider_id, "provider"); - assert_eq!(targets[0].config["region"], "test"); assert_eq!( targets[0].agent_json.as_ref().unwrap()["respond_to"], "owner-only" @@ -231,7 +202,6 @@ mod tests { assert_eq!(targets.len(), 1); assert_eq!(targets[0].pubkey, "pending-agent"); - assert_eq!(targets[0].provider_id, "pending-provider"); assert_eq!( targets[0].agent_json.as_ref().unwrap()["pubkey"], "pending-agent" diff --git a/desktop/src-tauri/src/commands/agents/provider_deploy.rs b/desktop/src-tauri/src/commands/agents/provider_deploy.rs index eb2d307898f..fde3baa7406 100644 --- a/desktop/src-tauri/src/commands/agents/provider_deploy.rs +++ b/desktop/src-tauri/src/commands/agents/provider_deploy.rs @@ -6,7 +6,8 @@ use crate::{ app_state::AppState, managed_agents::{ discover_provider_candidates, load_managed_agents, provider_deploy, - resolve_provider_binary, save_managed_agents, BackendKind, REPLAY_FLOOR_ENV_VAR, + resolve_provider_binary, save_managed_agents, BackendKind, ProviderFailure, + REPLAY_FLOOR_ENV_VAR, }, util::now_iso, }; @@ -21,6 +22,18 @@ pub struct RemoteSessionScope { thread_root: String, } +/// How an explicit user action recovers an ended thread session. Sent as +/// `provider_config.sandbox.recovery` for one invocation only; never stored +/// on the agent record, so no later deploy can resurrect a session. +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "lowercase")] +pub enum RemoteRecoveryMode { + /// Restore the ended session's verified checkpoint. + Checkpoint, + /// Start a new session with an empty workspace. + Fresh, +} + /// Replace a thread Sandbox's `sandbox: true` request with the published /// thread and the rebuilt payload's verified owner. Callers decide eligibility /// with [`BackendKind::is_thread_sandbox`] on the rebuilt record. @@ -29,6 +42,7 @@ fn bind_sandbox_scope( scope: Option<&RemoteSessionScope>, agent_pubkey: &str, payload: &serde_json::Value, + recovery: Option, ) -> Result<(), String> { let scope = scope.ok_or("Mention this agent in a channel thread to start its workspace")?; if scope.thread_root.len() != 64 @@ -46,56 +60,27 @@ fn bind_sandbox_scope( config["sandbox"] = serde_json::json!({ "channel_id": scope.channel_id.to_string(), "thread_root": scope.thread_root, }); + if let Some(mode) = recovery { + config["sandbox"]["recovery"] = serde_json::json!({ "mode": mode }); + } config["identity_policy"] = serde_json::json!({ "agent_pubkey": agent_pubkey, "owner_pubkey": owner, }); Ok(()) } -/// Deploy an agent to a provider backend. Resolves the binary, calls deploy via -/// spawn_blocking, and persists the result (backend_agent_id or last_error). +/// Serialize provider operations per agent. The guard must stay alive until +/// the provider invocation finishes. /// -/// Idempotency: calling deploy on an already-deployed agent sends the same payload -/// again. Providers are expected to handle this as an update-in-place or no-op. -/// The protocol has no explicit `undeploy` operation or acknowledgement that an -/// existing process stopped, so a successful redeploy delegates access-policy -/// revocation semantics to the provider implementation (deferred to v2). -/// Returns Ok(()) on success, Err(message) on failure. Either way the record is -/// updated and saved before returning. -/// -/// Callers with a captured tenant scope (Projects agent starts) pass -/// `expected_relay_url` / `expected_signer_pubkey`; they are asserted against -/// the payload REBUILT after the deploy lock — the exact value invoked — so a -/// workspace or identity switch landing while this call waited behind another -/// deployment fails closed instead of deploying a stale start into the new -/// tenant under the new tenant's owner identity. `None` preserves the -/// unscoped behavior for callers without a tenant boundary. -/// -/// `replay_floor_unix`: optional unix-seconds replay floor from a -/// publish-first mention send. It is injected into the rebuilt payload's -/// `launch.policy_env` as `BUZZ_ACP_REPLAY_FLOOR`, so the remote harness's -/// startup watermark replays back past the already-published triggering -/// message exactly like a local spawn. Per-invocation only — never persisted -/// on the record, so later redeploys do not carry a stale floor. -#[allow(clippy::too_many_arguments)] -pub(crate) async fn deploy_to_provider( - app: &AppHandle, +/// One lock per agent, not per thread: it serializes the record rebuild, +/// provider call, and result write. A thread Sandbox cold start holds it +/// through the provider's Pod waits (120s bound each), so starts for the same +/// agent's other threads queue behind it. +pub(super) async fn lock_provider_operations( state: &AppState, pubkey: &str, - _provider_id: &str, - _config: &serde_json::Value, - _agent_json: serde_json::Value, - _cached_binary_path: Option<&str>, - expected_relay_url: Option<&str>, - expected_signer_pubkey: Option<&str>, - replay_floor_unix: Option, - session_scope: Option<&RemoteSessionScope>, -) -> Result<(), String> { - // One lock per agent, not per thread: it serializes the record rebuild, - // provider call, and result write. A thread Sandbox cold start holds it - // through the provider's Pod waits (120s bound each), so starts for the - // same agent's other threads queue behind it. - let deploy_lock = { +) -> Result, String> { + let lock = { let mut locks = state .provider_deploy_locks .lock() @@ -106,11 +91,63 @@ pub(crate) async fn deploy_to_provider( .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))), ) }; - let _deploy_guard = deploy_lock.lock().await; - // The payload may have waited behind another deployment. Rebuild it from - // the current record so the final provider invocation always carries the - // newest saved policy rather than the stale snapshot captured by its caller. - let (provider_id, mut config, thread_sandbox, cached_binary_path, mut agent_json) = { + Ok(lock.lock_owned().await) +} + +/// A provider invocation rebuilt from the current record after the lock. +pub(super) struct ProviderInvocation { + pub(super) binary: std::path::PathBuf, + pub(super) config: serde_json::Value, + pub(super) agent_json: serde_json::Value, +} + +/// What one provider invocation does for the user. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum ProviderAction { + /// Start or update the agent (an ordinary deploy). + #[default] + Deploy, + /// End the thread's session with the provider's `stop`. + End, + /// Recover the thread's ended session: a deploy carrying the mode. + Recover(RemoteRecoveryMode), +} + +/// Session inputs for one provider invocation. None of them are persisted. +/// +/// Callers with a captured tenant scope (Projects agent starts, remote session +/// actions) set `expected_relay_url` / `expected_signer_pubkey`; they are +/// asserted against the payload REBUILT after the deploy lock, the exact value +/// invoked. `None` preserves the unscoped behavior for callers without a +/// tenant boundary. +#[derive(Default)] +pub(crate) struct InvocationScope<'a> { + pub(crate) expected_relay_url: Option<&'a str>, + pub(crate) expected_signer_pubkey: Option<&'a str>, + pub(crate) session: Option<&'a RemoteSessionScope>, + pub(crate) action: ProviderAction, +} + +impl InvocationScope<'_> { + fn recovery(&self) -> Option { + match self.action { + ProviderAction::Recover(mode) => Some(mode), + ProviderAction::Deploy | ProviderAction::End => None, + } + } +} + +/// Rebuild the invocation from the current record. Call only while holding +/// [`lock_provider_operations`]: the payload may have waited behind another +/// operation, so the final invocation must carry the newest saved policy, not +/// the stale snapshot its caller captured. +pub(super) fn prepare_provider_invocation( + app: &AppHandle, + state: &AppState, + pubkey: &str, + scope: &InvocationScope<'_>, +) -> Result { + let (provider_id, mut config, thread_sandbox, cached_binary_path, agent_json) = { let _store_guard = state .managed_agents_store_lock .lock() @@ -134,33 +171,93 @@ pub(crate) async fn deploy_to_provider( }; // The rebuild above re-read the live workspace relay and owner identity. // Assert the caller's captured scope against THIS payload — the exact - // value invoked below — not the pre-lock snapshot its caller validated. - assert_payload_scope(&agent_json, expected_relay_url, expected_signer_pubkey)?; + // value invoked — not the pre-lock snapshot its caller validated. + assert_payload_scope(&agent_json, scope)?; + // `config` is a copy: binding the session and any recovery mode never + // reaches the record's saved `sandbox: true`. if thread_sandbox { - bind_sandbox_scope(&mut config, session_scope, pubkey, &agent_json)?; + bind_sandbox_scope( + &mut config, + scope.session, + pubkey, + &agent_json, + scope.recovery(), + )?; + } else if scope.action != ProviderAction::Deploy { + return Err("Only thread workspaces can be ended or recovered".into()); } - // The floor is invocation state, not record state, so the post-lock - // rebuild cannot restore it — inject it into the payload actually invoked. - apply_replay_floor(&mut agent_json, replay_floor_unix); - // Resolve via discovered candidates only. Cached path must match BOTH - // "is a discovered candidate" AND "belongs to this provider_id". A tampered - // record cannot redirect deploys to a different provider's binary. - let bin_path = cached_binary_path - .as_deref() + Ok(ProviderInvocation { + binary: resolve_record_provider_binary(&provider_id, cached_binary_path.as_deref())?, + config, + agent_json, + }) +} + +/// Resolve via discovered candidates only. Cached path must match BOTH +/// "is a discovered candidate" AND "belongs to this provider_id". A tampered +/// record cannot redirect provider calls to a different provider's binary. +pub(super) fn resolve_record_provider_binary( + provider_id: &str, + cached_binary_path: Option<&str>, +) -> Result { + cached_binary_path .map(std::path::PathBuf::from) .filter(|p| p.exists()) .map(|p| p.canonicalize().unwrap_or(p)) .filter(|canonical| { discover_provider_candidates().iter().any(|(id, cp)| { - id == &provider_id && cp.canonicalize().ok().as_ref() == Some(canonical) + id == provider_id && cp.canonicalize().ok().as_ref() == Some(canonical) }) }) - .map_or_else(|| resolve_provider_binary(&provider_id), Ok)?; + .map_or_else(|| resolve_provider_binary(provider_id), Ok) +} + +/// Deploy an agent to a provider backend. Resolves the binary, calls deploy via +/// spawn_blocking, and persists the result (backend_agent_id or last_error). +/// +/// Idempotency: calling deploy on an already-deployed agent sends the same payload +/// again. Providers are expected to handle this as an update-in-place or no-op. +/// Thread-sandbox providers end a session with the separate `stop` operation +/// (`end_remote_agent_session`); other providers have no undeploy operation, so +/// a successful redeploy delegates access-policy revocation semantics to the +/// provider implementation. +/// Returns Ok(()) on success, Err on failure, with the provider's refusal code +/// when it sent one. Either way the record is updated and saved before +/// returning. +/// +/// `scope`'s tenant expectations are asserted against the payload rebuilt +/// after the deploy lock, so a workspace or identity switch landing while this +/// call waited behind another deployment fails closed instead of deploying a +/// stale start into the new tenant under the new tenant's owner identity. A +/// recovery mode in `scope.action` rides only this invocation's config. +/// +/// `replay_floor_unix`: optional unix-seconds replay floor from a +/// publish-first mention send. It is injected into the rebuilt payload's +/// `launch.policy_env` as `BUZZ_ACP_REPLAY_FLOOR`, so the remote harness's +/// startup watermark replays back past the already-published triggering +/// message exactly like a local spawn. Per-invocation only — never persisted +/// on the record, so later redeploys do not carry a stale floor. +pub(crate) async fn deploy_to_provider( + app: &AppHandle, + state: &AppState, + pubkey: &str, + scope: &InvocationScope<'_>, + replay_floor_unix: Option, +) -> Result<(), ProviderFailure> { + let _deploy_guard = lock_provider_operations(state, pubkey).await?; + let ProviderInvocation { + binary: bin_path, + config, + mut agent_json, + .. + } = prepare_provider_invocation(app, state, pubkey, scope)?; + // The floor is invocation state, not record state, so the post-lock + // rebuild cannot restore it — inject it into the payload actually invoked. + apply_replay_floor(&mut agent_json, replay_floor_unix); let deployed_agent_json = agent_json.clone(); - let config_clone = config.clone(); let deploy_result = - tokio::task::spawn_blocking(move || provider_deploy(&bin_path, &agent_json, &config_clone)) + tokio::task::spawn_blocking(move || provider_deploy(&bin_path, &agent_json, &config)) .await .map_err(|e| format!("spawn_blocking failed: {e}"))?; @@ -175,9 +272,14 @@ pub(crate) async fn deploy_to_provider( .find(|r| r.pubkey == pubkey) .ok_or_else(|| format!("agent {pubkey} not found"))?; - let result = apply_deploy_result(rec, deploy_result, &deployed_agent_json); + let code = deploy_result.as_ref().err().and_then(|e| e.code.clone()); + let result = apply_deploy_result( + rec, + deploy_result.map_err(String::from), + &deployed_agent_json, + ); save_managed_agents(app, &records)?; - result + result.map_err(|message| ProviderFailure { message, code }) } /// Assert a caller-captured tenant scope against the payload that will @@ -189,16 +291,29 @@ pub(crate) async fn deploy_to_provider( /// deploy on behalf of a scoped callback. fn assert_payload_scope( agent_json: &serde_json::Value, - expected_relay_url: Option<&str>, - expected_signer_pubkey: Option<&str>, + scope: &InvocationScope<'_>, ) -> Result<(), String> { + let (expected_relay_url, expected_signer_pubkey) = + (scope.expected_relay_url, scope.expected_signer_pubkey); + // The shared checks word a mismatch for a message send; say what this + // action did instead, keeping the "active ... changed" prefix. + let reword = |error: String, what: &str| match scope.action { + ProviderAction::Deploy => error, + ProviderAction::End => { + format!("active {what} changed before the session was ended; nothing was changed") + } + ProviderAction::Recover(_) => { + format!("active {what} changed before the session was recovered; nothing was changed") + } + }; let has_expectation = |expected: Option<&str>| expected.map(str::trim).filter(|s| !s.is_empty()).is_some(); match agent_json.get("relay_url").and_then(|v| v.as_str()) { Some(embedded_relay) => crate::relay::assert_expected_relay_scope( expected_relay_url, &crate::relay::relay_http_base_url(embedded_relay), - )?, + ) + .map_err(|error| reword(error, "community"))?, None if has_expectation(expected_relay_url) => { return Err("deploy payload carries no relay; not deployed".to_string()); } @@ -209,7 +324,8 @@ fn assert_payload_scope( .and_then(|launch| launch.get("owner_pubkey")) .and_then(|v| v.as_str()) { - Some(owner) => crate::relay::assert_expected_signer(expected_signer_pubkey, owner)?, + Some(owner) => crate::relay::assert_expected_signer(expected_signer_pubkey, owner) + .map_err(|error| reword(error, "identity"))?, None if has_expectation(expected_signer_pubkey) => { return Err("deploy payload carries no owner identity; not deployed".to_string()); } @@ -318,21 +434,103 @@ mod tests { }; let mut config = serde_json::json!({"sandbox":true}); let payload = serde_json::json!({"launch":{"owner_pubkey":"owner"}}); - assert!(bind_sandbox_scope(&mut config, None, "agent", &payload).is_err()); + assert!(bind_sandbox_scope(&mut config, None, "agent", &payload, None).is_err()); assert_eq!(config["sandbox"], true); - bind_sandbox_scope(&mut config, Some(&scope), "agent", &payload).unwrap(); + bind_sandbox_scope(&mut config, Some(&scope), "agent", &payload, None).unwrap(); assert_eq!(config["sandbox"]["thread_root"], scope.thread_root); assert_eq!(config["identity_policy"]["owner_pubkey"], "owner"); + assert!( + config["sandbox"].get("recovery").is_none(), + "an ordinary deploy must never carry a recovery mode" + ); let mut missing_owner = serde_json::json!({"sandbox":true}); assert!(bind_sandbox_scope( &mut missing_owner, Some(&scope), "agent", - &serde_json::json!({}) + &serde_json::json!({}), + None ) .is_err()); } + fn thread_scope() -> RemoteSessionScope { + RemoteSessionScope { + channel_id: uuid::Uuid::new_v4(), + thread_root: "b".repeat(64), + } + } + + #[test] + fn recovery_mode_rides_only_the_bound_invocation_config() { + let payload = serde_json::json!({"launch":{"owner_pubkey":"owner"}}); + for (mode, wire) in [ + (RemoteRecoveryMode::Checkpoint, "checkpoint"), + (RemoteRecoveryMode::Fresh, "fresh"), + ] { + let mut record = record(); + record.backend = BackendKind::Provider { + id: "kubernetes".into(), + config: serde_json::json!({"sandbox": true}), + }; + // Same copy-then-bind sequence as `prepare_provider_invocation`. + let BackendKind::Provider { config, .. } = &record.backend else { + unreachable!() + }; + let mut invoked = config.clone(); + bind_sandbox_scope( + &mut invoked, + Some(&thread_scope()), + "agent", + &payload, + Some(mode), + ) + .unwrap(); + assert_eq!( + invoked["sandbox"]["recovery"], + serde_json::json!({"mode": wire}) + ); + assert_eq!(invoked["sandbox"]["thread_root"], "b".repeat(64)); + + apply_deploy_result(&mut record, Ok("remote".into()), &payload).unwrap(); + let saved = serde_json::to_value(&record).unwrap(); + assert_eq!( + saved["backend"]["config"], + serde_json::json!({"sandbox": true}), + "recovery must never be persisted on the agent record" + ); + } + } + + #[test] + fn recovery_is_refused_without_a_thread_workspace() { + let payload = serde_json::json!({"launch":{"owner_pubkey":"owner"}}); + let mut sandbox = serde_json::json!({"sandbox": true}); + let error = bind_sandbox_scope( + &mut sandbox, + None, + "agent", + &payload, + Some(RemoteRecoveryMode::Fresh), + ) + .unwrap_err(); + assert!(error.contains("channel thread"), "{error}"); + assert_eq!(sandbox, serde_json::json!({"sandbox": true})); + } + + #[test] + fn recovery_mode_deserializes_from_lowercase_wire_names() { + assert_eq!( + serde_json::from_str::(r#""checkpoint""#).unwrap(), + RemoteRecoveryMode::Checkpoint + ); + assert_eq!( + serde_json::from_str::(r#""fresh""#).unwrap(), + RemoteRecoveryMode::Fresh + ); + assert!(serde_json::from_str::(r#""Checkpoint""#).is_err()); + } + fn record() -> crate::managed_agents::ManagedAgentRecord { serde_json::from_value(serde_json::json!({ "pubkey": "agent", "name": "Agent", "relay_url": "", "acp_command": "", @@ -358,12 +556,50 @@ mod tests { // ── assert_payload_scope: post-lock rebuilt-payload validation ────────── + fn tenant<'a>(relay: Option<&'a str>, signer: Option<&'a str>) -> InvocationScope<'a> { + InvocationScope { + expected_relay_url: relay, + expected_signer_pubkey: signer, + ..InvocationScope::default() + } + } + + #[test] + fn a_scope_mismatch_names_the_session_action_it_refused() { + let payload = scoped_payload("wss://tenant-b.example", "bb22"); + for (action, relay_text, signer_text) in [ + ( + ProviderAction::End, + "active community changed before the session was ended", + "active identity changed before the session was ended", + ), + ( + ProviderAction::Recover(RemoteRecoveryMode::Fresh), + "active community changed before the session was recovered", + "active identity changed before the session was recovered", + ), + ] { + let relay = InvocationScope { + action, + ..tenant(Some("wss://tenant-a.example"), None) + }; + let error = assert_payload_scope(&payload, &relay).unwrap_err(); + assert!(error.starts_with(relay_text), "{error}"); + let signer = InvocationScope { + action, + ..tenant(None, Some("aa11")) + }; + let error = assert_payload_scope(&payload, &signer).unwrap_err(); + assert!(error.starts_with(signer_text), "{error}"); + assert!(!error.contains("not sent"), "{error}"); + } + } + #[test] fn matching_scope_and_signer_pass_on_the_rebuilt_payload() { assert_payload_scope( &scoped_payload("wss://tenant-a.example", "aa11"), - Some("wss://tenant-a.example"), - Some("aa11"), + &tenant(Some("wss://tenant-a.example"), Some("aa11")), ) .unwrap(); } @@ -375,8 +611,7 @@ mod tests { // refused — the pre-lock snapshot its caller validated is irrelevant. let error = assert_payload_scope( &scoped_payload("wss://tenant-b.example", "aa11"), - Some("wss://tenant-a.example"), - Some("aa11"), + &tenant(Some("wss://tenant-a.example"), Some("aa11")), ) .unwrap_err(); assert!(error.contains("active community changed"), "{error}"); @@ -389,8 +624,7 @@ mod tests { // caller never validated. let error = assert_payload_scope( &scoped_payload("wss://tenant-a.example", "bb22"), - Some("wss://tenant-a.example"), - Some("aa11"), + &tenant(Some("wss://tenant-a.example"), Some("aa11")), ) .unwrap_err(); assert!(error.contains("active identity changed"), "{error}"); @@ -400,9 +634,10 @@ mod tests { fn scoped_caller_with_an_unverifiable_payload_fails_closed() { let payload = serde_json::json!({}); let relay_error = - assert_payload_scope(&payload, Some("wss://tenant-a.example"), None).unwrap_err(); + assert_payload_scope(&payload, &tenant(Some("wss://tenant-a.example"), None)) + .unwrap_err(); assert!(relay_error.contains("no relay"), "{relay_error}"); - let signer_error = assert_payload_scope(&payload, None, Some("aa11")).unwrap_err(); + let signer_error = assert_payload_scope(&payload, &tenant(None, Some("aa11"))).unwrap_err(); assert!(signer_error.contains("no owner identity"), "{signer_error}"); } @@ -410,11 +645,10 @@ mod tests { fn unscoped_callers_deploy_any_payload() { assert_payload_scope( &scoped_payload("wss://anywhere.example", "cc33"), - None, - None, + &tenant(None, None), ) .unwrap(); - assert_payload_scope(&serde_json::json!({}), None, None).unwrap(); + assert_payload_scope(&serde_json::json!({}), &tenant(None, None)).unwrap(); } // ── apply_replay_floor: publish-first floor threading into the payload ── diff --git a/desktop/src-tauri/src/commands/agents/remote_session.rs b/desktop/src-tauri/src/commands/agents/remote_session.rs new file mode 100644 index 00000000000..2a06d307b46 --- /dev/null +++ b/desktop/src-tauri/src/commands/agents/remote_session.rs @@ -0,0 +1,202 @@ +//! Explicit lifecycle actions for a thread-sandbox agent's remote session: +//! end it, or recover it after it ended. Both run only on user action and +//! share the deploy path's per-agent lock, post-lock payload rebuild, tenant +//! scope assertion, and session binding. +//! +//! Errors carry the provider's refusal `code` when it sent one (see +//! [`ProviderFailure`]), so the frontend branches on the code, not the text. + +use tauri::{AppHandle, State}; + +use crate::{ + app_state::AppState, + managed_agents::{ + load_managed_agents, provider_ops, provider_stop, BackendKind, ManagedAgentSummary, + ProviderFailure, + }, +}; + +use super::provider_deploy::{ + deploy_to_provider, lock_provider_operations, prepare_provider_invocation, + resolve_record_provider_binary, InvocationScope, ProviderAction, ProviderInvocation, + RemoteRecoveryMode, RemoteSessionScope, +}; + +/// The `info.ops` entry a provider advertises when it can end a thread +/// session (`stop`) and recover an ended one (`provider_config.sandbox.recovery`). +const RECOVER_OP: &str = "recover"; + +/// The tenant a session action was offered for. Both parts are required: an +/// action on a remote session is never unscoped. +fn session_tenant<'a>( + expected_relay_url: &'a str, + expected_signer_pubkey: &'a str, + session: &'a RemoteSessionScope, + action: ProviderAction, +) -> Result, ProviderFailure> { + if expected_relay_url.trim().is_empty() || expected_signer_pubkey.trim().is_empty() { + return Err("remote session actions require the community and identity they were offered for; nothing was changed".into()); + } + Ok(InvocationScope { + expected_relay_url: Some(expected_relay_url), + expected_signer_pubkey: Some(expected_signer_pubkey), + session: Some(session), + action, + }) +} + +/// End the remote session for one channel thread with the provider's `stop` +/// operation. Returns the session state the provider reports: `ending`, +/// `ended`, or `absent`. The agent record is not changed: the provider owns +/// the session lifecycle, and recovery is a separate explicit action. +#[tauri::command] +pub async fn end_remote_agent_session( + pubkey: String, + session_scope: RemoteSessionScope, + expected_relay_url: String, + expected_signer_pubkey: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result { + end_session( + &app, + &state, + &pubkey, + &session_scope, + &expected_relay_url, + &expected_signer_pubkey, + ) + .await +} + +async fn end_session( + app: &AppHandle, + state: &AppState, + pubkey: &str, + session_scope: &RemoteSessionScope, + expected_relay_url: &str, + expected_signer_pubkey: &str, +) -> Result { + let scope = session_tenant( + expected_relay_url, + expected_signer_pubkey, + session_scope, + ProviderAction::End, + )?; + let _guard = lock_provider_operations(state, pubkey).await?; + let ProviderInvocation { + binary, + config, + agent_json, + .. + } = prepare_provider_invocation(app, state, pubkey, &scope)?; + tokio::task::spawn_blocking(move || provider_stop(&binary, &agent_json, &config)) + .await + .map_err(|e| format!("spawn_blocking failed: {e}"))? +} + +/// Recover the ended remote session for one channel thread, from its +/// checkpoint or fresh. The mode is sent for this deploy only and is never +/// persisted. Returns the updated agent, like `start_managed_agent`. +#[tauri::command] +pub async fn recover_remote_agent_session( + pubkey: String, + session_scope: RemoteSessionScope, + mode: RemoteRecoveryMode, + expected_relay_url: String, + expected_signer_pubkey: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result { + recover_session( + &app, + &state, + &pubkey, + &session_scope, + mode, + &expected_relay_url, + &expected_signer_pubkey, + ) + .await?; + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(&app)?; + let runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + let record = records + .iter() + .find(|r| r.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + Ok(super::summarize_from_disk(&app, record, &runtimes)?) +} + +/// The deploy half of [`recover_remote_agent_session`]. +async fn recover_session( + app: &AppHandle, + state: &AppState, + pubkey: &str, + session_scope: &RemoteSessionScope, + mode: RemoteRecoveryMode, + expected_relay_url: &str, + expected_signer_pubkey: &str, +) -> Result<(), ProviderFailure> { + let scope = session_tenant( + expected_relay_url, + expected_signer_pubkey, + session_scope, + ProviderAction::Recover(mode), + )?; + deploy_to_provider(app, state, pubkey, &scope, None).await +} + +/// Whether the agent's provider can end and recover thread sessions, which +/// it advertises as `recover` in `info.ops`. Runs only `info`. Local agents +/// and provider agents without thread workspaces return `false`. +#[tauri::command] +pub async fn remote_agent_session_supports_lifecycle( + pubkey: String, + app: AppHandle, + state: State<'_, AppState>, +) -> Result { + supports_lifecycle(&app, &state, &pubkey).await +} + +async fn supports_lifecycle( + app: &AppHandle, + state: &AppState, + pubkey: &str, +) -> Result { + let target = { + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(app)?; + let record = records + .iter() + .find(|r| r.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + match &record.backend { + BackendKind::Provider { id, .. } if record.backend.is_thread_sandbox() => { + Some((id.clone(), record.provider_binary_path.clone())) + } + _ => None, + } + }; + let Some((provider_id, cached_binary_path)) = target else { + return Ok(false); + }; + let binary = resolve_record_provider_binary(&provider_id, cached_binary_path.as_deref())?; + let ops = tokio::task::spawn_blocking(move || provider_ops(&binary)) + .await + .map_err(|e| format!("spawn_blocking failed: {e}"))??; + Ok(ops.iter().any(|op| op == "stop") && ops.iter().any(|op| op == RECOVER_OP)) +} + +#[cfg(all(test, unix))] +#[path = "remote_session_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/commands/agents/remote_session_tests.rs b/desktop/src-tauri/src/commands/agents/remote_session_tests.rs new file mode 100644 index 00000000000..15e9df0a9fe --- /dev/null +++ b/desktop/src-tauri/src/commands/agents/remote_session_tests.rs @@ -0,0 +1,174 @@ +use super::*; +use crate::commands::agents::{tests::*, workspace_owner_hex}; +use crate::managed_agents::save_managed_agents; +use serde_json::{json, Value}; +use std::{os::unix::fs::PermissionsExt, path::Path}; +use tauri::Manager; + +fn install_provider(home: &Path, ops: Value, response: Value) { + let path = home.join("bin/buzz-backend-kubernetes"); + let info = json!({"ok": true, "name": "kubernetes", "version": "1", + "protocol_version": 1, "description": "Fixture provider", "config_schema": {}, "ops": ops}); + std::fs::write(&path, format!( + "#!/bin/sh\nIFS= read -r request\ncase \"$request\" in\n*'\"op\":\"info\"'*) printf '%s\\n' '{info}' ;;\n*) printf '%s\\n' \"$request\" > '{}/request.json'\nprintf '%s\\n' '{response}' ;;\nesac\n", + home.display(), + )).unwrap(); + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap(); +} + +#[test] +fn lifecycle_actions_enforce_scope_and_preserve_provider_codes() { + const CHILD: &str = "BUZZ_REMOTE_SESSION_TEST_HOME"; + let Some(home) = std::env::var_os(CHILD) else { + return run_in_confined_child( + "commands::agents::remote_session::tests::lifecycle_actions_enforce_scope_and_preserve_provider_codes", + CHILD, + ); + }; + let home = Path::new(&home); + let app = confined_mock_app(home); + let state = app.state::(); + let owner = workspace_owner_hex(&state).unwrap(); + let config = json!({"sandbox": true, "namespace": "agents"}); + save_managed_agents( + app.handle(), + &[provider_record( + "thread-agent", + "kubernetes", + config.clone(), + )], + ) + .unwrap(); + let session: RemoteSessionScope = serde_json::from_value(json!({ + "channelId": "f9ee507e-d03c-4570-baa6-02951c38e1cd", "threadRoot": "a".repeat(64), + })) + .unwrap(); + let relay = "http://127.0.0.1:9"; + let capture = home.join("request.json"); + let end = |relay: &str, signer: &str| { + tauri::async_runtime::block_on(end_session( + app.handle(), + &state, + "thread-agent", + &session, + relay, + signer, + )) + }; + let recover = |mode, relay: &str, signer: &str| { + tauri::async_runtime::block_on(recover_session( + app.handle(), + &state, + "thread-agent", + &session, + mode, + relay, + signer, + )) + }; + install_provider( + home, + json!(["stop", "recover"]), + json!({"ok": true, "agent_id": "remote", "state": "ended"}), + ); + for (expected_relay, signer, message) in [ + ("", owner.as_str(), "require the community and identity"), + (relay, " ", "require the community and identity"), + ( + "http://other.invalid", + owner.as_str(), + "active community changed", + ), + (relay, "wrong-owner", "active identity changed"), + ] { + for error in [ + end(expected_relay, signer).unwrap_err(), + recover(RemoteRecoveryMode::Fresh, expected_relay, signer).unwrap_err(), + ] { + assert!(error.message.contains(message), "{error}"); + assert!(error.message.contains("nothing was changed"), "{error}"); + } + assert!( + !capture.exists(), + "a refused action must not invoke the provider" + ); + } + for stop_state in ["ending", "ended", "absent"] { + install_provider( + home, + json!(["stop", "recover"]), + json!({"ok": true, "agent_id": "remote", "state": stop_state}), + ); + assert_eq!(end(relay, &owner).unwrap(), stop_state); + let request: Value = + serde_json::from_str(&std::fs::read_to_string(&capture).unwrap()).unwrap(); + assert_eq!(request["op"], "stop"); + assert_eq!( + request["provider_config"]["identity_policy"]["owner_pubkey"], + owner + ); + assert_eq!( + request["provider_config"]["sandbox"]["thread_root"], + "a".repeat(64) + ); + assert!(request["provider_config"]["sandbox"] + .get("recovery") + .is_none()); + } + for mode in [RemoteRecoveryMode::Fresh, RemoteRecoveryMode::Checkpoint] { + install_provider( + home, + json!(["stop", "recover"]), + json!({"ok": true, "agent_id": "remote"}), + ); + recover(mode, relay, &owner).unwrap(); + let request: Value = + serde_json::from_str(&std::fs::read_to_string(&capture).unwrap()).unwrap(); + assert_eq!(request["op"], "deploy"); + assert_eq!( + request["provider_config"]["sandbox"]["recovery"]["mode"], + json!(mode) + ); + assert_eq!(request["agent"]["launch"]["owner_pubkey"], owner); + let records = load_managed_agents(app.handle()).unwrap(); + assert!( + matches!(&records[0].backend, BackendKind::Provider { config: saved, .. } if saved == &config) + ); + } + for code in [Some("no_checkpoint"), None] { + let mut response = json!({"ok": false, "error": "fixture refusal"}); + if let Some(code) = code { + response["code"] = json!(code); + } + install_provider(home, json!(["stop", "recover"]), response); + for error in [ + end(relay, &owner).unwrap_err(), + recover(RemoteRecoveryMode::Checkpoint, relay, &owner).unwrap_err(), + ] { + assert_eq!(error.code.as_deref(), code); + assert!(error.message.contains("fixture refusal"), "{error}"); + let serialized = serde_json::to_value(&error).unwrap(); + if let Some(code) = code { + assert_eq!(serialized["code"], code); + } else { + assert_eq!(serialized, error.message); + } + } + } + for (ops, expected) in [ + (json!(["info", "deploy", "stop"]), false), + (json!(["info", "deploy", "recover"]), false), + (json!(["info", "deploy", "stop", "recover"]), true), + ] { + install_provider(home, ops, json!({"ok": true})); + assert_eq!( + tauri::async_runtime::block_on(supports_lifecycle( + app.handle(), + &state, + "thread-agent" + )) + .unwrap(), + expected + ); + } +} diff --git a/desktop/src-tauri/src/commands/agents_tests.rs b/desktop/src-tauri/src/commands/agents_tests.rs index 190ea79e7c5..fff809aeb1e 100644 --- a/desktop/src-tauri/src/commands/agents_tests.rs +++ b/desktop/src-tauri/src/commands/agents_tests.rs @@ -998,28 +998,24 @@ fn thread_sandbox_deploy_in_confined_child(home: &std::path::Path) { "threadRoot": "a".repeat(64), })) .unwrap(); - // Callers' pre-lock snapshots are stale on purpose: deploy must rebuild. - let stale_payload = serde_json::json!({"launch": {"owner_pubkey": "stale-owner"}}); let deploy = |pubkey: &str, scope: Option<&provider_deploy::RemoteSessionScope>| { + let scope = provider_deploy::InvocationScope { + session: scope, + ..Default::default() + }; tauri::async_runtime::block_on(deploy_to_provider( app.handle(), &state, pubkey, - "kubernetes", - &serde_json::json!({}), - stale_payload.clone(), + &scope, None, - None, - None, - None, - scope, )) }; // No provider binary exists yet, so this error proves the refusal happens // before binary resolution. let refused = deploy("thread-agent", None).unwrap_err(); - assert!(refused.contains("Mention this agent"), "{refused}"); + assert!(refused.message.contains("Mention this agent"), "{refused}"); let capture = home.join("deploy.json"); install_fake_provider(&home.join("bin"), "kubernetes", &capture); diff --git a/desktop/src-tauri/src/commands/personas/inbound.rs b/desktop/src-tauri/src/commands/personas/inbound.rs index 0ee9ee170d0..0b29750b533 100644 --- a/desktop/src-tauri/src/commands/personas/inbound.rs +++ b/desktop/src-tauri/src/commands/personas/inbound.rs @@ -30,9 +30,6 @@ enum InboundRuntimeRefresh { }, Provider { pubkey: String, - provider_id: String, - config: serde_json::Value, - cached_binary_path: Option, agent_json: Result, }, } @@ -104,40 +101,19 @@ pub async fn reconcile_inbound_persona_event( ) })?; } - Some(InboundRuntimeRefresh::Provider { - pubkey, - provider_id, - config, - cached_binary_path, - agent_json, - }) => { + Some(InboundRuntimeRefresh::Provider { pubkey, agent_json }) => { let state = app.state::(); - let agent_json = match agent_json { - Ok(agent_json) => agent_json, - Err(error) => { - let message = format!( - "Inbound agent access was saved, but its provider deployment could not be refreshed safely: {error}" - ); - super::super::agents::provider_access::persist_failure( - &app, &state, &pubkey, &message, - )?; - let _ = app.emit("agents-data-changed", ()); - return Err(message); - } - }; - super::super::agents::deploy_to_provider( - &app, - &state, - &pubkey, - &provider_id, - &config, - agent_json, - cached_binary_path.as_deref(), - None, - None, - None, - None, - ) + if let Err(error) = agent_json { + let message = format!( + "Inbound agent access was saved, but its provider deployment could not be refreshed safely: {error}" + ); + super::super::agents::provider_access::persist_failure( + &app, &state, &pubkey, &message, + )?; + let _ = app.emit("agents-data-changed", ()); + return Err(message); + } + super::super::agents::deploy_to_provider(&app, &state, &pubkey, &Default::default(), None) .await .map_err(|error| { format!( @@ -376,7 +352,7 @@ fn reconcile_inbound_persona_event_blocking( }); } } - crate::managed_agents::BackendKind::Provider { id, config } + crate::managed_agents::BackendKind::Provider { .. } if record.backend_agent_id.is_some() && !record.backend.is_thread_sandbox() => { @@ -387,9 +363,6 @@ fn reconcile_inbound_persona_event_blocking( record.provider_policy_pending = true; runtime_refresh = Some(InboundRuntimeRefresh::Provider { pubkey: d_tag.clone(), - provider_id: id.clone(), - config: config.clone(), - cached_binary_path: record.provider_binary_path.clone(), agent_json: super::super::agents::build_deploy_payload( &app, &state, record, ), diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 4497eb78e56..69efa4baa8b 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -714,6 +714,9 @@ pub fn run() { create_managed_agent, start_managed_agent, stop_managed_agent, + end_remote_agent_session, + recover_remote_agent_session, + remote_agent_session_supports_lifecycle, set_agent_managed_profiles, set_managed_agent_start_on_app_launch, set_managed_agent_auto_restart, diff --git a/desktop/src-tauri/src/managed_agents/backend.rs b/desktop/src-tauri/src/managed_agents/backend.rs index 1a18637c74e..4a68ddd8a2a 100644 --- a/desktop/src-tauri/src/managed_agents/backend.rs +++ b/desktop/src-tauri/src/managed_agents/backend.rs @@ -12,6 +12,74 @@ const STDERR_CAP: usize = 65536; const STDOUT_CAP: usize = 1_048_576; // 1 MB const PROVIDER_PROTOCOL_VERSION: u64 = 1; +/// A failed provider operation, or a command that ended in one. +/// +/// `code` is the provider's stable refusal code (`docs/remote-agents.md` +/// §Sandbox error codes). It is absent for transient failures, desktop-side +/// failures, and providers that predate codes. As a Tauri command error it +/// serializes as the plain message when there is no code, and as +/// `{"message": ..., "code": ...}` when there is one; the frontend's +/// `invokeTauri` accepts both shapes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProviderFailure { + pub message: String, + pub code: Option, +} + +impl From for ProviderFailure { + fn from(message: String) -> Self { + Self { + message, + code: None, + } + } +} + +impl From<&str> for ProviderFailure { + fn from(message: &str) -> Self { + message.to_string().into() + } +} + +impl From for String { + fn from(failure: ProviderFailure) -> Self { + failure.message + } +} + +impl std::fmt::Display for ProviderFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} + +impl serde::Serialize for ProviderFailure { + fn serialize(&self, serializer: S) -> Result { + use serde::ser::SerializeStruct; + let Some(code) = &self.code else { + return serializer.serialize_str(&self.message); + }; + let mut failure = serializer.serialize_struct("ProviderFailure", 2)?; + failure.serialize_field("message", &self.message)?; + failure.serialize_field("code", code)?; + failure.end() + } +} + +/// A provider's refusal code, if it is a well-formed snake_case token. +/// Anything else is dropped rather than passed to the frontend. +fn refusal_code(response: &serde_json::Value) -> Option { + response + .get("code") + .and_then(serde_json::Value::as_str) + .filter(|code| { + (1..=64).contains(&code.len()) + && code.starts_with(|c: char| c.is_ascii_lowercase()) + && code.bytes().all(|b| b.is_ascii_lowercase() || b == b'_') + }) + .map(String::from) +} + fn validate_provider_info(info: &serde_json::Value) -> Result<(), String> { let object = info .as_object() @@ -52,6 +120,7 @@ fn validate_provider_info(info: &serde_json::Value) -> Result<(), String> { "name", "version", "protocol_version", + "ops", "description", "config_schema", ]; @@ -63,7 +132,37 @@ fn validate_provider_info(info: &serde_json::Value) -> Result<(), String> { "provider info response contains unknown field {field}" )); } - Ok(()) + advertised_ops(info).map(drop) +} + +/// Operations a provider advertises in `info.ops`. The field is additive +/// within protocol v1: a provider that omits it answers only `info` and +/// `deploy`, so it has no `stop` and no session recovery. +pub(crate) fn advertised_ops(info: &serde_json::Value) -> Result, String> { + const MAX_OPS: usize = 32; + let Some(ops) = info.get("ops") else { + return Ok(vec!["info".to_string(), "deploy".to_string()]); + }; + ops.as_array() + .filter(|ops| ops.len() <= MAX_OPS) + .ok_or_else(|| format!("provider info ops must be an array of at most {MAX_OPS} strings"))? + .iter() + .map(|op| { + op.as_str() + .filter(|op| !op.is_empty()) + .map(String::from) + .ok_or_else(|| "provider info ops must contain non-empty strings".to_string()) + }) + .collect() +} + +/// The terminal state a `stop` response reports for the thread's session. +fn stop_response_state(response: &serde_json::Value) -> Result { + match response.get("state").and_then(serde_json::Value::as_str) { + Some(state @ ("ending" | "ended" | "absent")) => Ok(state.to_string()), + Some(state) => Err(format!("stop response has unknown state {state:?}")), + None => Err("stop response missing state".to_string()), + } } /// Invoke a provider binary: write JSON to stdin, read JSON from stdout. @@ -78,7 +177,7 @@ pub fn invoke_provider( binary: &Path, request: &serde_json::Value, timeout: Duration, -) -> Result { +) -> Result { let request_bytes = format!( "{}\n", serde_json::to_string(request).map_err(|e| e.to_string())? @@ -155,7 +254,7 @@ pub fn invoke_provider( if let Err(e) = stdin_result { let _ = child.kill(); let _ = child.wait(); - return Err(format!("stdin write failed: {e}")); + return Err(format!("stdin write failed: {e}").into()); } // Poll try_wait with a deadline, collecting stdout chunks and draining @@ -189,14 +288,14 @@ pub fn invoke_provider( if std::time::Instant::now() >= deadline { let _ = child.kill(); let _ = child.wait(); - return Err(format!("provider timed out after {timeout_secs}s")); + return Err(format!("provider timed out after {timeout_secs}s").into()); } std::thread::sleep(Duration::from_millis(50)); } Err(e) => { let _ = child.kill(); let _ = child.wait(); - return Err(format!("wait error: {e}")); + return Err(format!("wait error: {e}").into()); } } }; @@ -253,11 +352,9 @@ pub fn invoke_provider( if !exited_ok { let stderr_snippet = &stderr_redacted[..stderr_redacted.len().min(4096)]; if stderr_snippet.is_empty() { - return Err(format!("provider failed ({exit_info}, empty stderr)")); + return Err(format!("provider failed ({exit_info}, empty stderr)").into()); } else { - return Err(format!( - "provider failed ({exit_info}). stderr: {stderr_snippet}" - )); + return Err(format!("provider failed ({exit_info}). stderr: {stderr_snippet}").into()); } } @@ -282,7 +379,10 @@ pub fn invoke_provider( if response.get("ok").and_then(|v| v.as_bool()) == Some(false) { let error = response["error"].as_str().unwrap_or("unknown error"); - return Err(redact_secrets_with(error, &env_secret_refs)); + return Err(ProviderFailure { + message: redact_secrets_with(error, &env_secret_refs), + code: refusal_code(&response), + }); } Ok(response) @@ -505,20 +605,33 @@ fn stage_provider( )) } +/// Run `info` on a staged provider and return the operations it advertises. +fn negotiate_staged_provider(staged: &Path) -> Result, String> { + let info_request = serde_json::json!({ + "op": "info", + "request_id": uuid::Uuid::new_v4().to_string(), + }); + let info = invoke_provider(staged, &info_request, Duration::from_secs(10))?; + validate_provider_info(&info)?; + advertised_ops(&info) +} + +/// Operations a provider advertises. Runs only `info`, which carries no +/// agent secret, on a staged copy of the binary. +pub fn provider_ops(binary: &Path) -> Result, String> { + let (_directory, staged, _digest, _execution_guard) = stage_provider(binary)?; + negotiate_staged_provider(&staged) +} + /// Deploy through one immutable staged copy: negotiate protocol v1 before the /// secret-bearing request, then invoke deploy on those exact same bytes. pub fn provider_deploy( binary: &Path, agent: &serde_json::Value, provider_config: &serde_json::Value, -) -> Result { +) -> Result { let (_directory, staged, _digest, _execution_guard) = stage_provider(binary)?; - let info_request = serde_json::json!({ - "op": "info", - "request_id": uuid::Uuid::new_v4().to_string(), - }); - let info = invoke_provider(&staged, &info_request, Duration::from_secs(10))?; - validate_provider_info(&info)?; + negotiate_staged_provider(&staged)?; let request = serde_json::json!({ "op": "deploy", @@ -530,7 +643,35 @@ pub fn provider_deploy( resp["agent_id"] .as_str() .map(String::from) - .ok_or_else(|| "deploy response missing agent_id".to_string()) + .ok_or_else(|| "deploy response missing agent_id".into()) +} + +/// End a remote session through one immutable staged copy. The `stop` +/// request carries the same secret-bearing payload as `deploy`, so it is sent +/// only after the same bytes negotiated protocol v1 and advertised `stop`. +/// Returns the session state: `ending`, `ended`, or `absent`. +pub fn provider_stop( + binary: &Path, + agent: &serde_json::Value, + provider_config: &serde_json::Value, +) -> Result { + let (_directory, staged, _digest, _execution_guard) = stage_provider(binary)?; + if !negotiate_staged_provider(&staged)? + .iter() + .any(|op| op == "stop") + { + return Err( + "this provider cannot end remote sessions; update the provider and try again".into(), + ); + } + let request = serde_json::json!({ + "op": "stop", + "request_id": uuid::Uuid::new_v4().to_string(), + "agent": agent, + "provider_config": provider_config, + }); + let response = invoke_provider(&staged, &request, Duration::from_secs(120))?; + Ok(stop_response_state(&response)?) } /// Bound public provider configuration and reject secret-like keys at every depth. diff --git a/desktop/src-tauri/src/managed_agents/backend_tests.rs b/desktop/src-tauri/src/managed_agents/backend_tests.rs index 5dde2279ba3..956d2397b77 100644 --- a/desktop/src-tauri/src/managed_agents/backend_tests.rs +++ b/desktop/src-tauri/src/managed_agents/backend_tests.rs @@ -346,9 +346,9 @@ esac"#, &serde_json::json!({}), ) .unwrap_err(); - assert!(error.contains("protocol version 2"), "{error}"); + assert!(error.to_string().contains("protocol version 2"), "{error}"); assert!(!marker.exists()); - assert!(!error.contains("nsec1must-not-cross")); + assert!(!error.to_string().contains("nsec1must-not-cross")); } #[cfg(unix)] @@ -365,7 +365,9 @@ printf '%s\n' '{"ok":true,"version":"1.0.0"}'"#, let error = provider_deploy(&provider, &serde_json::json!({}), &serde_json::json!({})).unwrap_err(); assert!( - error.contains("missing integer protocol_version"), + error + .to_string() + .contains("missing integer protocol_version"), "{error}" ); } @@ -398,6 +400,127 @@ fn provider_info_requires_the_complete_flat_wire_shape() { .contains("unknown field provider")); } +fn info_with_ops(ops: Option) -> serde_json::Value { + let mut info = serde_json::json!({ + "ok": true, + "name": "kubernetes", + "version": "1.0.0", + "protocol_version": 1, + "description": "Kubernetes provider", + "config_schema": {} + }); + if let Some(ops) = ops { + info["ops"] = ops; + } + info +} + +#[test] +fn provider_info_ops_are_parsed_and_default_to_deploy_only() { + let advertised = info_with_ops(Some(serde_json::json!(["info", "deploy", "stop"]))); + assert!(validate_provider_info(&advertised).is_ok()); + assert_eq!( + advertised_ops(&advertised).unwrap(), + ["info", "deploy", "stop"] + ); + + // Providers predating `ops` keep working and have no stop. + let legacy = info_with_ops(None); + assert!(validate_provider_info(&legacy).is_ok()); + assert_eq!(advertised_ops(&legacy).unwrap(), ["info", "deploy"]); + + for malformed in [ + serde_json::json!("stop"), + serde_json::json!({"stop": true}), + serde_json::json!(["info", 1]), + serde_json::json!(["info", ""]), + serde_json::json!(vec!["stop"; 33]), + ] { + let info = info_with_ops(Some(malformed.clone())); + let error = validate_provider_info(&info).unwrap_err(); + assert!(error.to_string().contains("ops"), "{malformed}: {error}"); + } +} + +#[test] +fn stop_response_state_accepts_only_the_wire_states() { + for state in ["ending", "ended", "absent"] { + let response = serde_json::json!({"ok": true, "agent_id": "a", "state": state}); + assert_eq!(stop_response_state(&response).unwrap(), state); + } + let unknown = serde_json::json!({"ok": true, "agent_id": "a", "state": "running"}); + assert!(stop_response_state(&unknown) + .unwrap_err() + .contains("unknown state")); + let missing = serde_json::json!({"ok": true, "agent_id": "a"}); + assert!(stop_response_state(&missing) + .unwrap_err() + .contains("missing state")); +} + +#[cfg(unix)] +fn stop_test_provider(path: &Path, ops: &str, marker: &Path) { + let body = format!( + r#"read request +case "$request" in + *\"op\":\"info\"*) printf '%s\n' '{{"ok":true,"name":"test","version":"1.0.0","protocol_version":1,{ops}"description":"test provider","config_schema":{{}}}}' ;; + *\"op\":\"stop\"*) touch '{}'; printf '%s\n' '{{"ok":true,"agent_id":"remote-1","state":"ending"}}' ;; +esac"#, + marker.display() + ); + write_test_provider(path, &body); +} + +#[cfg(unix)] +#[test] +fn provider_stop_sends_stop_only_to_a_provider_that_advertises_it() { + let directory = tempfile::tempdir().unwrap(); + let provider = directory.path().join("provider"); + let marker = directory.path().join("stop-received"); + let agent = serde_json::json!({"private_key_nsec": "nsec1must-not-cross"}); + + stop_test_provider(&provider, "", &marker); + let error = provider_stop(&provider, &agent, &serde_json::json!({})).unwrap_err(); + assert!( + error.to_string().contains("cannot end remote sessions"), + "{error}" + ); + assert!(!marker.exists(), "the secret-bearing stop must not be sent"); + + stop_test_provider(&provider, r#""ops":["info","deploy"],"#, &marker); + assert!(provider_stop(&provider, &agent, &serde_json::json!({})).is_err()); + assert!(!marker.exists(), "the secret-bearing stop must not be sent"); + + stop_test_provider(&provider, r#""ops":["info","deploy","stop"],"#, &marker); + assert_eq!(provider_ops(&provider).unwrap(), ["info", "deploy", "stop"]); + assert!(!marker.exists(), "provider_ops runs only info"); + assert_eq!( + provider_stop(&provider, &agent, &serde_json::json!({})).unwrap(), + "ending" + ); + assert!(marker.exists()); +} + +#[cfg(unix)] +#[test] +fn provider_stop_errors_are_redacted_like_deploy() { + let directory = tempfile::tempdir().unwrap(); + let provider = directory.path().join("provider"); + write_test_provider( + &provider, + r#"read request +case "$request" in + *\"op\":\"info\"*) printf '%s\n' '{"ok":true,"name":"test","version":"1.0.0","protocol_version":1,"ops":["info","deploy","stop"],"description":"test provider","config_schema":{}}' ;; + *\"op\":\"stop\"*) printf '%s\n' '{"ok":false,"error":"refused env-secret-value and nsec1leak"}' ;; +esac"#, + ); + let agent = serde_json::json!({"env_vars": {"API": "env-secret-value"}}); + let error = provider_stop(&provider, &agent, &serde_json::json!({})).unwrap_err(); + assert!(!error.to_string().contains("env-secret-value"), "{error}"); + assert!(!error.to_string().contains("nsec1leak"), "{error}"); + assert!(error.to_string().contains("[REDACTED]"), "{error}"); +} + #[test] fn validate_provider_config_rejects_secret_key() { let cfg = serde_json::json!({"api_key": "val"}); diff --git a/desktop/src/shared/api/tauriManagedAgents.ts b/desktop/src/shared/api/tauriManagedAgents.ts index d9f6c1525bd..a2e044c7edf 100644 --- a/desktop/src/shared/api/tauriManagedAgents.ts +++ b/desktop/src/shared/api/tauriManagedAgents.ts @@ -38,6 +38,64 @@ export async function startManagedAgent( return fromRawManagedAgent(response); } +/** How the user chose to recover an ended thread session. */ +export type RemoteRecoveryMode = "checkpoint" | "fresh"; + +/** Tenant scope a recovery prompt captured; the backend fails closed on change. */ +export type RemoteSessionTenantScope = { + expectedRelayUrl: string; + expectedSignerPubkey: string; +}; + +/** Session state the provider reports after `stop`. */ +export type RemoteSessionStopState = "ending" | "ended" | "absent"; + +/** Ends a thread's remote session with the provider's `stop` operation. */ +export async function endRemoteAgentSession( + pubkey: string, + sessionScope: RemoteAgentSession, + scope: RemoteSessionTenantScope, +): Promise { + return invokeTauri("end_remote_agent_session", { + pubkey, + sessionScope, + expectedRelayUrl: scope.expectedRelayUrl, + expectedSignerPubkey: scope.expectedSignerPubkey, + }); +} + +/** + * Recovers a thread's ended remote session. Call only from an explicit user + * action: the mode applies to this one deploy and is never stored. + */ +export async function recoverRemoteAgentSession( + pubkey: string, + sessionScope: RemoteAgentSession, + mode: RemoteRecoveryMode, + scope: RemoteSessionTenantScope, +): Promise { + const response = await invokeTauri( + "recover_remote_agent_session", + { + pubkey, + sessionScope, + mode, + expectedRelayUrl: scope.expectedRelayUrl, + expectedSignerPubkey: scope.expectedSignerPubkey, + }, + ); + return fromRawManagedAgent(response); +} + +/** Whether the agent's provider advertises `recover`, so sessions can be ended and recovered. */ +export async function remoteAgentSessionSupportsLifecycle( + pubkey: string, +): Promise { + return invokeTauri("remote_agent_session_supports_lifecycle", { + pubkey, + }); +} + export async function stopManagedAgent(pubkey: string): Promise { const response = await invokeTauri("stop_managed_agent", { pubkey, diff --git a/docs/remote-agents.md b/docs/remote-agents.md index c5f96395da0..508efe963a9 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -401,12 +401,12 @@ response: {"ok": true, "name": str, "version": str, timeout: 10s ``` -**Planned (added in a later change):** an `ops: [str]` field that lists the -operations the provider answers. It will be additive within protocol -version 1: a provider that omits it answers only `info` and `deploy`. The -desktop will show Sandbox session controls (§Sandbox lifecycle) only when -`ops` contains `stop`. The provider does not advertise `ops` yet, so that -desktops that predate the field keep accepting `info`. +The additive `ops: [str]` field lists provider capabilities within protocol +version 1. Providers that omit it support only `info` and `deploy`. +The Kubernetes provider advertises `info`, `deploy`, `stop`, and `recover`. +`recover` means that `deploy` accepts `provider_config.sandbox.recovery`; +it is a capability marker, not a separate request operation. The desktop +shows session controls only when `ops` contains `recover`. `version` is the provider's *software* version — useful in error reports, useless for compatibility. `protocol_version` (this document: `1`) is the @@ -1649,6 +1649,11 @@ or reused, and a new refusal gets a new code. They are defined in | `recovery_interrupted` | A recovery deleted the tombstone but did not finish. | explicit recovery | | `recovery_mode_conflict` | A recovery in the other mode already created the replacement. | retry the same mode, or end the session first | +The desktop preserves provider refusals as `{ "message": str, "code": str }` +Tauri command errors. Errors without a code remain strings for compatibility +with older providers. Session end and recovery commands require the captured +relay URL and signer identity, and verify both after acquiring the agent lock. + The `session_ended` text names the ended reason only when it is one of the spellings in §Sandbox lifecycle; any other annotation value is not repeated. Transient and configuration failures (apiserver errors, timeouts) carry no