//! Generic immutable archive installation and Computer-local stdio MCP calls. use axum::{ Json, extract::{Path, State}, http::StatusCode, }; use lazyboy_control::{ AdapterContext, CommandRequest, ToolManifest, resolve_bot_workspace_path, validate_manifest, }; use serde::Deserialize; use serde_json::{Value, json}; use uuid::Uuid; use crate::{ computer, db::{Actor, parse_mode}, state::AppState, tools::ToolCtx, }; type ApiError = (StatusCode, Json); fn failure(message: impl ToString) -> ApiError { ( StatusCode::BAD_REQUEST, Json(json!({"message":message.to_string()})), ) } #[derive(Deserialize)] #[serde(deny_unknown_fields)] pub struct InstallRequest { manifest: ToolManifest, #[serde(rename = "artifactPath")] artifact_path: String, #[serde(default)] reauthorize: bool, } pub async fn install( State(state): State, actor: Actor, Path(bot_id): Path, Json(mut body): Json, ) -> Result, ApiError> { if !crate::tool_install::tool_install_enabled() { return Err(failure("tool install disabled")); } body.manifest.sha256 = body.manifest.sha256.to_ascii_lowercase(); validate_manifest(&body.manifest) .map_err(|error| failure(format!("invalid manifest: {error:?}")))?; state .db .get_bot(&actor, &bot_id) .await .map_err(failure)? .ok_or_else(|| failure("bot not found"))?; computer::boot_for(&state, &actor, &bot_id, false) .await .map_err(failure)?; let bot = state .db .get_bot(&actor, &bot_id) .await .map_err(failure)? .ok_or_else(|| failure("bot not found"))?; let computer_id = bot.computer_id.ok_or_else(|| failure("Computer missing"))?; let computer = state .db .get_computer(&computer_id) .await .map_err(failure)? .ok_or_else(|| failure("Computer missing"))?; let reference = computer::computer_ref(&computer).ok_or_else(|| failure("Computer unavailable"))?; let artifact = resolve_bot_workspace_path(parse_mode(&computer.scope), &bot_id, &body.artifact_path) .map_err(failure)?; let manifest = serde_json::to_value(&body.manifest).map_err(failure)?; let mut transaction = state.pool().begin().await.map_err(failure)?; sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('package-install:' || $1,0))") .bind(format!("{computer_id}:{}", body.manifest.id)) .execute(&mut *transaction) .await .map_err(failure)?; let pending: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM package_transitions WHERE bot_id=$1 AND package_id=$2 AND status IN ('pending','cancelling'))") .bind(&bot_id).bind(&body.manifest.id).fetch_one(&mut *transaction).await.map_err(failure)?; if pending { return Err(failure( "version transition pending; resume or revoke it before installing", )); } let package_id: String = sqlx::query_scalar( "INSERT INTO tool_packages(id,computer_id,package_id,version,sha256,status,manifest) VALUES ($1,$2,$3,$4,$5,'installing',$6) ON CONFLICT(computer_id,package_id,version) DO UPDATE SET status=tool_packages.status RETURNING id", ) .bind(Uuid::new_v4().to_string()) .bind(&computer_id) .bind(&body.manifest.id) .bind(&body.manifest.version) .bind(&body.manifest.sha256) .bind(&manifest) .fetch_one(&mut *transaction) .await .map_err(failure)?; let stored: (String, Option) = sqlx::query_as("SELECT sha256,manifest FROM tool_packages WHERE id=$1") .bind(&package_id) .fetch_one(&mut *transaction) .await .map_err(failure)?; if stored.0 != body.manifest.sha256 || stored.1.as_ref() != Some(&manifest) { return Err(failure("immutable package version conflict")); } // Lock live rows too: a concurrent revoke must not turn a normal retry // into an implicitly authorized new binding between this check and INSERT. let previous: Vec<(String, String)> = sqlx::query_as("SELECT id,status FROM tool_bindings WHERE package_row_id=$1 AND bot_id=$2 ORDER BY created_at FOR UPDATE") .bind(&package_id).bind(&bot_id).fetch_all(&mut *transaction).await.map_err(failure)?; let has_live = previous .iter() .any(|(_, status)| matches!(status.as_str(), "installing" | "ready")); if !has_live && !previous.is_empty() { if !body.reauthorize { return Err(failure( "binding revoked; explicit new authorization required", )); } for (old_id, _) in &previous { revoke_runtime(&state, &actor, &bot_id, old_id) .await .map_err(failure)?; } } let binding_id: String = sqlx::query_scalar( "INSERT INTO tool_bindings(id,package_row_id,bot_id,status) VALUES ($1,$2,$3,'installing') ON CONFLICT(package_row_id,bot_id) WHERE status IN ('installing','ready','switching','retiring') DO UPDATE SET status=tool_bindings.status RETURNING id", ) .bind(Uuid::new_v4().to_string()) .bind(&package_id) .bind(&bot_id) .fetch_one(&mut *transaction) .await .map_err(failure)?; let status: String = sqlx::query_scalar("SELECT status FROM tool_bindings WHERE id=$1") .bind(&binding_id) .fetch_one(&mut *transaction) .await .map_err(failure)?; if matches!(status.as_str(), "revoked" | "removed") { return Err(failure( "binding revoked; explicit new authorization required", )); } transaction.commit().await.map_err(failure)?; let adapter = AdapterContext { bot_id: Some(bot_id.clone()), computer_generation: Some(computer.generation), operation_id: format!("install:{binding_id}"), space_id: actor.space_id.clone(), user_id: actor.user_id.clone(), ..Default::default() }; let installed = state.sandbox.execute(&reference, CommandRequest { job_op: Some("package_install".into()), stdin: Some(json!({"manifest":manifest,"artifact_relative":artifact,"binding_id":binding_id}).to_string()), ..Default::default() }, &adapter).await.map_err(|_| failure("Computer installation incomplete; the recorded intent is retained"))?; if installed.code != 0 { return Err(failure("Computer installation incomplete")); } let discovery = state .sandbox .execute( &reference, CommandRequest { job_op: Some("package_call".into()), stdin: Some(json!({"binding_id":binding_id,"method":"tools/list"}).to_string()), ..Default::default() }, &adapter, ) .await .map_err(|_| failure("MCP discovery failed; binding is not ready"))?; if discovery.code != 0 { return Err(failure("MCP discovery failed")); } let tools: Value = serde_json::from_str(&discovery.stdout) .map_err(|_| failure("invalid MCP discovery response"))?; if !tools.is_array() { return Err(failure("invalid MCP tool list")); } let mut transaction = state.pool().begin().await.map_err(failure)?; sqlx::query("UPDATE tool_packages SET status='installed' WHERE id=$1") .bind(&package_id) .execute(&mut *transaction) .await .map_err(failure)?; let ready = sqlx::query("UPDATE tool_bindings b SET status='ready' FROM bots bot WHERE b.id=$1 AND bot.id=b.bot_id AND bot.computer_id=$2 AND b.status IN ('installing','ready')") .bind(&binding_id).bind(&computer_id).execute(&mut *transaction).await.map_err(failure)?; if ready.rows_affected() != 1 { return Err(failure("assignment or binding changed during installation")); } transaction.commit().await.map_err(failure)?; Ok(Json( json!({"bindingId":binding_id,"packageId":body.manifest.id,"version":body.manifest.version,"tools":tools,"executionLocation":"assigned_computer"}), )) } pub async fn any_bound( pool: &sqlx::PgPool, actor: &Actor, bot_id: &str, ) -> Result { sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM tool_bindings b JOIN tool_packages p ON p.id=b.package_row_id JOIN bots bot ON bot.id=b.bot_id WHERE bot.id=$1 AND bot.space_id=$2 AND bot.user_id=$3 AND p.computer_id=bot.computer_id AND b.status='ready' AND p.status='installed')") .bind(bot_id).bind(&actor.space_id).bind(&actor.user_id).fetch_one(pool).await } pub struct CallError { pub code: &'static str, pub message: &'static str, } fn rejected(message: &'static str) -> CallError { CallError { code: "MCP_REJECTED", message, } } fn unknown(message: &'static str) -> CallError { CallError { code: "UNKNOWN_EFFECT", message, } } pub async fn call(ctx: &ToolCtx, args: &Value) -> Result { call_on_computer( &ctx.pool, &ctx.actor, &ctx.bot_id, ctx.sandbox.as_ref(), &ctx.computer_ref(), &ctx.adapter(), args, ) .await } async fn call_on_computer( pool: &sqlx::PgPool, actor: &Actor, bot_id: &str, sandbox: &dyn lazyboy_control::SandboxProvider, computer_ref: &lazyboy_control::ComputerRef, adapter: &AdapterContext, args: &Value, ) -> Result { let rows: Vec<(String, String, String, String)> = sqlx::query_as( "SELECT b.id,p.package_id,p.version,p.sha256 FROM tool_bindings b JOIN tool_packages p ON p.id=b.package_row_id JOIN bots bot ON bot.id=b.bot_id WHERE bot.id=$1 AND bot.space_id=$2 AND bot.user_id=$3 AND p.computer_id=bot.computer_id AND b.status='ready' AND p.status='installed' AND p.manifest IS NOT NULL") .bind(bot_id).bind(&actor.space_id).bind(&actor.user_id).fetch_all(pool).await.map_err(|_| rejected("package binding lookup failed"))?; let action = args["action"].as_str().unwrap_or("call"); if action == "list" { return Ok( json!({"bindings":rows.into_iter().map(|(id,package,version,_)| json!({"bindingId":id,"packageId":package,"version":version})).collect::>()}), ); } if !matches!(action, "discover" | "call") { return Err(rejected("unsupported installed MCP action")); } let binding = args["bindingId"] .as_str() .ok_or_else(|| rejected("bindingId required; list installed packages first"))?; let bound = rows .iter() .find(|row| row.0 == binding) .ok_or_else(|| rejected("binding is not authorized on this Computer"))?; if action == "call" && (args["tool"] .as_str() .is_none_or(|name| name.trim().is_empty()) || args .get("arguments") .is_some_and(|value| !value.is_object())) { return Err(rejected("call requires a tool name and object arguments")); } let payload = json!({"binding_id":binding,"expected_sha256":bound.3,"expected_package_id":bound.1,"expected_version":bound.2,"method":if action == "discover" { "tools/list" } else { "tools/call" }, "name":args["tool"],"arguments":args.get("arguments").cloned().unwrap_or(json!({}))}); let result = sandbox .execute( computer_ref, CommandRequest { job_op: Some("package_call".into()), stdin: Some(payload.to_string()), ..Default::default() }, adapter, ) .await .map_err(|_| unknown("installed MCP failed; effects may be unknown"))?; if result.code != 0 { return Err(unknown("installed MCP failed; effects may be unknown")); } let value: Value = serde_json::from_str(&result.stdout).map_err(|_| unknown("invalid MCP response"))?; if value["isError"].as_bool() == Some(true) { return Err(unknown("MCP reported an error; effects may be unknown")); } Ok(value) } /// Re-read schemas from the authorized binding on its current Computer. pub async fn discover( State(state): State, actor: Actor, Path((bot_id, binding_id)): Path<(String, String)>, ) -> Result, ApiError> { let bot = state .db .get_bot(&actor, &bot_id) .await .map_err(failure)? .ok_or_else(|| { ( StatusCode::NOT_FOUND, Json(json!({"message":"bot not found"})), ) })?; let computer = state .db .get_computer(bot.computer_id.as_deref().unwrap_or("")) .await .map_err(failure)? .ok_or_else(|| failure("Computer missing"))?; if computer.state != "running" { return Err(failure("Computer is not running")); } let computer_ref = computer::computer_ref(&computer).ok_or_else(|| failure("Computer is not running"))?; let mut adapter = computer::adapter_context(&actor, &bot_id, "tool-discovery"); adapter.computer_generation = Some(computer.generation); let value = call_on_computer( state.pool(), &actor, &bot_id, state.sandbox.as_ref(), &computer_ref, &adapter, &json!({"action":"discover","bindingId":binding_id}), ) .await .map_err(|error| { ( StatusCode::BAD_GATEWAY, Json(json!({"message":error.message,"code":error.code})), ) })?; Ok(Json(value)) } #[derive(Deserialize)] #[serde(deny_unknown_fields, rename_all = "camelCase")] pub struct InvokeRequest { operation_id: Uuid, tool: String, arguments: serde_json::Map, } pub async fn invoke( State(state): State, actor: Actor, Path((bot_id, binding_id)): Path<(String, String)>, Json(body): Json, ) -> Result, ApiError> { if body.tool.trim().is_empty() { return Err(failure("tool name required")); } let bot = state .db .get_bot(&actor, &bot_id) .await .map_err(failure)? .ok_or_else(|| { ( StatusCode::NOT_FOUND, Json(json!({"message":"bot not found"})), ) })?; let computer = state .db .get_computer(bot.computer_id.as_deref().unwrap_or("")) .await .map_err(failure)? .ok_or_else(|| failure("Computer missing"))?; if computer.state != "running" { return Err(failure("Computer is not running")); } let computer_ref = computer::computer_ref(&computer).ok_or_else(|| failure("Computer is not running"))?; let ctx = ToolCtx::for_tool_manager( &state, &actor, &bot_id, computer_ref, parse_mode(&computer.scope), ); ctx.context.lock().unwrap().computer_generation = Some(computer.generation); let args = json!({"action":"call","bindingId":binding_id,"tool":body.tool,"arguments":body.arguments,"operationId":body.operation_id}); let outcome = crate::tools::dispatch(&ctx, "computer_mcp", &args).await; if let Some(code) = outcome.error_code { return Err(( StatusCode::CONFLICT, Json(json!({"message":outcome.text,"code":code,"operationId":body.operation_id})), )); } let result: Value = serde_json::from_str(&outcome.text).map_err(failure)?; Ok(Json( json!({"result":result,"operationId":body.operation_id}), )) } pub async fn revoke_runtime( state: &AppState, actor: &Actor, bot_id: &str, binding_id: &str, ) -> Result<(), String> { let computer_id: Option = sqlx::query_scalar("SELECT p.computer_id FROM tool_bindings b JOIN tool_packages p ON p.id=b.package_row_id JOIN bots bot ON bot.id=b.bot_id WHERE b.id=$1 AND b.bot_id=$2 AND bot.space_id=$3 AND bot.user_id=$4 AND p.manifest IS NOT NULL") .bind(binding_id).bind(bot_id).bind(&actor.space_id).bind(&actor.user_id).fetch_optional(state.pool()).await.map_err(|_| "binding lookup failed")?; let Some(id) = computer_id else { return Ok(()) }; let computer = state .db .get_computer(&id) .await .map_err(|_| "Computer lookup failed")? .ok_or("Computer unavailable")?; if computer.state != "running" { return Err( "binding is centrally revoked; local drain awaits Computer availability".into(), ); } let reference = computer::computer_ref(&computer).ok_or("Computer unavailable")?; let adapter = AdapterContext { bot_id: Some(bot_id.into()), computer_generation: Some(computer.generation), operation_id: format!("revoke:{binding_id}"), space_id: actor.space_id.clone(), user_id: actor.user_id.clone(), ..Default::default() }; let result = state .sandbox .execute( &reference, CommandRequest { job_op: Some("package_revoke".into()), stdin: Some(json!({"binding_id":binding_id}).to_string()), ..Default::default() }, &adapter, ) .await .map_err(|_| "binding is revoked; runtime drain not confirmed")?; if result.code != 0 { return Err("binding is revoked; runtime drain not confirmed".into()); } Ok(()) } #[cfg(test)] mod tests { use super::*; #[sqlx::test(migrations = "../../migrations")] async fn installed_visibility_follows_owner_binding_and_current_computer(pool: sqlx::PgPool) { for sql in [ "INSERT INTO users(id,name) VALUES ('u','owner'),('v','other')", "INSERT INTO spaces(id,user_id,name) VALUES ('s','u','test'),('other','v','other')", "INSERT INTO computers(id,space_id,user_id,scope,scope_key,home_key) VALUES ('c','s','u','team','team:s','home'),('d','s','u','dedicated','bot:d','private')", "INSERT INTO bots(id,space_id,user_id,name,computer_id) VALUES ('a','s','u','A','c'),('b','s','u','B','c')", "INSERT INTO tool_packages(id,computer_id,package_id,version,sha256,status,manifest) VALUES ('p','c','fixture','1.0.0','digest','installed','{}')", "INSERT INTO tool_bindings(id,package_row_id,bot_id,status) VALUES ('binding','p','a','ready')", ] { sqlx::query(sql).execute(&pool).await.unwrap(); } let owner = Actor { user_id: "u".into(), space_id: "s".into(), }; let other = Actor { user_id: "v".into(), space_id: "other".into(), }; let state = AppState { db: crate::db::Db { pool: pool.clone() }, sandbox: std::sync::Arc::new(lazyboy_sandbox::FakeSandbox::new()), data_dir: String::new(), auth: crate::auth::AuthConfig::from_env(), memory: crate::memory::MemoryService::from_env(), mcp: crate::mcp::McpHub::new(), calls: crate::state::CallRegistry::default(), wakes: crate::state::WakeBus::default(), }; sqlx::query("UPDATE computers SET state='running',provider_ref='fixture' WHERE id='c'") .execute(&pool) .await .unwrap(); sqlx::query("UPDATE computers SET state='stopped' WHERE id='c'") .execute(&pool) .await .unwrap(); let stopped = discover( State(state.clone()), owner.clone(), Path(("a".into(), "binding".into())), ) .await .unwrap_err(); assert_eq!(stopped.1.0["message"], "Computer is not running"); sqlx::query("UPDATE computers SET state='running' WHERE id='c'") .execute(&pool) .await .unwrap(); let operation_id = Uuid::new_v4(); let request = || InvokeRequest { operation_id, tool: "fixture_tool".into(), arguments: serde_json::Map::new(), }; let unauthorized = invoke( State(state.clone()), other.clone(), Path(("a".into(), "binding".into())), Json(request()), ) .await .unwrap_err(); assert_eq!(unauthorized.0, StatusCode::NOT_FOUND); let rejected = invoke( State(state.clone()), owner.clone(), Path(("b".into(), "binding".into())), Json(request()), ) .await .unwrap_err(); assert_eq!(rejected.1.0["code"], "MCP_REJECTED"); // The fake rejects MCP dispatch: treat transport failure as unknown, never replay it. let unknown_result = invoke( State(state.clone()), owner.clone(), Path(("a".into(), "binding".into())), Json(request()), ) .await .unwrap_err(); assert_eq!(unknown_result.1.0["code"], "UNKNOWN_EFFECT"); let stored_run: Option = sqlx::query_scalar("SELECT run_id FROM computer_operations WHERE id=$1") .bind(format!("a:{operation_id}")) .fetch_one(&pool) .await .unwrap(); assert!( stored_run.is_none(), "management calls must not invent a run" ); let mut changed = request(); changed.tool = "another_tool".into(); let mismatch = invoke( State(state.clone()), owner.clone(), Path(("a".into(), "binding".into())), Json(changed), ) .await .unwrap_err(); assert_eq!(mismatch.1.0["code"], "PAYLOAD_MISMATCH"); assert!(any_bound(&pool, &owner, "a").await.unwrap()); assert!(!any_bound(&pool, &other, "a").await.unwrap()); assert!(!any_bound(&pool, &owner, "b").await.unwrap()); sqlx::query("UPDATE bots SET computer_id='d' WHERE id='a'") .execute(&pool) .await .unwrap(); assert!(!any_bound(&pool, &owner, "a").await.unwrap()); sqlx::query("UPDATE bots SET computer_id='c' WHERE id='a'") .execute(&pool) .await .unwrap(); sqlx::query("UPDATE tool_bindings SET status='revoked' WHERE id='binding'") .execute(&pool) .await .unwrap(); assert!(!any_bound(&pool, &owner, "a").await.unwrap()); let replay = invoke( State(state.clone()), owner.clone(), Path(("a".into(), "binding".into())), Json(request()), ) .await .unwrap_err(); assert_eq!( replay.1.0["code"], "UNKNOWN_EFFECT", "replay must not dispatch to the now-revoked binding" ); let discovery = discover( State(state.clone()), owner.clone(), Path(("a".into(), "binding".into())), ) .await .unwrap_err(); assert_eq!(discovery.1.0["code"], "MCP_REJECTED"); sqlx::query("UPDATE operation_outbox SET created_at=now()-interval '5 seconds'") .execute(&pool) .await .unwrap(); crate::operations::flush_outbox(&pool).await; let pending: i64 = sqlx::query_scalar("SELECT count(*) FROM operation_outbox WHERE NOT delivered") .fetch_one(&pool) .await .unwrap(); assert_eq!( pending, 0, "management events must not retry an invalid run_activity foreign key forever" ); let manifest = ToolManifest { id: "fixture.reauthorized".into(), version: "1.0.0".into(), sha256: "0".repeat(64), entrypoint: vec!["python3".into(), "./server.py".into()], share_immutable_package: true, }; sqlx::query("UPDATE tool_packages SET package_id=$1,sha256=$2,manifest=$3 WHERE id='p'") .bind(&manifest.id) .bind(&manifest.sha256) .bind(serde_json::to_value(&manifest).unwrap()) .execute(&pool) .await .unwrap(); let install_request = |reauthorize| InstallRequest { manifest: serde_json::from_value(serde_json::to_value(&manifest).unwrap()).unwrap(), artifact_path: "server.tar.gz".into(), reauthorize, }; let denied = install( State(state.clone()), owner.clone(), Path("a".into()), Json(install_request(false)), ) .await .unwrap_err(); assert!( denied.1.0["message"] .as_str() .unwrap() .contains("explicit new authorization") ); // Fake rejects package installation dispatch: intent must survive as installing. let _ = install( State(state.clone()), owner.clone(), Path("a".into()), Json(install_request(true)), ) .await .unwrap_err(); let bindings: Vec<(String, String)> = sqlx::query_as( "SELECT id,status FROM tool_bindings WHERE package_row_id='p' AND bot_id='a'", ) .fetch_all(&pool) .await .unwrap(); assert_eq!(bindings.len(), 2); assert!(bindings.contains(&("binding".into(), "revoked".into()))); assert!( bindings .iter() .any(|(id, status)| id != "binding" && status == "installing") ); let _ = install( State(state), owner, Path("a".into()), Json(install_request(true)), ) .await .unwrap_err(); let count: i64 = sqlx::query_scalar( "SELECT count(*) FROM tool_bindings WHERE package_row_id='p' AND bot_id='a'", ) .fetch_one(&pool) .await .unwrap(); assert_eq!( count, 2, "retry reuses the new installing intent, never the retired id" ); } }