diff --git a/devtools/docs_surface.py b/devtools/docs_surface.py index 409b594a98..c91e49d9db 100644 --- a/devtools/docs_surface.py +++ b/devtools/docs_surface.py @@ -235,6 +235,12 @@ def _entry(title: str, path: str, description: str, tier: DocsTier) -> DocsEntry "Privacy-safe read-only census of cursor and accepted-head readiness evidence.", "evidence", ), + _entry( + "Topology Live-Proof Residue, 2026-08-06", + "evidence/polylogue-topology-live-proof-2026-08-06.md", + "Candidate topology census, production-route cycle evidence, and unexercised live-archive residue.", + "evidence", + ), _entry( "Proof Artifacts", "proof-artifacts.md", diff --git a/devtools/lineage_validation.py b/devtools/lineage_validation.py index 36c08e55a7..4252dfc456 100644 --- a/devtools/lineage_validation.py +++ b/devtools/lineage_validation.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +import hashlib import json import sys from collections.abc import Iterable @@ -18,6 +19,11 @@ SUPPORTED_PREFIX_ORIGINS = frozenset({"codex-session", "claude-code-session"}) REQUIRED_SESSION_LINK_COLUMNS = frozenset({"branch_point_message_id", "inheritance"}) +REQUIRED_TOPOLOGY_LINK_COLUMNS = frozenset( + {"dst_native_id", "evidence_json", "link_type", "method", "resolved_dst_session_id", "status"} +) +TOPOLOGY_EFFECTIVE_STATES = frozenset({"resolved", "unresolved", "repaired", "quarantined"}) +_SNAPSHOT_HASH_CHUNK_BYTES = 1024 * 1024 @dataclass(frozen=True, slots=True) @@ -27,6 +33,44 @@ class LineageValidationArgs: sample_prefix_sharing: int max_sample_stored_messages: int json: bool + sample_unresolved: int = 20 + index_db: Path | None = None + + +def _snapshot_identity(index_db: Path) -> dict[str, Any]: + """Describe the database files that make up one read-only index snapshot.""" + paths = [index_db, Path(f"{index_db}-wal"), Path(f"{index_db}-shm"), Path(f"{index_db}-journal")] + files: list[dict[str, Any]] = [] + for path in paths: + if not path.is_file(): + files.append({"path": str(path), "present": False}) + continue + stat = path.stat() + digest = _file_sha256(path) + files.append( + { + "path": str(path), + "present": True, + "size": stat.st_size, + "mtime_ns": stat.st_mtime_ns, + "inode": stat.st_ino, + "sha256": digest, + } + ) + encoded = json.dumps(files, sort_keys=True, separators=(",", ":")).encode("utf-8") + return { + "index_db": str(index_db), + "files": files, + "sha256": hashlib.sha256(encoded).hexdigest(), + } + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(_SNAPSHOT_HASH_CHUNK_BYTES), b""): + digest.update(chunk) + return digest.hexdigest() def _parser() -> argparse.ArgumentParser: @@ -52,6 +96,18 @@ def _parser() -> argparse.ArgumentParser: ), ) parser.add_argument("--json", action="store_true", help="Emit JSON report to stdout.") + parser.add_argument( + "--sample-unresolved", + type=int, + default=20, + help="Number of unresolved-parent rows to exercise through the archive read seam.", + ) + parser.add_argument( + "--index-db", + type=Path, + default=None, + help="Read a specific candidate/live index database instead of /index.db.", + ) return parser @@ -106,6 +162,34 @@ def _table_columns(conn: Connection, table: str) -> set[str]: return {str(row[1]) for row in conn.execute(f"PRAGMA table_info({table})").fetchall()} +def _quarantine_evidence_counts(conn: Connection) -> tuple[int, int]: + """Count valid cycle evidence and malformed quarantine evidence separately.""" + cycle_evidence_count = 0 + malformed_count = 0 + rows = conn.execute("SELECT evidence_json FROM session_links WHERE TRIM(status) = 'quarantined'").fetchall() + for (raw_evidence,) in rows: + try: + evidence = json.loads(raw_evidence) + except (TypeError, ValueError): + malformed_count += 1 + continue + cycle_path = evidence.get("cycle_path") if isinstance(evidence, dict) else None + detected_at_ms = evidence.get("detected_at_ms") if isinstance(evidence, dict) else None + if ( + isinstance(evidence, dict) + and evidence.get("reason") == "cycle_rejected" + and isinstance(cycle_path, list) + and len(cycle_path) >= 2 + and all(isinstance(session_id, str) and session_id.strip() for session_id in cycle_path) + and isinstance(detected_at_ms, int) + and not isinstance(detected_at_ms, bool) + ): + cycle_evidence_count += 1 + else: + malformed_count += 1 + return cycle_evidence_count, malformed_count + + def _logical_session_count(conn: Connection) -> int: return _scalar_int( conn, @@ -157,6 +241,209 @@ def _lineage_counts(conn: Connection) -> dict[str, Any]: } +def _topology_read_sample(conn: Connection, *, limit: int) -> dict[str, Any]: + """Exercise unresolved-parent rows through the production composition seam. + + An unresolved edge must remain a child-local read. The parent pointer is + retained in ``session_links`` for later repair, but the archive envelope + must not recurse into a parent that was not resolved. This uses + ``read_archive_session_envelope`` itself, rather than duplicating its + composition query in the census. + """ + if limit < 0: + raise ValueError("--sample-unresolved must be non-negative") + unresolved_count = _scalar_int( + conn, + """ + SELECT COUNT(*) + FROM session_links + WHERE resolved_dst_session_id IS NULL + AND COALESCE(NULLIF(TRIM(status), ''), 'unresolved') = 'unresolved' + """, + ) + rows = _rows( + conn, + """ + SELECT l.src_session_id AS session_id, l.dst_native_id AS parent_native_id, + COUNT(m.message_id) AS stored_messages + FROM session_links l + LEFT JOIN messages m ON m.session_id = l.src_session_id + WHERE l.resolved_dst_session_id IS NULL + AND COALESCE(NULLIF(TRIM(l.status), ''), 'unresolved') = 'unresolved' + GROUP BY l.src_session_id, l.dst_native_id + ORDER BY l.src_session_id, l.dst_native_id + LIMIT ? + """, + (limit,), + ) + samples: list[dict[str, Any]] = [] + errors: list[dict[str, str]] = [] + for row in rows: + session_id = str(row["session_id"]) + stored_messages = _int(row["stored_messages"]) + try: + envelope = read_archive_session_envelope(conn, session_id) + except Exception as exc: # pragma: no cover - defensive for live artifacts + errors.append({"session_id": session_id, "error": f"{type(exc).__name__}: {exc}"}) + samples.append({**row, "read_status": "error", "error": f"{type(exc).__name__}: {exc}"}) + continue + served_messages = len(envelope.messages) + safe = ( + envelope.parent_session_id is None + and envelope.lineage_inheritance != "prefix-sharing" + and served_messages == stored_messages + ) + samples.append( + { + **row, + "served_messages": served_messages, + "parent_session_id": envelope.parent_session_id, + "lineage_inheritance": envelope.lineage_inheritance, + "lineage_complete": envelope.lineage_complete, + "read_status": "safe" if safe else "unsafe", + } + ) + unsafe = sum(1 for row in samples if row.get("read_status") != "safe") + if unresolved_count == 0: + status = "not_applicable" + safe = True + elif not samples: + status = "not_observed" + safe = False + elif unsafe or errors: + status = "unsafe" + safe = False + else: + status = "safe" + safe = True + return { + "requested": limit, + "unresolved_count": unresolved_count, + "sampled": len(samples), + "status": status, + "safe": safe, + "unsafe": unsafe, + "errors": errors, + "rows": samples, + } + + +def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> dict[str, Any]: + """Return the typed topology census used by candidate and live reports. + + ``session_links.status`` is intentionally nullable for ordinary edges: + resolvedness is carried by ``resolved_dst_session_id``. The census reports + that raw fact separately and computes the public effective state as + resolved, unresolved, repaired, or quarantined. This makes an empty + effective state impossible to hide behind SQL ``NULL`` while preserving + the storage contract. + """ + if sample_unresolved < 0: + raise ValueError("sample_unresolved must be non-negative") + columns = _table_columns(conn, "session_links") + missing = sorted(REQUIRED_TOPOLOGY_LINK_COLUMNS - columns) + if missing: + return { + "checked": False, + "missing_columns": missing, + "total": 0, + "raw_status_empty_count": 0, + "empty_effective_status_count": 0, + "empty_method_count": 0, + "effective_status_counts": {}, + "method_counts": {}, + "unknown_effective_status_count": 0, + "unknown_effective_statuses": {}, + "cycle_evidence_count": 0, + "malformed_quarantine_evidence_count": 0, + "quarantined_without_cycle_evidence": 0, + "quarantined_with_resolved_parent_count": 0, + "unresolved_count": 0, + "unresolved_read_sample": { + "requested": sample_unresolved, + "unresolved_count": 0, + "sampled": 0, + "status": "not_observed", + "safe": False, + "unsafe": 0, + "errors": [], + "rows": [], + }, + } + + state_rows = _rows( + conn, + """ + SELECT CASE + WHEN NULLIF(TRIM(status), '') IS NOT NULL THEN TRIM(status) + WHEN resolved_dst_session_id IS NOT NULL THEN 'resolved' + ELSE 'unresolved' + END AS effective_status, + COUNT(*) AS links + FROM session_links + GROUP BY effective_status + ORDER BY effective_status + """, + ) + method_rows = _rows( + conn, + """ + SELECT COALESCE(NULLIF(TRIM(method), ''), '') AS method, COUNT(*) AS links + FROM session_links + GROUP BY method + ORDER BY method + """, + ) + effective_status_counts = {str(row["effective_status"]): _int(row["links"]) for row in state_rows} + method_counts = {str(row["method"]): _int(row["links"]) for row in method_rows} + raw_status_empty_count = _scalar_int( + conn, + "SELECT COUNT(*) FROM session_links WHERE status IS NULL OR TRIM(status) = ''", + ) + empty_effective_status_count = effective_status_counts.get("", 0) + empty_method_count = method_counts.get("", 0) + unknown_states = { + state: count for state, count in effective_status_counts.items() if state not in TOPOLOGY_EFFECTIVE_STATES + } + cycle_evidence_count, malformed_quarantine_evidence_count = _quarantine_evidence_counts(conn) + quarantined_count = effective_status_counts.get("quarantined", 0) + quarantined_without_cycle_evidence = max(0, quarantined_count - cycle_evidence_count) + quarantined_with_resolved_parent_count = _scalar_int( + conn, + """ + SELECT COUNT(*) + FROM session_links + WHERE TRIM(status) = 'quarantined' + AND resolved_dst_session_id IS NOT NULL + """, + ) + unresolved_read_sample = _topology_read_sample(conn, limit=sample_unresolved) + return { + "checked": True, + "missing_columns": [], + "total": sum(effective_status_counts.values()), + "raw_status_empty_count": raw_status_empty_count, + "empty_effective_status_count": empty_effective_status_count, + "empty_method_count": empty_method_count, + "effective_status_counts": effective_status_counts, + "method_counts": method_counts, + "unknown_effective_status_count": sum(unknown_states.values()), + "unknown_effective_statuses": unknown_states, + "cycle_evidence_count": cycle_evidence_count, + "malformed_quarantine_evidence_count": malformed_quarantine_evidence_count, + "quarantined_without_cycle_evidence": quarantined_without_cycle_evidence, + "quarantined_with_resolved_parent_count": quarantined_with_resolved_parent_count, + "unresolved_count": unresolved_read_sample["unresolved_count"], + "unresolved_read_sample": unresolved_read_sample, + } + + +def _receipt_sha256(payload: dict[str, Any]) -> str: + body = {key: value for key, value in payload.items() if key != "receipt_sha256"} + encoded = json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + def _lineage_integrity(conn: Connection) -> dict[str, Any]: prefix_missing_resolution = _scalar_int( conn, @@ -259,6 +546,7 @@ def _sample_prefix_sharing(conn: Connection, limit: int, *, max_stored_messages: FROM session_links l LEFT JOIN messages m ON m.session_id = l.src_session_id WHERE l.inheritance = 'prefix-sharing' + AND COALESCE(TRIM(l.status), '') != 'quarantined' GROUP BY l.src_session_id HAVING stored_messages <= ? ) @@ -276,6 +564,7 @@ def _sample_prefix_sharing(conn: Connection, limit: int, *, max_stored_messages: JOIN sessions s ON s.session_id = l.src_session_id LEFT JOIN messages m ON m.session_id = l.src_session_id WHERE l.inheritance = 'prefix-sharing' + AND COALESCE(TRIM(l.status), '') != 'quarantined' GROUP BY l.src_session_id, s.origin, s.native_id, l.resolved_dst_session_id, l.branch_point_message_id HAVING stored_messages <= ? ORDER BY stored_messages ASC, l.src_session_id @@ -348,6 +637,7 @@ def _demo_summary(report: dict[str, Any]) -> dict[str, Any]: "link_counts": report["lineage"]["counts"], "integrity": report["lineage"]["integrity"], "sample": report["lineage"]["prefix_sharing_read_sample"], + "topology": report["lineage"]["topology"], }, "caveats": verdict["reasons"] or [ @@ -387,6 +677,7 @@ def _write_readme(path: Path, report: dict[str, Any]) -> None: f"- logical sessions: `{counts['logical_sessions']}`", f"- physical/logical ratio: `{ratio_text}`", f"- stored messages: `{counts['stored_messages']}`", + f"- topology receipt SHA-256: `{report['receipt_sha256']}`", "", "## Files", "", @@ -410,9 +701,11 @@ def _write_artifacts(out_dir: Path, report: dict[str, Any]) -> None: def build_report(args: LineageValidationArgs) -> dict[str, Any]: config = _config_with_archive_root(get_config(), args.archive_root) - index_db = config.db_path + index_db = (args.index_db or config.db_path).expanduser().resolve() conn = open_readonly_connection(index_db) try: + conn.execute("BEGIN") + snapshot_before = _snapshot_identity(index_db) index_schema_version = _user_version(conn) link_columns = _table_columns(conn, "session_links") missing_link_columns = sorted(REQUIRED_SESSION_LINK_COLUMNS - link_columns) @@ -432,6 +725,7 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: ) lineage_counts = _lineage_counts(conn) integrity = _lineage_integrity(conn) + topology = census_topology_links(conn, sample_unresolved=args.sample_unresolved) prefix_sample = _sample_prefix_sharing( conn, args.sample_prefix_sharing, @@ -459,14 +753,56 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: reasons.append(f"prefix-sharing links found for unsupported origins: {origins}") if prefix_sample["errors"]: reasons.append(f"{len(prefix_sample['errors'])} sampled prefix-sharing composed reads failed") - + if not topology["checked"]: + reasons.append(f"topology census missing columns: {', '.join(topology['missing_columns'])}") + else: + if topology["empty_effective_status_count"]: + reasons.append( + f"{topology['empty_effective_status_count']} topology links have an empty effective status" + ) + if topology["empty_method_count"]: + reasons.append(f"{topology['empty_method_count']} topology links have an empty method") + if topology["unknown_effective_status_count"]: + reasons.append( + "topology census found unknown effective states: " + + ", ".join(sorted(topology["unknown_effective_statuses"])), + ) + if topology["quarantined_without_cycle_evidence"]: + reasons.append( + f"{topology['quarantined_without_cycle_evidence']} quarantined topology links lack cycle evidence" + ) + if topology["malformed_quarantine_evidence_count"]: + reasons.append( + f"{topology['malformed_quarantine_evidence_count']} quarantined topology links have malformed evidence" + ) + if topology["quarantined_with_resolved_parent_count"]: + reasons.append( + f"{topology['quarantined_with_resolved_parent_count']} quarantined topology links still resolve a parent" + ) + if topology["unresolved_read_sample"]["status"] == "not_observed": + reasons.append( + f"{topology['unresolved_count']} unresolved-parent links were not exercised through the reader" + ) + elif topology["unresolved_read_sample"]["status"] == "unsafe": + reasons.append("sampled unresolved-parent reads did not remain child-local") + + snapshot_after = _snapshot_identity(index_db) + snapshot_stable = snapshot_before["sha256"] == snapshot_after["sha256"] + if not snapshot_stable: + reasons.append("index snapshot changed during the read-only census") + snapshot_identity = { + "before": snapshot_before, + "after": snapshot_after, + "stable": snapshot_stable, + } report: dict[str, Any] = { - "report_version": 1, + "report_version": 2, "captured_at": datetime.now(UTC).isoformat(), "command": "devtools workspace lineage-validation", "archive_root": str(config.archive_root), "index_db": str(index_db), "index_schema_version": index_schema_version, + "snapshot_identity": snapshot_identity, "counts": counts, "schema": { "required_session_link_columns": sorted(REQUIRED_SESSION_LINK_COLUMNS), @@ -477,6 +813,7 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: "integrity": integrity, "missing_profile_samples": _missing_profile_samples(conn), "prefix_sharing_read_sample": prefix_sample, + "topology": topology, "supported_prefix_origins": sorted(SUPPORTED_PREFIX_ORIGINS), }, "verdict": { @@ -484,7 +821,9 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: "reasons": reasons, }, } + report["receipt_sha256"] = _receipt_sha256(report) finally: + conn.rollback() conn.close() if args.out_dir is not None: @@ -500,6 +839,8 @@ def main(argv: list[str] | None = None) -> int: sample_prefix_sharing=parsed.sample_prefix_sharing, max_sample_stored_messages=parsed.max_sample_stored_messages, json=parsed.json, + sample_unresolved=parsed.sample_unresolved, + index_db=parsed.index_db, ) report = build_report(args) if args.json: diff --git a/docs/README.md b/docs/README.md index f1d02b28ff..e8dbc68b70 100644 --- a/docs/README.md +++ b/docs/README.md @@ -85,6 +85,7 @@ Start with **Guides** for a task, **Reference** for a surface contract, and **Ar |----------|-------------| | [Demos and Proofs](demos.md) | Reproducible proofs, construct-valid demo doctrine, and flagship demonstrations. | | [Cursor Authority Census, 2026-08-04](evidence/polylogue-xeck9-cursor-authority-census-2026-08-04.md) | Privacy-safe read-only census of cursor and accepted-head readiness evidence. | +| [Topology Live-Proof Residue, 2026-08-06](evidence/polylogue-topology-live-proof-2026-08-06.md) | Candidate topology census, production-route cycle evidence, and unexercised live-archive residue. | | [Proof Artifacts](proof-artifacts.md) | Claim-to-proof map for public-facing demo and evidence claims. | | [README Public-Claims View](generated/public-claims/readme.md) | Generated compact status view for claims used in README-facing copy. | | [Launch Public-Claims View](generated/public-claims/launch.md) | Generated launch-copy claim status with evidence blockers and remediation refs. | diff --git a/docs/evidence/polylogue-topology-live-proof-2026-08-06.md b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md new file mode 100644 index 0000000000..6f2e1d2c7b --- /dev/null +++ b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md @@ -0,0 +1,34 @@ +# Topology live-proof residue, 2026-08-06 + +## Scope + +This report records the proof surface implemented for `polylogue-topology-live-proof`. The census reuses `devtools workspace lineage-validation` and the production topology write/read seams. The candidate evidence is a frozen test index populated by `write_parsed_session_to_archive`; it is not a claim about the operator's live archive. + +## Candidate proof + +The candidate fixture contains two resolved links and one unresolved native-parent link, all written through the production writer. The production writer supplies a non-empty method for all three rows. The census derives the ordinary `resolved` and `unresolved` states from `resolved_dst_session_id`, while preserving the nullable raw `status` column contract. The bounded unresolved-parent read sample exercises `read_archive_session_envelope` and proves the child remains child-local: no parent session is composed, and the served message count equals the child-owned count. Each receipt binds the report to the database and any SQLite sidecars by content digest, file identity, and a held read transaction. With a fixed capture time, an unchanged source reproduces the receipt, while a source mutation changes its binding. + +| Evidence | Result | +| --- | ---: | +| effective topology states | `resolved=2`, `unresolved=1` | +| empty effective states | `0` | +| empty methods | `0` | +| raw nullable status values | `3` ordinary NULLs, reported transparently | +| unresolved-parent reads sampled | `1` | +| unresolved-parent reads safe | `true` | +| cycle-quarantine evidence in candidate | `0` | +| candidate snapshot stable during census | `true` | + +The production-route cycle fixture separately proves a `quarantined` closing edge with `cycle_rejected` evidence. Its census has `resolved=1`, `quarantined=1`, zero empty effective states, zero empty methods, one valid cycle-evidence row, and zero malformed quarantine-evidence rows. Mutations that blank a method, introduce malformed quarantine JSON, or give a quarantined row a resolved parent each make the census fail, and the reader leaves the contradictory quarantined row uncomposed. + +## Live residue + +No live archive was opened or mutated in this lane. The live database path is outside the assigned worktree and is excluded by the repository operating boundary. Therefore this report does not claim live zero-empty counts, archive convergence, or a post-reindex status distribution. The remaining named follow-up is `polylogue-topology-live-proof`: run the read-only census against the approved live or activated candidate index, retain the generated receipt, and compare `effective_status_counts`, `empty_effective_status_count`, `empty_method_count`, `cycle_evidence_count`, and `unresolved_read_sample`. + +## Verification + +```text +devtools test tests/unit/devtools/test_lineage_validation.py tests/unit/storage/test_topology_cycle_quarantine_live.py +``` + +The tests include mutations that blank a method, introduce an unknown status, make an unresolved child claim a parent in `sessions.parent_session_id`, introduce malformed quarantine JSON, and make a quarantined row resolve a parent; each mutation makes the relevant proof fail. The live receipt step was not run, so the live census remains explicitly not observed. diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 69b6d0aa40..5baff3ee9f 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -2797,6 +2797,7 @@ def has_prefix_lineage(self, session_id: str) -> bool: WHERE src_session_id = ? AND inheritance = 'prefix-sharing' AND resolved_dst_session_id IS NOT NULL + AND COALESCE(TRIM(status), '') != 'quarantined' LIMIT 1 """, (session_id,), diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index 32bcd3fcbf..5b8ad300cd 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/polylogue/storage/sqlite/archive_tiers/write.py @@ -2782,7 +2782,8 @@ def _union_with_existing_rows( # changes which messages exist). is_prefix_sharing_parent = ( conn.execute( - "SELECT 1 FROM session_links WHERE resolved_dst_session_id = ? AND inheritance = 'prefix-sharing' LIMIT 1", + "SELECT 1 FROM session_links WHERE resolved_dst_session_id = ? AND inheritance = 'prefix-sharing' " + "AND COALESCE(TRIM(status), '') != 'quarantined' LIMIT 1", (session_id,), ).fetchone() is not None @@ -4019,6 +4020,7 @@ def _refresh_session_projection(conn: sqlite3.Connection, session_id: str, *, se SELECT resolved_dst_session_id, link_type FROM session_links WHERE src_session_id = ? AND resolved_dst_session_id IS NOT NULL + AND COALESCE(TRIM(status), '') != 'quarantined' ORDER BY observed_at_ms IS NULL, observed_at_ms, dst_origin, dst_native_id, link_type LIMIT 1 """, @@ -5700,6 +5702,7 @@ def own_signatures(target_session_id: str) -> list[tuple[str, str]]: AND inheritance = 'prefix-sharing' AND resolved_dst_session_id IS NOT NULL AND branch_point_message_id IS NOT NULL + AND COALESCE(TRIM(status), '') != 'quarantined' LIMIT 1 """, (cursor_session_id,), @@ -6104,6 +6107,7 @@ def _repair_stale_prefix_branch_points_db( WHERE l.inheritance = 'prefix-sharing' AND l.resolved_dst_session_id IS NOT NULL AND l.branch_point_message_id IS NOT NULL + AND COALESCE(TRIM(l.status), '') != 'quarantined' {scope_clause} AND NOT EXISTS ( SELECT 1 FROM messages m @@ -6430,6 +6434,7 @@ def _prefix_sharing_edge_sync(conn: sqlite3.Connection, session_id: str) -> tupl AND inheritance = 'prefix-sharing' AND resolved_dst_session_id IS NOT NULL AND branch_point_message_id IS NOT NULL + AND COALESCE(TRIM(status), '') != 'quarantined' LIMIT 1 """, (session_id,), diff --git a/polylogue/storage/sqlite/queries/message_query_reads.py b/polylogue/storage/sqlite/queries/message_query_reads.py index d7b48562a3..2edafa1e98 100644 --- a/polylogue/storage/sqlite/queries/message_query_reads.py +++ b/polylogue/storage/sqlite/queries/message_query_reads.py @@ -55,6 +55,7 @@ async def _prefix_sharing_edge(conn: aiosqlite.Connection, session_id: str) -> t AND inheritance = 'prefix-sharing' AND resolved_dst_session_id IS NOT NULL AND branch_point_message_id IS NOT NULL + AND COALESCE(TRIM(status), '') != 'quarantined' LIMIT 1 """, (session_id,), diff --git a/tests/unit/devtools/test_lineage_validation.py b/tests/unit/devtools/test_lineage_validation.py index 7880023694..a4906535de 100644 --- a/tests/unit/devtools/test_lineage_validation.py +++ b/tests/unit/devtools/test_lineage_validation.py @@ -8,9 +8,17 @@ from devtools import lineage_validation from devtools.command_catalog import COMMANDS +from polylogue.archive.message.roles import Role +from polylogue.archive.session.branch_type import BranchType +from polylogue.core.enums import BlockType, Provider +from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive +from tests.infra.frozen_clock import FrozenClock -def _make_index_db(root: Path, *, with_gap: bool = False) -> Path: +def _make_index_db(root: Path, *, with_gap: bool = False, with_unresolved: bool = False) -> Path: root.mkdir() db = root / "index.db" conn = sqlite3.connect(db) @@ -51,6 +59,8 @@ def _make_index_db(root: Path, *, with_gap: bool = False) -> Path: link_type TEXT, status TEXT, resolved_dst_session_id TEXT, + method TEXT, + evidence_json TEXT, branch_point_message_id TEXT, inheritance TEXT ); @@ -131,10 +141,27 @@ def _make_index_db(root: Path, *, with_gap: bool = False) -> Path: ('bc3', 'c3', 'text', 'child tail', 0), ('bf1', 'f1', 'text', 'fresh', 0); INSERT INTO session_links VALUES - ('child', 'codex-session', 'parent-native', 'continuation', 'resolved', 'parent', 'p2', 'prefix-sharing'), - ('fresh', 'claude-code-session', 'parent-native', 'subagent', 'resolved', 'parent', NULL, 'spawned-fresh'); + ('child', 'codex-session', 'parent-native', 'continuation', NULL, 'parent', 'parser-parent', '{}', 'p2', 'prefix-sharing'), + ('fresh', 'claude-code-session', 'parent-native', 'subagent', NULL, 'parent', 'parent-tool-use-id', '{}', NULL, 'spawned-fresh'); """ ) + if with_unresolved: + conn.executescript( + """ + INSERT INTO sessions(session_id, native_id, origin, title, root_session_id, branch_type, message_count) + VALUES ('orphan', 'orphan-native', 'codex-session', 'Orphan', 'orphan', 'continuation', 1); + INSERT INTO session_profiles VALUES ('orphan', 'orphan'); + INSERT INTO messages(message_id, session_id, native_id, role, position) + VALUES ('o1', 'orphan', 'o1', 'user', 0); + INSERT INTO blocks(block_id, message_id, block_type, text, position) + VALUES ('bo1', 'o1', 'text', 'orphan', 0); + INSERT INTO session_links + (src_session_id, dst_origin, dst_native_id, link_type, status, + resolved_dst_session_id, method, evidence_json, branch_point_message_id, inheritance) + VALUES ('orphan', 'codex-session', 'missing-parent', 'continuation', NULL, + NULL, 'parser-parent', '{}', NULL, 'spawned-fresh'); + """ + ) if with_gap: conn.executescript( """ @@ -150,16 +177,86 @@ def _make_index_db(root: Path, *, with_gap: bool = False) -> Path: return db -def _args(archive_root: Path, out_dir: Path | None = None) -> lineage_validation.LineageValidationArgs: +def _args( + archive_root: Path, + out_dir: Path | None = None, + *, + index_db: Path | None = None, +) -> lineage_validation.LineageValidationArgs: return lineage_validation.LineageValidationArgs( archive_root=archive_root, out_dir=out_dir, sample_prefix_sharing=10, max_sample_stored_messages=500, json=True, + index_db=index_db, ) +def _writer_message(provider_id: str, text: str, position: int, role: Role = Role.USER) -> ParsedMessage: + return ParsedMessage( + provider_message_id=provider_id, + role=role, + text=text, + position=position, + variant_index=0, + is_active_path=True, + is_active_leaf=False, + blocks=[ParsedContentBlock(type=BlockType.TEXT, text=text)], + ) + + +def _make_writer_candidate(root: Path) -> Path: + root.mkdir() + db = root / "index.db" + conn = sqlite3.connect(db) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + initialize_archive_tier(conn, ArchiveTier.INDEX) + try: + write_parsed_session_to_archive( + conn, + ParsedSession( + source_name=Provider.CODEX, + provider_session_id="parent", + title="parent", + messages=[ + _writer_message("p0", "hello", 0), + _writer_message("p1", "world", 1, Role.ASSISTANT), + ], + ), + ) + for provider_id, tail_text in (("child", "child tail"), ("sibling", "sibling tail")): + write_parsed_session_to_archive( + conn, + ParsedSession( + source_name=Provider.CODEX, + provider_session_id=provider_id, + title=provider_id, + parent_session_provider_id="parent", + branch_type=BranchType.FORK, + messages=[ + _writer_message(f"{provider_id}-p0", "hello", 0), + _writer_message(f"{provider_id}-p1", "world", 1, Role.ASSISTANT), + _writer_message(f"{provider_id}-tail", tail_text, 2), + ], + ), + ) + write_parsed_session_to_archive( + conn, + ParsedSession( + source_name=Provider.CODEX, + provider_session_id="orphan", + title="orphan", + parent_session_provider_id="missing-parent", + messages=[_writer_message("orphan-0", "orphan", 0)], + ), + ) + finally: + conn.close() + return db + + def test_lineage_validation_clean_archive_is_citable(tmp_path: Path) -> None: archive_root = tmp_path / "archive" _make_index_db(archive_root) @@ -177,6 +274,141 @@ def test_lineage_validation_clean_archive_is_citable(tmp_path: Path) -> None: assert sample["stored_messages"] == 1 assert sample["composed_messages"] == 3 assert sample["rows"][0]["served_exceeds_stored"] is True + topology = report["lineage"]["topology"] + assert topology["empty_effective_status_count"] == 0 + assert topology["empty_method_count"] == 0 + assert topology["effective_status_counts"] == {"resolved": 2} + assert topology["raw_status_empty_count"] == 2 + assert lineage_validation._receipt_sha256(report) == report["receipt_sha256"] + + +def test_lineage_validation_proves_unresolved_reads_stay_child_local(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + _make_index_db(archive_root, with_unresolved=True) + + report = lineage_validation.build_report(_args(archive_root)) + + topology = report["lineage"]["topology"] + assert topology["effective_status_counts"] == {"resolved": 2, "unresolved": 1} + sample = topology["unresolved_read_sample"] + assert sample["safe"] is True + assert sample["sampled"] == 1 + assert sample["rows"][0]["read_status"] == "safe" + assert report["verdict"]["external_counts_citable"] is True + + +def test_lineage_validation_proves_writer_candidate_and_snapshot_identity(tmp_path: Path) -> None: + archive_root = tmp_path / "candidate" + db = _make_writer_candidate(archive_root) + + report = lineage_validation.build_report(_args(archive_root, index_db=db)) + + topology = report["lineage"]["topology"] + assert topology["effective_status_counts"] == {"resolved": 2, "unresolved": 1} + assert topology["empty_effective_status_count"] == 0 + assert topology["empty_method_count"] == 0 + assert topology["raw_status_empty_count"] == 3 + assert topology["method_counts"] == {"parser-parent": 3} + assert topology["unresolved_read_sample"]["status"] == "safe" + assert topology["unresolved_read_sample"]["sampled"] == 1 + assert report["index_db"] == str(db.resolve()) + assert report["snapshot_identity"]["stable"] is True + assert report["snapshot_identity"]["before"]["sha256"] == report["snapshot_identity"]["after"]["sha256"] + + +def test_lineage_validation_rejects_unobserved_unresolved_reader_sample(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + _make_index_db(archive_root, with_unresolved=True) + args = lineage_validation.LineageValidationArgs( + archive_root=archive_root, + out_dir=None, + sample_prefix_sharing=10, + max_sample_stored_messages=500, + json=True, + sample_unresolved=0, + ) + + report = lineage_validation.build_report(args) + + sample = report["lineage"]["topology"]["unresolved_read_sample"] + assert sample["status"] == "not_observed" + assert sample["safe"] is False + assert report["verdict"]["external_counts_citable"] is False + assert "1 unresolved-parent links were not exercised through the reader" in report["verdict"]["reasons"] + + +@pytest.mark.frozen_clock_modules("devtools.lineage_validation") +def test_lineage_validation_receipt_reproduces_before_binding_mutation( + tmp_path: Path, frozen_clock: FrozenClock +) -> None: + configured_root = tmp_path / "configured" + candidate_root = tmp_path / "candidate" + _make_index_db(configured_root) + candidate_db = _make_index_db(candidate_root) + + first = lineage_validation.build_report(_args(configured_root, index_db=candidate_db)) + assert first["index_db"] == str(candidate_db.resolve()) + unchanged = lineage_validation.build_report(_args(configured_root, index_db=candidate_db)) + assert unchanged["captured_at"] == first["captured_at"] == frozen_clock.now().isoformat() + assert unchanged["snapshot_identity"] == first["snapshot_identity"] + assert unchanged["receipt_sha256"] == first["receipt_sha256"] + + with sqlite3.connect(candidate_db) as conn: + conn.execute("UPDATE session_links SET method = 'changed' WHERE src_session_id = 'child'") + conn.commit() + second = lineage_validation.build_report(_args(configured_root, index_db=candidate_db)) + + assert second["index_db"] == str(candidate_db.resolve()) + assert second["snapshot_identity"]["before"]["sha256"] != first["snapshot_identity"]["before"]["sha256"] + assert second["receipt_sha256"] != first["receipt_sha256"] + + +def test_lineage_validation_unchecked_census_has_checked_schema(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + db = _make_index_db(archive_root) + with sqlite3.connect(db) as checked_conn: + checked = lineage_validation.census_topology_links(checked_conn, sample_unresolved=0) + + missing_db = tmp_path / "missing.db" + with sqlite3.connect(missing_db) as missing_conn: + missing_conn.execute("CREATE TABLE session_links (src_session_id TEXT)") + unchecked = lineage_validation.census_topology_links(missing_conn, sample_unresolved=0) + + assert unchecked["checked"] is False + assert set(unchecked) == set(checked) + + +def test_lineage_validation_catches_empty_method_mutation(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + db = _make_index_db(archive_root) + with sqlite3.connect(db) as conn: + conn.execute("UPDATE session_links SET method = '' WHERE src_session_id = 'child'") + conn.commit() + + report = lineage_validation.build_report(_args(archive_root)) + + topology = report["lineage"]["topology"] + assert topology["empty_method_count"] == 1 + assert report["verdict"]["external_counts_citable"] is False + assert "1 topology links have an empty method" in report["verdict"]["reasons"] + + +def test_lineage_validation_catches_unknown_status_and_unsafe_reader_mutation(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + db = _make_index_db(archive_root, with_unresolved=True) + with sqlite3.connect(db) as conn: + conn.execute("UPDATE session_links SET status = 'made-up' WHERE src_session_id = 'child'") + conn.execute("UPDATE sessions SET parent_session_id = 'parent' WHERE session_id = 'orphan'") + conn.commit() + + report = lineage_validation.build_report(_args(archive_root)) + + topology = report["lineage"]["topology"] + assert topology["unknown_effective_status_count"] == 1 + assert topology["unresolved_read_sample"]["safe"] is False + assert report["verdict"]["external_counts_citable"] is False + assert any("unknown effective states" in reason for reason in report["verdict"]["reasons"]) + assert "sampled unresolved-parent reads did not remain child-local" in report["verdict"]["reasons"] def test_lineage_validation_reports_integrity_gaps(tmp_path: Path) -> None: @@ -206,6 +438,8 @@ def test_lineage_validation_writes_demo_artifacts(tmp_path: Path) -> None: summary = json.loads((out_dir / "summary.json").read_text(encoding="utf-8")) readme = (out_dir / "README.md").read_text(encoding="utf-8") assert written["counts"] == report["counts"] + assert written["receipt_sha256"] == report["receipt_sha256"] + assert lineage_validation._receipt_sha256(written) == written["receipt_sha256"] assert summary["artifact"] == "lineage-validation" assert summary["proof_report"]["external_counts_citable"] is True assert "external counts citable: `true`" in readme diff --git a/tests/unit/storage/test_topology_cycle_quarantine_live.py b/tests/unit/storage/test_topology_cycle_quarantine_live.py index 79b89fc3c3..fa3f103b31 100644 --- a/tests/unit/storage/test_topology_cycle_quarantine_live.py +++ b/tests/unit/storage/test_topology_cycle_quarantine_live.py @@ -27,13 +27,14 @@ from pathlib import Path from typing import cast +from devtools.lineage_validation import census_topology_links from polylogue.archive.message.roles import Role from polylogue.archive.topology.edge import TopologyEdgeStatus from polylogue.core.enums import BlockType, Provider from polylogue.sources.parsers.base import ParsedContentBlock, ParsedMessage, ParsedSession from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier -from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive +from polylogue.storage.sqlite.archive_tiers.write import read_archive_session_envelope, write_parsed_session_to_archive def _connect(path: Path) -> sqlite3.Connection: @@ -121,6 +122,47 @@ def test_cross_ingest_cycle_quarantines_the_closing_edge(tmp_path: Path) -> None assert b_link["status"] is None assert b_link["resolved_dst_session_id"] == a_id + census = census_topology_links(conn, sample_unresolved=0) + assert census["checked"] is True + assert census["empty_effective_status_count"] == 0 + assert census["empty_method_count"] == 0 + assert census["effective_status_counts"] == {"quarantined": 1, "resolved": 1} + assert census["cycle_evidence_count"] == 1 + assert census["malformed_quarantine_evidence_count"] == 0 + assert census["quarantined_with_resolved_parent_count"] == 0 + + valid_evidence = link["evidence_json"] + conn.execute("UPDATE session_links SET evidence_json = '{malformed' WHERE src_session_id = ?", (a_id,)) + malformed = census_topology_links(conn, sample_unresolved=0) + assert malformed["cycle_evidence_count"] == 0 + assert malformed["malformed_quarantine_evidence_count"] == 1 + assert malformed["quarantined_without_cycle_evidence"] == 1 + + parent_message_id = conn.execute( + "SELECT message_id FROM messages WHERE session_id = ? ORDER BY position LIMIT 1", (b_id,) + ).fetchone()[0] + conn.execute( + """ + UPDATE session_links + SET evidence_json = ?, resolved_dst_session_id = ?, branch_point_message_id = ?, + inheritance = 'prefix-sharing' + WHERE src_session_id = ? + """, + (valid_evidence, b_id, parent_message_id, a_id), + ) + quarantined_read = read_archive_session_envelope(conn, a_id) + assert quarantined_read.parent_session_id is None + assert quarantined_read.lineage_inheritance == "none" + contradictory = census_topology_links(conn, sample_unresolved=0) + assert contradictory["quarantined_with_resolved_parent_count"] == 1 + assert contradictory["cycle_evidence_count"] == 1 + + # Anti-vacuity: the census must observe a production-row mutation rather + # than merely restating the expected fixture shape. + conn.execute("UPDATE session_links SET method = '' WHERE src_session_id = ?", (b_id,)) + mutated = census_topology_links(conn, sample_unresolved=0) + assert mutated["empty_method_count"] == 1 + def test_self_referential_edge_quarantines_without_touching_projection(tmp_path: Path) -> None: db = tmp_path / "index.db"