diff --git a/continuityos/demo.py b/continuityos/demo.py index e29f15e3..f3875ddc 100644 --- a/continuityos/demo.py +++ b/continuityos/demo.py @@ -402,7 +402,7 @@ def main(argv: Sequence[str] | None = None) -> int: prog="cos demo", description="Run self-contained product proofs without touching your normal memory DB", ) - parser.add_argument("demo", choices=["continuity"]) + parser.add_argument("demo", choices=["continuity", "person-continuity"]) parser.add_argument("--json", action="store_true", dest="as_json") parser.add_argument("--timeout", type=float, default=20.0, help="fresh-process probe timeout in seconds") parser.add_argument("--db", dest="forbidden_db", default=None, help=argparse.SUPPRESS) @@ -411,11 +411,17 @@ def main(argv: Sequence[str] | None = None) -> int: if args.forbidden_db is not None: value = _hold("USER_DB_ARGUMENT_NOT_ALLOWED", effects=_effects(cleanup_pass=True)) code = 2 + elif args.demo == "person-continuity": + from .person_twin_demo import run_person_continuity + value, code = run_person_continuity(timeout=args.timeout) else: value, code = run_continuity(timeout=max(1.0, args.timeout)) if args.as_json: print(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True)) + elif args.demo == "person-continuity" and args.forbidden_db is None: + from .person_twin_demo import render_person_continuity + print(render_person_continuity(value)) else: print(_render(value)) return code diff --git a/continuityos/person_twin_admission_store.py b/continuityos/person_twin_admission_store.py new file mode 100644 index 00000000..aed05975 --- /dev/null +++ b/continuityos/person_twin_admission_store.py @@ -0,0 +1,534 @@ +"""Durable, synthetic-qualified R3 admission in an existing Memory SQLite file. + +This is a trusted-host storage adapter, not an authority issuer. It persists the +actual R3 receipt and verifies it after COMMIT through a new read-only connection. +Fresh consent, time, policy, session and disclosure authority remain external. +No CLI, running database, R3/R4 implementation or current-mode ceiling is changed. +""" +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +import math +from pathlib import Path +import sqlite3 +import stat +import uuid +from typing import Any, Callable, Mapping + +from .current_effect_boundary import assert_current_effect_allowed +from .person_twin_admission_contracts import validate_person_twin_identity +from .person_twin_admission_coordinator import ( + AdmissionStoreConflict, AdmissionStoreError, PersonTwinAdmissionCoordinator, +) +from .person_twin_authority_replay import _current_admission +from .person_twin_privacy_provenance import validate_person_twin_record +from .person_twin_r21h_evidence import _require_current_meta + +SCHEMA = "continuityos.person-twin.durable-admission/v1" +MAX_RECORD_BYTES = 1024 * 1024 # bounded lab input, not a calibrated production limit +_CEILING = dict(production_admission_status="NOT_PRODUCTION_ADMITTED", + execution_authority="NONE", can_execute=False, can_trade=False, + capital_permission="DENY") +_DDL = ( + "CREATE TABLE person_admission_meta (slot INTEGER PRIMARY KEY CHECK(slot=1), schema_id TEXT NOT NULL)", + "CREATE TABLE person_admission_candidates (twin_key TEXT NOT NULL, candidate_id TEXT NOT NULL, candidate BLOB NOT NULL, candidate_hash TEXT NOT NULL, anchors BLOB NOT NULL, anchors_hash TEXT NOT NULL, binding BLOB NOT NULL, binding_hash TEXT NOT NULL, PRIMARY KEY(twin_key,candidate_id))", + "CREATE TABLE person_admission_publications (publication_id TEXT PRIMARY KEY, twin_key TEXT NOT NULL, candidate_id TEXT NOT NULL, pointer BLOB NOT NULL, pointer_hash TEXT NOT NULL, receipt BLOB NOT NULL, receipt_hash TEXT NOT NULL, binding_hash TEXT NOT NULL, previous_publication_id TEXT, UNIQUE(publication_id,twin_key,pointer_hash), FOREIGN KEY(twin_key,candidate_id) REFERENCES person_admission_candidates(twin_key,candidate_id))", + "CREATE TABLE person_admission_current (twin_key TEXT PRIMARY KEY, publication_id TEXT NOT NULL UNIQUE, pointer BLOB NOT NULL, pointer_hash TEXT NOT NULL, FOREIGN KEY(publication_id,twin_key,pointer_hash) REFERENCES person_admission_publications(publication_id,twin_key,pointer_hash) DEFERRABLE INITIALLY DEFERRED)", +) +_TABLES = tuple(sql.split()[2] for sql in _DDL) + + +class DurableAdmissionError(AdmissionStoreError): + """Fixed-code error; no protected source text or filesystem path is returned.""" + + +def _canonical(value: Any) -> bytes: + raw = json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False).encode("utf-8") + if not 0 < len(raw) <= MAX_RECORD_BYTES: + raise DurableAdmissionError("ADMISSION_RECORD_SIZE_INVALID") + return raw + + +def _hash(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def _decode(raw: bytes) -> dict[str, Any]: + if type(raw) is not bytes or not 0 < len(raw) <= MAX_RECORD_BYTES: + raise DurableAdmissionError("ADMISSION_BYTES_INVALID") + try: + value = json.loads(raw) + if not isinstance(value, dict) or _canonical(value) != raw: + raise ValueError() + return value + except Exception: + raise DurableAdmissionError("ADMISSION_CANONICAL_JSON_INVALID") from None + + +def _identity(value: Mapping[str, Any]) -> tuple[dict[str, Any], str]: + try: + identity = _decode(_canonical(dict(value))) + validate_person_twin_identity(identity) + key = _hash(_canonical({k: identity[k] for k in + ("tenant_id", "twin_id", "identity_fingerprint")})) + return identity, key + except Exception: + raise DurableAdmissionError("ADMISSION_IDENTITY_INVALID") from None + + +def _direct_file(path: Path) -> tuple[int, int]: + try: + for member in (path, *path.parents): + info = member.lstat() + if member.is_symlink() or getattr(info, "st_file_attributes", 0) & 0x400: + raise ValueError() + info = path.stat() + if not stat.S_ISREG(info.st_mode): + raise ValueError() + return info.st_dev, info.st_ino + except Exception: + raise DurableAdmissionError("ADMISSION_DB_PATH_INVALID") from None + + +def _connect(path: Path, read_only: bool, timeout: float) -> sqlite3.Connection: + before = _direct_file(path) + con = None + try: + con = sqlite3.connect(path.as_uri() + ("?mode=ro" if read_only else "?mode=rw"), + uri=True, isolation_level=None, timeout=timeout) + con.row_factory = sqlite3.Row + con.execute("PRAGMA foreign_keys=ON") + con.execute("PRAGMA trusted_schema=OFF") + if read_only: + con.execute("PRAGMA query_only=ON") + else: + con.execute("PRAGMA synchronous=FULL") + if con.execute("PRAGMA foreign_keys").fetchone()[0] != 1: + raise ValueError() + if not read_only and con.execute("PRAGMA synchronous").fetchone()[0] != 2: + raise ValueError() + if before != _direct_file(path): + raise ValueError() + row = con.execute("SELECT type FROM sqlite_master WHERE name='items'").fetchone() + if row is None or row[0] != "table": + raise ValueError() + columns = {r[1] for r in con.execute("PRAGMA table_info(items)")} + if not {"id", "namespace", "text", "meta"}.issubset(columns): + raise ValueError() + return con + except Exception: + if con is not None: + con.close() + raise DurableAdmissionError("ADMISSION_DB_OPEN_FAILED") from None + + +def _schema(con: sqlite3.Connection) -> None: + if con.execute("PRAGMA foreign_keys").fetchone()[0] != 1 or con.execute("PRAGMA trusted_schema").fetchone()[0] != 0: + raise DurableAdmissionError("ADMISSION_CONNECTION_SETTINGS_CHANGED") + if con.execute("PRAGMA query_only").fetchone()[0] == 0 and con.execute("PRAGMA synchronous").fetchone()[0] != 2: + raise DurableAdmissionError("ADMISSION_DURABILITY_SETTING_CHANGED") + for name, expected in zip(_TABLES, _DDL): + row = con.execute("SELECT type,sql FROM sqlite_master WHERE name=?", (name,)).fetchone() + if row is None or row[0] != "table" or row[1] != expected: + raise DurableAdmissionError("ADMISSION_SCHEMA_MISMATCH") + if con.execute("SELECT 1 FROM sqlite_master WHERE type='trigger' AND tbl_name=?", (name,)).fetchone(): + raise DurableAdmissionError("ADMISSION_SCHEMA_TRIGGER_DENIED") + rows = con.execute("SELECT slot,schema_id FROM person_admission_meta").fetchall() + if len(rows) != 1 or tuple(rows[0]) != (1, SCHEMA): + raise DurableAdmissionError("ADMISSION_SCHEMA_VERSION_MISMATCH") + + +@dataclass(frozen=True) +class AdmissionSnapshot: + """Verified stored bytes, not a fresh authorization or a signed attestation.""" + publication_id: str + candidate_bytes: bytes + pointer_bytes: bytes + receipt_bytes: bytes + anchors_bytes: bytes + binding_bytes: bytes + + @property + def candidate(self) -> dict[str, Any]: + return _decode(self.candidate_bytes) + + @property + def pointer(self) -> dict[str, Any]: + return _decode(self.pointer_bytes) + + @property + def receipt(self) -> dict[str, Any]: + return _decode(self.receipt_bytes) + + @property + def anchors(self) -> dict[str, Any]: + return _decode(self.anchors_bytes) + + @property + def binding(self) -> dict[str, Any]: + return _decode(self.binding_bytes) + + +class _Readback: + """Already read SQLite snapshot used by the unchanged R4 structural validator.""" + def __init__(self, candidate_id: str, candidate: bytes, pointer: bytes): + self.cid, self.candidate, self.pointer = candidate_id, candidate, pointer + + def read_candidate_bytes(self, candidate_id: str) -> bytes | None: + return self.candidate if candidate_id == self.cid else None + + def read_current_pointer_bytes(self) -> bytes: + return self.pointer + + +class PersonTwinAdmissionStore: + """One thread-owned connection; independent writers use independent instances. + + Constructor is read-only by default and never creates a file or a schema. + Only admit() can invoke the R3 write callbacks. Direct promotion is rejected. + """ + def __init__(self, path: str | Path, *, identity: Mapping[str, Any], + read_only: bool = True, timeout: float = 2.0): + if type(timeout) not in (int, float) or not math.isfinite(timeout) or not 0 <= timeout <= 10: + raise DurableAdmissionError("ADMISSION_TIMEOUT_INVALID") + if type(read_only) is not bool: + raise DurableAdmissionError("ADMISSION_MODE_INVALID") + self.identity, self.twin_key = _identity(identity) + self.path = Path(path).absolute() + self.read_only, self.timeout = read_only, float(timeout) + if not read_only: + assert_current_effect_allowed("person.admission.open_writer") + self._file_identity = _direct_file(self.path) + self._con = _connect(self.path, read_only, self.timeout) + try: + _schema(self._con) + except Exception: + self._con.close() + raise DurableAdmissionError("ADMISSION_SCHEMA_UNAVAILABLE") from None + self._phase = None + self._active = None + self._expected_publication = None + self._next_publication = None + self._candidate_committed = False + self._candidate_commit_state = "NOT_ATTEMPTED" + self._suspended = False + self._commit_state = "NOT_ATTEMPTED" + + @classmethod + def initialize(cls, path: str | Path) -> None: + """Explicit schema preparation only, in an existing Memory database.""" + assert_current_effect_allowed("person.admission.initialize") + con = _connect(Path(path).absolute(), False, 2.0) + try: + con.execute("BEGIN IMMEDIATE") + existing = [con.execute("SELECT 1 FROM sqlite_master WHERE name=?", (n,)).fetchone() + for n in _TABLES] + if not any(existing): + for ddl in _DDL: + con.execute(ddl) + con.execute("INSERT INTO person_admission_meta VALUES(1,?)", (SCHEMA,)) + _schema(con) + assert_current_effect_allowed("person.admission.initialize_commit") + con.execute("COMMIT") + except Exception: + if con.in_transaction: + con.execute("ROLLBACK") + raise DurableAdmissionError("ADMISSION_SCHEMA_INITIALIZE_FAILED") from None + finally: + con.close() + + def __enter__(self) -> PersonTwinAdmissionStore: + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + def close(self) -> None: + self._con.close() + + def _current_row(self): + return self._con.execute("SELECT * FROM person_admission_current WHERE twin_key=?", + (self.twin_key,)).fetchone() + + def _row_binding(self, record, anchors, memory_id, namespace, *, at=None): + if type(memory_id) is not int or memory_id < 1 or type(namespace) is not str or not namespace.strip(): + raise DurableAdmissionError("ADMISSION_MEMORY_BINDING_INVALID") + row = self._con.execute("SELECT id,namespace,text,meta FROM items WHERE id=?", (memory_id,)).fetchone() + source = anchors["source_record"] + text = source["payload"]["text"] + if row is None or row["namespace"] != namespace or row["text"] != text or not isinstance(text, str) or not text.strip(): + raise DurableAdmissionError("ADMISSION_MEMORY_SOURCE_MISMATCH") + if at is not None: + _require_current_meta(row["meta"], at) + return dict(schema=SCHEMA + "/memory-binding", twin_key=self.twin_key, + memory_id=memory_id, namespace=namespace, candidate_id=record["id"], + candidate_sha256=_hash(_canonical(record)), + source_record_id=source["id"], source_content_hash=source["content_hash"], + text_sha256=_hash(text.encode("utf-8")), anchors_sha256=_hash(_canonical(anchors))) + + def _candidate_row(self, candidate_id): + row = self._con.execute("SELECT * FROM person_admission_candidates WHERE twin_key=? AND candidate_id=?", + (self.twin_key, candidate_id)).fetchone() + if row is None: + return None + for payload, digest in (("candidate", "candidate_hash"), ("anchors", "anchors_hash"), ("binding", "binding_hash")): + _decode(row[payload]) + if _hash(row[payload]) != row[digest]: + raise DurableAdmissionError("ADMISSION_STORED_HASH_MISMATCH") + record, anchors, binding = (_decode(row[k]) for k in ("candidate", "anchors", "binding")) + if set(anchors) != {"identity", "consent_receipt", "source_record", "purpose"} or anchors["identity"] != self.identity: + raise DurableAdmissionError("ADMISSION_STORED_IDENTITY_MISMATCH") + validate_person_twin_record(record, identity=self.identity, + consent_receipt=anchors["consent_receipt"], source_record=anchors["source_record"]) + if record["id"] != candidate_id: + raise DurableAdmissionError("ADMISSION_STORED_CANDIDATE_MISMATCH") + expected = self._row_binding(record, anchors, binding["memory_id"], binding["namespace"]) + if binding != expected: + raise DurableAdmissionError("ADMISSION_STORED_BINDING_MISMATCH") + return row + + def read_candidate_bytes(self, candidate_id: str) -> bytes | None: + try: + row = self._candidate_row(candidate_id) + return None if row is None else row["candidate"] + except Exception: + raise DurableAdmissionError("ADMISSION_CANDIDATE_READ_FAILED") from None + + def _snapshot(self) -> AdmissionSnapshot | None: + current = self._current_row() + if current is None: + return None + pub = self._con.execute("SELECT * FROM person_admission_publications WHERE publication_id=? AND twin_key=?", + (current["publication_id"], self.twin_key)).fetchone() + if pub is None or current["pointer"] != pub["pointer"] or current["pointer_hash"] != pub["pointer_hash"]: + raise DurableAdmissionError("ADMISSION_PUBLICATION_INCOMPLETE") + row = self._candidate_row(pub["candidate_id"]) + if row is None or pub["binding_hash"] != row["binding_hash"]: + raise DurableAdmissionError("ADMISSION_PUBLICATION_BINDING_MISMATCH") + pointer, receipt = _decode(pub["pointer"]), _decode(pub["receipt"]) + if _hash(pub["pointer"]) != pub["pointer_hash"] or _hash(pub["receipt"]) != pub["receipt_hash"]: + raise DurableAdmissionError("ADMISSION_PUBLICATION_HASH_MISMATCH") + if receipt.get("history") != ["STAGED", "VALIDATED", "COMMITTED_NOT_CURRENT", "POSTVERIFY_PASS", "CURRENT"]: + raise DurableAdmissionError("ADMISSION_RECEIPT_HISTORY_INVALID") + if receipt.get("reason") != "EXACT_POSTVERIFIED_CANDIDATE_PROMOTED_BY_OCC": + raise DurableAdmissionError("ADMISSION_RECEIPT_REASON_INVALID") + previous = pub["previous_publication_id"] + previous_hash = None + if previous is not None: + old = self._con.execute("SELECT pointer_hash FROM person_admission_publications WHERE publication_id=? AND twin_key=?", + (previous, self.twin_key)).fetchone() + if old is None or previous == pub["publication_id"]: + raise DurableAdmissionError("ADMISSION_PREDECESSOR_MISMATCH") + previous_hash = old[0] + if receipt.get("expected_pointer_sha256") != previous_hash: + raise DurableAdmissionError("ADMISSION_RECEIPT_PREDECESSOR_MISMATCH") + _current_admission(self.identity, _decode(row["candidate"]), receipt, pointer, + _Readback(pub["candidate_id"], row["candidate"], pub["pointer"])) + return AdmissionSnapshot(pub["publication_id"], row["candidate"], pub["pointer"], + pub["receipt"], row["anchors"], row["binding"]) + + def load_current(self) -> AdmissionSnapshot | None: + if self._phase is not None or self._suspended: + raise DurableAdmissionError("ADMISSION_ACTIVE_OR_SUSPENDED_READ_DENIED") + try: + if self._file_identity != _direct_file(self.path): + raise DurableAdmissionError("ADMISSION_DB_IDENTITY_CHANGED") + self._con.execute("BEGIN") + _schema(self._con) + result = self._snapshot() + self._con.execute("COMMIT") + return result + except Exception: + if self._con.in_transaction: + self._con.execute("ROLLBACK") + raise DurableAdmissionError("ADMISSION_CURRENT_READ_FAILED") from None + + def read_current_pointer_bytes(self) -> bytes | None: + if self._phase == "PROMOTING" and self._con.in_transaction: + row = self._current_row() + return None if row is None else row["pointer"] + # R3 callback reads and ordinary R4 reads use actual SQLite, not a cached ALLOW. + if self._phase is not None: + result = self._snapshot() + if self._phase == "INITIAL": + self._expected_publication = None if result is None else result.publication_id + return None if result is None else result.pointer_bytes + result = self.load_current() + return None if result is None else result.pointer_bytes + + def _check_expected(self) -> None: + row = self._current_row() + actual = None if row is None else row["publication_id"] + if actual != self._expected_publication: + raise AdmissionStoreConflict("ADMISSION_CURRENT_GENERATION_CONFLICT") + + def commit_candidate_not_current(self, candidate_id: str, candidate_bytes: bytes) -> None: + if self.read_only or self._phase != "CANDIDATE" or self._active is None: + raise DurableAdmissionError("ADMISSION_DIRECT_WRITE_DENIED") + active = self._active + if candidate_id != active["record"]["id"] or candidate_bytes != _canonical(active["record"]): + raise DurableAdmissionError("ADMISSION_CANDIDATE_INPUT_MISMATCH") + assert_current_effect_allowed("person.admission.candidate_write") + _schema(self._con) + self._con.execute("BEGIN IMMEDIATE") + try: + self._check_expected() + self._verify_binding(active["at"]) + existing = self._candidate_row(candidate_id) + anchors, binding = _canonical(active["anchors"]), _canonical(active["binding"]) + expected = (self.twin_key, candidate_id, candidate_bytes, _hash(candidate_bytes), + anchors, _hash(anchors), binding, _hash(binding)) + if existing is None: + self._con.execute("INSERT INTO person_admission_candidates VALUES(?,?,?,?,?,?,?,?)", expected) + elif tuple(existing) != expected: + raise AdmissionStoreConflict("ADMISSION_CANDIDATE_REBIND_DENIED") + assert_current_effect_allowed("person.admission.candidate_commit") + _schema(self._con) + self._candidate_commit_state = "UNKNOWN" + self._con.execute("COMMIT") + self._candidate_commit_state = "CONFIRMED" + self._candidate_committed = True + except Exception: + if self._con.in_transaction: + self._con.execute("ROLLBACK") + raise + + def promote_current_pointer(self, *, expected_pointer_sha256: str | None, pointer_bytes: bytes) -> None: + if self.read_only or self._phase != "PROMOTING" or not self._con.in_transaction or self._next_publication is None: + raise DurableAdmissionError("ADMISSION_DIRECT_PROMOTION_DENIED") + assert_current_effect_allowed("person.admission.pointer_write") + self._check_expected() + row = self._current_row() + actual_hash = None if row is None else _hash(row["pointer"]) + if actual_hash != expected_pointer_sha256: + raise AdmissionStoreConflict("ADMISSION_POINTER_HASH_CONFLICT") + _decode(pointer_bytes) + # Deferred FK makes COMMIT of this pointer without its real R3 receipt fail. + self._con.execute("INSERT INTO person_admission_current VALUES(?,?,?,?) ON CONFLICT(twin_key) DO UPDATE SET publication_id=excluded.publication_id,pointer=excluded.pointer,pointer_hash=excluded.pointer_hash", + (self.twin_key, self._next_publication, pointer_bytes, _hash(pointer_bytes))) + + def _verify_binding(self, at: str) -> None: + active = self._active + if active is None or self._row_binding(active["record"], active["anchors"], + active["binding"]["memory_id"], active["binding"]["namespace"], at=at) != active["binding"]: + raise DurableAdmissionError("ADMISSION_MEMORY_BINDING_CHANGED") + + def _fresh(self, reader: Callable[[], Mapping[str, Any]]) -> dict[str, Any]: + value = _decode(_canonical(dict(reader()))) + if set(value) != {"current_revocation_ledger", "at"} or not isinstance(value["at"], str): + raise DurableAdmissionError("ADMISSION_FRESH_INPUT_INVALID") + self._active["at"] = value["at"] + return value + + def _result(self, status: str, reason: str, snapshot: AdmissionSnapshot | None = None) -> dict[str, Any]: + if self._candidate_commit_state == "UNKNOWN": + status, reason = "HOLD", "CANDIDATE_COMMIT_OUTCOME_UNKNOWN" + return dict(schema=SCHEMA, status=status, reason=reason, + candidate_persisted=(None if self._candidate_commit_state == "UNKNOWN" else self._candidate_committed), + candidate_commit_state=self._candidate_commit_state, + commit_state=self._commit_state, + possible_persisted_effect=self._candidate_commit_state != "NOT_ATTEMPTED" or self._commit_state != "NOT_ATTEMPTED", + publication_id=self._next_publication, + admission_receipt=None if snapshot is None else snapshot.receipt, + **_CEILING) + + def admit(self, *, consent_receipt: Mapping[str, Any], source_record: Mapping[str, Any], + candidate_record: Mapping[str, Any], purpose: str, memory_id: int, namespace: str, + read_current_evidence: Callable[[], Mapping[str, Any]]) -> dict[str, Any]: + """Bounded owner-invoked composition; no fresh authority is loaded from SQL. + + Candidate staging commits separately. CURRENT pointer + the actual final + R3 receipt + exact row binding become visible in one promotion transaction. + Any uncertain commit or failed postcommit readback is HOLD, never retried. + """ + if self.read_only or self._phase is not None or self._suspended: + raise DurableAdmissionError("ADMISSION_WRITE_MODE_OR_REENTRY_DENIED") + self._candidate_committed, self._commit_state = False, "NOT_ATTEMPTED" + self._candidate_commit_state = "NOT_ATTEMPTED" + result = None + self._next_publication = None + try: + assert_current_effect_allowed("person.admission.admit") + if self._file_identity != _direct_file(self.path): + raise DurableAdmissionError("ADMISSION_DB_IDENTITY_CHANGED") + _schema(self._con) + record = _decode(_canonical(dict(candidate_record))) + anchors = _decode(_canonical(dict(identity=self.identity, consent_receipt=dict(consent_receipt), + source_record=dict(source_record), purpose=purpose))) + validate_person_twin_record(record, identity=self.identity, + consent_receipt=anchors["consent_receipt"], source_record=anchors["source_record"]) + binding = self._row_binding(record, anchors, memory_id, namespace) + self._active = dict(record=record, anchors=anchors, binding=binding, at=None) + self._phase = "INITIAL" + coordinator = PersonTwinAdmissionCoordinator(self, session_id=record["admission_session_id"], + identity=self.identity, consent_receipt=anchors["consent_receipt"], + source_record=anchors["source_record"], candidate_record=record, purpose=purpose) + for phase, method, expected in ( + ("VALIDATE", coordinator.validate, "VALIDATED"), + ("CANDIDATE", coordinator.commit_not_current, "COMMITTED_NOT_CURRENT"), + ("POSTVERIFY", coordinator.postverify, "POSTVERIFY_PASS"), + ): + self._phase = phase + evidence = self._fresh(read_current_evidence) + ledger_key = "revocation_ledger" if phase == "VALIDATE" else "current_revocation_ledger" + receipt = method(**{ledger_key: evidence["current_revocation_ledger"]}, at=evidence["at"]) + if receipt["state"] != expected: + result = self._result("HOLD", "R3_" + phase + "_HELD") + return result + assert_current_effect_allowed("person.admission.promotion_begin") + _schema(self._con) + self._con.execute("BEGIN IMMEDIATE") + self._phase = "PROMOTING" + self._check_expected() + self._next_publication = uuid.uuid4().hex + evidence = self._fresh(read_current_evidence) + self._verify_binding(evidence["at"]) + receipt = coordinator.promote_current(**evidence) + if receipt["state"] != "CURRENT": + raise DurableAdmissionError("ADMISSION_R3_PROMOTION_HELD") + row = self._current_row() + receipt_bytes = _canonical(receipt) + self._con.execute("INSERT INTO person_admission_publications VALUES(?,?,?,?,?,?,?,?,?)", + (self._next_publication, self.twin_key, record["id"], row["pointer"], row["pointer_hash"], + receipt_bytes, _hash(receipt_bytes), _hash(_canonical(binding)), self._expected_publication)) + before_commit = self._snapshot() + if before_commit is None or before_commit.receipt_bytes != receipt_bytes: + raise DurableAdmissionError("ADMISSION_TRANSACTION_READBACK_FAILED") + assert_current_effect_allowed("person.admission.promotion_commit") + _schema(self._con) + self._commit_state = "UNKNOWN" + self._con.execute("COMMIT") + self._commit_state = "CONFIRMED" + self._phase = None + with PersonTwinAdmissionStore(self.path, identity=self.identity, timeout=self.timeout) as fresh: + if fresh._file_identity != self._file_identity: + raise DurableAdmissionError("ADMISSION_POSTCOMMIT_FILE_CHANGED") + verified = fresh.load_current() + if verified != before_commit: + raise DurableAdmissionError("ADMISSION_POSTCOMMIT_SNAPSHOT_CHANGED") + result = self._result("PASS", "CURRENT_COMMITTED_AND_FRESH_READBACK_VERIFIED", verified) + return result + except Exception: + reason = ("COMMIT_OUTCOME_UNKNOWN" if self._commit_state == "UNKNOWN" else + "POSTCOMMIT_READBACK_FAILED" if self._commit_state == "CONFIRMED" else + "ADMISSION_FAILED_CLOSED") + result = self._result("HOLD", reason) + return result + finally: + try: + if self._con.in_transaction: + self._con.execute("ROLLBACK") + except Exception: + self._suspended = True + if result is not None: + result.update(status="HOLD", reason="ROLLBACK_UNVERIFIED", + possible_persisted_effect=True, admission_receipt=None) + try: + self._con.close() + except Exception: + pass + finally: + self._phase, self._active = None, None diff --git a/continuityos/person_twin_demo.py b/continuityos/person_twin_demo.py new file mode 100644 index 00000000..4f0a6e0a --- /dev/null +++ b/continuityos/person_twin_demo.py @@ -0,0 +1,919 @@ +"""Synthetic Person/R21H continuity through three finite, isolated children. + +The caller owns PREPARE, A and B directly. Only PREPARE writes the fixture; +both model workers recover the real R3 receipt from that same SQLite file and +construct fresh synthetic host bindings. No admission sidecar or reader-side +promotion is used. This is a product proof, not a vault or owner grant. +""" +from __future__ import annotations + +import argparse +import copy +from dataclasses import replace +from datetime import datetime, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import re +import secrets +import shutil +import stat +import subprocess +import sys +import tempfile +import threading +import time +from typing import Any, Mapping, Sequence + +from .current_effect_boundary import assert_current_effect_allowed + +SCHEMA = "continuityos.product_demo_person_continuity/v2" +FIXTURE_SCHEMA = "continuityos.product_demo_person_continuity_snapshot/v2" +WORKER_SCHEMA = "continuityos.product_demo_person_continuity_worker/v2" +PREPARE_SCHEMA = "continuityos.product_demo_person_continuity_prepare/v2" +WORK_ORDER = "COS_PERSON_CONTINUITY_DURABLE_DEMO_R2" +PASS = "COS_DEMO_PERSON_CONTINUITY_PASS" +HOLD = "COS_DEMO_PERSON_CONTINUITY_HOLD" +NAMESPACE = "synthetic-person-context" +QUERY = "Continue the current synthetic project context." +AT = "2026-08-25T12:50:00Z" # Declared fixture clock, not trusted current time. +PURPOSE = "person-memory" +ROOT_PREFIX = "continuityos-person-demo-" +MAX_DB_BYTES = 2097152 +MAX_OUTPUT_BYTES = 65536 +OLD_TEXT = "OLD_DECISION_SENTINEL: use the retired prototype." +FOREIGN_TEXT = "FOREIGN_PRIVATE_SENTINEL: another person's context." +_socket_events: list[str] = [] +_socket_guard_installed = False +WORKER_CHECKS = frozenset({"current_evidence_delivered", "old_and_foreign_excluded", "exact_query", + "query_only", "fresh_prompt_revalidation", "one_generation", "transport", "execute_denied", + "no_socket_events", "fixture_and_db_unchanged", "durable_receipt_recovered", "no_admission_sidecar"}) +PERSON_RECEIPT_KEYS = frozenset({"schema", "decision", "action", "evidence_count", "memory_refs", + "prompt_boundary_revalidated", "production_admission_status", "execution_authority", "can_execute", + "can_trade", "capital_permission", "intent_sha256", "r4_receipt_hash", "evidence_sha256", "receipt_hash"}) + + +class PersonDemoHold(ValueError): + """Only fixed public reason codes are raised by fixture helpers.""" + + +def _canonical(value: Any) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(",", ":"), allow_nan=False).encode("utf-8") + + +def _sha(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _hash(value: Any) -> str: + return _sha(_canonical(value)) + + +def _is_sha(value: Any) -> bool: + return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None + + +def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise PersonDemoHold("DUPLICATE_JSON_KEY") + value[key] = item + return value + + +def _decode(data: bytes) -> dict[str, Any]: + try: + value = json.loads(data.decode("utf-8"), object_pairs_hook=_pairs, + parse_constant=lambda _: (_ for _ in ()).throw(ValueError())) + if not isinstance(value, dict): + raise ValueError() + # JSON's finite syntax can still overflow a float (for example 1e999). + _canonical(value) + return value + except Exception: + raise PersonDemoHold("INVALID_JSON_OBJECT") from None + + +def _seal(value: dict[str, Any]) -> dict[str, Any]: + return {**value, "receipt_sha256": _hash(value)} + + +def _sealed(value: Any) -> bool: + try: + return (isinstance(value, dict) and _is_sha(value.get("receipt_sha256")) + and _hash({k: v for k, v in value.items() if k != "receipt_sha256"}) + == value["receipt_sha256"]) + except Exception: + return False + + +def _governance() -> dict[str, Any]: + return {"production_admission_status": "NOT_PRODUCTION_ADMITTED", + "execution_authority": "NONE", "can_execute": False, + "can_trade": False, "capital_permission": "DENY"} + + +def _effects() -> dict[str, Any]: + return {"ephemeral_filesystem_write": False, "ephemeral_memory_write": False, + "user_memory_read": False, "user_memory_write": False, + "client_config_write": False, "network_effect": False, + "external_model_call": False, "subprocess_execution": False, + "child_launches": 0, "server_started": False, "deployment": False, + "agent_dispatch": False, "trading": False, "wallet_access": False, + "temporary_cleanup_pass": None} + + +def _hold(reason: str, **details: Any) -> dict[str, Any]: + return {"schema": SCHEMA, "work_order": WORK_ORDER, "terminal": HOLD, + "reason": reason, "mode": "SYNTHETIC_ONLY", "effects": _effects(), + **_governance(), **details} + + +def _guard() -> None: + try: + assert_current_effect_allowed("person.demo") + except Exception: + raise PersonDemoHold("CURRENT_EFFECT_HOLD") from None + + +def _timeout(value: Any) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise PersonDemoHold("INVALID_TIMEOUT") + result = float(value) + if not math.isfinite(result) or not 0 < result <= 20.0: + raise PersonDemoHold("INVALID_TIMEOUT") + return result + + +def _reject_link(info: os.stat_result) -> None: + if stat.S_ISLNK(info.st_mode) or getattr(info, "st_file_attributes", 0) & 0x400: + raise PersonDemoHold("FIXTURE_LINK_FORBIDDEN") + + +def _owned_root(root: Path) -> Path: + if ".." in Path(root).parts: + raise PersonDemoHold("FIXTURE_ROOT_INVALID") + root = Path(root).absolute() + base = Path(tempfile.gettempdir()).resolve(strict=True) + try: + if not root.name.startswith(ROOT_PREFIX) or not root.is_relative_to(base): + raise PersonDemoHold("FIXTURE_ROOT_INVALID") + for node in (root, *root.parents): + info = node.lstat() + _reject_link(info) + if not stat.S_ISDIR(info.st_mode): + raise PersonDemoHold("FIXTURE_ROOT_INVALID") + if node == base: + break + info = root.lstat() + if hasattr(os, "getuid") and (info.st_uid != os.getuid() or info.st_mode & 0o022): + raise PersonDemoHold("FIXTURE_ROOT_NOT_OWNED") + return root + except PersonDemoHold: + raise + except Exception: + raise PersonDemoHold("FIXTURE_ROOT_INVALID") from None + + +def _read_bytes(root: Path, name: str, maximum: int) -> bytes: + root = _owned_root(root) + if name != "memory.db": + raise PersonDemoHold("FIXTURE_MEMBER_INVALID") + path = root / name + try: + before = path.lstat() + _reject_link(before) + if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or not 0 < before.st_size <= maximum: + raise PersonDemoHold("FIXTURE_MEMBER_INVALID") + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0) + fd = os.open(path, flags) + with os.fdopen(fd, "rb") as handle: + opened = os.fstat(handle.fileno()) + if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): + raise PersonDemoHold("FIXTURE_MEMBER_CHANGED") + data = handle.read(maximum + 1) + after = path.lstat() + _reject_link(after) + if len(data) != before.st_size or len(data) > maximum or ( + after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns + ) != (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns): + raise PersonDemoHold("FIXTURE_MEMBER_CHANGED") + _owned_root(root) + return data + except PersonDemoHold: + raise + except Exception: + raise PersonDemoHold("FIXTURE_MEMBER_UNAVAILABLE") from None + + +def _members(root: Path) -> None: + root = _owned_root(root) + if {item.name for item in root.iterdir()} != {"memory.db"}: + raise PersonDemoHold("FIXTURE_MEMBERS_CHANGED") + for item in root.iterdir(): + _reject_link(item.lstat()) + + +def _fixture_identity(run_id: str) -> dict[str, Any]: + """Controller-pinned synthetic identity; never selected from stored contents.""" + if type(run_id) is not str or re.fullmatch(r"[0-9a-f]{32}", run_id) is None: + raise PersonDemoHold("FIXTURE_RUN_ID_INVALID") + from .person_twin_admission_contracts import build_person_twin_identity + return build_person_twin_identity( + tenant_id="tenant:synthetic-person-demo", subject_id="subject:" + run_id, + owner_id="human:demo-owner", controller_id="human:demo-owner", + controller_kind="HUMAN", ownership_epoch=1, created_at="2026-08-25T00:00:00Z", + ) + +def _write_fixture(root: Path, run_id: str) -> dict[str, Any]: + _guard() + root = _owned_root(root) + if re.fullmatch(r"[0-9a-f]{32}", run_id) is None or any(root.iterdir()): + raise PersonDemoHold("FIXTURE_PREPARATION_INVALID") + from .company_twin_ingest import normalize_envelope + from .embed import HashingEmbedder + from .memory import Memory + from .person_twin_admission_contracts import ( + build_consent_revocation_ledger, build_source_consent_receipt, + ) + from .person_twin_admission_store import PersonTwinAdmissionStore + from .person_twin_privacy_provenance import ( + build_person_twin_provenance, canonical_person_private_scope, deterministic_source_identity_hash, + ) + identity = _fixture_identity(run_id) + scope = canonical_person_private_scope(identity["twin_id"]) + text = ("Goal: continue the approved synthetic project.\n" + "Decision: keep the read-only Person/R21H evidence path.\n" + "Constraint: synthetic data only; no external effects.\n" + "Next step: inspect the recovered current context in session B.\n" + "Run: " + run_id) + source = normalize_envelope({ + "schema_version": "company-twin-source-envelope/1", "tenant_id": identity["tenant_id"], + "connector_id": "connector:synthetic-fixture", "source_system": "synthetic_fixture", + "source_object_type": "document", "source_object_id": "document:" + run_id, + "revision_id": "current", "observed_at": "2026-08-25T10:01:00Z", + "effective_at": "2026-08-25T10:00:00Z", "acl": {"visibility": "PERSONAL", "scope": scope}, + "payload": {"title": "Synthetic current context", "text": text}, + "raw_ref": "synthetic://person-demo/" + run_id, "cursor": "cursor:" + run_id, + "actor": {"actor_id": "human:demo-owner", "actor_kind": "HUMAN", "authority_class": "OWNER"}, + "deleted": False, + }) + consent = build_source_consent_receipt( + identity, authorizing_principal="human:demo-owner", authorizing_principal_kind="HUMAN", + source_system=source["source_system"], source_identity_hash=deterministic_source_identity_hash(source), + allowed_object_types=["document"], allowed_scopes=[scope], purpose=PURPOSE, + issued_at="2026-08-25T09:00:00Z", expires_at="2026-08-26T00:00:00Z", revoked_at=None, + source_read_authority=False, memory_admission_authority=True, + policy_version="synthetic-person-demo/1", + ) + ledger = build_consent_revocation_ledger() + record = build_person_twin_provenance( + identity, consent, ledger, source, admission_session_id="admission:" + run_id, + purpose=PURPOSE, at="2026-08-25T11:00:00Z", + ) + writer = Memory(str(root / "memory.db"), embedder=HashingEmbedder()) + try: + old_id = writer.remember(OLD_TEXT, namespace=NAMESPACE) + writer.remember(FOREIGN_TEXT, namespace="foreign-person") + memory_id = writer.supersede(old_id, text, namespace=NAMESPACE) + wal = writer.store.con.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + if wal is None or int(wal[0]) != 0: + raise PersonDemoHold("FIXTURE_WAL_NOT_QUIESCENT") + mode = writer.store.con.execute("PRAGMA journal_mode=DELETE").fetchone() + if mode is None or str(mode[0]).lower() != "delete": + raise PersonDemoHold("FIXTURE_JOURNAL_NOT_QUIESCENT") + finally: + writer.store.con.close() + # The Memory writer is closed. The real store adds its tables to this same + # owned DB and persists the unchanged coordinator's actual final receipt. + PersonTwinAdmissionStore.initialize(root / "memory.db") + with PersonTwinAdmissionStore(root / "memory.db", identity=identity, read_only=False) as store: + result = store.admit(consent_receipt=consent, source_record=source, + candidate_record=record, purpose=PURPOSE, memory_id=memory_id, namespace=NAMESPACE, + read_current_evidence=lambda: {"current_revocation_ledger": build_consent_revocation_ledger(), "at": AT}) + snapshot = store.load_current() + if (result.get("status") != "PASS" or result.get("commit_state") != "CONFIRMED" + or snapshot is None or snapshot.receipt != result.get("admission_receipt")): + raise PersonDemoHold("FIXTURE_DURABLE_ADMISSION_HOLD") + _members(root) + db = _read_bytes(root, "memory.db", MAX_DB_BYTES) + _read_fixture(root, _sha(db), run_id) + # Only selectors and digests cross PREPARE stdout, never the R3 receipt/body. + return {"db_sha256": _sha(db), "memory_id": memory_id, "run_id": run_id, + "publication_id": snapshot.publication_id, "stored_receipt_sha256": _sha(snapshot.receipt_bytes), + "admission_source": "MEMORY_SQLITE", "admission_sidecar_count": 0} + + +def _read_fixture(root: Path, expected_sha256: str, run_id: str) -> dict[str, Any]: + """Reopen only the pinned DB; decoded data stays in this process, not a sidecar.""" + _guard() + identity = _fixture_identity(run_id) + if not _is_sha(expected_sha256): + raise PersonDemoHold("FIXTURE_HASH_INVALID") + _members(root) + db = _read_bytes(root, "memory.db", MAX_DB_BYTES) + if _sha(db) != expected_sha256: + raise PersonDemoHold("FIXTURE_DB_HASH_MISMATCH") + from .person_twin_admission_store import PersonTwinAdmissionStore + from .person_twin_admission_contracts import build_consent_revocation_ledger + try: + with PersonTwinAdmissionStore(root / "memory.db", identity=identity) as store: + snapshot = store.load_current() + if snapshot is None: + raise ValueError() + anchors, binding = snapshot.anchors, snapshot.binding + if (anchors["identity"] != identity or anchors["purpose"] != PURPOSE + or type(binding["memory_id"]) is not int or binding["memory_id"] < 1 + or binding["namespace"] != NAMESPACE): + raise ValueError() + text = anchors["source_record"]["payload"]["text"] + if not isinstance(text, str) or not 0 < len(text) <= 8192 or run_id not in text: + raise ValueError() + # The ledger is a new synthetic host input, not persisted authorization. + context = {"identity": identity, "source_record": anchors["source_record"], + "consent_receipt": anchors["consent_receipt"], + "current_revocation_ledger": build_consent_revocation_ledger(), + "person_record": snapshot.candidate, "admission_receipt": snapshot.receipt, + "current_pointer": snapshot.pointer} + fixture = {"schema": FIXTURE_SCHEMA, "run_id": run_id, + "memory": {"id": binding["memory_id"], "namespace": NAMESPACE, + "sha256": expected_sha256, "size_bytes": len(db)}, + "context": context, "publication_id": snapshot.publication_id, + "admission_bytes": {"candidate_utf8": snapshot.candidate_bytes.decode("utf-8"), + "pointer_utf8": snapshot.pointer_bytes.decode("utf-8"), + "candidate_sha256": _sha(snapshot.candidate_bytes), + "pointer_sha256": _sha(snapshot.pointer_bytes), + "receipt_sha256": _sha(snapshot.receipt_bytes)}} + except Exception: + raise PersonDemoHold("DURABLE_ADMISSION_READ_HOLD") from None + _members(root) + if _read_bytes(root, "memory.db", MAX_DB_BYTES) != db: + raise PersonDemoHold("FIXTURE_DB_CHANGED_DURING_READ") + return fixture + +class _EmptyReplayFixture: + def read_replay_receipt_bytes(self, replay_key): + return None + + +def _fresh_host_context(fixture: Mapping[str, Any], role: str) -> dict[str, Any]: + """New local fixture host bindings; no serialized session or inherited ALLOW.""" + if role not in {"A", "B"}: + raise PersonDemoHold("WORKER_ROLE_INVALID") + nonce = secrets.token_hex(16) + identity = fixture["context"]["identity"] + scope = fixture["context"]["person_record"]["scope"] + owner = identity["owner_id"] + session_effects = {name: False for name in ( + "legacy_fallback", "memory_write", "ledger_write", "filesystem_write", "network_effect", + "server_started", "subprocess_execution", "deployment", "current_state_apply", "canonical_mutation", + "external_message", "auto_dispatch", "trading", "wallet_access", "order_execution", "can_trade")} + session_effects.update(capital_permission="DENY", deploy_permission="DENY") + session = {"schema": "continuityos.current_effect_boundary/v1", "mode": "CURRENT", "declared": True, + "binding_verified": True, "reason": "EXACT_CURRENT_SESSION_VERIFIED", "authority_generation": "R64", + "challenge_id": "synthetic-challenge:" + nonce, "challenge_sha256": _hash([nonce, "challenge"]), + "ack_sha256": _hash([nonce, "ack"]), "session_effect_ceiling": "READ_ONLY", + "authority_ceiling": "NO_FURTHER_AGENT_WORK", "effects": session_effects} + base = {"projection_sha256": _hash([nonce, "projection"]), "event_cursor": 1, + "event_chain_head": _hash([nonce, "chain"])} + core = {"proposal_id": "synthetic-proposal:" + nonce, "proposal_file_sha256": _hash([nonce, "proposal"]), + "project_id": "synthetic-person-demo", "base": base, "operation_count": 1} + request = {"schema": "continuityos.operational_memory.apply_authorization_request/v1", + "terminal": "CURRENT_MEMORY_APPLY_AUTH_REQUEST_PASS", "authorization_granted": False, + "execution_authorized": False, "apply_status": "NOT_APPLIED", "request_id": "omar-" + _hash(core)[:40], + "proposal_id": core["proposal_id"], "proposal_file_sha256": core["proposal_file_sha256"], + "project_id": core["project_id"], "expected_base": base, "operation_count": 1} + packet_core = {key: request[key] for key in ("request_id", "project_id", "proposal_id", + "proposal_file_sha256", "expected_base", "operation_count")} + packet_core.update(authorization_file_sha256=_hash([nonce, "authorization"]), + authorization={"class": "HUMAN", "id": owner, "ref": "synthetic://approval/" + nonce}) + preflight = {"schema": "continuityos.operational_memory.project_update_packet_preflight/v1", + "terminal": "CURRENT_PROJECT_UPDATE_PREFLIGHT_READY", "packet_valid": True, + "authorization_record_valid": True, "apply_ready": True, "execution_authorized": False, + "packet_id": "r4p-" + _hash(packet_core)[:40], + **{k: v for k, v in packet_core.items() if k != "request_id"}} + policy = {"schema_version": "company-twin-p2c/1", "tenant_id": identity["tenant_id"], + "actors": [{"id": "actor:owner", "principal_id": owner, "actor_kind": "HUMAN", + "role": "DIRECTOR", "scopes": [scope]}, + {"id": "actor:agent", "principal_id": "agent:demo", "actor_kind": "AGENT", + "role": "AGENT", "scopes": [scope], "manager_actor_id": "actor:owner"}], + "grants": [{"id": "grant:owner", "actor_id": "actor:owner", "actions": ["READ", "DELEGATE"], + "scopes": [scope], "purposes": [PURPOSE]}], + "delegations": [{"id": "delegation:demo", "grantor_actor_id": "actor:owner", + "grantee_actor_id": "actor:agent", "actions": ["READ"], "scopes": [scope], + "purposes": [PURPOSE], "parent_id": None, "expires_at": "2026-08-26T00:00:00Z", "revoked": False}], + "explicit_denies": [], "revoked_delegation_ids": []} + return {"current_session": session, "authorization_request": request, "authorization_preflight": preflight, + "policy": policy, "principal_id": "agent:demo", "action": "READ", "requested_scope": scope, + "requested_privacy_class": fixture["context"]["person_record"]["privacy_class"], + "purpose": PURPOSE, "at": AT} + + +class _FixtureProvider: + """Synthetic host context plus fresh reads of the real SQLite admission store.""" + def __init__(self, root: Path, expected_sha256: str, host: Mapping[str, Any], *, + run_id: str, admission_store): + self.root, self.expected_sha256, self.run_id = root, expected_sha256, run_id + self.host = copy.deepcopy(dict(host)) + self.admission_store = admission_store + self.reads = 0 + + def read_context(self): + self.reads += 1 + fixture = _read_fixture(self.root, self.expected_sha256, self.run_id) + return {**copy.deepcopy(fixture["context"]), **copy.deepcopy(self.host), + "admission_store": self.admission_store, "replay_store": _EmptyReplayFixture()} + +class _RecordingModel: + """Transport fixture only. Its zero timing values are not measured latency.""" + def __init__(self, profile, *, cold: bool): + self.profile, self.loaded = profile, not cold + self.instance_id = "synthetic-instance:" + profile.model + self.calls: list[dict[str, Any]] = [] + self.events: list[str] = [] + + def models(self): + self.events.append("models") + instances = [{"id": self.instance_id, "config": { + "context_length": self.profile.context_length, "parallel": 1}}] if self.loaded else [] + return [{"key": self.profile.model, "loaded_instances": instances}] + + def _capture(self, kwargs, event): + from .sovereign_twin_runtime import LocalChatResult + self.events.append(event) + if self.calls: + raise PersonDemoHold("MULTIPLE_MODEL_CALLS") + self.calls.append(dict(kwargs)) + return LocalChatResult(text="Synthetic transport captured the context.", + model_instance_id=self.instance_id, stats={}, reasoning=None) + + def chat(self, **kwargs): + return self._capture(kwargs, "chat") + + def chat_fast_streaming_jit_reconciled(self, **kwargs): + result = self._capture(kwargs, "stream") + self.loaded = True + kwargs["on_model_load_end"](self.instance_id, 0.0) + return result, {"model_instance_id": self.instance_id, "model_load_time_seconds": 0.0, + "model_load_end_seen": True, "acquisition_signal": "model_load.end", + "acquisition_event_type": "model_load.end", + "event_types": ("chat.start", "model_load.end", "message.delta", "chat.end")} + + def embed(self, *args, **kwargs): + self.events.append("embed") + raise PersonDemoHold("UNEXPECTED_EMBEDDING") + + +def _audit_state() -> dict[str, Any]: + return {"installed": _socket_guard_installed, "events": len(_socket_events), "policy": "DENY_SOCKET_EVENTS"} + + +def _source_sha256() -> str: + return _sha(Path(__file__).read_bytes()) + + +def _context_fields(text: str) -> dict[str, str]: + labels = {"Goal": "goal", "Decision": "decision", "Constraint": "constraint", "Next step": "next_action"} + result = {} + for line in text.splitlines(): + label, sep, value = line.partition(": ") + if sep and label in labels: + if labels[label] in result: + raise PersonDemoHold("RECOVERED_CONTEXT_INVALID") + result[labels[label]] = value + if set(result) != set(labels.values()) or not all(result.values()): + raise PersonDemoHold("RECOVERED_CONTEXT_INVALID") + return result + + +def _worker(root: Path, expected_sha256: str, role: str, run_id: str = "") -> tuple[dict[str, Any], int]: + client = runtime = admission_store = None + started = time.monotonic() + try: + _guard() + fixture = _read_fixture(root, expected_sha256, run_id) + host = _fresh_host_context(fixture, role) + from .person_twin_authority_replay import evaluate_person_twin_authority_replay + from .person_twin_r21h_evidence import PersonReadIntent, PersonTwinR21HEvidenceAdapter + from .sovereign_twin_runtime import DEFAULT_PROFILES, SovereignTwinRuntime + context = fixture["context"] + intent = PersonReadIntent( + query_sha256=_sha(QUERY.encode("utf-8")), current_session_sha256=_hash(host["current_session"]), + request_id=host["authorization_request"]["request_id"], packet_id=host["authorization_preflight"]["packet_id"], + memory_id=fixture["memory"]["id"], namespace=NAMESPACE, + person_record_id=context["person_record"]["id"], source_record_id=context["source_record"]["id"], + principal_id=host["principal_id"], scope=host["requested_scope"], + privacy_class=host["requested_privacy_class"], purpose=PURPOSE, + ) + from .person_twin_admission_store import PersonTwinAdmissionStore + admission_store = PersonTwinAdmissionStore(root / "memory.db", identity=_fixture_identity(run_id)) + provider = _FixtureProvider(root, expected_sha256, host, run_id=run_id, admission_store=admission_store) + profile = replace(DEFAULT_PROFILES["fast"], model="synthetic-person-" + role) + client = _RecordingModel(profile, cold=role == "B") + _read_fixture(root, expected_sha256, run_id) # No-follow/hash checks directly before Memory opens. + runtime = SovereignTwinRuntime(str(root / "memory.db"), client=client, + profiles={**DEFAULT_PROFILES, "fast": profile}, + person_evidence_adapter=PersonTwinR21HEvidenceAdapter(intent, provider)) + query_only = runtime.memory.store.con.execute("PRAGMA query_only").fetchone()[0] == 1 + answer = runtime.ask(QUERY) + denied = evaluate_person_twin_authority_replay(**{**provider.read_context(), "action": "EXECUTE"}) + if len(client.calls) != 1: + raise PersonDemoHold("MODEL_CALL_COUNT_MISMATCH") + prompt = client.calls[0]["system_prompt"] + rows = json.loads(prompt.split("MEMORY_EVIDENCE_JSON:\n", 1)[1]) + delivered = [{k: row[k] for k in ("ref", "namespace", "text")} for row in rows] + expected = [{"ref": "mem:" + str(fixture["memory"]["id"]), "namespace": NAMESPACE, + "text": context["source_record"]["payload"]["text"]}] + person = dict(answer.stats["person_evidence"]) + checks = {"current_evidence_delivered": delivered == expected, + "old_and_foreign_excluded": OLD_TEXT not in prompt and FOREIGN_TEXT not in prompt, + "exact_query": client.calls[0]["input_text"] == QUERY, "query_only": query_only and admission_store._con.execute("PRAGMA query_only").fetchone()[0] == 1, + "fresh_prompt_revalidation": provider.reads >= 2 and person.get("prompt_boundary_revalidated") is True, + "one_generation": len(client.calls) == 1, + "transport": (client.events.count("chat"), client.events.count("stream")) == ((1, 0) if role == "A" else (0, 1)), + "execute_denied": denied.get("decision") == "DENY" and denied.get("execution_authorized") is False, + "no_socket_events": not _socket_events, + "durable_receipt_recovered": _sha(admission_store.load_current().receipt_bytes) == fixture["admission_bytes"]["receipt_sha256"], + "no_admission_sidecar": {item.name for item in root.iterdir()} == {"memory.db"}} + runtime.close() + runtime = None + admission_store.close() + admission_store = None + after = _read_fixture(root, expected_sha256, run_id) + checks["fixture_and_db_unchanged"] = after == fixture + if not all(checks.values()): + raise PersonDemoHold("WORKER_PROOF_MISMATCH") + value = {"schema": WORKER_SCHEMA, "terminal": "PERSON_DEMO_WORKER_PASS", "role": role, + "process_id": os.getpid(), "model_profile": profile.model, + "model_calls": len(client.calls), "model_events": client.events, + "transport": "warm_chat" if role == "A" else "cold_streaming", + "captured_evidence": delivered, "prompt_sha256": _sha(prompt.encode("utf-8")), + "recovered": _context_fields(delivered[0]["text"]), "session_sha256": _hash(host["current_session"]), + "request_sha256": _hash(host["authorization_request"]), + "packet_sha256": _hash(host["authorization_preflight"]), + "person_evidence": person, "context_reads": provider.reads, "checks": checks, + "db_sha256": expected_sha256, "publication_id": fixture["publication_id"], + "stored_receipt_sha256": fixture["admission_bytes"]["receipt_sha256"], + "admission_source": "MEMORY_SQLITE", "admission_sidecar_count": 0, + "unsupported_execute": {"decision": denied["decision"], "reason": denied.get("reason"), + "execution_authorized": denied.get("execution_authorized")}, + "socket_audit": _audit_state(), "synthetic_host_bindings": True, "fixture_clock_at": AT, + "source_sha256": _source_sha256(), + "duration_seconds": time.monotonic() - started, **_governance()} + return _seal(value), 0 + except Exception as exc: + reason = str(exc) if type(exc) is PersonDemoHold else "WORKER_VALIDATION_HOLD" + return _seal({"schema": WORKER_SCHEMA, "terminal": "PERSON_DEMO_WORKER_HOLD", "reason": reason, + "role": role, "process_id": os.getpid(), "model_calls": len(client.calls) if client else 0, + "model_events": list(client.events) if client else [], "socket_audit": _audit_state(), + **_governance()}), 2 + finally: + if runtime is not None: + try: + runtime.close() + except Exception: + pass # The failed normal close already produces a HOLD above. + if admission_store is not None: + try: + admission_store.close() + except Exception: + pass # Any failed normal close above is already a HOLD. + + +def _run_process(command: list[str], timeout: float) -> dict[str, Any]: + """Bound both pipes during execution, terminate and reap this owned child.""" + start = time.monotonic() + buffers = {"stdout": bytearray(), "stderr": bytearray()} + overflow = threading.Event() + child = None + readers: list[threading.Thread] = [] + result: dict[str, Any] = {"started": False, "returncode": None, "timed_out": False, + "output_limit_exceeded": False, "kill_attempted": False, "direct_child_reaped": False} + try: + child = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + result.update(started=True, process_id=child.pid) + def drain(name, pipe): + try: + while True: + chunk = pipe.read(4096) + if not chunk: + break + remaining = MAX_OUTPUT_BYTES - len(buffers[name]) + buffers[name].extend(chunk[:remaining]) + if len(chunk) > remaining: + overflow.set() + break + except Exception: + overflow.set() + finally: + pipe.close() + for name, pipe in (("stdout", child.stdout), ("stderr", child.stderr)): + thread = threading.Thread(target=drain, args=(name, pipe), daemon=True) + readers.append(thread) + thread.start() + while child.poll() is None and not overflow.is_set(): + remaining = timeout - (time.monotonic() - start) + if remaining <= 0: + result["timed_out"] = True + break + try: + child.wait(timeout=min(remaining, 0.05)) + except subprocess.TimeoutExpired: + pass + if child.poll() is None: + result["kill_attempted"] = True + child.kill() + child.wait(timeout=2.0) + result["direct_child_reaped"] = True + except Exception: + result["reason"] = "CHILD_PROCESS_HOLD" + finally: + if child is not None: + if child.poll() is None: + try: + result["kill_attempted"] = True + child.kill() + child.wait(timeout=2.0) + except Exception: + pass + result["returncode"] = child.poll() + result["direct_child_reaped"] = result["returncode"] is not None + for reader in readers: + reader.join(timeout=1.0) + if any(reader.is_alive() for reader in readers): + overflow.set() + result.update(output_limit_exceeded=overflow.is_set(), stdout=bytes(buffers["stdout"]), + stderr=bytes(buffers["stderr"]), duration_seconds=time.monotonic() - start) + return result + + +def _child_result(command: list[str], timeout: float) -> dict[str, Any]: + result = _run_process(command, timeout) + stdout = result.pop("stdout", b"") + stderr = result.pop("stderr", b"") + result.update(stdout_sha256=_sha(stdout), stderr_sha256=_sha(stderr), command_sha256=_hash(command)) + receipt = None + try: + receipt = _decode(stdout) + except PersonDemoHold: + pass + result["receipt"] = receipt + result["ok"] = (result.get("returncode") == 0 and result.get("direct_child_reaped") is True + and not result.get("timed_out") and not result.get("output_limit_exceeded") + and _sealed(receipt)) + return result + + +def _run_prepare(root: Path, run_id: str, timeout: float) -> dict[str, Any]: + return _child_result(_child_command("_prepare", "--root", str(root), "--run-id", run_id), timeout) + + +def _run_worker(root: Path, expected_sha256: str, role: str, timeout: float, run_id: str) -> dict[str, Any]: + return _child_result(_child_command("_worker", "--root", str(root), + "--db-sha256", expected_sha256, "--role", role, "--run-id", run_id), timeout) + +def _child_command(*arguments: str) -> list[str]: + # -I removes CWD/PYTHONPATH ambiguity; -B avoids ignored environment settings + # producing bytecode. Current-session environment bindings remain intact. + bootstrap = ( + "import hashlib,pathlib,runpy,sys; " + "root=sys.argv.pop(1); expected=sys.argv.pop(1); " + "source=pathlib.Path(root)/'continuityos'/'person_twin_demo.py'; " + "actual=hashlib.sha256(source.read_bytes()).hexdigest(); " + "sys.exit(2) if actual != expected else None; " + "sys.path.insert(0,root); " + "runpy.run_module('continuityos.person_twin_demo',run_name='__main__')" + ) + # Windows venv's python.exe may be a redirector: the visible Popen.pid is + # then NOT the worker PID. Use the verified base interpreter directly. + # The isolated bootstrap below still pins exact package-source SHA-256. + interpreter = sys.executable + if sys.platform == "win32" and sys.prefix != sys.base_prefix: + interpreter = getattr(sys, "_base_executable", "") + if not isinstance(interpreter, str) or not Path(interpreter).is_file(): + raise PersonDemoHold("PYTHON_BASE_EXECUTABLE_UNAVAILABLE") + return [interpreter, "-I", "-B", "-c", bootstrap, + str(Path(__file__).resolve().parent.parent), _source_sha256(), *arguments] + + +def _valid_preparation(result: Mapping[str, Any], run_id: str) -> bool: + try: + value = result["receipt"] + keys = {"schema", "terminal", "process_id", "socket_audit", "source_sha256", + "db_sha256", "memory_id", "run_id", "publication_id", "stored_receipt_sha256", + "admission_source", "admission_sidecar_count", "receipt_sha256", *_governance()} + return (result.get("ok") is True and result.get("direct_child_reaped") is True + and _sealed(value) and set(value) == keys + and value["schema"] == PREPARE_SCHEMA and value["terminal"] == "PERSON_DEMO_PREPARE_PASS" + and value["run_id"] == run_id and type(value["process_id"]) is int + and value["process_id"] == result["process_id"] + and type(value["memory_id"]) is int and value["memory_id"] > 0 + and value["source_sha256"] == _source_sha256() + and value["socket_audit"] == {"installed": True, "events": 0, "policy": "DENY_SOCKET_EVENTS"} + and _is_sha(value["db_sha256"]) and _is_sha(value["stored_receipt_sha256"]) + and isinstance(value["publication_id"], str) and re.fullmatch(r"[0-9a-f]{32}", value["publication_id"]) is not None + and value["admission_source"] == "MEMORY_SQLITE" and type(value["admission_sidecar_count"]) is int + and value["admission_sidecar_count"] == 0 + and all(type(value.get(k)) is type(v) and value[k] == v for k, v in _governance().items())) + except (KeyError, TypeError, ValueError): + return False + + +def _valid_worker(result: Mapping[str, Any], role: str, fixture: Mapping[str, Any], db_sha: str) -> bool: + try: + receipt = result["receipt"] + person = receipt["person_evidence"] + expected_evidence = [{"ref": "mem:" + str(fixture["memory"]["id"]), + "namespace": NAMESPACE, "text": fixture["context"]["source_record"]["payload"]["text"]}] + events = receipt["model_events"] + return (result.get("ok") is True and result.get("direct_child_reaped") is True and _sealed(receipt) + and receipt["schema"] == WORKER_SCHEMA and receipt["terminal"] == "PERSON_DEMO_WORKER_PASS" + and receipt["role"] == role and receipt["model_profile"] == "synthetic-person-" + role + and type(receipt["model_calls"]) is int and receipt["model_calls"] == 1 and type(receipt["process_id"]) is int + and receipt["process_id"] == result["process_id"] + and receipt["source_sha256"] == _source_sha256() + and receipt["db_sha256"] == db_sha == fixture["memory"]["sha256"] + and receipt["publication_id"] == fixture["publication_id"] + and receipt["stored_receipt_sha256"] == fixture["admission_bytes"]["receipt_sha256"] + and receipt["admission_source"] == "MEMORY_SQLITE" + and type(receipt["admission_sidecar_count"]) is int and receipt["admission_sidecar_count"] == 0 + and receipt["captured_evidence"] == expected_evidence + and receipt["recovered"] == _context_fields(expected_evidence[0]["text"]) + and receipt["transport"] == ("warm_chat" if role == "A" else "cold_streaming") + and isinstance(events, list) and set(events) <= {"models", "chat", "stream"} + and (events.count("chat"), events.count("stream")) == ((1, 0) if role == "A" else (0, 1)) + and receipt["socket_audit"]["installed"] is True and receipt["socket_audit"]["events"] == 0 + and set(receipt["checks"]) == WORKER_CHECKS and all(v is True for v in receipt["checks"].values()) + and type(receipt["context_reads"]) is int and receipt["context_reads"] >= 2 + and set(person) == PERSON_RECEIPT_KEYS + and person["receipt_hash"] == _hash({k: v for k, v in person.items() if k != "receipt_hash"}) + and person["schema"] == "continuityos.person-twin.r21h-evidence/v1" + and person["decision"] == "ALLOW" and person["action"] == "READ" + and person["evidence_count"] == 1 and person["memory_refs"] == [expected_evidence[0]["ref"]] + and person["prompt_boundary_revalidated"] is True + and all(person.get(k) == v and receipt.get(k) == v for k, v in _governance().items()) + and receipt["unsupported_execute"]["decision"] == "DENY" + and receipt["unsupported_execute"]["execution_authorized"] is False + and all(_is_sha(person[key]) for key in ("intent_sha256", "r4_receipt_hash", "evidence_sha256", "receipt_hash")) + and all(_is_sha(receipt[key]) for key in ("session_sha256", "request_sha256", "packet_sha256", "prompt_sha256"))) + except (KeyError, TypeError, ValueError): + return False + + +def _controller(*, timeout: float = 20.0) -> tuple[dict[str, Any], int]: + """Caller-side orchestrator; all three audited children are direct siblings.""" + root = None + root_identity = None + children: list[dict[str, Any]] = [] + effects = _effects() + value: dict[str, Any] = _hold("DEMO_NOT_STARTED") + started_at = datetime.now(timezone.utc).isoformat() + try: + _guard() + timeout = _timeout(timeout) + root = Path(tempfile.mkdtemp(prefix=ROOT_PREFIX)) + effects["ephemeral_filesystem_write"] = True + info = _owned_root(root).lstat() + root_identity = (info.st_dev, info.st_ino) + run_id = secrets.token_hex(16) + prepared = _run_prepare(root, run_id, timeout) + children.append(prepared) + if prepared.get("started") is True: + effects["ephemeral_memory_write"] = "UNKNOWN" + if not _valid_preparation(prepared, run_id): + raise PersonDemoHold("FIXTURE_PREPARATION_RECEIPT_INVALID") + preparation = prepared["receipt"] + db_sha = preparation["db_sha256"] + fixture = _read_fixture(root, db_sha, run_id) + if (preparation["memory_id"] != fixture["memory"]["id"] + or preparation["publication_id"] != fixture["publication_id"] + or preparation["stored_receipt_sha256"] != fixture["admission_bytes"]["receipt_sha256"]): + raise PersonDemoHold("FIXTURE_PREPARATION_SNAPSHOT_MISMATCH") + effects["ephemeral_memory_write"] = True + sessions = {} + for role in ("A", "B"): + result = _run_worker(root, db_sha, role, timeout, run_id) + children.append(result) + if not _valid_worker(result, role, fixture, db_sha): + raise PersonDemoHold("WORKER_RESULT_INVALID") + sessions[role] = result["receipt"] + if _read_fixture(root, db_sha, run_id) != fixture: + raise PersonDemoHold("FIXTURE_CHANGED_AFTER_WORKER") + a, b = sessions["A"], sessions["B"] + checks = {"separate_model_processes": a["process_id"] != b["process_id"] and os.getpid() not in (a["process_id"], b["process_id"]), + "same_current_evidence": a["captured_evidence"] == b["captured_evidence"] + and a["person_evidence"]["evidence_sha256"] == b["person_evidence"]["evidence_sha256"], + "fresh_session": a["session_sha256"] != b["session_sha256"], + "fresh_request_and_packet": a["request_sha256"] != b["request_sha256"] and a["packet_sha256"] != b["packet_sha256"], + "fresh_intent_and_r4": all(a["person_evidence"][key] != b["person_evidence"][key] + for key in ("intent_sha256", "r4_receipt_hash", "receipt_hash")), + "warm_a_cold_b": a["transport"] == "warm_chat" and b["transport"] == "cold_streaming", + "snapshot_bytes_unchanged": _read_fixture(root, db_sha, run_id) == fixture, + "same_saved_admission": a["stored_receipt_sha256"] == b["stored_receipt_sha256"] + == preparation["stored_receipt_sha256"] and a["publication_id"] == b["publication_id"] == preparation["publication_id"], + "no_admission_sidecar": {item.name for item in root.iterdir()} == {"memory.db"}} + if not all(checks.values()): + raise PersonDemoHold("CROSS_PROCESS_PROOF_MISMATCH") + value = {"schema": SCHEMA, "work_order": WORK_ORDER, "terminal": PASS, + "reason": "SYNTHETIC_PERSON_CONTEXT_RECOVERED", "mode": "SYNTHETIC_ONLY", "run_id": run_id, + "orchestrator_process_id": os.getpid(), "prepare": preparation, "sessions": sessions, + "checks": checks, "recovered": b["recovered"], "fixture_clock_at": AT, + "snapshot": {"admission_source": "MEMORY_SQLITE", "admission_sidecar_count": 0, + "db_sha256_before": fixture["memory"]["sha256"], "db_sha256_after": _sha(_read_bytes(root, "memory.db", MAX_DB_BYTES)), + "memory_ref": "mem:" + str(fixture["memory"]["id"]), + "candidate_sha256": fixture["admission_bytes"]["candidate_sha256"], + "pointer_sha256": fixture["admission_bytes"]["pointer_sha256"], + "stored_receipt_sha256": fixture["admission_bytes"]["receipt_sha256"], "publication_id": fixture["publication_id"]}, + "socket_audit": {"scope": "fixture_and_model_children", "audited_children": 3, "events": 0}, + "provenance": {"signed": False, "authority": "NONE", "synthetic_host_bindings": True}, **_governance()} + value["source_sha256"] = _source_sha256() + except Exception as exc: + value = _hold(str(exc) if type(exc) is PersonDemoHold else "DEMO_INTERNAL_HOLD") + finally: + effects["subprocess_execution"] = any(item.get("started") is True for item in children) + effects["child_launches"] = sum(item.get("started") is True for item in children) + if root is not None: + try: + info = _owned_root(root).lstat() + if root_identity != (info.st_dev, info.st_ino) or any( + item.get("started") and item.get("direct_child_reaped") is not True for item in children): + raise PersonDemoHold("CLEANUP_OWNERSHIP_UNCONFIRMED") + effects["ephemeral_memory_write"] = any(item.name == "memory.db" for item in root.iterdir()) + for item in root.iterdir(): + _reject_link(item.lstat()) + shutil.rmtree(root) + effects["temporary_cleanup_pass"] = not root.exists() + except Exception: + effects["temporary_cleanup_pass"] = False + if effects["temporary_cleanup_pass"] is not True: + value["terminal"], value["reason"] = HOLD, "TEMPORARY_CLEANUP_FAILED" + else: + effects["temporary_cleanup_pass"] = True + value["effects"] = effects + value["temporary_path_removed"] = effects["temporary_cleanup_pass"] + value["processes"] = [{k: v for k, v in item.items() if k not in {"receipt", "ok"}} for item in children] + value["started_at_utc"] = started_at + value["finished_at_utc"] = datetime.now(timezone.utc).isoformat() + return _seal(value), 0 if value["terminal"] == PASS else 2 + + +def run_person_continuity(*, timeout: float = 20.0) -> tuple[dict[str, Any], int]: + return _controller(timeout=timeout) + + +def render_person_continuity(value: Mapping[str, Any]) -> str: + passed = value.get("terminal") == PASS + lines = ["ContinuityOS Person continuity demo " + ("PASS" if passed else "HOLD")] + if passed: + lines.append("Boundary fresh A/B processes; synthetic model profiles; B cold") + lines.append("Admission same Memory SQLite; no receipt sidecar") + for key, label in (("goal", "Goal"), ("decision", "Decision"), ("constraint", "Constraint"), ("next_action", "Next step")): + lines.append(f"{label:<14}{value['recovered'][key]}") + lines.append("Memory ref " + str(value["snapshot"]["memory_ref"])) + else: + lines.append("Reason " + str(value.get("reason", "UNKNOWN"))) + lines.extend(["External AI NOT USED", "Production NOT_PRODUCTION_ADMITTED", + "Cleanup " + ("PASS" if value.get("temporary_path_removed") is True else "HOLD")]) + return "\n".join(lines) + + +def _install_child_socket_guard() -> None: + global _socket_guard_installed + def deny(event, args): + if event.startswith("socket."): + _socket_events.append(event) + raise RuntimeError("PERSON_DEMO_NETWORK_DENIED") + sys.addaudithook(deny) + _socket_guard_installed = True + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m continuityos.person_twin_demo", add_help=False) + parser.add_argument("operation", choices=["_prepare", "_worker"]) + parser.add_argument("--root", required=True) + parser.add_argument("--run-id") + parser.add_argument("--db-sha256") + parser.add_argument("--role", choices=["A", "B"]) + args = parser.parse_args(argv) + try: + _guard() + _install_child_socket_guard() + if args.operation == "_prepare": + result = _write_fixture(Path(args.root), args.run_id or "") + value, code = _seal({"schema": PREPARE_SCHEMA, "terminal": "PERSON_DEMO_PREPARE_PASS", + "process_id": os.getpid(), "socket_audit": _audit_state(), + "source_sha256": _source_sha256(), **result, **_governance()}), 0 + else: + value, code = _worker(Path(args.root), args.db_sha256 or "", args.role or "", args.run_id or "") + except Exception as exc: + value, code = _seal(_hold(str(exc) if type(exc) is PersonDemoHold else "CHILD_VALIDATION_HOLD")), 2 + print(_canonical(value).decode("utf-8")) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/DEMO_PERSON_CONTINUITY_R1.md b/docs/DEMO_PERSON_CONTINUITY_R1.md new file mode 100644 index 00000000..878d6313 --- /dev/null +++ b/docs/DEMO_PERSON_CONTINUITY_R1.md @@ -0,0 +1,207 @@ +# Person continuity demo — R1 + +Status: local candidate; scoped tests and installed-wheel demo PASS. +Work order: `COS_DEMO_PERSON_CONTINUITY_R1`. +Base commit: `28bfce3c1900d11bcc90fdd9428295ce18c68b14`. +Base tree: `fcb37ef8d9af2c349a62c576b5634798881a7fbe`. + +## User outcome + +`cos demo person-continuity` demonstrates a bounded Person/R21H read across a +real process boundary. A later process recovers the admitted synthetic goal, +current decision, constraint and next step from persisted inputs, and the demo +checks the exact evidence delivered to the recording model client. + +The existing Memory SQLite and R21H FAST runtime are reused. The demo composes +real R1 identity/consent, R2 privacy/provenance, R3 admission transitions and R4 +READ evaluation. Its model client records arguments; it performs no inference +and supplies no model-quality or model-latency evidence. + +## Entrypoint + +```bash +cos demo person-continuity +cos demo person-continuity --json +``` + +This command is a synthetic local product proof. The ordinary user's database, +external AI providers and native H3 are outside its input surface. + +The existing current-session containment continues to apply. A bound read-only +session must HOLD before demo filesystem or subprocess effects. Supplying a +user database through the public `--db` argument is rejected. + +## Process and data boundary + +The public orchestrator directly owns three sequential child processes: +PREPARE, A and B. Socket-denial instrumentation is installed inside each child; +the calling application's audit hooks are not changed. The parent owns each +process handle and its timeout/reap boundary, with no worker grandchildren. +Each read worker creates its own runtime, synthetic model profile, session, +request and intent. Worker A exercises warm FAST chat and worker B exercises +cold FAST acquisition. + +Child imports are pinned to the parent's actual package root and demo source +SHA-256, using isolated Python startup. A different working directory or +`PYTHONPATH` cannot silently select a different demo module. This is an import +consistency check over the trusted local package, not source-origin attestation. + +One owned temporary directory contains one Memory SQLite database and bounded +canonical fixture bytes. The PREPARE process uses the existing R3 transition +API before the workers run. It persists the candidate, CURRENT pointer and +admission receipt bytes required for subsequent R4 readback. + +PREPARE seeds the synthetic scenario. A and B independently recover that saved +scenario; neither reader is claimed to have originated a real owner's decision. + +The declared fixture clock is `2026-08-25T12:50:00Z`. Consent, delegation and +currentness checks use that synthetic clock. Real execution timestamps are +reported separately; the demo does not authenticate the current time or a live +owner's authorization. + +Worker B receives persisted input paths and commitments. It does not inherit +the earlier runtime, a ready prompt or an answer. Memory rows are not recreated +during B's read. The recording model's actual prompt is the proof surface. + +Three kinds of evidence must remain explicit in the result: + +| Input | What the demo can establish | +|---|---| +| Memory SQLite bytes | The synthetic memory survived closing and reopening in another process. | +| Persisted admission fixture bytes | The bounded R3 candidate, pointer and receipt were read back and checked. | +| Fresh synthetic host/session inputs | Each worker constructed different test session and request bindings. | + +The admission fixture is test input. It is not a production persistent +AdmissionStore, a second canonical database, an export, a Portable Twin Vault +or independently authenticated owner authority. A saved approval never becomes +a fresh authorization merely because it was persisted. + +## Acceptance requirements + +The implementation must establish all of the following before a PASS is recorded: + +- Successful, separately observed PREPARE, A and B process outcomes. +- A and B use distinct process/session/request identities and model profiles. +- Both protected READ paths bind the same exact current memory reference and + evidence content; old and foreign sentinels are excluded. +- The recovered goal, decision, constraint and next step come from the model + prompt's actual evidence projection. +- The SQLite and fixture bytes are unchanged by the worker reads. +- Tampered persisted evidence, revoked consent or stale currentness fails closed + before model generation. Existing R4 action semantics remain unchanged. +- Public database overrides, malformed worker inputs, invalid receipts and + containment failures do not become success. +- Socket operations are denied in the isolated PREPARE, A and B children; no live + model, connector, credential or external-network operation is part of the demo. +- Each child has a finite timeout of at most 20 seconds, plus bounded termination + and pipe-reader join grace periods; the argument is a per-child deadline. + Fixed fixture members are checked within the owned temporary directory without + following symlinks; parent traversal components and unexpected members fail closed. +- Timeout, worker failure or cleanup failure excludes PASS. Reported effects and + cleanup outcomes describe observations, not unconditional defaults. + +The tests target the new composition and process boundaries. Existing generic +demo, current-entrypoint containment and Person/R21H suites remain compatibility +evidence. + +## Scope of this result + +This is a synthetic fresh-process Person continuity proof. It does not establish +cross-model reasoning quality, complete authorized-state restore equivalence, +R5 lifecycle/export completion, Portable Twin Vault readiness, the full P3 +cross-gate acceptance family, native broker authentication or a real-data pilot. + +Project admission remains `NOT_PRODUCTION_ADMITTED`. +H3 C0 remains `UNVERIFIED/HOLD`; T11/T12 remain `UNEXECUTED`. +Execution, deployment and trading authority are not granted by a demo PASS. + +## Validation record + +The final scoped run on 2026-10-12 UTC used Python 3.12.14 and pytest 9.1.1: +**169 passed, 0 failed, 0 skipped**. The selected source/test hashes were identical +before and after execution. A separate parent audit hook denied socket events; +zero parent socket events were observed. Child audit scope is reported separately +by each real PREPARE/A/B receipt. + +| Suite | Passed | +|---|---:| +| `tests/test_product_person_continuity.py` | 27 | +| `tests/test_product_demo.py` | 8 | +| `tests/test_current_entrypoint_containment_v1.py` | 11 | +| `tests/test_person_twin_r21h_evidence.py` | 73 | +| `tests/test_person_twin_authority_replay.py` | 50 | +| Total | 169 | + +Reproduce the selected test set from the checkout with the existing test tooling: + +```bash +PYTHONDONTWRITEBYTECODE=1 PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 \ +python -m pytest -q -p no:cacheprovider \ + tests/test_product_person_continuity.py \ + tests/test_product_demo.py \ + tests/test_current_entrypoint_containment_v1.py \ + tests/test_person_twin_r21h_evidence.py \ + tests/test_person_twin_authority_replay.py +``` + +The recorded run used those pytest arguments inside the separate offline audit +recorder. The shell command above alone does not install that parent audit hook. +The developer's earlier 27-case run repeats the new cases and is not added to +the 169-case total. + +The new tests independently inspect SQLite through a read-only immutable +connection and compare its rows and bytes with both real workers' results. +The same positive run puts a failing shadow package in CWD and `PYTHONPATH`; +all children still use the expected parent package. Negative tests exercise +revoked consent, stale session bindings, changed or missing persisted bytes, +database override rejection, direct current-session containment, symlinks, +parent traversal and malformed or contradictory receipts. Real child processes +exercise timeout and output limits, with termination checked through their +actual `Popen` handles. A forced cleanup failure changes an otherwise successful +three-child run to HOLD. + +### Installed-package observation + +The wheel was built locally with the available setuptools backend, without +fetching dependencies, and installed into a new isolated Python environment. +The installed `cos` script was executed through `runpy.run_path` with arguments +`demo person-continuity --json`, under Python `-I -B` and a parent socket-denial +audit hook. The package imported from that environment's `site-packages`, with +the source checkout absent from `sys.path`; no tests are packaged in the wheel. + +The installed demo returned `COS_DEMO_PERSON_CONTINUITY_PASS` with three observed +direct children, each exited zero and reaped. PREPARE, A and B reported the same +frozen demo source SHA-256. A used one warm chat call and B one cold streaming +call. The exact current evidence and persisted SQLite/fixture hashes agreed; +session, request, packet, intent and R4 commitments differed. Parent and child +socket event counts were zero, and temporary cleanup was observed successful. + +The recovered synthetic context was: + +- Goal: continue the approved synthetic project. +- Decision: keep the read-only Person/R21H evidence path. +- Constraint: synthetic data only; no external effects. +- Next step: inspect the recovered current context in session B. + +These fields were extracted from the actual evidence delivered to the recording +client. They are not a generated model answer or real owner's remembered state. + +### Tested source identities + +| File | SHA-256 | +|---|---| +| `continuityos/person_twin_demo.py` | `2c1f95648f5c321555bc0ae7bf23e42195de6cd84bf4bbcf0ae4a7747556f417` | +| `continuityos/demo.py` | `e13dcc28ae1ee46e95b809ba3170808e6e3c3faf132b0b736d8074440cc62a49` | +| `tests/test_product_person_continuity.py` | `5098be2d8a60c6216e30f1113a7ca4a1a07c8241267df5be121f8706052762b7` | + +The local candidate manifest binds the final four-file payload and external +validation receipts. The observations establish this local synthetic composition; +they do not authenticate a hostile same-user filesystem, an independently +authenticated host or an atomic admission/SQLite/model-delivery transaction. +New GitHub CI and Windows execution were not performed for this local candidate. + +Independent source/test review passed on the exact hashes above. Review findings +about path traversal, child import consistency, receipt validation, process +identity and observed effects were resolved before the final scoped run. The +implementer, test author and reviewer were separate agent roles sharing the same +repository and fixture lineage; this is not independent owner attestation. diff --git a/docs/DEMO_PERSON_CONTINUITY_R2.md b/docs/DEMO_PERSON_CONTINUITY_R2.md new file mode 100644 index 00000000..88e31e12 --- /dev/null +++ b/docs/DEMO_PERSON_CONTINUITY_R2.md @@ -0,0 +1,170 @@ +# Person continuity demo R2 — durable SQLite admission + +Work order: `COS_PERSON_CONTINUITY_DURABLE_DEMO_R2`. +Specification: RC. Local synthetic candidate; no production admission. + +## Goal and baseline + +Wire the existing `cos demo person-continuity` command to the existing +`PersonTwinAdmissionStore`. A new process must recover the actual R3 admission +receipt from Memory SQLite, not receive a preassembled receipt file or recreate +that receipt by promoting the candidate again. + +Verified local parent: `8120fb04a592661e07ac26d7a20f21b639b3fa27`. +Parent tree: `fe58c85359aa448aa862c2432737c2a9161f4e25`. +Branch: `agent/person-continuity-durable-demo-r2`. + +Exactly three source paths are in scope: + +- `continuityos/person_twin_demo.py` (existing demo implementation). +- `tests/test_product_person_continuity.py` (existing product test suite). +- `docs/DEMO_PERSON_CONTINUITY_R2.md` (this new document). + +R1 documents/manifests, durable-store implementation/tests, R1–R4 contracts, +R21H, the command dispatcher and CI are not changed. In particular the shared +`_fresh_host_context`, `_EmptyReplayFixture` and `_RecordingModel` helpers retain +compatibility with the existing durable-store tests. No new runtime generation, +second Memory database or production service is introduced. + +## Run the candidate + +Run these commands only with the candidate checkout/package selected; this slice +does not install or switch the user's existing runtime: + +```text +cos demo person-continuity +cos demo person-continuity --json +``` + +The same existing dispatcher is usable from the candidate checkout: + +```text +py -3.11 -c "from continuityos.current_entrypoints import cos_main; raise SystemExit(cos_main(['demo','person-continuity','--json']))" +``` + +The public terminal names remain `COS_DEMO_PERSON_CONTINUITY_PASS` and +`COS_DEMO_PERSON_CONTINUITY_HOLD`. Structured demo, preparation and worker +reports use version 2 schemas and the R2 work-order name. They are unsigned +integrity records, not identity attestations or owner authorization. + +## Process and data flow + +The controller owns three bounded direct child processes: PREPARE, A and B. +It generates a fresh synthetic run ID before launching any child. The expected +Person identity is deterministically derived from that externally supplied run +ID and fixed fixture parameters; it is not selected from the database contents. + +PREPARE creates only an owned temporary `memory.db`. It writes the current +context and retained old/foreign sentinel records using existing Memory methods, +closes/checkpoints its own writer, and uses the existing durable-store API to +initialize admission tables and admit the current record through the real R3. +It requires confirmed publication and an exact saved receipt readback. +The database is quiescent in DELETE journal mode before reader handoff. This +journal choice concerns only the new temporary demo database. + +PREPARE's bounded, exact-shape result contains selectors and digests: run ID, +Memory ID, database hash, publication ID and saved-receipt hash. It cannot contain +the full admission receipt, candidate, source text or a context bundle. The +controller checks this shape before starting A. A/B receive only the owned root, +database digest, role and run ID. Neither A's context nor its ALLOW is input to B. + +A/B reopen the same database read-only. The store verifies its persisted candidate, +CURRENT pointer, publication and real R3 receipt. The demo constructs fresh +synthetic ledger, policy, session, request, packet and intent inputs, then passes +the actual read-only store to the existing Person evidence adapter and R21H. +The provider reopens/rechecks the pinned snapshot before each context projection. +The prepared database digest is checked before and after these reads. + +There is no admission sidecar, no renamed replacement file, and no receipt body +moved into command arguments or PREPARE stdout. Injecting the old sidecar is an +unexpected root member and causes HOLD. Missing or corrupt durable data never +falls back to the old file path, reseeds Memory or regenerates the R3 receipt. + +A uses warm chat; B uses the cold streaming path. Recording model fixtures capture +the actual outgoing prompt. Success requires one generation per worker, the same +current evidence and saved admission receipt, different fresh session/request/ +intent bindings, exclusion of old/foreign records, and DENY for EXECUTE. + +## Existing containment retained + +The public `--db` option remains rejected, and normal Memory database resolution +is not used. Declared current-session effect restrictions still stop this demo +before temporary I/O or child creation. Owned-root/no-link checks, source-pinned +isolated Python bootstrap, both-pipe output limits, child timeout/termination/reap, +and cleanup failure as HOLD are retained. + +Each child has the existing maximum 20-second execution budget; captured stdout +and stderr remain bounded at 64 KiB per pipe. These are specific demo budgets, not +universal performance or security thresholds. Cleanup is attempted only for the +owned temporary root after the direct children are confirmed reaped. Nothing in +this change cleans existing databases, WAL/SHM files or unrelated folders. + +## Validation and review evidence + +Validation records are stored outside the source tree under `evidence`. Successful +qualification must bind the current code/test hashes to the final scoped result, +a separate automated audit and an actual command-entrypoint observation. Earlier +failed runs remain separate historical evidence, never relabeled as success. + +The product suite covers real PREPARE/A/B processes and deliberately disables +`PersonTwinAdmissionStore.initialize`, `admit` and R3 promotion in reader children. +It also tests missing/altered schema, incomplete CURRENT/publication/receipt, +wrong identity, changed Memory rows, revoked consent, stale session/request at +initial and final prompt boundaries, sealed-but-invalid reports and receipt-body +injection into PREPARE. Deep structural mutations are tested with a recomputed +outer DB hash, so rejection is not credited solely to the file digest. + +Full selected compatibility includes the unchanged durable-store and R1–R4/R21H +suites. A separate deterministic audit checks scope, unchanged shared helper +ASTs and absence of sidecar and reader-write paths. This is automated evidence, +not an independent external human/model source review. + +## Limits + +All data, consent/time/policy/session inputs and model profiles are synthetic. +The host, Python environment and fixture issuer remain trusted. Hashes and the +run ID do not authenticate a hostile host or database owner. This is not C0, +source-origin or egress authentication, live-model inference, a Portable Twin +Vault restore, hardware power-loss qualification or production pilot admission. + +The SQLite receipt proves historical admission; it does not create current READ +or EXECUTE authority. External consent updates and later model delivery are not +made one atomic transaction by this demo. There is no durable model-delivery +replay ledger. Python socket audit observations are not OS packet capture or +containment of arbitrary native code. + +No push, PR edit, merge, deployment, Scheduler/TPM/service modification, real +Memory use or live model access is part of this local work order. C0 remains +UNVERIFIED/HOLD; native T11/T12 remain UNEXECUTED; the product remains +NOT_PRODUCTION_ADMITTED, execution_authority=NONE, can_execute=false, +can_trade=false and capital_permission=DENY. + +## Final execution record + +Final scoped run on Windows / Python 3.11.9 / pytest 9.1.1 / SQLite 3.45.1: +**320 passed, 1 skipped, 0 failures, 0 errors**. All 52 runnable product-demo +cases passed. The one pre-existing real-symlink case was skipped because this +Windows process could not create the link; it is not counted as PASS. Code and +test hashes were unchanged. The audit hook observed no socket events; the +validation temporary root was removed. + +A separate invocation of the existing cos_main command entrypoint returned PASS: +three real direct children, Memory reference mem:3, no admission sidecar, the +same stored receipt and actual prompt in A/B, distinct fresh sessions, one warm +chat and one cold streaming call, and verified temporary cleanup. Neither this +invocation nor the tests used a live model or the user Memory database. + +Failure history is retained: the first product-only run had 36 PASS, 1 SKIP and +16 setup errors caused by one shared fixture returning WORKER_RESULT_INVALID. +Its complete failed worker report was not retained, so its precise underlying +cause is not established. A targeted guarded A worker and the complete guarded +CWD/PYTHONPATH-isolation fixture then passed on the same runtime bytes. The only +subsequent test edit enriched the failure assertion with worker reports; no gate +or timeout was relaxed and no runtime fix is claimed for that early failure. +The final full scoped run and normal CLI observation above passed. The early +non-reproduced failure remains a reproducibility caveat, not erased evidence. + +The local result is Tested / PASS WITH CONDITIONS for the observed synthetic +Windows environment. Sealing additionally requires the separate automated audit +to pass. No Linux, installed-wheel or full-repository CI run is claimed for R2, +and no external independent human/model review was performed. diff --git a/docs/PERSON_TWIN_DURABLE_ADMISSION_R1.md b/docs/PERSON_TWIN_DURABLE_ADMISSION_R1.md new file mode 100644 index 00000000..77224f7b --- /dev/null +++ b/docs/PERSON_TWIN_DURABLE_ADMISSION_R1.md @@ -0,0 +1,277 @@ +# Person Twin durable admission — R1 local candidate + +Work order: `COS_PERSON_TWIN_DURABLE_ADMISSION_R1`. + +Implementation: **Tested in the bounded synthetic Windows environment described below**. +Review conclusion: **PASS WITH CONDITIONS for a local candidate**, not production admission. +Independent external human/model source review: **NOT PERFORMED**. Author review and a +separate deterministic audit process are identified explicitly; they are not independent agents. + +## Purpose and exact scope + +Persist the admission evidence required by the existing R4/R21H read path across a +process boundary without manufacturing a replacement R3 receipt at read time. +The ordinary Memory database remains the single SQLite file. An explicit store +adapter adds its own tables; it does not create a second memory database or modify +`items`, the existing R3 coordinator, R4 evaluator, Person evidence adapter, R21H, +CLI, MCP, or the earlier continuity demo. + +Exactly three new repository files: + +- `continuityos/person_twin_admission_store.py` +- `tests/test_person_twin_admission_store.py` +- `docs/PERSON_TWIN_DURABLE_ADMISSION_R1.md` + +The existing `cos demo person-continuity` command remains unchanged and still uses +its original fixture sidecar. This slice proves the new durable path through a +separate synthetic test harness; it does **not** switch that command to the store. + +## Baseline recovery + +The former local demo commit `fa642ba9b81c926d78a0d61dd4169d57b90ed65e` was +unavailable. The owner separately approved replacing it with a new recovery commit +only after proving full-tree identity. + +Verified public parent: `28bfce3c1900d11bcc90fdd9428295ce18c68b14`. +Parent tree: `fcb37ef8d9af2c349a62c576b5634798881a7fbe`. +Original supplied patch SHA-256: +`56867bd83c9e8ff4d15e93f81e020d02448fa91a2eff1d96c2589d2e65a0ab3e`. + +The patch applied cleanly to a fresh, full checkout. `git write-tree` reproduced +`b989e36d3703b0a6dd8b5edd60ea50367c864f1e` exactly, including all unchanged files. +New recovery commit: `3b166d4beacc1a81f17ee07e2a042f192914e326`. +This is a different commit with the same approved source tree, not the missing +original object. Git history is shallow at the verified public parent; checkout +contents are complete. Initial `git fsck --full` and clean-worktree checks passed. + +The local implementation branch is `agent/person-durable-admission-r1`. Work is +kept in a new isolated directory on the authorized owner workstation, not in the +older shared worktrees. No GitHub write, PR edit, merge, deployment, service, +Scheduler, TPM, credential, trading, or real-memory operation is part of this slice. + +## Storage and transaction contract + +`PersonTwinAdmissionStore` opens an explicitly supplied existing Memory database. +The constructor is read-only by default. Missing files or admission schemas are +errors; read-only opening never creates the database or admission schema. +`initialize(path)` is a separate explicit schema-write operation on an existing +Memory database. Initialization is idempotent only for the exact supported schema; +partial, altered or unexpected schemas are not silently repaired or migrated. + +The adapter owns a separate SQLite connection. Writers use explicit transactions, +`foreign_keys=ON`, `trusted_schema=OFF`, and `synchronous=FULL`; readers use +`query_only=ON`. These relevant settings and the exact schema are rechecked at +write boundaries. The existing database journal mode is not changed. Each store +instance belongs to one thread; independent writers open independent instances. + +The schema has four tables: + +1. A version marker. +2. Scoped candidate bytes, historical source/consent/identity anchors, and exact + Memory row binding, each with an integrity hash. +3. Append-only-by-API publications containing the actual R3 CURRENT receipt, + pointer, candidate binding and previous-publication identity. +4. One CURRENT pointer per tenant/twin identity fingerprint. + +Candidate staging is committed separately as non-current. Only the configured +`admit()` sequence enables the existing R3 storage callbacks. Direct candidate +writes and pointer promotion through those callbacks are rejected otherwise. + +R3 performs its unchanged state transitions: + +`STAGED -> VALIDATED -> COMMITTED_NOT_CURRENT -> POSTVERIFY_PASS -> CURRENT`. + +Before promotion, `BEGIN IMMEDIATE` acquires the writer transaction. The expected +publication identity and R3 pointer SHA are checked. R3 writes its pointer into +this transaction and produces its real final receipt. The adapter then inserts +that exact receipt and its linked publication before the single promotion COMMIT. +A deferred composite foreign key rejects COMMIT of a CURRENT pointer that has no +matching publication/receipt row. Candidate identity cannot be rebound to another +Memory ID or namespace, even if its text is identical. + +The transaction is verified before commit with the existing R4 structural +admission validator. After COMMIT, a **new read-only connection** reads and checks +the whole snapshot. Success requires exact equality with the committed candidate, +pointer, receipt, anchors and binding. A concurrent change in the intervening +window gives HOLD, not a stale success or automatic retry. + +Reference semantics: SQLite explicit transactions and deferred constraints are +specified in https://www.sqlite.org/lang_transaction.html and +https://www.sqlite.org/foreignkeys.html. Python explicit transaction handling is +described in https://docs.python.org/3.11/library/sqlite3.html. This implementation's +observed results are the tests below, not a blanket claim about every deployment. + +## Results and uncertain writes + +The operation reports separate candidate and promotion commit states: +`NOT_ATTEMPTED`, `UNKNOWN`, or `CONFIRMED`. + +- PASS requires confirmed promotion plus fresh readback verification. +- A lost candidate COMMIT acknowledgement returns HOLD with + `CANDIDATE_COMMIT_OUTCOME_UNKNOWN`, `candidate_persisted=null`, and + `possible_persisted_effect=true`. +- A lost promotion COMMIT acknowledgement returns HOLD with + `COMMIT_OUTCOME_UNKNOWN`. A new reader may subsequently find a complete saved + bundle; the original caller must not infer that no write happened. +- Confirmed commit followed by failed readback returns + `POSTCOMMIT_READBACK_FAILED`; it is not described as rollback. +- Failed rollback returns `ROLLBACK_UNVERIFIED`, marks a possible effect, suppresses + the successful receipt, and suspends/closes that connection. + +There is no automatic replay after uncertainty. A staged candidate may remain +when later validation fails. Initialization exceptions also must not be interpreted +as proof of an unchanged file after an uncertain commit. No compensating deletion +or implicit repair is performed. + +## Read contract and authority boundary + +`load_current()` verifies the persisted snapshot in a read transaction. Immutable +snapshot bytes and separately decoded dictionaries prevent accidental mutation of +previously returned bytes. Existing R3/R4 readback methods read actual SQLite data, +not a cached ALLOW. Candidate/current lookup is bound to the supplied tenant, twin +and identity fingerprint. + +A saved receipt is **historical admission evidence**, not current authorization. +The reader must still obtain fresh independently trusted time, revocation/consent, +policy, principal, session, request and intent bindings. The test harness combines +the saved bundle with newly generated synthetic host/session inputs and the +unchanged R4/R21H evaluator. No SQL column supplies a new grant. + +All operation results retain: + +```text +production_admission_status=NOT_PRODUCTION_ADMITTED +execution_authority=NONE +can_execute=false +can_trade=false +capital_permission=DENY +``` + +Current-session effect checks are reused on every public write boundary and +material commit boundary. The adapter does not bypass an active declared current +mode. In the lab with no such declaration, the trusted host is responsible for the +explicit owner-approved operation. This API is not an independently authenticated +owner signer or a production authorization service. + +## API shape + +This fragment documents the API, not runnable authority or real-data permission: + +```python +# path must be an explicitly approved existing synthetic Memory database. +PersonTwinAdmissionStore.initialize(path) +with PersonTwinAdmissionStore(path, identity=identity, read_only=False) as writer: + result = writer.admit( + consent_receipt=consent_receipt, + source_record=source_record, + candidate_record=candidate_record, + purpose="person-memory", + memory_id=memory_id, + namespace=namespace, + read_current_evidence=read_fresh_ledger_and_time, + ) +# A new process supplies the controller's expected identity again. +with PersonTwinAdmissionStore(path, identity=identity) as reader: + snapshot = reader.load_current() +# Still run fresh R4/R21H authorization before disclosing snapshot contents. +``` + +`read_fresh_ledger_and_time()` must return exactly +`{"current_revocation_ledger": ..., "at": ...}`. A missing/extra field, invalid time +or revoked/expired evidence is rejected. The 1 MiB record limit and bounded +connection timeout are lab limits, not calibrated universal production thresholds. + +## Executed validation + +Executed on 2026-10-13 on Windows, Python **3.11.9**, pytest **9.1.1**, SQLite +**3.45.1**. Source and tests remained hash-identical throughout the final run. + +Final scoped selection, importlib mode: **294 passed, 1 skipped, 0 failures, +0 errors**. All **52 new durable-admission cases passed**. The single skipped +pre-existing demo case requires a real filesystem symlink that this Windows +process could not create. It is not counted as a passed link-isolation test. + +Selected files: + +```text +tests/test_person_twin_admission_store.py +tests/test_person_twin_admission_contracts.py +tests/test_person_twin_privacy_provenance.py +tests/test_person_twin_admission_coordinator.py +tests/test_person_twin_authority_replay.py +tests/test_person_twin_r21h_evidence.py +tests/test_product_person_continuity.py +``` + +The final runner denied every `socket.*` Python audit event; zero occurred in its +process. Explicit spawned durable-test workers independently install that guard; +the existing demo workers retain their own guard. This is not OS packet capture +or an isolation claim against arbitrary native code. + +Coverage includes actual A/B process separation, receipt equality, fresh session +inequality, real R4 decisions, actual captured R21H prompt JSON, exactly one model +call per worker, and exclusion of old/foreign records. Process B receives only +the DB path and expected identity, not a serialized admission sidecar. R3 promotion +is made to fail if invoked in B, establishing that the reader did not regenerate +the receipt. The model is a recording fixture, not live inference. + +Two real competing writer processes produce one winner. Separate real processes +exit at code 73 after the transactional pointer write or immediately before COMMIT; +the old complete publication survives. Fault injection also tests commit +acknowledgement loss, failed postcommit reopen, unverified rollback, lock contention, +reentry, wrong identity/scope/Memory binding, changed consent/session/request, +corrupt bytes/schema, and disabled foreign-key enforcement. + +## Review and retained failure history + +The author's initial run was 36 PASS / 2 FAIL. One failure identified a real +candidate-commit uncertainty reporting defect: the write could persist while +`possible_persisted_effect` was false. Separate tri-state tracking corrected it. +The other failure was a test assertion against raw multiline text embedded in JSON; +the test now decodes and checks the actual evidence payload. The repaired initial +38-case suite passed, then 14 further boundary cases were added. + +Several validation-recorder attempts failed without changing source semantics. +One called `platform.platform()`, which invokes a local hostname socket audit event +on Windows and was correctly denied. Another gave pytest a missing parent folder; +a later custom temp root was outside the OS temporary directory required by the +unchanged demo and caused expected root/cleanup rejection. The final recorder uses +`sys.platform`, an existing OS-temp parent, a new dedicated temp directory, and the +same complete seven-file test selection. Earlier logs are retained, not relabeled. + +A separate Python process performed a deterministic AST/import/effect-label/Git +check and independently exercised the extracted DDL in an in-memory SQLite DB. +It confirmed that committing a pointer without its receipt is rejected and rollback +restores the empty pointer table. This audit passed on the exact current module and +test hashes. It is additional automated evidence, **not** an independent human or +model review of all source semantics. + +## Remaining limits and next boundary + +The host and controller identity inputs remain trusted. Hashes do not authenticate +a hostile DB owner; full-file rollback, malicious coordinated SQL/hash rewrites, +key custody, trusted monotonic time, source origin, broker/client identity and +egress consent remain separate problems. File identity and no-link checks narrow +ordinary path mistakes; they do not prove race-free native handle custody. + +Process exits tested here are not hardware power-loss, disk-full, filesystem or +storage-controller durability qualification. Read-only SQLite may interact with +existing WAL/SHM according to SQLite's own rules; this slice does not delete or +repair sidecars or claim zero filesystem metadata effects for every reader. + +The connection/settings, local snapshot checks and writer transaction do not make +external consent updates, the later model sink, or disclosure delivery one atomic +transaction. No durable model-delivery or execution replay ledger is added. +No full repository CI, installed-wheel test of this new slice, Linux test, live +model comparison, external independent source review, personal pilot or production +qualification is claimed. + +C0 remains **UNVERIFIED/HOLD**; native T11/T12 remain **UNEXECUTED**. The next +product wiring can replace the old demo's receipt sidecar with this store under a +separate reviewed change. This local commit does not perform that switch. + +Evidence, generated manifests, patches and Git bundles are stored outside the +repository source tree. A Git bundle with the verified public parent prerequisite +preserves both recovery and this implementation without requiring a push. The +final candidate manifest binds the commit/tree, all three payload files and actual +receipts. It is an unsigned integrity record, not an owner authority grant. diff --git a/tests/test_person_twin_admission_store.py b/tests/test_person_twin_admission_store.py new file mode 100644 index 00000000..539eaa18 --- /dev/null +++ b/tests/test_person_twin_admission_store.py @@ -0,0 +1,633 @@ +"""Offline durable admission proofs using temporary synthetic Memory databases. + +Subprocess B receives only the DB path and the controller's expected identity, +not A's source/candidate/pointer/receipt or a cached authorization decision. +Model clients record calls. These tests do not prove live host/owner authority. +""" +from __future__ import annotations + +import copy +from dataclasses import replace +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import runpy +import sqlite3 +import subprocess +import sys +import time + +import pytest + +from continuityos.embed import HashingEmbedder +from continuityos.memory import Memory +from continuityos import person_twin_admission_store as durable +from continuityos import person_twin_demo as demo +from continuityos.person_twin_admission_contracts import ( + build_consent_revocation_entry, build_consent_revocation_ledger, build_person_twin_identity, +) +from continuityos.person_twin_authority_replay import evaluate_person_twin_authority_replay +from continuityos.person_twin_privacy_provenance import build_person_twin_provenance +from continuityos.person_twin_r21h_evidence import PersonReadIntent, PersonTwinR21HEvidenceAdapter, PersonTwinEvidenceError +from continuityos.sovereign_twin_runtime import DEFAULT_PROFILES, SovereignTwinRuntime + +AT = "2026-08-25T12:50:00Z" +NS = "synthetic-durable-admission" +QUERY = "Restore the admitted current context." +TEXT = ("Goal: resume the project after a process restart.\n" + "Decision: keep the existing R21H engine.\n" + "Constraint: synthetic data only; no external effects.\n" + "Next step: inspect the recovered evidence and admission receipt.") + + +def _fixture_module(): + path = Path(__file__).with_name("test_person_twin_authority_replay.py") + spec = importlib.util.spec_from_file_location("_durable_r4_fixture", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _data(): + ctx = _fixture_module()._inputs() + source = copy.deepcopy(ctx["source_record"]) + source["payload"] = {"title": "Synthetic durable context", "text": TEXT} + source["content_hash"] = durable._hash(durable._canonical(source["payload"])) + record = build_person_twin_provenance(ctx["identity"], ctx["consent_receipt"], + ctx["current_revocation_ledger"], source, admission_session_id="admission:durable:r1", + purpose="person-memory", at="2026-08-25T11:00:00Z") + return {**ctx, "source_record": source, "person_record": record} + + +def _write_memory(path, ctx): + memory = Memory(str(path), embedder=HashingEmbedder()) + try: + memory.remember("OLD_DECISION_NOT_ADMITTED", namespace=NS) + memory.remember("FOREIGN_PRIVATE_NOT_ADMITTED", namespace="foreign-person") + rid = memory.remember(ctx["source_record"]["payload"]["text"], namespace=NS) + finally: + memory.store.con.close() + return rid + + +def _fresh(ctx): + return {"current_revocation_ledger": copy.deepcopy(ctx["current_revocation_ledger"]), "at": AT} + + +def _admit(store, ctx, rid, reader=None, namespace=NS): + return store.admit(consent_receipt=ctx["consent_receipt"], source_record=ctx["source_record"], + candidate_record=ctx["person_record"], purpose="person-memory", memory_id=rid, namespace=namespace, + read_current_evidence=(lambda: _fresh(ctx)) if reader is None else reader) + + +@pytest.fixture +def ready(tmp_path): + ctx, path = _data(), tmp_path / "memory.db" + rid = _write_memory(path, ctx) + durable.PersonTwinAdmissionStore.initialize(path) + return path, ctx, rid + + +def _context(snapshot, store, host=None, ledger=None): + anchors = snapshot.anchors + if host is None: + host = demo._fresh_host_context({"context": {"identity": anchors["identity"], + "person_record": snapshot.candidate}}, "B") + return {"identity": anchors["identity"], "consent_receipt": anchors["consent_receipt"], + "source_record": anchors["source_record"], "person_record": snapshot.candidate, + "admission_receipt": snapshot.receipt, "current_pointer": snapshot.pointer, + "admission_store": store, "replay_store": demo._EmptyReplayFixture(), + "current_revocation_ledger": build_consent_revocation_ledger() if ledger is None else ledger, **host} + + +def _run_r21h(path, store, *, drift=None): + snapshot = store.load_current() + assert snapshot is not None + ctx = _context(snapshot, store) + query_hash = durable._hash(QUERY.encode()) + session_hash = durable._hash(durable._canonical(ctx["current_session"])) + binding = snapshot.binding + intent = PersonReadIntent(query_sha256=query_hash, current_session_sha256=session_hash, + request_id=ctx["authorization_request"]["request_id"], packet_id=ctx["authorization_preflight"]["packet_id"], + memory_id=binding["memory_id"], namespace=binding["namespace"], + person_record_id=snapshot.candidate["id"], source_record_id=snapshot.anchors["source_record"]["id"], + principal_id=ctx["principal_id"], scope=ctx["requested_scope"], privacy_class=ctx["requested_privacy_class"], + purpose=ctx["purpose"]) + if drift is not None: + drift(ctx) + class Provider: + def read_context(self): + latest = store.load_current() + assert latest is not None + return {**ctx, "person_record": latest.candidate, "current_pointer": latest.pointer, + "admission_receipt": latest.receipt} + profile = replace(DEFAULT_PROFILES["fast"], model="synthetic-durable-fast") + model = demo._RecordingModel(profile, cold=False) + runtime = SovereignTwinRuntime(str(path), profiles={**DEFAULT_PROFILES, "fast": profile}, + client=model, person_evidence_adapter=PersonTwinR21HEvidenceAdapter(intent, Provider())) + try: + if drift is not None: + with pytest.raises(PersonTwinEvidenceError): + runtime.ask(QUERY) + assert model.calls == [] + return {"decision": "DENY", "model_calls": 0} + answer = runtime.ask(QUERY) + assert len(model.calls) == 1 + call = model.calls[0] + prompt = call["system_prompt"] + evidence = json.loads(prompt.split("MEMORY_EVIDENCE_JSON:\n", 1)[1]) + assert len(evidence) == 1 and evidence[0]["text"] == TEXT + assert "OLD_DECISION_NOT_ADMITTED" not in prompt and "FOREIGN_PRIVATE_NOT_ADMITTED" not in prompt + assert f"mem:{binding['memory_id']}" in prompt + assert answer.stats["person_evidence"]["decision"] == "ALLOW" + return {"decision": "ALLOW", "model_calls": 1, "prompt_hash": durable._hash(prompt.encode()), + "session_hash": session_hash, "receipt_hash": durable._hash(snapshot.receipt_bytes)} + finally: + runtime.close() + + +def test_actual_receipt_and_same_db_row_survive_reopen(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + outcome = _admit(store, ctx, rid) + assert outcome["status"] == "PASS", outcome + assert outcome["commit_state"] == "CONFIRMED" + first = store.load_current() + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + recovered = reader.load_current() + assert recovered == first + assert recovered.receipt_bytes == durable._canonical(outcome["admission_receipt"]) + assert recovered.binding["memory_id"] == rid + assert recovered.candidate["payload"]["text"] == TEXT + assert evaluate_person_twin_authority_replay(**_context(recovered, reader))["decision"] == "ALLOW" + assert reader._con.execute("PRAGMA query_only").fetchone()[0] == 1 + assert reader._con.execute("SELECT COUNT(*) FROM items").fetchone()[0] == 3 + assert outcome["production_admission_status"] == "NOT_PRODUCTION_ADMITTED" + assert outcome["execution_authority"] == "NONE" and outcome["can_execute"] is False + assert outcome["can_trade"] is False and outcome["capital_permission"] == "DENY" + + +@pytest.mark.parametrize("kind", ["missing", "memory-no-schema", "partial-schema", "wrong-version", "wrong-ddl", "trigger"]) +def test_read_only_missing_or_corrupt_schema_never_initializes(tmp_path, kind): + ctx, path = _data(), tmp_path / "isolated.db" + if kind != "missing": + _write_memory(path, ctx) + if kind not in {"missing", "memory-no-schema"}: + durable.PersonTwinAdmissionStore.initialize(path) + with sqlite3.connect(path) as con: + if kind == "partial-schema": + con.execute("DROP TABLE person_admission_current") + elif kind == "wrong-version": + con.execute("UPDATE person_admission_meta SET schema_id='other'") + elif kind == "wrong-ddl": + con.execute("ALTER TABLE person_admission_meta ADD COLUMN unexpected TEXT") + else: + con.execute("CREATE TRIGGER unsafe_admission AFTER INSERT ON person_admission_meta BEGIN SELECT 1; END") + before = None if not path.exists() else path.read_bytes() + with pytest.raises(durable.DurableAdmissionError): + durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) + assert (None if not path.exists() else path.read_bytes()) == before + + +def test_schema_creation_is_explicit_idempotent_and_preserves_memory(ready): + path, ctx, rid = ready + with sqlite3.connect(path) as con: + before = con.execute("SELECT * FROM items ORDER BY id").fetchall() + durable.PersonTwinAdmissionStore.initialize(path) + with sqlite3.connect(path) as con: + assert con.execute("SELECT * FROM items ORDER BY id").fetchall() == before + assert con.execute("SELECT COUNT(*) FROM person_admission_current").fetchone()[0] == 0 + + +@pytest.mark.parametrize("action", ["candidate", "pointer", "readonly-admit"]) +def test_direct_write_callbacks_cannot_bypass_real_r3(ready, action): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as store: + with pytest.raises(durable.DurableAdmissionError): + if action == "candidate": + store.commit_candidate_not_current(ctx["person_record"]["id"], durable._canonical(ctx["person_record"])) + elif action == "pointer": + store.promote_current_pointer(expected_pointer_sha256=None, pointer_bytes=b'{}') + else: + _admit(store, ctx, rid) + assert store.load_current() is None + + +@pytest.mark.parametrize("phase", [1, 2, 3, 4]) +def test_revocation_at_each_boundary_prevents_current_publication(ready, phase): + path, ctx, rid = ready + calls = 0 + entry = build_consent_revocation_entry(ctx["identity"], ctx["consent_receipt"], + revoking_principal="human:owner", revoking_principal_kind="HUMAN", + revoked_at="2026-08-25T12:00:00Z", reason="synthetic boundary revocation") + def reader(): + nonlocal calls + calls += 1 + value = _fresh(ctx) + if calls >= phase: + value["current_revocation_ledger"] = build_consent_revocation_ledger([entry]) + return value + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + value = _admit(store, ctx, rid, reader) + assert value["status"] == "HOLD" + assert value["admission_receipt"] is None + assert store.load_current() is None + assert store._con.execute("SELECT COUNT(*) FROM person_admission_publications").fetchone()[0] == 0 + + +@pytest.mark.parametrize("drift", ["text", "namespace", "superseded", "missing"]) +def test_memory_drift_before_promotion_never_publishes(ready, drift): + path, ctx, rid = ready + calls = 0 + def reader(): + nonlocal calls + calls += 1 + if calls == 3: + with sqlite3.connect(path) as con: + if drift == "text": + con.execute("UPDATE items SET text='SYNTHETIC_TAMPER' WHERE id=?", (rid,)) + elif drift == "namespace": + con.execute("UPDATE items SET namespace='foreign-person' WHERE id=?", (rid,)) + elif drift == "superseded": + con.execute('UPDATE items SET meta=? WHERE id=?', ('{"superseded_by":99}', rid)) + else: + con.execute("DELETE FROM items WHERE id=?", (rid,)) + return _fresh(ctx) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + outcome = _admit(store, ctx, rid, reader) + assert outcome["status"] == "HOLD" + assert store.load_current() is None + + +def test_same_candidate_cannot_be_rebound_to_another_memory_id(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + old = store.load_current() + memory = Memory(str(path), embedder=HashingEmbedder()) + try: + other = memory.remember(TEXT, namespace=NS) + finally: + memory.store.con.close() + value = _admit(store, ctx, other) + assert value["status"] == "HOLD" + assert store.load_current() == old + + +@pytest.mark.parametrize("table,column", [("person_admission_candidates", "candidate"), + ("person_admission_candidates", "anchors"), ("person_admission_candidates", "binding"), + ("person_admission_publications", "receipt"), ("person_admission_current", "pointer")]) +def test_corrupt_stored_bundle_fails_closed(ready, table, column): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + with sqlite3.connect(path) as con: + con.execute(f"UPDATE {table} SET {column}=?", (b'{}',)) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + with pytest.raises(durable.DurableAdmissionError): + reader.load_current() + + +def test_different_valid_twin_cannot_read_other_identity(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + other = build_person_twin_identity(tenant_id="tenant:other", subject_id="subject:other", owner_id="human:other", + controller_id="human:other", controller_kind="HUMAN", ownership_epoch=1, created_at="2026-08-25T00:00:00Z") + with durable.PersonTwinAdmissionStore(path, identity=other) as reader: + assert reader.load_current() is None + assert reader.read_candidate_bytes(ctx["person_record"]["id"]) is None + + +@pytest.mark.parametrize("which", ["session", "request", "revocation"]) +def test_fresh_read_authority_still_required_before_model_call(ready, which): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + def change(context): + if which == "session": + context["current_session"]["challenge_id"] += ":stale" + elif which == "request": + context["authorization_request"]["request_id"] += ":stale" + else: + entry = build_consent_revocation_entry(context["identity"], context["consent_receipt"], + revoking_principal="human:owner", revoking_principal_kind="HUMAN", + revoked_at="2026-08-25T12:00:00Z", reason="synthetic read revocation") + context["current_revocation_ledger"] = build_consent_revocation_ledger([entry]) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + assert _run_r21h(path, reader, drift=change)["model_calls"] == 0 + + +class _CommitAckLoss: + def __init__(self, store, phase): + self.store, self.real, self.phase, self.calls = store, store._con, phase, 0 + def __getattr__(self, name): + return getattr(self.real, name) + def execute(self, sql, args=()): + if sql == "COMMIT" and self.store._phase == self.phase: + self.calls += 1 + self.real.execute(sql, args) + raise OSError("SYNTHETIC_COMMIT_ACK_LOST_PRIVATE_TEXT") + return self.real.execute(sql, args) + + +@pytest.mark.parametrize("phase", ["CANDIDATE", "PROMOTING"]) +def test_lost_commit_ack_is_hold_with_possible_persisted_effect(ready, phase): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + fault = _CommitAckLoss(store, phase) + store._con = fault + value = _admit(store, ctx, rid) + assert value["status"] == "HOLD" and value["possible_persisted_effect"] is True, value + assert value["admission_receipt"] is None and fault.calls == 1 + assert "PRIVATE_TEXT" not in json.dumps(value) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + snapshot = reader.load_current() + assert (snapshot is not None) == (phase == "PROMOTING") + assert reader.read_candidate_bytes(ctx["person_record"]["id"]) is not None + + +def test_failed_postcommit_reopen_is_not_reported_as_rollback(ready, monkeypatch): + path, ctx, rid = ready + original = durable.PersonTwinAdmissionStore.load_current + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + def fail_new(instance): + if instance is not store: + raise OSError("SYNTHETIC_READBACK_UNAVAILABLE") + return original(instance) + with monkeypatch.context() as m: + m.setattr(durable.PersonTwinAdmissionStore, "load_current", fail_new) + value = _admit(store, ctx, rid) + assert value["status"] == "HOLD" and value["commit_state"] == "CONFIRMED" + assert value["reason"] == "POSTCOMMIT_READBACK_FAILED" and value["possible_persisted_effect"] + assert store.load_current() is not None + + +def test_declared_current_mode_blocks_writes_without_schema_mutation(ready, monkeypatch): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + monkeypatch.setenv("CONTINUITYOS_CURRENT_SESSION_REQUIRED", "1") + value = _admit(store, ctx, rid) + assert value["status"] == "HOLD" + assert store.load_current() is None + with pytest.raises(Exception): + durable.PersonTwinAdmissionStore.initialize(path) + with pytest.raises(Exception): + durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) + + +def _worker(mode, path, expected_identity, token=""): + if mode == "write": + ctx = _data() + rid = _write_memory(path, ctx) + durable.PersonTwinAdmissionStore.initialize(path) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + out = _admit(store, ctx, rid) + assert out["status"] == "PASS", out + observation = _run_r21h(path, store) + return {"pid": os.getpid(), "publication_id": out["publication_id"], **observation} + if mode == "read": + # A reader must not manufacture a replacement R3 receipt after restart. + def forbidden(*args, **kwargs): + raise AssertionError("R3_MUST_NOT_RUN_IN_READER") + durable.PersonTwinAdmissionCoordinator.promote_current = forbidden + with durable.PersonTwinAdmissionStore(path, identity=expected_identity) as reader: + snap = reader.load_current() + assert snap is not None + return {"pid": os.getpid(), "publication_id": snap.publication_id, **_run_r21h(path, reader)} + ctx = _data() + with durable.PersonTwinAdmissionStore(path, identity=expected_identity, read_only=False, timeout=3) as store: + if mode == "race": + calls = 0 + def barrier(): + nonlocal calls + calls += 1 + if calls == 1: + path.with_name(token + ".ready").write_text("ready", encoding="ascii") + end = time.monotonic() + 8 + while not all(path.with_name(x + ".ready").exists() for x in ("A", "B")): + if time.monotonic() > end: + raise TimeoutError("SYNTHETIC_BARRIER_TIMEOUT") + time.sleep(0.02) + return _fresh(ctx) + return _admit(store, ctx, 3, barrier) + def crash(sql): + if store._phase == "PROMOTING" and ( + (mode == "crash-after-pointer" and sql.startswith("INSERT INTO person_admission_publications")) or + (mode == "crash-precommit" and sql == "COMMIT")): + os._exit(73) + store._con.set_trace_callback(crash) + _admit(store, ctx, 3) + raise AssertionError("CRASH_CHECKPOINT_NOT_REACHED") + + +def _command(mode, path, identity, token=""): + root = Path(__file__).resolve().parents[1] + code = ("import sys,json,runpy; " + "sys.addaudithook(lambda e,a: (_ for _ in ()).throw(RuntimeError('OFFLINE_SOCKET_DENIED')) if e.startswith('socket.') else None); " + "sys.path.insert(0,sys.argv[1]); m=runpy.run_path(sys.argv[2]); " + "print(json.dumps(m['_worker'](sys.argv[3],m['Path'](sys.argv[4]),json.loads(sys.argv[5]),sys.argv[6]),sort_keys=True))") + return [sys.executable, "-I", "-B", "-c", code, str(root), str(Path(__file__).resolve()), + mode, str(path), json.dumps(identity, separators=(",", ":")), token] + + +def _run_child(mode, path, identity): + return subprocess.run(_command(mode, path, identity), capture_output=True, text=True, + encoding="utf-8", timeout=20, check=False) + + +def test_two_real_processes_recover_receipt_without_sidecar(tmp_path): + path = tmp_path / "process-memory.db" + identity = _data()["identity"] + first = _run_child("write", path, identity) + assert first.returncode == 0, first.stderr + second = _run_child("read", path, identity) + assert second.returncode == 0, second.stderr + a, b = json.loads(first.stdout), json.loads(second.stdout) + assert a["pid"] != b["pid"] and a["pid"] != os.getpid() and b["pid"] != os.getpid() + for key in ("publication_id", "receipt_hash", "prompt_hash"): + assert a[key] == b[key] + assert a["session_hash"] != b["session_hash"] + assert a["model_calls"] == b["model_calls"] == 1 + assert not list(tmp_path.glob("*.json")) + + +def test_two_process_writers_have_exactly_one_winner(ready): + path, ctx, rid = ready + workers = [subprocess.Popen(_command("race", path, ctx["identity"], token), stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, encoding="utf-8") for token in ("A", "B")] + results = [] + try: + for proc in workers: + stdout, stderr = proc.communicate(timeout=20) + assert proc.returncode == 0, stderr + results.append(json.loads(stdout)) + finally: + for proc in workers: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + assert sorted(r["status"] for r in results) == ["HOLD", "PASS"], results + winner = next(r for r in results if r["status"] == "PASS") + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + assert reader.load_current().publication_id == winner["publication_id"] + assert reader._con.execute("SELECT COUNT(*) FROM person_admission_publications").fetchone()[0] == 1 + + +@pytest.mark.parametrize("checkpoint", ["crash-after-pointer", "crash-precommit"]) +def test_actual_process_exit_before_commit_preserves_last_known_good(ready, checkpoint): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + old = store.load_current() + crashed = _run_child(checkpoint, path, ctx["identity"]) + assert crashed.returncode == 73, (crashed.stdout, crashed.stderr) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as reader: + assert reader.load_current() == old + assert reader._con.execute("PRAGMA foreign_key_check").fetchall() == [] + + +def test_deferred_fk_refuses_pointer_without_receipt(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + store._con.execute("BEGIN IMMEDIATE") + store._con.execute("INSERT INTO person_admission_current VALUES(?,?,?,?)", (store.twin_key, "missing", b'{}', "bad")) + with pytest.raises(sqlite3.IntegrityError): + store._con.execute("COMMIT") + store._con.execute("ROLLBACK") + assert store.load_current() is None + + +@pytest.mark.parametrize("memory_id", [True, 0, -1, "3"]) +def test_malformed_memory_identity_cannot_stage(ready, memory_id): + path, ctx, _ = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + value = _admit(store, ctx, memory_id) + assert value["status"] == "HOLD" and value["possible_persisted_effect"] is False + assert store.load_current() is None + + +@pytest.mark.parametrize("mode", ["missing-ledger", "extra-field", "bad-time", "expired"]) +def test_fresh_input_shape_or_expiry_cannot_stage(ready, mode): + path, ctx, rid = ready + def bad(): + value = _fresh(ctx) + if mode == "missing-ledger": + del value["current_revocation_ledger"] + elif mode == "extra-field": + value["authorization_granted"] = True + elif mode == "bad-time": + value["at"] = "not-a-timestamp" + else: + value["at"] = "2026-08-27T12:50:00Z" + return value + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + value = _admit(store, ctx, rid, bad) + assert value["status"] == "HOLD" + assert store.load_current() is None + + +def test_failed_postverify_retains_only_noncurrent_candidate(ready): + path, ctx, rid = ready + calls = 0 + def fail_third(): + nonlocal calls + calls += 1 + if calls == 3: + raise OSError("SYNTHETIC_PROVIDER_UNAVAILABLE") + return _fresh(ctx) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + result = _admit(store, ctx, rid, fail_third) + assert result["status"] == "HOLD" and result["candidate_persisted"] is True + assert result["commit_state"] == "NOT_ATTEMPTED" + assert store.read_candidate_bytes(ctx["person_record"]["id"]) == durable._canonical(ctx["person_record"]) + assert store.load_current() is None + + +def test_next_promotion_preserves_exact_receipt_predecessor(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + assert _admit(store, ctx, rid)["status"] == "PASS" + first = store.load_current() + assert _admit(store, ctx, rid)["status"] == "PASS" + second = store.load_current() + assert first.publication_id != second.publication_id + assert first.candidate_bytes == second.candidate_bytes + assert second.receipt["expected_pointer_sha256"] == durable._hash(first.pointer_bytes) + assert second.receipt_bytes != first.receipt_bytes + assert store._con.execute("SELECT COUNT(*) FROM person_admission_publications").fetchone()[0] == 2 + + +def test_reentrant_admit_does_not_replace_the_outer_transaction(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + calls = 0 + def provider(): + nonlocal calls + calls += 1 + with pytest.raises(durable.DurableAdmissionError): + _admit(store, ctx, rid) + return _fresh(ctx) + result = _admit(store, ctx, rid, provider) + assert result["status"] == "PASS" and calls == 4 + assert store._phase is None and store._active is None + assert store._con.execute("SELECT COUNT(*) FROM person_admission_publications").fetchone()[0] == 1 + + +def test_lock_contention_holds_without_automatic_replay(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False, timeout=0.05) as store: + lock = sqlite3.connect(path, isolation_level=None) + try: + lock.execute("BEGIN IMMEDIATE") + outcome = _admit(store, ctx, rid) + assert outcome["status"] == "HOLD" + finally: + lock.execute("ROLLBACK") + lock.close() + assert store.load_current() is None + assert store._con.execute("SELECT COUNT(*) FROM person_admission_candidates").fetchone()[0] == 0 + + +def test_changed_foreign_key_setting_cannot_publish(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + store._con.execute("PRAGMA foreign_keys=OFF") + result = _admit(store, ctx, rid) + assert result["status"] == "HOLD" and result["possible_persisted_effect"] is False + store._con.execute("PRAGMA foreign_keys=ON") + assert store.load_current() is None + + +def test_unverified_rollback_suspends_connection_and_keeps_hold(ready): + path, ctx, rid = ready + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"], read_only=False) as store: + class Fault: + def __init__(self, real): + self.real = real + def __getattr__(self, name): + return getattr(self.real, name) + def execute(self, sql, args=()): + if sql == "ROLLBACK": + raise OSError("SYNTHETIC_ROLLBACK_ACK_LOST") + return self.real.execute(sql, args) + store._con = Fault(store._con) + calls = 0 + def fail_promotion(): + nonlocal calls + calls += 1 + if calls == 4: + raise OSError("SYNTHETIC_PROVIDER_FAILURE") + return _fresh(ctx) + result = _admit(store, ctx, rid, fail_promotion) + assert result["status"] == "HOLD" and result["reason"] == "ROLLBACK_UNVERIFIED" + assert result["possible_persisted_effect"] is True and result["admission_receipt"] is None + with pytest.raises(durable.DurableAdmissionError): + store.load_current() + with pytest.raises(durable.DurableAdmissionError): + _admit(store, ctx, rid) + with durable.PersonTwinAdmissionStore(path, identity=ctx["identity"]) as fresh: + assert fresh.load_current() is None diff --git a/tests/test_product_person_continuity.py b/tests/test_product_person_continuity.py new file mode 100644 index 00000000..63ff3643 --- /dev/null +++ b/tests/test_product_person_continuity.py @@ -0,0 +1,758 @@ +"""Product proof of synthetic Person continuity across real Python processes. + +The Person contracts, admission readback, SQLite reads, and R21H prompt path are +real. Model clients only capture calls; synthetic host fixtures are not live OS +authority, cross-model inference, or a Portable Twin Vault restore. +""" +from __future__ import annotations + +import copy +import hashlib +import json +import os +from pathlib import Path +import secrets +import shutil +import sqlite3 +import subprocess +import sys + +import pytest + +from continuityos import current_entrypoints, db, demo +from continuityos import person_twin_demo as person_demo +from continuityos.person_twin_admission_contracts import ( + build_consent_revocation_entry, + build_consent_revocation_ledger, +) + + +def _fixture(tmp_path): + root = tmp_path / ("continuityos-person-demo-" + secrets.token_hex(16)) + root.mkdir() + written = person_demo._write_fixture(root, secrets.token_hex(16)) + return root, written + + +def _assert_hold(value, code): + assert code == 2 + assert value["terminal"].endswith("_HOLD") + + +def _controlled_root(tmp_path, monkeypatch): + root = tmp_path / ("continuityos-person-demo-" + secrets.token_hex(16)) + + def create_owned_root(*args, **kwargs): + root.mkdir() + return str(root) + + monkeypatch.setattr(person_demo.tempfile, "mkdtemp", create_owned_root) + return root + + +@pytest.fixture(scope="module") +def completed_person_demo(tmp_path_factory): + """One real PREPARE/A/B run; observe bytes without replacing either worker.""" + sandbox = tmp_path_factory.mktemp("person-continuity-proof") + user_db = sandbox / "user-memory.db" + user_db.write_bytes(b"synthetic-user-memory-sentinel") + user_before = user_db.read_bytes() + shadow = sandbox / "shadow-import-origin" + (shadow / "continuityos").mkdir(parents=True) + (shadow / "continuityos" / "__init__.py").write_text( + "raise RuntimeError('SYNTHETIC_WRONG_IMPORT_ORIGIN')\n", encoding="utf-8", + ) + observations = [] + commands = [] + original_command = person_demo._child_command + original_worker = person_demo._run_worker + + def observed_command(*arguments): + command = original_command(*arguments) + commands.append((arguments, list(command))) + if arguments[0] == "_worker": + # These real reader children must succeed with every write entry + # disabled: neither SQL admission nor R3 receipt regeneration is a fallback. + prelude = ("from continuityos.person_twin_admission_store import PersonTwinAdmissionStore as S\n" + "from continuityos.person_twin_admission_coordinator import PersonTwinAdmissionCoordinator as C\n" + "def forbidden(*a, **kw): raise AssertionError('READER_ADMISSION_WRITE_FORBIDDEN')\n" + "S.initialize = forbidden\nS.admit = forbidden\nC.promote_current = forbidden\n") + command[4] = command[4].replace("runpy.run_module(", "exec(" + repr(prelude) + "); runpy.run_module(", 1) + return command + + def observe_worker(root, expected_sha256, role, timeout, run_id): + assert {item.name for item in root.iterdir()} == {"memory.db"} + before = {"memory.db": (root / "memory.db").read_bytes()} + fixture = person_demo._read_fixture(root, expected_sha256, run_id) + con = sqlite3.connect((root / "memory.db").as_uri() + "?mode=ro&immutable=1", uri=True) + try: + rows = con.execute("SELECT id, namespace, text, meta FROM items ORDER BY id").fetchall() + finally: + con.close() + result = original_worker(root, expected_sha256, role, timeout, run_id) + after = {name: (root / name).read_bytes() for name in before} + observations.append({ + "role": role, "before": before, "after": after, + "fixture": fixture, "rows": rows, "result": result, + }) + return result + + with pytest.MonkeyPatch.context() as patch: + root = _controlled_root(sandbox, patch) + patch.setenv("CONTINUITYOS_DB", str(user_db)) + # Workers must use the already imported parent's package, even when a + # same-named package exists at both ambient import-origin entry points. + patch.chdir(shadow) + patch.setenv("PYTHONPATH", str(shadow)) + patch.setattr(person_demo, "_run_worker", observe_worker) + patch.setattr(person_demo, "_child_command", observed_command) + value, code = person_demo.run_person_continuity(timeout=20.0) + + assert code == 0, (value, [o['result'] for o in observations]) + assert len(commands) == 3 + for args, command in commands[1:]: + assert len(args) == 9 and args[0] == "_worker" + assert args[1] == "--root" and args[3] == "--db-sha256" + assert args[5] == "--role" and args[7] == "--run-id" + assert args[8] == value["run_id"] + assert not any("fixture.json" in item or '"admission_receipt"' in item for item in command[7:]) + assert not root.exists() + assert user_db.read_bytes() == user_before + return value, observations + + +def test_person_continuity_recovers_actual_current_evidence_across_processes(completed_person_demo): + value, observations = completed_person_demo + assert value["terminal"] == "COS_DEMO_PERSON_CONTINUITY_PASS" + assert value["orchestrator_process_id"] == os.getpid() + assert [entry["role"] for entry in observations] == ["A", "B"] + assert len(value["processes"]) == value["effects"]["child_launches"] == 3 + process_ids = [process["process_id"] for process in value["processes"]] + assert len(set(process_ids)) == 3 + assert os.getpid() not in process_ids + assert value["prepare"]["process_id"] == process_ids[0] + assert all(process["returncode"] == 0 and process["direct_child_reaped"] is True + for process in value["processes"]) + source_sha = hashlib.sha256(Path(person_demo.__file__).read_bytes()).hexdigest() + assert value["source_sha256"] == value["prepare"]["source_sha256"] == source_sha + + first, second = observations + assert first["before"] == first["after"] == second["before"] == second["after"] + fixture = first["fixture"] + memory_id = fixture["memory"]["id"] + current_row = next(row for row in first["rows"] if row[0] == memory_id) + assert current_row[1] == fixture["memory"]["namespace"] + assert current_row[2] == fixture["context"]["source_record"]["payload"]["text"] + old_rows = [row for row in first["rows"] + if json.loads(row[3]).get("superseded_by") == memory_id] + assert old_rows, "The persisted history must include a superseded decision" + expected = [{"ref": f"mem:{memory_id}", "namespace": current_row[1], "text": current_row[2]}] + a, b = (value["sessions"][role] for role in ("A", "B")) + assert a["captured_evidence"] == b["captured_evidence"] == expected + assert all(row[2] not in item["text"] for row in old_rows for item in b["captured_evidence"]) + assert a["process_id"] == process_ids[1] and b["process_id"] == process_ids[2] + assert a["transport"] == "warm_chat" and b["transport"] == "cold_streaming" + assert a["model_calls"] == b["model_calls"] == 1 + assert a["model_events"].count("chat") == 1 and "stream" not in a["model_events"] + assert b["model_events"].count("stream") == 1 and "chat" not in b["model_events"] + assert a["model_profile"] != b["model_profile"] + assert a["session_sha256"] != b["session_sha256"] + assert a["request_sha256"] != b["request_sha256"] + assert a["packet_sha256"] != b["packet_sha256"] + for key in ("intent_sha256", "r4_receipt_hash", "receipt_hash"): + assert a["person_evidence"][key] != b["person_evidence"][key] + assert a["person_evidence"]["evidence_sha256"] == b["person_evidence"]["evidence_sha256"] + for worker in (a, b): + assert worker["source_sha256"] == source_sha + assert worker["context_reads"] >= 2 + assert worker["person_evidence"]["prompt_boundary_revalidated"] is True + assert worker["checks"]["query_only"] is True + assert worker["unsupported_execute"]["decision"] == "DENY" + assert worker["unsupported_execute"]["execution_authorized"] is False + assert worker["socket_audit"] == { + "installed": True, "events": 0, "policy": "DENY_SOCKET_EVENTS", + } + + for member, key in (("memory.db", "db"),): + digest = hashlib.sha256(first["before"][member]).hexdigest() + assert value["snapshot"][key + "_sha256_before"] == digest + assert value["snapshot"][key + "_sha256_after"] == digest + assert value["schema"] == "continuityos.product_demo_person_continuity/v2" + assert value["work_order"] == "COS_PERSON_CONTINUITY_DURABLE_DEMO_R2" + assert value["snapshot"]["admission_source"] == "MEMORY_SQLITE" + assert value["snapshot"]["admission_sidecar_count"] == 0 + saved = fixture["context"]["admission_receipt"] + saved_hash = hashlib.sha256(person_demo._canonical(saved)).hexdigest() + assert a["stored_receipt_sha256"] == b["stored_receipt_sha256"] == saved_hash + assert a["publication_id"] == b["publication_id"] == fixture["publication_id"] + assert value["prepare"]["stored_receipt_sha256"] == saved_hash + assert value["snapshot"]["stored_receipt_sha256"] == saved_hash + assert "admission_receipt" not in value["prepare"] + assert "admission_bytes" not in value["prepare"] and "context" not in value["prepare"] + assert all(worker["checks"]["durable_receipt_recovered"] is True for worker in (a,b)) + for kind in ("candidate", "pointer"): + persisted = fixture["admission_bytes"][kind + "_utf8"].encode("utf-8") + assert value["snapshot"][kind + "_sha256"] == hashlib.sha256(persisted).hexdigest() + assert value["temporary_path_removed"] is True + assert value["effects"]["temporary_cleanup_pass"] is True + for effect in ("user_memory_read", "user_memory_write", "network_effect", "external_model_call"): + assert value["effects"][effect] is False + assert value["socket_audit"]["scope"] == "fixture_and_model_children" + assert value["production_admission_status"] == "NOT_PRODUCTION_ADMITTED" + assert value["execution_authority"] == "NONE" and value["can_execute"] is False + + +def test_person_demo_shows_recovered_work_context_and_routes_json( + completed_person_demo, monkeypatch, capsys, +): + value, observations = completed_person_demo + text = observations[1]["fixture"]["context"]["source_record"]["payload"]["text"] + rendered = person_demo.render_person_continuity(value) + for key, label in (("goal", "Goal"), ("decision", "Decision"), + ("constraint", "Constraint"), ("next_action", "Next step")): + persisted_value = next(line.split(": ", 1)[1] for line in text.splitlines() + if line.startswith(label + ": ")) + assert value["recovered"][key] == persisted_value + assert persisted_value in rendered + assert value["snapshot"]["memory_ref"] in rendered + assert "External AI NOT USED" in rendered + assert "NOT_PRODUCTION_ADMITTED" in rendered + seen = [] + + def reuse_observed_result(*, timeout): + seen.append(timeout) + return copy.deepcopy(value), 0 + + monkeypatch.setattr(person_demo, "run_person_continuity", reuse_observed_result) + monkeypatch.setattr(current_entrypoints, "current_binding_from_env", lambda env: (None, [])) + code = current_entrypoints.cos_main(["demo", "person-continuity", "--json", "--timeout", "7"]) + + assert code == 0 and seen == [7.0] + assert json.loads(capsys.readouterr().out) == value + + +@pytest.mark.parametrize("placement", ["leading", "trailing"]) +def test_person_demo_rejects_user_db_before_any_demo_io( + tmp_path, monkeypatch, capsys, placement, +): + user_db = tmp_path / "user-memory.db" + user_db.write_bytes(b"synthetic-user-memory-sentinel") + before = user_db.read_bytes() + + def forbidden(*args, **kwargs): + raise AssertionError("A public --db argument must stop before demo or DB resolution") + + monkeypatch.setattr(person_demo, "run_person_continuity", forbidden) + monkeypatch.setattr(db, "resolve_memory_db", forbidden) + monkeypatch.setattr(current_entrypoints, "current_binding_from_env", lambda env: (None, [])) + args = ["demo", "person-continuity", "--json"] + if placement == "leading": + args = ["--db", str(user_db), *args] + else: + args.extend(["--db", str(user_db)]) + + code = current_entrypoints.cos_main(args) + value = json.loads(capsys.readouterr().out) + + _assert_hold(value, code) + assert value["reason"] == "USER_DB_ARGUMENT_NOT_ALLOWED" + assert value["effects"]["ephemeral_filesystem_write"] is False + assert value["effects"]["ephemeral_memory_write"] is False + assert value["effects"]["subprocess_execution"] is False + assert user_db.read_bytes() == before + + +def test_person_demo_stays_inside_current_entrypoint_containment(monkeypatch): + def forbidden(*args, **kwargs): + raise AssertionError("A current-bound session must not enter the product demo") + + monkeypatch.setattr(person_demo, "run_person_continuity", forbidden) + monkeypatch.setattr( + current_entrypoints, + "current_binding_from_env", + lambda env: ({"challenge": "x", "ack": "y", "challenge_sha256": "z"}, []), + ) + monkeypatch.setattr( + current_entrypoints, + "_verify_binding", + lambda binding, surface, command: ( + { + "binding_verified": True, + "authority_generation": "R64", + "challenge_id": "synthetic-demo-containment", + "challenge_sha256": "a" * 64, + }, + None, + ), + ) + + assert current_entrypoints.cos_main(["demo", "person-continuity", "--json"]) == 3 + + +@pytest.mark.parametrize("surface", ["runner", "worker", "private-cli"]) +def test_person_demo_direct_surfaces_reject_declared_current_binding( + tmp_path, monkeypatch, capsys, surface, +): + calls = [] + + def forbidden(*args, **kwargs): + calls.append("forbidden-io") + raise AssertionError("Current binding must stop before fixture access or subprocess work") + + monkeypatch.setenv("CONTINUITYOS_CURRENT_SESSION_REQUIRED", "1") + monkeypatch.setattr(person_demo.tempfile, "mkdtemp", forbidden) + monkeypatch.setattr(person_demo, "_run_process", forbidden) + monkeypatch.setattr(person_demo, "_read_fixture", forbidden) + monkeypatch.setattr(person_demo, "_install_child_socket_guard", forbidden) + if surface == "runner": + value, code = person_demo.run_person_continuity() + elif surface == "worker": + value, code = person_demo._worker( + tmp_path / "continuityos-person-demo-unopened", "a" * 64, "B", + ) + else: + code = person_demo.main([ + "_worker", "--root", str(tmp_path / "continuityos-person-demo-unopened"), + "--db-sha256", "a" * 64, "--role", "B", + ]) + value = json.loads(capsys.readouterr().out) + + _assert_hold(value, code) + assert "CURRENT" in value["reason"] + assert not calls + + +@pytest.mark.parametrize("drift", ["stale-session", "revoked-consent", "stale-request"]) +@pytest.mark.parametrize("boundary", [1, 2]) +def test_person_demo_fresh_host_drift_holds_before_model(tmp_path, monkeypatch, drift, boundary): + root, written = _fixture(tmp_path) + original_read = person_demo._FixtureProvider.read_context + seen = [] + + def drifted_read(provider): + context = original_read(provider) + seen.append(drift) + if provider.reads < boundary: + return context + if drift == "stale-session": + context["current_session"] = copy.deepcopy(context["current_session"]) + context["current_session"]["challenge_sha256"] = "0" * 64 + elif drift == "stale-request": + context["authorization_request"]["request_id"] = "stale-request" + else: + identity = context["identity"] + consent = context["consent_receipt"] + revocation = build_consent_revocation_entry( + identity, consent, + revoking_principal=identity["controller_id"], + revoking_principal_kind=identity["controller_kind"], + revoked_at=consent["issued_at"], + reason="synthetic current consent revoked before Person READ", + ) + context["current_revocation_ledger"] = build_consent_revocation_ledger([revocation]) + return context + + monkeypatch.setattr(person_demo._FixtureProvider, "read_context", drifted_read) + value, code = person_demo._worker(root, written["db_sha256"], "B", written["run_id"]) + + _assert_hold(value, code) + assert value["model_calls"] == 0 + assert seen + + +@pytest.mark.parametrize("case", ["tampered-db", "injected-sidecar", "missing-db"]) +def test_person_demo_changed_persisted_bytes_hold_without_reseed(tmp_path, monkeypatch, case): + root, written = _fixture(tmp_path) + member = "fixture.json" if case == "injected-sidecar" else "memory.db" + path = root / member + writer_calls = [] + + def forbidden_writer(*args, **kwargs): + writer_calls.append("reseed") + raise AssertionError("B must consume persisted bytes without invoking a fixture writer") + + monkeypatch.setattr(person_demo, "_write_fixture", forbidden_writer) + if case == "missing-db": + path.unlink() + changed = None + elif case == "injected-sidecar": + changed = b'{"admission_receipt":{"state":"CURRENT"}}' + path.write_bytes(changed) + else: + # Keep the SQLite header intact and JSON parseable; the persisted hash + # binding must reject changed bytes before model or mutable DB work. + changed = path.read_bytes() + (b" " if member.endswith("json") else b"\x00") + path.write_bytes(changed) + + value, code = person_demo._worker(root, written["db_sha256"], "B", written["run_id"]) + + _assert_hold(value, code) + assert value["model_calls"] == 0 + assert not writer_calls + if changed is None: + assert not path.exists() + else: + assert path.read_bytes() == changed + + +def test_person_demo_worker_uses_only_the_explicit_read_only_fixture(tmp_path, monkeypatch): + from continuityos import sovereign_twin_runtime + + root, written = _fixture(tmp_path) + user_db = tmp_path / "user-memory.db" + user_db.write_bytes(b"synthetic-normal-db-sentinel") + before = (root / "memory.db").read_bytes() + opened = [] + original_runtime = sovereign_twin_runtime.SovereignTwinRuntime + + def observe_runtime(*args, **kwargs): + runtime = original_runtime(*args, **kwargs) + connection = runtime.memory.store.con + opened.append((connection.execute("PRAGMA database_list").fetchone()[2], + connection.execute("PRAGMA query_only").fetchone()[0])) + return runtime + + def forbidden_resolver(*args, **kwargs): + raise AssertionError("The fixed synthetic worker must not resolve a normal DB") + + monkeypatch.setenv("CONTINUITYOS_DB", str(user_db)) + monkeypatch.setattr(db, "resolve_memory_db", forbidden_resolver) + monkeypatch.setattr(sovereign_twin_runtime, "SovereignTwinRuntime", observe_runtime) + + value, code = person_demo._worker(root, written["db_sha256"], "B", written["run_id"]) + + assert code == 0, value + assert value["model_calls"] == 1 and value["transport"] == "cold_streaming" + assert len(opened) == 1 + assert Path(opened[0][0]).resolve() == (root / "memory.db").resolve() + assert opened[0][1] == 1 + assert (root / "memory.db").read_bytes() == before + assert user_db.read_bytes() == b"synthetic-normal-db-sentinel" + + +def test_person_demo_rejects_linked_db_before_following_it(tmp_path): + root, written = _fixture(tmp_path) + path = root / "memory.db" + outside = tmp_path / "outside-fixture-memory.db" + before = path.read_bytes() + outside.write_bytes(before) + path.unlink() + try: + path.symlink_to(outside) + except (NotImplementedError, OSError) as exc: + pytest.skip(f"Creating a real filesystem symlink is unavailable: {type(exc).__name__}") + + value, code = person_demo._worker(root, written["db_sha256"], "B", written["run_id"]) + + _assert_hold(value, code) + assert value["reason"] == "FIXTURE_LINK_FORBIDDEN" + assert value["model_calls"] == 0 + assert outside.read_bytes() == before + + +def test_person_demo_rejects_parent_traversal_even_to_valid_fixture(tmp_path): + root, written = _fixture(tmp_path) + redundant = tmp_path / "redundant-component" + redundant.mkdir() + traversal = redundant / ".." / root.name + + value, code = person_demo._worker(traversal, written["db_sha256"], "B", written["run_id"]) + + _assert_hold(value, code) + assert value["reason"] == "FIXTURE_ROOT_INVALID" + assert value["model_calls"] == 0 + + +@pytest.mark.parametrize("mutation", ["subset-checks", "corrupt-inner-hash", "stored-receipt", "publication", "sidecar", "source", "bool-calls"]) +def test_person_demo_sealed_outer_receipt_does_not_replace_worker_validation( + completed_person_demo, mutation, +): + _, observations = completed_person_demo + observed = observations[0] + result = copy.deepcopy(observed["result"]) + fixture = observed["fixture"] + fixture_sha = hashlib.sha256(observed["before"]["memory.db"]).hexdigest() + assert person_demo._valid_worker(result, "A", fixture, fixture_sha) is True + receipt = result["receipt"] + if mutation == "subset-checks": + receipt["checks"] = {"one_generation": True} + elif mutation == "corrupt-inner-hash": + receipt["person_evidence"]["receipt_hash"] = "0" * 64 + elif mutation == "stored-receipt": + receipt["stored_receipt_sha256"] = "0" * 64 + elif mutation == "publication": + receipt["publication_id"] = "0" * 32 + elif mutation == "sidecar": + receipt["admission_sidecar_count"] = 1 + elif mutation == "source": + receipt["admission_source"] = "SIDECAR" + else: + receipt["model_calls"] = True + receipt.pop("receipt_sha256") + result["receipt"] = person_demo._seal(receipt) + + assert person_demo._valid_worker(result, "A", fixture, fixture_sha) is False + + +@pytest.mark.parametrize("failure", ["timeout", "output-limit"]) +def test_person_demo_process_limits_reap_the_actual_direct_child(monkeypatch, failure): + created = [] + original_popen = subprocess.Popen + + def record_child(*args, **kwargs): + child = original_popen(*args, **kwargs) + created.append(child) + return child + + monkeypatch.setattr(person_demo.subprocess, "Popen", record_child) + if failure == "timeout": + command = [sys.executable, "-c", "import time; time.sleep(10)"] + timeout = 0.1 + else: + command = [sys.executable, "-c", + "import sys; sys.stdout.buffer.write(b'x' * 131072); sys.stdout.flush()"] + timeout = 5.0 + try: + result = person_demo._run_process(command, timeout) + assert len(created) == 1 + assert created[0].poll() is not None + assert result["process_id"] == created[0].pid + assert result["started"] is True and result["direct_child_reaped"] is True + assert len(result["stdout"]) <= person_demo.MAX_OUTPUT_BYTES + assert len(result["stderr"]) <= person_demo.MAX_OUTPUT_BYTES + if failure == "timeout": + assert result["timed_out"] is True and result["kill_attempted"] is True + assert result["returncode"] != 0 + else: + assert result["output_limit_exceeded"] is True + finally: + # A failing reap assertion must not leave the test's owned child alive. + for child in created: + if child.poll() is None: + child.kill() + child.wait(timeout=3.0) + + +@pytest.mark.parametrize("failure", ["timeout", "output-limit", "unreaped"]) +def test_person_demo_failed_process_cannot_reuse_a_valid_allow_receipt( + completed_person_demo, monkeypatch, failure, +): + value, _ = completed_person_demo + raw = json.dumps(value["sessions"]["A"], ensure_ascii=False, + sort_keys=True, separators=(",", ":")).encode("utf-8") + result = { + "started": True, "returncode": 0, "process_id": value["sessions"]["A"]["process_id"], + "direct_child_reaped": failure != "unreaped", "timed_out": failure == "timeout", + "output_limit_exceeded": failure == "output-limit", "stdout": raw, "stderr": b"", + } + monkeypatch.setattr(person_demo, "_run_process", lambda command, timeout: dict(result)) + + observed = person_demo._child_result(["synthetic-never-executed"], 1.0) + + assert observed["ok"] is False + + +@pytest.mark.parametrize("raw", [ + b"{not-json", + b'{"number":1e999,"receipt_sha256":"' + b"a" * 64 + b'"}', +]) +def test_person_demo_malformed_child_json_is_a_failed_result(monkeypatch, raw): + monkeypatch.setattr(person_demo, "_run_process", lambda command, timeout: { + "started": True, "returncode": 0, "direct_child_reaped": True, + "timed_out": False, "output_limit_exceeded": False, + "stdout": raw, "stderr": b"", + }) + + observed = person_demo._child_result(["synthetic-never-executed"], 1.0) + + assert observed["ok"] is False + + +def test_person_demo_invalid_worker_receipt_stops_before_b_and_cleans(tmp_path, monkeypatch): + root = _controlled_root(tmp_path, monkeypatch) + roles = [] + + def incomplete_worker(root, expected_sha256, role, timeout, run_id): + roles.append(role) + return { + "ok": True, "started": True, "returncode": 0, "direct_child_reaped": True, + "receipt": {"schema": person_demo.WORKER_SCHEMA, "terminal": "PERSON_DEMO_WORKER_PASS"}, + } + + monkeypatch.setattr(person_demo, "_run_worker", incomplete_worker) + value, code = person_demo._controller(timeout=20.0) + + _assert_hold(value, code) + assert value["reason"] == "WORKER_RESULT_INVALID" + assert roles == ["A"] + assert value["temporary_path_removed"] is True + assert not root.exists() + + +def test_person_demo_cleanup_failure_cannot_return_pass(tmp_path, monkeypatch): + root = _controlled_root(tmp_path, monkeypatch) + original_rmtree = shutil.rmtree + + def fail_cleanup(*args, **kwargs): + raise OSError("synthetic cleanup failure") + + monkeypatch.setattr(person_demo.shutil, "rmtree", fail_cleanup) + try: + value, code = person_demo._controller(timeout=20.0) + + _assert_hold(value, code) + assert value["reason"] == "TEMPORARY_CLEANUP_FAILED" + assert value["checks"] and all(value["checks"].values()) + assert value["effects"]["child_launches"] == 3 + assert value["effects"]["temporary_cleanup_pass"] is False + assert value["temporary_path_removed"] is False + assert root.exists() + finally: + if root.exists(): + original_rmtree(root) + + +@pytest.mark.parametrize("damage", ["missing-schema", "wrong-version", "missing-current", "missing-publication", + "missing-candidate", "corrupt-receipt", "rehashed-receipt", "pointer-hash", "text", "namespace", "superseded"]) +def test_durable_demo_rejects_structural_damage_even_with_recomputed_db_digest(tmp_path, monkeypatch, damage): + from continuityos.person_twin_admission_store import PersonTwinAdmissionStore + from continuityos.person_twin_admission_coordinator import PersonTwinAdmissionCoordinator + root, written = _fixture(tmp_path) + path = root / "memory.db" + with sqlite3.connect(path) as con: + if damage == "missing-schema": + con.execute("DROP TABLE person_admission_meta") + elif damage == "wrong-version": + con.execute("UPDATE person_admission_meta SET schema_id='unknown'") + elif damage == "missing-current": + con.execute("DELETE FROM person_admission_current") + elif damage == "missing-publication": + con.execute("DELETE FROM person_admission_publications") + elif damage == "missing-candidate": + con.execute("DELETE FROM person_admission_candidates") + elif damage == "corrupt-receipt": + con.execute("UPDATE person_admission_publications SET receipt=?", (b"{}",)) + elif damage == "rehashed-receipt": + con.execute("UPDATE person_admission_publications SET receipt=?,receipt_hash=?", (b"{}", hashlib.sha256(b"{}").hexdigest())) + elif damage == "pointer-hash": + con.execute("UPDATE person_admission_current SET pointer_hash=?", ("0" * 64,)) + elif damage == "text": + con.execute("UPDATE items SET text='ALTERED_CURRENT_TEXT' WHERE id=?", (written["memory_id"],)) + elif damage == "namespace": + con.execute("UPDATE items SET namespace='foreign-person' WHERE id=?", (written["memory_id"],)) + else: + row = con.execute("SELECT meta FROM items WHERE id=?", (written["memory_id"],)).fetchone() + meta = json.loads(row[0]); meta["superseded_by"] = 99999 + con.execute("UPDATE items SET meta=? WHERE id=?", (json.dumps(meta), written["memory_id"])) + con.close() + before = path.read_bytes() + calls = [] + def forbidden(*args, **kwargs): + calls.append("repair") + raise AssertionError("Read must not repair or regenerate admission") + monkeypatch.setattr(PersonTwinAdmissionStore, "initialize", forbidden) + monkeypatch.setattr(PersonTwinAdmissionStore, "admit", forbidden) + monkeypatch.setattr(PersonTwinAdmissionCoordinator, "promote_current", forbidden) + monkeypatch.setattr(person_demo, "_write_fixture", forbidden) + value, code = person_demo._worker(root, hashlib.sha256(before).hexdigest(), "B", written["run_id"]) + _assert_hold(value, code) + assert value["model_calls"] == 0 and not calls + assert path.read_bytes() == before + assert {item.name for item in root.iterdir()} == {"memory.db"} + + +def test_durable_demo_expected_identity_is_not_selected_from_database(tmp_path): + root, written = _fixture(tmp_path) + before = (root / "memory.db").read_bytes() + value, code = person_demo._worker(root, written["db_sha256"], "B", "f" * 32) + _assert_hold(value, code) + assert value["model_calls"] == 0 + assert (root / "memory.db").read_bytes() == before + + +@pytest.mark.parametrize("damage", ["full-receipt", "extra-context", "bool-memory-id", "bool-sidecars"]) +def test_durable_demo_prepare_shape_cannot_transport_admission_payload(completed_person_demo, tmp_path, monkeypatch, damage): + value, observations = completed_person_demo + prepared = {**value["processes"][0], "receipt": copy.deepcopy(value["prepare"]), "ok": True} + assert person_demo._valid_preparation(prepared, value["run_id"]) is True + def invalid_prepare(root, run_id, timeout): + result = copy.deepcopy(prepared) + receipt = result["receipt"] + receipt["run_id"] = run_id + if damage == "full-receipt": + receipt["admission_receipt"] = observations[0]["fixture"]["context"]["admission_receipt"] + elif damage == "extra-context": + receipt["context"] = observations[0]["fixture"]["context"] + elif damage == "bool-memory-id": + receipt["memory_id"] = True + else: + receipt["admission_sidecar_count"] = False + receipt.pop("receipt_sha256") + result["receipt"] = person_demo._seal(receipt) + return result + def forbidden_worker(*args, **kwargs): + raise AssertionError("Invalid PREPARE cannot reach a model worker") + root = _controlled_root(tmp_path, monkeypatch) + monkeypatch.setattr(person_demo, "_run_prepare", invalid_prepare) + monkeypatch.setattr(person_demo, "_run_worker", forbidden_worker) + outcome, code = person_demo.run_person_continuity() + _assert_hold(outcome, code) + assert outcome["reason"] == "FIXTURE_PREPARATION_RECEIPT_INVALID" + assert not root.exists() + + +def test_durable_demo_db_tampering_between_real_workers_stops_b(tmp_path, monkeypatch): + root = _controlled_root(tmp_path, monkeypatch) + original = person_demo._run_worker + observations = [] + def change_before_b(root, digest, role, timeout, run_id): + if role == "B": + with sqlite3.connect(root / "memory.db") as con: + con.execute("UPDATE person_admission_publications SET receipt=?", (b"{}",)) + con.close() + result = original(root, digest, role, timeout, run_id) + observations.append(result) + return result + monkeypatch.setattr(person_demo, "_run_worker", change_before_b) + value, code = person_demo.run_person_continuity() + _assert_hold(value, code) + assert value["reason"] == "WORKER_RESULT_INVALID" + assert len(observations) == 2 + assert observations[0]["receipt"]["model_calls"] == 1 + assert observations[1]["receipt"]["model_calls"] == 0 + assert observations[1]["direct_child_reaped"] is True + assert not root.exists() + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows venv redirector only") +def test_venv_child_pid_is_exactly_the_reaped_direct_python_process(monkeypatch): + base = getattr(sys, "_base_executable", "") + assert isinstance(base, str) and Path(base).is_file() + with monkeypatch.context() as m: + m.setattr(person_demo.sys, "prefix", "synthetic-venv") + m.setattr(person_demo.sys, "base_prefix", "synthetic-base") + command = person_demo._child_command("_prepare", "--root", "unused", "--run-id", "0" * 32) + assert command[0] == base and command[1:4] == ["-I", "-B", "-c"] + observation = person_demo._run_process( + [base, "-I", "-B", "-c", "import os; print(os.getpid())"], 15.0) + assert observation["returncode"] == 0 + assert observation["direct_child_reaped"] is True + assert observation["stdout"].decode("ascii").strip() == str(observation["process_id"]) + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows venv redirector only") +def test_venv_missing_direct_interpreter_fails_closed(monkeypatch): + with monkeypatch.context() as m: + m.setattr(person_demo.sys, "prefix", "synthetic-venv") + m.setattr(person_demo.sys, "base_prefix", "synthetic-base") + m.setattr(person_demo.sys, "_base_executable", "") + with pytest.raises(person_demo.PersonDemoHold, match="PYTHON_BASE_EXECUTABLE_UNAVAILABLE"): + person_demo._child_command("_prepare", "--root", "unused", "--run-id", "0" * 32) + + +def test_direct_nonvenv_launch_keeps_original_python_executable(monkeypatch): + with monkeypatch.context() as m: + m.setattr(person_demo.sys, "base_prefix", sys.prefix) + command = person_demo._child_command("_prepare", "--root", "unused", "--run-id", "0" * 32) + assert command[0] == sys.executable + assert command[1:4] == ["-I", "-B", "-c"]