Skip to content

refactor(storage): delete dead session_links cycle-quarantine engine #3655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sinity merged 2 commits into from
Aug 3, 2026
Merged

refactor(storage): delete dead session_links cycle-quarantine engine #3655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8d1d068
refactor(storage): delete dead session_links cycle-quarantine engine
Sinity Aug 3, 2026
6d5f8a1
test(storage): retarget session_links tests off dead cycle engine
Sinity Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
340 changes: 19 additions & 321 deletions polylogue/storage/sqlite/queries/session_links.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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",
]
Loading