feat(buzz-db): durable workflow agent delivery state machine

Add node C of the workflow-delivery DAG: the complete dormant DB
transition system for workflow messages addressed to managed agents,
built on B's canonical delivery protocol vocabulary.

Migration 0035 creates workflow_agent_deliveries: one canonical binding
per (community, run, step, target) with a pending -> claimed ->
finished | failed status enum, a monotonic lease_generation fence,
decomposed WorkflowDeliveryCause identity columns guarded by a CHECK
that makes an ambiguous or absent cause unrepresentable, cascading FKs
to workflows/workflow_runs/events, and the community write fence. The
desired-state schema.sql mirrors it inline.

The buzz-db module persists B's WorkflowDeliveryBinding verbatim (no
duplicate tuple spelling) and owns only the lifecycle around it:

  - lock_workflow_agent_delivery_identity + commit_workflow_agent_deliveries:
    the ONLY insert path, atomic per step, idempotent across producer
    retries via ON CONFLICT DO NOTHING.
  - claim_workflow_agent_delivery: FOR UPDATE SKIP LOCKED, scoped to the
    authenticated target and community; an optional expected binding must
    match every supplied field (workflow/run/step/definition/message and
    the full decomposed cause) so a forged or stale wake is a miss, never
    an alternate authority path. The winning row bumps its fence token.
  - renew / finish: fenced by lease_generation; a reaped or superseded
    holder matches zero rows and fails closed. finish is terminal-once and
    reconciles uncertain completion to a single terminal idempotently.
  - reap_expired_workflow_agent_deliveries: fleet-wide, filtered through
    community_write_allowed exactly like the scheduler prune scan, so a
    quiescing/fenced/deleted tenant is skipped before its write fence can
    abort healthy tenants in the same statement.

No producer, runtime, API, or ACP reachability: dormant by contract.

Fresh-Postgres invariant matrix (#[ignore]) proves create idempotency,
community/target claim isolation, mismatched-binding rejection (incl.
cause- and workflow-only disagreement), lease fencing by generation,
reap + prior-holder fencing, terminal-once/idempotent reconciliation,
the fleet-wide lifecycle fence, run-deletion cascade, and commit
rollback. migration.rs asserts len 35, the 0035 contract, and fence
parity.

Co-authored-by: Mongo <89eda16aff8223982a0d928ccc850c50a22abc23babd9f890dc8d6d5431876fb@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-08-25 16:59:22 -04:00
committed by Larry
parent 92a4800c11
commit 83b60e5e05
5 changed files with 1469 additions and 2 deletions

View File

@@ -31,6 +31,7 @@ CREATE TYPE member_role AS ENUM ('owner', 'admin', 'member', 'guest', 'bot');
CREATE TYPE workflow_status AS ENUM ('active', 'disabled', 'archived');
CREATE TYPE run_status AS ENUM ('pending', 'running', 'waiting_approval', 'completed', 'failed', 'cancelled');
CREATE TYPE approval_status AS ENUM ('pending', 'granted', 'denied', 'expired');
CREATE TYPE workflow_agent_delivery_status AS ENUM ('pending', 'claimed', 'finished', 'failed');
CREATE TYPE delivery_method AS ENUM ('webhook', 'websocket');
CREATE TYPE subscription_status AS ENUM ('active', 'paused', 'deleted');
CREATE TYPE pause_reason AS ENUM ('user', 'system', 'rate_limit');
@@ -475,6 +476,66 @@ CREATE TABLE scheduled_workflow_fires (
-- by claimed_at globally (operator concern). See plan §5 retention coupling.
CREATE INDEX idx_scheduled_fires_claimed_at ON scheduled_workflow_fires (claimed_at);
-- ── Workflow agent deliveries ───────────────────────────────────────────────
-- Durable, target-scoped delivery inbox and complete transition state machine
-- for workflow messages addressed to managed agents. Persists exactly one
-- canonical binding per (community, run, step, target) and owns the lifecycle
-- pending -> claimed -> finished | failed, with fenced leases reclaimed by a
-- fleet-wide reaper. See migration 0035 for the full contract.
CREATE TABLE workflow_agent_deliveries (
community_id UUID NOT NULL REFERENCES communities(id),
id UUID NOT NULL,
workflow_id UUID NOT NULL,
run_id UUID NOT NULL,
step_id VARCHAR(64) NOT NULL CHECK (length(btrim(step_id)) > 0),
target_pubkey BYTEA NOT NULL CHECK (octet_length(target_pubkey) = 32),
definition_event_id BYTEA NOT NULL CHECK (octet_length(definition_event_id) = 32),
message_event_id BYTEA NOT NULL CHECK (octet_length(message_event_id) = 32),
message_event_created_at TIMESTAMPTZ NOT NULL,
cause_kind TEXT NOT NULL CHECK (cause_kind IN ('event', 'schedule', 'webhook')),
cause_event_id BYTEA CHECK (cause_event_id IS NULL OR octet_length(cause_event_id) = 32),
cause_scheduled_for TIMESTAMPTZ,
cause_webhook_invocation_id UUID,
status workflow_agent_delivery_status NOT NULL DEFAULT 'pending',
lease_generation BIGINT NOT NULL DEFAULT 0 CHECK (lease_generation >= 0),
lease_until TIMESTAMPTZ,
claimed_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (community_id, id),
UNIQUE (community_id, run_id, step_id, target_pubkey),
FOREIGN KEY (community_id, workflow_id) REFERENCES workflows (community_id, id) ON DELETE CASCADE,
FOREIGN KEY (community_id, run_id) REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE,
FOREIGN KEY (community_id, message_event_created_at, message_event_id)
REFERENCES events (community_id, created_at, id) ON DELETE CASCADE,
CHECK (
(cause_kind = 'event'
AND cause_event_id IS NOT NULL
AND cause_scheduled_for IS NULL
AND cause_webhook_invocation_id IS NULL)
OR (cause_kind = 'schedule'
AND cause_event_id IS NULL
AND cause_scheduled_for IS NOT NULL
AND cause_webhook_invocation_id IS NULL)
OR (cause_kind = 'webhook'
AND cause_event_id IS NULL
AND cause_scheduled_for IS NULL
AND cause_webhook_invocation_id IS NOT NULL)
),
CHECK ((status = 'claimed') = (lease_until IS NOT NULL)),
CHECK ((status = 'claimed') = (claimed_at IS NOT NULL)),
CHECK ((status IN ('finished', 'failed')) = (finished_at IS NOT NULL))
);
CREATE INDEX idx_workflow_agent_deliveries_pending
ON workflow_agent_deliveries (community_id, target_pubkey, created_at)
WHERE status = 'pending';
CREATE INDEX idx_workflow_agent_deliveries_lease
ON workflow_agent_deliveries (lease_until)
WHERE status = 'claimed';
-- ── API tokens ────────────────────────────────────────────────────────────────
-- Conformance: "API tokens and NIP-98 replay". token_hash uniqueness scoped to
-- (community_id, token_hash); channel claims reference channels in same community.
@@ -1752,6 +1813,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_agent_deliveries');
SELECT attach_community_write_fence('workflow_approvals');
SELECT attach_community_write_fence('workflow_runs');
SELECT attach_community_write_fence('workflows');