surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge T55 step 1: messages branch uses the models service; test that loads every extension (fixes the T49/T50 conflict)

Browse Source
This commit is contained in:
surma
2026-11-07 14:41:46 +00:00
parent fe9306c38b a66a933dc4
commit ca9bd813db
5 changed files with 70 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/messages/branch.ts
+6 -7
View File
@@ -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)
host/js/p3-test-bundle.js
+28 -24
View File
@@ -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);
host/js/p3-test.ts
+5 -1
View File
@@ -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.
host/tests/messages.rs
+2
View File
@@ -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?;
host/tests/p3.rs
+29 -10
View File
@@ -22,7 +22,20 @@ fn git(repo: &Path, args: &[&str]) -> anyhow::Result<String> {
);
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"])