185 lines
18 KiB
TypeScript
185 lines
18 KiB
TypeScript
/** World-loaded branches, routing, tally catch-up, memory, and delivery recovery. */
|
|
import {BACKGROUND_CONTEXT as context, awaitWithContext} from '@earendil-works/chord/context';
|
|
import {createModels} from '@earendil-works/pi-ai/models';
|
|
import {fauxProvider, fauxAssistantMessage, fauxToolCall} from '@earendil-works/pi-ai/providers/faux';
|
|
import {Harness, createRegistry, defineExtension, defineTool, LiveDoc} from '@earendil-works/pi-durable';
|
|
import {obj} from '@earendil-works/pi-schema';
|
|
import {loadExtensions, fs} from './extensions';
|
|
import {conversationChains} from './environment';
|
|
import {openStorage} from './sqlite';
|
|
const assert = (value: any, message: string) => {if (!value) throw new Error(message)};
|
|
const until = async (test: () => any) => {const deadline = Date.now() + 15000; while (!await test()) {if (Date.now() > deadline) throw new Error('Messages deadline'); await new Promise(r => setTimeout(r, 0))}};
|
|
const models = createModels(), faux = fauxProvider({tokensPerSecond: 100000}); models.setProvider(faux.provider);
|
|
const template = models.getModels('faux')[0];
|
|
models.setProvider({...faux.provider, getModels: () => [template, {...template, id: 'coordinator'}]});
|
|
const registry = createRegistry(), storage = await openStorage(globalThis.storagePath);
|
|
const harness = await Harness.open(storage, {models, registry, settings: {get extensions() {return registry.snapshot().installed()}}}, context);
|
|
const root = await harness.root(context, {agent: {model: {provider: 'faux', modelId: 'faux-1'}, cwd: '/src/sausage'}});
|
|
const gate = Promise.withResolvers<void>(), holding = new Set<number>();
|
|
const workerGate = Promise.withResolvers<void>();
|
|
let workerStarted = false;
|
|
const worker = defineExtension({name: 'worker-hold', tools: [defineTool({name: 'worker_hold', description: 'Hold a background worker', parameters: obj({}), replay: 'safe', execute: async (_args, _api, ctx) => {workerStarted = true; await awaitWithContext(workerGate.promise, ctx); return {output: [{type: 'text', text: 'worker released'}]}}})]});
|
|
const held = defineExtension({name: 'hold', tools: [defineTool({name: 'hold', description: 'Wait for the test gate', parameters: obj({}), replay: 'safe', execute: async (_args, api, ctx) => {holding.add(api.conversationId); await awaitWithContext(gate.promise, ctx); return {output: [{type: 'text', text: 'released'}]}}})]});
|
|
if (globalThis.messagesMode === 'kill') {
|
|
const install = registry.install.bind(registry);
|
|
registry.install = (extension: any) => install(extension.name !== 'messages' ? extension : {...extension, tasks: extension.tasks.map((task: any) => task.definition.name !== 'board.delivery' ? task : {definition: {...task.definition, phases: {...task.definition.phases, reply: async (task: any, runtime: any, ctx: any) => {
|
|
const intercepted = new Proxy(runtime, {get(target, key) {if (key === 'commit') return async () => {await fs.write('/report-ready', 'yes'); await awaitWithContext(new Promise(() => {}), ctx)}; const value = Reflect.get(target, key); return typeof value === 'function' ? value.bind(target) : value}});
|
|
await taskDefinitionReply(task, intercepted, ctx);
|
|
}}}})});
|
|
// Capture the original phase before the registry wrapper replaces it.
|
|
var taskDefinitionReply: any;
|
|
const wrapped = registry.install;
|
|
registry.install = (extension: any) => {if (extension.name === 'messages') taskDefinitionReply = extension.tasks.find((t: any) => t.definition.name === 'board.delivery').definition.phases.reply; return wrapped(extension)};
|
|
}
|
|
let memoryAnswers = 0;
|
|
const route = (transcript: any): any => {
|
|
faux.appendResponses([route]);
|
|
const system = transcript.messages.find((message: any) => message.role === 'system');
|
|
const systemText = typeof system?.content === 'string' ? system.content : (system?.content ?? []).filter((part: any) => part.type === 'text').map((part: any) => part.text).join('');
|
|
if (systemText.startsWith('You keep a software')) {
|
|
assert(JSON.stringify(transcript.messages).includes('Latest state'), 'Distillation must include the frozen abstract');
|
|
memoryAnswers++;
|
|
return fauxAssistantMessage(memoryAnswers === 1 ? 'not JSON' : JSON.stringify({ops: [{op: 'add', slug: 'landing-lesson', kind: 'lesson', line: 'A durable lesson', body: 'Keep the reason for the decision.'}]}));
|
|
}
|
|
const last = transcript.messages.findLast((m: any) => m.role !== 'system');
|
|
const text = typeof last.content === 'string' ? last.content : last.content.filter((p: any) => p.type === 'text').map((p: any) => p.text).join('');
|
|
if (last.role === 'user' && text.startsWith('CALL ')) {const input = JSON.parse(text.slice(5)); return fauxAssistantMessage([fauxToolCall(input.name, input.args)], {stopReason: 'toolUse'})}
|
|
if (last.role === 'user' && text === 'HOLD WORKER') return fauxAssistantMessage([fauxToolCall('worker_hold', {})], {stopReason: 'toolUse'});
|
|
if (last.role === 'user' && text === 'HOLD') return fauxAssistantMessage([fauxToolCall('hold', {})], {stopReason: 'toolUse'});
|
|
return fauxAssistantMessage(last.role === 'user' && /^(Reply|Message|Note) from/.test(text) ? 'Received' : `Answer: ${text}`);
|
|
};
|
|
faux.setResponses([route]);
|
|
const loader = await loadExtensions(harness, registry, storage, models, context, [held, worker]);
|
|
assert(!loader.state().error, `Extension load: ${loader.state().error}`);
|
|
const snapshot = await fs.read('/src/sausage/.git/sausage-last-good');
|
|
const base = `file:///run/sausage/extensions/${snapshot}/extensions`;
|
|
const {Started} = await import(`${base}/messages/addresses.ts`);
|
|
const {Roots} = await import(`${base}/lineage/api.ts`);
|
|
const {SubmissionLanded} = await import(`${base}/lineage/refs.ts`);
|
|
const {Tally, abstract} = await import(`${base}/lineage/server.ts`);
|
|
const {normalizeMemoryOp, parseMemoryOps, applyMemoryOps, readMemory, memorizeTask} = await import(`${base}/lineage/memory.ts`);
|
|
const {Seen, mainOf} = await import(`${base}/lineage/catchup.ts`);
|
|
const primaryId = await globalThis.subagentService.standing('standing', context);
|
|
const primary = (await harness.conversation(primaryId, context))!;
|
|
const entries = async (conversation: any) => (await conversation.context(context)).entries;
|
|
const replies = async () => (await entries(primary)).filter((e: any) => e.kind === 'board.reply');
|
|
const tool = async (conversation: any, name: string, args: any) => {
|
|
await (await conversation.submit({type: 'input', content: `CALL ${JSON.stringify({name, args})}`}, context)).wait(context);
|
|
const result = (await entries(conversation)).filter((e: any) => e.kind === 'pi.tool-result').at(-1)?.model?.[0];
|
|
assert(result && !result.isError, `Tool ${name}: ${JSON.stringify(result)}`);
|
|
return result;
|
|
};
|
|
const branchId = async (result: any) => Number(/Branch #(\d+)/.exec(result.content[0].text)![1]);
|
|
harness.resume();
|
|
try {
|
|
if (globalThis.messagesMode === 'kill') {
|
|
await tool(primary, 'branch', {repo: '/src/sausage', prompt: 'crash answer'});
|
|
await until(async () => {try {return await fs.read('/report-ready')} catch {return false}});
|
|
await new Promise(() => {});
|
|
} else if (globalThis.messagesMode === 'resume') {
|
|
await until(async () => (await replies()).length === 1);
|
|
await harness.waitForIdle(context);
|
|
assert((await replies()).length === 1, 'Restart duplicated the reply');
|
|
const branches = (await entries(root)).filter((e: any) => e.kind === 'lineage.branched');
|
|
assert(branches.length === 1, 'Restart duplicated the fork');
|
|
const branch = (await harness.conversation(branches[0].data.branch, context))!;
|
|
assert((await entries(branch)).filter((e: any) => e.kind === 'board.message' && e.data.reply).length === 1, 'Restart duplicated the first input');
|
|
} else {
|
|
// No git or exec member exists on this host: baseline reads must use the VFS.
|
|
const refHost = {fs, want: () => undefined};
|
|
for (const directory of ['/ref-probe', '/ref-probe/.git', '/ref-probe/.git/refs', '/ref-probe/.git/refs/heads']) await fs.mkdir(directory);
|
|
await fs.write('/ref-probe/.git/HEAD', 'ref: refs/heads/main\n');
|
|
await fs.write('/ref-probe/.git/refs/heads/main', 'a'.repeat(40));
|
|
assert((await mainOf(refHost, root.id, context, undefined, '/ref-probe')).sha === 'a'.repeat(40), 'Loose VFS ref');
|
|
await fs.remove('/ref-probe/.git/refs/heads/main');
|
|
await fs.write('/ref-probe/.git/packed-refs', `${'b'.repeat(40)} refs/heads/main\n`);
|
|
assert((await mainOf(refHost, root.id, context, undefined, '/ref-probe')).sha === 'b'.repeat(40), 'Packed VFS ref');
|
|
for (const directory of ['/ref-worktree', '/ref-probe/.git/worktrees', '/ref-probe/.git/worktrees/test']) await fs.mkdir(directory);
|
|
await fs.write('/ref-worktree/.git', 'gitdir: /ref-probe/.git/worktrees/test\n');
|
|
await fs.write('/ref-probe/.git/worktrees/test/HEAD', 'ref: refs/heads/main\n');
|
|
await fs.write('/ref-probe/.git/worktrees/test/commondir', '../..\n');
|
|
assert((await mainOf(refHost, root.id, context, undefined, '/ref-worktree')).sha === 'b'.repeat(40), 'Worktree common VFS ref');
|
|
const branch = (await harness.conversation(await branchId(await tool(primary, 'branch', {repo: '/src/sausage', prompt: 'first task'})), context))!;
|
|
await until(async () => (await replies()).length === 1); await harness.waitForIdle(context);
|
|
assert((await replies())[0].data.text === 'Received', 'First answer did not reach the starter');
|
|
assert((await entries(branch)).some((e: any) => e.kind === 'board.message' && e.data.text === 'first task'), 'The first message did not reach the branch');
|
|
assert((await harness.commit((tx: any) => tx.conversation(branch.id), context)).parent.conversationId === root.id, 'Branch must fork its root');
|
|
assert((await harness.snapshot(Started, branch.id, context)).by === primary.id, 'Missing starter');
|
|
const starts = (await globalThis.testHost.eventLog.recent()).filter((event: any) => event.name === 'subagents.started' && event.data.id === branch.id);
|
|
assert(starts.length === 1 && starts[0].data.model === 'faux/coordinator', `Effective branch model: ${JSON.stringify(starts)}`);
|
|
assert((await branch.agent(context)).model.modelId === 'coordinator', 'The branch must use its advertised model');
|
|
assert(JSON.stringify(await conversationChains(harness, context)(branch.id)) === JSON.stringify([String(branch.id), String(root.id), String(primary.id)]), 'Fork and starter grant ancestry');
|
|
const before = (await entries(primary)).length;
|
|
await (await branch.submit({type: 'input', content: 'later work'}, context)).wait(context);
|
|
await until(async () => (await entries(primary)).some((e: any) => e.kind === 'board.message' && e.data.from === branch.id && e.data.text === 'Answer: later work' && e.data.delivery !== 'note'));
|
|
await harness.waitForIdle(context);
|
|
assert((await entries(primary)).length > before, 'An idle branch must report');
|
|
assert((await entries(branch)).some((entry: any) => entry.kind === 'board.message' && entry.data.from === primary.id && entry.data.routed === true), 'The final answer must route back to the branch');
|
|
await until(async () => {
|
|
const routeTasks = (await harness.commit((tx: any) => tx.scanTasks({kind: 'board.route'}, 1000), context)).items;
|
|
assert(routeTasks.every((task: any) => task.state.status !== 'terminal' || task.state.outcome.status === 'completed'), `Final-answer routing fault: ${JSON.stringify(routeTasks)}`);
|
|
return routeTasks.length > 0 && routeTasks.every((task: any) => task.state.status === 'terminal');
|
|
});
|
|
await tool(branch, 'subagent', {action: 'start', agent: 'scout', task: 'HOLD', background: true});
|
|
await until(() => holding.size > 0);
|
|
await until(async () => (await entries(primary)).some((e: any) => e.kind === 'board.message' && e.data.from === branch.id && e.data.delivery === 'note'));
|
|
const activeChild = [...holding][0];
|
|
assert((await harness.snapshot(LiveDoc, activeChild, context))?.run, 'Route fixture must retain an active child');
|
|
gate.resolve(); await harness.waitForIdle(context);
|
|
await harness.commit((tx: any) => abstract(tx, branch.id, {text: 'Latest state', at: 3}), context);
|
|
const landing = {submission: 's1', conversationId: branch.id, at: 3, sha: 'abcdef0123456789', base: '0000000', target: 'main', lander: primary.id, divergence: 'None', title: 'Land me', commits: [{sha: 'abc', subject: 'feat: work'}], commitCount: 1, paths: ['work.txt']};
|
|
for (let i = 0; i < 2; i++) await harness.commit((tx: any) => globalThis.testHost.emit(tx, SubmissionLanded, landing), context);
|
|
await until(async () => (await entries(root)).some((e: any) => e.kind === 'lineage.note' && e.data.sha === landing.sha));
|
|
await harness.waitForIdle(context);
|
|
const landings = (await entries(root)).filter((e: any) => e.kind === 'lineage.note' && e.data.sha === landing.sha);
|
|
assert(landings.length === 1 && landings[0].data.abstract.text === 'Latest state', 'Landing must freeze its abstract once');
|
|
assert(!(await harness.snapshot(Tally, root.id, context)).abstracts[branch.id], 'Landing must clear the liquid abstract');
|
|
await globalThis.lineageService.remember(root.id, [{op: 'add', slug: 'testing', kind: 'decision', line: 'Test it', body: 'Every change has a test.'}], context);
|
|
const path = `/src/sausage/.sausage-memory/${root.id}.json`;
|
|
assert(JSON.parse(await fs.read(path)).memories.testing.body === 'Every change has a test.', 'Memory must live in a world file');
|
|
const original = await fs.read(path);
|
|
try {await globalThis.lineageService.remember(root.id, [{op: 'update', slug: 'testing', line: 'Changed'}, {op: 'delete', slug: 'missing'}], context); throw new Error('Invalid batch accepted')} catch (error) {assert(String(error).includes('no memory missing'), String(error))}
|
|
assert(await fs.read(path) === original, 'Invalid batch changed memory');
|
|
assert(normalizeMemoryOp({slug: 'Upper Case', line: 'a\nb', body: 'x'.repeat(3000)}).slug === 'upper-case', 'Memory normalization');
|
|
assert(parseMemoryOps('```json\n{"ops":[]}\n```').length === 0, 'Fenced memory ops');
|
|
await applyMemoryOps(globalThis.testHost, root.id, [{op: 'update', slug: 'testing', line: 'Once'}], 10, 'once');
|
|
await applyMemoryOps(globalThis.testHost, root.id, [{op: 'update', slug: 'testing', line: 'Twice'}], 20, 'once');
|
|
assert((await readMemory(globalThis.testHost, root.id)).memories.testing.line === 'Once', 'File receipt must deduplicate writes');
|
|
globalThis.summaryRef = {provider: 'faux', modelId: 'faux-1'};
|
|
const memoryTask = await harness.commit((tx: any) => tx.createTask(memorizeTask(globalThis.testHost), {root: root.id, submission: 's1', from: branch.id, at: 3, note: 'Why this choice matters'}, {conversationId: root.id, ownership: {kind: 'conversation'}, background: true}), context);
|
|
await harness.waitForTask(memoryTask, context);
|
|
assert(memoryAnswers === 2, 'Memory must retry a refused answer once');
|
|
assert((await readMemory(globalThis.testHost, root.id)).memories['landing-lesson'].source === `#${branch.id}/3`, 'Memory must retain landing provenance');
|
|
const briefBranch = (await globalThis.lineageService.branch({from: root.id, repo: '/src/sausage', prompt: 'Use memory'}, context)).id;
|
|
await harness.waitForIdle(context);
|
|
const briefConversation = (await harness.conversation(briefBranch, context))!;
|
|
assert(JSON.stringify(await briefConversation.context(context)).includes('<project_memory'), 'Branch memory brief');
|
|
assert((await entries(briefConversation)).some((e: any) => e.kind === 'pi.user' && e.model?.[0].content === 'Use memory'), 'lineage.prompt must deliver the direct branch input');
|
|
await fs.write('/src/sausage/catchup.txt', 'new');
|
|
const host = globalThis.testHost;
|
|
const baseline = (await mainOf(host, branch.id, context)).sha;
|
|
await harness.commit(async (tx: any) => {(await tx.doc(Seen, branch.id)).main = baseline}, context);
|
|
await host.git('/src/sausage', 'add', 'catchup.txt');
|
|
const committed = await host.exec(['git', '-c', 'user.name=Test', '-c', 'user.email=test@example.com', 'commit', '--no-gpg-sign', '-m', 'Catch-up commit\n\nSausage-Conversation: ' + branch.id], {cwd: '/src/sausage'});
|
|
assert(committed.code === 0, `Catch-up fixture commit: ${committed.stderr}`);
|
|
await (await branch.submit({type: 'input', content: 'after commit'}, context)).wait(context);
|
|
assert((await entries(branch)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'main' && e.data.text.includes('Catch-up commit') && e.data.text.includes(`#${branch.id}`)), 'Attributed commit catch-up');
|
|
assert((await entries(branch)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'main' && e.data.text.includes('Use memory')), 'Tally catch-up must include the new branch');
|
|
await harness.waitForIdle(context);
|
|
await tool(branch, 'subagent', {action: 'start', agent: 'coordinator', task: 'HOLD WORKER', background: true});
|
|
await until(() => workerStarted);
|
|
await branch.waitForIdle(context);
|
|
await tool(primary, 'notify', {id: branch.id, wake: false});
|
|
// Allow the notify invocation to reach its wait. The active worker must prevent delivery.
|
|
await new Promise(resolve => setTimeout(resolve, 50));
|
|
assert(!(await entries(primary)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished'), 'Idle coordinator with a live worker was called finished');
|
|
workerGate.resolve();
|
|
await until(async () => (await entries(primary)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished'));
|
|
await harness.waitForIdle(context);
|
|
const finished = (await entries(primary)).filter((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished');
|
|
assert(finished.length === 1, 'Worker completion must notify once');
|
|
assert(finished[0].data.text.includes('worker released'), 'Notice must contain the coordinator answer after the worker report');
|
|
assert((await harness.snapshot(Roots, root.id, context)).roots.includes(root.id), 'Listed root');
|
|
}
|
|
} finally {gate.resolve(); workerGate.resolve(); await loader.close(); await harness.close(context)}
|