From 88eb4b2ec050a5d89306a966cfd9205668179a4d Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Tue, 29 Sep 2026 20:42:33 +0300 Subject: [PATCH] =?UTF-8?q?practice=20restructure=20=E2=80=94=20per-proble?= =?UTF-8?q?m=20jobs=20+=20ONE=20shared=20background-LLM=20pool=20(owner=20?= =?UTF-8?q?ruling=202026-09-29)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rulings: each problem generation is its own visible dock job (the extraction per-chunk pattern), and practice shares extraction's semaphore — one process-wide 3-permit pool for ALL background LLM jobs; total background LLM concurrency bounded at 3. A. SHARED POOL: new src/agent/subagents/bg-llm-semaphore.ts — globalThis __lhLlmSlots, PERMIT_COUNT 3 (the ruling), the fresh-take/handed-off-permit deadlock discipline preserved for BOTH callers. extraction.ts's withExtractSlot is now a thin wrapper (job-row signal → withBgLlmSlot); __lhExtractSlots and __lhPracticeSlots both die. The partitioner now acquires a permit too (it's a background LLM job — it didn't before). B. REGISTRY ROWS: the aggregate setId:generation / setId:grading rows are replaced by setId:partition (phase partitioning, label 'Partitioning — title'), setId:gen:problemId (queued→generating, 'Problem i/N — title'), setId:grade:problemId (queued→grading, 'Grading i/N — title'). Each row carries its OWN abort controller and its OWN usage (one PracticeStepLog per row — the aggregate partitioner/generator/grader folding dies). Row key = feed id = abort key (the dock dup-key law). C. ABORT SEMANTICS: abortPracticeRow(key) routes by shape — gen row: THAT problem only (error 'Aborted', siblings continue, set settles ready with one errored problem); partition row: the whole run; grade row: that grade drops (answer kept, re-submittable). abortPractice(setId) composes from row aborts + the synchronous persisted marks (a set-abort's verdict stands before the settles re-read it — the settle step keeps an existing error status). /api/activity/abort kind practice now takes the ROW key (charset admits ':'). D. FEED + DOCK: ActivityPractice/FeedPractice carry per-row identity (id = row key, problemIndex/problemTotal); dock blobs render per row, labels carry the i/N; the aggregate row is gone. E. Boot sweep unchanged at set grain; the live-row check is now a key-prefix scan (any live row of the set protects it). The updatedAt heartbeat covers the new row scheme as-is. Tests (20 in the pipeline suite): per-problem row shape/uniqueness/ i-N labels (partition row frozen behind a gated partitioner; per-row usage ISOLATION pin — b's log never shows on c's row); abort-one-gen (sibling completes, set ready with one errored problem); partition- row abort; queued grade-row abort (dead-ticket path: answer kept, re-submittable, sibling lands); shared-pool identity + contention (practice queues behind extraction-style holders); registry/dock feed mapping. Test isolation hardened: unique per-store set ids, afterEach registry/lock/pool reset (a timed-out test can no longer poison the rest). architecture.md: jobs-pattern paragraph rewritten (per-problem rows, shared pool, abort routing). 1149/1149, tsc 0, lint 0, build 0. --- app/api/activity/abort/route.ts | 13 +- components/activity-dock.tsx | 10 +- docs/architecture.md | 2 +- src/agent/subagents/bg-llm-semaphore.ts | 92 +++++++ src/agent/subagents/extraction.ts | 63 +---- src/agent/subagents/practice-pipeline.ts | 282 ++++++++++--------- src/server/observability.ts | 13 +- tests/practice-pipeline.test.ts | 334 ++++++++++++++++++++++- 8 files changed, 588 insertions(+), 221 deletions(-) create mode 100644 src/agent/subagents/bg-llm-semaphore.ts diff --git a/app/api/activity/abort/route.ts b/app/api/activity/abort/route.ts index fc74221..b6a348a 100644 --- a/app/api/activity/abort/route.ts +++ b/app/api/activity/abort/route.ts @@ -30,7 +30,9 @@ export async function POST(req: Request) { if ((kind !== "job" && kind !== "extraction" && kind !== "turn" && kind !== "practice") || typeof id !== "string" || id === "") { return Response.json({ error: 'Body must be {kind: "job"|"extraction"|"turn"|"practice", id: string}' }, { status: 400 }); } - if (!/^[A-Za-z0-9_-]+$/.test(id)) { + // Practice row keys carry ":" (setId:gen:problemId) — the charset + // admits it for every kind (job/turn/extraction ids never use it). + if (!/^[A-Za-z0-9:_-]+$/.test(id)) { return Response.json({ error: "Invalid id" }, { status: 400 }); } @@ -50,10 +52,11 @@ export async function POST(req: Request) { // kind === "extraction": true cancellation; idempotent (nothing in // flight is success — same dock outcome) if (kind === "practice") { - // practice slice 4 (2026-09-29): abort BOTH of a set's live runs - // (generation + grading) and mark the persisted "Aborted" state — - // idempotent like extraction (nothing in flight is success). - practice.abortPractice(id); + // practice restructure (owner, 2026-09-29): id is the ROW key — + // abort routes by shape (a gen row kills THAT problem only; the + // partition row ends the run; a grade row drops that grade). + // Idempotent like extraction (nothing in flight is success). + practice.abortPracticeRow(id); return Response.json({ ok: true }); } abortExtraction(id); diff --git a/components/activity-dock.tsx b/components/activity-dock.tsx index 0af7320..2e62e4c 100644 --- a/components/activity-dock.tsx +++ b/components/activity-dock.tsx @@ -63,14 +63,18 @@ export interface FeedExtraction { startedAt: number; } -/** A live practice run — the set's generation or grading row (practice - * slice 4). id is the SET id (the abort seam is set-keyed). */ +/** A live practice row — per-problem (owner ruling, 2026-09-29), + * matching the extraction per-chunk pattern. id is the ROW KEY + * (setId:partition | setId:gen:pid | setId:grade:pid) — the abort + * seam is row-keyed; the label carries the i/N. */ export interface FeedPractice { id: string; kind: "generation" | "grading"; title: string; phase: string; startedAt: number; + problemIndex?: number; + problemTotal?: number; } export interface ActivityFeed { @@ -251,7 +255,7 @@ export function toBlobs(feed: ActivityFeed): ActivityBlob[] { } for (const run of feed.practice ?? []) { blobs.push({ - id: run.id, // the SET id — practice abort is set-keyed + id: run.id, // the ROW key — practice abort is row-keyed process: "practice", practiceKind: run.kind, phase: run.phase, diff --git a/docs/architecture.md b/docs/architecture.md index 0b81c50..dcf826d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -380,7 +380,7 @@ Practice turns the graph into problem sets the learner reasons through out loud. **Grading is feedback-only.** Submitting answers fires one **grader** job per answered problem: the grader sees the problem, the propositions its chains touch, the pre-written solution and rubric, and the learner's answer — never the full set (minimal, by ruling). It returns a verdict (`solid`/`partial`/`missed`) and feedback that names the failed recall and teaches the boundary. The verdict lands on the problem and the UI reveals it with the solution and rubric; the answer locks. Nothing reaches the graph, no level, no count, anywhere. A grader failure leaves the answer gradeless and re-submittable. -**The jobs pattern — registry, not JobRecords.** Practice runs are NOT Workbench jobs: no `JobStore` records, no steer messages, nothing ever wakes the teacher — a practice run isn't chat-scoped (a set outlives its birth session and has no lesson to re-enter; its result is itself, persisted in its own store and read by its own page). What it shares with the Workbench is the observability surface: a `globalThis` run registry (the module-instance idiom — `__lhPracticeRuns`), one row per set-phase (`setId:generation`, `setId:grading`), feeding the activity dock (GraduationCap rows; abort is set-scoped and idempotent; a boot sweep marks orphaned `partitioning`/`generating` sets as `error "Interrupted (server restarted)"` — once per process, and only when the set's `updatedAt` heartbeat is >120s stale (a live run saves at every transition and is structurally never stamped; the guard trio closed a live-stamping bug from the first live test) — the job runner's crash-honesty at set grain). Store writes serialize behind a per-set mutex; the pipeline is the sole writer of `data/practice`. +**The jobs pattern — registry, not JobRecords.** Practice runs are NOT Workbench jobs: no `JobStore` records, no steer messages, nothing ever wakes the teacher — a practice run isn't chat-scoped (a set outlives its birth session and has no lesson to re-enter; its result is itself, persisted in its own store and read by its own page). What it shares with the Workbench is the observability surface: a `globalThis` run registry (the module-instance idiom — `__lhPracticeRuns`) with PER-PROBLEM ROWS (owner ruling, 2026-09-29 — matching the extraction per-chunk pattern): `setId:partition` while the partitioner runs ("Partitioning — title"), then one `setId:gen:problemId` per problem ("queued"→"generating", label "Problem i/N — title"), and one `setId:grade:problemId` per graded answer ("queued"→"grading", "Grading i/N — title") — each row its OWN abort controller and its OWN usage accumulation; the row key IS the feed id (the dock dup-key law) and the abort target. ABORT ROUTES BY ROW SHAPE: a gen-row abort kills THAT problem only (siblings continue; the set still settles `ready` with one errored problem), a partition-row abort ends the whole run, a grade-row abort drops that grade (answer kept, re-submittable); set-level abort (delete) composes from row aborts plus the synchronous persisted marks. Model calls — partitioner included — run under the SHARED background-LLM semaphore (`bg-llm-semaphore.ts`): one process-wide pool of 3 permits for ALL background LLM jobs, extraction chunks included (total background LLM concurrency bounded at 3, whoever calls). A boot sweep marks orphaned `partitioning`/`generating` sets as `error "Interrupted (server restarted)"` — once per process, and only when the set's `updatedAt` heartbeat is >120s stale (a live run saves at every transition and is structurally never stamped; the guard trio closed a live-stamping bug from the first live test) — the job runner's crash-honesty at set grain. Store writes serialize behind a per-set mutex; the pipeline is the sole writer of `data/practice`. **API + UI.** Creation is GRAPH-DOMAIN, not chat-domain (owner ruling 2026-09-29: "problem sets always take in nodes — graph objects"): `POST /api/practice/sets` takes `{nodeIds, params, title?}` ONLY — no chatId, no clusterId; the UI resolves a picked cluster to its member node ids and passes the cluster's label as the title. `GET /api/graph/clusters` serves the picker: the birth-session partition itself (`birthSessionPartition`, same ordering — size desc, ties by id) — EVERY group regardless of size (owner ruling, same evening: MIN_CLUSTER_SIZE is a hull-drawing rule; practice availability must not inherit a visual threshold), each as `{key, kind: "chat"|"notes", label, nodeIds, nodeCount}`; a cluster whose naming record is gone keeps `label: null` — honest, never guessed. 202 `{id}` with generation fire-and-forget; 400 empty selection (empty intersection with live nodes — the vanished-nodes edge) / bad params; 503 model/key refusal. `GET /api/practice/sets` · `GET/DELETE /api/practice/sets/[id]` (delete aborts live runs first) · `POST /api/practice/sets/[id]/answers` (202; 409 when not ready). Surfaces: `/practice` (the list — status/counts rows, delete confirm, the new-set dialog with the named-level controls (a 5-detent ladder slider, a 3-detent cue slider, a Theory|Application toggle — each showing the level's name + definition) and the cluster picker — the ONLY creation entry for now; the graph-view cluster entry comes later with the free-form selection UI) and `/practice/[id]` (the exam page — a levels line in the header, numbered problems, markdown statements with math healing at the display seam, a textarea + focus-gated dictation bar per problem, one Submit per set, the post-grade reveal). Entry: the sidebar's Practice nav row. diff --git a/src/agent/subagents/bg-llm-semaphore.ts b/src/agent/subagents/bg-llm-semaphore.ts new file mode 100644 index 0000000..a37f9dd --- /dev/null +++ b/src/agent/subagents/bg-llm-semaphore.ts @@ -0,0 +1,92 @@ +/** + * THE SHARED BACKGROUND-LLM SEMAPHORE (owner ruling, 2026-09-29): ONE + * process-wide pool of 3 permits for ALL background LLM jobs — extraction + * chunks AND practice partitioner/generators/graders. Total background + * LLM concurrency is bounded at 3, whoever is calling: no subsystem gets + * a private lane past the provider-facing stance. + * + * Extraction owned this gate alone first (owner, 2026-09-28: at most 3 + * concurrent extractor calls). Practice grew its own twin that evening; + * the ruling then unified them — this module is the single truth, so + * extraction.ts and practice-pipeline.ts must never cross-import. + * + * FIFO; abort-aware (a queued caller that aborts marks its ticket dead — + * the permit handoff skips it and hands to the next waiter). + * + * FRESH-TAKE / HANDED-OFF-PERMIT discipline (the double-count deadlock + * lesson, extraction 2026-09-28): fresh takers increment `active` + * themselves; a HANDED-OFF permit does not (the releaser left `active` + * untouched for the inheritor) — counting both sides is how a semaphore + * slowly deadlocks at 3-with-empty-queue. + * + * The audit side needs no slot: audits serialize behind their own global + * lock, so by the time a job releases its permit the audit queue is the + * only remaining throttle. + */ + +/** Total background LLM concurrency, process-wide, across ALL subsystems + * (owner ruling 2026-09-29: one shared pool — extraction chunks + + * practice partitioner/generators/graders). */ +const PERMIT_COUNT = 3; + +export interface LlmSlots { + active: number; + queue: { resolve: () => void; dead: boolean }[]; +} + +interface LhLlmSemaphoreGlobal { + __lhLlmSlots?: LlmSlots; +} + +/** The process's single permit pool (globalThis — Next dev compiles each + * route entry as its own module instance; a module-level object would + * split the pool per route. One process, one truth). */ +export function bgLlmSlots(): LlmSlots { + const g = globalThis as typeof globalThis & LhLlmSemaphoreGlobal; + return (g.__lhLlmSlots ??= { active: 0, queue: [] }); +} + +/** Run `critical` under one permit of the shared pool. `signal` (the + * caller's row/abort controller) kills a queued ticket: it marks itself + * dead and rejects AbortError — the handoff skips it. */ +export async function withBgLlmSlot( + signal: AbortSignal | undefined, + critical: () => Promise, +): Promise { + const slots = bgLlmSlots(); + let freshTake = true; + if (slots.active >= PERMIT_COUNT) { + freshTake = false; + await new Promise((resolve, reject) => { + const ticket: { resolve: () => void; dead: boolean } = { resolve, dead: false }; + slots.queue.push(ticket); + if (!signal) return; + const die = () => { + ticket.dead = true; // the permit handoff must skip us + const err = new Error("background LLM job aborted while queued for a permit"); + err.name = "AbortError"; + reject(err); + }; + if (signal.aborted) die(); + else signal.addEventListener("abort", die, { once: true }); + }); + } + if (freshTake) slots.active++; + try { + return await critical(); + } finally { + // Permit handoff: a waiting ticket inherits the permit directly + // (active stays); dead tickets are skipped — their job already + // left through the abort path. + for (;;) { + const next = slots.queue.shift(); + if (!next) { + slots.active--; + break; + } + if (next.dead) continue; + next.resolve(); + break; + } + } +} diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 6f0ff66..217ec41 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -14,6 +14,7 @@ import { } from "../../core/settings.ts"; import type { PiEngine } from "../models.ts"; import { runSubagent } from "./pi-run.ts"; +import { withBgLlmSlot } from "./bg-llm-semaphore.ts"; import type { AuditVerdicts, CallUsage, @@ -526,7 +527,6 @@ interface LhExtractionGlobal { * calls process-wide — the provider-facing rate stance (the import * route's old 3-wide pool doctrine, now global). Per-chunk jobs fan * out in parallel but queue here. */ - __lhExtractSlots?: { active: number; queue: { resolve: () => void; dead: boolean }[] }; } /** THE RUN REGISTRIES live on globalThis, not module level — the @@ -796,60 +796,19 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti } -/** THE EXTRACT SEMAPHORE (owner, 2026-09-28): at most 3 concurrent - * extractor calls, process-wide — the provider-facing stance. Per-chunk - * jobs fan out freely; this gate is what they queue behind ("queued"). - * FIFO; abort-aware (a queued job that aborts marks its ticket dead — - * the permit handoff skips it and hands to the next waiter). The audit - * side needs no slot: audits already serialize behind their own global - * lock, so by the time a job releases its extract slot the audit queue - * is the only remaining throttle. */ + /** THE EXTRACT SLOT (unified 2026-09-29): extraction chunk jobs run + * under the SHARED background-LLM semaphore — one process-wide pool + * of 3 permits for extraction chunks AND practice's partitioner/ + * generators/graders (owner ruling; see bg-llm-semaphore.ts). The + * job's registry-row signal kills a queued ticket (dead-ticket + * handoff). The audit side needs no slot: audits already serialize + * behind their own global lock. */ async function withExtractSlot(jobKey: string, critical: () => Promise): Promise { - const g = globalThis as typeof globalThis & LhExtractionGlobal; - const slots = (g.__lhExtractSlots ??= { active: 0, queue: [] }); - // Fresh takers increment `active` themselves; a HANDED-OFF permit - // does not (the releaser left `active` untouched for us) — counting - // both sides is how a semaphore slowly deadlocks at 3-with-empty- - // queue. - let freshTake = true; - if (slots.active >= 3) { - freshTake = false; - await new Promise((resolve, reject) => { - const ticket: { resolve: () => void; dead: boolean } = { resolve, dead: false }; - slots.queue.push(ticket); - const signal = runningExtractions.get(jobKey)?.controller.signal; - if (!signal) return; - const die = () => { - ticket.dead = true; // the permit handoff must skip us - const err = new Error("extraction aborted while queued for an extract slot"); - err.name = "AbortError"; - reject(err); - }; - if (signal.aborted) die(); - else signal.addEventListener("abort", die, { once: true }); - }); - } - if (freshTake) slots.active++; - try { - return await critical(); - } finally { - // Permit handoff: a waiting ticket inherits the slot directly - // (active stays); dead tickets are skipped — their job already - // left through the abort path. - for (;;) { - const next = slots.queue.shift(); - if (!next) { - slots.active--; - break; - } - if (next.dead) continue; - next.resolve(); - break; - } - } + const signal = runningExtractions.get(jobKey)?.controller.signal; + return withBgLlmSlot(signal, critical); } - /** Append the run's usage record to the chat's extractions log — success + /** Append the run's usage record to the chat's extractions log — success * or failure: a failed run still spent money, and its partial usage is * exactly what the record should show (stats absent = never merged). * Idempotent per startedAt (the store dedupes). */ diff --git a/src/agent/subagents/practice-pipeline.ts b/src/agent/subagents/practice-pipeline.ts index 771f9bb..fa1feef 100644 --- a/src/agent/subagents/practice-pipeline.ts +++ b/src/agent/subagents/practice-pipeline.ts @@ -7,17 +7,23 @@ * this module owns STATE, the agents module owns CALLS. * * ARCHITECTURE (extraction's idioms, deliberately): - * - Live runs live on globalThis.__lhPracticeRuns — one row per set per - * kind ("${setId}:generation" | "${setId}:grading") — because Next dev - * compiles each route entry as its own module instance; a module-level - * Map would read empty from /api/activity. One process, one truth - * (the turn/wake/abort/extraction registries' lesson). + * - Live runs live on globalThis.__lhPracticeRuns — PER-PROBLEM ROWS + * (owner ruling, 2026-09-29: matching the extraction per-chunk + * pattern): "${setId}:partition" while the partitioner runs, then one + * "${setId}:gen:${problemId}" per problem ("queued"→"generating"), + * and one "${setId}:grade:${problemId}" per graded answer + * ("queued"→"grading") — because Next dev compiles each route entry + * as its own module instance; a module-level Map would read empty + * from /api/activity. One process, one truth (the turn/wake/abort/ + * extraction registries' lesson). * - Every set-file mutation funnels through the per-set write mutex * (the audit-lock idiom, per-set): the STORE adapter stays dumb — * serialization is THIS layer's job (store module's contract). - * - Model calls (generators, graders) gate behind a process-wide - * 3-permit semaphore (__lhPracticeSlots) — the same provider-facing - * rate stance as extraction's slots; abort-aware while queued. + * - Model calls (partitioner, generators, graders) gate behind the + * SHARED background-LLM semaphore — one process-wide pool of 3 + * permits for ALL background LLM jobs, extraction chunks included + * (owner ruling, 2026-09-29; bg-llm-semaphore.ts); abort-aware + * while queued. * - The store adapter isSYNC (fs); reads may run unlocked, only * read-modify-write cycles take the lock. * @@ -44,6 +50,7 @@ import { type PreviousProblemRef, type PropositionSnapshot, } from "./practice.ts"; +import { withBgLlmSlot } from "./bg-llm-semaphore.ts"; // ─── typed errors (the service maps these to responses) ────────────────────── @@ -83,30 +90,34 @@ export class BadAnswersError extends Error { // ─── the live-run registry (globalThis — see module doc) ───────────────────── export type PracticeRunKind = "generation" | "grading"; -export type PracticePhase = "partitioning" | "generating" | "grading"; +export type PracticePhase = "partitioning" | "queued" | "generating" | "grading"; -/** One live run row — generation or grading, per set. Mirrors - * RunningExtraction's shape discipline (label carries the progress the - * dock renders; usage folds stage logs as they land). */ +/** One live ROW — per-problem, matching the extraction per-chunk pattern + * (owner ruling, 2026-09-29). The registry key IS the row's identity: + * the feed's id and the dock's abort target. Each row owns its OWN + * abort controller (abort routes by row: one problem dies alone) and + * its OWN usage accumulation (this row's model call only). */ export interface RunningPractice { - /** `${setId}:generation` | `${setId}:grading` — the registry key. */ + /** `${setId}:partition` | `${setId}:gen:${problemId}` | + * `${setId}:grade:${problemId}` — the registry key, the feed id, + * and the abort key in one. */ key: string; setId: string; kind: PracticeRunKind; phase: PracticePhase; - /** Set title + progress ("T — problem 2/5") — the dock's line. */ + /** "Partitioning — T" | "Problem i/N — T" | "Grading i/N — T" — the + * dock's line (progress rides the label). */ label: string; startedAt: number; - /** Stage usage, folded from each runner's returned PracticeStepLog - * as its call settles (v1: completion-folded, not per-step — the - * practice blobs ride no live cost channel). */ - usage: { - partitioner?: PracticeStepLog; - generator?: PracticeStepLog; - grader?: PracticeStepLog; - }; - /** True-cancellation (the dock's abort): shared across the run's - * model calls; abort() also kills queued semaphore tickets. */ + /** 1-based position within the fan-out (gen rows: the partition's + * order; grade rows: the submitted answers' order). */ + problemIndex?: number; + problemTotal?: number; + /** THIS row's model-call usage (one stage per row: partitioner, + * generator, or grader — completion-folded, not per-step). */ + usage?: PracticeStepLog; + /** True-cancellation for THIS row's model call (the dock's abort); + * abort() also kills a queued semaphore ticket. */ controller: AbortController; } @@ -114,10 +125,6 @@ interface LhPracticeGlobal { __lhPracticeRuns?: Map; /** Per-set write mutexes (audit-lock idiom, per-set). */ __lhPracticeLocks?: Map }>; - /** The PRACTICE SEMAPHORE: 3 concurrent practice model calls, - * process-wide — shared by generators and graders (the same - * provider-facing stance as extraction's extract slots). */ - __lhPracticeSlots?: { active: number; queue: { resolve: () => void; dead: boolean }[] }; } function practiceRuns(): Map { @@ -152,48 +159,6 @@ async function withSetLock(setId: string, critical: () => Promise | T): Pr } } -/** The 3-permit practice semaphore — extraction's withExtractSlot idiom, - * signal taken directly (the caller's row controller). A queued ticket - * that aborts marks itself dead: the permit handoff skips it (fresh - * takers count themselves; handed-off permits don't — the double-count - * deadlock lesson, extraction 2026-09-28). */ -async function withPracticeSlot(signal: AbortSignal | undefined, critical: () => Promise): Promise { - const g = globalThis as typeof globalThis & LhPracticeGlobal; - const slots = (g.__lhPracticeSlots ??= { active: 0, queue: [] }); - let freshTake = true; - if (slots.active >= 3) { - freshTake = false; - await new Promise((resolve, reject) => { - const ticket: { resolve: () => void; dead: boolean } = { resolve, dead: false }; - slots.queue.push(ticket); - if (!signal) return; - const die = () => { - ticket.dead = true; - const err = new Error("practice run aborted while queued for a slot"); - err.name = "AbortError"; - reject(err); - }; - if (signal.aborted) die(); - else signal.addEventListener("abort", die, { once: true }); - }); - } - if (freshTake) slots.active++; - try { - return await critical(); - } finally { - for (;;) { - const next = slots.queue.shift(); - if (!next) { - slots.active--; - break; - } - if (next.dead) continue; - next.resolve(); - break; - } - } -} - // ─── small helpers ─────────────────────────────────────────────────────────── /** "pp_" + ts36 + rand — mirrors the store's ps_ minting. */ @@ -209,20 +174,14 @@ function messageOf(err: unknown): string { return err instanceof Error ? err.message : String(err); } -function foldUsage( - target: RunningPractice["usage"], - stage: "partitioner" | "generator" | "grader", - log: PracticeStepLog, -): void { - const cur = target[stage]; - if (!cur) target[stage] = { model: log.model, steps: [...log.steps] }; - else { - if (!cur.model) cur.model = log.model; - cur.steps.push(...log.steps); - } -} - -function registerRow(key: string, setId: string, kind: PracticeRunKind, phase: PracticePhase, label: string): RunningPractice { +function registerRow( + key: string, + setId: string, + kind: PracticeRunKind, + phase: PracticePhase, + label: string, + part?: { i: number; n: number }, +): RunningPractice { const row: RunningPractice = { key, setId, @@ -230,7 +189,8 @@ function registerRow(key: string, setId: string, kind: PracticeRunKind, phase: P phase, label, startedAt: Date.now(), - usage: {}, + problemIndex: part?.i, + problemTotal: part?.n, controller: new AbortController(), }; practiceRuns().set(key, row); @@ -283,7 +243,9 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { const staleBefore = Date.now() - 120_000; for (const set of deps.practiceStore.listSets()) { if (set.status !== "partitioning" && set.status !== "generating") continue; - if (practiceRuns().has(`${set.id}:generation`)) continue; + let live = false; + for (const key of practiceRuns().keys()) if (key.startsWith(`${set.id}:`)) { live = true; break; } + if (live) continue; const heartbeat = Date.parse(set.updatedAt ?? set.createdAt); if (Number.isFinite(heartbeat) && heartbeat > staleBefore) continue; // grace console.error(`[practice ${set.id}] boot sweep: stale ${set.status} set marked interrupted`); @@ -309,29 +271,33 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { async function generateSet( set: ProblemSet, previousProblems: PreviousProblemRef[], - row: RunningPractice, + partitionRow: RunningPractice, ): Promise { - const controller = row.controller; + const controller = partitionRow.controller; - // Stage 1 — the partitioner. + // Stage 1 — the partitioner (its OWN dock row; acquires a shared + // permit like every background LLM job). let partitions: Array<{ chains: string[] }>; try { - const submitted = await agents.runPartitioner({ - chains: set.chains, - propositions: set.propositions, - params: set.params, - previousProblems, - signal: controller.signal, - }); - foldUsage(row.usage, "partitioner", submitted.log); + const submitted = await withBgLlmSlot(controller.signal, () => + agents.runPartitioner({ + chains: set.chains, + propositions: set.propositions, + params: set.params, + previousProblems, + signal: controller.signal, + }), + ); + partitionRow.usage = submitted.log; partitions = submitted.partitions; } catch (err) { await markSetError(set.id, isAbortLike(err) || controller.signal.aborted ? "Aborted" : messageOf(err)); - practiceRuns().delete(row.key); + practiceRuns().delete(partitionRow.key); return; } - // Stubs appear the moment the partition lands (numbered, generating). + // Stubs appear the moment the partition lands (numbered, generating); + // per-problem gen rows take the partition row's place in the dock. const stubs: Problem[] = partitions.map((part) => ({ id: newProblemId(), chains: part.chains, @@ -348,33 +314,42 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { deps.practiceStore.saveSet(cur); return cur; }); - if (!saved) { - practiceRuns().delete(row.key); // deleted mid-run — stop quietly - return; - } - row.phase = "generating"; - let settled = 0; - row.label = `${set.title} — problem 0/${stubs.length}`; + practiceRuns().delete(partitionRow.key); + if (!saved) return; // deleted mid-run — stop quietly - // Stage 2 — one generator per problem, failure-isolated, 3-wide. + const genRows = stubs.map((stub, i) => + registerRow( + `${set.id}:gen:${stub.id}`, + set.id, + "generation", + "queued", + `Problem ${i + 1}/${stubs.length} — ${set.title}`, + { i: i + 1, n: stubs.length }, + ), + ); + + // Stage 2 — one generator per problem, failure-isolated, each its + // own row + abort controller (a row abort kills THAT problem only). await Promise.all( - stubs.map(async (stub) => { + stubs.map(async (stub, i) => { + const row = genRows[i]!; // 1:1 with stubs by construction try { - await withPracticeSlot(controller.signal, async () => { - if (controller.signal.aborted) { + await withBgLlmSlot(row.controller.signal, async () => { + if (row.controller.signal.aborted) { const err = new Error("aborted while queued"); err.name = "AbortError"; throw err; } + row.phase = "generating"; const generated = await agents.runProblemGenerator({ target: stub.chains, fullChains: set.chains, propositions: set.propositions, params: set.params, previousProblems, - signal: controller.signal, + signal: row.controller.signal, }); - foldUsage(row.usage, "generator", generated.log); + row.usage = generated.log; await withSetLock(set.id, () => { const cur = deps.practiceStore.getSet(set.id); if (!cur) return; @@ -391,7 +366,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { }); } catch (err) { // Failure isolation: this problem errors, the set runs on. - const aborted = isAbortLike(err) || controller.signal.aborted; + const aborted = isAbortLike(err) || row.controller.signal.aborted; await withSetLock(set.id, () => { const cur = deps.practiceStore.getSet(set.id); if (!cur) return; @@ -402,27 +377,22 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { deps.practiceStore.saveSet(cur); }); } finally { - settled++; - row.label = `${set.title} — problem ${settled}/${stubs.length}`; + practiceRuns().delete(row.key); } }), ); - // Settle: ready even with errored problems (per-problem isolation); - // aborted runs keep the error abortPractice already wrote. + // Settle: ready even with errored problems (per-problem isolation). + // A SET-level abort already wrote its verdict synchronously + // (abortPractice) — re-reading under the lock, we keep it. await withSetLock(set.id, () => { const cur = deps.practiceStore.getSet(set.id); if (!cur) return; - if (controller.signal.aborted) { - cur.status = "error"; - cur.error = "Aborted"; - } else { - cur.status = "ready"; - cur.error = undefined; - } + if (cur.status === "error") return; // set-abort's mark stands + cur.status = "ready"; + cur.error = undefined; deps.practiceStore.saveSet(cur); }); - practiceRuns().delete(row.key); } /** @@ -465,7 +435,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { }); const previousProblems = dedupInput(deps.practiceStore, selectedIds); - const row = registerRow(`${set.id}:generation`, set.id, "generation", "partitioning", title); + const row = registerRow(`${set.id}:partition`, set.id, "generation", "partitioning", `Partitioning — ${title}`); const done = generateSet(set, previousProblems, row); return { set, done }; } @@ -482,8 +452,8 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { answers: Array<{ problemId: string; text: string }>, ): Promise<{ set: ProblemSet; done: Promise }> { agents.refuseIfStagesUnready(); // grading needs its engine — refuse BEFORE saving answers - if (practiceRuns().has(`${setId}:grading`)) { - throw new BadAnswersError("a grading run is already in flight for this set"); + for (const key of practiceRuns().keys()) { + if (key.startsWith(`${setId}:grade:`)) throw new BadAnswersError("a grading run is already in flight for this set"); } const set = await withSetLock(setId, () => { const cur = deps.practiceStore.getSet(setId); @@ -504,21 +474,31 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { return cur; }); - const row = registerRow(`${setId}:grading`, setId, "grading", "grading", `${set.title} — grading`); + const gradeRows = answers.map((a, i) => + registerRow( + `${setId}:grade:${a.problemId}`, + setId, + "grading", + "queued", + `Grading ${i + 1}/${answers.length} — ${set.title}`, + { i: i + 1, n: answers.length }, + ), + ); const done = (async () => { - await Promise.all(answers.map((a) => gradeOne(setId, a.problemId, row))); - practiceRuns().delete(row.key); + await Promise.all(answers.map((a, i) => gradeOne(setId, a.problemId, gradeRows[i]!))); // 1:1 })(); return { set, done }; } - /** One grading job — abort/failure drops the grade (answer kept, - * re-submittable later); success writes it under the set lock. */ + /** One grading job — its OWN row (key `${setId}:grade:${problemId}`); + * abort/failure drops the grade (answer kept, re-submittable later); + * success writes it under the set lock. */ async function gradeOne(setId: string, problemId: string, row: RunningPractice): Promise { const controller = row.controller; try { - await withPracticeSlot(controller.signal, async () => { + await withBgLlmSlot(controller.signal, async () => { if (controller.signal.aborted) return; // dropped — no grade, no error state + row.phase = "grading"; const set = deps.practiceStore.getSet(setId); if (!set) return; const problem = set.problems.find((p) => p.id === problemId); @@ -540,7 +520,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { answer: problem.answer ?? "", signal: controller.signal, }); - foldUsage(row.usage, "grader", graded.log); + row.usage = graded.log; await withSetLock(setId, () => { const cur = deps.practiceStore.getSet(setId); if (!cur) return; @@ -553,21 +533,34 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { } catch (err) { // Grading failure: the grade simply stays absent (re-submittable). console.error(`[practice] grading failed for ${problemId}:`, isAbortLike(err) ? "aborted" : messageOf(err)); + } finally { + practiceRuns().delete(row.key); } } // ── abort + delete ─────────────────────────────────────────────────── - /** True-cancellation for a set's live runs + the persisted "Aborted" - * marks. Idempotent by design (aborting with nothing in flight is - * success — the blob is gone either way), like abortExtraction. */ + /** True-cancellation for ONE row (the dock's abort — id = the row + * key). Routes by shape: a gen-row abort kills THAT problem only + * (siblings continue; the failure-isolation catch writes "Aborted"); + * a partition-row abort ends the whole run (the partitioner's catch + * writes the set error); a grade-row abort drops that grade (the + * answer stays, re-submittable). Idempotent by design (a gone row + * is success — the blob is gone either way). */ + function abortPracticeRow(key: string): void { + const row = practiceRuns().get(key); + if (!row) return; + row.controller.abort(); + practiceRuns().delete(key); + } + + /** SET-level abort — composes from row aborts (every live row of the + * set) plus the synchronous persisted "Aborted" marks (a set-abort's + * verdict must stand before the generators' settles re-read it). + * Idempotent like abortExtraction. */ function abortPractice(setId: string): void { - for (const key of [`${setId}:generation`, `${setId}:grading`]) { - const row = practiceRuns().get(key); - if (row) { - row.controller.abort(); - practiceRuns().delete(key); - } + for (const key of [...practiceRuns().keys()]) { + if (key.startsWith(`${setId}:`)) abortPracticeRow(key); } const set = deps.practiceStore.getSet(setId); if (!set) return; @@ -602,6 +595,7 @@ export function createPracticePipeline(deps: PracticePipelineDeps) { createProblemSet, submitAnswers, abortPractice, + abortPracticeRow, deleteProblemSet, }; } diff --git a/src/server/observability.ts b/src/server/observability.ts index 8214951..5e311c5 100644 --- a/src/server/observability.ts +++ b/src/server/observability.ts @@ -56,14 +56,19 @@ export interface ActivityExtraction { startedAt: number; } -/** A live practice run in the dock's feed — mirrors ActivityExtraction - * (kind names which pipeline half: generation or grading). */ +/** A live practice ROW in the dock's feed — per-problem, matching the + * extraction per-chunk pattern (owner ruling, 2026-09-29). id is the + * ROW KEY (`${setId}:partition` | `${setId}:gen:${pid}` | + * `${setId}:grade:${pid}`) — the dock's abort target; problemIndex/ + * problemTotal ride when the row belongs to a fan-out. */ export interface ActivityPractice { id: string; kind: "generation" | "grading"; title: string; phase: PracticePhase; startedAt: number; + problemIndex?: number; + problemTotal?: number; } /** The dock's whole world in one payload: every running process, every view. @@ -111,11 +116,13 @@ export function activityFeed(): { startedAt: run.startedAt, })); const practice: ActivityPractice[] = practiceStatuses().map((row) => ({ - id: row.setId, + id: row.key, kind: row.kind, title: row.label, phase: row.phase, startedAt: row.startedAt, + problemIndex: row.problemIndex, + problemTotal: row.problemTotal, })); // oldest-first on last activity: the stable blob order (ISO strings sort chronologically) jobs.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); diff --git a/tests/practice-pipeline.test.ts b/tests/practice-pipeline.test.ts index 8862cd8..e05dbc9 100644 --- a/tests/practice-pipeline.test.ts +++ b/tests/practice-pipeline.test.ts @@ -7,7 +7,7 @@ */ import assert from "node:assert/strict"; -import { describe, it } from "node:test"; +import { afterEach, describe, it } from "node:test"; import { createFauxCore, fauxAssistantMessage, @@ -16,6 +16,7 @@ import { type ToolCall, } from "@earendil-works/pi-ai"; import type { PiEngine } from "../src/agent/models.ts"; +import { bgLlmSlots, withBgLlmSlot } from "../src/agent/subagents/bg-llm-semaphore.ts"; import { createPracticePipeline, practiceStatuses, @@ -51,12 +52,14 @@ function memoryGraph(nodes: GraphNode[], edges: GraphEdge[]): GraphStore { } satisfies GraphStore; } +let storeSeq = 0; function memoryPracticeStore() { const sets = new Map(); + const tag = `u${++storeSeq}`; let n = 0; const store: PracticeStore = { createSet: (init) => { - const id = `ps_t${++n}`; + const id = `ps_t${tag}_${++n}`; const set = { ...init, id, problems: init.problems.map((p) => ({ ...p })) }; sets.set(id, set); return structuredClone(set); @@ -183,6 +186,38 @@ function partitionModel(partitions: Array<{ chains: string[] }>): PiEngine { /** Hangs until the call's signal aborts (then resolves no submit call) — * the extraction.test.ts cancellation-mimicking idiom. Queue one step * per concurrent call (faux queues are finite). */ +/** Abort-or-gate generator: each step resolves "late" when ITS call's + * signal aborts, or a valid submit when the gate opens. The aborted + * problem burns its retries and errors; the sibling completes on the + * gate — order-independent (every step behaves identically). */ +function abortOrGateModel(count: number, gate: Promise): PiEngine { + const factory: FauxResponseStep = (context, options) => + new Promise((resolve) => { + const sig = options?.signal; + if (sig?.aborted) { + resolve(fauxAssistantMessage("late")); + return; + } + let settled = false; + const done = (m: ReturnType) => { + if (settled) return; + settled = true; + resolve(m); + }; + sig?.addEventListener("abort", () => done(fauxAssistantMessage("late")), { once: true }); + void gate.then(() => { + const target = targetOf(lastUserText(context as never)); + done( + fauxAssistantMessage( + [fauxToolCall("submit_problem", { chains: target, statement: `S ${target.join(",")}`, solution: "v", rubric: "r" } as ToolCall["arguments"], { id: `g-${target.join("-")}` })], + { stopReason: "toolUse" }, + ), + ); + }); + }); + return fauxEngine(Array.from({ length: count }, () => factory)); +} + function hangingModel(count = 1): PiEngine { const factory: FauxResponseStep = (_context, options) => new Promise((resolve) => { @@ -226,6 +261,9 @@ const PARAMS = { ladderRung: 4, cueDistance: 1, presentation: "theory" } as cons function fixture(partial: Partial[0]> = {}) { const practice = memoryPracticeStore(); + // The shared background-LLM pool lives on globalThis — start every + // test with a fresh, empty one (no inherited held permits). + delete (globalThis as { __lhLlmSlots?: unknown }).__lhLlmSlots; const deps = { graphStore: diamondGraph(), practiceStore: practice.store, @@ -234,6 +272,19 @@ function fixture(partial: Partial[0]> return { practice, deps, make: () => createPracticePipeline(deps) }; } +// Failure isolation between tests: a timed-out test can leave rows, +// locks, and held permits behind — the NEXT test must start clean. +afterEach(() => { + const g = globalThis as { + __lhPracticeRuns?: Map; + __lhPracticeLocks?: Map; + __lhLlmSlots?: unknown; + }; + g.__lhPracticeRuns?.clear(); + g.__lhPracticeLocks?.clear(); + delete g.__lhLlmSlots; +}); + /** Reset the boot sweep's once-per-process flag — each call simulates a * fresh process for the sweep's semantics (production sets it once and * never re-sweeps; tests need controlled re-arming). */ @@ -379,13 +430,246 @@ describe("createProblemSet (generation)", () => { assert.equal(practiceStatuses().length, 0); }); - it("registry shape: kind/phase/label/usage while running, cleared after", async () => { - let seen: RunningPractice | undefined; + it("registry: partition row first, then UNIQUE per-problem rows with i/N labels; every row clears", async () => { + // BOTH generators hold on their own gates — the 2-row mid-flight + // window is stable, then b releases alone (per-row usage pin). + let releaseB!: () => void; + let releaseC!: () => void; + const gateB = new Promise((r) => (releaseB = r)); + const gateC = new Promise((r) => (releaseC = r)); + const step: FauxResponseStep = (context) => { + const target = targetOf(lastUserText(context as never)); + const gate = target.includes("c") ? gateC : gateB; + return gate.then(() => + fauxAssistantMessage( + [fauxToolCall("submit_problem", { chains: target, statement: "s", solution: "v", rubric: "r" } as ToolCall["arguments"], { id: `g-${target.join("-")}` })], + { stopReason: "toolUse" }, + ), + ); + }; + const gen = fauxEngine([step, step]); + // Freeze stage 1: the partitioner holds until released. + let releasePartition!: () => void; + const partitionGate = new Promise((r) => (releasePartition = r)); + const gatedPartitioner = fauxEngine([ + () => + partitionGate.then(() => + fauxAssistantMessage( + [fauxToolCall("submit_partition", { partitions: [{ chains: ["b"] }, { chains: ["c"] }] } as ToolCall["arguments"], { id: "p1" })], + { stopReason: "toolUse" }, + ), + ), + ]); + const f = fixture({ + partitioner: gatedPartitioner, + problemGenerator: gen, + grader: echoGrader(1), + }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet(["a", "b", "c"], PARAMS); + + // Stage 1: exactly ONE row — the partition row (frozen). + await until(() => practiceStatuses().length === 1); + { + const pr = practiceStatuses()[0]!; + assert.equal(pr.key, `${set.id}:partition`, "the partition row's key shape"); + assert.equal(pr.phase, "partitioning"); + assert.equal(pr.label, `Partitioning — ${set.title}`); + assert.equal(pr.kind, "generation"); + assert.equal(pr.problemIndex, undefined, "no fan-out yet"); + } + + releasePartition(); + + // Stage 2: partition row retired; per-problem rows carry i/N. + await until(() => practiceStatuses().length === 2 && practiceStatuses().every((r) => r.phase === "generating")); + { + const rows = practiceStatuses(); + assert.ok(!rows.some((r) => r.key === `${set.id}:partition`), "partition row retired at fan-out"); + const keys = rows.map((r) => r.key); + assert.equal(new Set(keys).size, keys.length, "row keys are unique (the dock dup-key law)"); + for (const r of rows) { + assert.match(r.key, new RegExp(`^${set.id}:gen:pp_`), "gen row key shape"); + assert.equal(r.kind, "generation"); + assert.match(r.label, /^Problem [12]\/2 — /, `label carries i/N: ${r.label}`); + } + const byIndex = [...rows].sort((a, b) => (a.problemIndex ?? 0) - (b.problemIndex ?? 0)); + assert.equal(byIndex[0]!.problemIndex, 1); + assert.equal(byIndex[0]!.problemTotal, 2); + assert.equal(byIndex[1]!.problemIndex, 2); + assert.equal(byIndex[1]!.problemTotal, 2); + } + + // PER-ROW USAGE ISOLATION: b submits and settles (its row clears); + // c's row is still generating and carries NO usage — b's log lives + // on b's row only (the aggregate row would have shown it here). + releaseB(); + await until(() => practiceStatuses().length === 1); + { + const held = practiceStatuses()[0]!; + assert.match(held.key, /:gen:pp_/, "the surviving row is c's"); + assert.equal(held.usage, undefined, "no cross-contamination of usage between rows"); + } + + releaseC(); + await done; + assert.equal(practiceStatuses().length, 0, "every row clears"); + }); + + it("grading rows: per-answer keys, Grading i/N labels, queued→grading, per-row usage", async () => { let release!: () => void; const gate = new Promise((r) => (release = r)); + // hold both grades at the model seam + const holdSteps: FauxResponseStep[] = Array.from({ length: 2 }, () => () => + gate.then(() => fauxAssistantMessage([fauxToolCall("submit_grade", { verdict: "solid", feedback: "yes" } as ToolCall["arguments"], { id: "gr" })], { stopReason: "toolUse" })), + ); + const f = fixture({ + partitioner: partitionModel([{ chains: ["b"] }, { chains: ["c"] }]), + problemGenerator: echoGenerator(2), + grader: fauxEngine(holdSteps), + }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet(["a", "b", "c"], PARAMS); + await done; + const ready = f.practice.store.getSet(set.id)!; + const ids = ready.problems.map((p) => p.id); + + const sub = await pipeline.submitAnswers(set.id, [ + { problemId: ids[0]!, text: "first answer" }, + { problemId: ids[1]!, text: "second answer" }, + ]); + await until(() => practiceStatuses().length === 2 && practiceStatuses().every((r) => r.phase === "grading")); + const rows = practiceStatuses(); + for (const r of rows) { + assert.equal(r.kind, "grading"); + assert.match(r.key, new RegExp(`^${set.id}:grade:pp_`), "grade row key shape"); + assert.match(r.label, /^Grading [12]\/2 — /, `label carries i/N: ${r.label}`); + assert.ok(r.problemIndex === 1 || r.problemIndex === 2); + assert.equal(r.problemTotal, 2); + } + const keys = rows.map((r) => r.key); + assert.equal(new Set(keys).size, keys.length, "grade row keys unique"); + + release(); + await sub.done; + assert.equal(practiceStatuses().length, 0); + const graded = f.practice.store.getSet(set.id)!; + assert.ok(graded.problems.every((p) => p.grade?.verdict === "solid")); + }); + + it("abort ONE gen row: that problem dies alone, siblings complete, set reaches ready with one errored", async () => { + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const f = fixture({ + partitioner: partitionModel([{ chains: ["b"] }, { chains: ["c"] }]), + problemGenerator: abortOrGateModel(8, gate), // victim burns retries; sibling submits on gate + grader: echoGrader(1), + }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet(["a", "b", "c"], PARAMS); + await until(() => f.practice.store.getSet(set.id)?.status === "generating"); + const mid = f.practice.store.getSet(set.id)!; + const victim = mid.problems[0]!.id; + + pipeline.abortPracticeRow(`${set.id}:gen:${victim}`); + release(); // the sibling completes on the gate + await done; + + const final = f.practice.store.getSet(set.id)!; + assert.equal(final.status, "ready", "the set completes despite the single row abort"); + const dead = final.problems.find((p) => p.id === victim)!; + assert.equal(dead.status, "error"); + assert.equal(dead.error, "Aborted"); + const sibling = final.problems.find((p) => p.id !== victim)!; + assert.equal(sibling.status, "ready", "the sibling ran on"); + assert.equal(practiceStatuses().length, 0); + }); + + it("abort the PARTITION row: the whole run dies (set error Aborted)", async () => { + const f = fixture({ partitioner: hangingModel(1), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet(["b"], PARAMS); + await until(() => practiceStatuses().length === 1 && practiceStatuses()[0]!.key === `${set.id}:partition`); + + pipeline.abortPracticeRow(`${set.id}:partition`); + await done; + + const final = f.practice.store.getSet(set.id)!; + assert.equal(final.status, "error"); + assert.equal(final.error, "Aborted"); + assert.equal(practiceStatuses().length, 0); + }); + + it("abort ONE grade row (queued): that grade drops (answer kept, re-submittable), sibling grade lands", async () => { + const f = fixture({ + partitioner: partitionModel([{ chains: ["b"] }, { chains: ["c"] }]), + problemGenerator: echoGenerator(2), + grader: echoGrader(2), // sibling grade + the retry's grade + }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet(["a", "b", "c"], PARAMS); + await done; + const ready = f.practice.store.getSet(set.id)!; + const ids = ready.problems.map((p) => p.id); + + // Hold ALL THREE shared permits (AFTER the fixture — it resets the + // pool): both grade rows sit "queued" behind them, so the abort + // exercises the dead-ticket path deterministically (no model call + // for the victim at all). + let rh1!: () => void; + let rh2!: () => void; + let rh3!: () => void; + const h1 = withBgLlmSlot(undefined, () => new Promise((r) => (rh1 = r))); + const h2 = withBgLlmSlot(undefined, () => new Promise((r) => (rh2 = r))); + const h3 = withBgLlmSlot(undefined, () => new Promise((r) => (rh3 = r))); + const slots = bgLlmSlots(); + await until(() => slots.active === 3); + + const sub = await pipeline.submitAnswers(set.id, [ + { problemId: ids[0]!, text: "first" }, + { problemId: ids[1]!, text: "second" }, + ]); + await until(() => practiceStatuses().length === 2 && practiceStatuses().every((r) => r.phase === "queued")); + const victimKey = practiceStatuses()[0]!.key; + const victimId = victimKey.split(":").pop()!; + + pipeline.abortPracticeRow(victimKey); + await until(() => practiceStatuses().length === 1); // the aborted row is gone at once + assert.equal(practiceStatuses()[0]!.key !== victimKey, true); + + rh3(); // free one permit → the sibling grades + await h3; + await sub.done; + const graded = f.practice.store.getSet(set.id)!; + const dropped = graded.problems.find((p) => p.id === victimId)!; + assert.equal(dropped.grade, undefined, "the aborted grade is absent"); + assert.equal(dropped.answer, "first", "the answer is kept"); + const other = graded.problems.find((p) => p.id !== victimId)!; + assert.equal(other.grade?.verdict, "partial", "the sibling grade landed (echoGrader: no 'solid' in its text)"); + assert.equal(practiceStatuses().length, 0); + + // re-submittable: the dropped answer grades on a second submit + const resub = await pipeline.submitAnswers(set.id, [{ problemId: victimId, text: "retry answer" }]); + await resub.done; + const retried = f.practice.store.getSet(set.id)!; + assert.equal(retried.problems.find((p) => p.id === victimId)!.grade?.verdict, "partial", "the retry grades (echoGrader verdict)"); + + rh1(); + rh2(); + await h1; + await h2; + assert.equal(bgLlmSlots().active, 0, "the pool drains"); + }); + + it("shared pool: practice queues behind extraction-style holders of the SAME semaphore (identity + contention)", async () => { + // identity: one process, one pool — practice and extraction share it + assert.equal(bgLlmSlots(), bgLlmSlots()); + + let releaseGen!: () => void; + const genGate = new Promise((r) => (releaseGen = r)); const gen = fauxEngine([ (context) => - gate.then(() => { + genGate.then(() => { const target = targetOf(lastUserText(context as never)); return fauxAssistantMessage( [fauxToolCall("submit_problem", { chains: target, statement: "s", solution: "v", rubric: "r" } as ToolCall["arguments"], { id: "g" })], @@ -395,17 +679,41 @@ describe("createProblemSet (generation)", () => { ]); const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }]), problemGenerator: gen, grader: echoGrader(1) }); const pipeline = f.make(); + + // Hold all THREE permits as an extraction-style caller would (AFTER + // the fixture — it resets the pool): the partitioner (a background + // LLM job) must QUEUE behind them. + let rh1!: () => void; + let rh2!: () => void; + let rh3!: () => void; + const h1 = withBgLlmSlot(undefined, () => new Promise((r) => (rh1 = r))); + const h2 = withBgLlmSlot(undefined, () => new Promise((r) => (rh2 = r))); + const h3 = withBgLlmSlot(undefined, () => new Promise((r) => (rh3 = r))); + const slots = bgLlmSlots(); + await until(() => slots.active === 3); + const { set, done } = await pipeline.createProblemSet(["b"], PARAMS); + + // the partition row EXISTS while its call queues for a permit + await until(() => slots.queue.length === 1); + assert.equal(practiceStatuses()[0]!.key, `${set.id}:partition`, "the row exists while its call queues"); + assert.equal(practiceStatuses()[0]!.phase, "partitioning"); + + rh3(); // free one permit → the partitioner runs and finishes + await h3; await until(() => practiceStatuses().length === 1 && practiceStatuses()[0]!.phase === "generating"); - seen = practiceStatuses()[0]!; - assert.equal(seen.setId, set.id); - assert.equal(seen.kind, "generation"); - assert.ok(seen.label.includes("problem"), `label carries progress: ${seen.label}`); - assert.ok(seen.startedAt > 0); - assert.ok(seen.usage.partitioner?.model, "partitioner usage folded at submit"); - release(); + assert.match(practiceStatuses()[0]!.key, /:gen:pp_/, "the gen row took over"); + await until(() => slots.active === 3); // two holders + the generator + + releaseGen(); // the generator settles; the set completes await done; - assert.equal(practiceStatuses().length, 0); + assert.equal(f.practice.store.getSet(set.id)!.status, "ready"); + + rh1(); + rh2(); + await h1; + await h2; + assert.equal(bgLlmSlots().active, 0, "the pool drains"); }); it("refusal gate fires BEFORE anything is created", async () => {