diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index aa448c0391..241ad9902f 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -8,7 +8,7 @@ {"_type":"issue","id":"polylogue-kmt1c","title":"test: close pathology-zoo anti-vacuity residuals from PR 3849","description":"Residual implementation scope from Codex review of merged PR #3849. The Claude vintage member is registered, but the merged proof train left stale manifest-size expectations, no red mutation for the new member, an overbroad vintage filename census, and a registered invariant that can pass while normalized hashes or membership decisions disagree.\n\nThis is implementation and test scope only. It does not produce a live cohort receipt or close polylogue-claude-vintage-live-proof. The production-owned pathology zoo must remain the source of truth for the red twin and maintenance invariant.\n","design":"Update the production-owned pathology-zoo contract and its focused tests. Change the manifest-size assertion to derive from the manifest or the exact registered count without making a textual fossil. Add a deterministic mutation for claude-vintage-live-proof that makes its invariant fail by changing normalized identity or membership verdict, not by merely deleting a raw row if the invariant no longer observes the intended relation. Narrow the existing vintage-reorder census to its member identity or an explicit fixture manifest so the Claude proof files cannot alter it. Keep the registered Claude invariant checking two rows, one normalized_content_hash, one applied decision, and one superseded_equivalent decision. Add an anti-vacuity test that mutates each of those semantic facts and observes a red report.\n","acceptance_criteria":"1. The expanded pathology-zoo contract passes without a stale hard-coded member count. 2. The Claude vintage member has a deterministic red mutation and the registry red-twin reaches its invariant. 3. The vintage-reorder census is scoped to its own fixture. 4. The registered Claude invariant detects hash and membership-decision drift. 5. No live mutation or live-proof closure is claimed. 6. Focused pathology-zoo/archive-verification tests and devtools verify --quick pass.","status":"open","priority":1,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:13:56Z","created_by":"Sinity","updated_at":"2026-08-06T15:13:56Z","labels":["area:verification","lane:reindex"],"dependencies":[{"issue_id":"polylogue-kmt1c","depends_on_id":"polylogue-yazae","type":"blocks","created_at":"2026-08-06T17:13:55Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-o5smo","title":"fix: preserve parser title provenance across public surfaces","description":"Residual P1 from Codex review of merged PR #3846. The title-presence gate drops genuine titles from Grok, Antigravity, browser capture, and Hermes parsers because those producers leave ParsedSession.title_source unset. The public summary then falls back to generated labels even though authored provider title evidence exists.\n","design":"Trace all active parser title producers and assign the existing title-provenance vocabulary at the parser boundary, or make the public conversion recognize a typed parser-title source. Do not broaden acceptance to arbitrary nonempty text. Cover Grok, Antigravity, browser capture, Hermes, and at least one already-supported origin in a shared production-ingest fixture. Verify the title ladder, full-session conversion, and display-label path consume the same typed provenance. Add a red mutation that removes or relabels the parser provenance and makes the public title disappear or degrade visibly.\n","acceptance_criteria":"1. Genuine titles from Grok, Antigravity, browser capture, and Hermes survive full-session conversion and display-label generation. 2. Synthetic and heuristic titles remain governed by existing provenance policy. 3. Tests exercise parser, storage, and public conversion with a red provenance mutation. 4. Focused tests and devtools verify --quick pass; no live mutation.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:14:03Z","created_by":"Sinity","updated_at":"2026-08-06T15:14:03Z","labels":["area:ingest","area:query","lane:reindex"],"dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-ohkfy","title":"test: make incident ledger current-set authoritative","description":"The incident ledger merged in PR #3839 validates its own checked-in graph fixture but does not prove that the current Beads forcing set equals the ledger forcing set. It also lacks typed dependency-kind validation, source resolution for fixtures and receipts, bead-linked receipt ownership, and an unconditional devtools gate. These are review findings 3726231031, 3726231040, 3726292151, 3726231046, 3726292133, 3726292138, and 3726292144.\n\nMake the existing incident coverage validator consume a structured current-Beads export or digest supplied by the verification control plane. Keep natural language out of the gate. Preserve the committed ledger as a reviewed artifact, but fail closed when a current direct forcing dependency or P0 live acceptance Bead is absent, when dependency kinds are unknown, or when catalog references do not resolve. This is implementation and verification scope only. Do not close the campaign ledger Bead until the dynamic equality proof is green.\n","design":"Extend the existing versioned JSON ledger schema with an explicit closed dependency-kind vocabulary, typed route and receipt ownership fields, and source references that resolve to committed fixtures or named live-proof receipt producers. Add a loader path that receives the current Beads forcing-set export from the devtools command rather than parsing prose or importing a stale graph fixture. The current forcing set is the transitive dependency closure relevant to 818fy, including open, in-progress, and closed implementation nodes with named residual successors. Compare it to the ledger row set and require exact equality after the declared implementation-to-successor normalization.\n\nWire one unconditional check into the reindex verification command. It must run even when no optional campaign environment is present and must fail with a structured report naming missing, extra, stale, and unresolved entries. Add red tests that delete one current forcing row, add one new P0 blocker, change a dependency kind, remove a fixture source, and detach a receipt from its owning Bead. Do not infer correctness from close-reason text.\n","acceptance_criteria":"1. Current Beads forcing-set equality is checked against the ledger on every relevant devtools verification run.\n2. Dependency kinds are a closed typed vocabulary and unknown kinds fail validation.\n3. Every fixture, check, snapshot, receipt producer, and successor reference resolves.\n4. Every live receipt is associated with its owning Bead and cannot satisfy another row by name alone.\n5. The check is unconditional for the reindex gate and emits machine-readable missing/extra/stale diagnostics.\n6. Controlled red mutations make the validator fail for one missing row, one extra blocker, one unknown dependency kind, one missing source, and one unowned receipt.\n7. Existing ledger tests pass and devtools verify --quick passes.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:25:08Z","created_by":"Sinity","updated_at":"2026-08-06T15:25:18Z","labels":["area:verification","lane:reindex"],"dependency_count":0,"dependent_count":2,"comment_count":0} -{"_type":"issue","id":"polylogue-q4qpl","title":"fix: bind rebuild receipts to immutable source evidence","description":"Codex review of merged PR #3803 left four reindex-safety gaps in the schema-inference and rebuild receipt contract: parser-affecting source metadata is omitted from the immutable snapshot (3724479717), large external ground-truth trees are fully rehashed at every checkpoint (3724479723), internal blob bytes are trusted from recorded hashes without a second integrity snapshot (3724479730), and a receipt validation failure after pointer promotion can leave an active generation paired with a ready transaction (3724479731). The CLI relative receipt path finding 3724479738 is included because it can make a valid daemon rebuild consume the wrong evidence.\n\nThis Bead owns implementation and tests only. It does not authorize production migration or promotion.\n","design":"Expand the source snapshot with every replay-affecting raw field, including capture mode, revision kind, logical source key, predecessor and authority fields, and any semantic fingerprint already consumed by replay. Compute one identity-bound external inventory token for a rebuild pass and reuse it for repeated validation transitions, while detecting source changes before a new pass or terminal acceptance. Add a non-mutating internal blob snapshot or equivalent verified capability bound to the exact referenced blob universe, and require it before candidate readiness.\n\nReorder the terminal rebuild transition so every fallible receipt and corpus validation completes before the pointer flip. After promotion, record the promoted transaction state and receipt through a non-failing durable transition; if post-promotion evidence collection itself fails, emit an explicit post-promotion-attestation failure rather than leaving a ready transaction that looks resumable. Resolve CLI receipt paths to absolute paths before daemon transport. Preserve all existing fail-closed behavior and do not add an unbounded rehash to every page.\n","acceptance_criteria":"1. Parser-affecting source metadata changes invalidate a saved rebuild receipt and transaction.\n2. Repeated validation in one pass reuses an identity-bound external inventory token without weakening change detection.\n3. Referenced internal blob bytes are verified against a bound snapshot before candidate readiness.\n4. A failure after promotion cannot leave a ready transaction paired with a different active generation.\n5. Relative daemon receipt paths are resolved at the CLI boundary.\n6. Red tests prove each finding is load-bearing and fail if the old behavior is restored.\n7. Focused maintenance/CLI tests and devtools verify --quick pass.\n8. Production migration and candidate promotion remain out of scope.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:27:46Z","created_by":"Sinity","updated_at":"2026-08-06T15:27:46Z","labels":["area:maintenance","area:verification","lane:reindex"],"dependencies":[{"issue_id":"polylogue-q4qpl","depends_on_id":"polylogue-dudtn","type":"blocks","created_at":"2026-08-06T17:27:45Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} +{"_type":"issue","id":"polylogue-q4qpl","title":"fix: bind rebuild receipts to immutable source evidence","description":"Codex review of merged PR #3803 left four reindex-safety gaps in the schema-inference and rebuild receipt contract: parser-affecting source metadata is omitted from the immutable snapshot (3724479717), large external ground-truth trees are fully rehashed at every checkpoint (3724479723), internal blob bytes are trusted from recorded hashes without a second integrity snapshot (3724479730), and a receipt validation failure after pointer promotion can leave an active generation paired with a ready transaction (3724479731). The CLI relative receipt path finding 3724479738 is included because it can make a valid daemon rebuild consume the wrong evidence.\n\nThis Bead owns implementation and tests only. It does not authorize production migration or promotion.\n","design":"Expand the source snapshot with every replay-affecting raw field, including capture mode, revision kind, logical source key, predecessor and authority fields, and any semantic fingerprint already consumed by replay. Compute one identity-bound external inventory token for a rebuild pass and reuse it for repeated validation transitions, while detecting source changes before a new pass or terminal acceptance. Add a non-mutating internal blob snapshot or equivalent verified capability bound to the exact referenced blob universe, and require it before candidate readiness.\n\nReorder the terminal rebuild transition so every fallible receipt and corpus validation completes before the pointer flip. After promotion, record the promoted transaction state and receipt through a non-failing durable transition; if post-promotion evidence collection itself fails, emit an explicit post-promotion-attestation failure rather than leaving a ready transaction that looks resumable. Resolve CLI receipt paths to absolute paths before daemon transport. Preserve all existing fail-closed behavior and do not add an unbounded rehash to every page.\n","acceptance_criteria":"1. Parser-affecting source metadata changes invalidate a saved rebuild receipt and transaction.\n2. Repeated validation in one pass reuses an identity-bound external inventory token without weakening change detection.\n3. Referenced internal blob bytes are verified against a bound snapshot before candidate readiness.\n4. A failure after promotion cannot leave a ready transaction paired with a different active generation.\n5. Relative daemon receipt paths are resolved at the CLI boundary.\n6. Red tests prove each finding is load-bearing and fail if the old behavior is restored.\n7. Focused maintenance/CLI tests and devtools verify --quick pass.\n8. Production migration and candidate promotion remain out of scope.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:27:46Z","created_by":"Sinity","updated_at":"2026-08-06T15:27:46Z","labels":["area:maintenance","area:verification","lane:reindex"],"dependencies":[{"issue_id":"polylogue-q4qpl","depends_on_id":"polylogue-dudtn","type":"blocks","created_at":"2026-08-06T17:27:45Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-tiozw","title":"fix: discard partial hook match stages before liveness checks","description":"Residual P1 from Codex review of merged PR #3847. A failed hook match-stage build can leave both temporary tables behind, and the next candidate can mistake the partial stage for a complete ready stage. That can make blob liveness verification accept a false negative and delete evidence after an interrupted query.\n","design":"Make match-stage construction transactional and readiness-bearing. A stage is visible to candidate evaluation only after every table, population statement, and integrity check succeeds. On any exception, drop all stage tables created by that attempt and clear the readiness marker. A later candidate must rebuild rather than reuse partial state. Add crash and exception injection tests for failure after each stage table creation and after population begins, then assert no candidate can report a clean dead-blob result from partial state.\n","acceptance_criteria":"1. Failed match-stage construction removes all created tables and readiness state. 2. Later candidates never reuse partial stages. 3. Successful reuse requires complete integrity proof. 4. Injected failures make blob-liveness verification fail closed. 5. Focused hook/blob-liveness tests and devtools verify --quick pass; no live mutation.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:14:07Z","created_by":"Sinity","updated_at":"2026-08-06T15:14:07Z","labels":["area:maintenance","area:storage","lane:reindex"],"dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-uecir","title":"acceptance: complete Antigravity source remediation before candidate freeze","description":"PR #3859 Codex review comment 3728626200 identified a phase-ordering hole: the 44 real Antigravity conversation reingest and 116 metadata-stub purge are source-side work required before the frozen source snapshot, but the current graph leaves polylogue-msia runnable only after promotion. That permits candidate construction from the old 328 MB-missing source state.\n\nThis Bead is the phase-2 live operation receipt. It does not implement the RPC parser or phantom purge actuator, which already have merged implementation Beads, and it does not perform production mutation without the operator boundary.\n","design":"Before candidate construction, require the selected deployed package and Antigravity language-server RPC availability, perform the real conversations/*.pb reingest through the ordinary source acquisition route, then run the existing backup-gated phantom purge actuator against the 116 metadata stubs. Bind the exact backup, package, source snapshot, before/after raw and session censuses, message-count distribution, zero metadata-only sessions, retained AGENT_SIDECAR_META provenance, RPC evidence, and quick_check results into an immutable receipt. If RPC is unavailable, emit typed not_applicable only with the exact unsupported-origin evidence and keep candidate acceptance blocked unless the operator explicitly accepts the exclusion. No direct SQL deletes.\n","acceptance_criteria":"1. The source phase cannot freeze or build a candidate while the Antigravity source-side receipt is missing.\n2. RPC availability is checked before any apply and absence is typed, never silently skipped.\n3. The real conversations/*.pb route produces the expected real-session/message cohort or a typed blocked outcome.\n4. The existing backup-gated phantom purge removes the metadata-only stubs while retaining AGENT_SIDECAR_META provenance.\n5. Before/after source and index-independent counts, origin distribution, package SHA, backup identity, and quick_check are recorded.\n6. The receipt is consumed by polylogue-reindex-source-remediation and polylogue-live-operation-receipts.\n7. No production mutation is performed by implementation lanes; the operator applies the named command in the maintenance window.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T15:35:00Z","created_by":"Sinity","updated_at":"2026-08-06T15:35:00Z","labels":["area:maintenance","area:verification","lane:reindex"],"dependency_count":0,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-x97cf","title":"maintenance: install typed live-proof receipt protocol","description":"Implement one static, typed live-proof receipt protocol for the reindex campaign. It must collect read-only, candidate, and already-produced apply receipts without becoming a task scheduler or mutation surface. Every receipt binds the proof and Bead IDs, exact code SHA, archive identity, source snapshot, schema versions, candidate identity when applicable, parser and lowering fingerprints, registry version, structured result, typed residues, input receipt digests, and private-path digests. This is the shared evidence protocol required before live-proof children can contribute to the terminal reindex proof.\n","design":"Add a versioned LiveProofSpec registry and receipt collector. Register the fixed proof modes read_only, candidate, and existing_apply_receipt. Expose polylogue ops maintenance live-proof with a fixed proof ID and receipt input/output contract. Read-only and candidate producers may execute only registered callables; existing-apply mode validates an immutable receipt. The command must never apply a mutation, stop or start the daemon, migrate a tier, promote a generation, accept arbitrary commands, or infer proof from Beads status. Private paths are represented by SHA-256 digest plus basename. Wire the collector into the live-operation aggregate and the candidate/final proof consumers. Reuse the existing archive-verification registry and canonical fingerprint helpers.\n","acceptance_criteria":"1. The three fixed proof modes and typed residue vocabulary are represented by one registry.\n2. The maintenance command accepts only registered proof IDs and the mode-specific input shape.\n3. Receipts are immutable, self-hashed, and bind code, archive, source snapshot, schema, semantic fingerprints, result, residues, and input receipts.\n4. Candidate receipts bind the exact inactive candidate generation; existing-apply receipts bind the validated input receipt.\n5. Private paths never appear in durable receipt payloads except as digest plus basename.\n6. The command has no mutation, daemon lifecycle, migration, promotion, or arbitrary-command path.\n7. Missing, stale, or malformed bindings fail closed, and a controlled mutation of any required binding makes validation fail.\n8. Focused tests cover registry completeness, receipt determinism, mode isolation, binding failures, and the real CLI dispatch path.\n9. The protocol is a prerequisite of polylogue-live-operation-receipts and remains open until its implementation and focused verification merge.","notes":"Compiled packet intake 2026-08-06. Source packet tar SHA-256: cae45456e8f25c491085c2035afc8fbf36545e4ac59c55bd53c114e6d4179189. Execution-spec SHA-256: 64fa47bd7d42e0d4e81b3e77db2216a88300dd81b08141dd6e79a54319607303. The packet graph basis is 685f2ca8, so current phase names and dependencies must be checked against current Beads before dispatch.\nGraph correction from Codex review on PR #3861 (comment 5205727476): closed duplicate polylogue-q8tpq no longer claims to supersede this canonical Bead. The supersedes relationship is now x97cf -\u003e q8tpq; x97cf remains the open implementation owner consumed by live-operation and candidate proof.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T13:50:01Z","created_by":"Sinity","updated_at":"2026-08-06T15:55:02Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-x97cf","depends_on_id":"polylogue-q8tpq","type":"supersedes","created_at":"2026-08-06T17:54:48Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":2,"comment_count":0} @@ -37,7 +37,7 @@ {"_type":"issue","id":"polylogue-claude-vintage-live-proof","title":"acceptance: prove measured Claude vintage reclassification","description":"Residual live or proof scope of polylogue-0qfy. The implementation mechanism may remain closed, but the reindex cannot claim convergence until this evidence exists.","design":"Run through the real production seam against a frozen source or candidate generation. Bind the result to an immutable receipt and state explicitly what was not exercised.","acceptance_criteria":"A sanitized measured-cohort old/new pair and read-only cohort-reclassification receipt prove the actual classifier branch and canonical identity decision.","notes":"Parent implementation bead: polylogue-0qfy. This child exists because the audit found the implementation closure did not prove the live effect.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:11Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:11Z","dependencies":[{"issue_id":"polylogue-claude-vintage-live-proof","depends_on_id":"polylogue-0qfy","type":"blocks","created_at":"2026-08-06T07:39:43Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-codex-804-live-proof","title":"acceptance: prove incident-scale Codex materialization and recovery","description":"Residual live or proof scope of polylogue-5iz4. The implementation mechanism may remain closed, but the reindex cannot claim convergence until this evidence exists.","design":"Run through the real production seam against a frozen source or candidate generation. Bind the result to an immutable receipt and state explicitly what was not exercised.","acceptance_criteria":"A real or faithfully sanitized 804-revision, approximately 90 MiB wire shape exercises recovery-copy ordering, crash/restart, source-authority terminal state, and candidate materialization with resource receipts.","notes":"Parent implementation bead: polylogue-5iz4. This child exists because the audit found the implementation closure did not prove the live effect.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:11Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:11Z","dependencies":[{"issue_id":"polylogue-codex-804-live-proof","depends_on_id":"polylogue-5iz4","type":"blocks","created_at":"2026-08-06T07:39:33Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-topology-live-proof","title":"acceptance: prove live topology status and safe public composition","description":"Residual live or proof scope of polylogue-4ts.10. The implementation mechanism may remain closed, but the reindex cannot claim convergence until this evidence exists.","design":"Run through the real production seam against a frozen source or candidate generation. Bind the result to an immutable receipt and state explicitly what was not exercised.","acceptance_criteria":"Candidate and live census show zero empty link status or method values, typed unresolved or cycle-broken edges, and visibly safe reader behavior for unresolved parents.","notes":"Parent implementation bead: polylogue-4ts.10. This child exists because the audit found the implementation closure did not prove the live effect.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:11Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:11Z","dependencies":[{"issue_id":"polylogue-topology-live-proof","depends_on_id":"polylogue-4ts.10","type":"blocks","created_at":"2026-08-06T07:39:23Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} -{"_type":"issue","id":"polylogue-live-operation-receipts","title":"acceptance: collect live operation receipts for reindex readiness","description":"Aggregate the named operational and live-proof children that convert implementation mechanisms into evidence about the actual archive. This remains open until each applicable child has an immutable receipt or an explicit typed non-applicability decision.","design":"Keep mechanism beads historically honest. Each child proves one live effect or operation and is independently rerunnable and idempotent.","acceptance_criteria":"Every required live-proof child has a receipt bound to the exact source snapshot, candidate or active generation, and semantic fingerprint. Residues are typed rather than silently omitted. The aggregate report is consumed by the terminal reindex proof.","notes":"Compiled packet intake 2026-08-06: all aggregate members must produce a validated polylogue.live-proof-receipt.v1 or an exact typed non-applicability decision. The aggregate consumes the typed live-proof protocol, current source/candidate bindings, and scoped apply receipts; it does not infer completion from Bead status or PR merge state. Promotion/restart is a downstream phase and is not itself a live-operation child.\nCompiled packet intake 2026-08-06: all aggregate members must produce a validated polylogue.live-proof-receipt.v1 or an exact typed non-applicability decision. The aggregate consumes the typed live-proof protocol, current source/candidate bindings, and scoped apply receipts; it does not infer completion from Bead status or PR merge state. Promotion/restart is a downstream phase and is not itself a live-operation child.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:10Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:22Z","dependencies":[{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-active-leaf-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-byte-supersession-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-chatgpt-content-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-claude-streaming-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-claude-vintage-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-codex-804-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-cursor-authority-live-proof","type":"blocks","created_at":"2026-08-06T13:54:17Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-excluded-cursor-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-hook-authority-conflict-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-hook-reconciliation-apply-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-raw-dedupe-apply-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-stalled-cursor-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-topology-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-uecir","type":"blocks","created_at":"2026-08-06T17:39:10Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-x97cf","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":15,"dependent_count":1,"comment_count":0} +{"_type":"issue","id":"polylogue-live-operation-receipts","title":"acceptance: collect live operation receipts for reindex readiness","description":"Aggregate the named operational and live-proof children that convert implementation mechanisms into evidence about the actual archive. This remains open until each applicable child has an immutable receipt or an explicit typed non-applicability decision.","design":"Keep mechanism beads historically honest. Each child proves one live effect or operation and is independently rerunnable and idempotent.","acceptance_criteria":"Every required live-proof child has a receipt bound to the exact source snapshot, candidate or active generation, and semantic fingerprint. Residues are typed rather than silently omitted. The aggregate report is consumed by the terminal reindex proof.","notes":"Compiled packet intake 2026-08-06: all aggregate members must produce a validated polylogue.live-proof-receipt.v1 or an exact typed non-applicability decision. The aggregate consumes the typed live-proof protocol, current source/candidate bindings, and scoped apply receipts; it does not infer completion from Bead status or PR merge state. Promotion/restart is a downstream phase and is not itself a live-operation child.\nCompiled packet intake 2026-08-06: all aggregate members must produce a validated polylogue.live-proof-receipt.v1 or an exact typed non-applicability decision. The aggregate consumes the typed live-proof protocol, current source/candidate bindings, and scoped apply receipts; it does not infer completion from Bead status or PR merge state. Promotion/restart is a downstream phase and is not itself a live-operation child.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:10Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:22Z","dependencies":[{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-active-leaf-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-byte-supersession-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-chatgpt-content-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-claude-streaming-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-claude-vintage-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-codex-804-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-cursor-authority-live-proof","type":"blocks","created_at":"2026-08-06T13:54:17Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-excluded-cursor-live-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-hook-authority-conflict-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-hook-reconciliation-apply-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-raw-dedupe-apply-proof","type":"blocks","created_at":"2026-08-06T07:02:58Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-stalled-cursor-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-topology-live-proof","type":"blocks","created_at":"2026-08-06T07:02:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-uecir","type":"blocks","created_at":"2026-08-06T17:39:10Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-x97cf","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-live-operation-receipts","depends_on_id":"polylogue-q4qpl","type":"blocks","created_at":"2026-08-06T17:42:00Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":16,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-reindex-final-proof","title":"acceptance: emit a proof-carrying production reindex receipt","description":"Terminal receipt-only aggregation gate for the production reindex. It authorizes neither code closure nor promotion by itself. It becomes satisfiable only when every incident, candidate, live-operation, fidelity, provenance, scale, and public-contract obligation has a current hash-bound receipt.","design":"Consume one self-describing invariant registry, one canonical comparator, one structured incident ledger, and immutable live receipts. Do not infer completion from a merged PR or a green synthetic test.","acceptance_criteria":"Every blocking incident-ledger row is green or has explicit operator acceptance. The candidate receipt binds the source snapshot, semantic fingerprints, registry version, canonical snapshot, and exact candidate generation. Promotion, daemon restart, convergence, public query tour, and rollback-retention receipts are present and current.","notes":"Evidence required: candidate acceptance receipt; promotion receipt; post-promotion daemon health receipt; canonical query-tour receipt; retained-generation rollback receipt.\nCompiled packet mapping 2026-08-06: this is phase 6 postflight proof, not a start gate. It consumes candidate acceptance, promotion/restart, live-operation, public query-tour, convergence, health, and rollback-retention receipts. Historical direct implementation edges must not be used as a second control surface.\nCompiled packet mapping 2026-08-06: this is phase 6 postflight proof, not a start gate. It consumes candidate acceptance, promotion/restart, live-operation, public query-tour, convergence, health, and rollback-retention receipts. Historical direct implementation edges must not be used as a second control surface.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:10Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:42Z","dependencies":[{"issue_id":"polylogue-reindex-final-proof","depends_on_id":"polylogue-i3i5k","type":"blocks","created_at":"2026-08-06T18:53:04Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-final-proof","depends_on_id":"polylogue-live-operation-receipts","type":"blocks","created_at":"2026-08-06T07:02:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-final-proof","depends_on_id":"polylogue-reindex-promotion-restart","type":"blocks","created_at":"2026-08-06T13:53:47Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":3,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-stalled-cursor-live-proof","title":"acceptance: drain and disposition all stalled append cursors","description":"Residual live or proof scope of polylogue-2qrx. The implementation mechanism may remain closed, but the reindex cannot claim convergence until this evidence exists.","design":"Run through the real production seam against a frozen source or candidate generation. Bind the result to an immutable receipt and state explicitly what was not exercised.","acceptance_criteria":"Ordinary daemon catch-up leaves zero unexplained stalled cursors and every residue has a typed disposition bound to the source snapshot.","notes":"Parent implementation bead: polylogue-2qrx. This child exists because the audit found the implementation closure did not prove the live effect.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:00:10Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:10Z","dependencies":[{"issue_id":"polylogue-stalled-cursor-live-proof","depends_on_id":"polylogue-2qrx","type":"blocks","created_at":"2026-08-06T07:39:14Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-dlfcx","title":"fix: gate fast-forward promotions on candidate corpus fidelity","description":"Independent review after #3748 found devtools/index_v37_fast_forward.py promotes or recovers activating candidates without the corpus fidelity candidate gate. Absence/revision tests also pass if runners accidentally resolve the active index. Every production promotion path must gate candidate index plus durable source evidence before activation or recovery.","design":"Route fast-forward and activating-recovery through the same candidate acceptance helper used by managed rebuild, retaining durable source root plus explicit candidate index path. Strengthen absence/revision fixtures so active index passes and only candidate fails. Do not duplicate the rebuild gate or mutate production.","acceptance_criteria":"1. Fast-forward normal and restart-recovery promotion refuse an invalid candidate before activation. 2. Candidate-only absence and revision drift tests fail only with index_path_override; active verification remains clear. 3. Focused real-route tests and quick gate pass. 4. No live archive mutation.","status":"in_progress","priority":0,"issue_type":"bug","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-08-04T07:44:31Z","created_by":"Sinity","updated_at":"2026-08-04T07:44:47Z","started_at":"2026-08-04T07:44:47Z","lease_expires_at":"2026-08-04T07:49:47Z","heartbeat_at":"2026-08-04T07:44:47Z","dependency_count":0,"dependent_count":1,"comment_count":0} diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index 1a8efa530e..6b76a8855e 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -465,6 +465,22 @@ def rebuild_index_command( if payload["status"] != "ready": raise click.ClickException("rebuild schema currency preflight failed; migrate or deploy before rebuilding") return + if schema_inference_receipt_path is not None: + # Resolve relative receipt handles against the configured archive + # command context before serializing the daemon request. The shared + # resolver also preserves the refusal for receipt paths inside the + # archive's durable file set. + from polylogue.maintenance.schema_inference_gate import ( + SchemaInferenceGateError, + resolve_schema_inference_receipt_reference, + ) + + try: + schema_inference_receipt_path = resolve_schema_inference_receipt_reference( + root, schema_inference_receipt_path + ) + except SchemaInferenceGateError as exc: + raise click.ClickException(str(exc)) from exc if use_daemon: payload = _run_daemon_rebuild( daemon_url, diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py index 02c2dde5ad..85b1aa740d 100644 --- a/polylogue/daemon/bulk_rebuild.py +++ b/polylogue/daemon/bulk_rebuild.py @@ -144,10 +144,12 @@ def _raise_daemon_cleanup_failures(cleanup_errors: list[BaseException], *, label #: Transaction statuses that mean "not resumable, retire and start fresh at #: the same well-known operation id": ``promoted`` (a prior build already -#: succeeded and is now the active index), ``stale`` (source evidence -#: changed mid-build), ``failed`` (a pass raised; automagic doctrine retries -#: rather than waiting on an operator to intervene). -_TERMINAL_NOT_RESUMABLE = frozenset({"promoted", "stale", "failed"}) +#: succeeded and is now the active index), ``promoted-attestation-failed`` (a +#: build succeeded and is active, but a post-promotion receipt write failed), +#: ``stale`` (source evidence changed mid-build), and ``failed`` (a pass +#: raised; automagic doctrine retries rather than waiting on an operator to +#: intervene). +_TERMINAL_NOT_RESUMABLE = frozenset({"promoted", "promoted-attestation-failed", "stale", "failed"}) def _preflight_raw_failure_lifecycle(root: Path) -> None: @@ -162,6 +164,41 @@ def _preflight_raw_failure_lifecycle(root: Path) -> None: raise RuntimeError(f"daemon bulk-rebuild raw failure lifecycle preflight failed: {reason}") +def _reconcile_active_generation_transaction( + store: IndexGenerationStore, transaction: IndexRebuildTransaction +) -> IndexRebuildTransaction: + """Turn a post-pointer-write transaction into terminal state on restart. + + Promotion changes the active pointer before the transaction attestation is + durable. If both the normal and recovery checkpoints fail, the persisted + transaction can still look resumable even though its generation is active. + The next resolver pass must record that observed fact before returning it + to the rebuild loop, otherwise the daemon retries an already-active + generation forever. + """ + + if transaction.status in _TERMINAL_NOT_RESUMABLE: + return transaction + try: + generation = store.load(transaction.generation_id) + active_path = store.active_pointer.resolve(strict=True) + generation_path = Path(generation.index_path).resolve(strict=True) + except (FileNotFoundError, OSError, ValueError): + return transaction + if generation.state != "active" or active_path != generation_path: + return transaction + return store.checkpoint_transaction( + transaction, + status="promoted-attestation-failed", + error="reconciled active generation after interrupted promotion attestation", + post_promotion_attestation={ + "status": "reconciled-after-restart", + "generation_id": generation.generation_id, + "generation_state": generation.state, + }, + ) + + def resolve_or_start_daemon_bulk_rebuild_transaction( root: Path, *, schema_inference_receipt_path: Path | None = None ) -> IndexRebuildTransaction: @@ -219,44 +256,65 @@ def resolve_or_start_daemon_bulk_rebuild_transaction( transaction = None if transaction is not None and transaction.status not in _TERMINAL_NOT_RESUMABLE: - _validate_rebuild_provenance_receipt(root, schema_inference_receipt_path) - return transaction + transaction = _reconcile_active_generation_transaction(store, transaction) + if transaction.status not in _TERMINAL_NOT_RESUMABLE: + _validate_rebuild_provenance_receipt(root, schema_inference_receipt_path) + return transaction if transaction is not None: - # Terminal: retire the old candidate/transaction record before - # reusing the well-known operation id. A "promoted" generation is - # already the active index (nothing to discard); "stale"/"failed" - # candidates are still inactive and safe to discard. - cleanup_errors: list[BaseException] = [] - if transaction.status != "promoted": + if transaction.status == "promoted-attestation-failed": + # The generation is already active. Preserve the terminal + # attestation failure while its source evidence is unchanged. + # A later, receipt-authorized source change needs a fresh + # daemon transaction, but must retain the active generation. + current_source_snapshot = rebuild_source_evidence_snapshot(root) + if current_source_snapshot == transaction.source_snapshot: + return transaction + if not store.discard_transaction(DAEMON_BULK_REBUILD_OPERATION_ID): + raise RuntimeError( + f"transaction {DAEMON_BULK_REBUILD_OPERATION_ID} was not discarded after source drift" + ) + transaction = None + if transaction is None: + pass + else: + # Terminal: retire the old candidate/transaction record before + # reusing the well-known operation id. A promoted generation, + # including one with a failed post-promotion attestation, is + # already the active index (nothing to discard); "stale/failed" + # candidates are still inactive and safe to discard. + cleanup_errors: list[BaseException] = [] + if transaction.status not in {"promoted", "promoted-attestation-failed"}: + try: + generation = store.load(transaction.generation_id) + except BaseException as exc: + logger.error("bulk-rebuild: terminal candidate load failed", exc_info=True) + cleanup_errors.append(exc) + else: + if generation.state != "inactive": + cleanup_errors.append(RuntimeError(f"candidate {generation.generation_id} is not inactive")) + else: + try: + if not store.discard_if_inactive(generation): + cleanup_errors.append( + RuntimeError(f"candidate {generation.generation_id} was not discarded") + ) + except BaseException as exc: + logger.error("bulk-rebuild: terminal candidate discard raised", exc_info=True) + cleanup_error = RuntimeError( + f"candidate {generation.generation_id} discard failed: {exc}" + ) + cleanup_error.__cause__ = exc + cleanup_errors.append(cleanup_error) try: - generation = store.load(transaction.generation_id) + if not store.discard_transaction(DAEMON_BULK_REBUILD_OPERATION_ID): + cleanup_errors.append( + RuntimeError(f"transaction {DAEMON_BULK_REBUILD_OPERATION_ID} was not discarded") + ) except BaseException as exc: - logger.error("bulk-rebuild: terminal candidate load failed", exc_info=True) + logger.error("bulk-rebuild: terminal transaction discard failed", exc_info=True) cleanup_errors.append(exc) - else: - if generation.state != "inactive": - cleanup_errors.append(RuntimeError(f"candidate {generation.generation_id} is not inactive")) - else: - try: - if not store.discard_if_inactive(generation): - cleanup_errors.append( - RuntimeError(f"candidate {generation.generation_id} was not discarded") - ) - except BaseException as exc: - logger.error("bulk-rebuild: terminal candidate discard raised", exc_info=True) - cleanup_error = RuntimeError(f"candidate {generation.generation_id} discard failed: {exc}") - cleanup_error.__cause__ = exc - cleanup_errors.append(cleanup_error) - try: - if not store.discard_transaction(DAEMON_BULK_REBUILD_OPERATION_ID): - cleanup_errors.append( - RuntimeError(f"transaction {DAEMON_BULK_REBUILD_OPERATION_ID} was not discarded") - ) - except BaseException as exc: - logger.error("bulk-rebuild: terminal transaction discard failed", exc_info=True) - cleanup_errors.append(exc) - _raise_daemon_cleanup_failures(cleanup_errors, label="daemon bulk-rebuild terminal") + _raise_daemon_cleanup_failures(cleanup_errors, label="daemon bulk-rebuild terminal") _validate_rebuild_provenance_receipt(root, schema_inference_receipt_path) source_snapshot = rebuild_source_evidence_snapshot(root) @@ -340,7 +398,7 @@ async def run_daemon_bulk_rebuild_pass( root, schema_inference_receipt_path=receipt_path, ) - if transaction.status == "promoted": + if transaction.status in {"promoted", "promoted-attestation-failed"}: return None location = ArchiveLocation.resolve(root) diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 3b15f24285..f82673d9f5 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -10,15 +10,17 @@ import asyncio import contextlib +import contextvars import json import os import sqlite3 import time +from collections.abc import Iterable, Sequence from dataclasses import asdict, dataclass, field from hashlib import sha256 from http import HTTPStatus from pathlib import Path -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from polylogue.config import Config from polylogue.core.errors import PolylogueError @@ -59,6 +61,10 @@ logger = get_logger(__name__) +_ACTIVE_EXTERNAL_INVENTORY_TOKEN: contextvars.ContextVar[dict[str, object] | None] = contextvars.ContextVar( + "rebuild_external_inventory_token", default=None +) + class RebuildProvenanceError(RuntimeError): """Raised when rebuild evidence is no longer valid for a mutation.""" @@ -125,7 +131,7 @@ def require_rebuild_schema_currency(root: Path) -> dict[str, object]: return diagnostic -@dataclass(frozen=True, slots=True) +@dataclass(slots=True) class RebuildProvenanceContext: """Validated evidence shared by every mutation in one rebuild pass. @@ -141,9 +147,22 @@ class RebuildProvenanceContext: receipt_path: Path | None source_snapshot: str consumed_evidence: dict[str, object] - - def validate(self) -> None: - _validate_rebuild_provenance_receipt(self.root, self.receipt_path) + external_inventory_token: dict[str, object] = field(default_factory=dict) + verified_blob_integrity_snapshot: dict[str, object] | None = None + + def validate(self, *, verify_blob_integrity: bool = False, refresh_blob_integrity: bool = False) -> None: + cached_snapshot = None if refresh_blob_integrity else self.verified_blob_integrity_snapshot + validated = _validate_rebuild_provenance_receipt( + self.root, + self.receipt_path, + inventory_token=self.external_inventory_token, + verify_blob_integrity=verify_blob_integrity, + verified_blob_integrity_snapshot=cached_snapshot, + ) + if verify_blob_integrity: + refreshed_snapshot = validated.pop("_verified_blob_integrity_snapshot", None) + if isinstance(refreshed_snapshot, dict): + self.verified_blob_integrity_snapshot = refreshed_snapshot from polylogue.maintenance.schema_inference_gate import rebuild_source_revision_snapshot if rebuild_source_revision_snapshot(self.root) != self.source_snapshot: @@ -157,23 +176,80 @@ def validate_cleanup(self) -> None: raise RuntimeError("rebuild cleanup has no validated provenance context") -def _validate_rebuild_provenance_receipt(root: Path, receipt_path: Path | None) -> dict[str, object]: +def _validate_rebuild_provenance_receipt( + root: Path, + receipt_path: Path | None, + *, + inventory_token: dict[str, object] | None = None, + verify_blob_integrity: bool = False, + verified_blob_integrity_snapshot: dict[str, object] | None = None, +) -> dict[str, object]: """Validate rebuild provenance at the current ownership boundary.""" from polylogue.maintenance.schema_inference_gate import ( SchemaInferenceGateError, validate_schema_inference_receipt, ) + if inventory_token is None: + inventory_token = _ACTIVE_EXTERNAL_INVENTORY_TOKEN.get() try: - return validate_schema_inference_receipt(root, receipt_path) + return validate_schema_inference_receipt( + root, + receipt_path, + inventory_token=inventory_token, + verify_blob_integrity=verify_blob_integrity, + verified_blob_integrity_snapshot=verified_blob_integrity_snapshot, + ) except SchemaInferenceGateError as exc: raise RebuildProvenanceError(f"rebuild schema-inference preflight gate failed: {exc}") from exc -def _validate_before_derived_state(provenance: RebuildProvenanceContext) -> None: +def _mark_rebuild_transaction_stale_after_provenance_failure( + root: Path, operation_id: str | None, error: RebuildProvenanceError +) -> None: + """Terminally classify a resumable pass whose next admission failed. + + The normal checkpoint helper validates first, which is correct for every + ordinary state transition. A failed source/receipt admission is the one + exception: re-running that validation would preserve a ``paused`` or + ``ready`` transaction indefinitely. The stale marker is lifecycle + evidence, so it is written directly through ``IndexGenerationStore`` and + never authorizes another replay. + """ + if operation_id is None: + return + from polylogue.storage.index_generation import IndexGenerationStore + + try: + store = IndexGenerationStore.for_archive_root(root) + transaction = store.load_transaction(operation_id) + except Exception as load_error: + error.add_note(f"could not load rebuild transaction to mark it stale: {load_error}") + return + if transaction.status in {"promoted", "promoted-attestation-failed", "stale"}: + return + try: + store.checkpoint_transaction( + transaction, + status="stale", + error=f"stale because source evidence or receipt validation failed: {error}", + ) + except Exception as checkpoint_error: + error.add_note(f"could not persist stale rebuild transaction: {checkpoint_error}") + + +def _validate_before_derived_state( + provenance: RebuildProvenanceContext, + *, + verify_blob_integrity: bool = False, + refresh_blob_integrity: bool = False, +) -> None: """Validate immediately before a derived-state mutation begins.""" try: - provenance.validate() + provenance.validate( + verify_blob_integrity=verify_blob_integrity, + refresh_blob_integrity=refresh_blob_integrity, + ) except Exception as exc: raise RebuildDerivedStateProvenanceError(str(exc)) from exc @@ -276,11 +352,15 @@ def _create_rebuild_transaction_after_receipt_validation( generation_store: IndexGenerationStore, request: RebuildIndexRequest, root: Path, + *, + inventory_token: dict[str, object] | None = None, ) -> IndexRebuildTransaction: """Create the first candidate only after an ownership-bound validation.""" from polylogue.storage.index_generation import rebuild_source_evidence_snapshot - consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + consumed_evidence = _validate_rebuild_provenance_receipt( + root, request.schema_inference_receipt_path, inventory_token=inventory_token + ) source_snapshot = rebuild_source_evidence_snapshot(root) transaction = generation_store.create_transaction( source_snapshot=source_snapshot, @@ -292,7 +372,9 @@ def _create_rebuild_transaction_after_receipt_validation( ), ) try: - _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + _validate_rebuild_provenance_receipt( + root, request.schema_inference_receipt_path, inventory_token=inventory_token + ) if rebuild_source_evidence_snapshot(root) != source_snapshot: raise RebuildProvenanceError( "rebuild schema-inference preflight gate failed: source evidence changed during transaction creation" @@ -323,9 +405,17 @@ def _checkpoint_rebuild_transaction_after_receipt_validation( processed_blob_bytes: int | None = None, error: str | None = None, derived_stores_cleared: bool | None = None, + post_promotion_attestation: dict[str, object] | None = None, + inventory_token: dict[str, object] | None = None, + verify_blob_integrity: bool = False, ) -> IndexRebuildTransaction: """Validate immediately before every persisted rebuild state transition.""" - _validate_rebuild_provenance_receipt(root, receipt_path) + _validate_rebuild_provenance_receipt( + root, + receipt_path, + inventory_token=inventory_token, + verify_blob_integrity=verify_blob_integrity, + ) return generation_store.checkpoint_transaction( transaction, status=status, @@ -335,6 +425,7 @@ def _checkpoint_rebuild_transaction_after_receipt_validation( processed_blob_bytes=processed_blob_bytes, error=error, derived_stores_cleared=derived_stores_cleared, + post_promotion_attestation=post_promotion_attestation, ) @@ -344,9 +435,10 @@ def _save_rebuild_pass_receipt_after_receipt_validation( pass_receipt: RebuildIndexReceipt, root: Path, receipt_path: Path | None, + inventory_token: dict[str, object] | None = None, ) -> None: """Validate before publishing a pass receipt tied to rebuild state.""" - _validate_rebuild_provenance_receipt(root, receipt_path) + _validate_rebuild_provenance_receipt(root, receipt_path, inventory_token=inventory_token) generation_store.save_pass_receipt(operation_id, pass_receipt.to_dict()) @@ -842,50 +934,117 @@ def _rebuild_replay_closure_evidence(archive_root: Path, raw_ids: list[str] | tu source_db = archive_root / "source.db" if not source_db.exists(): - return {"raw_ids": sorted(raw_ids), "logical_source_keys": [], "raw_session_memberships": []} + return { + "raw_ids": sorted(raw_ids), + "logical_source_keys": [], + "raw_session_evidence": [], + "raw_session_memberships": [], + } from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + bind_chunk_size = 900 + + def chunks(values: Sequence[str]) -> Iterable[list[str]]: + for start in range(0, len(values), bind_chunk_size): + yield list(values[start : start + bind_chunk_size]) + with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as connection: expanded, logical_keys = ArchiveStore.expand_raw_membership_selection_sync(connection, list(raw_ids)) - placeholders_raw = ",".join("?" for _ in expanded) - placeholders_key = ",".join("?" for _ in logical_keys) - clauses: list[str] = [] - parameters: list[str] = [] - if expanded: - clauses.append(f"raw_id IN ({placeholders_raw})") - parameters.extend(expanded) - if logical_keys: - clauses.append(f"logical_source_key IN ({placeholders_key})") - parameters.extend(logical_keys) + membership_rows: dict[tuple[str, str], tuple[Any, ...]] = {} + membership_columns = ( + "raw_id, logical_source_key, provider_session_id, source_revision, " + "normalized_content_hash, message_count, predecessor_raw_id, acquisition_generation, " + "revision_authority, decision, decided_at_ms" + ) + for column, values in (("raw_id", expanded), ("logical_source_key", logical_keys)): + for batch in chunks(values): + placeholders = ",".join("?" for _ in batch) + selected = connection.execute( + f"SELECT {membership_columns} FROM raw_session_memberships WHERE {column} IN ({placeholders})", + batch, + ) + for row in selected: + membership_rows[(str(row[0]), str(row[1]))] = tuple(row) + rows: list[dict[str, object]] = [] - if clauses: - selected = connection.execute( - "SELECT raw_id, logical_source_key, provider_session_id, source_revision, " - "normalized_content_hash, message_count, predecessor_raw_id, acquisition_generation, " - "revision_authority, decision, decided_at_ms " - "FROM raw_session_memberships WHERE " + " OR ".join(clauses) + " ORDER BY logical_source_key, raw_id", - parameters, + for row in sorted(membership_rows.values(), key=lambda item: (str(item[1]), str(item[0]))): + rows.append( + { + "raw_id": str(row[0]), + "logical_source_key": str(row[1]), + "provider_session_id": str(row[2]), + "source_revision": str(row[3]), + "normalized_content_hash": bytes(row[4]).hex(), + "message_count": int(row[5]), + "predecessor_raw_id": None if row[6] is None else str(row[6]), + "acquisition_generation": int(row[7]), + "revision_authority": str(row[8]), + "decision": None if row[9] is None else str(row[9]), + "decided_at_ms": None if row[10] is None else int(row[10]), + } ) - for row in selected: - rows.append( - { - "raw_id": str(row[0]), - "logical_source_key": str(row[1]), - "provider_session_id": str(row[2]), - "source_revision": str(row[3]), - "normalized_content_hash": bytes(row[4]).hex(), - "message_count": int(row[5]), - "predecessor_raw_id": None if row[6] is None else str(row[6]), - "acquisition_generation": int(row[7]), - "revision_authority": str(row[8]), - "decision": None if row[9] is None else str(row[9]), - "decided_at_ms": None if row[10] is None else int(row[10]), - } - ) + + raw_rows: list[tuple[Any, ...]] = [] + for batch in chunks(expanded): + placeholders = ",".join("?" for _ in batch) + raw_rows.extend( + tuple(row) + for row in connection.execute( + f""" + SELECT raw_id, origin, capture_mode, native_id, source_path, + source_index, blob_hash, blob_size, acquired_at_ms, + logical_source_key, revision_kind, source_revision, + predecessor_source_revision, predecessor_raw_id, + baseline_raw_id, append_start_offset, append_end_offset, + acquisition_generation, revision_authority, + revision_authority_evidence + FROM raw_sessions WHERE raw_id IN ({placeholders}) + """, + batch, + ).fetchall() + ) + raw_rows.sort(key=lambda row: str(row[0])) + from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin + + lowering = lowering_fingerprint() + parser_fingerprints: dict[str, str] = {} + for row in raw_rows: + origin = str(row[1]) + if origin not in parser_fingerprints: + parser_fingerprints[origin] = parser_fingerprint_for_origin(origin) + + raw_evidence = [ + { + "raw_id": str(row[0]), + "origin": str(row[1]), + "capture_mode": None if row[2] is None else str(row[2]), + "native_id": None if row[3] is None else str(row[3]), + "source_path": str(row[4]), + "source_index": int(row[5]), + "blob_hash": bytes(row[6]).hex(), + "blob_size": int(row[7]), + "acquired_at_ms": int(row[8]), + "logical_source_key": None if row[9] is None else str(row[9]), + "revision_kind": str(row[10]), + "source_revision": None if row[11] is None else str(row[11]), + "predecessor_source_revision": None if row[12] is None else str(row[12]), + "predecessor_raw_id": None if row[13] is None else str(row[13]), + "baseline_raw_id": None if row[14] is None else str(row[14]), + "append_start_offset": None if row[15] is None else int(row[15]), + "append_end_offset": None if row[16] is None else int(row[16]), + "acquisition_generation": None if row[17] is None else int(row[17]), + "revision_authority": str(row[18]), + "revision_authority_evidence": None if row[19] is None else str(row[19]), + "parser_fingerprint": parser_fingerprints[str(row[1])], + "lowering_fingerprint": lowering, + } + for row in raw_rows + ] return { "raw_ids": list(expanded), "logical_source_keys": list(logical_keys), + "raw_session_evidence": raw_evidence, "raw_session_memberships": rows, } @@ -1106,8 +1265,23 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde log_mapped_bytes_budget_check(logger, check_mapped_bytes_budget_against_cgroup_limit()) validate_rebuild_index_request(request) root = request.archive_root + # A prior owned pass may have left its token in this task's context. The + # new request must establish its own receipt-bound token at the first + # validation rather than accidentally reusing another archive/pass. + _ACTIVE_EXTERNAL_INVENTORY_TOKEN.set(None) require_rebuild_schema_currency(root) - consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + initial_provenance_error: RebuildProvenanceError | None = None + try: + consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + except RebuildProvenanceError as exc: + if request.operation_id is None: + raise + # A resumable operation may need to be retired because this admission + # failed, but that lifecycle mutation must wait until both ownership + # boundaries are held. Control-flow exceptions are intentionally not + # caught here and therefore never change resumability. + initial_provenance_error = exc + consumed_evidence = {} location = ArchiveLocation.resolve(root) # The joined raw-frontier projection is rooted at the co-located active # index. A split-root canary intentionally points that index elsewhere and @@ -1145,12 +1319,34 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde try: assert_owns_archive_location(owned, location) require_rebuild_schema_currency(root) - consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + if initial_provenance_error is None: + consumed_evidence = _validate_rebuild_provenance_receipt( + root, + request.schema_inference_receipt_path, + inventory_token=cast( + dict[str, object], consumed_evidence.get("external_ground_truth_inventory_token", {}) + ), + ) # The lease is itself lifecycle state guarded by the provenance gate. # Revalidate again under the lease immediately before the owned body # can create or mutate a candidate/transaction. with RebuildLease(root): - consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) + if initial_provenance_error is not None: + _mark_rebuild_transaction_stale_after_provenance_failure( + root, request.operation_id, initial_provenance_error + ) + raise initial_provenance_error + try: + consumed_evidence = _validate_rebuild_provenance_receipt( + root, + request.schema_inference_receipt_path, + inventory_token=cast( + dict[str, object], consumed_evidence.get("external_ground_truth_inventory_token", {}) + ), + ) + except RebuildProvenanceError as exc: + _mark_rebuild_transaction_stale_after_provenance_failure(root, request.operation_id, exc) + raise return await _rebuild_index_from_source_owned( request, root=root, owned=owned, consumed_evidence=consumed_evidence ) @@ -1183,6 +1379,8 @@ async def _rebuild_index_from_source_owned( from polylogue.storage.repair import repair_session_insights generation_store = IndexGenerationStore(owned.location) + inventory_token = cast(dict[str, object], consumed_evidence.get("external_ground_truth_inventory_token", {})) + _ACTIVE_EXTERNAL_INVENTORY_TOKEN.set(inventory_token) # ``rebuild_index_from_source`` already acquired this root's # ``RebuildLease`` before any operation mutation. Retain this scope only # to preserve the body's indentation and make the outer ownership boundary @@ -1217,9 +1415,10 @@ async def _rebuild_index_from_source_owned( generation_store, request, root, + inventory_token=inventory_token, ) transaction_created_here = True - if transaction.status in {"promoted", "stale"}: + if transaction.status in {"promoted", "promoted-attestation-failed", "stale"}: raise RuntimeError( f"rebuild operation {transaction.operation_id} is {transaction.status}; start a new operation" ) @@ -1309,6 +1508,7 @@ async def _rebuild_index_from_source_owned( receipt_path=request.schema_inference_receipt_path, source_snapshot=str(consumed_evidence.get("source_snapshot", "")), consumed_evidence=consumed_evidence, + external_inventory_token=inventory_token, ) precreate_provenance.validate() generation = generation_store.create(source_snapshot=rebuild_source_evidence_snapshot(root)) @@ -1337,6 +1537,7 @@ async def _rebuild_index_from_source_owned( receipt_path=request.schema_inference_receipt_path, source_snapshot=str(consumed_evidence.get("source_snapshot", "")), consumed_evidence=consumed_evidence, + external_inventory_token=inventory_token, ) sharded_replay = request.shard_count > 1 and len(selected_raw_ids) >= request.shard_count source_drifted = False @@ -1728,6 +1929,11 @@ def _check_pass_deadline() -> None: stage=f"bulk_build.{stage}", elapsed_s=round(elapsed_s, 3), ) + # Blob bytes are checked after replay and bulk derivation, before + # candidate acceptance/readiness can observe a corrupted source. + # The receipt's source.db digest is only a binding; BlobStore is + # the byte-level authority for the current referenced universe. + _validate_before_derived_state(provenance, verify_blob_integrity=True) terminal_started_at = time.perf_counter() # polylogue-t0m73: the index-only reindex acceptance gate -- every # ground-truth check whose universe is satisfiable from a @@ -1817,6 +2023,15 @@ def _check_pass_deadline() -> None: else "blocked surfaces: " + ", ".join(blocked) ) raise RuntimeError(f"inactive generation {generation.generation_id} is not exact-ready; {detail}") + # Candidate readiness is the last fallible validation boundary. + # Verify the bytes named by source.db through BlobStore here, + # before a pointer can be flipped, rather than treating a source + # tier hash or an earlier receipt as proof of current blob bytes. + _validate_before_derived_state( + provenance, + verify_blob_integrity=True, + refresh_blob_integrity=True, + ) if transaction is not None: transaction = _checkpoint_rebuild_transaction_after_receipt_validation( generation_store, @@ -1843,13 +2058,46 @@ def _check_pass_deadline() -> None: elapsed_s=round(terminal_timings_s["terminal.promote"], 3), ) if transaction is not None: - transaction = _checkpoint_rebuild_transaction_after_receipt_validation( - generation_store, - transaction, - root, - request.schema_inference_receipt_path, - status="promoted", - ) + # The pointer is active now. Persist only a non-failing + # lifecycle transition after this point. In particular, + # do not call the receipt validator after promotion: a + # changed external root or blob must never turn an active + # generation back into a resumable ``ready`` candidate. + attestation: dict[str, object] + try: + attestation = { + "status": "passed", + "generation_id": generation.generation_id, + "generation_state": generation.state, + "active_pointer": str(generation_store.active_pointer), + } + transaction = generation_store.checkpoint_transaction( + transaction, + status="promoted", + post_promotion_attestation=attestation, + ) + except Exception as attestation_error: + source_drifted = True + try: + transaction = generation_store.load_transaction(transaction.operation_id) + transaction = generation_store.checkpoint_transaction( + transaction, + status="promoted-attestation-failed", + error=str(attestation_error), + post_promotion_attestation={ + "status": "failed", + "generation_id": generation.generation_id, + "generation_state": "active", + "error": str(attestation_error), + }, + ) + except Exception as recovery_error: + attestation_error.add_note( + "active generation post-promotion attestation checkpoint failed: " + f"{type(recovery_error).__name__}: {recovery_error}" + ) + raise + raise except Exception as exc: fresh_provenance_failure = isinstance(exc, RebuildProvenanceError) and ( transaction_created_here or transaction is None or sharded_replay @@ -1923,15 +2171,67 @@ def _check_pass_deadline() -> None: ), consumed_evidence=consumed_evidence, ) - _persist_candidate_receipt(generation, final_receipt.to_dict()) + try: + _persist_candidate_receipt(generation, final_receipt.to_dict()) + except BaseException as attestation_error: + if transaction is not None and transaction.status in {"promoted", "promoted-attestation-failed"}: + source_drifted = True + try: + transaction = generation_store.load_transaction(transaction.operation_id) + transaction = generation_store.checkpoint_transaction( + transaction, + status="promoted-attestation-failed", + error=str(attestation_error), + post_promotion_attestation={ + "status": "failed", + "generation_id": generation.generation_id, + "generation_state": "active", + "error": str(attestation_error), + }, + ) + except BaseException as recovery_error: + attestation_error.add_note( + "active generation candidate-receipt attestation checkpoint failed: " + f"{type(recovery_error).__name__}: {recovery_error}" + ) + raise if transaction is not None: - _save_rebuild_pass_receipt_after_receipt_validation( - generation_store, - transaction.operation_id, - final_receipt, - root, - request.schema_inference_receipt_path, - ) + if transaction.status in {"promoted", "promoted-attestation-failed"}: + # Promotion already crossed the pointer boundary. Receipt + # persistence here is an attestation record, not another + # admission gate that can demote or leave the active + # generation resumable. + try: + generation_store.save_pass_receipt(transaction.operation_id, final_receipt.to_dict()) + except BaseException as attestation_error: + source_drifted = True + try: + transaction = generation_store.load_transaction(transaction.operation_id) + transaction = generation_store.checkpoint_transaction( + transaction, + status="promoted-attestation-failed", + error=str(attestation_error), + post_promotion_attestation={ + "status": "failed", + "generation_id": generation.generation_id, + "generation_state": "active", + "error": str(attestation_error), + }, + ) + except BaseException as recovery_error: + attestation_error.add_note( + "active generation receipt-attestation checkpoint failed: " + f"{type(recovery_error).__name__}: {recovery_error}" + ) + raise + else: + _save_rebuild_pass_receipt_after_receipt_validation( + generation_store, + transaction.operation_id, + final_receipt, + root, + request.schema_inference_receipt_path, + ) except BaseException as exc: if transaction is None: _cleanup_nonresumable_generation_failure( diff --git a/polylogue/maintenance/schema_inference_gate.py b/polylogue/maintenance/schema_inference_gate.py index 668666ea41..bf19d9d4f2 100644 --- a/polylogue/maintenance/schema_inference_gate.py +++ b/polylogue/maintenance/schema_inference_gate.py @@ -803,6 +803,8 @@ def _full_blob_hash_evidence(archive_root: Path, *, referenced_hashes: set[str]) errors.append("BlobStore.verify_all truncated before a complete result") if missing_references: errors.append("referenced source blobs are absent from the verified blob root") + failed_hashes = {failure.hash for failure in verification.failures if failure.hash} + verified_hashes = set() if verification.truncated else canonical_hashes - failed_hashes return { "passed": not errors, "verifier": { @@ -822,11 +824,61 @@ def _full_blob_hash_evidence(archive_root: Path, *, referenced_hashes: set[str]) for failure in verification.failures ], "missing_references": _sample(missing_references, DEFAULT_SAMPLE_LIMIT), + "referenced_blob_integrity_snapshot": _referenced_blob_integrity_snapshot( + archive_root, referenced_hashes=referenced_hashes, verified_hashes=verified_hashes + ), "errors": errors, "reason": "; ".join(errors) if errors else None, } +def _referenced_blob_integrity_snapshot( + archive_root: Path, + *, + referenced_hashes: set[str], + verified_hashes: set[str] | None = None, +) -> dict[str, object]: + """Verify and bind the current bytes for every source-referenced blob. + + The source.db hash is the expected content identity. When + ``verified_hashes`` is supplied, it is the completed ``verify_all`` result + and this function records that evidence without rehashing the referenced + blobs. The fallback performs direct verification for callers that do not + already own a complete scan. + """ + + store = BlobStore(archive_root / "blob") + entries: list[dict[str, object]] = [] + for blob_hash in sorted(referenced_hashes): + path = store.blob_path(blob_hash) + verified = blob_hash in verified_hashes if verified_hashes is not None else store.verify(blob_hash) + item: dict[str, object] = {"blob_hash": blob_hash, "verified": verified} + try: + item["size"] = path.stat().st_size + except OSError as exc: + item["stat_error"] = str(exc) + entries.append(item) + encoded = json.dumps(entries, sort_keys=True, separators=(",", ":")).encode("utf-8") + return { + "algorithm": "sha256-referenced-blob-integrity-v2", + "verifier": ( + "polylogue.storage.blob_store.BlobStore.verify_all" + if verified_hashes is not None + else "polylogue.storage.blob_store.BlobStore.verify" + ), + "referenced_count": len(entries), + "passed": all(bool(item.get("verified")) for item in entries), + "entries": entries, + "digest": hashlib.sha256(encoded).hexdigest(), + } + + +def _semantic_referenced_blob_integrity_snapshot(snapshot: Mapping[str, object]) -> dict[str, object]: + """Remove verifier provenance before comparing equivalent blob evidence.""" + + return {key: value for key, value in snapshot.items() if key != "verifier"} + + def _iter_ground_truth_files(root: Path) -> Iterable[Path]: if root.is_file(): yield root @@ -868,6 +920,40 @@ def _external_inventory(roots: Sequence[Path]) -> list[_ExternalGroundTruthFile] return inventory +def _external_inventory_change_detector(roots: Sequence[Path]) -> dict[str, object]: + """Return metadata evidence that detects writes before rehashing. + + The detector walks and stats files without opening them. ``ctime_ns`` is + included because an in-place rewrite can preserve size, inode, and mtime; + a changed detector then triggers the authoritative full inventory hash. + The detector is only a change signal, never content identity by itself. + """ + + entries: list[dict[str, object]] = [] + for root_index, root in enumerate(sorted({path.resolve() for path in roots}, key=str)): + for path in _iter_ground_truth_files(root): + resolved_path = path.resolve() + stat = resolved_path.stat() + relative_path = resolved_path.name if root.is_file() else resolved_path.relative_to(root).as_posix() + entries.append( + { + "root_index": root_index, + "relative_path": relative_path, + "size": stat.st_size, + "inode": stat.st_ino, + "mtime_ns": stat.st_mtime_ns, + "ctime_ns": stat.st_ctime_ns, + } + ) + entries.sort(key=lambda item: (int(cast(int, item["root_index"])), str(item["relative_path"]))) + encoded = json.dumps(entries, sort_keys=True, separators=(",", ":")).encode("utf-8") + return { + "algorithm": "sha256-stat-inventory-v2", + "entry_count": len(entries), + "digest": hashlib.sha256(encoded).hexdigest(), + } + + def _external_inventory_records(inventory: Sequence[_ExternalGroundTruthFile]) -> list[dict[str, object]]: return [ { @@ -1220,6 +1306,7 @@ def _ground_truth_evidence( "cross_origin_mismatches": cross_origin_mismatches[:DEFAULT_SAMPLE_LIMIT], "provenance": provenance, "external_inventory": _external_inventory_records(inventory), + "inventory_change_detector": _external_inventory_change_detector(declared_roots), "raw_external_mapping": mapping, "passed": not missing and not unmatched_mapping, } @@ -1594,13 +1681,48 @@ def _int_or_zero(value: object) -> int: return value if isinstance(value, int) and not isinstance(value, bool) else 0 -def _current_external_ground_truth_digest(archive_root: Path, ground_truth: Mapping[str, object]) -> str: +def _external_inventory_token( + archive_root: Path, + *, + receipt_path: Path, + receipt_nonce: object, + source_snapshot: object, + external_ground_truth_digest: object, + ground_truth: Mapping[str, object], +) -> dict[str, object]: + origins: dict[str, object] = {} + for origin, raw_evidence in _as_dict(ground_truth.get("origins")).items(): + evidence = _as_dict(raw_evidence) + origins[origin] = { + "exempt": bool(evidence.get("exempt")), + "declared_roots": evidence.get("declared_roots", []), + "inventory_change_detector": evidence.get("inventory_change_detector"), + } + return { + "schema": "polylogue.external-ground-truth-inventory-token.v1", + "archive_root": str(archive_root.resolve()), + "receipt_path": str(receipt_path.resolve()), + "receipt_nonce": receipt_nonce, + "source_snapshot": source_snapshot, + "external_ground_truth_digest": external_ground_truth_digest, + "origins": origins, + } + + +def _current_external_ground_truth_digest( + archive_root: Path, + ground_truth: Mapping[str, object], + *, + inventory_token: Mapping[str, object] | None = None, +) -> tuple[str, dict[str, dict[str, object]]]: origins = _as_dict(ground_truth.get("origins")) current: dict[str, object] = {} + current_token_origins: dict[str, dict[str, object]] = {} with open_readonly_connection(archive_root / "source.db") as source: for origin, raw_evidence in origins.items(): evidence = _as_dict(raw_evidence) if bool(evidence.get("exempt")): + current_token_origins[origin] = {"exempt": True} current[origin] = { "exempt": True, "reason": evidence.get("reason"), @@ -1616,13 +1738,34 @@ def _current_external_ground_truth_digest(archive_root: Path, ground_truth: Mapp raise SchemaInferenceGateError( f"ground truth roots for {origin} are unavailable: {', '.join(unavailable)}" ) + detector = _external_inventory_change_detector(resolved_roots) + current_token_origins[origin] = { + "exempt": False, + "declared_roots": [str(root) for root in resolved_roots], + "inventory_change_detector": detector, + } + token_origin = _as_dict(_as_dict(inventory_token).get("origins")).get(origin) + token_origin = _as_dict(token_origin) + if ( + inventory_token is not None + and token_origin.get("declared_roots") == [str(root) for root in resolved_roots] + and token_origin.get("inventory_change_detector") == detector + ): + current[origin] = { + "declared_roots": [str(root) for root in resolved_roots], + "external_inventory": evidence.get("external_inventory"), + "raw_external_mapping": evidence.get("raw_external_mapping"), + "inventory_change_detector": detector, + } + continue inventory = _external_inventory(resolved_roots) current[origin] = { "declared_roots": [str(root) for root in resolved_roots], "external_inventory": _external_inventory_records(inventory), "raw_external_mapping": _raw_external_mapping(source, origin=origin, inventory=inventory, exempt=False), + "inventory_change_detector": detector, } - return _canonical_external_ground_truth_digest(current) + return _canonical_external_ground_truth_digest(current), current_token_origins def validate_schema_inference_receipt( @@ -1630,6 +1773,9 @@ def validate_schema_inference_receipt( receipt_path: Path | None = None, *, now: datetime | None = None, + inventory_token: dict[str, object] | None = None, + verify_blob_integrity: bool = False, + verified_blob_integrity_snapshot: dict[str, object] | None = None, ) -> dict[str, object]: """Validate the evidence a rebuild is authorized to consume. @@ -1649,6 +1795,7 @@ def validate_schema_inference_receipt( raise SchemaInferenceGateError("schema-inference receipt root must be an object") errors: list[str] = [] + verified_snapshot: dict[str, object] | None = None if document.get("schema") != RECEIPT_SCHEMA: errors.append(f"schema must be {RECEIPT_SCHEMA!r}") if document.get("verdict") != "PASS": @@ -1708,6 +1855,7 @@ def validate_schema_inference_receipt( if _as_dict(document.get(field)).get("passed") is not True: errors.append(f"receipt subgate {field} is not PASS") + active_token: dict[str, object] | None = inventory_token ground_truth = document.get("ground_truth_inputs") if not isinstance(ground_truth, dict): errors.append("receipt ground_truth_inputs are missing or malformed") @@ -1722,11 +1870,58 @@ def validate_schema_inference_receipt( ) if recorded_digest != recorded_structure_digest: errors.append("receipt raw external mapping or inventory does not match its digest") - current_digest = _current_external_ground_truth_digest(root, ground_truth) + receipt_token = _as_dict(document.get("external_ground_truth_inventory_token")) + if inventory_token: + expected_token_fields = { + "schema": "polylogue.external-ground-truth-inventory-token.v1", + "archive_root": str(root.resolve()), + "receipt_path": str(path.resolve()), + "receipt_nonce": document.get("receipt_nonce"), + "source_snapshot": document.get("source_snapshot"), + "external_ground_truth_digest": recorded_digest, + } + if any(inventory_token.get(key) != value for key, value in expected_token_fields.items()): + errors.append("receipt external ground-truth inventory token is not bound to this pass") + current_digest, current_token_origins = _current_external_ground_truth_digest( + root, ground_truth, inventory_token=active_token + ) if recorded_digest != current_digest: errors.append("external ground-truth corpus changed since the receipt was produced") + if not inventory_token and receipt_token: + active_token = receipt_token + if active_token: + refreshed_token = dict(active_token) + refreshed_token["origins"] = { + origin: { + **_as_dict(_as_dict(active_token.get("origins")).get(origin)), + **current_origin, + } + for origin, current_origin in current_token_origins.items() + } + refreshed_token["external_ground_truth_digest"] = current_digest + active_token = refreshed_token except (OSError, SchemaInferenceGateError, sqlite3.Error) as exc: errors.append(str(exc)) + if verify_blob_integrity: + try: + with open_readonly_connection(root / "source.db") as source: + referenced_hashes = _referenced_blob_hashes(source) + verified_snapshot = ( + verified_blob_integrity_snapshot + if verified_blob_integrity_snapshot is not None + else _referenced_blob_integrity_snapshot(root, referenced_hashes=referenced_hashes) + ) + recorded_blob_snapshot = _as_dict( + _as_dict(document.get("full_blob_hash_verification")).get("referenced_blob_integrity_snapshot") + ) + if not verified_snapshot.get("passed"): + errors.append("referenced source blob integrity verification failed before candidate readiness") + if recorded_blob_snapshot and _semantic_referenced_blob_integrity_snapshot( + recorded_blob_snapshot + ) != _semantic_referenced_blob_integrity_snapshot(verified_snapshot): + errors.append("receipt referenced source blob integrity snapshot changed") + except (OSError, SchemaInferenceGateError, sqlite3.Error, ValueError) as exc: + errors.append(f"could not verify referenced source blob integrity: {exc}") try: with open_readonly_connection(root / "source.db") as source: source_origins = {str(row[0]) for row in source.execute("SELECT DISTINCT origin FROM raw_sessions")} @@ -1744,7 +1939,7 @@ def validate_schema_inference_receipt( raise SchemaInferenceGateError("schema-inference receipt rejected: " + "; ".join(errors)) identity = _as_dict(document.get("archive_identity")) - return { + result = { "receipt_path": str(path), "schema": document.get("schema"), "generated_at": document.get("generated_at"), @@ -1754,7 +1949,19 @@ def validate_schema_inference_receipt( "source_identity": document.get("source_identity"), "source_snapshot": document.get("source_snapshot"), "external_ground_truth_digest": document.get("external_ground_truth_digest"), + "external_ground_truth_inventory_token": active_token + or _external_inventory_token( + root, + receipt_path=path, + receipt_nonce=document.get("receipt_nonce"), + source_snapshot=document.get("source_snapshot"), + external_ground_truth_digest=document.get("external_ground_truth_digest"), + ground_truth=cast(Mapping[str, object], document.get("ground_truth_inputs", {})), + ), } + if verified_snapshot is not None: + result["_verified_blob_integrity_snapshot"] = verified_snapshot + return result def schema_inference_gate_receipt_digest(payload: Mapping[str, object]) -> str: @@ -2022,11 +2229,12 @@ def _run_schema_inference_gate_locked( if not bool(_as_dict(entry).get("matches_expected")): reasons.append(f"{tier_name}.db schema identity is stale or does not match the packaged schema") archive_payload = _as_dict(final_schema_identity.get("archive")) + receipt_nonce = uuid4().hex payload: dict[str, object] = { "schema": RECEIPT_SCHEMA, "gate_version": GATE_VERSION, "generated_at": datetime.now(UTC).isoformat(), - "receipt_nonce": uuid4().hex, + "receipt_nonce": receipt_nonce, "sample_limit": sample_limit, "verdict": "PASS" if not reasons and passed_hard_gates and schema_identity_ok else "FAIL", "archive_root": str(root), @@ -2037,6 +2245,14 @@ def _run_schema_inference_gate_locked( }, "source_snapshot": source_snapshot, "external_ground_truth_digest": ground_truth.get("external_ground_truth_digest"), + "external_ground_truth_inventory_token": _external_inventory_token( + root, + receipt_path=safe_receipt_path, + receipt_nonce=receipt_nonce, + source_snapshot=source_snapshot, + external_ground_truth_digest=ground_truth.get("external_ground_truth_digest"), + ground_truth=ground_truth, + ), "archive_identity_digest": archive_payload.get("authority_identity_digest"), "schema_identity": final_schema_identity, "source_schema_identity": source_entry, diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 0a9ea5c36e..121ce6decd 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -186,6 +186,11 @@ class IndexRebuildTransaction: # transaction created before this field existed is treated as not yet # cleared and clears exactly once on its next resume. derived_stores_cleared: bool = False + # Promotion is a terminal lifecycle boundary. This attestation is + # written after the pointer flip without re-running fallible admission + # checks, so a post-flip observation failure cannot leave a resumable + # transaction claiming that its candidate is merely ready. + post_promotion_attestation: dict[str, object] | None = None @property def cursor(self) -> str | None: @@ -569,7 +574,7 @@ def save_transaction(self, transaction: IndexRebuildTransaction) -> IndexRebuild path = self._transaction_path(transaction.operation_id) path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(".json.tmp") - temporary.write_text(json.dumps(asdict(updated), indent=2, sort_keys=True), encoding="utf-8") + temporary.write_text(json.dumps(asdict(updated), indent=2, sort_keys=True, default=str), encoding="utf-8") os.replace(temporary, path) _fsync_directory(path.parent) return updated @@ -585,6 +590,7 @@ def checkpoint_transaction( processed_blob_bytes: int | None = None, error: str | None = None, derived_stores_cleared: bool | None = None, + post_promotion_attestation: dict[str, object] | None = None, ) -> IndexRebuildTransaction: """Persist one state transition without changing candidate ownership.""" return self.save_transaction( @@ -606,6 +612,9 @@ def checkpoint_transaction( "derived_stores_cleared": derived_stores_cleared if derived_stores_cleared is not None else transaction.derived_stores_cleared, + "post_promotion_attestation": post_promotion_attestation + if post_promotion_attestation is not None + else transaction.post_promotion_attestation, } ) ) @@ -1150,12 +1159,15 @@ def rebuild_source_evidence_snapshot(archive_root: Path) -> str: Parse, validation, and revision-governance state are rebuild outputs or post-acquisition interpretation. They can legitimately change while the rebuild runs, so they must never invalidate its before/after source proof. - The selected columns capture durable raw identity, membership, and bytes, - together with acquired raw-payload references and capture-mode observations, - in deterministic order. + The selected columns capture every durable raw and authority field already + consumed by revision backfill, together with the parser/lowering semantic + fingerprints that give those fields meaning. Equivalent receipt snapshots + use the same ordered evidence below, so a resumable pass cannot cross a + changed revision authority boundary. """ import hashlib + from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin from polylogue.storage.blob_store import BlobStore digest = hashlib.sha256() @@ -1164,12 +1176,18 @@ def rebuild_source_evidence_snapshot(archive_root: Path) -> str: """ SELECT raw_id, origin, capture_mode, native_id, source_path, source_index, blob_hash, blob_size, acquired_at_ms, - file_mtime_ms + file_mtime_ms, logical_source_key, revision_kind, + source_revision, predecessor_source_revision, + predecessor_raw_id, baseline_raw_id, append_start_offset, + append_end_offset, acquisition_generation, + revision_authority, revision_authority_evidence FROM raw_sessions ORDER BY raw_id """ ) + origins: set[str] = set() for row in rows: + origins.add(str(row[1])) for value in row: if value is None: encoded = b"n" @@ -1181,6 +1199,18 @@ def rebuild_source_evidence_snapshot(archive_root: Path) -> str: encoded = b"i" + str(value).encode() digest.update(len(encoded).to_bytes(8, "big")) digest.update(encoded) + sorted_origins = sorted(origins) + digest.update(b"parser-lowering-semantic-fingerprints\0") + digest.update(b"lowering\0") + lowering = lowering_fingerprint() + digest.update(len(lowering).to_bytes(8, "big")) + digest.update(lowering.encode()) + for origin in sorted_origins: + parser = parser_fingerprint_for_origin(origin) + for value in (origin, parser): + encoded = value.encode() + digest.update(len(encoded).to_bytes(8, "big")) + digest.update(encoded) raw_blob_hashes = { bytes(row[0]).hex() if isinstance(row[0], (bytes, bytearray, memoryview)) else str(row[0]) for row in conn.execute("SELECT DISTINCT blob_hash FROM raw_sessions ORDER BY blob_hash") diff --git a/tests/infra/rebuild_receipt.py b/tests/infra/rebuild_receipt.py index 6bbcc2dced..7709fb96bc 100644 --- a/tests/infra/rebuild_receipt.py +++ b/tests/infra/rebuild_receipt.py @@ -61,6 +61,7 @@ def write_valid_rebuild_receipt( "exempt": False, "declared_roots": [str(external_root)], "external_inventory": inventory_records, + "inventory_change_detector": gate._external_inventory_change_detector((external_root,)), "raw_external_mapping": mapping, "passed": all(item["disposition"] == "matched-external" for item in mapping), } diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index 67b1397ede..b023bbec92 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -2382,6 +2382,61 @@ def fake_urlopen(request: object, *, timeout: int) -> Response: assert "Classified:" in result.output +def test_rebuild_index_daemon_resolves_relative_schema_receipt_before_post( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Relative receipt references are resolved before daemon serialization. + + Anti-vacuity: removing CLI-side resolution sends the relative filename in + the captured production daemon payload. + """ + root = cli_workspace["archive_root"] + absolute_receipt = write_valid_rebuild_receipt(root, tmp_path / "relative-receipt.json") + monkeypatch.chdir(tmp_path) + captured: dict[str, object] = {} + + class Response: + def read(self) -> bytes: + return json.dumps( + { + "archive_root": str(root), + "classified_full_count": 0, + "replayed_logical_source_count": 0, + "quarantined_raw_count": 0, + } + ).encode() + + def __enter__(self) -> Response: + return self + + def __exit__(self, *_args: object) -> None: + return None + + def fake_urlopen(request: object, *, timeout: int) -> Response: + captured["body"] = json.loads(request.data) # type: ignore[attr-defined] + return Response() + + monkeypatch.setattr(maintenance_rebuild_index, "urlopen", fake_urlopen) + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "rebuild-index", + "--daemon", + "--schema-inference-receipt", + absolute_receipt.name, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + body = captured["body"] + assert isinstance(body, dict) + assert body["schema_inference_receipt_path"] == str(absolute_receipt.resolve()) + + @pytest.mark.parametrize("selection_args", [["--only-missing"], ["--raw-id", "raw-a"]]) def test_partial_rebuild_requires_no_promote_before_archive_mutation( cli_workspace: dict[str, Path], cli_runner: CliRunner, selection_args: list[str] @@ -2448,6 +2503,22 @@ def test_rebuild_index_full_source_resumes_one_candidate_until_terminal_promotio source_path=f"{native_id}.jsonl", acquired_at_ms=acquired_at_ms, ) + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + UPDATE raw_sessions + SET logical_source_key = CASE + WHEN source_path = 'first.jsonl' THEN 'codex:first' + ELSE 'codex:second' + END, + revision_kind = 'full', + source_revision = raw_id, + baseline_raw_id = raw_id, + acquisition_generation = 0, + revision_authority = 'byte_proven' + """ + ) + source.commit() receipt_path = write_valid_rebuild_receipt(root, root.parent / "schema-inference-gate-receipt.json") monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) @@ -2528,6 +2599,22 @@ def test_rebuild_index_persists_durable_pass_receipt_alongside_transaction( source_path=f"{native_id}.jsonl", acquired_at_ms=acquired_at_ms, ) + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + UPDATE raw_sessions + SET logical_source_key = CASE + WHEN source_path = 'first.jsonl' THEN 'codex:first' + ELSE 'codex:second' + END, + revision_kind = 'full', + source_revision = raw_id, + baseline_raw_id = raw_id, + acquisition_generation = 0, + revision_authority = 'byte_proven' + """ + ) + source.commit() receipt_path = write_valid_rebuild_receipt(root, root.parent / "schema-inference-gate-receipt.json") monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) diff --git a/tests/unit/maintenance/test_rebuild_index_phase_timing.py b/tests/unit/maintenance/test_rebuild_index_phase_timing.py index be2511660f..bbe9c82f16 100644 --- a/tests/unit/maintenance/test_rebuild_index_phase_timing.py +++ b/tests/unit/maintenance/test_rebuild_index_phase_timing.py @@ -63,7 +63,7 @@ def _codex_session(native_id: str, messages: tuple[tuple[str, str], ...]) -> byt return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) -def _seed_distinct_codex_sessions(root: Path, count: int) -> list[str]: +def _seed_distinct_codex_sessions(root: Path, count: int, *, monkeypatch: pytest.MonkeyPatch) -> list[str]: initialize_active_archive_root(root) raw_ids: list[str] = [] with ArchiveStore.open_existing(root, read_only=False) as archive: @@ -77,6 +77,26 @@ def _seed_distinct_codex_sessions(root: Path, count: int) -> list[str]: acquired_at_ms=index + 1, ) ) + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + UPDATE raw_sessions + SET logical_source_key = CASE + WHEN source_path LIKE '%/0.jsonl' THEN 'codex:sess-0' + WHEN source_path LIKE '%/1.jsonl' THEN 'codex:sess-1' + WHEN source_path LIKE '%/2.jsonl' THEN 'codex:sess-2' + ELSE 'codex:sess-3' + END, + revision_kind = 'full', + source_revision = raw_id, + baseline_raw_id = raw_id, + acquisition_generation = 0, + revision_authority = 'byte_proven' + """ + ) + source.commit() + receipt_path = write_valid_rebuild_receipt(root, root.parent / f"{root.name}-schema-receipt.json") + monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) return raw_ids @@ -85,7 +105,7 @@ def test_replayed_receipt_carries_selection_phase_timing(tmp_path: Path, monkeyp raw-id selection took, distinct from replay/parse/apply/terminal costs.""" root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 2) + _seed_distinct_codex_sessions(root, 2, monkeypatch=monkeypatch) receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=True)) @@ -106,7 +126,7 @@ def test_real_receipt_accounts_for_all_rebuild_phases(tmp_path: Path, monkeypatc """ root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 3) + _seed_distinct_codex_sessions(root, 3, monkeypatch=monkeypatch) receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=True)) @@ -156,7 +176,7 @@ def test_phase_rollups_are_bound_to_production_stage_timing_mutation( """The receipt's phase rollups cannot pass from selection timing alone.""" root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 2) + _seed_distinct_codex_sessions(root, 2, monkeypatch=monkeypatch) original = rebuild_index._repopulate_bulk_build_derived_state @@ -184,7 +204,7 @@ def test_archive_wide_derived_refresh_runs_once_at_terminal_boundary( """ root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 4) + _seed_distinct_codex_sessions(root, 4, monkeypatch=monkeypatch) receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") calls = 0 @@ -216,7 +236,7 @@ def test_partial_rebuild_cannot_complete_when_candidate_omits_source_document( """ root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - raw_ids = _seed_distinct_codex_sessions(root, 2) + raw_ids = _seed_distinct_codex_sessions(root, 2, monkeypatch=monkeypatch) with pytest.raises(RuntimeError, match="reindex acceptance gate failed.*corpus-absences"): rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, raw_ids=(raw_ids[0],), promote=False)) @@ -226,17 +246,17 @@ def test_partial_rebuild_cannot_complete_when_candidate_omits_source_document( @pytest.mark.parametrize( - ("violation", "expected_check"), + ("violation", "expected_error"), ( - ("attachment", "corpus-attachment-fidelity"), - ("revision", "corpus-revision-fidelity"), + ("attachment", "reindex acceptance gate failed.*corpus-attachment-fidelity"), + ("revision", "referenced source blob integrity verification failed"), ), ) def test_rebuild_corpus_gate_blocks_mutated_inactive_candidate( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, violation: str, - expected_check: str, + expected_error: str, ) -> None: """The real rebuild acceptance stage rejects each corrupted candidate. @@ -249,7 +269,7 @@ def test_rebuild_corpus_gate_blocks_mutated_inactive_candidate( """ root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 1) + _seed_distinct_codex_sessions(root, 1, monkeypatch=monkeypatch) original_repopulate = rebuild_index._repopulate_bulk_build_derived_state def corrupt_candidate_before_acceptance(index_path: Path) -> dict[str, float]: @@ -296,7 +316,7 @@ def corrupt_candidate_before_acceptance(index_path: Path) -> dict[str, float]: monkeypatch.setattr(rebuild_index, "_repopulate_bulk_build_derived_state", corrupt_candidate_before_acceptance) - with pytest.raises(RuntimeError, match=f"reindex acceptance gate failed.*{expected_check}"): + with pytest.raises(RuntimeError, match=expected_error): rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=False)) with sqlite3.connect(root / "index.db") as conn: @@ -309,7 +329,7 @@ def test_deferred_pass_cost_carries_selection_phase_timing(tmp_path: Path, monke not a different shape for the deferred path.""" root = tmp_path / "archive" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed_distinct_codex_sessions(root, 2) + _seed_distinct_codex_sessions(root, 2, monkeypatch=monkeypatch) # A sub-millisecond deadline truncates to pass_deadline_ms=0, so the # first between-cohorts check always trips deterministically (see diff --git a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py index 7483c21d42..1a1ead2f7e 100644 --- a/tests/unit/maintenance/test_rebuild_index_provenance_gate.py +++ b/tests/unit/maintenance/test_rebuild_index_provenance_gate.py @@ -2,7 +2,9 @@ from __future__ import annotations +import asyncio import json +import os import sqlite3 import threading from collections.abc import Callable @@ -12,14 +14,19 @@ import pytest import polylogue.maintenance.rebuild_index as rebuild_index_module +import polylogue.maintenance.schema_inference_gate as schema_gate_module import polylogue.maintenance.sharded_rebuild as sharded_rebuild_module +import polylogue.sources.origin_specs as origin_specs_module +import polylogue.storage.index_generation as index_generation_module from polylogue.archive.revision_authority import RawRevisionAuthority, RawRevisionEnvelope, RawRevisionKind from polylogue.config import Config from polylogue.core.enums import Provider +from polylogue.daemon import bulk_rebuild as bulk_rebuild_module from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync from polylogue.maintenance.sharded_rebuild import shard_raw_ids from polylogue.sources.revision_backfill import RebuildDeadlineExceededError -from polylogue.storage.archive_identity import OwnedArchiveLocation +from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation +from polylogue.storage.blob_store import BlobStore from polylogue.storage.index_generation import ( IndexGeneration, IndexGenerationStore, @@ -64,6 +71,13 @@ def _seed(root: Path, count: int = 2) -> None: authority=RawRevisionAuthority.ASSERTED, ), ) + # The receipt is intentionally taken from a source tier whose authority + # classification is already settled. Replay must not manufacture a + # baseline or rewrite asserted authority before the first checkpoint, so + # any later mutation of these bindings is a real stale-pass condition. + with sqlite3.connect(root / "source.db") as source: + source.execute("UPDATE raw_sessions SET baseline_raw_id = raw_id, revision_authority = 'byte_proven'") + source.commit() def _active_bytes(root: Path) -> bytes: @@ -686,3 +700,528 @@ def unexpected_insight_repair(*args: Any, **kwargs: Any) -> object: assert shard_ids <= set(discard_calls) assert len([generation_id for generation_id in discard_calls if generation_id in shard_ids]) == len(shard_ids) assert _generation_ids(root) == set() + + +def test_revision_authority_binding_stales_a_resumed_real_rebuild( + tmp_path: Path, +) -> None: + """A replay-affecting raw authority mutation cannot cross a page boundary. + + Anti-vacuity: removing the authority fields from the source binding leaves + the transaction paused instead of stale, failing the terminal assertion. + """ + root = tmp_path / "archive" + _seed(root, count=2) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + first = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_batch_size=1, + promote=False, + ) + ) + assert first.status == "paused" + assert first.transaction is not None + operation_id = str(first.transaction["operation_id"]) + before_evidence_digest = rebuild_source_evidence_snapshot(root) + with sqlite3.connect(root / "source.db") as source: + source.execute( + "UPDATE raw_sessions SET revision_authority_evidence = 'live_source_verification_v1' " + "WHERE raw_id = (SELECT raw_id FROM raw_sessions ORDER BY raw_id LIMIT 1)" + ) + source.commit() + assert rebuild_source_evidence_snapshot(root) != before_evidence_digest + + with pytest.raises(RuntimeError, match="source snapshot does not match"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=operation_id, + raw_batch_size=1, + promote=False, + ) + ) + transaction = IndexGenerationStore.for_archive_root(root).load_transaction(operation_id) + assert transaction.status == "stale" + + +def test_blob_bytes_changed_after_replay_are_rejected_before_candidate_readiness( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The final real readiness route verifies bytes, not only source.db hashes. + + Anti-vacuity: removing the readiness blob binding removes the dedicated + integrity failure asserted here, even though source.db still names a + parseable row. + """ + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + original_repopulate = rebuild_index_module._repopulate_bulk_build_derived_state + + def corrupt_after_replay(index_path: Path) -> dict[str, float]: + timings = original_repopulate(index_path) + with sqlite3.connect(root / "source.db") as source: + blob_hash = bytes(source.execute("SELECT blob_hash FROM raw_sessions LIMIT 1").fetchone()[0]).hex() + BlobStore(root / "blob").blob_path(blob_hash).write_bytes(b"parseable but corrupted") + return timings + + monkeypatch.setattr(rebuild_index_module, "_repopulate_bulk_build_derived_state", corrupt_after_replay) + with pytest.raises(RuntimeError, match="referenced source blob integrity verification failed"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + promote=False, + ) + ) + assert not list((root / ".index-generations").glob("gen-*")) + + +def test_pointer_flip_records_post_promotion_attestation_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A post-flip checkpoint fault leaves an active, terminally classified operation. + + Anti-vacuity: removing the terminal attestation transition makes the + injected post-flip failure escape instead of returning an active failed + attestation. + """ + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + original_checkpoint = IndexGenerationStore.checkpoint_transaction + + def fail_promoted_checkpoint(self: IndexGenerationStore, transaction: object, **kwargs: object) -> object: + if kwargs.get("status") == "promoted": + raise OSError("simulated post-promotion attestation failure") + return original_checkpoint(self, transaction, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(IndexGenerationStore, "checkpoint_transaction", fail_promoted_checkpoint) + with pytest.raises(OSError, match="simulated post-promotion attestation failure"): + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path, promote=True) + ) + + store = IndexGenerationStore.for_archive_root(root) + operation_id = next(path.stem for path in store.transactions_root.glob("*.json")) + transaction = store.load_transaction(operation_id) + assert transaction.status == "promoted-attestation-failed" + attestation = transaction.post_promotion_attestation + assert isinstance(attestation, dict) + assert attestation["status"] == "failed" + assert store.load(transaction.generation_id).state == "active" + assert store.active_pointer.resolve(strict=True) == Path(store.load(transaction.generation_id).index_path).resolve( + strict=True + ) + + +def test_daemon_reconciles_active_generation_after_both_attestation_checkpoints_fail( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + store = IndexGenerationStore.for_archive_root(root) + seeded_transaction = store.create_transaction( + source_snapshot=rebuild_source_evidence_snapshot(root), + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + ) + original_checkpoint = IndexGenerationStore.checkpoint_transaction + + def fail_attestation_checkpoint(self: IndexGenerationStore, transaction: object, **kwargs: object) -> object: + if kwargs.get("status") in {"promoted", "promoted-attestation-failed"}: + raise OSError("simulated double attestation checkpoint failure") + return original_checkpoint(self, transaction, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(IndexGenerationStore, "checkpoint_transaction", fail_attestation_checkpoint) + with pytest.raises(OSError, match="simulated double attestation checkpoint failure"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + promote=True, + ) + ) + + transaction = store.load_transaction(bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID) + assert transaction.status == "ready" + assert transaction.generation_id == seeded_transaction.generation_id + assert store.load(transaction.generation_id).state == "active" + + monkeypatch.undo() + reconciled = bulk_rebuild_module.resolve_or_start_daemon_bulk_rebuild_transaction( + root, + schema_inference_receipt_path=receipt_path, + ) + + assert reconciled.status == "promoted-attestation-failed" + assert reconciled.post_promotion_attestation == { + "status": "reconciled-after-restart", + "generation_id": transaction.generation_id, + "generation_state": "active", + } + + +def test_daemon_does_not_route_promoted_attestation_failure_back_to_rebuild( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A terminal active operation is not handed to the rebuild engine again.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + store = IndexGenerationStore.for_archive_root(root) + store.create_transaction( + source_snapshot=rebuild_source_evidence_snapshot(root), + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + ) + original_checkpoint = IndexGenerationStore.checkpoint_transaction + + def fail_promoted_checkpoint(self: IndexGenerationStore, transaction: object, **kwargs: object) -> object: + if kwargs.get("status") == "promoted": + raise OSError("simulated post-promotion attestation failure") + return original_checkpoint(self, transaction, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(IndexGenerationStore, "checkpoint_transaction", fail_promoted_checkpoint) + with pytest.raises(OSError, match="simulated post-promotion attestation failure"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + promote=True, + ) + ) + terminal = store.load_transaction(bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID) + assert terminal.status == "promoted-attestation-failed" + assert bulk_rebuild_module.has_resumable_daemon_bulk_rebuild_transaction(root) is False + + monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) + rebuild_called = False + + def unexpected_rebuild(*args: object, **kwargs: object) -> object: + nonlocal rebuild_called + rebuild_called = True + raise AssertionError("terminal daemon operation was routed back to rebuild") + + monkeypatch.setattr(rebuild_index_module, "rebuild_index_from_source_sync", unexpected_rebuild) + result = asyncio.run( + bulk_rebuild_module.run_daemon_bulk_rebuild_pass( + config=Config(archive_root=root, render_root=root / "render", sources=[]), + parse_stage=cast(Any, object()), + max_payload_bytes=1, + ) + ) + assert result is None + assert rebuild_called is False + active_generation = store.load(terminal.generation_id) + assert store.active_pointer.resolve(strict=True) == Path(active_generation.index_path).resolve(strict=True) + + +def test_daemon_retires_attestation_failure_after_source_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A changed source gets a fresh daemon transaction without replacing the active index.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + store = IndexGenerationStore.for_archive_root(root) + terminal = store.create_transaction( + source_snapshot=rebuild_source_evidence_snapshot(root), + operation_id=bulk_rebuild_module.DAEMON_BULK_REBUILD_OPERATION_ID, + ) + store.promote(store.load(terminal.generation_id)) + store.checkpoint_transaction(terminal, status="promoted-attestation-failed") + monkeypatch.setattr(bulk_rebuild_module, "rebuild_source_evidence_snapshot", lambda _root: "changed-source") + + replacement = bulk_rebuild_module.resolve_or_start_daemon_bulk_rebuild_transaction( + root, schema_inference_receipt_path=receipt_path + ) + + assert replacement.status == "running" + assert replacement.generation_id != terminal.generation_id + assert replacement.source_snapshot == "changed-source" + assert store.load(terminal.generation_id).state == "active" + + +def test_validation_rejection_cannot_stale_transaction_before_ownership( + tmp_path: Path, +) -> None: + """A live archive owner rejects the invocation before transaction mutation.""" + root = tmp_path / "archive" + _seed(root, count=2) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + first = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_batch_size=1, + promote=False, + ) + ) + assert first.transaction is not None + operation_id = str(first.transaction["operation_id"]) + store = IndexGenerationStore.for_archive_root(root) + before = store.load_transaction(operation_id) + + owner = OwnedArchiveLocation.acquire(ArchiveLocation.resolve(root)) + try: + with pytest.raises(ArchiveOwnershipError): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=operation_id, + raw_batch_size=1, + promote=False, + ) + ) + finally: + owner.release() + + assert store.load_transaction(operation_id) == before + + +@pytest.mark.parametrize( + "exception_type", + [KeyboardInterrupt, asyncio.CancelledError, SystemExit], + ids=["keyboard-interrupt", "cancelled", "control-flow-base-exception"], +) +def test_validation_control_flow_does_not_change_resumability( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, exception_type: type[BaseException] +) -> None: + root = tmp_path / "archive" + _seed(root, count=2) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + first = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_batch_size=1, + promote=False, + ) + ) + assert first.transaction is not None + operation_id = str(first.transaction["operation_id"]) + + def raise_control_flow(*args: object, **kwargs: object) -> object: + raise exception_type("validation interrupted") + + monkeypatch.setattr(rebuild_index_module, "_validate_rebuild_provenance_receipt", raise_control_flow) + with pytest.raises(exception_type, match="validation interrupted"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=operation_id, + raw_batch_size=1, + promote=False, + ) + ) + assert IndexGenerationStore.for_archive_root(root).load_transaction(operation_id).status == "paused" + + +def test_external_rewrite_with_preserved_size_inode_and_mtime_is_rehashed( + tmp_path: Path, +) -> None: + root = tmp_path / "archive" + _seed(root, count=2) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + first = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + raw_batch_size=1, + promote=False, + ) + ) + assert first.transaction is not None + operation_id = str(first.transaction["operation_id"]) + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + origin = receipt["ground_truth_inputs"]["origins"]["codex-session"] + external_path = Path(origin["declared_roots"][0]) / origin["external_inventory"][0]["relative_path"] + before = external_path.stat() + original = external_path.read_bytes() + replacement = bytes(byte ^ 0xFF for byte in original) + external_path.write_bytes(replacement) + os.utime(external_path, ns=(before.st_atime_ns, before.st_mtime_ns)) + after = external_path.stat() + assert after.st_ino == before.st_ino + assert after.st_size == before.st_size + assert after.st_mtime_ns == before.st_mtime_ns + assert replacement != original + + with pytest.raises(RuntimeError, match="external ground-truth corpus changed"): + rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + schema_inference_receipt_path=receipt_path, + operation_id=operation_id, + raw_batch_size=1, + promote=False, + ) + ) + + +def test_full_blob_verification_supplies_referenced_snapshot_without_rehash( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive" + _seed(root, count=1) + with sqlite3.connect(root / "source.db") as source: + referenced_hashes = {bytes(row[0]).hex() for row in source.execute("SELECT blob_hash FROM raw_sessions")} + verify_calls = 0 + original_verify = BlobStore.verify + + def count_verify(self: BlobStore, blob_hash: str) -> bool: + nonlocal verify_calls + verify_calls += 1 + return original_verify(self, blob_hash) + + monkeypatch.setattr(BlobStore, "verify", count_verify) + evidence = schema_gate_module._full_blob_hash_evidence(root, referenced_hashes=referenced_hashes) + assert evidence["passed"] is True + assert verify_calls == 0 + snapshot = cast(dict[str, object], evidence["referenced_blob_integrity_snapshot"]) + assert snapshot["verifier"] == "polylogue.storage.blob_store.BlobStore.verify_all" + assert snapshot["passed"] is True + + +def test_rebuild_reuses_verified_blob_snapshot_across_readiness_boundaries( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The two production readiness checks share one byte-verification result.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + calls = 0 + original_snapshot = schema_gate_module._referenced_blob_integrity_snapshot + + def count_snapshot(*args: object, **kwargs: object) -> dict[str, object]: + nonlocal calls + calls += 1 + return original_snapshot(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(schema_gate_module, "_referenced_blob_integrity_snapshot", count_snapshot) + result = rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path, promote=False) + ) + assert result.status == "replayed" + assert calls == 2 + + +def test_rebuild_rechecks_blob_bytes_at_final_readiness_boundary( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Corruption after insight materialization is caught before readiness.""" + root = tmp_path / "archive" + _seed(root, count=1) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "receipt.json") + from polylogue.storage import repair as repair_module + + original_repair = repair_module.repair_session_insights + + def corrupt_after_insights(*args: object, **kwargs: object) -> object: + result = cast(Any, original_repair)(*args, **kwargs) + with sqlite3.connect(root / "source.db") as source: + blob_hash = bytes(source.execute("SELECT blob_hash FROM raw_sessions LIMIT 1").fetchone()[0]).hex() + BlobStore(root / "blob").blob_path(blob_hash).write_bytes(b"corrupted after readiness precursor") + return result + + monkeypatch.setattr("polylogue.storage.repair.repair_session_insights", corrupt_after_insights) + with pytest.raises(RuntimeError, match="referenced source blob integrity verification failed"): + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path, promote=False) + ) + + +def test_replay_closure_caches_fingerprints_per_origin(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Replay evidence computes parser and lowering fingerprints once per closure.""" + root = tmp_path / "archive" + _seed(root, count=2) + lower_calls = 0 + parser_calls: list[str] = [] + original_lowering = origin_specs_module.lowering_fingerprint + original_parser = origin_specs_module.parser_fingerprint_for_origin + + def count_lowering() -> str: + nonlocal lower_calls + lower_calls += 1 + return original_lowering() + + def count_parser(origin: str) -> str: + parser_calls.append(origin) + return original_parser(origin) + + monkeypatch.setattr(origin_specs_module, "lowering_fingerprint", count_lowering) + monkeypatch.setattr(origin_specs_module, "parser_fingerprint_for_origin", count_parser) + evidence = rebuild_index_module._rebuild_replay_closure_evidence(root, _raw_ids(root)) + assert evidence["raw_session_evidence"] + assert lower_calls == 1 + assert parser_calls == ["codex-session"] + + +def test_referenced_blob_snapshot_ignores_volatile_filesystem_metadata( + tmp_path: Path, +) -> None: + """Identical content remains valid when a blob's mtime changes. + + Anti-vacuity: restoring the old inode/mtime fields to the content snapshot + would make this metadata-only touch change the recorded evidence. + """ + root = tmp_path / "archive" + _seed(root, count=1) + with sqlite3.connect(root / "source.db") as source: + referenced_hashes = {bytes(row[0]).hex() for row in source.execute("SELECT blob_hash FROM raw_sessions")} + before = schema_gate_module._referenced_blob_integrity_snapshot(root, referenced_hashes=referenced_hashes) + blob_path = BlobStore(root / "blob").blob_path(next(iter(referenced_hashes))) + stat = blob_path.stat() + os.utime(blob_path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1)) + after = schema_gate_module._referenced_blob_integrity_snapshot(root, referenced_hashes=referenced_hashes) + assert all( + "inode" not in entry and "mtime_ns" not in entry for entry in cast(list[dict[str, object]], before["entries"]) + ) + assert after == before + + +def test_source_evidence_snapshot_streams_raw_session_rows(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "archive" + _seed(root, count=2) + sqlite_module = cast(Any, index_generation_module).sqlite3 + real_connect = sqlite_module.connect + + class GuardedCursor: + def __init__(self, cursor: sqlite3.Cursor, sql: str) -> None: + self._cursor = cursor + self._sql = sql + + def __iter__(self) -> GuardedCursor: + return self + + def __next__(self) -> tuple[object, ...]: + return next(self._cursor) + + def fetchall(self) -> list[tuple[object, ...]]: + if "FROM RAW_SESSIONS" in self._sql.upper(): + raise AssertionError("raw_sessions evidence must be consumed as a stream") + return self._cursor.fetchall() + + def __getattr__(self, name: str) -> object: + return getattr(self._cursor, name) + + class GuardedConnection: + def __init__(self, connection: sqlite3.Connection) -> None: + self._connection = connection + + def execute(self, sql: str, parameters: object = ()) -> GuardedCursor: + return GuardedCursor(self._connection.execute(sql, cast(Any, parameters)), sql) + + def __getattr__(self, name: str) -> object: + return getattr(self._connection, name) + + def guarded_connect(*args: object, **kwargs: object) -> GuardedConnection: + return GuardedConnection(real_connect(*args, **kwargs)) + + monkeypatch.setattr(sqlite_module, "connect", guarded_connect) + assert index_generation_module.rebuild_source_evidence_snapshot(root) diff --git a/tests/unit/maintenance/test_rebuild_index_resume_correctness.py b/tests/unit/maintenance/test_rebuild_index_resume_correctness.py index 897b4813da..2bec4fa1fd 100644 --- a/tests/unit/maintenance/test_rebuild_index_resume_correctness.py +++ b/tests/unit/maintenance/test_rebuild_index_resume_correctness.py @@ -22,6 +22,7 @@ from polylogue.storage.index_generation import IndexGenerationStore from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from tests.infra.rebuild_receipt import write_valid_rebuild_receipt class InjectedInterruptError(RuntimeError): @@ -57,7 +58,7 @@ def _payload(native_id: str, text: str, *, parent_native_id: str | None = None) return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) -def _seed(root: Path) -> None: +def _seed(root: Path, *, monkeypatch: pytest.MonkeyPatch) -> None: initialize_active_archive_root(root) with ArchiveStore.open_existing(root, read_only=False) as archive: for index, native_id, parent_native_id in ( @@ -71,6 +72,25 @@ def _seed(root: Path) -> None: source_path=f"resume/{index}.jsonl", acquired_at_ms=index + 1, ) + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + UPDATE raw_sessions + SET logical_source_key = CASE + WHEN source_path LIKE '%/0.jsonl' THEN 'codex:resume-parent' + WHEN source_path LIKE '%/1.jsonl' THEN 'codex:resume-child' + ELSE 'codex:resume-standalone' + END, + revision_kind = 'full', + source_revision = raw_id, + baseline_raw_id = raw_id, + acquisition_generation = 0, + revision_authority = 'byte_proven' + """ + ) + source.commit() + receipt_path = write_valid_rebuild_receipt(root, root.parent / f"{root.name}-schema-receipt.json") + monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) def _semantic_snapshot(root: Path) -> tuple[object, ...]: @@ -101,7 +121,7 @@ def test_committed_page_interrupt_resumes_only_suffix_and_matches_clean_rebuild( root = tmp_path / "resumed" clean_root = tmp_path / "clean" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) - _seed(root) + _seed(root, monkeypatch=monkeypatch) original_checkpoint = IndexGenerationStore.checkpoint_transaction interrupted = False @@ -155,7 +175,7 @@ async def recording_replay(*args: object, **kwargs: object) -> dict[str, object] assert receipt.operation["recovery_state"] == "promoted" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(clean_root)) - _seed(clean_root) + _seed(clean_root, monkeypatch=monkeypatch) clean = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=clean_root, raw_batch_size=1)) assert clean.status == "paused" assert clean.transaction is not None diff --git a/tests/unit/maintenance/test_schema_inference_gate.py b/tests/unit/maintenance/test_schema_inference_gate.py index d18edda3d0..1ac1d4d1eb 100644 --- a/tests/unit/maintenance/test_schema_inference_gate.py +++ b/tests/unit/maintenance/test_schema_inference_gate.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +import json import sqlite3 from datetime import UTC, datetime, timedelta from pathlib import Path @@ -133,6 +134,25 @@ def test_clean_archive_runs_actual_blobstore_verifier_and_external_reconciliatio } == before +def test_readiness_accepts_equivalent_blob_evidence_from_another_verifier(tmp_path: Path) -> None: + root = tmp_path / "archive" + ground_truth = _seed_archive(root) + receipt_path = tmp_path / "receipt" / RECEIPT_FILENAME + run_schema_inference_gate(root, receipt_path=receipt_path, ground_truth_roots={"codex-session": (ground_truth,)}) + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + recorded = dict(payload["full_blob_hash_verification"]["referenced_blob_integrity_snapshot"]) + recorded["verifier"] = "polylogue.storage.blob_store.BlobStore.verify" + + validated = gate.validate_schema_inference_receipt( + root, + receipt_path, + verify_blob_integrity=True, + verified_blob_integrity_snapshot=recorded, + ) + + assert validated["_verified_blob_integrity_snapshot"] == recorded + + def test_empty_archive_cannot_authorize_schema_inference(tmp_path: Path) -> None: root = tmp_path / "archive" initialize_active_archive_root(root) @@ -500,3 +520,92 @@ def test_gate_receipt_digest_is_canonical_and_content_bound() -> None: altered = dict(first) altered["verdict"] = "FAIL" assert schema_inference_gate_receipt_digest(first) != schema_inference_gate_receipt_digest(altered) + + +def test_external_inventory_token_reuses_metadata_detector_and_rejects_changed_bytes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The real receipt validator hashes once, then detects corpus drift cheaply. + + Anti-vacuity: removing the pass token binding makes the second validation + hash the full corpus again, so the call-count assertions fail. + """ + root = tmp_path / "archive" + ground_truth = _seed_archive(root) + receipt_path = tmp_path / "receipt" / RECEIPT_FILENAME + run_schema_inference_gate(root, receipt_path=receipt_path, ground_truth_roots={"codex-session": (ground_truth,)}) + + full_inventory_calls = 0 + original_inventory = gate._external_inventory + + def counted_inventory(roots: object) -> object: + nonlocal full_inventory_calls + full_inventory_calls += 1 + return original_inventory(roots) # type: ignore[arg-type] + + monkeypatch.setattr(gate, "_external_inventory", counted_inventory) + first = gate.validate_schema_inference_receipt(root, receipt_path) + token = cast(dict[str, object], first["external_ground_truth_inventory_token"]) + assert full_inventory_calls == 1, "the first validation establishes the authoritative full inventory" + gate.validate_schema_inference_receipt(root, receipt_path, inventory_token=token) + assert full_inventory_calls == 1, "unchanged pass validation must reuse the bound detector token" + + (ground_truth / "session.jsonl").write_bytes(b"changed external codex raw") + with pytest.raises(SchemaInferenceGateError, match="external ground-truth corpus changed"): + gate.validate_schema_inference_receipt(root, receipt_path, inventory_token=token) + assert full_inventory_calls == 2, "a changed detector must trigger one authoritative inventory recalculation" + + +def test_inventory_change_detector_triggers_rehash_without_changing_content_identity( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive" + ground_truth = _seed_archive(root) + receipt_path = tmp_path / "receipt" / RECEIPT_FILENAME + run_schema_inference_gate(root, receipt_path=receipt_path, ground_truth_roots={"codex-session": (ground_truth,)}) + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + origins = payload["ground_truth_inputs"]["origins"] + baseline = gate._canonical_external_ground_truth_digest(origins) + altered = json.loads(json.dumps(origins)) + altered["codex-session"]["inventory_change_detector"] = {"forced": "changed"} + assert gate._canonical_external_ground_truth_digest(altered) == baseline + + validated = gate.validate_schema_inference_receipt(root, receipt_path) + token = cast(dict[str, object], validated["external_ground_truth_inventory_token"]) + (ground_truth / "session.jsonl").touch() + inventory_calls = 0 + original_inventory = gate._external_inventory + + def counted_inventory(roots: tuple[Path, ...]) -> list[object]: + nonlocal inventory_calls + inventory_calls += 1 + return original_inventory(roots) # type: ignore[return-value] + + monkeypatch.setattr(gate, "_external_inventory", counted_inventory) + refreshed = gate.validate_schema_inference_receipt(root, receipt_path, inventory_token=token) + assert inventory_calls == 1 + refreshed_token = cast(dict[str, object], refreshed["external_ground_truth_inventory_token"]) + original_origin_token = cast(dict[str, object], token["origins"])["codex-session"] + refreshed_origin_token = cast(dict[str, object], refreshed_token["origins"])["codex-session"] + assert ( + cast(dict[str, object], refreshed_origin_token)["inventory_change_detector"] + != cast(dict[str, object], original_origin_token)["inventory_change_detector"] + ) + gate.validate_schema_inference_receipt(root, receipt_path, inventory_token=refreshed_token) + assert inventory_calls == 1, "a successful rehash must refresh the detector token" + + +def test_inventory_token_rejects_foreign_nonce_and_empty_token_falls_back_to_receipt( + tmp_path: Path, +) -> None: + root = tmp_path / "archive" + ground_truth = _seed_archive(root) + receipt_path = tmp_path / "receipt" / RECEIPT_FILENAME + run_schema_inference_gate(root, receipt_path=receipt_path, ground_truth_roots={"codex-session": (ground_truth,)}) + validated = gate.validate_schema_inference_receipt(root, receipt_path) + token = cast(dict[str, object], validated["external_ground_truth_inventory_token"]) + foreign = json.loads(json.dumps(token)) + foreign["receipt_nonce"] = "foreign-pass" + with pytest.raises(SchemaInferenceGateError, match="token is not bound"): + gate.validate_schema_inference_receipt(root, receipt_path, inventory_token=foreign) + gate.validate_schema_inference_receipt(root, receipt_path, inventory_token={})