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