mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-28 06:23:01 +00:00
feat: Minions v2 — agent orchestration primitives (pause/resume, inbox, tokens, replay)
Adds the foundation for Minions as universal agent orchestration infrastructure.
GBrain's Postgres-native job queue now supports durable, observable, steerable
background agents. The OpenClaw plugin (separate repo) will consume these via
library import, not MCP, for zero-latency local integration.
## New capabilities
- **Concurrent worker** — Promise pool replaces sequential loop. Per-job
AbortController for cooperative cancellation. Graceful shutdown waits for
all in-flight jobs via Promise.allSettled.
- **Pause/resume** — pauseJob clears the lock and fires AbortSignal on active
jobs. Handlers check ctx.signal.aborted and exit cleanly. resumeJob returns
paused jobs to waiting. Catch block skips failJob when signal.aborted.
- **Inbox (separate table)** — minion_inbox table for sidechannel messages.
sendMessage with sender validation (parent job or admin). readInbox is
token-fenced and marks read_at atomically. Separate table avoids row bloat
from rewriting JSONB on every send.
- **Token accounting** — tokens_input/tokens_output/tokens_cache_read columns.
updateTokens accumulates; completeJob rolls child tokens up to parent.
USD cost computed at read time (no cost_usd column — pricing too volatile).
- **Job replay** — replayJob clones a terminal job with optional data overrides.
New job, fresh attempts, no parent link.
## Handler contract additions
MinionJobContext now provides:
- `signal: AbortSignal` — cooperative cancellation
- `updateTokens(tokens)` — accumulate token usage
- `readInbox()` — check for sidechannel messages
- `log()` — now accepts string or TranscriptEntry
## MCP operations added
pause_job, resume_job, replay_job, send_job_message — all auto-generate CLI
commands and MCP server endpoints.
## Library exports
package.json exports map adds ./minions and ./engine-factory paths so plugins
can `import { MinionQueue } from 'gbrain/minions'` for direct library use.
## Instruction layer (the teaching)
- skills/minion-orchestrator/SKILL.md — when/how to use Minions, decision
matrix, lifecycle management, anti-patterns
- skills/conventions/subagent-routing.md — cross-cutting rule: all background
work goes through Minions
- RESOLVER.md — trigger entries for agent orchestration
- manifest.json — registered
## Schema migration v6
Additive: 3 token columns, paused status, minion_inbox table with unread index.
Full Postgres + PGLite support. No backfill needed.
## Tests
65 tests (was 43): pause/resume (5), inbox (6), tokens (4), replay (4),
concurrent worker context (3), plus all existing coverage.
## What's NOT in this commit
Deferred to follow-up PRs:
- LISTEN/NOTIFY subscribe (needs real Postgres E2E)
- Resource governor (depends on concurrent worker stress testing)
- Routing eval harness (needs API keys + benchmark data)
- OpenClaw plugin (separate @gbrain/openclaw-minions-plugin repo)
See docs/designs/MINIONS_AGENT_ORCHESTRATION.md for full CEO-approved design.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
34529ac60f
commit
79d09bc3e6
@@ -0,0 +1,448 @@
|
||||
---
|
||||
status: ACTIVE
|
||||
---
|
||||
# CEO Plan: Minions as Universal Agent Orchestration Protocol
|
||||
Generated by /plan-ceo-review on 2026-04-15
|
||||
Branch: garrytan/minions-jobs | Mode: SCOPE EXPANSION
|
||||
Repo: garrytan/gbrain
|
||||
|
||||
## Vision
|
||||
|
||||
### 10x Check
|
||||
Instead of "GBrain has a queue, OpenClaw uses it," make Minions a universal agent
|
||||
orchestration protocol. Any platform (OpenClaw, Hermes, Claude Code, Codex, custom
|
||||
scripts) submits, monitors, steers, and composes agents through the same Postgres-native
|
||||
protocol. GBrain IS the agent control plane.
|
||||
|
||||
### Platonic Ideal (aspirational North Star, NOT in v1 scope)
|
||||
Open a terminal, type `gbrain jobs dashboard`. See every agent across every platform.
|
||||
Their progress, tool calls, token spend. Click any agent for full execution trace.
|
||||
Type a message to redirect a running agent mid-flight. See the governor's decisions
|
||||
visualized. Run A/B tests between agent configurations. The feeling: complete
|
||||
situational awareness of your AI workforce.
|
||||
|
||||
**Note:** The dashboard, A/B testing, and visual governor are future phases. This plan
|
||||
builds the primitives they would sit on top of: real-time events, structured progress,
|
||||
token accounting, inbox with ack, and session transcripts.
|
||||
|
||||
## Scope Decisions
|
||||
|
||||
| # | Proposal | Effort | Decision | Reasoning |
|
||||
|---|----------|--------|----------|-----------|
|
||||
| 1 | pg LISTEN/NOTIFY real-time events | S | ACCEPTED | Sub-second event delivery vs 5s polling. Every platform benefits. |
|
||||
| 2 | Structured progress protocol | S | ACCEPTED | Standard progress makes unified dashboard possible. |
|
||||
| 3 | Job cost tracking (token accounting) | M | ACCEPTED | Token cost is #1 thing users want to know about agent work. |
|
||||
| 4 | Job replay | S | ACCEPTED | Small surface area, high utility for debugging failures. |
|
||||
| 5 | Job groups / waves | M | DEFERRED | Parent-child already provides grouping. Overlap concern. |
|
||||
| 6 | Inbox acknowledgment (read receipts) | S | ACCEPTED | Without it, inbox is fire-and-forget — same problem we're fixing. |
|
||||
| 7 | Universal agent protocol | S | ACCEPTED | Design framing, not extra code. Platform-agnostic naming/docs. |
|
||||
| 8 | Session transcript capture | M | ACCEPTED | Full audit trail of every agent run. |
|
||||
|
||||
## Accepted Scope — Implementation Detail
|
||||
|
||||
### 0a. Pause/resume (from base plan)
|
||||
|
||||
**Schema:** Add `'paused'` to `MinionJobStatus` (already in migration v6 constraint).
|
||||
|
||||
**New methods:**
|
||||
- `MinionQueue.pauseJob(id): MinionJob | null`
|
||||
Transitions `waiting` or `active` → `paused`. For `active` jobs, clears `lock_token`
|
||||
and `lock_until` (worker will detect lock loss and stop). Returns null if job not
|
||||
in pausable state.
|
||||
- `MinionQueue.resumeJob(id): MinionJob | null`
|
||||
Transitions `paused` → `waiting`. Resets for claiming. Returns null if not paused.
|
||||
|
||||
**Worker integration:** Worker's lock renewal loop checks `isActive()`. When a job
|
||||
is paused, the lock is cleared, so `renewLock()` returns false and the worker stops
|
||||
execution gracefully (same path as stall detection). The job's progress and state
|
||||
are preserved in the DB for when it resumes.
|
||||
|
||||
**MCP operations:** `pause_job`, `resume_job` (added in Step 3 of implementation plan).
|
||||
|
||||
**PGLite compatibility:** Full.
|
||||
|
||||
### 0b. Resource governor (from base plan)
|
||||
|
||||
**New file:** `src/core/minions/governor.ts`
|
||||
|
||||
```typescript
|
||||
interface GovernorConfig {
|
||||
maxConcurrency: number; // ceiling
|
||||
minConcurrency: number; // floor (default 1)
|
||||
checkIntervalMs: number; // default 10000
|
||||
cpuThreshold: number; // default 0.80 (80%)
|
||||
memoryThreshold: number; // default 0.85 (85%)
|
||||
circuitBreakerMemory: number; // default 0.90 (90%)
|
||||
}
|
||||
|
||||
class ResourceGovernor {
|
||||
getEffectiveConcurrency(): number; // current allowed concurrency
|
||||
start(): void; // begin polling system metrics
|
||||
stop(): void; // stop polling
|
||||
onCircuitBreak(cb: (jobId) => void): void; // kill callback
|
||||
}
|
||||
```
|
||||
|
||||
**System metrics:** Reuse `getSystemLoad()` from `src/core/backoff.ts` (already
|
||||
implements CPU and memory checks). Add event loop lag measurement via
|
||||
`perf_hooks.monitorEventLoopDelay()`.
|
||||
|
||||
**Worker integration:** `MinionWorker.start()` consults `governor.getEffectiveConcurrency()`
|
||||
before claiming new jobs. If current in-flight count >= effective concurrency, skip claim.
|
||||
|
||||
**Circuit breaker:** If memory > 90%, governor calls `onCircuitBreak` with the
|
||||
lowest-priority active job ID. Worker cancels that job via `failJob()` with
|
||||
`UnrecoverableError("circuit breaker: memory pressure")`.
|
||||
|
||||
**Prerequisite:** Concurrent job processing must be implemented first (see
|
||||
Concurrency Note below).
|
||||
|
||||
**PGLite compatibility:** Full (governor is app-level, not DB-level).
|
||||
|
||||
### 1. pg LISTEN/NOTIFY (real-time events)
|
||||
|
||||
**Schema:** No new columns. Add NOTIFY triggers to state transitions.
|
||||
|
||||
**SQL trigger:**
|
||||
```sql
|
||||
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;
|
||||
|
||||
CREATE TRIGGER minion_job_notify AFTER INSERT OR UPDATE OF status ON minion_jobs
|
||||
FOR EACH ROW EXECUTE FUNCTION notify_minion_job_change();
|
||||
```
|
||||
|
||||
**New method:** `MinionQueue.subscribe(callback: (event) => void): () => void`
|
||||
Returns unsubscribe function. Requires direct Postgres connection (NOT pooled).
|
||||
|
||||
**PGLite compatibility:** PGLite does NOT support LISTEN/NOTIFY. Fallback: polling
|
||||
via `getJob()` at configurable interval (default 2s). The `subscribe()` method
|
||||
detects engine type and uses polling fallback automatically.
|
||||
|
||||
**Supabase constraint:** Requires direct connection (port 5432), not pgBouncer
|
||||
pooler (port 6543). Document in skill file and setup guide.
|
||||
|
||||
### 2. Structured progress protocol
|
||||
|
||||
**TypeScript interface (convention, not enforced at DB level):**
|
||||
```typescript
|
||||
interface AgentProgress {
|
||||
step: number; // current step (1-based)
|
||||
total: number; // total expected steps (0 = unknown)
|
||||
message: string; // human-readable status
|
||||
tokens_in: number; // cumulative input tokens
|
||||
tokens_out: number; // cumulative output tokens
|
||||
last_tool: string; // name of last tool called
|
||||
started_at: string; // ISO 8601 when this step started
|
||||
}
|
||||
```
|
||||
|
||||
**Storage:** Existing `progress JSONB` column. No schema change needed.
|
||||
Handlers use `ctx.updateProgress(agentProgress)`. Non-agent jobs can use
|
||||
any JSONB shape (backward compatible).
|
||||
|
||||
**Validation:** `updateProgress()` accepts any JSONB. The `AgentProgress`
|
||||
interface is a convention enforced by the agent handler, not by the queue.
|
||||
|
||||
### 3. Job cost tracking (token accounting)
|
||||
|
||||
**Schema changes (migration v6):**
|
||||
```sql
|
||||
ALTER TABLE minion_jobs ADD COLUMN tokens_input INTEGER DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN tokens_output INTEGER DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN tokens_cache_read INTEGER DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN cost_usd NUMERIC(10,6) DEFAULT 0;
|
||||
```
|
||||
|
||||
**New method:** `MinionQueue.updateTokens(id, lockToken, { input, output, cache_read, cost_usd })`
|
||||
Accumulates (adds to existing values, does not replace).
|
||||
|
||||
**Parent rollup:** When `completeJob()` is called, if `parent_job_id` is set,
|
||||
add this job's token counts to the parent's via:
|
||||
```sql
|
||||
UPDATE minion_jobs SET
|
||||
tokens_input = tokens_input + $child_input,
|
||||
tokens_output = tokens_output + $child_output,
|
||||
tokens_cache_read = tokens_cache_read + $child_cache,
|
||||
cost_usd = cost_usd + $child_cost
|
||||
WHERE id = $parent_id;
|
||||
```
|
||||
|
||||
**PGLite compatibility:** Full support (standard columns).
|
||||
|
||||
### 4. Job replay
|
||||
|
||||
**New method:** `MinionQueue.replayJob(id, dataOverrides?: Record<string, unknown>): MinionJob`
|
||||
|
||||
Implementation: Read the completed/failed/dead job. Create a NEW job with:
|
||||
- Same `name`, `queue`, `priority`, `max_attempts`, `backoff_type`, `backoff_delay`
|
||||
- `data` = deep merge of original data + overrides
|
||||
- Fresh `attempts_made: 0`, `status: 'waiting'`
|
||||
- `parent_job_id` = null (replay is a new top-level job, not a child)
|
||||
- Does NOT clone children (replay is a single job, not a DAG)
|
||||
|
||||
**Constraint:** Only works on terminal statuses (completed/failed/dead).
|
||||
Returns the new job record.
|
||||
|
||||
**Idempotency:** Each replay creates a distinct new job. No deduplication.
|
||||
If the original had side effects, the replay may repeat them. Document this
|
||||
in the skill file as a user responsibility.
|
||||
|
||||
### 5. Inbox (sidechannel messaging)
|
||||
|
||||
**Schema changes (migration v6):**
|
||||
```sql
|
||||
ALTER TABLE minion_jobs ADD COLUMN inbox JSONB DEFAULT '[]';
|
||||
```
|
||||
|
||||
**Inbox message format:**
|
||||
```typescript
|
||||
interface InboxMessage {
|
||||
id: string; // UUIDv4
|
||||
sent_at: string; // ISO 8601
|
||||
read_at: string | null; // null until worker reads it
|
||||
sender: string; // 'parent' | 'user' | job ID
|
||||
payload: unknown; // arbitrary directive
|
||||
}
|
||||
```
|
||||
|
||||
**New methods:**
|
||||
- `MinionQueue.sendMessage(jobId, payload, sender?): InboxMessage`
|
||||
Appends message to inbox array via atomic JSONB append
|
||||
(`inbox = inbox || $1::jsonb`), not read-modify-write. Returns the message with id + sent_at.
|
||||
- `MinionQueue.readInbox(jobId, lockToken): InboxMessage[]`
|
||||
Returns unread messages (read_at = null). Marks them as read (sets read_at).
|
||||
Token-fenced: only the worker holding the lock can read.
|
||||
|
||||
**Worker integration:** Agent handler calls `readInbox()` on each iteration.
|
||||
If messages exist, injects them into the agent's context as system messages.
|
||||
|
||||
**PGLite compatibility:** Full support (standard JSONB column).
|
||||
|
||||
### 6. Inbox acknowledgment (read receipts)
|
||||
|
||||
Built into the inbox design above. The `read_at` field on each `InboxMessage`
|
||||
provides the receipt. `sendMessage()` returns the message ID; the sender can
|
||||
later check `getJob(id)` and inspect `inbox` to see which messages have been
|
||||
read.
|
||||
|
||||
No additional schema or methods needed beyond what's in #5.
|
||||
|
||||
### 7. Universal agent protocol (platform-agnostic framing)
|
||||
|
||||
**This is a design decision, not code.** It means:
|
||||
|
||||
1. The skill file (`skills/minion-orchestrator/SKILL.md`) is written for ANY
|
||||
agent platform, not just OpenClaw. Examples show MCP tool calls, not
|
||||
OpenClaw-specific commands.
|
||||
|
||||
2. The agent handler (`agent-handler.ts`) accepts a generic interface:
|
||||
```typescript
|
||||
interface AgentJobData {
|
||||
prompt: string;
|
||||
tools?: string[]; // MCP tool names
|
||||
model?: string; // e.g., 'claude-opus-4-6', 'gpt-4o'
|
||||
context?: string; // additional context
|
||||
platform?: string; // 'openclaw' | 'hermes' | 'claude-code' | 'custom'
|
||||
max_iterations?: number; // agent loop budget
|
||||
}
|
||||
```
|
||||
|
||||
3. The OpenClaw plugin is ONE consumer. Hermes, Claude Code extensions,
|
||||
or custom scripts can submit `agent` jobs through the same MCP operations.
|
||||
|
||||
4. **NOT in v1 scope:** Multi-tenant auth, cross-network connectivity,
|
||||
protocol versioning, API key isolation. These are Phase 2 concerns when
|
||||
actual multi-platform usage materializes. v1 is single-user, single-brain.
|
||||
|
||||
### Agent Handler Architecture (critical design decision)
|
||||
|
||||
The agent handler does NOT live in GBrain. GBrain provides the queue infrastructure
|
||||
and a clean handler contract. The actual agent execution lives in the platform plugin.
|
||||
|
||||
```
|
||||
GBrain (this repo):
|
||||
MinionQueue — queue/claim/complete/inbox/tokens/NOTIFY
|
||||
MinionWorker — poll/lock/stall/governor framework
|
||||
Handler contract — AgentJobData interface + MinionJobContext
|
||||
|
||||
OpenClaw plugin (separate repo):
|
||||
Registers "agent" handler with MinionWorker
|
||||
Handler calls OpenClaw's PI agent core (the actual LLM loop)
|
||||
Each iteration: readInbox → inject as system message, updateProgress, updateTokens
|
||||
Completion: store result + session transcript in job.result + job.stacktrace
|
||||
|
||||
GBrain ships a test/echo handler for unit testing only.
|
||||
```
|
||||
|
||||
**Handler contract (GBrain side):**
|
||||
```typescript
|
||||
// The handler receives this context (already exists in worker.ts)
|
||||
interface MinionJobContext {
|
||||
id: number;
|
||||
name: string;
|
||||
data: Record<string, unknown>; // AgentJobData when name="agent"
|
||||
attempts_made: number;
|
||||
updateProgress(progress: unknown): Promise<void>;
|
||||
updateTokens(tokens: TokenUpdate): Promise<void>; // NEW
|
||||
log(message: string | TranscriptEntry): Promise<void>;
|
||||
isActive(): Promise<boolean>;
|
||||
readInbox(): Promise<InboxMessage[]>; // NEW
|
||||
}
|
||||
```
|
||||
|
||||
**Why this is right:** GBrain is orchestration, not execution. OpenClaw has the
|
||||
PI agent core. Hermes has AIAgent. Claude Code has its own loop. Each platform
|
||||
brings its own engine and registers a handler. GBrain manages lifecycle, progress,
|
||||
steering, cost tracking, and persistence around it.
|
||||
|
||||
### 8. Session transcript capture
|
||||
|
||||
**Extends existing stacktrace mechanism.** The `stacktrace` field (JSONB array
|
||||
of strings) already captures log messages. Session transcripts use the same
|
||||
field with structured entries:
|
||||
|
||||
```typescript
|
||||
type TranscriptEntry =
|
||||
| { type: 'log'; message: string; ts: string }
|
||||
| { type: 'tool_call'; tool: string; args_size: number; result_size: number; ts: string }
|
||||
| { type: 'llm_turn'; model: string; tokens_in: number; tokens_out: number; ts: string }
|
||||
| { type: 'error'; message: string; stack?: string; ts: string };
|
||||
```
|
||||
|
||||
**Storage:** Existing `stacktrace JSONB` column. No schema change.
|
||||
The agent handler appends `TranscriptEntry` objects instead of plain strings.
|
||||
Backward compatible: non-agent jobs continue appending strings.
|
||||
|
||||
**Size concern:** Long agent runs could generate large transcripts. Add a
|
||||
`max_transcript_entries` option (default 1000) that rotates oldest entries
|
||||
when exceeded (FIFO). The full transcript for forensic analysis can be
|
||||
stored as a brain file via `gbrain files upload-raw`.
|
||||
|
||||
## Schema Migration v6
|
||||
|
||||
All schema changes are additive (ALTER TABLE ADD COLUMN). No backfill needed.
|
||||
Existing jobs continue to work with default values.
|
||||
|
||||
```sql
|
||||
-- Migration v6: Agent orchestration primitives
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_input INTEGER DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_output INTEGER DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_cache_read INTEGER DEFAULT 0;
|
||||
|
||||
-- Separate inbox table (not JSONB on job row)
|
||||
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;
|
||||
|
||||
-- Status constraint update: add 'paused'
|
||||
ALTER TABLE minion_jobs DROP CONSTRAINT IF EXISTS minion_jobs_status_check;
|
||||
ALTER TABLE minion_jobs ADD CONSTRAINT minion_jobs_status_check
|
||||
CHECK (status IN ('waiting','active','completed','failed','delayed','dead','cancelled','waiting-children','paused'));
|
||||
|
||||
-- NOTIFY trigger for real-time 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;
|
||||
|
||||
CREATE TRIGGER minion_job_notify AFTER INSERT OR UPDATE OF status ON minion_jobs
|
||||
FOR EACH ROW EXECUTE FUNCTION notify_minion_job_change();
|
||||
```
|
||||
|
||||
## PGLite Compatibility Matrix
|
||||
|
||||
| Feature | Postgres | PGLite | Fallback |
|
||||
|---|---|---|---|
|
||||
| Pause/resume | Full | Full | — |
|
||||
| Inbox + ack | Full | Full | — |
|
||||
| Token accounting | Full | Full | — |
|
||||
| Job replay | Full | Full | — |
|
||||
| LISTEN/NOTIFY | Full | NO | Polling (2s interval) |
|
||||
| NOTIFY trigger | Full | NO | Skipped in PGLite schema |
|
||||
| Structured progress | Full | Full | — |
|
||||
| Session transcripts | Full | Full | — |
|
||||
| Resource governor | Full | Full | — |
|
||||
| Worker daemon | Full | NO (existing limitation) | — |
|
||||
|
||||
## Concurrency Note
|
||||
|
||||
The current `MinionWorker.start()` processes jobs sequentially (one at a time)
|
||||
despite `concurrency` being declared in `MinionWorkerOpts`. Implementing actual
|
||||
concurrent job processing (Promise pool) is a prerequisite for the resource
|
||||
governor to be meaningful. The governor adjusts effective concurrency, which
|
||||
requires actual concurrent processing to exist.
|
||||
|
||||
**Action:** Implement concurrent job processing in `worker.ts` before or as
|
||||
part of the governor step. Use a semaphore pattern: maintain up to N in-flight
|
||||
promises, claim new jobs as slots free up.
|
||||
|
||||
## Outside Voice Decisions (from adversarial review)
|
||||
|
||||
1. **AbortController for pause/resume** — Handler contract gets `signal: AbortSignal`.
|
||||
Pause clears lock AND signals abort. Handler must check `signal.aborted` on each
|
||||
iteration. Without this, pausing active jobs creates duplicate execution.
|
||||
|
||||
2. **Drop cost_usd column** — Token counts (input/output/cache_read) are stable facts.
|
||||
USD pricing is volatile. Compute cost at display/read time from a pricing table,
|
||||
not at write time. Removes `cost_usd NUMERIC(10,6)` from migration v6.
|
||||
|
||||
3. **Separate minion_inbox table** — Instead of JSONB array on job row, use a dedicated
|
||||
table for inbox messages. Avoids row bloat from rewriting entire inbox on every send.
|
||||
Properly concurrent-safe with standard INSERT (no JSONB append concerns).
|
||||
```sql
|
||||
CREATE TABLE 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 idx_minion_inbox_unread ON minion_inbox (job_id) WHERE read_at IS NULL;
|
||||
```
|
||||
|
||||
4. **One release, not two** — Ship all features in one migration (v6). User prefers
|
||||
cohesive release over incremental delivery for this feature set.
|
||||
|
||||
5. **Selective column projection** — Fix SELECT * queries in getJobs(), claim(),
|
||||
handleStalled() to exclude stacktrace column. Include stacktrace only in getJob()
|
||||
detail view. Prevents transcript bloat from affecting query performance.
|
||||
|
||||
## Future Phases (accepted trajectory)
|
||||
|
||||
- **Phase 2: Dashboard CLI** — `gbrain jobs dashboard` live TUI showing all agents.
|
||||
Enabled by: LISTEN/NOTIFY, structured progress, token accounting.
|
||||
- **Phase 3: Multi-tenant auth** — Runtime MCP access control, per-platform API keys.
|
||||
Enabled by: platform-agnostic framing, sender validation on inbox.
|
||||
- **Phase 4: Agent composition patterns** — Map-reduce, pipeline, approval gates as
|
||||
first-class primitives. Enabled by: parent-child DAGs, inbox sidechannel.
|
||||
|
||||
## Deferred to TODOS.md
|
||||
- Job groups / waves (parent-child covers this; revisit if real grouping need emerges)
|
||||
- cost_usd column (compute from pricing table at read time when pricing API exists)
|
||||
|
||||
## Key Premises Confirmed
|
||||
1. GBrain is intentionally evolving from knowledge brain to agent infrastructure (user confirmed)
|
||||
2. Coupling between OpenClaw and GBrain's Postgres is acceptable (OpenClaw already depends on GBrain)
|
||||
3. Full Infrastructure approach (all 8+ steps) selected over Minimal Viable or Sidecar Tracking
|
||||
4. Prior learning [agent-dx-instruction-layer] validates that the teaching layer (skill + evals) is mandatory
|
||||
+3
-1
@@ -11,7 +11,9 @@
|
||||
".": "./src/core/index.ts",
|
||||
"./engine": "./src/core/engine.ts",
|
||||
"./types": "./src/core/types.ts",
|
||||
"./operations": "./src/core/operations.ts"
|
||||
"./operations": "./src/core/operations.ts",
|
||||
"./minions": "./src/core/minions/index.ts",
|
||||
"./engine-factory": "./src/core/engine-factory.ts"
|
||||
},
|
||||
"scripts": {
|
||||
"dev": "bun run src/cli.ts",
|
||||
|
||||
@@ -54,6 +54,7 @@ This is the dispatcher. Skills are the implementation. **Read the skill file bef
|
||||
| Cross-modal review, second opinion | `skills/cross-modal-review/SKILL.md` |
|
||||
| "Validate skills", skill health check | `skills/testing/SKILL.md` |
|
||||
| Webhook setup, external event processing | `skills/webhook-transforms/SKILL.md` |
|
||||
| "Spawn agent", "background task", "parallel tasks", "steer agent", "pause/resume agent" | `skills/minion-orchestrator/SKILL.md` |
|
||||
|
||||
## Setup & migration
|
||||
|
||||
@@ -87,5 +88,6 @@ When multiple skills could match:
|
||||
These apply to ALL brain-writing skills:
|
||||
- `skills/conventions/quality.md` — citations, back-links, notability gate
|
||||
- `skills/conventions/brain-first.md` — check brain before external APIs
|
||||
- `skills/conventions/subagent-routing.md` — when to use Minions vs inline work
|
||||
- `skills/_brain-filing-rules.md` — where files go
|
||||
- `skills/_output-rules.md` — output quality standards
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
# Subagent Routing Convention
|
||||
|
||||
## Rule: Minions for all background work
|
||||
|
||||
All background agent work MUST go through Minions (`submit_job`), never
|
||||
`sessions_spawn` with `runtime: "subagent"`. The native subagent runtime
|
||||
has no persistence, no observability, and no recovery.
|
||||
|
||||
Reference: `skills/minion-orchestrator/SKILL.md`
|
||||
|
||||
## Decision matrix
|
||||
|
||||
| Condition | Action |
|
||||
|---|---|
|
||||
| Single tool call, < 30s | Do it inline |
|
||||
| Multi-step, any duration | Submit as Minion job |
|
||||
| Parallel work (2+ streams) | Submit N Minion jobs with shared parent |
|
||||
| Needs to survive restart | Submit as Minion job |
|
||||
| User wants progress updates | Submit as Minion job with progress tracking |
|
||||
| Research / bulk operation | Submit as Minion job, always |
|
||||
|
||||
## Concurrency budget
|
||||
|
||||
- Check `get_job_stats` queue_health.active before submitting batch jobs
|
||||
- If active > 5, submit remaining jobs with `delay` to stagger execution
|
||||
- The resource governor auto-throttles, but don't dump 20 jobs at once
|
||||
@@ -128,6 +128,11 @@
|
||||
"name": "data-research",
|
||||
"path": "data-research/SKILL.md",
|
||||
"description": "Structured data research: search, extract, archive, deduplicate, track. Parameterized YAML recipes for investor updates, donations, company metrics."
|
||||
},
|
||||
{
|
||||
"name": "minion-orchestrator",
|
||||
"path": "minion-orchestrator/SKILL.md",
|
||||
"description": "Manage background agents via Minions job queue. Submit, monitor, steer, pause/resume, replay. Replaces sessions_spawn for durable observable agents."
|
||||
}
|
||||
],
|
||||
"dependencies": {
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
---
|
||||
name: minion-orchestrator
|
||||
version: 1.0.0
|
||||
description: |
|
||||
Manage background agents via Minions job queue. Use when: spawning subagents,
|
||||
checking agent progress, steering running agents, pausing/resuming work,
|
||||
parallel task execution, fan-out research. Replaces sessions_spawn for
|
||||
durable, observable, steerable agents.
|
||||
triggers:
|
||||
- "spawn agent"
|
||||
- "background task"
|
||||
- "run in background"
|
||||
- "check on agent"
|
||||
- "agent progress"
|
||||
- "what's running"
|
||||
- "steer agent"
|
||||
- "change direction"
|
||||
- "tell the agent"
|
||||
- "pause agent"
|
||||
- "stop agent"
|
||||
- "resume agent"
|
||||
- "parallel tasks"
|
||||
- "fan out"
|
||||
- "do these in parallel"
|
||||
tools:
|
||||
- submit_job
|
||||
- get_job
|
||||
- list_jobs
|
||||
- cancel_job
|
||||
- pause_job
|
||||
- resume_job
|
||||
- replay_job
|
||||
- send_job_message
|
||||
- get_job_progress
|
||||
- get_job_stats
|
||||
mutating: true
|
||||
---
|
||||
|
||||
# Minion Orchestrator
|
||||
|
||||
## Contract
|
||||
|
||||
Minions is a Postgres-native job queue for durable, observable agent orchestration.
|
||||
Every background agent task goes through Minions. No in-memory subagent spawning.
|
||||
|
||||
Guarantees:
|
||||
- Jobs survive gateway restart (Postgres-backed)
|
||||
- Every job has structured progress, token accounting, and session transcripts
|
||||
- Running agents can be steered mid-flight via inbox messages
|
||||
- Jobs can be paused, resumed, or cancelled at any time
|
||||
- Parent-child DAGs with configurable failure policies
|
||||
|
||||
## When to Use Minions vs Inline Work
|
||||
|
||||
| Condition | Action |
|
||||
|---|---|
|
||||
| Single tool call, < 30s | Do it inline |
|
||||
| Multi-step, any duration | Submit as Minion job |
|
||||
| Parallel work (2+ streams) | Submit N Minion jobs with shared parent |
|
||||
| Needs to survive restart | Submit as Minion job |
|
||||
| User wants progress updates | Submit as Minion job with progress tracking |
|
||||
| Research / bulk operation | Submit as Minion job, always |
|
||||
| File imports, bulk embeds | Submit as Minion job |
|
||||
|
||||
**Rule of thumb:** If it takes more than 3 tool calls, use a Minion.
|
||||
|
||||
## Phase 1: Submit
|
||||
|
||||
```
|
||||
submit_job name="research" data={"prompt":"Research Acme Corp revenue","tools":["search","web_search"]}
|
||||
```
|
||||
|
||||
Options:
|
||||
- `queue` — queue name (default: 'default')
|
||||
- `priority` — lower = higher priority (default: 0)
|
||||
- `max_attempts` — retry limit (default: 3)
|
||||
- `delay` — ms delay before eligible
|
||||
|
||||
For parallel work, submit a parent then children:
|
||||
```
|
||||
submit_job name="orchestrate" data={"task":"research 5 companies"}
|
||||
# Returns parent_id
|
||||
|
||||
submit_job name="research" data={"company":"Acme"} parent_job_id=PARENT_ID
|
||||
submit_job name="research" data={"company":"Beta"} parent_job_id=PARENT_ID
|
||||
submit_job name="research" data={"company":"Gamma"} parent_job_id=PARENT_ID
|
||||
```
|
||||
|
||||
Parent auto-enters `waiting-children` and unblocks when all children finish.
|
||||
|
||||
## Phase 2: Monitor
|
||||
|
||||
```
|
||||
list_jobs --status active # what's running?
|
||||
get_job ID # full details + logs + tokens
|
||||
get_job_progress ID # structured progress snapshot
|
||||
get_job_stats # health dashboard
|
||||
```
|
||||
|
||||
Progress includes: step count, total steps, message, token usage, last tool called.
|
||||
|
||||
## Phase 3: Steer
|
||||
|
||||
Send a message to redirect a running agent:
|
||||
```
|
||||
send_job_message id=ID payload={"directive":"focus on revenue, skip headcount"}
|
||||
```
|
||||
|
||||
The agent handler reads inbox messages on each iteration and injects them as
|
||||
context. Messages are acknowledged (read receipts tracked).
|
||||
|
||||
Only the parent job or admin can send messages (sender validation).
|
||||
|
||||
## Phase 4: Lifecycle
|
||||
|
||||
```
|
||||
pause_job id=ID # freeze without losing state
|
||||
resume_job id=ID # pick up where it left off
|
||||
cancel_job id=ID # hard stop
|
||||
replay_job id=ID # re-run with same or modified params
|
||||
replay_job id=ID data_overrides={"depth":"deep"} # replay with changes
|
||||
```
|
||||
|
||||
## Phase 5: Review Results
|
||||
|
||||
```
|
||||
get_job ID # result, token counts, transcript
|
||||
```
|
||||
|
||||
Token accounting: every job tracks `tokens_input`, `tokens_output`, `tokens_cache_read`.
|
||||
Child tokens roll up to parent automatically on completion.
|
||||
|
||||
## Output Format
|
||||
|
||||
When reporting job status to the user:
|
||||
|
||||
```
|
||||
Job #ID (name) — status
|
||||
Progress: step/total — last action
|
||||
Tokens: input_count in / output_count out (+ cache_read cached)
|
||||
Runtime: Xs
|
||||
Children: N pending, M completed
|
||||
```
|
||||
|
||||
When reporting completion:
|
||||
|
||||
```
|
||||
Job #ID completed in Xs
|
||||
Tokens used: input / output / cache_read
|
||||
Result: <summary>
|
||||
```
|
||||
|
||||
When reporting batch status (parent with children):
|
||||
|
||||
```
|
||||
Parent #ID — waiting-children
|
||||
#A research(Acme) — active, 3/5 steps, 2.5k tokens
|
||||
#B research(Beta) — completed, 1.8k tokens
|
||||
#C research(Gamma) — paused
|
||||
Total tokens so far: 4.3k
|
||||
```
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Don't spawn a Minion for a single search query (use search tool directly)
|
||||
- Don't fire-and-forget without checking results
|
||||
- Don't spawn > 5 concurrent agents without checking `get_job_stats` first
|
||||
- Don't use `sessions_spawn` with `runtime: "subagent"` when Minions is available
|
||||
- Don't poll `get_job` in a tight loop (use `get_job_progress` for lightweight checks)
|
||||
|
||||
## Tools Used
|
||||
|
||||
- Submit a background job (submit_job)
|
||||
- Get job details (get_job)
|
||||
- List jobs with filters (list_jobs)
|
||||
- Cancel a job (cancel_job)
|
||||
- Pause a job (pause_job)
|
||||
- Resume a paused job (resume_job)
|
||||
- Replay a completed/failed job (replay_job)
|
||||
- Send sidechannel message (send_job_message)
|
||||
- Get structured progress (get_job_progress)
|
||||
- Get job queue stats (get_job_stats)
|
||||
@@ -128,6 +128,32 @@ const MIGRATIONS: Migration[] = [
|
||||
CREATE INDEX IF NOT EXISTS idx_minion_jobs_parent ON minion_jobs(parent_job_id);
|
||||
`,
|
||||
},
|
||||
{
|
||||
version: 6,
|
||||
name: 'agent_orchestration_primitives',
|
||||
sql: `
|
||||
-- Token accounting columns
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_input INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_output INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE minion_jobs ADD COLUMN IF NOT EXISTS tokens_cache_read INTEGER NOT NULL DEFAULT 0;
|
||||
|
||||
-- Update status constraint to include 'paused'
|
||||
ALTER TABLE minion_jobs DROP CONSTRAINT IF EXISTS chk_status;
|
||||
ALTER TABLE minion_jobs ADD CONSTRAINT chk_status
|
||||
CHECK (status IN ('waiting','active','completed','failed','delayed','dead','cancelled','waiting-children','paused'));
|
||||
|
||||
-- Inbox table (separate from job row for clean concurrency)
|
||||
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;
|
||||
`,
|
||||
},
|
||||
];
|
||||
|
||||
export const LATEST_VERSION = MIGRATIONS.length > 0
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
export { MinionQueue } from './queue.ts';
|
||||
export { MinionWorker } from './worker.ts';
|
||||
export { calculateBackoff } from './backoff.ts';
|
||||
export { UnrecoverableError, rowToMinionJob } from './types.ts';
|
||||
export { UnrecoverableError, rowToMinionJob, rowToInboxMessage } from './types.ts';
|
||||
export type {
|
||||
MinionJob, MinionJobInput, MinionJobStatus, MinionJobContext,
|
||||
MinionHandler, MinionWorkerOpts, BackoffType, ChildFailPolicy,
|
||||
InboxMessage, TokenUpdate, AgentProgress, TranscriptEntry,
|
||||
} from './types.ts';
|
||||
|
||||
+122
-6
@@ -9,10 +9,10 @@
|
||||
*/
|
||||
|
||||
import type { BrainEngine } from '../engine.ts';
|
||||
import type { MinionJob, MinionJobInput, MinionJobStatus } from './types.ts';
|
||||
import { rowToMinionJob } from './types.ts';
|
||||
import type { MinionJob, MinionJobInput, MinionJobStatus, InboxMessage, TokenUpdate } from './types.ts';
|
||||
import { rowToMinionJob, rowToInboxMessage } from './types.ts';
|
||||
|
||||
const MIGRATION_VERSION = 5;
|
||||
const MIGRATION_VERSION = 6;
|
||||
|
||||
export class MinionQueue {
|
||||
constructor(private engine: BrainEngine) {}
|
||||
@@ -120,7 +120,7 @@ export class MinionQueue {
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET status = 'cancelled', lock_token = NULL, lock_until = NULL,
|
||||
finished_at = now(), updated_at = now()
|
||||
WHERE id = $1 AND status IN ('waiting', 'active', 'delayed', 'waiting-children')
|
||||
WHERE id = $1 AND status IN ('waiting', 'active', 'delayed', 'waiting-children', 'paused')
|
||||
RETURNING *`,
|
||||
[id]
|
||||
);
|
||||
@@ -235,7 +235,7 @@ export class MinionQueue {
|
||||
return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
|
||||
}
|
||||
|
||||
/** Complete a job (token-fenced). Returns null if token mismatch. */
|
||||
/** Complete a job (token-fenced). Rolls up token counts to parent if applicable. Returns null if token mismatch. */
|
||||
async completeJob(id: number, lockToken: string, result?: Record<string, unknown>): Promise<MinionJob | null> {
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET status = 'completed', result = $1,
|
||||
@@ -244,7 +244,24 @@ export class MinionQueue {
|
||||
RETURNING *`,
|
||||
[result ? JSON.stringify(result) : null, id, lockToken]
|
||||
);
|
||||
return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
|
||||
if (rows.length === 0) return null;
|
||||
|
||||
const completed = rowToMinionJob(rows[0]);
|
||||
|
||||
// Roll up token counts to parent (if parent exists and is not in a terminal state)
|
||||
if (completed.parent_job_id && (completed.tokens_input > 0 || completed.tokens_output > 0 || completed.tokens_cache_read > 0)) {
|
||||
await this.engine.executeRaw(
|
||||
`UPDATE minion_jobs SET
|
||||
tokens_input = tokens_input + $1,
|
||||
tokens_output = tokens_output + $2,
|
||||
tokens_cache_read = tokens_cache_read + $3,
|
||||
updated_at = now()
|
||||
WHERE id = $4 AND status NOT IN ('completed', 'dead', 'cancelled')`,
|
||||
[completed.tokens_input, completed.tokens_output, completed.tokens_cache_read, completed.parent_job_id]
|
||||
);
|
||||
}
|
||||
|
||||
return completed;
|
||||
}
|
||||
|
||||
/** Fail a job (token-fenced). Sets delayed for retry or dead/failed for terminal. */
|
||||
@@ -367,6 +384,105 @@ export class MinionQueue {
|
||||
return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
|
||||
}
|
||||
|
||||
/** Pause a waiting or active job. For active jobs, clears the lock so the worker's
|
||||
* AbortController fires and the handler stops gracefully. */
|
||||
async pauseJob(id: number): Promise<MinionJob | null> {
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET status = 'paused',
|
||||
lock_token = NULL, lock_until = NULL, updated_at = now()
|
||||
WHERE id = $1 AND status IN ('waiting', 'active', 'delayed')
|
||||
RETURNING *`,
|
||||
[id]
|
||||
);
|
||||
return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
|
||||
}
|
||||
|
||||
/** Resume a paused job back to waiting. */
|
||||
async resumeJob(id: number): Promise<MinionJob | null> {
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET status = 'waiting',
|
||||
lock_token = NULL, lock_until = NULL, updated_at = now()
|
||||
WHERE id = $1 AND status = 'paused'
|
||||
RETURNING *`,
|
||||
[id]
|
||||
);
|
||||
return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
|
||||
}
|
||||
|
||||
/** Send a message to a job's inbox. Sender must be the parent job or 'admin'. */
|
||||
async sendMessage(jobId: number, payload: unknown, sender: string): Promise<InboxMessage | null> {
|
||||
// Validate job exists and is in a messageable state
|
||||
const job = await this.getJob(jobId);
|
||||
if (!job) return null;
|
||||
if (['completed', 'dead', 'cancelled', 'failed'].includes(job.status)) return null;
|
||||
|
||||
// Sender validation: must be parent job ID or 'admin'
|
||||
if (sender !== 'admin' && sender !== String(job.parent_job_id)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`INSERT INTO minion_inbox (job_id, sender, payload)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING *`,
|
||||
[jobId, sender, JSON.stringify(payload)]
|
||||
);
|
||||
return rows.length > 0 ? rowToInboxMessage(rows[0]) : null;
|
||||
}
|
||||
|
||||
/** Read unread inbox messages for a job. Token-fenced. Marks messages as read. */
|
||||
async readInbox(jobId: number, lockToken: string): Promise<InboxMessage[]> {
|
||||
// Verify lock ownership
|
||||
const lockCheck = await this.engine.executeRaw<{ id: number }>(
|
||||
`SELECT id FROM minion_jobs WHERE id = $1 AND lock_token = $2 AND status = 'active'`,
|
||||
[jobId, lockToken]
|
||||
);
|
||||
if (lockCheck.length === 0) return [];
|
||||
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_inbox SET read_at = now()
|
||||
WHERE job_id = $1 AND read_at IS NULL
|
||||
RETURNING *`,
|
||||
[jobId]
|
||||
);
|
||||
return rows.map(rowToInboxMessage);
|
||||
}
|
||||
|
||||
/** Update token counts for a job. Accumulates (adds to existing). Token-fenced. */
|
||||
async updateTokens(id: number, lockToken: string, tokens: TokenUpdate): Promise<boolean> {
|
||||
const rows = await this.engine.executeRaw<Record<string, unknown>>(
|
||||
`UPDATE minion_jobs SET
|
||||
tokens_input = tokens_input + $1,
|
||||
tokens_output = tokens_output + $2,
|
||||
tokens_cache_read = tokens_cache_read + $3,
|
||||
updated_at = now()
|
||||
WHERE id = $4 AND status = 'active' AND lock_token = $5
|
||||
RETURNING id`,
|
||||
[tokens.input ?? 0, tokens.output ?? 0, tokens.cache_read ?? 0, id, lockToken]
|
||||
);
|
||||
return rows.length > 0;
|
||||
}
|
||||
|
||||
/** Replay a completed/failed/dead job with optional data overrides. Creates a new job. */
|
||||
async replayJob(id: number, dataOverrides?: Record<string, unknown>): Promise<MinionJob | null> {
|
||||
const source = await this.getJob(id);
|
||||
if (!source) return null;
|
||||
if (!['completed', 'failed', 'dead'].includes(source.status)) return null;
|
||||
|
||||
const data = dataOverrides
|
||||
? { ...source.data, ...dataOverrides }
|
||||
: source.data;
|
||||
|
||||
return this.add(source.name, data, {
|
||||
queue: source.queue,
|
||||
priority: source.priority,
|
||||
max_attempts: source.max_attempts,
|
||||
backoff_type: source.backoff_type,
|
||||
backoff_delay: source.backoff_delay,
|
||||
backoff_jitter: source.backoff_jitter,
|
||||
});
|
||||
}
|
||||
|
||||
/** Remove a child's dependency on its parent. */
|
||||
async removeChildDependency(childId: number): Promise<void> {
|
||||
await this.engine.executeRaw(
|
||||
|
||||
@@ -23,7 +23,8 @@ export type MinionJobStatus =
|
||||
| 'delayed'
|
||||
| 'dead'
|
||||
| 'cancelled'
|
||||
| 'waiting-children';
|
||||
| 'waiting-children'
|
||||
| 'paused';
|
||||
|
||||
export type BackoffType = 'fixed' | 'exponential';
|
||||
|
||||
@@ -60,6 +61,11 @@ export interface MinionJob {
|
||||
parent_job_id: number | null;
|
||||
on_child_fail: ChildFailPolicy;
|
||||
|
||||
// Token accounting
|
||||
tokens_input: number;
|
||||
tokens_output: number;
|
||||
tokens_cache_read: number;
|
||||
|
||||
// Results
|
||||
result: Record<string, unknown> | null;
|
||||
progress: unknown | null;
|
||||
@@ -105,16 +111,72 @@ export interface MinionJobContext {
|
||||
name: string;
|
||||
data: Record<string, unknown>;
|
||||
attempts_made: number;
|
||||
/** AbortSignal for cooperative cancellation (fires on pause or lock loss). */
|
||||
signal: AbortSignal;
|
||||
/** Update structured progress (not just 0-100). */
|
||||
updateProgress(progress: unknown): Promise<void>;
|
||||
/** Append a log message to the job's stacktrace array. */
|
||||
log(message: string): Promise<void>;
|
||||
/** Accumulate token usage for this job. */
|
||||
updateTokens(tokens: TokenUpdate): Promise<void>;
|
||||
/** Append a log message or transcript entry to the job's stacktrace array. */
|
||||
log(message: string | TranscriptEntry): Promise<void>;
|
||||
/** Check if the lock is still held (for long-running jobs). */
|
||||
isActive(): Promise<boolean>;
|
||||
/** Read unread inbox messages (marks as read). */
|
||||
readInbox(): Promise<InboxMessage[]>;
|
||||
}
|
||||
|
||||
export type MinionHandler = (job: MinionJobContext) => Promise<unknown>;
|
||||
|
||||
// --- Inbox Message ---
|
||||
|
||||
export interface InboxMessage {
|
||||
id: number;
|
||||
job_id: number;
|
||||
sender: string;
|
||||
payload: unknown;
|
||||
sent_at: Date;
|
||||
read_at: Date | null;
|
||||
}
|
||||
|
||||
export function rowToInboxMessage(row: Record<string, unknown>): InboxMessage {
|
||||
return {
|
||||
id: row.id as number,
|
||||
job_id: row.job_id as number,
|
||||
sender: row.sender as string,
|
||||
payload: typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload,
|
||||
sent_at: new Date(row.sent_at as string),
|
||||
read_at: row.read_at ? new Date(row.read_at as string) : null,
|
||||
};
|
||||
}
|
||||
|
||||
// --- Token Update ---
|
||||
|
||||
export interface TokenUpdate {
|
||||
input?: number;
|
||||
output?: number;
|
||||
cache_read?: number;
|
||||
}
|
||||
|
||||
// --- Structured Progress (convention, not enforced) ---
|
||||
|
||||
export interface AgentProgress {
|
||||
step: number;
|
||||
total: number;
|
||||
message: string;
|
||||
tokens_in: number;
|
||||
tokens_out: number;
|
||||
last_tool: string;
|
||||
started_at: string;
|
||||
}
|
||||
|
||||
// --- Transcript Entry ---
|
||||
|
||||
export type TranscriptEntry =
|
||||
| { type: 'log'; message: string; ts: string }
|
||||
| { type: 'tool_call'; tool: string; args_size: number; result_size: number; ts: string }
|
||||
| { type: 'llm_turn'; model: string; tokens_in: number; tokens_out: number; ts: string }
|
||||
| { type: 'error'; message: string; stack?: string; ts: string };
|
||||
|
||||
// --- Errors ---
|
||||
|
||||
/** Throw this from a handler to skip all retry logic and go straight to 'dead'. */
|
||||
@@ -148,6 +210,9 @@ export function rowToMinionJob(row: Record<string, unknown>): MinionJob {
|
||||
delay_until: row.delay_until ? new Date(row.delay_until as string) : null,
|
||||
parent_job_id: (row.parent_job_id as number) || null,
|
||||
on_child_fail: row.on_child_fail as ChildFailPolicy,
|
||||
tokens_input: (row.tokens_input as number) ?? 0,
|
||||
tokens_output: (row.tokens_output as number) ?? 0,
|
||||
tokens_cache_read: (row.tokens_cache_read as number) ?? 0,
|
||||
result: row.result ? (typeof row.result === 'string' ? JSON.parse(row.result) : row.result) as Record<string, unknown> : null,
|
||||
progress: row.progress ? (typeof row.progress === 'string' ? JSON.parse(row.progress) : row.progress) : null,
|
||||
error_text: (row.error_text as string) || null,
|
||||
|
||||
+90
-48
@@ -1,5 +1,8 @@
|
||||
/**
|
||||
* MinionWorker — In-process job worker with BullMQ-inspired patterns.
|
||||
* MinionWorker — Concurrent in-process job worker with BullMQ-inspired patterns.
|
||||
*
|
||||
* Processes up to `concurrency` jobs simultaneously using a Promise pool.
|
||||
* Each job gets its own AbortController, lock renewal timer, and isolated state.
|
||||
*
|
||||
* Usage:
|
||||
* const worker = new MinionWorker(engine);
|
||||
@@ -9,18 +12,26 @@
|
||||
*/
|
||||
|
||||
import type { BrainEngine } from '../engine.ts';
|
||||
import type { MinionJob, MinionJobContext, MinionHandler, MinionWorkerOpts } from './types.ts';
|
||||
import type { MinionJob, MinionJobContext, MinionHandler, MinionWorkerOpts, TokenUpdate } from './types.ts';
|
||||
import { UnrecoverableError } from './types.ts';
|
||||
import { MinionQueue } from './queue.ts';
|
||||
import { calculateBackoff } from './backoff.ts';
|
||||
import { randomUUID } from 'crypto';
|
||||
|
||||
/** Per-job in-flight state (isolated per job, not shared on the worker). */
|
||||
interface InFlightJob {
|
||||
job: MinionJob;
|
||||
lockToken: string;
|
||||
lockTimer: ReturnType<typeof setInterval>;
|
||||
abort: AbortController;
|
||||
promise: Promise<void>;
|
||||
}
|
||||
|
||||
export class MinionWorker {
|
||||
private queue: MinionQueue;
|
||||
private handlers = new Map<string, MinionHandler>();
|
||||
private running = false;
|
||||
private currentJob: MinionJob | null = null;
|
||||
private lockRenewalTimer: ReturnType<typeof setInterval> | null = null;
|
||||
private inFlight = new Map<number, InFlightJob>();
|
||||
private workerId = randomUUID();
|
||||
|
||||
private opts: Required<MinionWorkerOpts>;
|
||||
@@ -87,22 +98,28 @@ export class MinionWorker {
|
||||
console.error('Promotion error:', e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
|
||||
// Claim and execute
|
||||
const lockToken = `${this.workerId}:${Date.now()}`;
|
||||
const job = await this.queue.claim(
|
||||
lockToken,
|
||||
this.opts.lockDuration,
|
||||
this.opts.queue,
|
||||
this.registeredNames,
|
||||
);
|
||||
// Claim jobs up to concurrency limit
|
||||
if (this.inFlight.size < this.opts.concurrency) {
|
||||
const lockToken = `${this.workerId}:${Date.now()}`;
|
||||
const job = await this.queue.claim(
|
||||
lockToken,
|
||||
this.opts.lockDuration,
|
||||
this.opts.queue,
|
||||
this.registeredNames,
|
||||
);
|
||||
|
||||
if (job) {
|
||||
this.currentJob = job;
|
||||
await this.executeJob(job, lockToken);
|
||||
this.currentJob = null;
|
||||
if (job) {
|
||||
this.launchJob(job, lockToken);
|
||||
} else if (this.inFlight.size === 0) {
|
||||
// No jobs and nothing in flight, poll
|
||||
await new Promise(resolve => setTimeout(resolve, this.opts.pollInterval));
|
||||
} else {
|
||||
// Jobs are running but no new ones available, brief pause before re-checking
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
}
|
||||
} else {
|
||||
// No jobs available, poll
|
||||
await new Promise(resolve => setTimeout(resolve, this.opts.pollInterval));
|
||||
// At concurrency limit, wait briefly before re-checking for free slots
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
@@ -110,10 +127,14 @@ export class MinionWorker {
|
||||
process.removeListener('SIGTERM', shutdown);
|
||||
process.removeListener('SIGINT', shutdown);
|
||||
|
||||
// Wait for current job to finish (graceful shutdown)
|
||||
if (this.currentJob && this.lockRenewalTimer) {
|
||||
console.log('Waiting for current job to finish (30s timeout)...');
|
||||
await new Promise(resolve => setTimeout(resolve, 30000));
|
||||
// Graceful shutdown: wait for all in-flight jobs with timeout
|
||||
if (this.inFlight.size > 0) {
|
||||
console.log(`Waiting for ${this.inFlight.size} in-flight job(s) to finish (30s timeout)...`);
|
||||
const pending = Array.from(this.inFlight.values()).map(f => f.promise);
|
||||
await Promise.race([
|
||||
Promise.allSettled(pending),
|
||||
new Promise(resolve => setTimeout(resolve, 30000)),
|
||||
]);
|
||||
}
|
||||
|
||||
console.log('Minion worker stopped.');
|
||||
@@ -125,41 +146,61 @@ export class MinionWorker {
|
||||
this.running = false;
|
||||
}
|
||||
|
||||
private async executeJob(job: MinionJob, lockToken: string): Promise<void> {
|
||||
/** Launch a job as an independent in-flight promise. */
|
||||
private launchJob(job: MinionJob, lockToken: string): void {
|
||||
const abort = new AbortController();
|
||||
|
||||
// Start lock renewal (per-job timer, not shared)
|
||||
const lockTimer = setInterval(async () => {
|
||||
const renewed = await this.queue.renewLock(job.id, lockToken, this.opts.lockDuration);
|
||||
if (!renewed) {
|
||||
console.warn(`Lock lost for job ${job.id}, aborting execution`);
|
||||
clearInterval(lockTimer);
|
||||
abort.abort();
|
||||
}
|
||||
}, this.opts.lockDuration / 2);
|
||||
|
||||
const promise = this.executeJob(job, lockToken, abort, lockTimer)
|
||||
.finally(() => {
|
||||
clearInterval(lockTimer);
|
||||
this.inFlight.delete(job.id);
|
||||
});
|
||||
|
||||
this.inFlight.set(job.id, { job, lockToken, lockTimer, abort, promise });
|
||||
}
|
||||
|
||||
private async executeJob(
|
||||
job: MinionJob,
|
||||
lockToken: string,
|
||||
abort: AbortController,
|
||||
lockTimer: ReturnType<typeof setInterval>,
|
||||
): Promise<void> {
|
||||
const handler = this.handlers.get(job.name);
|
||||
if (!handler) {
|
||||
// This shouldn't happen (claim filters by registered names), but be safe
|
||||
await this.queue.failJob(job.id, lockToken, `No handler for job type '${job.name}'`, 'dead');
|
||||
return;
|
||||
}
|
||||
|
||||
// Start lock renewal
|
||||
this.lockRenewalTimer = setInterval(async () => {
|
||||
const renewed = await this.queue.renewLock(job.id, lockToken, this.opts.lockDuration);
|
||||
if (!renewed) {
|
||||
// Lock was stolen (stall detector reclaimed it)
|
||||
console.warn(`Lock lost for job ${job.id}, stopping execution`);
|
||||
this.lockRenewalTimer && clearInterval(this.lockRenewalTimer);
|
||||
this.lockRenewalTimer = null;
|
||||
}
|
||||
}, this.opts.lockDuration / 2);
|
||||
|
||||
// Build job context
|
||||
// Build job context with per-job AbortSignal
|
||||
const context: MinionJobContext = {
|
||||
id: job.id,
|
||||
name: job.name,
|
||||
data: job.data,
|
||||
attempts_made: job.attempts_made,
|
||||
signal: abort.signal,
|
||||
updateProgress: async (progress: unknown) => {
|
||||
await this.queue.updateProgress(job.id, lockToken, progress);
|
||||
},
|
||||
log: async (message: string) => {
|
||||
// Append to stacktrace as a log entry
|
||||
updateTokens: async (tokens: TokenUpdate) => {
|
||||
await this.queue.updateTokens(job.id, lockToken, tokens);
|
||||
},
|
||||
log: async (message: string | Record<string, unknown>) => {
|
||||
const value = typeof message === 'string' ? message : JSON.stringify(message);
|
||||
await this.engine.executeRaw(
|
||||
`UPDATE minion_jobs SET stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($1::text),
|
||||
updated_at = now()
|
||||
WHERE id = $2 AND status = 'active' AND lock_token = $3`,
|
||||
[message, job.id, lockToken]
|
||||
[value, job.id, lockToken]
|
||||
);
|
||||
},
|
||||
isActive: async () => {
|
||||
@@ -169,16 +210,15 @@ export class MinionWorker {
|
||||
);
|
||||
return rows.length > 0;
|
||||
},
|
||||
readInbox: async () => {
|
||||
return this.queue.readInbox(job.id, lockToken);
|
||||
},
|
||||
};
|
||||
|
||||
try {
|
||||
const result = await handler(context);
|
||||
|
||||
// Clear renewal timer
|
||||
if (this.lockRenewalTimer) {
|
||||
clearInterval(this.lockRenewalTimer);
|
||||
this.lockRenewalTimer = null;
|
||||
}
|
||||
clearInterval(lockTimer);
|
||||
|
||||
// Complete the job (token-fenced)
|
||||
const completed = await this.queue.completeJob(
|
||||
@@ -197,10 +237,12 @@ export class MinionWorker {
|
||||
await this.queue.resolveParent(job.parent_job_id);
|
||||
}
|
||||
} catch (err) {
|
||||
// Clear renewal timer
|
||||
if (this.lockRenewalTimer) {
|
||||
clearInterval(this.lockRenewalTimer);
|
||||
this.lockRenewalTimer = null;
|
||||
clearInterval(lockTimer);
|
||||
|
||||
// If aborted (paused or lock lost), don't try to fail the job
|
||||
if (abort.signal.aborted) {
|
||||
console.log(`Job ${job.id} (${job.name}) aborted (paused or lock lost)`);
|
||||
return;
|
||||
}
|
||||
|
||||
const errorText = err instanceof Error ? err.message : String(err);
|
||||
|
||||
@@ -762,6 +762,71 @@ const get_job_progress: Operation = {
|
||||
},
|
||||
};
|
||||
|
||||
const pause_job: Operation = {
|
||||
name: 'pause_job',
|
||||
description: 'Pause a waiting, active, or delayed job',
|
||||
params: {
|
||||
id: { type: 'number', required: true, description: 'Job ID' },
|
||||
},
|
||||
handler: async (ctx, p) => {
|
||||
const { MinionQueue } = await import('./minions/queue.ts');
|
||||
const queue = new MinionQueue(ctx.engine);
|
||||
const job = await queue.pauseJob(p.id as number);
|
||||
if (!job) throw new OperationError('invalid_params', `Job not found or not pausable: ${p.id}`);
|
||||
return { id: job.id, status: job.status };
|
||||
},
|
||||
};
|
||||
|
||||
const resume_job: Operation = {
|
||||
name: 'resume_job',
|
||||
description: 'Resume a paused job back to waiting',
|
||||
params: {
|
||||
id: { type: 'number', required: true, description: 'Job ID' },
|
||||
},
|
||||
handler: async (ctx, p) => {
|
||||
const { MinionQueue } = await import('./minions/queue.ts');
|
||||
const queue = new MinionQueue(ctx.engine);
|
||||
const job = await queue.resumeJob(p.id as number);
|
||||
if (!job) throw new OperationError('invalid_params', `Job not found or not paused: ${p.id}`);
|
||||
return { id: job.id, status: job.status };
|
||||
},
|
||||
};
|
||||
|
||||
const replay_job: Operation = {
|
||||
name: 'replay_job',
|
||||
description: 'Replay a completed/failed/dead job, optionally with modified data',
|
||||
params: {
|
||||
id: { type: 'number', required: true, description: 'Source job ID to replay' },
|
||||
data_overrides: { type: 'object', required: false, description: 'Data fields to override (merged with original)' },
|
||||
},
|
||||
handler: async (ctx, p) => {
|
||||
if (ctx.dryRun) return { dry_run: true, action: 'replay_job', id: p.id };
|
||||
const { MinionQueue } = await import('./minions/queue.ts');
|
||||
const queue = new MinionQueue(ctx.engine);
|
||||
const job = await queue.replayJob(p.id as number, p.data_overrides as Record<string, unknown> | undefined);
|
||||
if (!job) throw new OperationError('invalid_params', `Job not found or not in terminal state: ${p.id}`);
|
||||
return { id: job.id, name: job.name, status: job.status, source_id: p.id };
|
||||
},
|
||||
};
|
||||
|
||||
const send_job_message: Operation = {
|
||||
name: 'send_job_message',
|
||||
description: 'Send a sidechannel message to a running job\'s inbox',
|
||||
params: {
|
||||
id: { type: 'number', required: true, description: 'Job ID to message' },
|
||||
payload: { type: 'object', required: true, description: 'Message payload (arbitrary JSON)' },
|
||||
sender: { type: 'string', required: false, description: 'Sender identity (default: admin)' },
|
||||
},
|
||||
handler: async (ctx, p) => {
|
||||
if (ctx.dryRun) return { dry_run: true, action: 'send_job_message', id: p.id };
|
||||
const { MinionQueue } = await import('./minions/queue.ts');
|
||||
const queue = new MinionQueue(ctx.engine);
|
||||
const msg = await queue.sendMessage(p.id as number, p.payload, (p.sender as string) ?? 'admin');
|
||||
if (!msg) throw new OperationError('invalid_params', `Job not found, not messageable, or sender unauthorized: ${p.id}`);
|
||||
return { sent: true, message_id: msg.id, job_id: p.id };
|
||||
},
|
||||
};
|
||||
|
||||
// --- Exports ---
|
||||
|
||||
export const operations: Operation[] = [
|
||||
@@ -789,6 +854,7 @@ export const operations: Operation[] = [
|
||||
file_list, file_upload, file_url,
|
||||
// Jobs (Minions)
|
||||
submit_job, get_job, list_jobs, cancel_job, retry_job, get_job_progress,
|
||||
pause_job, resume_job, replay_job, send_job_message,
|
||||
];
|
||||
|
||||
export const operationsByName = Object.fromEntries(
|
||||
|
||||
@@ -182,6 +182,9 @@ 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,
|
||||
@@ -190,7 +193,7 @@ CREATE TABLE IF NOT EXISTS minion_jobs (
|
||||
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),
|
||||
@@ -204,6 +207,17 @@ CREATE INDEX IF NOT EXISTS idx_minion_jobs_stalled ON minion_jobs (lock_until) W
|
||||
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);
|
||||
|
||||
-- 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;
|
||||
|
||||
-- ============================================================
|
||||
-- Trigger-based search_vector (spans pages + timeline_entries)
|
||||
-- ============================================================
|
||||
|
||||
+29
-1
@@ -269,6 +269,9 @@ 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,
|
||||
@@ -277,7 +280,7 @@ CREATE TABLE IF NOT EXISTS minion_jobs (
|
||||
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),
|
||||
@@ -291,6 +294,31 @@ CREATE INDEX IF NOT EXISTS idx_minion_jobs_stalled ON minion_jobs (lock_until) W
|
||||
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);
|
||||
|
||||
-- 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;
|
||||
|
||||
-- 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;
|
||||
|
||||
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
|
||||
-- ============================================================
|
||||
|
||||
@@ -550,3 +550,262 @@ describe('MinionQueue: Cancel & Retry', () => {
|
||||
expect(retried!.error_text).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// --- Pause / Resume (5 tests) ---
|
||||
|
||||
describe('MinionQueue: Pause/Resume', () => {
|
||||
test('pause waiting job → paused', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
const paused = await queue.pauseJob(job.id);
|
||||
expect(paused!.status).toBe('paused');
|
||||
});
|
||||
|
||||
test('pause active job clears lock', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
const paused = await queue.pauseJob(job.id);
|
||||
expect(paused!.status).toBe('paused');
|
||||
expect(paused!.lock_token).toBeNull();
|
||||
expect(paused!.lock_until).toBeNull();
|
||||
});
|
||||
|
||||
test('pause completed job returns null', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await queue.completeJob(job.id, 'tok1');
|
||||
const paused = await queue.pauseJob(job.id);
|
||||
expect(paused).toBeNull();
|
||||
});
|
||||
|
||||
test('resume paused job → waiting', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.pauseJob(job.id);
|
||||
const resumed = await queue.resumeJob(job.id);
|
||||
expect(resumed!.status).toBe('waiting');
|
||||
});
|
||||
|
||||
test('resume non-paused job returns null', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
const resumed = await queue.resumeJob(job.id);
|
||||
expect(resumed).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// --- Inbox (6 tests) ---
|
||||
|
||||
describe('MinionQueue: Inbox', () => {
|
||||
beforeEach(async () => {
|
||||
await engine.executeRaw('DELETE FROM minion_inbox');
|
||||
});
|
||||
|
||||
test('send message to active job from admin', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
const msg = await queue.sendMessage(job.id, { directive: 'focus on X' }, 'admin');
|
||||
expect(msg).not.toBeNull();
|
||||
expect(msg!.sender).toBe('admin');
|
||||
expect(msg!.payload).toEqual({ directive: 'focus on X' });
|
||||
expect(msg!.read_at).toBeNull();
|
||||
});
|
||||
|
||||
test('send message from parent job succeeds', async () => {
|
||||
const parent = await queue.add('orchestrate', {});
|
||||
// Create child directly with waiting status so it's claimable
|
||||
const childRows = await engine.executeRaw<Record<string, unknown>>(
|
||||
`INSERT INTO minion_jobs (name, queue, status, data, parent_job_id)
|
||||
VALUES ('research', 'default', 'waiting', '{}', $1) RETURNING *`,
|
||||
[parent.id]
|
||||
);
|
||||
const childId = childRows[0].id as number;
|
||||
await queue.claim('tok1', 30000, 'default', ['research']);
|
||||
const msg = await queue.sendMessage(childId, { hint: 'dig deeper' }, String(parent.id));
|
||||
expect(msg).not.toBeNull();
|
||||
});
|
||||
|
||||
test('send message from unauthorized sender returns null', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
const msg = await queue.sendMessage(job.id, { hack: true }, 'rogue-agent');
|
||||
expect(msg).toBeNull();
|
||||
});
|
||||
|
||||
test('send message to completed job returns null', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await queue.completeJob(job.id, 'tok1');
|
||||
const msg = await queue.sendMessage(job.id, { too: 'late' }, 'admin');
|
||||
expect(msg).toBeNull();
|
||||
});
|
||||
|
||||
test('readInbox returns unread messages and marks read', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await queue.sendMessage(job.id, { msg: 1 }, 'admin');
|
||||
await queue.sendMessage(job.id, { msg: 2 }, 'admin');
|
||||
|
||||
const messages = await queue.readInbox(job.id, 'tok1');
|
||||
expect(messages).toHaveLength(2);
|
||||
expect(messages[0].payload).toEqual({ msg: 1 });
|
||||
expect(messages[0].read_at).not.toBeNull();
|
||||
|
||||
// Second read returns empty (all marked read)
|
||||
const empty = await queue.readInbox(job.id, 'tok1');
|
||||
expect(empty).toHaveLength(0);
|
||||
});
|
||||
|
||||
test('readInbox with wrong token returns empty', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['sync']);
|
||||
await queue.sendMessage(job.id, { msg: 1 }, 'admin');
|
||||
|
||||
const messages = await queue.readInbox(job.id, 'wrong-token');
|
||||
expect(messages).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
||||
// --- Token Accounting (4 tests) ---
|
||||
|
||||
describe('MinionQueue: Token Accounting', () => {
|
||||
test('updateTokens accumulates counts', async () => {
|
||||
const job = await queue.add('agent', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['agent']);
|
||||
|
||||
await queue.updateTokens(job.id, 'tok1', { input: 100, output: 50 });
|
||||
await queue.updateTokens(job.id, 'tok1', { input: 200, output: 100, cache_read: 50 });
|
||||
|
||||
const updated = await queue.getJob(job.id);
|
||||
expect(updated!.tokens_input).toBe(300);
|
||||
expect(updated!.tokens_output).toBe(150);
|
||||
expect(updated!.tokens_cache_read).toBe(50);
|
||||
});
|
||||
|
||||
test('updateTokens with wrong token returns false', async () => {
|
||||
const job = await queue.add('agent', {});
|
||||
await queue.claim('tok1', 30000, 'default', ['agent']);
|
||||
const result = await queue.updateTokens(job.id, 'wrong', { input: 100 });
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
test('completeJob rolls up tokens to parent', async () => {
|
||||
const parent = await queue.add('orchestrate', {});
|
||||
// Create child with parent_job_id but manually set to 'waiting' so it's claimable
|
||||
const childRows = await engine.executeRaw<Record<string, unknown>>(
|
||||
`INSERT INTO minion_jobs (name, queue, status, data, parent_job_id)
|
||||
VALUES ('research', 'default', 'waiting', '{}', $1) RETURNING *`,
|
||||
[parent.id]
|
||||
);
|
||||
const childId = childRows[0].id as number;
|
||||
await queue.claim('tok1', 30000, 'default', ['research']);
|
||||
await queue.updateTokens(childId, 'tok1', { input: 500, output: 200 });
|
||||
await queue.completeJob(childId, 'tok1', { done: true });
|
||||
|
||||
const parentJob = await queue.getJob(parent.id);
|
||||
expect(parentJob!.tokens_input).toBe(500);
|
||||
expect(parentJob!.tokens_output).toBe(200);
|
||||
});
|
||||
|
||||
test('new jobs start with zero tokens', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
expect(job.tokens_input).toBe(0);
|
||||
expect(job.tokens_output).toBe(0);
|
||||
expect(job.tokens_cache_read).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
// --- Job Replay (4 tests) ---
|
||||
|
||||
describe('MinionQueue: Replay', () => {
|
||||
test('replay completed job creates new job', async () => {
|
||||
const job = await queue.add('research', { topic: 'AI' }, { priority: 5 });
|
||||
await queue.claim('tok1', 30000, 'default', ['research']);
|
||||
await queue.completeJob(job.id, 'tok1', { result: 'done' });
|
||||
|
||||
const replay = await queue.replayJob(job.id);
|
||||
expect(replay).not.toBeNull();
|
||||
expect(replay!.id).not.toBe(job.id);
|
||||
expect(replay!.name).toBe('research');
|
||||
expect(replay!.data).toEqual({ topic: 'AI' });
|
||||
expect(replay!.status).toBe('waiting');
|
||||
expect(replay!.priority).toBe(5);
|
||||
expect(replay!.attempts_made).toBe(0);
|
||||
});
|
||||
|
||||
test('replay with data override merges data', async () => {
|
||||
const job = await queue.add('research', { topic: 'AI', depth: 'shallow' });
|
||||
await queue.claim('tok1', 30000, 'default', ['research']);
|
||||
await queue.completeJob(job.id, 'tok1');
|
||||
|
||||
const replay = await queue.replayJob(job.id, { depth: 'deep', focus: 'revenue' });
|
||||
expect(replay!.data).toEqual({ topic: 'AI', depth: 'deep', focus: 'revenue' });
|
||||
});
|
||||
|
||||
test('replay non-terminal job returns null', async () => {
|
||||
const job = await queue.add('sync', {});
|
||||
const replay = await queue.replayJob(job.id);
|
||||
expect(replay).toBeNull();
|
||||
});
|
||||
|
||||
test('replay nonexistent job returns null', async () => {
|
||||
const replay = await queue.replayJob(99999);
|
||||
expect(replay).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// --- Concurrent Worker (3 tests) ---
|
||||
|
||||
describe('MinionWorker: Concurrent', () => {
|
||||
test('worker provides AbortSignal in context', async () => {
|
||||
let receivedSignal: AbortSignal | null = null;
|
||||
const job = await queue.add('test-signal', {});
|
||||
|
||||
const worker = new MinionWorker(engine, { concurrency: 1, pollInterval: 100 });
|
||||
worker.register('test-signal', async (ctx) => {
|
||||
receivedSignal = ctx.signal;
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
const p = worker.start();
|
||||
await new Promise(r => setTimeout(r, 500));
|
||||
worker.stop();
|
||||
await p;
|
||||
|
||||
expect(receivedSignal).not.toBeNull();
|
||||
expect(receivedSignal!.aborted).toBe(false);
|
||||
});
|
||||
|
||||
test('worker provides readInbox in context', async () => {
|
||||
let hasReadInbox = false;
|
||||
const job = await queue.add('test-inbox', {});
|
||||
|
||||
const worker = new MinionWorker(engine, { concurrency: 1, pollInterval: 100 });
|
||||
worker.register('test-inbox', async (ctx) => {
|
||||
hasReadInbox = typeof ctx.readInbox === 'function';
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
const p = worker.start();
|
||||
await new Promise(r => setTimeout(r, 500));
|
||||
worker.stop();
|
||||
await p;
|
||||
|
||||
expect(hasReadInbox).toBe(true);
|
||||
});
|
||||
|
||||
test('worker provides updateTokens in context', async () => {
|
||||
let hasUpdateTokens = false;
|
||||
const job = await queue.add('test-tokens', {});
|
||||
|
||||
const worker = new MinionWorker(engine, { concurrency: 1, pollInterval: 100 });
|
||||
worker.register('test-tokens', async (ctx) => {
|
||||
hasUpdateTokens = typeof ctx.updateTokens === 'function';
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
const p = worker.start();
|
||||
await new Promise(r => setTimeout(r, 500));
|
||||
worker.stop();
|
||||
await p;
|
||||
|
||||
expect(hasUpdateTokens).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user