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

Merge durable background job delivery and recovery

Browse Source
This commit is contained in:
amosblomqvist
2026-10-26 02:39:59 +02:00
parent 85329290f2 90828cef49
commit 6370c9d618
19 changed files with 730 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/artifacts/dispatch-card.tsx
+2 -2
View File
@@ -15,7 +15,7 @@ import { BrailleSpinner } from "@/components/braille-spinner";
* *
* job unknown → "dispatched" pulse (the jobs feed hasn't seen it yet) * job unknown → "dispatched" pulse (the jobs feed hasn't seen it yet)
* running → live activity card (activity line + elapsed) * running → live activity card (activity line + elapsed)
* done/error → quiet one-liner ("delivered" / "failed") — the artifact * done/error → quiet one-liner ("finished" / "failed") — the artifact
* itself arrives in the steer message right below, so the * itself arrives in the steer message right below, so the
* box must not duplicate it. * box must not duplicate it.
*/ */
@@ -54,7 +54,7 @@ export function DispatchCard({ toolCallId, kind }: { toolCallId: string; kind: J
> >
{failed ? <XIcon className="size-3.5 text-red-500" /> : <CheckIcon className="size-3.5" />} {failed ? <XIcon className="size-3.5 text-red-500" /> : <CheckIcon className="size-3.5" />}
<span className="font-medium uppercase tracking-wide">{label}</span> <span className="font-medium uppercase tracking-wide">{label}</span>
<span>{failed ? "failed" : "delivered"}</span> <span>{failed ? "failed" : "finished"}</span>
</div> </div>
); );
} }
docs/architecture.md
+17 -11
View File
@@ -313,22 +313,28 @@ sequenceDiagram
### 5.1 The wake pipeline (Workbench → Classroom) ### 5.1 The wake pipeline (Workbench → Classroom)
On sub-agent completion the runner builds a **steer message** — a user-role `UIMessage` whose text part is the model-facing result and whose `data-subagent-artifact` part carries the full result for the client (provider conversion drops data parts) — and hands it to the wake pipeline: **Worker completion and session delivery are separate lifecycles.** Dispatch persists the tool input on the job before spawning. On completion, the runner saves its result, terminal worker status (`done`/`error`), and `delivery: {kind: "result", status: "pending"}` in one job-record update. It then queues the native custom **steer message** without awaiting delivery. Its content is model-facing text; details carry the UI artifact. Worker resources are released and the Background agents row disappears even if an unanswered question holds the result indefinitely. Dispatch cards say “finished,” not “delivered.” Only computing workers can be aborted; abort saves an independent pending abort notice.
```mermaid ```mermaid
flowchart TB flowchart TB
R["job runner finishes sub-agent"] -->|"deliver(steer) — injectable seam"| ENQ["enqueue steer in per-chat queue (globalThis)"] R["save worker outcome + pending delivery; release worker"] --> ENQ["one registry entry per steer ID, including wake cargo"]
ENQ --> PUMP["pump(chatId) — one loop per chat (pumping set serializes)"] ENQ --> LIVE{"live agent and background consent?"}
PUMP --> GATE{"turn-free AND no pending question? (400ms poll)"} LIVE -->|yes| STEER["pi steering at native boundaries; answer run takes parked results"]
GATE -->|no| PUMP LIVE -->|no| GATE["wait for turn-free, question-free, unpaused, consented session"]
GATE -->|yes| REFRESH["re-fetch fresh record, re-check steerDelivered (claim race)"] GATE --> RUN["idle wake through the normal turn composition"]
REFRESH --> RUN["turnRunner(chatId, record + steer) — the SAME handleChatRequest path"] RUN -->|"409"| BACKOFF["existing backoff 500→1000→2000→5000ms"]
RUN -->|"200"| FIN["finalizeSteer → finalizeJob (done/error)<br/>— status flips only after the wake turn persists"] BACKOFF --> GATE
RUN -->|"409"| BACKOFF["backoff 500→1000→2000→5000ms → retry"] RUN -->|"403"| HOLD["retain ownership until consent permits delivery"]
CRASH["crash recovery: listJobs → recoverStaleJobs (stale >15min)<br/>re-deliver results / respawn via setJobRespawn"] --> PUMP RUN -->|"other failed start"| QUIET["quiet pending; next learner run or restart recovery"]
RUN --> SAVE["canonical marker persistence, not merely HTTP 200"]
STEER --> SAVE
SAVE --> FIN["acknowledge delivery; worker outcome unchanged"]
CRASH["lazy recovery + live-agent registration restore known pending jobs"] --> ENQ
``` ```
Job status flips done/error only after the wake turn persists. Delivery is **exactly-once**: deterministic steer ids (`steer_<jobId>`, `steer_probe_<chatId>`), `steerDelivered` checks at every claim point, and `claimSteers` finalizing in-transcript steers instead of re-appending. The client adopts the wake turn through the same `turnInFlight` machinery as returning mid-stream (busy shimmer, composer gate, append-only `mergeAdopted` adoption) and renders steers as artifacts — never a bubble. The per-chat registry retains ownership across live handoff, idle-wake flight, and 409/403 return; repeated submissions share one entry/promise. A deterministic marker ID in the **saved conversation** completes delivery. Streaming visibility alone does not. A failed bookkeeping acknowledgement cannot replay a saved marker, and later teaching/model failure does not reopen delivery or retry the accepted turn. Failed starts leave results durably pending without a new error UI or retry timer; the in-memory entry stays parked until a learner run or process restart. Normal question/dismissal waits and consent deferrals are not failures.
Recovery reads saved job input before legacy transcript lookup. Missing research questions fail honestly without invoking the researcher. Legacy running jobs with saved results become finished workers with pending delivery; legacy terminal jobs without delivery metadata remain untouched, rather than guessing that old outcomes should be replayed. Pending result and abort notices are restored before a newly registered agent's first steering poll, even without a preceding jobs-feed read. Deleting a session discards its pending delivery obligations; only live turns and computing workers block deletion. These guarantees use the existing JSON storage and single-process delivery registry, not a transactional queue or a promise of exactly-once model execution.
Sub-agents **verify, never one-shot**: the demo-builder headlessly executes its HTML in system Chrome before delivery; the visual-maker renders each SVG and looks at it (needs a vision-capable model — the `vision` slot in settings). Sub-agents **verify, never one-shot**: the demo-builder headlessly executes its HTML in system Chrome before delivery; the visual-maker renders each SVG and looks at it (needs a vision-capable model — the `vision` slot in settings).
src/agent/subagents/aborts.ts
+5 -7
View File
@@ -41,18 +41,16 @@ export function unregisterJobAbort(jobId: string): void {
* the liveness gate for the stale-job sweep: a "running" record with a * the liveness gate for the stale-job sweep: a "running" record with a
* live registration belongs to a loop this process owns (hung or slow, it * live registration belongs to a loop this process owns (hung or slow, it
* is the one truth — never re-spawn beside it); without one the record is * is the one truth — never re-spawn beside it); without one the record is
* an orphan of a process death and recovers immediately. The span is * an orphan of a process death and recovers immediately. The registration
* loop-start to delivery-completion: the runner's unregistering finally * spans computation and cleanup only. Finished outcomes have independent,
* runs only after deliver() resolves, by which point finalizeJob has * persisted delivery obligations; no worker waits for their consumption. */
* flipped the record — so absence of registration is unambiguous. */
export function isJobLive(jobId: string): boolean { export function isJobLive(jobId: string): boolean {
return registry().has(jobId); return registry().has(jobId);
} }
/** Abort a running job's loop. Returns false when no controller is live — /** Abort a running job's loop. Returns false when no controller is live —
* the job's loop is not executing (crashed, or in the delivering-result * the job's loop is not executing. The endpoint separately refuses abort
* phase where the runner has already returned); the caller then owns the * for a finished worker, even while its session delivery remains pending. */
* whole abort on the record side. */
export function fireJobAbort(jobId: string): boolean { export function fireJobAbort(jobId: string): boolean {
const controller = registry().get(jobId); const controller = registry().get(jobId);
if (!controller) return false; if (!controller) return false;
src/agent/subagents/runner.ts
+35 -27
View File
@@ -3,6 +3,7 @@ import type { SearchProvider } from "../../core/search.ts";
import { import {
accumulateJobUsage, accumulateJobUsage,
buildSubagentSteer, buildSubagentSteer,
isSubagentFailure,
piUsageToStep, piUsageToStep,
type DemoResult, type DemoResult,
type DemoToolInput, type DemoToolInput,
@@ -40,13 +41,13 @@ import { runVisualMaker } from "./visual-maker.ts";
* (the job tools' execute fires this via the spawn seam, fire-and-forget), * (the job tools' execute fires this via the spawn seam, fire-and-forget),
* updating the job's activity line on every iteration. On completion the * updating the job's activity line on every iteration. On completion the
* runner builds the STEER MESSAGE (buildSubagentSteer) and hands it to * runner builds the STEER MESSAGE (buildSubagentSteer) and hands it to
* `deliver`; the wake pipeline fires the server-side teacher turn and * `deliver` without waiting. The worker's final outcome and pending
* flips the job to done/error when the lesson has actually resumed. * delivery are saved together; the session owns delivery from then on.
* *
* On failure the result is the failure shape (empty html/svg + "FAILED: …" * On failure the result is the failure shape (empty html/svg + "FAILED: …"
* notes; research: failure brief) — and it STILL steers, because the * notes; research: failure brief) — and it STILL steers, because the
* teacher must learn the outcome. If delivery itself fails, the job is * teacher must learn the outcome. Failed delivery starts stay quietly
* marked error (the learner's activity card shows it). * pending and never rewrite the worker's computation outcome.
*/ */
/** /**
@@ -67,9 +68,9 @@ export interface JobRunnerDeps {
chatStore: ChatStore; chatStore: ChatStore;
jobStore: JobStore; jobStore: JobStore;
/** /**
* Hand a finished job's steer to the wake pipeline — resolves when the * Hand a finished job's steer to the session delivery registry. The
* steer is delivered (the wake turn completed and the job status * promise tracks durable marker acceptance, but the worker never waits
* flipped); rejects when delivery fails unrecoverably. * on it. Its persisted pending obligation survives process restart.
*/ */
deliver: (steer: SteerMessage) => Promise<void>; deliver: (steer: SteerMessage) => Promise<void>;
/** Settings-resolved engine per role (model slots wave, 2026-09-20): /** Settings-resolved engine per role (model slots wave, 2026-09-20):
@@ -96,8 +97,8 @@ export interface JobRunnerDeps {
export interface JobRunner { export interface JobRunner {
/** Run one job's sub-agent loop to completion (background; never throws). /** Run one job's sub-agent loop to completion (background; never throws).
* The tool input rides along from dispatch (the spawn fires mid-stream, * The tool input rides along from dispatch (the spawn fires mid-stream,
* before the tool part is persisted — only crash-recovery re-spawns * before the tool part is persisted). Recovery reads saved job input,
* resolve the input from the transcript). */ * with transcript fallback for legacy jobs. Delivery is not awaited. */
runJob(jobId: string, input?: JobToolInput): Promise<void>; runJob(jobId: string, input?: JobToolInput): Promise<void>;
} }
@@ -270,14 +271,15 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
} }
} }
/** Resolve the job's tool input: dispatch passes it directly; only a /** Dispatch input wins, then the job's saved arguments. Legacy jobs
* crash-recovery re-spawn (no input) falls back to scanning the * without saved input fall back to the durable transcript, which may
* transcript (by then onFinish has long persisted the part). */ * lack a call if the dispatching turn died before settlement. */
function resolveInput( function resolveInput(
job: JobRecord, job: JobRecord,
input: JobToolInput | undefined, input: JobToolInput | undefined,
): JobToolInput | undefined { ): JobToolInput | undefined {
if (input) return input; if (input) return input;
if (job.input !== undefined) return job.input as JobToolInput;
const chat = chatStore.getChat(job.chatId); const chat = chatStore.getChat(job.chatId);
if (!chat) return undefined; if (!chat) return undefined;
// Wave D: the record is native history — the scan reads the native // Wave D: the record is native history — the scan reads the native
@@ -295,27 +297,28 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
} }
/** /**
* The tail of every job: store the result, build the steer, hand it to * Save the computation outcome and session delivery obligation together.
* the wake pipeline. The pipeline flips the status when the lesson has * Worker liveness ends after cleanup, not after a lesson consumes the result.
* actually resumed — until then the job reads "running" so the client's
* activity card keeps polling and re-syncs once, at the end.
*/ */
async function finishJob(jobId: string, result: ResearchResult | DemoResult | VisualResult | PdfResult): Promise<void> { async function finishJob(jobId: string, result: ResearchResult | DemoResult | VisualResult | PdfResult): Promise<void> {
const job = jobStore.getJob(jobId); const job = jobStore.getJob(jobId);
if (!job) return; if (!job) return;
// Aborted mid-run or mid-delivery (abortJob flips the record before // Aborted mid-run (abortJob flips the record before
// the loop unwinds): no result write, no steer — the abort path owns // the loop unwinds): no result write, no steer — the abort path owns
// the record and queues the abort steer itself. // the record and queues the abort steer itself.
if (job.status !== "running") return; if (job.status !== "running") return;
jobStore.updateJob(jobId, { result, activity: "delivering result" }); const failed = isSubagentFailure(job.kind, result);
const steer = buildSubagentSteer(job, result); jobStore.updateJob(jobId, {
try { result,
await deliver(steer); status: failed ? "error" : "done",
} catch (err) { activity: failed ? "failed" : "finished",
const reason = err instanceof Error ? err.message : String(err); delivery: { kind: "result", status: "pending" },
console.error(`[job ${jobId}] steer delivery failed:`, err); });
jobStore.updateJob(jobId, { status: "error", error: reason, activity: "delivery failed" }); // The saved result now belongs to the session. Do not keep the worker,
} // abort registration, or temporary resources alive behind a question.
void deliver(buildSubagentSteer(job, result)).catch((err: unknown) => {
console.error(`[job ${jobId}] pending delivery:`, err);
});
} }
/** Chat-record + input resolution shared by all three runners. */ /** Chat-record + input resolution shared by all three runners. */
@@ -343,7 +346,12 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
const { job } = ctx; const { job } = ctx;
const input = (passedInput ?? resolveInput(job, passedInput)) as ResearchToolInput | undefined; const input = (passedInput ?? resolveInput(job, passedInput)) as ResearchToolInput | undefined;
const question = typeof input?.question === "string" && input.question.trim() ? input.question : "(no question)"; if (!(typeof input?.question === "string" && input.question.trim())) {
jobStore.updateJob(jobId, { error: "research question unavailable" });
await finishJob(jobId, { brief: "Research failed: the research question is unavailable.", sources: [] });
return;
}
const question = input.question;
const context = typeof input?.context === "string" && input.context.trim() ? input.context : undefined; const context = typeof input?.context === "string" && input.context.trim() ? input.context : undefined;
const prompt = [ const prompt = [
src/agent/tools/jobs.ts
+1 -1
View File
@@ -78,7 +78,7 @@ function dispatch(
input: ResearchInput | DemoInput | VisualInput | PdfInput, input: ResearchInput | DemoInput | VisualInput | PdfInput,
): string { ): string {
const existing = ctx.jobStore.findJobByToolCallId(ctx.chatId, toolCallId); const existing = ctx.jobStore.findJobByToolCallId(ctx.chatId, toolCallId);
const job = existing ?? ctx.jobStore.createJob({ kind, chatId: ctx.chatId, toolCallId }); const job = existing ?? ctx.jobStore.createJob({ kind, chatId: ctx.chatId, toolCallId, input });
if (!existing) ctx.spawn(job, input); if (!existing) ctx.spawn(job, input);
return `Dispatched (job ${job.id}). ${ack}`; return `Dispatched (job ${job.id}). ${ack}`;
} }
src/core/contracts/jobs.ts
+28 -32
View File
@@ -21,6 +21,12 @@ export interface JobRecord {
chatId: string; chatId: string;
/** The tool part this job fulfills. */ /** The tool part this job fulfills. */
toolCallId: string; toolCallId: string;
/** Dispatch-time arguments, saved before spawn. Older jobs resolve from history. */
input?: unknown;
/** Session-owned delivery, independent of the worker's terminal outcome.
* Absent on legacy records: old terminal jobs must never be replayed. */
delivery?: { kind: "result" | "abort"; status: "pending" | "delivered" };
/** Worker computation only; a pending delivery does not keep it running. */
status: JobStatus; status: JobStatus;
/** Latest human-readable activity line, e.g. `web_search: "sigma.js v3"`. */ /** Latest human-readable activity line, e.g. `web_search: "sigma.js v3"`. */
activity: string; activity: string;
@@ -148,42 +154,32 @@ export function accumulateJobUsage(
* sourced briefs) — tracking a live activity line in the job record. * sourced briefs) — tracking a live activity line in the job record.
* ONE central client poll of GET /api/jobs?chatId=… (in Chat) drives every * ONE central client poll of GET /api/jobs?chatId=… (in Chat) drives every
* job surface: the transcript dispatch cards (live activity by * job surface: the transcript dispatch cards (live activity by
* toolCallId, quiet one-liner after delivery), the sub-agents sidebar * toolCallId, quiet one-liner after worker completion), the sub-agents sidebar
* panel, and the single re-sync when a job settles. * panel, and the single re-sync when a job settles.
* *
* 3. On completion the runner builds a STEER MESSAGE (buildSubagentSteer) * 3. Dispatch saves input on the job before spawn. On completion the
* and hands it to the wake pipeline (src/server/jobs.ts): a user-role * runner saves result + terminal worker status + pending delivery in
* result-carrying native marker — the content is what the MODEL needs * one job update, then queues buildSubagentSteer WITHOUT waiting for
* (the research brief, the builder's notes), plus a data-subagent-artifact * delivery. The native custom marker's content is model-facing text;
* part with what the UI renders (the full result: brief + sources / * details carry the full artifact payload for the UI. Worker resources
* SVG / demo HTML). Provider conversion drops data parts, so artifact * are released even if a question keeps the result pending indefinitely.
* payloads never enter the teacher's context. metadata.subagentResult * Only computing workers appear in Background agents or accept abort.
* identifies the job; the UI renders these as sub-agent result cards,
* NEVER as learner interjections.
* *
* 4. The wake pipeline delivers the steer by firing a SERVER-SIDE teacher * 4. Session delivery owns one registry entry per deterministic steer ID,
* turn through the same turn composition as a client request * including while a wake is in flight or a 409/403 returns its cargo.
* (handleChatRequest) — write-ahead persist, turn liveness, extraction * A live agent takes it via pi steering; registration restores saved
* trigger: one code path. A per-chat serialized queue: a steer waits * pending jobs before the answer run's first poll. Otherwise the idle
* while a turn is in flight OR a question tool part is pending (an * wake uses the normal turn composition and respects pending questions,
* unanswered client-side tool call breaks provider conversion — the * dismissal, and background consent. A saved transcript marker completes
* steer fires right after the learner's continuation), then fires; a 409 * delivery; HTTP 200 or streamed visibility alone does not.
* (a client turn raced in) retries. Delivery is idempotent per jobId, so
* a crashed wake can be re-entered by the stale-job recovery sweep. The
* job's status flips to done/error only AFTER the wake turn completes —
* a polling client that sees the flip re-syncs once and gets the
* finished record.
* *
* CONTINUATION COALESCING: the idle window above can starve — a * Failed starts remain quietly pending until a learner run or restart
* quiz-chaining teacher ends every turn with a new pending check, so * recovery, without new timers, visible delivery errors, or worker-status
* the gates never stay open long enough for a wake to fire. The fix: * changes. Existing 409 backoff and consent deferral remain. Accepted
* when a client turn wins the turn registry, the handler CLAIMS every * turns are never automatically replayed. Legacy running records with a
* queued, undelivered steer (jobs.ts claimSteers) and appends it after * result become finished/pending; legacy terminal records with no
* the learner's answer in that very turn — conversion-safe, because the * delivery metadata remain untouched. Session deletion discards pending
* answered quiz carries its output. Every answer flushes pending * deliveries rather than treating them as running work.
* steers immediately; the wake pump remains for turns that end with no
* pending question. The pump re-checks steerDelivered right before
* firing, so a claim racing the pump converges to exactly one delivery.
* *
* 5. No pending job parts ever exist, so the composer never gates on jobs * 5. No pending job parts ever exist, so the composer never gates on jobs
* (the learner can interject mid-job; the steer queues behind their * (the learner can interject mid-job; the steer queues behind their
src/core/ports.ts
+2
View File
@@ -136,6 +136,8 @@ export interface CreateJobInit {
toolCallId: string; toolCallId: string;
/** First activity line (defaults to "starting"). */ /** First activity line (defaults to "starting"). */
activity?: string; activity?: string;
/** Saved on the job before its worker is spawned. */
input?: unknown;
} }
/** Mergeable fields — identity (id, kind, chatId, toolCallId, createdAt) is immutable. */ /** Mergeable fields — identity (id, kind, chatId, toolCallId, createdAt) is immutable. */
src/server/chat-handler.ts
+15 -10
View File
@@ -17,7 +17,7 @@ import { createExtractionPipeline } from "../agent/subagents/extraction.ts";
import type { ModelRole } from "../../config.ts"; import type { ModelRole } from "../../config.ts";
import { import {
deliverJobResult, deliverJobResult,
noteSteerVisible, noteSteersDurable,
reconcileUserSteers, reconcileUserSteers,
registerLiveAgent, registerLiveAgent,
restoreUncommittedAutomatic, restoreUncommittedAutomatic,
@@ -58,7 +58,7 @@ import {
type TurnInput, type TurnInput,
} from "../core/input-events.ts"; } from "../core/input-events.ts";
import { steerDelivered } from "../core/contracts.ts"; import { steerDelivered } from "../core/contracts.ts";
import type {CallUsage, SteerMessage} from "../core/contracts.ts"; import type {CallUsage} from "../core/contracts.ts";
/** /**
* One chat turn over the server-owned native record. Framework-free (plain * One chat turn over the server-owned native record. Framework-free (plain
@@ -259,7 +259,8 @@ export async function handleChatInput(
chatId, chatId,
); );
const compactedRecord = stores.chatStore.getChat(chatId); const compactedRecord = stores.chatStore.getChat(chatId);
if (compactedRecord) record = compactedRecord; if (!compactedRecord) return Response.json({ error: "Unknown chat" }, { status: 404 });
record = compactedRecord;
// The input's native appends (pure; the honest 400 for an answer with no // The input's native appends (pure; the honest 400 for an answer with no
// pending question call). Computed BEFORE beginTurn so a refused input // pending question call). Computed BEFORE beginTurn so a refused input
@@ -329,9 +330,7 @@ export async function handleChatInput(
// Visibility bookkeeping is not the canonical save. Its failure must // Visibility bookkeeping is not the canonical save. Its failure must
// neither misreport a saved input as lost nor prevent the turn running. // neither misreport a saved input as lost nor prevent the turn running.
try { try {
for (const appended of stagedAutomatic ? [] : appends) { if (!stagedAutomatic) noteSteersDurable(chatId, merged);
if ((appended as { role?: string }).role === "custom") noteSteerVisible(appended as SteerMessage);
}
} catch (err) { } catch (err) {
console.error(`[chat ${chatId}] input visibility bookkeeping failed:`, err); console.error(`[chat ${chatId}] input visibility bookkeeping failed:`, err);
} }
@@ -383,7 +382,11 @@ export async function handleChatInput(
foregroundOwed: !callPurpose.background, foregroundOwed: !callPurpose.background,
purpose: callPurpose, purpose: callPurpose,
scopeCurrent: (scope) => connectionScopeCurrent(conversation.provider, scope), scopeCurrent: (scope) => connectionScopeCurrent(conversation.provider, scope),
commitStaged: () => { stores.chatStore.saveMessages(chatId, merged); persistBaseCount = merged.length; }, commitStaged: () => {
stores.chatStore.saveMessages(chatId, merged);
persistBaseCount = merged.length;
noteSteersDurable(chatId, merged);
},
restoreAutomatic: (automaticIds, retainedIds) => restoreUncommittedAutomatic(chatId, constructed, automaticIds, retainedIds), restoreAutomatic: (automaticIds, retainedIds) => restoreUncommittedAutomatic(chatId, constructed, automaticIds, retainedIds),
}); });
const agentOpts: AgentBuildOptions = { const agentOpts: AgentBuildOptions = {
@@ -481,7 +484,8 @@ export async function handleChatInput(
// run's first LLM call (the native replacement for continuation // run's first LLM call (the native replacement for continuation
// coalescing). Settlement still cannot beat the flow below: it needs // coalescing). Settlement still cannot beat the flow below: it needs
// a streamed response, which never happens inside the constructor. // a streamed response, which never happens inside the constructor.
registerLiveAgent(chatId, constructed); 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 // Debug recorder run marker (owner, 2026-10-04): correlates the
// JSONL event stream with the record's turns; no-op unless // JSONL event stream with the record's turns; no-op unless
// LH_PROVIDER_EVENTS=1. // LH_PROVIDER_EVENTS=1.
@@ -522,12 +526,13 @@ export async function handleChatInput(
// merely because usage/extraction/other settlement work failed. // merely because usage/extraction/other settlement work failed.
const persisted = stores.chatStore.getChat(chatId)?.messages ?? []; const persisted = stores.chatStore.getChat(chatId)?.messages ?? [];
reconcileUserSteers(chatId, persisted); reconcileUserSteers(chatId, persisted);
settleSteers(chatId, persisted); settleSteers(chatId, persisted, true, constructed);
}, },
onFinish: (turnMessages, isAborted) => { onFinish: (turnMessages, isAborted) => {
// Wave B: the record write is the agent's own final state — // Wave B: the record write is the agent's own final state —
// the new messages are the slice past the merged base. // the new messages are the slice past the merged base.
const stamped = persistTurn(stores.chatStore, chatId, persistBaseCount, turnMessages, isAborted); const stamped = persistTurn(stores.chatStore, chatId, persistBaseCount, turnMessages, isAborted);
noteSteersDurable(chatId, stores.chatStore.getChat(chatId)?.messages ?? stamped);
// usage log append (idempotent per turn id) — the stats bar // usage log append (idempotent per turn id) — the stats bar
// refolds this on the client after the turn settles. The // refolds this on the client after the turn settles. The
// provenance message is the turn's FIRST new assistant message. // provenance message is the turn's FIRST new assistant message.
@@ -573,7 +578,7 @@ export async function handleChatInput(
// resolve; stranded markers return to pending), then let // resolve; stranded markers return to pending), then let
// the pump take any stragglers down the idle wake. // the pump take any stragglers down the idle wake.
unregisterLiveAgent(chatId); unregisterLiveAgent(chatId);
settleSteers(chatId, stamped); settleSteers(chatId, stamped, false, constructed);
// Composer steers (2026-10-03): delivered ids dropped, stranded // Composer steers (2026-10-03): delivered ids dropped, stranded
// ones stay parked for the chat's next run (the registry is the // ones stay parked for the chat's next run (the registry is the
// stranding safety net — the agent's undrained copy died above). // stranding safety net — the agent's undrained copy died above).
src/server/chats.ts
+26 -6
View File
@@ -1,4 +1,5 @@
import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts"; import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts";
import { isJobLive } from "../agent/subagents/aborts.ts";
import { coverageOf } from "../core/coverage.ts"; import { coverageOf } from "../core/coverage.ts";
import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts"; import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts";
import { titleFromPrompt } from "../core/contracts.ts"; import { titleFromPrompt } from "../core/contracts.ts";
@@ -6,6 +7,7 @@ import { graphRevision } from "./graph.ts";
import { stores } from "./stores.ts"; import { stores } from "./stores.ts";
import { isTurnRunning } from "./turn-liveness.ts"; import { isTurnRunning } from "./turn-liveness.ts";
import { clearTurnFailure } from "./turn-runs.ts"; import { clearTurnFailure } from "./turn-runs.ts";
import { discardChatDeliveries } from "./jobs.ts";
/** /**
* Server-side chat entries for the transport layer (app/api/chats*, * Server-side chat entries for the transport layer (app/api/chats*,
@@ -66,10 +68,7 @@ export type ChatDeleteOutcome =
* the wind-down finds the chat gone at its merge and aborts with * the wind-down finds the chat gone at its merge and aborts with
* nothing merged. * nothing merged.
*/ */
export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): Promise<ChatDeleteOutcome> { function runningWorkBlocker(chatId: string): Extract<ChatDeleteOutcome, { status: 409 }> | null {
if (!stores.chatStore.getChat(chatId)) {
return { ok: false, status: 404, code: "not_found", error: "No such session" };
}
if (isTurnRunning(chatId)) { if (isTurnRunning(chatId)) {
return { return {
ok: false, ok: false,
@@ -79,7 +78,11 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
jobIds: [], 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) { if (runningJobs.length > 0) {
return { return {
ok: false, ok: false,
@@ -89,6 +92,15 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
jobIds: runningJobs.map((job) => job.id), jobIds: runningJobs.map((job) => job.id),
}; };
} }
return null;
}
export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): Promise<ChatDeleteOutcome> {
if (!stores.chatStore.getChat(chatId)) {
return { ok: false, status: 404, code: "not_found", error: "No such session" };
}
const blocker = runningWorkBlocker(chatId);
if (blocker) return blocker;
if (extractionStatuses().some((run) => run.chatId === chatId)) { if (extractionStatuses().some((run) => run.chatId === chatId)) {
abortExtraction(chatId); // true cancellation — spend stops, nothing merges abortExtraction(chatId); // true cancellation — spend stops, nothing merges
const deadline = Date.now() + WIND_DOWN_TIMEOUT_MS; const deadline = Date.now() + WIND_DOWN_TIMEOUT_MS;
@@ -105,11 +117,19 @@ export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): P
} }
} }
// A result wake or a worker may have started during extraction wind-down.
// Recheck immediately before the synchronous purge, with no await gap.
const lateBlocker = runningWorkBlocker(chatId);
if (lateBlocker) return lateBlocker;
// THE SYNC BLOCK — no awaits below (see the doc comment). Order: // THE SYNC BLOCK — no awaits below (see the doc comment). Order:
// ledgers → jobs → the chat record. // ledgers → jobs → the chat record.
const stats = deleteChatTraces(stores.graphStore, chatId, mode); const stats = deleteChatTraces(stores.graphStore, chatId, mode);
const jobsRemoved = stores.jobStore.deleteJobsForChat(chatId); const jobsRemoved = stores.jobStore.deleteJobsForChat(chatId);
if (stores.chatStore.deleteChat(chatId)) clearTurnFailure(chatId); if (stores.chatStore.deleteChat(chatId)) {
discardChatDeliveries(chatId);
clearTurnFailure(chatId);
}
return { ok: true, mode, stats, jobsRemoved, graphRevision: graphRevision() }; return { ok: true, mode, stats, jobsRemoved, graphRevision: graphRevision() };
} }
src/server/jobs.ts
+185 -163
View File
@@ -1,7 +1,7 @@
/** /**
* THE WAKE PIPELINE (contracts.ts "THE ASYNC PROTOCOL" §4) — server-side * THE WAKE PIPELINE (contracts.ts "THE ASYNC PROTOCOL" §4) — server-side
* steer delivery for finished sub-agent jobs. Wave C (2026-09-26): * steer delivery for finished sub-agent jobs. Wave C (2026-09-26):
* delivery rides pi's OWN follow-up queue when a run is live, and the * delivery rides pi's OWN steering queue when a run is live, and the
* idle wake otherwise. One registry, two vehicles. * idle wake otherwise. One registry, two vehicles.
* *
* deliverJobResult(steer) — the runner calls this on job completion. * deliverJobResult(steer) — the runner calls this on job completion.
@@ -40,15 +40,17 @@
* *
* settleSteers(chatId, persisted) — the handler calls this at turn * settleSteers(chatId, persisted) — the handler calls this at turn
* settlement (after unregisterLiveAgent): entries the run DELIVERED * settlement (after unregisterLiveAgent): entries the run DELIVERED
* resolve + flip their jobs; entries that were handed to the dying * resolve + acknowledge delivery; entries handed to the dying
* run but never injected (stopper exit, abort, a missed final poll) * run but never injected (stopper exit, abort, a missed final poll)
* return to pending and the pump restarts — no stranding, no loss. * return to pending and the pump restarts — no stranding, no loss.
* *
* Crash recovery (recoverStaleJobs): a process restart can orphan jobs — * Worker status ends at computation completion; JobRecord.delivery owns
* "running" with a result (the wake never fired) or without one (the * the pending result or abort notice independently. Registry entries
* sub-agent loop itself died). The sweep re-delivers finished results * survive in-flight wakes, so duplicate submissions share one promise.
* (idempotently) and re-spawns unfinished loops, both via injected * Failed starts stay pending without timers or worker errors. Recovery
* seams. It runs lazily off the /api/jobs poll, throttled. * restores known pending deliveries and re-spawns unfinished workers;
* registration also restores pending delivery before the first poll.
* Legacy terminal jobs without delivery metadata are never replayed.
* *
* All mutable state lives on globalThis (the turn-liveness lesson: Next dev * All mutable state lives on globalThis (the turn-liveness lesson: Next dev
* gives each route entry its own module instances — a module-level Map would * gives each route entry its own module instances — a module-level Map would
@@ -83,6 +85,14 @@ export type JobRespawn = (jobId: string) => void;
interface QueuedSteer { interface QueuedSteer {
steer: SteerMessage; steer: SteerMessage;
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; resolve: () => void;
reject: (reason: Error) => void; reject: (reason: Error) => void;
/** The live agent this marker was steered into (agent.steer) — set by /** The live agent this marker was steered into (agent.steer) — set by
@@ -166,18 +176,37 @@ function jobIdOf(steer: SteerMessage): string | null {
return typeof details?.jobId === "string" ? details.jobId : null; return typeof details?.jobId === "string" ? details.jobId : null;
} }
/** Dock bookkeeping: the marker reached the learner's view — flip its job /** Call only after canonical conversation persistence. Streaming visibility
* NOW instead of waiting for the delivering turn to settle. Called from * is not durable acceptance. The saved marker wins over delivery-bookkeeping
* BOTH visibility seams: the marker's message_end wire event (live steer * failures; acknowledging delivery never changes the worker's outcome. */
* injection — the card is on screen while the teacher keeps teaching) and export function noteSteersDurable(chatId: string, persisted: readonly LhMessage[]): void {
* the write-ahead persist in handleChatInput (the wake's input appends — for (const message of persisted) {
* the card is durable in the record). The registry's delivery promise if (message.role === "custom") finalizeSteer(message);
* still settles via settleSteers / the runner-200 path; finalizeJob is }
* idempotent, so the later flip is a no-op. Owner ruling (2026-09-26): a for (const entry of [...(state().queues.get(chatId) ?? [])]) {
* delivered subagent leaves the background dock the moment its result is if (persisted.some((message) => message.id === entry.steer.id)) completeEntry(chatId, entry);
* visible, not when the teacher's turn ends. */ }
export function noteSteerVisible(steer: SteerMessage): void { }
finalizeSteer(steer);
function removeEntry(chatId: string, entry: QueuedSteer): void {
const remaining = (state().queues.get(chatId) ?? []).filter((candidate) => candidate !== entry);
if (remaining.length > 0) state().queues.set(chatId, remaining);
else state().queues.delete(chatId);
}
function completeEntry(chatId: string, entry: QueuedSteer): void {
finalizeSteer(entry.steer);
removeEntry(chatId, entry);
entry.resolve();
}
/** Deletion discards the session's obligations, never wakes a deleted chat. */
export function discardChatDeliveries(chatId: string): void {
const s = state();
const queue = s.queues.get(chatId) ?? [];
s.queues.delete(chatId);
s.userSteers.delete(chatId);
for (const entry of queue) entry.resolve();
} }
/** Finalize a delivered steer's job bookkeeping — job steers only. */ /** Finalize a delivered steer's job bookkeeping — job steers only. */
@@ -186,24 +215,30 @@ function finalizeSteer(steer: SteerMessage): void {
if (jobId) finalizeJob(jobId); if (jobId) finalizeJob(jobId);
} }
/** Fail a steer's job on unrecoverable delivery failure — job steers only. */ /** Queue one logical delivery per ID, including in-flight wake cargo.
function failSteer(steer: SteerMessage, err: Error): void { * Job promises resolve on durable marker acceptance (or session deletion);
const jobId = jobIdOf(steer); * failed starts stay pending. Non-job probe handoffs keep their existing
if (jobId) failJob(jobId, err); * wake-acceptance/error semantics. Workers never await this promise. */
export function deliverJobResult(steer: SteerMessage): Promise<void> {
const chatId = chatIdOf(steer);
const entry = queueSteer(chatId, steer);
void pump(chatId);
return entry.promise;
} }
/** Queue ANY steer (result, abort, probe handoff) through the pipeline. function queueSteer(chatId: string, steer: SteerMessage): QueuedSteer {
* Resolves when the wake turn has completed; rejects when delivery fails const queue = state().queues.get(chatId) ?? [];
* unrecoverably. Never rejects for queueing itself. */ const existing = queue.find((entry) => entry.steer.id === steer.id);
export function deliverJobResult(steer: SteerMessage): Promise<void> { if (existing) return existing;
return new Promise<void>((resolve, reject) => { let resolve!: () => void;
const s = state(); let reject!: (error: Error) => void;
const chatId = chatIdOf(steer); const promise = new Promise<void>((res, rej) => { resolve = res; reject = rej; });
const queue = s.queues.get(chatId) ?? []; // Recovery/registration may own an entry with no waiting caller.
queue.push({ steer, resolve, reject }); void promise.catch(() => undefined);
s.queues.set(chatId, queue); const entry: QueuedSteer = { steer, promise, resolve, reject };
void pump(chatId); queue.push(entry);
}); state().queues.set(chatId, queue);
return entry;
} }
/** Hand every eligible pending marker to a live agent's steering queue /** Hand every eligible pending marker to a live agent's steering queue
@@ -227,7 +262,7 @@ function handOffEligible(chatId: string, agent: Agent): void {
// remain deliverable, including the first poll of an answer run. // remain deliverable, including the first poll of an answer run.
if (automaticFollowUpAllowed(chatId, agent) && queue && queue.length > 0) { if (automaticFollowUpAllowed(chatId, agent) && queue && queue.length > 0) {
for (const entry of queue) { for (const entry of queue) {
if (entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") { if (!entry.waking && !entry.deferred && entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") {
agent.steer(entry.steer); agent.steer(entry.steer);
entry.handedTo = agent; entry.handedTo = agent;
} }
@@ -254,8 +289,13 @@ function handOffEligible(chatId: string, agent: Agent): void {
* markers are steered into the run right here. The settle seam cannot * markers are steered into the run right here. The settle seam cannot
* beat the registration: settlement needs a streamed response, which * beat the registration: settlement needs a streamed response, which
* never happens inside the constructor call. */ * never happens inside the constructor call. */
export function registerLiveAgent(chatId: string, agent: Agent): void { export function registerLiveAgent(chatId: string, agent: Agent, allowDeferred = true, wakeSteerIds?: ReadonlySet<string>): void {
state().liveAgents.set(chatId, agent); state().liveAgents.set(chatId, agent);
restorePendingDeliveries(chatId);
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); handOffEligible(chatId, agent);
} }
@@ -326,7 +366,7 @@ export function restoreUncommittedAutomatic(chatId: string, agent: Agent, automa
* unregisterLiveAgent, with the run's persisted messages. Two jobs: * unregisterLiveAgent, with the run's persisted messages. Two jobs:
* *
* - entries whose marker the run DELIVERED (in-run follow-up injection — * - entries whose marker the run DELIVERED (in-run follow-up injection —
* their ids are in the persisted slice) resolve and flip their jobs; * their ids are in the persisted slice) resolve and acknowledge delivery;
* - entries handed to the dying run but never injected (the question * - entries handed to the dying run but never injected (the question
* stopper exited before the follow-up poll; an abort; a delivery that * stopper exited before the follow-up poll; an abort; a delivery that
* raced the final poll) return to PENDING (handedTo cleared) and the * raced the final poll) return to PENDING (handedTo cleared) and the
@@ -335,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 * 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 * 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). */ * the dead agent's copy is simply abandoned (GC'd with the agent). */
export function settleSteers(chatId: string, persisted: { id?: string }[]): void { export function settleSteers(chatId: string, persisted: { id?: string }[], failed = false, settledAgent?: Agent): void {
const queue = state().queues.get(chatId); const queue = state().queues.get(chatId);
if (!queue || queue.length === 0) return; if (!queue || queue.length === 0) return;
const deliveredIds = new Set(persisted.map((m) => m.id).filter(Boolean)); const deliveredIds = new Set(persisted.map((m) => m.id).filter(Boolean));
@@ -345,7 +385,11 @@ export function settleSteers(chatId: string, persisted: { id?: string }[]): void
finalizeSteer(entry.steer); finalizeSteer(entry.steer);
entry.resolve(); entry.resolve();
} else { } else {
entry.handedTo = undefined; 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); remaining.push(entry);
} }
} }
@@ -373,16 +417,10 @@ export interface AbortOutcome {
* runner registered it at loop start) — the in-flight model call * runner registered it at loop start) — the in-flight model call
* aborts, not just the next loop check. The runner unwinds silently * aborts, not just the next loop check. The runner unwinds silently
* (its catch checks abortSignal.aborted; finishJob checks status). * (its catch checks abortSignal.aborted; finishJob checks status).
* 2. DROP any queued result steer for this job (the delivering-result * 2. REFUSE if computation already finished. Queued results belong to
* phase): resolve its entry without firing the wake. If the wake * the session, not to an abortable background worker.
* turn is ALREADY in flight (the pump shifted the entry), it's too * 3. SAVE error "Aborted by learner" and a pending abort delivery together.
* late — the result steer lands, and the cancellation steer dedupes * Delivery acknowledgement never overwrites the worker's outcome.
* against it via steerDelivered (same jobId metadata).
* 3. FLIP the record to error "Aborted by learner" immediately — the
* poll reflects the user's action at once (the user chose
* error-with-reason over a distinct cancelled status). finalizeJob
* never overwrites a non-running status, so the flip is terminal no
* matter what the wake does.
* 4. STEER the teacher: a one-line abort note (buildAbortSteer) through * 4. STEER the teacher: a one-line abort note (buildAbortSteer) through
* the same wake pipeline, so the teacher's promise of a result * the same wake pipeline, so the teacher's promise of a result
* doesn't hang silently in the lesson. * doesn't hang silently in the lesson.
@@ -394,28 +432,18 @@ export function abortJob(jobId: string): AbortOutcome {
const job = stores.jobStore.getJob(jobId); const job = stores.jobStore.getJob(jobId);
if (!job) return { ok: false, reason: "not_found" }; if (!job) return { ok: false, reason: "not_found" };
if (job.status !== "running") return { ok: false, reason: "not_running" }; if (job.status !== "running") return { ok: false, reason: "not_running" };
if (job.result) {
fireJobAbort(jobId); // no-op when the loop already returned (delivering) normalizeLegacyResult(job);
return { ok: false, reason: "not_running" };
// Drop any queued steers for this job (their deliver() resolves unfired).
// HANDED entries (already in a live run's follow-up queue) are past
// recall — same ruling as a fired wake: the result lands, and the abort
// note below dedupes against it via steerDelivered (same jobId).
const queue = state().queues.get(job.chatId);
if (queue) {
const remaining = queue.filter(
(e) => jobIdOf(e.steer) !== jobId || e.handedTo !== undefined,
);
if (remaining.length === 0) state().queues.delete(job.chatId);
else state().queues.set(job.chatId, remaining);
for (const entry of queue) {
if (jobIdOf(entry.steer) === jobId && entry.handedTo === undefined) entry.resolve();
}
} }
stores.jobStore.updateJob(jobId, { status: "error", error: "Aborted by learner", activity: "aborted" }); fireJobAbort(jobId);
stores.jobStore.updateJob(jobId, {
status: "error", error: "Aborted by learner", activity: "aborted",
delivery: { kind: "abort", status: "pending" },
});
void deliverJobResult(buildAbortSteer(job)).catch(() => { void deliverJobResult(buildAbortSteer(job)).catch(() => {
/* deliverJobResult marks the job error on unrecoverable failure */ /* The saved pending abort notice remains owned by the session. */
}); });
return { ok: true }; return { ok: true };
} }
@@ -470,18 +498,9 @@ async function pump(chatId: string): Promise<void> {
// Idempotent delivery: the transcript already carries these steers // Idempotent delivery: the transcript already carries these steers
// (an in-run injection persisted, a wake completed, or a duplicate // (an in-run injection persisted, a wake completed, or a duplicate
// deliver raced in). Resolve and flip without re-firing anything. // deliver raced in). Resolve and flip without re-firing anything.
const stillPending = queue.filter((e) => !steerDelivered(record.messages, e.steer.id)); noteSteersDurable(chatId, record.messages);
for (const entry of queue) { const stillPending = (s.queues.get(chatId) ?? []).filter((entry) => !entry.waking && !entry.deferred);
if (!stillPending.includes(entry)) { if (stillPending.length === 0) { parked = true; return; }
finalizeSteer(entry.steer);
entry.resolve();
}
}
if (stillPending.length === 0) {
s.queues.delete(chatId);
return;
}
s.queues.set(chatId, stillPending);
if (!automaticFollowUpAllowed(chatId, s.liveAgents.get(chatId))) { parked = true; return; } if (!automaticFollowUpAllowed(chatId, s.liveAgents.get(chatId))) { parked = true; return; }
// VEHICLE 1 — THE LIVE RUN: pi's steering queue. The agent of the // VEHICLE 1 — THE LIVE RUN: pi's steering queue. The agent of the
@@ -526,28 +545,17 @@ async function pump(chatId: string): Promise<void> {
// The runner is required — chat-handler sets it at init. If some // The runner is required — chat-handler sets it at init. If some
// module instance missed the wiring, fail loudly rather than hang. // module instance missed the wiring, fail loudly rather than hang.
if (!s.turnRunner) { if (!s.turnRunner) {
s.queues.delete(chatId);
const err = new Error("wake pipeline: no turn runner wired (setTurnRunner)"); const err = new Error("wake pipeline: no turn runner wired (setTurnRunner)");
for (const entry of stillPending) { for (const entry of stillPending) deferOrReject(chatId, entry, err);
failSteer(entry.steer, err); parked = true;
entry.reject(err);
}
return; return;
} }
// A live run may have delivered some markers while the gates were // A live run may have delivered some markers while the gates were
// closed — re-check: only still-missing markers ride the wake. // closed — re-check: only still-missing markers ride the wake.
const fire = stillPending.filter((e) => !steerDelivered(fresh.messages, e.steer.id)); noteSteersDurable(chatId, fresh.messages);
for (const entry of stillPending) { const fire = (s.queues.get(chatId) ?? []).filter((entry) => !entry.waking && !entry.deferred && !entry.handedTo);
if (!fire.includes(entry)) { if (fire.length === 0) return;
finalizeSteer(entry.steer);
entry.resolve();
}
}
if (fire.length === 0) {
s.queues.delete(chatId);
return;
}
// THE DISMISSAL PAUSE (owner design 2026-10-03, "A2"): the record // THE DISMISSAL PAUSE (owner design 2026-10-03, "A2"): the record
// ends in a user dismissal — the session waits for the USER. Hold // ends in a user dismissal — the session waits for the USER. Hold
@@ -561,50 +569,37 @@ async function pump(chatId: string): Promise<void> {
return; return;
} }
// DEQUEUE-AT-FIRE (the Wave C lesson from the Wave B regression): // Keep ownership through the wake. The flag excludes cargo from
// the carried markers leave the registry BEFORE the turn runs — // its own live-agent handoff while duplicate callers share it.
// the wake run's OWN follow-up poll must never see its own cargo, for (const entry of fire) entry.waking = true;
// and a 409 (client raced in) must be able to push them back.
s.queues.delete(chatId);
const steers = fire.map((e) => e.steer); const steers = fire.map((e) => e.steer);
// Fire the wake (every marker as the turn's input appends). A 409 = // Fire the wake with cargo still registered. A 409 releases its
// a client turn raced in between idle and firing — push the cargo // vehicle flag and retries after the existing backoff, without
// back and retry after backoff (the racing turn's own settlement // appending another registry entry.
// re-pumps if it beats us to the re-queue).
const status = await s.turnRunner(chatId, { type: "steers", steers }).catch((err: unknown) => { const status = await s.turnRunner(chatId, { type: "steers", steers }).catch((err: unknown) => {
console.error(`[steer ${steers.map((x) => x.id).join(",")}] wake turn crashed:`, err); console.error(`[steer ${steers.map((x) => x.id).join(",")}] wake turn crashed:`, err);
return -1; return -1;
}); });
if (status === 403) { const saved = stores.chatStore.getChat(chatId);
// Consent may have changed after the last gate. Preserve completed if (saved) noteSteersDurable(chatId, saved.messages);
// job/result ownership and wait for an explicit connection change. const owned = fire.filter((entry) => s.queues.get(chatId)?.includes(entry));
const back = s.queues.get(chatId) ?? []; if (status === 403 || status === 409) {
back.push(...fire); for (const entry of owned) { entry.waking = false; entry.wakeAgent = undefined; }
s.queues.set(chatId, back); if (status === 403) { parked = true; return; }
parked = true;
return;
}
if (status === 409) {
const back = s.queues.get(chatId) ?? [];
back.push(...fire);
s.queues.set(chatId, back);
await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!); await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!);
continue; continue;
} }
attempt = 0; attempt = 0;
for (const entry of owned) {
if (status === 200) { if (status === 200 && !jobIdOf(entry.steer)) {
for (const entry of fire) { // Preserve the non-job probe handoff's acceptance semantics.
finalizeSteer(entry.steer); completeEntry(chatId, entry);
entry.resolve(); } else if (status !== 200 || !isTurnRunning(chatId)) {
} deferOrReject(chatId, entry, new Error(`wake turn did not persist steer (status ${status})`));
} else {
const err = new Error(`wake turn failed with status ${status}`);
for (const entry of fire) {
failSteer(entry.steer, err);
entry.reject(err);
} }
// An accepted live turn still owns its cargo until canonical save
// or settlement. HTTP 200 alone is not a durable marker receipt.
} }
} }
} finally { } finally {
@@ -614,7 +609,7 @@ async function pump(chatId: string): Promise<void> {
// a live run are NOT stragglers (settleSteers owns their resurrection) // a live run are NOT stragglers (settleSteers owns their resurrection)
// — restarting for them would just park in waitForIdle forever. // — restarting for them would just park in waitForIdle forever.
const q = state().queues.get(chatId) ?? []; const q = state().queues.get(chatId) ?? [];
if (!parked && automaticFollowUpAllowed(chatId, state().liveAgents.get(chatId)) && q.some((e) => e.handedTo === undefined)) void pump(chatId); if (!parked && automaticFollowUpAllowed(chatId, state().liveAgents.get(chatId)) && q.some((e) => e.handedTo === undefined && !e.waking && !e.deferred)) void pump(chatId);
} }
} }
@@ -638,25 +633,56 @@ state().stopConnectionWatch = onSubscriptionChange(() => {
} }
}); });
/** Flip the job to its final status now that the lesson has resumed. */ function deferOrReject(chatId: string, entry: QueuedSteer, error: Error): void {
function finalizeJob(jobId: string): void { entry.waking = false;
const job = stores.jobStore.getJob(jobId); entry.wakeAgent = undefined;
if (!job || job.status !== "running") return; if (jobIdOf(entry.steer)) {
const result = job.result; entry.deferred = true;
const failed = result ? isSubagentFailure(job.kind, result as never) : true; console.error(`[steer ${entry.steer.id}] delivery remains pending:`, error.message);
stores.jobStore.updateJob(jobId, { } else {
status: failed ? "error" : "done", removeEntry(chatId, entry);
activity: failed ? "failed" : "delivered", entry.reject(error);
}); }
} }
function failJob(jobId: string, err: Error): void { /** Acknowledge session delivery without changing the worker outcome. */
// An aborted job stays aborted even when its (abort) steer function finalizeJob(jobId: string): void {
// delivery fails — the user's action is the terminal fact, not the try {
// delivery error. Non-running records are never overwritten here. let job = stores.jobStore.getJob(jobId);
const job = stores.jobStore.getJob(jobId); if (!job) return;
if (!job || job.status !== "running") return; if (!job.delivery) job = normalizeLegacyResult(job);
stores.jobStore.updateJob(jobId, { status: "error", error: err.message, activity: "delivery failed" }); if (job.delivery?.status !== "pending") return;
stores.jobStore.updateJob(jobId, { delivery: { ...job.delivery, status: "delivered" } });
} catch (error) {
// The conversation is already durable. A bookkeeping failure must
// never replay that marker or turn a successful worker into a failure.
console.error(`[job ${jobId}] delivery acknowledgement failed:`, error);
}
}
function normalizeLegacyResult(job: JobRecord): JobRecord {
if (job.delivery || job.status !== "running" || !job.result) return job;
const failed = isSubagentFailure(job.kind, job.result as never);
return stores.jobStore.updateJob(job.id, {
status: failed ? "error" : "done", activity: failed ? "failed" : "finished",
delivery: { kind: "result", status: "pending" },
}) ?? job;
}
/** Restore known obligations only; old terminal records are never replayed. */
function restorePendingDeliveries(chatId: string): void {
const record = stores.chatStore.getChat(chatId);
if (!record) return;
noteSteersDurable(chatId, record.messages);
for (const stored of stores.jobStore.listJobs(chatId)) {
const job = isJobLive(stored.id) ? stored : normalizeLegacyResult(stored);
if (job.delivery?.status !== "pending") continue;
const steer = job.delivery.kind === "abort" ? buildAbortSteer(job)
: job.result ? buildSubagentSteer(job, job.result as never) : undefined;
if (!steer) continue;
if (steerDelivered(record.messages, steer.id)) finalizeSteer(steer);
else queueSteer(chatId, steer);
}
} }
/** /**
@@ -669,7 +695,7 @@ function failJob(jobId: string, err: Error): void {
async function waitForIdle(chatId: string): Promise<void> { async function waitForIdle(chatId: string): Promise<void> {
for (;;) { for (;;) {
const record = stores.chatStore.getChat(chatId); const record = stores.chatStore.getChat(chatId);
if (!record) return; if (!record || !(state().queues.get(chatId) ?? []).some((entry) => !entry.waking && !entry.deferred && !entry.handedTo)) return;
const blocked = isTurnRunning(chatId) || danglingQuestionOf(record.messages) !== null; const blocked = isTurnRunning(chatId) || danglingQuestionOf(record.messages) !== null;
if (!blocked) { if (!blocked) {
await delay(GRACE_MS); await delay(GRACE_MS);
@@ -699,9 +725,9 @@ const SWEEP_THROTTLE_MS = 30_000;
* hung-but-alive loops beside themselves — the figure-maker incident, * hung-but-alive loops beside themselves — the figure-maker incident,
* 2026-09-18). A record WITH a live registration is provably executing * 2026-09-18). A record WITH a live registration is provably executing
* in THIS process — never touched, however old: a hung first model call * in THIS process — never touched, however old: a hung first model call
* (zero usage ever billed) stays honestly "running" in the dock until * stays honestly running until END kills it. Terminal workers with known
* END kills it. The abort registration spans loop start to delivery * pending delivery are restored independently; old terminal records without
* completion, so a live registration can never belong to a dead run. * delivery metadata are never inferred to owe a delivery.
*/ */
/** Re-run a job's sub-agent loop — exposed for the sweep's tests (the /** Re-run a job's sub-agent loop — exposed for the sweep's tests (the
* /api/jobs poll is its only production caller, disabled under NODE_ENV=test). */ * /api/jobs poll is its only production caller, disabled under NODE_ENV=test). */
@@ -713,18 +739,14 @@ export function recoverStaleJobs(): void {
try { try {
for (const chat of stores.chatStore.listChats()) { for (const chat of stores.chatStore.listChats()) {
for (const job of stores.jobStore.listJobs(chat.id)) { for (const job of stores.jobStore.listJobs(chat.id)) {
if (job.status !== "running") continue; if (isJobLive(job.id)) continue; // never re-run beside live computation
if (isJobLive(job.id)) continue; // provably executing here — never re-run beside it
if (!stores.chatStore.getChat(job.chatId)) continue; if (!stores.chatStore.getChat(job.chatId)) continue;
if (isTurnRunning(job.chatId)) continue; // a wake turn may be delivering it right now if (isTurnRunning(job.chatId)) continue;
if (job.result) { const recovered = normalizeLegacyResult(job);
// The wake never fired (or crashed mid-flight): re-deliver. if (recovered.delivery?.status === "pending") {
const steer = buildSubagentSteer(job, job.result as never); restorePendingDeliveries(job.chatId);
stores.jobStore.updateJob(job.id, { activity: "recovering: re-delivering" }); void pump(job.chatId);
void deliverJobResult(steer).catch(() => { } else if (job.status === "running" && !job.result) {
/* deliverJobResult marks the job error on failure */
});
} else {
// The sub-agent loop itself died with the process: re-spawn. // The sub-agent loop itself died with the process: re-spawn.
if (!s.respawn) continue; if (!s.respawn) continue;
stores.jobStore.updateJob(job.id, { activity: "recovering: re-running" }); stores.jobStore.updateJob(job.id, { activity: "recovering: re-running" });
src/server/pi/ui-stream.ts
-7
View File
@@ -1,6 +1,4 @@
import type { Agent, AgentEvent, AgentMessage } from "@earendil-works/pi-agent-core"; import type { Agent, AgentEvent, AgentMessage } from "@earendil-works/pi-agent-core";
import type { SteerMessage } from "../../core/contracts.ts";
import { noteSteerVisible } from "../jobs.ts";
import { ToolCallIdUniquifier } from "./call-ids.ts"; import { ToolCallIdUniquifier } from "./call-ids.ts";
import type { ChatGPTRequestCycle } from "../../agent/chatgpt-request-cycle.ts"; import type { ChatGPTRequestCycle } from "../../agent/chatgpt-request-cycle.ts";
@@ -255,11 +253,6 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
if (event.type === "message_end") { if (event.type === "message_end") {
const m = event.message as { role?: string; stopReason?: string } | undefined; const m = event.message as { role?: string; stopReason?: string } | undefined;
if (m?.role === "assistant" && m.stopReason === "aborted") sawAbort = true; if (m?.role === "assistant" && m.stopReason === "aborted") sawAbort = true;
// Dock fix (owner ruling 2026-09-26): an injected steer marker's
// message_end = the card is visible to the learner — flip the
// job's bookkeeping now (the turn may keep teaching for minutes;
// settleSteers re-runs the flip as a no-op idempotently).
if (m?.role === "custom") noteSteerVisible(event.message as SteerMessage);
} }
if (event.type === "agent_end") { if (event.type === "agent_end") {
// Persist/report before the terminal closes every view to new frames. // Persist/report before the terminal closes every view to new frames.
src/store/jobs.ts
+1
View File
@@ -35,6 +35,7 @@ function createJob(init: CreateJobInit): JobRecord {
kind: init.kind, kind: init.kind,
chatId: init.chatId, chatId: init.chatId,
toolCallId: init.toolCallId, toolCallId: init.toolCallId,
...(init.input !== undefined ? { input: init.input } : {}),
status: "running", status: "running",
activity: init.activity ?? "starting", activity: init.activity ?? "starting",
createdAt: now, createdAt: now,
tests/chat-flow.test.ts
+2 -2
View File
@@ -522,7 +522,7 @@ describe("teacher chat flow (scripted model)", () => {
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-mid" }); const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-mid" });
jobIdMid = job.id; jobIdMid = job.id;
// The runner stores the result BEFORE delivery (finalizeJob reads it). // The runner stores the result BEFORE delivery (finalizeJob reads it).
jobStore.updateJob(job.id, { result: { brief: "Mid-run brief.", sources: [] } }); jobStore.updateJob(job.id, { status: "done", result: { brief: "Mid-run brief.", sources: [] }, delivery: { kind: "result", status: "pending" } });
const steer = buildSubagentSteer(job, { brief: "Mid-run brief.", sources: [] }); const steer = buildSubagentSteer(job, { brief: "Mid-run brief.", sources: [] });
const delivery = deliverJobResult(steer); const delivery = deliverJobResult(steer);
await new Promise((r) => setTimeout(r, 300)); // vehicle 1 hands off await new Promise((r) => setTimeout(r, 300)); // vehicle 1 hands off
@@ -534,7 +534,7 @@ describe("teacher chat flow (scripted model)", () => {
release2(); release2();
await drain(await res); await drain(await res);
await delivery; // resolves at settleSteers (the run persisted the marker) await delivery; // resolves at settleSteers (the run persisted the marker)
assert.equal(jobFlippedMidTurn, "done", "the job flipped at the marker's message_end — WHILE the turn still streamed"); assert.equal(jobFlippedMidTurn, "done", "worker completion is independent of the streaming teacher turn");
// Record shape: the marker sits BETWEEN completed turn units — after // Record shape: the marker sits BETWEEN completed turn units — after
// the toolResult it must never split, adjacent to it. // the toolResult it must never split, adjacent to it.
tests/chatgpt-background.test.ts
+3 -2
View File
@@ -43,7 +43,7 @@ for (const revokedAfterQueue of [false, true]) {
stores.chatStore.saveSettings(chat.id, { model: { provider: "chatgpt", id: "gpt-5" } }); stores.chatStore.saveSettings(chat.id, { model: { provider: "chatgpt", id: "gpt-5" } });
const job = stores.jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "scratch-job-call" }); const job = stores.jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "scratch-job-call" });
const result = { brief: "Scratch completed job result", sources: [] }; const result = { brief: "Scratch completed job result", sources: [] };
stores.jobStore.updateJob(job.id, { status: "running", result }); stores.jobStore.updateJob(job.id, { status: "done", result, delivery: { kind: "result", status: "pending" } });
const marker = buildSubagentSteer(job, result); const marker = buildSubagentSteer(job, result);
const contexts: string[] = []; const contexts: string[] = [];
const native = piModels().getModel("openai", "gpt-5")!; const native = piModels().getModel("openai", "gpt-5")!;
@@ -83,7 +83,8 @@ for (const revokedAfterQueue of [false, true]) {
unregisterLiveAgent(chat.id); endTurn(chat.id); settleSteers(chat.id, persisted); unregisterLiveAgent(chat.id); endTurn(chat.id); settleSteers(chat.id, persisted);
await new Promise((resolve) => setImmediate(resolve)); await new Promise((resolve) => setImmediate(resolve));
assert.equal(wakes, 0, "no background spin or spend while consent is absent"); assert.equal(wakes, 0, "no background spin or spend while consent is absent");
assert.equal(stores.jobStore.getJob(job.id)!.status, "running", "delivery ownership must not be finalized or failed by deferral"); assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending", "delivery ownership must not be finalized or failed by deferral");
assert.equal(stores.jobStore.getJob(job.id)!.status, "done", "worker completion is independent of consent deferral");
assert.deepEqual(stores.jobStore.getJob(job.id)!.result, result); assert.deepEqual(stores.jobStore.getJob(job.id)!.result, result);
connection(true); connection(true);
await delivered; await delivered;
tests/demos.test.tsx
+5 -5
View File
@@ -274,14 +274,14 @@ describe("dispatch card rendering (the tool-call ack box)", () => {
assert.match(html, /Dispatched…/); assert.match(html, /Dispatched…/);
}); });
it("settles to a quiet delivered one-liner when done (artifact lives in the steer)", () => { it("settles to a quiet finished one-liner when the worker is done", () => {
const html = renderToStaticMarkup( const html = renderToStaticMarkup(
<JobsContext.Provider value={[mkJob("done", "delivered")]}> <JobsContext.Provider value={[mkJob("done", "delivered")]}>
<JobCard part={ackPart as never} /> <JobCard part={ackPart as never} />
</JobsContext.Provider>, </JobsContext.Provider>,
); );
assert.match(html, /Research agent/); assert.match(html, /Research agent/);
assert.match(html, /delivered/); assert.match(html, /finished/);
assert.ok(!html.includes("web_search"), "no stale activity line once settled"); assert.ok(!html.includes("web_search"), "no stale activity line once settled");
}); });
@@ -325,7 +325,7 @@ describe("dispatch card rendering (the tool-call ack box)", () => {
assert.match(html, /pdf_render/); assert.match(html, /pdf_render/);
}); });
it("pdf dispatch settles to the delivered one-liner", () => { it("pdf dispatch settles to the finished one-liner", () => {
const pdfJob: JobRecord = ({ const pdfJob: JobRecord = ({
id: "jp1", id: "jp1",
kind: "pdf", kind: "pdf",
@@ -342,7 +342,7 @@ describe("dispatch card rendering (the tool-call ack box)", () => {
</JobsContext.Provider>, </JobsContext.Provider>,
); );
assert.match(html, /PDF reader/); assert.match(html, /PDF reader/);
assert.match(html, /delivered/); assert.match(html, /finished/);
}); });
}); });
@@ -839,7 +839,7 @@ describe("demo/visual job runners", () => {
const done = jobStore.getJob(job.id)!; const done = jobStore.getJob(job.id)!;
assert.equal(done.status, "done"); assert.equal(done.status, "done");
assert.equal(done.activity, "delivered", "the fake deliver flips status like the wake pipeline"); assert.equal(done.activity, "finished", "the worker finishes independently of delivery");
const result = done.result as { title: string; html: string; tested: boolean; notes?: string }; const result = done.result as { title: string; html: string; tested: boolean; notes?: string };
assert.equal(result.title, "Compound interest"); assert.equal(result.title, "Compound interest");
assert.equal(result.html, GOOD_HTML); assert.equal(result.html, GOOD_HTML);
tests/job-delivery-recovery.test.ts
+331
View File
@@ -0,0 +1,331 @@
import assert from "node:assert/strict";
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 { 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.
const directory = fs.mkdtempSync(path.join(os.tmpdir(), "lh-job-delivery-"));
process.env.LH_DATA_DIR = path.join(directory, "data");
process.env.LH_AUTH_FILE = path.join(directory, "auth.json");
fs.writeFileSync(process.env.LH_AUTH_FILE, "{}");
const { stores } = await import("../src/server/stores.ts");
const { jobTools } = await import("../src/agent/tools/jobs.ts");
const { createJobRunner, researchModelOverride } = await import("../src/agent/subagents/runner.ts");
const { isJobLive } = await import("../src/agent/subagents/aborts.ts");
const { beginTurn, endTurn } = await import("../src/server/turn-liveness.ts");
const {
deliverJobResult, discardChatDeliveries, noteSteersDurable, recoverStaleJobs,
registerLiveAgent, unregisterLiveAgent, settleSteers, setTurnRunner,
} = await import("../src/server/jobs.ts");
const { activityFeed } = await import("../src/server/observability.ts");
const { deleteChatSession } = await import("../src/server/chats.ts");
const chats: string[] = [];
const result = { brief: "Fixture research result", sources: [] };
const question = {
id: "question-message", role: "assistant", content: [{ type: "toolCall", id: "question-call", name: "quiz", arguments: {} }],
api: "openai-completions", provider: "faux", model: "faux", timestamp: 1,
} as unknown as LhMessage;
const delay = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
async function until(predicate: () => boolean): Promise<void> {
const end = Date.now() + 8000;
while (!predicate()) { assert(Date.now() < end, "fixture condition timed out"); await delay(10); }
}
function sweep(): void {
const global = globalThis as { __lhWakePipeline?: { lastSweepAt: number } };
if (global.__lhWakePipeline) global.__lhWakePipeline.lastSweepAt = 0;
recoverStaleJobs();
}
function chat(blocked = false) {
const value = stores.chatStore.createChat("Isolated delivery fixture");
chats.push(value.id);
if (blocked) stores.chatStore.saveMessages(value.id, [question]);
return value;
}
function pending(chatId: string) {
const value = stores.jobStore.createJob({ kind: "research", chatId, toolCallId: "fixture-call" });
const job = stores.jobStore.updateJob(value.id, {
status: "done", result, activity: "finished", delivery: { kind: "result", status: "pending" },
})!;
return { job, steer: buildSubagentSteer(job, result) };
}
function agent() {
const handed: SteerMessage[] = [];
return { handed, value: { steer: (marker: SteerMessage) => handed.push(marker) } as unknown as Parameters<typeof registerLiveAgent>[1] };
}
function persist(chatId: string, steers: readonly LhMessage[]): void {
stores.chatStore.saveMessages(chatId, [...stores.chatStore.getChat(chatId)!.messages, ...steers]);
noteSteersDurable(chatId, steers);
}
afterEach(() => {
researchModelOverride.current = undefined;
for (const id of chats.splice(0)) {
unregisterLiveAgent(id); endTurn(id); discardChatDeliveries(id);
stores.jobStore.deleteJobsForChat(id); stores.chatStore.deleteChat(id);
}
setTurnRunner(async () => 500);
});
after(() => fs.rmSync(directory, { recursive: true, force: true }));
for (const [kind, input] of Object.entries({
research: { question: "Saved research question", context: "Saved context" },
demo: { title: "Fixture", brief: "Saved demo brief" },
visual: { brief: "Saved figure brief" },
pdf: { path: "fixture.pdf", question: "Saved PDF question" },
})) test(`${kind}: dispatch saves input before spawn and does not overwrite it on repeated dispatch`, async () => {
const owner = chat();
let spawns = 0;
const tools = jobTools({ chatId: owner.id, jobStore: stores.jobStore, spawn: (job) => {
spawns++;
assert.deepEqual(stores.jobStore.getJob(job.id)!.input, input);
} });
await tools[kind]!.execute!("fixture-dispatch", input);
await tools[kind]!.execute!("fixture-dispatch", input);
assert.equal(spawns, 1);
assert.equal(stores.jobStore.listJobs(owner.id)[0]!.kind, kind as JobKind);
});
test("saved question survives absent transcript; runner ends while delivery is still pending", async () => {
const owner = chat(true);
const input = { question: "The dispatch turn was never saved", context: "Fixture context" };
const job = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "lost-turn", input });
const faux = createFauxCore({});
let seen = "";
faux.setResponses([(context) => { seen = JSON.stringify(context.messages); return fauxAssistantMessage("Fixture brief"); }]);
researchModelOverride.current = { model: faux.getModel(), streamFn: faux.streamSimple };
let delivery: Promise<void> | undefined;
const runner = createJobRunner({ ...stores, deliver: (steer) => { delivery = deliverJobResult(steer); return delivery; } });
await runner.runJob(job.id);
assert(seen.includes(input.question)); assert(seen.includes(input.context));
assert.equal(isJobLive(job.id), false, "worker registration no longer spans delivery");
assert.equal(stores.jobStore.getJob(job.id)!.status, "done");
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
assert(!activityFeed().jobs.some((item) => item.id === job.id), "no running background-agent row");
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); settleSteers(owner.id, next.handed); endTurn(owner.id);
await delivery;
});
test("missing or blank research question records an honest failure without invoking a model", async () => {
for (const input of [undefined, { question: " " }]) {
const owner = chat();
const job = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "missing-question", input });
const faux = createFauxCore({}); let calls = 0;
faux.setResponses([() => { calls++; return fauxAssistantMessage("must not run"); }]);
researchModelOverride.current = { model: faux.getModel(), streamFn: faux.streamSimple };
const steers: SteerMessage[] = [];
await createJobRunner({ ...stores, deliver: async (steer) => { steers.push(steer); } }).runJob(job.id);
assert.equal(calls, 0);
assert.equal(stores.jobStore.getJob(job.id)!.error, "research question unavailable");
assert.equal(stores.jobStore.getJob(job.id)!.status, "error");
assert.match(steers[0]!.content, /^Research failed: the research question is unavailable\./);
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
}
});
test("repeated recovery behind a question yields one handed and one durable marker", async () => {
const owner = chat(true); const { job, steer } = pending(owner.id);
let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });
sweep(); sweep(); sweep();
const first = deliverJobResult(steer);
assert.strictEqual(deliverJobResult(steer), first, "duplicate callers share one logical delivery");
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
assert.deepEqual(next.handed.map((message) => message.id), [steer.id]);
persist(owner.id, next.handed); unregisterLiveAgent(owner.id); settleSteers(owner.id, next.handed); endTurn(owner.id);
await first;
assert.equal(wakes, 0);
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
});
for (const refused of [409, 403]) test(`duplicate during a ${refused} wake retains one logical owner`, async () => {
const owner = chat(); const { steer } = pending(owner.id);
let release!: () => void; const gate = new Promise<void>((resolve) => { release = resolve; });
let attempts = 0;
setTurnRunner(async (id, input) => {
assert.equal(input.type, "steers"); if (input.type !== "steers") return 500;
assert.equal(input.steers.length, 1);
attempts++;
if (attempts === 1) { await gate; return refused; }
persist(id, input.steers); return 200;
});
const first = deliverJobResult(steer);
await until(() => attempts === 1);
assert.strictEqual(deliverJobResult(steer), first, "wake cargo is still in the registry");
release();
if (refused === 403) {
await delay(30);
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); settleSteers(owner.id, next.handed); endTurn(owner.id);
}
await first;
assert.equal(stores.chatStore.getChat(owner.id)!.messages.filter((message) => message.id === steer.id).length, 1);
});
test("failed start stays quiet across sweeps, then joins a learner run", async () => {
const owner = chat(); const { job, steer } = pending(owner.id);
let attempts = 0; setTurnRunner(async () => { attempts++; return 503; });
const delivery = deliverJobResult(steer); await until(() => attempts === 1); await delay(30);
sweep(); sweep(); await delay(900);
assert.equal(attempts, 1);
assert.equal(stores.jobStore.getJob(job.id)!.status, "done");
assert.equal(stores.jobStore.getJob(job.id)!.error, undefined);
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
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); settleSteers(owner.id, next.handed); endTurn(owner.id);
await delivery;
});
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;
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, 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);
const update = stores.jobStore.updateJob;
stores.jobStore.updateJob = (id, patch) => {
if (id === job.id && patch.delivery?.status === "delivered") throw new Error("fixture acknowledgement write failure");
return update(id, patch);
};
try { persist(owner.id, [steer]); await delivery; }
finally { stores.jobStore.updateJob = update; }
let wakes = 0; setTurnRunner(async () => { wakes++; return 500; });
sweep();
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
assert.equal(wakes, 0);
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
assert.equal(next.handed.length, 0); unregisterLiveAgent(owner.id); endTurn(owner.id);
});
test("old running results become finished pending work; old terminal records are untouched", () => {
const owner = chat(true);
const old = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-pending" });
stores.jobStore.updateJob(old.id, { result });
const terminal = stores.jobStore.createJob({ kind: "research", chatId: owner.id, toolCallId: "legacy-terminal" });
stores.jobStore.updateJob(terminal.id, { status: "done", result });
sweep();
assert.equal(stores.jobStore.getJob(old.id)!.status, "done");
assert.equal(stores.jobStore.getJob(old.id)!.delivery?.status, "pending");
assert.equal(stores.jobStore.getJob(terminal.id)!.delivery, undefined);
const next = agent(); assert(beginTurn(owner.id)); registerLiveAgent(owner.id, next.value);
assert.deepEqual(next.handed.map((message) => message.id), [`steer_${old.id}`]);
unregisterLiveAgent(owner.id); endTurn(owner.id);
});
test("fresh process restores a pending delivery on registration without a jobs poll", () => {
const owner = chat(true); const { job, steer } = pending(owner.id);
const script = `
const { stores } = await import('./src/server/stores.ts');
const { registerLiveAgent, noteSteersDurable, unregisterLiveAgent } = await import('./src/server/jobs.ts');
const { beginTurn, endTurn } = await import('./src/server/turn-liveness.ts');
const handed = []; beginTurn(${JSON.stringify(owner.id)});
registerLiveAgent(${JSON.stringify(owner.id)}, { steer: marker => handed.push(marker) });
const record = stores.chatStore.getChat(${JSON.stringify(owner.id)});
stores.chatStore.saveMessages(record.id, [...record.messages, ...handed]);
noteSteersDurable(record.id, handed); unregisterLiveAgent(record.id); endTurn(record.id);
console.log(JSON.stringify(handed.map(marker => marker.id)));
`;
const child = spawnSync(process.execPath, ["--import", "tsx", "--input-type=module", "-e", script], {
cwd: process.cwd(), encoding: "utf8", timeout: 15000,
env: { PATH: process.env.PATH, HOME: path.join(directory, "home"), NODE_ENV: "test", LH_DATA_DIR: process.env.LH_DATA_DIR, LH_AUTH_FILE: process.env.LH_AUTH_FILE },
});
assert.equal(child.status, 0, child.stderr);
assert.deepEqual(JSON.parse(child.stdout.trim()), [steer.id]);
assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "delivered");
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; });
const delivery = deliverJobResult(steer);
const outcome = await deleteChatSession(owner.id, "preserve-graph");
assert(outcome.ok);
await delivery; await delay(450);
assert.equal(stores.jobStore.getJob(job.id), null);
assert.equal(stores.chatStore.getChat(owner.id), null);
assert.equal(wakes, 0);
});
tests/jobs.test.ts
+68 -50
View File
@@ -52,18 +52,20 @@ import { runningTurn } from "../src/server/turn-runs.ts";
import { import {
abortJob, abortJob,
deliverJobResult, deliverJobResult,
noteSteerVisible, noteSteersDurable,
discardChatDeliveries,
reconcileUserSteers, reconcileUserSteers,
recoverStaleJobs, recoverStaleJobs,
registerLiveAgent, registerLiveAgent,
setJobRespawn, setJobRespawn,
setTurnRunner, setTurnRunner as setRawTurnRunner,
type TurnRunner,
settleSteers, settleSteers,
steerLiveChat, steerLiveChat,
unregisterLiveAgent, unregisterLiveAgent,
} from "../src/server/jobs.ts"; } from "../src/server/jobs.ts";
import { registerJobAbort, unregisterJobAbort } from "../src/agent/subagents/aborts.ts"; import { isJobLive, registerJobAbort, unregisterJobAbort } from "../src/agent/subagents/aborts.ts";
import { beginTurn, endTurn } from "../src/server/turn-liveness.ts"; import { beginTurn, endTurn, isTurnRunning } from "../src/server/turn-liveness.ts";
import { persistTurn } from "../src/server/turn-service.ts"; import { persistTurn } from "../src/server/turn-service.ts";
import { chatStore } from "../src/store/chats.ts"; import { chatStore } from "../src/store/chats.ts";
import { jobStore } from "../src/store/jobs.ts"; import { jobStore } from "../src/store/jobs.ts";
@@ -141,11 +143,13 @@ afterEach(() => {
if (job && job.status === "running") { if (job && job.status === "running") {
jobStore.updateJob(id, { status: "done", activity: "settled (fence)" }); jobStore.updateJob(id, { status: "done", activity: "settled (fence)" });
} }
if (job?.delivery?.status === "pending") jobStore.updateJob(id, { delivery: { ...job.delivery, status: "delivered" } });
unregisterJobAbort(id); unregisterJobAbort(id);
} catch { } catch {
/* already gone */ /* already gone */
} }
} }
for (const chatId of createdChats) discardChatDeliveries(chatId);
// Late wakes from a prior test must be inert: a no-op respawn can't // Late wakes from a prior test must be inert: a no-op respawn can't
// increment anyone's counter, a no-op runner can't spawn real work. // increment anyone's counter, a no-op runner can't spawn real work.
setJobRespawn(() => {}); setJobRespawn(() => {});
@@ -455,16 +459,25 @@ function fakeDeliver() {
steers, steers,
deliver: async (steer: SteerMessage) => { deliver: async (steer: SteerMessage) => {
steers.push(steer); steers.push(steer);
const jobId = (steer.details as { jobId: string }).jobId; const job = jobStore.getJob((steer.details as { jobId: string }).jobId)!;
const job = jobStore.getJob(jobId); chatStore.saveMessages(job.chatId, [...chatStore.getChat(job.chatId)!.messages, steer]);
if (job && job.status === "running") { noteSteersDurable(job.chatId, [steer]);
const failed = job.result ? isSubagentFailure(job.kind, job.result as never) : true;
jobStore.updateJob(jobId, { status: failed ? "error" : "done", activity: failed ? "failed" : "delivered" });
}
}, },
}; };
} }
/** Fake wakes model canonical input acceptance, not just HTTP acknowledgement. */
function setTurnRunner(run: TurnRunner): void {
setRawTurnRunner(async (chatId, input) => {
const status = await run(chatId, input);
if (status === 200 && input.type === "steers") {
const record = chatStore.getChat(chatId);
if (record) chatStore.saveMessages(chatId, [...record.messages, ...input.steers.filter((s) => !steerDelivered(record.messages, s.id))]);
}
return status;
});
}
// ─── 3. dispatch: the tool call acks mid-turn and spawns ───────────────────── // ─── 3. dispatch: the tool call acks mid-turn and spawns ─────────────────────
const RESEARCH_INPUT = { const RESEARCH_INPUT = {
@@ -527,7 +540,7 @@ describe("dispatch (async protocol §2)", () => {
const done = jobStore.getJob(job!.id)!; const done = jobStore.getJob(job!.id)!;
assert.equal(done.status, "done"); assert.equal(done.status, "done");
assert.equal(done.activity, "delivered"); assert.equal(done.activity, "finished");
// The steer carries the brief on its model-facing content; the // The steer carries the brief on its model-facing content; the
// artifact payload rides details for the (UI) renderer. // artifact payload rides details for the (UI) renderer.
@@ -604,7 +617,7 @@ describe("research runner (async protocol §3)", () => {
// The job is done, with the activity line from the tool call. // The job is done, with the activity line from the tool call.
const done = jobStore.getJob(job.id)!; const done = jobStore.getJob(job.id)!;
assert.equal(done.status, "done"); assert.equal(done.status, "done");
assert.equal(done.activity, "delivered", "the fake deliver flips the status like the pipeline"); assert.equal(done.activity, "finished", "delivery does not own the worker status");
const result = done.result as ResearchResult; const result = done.result as ResearchResult;
assert.equal(result.brief, BRIEF); assert.equal(result.brief, BRIEF);
assert.ok(result.sources.length > 0, "sources parsed from the brief"); assert.ok(result.sources.length > 0, "sources parsed from the brief");
@@ -768,7 +781,7 @@ describe("pdf runner (sources v0)", () => {
const done = jobStore.getJob(job.id)!; const done = jobStore.getJob(job.id)!;
assert.equal(done.status, "done"); assert.equal(done.status, "done");
assert.equal(done.activity, "delivered"); assert.equal(done.activity, "finished");
const result = done.result as PdfResult; const result = done.result as PdfResult;
assert.equal(result.path, PDF_FIXTURE); assert.equal(result.path, PDF_FIXTURE);
assert.match(result.brief, /integral rule/); assert.match(result.brief, /integral rule/);
@@ -880,7 +893,7 @@ describe("wake pipeline (src/server/jobs.ts)", () => {
const job = jobStore.createJob({ kind, chatId: chat.id, toolCallId: `call-${jobStore.listJobs(chat.id).length}` }); const job = jobStore.createJob({ kind, chatId: chat.id, toolCallId: `call-${jobStore.listJobs(chat.id).length}` });
trackJob(job.id); trackJob(job.id);
const result: ResearchResult = { brief: "Done: it works.", sources: [{ title: "s", url: "https://x.test" }] }; const result: ResearchResult = { brief: "Done: it works.", sources: [{ title: "s", url: "https://x.test" }] };
jobStore.updateJob(job.id, { result, activity: "delivering result" }); jobStore.updateJob(job.id, { result, status: "done", activity: "finished", delivery: { kind: "result", status: "pending" } });
const steer = buildSubagentSteer(job, result); const steer = buildSubagentSteer(job, result);
return { chatId: chat.id, jobId: job.id, steer }; return { chatId: chat.id, jobId: job.id, steer };
} }
@@ -935,7 +948,7 @@ describe("wake pipeline (src/server/jobs.ts)", () => {
await waitFor(() => jobStore.getJob(jobId)!.status !== "running", 8000, "job flip after delivery"); await waitFor(() => jobStore.getJob(jobId)!.status !== "running", 8000, "job flip after delivery");
assert.equal(jobStore.getJob(jobId)!.status, "done"); assert.equal(jobStore.getJob(jobId)!.status, "done");
assert.equal(jobStore.getJob(jobId)!.activity, "delivered"); assert.equal(jobStore.getJob(jobId)!.delivery?.status, "delivered");
assert.ok(seen, "the turn runner fired"); assert.ok(seen, "the turn runner fired");
assert.equal(seen.chatId, chatId); assert.equal(seen.chatId, chatId);
assert.equal(seen.steers.length, 1); assert.equal(seen.steers.length, 1);
@@ -973,7 +986,8 @@ describe("wake pipeline (src/server/jobs.ts)", () => {
await delay(1200); await delay(1200);
assert.equal(fired, 0, "no wake while the quiz is pending"); assert.equal(fired, 0, "no wake while the quiz is pending");
assert.equal(jobStore.getJob(jobId)!.status, "running", "job stays running while blocked"); assert.equal(jobStore.getJob(jobId)!.status, "done", "worker is finished while delivery waits");
assert.equal(jobStore.getJob(jobId)!.delivery?.status, "pending");
// The learner answers: the real tool result clears the dangling call. // The learner answers: the real tool result clears the dangling call.
const record = chatStore.getChat(chatId)!; const record = chatStore.getChat(chatId)!;
@@ -1037,7 +1051,7 @@ describe("wake pipeline (src/server/jobs.ts)", () => {
await delay(1200); await delay(1200);
assert.equal(fired, 0, "no wake while the session sits dismissal-paused"); assert.equal(fired, 0, "no wake while the session sits dismissal-paused");
assert.equal(jobStore.getJob(jobId)!.status, "running", "the delivery holds (queued, not lost)"); assert.equal(jobStore.getJob(jobId)!.delivery?.status, "pending", "the delivery holds (queued, not lost)");
// The user's next message is the wake — its turn settles, and the // The user's next message is the wake — its turn settles, and the
// settlement re-pump lands the held delivery right after it. // settlement re-pump lands the held delivery right after it.
@@ -1109,7 +1123,8 @@ describe("Wave C: live steering handoff + settlement reconciliation", () => {
await waitFor(() => handed.length === 1, 8000, "marker handed to the live agent"); await waitFor(() => handed.length === 1, 8000, "marker handed to the live agent");
await delay(1200); await delay(1200);
assert.equal(fired, 0, "no wake while the run is live"); assert.equal(fired, 0, "no wake while the run is live");
assert.equal(jobStore.getJob(jobId)!.status, "running", "job flips only at delivery"); assert.equal(jobStore.getJob(jobId)!.status, "done", "worker finished before delivery");
assert.equal(jobStore.getJob(jobId)!.delivery?.status, "pending");
// Settlement: the run persisted the injected marker. // Settlement: the run persisted the injected marker.
unregisterLiveAgent(chatId); unregisterLiveAgent(chatId);
@@ -1222,20 +1237,15 @@ describe("Wave C: live steering handoff + settlement reconciliation", () => {
assert.equal(jobStore.getJob(jobId)!.status, "done"); assert.equal(jobStore.getJob(jobId)!.status, "done");
}); });
it("noteSteerVisible flips the job at the visibility seam — before any settle (owner ruling 2026-09-26)", async () => { it("only a durable marker acknowledges delivery, independently of worker completion", () => {
// The dock fix: the marker reaching the learner's view (wire
// message_end / write-ahead persist) flips the job NOW — the
// delivering turn may keep teaching for minutes. Idempotent with
// the settle-path finalize.
const { chatId, jobId, steer } = setup(); const { chatId, jobId, steer } = setup();
assert.equal(jobStore.getJob(jobId)!.status, "running");
noteSteerVisible(steer);
assert.equal(jobStore.getJob(jobId)!.status, "done", "flipped at the seam, no turn settled");
noteSteerVisible(steer); // the settle-path flip re-runs — no-op
assert.equal(jobStore.getJob(jobId)!.status, "done"); assert.equal(jobStore.getJob(jobId)!.status, "done");
// A probe-handoff (no jobId) is a harmless no-op. assert.equal(jobStore.getJob(jobId)!.delivery?.status, "pending");
const handoff = buildProbeHandoffSteer(chatId); chatStore.saveMessages(chatId, [...chatStore.getChat(chatId)!.messages, steer]);
noteSteerVisible(handoff); noteSteersDurable(chatId, [steer]);
noteSteersDurable(chatId, [steer]);
assert.equal(jobStore.getJob(jobId)!.delivery?.status, "delivered");
assert.equal(jobStore.getJob(jobId)!.activity, "finished");
}); });
it("the idle wake carries EVERY pending marker in ONE turn", async () => { it("the idle wake carries EVERY pending marker in ONE turn", async () => {
@@ -1261,17 +1271,25 @@ describe("Wave C: live steering handoff + settlement reconciliation", () => {
}); });
it("fails the job (error status) when the wake turn errors unrecoverably", async () => { it("a failed start stays pending without changing the worker outcome", async () => {
const { jobId, steer } = setup(); const { chatId, jobId, steer } = setup();
setTurnRunner(async () => 500); let attempts = 0;
setTurnRunner(async () => { attempts++; return 500; });
await deliverJobResult(steer).then( const delivery = deliverJobResult(steer);
() => assert.fail("expected delivery to reject"), await waitFor(() => attempts === 1, 8000, "failed start attempted");
() => undefined, await delay(1000);
); assert.equal(attempts, 1, "no automatic retry loop");
const errored = jobStore.getJob(jobId)!; assert.equal(jobStore.getJob(jobId)!.status, "done");
assert.equal(errored.status, "error"); assert.equal(jobStore.getJob(jobId)!.error, undefined);
assert.match(errored.error ?? "", /status 500/); assert.equal(jobStore.getJob(jobId)!.delivery?.status, "pending");
const handed: SteerMessage[] = [];
assert.ok(beginTurn(chatId));
registerLiveAgent(chatId, { steer: (m: SteerMessage) => handed.push(m) } as never);
assert.deepEqual(handed.map((m) => m.id), [steer.id]);
unregisterLiveAgent(chatId);
settleSteers(chatId, [steer]);
endTurn(chatId);
await delivery;
}); });
}); });
@@ -1607,14 +1625,14 @@ describe("abort (src/server/jobs.ts abortJob)", () => {
} }
}); });
it("aborting a delivering job drops the queued result steer and wakes with the abort note instead", async () => { it("a finished worker with a queued result is no longer abortable", async () => {
const chat = chatStore.createChat("Test: abort delivering job"); const chat = chatStore.createChat("Test: abort delivering job");
trackChat(chat.id); trackChat(chat.id);
chatStore.saveMessages(chat.id, [{ id: "u1", role: "user", content: "hello", timestamp: 1 } as never]); chatStore.saveMessages(chat.id, [{ id: "u1", role: "user", content: "hello", timestamp: 1 } as never]);
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-abort3" }); const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-abort3" });
trackJob(job.id); trackJob(job.id);
const result: ResearchResult = { brief: "Done: it works.", sources: [{ title: "s", url: "https://x.test" }] }; const result: ResearchResult = { brief: "Done: it works.", sources: [{ title: "s", url: "https://x.test" }] };
jobStore.updateJob(job.id, { result, activity: "delivering result" }); jobStore.updateJob(job.id, { result, status: "done", activity: "finished", delivery: { kind: "result", status: "pending" } });
// Block the pump's idle gates so the result steer sits QUEUED, not in flight. // Block the pump's idle gates so the result steer sits QUEUED, not in flight.
assert.ok(beginTurn(chat.id)); assert.ok(beginTurn(chat.id));
@@ -1626,18 +1644,18 @@ describe("abort (src/server/jobs.ts abortJob)", () => {
void deliverJobResult(buildSubagentSteer(job, result)).catch(() => { void deliverJobResult(buildSubagentSteer(job, result)).catch(() => {
/* dropped by the abort below */ /* dropped by the abort below */
}); });
await waitFor(() => jobStore.getJob(job.id)!.activity === "delivering result", 8000, "queued"); assert.equal(jobStore.getJob(job.id)!.delivery?.status, "pending");
const outcome = abortJob(job.id); const outcome = abortJob(job.id);
assert.ok(outcome.ok); assert.deepEqual(outcome, { ok: false, reason: "not_running" });
assert.equal(jobStore.getJob(job.id)!.status, "error"); assert.equal(jobStore.getJob(job.id)!.status, "done");
assert.equal(jobStore.getJob(job.id)!.error, "Aborted by learner"); assert.equal(jobStore.getJob(job.id)!.error, undefined);
endTurn(chat.id); // gates open — whatever is left in the queue fires endTurn(chat.id); // gates open — whatever is left in the queue fires
await waitFor(() => fired.length > 0, 8000, "wake fired"); await waitFor(() => fired.length > 0, 8000, "wake fired");
assert.equal(fired.length, 1, "exactly one delivery — the result steer was dropped"); assert.equal(fired.length, 1, "exactly one result delivery");
assert.equal((fired[0]!.details as { aborted: boolean }).aborted, true, "the abort steer, not the result"); assert.equal(fired[0]!.customType, "subagent-steer");
assert.equal(jobStore.getJob(job.id)!.status, "error", "terminal after the wake"); assert.equal(jobStore.getJob(job.id)!.status, "done");
}); });
it("outcome reasons: not_found for unknown, not_running for settled jobs", () => { it("outcome reasons: not_found for unknown, not_running for settled jobs", () => {
tests/render-cards.test.tsx
+2 -2
View File
@@ -156,9 +156,9 @@ describe("DispatchCard — the BYOK refusal tone (owner, 2026-09-22)", () => {
assert.ok(!html.includes("dispatch-refused")); assert.ok(!html.includes("dispatch-refused"));
}); });
it("a delivered job reads delivered, untouched by the wave", () => { it("a finished worker reads finished, independently of delivery", () => {
const html = renderToStaticMarkup(inJobs([job({ status: "done", error: undefined })])); const html = renderToStaticMarkup(inJobs([job({ status: "done", error: undefined })]));
assert.ok(html.includes("delivered")); assert.ok(html.includes("finished"));
}); });
}); });
tests/subscription-background.test.ts
+2 -2
View File
@@ -42,7 +42,7 @@ for (const p of NATIVE_SUBSCRIPTIONS) for (const revokeAfterQueue of [false, tru
const chat = stores.chatStore.createChat("Fixture mixed queue"); const model = { ...piModels().getModels(p.providerId)[0]!, provider: p.logicalId }; const chat = stores.chatStore.createChat("Fixture mixed queue"); const model = { ...piModels().getModels(p.providerId)[0]!, provider: p.logicalId };
stores.chatStore.saveSettings(chat.id, { model: { provider: p.logicalId, id: model.id } }); stores.chatStore.saveSettings(chat.id, { model: { provider: p.logicalId, id: model.id } });
const job = stores.jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "fixture-job-call" }); const result = { brief: "Fixture helper result", sources: [] }; const job = stores.jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "fixture-job-call" }); const result = { brief: "Fixture helper result", sources: [] };
stores.jobStore.updateJob(job.id, { status: "running", result }); const steer = buildSubagentSteer(job, result); const contexts: string[] = []; stores.jobStore.updateJob(job.id, { status: "done", result, delivery: { kind: "result", status: "pending" } }); const steer = buildSubagentSteer(job, result); const contexts: string[] = [];
const agent = new Agent({ initialState: { model, messages: [{ id: "fixture-initial", role: "user", content: "Initial learner input", timestamp: 1 } as AgentMessage] }, convertToLlm: toLlmMessages, streamFn: () => { const agent = new Agent({ initialState: { model, messages: [{ id: "fixture-initial", role: "user", content: "Initial learner input", timestamp: 1 } as AgentMessage] }, convertToLlm: toLlmMessages, streamFn: () => {
contexts.push(JSON.stringify(agent.state.messages)); const stream = new AssistantMessageEventStream(); contexts.push(JSON.stringify(agent.state.messages)); const stream = new AssistantMessageEventStream();
const output = { role: "assistant" as const, provider: p.logicalId, api: model.api, model: model.id, content: [{ type: "text" as const, text: "Foreground response" }], stopReason: "stop" as const, timestamp: 2, usage: { input: 2, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 3, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } } }; const output = { role: "assistant" as const, provider: p.logicalId, api: model.api, model: model.id, content: [{ type: "text" as const, text: "Foreground response" }], stopReason: "stop" as const, timestamp: 2, usage: { input: 2, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 3, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } } };
@@ -57,6 +57,6 @@ for (const p of NATIVE_SUBSCRIPTIONS) for (const revokeAfterQueue of [false, tru
assert.equal(contexts.length, 1); assert(contexts[0]!.includes("Continue my learning")); assert(!contexts[0]!.includes(steer.id)); assert.equal(contexts.length, 1); assert(contexts[0]!.includes("Continue my learning")); assert(!contexts[0]!.includes(steer.id));
const persisted = persistTurn(stores.chatStore, chat.id, 0, agent.state.messages, false); const persisted = persistTurn(stores.chatStore, chat.id, 0, agent.state.messages, false);
unregisterLiveAgent(chat.id); endTurn(chat.id); settleSteers(chat.id, persisted); await new Promise((resolve) => setImmediate(resolve)); unregisterLiveAgent(chat.id); endTurn(chat.id); settleSteers(chat.id, persisted); await new Promise((resolve) => setImmediate(resolve));
assert.equal(wakes, 0); assert.equal(stores.jobStore.getJob(job.id)!.status, "running"); assert.equal(wakes, 0); assert.equal(stores.jobStore.getJob(job.id)!.status, "done"); assert.equal(stores.jobStore.getJob(job.id)!.delivery?.status, "pending");
connection(p.providerId, true); await delivered; assert.equal(wakes, 1); assert.equal(stores.jobStore.getJob(job.id)!.status, "done"); connection(p.providerId, true); await delivered; assert.equal(wakes, 1); assert.equal(stores.jobStore.getJob(job.id)!.status, "done");
}); });