diff --git a/polylogue/maintenance/archive_verification.py b/polylogue/maintenance/archive_verification.py index f07bb0d3a..4a2335bff 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/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 6be1512d0..ff2f6d399 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 @@ -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 ) diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index aafa98875..839b3ced4 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[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. diff --git a/tests/infra/reindex_differential.py b/tests/infra/reindex_differential.py index 2d2ee7052..bd7473c68 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/api/test_facade_contracts.py b/tests/unit/api/test_facade_contracts.py index 120ef43de..268e4692c 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 22bc7aaa3..6937040e2 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/maintenance/test_archive_verification.py b/tests/unit/maintenance/test_archive_verification.py index bd1e0dcad..db3461450 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: diff --git a/tests/unit/pipeline/test_content_hash_determinism.py b/tests/unit/pipeline/test_content_hash_determinism.py index 23c45fc4b..6b9f86920 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/pipeline/test_delegation_provider_fixtures.py b/tests/unit/pipeline/test_delegation_provider_fixtures.py index e79e90537..7d2e5443a 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", + }, + }, ] diff --git a/tests/unit/pipeline/test_ingest_batch_resource_bounds.py b/tests/unit/pipeline/test_ingest_batch_resource_bounds.py index 923699f4a..60f237cf6 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 2499b364a..97e620109 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_embedding_contracts.py b/tests/unit/storage/test_embedding_contracts.py index 044e9b86f..b305e2222 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 2dcfead28..070a03955 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] diff --git a/tests/unit/storage/test_session_insight_status_descriptors.py b/tests/unit/storage/test_session_insight_status_descriptors.py index 547b127cd..af6473399 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) 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 49543062e..be5f8dfb9 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: