From 90828cef49539d25d4ccac107e1f0a904ac79540 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Mon, 26 Oct 2026 02:04:18 +0200 Subject: [PATCH] fix: scope wake settlement ownership and allow legacy result deletion --- src/server/chat-handler.ts | 7 +-- src/server/chats.ts | 7 ++- src/server/jobs.ts | 23 ++++++--- tests/job-delivery-recovery.test.ts | 72 +++++++++++++++++++++++++++-- 4 files changed, 95 insertions(+), 14 deletions(-) diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 74ed3c8..8415688 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -484,7 +484,8 @@ export async function handleChatInput( // run's first LLM call (the native replacement for continuation // coalescing). Settlement still cannot beat the flow below: it needs // a streamed response, which never happens inside the constructor. - registerLiveAgent(chatId, constructed, input.type !== "steers" && input.type !== "continue"); + registerLiveAgent(chatId, constructed, input.type !== "steers" && input.type !== "continue", + input.type === "steers" ? new Set(appends.map((message) => message.id)) : undefined); // Debug recorder run marker (owner, 2026-10-04): correlates the // JSONL event stream with the record's turns; no-op unless // LH_PROVIDER_EVENTS=1. @@ -525,7 +526,7 @@ export async function handleChatInput( // merely because usage/extraction/other settlement work failed. const persisted = stores.chatStore.getChat(chatId)?.messages ?? []; reconcileUserSteers(chatId, persisted); - settleSteers(chatId, persisted, true); + settleSteers(chatId, persisted, true, constructed); }, onFinish: (turnMessages, isAborted) => { // Wave B: the record write is the agent's own final state — @@ -577,7 +578,7 @@ export async function handleChatInput( // resolve; stranded markers return to pending), then let // the pump take any stragglers down the idle wake. unregisterLiveAgent(chatId); - settleSteers(chatId, stamped); + settleSteers(chatId, stamped, false, constructed); // Composer steers (2026-10-03): delivered ids dropped, stranded // ones stay parked for the chat's next run (the registry is the // stranding safety net — the agent's undrained copy died above). diff --git a/src/server/chats.ts b/src/server/chats.ts index 2e6f2c9..857d3d1 100644 --- a/src/server/chats.ts +++ b/src/server/chats.ts @@ -1,4 +1,5 @@ import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts"; +import { isJobLive } from "../agent/subagents/aborts.ts"; import { coverageOf } from "../core/coverage.ts"; import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts"; import { titleFromPrompt } from "../core/contracts.ts"; @@ -77,7 +78,11 @@ function runningWorkBlocker(chatId: string): Extract job.status === "running"); + // Legacy records kept completed workers running until delivery. A saved + // result without a live loop is pending session data, not computing work; + // deletion must not depend on a prior recovery poll to recognize it. + const runningJobs = stores.jobStore.listJobs(chatId).filter((job) => + job.status === "running" && (!job.result || isJobLive(job.id))); if (runningJobs.length > 0) { return { ok: false, diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 98becef..f4b2cd2 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -88,6 +88,9 @@ interface QueuedSteer { promise: Promise; /** Wake cargo remains owned here, but must not steer into its own run. */ waking?: boolean; + /** The constructed vehicle, once startup claims a turn. Until then only + * the pump owns waking: an unrelated foreground settlement cannot free it. */ + wakeAgent?: Agent; /** A failed start waits for a learner run or process restart, not a timer. */ deferred?: boolean; resolve: () => void; @@ -286,10 +289,13 @@ function handOffEligible(chatId: string, agent: Agent): void { * markers are steered into the run right here. The settle seam cannot * beat the registration: settlement needs a streamed response, which * never happens inside the constructor call. */ -export function registerLiveAgent(chatId: string, agent: Agent, allowDeferred = true): void { +export function registerLiveAgent(chatId: string, agent: Agent, allowDeferred = true, wakeSteerIds?: ReadonlySet): void { state().liveAgents.set(chatId, agent); restorePendingDeliveries(chatId); - if (allowDeferred) for (const entry of state().queues.get(chatId) ?? []) entry.deferred = false; + for (const entry of state().queues.get(chatId) ?? []) { + if (allowDeferred) entry.deferred = false; + if (entry.waking && wakeSteerIds?.has(entry.steer.id)) entry.wakeAgent = agent; + } handOffEligible(chatId, agent); } @@ -369,7 +375,7 @@ export function restoreUncommittedAutomatic(chatId: string, agent: Agent, automa * This is what makes the live hand-off safe: pi owns no retry for a marker * stranded in a settled agent's queue, so the registry stays the truth and * the dead agent's copy is simply abandoned (GC'd with the agent). */ -export function settleSteers(chatId: string, persisted: { id?: string }[], failed = false): void { +export function settleSteers(chatId: string, persisted: { id?: string }[], failed = false, settledAgent?: Agent): void { const queue = state().queues.get(chatId); if (!queue || queue.length === 0) return; const deliveredIds = new Set(persisted.map((m) => m.id).filter(Boolean)); @@ -379,9 +385,11 @@ export function settleSteers(chatId: string, persisted: { id?: string }[], faile finalizeSteer(entry.steer); entry.resolve(); } else { - entry.handedTo = undefined; - entry.waking = false; - if (failed && jobIdOf(entry.steer)) entry.deferred = true; + const handedHere = entry.handedTo !== undefined && (!settledAgent || entry.handedTo === settledAgent); + const wakeHere = settledAgent !== undefined && entry.wakeAgent === settledAgent; + if (handedHere) entry.handedTo = undefined; + if (wakeHere) { entry.waking = false; entry.wakeAgent = undefined; } + if (failed && (handedHere || wakeHere) && jobIdOf(entry.steer)) entry.deferred = true; remaining.push(entry); } } @@ -577,7 +585,7 @@ async function pump(chatId: string): Promise { if (saved) noteSteersDurable(chatId, saved.messages); const owned = fire.filter((entry) => s.queues.get(chatId)?.includes(entry)); if (status === 403 || status === 409) { - for (const entry of owned) entry.waking = false; + for (const entry of owned) { entry.waking = false; entry.wakeAgent = undefined; } if (status === 403) { parked = true; return; } await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!); continue; @@ -627,6 +635,7 @@ state().stopConnectionWatch = onSubscriptionChange(() => { function deferOrReject(chatId: string, entry: QueuedSteer, error: Error): void { entry.waking = false; + entry.wakeAgent = undefined; if (jobIdOf(entry.steer)) { entry.deferred = true; console.error(`[steer ${entry.steer.id}] delivery remains pending:`, error.message); diff --git a/tests/job-delivery-recovery.test.ts b/tests/job-delivery-recovery.test.ts index 25618c7..8e34e6e 100644 --- a/tests/job-delivery-recovery.test.ts +++ b/tests/job-delivery-recovery.test.ts @@ -3,9 +3,10 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { spawnSync } from "node:child_process"; +import { Agent } from "@earendil-works/pi-agent-core"; import { after, afterEach, test } from "node:test"; import { createFauxCore, fauxAssistantMessage } from "@earendil-works/pi-ai"; -import type { LhMessage } from "../src/core/native-messages.ts"; +import { toLlmMessages, type LhMessage } from "../src/core/native-messages.ts"; import { buildSubagentSteer, type JobKind, type SteerMessage } from "../src/core/contracts.ts"; // Install isolation before importing any storage/configuration composition. @@ -186,16 +187,59 @@ test("failed start stays quiet across sweeps, then joins a learner run", async ( test("HTTP acceptance without a saved marker is not delivery completion or a retry trigger", async () => { const owner = chat(); const { job, steer } = pending(owner.id); let attempts = 0; - setTurnRunner(async () => { attempts++; assert(beginTurn(owner.id)); return 200; }); + const accepted = agent(); + setTurnRunner(async () => { + attempts++; assert(beginTurn(owner.id)); + registerLiveAgent(owner.id, accepted.value, false, new Set([steer.id])); + return 200; + }); const delivery = deliverJobResult(steer); await until(() => attempts === 1); await delay(30); assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending"); - unregisterLiveAgent(owner.id); endTurn(owner.id); settleSteers(owner.id, [], true); + unregisterLiveAgent(owner.id); endTurn(owner.id); settleSteers(owner.id, [], true, accepted.value); sweep(); await delay(900); assert.equal(attempts, 1); const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value); assert.equal(next.handed.length, 1); persist(owner.id, next.handed); unregisterLiveAgent(owner.id); endTurn(owner.id); await delivery; }); +test("a racing foreground settlement cannot release an outstanding staged wake", async () => { + const owner = chat(); const { steer } = pending(owner.id); + let started = false; + let release!: () => void; + const startup = new Promise((resolve) => { release = resolve; }); + let modelCopies = 0; + let nativeCopies = 0; + setTurnRunner(async (id, input) => { + assert.equal(input.type, "steers"); if (input.type !== "steers") return 500; + started = true; + // Subscription wake startup can await catalog refresh before claiming + // the session. Its cargo is not yet a durable input prefix. + await startup; + const faux = createFauxCore({}); + faux.setResponses([(context) => { + modelCopies = JSON.stringify(context.messages).split(result.brief).length - 1; + return fauxAssistantMessage("Fixture wake response"); + }]); + const wake = new Agent({ initialState: { model: faux.getModel(), messages: input.steers }, convertToLlm: toLlmMessages, streamFn: faux.streamSimple }); + assert(beginTurn(id)); + registerLiveAgent(id, wake, false, new Set(input.steers.map((message) => message.id))); + await wake.continue(); + nativeCopies = wake.state.messages.filter((message) => (message as LhMessage).id === steer.id).length; + persist(id, wake.state.messages.filter((message) => message.role !== "system") as LhMessage[]); + unregisterLiveAgent(id); endTurn(id); settleSteers(id, stores.chatStore.getChat(id)!.messages, false, wake); + return 200; + }); + const delivery = deliverJobResult(steer); + await until(() => started); + const foreground = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, foreground.value); + assert.equal(foreground.handed.length, 0, "outstanding wake owns the marker"); + unregisterLiveAgent(owner.id); endTurn(owner.id); settleSteers(owner.id, [], false, foreground.value); + release(); await delivery; + assert.equal(modelCopies, 1, "one result in the model's initial context"); + assert.equal(nativeCopies, 1, "no self-steering of staged wake cargo"); + assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1); +}); + test("canonical save wins even if delivery bookkeeping fails; no marker replay", async () => { const owner = chat(true); const { job, steer } = pending(owner.id); const delivery = deliverJobResult(steer); @@ -252,6 +296,28 @@ test("fresh process restores a pending delivery on registration without a jobs p assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1); }); +test("deleting legacy completed work needs no preceding recovery poll", async () => { + const owner = chat(true); + const old = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-delete" }); + stores.jobStore.updateJob(old.id, { result }); + const outcome = await deleteChatSession(owner.id, "preserve-graph"); + assert(outcome.ok); + assert.equal(stores.jobStore.getJob(old.id), null); + assert.equal(stores.chatStore.getChat(owner.id), null); +}); + +test("deletion still refuses a computing worker or a live turn", async () => { + const owner = chat(); + const computing = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "computing-delete" }); + const blocked = await deleteChatSession(owner.id, "preserve-graph"); + assert(!blocked.ok && blocked.code === "jobs_running"); + stores.jobStore.updateJob(computing.id, { status: "done" }); + assert(beginTurn(owner.id)); + const live = await deleteChatSession(owner.id, "preserve-graph"); + assert(!live.ok && live.code === "turn_in_flight"); + endTurn(owner.id); +}); + test("deleting a session discards its pending deliveries without starting a turn", async () => { const owner = chat(true); const { job, steer } = pending(owner.id); let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });