diff --git a/.env.example b/.env.example index ed6ce0e..c0ea090 100644 --- a/.env.example +++ b/.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" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cd58d65..6721d11 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..9b91de6 --- /dev/null +++ b/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" + } +} diff --git a/apps/worker/src/db/convoy.ts b/apps/worker/src/db/convoy.ts new file mode 100644 index 0000000..349ec7b --- /dev/null +++ b/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 { + 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, + }, + }); +} diff --git a/apps/worker/src/db/task-lifecycle.ts b/apps/worker/src/db/task-lifecycle.ts new file mode 100644 index 0000000..f899161 --- /dev/null +++ b/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, + }); + } +} diff --git a/apps/worker/src/env.ts b/apps/worker/src/env.ts new file mode 100644 index 0000000..ab42948 --- /dev/null +++ b/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; + +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; +} diff --git a/apps/worker/src/git/worktree.ts b/apps/worker/src/git/worktree.ts new file mode 100644 index 0000000..fd3d695 --- /dev/null +++ b/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 { + 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 { + try { + await execFileAsync("git", ["worktree", "remove", "--force", worktreePath], { + cwd: repoRoot, + }); + } catch { + // Best-effort cleanup; operator can prune stale worktrees manually. + } +} diff --git a/apps/worker/src/handle-job.ts b/apps/worker/src/handle-job.ts new file mode 100644 index 0000000..e77b2d7 --- /dev/null +++ b/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 { + 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); + } +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts new file mode 100644 index 0000000..b0be125 --- /dev/null +++ b/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 { + 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); +}); diff --git a/apps/worker/src/lib/repo-root.ts b/apps/worker/src/lib/repo-root.ts new file mode 100644 index 0000000..e13e578 --- /dev/null +++ b/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(); +} diff --git a/apps/worker/src/queue/dequeue.ts b/apps/worker/src/queue/dequeue.ts new file mode 100644 index 0000000..8c238fc --- /dev/null +++ b/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 { + 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; + } +} diff --git a/apps/worker/src/session/minimal-session.ts b/apps/worker/src/session/minimal-session.ts new file mode 100644 index 0000000..ef8088f --- /dev/null +++ b/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 }; diff --git a/apps/worker/src/tools/allowlist.test.ts b/apps/worker/src/tools/allowlist.test.ts new file mode 100644 index 0000000..17feb5e --- /dev/null +++ b/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); + }); +}); diff --git a/apps/worker/src/tools/allowlist.ts b/apps/worker/src/tools/allowlist.ts new file mode 100644 index 0000000..901f9c9 --- /dev/null +++ b/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} `)); +} diff --git a/apps/worker/src/tools/runner.ts b/apps/worker/src/tools/runner.ts new file mode 100644 index 0000000..aec8ade --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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`); +} diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json new file mode 100644 index 0000000..486abdd --- /dev/null +++ b/apps/worker/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "noEmit": true + }, + "include": ["src/**/*"] +} diff --git a/apps/worker/tsup.config.ts b/apps/worker/tsup.config.ts new file mode 100644 index 0000000..9a20f39 --- /dev/null +++ b/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, +}); diff --git a/apps/worker/vitest.config.ts b/apps/worker/vitest.config.ts new file mode 100644 index 0000000..f624398 --- /dev/null +++ b/apps/worker/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + environment: "node", + }, +}); diff --git a/docker/.dockerignore b/docker/.dockerignore index 1496c62..c47474d 100644 --- a/docker/.dockerignore +++ b/docker/.dockerignore @@ -9,6 +9,8 @@ .git .gitignore .cursor +!.cursor/agents +!.cursor/agents/** .idea .vscode *.swp diff --git a/docker/Dockerfile b/docker/Dockerfile index b4de996..8449e08 100644 --- a/docker/Dockerfile +++ b/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"] diff --git a/docker/docker-compose.coolify.yml b/docker/docker-compose.coolify.yml index 3f1b0d5..1a2d030 100644 --- a/docker/docker-compose.coolify.yml +++ b/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 diff --git a/docs/worker.md b/docs/worker.md new file mode 100644 index 0000000..92e884f --- /dev/null +++ b/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). diff --git a/packages/shared/package.json b/packages/shared/package.json index 4abaa09..32f330c 100644 --- a/packages/shared/package.json +++ b/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" diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index dbd445e..86dec53 100644 --- a/packages/shared/src/index.ts +++ b/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"; diff --git a/packages/shared/src/worker/index.ts b/packages/shared/src/worker/index.ts new file mode 100644 index 0000000..2f885a3 --- /dev/null +++ b/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"; diff --git a/packages/shared/src/worker/session.ts b/packages/shared/src/worker/session.ts new file mode 100644 index 0000000..a574f0b --- /dev/null +++ b/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; + +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; +}; + +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 }>; +} diff --git a/plans/Plan-automation-foundation/Epic-apps-worker-v1/Epic-apps-worker-v1.md b/plans/Plan-automation-foundation/Epic-apps-worker-v1/Epic-apps-worker-v1.md index 02346a8..11c94b7 100644 --- a/plans/Plan-automation-foundation/Epic-apps-worker-v1/Epic-apps-worker-v1.md +++ b/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 diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8f53537..bd546d5 100644 --- a/pnpm-lock.yaml +++ b/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-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-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-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