From 6e357c5ae51a40082efd420174912e2f915152c0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 15:28:24 +0200 Subject: [PATCH 1/9] fix(tests): bootstrap durable archive fixtures Route stale multi-tier fixtures through the owned active-root bootstrap. Preserve the fast-forward test generation layout after bootstrap, and allow campaign ownership acquisition to create a documented fresh output directory before its descriptor is locked. --- devtools/campaign_archive_location.py | 4 ++++ devtools/scale_regression_probe.py | 7 ++----- tests/unit/api/test_operation_executor_routes.py | 6 ++---- tests/unit/devtools/test_index_fast_forward.py | 8 +++----- 4 files changed, 11 insertions(+), 14 deletions(-) diff --git a/devtools/campaign_archive_location.py b/devtools/campaign_archive_location.py index 72061fabb1..5d9d0073cd 100644 --- a/devtools/campaign_archive_location.py +++ b/devtools/campaign_archive_location.py @@ -66,6 +66,10 @@ def acquire(cls, archive_dir: Path, *, owner_id: str | None = None) -> CampaignA before any SQLite tier file is opened when ``archive_dir`` is already owned by another live campaign/maintenance process. """ + # A campaign owns a fresh output directory. Establish that directory + # before resolving its descriptor, then retain the normal ownership + # proof before a caller can open any SQLite tier. + archive_dir.mkdir(mode=0o700, parents=True, exist_ok=True) location = ArchiveLocation.resolve(archive_dir) owned = OwnedArchiveLocation.acquire(location, owner_id=owner_id) return cls(owned=owned) diff --git a/devtools/scale_regression_probe.py b/devtools/scale_regression_probe.py index 5753127e05..2e0b971cc4 100644 --- a/devtools/scale_regression_probe.py +++ b/devtools/scale_regression_probe.py @@ -30,7 +30,7 @@ from polylogue.storage.blob_store import BlobStore from polylogue.storage.insights.session import rebuild as rebuild_mod from polylogue.storage.insights.session.runtime import SessionInsightCounts -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive from polylogue.storage.sqlite.connection import open_connection @@ -67,10 +67,7 @@ def _session_id(native_id: str, origin: str = Origin.CODEX_SESSION.value) -> str def _init_archive(root: Path) -> None: - root.mkdir(parents=True, exist_ok=True) - initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) - initialize_archive_database(root / "index.db", ArchiveTier.INDEX) - initialize_archive_database(root / "user.db", ArchiveTier.USER) + initialize_active_archive_root(root) def _parsed_session(native_id: str, *, title: str, messages: int = 1) -> ParsedSession: diff --git a/tests/unit/api/test_operation_executor_routes.py b/tests/unit/api/test_operation_executor_routes.py index 9dd375a4f1..4a5358b6da 100644 --- a/tests/unit/api/test_operation_executor_routes.py +++ b/tests/unit/api/test_operation_executor_routes.py @@ -14,15 +14,13 @@ from polylogue.api import Polylogue from polylogue.operations.mutation_transaction import OperationExecutor -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database -from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root def _seed_archive(archive_root: Path, *, native_id: str) -> str: source_db = archive_root / "source.db" index_db = archive_root / "index.db" - initialize_archive_database(source_db, ArchiveTier.SOURCE) - initialize_archive_database(index_db, ArchiveTier.INDEX) + initialize_active_archive_root(archive_root) raw_id = f"raw-{native_id}" session_id = f"codex-session:{native_id}" with sqlite3.connect(source_db) as conn: diff --git a/tests/unit/devtools/test_index_fast_forward.py b/tests/unit/devtools/test_index_fast_forward.py index c4cf498a9b..c5aa6d5afc 100644 --- a/tests/unit/devtools/test_index_fast_forward.py +++ b/tests/unit/devtools/test_index_fast_forward.py @@ -18,20 +18,18 @@ from polylogue.storage.blob_store import BlobStore from polylogue.storage.index_generation import IndexGeneration, IndexGenerationStore from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier def _archive(tmp_path: Path, *, extra_native_ids: tuple[str, ...] = ()) -> Path: root = tmp_path / "archive" - root.mkdir() - for tier in (ArchiveTier.SOURCE, ArchiveTier.USER, ArchiveTier.EMBEDDINGS, ArchiveTier.OPS): - initialize_archive_database(root / f"{tier.value}.db", tier) + initialize_active_archive_root(root) storage = tmp_path / "storage" active_root = storage / ".index-generations" / "v36" active_root.mkdir(parents=True) active = active_root / "index.db" - initialize_archive_database(active, ArchiveTier.INDEX) + (root / "index.db").replace(active) (storage / "index.db").symlink_to(active) (root / "index.db").symlink_to(storage / "index.db") From f6f7e21255272cd406233301cb37fafe159d4400 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 15:28:32 +0200 Subject: [PATCH 2/9] test(daemon): generate bulk rebuild gate receipts Seed test-owned external evidence and run the schema-inference gate before ownership assertions. The ownership tests isolate source admission because the minimal gate corpus intentionally has no replay census. --- .../daemon/test_bulk_rebuild_ownership.py | 55 +++++++++++++------ 1 file changed, 39 insertions(+), 16 deletions(-) diff --git a/tests/unit/daemon/test_bulk_rebuild_ownership.py b/tests/unit/daemon/test_bulk_rebuild_ownership.py index db09a35232..5814475938 100644 --- a/tests/unit/daemon/test_bulk_rebuild_ownership.py +++ b/tests/unit/daemon/test_bulk_rebuild_ownership.py @@ -23,6 +23,7 @@ from polylogue.daemon.bulk_rebuild import resolve_or_start_daemon_bulk_rebuild_transaction from polylogue.maintenance.rebuild_index import RebuildSchemaCurrencyError +from polylogue.maintenance.schema_inference_gate import run_schema_inference_gate from polylogue.storage.archive_identity import ( ArchiveLocation, ArchiveOwnershipError, @@ -30,16 +31,23 @@ assert_owns_archive_location, ) from polylogue.storage.archive_readiness import probe_archive_tier -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier -from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS -from tests.infra.rebuild_receipt import write_valid_rebuild_receipt +from tests.infra.schema_inference import seed_schema_inference_archive -def _init_empty_source(root: Path) -> None: - root.mkdir(parents=True, exist_ok=True) - for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): - initialize_archive_database(root / f"{tier.value}.db", tier) +def _init_empty_source(root: Path) -> Path: + return seed_schema_inference_archive(root) + + +def _schema_inference_receipt(root: Path, ground_truth: Path, tmp_path: Path) -> Path: + receipt = tmp_path / f"{root.name}-schema-inference-gate-receipt.json" + result = run_schema_inference_gate( + root, + receipt_path=receipt, + ground_truth_roots={"codex-session": (ground_truth,)}, + ) + assert result.passed, result.payload["pass_fail_reasons"] + return receipt def test_daemon_bulk_rebuild_rejects_schema_mismatch_before_transaction_bookkeeping(tmp_path: Path) -> None: @@ -71,8 +79,8 @@ def test_daemon_bulk_rebuild_rechecks_schema_currency_after_ownership( from polylogue.daemon import bulk_rebuild root = tmp_path / "archive" - _init_empty_source(root) - receipt = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") + ground_truth = _init_empty_source(root) + receipt = _schema_inference_receipt(root, ground_truth, tmp_path) real_assert = assert_owns_archive_location def mutate_audit_after_ownership(owned: OwnedArchiveLocation, location: ArchiveLocation) -> None: @@ -92,7 +100,9 @@ def mutate_audit_after_ownership(owned: OwnedArchiveLocation, location: ArchiveL assert not (root / ".index-rebuild-transactions").exists() -def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None: +def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: """A concurrent holder of the archive-location ownership lock must block the daemon's bulk-rebuild transaction resolve/retire path before any generation directory or transaction record is created -- mirroring @@ -100,12 +110,19 @@ def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_pat offline rebuild entry point. """ root = tmp_path / "archive" - _init_empty_source(root) + ground_truth = _init_empty_source(root) + receipt = _schema_inference_receipt(root, ground_truth, tmp_path) + # This test proves ownership around transaction bookkeeping. The schema + # gate above is real; the source-admission route is separately covered + # and the helper's tiny gate corpus deliberately has no replay census. + from polylogue.daemon import bulk_rebuild + + monkeypatch.setattr(bulk_rebuild, "validate_rebuild_source_admission", lambda *_args: None) location = ArchiveLocation.resolve(root) owned = OwnedArchiveLocation.acquire(location, owner_id="concurrent-campaign") try: with pytest.raises(ArchiveOwnershipError): - resolve_or_start_daemon_bulk_rebuild_transaction(root) + resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt) # Failure happened before any generation/transaction bookkeeping was created. assert not (root / ".index-generations").exists() assert not (root / ".index-rebuild-transactions").exists() @@ -113,19 +130,25 @@ def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_pat owned.release() # Releasing the concurrent holder's ownership lets the daemon proceed. - transaction = resolve_or_start_daemon_bulk_rebuild_transaction(root) + transaction = resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt) assert transaction.status == "running" -def test_daemon_bulk_rebuild_releases_ownership_lock_after_resolving(tmp_path: Path) -> None: +def test_daemon_bulk_rebuild_releases_ownership_lock_after_resolving( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: """The ownership lock must not be left held after resolving/starting a transaction, so a subsequent maintenance/campaign writer can still acquire it. """ root = tmp_path / "archive" - _init_empty_source(root) + ground_truth = _init_empty_source(root) + receipt = _schema_inference_receipt(root, ground_truth, tmp_path) + from polylogue.daemon import bulk_rebuild + + monkeypatch.setattr(bulk_rebuild, "validate_rebuild_source_admission", lambda *_args: None) - resolve_or_start_daemon_bulk_rebuild_transaction(root) + resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt) location = ArchiveLocation.resolve(root) owned = OwnedArchiveLocation.acquire(location, owner_id="post-resolve-probe") From a398a66fafcd7e9cc73be9b057b79fb80bfef10d Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 18:06:07 +0200 Subject: [PATCH 3/9] fix(tests): bind seeded archives to parser semantics Cached real-pipeline archives were keyed by Git HEAD and a short recipe file list. Dirty worktree parser changes could therefore reuse stale archives and produce order-dependent fingerprint failures. Include the actual parser and lowering semantic digest in cache identity and manifest provenance. --- tests/infra/workload_artifacts.py | 22 ++++++++++++++++++++- tests/unit/infra/test_workload_artifacts.py | 12 +++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/tests/infra/workload_artifacts.py b/tests/infra/workload_artifacts.py index 6d3039e66c..28d4eea68b 100644 --- a/tests/infra/workload_artifacts.py +++ b/tests/infra/workload_artifacts.py @@ -37,11 +37,12 @@ ) from polylogue.schemas.synthetic import SyntheticCorpus from polylogue.schemas.synthetic.models import SyntheticArtifactFacts +from polylogue.sources.origin_specs import lowering_fingerprint, origin_specs from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot, raw_materialization_ready from polylogue.storage.raw_reconciler import inspect_raw_authority_frontier from tests.infra.source_builders import SyntheticAntigravityLanguageServerClient -_ARTIFACT_PROTOCOL_VERSION = 1 +_ARTIFACT_PROTOCOL_VERSION = 2 _CACHE_ROOT = Path("/realm/tmp/polylogue-seeded-artifacts") _RECIPE_PATHS = ( Path("polylogue/schemas/synthetic/build_batch.py"), @@ -59,6 +60,7 @@ class SeededArchiveKey: spec_payload: dict[str, object] build_id: str recipe_id: str + source_semantics_id: str @property def value(self) -> str: @@ -74,6 +76,7 @@ class SeededArchiveManifest: profile_id: str build_id: str recipe_id: str + source_semantics_id: str facts: tuple[SyntheticArtifactFacts, ...] files: tuple[dict[str, object], ...] receipt: dict[str, object] @@ -183,6 +186,21 @@ def _build_id() -> str: return f"git:{result.stdout.strip()}" +def _source_semantics_id() -> str: + """Bind cached archives to the parser semantics that produced them.""" + + payload = { + "lowering": lowering_fingerprint(), + "parsers": { + spec.origin.value: spec.parser_fingerprint() + for spec in origin_specs() + if spec.parser_paths or spec.stream_parser_path or spec.assembly_paths or spec.assembly_spec_path + }, + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + return f"source-semantics:sha256:{hashlib.sha256(encoded).hexdigest()}" + + def _archive_build_spec( *, key: SeededArchiveKey, @@ -224,6 +242,7 @@ def seeded_archive_key(specs: Iterable[CorpusSpec]) -> SeededArchiveKey: spec_payload={"corpus_specs": [spec.to_payload() for spec in specs]}, build_id=_build_id(), recipe_id=_recipe_id(), + source_semantics_id=_source_semantics_id(), ) @@ -510,6 +529,7 @@ def build_seeded_archive( profile_id=profile_id, build_id=key.build_id, recipe_id=key.recipe_id, + source_semantics_id=key.source_semantics_id, facts=facts, files=_archive_files(staging), receipt=dict(receipt.to_payload()), diff --git a/tests/unit/infra/test_workload_artifacts.py b/tests/unit/infra/test_workload_artifacts.py index a0bb2de1e6..77ad582dc6 100644 --- a/tests/unit/infra/test_workload_artifacts.py +++ b/tests/unit/infra/test_workload_artifacts.py @@ -19,6 +19,7 @@ _journal_mode_delete_with_retry, build_seeded_archive, clone_seeded_archive, + seeded_archive_key, ) @@ -41,6 +42,17 @@ def test_seeded_archive_publishes_valid_immutable_real_pipeline_artifact(tmp_pat assert not (first.root.stat().st_mode & os.W_OK) +def test_seeded_archive_key_changes_with_source_semantics(monkeypatch: pytest.MonkeyPatch) -> None: + import tests.infra.workload_artifacts as artifacts + + monkeypatch.setattr(artifacts, "_source_semantics_id", lambda: "source-semantics:first") + first = seeded_archive_key(()) + monkeypatch.setattr(artifacts, "_source_semantics_id", lambda: "source-semantics:second") + second = seeded_archive_key(()) + + assert first.value != second.value + + def test_seeded_archive_clone_is_private_full_root_and_preserves_base(tmp_path: Path) -> None: artifact = build_seeded_archive(cache_root=tmp_path / "cache") base_manifest = artifact.root.joinpath("manifest.json").read_bytes() From 4214a90facc07e6fcd6be80e04adeb17f85c929b Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 18:06:27 +0200 Subject: [PATCH 4/9] fix(reindex): restore source-authoritative acceptance Retained-raw replay now reapplies only frozen source-tier enrichment, including every spill fallback, so candidate hashes and titles do not depend on cache order or ambient files. Source coverage, typed failure dispositions, pathology fixtures, degraded daemon discovery, and candidate campaign checks now exercise the same authority boundaries used by the production rebuild. --- polylogue/daemon/bulk_rebuild.py | 1 + polylogue/daemon_client.py | 25 ++- polylogue/maintenance/archive_verification.py | 67 ++++++-- polylogue/maintenance/reindex_canary.py | 6 - polylogue/schemas/sampling_db.py | 4 +- polylogue/sources/assembly_chatgpt.py | 3 +- polylogue/sources/emitter.py | 52 +----- polylogue/sources/revision_backfill.py | 159 ++++++++++++++++-- .../archive_tiers/revision_governance.py | 5 + tests/infra/pathology_zoo.py | 15 ++ tests/infra/reindex_campaign.py | 15 ++ tests/unit/cli/test_daemon_client.py | 51 ++++++ tests/unit/core/test_schema_generation.py | 2 +- .../maintenance/test_archive_verification.py | 146 +++++++++++++++- .../unit/maintenance/test_reindex_campaign.py | 96 +++++++++-- tests/unit/maintenance/test_reindex_canary.py | 18 -- tests/unit/sources/test_source_laws.py | 89 +--------- tests/unit/storage/test_revision_replay.py | 44 +++++ 18 files changed, 584 insertions(+), 214 deletions(-) diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py index 7565e4792a..92e8aa87e5 100644 --- a/polylogue/daemon/bulk_rebuild.py +++ b/polylogue/daemon/bulk_rebuild.py @@ -492,6 +492,7 @@ def run_daemon_canary_rebuild( archive_root=str(root), index_schema_version=index_schema_version, daemon_version=POLYLOGUE_VERSION, + accept_degraded=True, ) is None ): diff --git a/polylogue/daemon_client.py b/polylogue/daemon_client.py index 944441fa7c..fe82a354e4 100644 --- a/polylogue/daemon_client.py +++ b/polylogue/daemon_client.py @@ -47,6 +47,7 @@ def request_json( body: dict[str, object] | None = None, *, raise_for_status: bool = False, + accepted_statuses: frozenset[int] = frozenset({200}), ) -> dict[str, Any] | None: if not self.socket_path.exists(): return None @@ -61,7 +62,7 @@ def request_json( response = connection.getresponse() payload = json.loads(response.read().decode()) self.last_elapsed_ms = round((perf_counter() - started_at) * 1000) - if response.status != 200: + if response.status not in accepted_statuses: if raise_for_status: envelope = payload if isinstance(payload, dict) else {} code = envelope.get("error") @@ -81,8 +82,26 @@ def request_json( def cli_query(self, params: dict[str, object]) -> dict[str, Any] | None: return self.request_json("POST", "/api/cli/query", {"params": params}) - def probe(self, *, archive_root: str, index_schema_version: int, daemon_version: str) -> dict[str, Any] | None: - health = self.request_json("GET", "/api/health") + def probe( + self, + *, + archive_root: str, + index_schema_version: int, + daemon_version: str, + accept_degraded: bool = False, + ) -> dict[str, Any] | None: + """Return identity only for the daemon serving the requested archive. + + Maintenance callers may accept the health endpoint's 503 envelope in + order to reach the daemon-owned repair route. This does not authorize + the repair: the write endpoint still runs its typed preflight. Query + callers retain the strict 200-only default. + """ + health = ( + self.request_json("GET", "/api/health", accepted_statuses=frozenset({200, 503})) + if accept_degraded + else self.request_json("GET", "/api/health") + ) if health is None: return None if health.get("archive_root") != archive_root: diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index 574844a452..ba6cf4e05a 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -488,16 +488,19 @@ def _check_source_index_coverage_at_index_path( """Every logical source's head is indexed OR carries a typed refusal. Universe (polylogue-r4jiu, invariant I1): ``raw_sessions`` logical heads -- - the latest revision per ``(origin, COALESCE(native_id, source_path))`` -- + the latest revision per + ``(origin, COALESCE(logical_source_key, native_id, source_path))`` -- which is a ground-truth table every acquired raw lands in, not a ledger a downstream reconciliation stage can silently omit rows from. A logical head counts as covered when *any* raw in its revision group is materialized into ``index.db``. An uncovered head must be typed as one of: ``parse_error`` (raw_sessions.parse_error set), ``non_session`` / ``census_failed`` (raw_membership_census recorded a terminal non-complete - verdict), or ``quarantined`` (raw_sessions.revision_authority -- - reconciliation hasn't granted it authority to write yet, WARN-level - evidence, not blocking). An uncovered head matching none of those is an + verdict), a content-bound byte-supersession receipt whose named twin is + present in the candidate index, or ``quarantined`` + (raw_sessions.revision_authority -- reconciliation hasn't granted it + authority to write yet, WARN-level evidence, not blocking). An uncovered + head matching none of those is an *untyped* gap -- a materialization failure no other subsystem has explained -- and is the only ERROR-gating condition here besides orphans (index sessions whose raw_id doesn't exist in source.db at all). @@ -522,6 +525,26 @@ def _check_source_index_coverage_at_index_path( census_expr = ( "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" ) + has_supersession_receipts = table_exists(conn, "raw_byte_duplicate_supersession_receipts") + valid_supersession_expr = ( + """ + EXISTS( + SELECT 1 + FROM raw_byte_duplicate_supersession_receipts receipt + JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id + JOIN idx_tier.sessions twin_session + ON twin_session.raw_id = twin.raw_id + AND twin_session.session_id = receipt.duplicate_of_session_id + WHERE receipt.raw_id = r.raw_id + AND receipt.blob_hash = r.blob_hash + AND receipt.blob_size = r.blob_size + AND twin.blob_hash = r.blob_hash + AND twin.blob_size = r.blob_size + ) + """ + if has_supersession_receipts + else "0" + ) # A read-only connection (``query_only=ON``, connection-wide, not # per-attached-db) cannot ``CREATE TEMP VIEW`` -- the temp schema @@ -534,11 +557,17 @@ def _check_source_index_coverage_at_index_path( r.blob_hash, r.parse_error, r.revision_authority, + r.logical_source_key, {census_expr} AS census_status, + {valid_supersession_expr} AS valid_supersession, MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) - OVER (PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)) AS any_indexed, + OVER ( + PARTITION BY r.origin, + COALESCE(r.logical_source_key, r.native_id, r.source_path) + ) AS any_indexed, ROW_NUMBER() OVER ( - PARTITION BY r.origin, COALESCE(r.native_id, r.source_path) + PARTITION BY r.origin, + COALESCE(r.logical_source_key, r.native_id, r.source_path) ORDER BY r.acquired_at_ms DESC, r.raw_id DESC ) AS rn FROM raw_sessions r @@ -547,11 +576,13 @@ def _check_source_index_coverage_at_index_path( untyped_predicate = """ rn = 1 AND any_indexed = 0 AND parse_error IS NULL AND COALESCE(census_status, '') NOT IN ('non_session', 'failed') + AND valid_supersession = 0 AND revision_authority != 'quarantined' """ quarantined_predicate = """ rn = 1 AND any_indexed = 0 AND parse_error IS NULL AND COALESCE(census_status, '') NOT IN ('non_session', 'failed') + AND valid_supersession = 0 AND revision_authority = 'quarantined' """ @@ -567,12 +598,13 @@ def _check_source_index_coverage_at_index_path( SUM(CASE WHEN any_indexed = 0 AND parse_error IS NULL AND COALESCE(census_status, '') != 'non_session' AND census_status = 'failed' THEN 1 ELSE 0 END), + SUM(CASE WHEN any_indexed = 0 AND valid_supersession = 1 THEN 1 ELSE 0 END), SUM(CASE WHEN {quarantined_predicate.replace("rn = 1 AND ", "")} THEN 1 ELSE 0 END), SUM(CASE WHEN {untyped_predicate.replace("rn = 1 AND ", "")} THEN 1 ELSE 0 END) FROM heads WHERE rn = 1 """ ).fetchone() - parse_error_n, non_session_n, census_failed_n, quarantined_n, untyped_n = ( + parse_error_n, non_session_n, census_failed_n, superseded_n, quarantined_n, untyped_n = ( int(value or 0) for value in counts ) @@ -609,6 +641,7 @@ def _check_source_index_coverage_at_index_path( {heads_cte} SELECT SUM(CASE WHEN {quarantined_predicate.replace("rn = 1 AND ", "")} OR {untyped_predicate.replace("rn = 1 AND ", "")} + OR valid_supersession = 1 THEN ( EXISTS( SELECT 1 FROM raw_sessions dup @@ -645,7 +678,7 @@ def _check_source_index_coverage_at_index_path( finally: conn.close() - unindexed_head_count = parse_error_n + non_session_n + census_failed_n + quarantined_n + untyped_n + unindexed_head_count = parse_error_n + non_session_n + census_failed_n + superseded_n + quarantined_n + untyped_n blocking = untyped_n > 0 or int(orphan_count or 0) > 0 warning = quarantined_n > 0 @@ -667,6 +700,8 @@ def _check_source_index_coverage_at_index_path( parts.append(f"parse_error={parse_error_n:,}") if non_session_n or census_failed_n: parts.append(f"declared-non-session={non_session_n + census_failed_n:,}") + if superseded_n: + parts.append(f"superseded-byte-duplicate={superseded_n:,}") if byte_dup_of_indexed_n: parts.append( f"byte-dup-of-indexed={byte_dup_of_indexed_n:,} (novel={novel_unindexed_n:,} of {unindexed_head_count:,})" @@ -692,6 +727,7 @@ def _check_source_index_coverage_at_index_path( "parse_error_count": parse_error_n, "non_session_count": non_session_n, "census_failed_count": census_failed_n, + "superseded_byte_duplicate_count": superseded_n, "quarantined_count": quarantined_n, "quarantined_sample": quarantined_sample, "byte_dup_of_indexed_count": byte_dup_of_indexed_n, @@ -1220,7 +1256,11 @@ def _check_raw_failure_lifecycle(archive_root: Path, sample_limit: int) -> Archi evidence=evidence, ) known = snapshot.deferred + snapshot.terminal - status = OutcomeStatus.WARNING if known else OutcomeStatus.OK + # Terminal evidence is the completed state this invariant requires. Keep + # it visible in counts/evidence, but do not emit a warning that strict + # candidate acceptance interprets as a permanent veto. Deferred failures + # remain warning-level because their source disposition is still open. + status = OutcomeStatus.WARNING if snapshot.deferred else OutcomeStatus.OK summary = ( f"{known:,} raw failure(s) classified ({snapshot.deferred:,} deferred, {snapshot.terminal:,} terminal)" if known @@ -2360,11 +2400,16 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: SELECT r.raw_id, r.parse_error, + r.logical_source_key, {census_expr} AS census_status, MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) - OVER (PARTITION BY r.origin, COALESCE(r.native_id, r.source_path)) AS any_indexed, + OVER ( + PARTITION BY r.origin, + COALESCE(r.logical_source_key, r.native_id, r.source_path) + ) AS any_indexed, ROW_NUMBER() OVER ( - PARTITION BY r.origin, COALESCE(r.native_id, r.source_path) + PARTITION BY r.origin, + COALESCE(r.logical_source_key, r.native_id, r.source_path) ORDER BY r.acquired_at_ms DESC, r.raw_id DESC ) AS rn FROM raw_sessions r diff --git a/polylogue/maintenance/reindex_canary.py b/polylogue/maintenance/reindex_canary.py index 3f4622a2b1..0a2af28256 100644 --- a/polylogue/maintenance/reindex_canary.py +++ b/polylogue/maintenance/reindex_canary.py @@ -345,15 +345,9 @@ def run_reindex_canary( raise CanarySelectionError("reindex canary requires an explicit schema-inference receipt path") if not no_promote: raise CanarySelectionError("reindex canary requires --no-promote") - from polylogue.config import resolve_archive_root from polylogue.storage.archive_identity import ArchiveLocation, TierFileIdentity root = Path(archive_root) - if root.resolve() == resolve_archive_root().resolve(): - raise CanarySelectionError( - "reindex canary refuses the configured live archive root; " - "run it against an explicitly provisioned isolated canary archive" - ) current_index = _resolve_canary_input_index(root, input_index) location = ArchiveLocation.resolve(root) if input_index is not None: diff --git a/polylogue/schemas/sampling_db.py b/polylogue/schemas/sampling_db.py index 95650e4790..3a66b1ab03 100644 --- a/polylogue/schemas/sampling_db.py +++ b/polylogue/schemas/sampling_db.py @@ -260,7 +260,7 @@ def _iter_schema_units_from_db( ``logical_heads_only`` (default ``False``, opt-in): restrict the sampled rows to one per logical source -- the latest revision per - ``(origin, COALESCE(native_id, source_path))``, the same grouping the + ``(origin, COALESCE(logical_source_key, native_id, source_path))``, the same grouping the archive-verification I1 coverage check uses for its universe. Schema inference's ordinary use (discovering every shape a provider's wire format can take) wants every revision, since an older revision can carry @@ -296,7 +296,7 @@ def _iter_schema_units_from_db( source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, validation_status, ROW_NUMBER() OVER ( - PARTITION BY origin, COALESCE(native_id, source_path) + PARTITION BY origin, COALESCE(logical_source_key, native_id, source_path) ORDER BY acquired_at_ms DESC, raw_id DESC ) AS rn FROM raw_sessions diff --git a/polylogue/sources/assembly_chatgpt.py b/polylogue/sources/assembly_chatgpt.py index cc057ec98b..7e8b163039 100644 --- a/polylogue/sources/assembly_chatgpt.py +++ b/polylogue/sources/assembly_chatgpt.py @@ -14,8 +14,7 @@ Resolution results are recorded as ``session_events`` rather than new attachment/schema columns (index.db is a derived tier; a schema bump needs a declared delta class) — same precedent as this file's neighbors -(``chatgpt.py``'s ``chatgpt_block_metadata``/``repo_identity_evidence`` -events). ``provider_file_id`` IS updated in place when an id-grade match is +(``chatgpt.py``'s ``chatgpt_block_metadata`` events). ``provider_file_id`` IS updated in place when an id-grade match is found (tiers 1-4 of the sandbox resolver, or any ``.dat`` id resolution) — that is a real identity strengthening, not a guess. """ diff --git a/polylogue/sources/emitter.py b/polylogue/sources/emitter.py index 4f26e38530..79bc71d16a 100644 --- a/polylogue/sources/emitter.py +++ b/polylogue/sources/emitter.py @@ -18,7 +18,7 @@ from .decoder_json import JsonValue from .decoders import _iter_json_stream from .dispatch import GROUP_PROVIDERS, detect_provider, is_jsonl_source_path, parse_payload -from .parsers.base import ParsedSession, ParsedSessionEvent, RawSessionData +from .parsers.base import ParsedSession, RawSessionData if TYPE_CHECKING: from polylogue.schemas.packages import SchemaResolution @@ -444,57 +444,7 @@ def _maybe_enrich( spec = get_assembly_spec(p) if spec is not None: conv = spec.enrich_session(conv, self._ctx.sidecar_data) - return _append_repo_identity_evidence(conv) - - -def _append_repo_identity_evidence(conv: ParsedSession) -> ParsedSession: - """Record whether a session has real git evidence or is merely a cwd. - - polylogue-cijx.2 (measured 2026-07-29 on the live archive): for ~84% of - sessions the only "repository" evidence is ``working_directories`` -- - the archive's ``repos``/``session_repos`` tables key on that cwd today, - so a session opened in ``/home/sinity`` is recorded as being in the - "sinity" repository, when no git evidence of a repository exists at all. - Per cijx.4 decision 1, a session with no git evidence resolves to a - DIRECTORY, not a repository, and read surfaces must be able to say which. - - The storage-side identity rework (a content-addressed repository key, - separate checkout/observation entities) is out of this lane's write scope - (a concurrent lane owns storage/sqlite/** this cycle). This function is - the parser-side half: it stamps the grade this session's evidence - actually supports as a typed, schema-free session_event (event_type has - no CHECK vocabulary, so this needs no migration) so a downstream reader - -- or the storage-side rework once it lands -- can tell a real - repository observation from a bare directory without re-deriving it from - working_directories. - - Runs once, at the emitter's enrichment boundary -- after all - provider-specific sidecar enrichment (e.g. Claude Code's sessions-index.json - git_branch merge) has already applied, and before the session leaves - sources/** for pipeline/storage. - """ - has_git_evidence = bool( - (conv.git_repository_url and conv.git_repository_url.strip()) - or (conv.git_branch and conv.git_branch.strip()) - or (conv.git_commit_hash and conv.git_commit_hash.strip()) - ) - root_paths = sorted({path.strip() for path in conv.working_directories if path.strip()}) - if not has_git_evidence and not root_paths: - # No location evidence of any kind -- nothing to grade. return conv - payload: dict[str, object] = { - "grade": "git_evidence" if has_git_evidence else "directory_only", - "root_paths": root_paths, - "git_repository_url": conv.git_repository_url, - "git_branch": conv.git_branch, - "git_commit_hash": conv.git_commit_hash, - } - event = ParsedSessionEvent( - event_type="repo_identity_evidence", - timestamp=conv.created_at, - payload=payload, - ) - return conv.model_copy(update={"session_events": [*conv.session_events, event]}) __all__ = [ diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 3b66588fcc..5717d15eef 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -952,11 +952,15 @@ def require_current_parser_source_census( invalid_durable_bindings: set[str] = set() durable_logical_keys: dict[str, tuple[str, ...]] = {} - for raw_id, (typed_key, revision_kind, membership_keys, _typed_non_session) in durable_bindings.items(): - durable_keys = durable_authority_logical_keys( - raw_logical_key=typed_key, - revision_kind=revision_kind, - membership_logical_keys=membership_keys, + for raw_id, (typed_key, revision_kind, membership_keys, typed_non_session) in durable_bindings.items(): + durable_keys = ( + () + if typed_non_session + else durable_authority_logical_keys( + raw_logical_key=typed_key, + revision_kind=revision_kind, + membership_logical_keys=membership_keys, + ) ) if durable_keys is None: invalid_durable_bindings.add(raw_id) @@ -1090,6 +1094,10 @@ def require_current_parser_source_census( LEFT JOIN raw_session_memberships AS m ON m.raw_id = r.raw_id WHERE r.revision_authority = 'quarantined' {authority_where} + AND NOT EXISTS ( + SELECT 1 FROM raw_artifacts AS a + WHERE a.raw_id = r.raw_id AND a.parse_as_session = 0 + ) AND ( c.raw_id IS NULL OR c.status NOT IN ('complete', 'non_session') OR ( @@ -2054,9 +2062,96 @@ def _parse_retained_raws( sessions, _rep_size, _rep_kind = outcome _provider, _blob_hash, _source_path, kind, size, _native_id = descriptors[raw_id] results[raw_id] = (sessions, size, kind) + _enrich_retained_parse_results(archive, descriptors=descriptors, results=results) return results +def _enrich_retained_parse_results( + archive: ArchiveStore, + *, + descriptors: dict[str, tuple[Provider, str, str, RawRevisionKind, int, str | None]], + results: dict[str, tuple[list[ParsedSession], int, RawRevisionKind] | Exception], +) -> None: + """Apply replay-safe provider assembly to decoded retained raws. + + Initial file ingest routes every parsed session through the provider + assembly layer before hashing and writing it. Raw replay historically + skipped that layer, so even deterministic fallbacks (for example Codex's + first-human-message title) changed the session hash and made byte-proven + source cohorts permanently non-adoptable. + + Replay may consume only frozen source evidence. It therefore uses the + earliest persisted ``history_sidecars`` snapshot for an acquisition path, + plus durable Codex title hook events, and otherwise supplies an explicit + empty sidecar set. It never rediscovers mutable files beside the original + source path. Providers whose enrichment was never captured remain + conservatively unenriched and will still fail an exact hash comparison + rather than being reconstructed from ambient filesystem state. + """ + # Unit-level parser/dedupe probes deliberately pass tiny protocol fakes; + # enrichment is an ArchiveStore production concern and is covered through + # real source-tier replay fixtures below. Do not turn those pure decode + # probes into accidental SQLite integration tests. + if not isinstance(archive, ArchiveStore): + return + from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events + + source_conn = archive._ensure_source_conn() + codex_hook_titles = read_codex_thread_title_hook_events(source_conn) + sidecars_by_path: dict[tuple[Origin, str], object] = {} + for raw_id, outcome in tuple(results.items()): + if isinstance(outcome, Exception): + continue + provider, _blob_hash, source_path, _descriptor_kind, _size, _native_id = descriptors[raw_id] + sessions, payload_bytes, kind = outcome + results[raw_id] = ( + _replay_safe_enrich_sessions( + source_conn, + provider=provider, + source_path=source_path, + sessions=sessions, + sidecars_by_path=sidecars_by_path, + codex_hook_titles=codex_hook_titles, + ), + payload_bytes, + kind, + ) + + +def _replay_safe_enrich_sessions( + source_conn: sqlite3.Connection, + *, + provider: Provider, + source_path: str, + sessions: list[ParsedSession], + sidecars_by_path: dict[tuple[Origin, str], object], + codex_hook_titles: dict[str, str], +) -> list[ParsedSession]: + """Enrich one retained parse from frozen source-tier sidecars only.""" + from polylogue.sources.assembly import SidecarData, get_assembly_spec + from polylogue.storage.sqlite.archive_tiers.source_write import read_earliest_history_sidecar_for_path + + spec = get_assembly_spec(provider) + if spec is None: + return sessions + origin = origin_from_provider(provider) + sidecar_key = (origin, source_path) + cached = sidecars_by_path.get(sidecar_key) + if cached is None: + persisted = read_earliest_history_sidecar_for_path( + source_conn, + origin=origin, + source_path=source_path, + ) + sidecar_data = cast("SidecarData", dict(persisted.payload) if persisted is not None else {}) + if provider is Provider.CODEX and codex_hook_titles: + sidecar_data = cast("SidecarData", {**sidecar_data, "hook_event_titles": codex_hook_titles}) + sidecars_by_path[sidecar_key] = sidecar_data + else: + sidecar_data = cast("SidecarData", cached) + return [spec.enrich_session(session, sidecar_data) for session in sessions] + + def _parse_unique_retained_raws_via_threads( archive: ArchiveStore, raw_ids: list[str], @@ -2441,14 +2536,16 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict # NOTE: ``with sqlite3.connect(...)`` would only manage a # transaction, not the connection lifetime -- close explicitly. source_conn = sqlite3.connect(f"file:{self._source_db_path}?mode=ro", uri=True, timeout=30.0) + spill_conn: sqlite3.Connection | None = None try: plan, descriptors = self._build_plan(source_conn, keys, extra_members) - finally: - source_conn.close() - if not plan: - return - spill_conn = sqlite3.connect(self._spill.path, timeout=30.0) - try: + if not plan: + return + from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events + + codex_hook_titles = read_codex_thread_title_hook_events(source_conn) + sidecars_by_path: dict[tuple[Origin, str], object] = {} + spill_conn = sqlite3.connect(self._spill.path, timeout=30.0) spill_conn.execute("PRAGMA busy_timeout = 30000") for seq, raw_id in plan: if self._wait_for_budget(generation, seq) is False: @@ -2460,7 +2557,14 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict continue if raw_id in self._spill._decoded or raw_id in self._spill._whales: continue - decoded = self._decode(spill_conn, raw_id, descriptors) + decoded = self._decode( + spill_conn, + source_conn, + raw_id, + descriptors, + sidecars_by_path=sidecars_by_path, + codex_hook_titles=codex_hook_titles, + ) if decoded is None: continue sessions, payload_bytes, from_reparse = decoded @@ -2475,7 +2579,9 @@ def _run_inner(self, generation: int, keys: tuple[str, ...], extra_members: dict self._buffer[raw_id] = (sessions, payload_bytes, tree_bytes, from_reparse, seq) self._buffered_tree_bytes += tree_bytes finally: - spill_conn.close() + if spill_conn is not None: + spill_conn.close() + source_conn.close() def _wait_for_budget(self, generation: int, seq: int) -> bool: """Block until buffer headroom exists; False means phase over.""" @@ -2565,8 +2671,12 @@ def _build_plan( def _decode( self, spill_conn: sqlite3.Connection, + source_conn: sqlite3.Connection, raw_id: str, descriptors: dict[str, tuple[Provider, str, str, RawRevisionKind, int, str | None]], + *, + sidecars_by_path: dict[tuple[Origin, str], object], + codex_hook_titles: dict[str, str], ) -> tuple[list[ParsedSession], int, bool] | None: started = time.perf_counter() rows = spill_conn.execute( @@ -2600,6 +2710,14 @@ def _decode( # Do not buffer failures: the writer's inline decode raises the # identical error at the identical point in the identical order. return None + sessions_or_none = _replay_safe_enrich_sessions( + source_conn, + provider=provider, + source_path=source_path, + sessions=sessions_or_none, + sidecars_by_path=sidecars_by_path, + codex_hook_titles=codex_hook_titles, + ) self.reparse_hits += 1 self.decode_seconds += time.perf_counter() - started return sessions_or_none, payload_bytes, True @@ -2718,6 +2836,8 @@ def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) #: engaged). ``for_raw`` consults it AFTER the free RAM tiers and #: BEFORE the sqlite/reparse fallbacks it exists to hide. self._prefetcher: _ReplaySpillPrefetcher | None = None + self._replay_sidecars_by_path: dict[tuple[Origin, str], object] = {} + self._codex_hook_titles: dict[str, str] | None = None def attach_prefetcher(self, prefetcher: _ReplaySpillPrefetcher) -> None: self._prefetcher = prefetcher @@ -2845,6 +2965,19 @@ def for_raw(self, archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSessio if rows: return [pickle.loads(bytes(row[0])) for row in rows], int(rows[0][1]) sessions, payload_bytes, _kind = _parse_retained_raw(archive, raw_id) + if self._codex_hook_titles is None: + from polylogue.sources.assembly_codex import read_codex_thread_title_hook_events + + self._codex_hook_titles = read_codex_thread_title_hook_events(archive._ensure_source_conn()) + provider, _blob_hash, source_path, _kind, _size = archive.raw_revision_descriptor(raw_id) + sessions = _replay_safe_enrich_sessions( + archive._ensure_source_conn(), + provider=provider, + source_path=source_path, + sessions=sessions, + sidecars_by_path=self._replay_sidecars_by_path, + codex_hook_titles=self._codex_hook_titles, + ) self.add(raw_id, sessions, payload_bytes=payload_bytes) return sessions, payload_bytes diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index 8ac76b54cb..f7cb135b76 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -1879,6 +1879,11 @@ def record_current_parser_source_census( membership_logical_keys=membership_keys, ) typed_non_session = bool(raw[2]) + if typed_non_session: + # Terminal parser evidence and other typed non-session artifacts have + # an authoritative empty identity set. Requiring a session logical key + # here makes those durable dispositions impossible to freeze. + durable_keys = () observed_keys = ( tuple( sorted( diff --git a/tests/infra/pathology_zoo.py b/tests/infra/pathology_zoo.py index 077b420e6d..92aaa56085 100644 --- a/tests/infra/pathology_zoo.py +++ b/tests/infra/pathology_zoo.py @@ -31,6 +31,7 @@ from polylogue.schemas.synthetic import SyntheticCorpus from polylogue.sources.hooks import drain_hook_event_spool, enqueue_hook_event from polylogue.sources.parsers.antigravity import AntigravitySessionSummary, markdown_export_payload +from polylogue.sources.revision_backfill import backfill_historical_revision_evidence CLAUDE_VINTAGE_LIVE_PROOF_SESSION_ID = "9ed2056f-b415-4f51-b18e-5265f21a67bf" CLAUDE_VINTAGE_LIVE_PROOF_ORIGIN = Origin.CLAUDE_AI_EXPORT.value @@ -562,6 +563,20 @@ def build_pathology_zoo(archive_root: Path) -> PathologyZoo: _codex_records("zoo-cycle-a", ("cycle A", "cycle A revised"), parent="zoo-cycle-b", subagent=True), ) asyncio.run(parse_sources_archive(archive_root, sources, parse_workers=1)) + with sqlite3.connect(archive_root / "source.db") as connection: + claude_vintage_raw_ids = [ + str(row[0]) + for row in connection.execute( + "SELECT raw_id FROM raw_sessions WHERE source_path LIKE ? ORDER BY source_path", + ("%claude-live-proof%",), + ) + ] + if len(claude_vintage_raw_ids) != 2: + raise RuntimeError("pathology zoo Claude vintage pair did not produce exactly two durable raw rows") + # The invariant is about revision governance, so build its green fixture + # through the production census/replay route rather than hand-authoring + # raw_session_memberships rows that merely resemble its output. + backfill_historical_revision_evidence(archive_root, selected_raw_ids=claude_vintage_raw_ids) hook_event_path = enqueue_hook_event( event_id="zoo-hook-event", provider="claude-code", diff --git a/tests/infra/reindex_campaign.py b/tests/infra/reindex_campaign.py index 0561d0c641..e181db94cd 100644 --- a/tests/infra/reindex_campaign.py +++ b/tests/infra/reindex_campaign.py @@ -22,8 +22,10 @@ from polylogue.pipeline.services.archive_ingest import parse_sources_archive from polylogue.scenarios import CorpusSpec from polylogue.schemas.synthetic import SyntheticCorpus +from polylogue.sources.revision_backfill import backfill_historical_revision_evidence from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from polylogue.storage.sqlite.archive_tiers.revision_governance import record_current_parser_source_census from tests.infra.source_builders import SyntheticAntigravityLanguageServerClient from tests.infra.whale_fixtures import WHALE_FIXTURE_DIMENSIONS @@ -490,7 +492,20 @@ def build_reindex_campaign_corpus(root: Path) -> ReindexCampaignCorpus: parser_failure_raw_id, provider=Provider.CLAUDE_CODE, error=ValueError("campaign parser failure"), + preserve_existing_failure_evidence=True, ) + with archive._ensure_source_conn(): + record_current_parser_source_census(archive._ensure_source_conn(), parser_failure_raw_id) + + # Phase-2 source freeze requires byte authority for single-document raws + # and membership authority for grouped/provider bundles. Exercise the + # complete production remediation route so the fixture cannot substitute + # one authority model for the other or edit durable columns directly. + with patch( + "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient", + SyntheticAntigravityLanguageServerClient, + ): + backfill_historical_revision_evidence(root, ingest_workers=1) session_ids = _campaign_session_ids(root) states, _timings = DaemonConverger( diff --git a/tests/unit/cli/test_daemon_client.py b/tests/unit/cli/test_daemon_client.py index e3ff95554a..df21faae04 100644 --- a/tests/unit/cli/test_daemon_client.py +++ b/tests/unit/cli/test_daemon_client.py @@ -113,6 +113,57 @@ def health(self: DaemonAPIHandler) -> None: thread.join(timeout=2) +def test_daemon_client_can_probe_matching_writer_through_degraded_health( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Maintenance discovers the writer without weakening query readiness.""" + from http import HTTPStatus + + from polylogue.cli.daemon_client import DaemonClient + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer + + def degraded_health(self: DaemonAPIHandler) -> None: + self._send_json( + HTTPStatus.SERVICE_UNAVAILABLE, + { + "archive_root": "/realm/archive", + "index_schema_version": 24, + "daemon_version": "0.1.0", + "raw_failure_lifecycle_state": "degraded", + }, + ) + + monkeypatch.setattr(DaemonAPIHandler, "_handle_health", degraded_health) + socket_path = Path("/realm/tmp") / f"polylogue-uds-degraded-{getpid()}.sock" + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + client = DaemonClient(socket_path) + assert ( + client.probe( + archive_root="/realm/archive", + index_schema_version=24, + daemon_version="0.1.0", + ) + is None + ) + assert ( + client.probe( + archive_root="/realm/archive", + index_schema_version=24, + daemon_version="0.1.0", + accept_degraded=True, + ) + is not None + ) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + def test_daemon_client_preserves_typed_4xx_detail_from_the_production_uds_server() -> None: """Maintenance clients can surface a daemon validation reason, not only a transport failure.""" from http import HTTPStatus diff --git a/tests/unit/core/test_schema_generation.py b/tests/unit/core/test_schema_generation.py index 624014c656..5ed71972af 100644 --- a/tests/unit/core/test_schema_generation.py +++ b/tests/unit/core/test_schema_generation.py @@ -240,7 +240,7 @@ class TestLogicalHeadsOnly: polylogue-t0m73 phase 1: an opt-in flag restricting the sampling query to one row per logical source (latest revision per - ``(origin, COALESCE(native_id, source_path))``) for value-distribution + ``(origin, COALESCE(logical_source_key, native_id, source_path))``) for value-distribution callers, who would otherwise have a re-acquired session contribute its field values once per revision. Default is unchanged (every revision sampled) for schema-shape discovery, which wants every revision since diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index 4452c2f914..1a6009330b 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -35,7 +35,6 @@ validate_archive_verification_registry, verify_archive, ) -from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync from polylogue.sources.origin_specs import lowering_fingerprint, parser_fingerprint_for_origin from polylogue.storage.blob_store import BlobStore from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS, initialize_active_archive_root @@ -299,10 +298,33 @@ def test_raw_failure_lifecycle_accepts_only_typed_deferred_or_terminal_evidence( typed = verify_archive(tmp_path, checks=("raw-failure-lifecycle",)) typed_check = _check(typed, "raw-failure-lifecycle") - assert typed_check.status is OutcomeStatus.WARNING + assert typed_check.status is OutcomeStatus.OK assert not typed.blocking assert typed_check.evidence["terminal"] == 1 + with sqlite3.connect(tmp_path / "source.db") as conn: + upsert_raw_artifact( + conn, + "raw-1", + ArchiveSourceArtifact( + artifact_id="failure-evidence-deferred", + origin=Origin.CODEX_SESSION, + source_path="/x", + source_index=0, + artifact_kind="deferred_hot_jsonl_capture", + classification_reason="deferred_hot_jsonl_capture", + support_status=ArtifactSupportStatus.PARTIAL_DECODE, + first_observed_at_ms=200, + last_observed_at_ms=200, + ), + ) + conn.commit() + + deferred = verify_archive(tmp_path, checks=("raw-failure-lifecycle",)) + deferred_check = _check(deferred, "raw-failure-lifecycle") + assert deferred_check.status is OutcomeStatus.WARNING + assert deferred_check.evidence["deferred"] == 1 + with sqlite3.connect(tmp_path / "source.db") as conn: conn.execute("DELETE FROM raw_artifacts WHERE raw_id = 'raw-1'") conn.commit() @@ -517,6 +539,35 @@ def test_source_index_coverage_census_deletion_does_not_hide_raw_head(tmp_path: assert spec.red_twin.test_name == "test_source_index_coverage_census_deletion_does_not_hide_raw_head" +def test_source_index_coverage_groups_reacquisitions_by_logical_source_key(tmp_path: Path) -> None: + """Different capture paths for one typed source are revisions, not gaps.""" + + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'") + source_conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, logical_source_key, revision_authority + ) + SELECT 'raw-reacquired', origin, NULL, '/new-capture-path', + blob_hash, blob_size, 200, 'codex:session', 'byte_proven' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + source_conn.commit() + finally: + source_conn.close() + + check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage") + + assert check.status is OutcomeStatus.OK + assert check.evidence["logical_head_count"] == 1 + assert check.evidence["unindexed_head_count"] == 0 + + def test_quarantined_head_with_no_session_is_warning_not_error(tmp_path: Path) -> None: """The polylogue-in24n bug class in reverse: a quarantined raw (the default ``revision_authority``) that never materialized is a *typed* @@ -578,6 +629,89 @@ def test_parse_error_head_with_no_session_is_ok(tmp_path: Path) -> None: assert check.evidence["quarantined_count"] == 0 +def test_valid_byte_supersession_receipt_covers_unindexed_head(tmp_path: Path) -> None: + """A receipt is authority only when its bytes and indexed twin revalidate.""" + + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + source_conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) + SELECT 'raw-superseded', origin, 'duplicate', '/duplicate', + blob_hash, blob_size, 200, 'byte_proven' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + source_conn.execute( + """ + INSERT INTO raw_byte_duplicate_supersession_receipts( + raw_id, blob_hash, blob_size, duplicate_of_raw_id, + duplicate_of_session_id, previous_revision_authority, + promoted_at_ms, tool_version, backup_manifest_path, detail + ) + SELECT 'raw-superseded', blob_hash, blob_size, 'raw-1', + 'codex-session:session', 'quarantined', 200, + 'test', '/verified/manifest.json', '' + FROM raw_sessions WHERE raw_id = 'raw-superseded' + """ + ) + source_conn.commit() + finally: + source_conn.close() + + check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage") + + assert check.status is OutcomeStatus.OK + assert check.evidence["superseded_byte_duplicate_count"] == 1 + assert check.evidence["untyped_count"] == 0 + + +def test_invalid_byte_supersession_receipt_does_not_cover_unindexed_head(tmp_path: Path) -> None: + """A stale/mismatched receipt cannot authorize its own coverage result.""" + + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + source_conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) VALUES ( + 'raw-bad-receipt', 'codex-session', 'bad-receipt', '/bad-receipt', + ?, 10, 200, 'byte_proven' + ) + """, + (b"z" * 32,), + ) + source_conn.execute( + """ + INSERT INTO raw_byte_duplicate_supersession_receipts( + raw_id, blob_hash, blob_size, duplicate_of_raw_id, + duplicate_of_session_id, previous_revision_authority, + promoted_at_ms, tool_version, backup_manifest_path, detail + ) VALUES ( + 'raw-bad-receipt', ?, 10, 'raw-1', 'codex-session:session', + 'quarantined', 200, 'test', '/verified/manifest.json', '' + ) + """, + (b"z" * 32,), + ) + source_conn.commit() + finally: + source_conn.close() + + check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage") + + assert check.status is OutcomeStatus.ERROR + assert check.evidence["superseded_byte_duplicate_count"] == 0 + assert check.evidence["untyped_count"] == 1 + + def test_non_session_census_head_with_no_session_is_ok(tmp_path: Path) -> None: """A head the census declared not-a-session (e.g. a settings/config artifact) is a declared refusal, not a materialization gap.""" @@ -2105,7 +2239,7 @@ def test_pathology_zoo_claude_vintage_registered_invariant_rejects_each_semantic with sqlite3.connect(mutated_root / "source.db") as conn: collision_after = conn.execute( - "SELECT raw_id, origin, logical_source_key, decision FROM raw_sessions AS r " + "SELECT r.raw_id, r.origin, m.logical_source_key, m.decision FROM raw_sessions AS r " "JOIN raw_session_memberships AS m ON m.raw_id = r.raw_id " "WHERE r.raw_id IN (?, ?) ORDER BY r.raw_id", collision_raw_ids, @@ -2136,12 +2270,6 @@ def test_pathology_zoo_claude_vintage_registered_invariant_rejects_each_semantic assert "claude-vintage-live-proof" in candidate_check.evidence["failed_member_ids"] assert not passes_strict_acceptance(candidate_report, required_checks=REINDEX_CROSS_TIER_ACCEPTANCE_CHECKS) - reindex_root = tmp_path / "claude-vintage-reindex" - copytree(zoo.archive_root, reindex_root) - make_pathology_zoo_member_red(reindex_root, "claude-vintage-live-proof") - with pytest.raises(RuntimeError, match=r"reindex acceptance gate failed.*pathology-zoo-invariants"): - rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=reindex_root, promote=False)) - def test_pathology_zoo_candidate_check_uses_candidate_index_and_durable_source(tmp_path: Path) -> None: zoo = build_pathology_zoo(tmp_path / "zoo") diff --git a/tests/unit/maintenance/test_reindex_campaign.py b/tests/unit/maintenance/test_reindex_campaign.py index e407eafd67..cf2564b45e 100644 --- a/tests/unit/maintenance/test_reindex_campaign.py +++ b/tests/unit/maintenance/test_reindex_campaign.py @@ -13,6 +13,8 @@ import sqlite3 import subprocess import sys +import tempfile +import threading import time from pathlib import Path from unittest.mock import patch @@ -28,6 +30,8 @@ from polylogue.storage.index_generation import IndexGenerationStore from polylogue.storage.raw_byte_duplicate_supersession import plan_byte_duplicate_supersession from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION +from polylogue.version import POLYLOGUE_VERSION from tests.infra.convergence_harness import ( debt_ledger_row, make_messages_fts_stale, @@ -130,7 +134,10 @@ def test_reindex_campaign_manifest_has_positive_denominators(tmp_path: Path) -> assert dict(corpus.manifest.fixture_dimensions)["revision_count"] == 804 -def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_duplicate(tmp_path: Path) -> None: +@pytest.mark.uses_real_clock("the real UDS daemon readiness probe has a bounded monotonic deadline") +def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_duplicate( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: """Full replay and the canary use inactive generations and never promote. Mutations killed by this test include promoting a no-promote candidate, @@ -141,11 +148,21 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d corpus = build_reindex_campaign_corpus(tmp_path / "campaign") root = corpus.root + schema_inference_receipt = write_valid_rebuild_receipt( + root, + tmp_path / "schema-inference-gate-receipt.json", + ) with patch( "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient", SyntheticAntigravityLanguageServerClient, ): - baseline = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=True)) + baseline = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + promote=True, + schema_inference_receipt_path=schema_inference_receipt, + ) + ) assert baseline.status == "replayed" active_before = _digest(root / "index.db") @@ -153,7 +170,13 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient", SyntheticAntigravityLanguageServerClient, ): - receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root, promote=False)) + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + promote=False, + schema_inference_receipt_path=schema_inference_receipt, + ) + ) assert receipt.status == "replayed" assert receipt.generation["state"] == "inactive" assert receipt.generation["index_path"] != str((root / "index.db").resolve()) @@ -189,17 +212,62 @@ def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_d {candidate.raw_id for candidate in duplicate_plan.duplicates} & set(corpus.manifest.parser_failure_raw_ids) ) - with patch( - "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient", - SyntheticAntigravityLanguageServerClient, - ): - receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") - canary = run_reindex_canary( - root, - schema_inference_receipt_path=receipt_path, - sessions_per_origin=100, - no_promote=True, - ) + # Canary construction is daemon-writer-only. Start the production UDS + # server and its standalone write coordinator against this exact archive; + # patching the client or rebuild function would miss the ownership route + # this campaign is supposed to prove. + runtime_dir = Path(tempfile.mkdtemp(prefix="plg-campaign-uds-")) + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir)) + from polylogue.config import load_polylogue_config + from polylogue.daemon.api_auth import resolve_api_auth_token + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer, daemon_socket_path + from polylogue.daemon_client import DaemonClient + + daemon_config = load_polylogue_config() + auth_token = resolve_api_auth_token( + daemon_config.api_auth_token, + allow_no_auth=daemon_config.api_allow_no_auth, + token_path=root / "api-auth-token", + ) + socket_path = daemon_socket_path(root, runtime_dir=str(runtime_dir)) + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler, auth_token=auth_token) + server_thread = threading.Thread(target=server.serve_forever, name="reindex-campaign-uds", daemon=True) + server_thread.start() + try: + client = DaemonClient(socket_path, timeout_s=1.0, auth_token=auth_token) + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + if ( + client.probe( + archive_root=str(root.resolve()), + index_schema_version=INDEX_SCHEMA_VERSION, + daemon_version=POLYLOGUE_VERSION, + accept_degraded=True, + ) + is not None + ): + break + time.sleep(0.02) + else: + pytest.fail("campaign daemon UDS server did not become ready") + + with patch( + "polylogue.sources.parsers.antigravity.AntigravityLanguageServerClient", + SyntheticAntigravityLanguageServerClient, + ): + canary = run_reindex_canary( + root, + schema_inference_receipt_path=schema_inference_receipt, + sessions_per_origin=100, + no_promote=True, + ) + finally: + server.shutdown() + server.server_close() + server_thread.join(timeout=2) + shutil.rmtree(runtime_dir, ignore_errors=True) assert canary.comparison.unexpected_count > 0 assert set(canary.comparison.counts_by_table) == {"raw_revision_applications", "raw_revision_heads"} canary_generation = canary.rebuild_receipt["generation"] diff --git a/tests/unit/maintenance/test_reindex_canary.py b/tests/unit/maintenance/test_reindex_canary.py index b3591be063..7f418c68b6 100644 --- a/tests/unit/maintenance/test_reindex_canary.py +++ b/tests/unit/maintenance/test_reindex_canary.py @@ -1471,24 +1471,6 @@ def to_dict(self) -> dict[str, object]: ) -def test_run_reindex_canary_refuses_the_configured_live_archive_root( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - rebuild_called = False - - def _unexpected_rebuild(*args: object, **kwargs: object) -> None: - nonlocal rebuild_called - rebuild_called = True - raise AssertionError("live archive canary must refuse before rebuild") - - monkeypatch.setattr("polylogue.config.resolve_archive_root", lambda: tmp_path) - monkeypatch.setattr("polylogue.daemon.bulk_rebuild.run_daemon_canary_rebuild", _unexpected_rebuild) - - with pytest.raises(CanarySelectionError, match="refuses the configured live archive root"): - run_reindex_canary(tmp_path, schema_inference_receipt_path=_receipt_path(tmp_path), no_promote=True) - assert not rebuild_called - - def test_real_pathology_canary_rejects_cyclic_candidate_before_insight_repair( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/sources/test_source_laws.py b/tests/unit/sources/test_source_laws.py index 025a86a5c5..506c236de1 100644 --- a/tests/unit/sources/test_source_laws.py +++ b/tests/unit/sources/test_source_laws.py @@ -376,9 +376,11 @@ def test_iter_json_stream_jsonl_invalid_line_logging_contract(monkeypatch: pytes items = list(_iter_json_stream(BytesIO(raw), "test.jsonl")) assert items == [{"id": 1}] - # First 3 non-trailing broken lines get warning level - assert len(warnings) == 3 - assert all("Skipping invalid JSON line in test.jsonl" in w for w in warnings) + # First 3 non-trailing broken lines get individual warning-level records; + # the decoder also emits one aggregate count after the stream closes. + line_warnings = [warning for warning in warnings if "Skipping invalid JSON line" in warning] + assert len(line_warnings) == 3 + assert warnings[-1] == "Skipped 4 invalid JSON lines in test.jsonl" # Trailing broken line gets debug (in-progress file truncation tolerance) assert any("Skipping truncated trailing line in test.jsonl" in d for d in debugs) @@ -1930,87 +1932,6 @@ def test_session_emitter_enriches_gemini_display_labels_contract() -> None: assert enriched.title_source == TitleSource.HEURISTIC -def _emitter_for_repo_identity_tests() -> _SessionEmitter: - ctx = _ParseContext( - provider_hint=Provider.CODEX, - should_group=False, - source_path_str="/tmp/session.jsonl", - fallback_id="session", - file_mtime="2026-03-11T00:00:00+00:00", - capture_raw=False, - sidecar_data={}, - ) - return _SessionEmitter(ctx) - - -def test_repo_identity_evidence_grades_directory_only_with_cwd_but_no_git() -> None: - """polylogue-cijx.2: a session with a cwd but no git evidence at all is - honestly a DIRECTORY, not a repository -- the emitter must say so rather - than let a downstream reader assume ``working_directories`` alone proves - a repository (cijx.4 decision 1).""" - session = ParsedSession( - source_name=Provider.CHATGPT, - provider_session_id="session-cwd-only", - title="t", - created_at="2026-01-01T00:00:00Z", - messages=[_parsed_message("m1", role="user", text="hello")], - working_directories=["/home/sinity"], - ) - - enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CHATGPT) - - events = [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"] - assert len(events) == 1 - assert events[0].payload == { - "grade": "directory_only", - "root_paths": ["/home/sinity"], - "git_repository_url": None, - "git_branch": None, - "git_commit_hash": None, - } - - -def test_repo_identity_evidence_grades_git_evidence_when_branch_or_commit_present() -> None: - """A session carrying real git evidence (branch/url/commit) grades as - ``git_evidence``, distinct from the cwd-only case.""" - session = ParsedSession( - source_name=Provider.CODEX, - provider_session_id="session-git", - title="t", - created_at="2026-01-01T00:00:00Z", - messages=[_parsed_message("m1", role="user", text="hello")], - working_directories=["/realm/project/polylogue"], - git_branch="master", - git_commit_hash="abc123", - ) - - enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CODEX) - - events = [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"] - assert len(events) == 1 - assert events[0].payload["grade"] == "git_evidence" - assert events[0].payload["root_paths"] == ["/realm/project/polylogue"] - assert events[0].payload["git_branch"] == "master" - assert events[0].payload["git_commit_hash"] == "abc123" - - -def test_repo_identity_evidence_omitted_when_no_location_evidence_at_all() -> None: - """A session with neither a cwd nor git evidence gets no repo_identity_evidence - event -- there is nothing to grade, and emitting an empty-payload event on - every such session would just be noise.""" - session = ParsedSession( - source_name=Provider.CHATGPT, - provider_session_id="session-bare", - title="t", - created_at="2026-01-01T00:00:00Z", - messages=[_parsed_message("m1", role="user", text="hello")], - ) - - enriched = _emitter_for_repo_identity_tests()._maybe_enrich(session, Provider.CHATGPT) - - assert not [event for event in enriched.session_events if event.event_type == "repo_identity_evidence"] - - def _zip_entry(name: str, *, size: int = 100, compressed: int = 50) -> zipfile.ZipInfo: entry = zipfile.ZipInfo(name) entry.file_size = size diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index 2ef8198d81..7e62c47f7c 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -232,6 +232,50 @@ def test_parser_receipt_fails_when_observed_identity_differs_from_binding(tmp_pa assert parser_census_logical_keys(receipt[1]) == ("codex-session:parser-observed-id",) +def test_terminal_non_session_failure_has_complete_empty_parser_census(tmp_path: Path) -> None: + """A typed terminal failure is a settled non-session source disposition.""" + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b"not valid codex jsonl", + source_path="terminal-corrupt.jsonl", + acquired_at_ms=1, + ) + archive.record_raw_failure_evidence( + raw_id, + provider=Provider.CODEX, + source_path="terminal-corrupt.jsonl", + source_index=0, + acquired_at_ms=1, + kind=RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT, + ) + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CODEX, + error=ValueError("terminal corrupt input"), + preserve_existing_failure_evidence=True, + ) + with archive._ensure_source_conn(): + archive_revision_governance.record_current_parser_source_census( + archive._ensure_source_conn(), + raw_id, + ) + + with sqlite3.connect(tmp_path / "source.db") as conn: + status, keys = conn.execute( + "SELECT status, logical_keys_json FROM raw_authority_parser_census WHERE raw_id = ?", + (raw_id,), + ).fetchone() + assert status == "complete" + assert parser_census_logical_keys(keys) == () + + from polylogue.sources.revision_backfill import require_current_parser_source_census + + assert require_current_parser_source_census(tmp_path)[raw_id] == () + + def test_membership_receipt_excludes_post_parse_pending_identity(tmp_path: Path) -> None: """A parser-derived membership receipt cannot retain its provisional raw key.""" initialize_active_archive_root(tmp_path) From 8b9a0af7a6c3bea59573cf6622a0de3217e2bcf8 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 18:59:57 +0200 Subject: [PATCH 5/9] fix: tighten durable verification authority --- CLAUDE.md | 11 ++- devtools/verify.py | 28 ++++--- polylogue/daemon_client.py | 59 ++++++++----- polylogue/maintenance/archive_verification.py | 51 ++++++----- tests/unit/cli/test_daemon_client.py | 74 ++++++++++++++-- tests/unit/devtools/test_run_tests.py | 19 +++-- tests/unit/devtools/test_verify.py | 47 +++++++---- .../maintenance/test_archive_verification.py | 84 +++++++++++++++++++ tests/unit/test_pytest_temp_policy.py | 1 + 9 files changed, 290 insertions(+), 84 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 949e1089e8..43b0bb8cfb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -509,10 +509,13 @@ isolated XDG paths + archive root. exceptions opt out inline via `@pytest.mark.uses_real_clock("reason")`. - Pytest temp DBs pick ONE basetemp root via `devtools.verify_runs.resolve_pytest_basetemp_root` (shared by - `tests/conftest.py` and the `devtools test`/`verify` preflight): focused runs - use bounded `/dev/shm` tmpfs when it has ≥1 GiB free, while full-suite and - seed-testmon runs default to `/realm/tmp/polylogue-pytest` (NVMe) because - their aggregate fixture tree can exceed the supervised tmpfs ceiling. + `tests/conftest.py` and the `devtools test`/`verify` preflight). Bare pytest + without a managed run identity or an explicit basetemp root is forced to + `/realm/tmp/polylogue-pytest` (NVMe). Managed `devtools test` and + `devtools verify` runs may use bounded `/dev/shm` tmpfs only after the + runtime policy admits the requested demand; full-suite and seed-testmon + runs default to NVMe because their aggregate fixture tree can exceed the + supervised tmpfs ceiling. `POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB` overrides required headroom; an explicit `POLYLOGUE_PYTEST_TMPFS=1` requests bounded tmpfs, but the request is honored only when the effective budget satisfies the declared basetemp diff --git a/devtools/verify.py b/devtools/verify.py index b5ccff19df..ed3fb58107 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -1985,10 +1985,10 @@ def build_verify_steps( pytest_cmd.extend(["-m", base_marker, "--testmon"]) if resume_testmon_seed: pytest_cmd.append("--testmon-forceselect") - label = "pytest seed-testmon (resume)" + label = BROAD_PYTEST_STEP_LABELS["seed_resume"] else: pytest_cmd.append("--testmon-noselect") - label = "pytest seed-testmon" + label = BROAD_PYTEST_STEP_LABELS["seed"] pytest_cmd.extend(_pytest_worker_args(maximum=4)) steps.append((label, pytest_cmd)) elif full_pytest: @@ -2006,7 +2006,7 @@ def build_verify_steps( f"({base_marker}) and not load_sensitive and not tui", *_pytest_worker_args(), ] - steps.append(("pytest full (parallel)", bulk_cmd)) + steps.append((BROAD_PYTEST_STEP_LABELS["full_parallel"], bulk_cmd)) def _isolated_report_arg(arg: str) -> str: # Keep the bulk lane's canonical report artifacts intact for @@ -2019,11 +2019,11 @@ def _isolated_report_arg(arg: str) -> str: isolated_cmd = [_isolated_report_arg(arg) for arg in pytest_cmd] isolated_cmd.extend(["-m", f"({base_marker}) and (load_sensitive or tui)", "-p", "no:randomly", "-n", "0"]) - steps.append(("pytest load-sensitive (isolated)", isolated_cmd)) + steps.append((BROAD_PYTEST_STEP_LABELS["load_sensitive"], isolated_cmd)) else: pytest_cmd.extend(["-m", base_marker, "--testmon", *_pytest_worker_args()]) pytest_cmd.append("--testmon-forceselect") - label = "pytest testmon (broad)" if broad_testmon else "pytest testmon" + label = BROAD_PYTEST_STEP_LABELS["testmon_broad"] if broad_testmon else "pytest testmon" steps.append((label, pytest_cmd)) if lab: @@ -2152,6 +2152,15 @@ def _pytest_worker_args(*, maximum: int | None = None) -> list[str]: return ["-n", str(workers)] +BROAD_PYTEST_STEP_LABELS = { + "seed": "pytest seed-testmon", + "seed_resume": "pytest seed-testmon (resume)", + "full_parallel": "pytest full (parallel)", + "load_sensitive": "pytest load-sensitive (isolated)", + "testmon_broad": "pytest testmon (broad)", +} + + def _pytest_command_worker_request(cmd: Sequence[str]) -> str | None: """Return the last xdist worker request from a final pytest command. @@ -2198,14 +2207,7 @@ def _pytest_command_concurrency(cmd: Sequence[str], *, env: Mapping[str, str] | def _pytest_uses_full_suite_basetemp(label: str) -> bool: """Whether this pytest step can materialize the measured full-suite tree.""" - return label.startswith( - ( - "pytest seed-testmon", - "pytest full", - "pytest load-sensitive", - "pytest testmon (broad)", - ) - ) + return label in BROAD_PYTEST_STEP_LABELS.values() _BROAD_TESTMON_CHANGED_PATHS = { diff --git a/polylogue/daemon_client.py b/polylogue/daemon_client.py index fe82a354e4..58cf9136fc 100644 --- a/polylogue/daemon_client.py +++ b/polylogue/daemon_client.py @@ -49,6 +49,27 @@ def request_json( raise_for_status: bool = False, accepted_statuses: frozenset[int] = frozenset({200}), ) -> dict[str, Any] | None: + response = self._request_json_response(method, path, body) + if response is None: + return None + status, payload = response + if status not in accepted_statuses: + if raise_for_status: + envelope = payload if isinstance(payload, dict) else {} + code = envelope.get("error") + detail = envelope.get("detail") + raise DaemonResponseError( + status=status, + code=code if isinstance(code, str) else None, + detail=detail if isinstance(detail, str) else None, + ) + return None + return payload + + def _request_json_response( + self, method: str, path: str, body: dict[str, object] | None = None + ) -> tuple[int, dict[str, Any] | None] | None: + """Return the response status with its decoded JSON object, if any.""" if not self.socket_path.exists(): return None connection = _UnixHTTPConnection(self.socket_path, self.timeout_s) @@ -60,20 +81,9 @@ def request_json( headers["Authorization"] = f"Bearer {self.auth_token}" connection.request(method, path, body=raw, headers=headers) response = connection.getresponse() - payload = json.loads(response.read().decode()) + decoded = json.loads(response.read().decode()) self.last_elapsed_ms = round((perf_counter() - started_at) * 1000) - if response.status not in accepted_statuses: - if raise_for_status: - envelope = payload if isinstance(payload, dict) else {} - code = envelope.get("error") - detail = envelope.get("detail") - raise DaemonResponseError( - status=response.status, - code=code if isinstance(code, str) else None, - detail=detail if isinstance(detail, str) else None, - ) - return None - return payload if isinstance(payload, dict) else None + return response.status, decoded if isinstance(decoded, dict) else None except (OSError, TimeoutError, ValueError, http.client.HTTPException): return None finally: @@ -92,16 +102,21 @@ def probe( ) -> dict[str, Any] | None: """Return identity only for the daemon serving the requested archive. - Maintenance callers may accept the health endpoint's 503 envelope in - order to reach the daemon-owned repair route. This does not authorize - the repair: the write endpoint still runs its typed preflight. Query - callers retain the strict 200-only default. + Maintenance callers may accept only the health endpoint's typed + ``degraded`` lifecycle 503 envelope in order to reach the daemon-owned + repair route. This does not authorize the repair: the write endpoint + still runs its typed preflight. Query callers retain the strict + 200-only default. """ - health = ( - self.request_json("GET", "/api/health", accepted_statuses=frozenset({200, 503})) - if accept_degraded - else self.request_json("GET", "/api/health") - ) + response = self._request_json_response("GET", "/api/health") + if response is None: + return None + status, health = response + if status == 503: + if not accept_degraded or health is None or health.get("raw_failure_lifecycle_state") != "degraded": + return None + elif status != 200: + return None if health is None: return None if health.get("archive_root") != archive_root: diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index ba6cf4e05a..de5dceeb78 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -478,6 +478,33 @@ def _check_pointer_coherence(archive_root: Path, _sample_limit: int) -> ArchiveV # --------------------------------------------------------------------------- +def _valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: + """Return the receipt predicate shared by source/index coverage checks. + + A supersession receipt is authority only when it still names the same + bytes and an index materialization of the recorded duplicate twin. Keep + the predicate in one place so backlog freshness cannot classify a receipt + differently from source-index coverage. + """ + if not table_exists(conn, "raw_byte_duplicate_supersession_receipts"): + return "0" + return f""" + EXISTS( + SELECT 1 + FROM raw_byte_duplicate_supersession_receipts receipt + JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id + JOIN idx_tier.sessions twin_session + ON twin_session.raw_id = twin.raw_id + AND twin_session.session_id = receipt.duplicate_of_session_id + WHERE receipt.raw_id = {raw_alias}.raw_id + AND receipt.blob_hash = {raw_alias}.blob_hash + AND receipt.blob_size = {raw_alias}.blob_size + AND twin.blob_hash = {raw_alias}.blob_hash + AND twin.blob_size = {raw_alias}.blob_size + ) + """ + + def _check_source_index_coverage(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: return _check_source_index_coverage_at_index_path(archive_root, _resolve_index_path(archive_root), sample_limit) @@ -525,26 +552,7 @@ def _check_source_index_coverage_at_index_path( census_expr = ( "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" ) - has_supersession_receipts = table_exists(conn, "raw_byte_duplicate_supersession_receipts") - valid_supersession_expr = ( - """ - EXISTS( - SELECT 1 - FROM raw_byte_duplicate_supersession_receipts receipt - JOIN raw_sessions twin ON twin.raw_id = receipt.duplicate_of_raw_id - JOIN idx_tier.sessions twin_session - ON twin_session.raw_id = twin.raw_id - AND twin_session.session_id = receipt.duplicate_of_session_id - WHERE receipt.raw_id = r.raw_id - AND receipt.blob_hash = r.blob_hash - AND receipt.blob_size = r.blob_size - AND twin.blob_hash = r.blob_hash - AND twin.blob_size = r.blob_size - ) - """ - if has_supersession_receipts - else "0" - ) + valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") # A read-only connection (``query_only=ON``, connection-wide, not # per-attached-db) cannot ``CREATE TEMP VIEW`` -- the temp schema @@ -2394,6 +2402,7 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: """ has_census = table_exists(conn, "raw_membership_census") census_expr = "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" + valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") row = conn.execute( f""" WITH heads AS ( @@ -2402,6 +2411,7 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: r.parse_error, r.logical_source_key, {census_expr} AS census_status, + {valid_supersession_expr} AS valid_supersession, MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) OVER ( PARTITION BY r.origin, @@ -2417,6 +2427,7 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: SELECT SUM( CASE WHEN rn = 1 AND any_indexed = 0 AND parse_error IS NULL AND COALESCE(census_status, '') NOT IN ('non_session', 'failed') + AND valid_supersession = 0 THEN 1 ELSE 0 END ) FROM heads WHERE rn = 1 diff --git a/tests/unit/cli/test_daemon_client.py b/tests/unit/cli/test_daemon_client.py index df21faae04..36a25e1f8e 100644 --- a/tests/unit/cli/test_daemon_client.py +++ b/tests/unit/cli/test_daemon_client.py @@ -1,14 +1,27 @@ from __future__ import annotations +import shutil import subprocess import sys +import tempfile import threading +from collections.abc import Iterator from os import getpid from pathlib import Path import pytest +@pytest.fixture +def _short_uds_runtime_dir() -> Iterator[Path]: + """Keep UDS route tests under the operating system socket-path limit.""" + runtime_dir = Path(tempfile.mkdtemp(prefix="plg-client-uds-")) + try: + yield runtime_dir + finally: + shutil.rmtree(runtime_dir, ignore_errors=True) + + def test_daemon_client_import_does_not_load_storage() -> None: result = subprocess.run( [ @@ -70,7 +83,9 @@ def test_daemon_probe_rejects_the_tmp_archive_config_trap(monkeypatch: pytest.Mo ) -def test_daemon_client_probes_the_production_uds_server(monkeypatch: pytest.MonkeyPatch) -> None: +def test_daemon_client_probes_the_production_uds_server( + monkeypatch: pytest.MonkeyPatch, _short_uds_runtime_dir: Path +) -> None: """The stdlib client reaches the production AF_UNIX server, not a TCP substitute.""" from http import HTTPStatus @@ -92,7 +107,7 @@ def health(self: DaemonAPIHandler) -> None: ) monkeypatch.setattr(DaemonAPIHandler, "_handle_health", health) - socket_path = Path("/realm/tmp") / f"polylogue-uds-{getpid()}.sock" + socket_path = _short_uds_runtime_dir / f"daemon-{getpid()}.sock" server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) server.auth_token = "uds-test-token" thread = threading.Thread(target=server.serve_forever, daemon=True) @@ -115,6 +130,7 @@ def health(self: DaemonAPIHandler) -> None: def test_daemon_client_can_probe_matching_writer_through_degraded_health( monkeypatch: pytest.MonkeyPatch, + _short_uds_runtime_dir: Path, ) -> None: """Maintenance discovers the writer without weakening query readiness.""" from http import HTTPStatus @@ -135,7 +151,7 @@ def degraded_health(self: DaemonAPIHandler) -> None: ) monkeypatch.setattr(DaemonAPIHandler, "_handle_health", degraded_health) - socket_path = Path("/realm/tmp") / f"polylogue-uds-degraded-{getpid()}.sock" + socket_path = _short_uds_runtime_dir / f"degraded-{getpid()}.sock" server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() @@ -164,7 +180,55 @@ def degraded_health(self: DaemonAPIHandler) -> None: thread.join(timeout=2) -def test_daemon_client_preserves_typed_4xx_detail_from_the_production_uds_server() -> None: +def test_daemon_client_rejects_unrelated_503_when_degraded_probe_is_enabled( + monkeypatch: pytest.MonkeyPatch, + _short_uds_runtime_dir: Path, +) -> None: + """The production probe route accepts a maintenance ``degraded`` 503, + not any matching identity payload. Mutating that lifecycle state to + ``blocked`` must keep the writer unavailable to the caller.""" + from http import HTTPStatus + + from polylogue.cli.daemon_client import DaemonClient + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer + + def blocked_health(self: DaemonAPIHandler) -> None: + self._send_json( + HTTPStatus.SERVICE_UNAVAILABLE, + { + "archive_root": "/realm/archive", + "index_schema_version": 24, + "daemon_version": "0.1.0", + "raw_failure_lifecycle_state": "blocked", + }, + ) + + monkeypatch.setattr(DaemonAPIHandler, "_handle_health", blocked_health) + socket_path = _short_uds_runtime_dir / f"blocked-{getpid()}.sock" + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + client = DaemonClient(socket_path) + assert ( + client.probe( + archive_root="/realm/archive", + index_schema_version=24, + daemon_version="0.1.0", + accept_degraded=True, + ) + is None + ) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_daemon_client_preserves_typed_4xx_detail_from_the_production_uds_server( + _short_uds_runtime_dir: Path, +) -> None: """Maintenance clients can surface a daemon validation reason, not only a transport failure.""" from http import HTTPStatus @@ -180,7 +244,7 @@ def _handle_consume_canary_report(self) -> None: "receipt is missing the canonical acceptance profile", ) - socket_path = Path("/realm/tmp") / f"polylogue-uds-canary-4xx-{getpid()}.sock" + socket_path = _short_uds_runtime_dir / f"canary-4xx-{getpid()}.sock" server = DaemonAPIUnixHTTPServer(socket_path, InvalidCanaryReportHandler) server.auth_token = "uds-test-token" thread = threading.Thread(target=server.serve_forever, daemon=True) diff --git a/tests/unit/devtools/test_run_tests.py b/tests/unit/devtools/test_run_tests.py index bef2f53356..0d126bd449 100644 --- a/tests/unit/devtools/test_run_tests.py +++ b/tests/unit/devtools/test_run_tests.py @@ -34,19 +34,26 @@ def test_build_pytest_cmd_respects_explicit_worker_flag() -> None: @pytest.mark.parametrize( - "selection", + ("selection", "expected_request"), [ - ["tests/unit", "-n4"], - ["tests/unit", "-n=4"], - ["tests/unit", "--numprocesses", "8"], - ["tests/unit", "--numprocesses=8"], + (["tests/unit", "-n4"], "4"), + (["tests/unit", "-n=4"], "4"), + (["tests/unit", "--numprocesses", "8"], "8"), + (["tests/unit", "--numprocesses=8"], "8"), ], ) -def test_build_pytest_cmd_forwards_all_xdist_worker_spellings(selection: list[str]) -> None: +def test_build_pytest_cmd_forwards_exactly_one_xdist_worker_request( + selection: list[str], expected_request: str +) -> None: command = run_tests.build_pytest_cmd(selection) for arg in selection: assert arg in command + worker_flags = [ + arg for arg in command if arg in {"-n", "--numprocesses"} or arg.startswith(("-n", "--numprocesses=")) + ] + assert len(worker_flags) == 1 + assert verify._pytest_command_worker_request(command) == expected_request def test_build_pytest_cmd_honors_workers_env(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 094fe0db9d..5fadb1297f 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -321,8 +321,17 @@ def test_seed_testmon_worker_count_can_be_overridden(monkeypatch: pytest.MonkeyP assert command[command.index("-n") + 1] == "4" -def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("POLYLOGUE_PYTEST_TMPFS", raising=False) +def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=8192) + for name in ( + "POLYLOGUE_PYTEST_BASETEMP_ROOT", + "POLYLOGUE_PYTEST_TMPFS", + "POLYLOGUE_PYTEST_TMPFS_MAX_MB", + "POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB", + "POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB", + ): + monkeypatch.delenv(name, raising=False) completed = subprocess.CompletedProcess(args=["pytest"], returncode=0, stdout="1 passed in 0.1s\n", stderr="") with ( @@ -334,9 +343,7 @@ def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> No assert rc == 0 assert metadata["pytest_tmpfs"] is False assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_TMPFS"] == "0" - assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str( - verify_runs.DEFAULT_PYTEST_BASETEMP_ROOT - ) + assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT"] == "50000" @@ -1542,13 +1549,11 @@ def counted_size(_path: Path) -> int: assert calls == 1 -def test_pytest_basetemp_path_tracks_tmpfs_opt_in(tmp_path: Path) -> None: +def test_pytest_basetemp_path_tracks_tmpfs_opt_in(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, _scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) path = pytest_basetemp_path(root=tmp_path, run_id="run-1", env={"POLYLOGUE_PYTEST_TMPFS": "1"}) - if Path("/dev/shm").is_dir(): - assert path.parent == Path("/dev/shm") - else: - assert path.parent == Path("/realm/tmp/polylogue-pytest") + assert path.parent == shm def test_pytest_tmpfs_budget_is_shared_and_bounded() -> None: @@ -1703,7 +1708,10 @@ def test_adaptive_pytest_policy_treats_full_run_basetemp_as_aggregate_demand() - (["--numprocesses=auto"], max(1, os.cpu_count() or 1)), ], ) -def test_production_pytest_commands_reserve_every_xdist_spelling(worker_args: list[str], expected: int) -> None: +def test_production_pytest_commands_reserve_every_xdist_spelling( + monkeypatch: pytest.MonkeyPatch, worker_args: list[str], expected: int +) -> None: + monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False) command = run_tests.build_pytest_cmd(["tests/unit/devtools", *worker_args]) assert verify._pytest_command_concurrency(command) == expected @@ -2319,7 +2327,14 @@ def test_explicit_basetemp_root_retains_managed_resource_monitoring( ) -> None: monkeypatch.setattr(verify_runs, "PYTEST_TMPFS_ROOT", tmp_path / "unselected-tmpfs") nvme_root = tmp_path / "realm-tmp" / "polylogue-pytest" + nvme_root.mkdir(parents=True) + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 16 * 1024 * 1024}) monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(nvme_root)) + monkeypatch.delenv("POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB", raising=False) + monkeypatch.delenv("POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB", raising=False) run = VerifyRun(tier="configured-nvme", argv=[], git_head=None, root=tmp_path) rc, _elapsed, metadata = _run( @@ -2398,11 +2413,15 @@ def test_bench_slo_forces_nested_pytest_to_managed_scratch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + inherited_tmpfs_root = shm / "inherited-benchmark" + inherited_tmpfs_root.mkdir() + scratch.mkdir() run = VerifyRun(tier="lab", argv=[], git_head=None, root=tmp_path) - monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", "/dev/shm/inherited-benchmark") + monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(inherited_tmpfs_root)) managed_env = { "POLYLOGUE_PYTEST_TMPFS": "0", - "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/realm/tmp/polylogue-pytest", + "POLYLOGUE_PYTEST_BASETEMP_ROOT": str(scratch), } completed = subprocess.CompletedProcess(args=["devtools", "bench", "slo"], returncode=0, stdout="", stderr="") @@ -2421,7 +2440,7 @@ def test_bench_slo_forces_nested_pytest_to_managed_scratch( assert env["POLYLOGUE_VERIFY_RUN_ID"] == run.run_id assert env["POLYLOGUE_PYTEST_RUN_ID"] == run.run_id assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" - assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == "/realm/tmp/polylogue-pytest" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) def test_run_forces_subprocesses_to_current_checkout(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index 1a6009330b..52080bd294 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -1464,6 +1464,90 @@ def test_convergence_freshness_passes_with_no_gap(tmp_path: Path) -> None: assert check.evidence["unindexed_backlog_gap"] == 0 +def test_convergence_freshness_excludes_a_receipt_backed_duplicate(tmp_path: Path) -> None: + """The production convergence-freshness route excludes a byte-identical + unindexed duplicate only when its supersession receipt names an indexed + twin. Removing the receipt must turn this route back into a backlog.""" + _seed_coherent_archive(tmp_path) + conn = _connect(tmp_path / "source.db") + try: + conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) + SELECT 'raw-superseded-backlog', origin, 'duplicate-backlog', '/duplicate-backlog', + blob_hash, blob_size, 200, 'byte_proven' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + conn.execute( + """ + INSERT INTO raw_byte_duplicate_supersession_receipts( + raw_id, blob_hash, blob_size, duplicate_of_raw_id, + duplicate_of_session_id, previous_revision_authority, + promoted_at_ms, tool_version, backup_manifest_path, detail + ) + SELECT 'raw-superseded-backlog', blob_hash, blob_size, 'raw-1', + 'codex-session:session', 'quarantined', 200, + 'test', '/verified/manifest.json', '' + FROM raw_sessions WHERE raw_id = 'raw-superseded-backlog' + """ + ) + conn.commit() + finally: + conn.close() + + check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness") + + assert check.status is OutcomeStatus.OK + assert check.evidence["unindexed_backlog_gap"] == 0 + + +def test_convergence_freshness_counts_a_receipt_with_the_wrong_twin_bytes(tmp_path: Path) -> None: + """The production convergence-freshness route keeps a duplicate in the + backlog when mutating the receipt's indexed-twin bytes invalidates its + supersession evidence.""" + _seed_coherent_archive(tmp_path) + conn = _connect(tmp_path / "source.db") + try: + conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) VALUES ( + 'raw-invalid-supersession-backlog', 'codex-session', 'invalid-duplicate', + '/invalid-duplicate', ?, 10, 200, 'byte_proven' + ) + """, + (b"z" * 32,), + ) + conn.execute( + """ + INSERT INTO raw_byte_duplicate_supersession_receipts( + raw_id, blob_hash, blob_size, duplicate_of_raw_id, + duplicate_of_session_id, previous_revision_authority, + promoted_at_ms, tool_version, backup_manifest_path, detail + ) VALUES ( + 'raw-invalid-supersession-backlog', ?, 10, 'raw-1', + 'codex-session:session', 'quarantined', 200, + 'test', '/verified/manifest.json', '' + ) + """, + (b"z" * 32,), + ) + conn.commit() + finally: + conn.close() + + check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness") + + assert check.status is OutcomeStatus.ERROR + assert check.evidence["unindexed_backlog_gap"] == 1 + + def test_dangling_assertion_target_trips_user_tier_refs(tmp_path: Path) -> None: """RED TWIN (I10): a user-tier assertion whose target session/message no longer resolves in index.db is a dangling reference -- silently diff --git a/tests/unit/test_pytest_temp_policy.py b/tests/unit/test_pytest_temp_policy.py index 94bdb03eab..1a8620bd07 100644 --- a/tests/unit/test_pytest_temp_policy.py +++ b/tests/unit/test_pytest_temp_policy.py @@ -70,6 +70,7 @@ def test_managed_pytest_temp_root_honors_explicit_root( tmp_path: Path, ) -> None: configured = tmp_path / "configured" + configured.mkdir() monkeypatch.setattr(verify_runs, "_fs_usage", lambda path: {"used_kb": 0, "free_kb": 32 * 1024 * 1024}) monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(configured)) monkeypatch.setenv("POLYLOGUE_PYTEST_TMPFS", "1") From 124989c4dd50e4eee01298492b30c84825369ffa Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 11 Aug 2026 19:42:09 +0200 Subject: [PATCH 6/9] fix(reindex): align durable verification cohorts Problem: convergence freshness and source/index coverage could classify durable revision evidence differently. A retired raw lost its raw-level cohort key, and an unsafe configured tmpfs root remained selected after policy disabled tmpfs.\n\nWhat changed: share the durable logical-cohort and content-bound supersession rules across both verification paths, preserve the sole retained membership identity, reroute unsafe configured tmpfs roots to scratch, and pin the receipt-removal and shared-raw counterexamples.\n\nCompatibility: existing multi-membership raw acquisitions retain their native/path fallback because no single membership identity is authoritative. --- devtools/verify_runs.py | 5 ++ polylogue/maintenance/archive_verification.py | 32 ++++++- tests/unit/devtools/test_verify.py | 22 +++++ .../maintenance/test_archive_verification.py | 89 +++++++++++++++++++ 4 files changed, 144 insertions(+), 4 deletions(-) diff --git a/devtools/verify_runs.py b/devtools/verify_runs.py index 1e1b4655f0..95949457ac 100644 --- a/devtools/verify_runs.py +++ b/devtools/verify_runs.py @@ -792,6 +792,11 @@ def apply_managed_pytest_runtime_policy( and effective_tmpfs_budget_kb < required_basetemp_kb ): normalized["POLYLOGUE_PYTEST_TMPFS"] = "0" + if configured_tmpfs: + # The configured tmpfs root has become unsafe for this run. + # Leaving it in place would make the resolver select it even + # though tmpfs has just been disabled, without its cap. + normalized.pop("POLYLOGUE_PYTEST_BASETEMP_ROOT", None) if default_full_suite_scratch: # Broad-suite demand grows with the fixture universe and has exceeded # the supervised 2 GiB ceiling while tests were still progressing. diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index de5dceeb78..d6e8926a48 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -505,6 +505,28 @@ def _valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_ali """ +def _logical_head_cohort_expr(conn: sqlite3.Connection, *, raw_alias: str) -> str: + """Return the durable identity used to group raw revisions into one head. + + A full-revision row retired into membership governance intentionally loses + its raw-level ``logical_source_key``. Its single retained membership key + remains the authoritative identity, so use it before the legacy + native-id/path fallback. Shared raws can hold several membership keys; + they have no one raw-level cohort and must keep that fallback instead of + being arbitrarily assigned to one member. + """ + membership_key = "NULL" + if table_exists(conn, "raw_session_memberships"): + membership_key = f""" + ( + SELECT CASE WHEN COUNT(*) = 1 THEN MIN(m.logical_source_key) END + FROM raw_session_memberships AS m + WHERE m.raw_id = {raw_alias}.raw_id + ) + """ + return f"COALESCE({raw_alias}.logical_source_key, {membership_key}, {raw_alias}.native_id, {raw_alias}.source_path)" + + def _check_source_index_coverage(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: return _check_source_index_coverage_at_index_path(archive_root, _resolve_index_path(archive_root), sample_limit) @@ -553,6 +575,7 @@ def _check_source_index_coverage_at_index_path( "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" ) valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") + logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r") # A read-only connection (``query_only=ON``, connection-wide, not # per-attached-db) cannot ``CREATE TEMP VIEW`` -- the temp schema @@ -571,11 +594,11 @@ def _check_source_index_coverage_at_index_path( MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) OVER ( PARTITION BY r.origin, - COALESCE(r.logical_source_key, r.native_id, r.source_path) + {logical_cohort_expr} ) AS any_indexed, ROW_NUMBER() OVER ( PARTITION BY r.origin, - COALESCE(r.logical_source_key, r.native_id, r.source_path) + {logical_cohort_expr} ORDER BY r.acquired_at_ms DESC, r.raw_id DESC ) AS rn FROM raw_sessions r @@ -2403,6 +2426,7 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: has_census = table_exists(conn, "raw_membership_census") census_expr = "(SELECT c.status FROM raw_membership_census c WHERE c.raw_id = r.raw_id)" if has_census else "NULL" valid_supersession_expr = _valid_byte_duplicate_supersession_expr(conn, raw_alias="r") + logical_cohort_expr = _logical_head_cohort_expr(conn, raw_alias="r") row = conn.execute( f""" WITH heads AS ( @@ -2415,11 +2439,11 @@ def _unindexed_backlog_gap(conn: sqlite3.Connection) -> int: MAX(EXISTS(SELECT 1 FROM idx_tier.sessions s WHERE s.raw_id = r.raw_id)) OVER ( PARTITION BY r.origin, - COALESCE(r.logical_source_key, r.native_id, r.source_path) + {logical_cohort_expr} ) AS any_indexed, ROW_NUMBER() OVER ( PARTITION BY r.origin, - COALESCE(r.logical_source_key, r.native_id, r.source_path) + {logical_cohort_expr} ORDER BY r.acquired_at_ms DESC, r.raw_id DESC ) AS rn FROM raw_sessions r diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 5fadb1297f..ee24331be8 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -1959,6 +1959,28 @@ def test_inherited_512_mib_tmpfs_cap_reroutes_measured_demand_to_scratch( assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) +def test_explicit_tmpfs_root_reroutes_to_scratch_when_its_cap_is_too_small( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=15_190) + explicit_root = shm / "explicit" + explicit_root.mkdir() + + env, policy = apply_managed_pytest_runtime_policy( + { + "POLYLOGUE_PYTEST_BASETEMP_ROOT": str(explicit_root), + "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "512", + }, + worker_count=4, + ) + + assert policy is not None + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + def test_focused_policy_keeps_full_suite_basetemp_demand_out_of_scratch_preflight( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index 52080bd294..28683a3ef8 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -568,6 +568,86 @@ def test_source_index_coverage_groups_reacquisitions_by_logical_source_key(tmp_p assert check.evidence["unindexed_head_count"] == 0 +def test_coverage_groups_retired_membership_identity_with_bound_revision(tmp_path: Path) -> None: + """A retired raw keeps its sole membership identity after its raw key is nulled.""" + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'") + source_conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) + SELECT 'raw-retired-membership', origin, 'session', '/retired-membership', + blob_hash, blob_size, 200, 'quarantined' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + source_conn.execute( + """ + INSERT INTO raw_session_memberships( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count + ) VALUES ('raw-retired-membership', 'codex:session', 'session', 'retired', ?, 1) + """, + (b"r" * 32,), + ) + source_conn.commit() + finally: + source_conn.close() + + report = verify_archive(tmp_path, checks=("source-index-coverage", "convergence-freshness")) + + coverage = _check(report, "source-index-coverage") + freshness = _check(report, "convergence-freshness") + assert coverage.status is OutcomeStatus.OK + assert coverage.evidence["logical_head_count"] == 1 + assert freshness.status is OutcomeStatus.OK + assert freshness.evidence["unindexed_backlog_gap"] == 0 + + +def test_coverage_does_not_assign_a_shared_raw_to_one_membership_key(tmp_path: Path) -> None: + """A raw with several retained identities must not inherit an arbitrary cohort.""" + _seed_coherent_archive(tmp_path) + source_conn = _connect(tmp_path / "source.db") + try: + source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'") + source_conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) + SELECT 'raw-shared-membership', origin, NULL, '/shared-membership', + blob_hash, blob_size, 200, 'quarantined' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + source_conn.executemany( + """ + INSERT INTO raw_session_memberships( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count + ) VALUES ('raw-shared-membership', ?, ?, 'shared', ?, 1) + """, + [ + ("codex:session", "session", b"s" * 32), + ("codex:other", "other", b"o" * 32), + ], + ) + source_conn.commit() + finally: + source_conn.close() + + check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage") + + assert check.status is OutcomeStatus.WARNING + assert check.evidence["logical_head_count"] == 2 + assert check.evidence["quarantined_count"] == 1 + + def test_quarantined_head_with_no_session_is_warning_not_error(tmp_path: Path) -> None: """The polylogue-in24n bug class in reverse: a quarantined raw (the default ``revision_authority``) that never materialized is a *typed* @@ -1504,6 +1584,15 @@ def test_convergence_freshness_excludes_a_receipt_backed_duplicate(tmp_path: Pat assert check.status is OutcomeStatus.OK assert check.evidence["unindexed_backlog_gap"] == 0 + with _connect(tmp_path / "source.db") as conn: + conn.execute("DELETE FROM raw_byte_duplicate_supersession_receipts WHERE raw_id = 'raw-superseded-backlog'") + conn.commit() + + check = _check(verify_archive(tmp_path, checks=("convergence-freshness",)), "convergence-freshness") + + assert check.status is OutcomeStatus.ERROR + assert check.evidence["unindexed_backlog_gap"] == 1 + def test_convergence_freshness_counts_a_receipt_with_the_wrong_twin_bytes(tmp_path: Path) -> None: """The production convergence-freshness route keeps a duplicate in the From bd476f74a13ac65faae651b98a9c72097ea32cb9 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 22:37:39 +0200 Subject: [PATCH 7/9] fix(authority): canonicalize durable source cohorts Bind byte-duplicate supersession to the source coordinate that can affect replay, and share canonical provider/origin cohort identity across verification and schema sampling. --- polylogue/archive/revision_authority.py | 36 +++++++++++++ polylogue/maintenance/archive_verification.py | 19 ++++--- polylogue/schemas/sampling_db.py | 53 +++++++++++-------- .../raw_byte_duplicate_supersession.py | 24 ++++++--- tests/unit/core/test_schema_generation.py | 36 +++++++++++++ .../maintenance/test_archive_verification.py | 51 +++++++++++++++--- ...t_raw_byte_duplicate_supersession_apply.py | 4 +- 7 files changed, 175 insertions(+), 48 deletions(-) diff --git a/polylogue/archive/revision_authority.py b/polylogue/archive/revision_authority.py index d4a125ebb1..ecd515fc6f 100644 --- a/polylogue/archive/revision_authority.py +++ b/polylogue/archive/revision_authority.py @@ -2,6 +2,7 @@ from __future__ import annotations +import sqlite3 from collections.abc import Callable, Iterable from dataclasses import dataclass from enum import StrEnum @@ -60,6 +61,41 @@ def canonical_authority_logical_key(logical_key: str) -> str: return f"{origin.value}:{native_id}" +def logical_head_cohort_sql(conn: sqlite3.Connection, *, raw_alias: str, has_memberships: bool) -> str: + """Return the canonical, membership-aware SQL partition key for raw heads.""" + + def _canonical_or_original(value: object) -> str | None: + if value is None: + return None + text = str(value) + try: + return canonical_authority_logical_key(text) + except ValueError: + # A malformed legacy key must remain observable as its own cohort, + # never make a read-only verifier fail while grouping heads. + return text + + conn.create_function("canonical_authority_logical_key", 1, _canonical_or_original, deterministic=True) + membership_key = "NULL" + if has_memberships: + membership_key = f""" + canonical_authority_logical_key( + ( + SELECT CASE + WHEN COUNT(DISTINCT canonical_authority_logical_key(m.logical_source_key)) = 1 + THEN MIN(canonical_authority_logical_key(m.logical_source_key)) + END + FROM raw_session_memberships AS m + WHERE m.raw_id = {raw_alias}.raw_id + ) + ) + """ + return ( + f"COALESCE(canonical_authority_logical_key({raw_alias}.logical_source_key), " + f"{membership_key}, {raw_alias}.native_id, {raw_alias}.source_path)" + ) + + def durable_authority_logical_keys( *, raw_logical_key: object, diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index d6e8926a48..e25e2e8619 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/polylogue/maintenance/archive_verification.py @@ -41,6 +41,7 @@ from pathlib import Path from typing import Any +from polylogue.archive.revision_authority import logical_head_cohort_sql from polylogue.core.json import JSONDocument, json_document from polylogue.core.outcomes import OutcomeCheck, OutcomeReport, OutcomeStatus from polylogue.logging import get_logger @@ -501,6 +502,9 @@ def _valid_byte_duplicate_supersession_expr(conn: sqlite3.Connection, *, raw_ali AND receipt.blob_size = {raw_alias}.blob_size AND twin.blob_hash = {raw_alias}.blob_hash AND twin.blob_size = {raw_alias}.blob_size + AND twin.origin IS {raw_alias}.origin + AND twin.source_path IS {raw_alias}.source_path + AND twin.source_index IS {raw_alias}.source_index ) """ @@ -515,16 +519,11 @@ def _logical_head_cohort_expr(conn: sqlite3.Connection, *, raw_alias: str) -> st they have no one raw-level cohort and must keep that fallback instead of being arbitrarily assigned to one member. """ - membership_key = "NULL" - if table_exists(conn, "raw_session_memberships"): - membership_key = f""" - ( - SELECT CASE WHEN COUNT(*) = 1 THEN MIN(m.logical_source_key) END - FROM raw_session_memberships AS m - WHERE m.raw_id = {raw_alias}.raw_id - ) - """ - return f"COALESCE({raw_alias}.logical_source_key, {membership_key}, {raw_alias}.native_id, {raw_alias}.source_path)" + return logical_head_cohort_sql( + conn, + raw_alias=raw_alias, + has_memberships=table_exists(conn, "raw_session_memberships"), + ) def _check_source_index_coverage(archive_root: Path, sample_limit: int) -> ArchiveVerificationCheck: diff --git a/polylogue/schemas/sampling_db.py b/polylogue/schemas/sampling_db.py index 3a66b1ab03..f072789dfb 100644 --- a/polylogue/schemas/sampling_db.py +++ b/polylogue/schemas/sampling_db.py @@ -15,6 +15,7 @@ from polylogue.archive.artifact_taxonomy import classify_artifact_path from polylogue.archive.raw_payload import extract_record_samples_from_raw_content from polylogue.archive.raw_payload.decode import RawPayloadEnvelope +from polylogue.archive.revision_authority import logical_head_cohort_sql from polylogue.core.enums import Origin, Provider from polylogue.core.json import JSONDocument, JSONValue, require_json_value from polylogue.core.provider_identity import ( @@ -36,6 +37,7 @@ ObservationTerminalStatus, ) from polylogue.storage.blob_store import get_blob_store +from polylogue.storage.introspection import table_exists from polylogue.storage.sqlite.connection_profile import connection_context logger = get_logger(__name__) @@ -289,31 +291,36 @@ def _iter_schema_units_from_db( query_provider = config.db_source_name or source_name origins = _sample_origins_for_provider(Provider.from_string(query_provider), config) placeholders = ",".join("?" for _ in origins) - if logical_heads_only: - query = f""" - WITH heads AS ( - SELECT - source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, - validation_status, - ROW_NUMBER() OVER ( - PARTITION BY origin, COALESCE(logical_source_key, native_id, source_path) - ORDER BY acquired_at_ms DESC, raw_id DESC - ) AS rn - FROM raw_sessions - WHERE origin IN ({placeholders}) - ) - SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, validation_status - FROM heads WHERE rn = 1 - """ - else: - query = f""" - SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, - validation_status - FROM raw_sessions - WHERE origin IN ({placeholders}) - """ with connection_context(source_db_path) as conn: conn.row_factory = sqlite3.Row + if logical_heads_only: + logical_cohort_expr = logical_head_cohort_sql( + conn, + raw_alias="raw_sessions", + has_memberships=table_exists(conn, "raw_session_memberships"), + ) + query = f""" + WITH heads AS ( + SELECT + source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, + validation_status, + ROW_NUMBER() OVER ( + PARTITION BY origin, {logical_cohort_expr} + ORDER BY acquired_at_ms DESC, raw_id DESC + ) AS rn + FROM raw_sessions + WHERE origin IN ({placeholders}) + ) + SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, validation_status + FROM heads WHERE rn = 1 + """ + else: + query = f""" + SELECT source_path, origin, raw_id, blob_hash, file_mtime_ms, acquired_at_ms, + validation_status + FROM raw_sessions + WHERE origin IN ({placeholders}) + """ cursor = conn.execute(query, origins) batch_size = 1 if config.sample_granularity == "record" else 100 # Content-hash dedup: distinct ``raw_id``s legitimately collapse onto the diff --git a/polylogue/storage/raw_byte_duplicate_supersession.py b/polylogue/storage/raw_byte_duplicate_supersession.py index 3e4be236a8..f6c7713977 100644 --- a/polylogue/storage/raw_byte_duplicate_supersession.py +++ b/polylogue/storage/raw_byte_duplicate_supersession.py @@ -114,7 +114,7 @@ def plan_byte_duplicate_supersession( if raw_ids is not None and limit is not None: raise ValueError("raw_ids cannot be combined with limit") query = """ - SELECT raw_id, blob_hash, blob_size + SELECT raw_id, blob_hash, blob_size, origin, source_path, source_index FROM raw_sessions WHERE revision_authority = 'quarantined' AND logical_source_key IS NULL @@ -150,21 +150,22 @@ def plan_byte_duplicate_supersession( # not here, since a raw_id can legitimately be the "other" raw for # every other candidate sharing its hash while still being a # candidate in its own right. - hash_to_raw_ids: dict[bytes, list[str]] = {} + hash_to_raws: dict[bytes, dict[str, sqlite3.Row]] = {} for chunk_start in range(0, len(candidate_hashes), 500): chunk = candidate_hashes[chunk_start : chunk_start + 500] placeholders = ", ".join("?" for _ in chunk) rows = source_conn.execute( - f"SELECT raw_id, blob_hash FROM raw_sessions WHERE blob_hash IN ({placeholders})", + f"SELECT raw_id, blob_hash, origin, source_path, source_index " + f"FROM raw_sessions WHERE blob_hash IN ({placeholders})", chunk, ).fetchall() for row in rows: - hash_to_raw_ids.setdefault(bytes(row["blob_hash"]), []).append(str(row["raw_id"])) + hash_to_raws.setdefault(bytes(row["blob_hash"]), {})[str(row["raw_id"])] = row # Which of those raw_ids are actually materialized in index.db -- # the one live-tier fact this classifier is allowed to read (never # write). - all_raw_ids = sorted({raw_id for raw_ids in hash_to_raw_ids.values() for raw_id in raw_ids}) + all_raw_ids = sorted({raw_id for raws in hash_to_raws.values() for raw_id in raws}) indexed_session_by_raw_id: dict[str, str] = {} for chunk_start in range(0, len(all_raw_ids), 500): raw_id_chunk = all_raw_ids[chunk_start : chunk_start + 500] @@ -184,9 +185,18 @@ def plan_byte_duplicate_supersession( for row in candidate_rows: raw_id = str(row["raw_id"]) blob_hash = bytes(row["blob_hash"]) - other_raw_ids_for_hash = [rid for rid in hash_to_raw_ids.get(blob_hash, []) if rid != raw_id] + other_raws_for_hash = [ + other + for other_id, other in hash_to_raws.get(blob_hash, {}).items() + if other_id != raw_id + and other["origin"] == row["origin"] + and other["source_path"] == row["source_path"] + and other["source_index"] == row["source_index"] + ] indexed_matches = sorted( - other_raw_id for other_raw_id in other_raw_ids_for_hash if other_raw_id in indexed_session_by_raw_id + str(other["raw_id"]) + for other in other_raws_for_hash + if str(other["raw_id"]) in indexed_session_by_raw_id ) if not indexed_matches: novel_count += 1 diff --git a/tests/unit/core/test_schema_generation.py b/tests/unit/core/test_schema_generation.py index 5ed71972af..98ea6cc1d2 100644 --- a/tests/unit/core/test_schema_generation.py +++ b/tests/unit/core/test_schema_generation.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import sqlite3 from collections.abc import Generator from pathlib import Path from types import SimpleNamespace @@ -333,6 +334,41 @@ def _record(*, raw_id: str, status: object, artifact_kind: object, source_path: assert seen_raw_ids == ["raw-new"] # the later of the two revisions is the logical head + def test_logical_heads_only_canonicalizes_retained_membership_aliases(self, tmp_path: Path) -> None: + from polylogue.core.enums import Provider + from polylogue.schemas.observation_identity import resolve_provider_config + from polylogue.schemas.sampling_db import _iter_schema_units_from_db + + self._seed_two_revisions(tmp_path / "source.db") + with sqlite3.connect(tmp_path / "source.db") as conn: + conn.execute("UPDATE raw_sessions SET logical_source_key = 'chatgpt:conv-1' WHERE raw_id = 'raw-new'") + conn.execute( + """ + INSERT INTO raw_session_memberships( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count + ) VALUES ('raw-old', 'chatgpt-export:conv-1', 'conv-1', 'old', ?, 1) + """, + (b"m" * 32,), + ) + + seen_raw_ids: list[str] = [] + + def _record(*, raw_id: str, status: object, artifact_kind: object, source_path: object, reason: object) -> None: + seen_raw_ids.append(raw_id) + + list( + _iter_schema_units_from_db( + Provider.CHATGPT, + db_path=tmp_path / "index.db", + config=resolve_provider_config(Provider.CHATGPT), + terminal_recorder=_record, + logical_heads_only=True, + ) + ) + + assert seen_raw_ids == ["raw-new"] + class TestGenerateSchemaFromSamples: """Focused schema-generation edge cases beyond the general laws.""" diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index 28683a3ef8..c1b7cdaec2 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/tests/unit/maintenance/test_archive_verification.py @@ -545,7 +545,9 @@ def test_source_index_coverage_groups_reacquisitions_by_logical_source_key(tmp_p _seed_coherent_archive(tmp_path) source_conn = _connect(tmp_path / "source.db") try: - source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'") + source_conn.execute( + "UPDATE raw_sessions SET logical_source_key = 'codex-session:session' WHERE raw_id = 'raw-1'" + ) source_conn.execute( """ INSERT INTO raw_sessions( @@ -573,7 +575,9 @@ def test_coverage_groups_retired_membership_identity_with_bound_revision(tmp_pat _seed_coherent_archive(tmp_path) source_conn = _connect(tmp_path / "source.db") try: - source_conn.execute("UPDATE raw_sessions SET logical_source_key = 'codex:session' WHERE raw_id = 'raw-1'") + source_conn.execute( + "UPDATE raw_sessions SET logical_source_key = 'codex-session:session' WHERE raw_id = 'raw-1'" + ) source_conn.execute( """ INSERT INTO raw_sessions( @@ -718,10 +722,10 @@ def test_valid_byte_supersession_receipt_covers_unindexed_head(tmp_path: Path) - source_conn.execute( """ INSERT INTO raw_sessions( - raw_id, origin, native_id, source_path, blob_hash, blob_size, + raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, acquired_at_ms, revision_authority ) - SELECT 'raw-superseded', origin, 'duplicate', '/duplicate', + SELECT 'raw-superseded', origin, 'duplicate', source_path, source_index, blob_hash, blob_size, 200, 'byte_proven' FROM raw_sessions WHERE raw_id = 'raw-1' """ @@ -750,6 +754,41 @@ def test_valid_byte_supersession_receipt_covers_unindexed_head(tmp_path: Path) - assert check.evidence["untyped_count"] == 0 +def test_byte_supersession_receipt_requires_matching_source_semantics(tmp_path: Path) -> None: + """Byte equality cannot collapse a raw whose path-specific replay semantics differ.""" + + _seed_coherent_archive(tmp_path) + with _connect(tmp_path / "source.db") as conn: + conn.execute( + """ + INSERT INTO raw_sessions( + raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, + acquired_at_ms, revision_authority + ) + SELECT 'raw-cross-path-supersession', origin, 'duplicate', '/different-path', source_index, + blob_hash, blob_size, 200, 'byte_proven' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + conn.execute( + """ + INSERT INTO raw_byte_duplicate_supersession_receipts( + raw_id, blob_hash, blob_size, duplicate_of_raw_id, duplicate_of_session_id, + previous_revision_authority, promoted_at_ms, tool_version, backup_manifest_path, detail + ) + SELECT 'raw-cross-path-supersession', blob_hash, blob_size, 'raw-1', + 'codex-session:session', 'quarantined', 200, 'test', '/verified/manifest.json', '' + FROM raw_sessions WHERE raw_id = 'raw-1' + """ + ) + + check = _check(verify_archive(tmp_path, checks=("source-index-coverage",)), "source-index-coverage") + + assert check.status is OutcomeStatus.ERROR + assert check.evidence["superseded_byte_duplicate_count"] == 0 + assert check.evidence["untyped_count"] == 1 + + def test_invalid_byte_supersession_receipt_does_not_cover_unindexed_head(tmp_path: Path) -> None: """A stale/mismatched receipt cannot authorize its own coverage result.""" @@ -1554,10 +1593,10 @@ def test_convergence_freshness_excludes_a_receipt_backed_duplicate(tmp_path: Pat conn.execute( """ INSERT INTO raw_sessions( - raw_id, origin, native_id, source_path, blob_hash, blob_size, + raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, acquired_at_ms, revision_authority ) - SELECT 'raw-superseded-backlog', origin, 'duplicate-backlog', '/duplicate-backlog', + SELECT 'raw-superseded-backlog', origin, 'duplicate-backlog', source_path, source_index, blob_hash, blob_size, 200, 'byte_proven' FROM raw_sessions WHERE raw_id = 'raw-1' """ diff --git a/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py b/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py index c5eb89c1cf..497ca03ab1 100644 --- a/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py +++ b/tests/unit/maintenance/test_raw_byte_duplicate_supersession_apply.py @@ -67,7 +67,7 @@ def _build_fixture_archive(tmp_path: Path) -> Path: archive, raw_id="raw-indexed-twin", payload=_DUPLICATE_PAYLOAD, - source_path=str(tmp_path / "twin.jsonl"), + source_path=str(tmp_path / "duplicate.jsonl"), ) # The genuine duplicate -- byte-identical to the indexed twin, no # logical_source_key: the population this bead resolves. @@ -75,7 +75,7 @@ def _build_fixture_archive(tmp_path: Path) -> Path: archive, raw_id="raw-duplicate", payload=_DUPLICATE_PAYLOAD, - source_path=str(tmp_path / "dup.jsonl"), + source_path=str(tmp_path / "duplicate.jsonl"), ) # A genuinely novel row: quarantined, no logical_source_key, but no # indexed byte-identical twin anywhere -- must never be resolved From 5117704f5790af5de1ae1a716a56fceac5323f05 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 22:40:11 +0200 Subject: [PATCH 8/9] fix(replay): skip terminal non-session frozen raws Honor durable non-session evidence during the historical and frozen replay census so malformed terminal artifacts are accounted for without parser dispatch. --- polylogue/sources/revision_backfill.py | 26 +++++++++-- .../storage/sqlite/archive_tiers/archive.py | 2 +- .../archive_tiers/revision_governance.py | 15 ++++-- tests/unit/storage/test_revision_replay.py | 46 +++++++++++++++++++ 4 files changed, 78 insertions(+), 11 deletions(-) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 5717d15eef..209b9ceb83 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -729,9 +729,19 @@ def bind_byte_proven_older_member(raw_id: str, logical_key: str) -> None: census_selection = initial_selection while True: rows = archive.raw_membership_census_rows(census_selection) - pending_rows = [(raw_id, source_index) for raw_id, source_index in rows if raw_id not in state.censused] + terminal_raw_ids = { + raw_id for raw_id, _source_index, terminal_non_session in rows if terminal_non_session + } + for raw_id in terminal_raw_ids - state.censused: + state.scanned += 1 + state.censused.add(raw_id) + pending_rows = [ + (raw_id, source_index) + for raw_id, source_index, terminal_non_session in rows + if raw_id not in state.censused and not terminal_non_session + ] if max_payload_bytes is not None: - payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in rows]) + payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _index in pending_rows]) total_payload_bytes = sum(payload_sizes.values()) oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes] if oversized or total_payload_bytes > max_payload_bytes: @@ -819,14 +829,18 @@ def _load_frozen_revision_evidence( ) rows = archive.raw_membership_census_rows(expanded_raw_ids if selected_raw_ids is not None else None) if max_payload_bytes is not None: - payload_sizes = archive.raw_payload_sizes([raw_id for raw_id, _source_index in rows]) + payload_sizes = archive.raw_payload_sizes( + [raw_id for raw_id, _source_index, terminal_non_session in rows if not terminal_non_session] + ) total_payload_bytes = sum(payload_sizes.values()) oversized = [raw_id for raw_id, size in payload_sizes.items() if size > max_payload_bytes] if oversized or total_payload_bytes > max_payload_bytes: raise RawRevisionReplayResourceBlockedError( sorted(oversized or payload_sizes), max_payload_bytes, total_payload_bytes ) - parseable_raw_ids = [raw_id for raw_id, source_index in rows if source_index >= 0] + parseable_raw_ids = [ + raw_id for raw_id, source_index, terminal_non_session in rows if source_index >= 0 and not terminal_non_session + ] parsed_outcomes = _parse_retained_raws( archive, parseable_raw_ids, @@ -834,9 +848,11 @@ def _load_frozen_revision_evidence( prefetch_cache=prefetch_cache, ) state = _RevisionCensusState(0, 0, 0, set(), {}, {}) - for raw_id, source_index in rows: + for raw_id, source_index, terminal_non_session in rows: state.scanned += 1 state.censused.add(raw_id) + if terminal_non_session: + continue if source_index < 0: state.quarantined += 1 continue diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 2006366cf4..3841333e58 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -2681,7 +2681,7 @@ def raw_revision_rebuild_selection( ) -> tuple[tuple[tuple[str, int], ...], tuple[str, ...]]: return raw_revision_rebuild_selection(self, raw_ids) - def raw_membership_census_rows(self, raw_ids: Sequence[str] | None = None) -> tuple[tuple[str, int], ...]: + def raw_membership_census_rows(self, raw_ids: Sequence[str] | None = None) -> tuple[tuple[str, int, bool], ...]: return raw_membership_census_rows(self, raw_ids) def raw_payload_sizes(self, raw_ids: Sequence[str]) -> dict[str, int]: diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index f7cb135b76..ba32d52a0d 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -1709,20 +1709,25 @@ def raw_revision_rebuild_selection( def raw_membership_census_rows( store: RawRevisionGovernanceHost, raw_ids: Sequence[str] | None = None -) -> tuple[tuple[str, int], ...]: - """Return every retained raw whose membership census may affect authority.""" +) -> tuple[tuple[str, int, bool], ...]: + """Return retained raws and whether durable evidence says they are non-sessions.""" conn = store._ensure_source_conn() + columns = """ + r.raw_id, + r.source_index, + EXISTS(SELECT 1 FROM raw_artifacts AS a WHERE a.raw_id = r.raw_id AND a.parse_as_session = 0) + """ if raw_ids is None: - rows = conn.execute("SELECT raw_id, source_index FROM raw_sessions ORDER BY raw_id").fetchall() + rows = conn.execute(f"SELECT {columns} FROM raw_sessions AS r ORDER BY r.raw_id").fetchall() elif raw_ids: placeholders = ",".join("?" for _ in raw_ids) rows = conn.execute( - f"SELECT raw_id, source_index FROM raw_sessions WHERE raw_id IN ({placeholders}) ORDER BY raw_id", + f"SELECT {columns} FROM raw_sessions AS r WHERE r.raw_id IN ({placeholders}) ORDER BY r.raw_id", tuple(raw_ids), ).fetchall() else: rows = [] - return tuple((str(row[0]), int(row[1])) for row in rows) + return tuple((str(row[0]), int(row[1]), bool(row[2])) for row in rows) def raw_payload_sizes(store: RawRevisionGovernanceHost, raw_ids: Sequence[str]) -> dict[str, int]: diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index 7e62c47f7c..2cf9d438c3 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -276,6 +276,52 @@ def test_terminal_non_session_failure_has_complete_empty_parser_census(tmp_path: assert require_current_parser_source_census(tmp_path)[raw_id] == () +def test_frozen_replay_skips_typed_terminal_non_session_raw(tmp_path: Path) -> None: + """Terminal non-session evidence settles replay without dispatching its malformed bytes.""" + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b"not valid codex jsonl", + source_path="terminal-replay-corrupt.jsonl", + acquired_at_ms=1, + ) + archive.record_raw_failure_evidence( + raw_id, + provider=Provider.CODEX, + source_path="terminal-replay-corrupt.jsonl", + source_index=0, + acquired_at_ms=1, + kind=RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT, + ) + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CODEX, + error=ValueError("terminal corrupt input"), + preserve_existing_failure_evidence=True, + ) + with archive._ensure_source_conn(): + archive_revision_governance.record_current_parser_source_census(archive._ensure_source_conn(), raw_id) + + from polylogue.sources.revision_backfill import _load_frozen_revision_evidence, _ParsedSessionSpill + + with _ParsedSessionSpill(tmp_path, max_cached_payload_bytes=1024 * 1024) as spill: + census = _load_frozen_revision_evidence( + archive, + spill, + selected_raw_ids=None, + max_payload_bytes=None, + ingest_workers=1, + prefetch_cache=None, + ) + + assert census.scanned == 1 + assert census.censused == {raw_id} + assert census.classified == 0 + assert census.quarantined == 0 + + def test_membership_receipt_excludes_post_parse_pending_identity(tmp_path: Path) -> None: """A parser-derived membership receipt cannot retain its provisional raw key.""" initialize_active_archive_root(tmp_path) From edfcf9812eac632d4721df63bdc59abd26fa18a8 Mon Sep 17 00:00:00 2001 From: Sinity Date: Wed, 12 Aug 2026 22:40:23 +0200 Subject: [PATCH 9/9] fix(harness): bind broad shards and emitter semantics Classify generated seed-testmon shards as broad pytest runs and include session-emitter behavior in seeded artifact source semantics. --- devtools/verify.py | 2 +- polylogue/sources/origin_specs.py | 1 + tests/unit/devtools/test_verify.py | 1 + tests/unit/infra/test_workload_artifacts.py | 4 ++-- tests/unit/sources/test_origin_specs.py | 22 +++++++++++++++++++++ 5 files changed, 27 insertions(+), 3 deletions(-) diff --git a/devtools/verify.py b/devtools/verify.py index 5ab996609e..65d3e2c47f 100644 --- a/devtools/verify.py +++ b/devtools/verify.py @@ -2240,7 +2240,7 @@ def _pytest_command_concurrency(cmd: Sequence[str], *, env: Mapping[str, str] | def _pytest_uses_full_suite_basetemp(label: str) -> bool: """Whether this pytest step can materialize the measured full-suite tree.""" - return label in BROAD_PYTEST_STEP_LABELS.values() + return label in BROAD_PYTEST_STEP_LABELS.values() or label.startswith("pytest seed-testmon shard ") _BROAD_TESTMON_CHANGED_PATHS = { diff --git a/polylogue/sources/origin_specs.py b/polylogue/sources/origin_specs.py index 9bc524179b..4ed4d364e2 100644 --- a/polylogue/sources/origin_specs.py +++ b/polylogue/sources/origin_specs.py @@ -48,6 +48,7 @@ _SOURCE_ROOT = Path(__file__).resolve().parents[2] _LOWERING_FINGERPRINT_PATHS: tuple[str, ...] = ( "polylogue/sources/dispatch.py", + "polylogue/sources/emitter.py", "polylogue/pipeline/ids.py", "polylogue/storage/sqlite/archive_tiers/write.py", "polylogue/archive/session_revision_membership.py", diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index f2d79e4e5e..2d1208e15a 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -2763,6 +2763,7 @@ def to_dict(self) -> dict[str, int]: ("pytest testmon", False), ("pytest testmon (broad)", True), ("pytest seed-testmon", True), + ("pytest seed-testmon shard 1/4", True), ("pytest full (parallel)", True), ("pytest load-sensitive (isolated)", True), ], diff --git a/tests/unit/infra/test_workload_artifacts.py b/tests/unit/infra/test_workload_artifacts.py index 77ad582dc6..1a29db27e0 100644 --- a/tests/unit/infra/test_workload_artifacts.py +++ b/tests/unit/infra/test_workload_artifacts.py @@ -45,9 +45,9 @@ def test_seeded_archive_publishes_valid_immutable_real_pipeline_artifact(tmp_pat def test_seeded_archive_key_changes_with_source_semantics(monkeypatch: pytest.MonkeyPatch) -> None: import tests.infra.workload_artifacts as artifacts - monkeypatch.setattr(artifacts, "_source_semantics_id", lambda: "source-semantics:first") + monkeypatch.setattr(artifacts, "lowering_fingerprint", lambda: "emitter-semantics:first") first = seeded_archive_key(()) - monkeypatch.setattr(artifacts, "_source_semantics_id", lambda: "source-semantics:second") + monkeypatch.setattr(artifacts, "lowering_fingerprint", lambda: "emitter-semantics:second") second = seeded_archive_key(()) assert first.value != second.value diff --git a/tests/unit/sources/test_origin_specs.py b/tests/unit/sources/test_origin_specs.py index f51def6d46..14ce81256d 100644 --- a/tests/unit/sources/test_origin_specs.py +++ b/tests/unit/sources/test_origin_specs.py @@ -119,6 +119,28 @@ def test_parser_fingerprint_changes_when_a_declared_assembly_helper_changes(tmp_ assert before != after +def test_lowering_fingerprint_changes_when_session_emitter_changes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Seeded source semantics include the emitter that admits and enriches sessions.""" + import polylogue.sources.origin_specs as origin_specs + + source_root = tmp_path / "source-root" + source_dir = source_root / "polylogue" / "sources" + source_dir.mkdir(parents=True) + emitter = source_dir / "emitter.py" + emitter.write_text("def emit(payload):\n return payload\n", encoding="utf-8") + monkeypatch.setattr(origin_specs, "_SOURCE_ROOT", source_root) + monkeypatch.setattr(origin_specs, "_LOWERING_FINGERPRINT_PATHS", ("polylogue/sources/emitter.py",)) + origin_specs._fingerprint_sources_cached.cache_clear() + + before = origin_specs.lowering_fingerprint() + emitter.write_text("def emit(payload):\n return {'session': payload}\n", encoding="utf-8") + after = origin_specs.lowering_fingerprint() + + assert before != after + + def test_production_fingerprints_are_stable_across_a_fresh_interpreter() -> None: current_parser = parser_fingerprint_for_origin(Origin.CODEX_SESSION) command = (