feat(worker): minimal agent runtime for orchestrator jobs (Epic 7) (#11)

* feat(worker): minimal agent runtime for orchestrator jobs (Epic 7)

Adds apps/worker to consume Redis dispatch jobs, run a tool loop with model registry resolution, and record claim/complete + convoy role_complete events.

Co-authored-by: Cursor <cursoragent@cursor.com>

* chore(deploy): add orchestrator and worker to Coolify compose

Fix CI pnpm setup conflict and ship Docker targets for the automation stack alongside web/collab/mcp.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -37,6 +37,13 @@ REDIS_URL="redis://:CHANGE_ME@192.168.68.102:6379/11"
 # ORCHESTRATOR_STALL_MINUTES=120
 # ORCHESTRATOR_MAX_IMPLEMENTERS=2
 
+# ----- Worker (apps/worker) -----
+# WORKER_DRY_RUN=1 skips Redis dequeue loop
+# WORKER_BLPOP_TIMEOUT_S=5
+# WORKER_MAX_TOOL_STEPS=12
+# WORKER_REPO_ROOT=  # defaults to monorepo root (pnpm-workspace.yaml walk-up)
+# WORKER_WORKTREE_BASE=/tmp/echodo-worker-worktrees
+
 # ----- Auth (NextAuth v5) -----
 # Local: http://localhost:3000  •  Prod: https://echodo.stillwell.cloud
 NEXTAUTH_URL="http://localhost:3000"

.github/workflows/ci.yml

@@ -26,8 +26,6 @@ jobs:
       # versions are more reproducible and the lockfile is pinned to 9.x.
       - name: Set up pnpm
         uses: pnpm/action-setup@v4
-        with:
-          version: 9
 
       - name: Set up Node.js
         uses: actions/setup-node@v4

apps/worker/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "@tasks/worker",
+  "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/ai": "workspace:*",
+    "@tasks/database": "workspace:*",
+    "@tasks/shared": "workspace:*",
+    "ai": "^4.3.16",
+    "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"
+  }
+}

apps/worker/src/db/convoy.ts

@@ -0,0 +1,227 @@
+import { and, asc, eq } from "drizzle-orm";
+
+import type { db as dbInstance } from "@tasks/database";
+import {
+  convoyEvents,
+  modelProviders,
+  objects,
+  roleModelRoutes,
+} from "@tasks/database/schema";
+import {
+  briefObjectContentSchema,
+  convoyObjectContentSchema,
+} from "@tasks/shared/convoy-lifecycle";
+import {
+  providerTypeSchema,
+  resolveModel,
+  type ModelProvider,
+  type ResolvedModel,
+  type RoleModelRoute,
+} from "@tasks/shared/model-routing";
+
+type Db = typeof dbInstance;
+
+export type LoadedJobContext = {
+  convoyId: string;
+  convoyTitle: string;
+  convoySlug: string;
+  workspaceId: string;
+  roleId: string;
+  briefId?: string;
+  briefFiles: string[];
+  briefAcceptance?: string;
+};
+
+function mapProvider(row: typeof modelProviders.$inferSelect): ModelProvider {
+  return {
+    id: row.id,
+    workspaceId: row.workspaceId,
+    alias: row.alias,
+    displayName: row.displayName,
+    providerType: providerTypeSchema.parse(row.providerType),
+    modelId: row.modelId,
+    baseUrl: row.baseUrl,
+    apiKeyEnvVar: row.apiKeyEnvVar,
+    extraHeaders: row.extraHeaders,
+    defaultParams: row.defaultParams,
+    metadata: row.metadata,
+    enabled: row.enabled,
+  };
+}
+
+function mapRoute(row: typeof roleModelRoutes.$inferSelect): RoleModelRoute {
+  return {
+    id: row.id,
+    workspaceId: row.workspaceId,
+    roleId: row.roleId,
+    providerAliasId: row.providerAliasId,
+    enabled: row.enabled,
+    sortOrder: row.sortOrder,
+  };
+}
+
+function readApiKey(envVarName: string): string | undefined {
+  return process.env[envVarName]?.trim() || undefined;
+}
+
+export async function loadJobContext(
+  db: Db,
+  input: {
+    workspaceId: string;
+    convoyId: string;
+    roleId: string;
+    briefId?: string;
+  },
+): Promise<LoadedJobContext | null> {
+  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) return null;
+
+  const content = convoyObjectContentSchema.safeParse(convoy.content);
+  if (!content.success) return null;
+
+  let briefFiles: string[] = [];
+  let briefAcceptance: string | undefined;
+
+  if (input.briefId) {
+    const [brief] = await db
+      .select()
+      .from(objects)
+      .where(
+        and(
+          eq(objects.id, input.briefId),
+          eq(objects.workspaceId, input.workspaceId),
+          eq(objects.parentId, convoy.id),
+        ),
+      )
+      .limit(1);
+
+    if (brief) {
+      const briefContent = briefObjectContentSchema.safeParse(brief.content);
+      if (briefContent.success) {
+        briefFiles = briefContent.data.files;
+        briefAcceptance = briefContent.data.acceptance;
+      }
+    }
+  }
+
+  return {
+    convoyId: convoy.id,
+    convoyTitle: convoy.title,
+    convoySlug: content.data.convoySlug,
+    workspaceId: input.workspaceId,
+    roleId: input.roleId,
+    briefId: input.briefId,
+    briefFiles,
+    briefAcceptance,
+  };
+}
+
+export async function resolveModelForRole(
+  db: Db,
+  input: {
+    workspaceId: string;
+    roleId: string;
+    convoyId: string;
+    briefRecommendedModel?: string;
+  },
+): Promise<
+  | { ok: true; model: ResolvedModel; resolvedAlias: string; modelTier?: string }
+  | { ok: false; reason: string }
+> {
+  const [convoy] = await db
+    .select({ content: objects.content })
+    .from(objects)
+    .where(
+      and(eq(objects.id, input.convoyId), eq(objects.workspaceId, input.workspaceId)),
+    )
+    .limit(1);
+
+  const convoyContent = convoy
+    ? convoyObjectContentSchema.safeParse(convoy.content)
+    : null;
+
+  const providers = (
+    await db
+      .select()
+      .from(modelProviders)
+      .where(eq(modelProviders.workspaceId, input.workspaceId))
+      .orderBy(asc(modelProviders.alias))
+  ).map(mapProvider);
+
+  const routes = (
+    await db
+      .select()
+      .from(roleModelRoutes)
+      .where(eq(roleModelRoutes.workspaceId, input.workspaceId))
+      .orderBy(asc(roleModelRoutes.sortOrder))
+  ).map(mapRoute);
+
+  const result = resolveModel({
+    roleId: input.roleId,
+    providers,
+    routes,
+    convoyPolicy: convoyContent?.success ? convoyContent.data.modelPolicy : undefined,
+    briefRecommendedModel: input.briefRecommendedModel,
+    getApiKey: readApiKey,
+  });
+
+  if (!result.ok) {
+    return { ok: false, reason: result.error.message };
+  }
+
+  return {
+    ok: true,
+    model: result.model,
+    resolvedAlias: result.resolvedAlias,
+    modelTier: result.model.tier,
+  };
+}
+
+export async function logRoleComplete(
+  db: Db,
+  input: {
+    workspaceId: string;
+    convoyId: string;
+    roleId: string;
+    briefId?: string;
+    outcome: string;
+    durationS: number;
+    modelAlias?: string;
+    providerType?: string;
+    modelId?: string;
+    modelTier?: string;
+    tokensTotal?: number;
+    summary?: string;
+  },
+) {
+  await db.insert(convoyEvents).values({
+    workspaceId: input.workspaceId,
+    convoyId: input.convoyId,
+    briefId: input.briefId ?? null,
+    role: input.roleId,
+    actor: input.roleId,
+    eventType: "role_complete",
+    durationS: input.durationS,
+    outcome: input.outcome,
+    modelAlias: input.modelAlias ?? null,
+    providerType: input.providerType ?? null,
+    modelId: input.modelId ?? null,
+    modelTier: input.modelTier ?? null,
+    metadata: {
+      source: "worker",
+      tokensTotal: input.tokensTotal ?? null,
+      summary: input.summary ?? null,
+    },
+  });
+}

