admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-07-30 19:49:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(minions): migration v7 — agent_parity_layer schema

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Garry Tan
2026-04-18 06:31:38 +08:00
co-authored by Claude Opus 4.7
parent 5fed5dcb1d
commit da65b3af45
4 changed files with 200 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/migrate.ts
+65
View File
@@ -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
src/core/pglite-schema.ts
+38 -1
View File
@@ -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)
src/core/schema-embedded.ts
+63 -2
View File
@@ -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
src/schema.sql
+34 -1
View File
@@ -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();