diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index fc6fdd0fa2..9d2b17fc48 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -14,7 +14,7 @@ {"_type":"issue","id":"polylogue-cursor-authority-reconcile-implementation","title":"maintenance: add scoped cursor-authority reconciliation through normal ingest","description":"Implement the one missing maintenance surface needed to repair the known cursor-authority violation without disabling the global fail-closed gate. The existing fail-closed mechanism and census are recorded by polylogue-xeck9 and PR #3823; the remaining implementation is a scoped, backup-gated, single-path reconciliation command.","design":"Add `polylogue ops maintenance cursor-authority-reconcile` with dry-run default and explicit `--apply`. Dry-run takes `--source-path-file` (0600 file containing one private absolute path) and `--output-plan`; resolve `/realm/db/polylogue` explicitly, require the daemon stopped, inspect source/index/ops/audit schemas and versions, run raw_frontier_integrity_projection, require exactly one true cursor-ahead relation for the selected path, verify byte-authoritative accepted head, hash current source bytes through accepted_frontier with two stat observations, and emit self-hashed polylogue.cursor-authority-reconciliation-plan.v1. Apply takes `--plan`, `--backup-manifest`, `--receipt`, and `--apply`; require stopped daemon and verified full-evidence backup, acquire writer lease, revalidate every binding, create a one-use authorization carrying path digest, cursor offset, accepted frontier, and plan digest, invoke only the existing full-ingest/replay route, allow the gate bypass only for the exact planned violation, never directly update ingest_cursor or accepted heads, enforce reconciled or typed_deferred postconditions, reject worsening unrelated rows, quick_check touched tiers, and emit polylogue.cursor-authority-reconciliation-receipt.v1. Add the exact named tests in the acceptance criteria. Do not change frontier comparison semantics or typed reason codes.","acceptance_criteria":"1. Dry-run is deterministic and changes no SQLite pages, cursor rows, source rows, or files except the requested plan. 2. Apply refuses without a verified full-evidence backup or while daemon ownership is active. 3. The exact planned ahead path uses the existing full-ingest/replay route. 4. Normal unscoped ingestion remains blocked. 5. One-use authorization cannot be reused for another path or after cursor, head, source-prefix, schema, database, or code SHA changes. 6. Source mutation during hashing refuses. 7. Zero ahead rows produces typed not_applicable with no mutation. 8. More than one true ahead row refuses without guessing. 9. Existing 725/2 incomparable classes stay explicitly typed and do not become falsely healthy. 10. No direct cursor reset, accepted-head rewrite, or global force switch exists. 11. Crash before commit leaves pre-state unchanged; crash after ingest commit before receipt is recoverable and cannot run twice. 12. Removing path restriction or replacing scoped authorization with global bypass fails a controlled mutation test. 13. Exact-frontier ordinary ingestion remains allowed. 14. Tests exist in tests/unit/maintenance/test_cursor_authority_reconcile.py, tests/unit/sources/test_live_watcher.py, tests/unit/storage/test_raw_retention.py, and tests/unit/cli/test_archive_maintenance_cli.py. 15. devtools verify --quick passes. 16. PR scope is implementation-complete and live-proof-pending; no production receipt or Beads mutation is delivered by the worker.","notes":"Compiled Luna execution packet 2026-08-06 from the settled #3823 authority design. Required base commit: 5ed7a50a12b5ae6fe7dad00213b4a1ac160860b3. Allowed files: polylogue/maintenance/cursor_authority_reconcile.py; polylogue/cli/commands/maintenance/_cursor_authority.py; polylogue/cli/commands/maintenance/__init__.py; polylogue/sources/live/batch.py; tests/unit/maintenance/test_cursor_authority_reconcile.py; tests/unit/sources/test_live_watcher.py; tests/unit/storage/test_raw_retention.py; tests/unit/cli/test_archive_maintenance_cli.py; docs/devtools.md. Forbidden: production access, direct SQL cursor/head/source-row writes, global bypass, generic force flag, raw-frontier semantic changes, typed reason-code changes, Beads writes, subagents. Stop if base commit or an existing one-path normal-ingest API is absent, a required change leaves allowed files, or any ambiguity remains. Verification commands are the four named focused devtools tests plus devtools verify --quick.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:51:32Z","created_by":"Sinity","updated_at":"2026-08-06T11:51:32Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-cursor-authority-reconcile-implementation","depends_on_id":"polylogue-xeck9","type":"blocks","created_at":"2026-08-06T13:54:26Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":1,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-reindex-promotion-restart","title":"reindex: promote the accepted candidate and prove restart","description":"Phase 5 of the production reindex. Promote only the candidate named by a valid acceptance receipt, retain the rollback generation, restart the real daemon, and emit the ownership and pointer-transition receipt consumed by live-operation receipts and the terminal proof.","design":"Acquire the ordinary writer lease, revalidate source, candidate, semantics, acceptance receipt, and rollback identities, then perform one atomic pointer transition. Restart the daemon using the deployed package selected during preflight. Do not treat restart health or postflight convergence as already proven; downstream live-operation and final-proof beads consume this phase receipt and add those observations.","acceptance_criteria":"1. Candidate acceptance receipt is valid and fresh. 2. Pointer flip is atomic and binds old/new generation identities. 3. Previous generation remains available for rollback and its retention is recorded. 4. Real daemon restart receipt binds package, pointer, process, and initial health state. 5. No postflight receipt is synthesized; downstream proofs remain open until independently executed.","notes":"Compiled packet mapping 2026-08-06: this current phase name is authoritative for the stale packet alias reindex-promote-restart. It starts only after candidate acceptance and emits pointer, rollback-retention, ownership, and restart receipts; terminal postflight remains downstream.\nCompiled packet mapping 2026-08-06: this current phase name is authoritative for the stale packet alias reindex-promote-restart. It starts only after candidate acceptance and emits pointer, rollback-retention, ownership, and restart receipts; terminal postflight remains downstream.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:50:18Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:38Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-reindex-promotion-restart","depends_on_id":"polylogue-embeddings-retention","type":"blocks","created_at":"2026-08-06T13:53:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-promotion-restart","depends_on_id":"polylogue-reindex-candidate-acceptance","type":"blocks","created_at":"2026-08-06T13:53:21Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":2,"dependent_count":5,"comment_count":0} {"_type":"issue","id":"polylogue-reindex-candidate-acceptance","title":"reindex: accept an inactive candidate against all proof gates","description":"Phase 4 of the production reindex. After the inactive candidate is built by polylogue-818fy, run the candidate-targeted canary, incident ledger, registry, corpus fidelity, complexity, origin capability, and promotion-guard checks. This phase owns candidate acceptance, not candidate construction or promotion.","design":"Every check must target the exact inactive candidate generation and frozen source snapshot named by the phase receipts. Consume the dynamic forcing-set equality, canonical snapshot comparator, incident coverage ledger, inferred-origin matrix, and dlfcx candidate-fidelity guard. A missing receipt, stale semantics fingerprint, candidate/active target confusion, unexplained difference, or incomplete forcing row blocks acceptance. Emit one self-hashed candidate acceptance receipt.","acceptance_criteria":"1. polylogue-818fy and polylogue-0x7nh are prerequisites. 2. dlfcx is a direct prerequisite and cannot remain disconnected. 3. All registry checks, incident rows, red twins, corpus members, complexity laws, and expected deltas target the same candidate. 4. Candidate acceptance produces an immutable receipt consumed by promotion/restart. 5. No pointer flip or daemon restart occurs here.","notes":"Compiled packet mapping 2026-08-06: candidate acceptance consumes the shared live-proof protocol, canonical snapshot, dynamic incident forcing-set equality, origin matrix, inferred corpus, fidelity, complexity, and no-promote canary receipts. It never flips the active pointer or restarts the daemon.\nCompiled packet mapping 2026-08-06: candidate acceptance consumes the shared live-proof protocol, canonical snapshot, dynamic incident forcing-set equality, origin matrix, inferred corpus, fidelity, complexity, and no-promote canary receipts. It never flips the active pointer or restarts the daemon.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:50:14Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:34Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-0v4tn","type":"blocks","created_at":"2026-08-06T17:28:19Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-0x7nh","type":"blocks","created_at":"2026-08-06T13:53:13Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-818fy","type":"blocks","created_at":"2026-08-06T13:53:09Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-canonical-snapshot","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-csx21","type":"blocks","created_at":"2026-08-06T15:51:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-cw8l0","type":"blocks","created_at":"2026-08-06T17:00:37Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-dcrmm","type":"blocks","created_at":"2026-08-06T17:27:49Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-dlfcx","type":"blocks","created_at":"2026-08-06T13:53:17Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-f1vg","type":"blocks","created_at":"2026-08-06T15:51:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-incident-coverage-ledger","type":"blocks","created_at":"2026-08-06T14:07:12Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-kmt1c","type":"blocks","created_at":"2026-08-06T17:13:55Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-mn0si","type":"blocks","created_at":"2026-08-06T17:34:18Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-o5smo","type":"blocks","created_at":"2026-08-06T17:14:03Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-origin-capability-matrix","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-q4qpl","type":"blocks","created_at":"2026-08-06T17:27:45Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-reindex-registry-two-plane-subset","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-reindex-source-remediation","type":"blocks","created_at":"2026-08-06T14:07:08Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-rrxe4","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-rrxe4.1","type":"blocks","created_at":"2026-08-06T15:51:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-tiozw","type":"blocks","created_at":"2026-08-06T17:14:06Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-candidate-acceptance","depends_on_id":"polylogue-x97cf","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":21,"dependent_count":1,"comment_count":0} -{"_type":"issue","id":"polylogue-reindex-source-remediation","title":"reindex: freeze, deploy, migrate, and remediate source state","description":"Phase 2 of the production reindex. Starting from an accepted preflight authorization, perform the freeze/deploy/migrate boundary and the backup-gated source remediation required to make the durable source archive eligible for candidate construction. This carrier owns live application receipts split from raw-failure, cursor-authority, and redeploy residuals; it does not build or promote an index candidate.","design":"Require a fresh verified full-evidence backup, stopped-daemon ownership, the selected deployed package, durable schema currency, and immutable dry-run plans before any apply. Execute only named scoped maintenance routes, including the cursor-authority reconciliation route once its implementation exists. Recompute every plan binding after ownership acquisition, run quick_check on touched tiers, and emit immutable receipts consumed by candidate build and postflight proof. Never repair by direct cursor/head/source-row SQL edits.","acceptance_criteria":"1. The deployed runtime and durable tiers match the preflight contract. 2. Fresh backup and migration receipts bind exact source, index, ops, blob, and package identities. 3. dyica, xeck9, and msia source-side residuals have typed dry-run/apply outcomes or remain explicit blockers. 4. No candidate generation is built or promoted here. 5. The phase exposes exact source snapshot and receipts for the inactive candidate build.","notes":"Compiled packet mapping 2026-08-06: this current phase carrier subsumes the stale packet name reindex-freeze-deploy-migrate. It is phase 2 after preflight and before candidate construction. All live writes remain operator-authorized, dry-run-first, backup-gated, and receipt-bound.\nCompiled packet mapping 2026-08-06: this current phase carrier subsumes the stale packet name reindex-freeze-deploy-migrate. It is phase 2 after preflight and before candidate construction. All live writes remain operator-authorized, dry-run-first, backup-gated, and receipt-bound.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:50:11Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:30Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-a7gmk","type":"blocks","created_at":"2026-08-06T13:52:51Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-cursor-authority-reconcile-implementation","type":"blocks","created_at":"2026-08-06T13:54:06Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-dyica","type":"blocks","created_at":"2026-08-06T13:52:55Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-reindex-preflight-authorization","type":"blocks","created_at":"2026-08-06T13:52:47Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-s8s54","type":"blocks","created_at":"2026-08-06T18:53:04Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-uecir","type":"blocks","created_at":"2026-08-06T17:35:00Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-xeck9","type":"blocks","created_at":"2026-08-06T13:52:58Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":7,"dependent_count":5,"comment_count":0} +{"_type":"issue","id":"polylogue-reindex-source-remediation","title":"reindex: freeze, deploy, migrate, and remediate source state","description":"Phase 2 of the production reindex. Starting from an accepted preflight authorization, perform the freeze/deploy/migrate boundary and the backup-gated source remediation required to make the durable source archive eligible for candidate construction. This carrier owns live application receipts split from raw-failure, cursor-authority, and redeploy residuals; it does not build or promote an index candidate.","design":"Require a fresh verified full-evidence backup, stopped-daemon ownership, the selected deployed package, durable schema currency, and immutable dry-run plans before any apply. Execute only named scoped maintenance routes, including the cursor-authority reconciliation route once its implementation exists. Recompute every plan binding after ownership acquisition, run quick_check on touched tiers, and emit immutable receipts consumed by candidate build and postflight proof. Never repair by direct cursor/head/source-row SQL edits.","acceptance_criteria":"1. The deployed runtime and durable tiers match the preflight contract. 2. Fresh backup and migration receipts bind exact source, index, ops, blob, and package identities. 3. dyica, xeck9, and msia source-side residuals have typed dry-run/apply outcomes or remain explicit blockers. 4. No candidate generation is built or promoted here. 5. The phase exposes exact source snapshot and receipts for the inactive candidate build.","notes":"Compiled packet mapping 2026-08-06: this current phase carrier subsumes the stale packet name reindex-freeze-deploy-migrate. It is phase 2 after preflight and before candidate construction. All live writes remain operator-authorized, dry-run-first, backup-gated, and receipt-bound.\nCompiled packet mapping 2026-08-06: this current phase carrier subsumes the stale packet name reindex-freeze-deploy-migrate. It is phase 2 after preflight and before candidate construction. All live writes remain operator-authorized, dry-run-first, backup-gated, and receipt-bound.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:50:11Z","created_by":"Sinity","updated_at":"2026-08-06T13:53:30Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-6k0na","type":"blocks","created_at":"2026-08-07T01:49:25Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-a7gmk","type":"blocks","created_at":"2026-08-06T13:52:51Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-cursor-authority-reconcile-implementation","type":"blocks","created_at":"2026-08-06T13:54:06Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-dyica","type":"blocks","created_at":"2026-08-06T13:52:55Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-reindex-preflight-authorization","type":"blocks","created_at":"2026-08-06T13:52:47Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-s8s54","type":"blocks","created_at":"2026-08-06T18:53:04Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-uecir","type":"blocks","created_at":"2026-08-06T17:35:00Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-source-remediation","depends_on_id":"polylogue-xeck9","type":"blocks","created_at":"2026-08-06T13:52:58Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":8,"dependent_count":5,"comment_count":0} {"_type":"issue","id":"polylogue-reindex-preflight-authorization","title":"reindex: authorize a frozen production transition","description":"Phase 1 of the production reindex. Establish one immutable authorization binding the selected code and deployed package, source snapshot, raw-authority census, schema fingerprints, explicit waivers, and the exact preflight receipt before any freeze, migration, candidate build, or live mutation. The current graph has postconditions blocking the operation that produces them; this phase carrier provides the ordered precondition.","design":"The coordinator records a structured preflight contract and receipt. It consumes the structured PR scope contract and the current incident-forcing set, and it does not perform database mutation. It must reject stale code, stale Beads scope, unresolved disconnected P0 delivery defects, untyped raw failures, or a source/index authority census without an explicit typed disposition. Runtime values are produced only by named commands and stored in the receipt consumed by the next phase.","acceptance_criteria":"1. Exact code SHA, deployed package SHA, Beads digest, source/index/ops fingerprints, schema versions, raw-authority census, and waiver set are bound in a self-hashed preflight receipt. 2. polylogue-pr-scope-contract is a direct prerequisite. 3. The receipt is immutable and names the next source-remediation and candidate-build phase inputs. 4. No production write or promotion occurs in this phase. 5. A stale binding or disconnected P0 defect fails closed.","status":"open","priority":0,"issue_type":"epic","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T11:50:07Z","created_by":"Sinity","updated_at":"2026-08-06T11:50:07Z","labels":["area:maintenance","lane:reindex"],"dependencies":[{"issue_id":"polylogue-reindex-preflight-authorization","depends_on_id":"polylogue-incident-coverage-ledger","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-preflight-authorization","depends_on_id":"polylogue-pr-scope-contract","type":"blocks","created_at":"2026-08-06T13:52:44Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-reindex-preflight-authorization","depends_on_id":"polylogue-reindex-proof-edge-correction","type":"blocks","created_at":"2026-08-06T15:51:23Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":3,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-dudtn","title":"Implement durable schema-currency gate hardening","description":"Implementation slice extracted from polylogue-9qnzy after review found the original gate incomplete. This Bead owns the code and regression tests; polylogue-9qnzy remains the live deployment and migration operation.\n\nProblem: rebuild-index could bypass durable schema currency through audit drift, ownership races, daemon bulk setup, the empty-source CLI path, and an unstructured daemon error.\n\nScope: guard every canonical durable migration tier, recheck after ownership acquisition, guard daemon bulk bookkeeping, remove empty-source bypass, reject daemon preflight, preserve a structured conflict diagnostic, and document migration of audit.db.\n\nNo production mutation is performed by this implementation Bead.","design":"Use one canonical durable-tier set from the migration runner. Keep derived index mismatch outside this gate because rebuild-index owns replacement of the derived tier. Bind the implementation to a PR scope carrier and leave live operation as a separate receipt.","acceptance_criteria":"1. The predicate derives from DURABLE_MIGRATION_TIERS and covers source, user, and audit.\n2. Local rebuild checks before receipt/ownership/candidate creation and again after ownership acquisition.\n3. Daemon bulk entry checks before transaction bookkeeping.\n4. CLI empty-source execution and --preflight --daemon cannot bypass the guard.\n5. Daemon HTTP returns a structured 409 rebuild-schema-currency diagnostic.\n6. File-backed regression tests exercise every guard and the quick gate passes.\n7. Live migrations and deployment remain open under polylogue-9qnzy and polylogue-a7gmk.","notes":"Created 2026-08-06 to provide truthful implementation authority for PR #3856 after Terra/Sol review of polylogue-9qnzy noted that its AC5 explicitly excluded code changes.\nPublication carrier corrected 2026-08-06: PR #3857 now binds the full head c9d12abcf60941e9925ae22c475dc9a4230b3a8d. Implementation evidence remains PR #3856 at 1ac4749772bb9207c356ab9a32e6fa14c9db194a; live migration remains polylogue-9qnzy/a7gmk.\nCarrier publication commit is now a2ec4d8fbff5b19bed843df15f269f58b6881580; the PR body carrier is regenerated against that exact head and the committed branch snapshot.\nCI publication sequencing correction 2026-08-06: prepare the next exact-head carrier before pushing its commit, because Circle validates the PR body at push-trigger time.\nFinal carrier sequencing receipt 2026-08-06: the next PR head will be pushed only after its exact carrier has been published in the PR body; this prevents a stale push-triggered validation.\nAudit correction 2026-08-06: the merged tracker commit for PR #3857 had serialized literal backslash-n sequences in this Bead description and acceptance fields. Replaced them with actual paragraph/newline structure. The PR also carried unrelated tracker mutations for polylogue-z7sv3 and polylogue-csx21; those graph states are retained only where independently justified, while the attribution/process defect is tracked under polylogue-pr-scope-contract.","status":"closed","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T09:50:03Z","created_by":"Sinity","updated_at":"2026-08-06T19:24:35Z","closed_at":"2026-08-06T19:24:35Z","close_reason":"Closed after current-master audit: the durable schema-currency implementation and regression coverage landed in PR #3858. Live deploy and migration receipts remain under polylogue-9qnzy and polylogue-a7gmk.","dependency_count":0,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-reindex-proof-edge-correction","title":"Correct blocking edges for reindex live proof graph","description":"The proof graph merged in PR 3836 attached live-proof children to historical implementation Beads only with parent-child membership edges. The twelve live-proof children must also have explicit blocking dependencies on their implementation mechanism or acceptance owner, and the terminal reindex proof must consume the corrected graph.","design":"Beads rejects a blocks edge from a child to its ancestor, so the twelve live-proof records are standalone acceptance nodes rather than children of the historical implementation records. Each has an explicit blocks edge to the implementation owner. The edge-correction item itself blocks polylogue-reindex-final-proof. This preserves hard ordering without introducing an impossible ancestor dependency.","acceptance_criteria":"1. All twelve live-operation proof children have explicit blocks edges to their implementation or acceptance owner where the audit identified one. 2. The terminal reindex proof depends on the edge-correction item and therefore cannot be ready while those proof obligations are unbound. 3. Beads graph validation reports no cycles or dangling dependencies. 4. A graph fixture test or executable policy check fails when one required blocking edge is removed. 5. The change is delivered as one batched Beads export commit with the exact edge matrix in the PR body.","notes":"Graph constraint correction 2026-08-06: Beads rejects a blocks edge from a child to its ancestor. The twelve live-proof records were therefore detached from historical implementation parents and retain explicit blocks edges to those implementation records. This preserves hard ordering without creating an impossible ancestor dependency; the exact matrix is in the PR.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T05:35:24Z","created_by":"Sinity","updated_at":"2026-08-06T05:48:17Z","dependency_count":0,"dependent_count":2,"comment_count":0} @@ -1743,3 +1743,4 @@ {"_type":"issue","id":"polylogue-fs1.4","title":"Report: polylogue forensics for Hermes sessions","description":"Five-section per-session/per-corpus report, computed from the canonical archive (composition over existing primitives where possible): 1) session topology — parents, resumes, compactions, subagents, branches, long turns; 2) LLM/request economy — token lanes, cost, retry/fallback causes, model/provider shifts, cache-read amplification; 3) tool execution profile — durations, failures, approvals, repeated calls, parallel groups; 4) failure patterns — loops, stalls, empty-response retries, repeated shell failures, truncation, compaction-induced loss, reasoning burn; 5) local causal footprint — git diff/commits, commands, files, build/test runs. The 2-minute demo artifact (sanitized sessions, one command, README section) is the campaign-grade packaging of this report.","design":"Composition first: sections 1-3 and most of 4 should lower onto existing primitives — get_session_topology/logical session (topology), session_provider_usage_events + cost rollups (economy), actions/tool timing (tool profile), pathology detectors + structural outcomes (failure patterns), session_commits/git correlation (footprint). Only add new detectors where Hermes-specific (loop detection over repeated identical tool calls; stall = long gap between spans; reasoning burn = reasoning-token share per turn). Surface: a named read view/report profile (`polylogue forensics hermes --session \u003cid\u003e` or read --view forensics), rendered markdown + JSON. Demo packaging: sanitized fixture sessions, one command, \u003c2min, README section — that packaging is a legitimate one-off; the five sections' facts must be query-composable (capabilities-not-silos rule).","acceptance_criteria":"A Hermes forensic report regenerates from imported Hermes sessions and emits citable findings with coverage/fidelity caveats, raw evidence refs, and a single documented regeneration command. The report includes at least one happy-path fixture, one missing-field/degraded fixture, and one fidelity limitation that renders visibly instead of silently disappearing.","notes":"Executable upgrade (2026-07-04 sidecar):\nClassification: blocked on polylogue-fs1.1 for real Hermes state.db ingestion, but the report contract can be made executable now against synthetic/fixture sessions and later rerun on real Hermes rows.\nProduct question: can Polylogue produce a cold-reader forensic report that explains one Hermes/agent session better than the runtime itself, using canonical archive facts rather than a silo export?\nLikely modules/surfaces: read/query surfaces under polylogue/cli/read or command inventory, session topology/logical-session APIs, session_provider_usage_events/cost rollups, actions/tool timing readers, git/local footprint helpers, demo fixture/scenario generators, docs/demo shelf. Prefer a named report/read view that composes existing primitives; avoid a Hermes-only data path.\nArtifact shape: markdown + JSON report with the five existing sections, each section carrying source_refs/query names and missing-data caveats. Demo package must include sanitized fixture/session id, one command, expected runtime under 2 minutes, and a README snippet.\nAcceptance detail: for fixture data, each section has at least one asserted fact and one source reference; for missing Hermes fields, the report emits explicit unavailable/caveat rows rather than prose guesses; once fs1.1 lands, rerun against a real Hermes session and record diff between fixture and real coverage.\nVerification commands: focused unit/visual/demo command for the report surface, plus devtools render all --check if docs/README/demo surfaces are touched. If the implementation adds a new CLI command or view, verify command inventory and generated docs through devtools render all.\n[Delivery upgrade 2026-07-07T00:05:00Z] Release=K-interop-origin-export; lane=origin-interop-export; readiness=A-implementation-ready; proof=OriginSpec detector/parser/fixture/fidelity suite and content-hash export/import roundtrip. Original readiness=C-needs-acceptance-criteria.\n[Prework packet 2026-07-07] Static execution packet (anchors, mechanism, plan, tests, verification): .agent/handoffs/polylogue-gpt-pro-2026-07-07/prework-v2/task_packets/105_polylogue_fs1_4.md (depth: bead-localized-from-export; urgency: T2-foundation-before-feature-proof). Generated from master @ 8a975a40 2026-07-06 — verify source anchors before coding; line numbers are snapshot-relative.\n2026-07-10 integration-demo refinement: fs1.12 consumes this report. Include a claim-vs-tool-evidence canary with supported, contradicted, and unknown states; every displayed conclusion must resolve to structured tool outcome evidence and a fidelity caveat, not agent prose.\n2026-07-10 Nous follow-up technical refinement: use an evidence-status taxonomy that distinguishes supported, contradicted, later repaired/reversed, externally uncheckable, and unverifiable because required evidence was not retained. Preserve temporal ordering and evidence-retention caveats so later success cannot launder an earlier contradicted claim. Reuse the general claim-vs-evidence/claims-ledger substrate; do not create Hermes-only verdict semantics.\nDeferred (no new code) -- investigated and found the design's own preferred shape (\"composition first... avoid a Hermes-only data path... only add new detectors where Hermes-specific\") is already substantially satisfied by existing generic primitives, verified by reading their source (not assumed):\n\n- Section 1 (session topology): get_session_topology / logical session APIs -- already exist, origin-agnostic.\n- Sections 2-4 (LLM/request economy, tool execution profile, failure patterns): polylogue/insights/postmortem.py's compile_postmortem_bundle. Verified it is 100% origin-agnostic (SessionProfile.origin is a plain string field, zero origin-conditional branches in the aggregator) and already produces cost/token-lane metrics, tool-category profiles, and pathology-detector failure_mode/wasted_loop fields with evidence refs and honest degraded-not-fabricated behavior for missing signal (test_compile_postmortem_bundle_degrades_without_signal already covers this generically).\n- Section 5 (local causal footprint): polylogue/insights/session_commit.py's detect_session_commits -- git-commit attribution via time-window + file-overlap scoring, already exists, origin-agnostic.\n\nWhat's genuinely missing, and why I did not build it this pass: (a) a NAMED regeneration surface (a `polylogue forensics hermes --session \u003cid\u003e` command or `read --view forensics`) unifying these existing primitives under one command -- mechanical but real work that triggers the CLI-inventory/devtools-render-docs cascade; (b) the claim-vs-tool-evidence canary / evidence-status taxonomy the bead's own 2026-07-10 refinement explicitly says to build on \"the general claim-vs-evidence/claims-ledger substrate\" -- that substrate does not exist yet, so building Hermes-only verdict semantics here would violate the refinement's own instruction not to invent parallel machinery.\n\nDid not add a redundant \"hermes-flavored\" test of compile_postmortem_bundle: since the aggregator has zero origin-conditional logic, a test asserting it also works with origin=\"hermes-session\" would be vacuous (guaranteed to pass, proves nothing a mutation could break that the existing origin-agnostic tests don't already cover).\n\nRecommend: a follow-up scoped narrowly to (a) the CLI/read-view wiring only, composing the primitives above with zero new detector logic -- and treat the claim-vs-evidence canary as blocked on its own substrate bead, not this one.\n2026-07-18 (Claude Sonnet, branch feature/fix/hermes-atof-remaining-gaps): landed the verification-coverage correlation primitive that Phase 3's verification-ledger import (wj25) unblocked -- polylogue/insights/hermes_verification_coverage.py, a pure aggregator (no I/O) summarizing one Hermes session's verification_evidence.db coverage: structural event outcomes, final status, changed_paths, honest available=False (not fabricated) when no verification evidence exists. Also added hermes_verification.hermes_verification_session_id_for mirroring hermes_spans's existing observer-correlation helper. Verified via real archive ingestion (LiveBatchProcessor), not hand-built fixtures. 3/3 tests, devtools verify --quick green.\n\nDid NOT attempt in this pass, per this bead's own prior 2026-07-14 finding that sections 1-4 are 'substantially satisfied by existing generic primitives' and the recommended narrow follow-up is 'the CLI/read-view wiring only': the named CLI regeneration surface (read --view forensics or similar), the per-corpus aggregate ('sessions ended with failing/absent verification'), the 2-minute demo package (sanitized fixtures, one command, README section), and MCP tool wiring. This was a deliberate scoping decision given session budget, not an oversight -- the correlation primitive was the one piece genuinely blocked on Phase 3 landing first; the rest is composition/wiring work that deserves its own focused pass (CLI read-view registration triggers the docs-render cascade per this repo's own gotchas list).\n2026-07-18 merge: PR #3120 squash-merged to master as b563083188926b2078965ea41c2028a0a305577e. Verification-coverage correlation primitive (hermes_verification_coverage.py) is now live on master. Bead stays open: named CLI/read-view surface, per-corpus aggregate, 2-minute demo package, and MCP wiring remain undone per this pass's explicit scoping decision (see prior note).\n[Verification sweep 2026-07-31, bead-landing-check group5] Verdict: LIVE. Substantial composition-based delivery (PR #3120 verification-coverage correlation) but explicit remaining scope: named CLI/read-view surface, per-corpus aggregate, demo package, MCP wiring - \"deliberate scoping decision... not an oversight.\"\nUNBLOCKED 2026-07-31 (polylogue-pbuh/cijx.1 residual pass, worktree agent-aaffe89902b670d4b): the session-\u003ePR producer+reader chain this bead depends on is now real. session_refs carries typed pull_request evidence (18,949 rows live), and PR #3425 (merged 5525446a2) wired `read --view correlation` / Polylogue.session_correlation_payload to consume it as authoritative over the old regex/time-window heuristics, with disagreements surfaced rather than silently guessed. Verified live against /realm/db/polylogue/index.db (read-only) that the CLI path resolves real typed PR refs end-to-end (also fixed a pre-existing NameError in that path's GitHub-enrichment branch that had never been exercised with real refs before this pass). Full detail: polylogue-cijx.1 and polylogue-pbuh notes, 2026-07-31.\n\nNOT closed by this alone: this bead's own AC still needs its specific deliverable (see this bead's own description) beyond \"the correlation data is now readable\" -- that implementation work was not attempted in this pass (out of its declared scope: read-surface residual verification for pbuh/cijx.1 only). Re-triage this bead's own AC against the now-working session_commit.py/correlation_view.py surface when picked up next.\n","status":"open","priority":4,"issue_type":"feature","owner":"ezo.dev@gmail.com","created_at":"2026-07-03T04:31:41Z","created_by":"Sinity","updated_at":"2026-07-31T22:35:46Z","labels":["area:ingest","area:query","area:substrate","delivery:K-interop-origin-export","delivery:ac-patched","lane:origin-interop-export"],"dependencies":[{"issue_id":"polylogue-fs1.4","depends_on_id":"polylogue-cijx.1","type":"blocks","created_at":"2026-07-29T06:52:02Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-fs1.4","depends_on_id":"polylogue-fs1","type":"parent-child","created_at":"2026-07-03T06:31:40Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-fs1.4","depends_on_id":"polylogue-fs1.1","type":"blocks","created_at":"2026-07-03T06:31:41Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-fs1.4","depends_on_id":"polylogue-fs1.3","type":"blocks","created_at":"2026-07-10T11:03:54Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":3,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-fs1.2","title":"Importer: NeMo Relay ATOF/ATIF runtime spans","description":"Import Hermes observer-layer trace exports as runtime span evidence: pre/post_api_request -\u003e LLM request spans; pre/post_tool_call -\u003e tool execution spans with duration/status; approval hooks -\u003e high-risk decision points; subagent hooks -\u003e delegation graph; error hooks -\u003e retry/fallback taxonomy. ATIF import + enrichment beats inventing another trajectory format — respect Hermes's actual extension seams and make Polylogue the normalizer.","design":"VERIFY first: current NeMo Relay plugin output shape in the Hermes repo (ATOF JSONL / ATIF JSON exported from observer hooks). Ingest route: new artifact kinds in the taxonomy (archive/artifact_taxonomy/) + a spans parser under sources/parsers/, landing as ObservedEvents/actions attached to the session (join key: Hermes session id from the trace envelope -\u003e sessions.native_id). Map: pre/post_api_request pair -\u003e LLM request span (duration, model, provider, token fields if present); pre/post_tool_call -\u003e tool execution span with duration/status (structural outcome — feeds is_error/exit_code lanes where present); approval hooks -\u003e decision-point events; subagent lifecycle -\u003e topology_edges (subagent type); error hooks -\u003e retry/fallback taxonomy events. Spans without a matching archived session become explicit acquisition debt rows, not silent drops.","acceptance_criteria":"`polylogue-fs1.2` adds or updates an origin contract with detector, parser, raw fixture, normalized fixture, parser fingerprint, and fidelity/completeness notes. Ambiguous inputs are handled deterministically. The regression suite proves idempotent replay and visible degraded/missing-field behavior. Verification artifact: OriginSpec detector/parser/fixture/fidelity suite and content-hash export/import roundtrip.","notes":"[Delivery upgrade 2026-07-07T00:05:00Z] Release=K-interop-origin-export; lane=origin-interop-export; readiness=D-horizon-ready; proof=OriginSpec detector/parser/fixture/fidelity suite and content-hash export/import roundtrip. Original readiness=E-spec-needed.\n2026-07-10 Hermes contract refinement: ingest context_injected with profile/session/turn/snapshot-revision correlation; unpaired spans remain explicit acquisition debt. fs1.7 owns atomic spool/export production; this bead owns normalization and reconciliation.\nImplemented and PR opened (not merged): #2876 (feature/hermes/lifecycle-spool-and-bridge).\n\nScope understood: import Hermes observer-layer (NeMo Relay) trace exports as runtime span evidence, normalized and reconciled per the design's mapping (pre/post_api_request -\u003e LLM request spans, pre/post_tool_call -\u003e tool spans, approvals -\u003e decision points, subagent hooks -\u003e delegation evidence, error hooks -\u003e retry/fallback taxonomy).\n\nHonesty constraint documented explicitly in code + PR: the real ATOF/ATIF wire shape was not independently verifiable from this workspace -- no local checkout of the Hermes observer-plugin source was available. sources/parsers/hermes_spans.py implements a documented, testable, best-effort marker-based schema derived from this bead's own design notes and the shared lifecycle taxonomy (hermes_lifecycle.py, fs1.7). Every fidelity capability the parser declares tops out at \"inferred\", never \"exact\", for this reason -- filed as a concrete follow-up (fs1.2.1, not yet created as a bead by me -- flagging here so the orchestrator can file it) to re-verify against real Hermes source and tighten fidelity if it matches without changing the public contract.\n\nWhat changed: sources/parsers/hermes_spans.py (detector/parser/fidelity), wired into the real dispatch pipeline (sources/dispatch.py: detect_provider, lowering, parse_payload -- same path every other origin uses, not a bespoke test-only entrypoint); new artifact-taxonomy classification (archive/artifact_taxonomy/runtime.py).\n\nAC checklist: origin contract with detector/parser/raw fixture/normalized fixture/parser fingerprint/fidelity notes -- satisfied (marker_payload() is the raw-fixture generator used by every test; normalized output is the ParsedSession/session_events produced; fidelity via import_fidelity_declaration()). Ambiguous inputs handled deterministically -- satisfied: unrecognized hook_type -\u003e generic hermes_observer_span event (never dropped, never misclassified as a known kind); malformed span entries (missing hook_type/span_id, non-dict entries) are skipped and counted, not crashing. Idempotent replay -- satisfied and tested (test_atif_parse_is_idempotent_and_deterministic: same document parsed twice -\u003e byte-identical structural output). Visible degraded/missing-field behavior -- satisfied: unpaired spans (start without finish) are counted and surfaced as an explicit degraded fidelity capability with a caveat, never silently dropped.\n\nDesign gap explicitly NOT closed, documented not silently assumed: physical merge of observer spans into the state-db-ingested conversational session's message tree (the design's \"landing as ObservedEvents/actions attached to the session\"). This parser instead produces its own observer-evidence session (observer:\u003chermes_session_id\u003e) with a read-side correlation helper (hermes_observer_session_id_for) joining by the shared raw Hermes session id -- a physical content-tree merge across two independently-acquired artifacts is a session-identity/lineage design decision (topology_edges/session_links) I judged out of scope for this pass rather than improvising a schema-adjacent change.\n\nVerification: devtools test tests/unit/sources/parsers/test_hermes_spans.py -- 9/9 passed (subset of PR's 43-test combined run). devtools verify --quick exit 0.\n[gpt-5.6-terra integration refinement, 2026-07-14]\n\nReal producer evidence now exists: the bundled NousResearch Hermes observability/nemo_relay plugin emits ATIF v1.7 session documents and append-only ATOF JSONL through actual session, LLM, tool, approval, and subagent callbacks. ATIF import is live. The remaining producer-to-archive gap is ATOF materialization, not schema speculation.\n\nRefine this bead implementation order: retain byte-identified ATOF raw evidence first; incremental reader checkpoints file identity plus byte offset; tolerate partial final lines and rotation/truncation; validate/order/deduplicate events; materialize normalized lifecycle/action evidence idempotently; retain parent/child subagent links; surface unpaired/unmatched records as debt. Never synthesize ATIF from ATOF or duplicate transcript bodies into events. Update OriginSpec fidelity only where real exported fixtures prove a field mapping.","status":"closed","priority":4,"issue_type":"feature","owner":"ezo.dev@gmail.com","created_at":"2026-07-03T04:31:39Z","created_by":"Sinity","updated_at":"2026-07-31T22:35:43Z","closed_at":"2026-07-20T21:34:47Z","close_reason":"Complete in substance across the merged chain — every item of the 2026-07-14 refined implementation order shipped: byte-identified ATOF raw retention + incremental byte-offset reader with partial-line/rotation tolerance (pre-existing append-plan mechanism, verified fs1.2.1 notes); validate/order/dedup + idempotent lifecycle/action materialization (#3103); shared-file multi-session correctness (#3113/flxh); parent/child subagent links from producer-positive marks, fail-closed (#3231); unpaired/unmatched as explicit debt (#3103). Identity composed with profile+artifact-family qualification (#3224/#3225). OriginSpec detector/parser/fixture/fidelity satisfied against REAL producer fixtures with marker-only payloads as negative tests (#3231, fs1.2.1 closed). Force rationale: remaining blocker edge 2qx.1.1 (shared OriginSpec admission kernel/conformance law) is a lane-gate shared suite per the delivery-ac-template-interpretation adjudication (2026-07-07) — not a per-bead requirement; the Hermes origin will conform when that kernel lands, tracked there.","metadata":{"authored_by":"gpt-5.6-terra","authored_on":"2026-07-14"},"labels":["area:ingest","area:substrate","delivery:K-interop-origin-export","delivery:ac-patched","lane:origin-interop-export"],"dependencies":[{"issue_id":"polylogue-fs1.2","depends_on_id":"polylogue-2qx.1.1","type":"blocks","created_at":"2026-07-15T20:55:32Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-fs1.2","depends_on_id":"polylogue-fs1","type":"parent-child","created_at":"2026-07-03T06:31:39Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-fs1.2","depends_on_id":"polylogue-fs1.2.1","type":"blocks","created_at":"2026-07-14T11:39:19Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":2,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-n2dmn","title":"devtools: author execution contracts for graph missing-AC census","description":"The graph-policy verifier now exposes the complete missing-acceptance-criteria census, but the current graph still contains 220 open items without execution-grade acceptance criteria. Convert that census into durable, executable Beads scope. Preserve each item’s actual intent and existing dependency structure. Do not satisfy the policy by writing generic boilerplate, by closing work, or by treating a PR as evidence that its whole Bead is complete.","design":"Use the Terra graph-policy census as the authoritative input. Work in bounded clusters by subsystem and preserve the complete graph. The successor owns contract authoring and graph-state reconciliation; it does not weaken the verifier or close unrelated work.","acceptance_criteria":"1. Start from the current complete machine-readable census and retain a deterministic before/after manifest of every affected Bead ID. 2. Every affected open non-epic Bead has concrete acceptance criteria that name observable behavior, exact verification evidence, and any required live receipt, or has a structured policy-exemption record with owner, reason, and named successor. 3. Parent-child and blocking relationships remain valid and every successor is reachable from the owning campaign gate. 4. The graph-policy verifier reports zero unexplained missing-AC items and still reports all remaining exemptions or residual successors explicitly. 5. Add mutation coverage proving removal of one contract or successor makes the policy fail. 6. Run the focused devtools tests, graph policy, closure matrix, and quick verification on the final graph state.","notes":"Created as the named residual successor for the partial polylogue-8jg9.1 graph-policy implementation. Terra delivered execution-focus derivation, documentation guidance, parent-child integrity validation, and the complete 220-item census. This bead owns the remaining contract authoring and zero-unexplained-residual proof.","status":"open","priority":1,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T21:38:34Z","created_by":"Sinity","updated_at":"2026-08-06T21:38:34Z","labels":["area:beads","area:devtools","area:planning","horizon:frontier"],"dependencies":[{"issue_id":"polylogue-n2dmn","depends_on_id":"polylogue-8jg9.1","type":"discovered-from","created_at":"2026-08-06T23:38:34Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"polylogue-6k0na","title":"fix(storage): attest authorized source mutations across durable train continuity","description":"The durable source train continuity gate records exact content evidence at schema release. A backup-gated source liveness operation can legitimately mutate source.db after that release, but the next migration or daemon startup rejects the archive because source-028.json still carries the pre-mutation content hash. This blocks the production source-remediation phase after a valid apply and creates pressure to bypass the durable change-train gate. The fix must preserve fail-closed behavior while giving named source mutation routes a receipt-bound way to refresh continuity evidence.","design":"Add a typed, receipt-backed continuity refresh for authorized source-tier maintenance. It must require the released source train, stopped daemon, the archive ownership lease, a verified backup covering the pre-apply state, the exact mutation receipt and operation identity, unchanged archive identity and schema version, quick_check, and a post-mutation source evidence capture. Persist the refresh in the durable train manifest as an auditable successor to the released content evidence. Reject arbitrary file or SQL changes, stale receipts, wrong archive identity, version changes, missing backup attestation, and refreshes for non-source or non-released trains. Wire the blob-reference-liveness apply route through this seam and add real file-backed red tests for an unreceipted mutation and a successful authorized refresh. Do not weaken source identity, schema, or backup checks.","acceptance_criteria":"1. A valid backup-gated source liveness apply emits a typed mutation receipt that can refresh the released source train continuity evidence. 2. Startup and the next durable migration accept the refreshed train only when archive identity, schema version, quick_check, backup binding, operation identity, and post-mutation evidence match. 3. Missing, stale, malformed, wrong-archive, wrong-tier, or unreceipted mutations remain fail-closed. 4. No direct train-manifest editing or generic bypass flag is added. 5. Real file-backed tests cover successful refresh and each safety rejection. 6. Focused durable-train and blob-liveness tests plus devtools verify --quick pass. 7. Production source mutation remains under the phase-2 source-remediation receipt and is not closed by synthetic tests.","status":"open","priority":0,"issue_type":"bug","owner":"ezo.dev@gmail.com","created_at":"2026-08-06T23:49:12Z","created_by":"Sinity","updated_at":"2026-08-06T23:49:12Z","dependency_count":0,"dependent_count":1,"comment_count":0} diff --git a/polylogue/cli/commands/maintenance/_blob_integrity.py b/polylogue/cli/commands/maintenance/_blob_integrity.py index 485037bc85..aa6e193f99 100644 --- a/polylogue/cli/commands/maintenance/_blob_integrity.py +++ b/polylogue/cli/commands/maintenance/_blob_integrity.py @@ -93,8 +93,10 @@ def blob_reference_liveness_command( } if output_format == "json": click.echo(json.dumps(payload, indent=2, sort_keys=True)) - return - _render_blob_reference_liveness_plain(report, sample_limit=sample_limit) + else: + _render_blob_reference_liveness_plain(report, sample_limit=sample_limit) + if report.continuity_refresh_pending: + raise click.exceptions.Exit(1) def _render_blob_reference_liveness_plain(report: BlobRefLivenessReconciliationReport, *, sample_limit: int) -> None: @@ -120,6 +122,10 @@ def _render_blob_reference_liveness_plain(report: BlobRefLivenessReconciliationR ) if report.receipt_path is not None: click.echo(f"Receipt: {report.receipt_path}") + if report.continuity_refresh_receipt is not None: + click.echo(f"Train proof: {report.continuity_refresh_receipt}") + if report.continuity_refresh_error is not None: + click.echo(f"Train proof refresh failed: {report.continuity_refresh_error}") for candidate in report.classification.candidates[: max(0, sample_limit)]: click.echo( f" orphan {candidate.ref_type} ref_id={candidate.ref_id} " diff --git a/polylogue/maintenance/blob_ref_liveness_reconciliation.py b/polylogue/maintenance/blob_ref_liveness_reconciliation.py index 979d54713a..fc63f0ea10 100644 --- a/polylogue/maintenance/blob_ref_liveness_reconciliation.py +++ b/polylogue/maintenance/blob_ref_liveness_reconciliation.py @@ -7,11 +7,12 @@ import os import sqlite3 import time -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator from contextlib import suppress from dataclasses import dataclass +from functools import wraps from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Concatenate, ParamSpec, TypeVar, cast if TYPE_CHECKING: from polylogue.storage.blob_gc import OrphanedBlobRefCensus @@ -20,17 +21,31 @@ from polylogue.daemon.write_coordinator import daemon_write_lease_active from polylogue.maintenance.offline_guard import running_daemon_pid from polylogue.paths import render_root +from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation from polylogue.storage.blob_ref_liveness import ( BlobRefLivenessCandidate, + BlobRefLivenessCandidateDigest, BlobRefLivenessClassification, classify_blob_ref_liveness, + digest_blob_ref_liveness_candidates, stage_blob_ref_liveness, validated_blob_ref_liveness_joins, ) from polylogue.storage.hook_payload_ref_reconciliation import _deterministic_raw_session_id_udf from polylogue.storage.introspection import table_exists as _table_exists from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.durable_change_train import ( + DurableChangeTrainError, + DurableSourceContinuitySemanticError, + DurableSourceTrainMissingError, + assert_source_continuity_apply_allowed, + clear_source_continuity_pending_intent, + mark_source_continuity_pending_intent_terminal, + refresh_released_source_train_continuity, + write_source_continuity_pending_intent, +) from polylogue.storage.sqlite.migration_runner import ( + capture_durable_database_evidence, validate_migration_backup_live_fingerprint, validate_migration_backup_manifest, ) @@ -56,6 +71,30 @@ class BlobRefLivenessReconciliationError(RuntimeError): """Raised when reconciliation cannot prove a safe source-tier apply.""" +_ArchiveOwnedParams = ParamSpec("_ArchiveOwnedParams") +_ArchiveOwnedResult = TypeVar("_ArchiveOwnedResult") + + +def _archive_owned( + function: Callable[Concatenate[Path, _ArchiveOwnedParams], _ArchiveOwnedResult], +) -> Callable[Concatenate[Path, _ArchiveOwnedParams], _ArchiveOwnedResult]: + """Hold the archive lease across the complete liveness operation.""" + + @wraps(function) + def wrapped( + archive_root: Path, *args: _ArchiveOwnedParams.args, **kwargs: _ArchiveOwnedParams.kwargs + ) -> _ArchiveOwnedResult: + if bool(kwargs.get("dry_run", True)): + return function(archive_root, *args, **kwargs) + with OwnedArchiveLocation.acquire( + ArchiveLocation.resolve(archive_root), + owner_id=f"blob-ref-liveness:{os.getpid()}", + ): + return function(archive_root, *args, **kwargs) + + return cast(Callable[Concatenate[Path, _ArchiveOwnedParams], _ArchiveOwnedResult], wrapped) + + @dataclass(frozen=True, slots=True) class BlobRefLivenessReconciliationReport: source_db: str @@ -66,6 +105,9 @@ class BlobRefLivenessReconciliationReport: receipt_path: Path | None = None backup_manifest: Path | None = None post_classification: BlobRefLivenessClassification | None = None + continuity_refresh_receipt: Path | None = None + continuity_refresh_error: str | None = None + continuity_refresh_pending: bool = False def to_dict(self, *, sample_limit: int = 30) -> dict[str, object]: return { @@ -75,6 +117,11 @@ def to_dict(self, *, sample_limit: int = 30) -> dict[str, object]: "deleted_count": self.deleted_count, "receipt_path": str(self.receipt_path) if self.receipt_path is not None else None, "backup_manifest": str(self.backup_manifest) if self.backup_manifest is not None else None, + "continuity_refresh_receipt": ( + str(self.continuity_refresh_receipt) if self.continuity_refresh_receipt is not None else None + ), + "continuity_refresh_error": self.continuity_refresh_error, + "continuity_refresh_pending": self.continuity_refresh_pending, "post_classification": self.post_classification.to_dict() if self.post_classification is not None else None, **self.classification.to_dict(sample_limit=sample_limit), } @@ -120,16 +167,7 @@ def _source_data_version(conn: sqlite3.Connection) -> int: def _candidate_digest(candidates: Iterable[BlobRefLivenessCandidate]) -> str: - digest = hashlib.sha256() - digest.update(b"[") - first = True - for candidate in candidates: - if not first: - digest.update(b",") - digest.update(json.dumps(candidate.to_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")) - first = False - digest.update(b"]") - return digest.hexdigest() + return digest_blob_ref_liveness_candidates(candidates) def _iter_staged_candidates(conn: sqlite3.Connection, table_name: str) -> Iterator[BlobRefLivenessCandidate]: @@ -169,9 +207,13 @@ def _write_prepared_receipt( *, candidates: Iterable[BlobRefLivenessCandidate] | None = None, candidate_digest: str | None = None, + backup_manifest_sha256: str | None = None, ) -> None: + receipt_parent_existed = receipt_path.parent.is_dir() receipt_path.parent.mkdir(parents=True, exist_ok=True) - header = { + if not receipt_parent_existed: + _fsync_directory(receipt_path.parent.parent) + header: dict[str, object] = { "kind": "blob_ref_liveness_reconciliation", "phase": "prepared", "tool_version": TOOL_VERSION, @@ -187,6 +229,10 @@ def _write_prepared_receipt( for ref_type, table, column in classification.ref_type_joins ], } + if backup_manifest_sha256 is None and backup_manifest.is_file(): + backup_manifest_sha256 = hashlib.sha256(backup_manifest.read_bytes()).hexdigest() + if backup_manifest_sha256 is not None: + header["backup_manifest_sha256"] = backup_manifest_sha256 try: with receipt_path.open("x", encoding="utf-8") as handle: handle.write(json.dumps(header, sort_keys=True, separators=(",", ":"))) @@ -237,7 +283,12 @@ def _receipt_candidate(row: dict[str, object]) -> tuple[bytes, str, str]: raise BlobRefLivenessReconciliationError("prepared receipt contains an invalid candidate") from exc -def _stage_receipt_candidates(conn: sqlite3.Connection, receipt_path: Path) -> tuple[int, str, dict[str, object]]: +def _stage_receipt_candidates( + conn: sqlite3.Connection, + receipt_path: Path, + *, + allow_postcondition_failed: bool = False, +) -> tuple[int, str, dict[str, object]]: table_name = "blob_ref_liveness_receipt_candidates" conn.execute(f"DROP TABLE IF EXISTS temp.{table_name}") conn.execute( @@ -252,9 +303,7 @@ def _stage_receipt_candidates(conn: sqlite3.Connection, receipt_path: Path) -> t ) header: dict[str, object] | None = None terminal_phases: list[str] = [] - digest = hashlib.sha256() - digest.update(b"[") - first_candidate = True + digest = BlobRefLivenessCandidateDigest() candidate_count = 0 with receipt_path.open(encoding="utf-8") as handle: for line in handle: @@ -276,40 +325,43 @@ def _stage_receipt_candidates(conn: sqlite3.Connection, receipt_path: Path) -> t elif row.get("kind") == "candidate": candidate = _receipt_candidate(row) conn.execute(f"INSERT INTO {table_name} (blob_hash, ref_type, ref_id) VALUES (?, ?, ?)", candidate) - if not first_candidate: - digest.update(b",") - digest.update( - json.dumps( - { - "blob_hash": candidate[0].hex(), - "ref_id": candidate[2], - "ref_type": candidate[1], - "source_path": row.get("source_path"), - "size_bytes": row.get("size_bytes"), - "acquired_at_ms": row.get("acquired_at_ms"), - "referent_table": row.get("referent_table"), - "referent_column": row.get("referent_column"), - }, - sort_keys=True, - separators=(",", ":"), - ).encode("utf-8") - ) - first_candidate = False + try: + digest.update( + BlobRefLivenessCandidate( + blob_hash=candidate[0].hex(), + ref_type=candidate[1], + ref_id=candidate[2], + source_path=str(row["source_path"]) if row.get("source_path") is not None else None, + size_bytes=int(row["size_bytes"]), + acquired_at_ms=int(row["acquired_at_ms"]), + referent_table=str(row["referent_table"]), + referent_column=str(row["referent_column"]), + ) + ) + except (KeyError, TypeError, ValueError) as exc: + raise BlobRefLivenessReconciliationError( + f"prepared receipt contains an invalid candidate: {receipt_path}" + ) from exc candidate_count += 1 if header is None or header.get("phase") != "prepared": raise BlobRefLivenessReconciliationError(f"receipt does not contain a prepared plan: {receipt_path}") - if terminal_phases: + if terminal_phases and not (allow_postcondition_failed and terminal_phases == ["postcondition_failed"]): raise BlobRefLivenessReconciliationError(f"receipt is already terminal: {receipt_path}") receipt_candidate_count = header.get("candidate_count") if not isinstance(receipt_candidate_count, int) or receipt_candidate_count != candidate_count: raise BlobRefLivenessReconciliationError(f"prepared receipt candidate count mismatch: {receipt_path}") - digest.update(b"]") if str(header.get("candidate_digest")) != digest.hexdigest(): raise BlobRefLivenessReconciliationError(f"prepared receipt candidate digest mismatch: {receipt_path}") return candidate_count, table_name, header -def _recover_prepared_receipt(source_db: Path, receipt_path: Path) -> str: +def _recover_prepared_receipt( + source_db: Path, + receipt_path: Path, + *, + allow_postcondition_failed: bool = False, + postcondition_check: Callable[[], None] | None = None, +) -> str: """Resolve crashes between bounded batch commits and receipt progress. Each batch is one SQLite transaction. Exact receipt keys therefore show a @@ -318,7 +370,11 @@ def _recover_prepared_receipt(source_db: Path, receipt_path: Path) -> str: """ with sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) as conn: - candidate_count, table_name, _header = _stage_receipt_candidates(conn, receipt_path) + candidate_count, table_name, _header = _stage_receipt_candidates( + conn, + receipt_path, + allow_postcondition_failed=allow_postcondition_failed, + ) present_count = int( conn.execute( f""" @@ -337,6 +393,8 @@ def _recover_prepared_receipt(source_db: Path, receipt_path: Path) -> str: outcome = "recovered_committed" else: outcome = "recovered_partial" + if outcome == "recovered_committed" and postcondition_check is not None: + postcondition_check() _append_receipt_footer( receipt_path, phase=outcome, @@ -706,6 +764,7 @@ def _validate_locked_candidate_plan( ) +@_archive_owned def reconcile_blob_ref_liveness( archive_root: Path, *, @@ -721,6 +780,7 @@ def reconcile_blob_ref_liveness( after ``BEGIN IMMEDIATE`` before bounded deletes start. """ + archive_root = archive_root.resolve() source_db = archive_root / "source.db" if not source_db.exists(): raise FileNotFoundError(f"no source.db at {source_db}") @@ -744,6 +804,8 @@ def reconcile_blob_ref_liveness( raise BlobRefLivenessReconciliationError( "applying blob-ref liveness reconciliation requires a receipt path (--receipt-file)" ) + backup_manifest = backup_manifest.resolve() + receipt_path = receipt_path.resolve() if receipt_path.exists(): outcome = _recover_prepared_receipt(source_db, receipt_path) raise BlobRefLivenessReconciliationError( @@ -751,13 +813,26 @@ def reconcile_blob_ref_liveness( ) if reason := _offline_apply_block_reason(archive_root): raise BlobRefLivenessReconciliationError(reason) + try: + assert_source_continuity_apply_allowed(archive_root) + except DurableChangeTrainError as exc: + raise BlobRefLivenessReconciliationError(str(exc)) from exc pre_conn = sqlite3.connect(source_db) staged_plan = None staged_data_version: int | None = None try: _checkpoint_source_db(pre_conn) + validated_backup_digest = ( + hashlib.sha256(backup_manifest.read_bytes()).hexdigest() if backup_manifest.is_file() else None + ) validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=pre_conn) + if ( + validated_backup_digest is not None + and hashlib.sha256(backup_manifest.read_bytes()).hexdigest() != validated_backup_digest + ): + raise BlobRefLivenessReconciliationError("backup manifest changed during validation") + pre_mutation_evidence = capture_durable_database_evidence(pre_conn, ArchiveTier.SOURCE) staged_plan = stage_blob_ref_liveness(pre_conn) classification = staged_plan.classification if not classification.safe_to_apply: @@ -777,6 +852,15 @@ def reconcile_blob_ref_liveness( backup_manifest, candidates=candidates, candidate_digest=candidate_digest, + backup_manifest_sha256=validated_backup_digest, + ) + pending_intent = write_source_continuity_pending_intent( + archive_root, + mutation_receipt=receipt_path, + backup_manifest=backup_manifest, + pre_mutation_evidence=pre_mutation_evidence, + operation_id=candidate_digest, + evidence_ref=f"proof:blob-ref-liveness:{candidate_digest}", ) expected_count = classification.orphaned_count staged_data_version = _source_data_version(pre_conn) @@ -901,6 +985,56 @@ def reconcile_blob_ref_liveness( f"source.db committed but could not finalize receipt {receipt_path}" ) from exc + continuity_refresh_receipt: Path | None = None + continuity_refresh_error: str | None = None + continuity_refresh_pending = False + try: + continuity_refresh_receipt = refresh_released_source_train_continuity( + archive_root, + mutation_receipt=receipt_path, + backup_manifest=backup_manifest, + pre_mutation_evidence=pre_mutation_evidence, + operation_id=candidate_digest, + evidence_ref=f"proof:blob-ref-liveness:{candidate_digest}", + ) + try: + clear_source_continuity_pending_intent(pending_intent) + except Exception as cleanup_exc: + # The refresh is durable. Startup can consume the remaining + # intent idempotently, so preserve the committed report and mark + # only the cleanup residual as pending. + continuity_refresh_error = f"pending intent cleanup failed: {cleanup_exc}" + continuity_refresh_pending = True + except DurableSourceTrainMissingError as exc: + try: + clear_source_continuity_pending_intent(pending_intent) + except Exception as cleanup_exc: + continuity_refresh_error = f"{exc}; pending intent cleanup failed: {cleanup_exc}" + continuity_refresh_pending = True + else: + # A fresh archive has no released source train to refresh. The + # committed source mutation is complete and there is no pending + # continuity recovery obligation in this case. + continuity_refresh_error = None + except DurableSourceContinuitySemanticError as exc: + # Semantic continuity rejection cannot become valid by retrying the + # same committed source mutation. Preserve it durably for startup to + # consume without hiding the fail-closed train mismatch. + try: + mark_source_continuity_pending_intent_terminal(pending_intent, error=exc) + except Exception as terminalization_exc: + continuity_refresh_error = f"{exc}; pending intent terminalization failed: {terminalization_exc}" + else: + continuity_refresh_error = str(exc) + continuity_refresh_pending = True + except Exception as exc: + # The source deletion and its committed receipt are already durable. + # Keep the report truthful while leaving the train fail-closed until a + # separate continuity refresh succeeds. This boundary also normalizes + # filesystem and SQLite failures after the irreversible commit. + continuity_refresh_error = str(exc) + continuity_refresh_pending = True + assert staged_plan is not None return BlobRefLivenessReconciliationReport( source_db=str(source_db), @@ -911,6 +1045,9 @@ def reconcile_blob_ref_liveness( receipt_path=receipt_path, backup_manifest=backup_manifest, post_classification=post_classification, + continuity_refresh_receipt=continuity_refresh_receipt, + continuity_refresh_error=continuity_refresh_error, + continuity_refresh_pending=continuity_refresh_pending, ) diff --git a/polylogue/storage/blob_ref_liveness.py b/polylogue/storage/blob_ref_liveness.py index 16b0fc6272..ea79848360 100644 --- a/polylogue/storage/blob_ref_liveness.py +++ b/polylogue/storage/blob_ref_liveness.py @@ -8,8 +8,10 @@ from __future__ import annotations +import hashlib +import json import sqlite3 -from collections.abc import Iterator +from collections.abc import Iterable, Iterator from dataclasses import dataclass from typing import cast @@ -48,6 +50,33 @@ def to_dict(self) -> dict[str, object]: } +class BlobRefLivenessCandidateDigest: + """Incrementally encode the canonical candidate stream used by receipts.""" + + def __init__(self) -> None: + self._digest = hashlib.sha256(b"[") + self._first = True + + def update(self, candidate: BlobRefLivenessCandidate) -> None: + if not self._first: + self._digest.update(b",") + self._digest.update(json.dumps(candidate.to_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8")) + self._first = False + + def hexdigest(self) -> str: + digest = self._digest.copy() + digest.update(b"]") + return digest.hexdigest() + + +def digest_blob_ref_liveness_candidates(candidates: Iterable[BlobRefLivenessCandidate]) -> str: + """Hash candidates with the receipt's canonical JSON array framing.""" + digest = BlobRefLivenessCandidateDigest() + for candidate in candidates: + digest.update(candidate) + return digest.hexdigest() + + @dataclass(frozen=True, slots=True) class BlobRefLivenessClassification: """Complete read-only classification of source-tier blob references.""" diff --git a/polylogue/storage/sqlite/durable_change_train.py b/polylogue/storage/sqlite/durable_change_train.py index 7532b6f3a8..2816f3bf2c 100644 --- a/polylogue/storage/sqlite/durable_change_train.py +++ b/polylogue/storage/sqlite/durable_change_train.py @@ -2,19 +2,25 @@ from __future__ import annotations +import hashlib import importlib import inspect +import io import json import os import re import sqlite3 import tempfile from collections.abc import Callable, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from importlib import resources from pathlib import Path from typing import Final, cast +from polylogue.storage.blob_ref_liveness import ( + BlobRefLivenessCandidate, + BlobRefLivenessCandidateDigest, +) from polylogue.storage.sqlite import migration_runner as _migration_runner from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.migration_runner import ( @@ -24,11 +30,15 @@ DurableChangeTrainError, DurableChangeTrainRecoveryError, DurableChangeTrainState, + DurableDatabaseEvidence, + DurableFailureClassification, DurableFreshDDLParityProof, DurableMigrationClaim, DurableRuntimeConsumerResult, MigrationResult, _assert_durable_database_continuity, + _canonical_json_sha256, + _require_nonempty, _validate_riders, add_durable_change_train_rider, admit_durable_change_train, @@ -61,6 +71,15 @@ _SIDECAR_NAME_RE = re.compile(r"^(?P\d{3,})\.train\.json$") _MIGRATION_NAME_RE = re.compile(r"^(?P\d{3,})_[a-z0-9_]+\.sql$") _DROP_SQL_RE = re.compile(r"(?is)\bDROP\s+(?:TABLE|INDEX|TRIGGER|VIEW)\b") +_SOURCE_CONTINUITY_PENDING_FORMAT = "polylogue.source-continuity-pending.v1" + + +class DurableSourceTrainMissingError(DurableChangeTrainError): + """Raised when an archive has no released source train to refresh.""" + + +class DurableSourceContinuitySemanticError(DurableChangeTrainError): + """A committed source mutation cannot satisfy immutable train evidence.""" @dataclass(frozen=True, slots=True) @@ -290,6 +309,665 @@ def _persist_train_transition(path: Path, train: DurableChangeTrain, *, expected return load_durable_change_train_manifest(path) +def write_source_continuity_pending_intent( + archive_root: Path, + *, + mutation_receipt: Path, + backup_manifest: Path, + pre_mutation_evidence: DurableDatabaseEvidence, + operation_id: str, + evidence_ref: str, +) -> Path: + """Persist the recovery input before a source mutation can commit.""" + mutation_receipt = mutation_receipt.resolve() + backup_manifest = backup_manifest.resolve() + pending_root = archive_root / ".maintenance-state" / "source-continuity-pending" + pending_root_existed = pending_root.is_dir() + pending_root.mkdir(parents=True, exist_ok=True) + if not pending_root_existed: + _migration_runner._fsync_manifest_directory(pending_root.parent) + payload: dict[str, object] = { + "format": _SOURCE_CONTINUITY_PENDING_FORMAT, + "mutation_receipt": str(mutation_receipt), + "backup_manifest": str(backup_manifest), + "operation_id": operation_id, + "evidence_ref": evidence_ref, + "source_before": _migration_runner._manifest_json_value(pre_mutation_evidence), + } + pending_digest = _canonical_json_sha256(payload) + path = pending_root / f"{pending_digest}.json" + encoded = (json.dumps({**payload, "pending_sha256": pending_digest}, indent=2, sort_keys=True) + "\n").encode( + "utf-8" + ) + if path.exists(): + try: + existing = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"source continuity pending intent is unreadable: {path}") from exc + if existing != {**payload, "pending_sha256": pending_digest}: + raise DurableChangeTrainError("source continuity pending intent collision") + return path + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + dir=pending_root, prefix=f".{path.name}.", suffix=".tmp", delete=False + ) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + temporary = None + _migration_runner._fsync_manifest_directory(pending_root) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + return path + + +def _load_source_continuity_pending_intent(path: Path) -> dict[str, object]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"source continuity pending intent is unreadable: {path}") from exc + if not isinstance(raw, dict): + raise DurableChangeTrainError(f"source continuity pending intent is not an object: {path}") + pending_digest = raw.pop("pending_sha256", None) + if not isinstance(pending_digest, str) or pending_digest != _canonical_json_sha256(raw): + raise DurableChangeTrainError(f"source continuity pending intent checksum mismatch: {path}") + if raw.get("format") != _SOURCE_CONTINUITY_PENDING_FORMAT: + raise DurableChangeTrainError(f"unsupported source continuity pending intent: {path}") + return cast(dict[str, object], raw) + + +def _replace_source_continuity_pending_intent(path: Path, payload: dict[str, object]) -> None: + encoded = ( + json.dumps({**payload, "pending_sha256": _canonical_json_sha256(payload)}, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False + ) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + temporary = None + _migration_runner._fsync_manifest_directory(path.parent) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def mark_source_continuity_pending_intent_terminal(path: Path, *, error: DurableSourceContinuitySemanticError) -> None: + """Persist a semantic refresh rejection so startup does not retry it forever.""" + payload = _load_source_continuity_pending_intent(path) + terminal = {"kind": "continuity_refresh_rejected", "error": str(error)} + existing = payload.get("terminal_outcome") + if existing == terminal: + return + if existing is not None: + raise DurableChangeTrainError(f"source continuity pending intent has an unknown terminal outcome: {path}") + _replace_source_continuity_pending_intent(path, {**payload, "terminal_outcome": terminal}) + + +def clear_source_continuity_pending_intent(path: Path) -> None: + """Remove a consumed pending intent only after manifest refresh succeeds.""" + try: + path.unlink() + except FileNotFoundError: + return + _migration_runner._fsync_manifest_directory(path.parent) + + +def assert_source_continuity_apply_allowed(archive_root: Path) -> None: + """Reject a new source mutation that could invalidate continuity recovery.""" + archive_root = archive_root.resolve() + pending_root = archive_root / ".maintenance-state" / "source-continuity-pending" + pending_intents = tuple(sorted(pending_root.glob("*.json"))) if pending_root.is_dir() else () + if pending_intents: + raise DurableChangeTrainError("source liveness apply is blocked while source continuity recovery is pending") + + source_path = archive_root / "source.db" + with sqlite3.connect(f"file:{source_path}?mode=ro", uri=True) as connection: + current_version = int(connection.execute("PRAGMA user_version").fetchone()[0] or 0) + manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" + if not manifest_root.is_dir(): + return + unreleased: list[Path] = [] + released: list[DurableChangeTrain] = [] + for candidate in sorted(manifest_root.glob("source-*.json")): + train = load_durable_change_train_manifest(candidate) + rollback_failed_train = ( + train.state is DurableChangeTrainState.FAILED + and train.failure is not None + and train.failure.classification is DurableFailureClassification.ROLLED_BACK_TO_CURRENT + and train.current_version == current_version + ) + if ( + train.target_version != current_version + and not ( + train.reservation is not None and train.reservation.active and train.current_version == current_version + ) + and not rollback_failed_train + ): + continue + if train.state is DurableChangeTrainState.RELEASED: + released.append(train) + else: + unreleased.append(candidate) + if unreleased: + raise DurableChangeTrainError( + "source liveness apply is blocked by an unreleased source train for the live schema" + ) + if len(released) > 1: + raise DurableChangeTrainError( + "source liveness apply requires exactly one released source train for the live schema" + ) + if released: + with sqlite3.connect(f"file:{source_path}?mode=ro", uri=True) as connection: + _verify_released_train_live_tier(archive_root, connection, released[0]) + + +def _recover_pending_source_continuity_intents(archive_root: Path) -> None: + """Finish committed source mutations whose manifest refresh was interrupted.""" + pending_root = archive_root / ".maintenance-state" / "source-continuity-pending" + if not pending_root.is_dir(): + return + for path in sorted(pending_root.glob("*.json")): + raw = _load_source_continuity_pending_intent(path) + terminal = raw.get("terminal_outcome") + if terminal is not None: + if not ( + isinstance(terminal, dict) + and terminal.get("kind") == "continuity_refresh_rejected" + and isinstance(terminal.get("error"), str) + ): + raise DurableChangeTrainError( + f"source continuity pending intent has an invalid terminal outcome: {path}" + ) + clear_source_continuity_pending_intent(path) + continue + try: + evidence_raw = raw["source_before"] + if not isinstance(evidence_raw, dict): + raise TypeError("source_before is not an object") + pre_mutation_evidence = _migration_runner._decode_manifest_value( + DurableDatabaseEvidence, + evidence_raw, + label=f"{path}.source_before", + ) + if not isinstance(pre_mutation_evidence, DurableDatabaseEvidence): + raise TypeError("source_before decoded to the wrong type") + receipt = Path(str(raw["mutation_receipt"])) + backup = Path(str(raw["backup_manifest"])) + operation_id = str(raw["operation_id"]) + evidence_ref = str(raw["evidence_ref"]) + except (DurableChangeTrainError, KeyError, TypeError, ValueError) as exc: + raise DurableChangeTrainError(f"source continuity pending intent is malformed: {path}") from exc + receipt_phase = _liveness_receipt_phase(receipt) + if receipt_phase == "recovered_rolled_back": + clear_source_continuity_pending_intent(path) + continue + if receipt_phase in {"prepared", "batch_committed"}: + from polylogue.maintenance.blob_ref_liveness_reconciliation import _recover_prepared_receipt + + outcome = _recover_prepared_receipt(archive_root / "source.db", receipt) + if outcome == "recovered_rolled_back": + clear_source_continuity_pending_intent(path) + continue + if outcome == "recovered_partial": + raise DurableChangeTrainError(f"source continuity pending intent has a partial source mutation: {path}") + receipt_phase = outcome + if receipt_phase == "postcondition_failed": + from polylogue.maintenance.blob_ref_liveness_reconciliation import _recover_prepared_receipt + from polylogue.storage.blob_ref_liveness import classify_blob_ref_liveness + + def validate_recovered_postcondition(pending_path: Path = path) -> None: + with sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) as connection: + classification = classify_blob_ref_liveness(connection) + if not classification.safe_to_apply or classification.orphaned_count != 0: + raise DurableChangeTrainError( + f"source continuity pending intent postcondition remains unsafe: {pending_path}" + ) + + outcome = _recover_prepared_receipt( + archive_root / "source.db", + receipt, + allow_postcondition_failed=True, + postcondition_check=validate_recovered_postcondition, + ) + if outcome == "recovered_rolled_back": + clear_source_continuity_pending_intent(path) + continue + if outcome == "recovered_partial": + raise DurableChangeTrainError(f"source continuity pending intent has a partial source mutation: {path}") + receipt_phase = outcome + if receipt_phase not in {"committed", "recovered_committed"}: + raise DurableChangeTrainError(f"source continuity pending intent has no committed receipt: {path}") + try: + refresh_released_source_train_continuity( + archive_root, + mutation_receipt=receipt, + backup_manifest=backup, + pre_mutation_evidence=pre_mutation_evidence, + operation_id=operation_id, + evidence_ref=evidence_ref, + ) + except DurableSourceTrainMissingError: + clear_source_continuity_pending_intent(path) + except DurableSourceContinuitySemanticError as exc: + mark_source_continuity_pending_intent_terminal(path, error=exc) + else: + clear_source_continuity_pending_intent(path) + + +def _liveness_receipt_phase(receipt_path: Path) -> str: + """Read the last liveness phase before deciding how a pending intent recovers.""" + try: + last_record: object | None = None + with receipt_path.open(encoding="utf-8") as handle: + for line in handle: + if line.strip(): + last_record = json.loads(line) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"source continuity pending receipt is unreadable: {receipt_path}") from exc + if not isinstance(last_record, dict): + raise DurableChangeTrainError(f"source continuity pending receipt is incomplete: {receipt_path}") + phase = last_record.get("phase") + if not isinstance(phase, str): + raise DurableChangeTrainError(f"source continuity pending receipt has no phase: {receipt_path}") + return phase + + +def _validate_liveness_receipt_bytes( + receipt_bytes: bytes, + *, + source_path: Path, + backup_manifest: Path, + operation_id: str, +) -> dict[str, object]: + """Validate the exact candidate stream and terminal footer of a liveness receipt.""" + header: dict[str, object] | None = None + footer: dict[str, object] | None = None + candidate_digest = BlobRefLivenessCandidateDigest() + candidate_count = 0 + try: + for raw_line in io.BytesIO(receipt_bytes): + if not raw_line.strip(): + continue + record = json.loads(raw_line) + if not isinstance(record, dict): + raise DurableChangeTrainError("source mutation receipt contains a non-object record") + if header is None: + header = cast(dict[str, object], record) + continue + if footer is not None: + row = footer + if row.get("kind") == "blob_ref_liveness_reconciliation": + previous_phase = row.get("phase") + current_phase = ( + record.get("phase") if record.get("kind") == "blob_ref_liveness_reconciliation" else None + ) + if not ( + previous_phase == "batch_committed" + or previous_phase == "postcondition_failed" + and current_phase == "recovered_committed" + ): + raise DurableChangeTrainError( + "source mutation receipt contains an unexpected intermediate footer" + ) + elif row.get("kind") == "candidate": + try: + blob_hash = str(row["blob_hash"]) + bytes.fromhex(blob_hash) + size_bytes = row["size_bytes"] + acquired_at_ms = row["acquired_at_ms"] + if not isinstance(size_bytes, int) or isinstance(size_bytes, bool): + raise TypeError("candidate size_bytes is not an integer") + if not isinstance(acquired_at_ms, int) or isinstance(acquired_at_ms, bool): + raise TypeError("candidate acquired_at_ms is not an integer") + candidate_digest.update( + BlobRefLivenessCandidate( + blob_hash=blob_hash, + ref_type=str(row["ref_type"]), + ref_id=str(row["ref_id"]), + source_path=str(row["source_path"]) if row.get("source_path") is not None else None, + size_bytes=size_bytes, + acquired_at_ms=acquired_at_ms, + referent_table=str(row["referent_table"]), + referent_column=str(row["referent_column"]), + ) + ) + except (KeyError, TypeError, ValueError) as exc: + raise DurableChangeTrainError("source mutation receipt contains an invalid candidate") from exc + candidate_count += 1 + else: + raise DurableChangeTrainError("source mutation receipt contains an unexpected record") + footer = cast(dict[str, object], record) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError("source mutation receipt is not valid JSONL") from exc + if header is None or footer is None: + raise DurableChangeTrainError("source mutation receipt is incomplete") + if ( + header.get("kind") != "blob_ref_liveness_reconciliation" + or header.get("phase") != "prepared" + or header.get("source_db") != str(source_path) + or header.get("backup_manifest") != str(backup_manifest) + or header.get("candidate_digest") != operation_id + or footer.get("kind") != "blob_ref_liveness_reconciliation" + or footer.get("phase") not in {"committed", "recovered_committed"} + or footer.get("deleted_count") != header.get("candidate_count") + or (footer.get("phase") == "committed" and footer.get("post_orphaned_count") != 0) + ): + raise DurableChangeTrainError("source mutation receipt does not bind the named liveness operation") + + if ( + header.get("candidate_count") != candidate_count + or header.get("candidate_digest") != candidate_digest.hexdigest() + ): + raise DurableChangeTrainError("source mutation receipt candidate digest or count mismatch") + return header + + +def _validate_source_continuity_refresh_receipt( + archive_root: Path, + train: DurableChangeTrain, +) -> None: + """Require the latest source continuity evidence to retain its receipt.""" + if train.source_continuity_evidence is None: + return + expected_after = _migration_runner._manifest_json_value(train.source_continuity_evidence) + refresh_root = archive_root / ".maintenance-state" / "source-continuity-refreshes" + refresh_refs = [ + ref.removeprefix("proof:source-continuity-refresh:") + for ref in train.proof_refs + if ref.startswith("proof:source-continuity-refresh:") + ] + if not refresh_refs: + raise DurableChangeTrainError("source continuity evidence has no retained refresh receipt") + matches = 0 + for digest in refresh_refs: + receipt_path = refresh_root / f"{digest}.json" + payload = _read_source_continuity_refresh_receipt(receipt_path, digest=digest, train=train) + if payload.get("source_after") == expected_after: + matches += 1 + if matches != 1: + raise DurableChangeTrainError( + "source continuity evidence does not identify exactly one matching refresh receipt" + ) + + +def _read_source_continuity_refresh_receipt( + receipt_path: Path, + *, + digest: str, + train: DurableChangeTrain, +) -> dict[str, object]: + """Load one train-retained refresh artifact and authenticate its identity.""" + try: + raw = json.loads(receipt_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError(f"source continuity refresh receipt is unreadable: {receipt_path}") from exc + if not isinstance(raw, dict): + raise DurableChangeTrainError(f"source continuity refresh receipt is not an object: {receipt_path}") + payload = cast(dict[str, object], raw) + refresh_sha256 = payload.pop("refresh_sha256", None) + if refresh_sha256 != digest or _canonical_json_sha256(payload) != digest: + raise DurableChangeTrainError(f"source continuity refresh receipt checksum mismatch: {receipt_path}") + if payload.get("format") != "polylogue.source-continuity-refresh.v1": + raise DurableChangeTrainError(f"source continuity refresh receipt format mismatch: {receipt_path}") + if payload.get("train_id") != train.train_id: + raise DurableChangeTrainError(f"source continuity refresh receipt train mismatch: {receipt_path}") + return payload + + +def refresh_released_source_train_continuity( + archive_root: Path, + *, + mutation_receipt: Path, + backup_manifest: Path, + pre_mutation_evidence: DurableDatabaseEvidence, + operation_id: str, + evidence_ref: str, +) -> Path: + """Refresh released source-train continuity while retaining archive ownership.""" + from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation + + with OwnedArchiveLocation.acquire( + ArchiveLocation.resolve(archive_root), + owner_id=f"source-continuity-refresh:{os.getpid()}", + allow_reentrant=True, + ): + return _refresh_released_source_train_continuity_locked( + archive_root, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=pre_mutation_evidence, + operation_id=operation_id, + evidence_ref=evidence_ref, + ) + + +def _refresh_released_source_train_continuity_locked( + archive_root: Path, + *, + mutation_receipt: Path, + backup_manifest: Path, + pre_mutation_evidence: DurableDatabaseEvidence, + operation_id: str, + evidence_ref: str, +) -> Path: + """Record an authorized source mutation without weakening train checks. + + Source maintenance may change rows after a schema train is released. The + caller must first validate the named operation and backup against the + pre-mutation live tier. This helper then binds those exact receipt and + backup bytes to a separate current-evidence record. The original migration + evidence remains immutable in ``apply_evidence``. + """ + from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation + + archive_root = archive_root.resolve() + mutation_receipt = mutation_receipt.resolve() + backup_manifest = backup_manifest.resolve() + + _require_nonempty(operation_id, label="source mutation operation id") + _require_nonempty(evidence_ref, label="source continuity evidence ref") + if not mutation_receipt.is_file() or mutation_receipt.is_symlink(): + raise DurableChangeTrainError("source mutation receipt is not a real file") + if not backup_manifest.is_file() or backup_manifest.is_symlink(): + raise DurableChangeTrainError("source mutation backup manifest is not a real file") + + source_path = archive_root / "source.db" + try: + receipt_bytes = mutation_receipt.read_bytes() + except OSError as exc: + raise DurableChangeTrainError("source mutation receipt is not readable") from exc + header = _validate_liveness_receipt_bytes( + receipt_bytes, + source_path=source_path, + backup_manifest=backup_manifest, + operation_id=operation_id, + ) + + mutation_digest = hashlib.sha256(receipt_bytes).hexdigest() + backup_digest = hashlib.sha256(backup_manifest.read_bytes()).hexdigest() + receipt_backup_digest = header.get("backup_manifest_sha256") + if not isinstance(receipt_backup_digest, str) or receipt_backup_digest != backup_digest: + raise DurableChangeTrainError("source mutation receipt backup manifest digest mismatch") + with OwnedArchiveLocation.acquire( + ArchiveLocation.resolve(archive_root), + owner_id=f"source-continuity-refresh:{os.getpid()}", + allow_reentrant=True, + ): + with sqlite3.connect(f"file:{source_path}?mode=ro", uri=True) as connection: + current = capture_durable_database_evidence(connection, ArchiveTier.SOURCE) + + manifest_candidates = sorted( + (archive_root / ".maintenance-state" / "durable-change-trains").glob("source-*.json") + ) + if not manifest_candidates: + raise DurableSourceTrainMissingError("source continuity refresh found no released source train") + matching: list[tuple[Path, DurableChangeTrain]] = [] + for candidate in manifest_candidates: + candidate_train = load_durable_change_train_manifest(candidate) + if ( + candidate_train.state is DurableChangeTrainState.RELEASED + and candidate_train.tier is ArchiveTier.SOURCE + and candidate_train.target_version == current.user_version + ): + matching.append((candidate, candidate_train)) + if not matching: + raise DurableSourceTrainMissingError("source continuity refresh found no released source train") + if len(matching) != 1: + raise DurableChangeTrainError( + "source continuity refresh requires exactly one released source train for the live schema" + ) + manifest_path, train = matching[0] + if train.state is not DurableChangeTrainState.RELEASED: + raise DurableChangeTrainError("source continuity refresh requires a released source train") + if train.tier is not ArchiveTier.SOURCE: + raise DurableChangeTrainError("source continuity refresh selected a non-source train") + if train.apply_evidence is None: + raise DurableChangeTrainError("source continuity refresh requires apply evidence") + refresh_root = archive_root / ".maintenance-state" / "source-continuity-refreshes" + serialized_before = _migration_runner._manifest_json_value(pre_mutation_evidence) + serialized_current = _migration_runner._manifest_json_value(current) + retained_current = train.source_continuity_evidence + current_matches_retained = False + if retained_current is not None: + try: + _migration_runner._assert_durable_database_continuity( + current, + retained_current, + label="source continuity retained refresh", + ) + except DurableChangeTrainError: + pass + else: + current_matches_retained = True + serialized_retained_current = ( + _migration_runner._manifest_json_value(retained_current) if retained_current is not None else None + ) + retained_refreshes: list[tuple[Path, dict[str, object]]] = [] + retained_refs = { + ref.removeprefix("proof:source-continuity-refresh:") + for ref in train.proof_refs + if ref.startswith("proof:source-continuity-refresh:") + } + for existing_path in sorted(refresh_root.glob("*.json")) if refresh_root.is_dir() else (): + digest = existing_path.stem + if digest not in retained_refs: + continue + existing = _read_source_continuity_refresh_receipt(existing_path, digest=digest, train=train) + if existing.get("mutation_receipt_sha256") == mutation_digest: + retained_refreshes.append((existing_path, existing)) + for existing_path, existing in retained_refreshes: + # A crash after manifest persistence can leave its pending intent + # behind. The exact retained artifact, including its before/after + # evidence, authenticates this idempotent completion. + if ( + current_matches_retained + and existing.get("source_before") == serialized_before + and existing.get("source_after") == serialized_retained_current + ): + _validate_source_continuity_refresh_receipt(archive_root, train) + return existing_path + if pre_mutation_evidence.user_version != train.target_version: + raise DurableSourceContinuitySemanticError( + "source continuity refresh pre-state has the wrong schema version" + ) + if current.user_version != train.target_version: + raise DurableSourceContinuitySemanticError("source continuity refresh changed the schema version") + baseline = train.source_continuity_evidence or train.apply_evidence.post + try: + _migration_runner._assert_durable_database_continuity( + pre_mutation_evidence, + baseline, + label="source continuity pre-mutation", + ) + except DurableChangeTrainError as exc: + raise DurableSourceContinuitySemanticError( + "source continuity refresh pre-state contains unreceipted content drift" + ) from exc + if pre_mutation_evidence.archive_identity_digest != train.apply_evidence.post.archive_identity_digest: + raise DurableSourceContinuitySemanticError( + "source continuity refresh pre-state has the wrong archive identity" + ) + if current.archive_identity_digest != train.apply_evidence.post.archive_identity_digest: + raise DurableSourceContinuitySemanticError("source continuity refresh changed archive identity") + if pre_mutation_evidence.quick_check != ("ok",) or current.quick_check != ("ok",): + raise DurableSourceContinuitySemanticError( + "source continuity refresh requires successful quick_check evidence" + ) + + payload = { + "format": "polylogue.source-continuity-refresh.v1", + "operation_id": operation_id, + "evidence_ref": evidence_ref, + "backup_manifest": str(backup_manifest), + "backup_manifest_sha256": backup_digest, + "mutation_receipt": str(mutation_receipt), + "mutation_receipt_sha256": mutation_digest, + "train_id": train.train_id, + "source_before": serialized_before, + "source_after": serialized_current, + "refreshed_at_ms": current.observed_at_ms, + } + refresh_digest = _canonical_json_sha256(payload) + refresh_root_existed = refresh_root.is_dir() + refresh_root.mkdir(parents=True, exist_ok=True) + if not refresh_root_existed: + _migration_runner._fsync_manifest_directory(refresh_root.parent) + refresh_path = refresh_root / f"{refresh_digest}.json" + if refresh_path.exists(): + try: + existing = json.loads(refresh_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise DurableChangeTrainError( + f"source continuity refresh receipt is unreadable: {refresh_path}" + ) from exc + if existing != {**payload, "refresh_sha256": refresh_digest}: + raise DurableChangeTrainError("source continuity refresh receipt collision") + else: + encoded = ( + json.dumps({**payload, "refresh_sha256": refresh_digest}, indent=2, sort_keys=True) + "\n" + ).encode("utf-8") + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + dir=refresh_root, + prefix=f".{refresh_path.name}.", + suffix=".tmp", + delete=False, + ) as stream: + temporary = Path(stream.name) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, refresh_path) + temporary = None + _migration_runner._fsync_manifest_directory(refresh_root) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + refresh_ref = f"proof:source-continuity-refresh:{refresh_digest}" + references = _migration_runner._append_proof_refs(train.proof_refs, evidence_ref, refresh_ref) + if train.proof is None: + raise DurableChangeTrainError("source continuity refresh requires train proof") + updated = replace( + train, + revision=train.revision + 1, + source_continuity_evidence=current, + proof_refs=references, + ) + write_durable_change_train_manifest(manifest_path, updated, expected_revision=train.revision) + return refresh_path + + def _fresh_ddl_parity_for_train( train: DurableChangeTrain, *, @@ -702,7 +1380,7 @@ def _verify_persisted_live_tier_continuity(conn: sqlite3.Connection, train: Dura ) from exc -def _verify_released_train_live_tier(conn: sqlite3.Connection, train: DurableChangeTrain) -> None: +def _verify_released_train_live_tier(archive_root: Path, conn: sqlite3.Connection, train: DurableChangeTrain) -> None: """Verify a released train remains represented after later trains advance it.""" if train.apply_evidence is None: raise DurableChangeTrainError(f"{train.state.value} train lacks post-apply continuity evidence") @@ -713,7 +1391,15 @@ def _verify_released_train_live_tier(conn: sqlite3.Connection, train: DurableCha "target; refusing startup initialization" ) if actual.user_version == train.target_version: - _verify_persisted_live_tier_continuity(conn, train) + if train.source_continuity_evidence is not None: + _validate_source_continuity_refresh_receipt(archive_root, train) + _assert_durable_database_continuity( + actual, + train.source_continuity_evidence, + label="source continuity refresh", + ) + else: + _verify_persisted_live_tier_continuity(conn, train) return integrity = conn.execute("PRAGMA integrity_check").fetchone() if integrity != ("ok",): @@ -850,7 +1536,7 @@ def execute_durable_change_train( f"released {tier.value} train {train.train_id} expects live v{runtime_target_version}, " f"found v{live_version}; authorize a new execution" ) - _verify_persisted_live_tier_continuity(live, train) + _verify_released_train_live_tier(archive_root, live, train) return DurableChangeTrainExecution(train=train, manifest_path=manifest_path, migration_result=None) if train.state is DurableChangeTrainState.DECLARED: @@ -945,6 +1631,7 @@ def reconcile_durable_change_train_startup(archive_root: Path) -> tuple[Path, .. def _reconcile_durable_change_train_startup_locked(archive_root: Path) -> tuple[Path, ...]: """Reconcile persisted trains while the caller holds archive ownership.""" + _recover_pending_source_continuity_intents(archive_root) manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" if not manifest_root.is_dir(): return () @@ -980,7 +1667,7 @@ def _reconcile_durable_change_train_startup_locked(archive_root: Path) -> tuple[ train = _persist_train_transition(manifest_path, recovered, expected_revision=train.revision) if train.state is DurableChangeTrainState.RELEASED: with _open_existing_tier(archive_root / f"{train.tier.value}.db") as live: - _verify_released_train_live_tier(live, train) + _verify_released_train_live_tier(archive_root, live, train) reconciled.append(manifest_path) continue if train.state not in { @@ -1020,6 +1707,8 @@ def __getattr__(name: str) -> object: "DurableChangeTrain", "DurableChangeTrainState", "DurableChangeTrainError", + "DurableSourceTrainMissingError", + "DurableSourceContinuitySemanticError", "DurableChangeTrainApplyError", "DurableChangeTrainRecoveryError", "DurableMigrationClaim", @@ -1040,6 +1729,11 @@ def __getattr__(name: str) -> object: "record_durable_writer_release", "prove_durable_change_train", "release_durable_change_train", + "assert_source_continuity_apply_allowed", + "write_source_continuity_pending_intent", + "mark_source_continuity_pending_intent_terminal", + "clear_source_continuity_pending_intent", + "refresh_released_source_train_continuity", "write_durable_change_train_manifest", "load_durable_change_train_manifest", ] diff --git a/polylogue/storage/sqlite/migration_runner.py b/polylogue/storage/sqlite/migration_runner.py index 6ecbeccbb6..decd9e3a0d 100644 --- a/polylogue/storage/sqlite/migration_runner.py +++ b/polylogue/storage/sqlite/migration_runner.py @@ -1367,6 +1367,7 @@ class DurableChangeTrain: released_at_ms: int | None release_evidence_ref: str | None proof_refs: tuple[str, ...] + source_continuity_evidence: DurableDatabaseEvidence | None = None @property def contention_key(self) -> tuple[str, int, int]: @@ -1407,6 +1408,7 @@ def _normalize_schema_sql(sql: str | None) -> str: if sql is None: return "" unquoted: list[str] = [] + string_literals: list[str] = [] index = 0 while index < len(sql): character = sql[index] @@ -1423,7 +1425,8 @@ def _normalize_schema_sql(sql: str | None) -> str: continue index += 1 break - unquoted.append(sql[start:index]) + string_literals.append(sql[start:index]) + unquoted.append(f"\x00{len(string_literals) - 1}\x00") continue if character in {'"', "`", "["}: closing = "]" if character == "[" else character @@ -1463,6 +1466,8 @@ def _normalize_schema_sql(sql: str | None) -> str: r"CHECK(\g IN(\g))", collapsed, ) + for index, literal in enumerate(string_literals): + collapsed = collapsed.replace(f"\x00{index}\x00", literal) return collapsed @@ -2900,6 +2905,10 @@ def validate_durable_change_train_manifest(train: DurableChangeTrain) -> None: """Validate cross-field lifecycle invariants for loaded and transitioned manifests.""" if train.manifest_format != DURABLE_CHANGE_TRAIN_FORMAT: raise DurableChangeTrainError(f"unsupported durable change train format: {train.manifest_format}") + if train.source_continuity_evidence is not None and ( + train.tier is not ArchiveTier.SOURCE or train.state is not DurableChangeTrainState.RELEASED + ): + raise DurableChangeTrainError("source continuity evidence is only valid on a released source train") if train.tier not in DURABLE_MIGRATION_TIERS: raise DurableChangeTrainError(f"manifest tier is not durable: {train.tier.value}") if train.current_version < 1 or train.target_version != train.current_version + 1: @@ -3006,6 +3015,7 @@ def validate_durable_change_train_manifest(train: DurableChangeTrain) -> None: raise DurableChangeTrainError("backup-authorized manifest contains invalid later evidence") return _validate_apply_evidence(train) + apply_evidence = train.apply_evidence if train.state is DurableChangeTrainState.APPLIED: if train.proof is not None or train.released_at_ms is not None or train.release_evidence_ref is not None: raise DurableChangeTrainError("applied manifest contains proof/release evidence") @@ -3030,6 +3040,26 @@ def validate_durable_change_train_manifest(train: DurableChangeTrain) -> None: raise DurableChangeTrainError("train release timestamp predates proof") if release_ref not in train.proof_refs: raise DurableChangeTrainError("train release evidence is not retained by the manifest") + if apply_evidence is None: + raise DurableChangeTrainError("released manifest lacks apply evidence") + if train.source_continuity_evidence is not None: + _validate_database_evidence( + train.source_continuity_evidence, + train, + expected_version=train.target_version, + label="source continuity evidence", + ) + apply_post = apply_evidence.post + refreshed = train.source_continuity_evidence + if ( + refreshed.schema_inventory_sha256 != apply_post.schema_inventory_sha256 + or refreshed.archive_identity_digest != apply_post.archive_identity_digest + ): + raise DurableChangeTrainError("source continuity evidence changed schema or archive identity") + if refreshed.observed_at_ms < train.released_at_ms: + raise DurableChangeTrainError("source continuity evidence predates train release") + if not any(ref.startswith("proof:source-continuity-refresh:") for ref in train.proof_refs): + raise DurableChangeTrainError("source continuity evidence is not retained by the train") return raise DurableChangeTrainError(f"unknown durable change train state: {train.state}") @@ -3197,6 +3227,9 @@ def durable_change_train_from_payload(payload: Mapping[str, object]) -> DurableC checksum = mutable.pop("manifest_sha256", None) if not isinstance(checksum, str) or checksum != _canonical_json_sha256(mutable): raise DurableChangeTrainError("durable change train manifest checksum mismatch") + # v1 manifests written before source continuity refreshes omitted this + # optional field. Preserve their checksum and decode them as no refresh. + mutable.setdefault("source_continuity_evidence", None) decoded = _decode_manifest_value(DurableChangeTrain, mutable, label="train") if not isinstance(decoded, DurableChangeTrain): raise DurableChangeTrainError("durable change train payload decoded to the wrong type") diff --git a/tests/unit/storage/test_blob_ref_liveness.py b/tests/unit/storage/test_blob_ref_liveness.py index 9a2a94bcd7..4cfeeb25fc 100644 --- a/tests/unit/storage/test_blob_ref_liveness.py +++ b/tests/unit/storage/test_blob_ref_liveness.py @@ -4,6 +4,7 @@ import json import sqlite3 +from importlib import resources from pathlib import Path import pytest @@ -26,7 +27,9 @@ from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from polylogue.storage.sqlite.archive_tiers.source_write import deterministic_blob_hash, deterministic_raw_session_id from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.durable_change_train import DurableSourceContinuitySemanticError from polylogue.storage.sqlite.migration_runner import ( + DurableChangeTrainError, MigrationError, validate_migration_backup_live_fingerprint, validate_migration_backup_manifest, @@ -152,6 +155,21 @@ def test_dry_run_is_read_only_and_reports_attachment_parent_join(tmp_path: Path) assert after == before +def test_default_dry_run_does_not_acquire_the_archive_writer_lease( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = _source_archive(tmp_path) + + def unexpected_lease(*args: object, **kwargs: object) -> object: + raise AssertionError("default dry-run must not acquire the archive writer lease") + + monkeypatch.setattr("polylogue.storage.archive_identity.OwnedArchiveLocation.acquire", unexpected_lease) + + report = reconcile_blob_ref_liveness(archive_root) + + assert report.dry_run is True + + def test_legacy_hook_payload_ref_is_rekeyable_not_a_delete_candidate(tmp_path: Path) -> None: archive_root = _source_archive(tmp_path) blob_hash = b"h" * 32 @@ -189,6 +207,51 @@ def test_apply_requires_backup_and_receipt_before_mutation(tmp_path: Path) -> No assert conn.execute("SELECT COUNT(*) FROM blob_refs").fetchone() == (8,) +def test_apply_refuses_a_fresh_mutation_while_continuity_recovery_is_pending(tmp_path: Path) -> None: + archive_root = _source_archive(tmp_path) + pending_root = archive_root / ".maintenance-state" / "source-continuity-pending" + pending_root.mkdir(parents=True) + (pending_root / "pending.json").write_text("{}\n", encoding="utf-8") + receipt = tmp_path / "receipts" / "fresh.jsonl" + + with pytest.raises(BlobRefLivenessReconciliationError, match="continuity recovery is pending"): + reconcile_blob_ref_liveness( + archive_root, + backup_manifest=tmp_path / "backup.json", + receipt_path=receipt, + dry_run=False, + ) + + assert not receipt.exists() + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM blob_refs").fetchone() == (8,) + + +def test_apply_refuses_an_unreleased_source_train_before_writing_a_receipt(tmp_path: Path) -> None: + archive_root = _source_archive(tmp_path) + manifest_root = archive_root / ".maintenance-state" / "durable-change-trains" + manifest_root.mkdir(parents=True) + manifest_root.joinpath("source-029.json").write_text( + resources.files("polylogue.storage.sqlite.migrations.source") + .joinpath("029.train.json") + .read_text(encoding="utf-8"), + encoding="utf-8", + ) + receipt = tmp_path / "receipts" / "blocked-by-train.jsonl" + + with pytest.raises(BlobRefLivenessReconciliationError, match="unreleased source train"): + reconcile_blob_ref_liveness( + archive_root, + backup_manifest=tmp_path / "backup.json", + receipt_path=receipt, + dry_run=False, + ) + + assert not receipt.exists() + with sqlite3.connect(archive_root / "source.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM blob_refs").fetchone() == (8,) + + def test_apply_refuses_a_running_daemon_before_receipt_or_blob_ref_mutation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -260,6 +323,27 @@ def test_restart_recovers_prepared_receipt_after_committed_delete(tmp_path: Path assert rows[-1]["phase"] == "recovered_committed" +def test_prepared_receipt_persists_new_receipt_parent_entry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + archive_root = _source_archive(tmp_path) + receipt = tmp_path / "receipts" / "fresh.jsonl" + with sqlite3.connect(archive_root / "source.db") as conn: + classification = classify_blob_ref_liveness(conn) + fsync_calls: list[Path] = [] + real_fsync_directory = liveness_reconciliation._fsync_directory + + def record_fsync_directory(path: Path) -> None: + fsync_calls.append(path) + real_fsync_directory(path) + + monkeypatch.setattr(liveness_reconciliation, "_fsync_directory", record_fsync_directory) + liveness_reconciliation._write_prepared_receipt( + receipt, archive_root / "source.db", classification, tmp_path / "backup.json" + ) + + assert receipt.parent in fsync_calls + assert receipt.parent.parent in fsync_calls + + def test_restart_recovers_prepared_receipt_after_rollback(tmp_path: Path) -> None: archive_root = _source_archive(tmp_path) receipt = tmp_path / "receipts" / "rolled-back.jsonl" @@ -983,6 +1067,86 @@ def fake_validate(path: Path, tier: object, *, connection: sqlite3.Connection) - assert conn.execute("SELECT COUNT(*) FROM blob_refs").fetchone() == (8,) +def test_committed_delete_retains_pending_intent_for_retryable_refresh_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = _source_archive(tmp_path) + + def fake_validate(path: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return path + + monkeypatch.setattr( + "polylogue.maintenance.blob_ref_liveness_reconciliation.validate_migration_backup_manifest", + fake_validate, + ) + monkeypatch.setattr( + "polylogue.maintenance.blob_ref_liveness_reconciliation.validate_migration_backup_live_fingerprint", + fake_validate, + ) + monkeypatch.setattr( + "polylogue.maintenance.blob_ref_liveness_reconciliation.running_daemon_pid", lambda _config: None + ) + monkeypatch.setattr( + liveness_reconciliation, + "refresh_released_source_train_continuity", + lambda *args, **kwargs: (_ for _ in ()).throw(DurableChangeTrainError("continuity unavailable")), + ) + + backup_manifest = tmp_path / "backup" / "manifest.json" + backup_manifest.parent.mkdir() + backup_manifest.write_text("{}\n", encoding="utf-8") + receipt = tmp_path / "receipts" / "continuity-failed.jsonl" + report = reconcile_blob_ref_liveness( + archive_root, + backup_manifest=backup_manifest, + receipt_path=receipt, + dry_run=False, + ) + + assert report.applied is True + assert report.deleted_count == 4 + assert report.continuity_refresh_receipt is None + assert report.continuity_refresh_error == "continuity unavailable" + assert json.loads(receipt.read_text(encoding="utf-8").splitlines()[-1])["phase"] == "committed" + pending = next((archive_root / ".maintenance-state" / "source-continuity-pending").glob("*.json")) + assert "terminal_outcome" not in json.loads(pending.read_text(encoding="utf-8")) + + +def test_committed_delete_terminalizes_an_immutable_continuity_mismatch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = _source_archive(tmp_path) + + def fake_validate(path: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return path + + monkeypatch.setattr(liveness_reconciliation, "validate_migration_backup_manifest", fake_validate) + monkeypatch.setattr(liveness_reconciliation, "validate_migration_backup_live_fingerprint", fake_validate) + monkeypatch.setattr(liveness_reconciliation, "running_daemon_pid", lambda _config: None) + monkeypatch.setattr( + liveness_reconciliation, + "refresh_released_source_train_continuity", + lambda *args, **kwargs: (_ for _ in ()).throw( + DurableSourceContinuitySemanticError("continuity baseline rejected") + ), + ) + backup_manifest = tmp_path / "backup" / "manifest.json" + backup_manifest.parent.mkdir() + backup_manifest.write_text("{}\n", encoding="utf-8") + reconcile_blob_ref_liveness( + archive_root, + backup_manifest=backup_manifest, + receipt_path=tmp_path / "receipts" / "continuity-semantic.jsonl", + dry_run=False, + ) + + pending = next((archive_root / ".maintenance-state" / "source-continuity-pending").glob("*.json")) + assert json.loads(pending.read_text(encoding="utf-8"))["terminal_outcome"] == { + "kind": "continuity_refresh_rejected", + "error": "continuity baseline rejected", + } + + def test_shared_legacy_hook_path_fails_closed_without_cross_product( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/storage/test_durable_change_train.py b/tests/unit/storage/test_durable_change_train.py index 61097d217e..1a1a506d88 100644 --- a/tests/unit/storage/test_durable_change_train.py +++ b/tests/unit/storage/test_durable_change_train.py @@ -2,29 +2,39 @@ from __future__ import annotations +import hashlib import json import os import sqlite3 import sys -from collections.abc import Iterator +from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import replace from pathlib import Path +from types import SimpleNamespace from typing import cast import pytest +import polylogue.storage.sqlite.durable_change_train as durable_change_train_module from polylogue.storage.sqlite import migration_runner from polylogue.storage.sqlite.archive_tiers import ARCHIVE_DDL_BY_TIER, ARCHIVE_VERSION_BY_TIER from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.durable_change_train import ( DURABLE_MIGRATION_ADOPTION_FLOORS, + DurableSourceContinuitySemanticError, + DurableSourceTrainMissingError, _runtime_consumer_results, + assert_source_continuity_apply_allowed, durable_change_train_manifest_path, durable_change_train_policy_report, durable_migration_sidecar_for_slot, execute_durable_change_train, + mark_source_continuity_pending_intent_terminal, + reconcile_durable_change_train_startup, + refresh_released_source_train_continuity, validate_durable_migration_sidecars, + write_source_continuity_pending_intent, ) from polylogue.storage.sqlite.migration_runner import ( DurableChangeRider, @@ -60,6 +70,7 @@ _CURRENT_VERSION = 1 _TARGET_VERSION = 2 +_EMPTY_LIVENESS_DIGEST = hashlib.sha256(b"[]").hexdigest() _ADDITIVE_SQL = """-- migration-safety: additive-no-backup CREATE TABLE durable_items ( item_id TEXT PRIMARY KEY, @@ -354,12 +365,346 @@ def test_applied_train_release_requires_the_source_hook_event_writer_probe( assert released.state is DurableChangeTrainState.RELEASED assert released.proof is not None + assert released.apply_evidence is not None hook_writer = next( result for result in released.proof.runtime_consumers if result.consumer_id == "source-hook-event-writer" ) assert hook_writer.passed is True +def test_released_source_train_can_record_an_authorized_mutation_refresh( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + db_path = tmp_path / "source.db" + _create_current_database(db_path) + _install_synthetic_migration(tmp_path, monkeypatch, ArchiveTier.SOURCE) + train = _admitted(ArchiveTier.SOURCE) + with sqlite3.connect(db_path) as conn: + train = _reserve_and_authorize(conn, train, archive_root=tmp_path) + train = apply_durable_change_train(conn, train) + train = record_durable_writer_release(train, evidence_ref="proof:writer-release") + with sqlite3.connect(db_path) as conn: + runtime_results = _runtime_results() + restart = capture_durable_restart_convergence( + conn, + train, + runtime_consumers=runtime_results, + evidence_ref="proof:restart", + ) + train = prove_durable_change_train( + train, + fresh_ddl_parity=_parity(ArchiveTier.SOURCE), + runtime_consumers=runtime_results, + restart_convergence=restart, + ) + train = release_durable_change_train(train, evidence_ref="proof:release") + manifest = tmp_path / ".maintenance-state" / "durable-change-trains" / "source-002.json" + manifest.parent.mkdir(parents=True) + write_durable_change_train_manifest(manifest, train, expected_revision=-1) + released = load_durable_change_train_manifest(manifest) + with sqlite3.connect(db_path) as conn: + before = migration_runner.capture_durable_database_evidence(conn, ArchiveTier.SOURCE) + conn.execute("INSERT INTO base_items VALUES ('mutation-1', 'authorized')") + conn.commit() + + backup_manifest = tmp_path / "backup-manifest.json" + backup_manifest.write_text("{}\n", encoding="utf-8") + mutation_receipt = tmp_path / "mutation-receipt.jsonl" + mutation_receipt.write_text( + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "prepared", + "source_db": str(db_path), + "backup_manifest": str(backup_manifest), + "candidate_count": 0, + "candidate_digest": _EMPTY_LIVENESS_DIGEST, + "backup_manifest_sha256": hashlib.sha256(backup_manifest.read_bytes()).hexdigest(), + } + ) + + "\n" + + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "committed", + "deleted_count": 0, + "post_orphaned_count": 0, + } + ) + + "\n", + encoding="utf-8", + ) + refresh_fsync_calls: list[Path] = [] + real_fsync_manifest_directory = migration_runner._fsync_manifest_directory + + def record_refresh_fsync(path: Path) -> None: + refresh_fsync_calls.append(path) + real_fsync_manifest_directory(path) + + monkeypatch.setattr(migration_runner, "_fsync_manifest_directory", record_refresh_fsync) + refreshed_path = refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-1", + ) + assert tmp_path / ".maintenance-state" in refresh_fsync_calls + + refreshed = load_durable_change_train_manifest(manifest) + assert refreshed.state is DurableChangeTrainState.RELEASED + assert refreshed.source_continuity_evidence is not None + assert released.apply_evidence is not None + assert refreshed.apply_evidence == released.apply_evidence + assert refreshed.proof == released.proof + assert refreshed.revision == released.revision + 1 + assert any(ref.startswith("proof:source-continuity-refresh:") for ref in refreshed.proof_refs) + assert refreshed.source_continuity_evidence.content_sha256 != released.apply_evidence.post.content_sha256 + assert refreshed_path.is_file() + # A pending-intent retry after manifest persistence accepts only the + # receipt's exact pre/post evidence, rather than bypassing pre-state + # authentication because the receipt digest already exists. + assert ( + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-1", + ) + == refreshed_path + ) + with pytest.raises(DurableSourceContinuitySemanticError, match="unreceipted content drift"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=replace(before, content_sha256="f" * 64), + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-1", + ) + operator_cwd = tmp_path / "operator-cwd" + operator_cwd.mkdir() + monkeypatch.chdir(tmp_path) + pending_fsync_start = len(refresh_fsync_calls) + pending_path = write_source_continuity_pending_intent( + tmp_path, + mutation_receipt=Path("mutation-receipt.jsonl"), + backup_manifest=Path("backup-manifest.json"), + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-1", + ) + assert pending_path.is_file() + assert tmp_path / ".maintenance-state" in refresh_fsync_calls[pending_fsync_start:] + pending_payload = json.loads(pending_path.read_text(encoding="utf-8")) + assert pending_payload["mutation_receipt"] == str(mutation_receipt) + assert pending_payload["backup_manifest"] == str(backup_manifest) + monkeypatch.chdir(operator_cwd) + assert reconcile_durable_change_train_startup(tmp_path) == (manifest,) + assert not pending_path.exists() + terminal_pending_path = write_source_continuity_pending_intent( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-1", + ) + mark_source_continuity_pending_intent_terminal( + terminal_pending_path, + error=DurableSourceContinuitySemanticError("continuity precondition rejected"), + ) + assert reconcile_durable_change_train_startup(tmp_path) == (manifest,) + assert not terminal_pending_path.exists() + refreshed_path.unlink() + with pytest.raises(DurableChangeTrainError, match="refresh receipt"): + reconcile_durable_change_train_startup(tmp_path) + with pytest.raises(DurableChangeTrainError, match="refresh receipt"): + assert_source_continuity_apply_allowed(tmp_path) + + mutation_receipt.write_text( + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "prepared", + "source_db": str(db_path), + "backup_manifest": str(backup_manifest), + "candidate_count": 0, + "candidate_digest": _EMPTY_LIVENESS_DIGEST, + "backup_manifest_sha256": hashlib.sha256(backup_manifest.read_bytes()).hexdigest(), + } + ) + + "\n" + + json.dumps({"kind": "blob_ref_liveness_reconciliation", "phase": "prepared"}) + + "\n", + encoding="utf-8", + ) + with pytest.raises(DurableChangeTrainError, match="does not bind"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-invalid-footer", + ) + + mutation_receipt.write_text('{"kind": "blob_ref_liveness_reconciliation"}\n', encoding="utf-8") + with pytest.raises(DurableChangeTrainError, match="incomplete"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-incomplete", + ) + + mutation_receipt.write_text("not-json\n", encoding="utf-8") + with pytest.raises(DurableChangeTrainError, match="valid JSONL"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-malformed", + ) + + mutation_receipt.write_text( + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "prepared", + "source_db": str(db_path), + "backup_manifest": str(backup_manifest), + "candidate_count": 0, + "candidate_digest": "b" * 64, + "backup_manifest_sha256": hashlib.sha256(backup_manifest.read_bytes()).hexdigest(), + } + ) + + "\n" + + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "committed", + "deleted_count": 0, + "post_orphaned_count": 0, + } + ) + + "\n", + encoding="utf-8", + ) + with pytest.raises(DurableChangeTrainError, match="does not bind"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-wrong-operation", + ) + + manifest.unlink() + mutation_receipt.write_text( + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "prepared", + "source_db": str(db_path), + "backup_manifest": str(backup_manifest), + "candidate_count": 0, + "candidate_digest": _EMPTY_LIVENESS_DIGEST, + "backup_manifest_sha256": hashlib.sha256(backup_manifest.read_bytes()).hexdigest(), + } + ) + + "\n" + + json.dumps( + { + "kind": "blob_ref_liveness_reconciliation", + "phase": "committed", + "deleted_count": 0, + "post_orphaned_count": 0, + } + ) + + "\n", + encoding="utf-8", + ) + with pytest.raises(DurableSourceTrainMissingError, match="no released source train"): + refresh_released_source_train_continuity( + tmp_path, + mutation_receipt=mutation_receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id=_EMPTY_LIVENESS_DIGEST, + evidence_ref="proof:mutation-no-train", + ) + + +def test_startup_consumes_an_already_recovered_rollback_intent(tmp_path: Path) -> None: + db_path = tmp_path / "source.db" + _create_current_database(db_path) + with sqlite3.connect(db_path) as connection: + before = migration_runner.capture_durable_database_evidence(connection, ArchiveTier.SOURCE) + receipt = tmp_path / "rolled-back.jsonl" + receipt.write_text('{"phase": "recovered_rolled_back"}\n', encoding="utf-8") + backup_manifest = tmp_path / "backup-manifest.json" + backup_manifest.write_text("{}\n", encoding="utf-8") + pending_path = write_source_continuity_pending_intent( + tmp_path, + mutation_receipt=receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id="rolled-back-operation", + evidence_ref="proof:rolled-back-operation", + ) + + assert reconcile_durable_change_train_startup(tmp_path) == () + assert not pending_path.exists() + + +def test_postcondition_recovery_rejects_remaining_orphans(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + db_path = tmp_path / "source.db" + _create_current_database(db_path) + with sqlite3.connect(db_path) as connection: + before = migration_runner.capture_durable_database_evidence(connection, ArchiveTier.SOURCE) + receipt = tmp_path / "postcondition-failed.jsonl" + receipt.write_text('{"phase": "postcondition_failed"}\n', encoding="utf-8") + backup_manifest = tmp_path / "backup-manifest.json" + backup_manifest.write_text("{}\n", encoding="utf-8") + pending_path = write_source_continuity_pending_intent( + tmp_path, + mutation_receipt=receipt, + backup_manifest=backup_manifest, + pre_mutation_evidence=before, + operation_id="postcondition-failed-operation", + evidence_ref="proof:postcondition-failed-operation", + ) + + def recover_with_postcondition_check(*_args: object, **kwargs: object) -> str: + postcondition_check = cast(Callable[[], None], kwargs["postcondition_check"]) + postcondition_check() + return "recovered_committed" + + monkeypatch.setattr( + "polylogue.maintenance.blob_ref_liveness_reconciliation._recover_prepared_receipt", + recover_with_postcondition_check, + ) + monkeypatch.setattr( + "polylogue.storage.blob_ref_liveness.classify_blob_ref_liveness", + lambda _connection: SimpleNamespace(safe_to_apply=True, orphaned_count=1), + ) + + with pytest.raises(DurableChangeTrainError, match="postcondition remains unsafe"): + durable_change_train_module._recover_pending_source_continuity_intents(tmp_path) + + assert pending_path.exists() + + @pytest.mark.parametrize("tier", (ArchiveTier.SOURCE, ArchiveTier.USER)) def test_synthetic_source_and_user_trains_complete_the_full_lifecycle( tmp_path: Path, @@ -1178,6 +1523,12 @@ def test_failed_transaction_exposes_exact_retry_recovery( failure_manifest = tmp_path / "source-failed-train.json" write_durable_change_train_manifest(failure_manifest, failed, expected_revision=-1) failed = load_durable_change_train_manifest(failure_manifest) + released_failed = record_durable_writer_release(failed, evidence_ref="proof:failed-writer-release") + released_manifest = tmp_path / ".maintenance-state" / "durable-change-trains" / "source-002.json" + released_manifest.parent.mkdir(parents=True) + write_durable_change_train_manifest(released_manifest, released_failed, expected_revision=-1) + with pytest.raises(DurableChangeTrainError, match="unreleased source train"): + assert_source_continuity_apply_allowed(tmp_path) assert int(conn.execute("PRAGMA user_version").fetchone()[0]) == _CURRENT_VERSION assert conn.execute("SELECT name FROM sqlite_schema WHERE name='durable_items'").fetchone() is None recovered = recover_durable_change_train(