fix(tests): repair storage-cluster test drift from origin-vocab+clock-guard fixes

polylogue/sources/parsers/chatgpt.py

@@ -11,7 +11,7 @@
 from polylogue.archive.message.artifacts import classify_material_origin, classify_text_message_type
 from polylogue.archive.message.roles import Role
 from polylogue.archive.message.types import MessageType
-from polylogue.core.enums import BlockType, Provider, SessionKind, WebConstructType
+from polylogue.core.enums import BlockType, Provider, SessionKind, TitleSource, WebConstructType
 from polylogue.core.timestamps import parse_timestamp
 from polylogue.sources.providers.chatgpt_session_models import ChatGPTNode
 
@@ -1282,7 +1282,18 @@ def parse(payload: Mapping[str, object], fallback_id: str) -> ParsedSession:
     ]
     session_events.extend(_block_metadata_evidence_events(messages))
     duration_values = [message.duration_ms for message in messages if message.duration_ms is not None]
-    title = payload.get("title") or payload.get("name") or fallback_id
+    provider_title = payload.get("title") or payload.get("name")
+    title = provider_title or fallback_id
+    # polylogue-cijx.4 decision 3 / has_real_title (archive_tiers/archive.py):
+    # title_source is the sole gate distinguishing a genuine provider title
+    # from the bare native-id fallback this parser stores in `title` when
+    # ChatGPT's own export carries neither `title` nor `name` -- without it,
+    # every ChatGPT session (titled or not) silently degraded to the
+    # structural "N msgs" label once #3421 made that gate strict. Only a
+    # real payload title counts as ORIGIN evidence; the id fallback is
+    # exactly the "worse than the UUID it replaces" case that gate exists
+    # to catch.
+    title_source = TitleSource.ORIGIN if provider_title else None
     conv_id = payload.get("id") or payload.get("uuid") or payload.get("conversation_id")
     ingest_flags: list[str] = []
     if not messages and payload.get("conversation_id") and payload.get("id") and "mapping" not in payload:
@@ -1301,6 +1312,7 @@ def parse(payload: Mapping[str, object], fallback_id: str) -> ParsedSession:
         source_name=Provider.CHATGPT,
         provider_session_id=str(conv_id or fallback_id),
         title=str(title),
+        title_source=title_source,
         session_kind=session_kind,
         provider_project_ref=provider_project_ref,
         created_at=str(payload.get("create_time")) if payload.get("create_time") is not None else None,

polylogue/storage/repair.py

@@ -5681,24 +5681,40 @@ def _empty_session_debris_session_ids(conn: sqlite3.Connection) -> list[str]:
 
     Shared by ``count_empty_sessions_sync`` and ``repair_empty_sessions`` so
     the reported debt and the deleted rows can never diverge (polylogue-ne6k).
