feat(extraction): stage engines resolve the slot chain — engineFor on the pipeline
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.
This commit is contained in:
@@ -556,11 +556,17 @@ export function abortExtraction(chatId: string): boolean {
|
|||||||
// ─── the pipeline ───────────────────────────────────────────────────────────
|
// ─── the pipeline ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
export interface ExtractionModels {
|
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;
|
extractor?: PiEngine;
|
||||||
auditor?: PiEngine;
|
auditor?: PiEngine;
|
||||||
/** Test seam for the cluster namer (post-extraction garnish). */
|
/** Test seam for the cluster namer (post-extraction garnish). */
|
||||||
labeler?: PiEngine;
|
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 {
|
export interface ExtractionPipelineDeps extends GraphDeps, ExtractionModels {
|
||||||
@@ -609,6 +615,13 @@ export function callUsage(usage: Usage): CallUsage {
|
|||||||
|
|
||||||
export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
|
export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
|
||||||
const { chatStore, graphStore, embedder: embed } = deps;
|
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
|
// THE SHARED REGISTRIES (runs() above — globalThis): every pipeline
|
||||||
// instance (route entries in dev, tests) sees the same runs, the same
|
// instance (route entries in dev, tests) sees the same runs, the same
|
||||||
// single-flight guard. No module-level state left in this factory.
|
// 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 t0 = performance.now();
|
||||||
|
|
||||||
const findings = await runExtractor(
|
const findings = await runExtractor(
|
||||||
deps.extractor,
|
stageEngine("extractor", deps.extractor),
|
||||||
`Transcript:\n\n${view}`,
|
`Transcript:\n\n${view}`,
|
||||||
signal,
|
signal,
|
||||||
(usage, model, step, finishReason, tools) => {
|
(usage, model, step, finishReason, tools) => {
|
||||||
@@ -669,7 +682,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
|||||||
const t0 = performance.now();
|
const t0 = performance.now();
|
||||||
|
|
||||||
const auditResult = await runAuditor(
|
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` +
|
`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)}`,
|
`Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`,
|
||||||
signal,
|
signal,
|
||||||
@@ -800,7 +813,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
|||||||
// clusters. Fire-and-forget — naming never fails the run.
|
// clusters. Fire-and-forget — naming never fails the run.
|
||||||
refreshClusterLabels({
|
refreshClusterLabels({
|
||||||
graphStore,
|
graphStore,
|
||||||
engine: deps.labeler,
|
engine: stageEngine("labeler", deps.labeler),
|
||||||
signal: runningExtractions.get(chatId)?.controller.signal,
|
signal: runningExtractions.get(chatId)?.controller.signal,
|
||||||
}).catch(() => {});
|
}).catch(() => {});
|
||||||
return stats;
|
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)`);
|
console.info(`[extraction ${input.id}] notes extractor starting (${view.split("\n").length} source lines)`);
|
||||||
const t0 = performance.now();
|
const t0 = performance.now();
|
||||||
const findings = await runNotesExtractor(
|
const findings = await runNotesExtractor(
|
||||||
deps.extractor,
|
stageEngine("extractor", deps.extractor),
|
||||||
view,
|
view,
|
||||||
signal,
|
signal,
|
||||||
(usage, model, step, finishReason, tools) => {
|
(usage, model, step, finishReason, tools) => {
|
||||||
@@ -866,7 +879,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
|||||||
);
|
);
|
||||||
const t0 = performance.now();
|
const t0 = performance.now();
|
||||||
const auditResult = await runNotesAuditor(
|
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` +
|
`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)}`,
|
`Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`,
|
||||||
signal,
|
signal,
|
||||||
@@ -922,7 +935,7 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
|
|||||||
);
|
);
|
||||||
refreshClusterLabels({
|
refreshClusterLabels({
|
||||||
graphStore,
|
graphStore,
|
||||||
engine: deps.labeler,
|
engine: stageEngine("labeler", deps.labeler),
|
||||||
signal: runningExtractions.get(id)?.controller.signal,
|
signal: runningExtractions.get(id)?.controller.signal,
|
||||||
}).catch(() => {});
|
}).catch(() => {});
|
||||||
return stats;
|
return stats;
|
||||||
|
|||||||
@@ -60,6 +60,13 @@ export const extraction = createExtractionPipeline({
|
|||||||
// The REAL embedder (local transformer model): post-merge embedding
|
// The REAL embedder (local transformer model): post-merge embedding
|
||||||
// batch — semantic clusters are computed and persisted here.
|
// batch — semantic clusters are computed and persisted here.
|
||||||
embedder: embedNewNodes,
|
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
|
/** Graph context injection (vision-context-injection.md): the runner's
|
||||||
|
|||||||
@@ -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 () => {
|
it("checkpoints nothing when a stage fails", async () => {
|
||||||
const { chatId } = fixtureChat();
|
const { chatId } = fixtureChat();
|
||||||
const ledgers = memoryLedgers();
|
const ledgers = memoryLedgers();
|
||||||
|
|||||||
Reference in New Issue
Block a user