diff --git a/docs/data-model.md b/docs/data-model.md index 6c9bfbc565..2d5e65cf8b 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -72,7 +72,7 @@ Convenience properties resolve these: | Field | Type | Description | |-------|------|-------------| -| `id` | `str` | Message ID, `session_id:native_id` (or `session_id:position.variant`) | +| `id` | `str` | Message ID, `session_id:n:native_id` (or `session_id:p:position.variant`) | | `role` | `Role` | `user`, `assistant`, `system`, `tool`, `unknown` | | `text` | `str?` | Flattened message text | | `timestamp` | `datetime?` | Message timestamp | diff --git a/docs/examples/demo-tour/command-output/01-claim-versus-receipt.txt b/docs/examples/demo-tour/command-output/01-claim-versus-receipt.txt index 629dfa862f..613b302b53 100644 --- a/docs/examples/demo-tour/command-output/01-claim-versus-receipt.txt +++ b/docs/examples/demo-tour/command-output/01-claim-versus-receipt.txt @@ -3,21 +3,21 @@ archive: verdict: contradicted_at_claim_time_then_repaired assistant claim: All tests pass. The clock fix is complete. -claim evidence: block:codex-session:demo-receipts:receipts-a-claim:0 +claim evidence: block:codex-session:demo-receipts:n:receipts-a-claim:0 at claim time: tool: shell (exec_command) command: pytest tests/test_clock.py -q exit: 1 (failed=true) result: {"metadata": {"exit_code": 1}, "output": "F tests/test_clock.py::test_uses_monotonic_clock\n1 failed in 0.18s"} - evidence: block:codex-session:demo-receipts:call-receipts-test-fail:0 + evidence: block:codex-session:demo-receipts:n:call-receipts-test-fail:0 later recovery: tool: shell (exec_command) command: pytest tests/test_clock.py -q exit: 0 (failed=false) result: {"metadata": {"exit_code": 0}, "output": ". 1 passed in 0.16s"} - evidence: block:codex-session:demo-receipts:call-receipts-test-pass:0 + evidence: block:codex-session:demo-receipts:n:call-receipts-test-pass:0 anti-grep control: prose hits for 'error': 2 @@ -29,7 +29,7 @@ source material: blob_sha256: 9fd0dbdb080058070935924534a903cc63a8dcba571f6b2734f92a96576b59d7 completion-claim experiment: - sample manifest: ffbb3c4609d488d25510923848188422bb098ef700fd862e44e4315e14b473f6 + sample manifest: a7a39bdcced3d950886d5b555d2ceec7b5f776d961fa726a76946c857f863d13 denominator: 2 unsupported by structural evidence: 0 (0.0%) neutral prior outcome: 0 (0.0%) diff --git a/docs/examples/demo-tour/command-output/03-composed-lineage.txt b/docs/examples/demo-tour/command-output/03-composed-lineage.txt index 365203fd9e..db27303bb4 100644 --- a/docs/examples/demo-tour/command-output/03-composed-lineage.txt +++ b/docs/examples/demo-tour/command-output/03-composed-lineage.txt @@ -18,25 +18,25 @@ Map the demo lineage base context. -`codex-session:demo-lineage-parent:parent-u0` +`codex-session:demo-lineage-parent:n:parent-u0` ### 2026-07-04T10:00:02+00:00 - assistant / message I have the base context and can branch the analysis. -`codex-session:demo-lineage-parent:parent-a1` +`codex-session:demo-lineage-parent:n:parent-a1` ### 2026-07-04T10:01:03+00:00 - user / message Now take the forked branch and audit construct validity. -`codex-session:demo-lineage-fork:fork-u2` +`codex-session:demo-lineage-fork:n:fork-u2` ### 2026-07-04T10:01:04+00:00 - assistant / message The fork diverges into demo corpus construct checks. -`codex-session:demo-lineage-fork:fork-a3` +`codex-session:demo-lineage-fork:n:fork-a3` ### Last Messages diff --git a/docs/examples/demo-tour/report.json b/docs/examples/demo-tour/report.json index f0035ba397..d63167c127 100644 --- a/docs/examples/demo-tour/report.json +++ b/docs/examples/demo-tour/report.json @@ -28,7 +28,7 @@ "result": "pass", "triggered": false }, - "first_result_s": 2.065, + "first_result_s": 1.362, "non_claims": [ "The deterministic tour does not establish field prevalence, production scale, or provider completeness.", "The deterministic tour does not establish memory uplift, invoice accuracy, selective deletion, or the Sinex backend.", @@ -379,13 +379,13 @@ }, "steps": [ { - "bytes_written": 1399, + "bytes_written": 1405, "command": [ "polylogue", "demo", "receipts" ], - "duration_s": 2.064, + "duration_s": 1.362, "exit_code": 0, "name": "claim versus receipt", "output_path": "command-output/01-claim-versus-receipt.txt" @@ -396,13 +396,13 @@ "polylogue", "actions where is_error:true | group by tool | count" ], - "duration_s": 2.822, + "duration_s": 1.839, "exit_code": 0, "name": "failed actions aggregate", "output_path": "command-output/02-failed-actions-aggregate.txt" }, { - "bytes_written": 936, + "bytes_written": 944, "command": [ "polylogue", "--id", @@ -411,7 +411,7 @@ "--view", "chronicle" ], - "duration_s": 2.485, + "duration_s": 1.695, "exit_code": 0, "name": "composed lineage", "output_path": "command-output/03-composed-lineage.txt" @@ -423,13 +423,13 @@ "analyze", "--facets" ], - "duration_s": 2.435, + "duration_s": 1.629, "exit_code": 0, "name": "archive facets", "output_path": "command-output/04-archive-facets.txt" } ], - "total_duration_s": 15.747, + "total_duration_s": 27.096, "transcript_path": "transcript.txt", "verify": { "absolute_path_leaks": [], diff --git a/docs/examples/demo-tour/report.md b/docs/examples/demo-tour/report.md index 5660061872..360ae26834 100644 --- a/docs/examples/demo-tour/report.md +++ b/docs/examples/demo-tour/report.md @@ -40,8 +40,8 @@ The semantic fixture verifier runs before the narrated commands and checks plant ## Timings -- First evidence result: 2.065s (budget 30s) -- Full tour: 15.747s (budget 420s) +- First evidence result: 1.362s (budget 30s) +- Full tour: 27.096s (budget 420s) ## Archive @@ -55,10 +55,10 @@ The semantic fixture verifier runs before the narrated commands and checks plant | Step | Exit | Duration | Bytes | Output | | --- | ---: | ---: | ---: | --- | -| claim versus receipt | 0 | 2.064s | 1399 | `command-output/01-claim-versus-receipt.txt` | -| failed actions aggregate | 0 | 2.822s | 62 | `command-output/02-failed-actions-aggregate.txt` | -| composed lineage | 0 | 2.485s | 936 | `command-output/03-composed-lineage.txt` | -| archive facets | 0 | 2.435s | 1652 | `command-output/04-archive-facets.txt` | +| claim versus receipt | 0 | 1.362s | 1405 | `command-output/01-claim-versus-receipt.txt` | +| failed actions aggregate | 0 | 1.839s | 62 | `command-output/02-failed-actions-aggregate.txt` | +| composed lineage | 0 | 1.695s | 944 | `command-output/03-composed-lineage.txt` | +| archive facets | 0 | 1.629s | 1652 | `command-output/04-archive-facets.txt` | ## Problems diff --git a/docs/examples/demo-tour/transcript.txt b/docs/examples/demo-tour/transcript.txt index 52e472a2bb..258c8be69d 100644 --- a/docs/examples/demo-tour/transcript.txt +++ b/docs/examples/demo-tour/transcript.txt @@ -1,32 +1,32 @@ # prepare deterministic proof archive -Seeded 19 sessions, 71 messages, and 8 user-state assertions in 12.070s. Fixture audit: 40/40 declared constructs satisfied. +Seeded 19 sessions, 71 messages, and 8 user-state assertions in 20.556s. Fixture audit: 40/40 declared constructs satisfied. # verify evidence before presenting it -Verification passed in 0.017s; 0 path leaks and 0 semantic problems. The complete fixture and verification audit remains in report.json. +Verification passed in 0.013s; 0 path leaks and 0 semantic problems. The complete fixture and verification audit remains in report.json. $ polylogue demo receipts Start with a falsifiable disagreement: assistant prose claims the tests pass, while the provider-normalized tool result says exit 1. A later run repairs the result, and a prose-only 'error' control demonstrates why keyword matching is not the oracle. -exit=0 duration=2.267s bytes=1399 +exit=0 duration=1.362s bytes=1405 Polylogue evidence receipt archive: verdict: contradicted_at_claim_time_then_repaired assistant claim: All tests pass. The clock fix is complete. -claim evidence: block:codex-session:demo-receipts:receipts-a-claim:0 +claim evidence: block:codex-session:demo-receipts:n:receipts-a-claim:0 at claim time: tool: shell (exec_command) command: pytest tests/test_clock.py -q exit: 1 (failed=true) result: {"metadata": {"exit_code": 1}, "output": "F tests/test_clock.py::test_uses_monotonic_clock\n1 failed in 0.18s"} - evidence: block:codex-session:demo-receipts:call-receipts-test-fail:0 + evidence: block:codex-session:demo-receipts:n:call-receipts-test-fail:0 later recovery: tool: shell (exec_command) command: pytest tests/test_clock.py -q exit: 0 (failed=false) result: {"metadata": {"exit_code": 0}, "output": ". 1 passed in 0.16s"} - evidence: block:codex-session:demo-receipts:call-receipts-test-pass:0 + evidence: block:codex-session:demo-receipts:n:call-receipts-test-pass:0 anti-grep control: prose hits for 'error': 2 @@ -38,7 +38,7 @@ source material: blob_sha256: 9fd0dbdb080058070935924534a903cc63a8dcba571f6b2734f92a96576b59d7 completion-claim experiment: - sample manifest: ffbb3c4609d488d25510923848188422bb098ef700fd862e44e4315e14b473f6 + sample manifest: a7a39bdcced3d950886d5b555d2ceec7b5f776d961fa726a76946c857f863d13 denominator: 2 unsupported by structural evidence: 0 (0.0%) neutral prior outcome: 0 (0.0%) @@ -47,14 +47,14 @@ contradicted without recorded repair: 1 (50.0%) $ polylogue 'actions where is_error:true | group by tool | count' Now aggregate the same structural field across providers. This query counts normalized failed actions; it does not search prose for the word 'error'. -exit=0 duration=3.273s bytes=62 +exit=0 duration=1.839s bytes=62 tool=Bash count=4 tool=exec_command count=2 tool=Edit count=1 $ polylogue --id codex-session:demo-lineage-fork read --view chronicle Read a fork as one logical chronicle: inherited parent messages remain attributable to their origin while the fork contributes only its divergent tail. -exit=0 duration=2.708s bytes=936 +exit=0 duration=1.695s bytes=944 # Session Chronicle - Sessions: 1 @@ -75,25 +75,25 @@ exit=0 duration=2.708s bytes=936 Map the demo lineage base context. -`codex-session:demo-lineage-parent:parent-u0` +`codex-session:demo-lineage-parent:n:parent-u0` ### 2026-07-04T10:00:02+00:00 - assistant / message I have the base context and can branch the analysis. -`codex-session:demo-lineage-parent:parent-a1` +`codex-session:demo-lineage-parent:n:parent-a1` ### 2026-07-04T10:01:03+00:00 - user / message Now take the forked branch and audit construct validity. -`codex-session:demo-lineage-fork:fork-u2` +`codex-session:demo-lineage-fork:n:fork-u2` ### 2026-07-04T10:01:04+00:00 - assistant / message The fork diverges into demo corpus construct checks. -`codex-session:demo-lineage-fork:fork-a3` +`codex-session:demo-lineage-fork:n:fork-a3` ### Last Messages @@ -101,7 +101,7 @@ _No distinct matching prose in the last edge._ $ polylogue analyze --facets Only after inspecting evidence, zoom out to the archive across 8 origins, with deferred families labeled rather than silently guessed. -exit=0 duration=2.555s bytes=1652 +exit=0 duration=1.629s bytes=1652 Facets (global) — matched result set: readiness: ready (cost_class=cheap; budget 0.01s/2.00s) sessions: 19 messages: 71 diff --git a/docs/maintenance.md b/docs/maintenance.md index 7fcea64e81..ec3c3a6e33 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -227,6 +227,31 @@ reclassifies under `BEGIN IMMEDIATE` before fsyncing the prepared receipt and deleting the exact candidate set. Review the receipt's final `committed` line before treating the pass as complete. +### `polylogue ops maintenance blob-reference-closure` - acquired reference closure + +Read-only by default. It checks that each `raw_sessions` row has exactly one +matching `raw_payload` ref and that each acquired index attachment is reachable +through `attachment_refs`. Raw gaps are repaired from the retained raw row's +exact hash, path, size, and acquisition timestamp. Attachment gaps are repaired +only when a complete reparse of authoritative `source.db` bytes reproduces the +attachment identity and its owning message still exists. Other rows are +reported as typed blockers and remain untouched. + +```bash +polylogue ops maintenance blob-reference-closure --output-format json +polylogue ops maintenance blob-reference-closure --apply \ + --backup-manifest /path/to/verified-full-evidence-manifest.json \ + --receipt-file /path/to/new/blob-reference-closure.jsonl \ + --output-format json +``` + +Apply requires the daemon to be offline, a verified backup manifest covering +both `source.db` and `index.db`, and a new receipt path. It inserts exact refs +only, never deletes or replaces existing refs. The source and index commits are +recorded separately in the receipt so a retry can safely continue an additive +repair. Reindex acceptance runs the same closure check against the candidate +index before promotion. + ### `polylogue ops maintenance hook-payload-ref-reconcile` - legacy hook-ref repair Read-only by default. It classifies historical orphaned `raw_payload` refs and diff --git a/polylogue/cli/commands/maintenance/__init__.py b/polylogue/cli/commands/maintenance/__init__.py index 6e23ec38a3..3ddd56970f 100644 --- a/polylogue/cli/commands/maintenance/__init__.py +++ b/polylogue/cli/commands/maintenance/__init__.py @@ -121,6 +121,12 @@ "blob_reference_liveness_command", "Classify source-tier orphan refs; apply only with backup and receipt.", ), + ( + "blob-reference-closure", + "_blob_reference_closure", + "blob_reference_closure_command", + "Repair deterministic raw and acquired-attachment reference gaps; dry-run by default.", + ), ( "hook-payload-ref-reconcile", "_hook_payload_ref_reconciliation", diff --git a/polylogue/cli/commands/maintenance/_blob_reference_closure.py b/polylogue/cli/commands/maintenance/_blob_reference_closure.py new file mode 100644 index 0000000000..71593416e2 --- /dev/null +++ b/polylogue/cli/commands/maintenance/_blob_reference_closure.py @@ -0,0 +1,63 @@ +"""CLI adapter for acquired blob-reference closure repair.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import click + +from polylogue.paths import archive_root + + +@click.command("blob-reference-closure") +@click.option("--apply", "apply_changes", is_flag=True, help="Apply only deterministic exact reference repairs.") +@click.option( + "--backup-manifest", + type=click.Path(dir_okay=False, path_type=Path), + default=None, + help="Verified manifest covering source.db and index.db; required with --apply.", +) +@click.option( + "--receipt-file", + type=click.Path(dir_okay=False, path_type=Path), + default=None, + help="New immutable receipt path; required with --apply.", +) +@click.option("--output-format", type=click.Choice(["plain", "json"]), default="plain", show_default=True) +def blob_reference_closure_command( + apply_changes: bool, + backup_manifest: Path | None, + receipt_file: Path | None, + output_format: str, +) -> None: + """Audit closure, or repair exact refs from existing source evidence.""" + from polylogue.maintenance.blob_reference_closure import ( + BlobReferenceClosureError, + reconcile_blob_reference_closure, + ) + + try: + report = reconcile_blob_reference_closure( + archive_root(), + backup_manifest=backup_manifest, + receipt_path=receipt_file, + dry_run=not apply_changes, + ) + except BlobReferenceClosureError as exc: + raise click.ClickException(str(exc)) from exc + + payload = {"mode": "blob_reference_closure", **report.to_dict()} + if output_format == "json": + click.echo(json.dumps(payload, indent=2, sort_keys=True)) + return + click.echo("Blob-reference closure") + click.echo(f"Mode: {'apply' if report.applied else 'dry-run'}") + click.echo(f"Raw repair: {report.raw_repaired_count:,}") + click.echo(f"Attachment: {report.attachment_repaired_count:,}") + click.echo(f"Blockers: {len(report.plan.blockers):,}") + for blocker in report.plan.blockers: + click.echo(f" blocker [{blocker.kind.value}] {blocker.object_id}: {blocker.detail}") + + +__all__ = ["blob_reference_closure_command"] diff --git a/polylogue/core/identity_law.py b/polylogue/core/identity_law.py index 4a2567b568..e184eb7adb 100644 --- a/polylogue/core/identity_law.py +++ b/polylogue/core/identity_law.py @@ -43,13 +43,14 @@ def message_local_id( ) -> str: """Return the message-local identity component. - Provider-native message IDs win when present. When the provider omits a - native ID, archive falls back to ``position.variant_index`` so sibling - regeneration branches cannot collide. + Provider-native message IDs and position-derived coordinates occupy + disjoint tagged namespaces. This prevents a provider id such as ``0.0`` + from colliding with the positional identity for ``(position=0, + variant_index=0)`` while keeping both components opaque. """ if native_id is not None and native_id.strip(): - return _required_text("message native_id", native_id) - return f"{_required_non_negative('position', position)}.{_required_non_negative('variant_index', variant_index)}" + return f"n:{_required_text('message native_id', native_id)}" + return f"p:{_required_non_negative('position', position)}.{_required_non_negative('variant_index', variant_index)}" def message_id( diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index 848ac8fa87..be29a187e1 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -1063,6 +1063,89 @@ def _check_blob_refs_liveness(archive_root: Path, sample_limit: int) -> ArchiveV ) +def _check_blob_reference_closure_for_index( + archive_root: Path, index_path: Path, sample_limit: int +) -> ArchiveVerificationCheck: + """Require acquired rows to retain their exact canonical references.""" + source_path = _tier_path(archive_root, ArchiveTier.SOURCE) + if not source_path.exists() or not index_path.exists(): + return _skip_check("blob-reference-closure", "source.db or index.db not present") + try: + source_conn = _open_ro(source_path) + except sqlite3.Error as exc: + return _error_check("blob-reference-closure", f"could not open source/index tiers: {exc}", exc=exc) + try: + index_conn = _open_ro(index_path) + except sqlite3.Error as exc: + source_conn.close() + return _error_check("blob-reference-closure", f"could not open source/index tiers: {exc}", exc=exc) + try: + from polylogue.maintenance.blob_reference_closure import ( + closure_counts, + raw_reference_closure_predicate, + ) + + counts = closure_counts(source_conn, index_conn) + raw_sample = [ + str(row[0]) + for row in source_conn.execute( + f""" + SELECT r.raw_id FROM raw_sessions r + WHERE {raw_reference_closure_predicate()} + ORDER BY r.raw_id LIMIT ? + """, + (sample_limit,), + ) + ] + attachment_sample = [ + str(row[0]) + for row in index_conn.execute( + """ + SELECT a.attachment_id FROM attachments a + WHERE a.acquisition_status = 'acquired' + AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id) + ORDER BY a.attachment_id LIMIT ? + """, + (sample_limit,), + ) + ] + except sqlite3.Error as exc: + return _error_check("blob-reference-closure", f"could not read source/index tiers: {exc}", exc=exc) + finally: + index_conn.close() + source_conn.close() + + total = sum(counts.values()) + return ArchiveVerificationCheck( + name="blob-reference-closure", + status=OutcomeStatus.ERROR if total else OutcomeStatus.OK, + summary=( + f"{counts['raw_missing_exact_count']:,} raw session(s) and " + f"{counts['acquired_attachment_missing_ref_count']:,} acquired attachment(s) lack canonical refs" + if total + else "every raw session and acquired attachment has canonical reference closure" + ), + count=total, + details=[f"raw:{raw_id}" for raw_id in raw_sample] + + [f"attachment:{attachment_id}" for attachment_id in attachment_sample], + evidence={ + **counts, + "raw_sample": raw_sample, + "attachment_sample": attachment_sample, + }, + ) + + +def _check_blob_reference_closure(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: + return _check_blob_reference_closure_for_index(archive_root, _resolve_index_path(archive_root), sample_limit) + + +def _check_blob_reference_closure_at_index_path( + archive_root: Path, index_path: Path, sample_limit: int +) -> ArchiveVerificationCheck: + return _check_blob_reference_closure_for_index(archive_root, index_path, sample_limit) + + # --------------------------------------------------------------------------- # Check: raw failure lifecycle # --------------------------------------------------------------------------- @@ -2534,6 +2617,13 @@ def _check_user_tier_refs_at_candidate( ArchiveVerificationCheckClass.STATE_INVARIANT, _check_raw_failure_lifecycle_at_candidate, ), + ArchiveVerificationCheckSpec( + "blob-reference-closure", + "Every raw session has exactly one matching raw_payload ref and every acquired attachment is reachable by attachment_refs.", + _check_blob_reference_closure, + ArchiveVerificationCheckClass.STATE_INVARIANT, + _check_blob_reference_closure_at_index_path, + ), ArchiveVerificationCheckSpec( "pathology-zoo-invariants", "Every present pathology-zoo member satisfies its production-owned invariant.", @@ -2672,6 +2762,8 @@ def _check_user_tier_refs_at_candidate( #: candidate runner so ``index_path_override`` cannot silently inspect the #: active generation. REINDEX_CROSS_TIER_ACCEPTANCE_CHECKS: tuple[str, ...] = ( + "blob-reference-closure", + "raw-failure-lifecycle", "source-index-coverage", "blob-refs-liveness", "blob-integrity", diff --git a/polylogue/maintenance/blob_reference_closure.py b/polylogue/maintenance/blob_reference_closure.py new file mode 100644 index 0000000000..8beac17b88 --- /dev/null +++ b/polylogue/maintenance/blob_reference_closure.py @@ -0,0 +1,583 @@ +"""Read-only audit and guarded repair for acquired blob-reference closure.""" + +from __future__ import annotations + +import hashlib +import json +import os +import sqlite3 +import time +from contextlib import suppress +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path + +from polylogue.config import Config +from polylogue.maintenance.offline_guard import offline_maintenance_block_reason +from polylogue.paths import render_root +from polylogue.storage.attachment_relink import ( + MAX_ATTACHMENT_SAMPLE_LIMIT, + OrphanedAttachmentRelinkPlan, + RawSessionParser, + RelinkableAttachment, + UnrecoverableAttachmentReason, + plan_orphaned_attachment_relink, +) +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import ( + validate_backup_manifest_covers_derived_tier, + validate_migration_backup_manifest, +) + +TOOL_VERSION = "blob-reference-closure-v1" + + +class BlobReferenceClosureError(RuntimeError): + """Raised when a guarded closure repair cannot prove its write set.""" + + +class BlobReferenceBlockerKind(StrEnum): + """Typed reasons a closure row cannot be repaired by this route.""" + + RAW_NONEXACT_REFERENCE = "raw_nonexact_reference" + RAW_MISSING_AUTHORITATIVE_FIELD = "raw_missing_authoritative_field" + ATTACHMENT_NO_AUTHORITATIVE_RAW = "attachment_no_authoritative_raw" + ATTACHMENT_MESSAGE_MISSING = "attachment_message_missing" + + +@dataclass(frozen=True, slots=True) +class BlobReferenceClosureBlocker: + kind: BlobReferenceBlockerKind + object_id: str + detail: str + + def to_dict(self) -> dict[str, str]: + return {"kind": self.kind.value, "object_id": self.object_id, "detail": self.detail} + + +@dataclass(frozen=True, slots=True) +class RawBlobReferenceCandidate: + raw_id: str + blob_hash: bytes + source_path: str + blob_size: int + acquired_at_ms: int + + def to_dict(self) -> dict[str, object]: + return { + "raw_id": self.raw_id, + "blob_hash": self.blob_hash.hex(), + "source_path": self.source_path, + "blob_size": self.blob_size, + "acquired_at_ms": self.acquired_at_ms, + } + + +@dataclass(frozen=True, slots=True) +class BlobReferenceClosurePlan: + raw_candidates: tuple[RawBlobReferenceCandidate, ...] + attachment_candidates: tuple[RelinkableAttachment, ...] + blockers: tuple[BlobReferenceClosureBlocker, ...] + raw_rows_scanned: int + raw_rows_total: int + attachment_orphan_count: int + attachment_blockers_sampled: bool = False + + @property + def candidate_count(self) -> int: + return len(self.raw_candidates) + len(self.attachment_candidates) + + def to_dict(self) -> dict[str, object]: + return { + "raw_candidate_count": len(self.raw_candidates), + "attachment_candidate_count": len(self.attachment_candidates), + "candidate_count": self.candidate_count, + "raw_rows_scanned": self.raw_rows_scanned, + "raw_rows_total": self.raw_rows_total, + "attachment_orphan_count": self.attachment_orphan_count, + "attachment_blockers_sampled": self.attachment_blockers_sampled, + "blocker_count": len(self.blockers), + "blockers": [blocker.to_dict() for blocker in self.blockers], + } + + +@dataclass(frozen=True, slots=True) +class BlobReferenceClosureReport: + archive_root: str + dry_run: bool + applied: bool + plan: BlobReferenceClosurePlan + raw_repaired_count: int = 0 + attachment_repaired_count: int = 0 + backup_manifest: Path | None = None + receipt_path: Path | None = None + + def to_dict(self) -> dict[str, object]: + return { + "archive_root": self.archive_root, + "dry_run": self.dry_run, + "applied": self.applied, + "raw_repaired_count": self.raw_repaired_count, + "attachment_repaired_count": self.attachment_repaired_count, + "backup_manifest": str(self.backup_manifest) if self.backup_manifest is not None else None, + "receipt_path": str(self.receipt_path) if self.receipt_path is not None else None, + "plan": self.plan.to_dict(), + } + + +def _offline_config(archive_root: Path) -> Config: + return Config(archive_root=archive_root, render_root=render_root(), sources=[]) + + +def _open_ro(path: Path) -> sqlite3.Connection: + return sqlite3.connect(f"file:{path}?mode=ro", uri=True) + + +def raw_reference_closure_predicate(raw_alias: str = "r", ref_alias: str = "b") -> str: + """Return the canonical exact-one raw-payload reference predicate.""" + return f""" + ( + ( + SELECT COUNT(*) FROM blob_refs {ref_alias} + WHERE {ref_alias}.ref_type = 'raw_payload' + AND {ref_alias}.ref_id = {raw_alias}.raw_id + AND {ref_alias}.blob_hash = {raw_alias}.blob_hash + ) != 1 + OR ( + SELECT COUNT(*) FROM blob_refs {ref_alias} + WHERE {ref_alias}.ref_type = 'raw_payload' + AND {ref_alias}.ref_id = {raw_alias}.raw_id + ) != 1 + ) + """ + + +def _raw_candidates_and_blockers( + conn: sqlite3.Connection, +) -> tuple[list[RawBlobReferenceCandidate], list[BlobReferenceClosureBlocker], int]: + rows = conn.execute( + f""" + SELECT r.raw_id, r.blob_hash, r.source_path, r.blob_size, r.acquired_at_ms, + (SELECT COUNT(*) FROM blob_refs b + WHERE b.ref_type = 'raw_payload' AND b.ref_id = r.raw_id) AS ref_count, + (SELECT COUNT(*) FROM blob_refs b + WHERE b.ref_type = 'raw_payload' AND b.ref_id = r.raw_id + AND b.blob_hash = r.blob_hash) AS exact_count + FROM raw_sessions r + WHERE {raw_reference_closure_predicate()} + ORDER BY r.raw_id + """ + ).fetchall() + candidates: list[RawBlobReferenceCandidate] = [] + blockers: list[BlobReferenceClosureBlocker] = [] + for raw_id, blob_hash, source_path, blob_size, acquired_at_ms, ref_count, exact_count in rows: + if source_path is None or blob_size is None or acquired_at_ms is None: + blockers.append( + BlobReferenceClosureBlocker( + BlobReferenceBlockerKind.RAW_MISSING_AUTHORITATIVE_FIELD, + str(raw_id), + "raw_sessions lacks source_path, blob_size, or acquired_at_ms", + ) + ) + elif exact_count == 0 and ref_count == 0: + candidates.append( + RawBlobReferenceCandidate( + raw_id=str(raw_id), + blob_hash=bytes(blob_hash), + source_path=str(source_path), + blob_size=int(blob_size), + acquired_at_ms=int(acquired_at_ms), + ) + ) + else: + blockers.append( + BlobReferenceClosureBlocker( + BlobReferenceBlockerKind.RAW_NONEXACT_REFERENCE, + str(raw_id), + f"expected exactly one exact raw_payload ref; ref_count={ref_count}, exact_count={exact_count}", + ) + ) + total = int(conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone()[0]) + return candidates, blockers, total + + +def _attachment_blockers(plan: OrphanedAttachmentRelinkPlan) -> list[BlobReferenceClosureBlocker]: + blockers: list[BlobReferenceClosureBlocker] = [] + for item in plan.unrecoverable_samples[:MAX_ATTACHMENT_SAMPLE_LIMIT]: + if item.reason_kind is UnrecoverableAttachmentReason.MESSAGE_MISSING: + kind = BlobReferenceBlockerKind.ATTACHMENT_MESSAGE_MISSING + else: + kind = BlobReferenceBlockerKind.ATTACHMENT_NO_AUTHORITATIVE_RAW + blockers.append(BlobReferenceClosureBlocker(kind, item.attachment_id, item.reason)) + return blockers + + +def _acquired_attachment_ids(conn: sqlite3.Connection) -> set[str]: + rows = conn.execute( + """ + SELECT a.attachment_id + FROM attachments a + WHERE a.acquisition_status = 'acquired' + AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id) + """ + ).fetchall() + return {str(row[0]) for row in rows} + + +def _plan_connections( + index_conn: sqlite3.Connection, + source_conn: sqlite3.Connection, + *, + archive_root: Path, + sample_limit: int, + raw_session_parser: RawSessionParser | None = None, +) -> BlobReferenceClosurePlan: + raw_candidates, raw_blockers, raw_total = _raw_candidates_and_blockers(source_conn) + acquired_attachment_ids = _acquired_attachment_ids(index_conn) + attachment_plan = plan_orphaned_attachment_relink( + index_conn, + source_conn, + archive_root=archive_root, + blob_root=archive_root / "blob", + raw_row_limit=None, + sample_limit=MAX_ATTACHMENT_SAMPLE_LIMIT, + raw_session_parser=raw_session_parser, + ) + return BlobReferenceClosurePlan( + raw_candidates=tuple(raw_candidates), + attachment_candidates=tuple( + candidate for candidate in attachment_plan.eligible if candidate.attachment_id in acquired_attachment_ids + ), + blockers=tuple( + raw_blockers + + [ + blocker + for blocker in _attachment_blockers(attachment_plan) + if blocker.object_id in acquired_attachment_ids + ] + ), + raw_rows_scanned=attachment_plan.raw_rows_scanned, + raw_rows_total=raw_total, + attachment_orphan_count=len(acquired_attachment_ids), + attachment_blockers_sampled=attachment_plan.unrecoverable_samples_truncated, + ) + + +def plan_blob_reference_closure( + archive_root: Path, + *, + sample_limit: int = 30, + raw_session_parser: RawSessionParser | None = None, +) -> BlobReferenceClosurePlan: + """Build a complete, read-only plan from durable source and index evidence.""" + source_db = archive_root / "source.db" + index_db = archive_root / "index.db" + if not source_db.exists() or not index_db.exists(): + raise FileNotFoundError("blob-reference closure requires source.db and index.db") + source_conn = _open_ro(source_db) + index_conn = _open_ro(index_db) + try: + return _plan_connections( + index_conn, + source_conn, + archive_root=archive_root, + sample_limit=sample_limit, + raw_session_parser=raw_session_parser, + ) + finally: + index_conn.close() + source_conn.close() + + +def _plan_digest(plan: BlobReferenceClosurePlan) -> str: + attachment_payload: list[dict[str, object]] = [] + for attachment in plan.attachment_candidates: + attachment_payload.append( + { + "attachment_id": attachment.attachment_id, + "session_id": attachment.session_id, + "message_id": attachment.message_id, + "position": attachment.position, + "upload_origin": attachment.upload_origin, + "source_url": attachment.source_url, + "caption": attachment.caption, + "raw_id": attachment.raw_id, + "native_ids": attachment.native_ids, + } + ) + payload = { + "raw": [candidate.to_dict() for candidate in plan.raw_candidates], + "attachments": attachment_payload, + } + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def _write_receipt(path: Path, *, archive_root: Path, plan: BlobReferenceClosurePlan, backup_manifest: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + raise BlobReferenceClosureError(f"receipt already exists: {path}") + with path.open("x", encoding="utf-8") as handle: + json.dump( + { + "kind": "blob_reference_closure", + "tool_version": TOOL_VERSION, + "phase": "prepared", + "archive_root": str(archive_root), + "backup_manifest": str(backup_manifest), + "prepared_at_ms": int(time.time() * 1000), + "plan_digest": _plan_digest(plan), + "plan": plan.to_dict(), + }, + handle, + sort_keys=True, + ) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + _fsync_receipt_directory(path) + + +def _append_receipt(path: Path, phase: str, **extra: object) -> None: + with path.open("a", encoding="utf-8") as handle: + json.dump({"kind": "blob_reference_closure", "phase": phase, **extra}, handle, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + _fsync_receipt_directory(path) + + +def _fsync_receipt_directory(path: Path) -> None: + """Durably publish a newly created or extended receipt directory entry.""" + directory_fd = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + + +def _validate_backups(backup_manifest: Path, source_conn: sqlite3.Connection, index_conn: sqlite3.Connection) -> None: + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=source_conn) + validate_backup_manifest_covers_derived_tier(backup_manifest, ArchiveTier.INDEX, connection=index_conn) + + +def reconcile_blob_reference_closure( + archive_root: Path, + *, + backup_manifest: Path | None = None, + receipt_path: Path | None = None, + dry_run: bool = True, + sample_limit: int = 30, + raw_session_parser: RawSessionParser | None = None, +) -> BlobReferenceClosureReport: + """Plan closure repair, or add only deterministic exact references. + + Apply is offline, backup-gated, and additive. It never deletes or replaces + an existing reference. Attachment ownership is accepted only when a full + raw reparse reproduces the attachment identity and its message exists in + the current index. + """ + if dry_run: + return BlobReferenceClosureReport( + archive_root=str(archive_root), + dry_run=True, + applied=False, + plan=plan_blob_reference_closure( + archive_root, + sample_limit=sample_limit, + raw_session_parser=raw_session_parser, + ), + ) + if backup_manifest is None: + raise BlobReferenceClosureError("apply requires a verified backup manifest covering source.db and index.db") + if receipt_path is None: + raise BlobReferenceClosureError("apply requires an explicit receipt path") + if reason := offline_maintenance_block_reason(_offline_config(archive_root), active=True, dry_run=False): + raise BlobReferenceClosureError(reason) + + source_db = archive_root / "source.db" + index_db = archive_root / "index.db" + source_conn = sqlite3.connect(source_db) + index_conn: sqlite3.Connection | None = sqlite3.connect(index_db) + assert index_conn is not None + source_conn.execute("PRAGMA foreign_keys = ON") + index_conn.execute("PRAGMA foreign_keys = ON") + plan: BlobReferenceClosurePlan | None = None + source_repaired = 0 + attachment_repaired = 0 + prepared = False + committed = False + attached_index = False + try: + try: + assert index_conn is not None + _validate_backups(backup_manifest, source_conn, index_conn) + plan = _plan_connections( + index_conn, + source_conn, + archive_root=archive_root, + sample_limit=sample_limit, + raw_session_parser=raw_session_parser, + ) + _write_receipt(receipt_path, archive_root=archive_root, plan=plan, backup_manifest=backup_manifest) + prepared = True + + # A single connection and attached index database give SQLite one + # transaction boundary for both tiers. Planning and backup checks + # happen before ATTACH, so every conflict is known before either + # tier is mutated. + index_conn.close() + index_conn = None + source_conn.execute("ATTACH DATABASE ? AS index_tier", (str(index_db),)) + attached_index = True + source_conn.execute("BEGIN IMMEDIATE") + for candidate in plan.raw_candidates: + source_conn.execute( + """ + INSERT INTO blob_refs ( + blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms + ) VALUES (?, ?, 'raw_payload', ?, ?, ?) + """, + ( + candidate.blob_hash, + candidate.raw_id, + candidate.source_path, + candidate.blob_size, + candidate.acquired_at_ms, + ), + ) + exact = source_conn.execute( + """ + SELECT COUNT(*) FROM blob_refs b + JOIN raw_sessions r ON r.raw_id = b.ref_id AND r.blob_hash = b.blob_hash + WHERE b.ref_type = 'raw_payload' AND b.ref_id = ? + """, + (candidate.raw_id,), + ).fetchone()[0] + if exact != 1: + raise BlobReferenceClosureError(f"raw exact-match check failed after insert: {candidate.raw_id}") + source_repaired += 1 + for attachment_candidate in plan.attachment_candidates: + source_conn.execute( + """ + INSERT INTO index_tier.attachment_refs ( + attachment_id, session_id, message_id, position, upload_origin, source_url, caption + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + attachment_candidate.attachment_id, + attachment_candidate.session_id, + attachment_candidate.message_id, + attachment_candidate.position, + attachment_candidate.upload_origin, + attachment_candidate.source_url, + attachment_candidate.caption, + ), + ) + source_conn.execute( + """ + UPDATE index_tier.attachments + SET ref_count = ( + SELECT COUNT(*) FROM index_tier.attachment_refs + WHERE index_tier.attachment_refs.attachment_id = index_tier.attachments.attachment_id + ) + WHERE attachment_id = ? + """, + (attachment_candidate.attachment_id,), + ) + for id_kind, native_id in attachment_candidate.native_ids: + source_conn.execute( + """ + INSERT OR IGNORE INTO index_tier.attachment_native_ids (ref_id, id_kind, native_id) + VALUES (?, ?, ?) + """, + ( + f"{attachment_candidate.message_id}:attachment:{attachment_candidate.position}", + id_kind, + native_id, + ), + ) + exact = source_conn.execute( + "SELECT COUNT(*) FROM index_tier.attachment_refs WHERE attachment_id = ?", + (attachment_candidate.attachment_id,), + ).fetchone()[0] + if exact < 1: + raise BlobReferenceClosureError( + f"attachment reference check failed after insert: {attachment_candidate.attachment_id}" + ) + attachment_repaired += 1 + source_conn.commit() + committed = True + _append_receipt(receipt_path, "source_committed", repaired_count=source_repaired) + _append_receipt(receipt_path, "index_committed", repaired_count=attachment_repaired) + _append_receipt( + receipt_path, + "committed", + raw_repaired_count=source_repaired, + attachment_repaired_count=attachment_repaired, + ) + except Exception as exc: + if source_conn.in_transaction: + source_conn.rollback() + if prepared: + with suppress(OSError): + _append_receipt( + receipt_path, + "committed_receipt_incomplete" if committed else "aborted", + error=str(exc), + ) + raise + finally: + if attached_index: + with suppress(sqlite3.Error): + source_conn.execute("DETACH DATABASE index_tier") + if index_conn is not None: + index_conn.close() + source_conn.close() + + assert plan is not None + return BlobReferenceClosureReport( + archive_root=str(archive_root), + dry_run=False, + applied=True, + plan=plan, + raw_repaired_count=source_repaired, + attachment_repaired_count=attachment_repaired, + backup_manifest=backup_manifest, + receipt_path=receipt_path, + ) + + +def closure_counts(source_conn: sqlite3.Connection, index_conn: sqlite3.Connection) -> dict[str, int]: + """Return exact structural closure counts without parsing or mutation.""" + raw_missing = int( + source_conn.execute( + f""" + SELECT COUNT(*) FROM raw_sessions r + WHERE {raw_reference_closure_predicate()} + """ + ).fetchone()[0] + ) + attachment_missing = int( + index_conn.execute( + """ + SELECT COUNT(*) FROM attachments a + WHERE a.acquisition_status = 'acquired' + AND NOT EXISTS (SELECT 1 FROM attachment_refs r WHERE r.attachment_id = a.attachment_id) + """ + ).fetchone()[0] + ) + return {"raw_missing_exact_count": raw_missing, "acquired_attachment_missing_ref_count": attachment_missing} + + +__all__ = [ + "BlobReferenceBlockerKind", + "BlobReferenceClosureBlocker", + "BlobReferenceClosureError", + "BlobReferenceClosurePlan", + "BlobReferenceClosureReport", + "RawBlobReferenceCandidate", + "closure_counts", + "plan_blob_reference_closure", + "raw_reference_closure_predicate", + "reconcile_blob_reference_closure", +] diff --git a/polylogue/material_protocol/v1/records.py b/polylogue/material_protocol/v1/records.py index c1cba3bf65..8464f1a2dc 100644 --- a/polylogue/material_protocol/v1/records.py +++ b/polylogue/material_protocol/v1/records.py @@ -8,6 +8,7 @@ from __future__ import annotations +from polylogue.core.identity_law import message_local_id from polylogue.core.json import JSONValue from polylogue.material_protocol.v1.input_model import ( AttachmentInput, @@ -21,8 +22,8 @@ def message_native_component(message: MessageInput) -> str: - """The COALESCE(native_id, position||'.'||variant_index) component.""" - return message.native_id if message.native_id is not None else f"{message.position}.{message.variant_index}" + """The tagged native-id or position/variant message-id component.""" + return message_local_id(message.native_id, position=message.position, variant_index=message.variant_index) def message_id_for(session_id: str, message: MessageInput) -> str: @@ -107,7 +108,11 @@ def usage_record(session_id: str, usage: UsageInput) -> dict[str, JSONValue]: def message_record(session_id: str, message: MessageInput) -> dict[str, JSONValue]: message_id = message_id_for(session_id, message) - parent_message_id = f"{session_id}:{message.parent_native_id}" if message.parent_native_id is not None else None + parent_message_id = ( + f"{session_id}:{message_local_id(message.parent_native_id, position=0)}" + if message.parent_native_id is not None and message.parent_native_id.strip() + else None + ) return { "kind": "message", "record_id": message_id, diff --git a/polylogue/pipeline/services/ingest_batch/_core.py b/polylogue/pipeline/services/ingest_batch/_core.py index e21bc1566f..73002dba49 100644 --- a/polylogue/pipeline/services/ingest_batch/_core.py +++ b/polylogue/pipeline/services/ingest_batch/_core.py @@ -76,6 +76,7 @@ from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceBlobRef from polylogue.storage.sqlite.archive_tiers.write import ( _message_content_hash, + _normalized_message_native_id, _timestamp_ms, _write_repo_edges, replace_parser_ingest_flag_tags, @@ -373,7 +374,7 @@ def _needs_session_fts_repair(conn: sqlite3.Connection, session_id: str) -> bool def _existing_native_message_ids(conn: sqlite3.Connection, session_id: str) -> set[str]: return { - str(row[0]) + str(row[0]).strip() for row in conn.execute( "SELECT native_id FROM messages WHERE session_id = ? AND native_id IS NOT NULL", (session_id,), @@ -388,7 +389,8 @@ def _append_delta_payload( existing_native_ids = _existing_native_message_ids(conn, payload.session_id) delta_messages: list[ParsedMessage] = [] for message in payload.parsed_session.messages: - if message.provider_message_id in existing_native_ids: + native_id = _normalized_message_native_id(message) + if native_id is not None and native_id in existing_native_ids: continue delta_messages.append(message.model_copy(update={"position": None})) if not delta_messages and not payload.attachment_count: @@ -410,7 +412,7 @@ def _append_payload_changes_existing_message( session depends on arrival order. """ existing_rows = { - str(row[0]): (int(row[1]), int(row[2]), row[3]) + str(row[0]).strip(): (int(row[1]), int(row[2]), row[3]) for row in conn.execute( """ SELECT native_id, position, variant_index, content_hash @@ -421,7 +423,8 @@ def _append_payload_changes_existing_message( ).fetchall() } for message in payload.parsed_session.messages: - existing = existing_rows.get(message.provider_message_id) + native_id = _normalized_message_native_id(message) + existing = existing_rows.get(native_id) if native_id is not None else None if existing is None: continue position, variant_index, existing_hash = existing diff --git a/polylogue/storage/attachment_relink.py b/polylogue/storage/attachment_relink.py index f2a1b31ce9..a98397f74a 100644 --- a/polylogue/storage/attachment_relink.py +++ b/polylogue/storage/attachment_relink.py @@ -36,17 +36,24 @@ import sqlite3 from collections.abc import Callable, Mapping from dataclasses import dataclass +from enum import StrEnum from pathlib import Path from polylogue.logging import get_logger from polylogue.pipeline.services.ingest_worker import IngestRecordResult, SessionWritePayload, ingest_record +from polylogue.sources.parsers.base import ParsedAttachment, ParsedMessage from polylogue.storage.runtime.raw.records import RawSessionRecord from polylogue.storage.sqlite.archive_tiers.write import ( _attachment_caption, _attachment_id, - _attachment_position, + _attachment_message_id_maps, + _attachment_native_id_values, + _attachment_reference_positions, _attachment_source_url, - _message_id, + _duplicate_message_native_ids, + _message_content_hash, + _next_message_position, + _normalized_message_native_id, ) from polylogue.storage.sqlite.queries.mappers_archive import _row_to_raw_session @@ -55,6 +62,7 @@ # Raw parsing is real work (decode + provider dispatch + normalize); bound how # many raw_sessions rows one plan pass inspects unless the caller overrides it. DEFAULT_RAW_ROW_LIMIT = 5_000 +MAX_ATTACHMENT_SAMPLE_LIMIT = 1_000_000 _NO_RAW_MATCH_REASON = "no raw session in source.db reproduces this attachment's identity" _MESSAGE_MISSING_REASON = ( @@ -75,12 +83,19 @@ class RelinkableAttachment: source_url: str | None caption: str | None raw_id: str + native_ids: tuple[tuple[str, str], ...] + + +class UnrecoverableAttachmentReason(StrEnum): + NO_AUTHORITATIVE_RAW = "no_authoritative_raw" + MESSAGE_MISSING = "message_missing" @dataclass(frozen=True, slots=True) class UnrecoverableAttachment: attachment_id: str reason: str + reason_kind: UnrecoverableAttachmentReason = UnrecoverableAttachmentReason.NO_AUTHORITATIVE_RAW @dataclass(frozen=True, slots=True) @@ -101,6 +116,7 @@ class OrphanedAttachmentRelinkPlan: unrecoverable_samples: tuple[UnrecoverableAttachment, ...] raw_rows_scanned: int raw_rows_total: int + unrecoverable_samples_truncated: bool = False RawSessionParser = Callable[[RawSessionRecord], IngestRecordResult] @@ -130,6 +146,84 @@ def _message_exists(index_conn: sqlite3.Connection, message_id: str) -> bool: return index_conn.execute("SELECT 1 FROM messages WHERE message_id = ?", (message_id,)).fetchone() is not None +def _append_materialized_message_ids( + index_conn: sqlite3.Connection, + session_id: str, + messages: list[ParsedMessage], +) -> dict[int, str]: + """Find append payload messages already present in the current index. + + ``_write_session`` clears append message positions before the writer adds + the current tail offset. Once that append is materialized, deriving an + owner from the raw payload and the *new* ``MAX(position) + 1`` offset + points past the real row. Native ids identify materialized rows directly; + id-less messages use the writer's content hash at each stored position so + an old base message cannot be mistaken for the append tail. + """ + rows = index_conn.execute( + """ + SELECT message_id, native_id, position, variant_index, content_hash + FROM messages + WHERE session_id = ? + ORDER BY position, variant_index + """, + (session_id,), + ).fetchall() + duplicate_native_ids = _duplicate_message_native_ids(messages) + rows_by_content_hash: dict[bytes, list[sqlite3.Row]] = {} + for row in rows: + rows_by_content_hash.setdefault(bytes(row[4]), []).append(row) + content_hash_cache: dict[tuple[int, int, int], bytes] = {} + resolved: dict[int, str] = {} + for message_index, message in enumerate(messages): + native_id = _normalized_message_native_id(message) + if native_id is not None and native_id not in duplicate_native_ids: + matches = [row for row in rows if row[1] == native_id] + else: + matches = [] + for row in rows: + position = int(row[2]) + variant_index = int(row[3]) + cache_key = (message_index, position, variant_index) + message_hash = content_hash_cache.get(cache_key) + if message_hash is None: + message_hash = _message_content_hash( + session_id, + message, + position=position, + variant_index=variant_index, + ) + content_hash_cache[cache_key] = message_hash + matches.extend(rows_by_content_hash.get(message_hash, ())) + if len(matches) == 1: + resolved[message_index] = str(matches[0][0]) + return resolved + + +def _append_materialized_attachment_maps( + index_conn: sqlite3.Connection, + session_id: str, + messages: list[ParsedMessage], +) -> tuple[dict[str, str], dict[int, str]]: + resolved = _append_materialized_message_ids(index_conn, session_id, messages) + by_native_message_id: dict[str, str] = {} + by_message_position: dict[int, str] = {} + for message_index, message in enumerate(messages): + message_id = resolved.get(message_index) + if message_id is None: + continue + provider_message_id = message.provider_message_id + if provider_message_id: + by_native_message_id[str(provider_message_id)] = message_id + normalized = _normalized_message_native_id(message) + if normalized is not None: + by_native_message_id[normalized] = message_id + message_position = message.position + if message_position is not None: + by_message_position[int(message_position)] = message_id + return by_native_message_id, by_message_position + + def _iter_raw_session_rows(source_conn: sqlite3.Connection, *, raw_row_limit: int | None) -> list[sqlite3.Row]: original_row_factory = source_conn.row_factory source_conn.row_factory = sqlite3.Row @@ -179,11 +273,12 @@ def plan_orphaned_attachment_relink( unrecoverable_samples=(), raw_rows_scanned=0, raw_rows_total=0, + unrecoverable_samples_truncated=False, ) pending: set[str] = set(orphan_ids) recovered: dict[str, RelinkableAttachment] = {} - ineligible_reasons: dict[str, str] = {} + ineligible_reasons: dict[str, tuple[UnrecoverableAttachmentReason, str]] = {} parser = raw_session_parser or _default_raw_session_parser(archive_root, blob_root) @@ -212,14 +307,20 @@ def plan_orphaned_attachment_relink( _match_session_payload(payload, pending, recovered, ineligible_reasons, index_conn, raw_record.raw_id) reason_counts: dict[str, int] = {} + effective_sample_limit = min(max(sample_limit, 0), MAX_ATTACHMENT_SAMPLE_LIMIT) samples: list[UnrecoverableAttachment] = [] + unrecoverable_count = 0 for attachment_id in orphan_ids: if attachment_id in recovered: continue - reason = ineligible_reasons.get(attachment_id, _NO_RAW_MATCH_REASON) + reason_kind, reason = ineligible_reasons.get( + attachment_id, + (UnrecoverableAttachmentReason.NO_AUTHORITATIVE_RAW, _NO_RAW_MATCH_REASON), + ) reason_counts[reason] = reason_counts.get(reason, 0) + 1 - if len(samples) < sample_limit: - samples.append(UnrecoverableAttachment(attachment_id=attachment_id, reason=reason)) + unrecoverable_count += 1 + if len(samples) < effective_sample_limit: + samples.append(UnrecoverableAttachment(attachment_id=attachment_id, reason=reason, reason_kind=reason_kind)) return OrphanedAttachmentRelinkPlan( orphan_count=orphan_count, @@ -228,6 +329,7 @@ def plan_orphaned_attachment_relink( unrecoverable_samples=tuple(samples), raw_rows_scanned=scanned, raw_rows_total=raw_rows_total, + unrecoverable_samples_truncated=unrecoverable_count > len(samples), ) @@ -235,7 +337,7 @@ def _match_session_payload( payload: SessionWritePayload, pending: set[str], recovered: dict[str, RelinkableAttachment], - ineligible_reasons: dict[str, str], + ineligible_reasons: dict[str, tuple[UnrecoverableAttachmentReason, str]], index_conn: sqlite3.Connection, raw_id: str, ) -> None: @@ -243,36 +345,74 @@ def _match_session_payload( return session_id = payload.session_id messages = payload.parsed_session.messages - by_native_message_id = { - message.provider_message_id: _message_id(session_id, message, fallback_position) - for fallback_position, message in enumerate(messages) - if message.provider_message_id - } + position_offset = _next_message_position(index_conn, session_id) if payload.append_only else 0 + by_native_message_id, by_message_position = _attachment_message_id_maps( + session_id, + messages, + position_offset=position_offset, + ) + if payload.append_only: + materialized_by_native_id, materialized_by_position = _append_materialized_attachment_maps( + index_conn, + session_id, + messages, + ) + by_native_message_id.update(materialized_by_native_id) + by_message_position.update(materialized_by_position) # Attachments are session-level (``ParsedSession.attachments``), each # linked to its owning message via ``message_provider_id`` -- mirroring - # exactly how ``_write_attachments`` consumes them (write.py:3288-3318). + # exactly how ``_write_attachments`` consumes them (write.py:_attachment_message_id_maps). + resolved_message_ids: dict[int, str] = {} + attachments_by_message: dict[str, list[ParsedAttachment]] = {} for attachment in payload.parsed_session.attachments: attachment_id = _attachment_id(session_id, attachment) - if attachment_id not in pending: - continue message_id = ( by_native_message_id.get(attachment.message_provider_id) if attachment.message_provider_id else None ) + if message_id is None and attachment.message_position is not None: + message_id = by_message_position.get(attachment.message_position) if message_id is None: - ineligible_reasons.setdefault(attachment_id, _NO_RAW_MATCH_REASON) + if attachment_id in pending: + ineligible_reasons.setdefault( + attachment_id, + (UnrecoverableAttachmentReason.NO_AUTHORITATIVE_RAW, _NO_RAW_MATCH_REASON), + ) continue if not _message_exists(index_conn, message_id): - ineligible_reasons.setdefault(attachment_id, _MESSAGE_MISSING_REASON) + if attachment_id in pending: + ineligible_reasons.setdefault( + attachment_id, + (UnrecoverableAttachmentReason.MESSAGE_MISSING, _MESSAGE_MISSING_REASON), + ) + continue + resolved_message_ids[id(attachment)] = message_id + attachments_by_message.setdefault(message_id, []).append(attachment) + + attachment_positions: dict[int, int] = {} + for message_id, message_group in attachments_by_message.items(): + rows = index_conn.execute( + "SELECT position, attachment_id FROM attachment_refs WHERE message_id = ?", + (message_id,), + ).fetchall() + occupied = {int(row[0]) for row in rows if str(row[1]) not in pending} + attachment_positions.update(_attachment_reference_positions(message_group, occupied_positions=occupied)) + for attachment in payload.parsed_session.attachments: + attachment_id = _attachment_id(session_id, attachment) + if attachment_id not in pending: + continue + message_id = resolved_message_ids.get(id(attachment)) + if message_id is None: continue recovered[attachment_id] = RelinkableAttachment( attachment_id=attachment_id, session_id=session_id, message_id=message_id, - position=_attachment_position(attachment), + position=attachment_positions[id(attachment)], upload_origin=attachment.upload_origin, source_url=_attachment_source_url(attachment), caption=_attachment_caption(attachment), raw_id=raw_id, + native_ids=_attachment_native_id_values(attachment), ) pending.discard(attachment_id) @@ -305,10 +445,11 @@ def relink_orphaned_attachments( ``attachment_refs`` row per eligible orphan (mirroring the exact INSERT ``_write_attachments`` uses) and refreshes that attachment's ``ref_count`` -- it never touches an attachment this plan did not mark eligible, and it - is safe to call repeatedly (``INSERT OR REPLACE`` + a ref_count recompute, - both idempotent). ``index_conn`` must be a writable index-tier connection - when ``dry_run=False``; this function does not commit -- the caller owns - transaction/commit scope. + is safe to call repeatedly (the same reference identity is idempotent). + ``index_conn`` must be a writable index-tier connection when + ``dry_run=False``; this function does not commit -- the caller owns + transaction/commit scope. Any write error is propagated so callers cannot + accidentally commit a partial relink. """ plan = plan_orphaned_attachment_relink( index_conn, @@ -320,42 +461,43 @@ def relink_orphaned_attachments( raw_session_parser=raw_session_parser, ) relinked = 0 - errors: list[str] = [] if not dry_run: for item in plan.eligible: - try: - index_conn.execute( - """ - INSERT OR REPLACE INTO attachment_refs ( - attachment_id, session_id, message_id, position, upload_origin, source_url, caption - ) VALUES (?, ?, ?, ?, ?, ?, ?) - """, - ( - item.attachment_id, - item.session_id, - item.message_id, - item.position, - item.upload_origin, - item.source_url, - item.caption, - ), - ) + index_conn.execute( + """ + INSERT INTO attachment_refs ( + attachment_id, session_id, message_id, position, upload_origin, source_url, caption + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + item.attachment_id, + item.session_id, + item.message_id, + item.position, + item.upload_origin, + item.source_url, + item.caption, + ), + ) + for id_kind, native_id in item.native_ids: index_conn.execute( """ - UPDATE attachments - SET ref_count = ( - SELECT COUNT(*) FROM attachment_refs WHERE attachment_refs.attachment_id = attachments.attachment_id - ) - WHERE attachment_id = ? + INSERT OR IGNORE INTO attachment_native_ids (ref_id, id_kind, native_id) + VALUES (?, ?, ?) """, - (item.attachment_id,), + (f"{item.message_id}:attachment:{item.position}", id_kind, native_id), ) - relinked += 1 - except sqlite3.Error as exc: - logger.warning( - "attachment relink: failed to write ref for attachment_id=%s: %s", item.attachment_id, exc + index_conn.execute( + """ + UPDATE attachments + SET ref_count = ( + SELECT COUNT(*) FROM attachment_refs WHERE attachment_refs.attachment_id = attachments.attachment_id ) - errors.append(f"{item.attachment_id}: {exc}") + WHERE attachment_id = ? + """, + (item.attachment_id,), + ) + relinked += 1 return OrphanedAttachmentRelinkResult( orphan_count=plan.orphan_count, @@ -363,15 +505,17 @@ def relink_orphaned_attachments( relinked_count=relinked, unrecoverable_reason_counts=plan.unrecoverable_reason_counts, unrecoverable_samples=plan.unrecoverable_samples, - errors=tuple(errors), + errors=(), ) __all__ = [ + "MAX_ATTACHMENT_SAMPLE_LIMIT", "OrphanedAttachmentRelinkPlan", "OrphanedAttachmentRelinkResult", "RelinkableAttachment", "UnrecoverableAttachment", + "UnrecoverableAttachmentReason", "plan_orphaned_attachment_relink", "relink_orphaned_attachments", ] diff --git a/polylogue/storage/sqlite/archive_tiers/archive_tiers_specs.py b/polylogue/storage/sqlite/archive_tiers/archive_tiers_specs.py index 8ad7e8ef4f..7a7c8ec063 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive_tiers_specs.py +++ b/polylogue/storage/sqlite/archive_tiers/archive_tiers_specs.py @@ -83,7 +83,7 @@ def _make_messages_spec() -> TableColumnSpec: all_columns: tuple[ColumnSpec, ...] = ( _ddl( "message_id", - "TEXT GENERATED ALWAYS AS (session_id || ':' || COALESCE(native_id, position || '.' || variant_index)) STORED UNIQUE", + "TEXT GENERATED ALWAYS AS (session_id || ':' || CASE WHEN native_id IS NULL THEN 'p:' || position || '.' || variant_index ELSE 'n:' || native_id END) STORED UNIQUE", ), _ddl("session_id", "TEXT NOT NULL REFERENCES sessions(session_id) ON DELETE CASCADE"), _ddl("native_id", "TEXT"), diff --git a/polylogue/storage/sqlite/archive_tiers/index.py b/polylogue/storage/sqlite/archive_tiers/index.py index 9dcab6bd29..116acfea9c 100644 --- a/polylogue/storage/sqlite/archive_tiers/index.py +++ b/polylogue/storage/sqlite/archive_tiers/index.py @@ -440,7 +440,11 @@ # It is computed entirely from existing action_pairs outcome columns, so a # fast-forward only replaces the derived view. No raw session is reparsed and # no persisted source/blob value changes. -INDEX_SCHEMA_VERSION = 65 +# polylogue-fix-blob-reference-closure: v66 separates native message ids from +# positional coordinates in the generated message_id expression. Existing +# derived rows remain readable as opaque legacy ids, but new materialization +# must replay raw sessions to regenerate message/block/reference identities. +INDEX_SCHEMA_VERSION = 66 # polylogue-v6i3: shared WHEN-clause fragment gating the blocks_command_trigram # trigger BODIES on the same dedicated bulk-build guard row messages_fts's diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index fed424b951..32bcd3fcbf 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/polylogue/storage/sqlite/archive_tiers/write.py @@ -683,11 +683,7 @@ def add_timing(name: str, started_at: float) -> None: projection_carry_forward: _ProjectionCarryForward | None = None t0 = time.perf_counter() if merge_append: - row = conn.execute( - "SELECT COALESCE(MAX(position) + 1, 0) FROM messages WHERE session_id = ?", - (session_id,), - ).fetchone() - position_offset = int(row[0] or 0) if row is not None else 0 + position_offset = _next_message_position(conn, session_id) _assert_unique_message_coordinates(session_id, messages, position_offset=position_offset) conn.execute( """ @@ -1829,7 +1825,7 @@ def _build_message_rows( def _messages_insert_sql() -> str: spec = archive_tiers_specs.MESSAGES_SPEC return f""" - INSERT OR REPLACE INTO messages ( + INSERT INTO messages ( {spec.insert_column_names} ) VALUES ({spec.insert_placeholder_string}) """ @@ -3326,47 +3322,110 @@ def _increment_session_counts_for_append( ) -def _write_attachments( - conn: sqlite3.Connection, +def _attachment_message_id_maps( session_id: str, messages: list[ParsedMessage], - attachments: Iterable[ParsedAttachment], *, position_offset: int = 0, - duplicate_native_ids: frozenset[str] = frozenset(), - refresh_attachment_ids: set[str] | None = None, - preacquired_blobs: dict[int, tuple[bytes | None, int, str]] | None = None, -) -> None: - by_native_message_id = { - message.provider_message_id: _message_id( + duplicate_native_ids: frozenset[str] | None = None, +) -> tuple[dict[str, str], dict[int, str]]: + """Build the authoritative attachment-owner lookup maps. + + Native message ids are usable only when unique after the same SQLite + normalization used by the writer. Message positions remain the fallback + for id-less attachments and for attachments whose native id is ambiguous. + Keep this shared with repair/relink paths so they cannot invent a weaker + ownership rule than the production write. + """ + duplicates = duplicate_native_ids if duplicate_native_ids is not None else _duplicate_message_native_ids(messages) + by_native_message_id: dict[str, str] = {} + for fallback_position, message in enumerate(messages): + normalized = _normalized_message_native_id(message) + if normalized is None or normalized in duplicates: + continue + message_id = _message_id( session_id, message, fallback_position, position_offset=position_offset, - duplicate_native_ids=duplicate_native_ids, + duplicate_native_ids=duplicates, ) - for fallback_position, message in enumerate(messages) - if message.provider_message_id and message.provider_message_id not in duplicate_native_ids - } + by_native_message_id[normalized] = message_id + if message.provider_message_id != normalized: + by_native_message_id[message.provider_message_id] = message_id by_message_position = { message.position: _message_id( session_id, message, fallback_position, position_offset=position_offset, - duplicate_native_ids=duplicate_native_ids, + duplicate_native_ids=duplicates, ) for fallback_position, message in enumerate(messages) if message.position is not None } - touched_attachment_ids: set[str] = set() + return by_native_message_id, by_message_position + + +def _next_message_position(conn: sqlite3.Connection, session_id: str) -> int: + """Return the position offset used when appending messages to a session.""" + row = conn.execute( + "SELECT COALESCE(MAX(position) + 1, 0) FROM messages WHERE session_id = ?", + (session_id,), + ).fetchone() + return int(row[0] or 0) if row is not None else 0 + + +def _write_attachments( + conn: sqlite3.Connection, + session_id: str, + messages: list[ParsedMessage], + attachments: Iterable[ParsedAttachment], + *, + position_offset: int = 0, + duplicate_native_ids: frozenset[str] = frozenset(), + refresh_attachment_ids: set[str] | None = None, + preacquired_blobs: dict[int, tuple[bytes | None, int, str]] | None = None, +) -> None: + attachments = tuple(attachments) + by_native_message_id, by_message_position = _attachment_message_id_maps( + session_id, + messages, + position_offset=position_offset, + duplicate_native_ids=duplicate_native_ids, + ) + attachment_positions: dict[int, int] = {} + resolved_message_ids: dict[int, str] = {} + attachments_by_message: defaultdict[str, list[ParsedAttachment]] = defaultdict(list) for attachment in attachments: - attachment_id = _attachment_id(session_id, attachment) message_id = ( by_native_message_id.get(attachment.message_provider_id) if attachment.message_provider_id else None ) if message_id is None and attachment.message_position is not None: message_id = by_message_position.get(attachment.message_position) + if message_id is not None: + resolved_message_ids[id(attachment)] = message_id + attachments_by_message[message_id].append(attachment) + for message_id, message_group in attachments_by_message.items(): + current_ids = {_attachment_id(session_id, attachment) for attachment in message_group} + occupied = ( + { + int(row[0]) + for row in conn.execute( + "SELECT position FROM attachment_refs WHERE message_id = ? AND attachment_id NOT IN ({})".format( + ",".join("?" for _ in current_ids) + ), + (message_id, *sorted(current_ids)), + ).fetchall() + } + if current_ids + else set() + ) + attachment_positions.update(_attachment_reference_positions(message_group, occupied_positions=occupied)) + touched_attachment_ids: set[str] = set() + for attachment in attachments: + attachment_id = _attachment_id(session_id, attachment) + message_id = resolved_message_ids.get(id(attachment)) if message_id is None: continue touched_attachment_ids.add(attachment_id) @@ -3397,16 +3456,33 @@ def _write_attachments( acquisition_status, ), ) - ref_position = _attachment_position(attachment) + ref_position = attachment_positions[id(attachment)] ref_id = f"{message_id}:attachment:{ref_position}" # Bulk rebuilds may suspend FK enforcement. Mirror REPLACE's cascade # explicitly so identifiers from an older projection cannot survive. + existing_ref = conn.execute( + "SELECT attachment_id FROM attachment_refs WHERE message_id = ? AND position = ?", + (message_id, ref_position), + ).fetchone() + if existing_ref is not None and existing_ref[0] != attachment_id: + raise ValueError( + "distinct attachments resolved to one reference identity: " + f"message_id={message_id!r}, position={ref_position}, " + f"existing_attachment_id={existing_ref[0]!r}, incoming_attachment_id={attachment_id!r}" + ) conn.execute("DELETE FROM attachment_native_ids WHERE ref_id = ?", (ref_id,)) conn.execute( """ - INSERT OR REPLACE INTO attachment_refs ( + INSERT INTO attachment_refs ( attachment_id, session_id, message_id, position, upload_origin, source_url, caption ) VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(message_id, position) DO UPDATE SET + attachment_id = excluded.attachment_id, + session_id = excluded.session_id, + upload_origin = excluded.upload_origin, + source_url = excluded.source_url, + caption = excluded.caption + WHERE attachment_refs.attachment_id = excluded.attachment_id """, ( attachment_id, @@ -3526,7 +3602,7 @@ def _write_parent_links( duplicate_native_ids=duplicate_native_ids, ) for fallback_position, message in enumerate(messages) - if message.provider_message_id and message.provider_message_id not in duplicate_native_ids + if message.provider_message_id and _normalized_message_native_id(message) not in duplicate_native_ids } by_message_position = { message.position: _message_id( @@ -4268,8 +4344,8 @@ def _write_session_events( ) for fallback_position, message in enumerate(messages) if message.provider_message_id - and message.provider_message_id not in duplicate_native_ids - and message.provider_message_id not in ambiguous_source_provider_ids + and _normalized_message_native_id(message) not in duplicate_native_ids + and _normalized_message_native_id(message) not in ambiguous_source_provider_ids } wrote_provider_usage_events = False position = event_position_offset @@ -6338,7 +6414,7 @@ def _extract_prefix_tail( inherited_refs: dict[str, str] = {} for index, message in enumerate(messages[:k]): provider_id = message.provider_message_id - if provider_id and provider_id not in duplicate_native_ids: + if provider_id and _normalized_message_native_id(message) not in duplicate_native_ids: inherited_refs[provider_id] = parent_composed[index][0] return (branch_point_message_id, "prefix-sharing", messages[k:], inherited_refs) @@ -6448,29 +6524,38 @@ def _message_id( def _duplicate_message_native_ids(messages: Iterable[ParsedMessage]) -> frozenset[str]: - """Native ids that collide once normalized the same way ``messages.native_id`` stores them. - - Counts by the surrogate-substituted (``_sqlite_text``) form, not the raw - provider string, so two distinct raw ids that collapse onto the same - U+FFFD-substituted text are treated as ambiguous too -- otherwise the - ``messages`` UNIQUE generated ``message_id`` column would silently - resolve the collision via ``INSERT OR REPLACE`` (one message vanishes) - while Python-side code still believed both had distinct identities. + """Native ids that collide after the same normalization ``messages.native_id`` stores. + + Counts by the stripped, surrogate-substituted (``_sqlite_text``) form, + not the raw provider string, so whitespace variants and two distinct raw + ids that collapse onto the same U+FFFD-substituted text are treated as + ambiguous too. Otherwise the ``messages`` UNIQUE generated ``message_id`` + column would silently resolve the collision via ``INSERT OR REPLACE`` (one + message vanishes) while Python-side code still believed both had distinct + identities. """ counts = Counter( - normalized for message in messages if (normalized := _sqlite_text(message.provider_message_id)) is not None + normalized for message in messages if (normalized := _normalized_message_native_id(message)) is not None ) return frozenset(native_id for native_id, count in counts.items() if count > 1) +def _normalized_message_native_id(message: ParsedMessage) -> str | None: + native_id = _sqlite_text(message.provider_message_id) + if native_id is None: + return None + stripped = native_id.strip() + return stripped or None + + def _effective_message_native_id(message: ParsedMessage, duplicate_native_ids: frozenset[str]) -> str | None: - """Return the surrogate-normalized native id, or ``None`` if ambiguous. + """Return the storage-normalized native id, or ``None`` if ambiguous. ``duplicate_native_ids`` (from ``_duplicate_message_native_ids``) is keyed - by the same surrogate-substituted form computed here, so membership is - always compared apples-to-apples. + by the same stripped, surrogate-substituted form computed here, so + membership is always compared apples-to-apples. """ - native_id = _sqlite_text(message.provider_message_id) + native_id = _normalized_message_native_id(message) if native_id in duplicate_native_ids: return None return native_id @@ -6798,6 +6883,48 @@ def _attachment_position(attachment: ParsedAttachment) -> int: return int.from_bytes(digest.digest()[:4], "big") +def _attachment_reference_positions( + attachments: Iterable[ParsedAttachment], + *, + occupied_positions: Iterable[int] = (), +) -> dict[int, int]: + """Return stable per-object reference positions without silent collisions. + + The historical four-byte position remains the primary identity so ordinary + archives retain their existing reference ids. When distinct attachment + identities share that truncated value, the canonical attachment-id order + keeps the first identity at the historical position and assigns the rest a + deterministic full-identity-derived position with collision probing. The + result is independent of parser/list order and is shared by write and + closure relink routes. + """ + attachments_by_identity: dict[str, list[ParsedAttachment]] = {} + for attachment in attachments: + attachments_by_identity.setdefault(_attachment_id("", attachment), []).append(attachment) + + groups: dict[int, list[tuple[str, list[ParsedAttachment]]]] = defaultdict(list) + for attachment_id, equivalent_attachments in attachments_by_identity.items(): + groups[_attachment_position(equivalent_attachments[0])].append((attachment_id, equivalent_attachments)) + + occupied = {int(position) for position in occupied_positions} + assigned: dict[int, int] = {} + for base_position, identity_group in sorted(groups.items()): + for collision_index, (attachment_id, equivalent_attachments) in enumerate(sorted(identity_group)): + if collision_index == 0 and base_position not in occupied: + position = base_position + else: + position = int.from_bytes( + hashlib.sha256(f"attachment-reference:{attachment_id}".encode()).digest()[:8], + "big", + ) & ((1 << 63) - 1) + while position in occupied: + position = (position + 1) & ((1 << 63) - 1) + occupied.add(position) + for attachment in equivalent_attachments: + assigned[id(attachment)] = position + return assigned + + def _acquire_attachment_blob( conn: sqlite3.Connection, attachment: ParsedAttachment, @@ -6823,22 +6950,32 @@ def _attachment_caption(attachment: ParsedAttachment) -> str | None: return attachment.caption -def _write_attachment_native_ids(conn: sqlite3.Connection, ref_id: str, attachment: ParsedAttachment) -> None: +def _attachment_native_id_values(attachment: ParsedAttachment) -> tuple[tuple[str, str], ...]: + """Return SQLite-sanitized typed native identities for one attachment.""" native_values = ( ("attachment", attachment.provider_attachment_id), ("file", attachment.provider_file_id), ("drive", attachment.provider_drive_id), ("url", _attachment_source_url(attachment)), ) - for id_kind, native_id in native_values: - if native_id: - conn.execute( - """ - INSERT OR IGNORE INTO attachment_native_ids (ref_id, id_kind, native_id) - VALUES (?, ?, ?) - """, - (ref_id, id_kind, _sqlite_text(native_id)), - ) + return tuple( + (id_kind, sanitized) + for id_kind, native_id in native_values + if native_id is not None + for sanitized in (_sqlite_text(native_id),) + if sanitized + ) + + +def _write_attachment_native_ids(conn: sqlite3.Connection, ref_id: str, attachment: ParsedAttachment) -> None: + for id_kind, native_id in _attachment_native_id_values(attachment): + conn.execute( + """ + INSERT OR IGNORE INTO attachment_native_ids (ref_id, id_kind, native_id) + VALUES (?, ?, ?) + """, + (ref_id, id_kind, _sqlite_text(native_id)), + ) def _hash_bytes(*parts: str) -> bytes: diff --git a/polylogue/storage/sqlite/lifecycle.py b/polylogue/storage/sqlite/lifecycle.py index 7855a80e49..eb3ee6ede6 100644 --- a/polylogue/storage/sqlite/lifecycle.py +++ b/polylogue/storage/sqlite/lifecycle.py @@ -886,6 +886,14 @@ class IndexDeltaDeclarationReport(TypedDict): ), ), ), + IndexDeltaDeclaration( + version=66, + # Native and positional message identities now use disjoint tagged + # namespaces. Existing generated ids are still valid read values, but + # every newly written message and dependent reference must be produced + # by raw replay under the new expression. + classes=(DerivedDeltaClass.SEMANTIC_REPARSE,), + ), ) diff --git a/polylogue/storage/sqlite/queries/raw_writes.py b/polylogue/storage/sqlite/queries/raw_writes.py index 2a80510d80..2025a31f0d 100644 --- a/polylogue/storage/sqlite/queries/raw_writes.py +++ b/polylogue/storage/sqlite/queries/raw_writes.py @@ -120,6 +120,28 @@ async def save_raw_session( (file_mtime_ms, record.source_path, record.raw_id, file_mtime_ms, record.source_path), ) + # ``raw_sessions`` and ``blob_refs`` are one durable acquisition contract. + # This async writer predates the typed source-tier writer and used to stop + # after inserting the raw row, leaving the payload invisible to blob GC and + # source-to-index reindex closure checks. Read the retained row back so a + # duplicate save cannot manufacture reference metadata from a stale caller + # record, then write the exact persisted identity. + cursor = await conn.execute( + "SELECT source_path, blob_hash, blob_size, acquired_at_ms FROM raw_sessions WHERE raw_id = ?", + (record.raw_id,), + ) + persisted = await cursor.fetchone() + if persisted is None: + raise RuntimeError(f"raw session disappeared before its blob reference was written: {record.raw_id}") + await conn.execute( + """ + INSERT OR REPLACE INTO blob_refs ( + blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms + ) VALUES (?, ?, 'raw_payload', ?, ?, ?) + """, + (persisted[1], record.raw_id, persisted[0], persisted[2], persisted[3]), + ) + if record.blob_publication_receipt_id is not None: await conn.execute( "DELETE FROM blob_publication_reservations WHERE publication_id = ? AND blob_hash = ?", diff --git a/tests/unit/cli/test_maintenance_registration.py b/tests/unit/cli/test_maintenance_registration.py index ceb287ac80..5270b27031 100644 --- a/tests/unit/cli/test_maintenance_registration.py +++ b/tests/unit/cli/test_maintenance_registration.py @@ -6,6 +6,7 @@ from click.testing import CliRunner from polylogue.cli.click_app import cli as root_cli +from polylogue.cli.commands.maintenance._blob_reference_closure import blob_reference_closure_command from polylogue.cli.commands.maintenance._hook_payload_ref_reconciliation import hook_payload_ref_reconcile_command from polylogue.cli.commands.maintenance._plan import plan_command from polylogue.cli.commands.maintenance._run import run_command @@ -106,6 +107,10 @@ def test_hook_payload_reconcile_is_click_command() -> None: assert isinstance(hook_payload_ref_reconcile_command, click.Command) +def test_blob_reference_closure_is_click_command() -> None: + assert isinstance(blob_reference_closure_command, click.Command) + + def test_maintenance_group_has_status() -> None: """maintenance group lists status as a subcommand (#1197).""" maintenance_group = _registered_maintenance_command() @@ -113,6 +118,7 @@ def test_maintenance_group_has_status() -> None: cmds = maintenance_group.list_commands(ctx) # type: ignore[attr-defined] assert "status" in cmds assert "hook-payload-ref-reconcile" in cmds + assert "blob-reference-closure" in cmds def test_maintenance_status_help_output() -> None: diff --git a/tests/unit/core/test_identity_law.py b/tests/unit/core/test_identity_law.py index e8e3887b82..c0a213d927 100644 --- a/tests/unit/core/test_identity_law.py +++ b/tests/unit/core/test_identity_law.py @@ -31,7 +31,7 @@ def test_native_message_id_ignores_position_fallback( variant_index: int, ) -> None: sid = session_id("codex", parent) - assert message_id(sid, native_id, position=position, variant_index=variant_index) == f"{sid}:{native_id.strip()}" + assert message_id(sid, native_id, position=position, variant_index=variant_index) == f"{sid}:n:{native_id.strip()}" @given(parent=_TOKEN, position=_POSITION, left_variant=_POSITION, right_variant=_POSITION) @@ -44,8 +44,8 @@ def test_no_native_message_id_uses_variant_index_for_collision_avoidance( sid = session_id("codex", parent) left = message_id(sid, None, position=position, variant_index=left_variant) right = message_id(sid, None, position=position, variant_index=right_variant) - assert left == f"{sid}:{position}.{left_variant}" - assert right == f"{sid}:{position}.{right_variant}" + assert left == f"{sid}:p:{position}.{left_variant}" + assert right == f"{sid}:p:{position}.{right_variant}" assert (left == right) is (left_variant == right_variant) @@ -60,7 +60,13 @@ def test_native_ids_are_opaque_and_may_contain_colons() -> None: mid = message_id(sid, "cascade:0:planner_response", position=0) assert sid == "antigravity-session:cascade:with:colon" - assert mid == "antigravity-session:cascade:with:colon:cascade:0:planner_response" + assert mid == "antigravity-session:cascade:with:colon:n:cascade:0:planner_response" + + +def test_native_and_positional_message_ids_are_disjoint() -> None: + sid = session_id("codex", "collision") + + assert message_id(sid, "0.0", position=99) != message_id(sid, None, position=0, variant_index=0) @pytest.mark.parametrize( diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index c80ee264f7..d664ee2534 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -64,6 +64,13 @@ def _seed_coherent_archive(root: Path) -> None: VALUES ('raw-1', 'fp', 'complete', 1, 100) """ ) + source_conn.execute( + """ + INSERT INTO blob_refs(blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms) + VALUES (?, 'raw-1', 'raw_payload', '/x', 10, 100) + """, + (bytes.fromhex(blob_hash),), + ) source_conn.commit() finally: source_conn.close() @@ -712,18 +719,6 @@ def test_blob_ref_with_no_referent_trips_blob_refs_liveness(tmp_path: Path) -> N def test_blob_refs_liveness_passes_on_coherent_archive(tmp_path: Path) -> None: _seed_coherent_archive(tmp_path) - conn = _connect(tmp_path / "source.db") - try: - conn.execute( - """ - INSERT INTO blob_refs(blob_hash, ref_id, ref_type, size_bytes, acquired_at_ms) - VALUES (?, 'raw-1', 'raw_payload', 10, 100) - """, - (b"a" * 32,), - ) - conn.commit() - finally: - conn.close() report = verify_archive(tmp_path, checks=("blob-refs-liveness",)) @@ -737,6 +732,25 @@ def test_blob_refs_liveness_passes_on_coherent_archive(tmp_path: Path) -> None: } +def test_blob_reference_closure_rejects_acquired_attachment_without_ref(tmp_path: Path) -> None: + _seed_coherent_archive(tmp_path) + conn = _connect(tmp_path / "index.db") + try: + conn.execute( + "INSERT INTO attachments (attachment_id, byte_count, blob_hash, acquisition_status, ref_count) " + "VALUES ('orphan-acquired', 1, ?, 'acquired', 0)", + (b"a" * 32,), + ) + conn.commit() + finally: + conn.close() + + report = verify_archive(tmp_path, checks=("blob-reference-closure",)) + check = _check(report, "blob-reference-closure") + assert check.status is OutcomeStatus.ERROR + assert check.evidence["acquired_attachment_missing_ref_count"] == 1 + + def test_attachment_blob_ref_joins_its_parent_raw_session(tmp_path: Path) -> None: _seed_coherent_archive(tmp_path) conn = _connect(tmp_path / "source.db") @@ -1590,6 +1604,7 @@ def test_reindex_acceptance_subset_is_satisfiable_from_index_only_root(tmp_path: "lineage-sanity": "test_dangling_resolved_dst_trips_lineage_sanity", "enum-superset-check": "test_missing_enum_value_trips_enum_superset_check", "blob-refs-liveness": "test_blob_ref_with_no_referent_trips_blob_refs_liveness", + "blob-reference-closure": "test_blob_reference_closure_rejects_acquired_attachment_without_ref", "blob-integrity": "test_full_blob_integrity_red_twin", "attachment-acquisition-debt": "test_acquired_unreachable_attachment_debt_is_blocking", "raw-failure-lifecycle": "test_raw_failure_lifecycle_mutation_red_twin_for_validation_failures", diff --git a/tests/unit/maintenance/test_blob_reference_closure.py b/tests/unit/maintenance/test_blob_reference_closure.py new file mode 100644 index 0000000000..7e87193d19 --- /dev/null +++ b/tests/unit/maintenance/test_blob_reference_closure.py @@ -0,0 +1,691 @@ +"""Production-route proofs for acquired blob-reference closure repair.""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +import aiosqlite +import pytest + +from polylogue.archive.message.roles import Role +from polylogue.core.enums import Provider +from polylogue.core.outcomes import OutcomeStatus +from polylogue.maintenance.archive_verification import ArchiveVerificationCheck +from polylogue.maintenance.blob_reference_closure import ( + BlobReferenceBlockerKind, + BlobReferenceClosureError, + plan_blob_reference_closure, + reconcile_blob_reference_closure, +) +from polylogue.pipeline.services.ingest_worker import IngestRecordResult, SessionWritePayload, ingest_record +from polylogue.sources.parsers.base import ParsedAttachment, ParsedMessage, ParsedSession +from polylogue.storage.attachment_relink import RawSessionParser +from polylogue.storage.blob_store import BlobStore +from polylogue.storage.runtime.raw.records import RawSessionRecord +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.write import ( + _attachment_id, + _attachment_position, + write_parsed_session_to_archive, +) +from polylogue.storage.sqlite.queries.attachment_records import search_attachment_identity_evidence_hits + +_PAYLOAD = { + "uuid": "closure-session-1", + "name": "Closure test", + "chat_messages": [ + { + "uuid": "m0", + "sender": "human", + "text": "here is a file", + "attachments": [ + { + "file_name": "notes.md", + "file_type": "text/markdown", + "file_size": 11, + "extracted_content": "hello notes", + } + ], + } + ], +} + + +def _conn(path: Path, tier: ArchiveTier) -> sqlite3.Connection: + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + initialize_archive_tier(conn, tier) + return conn + + +def _fixture(tmp_path: Path) -> tuple[Path, sqlite3.Connection, sqlite3.Connection, str]: + root = tmp_path + blob_store = BlobStore(root / "blob") + source = _conn(root / "source.db", ArchiveTier.SOURCE) + index = _conn(root / "index.db", ArchiveTier.INDEX) + payload = json.dumps(_PAYLOAD).encode() + blob_hash, blob_size = blob_store.write_from_bytes(payload) + source.execute( + "INSERT INTO raw_sessions (raw_id, origin, source_path, source_index, blob_hash, blob_size, acquired_at_ms) " + "VALUES (?, 'claude-ai-export', ?, 0, ?, ?, 100)", + ("raw-closure", "conversations.json", bytes.fromhex(blob_hash), blob_size), + ) + source.commit() + record = RawSessionRecord( + raw_id="raw-closure", + source_name=Provider.CLAUDE_AI.value, + payload_provider=Provider.CLAUDE_AI, + source_path="conversations.json", + source_index=0, + blob_size=blob_size, + blob_hash=blob_hash, + acquired_at="2026-01-01T00:00:00+00:00", + ) + parsed = ingest_record(record, str(root), "advisory", blob_root_str=str(blob_store.root)) + assert parsed.error is None + session = parsed.sessions[0] + attachment = session.parsed_session.attachments[0] + attachment_hash, attachment_size = blob_store.write_from_bytes(attachment.inline_bytes or b"") + write_parsed_session_to_archive( + index, + session.parsed_session, + raw_id=record.raw_id, + preacquired_attachment_blobs={id(attachment): (bytes.fromhex(attachment_hash), attachment_size, "acquired")}, + ) + attachment_id = str(index.execute("SELECT attachment_id FROM attachments").fetchone()[0]) + index.execute("DELETE FROM attachment_refs WHERE attachment_id = ?", (attachment_id,)) + index.execute("UPDATE attachments SET ref_count = 0 WHERE attachment_id = ?", (attachment_id,)) + index.commit() + return root, source, index, attachment_id + + +def _mapping_fixture( + tmp_path: Path, + *, + messages: list[ParsedMessage], + attachment: ParsedAttachment, + append_only: bool = False, + existing_messages: list[ParsedMessage] | None = None, + orphan_attachment: bool = True, +) -> tuple[Path, str, str, RawSessionParser]: + """Build a real closure archive with a parser result for one session.""" + root = tmp_path + blob_store = BlobStore(root / "blob") + source = _conn(root / "source.db", ArchiveTier.SOURCE) + index = _conn(root / "index.db", ArchiveTier.INDEX) + raw_bytes = b"closure mapping raw fixture" + raw_hash, raw_size = blob_store.write_from_bytes(raw_bytes) + source.execute( + "INSERT INTO raw_sessions (raw_id, origin, source_path, source_index, blob_hash, blob_size, acquired_at_ms) " + "VALUES ('raw-mapping', 'claude-ai-export', 'mapping.json', 0, ?, ?, 100)", + (bytes.fromhex(raw_hash), raw_size), + ) + source.commit() + + session = ParsedSession( + source_name=Provider.CLAUDE_AI, + provider_session_id="closure-mapping", + title="Closure mapping", + messages=messages, + attachments=[attachment], + ) + attachment_hash, attachment_size = blob_store.write_from_bytes(attachment.inline_bytes or b"") + if append_only: + if existing_messages is None: + raise AssertionError("append fixtures require existing messages") + write_parsed_session_to_archive( + index, + session.model_copy(update={"messages": existing_messages, "attachments": []}), + raw_id="raw-mapping-base", + ) + session_id = write_parsed_session_to_archive( + index, + session, + raw_id="raw-mapping", + merge_append=append_only, + preacquired_attachment_blobs={id(attachment): (bytes.fromhex(attachment_hash), attachment_size, "acquired")}, + ) + attachment_id = _attachment_id(session_id, attachment) + index.execute( + """ + INSERT OR IGNORE INTO attachments ( + attachment_id, display_name, media_type, byte_count, blob_hash, acquisition_status, ref_count + ) VALUES (?, ?, ?, ?, ?, 'acquired', 0) + """, + ( + attachment_id, + attachment.name, + attachment.mime_type, + attachment_size, + bytes.fromhex(attachment_hash), + ), + ) + if orphan_attachment: + index.execute("DELETE FROM attachment_refs WHERE attachment_id = ?", (attachment_id,)) + index.execute("UPDATE attachments SET ref_count = 0 WHERE attachment_id = ?", (attachment_id,)) + index.commit() + source.close() + index.close() + + def parse(_raw_record: RawSessionRecord) -> IngestRecordResult: + return IngestRecordResult( + raw_id="raw-mapping", + sessions=[ + SessionWritePayload( + session_id=session_id, + content_hash="mapping-fixture", + parsed_session=session, + append_only=append_only, + ) + ], + ) + + return root, session_id, attachment_id, parse + + +def _accept_backup(manifest: Path, _tier: ArchiveTier, *, connection: sqlite3.Connection) -> Path: + assert connection.execute("SELECT 1").fetchone() == (1,) + return manifest + + +def _apply_mapping_fixture( + monkeypatch: pytest.MonkeyPatch, + root: Path, + parser: RawSessionParser, +) -> None: + manifest = root / "verified-backup" / "manifest.json" + receipt = root / "receipts" / "closure.jsonl" + monkeypatch.setattr( + "polylogue.maintenance.blob_reference_closure.validate_migration_backup_manifest", _accept_backup + ) + monkeypatch.setattr( + "polylogue.maintenance.blob_reference_closure.validate_backup_manifest_covers_derived_tier", _accept_backup + ) + reconcile_blob_reference_closure( + root, + backup_manifest=manifest, + receipt_path=receipt, + dry_run=False, + raw_session_parser=parser, + ) + + +def _legacy_collision_fixture( + tmp_path: Path, +) -> tuple[Path, str, str, str, RawSessionParser]: + root = tmp_path + blob_store = BlobStore(root / "blob") + source = _conn(root / "source.db", ArchiveTier.SOURCE) + index = _conn(root / "index.db", ArchiveTier.INDEX) + raw_bytes = b"legacy collision raw fixture" + raw_hash, raw_size = blob_store.write_from_bytes(raw_bytes) + source.execute( + "INSERT INTO raw_sessions (raw_id, origin, source_path, source_index, blob_hash, blob_size, acquired_at_ms) " + "VALUES ('raw-collision', 'claude-ai-export', 'collision.json', 0, ?, ?, 100)", + (bytes.fromhex(raw_hash), raw_size), + ) + source.commit() + attachments = [ + ParsedAttachment( + provider_attachment_id="cert-collision-50449", + message_provider_id="m1", + name="first.txt", + mime_type="text/plain", + inline_bytes=b"first", + ), + ParsedAttachment( + provider_attachment_id="cert-collision-111329", + message_provider_id="m1", + name="second.txt", + mime_type="text/plain", + inline_bytes=b"second", + ), + ] + session = ParsedSession( + source_name=Provider.CLAUDE_AI, + provider_session_id="legacy-collision", + messages=[ParsedMessage(provider_message_id="m1", role=Role.USER, text="files", position=0)], + attachments=attachments, + ) + preacquired: dict[int, tuple[bytes | None, int, str]] = {} + for attachment in attachments: + blob_hash, blob_size = blob_store.write_from_bytes(attachment.inline_bytes or b"") + preacquired[id(attachment)] = (bytes.fromhex(blob_hash), blob_size, "acquired") + session_id = write_parsed_session_to_archive( + index, + session, + raw_id="raw-collision", + preacquired_attachment_blobs=preacquired, + ) + base_position = _attachment_position(attachments[0]) + assert base_position == _attachment_position(attachments[1]) + kept = index.execute( + "SELECT attachment_id FROM attachment_refs WHERE message_id = ? AND position = ?", + (f"{session_id}:n:m1", base_position), + ).fetchone() + assert kept is not None + orphan = next( + _attachment_id(session_id, attachment) + for attachment in attachments + if _attachment_id(session_id, attachment) != kept[0] + ) + index.execute("DELETE FROM attachment_refs WHERE attachment_id = ?", (orphan,)) + index.execute("UPDATE attachments SET ref_count = 0 WHERE attachment_id = ?", (orphan,)) + index.commit() + source.close() + index.close() + + def parse(_raw_record: RawSessionRecord) -> IngestRecordResult: + return IngestRecordResult( + raw_id="raw-collision", + sessions=[ + SessionWritePayload( + session_id=session_id, + content_hash="legacy-collision", + parsed_session=session, + append_only=False, + ) + ], + ) + + return root, session_id, kept[0], orphan, parse + + +def test_closure_repairs_idless_attachment_by_authoritative_message_position( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + attachment = ParsedAttachment( + provider_attachment_id="idless-position", + message_position=1, + name="position.txt", + mime_type="text/plain", + size_bytes=8, + inline_bytes=b"position", + ) + root, session_id, attachment_id, parser = _mapping_fixture( + tmp_path, + messages=[ + ParsedMessage(provider_message_id="first", role=Role.USER, text="first", position=0), + ParsedMessage(provider_message_id="second", role=Role.ASSISTANT, text="second", position=1), + ], + attachment=attachment, + ) + + dry = reconcile_blob_reference_closure(root, raw_session_parser=parser) + assert dry.applied is False + assert dry.plan.attachment_candidates[0].message_id == f"{session_id}:n:second" + with sqlite3.connect(root / "index.db") as conn: + assert conn.execute("SELECT COUNT(*) FROM attachment_refs").fetchone()[0] == 0 + + _apply_mapping_fixture(monkeypatch, root, parser) + with sqlite3.connect(root / "index.db") as conn: + ref = conn.execute( + "SELECT message_id FROM attachment_refs WHERE attachment_id = ?", (attachment_id,) + ).fetchone() + assert ref == (f"{session_id}:n:second",) + + +def test_closure_recovers_legacy_attachment_position_collision_atomically( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root, session_id, kept_id, orphan_id, parser = _legacy_collision_fixture(tmp_path) + + plan = plan_blob_reference_closure(root, raw_session_parser=parser) + candidate = next(item for item in plan.attachment_candidates if item.attachment_id == orphan_id) + assert candidate.message_id == f"{session_id}:n:m1" + assert candidate.position != _attachment_position(ParsedAttachment(provider_attachment_id="cert-collision-50449")) + + _apply_mapping_fixture(monkeypatch, root, parser) + with sqlite3.connect(root / "index.db") as conn: + rows = conn.execute( + "SELECT attachment_id, position FROM attachment_refs WHERE message_id = ? ORDER BY position", + (f"{session_id}:n:m1",), + ).fetchall() + assert {row[0] for row in rows} == {kept_id, orphan_id} + assert len({row[1] for row in rows}) == 2 + + +@pytest.mark.asyncio +async def test_closure_relink_restores_explicit_ids_for_production_reads_and_search( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + attachment = ParsedAttachment( + provider_attachment_id="closure-provider-id", + provider_file_id="closure-file-id", + provider_drive_id="closure-drive-id", + message_position=0, + name="explicit.txt", + mime_type="text/plain", + size_bytes=8, + source_url="https://example.test/explicit.txt", + inline_bytes=b"explicit", + ) + root, session_id, attachment_id, parser = _mapping_fixture( + tmp_path, + messages=[ParsedMessage(provider_message_id="m1", role=Role.USER, text="explicit attachment", position=0)], + attachment=attachment, + ) + + _apply_mapping_fixture(monkeypatch, root, parser) + + async with aiosqlite.connect(root / "index.db") as conn: + conn.row_factory = aiosqlite.Row + native_rows = await ( + await conn.execute( + """ + SELECT ani.id_kind, ani.native_id + FROM attachment_native_ids ani + JOIN attachment_refs ar ON ar.ref_id = ani.ref_id + WHERE ar.attachment_id = ? + ORDER BY ani.id_kind + """, + (attachment_id,), + ) + ).fetchall() + assert [(row["id_kind"], row["native_id"]) for row in native_rows] == [ + ("attachment", "closure-provider-id"), + ("drive", "closure-drive-id"), + ("file", "closure-file-id"), + ("url", "https://example.test/explicit.txt"), + ] + + for native_id in ("closure-provider-id", "closure-file-id", "closure-drive-id"): + hits = await search_attachment_identity_evidence_hits(conn, query=native_id, limit=10) + assert len(hits) == 1 + assert hits[0].session_id == session_id + assert hits[0].match_surface == "attachment" + + +def test_production_append_attaches_idless_message_at_max_position_plus_one(tmp_path: Path) -> None: + attachment = ParsedAttachment( + provider_attachment_id="append-position", + message_position=0, + name="append.txt", + mime_type="text/plain", + size_bytes=6, + inline_bytes=b"append", + ) + root, session_id, attachment_id, _parser = _mapping_fixture( + tmp_path, + messages=[ParsedMessage(provider_message_id="", role=Role.ASSISTANT, text="appended", position=0)], + existing_messages=[ParsedMessage(provider_message_id="", role=Role.USER, text="older", position=0)], + attachment=attachment, + append_only=True, + orphan_attachment=False, + ) + + with sqlite3.connect(root / "index.db") as conn: + messages = conn.execute( + "SELECT message_id, native_id, position FROM messages WHERE session_id = ? ORDER BY position", + (session_id,), + ).fetchall() + assert messages == [ + (f"{session_id}:p:0.0", None, 0), + (f"{session_id}:p:1.0", None, 1), + ] + ref = conn.execute( + "SELECT message_id FROM attachment_refs WHERE attachment_id = ?", (attachment_id,) + ).fetchone() + assert ref == (f"{session_id}:p:1.0",) + + +def test_closure_relinks_idless_append_attachment_to_existing_tail( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Closure must reuse the already-materialized append tail position. + + The raw append payload still carries its relative position ``0``. The + current index already contains that message at position ``1``. Applying + ``MAX(position) + 1`` again would probe position ``2`` and incorrectly + report the acquired attachment as unrecoverable. + """ + attachment = ParsedAttachment( + provider_attachment_id="append-orphan", + message_position=0, + name="append-orphan.txt", + mime_type="text/plain", + size_bytes=6, + inline_bytes=b"append", + ) + root, session_id, attachment_id, parser = _mapping_fixture( + tmp_path, + messages=[ParsedMessage(provider_message_id="", role=Role.ASSISTANT, text="appended", position=0)], + existing_messages=[ParsedMessage(provider_message_id="", role=Role.USER, text="older", position=0)], + attachment=attachment, + append_only=True, + ) + + plan = plan_blob_reference_closure(root, raw_session_parser=parser) + assert plan.attachment_candidates[0].attachment_id == attachment_id + assert plan.attachment_candidates[0].message_id == f"{session_id}:p:1.0" + + _apply_mapping_fixture(monkeypatch, root, parser) + with sqlite3.connect(root / "index.db") as conn: + ref = conn.execute( + "SELECT message_id FROM attachment_refs WHERE attachment_id = ?", (attachment_id,) + ).fetchone() + assert ref == (f"{session_id}:p:1.0",) + + +def test_closure_fails_closed_for_whitespace_duplicate_native_id( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + attachment = ParsedAttachment( + provider_attachment_id="duplicate-native-position", + message_provider_id=" duplicate ", + message_position=None, + name="duplicate.txt", + mime_type="text/plain", + size_bytes=9, + inline_bytes=b"duplicate", + ) + root, session_id, attachment_id, parser = _mapping_fixture( + tmp_path, + messages=[ + ParsedMessage(provider_message_id="duplicate", role=Role.USER, text="first", position=0), + ParsedMessage(provider_message_id=" duplicate ", role=Role.ASSISTANT, text="second", position=1), + ], + attachment=attachment, + ) + + plan = plan_blob_reference_closure(root, raw_session_parser=parser) + assert not plan.attachment_candidates + assert any(blocker.object_id == attachment_id for blocker in plan.blockers) + + _apply_mapping_fixture(monkeypatch, root, parser) + with sqlite3.connect(root / "index.db") as conn: + ref = conn.execute( + "SELECT message_id FROM attachment_refs WHERE attachment_id = ?", (attachment_id,) + ).fetchone() + assert ref is None + + +def test_plan_is_complete_and_typed_for_deterministic_and_irreparable_rows(tmp_path: Path) -> None: + root, source, index, attachment_id = _fixture(tmp_path) + index.execute( + "INSERT INTO attachments (attachment_id, display_name, byte_count, blob_hash, acquisition_status, ref_count) " + "VALUES ('irreparable', 'ghost.bin', 5, ?, 'acquired', 0)", + (b"g" * 32,), + ) + index.commit() + plan = plan_blob_reference_closure(root) + assert [candidate.raw_id for candidate in plan.raw_candidates] == ["raw-closure"] + assert [candidate.attachment_id for candidate in plan.attachment_candidates] == [attachment_id] + assert any( + blocker.object_id == "irreparable" and blocker.kind is BlobReferenceBlockerKind.ATTACHMENT_NO_AUTHORITATIVE_RAW + for blocker in plan.blockers + ) + source.close() + index.close() + + +def test_plan_excludes_unfetched_orphans_from_closure_scope(tmp_path: Path) -> None: + root, source, index, _attachment_id = _fixture(tmp_path) + index.execute( + "INSERT INTO attachments (attachment_id, display_name, byte_count, blob_hash, acquisition_status, ref_count) " + "VALUES (?, ?, ?, ?, 'unfetched', 0)", + ("f" * 64, "not acquired", 0, b"z" * 32), + ) + index.commit() + + plan = plan_blob_reference_closure(root) + + assert plan.attachment_orphan_count == 1 + assert all(blocker.object_id != "f" * 64 for blocker in plan.blockers) + assert all(candidate.attachment_id != "f" * 64 for candidate in plan.attachment_candidates) + source.close() + index.close() + + +def test_dry_run_does_not_write_and_apply_requires_both_tier_backup(tmp_path: Path) -> None: + root, source, index, _attachment_id = _fixture(tmp_path) + dry = reconcile_blob_reference_closure(root) + assert dry.applied is False + assert dry.plan.candidate_count == 2 + assert source.execute("SELECT COUNT(*) FROM blob_refs").fetchone()[0] == 0 + assert index.execute("SELECT COUNT(*) FROM attachment_refs").fetchone()[0] == 0 + with pytest.raises(BlobReferenceClosureError, match="backup manifest"): + reconcile_blob_reference_closure(root, dry_run=False, receipt_path=tmp_path / "receipt.jsonl") + source.close() + index.close() + + +def test_apply_writes_only_exact_canonical_refs_and_is_receipted( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root, source, index, attachment_id = _fixture(tmp_path) + manifest = tmp_path / "verified-backup" / "manifest.json" + receipt = tmp_path / "receipts" / "closure.jsonl" + validated: list[ArchiveTier] = [] + + def accept(_manifest: Path, tier: ArchiveTier, *, connection: sqlite3.Connection) -> Path: + assert connection.execute("SELECT 1").fetchone() == (1,) + validated.append(tier) + return _manifest + + monkeypatch.setattr("polylogue.maintenance.blob_reference_closure.validate_migration_backup_manifest", accept) + monkeypatch.setattr( + "polylogue.maintenance.blob_reference_closure.validate_backup_manifest_covers_derived_tier", accept + ) + report = reconcile_blob_reference_closure(root, backup_manifest=manifest, receipt_path=receipt, dry_run=False) + assert report.raw_repaired_count == 1 + assert report.attachment_repaired_count == 1 + assert validated == [ArchiveTier.SOURCE, ArchiveTier.INDEX] + assert source.execute("SELECT COUNT(*) FROM blob_refs WHERE ref_type = 'raw_payload'").fetchone()[0] == 1 + assert ( + index.execute("SELECT COUNT(*) FROM attachment_refs WHERE attachment_id = ?", (attachment_id,)).fetchone()[0] + == 1 + ) + assert receipt.exists() + source.close() + index.close() + + +def test_apply_rolls_back_source_when_index_reference_write_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root, source, index, _attachment_id = _fixture(tmp_path) + index.execute( + """ + CREATE TRIGGER fail_closure_attachment_insert + BEFORE INSERT ON attachment_refs + BEGIN + SELECT RAISE(ABORT, 'forced closure index failure'); + END + """ + ) + index.commit() + source.close() + index.close() + manifest = tmp_path / "verified-backup" / "manifest.json" + receipt = tmp_path / "receipts" / "closure-rollback.jsonl" + monkeypatch.setattr( + "polylogue.maintenance.blob_reference_closure.validate_migration_backup_manifest", _accept_backup + ) + monkeypatch.setattr( + "polylogue.maintenance.blob_reference_closure.validate_backup_manifest_covers_derived_tier", _accept_backup + ) + + with pytest.raises(sqlite3.IntegrityError, match="forced closure index failure"): + reconcile_blob_reference_closure( + root, + backup_manifest=manifest, + receipt_path=receipt, + dry_run=False, + ) + + with sqlite3.connect(root / "source.db") as source_after, sqlite3.connect(root / "index.db") as index_after: + assert source_after.execute("SELECT COUNT(*) FROM blob_refs").fetchone()[0] == 0 + assert index_after.execute("SELECT COUNT(*) FROM attachment_refs").fetchone()[0] == 0 + phases = [json.loads(line)["phase"] for line in receipt.read_text().splitlines()] + assert phases[-1] == "aborted" + + +def test_closure_relink_sanitizes_unpaired_surrogate_native_ids( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + attachment = ParsedAttachment( + provider_attachment_id="closure-surrogate-\ud800", + provider_file_id="closure-file-\udfff", + message_provider_id="m1", + name="surrogate.txt", + mime_type="text/plain", + size_bytes=8, + inline_bytes=b"surrogate", + ) + root, _session_id, attachment_id, parser = _mapping_fixture( + tmp_path, + messages=[ParsedMessage(provider_message_id="m1", role=Role.USER, text="surrogate", position=0)], + attachment=attachment, + ) + plan = plan_blob_reference_closure(root, raw_session_parser=parser) + assert plan.attachment_candidates[0].native_ids == ( + ("attachment", "closure-surrogate-�"), + ("file", "closure-file-�"), + ) + + _apply_mapping_fixture(monkeypatch, root, parser) + with sqlite3.connect(root / "index.db") as conn: + native_ids = { + row[0] + for row in conn.execute( + """ + SELECT ani.native_id + FROM attachment_native_ids ani + JOIN attachment_refs ar ON ar.ref_id = ani.ref_id + WHERE ar.attachment_id = ? + """, + (attachment_id,), + ).fetchall() + } + assert native_ids == {"closure-surrogate-�", "closure-file-�"} + + +def test_integrity_check_fails_when_exact_raw_reference_is_tampered(tmp_path: Path) -> None: + root, source, index, _attachment_id = _fixture(tmp_path) + source.execute( + "INSERT INTO blob_refs (blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms) " + "SELECT blob_hash, raw_id, 'raw_payload', source_path, blob_size, acquired_at_ms FROM raw_sessions" + ) + source.execute( + "INSERT INTO blob_refs (blob_hash, ref_id, ref_type, source_path, size_bytes, acquired_at_ms) " + "SELECT ?, raw_id, 'raw_payload', source_path, blob_size, acquired_at_ms FROM raw_sessions", + (b"x" * 32,), + ) + source.commit() + from polylogue.maintenance.archive_verification import verify_archive + + check = next( + check + for check in verify_archive(root, checks=("blob-reference-closure",)).checks + if check.name == "blob-reference-closure" + ) + assert isinstance(check, ArchiveVerificationCheck) + assert check.status is OutcomeStatus.ERROR + assert check.evidence["raw_missing_exact_count"] == 1 + source.close() + index.close() diff --git a/tests/unit/material_protocol/v1/fixture.py b/tests/unit/material_protocol/v1/fixture.py index 1af630e589..b095e1568a 100644 --- a/tests/unit/material_protocol/v1/fixture.py +++ b/tests/unit/material_protocol/v1/fixture.py @@ -188,13 +188,13 @@ def build_small_session_material() -> SessionMaterial: fidelity_gaps = ( FidelityGapInput( scope="attachment", - record_id="claude-code-session:demo-session-1:msg-2:attachment:0", + record_id="claude-code-session:demo-session-1:n:msg-2:attachment:0", gap_kind="unavailable_attachment_bytes", detail="attachment referenced by the provider export but bytes were never fetched", ), FidelityGapInput( scope="message", - record_id="claude-code-session:demo-session-1:msg-3", + record_id="claude-code-session:demo-session-1:n:msg-3", gap_kind="missing_timestamp", detail="provider omitted occurred_at; position ordinal is authoritative", ), diff --git a/tests/unit/material_protocol/v1/test_round_trip.py b/tests/unit/material_protocol/v1/test_round_trip.py index bd840e1d6c..f70f7af029 100644 --- a/tests/unit/material_protocol/v1/test_round_trip.py +++ b/tests/unit/material_protocol/v1/test_round_trip.py @@ -149,7 +149,7 @@ def test_fidelity_gaps_are_declared_in_the_manifest() -> None: def test_resolve_anchor_reads_one_record_without_a_full_scan() -> None: _material, encoded = _encode_small() - record = resolve_anchor(encoded.manifest, encoded.segments, "claude-code-session:demo-session-1:msg-2:0") + record = resolve_anchor(encoded.manifest, encoded.segments, "claude-code-session:demo-session-1:n:msg-2:0") assert record["kind"] == "block" assert record["tool_id"] == "tool-ok-1" diff --git a/tests/unit/pipeline/test_archive_write.py b/tests/unit/pipeline/test_archive_write.py index 5013a56ac9..eb2f4e6b53 100644 --- a/tests/unit/pipeline/test_archive_write.py +++ b/tests/unit/pipeline/test_archive_write.py @@ -233,7 +233,7 @@ async def test_persists_compaction_session_event_summary(async_backend: SQLiteBa assert row is not None assert row["event_type"] == "compaction" assert row["summary"] == "Older turns compacted" - assert row["source_message_id"] == f"{session_id}:msg-1" + assert row["source_message_id"] == f"{session_id}:n:msg-1" async def test_agent_reasoning_event_filtered_but_sibling_orphan_type_kept( @@ -338,7 +338,7 @@ async def test_projects_agent_policy_event_into_typed_columns(async_backend: SQL assert row["approval_policy"] == "on-request" assert row["sandbox_policy"] == "workspace-write" assert row["network_policy"] == "restricted" - assert row["source_message_id"] == f"{session_id}:msg-1" + assert row["source_message_id"] == f"{session_id}:n:msg-1" # --------------------------------------------------------------------------- @@ -441,7 +441,7 @@ async def test_whitespace_only_native_message_id_falls_back_and_writes_blocks(as ) assert message_row["native_id"] is None - assert message_row["message_id"] == f"{session_id}:0.0" + assert message_row["message_id"] == f"{session_id}:p:0.0" assert block_count == 1 @@ -480,7 +480,7 @@ class from the same rebuild crash).""" ) assert message_row["native_id"] == "�tail" - assert message_row["message_id"] == f"{session_id}:�tail" + assert message_row["message_id"] == f"{session_id}:n:�tail" assert block_count == 1 diff --git a/tests/unit/pipeline/test_ingest_batch.py b/tests/unit/pipeline/test_ingest_batch.py index b0bb69be56..c82c58d811 100644 --- a/tests/unit/pipeline/test_ingest_batch.py +++ b/tests/unit/pipeline/test_ingest_batch.py @@ -745,6 +745,70 @@ def test_write_session_append_mode_preserves_existing_messages(tmp_path: Path) - assert (stats["message_count"], stats["word_count"]) == (2, 2) +def test_write_session_append_dedupes_whitespace_padded_native_id(tmp_path: Path) -> None: + """Append dedupe must use the normalized id before INSERT OR REPLACE. + + The writer stores ``" msg-1 "`` as ``"msg-1"``. If the append gate + compares the raw provider id, it admits the duplicate and the generated + message id makes INSERT OR REPLACE overwrite the original message. + """ + with open_connection(tmp_path / "index.db") as conn: + initial = _session_data( + "codex-session:append-whitespace-id", + content_hash="hash-initial", + message_tuples=[ + _message_tuple( + "msg-1", + "codex-session:append-whitespace-id", + role="user", + text="original message", + content_hash="msg-original", + sort_key=1.0, + ) + ], + ) + duplicate = _session_data( + "codex-session:append-whitespace-id", + content_hash="hash-append-duplicate", + message_tuples=[ + _message_tuple( + " msg-1 ", + "codex-session:append-whitespace-id", + role="assistant", + text="replacement must not win", + content_hash="msg-replacement", + sort_key=1.0, + ) + ], + append_only=True, + ) + + changed_initial, _ = _write_session(conn, initial) + changed_duplicate, counts_duplicate = _write_session(conn, duplicate) + conn.commit() + + rows = conn.execute( + "SELECT native_id, position FROM messages WHERE session_id = ?", + ("codex-session:append-whitespace-id",), + ).fetchall() + text = conn.execute( + """ + SELECT b.text + FROM blocks b + JOIN messages m ON m.message_id = b.message_id + WHERE m.session_id = ? AND b.block_type = 'text' + """, + ("codex-session:append-whitespace-id",), + ).fetchone() + + assert changed_initial is True + assert changed_duplicate is False + assert counts_duplicate["skipped_messages"] == 1 + assert [(row["native_id"], row["position"]) for row in rows] == [("msg-1", 0)] + assert text is not None + assert text["text"] == "original message" + + def test_write_session_append_no_delta_refreshes_raw_link(tmp_path: Path) -> None: with open_connection(tmp_path / "index.db") as conn: initial = _session_data( diff --git a/tests/unit/storage/test_archive_tiers_write.py b/tests/unit/storage/test_archive_tiers_write.py index 99da6c34d5..4f6fca60d3 100644 --- a/tests/unit/storage/test_archive_tiers_write.py +++ b/tests/unit/storage/test_archive_tiers_write.py @@ -109,6 +109,34 @@ def test_message_content_hash_tracks_same_identity_body_edits(tmp_path: Path) -> conn.close() +def test_writer_separates_native_and_positional_message_identity(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="native-positional-identity", + messages=[ + ParsedMessage(provider_message_id="0.0", role=Role.USER, text="native", position=1), + ParsedMessage(provider_message_id="", role=Role.ASSISTANT, text="positional", position=0), + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + + rows = conn.execute( + "SELECT message_id, native_id FROM messages WHERE session_id = ? ORDER BY position", + (session_id,), + ).fetchall() + assert [(row["message_id"], row["native_id"]) for row in rows] == [ + (f"{session_id}:p:0.0", None), + (f"{session_id}:n:0.0", "0.0"), + ] + + write_parsed_session_to_archive(conn, session.model_copy(update={"messages": list(reversed(session.messages))})) + assert conn.execute("SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)).fetchone()[0] == 2 + finally: + conn.close() + + def _block_hash(conn: sqlite3.Connection, session_id: str, native_message_id: str, position: int) -> bytes: row = conn.execute( """ @@ -324,10 +352,10 @@ def test_archive_tiers_writer_materializes_codex_session(tmp_path: Path) -> None assert envelope.session_id == "codex-session:codex-session-1" assert envelope.origin == "codex-session" - assert envelope.active_leaf_message_id == "codex-session:codex-session-1:a1" + assert envelope.active_leaf_message_id == "codex-session:codex-session-1:n:a1" assert [message.message_id for message in envelope.messages] == [ - "codex-session:codex-session-1:u1", - "codex-session:codex-session-1:a1", + "codex-session:codex-session-1:n:u1", + "codex-session:codex-session-1:n:a1", ] assert envelope.messages[1].is_active_leaf is True assert [block.block_type for block in envelope.messages[1].blocks] == ["tool_use", "tool_result"] @@ -341,7 +369,7 @@ def test_archive_tiers_writer_materializes_codex_session(tmp_path: Path) -> None "is_error": 0, "exit_code": 0, } - assert search_archive_blocks(conn, "focused") == ["codex-session:codex-session-1:u1:0"] + assert search_archive_blocks(conn, "focused") == ["codex-session:codex-session-1:n:u1:0"] def test_archive_tiers_writer_splits_provider_user_from_authored_user_counts(tmp_path: Path) -> None: @@ -520,13 +548,13 @@ def test_archive_tiers_writer_does_not_collapse_duplicate_message_native_ids(tmp (0, "user", None), (1, "assistant", None), ] - assert [row["message_id"] for row in message_rows] == [f"{session_id}:0.0", f"{session_id}:1.0"] + assert [row["message_id"] for row in message_rows] == [f"{session_id}:p:0.0", f"{session_id}:p:1.0"] assert [(row["message_id"], row["text"]) for row in block_rows] == [ - (f"{session_id}:0.0", "first"), - (f"{session_id}:1.0", "second"), + (f"{session_id}:p:0.0", "first"), + (f"{session_id}:p:1.0", "second"), ] assert session_row["message_count"] == 2 - assert session_row["active_leaf_message_id"] == f"{session_id}:1.0" + assert session_row["active_leaf_message_id"] == f"{session_id}:p:1.0" def test_archive_tiers_writer_normalizes_duplicate_idless_active_leaves_by_position(tmp_path: Path) -> None: @@ -654,16 +682,16 @@ def test_archive_tiers_writer_preserves_chatgpt_branch_variants(tmp_path: Path) ).fetchall() assert [row["message_id"] for row in rows] == [ - "chatgpt-export:chatgpt-branch-1:u1", - "chatgpt-export:chatgpt-branch-1:a-old", - "chatgpt-export:chatgpt-branch-1:a-new", + "chatgpt-export:chatgpt-branch-1:n:u1", + "chatgpt-export:chatgpt-branch-1:n:a-old", + "chatgpt-export:chatgpt-branch-1:n:a-new", ] - assert rows[1]["parent_message_id"] == "chatgpt-export:chatgpt-branch-1:u1" - assert rows[2]["parent_message_id"] == "chatgpt-export:chatgpt-branch-1:u1" + assert rows[1]["parent_message_id"] == "chatgpt-export:chatgpt-branch-1:n:u1" + assert rows[2]["parent_message_id"] == "chatgpt-export:chatgpt-branch-1:n:u1" assert [(row["is_active_path"], row["is_active_leaf"]) for row in rows] == [(1, 0), (0, 0), (1, 1)] assert ( read_archive_session_envelope(conn, session_id).active_leaf_message_id - == "chatgpt-export:chatgpt-branch-1:a-new" + == "chatgpt-export:chatgpt-branch-1:n:a-new" ) @@ -715,12 +743,12 @@ def test_archive_tiers_writer_uses_identity_law_for_messages_without_native_ids( envelope = read_archive_session_envelope(conn, session_id) assert [message.message_id for message in envelope.messages] == [ - "codex-session:codex-generated-ids:0.0", - "codex-session:codex-generated-ids:1.0", + "codex-session:codex-generated-ids:p:0.0", + "codex-session:codex-generated-ids:p:1.0", ] assert [block.block_id for message in envelope.messages for block in message.blocks] == [ - "codex-session:codex-generated-ids:0.0:0", - "codex-session:codex-generated-ids:1.0:0", + "codex-session:codex-generated-ids:p:0.0:0", + "codex-session:codex-generated-ids:p:1.0:0", ] @@ -1088,7 +1116,7 @@ def test_archive_tiers_writer_materializes_supported_session_events(tmp_path: Pa assert [dict(row) for row in rows] == [ { "event_id": f"{session_id}:0", - "source_message_id": f"{session_id}:m1", + "source_message_id": f"{session_id}:n:m1", "source_message_provider_id": "m1", "position": 0, "event_type": "compaction", @@ -1249,7 +1277,7 @@ def test_archive_tiers_writer_materializes_provider_usage_events(tmp_path: Path) ).fetchone() assert dict(usage) == { "usage_event_id": f"{session_id}:usage:0", - "source_message_id": f"{session_id}:m1", + "source_message_id": f"{session_id}:n:m1", "position": 0, "provider_event_type": "token_count", "last_input_tokens": 11, @@ -1849,14 +1877,14 @@ def test_provider_usage_events_append_preserves_prior_history(tmp_path: Path) -> assert [dict(row) for row in rows] == [ { "usage_event_id": f"{session_id}:usage:0", - "source_message_id": f"{session_id}:m1", + "source_message_id": f"{session_id}:n:m1", "position": 0, "total_input_tokens": 10, "total_output_tokens": 5, }, { "usage_event_id": f"{session_id}:usage:1", - "source_message_id": f"{session_id}:m2", + "source_message_id": f"{session_id}:n:m2", "position": 1, "total_input_tokens": 30, "total_output_tokens": 15, @@ -3450,7 +3478,7 @@ def test_archive_tiers_writer_replacement_clears_old_projection_rows(tmp_path: P } assert dict(session_row) == {"git_branch": None, "git_repository_url": None, "commit_hash": None} assert search_archive_blocks(conn, "old") == [] - assert search_archive_blocks(conn, "replacement") == [f"{session_id}:m1:0"] + assert search_archive_blocks(conn, "replacement") == [f"{session_id}:n:m1:0"] assert ( conn.execute( """ @@ -3518,7 +3546,7 @@ def test_archive_tiers_writer_materializes_attachments_and_refs(tmp_path: Path) attachment_hash.update(part.encode("utf-8", errors="surrogatepass")) attachment_hash.update(b"\0") attachment_id = attachment_hash.hexdigest() - message_id = f"{session_id}:m1" + message_id = f"{session_id}:n:m1" attachment = conn.execute( """ @@ -3571,6 +3599,36 @@ def test_archive_tiers_writer_materializes_attachments_and_refs(tmp_path: Path) } +def test_writer_sanitizes_unpaired_surrogates_in_attachment_native_ids(tmp_path: Path) -> None: + conn = _connect(tmp_path / "index.db") + try: + session = ParsedSession( + source_name=Provider.CHATGPT, + provider_session_id="attachment-surrogate-native-id", + messages=[ParsedMessage(provider_message_id="m1", role=Role.USER, text="surrogate")], + attachments=[ + ParsedAttachment( + provider_attachment_id="attachment-\ud800", + provider_file_id="file-\udfff", + message_provider_id="m1", + name="surrogate.txt", + mime_type="text/plain", + ) + ], + ) + session_id = write_parsed_session_to_archive(conn, session) + ref_id = conn.execute("SELECT ref_id FROM attachment_refs WHERE session_id = ?", (session_id,)).fetchone()[0] + native_ids = { + row[0] + for row in conn.execute( + "SELECT native_id FROM attachment_native_ids WHERE ref_id = ?", (ref_id,) + ).fetchall() + } + assert native_ids == {"attachment-�", "file-�"} + finally: + conn.close() + + # --------------------------------------------------------------------------- # ITEM 1: sessions.instructions_text round-trip # --------------------------------------------------------------------------- @@ -4711,6 +4769,62 @@ def test_reingest_restores_attachment_ref_for_reinjected_message(tmp_path: Path) conn.close() +def test_full_replace_preserves_distinct_attachments_with_colliding_positions(tmp_path: Path) -> None: + """A truncated positional hash must not let one attachment replace another. + + These are the certification witness ids: both produce the historical + four-byte ``_attachment_position`` value ``0xf5f6e7cd``. + """ + conn = _connect(tmp_path / "index.db") + try: + attachments = [ + ParsedAttachment( + provider_attachment_id="cert-collision-50449", + message_provider_id="m1", + name="first.txt", + mime_type="text/plain", + size_bytes=5, + ), + ParsedAttachment( + provider_attachment_id="cert-collision-111329", + message_provider_id="m1", + name="second.txt", + mime_type="text/plain", + size_bytes=6, + ), + ] + session = ParsedSession( + source_name=Provider.CHATGPT, + provider_session_id="attachment-position-collision", + messages=[ParsedMessage(provider_message_id="m1", role=Role.USER, text="two files")], + attachments=attachments, + ) + session_id = write_parsed_session_to_archive(conn, session) + + def rows() -> list[tuple[str, int]]: + return [ + (str(row["attachment_id"]), int(row["position"])) + for row in conn.execute( + "SELECT attachment_id, position FROM attachment_refs WHERE session_id = ? ORDER BY attachment_id", + (session_id,), + ).fetchall() + ] + + first_rows = rows() + assert len(first_rows) == 2 + assert len({attachment_id for attachment_id, _position in first_rows}) == 2 + assert len({position for _attachment_id, position in first_rows}) == 2 + + write_parsed_session_to_archive( + conn, + session.model_copy(update={"attachments": list(reversed(attachments))}), + force_replace=True, + ) + assert rows() == first_rows + finally: + conn.close() + + def test_reingest_recomputes_message_flags_and_hash_after_block_restoration(tmp_path: Path) -> None: """PR #3413 review (P2 write.py:2334): when a still-present message drops a tool-use block that field-path union restores, the message's diff --git a/tests/unit/storage/test_attachment_acquisition.py b/tests/unit/storage/test_attachment_acquisition.py index 22f6556ba2..566a2cd6e0 100644 --- a/tests/unit/storage/test_attachment_acquisition.py +++ b/tests/unit/storage/test_attachment_acquisition.py @@ -216,6 +216,13 @@ def test_claude_extracted_attachment_content_is_acquired(tmp_path: Path, monkeyp assert remote["acquisition_status"] == "unfetched" assert remote["blob_hash"] is None assert remote["byte_count"] == 1024 + ref = conn.execute( + "SELECT attachment_id, session_id, message_id FROM attachment_refs WHERE attachment_id = " + "(SELECT attachment_id FROM attachments WHERE display_name = 'notes.md')" + ).fetchone() + assert ref is not None + assert ref[1] == "claude-ai-export:claude-attachment-session" + assert ref[2] == "claude-ai-export:claude-attachment-session:m0" @pytest.mark.parametrize("payload", [b"must be reserved first", b""], ids=["nonempty", "empty"]) diff --git a/tests/unit/storage/test_attachment_relink.py b/tests/unit/storage/test_attachment_relink.py index adcf511d55..0f56666080 100644 --- a/tests/unit/storage/test_attachment_relink.py +++ b/tests/unit/storage/test_attachment_relink.py @@ -18,6 +18,7 @@ from polylogue.core.enums import Provider from polylogue.pipeline.services.ingest_worker import ingest_record from polylogue.storage.attachment_relink import ( + UnrecoverableAttachmentReason, plan_orphaned_attachment_relink, relink_orphaned_attachments, ) @@ -179,11 +180,12 @@ async def test_orphaned_attachment_is_reachable_via_production_read_path_after_r """Anti-vacuity: after relink, the production ``get_attachments`` read path -- the one every session/message attachment surface (MCP, CLI ``read --view``, transcript view) goes through -- can see it. Reverting - the ``INSERT OR REPLACE INTO attachment_refs`` write in - ``relink_orphaned_attachments`` makes this assertion fail: the row would + the attachment-ref insert in ``relink_orphaned_attachments`` makes this + assertion fail: the row would still be present in ``attachments`` (visible via a direct SELECT) but ``get_attachments`` INNER JOINs ``attachment_refs``, so a still-ref-less - row can never be returned. + row can never be returned. Reverting the attachment-ref insert or the + typed native-id restoration makes this production read assertion fail. """ import aiosqlite @@ -256,6 +258,7 @@ def test_orphan_with_no_matching_raw_is_reported_unrecoverable_not_guessed(tmp_p assert sum(plan.unrecoverable_reason_counts.values()) == 1 assert plan.unrecoverable_samples[0].attachment_id == "ghost-attachment-id" assert "no raw session" in plan.unrecoverable_samples[0].reason + assert plan.unrecoverable_samples[0].reason_kind is UnrecoverableAttachmentReason.NO_AUTHORITATIVE_RAW exec_result = relink_orphaned_attachments( index_conn, source_conn, archive_root=tmp_path, blob_root=blob_store.root, dry_run=False diff --git a/tests/unit/storage/test_raw.py b/tests/unit/storage/test_raw.py index 09e2dd6c2a..f19e37c473 100644 --- a/tests/unit/storage/test_raw.py +++ b/tests/unit/storage/test_raw.py @@ -60,6 +60,14 @@ async def test_save_raw_session_new(self, backend: SQLiteBackend) -> None: result = await backend.save_raw_session(record) assert result is True + async with backend._get_connection() as conn: + cursor = await conn.execute( + "SELECT COUNT(*) FROM blob_refs WHERE ref_type = 'raw_payload' AND ref_id = ?", + (record.raw_id,), + ) + row = await cursor.fetchone() + assert row is not None + assert row[0] == 1 async def test_repository_update_raw_state_uses_source_tier(self, tmp_path: Path) -> None: initialize_archive_database(tmp_path / "source.db", ArchiveTier.SOURCE)