feat(storage): dedupe unindexed byte-identical raw quarantine groups

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 <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",

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 <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())

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. |

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)