From d014222ac40d05a444127eb7c5d90d236b56948f Mon Sep 17 00:00:00 2001 From: Surma Date: Sat, 7 Nov 2026 14:45:57 +0000 Subject: [PATCH] Test ultimate branch replies and D63 ancestry across reopen --- host/js/p3-test-bundle.js | 173 ++++++++++++++++++++++++++------------ host/js/p3-test.ts | 31 ++++++- host/tests/p3.rs | 16 +++- 3 files changed, 162 insertions(+), 58 deletions(-) diff --git a/host/js/p3-test-bundle.js b/host/js/p3-test-bundle.js index a1fe2a6..088449f 100644 --- a/host/js/p3-test-bundle.js +++ b/host/js/p3-test-bundle.js @@ -7915,27 +7915,27 @@ var REPLACE_TAKE = { kind: "replace" }; var PendingOps = class { #state = EMPTY; /** Queue the ops of one commit. */ - push(ops2) { - if (ops2.length === 0) + push(ops3) { + if (ops3.length === 0) return; const state = this.#state; - if (ops2[0][0] === "r") { - this.#state = { kind: "batch", ops: ops2 }; + if (ops3[0][0] === "r") { + this.#state = { kind: "batch", ops: ops3 }; return; } switch (state.kind) { case "empty": - this.#state = { kind: "batch", ops: ops2 }; + this.#state = { kind: "batch", ops: ops3 }; return; case "batch": { const folding = { kind: "folding", ops: [], leaves: /* @__PURE__ */ new Map() }; this.#state = folding; if (this.#fold(state.ops, folding)) - this.#fold(ops2, folding); + this.#fold(ops3, folding); return; } case "folding": - this.#fold(ops2, state); + this.#fold(ops3, state); return; case "replace": return; @@ -7966,8 +7966,8 @@ var PendingOps = class { } } /** Fold `ops` into `folding`; false once the queue overflowed into `replace`. */ - #fold(ops2, folding) { - for (const op2 of ops2) { + #fold(ops3, folding) { + for (const op2 of ops3) { switch (op2[0]) { case "r": folding.ops.length = 0; @@ -8236,12 +8236,12 @@ var PathError = class extends Error { this.name = "PathError"; } }; -function apply(target, ops2) { - return applyOps(target, ops2); +function apply(target, ops3) { + return applyOps(target, ops3); } -function applyOps(target, ops2) { +function applyOps(target, ops3) { let root = target; - for (const op2 of ops2) { + for (const op2 of ops3) { assertValidOp(op2); if (op2[0] === "r") { root = op2[1]; @@ -8310,14 +8310,14 @@ function applyOps(target, ops2) { } return root; } -function applyImmutable(target, ops2) { - return applyImmutableBatches(target, [ops2]); +function applyImmutable(target, ops3) { + return applyImmutableBatches(target, [ops3]); } function applyImmutableBatches(target, batches) { let root = target; const owned = /* @__PURE__ */ new WeakSet(); - for (const ops2 of batches) { - for (const op2 of ops2) { + for (const ops3 of batches) { + for (const op2 of ops3) { assertValidOp(op2); if (op2[0] === "r") { root = op2[1]; @@ -8431,10 +8431,10 @@ function encoder() { let nextId = 0; let previous; return { - encode(ops2) { + encode(ops3) { previous = void 0; const out = []; - for (const op2 of ops2) { + for (const op2 of ops3) { if (op2[0] === "r") { out.push(op2); seen.clear(); @@ -8684,9 +8684,9 @@ var Source = class { return []; } }; -function unbox(ops2, value2) { +function unbox(ops3, value2) { const result = []; - for (const op2 of ops2) { + for (const op2 of ops3) { if (op2[0] === "r" || op2[1].length === 0) return [["r", value2]]; const path = op2[1].slice(1); @@ -8780,8 +8780,8 @@ var Cell = class extends Source { throw notReady(`state ${this.key} (its init has not run)`); const before = this.value; const diff = options.diff ?? sharesStructure(before, value2); - const ops2 = diff ? diffRevisions(before, value2) : void 0; - this.#publish(this.#tracker.prepareReplace({ v: value2 }), "ops", true, ops2); + const ops3 = diff ? diffRevisions(before, value2) : void 0; + this.#publish(this.#tracker.prepareReplace({ v: value2 }), "ops", true, ops3); } /** * Installs a migrated value and its schema (reload commit). The caller validated the value and the derivations @@ -8792,7 +8792,7 @@ var Cell = class extends Source { this.#publish(this.#tracker.prepareReplace({ v: value2 }), "reset", false); } /** Adopts this revision and every derived one, then queues all notifications (cell first, then derivations). */ - #publish(prepared, type, validate2 = true, ops2) { + #publish(prepared, type, validate2 = true, ops3) { if (prepared.ops.length === 0 && type === "ops") { this.#tracker.adopt(prepared); return; @@ -8809,7 +8809,7 @@ var Cell = class extends Source { } holdPublications(() => { this.#tracker.adopt(prepared); - const published = ops2 !== void 0 && ops2.length > 0 ? ops2 : unbox(prepared.ops, value2); + const published = ops3 !== void 0 && ops3.length > 0 ? ops3 : unbox(prepared.ops, value2); this.fanout.emit(type === "reset" ? { type, value: value2 } : { type, ops: published, value: value2 }); for (const [derived, next] of candidates) derived.install(next); @@ -8922,9 +8922,9 @@ var Derived = class extends Source { install(next) { const before = this.#value; this.#value = next; - const ops2 = before === void 0 || next === void 0 ? [["r", next ?? null]] : diffRevisions(before, next); - if (ops2.length > 0) - this.fanout.emit({ type: "ops", ops: ops2, value: next }); + const ops3 = before === void 0 || next === void 0 ? [["r", next ?? null]] : diffRevisions(before, next); + if (ops3.length > 0) + this.fanout.emit({ type: "ops", ops: ops3, value: next }); } }; var isSource = (value2) => value2 instanceof Source; @@ -12977,10 +12977,10 @@ var RemoteBinding = class { const target = pending?.values.get(name) ?? this.members.get(name); if (target === void 0) return; - const ops2 = target.decoder.decode(frame.ops ?? []); - target.value = applyImmutable(target.value, ops2); + const ops3 = target.decoder.decode(frame.ops ?? []); + target.value = applyImmutable(target.value, ops3); if (target === this.members.get(name)) - this.#deliver(name, { type: "ops", ops: ops2, value: target.value }); + this.#deliver(name, { type: "ops", ops: ops3, value: target.value }); return; } const fresh = { value: frame.value, decoder: decoder() }; @@ -14804,8 +14804,8 @@ var Watch = class extends WatchLifecycle { this.#retired = false; if (lone !== void 0) return lone; - const ops2 = batch.kind === "ops" ? batch.ops : [["r", value2]]; - return { kind: "value", value: value2, ops: ops2, retired }; + const ops3 = batch.kind === "ops" ? batch.ops : [["r", value2]]; + return { kind: "value", value: value2, ops: ops3, retired }; } /** The pending failure, or the value after one as a root replacement (no ops were kept since the failure). */ #nextAfterFailure() { @@ -16570,7 +16570,7 @@ function describe(plan, record2) { const change = plan.change; if (change === void 0 || !publishes(plan)) return void 0; - const ops2 = change.loaded === void 0 ? EMPTY_OPERATIONS : change.prepared.ops; + const ops3 = change.loaded === void 0 ? EMPTY_OPERATIONS : change.prepared.ops; return { type: "document", kind: "change", @@ -16578,11 +16578,11 @@ function describe(plan, record2) { conversationId, version: change.version, value: change.prepared.value, - ops: ops2 + ops: ops3 }; } -function storeHeld(hold, ops2, value2, version) { - hold.pending.push(ops2); +function storeHeld(hold, ops3, value2, version) { + hold.pending.push(ops3); const taken = hold.pending.take(); if (taken.kind === "replace") return { content: { version, kind: "base", value: value2 }, ops: [["r", value2]] }; @@ -17663,7 +17663,7 @@ var SessionImpl = class { return { kind: "adoption", error }; } const deadline = this.#clock() + withinMs; - for (const { loaded, ops: ops2, base } of held2) { + for (const { loaded, ops: ops3, base } of held2) { const id = loaded.record.id; let hold = this.#holds.get(id); if (hold === void 0) { @@ -17671,7 +17671,7 @@ var SessionImpl = class { hold = { loaded, pending: new PendingOps(), base: false, deadline, unpin }; this.#holds.set(id, hold); } - hold.pending.push(ops2); + hold.pending.push(ops3); if (base) hold.base = true; if (deadline < hold.deadline) @@ -26442,14 +26442,14 @@ var TaskGraphView = class { const mount = this.#mount; if (mount === void 0) return; - const ops2 = []; + const ops3 = []; for (const [key, record2] of mount.records) { const blocked = this.#blocking.blocked(record2); if (blocked === mount.value.tasks[key]?.blocked) continue; - ops2.push(blocked === void 0 ? ["d", ["tasks", key, "blocked"]] : ["s", ["tasks", key, "blocked"], blocked]); + ops3.push(blocked === void 0 ? ["d", ["tasks", key, "blocked"]] : ["s", ["tasks", key, "blocked"], blocked]); } - publish(mount, ops2, BACKGROUND_CONTEXT, (error) => this.#session.report(error)); + publish(mount, ops3, BACKGROUND_CONTEXT, (error) => this.#session.report(error)); }).catch(() => { }); } @@ -26473,7 +26473,7 @@ var TaskGraphView = class { } }; function advance(mount, publication, context, blocked, report2) { - const ops2 = []; + const ops3 = []; const changed = /* @__PURE__ */ new Map(); const node = (key) => changed.has(key) ? changed.get(key) : mount.value.tasks[key]; for (const change of publication.changes) { @@ -26486,7 +26486,7 @@ function advance(mount, publication, context, blocked, report2) { mount.records.delete(key); if (previous === void 0) continue; - ops2.push(["d", ["tasks", key]]); + ops3.push(["d", ["tasks", key]]); changed.set(key, void 0); continue; } @@ -26495,7 +26495,7 @@ function advance(mount, publication, context, blocked, report2) { const next = nodeOf2(record2, previous?.conversations ?? [], blocked(subject)); if (previous !== void 0 && JSON.stringify(next) === JSON.stringify(previous)) continue; - ops2.push(["s", ["tasks", key], next]); + ops3.push(["s", ["tasks", key], next]); changed.set(key, next); } const created = /* @__PURE__ */ new Map(); @@ -26508,16 +26508,16 @@ function advance(mount, publication, context, blocked, report2) { } for (const [key, ids] of created) { const conversations = [...node(key).conversations, ...ids].sort((a, b) => a - b); - ops2.push(["s", ["tasks", key, "conversations"], conversations]); + ops3.push(["s", ["tasks", key, "conversations"], conversations]); } - publish(mount, ops2, context, report2); + publish(mount, ops3, context, report2); } -function publish(mount, ops2, context, report2) { - if (ops2.length === 0) +function publish(mount, ops3, context, report2) { + if (ops3.length === 0) return; - mount.value = applyImmutable(mount.value, ops2); + mount.value = applyImmutable(mount.value, ops3); const frameContext = withoutAbortSignal(context); - const call = { kind: "value", value: mount.value, ops: ops2, retired: false }; + const call = { kind: "value", value: mount.value, ops: ops3, retired: false }; for (const observer of [...mount.observers]) contained(() => observer.advance(call, frameContext), report2); } @@ -26743,19 +26743,19 @@ function advance2(id, mount, publication, context, report2) { const previous = windowed.value; let entries2 = previous.entries; let before = previous.before; - let ops2 = docOps; + let ops3 = docOps; if (appended.length > 0) { - ops2 = [...docOps, ["p", ["entries"], entries2.length, 0, inserted]]; + ops3 = [...docOps, ["p", ["entries"], entries2.length, 0, inserted]]; entries2 = entries2.concat(appended); if (entries2.length > windowed.max) { const drop = entries2.length - windowed.min; before = entries2[drop - 1].id; entries2 = entries2.slice(drop); - ops2.push(["p", ["entries"], 0, drop, []], ["s", ["before"], before]); + ops3.push(["p", ["entries"], 0, drop, []], ["s", ["before"], before]); } } windowed.value = view(mount.conversation, entries2, before, mount.docs); - const call = { kind: "value", value: windowed.value, ops: ops2, retired: false }; + const call = { kind: "value", value: windowed.value, ops: ops3, retired: false }; contained(() => windowed.observer.advance(call, frameContext), report2); } } @@ -31637,6 +31637,43 @@ async function openStorage(path) { return SqliteStorage.open(database); } +// environment.ts +var CodingTools2 = { ...CodingTools, tools: CodingTools.tools.map((tool) => ({ + ...tool, + execute: (args, api, context) => tool.execute(args, { ...api, env: api.env.forTool(api.callId, tool.name) }, context) +})) }; +var Started = defineDoc({ + kind: "board.started", + version: 1, + checkpointWhen: () => false, + scope: "conversation", + history: "latest", + fork: "initial", + initial: () => ({}) +}); +function conversationChains(harness, context) { + const cache = /* @__PURE__ */ new Map(); + const chain = (id) => { + let result = cache.get(id); + if (!result) { + result = (async () => { + const record2 = await harness.commit((tx) => tx.conversation(id), context); + const up = record2?.owner?.conversationId ?? record2?.parent?.conversationId; + const starter = (await harness.snapshot(Started, id, context))?.by; + return [.../* @__PURE__ */ new Set([ + String(id), + ...up === void 0 ? [] : await chain(up), + ...starter === void 0 ? [] : await chain(starter) + ])]; + })(); + cache.set(id, result); + } + return result; + }; + return chain; +} +var ops2 = Deno.core.ops; + // scripted-model.ts var textOf = (content) => typeof content === "string" ? content : (content ?? []).filter((part) => part.type === "text").map((part) => part.text).join("\n"); function answerFor(rules, transcript) { @@ -31679,6 +31716,10 @@ var assert = (ok, message) => { testReplay(); if (globalThis.p3Mode !== "unit") { 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: "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-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" } }, @@ -31729,6 +31770,28 @@ if (globalThis.p3Mode !== "unit") { 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") { + 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)); + } + 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); diff --git a/host/js/p3-test.ts b/host/js/p3-test.ts index d258acb..2c92787 100644 --- a/host/js/p3-test.ts +++ b/host/js/p3-test.ts @@ -5,11 +5,16 @@ import { Harness, createRegistry, defineExtension, defineTool } from '@earendil- import { obj } from '@earendil-works/pi-schema'; import { loadExtensions, fs, exec } from './extensions'; import { openStorage } from './sqlite'; +import { conversationChains } from './environment'; 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 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: '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-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' } }, @@ -57,6 +62,29 @@ if (globalThis.p3Mode !== 'unit') { 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') { + 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 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 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)); + } + 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); @@ -85,7 +113,8 @@ if (globalThis.p3Mode !== 'unit') { } } finally { await loader.close(); await harness.close(context); } } -// PENDING T49: @ultimate calls branch to create the coordinator, then receives one board.reply. +// 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. diff --git a/host/tests/p3.rs b/host/tests/p3.rs index 233087f..a8eb70b 100644 --- a/host/tests/p3.rs +++ b/host/tests/p3.rs @@ -82,7 +82,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'}); return {name:'probe'}; }", + "export default host => { globalThis.p3Models=host.need({name:'models'}); globalThis.p3Subagents=host.need({name:'subagents'}); return {name:'probe'}; }", )?; std::fs::create_dir_all(repo.join("agents"))?; std::fs::write( @@ -98,7 +98,12 @@ fn prepare(dir: &Path, all_extensions: bool) -> anyhow::Result<()> { std::fs::write( repo.join(format!("agents/{role}.md")), format!( - "---\nname: {role}\ndescription: P3 {role}\nisolate: none\n{denial}---\nP3 role:{role}\n" + "---\nname: {role}\ndescription: P3 {role}\nisolate: none\n{denial}{}---\nP3 role:{role}\n", + if role == "ultimate" { + "resident: true\n" + } else { + "" + } ), )?; } @@ -138,6 +143,13 @@ async fn all_extensions_load() -> anyhow::Result<()> { run(dir.path(), "load-all").await } #[tokio::test(flavor = "current_thread")] +async fn branch_round_trip() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + prepare(dir.path(), true)?; + run(dir.path(), "branch").await?; + run(dir.path(), "branch-resume").await +} +#[tokio::test(flavor = "current_thread")] async fn restart_child() -> anyhow::Result<()> { if let Ok(dir) = std::env::var("P3_DIR") { run(Path::new(&dir), &std::env::var("P3_MODE")?).await?;