Workflows Engine

The workflow engine data model — versioned defs, runs, and per-node runs, with RLS. Idempotent DDL.

-- ---------------------------------------------------------------------------
-- Workflow Designer Studio — the engine data model (WFS Phase 1, issue #1109).
--
-- Generalizes the Agents Store job model into a versioned, JSON-defined GRAPH
-- without replacing it. A workflow is a DAG of nodes + edges; a run executes
-- that graph; each node run REUSES the agent_jobs lifecycle (operator nodes
-- queue/drain exactly as today) by linking back to agent_jobs.id.
--
--   workflow_defs       — versioned graph definitions (immutable once published;
--                         a new edit inserts a new (slug, version) row).
--   workflow_runs       — one row per execution of a def@version.
--   workflow_node_runs  — per-node execution; operator nodes point at the
--                         agent_jobs row the existing drain runs.
--
-- The web app GOVERNS (author/publish/enqueue/approve); the existing GitHub
-- Actions runner EXECUTES operator nodes via agent_jobs — zero new runner code.
--
-- RLS on, NO anon — service-role only. Apply to the MAIN project
-- (diziibeymowzpondeqxz). Additive, idempotent. Reuses iam_set_updated_at().
-- ---------------------------------------------------------------------------

-- --- workflow_defs (the authored graph, versioned) --------------------------
create table if not exists public.workflow_defs (
  id          uuid primary key default gen_random_uuid(),
  slug        text not null,                                       -- stable workflow identity (e.g. 'entity-wars.surface')
  version     integer not null default 1,                          -- bumps on every published edit; (slug, version) immutable
  graph       jsonb not null default '{"nodes":[],"edges":[]}'::jsonb,
  triggers    jsonb not null default '[]'::jsonb,                  -- [{ type: 'cron'|'event'|'manual'|'webhook', ... }]
  enabled     boolean not null default false,                      -- only an enabled def fires on a trigger
  created_by  text,
  created_at  timestamptz not null default now(),
  updated_at  timestamptz not null default now(),
  unique (slug, version)
);
comment on table public.workflow_defs is
  'WFS: versioned, JSON-defined workflow graphs. Immutable once published — a new edit inserts a new (slug, version). enabled gates trigger firing.';

create index if not exists workflow_defs_slug_idx
  on public.workflow_defs (slug, version desc);
create index if not exists workflow_defs_enabled_idx
  on public.workflow_defs (enabled, slug);

drop trigger if exists workflow_defs_set_updated_at on public.workflow_defs;
create trigger workflow_defs_set_updated_at
  before update on public.workflow_defs
  for each row execute function public.iam_set_updated_at();

alter table public.workflow_defs enable row level security;

-- --- workflow_runs (one per execution) --------------------------------------
create table if not exists public.workflow_runs (
  id           uuid primary key default gen_random_uuid(),
  workflow_id  uuid not null references public.workflow_defs (id) on delete cascade,
  slug         text not null,                                      -- denormalized for cheap listing/filtering
  version      integer not null,                                   -- the def version this run executes (defs are immutable)
  status       text not null default 'queued'
               check (status in ('queued', 'running', 'blocked', 'done', 'failed', 'cancelled')),
  trigger_ref  text,                                               -- 'manual:<actor>' | 'cron' | 'event:<id>' | 'webhook:<source>'
  inputs       jsonb not null default '{}'::jsonb,
  error        text,
  queued_at    timestamptz not null default now(),
  started_at   timestamptz,
  finished_at  timestamptz,
  created_by   text,
  created_at   timestamptz not null default now(),
  updated_at   timestamptz not null default now()
);
comment on table public.workflow_runs is
  'WFS: one row per workflow execution. blocked = paused on a human-gate node (resumes to running on approve). Mirrors the agent_jobs status discipline.';

create index if not exists workflow_runs_workflow_idx
  on public.workflow_runs (workflow_id, status, queued_at);
create index if not exists workflow_runs_status_idx
  on public.workflow_runs (status, queued_at);
create index if not exists workflow_runs_slug_idx
  on public.workflow_runs (slug, queued_at);

drop trigger if exists workflow_runs_set_updated_at on public.workflow_runs;
create trigger workflow_runs_set_updated_at
  before update on public.workflow_runs
  for each row execute function public.iam_set_updated_at();

alter table public.workflow_runs enable row level security;

-- --- workflow_node_runs (per-node, reuses agent_jobs for operator steps) -----
create table if not exists public.workflow_node_runs (
  id            uuid primary key default gen_random_uuid(),
  run_id        uuid not null references public.workflow_runs (id) on delete cascade,
  node_id       text not null,                                     -- the node's id WITHIN the graph
  node_type     text not null,                                     -- 'operator' | 'llm' | 'http' | 'human-gate'
  status        text not null default 'pending'
                check (status in ('pending', 'queued', 'running', 'blocked', 'done', 'failed', 'skipped', 'cancelled')),
  verdict       text check (verdict in ('pass', 'fail', 'risk')),  -- operator nodes inherit the agent_jobs verdict
  agent_job_id  uuid references public.agent_jobs (id) on delete set null,  -- the queued/drained job for operator nodes
  output        jsonb,                                             -- bounded node output handed to downstream ports
  error         text,
  attempts      integer not null default 0,
  started_at    timestamptz,
  finished_at   timestamptz,
  created_at    timestamptz not null default now(),
  updated_at    timestamptz not null default now(),
  unique (run_id, node_id)
);
comment on table public.workflow_node_runs is
  'WFS: per-node execution within a workflow_run. Operator nodes link agent_job_id and inherit its verdict — the existing drain runs them, zero new runner code.';

create index if not exists workflow_node_runs_run_idx
  on public.workflow_node_runs (run_id);
create index if not exists workflow_node_runs_job_idx
  on public.workflow_node_runs (agent_job_id);

drop trigger if exists workflow_node_runs_set_updated_at on public.workflow_node_runs;
create trigger workflow_node_runs_set_updated_at
  before update on public.workflow_node_runs
  for each row execute function public.iam_set_updated_at();

alter table public.workflow_node_runs enable row level security;