Complete the P3 landing exit test across three SIGKILL boundaries
This commit is contained in:
@@ -14,6 +14,7 @@ export const branchTool = (host: any) => defineTool({
|
||||
let id = await api.memo<number>('branch-id', ctx)
|
||||
if (id === undefined) {
|
||||
const lineage = host.need(LineageApi)
|
||||
const workspaces = input.isolated ? host.need({name:'workspaces'}) : undefined
|
||||
const roots = (await api.snapshot(Roots, ROOT_CONVERSATION_ID, ctx))?.roots ?? []
|
||||
const root = input.root ?? await lineage.rootOf(api.conversationId, ctx) ?? (roots.length === 1 ? roots[0] : undefined)
|
||||
if (root === undefined || !roots.includes(root)) throw new Error('Choose a listed root')
|
||||
@@ -21,9 +22,13 @@ export const branchTool = (host: any) => defineTool({
|
||||
const coordinator = definitions.find((definition: any) => definition.name === 'coordinator')
|
||||
if (!coordinator) throw new Error('Missing coordinator definition')
|
||||
const rootAgent = await (await host.harness().conversation(root, ctx))!.agent(ctx)
|
||||
const branch = await lineage.branch({from: root, repo: input.repo, prompt: input.prompt, title: input.title, isolated: input.isolated ?? false, deliver: false, by: api.conversationId, receipt: {taskId: api.taskId, callId: api.callId}}, ctx)
|
||||
const branch = await lineage.branch({from: root, repo: input.repo, prompt: input.prompt, title: input.title, deliver: false, by: api.conversationId, receipt: {taskId: api.taskId, callId: api.callId}}, ctx)
|
||||
const existing = await workspaces?.get(branch.id, ctx)
|
||||
const cwd = existing ? (await (await host.harness().conversation(branch.id, ctx)).agent(ctx)).cwd : input.repo
|
||||
const ancestors = workspaces ? (await workspaces.chain(branch.id, ctx)).slice(1) : []
|
||||
await host.harness().commit(async (tx: any) => {
|
||||
await configureChild(host, tx, branch.id, {definition: coordinator, parentAgent: rootAgent, cwd: input.repo})
|
||||
await configureChild(host, tx, branch.id, {definition: coordinator, parentAgent: rootAgent, cwd})
|
||||
if (workspaces && !existing) await workspaces.create(tx, branch.id, input.repo, ancestors)
|
||||
;(await tx.doc(host.metadata, branch.id)).label = 'coordinator'
|
||||
}, ctx)
|
||||
id = await api.memo('branch-id', branch.id, ctx)
|
||||
|
||||
+88
-55
@@ -31684,7 +31684,7 @@ function answerFor(rules, transcript) {
|
||||
const last = transcript.messages.findLast((message) => message.role !== "system");
|
||||
const rule = rules.find((rule2) => rule2.role === role && (!rule2.match.kind || rule2.match.kind === last?.role) && textOf(last?.content).includes(rule2.match.text));
|
||||
if (!rule) throw new Error(`No scripted rule for ${role}: ${JSON.stringify(last)}`);
|
||||
return { rule: rule.id, answer: rule.answer };
|
||||
return { rule: rule.id, answer: typeof rule.answer === "function" ? rule.answer(transcript) : rule.answer };
|
||||
}
|
||||
function scriptedModel(rules, report2) {
|
||||
const faux = fauxProvider({ tokensPerSecond: 1e5 });
|
||||
@@ -31707,6 +31707,11 @@ function testReplay() {
|
||||
const first = answerFor(rules, { messages: [system, input] });
|
||||
const replay = answerFor(JSON.parse(JSON.stringify(rules)), { messages: [system, { role: "assistant", content: [] }, input] });
|
||||
if (JSON.stringify(first) !== JSON.stringify(replay)) throw new Error("Scripted replay changed the answer");
|
||||
const dynamic = [{ id: "land", role: "worker", match: { text: "build" }, answer: (transcript) => ({ tools: [{ name: "land", args: { id: /Started subagent (\d+)/.exec(JSON.stringify(transcript.messages))?.[1] } }] }) }];
|
||||
const history = { role: "toolResult", content: "Started subagent 42." };
|
||||
const selected = answerFor(dynamic, { messages: [system, history, input] });
|
||||
const restarted = answerFor(dynamic, { messages: [system, { role: "assistant", content: [] }, history, input] });
|
||||
if (JSON.stringify(selected) !== JSON.stringify(restarted) || selected.answer.tools?.[0].args.id !== "42") throw new Error("Dynamic tool identity changed on replay");
|
||||
const toolRules = [{ id: "worker-done", role: "worker", match: { text: "committed" }, answer: { text: "done" } }];
|
||||
if (answerFor(toolRules, { messages: [system, { role: "toolResult", content: [{ type: "text", text: "committed" }] }] }).rule !== "worker-done") throw new Error("Tool result rule did not match");
|
||||
}
|
||||
@@ -31717,14 +31722,17 @@ var assert = (ok, message) => {
|
||||
};
|
||||
testReplay();
|
||||
if (globalThis.p3Mode !== "unit") {
|
||||
const workerId = (transcript) => Number(/Started subagent (\d+)/.exec(JSON.stringify(transcript.messages))?.[1]);
|
||||
const rules = [
|
||||
{ id: "ultimate-branch", role: "ultimate", match: { kind: "user", text: "launch branch" }, answer: { tools: [{ name: "branch", args: { repo: "/src/demo", prompt: "branch probe" } }] } },
|
||||
{ id: "ultimate-branch", role: "ultimate", match: { kind: "user", text: "launch scenario" }, answer: { tools: [{ name: "branch", args: { repo: "/src/demo", isolated: true, prompt: "begin" } }] } },
|
||||
{ id: "ultimate-probe", role: "ultimate", match: { kind: "user", text: "launch branch" }, answer: { tools: [{ name: "branch", args: { repo: "/src/demo", prompt: "branch probe" } }] } },
|
||||
{ id: "coordinator-probe", role: "coordinator", match: { kind: "user", text: "branch probe" }, answer: { text: "branch complete" } },
|
||||
{ id: "ultimate-receipt", role: "ultimate", match: { text: "" }, answer: { text: "Receipt noted." } },
|
||||
{ id: "coordinator-echo", role: "coordinator", match: { kind: "user", text: "Receipt noted." }, answer: { text: "Acknowledged." } },
|
||||
{ id: "start-worker", role: "coordinator", match: { kind: "user", text: "begin" }, answer: { tools: [{ name: "subagent", args: { action: "start", agent: "worker", isolate: "none", task: "build feature", background: true } }] } },
|
||||
{ id: "start-worker", role: "coordinator", match: { kind: "user", text: "begin" }, answer: { tools: [{ name: "subagent", args: { action: "start", agent: "worker", isolate: "worktree", task: "build feature", background: true } }] } },
|
||||
{ id: "start-reviewer", role: "coordinator", match: { kind: "user", text: "worker complete" }, answer: { tools: [{ name: "subagent", args: { action: "start", agent: "reviewer", isolate: "none", task: "review feature", background: true } }] } },
|
||||
{ id: "coordinator-done", role: "coordinator", match: { kind: "user", text: "approve" }, answer: { text: "scenario complete" } },
|
||||
{ id: "land-worker", role: "coordinator", match: { kind: "user", text: "approve" }, answer: (transcript) => ({ tools: [{ name: "subagent", args: { action: "land", id: workerId(transcript) } }] }) },
|
||||
{ id: "submit", role: "coordinator", match: { text: "Landed commits" }, answer: { tools: [{ name: "workspace", args: { action: "submit", divergence: "None" } }] } },
|
||||
{ id: "coordinator-wait", role: "coordinator", match: { text: "" }, answer: { text: "Wait for child report." } },
|
||||
{ id: "worker-write", role: "worker", match: { kind: "user", text: "build feature" }, answer: { tools: [{ name: "write-feature", args: {} }] } },
|
||||
{ id: "worker-commit", role: "worker", match: { text: "feature written" }, answer: { tools: [{ name: "commit-feature", args: {} }] } },
|
||||
@@ -31740,30 +31748,50 @@ if (globalThis.p3Mode !== "unit") {
|
||||
return registry.snapshot().installed();
|
||||
} } }, BACKGROUND_CONTEXT);
|
||||
const root = await harness.root(BACKGROUND_CONTEXT, { agent: { model: { provider: "faux", modelId: "faux-1" }, instructions: "P3 role:ultimate", cwd: "/src/demo" } });
|
||||
const git2 = async (...args) => {
|
||||
const result = await exec(["/bin/git.wasm", ...args], { cwd: "/src/demo", env: [["GIT_AUTHOR_NAME", "Test"], ["GIT_AUTHOR_EMAIL", "test@example.com"], ["GIT_COMMITTER_NAME", "Test"], ["GIT_COMMITTER_EMAIL", "test@example.com"]] });
|
||||
if (globalThis.p3Mode === "kill-forward") {
|
||||
const install = registry.install.bind(registry);
|
||||
registry.install = (extension) => install(extension.name !== "workspaces" ? extension : { ...extension, tasks: extension.tasks.map((task) => task.definition.name !== "workspaces.lander" ? task : { definition: { ...task.definition, phases: { ...task.definition.phases, land: async (record2, runtime, ctx) => {
|
||||
const commit = runtime.commit.bind(runtime);
|
||||
return task.definition.phases.land(record2, { ...runtime, commit: async (...args) => {
|
||||
const result = await commit(...args);
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, BACKGROUND_CONTEXT);
|
||||
if (index.queue.some((entry2) => entry2.landed)) {
|
||||
await fs.write("/forward-recorded.tmp", JSON.stringify(index.queue));
|
||||
await fs.rename("/forward-recorded.tmp", "/forward-recorded");
|
||||
await awaitWithContext(new Promise(() => {
|
||||
}), ctx);
|
||||
}
|
||||
return result;
|
||||
} }, ctx);
|
||||
} } } }) });
|
||||
}
|
||||
const git2 = async (cwd, ...args) => {
|
||||
const result = await exec(["/bin/git.wasm", ...args], { cwd, env: [["GIT_AUTHOR_NAME", "Test"], ["GIT_AUTHOR_EMAIL", "test@example.com"], ["GIT_COMMITTER_NAME", "Test"], ["GIT_COMMITTER_EMAIL", "test@example.com"]] });
|
||||
assert(result.code === 0, `World git ${args}: ${JSON.stringify(result)}`);
|
||||
return result;
|
||||
};
|
||||
const tools = defineExtension({ name: "p3-tools", tools: [
|
||||
defineTool({ name: "write-feature", description: "Write the demo feature", parameters: obj({}), replay: "safe", execute: async (_args, _api, ctx) => {
|
||||
defineTool({ name: "write-feature", description: "Write the isolated feature", parameters: obj({}), replay: "safe", execute: async (_args, api, ctx) => {
|
||||
const cwd = (await api.agent(ctx)).cwd;
|
||||
assert(cwd.startsWith("/worktrees/") && await fs.read(`${cwd}/README.md`) === "P3 demo\n", "Worker must start in a prepared checkout");
|
||||
if (globalThis.p3Mode === "kill-worker") {
|
||||
await fs.write("/worker-running", "yes");
|
||||
await awaitWithContext(new Promise(() => {
|
||||
}), ctx);
|
||||
}
|
||||
await fs.write("/src/demo/feature.txt", "P3 feature\n");
|
||||
await fs.write(`${cwd}/feature.txt`, "P3 feature\n");
|
||||
return { output: [{ type: "text", text: "feature written" }] };
|
||||
} }),
|
||||
defineTool({ name: "commit-feature", description: "Commit the demo feature", parameters: obj({}), replay: "safe", execute: async () => {
|
||||
await git2("add", "feature.txt");
|
||||
if ((await git2("status", "--porcelain")).stdout.trim()) await git2("commit", "--no-gpg-sign", "-m", "P3 feature");
|
||||
defineTool({ name: "commit-feature", description: "Commit the isolated feature", parameters: obj({}), replay: "safe", execute: async (_args, api, ctx) => {
|
||||
const cwd = (await api.agent(ctx)).cwd;
|
||||
await git2(cwd, "add", "feature.txt");
|
||||
if ((await git2(cwd, "status", "--porcelain")).stdout.trim()) await git2(cwd, "commit", "--no-gpg-sign", "-m", "P3 feature");
|
||||
return { output: [{ type: "text", text: "feature committed" }] };
|
||||
} }),
|
||||
defineTool({ name: "review-feature", description: "Verify the committed demo feature", parameters: obj({}), replay: "safe", execute: async () => {
|
||||
assert(await fs.read("/src/demo/feature.txt") === "P3 feature\n", "Reviewer lost feature");
|
||||
const result = await git2("show", "HEAD:feature.txt");
|
||||
assert(result.stdout === "P3 feature\n", "Reviewer did not read the committed feature");
|
||||
defineTool({ name: "review-feature", description: "Verify the worker commit before landing", parameters: obj({}), replay: "safe", execute: async (_args, _api, ctx) => {
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, ctx);
|
||||
const worker = Object.values(index.items).find((item) => item.source.startsWith("/worktrees/"));
|
||||
assert(worker && (await git2(worker.path, "show", "HEAD:feature.txt")).stdout === "P3 feature\n", "Reviewer must approve the worker commit");
|
||||
return { output: [{ type: "text", text: "feature verified" }] };
|
||||
} })
|
||||
] });
|
||||
@@ -31771,54 +31799,59 @@ if (globalThis.p3Mode !== "unit") {
|
||||
assert(!loader.state().error, `Extension load: ${loader.state().error}`);
|
||||
try {
|
||||
if (globalThis.p3Mode === "load-all") {
|
||||
assert((await root.agent(BACKGROUND_CONTEXT)).tools.some((tool) => tool.name === "branch"), "Every extension must load, including messages");
|
||||
} else if (globalThis.p3Mode === "branch" || globalThis.p3Mode === "branch-resume") {
|
||||
assert((await root.agent(BACKGROUND_CONTEXT)).tools.some((tool) => tool.name === "branch"), "Every extension must load");
|
||||
} else {
|
||||
const ultimate = await harness.conversation(await globalThis.p3Subagents.standing("ultimate", BACKGROUND_CONTEXT), BACKGROUND_CONTEXT);
|
||||
harness.resume();
|
||||
if (globalThis.p3Mode === "branch") await (await ultimate.submit({ type: "input", content: "launch branch", requestId: "p3-branch" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
|
||||
const replies = async () => (await ultimate.context(BACKGROUND_CONTEXT)).entries.filter((entry2) => entry2.kind === "board.reply");
|
||||
const deadline = Date.now() + 3e4;
|
||||
while ((await replies()).length === 0) {
|
||||
if (Date.now() > deadline) throw new Error(`No branch reply: ${JSON.stringify((await ultimate.context(BACKGROUND_CONTEXT)).entries)}`);
|
||||
await new Promise((resolve2) => setTimeout(resolve2, 0));
|
||||
const probe = globalThis.p3Mode.startsWith("branch");
|
||||
if (globalThis.p3Mode === "branch" || globalThis.p3Mode.startsWith("kill-")) {
|
||||
await (await ultimate.submit({ type: "input", content: probe ? "launch branch" : "launch scenario", requestId: "p3-launch" }, BACKGROUND_CONTEXT)).wait(BACKGROUND_CONTEXT);
|
||||
const entries3 = (await ultimate.context(BACKGROUND_CONTEXT)).entries;
|
||||
assert(!JSON.stringify(entries3).includes('"isError":true'), `Branch tool failed: ${JSON.stringify(entries3)}`);
|
||||
}
|
||||
await harness.waitForIdle(BACKGROUND_CONTEXT);
|
||||
while (Deno.core.ops.op_sql("all", "SELECT id FROM tasks WHERE json_extract(kind, '$') IN ('board.route', 'board.delivery') AND status NOT IN ('terminal', 'faulted')", []).length) {
|
||||
if (Date.now() > deadline) throw new Error("Branch routing did not settle");
|
||||
await new Promise((resolve2) => setTimeout(resolve2, 0));
|
||||
}
|
||||
const received = await replies();
|
||||
assert(received.length === 1 && received[0].data.text === "branch complete" && !received[0].data.failed, `Expected one coordinator answer: ${JSON.stringify(received)}`);
|
||||
const branch = await harness.conversation(received[0].data.from, BACKGROUND_CONTEXT);
|
||||
assert((await branch.agent(BACKGROUND_CONTEXT)).instructions === "P3 role:coordinator", "Branch must use the coordinator definition");
|
||||
assert((await conversationChains(harness, BACKGROUND_CONTEXT)(branch.id)).includes(String(ultimate.id)), "D63 chain must include ultimate");
|
||||
const faults = Deno.core.ops.op_sql("all", "SELECT id, record FROM tasks WHERE status = 'faulted'", []);
|
||||
assert(faults.length === 0, `Faulted branch tasks: ${JSON.stringify(faults)}`);
|
||||
} else {
|
||||
if (globalThis.p3Mode === "kill-worker") {
|
||||
const coordinator2 = await globalThis.p3Models.start({ agent: "coordinator", cwd: "/src/demo" }, BACKGROUND_CONTEXT);
|
||||
await fs.write("/coordinator", String(coordinator2.id));
|
||||
await coordinator2.submit({ type: "input", content: "begin", requestId: "p3-begin" }, BACKGROUND_CONTEXT);
|
||||
}
|
||||
harness.resume();
|
||||
if (globalThis.p3Mode === "kill-worker") await new Promise(() => {
|
||||
if (globalThis.p3Mode.startsWith("kill-")) await new Promise(() => {
|
||||
});
|
||||
const coordinator = await harness.conversation(Number(await fs.read("/coordinator")), BACKGROUND_CONTEXT);
|
||||
const deadline = Date.now() + 3e4;
|
||||
while (!JSON.stringify((await coordinator.context(BACKGROUND_CONTEXT)).entries).includes("scenario complete")) {
|
||||
if (Date.now() > deadline) throw new Error(`P3 did not finish: ${JSON.stringify((await coordinator.context(BACKGROUND_CONTEXT)).entries)}`);
|
||||
await new Promise((resolve2) => setTimeout(resolve2, 0));
|
||||
const deadline = Date.now() + 6e4;
|
||||
const wait = async (predicate, message) => {
|
||||
while (!await predicate()) {
|
||||
assert(Date.now() < deadline, `${message}: ${JSON.stringify(Deno.core.ops.op_sql("all", "SELECT id, kind, status, record FROM tasks WHERE status != 'terminal'", []))}`);
|
||||
await new Promise((resolve2) => setTimeout(resolve2, 0));
|
||||
}
|
||||
};
|
||||
const replies = async () => (await ultimate.context(BACKGROUND_CONTEXT)).entries.filter((entry2) => entry2.kind === "board.reply");
|
||||
await wait(async () => (await replies()).length > 0, "No branch reply");
|
||||
const received = await replies();
|
||||
const coordinator = await harness.conversation(received[0].data.from, BACKGROUND_CONTEXT);
|
||||
assert((await coordinator.agent(BACKGROUND_CONTEXT)).instructions === "P3 role:coordinator", "Branch must use coordinator definition");
|
||||
assert((await conversationChains(harness, BACKGROUND_CONTEXT)(coordinator.id)).includes(String(ultimate.id)), "D63 must include ultimate");
|
||||
if (!probe) {
|
||||
await wait(async () => Object.keys((await harness.snapshot(globalThis.p3Workspaces.doc, root.id, BACKGROUND_CONTEXT)).results).length === 1, "No landing");
|
||||
await wait(async () => (await harness.snapshot(globalThis.p3Workspaces.doc, root.id, BACKGROUND_CONTEXT)).queue.length === 0, "Lander did not report");
|
||||
}
|
||||
await harness.waitForIdle(BACKGROUND_CONTEXT);
|
||||
await wait(async () => Deno.core.ops.op_sql("all", "SELECT id FROM tasks WHERE status NOT IN ('terminal', 'faulted')", []).length === 0, "Background tasks did not settle");
|
||||
const entries2 = (await coordinator.context(BACKGROUND_CONTEXT)).entries;
|
||||
const reports = entries2.filter((entry2) => entry2.kind === "subagents.report");
|
||||
assert(reports.length === 2, `Expected two child reports: ${JSON.stringify(entries2)}`);
|
||||
assert(new Set(reports.map((report2) => report2.data.sender)).size === 2, "Expected one report from each child");
|
||||
for (const answer2 of ["worker complete", "approve"]) assert(reports.filter((report2) => JSON.stringify(report2).includes(answer2)).length === 1, `Expected exactly one report for ${answer2}`);
|
||||
assert(JSON.stringify(entries2).includes("scenario complete"), "Coordinator did not finish");
|
||||
const faults = Deno.core.ops.op_sql("all", "SELECT id FROM tasks WHERE status = 'faulted'", []);
|
||||
if (probe) assert(received.length === 1 && received[0].data.text === "branch complete" && !received[0].data.failed, "Expected one branch answer");
|
||||
else {
|
||||
const reports = entries2.filter((entry2) => entry2.kind === "subagents.report");
|
||||
assert(reports.length === 2 && new Set(reports.map((entry2) => entry2.data.sender)).size === 2, `Expected one report per child: ${JSON.stringify(reports)}`);
|
||||
for (const answer2 of ["worker complete", "approve"]) assert(reports.filter((entry2) => JSON.stringify(entry2).includes(answer2)).length === 1, `Duplicate child answer: ${answer2}`);
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, BACKGROUND_CONTEXT);
|
||||
const worker = Object.values(index.items).find((item) => item.source.startsWith("/worktrees/"));
|
||||
assert(worker?.state === "removed", "Worker checkout must be released");
|
||||
assert(!await fs.stat(worker.path).then(() => true, () => false), "Released checkout must not exist");
|
||||
assert(Object.values(index.results).every((result) => result.outcome === "landed"), `Landing failed: ${JSON.stringify(index)}`);
|
||||
assert((await globalThis.p3Host.eventLog.recent({ events: [{ kind: "event.workspaces.landed" }] })).length === 1, "Expected one landed event");
|
||||
const ultimateEntries = (await ultimate.context(BACKGROUND_CONTEXT)).entries;
|
||||
const landed = Object.values(index.results)[0];
|
||||
const cards = ultimateEntries.filter((entry2) => entry2.kind === "board.message" && entry2.data.text.startsWith("[lander] Submission") && entry2.data.text.includes(" landed as "));
|
||||
assert(cards.length === 1 && cards[0].data.from === coordinator.id && cards[0].data.text.includes(landed.sha), `Expected one landing card through messages: ${JSON.stringify(cards)}`);
|
||||
const repliesNow = await replies();
|
||||
assert(repliesNow.length === 1 && !repliesNow[0].data.failed && new Set(repliesNow.map((entry2) => entry2.data.id)).size === repliesNow.length, `Duplicate board.reply: ${JSON.stringify(repliesNow)}`);
|
||||
assert(entries2.filter((entry2) => entry2.kind === "workspaces.lander-report").length === 1, "Expected one lander report");
|
||||
}
|
||||
const faults = Deno.core.ops.op_sql("all", "SELECT id, record FROM tasks WHERE status = 'faulted'", []);
|
||||
assert(faults.length === 0, `Faulted tasks: ${JSON.stringify(faults)}`);
|
||||
await fs.write("/p3-idle", "yes");
|
||||
}
|
||||
} finally {
|
||||
await loader.close();
|
||||
|
||||
+83
-65
@@ -1,4 +1,4 @@
|
||||
/** P3 exit scaffold: real world git, scripted coordination, and worker SIGKILL recovery. */
|
||||
/** P3 exit contract: one branch, two child answers, and one landing across three crash boundaries. */
|
||||
import { BACKGROUND_CONTEXT as context, awaitWithContext } from '@earendil-works/chord/context';
|
||||
import { createModels } from '@earendil-works/pi-ai/models';
|
||||
import { Harness, createRegistry, defineExtension, defineTool } from '@earendil-works/pi-durable';
|
||||
@@ -10,14 +10,17 @@ import { scriptedModel, testReplay, type Rule } from './scripted-model';
|
||||
const assert = (ok: any, message: string) => { if (!ok) throw new Error(message); };
|
||||
testReplay();
|
||||
if (globalThis.p3Mode !== 'unit') {
|
||||
const workerId = (transcript: any) => Number(/Started subagent (\d+)/.exec(JSON.stringify(transcript.messages))?.[1]);
|
||||
const rules: Rule[] = [
|
||||
{ id: 'ultimate-branch', role: 'ultimate', match: { kind: 'user', text: 'launch branch' }, answer: { tools: [{ name: 'branch', args: { repo: '/src/demo', prompt: 'branch probe' } }] } },
|
||||
{ id: 'ultimate-branch', role: 'ultimate', match: { kind: 'user', text: 'launch scenario' }, answer: { tools: [{ name: 'branch', args: { repo: '/src/demo', isolated: true, prompt: 'begin' } }] } },
|
||||
{ id: 'ultimate-probe', role: 'ultimate', match: { kind: 'user', text: 'launch branch' }, answer: { tools: [{ name: 'branch', args: { repo: '/src/demo', prompt: 'branch probe' } }] } },
|
||||
{ id: 'coordinator-probe', role: 'coordinator', match: { kind: 'user', text: 'branch probe' }, answer: { text: 'branch complete' } },
|
||||
{ id: 'ultimate-receipt', role: 'ultimate', match: { text: '' }, answer: { text: 'Receipt noted.' } },
|
||||
{ id: 'coordinator-echo', role: 'coordinator', match: { kind: 'user', text: 'Receipt noted.' }, answer: { text: 'Acknowledged.' } },
|
||||
{ id: 'start-worker', role: 'coordinator', match: { kind: 'user', text: 'begin' }, answer: { tools: [{ name: 'subagent', args: { action: 'start', agent: 'worker', isolate: 'none', task: 'build feature', background: true } }] } },
|
||||
{ id: 'start-worker', role: 'coordinator', match: { kind: 'user', text: 'begin' }, answer: { tools: [{ name: 'subagent', args: { action: 'start', agent: 'worker', isolate: 'worktree', task: 'build feature', background: true } }] } },
|
||||
{ id: 'start-reviewer', role: 'coordinator', match: { kind: 'user', text: 'worker complete' }, answer: { tools: [{ name: 'subagent', args: { action: 'start', agent: 'reviewer', isolate: 'none', task: 'review feature', background: true } }] } },
|
||||
{ id: 'coordinator-done', role: 'coordinator', match: { kind: 'user', text: 'approve' }, answer: { text: 'scenario complete' } },
|
||||
{ id: 'land-worker', role: 'coordinator', match: { kind: 'user', text: 'approve' }, answer: transcript => ({ tools: [{ name: 'subagent', args: { action: 'land', id: workerId(transcript) } }] }) },
|
||||
{ id: 'submit', role: 'coordinator', match: { text: 'Landed commits' }, answer: { tools: [{ name: 'workspace', args: { action: 'submit', divergence: 'None' } }] } },
|
||||
{ id: 'coordinator-wait', role: 'coordinator', match: { text: '' }, answer: { text: 'Wait for child report.' } },
|
||||
{ id: 'worker-write', role: 'worker', match: { kind: 'user', text: 'build feature' }, answer: { tools: [{ name: 'write-feature', args: {} }] } },
|
||||
{ id: 'worker-commit', role: 'worker', match: { text: 'feature written' }, answer: { tools: [{ name: 'commit-feature', args: {} }] } },
|
||||
@@ -30,30 +33,49 @@ if (globalThis.p3Mode !== 'unit') {
|
||||
const storage = await openStorage(globalThis.storagePath), registry = createRegistry();
|
||||
const harness = await Harness.open(storage, { models, registry, settings: { retry: { enabled: false }, get extensions() { return registry.snapshot().installed(); } } }, context);
|
||||
const root = await harness.root(context, { agent: { model: { provider: 'faux', modelId: 'faux-1' }, instructions: 'P3 role:ultimate', cwd: '/src/demo' } });
|
||||
const git = async (...args: string[]) => {
|
||||
const result = await exec(['/bin/git.wasm', ...args], { cwd: '/src/demo', env: [['GIT_AUTHOR_NAME', 'Test'], ['GIT_AUTHOR_EMAIL', 'test@example.com'], ['GIT_COMMITTER_NAME', 'Test'], ['GIT_COMMITTER_EMAIL', 'test@example.com']] });
|
||||
// Stop after the durable lander commit records landed.sha, before its event/report commit.
|
||||
if (globalThis.p3Mode === 'kill-forward') {
|
||||
const install = registry.install.bind(registry);
|
||||
registry.install = (extension: any) => install(extension.name !== 'workspaces' ? extension : { ...extension, tasks: extension.tasks.map((task: any) => task.definition.name !== 'workspaces.lander' ? task : { definition: { ...task.definition, phases: { ...task.definition.phases, land: async (record: any, runtime: any, ctx: any) => {
|
||||
const commit = runtime.commit.bind(runtime);
|
||||
return task.definition.phases.land(record, { ...runtime, commit: async (...args: any[]) => {
|
||||
const result = await commit(...args);
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, context);
|
||||
if (index.queue.some((entry: any) => entry.landed)) {
|
||||
await fs.write('/forward-recorded.tmp', JSON.stringify(index.queue));
|
||||
await fs.rename('/forward-recorded.tmp', '/forward-recorded');
|
||||
await awaitWithContext(new Promise(() => {}), ctx);
|
||||
}
|
||||
return result;
|
||||
} }, ctx);
|
||||
} } } }) });
|
||||
}
|
||||
const git = async (cwd: string, ...args: string[]) => {
|
||||
const result = await exec(['/bin/git.wasm', ...args], { cwd, env: [['GIT_AUTHOR_NAME', 'Test'], ['GIT_AUTHOR_EMAIL', 'test@example.com'], ['GIT_COMMITTER_NAME', 'Test'], ['GIT_COMMITTER_EMAIL', 'test@example.com']] });
|
||||
assert(result.code === 0, `World git ${args}: ${JSON.stringify(result)}`);
|
||||
return result;
|
||||
};
|
||||
const tools = defineExtension({ name: 'p3-tools', tools: [
|
||||
defineTool({ name: 'write-feature', description: 'Write the demo feature', parameters: obj({}), replay: 'safe', execute: async (_args, _api, ctx) => {
|
||||
defineTool({ name: 'write-feature', description: 'Write the isolated feature', parameters: obj({}), replay: 'safe', execute: async (_args, api, ctx) => {
|
||||
const cwd = (await api.agent(ctx)).cwd;
|
||||
assert(cwd.startsWith('/worktrees/') && await fs.read(`${cwd}/README.md`) === 'P3 demo\n', 'Worker must start in a prepared checkout');
|
||||
if (globalThis.p3Mode === 'kill-worker') {
|
||||
await fs.write('/worker-running', 'yes');
|
||||
await awaitWithContext(new Promise(() => {}), ctx);
|
||||
}
|
||||
await fs.write('/src/demo/feature.txt', 'P3 feature\n');
|
||||
await fs.write(`${cwd}/feature.txt`, 'P3 feature\n');
|
||||
return { output: [{ type: 'text', text: 'feature written' }] };
|
||||
} }),
|
||||
defineTool({ name: 'commit-feature', description: 'Commit the demo feature', parameters: obj({}), replay: 'safe', execute: async () => {
|
||||
await git('add', 'feature.txt');
|
||||
// A replay after a committed tool side effect must not create a second commit.
|
||||
if ((await git('status', '--porcelain')).stdout.trim()) await git('commit', '--no-gpg-sign', '-m', 'P3 feature');
|
||||
defineTool({ name: 'commit-feature', description: 'Commit the isolated feature', parameters: obj({}), replay: 'safe', execute: async (_args, api, ctx) => {
|
||||
const cwd = (await api.agent(ctx)).cwd;
|
||||
await git(cwd, 'add', 'feature.txt');
|
||||
if ((await git(cwd, 'status', '--porcelain')).stdout.trim()) await git(cwd, 'commit', '--no-gpg-sign', '-m', 'P3 feature');
|
||||
return { output: [{ type: 'text', text: 'feature committed' }] };
|
||||
} }),
|
||||
defineTool({ name: 'review-feature', description: 'Verify the committed demo feature', parameters: obj({}), replay: 'safe', execute: async () => {
|
||||
assert(await fs.read('/src/demo/feature.txt') === 'P3 feature\n', 'Reviewer lost feature');
|
||||
const result = await git('show', 'HEAD:feature.txt');
|
||||
assert(result.stdout === 'P3 feature\n', 'Reviewer did not read the committed feature');
|
||||
defineTool({ name: 'review-feature', description: 'Verify the worker commit before landing', parameters: obj({}), replay: 'safe', execute: async (_args, _api, ctx) => {
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, ctx);
|
||||
const worker = Object.values(index.items).find((item: any) => item.source.startsWith('/worktrees/')) as any;
|
||||
assert(worker && (await git(worker.path, 'show', 'HEAD:feature.txt')).stdout === 'P3 feature\n', 'Reviewer must approve the worker commit');
|
||||
return { output: [{ type: 'text', text: 'feature verified' }] };
|
||||
} }),
|
||||
] });
|
||||
@@ -61,62 +83,58 @@ if (globalThis.p3Mode !== 'unit') {
|
||||
assert(!loader.state().error, `Extension load: ${loader.state().error}`);
|
||||
try {
|
||||
if (globalThis.p3Mode === 'load-all') {
|
||||
assert((await root.agent(context)).tools.some((tool: any) => tool.name === 'branch'), 'Every extension must load, including messages');
|
||||
} else if (globalThis.p3Mode === 'branch' || globalThis.p3Mode === 'branch-resume') {
|
||||
assert((await root.agent(context)).tools.some((tool: any) => tool.name === 'branch'), 'Every extension must load');
|
||||
} else {
|
||||
const ultimate = (await harness.conversation(await globalThis.p3Subagents.standing('ultimate', context), context))!;
|
||||
harness.resume();
|
||||
if (globalThis.p3Mode === 'branch') await (await ultimate.submit({ type: 'input', content: 'launch branch', requestId: 'p3-branch' }, context)).wait(context);
|
||||
const probe = globalThis.p3Mode.startsWith('branch');
|
||||
if (globalThis.p3Mode === 'branch' || globalThis.p3Mode.startsWith('kill-')) {
|
||||
await (await ultimate.submit({ type: 'input', content: probe ? 'launch branch' : 'launch scenario', requestId: 'p3-launch' }, context)).wait(context);
|
||||
const entries = (await ultimate.context(context)).entries;
|
||||
assert(!JSON.stringify(entries).includes('"isError":true'), `Branch tool failed: ${JSON.stringify(entries)}`);
|
||||
}
|
||||
if (globalThis.p3Mode.startsWith('kill-')) await new Promise(() => {});
|
||||
const deadline = Date.now() + 60000;
|
||||
const wait = async (predicate: () => Promise<boolean>, message: string) => {
|
||||
while (!await predicate()) {
|
||||
assert(Date.now() < deadline, `${message}: ${JSON.stringify(Deno.core.ops.op_sql('all', "SELECT id, kind, status, record FROM tasks WHERE status != 'terminal'", []))}`);
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
}
|
||||
};
|
||||
const replies = async () => (await ultimate.context(context)).entries.filter((entry: any) => entry.kind === 'board.reply');
|
||||
const deadline = Date.now() + 30000;
|
||||
while ((await replies()).length === 0) {
|
||||
if (Date.now() > deadline) throw new Error(`No branch reply: ${JSON.stringify((await ultimate.context(context)).entries)}`);
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
await wait(async () => (await replies()).length > 0, 'No branch reply');
|
||||
const received = await replies();
|
||||
const coordinator = (await harness.conversation(received[0].data.from, context))!;
|
||||
assert((await coordinator.agent(context)).instructions === 'P3 role:coordinator', 'Branch must use coordinator definition');
|
||||
assert((await conversationChains(harness, context)(coordinator.id)).includes(String(ultimate.id)), 'D63 must include ultimate');
|
||||
if (!probe) {
|
||||
await wait(async () => Object.keys((await harness.snapshot(globalThis.p3Workspaces.doc, root.id, context)).results).length === 1, 'No landing');
|
||||
await wait(async () => (await harness.snapshot(globalThis.p3Workspaces.doc, root.id, context)).queue.length === 0, 'Lander did not report');
|
||||
}
|
||||
await harness.waitForIdle(context);
|
||||
// Routes are background tasks. Idle alone does not prove that they settled without a fault.
|
||||
while (Deno.core.ops.op_sql('all', "SELECT id FROM tasks WHERE json_extract(kind, '$') IN ('board.route', 'board.delivery') AND status NOT IN ('terminal', 'faulted')", []).length) {
|
||||
if (Date.now() > deadline) throw new Error('Branch routing did not settle');
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
await wait(async () => Deno.core.ops.op_sql('all', "SELECT id FROM tasks WHERE status NOT IN ('terminal', 'faulted')", []).length === 0, 'Background tasks did not settle');
|
||||
const entries = (await coordinator.context(context)).entries;
|
||||
if (probe) assert(received.length === 1 && received[0].data.text === 'branch complete' && !received[0].data.failed, 'Expected one branch answer');
|
||||
else {
|
||||
const reports = entries.filter((entry: any) => entry.kind === 'subagents.report');
|
||||
assert(reports.length === 2 && new Set(reports.map((entry: any) => entry.data.sender)).size === 2, `Expected one report per child: ${JSON.stringify(reports)}`);
|
||||
for (const answer of ['worker complete', 'approve']) assert(reports.filter((entry: any) => JSON.stringify(entry).includes(answer)).length === 1, `Duplicate child answer: ${answer}`);
|
||||
const index = await harness.snapshot(globalThis.p3Workspaces.doc, root.id, context);
|
||||
const worker = Object.values(index.items).find((item: any) => item.source.startsWith('/worktrees/')) as any;
|
||||
assert(worker?.state === 'removed', 'Worker checkout must be released');
|
||||
assert(!await fs.stat(worker.path).then(() => true, () => false), 'Released checkout must not exist');
|
||||
assert(Object.values(index.results).every((result: any) => result.outcome === 'landed'), `Landing failed: ${JSON.stringify(index)}`);
|
||||
assert((await globalThis.p3Host.eventLog.recent({ events: [{ kind: 'event.workspaces.landed' }] })).length === 1, 'Expected one landed event');
|
||||
const ultimateEntries = (await ultimate.context(context)).entries;
|
||||
const landed = Object.values(index.results)[0] as any;
|
||||
const cards = ultimateEntries.filter((entry: any) => entry.kind === 'board.message' && entry.data.text.startsWith('[lander] Submission') && entry.data.text.includes(' landed as '));
|
||||
assert(cards.length === 1 && cards[0].data.from === coordinator.id && cards[0].data.text.includes(landed.sha), `Expected one landing card through messages: ${JSON.stringify(cards)}`);
|
||||
const repliesNow = await replies();
|
||||
assert(repliesNow.length === 1 && !repliesNow[0].data.failed && new Set(repliesNow.map((entry: any) => entry.data.id)).size === repliesNow.length, `Duplicate board.reply: ${JSON.stringify(repliesNow)}`);
|
||||
assert(entries.filter((entry: any) => entry.kind === 'workspaces.lander-report').length === 1, 'Expected one lander report');
|
||||
}
|
||||
const received = await replies();
|
||||
assert(received.length === 1 && received[0].data.text === 'branch complete' && !received[0].data.failed, `Expected one coordinator answer: ${JSON.stringify(received)}`);
|
||||
const branch = (await harness.conversation(received[0].data.from, context))!;
|
||||
assert((await branch.agent(context)).instructions === 'P3 role:coordinator', 'Branch must use the coordinator definition');
|
||||
assert((await conversationChains(harness, context)(branch.id)).includes(String(ultimate.id)), 'D63 chain must include ultimate');
|
||||
const faults = Deno.core.ops.op_sql('all', "SELECT id, record FROM tasks WHERE status = 'faulted'", []);
|
||||
assert(faults.length === 0, `Faulted branch tasks: ${JSON.stringify(faults)}`);
|
||||
} else {
|
||||
if (globalThis.p3Mode === 'kill-worker') {
|
||||
const coordinator = await globalThis.p3Models.start({ agent: 'coordinator', cwd: '/src/demo' }, context);
|
||||
await fs.write('/coordinator', String(coordinator.id));
|
||||
await coordinator.submit({ type: 'input', content: 'begin', requestId: 'p3-begin' }, context);
|
||||
}
|
||||
harness.resume();
|
||||
if (globalThis.p3Mode === 'kill-worker') await new Promise(() => {});
|
||||
const coordinator = (await harness.conversation(Number(await fs.read('/coordinator')), context))!;
|
||||
// waitForIdle does not wait for background children. Poll the durable outcome first.
|
||||
const deadline = Date.now() + 30000;
|
||||
while (!JSON.stringify((await coordinator.context(context)).entries).includes('scenario complete')) {
|
||||
if (Date.now() > deadline) throw new Error(`P3 did not finish: ${JSON.stringify((await coordinator.context(context)).entries)}`);
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
}
|
||||
await harness.waitForIdle(context);
|
||||
const entries = (await coordinator.context(context)).entries;
|
||||
const reports = entries.filter((entry: any) => entry.kind === 'subagents.report');
|
||||
assert(reports.length === 2, `Expected two child reports: ${JSON.stringify(entries)}`);
|
||||
assert(new Set(reports.map((report: any) => report.data.sender)).size === 2, 'Expected one report from each child');
|
||||
for (const answer of ['worker complete', 'approve']) assert(reports.filter((report: any) => JSON.stringify(report).includes(answer)).length === 1, `Expected exactly one report for ${answer}`);
|
||||
assert(JSON.stringify(entries).includes('scenario complete'), 'Coordinator did not finish');
|
||||
const faults = Deno.core.ops.op_sql('all', "SELECT id FROM tasks WHERE status = 'faulted'", []);
|
||||
assert(faults.length === 0, `Faulted tasks: ${JSON.stringify(faults)}`);
|
||||
await fs.write('/p3-idle', 'yes');
|
||||
assert(faults.length === 0, `Faulted tasks: ${JSON.stringify(faults)}`);
|
||||
}
|
||||
} finally { await loader.close(); await harness.close(context); }
|
||||
}
|
||||
// branch_round_trip covers @ultimate's branch call, one board.reply, D63, and an idle reopen.
|
||||
// The worker SIGKILL scaffold remains separate until the T52 scenario connects the landing path.
|
||||
// PENDING T52: worker isolate:worktree, subagent land, workspace submit, and worktree release.
|
||||
// Kill point 2 (ignored until T52): SIGKILL while the lander check waits for /check-release.
|
||||
// Restart, release the marker, and run to idle.
|
||||
// Kill point 3 (ignored until T52): SIGKILL after the target fast-forward log entry.
|
||||
// Restart and run to idle. Assert one target commit, workspaces.landed event, and landing card.
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/** Stateless role/input rules for deterministic faux-model replay in integration tests. */
|
||||
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
|
||||
export type Answer = { text?: string; tools?: { name: string; args: any }[] };
|
||||
export type Rule = { id: string; role: string; match: { kind?: string; text: string }; answer: Answer };
|
||||
export type Rule = { id: string; role: string; match: { kind?: string; text: string }; answer: Answer | ((transcript: any) => Answer) };
|
||||
export const textOf = (content: any): string => typeof content === 'string' ? content : (content ?? []).filter((part: any) => part.type === 'text').map((part: any) => part.text).join('\n');
|
||||
export function answerFor(rules: Rule[], transcript: any) {
|
||||
const system = JSON.stringify(transcript.messages.filter((message: any) => message.role === 'system'));
|
||||
@@ -9,7 +9,7 @@ export function answerFor(rules: Rule[], transcript: any) {
|
||||
const last = transcript.messages.findLast((message: any) => message.role !== 'system');
|
||||
const rule = rules.find(rule => rule.role === role && (!rule.match.kind || rule.match.kind === last?.role) && textOf(last?.content).includes(rule.match.text));
|
||||
if (!rule) throw new Error(`No scripted rule for ${role}: ${JSON.stringify(last)}`);
|
||||
return { rule: rule.id, answer: rule.answer };
|
||||
return { rule: rule.id, answer: typeof rule.answer === 'function' ? rule.answer(transcript) : rule.answer };
|
||||
}
|
||||
export function scriptedModel(rules: Rule[], report: (rule: string) => void) {
|
||||
const faux = fauxProvider({ tokensPerSecond: 100000 });
|
||||
@@ -33,6 +33,11 @@ export function testReplay() {
|
||||
const first = answerFor(rules, { messages: [system, input] });
|
||||
const replay = answerFor(JSON.parse(JSON.stringify(rules)), { messages: [system, { role: 'assistant', content: [] }, input] });
|
||||
if (JSON.stringify(first) !== JSON.stringify(replay)) throw new Error('Scripted replay changed the answer');
|
||||
const dynamic: Rule[] = [{ id: 'land', role: 'worker', match: { text: 'build' }, answer: transcript => ({ tools: [{ name: 'land', args: { id: /Started subagent (\d+)/.exec(JSON.stringify(transcript.messages))?.[1] } }] }) }];
|
||||
const history = { role: 'toolResult', content: 'Started subagent 42.' };
|
||||
const selected = answerFor(dynamic, { messages: [system, history, input] });
|
||||
const restarted = answerFor(dynamic, { messages: [system, { role: 'assistant', content: [] }, history, input] });
|
||||
if (JSON.stringify(selected) !== JSON.stringify(restarted) || selected.answer.tools?.[0].args.id !== '42') throw new Error('Dynamic tool identity changed on replay');
|
||||
const toolRules: Rule[] = [{ id: 'worker-done', role: 'worker', match: { text: 'committed' }, answer: { text: 'done' } }];
|
||||
if (answerFor(toolRules, { messages: [system, { role: 'toolResult', content: [{ type: 'text', text: 'committed' }] }] }).rule !== 'worker-done') throw new Error('Tool result rule did not match');
|
||||
}
|
||||
|
||||
+72
-24
@@ -1,4 +1,4 @@
|
||||
//! P3 exit scaffold with a worker SIGKILL and pending lander crash boundaries.
|
||||
//! P3 exit scenario across worker, check, and recorded fast-forward SIGKILL boundaries.
|
||||
use sausage::{harness::run_module, world::World};
|
||||
use std::{
|
||||
path::Path,
|
||||
@@ -69,6 +69,21 @@ fn prepare(dir: &Path, all_extensions: bool) -> anyhow::Result<()> {
|
||||
target.join("wasm32-wasip2/release/git.wasm"),
|
||||
disk.join("bin/git.wasm"),
|
||||
)?;
|
||||
if !target.join("wasm32-wasip2/release/brush.wasm").exists() {
|
||||
anyhow::ensure!(
|
||||
Command::new("nu")
|
||||
.arg("brush/build.nu")
|
||||
.env("CARGO_TARGET_DIR", target)
|
||||
.current_dir(&userland)
|
||||
.status()?
|
||||
.success(),
|
||||
"brush build failed"
|
||||
);
|
||||
}
|
||||
std::fs::copy(
|
||||
target.join("wasm32-wasip2/release/brush.wasm"),
|
||||
disk.join("bin/sh.wasm"),
|
||||
)?;
|
||||
if all_extensions {
|
||||
copy_tree(&source.join("extensions"), &repo.join("extensions"))?;
|
||||
} else {
|
||||
@@ -82,7 +97,7 @@ fn prepare(dir: &Path, all_extensions: bool) -> anyhow::Result<()> {
|
||||
std::fs::create_dir_all(repo.join("extensions/probe"))?;
|
||||
std::fs::write(
|
||||
repo.join("extensions/probe/server.ts"),
|
||||
"export default host => { globalThis.p3Models=host.need({name:'models'}); globalThis.p3Subagents=host.need({name:'subagents'}); return {name:'probe'}; }",
|
||||
"export default host => { globalThis.p3Models=host.need({name:'models'}); globalThis.p3Subagents=host.need({name:'subagents'}); globalThis.p3Workspaces=host.need({name:'workspaces'}); globalThis.p3Host=host; return {name:'probe'}; }",
|
||||
)?;
|
||||
std::fs::create_dir_all(repo.join("agents"))?;
|
||||
std::fs::write(
|
||||
@@ -107,11 +122,17 @@ fn prepare(dir: &Path, all_extensions: bool) -> anyhow::Result<()> {
|
||||
),
|
||||
)?;
|
||||
}
|
||||
std::fs::write(
|
||||
repo.join("settings.json"),
|
||||
r#"{"lander":{"checks":["sh -c ': > /check-running; until test -f /check-release; do :; done; test -f feature.txt'"]}}"#,
|
||||
)?;
|
||||
let demo = disk.join("src/demo");
|
||||
std::fs::create_dir_all(&demo)?;
|
||||
std::fs::write(demo.join("README.md"), "P3 demo\n")?;
|
||||
for repo in [&repo, &demo] {
|
||||
git(repo, &["init", "-b", "main"])?;
|
||||
git(repo, &["config", "user.name", "Test"])?;
|
||||
git(repo, &["config", "user.email", "test@example.com"])?;
|
||||
git(repo, &["add", "."])?;
|
||||
git(repo, &["commit", "--no-gpg-sign", "-m", "P3 fixture"])?;
|
||||
}
|
||||
@@ -156,10 +177,12 @@ async fn restart_child() -> anyhow::Result<()> {
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
#[test]
|
||||
fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
fn recovery(point: &str) -> anyhow::Result<()> {
|
||||
let dir = tempfile::tempdir()?;
|
||||
prepare(dir.path(), false)?;
|
||||
prepare(dir.path(), true)?;
|
||||
if point != "check" {
|
||||
std::fs::write(dir.path().join("world/check-release"), "yes")?;
|
||||
}
|
||||
let spawn = |mode: &str| {
|
||||
Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "restart_child", "--nocapture"])
|
||||
@@ -167,16 +190,22 @@ fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
.env("P3_MODE", mode)
|
||||
.spawn()
|
||||
};
|
||||
let mut child = spawn("kill-worker")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(45);
|
||||
while !dir.path().join("world/worker-running").exists() {
|
||||
let mut child = spawn(&format!("kill-{point}"))?;
|
||||
let marker = match point {
|
||||
"worker" => "worker-running",
|
||||
"check" => "check-running",
|
||||
"forward" => "forward-recorded",
|
||||
_ => unreachable!(),
|
||||
};
|
||||
let deadline = Instant::now() + Duration::from_secs(120);
|
||||
while !dir.path().join("world").join(marker).exists() {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
anyhow::bail!("Worker child exited before marker: {status}");
|
||||
anyhow::bail!("P3 child exited before {point} marker: {status}");
|
||||
}
|
||||
if Instant::now() > deadline {
|
||||
child.kill()?;
|
||||
child.wait()?;
|
||||
anyhow::bail!("Worker did not reach kill point 1");
|
||||
anyhow::bail!("P3 did not reach {point} marker");
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
@@ -184,10 +213,25 @@ fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
let killed = child.wait()?;
|
||||
use std::os::unix::process::ExitStatusExt;
|
||||
assert_eq!(killed.signal(), Some(9));
|
||||
// A second reopen proves that idle recovery does not duplicate child reports or commits.
|
||||
let demo = dir.path().join("world/src/demo");
|
||||
assert_eq!(
|
||||
git(&demo, &["rev-list", "--count", "main"])?,
|
||||
if point == "forward" { "3" } else { "1" }
|
||||
);
|
||||
if point == "forward" {
|
||||
let recorded: serde_json::Value = serde_json::from_str(&std::fs::read_to_string(
|
||||
dir.path().join("world/forward-recorded"),
|
||||
)?)?;
|
||||
assert_eq!(
|
||||
recorded[0]["landed"]["sha"].as_str(),
|
||||
Some(git(&demo, &["rev-parse", "main"])?.as_str())
|
||||
);
|
||||
}
|
||||
std::fs::write(dir.path().join("world/check-release"), "yes")?;
|
||||
// A second reopen proves that idle recovery does not duplicate reports, cards, events, or commits.
|
||||
for _ in 0..2 {
|
||||
let mut child = spawn("resume")?;
|
||||
let deadline = Instant::now() + Duration::from_secs(45);
|
||||
let deadline = Instant::now() + Duration::from_secs(120);
|
||||
loop {
|
||||
if let Some(status) = child.try_wait()? {
|
||||
anyhow::ensure!(status.success(), "P3 recovery failed: {status}");
|
||||
@@ -201,8 +245,14 @@ fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
std::thread::yield_now();
|
||||
}
|
||||
let demo = dir.path().join("world/src/demo");
|
||||
assert_eq!(git(&demo, &["rev-list", "--count", "HEAD"])?, "2");
|
||||
assert_eq!(git(&demo, &["log", "-1", "--format=%s"])?, "P3 feature");
|
||||
assert_eq!(git(&demo, &["rev-list", "--count", "main"])?, "3");
|
||||
assert_eq!(
|
||||
git(&demo, &["log", "--format=%s", "main"])?
|
||||
.lines()
|
||||
.filter(|line| *line == "P3 feature")
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(git(&demo, &["status", "--porcelain"])?, "");
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(demo.join("feature.txt"))?,
|
||||
@@ -212,16 +262,14 @@ fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
#[test]
|
||||
#[ignore = "Pending T52: kill point 2 needs the lander check and marker-file gate"]
|
||||
fn lander_check_sigkill_recovery() {
|
||||
// Wait for the check-start marker, SIGKILL, reopen, release /check-release, and run to idle.
|
||||
// Assert one landed event/card and no duplicate board reply (T49).
|
||||
panic!("Pending T52 lander check");
|
||||
fn worker_sigkill_recovery() -> anyhow::Result<()> {
|
||||
recovery("worker")
|
||||
}
|
||||
#[test]
|
||||
#[ignore = "Pending T52: kill point 3 needs the target fast-forward log entry"]
|
||||
fn target_fast_forward_sigkill_recovery() {
|
||||
// Detect the durable fast-forward log entry, SIGKILL, reopen, and run to idle.
|
||||
// Assert one target commit, one landed event/card, no faulted tasks, and worktree release.
|
||||
panic!("Pending T52 target fast-forward");
|
||||
fn lander_check_sigkill_recovery() -> anyhow::Result<()> {
|
||||
recovery("check")
|
||||
}
|
||||
#[test]
|
||||
fn target_fast_forward_sigkill_recovery() -> anyhow::Result<()> {
|
||||
recovery("forward")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user