From 8b01c1b8dd0fcfef0bffadb08e475b697fd1db62 Mon Sep 17 00:00:00 2001 From: Randall Stillwell Date: Tue, 7 Jul 2026 15:10:59 -0500 Subject: [PATCH] feat(orchestrator): convoy driver tick loop and Redis job queue (Epic 6) Adds apps/orchestrator with planOrchestratorAction in shared, dry-run mode, Redis role_dispatch enqueue, gate pauses, and stall detection logging. Co-authored-by: Cursor --- .env.example | 6 + apps/orchestrator/package.json | 28 +++ apps/orchestrator/src/db/convoy.ts | 162 ++++++++++++++++++ apps/orchestrator/src/env.ts | 42 +++++ apps/orchestrator/src/index.ts | 49 ++++++ apps/orchestrator/src/queue/redis-queue.ts | 31 ++++ apps/orchestrator/src/tick.ts | 122 +++++++++++++ apps/orchestrator/tsconfig.json | 9 + apps/orchestrator/tsup.config.ts | 11 ++ apps/orchestrator/vitest.config.ts | 7 + docs/orchestrator.md | 30 ++++ packages/shared/package.json | 1 + packages/shared/src/index.ts | 1 + .../src/orchestrator/convoy-driver.test.ts | 82 +++++++++ .../shared/src/orchestrator/convoy-driver.ts | 147 ++++++++++++++++ packages/shared/src/orchestrator/index.ts | 13 ++ packages/shared/src/orchestrator/job.ts | 19 ++ .../Epic-apps-orchestrator.md | 7 +- .../Task-apps-orchestrator-service.md | 23 +++ .../Task-shared-orchestrator-driver.md | 23 +++ .../Plan-automation-foundation.md | 2 +- pnpm-lock.yaml | 50 +++++- 22 files changed, 853 insertions(+), 12 deletions(-) create mode 100644 apps/orchestrator/package.json create mode 100644 apps/orchestrator/src/db/convoy.ts create mode 100644 apps/orchestrator/src/env.ts create mode 100644 apps/orchestrator/src/index.ts create mode 100644 apps/orchestrator/src/queue/redis-queue.ts create mode 100644 apps/orchestrator/src/tick.ts create mode 100644 apps/orchestrator/tsconfig.json create mode 100644 apps/orchestrator/tsup.config.ts create mode 100644 apps/orchestrator/vitest.config.ts create mode 100644 docs/orchestrator.md create mode 100644 packages/shared/src/orchestrator/convoy-driver.test.ts create mode 100644 packages/shared/src/orchestrator/convoy-driver.ts create mode 100644 packages/shared/src/orchestrator/index.ts create mode 100644 packages/shared/src/orchestrator/job.ts create mode 100644 plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-apps-orchestrator-service.md create mode 100644 plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-shared-orchestrator-driver.md diff --git a/.env.example b/.env.example index bfce677..ed6ce0e 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,12 @@ MARKDOWN_BACKLOG_WORKSPACE_ID="00000000-0000-0000-0000-000000000000" # Local dev: redis://localhost:6379 REDIS_URL="redis://:CHANGE_ME@192.168.68.102:6379/11" +# ----- Orchestrator (apps/orchestrator) ----- +# ORCHESTRATOR_DRY_RUN=1 logs dispatch/transition without Redis enqueue +# ORCHESTRATOR_POLL_MS=5000 +# ORCHESTRATOR_STALL_MINUTES=120 +# ORCHESTRATOR_MAX_IMPLEMENTERS=2 + # ----- Auth (NextAuth v5) ----- # Local: http://localhost:3000 • Prod: https://echodo.stillwell.cloud NEXTAUTH_URL="http://localhost:3000" diff --git a/apps/orchestrator/package.json b/apps/orchestrator/package.json new file mode 100644 index 0000000..db6d573 --- /dev/null +++ b/apps/orchestrator/package.json @@ -0,0 +1,28 @@ +{ + "name": "@tasks/orchestrator", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "dev": "tsx watch src/index.ts", + "start": "node dist/index.mjs", + "build": "tsup", + "type-check": "tsc --noEmit", + "test": "vitest run --passWithNoTests", + "test:watch": "vitest" + }, + "dependencies": { + "@tasks/database": "workspace:*", + "@tasks/shared": "workspace:*", + "drizzle-orm": "^0.38.0", + "ioredis": "^5.4.2", + "zod": "^3.24.0" + }, + "devDependencies": { + "@types/node": "^22.10.0", + "tsup": "^8.3.5", + "tsx": "^4.19.2", + "typescript": "^5.7.0", + "vitest": "^2.1.9" + } +} diff --git a/apps/orchestrator/src/db/convoy.ts b/apps/orchestrator/src/db/convoy.ts new file mode 100644 index 0000000..8d5c2ea --- /dev/null +++ b/apps/orchestrator/src/db/convoy.ts @@ -0,0 +1,162 @@ +import { and, desc, eq, inArray, isNull, lt, sql } from "drizzle-orm"; + +import type { db as dbInstance } from "@tasks/database"; +import { agentRuns, convoyEvents, objects } from "@tasks/database/schema"; +import { + briefObjectContentSchema, + convoyObjectContentSchema, + validateConvoyTransition, + type ConvoyStatus, +} from "@tasks/shared/convoy-lifecycle"; +import type { BriefSummary } from "@tasks/shared/orchestrator"; + +type Db = typeof dbInstance; + +export type LoadedConvoy = { + id: string; + workspaceId: string; + title: string; + status: ConvoyStatus; + skipFlags: string[]; + briefs: BriefSummary[]; + completedRoles: Set; + completedBriefIds: Set; +}; + +function parseStatus(raw: string | null): ConvoyStatus { + const status = raw ?? "draft"; + return status as ConvoyStatus; +} + +export async function listActiveConvoys(db: Db): Promise { + const rows = await db + .select() + .from(objects) + .where( + and( + eq(objects.type, "project"), + sql`${objects.content} ->> 'convoySlug' IS NOT NULL`, + inArray(objects.status, ["draft", "ready", "in_progress"]), + ), + ); + + const loaded: LoadedConvoy[] = []; + for (const row of rows) { + const content = convoyObjectContentSchema.safeParse(row.content); + if (!content.success) continue; + + const briefRows = await db + .select() + .from(objects) + .where(and(eq(objects.parentId, row.id), eq(objects.workspaceId, row.workspaceId))); + + const events = await db + .select({ role: convoyEvents.role, eventType: convoyEvents.eventType, briefId: convoyEvents.briefId }) + .from(convoyEvents) + .where(eq(convoyEvents.convoyId, row.id)) + .orderBy(desc(convoyEvents.occurredAt)) + .limit(200); + + const completedRoles = new Set(); + const completedBriefIds = new Set(); + for (const event of events) { + if (event.eventType === "role_complete") { + completedRoles.add(event.role); + if (event.briefId) completedBriefIds.add(event.briefId); + } + } + + const briefs: BriefSummary[] = briefRows.map((brief) => { + const parsed = briefObjectContentSchema.safeParse(brief.content); + return { + id: brief.id, + briefNumber: parsed.success ? parsed.data.briefNumber : brief.sortOrder, + status: brief.status, + sliceDependencies: parsed.success ? parsed.data.sliceDependencies : [], + }; + }); + + loaded.push({ + id: row.id, + workspaceId: row.workspaceId, + title: row.title, + status: parseStatus(row.status), + skipFlags: content.data.skipFlags, + briefs, + completedRoles, + completedBriefIds, + }); + } + + return loaded; +} + +export async function transitionConvoyInDb( + db: Db, + input: { + workspaceId: string; + convoyId: string; + toStatus: ConvoyStatus; + actor: string; + reason?: string; + }, +) { + const [convoy] = await db + .select() + .from(objects) + .where( + and( + eq(objects.id, input.convoyId), + eq(objects.workspaceId, input.workspaceId), + eq(objects.type, "project"), + ), + ) + .limit(1); + + if (!convoy) { + throw new Error(`Convoy ${input.convoyId} not found`); + } + + const fromStatus = parseStatus(convoy.status); + validateConvoyTransition({ + from: fromStatus, + to: input.toStatus, + actor: input.actor, + reason: input.reason, + }); + + await db + .update(objects) + .set({ status: input.toStatus, updatedAt: new Date() }) + .where(eq(objects.id, convoy.id)); + + await db.insert(convoyEvents).values({ + workspaceId: input.workspaceId, + convoyId: convoy.id, + role: input.actor, + actor: input.actor, + eventType: "status_change", + metadata: { + from: fromStatus, + to: input.toStatus, + reason: input.reason ?? null, + source: "orchestrator", + }, + }); +} + +export async function detectStalledConvoys(db: Db, stallMinutes: number): Promise { + const cutoff = new Date(Date.now() - stallMinutes * 60_000); + const openRuns = await db + .select({ + backlogItemId: agentRuns.backlogItemId, + startedAt: agentRuns.startedAt, + }) + .from(agentRuns) + .where(and(isNull(agentRuns.finishedAt), lt(agentRuns.startedAt, cutoff))) + .limit(50); + + if (openRuns.length === 0) return []; + + return openRuns.map((run) => run.backlogItemId); +} diff --git a/apps/orchestrator/src/env.ts b/apps/orchestrator/src/env.ts new file mode 100644 index 0000000..afd5d70 --- /dev/null +++ b/apps/orchestrator/src/env.ts @@ -0,0 +1,42 @@ +import { z } from "zod"; + +const envSchema = z.object({ + DATABASE_URL: z.string().min(1), + REDIS_URL: z.string().min(1).optional(), + ORCHESTRATOR_DRY_RUN: z + .enum(["0", "1", "true", "false", ""]) + .optional() + .transform((value) => value === "1" || value === "true"), + ORCHESTRATOR_POLL_MS: z + .string() + .optional() + .transform((value) => { + const parsed = Number(value ?? "5000"); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 5_000; + }), + ORCHESTRATOR_STALL_MINUTES: z + .string() + .optional() + .transform((value) => { + const parsed = Number(value ?? "120"); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 120; + }), + ORCHESTRATOR_MAX_IMPLEMENTERS: z + .string() + .optional() + .transform((value) => { + const parsed = Number(value ?? "2"); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 2; + }), +}); + +export type OrchestratorEnv = z.infer; + +export function loadEnv(): OrchestratorEnv { + const parsed = envSchema.safeParse(process.env); + if (!parsed.success) { + console.error("[orchestrator] Invalid environment:", parsed.error.flatten().fieldErrors); + process.exit(1); + } + return parsed.data; +} diff --git a/apps/orchestrator/src/index.ts b/apps/orchestrator/src/index.ts new file mode 100644 index 0000000..9b3d371 --- /dev/null +++ b/apps/orchestrator/src/index.ts @@ -0,0 +1,49 @@ +import { db } from "@tasks/database"; +import IoRedis from "ioredis"; + +import { detectStalledConvoys } from "./db/convoy.js"; +import { loadEnv } from "./env.js"; +import { runOrchestratorTick } from "./tick.js"; + +async function main(): Promise { + const env = loadEnv(); + const redis = env.REDIS_URL ? new IoRedis(env.REDIS_URL) : null; + + if (env.ORCHESTRATOR_DRY_RUN) { + console.info("[orchestrator] DRY-RUN mode — jobs will be logged, not enqueued"); + } else if (!redis) { + console.warn("[orchestrator] REDIS_URL not set — running in DRY-RUN mode"); + } + + console.info(`[orchestrator] polling every ${env.ORCHESTRATOR_POLL_MS}ms`); + + const loop = async () => { + try { + const results = await runOrchestratorTick(db, redis, { + ...env, + ORCHESTRATOR_DRY_RUN: env.ORCHESTRATOR_DRY_RUN || !redis, + }); + + const stalled = await detectStalledConvoys(db, env.ORCHESTRATOR_STALL_MINUTES); + if (stalled.length > 0) { + console.warn( + `[orchestrator] stall detector: ${stalled.length} open agent_runs older than ${env.ORCHESTRATOR_STALL_MINUTES}m`, + ); + } + + if (results.length === 0) { + console.info("[orchestrator] no active convoys"); + } + } catch (error) { + console.error("[orchestrator] tick failed:", error); + } + }; + + await loop(); + setInterval(loop, env.ORCHESTRATOR_POLL_MS); +} + +main().catch((error) => { + console.error("[orchestrator] fatal:", error); + process.exit(1); +}); diff --git a/apps/orchestrator/src/queue/redis-queue.ts b/apps/orchestrator/src/queue/redis-queue.ts new file mode 100644 index 0000000..1e6d771 --- /dev/null +++ b/apps/orchestrator/src/queue/redis-queue.ts @@ -0,0 +1,31 @@ +import { randomUUID } from "node:crypto"; + +import type IoRedis from "ioredis"; + +import { + ORCHESTRATOR_QUEUE_KEY, + orchestratorJobSchema, + type OrchestratorJob, +} from "@tasks/shared/orchestrator"; + +export async function enqueueJob(redis: IoRedis, job: Omit) { + const payload = orchestratorJobSchema.parse({ + ...job, + id: randomUUID(), + enqueuedAt: new Date().toISOString(), + }); + await redis.rpush(ORCHESTRATOR_QUEUE_KEY, JSON.stringify(payload)); + return payload; +} + +export async function dequeueJob(redis: IoRedis, timeoutSeconds = 5): Promise { + const result = await redis.blpop(ORCHESTRATOR_QUEUE_KEY, timeoutSeconds); + if (!result) return null; + const [, raw] = result; + try { + return orchestratorJobSchema.parse(JSON.parse(raw)); + } catch { + console.error("[orchestrator] Dropping invalid job payload:", raw); + return null; + } +} diff --git a/apps/orchestrator/src/tick.ts b/apps/orchestrator/src/tick.ts new file mode 100644 index 0000000..d4bdf13 --- /dev/null +++ b/apps/orchestrator/src/tick.ts @@ -0,0 +1,122 @@ +import type IoRedis from "ioredis"; + +import type { db as dbInstance } from "@tasks/database"; +import { + planOrchestratorAction, + type OrchestratorAction, + type OrchestratorJob, +} from "@tasks/shared/orchestrator"; + +import { listActiveConvoys, transitionConvoyInDb, type LoadedConvoy } from "./db/convoy.js"; +import { enqueueJob } from "./queue/redis-queue.js"; +import type { OrchestratorEnv } from "./env.js"; + +type Db = typeof dbInstance; + +export type TickResult = { + convoyId: string; + action: OrchestratorAction; + job?: OrchestratorJob; +}; + +async function applyAction( + db: Db, + redis: IoRedis | null, + env: OrchestratorEnv, + convoy: LoadedConvoy, + action: OrchestratorAction, +): Promise { + const base = { convoyId: convoy.id, action }; + + if (action.type === "noop" || action.type === "pause") { + console.info( + `[orchestrator] convoy=${convoy.title} status=${convoy.status} → ${action.type}: ${ + action.type === "pause" ? action.reason : action.reason + }`, + ); + return base; + } + + if (action.type === "transition") { + if (env.ORCHESTRATOR_DRY_RUN) { + console.info( + `[orchestrator] DRY-RUN transition convoy=${convoy.title} ${convoy.status} → ${action.toStatus} (actor=${action.actor})`, + ); + return base; + } + + await transitionConvoyInDb(db, { + workspaceId: convoy.workspaceId, + convoyId: convoy.id, + toStatus: action.toStatus, + actor: action.actor, + }); + console.info( + `[orchestrator] transition convoy=${convoy.title} → ${action.toStatus} (actor=${action.actor})`, + ); + return base; + } + + if (action.type === "dispatch") { + if ( + action.roleId === "role-implementer" && + convoy.status === "ready" && + !env.ORCHESTRATOR_DRY_RUN + ) { + await transitionConvoyInDb(db, { + workspaceId: convoy.workspaceId, + convoyId: convoy.id, + toStatus: "in_progress", + actor: "role-implementer", + }); + convoy.status = "in_progress"; + } + + if (env.ORCHESTRATOR_DRY_RUN || !redis) { + console.info( + `[orchestrator] DRY-RUN dispatch convoy=${convoy.title} role=${action.roleId}${ + action.briefId ? ` brief=${action.briefId}` : "" + }`, + ); + return base; + } + + const job = await enqueueJob(redis, { + type: "role_dispatch", + workspaceId: convoy.workspaceId, + convoyId: convoy.id, + roleId: action.roleId, + briefId: action.briefId, + }); + console.info( + `[orchestrator] enqueued job=${job.id} convoy=${convoy.title} role=${action.roleId}`, + ); + return { ...base, job }; + } + + return base; +} + +export async function runOrchestratorTick( + db: Db, + redis: IoRedis | null, + env: OrchestratorEnv, +): Promise { + const convoys = await listActiveConvoys(db); + const results: TickResult[] = []; + + for (const convoy of convoys) { + const action = planOrchestratorAction({ + convoyId: convoy.id, + status: convoy.status, + skipFlags: convoy.skipFlags, + briefs: convoy.briefs, + completedRoles: convoy.completedRoles, + completedBriefIds: convoy.completedBriefIds, + }); + + results.push(await applyAction(db, redis, env, convoy, action)); + } + + return results; +} diff --git a/apps/orchestrator/tsconfig.json b/apps/orchestrator/tsconfig.json new file mode 100644 index 0000000..486abdd --- /dev/null +++ b/apps/orchestrator/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "noEmit": true + }, + "include": ["src/**/*"] +} diff --git a/apps/orchestrator/tsup.config.ts b/apps/orchestrator/tsup.config.ts new file mode 100644 index 0000000..9a20f39 --- /dev/null +++ b/apps/orchestrator/tsup.config.ts @@ -0,0 +1,11 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: ["src/index.ts"], + format: ["esm"], + dts: true, + noExternal: [/^@tasks\//], + target: "node20", + platform: "node", + clean: true, +}); diff --git a/apps/orchestrator/vitest.config.ts b/apps/orchestrator/vitest.config.ts new file mode 100644 index 0000000..f624398 --- /dev/null +++ b/apps/orchestrator/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + environment: "node", + }, +}); diff --git a/docs/orchestrator.md b/docs/orchestrator.md new file mode 100644 index 0000000..5cafd4a --- /dev/null +++ b/docs/orchestrator.md @@ -0,0 +1,30 @@ +# Orchestrator (`apps/orchestrator`) + +Long-running service that drives convoy lifecycle: plans next pipeline role, enqueues worker jobs on Redis, pauses at human gates, and detects stalled agent runs. + +## Run locally + +```bash +# Dry-run (no Redis required) +ORCHESTRATOR_DRY_RUN=1 pnpm --filter @tasks/orchestrator dev + +# Live enqueue (requires REDIS_URL) +pnpm --filter @tasks/orchestrator dev +``` + +## Architecture + +- **Driver** — pure `planOrchestratorAction()` in `@tasks/shared/orchestrator` +- **Tick loop** — loads active convoys (`draft`, `ready`, `in_progress`), applies next action +- **Queue** — `echodo:orchestrator:jobs` Redis list (consumed by `apps/worker`, Epic 7) +- **Gates** — `convoy_review`, `pr_review`, `merged` pause until mission control approves + +## Env + +| Variable | Default | Purpose | +|----------|---------|---------| +| `DATABASE_URL` | required | Convoy + event reads/writes | +| `REDIS_URL` | optional | Job queue; missing ⇒ dry-run | +| `ORCHESTRATOR_DRY_RUN` | `false` | Log actions without enqueue | +| `ORCHESTRATOR_POLL_MS` | `5000` | Tick interval | +| `ORCHESTRATOR_STALL_MINUTES` | `120` | Stall detector threshold | diff --git a/packages/shared/package.json b/packages/shared/package.json index 62ffe87..4abaa09 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -9,6 +9,7 @@ ".": "./src/index.ts", "./connected-repos": "./src/connected-repos/index.ts", "./convoy-lifecycle": "./src/convoy-lifecycle/index.ts", + "./orchestrator": "./src/orchestrator/index.ts", "./model-routing": "./src/model-routing/index.ts", "./types": "./src/types/index.ts", "./utils": "./src/utils/index.ts" diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index 565af3f..dbd445e 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -1,5 +1,6 @@ export * from "./connected-repos"; export * from "./convoy-lifecycle"; export * from "./model-routing"; +export * from "./orchestrator"; export * from "./types"; export * from "./utils"; diff --git a/packages/shared/src/orchestrator/convoy-driver.test.ts b/packages/shared/src/orchestrator/convoy-driver.test.ts new file mode 100644 index 0000000..8ffc50d --- /dev/null +++ b/packages/shared/src/orchestrator/convoy-driver.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, it } from "vitest"; + +import { planOrchestratorAction } from "./convoy-driver"; + +describe("planOrchestratorAction", () => { + const convoyId = "00000000-0000-4000-8000-000000000001"; + + it("pauses at convoy_review gate", () => { + const action = planOrchestratorAction({ + convoyId, + status: "convoy_review", + skipFlags: [], + briefs: [], + completedRoles: new Set(), + completedBriefIds: new Set(), + }); + expect(action).toMatchObject({ type: "pause", gate: "1" }); + }); + + it("dispatches first planning role on draft", () => { + const action = planOrchestratorAction({ + convoyId, + status: "draft", + skipFlags: [], + briefs: [], + completedRoles: new Set(), + completedBriefIds: new Set(), + }); + expect(action).toMatchObject({ type: "dispatch", roleId: "role-ia-architect" }); + }); + + it("skips ia when in skip flags", () => { + const action = planOrchestratorAction({ + convoyId, + status: "draft", + skipFlags: ["ia"], + briefs: [], + completedRoles: new Set(), + completedBriefIds: new Set(), + }); + expect(action).toMatchObject({ type: "dispatch", roleId: "role-ux-reviewer" }); + }); + + it("transitions to convoy_review when planning roles complete", () => { + const action = planOrchestratorAction({ + convoyId, + status: "draft", + skipFlags: ["ia", "ux"], + briefs: [], + completedRoles: new Set(["role-architect"]), + completedBriefIds: new Set(), + }); + expect(action).toMatchObject({ + type: "transition", + toStatus: "convoy_review", + actor: "role-architect", + }); + }); + + it("dispatches implementer for ready brief", () => { + const action = planOrchestratorAction({ + convoyId, + status: "ready", + skipFlags: [], + briefs: [ + { + id: "00000000-0000-4000-8000-000000000010", + briefNumber: 1, + status: "ready", + sliceDependencies: [], + }, + ], + completedRoles: new Set(), + completedBriefIds: new Set(), + }); + expect(action).toMatchObject({ + type: "dispatch", + roleId: "role-implementer", + briefId: "00000000-0000-4000-8000-000000000010", + }); + }); +}); diff --git a/packages/shared/src/orchestrator/convoy-driver.ts b/packages/shared/src/orchestrator/convoy-driver.ts new file mode 100644 index 0000000..65b32de --- /dev/null +++ b/packages/shared/src/orchestrator/convoy-driver.ts @@ -0,0 +1,147 @@ +import type { ConvoyStatus } from "../convoy-lifecycle/schemas"; + +export type BriefSummary = { + id: string; + briefNumber: number; + status: string | null; + sliceDependencies: number[]; +}; + +export type OrchestratorPauseAction = { + type: "pause"; + reason: string; + gate: "1" | "2" | "3" | null; +}; + +export type OrchestratorDispatchAction = { + type: "dispatch"; + roleId: string; + convoyId: string; + briefId?: string; +}; + +export type OrchestratorTransitionAction = { + type: "transition"; + actor: string; + toStatus: ConvoyStatus; +}; + +export type OrchestratorNoopAction = { + type: "noop"; + reason: string; +}; + +export type OrchestratorAction = + | OrchestratorPauseAction + | OrchestratorDispatchAction + | OrchestratorTransitionAction + | OrchestratorNoopAction; + +const PLANNING_PIPELINE: Array<{ skipFlag: string; roleId: string }> = [ + { skipFlag: "ia", roleId: "role-ia-architect" }, + { skipFlag: "ux", roleId: "role-ux-reviewer" }, + { skipFlag: "arch", roleId: "role-architect" }, +]; + +const GATE_STATUSES: Record = { + draft: null, + convoy_review: "1", + ready: null, + in_progress: null, + pr_review: "2", + merged: "3", + released: null, + blocked: null, + cancelled: null, +}; + +function isBriefComplete(brief: BriefSummary): boolean { + const status = brief.status ?? ""; + return status === "done" || status === "completed" || status === "merged"; +} + +function briefsReadyToDispatch(briefs: BriefSummary[], completedBriefIds: Set): BriefSummary[] { + return briefs.filter((brief) => { + if (completedBriefIds.has(brief.id)) return false; + if (isBriefComplete(brief)) return false; + const deps = brief.sliceDependencies; + if (deps.length === 0) return true; + return deps.every((depNumber) => + briefs.some((candidate) => candidate.briefNumber === depNumber && isBriefComplete(candidate)), + ); + }); +} + +/** + * Pure convoy driver — given current status and completed roles/briefs, returns the next orchestrator action. + */ +export function planOrchestratorAction(input: { + convoyId: string; + status: ConvoyStatus; + skipFlags: string[]; + briefs: BriefSummary[]; + completedRoles: Set; + completedBriefIds: Set; +}): OrchestratorAction { + const gate = GATE_STATUSES[input.status]; + if (gate) { + return { + type: "pause", + reason: `Awaiting human gate ${gate}`, + gate, + }; + } + + if (input.status === "cancelled" || input.status === "released" || input.status === "blocked") { + return { type: "noop", reason: `Terminal or blocked status: ${input.status}` }; + } + + if (input.status === "draft") { + for (const step of PLANNING_PIPELINE) { + if (input.skipFlags.includes(step.skipFlag)) continue; + if (!input.completedRoles.has(step.roleId)) { + return { + type: "dispatch", + roleId: step.roleId, + convoyId: input.convoyId, + }; + } + } + return { + type: "transition", + actor: "role-architect", + toStatus: "convoy_review", + }; + } + + if (input.status === "ready" || input.status === "in_progress") { + const eligible = briefsReadyToDispatch(input.briefs, input.completedBriefIds); + if (eligible.length > 0) { + const nextBrief = eligible.sort((a, b) => a.briefNumber - b.briefNumber)[0]!; + return { + type: "dispatch", + roleId: "role-implementer", + convoyId: input.convoyId, + briefId: nextBrief.id, + }; + } + + const allBriefsDone = + input.briefs.length > 0 && input.briefs.every((brief) => isBriefComplete(brief)); + if (allBriefsDone && input.status === "in_progress") { + return { + type: "transition", + actor: "role-implementer", + toStatus: "pr_review", + }; + } + + if (input.briefs.length === 0) { + return { type: "noop", reason: "No implementer briefs defined yet" }; + } + + return { type: "noop", reason: "Waiting for implementer briefs to complete" }; + } + + return { type: "noop", reason: `No driver rule for status ${input.status}` }; +} diff --git a/packages/shared/src/orchestrator/index.ts b/packages/shared/src/orchestrator/index.ts new file mode 100644 index 0000000..621c14a --- /dev/null +++ b/packages/shared/src/orchestrator/index.ts @@ -0,0 +1,13 @@ +export { + ORCHESTRATOR_QUEUE_KEY, + orchestratorJobSchema, + orchestratorJobTypeSchema, + type OrchestratorJob, + type OrchestratorJobType, +} from "./job"; + +export { + planOrchestratorAction, + type BriefSummary, + type OrchestratorAction, +} from "./convoy-driver"; diff --git a/packages/shared/src/orchestrator/job.ts b/packages/shared/src/orchestrator/job.ts new file mode 100644 index 0000000..911b327 --- /dev/null +++ b/packages/shared/src/orchestrator/job.ts @@ -0,0 +1,19 @@ +import { z } from "zod"; + +export const orchestratorJobTypeSchema = z.enum(["role_dispatch", "stall_check"]); + +export type OrchestratorJobType = z.infer; + +export const orchestratorJobSchema = z.object({ + id: z.string().uuid(), + type: orchestratorJobTypeSchema, + workspaceId: z.string().uuid(), + convoyId: z.string().uuid(), + roleId: z.string().min(1).max(64), + briefId: z.string().uuid().optional(), + enqueuedAt: z.string().datetime(), +}); + +export type OrchestratorJob = z.infer; + +export const ORCHESTRATOR_QUEUE_KEY = "echodo:orchestrator:jobs"; diff --git a/plans/Plan-automation-foundation/Epic-apps-orchestrator/Epic-apps-orchestrator.md b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Epic-apps-orchestrator.md index 8f1b64e..61e9638 100644 --- a/plans/Plan-automation-foundation/Epic-apps-orchestrator/Epic-apps-orchestrator.md +++ b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Epic-apps-orchestrator.md @@ -3,7 +3,7 @@ kind: epic slug: apps-orchestrator title: apps/orchestrator — role dispatch and gate pauses plan_slug: automation-foundation -status: draft +status: done priority: P1 tenant_id: global cursor_epic_id: null @@ -47,12 +47,13 @@ Each job calls `resolveModel(workspace, roleId, convoy, brief)` and passes `Reso ## Acceptance criteria -- [ ] Dry-run: orchestrator logs next role for a convoy without worker. +- [x] Dry-run: orchestrator logs next role for a convoy without worker. - [ ] Live: one implementer brief dispatched and paused correctly at gate. ## Tasks -_To be decomposed when epic graduates to `ready`._ +- [x] [Task-shared-orchestrator-driver](./Task-shared-orchestrator-driver.md) +- [x] [Task-apps-orchestrator-service](./Task-apps-orchestrator-service.md) ## Note diff --git a/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-apps-orchestrator-service.md b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-apps-orchestrator-service.md new file mode 100644 index 0000000..7c3ee17 --- /dev/null +++ b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-apps-orchestrator-service.md @@ -0,0 +1,23 @@ +--- +kind: task +slug: apps-orchestrator-service +title: apps/orchestrator Redis tick loop and dry-run mode +plan_slug: automation-foundation +epic_slug: apps-orchestrator +status: done +priority: P1 +tenant_id: global +owner: unassigned +cursor_todo_id: null +updated_at: "2026-07-07" +--- + +# Task summary + +`@tasks/orchestrator` package: env validation, DB convoy loader, Redis enqueue, poll loop, stall detector stub. + +## Acceptance criteria + +- [x] `ORCHESTRATOR_DRY_RUN=1` logs next role without worker +- [x] Live mode enqueues `role_dispatch` jobs when `REDIS_URL` is set +- [x] `docs/orchestrator.md` documents run commands diff --git a/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-shared-orchestrator-driver.md b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-shared-orchestrator-driver.md new file mode 100644 index 0000000..ec3c582 --- /dev/null +++ b/plans/Plan-automation-foundation/Epic-apps-orchestrator/Task-shared-orchestrator-driver.md @@ -0,0 +1,23 @@ +--- +kind: task +slug: shared-orchestrator-driver +title: Shared planOrchestratorAction driver and job schemas +plan_slug: automation-foundation +epic_slug: apps-orchestrator +status: done +priority: P1 +tenant_id: global +owner: unassigned +cursor_todo_id: null +updated_at: "2026-07-07" +--- + +# Task summary + +Pure convoy driver in `@tasks/shared/orchestrator` with vitest coverage for planning, gates, and implementer dispatch. + +## Acceptance criteria + +- [x] Pauses at gate statuses +- [x] Serial planning roles respect `skip` flags +- [x] `orchestratorJobSchema` + `ORCHESTRATOR_QUEUE_KEY` diff --git a/plans/Plan-automation-foundation/Plan-automation-foundation.md b/plans/Plan-automation-foundation/Plan-automation-foundation.md index 39f0039..a2b5231 100644 --- a/plans/Plan-automation-foundation/Plan-automation-foundation.md +++ b/plans/Plan-automation-foundation/Plan-automation-foundation.md @@ -167,7 +167,7 @@ flowchart LR | 3 | [Connected repos](./Epic-connected-repos/Epic-connected-repos.md) | done | `repo_links` table, “paste remote URL”, forge adapter interface | — | | 4 | [Convoy lifecycle MCP](./Epic-convoy-lifecycle-mcp/Epic-convoy-lifecycle-mcp.md) | done | `create_convoy`, `transition_convoy_status`, state machine §7, `convoy_events` | 2, 3 | | 5 | [Mission control gates UI](./Epic-mission-control-gates-ui/Epic-mission-control-gates-ui.md) | done | Convoy kanban, gate inbox, approve/deny/revise, gate co-pilot chat | 4 | -| 6 | [Orchestrator](./Epic-apps-orchestrator/Epic-apps-orchestrator.md) | draft | `apps/orchestrator`, Redis queue, role dispatch, pause at gates | 1, 3, 4 | +| 6 | [Orchestrator](./Epic-apps-orchestrator/Epic-apps-orchestrator.md) | done | `apps/orchestrator`, Redis queue, role dispatch, pause at gates | 1, 3, 4 | | 7 | [Worker v1](./Epic-apps-worker-v1/Epic-apps-worker-v1.md) | draft | `apps/worker` minimal tool loop, worktree-per-brief, MCP claim/complete | 1, 3, 6 | | 8 | [Forge webhooks](./Epic-forge-webhooks/Epic-forge-webhooks.md) | draft | PR opened/merged/CI webhooks → convoy transitions | 3, 4, 7 | | 9 | [Fleet dashboard](./Epic-fleet-dashboard/Epic-fleet-dashboard.md) | draft | Cross-repo home, `convoys-*` workspaces, “needs you” rollup | 4, 5 | diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6fd4ff7..8f53537 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -95,6 +95,40 @@ importers: specifier: ^5.7.0 version: 5.9.3 + apps/orchestrator: + dependencies: + '@tasks/database': + specifier: workspace:* + version: link:../../packages/database + '@tasks/shared': + specifier: workspace:* + version: link:../../packages/shared + drizzle-orm: + specifier: ^0.38.0 + version: 0.38.4(@opentelemetry/api@1.9.0)(@types/react@19.2.14)(postgres@3.4.8)(react@19.2.4) + ioredis: + specifier: ^5.4.2 + version: 5.10.1 + zod: + specifier: ^3.24.0 + version: 3.25.76 + devDependencies: + '@types/node': + specifier: ^22.10.0 + version: 22.19.15 + tsup: + specifier: ^8.3.5 + version: 8.5.1(jiti@1.21.7)(postcss@8.5.8)(tsx@4.21.0)(typescript@5.9.3)(yaml@2.8.3) + tsx: + specifier: ^4.19.2 + version: 4.21.0 + typescript: + specifier: ^5.7.0 + version: 5.9.3 + vitest: + specifier: ^2.1.9 + version: 2.1.9(@types/node@22.19.15) + apps/web: dependencies: '@ai-sdk/openai': @@ -8315,8 +8349,8 @@ snapshots: '@typescript-eslint/parser': 8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3) eslint: 9.39.4(jiti@1.21.7) eslint-import-resolver-node: 0.3.9 - eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@1.21.7)) - eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@1.21.7)) + eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)) + eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)) eslint-plugin-jsx-a11y: 6.10.2(eslint@9.39.4(jiti@1.21.7)) eslint-plugin-react: 7.37.5(eslint@9.39.4(jiti@1.21.7)) eslint-plugin-react-hooks: 5.2.0(eslint@9.39.4(jiti@1.21.7)) @@ -8335,7 +8369,7 @@ snapshots: transitivePeerDependencies: - supports-color - eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@1.21.7)): + eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)): dependencies: '@nolyfill/is-core-module': 1.0.39 debug: 4.4.3 @@ -8346,22 +8380,22 @@ snapshots: tinyglobby: 0.2.15 unrs-resolver: 1.11.1 optionalDependencies: - eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@1.21.7)) + eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)) transitivePeerDependencies: - supports-color - eslint-module-utils@2.12.1(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-node@0.3.9)(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@1.21.7)): + eslint-module-utils@2.12.1(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-node@0.3.9)(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)): dependencies: debug: 3.2.7 optionalDependencies: '@typescript-eslint/parser': 8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3) eslint: 9.39.4(jiti@1.21.7) eslint-import-resolver-node: 0.3.9 - eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@1.21.7)) + eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)) transitivePeerDependencies: - supports-color - eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@1.21.7)): + eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)): dependencies: '@rtsao/scc': 1.1.0 array-includes: 3.1.9 @@ -8372,7 +8406,7 @@ snapshots: doctrine: 2.1.0 eslint: 9.39.4(jiti@1.21.7) eslint-import-resolver-node: 0.3.9 - eslint-module-utils: 2.12.1(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-node@0.3.9)(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@1.21.7)) + eslint-module-utils: 2.12.1(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint-import-resolver-node@0.3.9)(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)))(eslint@9.39.4(jiti@1.21.7)) hasown: 2.0.2 is-core-module: 2.16.1 is-glob: 4.0.3