From 9ca9d8ccbcbdc8e766b6a8716d05ab778f046e60 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 21 Aug 2026 21:22:14 +0000 Subject: [PATCH] fix(workflow): complete managed owner updates Co-authored-by: Larry Signed-off-by: Logan Johnson --- crates/buzz-acp/src/lib.rs | 159 ++++ crates/buzz-db/src/lib.rs | 130 +++ crates/buzz-db/src/workflow.rs | 856 +++++++++++++++++- crates/buzz-relay/src/api/workflows.rs | 389 +++++++- .../src/handlers/command_executor.rs | 105 +-- crates/buzz-relay/src/main.rs | 4 +- crates/buzz-relay/src/router.rs | 8 + crates/buzz-relay/src/webhook_secret.rs | 59 ++ migrations/0035_workflow_owner_commands.sql | 1 + schema/schema.sql | 1 + 10 files changed, 1640 insertions(+), 72 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index e16bc3024a..dd1401bae1 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -23,6 +23,7 @@ use anyhow::{ensure, Context, Result}; use buzz_core::kind::{ KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_WORKFLOW_DEF, }; use buzz_core::observer::{ decrypt_observer_payload, encrypt_observer_payload, OBSERVER_FRAME_TELEMETRY, @@ -314,6 +315,139 @@ struct ClaimedWorkflowDelivery { trigger_context: Option, } +#[derive(Clone, Debug, serde::Deserialize)] +struct PendingWorkflowOwnerCommand { + command_id: Uuid, + event_id: String, + owner_pubkey: String, + agent_pubkey: String, + workflow_id: Uuid, + channel_id: Uuid, + expected_revision: String, + proposed_yaml: String, +} + +#[derive(Clone, Debug, serde::Serialize)] +struct PendingWorkflowOwnerCompletion { + replacement_event: Option, + rejection_reason: Option, +} + +async fn list_pending_workflow_owner_commands( + rest_client: &relay::RestClient, +) -> Vec { + rest_client + .post_json("/workflows/owner-commands/pending", &serde_json::json!({})) + .await + .ok() + .and_then(|response| serde_json::from_value(response.get("commands")?.clone()).ok()) + .unwrap_or_default() +} + +fn one_exact_tag(event: &nostr::Event, name: &str, expected: &str) -> bool { + let values = exact_tags(event, name); + values.len() == 1 && values[0].as_slice().len() == 2 && values[0].as_slice()[1] == expected +} + +async fn prepare_workflow_owner_completion( + command: &PendingWorkflowOwnerCommand, + rest_client: &relay::RestClient, + immutable_owner: &str, +) -> PendingWorkflowOwnerCompletion { + let reject = |reason: &str| PendingWorkflowOwnerCompletion { + replacement_event: None, + rejection_reason: Some(reason.to_owned()), + }; + let Ok(command_event_id) = nostr::EventId::from_hex(&command.event_id) else { + return reject("invalid_command_event_id"); + }; + let Some(command_event) = query_exact_event(command_event_id, rest_client).await else { + return reject("command_event_unavailable"); + }; + let Ok(parsed) = buzz_core::workflow_owner_command::parse_owner_command(&command_event) else { + return reject("invalid_command_event"); + }; + let agent = rest_client.keys.public_key(); + let owner_matches = nostr::PublicKey::from_hex(immutable_owner) + .is_ok_and(|owner| owner == command_event.pubkey); + if !owner_matches + || command.owner_pubkey != command_event.pubkey.to_hex() + || command.agent_pubkey != agent.to_hex() + || parsed.command_id != command.command_id + || parsed.agent_pubkey != agent + || parsed.recipient != agent + || parsed.workflow_id != command.workflow_id + || parsed.expected_revision.to_hex() != command.expected_revision + || parsed.yaml_definition.as_deref() != Some(command.proposed_yaml.as_str()) + { + return reject("command_binding_mismatch"); + } + let Ok(revision) = nostr::EventId::from_hex(&command.expected_revision) else { + return reject("invalid_expected_revision"); + }; + let Some(definition) = query_exact_event(revision, rest_client).await else { + return reject("stale_workflow_revision"); + }; + if definition.kind.as_u16() as u32 != KIND_WORKFLOW_DEF + || definition.pubkey != agent + || !one_exact_tag(&definition, "d", &command.workflow_id.to_string()) + || !one_exact_tag(&definition, "h", &command.channel_id.to_string()) + { + return reject("workflow_binding_mismatch"); + } + if buzz_workflow::WorkflowEngine::parse_yaml(&command.proposed_yaml).is_err() { + return reject("invalid_workflow_yaml"); + } + let tags = [ + nostr::Tag::parse(["d", &command.workflow_id.to_string()]), + nostr::Tag::parse(["h", &command.channel_id.to_string()]), + nostr::Tag::parse(["expected-revision", &command.expected_revision]), + nostr::Tag::parse(["command", &command.command_id.to_string()]), + ]; + let Ok(tags) = tags.into_iter().collect::, _>>() else { + return reject("invalid_replacement_binding"); + }; + let Ok(event) = nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_WORKFLOW_DEF as u16), + command.proposed_yaml.clone(), + ) + .tags(tags) + .sign_with_keys(&rest_client.keys) else { + return reject("replacement_signing_failed"); + }; + PendingWorkflowOwnerCompletion { + replacement_event: Some(event), + rejection_reason: None, + } +} + +async fn reconcile_workflow_owner_completions( + rest_client: &relay::RestClient, + pending: &mut HashMap, +) { + let attempts = pending + .iter() + .map(|(command_id, completion)| (*command_id, completion.clone())) + .collect::>(); + for (command_id, completion) in attempts { + let path = format!("/workflows/owner-commands/{command_id}/complete"); + match rest_client + .post_json( + &path, + &serde_json::to_value(&completion).unwrap_or_default(), + ) + .await + { + Ok(_) => { + pending.remove(&command_id); + } + Err(error) => { + tracing::warn!(%command_id, %error, "workflow owner command completion pending reconciliation"); + } + } + } +} + async fn claim_workflow_delivery( rest_client: &relay::RestClient, delivery_id: Option, @@ -3340,6 +3474,10 @@ async fn tokio_main() -> Result<()> { // turning successful work into a future redelivery. let mut pending_workflow_finalizations: HashMap = HashMap::new(); + // Signed owner-command terminal intents survive lost HTTP responses and are + // replayed verbatim until the relay returns the stored terminal result. + let mut pending_workflow_owner_completions: HashMap = + HashMap::new(); // Native steers retain durable delivery ownership on the queued event until // the agent acknowledges success. Results that race ahead of that ack are // retained so success can finish against the exact fenced turn outcome. @@ -3643,6 +3781,27 @@ async fn tokio_main() -> Result<()> { // Remaining branches don't touch pool — evaluated when pool is idle. _ = workflow_delivery_poll.tick() => { let _ = result_rx; + reconcile_workflow_owner_completions( + &ctx.rest_client, + &mut pending_workflow_owner_completions, + ).await; + if let Some(immutable_owner) = owner_cache.get() { + for command in list_pending_workflow_owner_commands(&ctx.rest_client).await { + if pending_workflow_owner_completions.contains_key(&command.command_id) { + continue; + } + let completion = prepare_workflow_owner_completion( + &command, + &ctx.rest_client, + immutable_owner, + ).await; + pending_workflow_owner_completions.insert(command.command_id, completion); + } + } + reconcile_workflow_owner_completions( + &ctx.rest_client, + &mut pending_workflow_owner_completions, + ).await; if let Some(delivery) = claim_workflow_delivery( &ctx.rest_client, None, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f24ed0f250..b41cb97057 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3882,6 +3882,69 @@ impl Db { .await } + /// Insert or update a workflow within a caller-owned transaction. + #[datastore_span(name = "upsert_workflow_in_transaction", system = "postgresql")] + #[allow(clippy::too_many_arguments)] + pub async fn upsert_workflow_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], + definition_event_id: &[u8], + enabled: bool, + ) -> Result<()> { + workflow::upsert_workflow_in_transaction( + tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + definition_event_id, + enabled, + ) + .await + } + + /// Persist a signed workflow definition and executable projection in one transaction. + #[allow(clippy::too_many_arguments)] + pub async fn persist_workflow_definition_in_transaction( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + channel_id: Uuid, + owner_pubkey: &[u8], + event: &nostr::Event, + event_persistence: workflow::WorkflowDefinitionEventPersistence<'_>, + name: &str, + definition_json: &str, + definition_hash: &[u8], + enabled: bool, + ) -> Result<()> { + workflow::persist_workflow_definition_in_transaction( + tx, + community_id, + workflow_id, + channel_id, + owner_pubkey, + event, + event_persistence, + name, + definition_json, + definition_hash, + enabled, + ) + .await + } + /// Fetch a single workflow by ID, scoped to its community. #[datastore_span(name = "get_workflow", system = "postgresql")] pub async fn get_workflow( @@ -4073,6 +4136,35 @@ impl Db { workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await } + /// List pending owner update commands addressed to one managed agent. + #[datastore_span(name = "list_pending_workflow_owner_commands", system = "postgresql")] + pub async fn list_pending_workflow_owner_commands( + &self, + community_id: CommunityId, + agent_pubkey: &[u8], + limit: i64, + ) -> Result> { + workflow::list_pending_workflow_owner_commands( + &self.pool, + community_id, + agent_pubkey, + limit, + ) + .await + } + + /// Fetch one owner command for its authenticated target agent. + #[datastore_span(name = "get_workflow_owner_command", system = "postgresql")] + pub async fn get_workflow_owner_command( + &self, + community_id: CommunityId, + command_id: Uuid, + agent_pubkey: &[u8], + ) -> Result { + workflow::get_workflow_owner_command(&self.pool, community_id, command_id, agent_pubkey) + .await + } + /// Verify and record one owner-authored workflow command in the caller transaction. #[allow(clippy::too_many_arguments)] pub async fn admit_workflow_owner_command( @@ -4087,6 +4179,7 @@ impl Db { expected_revision: &[u8], operation: &str, proposed_yaml: Option<&str>, + receipt_event: Option<&nostr::Event>, ) -> Result { workflow::admit_workflow_owner_command( tx, @@ -4099,6 +4192,43 @@ impl Db { expected_revision, operation, proposed_yaml, + receipt_event, + ) + .await + } + + /// Atomically complete one pending owner-authored workflow update. + #[datastore_span(name = "complete_workflow_owner_command", system = "postgresql")] + #[allow(clippy::too_many_arguments)] + pub async fn complete_workflow_owner_command( + &self, + community_id: CommunityId, + command_id: Uuid, + agent_pubkey: &[u8], + replacement_event: Option<&nostr::Event>, + definition_name: Option<&str>, + definition_json: Option<&str>, + definition_hash: Option<&[u8]>, + definition_enabled: Option, + rejection_reason: Option<&str>, + receipt_event: &nostr::Event, + coordinate_conflict_receipt: &nostr::Event, + revision_conflict_receipt: &nostr::Event, + ) -> Result { + workflow::complete_workflow_owner_command( + &self.pool, + community_id, + command_id, + agent_pubkey, + replacement_event, + definition_name, + definition_json, + definition_hash, + definition_enabled, + rejection_reason, + receipt_event, + coordinate_conflict_receipt, + revision_conflict_receipt, ) .await } diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 196aa29e23..d0077a36c2 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -366,6 +366,38 @@ pub async fn upsert_workflow( definition_hash: &[u8], definition_event_id: &[u8], enabled: bool, +) -> Result<()> { + let mut tx = pool.begin().await?; + upsert_workflow_in_transaction( + &mut tx, + community_id, + id, + channel_id, + owner_pubkey, + name, + definition_json, + definition_hash, + definition_event_id, + enabled, + ) + .await?; + tx.commit().await?; + Ok(()) +} + +/// Insert or update a workflow on the caller's transaction. +#[allow(clippy::too_many_arguments)] +pub async fn upsert_workflow_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + id: Uuid, + channel_id: Option, + owner_pubkey: &[u8], + name: &str, + definition_json: &str, + definition_hash: &[u8], + definition_event_id: &[u8], + enabled: bool, ) -> Result<()> { let row = sqlx::query( r#" @@ -393,7 +425,7 @@ pub async fn upsert_workflow( .bind(definition_hash) .bind(definition_event_id) .bind(enabled) - .fetch_optional(pool) + .fetch_optional(&mut **tx) .await?; if row.is_none() { @@ -405,6 +437,87 @@ pub async fn upsert_workflow( Ok(()) } +/// How the shared definition persistence path handles the signed event row. +pub enum WorkflowDefinitionEventPersistence<'a> { + /// The ingest transaction already inserted and NIP-33-replaced the event. + AlreadyStored, + /// Completion must replace this exact current revision and insert the event. + ReplaceExpected(&'a [u8]), +} + +/// Persist one signed workflow definition and its executable projection together. +#[allow(clippy::too_many_arguments)] +pub async fn persist_workflow_definition_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + workflow_id: Uuid, + channel_id: Uuid, + owner_pubkey: &[u8], + event: &nostr::Event, + event_persistence: WorkflowDefinitionEventPersistence<'_>, + name: &str, + definition_json: &str, + definition_hash: &[u8], + enabled: bool, +) -> Result<()> { + if event.pubkey.as_bytes() != owner_pubkey { + return Err(DbError::AccessDenied( + "workflow definition signer does not match its projection owner".into(), + )); + } + if let WorkflowDefinitionEventPersistence::ReplaceExpected(expected_revision) = + event_persistence + { + let replaced = sqlx::query( + "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=30620 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL AND id=$4", + ) + .bind(community_id.as_uuid()) + .bind(owner_pubkey) + .bind(workflow_id.to_string()) + .bind(expected_revision) + .execute(&mut **tx) + .await? + .rows_affected(); + if replaced != 1 { + return Err(DbError::InvalidData( + "expected workflow definition event is not current".into(), + )); + } + let tags = serde_json::to_value(&event.tags) + .map_err(|error| DbError::InvalidData(error.to_string()))?; + let created_at = DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) + .ok_or_else(|| DbError::InvalidData("invalid definition timestamp".into()))?; + sqlx::query( + r#"INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id,d_tag) + VALUES ($1,$2,$3,$4,30620,$5,$6,$7,NOW(),$8,$9)"#, + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes()) + .bind(owner_pubkey) + .bind(created_at) + .bind(tags) + .bind(&event.content) + .bind(event.sig.serialize().as_slice()) + .bind(channel_id) + .bind(workflow_id.to_string()) + .execute(&mut **tx) + .await?; + } + upsert_workflow_in_transaction( + tx, + community_id, + workflow_id, + Some(channel_id), + owner_pubkey, + name, + definition_json, + definition_hash, + event.id.as_bytes(), + enabled, + ) + .await +} + /// Fetch a single workflow by ID, scoped to its community. /// /// `workflows` is keyed `(community_id, id)`; the same workflow UUID can exist @@ -839,6 +952,107 @@ pub async fn delete_workflow_for_owner( } } +/// One pending owner-authored update awaiting the target agent's signature. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingWorkflowOwnerCommand { + /// Stable command replay key. + pub command_id: Uuid, + /// Exact owner-authored command event id. + pub event_id: Vec, + /// Immutable registered owner of the target agent. + pub owner_pubkey: Vec, + /// Agent that must author the replacement definition. + pub agent_pubkey: Vec, + /// Target workflow UUID. + pub workflow_id: Uuid, + /// Exact definition revision the owner reviewed. + pub expected_revision: Vec, + /// Proposed workflow YAML. + pub proposed_yaml: String, + /// Channel containing the target workflow. + pub channel_id: Uuid, +} + +/// List pending update commands for one authenticated target agent. +pub async fn list_pending_workflow_owner_commands( + pool: &PgPool, + community_id: CommunityId, + agent_pubkey: &[u8], + limit: i64, +) -> Result> { + let rows = sqlx::query( + r#"SELECT c.command_id, c.event_id, c.owner_pubkey, c.agent_pubkey, c.workflow_id, + c.expected_revision, c.proposed_yaml, w.channel_id + FROM workflow_owner_commands c + JOIN workflows w ON w.community_id = c.community_id AND w.id = c.workflow_id + WHERE c.community_id = $1 AND c.agent_pubkey = $2 AND c.status = 'pending_agent' + ORDER BY created_at, command_id + LIMIT $3"#, + ) + .bind(community_id.as_uuid()) + .bind(agent_pubkey) + .bind(limit.clamp(1, 100)) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + Ok(PendingWorkflowOwnerCommand { + command_id: row.try_get("command_id")?, + event_id: row.try_get("event_id")?, + owner_pubkey: row.try_get("owner_pubkey")?, + agent_pubkey: row.try_get("agent_pubkey")?, + workflow_id: row.try_get("workflow_id")?, + expected_revision: row.try_get("expected_revision")?, + proposed_yaml: row.try_get("proposed_yaml")?, + channel_id: row + .try_get::, _>("channel_id")? + .ok_or_else(|| { + DbError::InvalidData( + "pending owner command workflow is not channel-scoped".into(), + ) + })?, + }) + }) + .collect() +} + +/// Fetch one owner command for its authenticated target agent, including terminal rows. +pub async fn get_workflow_owner_command( + pool: &PgPool, + community_id: CommunityId, + command_id: Uuid, + agent_pubkey: &[u8], +) -> Result { + let row = sqlx::query( + r#"SELECT c.command_id, c.event_id, c.owner_pubkey, c.agent_pubkey, c.workflow_id, + c.expected_revision, c.proposed_yaml, w.channel_id + FROM workflow_owner_commands c + JOIN workflows w ON w.community_id = c.community_id AND w.id = c.workflow_id + WHERE c.community_id = $1 AND c.command_id = $2 AND c.agent_pubkey = $3"#, + ) + .bind(community_id.as_uuid()) + .bind(command_id) + .bind(agent_pubkey) + .fetch_optional(pool) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow owner command {command_id}")))?; + Ok(PendingWorkflowOwnerCommand { + command_id: row.try_get("command_id")?, + event_id: row.try_get("event_id")?, + owner_pubkey: row.try_get("owner_pubkey")?, + agent_pubkey: row.try_get("agent_pubkey")?, + workflow_id: row.try_get("workflow_id")?, + expected_revision: row.try_get("expected_revision")?, + proposed_yaml: row.try_get("proposed_yaml")?, + channel_id: row + .try_get::, _>("channel_id")? + .ok_or_else(|| { + DbError::InvalidData("owner command workflow is not channel-scoped".into()) + })?, + }) +} + /// Relay-ingest result for a managed-workflow owner command. #[derive(Debug, Clone, PartialEq, Eq)] pub enum WorkflowOwnerCommandAdmission { @@ -865,6 +1079,7 @@ pub async fn admit_workflow_owner_command( expected_revision: &[u8], operation: &str, proposed_yaml: Option<&str>, + receipt_event: Option<&nostr::Event>, ) -> Result { let workflow = sqlx::query( "SELECT owner_pubkey, definition_event_id FROM workflows WHERE community_id = $1 AND id = $2 FOR UPDATE", @@ -899,13 +1114,18 @@ pub async fn admit_workflow_owner_command( )); } let pending = operation == "update"; + if pending != receipt_event.is_none() { + return Err(DbError::InvalidData( + "pending updates must not have a receipt; controls must have one".into(), + )); + } let inserted = sqlx::query( r#"INSERT INTO workflow_owner_commands (community_id, command_id, event_id, owner_pubkey, agent_pubkey, workflow_id, - expected_revision, operation, proposed_yaml, status, resulting_revision) + expected_revision, operation, proposed_yaml, status, resulting_revision, receipt_event_id) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9, CASE WHEN $10 THEN 'pending_agent'::workflow_owner_command_status ELSE 'applied'::workflow_owner_command_status END, - CASE WHEN $10 THEN NULL ELSE $7 END) + CASE WHEN $10 THEN NULL ELSE $7 END, $11) ON CONFLICT DO NOTHING"#, ) .bind(community_id.as_uuid()) @@ -918,15 +1138,73 @@ pub async fn admit_workflow_owner_command( .bind(operation) .bind(proposed_yaml) .bind(pending) + .bind(receipt_event.map(|event| event.id.as_bytes())) .execute(&mut **tx) .await? .rows_affected(); if inserted == 0 { - return Ok(WorkflowOwnerCommandAdmission::Duplicate); + let existing = sqlx::query( + r#"SELECT event_id, owner_pubkey, agent_pubkey, workflow_id, + expected_revision, operation, proposed_yaml + FROM workflow_owner_commands + WHERE community_id = $1 AND command_id = $2"#, + ) + .bind(community_id.as_uuid()) + .bind(command_id) + .fetch_optional(&mut **tx) + .await?; + let exact = existing.is_some_and(|row| { + row.try_get::, _>("event_id").ok().as_deref() == Some(event_id) + && row.try_get::, _>("owner_pubkey").ok().as_deref() == Some(owner_pubkey) + && row.try_get::, _>("agent_pubkey").ok().as_deref() == Some(agent_pubkey) + && row.try_get::("workflow_id").ok() == Some(workflow_id) + && row + .try_get::, _>("expected_revision") + .ok() + .as_deref() + == Some(expected_revision) + && row.try_get::("operation").ok().as_deref() == Some(operation) + && row + .try_get::, _>("proposed_yaml") + .ok() + .flatten() + .as_deref() + == proposed_yaml + }); + if exact { + return Ok(WorkflowOwnerCommandAdmission::Duplicate); + } + return Err(DbError::InvalidData( + "conflicting workflow owner command replay".into(), + )); } if !pending { + let receipt = receipt_event + .ok_or_else(|| DbError::InvalidData("control command receipt is missing".into()))?; + let tags = serde_json::to_value(&receipt.tags) + .map_err(|error| DbError::InvalidData(error.to_string()))?; + let created_at = DateTime::from_timestamp(receipt.created_at.as_secs() as i64, 0) + .ok_or_else(|| DbError::InvalidData("invalid receipt timestamp".into()))?; + sqlx::query( + r#"INSERT INTO events + (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id,d_tag) + SELECT $1,$2,$3,$4,$5,$6,$7,$8,NOW(),channel_id,$9 + FROM workflows WHERE community_id=$1 AND id=$10"#, + ) + .bind(community_id.as_uuid()) + .bind(receipt.id.as_bytes()) + .bind(receipt.pubkey.as_bytes()) + .bind(created_at) + .bind(receipt.kind.as_u16() as i32) + .bind(tags) + .bind(&receipt.content) + .bind(receipt.sig.serialize().as_slice()) + .bind(command_id.to_string()) + .bind(workflow_id) + .execute(&mut **tx) + .await?; let sql = match operation { - "enable" => "UPDATE workflows SET enabled = TRUE, updated_at = NOW() WHERE community_id = $1 AND id = $2", + "enable" => "UPDATE workflows SET enabled = TRUE, status = 'active', updated_at = NOW() WHERE community_id = $1 AND id = $2", "disable" => "UPDATE workflows SET enabled = FALSE, updated_at = NOW() WHERE community_id = $1 AND id = $2", "retire" => "UPDATE workflows SET enabled = FALSE, status = 'archived', updated_at = NOW() WHERE community_id = $1 AND id = $2", _ => { @@ -948,6 +1226,235 @@ pub async fn admit_workflow_owner_command( }) } +/// Stored terminal result of one owner command completion. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowOwnerCommandResult { + /// Terminal status (`applied` or `rejected`). + pub status: String, + /// Applied agent-authored definition revision, when successful. + pub resulting_revision: Option>, + /// Stable failure classification, when rejected. + pub terminal_reason: Option, + /// Relay-authored receipt event id. + pub receipt_event_id: Vec, + /// Whether this call performed the transition rather than replaying it. + pub transitioned: bool, +} + +/// Atomically apply or reject one pending update and persist its receipt. +#[allow(clippy::too_many_arguments)] +pub async fn complete_workflow_owner_command( + pool: &PgPool, + community_id: CommunityId, + command_id: Uuid, + agent_pubkey: &[u8], + replacement_event: Option<&nostr::Event>, + definition_name: Option<&str>, + definition_json: Option<&str>, + definition_hash: Option<&[u8]>, + definition_enabled: Option, + rejection_reason: Option<&str>, + receipt_event: &nostr::Event, + coordinate_conflict_receipt: &nostr::Event, + revision_conflict_receipt: &nostr::Event, +) -> Result { + let mut tx = pool.begin().await?; + let command = sqlx::query( + r#"SELECT event_id, owner_pubkey, agent_pubkey, workflow_id, expected_revision, + proposed_yaml, status::text AS status, resulting_revision, + terminal_reason, receipt_event_id + FROM workflow_owner_commands + WHERE community_id=$1 AND command_id=$2 FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .bind(command_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow owner command {command_id}")))?; + let stored_agent: Vec = command.try_get("agent_pubkey")?; + if stored_agent != agent_pubkey { + return Err(DbError::AccessDenied( + "owner command targets a different agent".into(), + )); + } + let status: String = command.try_get("status")?; + if status != "pending_agent" { + let stored_receipt: Option> = command.try_get("receipt_event_id")?; + let stored_receipt = stored_receipt.ok_or_else(|| { + DbError::InvalidData("terminal owner command is missing its receipt".into()) + })?; + if stored_receipt != receipt_event.id.as_bytes() { + return Err(DbError::InvalidData( + "conflicting owner command replay".into(), + )); + } + return Ok(WorkflowOwnerCommandResult { + status, + resulting_revision: command.try_get("resulting_revision")?, + terminal_reason: command.try_get("terminal_reason")?, + receipt_event_id: stored_receipt, + transitioned: false, + }); + } + + let workflow_id: Uuid = command.try_get("workflow_id")?; + let expected_revision: Vec = command.try_get("expected_revision")?; + let proposed_yaml: String = command.try_get("proposed_yaml")?; + let stored_owner: Vec = command.try_get("owner_pubkey")?; + let immutable_owner = sqlx::query_scalar::<_, bool>( + "SELECT agent_owner_pubkey = $3 FROM users WHERE community_id = $1 AND pubkey = $2 AND agent_owner_pubkey IS NOT NULL", + ) + .bind(community_id.as_uuid()) + .bind(agent_pubkey) + .bind(&stored_owner) + .fetch_optional(&mut *tx) + .await? + .unwrap_or(false); + if !immutable_owner { + return Err(DbError::AccessDenied( + "command author is no longer the immutable agent owner".into(), + )); + } + let workflow = sqlx::query( + "SELECT owner_pubkey, channel_id, definition_event_id FROM workflows WHERE community_id=$1 AND id=$2 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .fetch_one(&mut *tx) + .await?; + let workflow_owner: Vec = workflow.try_get("owner_pubkey")?; + let channel_id: Option = workflow.try_get("channel_id")?; + let current_revision: Option> = workflow.try_get("definition_event_id")?; + let completion_conflict = if workflow_owner != agent_pubkey { + Some("workflow_coordinate_changed") + } else if replacement_event.is_some() && current_revision.as_deref() != Some(&expected_revision) + { + Some("workflow_revision_changed") + } else { + None + }; + + let (terminal_status, resulting_revision, terminal_reason) = + match (completion_conflict, replacement_event) { + (Some(reason), _) => ("rejected", None, Some(reason.to_owned())), + (None, Some(event)) => { + let channel = channel_id + .ok_or_else(|| DbError::InvalidData("workflow is not channel-scoped".into()))?; + let one_tag = |name: &str, expected: &str| { + let values = event + .tags + .iter() + .filter_map(|tag| { + let values = tag.as_slice(); + (values.len() == 2 && values[0] == name).then_some(values[1].as_str()) + }) + .collect::>(); + values == [expected] + }; + if rejection_reason.is_some() + || event.verify().is_err() + || event.kind.as_u16() != 30620 + || event.pubkey.as_bytes() != agent_pubkey + || event.content != proposed_yaml + || !one_tag("d", &workflow_id.to_string()) + || !one_tag("h", &channel.to_string()) + || !one_tag("expected-revision", &hex::encode(&expected_revision)) + || !one_tag("command", &command_id.to_string()) + { + return Err(DbError::InvalidData( + "replacement does not match pending command".into(), + )); + } + let name = definition_name + .ok_or_else(|| DbError::InvalidData("missing definition name".into()))?; + let json = definition_json + .ok_or_else(|| DbError::InvalidData("missing definition JSON".into()))?; + let hash = definition_hash + .ok_or_else(|| DbError::InvalidData("missing definition hash".into()))?; + let enabled = definition_enabled + .ok_or_else(|| DbError::InvalidData("missing enabled state".into()))?; + let channel_id = channel; + + persist_workflow_definition_in_transaction( + &mut tx, + community_id, + workflow_id, + channel_id, + agent_pubkey, + event, + WorkflowDefinitionEventPersistence::ReplaceExpected(&expected_revision), + name, + json, + hash, + enabled, + ) + .await?; + ("applied", Some(event.id.as_bytes().to_vec()), None) + } + (None, None) => { + let reason = rejection_reason + .filter(|reason| !reason.is_empty()) + .ok_or_else(|| DbError::InvalidData("missing rejection reason".into()))?; + ("rejected", None, Some(reason.to_owned())) + } + }; + + let receipt_event = match completion_conflict { + Some("workflow_coordinate_changed") => coordinate_conflict_receipt, + Some("workflow_revision_changed") => revision_conflict_receipt, + Some(_) => unreachable!("completion conflicts are closed above"), + None => receipt_event, + }; + let receipt_tags = serde_json::to_value(&receipt_event.tags) + .map_err(|error| DbError::InvalidData(error.to_string()))?; + let receipt_created_at = DateTime::from_timestamp(receipt_event.created_at.as_secs() as i64, 0) + .ok_or_else(|| DbError::InvalidData("invalid receipt timestamp".into()))?; + sqlx::query( + r#"INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id,d_tag) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NOW(),$9,$10)"#, + ) + .bind(community_id.as_uuid()) + .bind(receipt_event.id.as_bytes()) + .bind(receipt_event.pubkey.as_bytes()) + .bind(receipt_created_at) + .bind(receipt_event.kind.as_u16() as i32) + .bind(receipt_tags) + .bind(&receipt_event.content) + .bind(receipt_event.sig.serialize().as_slice()) + .bind(channel_id) + .bind(command_id.to_string()) + .execute(&mut *tx) + .await?; + let terminalized = sqlx::query( + r#"UPDATE workflow_owner_commands + SET status=$3::workflow_owner_command_status, resulting_revision=$4, + terminal_reason=$5, receipt_event_id=$6, updated_at=NOW() + WHERE community_id=$1 AND command_id=$2 AND status='pending_agent'"#, + ) + .bind(community_id.as_uuid()) + .bind(command_id) + .bind(terminal_status) + .bind(resulting_revision.as_deref()) + .bind(terminal_reason.as_deref()) + .bind(receipt_event.id.as_bytes()) + .execute(&mut *tx) + .await? + .rows_affected(); + if terminalized != 1 { + return Err(DbError::InvalidData( + "workflow owner command terminalization lost its compare-and-swap".into(), + )); + } + tx.commit().await?; + Ok(WorkflowOwnerCommandResult { + status: terminal_status.into(), + resulting_revision, + terminal_reason, + receipt_event_id: receipt_event.id.as_bytes().to_vec(), + transitioned: true, + }) +} + /// Insert a new workflow run. Returns the new run's UUID. /// /// `trigger_context` is the serialized `TriggerContext` for this run. It is stored @@ -2411,7 +2918,8 @@ mod tests { let pool = setup_pool().await; let community = make_community(&pool).await; let owner = vec![0x41; 32]; - let agent = vec![0x42; 32]; + let agent_keys = nostr::Keys::generate(); + let agent = agent_keys.public_key().to_bytes().to_vec(); let stranger = vec![0x43; 32]; for key in [&owner, &agent, &stranger] { ensure_user(&pool, community, key).await.unwrap(); @@ -2435,7 +2943,32 @@ mod tests { ) .await .unwrap(); - let revision = vec![0x51; 32]; + let initial_yaml = "name: Initial\ntrigger:\n on: message_posted\nsteps: []\n"; + let initial_event = nostr::EventBuilder::new(nostr::Kind::Custom(30620), initial_yaml) + .tags([ + nostr::Tag::parse(["d", &workflow.to_string()]).unwrap(), + nostr::Tag::parse(["h", &channel.to_string()]).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(1)) + .sign_with_keys(&agent_keys) + .unwrap(); + let revision = initial_event.id.as_bytes().to_vec(); + let initial_tags = serde_json::to_value(&initial_event.tags).unwrap(); + sqlx::query( + "INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id,d_tag) \ + VALUES ($1,$2,$3,to_timestamp(1),30620,$4,$5,$6,NOW(),$7,$8)", + ) + .bind(community.as_uuid()) + .bind(initial_event.id.as_bytes()) + .bind(&agent) + .bind(initial_tags) + .bind(initial_yaml) + .bind(initial_event.sig.serialize().as_slice()) + .bind(channel) + .bind(workflow.to_string()) + .execute(&pool) + .await + .unwrap(); sqlx::query("UPDATE workflows SET definition_event_id=$3 WHERE community_id=$1 AND id=$2") .bind(community.as_uuid()) .bind(workflow) @@ -2460,6 +2993,11 @@ mod tests { command: TestCommand<'_>, ) -> Result { let mut tx = pool.begin().await?; + let receipt = (command.operation != "update").then(|| { + nostr::EventBuilder::new(nostr::Kind::Custom(46022), "{}") + .sign_with_keys(&nostr::Keys::generate()) + .expect("receipt") + }); let result = admit_workflow_owner_command( &mut tx, c, @@ -2471,6 +3009,7 @@ mod tests { rev, command.operation, command.yaml, + receipt.as_ref(), ) .await; if result.is_ok() { @@ -2478,6 +3017,7 @@ mod tests { } result } + let update_command = Uuid::new_v4(); assert_eq!( apply( &pool, @@ -2488,7 +3028,7 @@ mod tests { TestCommand { actor: &owner, operation: "update", - id: Uuid::new_v4(), + id: update_command, event: &[0x71; 32], yaml: Some("name: replacement") } @@ -2497,7 +3037,158 @@ mod tests { .unwrap(), WorkflowOwnerCommandAdmission::PendingAgent ); - let stored: Vec = sqlx::query_scalar( + let receipt_keys = nostr::Keys::generate(); + let receipt = nostr::EventBuilder::new(nostr::Kind::Custom(46022), "rejected") + .custom_created_at(nostr::Timestamp::from(1)) + .sign_with_keys(&receipt_keys) + .unwrap(); + let rejected = complete_workflow_owner_command( + &pool, + community, + update_command, + &agent, + None, + None, + None, + None, + None, + Some("webhook_secret_unavailable"), + &receipt, + &receipt, + &receipt, + ) + .await + .unwrap(); + assert_eq!(rejected.status, "rejected"); + assert!(rejected.transitioned); + let replay = complete_workflow_owner_command( + &pool, + community, + update_command, + &agent, + None, + None, + None, + None, + None, + Some("webhook_secret_unavailable"), + &receipt, + &receipt, + &receipt, + ) + .await + .unwrap(); + assert_eq!( + replay, + WorkflowOwnerCommandResult { + transitioned: false, + ..rejected.clone() + } + ); + let conflicting_receipt = nostr::EventBuilder::new(nostr::Kind::Custom(46022), "different") + .custom_created_at(nostr::Timestamp::from(1)) + .sign_with_keys(&receipt_keys) + .unwrap(); + assert!(matches!( + complete_workflow_owner_command( + &pool, + community, + update_command, + &agent, + None, + None, + None, + None, + None, + Some("different"), + &conflicting_receipt, + &conflicting_receipt, + &conflicting_receipt, + ) + .await + .unwrap_err(), + DbError::InvalidData(_) + )); + let receipt_count: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(receipt.id.as_bytes()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(receipt_count, 1); + let stored: (Vec, serde_json::Value) = sqlx::query_as( + "SELECT definition_event_id,definition FROM workflows WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(workflow) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stored.0, revision); + assert_eq!( + stored.1, + serde_json::json!({"trigger":{"on":"manual"},"steps":[]}) + ); + + let applied_command = Uuid::new_v4(); + let replacement_yaml = "name: Replacement\ntrigger:\n on: message_posted\nsteps: []\n"; + assert_eq!( + apply( + &pool, + community, + workflow, + &revision, + &agent, + TestCommand { + actor: &owner, + operation: "update", + id: applied_command, + event: &[0x74; 32], + yaml: Some(replacement_yaml), + } + ) + .await + .unwrap(), + WorkflowOwnerCommandAdmission::PendingAgent + ); + let replacement = nostr::EventBuilder::new(nostr::Kind::Custom(30620), replacement_yaml) + .tags([ + nostr::Tag::parse(["d", &workflow.to_string()]).unwrap(), + nostr::Tag::parse(["h", &channel.to_string()]).unwrap(), + nostr::Tag::parse(["expected-revision", &hex::encode(&revision)]).unwrap(), + nostr::Tag::parse(["command", &applied_command.to_string()]).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(2)) + .sign_with_keys(&agent_keys) + .unwrap(); + let applied_receipt = nostr::EventBuilder::new(nostr::Kind::Custom(46022), "applied") + .custom_created_at(nostr::Timestamp::from(2)) + .sign_with_keys(&receipt_keys) + .unwrap(); + let applied = complete_workflow_owner_command( + &pool, + community, + applied_command, + &agent, + Some(&replacement), + Some("Replacement"), + Some(r#"{"name":"Replacement","trigger":{"on":"message_posted"},"steps":[]}"#), + Some(&[0x75; 32]), + Some(true), + None, + &applied_receipt, + &applied_receipt, + &applied_receipt, + ) + .await + .unwrap(); + assert_eq!(applied.status, "applied"); + assert_eq!( + applied.resulting_revision.as_deref(), + Some(replacement.id.as_bytes().as_slice()) + ); + let current_revision: Vec = sqlx::query_scalar( "SELECT definition_event_id FROM workflows WHERE community_id=$1 AND id=$2", ) .bind(community.as_uuid()) @@ -2505,7 +3196,105 @@ mod tests { .fetch_one(&pool) .await .unwrap(); - assert_eq!(stored, revision); + assert_eq!(current_revision, replacement.id.as_bytes()); + let current_event_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30620 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(&agent) + .bind(workflow.to_string()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(current_event_count, 1); + + let stale_command = Uuid::new_v4(); + assert_eq!( + apply( + &pool, + community, + workflow, + replacement.id.as_bytes(), + &agent, + TestCommand { + actor: &owner, + operation: "update", + id: stale_command, + event: &[0x77; 32], + yaml: Some(replacement_yaml), + } + ) + .await + .unwrap(), + WorkflowOwnerCommandAdmission::PendingAgent + ); + let stale_replacement = + nostr::EventBuilder::new(nostr::Kind::Custom(30620), replacement_yaml) + .tags([ + nostr::Tag::parse(["d", &workflow.to_string()]).unwrap(), + nostr::Tag::parse(["h", &channel.to_string()]).unwrap(), + nostr::Tag::parse(["expected-revision", &replacement.id.to_hex()]).unwrap(), + nostr::Tag::parse(["command", &stale_command.to_string()]).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(3)) + .sign_with_keys(&agent_keys) + .unwrap(); + let newer_revision = vec![0x78; 32]; + sqlx::query("UPDATE workflows SET definition_event_id=$3 WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(workflow) + .bind(&newer_revision) + .execute(&pool) + .await + .unwrap(); + let proposed_receipt = nostr::EventBuilder::new(nostr::Kind::Custom(46022), "applied") + .custom_created_at(nostr::Timestamp::from(3)) + .sign_with_keys(&receipt_keys) + .unwrap(); + let stale_receipt = nostr::EventBuilder::new(nostr::Kind::Custom(46022), "stale") + .custom_created_at(nostr::Timestamp::from(3)) + .sign_with_keys(&receipt_keys) + .unwrap(); + let stale = complete_workflow_owner_command( + &pool, + community, + stale_command, + &agent, + Some(&stale_replacement), + Some("Replacement"), + Some(r#"{"name":"Replacement","trigger":{"on":"message_posted"},"steps":[]}"#), + Some(&[0x79; 32]), + Some(true), + None, + &proposed_receipt, + &stale_receipt, + &stale_receipt, + ) + .await + .unwrap(); + assert_eq!(stale.status, "rejected"); + assert_eq!( + stale.terminal_reason.as_deref(), + Some("workflow_revision_changed") + ); + assert_eq!(stale.receipt_event_id, stale_receipt.id.as_bytes()); + let unchanged: Vec = sqlx::query_scalar( + "SELECT definition_event_id FROM workflows WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(workflow) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(unchanged, newer_revision); + sqlx::query("UPDATE workflows SET definition_event_id=$3 WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(workflow) + .bind(replacement.id.as_bytes()) + .execute(&pool) + .await + .unwrap(); + let command = Uuid::new_v4(); let event = vec![0x72; 32]; assert_eq!( @@ -2513,7 +3302,7 @@ mod tests { &pool, community, workflow, - &revision, + replacement.id.as_bytes(), &agent, TestCommand { actor: &owner, @@ -2532,7 +3321,7 @@ mod tests { &pool, community, workflow, - &revision, + replacement.id.as_bytes(), &agent, TestCommand { actor: &owner, @@ -2559,7 +3348,7 @@ mod tests { &pool, community, workflow, - &revision, + replacement.id.as_bytes(), &agent, TestCommand { actor: &stranger, @@ -2594,13 +3383,13 @@ mod tests { &pool, community, workflow, - &revision, + replacement.id.as_bytes(), &agent, TestCommand { actor: &owner, operation: "enable", id: Uuid::new_v4(), - event: &[0x74; 32], + event: &[0x76; 32], yaml: None, }, ) @@ -2610,7 +3399,7 @@ mod tests { &pool, community, workflow, - &revision, + replacement.id.as_bytes(), &agent, TestCommand { actor: &owner, @@ -2623,7 +3412,40 @@ mod tests { .await .unwrap(); let row:(bool,String,Vec)=sqlx::query_as("SELECT enabled,status::text,definition_event_id FROM workflows WHERE community_id=$1 AND id=$2").bind(community.as_uuid()).bind(workflow).fetch_one(&pool).await.unwrap(); - assert_eq!(row, (false, "archived".into(), revision)); + assert_eq!( + row, + (false, "archived".into(), replacement.id.as_bytes().to_vec()) + ); + apply( + &pool, + community, + workflow, + replacement.id.as_bytes(), + &agent, + TestCommand { + actor: &owner, + operation: "enable", + id: Uuid::new_v4(), + event: &[0x79; 32], + yaml: None, + }, + ) + .await + .unwrap(); + let revived: (bool, String) = sqlx::query_as( + "SELECT enabled,status::text FROM workflows WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(workflow) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(revived, (true, "active".into())); + assert!(list_enabled_channel_workflows(&pool, community, channel) + .await + .unwrap() + .iter() + .any(|record| record.id == workflow)); } /// Confinement: a duplicate workflow UUID existing in both community A and diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index 8e8ed1001f..4b04bd6d1f 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -13,9 +13,11 @@ use axum::{ use chrono::{DateTime, Utc}; use serde::Deserialize; use serde_json::Value; +use sha2::{Digest, Sha256}; use uuid::Uuid; -use buzz_core::TenantContext; +use buzz_core::kind::{KIND_WORKFLOW_DEF, KIND_WORKFLOW_OWNER_RECEIPT}; +use buzz_core::{StoredEvent, TenantContext}; use crate::{ api::{api_error, bridge, internal_error}, @@ -280,6 +282,356 @@ async fn authorize_delivery_write( Ok((tenant, pubkey)) } +/// List pending owner update commands for the authenticated target agent. +pub async fn pending_owner_commands( + State(state): State>, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + let path = "/workflows/owner-commands/pending"; + let (tenant, agent) = authorize_delivery_write(&state, &headers, path, &body).await?; + let commands = state + .db + .list_pending_workflow_owner_commands(tenant.community(), agent.as_bytes(), 100) + .await + .map_err(|error| internal_error(&format!("list workflow owner commands: {error}")))?; + Ok(Json(serde_json::json!({ + "commands": commands.iter().map(|command| serde_json::json!({ + "command_id": command.command_id, + "event_id": hex::encode(&command.event_id), + "owner_pubkey": hex::encode(&command.owner_pubkey), + "agent_pubkey": hex::encode(&command.agent_pubkey), + "workflow_id": command.workflow_id, + "expected_revision": hex::encode(&command.expected_revision), + "proposed_yaml": command.proposed_yaml, + "channel_id": command.channel_id, + })).collect::>(), + }))) +} + +/// Agent-authenticated terminal intent for one owner command. +#[derive(Debug, Deserialize)] +pub struct CompleteOwnerCommandRequest { + replacement_event: Option, + rejection_reason: Option, +} + +pub(crate) struct OwnerCommandReceiptInput<'a> { + pub command_id: Uuid, + pub owner_pubkey: &'a [u8], + pub agent_pubkey: &'a nostr::PublicKey, + pub workflow_id: Uuid, + pub expected_revision: &'a [u8], + pub status: &'a str, + pub resulting_revision: Option<&'a str>, + pub reason: Option<&'a str>, +} + +pub(crate) fn build_owner_command_receipt( + state: &AppState, + input: OwnerCommandReceiptInput<'_>, +) -> Result { + let OwnerCommandReceiptInput { + command_id, + owner_pubkey, + agent_pubkey, + workflow_id, + expected_revision, + status, + resulting_revision, + reason, + } = input; + nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_WORKFLOW_OWNER_RECEIPT as u16), + serde_json::json!({ + "status": status, + "resulting_revision": resulting_revision, + "reason": reason, + }) + .to_string(), + ) + .tags([ + nostr::Tag::parse(["d", &command_id.to_string()]).map_err(|e| e.to_string())?, + nostr::Tag::parse(["p", &hex::encode(owner_pubkey)]).map_err(|e| e.to_string())?, + nostr::Tag::parse(["p", &agent_pubkey.to_hex()]).map_err(|e| e.to_string())?, + nostr::Tag::parse([ + "a", + &format!( + "{KIND_WORKFLOW_DEF}:{}:{workflow_id}", + agent_pubkey.to_hex() + ), + ]) + .map_err(|e| e.to_string())?, + nostr::Tag::parse(["revision", &hex::encode(expected_revision)]) + .map_err(|e| e.to_string())?, + nostr::Tag::parse(["status", status]).map_err(|e| e.to_string())?, + ]) + .custom_created_at(nostr::Timestamp::from(0)) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| e.to_string()) +} + +fn prepare_managed_definition( + name: String, + enabled: bool, + definition_json: &str, + existing: &serde_json::Value, + is_webhook: bool, +) -> Result<(String, String, Vec, bool), String> { + let mut definition: serde_json::Value = serde_json::from_str(definition_json) + .map_err(|error| format!("replacement JSON: {error}"))?; + crate::webhook_secret::prepare_definition(&mut definition, Some(existing), is_webhook, false) + .map_err(str::to_string)?; + let json = + serde_json::to_string(&definition).map_err(|error| format!("replacement JSON: {error}"))?; + let hash = Sha256::digest(json.as_bytes()).to_vec(); + Ok((name, json, hash, enabled)) +} + +/// Complete one pending owner update after agent-side verification/signing. +pub async fn complete_owner_command( + State(state): State>, + Path(command_id): Path, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/owner-commands/{command_id}/complete"); + let (tenant, agent) = authorize_delivery_write(&state, &headers, &path, &body).await?; + let request = match serde_json::from_slice::(&body) { + Ok(request) + if request.replacement_event.is_some() != request.rejection_reason.is_some() => + { + request + } + Ok(_) => CompleteOwnerCommandRequest { + replacement_event: None, + rejection_reason: Some("invalid_completion_shape".into()), + }, + Err(_) => CompleteOwnerCommandRequest { + replacement_event: None, + rejection_reason: Some("invalid_completion_json".into()), + }, + }; + + let pending = state + .db + .get_workflow_owner_command(tenant.community(), command_id, agent.as_bytes()) + .await + .map_err(|error| match error { + buzz_db::DbError::NotFound(_) => api_error( + StatusCode::NOT_FOUND, + "owner command was not found for this agent", + ), + other => internal_error(&format!("read owner command: {other}")), + })?; + let workflow = state + .db + .get_workflow(tenant.community(), pending.workflow_id) + .await + .map_err(|error| internal_error(&format!("read owner command workflow: {error}")))?; + let channel_id = workflow + .channel_id + .ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow is not channel-scoped"))?; + + let mut replacement_event = request.replacement_event; + let mut rejection_reason = request.rejection_reason; + let mut definition = None; + if let Some(event) = replacement_event.as_ref() { + let bindings_match = event.verify().is_ok() + && event.kind.as_u16() as u32 == KIND_WORKFLOW_DEF + && event.pubkey == agent + && event.content == pending.proposed_yaml + && event + .tags + .iter() + .filter(|tag| tag.as_slice().first().map(String::as_str) == Some("d")) + .filter_map(|tag| tag.content()) + .collect::>() + == [pending.workflow_id.to_string()] + && event + .tags + .iter() + .filter(|tag| tag.as_slice().first().map(String::as_str) == Some("h")) + .filter_map(|tag| tag.content()) + .collect::>() + == [channel_id.to_string()] + && event + .tags + .iter() + .filter(|tag| { + tag.as_slice().first().map(String::as_str) == Some("expected-revision") + }) + .filter_map(|tag| tag.content()) + .collect::>() + == [hex::encode(&pending.expected_revision)] + && event + .tags + .iter() + .filter(|tag| tag.as_slice().first().map(String::as_str) == Some("command")) + .filter_map(|tag| tag.content()) + .collect::>() + == [command_id.to_string()]; + if !bindings_match { + replacement_event = None; + rejection_reason = Some("replacement_binding_mismatch".into()); + } else { + match buzz_workflow::WorkflowEngine::parse_yaml(&event.content) { + Ok((def, json)) => { + let authority_ok = if def.requires_elevated_authority() { + let role = state + .db + .get_member_role(tenant.community(), channel_id, agent.as_bytes()) + .await + .map_err(|error| { + internal_error(&format!("replacement authority lookup: {error}")) + })?; + matches!(role.as_deref(), Some("owner") | Some("admin")) + } else { + true + }; + if authority_ok { + match prepare_managed_definition( + def.name.clone(), + def.enabled, + &json, + &workflow.definition, + matches!(def.trigger, buzz_workflow::TriggerDef::Webhook), + ) { + Ok(prepared) => definition = Some(prepared), + Err(reason) => { + replacement_event = None; + rejection_reason = Some(reason); + } + } + } else { + replacement_event = None; + rejection_reason = Some("insufficient_workflow_authority".into()); + } + } + Err(_) => { + replacement_event = None; + rejection_reason = Some("invalid_workflow_yaml".into()); + } + } + } + } + + let status = if replacement_event.is_some() { + "applied" + } else { + "rejected" + }; + let resulting_revision = replacement_event.as_ref().map(|event| event.id.to_hex()); + let receipt = build_owner_command_receipt( + &state, + OwnerCommandReceiptInput { + command_id, + owner_pubkey: &pending.owner_pubkey, + agent_pubkey: &agent, + workflow_id: pending.workflow_id, + expected_revision: &pending.expected_revision, + status, + resulting_revision: resulting_revision.as_deref(), + reason: rejection_reason.as_deref(), + }, + ) + .map_err(|error| internal_error(&format!("sign owner command receipt: {error}")))?; + let conflict_receipt = |reason| { + build_owner_command_receipt( + &state, + OwnerCommandReceiptInput { + command_id, + owner_pubkey: &pending.owner_pubkey, + agent_pubkey: &agent, + workflow_id: pending.workflow_id, + expected_revision: &pending.expected_revision, + status: "rejected", + resulting_revision: None, + reason: Some(reason), + }, + ) + .map_err(|error| internal_error(&format!("sign owner command conflict receipt: {error}"))) + }; + let coordinate_conflict_receipt = conflict_receipt("workflow_coordinate_changed")?; + let revision_conflict_receipt = conflict_receipt("workflow_revision_changed")?; + let result = state + .db + .complete_workflow_owner_command( + tenant.community(), + command_id, + agent.as_bytes(), + replacement_event.as_ref(), + definition.as_ref().map(|value| value.0.as_str()), + definition.as_ref().map(|value| value.1.as_str()), + definition.as_ref().map(|value| value.2.as_slice()), + definition.as_ref().map(|value| value.3), + rejection_reason.as_deref(), + &receipt, + &coordinate_conflict_receipt, + &revision_conflict_receipt, + ) + .await + .map_err(|error| match error { + buzz_db::DbError::AccessDenied(_) => api_error( + StatusCode::FORBIDDEN, + "owner command targets a different agent", + ), + buzz_db::DbError::InvalidData(_) => { + api_error(StatusCode::CONFLICT, "owner command completion conflict") + } + other => internal_error(&format!("complete owner command: {other}")), + })?; + + if result.transitioned { + if let Some(event) = replacement_event.as_ref() { + state + .workflow_engine + .invalidate_channel_workflows(tenant.community(), channel_id); + publish_internal_event(&state, &tenant, event, Some(channel_id)).await; + } + let published_receipt = if result.receipt_event_id == receipt.id.as_bytes() { + &receipt + } else if result.receipt_event_id == coordinate_conflict_receipt.id.as_bytes() { + &coordinate_conflict_receipt + } else { + &revision_conflict_receipt + }; + publish_internal_event(&state, &tenant, published_receipt, Some(channel_id)).await; + } + Ok(Json(serde_json::json!({ + "status": result.status, + "resulting_revision": result.resulting_revision.as_deref().map(hex::encode), + "reason": result.terminal_reason, + "receipt_event_id": hex::encode(result.receipt_event_id), + "transitioned": result.transitioned, + }))) +} + +pub(crate) async fn publish_internal_event( + state: &Arc, + tenant: &TenantContext, + event: &nostr::Event, + channel_id: Option, +) { + let topic = channel_id + .map(buzz_pubsub::EventTopic::Channel) + .unwrap_or(buzz_pubsub::EventTopic::Global); + state.mark_local_event(tenant.community(), &event.id); + if state + .pubsub + .publish_event(tenant, topic, event) + .await + .is_err() + { + state + .local_event_ids + .invalidate(&(tenant.community(), event.id.to_bytes())); + } + let stored = StoredEvent::new(event.clone(), channel_id); + crate::handlers::event::fan_out_event_to_local_subscribers(state, tenant.community(), &stored) + .await; +} + /// Claim the oldest due workflow delivery for the authenticated managed agent. pub async fn claim_agent_delivery( State(state): State>, @@ -494,6 +846,41 @@ mod tests { ); } + #[test] + fn managed_webhook_update_preserves_existing_secret() { + let existing = serde_json::json!({ + "name": "existing", + "_webhook_secret": "keep-me", + }); + let (_, json, hash, _) = prepare_managed_definition( + "replacement".into(), + true, + r#"{"name":"replacement"}"#, + &existing, + true, + ) + .unwrap(); + let prepared: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(prepared["_webhook_secret"], "keep-me"); + assert_eq!(hash, Sha256::digest(json.as_bytes()).to_vec()); + } + + #[test] + fn managed_transition_to_webhook_rejects_undisclosable_secret() { + let existing = serde_json::json!({"name": "existing"}); + assert_eq!( + prepare_managed_definition( + "replacement".into(), + true, + r#"{"name":"replacement"}"#, + &existing, + true, + ), + Err("webhook_secret_unavailable".into()) + ); + assert!(existing.get("_webhook_secret").is_none()); + } + #[test] fn approval_wire_does_not_expose_hash_as_token() { let approval = buzz_db::workflow::ApprovalRecord { diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 49e4e2b902..e376344ead 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -795,26 +795,17 @@ async fn handle_workflow_def( } }; - // Preserve the existing webhook secret across updates. A new secret is - // returned only when the workflow first gains a webhook trigger. - let webhook_secret = if matches!(def.trigger, buzz_workflow::TriggerDef::Webhook) { - let existing_secret = existing_workflow + // Apply the shared secret projection before hashing. This synchronous path + // can securely return a newly generated webhook secret to its caller. + let webhook_secret = webhook_secret::prepare_definition( + &mut definition_json, + existing_workflow .as_ref() - .and_then(|workflow| webhook_secret::extract_secret(&workflow.definition)); - let secret = existing_secret.unwrap_or_else(webhook_secret::generate_webhook_secret); - webhook_secret::inject_secret(&mut definition_json, &secret); - if existing_workflow - .as_ref() - .and_then(|workflow| webhook_secret::extract_secret(&workflow.definition)) - .is_none() - { - Some(secret) - } else { - None - } - } else { - None - }; + .map(|workflow| &workflow.definition), + matches!(def.trigger, buzz_workflow::TriggerDef::Webhook), + true, + ) + .map_err(|error| IngestError::Internal(format!("error: prepare webhook secret: {error}")))?; // Compute hash AFTER secret injection let definition_json_final = serde_json::to_string(&definition_json) @@ -822,7 +813,7 @@ async fn handle_workflow_def( let hash = compute_definition_hash(&definition_json_final); // Persist the command event — returns open transaction - let tx = match persist_command_event(state, tenant, event, None).await? { + let mut tx = match persist_command_event(state, tenant, event, None).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -833,7 +824,8 @@ async fn handle_workflow_def( PersistResult::Inserted(tx) => tx, }; - // 4. Execute: upsert by the NIP-33 d-tag UUID. A retry updates the same + // 4. Execute in the same locked transaction that persists the NIP-33 + // definition. A retry updates the same // row instead of creating another enabled workflow that would fan out on // every matching event. The workflow's community is the request's // server-bound tenant — never re-derived from the (client-supplied) channel @@ -853,15 +845,17 @@ async fn handle_workflow_def( state .db - .upsert_workflow( + .persist_workflow_definition_in_transaction( + &mut tx, community_id, workflow_id, - Some(channel_id), + channel_id, &self_bytes, + event, + buzz_db::workflow::WorkflowDefinitionEventPersistence::AlreadyStored, &workflow_name, &definition_json_final, &hash, - event.id.as_bytes(), def.enabled, ) .await @@ -872,17 +866,16 @@ async fn handle_workflow_def( other => IngestError::Internal(format!("error: db upsert_workflow: {other}")), })?; - // Drop the trigger-path cache entry so the new/updated definition fires on - // the next matching event instead of after the cache TTL. - state - .workflow_engine - .invalidate_channel_workflows(community_id, channel_id); - - // Commit the event transaction after the idempotent workflow upsert succeeds. + // Commit the definition event and executable workflow projection together. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; + // Drop the trigger-path cache only after both representations commit. + state + .workflow_engine + .invalidate_channel_workflows(community_id, channel_id); + // 5. Return response let mut resp = serde_json::json!({ "workflow_id": workflow_id.to_string(), @@ -928,27 +921,30 @@ async fn handle_workflow_owner_command( let channel_id = workflow .channel_id .ok_or_else(|| IngestError::Rejected("invalid: workflow is not channel-scoped".into()))?; - if workflow.owner_pubkey != command.agent_pubkey.to_bytes() - || workflow.definition_event_id.as_deref() != Some(command.expected_revision.as_bytes()) - { - return Err(IngestError::Rejected( - "conflict: workflow coordinate or revision changed".into(), - )); - } - if !state - .db - .is_agent_owner( - tenant.community(), - command.agent_pubkey.as_bytes(), - event.pubkey.as_bytes(), - ) - .await - .map_err(|error| IngestError::Internal(format!("error: owner lookup: {error}")))? - { - return Err(IngestError::Rejected( - "forbidden: command author is not the immutable agent owner".into(), - )); - } + // The workflow lookup supplies only the channel binding needed to persist and + // publish the command event. Immutable owner, coordinate author, and revision + // authority are rechecked under the workflow row lock during admission. + let control_receipt = (operation != "update") + .then(|| { + let resulting_revision = command.expected_revision.to_hex(); + crate::api::workflows::build_owner_command_receipt( + state, + crate::api::workflows::OwnerCommandReceiptInput { + command_id: command.command_id, + owner_pubkey: event.pubkey.as_bytes(), + agent_pubkey: &command.agent_pubkey, + workflow_id: command.workflow_id, + expected_revision: command.expected_revision.as_bytes(), + status: "applied", + resulting_revision: Some(&resulting_revision), + reason: None, + }, + ) + }) + .transpose() + .map_err(|error| { + IngestError::Internal(format!("error: sign owner command receipt: {error}")) + })?; let mut tx = match persist_command_event(state, tenant, event, Some(channel_id)).await? { PersistResult::Duplicate => { return Ok(IngestResult { @@ -972,6 +968,7 @@ async fn handle_workflow_owner_command( command.expected_revision.as_bytes(), operation, command.yaml_definition.as_deref(), + control_receipt.as_ref(), ) .await .map_err(|error| match error { @@ -989,6 +986,10 @@ async fn handle_workflow_owner_command( state .workflow_engine .invalidate_channel_workflows(tenant.community(), channel_id); + if let Some(receipt) = control_receipt.as_ref() { + crate::api::workflows::publish_internal_event(state, tenant, receipt, Some(channel_id)) + .await; + } let status = match admission { WorkflowOwnerCommandAdmission::PendingAgent => "pending_agent", WorkflowOwnerCommandAdmission::Applied => "applied", diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 90ba61f3d4..08cf4cda5f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -657,8 +657,8 @@ async fn main() -> anyhow::Result<()> { let wf_cron = Arc::clone(&workflow_engine); tokio::spawn(async move { wf_cron.run().await }); - // Durable managed-agent delivery reaper. Expired leases become retryable by - // the claim query; exhausted or TTL-expired rows become terminal here and + // Durable managed-agent delivery reaper. Expired leases become terminal failed + // rows and are never reassigned; TTL-expired rows are terminalized here and // emit one owner-visible channel signal from the rows returned by the // guarded UPDATE. { diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 1f076474fb..20f7590474 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -72,6 +72,14 @@ pub fn build_router(state: Arc) -> Router { .route("/events", post(api::bridge::submit_event)) .route("/query", post(api::bridge::query_events)) .route("/count", post(api::bridge::count_events)) + .route( + "/workflows/owner-commands/pending", + post(api::workflows::pending_owner_commands), + ) + .route( + "/workflows/owner-commands/{command_id}/complete", + post(api::workflows::complete_owner_command), + ) .route( "/workflows/agent-deliveries/claim", post(api::workflows::claim_agent_delivery), diff --git a/crates/buzz-relay/src/webhook_secret.rs b/crates/buzz-relay/src/webhook_secret.rs index fc73e1f830..6f606b2483 100644 --- a/crates/buzz-relay/src/webhook_secret.rs +++ b/crates/buzz-relay/src/webhook_secret.rs @@ -68,6 +68,32 @@ pub fn strip_secret(def: &serde_json::Value) -> serde_json::Value { } } +/// Apply webhook-secret projection semantics before hashing and persistence. +/// +/// Existing webhook workflows retain their secret. Callers that can securely +/// return a newly minted secret may set `allow_generate`; background owner +/// command completion must not mint an undiscoverable credential. +pub fn prepare_definition( + definition: &mut serde_json::Value, + existing: Option<&serde_json::Value>, + is_webhook: bool, + allow_generate: bool, +) -> Result, &'static str> { + if !is_webhook { + return Ok(None); + } + if let Some(secret) = existing.and_then(extract_secret) { + inject_secret(definition, &secret); + return Ok(None); + } + if !allow_generate { + return Err("webhook_secret_unavailable"); + } + let secret = generate_webhook_secret(); + inject_secret(definition, &secret); + Ok(Some(secret)) +} + /// Compare `provided` against `stored` in constant time. /// /// Returns `true` only when the two strings are identical. The XOR-fold @@ -139,6 +165,39 @@ mod tests { assert_eq!(stripped.get("b").and_then(|v| v.as_i64()), Some(2)); } + #[test] + fn prepare_definition_preserves_existing_secret() { + let existing = serde_json::json!({"_webhook_secret": "keep-me"}); + let mut replacement = serde_json::json!({"name": "replacement"}); + assert_eq!( + prepare_definition(&mut replacement, Some(&existing), true, false), + Ok(None) + ); + assert_eq!(extract_secret(&replacement).as_deref(), Some("keep-me")); + } + + #[test] + fn prepare_definition_rejects_undeliverable_new_secret() { + let mut replacement = serde_json::json!({"name": "replacement"}); + assert_eq!( + prepare_definition(&mut replacement, None, true, false), + Err("webhook_secret_unavailable") + ); + assert!(extract_secret(&replacement).is_none()); + } + + #[test] + fn prepare_definition_generates_when_caller_can_return_secret() { + let mut replacement = serde_json::json!({"name": "replacement"}); + let generated = prepare_definition(&mut replacement, None, true, true) + .unwrap() + .expect("new secret"); + assert_eq!( + extract_secret(&replacement).as_deref(), + Some(generated.as_str()) + ); + } + #[test] fn verify_secret_matches() { assert!(verify_secret("hello-world", "hello-world")); diff --git a/migrations/0035_workflow_owner_commands.sql b/migrations/0035_workflow_owner_commands.sql index 6adfc7a1bc..2e4aedc7e7 100644 --- a/migrations/0035_workflow_owner_commands.sql +++ b/migrations/0035_workflow_owner_commands.sql @@ -13,6 +13,7 @@ CREATE TABLE workflow_owner_commands ( proposed_yaml TEXT, status workflow_owner_command_status NOT NULL, resulting_revision BYTEA, + receipt_event_id BYTEA, terminal_reason TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), diff --git a/schema/schema.sql b/schema/schema.sql index b5c001ad18..2f991e659b 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1776,6 +1776,7 @@ CREATE TABLE workflow_owner_commands ( proposed_yaml TEXT, status workflow_owner_command_status NOT NULL, resulting_revision BYTEA, + receipt_event_id BYTEA, terminal_reason TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),