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

test: regenerate P2 bundle with patched pi-durable

Browse Source
This commit is contained in:
surma
2026-11-07 07:03:07 +00:00
parent 7503f093d0
commit 9d40a3fd73
1 changed files with 58 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/js/p2-test-bundle.js
+58 -11
View File
@@ -8940,6 +8940,9 @@ var Transaction = class {
this.#assertOpen();
return record;
}
beforeInput(conversationId, input) {
return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
}
appendEntry(first, second, third) {
if (typeof first === "number")
return this.#appendEntry(first, second);
@@ -10122,6 +10125,7 @@ var SessionImpl = class {
if (this.#documents.get(id)?.record.id === recordId)
this.#dropDocument(id);
},
beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -10446,6 +10450,9 @@ var SessionImpl = class {
*/
report(_error, _info = LISTENER, _context) {
}
async beforeInput(_tx, _id, _input, _context) {
return [];
}
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
@@ -12250,10 +12257,14 @@ var InboxDoc = defineDoc({
initial: () => ({ items: [] }),
checkpointWhen: (value2) => value2.items.length === 0
});
async function prepareBoundary(tx, conversationId, modes) {
async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
const head = (await tx.latestHeadMarker(conversationId))?.head;
const inbox = await tx.doc(InboxDoc, conversationId);
const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
return {
entries: entries2,
conversationId,
inbox,
steeringMode: modes.steeringMode,
@@ -12289,6 +12300,9 @@ async function applyBoundary(tx, boundary, at, now) {
}
const placed = [];
for (const index of users) {
if (placed.length === 0)
for (const draft of boundary.entries)
await tx.appendEntry(conversationId, draft);
const item = items[index];
const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
tx.placeSubmission(item.id, entry2.id);
@@ -12301,10 +12315,11 @@ async function applyBoundary(tx, boundary, at, now) {
return { users: placed, first, reset };
}
function appendInput(tx, conversationId, content, entry2, now) {
return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
}
function inputDraft(content, entry2, now) {
const message = { role: "user", content, timestamp: now };
if (entry2 === void 0)
return tx.appendEntry(UserEntry, conversationId, { model: [message] });
return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
}
function pickInputEntry({ kind, data }) {
const picked = data === void 0 ? { kind } : { kind, data };
@@ -13016,6 +13031,7 @@ var TaskScheduler = class {
#unsubscribeRegistry = () => {
};
#enabled = false;
#held = false;
#readOnly;
#closing = false;
#dirty = false;
@@ -13044,6 +13060,14 @@ var TaskScheduler = class {
this.#context = options.context;
this.#readOnly = options.readOnly === true;
}
/** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
hold() {
this.#held = true;
}
release() {
this.#held = false;
this.#kick();
}
/**
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -14498,6 +14522,8 @@ var TaskScheduler = class {
for (const record of plan.ready) {
const id = record.id;
const mode = record.abortRequested ? "abort" : "run";
if (mode === "run" && this.#held)
continue;
const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -14683,7 +14709,7 @@ var TaskScheduler = class {
const tasks = [];
for (const record of this.#live.values())
tasks.push({ record, state: this.#inspectTask(record, snapshot) });
const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
return { scheduling, tasks };
}
#inspectTask(record, snapshot) {
@@ -14838,6 +14864,10 @@ var TaskScheduler = class {
* invocation continues.
*/
#progress(tx, current, state, extension, context) {
if (this.#held) {
tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
return false;
}
state.snapshot = this.#registry.snapshot();
const next = state.snapshot.task(current.kind);
if (next === state.task)
@@ -17571,7 +17601,7 @@ async function prepareFailure(tx, runtime) {
const policy = settings.afterFailedRun;
if (policy !== "continue")
return { policy };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
}
async function endFailedRun(tx, runtime, live, failure, end) {
const conversationId = runtime.conversationId;
@@ -17739,7 +17769,7 @@ async function answer(runtime, message, context) {
const afterRun = await asksAfterRun(runtime, context);
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const entry2 = await appendAssistant(tx, conversationId, message);
const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -17812,7 +17842,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
async function endAnsweredRun(runtime, answer2, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -17885,7 +17915,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
const maxTurns = (await runtime.agent(context)).maxTurns;
const afterRun = terminate && await asksAfterRun(runtime, context);
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
if (added.length > 0)
await addTools(tx, conversationId, added);
const live = await tx.doc(LiveDoc, conversationId);
@@ -18301,13 +18331,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
if (draft.type === "continue") {
if (busy)
throw new ConversationBusy(conversationId);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
const { users, first } = await applyBoundary(tx, boundary2, "final", now);
await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
return { id: id2, status: "placed" };
}
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
if (boundary === void 0 || boundary.inbox.items.length > 0) {
const { id: id2 } = await tx.createSubmission({
...fields,
@@ -18345,6 +18375,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
const write = { ...fields, type: "write", status: "done", entry: entry3.id };
return { id: (await tx.createSubmission(write)).id, status: "done" };
}
for (const before of boundary.entries)
await tx.appendEntry(conversationId, before);
const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
const { id } = await tx.createSubmission(input);
@@ -19360,6 +19392,15 @@ var HarnessImpl = class extends SessionImpl {
create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
};
}
async beforeInput(tx, conversationId, input, context) {
const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
const entries2 = [];
for (const { handlers } of agentHooks(agent, "pi.generation")) {
const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
entries2.push(...result?.entries ?? []);
}
return entries2;
}
/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
async resolveAgent(id, snapshot, context) {
const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -19382,6 +19423,12 @@ var HarnessImpl = class extends SessionImpl {
this.assertUsable();
this.#tasks.resume();
}
hold() {
this.#tasks.hold();
}
release() {
this.#tasks.release();
}
wake() {
this.assertUsable();
return this.#tasks.wake();