+
+    Both ``_empty_session_candidate_ids`` and
+    ``_raw_artifact_positively_fails_classification`` access rows by column
+    name, so this sets ``row_factory = sqlite3.Row`` defensively on entry
+    (restoring the caller's original factory on exit) rather than assuming
+    every caller-supplied connection already has it -- ``count_empty_sessions_sync``'s
+    own docstring documents that it is "called with a caller-supplied,
+    possibly read-only, connection" (polylogue-9rdky: the maintenance
+    planner's preview/execute path opens a plain tuple-row connection via
+    ``open_readonly_connection``, which crashed both helpers with
+    ``TypeError: tuple indices must be integers or slices, not str``).
     """
-    candidates = _empty_session_candidate_ids(conn)
-    if not candidates:
-        return []
-    source_db = _sibling_source_db_path(conn)
-    if source_db is None or not source_db.exists():
-        # No source tier reachable -> no way to obtain positive evidence for
-        # any candidate -> retain all of them.
-        return []
-    conn.execute("ATTACH DATABASE ? AS source", (str(source_db),))
+    original_row_factory = conn.row_factory
+    conn.row_factory = sqlite3.Row
     try:
-        return [
-            session_id
-            for session_id, raw_id in candidates
-            if _raw_artifact_positively_fails_classification(conn, raw_id)
-        ]
+        candidates = _empty_session_candidate_ids(conn)
+        if not candidates:
+            return []
+        source_db = _sibling_source_db_path(conn)
+        if source_db is None or not source_db.exists():
+            # No source tier reachable -> no way to obtain positive evidence for
+            # any candidate -> retain all of them.
+            return []
+        conn.execute("ATTACH DATABASE ? AS source", (str(source_db),))
+        try:
+            return [
+                session_id
+                for session_id, raw_id in candidates
+                if _raw_artifact_positively_fails_classification(conn, raw_id)
+            ]
+        finally:
+            conn.execute("DETACH DATABASE source")
     finally:
-        conn.execute("DETACH DATABASE source")
+        conn.row_factory = original_row_factory
 
 
 def repair_empty_sessions(config: Config, dry_run: bool = False) -> RepairResult:

polylogue/storage/sqlite/queries/raw_writes.py

@@ -45,8 +45,8 @@ async def save_raw_session(
             validated_at_ms, validation_status, validation_error, validation_drift_count,
             validation_mode, detection_warnings_json, logical_source_key, revision_kind,
             source_revision, predecessor_source_revision, predecessor_raw_id, baseline_raw_id, append_start_offset,
-            append_end_offset, acquisition_generation, revision_authority
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            append_end_offset, acquisition_generation, revision_authority, revision_authority_evidence
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """,
         (
             record.raw_id,
@@ -77,6 +77,16 @@ async def save_raw_session(
             record.revision.append_end_offset if record.revision else None,
             record.revision.acquisition_generation if record.revision else None,
             record.revision.authority.value if record.revision else "quarantined",
+            # revision_authority_evidence (migration 017) is never computed at
+            # initial-write time -- it is only ever populated later by a
+            # dedicated, explicitly operator-invoked maintenance actuator
+            # (raw_live_source_reconciliation_apply.py /
+            # raw_append_chain_backfill_apply.py) re-verifying the raw
+            # against still-present live source bytes. This is `INSERT OR
+            # IGNORE`, so binding NULL here for a brand-new row is correct
+            # and a duplicate-key insert attempt never overwrites an
+            # already-recorded verification verdict.
+            None,
         ),
     )
     inserted = bool(cursor.rowcount > 0)

tests/infra/storage_records.py

@@ -13,7 +13,15 @@
 
 from polylogue.archive.message.roles import Role
 from polylogue.archive.session.branch_type import BranchType
-from polylogue.core.enums import BlockType, Origin, Provider, SemanticBlockType, ValidationMode, ValidationStatus
+from polylogue.core.enums import (
+    BlockType,
+    Origin,
+    Provider,
+    SemanticBlockType,
+    TitleSource,
+    ValidationMode,
+    ValidationStatus,
+)
 from polylogue.core.json import dumps, loads, require_json_document, require_json_value
 from polylogue.core.sources import origin_from_provider, provider_from_origin
 from polylogue.core.timestamps import _timestamp_sort_key
@@ -937,6 +945,16 @@ def _blocks(message: MessageRecord) -> list[ParsedContentBlock]:
         source_name=provider_from_origin(session.origin),
         provider_session_id=session.native_id,
         title=session.title,
+        # A real parser that sets a title always sets title_source alongside
+        # it (assembly_codex.py, assembly_gemini.py, etc.) -- write.py's
+        # session upsert treats title_source as the sole gate for "is this a
+        # real title" (archive_tiers/archive.py's has_real_title check,
+        # polylogue-cijx.4 decision 3), so a builder-set title with no
+        # title_source silently degrades to the structural "N msgs" fallback
+        # at read time. Mirror real-parser provenance here rather than
+        # leaving every test-built session's explicit title invisible to
+        # that gate.
+        title_source=TitleSource.ORIGIN if session.title else None,
         created_at=session.created_at,
         updated_at=session.updated_at,
         messages=parsed_messages,
