From 27c5d9df192169431f0b507f2d0828eb5b558e06 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Tue, 29 Sep 2026 18:31:53 +0300 Subject: [PATCH] =?UTF-8?q?practice=20slice=204=20=E2=80=94=20pipeline,=20?= =?UTF-8?q?registry,=20API,=20feed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/agent/subagents/practice-pipeline.ts (sibling of practice.ts — orchestration vs calls): createPracticePipeline(graphStore+practiceStore +chatStore+PracticeModels). Fire-and-forget front doors returning {set, done}: createProblemSet (refuse → resolve selection (chatId = ANY-provenance membership per brief; nodeIds raw) → chains → snapshot → createSet partitioning → registry row → runPartitioner → stubs land (generating) → ONE generator per problem behind the process-wide 3-permit abort-aware semaphore (__lhPracticeSlots, extraction idiom) → per-problem failure isolation → ready (errored problems included); abortPractice (idempotent: controllers + 'Aborted' marks; dropped grades keep answers, re-submittable); submitAnswers (ready-only, already-graded rejects, one grading job per answer under the same semaphore); deleteProblemSet (abort-first). Per-set write mutex on globalThis (__lhPracticeLocks, audit-lock idiom); registry __lhPracticeRuns one row per set per kind; boot sweep marks orphaned partitioning/generating sets 'Interrupted (server restarted)'. Typed errors: EmptySelection/UnknownChat/SetNotFound/SetNotReady/BadAnswers. Usage folds stage logs into the registry row (v1: completion-folded). src/server/practice.ts: thin service — strict params parsing, typed error → response mapping (503 envelope family, 400/404/409), list/get. Routes: POST+GET /api/practice/sets (202 {id}), GET+DELETE /api/practice/sets/[id], POST /api/practice/sets/[id]/answers (202). Composition root: chat-handler wires createPracticePipeline (engineFor fresh + providerKeyMissing gate) exported as ; barrel re- exports. Feed: observability activityFeed() += practice rows (ActivityPractice); dock blobs (process 'practice', GraduationCap icon, set-keyed abort); /api/activity/abort kind 'practice'. tests/practice-pipeline.test.ts — 13 tests: end-to-end (stubs mid- flight via gated generators, coverage complete), failure isolation, 3-permit semaphore pin, abort mid-generation, registry shape pins, refusal-before-create, chatId any-provenance selection, typed errors, answers→grades land + graded rejects, boot sweep (orphan + live-row), delete. 1114/1114, tsc 0, lint 0, build 0. --- app/api/activity/abort/route.ts | 13 +- app/api/practice/sets/[id]/answers/route.ts | 19 + app/api/practice/sets/[id]/route.ts | 29 + app/api/practice/sets/route.ts | 27 + components/activity-dock.tsx | 61 +- src/agent/subagents/practice-pipeline.ts | 613 ++++++++++++++++++++ src/server/chat-handler.ts | 19 + src/server/index.ts | 1 + src/server/observability.ts | 27 +- src/server/practice.ts | 132 +++++ tests/practice-pipeline.test.ts | 529 +++++++++++++++++ 11 files changed, 1451 insertions(+), 19 deletions(-) create mode 100644 app/api/practice/sets/[id]/answers/route.ts create mode 100644 app/api/practice/sets/[id]/route.ts create mode 100644 app/api/practice/sets/route.ts create mode 100644 src/agent/subagents/practice-pipeline.ts create mode 100644 src/server/practice.ts create mode 100644 tests/practice-pipeline.test.ts diff --git a/app/api/activity/abort/route.ts b/app/api/activity/abort/route.ts index fac8b14..fc74221 100644 --- a/app/api/activity/abort/route.ts +++ b/app/api/activity/abort/route.ts @@ -17,7 +17,7 @@ * Extraction abort is idempotent by design (aborting with nothing in * flight is success — the blob is gone either way) -> 200. */ -import { abortExtraction, abortJob, abortTurn } from "@/src/server"; +import { abortExtraction, abortJob, abortTurn, practice } from "@/src/server"; export async function POST(req: Request) { let body: unknown; @@ -27,8 +27,8 @@ export async function POST(req: Request) { return Response.json({ error: "Invalid JSON body" }, { status: 400 }); } const { kind, id } = (body ?? {}) as { kind?: string; id?: string }; - if ((kind !== "job" && kind !== "extraction" && kind !== "turn") || typeof id !== "string" || id === "") { - return Response.json({ error: 'Body must be {kind: "job"|"extraction"|"turn", id: string}' }, { status: 400 }); + 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)) { return Response.json({ error: "Invalid id" }, { status: 400 }); @@ -49,6 +49,13 @@ 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); + return Response.json({ ok: true }); + } abortExtraction(id); return Response.json({ ok: true }); } diff --git a/app/api/practice/sets/[id]/answers/route.ts b/app/api/practice/sets/[id]/answers/route.ts new file mode 100644 index 0000000..c6c3475 --- /dev/null +++ b/app/api/practice/sets/[id]/answers/route.ts @@ -0,0 +1,19 @@ +/** + * POST /api/practice/sets/[id]/answers — submit answers for grading + * (202; one grading job per answered problem runs in the background). + * Body: {answers: [{problemId, text}]}. Set must be ready; answers must + * be non-empty text on known, not-yet-graded problems (400/409 via the + * service's typed-error mapping). + */ +import { submitPracticeAnswersResponse } from "@/src/server"; + +export async function POST(req: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + let body: unknown; + try { + body = await req.json(); + } catch { + return Response.json({ error: "Invalid JSON body" }, { status: 400 }); + } + return submitPracticeAnswersResponse(id, body); +} diff --git a/app/api/practice/sets/[id]/route.ts b/app/api/practice/sets/[id]/route.ts new file mode 100644 index 0000000..54c4e4c --- /dev/null +++ b/app/api/practice/sets/[id]/route.ts @@ -0,0 +1,29 @@ +/** + * GET /api/practice/sets/[id] — one problem set (the exam page polls + * this while status ≠ ready; graph-view's reveal-poll idiom). + * DELETE — abort the set's live jobs, then drop the record (the unified + * delete doctrine: no orphans, no zombies). 404 when never existed. + */ +import { getPracticeSet, practice, practiceErrorResponse } from "@/src/server"; + +export async function GET(_req: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + try { + const set = getPracticeSet(id); + if (!set) return Response.json({ error: "no practice set", code: "not_found" }, { status: 404 }); + return Response.json({ set }); + } catch (err) { + return practiceErrorResponse(err); + } +} + +export async function DELETE(_req: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + try { + const existed = await practice.deleteProblemSet(id); + if (!existed) return Response.json({ error: "no practice set", code: "not_found" }, { status: 404 }); + return Response.json({ ok: true }); + } catch (err) { + return practiceErrorResponse(err); + } +} diff --git a/app/api/practice/sets/route.ts b/app/api/practice/sets/route.ts new file mode 100644 index 0000000..4ce615e --- /dev/null +++ b/app/api/practice/sets/route.ts @@ -0,0 +1,27 @@ +/** + * POST /api/practice/sets — create a problem set (202 {id}; generation + * runs in the background — the fire-and-forget pipeline). Body: + * {chatId, params} — everything this session touched + * {nodeIds, params} — a manual node selection + * Validation + typed-error mapping live in the service (server/practice). + * GET — the set list (createdAt desc, the store's order). + */ +import { createPracticeSetResponse, listPracticeSets, practiceErrorResponse } from "@/src/server"; + +export async function POST(req: Request) { + let body: unknown; + try { + body = await req.json(); + } catch { + return Response.json({ error: "Invalid JSON body" }, { status: 400 }); + } + return createPracticeSetResponse(body); +} + +export async function GET() { + try { + return Response.json({ sets: listPracticeSets() }); + } catch (err) { + return practiceErrorResponse(err); + } +} diff --git a/components/activity-dock.tsx b/components/activity-dock.tsx index dc8dd56..0af7320 100644 --- a/components/activity-dock.tsx +++ b/components/activity-dock.tsx @@ -1,7 +1,7 @@ "use client"; import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState } from "react"; -import { ActivityIcon, BotIcon, FileTextIcon, PenToolIcon, PlayIcon, SearchIcon, SquareIcon, WaypointsIcon } from "lucide-react"; +import { ActivityIcon, BotIcon, FileTextIcon, GraduationCapIcon, PenToolIcon, PlayIcon, SearchIcon, SquareIcon, WaypointsIcon } from "lucide-react"; import { cn } from "@/lib/utils"; import type { CallUsage, JobKind, JobRecord } from "@/src/core"; import { isModelsRefusal, isProviderKeyRefusal, jobCost } from "@/src/core"; @@ -63,9 +63,20 @@ 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). */ +export interface FeedPractice { + id: string; + kind: "generation" | "grading"; + title: string; + phase: string; + startedAt: number; +} + export interface ActivityFeed { jobs: FeedJob[]; extractions: FeedExtraction[]; + practice?: FeedPractice[]; } const EMPTY_FEED: ActivityFeed = { jobs: [], extractions: [] }; @@ -95,7 +106,7 @@ export function ActivityProvider({ children }: { children: React.ReactNode }) { const [feed, setFeed] = useState(EMPTY_FEED); const [nonce, setNonce] = useState(0); const refresh = useCallback(() => setNonce((n) => n + 1), []); - const hasRunning = feed.jobs.length > 0 || feed.extractions.length > 0; + const hasRunning = feed.jobs.length > 0 || feed.extractions.length > 0 || (feed.practice?.length ?? 0) > 0; useEffect(() => { let alive = true; @@ -106,7 +117,7 @@ export function ActivityProvider({ children }: { children: React.ReactNode }) { if (res.ok) { const data = (await res.json()) as Partial; if (alive) - setFeed({ jobs: data.jobs ?? [], extractions: data.extractions ?? [] }); + setFeed({ jobs: data.jobs ?? [], extractions: data.extractions ?? [], practice: data.practice ?? [] }); } } catch { // transport blip: keep the last feed, try again @@ -146,14 +157,18 @@ export function useActivity(): { feed: ActivityFeed; refresh: () => void } { // ─── Pure presentation helpers (exported for tests) ────────────────────────── -/** One dock blob: the unified view of a job or an extraction run. */ +/** One dock blob: the unified view of a job, an extraction run, or a + * practice run. */ export interface ActivityBlob { - /** job.id, or the extraction job's row id (source id, or "src#i" for - * chunk jobs — unique per row, per-chunk jobs owner 2026-09-28). */ + /** job.id, the extraction job's row id (source id, or "src#i" for + * chunk jobs — unique per row, per-chunk jobs owner 2026-09-28), or + * the practice SET id (the abort seam is set-keyed). */ id: string; - process: "job" | "extraction"; - /** Sub-agent kind for jobs (extraction blobs have none). */ + process: "job" | "extraction" | "practice"; + /** Sub-agent kind for jobs (extraction/practice blobs have none). */ jobKind?: JobKind; + /** Practice run kind (generation | grading) — practice blobs only. */ + practiceKind?: "generation" | "grading"; /** Extraction phase (extracting/auditing/merging/embedding); jobs carry none. */ phase?: string; title: string; @@ -234,6 +249,20 @@ export function toBlobs(feed: ActivityFeed): ActivityBlob[] { activity: "", }); } + for (const run of feed.practice ?? []) { + blobs.push({ + id: run.id, // the SET id — practice abort is set-keyed + process: "practice", + practiceKind: run.kind, + phase: run.phase, + title: run.title, + chatId: "", // practice is not chat-scoped + startedAt: run.startedAt, + cost: null, // practice runs ride no live cost channel (v1) + context: null, + activity: "", + }); + } blobs.sort((a, b) => a.startedAt - b.startedAt); return blobs; } @@ -295,11 +324,10 @@ export function ActivitySection({ feed: feedProp }: { feed?: ActivityFeed }) { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ - kind: blob.process === "job" ? "job" : "extraction", - // Extractions abort by SOURCE id (chatId) — chunk rows - // share one source and die together (the route's id - // charset rejects the "#" of a "src#i" job key anyway). - id: blob.process === "job" ? blob.id : blob.chatId, + kind: blob.process, + // Extractions abort by SOURCE id (chatId) — chunk rows share one + // source and die together; practice aborts by SET id (blob.id). + id: blob.process === "extraction" ? blob.chatId : blob.id, }), }); if (!res.ok) { @@ -372,11 +400,14 @@ function ActivityBlobRow({ abortState: AbortState | undefined; onAbort: () => void; }) { - const Icon = blob.process === "extraction" ? WaypointsIcon : (blob.jobKind && KIND_ICON[blob.jobKind]) || BotIcon; + const Icon = + blob.process === "extraction" ? WaypointsIcon : blob.process === "practice" ? GraduationCapIcon : (blob.jobKind && KIND_ICON[blob.jobKind]) || BotIcon; const label = blob.process === "extraction" ? `Extraction — ${blob.phase ?? "running"}` - : blob.jobKind === "research" + : blob.process === "practice" + ? `Practice — ${blob.phase ?? "running"}` + : blob.jobKind === "research" ? "Researcher" : blob.jobKind === "demo" ? "Demo builder" diff --git a/src/agent/subagents/practice-pipeline.ts b/src/agent/subagents/practice-pipeline.ts new file mode 100644 index 0000000..f0b1d96 --- /dev/null +++ b/src/agent/subagents/practice-pipeline.ts @@ -0,0 +1,613 @@ +/** + * THE PRACTICE PIPELINE (practice arc, slice 4 — 2026-09-29): the + * orchestration around the practice agents (slice 3) — set creation, + * two-stage generation (partitioner → one generator per problem), answer + * submission, grading, abort, and the live-run registry the activity + * feed reads. Sibling of practice.ts (the agents + their doctrines); + * 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). + * - 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. + * - The store adapter isSYNC (fs); reads may run unlocked, only + * read-modify-write cycles take the lock. + * + * FIRE-AND-FORGET front doors: createProblemSet persists the set at + * "partitioning" and returns immediately — the pipeline runs in the + * background; the returned `done` promise is the test/diagnostic seam + * (routes ignore it). submitAnswers mirrors it for grading. All + * background failure paths settle the persisted record (never an + * unhandled rejection); aborts mark "Aborted". + * + * GRADING FAILURE DISCIPLINE: an aborted or failed grade leaves the + * answer in place, the grade absent — the problem is re-submittable + * later (a new grading job). A second submitAnswers while grading is in + * flight is rejected (single live grading run per set). + */ + +import type { ChatStore, GraphStore, PracticeStore } from "../../core/ports.ts"; +import type { Problem, ProblemSet, ReasoningChain } from "../../core/contracts.ts"; +import { chainsForSelection } from "../../core/practice.ts"; +import { + createPracticeAgents, + type PracticeModels, + type PracticeStepLog, + type PreviousProblemRef, + type PropositionSnapshot, +} from "./practice.ts"; + +// ─── typed errors (the service maps these to responses) ────────────────────── + +/** The selector resolved to nothing practicable (400 at the route). */ +export class EmptySelectionError extends Error { + constructor(message = "selection has no propositions to practice") { + super(message); + this.name = "EmptySelectionError"; + } +} + +/** chatId path: the chat record is gone (404). */ +export class UnknownChatError extends Error { + constructor(chatId: string) { + super(`no chat ${chatId}`); + this.name = "UnknownChatError"; + } +} + +/** Set id unknown (404). */ +export class PracticeSetNotFound extends Error { + constructor(setId: string) { + super(`no practice set ${setId}`); + this.name = "PracticeSetNotFound"; + } +} + +/** The set exists but is not in the state the caller needs (409). */ +export class PracticeSetNotReady extends Error { + constructor(status: ProblemSet["status"]) { + super(`practice set is ${status}, not ready`); + this.name = "PracticeSetNotReady"; + } +} + +/** Malformed answers: unknown problem, empty text, already graded, or a + * grading run already in flight (400). */ +export class BadAnswersError extends Error { + constructor(message: string) { + super(message); + this.name = "BadAnswersError"; + } +} + +// ─── the live-run registry (globalThis — see module doc) ───────────────────── + +export type PracticeRunKind = "generation" | "grading"; +export type PracticePhase = "partitioning" | "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). */ +export interface RunningPractice { + /** `${setId}:generation` | `${setId}:grading` — the registry key. */ + key: string; + setId: string; + kind: PracticeRunKind; + phase: PracticePhase; + /** Set title + progress ("T — problem 2/5") — the dock's line. */ + 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. */ + controller: AbortController; +} + +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 { + const g = globalThis as typeof globalThis & LhPracticeGlobal; + return (g.__lhPracticeRuns ??= new Map()); +} + +function practiceLocks(): Map }> { + const g = globalThis as typeof globalThis & LhPracticeGlobal; + return (g.__lhPracticeLocks ??= new Map()); +} + +/** In-flight practice runs, oldest first (the feed sorts again; this is + * the stable, deterministic order tests pin). */ +export function practiceStatuses(): RunningPractice[] { + return [...practiceRuns().values()].sort((a, b) => a.startedAt - b.startedAt); +} + +/** The per-set serialized write section: read-modify-write cycles around + * one set's file queue here, process-wide. */ +async function withSetLock(setId: string, critical: () => Promise | T): Promise { + const locks = practiceLocks(); + const prev = locks.get(setId)?.tail ?? Promise.resolve(); + let release!: () => void; + const mine = new Promise((resolve) => (release = resolve)); + locks.set(setId, { tail: prev.then(() => mine) }); + await prev; + try { + return await critical(); + } finally { + release(); + } +} + +/** 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. */ +function newProblemId(): string { + return `pp_${Date.now().toString(36)}_${crypto.randomUUID().slice(0, 8)}`; +} + +function isAbortLike(err: unknown): boolean { + return err instanceof Error && err.name === "AbortError"; +} + +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 { + const row: RunningPractice = { + key, + setId, + kind, + phase, + label, + startedAt: Date.now(), + usage: {}, + controller: new AbortController(), + }; + practiceRuns().set(key, row); + return row; +} + +/** DEDUP INPUT (computed ONCE at creation, passed to partitioner and + * every generator): ready problems from the most recent ≤5 sets sharing + * at least one selected node. */ +function dedupInput(practiceStore: PracticeStore, selected: ReadonlySet): PreviousProblemRef[] { + const refs: PreviousProblemRef[] = []; + let setsSeen = 0; + for (const set of practiceStore.listSets()) { + if (setsSeen >= 5) break; + if (!set.nodeIds.some((id) => selected.has(id))) continue; + setsSeen++; + for (const p of set.problems) { + if (p.status === "ready" && p.statement !== "") refs.push({ statement: p.statement, chains: p.chains }); + } + } + return refs; +} + +// ─── the pipeline ──────────────────────────────────────────────────────────── + +export interface PracticePipelineDeps extends PracticeModels { + graphStore: GraphStore; + practiceStore: PracticeStore; + /** Titles for the chat-born path (and existence: unknown chat → 404). */ + chatStore: Pick; +} + +export function createPracticePipeline(deps: PracticePipelineDeps) { + const agents = createPracticeAgents(deps); + + // BOOT SWEEP: a persisted partitioning/generating set with no live row + // is a crash/restart orphan (the registry is memory-only) — mark it, + // idempotently. Runs at every pipeline instantiation (dev route + // entries re-run it; live rows are globalThis-shared and skipped). + for (const set of deps.practiceStore.listSets()) { + if ( + (set.status === "partitioning" || set.status === "generating") && + !practiceRuns().has(`${set.id}:generation`) + ) { + set.status = "error"; + set.error = "Interrupted (server restarted)"; + deps.practiceStore.saveSet(set); + } + } + + /** Persist the error state on a set (partitioner failure / abort). */ + async function markSetError(setId: string, message: string): Promise { + await withSetLock(setId, () => { + const cur = deps.practiceStore.getSet(setId); + if (!cur) return; + cur.status = "error"; + cur.error = message; + deps.practiceStore.saveSet(cur); + }); + } + + // ── generation ─────────────────────────────────────────────────────── + + async function generateSet( + set: ProblemSet, + previousProblems: PreviousProblemRef[], + row: RunningPractice, + ): Promise { + const controller = row.controller; + + // Stage 1 — the partitioner. + 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); + partitions = submitted.partitions; + } catch (err) { + await markSetError(set.id, isAbortLike(err) || controller.signal.aborted ? "Aborted" : messageOf(err)); + practiceRuns().delete(row.key); + return; + } + + // Stubs appear the moment the partition lands (numbered, generating). + const stubs: Problem[] = partitions.map((part) => ({ + id: newProblemId(), + chains: part.chains, + statement: "", + solution: "", + rubric: "", + status: "generating", + })); + const saved = await withSetLock(set.id, () => { + const cur = deps.practiceStore.getSet(set.id); + if (!cur) return null; + cur.problems = stubs; + cur.status = "generating"; + 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}`; + + // Stage 2 — one generator per problem, failure-isolated, 3-wide. + await Promise.all( + stubs.map(async (stub) => { + try { + await withPracticeSlot(controller.signal, async () => { + if (controller.signal.aborted) { + const err = new Error("aborted while queued"); + err.name = "AbortError"; + throw err; + } + const generated = await agents.runProblemGenerator({ + target: stub.chains, + fullChains: set.chains, + propositions: set.propositions, + params: set.params, + previousProblems, + signal: controller.signal, + }); + foldUsage(row.usage, "generator", generated.log); + await withSetLock(set.id, () => { + const cur = deps.practiceStore.getSet(set.id); + if (!cur) return; + const p = cur.problems.find((q) => q.id === stub.id); + if (!p) return; + p.chains = generated.problem.chains; + p.statement = generated.problem.statement; + p.solution = generated.problem.solution; + p.rubric = generated.problem.rubric; + p.status = "ready"; + p.error = undefined; + deps.practiceStore.saveSet(cur); + }); + }); + } catch (err) { + // Failure isolation: this problem errors, the set runs on. + const aborted = isAbortLike(err) || controller.signal.aborted; + await withSetLock(set.id, () => { + const cur = deps.practiceStore.getSet(set.id); + if (!cur) return; + const p = cur.problems.find((q) => q.id === stub.id); + if (!p) return; + p.status = "error"; + p.error = aborted ? "Aborted" : messageOf(err); + deps.practiceStore.saveSet(cur); + }); + } finally { + settled++; + row.label = `${set.title} — problem ${settled}/${stubs.length}`; + } + }), + ); + + // Settle: ready even with errored problems (per-problem isolation); + // aborted runs keep the error abortPractice already wrote. + 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; + } + deps.practiceStore.saveSet(cur); + }); + practiceRuns().delete(row.key); + } + + /** + * Create a problem set — fire-and-forget: persists the set at + * "partitioning" and returns it immediately; `done` resolves when + * generation has fully settled (any end state). Refusal gates + * (models/key) fire BEFORE anything is created. + */ + async function createProblemSet( + selector: { chatId: string } | { nodeIds: string[] }, + params: ProblemSet["params"], + titleHint?: string, + ): Promise<{ set: ProblemSet; done: Promise }> { + agents.refuseIfStagesUnready(); + + // Resolve the selection: chatId = any-provenance membership (owner: + // a node belongs to a session when ANY provenance point names it — + // birth-session clustering deliberately NOT used here); nodeIds = + // the raw list, unknown ids simply drop. + let sourceTitle: string | undefined; + let selected; + if ("chatId" in selector) { + const chat = deps.chatStore.getChat(selector.chatId); + if (!chat) throw new UnknownChatError(selector.chatId); + sourceTitle = chat.title; + selected = deps.graphStore.readNodes().filter((n) => n.provenance.some((pt) => pt.chatId === selector.chatId)); + } else { + const wanted = new Set(selector.nodeIds); + selected = deps.graphStore.readNodes().filter((n) => wanted.has(n.id)); + } + const selectedIds = new Set(selected.map((n) => n.id)); + const edges = deps + .graphStore.readEdges() + .filter((e) => e.type === "follows" && selectedIds.has(e.from) && selectedIds.has(e.to)); + const chains = chainsForSelection(selected, edges); + if (chains.length === 0) throw new EmptySelectionError(); + const propositions: PropositionSnapshot = {}; + for (const n of selected) propositions[n.id] = { name: n.name, statement: n.statement }; + const title = titleHint ?? sourceTitle ?? `Practice — ${new Date().toISOString().slice(0, 10)}`; + + const set = deps.practiceStore.createSet({ + title, + createdAt: new Date().toISOString(), + params, + nodeIds: [...selectedIds].sort(), + propositions, + chains, + status: "partitioning", + problems: [], + }); + + const previousProblems = dedupInput(deps.practiceStore, selectedIds); + const row = registerRow(`${set.id}:generation`, set.id, "generation", "partitioning", title); + const done = generateSet(set, previousProblems, row); + return { set, done }; + } + + // ── grading ────────────────────────────────────────────────────────── + + /** + * Save answers and spawn one grading job per answered problem — + * fire-and-forget like creation: the persisted set carries the + * answers immediately; grades land as their jobs settle. + */ + async function submitAnswers( + setId: string, + 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"); + } + const set = await withSetLock(setId, () => { + const cur = deps.practiceStore.getSet(setId); + if (!cur) throw new PracticeSetNotFound(setId); + if (cur.status !== "ready") throw new PracticeSetNotReady(cur.status); + const ids = new Set(cur.problems.map((p) => p.id)); + for (const a of answers) { + if (typeof a.problemId !== "string" || !ids.has(a.problemId)) throw new BadAnswersError(`unknown problem id`); + if (typeof a.text !== "string" || a.text.trim() === "") throw new BadAnswersError("answers must be non-empty text"); + const p = cur.problems.find((q) => q.id === a.problemId)!; + if (p.grade) throw new BadAnswersError(`problem ${a.problemId} is already graded`); + } + for (const a of answers) { + const p = cur.problems.find((q) => q.id === a.problemId)!; + p.answer = a.text; + } + deps.practiceStore.saveSet(cur); + return cur; + }); + + const row = registerRow(`${setId}:grading`, setId, "grading", "grading", `${set.title} — grading`); + const done = (async () => { + await Promise.all(answers.map((a) => gradeOne(setId, a.problemId, row))); + practiceRuns().delete(row.key); + })(); + return { set, done }; + } + + /** One grading job — 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 () => { + if (controller.signal.aborted) return; // dropped — no grade, no error state + const set = deps.practiceStore.getSet(setId); + if (!set) return; + const problem = set.problems.find((p) => p.id === problemId); + if (!problem || problem.grade) return; + // The legend scoped to THIS problem's chains (its premises + + // conclusions) — the grader sees only what it grades. + const legend: PropositionSnapshot = {}; + for (const chainId of problem.chains) { + const chain: ReasoningChain | undefined = set.chains.find((c) => c.id === chainId); + if (!chain) continue; + for (const nodeId of [chain.conclusion, ...chain.premises]) { + const p = set.propositions[nodeId]; + if (p) legend[nodeId] = p; + } + } + const graded = await agents.runGrader({ + problem: { statement: problem.statement, solution: problem.solution, rubric: problem.rubric }, + propositions: legend, + answer: problem.answer ?? "", + signal: controller.signal, + }); + foldUsage(row.usage, "grader", graded.log); + await withSetLock(setId, () => { + const cur = deps.practiceStore.getSet(setId); + if (!cur) return; + const p = cur.problems.find((q) => q.id === problemId); + if (!p || p.grade) return; + p.grade = { verdict: graded.verdict, feedback: graded.feedback, gradedAt: new Date().toISOString() }; + deps.practiceStore.saveSet(cur); + }); + }); + } catch (err) { + // Grading failure: the grade simply stays absent (re-submittable). + console.error(`[practice] grading failed for ${problemId}:`, isAbortLike(err) ? "aborted" : messageOf(err)); + } + } + + // ── 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. */ + 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); + } + } + const set = deps.practiceStore.getSet(setId); + if (!set) return; + let dirty = false; + for (const p of set.problems) { + if (p.status === "generating") { + p.status = "error"; + p.error = "Aborted"; + dirty = true; + } + } + if (set.status === "partitioning" || set.status === "generating") { + set.status = "error"; + set.error = "Aborted"; + dirty = true; + } + if (dirty) deps.practiceStore.saveSet(set); + } + + /** Delete a set: live jobs die first, then the record. Boolean like + * the store (false = never existed → the route's 404). */ + async function deleteProblemSet(setId: string): Promise { + abortPractice(setId); + let existed = false; + await withSetLock(setId, () => { + existed = deps.practiceStore.deleteSet(setId); + }); + return existed; + } + + return { + createProblemSet, + submitAnswers, + abortPractice, + deleteProblemSet, + }; +} diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 814ff8b..7b0e041 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -3,6 +3,7 @@ import { createProberAgent, createTeacherAgent, type AgentBuildOptions, type Tea import { piEngineForSpec, type PiEngine } from "../agent/models.ts"; import { createJobRunner } from "../agent/subagents/runner.ts"; import { createExtractionPipeline } from "../agent/subagents/extraction.ts"; +import { createPracticePipeline } from "../agent/subagents/practice-pipeline.ts"; import type { ModelRole } from "../../config.ts"; import { deliverJobResult, @@ -97,6 +98,24 @@ export const extraction = createExtractionPipeline({ providerKeyPresent(provider) ? null : providerKeyMissingEnvelope(provider, [tier]), }); +/** The app's ONE practice pipeline instance — the API routes reach it + * through the server barrel / service, same discipline as extraction. */ +export const practice = createPracticePipeline({ + graphStore: stores.graphStore, + practiceStore: stores.practiceStore, + chatStore: stores.chatStore, + // BYOK (owner, 2026-09-22): the practice stages resolve the chain — + // role override → slot → NOTHING — read FRESH per stage call; an empty + // chain flows through as undefined and the pipeline refuses the run + // with the typed ModelsNotConfiguredError. Direct test values still win. + engineFor: (role) => + resolveEngine(role), + // Key-presence gate: same check extraction gets — a stage engine whose + // provider key is gone refuses BEFORE any set is created. + providerKeyMissing: (provider, tier) => + providerKeyPresent(provider) ? null : providerKeyMissingEnvelope(provider, [tier]), +}); + // Wake-pipeline wiring (contracts.ts "THE ASYNC PROTOCOL" §4): the pipeline // delivers its steers through THIS composition root (one event: the steer // marker, appended natively under the turn claim), and crash recovery diff --git a/src/server/index.ts b/src/server/index.ts index c1ead82..2191837 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -6,6 +6,7 @@ export * from "./chat-handler.ts"; export * from "./chats.ts"; +export * from "./practice.ts"; export * from "./graph.ts"; export * from "./commit-positions.ts"; export * from "./jobs.ts"; diff --git a/src/server/observability.ts b/src/server/observability.ts index 4644a2f..8214951 100644 --- a/src/server/observability.ts +++ b/src/server/observability.ts @@ -20,6 +20,7 @@ */ import { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; +import { practiceStatuses, type PracticePhase } from "../agent/subagents/practice-pipeline.ts"; import { chatStore } from "../store/chats.ts"; import { graphRevision } from "./graph.ts"; import { catalogContextLimits } from "./model-catalog.ts"; @@ -27,6 +28,7 @@ import { listJobs } from "./jobs.ts"; import type { JobRecord } from "../core/contracts.ts"; export { abortExtraction, extractionStatuses, type ExtractionPhase, type RunningExtraction } from "../agent/subagents/extraction.ts"; +export { practiceStatuses, type PracticePhase, type RunningPractice } from "../agent/subagents/practice-pipeline.ts"; export { chatStore } from "../store/chats.ts"; /** A running sub-agent job, joined with its chat's title for the dock. @@ -54,6 +56,16 @@ export interface ActivityExtraction { startedAt: number; } +/** A live practice run in the dock's feed — mirrors ActivityExtraction + * (kind names which pipeline half: generation or grading). */ +export interface ActivityPractice { + id: string; + kind: "generation" | "grading"; + title: string; + phase: PracticePhase; + startedAt: number; +} + /** The dock's whole world in one payload: every running process, every view. * Jobs are RUNNING-ONLY — finished jobs belong to their chat's transcript; * the blob disappears when the process ends. Cost math is NOT done here: @@ -65,7 +77,12 @@ export interface ActivityExtraction { * graph changed underneath me" (a merge landed, a deletion purged) and * refetch clusters/layout. Reads never move it; the DELETE response * returns the same stamp. */ -export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtraction[]; graphRevision: number } { +export function activityFeed(): { + jobs: ActivityJob[]; + extractions: ActivityExtraction[]; + practice: ActivityPractice[]; + graphRevision: number; +} { const chats = chatStore.listChats(); const limits = catalogContextLimits(); const jobs: ActivityJob[] = []; @@ -93,11 +110,19 @@ export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtr phase: run.phase, startedAt: run.startedAt, })); + const practice: ActivityPractice[] = practiceStatuses().map((row) => ({ + id: row.setId, + kind: row.kind, + title: row.label, + phase: row.phase, + startedAt: row.startedAt, + })); // oldest-first on last activity: the stable blob order (ISO strings sort chronologically) jobs.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); return { jobs, extractions: extractions.sort((a, b) => a.startedAt - b.startedAt), + practice: practice.sort((a, b) => a.startedAt - b.startedAt), graphRevision: graphRevision(), }; } diff --git a/src/server/practice.ts b/src/server/practice.ts new file mode 100644 index 0000000..6dd85d9 --- /dev/null +++ b/src/server/practice.ts @@ -0,0 +1,132 @@ +/** + * THE PRACTICE SERVICE (slice 4, 2026-09-29): the thin, route-facing + * surface over the practice pipeline — request parsing/validation, typed + * error → response mapping. Routes call THIS, never the pipeline + * internals (the chats/graph service discipline). + * + * The pipeline instance is the chat-handler composition root's export + * (`practice`, next to `extraction`): engines resolve fresh per stage + * from BYOK settings, and refusals map to the 503 envelope family. + */ + +import { practice } from "./chat-handler.ts"; +import { stores } from "./stores.ts"; +import type { PracticeParams, ProblemSet } from "../core/contracts.ts"; +import { + BadAnswersError, + EmptySelectionError, + PracticeSetNotFound, + PracticeSetNotReady, + UnknownChatError, +} from "../agent/subagents/practice-pipeline.ts"; +import { + ModelsNotConfiguredError, + ProviderKeyMissingError, + PROVIDER_KEY_MISSING_CODE, +} from "../core/settings.ts"; + +// app/api uses base-URL charset ids (the activity/abort precedent). +const ID_OK = /^[A-Za-z0-9_-]+$/; + +/** GET /api/practice/sets — the list, createdAt desc (store's order). */ +export function listPracticeSets(): ProblemSet[] { + return stores.practiceStore.listSets(); +} + +/** GET /api/practice/sets/[id] — one set, or null (→ 404). */ +export function getPracticeSet(id: string): ProblemSet | null { + return stores.practiceStore.getSet(id); +} + +/** Strict params: three finite numbers in [0, 1]. */ +function parseParams(raw: unknown): PracticeParams | null { + if (typeof raw !== "object" || raw === null) return null; + const p = raw as Record; + const dial = (v: unknown): v is number => typeof v === "number" && Number.isFinite(v) && v >= 0 && v <= 1; + if (!dial(p.searchSolutionRatio) || !dial(p.cueDistance) || !dial(p.presentation)) return null; + return { + searchSolutionRatio: p.searchSolutionRatio, + cueDistance: p.cueDistance, + presentation: p.presentation, + }; +} + +/** The typed pipeline errors → HTTP (refusals → the 503 envelope family, + * exactly like the import route's extraction mapping). */ +export function practiceErrorResponse(err: unknown): Response { + if (err instanceof ModelsNotConfiguredError) { + return Response.json( + { error: err.envelope.error, code: "models_not_configured", missing: err.envelope.missing }, + { status: 503 }, + ); + } + if (err instanceof ProviderKeyMissingError) { + return Response.json( + { + error: err.envelope.error, + code: PROVIDER_KEY_MISSING_CODE, + provider: err.envelope.provider, + slots: err.envelope.slots, + }, + { status: 503 }, + ); + } + if (err instanceof EmptySelectionError) return Response.json({ error: err.message, code: "empty_selection" }, { status: 400 }); + if (err instanceof UnknownChatError) return Response.json({ error: err.message, code: "unknown_chat" }, { status: 404 }); + if (err instanceof BadAnswersError) return Response.json({ error: err.message, code: "bad_answers" }, { status: 400 }); + if (err instanceof PracticeSetNotFound) return Response.json({ error: err.message, code: "not_found" }, { status: 404 }); + if (err instanceof PracticeSetNotReady) return Response.json({ error: err.message, code: "not_ready" }, { status: 409 }); + console.error("[practice] unexpected:", err); + return Response.json({ error: "Internal error" }, { status: 500 }); +} + +/** POST /api/practice/sets — 202 {id}; the set generates in the + * background (fire-and-forget pipeline). */ +export async function createPracticeSetResponse(body: unknown): Promise { + const b = (body ?? {}) as Record; + const params = parseParams(b.params); + if (!params) { + return Response.json( + { error: "params must carry searchSolutionRatio, cueDistance, presentation as numbers in [0,1]" }, + { status: 400 }, + ); + } + try { + if (typeof b.chatId === "string" && b.chatId !== "") { + if (!ID_OK.test(b.chatId)) return Response.json({ error: "Invalid chatId" }, { status: 400 }); + const { set } = await practice.createProblemSet({ chatId: b.chatId }, params); + return Response.json({ id: set.id }, { status: 202 }); + } + if (Array.isArray(b.nodeIds) && b.nodeIds.every((id): id is string => typeof id === "string" && ID_OK.test(id))) { + if (b.nodeIds.length === 0) { + return Response.json({ error: "nodeIds is empty", code: "empty_selection" }, { status: 400 }); + } + const { set } = await practice.createProblemSet({ nodeIds: b.nodeIds }, params); + return Response.json({ id: set.id }, { status: 202 }); + } + } catch (err) { + return practiceErrorResponse(err); + } + return Response.json({ error: 'Body must be {chatId, params} or {nodeIds, params}' }, { status: 400 }); +} + +/** POST /api/practice/sets/[id]/answers — 202; grades land in the + * background. Body: {answers: [{problemId, text}]}. */ +export async function submitPracticeAnswersResponse(setId: string, body: unknown): Promise { + if (!ID_OK.test(setId)) return Response.json({ error: "Invalid set id" }, { status: 400 }); + const b = (body ?? {}) as Record; + const raw = b.answers; + if (!Array.isArray(raw) || raw.length === 0 || !raw.every((a): a is { problemId: string; text: string } => + typeof a === "object" && a !== null && + typeof (a as Record).problemId === "string" && + typeof (a as Record).text === "string", + )) { + return Response.json({ error: 'Body must be {answers: [{problemId: string, text: string}]}' }, { status: 400 }); + } + try { + const { set } = await practice.submitAnswers(setId, raw); + return Response.json({ id: set.id }, { status: 202 }); + } catch (err) { + return practiceErrorResponse(err); + } +} diff --git a/tests/practice-pipeline.test.ts b/tests/practice-pipeline.test.ts new file mode 100644 index 0000000..ad611d4 --- /dev/null +++ b/tests/practice-pipeline.test.ts @@ -0,0 +1,529 @@ +/** + * THE PRACTICE PIPELINE (slice 4, 2026-09-29): set creation → partition → + * stubs → 3-wide generator fan-out → ready; answers → grading; abort; + * boot sweep; registry shape. Faux engines (the extraction.test.ts + * seams) + in-memory stores; the globalThis registry/slots/locks are + * per-process (node:test isolates files), reset implicitly per fixture. + */ + +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { + createFauxCore, + fauxAssistantMessage, + fauxToolCall, + type FauxResponseStep, + type ToolCall, +} from "@earendil-works/pi-ai"; +import type { PiEngine } from "../src/agent/models.ts"; +import { + createPracticePipeline, + practiceStatuses, + EmptySelectionError, + UnknownChatError, + PracticeSetNotFound, + PracticeSetNotReady, + BadAnswersError, + type RunningPractice, +} from "../src/agent/subagents/practice-pipeline.ts"; +import { chainCoverage } from "../src/core/practice.ts"; +import type { GraphEdge, GraphNode, ProblemSet } from "../src/core/contracts.ts"; +import type { ChatStore, GraphStore, PracticeStore } from "../src/core/ports.ts"; +import { ModelsNotConfiguredError } from "../src/core/settings.ts"; + +// ─── in-memory stores ──────────────────────────────────────────────────────── + +function memoryGraph(nodes: GraphNode[], edges: GraphEdge[]): GraphStore { + return { + appendNode: () => {}, + readNodes: () => nodes.map((n) => ({ ...n, provenance: [...n.provenance], aliases: [...n.aliases] })), + appendEdge: () => {}, + readEdges: () => edges.map((e) => ({ ...e, provenance: [...e.provenance] })), + appendCheckpoint: () => {}, + readCheckpoints: () => [], + appendCoverage: () => {}, + readCoverage: () => [], + upsertSource: () => {}, + readSources: () => [], + appendSourceUsage: () => {}, + readPositions: () => ({}), + writePositions: () => {}, + rewriteLedgers: () => {}, + } satisfies GraphStore; +} + +function memoryPracticeStore() { + const sets = new Map(); + let n = 0; + const store: PracticeStore = { + createSet: (init) => { + const id = `ps_t${++n}`; + const set = { ...init, id, problems: init.problems.map((p) => ({ ...p })) }; + sets.set(id, set); + return structuredClone(set); + }, + getSet: (id) => { + const s = sets.get(id); + return s ? structuredClone(s) : null; + }, + listSets: () => + [...sets.values()].map((s) => structuredClone(s)).sort((a, b) => (a.createdAt < b.createdAt ? 1 : -1)), + saveSet: (set) => void sets.set(set.id, structuredClone(set)), + deleteSet: (id) => { + const had = sets.has(id); + sets.delete(id); + return had; + }, + }; + return { store, sets }; +} + +function memoryChats(): Pick { + const chats = new Map(); + chats.set("chat_1", { + id: "chat_1", + title: "Ethics lesson", + createdAt: "2026-09-29T00:00:00.000Z", + updatedAt: "2026-09-29T00:00:00.000Z", + messages: [], + }); + return { + getChat: (id) => chats.get(id) ?? null, + listChats: () => [...chats.values()].map((c) => ({ ...c, messages: 0 }) as never), + }; +} + +// ─── faux engines ──────────────────────────────────────────────────────────── + +function fauxEngine(steps: FauxResponseStep[]): PiEngine { + const faux = createFauxCore({}); + faux.setResponses(steps); + return { model: faux.getModel(), streamFn: faux.streamSimple }; +} + +/** The target chains parsed from a generator prompt ("YOUR TARGET" block). */ +function targetOf(prompt: string): string[] { + const section = prompt.split("THE FULL SET")[0] ?? ""; + const ids: string[] = []; + for (const line of section.split("\n")) { + if (line.includes("→")) ids.push(line.split("→")[1]!.trim()); + else if (line.includes("(no premises)")) ids.push(line.replace("(no premises)", "").trim()); + } + return ids; +} + +function lastUserText(context: { messages: Array<{ role: string; content: unknown }> }): string { + for (let i = context.messages.length - 1; i >= 0; i--) { + const m = context.messages[i]!; + if (m.role !== "user") continue; + return typeof m.content === "string" ? m.content : JSON.stringify(m.content); + } + return ""; +} + +/** A generator that echoes whatever target it was assigned (law-compliant). + * Queued generously: one response per ATTEMPT, and a failing target + * burns maxTurns (4) attempts before neverSubmitted fires. */ +function echoGenerator(count = 24, failTargets: string[] = []): PiEngine { + const factory: FauxResponseStep = (context) => { + const prompt = lastUserText(context as never); + const target = targetOf(prompt); + if (target.some((id) => failTargets.includes(id))) { + return fauxAssistantMessage("I would rather not submit a problem."); + } + return fauxAssistantMessage( + [ + fauxToolCall( + "submit_problem", + { + chains: target, + statement: `Problem on ${target.join(", ")}`, + solution: "The model solution.", + rubric: "- names it", + } as ToolCall["arguments"], + { id: `gen-${target.join("-")}` }, + ), + ], + { stopReason: "toolUse" }, + ); + }; + return fauxEngine(Array.from({ length: count }, () => factory)); +} + +/** A grader whose verdict encodes in the answer text ("solid" → solid). */ +function echoGrader(count: number): PiEngine { + const factory: FauxResponseStep = (context) => { + const answer = lastUserText(context as never).split("THE LEARNER'S ANSWER\n")[1] ?? ""; + return fauxAssistantMessage( + [ + fauxToolCall( + "submit_grade", + { + verdict: answer.includes("solid") ? "solid" : "partial", + feedback: "The boundary.", + } as ToolCall["arguments"], + { id: "grade-call" }, + ), + ], + { stopReason: "toolUse" }, + ); + }; + return fauxEngine(Array.from({ length: count }, () => factory)); +} + +/** A partitioner submitting the given partition. */ +function partitionModel(partitions: Array<{ chains: string[] }>): PiEngine { + return fauxEngine([ + () => + fauxAssistantMessage([fauxToolCall("submit_partition", { partitions } as ToolCall["arguments"], { id: "p-call" })], { + stopReason: "toolUse", + }), + ]); +} + +/** 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). */ +function hangingModel(count = 1): PiEngine { + const factory: FauxResponseStep = (_context, options) => + new Promise((resolve) => { + const sig = options?.signal; + if (sig?.aborted) { + resolve(fauxAssistantMessage("late")); + return; + } + sig?.addEventListener("abort", () => resolve(fauxAssistantMessage("late")), { once: true }); + }); + return fauxEngine(Array.from({ length: count }, () => factory)); +} + +// ─── fixtures ──────────────────────────────────────────────────────────────── + +const NOW = "2026-09-29T12:00:00.000Z"; +const pt = (chatId: string) => [{ chatId, messageId: "m1" }]; +/** The diamond + isolated node: chains b, c, d (multi/linked) + e (singleton). */ +function diamondGraph(): GraphStore { + const node = (id: string, name: string): GraphNode => ({ + id, + name, + statement: `${name} statement`, + aliases: [], + createdAt: NOW, + provenance: pt("chat_1"), + }); + const edge = (from: string, to: string, type: "follows" | "supports" = "follows"): GraphEdge => ({ + from, + to, + type, + provenance: pt("chat_1"), + }); + return memoryGraph( + [node("a", "A"), node("b", "B"), node("c", "C"), node("d", "D"), node("e", "E")], + [edge("a", "b"), edge("a", "c"), edge("b", "d"), edge("c", "d")], + ); +} + +const PARAMS = { searchSolutionRatio: 0.6, cueDistance: 0.5, presentation: 0.4 }; + +function fixture(partial: Partial[0]> = {}) { + const practice = memoryPracticeStore(); + const deps = { + graphStore: diamondGraph(), + practiceStore: practice.store, + chatStore: memoryChats(), + ...partial, + }; + return { practice, deps, make: () => createPracticePipeline(deps) }; +} + +/** Await a condition (the extraction until() idiom, trimmed). */ +async function until(cond: () => boolean, ms = 2000): Promise { + const t0 = Date.now(); + while (!cond()) { + if (Date.now() - t0 > ms) throw new Error("condition never became true"); + await new Promise((r) => setTimeout(r, 5)); + } +} + +// ─── generation ────────────────────────────────────────────────────────────── + +describe("createProblemSet (generation)", () => { + it("end-to-end: partition → stubs → generators fill → ready, coverage complete", async () => { + // Generators gate on our promise: the stub state is assertable + // deterministically (ungated, the whole run finishes between polls). + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const factory: FauxResponseStep = (context) => + gate.then(() => { + const target = targetOf(lastUserText(context as never)); + return fauxAssistantMessage( + [ + fauxToolCall( + "submit_problem", + { chains: target, statement: `Problem on ${target.join(", ")}`, solution: "The model solution.", rubric: "- names it" } as ToolCall["arguments"], + { id: `gen-${target.join("-")}` }, + ), + ], + { stopReason: "toolUse" }, + ); + }); + const f = fixture({ + partitioner: partitionModel([{ chains: ["b", "c"] }, { chains: ["d"] }, { chains: ["e"] }]), + problemGenerator: fauxEngine(Array.from({ length: 3 }, () => factory)), + grader: echoGrader(1), + }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["a", "b", "c", "d", "e"] }, PARAMS); + + assert.match(set.id, /^ps_/); + assert.equal(set.status, "partitioning"); + assert.equal(set.title, "Practice — 2026-09-29"); // nodeIds path: dated fallback title + + // Stubs appear the moment the partition lands — numbered, generating. + await until(() => f.practice.store.getSet(set.id)?.status === "generating"); + const mid = f.practice.store.getSet(set.id)!; + assert.equal(mid.problems.length, 3); + for (const p of mid.problems) { + assert.equal(p.status, "generating"); + assert.equal(p.statement, ""); + } + + release(); + await done; + const final = f.practice.store.getSet(set.id)!; + assert.equal(final.status, "ready"); + assert.equal(final.problems.filter((p) => p.status === "ready").length, 3); + for (const p of final.problems) assert.match(p.statement, /^Problem on /); + // Coverage: the union of problems' chains equals the set's chains. + assert.deepEqual(chainCoverage({ chains: final.chains, problems: final.problems }).missing, []); + assert.deepEqual( + final.problems.flatMap((p) => p.chains).sort(), + ["b", "c", "d", "e"], + ); + // Registry cleared. + assert.equal(practiceStatuses().length, 0); + }); + + it("per-problem failure isolation: one generator fails, the set still completes", async () => { + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }, { chains: ["e"] }]), problemGenerator: echoGenerator(8, ["e"]), grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["a", "b", "e"] }, PARAMS); // chains b + e + await done; + const final = f.practice.store.getSet(set.id)!; + assert.equal(final.status, "ready"); + const errored = final.problems.find((p) => p.status === "error"); + const ready = final.problems.find((p) => p.status === "ready"); + assert.ok(errored && /never submitted/.test(errored.error ?? "")); + assert.ok(ready); + }); + + it("the 3-permit semaphore: five generators, never more than three in flight", async () => { + const nodes = ["e1", "e2", "e3", "e4", "e5"].map((id) => ({ + id, + name: id.toUpperCase(), + statement: `${id} statement`, + aliases: [], + createdAt: NOW, + provenance: pt("chat_1"), + })); + const f = fixture({ graphStore: memoryGraph(nodes, []), partitioner: partitionModel(nodes.map((n) => ({ chains: [n.id] }))), grader: echoGrader(1) }); + + let active = 0; + let max = 0; + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const factory: FauxResponseStep = (context) => { + active++; + max = Math.max(max, active); + return gate.then(() => { + active--; + 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" })], + { stopReason: "toolUse" }, + ); + }); + }; + const deps = { ...f.deps, problemGenerator: fauxEngine(Array.from({ length: 5 }, () => factory)), grader: echoGrader(1) }; + const pipeline = createPracticePipeline(deps); + + const { done } = await pipeline.createProblemSet({ nodeIds: nodes.map((n) => n.id) }, PARAMS); + await until(() => active === 3); + assert.equal(max, 3, "exactly three generators in flight"); + release(); + await done; + assert.equal(max, 3); + }); + + it("abort mid-generation: controllers die, everything generating is 'Aborted'", async () => { + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }, { chains: ["c"] }, { chains: ["d"] }, { chains: ["e"] }]), problemGenerator: hangingModel(4), grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["a", "b", "c", "d", "e"] }, PARAMS); + await until(() => f.practice.store.getSet(set.id)?.status === "generating"); + + pipeline.abortPractice(set.id); + await done; + + const final = f.practice.store.getSet(set.id)!; + assert.equal(final.status, "error"); + assert.equal(final.error, "Aborted"); + for (const p of final.problems) { + assert.equal(p.status, "error"); + assert.equal(p.error, "Aborted"); + } + assert.equal(practiceStatuses().length, 0); + }); + + it("registry shape: kind/phase/label/usage while running, cleared after", async () => { + let seen: RunningPractice | undefined; + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const gen = fauxEngine([ + (context) => + gate.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" })], + { stopReason: "toolUse" }, + ); + }), + ]); + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }]), problemGenerator: gen, grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["b"] }, PARAMS); + 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(); + await done; + assert.equal(practiceStatuses().length, 0); + }); + + it("refusal gate fires BEFORE anything is created", async () => { + const f = fixture({}); // no engines, no engineFor + const pipeline = f.make(); + await assert.rejects(pipeline.createProblemSet({ chatId: "chat_1" }, PARAMS), ModelsNotConfiguredError); + assert.equal(f.practice.sets.size, 0, "no set persisted on refusal"); + }); +}); + +// ─── selection + typed errors ──────────────────────────────────────────────── + +describe("selection resolution + typed errors", () => { + it("chatId: every node with ANY provenance point in the chat", async () => { + const f = fixture({ partitioner: partitionModel([{ chains: ["b", "c", "d", "e"] }]), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ chatId: "chat_1" }, PARAMS); + await done; + const final = f.practice.store.getSet(set.id)!; + assert.deepEqual(final.nodeIds.sort(), ["a", "b", "c", "d", "e"]); + assert.deepEqual(final.chains.map((c) => c.id).sort(), ["b", "c", "d", "e"]); + }); + + it("unknown chat → UnknownChatError; empty selection → EmptySelectionError", async () => { + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }]), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const pipeline = f.make(); + await assert.rejects(pipeline.createProblemSet({ chatId: "chat_x" }, PARAMS), UnknownChatError); + await assert.rejects(pipeline.createProblemSet({ nodeIds: [] }, PARAMS), EmptySelectionError); + await assert.rejects(pipeline.createProblemSet({ nodeIds: ["n_ghost"] }, PARAMS), EmptySelectionError); + }); +}); + +// ─── answers + grading ─────────────────────────────────────────────────────── + +describe("submitAnswers (grading)", () => { + async function readySet() { + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }, { chains: ["e"] }]), problemGenerator: echoGenerator(2), grader: echoGrader(2) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["a", "b", "e"] }, PARAMS); + await done; + return { f, pipeline, set: f.practice.store.getSet(set.id)! }; + } + + it("saves answers, grades land per problem, grading row clears; graded refuses re-answer", async () => { + const { f, pipeline, set } = await readySet(); + const p1 = set.problems[0]!; + const p2 = set.problems[1]!; + const { done } = await pipeline.submitAnswers(set.id, [ + { problemId: p1.id, text: "a solid answer" }, + { problemId: p2.id, text: "a weak answer" }, + ]); + await done; + const final = f.practice.store.getSet(set.id)!; + const g1 = final.problems.find((p) => p.id === p1.id)!.grade; + const g2 = final.problems.find((p) => p.id === p2.id)!.grade; + assert.ok(g1 && g1.verdict === "solid", "verdict encodes from the answer text"); + assert.ok(g2 && g2.verdict === "partial"); + assert.ok(g1!.gradedAt >= NOW); + assert.equal(practiceStatuses().length, 0, "the grading row clears"); + + // Graded problems refuse re-answers; the answers persist. + await assert.rejects(pipeline.submitAnswers(set.id, [{ problemId: p1.id, text: "again" }]), BadAnswersError); + assert.equal(final.problems.find((p) => p.id === p1.id)!.answer, "a solid answer"); + }); + + it("typed errors: unknown set, not-ready set, unknown problem, empty text", async () => { + const { pipeline } = await readySet(); + await assert.rejects(pipeline.submitAnswers("ps_nope", [{ problemId: "pp_x", text: "t" }]), PracticeSetNotFound); + + const f2 = fixture({ partitioner: hangingModel(), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const p2 = f2.make(); + const { set: s2 } = await p2.createProblemSet({ nodeIds: ["b"] }, PARAMS); // stuck partitioning + await assert.rejects(p2.submitAnswers(s2.id, [{ problemId: "pp_x", text: "t" }]), PracticeSetNotReady); + p2.abortPractice(s2.id); + + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }]), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const pl = f.make(); + const { set: s3, done } = await pl.createProblemSet({ nodeIds: ["b"] }, PARAMS); + await done; + await assert.rejects(pl.submitAnswers(s3.id, [{ problemId: "pp_ghost", text: "t" }]), BadAnswersError); + const settled = f.practice.store.getSet(s3.id)!; // the post-done record (s3 itself is the creation stub) + await assert.rejects(pl.submitAnswers(s3.id, [{ problemId: settled.problems[0]!.id, text: " " }]), BadAnswersError); + }); +}); + +// ─── boot sweep + delete ───────────────────────────────────────────────────── + +describe("boot sweep + delete", () => { + it("an orphaned generating set (no live row) is marked interrupted", async () => { + const f = fixture({}); + const orphan = f.practice.store.createSet({ + title: "orphan", + createdAt: NOW, + params: PARAMS, + nodeIds: ["b"], + propositions: { b: { name: "B", statement: "s" } }, + chains: [{ id: "b", premises: ["a"], conclusion: "b" }], + status: "generating", + problems: [{ id: "pp_orphan", chains: ["b"], statement: "", solution: "", rubric: "", status: "generating" }], + }); + f.make(); // the sweep runs at construction + const swept = f.practice.store.getSet(orphan.id)!; + assert.equal(swept.status, "error"); + assert.equal(swept.error, "Interrupted (server restarted)"); + }); + + it("a LIVE generating set (registry row) is NOT swept by a second instance", async () => { + const f = fixture({ partitioner: hangingModel(), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const a = f.make(); + const { set, done } = await a.createProblemSet({ nodeIds: ["b"] }, PARAMS); // hangs partitioning + f.make(); // second pipeline instance, same process → same registry + assert.equal(f.practice.store.getSet(set.id)!.status, "partitioning"); + a.abortPractice(set.id); + await done; + }); + + it("delete: aborts live jobs first, drops the record; unknown → false", async () => { + const f = fixture({ partitioner: partitionModel([{ chains: ["b"] }]), problemGenerator: echoGenerator(1), grader: echoGrader(1) }); + const pipeline = f.make(); + const { set, done } = await pipeline.createProblemSet({ nodeIds: ["b"] }, PARAMS); + await done; + assert.equal(await pipeline.deleteProblemSet(set.id), true); + assert.equal(f.practice.store.getSet(set.id), null); + assert.equal(await pipeline.deleteProblemSet("ps_never"), false); + }); +});