From a66a933dc47ccf2b861949d703894e547b06d9c9 Mon Sep 17 00:00:00 2001 From: Surma Date: Sat, 7 Nov 2026 14:41:02 +0000 Subject: [PATCH] Fix branch model configuration and load all world extensions in a test --- extensions/messages/branch.ts | 13 ++++----- host/js/p3-test-bundle.js | 52 +++++++++++++++++++---------------- host/js/p3-test.ts | 6 +++- host/tests/messages.rs | 2 ++ host/tests/p3.rs | 39 +++++++++++++++++++------- 5 files changed, 70 insertions(+), 42 deletions(-) diff --git a/extensions/messages/branch.ts b/extensions/messages/branch.ts index aba6f9d..ee3bc82 100644 --- a/extensions/messages/branch.ts +++ b/extensions/messages/branch.ts @@ -1,8 +1,8 @@ /** Fork a tree root, record its starter, and admit the first message exactly once. */ import { bool, int, obj, opt, str } from '@earendil-works/pi-schema' -import { configure, defineTool, ROOT_CONVERSATION_ID } from '@earendil-works/pi-durable' -import { discover } from '../subagents/definitions.ts' -import { childConfiguration } from '../subagents/children.ts' +import { defineTool, ROOT_CONVERSATION_ID } from '@earendil-works/pi-durable' +import { ModelsApi } from '../models/api.ts' +import { configureChild } from '../subagents/children.ts' import { LineageApi, Roots } from '../lineage/api.ts' import { imagesParameter } from './images.ts' import { send } from './send.ts' @@ -17,14 +17,13 @@ export const branchTool = (host: any) => defineTool({ 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') - const {definitions} = await discover(host, input.repo) + const {definitions} = await host.need(ModelsApi).discover(input.repo) 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 config = childConfiguration({definition: coordinator, parentAgent: rootAgent, cwd: input.repo}) - 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}, model: config.model, thinkingLevel: config.thinkingLevel}, 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) await host.harness().commit(async (tx: any) => { - await configure(tx, branch.id, config) + await configureChild(host, tx, branch.id, {definition: coordinator, parentAgent: rootAgent, cwd: input.repo}) ;(await tx.doc(host.metadata, branch.id)).label = 'coordinator' }, ctx) id = await api.memo('branch-id', branch.id, ctx) diff --git a/host/js/p3-test-bundle.js b/host/js/p3-test-bundle.js index 0a2192e..a1fe2a6 100644 --- a/host/js/p3-test-bundle.js +++ b/host/js/p3-test-bundle.js @@ -31696,7 +31696,7 @@ if (globalThis.p3Mode !== "unit") { const harness = await Harness.open(storage, { models, registry, settings: { retry: { enabled: false }, get extensions() { return registry.snapshot().installed(); } } }, BACKGROUND_CONTEXT); - await harness.root(BACKGROUND_CONTEXT, { agent: { model: { provider: "faux", modelId: "faux-1" }, instructions: "P3 role:ultimate", cwd: "/src/demo" } }); + 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"]] }); assert(result.code === 0, `World git ${args}: ${JSON.stringify(result)}`); @@ -31727,30 +31727,34 @@ if (globalThis.p3Mode !== "unit") { const loader = await loadExtensions(harness, registry, storage, models, BACKGROUND_CONTEXT, [tools]); assert(!loader.state().error, `Extension load: ${loader.state().error}`); try { - 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); + 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 === "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(() => { + }); + 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)); + } + await harness.waitForIdle(BACKGROUND_CONTEXT); + 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'", []); + assert(faults.length === 0, `Faulted tasks: ${JSON.stringify(faults)}`); + await fs.write("/p3-idle", "yes"); } - harness.resume(); - if (globalThis.p3Mode === "kill-worker") 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)); - } - await harness.waitForIdle(BACKGROUND_CONTEXT); - 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'", []); - assert(faults.length === 0, `Faulted tasks: ${JSON.stringify(faults)}`); - await fs.write("/p3-idle", "yes"); } finally { await loader.close(); await harness.close(BACKGROUND_CONTEXT); diff --git a/host/js/p3-test.ts b/host/js/p3-test.ts index e397b08..d258acb 100644 --- a/host/js/p3-test.ts +++ b/host/js/p3-test.ts @@ -24,7 +24,7 @@ if (globalThis.p3Mode !== 'unit') { models.setProvider(scriptedModel(rules, rule => Deno.core.ops.op_output(`scripted rule: ${rule}\n`))); 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); - await harness.root(context, { agent: { model: { provider: 'faux', modelId: 'faux-1' }, instructions: 'P3 role:ultimate', cwd: '/src/demo' } }); + 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']] }); assert(result.code === 0, `World git ${args}: ${JSON.stringify(result)}`); @@ -55,6 +55,9 @@ if (globalThis.p3Mode !== 'unit') { const loader = await loadExtensions(harness, registry, storage, models, context, [tools]); 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 === 'kill-worker') { const coordinator = await globalThis.p3Models.start({ agent: 'coordinator', cwd: '/src/demo' }, context); await fs.write('/coordinator', String(coordinator.id)); @@ -79,6 +82,7 @@ if (globalThis.p3Mode !== 'unit') { 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'); + } } finally { await loader.close(); await harness.close(context); } } // PENDING T49: @ultimate calls branch to create the coordinator, then receives one board.reply. diff --git a/host/tests/messages.rs b/host/tests/messages.rs index c6c176b..628d3b5 100644 --- a/host/tests/messages.rs +++ b/host/tests/messages.rs @@ -46,6 +46,7 @@ fn prepare(dir: &Path) -> anyhow::Result<()> { "extensions/subagents", "extensions/messages", "extensions/lineage", + "extensions/models", "extensions/core", "agents", ] { @@ -55,6 +56,7 @@ fn prepare(dir: &Path) -> anyhow::Result<()> { "extensions/subagents", "extensions/messages", "extensions/lineage", + "extensions/models", ] { for entry in std::fs::read_dir(source.join(directory))? { let entry = entry?; diff --git a/host/tests/p3.rs b/host/tests/p3.rs index 2d7b37d..233087f 100644 --- a/host/tests/p3.rs +++ b/host/tests/p3.rs @@ -22,7 +22,20 @@ fn git(repo: &Path, args: &[&str]) -> anyhow::Result { ); Ok(String::from_utf8(output.stdout)?.trim().into()) } -fn prepare(dir: &Path) -> anyhow::Result<()> { +fn copy_tree(source: &Path, destination: &Path) -> anyhow::Result<()> { + std::fs::create_dir_all(destination)?; + for entry in std::fs::read_dir(source)? { + let entry = entry?; + let target = destination.join(entry.file_name()); + if entry.file_type()?.is_dir() { + copy_tree(&entry.path(), &target)?; + } else { + std::fs::copy(entry.path(), target)?; + } + } + Ok(()) +} +fn prepare(dir: &Path, all_extensions: bool) -> anyhow::Result<()> { World::open(dir)?; let disk = dir.join("world"); let repo = disk.join("src/sausage"); @@ -56,14 +69,14 @@ fn prepare(dir: &Path) -> anyhow::Result<()> { target.join("wasm32-wasip2/release/git.wasm"), disk.join("bin/git.wasm"), )?; - for extension in ["subagents", "models", "core"] { - let destination = repo.join("extensions").join(extension); - std::fs::create_dir_all(&destination)?; - for entry in std::fs::read_dir(source.join("extensions").join(extension))? { - let entry = entry?; - if entry.path().extension().is_some_and(|ext| ext == "ts") { - std::fs::copy(entry.path(), destination.join(entry.file_name()))?; - } + if all_extensions { + copy_tree(&source.join("extensions"), &repo.join("extensions"))?; + } else { + for extension in ["subagents", "models", "core"] { + copy_tree( + &source.join("extensions").join(extension), + &repo.join("extensions").join(extension), + )?; } } std::fs::create_dir_all(repo.join("extensions/probe"))?; @@ -119,6 +132,12 @@ async fn scripted_model_replays() -> anyhow::Result<()> { run(dir.path(), "unit").await } #[tokio::test(flavor = "current_thread")] +async fn all_extensions_load() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + prepare(dir.path(), true)?; + run(dir.path(), "load-all").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?; @@ -128,7 +147,7 @@ async fn restart_child() -> anyhow::Result<()> { #[test] fn worker_sigkill_recovery() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; - prepare(dir.path())?; + prepare(dir.path(), false)?; let spawn = |mode: &str| { Command::new(std::env::current_exe().unwrap()) .args(["--exact", "restart_child", "--nocapture"])