From a31a95a1bc029e9d6297236c2e2608d5bd7a5ba0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 10:19:04 +0200 Subject: [PATCH 1/5] test(lineage): prove typed topology census Problem: topology status and method claims lacked a reusable candidate/live census, and unresolved-parent composition had no report-level proof through the archive read seam. What changed: extend lineage-validation with effective topology states, method and cycle-evidence counts, bounded unresolved-parent read checks, candidate index selection, and a receipt digest. Exercise the helper through the production cycle-quarantine writer and mutation fixtures. Compatibility/migration: preserve nullable status for ordinary resolved and unresolved rows; no schema or archive writes are introduced. Co-Authored-By: Claude --- devtools/lineage_validation.py | 219 +++++++++++++++++- .../unit/devtools/test_lineage_validation.py | 81 ++++++- .../test_topology_cycle_quarantine_live.py | 14 ++ 3 files changed, 309 insertions(+), 5 deletions(-) diff --git a/devtools/lineage_validation.py b/devtools/lineage_validation.py index 36c08e55a7..73859634ac 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,10 @@ 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"}) @dataclass(frozen=True, slots=True) @@ -27,6 +32,8 @@ class LineageValidationArgs: sample_prefix_sharing: int max_sample_stored_messages: int json: bool + sample_unresolved: int = 20 + index_db: Path | None = None def _parser() -> argparse.ArgumentParser: @@ -52,6 +59,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 @@ -157,6 +176,175 @@ 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") + 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") + return { + "requested": limit, + "sampled": len(samples), + "safe": unsafe == 0 and not errors, + "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, + "empty_effective_status_count": 0, + "empty_method_count": 0, + "effective_status_counts": {}, + "method_counts": {}, + "unknown_effective_status_count": 0, + "cycle_evidence_count": 0, + "quarantined_without_cycle_evidence": 0, + "unresolved_read_sample": { + "requested": sample_unresolved, + "sampled": 0, + "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 = _scalar_int( + conn, + """ + SELECT COUNT(*) + FROM session_links + WHERE TRIM(status) = 'quarantined' + AND json_extract(evidence_json, '$.reason') = 'cycle_rejected' + """, + ) + quarantined_count = effective_status_counts.get("quarantined", 0) + quarantined_without_cycle_evidence = max(0, quarantined_count - cycle_evidence_count) + 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, + "quarantined_without_cycle_evidence": quarantined_without_cycle_evidence, + "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, @@ -348,6 +536,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 +576,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,7 +600,7 @@ 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: index_schema_version = _user_version(conn) @@ -432,6 +622,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,9 +650,29 @@ 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 not topology["unresolved_read_sample"]["safe"]: + reasons.append("sampled unresolved-parent reads did not remain child-local") 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), @@ -477,6 +688,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,6 +696,7 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: "reasons": reasons, }, } + report["receipt_sha256"] = _receipt_sha256(report) finally: conn.close() @@ -500,6 +713,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/tests/unit/devtools/test_lineage_validation.py b/tests/unit/devtools/test_lineage_validation.py index 7880023694..45bb244658 100644 --- a/tests/unit/devtools/test_lineage_validation.py +++ b/tests/unit/devtools/test_lineage_validation.py @@ -10,7 +10,7 @@ from devtools.command_catalog import COMMANDS -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 +51,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 +133,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( """ @@ -177,6 +196,60 @@ 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_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 +279,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..db314d7777 100644 --- a/tests/unit/storage/test_topology_cycle_quarantine_live.py +++ b/tests/unit/storage/test_topology_cycle_quarantine_live.py @@ -27,6 +27,7 @@ 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 @@ -121,6 +122,19 @@ 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 + + # 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" From 89f7d1cb5870bf0f50a1422219e9d337b399a9eb Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 10:19:23 +0200 Subject: [PATCH 2/5] docs: record topology live-proof residue Problem: the topology implementation checkpoint must distinguish candidate proof from unavailable live evidence. What changed: record the candidate census results, production-route cycle evidence, read-safety proof, and the exact unexercised live step in the established evidence-report format. Compatibility/migration: this is a read-only evidence record and makes no archive or Beads changes. Co-Authored-By: Claude --- ...olylogue-topology-live-proof-2026-08-06.md | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 docs/evidence/polylogue-topology-live-proof-2026-08-06.md 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..250435036f --- /dev/null +++ b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md @@ -0,0 +1,33 @@ +# 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 focused archive fixture written 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. 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. + +| 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` | + +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, and one cycle-evidence row. + +## 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, and make an unresolved child claim a parent in `sessions.parent_session_id`; each mutation makes the census fail. The live receipt step was not run. From b78e05bb4c2704378136e3f5417ef6a8baf16e27 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 10:24:41 +0200 Subject: [PATCH 3/5] docs: index topology proof report Problem: the evidence report was rejected by the generated docs-surface gate because it lacked a registry entry. What changed: register the topology live-proof residue under the evidence tier and regenerate docs/README.md. Compatibility/migration: documentation-only change; no runtime or archive state changes. Co-Authored-By: Claude --- devtools/docs_surface.py | 6 ++++++ docs/README.md | 1 + 2 files changed, 7 insertions(+) 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/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. | From bdd2eeeca34a6c8da083acc444784f1547f5b3ca Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 10:35:44 +0200 Subject: [PATCH 4/5] test(lineage): bind topology proofs to live seams Problem: The topology census could pass with hand-built rows, a self-hash without source identity, or an empty unresolved sample.\n\nWhat changed: Build a writer-backed candidate fixture, bind each report to database and SQLite sidecar identities across a held read transaction, and make unresolved-reader sampling report not_observed instead of passing vacuously. Add explicit candidate override and mutation tests.\n\nCompatibility/migration: The nullable raw status contract remains unchanged; effective resolved, unresolved, repaired, and quarantined states are still reported separately. Co-Authored-By: Claude --- devtools/lineage_validation.py | 75 +++++++++- ...olylogue-topology-live-proof-2026-08-06.md | 5 +- .../unit/devtools/test_lineage_validation.py | 138 +++++++++++++++++- 3 files changed, 213 insertions(+), 5 deletions(-) diff --git a/devtools/lineage_validation.py b/devtools/lineage_validation.py index 73859634ac..d8639fc380 100644 --- a/devtools/lineage_validation.py +++ b/devtools/lineage_validation.py @@ -36,6 +36,34 @@ class LineageValidationArgs: 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 = hashlib.sha256(path.read_bytes()).hexdigest() + 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 _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="devtools workspace lineage-validation", @@ -187,6 +215,15 @@ def _topology_read_sample(conn: Connection, *, limit: int) -> dict[str, Any]: """ 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, """ @@ -230,10 +267,24 @@ def _topology_read_sample(conn: Connection, *, limit: int) -> dict[str, Any]: } ) 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), - "safe": unsafe == 0 and not errors, + "status": status, + "safe": safe, "unsafe": unsafe, "errors": errors, "rows": samples, @@ -268,7 +319,9 @@ def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> d "quarantined_without_cycle_evidence": 0, "unresolved_read_sample": { "requested": sample_unresolved, + "unresolved_count": 0, "sampled": 0, + "status": "not_observed", "safe": False, "unsafe": 0, "errors": [], @@ -335,6 +388,7 @@ def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> d "unknown_effective_statuses": unknown_states, "cycle_evidence_count": cycle_evidence_count, "quarantined_without_cycle_evidence": quarantined_without_cycle_evidence, + "unresolved_count": unresolved_read_sample["unresolved_count"], "unresolved_read_sample": unresolved_read_sample, } @@ -601,8 +655,10 @@ 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 = (args.index_db or config.db_path).expanduser().resolve() + snapshot_before = _snapshot_identity(index_db) conn = open_readonly_connection(index_db) try: + conn.execute("BEGIN") index_schema_version = _user_version(conn) link_columns = _table_columns(conn, "session_links") missing_link_columns = sorted(REQUIRED_SESSION_LINK_COLUMNS - link_columns) @@ -668,9 +724,22 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: reasons.append( f"{topology['quarantined_without_cycle_evidence']} quarantined topology links lack cycle evidence" ) - if not topology["unresolved_read_sample"]["safe"]: + 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": 2, "captured_at": datetime.now(UTC).isoformat(), @@ -678,6 +747,7 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: "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), @@ -698,6 +768,7 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: } report["receipt_sha256"] = _receipt_sha256(report) finally: + conn.rollback() conn.close() if args.out_dir is not None: diff --git a/docs/evidence/polylogue-topology-live-proof-2026-08-06.md b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md index 250435036f..123ed9234a 100644 --- a/docs/evidence/polylogue-topology-live-proof-2026-08-06.md +++ b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md @@ -2,11 +2,11 @@ ## 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 focused archive fixture written by `write_parsed_session_to_archive`; it is not a claim about the operator's live archive. +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. 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. +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; a source change produces a different receipt binding. | Evidence | Result | | --- | ---: | @@ -17,6 +17,7 @@ The candidate fixture contains two resolved links and one unresolved native-pare | 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, and one cycle-evidence row. diff --git a/tests/unit/devtools/test_lineage_validation.py b/tests/unit/devtools/test_lineage_validation.py index 45bb244658..8dfe932985 100644 --- a/tests/unit/devtools/test_lineage_validation.py +++ b/tests/unit/devtools/test_lineage_validation.py @@ -8,6 +8,13 @@ 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 def _make_index_db(root: Path, *, with_gap: bool = False, with_unresolved: bool = False) -> Path: @@ -169,16 +176,86 @@ def _make_index_db(root: Path, *, with_gap: bool = False, with_unresolved: bool 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) @@ -219,6 +296,65 @@ def test_lineage_validation_proves_unresolved_reads_stay_child_local(tmp_path: P 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["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"] + + +def test_lineage_validation_binds_explicit_candidate_and_mutation(tmp_path: Path) -> 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()) + first_snapshot = first["snapshot_identity"]["before"]["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 + assert second["receipt_sha256"] != first["receipt_sha256"] + + def test_lineage_validation_catches_empty_method_mutation(tmp_path: Path) -> None: archive_root = tmp_path / "archive" db = _make_index_db(archive_root) From f870283b86c79c3a4babc8da7c73745bcef32e01 Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 11:25:32 +0200 Subject: [PATCH 5/5] fix(lineage): reject unsafe topology proof rows Problem: topology receipts hashed whole files, emitted variant census shapes, and accepted malformed or contradictory quarantine evidence. Reader composition also treated a quarantined resolved edge as traversable, while the receipt mutation test changed its capture time on every run. What changed: stream snapshot hashing from a transaction-bound read, stabilize the census schema, validate cycle evidence structurally, report contradictory quarantine rows, and exclude quarantined edges from lineage readers and repair traversal. Add fixed-clock receipt reproducibility and production-route mutation tests. Compatibility/migration: no schema or archive data changes. Live census evidence remains explicitly unobserved. --- devtools/lineage_validation.py | 67 +++++++++++++++++-- ...olylogue-topology-live-proof-2026-08-06.md | 6 +- .../storage/sqlite/archive_tiers/archive.py | 1 + .../storage/sqlite/archive_tiers/write.py | 7 +- .../sqlite/queries/message_query_reads.py | 1 + .../unit/devtools/test_lineage_validation.py | 29 +++++++- .../test_topology_cycle_quarantine_live.py | 30 ++++++++- 7 files changed, 127 insertions(+), 14 deletions(-) diff --git a/devtools/lineage_validation.py b/devtools/lineage_validation.py index d8639fc380..4252dfc456 100644 --- a/devtools/lineage_validation.py +++ b/devtools/lineage_validation.py @@ -23,6 +23,7 @@ {"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) @@ -45,7 +46,7 @@ def _snapshot_identity(index_db: Path) -> dict[str, Any]: files.append({"path": str(path), "present": False}) continue stat = path.stat() - digest = hashlib.sha256(path.read_bytes()).hexdigest() + digest = _file_sha256(path) files.append( { "path": str(path), @@ -64,6 +65,14 @@ def _snapshot_identity(index_db: Path) -> dict[str, Any]: } +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: parser = argparse.ArgumentParser( prog="devtools workspace lineage-validation", @@ -153,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, @@ -310,13 +347,18 @@ def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> d "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, @@ -363,17 +405,18 @@ def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> d unknown_states = { state: count for state, count in effective_status_counts.items() if state not in TOPOLOGY_EFFECTIVE_STATES } - cycle_evidence_count = _scalar_int( + 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 json_extract(evidence_json, '$.reason') = 'cycle_rejected' + AND resolved_dst_session_id IS NOT NULL """, ) - quarantined_count = effective_status_counts.get("quarantined", 0) - quarantined_without_cycle_evidence = max(0, quarantined_count - cycle_evidence_count) unresolved_read_sample = _topology_read_sample(conn, limit=sample_unresolved) return { "checked": True, @@ -387,7 +430,9 @@ def census_topology_links(conn: Connection, *, sample_unresolved: int = 20) -> d "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, } @@ -501,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 <= ? ) @@ -518,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 @@ -655,10 +702,10 @@ 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 = (args.index_db or config.db_path).expanduser().resolve() - snapshot_before = _snapshot_identity(index_db) 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) @@ -724,6 +771,14 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: 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" diff --git a/docs/evidence/polylogue-topology-live-proof-2026-08-06.md b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md index 123ed9234a..6f2e1d2c7b 100644 --- a/docs/evidence/polylogue-topology-live-proof-2026-08-06.md +++ b/docs/evidence/polylogue-topology-live-proof-2026-08-06.md @@ -6,7 +6,7 @@ This report records the proof surface implemented for `polylogue-topology-live-p ## 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; a source change produces a different receipt binding. +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 | | --- | ---: | @@ -19,7 +19,7 @@ The candidate fixture contains two resolved links and one unresolved native-pare | 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, and one cycle-evidence row. +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 @@ -31,4 +31,4 @@ No live archive was opened or mutated in this lane. The live database path is ou 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, and make an unresolved child claim a parent in `sessions.parent_session_id`; each mutation makes the census fail. The live receipt step was not run. +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 8dfe932985..a4906535de 100644 --- a/tests/unit/devtools/test_lineage_validation.py +++ b/tests/unit/devtools/test_lineage_validation.py @@ -15,6 +15,7 @@ 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, with_unresolved: bool = False) -> Path: @@ -306,6 +307,7 @@ def test_lineage_validation_proves_writer_candidate_and_snapshot_identity(tmp_pa 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 @@ -335,7 +337,10 @@ def test_lineage_validation_rejects_unobserved_unresolved_reader_sample(tmp_path assert "1 unresolved-parent links were not exercised through the reader" in report["verdict"]["reasons"] -def test_lineage_validation_binds_explicit_candidate_and_mutation(tmp_path: Path) -> None: +@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) @@ -343,7 +348,10 @@ def test_lineage_validation_binds_explicit_candidate_and_mutation(tmp_path: Path first = lineage_validation.build_report(_args(configured_root, index_db=candidate_db)) assert first["index_db"] == str(candidate_db.resolve()) - first_snapshot = first["snapshot_identity"]["before"]["sha256"] + 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'") @@ -351,10 +359,25 @@ def test_lineage_validation_binds_explicit_candidate_and_mutation(tmp_path: Path 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 + 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) diff --git a/tests/unit/storage/test_topology_cycle_quarantine_live.py b/tests/unit/storage/test_topology_cycle_quarantine_live.py index db314d7777..fa3f103b31 100644 --- a/tests/unit/storage/test_topology_cycle_quarantine_live.py +++ b/tests/unit/storage/test_topology_cycle_quarantine_live.py @@ -34,7 +34,7 @@ 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: @@ -128,6 +128,34 @@ def test_cross_ingest_cycle_quarantines_the_closing_edge(tmp_path: Path) -> None 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.