fix(workflows): make forward definition lifecycle atomic and discoverable

.env.example

@@ -69,6 +69,11 @@ RELAY_URL=ws://localhost:3000
 # Stable relay signing key (required). `just bootstrap` generates a random key in
 # the gitignored .env file. Preserve that value across restarts and backups.
 # BUZZ_RELAY_PRIVATE_KEY=<32-byte hex private key>
+# Host-wide NIP-11 workflow lifecycle advertisement: default off; only true/1 opts in.
+# Enable only AFTER all incompatible endpoints AND existing connections are drained.
+# Gates discovery only, not lifecycle enforcement. See the two-phase operator sequence:
+# deploy/charts/buzz/README.md#workflow-lifecycle-activation
+BUZZ_ADVERTISE_WORKFLOW_LIFECYCLE=false
 # Optional: path to the web UI dist directory. When set, the relay serves
 # the web frontend at / for browser requests. Leave unset for local dev
 # (use `just web` for Vite HMR instead).

crates/buzz-agent/tests/fake_llm.rs

@@ -1320,7 +1320,7 @@ async fn mid_turn_usage_includes_earlier_turns() {
 /// Setup: round 1 is a tool call WITH usage (tokens are captured). After the
 /// tool_call_update notification (proving round 1 is fully processed), we gate
 /// the round-2 LLM response behind a `oneshot` barrier that only releases after
-/// cancel is sent. This guarantees the turn exits with `stopReason: "cancelled"`
+/// cancel is acknowledged. This guarantees the turn exits with `stopReason: "cancelled"`
 /// deterministically, even on a slow CI worker.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn cancelled_turn_with_usage_emits_notification_before_response() {
@@ -1331,11 +1331,9 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() {
     let gate_rx = Arc::new(tokio::sync::Mutex::new(Some(gate_rx)));
 
     // Round 1: tool call with usage — sets turn_input/output_tokens.
-    // Round 2: gated — blocked until cancel fires, then released so the
-    // in-flight TCP request can resolve. The queue is empty for round 2, so the
-    // agent receives the fallback "no canned response" body which it treats as
-    // an LLM error; the cancel check at the round boundary fires first because
-    // the gate is only released after cancel is enqueued.
+    // Round 2 stays blocked until the agent acknowledges cancellation. Writing
+    // cancel to stdin is not a barrier: the HTTP error could win before the
+    // agent reads stdin. handle_request sends the ack after cancel_session.
     let responses = vec![openai_tool_call_with_usage(
         "call_cancel_test",
         "fake__noop",
@@ -1371,7 +1369,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() {
                     }
                 }
                 // For request 2+ (round 2), wait for the gate to open before
-                // responding. This ensures cancel is sent before round 2 resolves,
+                // responding. This ensures cancel is acknowledged before round 2 resolves,
                 // making stopReason: cancelled deterministic.
                 if req_num >= 2 {
                     let rx = gate.lock().await.take();
@@ -1415,10 +1413,10 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() {
     })
     .await;
 
-    // Now send cancel and release the round-2 gate. Cancel is enqueued before
-    // round 2 can respond, so the turn exits with stopReason: cancelled.
+    // Collect every message while waiting for the ack: usage and the prompt
+    // response can precede it, so recv_until would discard evidence of ordering.
     let c_id = h.send("session/cancel", json!({"sessionId": sid})).await;
-    let _ = gate_tx.send(()); // unblock round 2
+    let mut gate_tx = Some(gate_tx);
 
     let mut saw_usage_before_prompt_response = false;
     let mut saw_usage = false;
@@ -1427,7 +1425,12 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() {
     for _ in 0..40 {
         let v = h.recv().await;
         if v["id"] == json!(c_id) {
+            assert_eq!(v.get("result"), Some(&Value::Null), "cancel must succeed");
+            assert!(v.get("error").is_none(), "cancel must not return an error");
             saw_cancel_ok = true;
+            if let Some(gate_tx) = gate_tx.take() {
+                let _ = gate_tx.send(()); // cancellation is now installed
+            }
         } else if is_usage_update(&v) {
             saw_usage = true;
             if !saw_prompt_response {
@@ -1446,6 +1449,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() {
         }
     }
     assert!(saw_cancel_ok, "session/cancel was not acknowledged");
+    assert!(saw_prompt_response, "session/prompt did not complete");
     assert!(
         saw_usage,
         "expected usage_update notification for cancelled turn with observed tokens"

crates/buzz-db/src/runtime/migration.rs

@@ -702,7 +702,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 44);
+        assert_eq!(migrations.len(), 45);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1368,6 +1368,35 @@ mod postgres_tests {
         assert!(include_str!("../../../../schema/schema.sql").contains("error_code          TEXT"));
     }
 
+    #[test]
+    fn workflow_deletion_cutoff_matches_desired_schema_without_legacy_backfill() {
+        let migration = MIGRATOR
+            .iter()
+            .find(|m| m.version == 45)
+            .expect("migration 45");
+        let schema = include_str!("../../../../schema/schema.sql");
+        let table = |sql: &str| {
+            split_sql_statements(sql)
+                .into_iter()
+                .find(|statement| {
+                    normalize_sql(statement).starts_with("create table workflow_deletions")
+                })
+                .map(|statement| normalize_sql(&statement))
+                .expect("deletion table")
+        };
+        assert_eq!(table(migration.sql.as_ref()), table(schema));
+        let statements = split_sql_statements(migration.sql.as_ref());
+        assert_eq!(
+            statements.len(),
+            2,
+            "forward proof must never be inferred from legacy data"
+        );
+        assert_eq!(
+            normalize_sql(&statements[1]),
+            "select attach_community_write_fence('workflow_deletions')"
+        );
+    }
+
     #[test]
     fn push_match_trigger_is_narrowed_to_message_kinds_additively() {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
@@ -1822,6 +1851,11 @@ mod postgres_tests {
         let mut expected_fences = migration.fence_attachments.clone();
         expected_fences.remove("product_feedback");
         expected_fences.remove("rate_limit_violations");
+        let workflow_lifecycle = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 45)
+            .expect("embedded workflow lifecycle migration");
+        expected_fences.extend(surface(workflow_lifecycle.sql.as_ref()).fence_attachments);
         assert_eq!(
             expected_fences, schema.fence_attachments,
             "write-fence attachment targets differ after recovery policy"
@@ -2752,10 +2786,16 @@ mod postgres_tests {
             "all NIP-FI tables must be absent after migration 0044: {present:?}"
         );
 
-        // The deletion catalog must validate with ledger relations gone.
+        // Ledger absence above is checked at 0044. The current binary's deletion
+        // catalog describes the latest schema (including later scoped tables),
+        // so finish the upgrade before asking it to validate serving readiness.
+        MIGRATOR
+            .run(&pool)
+            .await
+            .expect("finish migrations after ledger removal");
         crate::deletion::DeletionStore::new(pool.clone())
             .validate_catalog()
             .await
-            .expect("deletion catalog validates after migration 0044");
+            .expect("current deletion catalog validates after upgrading from migration 0044");
     }
 }

crates/buzz-db/src/store/deletion.rs

@@ -81,12 +81,14 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "thread_metadata",
     "users",
     "workflow_approvals",
+    "workflow_deletions",
     "workflow_runs",
     "workflows",
 ];
 
 /// Foreign-key-safe child-before-parent order for the PostgreSQL purge.
 pub const PURGE_SCOPED_TABLES: &[&str] = &[
+    "workflow_deletions",
     "workflow_approvals",
     "scheduled_workflow_fires",
     "workflow_runs",
@@ -4997,3 +4999,7 @@ mod postgres_tests {
         .expect("drop probe database");
     }
 }
+
+#[cfg(test)]
+#[path = "deletion/workflow_postgres_tests.rs"]
+mod workflow_postgres_tests;

crates/buzz-db/src/store/deletion/workflow_postgres_tests.rs

@@ -0,0 +1,158 @@
+//! The forward workflow proof is tenant data, including during whole-community purge.
+use super::*;
+use buzz_core::channel::{ChannelType, ChannelVisibility};
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn desired_workflow_deletion_proofs_are_fenced_and_purged() {
+    check_workflow_deletion_proofs("desired").await;
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_workflow_deletion_proofs_are_fenced_and_purged() {
+    check_workflow_deletion_proofs("migration").await;
+}
+
+async fn check_workflow_deletion_proofs(mode: &str) {
+    assert_eq!(std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref(), Ok(mode));
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .expect("isolated database");
+    let db = Db::from_pool(pool.clone());
+    if mode == "migration" {
+        db.migrate().await.expect("all migrations");
+    }
+    let store = db.deletion_store();
+    let mut tenants = Vec::new();
+    for label in ["target", "control"] {
+        let host = format!("workflow-purge-{label}-{}.example", Uuid::new_v4());
+        let community = db
+            .ensure_configured_community(&host)
+            .await
+            .expect("community")
+            .id;
+        let channel = db
+            .create_channel(
+                community,
+                "workflows",
+                ChannelType::Stream,
+                ChannelVisibility::Private,
+                None,
+                &[7; 32],
+                None,
+            )
+            .await
+            .expect("channel")
+            .id;
+        sqlx::query(
+            "INSERT INTO workflow_deletions \
+            (community_id, owner_pubkey, workflow_id, channel_id, deleted_through, event_id) \
+            VALUES ($1, $2, $3, $4, now(), $5)",
+        )
+        .bind(community.as_uuid())
+        .bind(vec![7_u8; 32])
+        .bind(Uuid::new_v4())
+        .bind(channel)
+        .bind(vec![8_u8; 32])
+        .execute(&pool)
+        .await
+        .expect("proof");
+        tenants.push((community, host));
+    }
+    let (target, host) = &tenants[0];
+    let schema = store
+        .inventory_schema(*target)
+        .await
+        .expect("catalog includes proof");
+    assert_eq!(schema.row_counts["workflow_deletions"], 1);
+    assert!(schema
+        .fenced_tables
+        .contains(&"workflow_deletions".to_string()));
+    let storage = StorageManifest {
+        version: 5,
+        prefixes: ["_meta", "_uploads", "repos"]
+            .into_iter()
+            .map(|prefix| {
+                let (keys_digest, object_count) = KeyStreamDigest::new().finish();
+                PrefixManifest {
+                    prefix: format!("{prefix}/{target}/"),
+                    keys_digest,
+                    object_count,
+                    total_bytes: 0,
+                }
+            })
+            .collect(),
+    };
+    let request = store
+        .submit(host, "test-operator", None)
+        .await
+        .expect("request");
+    store
+        .freeze_inventory(
+            request.id,
+            &FrozenInventory {
+                schema,
+                storage: storage.clone(),
+            },
+        )
+        .await
+        .expect("freeze");
+    store
+        .approve(request.id, "test-approver", None)
+        .await
+        .expect("approve");
+    let claim = store
+        .claim_specific(request.id, "test-executor", DEFAULT_LEASE_DURATION)
+        .await
+        .expect("claim")
+        .expect("won claim");
+    store.begin_quiescing(&claim.lease).await.expect("quiesce");
+    let generation = store.fence(&claim.lease).await.expect("fence");
+    let token = LeaseToken {
+        fence_generation: Some(generation),
+        ..claim.lease
+    };
+    let error = sqlx::query(
+        "UPDATE workflow_deletions SET deleted_through = now() WHERE community_id = $1",
+    )
+    .bind(target.as_uuid())
+    .execute(&pool)
+    .await
+    .expect_err("proof updates must be fenced");
+    assert!(
+        error.to_string().contains("community write fenced"),
+        "{error}"
+    );
+    store
+        .freeze_destructive_storage_manifest(&token, &storage)
+        .await
+        .expect("storage");
+    store.mark_drained(&token).await.expect("drain");
+    store
+        .mark_bindings_removed(&token, serde_json::json!({"keys": 0}))
+        .await
+        .expect("bindings");
+    let counts = store
+        .purge_postgres(&token)
+        .await
+        .expect("purge child before channel");
+    assert_eq!(counts["workflow_deletions"], 1);
+    store
+        .mark_cache_purged(&token, serde_json::json!({"keys": 0}))
+        .await
+        .expect("cache");
+    store
+        .verify_postgres_logically_deleted(&token)
+        .await
+        .expect("logical absence");
+    let remaining: Vec<Uuid> = sqlx::query_scalar("SELECT community_id FROM workflow_deletions")
+        .fetch_all(&pool)
+        .await
+        .expect("remaining proofs");
+    assert_eq!(
+        remaining,
+        vec![*tenants[1].0.as_uuid()],
+        "control tenant stays intact"
+    );
+}

crates/buzz-db/src/store/relay_admin_actions.rs

@@ -693,8 +693,9 @@ pub async fn execute_kick_with_marker(
 /// Atomically execute a soft-delete mutation and commit the step marker in one
 /// transaction, fenced by `action_id` AND the caller's `lease_token`.
 ///
-/// The delete is idempotent: if the event is already deleted the marker is still
-/// committed (soft-delete is already-done = success).
+/// Workflow definitions are rejected, including already-deleted ones; they require
+/// canonical author-signed lifecycle deletion. For other events, an already-deleted
+/// target is idempotent success and the marker is still committed.
 pub async fn execute_delete_with_marker(
     pool: &PgPool,
     action_id: Uuid,
@@ -727,6 +728,25 @@ pub async fn execute_delete_with_marker(
         return Ok(false);
     }
 
+    // Workflow definitions are one projection of executable state. This shared
+    // boundary also serves recovery-worker retries, so HTTP-only validation is
+    // insufficient. Reject tombstoned definitions too: they do not make a new
+    // moderation action an author-signed lifecycle deletion. Event kind is immutable.
+    let workflow: bool = sqlx::query_scalar(
+        "SELECT EXISTS (SELECT 1 FROM events WHERE community_id = $1 AND id = $2 AND kind = $3)",
+    )
+    .bind(community_id.as_uuid())
+    .bind(target_event_id)
+    .bind(buzz_core::kind::KIND_WORKFLOW_DEF as i32)
+    .fetch_one(&mut *tx)
+    .await?;
+    if workflow {
+        tx.rollback().await?;
+        return Err(crate::DbError::InvalidData(
+            "workflow definitions require canonical author-signed deletion".into(),
+        ));
+    }
+
     // Soft-delete the event and update thread metadata (idempotent: already-deleted is a no-op).
     sqlx::query(
         r#"