From f85256650a23c2fbff8775ea033010eb02bd5e7a Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 14:32:31 +0200 Subject: [PATCH 1/9] test: read profile totals and ingest seams at their current locations session_profiles carries no token columns; session_model_usage is the sole authority and the reader overlays it. _profile_totals now composes the production read path instead of selecting deleted columns. wal_checkpoint opens through open_daemon_connection, and the batch drain prelude reads the connection for stale-session cleanup and an FTS staleness verdict; both monkeypatch targets follow. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- .../test_ingest_batch_resource_bounds.py | 7 ++++ .../test_ingest_batch_wal_checkpoint.py | 2 +- ...session_profile_model_usage_consistency.py | 42 ++++++++++++++----- 3 files changed, 39 insertions(+), 12 deletions(-) diff --git a/tests/unit/pipeline/test_ingest_batch_resource_bounds.py b/tests/unit/pipeline/test_ingest_batch_resource_bounds.py index 923699f4ac..60f237cf6e 100644 --- a/tests/unit/pipeline/test_ingest_batch_resource_bounds.py +++ b/tests/unit/pipeline/test_ingest_batch_resource_bounds.py @@ -196,6 +196,13 @@ def fake_write(*args: object, **kwargs: object) -> bool: return True monkeypatch.setattr(ingest_batch_core, "_write_session_entry", fake_write) + # The drain prelude reads the connection for stale-session cleanup and an + # archive-wide FTS staleness verdict; neither is what this test measures. + monkeypatch.setattr(ingest_batch_core, "_delete_stale_sessions_for_raw_entries", lambda *_a, **_k: None) + monkeypatch.setattr( + "polylogue.storage.fts.freshness.message_fts_recorded_exact_stale_sync", + lambda *_a, **_k: False, + ) ingest_batch_core._drain_ready_session_entries( object(), # type: ignore[arg-type] diff --git a/tests/unit/pipeline/test_ingest_batch_wal_checkpoint.py b/tests/unit/pipeline/test_ingest_batch_wal_checkpoint.py index 2499b364ab..97e620109f 100644 --- a/tests/unit/pipeline/test_ingest_batch_wal_checkpoint.py +++ b/tests/unit/pipeline/test_ingest_batch_wal_checkpoint.py @@ -390,7 +390,7 @@ def close(self) -> None: monkeypatch.setattr("polylogue.storage.sqlite.wal_checkpoint._wal_size", lambda db: 1024) monkeypatch.setattr( - "polylogue.storage.sqlite.wal_checkpoint.open_connection", lambda *_args, **_kwargs: FakeConnection() + "polylogue.storage.sqlite.wal_checkpoint.open_daemon_connection", lambda *_args, **_kwargs: FakeConnection() ) monkeypatch.setattr( "polylogue.storage.sqlite.wal_checkpoint._sqlite_file_holders", diff --git a/tests/unit/storage/test_session_profile_model_usage_consistency.py b/tests/unit/storage/test_session_profile_model_usage_consistency.py index 49543062e1..be5f8dfb99 100644 --- a/tests/unit/storage/test_session_profile_model_usage_consistency.py +++ b/tests/unit/storage/test_session_profile_model_usage_consistency.py @@ -1,4 +1,4 @@ -"""session_profiles token/cost columns must agree with session_model_usage. +"""Session-profile token totals must agree with session_model_usage. polylogue-r7p6: for Codex sessions, session_profiles token columns undercounted session_model_usage by roughly 1000x (6.43M vs 6.74B input tokens across the @@ -17,9 +17,14 @@ The fix (``ModelUsageTotals`` plumbed through ``compute_session_cost`` / ``build_session_profile`` / ``build_session_insight_records``) makes profile building read ``session_model_usage`` back directly -- the same substrate the -archive's own cost/usage rollups are built from -- so profile columns are +archive's own cost/usage rollups are built from -- so profile totals are identical to that rollup by construction, for every origin, not just Codex. +``session_profiles`` stores no token or cost columns; ``session_model_usage`` +is the sole authority and the reader overlays it onto each record. These tests +assert on the composed read path, so they go red if the overlay is dropped or +the write path stops populating ``session_model_usage``. + These tests exercise the real production write path (``write_parsed_session_to_archive``) and both session-insight materializer twins (``rebuild_session_insights_sync`` / ``rebuild_session_insights_async``), @@ -36,12 +41,17 @@ from polylogue.archive.message.roles import Role from polylogue.core.enums import BlockType, Provider from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession, ParsedSessionEvent +from polylogue.storage.derived.session.profile_cost import ( + apply_profile_cost_lanes, + read_model_usage_batch_sync, +) from polylogue.storage.derived.session.rebuild import ( rebuild_session_insights_async, rebuild_session_insights_sync, ) from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive +from polylogue.storage.sqlite.queries.mappers import _row_to_session_profile_record # Realistic Codex cumulative usage: input is inclusive of cached (96% cached, # matching the corpus finding in _provider_usage_disjoint_lanes's docstring), @@ -148,16 +158,26 @@ def _model_usage_totals(conn: sqlite3.Connection, session_id: str) -> tuple[int, def _profile_totals(conn: sqlite3.Connection, session_id: str) -> tuple[int, int, int, int]: - row = conn.execute( - """ - SELECT total_input_tokens, total_output_tokens, total_cache_read_tokens, total_cache_write_tokens - FROM session_profiles - WHERE session_id = ? - """, - (session_id,), - ).fetchone() + """Session-profile token totals as a reader sees them. + + ``session_profiles`` stores no token columns; ``session_model_usage`` is + the sole authority and the reader overlays it onto the profile record. + This helper composes exactly the production read path + (``session_insight_profile_reads``) so the assertion covers the overlay a + consumer of ``SessionProfileRecord`` actually depends on. + """ + row = conn.execute("SELECT * FROM session_profiles WHERE session_id = ?", (session_id,)).fetchone() assert row is not None, f"no session_profiles row for {session_id}" - return (int(row[0]), int(row[1]), int(row[2]), int(row[3])) + record = apply_profile_cost_lanes( + _row_to_session_profile_record(row), + read_model_usage_batch_sync(conn, [session_id]), + ) + return ( + int(record.total_input_tokens), + int(record.total_output_tokens), + int(record.total_cache_read_tokens), + int(record.total_cache_write_tokens), + ) def test_codex_profile_tokens_match_model_usage_after_sync_rebuild(tmp_path: Path) -> None: From 7d1627977a8680b8306e0217024bc79e73e10b78 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 14:52:01 +0200 Subject: [PATCH 2/9] fix(maintenance): absent stats for an empty table are not missing coverage ANALYZE writes no sqlite_stat1 row for a table with no rows, so planner-stats warned on every archive whose action_pairs is empty. Demand coverage only for populated tables. The per-table coverage test targeted session_links, which the covered-table list has never contained. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/maintenance/archive_verification.py | 10 ++++++- .../maintenance/test_archive_verification.py | 26 +++++++++++++++++-- 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index f07bb0d3a8..4a2335bfff 100644 --- a/polylogue/maintenance/archive_verification.py +++ b/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", diff --git a/tests/unit/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index bd1e0dcad1..db34614504 100644 --- a/tests/unit/maintenance/test_archive_verification.py +++ b/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: From 1c94a83e44f75f2d3dc02d99b080a281b7140463 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 14:52:01 +0200 Subject: [PATCH 3/9] test: supply parent-dispatch evidence delegation resolution now requires Resolution joins a child to its dispatch through session_links.parent_tool_use_block_id, which the writer sets only from a claude_delegation_progress event on the parent. Fixtures that predate that contract resolved to unresolved/edge_only. The provider fixture carries the evidence as a real progress record so the Claude Code parser still produces it; the materializer fixture, which inserts session_links directly, sets the block id directly. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- tests/unit/api/test_facade_contracts.py | 53 ++++++++++++++++--- ...t_delegation_work_evidence_materializer.py | 23 +++++--- .../test_delegation_provider_fixtures.py | 18 +++++++ 3 files changed, 79 insertions(+), 15 deletions(-) diff --git a/tests/unit/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py index 120ef43de5..268e4692cc 100644 --- a/tests/unit/api/test_facade_contracts.py +++ b/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 diff --git a/tests/unit/insights/test_delegation_work_evidence_materializer.py b/tests/unit/insights/test_delegation_work_evidence_materializer.py index 22bc7aaa36..6937040e2d 100644 --- a/tests/unit/insights/test_delegation_work_evidence_materializer.py +++ b/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: diff --git a/tests/unit/pipeline/test_delegation_provider_fixtures.py b/tests/unit/pipeline/test_delegation_provider_fixtures.py index e79e905376..7d2e5443ad 100644 --- a/tests/unit/pipeline/test_delegation_provider_fixtures.py +++ b/tests/unit/pipeline/test_delegation_provider_fixtures.py @@ -120,6 +120,24 @@ def _claude_code_task_dispatch_payload(*, session_id: str) -> list[dict[str, obj "content": [{"type": "tool_result", "tool_use_id": "toolu_task1", "content": "Subagent finished."}], }, }, + # Real wire shape read by _accumulate_delegation_progress + # (code_parser.py:1841-1884): a top-level "progress" record whose + # "data" is an "agent_progress" tick naming the dispatching + # parentToolUseID and the spawned child's composed session id. + # Without this record parentToolUseID is never wired to + # session_links.parent_tool_use_block_id (origin_specs.py:1040-1043). + { + "type": "progress", + "uuid": "p1", + "parentUuid": "a1", + "sessionId": session_id, + "timestamp": "2025-01-01T10:00:06Z", + "parentToolUseID": "toolu_task1", + "data": { + "type": "agent_progress", + "childSessionId": f"{session_id}:agent-delegation-child", + }, + }, ] From f047650a20689bc7149529eed8713f182fc3ba2b Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 15:00:49 +0200 Subject: [PATCH 4/9] fix(storage): select the input_row_count the profile record reads get_session_profile_record indexes row["input_row_count"] unconditionally, but the profile SELECT never projected the column, so every read raised IndexError and the daemon session-insights route answered 500. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/storage/sqlite/archive_tiers/archive.py | 1 + 1 file changed, 1 insertion(+) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 6be1512d09..055ae6747e 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -3376,6 +3376,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 From 3654aec0b28ba7c0beb6aedefcbc6638f516aa22 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 15:16:58 +0200 Subject: [PATCH 5/9] fix(storage): insight presence counts views, not only tables threads, delegations and actions are query-time views. Testing presence with a table-only lookup reported them absent, which marks the insight diverged and withholds every row from an export as "insight table is absent". Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/storage/sqlite/archive_tiers/archive.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 055ae6747e..ff2f6d3999 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/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 @@ -5558,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 ) From a66edc8ca9c2043e7a0446650628141a4f05a327 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 15:16:58 +0200 Subject: [PATCH 6/9] test: classify new index tables and satisfy the threads view dependency schema_identity stamps the DDL's own hash and is not comparable; session_identity_claims is ordinary derived data. The status fixture needs session_work_events to exist before threads counts as readable. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- tests/infra/reindex_differential.py | 2 ++ tests/unit/storage/test_session_insight_status_descriptors.py | 1 + 2 files changed, 3 insertions(+) diff --git a/tests/infra/reindex_differential.py b/tests/infra/reindex_differential.py index 2d2ee7052b..bd7473c686 100644 --- a/tests/infra/reindex_differential.py +++ b/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(), diff --git a/tests/unit/storage/test_session_insight_status_descriptors.py b/tests/unit/storage/test_session_insight_status_descriptors.py index 547b127cd6..af6473399f 100644 --- a/tests/unit/storage/test_session_insight_status_descriptors.py +++ b/tests/unit/storage/test_session_insight_status_descriptors.py @@ -351,6 +351,7 @@ async def test_lightweight_status_sync_and_async_match_with_freshness_tables(tmp phase_count INTEGER NOT NULL ); CREATE TABLE session_profiles_fts (session_id TEXT NOT NULL); + CREATE TABLE session_work_events (session_id TEXT NOT NULL); CREATE TABLE threads (thread_id TEXT PRIMARY KEY); INSERT INTO sessions (session_id, parent_session_id, sort_key_ms, updated_at_ms) From 5ee7064c0769f64fdc23fc4916992b24e0b6dc73 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 15:16:59 +0200 Subject: [PATCH 7/9] test: follow embedding reuse and the declared content-hash partition Embedding finalization moved to finalize_embedding_attempt_success, and content-addressed vectors are reused across roots, so identical content in a second session makes no second provider call. source_name is a declared member of the ParsedSession semantic partition. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- .../pipeline/test_content_hash_determinism.py | 18 +++++++++++------- tests/unit/storage/test_embedding_contracts.py | 2 +- tests/unit/storage/test_embedding_dedup.py | 14 ++++++-------- 3 files changed, 18 insertions(+), 16 deletions(-) diff --git a/tests/unit/pipeline/test_content_hash_determinism.py b/tests/unit/pipeline/test_content_hash_determinism.py index 23c45fc4b3..6b9f86920e 100644 --- a/tests/unit/pipeline/test_content_hash_determinism.py +++ b/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 ─────────────────────────────────────── diff --git a/tests/unit/storage/test_embedding_contracts.py b/tests/unit/storage/test_embedding_contracts.py index 044e9b86f0..b305e22227 100644 --- a/tests/unit/storage/test_embedding_contracts.py +++ b/tests/unit/storage/test_embedding_contracts.py @@ -1622,7 +1622,7 @@ def test_archive_local_fault_is_not_ledgered_as_provider_failure( def _raise_write_fault(*args: object, **kwargs: object) -> None: raise sqlite3.OperationalError("database is locked") - monkeypatch.setattr(embedding_write_module, "complete_embedding_attempt_success", _raise_write_fault) + monkeypatch.setattr(embedding_write_module, "finalize_embedding_attempt_success", _raise_write_fault) outcome = embed_archive_session_sync(index_db, _FakeV1VectorProvider(), session_id) assert outcome.status == "error" diff --git a/tests/unit/storage/test_embedding_dedup.py b/tests/unit/storage/test_embedding_dedup.py index 2dcfead28e..070a03955a 100644 --- a/tests/unit/storage/test_embedding_dedup.py +++ b/tests/unit/storage/test_embedding_dedup.py @@ -101,14 +101,12 @@ def test_identical_content_across_two_sessions_embeds_once(tmp_path: Path) -> No outcome_b = embed_archive_session_sync(index_db, provider, session_b) assert outcome_a.status == "embedded" assert outcome_b.status == "embedded" - # Each session's embed pass still calls the provider once (per-session - # materialization does not consult prior vectors before calling out) -- - # the dedup win is in STORAGE, not in avoiding this particular pass's - # API call. The real cost saving is a subsequent session that reaches - # the freshness predicate already-fresh (see - # test_embedding_rebuild_survival.py), or the rescue path skipping the - # API entirely (see test_embedding_rescue_content_addressed_layout.py). - assert len(provider.calls) == 2 + # Session b's materialization pass consults present vector addresses by + # content hash before calling the provider (_present_vector_addresses) + # and reuses session a's vector, so the provider is called once total. + # Anti-vacuity: a regression that drops the reuse check before the + # provider call makes this 2. + assert len(provider.calls) == 1 with _connect_vec(embeddings_db) as conn: vector_rows = conn.execute("SELECT COUNT(*) FROM message_embeddings").fetchone()[0] From cd37e20f298058080b4e18f2cc8cef930bb771ec Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 15:21:37 +0200 Subject: [PATCH 8/9] fix(storage): retain an ambiguous-owner attachment as typed unowned An attachment whose owning message cannot be resolved was skipped before its own row was written, so identity and bytes were lost rather than kept without a guessed ref. Write the row and keep it out of the ref-count sweep, which collects rows whose refs went away. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- .../storage/sqlite/archive_tiers/write.py | 69 +++++++++++-------- 1 file changed, 42 insertions(+), 27 deletions(-) diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index aafa988752..509afb337c 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/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[str | None, int | None, 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. From c0b77901a64e68a047afbbedfdb3ee52581a9e34 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 17:34:02 +0200 Subject: [PATCH 9/9] fix(storage): the attachment row writer takes the caller's preacquired blob shape --- polylogue/storage/sqlite/archive_tiers/write.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index 509afb337c..839b3ced4f 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/polylogue/storage/sqlite/archive_tiers/write.py @@ -3849,7 +3849,7 @@ def _write_attachment_row( conn: sqlite3.Connection, attachment_id: str, attachment: ParsedAttachment, - preacquired_blobs: dict[int, tuple[str | None, int | None, str]] | None, + 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))