apps/worker/src/db/task-lifecycle.ts

@@ -0,0 +1,215 @@
+import { and, eq, isNull } from "drizzle-orm";
+
+import type { db as dbInstance } from "@tasks/database";
+import { agentRuns, auditLog, markdownBacklogItems } from "@tasks/database/schema";
+import { exportBacklogItemToMarkdown, resolveWorkflowPrompt } from "@tasks/database/markdown-backlog";
+
+type Db = typeof dbInstance;
+
+export async function findBacklogItemByRepoPath(
+  db: Db,
+  workspaceId: string,
+  repoPath: string,
+) {
+  const [row] = await db
+    .select({
+      id: markdownBacklogItems.id,
+      title: markdownBacklogItems.title,
+      status: markdownBacklogItems.status,
+      repoPath: markdownBacklogItems.repoPath,
+    })
+    .from(markdownBacklogItems)
+    .where(
+      and(
+        eq(markdownBacklogItems.workspaceId, workspaceId),
+        eq(markdownBacklogItems.repoPath, repoPath),
+      ),
+    )
+    .limit(1);
+
+  return row ?? null;
+}
+
+export function pickTaskRepoPath(files: string[]): string | null {
+  const taskFile = files.find((file) => /Task-[^/]+\.md$/i.test(file));
+  return taskFile ?? files[0] ?? null;
+}
+
+export async function claimTask(
+  db: Db,
+  input: {
+    workspaceId: string;
+    backlogItemId: string;
+    client?: string;
+    model?: string;
+  },
+): Promise<{ runId: string; workflowPrompt: string }> {
+  const [item] = await db
+    .select({
+      id: markdownBacklogItems.id,
+      workspaceId: markdownBacklogItems.workspaceId,
+      status: markdownBacklogItems.status,
+    })
+    .from(markdownBacklogItems)
+    .where(eq(markdownBacklogItems.id, input.backlogItemId))
+    .limit(1);
+
+  if (!item || item.workspaceId !== input.workspaceId) {
+    throw new Error(`Backlog item ${input.backlogItemId} not found in workspace`);
+  }
+
+  const [existing] = await db
+    .select({ id: agentRuns.id })
+    .from(agentRuns)
+    .where(
+      and(eq(agentRuns.backlogItemId, input.backlogItemId), isNull(agentRuns.finishedAt)),
+    )
+    .limit(1);
+
+  if (existing) {
+    const resolved = await resolveWorkflowPrompt(db, {
+      workspaceId: input.workspaceId,
+      backlogItemId: input.backlogItemId,
+    });
+    return { runId: existing.id, workflowPrompt: resolved.prompt };
+  }
+
+  const priorStatus = item.status;
+  const shouldTransition = priorStatus === "ready" || priorStatus === "draft";
+  const metadata =
+    input.client || input.model
+      ? {
+          ...(input.client ? { client: input.client } : {}),
+          ...(input.model ? { model: input.model } : {}),
+        }
+      : null;
+
+  const runId = await db.transaction(async (tx) => {
+    const [created] = await tx
+      .insert(agentRuns)
+      .values({
+        workspaceId: input.workspaceId,
+        backlogItemId: input.backlogItemId,
+        actorUserId: null,
+        metadata,
+      })
+      .returning({ id: agentRuns.id });
+
+    if (shouldTransition) {
+      await tx
+        .update(markdownBacklogItems)
+        .set({ status: "in_progress", updatedAt: new Date() })
+        .where(eq(markdownBacklogItems.id, input.backlogItemId));
+    }
+
+    await tx.insert(auditLog).values({
+      workspaceId: input.workspaceId,
+      actorUserId: null,
+      action: "task.claimed",
+      targetType: "agent_run",
+      targetId: created.id,
+      metadata: {
+        backlog_item_id: input.backlogItemId,
+        prior_status: priorStatus,
+        transitioned: shouldTransition,
+        system_actor: "echodo-worker",
+        ...(metadata ?? {}),
+      },
+    });
+
+    return created.id;
+  });
+
+  const resolved = await resolveWorkflowPrompt(db, {
+    workspaceId: input.workspaceId,
+    backlogItemId: input.backlogItemId,
+  });
+
+  return { runId, workflowPrompt: resolved.prompt };
+}
+
+export async function completeTask(
+  db: Db,
+  input: {
+    runId: string;
+    outcome: "succeeded" | "failed" | "cancelled" | "stalled";
+    tokensTotal?: number;
+    tokensInput?: number;
+    tokensOutput?: number;
+    notes?: string;
+    error?: string;
+    finalStatus?: "done" | "blocked" | "ready" | "in_progress" | "cancelled";
+    modelAlias?: string;
+    modelTier?: string;
+    providerType?: string;
+    modelId?: string;
+    repoRoot?: string | null;
+  },
+) {
+  const [run] = await db.select().from(agentRuns).where(eq(agentRuns.id, input.runId)).limit(1);
+  if (!run) throw new Error("Run not found");
+  if (run.finishedAt !== null) throw new Error("Run already finished");
+
+  const finalStatus =
+    input.finalStatus ??
+    (input.outcome === "succeeded" ? "done" : input.outcome === "failed" ? "blocked" : null);
+
+  const tokensTotal =
+    input.tokensTotal ??
+    (typeof input.tokensInput === "number" && typeof input.tokensOutput === "number"
+      ? input.tokensInput + input.tokensOutput
+      : null);
+
+  await db.transaction(async (tx) => {
+    await tx
+      .update(agentRuns)
+      .set({
+        finishedAt: new Date(),
+        outcome: input.outcome,
+        tokensInput: input.tokensInput ?? null,
+        tokensOutput: input.tokensOutput ?? null,
+        tokensTotal,
+        notes: input.notes ?? null,
+        error: input.error ?? null,
+        modelAlias: input.modelAlias ?? null,
+        modelTier: input.modelTier ?? null,
+        providerType: input.providerType ?? null,
+        modelId: input.modelId ?? null,
+        metadata: {
+          ...(run.metadata ?? {}),
+          client: "echodo-worker",
+        },
+      })
+      .where(eq(agentRuns.id, input.runId));
+
+    if (finalStatus) {
+      await tx
+        .update(markdownBacklogItems)
+        .set({ status: finalStatus, updatedAt: new Date() })
+        .where(eq(markdownBacklogItems.id, run.backlogItemId));
+    }
+
+    await tx.insert(auditLog).values({
+      workspaceId: run.workspaceId,
+      actorUserId: null,
+      action: "task.completed",
+      targetType: "agent_run",
+      targetId: input.runId,
+      metadata: {
+        outcome: input.outcome,
+        finalStatus,
+        backlogItemId: run.backlogItemId,
+        tokensTotal,
+        system_actor: "echodo-worker",
+      },
+    });
+  });
+
+  if (input.repoRoot && finalStatus) {
+    await exportBacklogItemToMarkdown(db, {
+      workspaceId: run.workspaceId,
+      backlogItemId: run.backlogItemId,
+      repoRootAbs: input.repoRoot,
+    });
+  }
+}

