sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

chunk notes like chats; both submit tools require edges/links (owner 2026-09-28)

Browse Source
This commit is contained in:
amosblomqvist
2026-09-28 21:28:39 +03:00
parent f11b1664dc
commit c17405a3d7
3 changed files with 159 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+1 -1
View File
@@ -346,7 +346,7 @@ flowchart TB
**Coverage, not quiz-counting** (supersedes the v1 every-N-quiz-answers trigger): each message's *view* (the compact line format `core/coverage.ts` has always hashed) is hashed into a per-chat fingerprint ledger (`coverage.jsonl`); "covered" means exactly "a run committed on this content." The prompt view is the richer markdown document — a deliberate 2026-09-27 split: re-formatting presentation never re-opens covered history. Uncovered = fingerprint diff, never dual-tracked; runs are greedy token-budgeted batches (`graph.extractionChunkTokens` — a settings knob, default 4096, floor 500, read fresh each turn) of the oldest uncovered messages — quiz-less sessions extract too. A coarse `sessions.jsonl` checkpoint still marks each merged run. The trigger seam (`turn-service.ts`) is trigger-agnostic from the pipeline's side: when a better mechanism lands (session end, idle, explicit flush), only the seam changes. Single-flight per source; the run registry powers the coverage UI and true abort.
**Two front doors, one pipeline**: chat transcripts (quiz-triggered) and raw personal notes (`POST /api/graph/import` — each file becomes a `NotesSource` playing the `chatId` role, deterministic id from the path hash).
**Two front doors, one pipeline**: chat transcripts (quiz-triggered) and raw personal notes (`POST /api/graph/import` — each file becomes a `NotesSource` playing the `chatId` role, deterministic id from the path hash). Notes chunk like chats (owner, 2026-09-28): large files split at line boundaries under the same token budget, each chunk a standalone extract → audit → merge under the serialized gate — no single model call carries a whole file. Both submit tools require their top-level fields (`edges`/`links` present even when empty): a payload missing them fails validation rather than silently merging structure-less nodes.
**The read channel is pure derivation.** `GET /api/graph` reads the ledgers (`graphSnapshot`: merged nodes/edges, birth-session clusters, positions) — the home view never triggers a write. Node detail cards show a node's edges — outgoing under "What follows from this" / "What this supports", incoming under "Follows from" / "Supported by" (an edge A→B means A implies or supports B, owner semantics 2026-09-28) — and its birth sessions.
src/agent/subagents/extraction.ts
+80 -40
View File
@@ -142,12 +142,10 @@ const findingsProps = {
propositions: Type.Array(Type.Object(propositionProps), {
description: "Every distinct proposition the session touched — true statements, misconception-free.",
}),
edges: Type.Optional(
Type.Array(edgeSchema, {
description:
"Relations between propositions: follows (derivable) / supports (grounds without entailing). Pass [] when there are none — never omit the field.",
}),
),
edges: Type.Array(edgeSchema, {
description:
"Relations between propositions: follows (derivable) / supports (grounds without entailing). Pass [] when there are none — never omit the field.",
}),
};
const findingsSchema = Type.Object(findingsProps);
@@ -157,24 +155,20 @@ const findingsSchema = Type.Object(findingsProps);
// edges are its entire deliverable. An order-of-magnitude smaller submit
// than the old full-findings echo.
const auditSchema = Type.Object({
links: Type.Optional(
Type.Array(
Type.Object({
id: Type.String({ description: 'The proposition id from the findings ("p1", …).' }),
linkTo: Type.String({
description:
"The id of the EXISTING graph node this proposition IS — your identity verdict. EQUIVALENCE BIAS (2026-09-27): paraphrases, rewordings, and compressions of the same claim MERGE; only genuinely distinct propositions stay apart.",
}),
links: Type.Array(
Type.Object({
id: Type.String({ description: 'The proposition id from the findings ("p1", …).' }),
linkTo: Type.String({
description:
"The id of the EXISTING graph node this proposition IS — your identity verdict. EQUIVALENCE BIAS (2026-09-27): paraphrases, rewordings, and compressions of the same claim MERGE; only genuinely distinct propositions stay apart.",
}),
{ description: "Identity verdicts — which findings propositions ARE existing graph nodes. A proposition absent from here = create a new node." },
),
),
edges: Type.Optional(
Type.Array(edgeSchema, {
description:
"BRIDGING edges ONLY — new edges welding this batch into the existing graph, with an EXISTING graph node id as one endpoint (or both) and/or a findings proposition id. NEVER restate the extractor's own edges.",
}),
{ description: "Identity verdicts — which findings propositions ARE existing graph nodes. A proposition absent from here = create a new node. Pass [] when none — never omit the field." },
),
edges: Type.Array(edgeSchema, {
description:
"BRIDGING edges ONLY — new edges welding this batch into the existing graph, with an EXISTING graph node id as one endpoint (or both) and/or a findings proposition id. NEVER restate the extractor's own edges. Pass [] when none — never omit the field.",
}),
});
// The notes front door's shapes are IDENTICAL to the session shapes now
@@ -209,19 +203,26 @@ function healStringifiedArrays(raw: unknown): unknown {
return out;
}
// ── the weak-model ??-fills (the old zod defaults, applied at submit) ──
// ── the submit seam (owner, 2026-09-28) ──
//
// Both top-level fields on both submit tools are REQUIRED — a payload
// missing `edges`/`links` fails validation instead of silently merging
// edge-less (the degraded-retry hole: the model's first truncated
// attempt failed validation, then the retry passed with propositions
// only — optional edges ??-filled [] — merging a pile of nodes with no
// structure). Only the per-item aliases stay optional with a ??-fill.
function fillFindings(raw: Static<typeof findingsSchema>): SessionFindings {
return {
propositions: raw.propositions.map((p) => ({ ...p, aliases: p.aliases ?? [] })),
edges: raw.edges ?? [],
edges: raw.edges,
};
}
function fillAudit(raw: Static<typeof auditSchema>): AuditVerdicts {
return {
links: raw.links ?? [],
edges: raw.edges ?? [],
links: raw.links,
edges: raw.edges,
};
}
@@ -231,6 +232,8 @@ report what those notes show the author engaging with, as structured findings fo
There is no session and no chat structure here: the source is raw text, the author's own material. Read it as such.
You are the ONLY stage that sees this source — you are fully trusted; report carefully.
You may be seeing ONE PART of a longer file (large files are chunked): extract exactly what THIS part holds —
the auditor links equivalent propositions across parts, so overlap worries are not yours.
The graph is a set of TRUE propositions about the world, full stop. A node is never anything about the author or the
source — no "the author understands X", no "mentioned in these notes": the node is just X. A misconception is NOT a
@@ -954,11 +957,34 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
/** The raw source view the notes stages see: the file path + the text
* VERBATIM. No message ids — provenance is the file itself. */
function notesView(input: { filePath: string; text: string }): string {
return `SOURCE (the author's personal notes): ${input.filePath}\n\n${input.text}`;
function notesView(input: { filePath: string; text: string; part?: { i: number; n: number } }): string {
const part = input.part && input.part.n > 1 ? ` (part ${input.part.i}/${input.part.n})` : "";
return `SOURCE (the author's personal notes): ${input.filePath}${part}\n\n${input.text}`;
}
async function notesExtract(input: { id: string; filePath: string; text: string }): Promise<NotesFindings> {
/** Chunk the notes text under the same budget the chat path uses
* (owner, 2026-09-28: the whole-file-in-one-call design collapsed a
* weak model — 26 propositions, structure bleed mid-emission).
* Greedy by line, chars/4 like every other estimate in the app; a
* single over-budget line rides alone (never split a line). */
function chunkNotes(text: string, budget: number): string[] {
if (text.trim().length === 0) return [];
const out: string[] = [];
let current = "";
for (const line of text.split("\n")) {
const candidate = current.length === 0 ? line : current + "\n" + line;
if (current.length > 0 && Math.ceil(candidate.length / 4) > budget) {
out.push(current);
current = line;
} else {
current = candidate;
}
}
if (current.trim().length > 0) out.push(current);
return out;
}
async function notesExtract(input: { id: string; filePath: string; text: string; part?: { i: number; n: number } }): Promise<NotesFindings> {
const signal = runningExtractions.get(input.id)?.controller.signal;
signal?.throwIfAborted(); // an abort during a sync gap ends the run here
const view = notesView(input);
@@ -990,18 +1016,32 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// (2026-09-25): same refusal as the chat front door — no
// extractor/auditor engines, or a stage key gone, no run.
refuseIfStagesUnready();
console.info(`[extraction ${id}] notes pipeline start (${input.filePath})`);
setPhase(id, "extracting", "notes", input.filePath);
const findings = await notesExtract(input);
// SERIALIZED AUDIT (owner 2026-09-28): the same gate as the chat
// path — a notes import and a chat run may extract in parallel,
// but audit + merge one at a time, process-wide.
const stats = await withAuditLock(id, "notes", input.filePath, async () => {
setPhase(id, "auditing", "notes", input.filePath);
const auditResult = await audit(id, findings);
setPhase(id, "merging", "notes", input.filePath);
return mergeFindings(id, findings, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt });
});
// CHUNKED like the chat path (owner, 2026-09-28): each chunk is a
// standalone extract → audit → merge, so no single call carries a
// whole large file. Abort mid-run keeps earlier chunks merged
// (same doctrine as chat batches).
const chunkTokens = deps.getChunkTokens?.() ?? FACTORY_SETTINGS.graph.extractionChunkTokens;
const chunks = chunkNotes(input.text, chunkTokens);
console.info(
`[extraction ${id}] notes pipeline start (${input.filePath}) — ${chunks.length} chunk(s) of ≤${chunkTokens} tok`,
);
const stats: EmissionStats = { nodes: 0, edges: 0 };
for (const [i, chunk] of chunks.entries()) {
setPhase(id, "extracting", "notes", input.filePath);
const findings = await notesExtract({ ...input, text: chunk, part: { i: i + 1, n: chunks.length } });
// SERIALIZED AUDIT (owner, 2026-09-28): the same gate as the chat
// path — a notes import and a chat run may extract in parallel,
// but audit + merge one at a time, process-wide. Reacquired per
// chunk so multi-chunk runs interleave fairly.
const merged = await withAuditLock(id, "notes", input.filePath, async () => {
setPhase(id, "auditing", "notes", input.filePath);
const auditResult = await audit(id, findings);
setPhase(id, "merging", "notes", input.filePath);
return mergeFindings(id, findings, auditResult, { chatStore, graphStore }, { foldedAt: input.importedAt });
});
stats.nodes += merged.nodes;
stats.edges += merged.edges;
}
// Same post-merge settle as the chat path.
settleGraphLayout(graphStore);
persistRunUsage(id, Math.round(performance.now() - t0), stats);
tests/extraction.test.ts
+78 -9
View File
@@ -598,7 +598,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: silentModel(),
auditor: submitModel("submit_review", FINDINGS),
auditor: submitModel("submit_review", { links: [], edges: [] }),
});
await assert.rejects(failingExtractor.runExtraction(chatId), /the extractor never submitted findings/);
assert.equal(ledgers.readCheckpoints().length, 0);
@@ -654,7 +654,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: submitModel("submit_findings", findingsA),
auditor: gatedSubmitModel("submit_review", findingsA, gate, () => {
auditor: gatedSubmitModel("submit_review", { links: [], edges: [] }, gate, () => {
aAuditing = true;
}),
});
@@ -671,7 +671,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
// THE pin: when B's auditor is finally invoked, A's merge must
// already be in the ledgers — B explores the post-merge graph,
// which is the entire point of the gate.
auditor: submitModel("submit_review", findingsB, () => {
auditor: submitModel("submit_review", { links: [], edges: [] }, () => {
bAuditorCalled = true;
bSawAlphasMerge = ledgers.readNodes().some((n) => n.name === "Alpha claim");
}),
@@ -708,7 +708,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: submitModel("submit_findings", single("pa", "Alpha claim", "Alpha statement")),
auditor: gatedSubmitModel("submit_review", single("pa", "Alpha claim", "Alpha statement"), gate),
auditor: gatedSubmitModel("submit_review", { links: [], edges: [] }, gate),
});
const runA = pipelineA.runExtraction(a.chatId);
await until(() => extractionStatuses().some((r) => r.chatId === a.chatId && r.phase === "auditing"));
@@ -718,7 +718,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: submitModel("submit_findings", single("pb", "Beta claim", "Beta statement")),
auditor: submitModel("submit_review", single("pb", "Beta claim", "Beta statement"), () => {
auditor: submitModel("submit_review", { links: [], edges: [] }, () => {
bAuditorCalled = true;
}),
});
@@ -740,7 +740,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: submitModel("submit_findings", single("pc", "Gamma claim", "Gamma statement")),
auditor: submitModel("submit_review", single("pc", "Gamma claim", "Gamma statement")),
auditor: submitModel("submit_review", { links: [], edges: [] }),
});
assert.deepEqual(await pipelineC.runExtraction(c.chatId), { nodes: 1, edges: 0 });
});
@@ -806,7 +806,7 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => {
chatStore,
graphStore: ledgers,
extractor: hangingModel(), // hangs until aborted — no step finishes
auditor: submitModel("submit_review", { propositions: [], edges: [] }),
auditor: submitModel("submit_review", { links: [], edges: [] }),
});
const runPromise = pipeline.runExtraction(chatId);
@@ -954,7 +954,7 @@ const NOTES_FINDINGS = {
} as const;
const NOTES_VERDICT = {
propositions: [NOTES_FINDINGS.propositions[0]],
links: [],
edges: [],
} as const;
@@ -999,6 +999,75 @@ describe("the notes front door (importNotes — format-agnostic)", () => {
assert.equal(ledgers.readCheckpoints()[0]!.chatId, source.id);
});
it("chunks a large file: one extract→audit→merge per chunk, part headers in each view (owner, 2026-09-28)", async () => {
const ledgers = memoryLedgers();
const views: string[] = [];
let calls = 0;
// Fresh engine PER CHUNK via the engineFor factory — same seam
// production uses (one scripted response per model call).
const extractorStep = () =>
fauxEngine([
(context) => {
calls++;
views.push(capturedView(context));
return fauxAssistantMessage(
[
fauxToolCall("submit_findings", NOTES_FINDINGS as unknown as ToolCall["arguments"], {
id: `findings-${calls}`,
}),
],
{ stopReason: "toolUse" },
);
},
]);
const text = Array.from({ length: 8 }, (_, i) => `note line ${i}: ${"x".repeat(30)}`).join("\n");
const pipeline = createExtractionPipeline({
chatStore,
graphStore: ledgers,
engineFor: (role) =>
role === "extractor"
? extractorStep()
: submitModel("submit_review", { links: [], edges: [] }),
getChunkTokens: () => 20, // 80 chars/line-groups → several chunks
});
const stats = await pipeline.importNotes({ filePath: "/vault/big.md", text });
assert.ok(calls >= 2, `expected ≥2 extractor calls, got ${calls}`);
assert.match(views[0]!, new RegExp(`part 1/${calls}`));
assert.match(views.at(-1)!, new RegExp(`part ${calls}/${calls}`));
// Same findings in every chunk → ONE node total (mapping dedupe
// across chunks), stats never double-count.
assert.deepEqual(stats, { nodes: 1, edges: 0 });
assert.equal(ledgers.readNodes().length, 1);
});
it("required top-level fields (owner, 2026-09-28): a findings submit MISSING edges fails validation — no silent edge-less merge", async () => {
const { chatId } = fixtureChat();
const ledgers = memoryLedgers();
// The degraded-retry shape from the OpenRouter log: propositions
// only, edges omitted. Required schema → validation fails on every
// turn → the run rejects instead of merging a pile of structureless
// nodes. Four queued attempts exhaust the extractor's turn budget.
// Four queued attempts — one per turn of the extractor's budget.
const noEdges = () =>
fauxEngine(
Array.from({ length: 4 }, () => (context) => {
void context;
return fauxAssistantMessage(
[fauxToolCall("submit_findings", { propositions: FINDINGS.propositions } as unknown as ToolCall["arguments"])],
{ stopReason: "toolUse" },
);
}),
);
const pipeline = createExtractionPipeline({
chatStore,
graphStore: ledgers,
engineFor: (role) => (role === "extractor" ? noEdges() : submitModel("submit_review", { links: [], edges: [] })),
});
await assert.rejects(pipeline.runExtraction(chatId), /never submitted findings/);
assert.equal(ledgers.readNodes().length, 0);
});
it("re-importing the same file is idempotent: same source identity, refs dedupe, usage history grows", async () => {
const ledgers = memoryLedgers();
const mk = () =>
@@ -1270,7 +1339,7 @@ describe("the session-deletion guard (chat deleted mid-run)", () => {
chatStore,
graphStore: ledgers,
extractor: submitModel("submit_findings", FINDINGS),
auditor: submitModel("submit_review", { propositions: FINDINGS.propositions, edges: [] }, () => chatStore.deleteChat(chatId)),
auditor: submitModel("submit_review", { links: [], edges: [] }, () => chatStore.deleteChat(chatId)),
});
const stats = await pipeline.runExtraction(chatId);