perf(storage): stop duplicating redundant events into session_events

docs/internals.md

@@ -173,6 +173,43 @@ Polylogue has two schema-evolution regimes, keyed by tier durability.
   hash, so concurrent publishers of identical bytes remain independent.
   Existing v3 tiers migrate additively through
   `004_blob_publication_reservations.sql` after a verified backup manifest.
+- Index schema version 42 stops materializing `session_events` rows for four
+  event types that are fully redundant with a sibling typed table
+  (`token_count`, `message_usage`, `agent_policy`, `agent_message`;
+  polylogue-bo9n consumer audit). `token_count`/`message_usage` are already
+  unpacked into `session_provider_usage_events`'s typed columns (the cost
+  model's sole read path, `storage/usage.py`); `agent_policy` is already
+  unpacked into `session_agent_policies` (the sole confirmed reader,
+  `read_session_agent_policies`); `agent_message` payloads never carry text
+  (Codex never populates it there) and the real text is guaranteed to exist
+  as a twin `ParsedMessage` via `_codex_event_message`. On a live generation
+  these four types combined were 2,496,806 rows (37% of `session_events`) and
+  694.4MB of `payload_json`, blended-share estimate ≈846MB of the table's
+  2.31GB. Parsers keep emitting all four event types unchanged (feeding
+  `material_protocol`'s parse-time transcript encode, which reads
+  `ParsedSession.session_events` directly and is unaffected); only the writer
+  (`_write_session_events` in `storage/sqlite/archive_tiers/write.py`) filters
+  them out of the `session_events` INSERT. `reasoning`/`agent_reasoning`
+  (zero captured content today -- an evidence gap, not duplication) and
+  `turn_context` (low urgency, no confirmed payload reader) are deliberately
+  excluded pending an explicit operator evidence-doctrine call; `function_call`/
+  `function_call_output` payload-slimming is a separate, not-yet-decided
+  change. This is a writer-behavior change with no DDL delta on
+  `session_events` itself, so there is no declared clone-safe SQL
+  fast-forward (see `IndexDeltaDeclaration` for v42 in
+  `polylogue/storage/sqlite/lifecycle.py`); existing index tiers rebuild from
+  source evidence (`polylogue ops reset --index && polylogued run`). A
+  companion finding from the same audit -- `session_provider_usage_events.
+  payload_json` totaling ~700MB with zero readers in `storage/usage.py` --
+  was **not** dropped in this version: `_reextract_provider_usage_tail_db`
+  (branch-tail re-extraction for prefix-sharing forks/resumes) reads
+  `json_extract(payload_json, '$.estimated_cost_usd' | ...)` to decide
+  whether a zero-token-count row still carries Hermes cost-provenance
+  evidence before deleting it, a reader the audit's `storage/usage.py`-only
+  sweep missed. Dropping the column outright would break that query; keeping
+  it requires either preserving the JSON column or promoting the eight
+  provenance keys (`hermes_state.py`) to typed columns, a follow-up decision
+  left to polylogue-c3ip.
 - Index schema version 41 stops materializing `tool_input`/`output_text` text
   copies on `action_pairs` (polylogue-2i2w). `action_pairs` keeps only
   join/rank/outcome columns for a paired `tool_use`/`tool_result` block

polylogue/storage/sqlite/archive_tiers/index.py

@@ -24,7 +24,7 @@
 )
 from polylogue.storage.sqlite.delegation_facts import delegation_facts_insert_sql
 