@@ -1423,3 +1441,75 @@ def create_session(
 
         builder.save()
         return cid
+
+    def mark_as_phantom_debris(self, native_id: str, *, provider: str = "test") -> str:
+        """Attach an ``agent-*.meta.json``-shaped phantom raw artifact to an
+        already-created session and link it via ``sessions.raw_id``.
+
+        ``count_empty_sessions_sync``/``repair_empty_sessions``
+        (``polylogue/storage/repair.py``) only ever count a message-less (or
+        all-zero-word) session as debris when its raw artifact *positively
+        fails* the current record-shape classifier
+        (``_raw_artifact_positively_fails_classification``) -- a session
+        created via :meth:`create_session` with no raw content at all has
+        ``raw_id IS NULL``, which the classifier treats as "no evidence
+        either way" and therefore always retains (never counted as debt).
+        This seeds the same phantom shape
+        ``tests/unit/storage/test_empty_session_repair_provenance.py``'s
+        ``_seed`` helper uses (an ``agent-*.meta.json`` sidecar path, a
+        genuinely-debris shape the classifier positively refuses), so a
+        caller that wants an "empty" session to actually register as
+        maintenance debt must call this after :meth:`create_session`.
+
+        Resolves the blob store from ``self.db_path``'s own parent directory
+        (this factory's archive root), never the ambient
+        ``POLYLOGUE_ARCHIVE_ROOT``/``blob_store_root()`` config -- a caller
+        that seeds a ``DbFactory`` pointed at an archive root distinct from
+        the ambient one (e.g. verifying config-supplied paths win over
+        ambient defaults) must have the phantom blob land in the same
+        archive the classifier will actually read back from.
+        """
+        from polylogue.storage.blob_store import BlobStore
+
+        # SessionBuilder.__init__ always stores native_id as f"ext-{session_id}"
+        # (the "id" callers pass to create_session), so the lookup below must
+        # match that same transform, not the bare caller-supplied id.
+        stored_native_id = f"ext-{native_id}"
+        origin = _origin_value(provider)
+        archive_root = self.db_path.parent
+        source_db = archive_root / "source.db"
+        store = BlobStore(archive_root / "blob")
+        raw_id, blob_size = store.write_from_bytes(
+            f'{{"agentType":"general-purpose","for":"{stored_native_id}"}}'.encode()
+        )
+
+        with sqlite3.connect(source_db) as source_conn:
+            source_conn.execute(
+                """
+                INSERT INTO raw_sessions (
+                    raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, acquired_at_ms
+                ) VALUES (?, ?, ?, ?, 0, ?, ?, 1)
+                """,
+                (
+                    raw_id,
+                    origin.value,
+                    stored_native_id,
+                    f"agent-{stored_native_id}.meta.json",
+                    bytes.fromhex(raw_id),
+                    blob_size,
+                ),
+            )
+            source_conn.commit()
+
+        with sqlite3.connect(self.db_path) as index_conn:
+            cursor = index_conn.execute(
+                "UPDATE sessions SET raw_id = ? WHERE native_id = ? AND origin = ?",
+                (raw_id, stored_native_id, origin.value),
+            )
+            if cursor.rowcount != 1:
+                raise AssertionError(
+                    f"mark_as_phantom_debris: expected exactly one session row for "
+                    f"native_id={stored_native_id!r} origin={origin.value!r}, updated {cursor.rowcount}"
+                )
+            index_conn.commit()
+        return raw_id

tests/unit/api/test_assertion_candidate_evidence_disclosure.py

@@ -15,7 +15,7 @@
 
 from polylogue import Polylogue
 from polylogue.archive.message.roles import Role
-from polylogue.core.enums import AssertionKind, BlockType, Provider
+from polylogue.core.enums import AssertionKind, BlockType, Provider, TitleSource
 from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession
 from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
 from polylogue.storage.sqlite.archive_tiers.user_write import upsert_assertion
@@ -29,6 +29,7 @@ def _seed_candidate(root: Path) -> tuple[str, str]:
                 source_name=Provider.CODEX,
                 provider_session_id="evidence-review",
                 title="Evidence review source",
+                title_source=TitleSource.ORIGIN,
                 messages=[
                     ParsedMessage(
                         provider_message_id="m1",

tests/unit/api/test_facade_contracts.py

@@ -4444,7 +4444,10 @@ async def test_archive_tiers_api_delete_uses_index_tier_and_keeps_user_overlay(t
         await archive.close()
 
 
-@pytest.mark.frozen_clock_modules("polylogue.storage.sqlite.archive_tiers.archive")
+@pytest.mark.frozen_clock_modules(
+    "polylogue.storage.sqlite.archive_tiers.archive",
+    "polylogue.storage.sqlite.archive_tiers.revision_governance",
+)
 async def test_archive_tiers_api_raw_artifacts_read_source_tier(tmp_path: Path, frozen_clock: FrozenClock) -> None:
     """Raw artifact facade reads ``source.db`` rows and their parse-lifecycle timestamp.
 

tests/unit/cli/__snapshots__/test_help_snapshots.ambr

@@ -118,6 +118,11 @@
                                     Sort by field
     --reverse                       Reverse sort order
     --sample INTEGER                Random sample of N sessions
+    --root / --no-root              Only top-level sessions (--root, the implicit
+                                    default when unset) or only subagent/branch
+                                    children (--no-root). Session counts count
+                                    top-level sessions unless --no-root or
+                                    root:false is used (polylogue-j8u2).
     -o, --output TEXT               Output destinations: browser, clipboard,
                                     stdout (comma-separated)
     --json                          Shortcut for --format json. Disables color and
@@ -192,6 +197,7 @@
     Other commands:
       agent        Install executable agent guidance.
       annotations  Import typed annotation batches.
+      compare      Blind pairwise comparative judgment and calibration.
 
   '''
 # ---

tests/unit/cli/__snapshots__/test_plain_cli_snapshots.ambr

@@ -50,8 +50,8 @@
           "total_cost_usd": 0.0,
           "cost_is_estimated": true,
           "tokens": {
-            "input_tokens": 191,
-            "output_tokens": 0,
+            "input_tokens": 100,
+            "output_tokens": 91,
             "cache_read_tokens": 0,
             "cache_write_tokens": 0
           },
@@ -162,8 +162,8 @@
           "total_cost_usd": 0.0,
           "cost_is_estimated": true,
           "tokens": {
-            "input_tokens": 191,
-            "output_tokens": 0,
+            "input_tokens": 100,
+            "output_tokens": 91,
             "cache_read_tokens": 0,
             "cache_write_tokens": 0
           },
@@ -524,7 +524,8 @@
     "next_offset": null,
     "offset": 0,
     "origin": null,
-    "total": 2
+    "total": 2,
+    "total_unit": "top-level sessions"
   }
 
   '''
@@ -546,6 +547,7 @@
     "sessions": 2,
     "messages": 12,
     "raw_records": 2,
+    "unidentified_artifacts": 0,
     "next_action": "polylogue init",
     "component_readiness": {
       "raw_materialization": {
@@ -1015,8 +1017,8 @@
         "path": "<PATH>",
         "exists": true,
         "size_bytes": <SIZE_BYTES>,
-        "expected_user_version": 15,
-        "user_version": 15,
+        "expected_user_version": 20,
+        "user_version": 20,
         "version_status": "ok",
         "table_counts": {
           "raw_sessions": 2,
@@ -1039,8 +1041,8 @@
         "path": "<PATH>",
         "exists": true,
         "size_bytes": <SIZE_BYTES>,
-        "expected_user_version": 46,
-        "user_version": 46,
+        "expected_user_version": 57,
+        "user_version": 57,
         "version_status": "ok",
         "table_counts": {
           "sessions": 2,
@@ -1150,7 +1152,7 @@
           "path": "<PATH>",
           "exists": true,
           "wal_bytes": 0,
-          "sqlite_stat1_rows": 10,
+          "sqlite_stat1_rows": 12,
           "planner_stats_present": true
         },
         "index": {
@@ -1350,7 +1352,7 @@
   - **Sessions:** 4 analyzed / 4 matched
   - **Origins:** chatgpt-export (2), claude-code-session (2)
   - **Total cost:** $0.00 _(estimated)_
-  - **Tokens:** in 191, out 0, cache-read 0, cache-write 0
+  - **Tokens:** in 100, out 91, cache-read 0, cache-write 0
 
   ## Distributions
 
@@ -1383,7 +1385,7 @@
   | session_count | 4 |
   | wallclock_span | span_ms=75663126945, summed_wall_ms=<HASH> |
   | estimated_cost_usd | $0.000000 (estimated) |
-  | token_lanes | input=191, output=0, cache_read=0, cache_write=0 |
+  | token_lanes | input=100, output=91, cache_read=0, cache_write=0 |
   | top_expensive_session | no_signal: no session in scope carries a positive cost figure |
   | repos_touched | (none) |
   | subagent_branch_count | 0 |
@@ -1417,7 +1419,7 @@
   sessions: analyzed=4 matched=4
   origins: chatgpt-export=2, claude-code-session=2
   total_cost: $0.00 (estimated)
-  tokens: in=191 out=0 cache_read=0 cache_write=0
+  tokens: in=100 out=91 cache_read=0 cache_write=0
   cost_per_session_usd: no signal
   wall_per_session_s: n=4 total=1.45443e+<DURATION> min=240 p50=360 p90=7.38324e+07 max=7.38324e+07 mean=3.63606e+07
   wallclock_span_ms: 75663126945
@@ -1433,7 +1435,7 @@
     sessions: 4
     wallclock: span_ms=75663126945 summed_wall_ms=<HASH> (<TIMESTAMP> -> <TIMESTAMP>)
     cost: $0.000000 (estimated)
-      tokens: input=191 output=0 cache_read=0 cache_write=0
+      tokens: input=100 output=91 cache_read=0 cache_write=0
     top_expensive_session: no_signal (no session in scope carries a positive cost figure)
     repos_touched: (none)
     subagent_branch_count: 0