diff --git a/apps/worker/src/executor/index.ts b/apps/worker/src/executor/index.ts new file mode 100644 index 0000000..2a4406c --- /dev/null +++ b/apps/worker/src/executor/index.ts @@ -0,0 +1,71 @@ +import type { HarnessType, WorkerExecutor, WorkerJobContext } from "@tasks/shared/worker"; +import { createMinimalSession } from "../session/minimal-session.js"; +import { createRemoteAgUiExecutor } from "./remote-ag-ui.js"; + +export type ExecutorDeps = { + fetchImpl?: typeof fetch; + getEnvVar?: (name: string) => string | undefined; +}; + +function buildMinimalExecutor(): WorkerExecutor { + return { + id: "minimal", + harnessType: "minimal" as HarnessType, + async run(ctx: WorkerJobContext, opts: any): Promise { + const session = createMinimalSession(ctx, { + resolvedModel: opts.resolvedModel, + resolvedAlias: opts.resolvedAlias, + rolePromptPath: opts.rolePromptPath, + workflowPrompt: opts.workflowPrompt, + briefAcceptance: opts.briefAcceptance, + briefFiles: opts.briefFiles, + maxSteps: opts.maxSteps, + dryRun: opts.dryRun, + checkpointEveryNSteps: opts.checkpointEveryNSteps, + resumeToolHistory: opts.resumeToolHistory, + resumeFilesTouched: opts.resumeFilesTouched, + allowedCommands: opts.allowedCommands, + onCheckpoint: opts.onCheckpoint, + }); + const convoyTitle = opts.convoyTitle ?? "unknown"; + const result = await session.run(opts.buildPrompt ? opts.buildPrompt(ctx, convoyTitle) : ""); + return { + summary: result.summary, + toolCalls: result.toolCalls, + tokensTotal: result.tokensTotal, + outcome: "succeeded", + }; + }, + }; +} + +function buildRemoteAgUiExecutor( + endpointUrl: string, + authEnvVar: string, + deps: ExecutorDeps, +): WorkerExecutor { + return createRemoteAgUiExecutor({ endpointUrl, authEnvVar }, { + fetchImpl: deps.fetchImpl ?? fetch, + getEnvVar: deps.getEnvVar ?? ((name: string) => process.env[name] ?? undefined), + }); +} + +export type HarnessExecutorConfig = { + harnessType: HarnessType; + endpointUrl?: string | null; + authEnvVar?: string | null; +}; + +export function getExecutor( + config: HarnessExecutorConfig, + deps: ExecutorDeps = {}, +): WorkerExecutor | null { + switch (config.harnessType) { + case "minimal": return buildMinimalExecutor(); + case "remote-ag-ui": + if (!config.endpointUrl || !config.authEnvVar) return null; + return buildRemoteAgUiExecutor(config.endpointUrl, config.authEnvVar, deps); + case "crewai": case "langgraph": return null; + default: return null; + } +} diff --git a/apps/worker/src/executor/remote-ag-ui.ts b/apps/worker/src/executor/remote-ag-ui.ts new file mode 100644 index 0000000..9da297b --- /dev/null +++ b/apps/worker/src/executor/remote-ag-ui.ts @@ -0,0 +1,57 @@ +import type { + HarnessType, + WorkerExecutor, + WorkerJobContext, +} from "@tasks/shared/worker"; + +export type RemoteAgUiConfig = { + endpointUrl: string; + authEnvVar: string; +}; + +export type RemoteAgUiExecutorDeps = { + fetchImpl: typeof fetch; + getEnvVar: (name: string) => string | undefined; +}; + +export function createRemoteAgUiExecutor( + config: RemoteAgUiConfig, + deps: RemoteAgUiExecutorDeps, +): WorkerExecutor { + return { + id: "remote-ag-ui", + harnessType: "remote-ag-ui" as HarnessType, + async run(ctx: WorkerJobContext, opts: any): Promise { + const authToken = deps.getEnvVar(config.authEnvVar); + if (!authToken) { + return { + summary: `Missing auth env var: ${config.authEnvVar}`, + toolCalls: 0, + outcome: "failed", + }; + } + const payload = { + workspaceId: ctx.workspaceId, + convoyId: ctx.convoyId, + roleId: ctx.roleId, + briefId: ctx.briefId ?? null, + prompt: opts.buildPrompt ? opts.buildPrompt(ctx, opts.convoyTitle ?? "unknown") : "", + repoRoot: ctx.repoRoot, + maxSteps: opts.maxSteps ?? 10, + }; + try { + const res = await deps.fetchImpl(config.endpointUrl, { + method: "POST", + headers: { "Content-Type": "application/json", Authorization: `Bearer ${authToken}` }, + body: JSON.stringify(payload), + }); + if (!res.ok) { + return { summary: `Remote agent rejected: ${res.status}`, toolCalls: 0, outcome: "failed" }; + } + return { summary: "Remote agent accepted (polling not fully implemented in brief)", toolCalls: 0, outcome: "succeeded" }; + } catch (e: any) { + return { summary: `Remote agent error: ${e.message ?? e}`, toolCalls: 0, outcome: "failed" }; + } + }, + }; +} diff --git a/apps/worker/src/handle-job.ts b/apps/worker/src/handle-job.ts index 10d7a0b..0091fdd 100644 --- a/apps/worker/src/handle-job.ts +++ b/apps/worker/src/handle-job.ts @@ -30,11 +30,11 @@ import { } from "./git/worktree.js"; import { openDraftPR } from "./git/pr.js"; import { resolveHarnessForJob, getHarnessExecutor } from "./harness.js"; +import { getExecutor, type HarnessExecutorConfig } from "./executor/index.js"; import { resolveHarnessForRoleDb } from "./db/harness.js"; import { resolveRepoRoot } from "./lib/repo-root.js"; import { buildSessionPrompt, - createMinimalSession, resolveRolePromptPath, } from "./session/minimal-session.js"; import { connectedRepos, convoyEvents, objects } from "@tasks/database/schema"; @@ -88,9 +88,9 @@ export async function handleOrchestratorJob( } catch (e) { console.warn("[worker] harness DB resolve failed, using local fallback:", e); } - const executor = getHarnessExecutor(harness.alias); + const executorMeta = getHarnessExecutor(harness.alias); const needsWorktree = harness.needsWorktree; - void executor; + void executorMeta; let repoRoot = monorepoRoot; let worktreePath: string | null = null; @@ -213,7 +213,8 @@ export async function handleOrchestratorJob( repoRoot, }; - // WorkerAdapter dispatch — minimal adapter delegates to createMinimalSession; remote adapters would POST to AG-UI. + // WorkerAdapter dispatch — registry routes to the correct executor based on + // harnessType. "minimal" runs locally; "remote-ag-ui" POSTs to the remote agent. const operatorNotesProvider = { harnessAlias: harness.alias, poll: async () => { @@ -230,52 +231,74 @@ export async function handleOrchestratorJob( }, }; - 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, - checkpointEveryNSteps: env.WORKER_CHECKPOINT_EVERY_N_STEPS, - resumeToolHistory, - resumeFilesTouched, - allowedCommands: harness.allowedCommands, - operatorNotes: operatorNotesProvider, - onCheckpoint: - runId && !dryRun - ? async ({ checkpointSeq, toolHistory, filesTouched }) => { - await saveWorkerCheckpoint(db, { - workspaceId: job.workspaceId, - agentRunId: runId, - convoyId: job.convoyId, - briefId: job.briefId, - roleId: job.roleId, - checkpointSeq, - toolHistory, - filesTouched, - branch: worktreeBranch, - worktreePath, - }); - wroteCheckpoint = true; - console.info(`[worker] checkpoint seq=${checkpointSeq} run=${runId} harness=${harness.alias}`); - } - : undefined, - }); + const onCheckpointSink = + runId && !dryRun + ? async ({ checkpointSeq, toolHistory, filesTouched }: { + checkpointSeq: number; + toolHistory: unknown[]; + filesTouched: unknown[]; + }) => { + await saveWorkerCheckpoint(db, { + workspaceId: job.workspaceId, + agentRunId: runId, + convoyId: job.convoyId, + briefId: job.briefId, + roleId: job.roleId, + checkpointSeq, + toolHistory: toolHistory as never, + filesTouched: filesTouched as never, + branch: worktreeBranch, + worktreePath, + }); + wroteCheckpoint = true; + console.info(`[worker] checkpoint seq=${checkpointSeq} run=${runId} harness=${harness.alias}`); + } + : 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) { + // Resolve the executor from the registry based on the harness type + const harnessExecutorConfig: HarnessExecutorConfig = { + harnessType: harness.harnessType, + endpointUrl: harness.endpointUrl, + authEnvVar: harness.authEnvVar, + }; + const executor = getExecutor(harnessExecutorConfig); + + if (!executor) { outcome = "failed"; - summary = error instanceof Error ? error.message : String(error); - console.error(`[worker] job=${job.id} session failed:`, error); + summary = `No executor available for harness type: ${harness.harnessType}`; + console.error(`[worker] job=${job.id} ${summary}`); + } else { + const sessionOptions = { + 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, + checkpointEveryNSteps: env.WORKER_CHECKPOINT_EVERY_N_STEPS, + resumeToolHistory, + resumeFilesTouched, + allowedCommands: harness.allowedCommands, + onCheckpoint: onCheckpointSink, + operatorNotes: operatorNotesProvider, + buildPrompt: buildSessionPrompt, + convoyTitle: ctx.convoyTitle, + }; + + try { + const result = await executor.run(sessionContext, modelResult.model, sessionOptions); + summary = result.summary; + tokensTotal = result.tokensTotal; + console.info( + `[worker] job=${job.id} role=${job.roleId} harness=${executor.id} 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); + } } // Draft PR via forge adapter after successful implementer session, before teardown diff --git a/packages/shared/src/worker/executor.ts b/packages/shared/src/worker/executor.ts index 1bc18e4..cd678dc 100644 --- a/packages/shared/src/worker/executor.ts +++ b/packages/shared/src/worker/executor.ts @@ -138,3 +138,23 @@ export interface WorkerExecutor { }, ): Promise<{ summary: string; toolCalls: number; tokensTotal?: number }>; } + +// --- WorkerExecutor types (used by worker dispatch) --- + +export type WorkerRunResult = { + summary: string; + toolCalls: number; + tokensTotal?: number; + outcome: "succeeded" | "failed"; +}; + +export type RemoteRunOptions = { + rolePromptPath: string; + workflowPrompt?: string; + briefAcceptance?: string; + briefFiles?: string[]; + maxSteps: number; + allowedCommands?: readonly string[] | null; + dryRun?: boolean; + convoyTitle?: string; +};