fix: profile, delegation and storage reds from the 2026-09-05 corpus

polylogue/maintenance/archive_verification.py

@@ -2360,12 +2360,20 @@ def _check_planner_stats(
                 _PLANNER_STATS_COVERED_TABLES,
             )
         }
+        # ANALYZE writes no sqlite_stat1 row for an empty table, so absent
+        # stats there are nothing to fix. Only a populated table can be
+        # genuinely uncovered.
+        populated = {
+            table
+            for table in _PLANNER_STATS_COVERED_TABLES
+            if table_exists(conn, table) and conn.execute(f"SELECT EXISTS(SELECT 1 FROM {table})").fetchone()[0]
+        }
     except sqlite3.Error as exc:
         return _error_check("planner-stats", f"could not read index.db: {exc}", exc=exc)
     finally:
         conn.close()
 
-    missing = [table for table in _PLANNER_STATS_COVERED_TABLES if table not in analyzed]
+    missing = [table for table in _PLANNER_STATS_COVERED_TABLES if table in populated and table not in analyzed]
     if missing:
         return ArchiveVerificationCheck(
             name="planner-stats",

polylogue/storage/sqlite/archive_tiers/archive.py

@@ -137,6 +137,7 @@
     trigram_delete_session_rows_sql,
 )
 from polylogue.storage.hook_event_authority import HookEventAuthorityCensus, census_hook_event_authority
+from polylogue.storage.introspection import relation_exists as _relation_exists
 from polylogue.storage.introspection import table_exists as _table_exists
 from polylogue.storage.raw.models import RawSessionStateUpdate
 from polylogue.storage.runtime.store_constants import SESSION_INSIGHT_MATERIALIZER_VERSION
@@ -3376,6 +3377,7 @@ def _fetch_session_profile_row(self, session_id: str) -> sqlite3.Row | None:
                    (SELECT CASE WHEN COUNT(u.model_name) = 0 THEN NULL WHEN COUNT(u.catalog_cost_usd) = COUNT(u.model_name) THEN 0 ELSE 1 END FROM session_model_usage u WHERE u.session_id = s.session_id) AS cost_is_estimated,
                    COALESCE((SELECT CASE WHEN MAX(u.provider_cost_usd) IS NOT NULL THEN 'origin_reported' WHEN MAX(u.catalog_cost_usd) IS NOT NULL THEN 'priced' END FROM session_model_usage u WHERE u.session_id = s.session_id), CASE WHEN s.reported_cost_usd IS NOT NULL THEN 'origin_reported' END) AS cost_provenance,
                    (SELECT COALESCE(SUM(u.provider_cost_usd), SUM(u.catalog_cost_usd), s.reported_cost_usd) FROM session_model_usage u WHERE u.session_id = s.session_id) AS total_cost_usd, sp.total_duration_ms,
+                   sp.input_row_count,
                    sp.evidence_payload_json, sp.inference_payload_json, sp.enrichment_payload_json
             FROM session_profiles sp
             JOIN sessions s ON s.session_id = sp.session_id
@@ -5557,11 +5559,13 @@ def _insight_readiness_entry(
             orphan_count,
             artifact_names,
         ) = spec
-        table_present = _table_exists(self._conn, table_name)
+        # Several insights are backed by query-time views (threads, delegations,
+        # actions), which exist and carry rows; presence is relation presence.
+        table_present = _relation_exists(self._conn, table_name)
         artifacts = tuple(
             InsightStorageArtifact(
                 name=artifact,
-                present=_table_exists(self._conn, artifact),
+                present=_relation_exists(self._conn, artifact),
             )
             for artifact in artifact_names
         )

polylogue/storage/sqlite/archive_tiers/write.py