apps/worker/src/env.ts

@@ -0,0 +1,40 @@
+import { z } from "zod";
+
+const envSchema = z.object({
+  DATABASE_URL: z.string().min(1),
+  REDIS_URL: z.string().min(1).optional(),
+  WORKER_DRY_RUN: z
+    .enum(["0", "1", "true", "false", ""])
+    .optional()
+    .transform((value) => value === "1" || value === "true"),
+  WORKER_BLPOP_TIMEOUT_S: z
+    .string()
+    .optional()
+    .transform((value) => {
+      const parsed = Number(value ?? "5");
+      return Number.isFinite(parsed) && parsed > 0 ? parsed : 5;
+    }),
+  WORKER_MAX_TOOL_STEPS: z
+    .string()
+    .optional()
+    .transform((value) => {
+      const parsed = Number(value ?? "12");
+      return Number.isFinite(parsed) && parsed > 0 ? parsed : 12;
+    }),
+  WORKER_REPO_ROOT: z.string().optional(),
+  WORKER_WORKTREE_BASE: z
+    .string()
+    .optional()
+    .transform((value) => value?.trim() || "/tmp/echodo-worker-worktrees"),
+});
+
+export type WorkerEnv = z.infer<typeof envSchema>;
+
+export function loadEnv(): WorkerEnv {
+  const parsed = envSchema.safeParse(process.env);
+  if (!parsed.success) {
+    console.error("[worker] Invalid environment:", parsed.error.flatten().fieldErrors);
+    process.exit(1);
+  }
+  return parsed.data;
+}

apps/worker/src/git/worktree.ts

@@ -0,0 +1,45 @@
+import { mkdir } from "node:fs/promises";
+import { join } from "node:path";
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+
+const execFileAsync = promisify(execFile);
+
+export type WorktreeContext = {
+  repoRoot: string;
+  worktreePath: string;
+  branch: string;
+};
+
+export async function ensureWorktree(
+  repoRoot: string,
+  worktreeBase: string,
+  jobId: string,
+  branchSuffix: string,
+): Promise<WorktreeContext> {
+  const branch = `worker/${branchSuffix}`.replace(/[^a-zA-Z0-9/_-]/g, "-");
+  const worktreePath = join(worktreeBase, jobId);
+  await mkdir(worktreeBase, { recursive: true });
+
+  try {
+    await execFileAsync("git", ["worktree", "add", "-B", branch, worktreePath], {
+      cwd: repoRoot,
+    });
+  } catch {
+    await execFileAsync("git", ["worktree", "add", worktreePath, branch], {
+      cwd: repoRoot,
+    });
+  }
+
+  return { repoRoot, worktreePath, branch };
+}
+
+export async function removeWorktree(repoRoot: string, worktreePath: string): Promise<void> {
+  try {
+    await execFileAsync("git", ["worktree", "remove", "--force", worktreePath], {
+      cwd: repoRoot,
+    });
+  } catch {
+    // Best-effort cleanup; operator can prune stale worktrees manually.
+  }
+}

apps/worker/src/handle-job.ts

