feat(worker): tighten harness executor types + add tests

- Extract WorkerExecutorOptions from inline type in shared executor
- Use WorkerExecutorOptions throughout remote-ag-ui.ts and index.ts
  instead of loose any types
- Extend WorkerExecutor.run return to include outcome
- Add 8 test cases in executor.test.ts covering:
  - Registry dispatch for all harness types
  - Missing config returns null
  - Auth token missing returns failed
  - POST + poll roundtrip behavior
This commit is contained in:
Randall Stillwell 2026-09-04 10:33:40 -05:00
parent 3810cd9dab
commit f6028615e8
4 changed files with 295 additions and 36 deletions

View file

@ -0,0 +1,147 @@
import { describe, expect, it, vi } from "vitest";
import type { WorkerJobContext } from "@tasks/shared/worker";
import { getExecutor } from "./index.js";
const FAKE_CTX: WorkerJobContext = {
workspaceId: "00000000-0000-0000-0000-000000000001",
convoyId: "00000000-0000-0000-0000-000000000002",
roleId: "role-implementer",
briefId: undefined,
repoRoot: "/tmp/fake-repo",
};
const FAKE_OPTS = {
resolvedModel: { modelId: "claude-sonnet-4-5", providerType: "anthropic", alias: "claude-planning" },
resolvedAlias: "claude-planning",
rolePromptPath: "/tmp/role.md",
maxSteps: 5,
buildPrompt: (ctx: WorkerJobContext, title: string) => `Convoy: ${title} Role: ${ctx.roleId}`,
convoyTitle: "test-convoy",
};
describe("getExecutor — harness registry dispatch", () => {
it("returns a minimal executor for harnessType: minimal", () => {
const exec = getExecutor({ harnessType: "minimal" });
expect(exec).not.toBeNull();
expect(exec?.id).toBe("minimal");
expect(exec?.harnessType).toBe("minimal");
});
it("returns a remote-ag-ui executor when endpointUrl + authEnvVar are provided", () => {
const exec = getExecutor({
harnessType: "remote-ag-ui",
endpointUrl: "https://agents.example.com/run",
authEnvVar: "OPENHANDS_API_KEY",
});
expect(exec).not.toBeNull();
expect(exec?.id).toBe("remote-ag-ui");
expect(exec?.harnessType).toBe("remote-ag-ui");
});
it("returns null for remote-ag-ui when endpointUrl is missing", () => {
const exec = getExecutor({ harnessType: "remote-ag-ui", authEnvVar: "X" });
expect(exec).toBeNull();
});
it("returns null for remote-ag-ui when authEnvVar is missing", () => {
const exec = getExecutor({ harnessType: "remote-ag-ui", endpointUrl: "https://x" });
expect(exec).toBeNull();
});
it("returns null for crewai (not yet implemented)", () => {
expect(getExecutor({ harnessType: "crewai" })).toBeNull();
});
it("returns null for langgraph (not yet implemented)", () => {
expect(getExecutor({ harnessType: "langgraph" })).toBeNull();
});
});
describe("minimal executor — runs createMinimalSession", () => {
it("invokes the session with the provided buildPrompt and returns the result", async () => {
const exec = getExecutor({ harnessType: "minimal" });
expect(exec).not.toBeNull();
if (!exec) return;
// We can't easily mock createMinimalSession from this test, so we verify
// the executor exists and has the right shape. The dispatch integration
// is covered by handle-job.test.ts.
expect(typeof exec.run).toBe("function");
});
});
describe("remote-ag-ui executor — POST + poll", () => {
it("returns failed when auth env var is missing", async () => {
const exec = getExecutor({
harnessType: "remote-ag-ui",
endpointUrl: "https://agents.example.com/run",
authEnvVar: "MISSING_TOKEN_VAR",
});
expect(exec).not.toBeNull();
if (!exec) return;
const result = await exec.run(FAKE_CTX, FAKE_OPTS);
expect(result.outcome).toBe("failed");
expect(result.summary).toMatch(/MISSING_TOKEN_VAR/);
});
it("POSTs the task payload and returns succeeded on immediate completion", async () => {
const fetchImpl = vi.fn().mockResolvedValueOnce(
new Response(JSON.stringify({ runId: "run-123", status: "succeeded" }), {
status: 200,
headers: { "Content-Type": "application/json" },
}),
);
const exec = getExecutor(
{ harnessType: "remote-ag-ui", endpointUrl: "https://agents.example.com/run", authEnvVar: "AUTH_TOKEN" },
{ fetchImpl, getEnvVar: () => "test-token" },
);
expect(exec).not.toBeNull();
if (!exec) return;
const result = await exec.run(FAKE_CTX, FAKE_OPTS);
expect(result.outcome).toBe("succeeded");
expect(fetchImpl).toHaveBeenCalledTimes(1);
const [url, init] = fetchImpl.mock.calls[0]!;
expect(url).toBe("https://agents.example.com/run");
expect((init as RequestInit).method).toBe("POST");
expect((init as RequestInit).headers).toMatchObject({ Authorization: "Bearer test-token" });
});
it("polls the status endpoint when initial response is pending", async () => {
const fetchImpl = vi
.fn()
.mockResolvedValueOnce(
new Response(JSON.stringify({ runId: "run-456", status: "pending" }), { status: 200 }),
)
.mockResolvedValueOnce(
new Response(JSON.stringify({ status: "pending" }), { status: 200 }),
)
.mockResolvedValueOnce(
new Response(
JSON.stringify({ status: "succeeded", summary: "done", toolCalls: 3, tokensTotal: 1000 }),
{ status: 200 },
),
);
const exec = getExecutor(
{ harnessType: "remote-ag-ui", endpointUrl: "https://agents.example.com/run", authEnvVar: "AUTH_TOKEN" },
{ fetchImpl, getEnvVar: () => "test-token" },
);
expect(exec).not.toBeNull();
if (!exec) return;
// Use a short poll interval by passing a custom executor — but we can't
// easily here. Just verify it eventually returns and the status URL was hit.
const result = await exec.run(FAKE_CTX, FAKE_OPTS);
expect(result.outcome).toBe("succeeded");
expect(result.summary).toBe("done");
expect(result.toolCalls).toBe(3);
expect(result.tokensTotal).toBe(1000);
// 1 initial POST + at least 1 status poll
expect(fetchImpl.mock.calls.length).toBeGreaterThanOrEqual(2);
});
});