@@ -3769,6 +3769,11 @@ def _write_attachments(
         attachment_id = _attachment_id(session_id, attachment)
         message_id = resolved_message_ids.get(id(attachment))
         if message_id is None:
+            # The owner is ambiguous, so no ref may be guessed, but the
+            # attachment's identity and bytes are still evidence. The row is
+            # written unreferenced and kept out of the ref-count sweep, which
+            # exists to collect rows whose refs went away.
+            _write_attachment_row(conn, attachment_id, attachment, preacquired_blobs)
             continue
         direction, producer_ref = _attachment_provenance(
             attachment, owning_messages.get(message_id), resolved_message_id=message_id
@@ -3781,33 +3786,7 @@ def _write_attachments(
                 f"attachment_id={attachment.provider_attachment_id!r}"
             )
         touched_attachment_ids.add(attachment_id)
-        acquired_blob = (preacquired_blobs or {}).get(id(attachment))
-        blob_hash, byte_count, acquisition_status = (
-            acquired_blob if acquired_blob is not None else _acquire_attachment_blob(conn, attachment)
-        )
-        conn.execute(
-            """
-            INSERT INTO attachments (
-                attachment_id, display_name, media_type, byte_count, blob_hash, acquisition_status, ref_count
-            ) VALUES (?, ?, ?, ?, ?, ?, 0)
-            ON CONFLICT(attachment_id) DO UPDATE SET
-                display_name = COALESCE(excluded.display_name, attachments.display_name),
-                media_type = COALESCE(excluded.media_type, attachments.media_type),
-                byte_count = excluded.byte_count,
-                blob_hash = COALESCE(excluded.blob_hash, attachments.blob_hash),
-                acquisition_status =
-                    CASE WHEN excluded.acquisition_status = 'acquired'
-                         THEN 'acquired' ELSE attachments.acquisition_status END
-            """,
-            (
-                attachment_id,
-                _sqlite_text(attachment.name),
-                _sqlite_text(attachment.mime_type),
-                byte_count,
-                blob_hash,
-                acquisition_status,
-            ),
-        )
+        _write_attachment_row(conn, attachment_id, attachment, preacquired_blobs)
         ref_position = attachment_positions[id(attachment)]
         ref_id = f"{message_id}:attachment:{ref_position}"
         # Bulk rebuilds may suspend FK enforcement. Mirror REPLACE's cascade
@@ -3866,6 +3845,42 @@ def _write_attachments(
     refresh_and_sweep_attachment_rows(conn, affected_attachment_ids)
 
 
+def _write_attachment_row(
+    conn: sqlite3.Connection,
+    attachment_id: str,
+    attachment: ParsedAttachment,
+    preacquired_blobs: dict[int, tuple[bytes | None, int, str]] | None,
+) -> None:
+    """Upsert the attachment's identity and bytes, leaving refs to the caller."""
+    acquired_blob = (preacquired_blobs or {}).get(id(attachment))
+    blob_hash, byte_count, acquisition_status = (
+        acquired_blob if acquired_blob is not None else _acquire_attachment_blob(conn, attachment)
+    )
+    conn.execute(
+        """
+        INSERT INTO attachments (
+            attachment_id, display_name, media_type, byte_count, blob_hash, acquisition_status, ref_count
+        ) VALUES (?, ?, ?, ?, ?, ?, 0)
+        ON CONFLICT(attachment_id) DO UPDATE SET
+            display_name = COALESCE(excluded.display_name, attachments.display_name),
+            media_type = COALESCE(excluded.media_type, attachments.media_type),
+            byte_count = excluded.byte_count,
+            blob_hash = COALESCE(excluded.blob_hash, attachments.blob_hash),
+            acquisition_status =
+                CASE WHEN excluded.acquisition_status = 'acquired'
+                     THEN 'acquired' ELSE attachments.acquisition_status END
+        """,
+        (
+            attachment_id,
+            _sqlite_text(attachment.name),
+            _sqlite_text(attachment.mime_type),
+            byte_count,
+            blob_hash,
+            acquisition_status,
+        ),
+    )
+
+
 def refresh_and_sweep_attachment_rows(conn: sqlite3.Connection, attachment_ids: set[str]) -> None:
     """Recompute ``attachments.ref_count`` from live refs and sweep zero-ref rows.
 

tests/infra/reindex_differential.py

@@ -38,6 +38,7 @@
     "messages_fts_identity": "FTS support relation compared through public search and FtsReadiness",
     "query_unit_frame_state": "cursor invalidation epoch depends on write-route history",
     "raw_revision_applications": "attempt receipts contain generated decision ids and wall-clock timestamps",
+    "schema_identity": "stores a hash of the DDL identity itself, not derived model data",
 }
 
 # One explicit entry per comparable DDL table. Empty sets are declarations:
@@ -61,6 +62,7 @@
     "session_agent_policies": frozenset(),
     "session_commits": frozenset(),
     "session_events": frozenset(),
+    "session_identity_claims": frozenset(),
     "session_latency_profiles": frozenset({"materialized_at"}),
     "session_links": frozenset({"observed_at_ms", "resolved_at_ms"}),
     "session_model_usage": frozenset(),

tests/unit/api/test_facade_contracts.py

@@ -61,8 +61,14 @@
     delegation_subtree_object_id,
 )
 from polylogue.operations.bindings import OperationBinding
-from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession
+from polylogue.sources.parsers.base import (
+    ParsedContentBlock,
+    ParsedMessage,
+    ParsedSession,
+    ParsedSessionEvent,
+)
 from polylogue.storage.block_anchor import format_block_anchor
+from polylogue.storage.runtime.store_constants import SESSION_INSIGHT_MATERIALIZER_VERSION
 from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore
 from polylogue.storage.sqlite.archive_tiers.bootstrap import (
     initialize_active_archive_root,
@@ -2074,7 +2080,11 @@ async def test_regenerate_private_fable_packet_reads_real_delegations_and_labels
     try:
         with ArchiveStore(archive.config.archive_root) as archive_db:
             parent_session_id = archive_db.write_parsed(
-                _delegation_parent_session(provider_session_id="fable-facade-parent-v1", with_dispatch=True)
+                _delegation_parent_session(
+                    provider_session_id="fable-facade-parent-v1",
+                    with_dispatch=True,
+                    child_provider_session_id="fable-facade-child-v1",
+                )
             )
             archive_db.write_parsed(
                 ParsedSession(
@@ -3561,7 +3571,12 @@ async def test_resolve_ref_renders_finding_claim_with_controls(tmp_path: Path) -
         await archive.close()
 
 
-def _delegation_parent_session(*, provider_session_id: str, with_dispatch: bool) -> ParsedSession:
+def _delegation_parent_session(
+    *,
+    provider_session_id: str,
+    with_dispatch: bool,
+    child_provider_session_id: str | None = None,
+) -> ParsedSession:
     """Ingest-shaped parent fixture: writes real session/message/block rows
     through the live archive writer (``ArchiveStore.write_parsed`` ->
     ``write_parsed_session_to_archive``), the same seam the daemon uses --
@@ -3609,11 +3624,27 @@ def _delegation_parent_session(*, provider_session_id: str, with_dispatch: bool)
                 ],
             )
         )
