863 lines
28 KiB
TypeScript
863 lines
28 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { execFile } from "node:child_process";
|
|
import { access, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import * as path from "node:path";
|
|
import test from "node:test";
|
|
import { promisify } from "node:util";
|
|
import type {
|
|
DefaultResourceLoader,
|
|
ExtensionContext,
|
|
SettingsManager,
|
|
} from "@earendil-works/pi-coding-agent";
|
|
import { emptyUsage } from "../_workflows/model.ts";
|
|
import type {
|
|
RunAgentOptions,
|
|
ThinkingLevel,
|
|
WorkflowModel,
|
|
} from "../_workflows/runner.ts";
|
|
import { DEFAULT_WORKFLOW_CONFIG, type WorkflowRun } from "./domain.ts";
|
|
import { WorkflowEngine, type WorkspaceManager } from "./engine.ts";
|
|
import {
|
|
PiAssignmentExecutor,
|
|
ProcessAssignmentScheduler,
|
|
resolveWorkflowModel,
|
|
} from "./execution.ts";
|
|
import { loadAgentProfiles } from "./profiles.ts";
|
|
import { WorkflowStore } from "./store.ts";
|
|
|
|
const execFileAsync = promisify(execFile);
|
|
|
|
async function withTempDir(run: (directory: string) => Promise<void>) {
|
|
const directory = await mkdtemp(path.join(tmpdir(), "pi-workflow-execution-"));
|
|
try {
|
|
await run(directory);
|
|
} finally {
|
|
await rm(directory, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
function model(provider: string, id: string) {
|
|
return {
|
|
provider,
|
|
id,
|
|
contextWindow: 128_000,
|
|
} as unknown as WorkflowModel;
|
|
}
|
|
|
|
function registry(models: WorkflowModel[]) {
|
|
return {
|
|
find(provider: string, id: string) {
|
|
return models.find((candidate) => candidate.provider === provider && candidate.id === id);
|
|
},
|
|
getAll() {
|
|
return models;
|
|
},
|
|
} as unknown as ExtensionContext["modelRegistry"];
|
|
}
|
|
|
|
function deferred() {
|
|
let resolve: (() => void) | undefined;
|
|
const promise = new Promise<void>((done) => {
|
|
resolve = done;
|
|
});
|
|
return { promise, resolve: () => resolve?.() };
|
|
}
|
|
|
|
test("process scheduler enforces aggregate and per-run capacity with queued cancellation", async () => {
|
|
const scheduler = new ProcessAssignmentScheduler();
|
|
scheduler.lowerCeiling(2);
|
|
const waits = [deferred(), deferred(), deferred(), deferred()];
|
|
const controllers = waits.map(() => new AbortController());
|
|
let active = 0;
|
|
let maximum = 0;
|
|
const tasks = waits.map((wait, index) =>
|
|
scheduler.schedule({
|
|
runId: index < 3 ? "run-a" : "run-b",
|
|
runLimit: 2,
|
|
signal: controllers[index].signal,
|
|
task: async () => {
|
|
active++;
|
|
maximum = Math.max(maximum, active);
|
|
try {
|
|
await wait.promise;
|
|
return index;
|
|
} finally {
|
|
active--;
|
|
}
|
|
},
|
|
}),
|
|
);
|
|
const cancelled = tasks[2].catch((error: unknown) => error);
|
|
assert.deepEqual(scheduler.snapshot(), {
|
|
ceiling: 2,
|
|
inUse: 2,
|
|
running: 0,
|
|
queued: 2,
|
|
});
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
assert.deepEqual(scheduler.snapshot(), {
|
|
ceiling: 2,
|
|
inUse: 2,
|
|
running: 2,
|
|
queued: 2,
|
|
});
|
|
controllers[2].abort(new Error("cancel queued"));
|
|
assert.match(String(await cancelled), /cancel queued/);
|
|
waits[0].resolve();
|
|
waits[1].resolve();
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
waits[3].resolve();
|
|
assert.deepEqual(await Promise.all([tasks[0], tasks[1], tasks[3]]), [0, 1, 3]);
|
|
assert.equal(maximum, 2);
|
|
assert.deepEqual(scheduler.snapshot(), {
|
|
ceiling: 2,
|
|
inUse: 0,
|
|
running: 0,
|
|
queued: 0,
|
|
});
|
|
});
|
|
|
|
test("exclusive mutation reservations serialize a run without blocking other runs", async () => {
|
|
const scheduler = new ProcessAssignmentScheduler();
|
|
scheduler.lowerCeiling(3);
|
|
const waits = [deferred(), deferred(), deferred(), deferred()];
|
|
const started: number[] = [];
|
|
const schedule = (index: number, runId: string, exclusive = false) =>
|
|
scheduler.schedule({
|
|
runId,
|
|
runLimit: 3,
|
|
exclusive,
|
|
signal: new AbortController().signal,
|
|
async task() {
|
|
started.push(index);
|
|
await waits[index].promise;
|
|
},
|
|
});
|
|
const firstRead = schedule(0, "run-a");
|
|
const mutation = schedule(1, "run-a", true);
|
|
const laterRead = schedule(2, "run-a");
|
|
const otherRun = schedule(3, "run-b");
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
assert.deepEqual(started, [0, 3]);
|
|
waits[0].resolve();
|
|
await firstRead;
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
assert.deepEqual(started, [0, 3, 1]);
|
|
waits[1].resolve();
|
|
await mutation;
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
assert.deepEqual(started, [0, 3, 1, 2]);
|
|
waits[2].resolve();
|
|
waits[3].resolve();
|
|
await Promise.all([laterRead, otherRun]);
|
|
});
|
|
|
|
test("running cancellation releases a scheduler reservation exactly once", async () => {
|
|
const scheduler = new ProcessAssignmentScheduler();
|
|
const controller = new AbortController();
|
|
let reserved = false;
|
|
let started = false;
|
|
const task = scheduler.schedule({
|
|
runId: "run-a",
|
|
runLimit: 1,
|
|
signal: controller.signal,
|
|
async onReserved() {
|
|
reserved = true;
|
|
},
|
|
async task(signal) {
|
|
started = true;
|
|
return new Promise<never>((_resolve, reject) => {
|
|
signal.addEventListener("abort", () => reject(signal.reason), {
|
|
once: true,
|
|
});
|
|
});
|
|
},
|
|
});
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
controller.abort(new Error("cancel running"));
|
|
await assert.rejects(task, /cancel running/);
|
|
assert.equal(reserved, true);
|
|
assert.equal(started, true);
|
|
assert.deepEqual(scheduler.snapshot(), {
|
|
ceiling: 32,
|
|
inUse: 0,
|
|
running: 0,
|
|
queued: 0,
|
|
});
|
|
});
|
|
|
|
test("model resolution is exact, inherited-provider aware, and ambiguity safe", () => {
|
|
const inherited = model("provider-a", "parent");
|
|
const sharedA = model("provider-a", "shared");
|
|
const sharedB = model("provider-b", "shared");
|
|
const models = registry([inherited, sharedA, sharedB]);
|
|
assert.equal(resolveWorkflowModel(models, {}, inherited), inherited);
|
|
assert.throws(
|
|
() => resolveWorkflowModel(registry([]), {}, inherited),
|
|
/Unknown inherited model/,
|
|
);
|
|
assert.equal(resolveWorkflowModel(models, { model: "shared" }, inherited), sharedA);
|
|
assert.equal(
|
|
resolveWorkflowModel(models, { provider: "provider-b", model: "shared" }, inherited),
|
|
sharedB,
|
|
);
|
|
assert.throws(
|
|
() => resolveWorkflowModel(models, { model: "shared" }, undefined),
|
|
/multiple providers/,
|
|
);
|
|
assert.throws(
|
|
() => resolveWorkflowModel(models, { provider: "provider-a" }, inherited),
|
|
/requires a model/,
|
|
);
|
|
});
|
|
|
|
test("the immutable process ceiling caps aggregate reservations at 32", async () => {
|
|
const scheduler = new ProcessAssignmentScheduler();
|
|
const waits = Array.from({ length: 40 }, deferred);
|
|
let maximum = 0;
|
|
let active = 0;
|
|
const tasks = waits.map((wait, index) =>
|
|
scheduler.schedule({
|
|
runId: `run-${index % 4}`,
|
|
runLimit: 32,
|
|
signal: new AbortController().signal,
|
|
task: async () => {
|
|
active++;
|
|
maximum = Math.max(maximum, active);
|
|
try {
|
|
await wait.promise;
|
|
} finally {
|
|
active--;
|
|
}
|
|
},
|
|
}),
|
|
);
|
|
assert.equal(scheduler.snapshot().inUse, 32);
|
|
assert.equal(scheduler.snapshot().queued, 8);
|
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
|
waits.forEach((wait) => wait.resolve());
|
|
await Promise.all(tasks);
|
|
assert.equal(maximum, 32);
|
|
assert.equal(scheduler.snapshot().inUse, 0);
|
|
});
|
|
|
|
test("production executor resolves trust and creates fresh structured child runs", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const profiles = (
|
|
await loadAgentProfiles({ cwd: directory, projectTrusted: false })
|
|
).profiles;
|
|
const profile = profiles.find((candidate) => candidate.roles.includes("discover"));
|
|
assert.ok(profile);
|
|
const inherited = model("provider-a", "parent");
|
|
const models = registry([inherited]);
|
|
const scheduler = new ProcessAssignmentScheduler();
|
|
let resources = 0;
|
|
let sessions = 0;
|
|
const trustedChildren: string[] = [];
|
|
const observedTools: string[][] = [];
|
|
const observedCustomTools: string[][] = [];
|
|
const fakeResources = {
|
|
loader: {} as DefaultResourceLoader,
|
|
settingsManager: {} as SettingsManager,
|
|
};
|
|
const fakeRunAgent = async (options: RunAgentOptions) => {
|
|
sessions++;
|
|
observedTools.push(options.tools ?? []);
|
|
observedCustomTools.push(
|
|
options.customTools?.map((tool) => tool.name) ?? [],
|
|
);
|
|
options.onSessionCreated?.({
|
|
childSessionId: `child-${sessions}`,
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
});
|
|
return {
|
|
ok: true,
|
|
output: "",
|
|
structured: {
|
|
questions: [],
|
|
candidatePaths: ["src/index.ts"],
|
|
languages: ["typescript"],
|
|
tags: ["backend"],
|
|
riskTags: [],
|
|
},
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: `child-${sessions}`,
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [{ role: "assistant" as const, text: "done" }],
|
|
};
|
|
};
|
|
const executor = new PiAssignmentExecutor({
|
|
modelRegistry: models,
|
|
parentCwd: directory,
|
|
parentProjectTrusted: true,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 4,
|
|
runConcurrency: 2,
|
|
scheduler,
|
|
runAgent: fakeRunAgent,
|
|
async createResources() {
|
|
resources++;
|
|
return fakeResources;
|
|
},
|
|
resolveProjectTrust(options) {
|
|
trustedChildren.push(options.childCwd);
|
|
return false;
|
|
},
|
|
});
|
|
const makeRequest = (runId: string) => ({
|
|
runId,
|
|
role: "discover" as const,
|
|
request: "Inspect the repository",
|
|
profile,
|
|
facts: {
|
|
stage: "discover" as const,
|
|
taskKind: "feature" as const,
|
|
requestTerms: ["inspect"],
|
|
candidatePaths: [],
|
|
languages: [],
|
|
tags: [],
|
|
riskTags: [],
|
|
},
|
|
decisions: [],
|
|
cwd: path.join(directory, "worktree"),
|
|
signal: new AbortController().signal,
|
|
});
|
|
const first = await executor.run(makeRequest("run-a"));
|
|
const second = await executor.run(makeRequest("run-b"));
|
|
await executor.run({
|
|
...makeRequest("run-c"),
|
|
allowedReadTools: ["grep"],
|
|
});
|
|
assert.equal(first.kind, "pi-workflow-assignment-result");
|
|
assert.equal(first.metadata.childSessionId, "child-1");
|
|
assert.equal(second.metadata.childSessionId, "child-2");
|
|
assert.equal(resources, 3);
|
|
assert.equal(sessions, 3);
|
|
assert.deepEqual(trustedChildren, [
|
|
path.join(directory, "worktree"),
|
|
path.join(directory, "worktree"),
|
|
path.join(directory, "worktree"),
|
|
]);
|
|
assert.ok(observedTools.every((tools) => tools.includes("structured_output")));
|
|
assert.equal(observedTools[2]?.includes("read"), false);
|
|
assert.equal(observedTools[2]?.includes("grep"), true);
|
|
assert.deepEqual(observedCustomTools[2], ["grep"]);
|
|
|
|
const failedExecutor = new PiAssignmentExecutor({
|
|
modelRegistry: models,
|
|
parentCwd: directory,
|
|
parentProjectTrusted: true,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 4,
|
|
runConcurrency: 2,
|
|
scheduler: new ProcessAssignmentScheduler(),
|
|
async createResources() {
|
|
return fakeResources;
|
|
},
|
|
async runAgent(options) {
|
|
options.onSessionCreated?.({
|
|
childSessionId: "child-failed",
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
});
|
|
return {
|
|
ok: false,
|
|
output: "partial",
|
|
error: "provider failed",
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: "child-failed",
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [{ role: "assistant" as const, text: "partial" }],
|
|
};
|
|
},
|
|
});
|
|
await assert.rejects(
|
|
failedExecutor.run(makeRequest("run-failed")),
|
|
(error: unknown) => {
|
|
assert.ok(error instanceof Error);
|
|
assert.equal(error.message, "provider failed");
|
|
assert.equal(
|
|
(error as Error & { assignmentMetadata?: { childSessionId?: string } })
|
|
.assignmentMetadata?.childSessionId,
|
|
"child-failed",
|
|
);
|
|
return true;
|
|
},
|
|
);
|
|
});
|
|
});
|
|
|
|
test("mutating children receive guarded file tools without process execution", async () => {
|
|
await withTempDir(async (directory) => {
|
|
await execFileAsync("git", ["init", "-q", directory]);
|
|
await execFileAsync("git", ["-C", directory, "config", "user.email", "test@example.com"]);
|
|
await execFileAsync("git", ["-C", directory, "config", "user.name", "Test User"]);
|
|
await mkdir(path.join(directory, "src"), { recursive: true });
|
|
await writeFile(path.join(directory, "src", "existing.ts"), "export {};\n", "utf8");
|
|
await execFileAsync("git", ["-C", directory, "add", "."]);
|
|
await execFileAsync("git", ["-C", directory, "commit", "-qm", "initial"]);
|
|
const profiles = (
|
|
await loadAgentProfiles({ cwd: directory, projectTrusted: false })
|
|
).profiles;
|
|
const profile = profiles.find((candidate) => candidate.roles.includes("implement"));
|
|
assert.ok(profile);
|
|
const inherited = model("provider-a", "parent");
|
|
const executor = new PiAssignmentExecutor({
|
|
modelRegistry: registry([inherited]),
|
|
parentCwd: directory,
|
|
parentProjectTrusted: false,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 2,
|
|
runConcurrency: 1,
|
|
scheduler: new ProcessAssignmentScheduler(),
|
|
async createResources() {
|
|
return {
|
|
loader: {} as DefaultResourceLoader,
|
|
settingsManager: {} as SettingsManager,
|
|
};
|
|
},
|
|
async runAgent(options) {
|
|
assert.deepEqual(options.tools, [
|
|
"read",
|
|
"grep",
|
|
"find",
|
|
"ls",
|
|
"edit",
|
|
"write",
|
|
"structured_output",
|
|
]);
|
|
assert.equal(options.tools?.includes("bash"), false);
|
|
const writeTool = options.customTools?.find((tool) => tool.name === "write");
|
|
assert.ok(writeTool);
|
|
await writeTool.execute(
|
|
"write-1",
|
|
{ path: "src/generated.ts", content: "export const generated = true;\n" },
|
|
options.signal ?? new AbortController().signal,
|
|
undefined,
|
|
{} as ExtensionContext,
|
|
);
|
|
return {
|
|
ok: true,
|
|
output: "",
|
|
structured: {
|
|
plan: "# Plan",
|
|
uiImpact: "none",
|
|
uiImpactEvidence: ["No user-facing change"],
|
|
affectedUiSurfaces: [],
|
|
assignments: [],
|
|
},
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: "mutating-child",
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [],
|
|
};
|
|
},
|
|
});
|
|
const result = await executor.run({
|
|
runId: "run-mutation",
|
|
role: "design",
|
|
request: "Generate an owned file",
|
|
profile,
|
|
facts: {
|
|
stage: "design",
|
|
taskKind: "feature",
|
|
requestTerms: ["generate"],
|
|
candidatePaths: ["src/generated.ts"],
|
|
languages: ["typescript"],
|
|
tags: [],
|
|
riskTags: [],
|
|
},
|
|
decisions: [],
|
|
cwd: directory,
|
|
signal: new AbortController().signal,
|
|
mutates: true,
|
|
ownedPaths: ["src"],
|
|
});
|
|
assert.deepEqual(result.metadata.gitAfter?.changedPaths, ["src/generated.ts"]);
|
|
|
|
const prototypePath = path.join(directory, "prototypes", "candidate-1");
|
|
await mkdir(prototypePath, { recursive: true });
|
|
await writeFile(path.join(prototypePath, "route.ts"), "prototype\n", "utf8");
|
|
const cleanupExecutor = new PiAssignmentExecutor({
|
|
modelRegistry: registry([inherited]),
|
|
parentCwd: directory,
|
|
parentProjectTrusted: false,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 2,
|
|
runConcurrency: 1,
|
|
scheduler: new ProcessAssignmentScheduler(),
|
|
async createResources() {
|
|
return {
|
|
loader: {} as DefaultResourceLoader,
|
|
settingsManager: {} as SettingsManager,
|
|
};
|
|
},
|
|
async runAgent(options) {
|
|
assert.equal(options.tools?.includes("workflow_remove"), true);
|
|
const removeTool = options.customTools?.find(
|
|
(tool) => tool.name === "workflow_remove",
|
|
);
|
|
assert.ok(removeTool);
|
|
await removeTool.execute(
|
|
"remove-prototype",
|
|
{ path: "prototypes/candidate-1" },
|
|
options.signal ?? new AbortController().signal,
|
|
undefined,
|
|
{} as ExtensionContext,
|
|
);
|
|
return {
|
|
ok: true,
|
|
output: "",
|
|
structured: {
|
|
summary: "Removed prototype-only route",
|
|
changedPaths: ["prototypes/candidate-1"],
|
|
focusedTests: [],
|
|
},
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: "cleanup-child",
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [],
|
|
};
|
|
},
|
|
});
|
|
await cleanupExecutor.run({
|
|
runId: "run-cleanup",
|
|
role: "remediate",
|
|
request: "Remove prototype-only routes",
|
|
profile,
|
|
facts: {
|
|
stage: "implement",
|
|
taskKind: "feature",
|
|
requestTerms: ["cleanup"],
|
|
candidatePaths: ["prototypes/candidate-1"],
|
|
languages: ["typescript"],
|
|
tags: ["frontend"],
|
|
riskTags: [],
|
|
},
|
|
decisions: [],
|
|
cwd: directory,
|
|
signal: new AbortController().signal,
|
|
mutates: true,
|
|
ownedPaths: ["prototypes/candidate-1"],
|
|
});
|
|
await assert.rejects(access(prototypePath));
|
|
|
|
const reviewProfile = profiles.find((candidate) => candidate.roles.includes("review"));
|
|
assert.ok(reviewProfile);
|
|
const reviewRequest = {
|
|
runId: "run-review",
|
|
role: "review" as const,
|
|
request: "Review the feature",
|
|
profile: reviewProfile,
|
|
facts: {
|
|
stage: "review" as const,
|
|
taskKind: "feature" as const,
|
|
requestTerms: ["review"],
|
|
candidatePaths: ["src/generated.ts"],
|
|
languages: ["typescript"],
|
|
tags: [],
|
|
riskTags: [],
|
|
},
|
|
decisions: [],
|
|
cwd: directory,
|
|
signal: new AbortController().signal,
|
|
captureGitState: true,
|
|
approvedPlan: "# Approved plan",
|
|
verificationEvidence: {
|
|
commands: [],
|
|
focusedBehavioralTests: true,
|
|
verifierSucceeded: true,
|
|
},
|
|
reviewDiff: "a😀b",
|
|
};
|
|
const reviewExecutor = (
|
|
readCompleteDiff: boolean,
|
|
readCompletePlan = readCompleteDiff,
|
|
) =>
|
|
new PiAssignmentExecutor({
|
|
modelRegistry: registry([inherited]),
|
|
parentCwd: directory,
|
|
parentProjectTrusted: false,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 2,
|
|
runConcurrency: 1,
|
|
scheduler: new ProcessAssignmentScheduler(),
|
|
async createResources() {
|
|
return {
|
|
loader: {} as DefaultResourceLoader,
|
|
settingsManager: {} as SettingsManager,
|
|
};
|
|
},
|
|
async runAgent(options) {
|
|
assert.equal(options.tools?.includes("workflow_diff"), true);
|
|
assert.equal(options.tools?.includes("workflow_plan"), true);
|
|
assert.equal(
|
|
options.customTools?.some((tool) => tool.name === "workflow_diff"),
|
|
true,
|
|
);
|
|
assert.equal(
|
|
options.customTools?.some((tool) => tool.name === "workflow_plan"),
|
|
true,
|
|
);
|
|
const diffTool = options.customTools?.find(
|
|
(tool) => tool.name === "workflow_diff",
|
|
);
|
|
assert.ok(diffTool);
|
|
const executeTool = diffTool.execute as unknown as (
|
|
id: string,
|
|
params: { offset: number; limit: number },
|
|
) => Promise<{ content: Array<{ type: string; text: string }> }>;
|
|
const unicodePage = await executeTool("unicode-page", {
|
|
offset: 1,
|
|
limit: 1,
|
|
});
|
|
assert.equal(unicodePage.content[0]?.text, "😀");
|
|
return {
|
|
ok: true,
|
|
output: "",
|
|
structured: {
|
|
specification: { verdict: "approved", findings: [] },
|
|
standards: { verdict: "approved", findings: [] },
|
|
},
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: "review-child",
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [
|
|
...(readCompletePlan
|
|
? [
|
|
{
|
|
role: "tool" as const,
|
|
name: "workflow_plan",
|
|
toolCallId: "plan-call",
|
|
text: '{"offset":0,"limit":49152}',
|
|
},
|
|
{
|
|
role: "toolResult" as const,
|
|
name: "workflow_plan",
|
|
toolCallId: "plan-call",
|
|
text: "plan page",
|
|
isError: false,
|
|
},
|
|
]
|
|
: []),
|
|
...(readCompleteDiff
|
|
? [
|
|
{
|
|
role: "tool" as const,
|
|
name: "workflow_diff",
|
|
toolCallId: "diff-call",
|
|
text: '{"offset":0,"limit":49152}',
|
|
},
|
|
{
|
|
role: "toolResult" as const,
|
|
name: "workflow_diff",
|
|
toolCallId: "diff-call",
|
|
text: "diff page",
|
|
isError: false,
|
|
},
|
|
]
|
|
: []),
|
|
],
|
|
};
|
|
},
|
|
});
|
|
await reviewExecutor(true).run(reviewRequest);
|
|
await assert.rejects(
|
|
reviewExecutor(false).run(reviewRequest),
|
|
/did not read the complete workflow diff/,
|
|
);
|
|
await assert.rejects(
|
|
reviewExecutor(true, false).run(reviewRequest),
|
|
/did not read the complete approved workflow plan/,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("production executor drives Discover and Design through Plan Approval", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const profiles = (
|
|
await loadAgentProfiles({ cwd: directory, projectTrusted: false })
|
|
).profiles;
|
|
const profileName = (role: "implement" | "verify" | "review") => {
|
|
const profile = profiles.find((candidate) => candidate.roles.includes(role));
|
|
assert.ok(profile);
|
|
return profile.name;
|
|
};
|
|
const inherited = model("provider-a", "parent");
|
|
const models = registry([inherited]);
|
|
let sessions = 0;
|
|
const fakeRunAgent = async (options: RunAgentOptions) => {
|
|
sessions++;
|
|
options.onSessionCreated?.({
|
|
childSessionId: `pipeline-child-${sessions}`,
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
});
|
|
const structured = options.prompt.includes("Assignment role: discover")
|
|
? {
|
|
questions: [],
|
|
candidatePaths: ["src/index.ts"],
|
|
languages: ["typescript"],
|
|
tags: ["backend"],
|
|
riskTags: [],
|
|
}
|
|
: {
|
|
plan: "# Plan\n\nImplement and verify the feature.",
|
|
uiImpact: "none",
|
|
uiImpactEvidence: ["No user-facing change"],
|
|
affectedUiSurfaces: [],
|
|
acceptanceCriteria: ["Feature behavior passes focused tests"],
|
|
verificationCommands: [
|
|
{
|
|
executable: "npm",
|
|
args: ["test"],
|
|
ownedPaths: [],
|
|
kind: "test",
|
|
},
|
|
],
|
|
assignments: [
|
|
{
|
|
role: "implement",
|
|
profileName: profileName("implement"),
|
|
rationale: "Implement the approved scope",
|
|
mutates: true,
|
|
ownedPaths: ["src"],
|
|
},
|
|
{
|
|
role: "verify",
|
|
profileName: profileName("verify"),
|
|
rationale: "Run focused verification",
|
|
mutates: false,
|
|
ownedPaths: [],
|
|
},
|
|
{
|
|
role: "review",
|
|
profileName: profileName("review"),
|
|
rationale: "Review independently",
|
|
mutates: false,
|
|
ownedPaths: [],
|
|
},
|
|
],
|
|
};
|
|
return {
|
|
ok: true,
|
|
output: "",
|
|
structured,
|
|
aborted: false,
|
|
usage: emptyUsage(),
|
|
provider: inherited.provider,
|
|
model: inherited.id,
|
|
thinkingLevel: "medium" as ThinkingLevel,
|
|
childSessionId: `pipeline-child-${sessions}`,
|
|
contextWindow: inherited.contextWindow,
|
|
transcript: [{ role: "assistant" as const, text: "done" }],
|
|
};
|
|
};
|
|
const executor = new PiAssignmentExecutor({
|
|
modelRegistry: models,
|
|
parentCwd: directory,
|
|
parentProjectTrusted: false,
|
|
parentModel: inherited,
|
|
parentThinkingLevel: "medium" as ThinkingLevel,
|
|
concurrencyCeiling: 4,
|
|
runConcurrency: 2,
|
|
scheduler: new ProcessAssignmentScheduler(),
|
|
runAgent: fakeRunAgent,
|
|
async createResources() {
|
|
return {
|
|
loader: {} as DefaultResourceLoader,
|
|
settingsManager: {} as SettingsManager,
|
|
};
|
|
},
|
|
});
|
|
const workspace: WorkspaceManager = {
|
|
async prepare(options) {
|
|
return {
|
|
sourceRoot: options.sourceRoot,
|
|
worktreeRoot: path.join(options.sourceRoot, ".worktrees"),
|
|
worktreePath: path.join(options.sourceRoot, ".worktrees", options.runId),
|
|
branch: `feature/${options.featureName}`,
|
|
baseHead: "abc123",
|
|
branchExists: false,
|
|
};
|
|
},
|
|
async create(plan): Promise<WorkflowRun["workspace"]> {
|
|
return {
|
|
state: "ready",
|
|
sourceRoot: plan.sourceRoot,
|
|
root: plan.worktreePath,
|
|
branch: plan.branch,
|
|
baseHead: plan.baseHead,
|
|
};
|
|
},
|
|
async reconcile(current) {
|
|
return current;
|
|
},
|
|
};
|
|
let id = 0;
|
|
const engine = new WorkflowEngine({
|
|
store: new WorkflowStore(path.join(directory, "state")),
|
|
workspace,
|
|
executor,
|
|
sourceRoot: directory,
|
|
parentSessionId: "session-1",
|
|
effectiveConfig: structuredClone(DEFAULT_WORKFLOW_CONFIG),
|
|
profiles,
|
|
now: () => 10_000,
|
|
createId: () => `pipeline-${++id}`,
|
|
processId: 100,
|
|
isProcessAlive: () => true,
|
|
});
|
|
const result = await engine.execute({
|
|
action: "start",
|
|
request: "Build the feature",
|
|
featureName: "build-feature",
|
|
idempotencyKey: "pipeline-start",
|
|
});
|
|
assert.equal(result.openGate?.kind, "plan-approval");
|
|
assert.equal(sessions, 2);
|
|
assert.deepEqual(
|
|
result.assignments
|
|
.filter((assignment) => assignment.childSessionId)
|
|
.map((assignment) => assignment.childSessionId),
|
|
["pipeline-child-1", "pipeline-child-2"],
|
|
);
|
|
assert.deepEqual(
|
|
result.assignments
|
|
.filter((assignment) => assignment.status === "queued")
|
|
.map((assignment) => assignment.role),
|
|
["implement", "verify", "review"],
|
|
);
|
|
});
|
|
});
|