@@ -0,0 +1,192 @@
+import type IoRedis from "ioredis";
+
+import type { db as dbInstance } from "@tasks/database";
+import type { OrchestratorJob } from "@tasks/shared/orchestrator";
+
+import {
+  loadJobContext,
+  logRoleComplete,
+  resolveModelForRole,
+} from "./db/convoy.js";
+import {
+  claimTask,
+  completeTask,
+  findBacklogItemByRepoPath,
+  pickTaskRepoPath,
+} from "./db/task-lifecycle.js";
+import type { WorkerEnv } from "./env.js";
+import { ensureWorktree, removeWorktree } from "./git/worktree.js";
+import { resolveRepoRoot } from "./lib/repo-root.js";
+import {
+  buildSessionPrompt,
+  createMinimalSession,
+  resolveRolePromptPath,
+} from "./session/minimal-session.js";
+
+type Db = typeof dbInstance;
+
+export async function handleOrchestratorJob(
+  db: Db,
+  _redis: IoRedis | null,
+  env: WorkerEnv,
+  job: OrchestratorJob,
+): Promise<void> {
+  if (job.type !== "role_dispatch") {
+    console.info(`[worker] skipping job type=${job.type} id=${job.id}`);
+    return;
+  }
+
+  const started = Date.now();
+  const dryRun = env.WORKER_DRY_RUN || !env.REDIS_URL;
+  const monorepoRoot = resolveRepoRoot(env.WORKER_REPO_ROOT);
+
+  const ctx = await loadJobContext(db, {
+    workspaceId: job.workspaceId,
+    convoyId: job.convoyId,
+    roleId: job.roleId,
+    briefId: job.briefId,
+  });
+
+  if (!ctx) {
+    console.error(`[worker] job=${job.id} convoy context not found`);
+    return;
+  }
+
+  const modelResult = await resolveModelForRole(db, {
+    workspaceId: job.workspaceId,
+    roleId: job.roleId,
+    convoyId: job.convoyId,
+  });
+
+  if (!modelResult.ok) {
+    console.error(`[worker] job=${job.id} model resolve failed: ${modelResult.reason}`);
+    await logRoleComplete(db, {
+      workspaceId: job.workspaceId,
+      convoyId: job.convoyId,
+      roleId: job.roleId,
+      briefId: job.briefId,
+      outcome: "failed",
+      durationS: Math.round((Date.now() - started) / 1000),
+      summary: modelResult.reason,
+    });
+    return;
+  }
+
+  let repoRoot = monorepoRoot;
+  let worktreePath: string | null = null;
+
+  if (!dryRun && job.briefId) {
+    try {
+      const worktree = await ensureWorktree(
+        monorepoRoot,
+        env.WORKER_WORKTREE_BASE,
+        job.id,
+        `${ctx.convoySlug}-brief-${job.briefId.slice(0, 8)}`,
+      );
+      repoRoot = worktree.worktreePath;
+      worktreePath = worktree.worktreePath;
+      console.info(`[worker] worktree=${worktree.worktreePath} branch=${worktree.branch}`);
+    } catch (error) {
+      console.warn("[worker] worktree setup failed, using monorepo root:", error);
+    }
+  }
+
+  let runId: string | null = null;
+  let workflowPrompt: string | undefined;
+  const taskRepoPath = pickTaskRepoPath(ctx.briefFiles);
+
+  if (taskRepoPath) {
+    const backlogItem = await findBacklogItemByRepoPath(db, job.workspaceId, taskRepoPath);
+    if (backlogItem) {
+      try {
+        const claim = await claimTask(db, {
+          workspaceId: job.workspaceId,
+          backlogItemId: backlogItem.id,
+          client: "echodo-worker",
+          model: modelResult.resolvedAlias,
+        });
+        runId = claim.runId;
+        workflowPrompt = claim.workflowPrompt;
+        console.info(`[worker] claimed backlog=${backlogItem.id} run=${runId}`);
+      } catch (error) {
+        console.error("[worker] claim_task failed:", error);
+      }
+    } else {
+      console.warn(`[worker] no backlog row for repo_path=${taskRepoPath}`);
+    }
+  }
+
+  const sessionContext = {
+    workspaceId: job.workspaceId,
+    convoyId: job.convoyId,
+    roleId: job.roleId,
+    briefId: job.briefId,
+    repoRoot,
+  };
+
+  const session = createMinimalSession(sessionContext, {
+    resolvedModel: modelResult.model,
+    resolvedAlias: modelResult.resolvedAlias,
+    rolePromptPath: resolveRolePromptPath(monorepoRoot, job.roleId),
+    workflowPrompt,
+    briefAcceptance: ctx.briefAcceptance,
+    briefFiles: ctx.briefFiles,
+    maxSteps: env.WORKER_MAX_TOOL_STEPS,
+    dryRun,
+  });
+
+  let outcome: "succeeded" | "failed" = "succeeded";
+  let summary = "";
+  let tokensTotal: number | undefined;
+
+  try {
+    const result = await session.run(buildSessionPrompt(sessionContext, ctx.convoyTitle));
+    summary = result.summary;
+    tokensTotal = result.tokensTotal;
+    console.info(
+      `[worker] job=${job.id} role=${job.roleId} tools=${result.toolCalls} dryRun=${dryRun}`,
+    );
+  } catch (error) {
+    outcome = "failed";
+    summary = error instanceof Error ? error.message : String(error);
+    console.error(`[worker] job=${job.id} session failed:`, error);
+  }
+
+  if (runId) {
+    try {
+      await completeTask(db, {
+        runId,
+        outcome,
+        notes: summary.slice(0, 2000),
+        error: outcome === "failed" ? summary.slice(0, 2000) : undefined,
+        tokensTotal,
+        modelAlias: modelResult.resolvedAlias,
+        modelTier: modelResult.modelTier,
+        providerType: modelResult.model.providerType,
+        modelId: modelResult.model.modelId,
+        repoRoot: monorepoRoot,
+      });
+    } catch (error) {
+      console.error("[worker] complete_task failed:", error);
+    }
+  }
+
+  await logRoleComplete(db, {
+    workspaceId: job.workspaceId,
+    convoyId: job.convoyId,
+    roleId: job.roleId,
+    briefId: job.briefId,
+    outcome,
+    durationS: Math.round((Date.now() - started) / 1000),
+    modelAlias: modelResult.resolvedAlias,
+    providerType: modelResult.model.providerType,
+    modelId: modelResult.model.modelId,
+    modelTier: modelResult.modelTier,
+    tokensTotal,
+    summary,
+  });
+
+  if (worktreePath) {
+    await removeWorktree(monorepoRoot, worktreePath);
+  }
+}

apps/worker/src/index.ts

@@ -0,0 +1,40 @@
+import { db } from "@tasks/database";
+import IoRedis from "ioredis";
+
+import { loadEnv } from "./env.js";
+import { handleOrchestratorJob } from "./handle-job.js";
+import { dequeueJob } from "./queue/dequeue.js";
+
+async function main(): Promise<void> {
+  const env = loadEnv();
+  const dryRun = env.WORKER_DRY_RUN || !env.REDIS_URL;
+  const redis = env.REDIS_URL && !env.WORKER_DRY_RUN ? new IoRedis(env.REDIS_URL) : null;
+
+  if (dryRun) {
+    console.info("[worker] DRY-RUN mode — dequeue loop idle (set REDIS_URL and WORKER_DRY_RUN=0 to consume)");
+    return;
+  }
+
+  if (!redis) {
+    console.error("[worker] REDIS_URL required when WORKER_DRY_RUN is off");
+    process.exit(1);
+  }
+
+  console.info(`[worker] consuming ${env.WORKER_BLPOP_TIMEOUT_S}s BLPOP on orchestrator queue`);
+
+  for (;;) {
+    try {
+      const job = await dequeueJob(redis, env.WORKER_BLPOP_TIMEOUT_S);
+      if (!job) continue;
+      console.info(`[worker] dequeued job=${job.id} role=${job.roleId} convoy=${job.convoyId}`);
+      await handleOrchestratorJob(db, redis, env, job);
+    } catch (error) {
+      console.error("[worker] loop error:", error);
+    }
+  }
+}
+
+main().catch((error) => {
+  console.error("[worker] fatal:", error);
+  process.exit(1);
+});

apps/worker/src/lib/repo-root.ts