-INDEX_SCHEMA_VERSION = 41
+INDEX_SCHEMA_VERSION = 42
 
 FTS_FRESHNESS_STATE_DDL = """
 CREATE TABLE IF NOT EXISTS fts_freshness_state (

polylogue/storage/sqlite/archive_tiers/write.py

@@ -2570,6 +2570,31 @@ def _next_session_event_position(conn: sqlite3.Connection, session_id: str) -> i
     return int(row[0] or 0) if row is not None else 0
 
 
+# Event types whose full evidence already lives durably in a sibling typed
+# table -- materializing a second copy into ``session_events`` is a pure,
+# zero-evidence-loss duplication (polylogue-bo9n consumer audit, 2026-07-19):
+#
+# - ``token_count`` / ``message_usage``: fully re-derivable from
+#   ``session_provider_usage_events`` (the cost model's sole read path,
+#   ``storage/usage.py``); every field the writer would otherwise copy into
+#   ``session_events.payload_json`` is already unpacked into that table's
+#   typed columns.
+# - ``agent_policy``: fully re-derivable from ``session_agent_policies``
+#   (dedicated typed table, identical fields, sole confirmed reader
+#   ``read_session_agent_policies``).
+# - ``agent_message``: the payload never carries text (Codex never populates
+#   it there); the real text is guaranteed to exist as a ``ParsedMessage``
+#   via ``_codex_event_message`` -- this is a pure existence marker with a
+#   message-shaped twin already present.
+#
+# ``reasoning``/``agent_reasoning``/``turn_context`` are deliberately excluded
+# (need an operator evidence-doctrine call per the audit); ``function_call``/
+# ``function_call_output`` payload-slimming is a separate, not-yet-decided
+# change. Parsers keep emitting all of these events unchanged -- only this
+# writer materialization step filters them.
+_SESSION_EVENTS_REDUNDANT_TYPES = frozenset({"token_count", "message_usage", "agent_policy", "agent_message"})
+
+
 def _write_session_events(
     conn: sqlite3.Connection,
     session_id: str,
@@ -2606,18 +2631,19 @@ def _write_session_events(
         source_message_id = by_native_id.get(source_message_provider_id or "")
         if source_message_id is None and inherited_source_message_ids is not None:
             source_message_id = inherited_source_message_ids.get(source_message_provider_id or "")
-        session_event_rows.append(
-            (
-                session_id,
-                source_message_id,
-                _sqlite_text(source_message_provider_id),
-                position,
-                _sqlite_text(event.event_type),
-                _sqlite_text(_event_summary(event) or ""),
-                _json_dumps(event.payload),
-                _timestamp_ms(event.timestamp),
-            ),
-        )
+        if event.event_type not in _SESSION_EVENTS_REDUNDANT_TYPES:
+            session_event_rows.append(
+                (
+                    session_id,
+                    source_message_id,
+                    _sqlite_text(source_message_provider_id),
+                    position,
+                    _sqlite_text(event.event_type),
+                    _sqlite_text(_event_summary(event) or ""),
+                    _json_dumps(event.payload),
+                    _timestamp_ms(event.timestamp),
+                ),
+            )
         if event.event_type == "agent_policy":
             agent_policy_rows.append(
                 (

polylogue/storage/sqlite/lifecycle.py

@@ -256,6 +256,22 @@ class IndexDeltaDeclarationReport(TypedDict):
             ),
         ),
     ),
+    IndexDeltaDeclaration(
+        version=42,
+        # The writer stops materializing `token_count`/`message_usage`/
+        # `agent_policy`/`agent_message` rows into `session_events`
+        # (polylogue-bo9n zero-evidence-loss filtering) -- each is fully
+        # re-derivable from a sibling typed table already written at the
+        # same commit (`session_provider_usage_events`, `session_agent_policies`)
+        # or from a twin `ParsedMessage`, so no unique evidence is lost. This
+        # is a writer-materialization change, not a DDL change (session_events
+        # keeps its existing columns) -- there is no declared clone-safe SQL
+        # delta because a fast-forward would require re-deriving which
+        # already-persisted rows a fresh parse would have skipped; existing
+        # index tiers rebuild from source evidence instead
+        # (`polylogue ops reset --index && polylogued run`).
+        classes=(DerivedDeltaClass.SEMANTIC_REPARSE,),
+    ),
 )
 
 

tests/unit/storage/test_archive_tiers_write.py

@@ -1065,16 +1065,11 @@ def test_archive_tiers_writer_materializes_supported_session_events(tmp_path: Pa
             "payload_json": '{"cwd":"/tmp"}',
             "occurred_at_ms": None,
         },
-        {
-            "event_id": f"{session_id}:3",
-            "source_message_id": None,
-            "source_message_provider_id": None,
-            "position": 3,
-            "event_type": "agent_policy",
-            "summary": "",
-            "payload_json": '{"approval":"on-request"}',
-            "occurred_at_ms": 1_767_225_603_000,
-        },
+        # NOTE: the `agent_policy` event (position 3) is deliberately absent
+        # here -- it is fully redundant with `session_agent_policies` (see
+        # assertion below) and the writer no longer materializes a second
+        # copy into `session_events` (polylogue-bo9n zero-evidence-loss
+        # filtering, index schema v42).
     ]
     hydrated = sync_session_events_batch(conn, [session_id])[session_id]
     assert hydrated[0].timestamp == "2026-01-01T00:00:01+00:00"