fix: converge raw components holding byte-governed append fragments

polylogue/storage/sqlite/archive_tiers/revision_governance.py

@@ -114,6 +114,7 @@
 from polylogue.archive.artifact_taxonomy import ArtifactClassification
 from polylogue.archive.ingest_flags import DOM_FALLBACK_INGEST_FLAG, NATIVE_BROWSER_CAPTURE_FLAGS
 from polylogue.archive.revision_authority import (
+    BYTE_AUTHORITY_CENSUS_DETAIL,
     RAW_AUTHORITY_PARSER_FINGERPRINT,
     RETIRED_FULL_REVISION_GOVERNANCE_DETAILS,
     HistoricalRawRevisionStream,
@@ -1964,7 +1965,8 @@ def record_current_parser_source_census(
     raw = conn.execute(
         """
         SELECT logical_source_key, revision_kind,
-               EXISTS(SELECT 1 FROM raw_artifacts WHERE raw_id = raw_sessions.raw_id AND parse_as_session = 0)
+               EXISTS(SELECT 1 FROM raw_artifacts WHERE raw_id = raw_sessions.raw_id AND parse_as_session = 0),
+               source_index
         FROM raw_sessions WHERE raw_id = ?
         """,
         (raw_id,),
@@ -1991,7 +1993,28 @@ def record_current_parser_source_census(
         membership_logical_keys=membership_keys,
     )
     typed_non_session = bool(raw[2])
-    if typed_non_session:
+    # polylogue-39kcs: an append fragment is never parsed for identity --
+    # ``_persist_revision_census`` routes every ``source_index < 0`` raw
+    # straight to the byte-authority membership receipt because appends are
+    # governed by byte revision authority, not by semantic membership. That
+    # is a COMPLETE observation with an authoritative empty identity set,
+    # exactly like a typed non-session artifact, so it must be receipted as
+    # such. Recording it as ``failed`` instead left the fragment matching
+    # neither branch of ``uncensused_historical_revision_raw_ids``'s gate
+    # (complete + ``parser-observed:%``, or ``failed`` at the current
+    # resource-blocked fingerprint), so every pass re-censused it, rewrote
+    # the same receipt, and left raw-replay planning "paused until the
+    # persisted parser census completes" forever -- taking every ``full``
+    # snapshot in the same authority component down with it (the live
+    # codex 019f49d8 rollout: 767 fragments, 20 parsed fulls, ~20k messages
+    # acquired and censused but never materialized).
+    byte_governed_fragment = (
+        int(raw[3]) < 0
+        and membership_census is not None
+        and str(membership_census[0]) == "failed"
+        and str(membership_census[1]) == BYTE_AUTHORITY_CENSUS_DETAIL
+    )
+    if typed_non_session or byte_governed_fragment:
         # 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.
@@ -2009,7 +2032,7 @@ def record_current_parser_source_census(
         else tuple(sorted(canonical_authority_logical_key(key) for key in inherited_logical_keys or ()))
         if inherited_logical_keys is not None
         else ()
-        if typed_non_session
+        if typed_non_session or byte_governed_fragment
         else None
     )
     complete = (
@@ -2019,11 +2042,14 @@ def record_current_parser_source_census(
         and (
             bool(observed_keys)
             or typed_non_session
+            or byte_governed_fragment
             or (membership_census is not None and str(membership_census[0]) == "non_session")
         )
     )
     detail = (
-        "parser-observed: typed non-session admission established no parser identity"
+        "parser-observed: append fragment governed by byte revision authority"
+        if byte_governed_fragment and complete
+        else "parser-observed: typed non-session admission established no parser identity"
         if typed_non_session and complete
         else "parser-observed: membership census established durable authority identity"
         if membership_census is not None and complete

tests/unit/sources/test_revision_backfill.py

@@ -1470,7 +1470,15 @@ def test_historical_backfill_selects_prefix_newest_independent_of_acquisition_or
             """,
             (RAW_AUTHORITY_PARSER_FINGERPRINT,),
         ).fetchall()
-    assert parser_census == [("complete", 2), ("failed", 1)]
+    # polylogue-39kcs: all three raws are census-complete, including the
+    # legacy append fragment. Its receipt records the authoritative empty
+    # identity set that byte revision governance gives it -- a ``failed``
+    # receipt there matched neither branch of
+    # ``uncensused_historical_revision_raw_ids``'s gate, so the fragment was
+    # re-selected for census forever and raw-replay planning never started.
+    # The fragment is still ``quarantined`` (asserted above); only the
+    # census receipt changed, not its authority.
+    assert parser_census == [("complete", 3)]
     with sqlite3.connect(tmp_path / "index.db") as conn:
         assert conn.execute("SELECT message_count, raw_id FROM sessions").fetchone() == (2, newest_raw_id)
 

tests/unit/storage/test_repair.py

@@ -2330,8 +2330,16 @@ def test_raw_materialization_reports_uncensused_append_fragments_as_pending_debt
     assert backlog["byte_authority_pending_count"] == 1
     assert targeted.success is False
     assert targeted.census_receipt is not None
-    assert targeted.census_receipt.quiescent is False
-    assert "persisted parser census" in targeted.detail
+    # polylogue-39kcs: the census IS quiescent here -- the current parser has
+    # fully observed this fragment and established that byte revision
+    # authority, not semantic membership, governs it. Reporting non-quiescence
+    # instead made the pass claim it was "paused until the persisted parser
+    # census completes" for a raw the census can never say anything more
+    # about, which livelocked every ``full`` snapshot sharing the component.
+    # The fragment is still debt -- just typed as the authority quarantine it
+    # actually is, rather than as unfinished census work.
+    assert targeted.census_receipt.quiescent is True
+    assert "append authority quarantine" in targeted.detail
 
     with sqlite3.connect(tmp_path / "source.db") as source_conn:
         cursor = source_conn.execute(
@@ -5042,3 +5050,115 @@ def counting_commit(self: ArchiveStore) -> None:
     assert coarse_count >= 1
     assert fine_count > coarse_count
     assert fine_count <= raw_count
+
+
+def test_raw_materialization_converges_component_with_byte_governed_append_fragment(
+    tmp_path: Path,
+) -> None:
+    """polylogue-39kcs: a byte-governed append fragment must not pause planning forever.
+
+    Codex rollout captures grow in place, so one logical source accumulates
+    both ``full`` snapshots and ``append`` fragments. Append fragments are
+    deliberately never parsed for identity -- ``_persist_revision_census``
+    routes every ``source_index < 0`` raw straight to the byte-authority
+    membership receipt (``BYTE_AUTHORITY_CENSUS_DETAIL``) instead. That in
+    turn makes ``record_raw_authority_parser_census`` write a ``failed``
+    parser receipt carrying the same detail, which satisfies neither branch
+    of ``uncensused_historical_revision_raw_ids``'s census-complete gate
+    (``parser-observed:%`` complete, or a resource-blocked ``failed``
+    receipt at the current envelope). The fragment is therefore reported
+    uncensused on every pass, planning stays "paused until the persisted
+    parser census completes", and every ``full`` snapshot sharing the
+    component is never replayed -- the live 019f49d8 rollout
+    (767 append fragments, 20 cleanly parsed fulls, ~20k messages) has sat
+    unmaterialized in exactly this state.
+
+    Drives the real ``repair_raw_materialization`` entry point, not the
+    census helper directly, because the livelock is a property of the pass's
+    census/planning handshake rather than of either half alone.
+    """
+    from polylogue.archive.revision_authority import (
+        RawRevisionAuthority,
+        RawRevisionEnvelope,
+        RawRevisionKind,
+    )
+    from polylogue.sources.revision_backfill import uncensused_historical_revision_raw_ids
+    from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
+
+    key = "codex:growing-rollout"
+    baseline = _codex_conversation_bytes("growing-rollout")
+    grown = baseline + (
+        b'{"type":"response_item","payload":{"type":"message","id":"m-second",'
+        b'"role":"assistant","content":[{"type":"output_text","text":"tail"}]}}\n'
+    )
+    tail = grown[len(baseline) :]
+
+    initialize_active_archive_root(tmp_path)
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as store:
+        full_raw_ids = []
+        for index, payload in enumerate((baseline, grown)):
+            raw_id = store.write_raw_payload(
+                provider=Provider.CODEX,
+                payload=payload,
+                source_path="rollout.jsonl",
+                acquired_at_ms=index + 1,
+            )
+            store.bind_raw_revision(
+                raw_id,
+                RawRevisionEnvelope(
+                    key,
+                    RawRevisionKind.FULL,
+                    raw_id,
+                    0,
+                    authority=RawRevisionAuthority.QUARANTINED,
+                ),
+            )
+            full_raw_ids.append(raw_id)
+        # An orphan append fragment: byte-governed, never parsed, and (like
+        # the live 019f49d8 fragments) chained to a predecessor revision that
+        # has no full row of its own, so it can never be promoted.
+        append_raw_id = store.write_raw_payload(
+            provider=Provider.CODEX,
+            payload=tail,
+            source_path="rollout.jsonl",
+            source_index=-1,
+            acquired_at_ms=3,
+        )
+        store.bind_raw_revision(
+            append_raw_id,
+            RawRevisionEnvelope(
+                key,
+                RawRevisionKind.APPEND,
+                append_raw_id,
+                0,
+                authority=RawRevisionAuthority.QUARANTINED,
+                predecessor_source_revision="0" * 64,
+                append_start_offset=len(baseline),
+                append_end_offset=len(grown),
+            ),
+        )
+        store.commit()
+
+    config = _config(tmp_path)
+    result = repair_mod.repair_raw_materialization(config)
+    for _ in range(2):
+        if result.repaired_count:
+            break
+        result = repair_mod.repair_raw_materialization(config)
+
+    with sqlite3.connect(tmp_path / "source.db") as conn:
+        append_receipt = conn.execute(
+            "SELECT status, detail FROM raw_authority_parser_census WHERE raw_id = ?",
+            (append_raw_id,),
+        ).fetchone()
+
+    # The fragment's parser receipt must be a terminal, census-complete
+    # answer -- not a 'failed' row the census gate re-selects forever.
+    assert append_receipt is not None
+    assert uncensused_historical_revision_raw_ids(tmp_path, [append_raw_id]) == ()
+
+    with sqlite3.connect(tmp_path / "index.db") as conn:
+        sessions = conn.execute("SELECT session_id FROM sessions").fetchall()
+    assert sessions == [("codex-session:growing-rollout",)], (
+        f"component never materialized; append receipt={append_receipt!r} detail={result.detail!r}"
+    )