@@ -0,0 +1,33 @@
+import { existsSync } from "node:fs";
+import { dirname, join } from "node:path";
+import { fileURLToPath } from "node:url";
+
+const WORKSPACE_MARKER = "pnpm-workspace.yaml";
+const MAX_WALK = 10;
+
+function findByMarker(start: string): string | null {
+  let dir = start;
+  for (let i = 0; i < MAX_WALK; i++) {
+    if (existsSync(join(dir, WORKSPACE_MARKER))) return dir;
+    const parent = dirname(dir);
+    if (parent === dir) return null;
+    dir = parent;
+  }
+  return null;
+}
+
+export function resolveRepoRoot(explicit?: string): string {
+  const env = explicit?.trim();
+  if (env && env.length > 0 && env !== "0" && env.toLowerCase() !== "off") {
+    return env;
+  }
+
+  const fileDir = dirname(fileURLToPath(import.meta.url));
+  const fromFile = findByMarker(fileDir);
+  if (fromFile) return fromFile;
+
+  const fromCwd = findByMarker(process.cwd());
+  if (fromCwd) return fromCwd;
+
+  return process.cwd();
+}

apps/worker/src/queue/dequeue.ts

@@ -0,0 +1,22 @@
+import type IoRedis from "ioredis";
+
+import {
+  ORCHESTRATOR_QUEUE_KEY,
+  orchestratorJobSchema,
+  type OrchestratorJob,
+} from "@tasks/shared/orchestrator";
+
+export async function dequeueJob(
+  redis: IoRedis,
+  timeoutSeconds = 5,
+): Promise<OrchestratorJob | null> {
+  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("[worker] Dropping invalid job payload:", raw);
+    return null;
+  }
+}

apps/worker/src/session/minimal-session.ts

@@ -0,0 +1,144 @@
+import { readFile } from "node:fs/promises";
+
+import { createLanguageModelFromResolved } from "@tasks/ai";
+import type { ResolvedModel } from "@tasks/shared/model-routing";
+import type { WorkerJobContext, WorkerSession } from "@tasks/shared/worker";
+import { generateText, tool } from "ai";
+import { z } from "zod";
+
+import {
+  grepTool,
+  readFileTool,
+  resolveRolePromptPath,
+  runCommandTool,
+  writeFileTool,
+} from "../tools/runner.js";
+
+export type MinimalSessionOptions = {
+  resolvedModel: ResolvedModel;
+  resolvedAlias: string;
+  rolePromptPath: string;
+  workflowPrompt?: string;
+  briefAcceptance?: string;
+  briefFiles?: string[];
+  maxSteps: number;
+  dryRun?: boolean;
+};
+
+export function createMinimalSession(
+  context: WorkerJobContext,
+  options: MinimalSessionOptions,
+): WorkerSession {
+  return {
+    context,
+    async run(prompt: string) {
+      if (options.dryRun) {
+        return {
+          summary: `[DRY-RUN] Would run ${context.roleId} on ${context.repoRoot}: ${prompt.slice(0, 200)}`,
+          toolCalls: 0,
+        };
+      }
+
+      let rolePrompt = "";
+      try {
+        rolePrompt = await readFile(options.rolePromptPath, "utf8");
+      } catch {
+        rolePrompt = `You are ${context.roleId} for convoy automation.`;
+      }
+
+      const modelResult = createLanguageModelFromResolved(options.resolvedModel);
+      if (!modelResult.ok) {
+        throw new Error(modelResult.reason);
+      }
+
+      const system = [
+        rolePrompt.trim(),
+        options.workflowPrompt ? `\n\n## Workflow\n${options.workflowPrompt.trim()}` : "",
+        options.briefFiles?.length
+          ? `\n\n## Brief files\n${options.briefFiles.map((f) => `- ${f}`).join("\n")}`
+          : "",
+        options.briefAcceptance
+          ? `\n\n## Acceptance\n${options.briefAcceptance.trim()}`
+          : "",
+        "\n\nUse tools to inspect and modify the repo. Finish with a concise summary of work done.",
+      ].join("");
+
+      let toolCalls = 0;
+      const repoRoot = context.repoRoot;
+
+      const result = await generateText({
+        model: modelResult.model,
+        maxSteps: options.maxSteps,
+        system,
+        prompt,
+        tools: {
+          read_file: tool({
+            description: "Read a file relative to the repo root",
+            parameters: z.object({
+              path: z.string().min(1),
+            }),
+            execute: async ({ path }) => {
+              toolCalls += 1;
+              const res = await readFileTool(repoRoot, path);
+              return res.ok ? res.output : `ERROR: ${res.error}`;
+            },
+          }),
+          write_file: tool({
+            description: "Write a file relative to the repo root",
+            parameters: z.object({
+              path: z.string().min(1),
+              content: z.string(),
+            }),
+            execute: async ({ path, content }) => {
+              toolCalls += 1;
+              const res = await writeFileTool(repoRoot, path, content);
+              return res.ok ? res.output : `ERROR: ${res.error}`;
+            },
+          }),
+          grep: tool({
+            description: "Search the repo with ripgrep",
+            parameters: z.object({
+              pattern: z.string().min(1),
+              path: z.string().optional(),
+            }),
+            execute: async ({ pattern, path }) => {
+              toolCalls += 1;
+              const res = await grepTool(repoRoot, pattern, path ?? ".");
+              return res.ok ? res.output : `ERROR: ${res.error}`;
+            },
+          }),
+          run_command: tool({
+            description: "Run an allowlisted shell command in the repo root",
+            parameters: z.object({
+              command: z.string().min(1),
+            }),
+            execute: async ({ command }) => {
+              toolCalls += 1;
+              const res = await runCommandTool(repoRoot, command);
+              return res.ok ? res.output : `ERROR: ${res.error}`;
+            },
+          }),
+        },
+      });
+
+      return {
+        summary: result.text.trim() || "Session completed with no summary text.",
+        toolCalls,
+        tokensTotal: result.usage?.totalTokens,
+      };
+    },
+  };
+}
+
+export function buildSessionPrompt(context: WorkerJobContext, convoyTitle: string): string {
+  return [
+    `Convoy: ${convoyTitle}`,
+    `Role: ${context.roleId}`,
+    context.briefId ? `Brief: ${context.briefId}` : null,
+    "Complete the assigned brief using the repo tools. Run lint/type-check/test when appropriate.",
+  ]
+    .filter(Boolean)
+    .join("\n");
+}
+
+export { resolveRolePromptPath };

apps/worker/src/tools/allowlist.test.ts

@@ -0,0 +1,16 @@
+import { describe, expect, it } from "vitest";
+
+import { isAllowedCommand } from "./allowlist.js";
+
+describe("isAllowedCommand", () => {
+  it("allows exact allowlisted commands", () => {
+    expect(isAllowedCommand("pnpm lint")).toBe(true);
+    expect(isAllowedCommand("pnpm test")).toBe(true);
+    expect(isAllowedCommand("git status")).toBe(true);
+  });
+
+  it("rejects arbitrary shell", () => {
+    expect(isAllowedCommand("rm -rf /")).toBe(false);
+    expect(isAllowedCommand("curl evil.example")).toBe(false);
+  });
+});