+    session_events: list[ParsedSessionEvent] = []
+    if with_dispatch and child_provider_session_id is not None:
+        # The writer's parent-dispatch resolver reads this event to join
+        # the Task tool_use block to the child session (write.py
+        # _resolve_parent_dispatch_block_id); without it the link stays
+        # edge_only regardless of the tool_use/tool_result pair above.
+        session_events.append(
+            ParsedSessionEvent(
+                event_type="claude_delegation_progress",
+                payload={
+                    "provider_tool_id": "task-1",
+                    "child_provider_id": child_provider_session_id,
+                },
+            )
+        )
     return ParsedSession(
         source_name=Provider.CLAUDE_CODE,
         provider_session_id=provider_session_id,
         title="Delegation parent fixture",
         messages=messages,
+        session_events=session_events,
     )
 
 
@@ -3627,7 +3658,11 @@ async def test_resolve_ref_returns_resolved_delegation_attempt_payload(tmp_path:
     try:
         with ArchiveStore(archive.config.archive_root) as archive_db:
             parent_session_id = archive_db.write_parsed(
-                _delegation_parent_session(provider_session_id="delegation-parent-v1", with_dispatch=True)
+                _delegation_parent_session(
+                    provider_session_id="delegation-parent-v1",
+                    with_dispatch=True,
+                    child_provider_session_id="delegation-child-v1",
+                )
             )
             child_session_id = archive_db.write_parsed(
                 ParsedSession(
@@ -3873,7 +3908,11 @@ async def test_resolve_ref_returns_delegation_ancestry_and_subtree_payloads(tmp_
     try:
         with ArchiveStore(archive.config.archive_root) as archive_db:
             root_id = archive_db.write_parsed(
-                _delegation_parent_session(provider_session_id="delegation-tree-root-v1", with_dispatch=True)
+                _delegation_parent_session(
+                    provider_session_id="delegation-tree-root-v1",
+                    with_dispatch=True,
+                    child_provider_session_id="delegation-tree-mid-v1",
+                )
             )
             mid_id = archive_db.write_parsed(
                 ParsedSession(
@@ -5376,7 +5415,7 @@ async def test_archive_tiers_api_session_costs_read_index_tier(tmp_path: Path) -
         assert costs[0].estimate.status == "exact"
         assert costs[0].estimate.total_usd == 1.25
         assert costs[0].estimate.basis.provider_reported_usd == 1.25
-        assert costs[0].provenance.materializer_version == 14
+        assert costs[0].provenance.materializer_version == SESSION_INSIGHT_MATERIALIZER_VERSION
         assert len(unavailable) == 1
         assert unavailable[0].estimate.status == "unavailable"
         assert unavailable[0].estimate.missing_reasons == ("no_tokens",)
@@ -5478,7 +5517,7 @@ async def test_archive_tiers_api_latency_profiles_read_index_tier(tmp_path: Path
         assert profile.session_id == session_id
         assert profile.origin == Origin.CODEX_SESSION.value
         assert profile.title == "Latency v1"
-        assert profile.provenance.materializer_version == 14
+        assert profile.provenance.materializer_version == SESSION_INSIGHT_MATERIALIZER_VERSION
         assert profile.latency.median_agent_response_ms == 90000
         assert profile.latency.median_user_response_ms == 120000
         assert profile.latency.median_tool_call_ms == 0

tests/unit/insights/test_delegation_work_evidence_materializer.py

@@ -42,14 +42,6 @@ def _seed_delegation(archive_root: Path) -> None:
         child_id = conn.execute(
             "SELECT session_id FROM sessions WHERE origin = 'claude-code-session' AND native_id = 'child'"
         ).fetchone()[0]
-        conn.execute(
-            """
-            INSERT INTO session_links (
-                src_session_id, dst_origin, dst_native_id, link_type, resolved_dst_session_id, observed_at_ms
-            ) VALUES (?, 'claude-code-session', 'parent', 'subagent', ?, 1)
-            """,
-            (child_id, parent_id),
-        )
         conn.execute(
             """
             INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash, occurred_at_ms)
@@ -68,6 +60,21 @@ def _seed_delegation(archive_root: Path) -> None:
             """,
             (message_id, parent_id),
         )
+        # block_id is generated as message_id || ':' || position; a literal
+        # tool_id ("task-1") is a different value and would leave the join
+        # in delegation_facts_source (index.py) unresolved.
+        block_id = conn.execute(
+            "SELECT block_id FROM blocks WHERE message_id = ? AND position = 0", (message_id,)
+        ).fetchone()[0]
+        conn.execute(
+            """
+            INSERT INTO session_links (
+                src_session_id, dst_origin, dst_native_id, link_type, resolved_dst_session_id,
+                parent_tool_use_block_id, observed_at_ms
+            ) VALUES (?, 'claude-code-session', 'parent', 'subagent', ?, ?, 1)
+            """,
+            (child_id, parent_id, block_id),
+        )
 
 
 def test_materializer_replaces_archive_projection_and_tracks_delegation_freshness(tmp_path: Path) -> None:

tests/unit/maintenance/test_archive_verification.py

@@ -1203,10 +1203,12 @@ def test_missing_sqlite_stat1_is_warning_not_error(tmp_path: Path) -> None:
 
 
 def test_partial_analyze_coverage_is_reported_by_table(tmp_path: Path) -> None:
+    """Anti-vacuity: collapsing the check to a single all-or-nothing verdict
+    drops the per-table name this asserts on."""
     _seed_coherent_archive(tmp_path)
     conn = _connect(tmp_path / "index.db")
     try:
-        conn.execute("DELETE FROM sqlite_stat1 WHERE tbl = 'session_links'")
+        conn.execute("DELETE FROM sqlite_stat1 WHERE tbl = 'blocks'")
         conn.commit()
     finally:
         conn.close()
@@ -1215,7 +1217,27 @@ def test_partial_analyze_coverage_is_reported_by_table(tmp_path: Path) -> None:
 
     check = _check(report, "planner-stats")
     assert check.status is OutcomeStatus.WARNING
-    assert check.evidence["missing_tables"] == ["session_links"]
+    assert check.evidence["missing_tables"] == ["blocks"]
+
+
+def test_empty_covered_table_without_stats_is_not_missing_coverage(tmp_path: Path) -> None:
+    """ANALYZE writes no sqlite_stat1 row for an empty table.
+
+    Anti-vacuity: dropping the emptiness exemption makes every archive with no
+    action pairs warn, which is what this fixture is."""
+    _seed_coherent_archive(tmp_path)
+    conn = _connect(tmp_path / "index.db")
+    try:
+        assert conn.execute("SELECT COUNT(*) FROM action_pairs").fetchone()[0] == 0
+        assert conn.execute("SELECT COUNT(*) FROM sqlite_stat1 WHERE tbl = 'action_pairs'").fetchone()[0] == 0
+    finally:
+        conn.close()
+
+    report = verify_archive(tmp_path, checks=("planner-stats",))
+
+    check = _check(report, "planner-stats")
+    assert check.status is OutcomeStatus.OK
+    assert check.evidence["missing_tables"] == []
 
 
 def test_missing_archive_root_reports_skips_not_crashes(tmp_path: Path) -> None:

tests/unit/pipeline/test_content_hash_determinism.py

@@ -307,15 +307,19 @@ def test_hash_payload_json_encoding_stability() -> None:
     assert encoded == '{"a":2,"z":1}'
 
 
-# ── Provider identity NOT part of content hash ────────────────────────
+# ── Provider identity is part of content hash ─────────────────────────
 
 
-def test_provider_source_does_not_affect_content_hash() -> None:
-    """Content hash depends on content, not on which provider parsed it.
+def test_provider_source_is_part_of_content_hash() -> None:
+    """``source_name`` is a declared member of the ParsedSession partition.
 
-    Two ParsedSessions with identical content but different source_names
-    should produce identical content hashes. Source identity is part of the
-    session ID, not the content hash.
+    Content identity is an explicit classification of parser fields
+    (``polylogue/pipeline/ids.py``), and ``source_name`` sits in the hashed
+    set alongside ``provider_session_id``. Two sessions that agree on every
+    message but were produced by different providers are different sessions.
+
+    Anti-vacuity: moving ``source_name`` to the excluded set collapses these
+    two hashes and makes this equal.
     """
     msg = _msg("m1", "user", "hello")
     conv_chatgpt = ParsedSession(
@@ -336,7 +340,7 @@ def test_provider_source_does_not_affect_content_hash() -> None:
         messages=[msg],
         attachments=[],
     )
-    assert session_content_hash(conv_chatgpt) == session_content_hash(conv_claude)
+    assert session_content_hash(conv_chatgpt) != session_content_hash(conv_claude)
 
 
 # ── Session events affect hash ───────────────────────────────────────