practice slice 4 — pipeline, registry, API, feed

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.
This commit is contained in:
amosblomqvist
2026-09-29 18:31:53 +03:00
parent da0cd558d0
commit 27c5d9df19
11 changed files with 1451 additions and 19 deletions
+10 -3
View File
@@ -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 });
}
@@ -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);
}
+29
View File
@@ -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);
}
}
+27
View File
@@ -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);
}
}
+46 -15
View File
@@ -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<ActivityFeed>(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<ActivityFeed>;
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"
+613
View File
@@ -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<string, RunningPractice>;
/** Per-set write mutexes (audit-lock idiom, per-set). */
__lhPracticeLocks?: Map<string, { tail: Promise<void> }>;
/** 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<string, RunningPractice> {
const g = globalThis as typeof globalThis & LhPracticeGlobal;
return (g.__lhPracticeRuns ??= new Map());
}
function practiceLocks(): Map<string, { tail: Promise<void> }> {
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<T>(setId: string, critical: () => Promise<T> | T): Promise<T> {
const locks = practiceLocks();
const prev = locks.get(setId)?.tail ?? Promise.resolve();
let release!: () => void;
const mine = new Promise<void>((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<T>(signal: AbortSignal | undefined, critical: () => Promise<T>): Promise<T> {
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<void>((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<string>): 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<ChatStore, "getChat" | "listChats">;
}
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<void> {
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<void> {
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<void> }> {
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<void> }> {
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<void> {
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<boolean> {
abortPractice(setId);
let existed = false;
await withSetLock(setId, () => {
existed = deps.practiceStore.deleteSet(setId);
});
return existed;
}
return {
createProblemSet,
submitAnswers,
abortPractice,
deleteProblemSet,
};
}
+19
View File
@@ -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
+1
View File
@@ -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";
+26 -1
View File
@@ -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(),
};
}
+132
View File
@@ -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<string, unknown>;
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<Response> {
const b = (body ?? {}) as Record<string, unknown>;
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<Response> {
if (!ID_OK.test(setId)) return Response.json({ error: "Invalid set id" }, { status: 400 });
const b = (body ?? {}) as Record<string, unknown>;
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<string, unknown>).problemId === "string" &&
typeof (a as Record<string, unknown>).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);
}
}
+529
View File
@@ -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<string, ProblemSet>();
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<ChatStore, "getChat" | "listChats"> {
const chats = new Map<string, { id: string; title: string; createdAt: string; updatedAt: string; messages: [] }>();
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<Parameters<typeof createPracticePipeline>[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<void> {
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<void>((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<void>((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<void>((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);
});
});