From 372d7f99752caa19044046fddad2ba4b6c5fb556 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Tue, 22 Sep 2026 13:59:16 +0300 Subject: [PATCH] =?UTF-8?q?feat(extraction):=20stage=20engines=20resolve?= =?UTF-8?q?=20the=20slot=20chain=20=E2=80=94=20engineFor=20on=20the=20pipe?= =?UTF-8?q?line?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chunk B, item 3 of the search port + slots wave (2026-09-20): the flagged gap closed — ExtractionModels deps were engine VALUES, keeping extractor/auditor/labeler config-pin-bound while the job runner went slot-native. The pipeline now takes the runner's exact treatment: - ExtractionPipelineDeps.engineFor: (role) => PiEngine — resolved FRESH per stage call, so a settings PUT between stages lands. Direct test values (extractor/auditor/labeler) remain the priority seam, checked FIRST; absent values fall to the chain; absent chain → config pins (unchanged). - All six stage call sites (chat + notes extract/audit, the cluster namer x2) resolve through stageEngine(role, direct). - chat-handler's extraction composition injects the full chain — piEngineForSpec(resolveRoleSpec(role, readSettings(), MODEL_PINS)) — same wiring as the job runner. Chunk B, item 4 (session-level slot-anchored model picks): VERIFIED already complete from Platform's model-slots wave — the conversation engine chain (chat-handler) anchors on effective.models.strong, the picker's default surface (catalogDefaultsFor) falls back to strong, both pinned in tests. No phase split (user decision, 2026-09-20). Capability-filtered pickers live in the settings UI (main 343be10). Tests: engineFor fills the stages end-to-end (both propositions merged through the chain, consulted in pipeline order) + the direct-value priority pin (engineFor that throws when consulted). Gates: typecheck, 795/795 tests, lint, build — all green. --- src/agent/subagents/extraction.ts | 27 ++++++++++---- src/server/chat-handler.ts | 7 ++++ tests/extraction.test.ts | 58 +++++++++++++++++++++++++++++++ 3 files changed, 85 insertions(+), 7 deletions(-) diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 53766d9..468cdca 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -556,11 +556,17 @@ export function abortExtraction(chatId: string): boolean { // ─── the pipeline ─────────────────────────────────────────────────────────── export interface ExtractionModels { - /** Test seam: a scripted engine per stage (production uses the roles). */ + /** Test seam: a scripted engine per stage — checked FIRST, ahead of the + * engineFor chain (production wires no direct values). */ extractor?: PiEngine; auditor?: PiEngine; /** Test seam for the cluster namer (post-extraction garnish). */ labeler?: PiEngine; + /** Settings-resolved engine per role (model slots wave, 2026-09-20) — + * the same slot treatment the job runner got: the role-override → + * slot → factory-pin chain, read FRESH per stage call (a settings PUT + * between stages lands). Absent → config pins, unchanged. */ + engineFor?: (role: ModelRole) => PiEngine; } export interface ExtractionPipelineDeps extends GraphDeps, ExtractionModels { @@ -609,6 +615,13 @@ export function callUsage(usage: Usage): CallUsage { export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline { const { chatStore, graphStore, embedder: embed } = deps; + // Stage-engine resolution (model slots wave, 2026-09-20): a direct test + // value wins; otherwise the engineFor chain (role override → slot → + // factory pin, resolved FRESH per stage call server-side) — the job + // runner's exact treatment, closing the flagged gap where extraction + // stayed config-pin-bound. + const stageEngine = (role: ModelRole, override?: PiEngine): PiEngine | undefined => + override ?? deps.engineFor?.(role); // THE SHARED REGISTRIES (runs() above — globalThis): every pipeline // instance (route entries in dev, tests) sees the same runs, the same // single-flight guard. No module-level state left in this factory. @@ -646,7 +659,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti const t0 = performance.now(); const findings = await runExtractor( - deps.extractor, + stageEngine("extractor", deps.extractor), `Transcript:\n\n${view}`, signal, (usage, model, step, finishReason, tools) => { @@ -669,7 +682,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti const t0 = performance.now(); const auditResult = await runAuditor( - deps.auditor, + stageEngine("auditor", deps.auditor), `Transcript:\n\n${view}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}\n\n` + `Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`, signal, @@ -800,7 +813,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti // clusters. Fire-and-forget — naming never fails the run. refreshClusterLabels({ graphStore, - engine: deps.labeler, + engine: stageEngine("labeler", deps.labeler), signal: runningExtractions.get(chatId)?.controller.signal, }).catch(() => {}); return stats; @@ -844,7 +857,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti console.info(`[extraction ${input.id}] notes extractor starting (${view.split("\n").length} source lines)`); const t0 = performance.now(); const findings = await runNotesExtractor( - deps.extractor, + stageEngine("extractor", deps.extractor), view, signal, (usage, model, step, finishReason, tools) => { @@ -866,7 +879,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti ); const t0 = performance.now(); const auditResult = await runNotesAuditor( - deps.auditor, + stageEngine("auditor", deps.auditor), `Source:\n\n${notesView(input)}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}\n\n` + `Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`, signal, @@ -922,7 +935,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti ); refreshClusterLabels({ graphStore, - engine: deps.labeler, + engine: stageEngine("labeler", deps.labeler), signal: runningExtractions.get(id)?.controller.signal, }).catch(() => {}); return stats; diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 651de89..07347b5 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -60,6 +60,13 @@ export const extraction = createExtractionPipeline({ // The REAL embedder (local transformer model): post-merge embedding // batch — semantic clusters are computed and persisted here. embedder: embedNewNodes, + // Model slots wave (2026-09-20): the extraction stages resolve the FULL + // chain — role override → slot → factory pin — read FRESH per stage call + // (the flagged gap: ExtractionModels deps were engine VALUES, keeping + // extractor/auditor/labeler config-pin-bound while the job runner went + // slot-native). Direct test values still win inside the pipeline. + engineFor: (role) => + piEngineForSpec(resolveRoleSpec(role, stores.settingsStore.readSettings(), MODEL_PINS)), }); /** Graph context injection (vision-context-injection.md): the runner's diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index a1e8856..1bf755f 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -401,6 +401,64 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { ); }); + it("engineFor fills the stages (model slots wave): the chain replaces config pins", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + const verdict: AuditResult = { + propositions: FINDINGS.propositions, + edges: [], // the edge is dropped with it + attribution: FINDINGS.attribution, + dropped: [{ ref: "p1→p2", reason: "endpoint dropped" }], + }; + const consulted: string[] = []; + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + // NO direct values — the stages resolve ENTIRELY through engineFor + // (the role override → slot → factory-pin chain, read fresh per + // stage call in production; here, the scripted chain). + engineFor: (role) => { + consulted.push(role); + if (role === "extractor") return submitModel("submit_findings", FINDINGS); + if (role === "auditor") return submitModel("submit_review", verdict); + return submitModel("submit_labels", { labels: [] }); + }, + }); + const stats = await pipeline.runExtraction(chatId); + assert.equal(stats.nodes, 2, "the chain-fed pipeline ran end to end — both propositions merged"); + // The stages were resolved through the chain, in pipeline order (the + // labeler is fire-and-forget garnish — extractor + auditor are the + // pinned order). + assert.equal(consulted[0], "extractor"); + assert.equal(consulted[1], "auditor"); + }); + + it("direct test values still WIN over engineFor (the priority pin)", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + const verdict: AuditResult = { + propositions: FINDINGS.propositions, + edges: [], + attribution: FINDINGS.attribution, + dropped: [{ ref: "p1→p2", reason: "endpoint dropped" }], + }; + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: submitModel("submit_findings", FINDINGS), + auditor: submitModel("submit_review", verdict), + labeler: submitModel("submit_labels", { labels: [] }), + // If the direct values did NOT win, this throw would surface. + engineFor: (role) => { + throw new Error(`engineFor should not be consulted for ${role} when a direct value is set`); + }, + }); + const stats = await pipeline.runExtraction(chatId); + assert.equal(stats.nodes, 2, "the direct-value seams ran, untouched by engineFor"); + }); + it("checkpoints nothing when a stage fails", async () => { const { chatId } = fixtureChat(); const ledgers = memoryLedgers();