From 9d40a3fd735084965d5c67b6b2a51c0c042f9a22 Mon Sep 17 00:00:00 2001 From: Surma Date: Sat, 7 Nov 2026 07:03:07 +0000 Subject: [PATCH] test: regenerate P2 bundle with patched pi-durable --- host/js/p2-test-bundle.js | 69 ++++++++++++++++++++++++++++++++------- 1 file changed, 58 insertions(+), 11 deletions(-) diff --git a/host/js/p2-test-bundle.js b/host/js/p2-test-bundle.js index a5347be..e287af4 100644 --- a/host/js/p2-test-bundle.js +++ b/host/js/p2-test-bundle.js @@ -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();