Merge T55 item 1: @ultimate branch round trip with D63 ancestry (T55, P3-p part)
This commit is contained in:
+118
-55
@@ -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);
|
||||
|
||||
+30
-1
@@ -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.
|
||||
|
||||
+14
-2
@@ -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?;
|
||||
|
||||
Reference in New Issue
Block a user