sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

fix: scope wake settlement ownership and allow legacy result deletion

Browse Source
This commit is contained in:
amosblomqvist
2026-10-26 02:04:18 +02:00
parent e851432f4c
commit 90828cef49
4 changed files with 95 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/chat-handler.ts
+4 -3
View File
@@ -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).
src/server/chats.ts
+6 -1
View File
@@ -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<ChatDeleteOutcome, { status
jobIds: [],
};
}
const runningJobs = stores.jobStore.listJobs(chatId).filter((job) => 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,
src/server/jobs.ts
+16 -7
View File
@@ -88,6 +88,9 @@ interface QueuedSteer {
promise: Promise<void>;
/** 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<string>): 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<void> {
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);
tests/job-delivery-recovery.test.ts
+69 -3
View File
@@ -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<void>((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; });