diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 3d96ecfd79..7836cb4e67 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -2164,6 +2164,14 @@ def _promote_contiguous_append_evidence(conn: sqlite3.Connection, logical_source if not changed: return + def _raw_revision_authority(self, raw_id: str) -> str | None: + row = ( + self._ensure_source_conn() + .execute("SELECT revision_authority FROM raw_sessions WHERE raw_id = ?", (raw_id,)) + .fetchone() + ) + return None if row is None or row[0] is None else str(row[0]) + def raw_revision_replay_plan(self, logical_source_key: str) -> RevisionReplayPlan: return plan_revision_replay(self._raw_revision_candidates(logical_source_key)) @@ -3044,6 +3052,29 @@ def apply_raw_revision_replay( ) else: accepted_frontier = None + if ( + existing_head is not None + and accepted_frontier_kind == "semantic" + and accepted_frontier is not None + and str(existing_head[4]) == "semantic" + and self._raw_revision_authority(str(existing_head[1])) == "quarantined" + ): + # The current head was written by membership replay for a + # quarantined-authority raw (e.g. a browser-capture snapshot of + # the same provider conversation). Chain evidence with + # source-tier revision governance outranks quarantined capture + # evidence unconditionally: a scalar semantic frontier cannot + # prove the capture is a content-superset (a divergent capture + # with more units is not "ahead"), so no count comparison is + # attempted. The capture raw stays in the source tier and its + # earlier receipts remain; re-adopting genuinely + # content-ahead capture tails needs a real prefix-dominance + # proof (follow-up bead), not a unit count. + self._conn.execute( + "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", + (plan.logical_source_key,), + ) + existing_head = None for position, raw_id in enumerate(plan.accepted_raw_ids): index_started = time.perf_counter() result = self._index_parsed_for_retained_raw( @@ -3192,106 +3223,168 @@ def apply_raw_membership_classification( with self._conn if manage_transaction else nullcontext(): existing_head = self._conn.execute( """ - SELECT accepted_raw_id, accepted_content_hash, accepted_frontier_kind, session_id + SELECT accepted_raw_id, accepted_content_hash, accepted_frontier_kind, session_id, + accepted_frontier FROM raw_revision_heads WHERE logical_source_key = ? """, (logical_source_key,), ).fetchone() + yield_to_head_raw_id: str | None = None if existing_head is not None: existing_raw_id = str(existing_head[0]) classified_raw_ids = { *classification.accepted_raw_ids, *classification.equivalent_raw_ids, } - persisted_session = self._conn.execute( - "SELECT raw_id, content_hash FROM sessions WHERE session_id = ?", - (str(existing_head[3]),), - ).fetchone() - if ( - existing_raw_id not in classified_raw_ids - or persisted_session is None - or str(persisted_session[0]) != existing_raw_id - or not isinstance(persisted_session[1], bytes) - or bytes(existing_head[1]) != bytes(persisted_session[1]) + chain_head_authority = ( + self._raw_revision_authority(existing_raw_id) + if existing_raw_id not in classified_raw_ids + else None + ) + if existing_raw_id not in classified_raw_ids and chain_head_authority not in ( + None, + "quarantined", ): - raise RuntimeError("membership replay cannot retire an unrelated accepted head") - existing_is_byte_governed = conn.execute( - "SELECT 1 FROM raw_sessions WHERE raw_id = ? AND logical_source_key = ?", - (existing_raw_id, logical_source_key), - ).fetchone() - if existing_is_byte_governed is not None and accepted_raw_id != existing_raw_id: - raise RuntimeError("membership replay cannot replace an unconvertible byte head") - self._conn.execute( - "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", - (logical_source_key,), + # The head is owned by chain-governed (non-quarantined) + # source evidence outside this quarantined membership + # cohort -- e.g. a byte-proven provider export of the + # same conversation this browser capture snapshotted. + # Provider-export evidence outranks quarantined capture + # evidence unconditionally: a scalar semantic frontier + # cannot prove the capture is a content-superset, so + # the cohort always yields (receipted below); the + # capture bytes stay in the source tier. Re-adopting a + # genuinely content-ahead capture tail needs a real + # prefix-dominance proof (follow-up bead). + yield_to_head_raw_id = existing_raw_id + else: + persisted_session = self._conn.execute( + "SELECT raw_id, content_hash FROM sessions WHERE session_id = ?", + (str(existing_head[3]),), + ).fetchone() + if ( + existing_raw_id not in classified_raw_ids + or persisted_session is None + or str(persisted_session[0]) != existing_raw_id + or not isinstance(persisted_session[1], bytes) + or bytes(existing_head[1]) != bytes(persisted_session[1]) + ): + raise RuntimeError( + "membership replay cannot retire an unrelated accepted head: " + f"logical_source_key={logical_source_key!r} " + f"existing_head(raw_id={existing_raw_id!r}, session_id={str(existing_head[3])!r}, " + f"authority={chain_head_authority!r}) " + f"cohort(accepted={classification.accepted_raw_ids!r}, " + f"equivalent={classification.equivalent_raw_ids!r}, " + f"ambiguous={classification.ambiguous_raw_ids!r}) " + f"persisted_session_raw={None if persisted_session is None else str(persisted_session[0])!r}" + ) + existing_is_byte_governed = conn.execute( + "SELECT 1 FROM raw_sessions WHERE raw_id = ? AND logical_source_key = ?", + (existing_raw_id, logical_source_key), + ).fetchone() + if existing_is_byte_governed is not None and accepted_raw_id != existing_raw_id: + raise RuntimeError("membership replay cannot replace an unconvertible byte head") + self._conn.execute( + "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", + (logical_source_key,), + ) + if yield_to_head_raw_id is not None: + assert existing_head is not None + session_id = str(existing_head[3]) + cohort_raw_ids = ( + *classification.accepted_raw_ids, + *classification.equivalent_raw_ids, + *classification.ambiguous_raw_ids, ) - index_started = time.perf_counter() - result = self._index_parsed_for_retained_raw( - accepted_session, - raw_id=accepted_raw_id, - source_index=0, - stage_timings_s=stage_timings_s, - stage_timing_prefix=stage_timing_prefix, - manage_transaction=False, - preacquired_attachment_blobs=attachments, - finalize_raw_parse=False, - revision_authoritative=True, - bulk_fts=bulk_fts, - bulk_build=bulk_build, - ) - if stage_timings_s is not None: - key = f"{stage_timing_prefix}.index_parsed_write" - stage_timings_s[key] = stage_timings_s.get(key, 0.0) + (time.perf_counter() - index_started) - session_id = result.session_id - if not bulk_build: - repair_message_fts_index_sync(self._conn, [session_id], record_exact_snapshot=False) - assert_session_fts_exact_sync(self._conn, session_id, bulk_build=bulk_build) - stored = self._conn.execute( - "SELECT content_hash FROM sessions WHERE session_id = ?", (session_id,) - ).fetchone() - if stored is None or not isinstance(stored[0], bytes): - raise RuntimeError("accepted membership did not produce a hashed session") - accepted_projection = projections_by_raw_id[accepted_raw_id] - semantic_frontier = ( - len(accepted_projection.message_hashes) - + len(accepted_projection.event_hashes) - + len(accepted_projection.attachment_hashes) - ) - cohort_raw_ids = ( - *classification.accepted_raw_ids, - *classification.equivalent_raw_ids, - *classification.ambiguous_raw_ids, - ) - for generation, raw_id in enumerate(cohort_raw_ids): - projection = projections_by_raw_id[raw_id] - decision = decisions.get(raw_id, "applied") - record_revision_application_sync( - self._conn, - RevisionApplicationReceipt( - raw_id=raw_id, - session_id=session_id, - logical_source_key=logical_source_key, - source_revision=projection.session_hash.hex(), - acquisition_generation=generation, - decision=( - ApplicationDecision.AMBIGUOUS - if decision == "ambiguous" - else ApplicationDecision.SUPERSEDED - if decision.startswith("superseded") - else ApplicationDecision.SELECTED_BASELINE - ), - accepted_raw_id=accepted_raw_id if decision != "ambiguous" else None, - accepted_source_revision=( - accepted_projection.session_hash.hex() if decision != "ambiguous" else None + for generation, raw_id in enumerate(cohort_raw_ids): + projection = projections_by_raw_id[raw_id] + decisions[raw_id] = "superseded_equivalent" + record_revision_application_sync( + self._conn, + RevisionApplicationReceipt( + raw_id=raw_id, + session_id=session_id, + logical_source_key=logical_source_key, + source_revision=projection.session_hash.hex(), + acquisition_generation=generation, + decision=ApplicationDecision.SUPERSEDED, + accepted_raw_id=None, + accepted_source_revision=None, + accepted_content_hash=None, + detail=f"membership:superseded_by_chain_governed_head:{yield_to_head_raw_id}", ), - accepted_content_hash=stored[0] if decision != "ambiguous" else None, - accepted_frontier_kind="semantic" if decision != "ambiguous" else None, - accepted_frontier=semantic_frontier if decision != "ambiguous" else None, - detail=f"membership:{decision}", - ), - decided_at_ms=decided_at_ms, + decided_at_ms=decided_at_ms, + ) + else: + index_started = time.perf_counter() + result = self._index_parsed_for_retained_raw( + accepted_session, + raw_id=accepted_raw_id, + source_index=0, + stage_timings_s=stage_timings_s, + stage_timing_prefix=stage_timing_prefix, + manage_transaction=False, + preacquired_attachment_blobs=attachments, + finalize_raw_parse=False, + revision_authoritative=True, + bulk_fts=bulk_fts, + bulk_build=bulk_build, + ) + if stage_timings_s is not None: + key = f"{stage_timing_prefix}.index_parsed_write" + stage_timings_s[key] = stage_timings_s.get(key, 0.0) + (time.perf_counter() - index_started) + session_id = result.session_id + if not bulk_build: + repair_message_fts_index_sync(self._conn, [session_id], record_exact_snapshot=False) + assert_session_fts_exact_sync(self._conn, session_id, bulk_build=bulk_build) + stored = self._conn.execute( + "SELECT content_hash FROM sessions WHERE session_id = ?", (session_id,) + ).fetchone() + if stored is None or not isinstance(stored[0], bytes): + raise RuntimeError("accepted membership did not produce a hashed session") + accepted_projection = projections_by_raw_id[accepted_raw_id] + semantic_frontier = ( + len(accepted_projection.message_hashes) + + len(accepted_projection.event_hashes) + + len(accepted_projection.attachment_hashes) + ) + cohort_raw_ids = ( + *classification.accepted_raw_ids, + *classification.equivalent_raw_ids, + *classification.ambiguous_raw_ids, ) - decisions[accepted_raw_id] = "applied" + for generation, raw_id in enumerate(cohort_raw_ids): + projection = projections_by_raw_id[raw_id] + decision = decisions.get(raw_id, "applied") + record_revision_application_sync( + self._conn, + RevisionApplicationReceipt( + raw_id=raw_id, + session_id=session_id, + logical_source_key=logical_source_key, + source_revision=projection.session_hash.hex(), + acquisition_generation=generation, + decision=( + ApplicationDecision.AMBIGUOUS + if decision == "ambiguous" + else ApplicationDecision.SUPERSEDED + if decision.startswith("superseded") + else ApplicationDecision.SELECTED_BASELINE + ), + accepted_raw_id=accepted_raw_id if decision != "ambiguous" else None, + accepted_source_revision=( + accepted_projection.session_hash.hex() if decision != "ambiguous" else None + ), + accepted_content_hash=stored[0] if decision != "ambiguous" else None, + accepted_frontier_kind="semantic" if decision != "ambiguous" else None, + accepted_frontier=semantic_frontier if decision != "ambiguous" else None, + detail=f"membership:{decision}", + ), + decided_at_ms=decided_at_ms, + ) + if yield_to_head_raw_id is None: + decisions[accepted_raw_id] = "applied" with conn if manage_transaction else nullcontext(): for raw_id, decision in decisions.items(): diff --git a/tests/unit/storage/test_revision_replay.py b/tests/unit/storage/test_revision_replay.py index 1630edf0a1..0050e94e6e 100644 --- a/tests/unit/storage/test_revision_replay.py +++ b/tests/unit/storage/test_revision_replay.py @@ -753,3 +753,224 @@ def durable_index_state(archive: ArchiveStore) -> tuple[object, ...]: assert archive._ensure_source_conn().execute( "SELECT parsed_at_ms FROM raw_sessions WHERE raw_id = ?", (rejected_raw,) ).fetchone() == (None,) + + +def _parsed_session(*messages: tuple[str, str]) -> ParsedSession: + return ParsedSession( + source_name=Provider.CODEX, + provider_session_id="session", + messages=[ + ParsedMessage(provider_message_id=message_id, role=Role.USER, text=text) for message_id, text in messages + ], + ) + + +def _write_quarantined_member(archive: ArchiveStore, label: str, session: ParsedSession) -> str: + """A capture-style raw: no revision envelope, default quarantined authority.""" + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=label.encode(), + source_path=f"{label}.json", + acquired_at_ms=1, + ) + archive.replace_raw_membership_census( + raw_id, + [session], + parser_fingerprint="test-parser", + censused_at_ms=1, + ) + return raw_id + + +def _write_chain_full(archive: ArchiveStore, label: str, generation: int) -> str: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=label.encode(), + source_path="session.json", + acquired_at_ms=generation, + ) + archive.bind_raw_revision( + raw_id, + RawRevisionEnvelope( + "codex:session", + RawRevisionKind.FULL, + f"revision-{label}", + generation, + authority=RawRevisionAuthority.BYTE_PROVEN, + ), + ) + return raw_id + + +def _apply_membership_head(archive: ArchiveStore, raw_id: str, session: ParsedSession) -> None: + archive.apply_raw_membership_classification( + "codex:session", + MembershipClassification((raw_id,), (), ()), + {raw_id: session}, + {raw_id: session_revision_projection(session)}, + acquired_at_ms=0, + ) + + +def _head_row(archive: ArchiveStore) -> tuple[object, ...] | None: + row = archive._conn.execute( + """SELECT accepted_raw_id, accepted_frontier_kind, accepted_frontier + FROM raw_revision_heads WHERE logical_source_key = 'codex:session'""" + ).fetchone() + return None if row is None else tuple(row) + + +def test_chain_replay_supersedes_equal_frontier_quarantined_membership_head(tmp_path: Path) -> None: + """Capture-vs-export head collision (the v42 rebuild crash): a byte-proven + + chain full at an EQUAL semantic frontier with different content must take + the head from a quarantined membership (browser-capture) raw instead of + the CAS rejecting the whole replay. + """ + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + capture_session = _parsed_session(("m0", "zero"), ("m1", "capture flavour")) + capture = _write_quarantined_member(archive, "capture", capture_session) + _apply_membership_head(archive, capture, capture_session) + assert _head_row(archive) == (capture, "semantic", 2) + + export_session = _parsed_session(("m0", "zero"), ("m1", "export flavour")) + export = _write_chain_full(archive, "export", 2) + plan = plan_revision_replay([_candidate(export, RawRevisionKind.FULL, 2, size=len("export"))]) + session_id, applied = archive.apply_raw_revision_replay(plan, {export: export_session}, acquired_at_ms=0) + + assert applied == (export,) + assert _head_row(archive) == (export, "semantic", 2) + stored = archive._conn.execute( + "SELECT content_hash FROM sessions WHERE session_id = ?", (session_id,) + ).fetchone() + assert stored is not None + assert bytes(stored[0]).hex() == session_content_hash(export_session) + + +def test_chain_replay_supersedes_quarantined_membership_head_even_when_capture_has_more_units(tmp_path: Path) -> None: + """Chain evidence wins unconditionally: a scalar frontier cannot prove a + + capture is a content-superset, so even a capture with MORE semantic units + hands the head to chain-governed evidence (the capture raw stays in the + source tier; re-adoption needs a real prefix-dominance proof). + """ + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + capture_session = _parsed_session(("m0", "zero"), ("m1", "one"), ("m2", "two")) + capture = _write_quarantined_member(archive, "capture", capture_session) + _apply_membership_head(archive, capture, capture_session) + assert _head_row(archive) == (capture, "semantic", 3) + + export_session = _parsed_session(("m0", "zero"), ("m1", "one")) + export = _write_chain_full(archive, "export", 2) + plan = plan_revision_replay([_candidate(export, RawRevisionKind.FULL, 2, size=len("export"))]) + session_id, applied = archive.apply_raw_revision_replay(plan, {export: export_session}, acquired_at_ms=0) + + assert applied == (export,) + assert _head_row(archive) == (export, "semantic", 2) + stored = archive._conn.execute( + "SELECT content_hash FROM sessions WHERE session_id = ?", (session_id,) + ).fetchone() + assert stored is not None + assert bytes(stored[0]).hex() == session_content_hash(export_session) + + +def test_membership_replay_yields_to_chain_governed_head(tmp_path: Path) -> None: + """Reverse arrival order: the chain head exists first; an equal-frontier + + quarantined capture cohort must yield (superseded receipts, memberships + terminally decided) instead of raising 'cannot retire an unrelated + accepted head'. + """ + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + export_session = _parsed_session(("m0", "zero"), ("m1", "export flavour")) + export = _write_chain_full(archive, "export", 1) + plan = plan_revision_replay([_candidate(export, RawRevisionKind.FULL, 1, size=len("export"))]) + archive.apply_raw_revision_replay(plan, {export: export_session}, acquired_at_ms=0) + # A chain-first head is byte-kind: its frontier is never comparable to + # a capture's semantic frontier, so the capture must always yield. + assert _head_row(archive) == (export, "byte", 6) + + capture_session = _parsed_session(("m0", "zero"), ("m1", "capture flavour")) + capture = _write_quarantined_member(archive, "capture", capture_session) + result = archive.apply_raw_membership_classification( + "codex:session", + MembershipClassification((capture,), (), ()), + {capture: capture_session}, + {capture: session_revision_projection(capture_session)}, + acquired_at_ms=0, + ) + + assert result == "codex-session:session" + assert _head_row(archive) == (export, "byte", 6) + receipts = archive._conn.execute( + """SELECT decision, detail FROM raw_revision_applications + WHERE raw_id = ? AND logical_source_key = 'codex:session'""", + (capture,), + ).fetchall() + assert [str(row[0]) for row in receipts] == ["superseded"] + assert f"superseded_by_chain_governed_head:{export}" in str(receipts[0][1]) + membership = ( + archive._ensure_source_conn() + .execute( + "SELECT decision, revision_authority FROM raw_session_memberships WHERE raw_id = ?", + (capture,), + ) + .fetchone() + ) + assert membership is not None and tuple(membership) == ("superseded_equivalent", "byte_proven") + stored = archive._conn.execute( + "SELECT content_hash FROM sessions WHERE session_id = 'codex-session:session'" + ).fetchone() + assert stored is not None + assert bytes(stored[0]).hex() == session_content_hash(export_session) + + +def test_membership_replay_yields_to_semantic_chain_head_even_when_capture_has_more_units(tmp_path: Path) -> None: + """A capture cohort with more semantic units still yields to a + chain-governed semantic head: unit counts are not a dominance proof.""" + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + capture1_session = _parsed_session(("m0", "zero")) + capture1 = _write_quarantined_member(archive, "capture1", capture1_session) + _apply_membership_head(archive, capture1, capture1_session) + export_session = _parsed_session(("m0", "zero")) + export = _write_chain_full(archive, "export", 2) + plan = plan_revision_replay([_candidate(export, RawRevisionKind.FULL, 2, size=len("export"))]) + archive.apply_raw_revision_replay(plan, {export: export_session}, acquired_at_ms=0) + assert _head_row(archive) == (export, "semantic", 1) + + capture2_session = _parsed_session(("m0", "zero"), ("m1", "the conversation continued")) + capture2 = _write_quarantined_member(archive, "capture2", capture2_session) + revisions = [ + MembershipRevision(capture1, session_revision_projection(capture1_session)), + MembershipRevision(capture2, session_revision_projection(capture2_session)), + ] + classification = classify_membership_revisions(revisions) + assert capture2 in classification.accepted_raw_ids + archive.apply_raw_membership_classification( + "codex:session", + classification, + {capture1: capture1_session, capture2: capture2_session}, + { + capture1: session_revision_projection(capture1_session), + capture2: session_revision_projection(capture2_session), + }, + acquired_at_ms=0, + ) + + assert _head_row(archive) == (export, "semantic", 1) + receipts = archive._conn.execute( + """SELECT decision, detail FROM raw_revision_applications + WHERE raw_id = ? AND logical_source_key = 'codex:session'""", + (capture2,), + ).fetchall() + assert [str(row[0]) for row in receipts] == ["superseded"] + assert f"superseded_by_chain_governed_head:{export}" in str(receipts[0][1]) + stored = archive._conn.execute( + "SELECT content_hash FROM sessions WHERE session_id = 'codex-session:session'" + ).fetchone() + assert stored is not None + assert bytes(stored[0]).hex() == session_content_hash(export_session)