Files
learnharness/src/agent/subagents/practice-pipeline.ts
T

653 lines
25 KiB
TypeScript

/**
* 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 — PER-PROBLEM ROWS
* (owner ruling, 2026-09-29: matching the extraction per-chunk
* pattern): "${setId}:partition" while the partitioner runs, then one
* "${setId}:gen:${problemId}" per problem ("queued"→"generating"),
* and one "${setId}:grade:${problemId}" per graded answer
* ("queued"→"grading") — because Next dev compiles each route entry
* as its own module instance; a module-level Map would read empty
* from /api/activity. One process, one truth (the turn/wake/abort/
* extraction registries' lesson).
* - Every set-file mutation funnels through the per-set write mutex
* (the audit-lock idiom, per-set): the STORE adapter stays dumb —
* serialization is THIS layer's job (store module's contract).
* - Model calls (partitioner, generators, graders) gate behind the
* SHARED background-LLM semaphore — one process-wide pool of 3
* permits for ALL background LLM jobs, extraction chunks included
* (owner ruling, 2026-09-29; bg-llm-semaphore.ts); abort-aware
* while queued.
* - The store adapter isSYNC (fs); reads may run unlocked, only
* read-modify-write cycles take the lock.
*
* 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".
*
* ONE ATTEMPT PER PROBLEM (owner ruling 2026-09-29): a problem accepts
* a single submission — once `answer` is set the problem is done, the
* route refuses re-submission. Grades are per-problem: submitting
* another problem while one grades is fine (the shared LLM pool bounds
* parallelism); only the SAME problem's live grade row blocks (which
* the one-attempt rule already prevents client-side).
*/
import type { GraphStore, PracticeStore } from "../../core/ports.ts";
import { emit } from "../../core/events.ts";
import type { Problem, ProblemSet, ReasoningChain } from "../../core/contracts.ts";
import { chainsForSelection } from "../../core/practice.ts";
import { getGraph } from "../../core/graph.ts";
import {
createPracticeAgents,
type PracticeModels,
type PracticeStepLog,
type PreviousProblemRef,
type PropositionSnapshot,
} from "./practice.ts";
import { withBgLlmSlot } from "./bg-llm-semaphore.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";
}
}
/** 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" | "queued" | "generating" | "grading";
/** One live ROW — per-problem, matching the extraction per-chunk pattern
* (owner ruling, 2026-09-29). The registry key IS the row's identity:
* the feed's id and the dock's abort target. Each row owns its OWN
* abort controller (abort routes by row: one problem dies alone) and
* its OWN usage accumulation (this row's model call only). */
export interface RunningPractice {
/** `${setId}:partition` | `${setId}:gen:${problemId}` |
* `${setId}:grade:${problemId}` — the registry key, the feed id,
* and the abort key in one. */
key: string;
setId: string;
kind: PracticeRunKind;
phase: PracticePhase;
/** "Partitioning — T" | "Problem i/N — T" | "Grading i/N — T" — the
* dock's line (progress rides the label). */
label: string;
startedAt: number;
/** 1-based position within the fan-out (gen rows: the partition's
* order; grade rows: the submitted answers' order). */
problemIndex?: number;
problemTotal?: number;
/** THIS row's model-call usage (one stage per row: partitioner,
* generator, or grader — completion-folded, not per-step). */
usage?: PracticeStepLog;
/** True-cancellation for THIS row's model call (the dock's abort);
* abort() also kills a queued semaphore ticket. */
controller: AbortController;
}
interface LhPracticeGlobal {
__lhPracticeRuns?: Map<string, RunningPractice>;
/** Per-set write mutexes (audit-lock idiom, per-set). */
__lhPracticeLocks?: Map<string, { tail: Promise<void> }>;
}
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();
}
}
// ─── 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 registerRow(
key: string,
setId: string,
kind: PracticeRunKind,
phase: PracticePhase,
label: string,
part?: { i: number; n: number },
): RunningPractice {
const row: RunningPractice = {
key,
setId,
kind,
phase,
label,
startedAt: Date.now(),
problemIndex: part?.i,
problemTotal: part?.n,
controller: new AbortController(),
};
practiceRuns().set(key, row);
// The dock learns the row at the moment of cause (push S5, 2026-10-02):
// notify-after-write — the registry is the feed's truth.
emit("activity-changed", {});
return row;
}
/** Row over (settled, failed, aborted — every exit): the row leaves the
* feed. ONE seam for all delete sites so no exit path can miss the
* announce (push S5, 2026-10-02). */
function finishPractice(key: string): void {
practiceRuns().delete(key);
emit("activity-changed", {});
}
/** 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;
}
export function createPracticePipeline(deps: PracticePipelineDeps) {
const agents = createPracticeAgents(deps);
// BOOT SWEEP (owner bug fix, live test 2026-09-29 — the sweep once
// stamped a LIVE run): a persisted partitioning/generating set with no
// live row AND a heartbeat older than the grace window is a crash/
// restart orphan (the registry is memory-only) — mark it, idempotently.
// THREE guards against stamping the living:
// 1. ONCE PER PROCESS (globalThis flag) — an instance could otherwise
// evaluate against a registry whose fresh row it can't observe
// (dev route bundles, restart paths).
// 2. GRACE WINDOW (120s over updatedAt ?? createdAt) — a live run's
// WATCHDOG re-stamps the file every 30s (cross-worker liveness —
// the registry is NOT shared between dev route workers), so a
// live run is structurally never 120s stale; a true crash stops
// the beats and is minutes stale at the next boot.
// 3. Live-row check (globalThis-shared registry) as before.
const sweepFlag = globalThis as { __lhPracticeSweepDone?: boolean };
if (!sweepFlag.__lhPracticeSweepDone) {
sweepFlag.__lhPracticeSweepDone = true;
const staleBefore = Date.now() - 120_000;
for (const set of deps.practiceStore.listSets()) {
if (set.status !== "partitioning" && set.status !== "generating") continue;
let live = false;
for (const key of practiceRuns().keys()) if (key.startsWith(`${set.id}:`)) { live = true; break; }
if (live) continue;
const heartbeat = Date.parse(set.updatedAt ?? set.createdAt);
if (Number.isFinite(heartbeat) && heartbeat > staleBefore) continue; // grace
console.error(`[practice ${set.id}] boot sweep: stale ${set.status} set marked interrupted`);
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[],
partitionRow: RunningPractice,
note?: string,
): Promise<void> {
const controller = partitionRow.controller;
// PERSISTENT HEARTBEAT (owner live test #2, 2026-09-29 — the sweep
// stamped LIVE runs again): dev serves routes from workers that do
// NOT share this process's registry, so memory liveness cannot
// protect a cross-worker sweep, and a long-running last problem can
// exceed the 120s grace between store writes. The watchdog
// re-stamps updatedAt on the FILE every 30s while THIS run is live
// (4-beat tolerance inside the grace window) — any process's sweep
// then sees a fresh heartbeat; a true crash stops the beats and the
// orphan goes stale honestly.
const beat = setInterval(() => {
void withSetLock(set.id, () => {
const cur = deps.practiceStore.getSet(set.id);
if (!cur) return;
if (cur.status !== "partitioning" && cur.status !== "generating") return;
deps.practiceStore.saveSet(cur);
});
}, 30_000);
beat.unref?.();
try {
// Stage 1 — the partitioner (its OWN dock row; acquires a shared
// permit like every background LLM job).
let partitions: Array<{ chains: string[] }>;
try {
const submitted = await withBgLlmSlot(controller.signal, () =>
agents.runPartitioner({
chains: set.chains,
propositions: set.propositions,
params: set.params,
previousProblems,
note,
signal: controller.signal,
}),
);
partitionRow.usage = submitted.log;
partitions = submitted.partitions;
} catch (err) {
await markSetError(set.id, isAbortLike(err) || controller.signal.aborted ? "Aborted" : messageOf(err));
finishPractice(partitionRow.key);
return;
}
// Stubs appear the moment the partition lands (numbered, generating);
// per-problem gen rows take the partition row's place in the dock.
const stubs: Problem[] = partitions.map((part) => ({
id: newProblemId(),
chains: part.chains,
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;
});
finishPractice(partitionRow.key);
if (!saved) return; // deleted mid-run — stop quietly
const genRows = stubs.map((stub, i) =>
registerRow(
`${set.id}:gen:${stub.id}`,
set.id,
"generation",
"queued",
`Problem ${i + 1}/${stubs.length} — ${set.title}`,
{ i: i + 1, n: stubs.length },
),
);
// Stage 2 — one generator per problem, failure-isolated, each its
// own row + abort controller (a row abort kills THAT problem only).
await Promise.all(
stubs.map(async (stub, i) => {
const row = genRows[i]!; // 1:1 with stubs by construction
try {
await withBgLlmSlot(row.controller.signal, async () => {
if (row.controller.signal.aborted) {
const err = new Error("aborted while queued");
err.name = "AbortError";
throw err;
}
row.phase = "generating";
emit("activity-changed", {}); // in-place phase write — announce it
const generated = await agents.runProblemGenerator({
target: stub.chains,
fullChains: set.chains,
propositions: set.propositions,
params: set.params,
previousProblems,
note,
signal: row.controller.signal,
});
row.usage = 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) || row.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 {
finishPractice(row.key);
}
}),
);
// Settle: ready even with errored problems (per-problem isolation).
// A SET-level abort already wrote its verdict synchronously
// (abortPractice) — re-reading under the lock, we keep it.
await withSetLock(set.id, () => {
const cur = deps.practiceStore.getSet(set.id);
if (!cur) return;
if (cur.status === "error") return; // set-abort's mark stands
cur.status = "ready";
cur.error = undefined;
deps.practiceStore.saveSet(cur);
});
} finally {
clearInterval(beat);
}
}
/**
* 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(
nodeIds: string[],
params: ProblemSet["params"],
titleHint?: string,
note?: string,
): Promise<{ set: ProblemSet; done: Promise<void> }> {
agents.refuseIfStagesUnready();
// Graph-domain selection (owner ruling 2026-09-29: problem sets
// take NODES — the UI resolves a picked cluster to its member ids;
// unknown ids simply drop, an empty live intersection refuses).
const wanted = new Set(nodeIds);
const selected = deps.graphStore.readNodes().filter((n) => wanted.has(n.id));
const selectedIds = new Set(selected.map((n) => n.id));
// The STORED chains (chains-primary, owner 2026-09-30): chainsForSelection
// keeps a chain iff every member is selected — the same inclusion law as
// the pre-chains edge grouping; never the derived fan-out.
const chains = chainsForSelection(selected, getGraph(deps.graphStore).chains);
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 ?? `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: [],
...(note ? { note } : {}),
});
const previousProblems = dedupInput(deps.practiceStore, selectedIds);
const row = registerRow(`${set.id}:partition`, set.id, "generation", "partitioning", `Partitioning — ${title}`);
const done = generateSet(set, previousProblems, row, note);
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
for (const a of answers) {
if (practiceRuns().has(`${setId}:grade:${(a as { problemId?: unknown }).problemId}`)) {
throw new BadAnswersError("this problem's grade is already in flight");
}
}
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)!;
// ONE attempt (owner 2026-09-29): an existing answer ends the
// problem — graded or not, it is never re-submitted.
if (p.answer !== undefined) throw new BadAnswersError(`problem ${a.problemId} was already answered`);
}
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 gradeRows = answers.map((a, i) =>
registerRow(
`${setId}:grade:${a.problemId}`,
setId,
"grading",
"queued",
`Grading ${i + 1}/${answers.length} — ${set.title}`,
{ i: i + 1, n: answers.length },
),
);
const done = (async () => {
await Promise.all(answers.map((a, i) => gradeOne(setId, a.problemId, gradeRows[i]!))); // 1:1
})();
return { set, done };
}
/** One grading job — its OWN row (key `${setId}:grade:${problemId}`);
* abort/failure drops the grade (answer kept, re-submittable later);
* success writes it under the set lock. */
async function gradeOne(setId: string, problemId: string, row: RunningPractice): Promise<void> {
const controller = row.controller;
try {
await withBgLlmSlot(controller.signal, async () => {
if (controller.signal.aborted) return; // dropped — no grade, no error state
row.phase = "grading";
emit("activity-changed", {}); // in-place phase write — announce it
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,
});
row.usage = 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 = { score: graded.score, reasoning: graded.reasoning, gradedAt: new Date().toISOString() };
deps.practiceStore.saveSet(cur);
});
});
} catch (err) {
// Grading failure: the grade simply stays absent (the answer
// stands — one attempt; the UI shows "grade interrupted").
console.error(`[practice] grading failed for ${problemId}:`, isAbortLike(err) ? "aborted" : messageOf(err));
} finally {
finishPractice(row.key);
}
}
// ── abort + delete ───────────────────────────────────────────────────
/** True-cancellation for ONE row (the dock's abort — id = the row
* key). Routes by shape: a gen-row abort kills THAT problem only
* (siblings continue; the failure-isolation catch writes "Aborted");
* a partition-row abort ends the whole run (the partitioner's catch
* writes the set error); a grade-row abort drops that grade (the
* answer stays, re-submittable). Idempotent by design (a gone row
* is success — the blob is gone either way). */
function abortPracticeRow(key: string): void {
const row = practiceRuns().get(key);
if (!row) return;
row.controller.abort();
finishPractice(key);
}
/** SET-level abort — composes from row aborts (every live row of the
* set) plus the synchronous persisted "Aborted" marks (a set-abort's
* verdict must stand before the generators' settles re-read it).
* Idempotent like abortExtraction. */
function abortPractice(setId: string): void {
for (const key of [...practiceRuns().keys()]) {
if (key.startsWith(`${setId}:`)) abortPracticeRow(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,
abortPracticeRow,
deleteProblemSet,
};
}