apps/worker/src/tools/allowlist.ts

@@ -0,0 +1,6 @@
+import { ALLOWED_COMMANDS } from "@tasks/shared/worker";
+
+export function isAllowedCommand(command: string): boolean {
+  const normalized = command.trim().replace(/\s+/g, " ");
+  return ALLOWED_COMMANDS.some((allowed) => normalized === allowed || normalized.startsWith(`${allowed} `));
+}

apps/worker/src/tools/runner.ts

@@ -0,0 +1,100 @@
+import { execFile } from "node:child_process";
+import { readFile, writeFile } from "node:fs/promises";
+import { join, resolve } from "node:path";
+import { promisify } from "node:util";
+
+import type { WorkerToolResult } from "@tasks/shared/worker";
+
+import { isAllowedCommand } from "./allowlist.js";
+
+const execFileAsync = promisify(execFile);
+
+function assertInsideRepo(repoRoot: string, targetPath: string): string {
+  const root = resolve(repoRoot);
+  const abs = resolve(root, targetPath);
+  if (!abs.startsWith(root)) {
+    throw new Error(`Path escapes repo root: ${targetPath}`);
+  }
+  return abs;
+}
+
+export async function readFileTool(repoRoot: string, path: string): Promise<WorkerToolResult> {
+  try {
+    const abs = assertInsideRepo(repoRoot, path);
+    const content = await readFile(abs, "utf8");
+    return { ok: true, output: content };
+  } catch (error) {
+    const message = error instanceof Error ? error.message : String(error);
+    return { ok: false, output: "", error: message };
+  }
+}
+
+export async function writeFileTool(
+  repoRoot: string,
+  path: string,
+  content: string,
+): Promise<WorkerToolResult> {
+  try {
+    const abs = assertInsideRepo(repoRoot, path);
+    await writeFile(abs, content, "utf8");
+    return { ok: true, output: `Wrote ${path} (${content.length} bytes)` };
+  } catch (error) {
+    const message = error instanceof Error ? error.message : String(error);
+    return { ok: false, output: "", error: message };
+  }
+}
+
+export async function grepTool(
+  repoRoot: string,
+  pattern: string,
+  path = ".",
+): Promise<WorkerToolResult> {
+  try {
+    const abs = assertInsideRepo(repoRoot, path);
+    const { stdout, stderr } = await execFileAsync(
+      "rg",
+      ["--line-number", "--max-count", "50", pattern, abs],
+      { cwd: repoRoot, maxBuffer: 512 * 1024 },
+    );
+    return { ok: true, output: stdout || stderr || "(no matches)" };
+  } catch (error) {
+    const err = error as { stdout?: string; stderr?: string; message?: string };
+    if (err.stdout) {
+      return { ok: true, output: err.stdout };
+    }
+    const message = err.message ?? String(error);
+    return { ok: false, output: "", error: message };
+  }
+}
+
+export async function runCommandTool(repoRoot: string, command: string): Promise<WorkerToolResult> {
+  if (!isAllowedCommand(command)) {
+    return {
+      ok: false,
+      output: "",
+      error: `Command not allowlisted: ${command}`,
+    };
+  }
+
+  try {
+    const { stdout, stderr } = await execFileAsync(command, {
+      cwd: repoRoot,
+      shell: true,
+      maxBuffer: 2 * 1024 * 1024,
+      env: process.env,
+    });
+    return { ok: true, output: [stdout, stderr].filter(Boolean).join("\n") || "(no output)" };
+  } catch (error) {
+    const err = error as { stdout?: string; stderr?: string; message?: string; code?: number };
+    const output = [err.stdout, err.stderr].filter(Boolean).join("\n");
+    return {
+      ok: false,
+      output: output || "",
+      error: err.message ?? `exit ${err.code ?? "unknown"}`,
+    };
+  }
+}
+
+export function resolveRolePromptPath(repoRoot: string, roleId: string): string {
+  return join(repoRoot, ".cursor", "agents", `${roleId}.md`);
+}

apps/worker/tsconfig.json

@@ -0,0 +1,9 @@
+{
+  "extends": "../../tsconfig.json",
+  "compilerOptions": {
+    "outDir": "dist",
+    "rootDir": "src",
+    "noEmit": true
+  },
+  "include": ["src/**/*"]
+}

apps/worker/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,
+});

apps/worker/vitest.config.ts

@@ -0,0 +1,7 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+  test: {
+    environment: "node",
+  },
+});

docker/.dockerignore

@@ -9,6 +9,8 @@
 .git
 .gitignore
 .cursor
