mirror of
https://github.com/Heretek-AI/agency-control-plane.git
synced 2026-11-07 10:46:08 -10:00
feat(dispatcher): land approved tasks, release worktrees, role packs, tier failover, exact pins
Closes the readiness gaps tracked in #2: - G-1: APPROVE now calls landTask(): fleet/<task> is merged --no-ff into the mission integration branch in a throwaway worktree, serialized across approvals. Success records landedSha and moves the task to done; a conflict (or any landing failure) keeps it waitingHuman, answers 409 landing-failed, and lists the conflicting files. - G-2: done/failed/cancelled tasks release their worktree once no worker runs in it (also swept on startup); failed/cancelled branches are then anchored at refs/fleet-salvage/<task> via salvageWorktree(). - G-3: --roles-dir role packs (roles.ts). An expert task runs with XDG_CONFIG_HOME=<roles-dir>/<role>, the role opencode.json copied into the worktree .opencode/, the role prompt and SKILL.md paths in its task prompt, and role.json maxConcurrent as its per-expert cap. - G-4: --tier-models maps each tier to a provider/model passed as the v2 session.create model. Quota-shaped worker failures feed the tier failover and re-queue the task without spending an attempt; quota on the last tier pauses new launches until the retry time. - G-5: exact dependency pins in dispatcher/ and web/ (the versions the lockfile already resolved). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01Pk7RD6oqBpDdpbWAYqj539
This commit is contained in:
14 files changed
+904
-73
No files matched your search
@@ -42,6 +42,9 @@ Dispatcher flags (also documented in the header of `dispatcher/src/main.ts`):
|
||||
| `--director-agent` | `director` | OpenCode agent of the Director session (define it in your OpenCode config) |
|
||||
| `--task-agent` | server default | OpenCode agent for task sessions |
|
||||
| `--fleet-cap` | `4` | Max concurrent workers (scheduler cap) |
|
||||
| `--per-expert-cap` | fleet cap | Max concurrent workers per `expert` when its role pack sets no `maxConcurrent` |
|
||||
| `--roles-dir` / `FLEET_ROLES_DIR` | unset | Role packs for `expert` tasks (see **Role packs** below) |
|
||||
| `--tier-models` / `FLEET_TIER_MODELS` | unset | JSON `{"tier-1-opencode": "provider/model", "tier-2-commercial": …, "tier-3-local": …}`. New worker sessions run on the active tier's model; unset tiers use the worker's default model |
|
||||
| `--task-timeout-ms` | 30 min | Per-task worker timeout |
|
||||
| `--no-workers`, `--no-director` | | Disable the launcher / Director |
|
||||
| `NATS_URL` | unset | Mirror events + board into NATS JetStream / KV (start fails if set but unreachable) |
|
||||
@@ -79,8 +82,12 @@ Dispatcher flags (also documented in the header of `dispatcher/src/main.ts`):
|
||||
exiting / timing out) ends the run. Successful runs are committed onto
|
||||
the task branch (never the `.opencode` state); a non-empty diff moves
|
||||
the card to **Waiting Approval / QA**, an empty one to **Done**.
|
||||
Failures follow the task's retry policy. Shutdown (SIGINT/SIGTERM)
|
||||
kills every worker and re-queues interrupted tasks.
|
||||
Failures follow the task's retry policy, except quota-shaped provider
|
||||
errors (`limits.ts`): those move the fleet to the next model tier and
|
||||
re-queue the task without spending an attempt; quota on
|
||||
`tier-3-local` holds new launches until the provider's retry time.
|
||||
Shutdown (SIGINT/SIGTERM) kills every worker and re-queues
|
||||
interrupted tasks.
|
||||
3. **Approval.** The human opens the card, checks the diff, and mints a
|
||||
ticket. The **server** computes `subjectHash = sha256(concatenated
|
||||
per-file patches of base...fleet/<task>)` and returns the exact
|
||||
@@ -96,17 +103,48 @@ Dispatcher flags (also documented in the header of `dispatcher/src/main.ts`):
|
||||
sends only `{ ticket, approve, subjectHash, signatureHex }`. The
|
||||
dispatcher burns the single-use 60 s ticket, recomputes the diff hash,
|
||||
and verifies against the pinned `human.pub` only (any client key or
|
||||
passphrase is ignored). APPROVE → `done`, REJECT → `failed`; the full
|
||||
es-core-shaped signed record goes into the hash-chained audit ledger.
|
||||
4. **Persistence.** Tasks, missions, chat, counters and per-task tool
|
||||
passphrase is ignored). APPROVE lands the task: `fleet/<task>` is
|
||||
merged `--no-ff` into `fleet/integration/<mission>` in a throwaway
|
||||
worktree (`landing.ts`), the task moves to `done` with its
|
||||
`landedSha`. A merge conflict leaves the integration branch untouched
|
||||
and the task in **Waiting Approval / QA** (409, conflicting files in
|
||||
the response and the task reason); resolve on the task branch and
|
||||
approve again. REJECT → `failed`. The full es-core-shaped signed
|
||||
record goes into the hash-chained audit ledger.
|
||||
4. **Worktree release.** When a task reaches `done`, `failed`, or
|
||||
`cancelled` and no worker runs in it, its worktree is removed (the
|
||||
branch stays). Branches of `failed` / `cancelled` tasks are then
|
||||
anchored at `refs/fleet-salvage/<task>` and deleted, so no work is
|
||||
lost. A restart releases worktrees of tasks that finished while the
|
||||
dispatcher was down.
|
||||
5. **Persistence.** Tasks, missions, chat, counters and per-task tool
|
||||
logs live in `<state-dir>/dispatcher.db`; a restart resumes the board
|
||||
(tasks that were running are re-queued).
|
||||
5. **NATS (optional).** With `NATS_URL`, every bus event is published to
|
||||
6. **NATS (optional).** With `NATS_URL`, every bus event is published to
|
||||
JetStream subject `fleet.events.<type>` on stream `FLEET_EVENTS`
|
||||
(drop-oldest at 10 000 messages) and every task document is mirrored to
|
||||
KV bucket `fleet-board` (`nats@2.29.3`). The in-process ring buffer
|
||||
remains the SSE source of truth.
|
||||
|
||||
## Role packs (`--roles-dir`)
|
||||
|
||||
A task's `expert` names a role pack at `<roles-dir>/<role-id>/`, laid out
|
||||
as an OpenCode config home:
|
||||
|
||||
```text
|
||||
<roles-dir>/<role-id>/
|
||||
role.json optional: {"maxConcurrent": 2}
|
||||
opencode/opencode.json copied (not symlinked) into the worktree .opencode/
|
||||
opencode/agent/<role-id>.md role prompt, frontmatter stripped, put in the task prompt
|
||||
opencode/skill/<name>/SKILL.md listed in the task prompt
|
||||
```
|
||||
|
||||
The worker runs with `XDG_CONFIG_HOME=<roles-dir>/<role-id>`, so OpenCode
|
||||
also picks the role's agents and skills up as its global config.
|
||||
`agents/` and `skills/` work as plural aliases. Only slug role ids
|
||||
resolve; an `expert` without a pack runs with the generic prompt and a
|
||||
`role.missing` event.
|
||||
|
||||
## Control API (`dispatcher/src/server.ts`, Bun HTTP + SSE, `127.0.0.1` only)
|
||||
|
||||
| Method & Path | Purpose |
|
||||
@@ -115,7 +153,7 @@ Dispatcher flags (also documented in the header of `dispatcher/src/main.ts`):
|
||||
| `POST /api/director/message` | `/ship <spec>` → cards now (200); otherwise → Director turn (202; reply + cards arrive as `director.message`). 503 without a Director. |
|
||||
| `GET /api/tasks/:id/drilldown` | Session, stage gates, tool log, diff hunks, `subjectHash`. |
|
||||
| `POST /api/approvals/:taskId/ticket` | Single-use 60 s ticket; server-computed `subjectHash`, payloads to sign, pinned key fingerprint. 503 without a pinned key, 409 unless `waitingHuman`. |
|
||||
| `POST /api/approvals/:taskId/sign` | Burn ticket, re-hash diff, verify Ed25519 against the pinned key; `waitingHuman` → `done` / `failed`. |
|
||||
| `POST /api/approvals/:taskId/sign` | Burn ticket, re-hash diff, verify Ed25519 against the pinned key; APPROVE lands the task (`done` + `landedSha`, or 409 `landing-failed` with conflicting files and the task kept `waitingHuman`), REJECT → `failed`. |
|
||||
| `GET /api/events` | SSE (ring-buffered, `Last-Event-ID` replay, gap markers). |
|
||||
| `GET /api/csrf` | CSRF token for POSTs. |
|
||||
| `GET /*` | The built WebUI (`web/dist`) with a strict same-origin CSP. |
|
||||
@@ -127,7 +165,7 @@ Dispatcher flags (also documented in the header of `dispatcher/src/main.ts`):
|
||||
`approval-tickets.ts`, `human-key.ts` (es-core-compatible pinned key),
|
||||
`board-store.ts` (SQLite board), `opencode-client.ts` (worker process +
|
||||
OpenCode v2 HTTP client behind `WorkerApi`), `worker-session.ts`,
|
||||
`worker-launcher.ts`, `director.ts`, `nats-bridge.ts`, `server.ts`,
|
||||
`worker-launcher.ts`, `roles.ts` (role packs), `director.ts`, `nats-bridge.ts`, `server.ts`,
|
||||
`main.ts` (daemon), `keygen.ts`.
|
||||
- `dispatcher/test/fixtures/fake-opencode.ts` — fake `opencode serve`
|
||||
implementing the routes above, used by the launcher/Director tests and
|
||||
|
||||
@@ -13,11 +13,11 @@
|
||||
"version": "0.1.0",
|
||||
"dependencies": {
|
||||
"nats": "2.29.3",
|
||||
"zod": "^3.23.8",
|
||||
"zod": "3.25.76",
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/bun": "latest",
|
||||
"typescript": "^5.5.4",
|
||||
"@types/bun": "1.4.3",
|
||||
"typescript": "5.9.3",
|
||||
},
|
||||
},
|
||||
"web": {
|
||||
@@ -25,14 +25,14 @@
|
||||
"version": "0.1.0",
|
||||
"dependencies": {
|
||||
"@noble/hashes": "1.8.0",
|
||||
"react": "^18.3.1",
|
||||
"react-dom": "^18.3.1",
|
||||
"react": "18.3.1",
|
||||
"react-dom": "18.3.1",
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/bun": "latest",
|
||||
"@types/react": "^18.3.3",
|
||||
"@types/react-dom": "^18.3.0",
|
||||
"typescript": "^5.5.4",
|
||||
"@types/bun": "1.4.3",
|
||||
"@types/react": "18.3.31",
|
||||
"@types/react-dom": "18.3.7",
|
||||
"typescript": "5.9.3",
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -9,10 +9,10 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"nats": "2.29.3",
|
||||
"zod": "^3.23.8"
|
||||
"zod": "3.25.76"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/bun": "latest",
|
||||
"typescript": "^5.5.4"
|
||||
"@types/bun": "1.4.3",
|
||||
"typescript": "5.9.3"
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import type { ModelTier } from "./limits.js";
|
||||
|
||||
/**
|
||||
* Minimal `--flag value` / `--flag=value` / `--switch` parser shared by the CLIs.
|
||||
* Names in `knownSwitches` take no operand; every other flag requires one.
|
||||
@@ -49,3 +51,28 @@ export function parseCommand(value: string): string[] {
|
||||
if (parts.length === 0) throw new Error("command must not be empty");
|
||||
return parts;
|
||||
}
|
||||
|
||||
const TIERS: readonly ModelTier[] = ["tier-1-opencode", "tier-2-commercial", "tier-3-local"];
|
||||
|
||||
/** Parse `--tier-models` JSON: an object from tier name to `provider/model`. */
|
||||
export function parseTierModels(value: string | undefined): Partial<Record<ModelTier, string>> | undefined {
|
||||
if (value === undefined) return undefined;
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(value);
|
||||
} catch (err) {
|
||||
throw new Error(`--tier-models must be JSON: ${(err as Error).message}`);
|
||||
}
|
||||
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
|
||||
throw new Error("--tier-models must be a JSON object of tier -> provider/model");
|
||||
}
|
||||
const models: Partial<Record<ModelTier, string>> = {};
|
||||
for (const [tier, model] of Object.entries(parsed)) {
|
||||
if (!TIERS.includes(tier as ModelTier)) throw new Error(`--tier-models: unknown tier ${JSON.stringify(tier)}`);
|
||||
if (typeof model !== "string" || !/^[^/#]+\/[^#]+(#.+)?$/.test(model)) {
|
||||
throw new Error(`--tier-models: ${tier} must map to "provider/model", got ${JSON.stringify(model)}`);
|
||||
}
|
||||
models[tier as ModelTier] = model;
|
||||
}
|
||||
return models;
|
||||
}
|
||||
@@ -88,7 +88,8 @@ export function extractRetryAfterMs(
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function isQuotaFailure(failure: ProviderFailure): boolean {
|
||||
/** True for usage-limit / quota / payment failures (the shapes that route to the next tier). */
|
||||
export function isQuotaFailure(failure: ProviderFailure): boolean {
|
||||
const codeQuota = failure.code !== undefined && QUOTA_CODE_PATTERN.test(failure.code);
|
||||
const messageQuota = QUOTA_MESSAGE_PATTERN.test(failure.message ?? "");
|
||||
return failure.status === 402 || codeQuota || messageQuota;
|
||||
|
||||
@@ -17,14 +17,18 @@
|
||||
* --director-cmd <cmd> Director server command (default: worker command)
|
||||
* --director-agent <name> OpenCode agent for the Director (default `director`)
|
||||
* --task-agent <name> OpenCode agent for task workers (default: server default)
|
||||
* --roles-dir <dir> role packs for `expert` tasks (FLEET_ROLES_DIR); see dispatcher/src/roles.ts
|
||||
* --tier-models <json> `{"tier-1-opencode": "provider/model", ...}` (FLEET_TIER_MODELS);
|
||||
* tier failover moves new worker sessions to the next tier's model
|
||||
* --fleet-cap <n> max concurrent workers (default 4)
|
||||
* --per-expert-cap <n> max concurrent workers per expert without a role cap (default: fleet cap)
|
||||
* --task-timeout-ms <n> per-task worker timeout (default 30 min)
|
||||
* --no-workers / --no-director
|
||||
* NATS_URL mirror events + board into NATS JetStream / KV
|
||||
*/
|
||||
import { homedir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { parseArgs, parseCommand } from "./cli-args.js";
|
||||
import { parseArgs, parseCommand, parseTierModels } from "./cli-args.js";
|
||||
import { NatsJetStreamBridge, natsUrlFromEnv, redactNatsUrl } from "./nats-bridge.js";
|
||||
import { startServer } from "./server.js";
|
||||
|
||||
@@ -58,6 +62,9 @@ async function main(): Promise<void> {
|
||||
: {
|
||||
command: workerCommand,
|
||||
agent: flags.get("task-agent"),
|
||||
rolesDir: flags.get("roles-dir") ?? env["FLEET_ROLES_DIR"],
|
||||
tierModels: parseTierModels(flags.get("tier-models") ?? env["FLEET_TIER_MODELS"]),
|
||||
perExpertCap: flags.has("per-expert-cap") ? Math.max(1, int(flags.get("per-expert-cap"), 1, "--per-expert-cap")) : undefined,
|
||||
taskTimeoutMs: flags.has("task-timeout-ms") ? int(flags.get("task-timeout-ms"), 0, "--task-timeout-ms") : undefined,
|
||||
},
|
||||
director: switches.has("no-director")
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
* 2. parses the listening URL from stdout (`server listening on <url>`,
|
||||
* or the `--stdio` form `{"url": ...}`);
|
||||
* 3. talks to the v2 API with HTTP Basic auth (`opencode:<password>`):
|
||||
* - `POST /api/session` `{ location: { directory }, agent?, title? }` -> `{ data: { id } }`
|
||||
* - `POST /api/session` `{ location: { directory }, agent?, title?, model? }` -> `{ data: { id } }`
|
||||
* - `POST /api/session/:id/prompt` `{ text }` (durably admits the input and schedules the agent loop)
|
||||
* - `GET /api/event` SSE, one `data: <json>` frame per event
|
||||
* (`session.tool.*`, `session.text.ended`, `session.status`,
|
||||
@@ -30,9 +30,23 @@ export interface WorkerEvent {
|
||||
data: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** OpenCode v2 `Model.Ref`: which provider model a session runs on. */
|
||||
export interface ModelRef {
|
||||
providerID: string;
|
||||
id: string;
|
||||
variant?: string | undefined;
|
||||
}
|
||||
|
||||
export interface CreateSessionInput {
|
||||
directory: string;
|
||||
agent?: string | undefined;
|
||||
title?: string | undefined;
|
||||
model?: ModelRef | undefined;
|
||||
}
|
||||
|
||||
/** The slice of the OpenCode v2 API the dispatcher uses. */
|
||||
export interface WorkerApi {
|
||||
createSession(input: { directory: string; agent?: string | undefined; title?: string | undefined }): Promise<{ id: string }>;
|
||||
createSession(input: CreateSessionInput): Promise<{ id: string }>;
|
||||
prompt(sessionId: string, text: string): Promise<void>;
|
||||
/** Open the event stream. Resolves once the stream is connected; events flow to `onEvent` until `signal` aborts or the stream ends. */
|
||||
subscribe(onEvent: (event: WorkerEvent) => void, signal: AbortSignal): Promise<{ closed: Promise<void> }>;
|
||||
@@ -70,10 +84,11 @@ export class OpencodeHttpClient implements WorkerApi {
|
||||
}
|
||||
}
|
||||
|
||||
async createSession(input: { directory: string; agent?: string | undefined; title?: string | undefined }): Promise<{ id: string }> {
|
||||
async createSession(input: CreateSessionInput): Promise<{ id: string }> {
|
||||
const payload: Record<string, unknown> = { location: { directory: input.directory } };
|
||||
if (input.agent !== undefined) payload["agent"] = input.agent;
|
||||
if (input.title !== undefined) payload["title"] = input.title;
|
||||
if (input.model !== undefined) payload["model"] = input.model;
|
||||
const result = (await this.request("POST", "/api/session", payload)) as { data?: { id?: unknown } } | null;
|
||||
const id = result?.data?.id;
|
||||
if (typeof id !== "string" || id.length === 0) {
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
/**
|
||||
* Per-role worker shaping for `expert` tasks.
|
||||
*
|
||||
* A role pack lives at `<rolesDir>/<role-id>/` and is laid out as an
|
||||
* OpenCode XDG config home, so the worker can run with
|
||||
* `XDG_CONFIG_HOME=<rolesDir>/<role-id>`:
|
||||
*
|
||||
* ```text
|
||||
* <rolesDir>/<role-id>/
|
||||
* role.json optional: { "maxConcurrent": n }
|
||||
* opencode/opencode.json copied into the worktree `.opencode/`
|
||||
* opencode/agent/<role-id>.md role system prompt (frontmatter stripped)
|
||||
* opencode/skill/<name>/SKILL.md skills listed in the task prompt
|
||||
* ```
|
||||
*
|
||||
* `agents/` and `skills/` are accepted as plural aliases. Role ids come
|
||||
* from Director output, so only slugs resolve: anything else (or a
|
||||
* missing directory) is "no role" and the worker runs with the generic
|
||||
* prompt.
|
||||
*/
|
||||
import { copyFileSync, existsSync, readdirSync, readFileSync, statSync } from "node:fs";
|
||||
import path from "node:path";
|
||||
|
||||
export interface RoleProfile {
|
||||
id: string;
|
||||
/** Role directory, used as the worker's `XDG_CONFIG_HOME`. */
|
||||
dir: string;
|
||||
/** Role agent prompt body, or undefined when the role has none. */
|
||||
prompt?: string | undefined;
|
||||
/** Absolute paths of the role's `SKILL.md` files, sorted. */
|
||||
skills: string[];
|
||||
/** Role `opencode.json`, copied into each worktree's `.opencode/`. */
|
||||
configFile?: string | undefined;
|
||||
/** Per-role cap on concurrently running workers. */
|
||||
maxConcurrent?: number | undefined;
|
||||
}
|
||||
|
||||
const ROLE_ID = /^[a-z0-9][a-z0-9-]*$/;
|
||||
|
||||
function firstExisting(candidates: string[]): string | undefined {
|
||||
return candidates.find((candidate) => existsSync(candidate) && statSync(candidate).isFile());
|
||||
}
|
||||
|
||||
/** Drop a leading `---` YAML frontmatter block from an agent markdown file. */
|
||||
export function stripFrontmatter(text: string): string {
|
||||
const normalized = text.replace(/^/, "");
|
||||
if (!normalized.startsWith("---\n") && !normalized.startsWith("---\r\n")) return normalized.trim();
|
||||
const end = normalized.indexOf("\n---", 3);
|
||||
if (end === -1) return normalized.trim();
|
||||
const after = normalized.indexOf("\n", end + 4);
|
||||
return (after === -1 ? "" : normalized.slice(after + 1)).trim();
|
||||
}
|
||||
|
||||
function listSkills(opencodeDir: string): string[] {
|
||||
const skills: string[] = [];
|
||||
for (const name of ["skill", "skills"]) {
|
||||
const root = path.join(opencodeDir, name);
|
||||
if (!existsSync(root) || !statSync(root).isDirectory()) continue;
|
||||
for (const entry of readdirSync(root, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
const file = path.join(root, entry.name, "SKILL.md");
|
||||
if (existsSync(file)) skills.push(file);
|
||||
}
|
||||
}
|
||||
return skills.sort();
|
||||
}
|
||||
|
||||
function readMaxConcurrent(roleDir: string): number | undefined {
|
||||
const file = path.join(roleDir, "role.json");
|
||||
if (!existsSync(file)) return undefined;
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(readFileSync(file, "utf8"));
|
||||
} catch (err) {
|
||||
throw new Error(`role ${path.basename(roleDir)}: role.json is not JSON: ${(err as Error).message}`);
|
||||
}
|
||||
const value = typeof parsed === "object" && parsed !== null ? (parsed as { maxConcurrent?: unknown }).maxConcurrent : undefined;
|
||||
if (value === undefined) return undefined;
|
||||
if (typeof value !== "number" || !Number.isInteger(value) || value < 1) {
|
||||
throw new Error(`role ${path.basename(roleDir)}: role.json maxConcurrent must be a positive integer`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Load the role pack for `roleId`, or undefined when there is none. */
|
||||
export function loadRole(rolesDir: string, roleId: string): RoleProfile | undefined {
|
||||
if (!ROLE_ID.test(roleId)) return undefined;
|
||||
const dir = path.join(path.resolve(rolesDir), roleId);
|
||||
if (!existsSync(dir) || !statSync(dir).isDirectory()) return undefined;
|
||||
const opencodeDir = path.join(dir, "opencode");
|
||||
const promptFile = firstExisting([
|
||||
path.join(opencodeDir, "agent", `${roleId}.md`),
|
||||
path.join(opencodeDir, "agents", `${roleId}.md`),
|
||||
]);
|
||||
const prompt = promptFile === undefined ? undefined : stripFrontmatter(readFileSync(promptFile, "utf8"));
|
||||
return {
|
||||
id: roleId,
|
||||
dir,
|
||||
prompt: prompt !== undefined && prompt.length > 0 ? prompt : undefined,
|
||||
skills: listSkills(opencodeDir),
|
||||
configFile: firstExisting([path.join(opencodeDir, "opencode.json"), path.join(opencodeDir, "opencode.jsonc")]),
|
||||
maxConcurrent: readMaxConcurrent(dir),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Copy (never symlink) the role's OpenCode config into the worktree's
|
||||
* isolated `.opencode/`, so workers never share config state.
|
||||
*/
|
||||
export function installRoleConfig(role: RoleProfile, worktreeDir: string): string | undefined {
|
||||
if (role.configFile === undefined) return undefined;
|
||||
const target = path.join(worktreeDir, ".opencode", path.basename(role.configFile));
|
||||
copyFileSync(role.configFile, target);
|
||||
return target;
|
||||
}
|
||||
+163
-24
@@ -19,8 +19,11 @@
|
||||
* returns the exact es-core-compatible payloads to sign.
|
||||
* - `POST /api/approvals/:taskId/sign` — burn the ticket, recompute the
|
||||
* diff hash, and verify the Ed25519 signature against the PINNED human
|
||||
* key (`human.pub`) only; `waitingHuman` -> `done` (APPROVE) or
|
||||
* `failed` (REJECT); audit-chained with the full signed record.
|
||||
* key (`human.pub`) only. APPROVE lands `fleet/<taskId>` onto the
|
||||
* mission's integration branch (`landing.ts`) and moves the task to
|
||||
* `done` with its `landedSha`; a landing conflict keeps it
|
||||
* `waitingHuman` (409, conflicting files listed). REJECT -> `failed`.
|
||||
* Audit-chained with the full signed record.
|
||||
* - `GET /api/events` — SSE stream with `Last-Event-ID` replay.
|
||||
* - `GET /api/csrf` — issue the CSRF token POST routes require.
|
||||
* - `GET /*` — the built WebUI (`web/dist`) when `webDistDir` is set.
|
||||
@@ -29,6 +32,10 @@
|
||||
* matching `x-csrf-token` → 403; no pinned human key → 503 on approval
|
||||
* routes; unknown/expired/reused/mismatched tickets → 403; bad
|
||||
* signatures → 403; approving a task that is not `waitingHuman` → 409.
|
||||
*
|
||||
* Worktree lifecycle: a task reaching `done` / `failed` / `cancelled`
|
||||
* releases its worktree once no worker runs in it; `failed` and
|
||||
* `cancelled` branches are then salvaged to `refs/fleet-salvage/<id>`.
|
||||
*/
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { existsSync, mkdirSync, mkdtempSync, statSync } from "node:fs";
|
||||
@@ -47,20 +54,32 @@ import {
|
||||
type ApprovalPayload,
|
||||
type PinnedHumanKey,
|
||||
} from "./human-key.js";
|
||||
import { classifyProviderFailure, type ModelTier, type ProviderFailure } from "./limits.js";
|
||||
import { landTask, type LandResult } from "./landing.js";
|
||||
import { classifyProviderFailure, type ModelTier, type ProviderDisposition, type ProviderFailure } from "./limits.js";
|
||||
import { TaskSchema, TaskStatuses, type Task, type TaskStatus } from "./model.js";
|
||||
import { attachBusToNats, type NatsJetStreamBridge } from "./nats-bridge.js";
|
||||
import { DEFAULT_WORKER_COMMAND } from "./opencode-client.js";
|
||||
import { schedule } from "./scheduler.js";
|
||||
import { WorkerLauncher } from "./worker-launcher.js";
|
||||
import type { WorkerClientFactory } from "./worker-session.js";
|
||||
import { WorktreeStore, cleanupUnconfirmed, runGit, type WorktreeState } from "./worktrees.js";
|
||||
import {
|
||||
WorktreeStore,
|
||||
cleanupUnconfirmed,
|
||||
integrationBranch,
|
||||
releaseWorktree,
|
||||
runGit,
|
||||
salvageWorktree,
|
||||
type WorktreeState,
|
||||
} from "./worktrees.js";
|
||||
|
||||
export const LOOPBACK_HOSTS: readonly string[] = ["127.0.0.1", "localhost", "::1"] as const;
|
||||
export const CSRF_HEADER = "x-csrf-token";
|
||||
export const SCHEDULER_FLEET_CAP = 4;
|
||||
export const MAX_TASKS_PER_MESSAGE = 50;
|
||||
|
||||
/** Statuses after which a task never runs again: its worktree is released. */
|
||||
const FINISHED: ReadonlySet<TaskStatus> = new Set<TaskStatus>(["done", "failed", "cancelled"]);
|
||||
|
||||
export type BoardColumn = "Backlog" | "Ready" | "In Progress" | "Waiting Approval / QA" | "Done";
|
||||
|
||||
export const BOARD_COLUMNS: readonly BoardColumn[] = [
|
||||
@@ -98,7 +117,8 @@ export interface WorktreeStatusView {
|
||||
taskId: string;
|
||||
branch: string;
|
||||
path: string;
|
||||
state: WorktreeState;
|
||||
/** `released`: the task finished and its worktree was removed. */
|
||||
state: WorktreeState | "released";
|
||||
}
|
||||
|
||||
export interface DagEdge {
|
||||
@@ -187,6 +207,10 @@ export interface WorkerConfig {
|
||||
startTimeoutMs?: number | undefined;
|
||||
taskTimeoutMs?: number | undefined;
|
||||
perExpertCap?: number | undefined;
|
||||
/** Role packs for `expert` tasks (`<rolesDir>/<role-id>/`, see `roles.ts`). */
|
||||
rolesDir?: string | undefined;
|
||||
/** `provider/model` per model tier; tier failover switches new sessions to the next tier's model. */
|
||||
tierModels?: Partial<Record<ModelTier, string>> | undefined;
|
||||
env?: Record<string, string | undefined> | undefined;
|
||||
clientFactory?: WorkerClientFactory | undefined;
|
||||
}
|
||||
@@ -234,7 +258,7 @@ export interface FleetServer {
|
||||
getTask(id: string): Task | undefined;
|
||||
upsertTask(input: UpsertTaskInput): Task;
|
||||
setTier(tier: ModelTier): void;
|
||||
reportProviderFailure(failure: ProviderFailure): { action: string };
|
||||
reportProviderFailure(failure: ProviderFailure): ProviderDisposition;
|
||||
reportProviderSuccess(): void;
|
||||
consecutiveFailures(): number;
|
||||
updateTask(id: string, patch: Partial<Task>): Task;
|
||||
@@ -446,6 +470,8 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
let worktreeStore: WorktreeStore | null = null;
|
||||
let natsBridge: NatsJetStreamBridge | null = null;
|
||||
let detachNats: (() => void) | null = null;
|
||||
// Landings onto shared integration branches run one at a time.
|
||||
let landingQueue: Promise<unknown> = Promise.resolve();
|
||||
|
||||
const boardStore = stateDir !== undefined ? new BoardStore(join(stateDir, "dispatcher.db")) : null;
|
||||
|
||||
@@ -593,10 +619,69 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
storeTask(task);
|
||||
if (task.status !== existing.status) {
|
||||
bus.publish("task.updated", { taskId: id, status: task.status, reason: task.reason ?? null });
|
||||
releaseIfFinished(id);
|
||||
}
|
||||
return task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release a finished task's worktree, then salvage the branch of a task
|
||||
* that did not succeed. Never throws: failures are audited and published
|
||||
* so a stuck worktree cannot wedge a status change.
|
||||
*/
|
||||
function releaseIfFinished(taskId: string): void {
|
||||
const task = tasks.get(taskId);
|
||||
if (repoDir === undefined || worktreeStore === null || task === undefined || !FINISHED.has(task.status)) return;
|
||||
if (!/^[a-z0-9][a-z0-9-]*$/.test(taskId)) return;
|
||||
// A worker still running in the tree: released when its run settles.
|
||||
if (launcher?.runningIds().includes(taskId) === true) return;
|
||||
const store = worktreeStore;
|
||||
const record = store.get(taskId);
|
||||
try {
|
||||
if (record !== undefined && record.state !== "salvage") {
|
||||
try {
|
||||
releaseWorktree({ repoDir, taskId, store });
|
||||
} catch (err) {
|
||||
if (existsSync(record.path)) throw err;
|
||||
// The directory is already gone: drop the stale registration.
|
||||
runGit(repoDir, ["worktree", "prune"]);
|
||||
store.remove(taskId);
|
||||
}
|
||||
audit("worktree.released", { taskId, path: record.path, status: task.status });
|
||||
bus.publish("worktree.released", { taskId, status: task.status });
|
||||
}
|
||||
if (task.status !== TaskStatuses.done) {
|
||||
const salvaged = salvageWorktree({ repoDir, taskId, store });
|
||||
if (salvaged !== undefined) {
|
||||
audit("worktree.salvaged", { taskId, ref: salvaged.ref, tip: salvaged.tip });
|
||||
bus.publish("worktree.salvaged", { taskId, ref: salvaged.ref, tip: salvaged.tip });
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
audit("worktree.release-failed", { taskId, error: (err as Error).message });
|
||||
bus.publish("worktree.release-failed", { taskId, error: (err as Error).message });
|
||||
}
|
||||
}
|
||||
|
||||
function reportProviderFailure(failure: ProviderFailure): ProviderDisposition {
|
||||
failures += 1;
|
||||
const disposition = classifyProviderFailure(failure, failures, activeTier);
|
||||
if (disposition.action === "route-fallback") {
|
||||
activeTier = disposition.targetTier;
|
||||
failures = 0;
|
||||
audit("tier.fallback", { targetTier: activeTier });
|
||||
bus.publish("tier.fallback", { targetTier: activeTier });
|
||||
} else if (disposition.action === "pause-missions") {
|
||||
const resumeAtMs = Date.now() + disposition.resumeAfterMs;
|
||||
launcher?.pauseUntil(resumeAtMs);
|
||||
audit("missions.paused", { reason: disposition.reason, resumeAtMs });
|
||||
bus.publish("missions.paused", { reason: disposition.reason, resumeAtMs });
|
||||
} else {
|
||||
bus.publish("provider.retry", { action: disposition.action, delayMs: disposition.delayMs });
|
||||
}
|
||||
return disposition;
|
||||
}
|
||||
|
||||
function getBoard(): BoardView {
|
||||
const all = [...tasks.values()].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0));
|
||||
const columns: BoardColumnView[] = BOARD_COLUMNS.map((name) => ({ name, taskIds: [] }));
|
||||
@@ -614,7 +699,7 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
taskId: task.id,
|
||||
branch: stored?.branch ?? task.branch ?? `fleet/${task.id}`,
|
||||
path: stored?.path ?? task.worktree ?? "",
|
||||
state: stored?.state ?? "allocated",
|
||||
state: stored?.state ?? (FINISHED.has(task.status) && task.worktree !== undefined ? "released" : "allocated"),
|
||||
});
|
||||
}
|
||||
const dagEdges: DagEdge[] = [];
|
||||
@@ -760,6 +845,9 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
bus.publish(type, data);
|
||||
},
|
||||
audit,
|
||||
onRunSettled: releaseIfFinished,
|
||||
activeTier: () => activeTier,
|
||||
reportProviderFailure,
|
||||
},
|
||||
{
|
||||
repoDir,
|
||||
@@ -768,6 +856,8 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
command: workerCommand,
|
||||
caps: perExpertCap === undefined ? { fleet: fleetCap } : { fleet: fleetCap, perExpert: () => perExpertCap },
|
||||
agent: workerConfig.agent,
|
||||
rolesDir: workerConfig.rolesDir === undefined ? undefined : resolve(workerConfig.rolesDir),
|
||||
tierModels: workerConfig.tierModels,
|
||||
startTimeoutMs: workerConfig.startTimeoutMs,
|
||||
taskTimeoutMs: workerConfig.taskTimeoutMs,
|
||||
env: workerConfig.env,
|
||||
@@ -792,6 +882,11 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
}
|
||||
}
|
||||
|
||||
// Finished tasks whose worktree outlived the previous process.
|
||||
if (worktreeStore !== null) {
|
||||
for (const record of worktreeStore.list()) releaseIfFinished(record.taskId);
|
||||
}
|
||||
|
||||
function handleEvents(req: Request): Response {
|
||||
const url = new URL(req.url);
|
||||
const headerId = req.headers.get("last-event-id");
|
||||
@@ -1056,15 +1151,70 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
if (task.status !== TaskStatuses.waitingHuman) {
|
||||
return fail(409, `task ${JSON.stringify(taskId)} is ${task.status}, only waitingHuman tasks can be signed`);
|
||||
}
|
||||
const signed = signedRecord(humanKey, payload, signatureHex as string);
|
||||
let landedSha: string | undefined;
|
||||
if (approve && repoDir !== undefined && stateDir !== undefined) {
|
||||
const integration = missions.get(task.missionId)?.integrationBranch ?? integrationBranch(task.missionId);
|
||||
const landed = await landApproved(task.id, integration);
|
||||
if (landed === null) {
|
||||
return fail(409, `task ${JSON.stringify(taskId)} changed state while landing was queued`);
|
||||
}
|
||||
if (!landed.ok) {
|
||||
const files = landed.files ?? [];
|
||||
const reason =
|
||||
landed.reason === "conflict"
|
||||
? `approved, but landing on ${integration} conflicts in: ${files.join(", ")}`
|
||||
: `approved, but landing on ${integration} failed at ${landed.stage}: ${landed.reason}`;
|
||||
updateTask(task.id, { status: TaskStatuses.waitingHuman, reason: reason.slice(0, 2000) });
|
||||
audit("approval.signed", { taskId, approve, status: TaskStatuses.waitingHuman, landed: false, stage: landed.stage, files, record: signed });
|
||||
bus.publish("approval.landing-failed", { taskId, integration, stage: landed.stage, reason: landed.reason, files });
|
||||
return fail(409, reason, { code: "landing-failed", stage: landed.stage, files, record: signed });
|
||||
}
|
||||
landedSha = landed.mergedSha;
|
||||
}
|
||||
const next: TaskStatus = approve ? TaskStatuses.done : TaskStatuses.failed;
|
||||
updateTask(task.id, {
|
||||
status: next,
|
||||
reason: approve ? "approved by human (Ed25519)" : "rejected by human (Ed25519)",
|
||||
reason: approve
|
||||
? landedSha !== undefined
|
||||
? `approved by human (Ed25519); landed ${landedSha.slice(0, 12)}`
|
||||
: "approved by human (Ed25519)"
|
||||
: "rejected by human (Ed25519)",
|
||||
...(landedSha !== undefined ? { landedSha } : {}),
|
||||
});
|
||||
const signed = signedRecord(humanKey, payload, signatureHex as string);
|
||||
audit("approval.signed", { taskId, approve, status: next, record: signed });
|
||||
bus.publish("approval.signed", { taskId, status: next, approve, subjectHash: binding.subjectHash });
|
||||
return json({ taskId, status: next, subjectHash: binding.subjectHash, record: signed }, 200);
|
||||
audit("approval.signed", { taskId, approve, status: next, landedSha: landedSha ?? null, record: signed });
|
||||
bus.publish("approval.signed", { taskId, status: next, approve, subjectHash: binding.subjectHash, landedSha: landedSha ?? null });
|
||||
return json({ taskId, status: next, subjectHash: binding.subjectHash, landedSha: landedSha ?? null, record: signed }, 200);
|
||||
}
|
||||
|
||||
/**
|
||||
* Land an approved task onto its mission's integration branch, one
|
||||
* landing at a time. Null when the task left `waitingHuman` while queued.
|
||||
*/
|
||||
function landApproved(taskId: string, integration: string): Promise<LandResult | null> {
|
||||
const run = async (): Promise<LandResult | null> => {
|
||||
const task = tasks.get(taskId);
|
||||
if (repoDir === undefined || stateDir === undefined || task === undefined || task.status !== TaskStatuses.waitingHuman) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return await landTask({
|
||||
repoDir,
|
||||
landingRoot: join(stateDir, "landing"),
|
||||
taskId,
|
||||
missionId: task.missionId,
|
||||
taskBranch: task.branch,
|
||||
integration,
|
||||
base: task.input.base,
|
||||
audit,
|
||||
});
|
||||
} catch (err) {
|
||||
return { ok: false, reason: (err as Error).message, stage: "merge", rolledBack: false };
|
||||
}
|
||||
};
|
||||
const result = landingQueue.then(run, run);
|
||||
landingQueue = result.catch(() => null);
|
||||
return result;
|
||||
}
|
||||
|
||||
async function route(req: Request): Promise<Response> {
|
||||
@@ -1129,18 +1279,7 @@ export function createServer(options: CreateServerOptions = {}): FleetServer {
|
||||
setTier: (tier: ModelTier) => {
|
||||
activeTier = tier;
|
||||
},
|
||||
reportProviderFailure: (failure: ProviderFailure) => {
|
||||
failures += 1;
|
||||
const disposition = classifyProviderFailure(failure, failures, activeTier);
|
||||
if (disposition.action === "route-fallback") {
|
||||
activeTier = disposition.targetTier;
|
||||
failures = 0;
|
||||
bus.publish("tier.fallback", { targetTier: activeTier });
|
||||
} else {
|
||||
bus.publish("provider.retry", { action: disposition.action });
|
||||
}
|
||||
return { action: disposition.action };
|
||||
},
|
||||
reportProviderFailure,
|
||||
reportProviderSuccess: () => {
|
||||
failures = 0;
|
||||
},
|
||||
|
||||
@@ -12,21 +12,32 @@
|
||||
* 1. allocate (or reuse) its worktree via `worktrees.ts`, pinning the
|
||||
* base commit so the approval diff is stable;
|
||||
* 2. spawn the worker command in the worktree with isolated
|
||||
* `.opencode` state;
|
||||
* 3. create a session, send the task prompt, and republish tool /
|
||||
* `.opencode` state; an `expert` task with a role pack (`roles.ts`)
|
||||
* gets `XDG_CONFIG_HOME=<rolesDir>/<role>`, the role `opencode.json`
|
||||
* copied into its `.opencode/`, and the role prompt + skills in its
|
||||
* task prompt;
|
||||
* 3. create a session on the active model tier's model (when
|
||||
* `tierModels` maps it), send the task prompt, and republish tool /
|
||||
* status / text events on the dispatcher bus (drill-down live log);
|
||||
* 4. on success commit the worktree changes onto `fleet/<taskId>` and
|
||||
* move the task to `waitingHuman` (non-empty diff: a human must sign
|
||||
* it) or `done` (no changes); on failure / exit / timeout apply the
|
||||
* task's retry policy (`nextStatus`).
|
||||
* it) or `done` (no changes); on a quota-shaped provider failure
|
||||
* report it to the tier failover and re-queue the task without
|
||||
* spending an attempt; on any other failure / exit / timeout apply
|
||||
* the task's retry policy (`nextStatus`).
|
||||
*
|
||||
* While missions are paused (last tier exhausted) nothing new launches
|
||||
* until the provider's retry time.
|
||||
*
|
||||
* `stop()` aborts every run, kills every worker, and re-queues
|
||||
* interrupted tasks as `pending` so a restart resumes them.
|
||||
*/
|
||||
import { existsSync } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { isQuotaFailure, type ModelTier, type ProviderDisposition, type ProviderFailure } from "./limits.js";
|
||||
import type { Task, TaskStatus } from "./model.js";
|
||||
import type { WorkerEvent } from "./opencode-client.js";
|
||||
import type { ModelRef, WorkerEvent } from "./opencode-client.js";
|
||||
import { installRoleConfig, loadRole, type RoleProfile } from "./roles.js";
|
||||
import { nextStatus, schedule, type ScheduleCaps } from "./scheduler.js";
|
||||
import { describeError, openWorker, runTurn, type WorkerClientFactory, type WorkerConnection } from "./worker-session.js";
|
||||
import { allocateWorktree, runGit, type WorktreeStore } from "./worktrees.js";
|
||||
@@ -41,6 +52,12 @@ export interface LauncherHost {
|
||||
audit(event: string, data: unknown): void;
|
||||
/** Called after a run finishes and the task branch holds the worker's commit. */
|
||||
afterRun?: ((taskId: string) => void) | undefined;
|
||||
/** Called once a run is fully over: worker killed, task no longer running here. */
|
||||
onRunSettled?: ((taskId: string) => void) | undefined;
|
||||
/** The model tier new sessions run on. */
|
||||
activeTier?: (() => ModelTier) | undefined;
|
||||
/** Tier failover: classify a provider failure and apply its disposition. */
|
||||
reportProviderFailure?: ((failure: ProviderFailure) => ProviderDisposition) | undefined;
|
||||
}
|
||||
|
||||
export interface WorkerLauncherOptions {
|
||||
@@ -51,6 +68,10 @@ export interface WorkerLauncherOptions {
|
||||
caps: ScheduleCaps;
|
||||
/** OpenCode agent for task sessions (omit for the server default). */
|
||||
agent?: string | undefined;
|
||||
/** Role packs for `expert` tasks (see `roles.ts`). */
|
||||
rolesDir?: string | undefined;
|
||||
/** `provider/model` per tier; a mapped tier pins the session model. */
|
||||
tierModels?: Partial<Record<ModelTier, string>> | undefined;
|
||||
startTimeoutMs?: number | undefined;
|
||||
taskTimeoutMs?: number | undefined;
|
||||
env?: Record<string, string | undefined> | undefined;
|
||||
@@ -85,13 +106,32 @@ function contentText(content: unknown): string {
|
||||
.join("\n");
|
||||
}
|
||||
|
||||
/** The prompt a task worker receives. */
|
||||
export function taskPrompt(task: Task): string {
|
||||
/** Parse a `provider/model[#variant]` reference into the OpenCode v2 `Model.Ref` shape. */
|
||||
export function parseModelRef(value: string): ModelRef {
|
||||
const slash = value.indexOf("/");
|
||||
const hash = value.indexOf("#", slash + 1);
|
||||
const providerID = value.slice(0, Math.max(slash, 0));
|
||||
const id = value.slice(slash + 1, hash === -1 ? undefined : hash);
|
||||
const variant = hash === -1 ? undefined : value.slice(hash + 1);
|
||||
if (slash <= 0 || id.length === 0 || providerID.includes("#") || variant === "") {
|
||||
throw new Error(`invalid model reference ${JSON.stringify(value)} (want provider/model)`);
|
||||
}
|
||||
return variant === undefined ? { providerID, id } : { providerID, id, variant };
|
||||
}
|
||||
|
||||
/** The prompt a task worker receives, shaped by its role pack when it has one. */
|
||||
export function taskPrompt(task: Task, role?: RoleProfile): string {
|
||||
const lines = [
|
||||
`You are a fleet worker for task ${task.id}${task.title !== undefined ? `: ${task.title}` : ""}.`,
|
||||
"",
|
||||
`Objective: ${task.input.objective}`,
|
||||
];
|
||||
if (role !== undefined) {
|
||||
lines.push("", `Role: ${role.id}`);
|
||||
if (role.prompt !== undefined) lines.push("", role.prompt);
|
||||
if (role.skills.length > 0) {
|
||||
lines.push("", "Role skills (read the SKILL.md that fits before starting):", ...role.skills.map((file) => `- ${file}`));
|
||||
}
|
||||
}
|
||||
lines.push("", `Objective: ${task.input.objective}`);
|
||||
if (task.input.context !== undefined) lines.push("", `Context: ${task.input.context}`);
|
||||
if (task.input.constraints !== undefined) lines.push("", `Constraints: ${task.input.constraints}`);
|
||||
if (task.input.acceptance.length > 0) {
|
||||
@@ -113,6 +153,8 @@ export class WorkerLauncher {
|
||||
private readonly active = new Map<string, ActiveRun>();
|
||||
private stopping = false;
|
||||
private scheduled = false;
|
||||
private pausedUntilMs = 0;
|
||||
private resumeTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly host: LauncherHost,
|
||||
@@ -123,6 +165,50 @@ export class WorkerLauncher {
|
||||
return [...this.active.keys()].sort();
|
||||
}
|
||||
|
||||
/** Hold every new launch until `untilMs` (epoch ms); running workers continue. */
|
||||
pauseUntil(untilMs: number): void {
|
||||
if (untilMs <= this.pausedUntilMs) return;
|
||||
this.pausedUntilMs = untilMs;
|
||||
if (this.resumeTimer !== null) clearTimeout(this.resumeTimer);
|
||||
this.resumeTimer = setTimeout(() => {
|
||||
this.resumeTimer = null;
|
||||
this.kick();
|
||||
}, Math.max(0, untilMs - Date.now()));
|
||||
}
|
||||
|
||||
/** Epoch ms until which launches are held, or 0. */
|
||||
pausedUntil(): number {
|
||||
return this.pausedUntilMs > Date.now() ? this.pausedUntilMs : 0;
|
||||
}
|
||||
|
||||
private roleFor(expert: string | undefined): RoleProfile | undefined {
|
||||
if (expert === undefined || this.options.rolesDir === undefined) return undefined;
|
||||
return loadRole(this.options.rolesDir, expert);
|
||||
}
|
||||
|
||||
/** Scheduler caps with per-role `maxConcurrent` taking precedence over the configured per-expert cap. */
|
||||
private capsForTick(): ScheduleCaps {
|
||||
const base = this.options.caps;
|
||||
if (this.options.rolesDir === undefined) return base;
|
||||
const cache = new Map<string, number | undefined>();
|
||||
return {
|
||||
...base,
|
||||
perExpert: (expert: string) => {
|
||||
if (!cache.has(expert)) {
|
||||
let cap: number | undefined;
|
||||
try {
|
||||
cap = this.roleFor(expert)?.maxConcurrent;
|
||||
} catch (err) {
|
||||
this.host.publish("role.invalid", { role: expert, error: (err as Error).message });
|
||||
cap = undefined;
|
||||
}
|
||||
cache.set(expert, cap);
|
||||
}
|
||||
return cache.get(expert) ?? base.perExpert?.(expert) ?? base.fleet;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Re-run scheduling soon (coalesced). Safe to call from any board mutation. */
|
||||
kick(): void {
|
||||
if (this.stopping || this.scheduled) return;
|
||||
@@ -136,7 +222,7 @@ export class WorkerLauncher {
|
||||
private tick(): void {
|
||||
if (this.stopping) return;
|
||||
const tasks = this.host.listTasks();
|
||||
const plan = schedule(tasks, this.active.keys(), this.options.caps);
|
||||
const plan = schedule(tasks, this.active.keys(), this.capsForTick());
|
||||
for (const id of plan.cancel) {
|
||||
this.host.updateTask(id, { status: "cancelled", reason: "a dependency can never succeed" });
|
||||
}
|
||||
@@ -146,6 +232,7 @@ export class WorkerLauncher {
|
||||
this.host.updateTask(id, { status: "ready" });
|
||||
}
|
||||
}
|
||||
const paused = this.pausedUntil() > 0;
|
||||
for (const id of plan.start) {
|
||||
const task = this.host.getTask(id);
|
||||
if (task === undefined) continue;
|
||||
@@ -153,6 +240,12 @@ export class WorkerLauncher {
|
||||
this.host.updateTask(id, { status: "waitingHuman", reason: "human gate" });
|
||||
continue;
|
||||
}
|
||||
if (paused) {
|
||||
if (task.status === "pending" || task.status === "backlog") {
|
||||
this.host.updateTask(id, { status: "ready", reason: "missions paused: model quota exhausted on every tier" });
|
||||
}
|
||||
continue;
|
||||
}
|
||||
this.launch(task);
|
||||
}
|
||||
}
|
||||
@@ -168,6 +261,7 @@ export class WorkerLauncher {
|
||||
})
|
||||
.finally(() => {
|
||||
this.active.delete(task.id);
|
||||
this.host.onRunSettled?.(task.id);
|
||||
this.kick();
|
||||
});
|
||||
}
|
||||
@@ -196,6 +290,22 @@ export class WorkerLauncher {
|
||||
const task = this.host.getTask(taskId);
|
||||
if (task === undefined) return;
|
||||
const { dir, base } = this.ensureWorktree(task);
|
||||
let role: RoleProfile | undefined;
|
||||
try {
|
||||
role = this.roleFor(task.expert);
|
||||
} catch (err) {
|
||||
this.finish(taskId, "failure", (err as Error).message);
|
||||
return;
|
||||
}
|
||||
if (role !== undefined) {
|
||||
installRoleConfig(role, dir);
|
||||
this.host.audit("worker.role", { taskId, role: role.id, skills: role.skills.length, config: role.configFile !== undefined });
|
||||
} else if (task.expert !== undefined && this.options.rolesDir !== undefined) {
|
||||
this.host.publish("role.missing", { taskId, role: task.expert });
|
||||
}
|
||||
const tier = this.host.activeTier?.();
|
||||
const modelName = tier === undefined ? undefined : this.options.tierModels?.[tier];
|
||||
const model = modelName === undefined ? undefined : parseModelRef(modelName);
|
||||
this.host.updateTask(taskId, {
|
||||
worktree: dir,
|
||||
branch: `fleet/${taskId}`,
|
||||
@@ -211,7 +321,7 @@ export class WorkerLauncher {
|
||||
command: this.options.command,
|
||||
cwd: dir,
|
||||
stateDir: path.join(dir, ".opencode"),
|
||||
env: this.options.env,
|
||||
env: role === undefined ? this.options.env : { ...this.options.env, XDG_CONFIG_HOME: role.dir },
|
||||
startTimeoutMs: this.options.startTimeoutMs,
|
||||
clientFactory: this.options.clientFactory,
|
||||
});
|
||||
@@ -223,18 +333,22 @@ export class WorkerLauncher {
|
||||
directory: dir,
|
||||
agent: this.options.agent,
|
||||
title: current.title ?? taskId,
|
||||
model,
|
||||
});
|
||||
this.host.updateTask(taskId, { sessionId: session.id, reason: "worker running" });
|
||||
this.host.publish("worker.started", { taskId, sessionId: session.id, pid: conn.proc.pid });
|
||||
this.host.audit("worker.started", { taskId, sessionId: session.id, command: this.options.command[0] });
|
||||
this.host.publish("worker.started", { taskId, sessionId: session.id, pid: conn.proc.pid, tier: tier ?? null, model: modelName ?? null });
|
||||
this.host.audit("worker.started", { taskId, sessionId: session.id, command: this.options.command[0], tier: tier ?? null, model: modelName ?? null });
|
||||
const toolNames = new Map<string, string>();
|
||||
const toolInputs = new Map<string, string>();
|
||||
const result = await runTurn(conn, session.id, taskPrompt(current), {
|
||||
const result = await runTurn(conn, session.id, taskPrompt(current, role), {
|
||||
timeoutMs: this.options.taskTimeoutMs ?? 30 * 60_000,
|
||||
signal: run.abort.signal,
|
||||
onEvent: (event) => this.relay(taskId, event, toolNames, toolInputs),
|
||||
});
|
||||
if (result.outcome === "aborted") return;
|
||||
if (result.outcome === "failed" && result.error !== undefined && this.requeueOnQuota(taskId, result.error) !== null) {
|
||||
return;
|
||||
}
|
||||
if (result.outcome !== "succeeded") {
|
||||
this.finish(taskId, "failure", result.error ?? result.outcome);
|
||||
return;
|
||||
@@ -284,6 +398,35 @@ export class WorkerLauncher {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A quota-shaped provider failure is the model tier's fault, not the
|
||||
* task's: report it to the tier failover and put the task back in the
|
||||
* queue without spending an attempt. Returns the applied disposition,
|
||||
* or null for any other error (the caller fails the task).
|
||||
*/
|
||||
private requeueOnQuota(taskId: string, error: string): ProviderDisposition | null {
|
||||
const failure: ProviderFailure = { message: error, code: error };
|
||||
if (this.host.reportProviderFailure === undefined || !isQuotaFailure(failure)) return null;
|
||||
const task = this.host.getTask(taskId);
|
||||
if (task === undefined || task.status !== "running") return null;
|
||||
const disposition = this.host.reportProviderFailure(failure);
|
||||
const reason =
|
||||
disposition.action === "route-fallback"
|
||||
? `provider quota exhausted; re-queued on ${disposition.targetTier}`
|
||||
: disposition.action === "pause-missions"
|
||||
? `provider quota exhausted on every tier; re-queued until missions resume`
|
||||
: `provider quota error; re-queued`;
|
||||
this.host.setStageGates(taskId, [
|
||||
{ name: "precheck", status: "pass" },
|
||||
{ name: "worker", status: "pending" },
|
||||
{ name: "verify", status: "pending" },
|
||||
]);
|
||||
this.host.updateTask(taskId, { status: "pending", attempts: Math.max(0, task.attempts - 1), reason });
|
||||
this.host.publish("worker.requeued", { taskId, reason, action: disposition.action });
|
||||
this.host.audit("worker.requeued", { taskId, error: clip(error), action: disposition.action });
|
||||
return disposition;
|
||||
}
|
||||
|
||||
/** Commit worktree changes (never the isolated `.opencode` state) onto the task branch. */
|
||||
private commitWork(task: Task): boolean {
|
||||
const dir = task.worktree;
|
||||
@@ -344,6 +487,8 @@ export class WorkerLauncher {
|
||||
/** Abort every run, kill every worker, re-queue interrupted tasks. */
|
||||
async stop(): Promise<void> {
|
||||
this.stopping = true;
|
||||
if (this.resumeTimer !== null) clearTimeout(this.resumeTimer);
|
||||
this.resumeTimer = null;
|
||||
const runs = [...this.active.entries()];
|
||||
for (const [, run] of runs) run.abort.abort();
|
||||
await Promise.all(runs.map(async ([, run]) => run.conn?.close()));
|
||||
|
||||
+18
-3
@@ -13,11 +13,16 @@
|
||||
* `[crash]` exits the process, `[hang]` never finishes, `[noop]`
|
||||
* succeeds without touching files; otherwise the worker writes
|
||||
* `fleet-output/<session>.txt` in its cwd via a `write` tool call.
|
||||
* A session created on model id `quota-exhausted` fails every prompt
|
||||
* with a quota-shaped provider error (tier failover tests).
|
||||
* - Every prompt is recorded in `<dir of $OPENCODE_DB>/fake-prompts.jsonl`
|
||||
* with the session's agent, model, `XDG_CONFIG_HOME`, and text, so tests
|
||||
* can assert what a worker was started with.
|
||||
* - Director sessions (agent `director`) reply with a valid ```tasks
|
||||
* block when asked to "plan", an invalid one for "bad plan", and plain
|
||||
* text otherwise; every reply states the turn number.
|
||||
*/
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { appendFileSync, mkdirSync, writeFileSync } from "node:fs";
|
||||
import path from "node:path";
|
||||
|
||||
const password = process.env["OPENCODE_SERVER_PASSWORD"] ?? "";
|
||||
@@ -30,6 +35,7 @@ if (dbPath !== undefined) {
|
||||
interface FakeSession {
|
||||
id: string;
|
||||
agent: string | undefined;
|
||||
model: { providerID?: string; id?: string } | undefined;
|
||||
directory: string;
|
||||
turns: number;
|
||||
}
|
||||
@@ -82,6 +88,11 @@ async function runWorker(session: FakeSession, text: string): Promise<void> {
|
||||
process.exit(7);
|
||||
}
|
||||
if (text.includes("[hang]")) return;
|
||||
if (session.model?.id === "quota-exhausted") {
|
||||
await sleep(5);
|
||||
emit("session.execution.failed", { sessionID, error: { type: "provider", message: "GoUsageLimitError: usage limit reached" } });
|
||||
return;
|
||||
}
|
||||
if (text.includes("[fail]")) {
|
||||
await sleep(5);
|
||||
emit("session.execution.failed", { sessionID, error: { type: "provider", message: "model exploded" } });
|
||||
@@ -127,13 +138,13 @@ const server = Bun.serve({
|
||||
return new Response(stream, { headers: { "content-type": "text/event-stream" } });
|
||||
}
|
||||
if (req.method === "POST" && url.pathname === "/api/session") {
|
||||
const body = (await req.json()) as { location?: { directory?: string }; agent?: string };
|
||||
const body = (await req.json()) as { location?: { directory?: string }; agent?: string; model?: FakeSession["model"] };
|
||||
const directory = body.location?.directory;
|
||||
if (typeof directory !== "string" || !path.isAbsolute(directory)) {
|
||||
return Response.json({ error: "location.directory must be absolute" }, { status: 400 });
|
||||
}
|
||||
const id = `ses_${sessions.size + 1}_${process.pid}`;
|
||||
sessions.set(id, { id, agent: body.agent, directory, turns: 0 });
|
||||
sessions.set(id, { id, agent: body.agent, model: body.model, directory, turns: 0 });
|
||||
return Response.json({ data: { id, agent: body.agent ?? null, location: { directory } } });
|
||||
}
|
||||
const prompt = /^\/api\/session\/([^/]+)\/prompt$/.exec(url.pathname);
|
||||
@@ -146,6 +157,10 @@ const server = Bun.serve({
|
||||
}
|
||||
session.turns += 1;
|
||||
const text = body.text;
|
||||
if (dbPath !== undefined) {
|
||||
const record = { session: session.id, agent: session.agent ?? null, model: session.model ?? null, xdg: process.env["XDG_CONFIG_HOME"] ?? null, text };
|
||||
appendFileSync(path.join(path.dirname(dbPath), "fake-prompts.jsonl"), `${JSON.stringify(record)}\n`);
|
||||
}
|
||||
setTimeout(() => {
|
||||
emit("session.execution.started", { sessionID: session.id });
|
||||
emit("session.status", { sessionID: session.id, status: { type: "busy" } });
|
||||
|
||||
@@ -0,0 +1,329 @@
|
||||
/**
|
||||
* Task lifecycle wiring: approval lands the task branch (G-1), finished
|
||||
* tasks release their worktrees (G-2), `expert` tasks run with their role
|
||||
* pack (G-3), and quota failures move workers to the next model tier (G-4).
|
||||
* Workers are the fake OpenCode v2 server spawned as a real child process.
|
||||
*/
|
||||
import { afterEach, describe, expect, test } from "bun:test";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { existsSync, mkdirSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { parseTierModels } from "../src/cli-args.js";
|
||||
import { sealHumanKey, signApprovalPayload, unsealHumanKey, type ApprovalPayload } from "../src/human-key.js";
|
||||
import { loadRole, stripFrontmatter } from "../src/roles.js";
|
||||
import { createServer, type CreateServerOptions, type FleetServer } from "../src/server.js";
|
||||
import { parseModelRef, taskPrompt } from "../src/worker-launcher.js";
|
||||
import { TaskSchema } from "../src/model.js";
|
||||
|
||||
const FAKE = join(import.meta.dir, "fixtures", "fake-opencode.ts");
|
||||
const FAKE_COMMAND = [process.execPath, FAKE, "serve", "--port", "0", "--hostname", "127.0.0.1"];
|
||||
const PASSPHRASE = "lifecycle passphrase";
|
||||
|
||||
function git(cwd: string, args: string[]): string {
|
||||
return execFileSync("git", ["-c", "user.email=t@t", "-c", "user.name=t", ...args], { cwd, encoding: "utf8" }).trim();
|
||||
}
|
||||
|
||||
function scratchRepo(): string {
|
||||
const repo = mkdtempSync(join(tmpdir(), "lifecycle-repo-"));
|
||||
git(repo, ["init", "-q", "-b", "main"]);
|
||||
writeFileSync(join(repo, "README.md"), "# repo\n");
|
||||
git(repo, ["add", "."]);
|
||||
git(repo, ["commit", "-qm", "init"]);
|
||||
return repo;
|
||||
}
|
||||
|
||||
const opened: FleetServer[] = [];
|
||||
|
||||
function server(options: CreateServerOptions): FleetServer {
|
||||
const app = createServer({
|
||||
workers: { command: FAKE_COMMAND, startTimeoutMs: 15_000, taskTimeoutMs: 15_000 },
|
||||
director: false,
|
||||
...options,
|
||||
});
|
||||
opened.push(app);
|
||||
return app;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(opened.splice(0).map((app) => app.shutdown()));
|
||||
});
|
||||
|
||||
async function until(predicate: () => boolean, timeoutMs = 15_000): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (!predicate()) {
|
||||
if (Date.now() > deadline) throw new Error("condition not reached in time");
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
}
|
||||
|
||||
function post(app: FleetServer, path: string, body: unknown): Promise<Response> {
|
||||
return app.fetch(
|
||||
new Request(`http://127.0.0.1${path}`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json", "x-csrf-token": app.csrfToken },
|
||||
body: JSON.stringify(body),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async function approve(app: FleetServer, humanKey: Buffer, taskId: string): Promise<Response> {
|
||||
const minted = await post(app, `/api/approvals/${taskId}/ticket`, {});
|
||||
expect(minted.status).toBe(200);
|
||||
const ticket = (await minted.json()) as { ticket: string; payloads: { approve: ApprovalPayload } };
|
||||
return post(app, `/api/approvals/${taskId}/sign`, {
|
||||
ticket: ticket.ticket,
|
||||
signatureHex: signApprovalPayload(humanKey, ticket.payloads.approve),
|
||||
});
|
||||
}
|
||||
|
||||
function keyedState(): { stateDir: string; humanKey: Buffer } {
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
sealHumanKey(stateDir, PASSPHRASE);
|
||||
return { stateDir, humanKey: unsealHumanKey(stateDir, PASSPHRASE) };
|
||||
}
|
||||
|
||||
function readPrompts(file: string): Array<{ model: { providerID: string; id: string } | null; xdg: string | null; text: string }> {
|
||||
return readFileSync(file, "utf8")
|
||||
.trim()
|
||||
.split("\n")
|
||||
.map((line) => JSON.parse(line) as { model: { providerID: string; id: string } | null; xdg: string | null; text: string });
|
||||
}
|
||||
|
||||
describe("approval lands the task (G-1) and releases its worktree (G-2)", () => {
|
||||
test("APPROVE merges fleet/<task> into the mission integration branch and records landedSha", async () => {
|
||||
const repo = scratchRepo();
|
||||
const { stateDir, humanKey } = keyedState();
|
||||
const app = server({ repoDir: repo, stateDir });
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => app.getTask("task-1")?.status === "waitingHuman" && app.runningWorkers().length === 0);
|
||||
const worktree = app.getTask("task-1")!.worktree!;
|
||||
expect(existsSync(worktree)).toBe(true);
|
||||
|
||||
const res = await approve(app, humanKey, "task-1");
|
||||
expect(res.status).toBe(200);
|
||||
const body = (await res.json()) as { status: string; landedSha: string };
|
||||
expect(body.status).toBe("done");
|
||||
const task = app.getTask("task-1")!;
|
||||
expect(task.landedSha).toBe(body.landedSha);
|
||||
expect(git(repo, ["rev-parse", "fleet/integration/mission-1"])).toBe(body.landedSha);
|
||||
// A --no-ff merge commit whose second parent is the task branch tip.
|
||||
expect(git(repo, ["rev-parse", `${body.landedSha}^2`])).toBe(git(repo, ["rev-parse", "fleet/task-1"]));
|
||||
// Worktree released, branch kept, no landing worktree left behind.
|
||||
expect(existsSync(worktree)).toBe(false);
|
||||
expect(git(repo, ["worktree", "list", "--porcelain"])).not.toContain(worktree);
|
||||
expect(git(repo, ["worktree", "list", "--porcelain"])).not.toContain("landing-");
|
||||
expect(app.getBoard().worktrees.find((entry) => entry.taskId === "task-1")?.state).toBe("released");
|
||||
});
|
||||
|
||||
test("a landing conflict keeps the task waitingHuman with the conflicting files listed", async () => {
|
||||
const repo = scratchRepo();
|
||||
const { stateDir, humanKey } = keyedState();
|
||||
const app = server({ repoDir: repo, stateDir });
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => app.getTask("task-1")?.status === "waitingHuman" && app.runningWorkers().length === 0);
|
||||
const sessionId = app.getTask("task-1")!.sessionId!;
|
||||
// Someone else lands a different version of the same file first.
|
||||
git(repo, ["branch", "fleet/integration/mission-1", "main"]);
|
||||
git(repo, ["checkout", "-q", "fleet/integration/mission-1"]);
|
||||
mkdirSync(join(repo, "fleet-output"), { recursive: true });
|
||||
writeFileSync(join(repo, "fleet-output", `${sessionId}.txt`), "conflicting\n");
|
||||
git(repo, ["add", "."]);
|
||||
git(repo, ["commit", "-qm", "conflict"]);
|
||||
git(repo, ["checkout", "-q", "main"]);
|
||||
const tip = git(repo, ["rev-parse", "fleet/integration/mission-1"]);
|
||||
|
||||
const res = await approve(app, humanKey, "task-1");
|
||||
expect(res.status).toBe(409);
|
||||
const body = (await res.json()) as { code: string; files: string[] };
|
||||
expect(body.code).toBe("landing-failed");
|
||||
expect(body.files).toEqual([`fleet-output/${sessionId}.txt`]);
|
||||
const task = app.getTask("task-1")!;
|
||||
expect(task.status).toBe("waitingHuman");
|
||||
expect(task.reason).toContain(`fleet-output/${sessionId}.txt`);
|
||||
expect(task.landedSha).toBeUndefined();
|
||||
expect(git(repo, ["rev-parse", "fleet/integration/mission-1"])).toBe(tip);
|
||||
// Still waiting on a human: the worktree stays.
|
||||
expect(existsSync(task.worktree!)).toBe(true);
|
||||
});
|
||||
|
||||
test("REJECT releases the worktree and salvages the branch", async () => {
|
||||
const repo = scratchRepo();
|
||||
const { stateDir, humanKey } = keyedState();
|
||||
const app = server({ repoDir: repo, stateDir });
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => app.getTask("task-1")?.status === "waitingHuman" && app.runningWorkers().length === 0);
|
||||
const worktree = app.getTask("task-1")!.worktree!;
|
||||
const tip = git(repo, ["rev-parse", "fleet/task-1"]);
|
||||
const minted = (await (await post(app, "/api/approvals/task-1/ticket", {})).json()) as {
|
||||
ticket: string;
|
||||
payloads: { reject: ApprovalPayload };
|
||||
};
|
||||
const res = await post(app, "/api/approvals/task-1/sign", {
|
||||
ticket: minted.ticket,
|
||||
approve: false,
|
||||
signatureHex: signApprovalPayload(humanKey, minted.payloads.reject),
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
expect(app.getTask("task-1")?.status).toBe("failed");
|
||||
expect(existsSync(worktree)).toBe(false);
|
||||
expect(git(repo, ["rev-parse", "refs/fleet-salvage/task-1"])).toBe(tip);
|
||||
expect(git(repo, ["branch", "--list", "fleet/task-1"])).toBe("");
|
||||
});
|
||||
});
|
||||
|
||||
describe("terminal states release worktrees (G-2)", () => {
|
||||
test("no-change done and worker failure both release; failure salvages", async () => {
|
||||
const repo = scratchRepo();
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
const app = server({ repoDir: repo, stateDir });
|
||||
app.upsertTask({ id: "task-n", status: "pending", title: "nothing [noop]" });
|
||||
app.upsertTask({ id: "task-f", status: "pending", title: "explode [fail]" });
|
||||
await until(
|
||||
() => app.getTask("task-n")?.status === "done" && app.getTask("task-f")?.status === "failed" && app.runningWorkers().length === 0,
|
||||
);
|
||||
await until(() => !existsSync(join(stateDir, "worktrees", "task-n")) && !existsSync(join(stateDir, "worktrees", "task-f")));
|
||||
expect(git(repo, ["worktree", "list", "--porcelain"]).split("\n").filter((line) => line.startsWith("worktree "))).toHaveLength(1);
|
||||
expect(git(repo, ["rev-parse", "--verify", "--quiet", "refs/fleet-salvage/task-f"]).length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
test("restart releases worktrees of tasks that finished before the crash", async () => {
|
||||
const repo = scratchRepo();
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
const first = server({ repoDir: repo, stateDir });
|
||||
first.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => first.getTask("task-1")?.status === "waitingHuman" && first.runningWorkers().length === 0);
|
||||
const worktree = first.getTask("task-1")!.worktree!;
|
||||
await first.shutdown();
|
||||
// Simulate a finish recorded while the old process could not release.
|
||||
const offline = createServer({ stateDir, workers: false, director: false });
|
||||
offline.updateTask("task-1", { status: "cancelled", reason: "cancelled offline" });
|
||||
await offline.shutdown();
|
||||
expect(existsSync(worktree)).toBe(true);
|
||||
server({ repoDir: repo, stateDir, workers: false });
|
||||
expect(existsSync(worktree)).toBe(false);
|
||||
expect(git(repo, ["rev-parse", "--verify", "--quiet", "refs/fleet-salvage/task-1"]).length).toBeGreaterThan(0);
|
||||
});
|
||||
});
|
||||
|
||||
function rolePack(): string {
|
||||
const roles = mkdtempSync(join(tmpdir(), "lifecycle-roles-"));
|
||||
const base = join(roles, "dba-specialist", "opencode");
|
||||
mkdirSync(join(base, "agent"), { recursive: true });
|
||||
mkdirSync(join(base, "skill", "postgres-migrations"), { recursive: true });
|
||||
writeFileSync(
|
||||
join(base, "agent", "dba-specialist.md"),
|
||||
"---\ndescription: database specialist\nmode: primary\n---\n\nYou are the DBA specialist. Never drop tables.\n",
|
||||
);
|
||||
writeFileSync(join(base, "skill", "postgres-migrations", "SKILL.md"), "---\nname: postgres-migrations\n---\nMigrate safely.\n");
|
||||
writeFileSync(join(base, "opencode.json"), `${JSON.stringify({ $schema: "https://opencode.ai/config.json", share: "disabled" })}\n`);
|
||||
writeFileSync(join(roles, "dba-specialist", "role.json"), `${JSON.stringify({ maxConcurrent: 1 })}\n`);
|
||||
return roles;
|
||||
}
|
||||
|
||||
describe("role packs shape expert workers (G-3)", () => {
|
||||
test("loadRole reads prompt, skills, config, and cap; rejects non-slug ids", () => {
|
||||
const roles = rolePack();
|
||||
const role = loadRole(roles, "dba-specialist")!;
|
||||
expect(role.prompt).toBe("You are the DBA specialist. Never drop tables.");
|
||||
expect(role.skills).toEqual([join(roles, "dba-specialist", "opencode", "skill", "postgres-migrations", "SKILL.md")]);
|
||||
expect(role.configFile).toBe(join(roles, "dba-specialist", "opencode", "opencode.json"));
|
||||
expect(role.maxConcurrent).toBe(1);
|
||||
expect(loadRole(roles, "unknown-role")).toBeUndefined();
|
||||
expect(loadRole(roles, "../dba-specialist")).toBeUndefined();
|
||||
expect(stripFrontmatter("no frontmatter\n")).toBe("no frontmatter");
|
||||
});
|
||||
|
||||
test("taskPrompt injects the role prompt and skill paths", () => {
|
||||
const role = loadRole(rolePack(), "dba-specialist")!;
|
||||
const task = TaskSchema.parse({
|
||||
id: "task-9",
|
||||
missionId: "mission-1",
|
||||
kind: "expert-run",
|
||||
expert: "dba-specialist",
|
||||
input: { objective: "Add an index" },
|
||||
status: "running",
|
||||
});
|
||||
const prompt = taskPrompt(task, role);
|
||||
expect(prompt).toContain("Role: dba-specialist");
|
||||
expect(prompt).toContain("Never drop tables.");
|
||||
expect(prompt).toContain(role.skills[0]!);
|
||||
expect(prompt.indexOf("Never drop tables.")).toBeLessThan(prompt.indexOf("Objective: Add an index"));
|
||||
expect(taskPrompt(task)).not.toContain("Role:");
|
||||
});
|
||||
|
||||
test("expert task: XDG_CONFIG_HOME, copied opencode.json, role prompt, per-role cap", async () => {
|
||||
const repo = scratchRepo();
|
||||
const roles = rolePack();
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
const app = server({
|
||||
repoDir: repo,
|
||||
stateDir,
|
||||
workers: { command: FAKE_COMMAND, startTimeoutMs: 15_000, taskTimeoutMs: 15_000, rolesDir: roles },
|
||||
});
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "slow [hang]", expert: "dba-specialist" });
|
||||
app.upsertTask({ id: "task-2", status: "pending", title: "second", expert: "dba-specialist" });
|
||||
await until(() => app.getTask("task-1")?.status === "running" && app.getTask("task-1")?.sessionId?.startsWith("ses_") === true);
|
||||
// role.json maxConcurrent = 1: the second dba task waits.
|
||||
expect(app.getTask("task-2")?.status).toBe("ready");
|
||||
const dir = app.getTask("task-1")!.worktree!;
|
||||
const copied = join(dir, ".opencode", "opencode.json");
|
||||
expect(readFileSync(copied, "utf8")).toContain('"share":"disabled"');
|
||||
const [prompt] = readPrompts(join(dir, ".opencode", "fake-prompts.jsonl"));
|
||||
expect(prompt?.xdg).toBe(join(roles, "dba-specialist"));
|
||||
expect(prompt?.text).toContain("You are the DBA specialist.");
|
||||
});
|
||||
});
|
||||
|
||||
describe("tier failover reaches the workers (G-4)", () => {
|
||||
test("a quota failure re-queues the task on the next tier's model without spending an attempt", async () => {
|
||||
const repo = scratchRepo();
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
const app = server({
|
||||
repoDir: repo,
|
||||
stateDir,
|
||||
workers: {
|
||||
command: FAKE_COMMAND,
|
||||
startTimeoutMs: 15_000,
|
||||
taskTimeoutMs: 15_000,
|
||||
tierModels: { "tier-1-opencode": "opencode/quota-exhausted", "tier-2-commercial": "bifrost/backup-model" },
|
||||
},
|
||||
});
|
||||
const seen: string[] = [];
|
||||
app.bus.subscribe(1, (event) => seen.push(event.type));
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => app.getTask("task-1")?.status === "waitingHuman" && app.runningWorkers().length === 0);
|
||||
expect(app.getBoard().activeTier).toBe("tier-2-commercial");
|
||||
expect(app.getTask("task-1")?.attempts).toBe(1);
|
||||
expect(seen).toContain("tier.fallback");
|
||||
expect(seen).toContain("worker.requeued");
|
||||
const prompts = readPrompts(join(app.getTask("task-1")!.worktree!, ".opencode", "fake-prompts.jsonl"));
|
||||
expect(prompts.map((entry) => entry.model?.id)).toEqual(["quota-exhausted", "backup-model"]);
|
||||
expect(prompts[1]?.model?.providerID).toBe("bifrost");
|
||||
});
|
||||
|
||||
test("quota on the last tier pauses launches instead of failing tasks", async () => {
|
||||
const repo = scratchRepo();
|
||||
const stateDir = mkdtempSync(join(tmpdir(), "lifecycle-state-"));
|
||||
const app = server({ repoDir: repo, stateDir, workers: { command: FAKE_COMMAND, startTimeoutMs: 15_000, taskTimeoutMs: 15_000 } });
|
||||
app.setTier("tier-3-local");
|
||||
const disposition = app.reportProviderFailure({ status: 402, code: "insufficient_quota", headers: { "retry-after": "60" } });
|
||||
expect(disposition.action).toBe("pause-missions");
|
||||
app.upsertTask({ id: "task-1", status: "pending", title: "Write output" });
|
||||
await until(() => app.getTask("task-1")?.status === "ready");
|
||||
await new Promise((resolve) => setTimeout(resolve, 200));
|
||||
expect(app.getTask("task-1")?.status).toBe("ready");
|
||||
expect(app.runningWorkers()).toEqual([]);
|
||||
expect(app.getTask("task-1")?.reason).toContain("missions paused");
|
||||
});
|
||||
|
||||
test("model references and --tier-models parse strictly", () => {
|
||||
expect(parseModelRef("bifrost/gpt-5#high")).toEqual({ providerID: "bifrost", id: "gpt-5", variant: "high" });
|
||||
expect(parseModelRef("opencode/glm-4.6")).toEqual({ providerID: "opencode", id: "glm-4.6" });
|
||||
expect(() => parseModelRef("no-provider")).toThrow("invalid model reference");
|
||||
expect(parseTierModels('{"tier-3-local":"ollama/qwen3"}')).toEqual({ "tier-3-local": "ollama/qwen3" });
|
||||
expect(() => parseTierModels('{"tier-9":"a/b"}')).toThrow("unknown tier");
|
||||
expect(() => parseTierModels('{"tier-1-opencode":"nomodel"}')).toThrow("provider/model");
|
||||
expect(parseTierModels(undefined)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -34,7 +34,7 @@ if grep -n -E 'record\["publicKey|passphraseToKeypair|verifyApprovalSignature' d
|
||||
fi
|
||||
|
||||
echo "[wave-4] approval attacks, es-core key compatibility, launcher, Director, persistence, NATS"
|
||||
bun test dispatcher/test/approvals.test.ts dispatcher/test/human-key.test.ts dispatcher/test/launcher.test.ts dispatcher/test/nats-bridge.test.ts
|
||||
bun test dispatcher/test/approvals.test.ts dispatcher/test/human-key.test.ts dispatcher/test/launcher.test.ts dispatcher/test/nats-bridge.test.ts dispatcher/test/lifecycle.test.ts
|
||||
|
||||
echo "[wave-4] browser build"
|
||||
bun run build:web
|
||||
|
||||
+6
-6
@@ -8,14 +8,14 @@
|
||||
"test": "bun test"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/bun": "latest",
|
||||
"@types/react": "^18.3.3",
|
||||
"@types/react-dom": "^18.3.0",
|
||||
"typescript": "^5.5.4"
|
||||
"@types/bun": "1.4.3",
|
||||
"@types/react": "18.3.31",
|
||||
"@types/react-dom": "18.3.7",
|
||||
"typescript": "5.9.3"
|
||||
},
|
||||
"dependencies": {
|
||||
"@noble/hashes": "1.8.0",
|
||||
"react": "^18.3.1",
|
||||
"react-dom": "^18.3.1"
|
||||
"react": "18.3.1",
|
||||
"react-dom": "18.3.1"
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user