diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index d3491e781c..037b108803 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -1090,6 +1090,34 @@ def to_dict(self) -> dict[str, object]: "--backup-manifest /realm/staging/polylogue-backup/manifest.json", ), ), + CommandSpec( + "workspace raw-quarantine-group-dedup-apply", + "workspace", + "Promote one representative raw per fully-quarantined byte-identical (source_path, blob_hash) group.", + "devtools.raw_quarantine_group_dedup_apply", + use_when=( + "polylogue-zm4w8 (measured live 2026-08-03): 1,777 raw_sessions rows (22.2 GiB) among " + "the codex-session quarantine backlog are pure redundant duplicates -- same source_path " + "AND same blob_hash as another raw_sessions row -- where EVERY member of the group is " + "still quarantined (no indexed twin anywhere), invisible to " + "raw-byte-duplicate-supersession-apply (which only matches a quarantined raw against an " + "already-INDEXED twin). Default is dry-run; --apply requires --backup-manifest pointing " + "at a verified source-tier backup (polylogue backup --output-dir --verify). " + "Materializes exactly one representative raw per group through the real ingest pipeline " + "(ParsingService.parse_from_raw -> write_parsed_session_to_archive -> " + "refresh_session_insights_bulk) so it becomes a genuine indexed session, then marks the " + "rest revision_authority='byte_proven' with an immutable per-row receipt " + "(raw_quarantine_group_dedup_receipts) pointing at the promoted representative's raw_id " + "and new session_id. Never deletes blobs or runs GC/VACUUM -- those are separate, later " + "steps." + ), + examples=( + "devtools workspace raw-quarantine-group-dedup-apply", + "devtools workspace raw-quarantine-group-dedup-apply --json", + "devtools workspace raw-quarantine-group-dedup-apply --apply " + "--backup-manifest /realm/staging/polylogue-backup/manifest.json", + ), + ), CommandSpec( "workspace binary-artifact-sweep", "workspace", diff --git a/devtools/raw_quarantine_group_dedup_apply.py b/devtools/raw_quarantine_group_dedup_apply.py new file mode 100644 index 0000000000..e95f50d8d3 --- /dev/null +++ b/devtools/raw_quarantine_group_dedup_apply.py @@ -0,0 +1,136 @@ +"""Actuator: promote one representative raw per fully-quarantined byte-identical group. + +polylogue-zm4w8: 1,777 raw_sessions rows (22.2 GiB, measured 2026-08-03) among +the codex-session quarantine backlog are pure redundant duplicates -- same +``source_path`` AND same ``blob_hash`` as another raw_sessions row, with +every member of the group still quarantined (no indexed twin anywhere) -- +invisible to ``raw-byte-duplicate-supersession-apply``, which only matches a +quarantined raw against an already-INDEXED twin. + +Default mode is dry-run (report only, zero mutation). Pass ``--apply`` to +actually promote a representative per group and mark the rest, which +additionally requires ``--backup-manifest`` pointing at a verified backup +manifest for the ``source`` tier (see ``polylogue backup --output-dir +--verify``). This never runs blob GC or ``VACUUM`` -- that is a separate, +later, operator-invoked step. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +from pathlib import Path +from typing import TextIO + +from polylogue.maintenance.raw_quarantine_group_dedup_apply import ( + RawQuarantineGroupDedupApplyError, + apply_raw_quarantine_group_dedup, +) +from polylogue.paths import archive_root as default_archive_root + + +def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--archive-root", + type=Path, + default=None, + help="Archive root to operate on; defaults to the configured active archive root.", + ) + parser.add_argument( + "--limit", + type=int, + default=None, + help="Cap the number of (source_path, blob_hash) groups classified/promoted (unbounded by default).", + ) + parser.add_argument( + "--apply", + action="store_true", + help="Actually promote representatives and mark duplicates. Without this flag, nothing is mutated.", + ) + parser.add_argument( + "--backup-manifest", + type=Path, + default=None, + help="Verified backup manifest for the source tier. Required with --apply.", + ) + parser.add_argument("--json", action="store_true", help="Emit the report as JSON.") + args = parser.parse_args(argv) + + root = args.archive_root if args.archive_root is not None else default_archive_root() + + try: + report = asyncio.run( + apply_raw_quarantine_group_dedup( + root, + backup_manifest=args.backup_manifest, + limit=args.limit, + dry_run=not args.apply, + ) + ) + except (RawQuarantineGroupDedupApplyError, FileNotFoundError) as exc: + print(f"refused: {exc}", file=stdout) + return 1 + + if args.json: + payload = { + "applied": report.applied, + "scanned_count": report.scanned_count, + "group_count": report.group_count, + "already_resolved_group_count": report.already_resolved_group_count, + "promoted_count": report.promoted_count, + "marked_duplicate_count": report.marked_duplicate_count, + "marked_duplicate_bytes": report.marked_duplicate_bytes, + "promotions": [ + { + "source_path": promotion.source_path, + "blob_size": promotion.blob_size, + "representative_raw_id": promotion.representative_raw_id, + "representative_session_id": promotion.representative_session_id or None, + "duplicate_raw_ids": list(promotion.duplicate_raw_ids), + } + for promotion in report.promotions + ], + "backup_manifest": str(report.backup_manifest) if report.backup_manifest is not None else None, + } + print(json.dumps(payload, indent=2, sort_keys=True), file=stdout) + return 0 + + def _gib(byte_count: int) -> str: + return f"{byte_count / (1024**3):.2f} GiB" + + mode = "APPLIED" if report.applied else "dry-run (no mutation performed -- pass --apply to promote)" + print(f"mode: {mode}", file=stdout) + print(f"quarantined, source_path-bearing rows scanned: {report.scanned_count}", file=stdout) + print( + f"fully-quarantined duplicate groups {'processed' if report.applied else 'found'}: {report.group_count}", + file=stdout, + ) + print( + f"already-resolved groups skipped (indexed twin or non-quarantined member elsewhere): " + f"{report.already_resolved_group_count}", + file=stdout, + ) + print( + f"{'materialized' if report.applied else 'would materialize'} representative(s): {report.promoted_count}", + file=stdout, + ) + print( + f"{'marked' if report.applied else 'would mark'} duplicate(s) byte_proven: " + f"{report.marked_duplicate_count:>7} ({_gib(report.marked_duplicate_bytes)})", + file=stdout, + ) + if report.applied: + print(f"backup manifest used: {report.backup_manifest}", file=stdout) + print( + "Each marked duplicate has an immutable receipt in " + "raw_quarantine_group_dedup_receipts. No blob GC or VACUUM was run -- " + "that is a separate, later step.", + file=stdout, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/devtools.md b/docs/devtools.md index d081c2f77a..850c30bec7 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -253,6 +253,7 @@ These are the commands worth remembering during normal repo work: | `devtools workspace raw-live-source-reconciliation` | Classify quarantined raw evidence against its live source file's current bytes. | | `devtools workspace raw-live-source-reconciliation-apply` | Promote quarantined raw evidence proven correct by live-source verification. | | `devtools workspace raw-membership-writeback-apply` | Propagate already-decided membership verdicts onto raw_sessions.revision_authority. | +| `devtools workspace raw-quarantine-group-dedup-apply` | Promote one representative raw per fully-quarantined byte-identical (source_path, blob_hash) group. | | `devtools workspace read-package` | Render a declarative package of Polylogue read artifacts. | | `devtools workspace scale-regression` | Run the seeded large-archive scale-regression probe. | | `devtools workspace tasks` | Record and query local agent task execution history. | diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index 87e4348469..12b5526466 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -1347,6 +1347,96 @@ def _check_excluded_cursor_vocabulary_honesty(archive_root: Path, sample_limit: ) +# --------------------------------------------------------------------------- +# Check: raw quarantine group dedup (polylogue-zm4w8) +# --------------------------------------------------------------------------- + + +def _check_raw_quarantine_group_dedup(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: + """No raw_sessions row is an unindexed byte-identical duplicate of another sharing its source_path. + + polylogue-zm4w8: a raw_sessions row is flagged when it belongs to a + ``(source_path, blob_hash)`` group of more than one quarantined row AND + no member of that group (nor any other raw sharing that blob_hash + anywhere) already has a materialized ``index.db`` session or a + non-quarantined ``revision_authority``. This is the residual, + genuinely-unresolved population: content acquired more than once that + was never chosen as its group's representative and materialized -- + distinct from ``source-index-coverage``'s ``byte_dup_of_indexed_count`` + (which requires an *already-indexed* twin) and from what + ``raw-byte-duplicate-supersession-apply`` (the actuator for that + already-indexed case) can catch by construction. The one-shot + ``raw-quarantine-group-dedup-apply`` actuator resolves a flagged group by + materializing exactly one representative raw and marking the rest + ``byte_proven`` -- once run, this check should report clean, and stays + part of the registry as the standing regression guard against the + pattern recurring. + """ + from polylogue.storage.raw_quarantine_group_dedup import plan_raw_quarantine_group_dedup + + source_path = _tier_path(archive_root, ArchiveTier.SOURCE) + index_path = _resolve_index_path(archive_root) + if not source_path.exists() or not index_path.exists(): + return _skip_check("raw-quarantine-group-dedup", "source.db or index.db not present") + + try: + source_conn = _open_ro(source_path) + except sqlite3.Error as exc: + return _error_check("raw-quarantine-group-dedup", f"could not open source.db: {exc}", exc=exc) + + try: + index_conn = _open_ro(index_path) + except sqlite3.Error as exc: + source_conn.close() + return _error_check("raw-quarantine-group-dedup", f"could not open index.db: {exc}", exc=exc) + + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn) + except sqlite3.Error as exc: + return _error_check("raw-quarantine-group-dedup", f"could not read source/index tiers: {exc}", exc=exc) + finally: + index_conn.close() + source_conn.close() + + group_count = len(plan.groups) + duplicate_count = plan.duplicate_count + sample = [ + { + "source_path": group.source_path, + "representative_raw_id": group.representative_raw_id, + "duplicate_raw_ids": list(group.duplicate_raw_ids), + } + for group in plan.groups[:sample_limit] + ] + + status = OutcomeStatus.ERROR if group_count else OutcomeStatus.OK + summary = ( + f"{group_count:,} fully-quarantined duplicate group(s), {duplicate_count:,} unindexed duplicate row(s) " + f"({_gib_str(plan.duplicate_bytes)}); run raw-quarantine-group-dedup-apply" + if group_count + else f"no fully-quarantined byte-identical duplicate groups ({plan.scanned_count:,} quarantined row(s) scanned)" + ) + return ArchiveVerificationCheck( + name="raw-quarantine-group-dedup", + status=status, + summary=summary, + count=duplicate_count, + details=[f"group:{group.source_path}" for group in plan.groups[:sample_limit]], + evidence={ + "scanned_count": plan.scanned_count, + "group_count": group_count, + "duplicate_count": duplicate_count, + "duplicate_bytes": plan.duplicate_bytes, + "already_resolved_group_count": plan.already_resolved_group_count, + "group_sample": sample, + }, + ) + + +def _gib_str(byte_count: int) -> str: + return f"{byte_count / (1024**3):.2f} GiB" + + # --------------------------------------------------------------------------- # Check: stalled append-cursor freshness (polylogue-2qrx) # --------------------------------------------------------------------------- @@ -1838,6 +1928,13 @@ def _check_user_tier_refs(archive_root: Path, sample_limit: int) -> ArchiveVerif _check_stalled_append_cursor_freshness, ArchiveVerificationCheckClass.LIVENESS, ), + ArchiveVerificationCheckSpec( + "raw-quarantine-group-dedup", + "No raw_sessions row is an unindexed byte-identical duplicate of another sharing its " + "source_path within a fully-quarantined group (polylogue-zm4w8).", + _check_raw_quarantine_group_dedup, + ArchiveVerificationCheckClass.STATE_INVARIANT, + ), ) ARCHIVE_VERIFICATION_CHECK_NAMES: tuple[str, ...] = tuple(spec.name for spec in ARCHIVE_VERIFICATION_CHECKS) diff --git a/polylogue/maintenance/raw_quarantine_group_dedup_apply.py b/polylogue/maintenance/raw_quarantine_group_dedup_apply.py new file mode 100644 index 0000000000..4996180226 --- /dev/null +++ b/polylogue/maintenance/raw_quarantine_group_dedup_apply.py @@ -0,0 +1,438 @@ +"""Promote one representative raw per fully-quarantined byte-identical group; mark the rest proven duplicates. + +polylogue-zm4w8: :mod:`polylogue.storage.raw_quarantine_group_dedup` classifies +``(source_path, blob_hash)`` groups among quarantined ``raw_sessions`` rows +where every member is quarantined and none has an indexed twin anywhere -- +genuine repeated acquisitions of the same real content, never materialized. +This module is the "act" half, following the same safety pattern as every +other actuator in this family (``raw_byte_duplicate_supersession_apply``, +``raw_live_source_reconciliation_apply``, ``raw_membership_writeback_apply``): + +* Dry-run by default (report only, zero mutation). +* ``dry_run=False`` requires a verified backup manifest for the ``source`` + tier, validated with the same gate durable-tier schema migrations use. +* Every marked-duplicate row gets an immutable receipt in + ``raw_quarantine_group_dedup_receipts`` recording exactly which + representative raw and materialized session it was superseded by. +* Never deletes blobs or runs GC/VACUUM -- that is a separate, later, + operator-invoked step. + +Unlike every sibling actuator, this one is genuinely two-phase per group and +cannot run as a single locked ``source.db`` transaction end to end: promoting +the representative raw to a real indexed session goes through the production +async ingest pipeline (``ParsingService.parse_from_raw`` -> +``write_parsed_session_to_archive``, which owns its own transaction against +``index.db``, plus ``refresh_session_insights_bulk``). Only once that +materialization has genuinely landed (re-verified by reading ``index.db`` +for a ``sessions`` row at that raw_id, not merely trusted from the ingest +call's return value) does phase two run for THAT group: a single locked +``source.db`` transaction that re-verifies each duplicate is still +quarantined and marks it, exactly like the sibling actuators' single-phase +writes. + +Materialize-then-mark runs **per group**, not batched across the whole plan +(explicit design decision, CodeRabbit review on PR #3697): if group N's +representative fails to materialize or raises, groups 1..N-1 already +recorded their receipts and are durably done, and group N+1 onward still get +attempted. Batching phase two until after every group's phase one completed +would mean one failing group anywhere in the plan leaves every +already-materialized group's duplicates permanently unreachable -- the +classifier excludes a group from its universe the moment any member's +``blob_hash`` has an indexed twin (see +``polylogue.storage.raw_quarantine_group_dedup``), so an already-materialized +representative with un-marked duplicates would never be reclassified as +"in scope" again by a later run. + +If materialization produces anything other than **exactly one** indexed +session for a group's representative (parse error, refused write, +non-session content, or a multi-session capture file materializing more than +one ``sessions`` row from a single raw), that whole group is left untouched +-- never guessed at. The exactly-one invariant is a deliberate, conservative +choice: this actuator's whole premise is "one raw, one piece of content, one +representative session" and a raw that turns out to violate that (a grouped +multi-session payload) doesn't fit the receipt schema's singular +``representative_session_id`` column without inventing ambiguous +multi-session semantics -- skipping is safer than guessing which of several +materialized sessions is "the" representative. + +If materialization *raises* for a group's representative (a genuine parse +error, a database error, ...), that exception is caught and logged, and that +one group is skipped exactly like the no-session case -- it does NOT abort +the run. Every earlier group's promotion is already durably committed (phase +two per group, above), and every later group still gets attempted. A single +bad row in an otherwise-clean ~1,800-group backlog must not block the rest. +""" + +from __future__ import annotations + +import sqlite3 +import time +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING + +from polylogue.config import Config +from polylogue.logging import get_logger +from polylogue.maintenance.offline_guard import offline_maintenance_block_reason +from polylogue.paths import render_root +from polylogue.storage.raw_quarantine_group_dedup import ( + RawQuarantineGroup, + RawQuarantineGroupDedupPlan, + plan_raw_quarantine_group_dedup, +) +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import validate_migration_backup_manifest + +if TYPE_CHECKING: + from polylogue.pipeline.services.parsing import ParsingService + from polylogue.storage.sqlite.async_sqlite import SQLiteBackend + +logger = get_logger(__name__) + +TOOL_VERSION = "raw-quarantine-group-dedup-apply-v1" + + +class RawQuarantineGroupDedupApplyError(RuntimeError): + """Raised when applying a quarantine-group-dedup promotion is refused.""" + + +@dataclass(frozen=True, slots=True) +class RawQuarantineGroupDedupPromotion: + """One group's outcome: the representative materialized, and the duplicates marked (or planned).""" + + source_path: str + blob_hash: bytes + blob_size: int + representative_raw_id: str + #: Empty string in a dry-run report (nothing was actually materialized). + representative_session_id: str + duplicate_raw_ids: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class RawQuarantineGroupDedupApplyReport: + scanned_count: int + group_count: int + already_resolved_group_count: int + promotions: tuple[RawQuarantineGroupDedupPromotion, ...] + applied: bool + backup_manifest: Path | None = None + + @property + def promoted_count(self) -> int: + return len(self.promotions) + + @property + def marked_duplicate_count(self) -> int: + return sum(len(promotion.duplicate_raw_ids) for promotion in self.promotions) + + @property + def marked_duplicate_bytes(self) -> int: + return sum(promotion.blob_size * len(promotion.duplicate_raw_ids) for promotion in self.promotions) + + +def _offline_config(archive_root: Path) -> Config: + return Config(archive_root=archive_root, render_root=render_root(), sources=[]) + + +def _checkpoint_live_tier(conn: sqlite3.Connection) -> None: + """Checkpoint the WAL and fail closed if it could not be fully truncated. + + ``PRAGMA wal_checkpoint(TRUNCATE)`` always returns one row + ``(busy, log, checkpointed)``. ``busy=1`` means another connection held a + lock that blocked the checkpoint -- the WAL was NOT truncated, and a + subsequent backup-manifest fingerprint check would silently attest + against a tier that still has uncheckpointed frames. Only ``busy=0`` is a + genuine success. + """ + try: + row = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + except sqlite3.Error as exc: + raise RawQuarantineGroupDedupApplyError("could not checkpoint source.db before backup validation") from exc + if row is None: + raise RawQuarantineGroupDedupApplyError("could not checkpoint source.db before backup validation") + if int(row[0]) != 0: + raise RawQuarantineGroupDedupApplyError( + "source.db WAL checkpoint was blocked by another connection; retry when the tier is idle" + ) + + +async def _materialize_group_representative( + parser: ParsingService, + backend: SQLiteBackend, + group: RawQuarantineGroup, +) -> str | None: + """Materialize a group's representative raw; return its session_id iff exactly one session resulted. + + Returns ``None`` (the whole group is left untouched by the caller) when + materialization produces zero sessions or produces MORE than one session + for this raw_id (a multi-session capture file) -- see the module + docstring for why more-than-one is treated the same as zero rather than + guessed at. Propagates any exception ``parse_from_raw`` raises; the + caller is responsible for catching it so one bad group doesn't abort the + rest of the run (see the module docstring). + """ + from polylogue.pipeline.services.ingest_batch import refresh_session_insights_bulk + + await parser.parse_from_raw(raw_ids=[group.representative_raw_id], force_write=True) + async with backend.connection() as conn: + cursor = await conn.execute( + "SELECT session_id FROM sessions WHERE raw_id = ? ORDER BY session_id", + (group.representative_raw_id,), + ) + session_rows = list(await cursor.fetchall()) + if len(session_rows) != 1: + # Zero: parse error, refused write, non-session content. More than + # one: a multi-session capture file -- this actuator's whole premise + # is one raw/one representative session, so an ambiguous + # materialization is left untouched rather than guessed at. + return None + representative_session_id = str(session_rows[0][0]) + await refresh_session_insights_bulk(backend, [representative_session_id]) + return representative_session_id + + +def _mark_group_duplicates( + source_db: Path, + backup_manifest: Path, + group: RawQuarantineGroup, + representative_session_id: str, +) -> RawQuarantineGroupDedupPromotion | None: + """Mark one group's duplicates 'byte_proven' with a receipt, in its own locked transaction. + + Runs per group (not batched across the whole plan) so a later group's + materialization failure can never leave an earlier, already-materialized + group's duplicates permanently unmarked -- see the module docstring. + Returns ``None`` only in the defensive case where every duplicate in the + group is no longer quarantined by the time this transaction's lock is + held (the representative's materialization still stands regardless). + """ + write_conn = sqlite3.connect(source_db) + try: + _checkpoint_live_tier(write_conn) + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=write_conn) + + write_conn.execute("BEGIN IMMEDIATE") + try: + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=write_conn) + marked_at_ms = int(time.time() * 1000) + marked_any = False + for duplicate_raw_id in group.duplicate_raw_ids: + existing_cursor = write_conn.execute( + "SELECT blob_size FROM raw_sessions WHERE raw_id = ? AND revision_authority = 'quarantined'", + (duplicate_raw_id,), + ) + existing = existing_cursor.fetchone() + if existing is None: + # Defensive: no longer quarantined under this same + # locked transaction's own read -- skip rather than + # assert, exactly like the sibling actuators. + continue + update_cursor = write_conn.execute( + """ + UPDATE raw_sessions + SET revision_authority = 'byte_proven' + WHERE raw_id = ? AND revision_authority = 'quarantined' + """, + (duplicate_raw_id,), + ) + if update_cursor.rowcount != 1: + continue + write_conn.execute( + """ + INSERT INTO raw_quarantine_group_dedup_receipts ( + raw_id, source_path, blob_hash, blob_size, + representative_raw_id, representative_session_id, + promoted_at_ms, tool_version, backup_manifest_path, detail + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + duplicate_raw_id, + group.source_path, + group.blob_hash, + int(existing[0]), + group.representative_raw_id, + representative_session_id, + marked_at_ms, + TOOL_VERSION, + str(backup_manifest), + "", + ), + ) + marked_any = True + + quick_check = write_conn.execute("PRAGMA quick_check").fetchone() + if quick_check is None or str(quick_check[0]).lower() != "ok": + raise RawQuarantineGroupDedupApplyError( + f"source.db quick_check failed after promotion: {quick_check!r}" + ) + except Exception: + if write_conn.in_transaction: + write_conn.rollback() + raise + else: + write_conn.commit() + finally: + write_conn.close() + + if not marked_any: + return None + return RawQuarantineGroupDedupPromotion( + source_path=group.source_path, + blob_hash=group.blob_hash, + blob_size=group.blob_size, + representative_raw_id=group.representative_raw_id, + representative_session_id=representative_session_id, + duplicate_raw_ids=group.duplicate_raw_ids, + ) + + +def _dry_run_report(plan: RawQuarantineGroupDedupPlan) -> RawQuarantineGroupDedupApplyReport: + promotions = tuple( + RawQuarantineGroupDedupPromotion( + source_path=group.source_path, + blob_hash=group.blob_hash, + blob_size=group.blob_size, + representative_raw_id=group.representative_raw_id, + representative_session_id="", + duplicate_raw_ids=group.duplicate_raw_ids, + ) + for group in plan.groups + ) + return RawQuarantineGroupDedupApplyReport( + scanned_count=plan.scanned_count, + group_count=len(plan.groups), + already_resolved_group_count=plan.already_resolved_group_count, + promotions=promotions, + applied=False, + ) + + +async def apply_raw_quarantine_group_dedup( + archive_root: Path, + *, + backup_manifest: Path | None = None, + limit: int | None = None, + dry_run: bool = True, +) -> RawQuarantineGroupDedupApplyReport: + """Classify quarantined groups, materialize one representative per group, mark the rest. + + ``dry_run=True`` (the default) never opens a write transaction on either + tier and never runs the ingest pipeline. It runs the same classifier a + real apply would and reports what it would do. + + ``dry_run=False`` requires ``backup_manifest``. Re-classifies live + (read-only) once, then processes each group in turn: materialize the + representative through the real production ingest pipeline, then -- + immediately, only for that group -- re-verify and mark its duplicates + inside their own locked ``source.db`` transaction. See the module + docstring for why this runs per group rather than as one batched pass or + a single atomic transaction like the sibling actuators. + """ + source_db = archive_root / "source.db" + if not source_db.exists(): + raise FileNotFoundError(f"no source.db at {source_db}") + + from polylogue.storage.archive_identity import resolve_active_index_path + + index_db = resolve_active_index_path(archive_root) + if not index_db.exists(): + raise FileNotFoundError(f"no index.db at {index_db}") + + if dry_run: + source_conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{index_db}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn, limit=limit) + finally: + source_conn.close() + index_conn.close() + return _dry_run_report(plan) + + if backup_manifest is None: + raise RawQuarantineGroupDedupApplyError( + "applying raw-quarantine-group-dedup requires a verified backup manifest (--backup-manifest)" + ) + + config = _offline_config(archive_root) + if reason := offline_maintenance_block_reason(config, active=True, dry_run=False): + raise RawQuarantineGroupDedupApplyError(reason) + + precheck_conn = sqlite3.connect(source_db) + try: + _checkpoint_live_tier(precheck_conn) + validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=precheck_conn) + finally: + precheck_conn.close() + + source_conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{index_db}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn, limit=limit) + finally: + source_conn.close() + index_conn.close() + + if not plan.groups: + return RawQuarantineGroupDedupApplyReport( + scanned_count=plan.scanned_count, + group_count=0, + already_resolved_group_count=plan.already_resolved_group_count, + promotions=(), + applied=True, + backup_manifest=backup_manifest, + ) + + from polylogue.pipeline.services.parsing import ParsingService + from polylogue.storage.repository import SessionRepository + from polylogue.storage.sqlite import create_backend + + backend = create_backend(db_path=config.db_path) + repository = SessionRepository(backend=backend) + parser = ParsingService(repository=repository, archive_root=archive_root, config=config) + + promotions: list[RawQuarantineGroupDedupPromotion] = [] + try: + for group in plan.groups: + try: + representative_session_id = await _materialize_group_representative(parser, backend, group) + except Exception: + # One bad group (a genuine parse/database error) must not + # abort the rest of an otherwise-clean run -- every earlier + # group's promotion is already durably committed below, and + # every later group still gets attempted. See module docstring. + logger.exception( + "raw-quarantine-group-dedup: materializing representative %s failed; leaving group at %s untouched", + group.representative_raw_id, + group.source_path, + ) + continue + if representative_session_id is None: + # Zero or more-than-one indexed session for this raw -- leave + # the whole group untouched rather than guessing (see + # _materialize_group_representative's docstring). + continue + + promotion = _mark_group_duplicates(source_db, backup_manifest, group, representative_session_id) + if promotion is not None: + promotions.append(promotion) + finally: + await backend.close() + + return RawQuarantineGroupDedupApplyReport( + scanned_count=plan.scanned_count, + group_count=len(plan.groups), + already_resolved_group_count=plan.already_resolved_group_count, + promotions=tuple(promotions), + applied=True, + backup_manifest=backup_manifest, + ) + + +__all__ = [ + "TOOL_VERSION", + "RawQuarantineGroupDedupApplyError", + "RawQuarantineGroupDedupApplyReport", + "RawQuarantineGroupDedupPromotion", + "apply_raw_quarantine_group_dedup", +] diff --git a/polylogue/storage/raw_quarantine_group_dedup.py b/polylogue/storage/raw_quarantine_group_dedup.py new file mode 100644 index 0000000000..48665cd6f8 --- /dev/null +++ b/polylogue/storage/raw_quarantine_group_dedup.py @@ -0,0 +1,219 @@ +"""Read-only classification: fully-quarantined raw_sessions groups sharing (source_path, blob_hash). + +polylogue-zm4w8 (measured live 2026-08-03, source.db mode=ro): of 5,203 +quarantined ``codex-session`` raw_sessions rows (45.73 GB), 3,426 distinct +``(source_path, blob_hash)`` pairs exist but 1,777 rows are pure redundant +duplicates -- same ``source_path`` AND same ``blob_hash`` as another already- +counted row -- 22.19 GB reclaimable. Sample: one file has NINE separate +``raw_id`` rows, all byte-identical, all ``revision_kind='unknown'``, +``revision_authority='quarantined'``. + +This is a distinct gap from :mod:`polylogue.storage.raw_byte_duplicate_supersession` +(polylogue-6753s): that module matches a quarantined row against an +**already-indexed** twin (a different ``raw_id`` with a materialized +``sessions`` row in ``index.db``). The population here never has an indexed +twin at all -- every member of a qualifying group starts out quarantined, +with nothing yet materialized for any of them. Confirmed live: ZERO of these +duplicate ``blob_hash`` values have any non-quarantined twin anywhere in +``raw_sessions``, so the existing actuator's classifier (whose universe is +exactly "quarantined rows with an indexed twin") returns zero candidates for +this class by construction. + +This module answers exactly one question per ``(source_path, blob_hash)`` +group among quarantined rows: does more than one raw_sessions row share this +exact source_path and blob_hash, and does NONE of them (nor any other raw +sharing that blob_hash, regardless of source_path) already have a +materialized session in ``index.db`` or a non-quarantined +``revision_authority``? If so, this group is a genuine unresolved cluster of +repeated acquisitions of the same content -- real, legitimate content that +was simply never chosen as the group's representative and materialized. + +Deterministic representative selection: the lowest (lexicographically +smallest) ``raw_id`` in each group is the representative a corresponding +actuator would materialize; the rest are the group's duplicates. This +mirrors ``raw_byte_duplicate_supersession``'s own "first (lowest) id wins" +precedent for a tie with no other governing signal. + +:mod:`polylogue.maintenance.raw_quarantine_group_dedup_apply` is the "act" +half: it promotes the representative through the real materialization path +(``ParsingService.parse_from_raw`` -> ``write_parsed_session_to_archive`` -> +``refresh_session_insights_bulk``) and marks the rest ``byte_proven`` with a +receipt, exactly the same safety pattern as every other actuator in this +family (dry-run by default, verified-backup-required-to-apply, immutable +receipt, never touches blob storage or runs GC). + +This module itself is **strictly read-only**: it never mutates ``source.db`` +or ``index.db``, and both connections are safe to open ``mode=ro``. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class RawQuarantineGroup: + """One (source_path, blob_hash) group of >1 fully-quarantined byte-identical raws.""" + + source_path: str + blob_hash: bytes + blob_size: int + #: All raw_ids in the group, sorted ascending -- deterministic. + raw_ids: tuple[str, ...] + + @property + def representative_raw_id(self) -> str: + """The deterministic (lowest raw_id) member an actuator would materialize.""" + return self.raw_ids[0] + + @property + def duplicate_raw_ids(self) -> tuple[str, ...]: + """Every group member other than the representative.""" + return self.raw_ids[1:] + + +@dataclass(frozen=True, slots=True) +class RawQuarantineGroupDedupPlan: + """Read-only projection: how much quarantined authority is a within-group duplicate cluster.""" + + #: Total quarantined, source_path-bearing rows examined. + scanned_count: int + groups: tuple[RawQuarantineGroup, ...] + #: (source_path, blob_hash) pairs with count > 1 among quarantined rows + #: that were EXCLUDED because some raw sharing that blob_hash already has + #: a materialized session or a non-quarantined revision_authority -- + #: already resolved, or raw_byte_duplicate_supersession's own territory. + already_resolved_group_count: int + + @property + def duplicate_count(self) -> int: + return sum(len(group.duplicate_raw_ids) for group in self.groups) + + @property + def duplicate_bytes(self) -> int: + return sum(group.blob_size * len(group.duplicate_raw_ids) for group in self.groups) + + +def plan_raw_quarantine_group_dedup( + source_conn: sqlite3.Connection, + index_conn: sqlite3.Connection, + *, + limit: int | None = None, +) -> RawQuarantineGroupDedupPlan: + """Read-only: classify quarantined raws into fully-unresolved duplicate groups. + + ``source_conn`` and ``index_conn`` are both used strictly for reads -- + safe to pass connections opened ``file:...?mode=ro``. Never mutates + either database. ``limit`` caps the number of qualifying *groups* + returned (not the number of rows scanned). + """ + original_source_row_factory = source_conn.row_factory + original_index_row_factory = index_conn.row_factory + source_conn.row_factory = sqlite3.Row + index_conn.row_factory = sqlite3.Row + try: + candidate_rows = source_conn.execute( + """ + SELECT raw_id, source_path, blob_hash, blob_size + FROM raw_sessions + WHERE revision_authority = 'quarantined' AND source_path IS NOT NULL + ORDER BY source_path, blob_hash, raw_id + """ + ).fetchall() + if not candidate_rows: + return RawQuarantineGroupDedupPlan(scanned_count=0, groups=(), already_resolved_group_count=0) + + grouped: dict[tuple[str, bytes], list[sqlite3.Row]] = {} + for row in candidate_rows: + key = (str(row["source_path"]), bytes(row["blob_hash"])) + grouped.setdefault(key, []).append(row) + + multi_member_keys = [key for key, rows in grouped.items() if len(rows) > 1] + if not multi_member_keys: + return RawQuarantineGroupDedupPlan( + scanned_count=len(candidate_rows), groups=(), already_resolved_group_count=0 + ) + + # Which blob_hash values (across ALL raw_sessions rows, any + # source_path/revision_authority) already have a materialized + # index.db session or a non-quarantined revision_authority -- those + # groups are out of scope: either already resolved, or + # raw_byte_duplicate_supersession's own territory. + distinct_hashes = sorted({blob_hash for _, blob_hash in multi_member_keys}) + hash_already_resolved: dict[bytes, bool] = {} + for chunk_start in range(0, len(distinct_hashes), 500): + hash_chunk = distinct_hashes[chunk_start : chunk_start + 500] + placeholders = ", ".join("?" for _ in hash_chunk) + rows = source_conn.execute( + f"SELECT raw_id, blob_hash, revision_authority FROM raw_sessions WHERE blob_hash IN ({placeholders})", + hash_chunk, + ).fetchall() + raw_ids_by_hash: dict[bytes, list[str]] = {} + non_quarantined_hashes: set[bytes] = set() + for row in rows: + blob_hash = bytes(row["blob_hash"]) + raw_ids_by_hash.setdefault(blob_hash, []).append(str(row["raw_id"])) + if str(row["revision_authority"]) != "quarantined": + non_quarantined_hashes.add(blob_hash) + + all_raw_ids = sorted({raw_id for hash_raw_ids in raw_ids_by_hash.values() for raw_id in hash_raw_ids}) + indexed_raw_ids: set[str] = set() + for id_chunk_start in range(0, len(all_raw_ids), 500): + id_chunk = all_raw_ids[id_chunk_start : id_chunk_start + 500] + id_placeholders = ", ".join("?" for _ in id_chunk) + for index_row in index_conn.execute( + f"SELECT DISTINCT raw_id FROM sessions WHERE raw_id IN ({id_placeholders})", id_chunk + ): + indexed_raw_ids.add(str(index_row[0])) + + for blob_hash, hash_raw_ids in raw_ids_by_hash.items(): + hash_already_resolved[blob_hash] = blob_hash in non_quarantined_hashes or any( + raw_id in indexed_raw_ids for raw_id in hash_raw_ids + ) + + groups: list[RawQuarantineGroup] = [] + already_resolved_group_count = 0 + for key in multi_member_keys: + source_path, blob_hash = key + if hash_already_resolved.get(blob_hash, False): + already_resolved_group_count += 1 + continue + # Check the cap BEFORE appending, not after: an after-the-fact + # check appends one group even when limit=0, silently + # dishonoring a caller's explicit "zero groups" request (the + # apply path iterates plan.groups directly, so a limit=0 dry-run + # would still classify -- and a limit=0 apply would still + # promote and mark -- exactly one group). Continue rather than + # break so already_resolved_group_count still reflects every + # remaining already-resolved key, not just the ones seen before + # the cap. + if limit is not None and len(groups) >= limit: + continue + rows = grouped[key] + raw_ids = tuple(sorted(str(row["raw_id"]) for row in rows)) + blob_size = int(rows[0]["blob_size"]) + groups.append( + RawQuarantineGroup( + source_path=source_path, + blob_hash=blob_hash, + blob_size=blob_size, + raw_ids=raw_ids, + ) + ) + + return RawQuarantineGroupDedupPlan( + scanned_count=len(candidate_rows), + groups=tuple(groups), + already_resolved_group_count=already_resolved_group_count, + ) + finally: + source_conn.row_factory = original_source_row_factory + index_conn.row_factory = original_index_row_factory + + +__all__ = [ + "RawQuarantineGroup", + "RawQuarantineGroupDedupPlan", + "plan_raw_quarantine_group_dedup", +] diff --git a/polylogue/storage/sqlite/archive_tiers/source.py b/polylogue/storage/sqlite/archive_tiers/source.py index 72abe2960a..5bccc12b96 100644 --- a/polylogue/storage/sqlite/archive_tiers/source.py +++ b/polylogue/storage/sqlite/archive_tiers/source.py @@ -21,7 +21,7 @@ from polylogue.storage.sqlite.archive_tiers.common import check, literal_check, nullable_check from polylogue.storage.sqlite.archive_tiers.types import ProvenRevisionAuthority -SOURCE_SCHEMA_VERSION = 24 +SOURCE_SCHEMA_VERSION = 25 SOURCE_DDL = f""" CREATE TABLE IF NOT EXISTS raw_sessions ( @@ -265,6 +265,39 @@ CREATE INDEX IF NOT EXISTS idx_raw_byte_duplicate_supersession_receipts_duplicate_of ON raw_byte_duplicate_supersession_receipts(duplicate_of_raw_id); +-- v25 (polylogue-zm4w8): one immutable receipt per raw_sessions row marked a +-- proven duplicate within a fully-quarantined (source_path, blob_hash) group +-- -- a group where EVERY member starts out quarantined (unlike v22's +-- raw_byte_duplicate_supersession_receipts above, which requires an already- +-- INDEXED twin). A distinct actuator (devtools workspace +-- raw-quarantine-group-dedup-apply) promotes exactly one representative raw +-- per group through the real ingest/materialization path so it becomes a +-- genuine indexed session, then marks the rest of the group 'byte_proven' +-- (reusing the existing closed revision_authority vocabulary -- there is no +-- 'superseded' member and widening it needs a full raw_sessions table +-- rebuild, migration 021's own precedent) with a receipt here pointing at +-- the representative raw and its newly materialized session. See +-- polylogue.storage.raw_quarantine_group_dedup + +-- polylogue.maintenance.raw_quarantine_group_dedup_apply. +CREATE TABLE IF NOT EXISTS raw_quarantine_group_dedup_receipts ( + raw_id TEXT PRIMARY KEY REFERENCES raw_sessions(raw_id) ON DELETE CASCADE, + source_path TEXT NOT NULL, + blob_hash BLOB NOT NULL CHECK(length(blob_hash) = 32), + blob_size INTEGER NOT NULL CHECK(blob_size >= 0), + representative_raw_id TEXT NOT NULL, + representative_session_id TEXT NOT NULL, + promoted_at_ms INTEGER NOT NULL CHECK(promoted_at_ms >= 0), + tool_version TEXT NOT NULL, + backup_manifest_path TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '' +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_raw_quarantine_group_dedup_receipts_promoted_at +ON raw_quarantine_group_dedup_receipts(promoted_at_ms); + +CREATE INDEX IF NOT EXISTS idx_raw_quarantine_group_dedup_receipts_representative +ON raw_quarantine_group_dedup_receipts(representative_raw_id); + -- Durable authority reconciliation ledger. The source tier owns this -- evidence because index.db and ops.db are rebuildable/disposable: neither -- can be the authority for whether an accepted replay plan was conserved. diff --git a/polylogue/storage/sqlite/migrations/source/025_raw_quarantine_group_dedup_receipts.sql b/polylogue/storage/sqlite/migrations/source/025_raw_quarantine_group_dedup_receipts.sql new file mode 100644 index 0000000000..ebf8a4a00a --- /dev/null +++ b/polylogue/storage/sqlite/migrations/source/025_raw_quarantine_group_dedup_receipts.sql @@ -0,0 +1,48 @@ +-- polylogue-zm4w8: 1,777 raw_sessions rows (22.2 GiB) were the bead's +-- original filing-time measurement (2026-08-03, morning) of pure redundant +-- duplicates -- same source_path AND same blob_hash as another raw_sessions +-- row -- among the codex-session quarantine backlog. A later, larger +-- read-only re-measurement the same day via this migration's own actuator +-- (devtools workspace raw-quarantine-group-dedup-apply --json, live archive, +-- no mutation) found 1,822 groups / 1,837 duplicate rows (8.22 GiB) -- +-- consistent with ordinary concurrent archive activity between the two +-- measurements, not a discrepancy in the classifier itself. ZERO of these +-- duplicate blob_hash values have any non-quarantined ("indexed") twin +-- anywhere in raw_sessions, so raw_byte_duplicate_supersession_apply (which +-- only matches a quarantined raw against an ALREADY-INDEXED twin) does not +-- and cannot catch this class: every member of one of these groups starts +-- out quarantined, with nothing yet materialized for any of them. +-- +-- A new, explicitly operator-invoked actuator (devtools workspace +-- raw-quarantine-group-dedup-apply) promotes exactly ONE representative raw +-- per (source_path, blob_hash) group through the real ingest/materialization +-- path (ParsingService.parse_from_raw -> write_parsed_session_to_archive -> +-- refresh_session_insights_bulk) so it becomes a genuine indexed session, +-- then marks the rest of the group's raws revision_authority='byte_proven' +-- (reusing raw_byte_duplicate_supersession's exact "quarantined, now proven +-- byte-identical to an indexed twin" precedent -- revision_authority is a +-- closed 3-value CHECK vocabulary with no 'superseded' member, and widening +-- it needs a full raw_sessions table rebuild, migration 021's own precedent +-- for why that is expensive). This receipt table is the durable, per-row +-- record of *which* representative raw and materialized session each +-- duplicate was superseded by -- never silently, exactly like +-- raw_byte_duplicate_supersession_receipts (migration 023) records its own +-- distinct evidence mechanism. +CREATE TABLE IF NOT EXISTS raw_quarantine_group_dedup_receipts ( + raw_id TEXT PRIMARY KEY REFERENCES raw_sessions(raw_id) ON DELETE CASCADE, + source_path TEXT NOT NULL, + blob_hash BLOB NOT NULL CHECK(length(blob_hash) = 32), + blob_size INTEGER NOT NULL CHECK(blob_size >= 0), + representative_raw_id TEXT NOT NULL, + representative_session_id TEXT NOT NULL, + promoted_at_ms INTEGER NOT NULL CHECK(promoted_at_ms >= 0), + tool_version TEXT NOT NULL, + backup_manifest_path TEXT NOT NULL, + detail TEXT NOT NULL DEFAULT '' +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_raw_quarantine_group_dedup_receipts_promoted_at +ON raw_quarantine_group_dedup_receipts(promoted_at_ms); + +CREATE INDEX IF NOT EXISTS idx_raw_quarantine_group_dedup_receipts_representative +ON raw_quarantine_group_dedup_receipts(representative_raw_id); diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index 77fb7bc442..d8585eb4cc 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -992,6 +992,83 @@ def test_stalled_append_cursor_freshness_passes_on_coherent_archive(tmp_path: Pa assert check.evidence["stalled_count"] == 0 +def test_fully_quarantined_duplicate_group_trips_raw_quarantine_group_dedup(tmp_path: Path) -> None: + """polylogue-zm4w8: two quarantined raws sharing (source_path, blob_hash), + with no indexed twin anywhere for that blob_hash, is exactly the residual + gap raw-byte-duplicate-supersession-apply cannot see (it requires an + already-indexed twin). This must trip ERROR, not the WARN + source-index-coverage already gives quarantined-but-unindexed heads. + """ + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + for raw_id in ("raw-dup-a", "raw-dup-b"): + source_conn.execute( + """ + INSERT INTO raw_sessions(raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms) + VALUES (?, 'codex-session', ?, '/rollout-repeated.jsonl', ?, 10, 100) + """, + (raw_id, f"native-{raw_id}", b"d" * 32), + ) + source_conn.commit() + finally: + source_conn.close() + + report = verify_archive(tmp_path, checks=("raw-quarantine-group-dedup",)) + + check = _check(report, "raw-quarantine-group-dedup") + assert check.status is OutcomeStatus.ERROR + assert check.evidence["group_count"] == 1 + assert check.evidence["duplicate_count"] == 1 + group_sample = check.evidence["group_sample"] + assert len(group_sample) == 1 + assert group_sample[0]["source_path"] == "/rollout-repeated.jsonl" + assert group_sample[0]["representative_raw_id"] == "raw-dup-a" + assert group_sample[0]["duplicate_raw_ids"] == ["raw-dup-b"] + + +def test_quarantined_duplicate_with_indexed_twin_elsewhere_does_not_trip(tmp_path: Path) -> None: + """A (source_path, blob_hash) group of >1 quarantined rows whose + blob_hash ALSO appears on an already-indexed raw elsewhere is + raw-byte-duplicate-supersession-apply's territory, not this check's -- + it must not double-flag content that actuator can already resolve. + """ + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + for raw_id in ("raw-dup-c", "raw-dup-d"): + source_conn.execute( + """ + INSERT INTO raw_sessions(raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms) + VALUES (?, 'codex-session', ?, '/rollout-also-indexed.jsonl', ?, 10, 100) + """, + (raw_id, f"native-{raw_id}", b"s" * 32), + ) + # raw-1's blob_hash (b"s" * 32) is the coherent-archive fixture's + # already-indexed raw -- share it here to simulate an indexed twin. + source_conn.execute("UPDATE raw_sessions SET blob_hash = ? WHERE raw_id = 'raw-1'", (b"s" * 32,)) + source_conn.commit() + finally: + source_conn.close() + + report = verify_archive(tmp_path, checks=("raw-quarantine-group-dedup",)) + + check = _check(report, "raw-quarantine-group-dedup") + assert check.status is OutcomeStatus.OK + assert check.evidence["group_count"] == 0 + assert check.evidence["already_resolved_group_count"] == 1 + + +def test_raw_quarantine_group_dedup_passes_on_coherent_archive(tmp_path: Path) -> None: + _seed_coherent_archive(tmp_path) + + report = verify_archive(tmp_path, checks=("raw-quarantine-group-dedup",)) + + check = _check(report, "raw-quarantine-group-dedup") + assert check.status is OutcomeStatus.OK + assert check.evidence["group_count"] == 0 + + @pytest.mark.parametrize("check_name", ARCHIVE_VERIFICATION_CHECK_NAMES) def test_every_registry_check_does_not_error_on_the_real_pipeline_corpus( check_name: str, seeded_archive: SeededArchiveArtifact @@ -1185,6 +1262,7 @@ def test_reindex_acceptance_subset_is_satisfiable_from_index_only_root(tmp_path: "user-tier-refs": "test_dangling_assertion_target_trips_user_tier_refs", "excluded-cursor-vocabulary-honesty": "test_excluded_cursor_with_live_next_retry_at_trips_vocabulary_honesty", "stalled-append-cursor-freshness": "test_stalled_append_cursor_trips_freshness_check", + "raw-quarantine-group-dedup": "test_fully_quarantined_duplicate_group_trips_raw_quarantine_group_dedup", } diff --git a/tests/unit/maintenance/test_raw_quarantine_group_dedup_apply.py b/tests/unit/maintenance/test_raw_quarantine_group_dedup_apply.py new file mode 100644 index 0000000000..8632854d27 --- /dev/null +++ b/tests/unit/maintenance/test_raw_quarantine_group_dedup_apply.py @@ -0,0 +1,414 @@ +"""polylogue-zm4w8: the actuator for fully-quarantined byte-identical group dedup. + +Builds a fixture archive with a genuine same-source_path, byte-identical +quarantined group (never indexed anywhere) plus a genuine singleton +(non-grouped) quarantined raw, and proves: + +* dry-run (the default) never mutates anything, and never runs the ingest + pipeline; +* --apply materializes exactly one representative raw per group through the + real production ingest pipeline (it becomes a genuine ``sessions`` row in + index.db), marks the rest ``revision_authority='byte_proven'``, and writes + an immutable receipt naming the representative raw_id/session_id; +* the singleton (non-grouped) row is left untouched; +* applying without a backup manifest is refused before anything is touched; +* applying with an invalid/stale backup manifest is refused before the + ingest pipeline ever runs. +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import cast + +import pytest + +from polylogue.core.enums import Provider +from polylogue.maintenance.raw_quarantine_group_dedup_apply import ( + TOOL_VERSION, + RawQuarantineGroupDedupApplyError, + _checkpoint_live_tier, + _mark_group_duplicates, + apply_raw_quarantine_group_dedup, +) +from polylogue.storage.raw_quarantine_group_dedup import RawQuarantineGroup +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + +def _codex_raw_payload(session_id: str, *, text: str) -> bytes: + """A minimal, genuinely-parseable single-session Codex JSONL raw.""" + session_meta = json.dumps( + {"type": "session_meta", "payload": {"id": session_id, "timestamp": "2026-06-01T00:00:00Z"}}, + separators=(",", ":"), + ) + response_item = json.dumps( + { + "type": "response_item", + "payload": { + "type": "message", + "id": "one", + "role": "user", + "content": [{"type": "input_text", "text": text}], + }, + }, + separators=(",", ":"), + ) + return (session_meta + "\n" + response_item + "\n").encode() + + +def _write_quarantined_raw(archive: ArchiveStore, *, raw_id: str, payload: bytes, source_path: str) -> None: + archive.write_raw_payload( + provider=Provider.CODEX, + payload=payload, + source_path=source_path, + source_index=-1, + acquired_at_ms=1_700_000_000_000, + raw_id=raw_id, + ) + + +_GROUP_PAYLOAD = _codex_raw_payload("zm4w8-repeated-session", text="repeated content") +_SINGLETON_PAYLOAD = _codex_raw_payload("zm4w8-singleton-session", text="singleton content") + + +def _build_fixture_archive(tmp_path: Path) -> Path: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + # Three separate acquisitions of the same source file -- the + # population this actuator resolves. + for raw_id in ("raw-a", "raw-b", "raw-c"): + _write_quarantined_raw( + archive, raw_id=raw_id, payload=_GROUP_PAYLOAD, source_path=str(tmp_path / "repeated.jsonl") + ) + # A genuine singleton -- never grouped, must never be touched. + _write_quarantined_raw( + archive, raw_id="raw-singleton", payload=_SINGLETON_PAYLOAD, source_path=str(tmp_path / "singleton.jsonl") + ) + archive.commit() + + return archive_root + + +def _revision_authority_rows(archive_root: Path) -> dict[str, str]: + conn = sqlite3.connect(archive_root / "source.db") + try: + rows = conn.execute("SELECT raw_id, revision_authority FROM raw_sessions").fetchall() + finally: + conn.close() + return dict(rows) + + +def _receipt_rows(archive_root: Path) -> dict[str, tuple[str, str, str]]: + conn = sqlite3.connect(archive_root / "source.db") + try: + rows = conn.execute( + "SELECT raw_id, representative_raw_id, representative_session_id, tool_version " + "FROM raw_quarantine_group_dedup_receipts" + ).fetchall() + finally: + conn.close() + return { + raw_id: (rep_raw_id, rep_session_id, tool_version) for raw_id, rep_raw_id, rep_session_id, tool_version in rows + } + + +def _index_sessions_rows(archive_root: Path) -> list[tuple[str, str | None]]: + conn = sqlite3.connect(archive_root / "index.db") + try: + rows = conn.execute("SELECT session_id, raw_id FROM sessions ORDER BY session_id").fetchall() + finally: + conn.close() + return [(session_id, raw_id) for session_id, raw_id in rows] + + +async def test_dry_run_makes_zero_mutations(tmp_path: Path) -> None: + archive_root = _build_fixture_archive(tmp_path) + before_source = _revision_authority_rows(archive_root) + before_index = _index_sessions_rows(archive_root) + assert all(authority == "quarantined" for authority in before_source.values()) + + report = await apply_raw_quarantine_group_dedup(archive_root, dry_run=True) + + assert report.applied is False + assert report.scanned_count == 4 + assert report.group_count == 1 + assert report.promoted_count == 1 + assert report.marked_duplicate_count == 2 + promotion = report.promotions[0] + assert promotion.representative_raw_id == "raw-a" + assert promotion.representative_session_id == "" + assert promotion.duplicate_raw_ids == ("raw-b", "raw-c") + + assert _revision_authority_rows(archive_root) == before_source + assert _index_sessions_rows(archive_root) == before_index + assert _receipt_rows(archive_root) == {} + + +async def test_apply_refuses_without_backup_manifest(tmp_path: Path) -> None: + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + + with pytest.raises(RawQuarantineGroupDedupApplyError, match="backup manifest"): + await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=None, dry_run=False) + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + + +async def test_apply_refuses_when_backup_manifest_invalid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + + def _reject(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + raise ValueError("backup manifest does not match live source.db") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _reject, + ) + + manifest = tmp_path / "stale-backup" / "manifest.json" + with pytest.raises(ValueError, match="does not match"): + await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=manifest, dry_run=False) + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + + +async def test_apply_materializes_one_representative_and_marks_duplicates( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = _build_fixture_archive(tmp_path) + + validated: list[tuple[Path, object]] = [] + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + validated.append((manifest, tier)) + assert connection.execute("SELECT 1").fetchone() == (1,) + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + manifest = tmp_path / "verified-backup" / "manifest.json" + report = await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=manifest, dry_run=False) + + assert report.applied is True + assert report.group_count == 1 + assert report.promoted_count == 1 + assert report.marked_duplicate_count == 2 + assert report.backup_manifest == manifest + assert len(validated) >= 2 # precheck + at least one locked-transaction re-validation + + promotion = report.promotions[0] + assert promotion.representative_raw_id == "raw-a" + assert promotion.representative_session_id != "" + assert promotion.duplicate_raw_ids == ("raw-b", "raw-c") + + # The representative raw is now genuinely indexed. + index_sessions = _index_sessions_rows(archive_root) + assert (promotion.representative_session_id, "raw-a") in index_sessions + # No session materialized for the duplicates or the untouched singleton. + indexed_raw_ids = {raw_id for _session_id, raw_id in index_sessions} + assert "raw-b" not in indexed_raw_ids + assert "raw-c" not in indexed_raw_ids + assert "raw-singleton" not in indexed_raw_ids + + rows = _revision_authority_rows(archive_root) + assert rows["raw-a"] == "quarantined" # untouched -- proof is its own indexed session, not this field + assert rows["raw-b"] == "byte_proven" + assert rows["raw-c"] == "byte_proven" + # The genuine singleton is never touched. + assert rows["raw-singleton"] == "quarantined" + + receipts = _receipt_rows(archive_root) + assert receipts == { + "raw-b": ("raw-a", promotion.representative_session_id, TOOL_VERSION), + "raw-c": ("raw-a", promotion.representative_session_id, TOOL_VERSION), + } + + +async def test_apply_leaves_group_untouched_when_representative_produces_no_session( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A representative whose bytes fail to parse (decode error, no session + materialized) must leave the WHOLE group untouched: no duplicate marked, + no receipt, and the representative itself still quarantined -- never + guessed at. Covers raw_quarantine_group_dedup_apply.py's "zero sessions" + branch (CodeRabbit PR #3697).""" + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + unparseable_payload = b"this is not valid json at all, just garbage bytes\n" + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for raw_id in ("raw-x", "raw-y"): + _write_quarantined_raw( + archive, raw_id=raw_id, payload=unparseable_payload, source_path=str(tmp_path / "garbage.jsonl") + ) + archive.commit() + + before = _revision_authority_rows(archive_root) + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + manifest = tmp_path / "verified-backup" / "manifest.json" + report = await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=manifest, dry_run=False) + + assert report.applied is True + assert report.group_count == 1 # classified as a group -- classification doesn't require parseability + assert report.promoted_count == 0 + assert report.marked_duplicate_count == 0 + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + assert _index_sessions_rows(archive_root) == [] + + +async def test_apply_leaves_group_untouched_when_representative_materializes_multiple_sessions( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A representative raw that is a genuine multi-session capture file + (e.g. a grouped Claude Code JSONL split by sessionId) materializes MORE + than one sessions row from one raw_id -- this actuator's whole premise + is one raw/one representative session, so it must refuse to guess which + materialized session is "the" representative and leave the group + untouched entirely, exactly like the zero-session case.""" + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + multi_session_payload = ( + b'{"type":"user","sessionId":"first-session","uuid":"u1",' + b'"message":{"role":"user","content":"one"}}\n' + b'{"type":"user","sessionId":"second-session","uuid":"u2",' + b'"message":{"role":"user","content":"two"}}\n' + ) + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for raw_id in ("raw-multi-a", "raw-multi-b"): + archive.write_raw_payload( + provider=Provider.CLAUDE_CODE, + payload=multi_session_payload, + source_path=str(tmp_path / "multi.jsonl"), + source_index=-1, + acquired_at_ms=1_700_000_000_000, + raw_id=raw_id, + ) + archive.commit() + + before = _revision_authority_rows(archive_root) + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + manifest = tmp_path / "verified-backup" / "manifest.json" + report = await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=manifest, dry_run=False) + + assert report.applied is True + assert report.group_count == 1 + assert report.promoted_count == 0 + assert report.marked_duplicate_count == 0 + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + # The representative's own content DID materialize (real, legitimate + # sessions) -- this actuator just declines to build a receipt around it. + index_sessions = _index_sessions_rows(archive_root) + assert len(index_sessions) == 2 + assert all(raw_id == "raw-multi-a" for _session_id, raw_id in index_sessions) + + +def test_mark_group_duplicates_rolls_back_on_receipt_constraint_violation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Covers the phase-two rollback branch directly: a receipt INSERT that + violates a CHECK constraint must roll back the WHOLE transaction -- + including any earlier duplicate's revision_authority UPDATE already + applied within the same locked transaction -- not partially commit + (CodeRabbit PR #3697).""" + archive_root = _build_fixture_archive(tmp_path) + before = _revision_authority_rows(archive_root) + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + # blob_hash with the wrong length violates raw_quarantine_group_dedup_receipts' + # CHECK(length(blob_hash) = 32) on the very first duplicate's INSERT. + bad_group = RawQuarantineGroup( + source_path=str(tmp_path / "repeated.jsonl"), + blob_hash=b"\x00" * 31, + blob_size=len(_GROUP_PAYLOAD), + raw_ids=("raw-a", "raw-b", "raw-c"), + ) + manifest = tmp_path / "verified-backup" / "manifest.json" + + with pytest.raises(sqlite3.IntegrityError): + _mark_group_duplicates(archive_root / "source.db", manifest, bad_group, "fake-representative-session-id") + + assert _revision_authority_rows(archive_root) == before + assert _receipt_rows(archive_root) == {} + + +def test_checkpoint_live_tier_refuses_when_wal_checkpoint_reports_busy(tmp_path: Path) -> None: + """``PRAGMA wal_checkpoint(TRUNCATE)`` always returns one row + ``(busy, log, checkpointed)``. ``busy=1`` means the WAL was NOT fully + truncated (another connection held a blocking lock); a subsequent + backup-manifest fingerprint check must not silently attest against a + tier that still has uncheckpointed frames (CodeRabbit PR #3697).""" + + class _BusyCursor: + def fetchone(self) -> tuple[int, int, int]: + return (1, 0, 0) # busy=1 + + class _BusyConnection: + def execute(self, _sql: str) -> _BusyCursor: + return _BusyCursor() + + with pytest.raises(RawQuarantineGroupDedupApplyError, match="blocked by another connection"): + _checkpoint_live_tier(cast(sqlite3.Connection, _BusyConnection())) + + +async def test_apply_with_no_qualifying_groups_is_a_clean_no_op( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + _write_quarantined_raw( + archive, raw_id="raw-only", payload=_SINGLETON_PAYLOAD, source_path=str(tmp_path / "only.jsonl") + ) + archive.commit() + + def _fake_validate(manifest: Path, tier: object, *, connection: sqlite3.Connection) -> Path: + return manifest.with_name("verification-receipt.json") + + monkeypatch.setattr( + "polylogue.maintenance.raw_quarantine_group_dedup_apply.validate_migration_backup_manifest", + _fake_validate, + ) + + manifest = tmp_path / "verified-backup" / "manifest.json" + report = await apply_raw_quarantine_group_dedup(archive_root, backup_manifest=manifest, dry_run=False) + + assert report.applied is True + assert report.group_count == 0 + assert report.promoted_count == 0 + assert report.marked_duplicate_count == 0 + assert _receipt_rows(archive_root) == {} diff --git a/tests/unit/storage/test_raw_quarantine_group_dedup.py b/tests/unit/storage/test_raw_quarantine_group_dedup.py new file mode 100644 index 0000000000..9102826d79 --- /dev/null +++ b/tests/unit/storage/test_raw_quarantine_group_dedup.py @@ -0,0 +1,205 @@ +"""polylogue-zm4w8: fully-quarantined byte-identical group dedup classifier. + +Proves the read-only classifier scopes strictly to (source_path, blob_hash) +groups among quarantined raw_sessions rows where every member is quarantined +and NONE of them (nor any other raw sharing that blob_hash anywhere) already +has a materialized index.db session or a non-quarantined revision_authority +-- the residual population raw-byte-duplicate-supersession (which requires +an already-indexed twin) cannot see by construction. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from polylogue.core.enums import Provider +from polylogue.storage.raw_quarantine_group_dedup import plan_raw_quarantine_group_dedup +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + +def _write_quarantined_raw(archive: ArchiveStore, *, raw_id: str, payload: bytes, source_path: str) -> None: + """Write a raw the way real acquisition does: no revision envelope, default quarantined authority.""" + archive.write_raw_payload( + provider=Provider.CODEX, + payload=payload, + source_path=source_path, + source_index=-1, + acquired_at_ms=1_700_000_000_000, + raw_id=raw_id, + ) + + +def _index_session(conn: sqlite3.Connection, *, raw_id: str, native_id: str) -> None: + conn.execute( + "INSERT INTO sessions (origin, native_id, content_hash, raw_id, created_at_ms, updated_at_ms) " + "VALUES ('codex-session', ?, ?, ?, 0, 0)", + (native_id, b"\x22" * 32, raw_id), + ) + + +def test_flags_fully_quarantined_same_source_path_group(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + repeated_payload = b'{"messages":["repeated"]}\n' + unique_payload = b'{"messages":["unique"]}\n' + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + # Three separate acquisitions of the SAME source file, byte-identical + # -- the target population. Deliberately written raw-c, raw-a, raw-b + # out of id order to prove representative selection is by raw_id + # value, not insertion order. + for raw_id in ("raw-c", "raw-a", "raw-b"): + _write_quarantined_raw( + archive, + raw_id=raw_id, + payload=repeated_payload, + source_path=str(tmp_path / "repeated.jsonl"), + ) + # A single, never-repeated acquisition -- not a group (count == 1). + _write_quarantined_raw( + archive, + raw_id="raw-singleton", + payload=unique_payload, + source_path=str(tmp_path / "singleton.jsonl"), + ) + archive.commit() + + source_conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{archive_root / 'index.db'}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn) + finally: + source_conn.close() + index_conn.close() + + assert plan.scanned_count == 4 + assert len(plan.groups) == 1 + group = plan.groups[0] + assert group.source_path == str(tmp_path / "repeated.jsonl") + assert group.raw_ids == ("raw-a", "raw-b", "raw-c") + assert group.representative_raw_id == "raw-a" + assert group.duplicate_raw_ids == ("raw-b", "raw-c") + assert group.blob_size == len(repeated_payload) + assert plan.duplicate_count == 2 + assert plan.duplicate_bytes == len(repeated_payload) * 2 + assert plan.already_resolved_group_count == 0 + + +def test_different_source_paths_never_group_even_if_byte_identical(tmp_path: Path) -> None: + """Same bytes, different source_path -- not this classifier's group key + (that would be raw-byte-duplicate-supersession's territory if one side + were indexed; here neither is, so it stays untouched by both).""" + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + shared_payload = b'{"messages":["shared-bytes-different-files"]}\n' + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + _write_quarantined_raw(archive, raw_id="raw-x", payload=shared_payload, source_path=str(tmp_path / "x.jsonl")) + _write_quarantined_raw(archive, raw_id="raw-y", payload=shared_payload, source_path=str(tmp_path / "y.jsonl")) + archive.commit() + + source_conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{archive_root / 'index.db'}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn) + finally: + source_conn.close() + index_conn.close() + + assert plan.scanned_count == 2 + assert plan.groups == () + assert plan.already_resolved_group_count == 0 + + +def test_group_with_indexed_twin_elsewhere_is_already_resolved_not_flagged(tmp_path: Path) -> None: + """raw-byte-duplicate-supersession-apply's own territory: if ANY raw + sharing this blob_hash (any source_path) already has a materialized + session, this classifier must defer to that actuator, not double-act. + """ + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + payload = b'{"messages":["already-has-an-indexed-twin"]}\n' + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for raw_id in ("raw-dup-1", "raw-dup-2"): + _write_quarantined_raw(archive, raw_id=raw_id, payload=payload, source_path=str(tmp_path / "twinned.jsonl")) + # A THIRD raw, different source_path, same bytes -- and this one is + # indexed. The group above must not be flagged. + _write_quarantined_raw( + archive, raw_id="raw-indexed-elsewhere", payload=payload, source_path=str(tmp_path / "elsewhere.jsonl") + ) + archive.commit() + _index_session(archive._conn, raw_id="raw-indexed-elsewhere", native_id="native-indexed-elsewhere") + archive.commit() + + source_conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{archive_root / 'index.db'}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn) + finally: + source_conn.close() + index_conn.close() + + assert plan.scanned_count == 3 + assert plan.groups == () + assert plan.already_resolved_group_count == 1 + + +def test_limit_caps_number_of_groups_returned(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for group_index in range(3): + payload = f'{{"n":{group_index}}}\n'.encode() + for member_index in range(2): + _write_quarantined_raw( + archive, + raw_id=f"raw-{group_index}-{member_index}", + payload=payload, + source_path=str(tmp_path / f"group-{group_index}.jsonl"), + ) + archive.commit() + + source_conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{archive_root / 'index.db'}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn, limit=2) + finally: + source_conn.close() + index_conn.close() + + assert len(plan.groups) == 2 + + +def test_limit_zero_returns_no_groups(tmp_path: Path) -> None: + """Regression (CodeRabbit PR #3697): the cap must be checked BEFORE a + group is appended, not after -- an after-the-fact check silently + appended exactly one group even when the caller explicitly asked for + zero via limit=0. The apply path iterates plan.groups directly, so this + bug would have promoted and marked one duplicate group despite a + limit=0 dry-run/apply call asking for none.""" + archive_root = tmp_path / "archive" + initialize_active_archive_root(archive_root) + + with ArchiveStore.open_existing(archive_root, read_only=False) as archive: + for raw_id in ("raw-a", "raw-b"): + _write_quarantined_raw( + archive, raw_id=raw_id, payload=b'{"n":0}\n', source_path=str(tmp_path / "group.jsonl") + ) + archive.commit() + + source_conn = sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True) + index_conn = sqlite3.connect(f"file:{archive_root / 'index.db'}?mode=ro", uri=True) + try: + plan = plan_raw_quarantine_group_dedup(source_conn, index_conn, limit=0) + finally: + source_conn.close() + index_conn.close() + + assert plan.groups == ()