View file

@ -1,4 +1,9 @@
import type { HarnessType, WorkerExecutor, WorkerJobContext } from "@tasks/shared/worker";
import type {
HarnessType,
WorkerExecutor,
WorkerExecutorOptions,
WorkerJobContext,
} from "@tasks/shared/worker";
import { createMinimalSession } from "../session/minimal-session.js";
import { createRemoteAgUiExecutor } from "./remote-ag-ui.js";
@ -11,9 +16,9 @@ function buildMinimalExecutor(): WorkerExecutor {
return {
id: "minimal",
harnessType: "minimal" as HarnessType,
async run(ctx: WorkerJobContext, opts: any): Promise<any> {
async run(ctx: WorkerJobContext, opts: WorkerExecutorOptions): Promise<{ summary: string; toolCalls: number; tokensTotal?: number }> {
const session = createMinimalSession(ctx, {
resolvedModel: opts.resolvedModel,
resolvedModel: opts.resolvedModel as never,
resolvedAlias: opts.resolvedAlias,
rolePromptPath: opts.rolePromptPath,
workflowPrompt: opts.workflowPrompt,
@ -22,18 +27,17 @@ function buildMinimalExecutor(): WorkerExecutor {
maxSteps: opts.maxSteps,
dryRun: opts.dryRun,
checkpointEveryNSteps: opts.checkpointEveryNSteps,
resumeToolHistory: opts.resumeToolHistory,
resumeFilesTouched: opts.resumeFilesTouched,
resumeToolHistory: opts.resumeToolHistory as never,
resumeFilesTouched: opts.resumeFilesTouched as never,
allowedCommands: opts.allowedCommands,
onCheckpoint: opts.onCheckpoint,
onCheckpoint: opts.onCheckpoint as never,
});
const convoyTitle = opts.convoyTitle ?? "unknown";
const result = await session.run(opts.buildPrompt ? opts.buildPrompt(ctx, convoyTitle) : "");
const result = await session.run(opts.buildPrompt(ctx, convoyTitle));
return {
summary: result.summary,
toolCalls: result.toolCalls,
tokensTotal: result.tokensTotal,
outcome: "succeeded",
};
},
};

View file

@ -1,6 +1,7 @@
import type {
HarnessType,
WorkerExecutor,
WorkerExecutorOptions,
WorkerJobContext,
} from "@tasks/shared/worker";
@ -14,6 +15,18 @@ export type RemoteAgUiExecutorDeps = {
getEnvVar: (name: string) => string | undefined;
};
export type RemoteAgUiRunOptions = WorkerExecutorOptions;
export type RemoteStatusResponse = {
status?: string;
summary?: string;
toolCalls?: number;
tokensTotal?: number;
};
const DEFAULT_TIMEOUT_MS = 300_000; // 5 min
const DEFAULT_POLL_INTERVAL_MS = 5_000;
export function createRemoteAgUiExecutor(
config: RemoteAgUiConfig,
deps: RemoteAgUiExecutorDeps,
@ -21,7 +34,7 @@ export function createRemoteAgUiExecutor(
return {
id: "remote-ag-ui",
harnessType: "remote-ag-ui" as HarnessType,
async run(ctx: WorkerJobContext, opts: any): Promise<any> {
async run(ctx: WorkerJobContext, opts: WorkerExecutorOptions): Promise<{ summary: string; toolCalls: number; tokensTotal?: number }> {
const authToken = deps.getEnvVar(config.authEnvVar);
if (!authToken) {
return {
@ -30,28 +43,118 @@ export function createRemoteAgUiExecutor(
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") : "",
model: opts.resolvedModel.modelId,
providerType: opts.resolvedModel.providerType,
modelAlias: opts.resolvedModel.alias,
prompt: opts.buildPrompt(ctx, opts.convoyTitle ?? "unknown"),
workflowPrompt: opts.workflowPrompt ?? null,
briefAcceptance: opts.briefAcceptance ?? null,
briefFiles: opts.briefFiles ?? null,
repoRoot: ctx.repoRoot,
maxSteps: opts.maxSteps ?? 10,
maxSteps: opts.maxSteps,
allowedCommands: opts.allowedCommands ?? null,
};
let postResponse: Response;
try {
const res = await deps.fetchImpl(config.endpointUrl, {
postResponse = await deps.fetchImpl(config.endpointUrl, {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: `Bearer ${authToken}` },
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" };
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
return {
summary: `Failed to POST to remote agent: ${message}`,
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" };
if (!postResponse.ok) {
return {
summary: `Remote agent rejected task: HTTP ${postResponse.status}`,
toolCalls: 0,
outcome: "failed",
};
}
// Parse the initial response — expect a { runId, status } shape
let runId: string | null = null;
try {
const body = (await postResponse.json()) as { runId?: string; status?: string };
runId = body.runId ?? null;
if (body.status === "completed" || body.status === "succeeded") {
return {
summary: "Completed synchronously",
toolCalls: 0,
outcome: "succeeded",
};
}
if (body.status === "failed") {
return {
summary: "Failed synchronously",
toolCalls: 0,
outcome: "failed",
};
}
} catch {
// Non-JSON response; assume fire-and-forget
}
// Poll the status endpoint until completion
const baseUrl = config.endpointUrl.endsWith("/")
? config.endpointUrl.slice(0, -1)
: config.endpointUrl;
const statusUrl = runId
? `${baseUrl}/runs/${runId}/status`
: `${baseUrl}/status`;
const startTime = Date.now();
while (Date.now() - startTime < DEFAULT_TIMEOUT_MS) {
await new Promise((r) => setTimeout(r, DEFAULT_POLL_INTERVAL_MS));
try {
const statusResponse = await deps.fetchImpl(statusUrl, {
method: "GET",
headers: { Authorization: `Bearer ${authToken}` },
});
if (!statusResponse.ok) continue;
const status = (await statusResponse.json()) as RemoteStatusResponse;
if (status.status === "completed" || status.status === "succeeded") {
return {
summary: status.summary ?? "Remote agent completed",
toolCalls: status.toolCalls ?? 0,
tokensTotal: status.tokensTotal,
outcome: "succeeded",
};
}
if (status.status === "failed") {
return {
summary: status.summary ?? "Remote agent failed",
toolCalls: status.toolCalls ?? 0,
tokensTotal: status.tokensTotal,
outcome: "failed",
};
}
} catch {
// Continue polling on transient errors
}
}
return {
summary: "Remote agent timed out",
toolCalls: 0,
outcome: "failed",
};
},
};
}

View file

@ -115,12 +115,8 @@ export type ResolvedModelLike = {
alias: string;
};
export interface WorkerExecutor {
id: string;
harnessType: HarnessType;
run(
ctx: WorkerJobContext,
opts: {
/** Options passed to WorkerExecutor.run — extracted for reuse across executors. */
export type WorkerExecutorOptions = {
resolvedModel: ResolvedModelLike;
resolvedAlias: string;
rolePromptPath: string;
@ -135,8 +131,17 @@ export interface WorkerExecutor {
allowedCommands?: readonly string[] | null;
onCheckpoint?: (input: unknown) => Promise<void>;
buildPrompt: (ctx: WorkerJobContext, title: string) => string;
},
): Promise<{ summary: string; toolCalls: number; tokensTotal?: number }>;
/** Optional: used by remote executors that need the convoy title for prompt building. */
convoyTitle?: string;
};
export interface WorkerExecutor {
id: string;
harnessType: HarnessType;
run(
ctx: WorkerJobContext,
opts: WorkerExecutorOptions,
): Promise<{ summary: string; toolCalls: number; tokensTotal?: number; outcome?: "succeeded" | "failed" }>;
}
// --- WorkerExecutor types (used by worker dispatch) ---