diff --git a/AGENTS.md b/AGENTS.md index db4c707..9f19bf1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,6 +84,15 @@ Use `knowledgeReleaseReport()` before promotion. It folds the candidate and base ## Integration Boundaries - Use `KbStore` for storage. Applications may provide any durable backend that implements it. +- Use `new FileSystemKbStore({ root })` when opening a knowledge-base root; it keeps every record under `/.agent-knowledge/` — index, event log, and per-run claim ledgers. + The string constructor retains the published direct-directory behavior. + A legacy string that names `/.agent-knowledge` is canonicalized to the same root lock, so the two constructor forms cannot race on one file under different locks. + The store is the single writer of `index.json`; `writeKnowledgeIndex` goes through it. + Do not add a second writer for a record this store owns. +- Research state is durable state. A driver that accumulates belief across rounds takes a store and a `ledgerId` (`createPersistentResearchDrivingDriver`) so corroboration counts, contradiction edges, and open questions survive the process. `runVerifiedResearchLoop` durably announces a fold before its synchronous question generation, calls `driver.checkpoint()` before publishing the round event, and reconstructs an interrupted fold on resume. +- `TrackedClaim` remains the live Set-based driver API; `ResearchClaimRecord` is its sorted-array durable form. Convert at the persistence boundary rather than changing the published live shape. +- More than one writer per ledger means `mergeClaimLedger(id, merge)`, never `putClaimLedger`. `putClaimLedger` writes the whole record, so two writers accumulating into one ledger each write what they built from a stale read and the later write erases the earlier writer's claims. `mergeClaimLedger` holds the store's lock across read, merge, and write; `mergeClaimLedgers` is the combining rule and is commutative, associative, and idempotent, so replay and arrival order cannot change the result. +- Use `writeFileDurable` / `writeJsonDurableWithinRoot` from the entrypoint for any file that must survive a crash. They are atomic, fsynced, and symlink-safe; a hand-rolled `writeFile` is none of those. - Use `KnowledgeDiscoveryDispatcher` for research workers. Applications should connect it to their own runtime. - Do not bypass `lint` or `validate` before using generated knowledge in an agent. diff --git a/docs/architecture.md b/docs/architecture.md index 44feb7b..b78c1e9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -28,6 +28,36 @@ Product apps own domain policies, provider accounts, vector stores, source adapt Core does not own a D1 schema or fleet dispatcher. Apps wire `KbStore` and `KnowledgeDiscoveryDispatcher` to their tenancy, queue, budget, auth, and sandbox systems. +## On-disk layout + +`new FileSystemKbStore({ root })` is the explicit knowledge-base-root form and owns everything under `/.agent-knowledge/`. +The published `new FileSystemKbStore(directory)` form remains a direct record directory, so upgrading does not silently move an existing store. +When that string is the canonical `/.agent-knowledge` directory, both forms use the root's one mutation lock; retaining the path must not create a second lock for the same files. + +| Path | Record | +| --- | --- | +| `.agent-knowledge/index.json` | the built knowledge index (`writeKnowledgeIndex` writes it through this store) | +| `.agent-knowledge/events.json` | the knowledge event log, including one `research.iteration` per research-loop round | +| `.agent-knowledge/claim-ledgers/.json` | one research run's claim ledger — corroboration counts, contradiction edges, open deep questions | +| `.agent-knowledge/sources.json` | the immutable source registry | +| `.agent-knowledge/mutation.lock.durable`, `mutation-epoch.json`, `file-transactions/` | the cross-process mutation lock and its crash-recovery state | + +The root is also the directory `withKnowledgeMutation` locks, so every record above is written under one lock and one epoch. +There is exactly one writer per file: a second index writer alongside this one is a defect, not a variation. + +A claim ledger is the one record several writers legitimately share — a resumed run beside a live one, or several workers researching one goal in parallel. +They reach it through `mergeClaimLedger(id, merge)`, which holds the mutation lock across the read, the merge, and the write, so no writer can build its record from a value another writer has already replaced. +`putClaimLedger` writes the whole record and is correct only for a single writer. +The combining rule is `mergeClaimLedgers`: support and contradiction edges union, `contested` and `addressed` latch on, `firstSeenRound` moves earlier, and every collection is sorted — so the merge is commutative, associative, and idempotent, and the bytes on disk depend on the evidence rather than on scheduling. +Ledgers for two different goals refuse to merge (`ClaimLedgerGoalConflictError`) rather than pooling unrelated evidence into one corroboration count. +The live driver exposes the published Set-based `TrackedClaim`; the ledger stores a separate `ResearchClaimRecord` with sorted arrays so JSON serialization cannot erase those sets. +Source verification first persists a `ResearchClaimEvidence` observation that cannot affect claim support or completion, then `runVerifiedResearchLoop` calls `commitSources` only after the source registry write succeeds. +The ledger unions those observations with exact confirmed original source URIs and materializes only their intersection, so a crash on either side resumes safely without treating an absent source as evidence or losing a registered source's claim. +Before synchronous question generation, the persistent driver records `preparedRounds`; a resume reconstructs and checkpoints any prepared round whose questions were interrupted, and the loop publishes its `research.iteration` event only after that checkpoint succeeds. + +Every write in this layer goes through `durable-fs` (`writeFileDurable`, `writeJsonDurableWithinRoot`) — temp file, fsync, atomic rename, fsync parent, through `O_NOFOLLOW` descriptors anchored via `/proc/self/fd` so a directory swapped for a symlink mid-write cannot redirect it outside the root. +These are exported from the package entrypoint; consumers that keep their own journals should use them rather than reimplement them. + ## Runtime Loop 1. Normalize sources into immutable source records. diff --git a/src/claim-ledger.ts b/src/claim-ledger.ts new file mode 100644 index 0000000..3b6969d --- /dev/null +++ b/src/claim-ledger.ts @@ -0,0 +1,526 @@ +/** + * The algebra of a research claim ledger: claim identity, and how two ledgers + * that accumulated evidence for the same goal combine into one. + * + * This lives apart from `research-driving-driver.ts` because it is no longer + * that driver's private business. A ledger is a durable record now, and a + * durable record addressed by id is a record two writers can reach: two rounds + * of one run resuming from disk, or two workers researching one goal in + * parallel. `putClaimLedger` writes the whole record, so the second writer's + * write erases the first writer's claims — the ledger persists and the + * knowledge still does not compound. Combining is therefore part of what the + * record MEANS, and it belongs next to the record rather than inside one + * consumer of it. + * + * Every combination here is monotone: support only grows, contradiction edges + * only accumulate, `contested` only latches on, `firstSeenRound` only moves + * earlier. That is what makes it safe to apply twice — a retried write cannot + * produce a different ledger than a single write did. + */ + +import { sha256 } from './ids' +import type { + DeepQuestion, + ResearchClaimEvidence, + ResearchClaimLedger, + ResearchClaimRecord, +} from './types' + +/** + * Claim identity = sha256 of the normalized claim text, so the same assertion + * discovered independently by two workers is ONE claim with two supporting + * sources rather than two claims with one each — which is the difference + * between corroborated and unsupported. + */ +export function claimId(text: string): string { + return `c_${sha256(normalizeClaimText(text)).slice(0, 16)}` +} + +/** Stable identity for one extracted claim/source/contradiction observation. */ +export function claimEvidenceId( + claim: Pick, +): string { + return `e_${sha256( + JSON.stringify([claim.claimId, claim.sourceUri, claim.contradictsClaimId ?? null]), + ).slice(0, 16)}` +} + +/** Case-, whitespace-, and stylistic-punctuation-insensitive claim identity form. */ +export function normalizeClaimText(text: string): string { + return ( + text + .normalize('NFKC') + .toLowerCase() + // Direction and polarity change scientific meaning. Preserve them as word + // tokens before removing stylistic punctuation, so `x > y` cannot merge + // with `x < y`, and `+5%` cannot corroborate `-5%`. + .replace(/<=|≤/gu, ' symbol_less_than_or_equal ') + .replace(/>=|≥/gu, ' symbol_greater_than_or_equal ') + .replace(/!=|≠/gu, ' symbol_not_equal ') + .replace(/==|=/gu, ' symbol_equal ') + .replace(/±/gu, ' symbol_plus_or_minus ') + .replace(/[≈~]/gu, ' symbol_approximately ') + .replace(/<|←|⇐/gu, ' symbol_less_or_left ') + .replace(/>|→|⇒/gu, ' symbol_greater_or_right ') + .replace(/[+]/gu, ' symbol_plus_or_positive ') + .replace(/[-−]/gu, ' symbol_minus_or_negative ') + .replace(/[^\p{L}\p{N}\s]+/gu, ' ') + .replace(/\s+/g, ' ') + .trim() + ) +} + +/** + * The canonical host a source uri counts as, which is what makes two sources + * INDEPENDENT: corroboration is "distinct hosts", so this function is the rule + * that decides whether a claim is confirmed or merely repeated. Exported so a + * consumer building a `ResearchClaimRecord` cannot answer it a different way — a + * consumer that counted raw uris would report two pages of one site as + * independent confirmation. + */ +export function claimSourceHost(uri: string): string { + try { + return new URL(uri.trim()).hostname.toLowerCase().replace(/^www\./, '') + } catch { + // Non-URL identifier (offline corpus uris like `web/foo`): canonicalize so + // distinct identifiers still count as distinct independent sources. + return uri.trim().toLowerCase() + } +} + +/** Stable identity for one deep question. */ +export function deepQuestionId(kind: DeepQuestion['kind'], text: string): string { + return `q_${sha256(`${kind}:${text}`).slice(0, 16)}` +} + +/** + * Refuse a claim record whose identity or source count disagrees with its evidence. + * + * `supportingHosts` is used as the independent-source count, so accepting hosts + * that cannot be derived from `supportingUris` would let a malformed record + * manufacture corroboration. Canonical ordering also makes equal records have + * equal bytes regardless of which process assembled them. + */ +export function assertTrackedClaimIntegrity(claim: ResearchClaimRecord): void { + if (claim.id !== claimId(claim.text)) { + throw new Error(`claim '${claim.id}' does not match its text-derived identity`) + } + if (claim.text !== claim.text.trim()) { + throw new Error(`claim '${claim.id}' text must not have surrounding whitespace`) + } + assertSortedUnique(`claim '${claim.id}' supportingUris`, claim.supportingUris) + assertSortedUnique(`claim '${claim.id}' supportingHosts`, claim.supportingHosts) + assertSortedUnique(`claim '${claim.id}' contradicts`, claim.contradicts) + const expectedHosts = [ + ...new Set(claim.supportingUris.map(claimSourceHost).filter(Boolean)), + ].sort() + if (!sameStrings(claim.supportingHosts, expectedHosts)) { + throw new Error( + `claim '${claim.id}' supportingHosts must equal the hosts derived from supportingUris`, + ) + } + if (claim.contradicts.includes(claim.id)) { + throw new Error(`claim '${claim.id}' cannot contradict itself`) + } + if (claim.contradicts.length > 0 && !claim.contested) { + throw new Error(`claim '${claim.id}' with a contradiction must be contested`) + } +} + +/** Refuse a deep question whose stable identity or set fields are malformed. */ +export function assertDeepQuestionIntegrity(question: DeepQuestion): void { + if (question.id !== deepQuestionId(question.kind, question.text)) { + throw new Error(`question '${question.id}' does not match its kind-and-text identity`) + } + if (question.text !== question.text.trim()) { + throw new Error(`question '${question.id}' text must not have surrounding whitespace`) + } + assertSortedUnique(`question '${question.id}' claimIds`, question.claimIds) +} + +/** Refuse an extracted observation whose identity or content is malformed. */ +export function assertResearchClaimEvidenceIntegrity(evidence: ResearchClaimEvidence): void { + if (evidence.claimId !== claimId(evidence.text)) { + throw new Error( + `claim evidence '${evidence.id}' does not match its text-derived claim identity`, + ) + } + if (evidence.id !== claimEvidenceId(evidence)) { + throw new Error(`claim evidence '${evidence.id}' does not match its content-derived identity`) + } + if (evidence.text !== evidence.text.trim()) { + throw new Error(`claim evidence '${evidence.id}' text must not have surrounding whitespace`) + } + if (evidence.contradictsClaimId === evidence.claimId) { + throw new Error(`claim evidence '${evidence.id}' cannot contradict its own claim`) + } +} + +/** Refuse a ledger that is not one canonical, internally consistent record. */ +export function assertResearchClaimLedgerIntegrity(ledger: ResearchClaimLedger): void { + if (ledger.preparedRounds !== undefined && ledger.preparedRounds <= ledger.rounds) { + throw new Error( + `claim ledger '${ledger.id}' preparedRounds must be greater than completed rounds`, + ) + } + assertSortedUnique( + `claim ledger '${ledger.id}' claimEvidence`, + ledger.claimEvidence.map((evidence) => evidence.id), + ) + assertSortedUnique( + `claim ledger '${ledger.id}' registeredSourceUris`, + ledger.registeredSourceUris, + ) + assertSortedUnique( + `claim ledger '${ledger.id}' claims`, + ledger.claims.map((claim) => claim.id), + ) + assertSortedUnique( + `claim ledger '${ledger.id}' questions`, + ledger.questions.map((question) => question.id), + ) + for (const evidence of ledger.claimEvidence) assertResearchClaimEvidenceIntegrity(evidence) + const registeredSourceUris = new Set(ledger.registeredSourceUris) + for (const claim of ledger.claims) { + assertTrackedClaimIntegrity(claim) + for (const sourceUri of claim.supportingUris) { + if (!registeredSourceUris.has(sourceUri)) { + throw new Error( + `claim '${claim.id}' counts source '${sourceUri}' before its registration is confirmed`, + ) + } + } + } + const claimsById = new Map(ledger.claims.map((claim) => [claim.id, claim])) + const claimIds = new Set(claimsById.keys()) + for (const evidence of ledger.claimEvidence) { + if (!registeredSourceUris.has(evidence.sourceUri)) continue + const claim = claimsById.get(evidence.claimId) + if (!claim?.supportingUris.includes(evidence.sourceUri)) { + throw new Error( + `registered claim evidence '${evidence.id}' must be materialized in its claim`, + ) + } + if ( + evidence.contradictsClaimId !== undefined && + claimsById.has(evidence.contradictsClaimId) && + !claim.contradicts.includes(evidence.contradictsClaimId) + ) { + throw new Error( + `registered claim evidence '${evidence.id}' must materialize its contradiction`, + ) + } + } + for (const question of ledger.questions) { + assertDeepQuestionIntegrity(question) + for (const claimId of question.claimIds) { + if (!claimIds.has(claimId)) { + throw new Error( + `question '${question.id}' references claim '${claimId}' outside its ledger`, + ) + } + } + } +} + +/** A ledger with nothing in it yet. */ +export function emptyClaimLedger(id: string, goal?: string): ResearchClaimLedger { + return { + id, + ...(goal === undefined ? {} : { goal }), + updatedAt: new Date(0).toISOString(), + rounds: 0, + claimEvidence: [], + registeredSourceUris: [], + claims: [], + questions: [], + } +} + +/** + * Turn only evidence backed by a confirmed source registration into support. + * + * This is a monotone closure: it never removes claims or evidence, and running + * it twice is a no-op. Keeping it in the ledger algebra means a source-confirming + * writer and an evidence-producing writer can arrive in either order. + */ +export function materializeRegisteredClaimEvidence( + ledger: ResearchClaimLedger, +): ResearchClaimLedger { + const registered = new Set(ledger.registeredSourceUris) + const claims = new Map( + ledger.claims.map((claim) => [claim.id, claim]), + ) + for (const evidence of ledger.claimEvidence) { + if (!registered.has(evidence.sourceUri)) continue + const host = claimSourceHost(evidence.sourceUri) + const observed: ResearchClaimRecord = { + id: evidence.claimId, + text: evidence.text, + supportingHosts: host ? [host] : [], + supportingUris: [evidence.sourceUri], + contradicts: [], + contested: false, + firstSeenRound: evidence.firstSeenRound, + } + const existing = claims.get(observed.id) + claims.set(observed.id, existing ? mergeTrackedClaims(existing, observed) : observed) + } + // A contradiction is active only when BOTH claims have registered support. + // Keep the observation while its counterpart is pending, then close the edge + // automatically when that counterpart is materialized by a later merge. + for (const evidence of ledger.claimEvidence) { + const otherId = evidence.contradictsClaimId + if (!registered.has(evidence.sourceUri) || !otherId || !claims.has(otherId)) continue + const claim = claims.get(evidence.claimId) + if (!claim) continue + claims.set(claim.id, { + ...claim, + contradicts: union(claim.contradicts, [otherId]), + contested: true, + }) + } + return linkClaimContradictions({ + ...ledger, + claims: [...claims.values()].sort((left, right) => left.id.localeCompare(right.id)), + }) +} + +/** + * Raised when two ledgers that accumulated evidence for DIFFERENT goals are + * combined. Merging them would pool two questions' evidence into one + * corroboration count, which reports a claim as independently confirmed when + * nobody confirmed it — strictly worse than losing the ledger, so this refuses. + */ +export class ClaimLedgerGoalConflictError extends Error { + constructor( + readonly ledgerId: string, + readonly existingGoal: string, + readonly incomingGoal: string, + ) { + super( + `claim ledger '${ledgerId}' accumulated evidence for goal '${existingGoal}' ` + + `and cannot be merged with evidence for '${incomingGoal}'`, + ) + this.name = 'ClaimLedgerGoalConflictError' + } +} + +/** + * Combine two records of the same claim. + * + * Union on every support collection, because a claim asserted by hosts {a} in + * one writer and {b} in another is asserted by two independent hosts and the + * whole completion oracle turns on that count. `contested` is OR — one writer + * seeing a contradiction is enough for the claim to be contested, and no later + * writer that simply did not see it may clear the flag. + */ +export function mergeTrackedClaims( + base: ResearchClaimRecord, + incoming: ResearchClaimRecord, +): ResearchClaimRecord { + assertTrackedClaimIntegrity(base) + assertTrackedClaimIntegrity(incoming) + if (base.id !== incoming.id) { + throw new Error(`cannot merge claim '${base.id}' with a different claim '${incoming.id}'`) + } + const text = + incoming.firstSeenRound < base.firstSeenRound + ? incoming.text + : incoming.firstSeenRound > base.firstSeenRound + ? base.text + : incoming.text < base.text + ? incoming.text + : base.text + return { + id: base.id, + // The earlier-seen text wins, so the claim's wording is stable across + // merges rather than flipping with whichever writer wrote last. Equal + // rounds use a lexical tie-break, preserving commutativity too. + text, + supportingHosts: union(base.supportingHosts, incoming.supportingHosts), + supportingUris: union(base.supportingUris, incoming.supportingUris), + contradicts: union(base.contradicts, incoming.contradicts), + contested: base.contested || incoming.contested, + firstSeenRound: Math.min(base.firstSeenRound, incoming.firstSeenRound), + } +} + +/** + * Combine two ledgers for the same run. + * + * `addressed` on a question is OR for the same reason `contested` is: a writer + * that answered a question has answered it, and a writer that never saw the + * answer must not reopen it. Everything else is a union or an extreme, so this + * is associative and idempotent — merge order cannot change the result and a + * replayed merge is a no-op. + */ +export function mergeClaimLedgers( + base: ResearchClaimLedger, + incoming: ResearchClaimLedger, +): ResearchClaimLedger { + assertResearchClaimLedgerIntegrity(base) + assertResearchClaimLedgerIntegrity(incoming) + if (base.id !== incoming.id) { + throw new Error( + `cannot merge claim ledger '${base.id}' with a different ledger '${incoming.id}'`, + ) + } + if (base.goal !== undefined && incoming.goal !== undefined && base.goal !== incoming.goal) { + throw new ClaimLedgerGoalConflictError(base.id, base.goal, incoming.goal) + } + const goal = base.goal ?? incoming.goal + + const claimEvidence = new Map( + base.claimEvidence.map((evidence) => [evidence.id, evidence]), + ) + for (const evidence of incoming.claimEvidence) { + const existing = claimEvidence.get(evidence.id) + claimEvidence.set(evidence.id, existing ? mergeClaimEvidence(existing, evidence) : evidence) + } + + const claims = new Map(base.claims.map((claim) => [claim.id, claim])) + for (const claim of incoming.claims) { + const existing = claims.get(claim.id) + claims.set(claim.id, existing ? mergeTrackedClaims(existing, claim) : claim) + } + + const questions = new Map( + base.questions.map((question) => [question.id, question]), + ) + for (const question of incoming.questions) { + const existing = questions.get(question.id) + if (existing && (existing.kind !== question.kind || existing.text !== question.text)) { + throw new Error(`question '${question.id}' has conflicting immutable content`) + } + questions.set( + question.id, + existing + ? { + ...existing, + claimIds: union(existing.claimIds, question.claimIds), + addressed: existing.addressed || question.addressed, + raisedRound: Math.min(existing.raisedRound, question.raisedRound), + } + : question, + ) + } + + const rounds = Math.max(base.rounds, incoming.rounds) + const preparedRounds = Math.max( + base.preparedRounds ?? base.rounds, + incoming.preparedRounds ?? incoming.rounds, + ) + return materializeRegisteredClaimEvidence({ + id: base.id, + ...(goal === undefined ? {} : { goal }), + updatedAt: + incoming.updatedAt.localeCompare(base.updatedAt) > 0 ? incoming.updatedAt : base.updatedAt, + rounds, + ...(preparedRounds > rounds ? { preparedRounds } : {}), + claimEvidence: [...claimEvidence.values()].sort((left, right) => + left.id.localeCompare(right.id), + ), + registeredSourceUris: union(base.registeredSourceUris, incoming.registeredSourceUris), + // Sorted by id so the bytes on disk depend on the ledger's content and not + // on the order two writers happened to arrive in. + claims: [...claims.values()].sort((a, b) => a.id.localeCompare(b.id)), + questions: [...questions.values()].sort((a, b) => a.id.localeCompare(b.id)), + }) +} + +function mergeClaimEvidence( + base: ResearchClaimEvidence, + incoming: ResearchClaimEvidence, +): ResearchClaimEvidence { + assertResearchClaimEvidenceIntegrity(base) + assertResearchClaimEvidenceIntegrity(incoming) + if ( + base.id !== incoming.id || + base.claimId !== incoming.claimId || + base.sourceUri !== incoming.sourceUri || + base.contradictsClaimId !== incoming.contradictsClaimId + ) { + throw new Error(`claim evidence '${base.id}' has conflicting immutable content`) + } + const text = + incoming.firstSeenRound < base.firstSeenRound + ? incoming.text + : incoming.firstSeenRound > base.firstSeenRound + ? base.text + : incoming.text < base.text + ? incoming.text + : base.text + return { + ...base, + text, + firstSeenRound: Math.min(base.firstSeenRound, incoming.firstSeenRound), + } +} + +/** + * Make every contradiction edge symmetric and mark both ends contested. + * + * A contradiction is a property of a PAIR, and a writer only ever sees one side + * of it: the worker that found the refuting source records "X contradicts Y" and + * knows nothing about Y's record. Left one-sided, Y reads as an uncontested + * claim, and the completion oracle would settle a question two sources disagree + * about. `createResearchDrivingDriver` does this pairwise as it records; this is + * the same rule stated over a whole ledger, for writers that assemble one from + * events rather than from a live loop. + * + * Idempotent and monotone like every other rule here: edges only appear and + * `contested` only latches on, so applying it twice changes nothing. An edge + * pointing at a claim this ledger does not hold is KEPT — the other side may + * arrive from another writer later, and discarding evidence of disagreement + * because the counterpart has not shown up yet is the failure this prevents. + */ +export function linkClaimContradictions(ledger: ResearchClaimLedger): ResearchClaimLedger { + const inbound = new Map() + for (const claim of ledger.claims) { + for (const other of claim.contradicts) { + if (other === claim.id) continue + const edges = inbound.get(other) + if (edges) edges.push(claim.id) + else inbound.set(other, [claim.id]) + } + } + return { + ...ledger, + claims: ledger.claims + .map((claim) => { + const contradicts = union( + claim.contradicts.filter((other) => other !== claim.id), + inbound.get(claim.id) ?? [], + ) + return { ...claim, contradicts, contested: claim.contested || contradicts.length > 0 } + }) + .sort((a, b) => a.id.localeCompare(b.id)), + } +} + +/** + * Set union, SORTED. + * + * Sorted because these collections are sets and merging must be commutative: + * arrival order is not part of what the ledger says, so two writers arriving in + * either order have to produce identical bytes. Preserving first-seen order + * instead made `merge(a, b)` and `merge(b, a)` differ, which the order- + * independence test caught — and a non-commutative merge under a filesystem + * lock means the record depends on scheduling. + */ +function union(base: readonly string[], incoming: readonly string[]): string[] { + return [...new Set([...base, ...incoming])].sort() +} + +function assertSortedUnique(label: string, values: readonly string[]): void { + for (let index = 1; index < values.length; index += 1) { + if ((values[index - 1] ?? '') >= (values[index] ?? '')) { + throw new Error(`${label} must be sorted and contain no duplicates`) + } + } +} + +function sameStrings(left: readonly string[], right: readonly string[]): boolean { + return left.length === right.length && left.every((value, index) => value === right[index]) +} diff --git a/src/index.ts b/src/index.ts index 580666b..962c0a4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,8 +5,13 @@ export * from './benchmarks/index' export * from './changes' export * from './chunking' export * from './claim-grounding' +export * from './claim-ledger' export * from './collection-research-driver' export * from './discovery' +// Atomic, fsync-durable, symlink-safe writes. Exported because every consumer +// that keeps a journal needs a write that survives `kill -9`, and the only +// alternative to reaching this is reimplementing it worse. +export * from './durable-fs' export * from './eval-readiness' export * from './events' export * from './filesystem-search-provider' diff --git a/src/indexer.ts b/src/indexer.ts index 1a0e1ab..cacb045 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -1,5 +1,5 @@ -import { writeJsonDurableWithinRoot } from './durable-fs' import { buildKnowledgeGraph } from './graph' +import { FileSystemKbStore } from './kb-store' import { withKnowledgeMutation, withKnowledgeRead } from './mutation-lock' import { loadSourceRegistry } from './sources' import { loadKnowledgePages } from './store' @@ -24,10 +24,19 @@ async function buildKnowledgeIndexUnlocked(root: string): Promise { return withKnowledgeMutation(root, async () => { const index = await buildKnowledgeIndexUnlocked(root) - await writeJsonDurableWithinRoot(root, '.agent-knowledge/index.json', index) + await new FileSystemKbStore({ root }).putIndex(index) return index }) } diff --git a/src/kb-store.ts b/src/kb-store.ts index 0ae813d..a6a296c 100644 --- a/src/kb-store.ts +++ b/src/kb-store.ts @@ -1,5 +1,11 @@ +import { basename, dirname, resolve } from 'node:path' import { z } from 'zod' -import { readRegularFileWithinRoot, writeJsonDurableWithinRoot } from './durable-fs' +import { + isMissingFile, + listRegularFilesWithinRoot, + readRegularFileWithinRoot, + writeJsonDurableWithinRoot, +} from './durable-fs' import type { KnowledgeEventQuery } from './events' import { buildKnowledgeGraph } from './graph' import { withKnowledgeMutation, withKnowledgeRead } from './mutation-lock' @@ -7,9 +13,48 @@ import { KnowledgeEventSchema, KnowledgeIndexSchema, KnowledgePageSchema, + ResearchClaimLedgerSchema, SourceRecordSchema, } from './schemas' -import type { KnowledgeEvent, KnowledgeIndex, KnowledgePage, SourceRecord } from './types' +import type { + KnowledgeEvent, + KnowledgeIndex, + KnowledgePage, + ResearchClaimLedger, + SourceRecord, +} from './types' + +/** + * Where a filesystem knowledge base keeps its machine-written records, relative + * to the knowledge-base root. + * + * This is the ONE location. `FileSystemKbStore` used to write `/index.json` + * while `writeKnowledgeIndex` wrote `/.agent-knowledge/index.json` — two + * index writers, two files, and only the second one reachable, so a store that + * had just been written to reported an empty knowledge base. Both now go through + * `FileSystemKbStore`, anchored on the knowledge-base root, which is also the + * directory `withKnowledgeMutation` locks: one file, one lock domain. + */ +export const KB_STORE_DIR = '.agent-knowledge' +export const KB_INDEX_PATH = `${KB_STORE_DIR}/index.json` +export const KB_EVENTS_PATH = `${KB_STORE_DIR}/events.json` +export const KB_CLAIM_LEDGER_DIR = `${KB_STORE_DIR}/claim-ledgers` + +/** + * A claim-ledger id is used as a filename, so it is restricted to one safe path + * segment. Rejecting rather than sanitising is deliberate: a sanitised id maps + * two different runs onto one file and silently merges their belief state. + */ +const LEDGER_ID_PATTERN = /^[A-Za-z0-9_-][A-Za-z0-9._-]*$/ + +export function assertClaimLedgerId(id: string): string { + if (!LEDGER_ID_PATTERN.test(id) || id === '.' || id === '..') { + throw new Error( + `claim ledger id must match ${LEDGER_ID_PATTERN} and cannot be a path segment: ${id}`, + ) + } + return id +} export interface KbStore { putSource(source: SourceRecord): Promise @@ -24,10 +69,42 @@ export interface KbStore { listEvents(query?: KnowledgeEventQuery): Promise } -export class MemoryKbStore implements KbStore { +/** Optional durable capability for stores that retain per-run research belief. */ +export interface ClaimLedgerStore { + /** + * Persist one research run's claim ledger — the corroboration counts, + * contradiction edges, and open deep questions that make up its belief state. + * Addressed by `ledger.id` so two runs against one knowledge base cannot + * overwrite each other. + */ + putClaimLedger(ledger: ResearchClaimLedger): Promise + getClaimLedger(id: string): Promise + listClaimLedgers(): Promise + /** + * Read one ledger, hand it to `merge`, and write what comes back — with the + * store's exclusive lock held across all three steps. + * + * This is the call a concurrent writer must use, and `putClaimLedger` is the + * one it must not. `putClaimLedger` writes the whole record, so two writers + * accumulating into one ledger each write a record built from what they read + * before the other wrote, and the later write erases the earlier writer's + * claims. Persisting a ledger that loses half its evidence is not compounding + * knowledge; it is losing it more slowly. + * + * `merge` is SYNCHRONOUS on purpose: it runs inside the critical section, and + * an `await` in there is a lock held across arbitrary I/O. + */ + mergeClaimLedger( + id: string, + merge: (current: ResearchClaimLedger | null) => ResearchClaimLedger, + ): Promise +} + +export class MemoryKbStore implements KbStore, ClaimLedgerStore { private readonly sources = new Map() private readonly pages = new Map() private readonly events: KnowledgeEvent[] = [] + private readonly claimLedgers = new Map() private index: KnowledgeIndex | null = null async putSource(source: SourceRecord): Promise { @@ -86,12 +163,74 @@ export class MemoryKbStore implements KbStore { out = [...out].sort((a, b) => a.createdAt.localeCompare(b.createdAt)) return out.slice(-(query.limit ?? out.length)).map(clone) } + + async putClaimLedger(ledger: ResearchClaimLedger): Promise { + const parsed = ResearchClaimLedgerSchema.parse(ledger) as ResearchClaimLedger + this.claimLedgers.set(assertClaimLedgerId(parsed.id), clone(parsed)) + } + + async getClaimLedger(id: string): Promise { + return clone(this.claimLedgers.get(assertClaimLedgerId(id)) ?? null) + } + + async listClaimLedgers(): Promise { + return [...this.claimLedgers.values()].map(clone).sort((a, b) => a.id.localeCompare(b.id)) + } + + async mergeClaimLedger( + id: string, + merge: (current: ResearchClaimLedger | null) => ResearchClaimLedger, + ): Promise { + // No lock: `merge` is synchronous and this is one process, so nothing can + // interleave between the read and the write. + const key = assertClaimLedgerId(id) + const next = assertMergedLedgerId(key, merge(clone(this.claimLedgers.get(key) ?? null))) + await this.putClaimLedger(next) + return clone(next) + } } const knowledgeEventsSchema = z.array(KnowledgeEventSchema) -export class FileSystemKbStore implements KbStore { - constructor(private readonly dir: string) {} +/** + * The durable record store for one knowledge base. + * + * The object constructor is anchored on the knowledge-base root and keeps its + * records under `/.agent-knowledge/`, which is the layout used by + * `writeKnowledgeIndex`. The string constructor preserves the published direct + * record-directory layout (`/index.json`). + */ +export interface FileSystemKbStoreOptions { + /** Knowledge-base root; records are stored under `/.agent-knowledge/`. */ + root: string +} + +export class FileSystemKbStore implements KbStore, ClaimLedgerStore { + private readonly root: string + private readonly indexPath: string + private readonly eventsPath: string + private readonly claimLedgerDir: string + + /** + * A string retains the published direct-directory contract. + * The object form explicitly selects a knowledge-base root and the canonical + * `.agent-knowledge/` layout. A string naming that exact canonical directory + * keeps its file paths but shares the root form's lock domain. + */ + constructor(input: string | FileSystemKbStoreOptions) { + const directDirectory = typeof input === 'string' ? resolve(input) : undefined + const aliasesCanonicalDirectory = + directDirectory !== undefined && basename(directDirectory) === KB_STORE_DIR + this.root = aliasesCanonicalDirectory + ? dirname(directDirectory) + : typeof input === 'string' + ? input + : input.root + const canonicalLayout = typeof input !== 'string' || aliasesCanonicalDirectory + this.indexPath = canonicalLayout ? KB_INDEX_PATH : 'index.json' + this.eventsPath = canonicalLayout ? KB_EVENTS_PATH : 'events.json' + this.claimLedgerDir = canonicalLayout ? KB_CLAIM_LEDGER_DIR : 'claim-ledgers' + } async putSource(source: SourceRecord): Promise { const parsed = SourceRecordSchema.parse(source) as SourceRecord @@ -103,14 +242,14 @@ export class FileSystemKbStore implements KbStore { } async getSource(id: string): Promise { - return withKnowledgeRead(this.dir, async () => { + return withKnowledgeRead(this.root, async () => { const index = await this.readIndex() return clone(index?.sources.find((source) => source.id === id) ?? null) }) } async listSources(): Promise { - return withKnowledgeRead(this.dir, async () => clone((await this.readIndex())?.sources ?? [])) + return withKnowledgeRead(this.root, async () => clone((await this.readIndex())?.sources ?? [])) } async putPage(page: KnowledgePage): Promise { @@ -127,7 +266,7 @@ export class FileSystemKbStore implements KbStore { } async getPage(idOrPath: string): Promise { - return withKnowledgeRead(this.dir, async () => { + return withKnowledgeRead(this.root, async () => { const index = await this.readIndex() return clone( index?.pages.find((page) => page.id === idOrPath || page.path === idOrPath) ?? null, @@ -136,33 +275,33 @@ export class FileSystemKbStore implements KbStore { } async listPages(): Promise { - return withKnowledgeRead(this.dir, async () => clone((await this.readIndex())?.pages ?? [])) + return withKnowledgeRead(this.root, async () => clone((await this.readIndex())?.pages ?? [])) } async putIndex(index: KnowledgeIndex): Promise { const parsed = KnowledgeIndexSchema.parse(index) as KnowledgeIndex - await withKnowledgeMutation(this.dir, () => - writeJsonDurableWithinRoot(this.dir, 'index.json', parsed), + await withKnowledgeMutation(this.root, () => + writeJsonDurableWithinRoot(this.root, this.indexPath, parsed), ) } async getIndex(): Promise { - return withKnowledgeRead(this.dir, () => this.readIndex()) + return withKnowledgeRead(this.root, () => this.readIndex()) } async putEvent(event: KnowledgeEvent): Promise { const parsed = KnowledgeEventSchema.parse(event) as KnowledgeEvent - await withKnowledgeMutation(this.dir, async () => { + await withKnowledgeMutation(this.root, async () => { const current = await this.readEvents() const next = [...current.filter((entry) => entry.id !== parsed.id), parsed].sort((a, b) => a.createdAt.localeCompare(b.createdAt), ) - await writeJsonDurableWithinRoot(this.dir, 'events.json', next) + await writeJsonDurableWithinRoot(this.root, this.eventsPath, next) }) } async listEvents(query: KnowledgeEventQuery = {}): Promise { - return withKnowledgeRead(this.dir, async () => { + return withKnowledgeRead(this.root, async () => { let events = await this.readEvents() if (query.type) events = events.filter((event) => event.type === query.type) if (query.target) events = events.filter((event) => event.target === query.target) @@ -170,26 +309,106 @@ export class FileSystemKbStore implements KbStore { }) } + async putClaimLedger(ledger: ResearchClaimLedger): Promise { + const parsed = ResearchClaimLedgerSchema.parse(ledger) as ResearchClaimLedger + const path = this.claimLedgerPath(parsed.id) + await withKnowledgeMutation(this.root, () => + writeJsonDurableWithinRoot(this.root, path, parsed), + ) + } + + async getClaimLedger(id: string): Promise { + const path = this.claimLedgerPath(id) + return withKnowledgeRead( + this.root, + () => + readJsonFile( + this.root, + path, + ResearchClaimLedgerSchema, + ) as Promise, + ) + } + + async listClaimLedgers(): Promise { + return withKnowledgeRead(this.root, async () => { + let files: Awaited> + try { + files = await listRegularFilesWithinRoot(this.root, this.claimLedgerDir) + } catch (error) { + if (isMissingFile(error)) return [] + throw error + } + const ledgers: ResearchClaimLedger[] = [] + for (const file of files) { + if (!file.path.endsWith('.json')) continue + ledgers.push( + ResearchClaimLedgerSchema.parse( + JSON.parse(file.bytes.toString('utf8')), + ) as ResearchClaimLedger, + ) + } + return ledgers.sort((a, b) => a.id.localeCompare(b.id)) + }) + } + + async mergeClaimLedger( + id: string, + merge: (current: ResearchClaimLedger | null) => ResearchClaimLedger, + ): Promise { + const key = assertClaimLedgerId(id) + // One mutation scope spans the read AND the write, so a second process + // cannot read the same `current` this one did. `withKnowledgeMutation` is + // reentrant, so the nested lock inside `putClaimLedger` joins this scope + // rather than deadlocking against it. + return withKnowledgeMutation(this.root, async () => { + const current = (await readJsonFile( + this.root, + this.claimLedgerPath(key), + ResearchClaimLedgerSchema, + )) as ResearchClaimLedger | null + const next = assertMergedLedgerId(key, merge(current)) + await this.putClaimLedger(next) + return next + }) + } + private async updateIndex(change: (index: KnowledgeIndex) => KnowledgeIndex): Promise { - await withKnowledgeMutation(this.dir, async () => { - const current = (await this.readIndex()) ?? emptyIndex(this.dir) + await withKnowledgeMutation(this.root, async () => { + const current = (await this.readIndex()) ?? emptyIndex(this.root) const next = KnowledgeIndexSchema.parse(change(current)) as KnowledgeIndex - await writeJsonDurableWithinRoot(this.dir, 'index.json', next) + await writeJsonDurableWithinRoot(this.root, this.indexPath, next) }) } private async readIndex(): Promise { return readJsonFile( - this.dir, - 'index.json', + this.root, + this.indexPath, KnowledgeIndexSchema, ) as Promise } private async readEvents(): Promise { - return ((await readJsonFile(this.dir, 'events.json', knowledgeEventsSchema)) ?? + return ((await readJsonFile(this.root, this.eventsPath, knowledgeEventsSchema)) ?? []) as KnowledgeEvent[] } + + private claimLedgerPath(id: string): string { + return `${this.claimLedgerDir}/${assertClaimLedgerId(id)}.json` + } +} + +/** + * A merge that returns a ledger under a different id would write that ledger to + * the file the caller asked to merge, giving one file two identities. Refuse + * rather than trust the merge function to be well behaved. + */ +function assertMergedLedgerId(id: string, merged: ResearchClaimLedger): ResearchClaimLedger { + if (merged.id !== id) { + throw new Error(`merge of claim ledger '${id}' returned a ledger with id '${merged.id}'`) + } + return merged } function emptyIndex(root: string): KnowledgeIndex { diff --git a/src/research-driving-driver.ts b/src/research-driving-driver.ts index 08f3956..c6544b0 100644 --- a/src/research-driving-driver.ts +++ b/src/research-driving-driver.ts @@ -38,12 +38,28 @@ * contested IS. * * It reuses `runVerifiedResearchLoop` (it is a plain `ResearchDriver`), the web - * worker, `sha256` (claim identity), `canonicalizeUrl` (independent-source - * identity), and the `RouterClient` chat surface; it reinvents none of them. + * worker, `claim-ledger.ts` (claim identity, independent-source identity, and + * the merge rule), and the `RouterClient` chat surface; it reinvents none of + * them. */ -import { canonicalizeUrl } from './adaptive-driver' -import { sha256 } from './ids' +import { + claimEvidenceId, + claimId, + deepQuestionId, + claimSourceHost as hostOf, + mergeClaimLedgers, + normalizeClaimText, +} from './claim-ledger' +import { assertClaimLedgerId, type ClaimLedgerStore } from './kb-store' +import type { + DeepQuestion, + DeepQuestionKind, + ResearchClaimEvidence, + ResearchClaimLedger, + ResearchClaimRecord, + TrackedClaim, +} from './types' import type { KnowledgeGap, ResearchDriver, @@ -57,46 +73,21 @@ import { type TangleRouterOptions, } from './web-research-worker' -/** The four deep sub-question kinds the driver generates to drive depth. */ -export type DeepQuestionKind = 'comparative' | 'mechanism' | 'gap' | 'contradiction' - -/** A deep sub-question the driver folds into the worker's next prompt. */ -export interface DeepQuestion { - kind: DeepQuestionKind - text: string - /** sha256-derived stable id, so "addressed" can be tracked across rounds. */ - id: string - /** Claim id(s) this question interrogates (for contradiction/mechanism kinds). */ - claimIds: string[] - /** True once a later round's evidence addressed it (see `markAddressed`). */ - addressed: boolean - /** The round this question was raised in. */ - raisedRound: number -} - -/** One tracked claim plus the independent sources that assert it. */ -export interface TrackedClaim { - id: string - /** The claim text as first extracted (kept for prompts/audit). */ - text: string - /** Canonical hosts of the INDEPENDENT sources that assert this claim. */ - supportingHosts: Set - /** Source URIs that assert this claim (provenance; may share a host). */ - supportingUris: string[] - /** Claim ids this claim was found to CONTRADICT (and vice versa). */ - contradicts: Set - /** - * CONTESTED = a contradiction the loop surfaced but could not resolve to a - * single supported claim. A contested claim counts as "settled enough to be - * done" (we report the disagreement) even with < 2 independent sources. - */ - contested: boolean - firstSeenRound: number +// The live claim type and its durable ledger records live in `types.ts` with +// the package's other public shapes. Re-exported here so existing importers +// keep working. +export type { + DeepQuestion, + DeepQuestionKind, + ResearchClaimEvidence, + ResearchClaimLedger, + ResearchClaimRecord, + TrackedClaim, } /** The driver's accumulated research state — the completion oracle reads this. */ export interface ResearchDrivingState { - /** Every claim extracted from the worker's sources, by id. */ + /** Every extracted claim backed by a registered source, by id. */ claims: TrackedClaim[] /** Every deep sub-question raised, by id. */ questions: DeepQuestion[] @@ -135,6 +126,21 @@ export interface ResearchDrivingDriverOptions { onSteer?: (steer: ResearchDrivingSteer) => void } +/** + * Options for the durable driver: the same driver plus a store to keep its + * belief state in. + */ +export interface PersistentResearchDrivingDriverOptions extends ResearchDrivingDriverOptions { + /** Where the claim ledger is read from and written to. */ + store: ClaimLedgerStore + /** + * Names this run's ledger. Stable across resumes (that is what makes a resume + * a resume) and distinct per run against one knowledge base. Must be a single + * safe path segment — see `assertClaimLedgerId`. + */ + ledgerId: string +} + /** What the driver folded into one round's worker prompt, surfaced for audit. */ export interface ResearchDrivingSteer { round: number @@ -168,6 +174,25 @@ export interface ResearchDrivingDriver extends ResearchDriver { * to assert the driver produced deeper questions / invalidation challenges. */ lastSteer(): ResearchDrivingSteer | undefined + /** + * Write the current belief state to the store. `verifySource` already persists + * each pending observation; this exists for the state `foldGaps` produces — + * the deep questions — which is raised by a synchronous hook and would + * otherwise be lost if the process died before the next source arrived. + * + * `runVerifiedResearchLoop` calls this at the end of every round. A driver + * built without a store has nothing to write and resolves immediately. + */ + checkpoint(): Promise + /** Durably announce the next synchronous fold before it begins. */ + prepareFold(): Promise + /** + * Confirm that source registration completed for these exact original URIs. + * Pending extracted evidence cannot affect claims or completion before this. + */ + commitSources(sourceUris: readonly string[]): Promise + /** The ledger record as it would be written right now. */ + toLedger(): ResearchClaimLedger } /** A claim the extractor returns for one source. */ @@ -177,8 +202,47 @@ interface ExtractedClaim { contradictsExistingId?: string } +/** + * The in-memory driver. Its belief state lives for exactly as long as the + * process does — use `createPersistentResearchDrivingDriver` when the run must + * survive a crash or resume. + */ export function createResearchDrivingDriver( options: ResearchDrivingDriverOptions = {}, +): ResearchDrivingDriver { + return buildDriver(options) +} + +/** + * The durable driver: same behaviour, plus its claim ledger is read from the + * store at construction and written back after every claim and every round. + * + * Construction is asynchronous because loading is I/O, and loading has to happen + * before the caller can read `researchState()` or `isComplete()` — a driver that + * loaded lazily would answer "nothing researched, not complete" for a run that + * had already corroborated everything. + */ +export async function createPersistentResearchDrivingDriver( + options: PersistentResearchDrivingDriverOptions, +): Promise { + const ledgerId = assertClaimLedgerId(options.ledgerId) + const existing = await options.store.getClaimLedger(ledgerId) + const driver = buildDriver(options, { store: options.store, ledgerId }, existing ?? undefined) + if (existing && (existing.preparedRounds ?? existing.rounds) > existing.rounds) { + await driver.checkpoint() + } + return driver +} + +interface DriverPersistence { + store: ClaimLedgerStore + ledgerId: string +} + +function buildDriver( + options: ResearchDrivingDriverOptions, + persistence?: DriverPersistence, + restored?: ResearchClaimLedger, ): ResearchDrivingDriver { const minIndependentSources = Math.max(2, options.minIndependentSources ?? 2) const maxQuestionsPerRound = Math.max(1, options.maxQuestionsPerRound ?? 6) @@ -186,12 +250,124 @@ export function createResearchDrivingDriver( const deterministicFallback = options.deterministicFallback ?? true // The claim ledger, keyed by claim id (sha256 of the normalized claim text). - const claims = new Map() + const claims = new Map( + (restored?.claims ?? []).map((claim) => [claim.id, fromRecord(claim)]), + ) + // Claim extraction and source registration are two separate durable writes. + // Evidence remains inert until its exact source URI is confirmed here. + const claimEvidence = new Map( + (restored?.claimEvidence ?? []).map((evidence) => [evidence.id, evidence]), + ) + const registeredSourceUris = new Set(restored?.registeredSourceUris ?? []) // Every deep question raised, by id — so we can mark them addressed later. - const questions = new Map() - let rounds = 0 + const questions = new Map( + (restored?.questions ?? []).map((question) => [question.id, question]), + ) + let rounds = restored?.rounds ?? 0 + let preparedRounds = Math.max(rounds, restored?.preparedRounds ?? rounds) + let goal = restored?.goal let lastSteer: ResearchDrivingSteer | undefined + // `prepareFold` is persisted before the synchronous fold starts. If the + // process died during that fold, regenerate its deterministic questions now + // and advance the in-memory round; `createPersistentResearchDrivingDriver` + // checkpoints this recovered state before returning it. + if (preparedRounds > rounds) { + for (let round = rounds + 1; round <= preparedRounds; round += 1) { + for (const question of synthesizeDeepQuestions([...claims.values()], round).slice( + 0, + maxQuestionsPerRound, + )) { + if (!questions.has(question.id)) questions.set(question.id, question) + } + } + rounds = preparedRounds + } + + function toLedger(): ResearchClaimLedger { + return { + id: persistence?.ledgerId ?? 'in-memory', + ...(goal === undefined ? {} : { goal }), + updatedAt: new Date().toISOString(), + rounds, + ...(preparedRounds > rounds ? { preparedRounds } : {}), + claimEvidence: [...claimEvidence.values()].sort((left, right) => + left.id.localeCompare(right.id), + ), + registeredSourceUris: [...registeredSourceUris].sort(), + claims: [...claims.values()] + .map((claim) => ({ + ...claim, + supportingHosts: [...new Set(claim.supportingHosts)].sort(), + supportingUris: [...new Set(claim.supportingUris)].sort(), + contradicts: [...new Set(claim.contradicts)].sort(), + })) + .sort((a, b) => a.id.localeCompare(b.id)), + questions: [...questions.values()] + .map((question) => ({ + ...question, + claimIds: [...new Set(question.claimIds)].sort(), + })) + .sort((a, b) => a.id.localeCompare(b.id)), + } + } + + /** + * Write this driver's belief state into the stored ledger and adopt the + * result. + * + * It merges rather than overwrites, and then rehydrates from the merged + * record, which is the whole of what makes knowledge compound across + * workers. Two drivers on one ledger id — a resumed run beside a still-live + * one, or two workers researching one goal in parallel — would otherwise each + * write a whole record built from what it read before the other wrote, and + * the later write would erase the earlier writer's claims. Rehydrating means + * a claim another worker corroborated counts toward THIS driver's completion + * oracle from the next round onward. + */ + async function persist(): Promise { + if (!persistence) return + const mine = toLedger() + const merged = await persistence.store.mergeClaimLedger(persistence.ledgerId, (current) => + current === null ? mine : mergeClaimLedgers(current, mine), + ) + claims.clear() + for (const claim of merged.claims) claims.set(claim.id, fromRecord(claim)) + claimEvidence.clear() + for (const evidence of merged.claimEvidence) claimEvidence.set(evidence.id, evidence) + registeredSourceUris.clear() + for (const sourceUri of merged.registeredSourceUris) registeredSourceUris.add(sourceUri) + questions.clear() + for (const question of merged.questions) questions.set(question.id, question) + rounds = Math.max(rounds, merged.rounds) + preparedRounds = Math.max(rounds, merged.preparedRounds ?? merged.rounds) + goal = merged.goal ?? goal + } + + async function prepareFold(): Promise { + if (!persistence) return + preparedRounds = Math.max(preparedRounds, rounds + 1) + await persist() + } + + /** + * A ledger accumulates evidence FOR a goal. Reusing one id across two goals + * merges two runs' beliefs into one corroboration count, which is worse than + * losing them, so it fails rather than merging. + */ + function bindGoal(nextGoal: string): void { + if (goal === undefined) { + goal = nextGoal + return + } + if (goal !== nextGoal) { + throw new Error( + `claim ledger '${persistence?.ledgerId ?? 'in-memory'}' accumulated evidence for goal ` + + `'${goal}' and cannot be reused for '${nextGoal}'`, + ) + } + } + function resolveRouter(): RouterClient { return options.router ?? createTangleRouterClient(options.router_options) } @@ -203,7 +379,7 @@ export function createResearchDrivingDriver( const existing = claims.get(id) if (existing) { if (host) existing.supportingHosts.add(host) - if (!existing.supportingUris.includes(sourceUri)) existing.supportingUris.push(sourceUri) + addUnique(existing.supportingUris, sourceUri) linkContradiction(existing, extracted.contradictsExistingId) return existing } @@ -221,6 +397,78 @@ export function createResearchDrivingDriver( return tracked } + /** Persist an extraction observation without treating its source as registered. */ + function recordEvidence( + extracted: ExtractedClaim, + sourceUri: string, + round: number, + ): ResearchClaimEvidence { + const text = extracted.text.trim() + const observedClaimId = claimId(text) + const contradictsClaimId = + extracted.contradictsExistingId === observedClaimId + ? undefined + : extracted.contradictsExistingId + const evidence: ResearchClaimEvidence = { + id: claimEvidenceId({ claimId: observedClaimId, sourceUri, contradictsClaimId }), + claimId: observedClaimId, + text, + sourceUri, + ...(contradictsClaimId === undefined ? {} : { contradictsClaimId }), + firstSeenRound: round, + } + const existing = claimEvidence.get(evidence.id) + if (!existing) { + claimEvidence.set(evidence.id, evidence) + return evidence + } + const earlier = + evidence.firstSeenRound < existing.firstSeenRound || + (evidence.firstSeenRound === existing.firstSeenRound && evidence.text < existing.text) + ? evidence + : existing + const merged = { + ...earlier, + firstSeenRound: Math.min(existing.firstSeenRound, evidence.firstSeenRound), + } + claimEvidence.set(merged.id, merged) + return merged + } + + /** Materialize evidence for newly confirmed source URIs into live claims. */ + function materializeEvidenceFor(sourceUris: ReadonlySet): string[] { + const evidence = [...claimEvidence.values()].filter((item) => sourceUris.has(item.sourceUri)) + const texts: string[] = [] + for (const item of evidence) { + recordClaim( + { text: item.text, contradictsExistingId: item.contradictsClaimId }, + item.sourceUri, + item.firstSeenRound, + ) + texts.push(item.text) + } + // A refuter can sort before the original claim, so close edges only after + // every claim for this confirmation batch exists. + for (const item of evidence) { + const claim = claims.get(item.claimId) + if (claim) linkContradiction(claim, item.contradictsClaimId) + } + return texts + } + + async function commitSources(sourceUris: readonly string[]): Promise { + const newlyRegistered = new Set() + for (const sourceUri of sourceUris) { + if (registeredSourceUris.has(sourceUri)) continue + registeredSourceUris.add(sourceUri) + newlyRegistered.add(sourceUri) + } + if (newlyRegistered.size > 0) { + markAddressed(materializeEvidenceFor(newlyRegistered)) + } + await persist() + } + /** Wire a bidirectional contradiction edge and mark BOTH claims contested. */ function linkContradiction(claim: TrackedClaim, otherId: string | undefined): void { if (!otherId || otherId === claim.id) return @@ -305,6 +553,7 @@ export function createResearchDrivingDriver( source: ResearchSourceProposal, ctx: SourceVerificationContext, ): Promise { + bindGoal(ctx.goal) const extracted = await extractClaims(source, ctx) if (extracted.length === 0) { return { @@ -312,12 +561,21 @@ export function createResearchDrivingDriver( reason: 'no extractable claim: source yields nothing to drive the research deeper', } } - const newTexts: string[] = [] for (const claim of extracted) { - recordClaim(claim, source.uri, ctx.round) - newTexts.push(claim.text) + recordEvidence(claim, source.uri, ctx.round) } - markAddressed(newTexts) + if (!persistence) { + registeredSourceUris.add(source.uri) + markAddressed(materializeEvidenceFor(new Set([source.uri]))) + } else if (registeredSourceUris.has(source.uri)) { + // Direct callers can verify a URI already present in the registry. Its + // newly extracted evidence is safe to materialize immediately. + markAddressed(materializeEvidenceFor(new Set([source.uri]))) + } + // Persist the observation BEFORE accepting. It remains pending until the + // loop confirms source registration through `commitSources`, closing both + // possible crash directions without a cross-store transaction. + await persist() return { accept: true } }, @@ -328,6 +586,9 @@ export function createResearchDrivingDriver( * (3) INVALIDATION challenges for weakly-supported / contradicted claims. */ foldGaps(gaps: KnowledgeGap[]): string { + if (persistence && preparedRounds <= rounds) { + throw new Error('persistent research driver must prepareFold before foldGaps') + } rounds += 1 const round = rounds const ledger = [...claims.values()] @@ -365,6 +626,14 @@ export function createResearchDrivingDriver( lastSteer(): ResearchDrivingSteer | undefined { return lastSteer }, + + checkpoint: persist, + + prepareFold, + + commitSources, + + toLedger, } // -- claim extraction ------------------------------------------------------ @@ -373,13 +642,30 @@ export function createResearchDrivingDriver( source: ResearchSourceProposal, ctx: SourceVerificationContext, ): Promise { - const ledger = [...claims.values()] + const ledger = claimsForExtraction() const fromLlm = await extractClaimsWithLlm(source, ctx, ledger) if (fromLlm.length > 0) return fromLlm.slice(0, maxClaimsPerSource) if (deterministicFallback) return deterministicClaims(source).slice(0, maxClaimsPerSource) return [] } + function claimsForExtraction(): TrackedClaim[] { + const known = new Map(claims) + for (const evidence of claimEvidence.values()) { + if (known.has(evidence.claimId)) continue + known.set(evidence.claimId, { + id: evidence.claimId, + text: evidence.text, + supportingHosts: new Set(), + supportingUris: [], + contradicts: new Set(), + contested: false, + firstSeenRound: evidence.firstSeenRound, + }) + } + return [...known.values()] + } + async function extractClaimsWithLlm( source: ResearchSourceProposal, ctx: SourceVerificationContext, @@ -538,34 +824,28 @@ function makeQuestion( return { kind, text, - id: `q_${sha256(`${kind}:${text}`).slice(0, 16)}`, - claimIds, + id: deepQuestionId(kind, text), + claimIds: [...new Set(claimIds)].sort(), addressed: false, raisedRound, } } -/** Claim identity = sha256 of the normalized claim text (same words ⇒ same claim). */ -function claimId(text: string): string { - return `c_${sha256(normalizeText(text)).slice(0, 16)}` -} - -function hostOf(uri: string): string { - try { - return new URL(uri.trim()).hostname.toLowerCase().replace(/^www\./, '') - } catch { - // Non-URL identifier (offline corpus uris like `web/foo`): canonicalize so - // distinct identifiers still count as distinct independent sources. - return canonicalizeUrl(uri) +function fromRecord(claim: ResearchClaimRecord): TrackedClaim { + return { + ...claim, + supportingHosts: new Set(claim.supportingHosts), + supportingUris: [...claim.supportingUris], + contradicts: new Set(claim.contradicts), } } -function normalizeText(text: string): string { - return text - .toLowerCase() - .replace(/[^\p{L}\p{N}\s]+/gu, ' ') - .replace(/\s+/g, ' ') - .trim() +/** + * Set-like insert into the array form. The ledger's collections are arrays so + * they survive `JSON.stringify`; dedup is enforced here instead of by the type. + */ +function addUnique(values: string[], value: string): void { + if (!values.includes(value)) values.push(value) } const stopwords = new Set([ @@ -622,7 +902,7 @@ const stopwords = new Set([ function contentWordSet(text: string): Set { return new Set( - normalizeText(text) + normalizeClaimText(text) .split(' ') .filter((word) => word.length >= 3 && !stopwords.has(word)), ) diff --git a/src/schemas.ts b/src/schemas.ts index fbb3660..b2b27a6 100644 --- a/src/schemas.ts +++ b/src/schemas.ts @@ -1,4 +1,11 @@ import { z } from 'zod' +import { + assertDeepQuestionIntegrity, + assertResearchClaimEvidenceIntegrity, + assertResearchClaimLedgerIntegrity, + assertTrackedClaimIntegrity, +} from './claim-ledger' +import { KNOWLEDGE_EVENT_TYPES } from './types' export const SourceAnchorSchema = z.object({ id: z.string().min(1), @@ -68,21 +75,87 @@ export const KnowledgeIndexSchema = z.object({ export const KnowledgeEventSchema = z.object({ id: z.string().min(1), - type: z.enum([ - 'source.added', - 'proposal.applied', - 'index.built', - 'lint.run', - 'optimization.run', - 'release.promoted', - 'release.rejected', - ]), + // Derived from the type union's own value list — see KNOWLEDGE_EVENT_TYPES. + // A hand-restated copy of this enum drifted and silently rejected the one + // event the research loop emits. + type: z.enum(KNOWLEDGE_EVENT_TYPES), createdAt: z.string().min(1), actor: z.string().optional(), target: z.string().optional(), metadata: z.record(z.string(), z.unknown()).optional(), }) +export const DeepQuestionSchema = z + .object({ + kind: z.enum(['comparative', 'mechanism', 'gap', 'contradiction']), + text: z.string().min(1), + id: z.string().min(1), + claimIds: z.array(z.string().min(1)), + addressed: z.boolean(), + raisedRound: z.number().int().nonnegative(), + }) + .strict() + .superRefine((question, context) => { + reportIntegrityError(context, () => assertDeepQuestionIntegrity(question)) + }) + +export const ResearchClaimRecordSchema = z + .object({ + id: z.string().min(1), + text: z.string().min(1), + supportingHosts: z.array(z.string().min(1)), + supportingUris: z.array(z.string().min(1)), + contradicts: z.array(z.string().min(1)), + contested: z.boolean(), + firstSeenRound: z.number().int().nonnegative(), + }) + .strict() + .superRefine((claim, context) => { + reportIntegrityError(context, () => assertTrackedClaimIntegrity(claim)) + }) + +export const ResearchClaimEvidenceSchema = z + .object({ + id: z.string().min(1), + claimId: z.string().min(1), + text: z.string().min(1), + sourceUri: z.string().min(1), + contradictsClaimId: z.string().min(1).optional(), + firstSeenRound: z.number().int().nonnegative(), + }) + .strict() + .superRefine((evidence, context) => { + reportIntegrityError(context, () => assertResearchClaimEvidenceIntegrity(evidence)) + }) + +export const ResearchClaimLedgerSchema = z + .object({ + id: z.string().min(1), + goal: z.string().trim().min(1).optional(), + updatedAt: z.iso.datetime(), + rounds: z.number().int().nonnegative(), + preparedRounds: z.number().int().nonnegative().optional(), + claimEvidence: z.array(ResearchClaimEvidenceSchema), + registeredSourceUris: z.array(z.string().min(1)), + claims: z.array(ResearchClaimRecordSchema), + questions: z.array(DeepQuestionSchema), + }) + .strict() + .superRefine((ledger, context) => { + reportIntegrityError(context, () => assertResearchClaimLedgerIntegrity(ledger)) + }) + +function reportIntegrityError(context: z.core.$RefinementCtx, check: () => void): void { + try { + check() + } catch (error) { + context.addIssue({ + code: 'custom', + message: error instanceof Error ? error.message : String(error), + }) + } +} + export const KnowledgeBaseCandidateSchema = z.object({ id: z.string().min(1), units: z.array( diff --git a/src/types.ts b/src/types.ts index e24053e..180ef99 100644 --- a/src/types.ts +++ b/src/types.ts @@ -179,15 +179,26 @@ export interface KnowledgeWriteParseResult { warnings: string[] } -export type KnowledgeEventType = - | 'source.added' - | 'proposal.applied' - | 'index.built' - | 'lint.run' - | 'research.iteration' - | 'optimization.run' - | 'release.promoted' - | 'release.rejected' +/** + * The event vocabulary, as a value so the runtime schema is DERIVED from it + * rather than restated. A restated copy in `schemas.ts` drifted: it omitted + * `research.iteration`, which is the only event `runVerifiedResearchLoop` + * produces, so every attempt to store one would have been rejected. Nothing + * caught it because nothing ever stored an event. Add a type here and the + * schema accepts it in the same edit. + */ +export const KNOWLEDGE_EVENT_TYPES = [ + 'source.added', + 'proposal.applied', + 'index.built', + 'lint.run', + 'research.iteration', + 'optimization.run', + 'release.promoted', + 'release.rejected', +] as const + +export type KnowledgeEventType = (typeof KNOWLEDGE_EVENT_TYPES)[number] export interface KnowledgeEvent { id: string @@ -198,6 +209,119 @@ export interface KnowledgeEvent { metadata?: Record } +/** The four deep sub-question kinds a research driver raises to drive depth. */ +export type DeepQuestionKind = 'comparative' | 'mechanism' | 'gap' | 'contradiction' + +/** + * A deep sub-question the research driver folds into the worker's next prompt. + * + * Lives here, next to the other record types, because it is persisted state: + * `addressed` is the half of the completion oracle that cannot be recomputed + * from the claim ledger alone, so a run that loses it reports "complete" for + * questions nobody ever answered. + */ +export interface DeepQuestion { + kind: DeepQuestionKind + text: string + /** sha256-derived stable id, so "addressed" can be tracked across rounds. */ + id: string + /** Claim id(s) this question interrogates (for contradiction/mechanism kinds). */ + claimIds: string[] + /** True once a later round's evidence addressed it. */ + addressed: boolean + /** The round this question was raised in. */ + raisedRound: number +} + +/** One live tracked claim exposed by the research-driving API. */ +export interface TrackedClaim { + id: string + /** The claim text as first extracted (kept for prompts/audit). */ + text: string + /** Canonical hosts of the INDEPENDENT sources that assert this claim. */ + supportingHosts: Set + /** Source URIs that assert this claim (provenance; may share a host). */ + supportingUris: string[] + /** Claim ids this claim was found to CONTRADICT (and vice versa). */ + contradicts: Set + /** + * CONTESTED = a contradiction the loop surfaced but could not resolve to a + * single supported claim. A contested claim counts as "settled enough to be + * done" (we report the disagreement) even with < 2 independent sources. + */ + contested: boolean + firstSeenRound: number +} + +/** + * JSON-safe form of a tracked claim stored in a research claim ledger. + * + * The live `TrackedClaim` contract retains its published `Set` fields. + * Durable records use sorted arrays because `JSON.stringify` turns a `Set` + * into `{}`, which would erase every corroboration count and contradiction. + */ +export interface ResearchClaimRecord { + id: string + text: string + supportingHosts: string[] + supportingUris: string[] + contradicts: string[] + contested: boolean + firstSeenRound: number +} + +/** + * One immutable claim extraction observed while a source is being verified. + * + * An observation is deliberately separate from `ResearchClaimRecord`: source + * verification happens before source registration, and a process can die in + * between. The observation is durable immediately, but it contributes support + * to a claim only after `sourceUri` appears in the ledger's independently + * confirmed `registeredSourceUris` set. + */ +export interface ResearchClaimEvidence { + /** Stable identity of this claim/source/contradiction observation. */ + id: string + claimId: string + text: string + sourceUri: string + /** Existing claim this observation directly contradicts, when reported. */ + contradictsClaimId?: string + firstSeenRound: number +} + +/** + * The durable record of one research run's belief state: which claims were + * extracted, how independently each is supported, which contradict which, and + * which deep sub-questions are still open. + * + * `id` names the run — one knowledge base can host several, and they must not + * overwrite each other, so the store addresses ledgers by this id. + */ +export interface ResearchClaimLedger { + id: string + /** The research goal this ledger accumulated evidence for. */ + goal?: string + /** ISO timestamp of the last write. */ + updatedAt: string + /** How many rounds the driver has folded steer for. */ + rounds: number + /** + * Highest round durably announced before its synchronous question-generation + * step began. Greater than `rounds` only while a round needs crash recovery. + */ + preparedRounds?: number + /** + * Extracted evidence, including observations whose source registration has + * not yet been confirmed. Pending observations never count toward claims. + */ + claimEvidence: ResearchClaimEvidence[] + /** Exact original source URIs confirmed present in the source registry. */ + registeredSourceUris: string[] + claims: ResearchClaimRecord[] + questions: DeepQuestion[] +} + export interface KnowledgeRelease { id: string candidateId: string diff --git a/src/verified-research-loop.ts b/src/verified-research-loop.ts index 0ff3353..e3f6363 100644 --- a/src/verified-research-loop.ts +++ b/src/verified-research-loop.ts @@ -7,6 +7,7 @@ import { } from './eval-readiness' import { createKnowledgeEvent } from './events' import { buildKnowledgeIndex } from './indexer' +import { FileSystemKbStore } from './kb-store' import { applyKnowledgeWriteBlocks } from './proposals' import { readinessFor } from './readiness-helpers' import { searchKnowledge } from './search' @@ -117,6 +118,13 @@ export interface DriverResearchContext { * open. Only invoked when `driverResearches` is true. * - `foldGaps` — turn the remaining gaps into a steer string for the worker's * next prompt. Defaults to a compact bulleted list when omitted. + * - `checkpoint` — write whatever state the driver accumulated to durable + * storage. Called at the end of every round, after `foldGaps`, so state that + * a synchronous hook produced is on disk before the next round can crash. + * - `prepareFold` — durably announce the next synchronous fold before it runs, + * so a crash between question generation and `checkpoint` can be recovered. + * - `commitSources` — confirm source records are durable after verification; + * drivers with pending evidence must not count it before this callback. */ export interface ResearchDriver { verifySource( @@ -125,6 +133,9 @@ export interface ResearchDriver { ): Promise | SourceVerdict research?(ctx: DriverResearchContext): Promise | ResearchContribution foldGaps?(gaps: KnowledgeGap[]): string + prepareFold?(): Promise | void + commitSources?(sourceUris: readonly string[]): Promise | void + checkpoint?(): Promise | void } export type SourceVerdict = { accept: true } | { accept: false; reason: string } @@ -213,8 +224,13 @@ export async function runVerifiedResearchLoop( ): Promise { const maxRounds = Math.max(1, options.maxRounds ?? 3) await initKnowledgeBase(options.root) + const store = new FileSystemKbStore({ root: options.root }) const steps: VerifiedResearchRound[] = [] let index = await buildKnowledgeIndex(options.root) + // Reconcile a source write that completed before a previous process died + // while confirming it to the driver. Exact original URIs are the shared + // identity; stored `record.uri` values are rewritten raw-file paths. + await confirmRegisteredSources(options.driver, index.sources) let readiness = readinessFor(options, index) let ready = isReady(readiness?.report) let steer: string | undefined @@ -270,6 +286,7 @@ export async function runVerifiedResearchLoop( // pages — but only when at least one source survived verification, so a // page never cites a rejected source. const acceptedWorkerSources = await registerSources(options, accepted) + await confirmRegisteredSources(options.driver, acceptedWorkerSources) const writtenPages: string[] = [] writtenPages.push( ...(await applyPages(options.root, workerContribution, acceptedWorkerSources)), @@ -295,6 +312,7 @@ export async function runVerifiedResearchLoop( }) driverNotes = driverContribution.notes driverSources = await registerSources(options, driverContribution.sources ?? []) + await confirmRegisteredSources(options.driver, driverSources) writtenPages.push(...(await applyPages(options.root, driverContribution, driverSources))) index = await buildKnowledgeIndex(options.root) readiness = readinessFor(options, index) @@ -303,7 +321,12 @@ export async function runVerifiedResearchLoop( // 4. DRIVER GATES on readiness and folds the remainder into the next prompt. ready = isReady(readiness?.report) const remainingGaps = gapsFromReadiness(readiness) - steer = ready ? undefined : foldGaps(options.driver, remainingGaps) + if (ready || remainingGaps.length === 0) { + steer = undefined + } else { + await options.driver.prepareFold?.() + steer = foldGaps(options.driver, remainingGaps) + } const step: VerifiedResearchRound = { round, @@ -331,6 +354,11 @@ export async function runVerifiedResearchLoop( }), notes: { worker: workerContribution.notes, driver: driverNotes }, } + // Commit the driver's state before publishing the round event. A persisted + // event therefore never claims a round whose generated questions were lost. + await options.driver.checkpoint?.() + await store.putEvent(step.event) + steps.push(step) await options.onRound?.(step) } @@ -405,6 +433,18 @@ async function registerSources( return records } +async function confirmRegisteredSources( + driver: ResearchDriver, + sources: readonly SourceRecord[], +): Promise { + if (!driver.commitSources) return + const originalUris = sources.flatMap((source) => + typeof source.metadata?.originalUri === 'string' ? [source.metadata.originalUri] : [], + ) + if (originalUris.length === 0) return + await driver.commitSources([...new Set(originalUris)].sort()) +} + /** * Apply a contribution's curated pages. Static `proposalText` plus a * `buildPages(acceptedSources)` result are concatenated and run through the safe diff --git a/tests/claim-persistence.test.ts b/tests/claim-persistence.test.ts new file mode 100644 index 0000000..3a261cd --- /dev/null +++ b/tests/claim-persistence.test.ts @@ -0,0 +1,1172 @@ +import { access, mkdir, mkdtemp, readdir, rm, symlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' +import type { + ResearchClaimEvidence, + ResearchClaimLedger, + ResearchClaimRecord, + ResearchSourceProposal, + SourceVerificationContext, +} from '../src/index' +import { + buildKnowledgeIndex, + ClaimLedgerGoalConflictError, + claimEvidenceId, + claimId, + createKnowledgeEvent, + createPersistentResearchDrivingDriver, + createResearchDrivingDriver, + DeepQuestionSchema, + deepQuestionId, + defineReadinessSpec, + FileSystemKbStore, + initKnowledgeBase, + KB_CLAIM_LEDGER_DIR, + KB_STORE_DIR, + KNOWLEDGE_EVENT_TYPES, + KnowledgeEventSchema, + linkClaimContradictions, + MemoryKbStore, + mergeClaimLedgers, + ResearchClaimLedgerSchema, + runVerifiedResearchLoop, + withSafeDescendant, + writeFileDurable, + writeJsonDurableWithinRoot, + writeKnowledgeIndex, +} from '../src/index' +import type { RouterClient } from '../src/web-research-worker' + +const GOAL = 'self-speculative decoding' + +async function withRoot(fn: (root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'agent-knowledge-claims-')) + try { + await fn(root) + } finally { + await rm(root, { recursive: true, force: true }) + } +} + +/** A RouterClient whose `chat` returns scripted claim-extraction JSON by token. */ +function stubRouter(repliesByToken: Record): RouterClient { + return { + search: async () => [], + chat: async (messages) => { + const user = messages.find((message) => message.role === 'user')?.content ?? '' + for (const [token, reply] of Object.entries(repliesByToken)) { + if (user.includes(token)) return reply + } + return '[]' + }, + usage: () => ({ + chatCalls: 0, + searchCalls: 0, + promptTokens: 0, + completionTokens: 0, + usd: 0, + wallMs: 0, + }), + } +} + +function ctx(round: number, goal = GOAL): SourceVerificationContext { + return { + root: '/tmp/x', + goal, + round, + index: { + root: '/tmp/x', + generatedAt: '', + sources: [], + pages: [], + graph: { nodes: [], edges: [] }, + }, + gaps: [], + acceptedThisRound: [], + } +} + +function source(uri: string, text: string): ResearchSourceProposal { + return { uri, text, title: uri } +} + +const CLAIM_A = '[{"claim":"layer skipping gives a 1.73x speedup","contradicts":null}]' + +// =========================================================================== +// Claims must survive the process that discovered them. +// =========================================================================== + +describe('research claim ledger — persistence', () => { + it('restores corroboration counts and open questions in a NEW driver instance', async () => { + const store = new MemoryKbStore() + const router = stubRouter({ 'PAGE-A': CLAIM_A, 'PAGE-B': CLAIM_A }) + + const first = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-1' }) + await first.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await first.commitSources(['https://arxiv.org/a']) + await first.prepareFold() + first.foldGaps?.([]) + await first.checkpoint() + + const before = first.researchState() + expect(before.claims).toHaveLength(1) + expect(before.claims[0]?.supportingHosts).toEqual(new Set(['arxiv.org'])) + expect(before.openQuestions.length).toBeGreaterThan(0) + + // The process dies here. A NEW driver over the same store is the resume. + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'run-1', + }) + const restored = resumed.researchState() + expect(restored.claims).toHaveLength(1) + expect(restored.claims[0]?.supportingHosts).toEqual(new Set(['arxiv.org'])) + expect(restored.weaklySupported).toHaveLength(1) + expect(restored.questions.map((question) => question.id).sort()).toEqual( + before.questions.map((question) => question.id).sort(), + ) + expect(restored.rounds).toBe(1) + + // And the resumed run keeps ACCUMULATING onto the restored ledger rather + // than starting a second, parallel belief state. + await resumed.verifySource(source('https://acm.org/b', 'PAGE-B body'), ctx(2)) + await resumed.commitSources(['https://acm.org/b']) + const grown = resumed.researchState() + expect(grown.claims).toHaveLength(1) + expect([...(grown.claims[0]?.supportingHosts ?? [])].sort()).toEqual(['acm.org', 'arxiv.org']) + expect(grown.corroborated).toHaveLength(1) + }) + + it('keeps accepted evidence pending until its source registration is confirmed', async () => { + const store = new MemoryKbStore() + const router = stubRouter({ 'PAGE-A': CLAIM_A, 'PAGE-B': CLAIM_A }) + const driver = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'mid-round', + }) + + // Verification persists the extraction first, but acceptance is not proof + // that the separate source-registry write completed. + const verdict = await driver.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + expect(verdict.accept).toBe(true) + + const pending = await store.getClaimLedger('mid-round') + expect(pending?.claimEvidence).toHaveLength(1) + expect(pending?.registeredSourceUris).toEqual([]) + expect(pending?.claims).toEqual([]) + expect(driver.isComplete()).toBe(false) + + await driver.commitSources(['https://arxiv.org/a']) + const afterFirst = await store.getClaimLedger('mid-round') + expect(afterFirst?.claims[0]?.supportingHosts).toEqual(['arxiv.org']) + + await driver.verifySource(source('https://acm.org/b', 'PAGE-B body'), ctx(1)) + const secondPending = await store.getClaimLedger('mid-round') + expect(secondPending?.registeredSourceUris).toEqual(['https://arxiv.org/a']) + expect(secondPending?.claims[0]?.supportingHosts).toEqual(['arxiv.org']) + + await driver.commitSources(['https://acm.org/b']) + const afterSecond = await store.getClaimLedger('mid-round') + expect([...(afterSecond?.claims[0]?.supportingHosts ?? [])].sort()).toEqual([ + 'acm.org', + 'arxiv.org', + ]) + + // The crash lands here, between sources and before the round ends. + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'mid-round', + }) + expect(resumed.researchState().corroborated).toHaveLength(1) + }) + + it('recovers when sources register but the evidence confirmation crashes', async () => { + await withRoot(async (root) => { + const store = new FileSystemKbStore({ root }) + const router = stubRouter({ BODY: CLAIM_A }) + const driver = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'source-confirmation-crash', + }) + const interrupted = { + ...driver, + commitSources: async (sourceUris: readonly string[]) => { + if (sourceUris.length > 0) { + throw new Error('simulated crash after source registration') + } + await driver.commitSources(sourceUris) + }, + } + const readinessSpecs = [ + defineReadinessSpec({ + id: 'two-results', + description: 'two independent results', + query: 'BODY', + requiredFor: ['ResearchAgent'], + importance: 'blocking', + minSources: 2, + minHits: 1, + }), + ] + + await expect( + runVerifiedResearchLoop({ + root, + goal: GOAL, + maxRounds: 1, + readinessSpecs, + worker: async () => ({ + sources: [ + source('https://a.org/result', 'BODY one'), + source('https://b.org/result', 'BODY two'), + ], + }), + driver: interrupted, + }), + ).rejects.toThrow(/simulated crash after source registration/) + + // Exact failed state: both source writes completed, while both claim + // observations remain pending and therefore cannot report completion. + const registeredIndex = await buildKnowledgeIndex(root) + expect(registeredIndex.sources).toHaveLength(2) + const pending = await store.getClaimLedger('source-confirmation-crash') + expect(pending?.claimEvidence).toHaveLength(2) + expect(pending?.registeredSourceUris).toEqual([]) + expect(pending?.claims).toEqual([]) + const afterCrash = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'source-confirmation-crash', + }) + expect(afterCrash.isComplete()).toBe(false) + await expect(store.listEvents({ type: 'research.iteration' })).resolves.toHaveLength(0) + + // A fresh loop reconciles exact original URIs from the source registry + // before it decides readiness or launches another worker round. + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'source-confirmation-crash', + }) + let completeBeforeWorker = false + await expect( + runVerifiedResearchLoop({ + root, + goal: GOAL, + maxRounds: 1, + readinessSpecs, + worker: async () => { + completeBeforeWorker = resumed.isComplete() + throw new Error('stop after restart reconciliation') + }, + driver: resumed, + }), + ).rejects.toThrow(/stop after restart reconciliation/) + expect(completeBeforeWorker).toBe(true) + expect(resumed.isComplete()).toBe(true) + const recovered = await store.getClaimLedger('source-confirmation-crash') + expect(recovered?.registeredSourceUris).toEqual([ + 'https://a.org/result', + 'https://b.org/result', + ]) + expect(recovered?.claims[0]?.supportingHosts).toEqual(['a.org', 'b.org']) + }) + }) + + it('survives JSON round-tripping — the failure that made the old ledger unstorable', async () => { + const router = stubRouter({ + SEED: '[{"claim":"the speedup is 5x","contradicts":null}]', + REFUTE: '[{"claim":"the speedup is only 2x","contradicts":"[__ID__]"}]', + }) + const driver = createResearchDrivingDriver({ + router: { + ...router, + chat: async (messages) => { + const user = messages.find((message) => message.role === 'user')?.content ?? '' + if (user.includes('SEED')) return '[{"claim":"the speedup is 5x","contradicts":null}]' + if (user.includes('REFUTE')) { + const id = user.match(/\[(c_[0-9a-f]+)\]/)?.[1] ?? '' + return `[{"claim":"the speedup is only 2x","contradicts":"[${id}]"}]` + } + return '[]' + }, + }, + }) + await driver.verifySource(source('https://a.org/x', 'SEED'), ctx(1)) + await driver.verifySource(source('https://b.org/y', 'REFUTE'), ctx(1)) + + const live = driver.researchState() + expect(live.contested).toHaveLength(2) + expect(live.claims[0]?.supportingHosts).toBeInstanceOf(Set) + + // The published live API retains Sets; the durable ledger converts them to + // arrays at the persistence boundary so JSON cannot erase their contents. + const durable = driver.toLedger() + const roundTripped = JSON.parse(JSON.stringify(durable)) as typeof durable + expect(roundTripped.claims[0]?.supportingHosts).toEqual(durable.claims[0]?.supportingHosts) + expect(roundTripped.claims[0]?.supportingHosts.length).toBe(1) + expect(roundTripped.claims[1]?.contradicts).toEqual(durable.claims[1]?.contradicts) + expect(roundTripped.claims[1]?.contradicts.length).toBe(1) + for (const claim of roundTripped.claims) { + expect(Array.isArray(claim.supportingHosts)).toBe(true) + expect(Array.isArray(claim.contradicts)).toBe(true) + } + }) + + it('does not report a resumed run complete while its questions are still open', async () => { + const store = new MemoryKbStore() + const router = stubRouter({ 'PAGE-A': CLAIM_A, 'PAGE-B': CLAIM_A }) + + const first = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-2' }) + await first.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await first.commitSources(['https://arxiv.org/a']) + // The questions this raises exist ONLY in the synchronous fold, so without a + // checkpoint they would die here and the resumed run would call itself done. + await first.prepareFold() + first.foldGaps?.([]) + await first.checkpoint() + expect(first.isComplete()).toBe(false) + + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'run-2', + }) + expect(resumed.researchState().openQuestions.length).toBeGreaterThan(0) + expect(resumed.isComplete()).toBe(false) + + // Corroborating the claim settles the CLAIM half, and completion still + // refuses while a question raised before the crash remains unanswered. + await resumed.verifySource(source('https://acm.org/b', 'PAGE-B body'), ctx(2)) + await resumed.commitSources(['https://acm.org/b']) + const settled = resumed.researchState() + expect(settled.corroborated).toHaveLength(1) + expect(settled.weaklySupported).toHaveLength(0) + expect(settled.openQuestions.length).toBeGreaterThan(0) + expect(resumed.isComplete()).toBe(false) + + // Why that matters, stated as its own violation: a ledger that kept the + // claims but LOST the questions — which is what a driver without + // `checkpoint` writes — reports the very same run complete. + const lossy = (await store.getClaimLedger('run-2'))! + await store.putClaimLedger({ + ...lossy, + claims: resumed.toLedger().claims, + questions: [], + }) + const lied = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-2' }) + expect(lied.isComplete()).toBe(true) + }) + + it('recovers a fold interrupted before checkpoint without publishing its event', async () => { + await withRoot(async (root) => { + const store = new FileSystemKbStore({ root }) + const router = stubRouter({ + BODY: '[{"claim":"the method is 2x faster","contradicts":null}]', + }) + const driver = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'crashed-fold', + }) + const interrupted = { + ...driver, + checkpoint: async () => { + throw new Error('simulated crash before checkpoint') + }, + } + + await expect( + runVerifiedResearchLoop({ + root, + goal: GOAL, + maxRounds: 1, + readinessSpecs: [ + defineReadinessSpec({ + id: 'still-open', + description: 'a result the sources do not close', + query: 'never-present', + requiredFor: ['ResearchAgent'], + importance: 'blocking', + minSources: 3, + minHits: 1, + }), + ], + worker: async () => ({ + sources: [ + source('https://a.org/result', 'BODY one'), + source('https://b.org/result', 'BODY two'), + ], + }), + driver: interrupted, + }), + ).rejects.toThrow(/simulated crash/) + + const pending = await store.getClaimLedger('crashed-fold') + expect(pending?.rounds).toBe(0) + expect(pending?.preparedRounds).toBe(1) + await expect(store.listEvents({ type: 'research.iteration' })).resolves.toHaveLength(0) + + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'crashed-fold', + }) + expect(resumed.researchState().corroborated).toHaveLength(1) + expect(resumed.researchState().openQuestions.length).toBeGreaterThan(0) + expect(resumed.isComplete()).toBe(false) + const recovered = await store.getClaimLedger('crashed-fold') + expect(recovered?.rounds).toBe(1) + expect(recovered?.preparedRounds).toBeUndefined() + expect(recovered?.questions.length).toBeGreaterThan(0) + }) + }) + + it('refuses to merge two research goals into one ledger', async () => { + const store = new MemoryKbStore() + const router = stubRouter({ 'PAGE-A': CLAIM_A }) + const first = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-3' }) + await first.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await first.checkpoint() + + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'run-3', + }) + await expect( + resumed.verifySource(source('https://acm.org/b', 'PAGE-A body'), ctx(1, 'a different goal')), + ).rejects.toThrow(/cannot be reused/) + }) + + it('rejects a ledger id that would escape its directory', async () => { + const store = new MemoryKbStore() + const router = stubRouter({}) + for (const ledgerId of ['../escape', 'nested/id', '..', '.', '', 'a\0b']) { + await expect( + createPersistentResearchDrivingDriver({ router, store, ledgerId }), + ).rejects.toThrow(/claim ledger id/) + await expect(store.getClaimLedger(ledgerId)).rejects.toThrow(/claim ledger id/) + } + }) + + it('keeps two runs against one knowledge base from overwriting each other', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + const store = new FileSystemKbStore({ root }) + const router = stubRouter({ + 'PAGE-A': CLAIM_A, + 'PAGE-C': '[{"claim":"a different claim about caches","contradicts":null}]', + }) + + const runA = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-a' }) + await runA.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1, 'goal A')) + await runA.commitSources(['https://arxiv.org/a']) + const runB = await createPersistentResearchDrivingDriver({ router, store, ledgerId: 'run-b' }) + await runB.verifySource(source('https://acm.org/c', 'PAGE-C body'), ctx(1, 'goal B')) + await runB.commitSources(['https://acm.org/c']) + + const ledgers = await store.listClaimLedgers() + expect(ledgers.map((ledger) => ledger.id)).toEqual(['run-a', 'run-b']) + expect(ledgers[0]?.goal).toBe('goal A') + expect(ledgers[1]?.goal).toBe('goal B') + expect(ledgers[0]?.claims[0]?.text).not.toBe(ledgers[1]?.claims[0]?.text) + + // On disk, under the one store directory, one file per run. + const files = await readdir(join(root, KB_CLAIM_LEDGER_DIR)) + expect(files.sort()).toEqual(['run-a.json', 'run-b.json']) + }) + }) + + it('persists a claim ledger to disk that a fresh store instance reads back', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + const router = stubRouter({ 'PAGE-A': CLAIM_A }) + const driver = await createPersistentResearchDrivingDriver({ + router, + store: new FileSystemKbStore({ root }), + ledgerId: 'run-disk', + }) + await driver.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await driver.commitSources(['https://arxiv.org/a']) + await driver.prepareFold() + driver.foldGaps?.([]) + await driver.checkpoint() + + // A different store object over the same root — the durable read path. + const reader = new FileSystemKbStore({ root }) + const ledger = await reader.getClaimLedger('run-disk') + expect(ledger?.claims).toHaveLength(1) + expect(ledger?.claims[0]?.supportingHosts).toEqual(['arxiv.org']) + expect(ledger?.questions.length).toBeGreaterThan(0) + expect(ledger?.goal).toBe(GOAL) + expect(await reader.getClaimLedger('never-written')).toBeNull() + }) + }) + + it('checkpoint on a store-less driver is a no-op rather than a silent write', async () => { + const driver = createResearchDrivingDriver({ router: stubRouter({ 'PAGE-A': CLAIM_A }) }) + await driver.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await expect(driver.checkpoint()).resolves.toBeUndefined() + expect(driver.toLedger().id).toBe('in-memory') + expect(driver.toLedger().claims).toHaveLength(1) + }) +}) + +// =========================================================================== +// One store, one index file, and an event log with a producer. +// =========================================================================== + +describe('knowledge store — one writer, one location', () => { + it('shows the indexer’s work through the store, and writes exactly one index file', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + await writeFile(join(root, 'knowledge', 'page.md'), '# Page\n\nBody text.\n') + + const built = await writeKnowledgeIndex(root) + const store = new FileSystemKbStore({ root }) + const stored = await store.getIndex() + + // The exact reproduction that used to resolve to `null`. + expect(stored).not.toBeNull() + expect(stored?.pages.map((page) => page.path)).toEqual(built.pages.map((page) => page.path)) + + // Violation attempt: no SECOND index file anywhere under the root. + const found = await findFiles(root, 'index.json') + expect(found).toEqual([join(root, '.agent-knowledge', 'index.json')]) + await expect(access(join(root, 'index.json'))).rejects.toThrow() + }) + }) + + it('accepts every event type the package declares', async () => { + await withRoot(async (root) => { + const store = new FileSystemKbStore({ root }) + for (const type of KNOWLEDGE_EVENT_TYPES) { + const event = createKnowledgeEvent({ type, target: `target-${type}` }) + expect(() => KnowledgeEventSchema.parse(event)).not.toThrow() + await store.putEvent(event) + } + const stored = await store.listEvents() + expect(stored.map((event) => event.type).sort()).toEqual([...KNOWLEDGE_EVENT_TYPES].sort()) + }) + }) + + it('records the research loop’s round events instead of discarding them', async () => { + await withRoot(async (root) => { + const result = await runVerifiedResearchLoop({ + root, + goal: GOAL, + maxRounds: 2, + actor: 'test', + worker: async ({ round }) => ({ + sources: [source(`https://arxiv.org/r${round}`, `body for round ${round}`)], + }), + driver: { verifySource: () => ({ accept: true }) }, + }) + expect(result.rounds).toBe(2) + + const stored = await new FileSystemKbStore({ root }).listEvents({ + type: 'research.iteration', + }) + expect(stored).toHaveLength(2) + expect(stored.map((event) => event.metadata?.round)).toEqual([1, 2]) + expect(stored.every((event) => event.actor === 'test')).toBe(true) + }) + }) + + it('checkpoints the driver through a real loop so the run resumes from disk', async () => { + await withRoot(async (root) => { + const store = new FileSystemKbStore({ root }) + const router = stubRouter({ 'body for round': CLAIM_A }) + const driver = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'loop-run', + }) + + // A readiness spec the single source cannot satisfy keeps the loop + // not-ready, which is what makes it fold steer — the driver's synchronous + // question-raising hook, whose output only reaches disk via `checkpoint`. + await runVerifiedResearchLoop({ + root, + goal: GOAL, + maxRounds: 1, + readinessSpecs: [ + defineReadinessSpec({ + id: 'topic/definition', + description: 'what the method is and how it works', + query: 'body for round', + requiredFor: ['ResearchAgent'], + importance: 'blocking', + minSources: 2, + minHits: 1, + }), + ], + worker: async ({ round }) => ({ + sources: [source(`https://arxiv.org/r${round}`, `body for round ${round}`)], + }), + driver, + }) + + const resumed = await createPersistentResearchDrivingDriver({ + router, + store, + ledgerId: 'loop-run', + }) + const state = resumed.researchState() + expect(state.claims).toHaveLength(1) + expect(state.claims[0]?.supportingHosts).toEqual(new Set(['arxiv.org'])) + expect(state.rounds).toBe(1) + expect(state.openQuestions.length).toBeGreaterThan(0) + }) + }) +}) + +// =========================================================================== +// durable-fs is reachable, and still refuses to be redirected. +// =========================================================================== + +describe('durable-fs on the package entrypoint', () => { + it('exports the durable write primitives', () => { + expect(typeof writeFileDurable).toBe('function') + expect(typeof writeJsonDurableWithinRoot).toBe('function') + expect(typeof withSafeDescendant).toBe('function') + }) + + it('still refuses a write redirected through a symbolic link', async () => { + await withRoot(async (root) => { + const outside = join(root, 'outside') + const base = join(root, 'base') + await mkdir(outside, { recursive: true }) + await mkdir(base, { recursive: true }) + await symlink(outside, join(base, 'records')) + + await expect( + writeJsonDurableWithinRoot(base, 'records/leak.json', { leaked: true }), + ).rejects.toThrow(/unsafe directory/) + await expect(access(join(outside, 'leak.json'))).rejects.toThrow() + + // The traversal guard is on the relative path itself, too. + await expect(writeJsonDurableWithinRoot(base, '../escape.json', {})).rejects.toThrow( + /unsafe segment/, + ) + }) + }) + + it('replaces a file atomically, leaving no temporary behind', async () => { + await withRoot(async (root) => { + const path = join(root, 'record.json') + await writeFileDurable(path, '{"generation":1}\n', { encoding: 'utf8' }) + await writeFileDurable(path, '{"generation":2}\n', { encoding: 'utf8' }) + expect(await readdir(root)).toEqual(['record.json']) + }) + }) +}) + +// =========================================================================== +// Persisting is not enough: two writers must ACCUMULATE, not overwrite. +// =========================================================================== + +function ledgerOf( + id: string, + claims: readonly ResearchClaimRecord[], + goal = GOAL, +): ResearchClaimLedger { + return { + id, + goal, + updatedAt: '2026-07-28T00:00:00.000Z', + rounds: 1, + claimEvidence: [], + registeredSourceUris: [...new Set(claims.flatMap((claim) => claim.supportingUris))].sort(), + claims: [...claims].sort((a, b) => a.id.localeCompare(b.id)), + questions: [], + } +} + +function claimFrom(text: string, host: string, round = 1): ResearchClaimRecord { + return { + id: claimId(text), + text, + supportingHosts: [host], + supportingUris: [`https://${host}/x`], + contradicts: [], + contested: false, + firstSeenRound: round, + } +} + +function evidenceFrom( + text: string, + sourceUri: string, + round = 1, + contradictsClaimId?: string, +): ResearchClaimEvidence { + const observedClaimId = claimId(text) + return { + id: claimEvidenceId({ claimId: observedClaimId, sourceUri, contradictsClaimId }), + claimId: observedClaimId, + text, + sourceUri, + ...(contradictsClaimId === undefined ? {} : { contradictsClaimId }), + firstSeenRound: round, + } +} + +describe('claim ledger — concurrent accumulation', () => { + /** + * The negative control for the whole merge path. If `putClaimLedger` did not + * lose a concurrent writer's claims, `mergeClaimLedger` would be ceremony — + * so the loss is asserted here, and the next test asserts the fix. Weakening + * either one makes the pair vacuous. + */ + it('loses a concurrent writer’s claims when each writes the whole ledger', async () => { + const store = new MemoryKbStore() + const mine = claimFrom('layer skipping gives a 1.73x speedup', 'arxiv.org') + const theirs = claimFrom('draft heads cost 8% of parameters', 'acm.org') + + // Both read the empty ledger, then both write what they built from it. + const readByA = await store.getClaimLedger('shared') + const readByB = await store.getClaimLedger('shared') + expect(readByA).toBeNull() + expect(readByB).toBeNull() + await store.putClaimLedger(ledgerOf('shared', [mine])) + await store.putClaimLedger(ledgerOf('shared', [theirs])) + + const after = await store.getClaimLedger('shared') + expect(after?.claims.map((claim) => claim.text)).toEqual([theirs.text]) + }) + + it('keeps both writers’ claims when each merges', async () => { + const store = new MemoryKbStore() + const mine = claimFrom('layer skipping gives a 1.73x speedup', 'arxiv.org') + const theirs = claimFrom('draft heads cost 8% of parameters', 'acm.org') + + for (const claim of [mine, theirs]) { + await store.mergeClaimLedger('shared', (current) => + current === null + ? ledgerOf('shared', [claim]) + : mergeClaimLedgers(current, ledgerOf('shared', [claim])), + ) + } + + const after = await store.getClaimLedger('shared') + expect(after?.claims.map((claim) => claim.text).sort()).toEqual([mine.text, theirs.text].sort()) + }) + + it('grows one claim’s independent-source count across separate writers', async () => { + const store = new MemoryKbStore() + const text = 'layer skipping gives a 1.73x speedup' + for (const host of ['arxiv.org', 'acm.org', 'arxiv.org']) { + await store.mergeClaimLedger('shared', (current) => { + const incoming = ledgerOf('shared', [claimFrom(text, host)]) + return current === null ? incoming : mergeClaimLedgers(current, incoming) + }) + } + + const after = await store.getClaimLedger('shared') + expect(after?.claims).toHaveLength(1) + // Two DISTINCT hosts, and the repeat did not inflate the count — that count + // is the corroboration threshold, so double-counting one host would report + // an unconfirmed claim as independently confirmed. + expect(after?.claims[0]?.supportingHosts.sort()).toEqual(['acm.org', 'arxiv.org']) + }) + + it('serialises concurrent merges on disk so no writer’s claim is dropped', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + const hosts = ['a.org', 'b.org', 'c.org', 'd.org', 'e.org', 'f.org'] + // A separate store instance per writer: same root, no shared memory, which + // is what two workers in two processes look like to the filesystem. + await Promise.all( + hosts.map((host) => + new FileSystemKbStore({ root }).mergeClaimLedger('pursuit', (current) => { + const incoming = ledgerOf('pursuit', [claimFrom(`claim from ${host}`, host)]) + return current === null ? incoming : mergeClaimLedgers(current, incoming) + }), + ), + ) + + const after = await new FileSystemKbStore({ root }).getClaimLedger('pursuit') + expect(after?.claims.map((claim) => claim.text).sort()).toEqual( + hosts.map((host) => `claim from ${host}`).sort(), + ) + }) + }) + + it('uses one lock when legacy and root constructors address the same files', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + const rootStore = new FileSystemKbStore({ root }) + const legacyStore = new FileSystemKbStore(join(root, KB_STORE_DIR)) + const hosts = Array.from({ length: 12 }, (_, index) => `host-${index}.org`) + + await Promise.all( + hosts.map((host, index) => { + const store = index % 2 === 0 ? rootStore : legacyStore + return store.mergeClaimLedger('aliased', (current) => { + const incoming = ledgerOf('aliased', [claimFrom(`claim ${index}`, host)]) + return current === null ? incoming : mergeClaimLedgers(current, incoming) + }) + }), + ) + + const stored = await rootStore.getClaimLedger('aliased') + expect(stored?.claims).toHaveLength(hosts.length) + }) + }) + + it('two persistent drivers on one ledger see each other’s corroboration', async () => { + await withRoot(async (root) => { + await initKnowledgeBase(root) + const router = stubRouter({ 'PAGE-A': CLAIM_A, 'PAGE-B': CLAIM_A }) + + const workerThree = await createPersistentResearchDrivingDriver({ + router, + store: new FileSystemKbStore({ root }), + ledgerId: 'pursuit', + }) + const workerForty = await createPersistentResearchDrivingDriver({ + router, + store: new FileSystemKbStore({ root }), + ledgerId: 'pursuit', + }) + + await workerThree.verifySource(source('https://arxiv.org/a', 'PAGE-A body'), ctx(1)) + await workerThree.commitSources(['https://arxiv.org/a']) + await workerForty.verifySource(source('https://acm.org/b', 'PAGE-B body'), ctx(1)) + await workerForty.commitSources(['https://acm.org/b']) + + // Worker 40 wrote second and read worker 3's evidence back: one claim, + // two independent hosts, corroborated. Under a whole-record write worker + // 40 would report one host and the claim would still be weak. + const seen = workerForty.researchState() + expect(seen.claims).toHaveLength(1) + expect([...(seen.claims[0]?.supportingHosts ?? [])].sort()).toEqual(['acm.org', 'arxiv.org']) + expect(seen.corroborated).toHaveLength(1) + expect(workerForty.isComplete()).toBe(true) + }) + }) + + it('refuses a merge that returns a ledger under a different id', async () => { + const store = new MemoryKbStore() + await expect( + store.mergeClaimLedger('pursuit', () => ledgerOf('somewhere-else', [])), + ).rejects.toThrow(/returned a ledger with id 'somewhere-else'/) + expect(await store.getClaimLedger('pursuit')).toBeNull() + + await withRoot(async (root) => { + await initKnowledgeBase(root) + const fileStore = new FileSystemKbStore({ root }) + await expect( + fileStore.mergeClaimLedger('pursuit', () => ledgerOf('somewhere-else', [])), + ).rejects.toThrow(/returned a ledger with id 'somewhere-else'/) + expect(await fileStore.getClaimLedger('pursuit')).toBeNull() + expect(await fileStore.getClaimLedger('somewhere-else')).toBeNull() + }) + }) + + it('refuses to pool evidence gathered for two different goals', () => { + const base = ledgerOf('pursuit', [claimFrom('x speeds up y', 'a.org')], 'speculative decoding') + const other = ledgerOf('pursuit', [claimFrom('x speeds up y', 'b.org')], 'quantization') + expect(() => mergeClaimLedgers(base, other)).toThrow(ClaimLedgerGoalConflictError) + // The claim would otherwise have read as corroborated by two independent + // hosts, on evidence collected for two unrelated questions. + expect(() => mergeClaimLedgers(base, other)).toThrow(/'speculative decoding'/) + }) + + it('is order-independent and idempotent, so a replayed write changes nothing', () => { + const a = ledgerOf('pursuit', [claimFrom('claim one', 'a.org', 3)]) + const b = ledgerOf('pursuit', [claimFrom('claim one', 'b.org', 1), claimFrom('two', 'b.org')]) + + const ab = mergeClaimLedgers(a, b) + const ba = mergeClaimLedgers(b, a) + expect(ab).toEqual(ba) + expect(mergeClaimLedgers(ab, b)).toEqual(ab) + expect(mergeClaimLedgers(ab, a)).toEqual(ab) + // The earliest round a claim was seen in survives the merge; a later + // sighting must not make the claim look newer than it is. + expect(ab.claims.find((claim) => claim.id === claimId('claim one'))?.firstSeenRound).toBe(1) + }) + + it('is associative across three independently accumulated ledgers', () => { + const a = ledgerOf('pursuit', [claimFrom('claim one', 'a.org', 3)]) + const b = ledgerOf('pursuit', [claimFrom('claim one', 'b.org', 1)]) + const c = ledgerOf('pursuit', [claimFrom('claim two', 'c.org', 2)]) + + expect(mergeClaimLedgers(mergeClaimLedgers(a, b), c)).toEqual( + mergeClaimLedgers(a, mergeClaimLedgers(b, c)), + ) + }) + + it('materializes split evidence and source confirmation in either merge order', () => { + const sourceUri = 'https://a.org/result' + const evidence = evidenceFrom('claim one', sourceUri) + const observed = { ...ledgerOf('pursuit', []), claimEvidence: [evidence] } + const registered = { ...ledgerOf('pursuit', []), registeredSourceUris: [sourceUri] } + + const evidenceThenRegistration = mergeClaimLedgers(observed, registered) + const registrationThenEvidence = mergeClaimLedgers(registered, observed) + expect(evidenceThenRegistration).toEqual(registrationThenEvidence) + expect(evidenceThenRegistration.claims[0]?.supportingUris).toEqual([sourceUri]) + expect(mergeClaimLedgers(evidenceThenRegistration, observed)).toEqual(evidenceThenRegistration) + expect(mergeClaimLedgers(evidenceThenRegistration, registered)).toEqual( + evidenceThenRegistration, + ) + }) + + it('keeps the two-phase closure associative across independent writers', () => { + const uriOne = 'https://a.org/result' + const uriTwo = 'https://b.org/result' + const evidence = { + ...ledgerOf('pursuit', []), + claimEvidence: [ + evidenceFrom('claim one', uriOne, 2), + evidenceFrom('claim one', uriTwo, 1), + ].sort((left, right) => left.id.localeCompare(right.id)), + } + const firstRegistration = { + ...ledgerOf('pursuit', []), + registeredSourceUris: [uriOne], + } + const secondRegistration = { + ...ledgerOf('pursuit', []), + registeredSourceUris: [uriTwo], + } + + const left = mergeClaimLedgers( + mergeClaimLedgers(evidence, firstRegistration), + secondRegistration, + ) + const right = mergeClaimLedgers( + evidence, + mergeClaimLedgers(firstRegistration, secondRegistration), + ) + expect(left).toEqual(right) + expect(left.claims[0]?.supportingHosts).toEqual(['a.org', 'b.org']) + }) + + it('does not contest a claim against an unregistered counterpart', () => { + const originalUri = 'https://a.org/original' + const refuterUri = 'https://b.org/refuter' + const original = evidenceFrom('the speedup is 5x', originalUri) + const refuter = evidenceFrom('the speedup is only 2x', refuterUri, 1, original.claimId) + const observed = { + ...ledgerOf('pursuit', []), + claimEvidence: [original, refuter].sort((left, right) => left.id.localeCompare(right.id)), + } + const onlyRefuterRegistered = { + ...ledgerOf('pursuit', []), + registeredSourceUris: [refuterUri], + } + + const oneSided = mergeClaimLedgers(observed, onlyRefuterRegistered) + expect(oneSided.claims).toHaveLength(1) + expect(oneSided.claims[0]?.contested).toBe(false) + expect(oneSided.claims[0]?.contradicts).toEqual([]) + + const bothRegistered = mergeClaimLedgers(oneSided, { + ...ledgerOf('pursuit', []), + registeredSourceUris: [originalUri], + }) + expect(bothRegistered.claims).toHaveLength(2) + expect(bothRegistered.claims.every((claim) => claim.contested)).toBe(true) + expect(bothRegistered.claims.every((claim) => claim.contradicts.length === 1)).toBe(true) + }) + + it('does not merge opposite directional or polarity claims', () => { + for (const [left, right] of [ + ['accuracy > 90%', 'accuracy < 90%'], + ['effect is +5%', 'effect is -5%'], + ['result is x + y', 'result is x - y'], + ['result ≥ baseline', 'result ≤ baseline'], + ]) { + expect(claimId(left)).not.toBe(claimId(right)) + const merged = mergeClaimLedgers( + ledgerOf('pursuit', [claimFrom(left, 'a.org')]), + ledgerOf('pursuit', [claimFrom(right, 'b.org')]), + ) + expect(merged.claims).toHaveLength(2) + expect(merged.claims.every((claim) => claim.supportingHosts.length === 1)).toBe(true) + } + }) + + it('uses a deterministic wording when equal-round writers spell one claim differently', () => { + const upper = claimFrom('Layer skipping gives a 1.73x speedup!', 'a.org') + const lower = claimFrom('layer skipping gives a 1 73x speedup', 'b.org') + expect(upper.id).toBe(lower.id) + + const forward = mergeClaimLedgers(ledgerOf('pursuit', [upper]), ledgerOf('pursuit', [lower])) + const reverse = mergeClaimLedgers(ledgerOf('pursuit', [lower]), ledgerOf('pursuit', [upper])) + expect(forward).toEqual(reverse) + expect(forward.claims[0]?.text).toBe( + [upper.text, lower.text].sort((a, b) => (a < b ? -1 : a > b ? 1 : 0))[0], + ) + }) + + it('never clears a contradiction a later writer did not happen to see', () => { + const contested: ResearchClaimRecord = { + ...claimFrom('x speeds up y', 'a.org'), + contradicts: [claimId('x slows down y')], + contested: true, + } + const oblivious = claimFrom('x speeds up y', 'b.org') + const merged = mergeClaimLedgers(ledgerOf('p', [contested]), ledgerOf('p', [oblivious])) + expect(merged.claims[0]?.contested).toBe(true) + expect(merged.claims[0]?.contradicts).toEqual([claimId('x slows down y')]) + }) + + it('makes a one-sided contradiction symmetric and contests both ends', () => { + // Only the refuting worker knows about the disagreement: it recorded the + // edge, the original claim's writer never saw it. + const refuter: ResearchClaimRecord = { + ...claimFrom('the speedup is only 2x', 'b.org'), + contradicts: [claimId('the speedup is 5x')], + contested: true, + } + const original = claimFrom('the speedup is 5x', 'a.org') + const linked = linkClaimContradictions(ledgerOf('p', [original, refuter])) + const byId = new Map(linked.claims.map((claim) => [claim.id, claim])) + + expect(byId.get(original.id)?.contested).toBe(true) + expect(byId.get(original.id)?.contradicts).toEqual([refuter.id]) + expect(byId.get(refuter.id)?.contradicts).toEqual([original.id]) + // Idempotent: a second pass finds the edges already there. + expect(linkClaimContradictions(linked)).toEqual(linked) + }) + + it('closes a one-sided contradiction when its counterpart arrives in a merge', () => { + const original = claimFrom('the speedup is 5x', 'a.org') + const refuter: ResearchClaimRecord = { + ...claimFrom('the speedup is only 2x', 'b.org'), + contradicts: [original.id], + contested: true, + } + const merged = mergeClaimLedgers(ledgerOf('p', [refuter]), ledgerOf('p', [original])) + const byId = new Map(merged.claims.map((claim) => [claim.id, claim])) + + expect(byId.get(original.id)?.contested).toBe(true) + expect(byId.get(original.id)?.contradicts).toEqual([refuter.id]) + expect(byId.get(refuter.id)?.contradicts).toEqual([original.id]) + }) + + it('keeps an edge whose counterpart claim has not arrived yet', () => { + const orphan: ResearchClaimRecord = { + ...claimFrom('x speeds up y', 'a.org'), + contradicts: [claimId('nobody has written this down yet')], + contested: true, + } + const linked = linkClaimContradictions(ledgerOf('p', [orphan])) + // Dropping the edge would report the claim as settled on the strength of + // the one writer that had not yet met its refutation. + expect(linked.claims[0]?.contradicts).toEqual([claimId('nobody has written this down yet')]) + expect(linked.claims[0]?.contested).toBe(true) + }) +}) + +describe('claim ledger — record integrity', () => { + const claim = claimFrom('layer skipping gives a 1.73x speedup', 'arxiv.org') + const questionText = 'What independent result corroborates the speedup?' + const question = { + kind: 'gap' as const, + text: questionText, + id: deepQuestionId('gap', questionText), + claimIds: [claim.id], + addressed: false, + raisedRound: 1, + } + + it('accepts a canonical claim and question record', () => { + const ledger = { ...ledgerOf('pursuit', [claim]), questions: [question] } + expect(ResearchClaimLedgerSchema.parse(ledger)).toEqual(ledger) + expect(DeepQuestionSchema.parse(question)).toEqual(question) + }) + + it('refuses a forged claim identity and leaves the store unchanged', async () => { + const store = new MemoryKbStore() + const forged = { + ...ledgerOf('pursuit', [claim]), + claims: [{ ...claim, id: 'c_forged' }], + } + await expect(store.putClaimLedger(forged)).rejects.toThrow(/text-derived identity/) + await expect(store.getClaimLedger('pursuit')).resolves.toBeNull() + }) + + it('refuses an independent-source count not backed by source URIs', () => { + const inflated = { + ...ledgerOf('pursuit', [claim]), + claims: [{ ...claim, supportingHosts: ['acm.org', 'arxiv.org'] }], + } + expect(() => ResearchClaimLedgerSchema.parse(inflated)).toThrow( + /hosts derived from supportingUris/, + ) + }) + + it('refuses duplicate evidence, self-contradictions, and unbound questions', () => { + const duplicateEvidence = { + ...ledgerOf('pursuit', [claim]), + claims: [{ ...claim, supportingUris: [...claim.supportingUris, ...claim.supportingUris] }], + } + expect(() => ResearchClaimLedgerSchema.parse(duplicateEvidence)).toThrow( + /sorted and contain no duplicates/, + ) + + const selfContradiction = { + ...ledgerOf('pursuit', [claim]), + claims: [{ ...claim, contradicts: [claim.id], contested: true }], + } + expect(() => ResearchClaimLedgerSchema.parse(selfContradiction)).toThrow(/cannot contradict/) + + const unboundQuestion = { + ...ledgerOf('pursuit', [claim]), + questions: [{ ...question, claimIds: ['c_missing'] }], + } + expect(() => ResearchClaimLedgerSchema.parse(unboundQuestion)).toThrow(/outside its ledger/) + }) + + it('refuses forged, unregistered, or unmaterialized evidence state', () => { + const evidence = evidenceFrom(claim.text, claim.supportingUris[0]!) + const forged = { + ...ledgerOf('pursuit', []), + claimEvidence: [{ ...evidence, id: 'e_forged' }], + } + expect(() => ResearchClaimLedgerSchema.parse(forged)).toThrow(/content-derived identity/) + + const unregistered = { + ...ledgerOf('pursuit', [claim]), + registeredSourceUris: [], + } + expect(() => ResearchClaimLedgerSchema.parse(unregistered)).toThrow( + /before its registration is confirmed/, + ) + + const unmaterialized = { + ...ledgerOf('pursuit', []), + claimEvidence: [evidence], + registeredSourceUris: [evidence.sourceUri], + } + expect(() => ResearchClaimLedgerSchema.parse(unmaterialized)).toThrow(/must be materialized/) + }) + + it('refuses a question whose content is not bound to its id', () => { + expect(() => DeepQuestionSchema.parse({ ...question, text: 'Different question' })).toThrow( + /kind-and-text identity/, + ) + }) +}) + +async function findFiles(root: string, name: string): Promise { + const out: string[] = [] + for (const entry of await readdir(root, { withFileTypes: true })) { + const path = join(root, entry.name) + if (entry.isDirectory()) out.push(...(await findFiles(path, name))) + else if (entry.name === name) out.push(path) + } + return out.sort() +} diff --git a/tests/contracts/tracked-claim-compatibility.ts b/tests/contracts/tracked-claim-compatibility.ts new file mode 100644 index 0000000..98e2bd9 --- /dev/null +++ b/tests/contracts/tracked-claim-compatibility.ts @@ -0,0 +1,13 @@ +import type { ResearchClaimRecord, TrackedClaim } from '../../src/index' + +/** Published live-driver callers retain the Set operations available in 6.1.11. */ +export function inspectAndExtendTrackedClaim(claim: TrackedClaim): number { + if (!claim.supportingHosts.has('example.org')) claim.supportingHosts.add('example.org') + if (!claim.contradicts.has('c_other')) claim.contradicts.add('c_other') + return claim.supportingHosts.size + claim.contradicts.size +} + +/** Durable records remain ordinary JSON arrays. */ +export function countDurableClaimEvidence(claim: ResearchClaimRecord): number { + return claim.supportingHosts.length + claim.contradicts.length +} diff --git a/tests/core.test.ts b/tests/core.test.ts index 46d9fbb..f31f34e 100644 --- a/tests/core.test.ts +++ b/tests/core.test.ts @@ -95,16 +95,17 @@ describe('source registry integrity', () => { it('does not replace a malformed filesystem index with generated data', async () => { await withProject(async (root) => { - const storeDir = join(root, '.store') - await mkdir(storeDir, { recursive: true }) - await writeFile(join(storeDir, 'index.json'), '{broken') + const storeRoot = join(root, '.store') + const indexPath = join(storeRoot, 'index.json') + await mkdir(storeRoot, { recursive: true }) + await writeFile(indexPath, '{broken') - await expect(new FileSystemKbStore(storeDir).getIndex()).rejects.toThrow() - await expect(readFile(join(storeDir, 'index.json'), 'utf8')).resolves.toBe('{broken') + await expect(new FileSystemKbStore(storeRoot).getIndex()).rejects.toThrow() + await expect(readFile(indexPath, 'utf8')).resolves.toBe('{broken') - await writeFile(join(storeDir, 'index.json'), '{}') - await expect(new FileSystemKbStore(storeDir).getIndex()).rejects.toThrow() - await expect(readFile(join(storeDir, 'index.json'), 'utf8')).resolves.toBe('{}') + await writeFile(indexPath, '{}') + await expect(new FileSystemKbStore(storeRoot).getIndex()).rejects.toThrow() + await expect(readFile(indexPath, 'utf8')).resolves.toBe('{}') }) })