diff --git a/crates/buzz-db/src/deletion.rs b/crates/buzz-db/src/deletion.rs index 98a039d62f..d39bb2b8aa 100644 --- a/crates/buzz-db/src/deletion.rs +++ b/crates/buzz-db/src/deletion.rs @@ -80,6 +80,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "thread_metadata", "users", "workflow_approvals", + "workflow_owner_commands", "workflow_runs", "workflows", ]; @@ -87,6 +88,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ "workflow_approvals", + "workflow_owner_commands", "scheduled_workflow_fires", "workflow_runs", "push_wake_outbox", diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index fcf4058079..27f3fe51e8 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3544,6 +3544,67 @@ impl Db { workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await } + /// Resolve the exact managed-workflow coordinate visible to its immutable human owner. + #[datastore_span(name = "get_workflow_owner_target", system = "postgresql")] + pub async fn get_workflow_owner_target( + &self, + community_id: CommunityId, + workflow_id: Uuid, + owner_pubkey: &[u8], + ) -> Result { + workflow::get_workflow_owner_target(&self.pool, community_id, workflow_id, owner_pubkey) + .await + } + + /// Admit one parsed owner-signed command into private durable state. + #[datastore_span(name = "admit_workflow_owner_command", system = "postgresql")] + pub async fn admit_workflow_owner_command( + &self, + community_id: CommunityId, + command: &buzz_core::workflow_owner_command::WorkflowOwnerCommand, + ) -> Result { + workflow::admit_workflow_owner_command(&self.pool, community_id, command).await + } + + /// List pending owner commands for one authenticated target 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 only for its immutable owner or 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, + principal_pubkey: &[u8], + ) -> Result { + workflow::get_workflow_owner_command(&self.pool, community_id, command_id, principal_pubkey) + .await + } + + /// Apply one exact agent-signed terminal result with compare-and-swap semantics. + #[datastore_span(name = "complete_workflow_owner_command", system = "postgresql")] + pub async fn complete_workflow_owner_command( + &self, + community_id: CommunityId, + result: &buzz_core::workflow_owner_command::WorkflowOwnerResult, + ) -> Result { + workflow::complete_workflow_owner_command(&self.pool, community_id, result).await + } + /// Find a workflow by owner pubkey and name within a community. Used for /// NIP-09 a-tag deletion where the d-tag is the workflow name (not UUID). #[datastore_span(name = "find_workflow_by_owner_and_name", system = "postgresql")] diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 1508197a79..a7bc0e2bae 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -645,7 +645,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 34); + assert_eq!(migrations.len(), 35); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1087,6 +1087,14 @@ mod tests { assert!(run_revision.contains("ADD COLUMN definition_event_id BYTEA")); assert!(run_revision.contains("octet_length(definition_event_id) = 32")); + // Owner commands are private durable state, fenced with their community + // and bound to an exact workflow revision. + assert_eq!(migrations[34].version, 35); + let owner_commands = migrations[34].sql.as_str(); + assert!(owner_commands.contains("CREATE TABLE workflow_owner_commands")); + assert!(owner_commands.contains("expected_revision BYTEA NOT NULL")); + assert!(owner_commands.contains("attach_community_write_fence('workflow_owner_commands')")); + // Fresh desired-state bootstrap must install the identical executable // fence as migration 0032. CI and isolated relay startup use schema.sql // without running migrations, so drift reopens rolling-deploy races. @@ -1559,9 +1567,16 @@ mod tests { "schema.sql is missing operator-global registry row {row:?}" ); } + // Recovery policy intentionally excludes two deployment-private tables. + // Later additive migrations attach newly introduced community-scoped + // tables to the same fence, so include those targets in the expected + // desired-state surface. let mut expected_fences = migration.fence_attachments.clone(); expected_fences.remove("product_feedback"); expected_fences.remove("rate_limit_violations"); + for migration in MIGRATOR.iter().filter(|migration| migration.version > 29) { + expected_fences.extend(surface(migration.sql.as_ref()).fence_attachments); + } assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index b06352c333..4065cb4d83 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -15,6 +15,9 @@ use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row}; use uuid::Uuid; +use buzz_core::workflow_owner_command::{ + WorkflowOwnerCommand, WorkflowOwnerOperation, WorkflowOwnerResult, WorkflowOwnerResultStatus, +}; use buzz_core::CommunityId; use crate::error::{DbError, Result}; @@ -829,6 +832,407 @@ pub async fn delete_workflow_for_owner( } } +// -- Workflow owner command state --------------------------------------------- + +/// Durable state of an owner-signed workflow command. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkflowOwnerCommandStatus { + /// Awaiting the target agent's signed terminal result. + Pending, + /// The target agent reported that it applied the operation. + Applied, + /// The target agent rejected the operation. + Rejected, +} + +impl FromStr for WorkflowOwnerCommandStatus { + type Err = DbError; + + fn from_str(value: &str) -> std::result::Result { + match value { + "pending" => Ok(Self::Pending), + "applied" => Ok(Self::Applied), + "rejected" => Ok(Self::Rejected), + other => Err(DbError::InvalidData(format!( + "unknown workflow owner command status: {other}" + ))), + } + } +} + +/// Private command record visible only to its immutable owner or target agent. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowOwnerCommandRecord { + /// Server-bound tenant containing the command. + pub community_id: CommunityId, + /// Stable replay identity from the signed `d` tag. + pub command_id: Uuid, + /// Monotonic insertion order used for target discovery. + pub command_sequence: i64, + /// Exact owner-signed command event. + pub event_id: Vec, + /// Immutable human owner that signed the command. + pub owner_pubkey: Vec, + /// Agent that owns the workflow and executes the command. + pub agent_pubkey: Vec, + /// Workflow UUID from the signed coordinate. + pub workflow_id: Uuid, + /// Exact signed definition revision reviewed by the owner. + pub expected_revision: Vec, + /// Requested operation. + pub operation: WorkflowOwnerOperation, + /// Current persistence state. + pub status: WorkflowOwnerCommandStatus, + /// Exact agent-signed result event, after terminalization. + pub result_event_id: Option>, + /// Optional terminal context from the signed result. + pub result_reason: Option, +} + +/// Result of admitting an owner command. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkflowOwnerCommandAdmission { + /// A new pending row was created. + Inserted(WorkflowOwnerCommandRecord), + /// The exact signed command was already present. + Duplicate(WorkflowOwnerCommandRecord), +} + +/// Result of applying an agent-signed terminal result. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkflowOwnerResultTransition { + /// This call performed the pending-to-terminal transition. + Applied(WorkflowOwnerCommandRecord), + /// The exact terminal result had already been applied. + Replay(WorkflowOwnerCommandRecord), +} + +/// Minimal workflow coordinate returned only to its immutable human owner. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowOwnerTarget { + /// Agent author of the workflow definition. + pub agent_pubkey: Vec, + /// Exact current kind:30620 event id. + pub expected_revision: Vec, +} + +/// Resolve an owner-management target without granting channel-content access. +/// +/// Unauthorized and absent workflows deliberately share [`DbError::NotFound`] +/// so this private discovery path does not disclose workflow existence. +pub async fn get_workflow_owner_target( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + owner_pubkey: &[u8], +) -> Result { + let row = sqlx::query( + r#"SELECT w.owner_pubkey AS agent_pubkey, w.definition_event_id AS expected_revision + FROM workflows w + JOIN users u ON u.community_id = w.community_id AND u.pubkey = w.owner_pubkey + WHERE w.community_id = $1 AND w.id = $2 + AND u.agent_owner_pubkey = $3 + AND w.definition_event_id IS NOT NULL"#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(owner_pubkey) + .fetch_optional(pool) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow {workflow_id}")))?; + Ok(WorkflowOwnerTarget { + agent_pubkey: row.try_get("agent_pubkey")?, + expected_revision: row.try_get("expected_revision")?, + }) +} + +/// Admit one owner-signed command after immutable-owner authorization. +/// +/// The immutable owner lookup occurs before the transaction opens, avoiding a +/// second pool acquisition while a writer connection is held. The workflow row +/// then serializes revision checks and command order. At most one command per +/// workflow may remain pending, so operations cannot be observed out of order. +pub async fn admit_workflow_owner_command( + pool: &PgPool, + community_id: CommunityId, + command: &WorkflowOwnerCommand, +) -> Result { + let owner = command.owner_pubkey.to_bytes(); + let agent = command.agent_pubkey.to_bytes(); + if !crate::user::is_agent_owner(pool, community_id, &agent, &owner).await? { + return Err(DbError::AccessDenied( + "command author is not the immutable agent owner".into(), + )); + } + + let mut tx = pool.begin().await?; + let workflow = sqlx::query( + r#"SELECT owner_pubkey, definition_event_id + FROM workflows + WHERE community_id = $1 AND id = $2 + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .bind(command.workflow_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow {}", command.workflow_id)))?; + let workflow_owner: Vec = workflow.try_get("owner_pubkey")?; + if workflow_owner != agent { + return Err(DbError::AccessDenied( + "workflow coordinate author mismatch".into(), + )); + } + + // Resolve replay identity before revalidating the mutable current revision. + // An exact retry remains an acknowledgement after the workflow advances; + // the recorded signed command, not today's revision, defines that replay. + let existing = sqlx::query( + r#"SELECT community_id, command_id, command_sequence, event_id, owner_pubkey, + agent_pubkey, workflow_id, expected_revision, operation, + status::text AS status, result_event_id, result_reason + FROM workflow_owner_commands + WHERE community_id = $1 AND (command_id = $2 OR event_id = $3) + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .bind(command.command_id) + .bind(command.event_id.as_bytes()) + .fetch_optional(&mut *tx) + .await?; + if let Some(row) = existing { + let record = row_to_workflow_owner_command(row)?; + if command_matches_record(command, &record) { + tx.commit().await?; + return Ok(WorkflowOwnerCommandAdmission::Duplicate(record)); + } + return Err(DbError::InvalidData( + "conflicting workflow owner command replay".into(), + )); + } + + let revision: Option> = workflow.try_get("definition_event_id")?; + if revision.as_deref() != Some(command.expected_revision.as_bytes()) { + return Err(DbError::InvalidData("workflow revision conflict".into())); + } + + let row = sqlx::query( + r#"INSERT INTO workflow_owner_commands + (community_id, command_id, event_id, owner_pubkey, agent_pubkey, + workflow_id, expected_revision, operation) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8) + RETURNING community_id, command_id, command_sequence, event_id, + owner_pubkey, agent_pubkey, workflow_id, expected_revision, + operation, status::text AS status, result_event_id, result_reason"#, + ) + .bind(community_id.as_uuid()) + .bind(command.command_id) + .bind(command.event_id.as_bytes()) + .bind(owner.as_slice()) + .bind(agent.as_slice()) + .bind(command.workflow_id) + .bind(command.expected_revision.as_bytes()) + .bind(command.operation.as_str()) + .fetch_one(&mut *tx) + .await + .map_err(map_owner_command_insert_error)?; + let record = row_to_workflow_owner_command(row)?; + tx.commit().await?; + Ok(WorkflowOwnerCommandAdmission::Inserted(record)) +} + +/// List pending commands for one authenticated target agent in stable order. +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 community_id, command_id, command_sequence, event_id, owner_pubkey, + agent_pubkey, workflow_id, expected_revision, operation, + status::text AS status, result_event_id, result_reason + FROM workflow_owner_commands + WHERE community_id = $1 AND agent_pubkey = $2 AND status = 'pending' + ORDER BY command_sequence + LIMIT $3"#, + ) + .bind(community_id.as_uuid()) + .bind(agent_pubkey) + .bind(limit.clamp(1, LIST_MAX_LIMIT)) + .fetch_all(pool) + .await?; + rows.into_iter() + .map(row_to_workflow_owner_command) + .collect() +} + +/// Fetch one command only for its immutable owner or its target agent. +pub async fn get_workflow_owner_command( + pool: &PgPool, + community_id: CommunityId, + command_id: Uuid, + principal_pubkey: &[u8], +) -> Result { + let row = sqlx::query( + r#"SELECT community_id, command_id, command_sequence, event_id, owner_pubkey, + agent_pubkey, workflow_id, expected_revision, operation, + status::text AS status, result_event_id, result_reason + FROM workflow_owner_commands + WHERE community_id = $1 AND command_id = $2 + AND (owner_pubkey = $3 OR agent_pubkey = $3)"#, + ) + .bind(community_id.as_uuid()) + .bind(command_id) + .bind(principal_pubkey) + .fetch_optional(pool) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow owner command {command_id}")))?; + row_to_workflow_owner_command(row) +} + +/// Apply one exact agent-signed terminal result with compare-and-swap semantics. +pub async fn complete_workflow_owner_command( + pool: &PgPool, + community_id: CommunityId, + result: &WorkflowOwnerResult, +) -> Result { + let mut tx = pool.begin().await?; + let row = sqlx::query( + r#"SELECT community_id, command_id, command_sequence, event_id, owner_pubkey, + agent_pubkey, workflow_id, expected_revision, operation, + status::text AS status, result_event_id, result_reason + FROM workflow_owner_commands + WHERE community_id = $1 AND command_id = $2 + FOR UPDATE"#, + ) + .bind(community_id.as_uuid()) + .bind(result.command_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("workflow owner command {}", result.command_id)))?; + let current = row_to_workflow_owner_command(row)?; + if !result_matches_command(result, ¤t) { + return Err(DbError::InvalidData( + "result does not match its workflow owner command".into(), + )); + } + if current.status != WorkflowOwnerCommandStatus::Pending { + if current.result_event_id.as_deref() == Some(result.event_id.as_bytes()) + && current.status == status_from_result(result.status) + && current.result_reason == result.reason + { + tx.commit().await?; + return Ok(WorkflowOwnerResultTransition::Replay(current)); + } + return Err(DbError::InvalidData( + "conflicting workflow owner result replay".into(), + )); + } + + let row = sqlx::query( + r#"UPDATE workflow_owner_commands + SET status = $3::workflow_owner_command_status, + result_event_id = $4, result_reason = $5, + completed_at = NOW() + WHERE community_id = $1 AND command_id = $2 AND status = 'pending' + RETURNING community_id, command_id, command_sequence, event_id, + owner_pubkey, agent_pubkey, workflow_id, expected_revision, + operation, status::text AS status, result_event_id, result_reason"#, + ) + .bind(community_id.as_uuid()) + .bind(result.command_id) + .bind(result.status.as_str()) + .bind(result.event_id.as_bytes()) + .bind(result.reason.as_deref()) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::InvalidData( + "workflow owner command terminalization lost its compare-and-swap".into(), + ) + })?; + let record = row_to_workflow_owner_command(row)?; + tx.commit().await?; + Ok(WorkflowOwnerResultTransition::Applied(record)) +} + +fn command_matches_record( + command: &WorkflowOwnerCommand, + record: &WorkflowOwnerCommandRecord, +) -> bool { + record.command_id == command.command_id + && record.event_id == command.event_id.as_bytes() + && record.owner_pubkey == command.owner_pubkey.to_bytes() + && record.agent_pubkey == command.agent_pubkey.to_bytes() + && record.workflow_id == command.workflow_id + && record.expected_revision == command.expected_revision.as_bytes() + && record.operation == command.operation +} + +fn result_matches_command( + result: &WorkflowOwnerResult, + command: &WorkflowOwnerCommandRecord, +) -> bool { + command.agent_pubkey == result.agent_pubkey.to_bytes() + && command.agent_pubkey == result.target_agent_pubkey.to_bytes() + && command.owner_pubkey == result.owner_pubkey.to_bytes() + && result.command_id == command.command_id + && result.workflow_id == command.workflow_id + && command.expected_revision == result.expected_revision.as_bytes() + && result.operation == command.operation +} + +fn status_from_result(status: WorkflowOwnerResultStatus) -> WorkflowOwnerCommandStatus { + match status { + WorkflowOwnerResultStatus::Applied => WorkflowOwnerCommandStatus::Applied, + WorkflowOwnerResultStatus::Rejected => WorkflowOwnerCommandStatus::Rejected, + } +} + +fn row_to_workflow_owner_command(row: sqlx::postgres::PgRow) -> Result { + let community_id: Uuid = row.try_get("community_id")?; + let operation: String = row.try_get("operation")?; + let operation = match operation.as_str() { + "start" => WorkflowOwnerOperation::Start, + "pause" => WorkflowOwnerOperation::Pause, + "resume" => WorkflowOwnerOperation::Resume, + "cancel" => WorkflowOwnerOperation::Cancel, + "restore" => WorkflowOwnerOperation::Restore, + other => { + return Err(DbError::InvalidData(format!( + "unknown workflow owner operation: {other}" + ))) + } + }; + let status: String = row.try_get("status")?; + Ok(WorkflowOwnerCommandRecord { + community_id: CommunityId::from_uuid(community_id), + command_id: row.try_get("command_id")?, + command_sequence: row.try_get("command_sequence")?, + 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")?, + operation, + status: status.parse()?, + result_event_id: row.try_get("result_event_id")?, + result_reason: row.try_get("result_reason")?, + }) +} + +fn map_owner_command_insert_error(error: sqlx::Error) -> DbError { + if let sqlx::Error::Database(database) = &error { + if database.constraint() == Some("workflow_owner_commands_workflow_pending_idx") { + return DbError::InvalidData( + "an earlier workflow owner command is still pending".into(), + ); + } + } + DbError::Sqlx(error) +} + // -- Workflow Run CRUD -------------------------------------------------------- /// Insert a new workflow run on the caller's transaction. @@ -1823,6 +2227,342 @@ mod tests { assert_eq!(cloned.status, ApprovalStatus::Granted); } + #[test] + fn workflow_owner_command_status_round_trips_and_rejects_unknown() { + assert_eq!( + "pending".parse::().unwrap(), + WorkflowOwnerCommandStatus::Pending + ); + assert_eq!( + "applied".parse::().unwrap(), + WorkflowOwnerCommandStatus::Applied + ); + assert_eq!( + "rejected".parse::().unwrap(), + WorkflowOwnerCommandStatus::Rejected + ); + assert!(matches!( + "lost".parse::(), + Err(DbError::InvalidData(message)) + if message == "unknown workflow owner command status: lost" + )); + } + + // -- Managed owner-command persistence ------------------------------------ + + use crate::user::ensure_user; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + struct OwnerCommandFixture { + admin: PgPool, + pool: PgPool, + database_name: String, + community: CommunityId, + workflow_id: Uuid, + owner_keys: Keys, + agent_keys: Keys, + revision: nostr::EventId, + } + + impl OwnerCommandFixture { + async fn new() -> Self { + let admin_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + let admin = PgPool::connect(&admin_url) + .await + .expect("connect isolated PG admin"); + let database_name = format!("buzz_owner_command_{}", Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE DATABASE {database_name}" + ))) + .execute(&admin) + .await + .expect("create isolated database"); + let (prefix, _) = admin_url.rsplit_once('/').expect("admin URL path"); + let pool = PgPool::connect(&format!("{prefix}/{database_name}")) + .await + .expect("connect isolated database"); + crate::migration::run_migrations(&pool) + .await + .expect("migrate isolated database"); + + let community = make_community(&pool).await; + let owner_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let owner = owner_keys.public_key().to_bytes(); + let agent = agent_keys.public_key().to_bytes(); + ensure_user(&pool, community, &owner) + .await + .expect("ensure owner"); + ensure_user(&pool, community, &agent) + .await + .expect("ensure agent"); + assert!( + crate::user::set_agent_owner(&pool, community, &agent, &owner) + .await + .expect("set immutable owner") + ); + + let workflow_id = Uuid::new_v4(); + let revision = nostr::EventId::from_hex(&"33".repeat(32)).unwrap(); + let channel = make_channel(&pool, community, &agent).await; + let mut conn = pool.acquire().await.expect("acquire workflow connection"); + upsert_workflow( + &mut conn, + community, + workflow_id, + Some(channel), + &agent, + "owner-command", + r#"{"trigger":{"on":"message_posted"},"steps":[]}"#, + &[0x44; 32], + revision.as_bytes(), + ) + .await + .expect("materialize agent workflow"); + drop(conn); + + Self { + admin, + pool, + database_name, + community, + workflow_id, + owner_keys, + agent_keys, + revision, + } + } + + fn command( + &self, + command_id: Uuid, + operation: WorkflowOwnerOperation, + ) -> WorkflowOwnerCommand { + let body = buzz_core::workflow_owner_command::WorkflowOwnerCommandBody { operation }; + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_WORKFLOW_OWNER_COMMAND as u16), + serde_json::to_string(&body).expect("serialize command"), + ) + .tags([ + Tag::parse(["d", &command_id.to_string()]).expect("command id tag"), + Tag::parse([ + "a", + &format!( + "{}:{}:{}", + buzz_core::kind::KIND_WORKFLOW_DEF, + self.agent_keys.public_key().to_hex(), + self.workflow_id + ), + ]) + .expect("workflow coordinate tag"), + Tag::parse(["revision", &self.revision.to_hex()]).expect("revision tag"), + Tag::parse(["p", &self.agent_keys.public_key().to_hex()]).expect("recipient tag"), + ]) + .sign_with_keys(&self.owner_keys) + .expect("sign command"); + buzz_core::workflow_owner_command::parse_owner_command(&event).expect("parse command") + } + + fn result( + &self, + command: &WorkflowOwnerCommand, + status: WorkflowOwnerResultStatus, + reason: Option<&str>, + ) -> WorkflowOwnerResult { + let body = buzz_core::workflow_owner_command::WorkflowOwnerResultBody { + status, + reason: reason.map(str::to_owned), + }; + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_WORKFLOW_OWNER_RESULT as u16), + serde_json::to_string(&body).expect("serialize result"), + ) + .tags([ + Tag::parse(["d", &command.command_id.to_string()]).expect("command id tag"), + Tag::parse(["a", &command.workflow_coordinate()]).expect("workflow coordinate tag"), + Tag::parse(["revision", &command.expected_revision.to_hex()]) + .expect("revision tag"), + Tag::parse(["p", &command.owner_pubkey.to_hex()]).expect("owner tag"), + Tag::parse(["operation", command.operation.as_str()]).expect("operation tag"), + Tag::parse(["status", status.as_str()]).expect("status tag"), + ]) + .sign_with_keys(&self.agent_keys) + .expect("sign result"); + buzz_core::workflow_owner_command::parse_owner_result(&event).expect("parse result") + } + + async fn cleanup(self) { + self.pool.close().await; + sqlx::query(sqlx::AssertSqlSafe(format!( + "DROP DATABASE {} WITH (FORCE)", + self.database_name + ))) + .execute(&self.admin) + .await + .expect("drop isolated database"); + self.admin.close().await; + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_command_persistence_is_private_ordered_and_replay_stable() { + let fixture = OwnerCommandFixture::new().await; + let command = fixture.command(Uuid::new_v4(), WorkflowOwnerOperation::Restore); + let admitted = admit_workflow_owner_command(&fixture.pool, fixture.community, &command) + .await + .expect("admit command"); + let inserted = match admitted { + WorkflowOwnerCommandAdmission::Inserted(record) => record, + other => panic!("expected inserted command, got {other:?}"), + }; + assert_eq!(inserted.status, WorkflowOwnerCommandStatus::Pending); + + let target = get_workflow_owner_target( + &fixture.pool, + fixture.community, + fixture.workflow_id, + &fixture.owner_keys.public_key().to_bytes(), + ) + .await + .expect("owner target"); + assert_eq!( + target.agent_pubkey, + fixture.agent_keys.public_key().to_bytes() + ); + assert_eq!(target.expected_revision, fixture.revision.as_bytes()); + assert!(matches!( + get_workflow_owner_target( + &fixture.pool, + fixture.community, + fixture.workflow_id, + &Keys::generate().public_key().to_bytes(), + ) + .await, + Err(DbError::NotFound(_)) + )); + + assert!(matches!( + get_workflow_owner_command( + &fixture.pool, + fixture.community, + command.command_id, + &Keys::generate().public_key().to_bytes(), + ) + .await, + Err(DbError::NotFound(_)) + )); + let pending = list_pending_workflow_owner_commands( + &fixture.pool, + fixture.community, + &fixture.agent_keys.public_key().to_bytes(), + 100, + ) + .await + .expect("list pending"); + assert_eq!(pending, vec![inserted.clone()]); + + let blocked = fixture.command(Uuid::new_v4(), WorkflowOwnerOperation::Pause); + assert!(matches!( + admit_workflow_owner_command(&fixture.pool, fixture.community, &blocked).await, + Err(DbError::InvalidData(message)) + if message == "an earlier workflow owner command is still pending" + )); + + let result = fixture.result( + &command, + WorkflowOwnerResultStatus::Applied, + Some("restored"), + ); + let completed = complete_workflow_owner_command(&fixture.pool, fixture.community, &result) + .await + .expect("complete command"); + assert!(matches!( + completed, + WorkflowOwnerResultTransition::Applied(ref record) + if record.status == WorkflowOwnerCommandStatus::Applied + && record.result_reason.as_deref() == Some("restored") + )); + assert!(matches!( + complete_workflow_owner_command(&fixture.pool, fixture.community, &result) + .await + .expect("replay result"), + WorkflowOwnerResultTransition::Replay(_) + )); + assert!(list_pending_workflow_owner_commands( + &fixture.pool, + fixture.community, + &fixture.agent_keys.public_key().to_bytes(), + 100, + ) + .await + .expect("list after terminal") + .is_empty()); + + // Exact retries acknowledge the recorded command even if the workflow's + // mutable current revision has advanced since admission. + sqlx::query( + "UPDATE workflows SET definition_event_id = $3 WHERE community_id = $1 AND id = $2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.workflow_id) + .bind(vec![0x55_u8; 32]) + .execute(&fixture.pool) + .await + .expect("advance revision"); + assert!(matches!( + admit_workflow_owner_command(&fixture.pool, fixture.community, &command) + .await + .expect("replay command"), + WorkflowOwnerCommandAdmission::Duplicate(_) + )); + + fixture.cleanup().await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_command_rejects_stale_revision_wrong_owner_and_conflicting_result() { + let fixture = OwnerCommandFixture::new().await; + let mut stale = fixture.command(Uuid::new_v4(), WorkflowOwnerOperation::Start); + stale.expected_revision = nostr::EventId::from_hex(&"66".repeat(32)).unwrap(); + assert!(matches!( + admit_workflow_owner_command(&fixture.pool, fixture.community, &stale).await, + Err(DbError::InvalidData(message)) if message == "workflow revision conflict" + )); + + let mut wrong_owner = fixture.command(Uuid::new_v4(), WorkflowOwnerOperation::Start); + wrong_owner.owner_pubkey = Keys::generate().public_key(); + assert!(matches!( + admit_workflow_owner_command(&fixture.pool, fixture.community, &wrong_owner).await, + Err(DbError::AccessDenied(message)) + if message == "command author is not the immutable agent owner" + )); + + let command = fixture.command(Uuid::new_v4(), WorkflowOwnerOperation::Start); + admit_workflow_owner_command(&fixture.pool, fixture.community, &command) + .await + .expect("admit valid command"); + let result = fixture.result(&command, WorkflowOwnerResultStatus::Rejected, Some("no")); + complete_workflow_owner_command(&fixture.pool, fixture.community, &result) + .await + .expect("reject command"); + let conflicting = fixture.result( + &command, + WorkflowOwnerResultStatus::Applied, + Some("different terminal result"), + ); + assert!(matches!( + complete_workflow_owner_command(&fixture.pool, fixture.community, &conflicting).await, + Err(DbError::InvalidData(message)) + if message == "conflicting workflow owner result replay" + )); + + fixture.cleanup().await; + } + // -- Scheduled workflow claim confinement --------------------------------- // // RECONCILED spec (supersedes the earlier S1 lock; Eva/Max 2026-06-27). @@ -1857,8 +2597,6 @@ mod tests { // race must yield exactly one winner, and pruning below the largest // interval breaks `latest_*` (the ยง5c retention rule Sami flagged). - use crate::user::ensure_user; - const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; async fn setup_pool() -> PgPool { diff --git a/migrations/0035_workflow_owner_commands.sql b/migrations/0035_workflow_owner_commands.sql new file mode 100644 index 0000000000..da3440f167 --- /dev/null +++ b/migrations/0035_workflow_owner_commands.sql @@ -0,0 +1,36 @@ +-- Durable, private state for owner-signed managed-workflow commands. +CREATE TYPE workflow_owner_command_status AS ENUM ('pending', 'applied', 'rejected'); + +CREATE TABLE workflow_owner_commands ( + community_id UUID NOT NULL REFERENCES communities(id), + command_id UUID NOT NULL, + command_sequence BIGSERIAL NOT NULL, + event_id BYTEA NOT NULL CHECK (length(event_id) = 32), + owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32), + agent_pubkey BYTEA NOT NULL CHECK (length(agent_pubkey) = 32), + workflow_id UUID NOT NULL, + expected_revision BYTEA NOT NULL CHECK (length(expected_revision) = 32), + operation TEXT NOT NULL CHECK (operation IN ('start', 'pause', 'resume', 'cancel', 'restore')), + status workflow_owner_command_status NOT NULL DEFAULT 'pending', + result_event_id BYTEA CHECK (result_event_id IS NULL OR length(result_event_id) = 32), + result_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + PRIMARY KEY (community_id, command_id), + UNIQUE (community_id, event_id), + FOREIGN KEY (community_id, workflow_id) REFERENCES workflows (community_id, id) ON DELETE CASCADE, + CHECK ((status = 'pending') = (result_event_id IS NULL)), + CHECK ((status = 'pending') = (completed_at IS NULL)) +); + +CREATE UNIQUE INDEX workflow_owner_commands_result_event_idx + ON workflow_owner_commands (community_id, result_event_id) + WHERE result_event_id IS NOT NULL; +CREATE INDEX workflow_owner_commands_agent_pending_idx + ON workflow_owner_commands (community_id, agent_pubkey, command_sequence) + WHERE status = 'pending'; +CREATE UNIQUE INDEX workflow_owner_commands_workflow_pending_idx + ON workflow_owner_commands (community_id, workflow_id) + WHERE status = 'pending'; + +SELECT attach_community_write_fence('workflow_owner_commands'); diff --git a/schema/schema.sql b/schema/schema.sql index f2230e7747..bf9af01a42 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1723,6 +1723,42 @@ BEGIN END $$; +-- Durable, private state for owner-signed managed-workflow commands. +CREATE TYPE workflow_owner_command_status AS ENUM ('pending', 'applied', 'rejected'); + +CREATE TABLE workflow_owner_commands ( + community_id UUID NOT NULL REFERENCES communities(id), + command_id UUID NOT NULL, + command_sequence BIGSERIAL NOT NULL, + event_id BYTEA NOT NULL CHECK (length(event_id) = 32), + owner_pubkey BYTEA NOT NULL CHECK (length(owner_pubkey) = 32), + agent_pubkey BYTEA NOT NULL CHECK (length(agent_pubkey) = 32), + workflow_id UUID NOT NULL, + expected_revision BYTEA NOT NULL CHECK (length(expected_revision) = 32), + operation TEXT NOT NULL CHECK (operation IN ('start', 'pause', 'resume', 'cancel', 'restore')), + status workflow_owner_command_status NOT NULL DEFAULT 'pending', + result_event_id BYTEA CHECK (result_event_id IS NULL OR length(result_event_id) = 32), + result_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + PRIMARY KEY (community_id, command_id), + UNIQUE (community_id, event_id), + FOREIGN KEY (community_id, workflow_id) REFERENCES workflows (community_id, id) ON DELETE CASCADE, + CHECK ((status = 'pending') = (result_event_id IS NULL)), + CHECK ((status = 'pending') = (completed_at IS NULL)) +); + +CREATE UNIQUE INDEX workflow_owner_commands_result_event_idx + ON workflow_owner_commands (community_id, result_event_id) + WHERE result_event_id IS NOT NULL; +CREATE INDEX workflow_owner_commands_agent_pending_idx + ON workflow_owner_commands (community_id, agent_pubkey, command_sequence) + WHERE status = 'pending'; +CREATE UNIQUE INDEX workflow_owner_commands_workflow_pending_idx + ON workflow_owner_commands (community_id, workflow_id) + WHERE status = 'pending'; + + -- Desired-state schema application does not replay migration history, so keep -- these explicit calls as first-class catalog declarations. They also make the -- fence contract visible to migration linting instead of hiding it only in the @@ -1752,6 +1788,7 @@ SELECT attach_community_write_fence('scheduled_workflow_fires'); SELECT attach_community_write_fence('subscriptions'); SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users'); +SELECT attach_community_write_fence('workflow_owner_commands'); SELECT attach_community_write_fence('workflow_approvals'); SELECT attach_community_write_fence('workflow_runs'); SELECT attach_community_write_fence('workflows');