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..8d9f2116f2 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 @@ -10,7 +11,7 @@ from datetime import UTC, datetime from pathlib import Path from sqlite3 import Connection -from typing import Any +from typing import Any, cast from polylogue.config import Config, get_config from polylogue.storage.sqlite.archive_tiers.write import read_archive_session_envelope @@ -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 @@ -74,6 +130,12 @@ def _user_version(conn: Connection) -> int: return int(row[0]) if row else 0 +def _data_version(conn: Connection) -> int: + """Return this observer connection's external-commit generation.""" + row = conn.execute("PRAGMA data_version").fetchone() + return int(row[0]) if row else 0 + + def _count(conn: Connection, table: str) -> int: row = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone() return int(row[0]) if row else 0 @@ -106,6 +168,95 @@ 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 _cycle_path_matches_projection(conn: Connection, path: list[str]) -> bool: + """Verify every recorded hop after the proposed edge against projection.""" + for child_id, parent_id in zip(path[1:-1], path[2:], strict=True): + row = conn.execute( + "SELECT parent_session_id FROM sessions WHERE session_id = ?", + (child_id,), + ).fetchone() + if row is None or row[0] != parent_id: + return False + return True + + +def _quarantine_evidence_counts(conn: Connection) -> tuple[int, int, int]: + """Count proven cycles, malformed evidence, and walk-budget exhaustion.""" + cycle_evidence_count = 0 + malformed_count = 0 + budget_exhausted_count = 0 + rows = conn.execute( + """ + SELECT links.src_session_id, + links.evidence_json, + ( + SELECT destination.session_id + FROM sessions destination + WHERE destination.origin = links.dst_origin + AND destination.native_id = links.dst_native_id + ORDER BY destination.session_id + LIMIT 1 + ) AS asserted_parent_session_id + FROM session_links links + WHERE TRIM(links.status) = 'quarantined' + """ + ).fetchall() + for src_session_id, raw_evidence, asserted_parent_session_id in rows: + try: + evidence = json.loads(raw_evidence) + except (TypeError, ValueError): + malformed_count += 1 + continue + reason = evidence.get("reason") if isinstance(evidence, dict) else None + detected_at_ms = evidence.get("detected_at_ms") if isinstance(evidence, dict) else None + timestamp_valid = ( + isinstance(evidence, dict) and isinstance(detected_at_ms, int) and not isinstance(detected_at_ms, bool) + ) + if reason == "cycle_walk_budget_exhausted": + walk_path = evidence.get("walk_path") if isinstance(evidence, dict) else None + walk_budget = evidence.get("walk_budget") if isinstance(evidence, dict) else None + if ( + timestamp_valid + and isinstance(walk_path, list) + and len(walk_path) >= 2 + and all(isinstance(session_id, str) and session_id.strip() for session_id in walk_path) + and isinstance(walk_budget, int) + and not isinstance(walk_budget, bool) + and walk_budget > 0 + and walk_path[0] == src_session_id + and walk_path[1] == asserted_parent_session_id + and walk_path[-1] != src_session_id + and len(walk_path) - 2 == walk_budget + and _cycle_path_matches_projection(conn, cast(list[str], walk_path)) + ): + budget_exhausted_count += 1 + else: + malformed_count += 1 + continue + cycle_path = evidence.get("cycle_path") if isinstance(evidence, dict) else None + if not ( + timestamp_valid + and 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) + ): + malformed_count += 1 + continue + typed_cycle_path = cast(list[str], cycle_path) + if ( + asserted_parent_session_id is not None + and typed_cycle_path[0] == src_session_id + and typed_cycle_path[-1] == src_session_id + and typed_cycle_path[1] == asserted_parent_session_id + and _cycle_path_matches_projection(conn, typed_cycle_path) + ): + cycle_evidence_count += 1 + else: + malformed_count += 1 + return cycle_evidence_count, malformed_count, budget_exhausted_count + + def _logical_session_count(conn: Connection) -> int: return _scalar_int( conn, @@ -157,6 +308,240 @@ 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_origin AS parent_origin, + l.dst_native_id AS parent_native_id, + l.link_type, + COUNT(DISTINCT 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_origin, l.dst_native_id, l.link_type + ORDER BY l.src_session_id, l.dst_origin, l.dst_native_id, l.link_type + 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, + "budget_exhausted_quarantine_evidence_count": 0, + "quarantined_without_cycle_evidence": 0, + "quarantined_with_resolved_parent_count": 0, + "quarantined_with_stale_projection_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, + budget_exhausted_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 + """, + ) + quarantined_with_stale_projection_count = _scalar_int( + conn, + """ + SELECT COUNT(*) + FROM session_links l + JOIN sessions s ON s.session_id = l.src_session_id + JOIN sessions asserted_parent + ON asserted_parent.origin = l.dst_origin + AND asserted_parent.native_id = l.dst_native_id + WHERE TRIM(l.status) = 'quarantined' + AND s.parent_session_id = asserted_parent.session_id + AND NOT EXISTS ( + SELECT 1 + FROM session_links valid + WHERE valid.src_session_id = l.src_session_id + AND valid.resolved_dst_session_id = s.parent_session_id + AND COALESCE(TRIM(valid.status), '') != 'quarantined' + ) + """, + ) + 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, + "budget_exhausted_quarantine_evidence_count": budget_exhausted_quarantine_evidence_count, + "quarantined_without_cycle_evidence": quarantined_without_cycle_evidence, + "quarantined_with_resolved_parent_count": quarantined_with_resolved_parent_count, + "quarantined_with_stale_projection_count": quarantined_with_stale_projection_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 +644,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 +662,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 +735,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 +775,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,10 +799,18 @@ 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) + observer: Connection | None = None try: + observer = open_readonly_connection(index_db) + observer_data_version_before = _data_version(observer) + conn.execute("BEGIN") + # BEGIN is deferred. Force the first SQLite read before hashing WAL + # sidecars so this census's own reader mark cannot make a quiescent + # snapshot appear to change between the before and after identities. index_schema_version = _user_version(conn) + snapshot_before = _snapshot_identity(index_db) link_columns = _table_columns(conn, "session_links") missing_link_columns = sorted(REQUIRED_SESSION_LINK_COLUMNS - link_columns) physical_sessions = _count(conn, "sessions") @@ -432,6 +829,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 +857,74 @@ 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["budget_exhausted_quarantine_evidence_count"]: + reasons.append( + f"{topology['budget_exhausted_quarantine_evidence_count']} quarantined topology links only have " + "cycle-walk budget exhaustion 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["quarantined_with_stale_projection_count"]: + reasons.append( + f"{topology['quarantined_with_stale_projection_count']} quarantined topology links retain a parent projection" + ) + 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) + observer_data_version_after = _data_version(observer) + file_set_stable = snapshot_before["sha256"] == snapshot_after["sha256"] + no_concurrent_commits = observer_data_version_before == observer_data_version_after + snapshot_stable = file_set_stable and no_concurrent_commits + if not file_set_stable: + reasons.append("index file set changed during the read-only census") + if not no_concurrent_commits: + reasons.append("index received a concurrent commit during the read-only census") + snapshot_identity = { + "before": snapshot_before, + "after": snapshot_after, + "file_set_stable": file_set_stable, + "observer_data_version_before": observer_data_version_before, + "observer_data_version_after": observer_data_version_after, + "no_concurrent_commits": no_concurrent_commits, + "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 +935,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,8 +943,12 @@ def build_report(args: LineageValidationArgs) -> dict[str, Any]: "reasons": reasons, }, } + report["receipt_sha256"] = _receipt_sha256(report) finally: + conn.rollback() conn.close() + if observer is not None: + observer.close() if args.out_dir is not None: _write_artifacts(args.out_dir, report) @@ -500,6 +963,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..d3f5027f03 --- /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, a held read transaction, and a second SQLite observer that rejects any concurrent commit between the snapshot read and both file-set hashes. 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. Valid evidence must carry a closed cycle path anchored to the quarantined source and asserted parent, with every return hop present in the stored projection. The production writer records walk-budget exhaustion as an indeterminate `cycle_walk_budget_exhausted` quarantine, never as a demonstrated cycle, and preserves the child's full transcript. 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, provide fabricated closed-cycle JSON, exhaust the walk budget, 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-live-operation-receipts`: 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`, provide malformed or unrelated cycle evidence, exhaust cycle-walk budget, make a quarantined row resolve a parent, and commit through a second WAL connection between the held reader snapshot and file hashing. 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/docs/maintenance.md b/docs/maintenance.md index 62f1375a44..ff7e9dbd7e 100644 --- a/docs/maintenance.md +++ b/docs/maintenance.md @@ -9,8 +9,9 @@ common operational incidents. ## Applying a durable schema change train Durable schema changes are an offline release operation. Before applying a -`source.db` or `user.db` migration above its adoption floor, confirm that the -release contains the matching `migrations/{source,user}/NNN.train.json` +`source.db`, `user.db`, or `audit.db` migration above its adoption floor, +confirm that the release contains the matching +`migrations/{source,user,audit}/NNN.train.json` sidecar. The sidecar reserves the exact slot and SQL hash and records the runtime and restart evidence needed for the change. @@ -38,6 +39,53 @@ Restart health and runtime-consumer convergence are the final lifecycle proof and are recorded by the durable train lifecycle API, not inferred from this command's migration result alone. +### Rebuild deployment-currency preflight + +Before a managed `rebuild-index`, confirm that the package selected for the +operation owns the live durable schemas. The read-only preflight checks every +canonical durable migration tier: `source.db`, `user.db`, and `audit.db`. +`index.db` may be behind because the rebuild is the supported way to replace +that derived tier. + +```bash +polylogue ops maintenance rebuild-index --preflight --output-format json +``` + +It emits `rebuild-schema-currency` JSON with each durable tier's observed and +package-expected `user_version`, and exits nonzero when a durable tier differs. +The execution route checks it before consuming the schema-inference receipt, +repeats it after archive ownership acquisition, and rejects daemon bulk +transaction creation before any bookkeeping or candidate generation. + +For a safe deployment recovery, first choose the exact target package commit. +With the daemon stopped, create a fresh verified full-evidence backup. If the +preflight reports that a newly introduced durable tier is absent, initialize +only that absent file through the archive ownership gate: + +```bash +polylogue ops maintenance migrate-tier audit --initialize-missing --output-format json +``` + +The flag builds the canonical database in memory, writes it directly into an +anonymous inode, and publishes that inode with an atomic no-replace link. It +never exposes a writable staging name, refuses any existing target including +one created concurrently, and never replaces durable data. For each existing +tier that the selected package reports behind, run its numbered +migration with the verified full-evidence backup manifest: + +```bash +polylogue ops maintenance migrate-tier source --backup-manifest /path/to/verified-full-backup/manifest.json --output-format json +polylogue ops maintenance migrate-tier user --backup-manifest /path/to/verified-full-backup/manifest.json --output-format json +polylogue ops maintenance migrate-tier audit --backup-manifest /path/to/verified-full-backup/manifest.json --output-format json +``` + +Deploy that exact package after every required durable migration. Run the +preflight again and require a ready result before invoking `polylogue ops +maintenance rebuild-index`; use that blue-green command rather than `ops reset +--index` for an active managed generation. Restart the daemon only after the +rebuilt generation is promoted and the post-deploy status shows no durable-tier +mismatch. + For the conceptual model behind derived insights and the FTS / blob substrate, see [architecture.md](architecture.md) and [internals.md](internals.md). For daemon ownership of the inline @@ -791,8 +839,9 @@ escalate. `SchemaVersionError: database is version N, code expects version M`. Polylogue uses durability-keyed schema versioning (see [internals.md ยง Schema Versioning Model](internals.md#schema-versioning-model)): -derived tiers rebuild, while durable `source.db` and `user.db` may advance only -through explicit additive numbered migrations. There is no auto-downgrade. +derived tiers rebuild, while durable `source.db`, `user.db`, and `audit.db` may +advance only through explicit additive numbered migrations. There is no +auto-downgrade. **Root cause.** A new release advanced one tier's schema version and the database is on the previous version. There is no reverse in-place migration. @@ -814,7 +863,7 @@ systemctl --user stop polylogued.service # install the previous polylogue version, leave the database # alone, restart the daemon. -# 3b. Derived-tier forward rebuild: keep the source/user/embedding tiers safe, +# 3b. Derived-tier forward rebuild: keep source/user/audit/embedding tiers safe, # move the mismatched index database aside, and re-ingest/rederive # the rebuildable index with the new polylogue binary. cp ~/.local/share/polylogue/index.db /tmp/index-before-rebuild.db diff --git a/polylogue/cli/commands/maintenance/_migrate_tier.py b/polylogue/cli/commands/maintenance/_migrate_tier.py index b14c216e08..7859bdb25c 100644 --- a/polylogue/cli/commands/maintenance/_migrate_tier.py +++ b/polylogue/cli/commands/maintenance/_migrate_tier.py @@ -26,6 +26,7 @@ ArchiveOwnershipError, acquire_durable_archive_ownership, execute_durable_change_train, + initialize_missing_durable_tier, ) from polylogue.paths import archive_root from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier @@ -58,8 +59,18 @@ def _require_stopped_daemon(root: Path) -> str: type=click.Path(path_type=Path, exists=True), help="Verified backup manifest. Required only when a selected migration changes existing durable data.", ) +@click.option( + "--initialize-missing", + is_flag=True, + help="Initialize this durable tier only when its database file is absent; never replaces an existing file.", +) @click.option("--output-format", type=click.Choice(["plain", "json"]), default="plain", show_default=True) -def migrate_tier_command(tier: str, backup_manifest: Path | None, output_format: str) -> None: +def migrate_tier_command( + tier: str, + backup_manifest: Path | None, + initialize_missing: bool, + output_format: str, +) -> None: """Apply additive migrations for one durable archive tier. Derived tiers are intentionally excluded from this command; rebuild or @@ -71,17 +82,24 @@ def migrate_tier_command(tier: str, backup_manifest: Path | None, output_format: spec = ARCHIVE_TIER_SPECS[archive_tier] path = archive_root() / spec.filename stopped_daemon_evidence_ref: str | None = None + initialized = False + initialized_version: int | None = None try: with acquire_durable_archive_ownership(path.parent, owner_id=f"migrate-tier:{os.getpid()}") as archive_owner: stopped_daemon_evidence_ref = _require_stopped_daemon(path.parent) - execution = execute_durable_change_train( - path.parent, - archive_tier, - backup_manifest=backup_manifest, - daemon_stopped_evidence_ref=stopped_daemon_evidence_ref, - single_writer_evidence_ref="proof:archive-ownership-lock", - release_archive_ownership=archive_owner.release, - ) + if initialize_missing: + initialized_version = initialize_missing_durable_tier(path, archive_tier) + initialized = True + execution = None + else: + execution = execute_durable_change_train( + path.parent, + archive_tier, + backup_manifest=backup_manifest, + daemon_stopped_evidence_ref=stopped_daemon_evidence_ref, + single_writer_evidence_ref="proof:archive-ownership-lock", + release_archive_ownership=archive_owner.release, + ) except (sqlite3.Error, MigrationError, ArchiveOwnershipError) as exc: if output_format == "json": click.echo( @@ -102,26 +120,32 @@ def migrate_tier_command(tier: str, backup_manifest: Path | None, output_format: click.echo(f"Migration blocked for {tier}: {exc}", err=True) raise SystemExit(1) from exc - result = execution.migration_result + result = execution.migration_result if execution is not None else None payload = { "ok": True, "tier": tier, "path": str(path), + "initialized": initialized, "backup_manifest": str(backup_manifest) if backup_manifest is not None else None, "stopped_daemon_evidence_ref": stopped_daemon_evidence_ref, - "train_manifest": str(execution.manifest_path) if execution.manifest_path is not None else None, - "train_state": execution.train.state.value if execution.train is not None else None, + "train_manifest": ( + str(execution.manifest_path) if execution is not None and execution.manifest_path is not None else None + ), + "train_state": execution.train.state.value if execution is not None and execution.train is not None else None, "backup_receipt": str(result.backup_receipt) if result is not None and result.backup_receipt is not None else None, - "from_version": result.from_version if result is not None else None, - "to_version": result.to_version if result is not None else None, + "from_version": result.from_version if result is not None else 0 if initialized else None, + "to_version": result.to_version if result is not None else initialized_version, "applied_versions": list(result.applied_versions) if result is not None else [], } if output_format == "json": click.echo(json.dumps(payload, indent=2, sort_keys=True)) return + if initialized: + click.echo(f"Initialized missing {tier} tier at schema version {initialized_version}.") + return if result is None: click.echo(f"No pending durable migration for {tier}.") return diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index ed2e128aae..1a8efa530e 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -388,6 +388,11 @@ def _rebuild_index_selection_plan( "single-writer path; unsupported with --daemon." ), ) +@click.option( + "--preflight", + is_flag=True, + help="Read-only: report whether every durable tier matches this package before rebuilding index.db.", +) def rebuild_index_command( only_missing: bool, raw_ids: tuple[str, ...], @@ -404,6 +409,7 @@ def rebuild_index_command( use_daemon: bool, daemon_url: str, shard_count: int, + preflight: bool, ) -> None: """Inspect or execute an authority-safe source-to-index rebuild. @@ -423,6 +429,8 @@ def rebuild_index_command( raise click.BadParameter("plan limit must be positive", param_hint="--plan-limit") if use_daemon and plan_only: raise click.UsageError("--daemon executes a rebuild; --plan is always a local read-only preview") + if use_daemon and preflight: + raise click.UsageError("--preflight cannot be combined with --daemon") if shard_count <= 0: raise click.BadParameter("shard count must be positive", param_hint="--shard-count") if use_daemon and shard_count > 1: @@ -441,6 +449,22 @@ def rebuild_index_command( raise click.UsageError("resumed rebuild budgets are durable; omit pass budget options with --operation-id") root = archive_root() + if preflight: + from polylogue.maintenance.rebuild_index import rebuild_schema_currency_preflight + + payload = rebuild_schema_currency_preflight(root) + if output_format == "json": + click.echo(json.dumps(payload, indent=2, sort_keys=True)) + else: + click.echo(f"Archive root: {root}") + for tier in cast(list[dict[str, object]], payload["tiers"]): + click.echo( + f"{tier['tier']}.db: {tier['actual_user_version']} (package expects " + f"{tier['expected_user_version']}; {tier['status']})" + ) + if payload["status"] != "ready": + raise click.ClickException("rebuild schema currency preflight failed; migrate or deploy before rebuilding") + return if use_daemon: payload = _run_daemon_rebuild( daemon_url, @@ -462,23 +486,8 @@ def rebuild_index_command( click.echo(f"Replayed: {int(cast(Any, payload['replayed_logical_source_count'])):,} logical source(s)") click.echo(f"Quarantined: {int(cast(Any, payload['quarantined_raw_count'])):,} raw row(s)") return - raw_count = _count_source_raw_sessions(root) - if raw_count == 0: - payload = { - "archive_root": str(root), - "raw_session_count": 0, - "selected_raw_count": 0, - "skipped_by_blob_limit_count": 0, - "status": "empty-source", - "materialized": False, - } - if output_format == "json": - click.echo(json.dumps(payload, indent=2, sort_keys=True)) - else: - click.echo(f"Archive root: {root}") - click.echo("No source.db raw_sessions rows found.") - return if plan_only: + raw_count = _count_source_raw_sessions(root) selected_raw_ids = ( list(dict.fromkeys(raw_ids)) if raw_ids @@ -533,7 +542,11 @@ def rebuild_index_command( f"blob={int(group['blob_bytes']):,} source={group['source_path']}" ) return - from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync + from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + RebuildSchemaCurrencyError, + rebuild_index_from_source_sync, + ) try: receipt = rebuild_index_from_source_sync( @@ -551,7 +564,7 @@ def rebuild_index_command( shard_count=shard_count, ) ) - except (RuntimeError, ValueError) as exc: + except (RebuildSchemaCurrencyError, RuntimeError, ValueError) as exc: raise click.ClickException(str(exc)) from exc payload = receipt.to_dict() result = payload @@ -559,6 +572,9 @@ def rebuild_index_command( click.echo(json.dumps(payload, indent=2, sort_keys=True)) return click.echo(f"Archive root: {root}") + if receipt.status == "empty-source": + click.echo("No source.db raw_sessions rows found.") + return click.echo(f"Classified: {int(cast(Any, result['classified_full_count'])):,} full revision(s)") click.echo(f"Replayed: {int(cast(Any, result['replayed_logical_source_count'])):,} logical source(s)") click.echo(f"Quarantined: {int(cast(Any, result['quarantined_raw_count'])):,} raw row(s)") diff --git a/polylogue/daemon/bulk_rebuild.py b/polylogue/daemon/bulk_rebuild.py index a5f0d7c5fc..02c2dde5ad 100644 --- a/polylogue/daemon/bulk_rebuild.py +++ b/polylogue/daemon/bulk_rebuild.py @@ -182,6 +182,9 @@ def resolve_or_start_daemon_bulk_rebuild_transaction( generation directory, so both must fail closed against a foreign/rotated archive location before touching disk, not just the eventual write pass. """ + from polylogue.maintenance.rebuild_index import require_rebuild_schema_currency + + require_rebuild_schema_currency(root) _validate_rebuild_provenance_receipt(root, schema_inference_receipt_path) # This must precede transaction resolution, because retiring a terminal # transaction and creating its replacement also creates generation state. @@ -193,6 +196,10 @@ def resolve_or_start_daemon_bulk_rebuild_transaction( owned = OwnedArchiveLocation.acquire(location) try: assert_owns_archive_location(owned, location) + # The early check is a cheap rejection before receipt work. Repeat it + # under archive ownership because a previous owner can migrate a + # durable tier while this caller waits for the lock. + require_rebuild_schema_currency(root) # The first validation is only a cheap early rejection. Revalidate # after ownership acquisition so receipt expiry, source revision, or # external-corpus drift cannot reach generation bookkeeping. @@ -319,7 +326,11 @@ async def run_daemon_bulk_rebuild_pass( never opens a second writer connection of its own). """ from polylogue.daemon.write_coordinator import daemon_write_coordinator - from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync + from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + rebuild_index_from_source_sync, + require_rebuild_schema_currency, + ) from polylogue.maintenance.schema_inference_gate import resolve_schema_inference_receipt_reference root = Path(config.archive_root) @@ -336,6 +347,10 @@ async def run_daemon_bulk_rebuild_pass( owned = await asyncio.to_thread(OwnedArchiveLocation.acquire, location) try: await asyncio.to_thread(assert_owns_archive_location, owned, location) + # Transaction resolution has its own ownership-bound currency check, + # but a migration can complete before this later page-selection hold. + # Recheck before consuming the receipt or selecting source material. + await asyncio.to_thread(require_rebuild_schema_currency, root) await asyncio.to_thread(_validate_rebuild_provenance_receipt, root, receipt_path) store = IndexGenerationStore(location) await asyncio.to_thread(_validate_rebuild_provenance_receipt, root, receipt_path) diff --git a/polylogue/daemon/http.py b/polylogue/daemon/http.py index b7c9b499bb..449f6cfd2f 100644 --- a/polylogue/daemon/http.py +++ b/polylogue/daemon/http.py @@ -1085,6 +1085,10 @@ def wrapper(self: DaemonAPIHandler, *args: object, **kwargs: object) -> None: if 100 <= exc.http_status_code <= 599 else HTTPStatus.INTERNAL_SERVER_ERROR ) + diagnostic = getattr(exc, "diagnostic", None) + if isinstance(diagnostic, dict): + self._send_json(status, diagnostic) + return field = getattr(exc, "field", None) self._send_json( status, diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 2805347939..3b15f24285 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -16,10 +16,12 @@ import time from dataclasses import asdict, dataclass, field from hashlib import sha256 +from http import HTTPStatus from pathlib import Path from typing import TYPE_CHECKING, cast from polylogue.config import Config +from polylogue.core.errors import PolylogueError from polylogue.logging import get_logger from polylogue.maintenance.offline_guard import offline_maintenance_block_reason from polylogue.paths import render_root @@ -66,6 +68,63 @@ class RebuildDerivedStateProvenanceError(RebuildProvenanceError): """A derived-state stage was blocked by a failed provenance recheck.""" +class RebuildSchemaCurrencyError(PolylogueError): + """The durable tiers do not match the package that would rebuild them.""" + + http_status_code = HTTPStatus.CONFLICT + + def __init__(self, diagnostic: dict[str, object]) -> None: + self.diagnostic = diagnostic + blocked = diagnostic["blocking_tiers"] + assert isinstance(blocked, list) + detail = ", ".join( + f"{item['tier']}.db:{item['actual_user_version']}!={item['expected_user_version']}" + for item in blocked + if isinstance(item, dict) + ) + super().__init__(f"rebuild schema currency preflight failed: {detail}") + + +def rebuild_schema_currency_preflight(root: Path) -> dict[str, object]: + """Report whether every durable tier matches this runtime package. + + ``index.db`` is intentionally absent: rebuilding it is the operation's + purpose, while a durable-tier mismatch means this package can interpret or + write durable evidence using a schema it does not own. + """ + from polylogue.storage.archive_readiness import probe_archive_tier + from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS + + checks: list[dict[str, object]] = [] + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + probe = probe_archive_tier(tier, root / f"{tier.value}.db") + checks.append( + { + "tier": tier.value, + "path": probe.path, + "actual_user_version": probe.user_version, + "expected_user_version": probe.expected_user_version, + "status": probe.version_status, + } + ) + blocking = [check for check in checks if check["status"] != "ok"] + return { + "kind": "rebuild-schema-currency", + "archive_root": str(root), + "status": "ready" if not blocking else "blocked", + "tiers": checks, + "blocking_tiers": blocking, + } + + +def require_rebuild_schema_currency(root: Path) -> dict[str, object]: + """Reject a rebuild before it consumes evidence or creates a generation.""" + diagnostic = rebuild_schema_currency_preflight(root) + if diagnostic["status"] != "ready": + raise RebuildSchemaCurrencyError(diagnostic) + return diagnostic + + @dataclass(frozen=True, slots=True) class RebuildProvenanceContext: """Validated evidence shared by every mutation in one rebuild pass. @@ -1047,6 +1106,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde log_mapped_bytes_budget_check(logger, check_mapped_bytes_budget_against_cgroup_limit()) validate_rebuild_index_request(request) root = request.archive_root + require_rebuild_schema_currency(root) consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) location = ArchiveLocation.resolve(root) # The joined raw-frontier projection is rooted at the co-located active @@ -1084,6 +1144,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde owned = OwnedArchiveLocation.acquire(location) try: assert_owns_archive_location(owned, location) + require_rebuild_schema_currency(root) consumed_evidence = _validate_rebuild_provenance_receipt(root, request.schema_inference_receipt_path) # The lease is itself lifecycle state guarded by the provenance gate. # Revalidate again under the lease immediately before the owned body diff --git a/polylogue/operations/durable_change_train.py b/polylogue/operations/durable_change_train.py index 3d448215b5..f2cd93bf61 100644 --- a/polylogue/operations/durable_change_train.py +++ b/polylogue/operations/durable_change_train.py @@ -2,6 +2,9 @@ from __future__ import annotations +import os +import sqlite3 +import stat from collections.abc import Callable from pathlib import Path @@ -16,7 +19,7 @@ from polylogue.storage.sqlite.durable_change_train import ( reconcile_durable_change_train_startup as _reconcile_durable_change_train_startup, ) -from polylogue.storage.sqlite.migration_runner import DurableRuntimeConsumerResult +from polylogue.storage.sqlite.migration_runner import DurableRuntimeConsumerResult, MigrationError def acquire_durable_archive_ownership(root: Path, *, owner_id: str) -> OwnedArchiveLocation: @@ -25,6 +28,108 @@ def acquire_durable_archive_ownership(root: Path, *, owner_id: str) -> OwnedArch return OwnedArchiveLocation.acquire(location, owner_id=owner_id) +def initialize_missing_durable_tier(path: Path, tier: ArchiveTier) -> int: + """Initialize one absent durable tier while the caller owns the archive. + + This is deliberately separate from migration. A missing tier has no + historical schema version to advance, while an existing path must never be + replaced or interpreted as empty by this recovery route. + """ + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier + + try: + parent_metadata = path.parent.lstat() + except FileNotFoundError as exc: + raise MigrationError(f"durable tier parent directory is missing: {path.parent}") from exc + if ( + stat.S_ISLNK(parent_metadata.st_mode) + or not stat.S_ISDIR(parent_metadata.st_mode) + or parent_metadata.st_uid != os.geteuid() + or stat.S_IMODE(parent_metadata.st_mode) & 0o022 + ): + raise MigrationError(f"durable tier parent is not a private owned directory: {path.parent}") + + try: + path.lstat() + except FileNotFoundError: + pass + else: + raise MigrationError(f"{tier.value} tier already exists; refusing missing-tier initialization: {path}") + + # Build the canonical database in memory, copy its serialized image into + # an anonymous inode, then publish that exact inode with link(2). No named + # staging path exists for a concurrent same-UID process to replace or + # mutate, and link cannot replace a target that appears concurrently. + anonymous_flag = getattr(os, "O_TMPFILE", 0) + if not anonymous_flag: + raise MigrationError("missing-tier initialization requires anonymous-file publication support") + publication_descriptor: int | None = None + try: + try: + publication_descriptor = os.open( + path.parent, + os.O_RDWR | anonymous_flag | getattr(os, "O_CLOEXEC", 0), + 0o600, + ) + except OSError as exc: + raise MigrationError(f"cannot create anonymous durable-tier publication inode: {path.parent}") from exc + + memory_database = sqlite3.connect(":memory:") + try: + initialize_archive_tier(memory_database, tier) + initialized_image = memory_database.serialize() + finally: + memory_database.close() + if not initialized_image: + raise MigrationError(f"canonical {tier.value} tier initialization produced an empty database image") + + source_offset = 0 + while source_offset < len(initialized_image): + written_offset = 0 + chunk = initialized_image[source_offset : source_offset + 1024 * 1024] + while written_offset < len(chunk): + written = os.write(publication_descriptor, chunk[written_offset:]) + if written <= 0: + raise MigrationError("durable-tier publication copy made no progress") + written_offset += written + source_offset += len(chunk) + os.fsync(publication_descriptor) + publication_metadata = os.fstat(publication_descriptor) + if ( + not stat.S_ISREG(publication_metadata.st_mode) + or publication_metadata.st_nlink != 0 + or publication_metadata.st_size != len(initialized_image) + ): + raise MigrationError(f"anonymous durable-tier publication image is incomplete: {path}") + publication_identity = (publication_metadata.st_dev, publication_metadata.st_ino) + try: + # O_TMPFILE plus link(2) publishes one descriptor-backed inode + # without resolving the replaceable named staging path again. + os.link(f"/proc/self/fd/{publication_descriptor}", path, follow_symlinks=True) + except FileExistsError as exc: + raise MigrationError( + f"{tier.value} tier appeared during initialization; refusing to replace it: {path}" + ) from exc + published_metadata = path.lstat() + if ( + not stat.S_ISREG(published_metadata.st_mode) + or (published_metadata.st_dev, published_metadata.st_ino) != publication_identity + ): + raise MigrationError(f"published durable tier identity does not match the staged database: {path}") + directory_descriptor = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) + finally: + if publication_descriptor is not None: + os.close(publication_descriptor) + + from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER + + return ARCHIVE_VERSION_BY_TIER[tier] + + def execute_durable_change_train( archive_root: Path, tier: ArchiveTier, @@ -56,5 +161,6 @@ def reconcile_durable_change_trains_on_startup(root: Path) -> tuple[Path, ...]: "acquire_durable_archive_ownership", "ArchiveOwnershipError", "execute_durable_change_train", + "initialize_missing_durable_tier", "reconcile_durable_change_trains_on_startup", ] 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..a84449b4da 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/polylogue/storage/sqlite/archive_tiers/write.py @@ -19,7 +19,7 @@ from contextlib import contextmanager, nullcontext from dataclasses import dataclass, replace from pathlib import Path -from typing import cast +from typing import Literal, cast from urllib.parse import urlparse from polylogue.archive.message.types import MessageType @@ -451,6 +451,18 @@ def add_timing(name: str, started_at: float) -> None: force_spawned_fresh = False if parent_session_id is not None and messages: parent_composed: list[tuple[str, str]] | None = None + # Link quarantine happens after the session/link rows are written, + # but prefix normalization happens here. Classify the proposed edge + # against the current projection before deleting the copied prefix. + # The graph resolver remains the authority that persists quarantine + # evidence; this early check preserves the full child transcript for + # both proven cycles and indeterminate over-budget walks. + cycle_walk = _would_create_cycle( + conn, + child_id=session_id, + proposed_parent_id=parent_session_id, + ) + force_spawned_fresh = cycle_walk.outcome != "acyclic" if acompact: parent_composed = _composed_db_signatures(conn, parent_session_id, cache=signature_cache) membership = _acompact_content_membership_ratio( @@ -2782,7 +2794,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 @@ -3727,40 +3740,46 @@ def _branch_type_from_link_type(link_type: object) -> str | None: _CYCLE_WALK_BUDGET = 1024 +@dataclass(frozen=True, slots=True) +class _CycleWalkResult: + outcome: Literal["acyclic", "cycle", "budget_exhausted"] + path: tuple[str, ...] + + def _would_create_cycle( conn: sqlite3.Connection, *, child_id: str, proposed_parent_id: str, -) -> list[str] | None: - """Return the cycle path if resolving child->proposed_parent would close a loop. +) -> _CycleWalkResult: + """Classify the proposed edge without conflating exhaustion with a cycle. Walks ``sessions.parent_session_id`` upward from ``proposed_parent_id``. - Returns ``None`` for a legitimate (acyclic, or not-yet-resolvable) shape. + Budget exhaustion is indeterminate and must remain quarantined, but it is + not evidence that the proposed edge closes a cycle. """ if proposed_parent_id == child_id: - return [child_id, child_id] + return _CycleWalkResult("cycle", (child_id, child_id)) path: list[str] = [child_id, proposed_parent_id] current = proposed_parent_id steps = 0 while True: if steps >= _CYCLE_WALK_BUDGET: - path.append("...budget-exceeded") - return path + return _CycleWalkResult("budget_exhausted", tuple(path)) row = conn.execute( "SELECT parent_session_id FROM sessions WHERE session_id = ?", (current,), ).fetchone() if row is None: - return None + return _CycleWalkResult("acyclic", tuple(path)) next_parent = row[0] if next_parent is None: - return None + return _CycleWalkResult("acyclic", tuple(path)) if next_parent == child_id: path.append(child_id) - return path - path.append(next_parent) - current = next_parent + return _CycleWalkResult("cycle", tuple(path)) + path.append(str(next_parent)) + current = str(next_parent) steps += 1 @@ -3771,17 +3790,26 @@ def _quarantine_session_link( dst_origin: str, dst_native_id: str, link_type: str, - cycle_path: list[str], + cycle_walk: _CycleWalkResult, observed_at_ms: int, ) -> None: - """Mark one edge quarantined instead of resolving it, with evidence.""" - evidence = _json_dumps( - { + """Mark one unsafe edge quarantined with accurately typed evidence.""" + if cycle_walk.outcome == "cycle": + evidence_payload: dict[str, JSONValue] = { "reason": "cycle_rejected", - "cycle_path": cycle_path, + "cycle_path": list(cycle_walk.path), "detected_at_ms": observed_at_ms, } - ) + elif cycle_walk.outcome == "budget_exhausted": + evidence_payload = { + "reason": "cycle_walk_budget_exhausted", + "walk_path": list(cycle_walk.path), + "walk_budget": _CYCLE_WALK_BUDGET, + "detected_at_ms": observed_at_ms, + } + else: + raise ValueError("acyclic session link cannot be quarantined as a cycle risk") + evidence = _json_dumps(evidence_payload) conn.execute( """ UPDATE session_links @@ -3862,16 +3890,16 @@ def record_substage(name: str, started_at: float) -> None: child_id, link_type = str(row[0]), str(row[1]) # polylogue-4ts.10: session_id is about to become child_id's parent -- # refuse (quarantine, with evidence) rather than silently resolve if - # that would close a cycle in sessions.parent_session_id. - cycle_path = _would_create_cycle(conn, child_id=child_id, proposed_parent_id=session_id) - if cycle_path is not None: + # that would close a cycle or cannot be decided within the walk budget. + cycle_walk = _would_create_cycle(conn, child_id=child_id, proposed_parent_id=session_id) + if cycle_walk.outcome != "acyclic": _quarantine_session_link( conn, src_session_id=child_id, dst_origin=origin, dst_native_id=native_id, link_type=link_type, - cycle_path=cycle_path, + cycle_walk=cycle_walk, observed_at_ms=int(time.time() * 1000), ) continue @@ -3964,9 +3992,9 @@ def _resolve_outbound_session_links(conn: sqlite3.Connection, session_id: str, o """Resolve ``session_id``'s own unresolved outbound edges (it is the child). polylogue-4ts.10: candidates are evaluated one at a time (rather than a - single blanket UPDATE) so each can be cycle-checked against - ``sessions.parent_session_id`` before being resolved -- a candidate whose - resolution would close a loop is quarantined instead, never resolved. + single blanket UPDATE) so each can be checked against + ``sessions.parent_session_id`` before being resolved. A candidate whose + resolution would close a loop or exhaust the walk budget is quarantined. """ candidates = conn.execute( """ @@ -3982,15 +4010,15 @@ def _resolve_outbound_session_links(conn: sqlite3.Connection, session_id: str, o (session_id,), ).fetchall() for dst_origin, dst_native_id, link_type, proposed_parent_id in candidates: - cycle_path = _would_create_cycle(conn, child_id=session_id, proposed_parent_id=proposed_parent_id) - if cycle_path is not None: + cycle_walk = _would_create_cycle(conn, child_id=session_id, proposed_parent_id=proposed_parent_id) + if cycle_walk.outcome != "acyclic": _quarantine_session_link( conn, src_session_id=session_id, dst_origin=dst_origin, dst_native_id=dst_native_id, link_type=link_type, - cycle_path=cycle_path, + cycle_walk=cycle_walk, observed_at_ms=int(time.time() * 1000), ) continue @@ -4019,6 +4047,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 +5729,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 +6134,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 +6461,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/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index fd9222171d..b4ff8048a6 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1990,6 +1990,245 @@ def test_rebuild_index_force_write_option_is_retired(cli_runner: CliRunner) -> N assert "--force-write" in result.output +def test_rebuild_index_preflight_reports_durable_schema_currency( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + root = cli_workspace["archive_root"] + with sqlite3.connect(root / "source.db") as conn: + conn.execute("DROP INDEX idx_raw_failure_disposition_receipts_disposed_at") + conn.execute("DROP TABLE raw_failure_disposition_receipts") + conn.execute("PRAGMA user_version = 28") + + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "rebuild-index", "--preflight", "--output-format", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + payload = json.loads(result.stdout) + assert payload["kind"] == "rebuild-schema-currency" + assert payload["status"] == "blocked" + assert [tier["tier"] for tier in payload["tiers"]] == ["audit", "source", "user"] + assert payload["blocking_tiers"][0]["tier"] == "source" + assert payload["blocking_tiers"][0]["actual_user_version"] == 28 + assert payload["blocking_tiers"][0]["expected_user_version"] == 29 + assert "migrate or deploy before rebuilding" in result.stderr + + +def test_migrate_tier_cli_initializes_only_an_absent_durable_tier( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + audit_db = cli_workspace["archive_root"] / "audit.db" + audit_db.unlink() + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "migrate-tier", + "audit", + "--initialize-missing", + "--output-format", + "json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.stdout) + assert payload["ok"] is True + assert payload["tier"] == "audit" + assert payload["initialized"] is True + assert payload["from_version"] == 0 + assert payload["to_version"] == 1 + with sqlite3.connect(audit_db) as conn: + assert conn.execute("PRAGMA user_version").fetchone() == (1,) + assert conn.execute("PRAGMA integrity_check").fetchone() == ("ok",) + + +def test_migrate_tier_cli_missing_initialization_refuses_an_existing_tier( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + audit_db = cli_workspace["archive_root"] / "audit.db" + before = audit_db.read_bytes() + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "migrate-tier", + "audit", + "--initialize-missing", + "--output-format", + "json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + assert "already exists; refusing missing-tier initialization" in json.loads(result.stdout)["error"] + assert audit_db.read_bytes() == before + + +def test_migrate_tier_cli_missing_initialization_loses_publish_race_without_replacement( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + audit_db = cli_workspace["archive_root"] / "audit.db" + audit_db.unlink() + raced_bytes = b"concurrent durable owner\n" + real_link = os.link + + def create_target_before_publish( + source: os.PathLike[str] | str, + destination: os.PathLike[str] | str, + *, + src_dir_fd: int | None = None, + dst_dir_fd: int | None = None, + follow_symlinks: bool = True, + ) -> None: + Path(destination).write_bytes(raced_bytes) + real_link( + source, + destination, + src_dir_fd=src_dir_fd, + dst_dir_fd=dst_dir_fd, + follow_symlinks=follow_symlinks, + ) + + monkeypatch.setattr("polylogue.operations.durable_change_train.os.link", create_target_before_publish) + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "migrate-tier", + "audit", + "--initialize-missing", + "--output-format", + "json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + assert "appeared during initialization; refusing to replace it" in json.loads(result.stdout)["error"] + assert audit_db.read_bytes() == raced_bytes + assert not list(audit_db.parent.glob(".audit.db.initialize-*.tmp")) + + +def test_migrate_tier_cli_exposes_no_named_staging_inode_before_publication( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + audit_db = cli_workspace["archive_root"] / "audit.db" + audit_db.unlink() + real_link = os.link + + def assert_no_named_stage_before_publish( + source: os.PathLike[str] | str, + destination: os.PathLike[str] | str, + *, + src_dir_fd: int | None = None, + dst_dir_fd: int | None = None, + follow_symlinks: bool = True, + ) -> None: + assert not list(audit_db.parent.glob(".audit.db.initialize-*.tmp")) + real_link( + source, + destination, + src_dir_fd=src_dir_fd, + dst_dir_fd=dst_dir_fd, + follow_symlinks=follow_symlinks, + ) + + monkeypatch.setattr( + "polylogue.operations.durable_change_train.os.link", + assert_no_named_stage_before_publish, + ) + + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "migrate-tier", + "audit", + "--initialize-missing", + "--output-format", + "json", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + with sqlite3.connect(audit_db) as conn: + assert conn.execute("PRAGMA user_version").fetchone() == (1,) + assert conn.execute("PRAGMA integrity_check").fetchone() == ("ok",) + assert not list(audit_db.parent.glob(".audit.db.initialize-*.tmp")) + + +def test_rebuild_index_empty_source_still_runs_the_schema_currency_guard( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + root = cli_workspace["archive_root"] + with sqlite3.connect(root / "audit.db") as conn: + expected = int(conn.execute("PRAGMA user_version").fetchone()[0]) + conn.execute(f"PRAGMA user_version = {expected + 1}") + + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "rebuild-index", "--output-format", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + assert "audit.db" in result.stderr + assert not (root / ".index-generations").exists() + + +def test_rebuild_index_empty_source_preserves_plain_receipt_output_after_guard( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + """The real empty receipt must render without replay-only counter keys. + + Mutation: removing the status branch reaches the production counter + formatter and raises KeyError before this exact plain output is emitted. + """ + root = cli_workspace["archive_root"] + receipt_path = write_valid_rebuild_receipt(root, root.parent / "schema-inference-gate-receipt.json") + monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) + + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "rebuild-index"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert result.stdout == f"Archive root: {root}\nNo source.db raw_sessions rows found.\n" + assert not (root / ".index-generations").exists() + + +def test_rebuild_index_rejects_daemon_schema_preflight_combination( + cli_workspace: dict[str, Path], cli_runner: CliRunner +) -> None: + result = cli_runner.invoke( + cli, + ["--plain", "ops", "maintenance", "rebuild-index", "--preflight", "--daemon"], + catch_exceptions=False, + ) + + assert result.exit_code == 2 + assert "--preflight cannot be combined with --daemon" in result.output + + def test_rebuild_index_daemon_path_posts_the_real_selection_request( cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/daemon/test_bulk_rebuild.py b/tests/unit/daemon/test_bulk_rebuild.py index 2ee3079a9c..342a87c2e7 100644 --- a/tests/unit/daemon/test_bulk_rebuild.py +++ b/tests/unit/daemon/test_bulk_rebuild.py @@ -48,7 +48,13 @@ run_daemon_bulk_rebuild_pass, ) from polylogue.daemon.parse_prefetch import DaemonParseStage -from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + RebuildSchemaCurrencyError, + rebuild_index_from_source_sync, +) +from polylogue.storage.archive_identity import ArchiveLocation, OwnedArchiveLocation, assert_owns_archive_location +from polylogue.storage.archive_readiness import probe_archive_tier from polylogue.storage.index_generation import ( IndexGenerationStore, rebuild_source_evidence_snapshot, @@ -56,6 +62,7 @@ ) from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from tests.infra.rebuild_receipt import write_valid_rebuild_receipt _RAW_COUNT = 6 @@ -144,6 +151,57 @@ def test_daemon_bulk_rebuild_refuses_unexplained_failures_before_generation_or_p parse_stage.warm_raw_ids.assert_not_called() +def test_daemon_bulk_pass_rechecks_schema_currency_in_page_selection_hold( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A migration between transaction resolution and page selection blocks. + + Production dependency: the second ownership-bound currency check in + ``run_daemon_bulk_rebuild_pass``. Mutation: removing that check reaches the + fail-fast ``next_raw_page`` replacement below instead of raising the schema + diagnostic before receipt or source-page consumption. + """ + from polylogue.daemon import bulk_rebuild + + _seed_corpus(tmp_path, count=1) + receipt_path = write_valid_rebuild_receipt(tmp_path, tmp_path.parent / f"{tmp_path.name}-receipt.json") + monkeypatch.setenv("POLYLOGUE_SCHEMA_INFERENCE_RECEIPT", str(receipt_path)) + real_assert = assert_owns_archive_location + assertion_count = 0 + + def advance_audit_after_page_ownership(owned: OwnedArchiveLocation, location: ArchiveLocation) -> None: + nonlocal assertion_count + real_assert(owned, location) + assertion_count += 1 + if assertion_count != 2: + return + audit_probe = probe_archive_tier(ArchiveTier.AUDIT, tmp_path / "audit.db") + with sqlite3.connect(tmp_path / "audit.db") as conn: + conn.execute(f"PRAGMA user_version = {audit_probe.expected_user_version + 1}") + + monkeypatch.setattr(bulk_rebuild, "assert_owns_archive_location", advance_audit_after_page_ownership) + next_raw_page = Mock(side_effect=AssertionError("source page selected before schema currency recheck")) + monkeypatch.setattr(IndexGenerationStore, "next_raw_page", next_raw_page) + parse_stage = Mock() + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + asyncio.run( + run_daemon_bulk_rebuild_pass( + config=_config(tmp_path), + parse_stage=parse_stage, + batch_size=1, + max_payload_bytes=10_000, + ) + ) + + blocking_tiers = exc_info.value.diagnostic["blocking_tiers"] + assert isinstance(blocking_tiers, list) + assert blocking_tiers[0]["tier"] == "audit" + assert assertion_count == 2 + next_raw_page.assert_not_called() + parse_stage.warm_raw_ids.assert_not_called() + + def _table_rows(conn: sqlite3.Connection, table: str) -> tuple[tuple[Any, ...], ...]: columns = tuple(row["name"] for row in conn.execute(f'PRAGMA table_xinfo("{table}")')) quoted = ", ".join(f'"{column}"' for column in columns) diff --git a/tests/unit/daemon/test_bulk_rebuild_ownership.py b/tests/unit/daemon/test_bulk_rebuild_ownership.py index 1868b7eb16..db09a35232 100644 --- a/tests/unit/daemon/test_bulk_rebuild_ownership.py +++ b/tests/unit/daemon/test_bulk_rebuild_ownership.py @@ -15,19 +15,81 @@ from __future__ import annotations +import sqlite3 from pathlib import Path +from typing import cast import pytest from polylogue.daemon.bulk_rebuild import resolve_or_start_daemon_bulk_rebuild_transaction -from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation +from polylogue.maintenance.rebuild_index import RebuildSchemaCurrencyError +from polylogue.storage.archive_identity import ( + ArchiveLocation, + ArchiveOwnershipError, + OwnedArchiveLocation, + assert_owns_archive_location, +) +from polylogue.storage.archive_readiness import probe_archive_tier from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS +from tests.infra.rebuild_receipt import write_valid_rebuild_receipt def _init_empty_source(root: Path) -> None: root.mkdir(parents=True, exist_ok=True) - initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) + + +def test_daemon_bulk_rebuild_rejects_schema_mismatch_before_transaction_bookkeeping(tmp_path: Path) -> None: + """The daemon's direct transaction entry cannot bypass the shared gate.""" + root = tmp_path / "archive" + _init_empty_source(root) + source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db") + with sqlite3.connect(root / "source.db") as conn: + conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}") + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + resolve_or_start_daemon_bulk_rebuild_transaction(root) + + blocking_tiers = cast(list[dict[str, object]], exc_info.value.diagnostic["blocking_tiers"]) + assert blocking_tiers[0]["tier"] == "source" + assert not (root / ".index-generations").exists() + assert not (root / ".index-rebuild-transactions").exists() + + +def test_daemon_bulk_rebuild_rechecks_schema_currency_after_ownership( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A durable migration while lock acquisition waits must block bookkeeping. + + Production dependency: the second shared currency probe after ownership. + Mutation: removing that probe creates generation bookkeeping after the + injected audit migration and makes this test fail. + """ + from polylogue.daemon import bulk_rebuild + + root = tmp_path / "archive" + _init_empty_source(root) + receipt = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") + real_assert = assert_owns_archive_location + + def mutate_audit_after_ownership(owned: OwnedArchiveLocation, location: ArchiveLocation) -> None: + real_assert(owned, location) + audit_probe = probe_archive_tier(ArchiveTier.AUDIT, root / "audit.db") + with sqlite3.connect(root / "audit.db") as conn: + conn.execute(f"PRAGMA user_version = {audit_probe.expected_user_version + 1}") + + monkeypatch.setattr(bulk_rebuild, "assert_owns_archive_location", mutate_audit_after_ownership) + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt) + + blocking_tiers = cast(list[dict[str, object]], exc_info.value.diagnostic["blocking_tiers"]) + assert blocking_tiers[0]["tier"] == "audit" + assert not (root / ".index-generations").exists() + assert not (root / ".index-rebuild-transactions").exists() def test_daemon_bulk_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None: diff --git a/tests/unit/daemon/test_daemon_http_contracts.py b/tests/unit/daemon/test_daemon_http_contracts.py index ea6e098c10..84a14eae43 100644 --- a/tests/unit/daemon/test_daemon_http_contracts.py +++ b/tests/unit/daemon/test_daemon_http_contracts.py @@ -40,6 +40,7 @@ from http import HTTPStatus from io import BytesIO from pathlib import Path +from types import SimpleNamespace from typing import TYPE_CHECKING, cast from unittest.mock import MagicMock @@ -193,6 +194,48 @@ def _archive_state_hash(archive_root: Path) -> str: return h.hexdigest() +def test_rebuild_index_schema_currency_conflict_preserves_preflight_diagnostic( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The actual maintenance route returns the shared diagnostic, not a 500.""" + from polylogue.storage.archive_readiness import probe_archive_tier + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS + + root = tmp_path / "archive" + root.mkdir() + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) + source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db") + with sqlite3.connect(root / "source.db") as conn: + conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}") + monkeypatch.setattr("polylogue.paths.archive_root", lambda: root) + + handler = _make_handler("POST", "/api/maintenance/rebuild-index", body=b"{}") + handler.server.write_bridge = SimpleNamespace( # type: ignore[assignment] + run_sync_with_timeout=lambda _actor, _timeout, operation, request: operation(request) + ) + send_error, send_json = _capture_responses(handler) + + handler._handle_rebuild_index() + + send_error.assert_not_called() + status, payload = send_json.call_args.args + assert status == HTTPStatus.CONFLICT + assert payload["kind"] == "rebuild-schema-currency" + assert payload["status"] == "blocked" + assert payload["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": source_probe.expected_user_version + 1, + "expected_user_version": source_probe.expected_user_version, + "status": "mismatch", + } + ] + + def test_cli_query_post_forwards_root_request_to_daemon_compiler() -> None: """The UDS-only envelope carries raw root flags, not a client-built SQL query.""" diff --git a/tests/unit/devtools/test_lineage_validation.py b/tests/unit/devtools/test_lineage_validation.py index 7880023694..dbebd74b25 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,254 @@ 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_samples_distinct_unresolved_edges_without_multiplying_messages(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( + """ + 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', 'subagent', NULL, + NULL, 'parent-tool-use-id', '{}', NULL, 'spawned-fresh') + """ + ) + conn.commit() + + report = lineage_validation.build_report(_args(archive_root)) + + sample = report["lineage"]["topology"]["unresolved_read_sample"] + assert sample["safe"] is True + assert sample["sampled"] == 2 + assert {row["link_type"] for row in sample["rows"]} == {"continuation", "subagent"} + assert {row["stored_messages"] for row in sample["rows"]} == {1} + assert {row["served_messages"] for row in sample["rows"]} == {1} + 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_snapshot_is_stable_for_a_quiescent_wal_database(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + db = _make_index_db(archive_root) + with sqlite3.connect(db) as writer: + assert writer.execute("PRAGMA journal_mode = WAL").fetchone() == ("wal",) + writer.execute("UPDATE session_links SET method = 'wal-proof' WHERE src_session_id = 'child'") + writer.commit() + + report = lineage_validation.build_report(_args(archive_root)) + + assert report["snapshot_identity"]["stable"] is True + assert report["snapshot_identity"]["file_set_stable"] is True + assert report["snapshot_identity"]["no_concurrent_commits"] is True + assert report["verdict"]["external_counts_citable"] is True + + +def test_lineage_validation_rejects_commit_between_reader_snapshot_and_file_hash( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive_root = tmp_path / "archive" + db = _make_index_db(archive_root) + writer = sqlite3.connect(db) + assert writer.execute("PRAGMA journal_mode = WAL").fetchone() == ("wal",) + original_snapshot_identity = lineage_validation._snapshot_identity + snapshot_calls = 0 + + def commit_before_first_file_hash(index_db: Path) -> dict[str, object]: + nonlocal snapshot_calls + if snapshot_calls == 0: + writer.execute("UPDATE session_links SET method = 'concurrent' WHERE src_session_id = 'child'") + writer.commit() + snapshot_calls += 1 + return original_snapshot_identity(index_db) + + monkeypatch.setattr(lineage_validation, "_snapshot_identity", commit_before_first_file_hash) + try: + report = lineage_validation.build_report(_args(archive_root)) + finally: + writer.close() + + identity = report["snapshot_identity"] + assert identity["file_set_stable"] is True + assert identity["no_concurrent_commits"] is False + assert identity["observer_data_version_after"] > identity["observer_data_version_before"] + assert identity["stable"] is False + assert report["verdict"]["external_counts_citable"] is False + assert "index received a concurrent commit during the read-only census" in report["verdict"]["reasons"] + + +def test_lineage_validation_rejects_budget_exhaustion_as_cycle_proof(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 status = 'quarantined', + resolved_dst_session_id = NULL, + evidence_json = ? + WHERE src_session_id = 'child' + """, + ( + json.dumps( + { + "reason": "cycle_walk_budget_exhausted", + "walk_path": ["child", "parent", "fresh"], + "walk_budget": 1, + "detected_at_ms": 1, + } + ), + ), + ) + conn.execute("UPDATE sessions SET parent_session_id = NULL WHERE session_id = 'child'") + conn.execute("UPDATE sessions SET parent_session_id = 'fresh' WHERE session_id = 'parent'") + + report = lineage_validation.build_report(_args(archive_root)) + + topology = report["lineage"]["topology"] + assert topology["cycle_evidence_count"] == 0 + assert topology["budget_exhausted_quarantine_evidence_count"] == 1 + assert topology["quarantined_without_cycle_evidence"] == 1 + assert report["verdict"]["external_counts_citable"] is False + assert ( + "1 quarantined topology links only have cycle-walk budget exhaustion evidence" in report["verdict"]["reasons"] + ) + + +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 +551,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/maintenance/test_rebuild_index_ownership.py b/tests/unit/maintenance/test_rebuild_index_ownership.py index ab5b43fcf8..afa31a2bd0 100644 --- a/tests/unit/maintenance/test_rebuild_index_ownership.py +++ b/tests/unit/maintenance/test_rebuild_index_ownership.py @@ -13,18 +13,145 @@ import sqlite3 from pathlib import Path +from typing import cast import pytest -from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + RebuildSchemaCurrencyError, + rebuild_index_from_source_sync, +) from polylogue.storage.archive_identity import ArchiveLocation, ArchiveOwnershipError, OwnedArchiveLocation -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.archive_readiness import probe_archive_tier +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS +from tests.infra.rebuild_receipt import write_valid_rebuild_receipt def _init_empty_source(root: Path) -> None: root.mkdir(parents=True, exist_ok=True) - initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) + + +def test_rebuild_rejects_source_schema_behind_runtime_before_candidate_creation(tmp_path: Path) -> None: + """A real v28 source tier must not reach the v29 rebuild package. + + The test builds ordinary file-backed archive tiers, removes exactly v29's + additive objects, and supplies a valid rebuild receipt. The production + rebuild route used to accept this archive and return ``empty-source``. + """ + root = tmp_path / "archive" + initialize_active_archive_root(root) + with sqlite3.connect(root / "source.db") as conn: + conn.execute("DROP INDEX idx_raw_failure_disposition_receipts_disposed_at") + conn.execute("DROP TABLE raw_failure_disposition_receipts") + conn.execute("PRAGMA user_version = 28") + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) + + diagnostic = exc_info.value.diagnostic + assert diagnostic["status"] == "blocked" + assert diagnostic["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": 28, + "expected_user_version": 29, + "status": "mismatch", + } + ] + assert not (root / ".index-generations").exists() + assert not (root / ".index-rebuild-transactions").exists() + + +def test_rebuild_rejects_source_schema_ahead_of_runtime_before_candidate_creation(tmp_path: Path) -> None: + """A newer source tier is as unsafe to rebuild as an older one.""" + root = tmp_path / "archive" + _init_empty_source(root) + source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db") + with sqlite3.connect(root / "source.db") as conn: + conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}") + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + + assert exc_info.value.diagnostic["blocking_tiers"] == [ + { + "tier": "source", + "path": str(root / "source.db"), + "actual_user_version": source_probe.expected_user_version + 1, + "expected_user_version": source_probe.expected_user_version, + "status": "mismatch", + } + ] + assert not (root / ".index-generations").exists() + + +@pytest.mark.parametrize("mode", ["missing", "mismatched"]) +def test_rebuild_rejects_missing_or_mismatched_audit_tier_before_candidate_creation(tmp_path: Path, mode: str) -> None: + """Every canonical durable tier, including audit, must be package-current.""" + root = tmp_path / "archive" + _init_empty_source(root) + audit_path = root / "audit.db" + expected = probe_archive_tier(ArchiveTier.AUDIT, audit_path).expected_user_version + if mode == "missing": + audit_path.unlink() + actual: int | None = None + status = "missing" + else: + with sqlite3.connect(audit_path) as conn: + conn.execute(f"PRAGMA user_version = {expected + 1}") + actual = expected + 1 + status = "mismatch" + + with pytest.raises(RebuildSchemaCurrencyError) as exc_info: + rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + + assert exc_info.value.diagnostic["blocking_tiers"] == [ + { + "tier": "audit", + "path": str(audit_path), + "actual_user_version": actual, + "expected_user_version": expected, + "status": status, + } + ] + assert not (root / ".index-generations").exists() + + +def test_rebuild_rechecks_schema_currency_after_acquiring_archive_ownership( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Schema drift after the early guard cannot reach the candidate path.""" + root = tmp_path / "archive" + _init_empty_source(root) + receipt_path = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-receipt.json") + source_probe = probe_archive_tier(ArchiveTier.SOURCE, root / "source.db") + original_acquire = OwnedArchiveLocation.acquire + + def acquire_then_advance_schema(location: ArchiveLocation) -> OwnedArchiveLocation: + owned = original_acquire(location) + with sqlite3.connect(root / "source.db") as conn: + conn.execute(f"PRAGMA user_version = {source_probe.expected_user_version + 1}") + return owned + + monkeypatch.setattr("polylogue.maintenance.rebuild_index.OwnedArchiveLocation.acquire", acquire_then_advance_schema) + + with pytest.raises(RebuildSchemaCurrencyError, match="schema currency") as exc_info: + rebuild_index_from_source_sync( + RebuildIndexRequest(archive_root=root, schema_inference_receipt_path=receipt_path) + ) + + blocking_tiers = cast(list[dict[str, object]], exc_info.value.diagnostic["blocking_tiers"]) + assert blocking_tiers[0]["tier"] == "source" + assert not (root / ".index-generations").exists() def test_rebuild_refuses_when_archive_location_already_owned(tmp_path: Path) -> None: diff --git a/tests/unit/maintenance/test_rebuild_status.py b/tests/unit/maintenance/test_rebuild_status.py index 222713cce0..d2c619dee1 100644 --- a/tests/unit/maintenance/test_rebuild_status.py +++ b/tests/unit/maintenance/test_rebuild_status.py @@ -26,14 +26,16 @@ from polylogue.storage.index_generation import IndexGenerationStore, rebuild_source_evidence_snapshot from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database -from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.migration_runner import DURABLE_MIGRATION_TIERS +from tests.infra.rebuild_receipt import write_valid_rebuild_receipt _DEFINITELY_DEAD_PID = 2**31 - 1 def _init_empty_source(root: Path) -> None: root.mkdir(parents=True, exist_ok=True) - initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) + for tier in sorted(DURABLE_MIGRATION_TIERS, key=lambda item: item.value): + initialize_archive_database(root / f"{tier.value}.db", tier) def _codex_session(native_id: str) -> bytes: @@ -192,7 +194,8 @@ def test_falls_back_to_the_daemon_well_known_operation_id_by_default(tmp_path: P root = tmp_path / "archive" _init_empty_source(root) - resolve_or_start_daemon_bulk_rebuild_transaction(root) + receipt = write_valid_rebuild_receipt(root, tmp_path / "schema-inference-gate-receipt.json") + resolve_or_start_daemon_bulk_rebuild_transaction(root, schema_inference_receipt_path=receipt) status = rebuild_status(root) diff --git a/tests/unit/storage/test_topology_cycle_quarantine_live.py b/tests/unit/storage/test_topology_cycle_quarantine_live.py index 79b89fc3c3..b6ac34d0ca 100644 --- a/tests/unit/storage/test_topology_cycle_quarantine_live.py +++ b/tests/unit/storage/test_topology_cycle_quarantine_live.py @@ -27,13 +27,18 @@ from pathlib import Path from typing import cast +import aiosqlite +import pytest + +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 +from polylogue.storage.sqlite.queries.message_query_reads import get_messages def _connect(path: Path) -> sqlite3.Connection: @@ -66,7 +71,8 @@ def _link_row(conn: sqlite3.Connection, src_session_id: str) -> sqlite3.Row: return cast(sqlite3.Row, row) -def test_cross_ingest_cycle_quarantines_the_closing_edge(tmp_path: Path) -> None: +@pytest.mark.asyncio +async def test_cross_ingest_cycle_quarantines_the_closing_edge_without_losing_prefix(tmp_path: Path) -> None: db = tmp_path / "index.db" conn = _connect(db) @@ -99,7 +105,10 @@ def test_cross_ingest_cycle_quarantines_the_closing_edge(tmp_path: Path) -> None provider_session_id="A", title="A", parent_session_provider_id="B", - messages=[_msg("a0", Role.USER, "start", 0), _msg("a1", Role.ASSISTANT, "revised", 1)], + messages=[ + _msg("a-copy-b0", Role.USER, "child of A", 0), + _msg("a1", Role.ASSISTANT, "revised", 1), + ], ) write_parsed_session_to_archive(conn, session_a_v2, force_replace=True) @@ -112,6 +121,25 @@ def test_cross_ingest_cycle_quarantines_the_closing_edge(tmp_path: Path) -> None assert a_id in evidence["cycle_path"] assert b_id in evidence["cycle_path"] + # Anti-vacuity: the first A message exactly matches B's full stored + # transcript. Without the pre-normalization cycle check, the writer slices + # it as an inherited prefix before quarantining A -> B, and both production + # readers then serve only the second message. + own_message_ids = [ + str(row[0]) + for row in conn.execute( + "SELECT message_id FROM messages WHERE session_id = ? ORDER BY position, variant_index", + (a_id,), + ).fetchall() + ] + assert len(own_message_ids) == 2 + quarantined_envelope = read_archive_session_envelope(conn, a_id) + assert [message.message_id for message in quarantined_envelope.messages] == own_message_ids + async with aiosqlite.connect(db) as reader: + reader.row_factory = sqlite3.Row + async_messages = await get_messages(reader, a_id) + assert [message.message_id for message in async_messages] == own_message_ids + # A's parent_session_id fast-path projection must stay NULL -- the # composition/ancestry walk must never enter the cycle. assert conn.execute("SELECT parent_session_id FROM sessions WHERE session_id = ?", (a_id,)).fetchone()[0] is None @@ -121,6 +149,89 @@ 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["budget_exhausted_quarantine_evidence_count"] == 0 + assert census["quarantined_with_resolved_parent_count"] == 0 + assert census["quarantined_with_stale_projection_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 + + unrelated_evidence = json.dumps( + { + "reason": "cycle_rejected", + "cycle_path": ["unrelated-a", "unrelated-b"], + "detected_at_ms": 1, + } + ) + conn.execute( + "UPDATE session_links SET evidence_json = ? WHERE src_session_id = ?", + (unrelated_evidence, a_id), + ) + unrelated = census_topology_links(conn, sample_unresolved=0) + assert unrelated["cycle_evidence_count"] == 0 + assert unrelated["malformed_quarantine_evidence_count"] == 1 + assert unrelated["budget_exhausted_quarantine_evidence_count"] == 0 + + fabricated_cycle_evidence = json.dumps( + { + "reason": "cycle_rejected", + "cycle_path": [a_id, b_id, a_id], + "detected_at_ms": 1, + } + ) + conn.execute("UPDATE sessions SET parent_session_id = NULL WHERE session_id = ?", (b_id,)) + conn.execute( + "UPDATE session_links SET evidence_json = ? WHERE src_session_id = ?", + (fabricated_cycle_evidence, a_id), + ) + fabricated = census_topology_links(conn, sample_unresolved=0) + assert fabricated["cycle_evidence_count"] == 0 + assert fabricated["malformed_quarantine_evidence_count"] == 1 + assert fabricated["budget_exhausted_quarantine_evidence_count"] == 0 + assert fabricated["quarantined_without_cycle_evidence"] == 1 + conn.execute("UPDATE sessions SET parent_session_id = ? WHERE session_id = ?", (a_id, b_id)) + + 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), + ) + conn.execute( + "UPDATE sessions SET parent_session_id = ?, root_session_id = ? WHERE session_id = ?", + (b_id, b_id, a_id), + ) + quarantined_read = read_archive_session_envelope(conn, a_id) + assert quarantined_read.lineage_inheritance == "none" + assert [message.message_id for message in quarantined_read.messages] == own_message_ids + contradictory = census_topology_links(conn, sample_unresolved=0) + assert contradictory["quarantined_with_resolved_parent_count"] == 1 + assert contradictory["quarantined_with_stale_projection_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" @@ -152,6 +263,138 @@ def test_self_referential_edge_quarantines_without_touching_projection(tmp_path: ) +def test_over_budget_acyclic_walk_is_not_recorded_as_a_cycle_and_keeps_prefix(tmp_path: Path) -> None: + """The live writer must distinguish an indeterminate deep walk from a cycle. + + Production dependencies: pre-slice cycle classification, outbound link + quarantine, and the synchronous composed reader. Mutation: returning a + cycle path at the walk budget records `cycle_rejected`; treating exhaustion + as acyclic slices the copied parent prefix and serves only the tail. + """ + db = tmp_path / "index.db" + conn = _connect(db) + parent = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="deep-0", + title="deep parent", + messages=[_msg("p0", Role.USER, "copied parent prefix", 0)], + ) + parent_id = write_parsed_session_to_archive(conn, parent) + child_v1 = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="deep-child", + title="deep child", + messages=[_msg("c0", Role.USER, "original child", 0)], + ) + child_id = write_parsed_session_to_archive(conn, child_v1) + + for position in range(1024, 0, -1): + native_id = f"deep-{position}" + parent_session_id = None if position == 1024 else f"codex-session:deep-{position + 1}" + conn.execute( + """ + INSERT INTO sessions(native_id, origin, parent_session_id, content_hash) + VALUES (?, 'codex-session', ?, ?) + """, + (native_id, parent_session_id, bytes(32)), + ) + conn.execute( + "UPDATE sessions SET parent_session_id = ? WHERE session_id = ?", + ("codex-session:deep-1", parent_id), + ) + + child_v2 = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="deep-child", + title="deep child", + parent_session_provider_id="deep-0", + messages=[ + _msg("copy-p0", Role.USER, "copied parent prefix", 0), + _msg("c1", Role.ASSISTANT, "child tail", 1), + ], + ) + write_parsed_session_to_archive(conn, child_v2, force_replace=True) + + link = _link_row(conn, child_id) + evidence = json.loads(link["evidence_json"]) + assert link["status"] == TopologyEdgeStatus.QUARANTINED.value + assert evidence["reason"] == "cycle_walk_budget_exhausted" + assert "cycle_path" not in evidence + assert evidence["walk_budget"] == 1024 + assert len(evidence["walk_path"]) == 1026 + own_message_ids = [ + str(row[0]) + for row in conn.execute( + "SELECT message_id FROM messages WHERE session_id = ? ORDER BY position, variant_index", + (child_id,), + ).fetchall() + ] + assert len(own_message_ids) == 2 + envelope = read_archive_session_envelope(conn, child_id) + assert [message.message_id for message in envelope.messages] == own_message_ids + census = census_topology_links(conn, sample_unresolved=0) + assert census["cycle_evidence_count"] == 0 + assert census["malformed_quarantine_evidence_count"] == 0 + assert census["budget_exhausted_quarantine_evidence_count"] == 1 + assert census["quarantined_without_cycle_evidence"] == 1 + + +def test_quarantined_alternate_parent_does_not_invalidate_resolved_projection(tmp_path: Path) -> None: + """A valid parent projection can coexist with a rejected alternate edge. + + Production dependencies: repeated writer assertions, cycle quarantine, + projection refresh, and the topology census. Mutation: counting any parent + pointer on a child with a quarantined edge reports this valid projection as + stale even though the child's earlier resolved edge still supports it. + """ + conn = _connect(tmp_path / "index.db") + session_a = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="multi-A", + title="A", + messages=[_msg("a0", Role.USER, "root A", 0)], + ) + a_id = write_parsed_session_to_archive(conn, session_a) + session_b_v1 = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="multi-B", + title="B", + parent_session_provider_id="multi-A", + messages=[_msg("b0", Role.USER, "B follows A", 0)], + ) + b_id = write_parsed_session_to_archive(conn, session_b_v1) + session_c = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="multi-C", + title="C", + parent_session_provider_id="multi-B", + messages=[_msg("c0", Role.USER, "C follows B", 0)], + ) + write_parsed_session_to_archive(conn, session_c) + + session_b_v2 = ParsedSession( + source_name=Provider.CODEX, + provider_session_id="multi-B", + title="B", + parent_session_provider_id="multi-C", + messages=[_msg("b1", Role.USER, "B asserts unsafe alternate C", 0)], + ) + write_parsed_session_to_archive(conn, session_b_v2, merge_append=True) + + links = conn.execute( + "SELECT dst_native_id, status FROM session_links WHERE src_session_id = ? ORDER BY dst_native_id", + (b_id,), + ).fetchall() + assert [(row[0], row[1]) for row in links] == [ + ("multi-A", None), + ("multi-C", TopologyEdgeStatus.QUARANTINED.value), + ] + assert conn.execute("SELECT parent_session_id FROM sessions WHERE session_id = ?", (b_id,)).fetchone()[0] == a_id + census = census_topology_links(conn, sample_unresolved=0) + assert census["cycle_evidence_count"] == 1 + assert census["quarantined_with_stale_projection_count"] == 0 + + def test_diamond_dag_is_not_mistaken_for_a_cycle(tmp_path: Path) -> None: """B -> D and C -> D (both children of D) is a legitimate shared-parent shape, not a cycle -- the resolver must resolve both edges cleanly."""