diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 2786d2c508..7f10eb8f6b 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use sha2::{Digest, Sha256}; use crate::client::{ @@ -9,6 +11,16 @@ use crate::validate::{parse_uuid, read_or_stdin, sdk_err, validate_uuid}; // TODO(phase-4): Replace raw nostr::EventBuilder usage with buzz-sdk builder functions +fn lifecycle_precedence(kind: u64) -> u8 { + match kind { + 46005..=46007 => 4, + 46011..=46012 => 3, + 46010 => 2, + 46001 => 1, + _ => 0, + } +} + /// List workflows in a channel — query kind:30620 workflow definition events. pub async fn cmd_list_workflows(client: &BuzzClient, channel_id: &str) -> Result<(), CliError> { validate_uuid(channel_id)?; @@ -57,12 +69,7 @@ pub async fn cmd_get_workflow(client: &BuzzClient, workflow_id: &str) -> Result< Ok(()) } -/// Get workflow run history — query kinds [46001, 46002, 46003]. -/// -/// NOTE: The relay does not currently emit workflow execution events (46001-46003). -/// Run history is stored in the workflow_runs DB table, not as Nostr events. -/// This command will return an empty array until the relay adds event emission -/// or a dedicated REST endpoint for run history. +/// Get workflow run history from durable lifecycle events. pub async fn cmd_get_workflow_runs( client: &BuzzClient, workflow_id: &str, @@ -71,24 +78,51 @@ pub async fn cmd_get_workflow_runs( validate_uuid(workflow_id)?; let limit = limit.unwrap_or(20).min(100); let filter = serde_json::json!({ - "kinds": [46001, 46002, 46003], + "kinds": [46001, 46005, 46006, 46007, 46010, 46011, 46012], "#d": [workflow_id], - "limit": limit + "limit": (limit * 10).min(500) }); let resp = client.query(&filter).await?; let events: Vec = serde_json::from_str(&resp).unwrap_or_default(); - let normalized: Vec = events - .iter() - .map(|e| { - serde_json::json!({ - "event_id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""), - "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0), - "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""), - "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), - "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])), + let mut latest: HashMap = HashMap::new(); + for event in events { + let Some(content) = event.get("content").and_then(|v| v.as_str()) else { + continue; + }; + let Ok(payload) = serde_json::from_str::(content) else { + continue; + }; + let kind = event.get("kind").and_then(|v| v.as_u64()).unwrap_or(0); + let run = if (46010..=46012).contains(&kind) { + payload.get("run").cloned().unwrap_or(payload) + } else { + payload + }; + if run.get("workflow_id").and_then(|v| v.as_str()) != Some(workflow_id) { + continue; + } + let Some(run_id) = run.get("id").and_then(|v| v.as_str()) else { + continue; + }; + let created_at = event + .get("created_at") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let precedence = lifecycle_precedence(kind); + if latest + .get(run_id) + .is_none_or(|(seen_at, seen_precedence, _)| { + (created_at, precedence) >= (*seen_at, *seen_precedence) }) - }) - .collect(); + { + latest.insert(run_id.to_owned(), (created_at, precedence, run)); + } + } + let mut normalized: Vec<(u64, u8, serde_json::Value)> = latest.into_values().collect(); + normalized.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| b.1.cmp(&a.1))); + normalized.truncate(limit as usize); + let normalized: Vec = + normalized.into_iter().map(|(_, _, run)| run).collect(); let output = serde_json::to_string(&normalized).unwrap_or_default(); println!("{output}"); Ok(()) @@ -196,12 +230,18 @@ pub async fn cmd_approve_step( approved: bool, note: Option<&str>, ) -> Result<(), CliError> { - validate_uuid(approval_token)?; - let content = note.unwrap_or(""); // The relay expects d-tag = hex(SHA256(token)), not the raw token UUID. - let token_hash = hex::encode(Sha256::digest(approval_token.as_bytes())); + // Desktop approval cards already expose the stored hash, so accept either + // that 64-character hash or the original raw UUID for CLI compatibility. + let token_hash = + if approval_token.len() == 64 && approval_token.chars().all(|c| c.is_ascii_hexdigit()) { + approval_token.to_ascii_lowercase() + } else { + validate_uuid(approval_token)?; + hex::encode(Sha256::digest(approval_token.as_bytes())) + }; let builder = buzz_sdk::build_workflow_approval(&token_hash, approved, content).map_err(sdk_err)?; let event = client.sign_event(builder)?; diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..9de209e271 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3825,6 +3825,16 @@ impl Db { workflow::create_approval(&self.pool, params).await } + /// Atomically suspend a workflow run and create its pending approval. + pub async fn suspend_workflow_run_for_approval( + &self, + params: workflow::CreateApprovalParams<'_>, + current_step: i32, + trace: &serde_json::Value, + ) -> Result { + workflow::suspend_workflow_run_for_approval(&self.pool, params, current_step, trace).await + } + /// Fetch an approval by raw token. pub async fn get_approval( &self, diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..e46b285afc 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -979,6 +979,88 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> Ok(()) } +/// Atomically suspend a run and create its pending approval record. +/// +/// A waiting run without its approval row is unreachable, while an approval +/// row without a waiting run is actionable garbage. Keep the two writes in the +/// same transaction so the relay never commits either half of a human gate. +pub async fn suspend_workflow_run_for_approval( + pool: &PgPool, + params: CreateApprovalParams<'_>, + current_step: i32, + trace: &serde_json::Value, +) -> Result { + let CreateApprovalParams { + community_id, + token, + workflow_id, + run_id, + step_id, + step_index, + approver_spec, + expires_at, + } = params; + let token_hash = hash_approval_token(token); + let mut tx = pool.begin().await?; + + let updated = sqlx::query( + r#" + UPDATE workflow_runs + SET status = 'waiting_approval', + current_step = $1, + execution_trace = $2, + error_message = NULL + WHERE community_id = $3 AND id = $4 + "#, + ) + .bind(current_step) + .bind(trace) + .bind(community_id.as_uuid()) + .bind(run_id) + .execute(&mut *tx) + .await? + .rows_affected(); + + if updated == 0 { + return Err(DbError::NotFound(format!("workflow_run {run_id}"))); + } + + let created_at: DateTime = sqlx::query_scalar( + r#" + INSERT INTO workflow_approvals + (community_id, token, workflow_id, run_id, step_id, step_index, approver_spec, status, expires_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8) + RETURNING created_at + "#, + ) + .bind(community_id.as_uuid()) + .bind(&token_hash) + .bind(workflow_id) + .bind(run_id) + .bind(step_id) + .bind(step_index) + .bind(approver_spec) + .bind(expires_at) + .fetch_one(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(ApprovalRecord { + token: token_hash, + workflow_id, + run_id, + step_id: step_id.to_owned(), + step_index, + approver_spec: approver_spec.to_owned(), + status: ApprovalStatus::Pending, + approver_pubkey: None, + note: None, + expires_at, + created_at, + }) +} + /// Fetch an approval record by raw token. /// /// The token is hashed before the DB lookup so plaintext tokens are never diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453f..e427a3a765 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1918,6 +1918,11 @@ pub async fn workflow_webhook( .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; + state + .workflow_engine + .record_run_triggered(community_id, run_id) + .await; + // Spawn workflow execution asynchronously. let engine = Arc::clone(&state.workflow_engine); let db = state.db.clone(); diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..c261a470f2 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -939,6 +939,11 @@ async fn handle_workflow_trigger( .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + state + .workflow_engine + .record_run_triggered(community_id, run_id) + .await; + // 5. Spawn workflow execution let engine = Arc::clone(&state.workflow_engine); let db = state.db.clone(); @@ -1111,8 +1116,17 @@ async fn handle_approval_grant( .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // 6. Resume workflow execution (post-commit, async) let community_id = tenant.community(); + let mut resolved_approval = approval.clone(); + resolved_approval.status = ApprovalStatus::Granted; + resolved_approval.approver_pubkey = Some(self_bytes.clone()); + resolved_approval.note = note.map(str::to_owned); + state + .workflow_engine + .record_approval_resolution(community_id, &resolved_approval) + .await; + + // 6. Resume workflow execution (post-commit, async) let run_id = approval.run_id; let workflow_id = approval.workflow_id; let resume_index = approval.step_index as usize + 1; @@ -1222,44 +1236,48 @@ async fn handle_approval_deny( .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // 6. Cancel the workflow run (post-commit, async) + // 6. Cancel the workflow run and persist its lifecycle receipt. let community_id = tenant.community(); let run_id = approval.run_id; let pubkey_hex = self_hex.clone(); - let db = state.db.clone(); - - tokio::spawn(async move { - let run = match db.get_workflow_run(community_id, run_id).await { - Ok(r) => r, - Err(e) => { - tracing::error!("approval_deny: failed to fetch run {run_id}: {e}"); - return; - } - }; + let run = state + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| IngestError::Internal(format!("error: db get_workflow_run: {e}")))?; + if run.status != RunStatus::WaitingApproval { + return Err(IngestError::Rejected(format!( + "invalid: workflow run is {} rather than waiting_approval", + run.status + ))); + } - if run.status != RunStatus::WaitingApproval { - tracing::warn!( - "approval_deny: run {run_id} has status '{}', expected 'waiting_approval'", - run.status - ); - return; - } + let cancel_msg = format!("workflow cancelled: approval denied by {pubkey_hex}"); + state + .db + .update_workflow_run( + community_id, + run_id, + RunStatus::Cancelled, + run.current_step, + &run.execution_trace, + Some(&cancel_msg), + ) + .await + .map_err(|e| IngestError::Internal(format!("error: db cancel workflow_run: {e}")))?; - let cancel_msg = format!("workflow cancelled: approval denied by {pubkey_hex}"); - if let Err(e) = db - .update_workflow_run( - community_id, - run_id, - RunStatus::Cancelled, - run.current_step, - &run.execution_trace, - Some(&cancel_msg), - ) - .await - { - tracing::error!("approval_deny: failed to cancel run {run_id}: {e}"); - } - }); + let mut resolved_approval = approval.clone(); + resolved_approval.status = ApprovalStatus::Denied; + resolved_approval.approver_pubkey = Some(self_bytes); + resolved_approval.note = note.map(str::to_owned); + state + .workflow_engine + .record_approval_resolution(community_id, &resolved_approval) + .await; + state + .workflow_engine + .record_run_cancelled(community_id, run_id) + .await; // 7. Return response Ok(IngestResult { @@ -1353,7 +1371,16 @@ async fn resume_workflow_after_approval( .unwrap_or_default(); // Execute remaining steps - let existing_trace = run.execution_trace.as_array().cloned(); + let mut existing_trace = run.execution_trace.as_array().cloned().unwrap_or_default(); + if resume_index > 0 { + if let Some(entry) = existing_trace.get_mut(resume_index - 1) { + if entry.get("status").and_then(serde_json::Value::as_str) == Some("waiting_approval") { + entry["status"] = serde_json::json!("completed"); + entry["output"] = serde_json::json!({"approval": "granted"}); + } + } + } + let existing_trace = Some(existing_trace); let result = buzz_workflow::executor::execute_from_step( &engine, community_id, diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..39ce0ada48 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,9 +8,9 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_core::kind::{is_workflow_execution_kind, KIND_STREAM_MESSAGE}; use buzz_core::tenant::CommunityId; -use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; +use buzz_workflow::action_sink::{ActionSink, ActionSinkError, WorkflowLifecycleEvent}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; use tracing::info; @@ -362,6 +362,102 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn emit_workflow_lifecycle( + &self, + community_id: CommunityId, + lifecycle: WorkflowLifecycleEvent, + ) -> Pin> + Send + '_>> { + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + if !is_workflow_execution_kind(lifecycle.kind) { + return Err(ActionSinkError::InvalidInput(format!( + "invalid workflow lifecycle kind {}", + lifecycle.kind + ))); + } + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow lifecycle community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + let WorkflowLifecycleEvent { + kind, + workflow_id, + run_id, + channel_id, + content, + token_hash, + target_pubkeys, + } = lifecycle; + + let mut tags = vec![ + Tag::parse(["d", &workflow_id.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + Tag::parse(["run", &run_id.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("run tag: {e}")))?, + ]; + if let Some(channel_id) = channel_id { + tags.push( + Tag::parse(["h", &channel_id.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("channel tag: {e}")))?, + ); + } + if let Some(token_hash) = token_hash { + tags.push( + Tag::parse(["t", &token_hash]) + .map_err(|e| ActionSinkError::EventBuild(format!("token tag: {e}")))?, + ); + } + for pubkey in target_pubkeys { + nostr::PublicKey::from_hex(&pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid lifecycle target pubkey: {e}")) + })?; + tags.push( + Tag::parse(["p", &pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("recipient tag: {e}")))?, + ); + } + + let event = EventBuilder::new(Kind::Custom(kind as u16), content.to_string()) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + let event_id_hex = event.id.to_hex(); + let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); + + let (stored_event, was_inserted) = state + .db + .insert_event(community_id, &event, channel_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind, + &relay_pubkey_hex, + None, + ) + .await; + } + + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..698dd3a99b 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -7,6 +7,8 @@ use std::future::Future; use std::pin::Pin; use buzz_core::tenant::CommunityId; +use serde_json::Value; +use uuid::Uuid; /// Errors from action sink operations. #[derive(Debug, thiserror::Error)] @@ -37,6 +39,30 @@ impl From for crate::WorkflowError { } } +/// A durable workflow lifecycle event to be signed and published by the relay. +/// +/// The workflow engine owns the state transition; the relay owns the Nostr +/// event, its signature, persistence, and fan-out. Keeping the raw approval +/// token out of this shape ensures lifecycle events can expose only the stored +/// token hash used by approval actions. +#[derive(Clone, Debug)] +pub struct WorkflowLifecycleEvent { + /// The workflow lifecycle kind (for example, 46001 or 46010). + pub kind: u32, + /// The workflow that owns this lifecycle record. + pub workflow_id: Uuid, + /// The workflow run represented by this record. + pub run_id: Uuid, + /// The channel that scopes the workflow, if it has one. + pub channel_id: Option, + /// The JSON wire payload consumed by Buzz clients. + pub content: Value, + /// The SHA-256 approval token hash, when this lifecycle event represents an approval. + pub token_hash: Option, + /// Exact recipients for a targeted approval notification. + pub target_pubkeys: Vec, +} + /// Interface for workflow actions that produce side effects. /// /// Implemented by the relay to provide direct DB/event access to the executor. @@ -66,4 +92,11 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Persist and fan out a relay-signed workflow lifecycle event. + fn emit_workflow_lifecycle( + &self, + community_id: CommunityId, + lifecycle: WorkflowLifecycleEvent, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..295dfa201c 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -451,6 +451,24 @@ pub fn resolve_step_templates( } } +/// Approval data captured when a workflow pauses at a human gate. +/// +/// The raw token exists only until the finalizer hashes and persists it. It is +/// never put in an execution trace or lifecycle event. +#[derive(Debug)] +pub struct PendingApproval { + /// Raw one-time token that is hashed before persistence. + pub(crate) token: String, + /// The exact approver policy from the workflow definition. + pub(crate) approver_spec: String, + /// The step that requested this human gate. + pub(crate) step_id: String, + /// The human-facing decision request after template resolution. + pub(crate) message: String, + /// How long the human gate remains actionable. + pub(crate) timeout_secs: u64, +} + /// Result of dispatching a single step action. #[derive(Debug)] pub enum StepResult { @@ -458,8 +476,8 @@ pub enum StepResult { Completed(JsonValue), /// Step requests suspension (approval gate). Execution must pause. Suspended { - /// Token used to resume or reject this approval gate. - approval_token: String, + /// Approval state that the finalizer must persist before returning. + approval: PendingApproval, }, /// Step was skipped due to `if:` condition being false. Skipped, @@ -653,6 +671,7 @@ pub async fn dispatch_action( timeout, } => { let timeout_str = timeout.as_deref().unwrap_or("24h"); + let timeout_secs = parse_duration_secs(timeout_str)?; info!( run_id = %run_id, step = step_id, "RequestApproval from={from} timeout={timeout_str}: {message}" @@ -660,11 +679,14 @@ pub async fn dispatch_action( let token = generate_approval_token(run_id, step_id); - // TODO (WF-08): create approval record in DB, emit kind:46010. - // For now, return Suspended with the token so the caller can persist state. - Ok(StepResult::Suspended { - approval_token: token, + approval: PendingApproval { + token, + approver_spec: from.to_owned(), + step_id: step_id.to_owned(), + message: message.to_owned(), + timeout_secs, + }, }) } @@ -942,7 +964,7 @@ async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result, + pub pending_approval: Option, /// Index of the step that suspended (or the total step count on completion). pub step_index: usize, /// Accumulated step outputs at the point of suspension or completion. @@ -959,10 +981,10 @@ pub struct ExecutionResult { /// 3. Dispatches the action. /// 4. Stores the step output for use by later steps. /// -/// On `RequestApproval`: returns `ExecutionResult` with `approval_token = Some(token)`. +/// On `RequestApproval`: returns `ExecutionResult` with `pending_approval` set. /// Caller must persist the approval record and update the run status. /// -/// Returns `ExecutionResult` with `approval_token = None` on normal completion. +/// Returns `ExecutionResult` with `pending_approval = None` on normal completion. /// /// Enforces `engine.config.max_concurrent` via a semaphore — returns /// [`WorkflowError::CapacityExceeded`] immediately if all permits are taken. @@ -1034,7 +1056,7 @@ pub async fn execute_from_step( // Mark run as Running now that we have a permit (resume from approval). // Preserve the existing execution trace from pre-approval steps. - let existing_trace = match engine.db.get_workflow_run(community_id, run_id).await { + let mut existing_trace = match engine.db.get_workflow_run(community_id, run_id).await { Ok(r) => r.execution_trace, Err(e) => { warn!( @@ -1044,6 +1066,17 @@ pub async fn execute_from_step( serde_json::json!([]) } }; + if start_index > 0 { + if let Some(entry) = existing_trace + .as_array_mut() + .and_then(|trace| trace.get_mut(start_index - 1)) + { + if entry.get("status").and_then(JsonValue::as_str) == Some("waiting_approval") { + entry["status"] = serde_json::json!("completed"); + entry["output"] = serde_json::json!({"approval": "granted"}); + } + } + } engine .db .update_workflow_run( @@ -1183,15 +1216,17 @@ async fn execute_steps( })); step_outputs.insert(step.id.clone(), output); } - StepResult::Suspended { approval_token } => { + StepResult::Suspended { approval } => { info!( run_id = %run_id, step = %step.id, "Step suspended — awaiting approval (token: )" ); - // Return the token and current state so the caller can persist the - // approval record and update the run's execution trace. + trace.push(serde_json::json!({ + "step_id": step.id, + "status": "waiting_approval", + })); return Ok(ExecutionResult { - approval_token: Some(approval_token), + pending_approval: Some(approval), step_index: i, step_outputs, trace, @@ -1209,7 +1244,7 @@ async fn execute_steps( info!(run_id = %run_id, "Workflow run completed"); Ok(ExecutionResult { - approval_token: None, + pending_approval: None, step_index: def.steps.len(), step_outputs, trace, diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index e142221169..069cf748e9 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -35,7 +35,7 @@ pub mod error; pub mod executor; pub mod schema; -pub use action_sink::{ActionSink, ActionSinkError}; +pub use action_sink::{ActionSink, ActionSinkError, WorkflowLifecycleEvent}; pub use error::{PartialProgress, WorkflowError}; pub use executor::ExecutionResult; pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef}; @@ -44,9 +44,13 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::OnceLock; -use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION}; +use buzz_core::kind::{ + event_kind_u32, is_workflow_execution_kind, KIND_REACTION, KIND_WORKFLOW_APPROVAL_DENIED, + KIND_WORKFLOW_APPROVAL_GRANTED, KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_WORKFLOW_COMPLETED, + KIND_WORKFLOW_FAILED, KIND_WORKFLOW_TRIGGERED, +}; use buzz_core::tenant::CommunityId; -use buzz_db::workflow::RunStatus; +use buzz_db::workflow::{ApprovalRecord, ApprovalStatus, CreateApprovalParams, RunStatus}; use buzz_db::Db; use chrono::{DateTime, Utc}; use dashmap::DashMap; @@ -104,6 +108,12 @@ pub struct WorkflowEngine { moka::sync::Cache<(CommunityId, Uuid), Arc>>, } +fn exact_pubkey_target(approver_spec: &str) -> Option { + let normalized = approver_spec.trim(); + (normalized.len() == 64 && normalized.chars().all(|c| c.is_ascii_hexdigit())) + .then(|| normalized.to_ascii_lowercase()) +} + impl WorkflowEngine { /// Create a new `WorkflowEngine`. pub fn new(db: Db, config: WorkflowConfig) -> Self { @@ -194,6 +204,160 @@ impl WorkflowEngine { }) } + async fn emit_lifecycle( + &self, + community_id: CommunityId, + lifecycle: WorkflowLifecycleEvent, + ) -> Result<(), WorkflowError> { + let Some(sink) = self.action_sink.get() else { + // DB-only engine tests intentionally run without a relay. The real + // relay always installs its sink during AppState construction. + tracing::debug!( + workflow_id = %lifecycle.workflow_id, + run_id = %lifecycle.run_id, + kind = lifecycle.kind, + "Skipping workflow lifecycle event because no relay sink is installed" + ); + return Ok(()); + }; + + sink.emit_workflow_lifecycle(community_id, lifecycle) + .await + .map(|_| ()) + .map_err(WorkflowError::from) + } + + fn run_lifecycle_payload(run: &buzz_db::workflow::WorkflowRunRecord) -> serde_json::Value { + serde_json::json!({ + "schema_version": 1, + "id": run.id.to_string(), + "workflow_id": run.workflow_id.to_string(), + "status": run.status.to_string(), + "current_step": run.current_step, + "execution_trace": run.execution_trace, + "started_at": run.started_at.map(|at| at.timestamp()), + "completed_at": run.completed_at.map(|at| at.timestamp()), + "error_message": run.error_message, + "created_at": run.created_at.timestamp(), + }) + } + + async fn emit_run_lifecycle( + &self, + community_id: CommunityId, + run_id: Uuid, + kind: u32, + ) -> Result<(), WorkflowError> { + let run = self.db.get_workflow_run(community_id, run_id).await?; + let workflow = self.db.get_workflow(community_id, run.workflow_id).await?; + let lifecycle = WorkflowLifecycleEvent { + kind, + workflow_id: run.workflow_id, + run_id, + channel_id: workflow.channel_id, + content: Self::run_lifecycle_payload(&run), + token_hash: None, + target_pubkeys: Vec::new(), + }; + self.emit_lifecycle(community_id, lifecycle).await + } + + async fn emit_approval_lifecycle( + &self, + community_id: CommunityId, + approval: &ApprovalRecord, + kind: u32, + message: Option<&str>, + ) -> Result<(), WorkflowError> { + let workflow = self + .db + .get_workflow(community_id, approval.workflow_id) + .await?; + let run = self + .db + .get_workflow_run(community_id, approval.run_id) + .await?; + let token_hash = hex::encode(&approval.token); + let mut content = serde_json::json!({ + "schema_version": 1, + "token": token_hash, + "workflow_id": approval.workflow_id.to_string(), + "run_id": approval.run_id.to_string(), + "step_id": approval.step_id, + "step_index": approval.step_index, + "approver_spec": approval.approver_spec, + "status": approval.status.to_string(), + "approver_pubkey": approval.approver_pubkey.as_ref().map(hex::encode), + "note": approval.note, + "expires_at": approval.expires_at.to_rfc3339(), + "created_at": approval.created_at.timestamp(), + "run": Self::run_lifecycle_payload(&run), + }); + if let Some(message) = message { + content["message"] = serde_json::Value::String(message.to_owned()); + } + + let target_pubkeys = exact_pubkey_target(&approval.approver_spec) + .into_iter() + .collect(); + let lifecycle = WorkflowLifecycleEvent { + kind, + workflow_id: approval.workflow_id, + run_id: approval.run_id, + channel_id: workflow.channel_id, + content, + token_hash: Some(token_hash), + target_pubkeys, + }; + self.emit_lifecycle(community_id, lifecycle).await + } + + /// Emit the durable lifecycle event for a newly created workflow run. + pub async fn record_run_triggered(&self, community_id: CommunityId, run_id: Uuid) { + if let Err(e) = self + .emit_run_lifecycle(community_id, run_id, KIND_WORKFLOW_TRIGGERED) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-triggered lifecycle event: {e}"); + } + } + + /// Emit the durable lifecycle event for a cancelled workflow run. + pub async fn record_run_cancelled(&self, community_id: CommunityId, run_id: Uuid) { + if let Err(e) = self + .emit_run_lifecycle( + community_id, + run_id, + buzz_core::kind::KIND_WORKFLOW_CANCELLED, + ) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-cancelled lifecycle event: {e}"); + } + } + + /// Emit the durable lifecycle event for an approval decision. + pub async fn record_approval_resolution( + &self, + community_id: CommunityId, + approval: &ApprovalRecord, + ) { + let kind = match approval.status { + ApprovalStatus::Granted => KIND_WORKFLOW_APPROVAL_GRANTED, + ApprovalStatus::Denied => KIND_WORKFLOW_APPROVAL_DENIED, + ApprovalStatus::Pending | ApprovalStatus::Expired => return, + }; + if let Err(e) = self + .emit_approval_lifecycle(community_id, approval, kind, None) + .await + { + tracing::error!( + run_id = %approval.run_id, + "Failed to emit approval-resolution lifecycle event: {e}" + ); + } + } + /// Parse and validate a YAML workflow definition. /// /// Returns `(WorkflowDef, canonical_json)` on success. The canonical JSON @@ -221,35 +385,116 @@ impl WorkflowEngine { match result { Ok(result) => { + let ExecutionResult { + pending_approval, + step_index, + trace, + .. + } = result; let mut full_trace = prefix; - full_trace.extend(result.trace); + full_trace.extend(trace); let trace_json = serde_json::Value::Array(full_trace); - let step_count = result.step_index as i32; + let step_count = step_index as i32; + + if let Some(pending) = pending_approval { + let expires_at = i64::try_from(pending.timeout_secs) + .ok() + .and_then(|seconds| { + Utc::now().checked_add_signed(chrono::Duration::seconds(seconds)) + }); + let Some(expires_at) = expires_at else { + let error = "approval timeout is out of range"; + tracing::error!(run_id = %run_id, "{error}"); + if let Err(db_err) = self + .db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + step_count, + &trace_json, + Some(error), + ) + .await + { + tracing::error!(run_id = %run_id, "Failed to mark invalid approval timeout as failed: {db_err}"); + } else if let Err(emit_err) = self + .emit_run_lifecycle(community_id, run_id, KIND_WORKFLOW_FAILED) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-failed lifecycle event: {emit_err}"); + } + return; + }; - if result.approval_token.is_some() { - // Approval gates are not yet implemented (WF-08). - // Fail explicitly rather than creating unreachable WaitingApproval rows. - tracing::warn!( - run_id = %run_id, - step_index = result.step_index, - "Workflow hit approval gate — not yet implemented, marking as failed" - ); - if let Err(e) = self + let workflow_id = match self.db.get_workflow_run(community_id, run_id).await { + Ok(run) => run.workflow_id, + Err(e) => { + tracing::error!(run_id = %run_id, "Failed to resolve workflow for approval: {e}"); + return; + } + }; + + let approval = self .db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, + .suspend_workflow_run_for_approval( + CreateApprovalParams { + community_id, + token: &pending.token, + workflow_id, + run_id, + step_id: &pending.step_id, + step_index: step_count, + approver_spec: &pending.approver_spec, + expires_at, + }, step_count, &trace_json, - Some("approval gates not yet implemented — see WF-08"), ) - .await - { - tracing::error!( - run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" - ); + .await; + + match approval { + Ok(approval) => { + tracing::info!( + run_id = %run_id, + step_index, + "Workflow suspended awaiting approval" + ); + if let Err(e) = self + .emit_approval_lifecycle( + community_id, + &approval, + KIND_WORKFLOW_APPROVAL_REQUESTED, + Some(&pending.message), + ) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit approval-requested lifecycle event: {e}"); + } + } + Err(e) => { + let error = format!("failed to persist approval gate: {e}"); + tracing::error!(run_id = %run_id, "{error}"); + if let Err(db_err) = self + .db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + step_count, + &trace_json, + Some(&error), + ) + .await + { + tracing::error!(run_id = %run_id, "Failed to mark approval-persistence error as failed: {db_err}"); + } else if let Err(emit_err) = self + .emit_run_lifecycle(community_id, run_id, KIND_WORKFLOW_FAILED) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-failed lifecycle event: {emit_err}"); + } + } } } else { tracing::info!(run_id = %run_id, "Workflow run completed"); @@ -269,6 +514,11 @@ impl WorkflowEngine { run_id = %run_id, "Failed to update run to Completed: {e}" ); + } else if let Err(e) = self + .emit_run_lifecycle(community_id, run_id, KIND_WORKFLOW_COMPLETED) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-completed lifecycle event: {e}"); } } } @@ -293,6 +543,11 @@ impl WorkflowEngine { run_id = %run_id, "Failed to update run to Failed: {db_err}" ); + } else if let Err(emit_err) = self + .emit_run_lifecycle(community_id, run_id, KIND_WORKFLOW_FAILED) + .await + { + tracing::error!(run_id = %run_id, "Failed to emit workflow-failed lifecycle event: {emit_err}"); } } } @@ -419,6 +674,8 @@ impl WorkflowEngine { "Workflow triggered — spawning execution" ); + self.record_run_triggered(community_id, run_id).await; + let engine = Arc::clone(self); let def_clone = def.clone(); let ctx_clone = trigger_ctx.clone(); @@ -714,6 +971,8 @@ impl WorkflowEngine { "Cron trigger fired" ); + self.record_run_triggered(community_id, run_id).await; + let engine = Arc::clone(self); let def_clone = def.clone(); let ctx_clone = trigger_ctx.clone(); diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index dd61fc9398..f88741a53e 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -4,14 +4,17 @@ use tauri::{AppHandle, State}; use crate::{ app_state::AppState, managed_agents::{ - build_managed_agent_summary, current_instance_id, discover_provider_candidates, - ensure_persona_is_active, find_managed_agent_mut, load_managed_agents, load_personas, - load_teams, managed_agent_avatar_url, normalize_agent_args, provider_deploy, - resolve_provider_binary, save_managed_agents, start_managed_agent_process, - stop_managed_agent_process, stop_managed_agent_workspace_pair, + build_external_agent_runtime, build_external_runtime_event, build_managed_agent_summary, + current_instance_id, discover_provider_candidates, ensure_persona_is_active, + external_runtime_projection_from_event, find_external_runtime_conflict, + find_managed_agent_mut, load_external_agent_runtimes, load_managed_agents, load_personas, + load_teams, managed_agent_avatar_url, normalize_agent_args, normalize_public_key, + provider_deploy, resolve_provider_binary, save_external_agent_runtimes, save_managed_agents, + start_managed_agent_process, stop_managed_agent_process, stop_managed_agent_workspace_pair, sync_managed_agent_processes, try_regenerate_nest, validate_provider_config, BackendKind, - CreateManagedAgentRequest, CreateManagedAgentResponse, ManagedAgentRecord, - ManagedAgentSummary, RelayMeshConfig, DEFAULT_ACP_COMMAND, DEFAULT_AGENT_PARALLELISM, + CreateManagedAgentRequest, CreateManagedAgentResponse, ExternalAgentRuntime, + ManagedAgentRecord, ManagedAgentSummary, RegisterExternalAgentRuntimeRequest, + RelayMeshConfig, DEFAULT_ACP_COMMAND, DEFAULT_AGENT_PARALLELISM, DEFAULT_AGENT_TURN_TIMEOUT_SECONDS, }, relay::{relay_ws_url_with_override, sync_managed_agent_profile}, @@ -59,6 +62,38 @@ pub(super) fn retain_managed_agent_pending( } } +/// Queue an owner-signed external-runtime provenance projection. This uses +/// the existing kind:30177 retention pipe but never creates a +/// ManagedAgentRecord, private key, provider deployment, or process handle. +fn retain_external_agent_pending( + app: &AppHandle, + state: &AppState, + record: &ExternalAgentRuntime, +) -> Result<(), String> { + use crate::managed_agents::retention::{ + active_retention_scope, open_retention_db, retain_event, RetainedEvent, + }; + use buzz_core_pkg::kind::KIND_MANAGED_AGENT; + use nostr::JsonUtil; + + let scope = active_retention_scope(app, state)?; + let owner_pubkey = scope.owner_keys.public_key().to_hex(); + let event = build_external_runtime_event(&scope.owner_keys, record)?; + let conn = open_retention_db(&scope.db_path)?; + retain_event( + &conn, + &RetainedEvent { + kind: KIND_MANAGED_AGENT, + pubkey: owner_pubkey, + d_tag: record.agent_pubkey.clone(), + content: event.content.to_string(), + created_at: event.created_at.as_secs() as i64, + raw_event: event.as_json(), + pending_sync: true, + }, + ) +} + /// Purge a deleted agent's pending row and enqueue a NIP-09 tombstone, both /// inside the `managed_agents_store_lock`-held delete body and NEVER across an /// `.await`. @@ -1350,6 +1385,187 @@ pub async fn delete_managed_agent( .map_err(|e| format!("spawn_blocking failed: {e}"))? } +/// Read the owner-only external-runtime register. This command has no +/// lifecycle side effects and intentionally does not merge entries into the +/// managed-agent list. +#[tauri::command] +pub async fn list_external_agent_runtimes( + app: AppHandle, +) -> Result, String> { + use tauri::Manager; + tokio::task::spawn_blocking(move || { + let state = app.state::(); + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + load_external_agent_runtimes(&app) + }) + .await + .map_err(|error| format!("spawn_blocking failed: {error}"))? +} + +/// Register an already-running external identity without importing its key or +/// starting a second executor. The NIP-OA tag is verified against the current +/// workspace owner and then discarded; only public provenance metadata is +/// stored locally and queued as an owner-signed kind:30177 projection. +#[tauri::command] +pub async fn register_external_agent_runtime( + input: RegisterExternalAgentRuntimeRequest, + app: AppHandle, +) -> Result { + use tauri::Manager; + use buzz_core_pkg::kind::KIND_MANAGED_AGENT; + + let owner_hex = { + let state = app.state::(); + workspace_owner_hex(&state)? + }; + let record = build_external_agent_runtime(&input, &owner_hex, &now_iso())?; + + // A second desktop may have registered the same identity since this + // device's local register was last read. Query the current replaceable + // provenance head before allowing another scope; a failed read is a + // fail-closed registration error rather than an invitation to duplicate. + let existing_events = { + let state = app.state::(); + crate::relay::query_relay( + &state, + &[serde_json::json!({ + "kinds": [KIND_MANAGED_AGENT], + "authors": [owner_hex], + "#d": [record.agent_pubkey], + })], + ) + .await? + }; + for event in &existing_events { + let has_agent_coordinate = event.tags.iter().any(|tag| { + let values: Vec<&str> = tag.as_slice().iter().map(|value| value.as_str()).collect(); + values.first() == Some(&"d") + && values.get(1).copied() == Some(record.agent_pubkey.as_str()) + }); + if !has_agent_coordinate { + continue; + } + let Some(existing) = external_runtime_projection_from_event(event, &record.agent_pubkey)? + else { + return Err(format!( + "agent {} already has a managed-agent provenance event on the active relay; stop/remove that runner before registering an external scope", + record.agent_pubkey + )); + }; + if existing.archived { + continue; + } + if existing.deployment_scope == record.deployment_scope { + return Err(format!( + "external agent {} is already registered in this deployment scope on the active relay", + record.agent_pubkey + )); + } + return Err(format!( + "external agent {} is already active in deployment scope '{}' on the active relay; one identity may have only one live runner scope", + record.agent_pubkey, existing.deployment_scope + )); + } + + tokio::task::spawn_blocking(move || { + let state = app.state::(); + let agent_pubkey = record.agent_pubkey.clone(); + let deployment_scope = record.deployment_scope.clone(); + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + + // A local managed record would share the same relay identity and + // therefore create two possible executor owners. Registration fails + // closed instead of trying to reconcile private keys. + if load_managed_agents(&app)? + .iter() + .any(|managed| managed.pubkey == agent_pubkey) + { + return Err(format!( + "agent {agent_pubkey} already has a local managed-agent record; stop/remove that runner before registering an external scope" + )); + } + + let mut records = load_external_agent_runtimes(&app)?; + if let Some(conflict) = + find_external_runtime_conflict(&records, &agent_pubkey, &deployment_scope) + { + return Err(conflict.message(&agent_pubkey)); + } + + records.push(record.clone()); + save_external_agent_runtimes(&app, &records)?; + if let Err(error) = retain_external_agent_pending(&app, &state, &record) { + records.pop(); + let rollback_error = save_external_agent_runtimes(&app, &records).err(); + return Err(match rollback_error { + Some(rollback_error) => format!( + "failed to retain external-runtime provenance: {error}; rollback also failed: {rollback_error}" + ), + None => format!("failed to retain external-runtime provenance: {error}"), + }); + } + Ok(record) + }) + .await + .map_err(|error| format!("spawn_blocking failed: {error}"))? +} + +/// Archive an external registration while preserving its local and relay +/// history. This is not access revocation and does not claim the external +/// runner stopped; the runner owner remains responsible for its shutdown path. +#[tauri::command] +pub async fn archive_external_agent_runtime( + agent_pubkey: String, + app: AppHandle, +) -> Result { + use tauri::Manager; + tokio::task::spawn_blocking(move || { + let state = app.state::(); + let owner_hex = workspace_owner_hex(&state)?; + let normalized = normalize_public_key(&agent_pubkey, "agent")?; + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + let mut records = load_external_agent_runtimes(&app)?; + let record_index = records + .iter() + .position(|record| record.agent_pubkey == normalized && !record.archived) + .ok_or_else(|| format!("active external agent {normalized} not found"))?; + let previous = records[record_index].clone(); + if previous.owner_pubkey != owner_hex { + return Err( + "external runtime is owned by a different workspace identity; switch to that identity before archiving" + .to_string(), + ); + } + let mut archived = previous.clone(); + archived.archived = true; + archived.updated_at = now_iso(); + records[record_index] = archived.clone(); + save_external_agent_runtimes(&app, &records)?; + if let Err(error) = retain_external_agent_pending(&app, &state, &archived) { + records[record_index] = previous; + let rollback_error = save_external_agent_runtimes(&app, &records).err(); + return Err(match rollback_error { + Some(rollback_error) => format!( + "failed to retain external-runtime archive provenance: {error}; rollback also failed: {rollback_error}" + ), + None => format!("failed to retain external-runtime archive provenance: {error}"), + }); + } + Ok(archived) + }) + .await + .map_err(|error| format!("spawn_blocking failed: {error}"))? +} + // Remote agent shutdown is handled entirely by the frontend: // 1. Frontend sends "!shutdown" @mention via WebSocket (signed by user's key) // 2. Harness sees it, exits gracefully, sets presence to "offline" diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index 1d5f309fb5..efce35620a 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use serde::Serialize; use serde_json::Value; use tauri::State; @@ -47,6 +49,35 @@ pub struct WorkflowSaveWire { pub webhook_secret: Option, } +fn lifecycle_payload(event: &nostr::Event) -> Option { + serde_json::from_str::(&event.content) + .ok() + .filter(Value::is_object) +} + +fn lifecycle_run_payload(event: &nostr::Event) -> Option { + let payload = lifecycle_payload(event)?; + let kind = event.kind.as_u16() as u32; + let run = if (46010..=46012).contains(&kind) { + payload.get("run").cloned().unwrap_or(payload) + } else { + payload + }; + (run.get("id").and_then(Value::as_str).is_some() + && run.get("workflow_id").and_then(Value::as_str).is_some()) + .then_some(run) +} + +fn lifecycle_precedence(kind: u32) -> u8 { + match kind { + 46005..=46007 => 4, + 46011..=46012 => 3, + 46010 => 2, + 46001 => 1, + _ => 0, + } +} + // ── Reads ──────────────────────────────────────────────────────────────────── #[tauri::command] @@ -121,26 +152,46 @@ pub async fn get_workflow( pub async fn get_workflow_runs( workflow_id: String, limit: Option, - _state: State<'_, AppState>, + state: State<'_, AppState>, ) -> Result, String> { - // TODO(workflow-runs): Run reconstruction is a clearly-scoped follow-up. - // The authoritative run record the frontend's `WorkflowRun` shape needs - // (status / current_step / execution_trace / error_message) lives in the - // relay DB and is not exposed to the desktop client as a single queryable - // record. If the relay starts emitting lifecycle events (46001–46007, …), - // folding that stream into `WorkflowRun` would be another viable design. - // The important bit for this command is that raw lifecycle events are not - // the `RawWorkflowRun` contract. - // - // Until then we return a bare empty array — NOT a raw-event wrapper. The - // frontend wrapper (`getWorkflowRuns`) does `raw.map(fromRawWorkflowRun)`, - // so it must receive an array; the wrapped `{ runs: [...] }` shape would - // make `.map()` throw and crash the detail panel (the same TypeError class - // as the original page bug). Raw lifecycle events also don't carry the - // `id`/`workflow_id`/`status`/… fields `RawWorkflowRun` expects, so an - // empty list is the honest, safe placeholder. - let _ = (workflow_id, limit); - Ok(Vec::new()) + let requested_limit = limit.unwrap_or(20).min(100); + let events = query_relay( + &state, + &[serde_json::json!({ + "kinds": [46001, 46005, 46006, 46007, 46010, 46011, 46012], + "#d": [workflow_id], + "limit": 500 + })], + ) + .await?; + + let mut latest: HashMap = HashMap::new(); + for event in events { + let Some(run) = lifecycle_run_payload(&event) else { + continue; + }; + if run.get("workflow_id").and_then(Value::as_str) != Some(workflow_id.as_str()) { + continue; + } + let Some(run_id) = run.get("id").and_then(Value::as_str) else { + continue; + }; + let created_at = event.created_at.as_secs(); + let precedence = lifecycle_precedence(event.kind.as_u16() as u32); + if latest + .get(run_id) + .is_none_or(|(seen_at, seen_precedence, _)| { + (created_at, precedence) >= (*seen_at, *seen_precedence) + }) + { + latest.insert(run_id.to_owned(), (created_at, precedence, run)); + } + } + + let mut runs: Vec<(u64, u8, Value)> = latest.into_values().collect(); + runs.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| b.1.cmp(&a.1))); + runs.truncate(requested_limit as usize); + Ok(runs.into_iter().map(|(_, _, run)| run).collect()) } // ── Writes ─────────────────────────────────────────────────────────────────── @@ -245,7 +296,13 @@ pub async fn trigger_workflow( ) -> Result { let builder = events::build_workflow_trigger(&workflow_id)?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + let response = parse_command_response::(&result.message).unwrap_or(Value::Null); + Ok(serde_json::json!({ + "event_id": result.event_id, + "run_id": response.get("run_id").and_then(Value::as_str).unwrap_or(""), + "workflow_id": workflow_id, + "status": "pending", + })) } // ── Approvals ──────────────────────────────────────────────────────────────── @@ -254,15 +311,63 @@ pub async fn trigger_workflow( pub async fn get_run_approvals( workflow_id: String, run_id: String, - _state: State<'_, AppState>, + state: State<'_, AppState>, ) -> Result, String> { - // TODO(workflow-runs): Like runs (see `get_workflow_runs`), reconstructing - // approvals into the frontend's `WorkflowApproval` shape from lifecycle - // events (46010/46011/46012) is a clearly-scoped follow-up tracked under - // TODO(workflow-runs). Return a bare empty array so the frontend's - // `getRunApprovals` (`raw.map(fromRawApproval)`) is safe. - let _ = (workflow_id, run_id); - Ok(Vec::new()) + let mut events = query_relay( + &state, + &[serde_json::json!({ + "kinds": [46010, 46011, 46012], + "#d": [workflow_id], + "limit": 500 + })], + ) + .await?; + events.sort_by_key(|event| event.created_at.as_secs()); + + let mut latest: HashMap = HashMap::new(); + for event in events { + let Some(mut approval) = lifecycle_payload(&event) else { + continue; + }; + if approval.get("workflow_id").and_then(Value::as_str) != Some(workflow_id.as_str()) + || approval.get("run_id").and_then(Value::as_str) != Some(run_id.as_str()) + { + continue; + } + let Some(token) = approval + .get("token") + .and_then(Value::as_str) + .map(str::to_owned) + else { + continue; + }; + let created_at = event.created_at.as_secs(); + let precedence = match approval.get("status").and_then(Value::as_str) { + Some("granted" | "denied") => 2, + Some("pending") => 1, + _ => 0, + }; + if approval.get("message").is_none() { + if let Some((_, _, previous)) = latest.get(&token) { + if let Some(message) = previous.get("message") { + approval["message"] = message.clone(); + } + } + } + let should_replace = latest.get(&token).is_none_or( + |(seen_at, seen_precedence, _)| { + (created_at, precedence) >= (*seen_at, *seen_precedence) + }, + ); + if should_replace { + latest.insert(token, (created_at, precedence, approval)); + } + } + + Ok(latest + .into_values() + .map(|(_, _, approval)| approval) + .collect()) } #[tauri::command] @@ -273,7 +378,14 @@ pub async fn grant_approval( ) -> Result { let builder = events::build_approval_grant(&token, note.as_deref())?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + let response = parse_command_response::(&result.message).unwrap_or(Value::Null); + Ok(serde_json::json!({ + "event_id": result.event_id, + "token": token, + "status": response.get("status").and_then(Value::as_str).unwrap_or("granted"), + "run_id": response.get("run_id").and_then(Value::as_str).unwrap_or(""), + "workflow_id": "", + })) } #[tauri::command] @@ -284,7 +396,14 @@ pub async fn deny_approval( ) -> Result { let builder = events::build_approval_deny(&token, note.as_deref())?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + let response = parse_command_response::(&result.message).unwrap_or(Value::Null); + Ok(serde_json::json!({ + "event_id": result.event_id, + "token": token, + "status": response.get("status").and_then(Value::as_str).unwrap_or("denied"), + "run_id": response.get("run_id").and_then(Value::as_str).unwrap_or(""), + "workflow_id": "", + })) } // ── Helpers (pure, unit-tested in workflows_tests.rs) ───────────────────────── diff --git a/desktop/src-tauri/src/commands/workflows_tests.rs b/desktop/src-tauri/src/commands/workflows_tests.rs index f07f4b0f42..d019718144 100644 --- a/desktop/src-tauri/src/commands/workflows_tests.rs +++ b/desktop/src-tauri/src/commands/workflows_tests.rs @@ -189,16 +189,37 @@ fn workflow_wire_serializes_with_snake_case_keys() { } #[test] -fn runs_and_approvals_serialize_to_bare_empty_array() { - // Regression guard for the crash class this fix closed. The frontend - // wrappers `getWorkflowRuns` / `getRunApprovals` do `raw.map(...)`, so the - // Rust side MUST return a bare JSON array. A wrapped `{ runs: [...] }` / - // `{ approvals: [...] }` shape would make `.map()` throw and crash the - // detail panel — the same TypeError class as the original page bug. - // - // The commands take `State`, so we can't invoke them directly in - // a unit test; instead we pin the exact value they return (`Vec::new()` of - // their `Vec` element type) and assert its serialized shape. +fn lifecycle_run_payload_reads_waiting_approval_snapshot() { + let keys = Keys::generate(); + let event = EventBuilder::new( + Kind::Custom(46010), + serde_json::json!({ + "token": "a".repeat(64), + "workflow_id": WF, + "run_id": "33333333-3333-3333-3333-333333333333", + "run": { + "id": "33333333-3333-3333-3333-333333333333", + "workflow_id": WF, + "status": "waiting_approval", + "current_step": 0, + "execution_trace": [{"step_id": "gate", "status": "waiting_approval"}], + "created_at": 100 + } + }).to_string(), + ) + .tags([Tag::parse(["d", WF]).expect("tag")]) + .sign_with_keys(&keys) + .expect("sign"); + + let run = lifecycle_run_payload(&event).expect("run payload"); + assert_eq!(run.get("status").and_then(Value::as_str), Some("waiting_approval")); + assert_eq!(run.get("workflow_id").and_then(Value::as_str), Some(WF)); +} + +#[test] +fn lifecycle_commands_return_bare_arrays_for_frontend_mapping() { + // The frontend wrappers `getWorkflowRuns` / `getRunApprovals` do + // `raw.map(...)`; a wrapped object would crash the detail panel. let runs: Vec = Vec::new(); let approvals: Vec = Vec::new(); assert_eq!(serde_json::to_string(&runs).expect("serialize runs"), "[]"); diff --git a/desktop/src-tauri/src/events.rs b/desktop/src-tauri/src/events.rs index 777d56d02e..323c14f33c 100644 --- a/desktop/src-tauri/src/events.rs +++ b/desktop/src-tauri/src/events.rs @@ -137,6 +137,17 @@ fn check_pubkey(pubkey: &str) -> Result<(), String> { Ok(()) } +/// Validate a stored approval token hash reference. +fn check_token_hash(token_hash: &str) -> Result<(), String> { + if token_hash.len() != 64 || !token_hash.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(format!( + "approval token hash must be a 64-character hex string (got {} chars)", + token_hash.len() + )); + } + Ok(()) +} + // ── Channel operations ─────────────────────────────────────────────────────── /// Kind 9007 — create channel. @@ -832,15 +843,17 @@ pub fn build_workflow_trigger(workflow_id: &str) -> Result Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags)) } -/// Kind 46030 — grant an approval token (with optional note). -pub fn build_approval_grant(token: &str, note: Option<&str>) -> Result { - let tags = vec![tag(vec!["t", token])?]; +/// Kind 46030 — grant an approval token hash (with optional note). +pub fn build_approval_grant(token_hash: &str, note: Option<&str>) -> Result { + check_token_hash(token_hash)?; + let tags = vec![tag(vec!["d", &token_hash.to_ascii_lowercase()])?]; Ok(EventBuilder::new(Kind::Custom(46030), note.unwrap_or("")).tags(tags)) } -/// Kind 46031 — deny an approval token (with optional note). -pub fn build_approval_deny(token: &str, note: Option<&str>) -> Result { - let tags = vec![tag(vec!["t", token])?]; +/// Kind 46031 — deny an approval token hash (with optional note). +pub fn build_approval_deny(token_hash: &str, note: Option<&str>) -> Result { + check_token_hash(token_hash)?; + let tags = vec![tag(vec!["d", &token_hash.to_ascii_lowercase()])?]; Ok(EventBuilder::new(Kind::Custom(46031), note.unwrap_or("")).tags(tags)) } diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 2847b87877..c5b607d36a 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -758,6 +758,9 @@ pub fn run() { resolve_oa_owner, list_relay_agents, list_managed_agents, + list_external_agent_runtimes, + register_external_agent_runtime, + archive_external_agent_runtime, list_managed_agent_runtimes, start_managed_agent_runtime, stop_managed_agent_runtime, diff --git a/desktop/src-tauri/src/managed_agents/external_runtimes.rs b/desktop/src-tauri/src/managed_agents/external_runtimes.rs new file mode 100644 index 0000000000..e74821e8ac --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/external_runtimes.rs @@ -0,0 +1,521 @@ +//! Durable registration for agent runtimes owned by another launcher. +//! +//! An external runtime is deliberately not a ManagedAgentRecord. It has no +//! private key, process handle, provider configuration, or start path in Buzz +//! Desktop. The registry is provenance and operating-contract data only: +//! registering an entry can never start a second executor. + +use std::{fs, path::PathBuf}; + +use nostr::{FromBech32, PublicKey}; +use serde::{Deserialize, Serialize}; +use tauri::AppHandle; + +use super::storage::{atomic_write_json_restricted, backup_invalid_store, managed_agents_base_dir}; + +const EXTERNAL_RUNTIMES_FILE: &str = "external-runtimes.json"; +const MAX_TEXT_LEN: usize = 256; +const MAX_CHANNELS: usize = 64; +const MAX_RATE_PER_MINUTE: u32 = 60; + +fn default_true() -> bool { + true +} + +/// The external runner's durable, non-secret operating contract. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExternalAgentRuntime { + /// The public key of the already-running agent. Buzz never stores its + /// nsec or any credential that could start it. + pub agent_pubkey: String, + /// Owner recovered from the verified NIP-OA attestation. + pub owner_pubkey: String, + pub name: String, + pub purpose: String, + pub deployment_scope: String, + pub runner_owner: String, + pub health_source: String, + pub shutdown_path: String, + pub allowed_channels: Vec, + /// Safety defaults are persisted with the registration so a later UI + /// cannot silently reinterpret an external agent as an unrestricted bot. + pub mention_only: bool, + pub mention_filter: bool, + pub rate_limit_per_minute: u32, + pub retirement_date: String, + /// Archiving preserves the register and history. It is not a relay + /// membership revocation and does not claim that the runner stopped. + pub archived: bool, + /// The NIP-OA conditions are public provenance metadata. The reusable + /// auth tag and its signature are intentionally not persisted here. + pub attestation_conditions: String, + pub created_at: String, + pub updated_at: String, +} + +/// Input for the explicit external-runtime registration command. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RegisterExternalAgentRuntimeRequest { + pub agent_pubkey: String, + /// A NIP-OA auth tag signed by the current workspace owner for + /// agent_pubkey. It is verified and then discarded. + pub owner_auth_tag: String, + pub name: String, + pub purpose: String, + pub deployment_scope: String, + pub runner_owner: String, + pub health_source: String, + pub shutdown_path: String, + pub allowed_channels: Vec, + #[serde(default = "default_true")] + pub mention_only: bool, + #[serde(default = "default_true")] + pub mention_filter: bool, + pub rate_limit_per_minute: u32, + pub retirement_date: String, +} + +/// Why an external registration cannot be activated. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ExternalRuntimeConflict { + /// The same identity is already registered in this deployment scope. + SameScope, + /// The same identity is registered in another live deployment scope. + DifferentScope { existing_scope: String }, +} + +impl ExternalRuntimeConflict { + /// Render the high-signal duplicate warning shown to the owner. + pub fn message(&self, agent_pubkey: &str) -> String { + match self { + Self::SameScope => format!( + "external agent {agent_pubkey} is already registered in this deployment scope" + ), + Self::DifferentScope { existing_scope } => format!( + "external agent {agent_pubkey} is already active in deployment scope '{existing_scope}'; one identity may have only one live runner scope" + ), + } + } +} + +/// Normalize a public key supplied as either hex or npub. +pub fn normalize_public_key(value: &str, label: &str) -> Result { + let trimmed = value.trim(); + if trimmed.is_empty() { + return Err(format!("{label} is required")); + } + let key = if trimmed.starts_with("npub1") { + PublicKey::from_bech32(trimmed).map_err(|error| format!("invalid {label} npub: {error}"))? + } else { + PublicKey::from_hex(trimmed) + .map_err(|error| format!("invalid {label} hex pubkey: {error}"))? + }; + Ok(key.to_hex()) +} + +fn bounded_text(value: &str, label: &str) -> Result { + let trimmed = value.trim(); + if trimmed.is_empty() { + return Err(format!("{label} is required")); + } + if trimmed.chars().count() > MAX_TEXT_LEN { + return Err(format!( + "{label} is too long (maximum {MAX_TEXT_LEN} characters)" + )); + } + if trimmed.chars().any(char::is_control) { + return Err(format!("{label} contains a control character")); + } + Ok(trimmed.to_string()) +} + +fn normalize_channels(channels: &[String]) -> Result, String> { + if channels.is_empty() { + return Err("at least one allowed channel is required".to_string()); + } + if channels.len() > MAX_CHANNELS { + return Err(format!( + "too many allowed channels (maximum {MAX_CHANNELS})" + )); + } + let mut out = Vec::with_capacity(channels.len()); + for channel in channels { + let normalized = bounded_text(channel, "allowed channel")?; + if !out.iter().any(|existing| existing == &normalized) { + out.push(normalized); + } + } + if out.is_empty() { + return Err("at least one allowed channel is required".to_string()); + } + Ok(out) +} + +fn attestation_conditions(auth_tag: &str) -> Result { + let value: serde_json::Value = serde_json::from_str(auth_tag) + .map_err(|error| format!("owner auth tag is not valid JSON: {error}"))?; + value + .as_array() + .and_then(|parts| parts.get(2)) + .and_then(serde_json::Value::as_str) + .map(str::to_string) + .ok_or_else(|| "owner auth tag is missing its conditions".to_string()) +} + +/// Build a validated registry record. This pure constructor is also the +/// command's security boundary and is intentionally independent of disk I/O. +pub fn build_external_agent_runtime( + request: &RegisterExternalAgentRuntimeRequest, + expected_owner_pubkey: &str, + now: &str, +) -> Result { + let agent_pubkey = normalize_public_key(&request.agent_pubkey, "agent")?; + let expected_owner_pubkey = normalize_public_key(expected_owner_pubkey, "workspace owner")?; + let agent_compat = nostr::PublicKey::from_hex(&agent_pubkey) + .map_err(|error| format!("invalid agent pubkey for attestation: {error}"))?; + let attested_owner = + buzz_sdk_pkg::nip_oa::verify_auth_tag(request.owner_auth_tag.trim(), &agent_compat) + .map_err(|error| format!("owner auth tag verification failed: {error}"))? + .to_hex(); + if attested_owner != expected_owner_pubkey { + return Err("owner auth tag is not signed by the current workspace owner".to_string()); + } + if !request.mention_only || !request.mention_filter { + return Err( + "external runtimes must start in mentions-only mode with the mention filter enabled" + .to_string(), + ); + } + if !(1..=MAX_RATE_PER_MINUTE).contains(&request.rate_limit_per_minute) { + return Err(format!( + "rate limit must be between 1 and {MAX_RATE_PER_MINUTE} messages per minute" + )); + } + + Ok(ExternalAgentRuntime { + agent_pubkey, + owner_pubkey: attested_owner, + name: bounded_text(&request.name, "agent name")?, + purpose: bounded_text(&request.purpose, "agent purpose")?, + deployment_scope: bounded_text(&request.deployment_scope, "deployment scope")?, + runner_owner: bounded_text(&request.runner_owner, "runner owner")?, + health_source: bounded_text(&request.health_source, "health source")?, + shutdown_path: bounded_text(&request.shutdown_path, "shutdown path")?, + allowed_channels: normalize_channels(&request.allowed_channels)?, + mention_only: true, + mention_filter: true, + rate_limit_per_minute: request.rate_limit_per_minute, + retirement_date: bounded_text(&request.retirement_date, "retirement date")?, + archived: false, + attestation_conditions: attestation_conditions(request.owner_auth_tag.trim())?, + created_at: now.to_string(), + updated_at: now.to_string(), + }) +} + +/// Find a live duplicate before any local or relay state is changed. +pub fn find_external_runtime_conflict( + records: &[ExternalAgentRuntime], + agent_pubkey: &str, + deployment_scope: &str, +) -> Option { + records + .iter() + .filter(|record| !record.archived && record.agent_pubkey == agent_pubkey) + .map(|record| { + if record.deployment_scope == deployment_scope { + ExternalRuntimeConflict::SameScope + } else { + ExternalRuntimeConflict::DifferentScope { + existing_scope: record.deployment_scope.clone(), + } + } + }) + .next() +} + +/// The local, owner-only registry path. It is separate from +/// managed-agents.json so no existing start/reconcile code can accidentally +/// treat an external identity as a local executor. +pub fn external_runtimes_store_path(app: &AppHandle) -> Result { + Ok(managed_agents_base_dir(app)?.join(EXTERNAL_RUNTIMES_FILE)) +} + +/// Load the owner-only external-runtime register from disk. +pub fn load_external_agent_runtimes(app: &AppHandle) -> Result, String> { + let path = external_runtimes_store_path(app)?; + if !path.exists() { + return Ok(Vec::new()); + } + let content = fs::read_to_string(&path) + .map_err(|error| format!("failed to read external runtime registry: {error}"))?; + serde_json::from_str(&content).map_err(|error| { + backup_invalid_store(&path); + format!("failed to parse external runtime registry (preserved as .invalid): {error}") + }) +} + +/// Persist the external-runtime register with deterministic ordering. +pub fn save_external_agent_runtimes( + app: &AppHandle, + records: &[ExternalAgentRuntime], +) -> Result<(), String> { + let mut sorted = records.to_vec(); + sorted.sort_by(|left, right| { + left.agent_pubkey + .cmp(&right.agent_pubkey) + .then_with(|| left.deployment_scope.cmp(&right.deployment_scope)) + }); + let payload = serde_json::to_vec_pretty(&sorted) + .map_err(|error| format!("failed to serialize external runtime registry: {error}"))?; + atomic_write_json_restricted(&external_runtimes_store_path(app)?, &payload) +} + +/// The public body of the owner-signed kind:30177 provenance projection. +/// It carries only operating-contract fields; it never carries the NIP-OA auth +/// tag, nsec, provider config, or a process handle. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExternalAgentRuntimeEventContent { + pub schema_version: u8, + pub owner_pubkey: String, + pub name: String, + pub purpose: String, + pub deployment_scope: String, + pub runner_owner: String, + pub health_source: String, + pub shutdown_path: String, + pub allowed_channels: Vec, + pub mention_only: bool, + pub mention_filter: bool, + pub rate_limit_per_minute: u32, + pub retirement_date: String, + pub archived: bool, + pub attestation_conditions: String, +} + +/// Project a register entry onto the public provenance event body. +pub fn external_runtime_event_content( + record: &ExternalAgentRuntime, +) -> ExternalAgentRuntimeEventContent { + ExternalAgentRuntimeEventContent { + schema_version: 1, + owner_pubkey: record.owner_pubkey.clone(), + name: record.name.clone(), + purpose: record.purpose.clone(), + deployment_scope: record.deployment_scope.clone(), + runner_owner: record.runner_owner.clone(), + health_source: record.health_source.clone(), + shutdown_path: record.shutdown_path.clone(), + allowed_channels: record.allowed_channels.clone(), + mention_only: record.mention_only, + mention_filter: record.mention_filter, + rate_limit_per_minute: record.rate_limit_per_minute, + retirement_date: record.retirement_date.clone(), + archived: record.archived, + attestation_conditions: record.attestation_conditions.clone(), + } +} + +/// Building and signing this event never starts or contacts a runtime. +pub fn build_external_runtime_event( + owner_keys: &nostr::Keys, + record: &ExternalAgentRuntime, +) -> Result { + if owner_keys.public_key().to_hex() != record.owner_pubkey { + return Err("external runtime owner does not match signing identity".to_string()); + } + // Keep the three mandatory kind:30177 projection fields present so older + // clients can parse the event, while the external contract lives in its + // namespaced extension object. + let content = serde_json::json!({ + "name": record.name, + "parallelism": 1, + "respond_to": "owner-only", + "respond_to_allowlist": [], + "external_runtime": external_runtime_event_content(record), + }) + .to_string(); + let d_tag = nostr::Tag::parse(["d", record.agent_pubkey.as_str()]) + .map_err(|error| format!("invalid external runtime d-tag: {error}"))?; + nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core_pkg::kind::KIND_MANAGED_AGENT as u16), + content, + ) + .tags([d_tag]) + .sign_with_keys(owner_keys) + .map_err(|error| format!("failed to sign external runtime event: {error}")) +} + +/// Read and verify an external-runtime extension from a relay event. +/// +/// A kind:30177 event without the namespaced extension is a normal managed +/// agent and returns `Ok(None)`. An extension must be signed by its event +/// author and carry the requested agent identity in its `d` tag; otherwise it +/// is rejected before it can suppress a duplicate-scope warning. +pub fn external_runtime_projection_from_event( + event: &nostr::Event, + agent_pubkey: &str, +) -> Result, String> { + if event.kind.as_u16() as u32 != buzz_core_pkg::kind::KIND_MANAGED_AGENT { + return Ok(None); + } + let d_tag = event.tags.iter().find_map(|tag| { + let values: Vec<&str> = tag.as_slice().iter().map(|value| value.as_str()).collect(); + (values.first() == Some(&"d")).then(|| values.get(1).copied()).flatten() + }); + if d_tag != Some(agent_pubkey) { + return Ok(None); + } + event + .verify() + .map_err(|error| format!("external runtime provenance event failed signature verification: {error}"))?; + let value: serde_json::Value = serde_json::from_str(event.content.as_ref()) + .map_err(|error| format!("external runtime provenance event is not JSON: {error}"))?; + let Some(extension) = value.get("external_runtime") else { + return Ok(None); + }; + let projection: ExternalAgentRuntimeEventContent = + serde_json::from_value(extension.clone()).map_err(|error| { + format!("external runtime provenance extension is invalid: {error}") + })?; + if projection.owner_pubkey != event.pubkey.to_hex() { + return Err("external runtime provenance owner does not match event author".to_string()); + } + Ok(Some(projection)) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::ToBech32; + + fn request(agent: &nostr::Keys, owner: &nostr::Keys) -> RegisterExternalAgentRuntimeRequest { + let agent_compat = nostr::PublicKey::from_hex(&agent.public_key().to_hex()).unwrap(); + let owner_nsec = owner.secret_key().to_bech32().unwrap(); + let owner_compat = nostr::Keys::parse(&owner_nsec).unwrap(); + let auth = + buzz_sdk_pkg::nip_oa::compute_auth_tag(&owner_compat, &agent_compat, "").unwrap(); + RegisterExternalAgentRuntimeRequest { + agent_pubkey: agent.public_key().to_bech32().unwrap(), + owner_auth_tag: auth, + name: "Kaya".to_string(), + purpose: "External Hermes runner".to_string(), + deployment_scope: "hetzner-hermes".to_string(), + runner_owner: "hermes-gateway-selim-pro".to_string(), + health_source: "systemd + relay presence".to_string(), + shutdown_path: "systemctl stop hermes-gateway-selim-pro".to_string(), + allowed_channels: vec!["approvals".to_string(), "alerts".to_string()], + mention_only: true, + mention_filter: true, + rate_limit_per_minute: 12, + retirement_date: "2027-01-01".to_string(), + } + } + + #[test] + fn validates_owner_attestation_and_discards_auth_tag() { + let owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let input = request(&agent, &owner); + let record = build_external_agent_runtime(&input, &owner.public_key().to_hex(), "now") + .expect("valid external registration"); + assert_eq!(record.agent_pubkey, agent.public_key().to_hex()); + assert_eq!(record.owner_pubkey, owner.public_key().to_hex()); + assert!(record.attestation_conditions.is_empty()); + let serialized = serde_json::to_string(&record).unwrap(); + assert!(!serialized.contains("auth_tag")); + assert!(!serialized.contains("sig")); + } + + #[test] + fn rejects_owner_mismatch_and_unsafe_noise_defaults() { + let owner = nostr::Keys::generate(); + let other_owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let input = request(&agent, &owner); + assert!( + build_external_agent_runtime(&input, &other_owner.public_key().to_hex(), "now") + .unwrap_err() + .contains("current workspace owner") + ); + + let mut unsafe_input = request(&agent, &owner); + unsafe_input.mention_only = false; + assert!( + build_external_agent_runtime(&unsafe_input, &owner.public_key().to_hex(), "now") + .unwrap_err() + .contains("mentions-only") + ); + } + + #[test] + fn blocks_same_identity_across_live_scopes_but_ignores_archived_history() { + let records = vec![ExternalAgentRuntime { + agent_pubkey: "a".to_string(), + owner_pubkey: "o".to_string(), + name: "a".to_string(), + purpose: "p".to_string(), + deployment_scope: "old".to_string(), + runner_owner: "r".to_string(), + health_source: "h".to_string(), + shutdown_path: "s".to_string(), + allowed_channels: vec!["alerts".to_string()], + mention_only: true, + mention_filter: true, + rate_limit_per_minute: 1, + retirement_date: "today".to_string(), + archived: false, + attestation_conditions: String::new(), + created_at: "now".to_string(), + updated_at: "now".to_string(), + }]; + assert!(matches!( + find_external_runtime_conflict(&records, "a", "old"), + Some(ExternalRuntimeConflict::SameScope) + )); + assert!(matches!( + find_external_runtime_conflict(&records, "a", "new"), + Some(ExternalRuntimeConflict::DifferentScope { .. }) + )); + let mut archived = records; + archived[0].archived = true; + assert!(find_external_runtime_conflict(&archived, "a", "new").is_none()); + } + + #[test] + fn owner_signed_projection_carries_no_runtime_secret() { + let owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let input = request(&agent, &owner); + let record = + build_external_agent_runtime(&input, &owner.public_key().to_hex(), "now").unwrap(); + let event = build_external_runtime_event(&owner, &record).unwrap(); + assert_eq!( + event.kind.as_u16() as u32, + buzz_core_pkg::kind::KIND_MANAGED_AGENT + ); + assert!(event.content.contains("deploymentScope")); + assert!(!event.content.contains("ownerAuthTag")); + assert!(!event.content.contains("private")); + } + + #[test] + fn relay_projection_parser_rejects_wrong_identity_and_accepts_verified_extension() { + let owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let input = request(&agent, &owner); + let record = + build_external_agent_runtime(&input, &owner.public_key().to_hex(), "now").unwrap(); + let event = build_external_runtime_event(&owner, &record).unwrap(); + let other_agent = "f".repeat(64); + assert!(external_runtime_projection_from_event(&event, &other_agent) + .unwrap() + .is_none()); + let projection = + external_runtime_projection_from_event(&event, &record.agent_pubkey).unwrap(); + assert_eq!(projection.unwrap().deployment_scope, "hetzner-hermes"); + } +} diff --git a/desktop/src-tauri/src/managed_agents/mod.rs b/desktop/src-tauri/src/managed_agents/mod.rs index fe90ce430f..54bcc91322 100644 --- a/desktop/src-tauri/src/managed_agents/mod.rs +++ b/desktop/src-tauri/src/managed_agents/mod.rs @@ -14,6 +14,7 @@ pub(crate) mod custom_harnesses; mod discovery; pub(crate) mod effective_config; mod env_vars; +pub(crate) mod external_runtimes; pub(crate) mod git_bash; pub(crate) mod global_config; mod managed_node_paths; @@ -53,6 +54,7 @@ pub(crate) fn lock_path_mutex() -> std::sync::MutexGuard<'static, ()> { pub use backend::*; pub use discovery::*; pub use env_vars::*; +pub use external_runtimes::*; #[cfg(windows)] pub(crate) use git_bash::git_bash_available; pub(crate) use git_bash::{discover_git_bash, GitBashPrerequisite}; diff --git a/desktop/src/features/agents/hooks.ts b/desktop/src/features/agents/hooks.ts index b3d4c5315e..2bc1885883 100644 --- a/desktop/src/features/agents/hooks.ts +++ b/desktop/src/features/agents/hooks.ts @@ -41,10 +41,14 @@ import { } from "@/shared/api/tauri"; import type { HarnessDefinitionInput } from "@/shared/api/tauri"; import { + archiveExternalAgentRuntime, + listExternalAgentRuntimes, + registerExternalAgentRuntime, setManagedAgentAutoRestart, setManagedAgentStartOnAppLaunch, startManagedAgent, stopManagedAgent, + type RegisterExternalAgentRuntimeInput, } from "@/shared/api/tauriManagedAgents"; import { bootstrapManagedAgentRuntimePairs } from "@/features/agents/managedAgentRuntimeHooks"; import { @@ -71,6 +75,7 @@ import type { Channel, CreateManagedAgentInput, CreatePersonaInput, + ExternalAgentRuntime, ManagedAgent, UpdateManagedAgentInput, UpdatePersonaInput, @@ -107,6 +112,9 @@ export type { export const relayAgentsQueryKey = ["relay-agents"] as const; export const managedAgentsQueryKey = ["managed-agents"] as const; +export const externalAgentRuntimesQueryKey = [ + "external-agent-runtimes", +] as const; export const personasQueryKey = ["personas"] as const; export const acpRuntimesQueryKey = ["acp-runtimes"] as const; export const acpAuthMethodsQueryKey = ["acp-auth-methods"] as const; @@ -360,6 +368,52 @@ export function useManagedAgentsQuery(options?: { enabled?: boolean }) { }); } +export function useExternalAgentRuntimesQuery() { + return useQuery({ + queryKey: externalAgentRuntimesQueryKey, + queryFn: listExternalAgentRuntimes, + }); +} + +export function useRegisterExternalAgentRuntimeMutation() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (input: RegisterExternalAgentRuntimeInput) => + registerExternalAgentRuntime(input), + onSuccess: (runtime) => { + queryClient.setQueryData( + externalAgentRuntimesQueryKey, + (current) => { + const entries = current ?? []; + return [ + ...entries.filter( + (entry) => entry.agentPubkey !== runtime.agentPubkey, + ), + runtime, + ]; + }, + ); + }, + }); +} + +export function useArchiveExternalAgentRuntimeMutation() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (agentPubkey: string) => + archiveExternalAgentRuntime(agentPubkey), + onSuccess: (runtime) => { + queryClient.setQueryData( + externalAgentRuntimesQueryKey, + (current) => + (current ?? []).map((entry) => + entry.agentPubkey === runtime.agentPubkey ? runtime : entry, + ), + ); + }, + }); +} + export function useCreateManagedAgentMutation() { const queryClient = useQueryClient(); diff --git a/desktop/src/features/agents/ui/AgentsView.tsx b/desktop/src/features/agents/ui/AgentsView.tsx index 720d6e62ad..b80d38132b 100644 --- a/desktop/src/features/agents/ui/AgentsView.tsx +++ b/desktop/src/features/agents/ui/AgentsView.tsx @@ -21,6 +21,7 @@ import { TeamDeleteDialog } from "./TeamDeleteDialog"; import { TeamDialog } from "./TeamDialog"; import { TeamsSection } from "./TeamsSection"; import { UnifiedAgentsSection } from "./UnifiedAgentsSection"; +import { ExternalRuntimesSection } from "./ExternalRuntimesSection"; import { useManagedAgentActions } from "./useManagedAgentActions"; import { usePersonaActions } from "./usePersonaActions"; import { useTeamActions } from "./useTeamActions"; @@ -271,6 +272,8 @@ export function AgentsView() { }} /> + + (EMPTY_FORM); + const [formError, setFormError] = React.useState(null); + + function update(field: keyof FormState, value: string) { + setForm((current) => ({ ...current, [field]: value })); + } + + async function submit(event: React.FormEvent) { + event.preventDefault(); + setFormError(null); + const allowedChannels = form.allowedChannels + .split(/[\n,]/u) + .map((channel) => channel.trim()) + .filter(Boolean); + const rateLimit = Number.parseInt(form.rateLimitPerMinute, 10); + if (!Number.isInteger(rateLimit) || rateLimit < 1 || rateLimit > 60) { + setFormError("Rate limit must be between 1 and 60 messages per minute."); + return; + } + if (allowedChannels.length === 0) { + setFormError("Add at least one allowed channel."); + return; + } + try { + await registerMutation.mutateAsync({ + agentPubkey: form.agentPubkey, + ownerAuthTag: form.ownerAuthTag, + name: form.name, + purpose: form.purpose, + deploymentScope: form.deploymentScope, + runnerOwner: form.runnerOwner, + healthSource: form.healthSource, + shutdownPath: form.shutdownPath, + allowedChannels, + mentionOnly: true, + mentionFilter: true, + rateLimitPerMinute: rateLimit, + retirementDate: form.retirementDate, + }); + setForm(EMPTY_FORM); + setOpen(false); + } catch (error) { + setFormError(error instanceof Error ? error.message : String(error)); + } + } + + const runtimes = runtimesQuery.data ?? []; + + return ( + <> + + +
+ External runtimes + + Provenance and shutdown register only. Registration never imports + a key or starts a Buzz executor. + +
+ + + + {runtimesQuery.isLoading ? ( +

Loading register…

+ ) : runtimesQuery.error instanceof Error ? ( +

+ {runtimesQuery.error.message} +

+ ) : runtimes.length === 0 ? ( +

+ No external runner is registered on this device. +

+ ) : ( + runtimes.map((runtime) => ( + { + void archiveMutation.mutateAsync(runtime.agentPubkey); + }} + /> + )) + )} + + + + { + if (!registerMutation.isPending) setOpen(nextOpen); + }} + > + + + Register an external runner + + Paste the owner-signed NIP-OA tag for this public identity. Buzz + stores only the verified owner and operating contract; it never + receives the runner key. + + +
+
+ update("agentPubkey", value)} + /> + update("deploymentScope", value)} + /> + update("name", value)} + /> + update("runnerOwner", value)} + /> + update("healthSource", value)} + /> + update("shutdownPath", value)} + /> + update("rateLimitPerMinute", value)} + /> + update("retirementDate", value)} + /> +
+ update("purpose", value)} + /> + update("allowedChannels", value)} + /> +