From bb6b7ed0c530622f5f0ba5399902d6831e79e9c3 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 28 Jul 2026 13:47:40 +0200 Subject: [PATCH] fix(storage): scope quarantine-refinement inspection by logical source key ## Problem Discovered live 2026-07-28 while investigating polylogue-zaiz (the residual "3 fan-out sessions stuck" finding from the ihc8/ewfp incident chain): `_inspect_quarantined_accepted_raw` looked up the accepted-head row by `accepted_raw_id` alone, requiring exactly one match system-wide -- the same architectural gap `_inspect_duplicate_raw_identity` had before polylogue-ihc8's fix. Forked/subagent/resumed sessions can physically replay the identical parent evidence, so the exact same raw_id can legitimately be the accepted head of several logical source keys/sessions at once. For a fan-out raw, this unconditionally raised "expected one accepted head, found N" for EVERY sibling, regardless of the underlying quarantine reason -- permanently non-actionable, not a crash this time (the census-level `unresolved_provenance` fallback in `_classify_frontier` degrades gracefully), but structurally blocking any of them from ever being refined. Investigation also found a much larger, separate fact while scoping this down: 15,798 of ~41,334 `raw_sessions` rows (38% of the archive) carry `revision_kind='unknown'` -- a pre-existing, already-known backlog predating this session, unrelated to the scoping bug and out of scope for a single fix. Confirmed the two are independent: the untyped/quarantined envelope path already handles `revision_kind='unknown'` correctly: the sole blocker for genuine fan-out cases is the head lookup's scoping. ## Solution - `_inspect_quarantined_accepted_raw` now requires a `logical_source_key` parameter, scoping the `heads`/`session_rows`/`applications` queries to it (mirroring `_inspect_duplicate_raw_identity`'s fix). - `inspect_quarantined_accepted_raws` now takes `(raw_id, logical_source_key)` pairs instead of bare raw_ids (blob-budget partitioning still dedupes by raw_id alone; the per-session identity proof runs once per pair). - `_strategy_overrides` builds per-session override keys (`_quarantine_override_key`) instead of bare-raw_id keys for the quarantine path specifically; the frontier-item lookup tries the session-scoped key first, falling back to the bare key unchanged for browser-origin/conflict overrides (which are properties of the raw itself, not per-session, and don't have this fan-out shape). - `_apply_strategy`'s REFINE_QUARANTINE branch threads the same `logical_source_key` through, and gains the identical graceful "ineligible" no-op handling already shipped for FOLD_DUPLICATE_ALIAS (polylogue-ewfp, #3369) -- a fan-out raw's refinement mutates the SHARED `raw_sessions` row itself (not a distinct canonical twin), so only one sibling can ever win; the others must resolve as a permanent, non-crashing ineligibility if selected in the same apply batch. ## Verification - `devtools test tests/unit/storage/test_quarantined_accepted_raw_repair.py tests/unit/storage/test_duplicate_raw_identity_repair.py tests/unit/storage/test_raw_authority_ledger.py` -> 52 passed (new regression tests reproduce the exact fan-out shape; anti-vacuity confirmed by reverting the scoping fix and observing the exact pre-fix failure) - `devtools test tests/unit/storage/ -k "raw_reconciler or raw_authority or duplicate or quarantine or repair"` -> 228 passed, 1 pre-existing unrelated failure (confirmed via git stash on master) - `mypy polylogue/storage/raw_reconciler.py polylogue/storage/repair.py` -> clean - `devtools verify --quick` -> exit 0 Ref polylogue-zaiz, polylogue-ihc8 Co-Authored-By: Claude --- .beads/issues.jsonl | 2 +- polylogue/storage/raw_reconciler.py | 95 ++++++-- polylogue/storage/repair.py | 75 ++++-- .../test_quarantined_accepted_raw_repair.py | 225 ++++++++++++++++++ 4 files changed, 361 insertions(+), 36 deletions(-) diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 06fb1afdcf..beecb7a915 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -397,7 +397,7 @@ {"_type":"issue","id":"polylogue-sru.2","title":"Characterize ambiguous bucket: wordless continuation vs prose-without-markers","description":"Split next-turn-is-tool-call (wordless continuation) from prose-lacking-ack-markers; state counts for both. Opus-4-7 74% ambiguous vs deepseek 17% is likely turn-structure variance, not behavior — this split disambiguates.","design":"Implementation home: the claim-vs-evidence classifier in devtools (devtools/ module behind `devtools workspace claim-vs-evidence`; tests tests/unit/devtools/test_claim_vs_evidence.py). Wordless-continuation detection: for each failure's paired next assistant message, check whether its blocks contain tool_use and no text block with \u003eN chars before the first tool_use — that is 'wordless continuation'; prose without matched ack markers stays 'ambiguous-prose'. Emit both as classification_reason variants (field already exists) and add the two counts to the report summary + by_model/by_tool cuts. Regen: `devtools workspace claim-vs-evidence --limit 5000 --out-dir .agent/demos/claim-vs-evidence --json`. Acceptance: report shows ambiguous split into wordless_continuation vs prose_no_marker with counts; per-model ambiguous variance (opus-4-7 74% vs deepseek 17%) re-examined after the split.","notes":"2026-07-03 Codex WIP: unit implementation for ambiguous split passes focused tests, but live regeneration with --limit 5000 became too slow and had to be killed twice. First attempt used correlated subqueries for next-message block shape; second used set-based CTE; third used chunked second query after sampled rows, but the full command still exceeded 90s on active archive and ignored SIGINT while inside SQLite. Do not close or commit this slice until the live regeneration path is profiled/fixed. Dirty files currently show the WIP implementation: devtools/claim_vs_evidence.py and tests/unit/devtools/test_claim_vs_evidence.py. Last passing focused proof: python -m py_compile + ruff check + devtools test tests/unit/devtools/test_claim_vs_evidence.py -\u003e 3 passed.","status":"closed","priority":1,"issue_type":"task","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-07-03T04:31:28Z","created_by":"Sinity","updated_at":"2026-07-03T07:45:10Z","started_at":"2026-07-03T07:09:21Z","closed_at":"2026-07-03T07:45:10Z","close_reason":"Completed: claim-vs-evidence now splits ambiguous follow-ups into wordless tool continuations and prose-without-marker buckets, reports the counts in JSON/README summaries, and regenerates the current demo on the active archive. Focused tests pass; live regen/check completed.","labels":["area:substrate","campaign"],"dependencies":[{"issue_id":"polylogue-sru.2","depends_on_id":"polylogue-sru","type":"parent-child","created_at":"2026-07-03T06:31:27Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-sru.3","title":"Benign-recovery vs consequential-silence split by handler kind","description":"Read failures are ~94% silent but 'tried another path' is usually benign; Bash/test failures are the consequential class. Scope the headline to consequential handler kinds or add an explicit split — credibility depends on not inflating with trivial recoveries.","design":"Handler kind is already available on the paired failure row (actions lane exposes handler/tool). Define the consequential set explicitly in code (Bash/test/build/write-class handlers) and the benign-recovery set (Read/Glob/Grep-class 'tried another path'), emit split headline rows: silent-proceed among consequential vs among all. Keep the mapping a named constant with a rationale comment so reviewers can argue with it. Report both; never let the headline mix classes silently. Same regen/tests as the other methodology children.","status":"closed","priority":1,"issue_type":"task","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-07-03T04:31:28Z","created_by":"Sinity","updated_at":"2026-07-03T07:58:08Z","started_at":"2026-07-03T07:55:37Z","closed_at":"2026-07-03T07:58:08Z","close_reason":"Completed: claim-vs-evidence now reports a first-class handler-class split separating consequential shell/edit/write-class tool failures from benign read/search/path-discovery failures and other tools. The regenerated active-archive artifact shows consequential=4,177 failures with 921 silent-proceed (22.0% lower bound), benign_recovery=633 with 166 silent-proceed (26.2%), and other=190 with 92 silent-proceed (48.4%). Focused tests and demo shelf checks passed.","labels":["area:substrate","campaign"],"dependencies":[{"issue_id":"polylogue-sru.3","depends_on_id":"polylogue-sru","type":"parent-child","created_at":"2026-07-03T06:31:28Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-sru.1","title":"Expose action-unit outcome fields + followup_class as product capability","description":"Capabilities-may-not-be-silos gate for the campaign: the facts the report needs must become composable query capability. After this, the whole report is `actions where is_error:true | group by session.origin, followup_class | count` and every future cut (model/tool/repo/time) is free.","design":"1) is_error/exit_code are normalized at parse time (sources/parsers/base_models.py:74-75) but ActionQueryRowPayload (surfaces/payloads.py:~1298) carries neither — add as filterable/groupable action-unit fields. 2) Add derived followup_class (acknowledged|silent_proceed|wordless_continuation|ambiguous) + followup_message_ref computed in the source-derived lowering (no cache tables). 3) Reduce devtools workspace claim-vs-evidence to a render preset over these query strings, or retire it. Touchpoint chain: stage parser -\u003e AST to_payload -\u003e executor -\u003e metadata.py aggregate_group_fields -\u003e shell_completion_values.py -\u003e devtools render openapi + cli-output-schemas + cli-reference. Line refs pre-07-03; re-locate.","acceptance_criteria":"Fixture session with known unacknowledged failure fires via pure query strings; report README numbers reproducible from the printed queries.","notes":"Completed: action-unit outcome follow-up classification is now shared query capability. is_error/exit_code were already wired; this slice added source-derived followup_class and followup_message_ref over existing actions/messages/blocks, exposed followup_class as filterable/groupable action metadata, added action row payload fields, routed root CLI terminal-unit aggregate expressions before session-selector compilation, and moved the report classifier from scripts into polylogue.archive.actions.followup. Reproduction/query forms are now printed in .agent/demos/claim-vs-evidence/PUBLIC_REPRODUCTION.md: actions where is_error:true | group by followup_class | count; actions where followup_class:silent_proceed. Verification: focused DSL/report/CLI tests passed; active demo packet regenerated over archive root /home/sinity/.local/share/polylogue schema v23 with 41,886 structured failures and 5,000 inspected; devtools verify --quick passed run 20260703T092510Z-quick-718233-46e8b587.","status":"closed","priority":1,"issue_type":"feature","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-07-03T04:31:27Z","created_by":"Sinity","updated_at":"2026-07-03T09:25:36Z","started_at":"2026-07-03T09:05:37Z","closed_at":"2026-07-03T09:25:36Z","close_reason":"Completed","labels":["area:query","area:substrate","campaign"],"dependencies":[{"issue_id":"polylogue-sru.1","depends_on_id":"polylogue-sru","type":"parent-child","created_at":"2026-07-03T06:31:26Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} -{"_type":"issue","id":"polylogue-zaiz","title":"Resolve quarantined fan-out sessions via refine_quarantined_raw actuator","status":"open","priority":2,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-07-28T11:10:11Z","created_by":"Sinity","updated_at":"2026-07-28T11:10:11Z","dependencies":[{"issue_id":"polylogue-zaiz","depends_on_id":"polylogue-ihc8","type":"relates-to","created_at":"2026-07-28T13:10:35Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"polylogue-zaiz","title":"Resolve quarantined fan-out sessions via refine_quarantined_raw actuator","notes":"2026-07-28 investigation (proper, careful treatment as requested): read\n_inspect_quarantined_accepted_raw (polylogue/storage/repair.py:698) in\nfull before touching anything.\n\nFound TWO layered facts, not one isolated bug:\n\n1. Same architectural gap class as ihc8/dmvo/ewfp: this function's `heads`\n lookup (`SELECT ... FROM raw_revision_heads WHERE accepted_raw_id = ?`)\n is unscoped by logical_source_key and requires `len(heads) == 1`,\n exactly like ihc8's original bug in _inspect_duplicate_raw_identity.\n For this fan-out shape (3 sessions sharing one stale raw_id), this\n ALWAYS returns \"expected one accepted head, found 3\" -- structurally\n ineligible for refinement regardless of anything else. This part alone\n would need the same per-session scoping fix ihc8 got.\n\n2. MORE IMPORTANTLY -- a much bigger, pre-existing finding that changes\n the scope entirely. Direct read-only query:\n\n SELECT revision_kind, revision_authority, COUNT(*) FROM raw_sessions\n GROUP BY revision_kind, revision_authority;\n\n unknown|quarantined|15798\n full|byte_proven|13467\n full|quarantined|7773\n append|byte_proven|2396\n append|quarantined|1900\n\n 15,798 of ~41,334 raw_sessions rows (38% of the ENTIRE archive) carry\n revision_kind='unknown' -- the schema's bare default, meaning these rows\n predate whatever backfill/migration established byte-proven revision\n classification and were simply never classified. This is NOT specific\n to the 3 stuck fan-out sessions; it is the exact same large-scale\n pre-existing gap already surfaced this session as \"2214 active index\n raw seeds with a broken predecessor chain\" in `polylogue status --full`.\n\n Critically: `_inspect_quarantined_accepted_raw`'s eligibility proof\n compares the raw's actual envelope against an EXPECTED envelope that\n hardcodes `RawRevisionKind.FULL.value`\n (polylogue/storage/repair.py, expected_envelope construction). A raw\n whose OWN stored revision_kind is 'unknown' can never match this\n envelope -- meaning the refine_quarantined_raw actuator, AS DESIGNED,\n cannot resolve ANY of the 15,798 unknown-kind rows, not just these 3.\n Refining them requires first backfilling their revision_kind\n classification post-hoc (determining whether each was originally a full\n snapshot or an append delta from other evidence), which is an entirely\n separate, much larger undertaking than fixing a code bug -- likely the\n real substance behind the \"archive convergence\" work already assessed\n this session (z9gh's dependency tree, b5l durable-tier transition,\n 1xc scale-hardening) as multi-week, not something to force here.\n\nConclusion: these 3 sessions are not a small scoped follow-up like\nihc8/ewfp were. They are the visible tip of the archive's large,\nalready-known 38%-of-rows revision-classification backfill gap. Scoping\nthis bead down to \"fix these 3 sessions\" would be misleading -- either\n(a) fix only the head-scoping bug (matching ihc8's pattern), which would\nstill not let these 3 refine successfully since their revision_kind is\n'unknown', not FULL, or (b) undertake the real fix (revision_kind\nbackfill for 15,798 rows), which is out of scope for a single bead and\nbelongs with the already-tracked large-scale convergence program.\n\nRecommend: close this bead's narrow framing as \"investigated, correctly\nre-scoped\" rather than continuing to treat it as an isolated fix. The\nhead-scoping bug (item 1 above) is real and independently worth fixing\n(it affects the refine_quarantined_raw actuator for EVERY future fan-out\ncase, not just these 3 sessions) -- filing that narrowly-scoped slice\nseparately, since it's genuinely small and safe regardless of the larger\nbackfill question.\n","status":"open","priority":2,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-07-28T11:10:11Z","created_by":"Sinity","updated_at":"2026-07-28T11:22:47Z","dependencies":[{"issue_id":"polylogue-zaiz","depends_on_id":"polylogue-ihc8","type":"relates-to","created_at":"2026-07-28T13:10:35Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-dmvo","title":"fold_duplicate_alias: N:1 stale-raw fan-out leaves N-1 sessions permanently unresolvable","description":"Follow-up from polylogue-ihc8 (fix PR #3326, commit 41baf9935). The\nscoping fix made _inspect_duplicate_raw_identity correctly disambiguate\nby logical_source_key, so it no longer cross-contaminates a strategy\nwitness between fan-out siblings. But it did not address a deeper\nstructural limitation confirmed live against /realm/db/polylogue:\n\nThe same physical stale raw (old native-id-inclusive scheme) can be the\naccepted head of MULTIPLE sessions simultaneously -- forked/subagent/\nresumed Claude Code sessions replay the identical parent JSONL, so each\nsession's materialization independently accepts that raw as its own\nhead. Observed live: raw_id 08f40243e9...ce9e0 is the accepted head of\nFOUR sessions (560a3328-..., 0f5e001c-..., 850e32cf-..., 896c6b64-...)\nsimultaneously, but there is only ONE dangling canonical twin raw\n(e869e6bf...8d6f0) available to fold onto.\n\nOnce the scoping fix (#3326) is deployed and the daemon actually applies\na fold, exactly one of the four sessions will repoint onto the canonical\nraw (whichever plan gets selected/applied first). The other three\nsessions' fold_duplicate_alias plans will then re-classify: canonical_head\nwill no longer be None (canonical_raw is now claimed by the session that\nwon), so _inspect_duplicate_raw_identity's ineligible(\"canonical raw is\nalready an accepted head; not a dangling duplicate\") branch fires for\nthem going forward -- NOT \"already_repaired\". _classify_frontier's\ncurrent filter only treats {\"eligible\",\"already_repaired\"} as\nselectable, so these three should fall OUT of the retryable frontier\nselection on the next census cycle rather than looping -- but this needs\nlive confirmation once #3326 actually deploys and runs, since it has\nnever been observed post-fix.\n\nNeeds: (1) confirm live post-deploy behavior matches this expectation\n(3 remaining sessions correctly drop out of the retryable frontier\ninstead of continuing to loop under a different symptom); (2) decide\nwhether those 3 sessions' stale-raw pointers are a legitimate permanent\nstate (each session's own accepted revision is still intact and\nreadable via the stale raw -- nothing is lost, just not deduped) or\nwhether a different actuator/design (e.g. per-session copies, or\naccepting N:1 as permanently-undeduped and suppressing them from the\nfrontier with an honest terminal reason) is needed so they don't sit as\ninvisible/silent debt.\n\nRef polylogue-ihc8","status":"closed","priority":2,"issue_type":"bug","owner":"ezo.dev@gmail.com","created_at":"2026-07-27T22:03:55Z","created_by":"Sinity","updated_at":"2026-07-28T09:54:45Z","closed_at":"2026-07-28T09:54:45Z","close_reason":"Fixed and confirmed live: PR #3368 (merged, deployed) added the missing\n\"ineligible\" branch to _classify_frontier, resolving the original crash\nthis bead tracked. Confirmed live: one fan-out sibling (896c6b64) has now\nsuccessfully folded onto the canonical raw, proving the fix works\nend-to-end in production, not just in tests.\n\nA NEW, distinct bug surfaced immediately after in the postflight\nverification layer (raw_authority.py:1417) -- tracked separately as the\nnewly-filed bead (see notes) since it's a genuinely different invariant\nat a different code layer, not a recurrence of what this bead tracked.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-0twa","title":"Stale 'query_units' vs 'api.query_units' call-log naming assertion in test_execution_control.py","description":"tests/unit/archive/query/test_execution_control.py::test_api_query_units_routes_through_execution_control and\n::test_api_multi_aggregate_receipt_reports_real_work_selection_and_delivery both fail on current master\n(verified 2026-07-27, unrelated to polylogue-1ldl/polylogue-5202): they assert the execution-control\ncall log records the operation name as \"api.query_units\", but the production code now logs it as plain\n\"query_units\" (assert 'query_units' == 'api.query_units' / assert ['query_units'] == ['api.query_units']).\n\nOriginally noted as an aside in polylogue-1ldl's investigation (\"Also noted in the same run ... separate\nstale assertion, same file\"), filed here as its own tracked item since it is a distinct assertion in\ndistinct tests, not part of 1ldl's VM-step-canary scope.\n\nNeeds the same \"verify current behavior is correct first\" treatment as 1ldl/5202: confirm whether the\n\"api.\" prefix was deliberately dropped by whatever call-site changed the logged operation name (grep\ncall-log call sites in polylogue/archive/query/execution_control.py and wherever query_units is invoked),\nand only then update the two assertions to match -- or, if the prefix drop was accidental, restore it in\nproduction instead of the tests.","status":"closed","priority":2,"issue_type":"bug","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-07-27T17:31:37Z","created_by":"Sinity","updated_at":"2026-07-27T20:37:20Z","started_at":"2026-07-27T20:37:19Z","closed_at":"2026-07-27T20:37:20Z","close_reason":"Fixed in PR #3352: confirmed via git history (query_units_transaction_request, introduced #3068) that plain 'query_units' is the intentional shared operation name across API/MCP/daemon surfaces, not 'api.query_units'. Updated both stale test assertions to match; devtools test passes 23/23, mypy --strict and ruff clean.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-lbgc","title":"seeded-archive corpus build hits 'database is locked' under xdist parallel first-build","description":"Multiple test files that depend on tests/infra/workload_artifacts.py's build_seeded_archive()/named_seeded_archive() (a cached, cross-run reusable real-pipeline archive artifact) fail with sqlite3.OperationalError: database is locked at tests/infra/workload_artifacts.py:274 (_sqlite_integrity's PRAGMA journal_mode=DELETE step), when the named corpus cache is cold and must be built fresh under devtools verify --all's default -n 2 xdist parallelism.\n\nConfirmed reproducible in isolation: cleared /realm/tmp/polylogue-pytest and /dev/shm/pytest-polylogue-seeded-* caches, then ran pytest tests/unit/cli/test_plain_cli_snapshots.py -q -n 2 by itself (not the full suite) -- identical failure, same file, same line, same shape (1 failed + 8 errors: \"database is locked\" during corpus build, plus a separate real snapshot-staleness failure already fixed elsewhere). Affects at minimum tests/unit/cli/test_plain_cli_snapshots.py (8 fixtures use postmortem_seeded_env -\u003e named_seeded_archive('cli-mixed')) and tests/unit/core/test_schema_generation.py (seeded_archive -\u003e schema_coverage_corpus_specs()), likely others sharing the same first-build race.\n\nRoot cause not fully diagnosed, but NOT simple cross-worker contention: build_seeded_archive() already holds an exclusive fcntl.flock() on a per-corpus-key lock file for the ENTIRE build+integrity-check critical section, so two xdist workers building the SAME named corpus should serialize correctly, not race each other's PRAGMA journal_mode=DELETE call. The lock is more likely intra-process: parse_sources_archive(staging, sources) (called inside the same flock'd section, before _sqlite_integrity) may leave a second sqlite3 connection to the same staging index.db open (e.g. a background convergence/materialization step, or a connection-pool handle not fully closed) by the time _sqlite_integrity tries to exclusively switch journal_mode=DELETE on that file -- SQLite locking is per-file across all connections in the process, not just cross-process, so an unclosed sibling connection within the same worker process would produce exactly this symptom under load (more real time elapses per stage under -n 2 parallel CPU contention, widening the race window).\n\nSuggest: audit the parse_sources_archive/materialize/index call chain (used inside build_seeded_archive) for any sqlite3.connect() that isn't guaranteed-closed (via context manager or explicit close()) before _sqlite_integrity runs, and/or check whether the daemon convergence/embed-catchup paths spawn a background thread holding a connection open past the synchronous ingest call's return.","status":"closed","priority":2,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-07-27T17:25:39Z","created_by":"Sinity","updated_at":"2026-07-27T21:21:06Z","closed_at":"2026-07-27T21:21:06Z","close_reason":"Fixed in PR #3363: root cause was CPython sqlite3's deferred sqlite3_close_v2 zombie-connection release colliding with PRAGMA journal_mode=DELETE (SQLITE_LOCKED, not retried by busy_timeout), not cross-worker flock contention (confirmed the existing per-key fcntl.flock already serializes correctly across processes). Fixed via contextlib.closing + a bounded gc.collect()+retry in tests/infra/workload_artifacts.py. Verified: cold-cache pytest -n 2 repro went from reliably failing (8 errors) to 24 passed across 4 repeated runs.","dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/polylogue/storage/raw_reconciler.py b/polylogue/storage/raw_reconciler.py index 86c3b13e3f..03b6c0c02f 100644 --- a/polylogue/storage/raw_reconciler.py +++ b/polylogue/storage/raw_reconciler.py @@ -19,7 +19,7 @@ from dataclasses import dataclass from enum import StrEnum from pathlib import Path -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, TypeVar, cast from polylogue.config import Config from polylogue.core.json import JSONDocument, json_document @@ -81,6 +81,24 @@ class RawAuthorityActuator(StrEnum): _VERIFIED_BLOB_STATS: dict[str, tuple[int, int, int, int, int]] = {} +_ChunkT = TypeVar("_ChunkT") + +_QUARANTINE_OVERRIDE_KEY_SEP = "\x00" + + +def _quarantine_override_key(raw_id: str, logical_source_key: str) -> str: + """Session-scoped override key (polylogue-zaiz): distinct from bare raw_id keys. + + Browser-origin/conflict overrides stay keyed by bare raw_id (those + proofs are properties of the raw itself, not per-session) -- only + quarantine-refinement needs per-session scoping, so this uses a + reserved separator no real raw_id/logical_source_key can contain + (raw_ids are hex digests; logical_source_keys are colon-joined + provider identifiers) to guarantee it never collides with a bare + raw_id key. + """ + return f"{raw_id}{_QUARANTINE_OVERRIDE_KEY_SEP}{logical_source_key}" + def _canonical_json(value: object) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) @@ -197,7 +215,7 @@ def _rows(cursor: sqlite3.Cursor) -> list[dict[str, object]]: return [{name: _json_value(value) for name, value in zip(names, row, strict=True)} for row in cursor.fetchall()] -def _chunks(values: Sequence[str], size: int = 100) -> Iterator[list[str]]: +def _chunks(values: Sequence[_ChunkT], size: int = 100) -> Iterator[list[_ChunkT]]: """Yield bounded strategy-proof requests in deterministic order.""" for start in range(0, len(values), size): yield list(values[start : start + size]) @@ -644,18 +662,26 @@ def _strategy_overrides( ), input_raw_ids=tuple(sorted({conflict_item.raw_id, conflict_item.competing_raw_id})), ) - quarantine_ids = sorted( + # (raw_id, logical_source_key) pairs, not bare raw_ids: a fan-out raw + # shared by several sessions (polylogue-zaiz, mirroring polylogue-ihc8) + # needs a proof scoped to each session, not one shared proof that + # either raises "expected one accepted head, found N" for every + # sibling or silently proves a witness against the wrong session's + # head. Override keys below therefore include the logical_source_key. + quarantine_pairs = sorted( { - str(row["accepted_raw_id"]) + (str(row["accepted_raw_id"]), str(row["logical_source_key"])) for row in rows - if row.get("revision_authority") == "quarantined" and str(row["accepted_raw_id"]) not in browser_ids + if row.get("revision_authority") == "quarantined" + and str(row["accepted_raw_id"]) not in browser_ids + and row.get("logical_source_key") is not None } ) - for quarantine_chunk in _chunks(quarantine_ids): + for quarantine_chunk in _chunks(quarantine_pairs, size=100): quarantine_items = inspect_quarantined_accepted_raws(config, quarantine_chunk) - for quarantine_item in quarantine_items: + for (raw_id, logical_source_key), quarantine_item in zip(quarantine_chunk, quarantine_items, strict=True): if quarantine_item.status in {"eligible", "already_repaired"}: - overrides[quarantine_item.raw_id] = _StrategyOverride( + overrides[_quarantine_override_key(raw_id, logical_source_key)] = _StrategyOverride( state=RawAuthorityFrontierState.SAFELY_REKEYABLE, actuator=RawAuthorityActuator.REFINE_QUARANTINE, reason="quarantined-raw strategy proved exact accepted-byte and semantic authority", @@ -965,10 +991,17 @@ def _frontier_items(config: Config) -> tuple[tuple[RawAuthorityFrontierItem, ... conn.execute("ATTACH DATABASE ? AS index_tier", (str(index_db),)) head_rows = _frontier_rows(conn) overrides = _strategy_overrides(config, head_rows) - head_items = [ - _classify_frontier(conn, BlobStore(root / "blob"), row, overrides.get(str(row["accepted_raw_id"]))) - for row in head_rows - ] + + def _override_for(row: dict[str, object]) -> _StrategyOverride | None: + raw_id = str(row["accepted_raw_id"]) + logical_source_key = row.get("logical_source_key") + if logical_source_key is not None: + scoped = overrides.get(_quarantine_override_key(raw_id, str(logical_source_key))) + if scoped is not None: + return scoped + return overrides.get(raw_id) + + head_items = [_classify_frontier(conn, BlobStore(root / "blob"), row, _override_for(row)) for row in head_rows] superseded_items = _terminal_superseded_items(conn) all_items = _apply_judgment_dispositions(config, (*head_items, *superseded_items)) return ( @@ -1253,6 +1286,9 @@ def _apply_strategy( } ) if item.actuator is RawAuthorityActuator.REFINE_QUARANTINE: + if item.logical_source_key is None: + raise RuntimeError("quarantine-refinement plan is missing the logical source key it was proven against") + logical_source_key = item.logical_source_key with RebuildLease(root), closing(sqlite3.connect(f"file:{source_db}?mode=rw", uri=True)) as source_conn: source_conn.execute("PRAGMA foreign_keys = ON") _attach_repair_index(source_conn, index_db) @@ -1265,17 +1301,46 @@ def _apply_strategy( raise RuntimeError( f"quarantine repair blob budget refused {item.raw_id}: {budget_excluded[0].reason}" ) - quarantine_locked = _inspect_quarantined_accepted_raw(root, item.raw_id, conn=source_conn) + quarantine_locked = _inspect_quarantined_accepted_raw( + root, item.raw_id, conn=source_conn, logical_source_key=logical_source_key + ) + if quarantine_locked.status == "ineligible": + # polylogue-zaiz: mirrors the fold_duplicate_alias + # batch-race fix (polylogue-ewfp, #3369). A fan-out raw + # shared by several sessions can have more than one + # sibling selected together in the same apply batch + # (all looked eligible at the pre-apply census + # snapshot); once one sibling's refinement commits, + # every other sibling's own re-inspection legitimately + # finds "ineligible" -- permanently, not transiently. + # The witness comparison below is skipped deliberately: + # it exists to catch drift that would invalidate an + # ELIGIBLE refinement between census and apply, which + # does not apply once no refinement is being attempted + # at all. + source_conn.commit() + return json_document( + { + "strategy": item.actuator.value, + "repaired_count": 0, + "already_repaired_count": 0, + "ineligible_reason": quarantine_locked.reason, + } + ) if _quarantine_strategy_witness(quarantine_locked) != item.strategy_witness: raise RuntimeError("quarantine strategy proof changed after plan authorization") if quarantine_locked.status == "eligible" and quarantine_locked.census_stage_raw_ids: _stage_quarantined_census_cohort(source_conn, quarantine_locked) - quarantine_locked = _inspect_quarantined_accepted_raw(root, item.raw_id, conn=source_conn) + quarantine_locked = _inspect_quarantined_accepted_raw( + root, item.raw_id, conn=source_conn, logical_source_key=logical_source_key + ) if quarantine_locked.status == "eligible": _cas_refine_quarantined_accepted_raw(source_conn, quarantine_locked) elif quarantine_locked.status != "already_repaired": raise RuntimeError(f"quarantine strategy lost its exact proof: {quarantine_locked.reason}") - quarantine_after = _inspect_quarantined_accepted_raw(root, item.raw_id, conn=source_conn) + quarantine_after = _inspect_quarantined_accepted_raw( + root, item.raw_id, conn=source_conn, logical_source_key=logical_source_key + ) if quarantine_after.status != "already_repaired": raise RuntimeError("quarantine strategy did not reach its typed terminal postcondition") source_conn.commit() diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 8a74f19dc7..037d17d628 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -700,8 +700,26 @@ def _inspect_quarantined_accepted_raw( raw_id: str, *, conn: sqlite3.Connection, + logical_source_key: str, ) -> QuarantinedAcceptedRawRepairItem: - """Prove one accepted head against source main + attached read-only index.""" + """Prove one accepted head against source main + attached read-only index. + + ``logical_source_key`` disambiguates which accepted-head row this proof + is about. A single physical raw acquisition (one ``raw_id``, most + commonly the pre-#2729 native-id-inclusive shape) can legitimately be + the accepted head for *several* logical source keys/sessions at once -- + forked/subagent/resumed sessions replay the same parent JSONL, so their + materialization can each independently accept the identical raw as + their own head (polylogue-ihc8). Scoping every lookup below to this + specific key mirrors the fix already applied to + ``_inspect_duplicate_raw_identity`` for the same fan-out shape + (polylogue-zaiz) -- an unscoped ``WHERE accepted_raw_id = ?`` lookup + here would either raise "expected one accepted head, found N" for every + fan-out sibling (permanently ineligible, regardless of the underlying + quarantine reason) or, worse, silently prove a witness against a + *different* session's head than the one the caller actually intends to + refine. + """ capture_mode_available = _raw_sessions_capture_mode_available(conn) capture_mode_projection = "capture_mode" if capture_mode_available else "NULL AS capture_mode" try: @@ -724,20 +742,21 @@ def _inspect_quarantined_accepted_raw( accepted_source_revision, accepted_content_hash, accepted_frontier_kind, accepted_frontier, acquisition_generation, append_end_offset, decided_at_ms - FROM index_tier.raw_revision_heads WHERE accepted_raw_id = ? + FROM index_tier.raw_revision_heads WHERE accepted_raw_id = ? AND logical_source_key = ? """, - (raw_id,), + (raw_id, logical_source_key), ).fetchall() if len(heads) != 1: - return _quarantined_raw_item(raw_id, f"expected one accepted head, found {len(heads)}") + return _quarantined_raw_item( + raw_id, f"expected one accepted head for {logical_source_key}, found {len(heads)}" + ) head = heads[0] - logical_source_key = str(head["logical_source_key"]) session_id = str(head["session_id"]) session_rows = conn.execute( - "SELECT session_id, raw_id, content_hash FROM index_tier.sessions WHERE raw_id = ?", - (raw_id,), + "SELECT session_id, raw_id, content_hash FROM index_tier.sessions WHERE raw_id = ? AND session_id = ?", + (raw_id, session_id), ).fetchall() - if len(session_rows) != 1 or str(session_rows[0]["session_id"]) != session_id: + if len(session_rows) != 1: return _quarantined_raw_item(raw_id, "accepted head is not the raw's unique indexed session") session_row = session_rows[0] applications = conn.execute( @@ -747,9 +766,9 @@ def _inspect_quarantined_accepted_raw( accepted_source_revision, accepted_content_hash, append_end_offset, baseline_raw_id, predecessor_raw_id, detail, decided_at_ms FROM index_tier.raw_revision_applications - WHERE raw_id = ? OR accepted_raw_id = ? OR logical_source_key = ? + WHERE logical_source_key = ? AND (raw_id = ? OR accepted_raw_id = ?) """, - (raw_id, raw_id, logical_source_key), + (logical_source_key, raw_id, raw_id), ).fetchall() if len(applications) != 1: return _quarantined_raw_item(raw_id, "competing raw-revision application authority exists") @@ -1179,13 +1198,21 @@ def _cas_refine_quarantined_accepted_raw( def inspect_quarantined_accepted_raws( config: Config, - raw_ids: list[str], + raw_ids_with_keys: list[tuple[str, str]], ) -> tuple[QuarantinedAcceptedRawRepairItem, ...]: - """Return exact typed quarantine-refinement proofs without mutation.""" - if len(set(raw_ids)) != len(raw_ids): - raise ValueError("duplicate raw ids are not allowed") - if not raw_ids or len(raw_ids) > _QUARANTINED_ACCEPTED_RAW_REPAIR_LIMIT: + """Return exact typed quarantine-refinement proofs without mutation. + + Each entry is ``(raw_id, logical_source_key)``: a fan-out raw shared by + several sessions (polylogue-zaiz, mirroring polylogue-ihc8) needs one + proof per session, not one per physical raw. Blob-budget partitioning + still dedupes by raw_id alone (blob content doesn't vary by session); + only the per-session identity proof runs once per pair. + """ + if len(set(raw_ids_with_keys)) != len(raw_ids_with_keys): + raise ValueError("duplicate (raw id, logical source key) pairs are not allowed") + if not raw_ids_with_keys or len(raw_ids_with_keys) > _QUARANTINED_ACCEPTED_RAW_REPAIR_LIMIT: raise ValueError(f"raw-id list must contain 1..{_QUARANTINED_ACCEPTED_RAW_REPAIR_LIMIT} entries") + raw_ids = sorted({raw_id for raw_id, _logical_source_key in raw_ids_with_keys}) if any(re.fullmatch(r"[0-9a-f]{64}", raw_id) is None for raw_id in raw_ids): raise ValueError("raw ids must be lowercase SHA-256 identifiers") archive_root = _raw_materialization_archive_root(config) @@ -1196,11 +1223,19 @@ def inspect_quarantined_accepted_raws( with closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True)) as conn: _attach_repair_index(conn, index_db) inspectable, excluded = _partition_quarantined_raw_repair_blob_budget(conn, raw_ids) - by_id = {item.raw_id: item for item in excluded} - by_id.update( - {raw_id: _inspect_quarantined_accepted_raw(archive_root, raw_id, conn=conn) for raw_id in inspectable} - ) - return tuple(by_id[raw_id] for raw_id in raw_ids) + inspectable_ids = set(inspectable) + excluded_by_id = {item.raw_id: item for item in excluded} + results: list[QuarantinedAcceptedRawRepairItem] = [] + for raw_id, logical_source_key in raw_ids_with_keys: + if raw_id in inspectable_ids: + results.append( + _inspect_quarantined_accepted_raw( + archive_root, raw_id, conn=conn, logical_source_key=logical_source_key + ) + ) + else: + results.append(excluded_by_id[raw_id]) + return tuple(results) def _browser_origin_ineligible(raw_id: str, reason: str) -> BrowserCaptureOriginRepairItem: diff --git a/tests/unit/storage/test_quarantined_accepted_raw_repair.py b/tests/unit/storage/test_quarantined_accepted_raw_repair.py index eb9a5b0b2b..e8ee18c776 100644 --- a/tests/unit/storage/test_quarantined_accepted_raw_repair.py +++ b/tests/unit/storage/test_quarantined_accepted_raw_repair.py @@ -14,10 +14,12 @@ from polylogue.sources.revision_backfill import _parse_one from polylogue.storage.blob_store import BlobStore from polylogue.storage.raw_reconciler import ( + RawAuthorityActuator, RawAuthorityFrontierState, apply_raw_authority_frontier, inspect_raw_authority_frontier, ) +from polylogue.storage.repair import inspect_quarantined_accepted_raws 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_application import ( @@ -260,3 +262,226 @@ def test_unified_quarantine_strategy_rejects_mutated_authority_witness(tmp_path: assert item.state is not RawAuthorityFrontierState.SAFELY_REKEYABLE assert _logical_state(tmp_path, raw_id) == before + + +def _seed_quarantined_raw_fanout(root: Path) -> tuple[str, tuple[tuple[str, str], ...]]: + """One quarantined raw accepted as the head by TWO different sessions. + + Mirrors the fan-out shape in ``test_duplicate_raw_identity_repair.py``'s + ``_seed_duplicate_raw_fanout`` (forked/subagent/resumed sessions can + physically replay the identical parent evidence, so the exact same + ``raw_id`` can legitimately be the accepted head of more than one + logical source key/session at once, polylogue-ihc8) -- but for + quarantine-refinement specifically, the raw's own bytes can only ever + re-parse into ONE genuine session's content + (``_inspect_quarantined_accepted_raw`` requires exactly one normalized + session and an exact content-hash match). So unlike duplicate-alias + fan-out, at most ONE sibling here can ever be genuinely eligible; + ``session_a`` is that one (its accepted content-hash/session-id + genuinely match what the raw re-parses to). ``session_b`` represents a + second, stale accepted-head row for the SAME raw_id whose own + identity/content does not match -- exactly the shape observed live + (a physically shared raw whose accepted-head rows for OTHER sessions + can never re-verify against it). Before the fix, inspecting EITHER + session crashed with "expected one accepted head, found 2" regardless + of this distinction; after the fix, session_a is provably eligible and + session_b is provably (and permanently) ineligible via the real + content-mismatch check -- neither crashes the census. + """ + initialize_active_archive_root(root) + records = [_chatgpt_session("fanout-quarantine", "proof text")] + payload = json.dumps(records[0], sort_keys=True).encode() + source_path = "fanout-quarantine.json" + parsed = _parse_one(Provider.CHATGPT, payload, source_path) + assert parsed + session_a = parsed[0] + source_revision = hashlib.sha256(payload).hexdigest() + content_hash_a = bytes.fromhex(session_content_hash(session_a)) + key_a = "chatgpt:fanout-quarantine" + + # Same origin (chatgpt-export) as session_a throughout -- a genuine + # origin mismatch is a DIFFERENT, unrelated repair path (browser-origin + # conflict), not the quarantine-refinement fan-out this test targets. + stale_records = [_chatgpt_session("fanout-quarantine-stale-sibling", "unrelated stale content")] + stale_payload = json.dumps(stale_records[0], sort_keys=True).encode() + stale_parsed = _parse_one(Provider.CHATGPT, stale_payload, "fanout-quarantine-stale.json") + assert stale_parsed + stale_session = stale_parsed[0] + content_hash_b = bytes.fromhex(session_content_hash(stale_session)) + key_b = "chatgpt:fanout-quarantine-stale-sibling" + + with ArchiveStore.open_existing(root, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CHATGPT, payload=payload, source_path=source_path, acquired_at_ms=1 + ) + _raw_id, session_id_a = archive.write_parsed_for_retained_raw( + session_a, raw_id=raw_id, source_path=source_path, acquired_at_ms=1, revision_authoritative=True + ) + assert session_id_a == "chatgpt-export:fanout-quarantine" + # session_b's own row is written to a DISTINCT raw first (it needs a + # real, materialized session/index row to exist), then retargeted + # onto the SAME shared raw_id -- reproducing "this session's own + # accepted head points at a raw it doesn't actually match" without + # needing a second real parseable payload. + stale_raw_id = archive.write_raw_payload( + provider=Provider.CHATGPT, + payload=stale_payload, + source_path="fanout-quarantine-stale.json", + acquired_at_ms=1, + ) + _stale_raw_id, session_id_b = archive.write_parsed_for_retained_raw( + stale_session, + raw_id=stale_raw_id, + source_path="fanout-quarantine-stale.json", + acquired_at_ms=1, + revision_authoritative=True, + ) + for session_id, logical_source_key, content_hash in ( + (session_id_a, key_a, content_hash_a), + (session_id_b, key_b, content_hash_b), + ): + record_revision_application_sync( + archive._conn, + RevisionApplicationReceipt( + raw_id=raw_id, + session_id=session_id, + logical_source_key=logical_source_key, + source_revision=source_revision, + acquisition_generation=0, + decision=ApplicationDecision.SELECTED_BASELINE, + accepted_raw_id=raw_id, + accepted_source_revision=source_revision, + accepted_content_hash=content_hash, + accepted_frontier_kind="byte", + accepted_frontier=len(payload), + baseline_raw_id=raw_id, + detail=f"pre-quarantine fanout fixture ({logical_source_key})", + ), + decided_at_ms=2, + ) + archive.commit() + with sqlite3.connect(root / "index.db") as index: + # Retarget session_b's own materialized session + accepted head onto + # the shared raw_id (it was written against stale_raw_id above only + # to get a real row to retarget). + index.execute("UPDATE sessions SET raw_id = ? WHERE session_id = ?", (raw_id, session_id_b)) + index.execute("UPDATE raw_revision_heads SET accepted_raw_id = ? WHERE session_id = ?", (raw_id, session_id_b)) + index.commit() + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + INSERT INTO raw_session_memberships ( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count, acquisition_generation, + revision_authority + ) VALUES (?, ?, ?, ?, ?, ?, 0, 'quarantined') + """, + (raw_id, key_a, "fanout-quarantine", content_hash_a.hex(), content_hash_a, len(session_a.messages)), + ) + source.execute( + """ + INSERT INTO raw_membership_census ( + raw_id, parser_fingerprint, status, member_count, censused_at_ms + ) VALUES (?, 'revision-membership-v1', 'complete', 1, 0) + """, + (raw_id,), + ) + source.commit() + with sqlite3.connect(root / "source.db") as source: + source.execute( + """ + UPDATE raw_sessions + SET logical_source_key = NULL, revision_kind = 'unknown', source_revision = NULL, + baseline_raw_id = NULL, acquisition_generation = NULL, + revision_authority = 'quarantined' + WHERE raw_id = ? + """, + (raw_id,), + ) + source.commit() + return raw_id, ((session_id_a, key_a), (session_id_b, key_b)) + + +def test_quarantine_refinement_fanout_scopes_by_logical_source_key(tmp_path: Path) -> None: + """polylogue-zaiz regression: fan-out siblings must not share one proof. + + Before the fix, ``_inspect_quarantined_accepted_raw`` looked up the + accepted-head row by ``accepted_raw_id`` alone, requiring exactly one + match system-wide. For this fan-out shape (two sessions sharing one + quarantined raw), that unconditionally raised "expected one accepted + head, found 2" for BOTH sessions -- crashing the whole census before + ever reaching the real, content-based eligibility question. After the + fix, each session gets an independent, correctly-scoped classification: + ``session_a`` (whose accepted content genuinely matches what the raw + re-parses to) is provably eligible; ``session_b`` (a stale sibling + accepted-head row sharing the same raw_id, whose own content does not + match) is provably -- and permanently -- ineligible via the real + content-mismatch check, not a crash. + """ + raw_id, heads = _seed_quarantined_raw_fanout(tmp_path) + (session_a, key_a), (session_b, key_b) = heads + + census = inspect_raw_authority_frontier(_config(tmp_path)) + by_key = {item.logical_source_key: item for item in census.items if item.raw_id == raw_id} + + assert set(by_key) == {key_a, key_b} + assert by_key[key_a].state is RawAuthorityFrontierState.SAFELY_REKEYABLE + assert by_key[key_a].actuator is RawAuthorityActuator.REFINE_QUARANTINE + assert by_key[key_a].executable + + assert by_key[key_b].state is RawAuthorityFrontierState.UNRESOLVED_PROVENANCE + assert by_key[key_b].actuator is RawAuthorityActuator.REFINE_QUARANTINE + assert not by_key[key_b].executable + + # The deep proof itself (not just the census-level fallback reason) + # must show the real content-mismatch cause, not the old scoping crash. + inspected = inspect_quarantined_accepted_raws(_config(tmp_path), [(raw_id, key_b)]) + assert inspected[0].status == "ineligible" + assert "expected one accepted head" not in inspected[0].reason + assert "differs from the accepted session" in inspected[0].reason + + +def test_quarantine_refinement_applies_for_the_matching_sibling_only(tmp_path: Path) -> None: + """polylogue-zaiz regression: the genuinely-matching sibling refines cleanly. + + Unlike duplicate-alias fan-out (where eligibility is a race -- any + sibling looks eligible until a canonical twin is claimed), quarantine- + refinement eligibility is a fixed content-match fact: only + ``session_a`` (whose accepted content genuinely matches the raw) is + ever ``SAFELY_REKEYABLE``. Applying its plan must succeed end-to-end + without disturbing ``session_b``'s own (permanently ineligible, + unaffected) accepted-head row. + """ + raw_id, heads = _seed_quarantined_raw_fanout(tmp_path) + (session_a, key_a), (session_b, key_b) = heads + + preview = inspect_raw_authority_frontier(_config(tmp_path)) + selected = next( + item + for item in preview.items + if item.raw_id == raw_id + and item.logical_source_key == key_a + and item.state is RawAuthorityFrontierState.SAFELY_REKEYABLE + ) + + report = apply_raw_authority_frontier( + _config(tmp_path), + preview_census_id=preview.census_id, + selected_plan_ids=(selected.plan_id,), + ) + + assert report.executed_plan_count == 1 + assert report.retryable_plan_count == 0 + assert report.success + + with sqlite3.connect(tmp_path / "source.db") as source: + assert source.execute( + "SELECT revision_authority, baseline_raw_id FROM raw_sessions WHERE raw_id = ?", + (raw_id,), + ).fetchone() == ("byte_proven", raw_id) + with sqlite3.connect(tmp_path / "index.db") as index: + # session_b's own head still points at the shared raw_id -- the + # refinement is scoped to session_a only, not a global side effect. + assert index.execute( + "SELECT accepted_raw_id FROM raw_revision_heads WHERE session_id = ?", (session_b,) + ).fetchone() == (raw_id,)