From da65b3af45615a121342fa77e72fdeace84c8f56 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 18 Apr 2026 06:31:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(minions):=20migration=20v7=20=E2=80=94=20a?= =?UTF-8?q?gent=5Fparity=5Flayer=20schema?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds columns on minion_jobs (depth, max_children, timeout_ms, timeout_at, remove_on_complete, remove_on_fail, idempotency_key) plus the new minion_attachments table. Three partial indexes for bounded scans: idx_minion_jobs_timeout, idx_minion_jobs_parent_status, and uniq_minion_jobs_idempotency. Check constraints enforce non-negative depth and positive child cap / timeout. Additive migration — existing installs pick it up via ensureSchema on next use. No user action required. Co-Authored-By: Claude Opus 4.7 --- src/core/migrate.ts | 65 +++++++++++++++++++++++++++++++++++++ src/core/pglite-schema.ts | 39 +++++++++++++++++++++- src/core/schema-embedded.ts | 65 +++++++++++++++++++++++++++++++++++-- src/schema.sql | 35 +++++++++++++++++++- 4 files changed, 200 insertions(+), 4 deletions(-) diff --git a/src/core/migrate.ts b/src/core/migrate.ts index e6ee4f425..052b977e2 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -154,6 +154,71 @@ const MIGRATIONS: Migration[] = [ CREATE INDEX IF NOT EXISTS idx_minion_inbox_unread ON minion_inbox (job_id) WHERE read_at IS NULL; `, }, + { + version: 7, + name: 'agent_parity_layer', + sql: ` + -- Subagent primitives + BullMQ parity columns + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS depth INTEGER NOT NULL DEFAULT 0; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS max_children INTEGER; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS timeout_ms INTEGER; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS timeout_at TIMESTAMPTZ; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS remove_on_complete BOOLEAN NOT NULL DEFAULT FALSE; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS remove_on_fail BOOLEAN NOT NULL DEFAULT FALSE; + ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS idempotency_key TEXT; + + -- Tighten constraints (drop-then-add for idempotency) + ALTER TABLE minion_jobs DROP CONSTRAINT IF EXISTS chk_depth_nonnegative; + ALTER TABLE minion_jobs ADD CONSTRAINT chk_depth_nonnegative CHECK (depth >= 0); + ALTER TABLE minion_jobs DROP CONSTRAINT IF EXISTS chk_max_children_positive; + ALTER TABLE minion_jobs ADD CONSTRAINT chk_max_children_positive CHECK (max_children IS NULL OR max_children > 0); + ALTER TABLE minion_jobs DROP CONSTRAINT IF EXISTS chk_timeout_positive; + ALTER TABLE minion_jobs ADD CONSTRAINT chk_timeout_positive CHECK (timeout_ms IS NULL OR timeout_ms > 0); + + -- Bounded scan for handleTimeouts + CREATE INDEX IF NOT EXISTS idx_minion_jobs_timeout ON minion_jobs (timeout_at) + WHERE status = 'active' AND timeout_at IS NOT NULL; + + -- O(children) child-count check in add() + CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent_status ON minion_jobs (parent_job_id, status) + WHERE parent_job_id IS NOT NULL; + + -- Idempotency: enforce "only one job per key" at the DB layer + CREATE UNIQUE INDEX IF NOT EXISTS uniq_minion_jobs_idempotency ON minion_jobs (idempotency_key) + WHERE idempotency_key IS NOT NULL; + + -- Fast lookup of child_done messages for readChildCompletions + CREATE INDEX IF NOT EXISTS idx_minion_inbox_child_done ON minion_inbox (job_id, sent_at) + WHERE (payload->>'type') = 'child_done'; + + -- Attachment manifest (BYTEA inline + forward-compat storage_uri) + CREATE TABLE IF NOT EXISTS minion_attachments ( + id SERIAL PRIMARY KEY, + job_id INTEGER NOT NULL REFERENCES minion_jobs(id) ON DELETE CASCADE, + filename TEXT NOT NULL, + content_type TEXT NOT NULL, + content BYTEA, + storage_uri TEXT, + size_bytes INTEGER NOT NULL, + sha256 TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uniq_minion_attachments_job_filename UNIQUE (job_id, filename), + CONSTRAINT chk_attachment_storage CHECK (content IS NOT NULL OR storage_uri IS NOT NULL), + CONSTRAINT chk_attachment_size CHECK (size_bytes >= 0) + ); + CREATE INDEX IF NOT EXISTS idx_minion_attachments_job ON minion_attachments (job_id); + + -- TOAST tuning: store attachment bytes out-of-line, skip compression. + -- Attachments are usually already-compressed formats; compression burns CPU for no win. + DO $$ + BEGIN + ALTER TABLE minion_attachments ALTER COLUMN content SET STORAGE EXTERNAL; + EXCEPTION WHEN OTHERS THEN + -- PGLite may not support SET STORAGE EXTERNAL. Storage tuning is an optimization, not correctness. + NULL; + END $$; + `, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/pglite-schema.ts b/src/core/pglite-schema.ts index 86c582672..351d2a798 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -185,6 +185,13 @@ CREATE TABLE IF NOT EXISTS minion_jobs ( tokens_input INTEGER NOT NULL DEFAULT 0, tokens_output INTEGER NOT NULL DEFAULT 0, tokens_cache_read INTEGER NOT NULL DEFAULT 0, + depth INTEGER NOT NULL DEFAULT 0, + max_children INTEGER, + timeout_ms INTEGER, + timeout_at TIMESTAMPTZ, + remove_on_complete BOOLEAN NOT NULL DEFAULT FALSE, + remove_on_fail BOOLEAN NOT NULL DEFAULT FALSE, + idempotency_key TEXT, result JSONB, progress JSONB, error_text TEXT, @@ -198,7 +205,10 @@ CREATE TABLE IF NOT EXISTS minion_jobs ( CONSTRAINT chk_on_child_fail CHECK (on_child_fail IN ('fail_parent','remove_dep','ignore','continue')), CONSTRAINT chk_jitter_range CHECK (backoff_jitter >= 0.0 AND backoff_jitter <= 1.0), CONSTRAINT chk_attempts_order CHECK (attempts_made <= attempts_started), - CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0) + CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0), + CONSTRAINT chk_depth_nonnegative CHECK (depth >= 0), + CONSTRAINT chk_max_children_positive CHECK (max_children IS NULL OR max_children > 0), + CONSTRAINT chk_timeout_positive CHECK (timeout_ms IS NULL OR timeout_ms > 0) ); CREATE INDEX IF NOT EXISTS idx_minion_jobs_claim ON minion_jobs (queue, priority ASC, created_at ASC) WHERE status = 'waiting'; @@ -206,6 +216,12 @@ CREATE INDEX IF NOT EXISTS idx_minion_jobs_status ON minion_jobs(status); CREATE INDEX IF NOT EXISTS idx_minion_jobs_stalled ON minion_jobs (lock_until) WHERE status = 'active'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_delayed ON minion_jobs (delay_until) WHERE status = 'delayed'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent ON minion_jobs(parent_job_id); +CREATE INDEX IF NOT EXISTS idx_minion_jobs_timeout ON minion_jobs (timeout_at) + WHERE status = 'active' AND timeout_at IS NOT NULL; +CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent_status ON minion_jobs (parent_job_id, status) + WHERE parent_job_id IS NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS uniq_minion_jobs_idempotency ON minion_jobs (idempotency_key) + WHERE idempotency_key IS NOT NULL; -- Inbox table for sidechannel messaging CREATE TABLE IF NOT EXISTS minion_inbox ( @@ -217,6 +233,27 @@ CREATE TABLE IF NOT EXISTS minion_inbox ( read_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_minion_inbox_unread ON minion_inbox (job_id) WHERE read_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_minion_inbox_child_done ON minion_inbox (job_id, sent_at) + WHERE (payload->>'type') = 'child_done'; + +-- Attachment manifest (BYTEA inline + forward-compat storage_uri) +CREATE TABLE IF NOT EXISTS minion_attachments ( + id SERIAL PRIMARY KEY, + job_id INTEGER NOT NULL REFERENCES minion_jobs(id) ON DELETE CASCADE, + filename TEXT NOT NULL, + content_type TEXT NOT NULL, + content BYTEA, + storage_uri TEXT, + size_bytes INTEGER NOT NULL, + sha256 TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uniq_minion_attachments_job_filename UNIQUE (job_id, filename), + CONSTRAINT chk_attachment_storage CHECK (content IS NOT NULL OR storage_uri IS NOT NULL), + CONSTRAINT chk_attachment_size CHECK (size_bytes >= 0) +); +CREATE INDEX IF NOT EXISTS idx_minion_attachments_job ON minion_attachments (job_id); +-- NOTE: SET STORAGE EXTERNAL is omitted on PGLite; it's a Postgres TOAST optimization +-- and PGLite may not support it. Postgres path applies it via migration v7. -- ============================================================ -- Trigger-based search_vector (spans pages + timeline_entries) diff --git a/src/core/schema-embedded.ts b/src/core/schema-embedded.ts index 3b68e8d8b..ca83290de 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -273,20 +273,33 @@ CREATE TABLE IF NOT EXISTS minion_jobs ( delay_until TIMESTAMPTZ, parent_job_id INTEGER REFERENCES minion_jobs(id) ON DELETE SET NULL, on_child_fail TEXT NOT NULL DEFAULT 'fail_parent', + tokens_input INTEGER NOT NULL DEFAULT 0, + tokens_output INTEGER NOT NULL DEFAULT 0, + tokens_cache_read INTEGER NOT NULL DEFAULT 0, result JSONB, progress JSONB, error_text TEXT, stacktrace JSONB DEFAULT '[]', + depth INTEGER NOT NULL DEFAULT 0, + max_children INTEGER, + timeout_ms INTEGER, + timeout_at TIMESTAMPTZ, + remove_on_complete BOOLEAN NOT NULL DEFAULT FALSE, + remove_on_fail BOOLEAN NOT NULL DEFAULT FALSE, + idempotency_key TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ, updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), - CONSTRAINT chk_status CHECK (status IN ('waiting','active','completed','failed','delayed','dead','cancelled','waiting-children')), + CONSTRAINT chk_status CHECK (status IN ('waiting','active','completed','failed','delayed','dead','cancelled','waiting-children','paused')), CONSTRAINT chk_backoff_type CHECK (backoff_type IN ('fixed','exponential')), CONSTRAINT chk_on_child_fail CHECK (on_child_fail IN ('fail_parent','remove_dep','ignore','continue')), CONSTRAINT chk_jitter_range CHECK (backoff_jitter >= 0.0 AND backoff_jitter <= 1.0), CONSTRAINT chk_attempts_order CHECK (attempts_made <= attempts_started), - CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0) + CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0), + CONSTRAINT chk_depth_nonnegative CHECK (depth >= 0), + CONSTRAINT chk_max_children_positive CHECK (max_children IS NULL OR max_children > 0), + CONSTRAINT chk_timeout_positive CHECK (timeout_ms IS NULL OR timeout_ms > 0) ); CREATE INDEX IF NOT EXISTS idx_minion_jobs_claim ON minion_jobs (queue, priority ASC, created_at ASC) WHERE status = 'waiting'; @@ -294,6 +307,54 @@ CREATE INDEX IF NOT EXISTS idx_minion_jobs_status ON minion_jobs(status); CREATE INDEX IF NOT EXISTS idx_minion_jobs_stalled ON minion_jobs (lock_until) WHERE status = 'active'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_delayed ON minion_jobs (delay_until) WHERE status = 'delayed'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent ON minion_jobs(parent_job_id); +CREATE INDEX IF NOT EXISTS idx_minion_jobs_timeout ON minion_jobs (timeout_at) WHERE status = 'active' AND timeout_at IS NOT NULL; +CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent_status ON minion_jobs (parent_job_id, status) WHERE parent_job_id IS NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS uniq_minion_jobs_idempotency ON minion_jobs (idempotency_key) WHERE idempotency_key IS NOT NULL; + +-- Inbox table for sidechannel messaging +CREATE TABLE IF NOT EXISTS minion_inbox ( + id SERIAL PRIMARY KEY, + job_id INTEGER NOT NULL REFERENCES minion_jobs(id) ON DELETE CASCADE, + sender TEXT NOT NULL, + payload JSONB NOT NULL, + sent_at TIMESTAMPTZ NOT NULL DEFAULT now(), + read_at TIMESTAMPTZ +); +CREATE INDEX IF NOT EXISTS idx_minion_inbox_unread ON minion_inbox (job_id) WHERE read_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_minion_inbox_child_done ON minion_inbox (job_id, sent_at) WHERE payload->>'type' = 'child_done'; + +-- Attachments table: per-job binary blobs (manifests, agent outputs, files) +CREATE TABLE IF NOT EXISTS minion_attachments ( + id SERIAL PRIMARY KEY, + job_id INTEGER NOT NULL REFERENCES minion_jobs(id) ON DELETE CASCADE, + filename TEXT NOT NULL, + content_type TEXT NOT NULL, + content BYTEA, + storage_uri TEXT, + size_bytes INTEGER NOT NULL, + sha256 TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uniq_minion_attachments_job_filename UNIQUE (job_id, filename), + CONSTRAINT chk_attachment_storage CHECK (content IS NOT NULL OR storage_uri IS NOT NULL), + CONSTRAINT chk_attachment_size CHECK (size_bytes >= 0) +); +CREATE INDEX IF NOT EXISTS idx_minion_attachments_job ON minion_attachments (job_id); +ALTER TABLE minion_attachments ALTER COLUMN content SET STORAGE EXTERNAL; + +-- NOTIFY trigger for real-time job events (Postgres only, not PGLite) +CREATE OR REPLACE FUNCTION notify_minion_job_change() RETURNS trigger AS \$\$ +BEGIN + PERFORM pg_notify('minion_jobs', json_build_object( + 'id', NEW.id, 'status', NEW.status, 'name', NEW.name, + 'queue', NEW.queue, 'prev_status', COALESCE(OLD.status, 'new') + )::text); + RETURN NEW; +END; +\$\$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS minion_job_notify ON minion_jobs; +CREATE TRIGGER minion_job_notify AFTER INSERT OR UPDATE OF status ON minion_jobs + FOR EACH ROW EXECUTE FUNCTION notify_minion_job_change(); -- ============================================================ -- Row Level Security: block anon access, postgres role bypasses diff --git a/src/schema.sql b/src/schema.sql index 4b1f48be4..18162031f 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -276,6 +276,13 @@ CREATE TABLE IF NOT EXISTS minion_jobs ( progress JSONB, error_text TEXT, stacktrace JSONB DEFAULT '[]', + depth INTEGER NOT NULL DEFAULT 0, + max_children INTEGER, + timeout_ms INTEGER, + timeout_at TIMESTAMPTZ, + remove_on_complete BOOLEAN NOT NULL DEFAULT FALSE, + remove_on_fail BOOLEAN NOT NULL DEFAULT FALSE, + idempotency_key TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ, @@ -285,7 +292,10 @@ CREATE TABLE IF NOT EXISTS minion_jobs ( CONSTRAINT chk_on_child_fail CHECK (on_child_fail IN ('fail_parent','remove_dep','ignore','continue')), CONSTRAINT chk_jitter_range CHECK (backoff_jitter >= 0.0 AND backoff_jitter <= 1.0), CONSTRAINT chk_attempts_order CHECK (attempts_made <= attempts_started), - CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0) + CONSTRAINT chk_nonnegative CHECK (attempts_made >= 0 AND attempts_started >= 0 AND stalled_counter >= 0 AND max_attempts >= 1 AND max_stalled >= 0), + CONSTRAINT chk_depth_nonnegative CHECK (depth >= 0), + CONSTRAINT chk_max_children_positive CHECK (max_children IS NULL OR max_children > 0), + CONSTRAINT chk_timeout_positive CHECK (timeout_ms IS NULL OR timeout_ms > 0) ); CREATE INDEX IF NOT EXISTS idx_minion_jobs_claim ON minion_jobs (queue, priority ASC, created_at ASC) WHERE status = 'waiting'; @@ -293,6 +303,9 @@ CREATE INDEX IF NOT EXISTS idx_minion_jobs_status ON minion_jobs(status); CREATE INDEX IF NOT EXISTS idx_minion_jobs_stalled ON minion_jobs (lock_until) WHERE status = 'active'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_delayed ON minion_jobs (delay_until) WHERE status = 'delayed'; CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent ON minion_jobs(parent_job_id); +CREATE INDEX IF NOT EXISTS idx_minion_jobs_timeout ON minion_jobs (timeout_at) WHERE status = 'active' AND timeout_at IS NOT NULL; +CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent_status ON minion_jobs (parent_job_id, status) WHERE parent_job_id IS NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS uniq_minion_jobs_idempotency ON minion_jobs (idempotency_key) WHERE idempotency_key IS NOT NULL; -- Inbox table for sidechannel messaging CREATE TABLE IF NOT EXISTS minion_inbox ( @@ -304,6 +317,25 @@ CREATE TABLE IF NOT EXISTS minion_inbox ( read_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_minion_inbox_unread ON minion_inbox (job_id) WHERE read_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_minion_inbox_child_done ON minion_inbox (job_id, sent_at) WHERE payload->>'type' = 'child_done'; + +-- Attachments table: per-job binary blobs (manifests, agent outputs, files) +CREATE TABLE IF NOT EXISTS minion_attachments ( + id SERIAL PRIMARY KEY, + job_id INTEGER NOT NULL REFERENCES minion_jobs(id) ON DELETE CASCADE, + filename TEXT NOT NULL, + content_type TEXT NOT NULL, + content BYTEA, + storage_uri TEXT, + size_bytes INTEGER NOT NULL, + sha256 TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uniq_minion_attachments_job_filename UNIQUE (job_id, filename), + CONSTRAINT chk_attachment_storage CHECK (content IS NOT NULL OR storage_uri IS NOT NULL), + CONSTRAINT chk_attachment_size CHECK (size_bytes >= 0) +); +CREATE INDEX IF NOT EXISTS idx_minion_attachments_job ON minion_attachments (job_id); +ALTER TABLE minion_attachments ALTER COLUMN content SET STORAGE EXTERNAL; -- NOTIFY trigger for real-time job events (Postgres only, not PGLite) CREATE OR REPLACE FUNCTION notify_minion_job_change() RETURNS trigger AS $$ @@ -316,6 +348,7 @@ BEGIN END; $$ LANGUAGE plpgsql; +DROP TRIGGER IF EXISTS minion_job_notify ON minion_jobs; CREATE TRIGGER minion_job_notify AFTER INSERT OR UPDATE OF status ON minion_jobs FOR EACH ROW EXECUTE FUNCTION notify_minion_job_change();