diff --git a/polylogue/storage/sqlite/queries/session_links.py b/polylogue/storage/sqlite/queries/session_links.py index 3b46a791c0..7ecb53845d 100644 --- a/polylogue/storage/sqlite/queries/session_links.py +++ b/polylogue/storage/sqlite/queries/session_links.py @@ -1,324 +1,27 @@ -"""SQL helpers for the current ``session_links`` table.""" +"""Read helper for the ``session_links`` table. + +polylogue-4ts.10: this module used to also carry a full async +resolve/cycle-detect/quarantine write engine +(``upsert_session_links``/``resolve_session_links_for_session``/ +``resolve_unresolved_links_for_child``/``_would_create_cycle``/ +``_quarantine_link``/``count_quarantined_session_links``). A 2026-08-03 +structural audit found that engine had zero production callers -- the sole +production writer is ``_resolve_session_graph``/``_resolve_outbound_session_links`` +in ``storage/sqlite/archive_tiers/write.py`` (invoked from +``write_parsed_session_to_archive``), which grew its own equivalent +cycle-detection + quarantine implementation (``_would_create_cycle``/ +``_quarantine_session_link`` there) in #3643. The dead engine was exercised +only by wrong-oracle tests certifying behavior production code could not +exhibit; those tests were retargeted at the live path and the dead engine +was deleted here rather than kept as an unreachable second implementation. +Only ``list_session_links_for_session`` (a genuine production read path via +``query_store_archive.py``) remains. +""" from __future__ import annotations -import json -from collections.abc import Iterable -from datetime import datetime - import aiosqlite -from polylogue.archive.session.branch_type import BranchType -from polylogue.archive.topology.edge import TopologyEdgeRecord, TopologyEdgeStatus - - -def _timestamp_ms(value: str | None) -> int | None: - if not value: - return None - try: - parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) - except ValueError: - return None - return int(parsed.timestamp() * 1000) - - -def _status_value(status: TopologyEdgeStatus | None) -> str | None: - # polylogue-5dfu: TopologyEdgeStatus now only has the two members this - # column ever stores (REPAIRED/QUARANTINED), so no narrowing is needed - # beyond passing the value (or its absence) straight through. - return status.value if status is not None else None - - -async def upsert_session_links( - conn: aiosqlite.Connection, - links: Iterable[TopologyEdgeRecord], -) -> int: - """Upsert session links. Returns number of rows written.""" - written = 0 - for link in links: - await conn.execute( - """ - INSERT INTO session_links ( - src_session_id, - dst_origin, - dst_native_id, - link_type, - resolved_dst_session_id, - status, - parent_tool_use_block_id, - method, - confidence, - evidence_json, - observed_at_ms, - resolved_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT (src_session_id, dst_origin, dst_native_id, link_type) DO UPDATE SET - resolved_dst_session_id = COALESCE( - excluded.resolved_dst_session_id, - session_links.resolved_dst_session_id - ), - status = CASE - WHEN session_links.status = 'quarantined' THEN session_links.status - ELSE excluded.status - END, - parent_tool_use_block_id = COALESCE( - excluded.parent_tool_use_block_id, - session_links.parent_tool_use_block_id - ), - method = COALESCE(excluded.method, session_links.method), - confidence = excluded.confidence, - evidence_json = COALESCE(NULLIF(excluded.evidence_json, '[]'), session_links.evidence_json), - resolved_at_ms = COALESCE(excluded.resolved_at_ms, session_links.resolved_at_ms) - """, - ( - str(link.src_session_id), - link.dst_origin.value, - link.dst_native_id, - str(link.link_type), - str(link.resolved_dst_session_id) if link.resolved_dst_session_id else None, - _status_value(link.status), - link.parent_tool_use_block_id, - "parser-parent" if link.parent_tool_use_block_id is None else "parent-tool-use-id", - link.confidence, - link.evidence_json, - _timestamp_ms(link.observed_at) or 0, - _timestamp_ms(link.resolved_at), - ), - ) - written += 1 - return written - - -_CYCLE_WALK_BUDGET = 1024 - - -async def _would_create_cycle( - conn: aiosqlite.Connection, - *, - child_id: str, - proposed_parent_id: str, -) -> list[str] | None: - if proposed_parent_id == child_id: - return [child_id, child_id] - - path: list[str] = [child_id, proposed_parent_id] - current = proposed_parent_id - steps = 0 - while True: - if steps >= _CYCLE_WALK_BUDGET: - path.append("...budget-exceeded") - return path - row = await ( - await conn.execute( - "SELECT parent_session_id FROM sessions WHERE session_id = ?", - (current,), - ) - ).fetchone() - if row is None: - return None - next_parent = row["parent_session_id"] - if next_parent is None: - return None - if next_parent == child_id: - path.append(child_id) - return path - path.append(next_parent) - current = next_parent - steps += 1 - - -async def _quarantine_link( - conn: aiosqlite.Connection, - *, - src_session_id: str, - dst_origin: str | None, - dst_native_id: str | None, - link_type: str | None, - cycle_path: list[str], - observed_at_ms: int, -) -> None: - evidence = json.dumps( - { - "reason": "cycle_rejected", - "cycle_path": cycle_path, - "detected_at_ms": observed_at_ms, - }, - sort_keys=True, - ) - if dst_origin is None or dst_native_id is None or link_type is None: - await conn.execute( - """ - UPDATE session_links - SET status = 'quarantined', - evidence_json = ?, - resolved_at_ms = ? - WHERE src_session_id = ? - """, - (evidence, observed_at_ms, src_session_id), - ) - return - await conn.execute( - """ - UPDATE session_links - SET status = 'quarantined', - evidence_json = ?, - resolved_at_ms = ? - WHERE src_session_id = ? - AND dst_origin = ? - AND dst_native_id = ? - AND link_type = ? - """, - (evidence, observed_at_ms, src_session_id, dst_origin, dst_native_id, link_type), - ) - - -async def count_quarantined_session_links(conn: aiosqlite.Connection) -> int: - row = await (await conn.execute("SELECT COUNT(*) AS n FROM session_links WHERE status = 'quarantined'")).fetchone() - return int(row["n"]) if row is not None else 0 - - -async def resolve_session_links_for_session( - conn: aiosqlite.Connection, - *, - session_id: str, - origin: str, - native_id: str, - resolved_at: str, -) -> int: - observed_at_ms = _timestamp_ms(resolved_at) or 0 - cursor = await conn.execute( - """ - SELECT src_session_id, link_type - FROM session_links - WHERE resolved_dst_session_id IS NULL - AND status IS NULL - AND dst_origin = ? - AND dst_native_id = ? - """, - (origin, native_id), - ) - pending_rows = list(await cursor.fetchall()) - if not pending_rows: - return 0 - - valid_branch_types: set[str] = {bt.value for bt in BranchType} - flipped = 0 - for row in pending_rows: - child_id = row["src_session_id"] - link_type = row["link_type"] - - cycle_path = await _would_create_cycle(conn, child_id=child_id, proposed_parent_id=session_id) - if cycle_path is not None: - await _quarantine_link( - conn, - src_session_id=child_id, - dst_origin=origin, - dst_native_id=native_id, - link_type=link_type, - cycle_path=cycle_path, - observed_at_ms=observed_at_ms, - ) - continue - - await conn.execute( - """ - UPDATE session_links - SET resolved_dst_session_id = ?, - resolved_at_ms = ? - WHERE resolved_dst_session_id IS NULL - AND status IS NULL - AND src_session_id = ? - AND dst_origin = ? - AND dst_native_id = ? - AND link_type = ? - """, - (session_id, observed_at_ms, child_id, origin, native_id, link_type), - ) - flipped += 1 - branch_type: str | None = link_type if link_type in valid_branch_types else None - await conn.execute( - """ - UPDATE sessions - SET parent_session_id = COALESCE(parent_session_id, ?), - branch_type = COALESCE(branch_type, ?) - WHERE session_id = ? - """, - (session_id, branch_type, child_id), - ) - - return flipped - - -async def resolve_unresolved_links_for_child( - conn: aiosqlite.Connection, - *, - src_session_id: str, - resolved_at: str, -) -> int: - observed_at_ms = _timestamp_ms(resolved_at) or 0 - cursor = await conn.execute( - """ - SELECT link.link_type, link.dst_origin, link.dst_native_id, dst.session_id AS parent_id - FROM session_links AS link - JOIN sessions AS dst - ON dst.origin = link.dst_origin - AND dst.native_id = link.dst_native_id - WHERE link.src_session_id = ? - AND link.resolved_dst_session_id IS NULL - AND link.status IS NULL - """, - (src_session_id,), - ) - rows = list(await cursor.fetchall()) - if not rows: - return 0 - - valid_branch_types: set[str] = {bt.value for bt in BranchType} - resolved = 0 - for row in rows: - link_type = row["link_type"] - parent_id = row["parent_id"] - - cycle_path = await _would_create_cycle(conn, child_id=src_session_id, proposed_parent_id=parent_id) - if cycle_path is not None: - await _quarantine_link( - conn, - src_session_id=src_session_id, - dst_origin=row["dst_origin"], - dst_native_id=row["dst_native_id"], - link_type=link_type, - cycle_path=cycle_path, - observed_at_ms=observed_at_ms, - ) - continue - - await conn.execute( - """ - UPDATE session_links - SET resolved_dst_session_id = ?, - resolved_at_ms = ? - WHERE src_session_id = ? - AND dst_origin = ? - AND dst_native_id = ? - AND link_type = ? - AND resolved_dst_session_id IS NULL - AND status IS NULL - """, - (parent_id, observed_at_ms, src_session_id, row["dst_origin"], row["dst_native_id"], link_type), - ) - branch_type: str | None = link_type if link_type in valid_branch_types else None - await conn.execute( - """ - UPDATE sessions - SET parent_session_id = COALESCE(parent_session_id, ?), - branch_type = COALESCE(branch_type, ?) - WHERE session_id = ? - """, - (parent_id, branch_type, src_session_id), - ) - resolved += 1 - return resolved - async def list_session_links_for_session( conn: aiosqlite.Connection, @@ -340,10 +43,5 @@ async def list_session_links_for_session( __all__ = [ - "TopologyEdgeStatus", - "count_quarantined_session_links", "list_session_links_for_session", - "resolve_session_links_for_session", - "resolve_unresolved_links_for_child", - "upsert_session_links", ] diff --git a/tests/property/test_write_path_state_machine.py b/tests/property/test_write_path_state_machine.py index 34373ed8fd..ae3c15851e 100644 --- a/tests/property/test_write_path_state_machine.py +++ b/tests/property/test_write_path_state_machine.py @@ -8,13 +8,13 @@ from __future__ import annotations import asyncio +import json import sqlite3 import tempfile from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path -import aiosqlite from hypothesis import HealthCheck, settings from hypothesis.stateful import RuleBasedStateMachine, initialize, rule @@ -30,8 +30,6 @@ read_archive_session_envelope, write_parsed_session_to_archive, ) -from polylogue.storage.sqlite.async_sqlite import configure_connection -from polylogue.storage.sqlite.queries.session_links import resolve_unresolved_links_for_child from polylogue.storage.sqlite.schema import _ensure_schema @@ -616,14 +614,21 @@ async def read_texts(session_id: str) -> list[str]: def test_session_link_resolver_quarantines_cycle() -> None: - """The async resolver quarantines a late link that would close a cycle.""" + """The live write path (``write_parsed_session_to_archive`` -> + ``_resolve_outbound_session_links``) quarantines a late link that would + close a cycle, instead of the dead async engine at + ``storage/sqlite/queries/session_links.py`` (zero production callers, + deleted for polylogue-4ts.10; see + ``tests/unit/storage/test_topology_cycle_quarantine_live.py`` for the + fuller cross-ingest/self-loop/diamond-DAG coverage of this same live + path).""" with tempfile.TemporaryDirectory(prefix="polylogue-write-model-", dir="/realm/tmp") as root_text: archive_root = Path(root_text) initialize_active_archive_root(archive_root) db_path = archive_root / "index.db" conn = sqlite3.connect(str(db_path)) try: - parent = ParsedSession( + parent_v1 = ParsedSession( source_name=Provider.CLAUDE_CODE, provider_session_id="cycle-parent", messages=[ParsedMessage(provider_message_id="parent-0", role=Role.USER, text="parent", position=0)], @@ -635,33 +640,28 @@ def test_session_link_resolver_quarantines_cycle() -> None: branch_type=BranchType.FORK, messages=[ParsedMessage(provider_message_id="child-0", role=Role.USER, text="child", position=0)], ) - parent_id = write_parsed_session_to_archive(conn, parent, content_hash=session_content_hash(parent)) + parent_id = write_parsed_session_to_archive(conn, parent_v1, content_hash=session_content_hash(parent_v1)) write_parsed_session_to_archive(conn, child, content_hash=session_content_hash(child)) - conn.execute( - """ - INSERT INTO session_links ( - src_session_id, dst_origin, dst_native_id, link_type, - status, method, confidence, evidence_json, observed_at_ms - ) VALUES (?, 'claude-code-session', 'cycle-child', 'fork', NULL, 'test', 1.0, '[]', 0) - """, - (parent_id,), + + # Re-ingest the parent now claiming the child as ITS parent -- + # closing a two-node cycle parent -> child -> parent. This must + # be rejected (quarantined), not silently resolved. + parent_v2 = ParsedSession( + source_name=Provider.CLAUDE_CODE, + provider_session_id="cycle-parent", + parent_session_provider_id="cycle-child", + messages=[ + ParsedMessage(provider_message_id="parent-0", role=Role.USER, text="parent", position=0), + ParsedMessage(provider_message_id="parent-1", role=Role.ASSISTANT, text="revised", position=1), + ], + ) + write_parsed_session_to_archive( + conn, parent_v2, content_hash=session_content_hash(parent_v2), force_replace=True ) conn.commit() finally: conn.close() - async def resolve_cycle() -> int: - async with aiosqlite.connect(db_path) as async_conn: - await configure_connection(async_conn) - resolved = await resolve_unresolved_links_for_child( - async_conn, - src_session_id=parent_id, - resolved_at="2026-01-01T00:00:00Z", - ) - await async_conn.commit() - return resolved - - assert asyncio.run(resolve_cycle()) == 0 with sqlite3.connect(str(db_path)) as verify_conn: row = verify_conn.execute( "SELECT resolved_dst_session_id, status, evidence_json FROM session_links WHERE src_session_id = ?", @@ -670,7 +670,7 @@ async def resolve_cycle() -> int: assert row is not None assert row[0] is None assert row[1] == TopologyEdgeStatus.QUARANTINED.value - assert '"reason": "cycle_rejected"' in row[2] + assert json.loads(row[2])["reason"] == "cycle_rejected" TestWritePathStateMachine = WritePathStateMachine.TestCase diff --git a/tests/unit/insights/test_topology_cycle_rejection.py b/tests/unit/insights/test_topology_cycle_rejection.py deleted file mode 100644 index 94ed2e6b8e..0000000000 --- a/tests/unit/insights/test_topology_cycle_rejection.py +++ /dev/null @@ -1,621 +0,0 @@ -"""Topology cycle rejection and quarantine (#1260 / #866 slice C). - -When resolving a topology edge would create a cycle in -``sessions.parent_session_id`` (A → B → A, longer cycles, or -the self-cycle A → A), the resolver must: - -- Refuse to backfill ``parent_session_id`` (the fast-path graph - must never enter the cycle). -- Refuse to resolve the link; instead, mark it - ``status='quarantined'`` with an ``evidence_json`` document that - records the detected cycle path so the operator can audit which - session chain was rejected. -- Leave non-cyclic edges in the same resolver batch untouched (a cycle - on one child does not poison legitimate siblings). -- Stay idempotent: re-running the resolver on an archive that already - contains quarantined edges produces no further changes. - -These tests exercise both resolver entry points -(:func:`resolve_session_links_for_session` and -:func:`resolve_unresolved_links_for_child`) and also assert the -no-false-positive case on a diamond DAG, which is a legitimate shape -(B→D, C→D both pointing at D) that some prior implementations confuse -with a cycle. -""" - -from __future__ import annotations - -import json -import sqlite3 -from collections.abc import Iterator -from pathlib import Path - -import pytest - -from polylogue.archive.topology.edge import ( - TopologyEdgeRecord, - TopologyEdgeStatus, - TopologyEdgeType, -) -from polylogue.core.enums import Origin -from polylogue.core.types import SessionId -from polylogue.storage.sqlite.queries.session_links import ( - count_quarantined_session_links, - resolve_session_links_for_session, - resolve_unresolved_links_for_child, - upsert_session_links, -) -from polylogue.storage.sqlite.schema import SCHEMA_DDL, SCHEMA_VERSION -from tests.infra.frozen_clock import fixed_now - - -def _now() -> str: - return fixed_now().isoformat() - - -def _sid(value: str) -> str: - if value.startswith("codex-session:"): - return value - if value.startswith("conv-"): - return f"codex-session:native-{value.removeprefix('conv-')}" - return value - - -def _hash_blob(value: str) -> bytes: - import hashlib - - return hashlib.sha256(value.encode("utf-8")).digest() - - -def _bootstrap_db(path: Path) -> None: - with sqlite3.connect(path) as conn: - conn.executescript(SCHEMA_DDL) - conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") - conn.commit() - - -def _insert_session( - conn: sqlite3.Connection, - *, - session_id: str, - source_name: str, - provider_session_id: str, - parent_session_id: str | None = None, -) -> None: - conn.execute( - """ - INSERT INTO sessions ( - native_id, origin, title, parent_session_id, - content_hash, created_at_ms, updated_at_ms - ) VALUES (?, ?, ?, ?, ?, ?, ?) - """, - ( - provider_session_id, - Origin.CODEX_SESSION.value, - f"conv {session_id}", - _sid(parent_session_id) if parent_session_id else None, - _hash_blob(session_id), - 1, - 1, - ), - ) - conn.commit() - - -class _AsyncSqliteAdapter: - """Minimal aiosqlite-compatible wrapper around stdlib sqlite3. - - The production query helpers are written against ``aiosqlite``; the - only async operations they perform are ``execute``/``fetchone``/ - ``fetchall``. This adapter lets the unit tests drive them against a - plain sqlite3 connection without spinning up the full async runtime. - """ - - def __init__(self, conn: sqlite3.Connection) -> None: - conn.row_factory = sqlite3.Row - self._conn = conn - - async def execute(self, sql: str, params: tuple[object, ...] = ()) -> _AsyncCursorAdapter: - cursor = self._conn.execute(sql, params) - return _AsyncCursorAdapter(cursor) - - def commit(self) -> None: - self._conn.commit() - - -class _AsyncCursorAdapter: - def __init__(self, cursor: sqlite3.Cursor) -> None: - self._cursor = cursor - - async def fetchall(self) -> list[sqlite3.Row]: - return list(self._cursor.fetchall()) - - async def fetchone(self) -> sqlite3.Row | None: - result: sqlite3.Row | None = self._cursor.fetchone() - return result - - @property - def rowcount(self) -> int: - return self._cursor.rowcount - - -@pytest.fixture -def cycle_db(tmp_path: Path) -> Iterator[sqlite3.Connection]: - db_path = tmp_path / "cycle.sqlite" - _bootstrap_db(db_path) - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - try: - yield conn - finally: - conn.close() - - -def _seed_edge( - conn: sqlite3.Connection, - *, - src_session_id: str, - dst_native_id: str, - dst_origin: Origin = Origin.CODEX_SESSION, - link_type: TopologyEdgeType = TopologyEdgeType.CONTINUATION, -) -> None: - """Insert one unresolved session link through the production upsert.""" - - import asyncio - - edge = TopologyEdgeRecord( - src_session_id=SessionId(_sid(src_session_id)), - dst_origin=dst_origin, - dst_native_id=dst_native_id, - link_type=link_type, - ) - asyncio.run(upsert_session_links(_AsyncSqliteAdapter(conn), [edge])) # type: ignore[arg-type] - - -def _run_resolve_for_parent( - conn: sqlite3.Connection, - *, - session_id: str, - origin: str, - native_id: str, -) -> int: - import asyncio - - return asyncio.run( - resolve_session_links_for_session( - _AsyncSqliteAdapter(conn), # type: ignore[arg-type] - session_id=_sid(session_id), - origin=origin, - native_id=native_id, - resolved_at=_now(), - ) - ) - - -def _run_resolve_for_child(conn: sqlite3.Connection, *, src_session_id: str) -> int: - import asyncio - - return asyncio.run( - resolve_unresolved_links_for_child( - _AsyncSqliteAdapter(conn), # type: ignore[arg-type] - src_session_id=_sid(src_session_id), - resolved_at=_now(), - ) - ) - - -def _fetch_edge_status(conn: sqlite3.Connection, src_session_id: str) -> tuple[str, str | None]: - row = conn.execute( - "SELECT status, evidence_json FROM session_links WHERE src_session_id = ?", - (_sid(src_session_id),), - ).fetchone() - assert row is not None - # "resolved"/"unresolved" here are test-only derived labels (polylogue-5dfu: - # TopologyEdgeStatus no longer has UNRESOLVED/RESOLVED members -- neither - # was ever storable in this column, since resolvedness is already carried - # by resolved_dst_session_id) computed the same way production code - # derives it: a NULL status column plus a resolved parent means resolved. - status = row["status"] - if status is None and row["evidence_json"] == "[]": - status = "resolved" if _fetch_parent(conn, src_session_id) else "unresolved" - return str(status), row["evidence_json"] - - -def _fetch_parent(conn: sqlite3.Connection, session_id: str) -> str | None: - row = conn.execute( - "SELECT parent_session_id FROM sessions WHERE session_id = ?", - (_sid(session_id),), - ).fetchone() - return None if row is None else row["parent_session_id"] - - -class TestTwoNodeCycle: - """A → B → A (B was already saved with A as parent; A is now landing - with a topology edge that would make B its parent).""" - - def test_two_node_cycle_quarantines_edge_and_leaves_parent_null(self, cycle_db: sqlite3.Connection) -> None: - # Topology: A's parent_session_id will eventually be set to B, - # but B already has A as its parent. Resolving A→B is the cycle. - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", # B → A already. - ) - # A asserts an unresolved edge to B (this would normally flip to - # resolved when B's row was already in place). - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-B", - ) - - # The resolver-by-child code path is what runs when A's own edge - # is upserted and we then sweep for matching parent rows. - _run_resolve_for_child(cycle_db, src_session_id="conv-A") - - status, evidence = _fetch_edge_status(cycle_db, "conv-A") - assert status == TopologyEdgeStatus.QUARANTINED.value - assert evidence is not None - payload = json.loads(evidence) - assert payload["reason"] == "cycle_rejected" - # The cycle path must include both endpoints. - assert _sid("conv-A") in payload["cycle_path"] - assert _sid("conv-B") in payload["cycle_path"] - - # And critically, A.parent_session_id stays NULL so the - # fast-path ancestry walk does not enter the cycle. - assert _fetch_parent(cycle_db, "conv-A") is None - - def test_two_node_cycle_via_parent_first_path(self, cycle_db: sqlite3.Connection) -> None: - # Same topology but exercised through the parent-first entry point - # (parent X is being saved; we look for unresolved edges pointing at X). - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", - ) - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-B", - ) - - # B is now being "saved" (already present in sessions from - # the fixture), and we sweep edges that point at native-B. Without - # cycle detection this would resolve the edge and backfill - # A.parent_session_id = B, closing the loop. - flipped = _run_resolve_for_parent( - cycle_db, - session_id="conv-B", - origin=Origin.CODEX_SESSION.value, - native_id="native-B", - ) - assert flipped == 0 # nothing was successfully resolved - - status, evidence = _fetch_edge_status(cycle_db, "conv-A") - assert status == TopologyEdgeStatus.QUARANTINED.value - assert evidence is not None - assert _fetch_parent(cycle_db, "conv-A") is None - - -class TestThreeNodeCycle: - """A → B → C → A. C is being saved; resolving A→C would close the loop.""" - - def test_three_node_cycle_quarantined(self, cycle_db: sqlite3.Connection) -> None: - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", - ) - _insert_session( - cycle_db, - session_id="conv-C", - source_name="codex", - provider_session_id="native-C", - parent_session_id="conv-B", - ) - # A asserts unresolved edge to C → resolving would mean - # A.parent = C → B → A → cycle. - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-C", - ) - - _run_resolve_for_child(cycle_db, src_session_id="conv-A") - - status, evidence = _fetch_edge_status(cycle_db, "conv-A") - assert status == TopologyEdgeStatus.QUARANTINED.value - payload = json.loads(evidence or "{}") - # The recorded path should walk A → C → B → A. - assert payload["cycle_path"][0] == _sid("conv-A") - assert payload["cycle_path"][-1] == _sid("conv-A") - assert _sid("conv-B") in payload["cycle_path"] - assert _sid("conv-C") in payload["cycle_path"] - assert _fetch_parent(cycle_db, "conv-A") is None - - -class TestSelfCycle: - """A → A. The most pathological cycle shape.""" - - def test_self_cycle_quarantined(self, cycle_db: sqlite3.Connection) -> None: - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - # A's edge points at its own native id. - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-A", - ) - - flipped = _run_resolve_for_parent( - cycle_db, - session_id="conv-A", - origin=Origin.CODEX_SESSION.value, - native_id="native-A", - ) - assert flipped == 0 - - status, evidence = _fetch_edge_status(cycle_db, "conv-A") - assert status == TopologyEdgeStatus.QUARANTINED.value - payload = json.loads(evidence or "{}") - assert payload["cycle_path"] == [_sid("conv-A"), _sid("conv-A")] - assert _fetch_parent(cycle_db, "conv-A") is None - - -class TestDiamondDagNoFalsePositive: - """B → D and C → D — D is a shared parent for two children. This is a - legitimate diamond DAG, not a cycle, and the resolver must process it - cleanly without quarantining either edge.""" - - def test_diamond_dag_resolves_both_children(self, cycle_db: sqlite3.Connection) -> None: - _insert_session( - cycle_db, - session_id="conv-D", - source_name="codex", - provider_session_id="native-D", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-C", - source_name="codex", - provider_session_id="native-C", - parent_session_id=None, - ) - _seed_edge( - cycle_db, - src_session_id="conv-B", - dst_native_id="native-D", - ) - _seed_edge( - cycle_db, - src_session_id="conv-C", - dst_native_id="native-D", - ) - - flipped = _run_resolve_for_parent( - cycle_db, - session_id="conv-D", - origin=Origin.CODEX_SESSION.value, - native_id="native-D", - ) - assert flipped == 2 # both edges resolved - - for child in ("conv-B", "conv-C"): - status, _ = _fetch_edge_status(cycle_db, child) - assert status == "resolved" - assert _fetch_parent(cycle_db, child) == _sid("conv-D") - - -class TestSiblingNotQuarantined: - """A cycle on one child must not quarantine a non-cyclic sibling that - happens to share the resolver batch (both pointing at the same parent - native id).""" - - def test_cycle_on_one_child_leaves_sibling_resolved(self, cycle_db: sqlite3.Connection) -> None: - # X is the parent native id. Two children share it: - # - cyclic: X is descendant of cycle-child, so resolving creates cycle. - # - clean: independent child with no ancestry conflict. - _insert_session( - cycle_db, - session_id="conv-cycle", - source_name="codex", - provider_session_id="native-cycle", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-X", - source_name="codex", - provider_session_id="native-X", - parent_session_id="conv-cycle", # X already descends from cycle-child - ) - _insert_session( - cycle_db, - session_id="conv-clean", - source_name="codex", - provider_session_id="native-clean", - parent_session_id=None, - ) - _seed_edge( - cycle_db, - src_session_id="conv-cycle", - dst_native_id="native-X", - ) - _seed_edge( - cycle_db, - src_session_id="conv-clean", - dst_native_id="native-X", - ) - - flipped = _run_resolve_for_parent( - cycle_db, - session_id="conv-X", - origin=Origin.CODEX_SESSION.value, - native_id="native-X", - ) - # Only the clean child resolved; the cyclic one was quarantined. - assert flipped == 1 - - cycle_status, _ = _fetch_edge_status(cycle_db, "conv-cycle") - clean_status, _ = _fetch_edge_status(cycle_db, "conv-clean") - assert cycle_status == TopologyEdgeStatus.QUARANTINED.value - assert clean_status == "resolved" - assert _fetch_parent(cycle_db, "conv-cycle") is None - assert _fetch_parent(cycle_db, "conv-clean") == _sid("conv-X") - - -class TestIdempotency: - """Re-running the resolver after quarantine produces no further state - changes — the edge stays quarantined, the sessions row stays - untouched.""" - - def test_resolver_idempotent_on_quarantined_edge(self, cycle_db: sqlite3.Connection) -> None: - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", - ) - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-B", - ) - - # First pass quarantines. - _run_resolve_for_parent( - cycle_db, - session_id="conv-B", - origin=Origin.CODEX_SESSION.value, - native_id="native-B", - ) - status1, evidence1 = _fetch_edge_status(cycle_db, "conv-A") - assert status1 == TopologyEdgeStatus.QUARANTINED.value - - # Re-running the parent resolver: the edge is no longer - # ``unresolved``, so the SELECT returns no candidates and nothing - # changes. No spurious second quarantine event, no parent backfill. - _run_resolve_for_parent( - cycle_db, - session_id="conv-B", - origin=Origin.CODEX_SESSION.value, - native_id="native-B", - ) - status2, evidence2 = _fetch_edge_status(cycle_db, "conv-A") - assert status2 == TopologyEdgeStatus.QUARANTINED.value - assert evidence2 == evidence1 # same payload, untouched - assert _fetch_parent(cycle_db, "conv-A") is None - - def test_resolver_for_child_idempotent_on_quarantined_edge(self, cycle_db: sqlite3.Connection) -> None: - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", - ) - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-B", - ) - - _run_resolve_for_child(cycle_db, src_session_id="conv-A") - _run_resolve_for_child(cycle_db, src_session_id="conv-A") - - status, _ = _fetch_edge_status(cycle_db, "conv-A") - assert status == TopologyEdgeStatus.QUARANTINED.value - assert _fetch_parent(cycle_db, "conv-A") is None - - -class TestQuarantineCount: - """The diagnostic counter surfaces quarantined edges so the daemon - workload probe can warn the operator.""" - - def test_count_quarantined_session_links(self, cycle_db: sqlite3.Connection) -> None: - import asyncio - - # Start with zero. - count0 = asyncio.run(count_quarantined_session_links(_AsyncSqliteAdapter(cycle_db))) # type: ignore[arg-type] - assert count0 == 0 - - # Create a cycle. - _insert_session( - cycle_db, - session_id="conv-A", - source_name="codex", - provider_session_id="native-A", - parent_session_id=None, - ) - _insert_session( - cycle_db, - session_id="conv-B", - source_name="codex", - provider_session_id="native-B", - parent_session_id="conv-A", - ) - _seed_edge( - cycle_db, - src_session_id="conv-A", - dst_native_id="native-B", - ) - _run_resolve_for_child(cycle_db, src_session_id="conv-A") - - count1 = asyncio.run(count_quarantined_session_links(_AsyncSqliteAdapter(cycle_db))) # type: ignore[arg-type] - assert count1 == 1 diff --git a/tests/unit/storage/test_delegations_view.py b/tests/unit/storage/test_delegations_view.py index aa1f47cc4f..a9ef746f6e 100644 --- a/tests/unit/storage/test_delegations_view.py +++ b/tests/unit/storage/test_delegations_view.py @@ -2,15 +2,15 @@ parent-dispatched subagent attempt from the PARENT's own dispatch actions (`actions` rows, semantic_type='subagent'), corroborated against resolved children via canonical `session_links` (child in `src_session_id`, parent in -`resolved_dst_session_id` -- see resolve_session_links_for_session). The -prior shipped view aliased these backwards; these fixtures use the canonical -direction throughout and would fail against that reversed view. Model -identity is separated into dispatch-turn / requested / child-observed / -session-dominant-fallback columns rather than one "orchestrator model".""" +`resolved_dst_session_id` -- see ``_resolve_outbound_session_links``, +``storage/sqlite/archive_tiers/write.py``). The prior shipped view aliased +these backwards; these fixtures use the canonical direction throughout and +would fail against that reversed view. Model identity is separated into +dispatch-turn / requested / child-observed / session-dominant-fallback +columns rather than one "orchestrator model".""" from __future__ import annotations -import asyncio import hashlib import json import sqlite3 @@ -18,18 +18,15 @@ import pytest +from polylogue.archive.message.roles import Role from polylogue.archive.query.unit_results import query_unit_envelope, query_unit_request -from polylogue.archive.topology.edge import TopologyEdgeRecord -from polylogue.core.enums import LinkType as TopologyEdgeType -from polylogue.core.enums import Origin -from polylogue.core.types import SessionId +from polylogue.core.enums import BranchType, Provider +from polylogue.pipeline.ids import session_content_hash +from polylogue.sources.parsers.base import ParsedMessage, ParsedSession from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database, initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier -from polylogue.storage.sqlite.queries.session_links import ( - resolve_session_links_for_session, - upsert_session_links, -) +from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive from polylogue.surfaces.payloads import DelegationCardPayload, QueryUnitAggregateRowPayload _HASH = b"x" * 32 @@ -164,10 +161,10 @@ def _insert_session_link( ) -> None: """Canonical direction: the CHILD asserts the link (src_session_id), the PARENT is the resolved destination -- matching - resolve_session_links_for_session, where `child_id = - row["src_session_id"]` and the resolved session is written into - `sessions.parent_session_id` keyed by that child. This is the reverse of - the pre-y964 test fixtures, which is exactly the bug: those fixtures + ``_resolve_outbound_session_links`` (``storage/sqlite/archive_tiers/write.py``), + where `child_id = row["src_session_id"]` and the resolved session is + written into `sessions.parent_session_id` keyed by that child. This is + the reverse of the pre-y964 test fixtures, which is exactly the bug: those fixtures matched the (wrong) shipped view, not real ingestion.""" conn.execute( """ @@ -189,40 +186,6 @@ def _insert_session_link( ) -class _AsyncSqliteAdapter: - """Minimal aiosqlite-compatible wrapper around stdlib sqlite3, matching - the pattern in tests/unit/insights/test_topology_cycle_rejection.py -- - lets a real production async query helper run against a plain sqlite3 - connection inside a synchronous test.""" - - def __init__(self, conn: sqlite3.Connection) -> None: - conn.row_factory = sqlite3.Row - self._conn = conn - - async def execute(self, sql: str, params: tuple[object, ...] = ()) -> _AsyncCursorAdapter: - cursor = self._conn.execute(sql, params) - return _AsyncCursorAdapter(cursor) - - def commit(self) -> None: - self._conn.commit() - - -class _AsyncCursorAdapter: - def __init__(self, cursor: sqlite3.Cursor) -> None: - self._cursor = cursor - - async def fetchall(self) -> list[sqlite3.Row]: - return list(self._cursor.fetchall()) - - async def fetchone(self) -> sqlite3.Row | None: - result: sqlite3.Row | None = self._cursor.fetchone() - return result - - @property - def rowcount(self) -> int: - return self._cursor.rowcount - - def test_delegation_resolves_with_canonical_child_to_parent_direction(tmp_path: Path) -> None: conn = _connect(tmp_path / "index.db") @@ -625,44 +588,48 @@ def test_delegation_requested_model_unknown_when_not_recorded(tmp_path: Path) -> def test_delegation_direction_matches_real_link_resolver(tmp_path: Path) -> None: """Real-route regression: drive the ACTUAL production write path - (upsert_session_links / resolve_session_links_for_session -- the same - functions the daemon calls after parsing a session) instead of a + (``write_parsed_session_to_archive`` -> ``_resolve_outbound_session_links``, + ``storage/sqlite/archive_tiers/write.py`` -- the sole production writer, + the same path the daemon calls after parsing a session) instead of a hand-built row shape, then confirm the view reads parent/child in the correct direction against whatever the real resolver produced. This is - the test that would fail outright against the pre-y964 reversed view.""" + the test that would fail outright against the pre-y964 reversed view. + + polylogue-4ts.10: previously drove ``queries/session_links.py``'s + ``upsert_session_links``/``resolve_session_links_for_session`` under the + same "ACTUAL production write path" claim -- a 2026-08-03 structural + audit found that engine has zero production callers and was deleted; + this test now drives the real writer directly.""" conn = _connect(tmp_path / "index.db") parent_id = _insert_session(conn, native_id="parent", origin="codex-session") - child_id = _insert_session(conn, native_id="child", origin="codex-session") dispatch_message_id = _insert_message(conn, session_id=parent_id, native_id="dispatch", position=0) _insert_dispatch_action(conn, message_id=dispatch_message_id, session_id=parent_id, position=0, tool_id="task-1") + conn.commit() - adapter = _AsyncSqliteAdapter(conn) # The CHILD is the one that asserts the (initially unresolved) link to # its parent -- mirroring what a real subagent-session parser does. - edge = TopologyEdgeRecord( - src_session_id=SessionId(child_id), - dst_origin=Origin.CODEX_SESSION, - dst_native_id="parent", - link_type=TopologyEdgeType.SUBAGENT, - ) - asyncio.run(upsert_session_links(adapter, [edge])) # type: ignore[arg-type] - resolved_count = asyncio.run( - resolve_session_links_for_session( - adapter, # type: ignore[arg-type] - session_id=parent_id, - origin="codex-session", - native_id="parent", - resolved_at="2026-07-10T00:00:00+00:00", - ) - ) + child_session = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="child", + parent_session_provider_id="parent", + branch_type=BranchType.SUBAGENT, + messages=[ParsedMessage(provider_message_id="c0", role=Role.USER, text="go", position=0)], + ) + child_id = write_parsed_session_to_archive(conn, child_session, content_hash=session_content_hash(child_session)) conn.commit() - assert resolved_count == 1 # The resolver must have written the PARENT into sessions.parent_session_id # keyed by the CHILD -- confirming our fixture direction matches reality. sessions_row = conn.execute("SELECT parent_session_id FROM sessions WHERE session_id = ?", (child_id,)).fetchone() assert sessions_row["parent_session_id"] == parent_id + link_row = conn.execute( + "SELECT status, resolved_dst_session_id FROM session_links WHERE src_session_id = ?", (child_id,) + ).fetchone() + assert link_row is not None + assert link_row["status"] is None + assert link_row["resolved_dst_session_id"] == parent_id + row = conn.execute("SELECT * FROM delegations WHERE parent_session_id = ?", (parent_id,)).fetchone() assert row is not None assert row["parent_session_id"] == parent_id