+!.cursor/agents
+!.cursor/agents/**
 .idea
 .vscode
 *.swp

docker/Dockerfile

@@ -1,7 +1,7 @@
 # syntax=docker/dockerfile:1
 #
 # Single multi-stage Dockerfile for ECHODO. Build via Docker Compose with
-# `target:` to select web | collab | mcp.
+# `target:` to select web | collab | mcp | orchestrator | worker.
 #
 # Why one file instead of three?
 #   - The `deps` stage runs `pnpm install` ONCE for all 723 workspace packages.
@@ -33,6 +33,8 @@ COPY package.json pnpm-lock.yaml pnpm-workspace.yaml turbo.json tsconfig.json ./
 COPY apps/web/package.json ./apps/web/
 COPY apps/collab-server/package.json ./apps/collab-server/
 COPY apps/mcp-server/package.json ./apps/mcp-server/
+COPY apps/orchestrator/package.json ./apps/orchestrator/
+COPY apps/worker/package.json ./apps/worker/
 COPY packages/database/package.json ./packages/database/
 COPY packages/shared/package.json ./packages/shared/
 COPY packages/ai/package.json ./packages/ai/
@@ -70,6 +72,31 @@ ENV NODE_ENV=production
 RUN pnpm exec turbo build --filter=@tasks/mcp-server
 
 
+# ============================================================================
+# orchestrator-build — tsup build for convoy driver
+# ============================================================================
+FROM deps AS orchestrator-build
+COPY apps/orchestrator ./apps/orchestrator
+COPY packages/database ./packages/database
+COPY packages/shared ./packages/shared
+
+ENV NODE_ENV=production
+RUN pnpm exec turbo build --filter=@tasks/orchestrator
+
+
+# ============================================================================
+# worker-build — tsup build for agent runtime
+# ============================================================================
+FROM deps AS worker-build
+COPY apps/worker ./apps/worker
+COPY packages/ai ./packages/ai
+COPY packages/database ./packages/database
+COPY packages/shared ./packages/shared
+
+ENV NODE_ENV=production
+RUN pnpm exec turbo build --filter=@tasks/worker
+
+
 # ============================================================================
 # web-build — Next.js production build → standalone output
 # ============================================================================
@@ -89,6 +116,8 @@ COPY docker/next.config.docker.ts ./apps/web/next.config.ts
 # runtime web image never reads /tmp.
 COPY --from=collab-build /app/apps/collab-server/dist/index.mjs /tmp/.collab-built
 COPY --from=mcp-build /app/apps/mcp-server/dist/index.js /tmp/.mcp-built
+COPY --from=orchestrator-build /app/apps/orchestrator/dist/index.mjs /tmp/.orchestrator-built
+COPY --from=worker-build /app/apps/worker/dist/index.mjs /tmp/.worker-built
 
 ENV NEXT_TELEMETRY_DISABLED=1
 ENV NODE_ENV=production
@@ -189,3 +218,76 @@ HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
   CMD pgrep -f "node.*dist/index\\.js" > /dev/null || exit 1
 
 CMD ["node", "dist/index.js"]
+
+
+# ============================================================================
+# orchestrator — runtime image (convoy driver + Redis enqueue)
+# ============================================================================
+FROM node:${NODE_VERSION}-alpine AS orchestrator
+RUN addgroup --system --gid 1001 nodejs \
+  && adduser --system --uid 1001 orchestrator --ingroup nodejs
+
+WORKDIR /app
+
+ENV NODE_ENV=production
+
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/node_modules ./node_modules
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/packages/database ./packages/database
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/packages/shared ./packages/shared
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/apps/orchestrator/dist ./apps/orchestrator/dist
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/apps/orchestrator/package.json ./apps/orchestrator/package.json
+COPY --from=orchestrator-build --chown=orchestrator:nodejs /app/apps/orchestrator/node_modules ./apps/orchestrator/node_modules
+
+USER orchestrator
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+  CMD pgrep -f "node.*dist/index\\.mjs" > /dev/null || exit 1
+
+WORKDIR /app/apps/orchestrator
+CMD ["node", "dist/index.mjs"]
+
+
+# ============================================================================
+# worker — runtime image (orchestrator job consumer + tool loop)
+# ============================================================================
+FROM node:${NODE_VERSION}-alpine AS worker
+RUN apk add --no-cache git ripgrep procps \
+  && addgroup --system --gid 1001 nodejs \
+  && adduser --system --uid 1001 worker --ingroup nodejs
+
+WORKDIR /app
+
+ENV NODE_ENV=production
+ENV WORKER_REPO_ROOT=/app/repo
+
+COPY --from=worker-build --chown=worker:nodejs /app/node_modules ./node_modules
+COPY --from=worker-build --chown=worker:nodejs /app/packages/ai ./packages/ai
+COPY --from=worker-build --chown=worker:nodejs /app/packages/database ./packages/database
+COPY --from=worker-build --chown=worker:nodejs /app/packages/shared ./packages/shared
+COPY --from=worker-build --chown=worker:nodejs /app/apps/worker/dist ./apps/worker/dist
+COPY --from=worker-build --chown=worker:nodejs /app/apps/worker/package.json ./apps/worker/package.json
+COPY --from=worker-build --chown=worker:nodejs /app/apps/worker/node_modules ./apps/worker/node_modules
+
+# Repo snapshot for tool loop + git worktrees (dogfood / connected-repo jobs).
+COPY --chown=worker:nodejs apps ./repo/apps
+COPY --chown=worker:nodejs packages ./repo/packages
+COPY --chown=worker:nodejs plans ./repo/plans
+COPY --chown=worker:nodejs .cursor ./repo/.cursor
+COPY --chown=worker:nodejs pnpm-workspace.yaml package.json turbo.json tsconfig.json ./repo/
+
+# Minimal git repo so worktrees + allowlisted git commands work on the deploy snapshot.
+USER root
+RUN cd /app/repo \
+  && git init -q \
+  && git config user.email "worker@echodo" \
+  && git config user.name "echodo-worker" \
+  && git add -A \
+  && git -c commit.gpgsign=false commit -q -m "deploy snapshot" \
+  || true
+USER worker
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+  CMD pgrep -f "node.*dist/index\\.mjs" > /dev/null || exit 1
+
+WORKDIR /app/apps/worker
+CMD ["node", "dist/index.mjs"]

docker/docker-compose.coolify.yml

@@ -18,7 +18,7 @@
 #   - Uses Coolify SERVICE_FQDN_* magic env vars so Coolify auto-wires Traefik labels
 #
 # Build strategy:
-#   All three services share docker/Dockerfile and select runtime via `target:`.
+#   All five services share docker/Dockerfile and select runtime via `target:`.
 #   The shared `deps` stage is hashed identically across targets, so BuildKit
 #   runs `pnpm install` ONCE.
 #
@@ -139,6 +139,62 @@ services:
     networks:
       - echodo
 
+  orchestrator:
+    build:
+      context: .
+      dockerfile: docker/Dockerfile
+      target: orchestrator
+    restart: unless-stopped
+    environment:
+      - DATABASE_URL=${DATABASE_URL}
+      - REDIS_URL=${REDIS_URL}
+      - ORCHESTRATOR_DRY_RUN=${ORCHESTRATOR_DRY_RUN:-0}
+      - ORCHESTRATOR_POLL_MS=${ORCHESTRATOR_POLL_MS:-5000}
+      - ORCHESTRATOR_STALL_MINUTES=${ORCHESTRATOR_STALL_MINUTES:-120}
+      - ORCHESTRATOR_MAX_IMPLEMENTERS=${ORCHESTRATOR_MAX_IMPLEMENTERS:-2}
+      - NODE_ENV=production
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - "pgrep -f 'node.*dist/index\\.mjs' > /dev/null || exit 1"
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 25s
+    networks:
+      - echodo
+
+  worker:
+    build:
+      context: .
+      dockerfile: docker/Dockerfile
+      target: worker
+    depends_on:
+      - orchestrator
+    restart: unless-stopped
+    environment:
+      - DATABASE_URL=${DATABASE_URL}
+      - REDIS_URL=${REDIS_URL}
+      - WORKER_DRY_RUN=${WORKER_DRY_RUN:-0}
+      - WORKER_BLPOP_TIMEOUT_S=${WORKER_BLPOP_TIMEOUT_S:-5}
+      - WORKER_MAX_TOOL_STEPS=${WORKER_MAX_TOOL_STEPS:-12}
+      - WORKER_REPO_ROOT=/app/repo
+      - WORKER_WORKTREE_BASE=/tmp/echodo-worker-worktrees
+      - OPENAI_API_KEY=${OPENAI_API_KEY}
+      - OPENAI_BASE_URL=${OPENAI_BASE_URL}
+      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
+      - NODE_ENV=production
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - "pgrep -f 'node.*dist/index\\.mjs' > /dev/null || exit 1"
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 25s
+    networks:
+      - echodo
+
 networks:
   echodo:
     driver: bridge

docs/worker.md

@@ -0,0 +1,42 @@
+# Worker (`apps/worker`)
+
+Minimal first-party agent runtime for convoy automation. Consumes orchestrator jobs from Redis, resolves models via the workspace registry, runs a tool loop against a git worktree, and records `claim_task` / `complete_task` + `convoy_events.role_complete`.
+
+## Run locally
+
+```bash
+# Dry-run (no Redis consumer — exits after banner)
+WORKER_DRY_RUN=1 pnpm --filter @tasks/worker dev
+
+# Live consumer (requires REDIS_URL + DATABASE_URL)
+pnpm --filter @tasks/worker dev
+```
+
+Pair with the orchestrator:
+
+```bash
+ORCHESTRATOR_DRY_RUN=0 REDIS_URL=... pnpm --filter @tasks/orchestrator dev
+WORKER_DRY_RUN=0 REDIS_URL=... pnpm --filter @tasks/worker dev
+```
+
+## Architecture
+
+- **Queue** — `BLPOP` on `echodo:orchestrator:jobs` (same key as `apps/orchestrator`)
+- **Session** — `WorkerSession` interface in `@tasks/shared/worker`; v1 implementation uses Vercel AI SDK `generateText` + tools
+- **Tools** — `read_file`, `write_file`, `grep` (ripgrep), `run_command` (allowlist: `pnpm lint`, `pnpm type-check`, `pnpm test`, `git status`, `git diff`)
+- **Worktree** — one git worktree per brief job under `WORKER_WORKTREE_BASE`
+- **Lifecycle** — DB writes mirroring MCP `claim_task` / `complete_task`; `role_complete` convoy event on finish
+
+## Env
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `DATABASE_URL` | required | Convoy, backlog, model registry |
+| `REDIS_URL` | optional | Job queue; missing ⇒ dry-run |
+| `WORKER_DRY_RUN` | `false` | Skip dequeue loop |
+| `WORKER_BLPOP_TIMEOUT_S` | `5` | Blocking pop timeout |
+| `WORKER_MAX_TOOL_STEPS` | `12` | Max tool rounds per session |
+| `WORKER_REPO_ROOT` | monorepo root | Repo for role prompts + worktree base |
+| `WORKER_WORKTREE_BASE` | `/tmp/echodo-worker-worktrees` | Worktree parent directory |
+
+Provider API keys are read from env vars named in `model_providers.api_key_env_var` (same as web settings).

packages/shared/package.json

@@ -10,6 +10,7 @@
     "./connected-repos": "./src/connected-repos/index.ts",
     "./convoy-lifecycle": "./src/convoy-lifecycle/index.ts",
     "./orchestrator": "./src/orchestrator/index.ts",
+    "./worker": "./src/worker/index.ts",
     "./model-routing": "./src/model-routing/index.ts",
     "./types": "./src/types/index.ts",
     "./utils": "./src/utils/index.ts"

packages/shared/src/index.ts

@@ -2,5 +2,6 @@ export * from "./connected-repos";
 export * from "./convoy-lifecycle";
 export * from "./model-routing";
 export * from "./orchestrator";
+export * from "./worker";
 export * from "./types";
 export * from "./utils";

packages/shared/src/worker/index.ts

@@ -0,0 +1,9 @@
+export {
+  ALLOWED_COMMANDS,
+  type WorkerJobContext,
+  type WorkerSession,
+  type WorkerToolCall,
+  type WorkerToolName,
+  type WorkerToolResult,
+  workerToolNameSchema,
+} from "./session";

packages/shared/src/worker/session.ts

@@ -0,0 +1,42 @@
+import { z } from "zod";
+
+export const workerToolNameSchema = z.enum([
+  "read_file",
+  "write_file",
+  "grep",
+  "run_command",
+]);
+
+export type WorkerToolName = z.infer<typeof workerToolNameSchema>;
+
+export const ALLOWED_COMMANDS = [
+  "pnpm lint",
+  "pnpm type-check",
+  "pnpm test",
+  "git status",
+  "git diff",
+] as const;
+
+export type WorkerJobContext = {
+  workspaceId: string;
+  convoyId: string;
+  roleId: string;
+  briefId?: string;
+  repoRoot: string;
+};
+
+export type WorkerToolCall = {
+  tool: WorkerToolName;
+  input: Record<string, unknown>;
+};
+
+export type WorkerToolResult = {
+  ok: boolean;
+  output: string;
+  error?: string;
+};
+
+export interface WorkerSession {
+  readonly context: WorkerJobContext;
+  run(prompt: string): Promise<{ summary: string; toolCalls: number; tokensTotal?: number }>;
+}

plans/Plan-automation-foundation/Epic-apps-worker-v1/Epic-apps-worker-v1.md

@@ -3,7 +3,7 @@ kind: epic
 slug: apps-worker-v1
 title: apps/worker v1 — minimal Echodo coding agent
 plan_slug: automation-foundation
-status: draft
+status: in_progress
 priority: P1
 tenant_id: global
 cursor_epic_id: null

pnpm-lock.yaml

@@ -367,6 +367,46 @@ importers:
         specifier: ^5.7.0
         version: 5.9.3
 
+  apps/worker:
+    dependencies:
+      '@tasks/ai':
+        specifier: workspace:*
+        version: link:../../packages/ai
+      '@tasks/database':
+        specifier: workspace:*
+        version: link:../../packages/database
+      '@tasks/shared':
+        specifier: workspace:*
+        version: link:../../packages/shared
+      ai:
+        specifier: ^4.3.16
+        version: 4.3.19(react@19.2.4)(zod@3.25.76)
+      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)
+
   packages/ai:
     dependencies:
       '@ai-sdk/anthropic':
@@ -8349,8 +8389,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(@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-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-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-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-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))
@@ -8369,7 +8409,7 @@ snapshots:
     transitivePeerDependencies:
       - supports-color
 
-  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-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@1.21.7)):
     dependencies:
       '@nolyfill/is-core-module': 1.0.39
       debug: 4.4.3
@@ -8380,22 +8420,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-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-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))
     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-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-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)):
     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(@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-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(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-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-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)):
     dependencies:
       '@rtsao/scc': 1.1.0
       array-includes: 3.1.9
@@ -8406,7 +8446,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-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-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))
       hasown: 2.0.2
       is-core-module: 2.16.1
       is-glob: 4.0.3