From 75039e96e110fb9f008b613afe053817952d1ebb Mon Sep 17 00:00:00 2001 From: bitmaster162 <115934939+bitmaster162@users.noreply.github.com> Date: Mon, 12 Oct 2026 12:15:14 -0700 Subject: [PATCH 1/4] checkpoint: compose exact PR150 R4 source on current master Import the unchanged R4 module and original tests from PR #150 head 4515bea19978b522328607deb493a55234ef7d4c onto master 719ed32a64301fc8083a5aad0c4ed7bed316040e. Tree 0a7e76d29612f6a10102e9dd57e0a420a63cb686 matches reviewed local composition checkpoint 0a08433c993017d877af95e6d679da4c964bd379. Composition only; no production admission or merge of PR #150. --- continuityos/person_twin_authority_replay.py | 709 +++++++++++++++++++ tests/test_person_twin_authority_replay.py | 683 ++++++++++++++++++ 2 files changed, 1392 insertions(+) create mode 100644 continuityos/person_twin_authority_replay.py create mode 100644 tests/test_person_twin_authority_replay.py diff --git a/continuityos/person_twin_authority_replay.py b/continuityos/person_twin_authority_replay.py new file mode 100644 index 00000000..f87704d2 --- /dev/null +++ b/continuityos/person_twin_authority_replay.py @@ -0,0 +1,709 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any, Mapping, Protocol + +from .company_twin_policy import evaluate as evaluate_policy, validate_policy +from .current_effect_boundary import MODE_CURRENT, SCHEMA as CURRENT_EFFECT_SCHEMA +from .person_twin_admission_contracts import ( + NOT_PRODUCTION_ADMITTED, + evaluate_consent, + validate_consent_revocation_ledger, + validate_person_twin_identity, + validate_source_consent_receipt, +) +from .person_twin_privacy_provenance import ( + validate_person_twin_record, + validate_privacy_scope, +) + +SCHEMA_VERSION = "continuityos.person-twin.authority-replay/v2" +ADMISSION_SCHEMA = "continuityos.person-twin.admission-coordinator/v2" +CURRENT_POINTER_SCHEMA = "continuityos.person-twin.current-pointer/v1" +AUTH_REQUEST_SCHEMA = "continuityos.operational_memory.apply_authorization_request/v1" +PREFLIGHT_SCHEMA = "continuityos.operational_memory.project_update_packet_preflight/v1" +SUPPORTED_ACTIONS = {"READ", "PROPOSE", "APPROVE"} +EXACT_CURRENT_AUTHORITY_CEILING = "NO_FURTHER_AGENT_WORK" +EXACT_CURRENT_REASON = "EXACT_CURRENT_SESSION_VERIFIED" + +PUBLIC_INPUT_KEYS = frozenset({ + "identity", "consent_receipt", "current_revocation_ledger", "source_record", + "person_record", "admission_receipt", "current_pointer", "admission_store", + "policy", "principal_id", "action", "requested_scope", + "requested_privacy_class", "purpose", "at", "current_session", + "authorization_request", "authorization_preflight", "replay_store", +}) +ADMISSION_KEYS = frozenset({ + "schema_version", "state", "reason", "session_id", "candidate_id", + "candidate_sha256", "identity_fingerprint", "expected_pointer_sha256", + "history", "production_admission_status", "execution_authority", + "can_execute", "can_trade", "capital_permission", +}) +CURRENT_POINTER_KEYS = frozenset({ + "schema_version", "state", "twin_id", "candidate_id", "candidate_sha256", + "identity_fingerprint", "production_admission_status", "execution_authority", + "can_execute", "can_trade", "capital_permission", +}) +CURRENT_SESSION_KEYS = frozenset({ + "schema", "mode", "declared", "binding_verified", "reason", + "authority_generation", "challenge_id", "challenge_sha256", "ack_sha256", + "session_effect_ceiling", "authority_ceiling", "effects", +}) +CURRENT_SESSION_EFFECT_KEYS = frozenset({ + "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", "capital_permission", "deploy_permission", +}) +AUTH_REQUEST_KEYS = frozenset({ + "schema", "terminal", "authorization_granted", "execution_authorized", + "apply_status", "request_id", "proposal_id", "proposal_file_sha256", + "project_id", "expected_base", "operation_count", +}) +EXPECTED_BASE_KEYS = frozenset({ + "projection_sha256", "event_cursor", "event_chain_head", +}) +PREFLIGHT_KEYS = frozenset({ + "schema", "terminal", "packet_valid", "authorization_record_valid", + "apply_ready", "execution_authorized", "packet_id", "project_id", + "proposal_id", "proposal_file_sha256", "expected_base", "operation_count", + "authorization_file_sha256", "authorization", +}) +AUTHORIZATION_KEYS = frozenset({"class", "id", "ref"}) +POLICY_REQUIRED_KEYS = frozenset({ + "schema_version", "tenant_id", "actors", "grants", "delegations", + "explicit_denies", "revoked_delegation_ids", +}) +POLICY_OPTIONAL_KEYS = frozenset({"max_delegation_depth"}) +SOURCE_RECORD_KEYS = frozenset({ + "actor_id", "actor_kind", "actor_role", "authority_class", "connector_id", + "content_hash", "cursor", "deleted", "duplicate_of", "effective_at", "id", + "manager_actor_id", "observed_at", "parser_version", "payload", "raw_ref", + "revision_id", "schema_version", "scope", "source_acl", + "source_envelope_id", "source_object_id", "source_object_type", + "source_system", "supersedes", "tenant_id", "truth_class", "visibility", +}) + + +class PersonTwinAuthorityReplayError(ValueError): + pass + + +def _require_exact_keys(value: Mapping[str, Any], expected: frozenset[str], label: str) -> None: + if not isinstance(value, Mapping): + raise PersonTwinAuthorityReplayError(f"{label} must be object") + actual = set(value) + if actual != set(expected): + missing = sorted(set(expected) - actual) + unexpected = sorted(actual - set(expected)) + raise PersonTwinAuthorityReplayError( + f"{label} keys mismatch: missing={missing}, unexpected={unexpected}" + ) + + +def _require_closed_keys( + value: Mapping[str, Any], + required: frozenset[str], + optional: frozenset[str], + label: str, +) -> None: + if not isinstance(value, Mapping): + raise PersonTwinAuthorityReplayError(f"{label} must be object") + actual = set(value) + missing = sorted(set(required) - actual) + unexpected = sorted(actual - set(required) - set(optional)) + if missing or unexpected: + raise PersonTwinAuthorityReplayError( + f"{label} keys mismatch: missing={missing}, unexpected={unexpected}" + ) + + +def _safe_receipt_scalar(value: Any) -> str: + return value if isinstance(value, str) else "" + + +class AdmissionReadStore(Protocol): + """Read-only subset of the R3 admission store used to prove CURRENT.""" + + def read_current_pointer_bytes(self) -> bytes | None: ... + + def read_candidate_bytes(self, candidate_id: str) -> bytes | None: ... + + +class DurableReplayReadStore(Protocol): + """Read-only durable replay lookup. R4 never writes this store.""" + + def read_replay_receipt_bytes(self, replay_key: str) -> bytes | None: ... + + +def _canonical_bytes(value: Mapping[str, Any]) -> bytes: + return json.dumps( + dict(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _hash(value: Any) -> str: + raw = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(raw).hexdigest() + + +def _text(value: Any, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise PersonTwinAuthorityReplayError(f"{field} must be non-empty") + return value + + +def _sha(value: Any, field: str) -> str: + value = _text(value, field) + if len(value) != 64 or any(ch not in "0123456789abcdef" for ch in value): + raise PersonTwinAuthorityReplayError(f"{field} must be lowercase sha256") + return value + + +def _mapping_from_canonical_bytes(payload: bytes, label: str) -> dict[str, Any]: + if not isinstance(payload, bytes) or not payload: + raise PersonTwinAuthorityReplayError(f"{label} bytes missing") + try: + decoded = json.loads(payload.decode("utf-8")) + except Exception as exc: + raise PersonTwinAuthorityReplayError(f"{label} bytes invalid JSON") from exc + if not isinstance(decoded, Mapping): + raise PersonTwinAuthorityReplayError(f"{label} root must be object") + result = dict(decoded) + if _canonical_bytes(result) != payload: + raise PersonTwinAuthorityReplayError(f"{label} bytes are not canonical") + return result + + +def _ceiling(value: Mapping[str, Any], label: str) -> None: + if value.get("production_admission_status") != NOT_PRODUCTION_ADMITTED: + raise PersonTwinAuthorityReplayError(f"{label} production admission escalation") + if value.get("execution_authority") != "NONE" or value.get("can_execute") is not False: + raise PersonTwinAuthorityReplayError(f"{label} execution authority escalation") + if value.get("can_trade") is not False: + raise PersonTwinAuthorityReplayError(f"{label} trading authority escalation") + if value.get("capital_permission") != "DENY": + raise PersonTwinAuthorityReplayError(f"{label} capital authority escalation") + + +def _current_admission( + identity: Mapping[str, Any], + record: Mapping[str, Any], + admission: Mapping[str, Any], + pointer: Mapping[str, Any], + admission_store: AdmissionReadStore, +) -> dict[str, str]: + _require_exact_keys(admission, ADMISSION_KEYS, "admission") + _require_exact_keys(pointer, CURRENT_POINTER_KEYS, "current pointer") + if admission.get("schema_version") != ADMISSION_SCHEMA or admission.get("state") != "CURRENT": + raise PersonTwinAuthorityReplayError("admission receipt is not exact CURRENT") + _ceiling(admission, "admission") + session_id = _text(admission.get("session_id"), "admission.session_id") + candidate_id = _text(admission.get("candidate_id"), "admission.candidate_id") + candidate_sha = _sha(admission.get("candidate_sha256"), "admission.candidate_sha256") + if session_id != record.get("admission_session_id"): + raise PersonTwinAuthorityReplayError("admission session identity mismatch") + expected_candidate_bytes = _canonical_bytes(record) + if candidate_id != record.get("id") or candidate_sha != hashlib.sha256(expected_candidate_bytes).hexdigest(): + raise PersonTwinAuthorityReplayError("admission candidate identity mismatch") + if admission.get("identity_fingerprint") != identity.get("identity_fingerprint"): + raise PersonTwinAuthorityReplayError("admission identity fingerprint mismatch") + + try: + stored_candidate = admission_store.read_candidate_bytes(candidate_id) + stored_pointer = admission_store.read_current_pointer_bytes() + except Exception as exc: + raise PersonTwinAuthorityReplayError("admission readback failed") from exc + if stored_candidate is None: + raise PersonTwinAuthorityReplayError("current candidate missing from admission store") + if stored_candidate != expected_candidate_bytes: + raise PersonTwinAuthorityReplayError("candidate byte readback mismatch") + if hashlib.sha256(stored_candidate).hexdigest() != candidate_sha: + raise PersonTwinAuthorityReplayError("candidate hash readback mismatch") + if stored_pointer is None: + raise PersonTwinAuthorityReplayError("current pointer missing from admission store") + + stored_pointer_value = _mapping_from_canonical_bytes(stored_pointer, "current pointer") + if _canonical_bytes(pointer) != stored_pointer: + raise PersonTwinAuthorityReplayError("caller current pointer differs from admission readback") + if stored_pointer_value != dict(pointer): + raise PersonTwinAuthorityReplayError("current pointer readback mismatch") + if stored_pointer_value.get("schema_version") != CURRENT_POINTER_SCHEMA or stored_pointer_value.get("state") != "CURRENT": + raise PersonTwinAuthorityReplayError("current pointer is not exact CURRENT") + _ceiling(stored_pointer_value, "current pointer") + expected = { + "twin_id": identity.get("twin_id"), + "candidate_id": candidate_id, + "candidate_sha256": candidate_sha, + "identity_fingerprint": identity.get("identity_fingerprint"), + } + if any(stored_pointer_value.get(key) != val for key, val in expected.items()): + raise PersonTwinAuthorityReplayError("current pointer identity mismatch") + return { + "session_id": session_id, + "candidate_id": candidate_id, + "candidate_sha256": candidate_sha, + "current_pointer_hash": hashlib.sha256(stored_pointer).hexdigest(), + } + + +def _current_session(value: Mapping[str, Any]) -> str: + _require_exact_keys(value, CURRENT_SESSION_KEYS, "current session") + if value.get("schema") != CURRENT_EFFECT_SCHEMA or value.get("mode") != MODE_CURRENT: + raise PersonTwinAuthorityReplayError("current session is not CURRENT") + if value.get("declared") is not True or value.get("reason") != EXACT_CURRENT_REASON: + raise PersonTwinAuthorityReplayError("current session declaration/reason invalid") + if value.get("binding_verified") is not True or value.get("session_effect_ceiling") != "READ_ONLY": + raise PersonTwinAuthorityReplayError("current session binding/ceiling invalid") + if value.get("authority_ceiling") != EXACT_CURRENT_AUTHORITY_CEILING: + raise PersonTwinAuthorityReplayError("current session authority ceiling mismatch") + effects = value.get("effects") + if not isinstance(effects, Mapping): + raise PersonTwinAuthorityReplayError("current session effects missing") + _require_exact_keys(effects, CURRENT_SESSION_EFFECT_KEYS, "current session effects") + if effects.get("legacy_fallback") is not False: + raise PersonTwinAuthorityReplayError("current session legacy fallback enabled") + denied = CURRENT_SESSION_EFFECT_KEYS - {"legacy_fallback", "capital_permission", "deploy_permission"} + if any(effects.get(key) is not False for key in denied): + raise PersonTwinAuthorityReplayError("current session effect ceiling widened") + if effects.get("capital_permission") != "DENY" or effects.get("deploy_permission") != "DENY": + raise PersonTwinAuthorityReplayError("current session deny ceiling widened") + for field in ("authority_generation", "challenge_id"): + _text(value.get(field), f"current_session.{field}") + _sha(value.get("challenge_sha256"), "current_session.challenge_sha256") + _sha(value.get("ack_sha256"), "current_session.ack_sha256") + return _hash(dict(value)) + + +def _authorization_chain(request: Mapping[str, Any], preflight: Mapping[str, Any]) -> dict[str, str]: + _require_exact_keys(request, AUTH_REQUEST_KEYS, "authorization request") + _require_exact_keys(preflight, PREFLIGHT_KEYS, "authorization preflight") + if request.get("schema") != AUTH_REQUEST_SCHEMA or request.get("terminal") != "CURRENT_MEMORY_APPLY_AUTH_REQUEST_PASS": + raise PersonTwinAuthorityReplayError("authorization request is not PASS") + if request.get("authorization_granted") is not False or request.get("execution_authorized") is not False: + raise PersonTwinAuthorityReplayError("authorization request manufactured authority") + if request.get("apply_status") != "NOT_APPLIED": + raise PersonTwinAuthorityReplayError("authorization request is not pre-apply") + + proposal_id = _text(request.get("proposal_id"), "request.proposal_id") + proposal_sha = _sha(request.get("proposal_file_sha256"), "request.proposal_file_sha256") + project_id = _text(request.get("project_id"), "request.project_id") + count = request.get("operation_count") + base = request.get("expected_base") + if not isinstance(base, Mapping) or not isinstance(count, int) or isinstance(count, bool) or count < 1: + raise PersonTwinAuthorityReplayError("authorization request base/count invalid") + _require_exact_keys(base, EXPECTED_BASE_KEYS, "authorization request expected_base") + snapshot_sha = _sha(base.get("projection_sha256"), "request.base.projection_sha256") + _sha(base.get("event_chain_head"), "request.base.event_chain_head") + cursor = base.get("event_cursor") + if not isinstance(cursor, int) or isinstance(cursor, bool) or cursor < 0: + raise PersonTwinAuthorityReplayError("authorization request cursor invalid") + request_id = _text(request.get("request_id"), "request.request_id") + expected_request_id = "omar-" + _hash({ + "proposal_id": proposal_id, + "proposal_file_sha256": proposal_sha, + "project_id": project_id, + "base": dict(base), + "operation_count": count, + })[:40] + if request_id != expected_request_id: + raise PersonTwinAuthorityReplayError("authorization request_id integrity mismatch") + + if preflight.get("schema") != PREFLIGHT_SCHEMA or preflight.get("terminal") != "CURRENT_PROJECT_UPDATE_PREFLIGHT_READY": + raise PersonTwinAuthorityReplayError("authorization preflight is not READY") + required_true = ("packet_valid", "authorization_record_valid", "apply_ready") + if any(preflight.get(key) is not True for key in required_true) or preflight.get("execution_authorized") is not False: + raise PersonTwinAuthorityReplayError("authorization preflight failed closed") + if preflight.get("project_id") != project_id: + raise PersonTwinAuthorityReplayError("authorization project identity mismatch") + if preflight.get("operation_count") != count: + raise PersonTwinAuthorityReplayError("authorization operation-count mismatch") + if preflight.get("proposal_id") != proposal_id or preflight.get("proposal_file_sha256") != proposal_sha: + raise PersonTwinAuthorityReplayError("authorization proposal identity mismatch") + preflight_base = preflight.get("expected_base") + if not isinstance(preflight_base, Mapping): + raise PersonTwinAuthorityReplayError("authorization preflight base missing") + _require_exact_keys(preflight_base, EXPECTED_BASE_KEYS, "authorization preflight expected_base") + if dict(preflight_base) != dict(base): + raise PersonTwinAuthorityReplayError("authorization snapshot identity mismatch") + + auth = preflight.get("authorization") + if not isinstance(auth, Mapping): + raise PersonTwinAuthorityReplayError("authorization identity missing") + _require_exact_keys(auth, AUTHORIZATION_KEYS, "authorization identity") + auth_class = _text(auth.get("class"), "authorization.class").upper() + if auth_class not in {"HUMAN", "DETERMINISTIC_CONTROLLER"}: + raise PersonTwinAuthorityReplayError("authorization class invalid") + auth_id = _text(auth.get("id"), "authorization.id") + auth_ref = _text(auth.get("ref"), "authorization.ref") + auth_file_sha = _sha(preflight.get("authorization_file_sha256"), "preflight.authorization_file_sha256") + packet_id = _text(preflight.get("packet_id"), "preflight.packet_id") + expected_packet_id = "r4p-" + _hash({ + "request_id": request_id, + "project_id": project_id, + "proposal_id": proposal_id, + "proposal_file_sha256": proposal_sha, + "expected_base": dict(base), + "operation_count": count, + "authorization_file_sha256": auth_file_sha, + "authorization": {"class": auth_class, "id": auth_id, "ref": auth_ref}, + })[:40] + if packet_id != expected_packet_id: + raise PersonTwinAuthorityReplayError("authorization packet identity mismatch") + return { + "request_id": request_id, + "proposal_id": proposal_id, + "proposal_file_sha256": proposal_sha, + "project_id": project_id, + "operation_count": count, + "packet_id": packet_id, + "authorization_file_sha256": auth_file_sha, + "authorization_class": auth_class, + "authorization_id": auth_id, + "authorization_ref": auth_ref, + "snapshot_sha256": snapshot_sha, + "request_hash": _hash(dict(request)), + "preflight_hash": _hash(dict(preflight)), + } + + +def _actor(policy: Mapping[str, Any], actor_id: str) -> dict[str, Any]: + actors = [row for row in policy.get("actors", []) if isinstance(row, Mapping) and row.get("id") == actor_id] + if len(actors) != 1: + raise PersonTwinAuthorityReplayError("policy actor binding ambiguous") + row = actors[0] + return { + "actor_id": actor_id, + "actor_kind": _text(row.get("actor_kind"), "actor.actor_kind"), + "role": _text(row.get("role"), "actor.role"), + "delegation_ids": sorted( + _text(d.get("id"), "delegation.id") + for d in policy.get("delegations", []) + if isinstance(d, Mapping) and d.get("grantee_actor_id") == actor_id and d.get("id") + ), + } + + +def _stable_replay_key(binding: Mapping[str, Any]) -> str: + return _hash({ + "tenant_id": binding["tenant_id"], + "twin_id": binding["twin_id"], + "person_record_id": binding["person_record_id"], + "candidate_id": binding["candidate_id"], + "proposal_id": binding["proposal_id"], + "action": binding["action"], + }) + + +def _receipt( + decision: str, + reason: str, + action: str, + at: str, + *, + binding: Mapping[str, Any] | None = None, + replay_status: str = "DENIED", + policy_sha: str | None = None, +) -> dict[str, Any]: + bound = dict(binding or {}) + body = { + "schema_version": SCHEMA_VERSION, + "decision": decision, + "reason": reason, + "action": action, + "evaluated_at": at, + "replay_status": replay_status, + "effect": "READ_ONLY_DECISION", + "mutated": False, + "current_state_apply": False, + "canonical_mutation": False, + "execution_authorized": False, + "production_admission_status": NOT_PRODUCTION_ADMITTED, + "execution_authority": "NONE", + "can_execute": False, + "can_trade": False, + "capital_permission": "DENY", + "deploy_permission": "DENY", + "binding": bound, + "replay_key": bound.get("replay_key"), + "replay_identity_hash": bound.get("replay_identity_hash"), + "policy_receipt_sha256": policy_sha, + } + body["receipt_hash"] = _hash(body) + return body + + +def _validate_durable(receipt: Mapping[str, Any]) -> dict[str, Any]: + if receipt.get("schema_version") != SCHEMA_VERSION: + raise PersonTwinAuthorityReplayError("durable replay schema mismatch") + if receipt.get("decision") != "ALLOW": + raise PersonTwinAuthorityReplayError("durable replay receipt is not ALLOW") + if receipt.get("effect") != "READ_ONLY_DECISION": + raise PersonTwinAuthorityReplayError("durable replay effect mismatch") + if any(( + receipt.get("mutated") is not False, + receipt.get("current_state_apply") is not False, + receipt.get("canonical_mutation") is not False, + receipt.get("execution_authorized") is not False, + receipt.get("production_admission_status") != NOT_PRODUCTION_ADMITTED, + receipt.get("execution_authority") != "NONE", + receipt.get("can_execute") is not False, + receipt.get("can_trade") is not False, + receipt.get("capital_permission") != "DENY", + receipt.get("deploy_permission") != "DENY", + )): + raise PersonTwinAuthorityReplayError("durable replay authority ceiling mismatch") + + claimed_receipt_hash = _sha(receipt.get("receipt_hash"), "durable.receipt_hash") + body = dict(receipt) + body.pop("receipt_hash", None) + if claimed_receipt_hash != _hash(body): + raise PersonTwinAuthorityReplayError("durable replay receipt hash mismatch") + + binding = receipt.get("binding") + if not isinstance(binding, Mapping): + raise PersonTwinAuthorityReplayError("durable replay binding missing") + binding_value = dict(binding) + claimed_replay_hash = _sha(binding_value.get("replay_identity_hash"), "durable.binding.replay_identity_hash") + if receipt.get("replay_identity_hash") != claimed_replay_hash: + raise PersonTwinAuthorityReplayError("durable replay top-level identity mismatch") + claimed_replay_key = _sha(binding_value.get("replay_key"), "durable.binding.replay_key") + if receipt.get("replay_key") != claimed_replay_key: + raise PersonTwinAuthorityReplayError("durable replay top-level key mismatch") + replay_body = dict(binding_value) + replay_body.pop("replay_identity_hash", None) + if _hash(replay_body) != claimed_replay_hash: + raise PersonTwinAuthorityReplayError("durable replay identity binding mismatch") + if _stable_replay_key(binding_value) != claimed_replay_key: + raise PersonTwinAuthorityReplayError("durable replay stable key mismatch") + return binding_value + + +def _read_durable_replay(replay_store: DurableReplayReadStore, *, replay_key: str) -> dict[str, Any] | None: + try: + payload = replay_store.read_replay_receipt_bytes(replay_key) + except Exception as exc: + raise PersonTwinAuthorityReplayError("durable replay readback failed") from exc + if payload is None: + return None + value = _mapping_from_canonical_bytes(payload, "durable replay receipt") + _validate_durable(value) + return value + + +def evaluate_person_twin_authority_replay(*args: Any, **kwargs: Any) -> dict[str, Any]: + """Public fail-closed R4 boundary. Argument-shape errors never escape as TypeError.""" + action = _safe_receipt_scalar(kwargs.get("action")) + at = _safe_receipt_scalar(kwargs.get("at")) + try: + if args: + raise PersonTwinAuthorityReplayError("positional arguments are unsupported") + actual = set(kwargs) + missing = sorted(set(PUBLIC_INPUT_KEYS) - actual) + unexpected = sorted(actual - set(PUBLIC_INPUT_KEYS)) + if missing or unexpected: + raise PersonTwinAuthorityReplayError( + f"public input keys mismatch: missing={missing}, unexpected={unexpected}" + ) + return _evaluate_person_twin_authority_replay_impl(**kwargs) + except Exception as exc: + return _receipt( + "DENY", + f"FAIL_CLOSED:{type(exc).__name__}:{exc}", + action, + at, + ) + + +def _evaluate_person_twin_authority_replay_impl( + *, + identity: Mapping[str, Any], + consent_receipt: Mapping[str, Any], + current_revocation_ledger: Mapping[str, Any], + source_record: Mapping[str, Any], + person_record: Mapping[str, Any], + admission_receipt: Mapping[str, Any], + current_pointer: Mapping[str, Any], + admission_store: AdmissionReadStore, + policy: Mapping[str, Any], + principal_id: str, + action: str, + requested_scope: str, + requested_privacy_class: str, + purpose: str, + at: str, + current_session: Mapping[str, Any], + authorization_request: Mapping[str, Any], + authorization_preflight: Mapping[str, Any], + replay_store: DurableReplayReadStore, +) -> dict[str, Any]: + """Pure R4 Person Twin authority/replay decision. Never executes or mutates.""" + try: + action = _text(action, "action").upper() + at = _text(at, "at") + purpose = _text(purpose, "purpose") + principal_id = _text(principal_id, "principal_id") + requested_scope = requested_scope + requested_privacy_class = _text(requested_privacy_class, "requested_privacy_class") + _require_exact_keys(source_record, SOURCE_RECORD_KEYS, "source record") + _require_closed_keys(policy, POLICY_REQUIRED_KEYS, POLICY_OPTIONAL_KEYS, "policy") + if "max_delegation_depth" in policy: + depth = policy.get("max_delegation_depth") + if not isinstance(depth, int) or isinstance(depth, bool) or depth < 1: + raise PersonTwinAuthorityReplayError("policy max_delegation_depth must be positive integer") + validate_person_twin_identity(identity) + validate_source_consent_receipt(identity, consent_receipt) + validate_consent_revocation_ledger(current_revocation_ledger) + validate_person_twin_record( + person_record, + identity=identity, + consent_receipt=consent_receipt, + source_record=source_record, + ) + scope = requested_scope + privacy = requested_privacy_class + if scope != person_record.get("scope") or privacy != person_record.get("privacy_class"): + raise PersonTwinAuthorityReplayError("requested privacy/scope identity drift") + validate_privacy_scope(twin_id=_text(identity.get("twin_id"), "identity.twin_id"), privacy_class=privacy, scope=scope) + + ledger_hash = _sha(current_revocation_ledger.get("ledger_hash"), "revocation_ledger.ledger_hash") + if person_record.get("revocation_ledger_hash") != ledger_hash: + raise PersonTwinAuthorityReplayError("current revocation ledger identity drift") + consent = evaluate_consent( + identity, + consent_receipt, + revocation_ledger=current_revocation_ledger, + at=at, + requested_object_type=_text(source_record.get("source_object_type"), "source_record.source_object_type"), + requested_scope=scope, + purpose=purpose, + require_source_read=False, + require_memory_admission=True, + ) + if consent.get("decision") != "ALLOW" or consent.get("revocation_ledger_hash") != ledger_hash: + raise PersonTwinAuthorityReplayError(f"current consent denied:{consent.get('reason')}") + + admission = _current_admission(identity, person_record, admission_receipt, current_pointer, admission_store) + session_hash = _current_session(current_session) + auth = _authorization_chain(authorization_request, authorization_preflight) + validate_policy(policy) + if policy.get("tenant_id") != identity.get("tenant_id"): + raise PersonTwinAuthorityReplayError("policy tenant identity mismatch") + if action == "EXECUTE": + raise PersonTwinAuthorityReplayError("EXECUTE unsupported in R4") + if action not in SUPPORTED_ACTIONS: + raise PersonTwinAuthorityReplayError("action outside R4 scope") + + principal = principal_id + source_acl = source_record.get("source_acl") + if not isinstance(source_acl, Mapping): + raise PersonTwinAuthorityReplayError("source_record.source_acl must be object") + resource = { + "id": _text(person_record.get("id"), "person_record.id"), + "tenant_id": _text(identity.get("tenant_id"), "identity.tenant_id"), + "scope": scope, + "source_acl_scopes": [_text(source_acl.get("scope"), "source_record.source_acl.scope")], + "classification": privacy, + } + policy_decision = evaluate_policy( + policy, + principal_id=principal, + resource=resource, + action=action, + context={"purpose": purpose}, + at=at, + ) + if policy_decision.get("decision") != "ALLOW": + raise PersonTwinAuthorityReplayError(f"policy denied:{policy_decision.get('reason')}") + actor = _actor(policy, _text(policy_decision.get("actor_id"), "policy.actor_id")) + if action == "APPROVE": + if actor["actor_kind"] != "HUMAN": + raise PersonTwinAuthorityReplayError("APPROVE requires HUMAN actor") + if auth["authorization_class"] != "HUMAN": + raise PersonTwinAuthorityReplayError("APPROVE requires HUMAN authorization identity") + if auth["authorization_id"] != principal: + raise PersonTwinAuthorityReplayError("APPROVE principal does not match authorization identity") + + policy_receipt_sha = _sha(policy_decision.get("receipt_sha256"), "policy.receipt_sha256") + replay_identity: dict[str, Any] = { + "tenant_id": identity["tenant_id"], + "twin_id": identity["twin_id"], + "identity_fingerprint": identity["identity_fingerprint"], + "person_record_id": person_record["id"], + "provenance_hash": _sha(person_record.get("provenance_hash"), "record.provenance_hash"), + "source_record_id": person_record["source_record_id"], + "source_content_hash": _sha(person_record.get("content_hash"), "record.content_hash"), + "privacy_class": privacy, + "scope": scope, + "consent_receipt_id": consent_receipt["consent_receipt_id"], + "consent_receipt_hash": consent_receipt["receipt_hash"], + "current_revocation_ledger_hash": ledger_hash, + "admission_session_id": admission["session_id"], + "candidate_id": admission["candidate_id"], + "candidate_sha256": admission["candidate_sha256"], + "current_pointer_hash": admission["current_pointer_hash"], + "policy_hash": _hash(dict(policy)), + "policy_receipt_sha256": policy_receipt_sha, + "principal_id": principal, + **actor, + "current_session_hash": session_hash, + "authority_generation": current_session["authority_generation"], + "challenge_id": current_session["challenge_id"], + "challenge_sha256": current_session["challenge_sha256"], + "ack_sha256": current_session["ack_sha256"], + **auth, + "action": action, + "purpose": purpose, + } + replay_key = _stable_replay_key(replay_identity) + replay_identity["replay_key"] = replay_key + replay_hash = _hash(replay_identity) + binding = dict(replay_identity) + binding["replay_identity_hash"] = replay_hash + + durable = _read_durable_replay(replay_store, replay_key=replay_key) + if durable is not None: + if durable.get("replay_key") != replay_key: + raise PersonTwinAuthorityReplayError("durable replay key lookup mismatch") + if durable.get("replay_identity_hash") != replay_hash: + return _receipt( + "DENY", + "REPLAY_IDENTITY_MISMATCH", + action, + at, + binding=binding, + replay_status="DIVERGENT_DENIED", + policy_sha=policy_receipt_sha, + ) + return _receipt( + "ALLOW", + "EXACT_REPLAY_IDEMPOTENT", + action, + at, + binding=binding, + replay_status="EXACT_IDEMPOTENT", + policy_sha=policy_receipt_sha, + ) + return _receipt( + "ALLOW", + "PERSON_TWIN_AUTHORITY_ALLOW", + action, + at, + binding=binding, + replay_status="NEW", + policy_sha=policy_receipt_sha, + ) + except Exception as exc: + return _receipt("DENY", f"FAIL_CLOSED:{type(exc).__name__}:{exc}", action, at) diff --git a/tests/test_person_twin_authority_replay.py b/tests/test_person_twin_authority_replay.py new file mode 100644 index 00000000..eb1c9aaf --- /dev/null +++ b/tests/test_person_twin_authority_replay.py @@ -0,0 +1,683 @@ +from __future__ import annotations + +import copy +import hashlib +import json + +import pytest + +from continuityos.company_twin_ingest import normalize_envelope +from continuityos.person_twin_admission_contracts import ( + build_consent_revocation_entry, + build_consent_revocation_ledger, + build_person_twin_identity, + build_source_consent_receipt, +) +from continuityos.person_twin_admission_coordinator import ( + AdmissionStoreConflict, + PersonTwinAdmissionCoordinator, +) +from continuityos.person_twin_authority_replay import ( + SCHEMA_VERSION, + evaluate_person_twin_authority_replay, +) +from continuityos.person_twin_privacy_provenance import ( + build_person_twin_provenance, + canonical_person_private_scope, + deterministic_source_identity_hash, +) + +AT = "2026-08-25T12:50:00Z" + + +def _canonical_bytes(value) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _hash(value) -> str: + return hashlib.sha256(_canonical_bytes(value)).hexdigest() + + +def _sha(value: bytes | None) -> str | None: + return None if value is None else hashlib.sha256(value).hexdigest() + + +class Store: + def __init__(self): + self.pointer = b'{"candidate":"last-known-good"}' + self.candidates: dict[str, bytes] = {} + + def read_current_pointer_bytes(self): + return self.pointer + + def read_candidate_bytes(self, candidate_id): + return self.candidates.get(candidate_id) + + def commit_candidate_not_current(self, candidate_id, candidate_bytes): + old = self.candidates.get(candidate_id) + if old is not None and old != candidate_bytes: + raise AdmissionStoreConflict("candidate conflict") + self.candidates[candidate_id] = candidate_bytes + + def promote_current_pointer(self, *, expected_pointer_sha256, pointer_bytes): + if _sha(self.pointer) != expected_pointer_sha256: + raise AdmissionStoreConflict("stale pointer") + self.pointer = pointer_bytes + + +class NullAdmissionStore: + def read_current_pointer_bytes(self): + return None + + def read_candidate_bytes(self, candidate_id): + return None + + +class ReplayStore: + def __init__(self): + self.receipts: dict[str, bytes] = {} + + def read_replay_receipt_bytes(self, replay_key): + return self.receipts.get(replay_key) + + def seed(self, receipt): + self.receipts[receipt["replay_key"]] = _canonical_bytes(receipt) + + +class ExplodingReplayStore: + def read_replay_receipt_bytes(self, replay_key): + raise RuntimeError("synthetic read failure") + + +def _person(): + identity = build_person_twin_identity( + tenant_id="tenant:person-pilot", + subject_id="subject:owner", + owner_id="human:owner", + controller_id="human:owner", + controller_kind="HUMAN", + ownership_epoch=1, + created_at="2026-08-25T00:00:00Z", + ) + scope = canonical_person_private_scope(identity["twin_id"]) + source = normalize_envelope({ + "schema_version": "company-twin-source-envelope/1", + "tenant_id": identity["tenant_id"], + "connector_id": "connector:synthetic", + "source_system": "synthetic_drive", + "source_object_type": "document", + "source_object_id": "doc-r4", + "revision_id": "r1", + "observed_at": "2026-08-25T10:01:00Z", + "effective_at": "2026-08-25T10:00:00Z", + "acl": {"visibility": "PERSONAL", "scope": scope}, + "payload": {"title": "R4", "text": "synthetic"}, + "raw_ref": "synthetic://r4/doc", + "cursor": "cursor:r4", + "actor": { + "actor_id": "human:owner", + "actor_kind": "HUMAN", + "authority_class": "OWNER", + }, + "deleted": False, + }) + consent = build_source_consent_receipt( + identity, + authorizing_principal="human: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="person-memory", + 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="p3-p1-r4r1-test/1", + ) + ledger = build_consent_revocation_ledger() + record = build_person_twin_provenance( + identity, + consent, + ledger, + source, + admission_session_id="admission:r4r1:1", + purpose="person-memory", + at="2026-08-25T11:00:00Z", + ) + store = Store() + coordinator = PersonTwinAdmissionCoordinator( + store, + session_id=record["admission_session_id"], + identity=identity, + consent_receipt=consent, + source_record=source, + candidate_record=record, + purpose="person-memory", + ) + assert coordinator.validate(revocation_ledger=ledger, at="2026-08-25T11:10:00Z")["state"] == "VALIDATED" + assert coordinator.commit_not_current(current_revocation_ledger=ledger, at="2026-08-25T11:20:00Z")["state"] == "COMMITTED_NOT_CURRENT" + assert coordinator.postverify(current_revocation_ledger=ledger, at="2026-08-25T11:30:00Z")["state"] == "POSTVERIFY_PASS" + admission = coordinator.promote_current(current_revocation_ledger=ledger, at="2026-08-25T11:40:00Z") + assert admission["state"] == "CURRENT" + pointer = json.loads(store.pointer.decode("utf-8")) + return identity, source, consent, ledger, record, admission, pointer, store + + +def _policy(identity, *, revoked=False): + scope = canonical_person_private_scope(identity["twin_id"]) + return { + "schema_version": "company-twin-p2c/1", + "tenant_id": identity["tenant_id"], + "actors": [ + { + "id": "actor:owner", + "principal_id": "human:owner", + "actor_kind": "HUMAN", + "role": "DIRECTOR", + "scopes": [scope], + }, + { + "id": "actor:agent", + "principal_id": "agent:assistant", + "actor_kind": "AGENT", + "role": "AGENT", + "scopes": [scope], + "manager_actor_id": "actor:owner", + }, + ], + "grants": [{ + "id": "grant:owner", + "actor_id": "actor:owner", + "actions": ["READ", "PROPOSE", "APPROVE", "DELEGATE"], + "scopes": [scope], + "purposes": ["person-memory"], + }], + "delegations": [{ + "id": "delegation:agent", + "grantor_actor_id": "actor:owner", + "grantee_actor_id": "actor:agent", + "actions": ["READ", "PROPOSE"], + "scopes": [scope], + "purposes": ["person-memory"], + "parent_id": None, + "expires_at": "2026-08-26T00:00:00Z", + "revoked": False, + }], + "explicit_denies": [], + "revoked_delegation_ids": ["delegation:agent"] if revoked else [], + } + + +def _session(): + return { + "schema": "continuityos.current_effect_boundary/v1", + "mode": "CURRENT", + "declared": True, + "binding_verified": True, + "reason": "EXACT_CURRENT_SESSION_VERIFIED", + "authority_generation": "R64", + "challenge_id": "challenge:r4r1:1", + "challenge_sha256": "1" * 64, + "ack_sha256": "2" * 64, + "session_effect_ceiling": "READ_ONLY", + "authority_ceiling": "NO_FURTHER_AGENT_WORK", + "effects": { + "legacy_fallback": False, + "memory_write": False, + "ledger_write": False, + "filesystem_write": False, + "network_effect": False, + "server_started": False, + "subprocess_execution": False, + "deployment": False, + "current_state_apply": False, + "canonical_mutation": False, + "external_message": False, + "auto_dispatch": False, + "trading": False, + "wallet_access": False, + "order_execution": False, + "can_trade": False, + "capital_permission": "DENY", + "deploy_permission": "DENY", + }, + } + + +def _authorization( + *, + auth_sha="6" * 64, + auth_class="HUMAN", + auth_id="human:owner", +): + base = { + "projection_sha256": "3" * 64, + "event_cursor": 7, + "event_chain_head": "4" * 64, + } + core = { + "proposal_id": "proposal:r4r1:1", + "proposal_file_sha256": "5" * 64, + "project_id": "project:r4r1", + "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, + } + auth_ref = "synthetic://approval/r4r1" + packet_core = { + "request_id": request["request_id"], + "project_id": core["project_id"], + "proposal_id": core["proposal_id"], + "proposal_file_sha256": core["proposal_file_sha256"], + "expected_base": base, + "operation_count": 1, + "authorization_file_sha256": auth_sha, + "authorization": {"class": auth_class, "id": auth_id, "ref": auth_ref}, + } + 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], + "project_id": core["project_id"], + "proposal_id": core["proposal_id"], + "proposal_file_sha256": core["proposal_file_sha256"], + "expected_base": base, + "operation_count": 1, + "authorization_file_sha256": auth_sha, + "authorization": { + "class": auth_class, + "id": auth_id, + "ref": auth_ref, + }, + } + return request, preflight + + +def _inputs(*, principal="agent:assistant", action="READ"): + identity, source, consent, ledger, record, admission, pointer, admission_store = _person() + request, preflight = _authorization() + return dict( + identity=identity, + consent_receipt=consent, + current_revocation_ledger=ledger, + source_record=source, + person_record=record, + admission_receipt=admission, + current_pointer=pointer, + admission_store=admission_store, + policy=_policy(identity), + principal_id=principal, + action=action, + requested_scope=record["scope"], + requested_privacy_class=record["privacy_class"], + purpose="person-memory", + at=AT, + current_session=_session(), + authorization_request=request, + authorization_preflight=preflight, + replay_store=ReplayStore(), + ) + + +def test_agent_read_and_propose_are_non_effectful_allows(): + for action in ("READ", "PROPOSE"): + result = evaluate_person_twin_authority_replay(**_inputs(action=action)) + assert result["schema_version"] == SCHEMA_VERSION + assert result["decision"] == "ALLOW" + assert result["replay_status"] == "NEW" + assert result["mutated"] is False + assert result["current_state_apply"] is False + assert result["canonical_mutation"] is False + assert result["execution_authorized"] is False + assert result["production_admission_status"] == "NOT_PRODUCTION_ADMITTED" + assert result["execution_authority"] == "NONE" + assert result["can_execute"] is False + assert result["can_trade"] is False + assert result["capital_permission"] == "DENY" + + +def test_human_approve_is_decision_only_and_agent_approve_is_denied(): + human = evaluate_person_twin_authority_replay( + **_inputs(principal="human:owner", action="APPROVE") + ) + assert human["decision"] == "ALLOW" + assert human["execution_authorized"] is False + agent = evaluate_person_twin_authority_replay(**_inputs(action="APPROVE")) + assert agent["decision"] == "DENY" + assert "AGENT_AUTHORITY_CEILING" in agent["reason"] + + +@pytest.mark.parametrize( + "action, needle", + [("EXECUTE", "EXECUTE unsupported"), ("EXPORT", "action outside R4 scope")], +) +def test_effectful_or_r5_actions_are_not_in_r4(action, needle): + result = evaluate_person_twin_authority_replay(**_inputs(action=action)) + assert result["decision"] == "DENY" + assert needle in result["reason"] + assert result["execution_authorized"] is False + + +def test_caller_self_attested_current_without_store_readback_is_denied(): + x = _inputs() + x["admission_store"] = NullAdmissionStore() + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "current candidate missing from admission store" in result["reason"] + + +def test_caller_current_pointer_must_equal_independent_store_readback(): + x = _inputs() + x["current_pointer"] = {**x["current_pointer"], "challenge": "forged"} + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "current pointer keys mismatch" in result["reason"] + + +def test_candidate_byte_drift_in_admission_store_is_denied(): + x = _inputs() + x["admission_store"].candidates[x["person_record"]["id"]] = b"{}" + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "candidate byte readback mismatch" in result["reason"] + + +def test_pointer_byte_drift_in_admission_store_is_denied(): + x = _inputs() + x["admission_store"].pointer = _canonical_bytes({"state": "CURRENT", "candidate_id": "other"}) + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "caller current pointer differs from admission readback" in result["reason"] + + +def test_current_session_requires_exact_no_further_agent_work_ceiling(): + x = _inputs() + x["current_session"] = { + **x["current_session"], + "authority_ceiling": "EXECUTION_ALLOWED", + } + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "current session authority ceiling mismatch" in result["reason"] + + +@pytest.mark.parametrize( + "mutation, needle", + [ + ("scope", "requested privacy/scope identity drift"), + ("privacy", "requested privacy/scope identity drift"), + ("tenant", "policy tenant identity mismatch"), + ("effect", "current session effect ceiling widened"), + ("proposal", "authorization proposal identity mismatch"), + ("request", "authorization request_id integrity mismatch"), + ("provenance", "provenance_hash mismatch"), + ], +) +def test_identity_and_authority_drift_fail_closed(mutation, needle): + x = _inputs() + if mutation == "scope": + x["requested_scope"] = "team:ops" + elif mutation == "privacy": + x["requested_privacy_class"] = "COMPANY" + elif mutation == "tenant": + x["policy"] = {**x["policy"], "tenant_id": "tenant:other"} + elif mutation == "effect": + x["current_session"] = copy.deepcopy(x["current_session"]) + x["current_session"]["effects"]["filesystem_write"] = True + elif mutation == "proposal": + x["authorization_preflight"] = { + **x["authorization_preflight"], + "proposal_id": "proposal:other", + } + elif mutation == "request": + x["authorization_request"] = { + **x["authorization_request"], + "request_id": "omar-forged", + } + elif mutation == "provenance": + x["person_record"] = { + **x["person_record"], + "provenance_hash": "f" * 64, + } + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert needle in result["reason"] + + +def test_revocation_and_ledger_drift_fail_closed_even_for_prior_allow(): + for revoked_at in ("2026-08-25T12:45:00Z", "2026-08-25T23:30:00Z"): + x = _inputs() + entry = build_consent_revocation_entry( + x["identity"], + x["consent_receipt"], + revoking_principal="human:owner", + revoking_principal_kind="HUMAN", + revoked_at=revoked_at, + reason="synthetic R4R1 revoke", + ) + x["current_revocation_ledger"] = build_consent_revocation_ledger([entry]) + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "current revocation ledger identity drift" in result["reason"] + + +def test_revoked_delegation_denies_agent(): + x = _inputs() + x["policy"] = _policy(x["identity"], revoked=True) + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "NO_MATCHING_GRANT" in result["reason"] + + +def test_human_approve_principal_must_match_human_authorization_identity(): + x = _inputs(principal="human:owner", action="APPROVE") + request, preflight = _authorization(auth_id="human:other") + x["authorization_request"] = request + x["authorization_preflight"] = preflight + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "APPROVE principal does not match authorization identity" in result["reason"] + + +def test_human_approve_rejects_deterministic_controller_authorization(): + x = _inputs(principal="human:owner", action="APPROVE") + request, preflight = _authorization( + auth_class="DETERMINISTIC_CONTROLLER", + auth_id="controller:r4r1", + ) + x["authorization_request"] = request + x["authorization_preflight"] = preflight + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "APPROVE requires HUMAN authorization identity" in result["reason"] + + +def test_exact_replay_requires_anchored_readback_and_is_idempotent(): + x = _inputs() + first = evaluate_person_twin_authority_replay(**x) + assert first["decision"] == "ALLOW" and first["replay_status"] == "NEW" + x["replay_store"].seed(first) + exact = evaluate_person_twin_authority_replay(**copy.deepcopy(x)) + assert exact["decision"] == "ALLOW" + assert exact["reason"] == "EXACT_REPLAY_IDEMPOTENT" + assert exact["replay_status"] == "EXACT_IDEMPOTENT" + assert exact["replay_identity_hash"] == first["replay_identity_hash"] + assert exact["mutated"] is False + + +def test_divergent_authorization_under_same_stable_replay_key_is_denied(): + x = _inputs() + first = evaluate_person_twin_authority_replay(**x) + x["replay_store"].seed(first) + request, preflight = _authorization(auth_sha="7" * 64) + x["authorization_request"] = request + x["authorization_preflight"] = preflight + denied = evaluate_person_twin_authority_replay(**x) + assert denied["decision"] == "DENY" + assert denied["reason"] == "REPLAY_IDENTITY_MISMATCH" + assert denied["replay_status"] == "DIVERGENT_DENIED" + + +def test_rehashed_forged_durable_receipt_is_rejected_by_binding_rederivation(): + x = _inputs() + first = evaluate_person_twin_authority_replay(**x) + forged = copy.deepcopy(first) + forged["binding"]["principal_id"] = "attacker" + body = dict(forged) + body.pop("receipt_hash") + forged["receipt_hash"] = _hash(body) + x["replay_store"].receipts[first["replay_key"]] = _canonical_bytes(forged) + denied = evaluate_person_twin_authority_replay(**x) + assert denied["decision"] == "DENY" + assert "durable replay identity binding mismatch" in denied["reason"] + + +def test_durable_replay_byte_drift_is_denied(): + x = _inputs() + first = evaluate_person_twin_authority_replay(**x) + x["replay_store"].receipts[first["replay_key"]] = _canonical_bytes(first) + b"\n" + denied = evaluate_person_twin_authority_replay(**x) + assert denied["decision"] == "DENY" + assert "durable replay receipt bytes are not canonical" in denied["reason"] + + +def test_durable_replay_read_failure_fails_closed(): + x = _inputs() + x["replay_store"] = ExplodingReplayStore() + denied = evaluate_person_twin_authority_replay(**x) + assert denied["decision"] == "DENY" + assert "durable replay readback failed" in denied["reason"] + + +def test_deterministic_new_receipt_for_identical_inputs(): + x = _inputs() + first = evaluate_person_twin_authority_replay(**x) + second = evaluate_person_twin_authority_replay(**copy.deepcopy(x)) + assert first == second + + +def test_public_call_boundary_missing_and_unexpected_keys_fail_closed_without_typeerror(): + missing = _inputs() + missing.pop("policy") + result = evaluate_person_twin_authority_replay(**missing) + assert result["decision"] == "DENY" + assert "public input keys mismatch" in result["reason"] + assert "TypeError" not in result["reason"] + + unexpected = _inputs() + unexpected["unexpected"] = "x" + result = evaluate_person_twin_authority_replay(**unexpected) + assert result["decision"] == "DENY" + assert "public input keys mismatch" in result["reason"] + assert "TypeError" not in result["reason"] + + +def test_public_call_boundary_rejects_positional_and_non_string_scalars_without_raw_exception(): + result = evaluate_person_twin_authority_replay({}) + assert result["decision"] == "DENY" + assert "positional arguments are unsupported" in result["reason"] + x = _inputs() + x["action"] = {"READ": True} + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "action must be non-empty" in result["reason"] + + +@pytest.mark.parametrize("target", ["admission_receipt", "current_pointer"]) +@pytest.mark.parametrize("field,value", [("can_trade", None), ("capital_permission", None)]) +def test_admission_and_pointer_safety_fields_must_be_explicit(target, field, value): + x = _inputs() + x[target] = copy.deepcopy(x[target]) + x[target][field] = value + if target == "current_pointer": + x["admission_store"].pointer = _canonical_bytes(x[target]) + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "authority escalation" in result["reason"] + + +@pytest.mark.parametrize("mutation", ["declared", "reason", "legacy_fallback"]) +def test_current_session_exact_contour_rejects_forged_declaration_reason_and_legacy(mutation): + x = _inputs() + x["current_session"] = copy.deepcopy(x["current_session"]) + if mutation == "declared": + x["current_session"]["declared"] = False + elif mutation == "reason": + x["current_session"]["reason"] = "forged" + else: + x["current_session"]["effects"]["legacy_fallback"] = True + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + + +@pytest.mark.parametrize( + "target", + ["source_record", "admission_receipt", "current_pointer", "policy", "current_session", "authorization_request", "authorization_preflight"], +) +def test_closed_trust_anchors_reject_unexpected_top_level_keys(target): + x = _inputs() + x[target] = copy.deepcopy(x[target]) + x[target]["unexpected"] = True + if target == "current_pointer": + x["admission_store"].pointer = _canonical_bytes(x[target]) + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "keys mismatch" in result["reason"] + + +@pytest.mark.parametrize("field", ["project_id", "packet_id", "operation_count"]) +def test_preflight_request_cross_binding_rejects_divergence(field): + x = _inputs() + x["authorization_preflight"] = copy.deepcopy(x["authorization_preflight"]) + if field == "project_id": + x["authorization_preflight"][field] = "project:other" + elif field == "packet_id": + x["authorization_preflight"][field] = "r4p-" + "0" * 40 + else: + x["authorization_preflight"][field] = 2 + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + + +@pytest.mark.parametrize("field,value", [("id", "human:other"), ("ref", "synthetic://other"), ("class", "DETERMINISTIC_CONTROLLER")]) +def test_preflight_authorization_identity_is_packet_integrity_bound(field, value): + x = _inputs() + x["authorization_preflight"] = copy.deepcopy(x["authorization_preflight"]) + x["authorization_preflight"]["authorization"][field] = value + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "authorization packet identity mismatch" in result["reason"] or "APPROVE" in result["reason"] + + +@pytest.mark.parametrize("value", [True, 1.5]) +def test_policy_max_delegation_depth_rejects_boolean_and_fractional(value): + x = _inputs() + x["policy"] = {**x["policy"], "max_delegation_depth": value} + result = evaluate_person_twin_authority_replay(**x) + assert result["decision"] == "DENY" + assert "max_delegation_depth must be positive integer" in result["reason"] From 4d8cf5d3028e11d4990b836e9e15ee881f446e6e Mon Sep 17 00:00:00 2001 From: bitmaster162 <115934939+bitmaster162@users.noreply.github.com> Date: Mon, 12 Oct 2026 12:15:25 -0700 Subject: [PATCH 2/4] feat(person-twin): gate R21H FAST evidence with controller-bound reads Compose R2 and R4 with a pinned CURRENT snapshot and repeat validation before model prompt construction. Preserve generic runtime/DeepLite behavior and reject protected fallback and nested dispatch. Publish the user-approved tree 1d97123f6b08fde74754117128eabb1dbb75ab7a from reviewed local candidate f87f9e496088347b314cf731b4170a9ad8d51b8d. Local synthetic qualification: 154 scoped tests passed with zero socket events observed. Draft review scope only. Trusted host/provider/intent inputs remain a precondition; no production admission, real personal-data ingestion or native H3 qualification. --- continuityos/person_twin_r21h_evidence.py | 245 +++++++ continuityos/sovereign_twin_runtime.py | 123 ++++ docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md | 158 +++++ tests/test_person_twin_r21h_evidence.py | 685 +++++++++++++++++++ 4 files changed, 1211 insertions(+) create mode 100644 continuityos/person_twin_r21h_evidence.py create mode 100644 docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md create mode 100644 tests/test_person_twin_r21h_evidence.py diff --git a/continuityos/person_twin_r21h_evidence.py b/continuityos/person_twin_r21h_evidence.py new file mode 100644 index 00000000..9c2ec4c0 --- /dev/null +++ b/continuityos/person_twin_r21h_evidence.py @@ -0,0 +1,245 @@ +"""Optional, host-configured read boundary for one R3 CURRENT Person snapshot. + +The host supplies the intent independently of a fresh R4 context provider. Neither +query text nor memory metadata supplies authority. The provider is a trusted host +integration boundary: synthetic tests do not prove live OS/session authority. +This adapter reads existing stores and grants no model egress or execution rights. +""" +from __future__ import annotations + +from dataclasses import asdict, dataclass, fields +from datetime import datetime +import hashlib +import json +import math +from typing import Any, Mapping, Protocol, Sequence + +from .person_twin_authority_replay import ( + PUBLIC_INPUT_KEYS, + evaluate_person_twin_authority_replay, +) +from .person_twin_privacy_provenance import filter_person_twin_records +from .sovereign_twin_runtime_r21b import TwinEvidence + + +class PersonTwinEvidenceError(ValueError): + """Fail-closed, fixed-code error containing no protected input or anchors.""" + + +def _canonical_hash(value: Any) -> str: + return hashlib.sha256(json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), + allow_nan=False, + ).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class PersonReadIntent: + """Independent controller pins for this exact query, session and selected row.""" + + query_sha256: str + current_session_sha256: str + request_id: str + packet_id: str + memory_id: int + namespace: str + person_record_id: str + source_record_id: str + principal_id: str + scope: str + privacy_class: str + purpose: str + + def __post_init__(self) -> None: + if type(self.memory_id) is not int or self.memory_id < 1: + raise PersonTwinEvidenceError("PERSON_INTENT_MEMORY_ID_INVALID") + for field in fields(self): + if field.name == "memory_id": + continue + value = getattr(self, field.name) + if not isinstance(value, str) or not value.strip(): + raise PersonTwinEvidenceError("PERSON_INTENT_FIELD_INVALID") + for value in (self.query_sha256, self.current_session_sha256): + if len(value) != 64 or any(char not in "0123456789abcdef" for char in value): + raise PersonTwinEvidenceError("PERSON_INTENT_HASH_INVALID") + + +class PersonReadContextProvider(Protocol): + def read_context(self) -> Mapping[str, Any]: + """Return fresh exact R4 inputs, including independent live host anchors.""" + ... + + +class MemoryReadStore(Protocol): + def get(self, rid: int) -> Any: ... + + +def _unique_json_object(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise PersonTwinEvidenceError("PERSON_MEMORY_META_INVALID") + result[key] = value + return result + + +def _require_current_meta(raw: Any, at: str) -> None: + if not isinstance(raw, str): + raise PersonTwinEvidenceError("PERSON_MEMORY_META_INVALID") + meta = json.loads(raw, object_pairs_hook=_unique_json_object) + if not isinstance(meta, dict): + raise PersonTwinEvidenceError("PERSON_MEMORY_META_INVALID") + # Metadata can veto a row; it never grants READ or supplies R4 anchors. + json.dumps(meta, allow_nan=False) + if "superseded_by" in meta or meta.get("valid_to") is not None: + raise PersonTwinEvidenceError("PERSON_MEMORY_NOT_CURRENT") + if "valid_from" in meta: + value = meta["valid_from"] + if type(value) not in (int, float) or not math.isfinite(value): + raise PersonTwinEvidenceError("PERSON_MEMORY_VALID_FROM_INVALID") + current = datetime.fromisoformat(at[:-1] + "+00:00" if at.endswith("Z") else at) + if current.tzinfo is None or value > current.timestamp(): + raise PersonTwinEvidenceError("PERSON_MEMORY_NOT_YET_CURRENT") + + +class PersonTwinR21HEvidenceAdapter: + """R2 filter + R4 READ + exact existing-memory projection, without retrieval. + + A new query/session needs independently updated intent pins. Every prepare + and prompt validation calls the provider and R4 again; ALLOW is never cached. + The one source payload may contain the host's goal/decision/constraint/next + step snapshot. Its exact payload.text is preserved, not generated here. + """ + + def __init__(self, intent: PersonReadIntent, provider: PersonReadContextProvider): + if type(intent) is not PersonReadIntent or not callable(getattr(provider, "read_context", None)): + raise PersonTwinEvidenceError("PERSON_ADAPTER_CONFIGURATION_INVALID") + self._intent = intent + self._provider = provider + + def _read( + self, query: str, store: MemoryReadStore, + ) -> tuple[tuple[TwinEvidence, ...], dict[str, str]]: + try: + intent = self._intent + if not isinstance(query, str) or hashlib.sha256(query.encode("utf-8")).hexdigest() != intent.query_sha256: + raise PersonTwinEvidenceError("PERSON_QUERY_BINDING_MISMATCH") + try: + raw = self._provider.read_context() + except Exception: + raise PersonTwinEvidenceError("PERSON_HOST_CONTEXT_READ_FAILED") from None + try: + if not isinstance(raw, Mapping) or set(raw) != set(PUBLIC_INPUT_KEYS): + raise ValueError("invalid context shape") + # Freeze JSON inputs, preserving only the two read-store interfaces. + stores = {name: raw[name] for name in ("admission_store", "replay_store")} + context = json.loads(json.dumps( + {key: value for key, value in raw.items() if key not in stores}, + ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, + )) + context.update(stores) + except Exception: + raise PersonTwinEvidenceError("PERSON_HOST_CONTEXT_SHAPE_INVALID") from None + pins = { + "action": "READ", "principal_id": intent.principal_id, + "requested_scope": intent.scope, "requested_privacy_class": intent.privacy_class, + "purpose": intent.purpose, + } + if any(context[key] != value for key, value in pins.items()): + raise PersonTwinEvidenceError("PERSON_READ_INTENT_MISMATCH") + if ( + _canonical_hash(context["current_session"]) != intent.current_session_sha256 + or context["authorization_request"]["request_id"] != intent.request_id + or context["authorization_preflight"]["packet_id"] != intent.packet_id + ): + raise PersonTwinEvidenceError("PERSON_SESSION_REQUEST_BINDING_MISMATCH") + record = context["person_record"] + source = context["source_record"] + if record["id"] != intent.person_record_id or source["id"] != intent.source_record_id: + raise PersonTwinEvidenceError("PERSON_RECORD_BINDING_MISMATCH") + if record["deleted"] is not False or source["deleted"] is not False: + raise PersonTwinEvidenceError("PERSON_DELETED_RECORD_DENIED") + consent = context["consent_receipt"] + filtered = filter_person_twin_records( + [record], context["identity"], authorized_scopes=[intent.scope], + source_records={source["id"]: source}, + consent_receipts={consent["consent_receipt_id"]: consent}, + current_revocation_ledger=context["current_revocation_ledger"], + purpose=intent.purpose, at=context["at"], + ) + if len(filtered) != 1 or filtered[0] != record: + raise PersonTwinEvidenceError("PERSON_R2_RECORD_DENIED") + decision = evaluate_person_twin_authority_replay(**context) + if decision.get("decision") != "ALLOW" or decision.get("action") != "READ": + raise PersonTwinEvidenceError("PERSON_R4_READ_DENIED") + expected_binding = { + "current_session_hash": intent.current_session_sha256, + "request_id": intent.request_id, "packet_id": intent.packet_id, + "person_record_id": intent.person_record_id, + "source_record_id": intent.source_record_id, + "principal_id": intent.principal_id, "scope": intent.scope, + "privacy_class": intent.privacy_class, "purpose": intent.purpose, + "action": "READ", + } + binding = decision.get("binding", {}) + if any(binding.get(key) != value for key, value in expected_binding.items()): + raise PersonTwinEvidenceError("PERSON_R4_BINDING_MISMATCH") + payload = source["payload"] + text = payload.get("text") if isinstance(payload, dict) else None + if not isinstance(text, str) or not text.strip(): + raise PersonTwinEvidenceError("PERSON_SOURCE_TEXT_INVALID") + if _canonical_hash(payload) != source["content_hash"] or record["payload"] != payload: + raise PersonTwinEvidenceError("PERSON_SOURCE_PAYLOAD_MISMATCH") + # No bulk recall, embeddings, metadata-derived authority, or hidden-store creation. + try: + selected = store.get(intent.memory_id) + row = None if selected is None else { + key: selected[key] for key in ("id", "namespace", "text", "meta") + } + except Exception: + raise PersonTwinEvidenceError("PERSON_MEMORY_READ_FAILED") from None + if row is None or type(row["id"]) is not int or row["id"] != intent.memory_id: + raise PersonTwinEvidenceError("PERSON_MEMORY_ID_MISMATCH") + if row["namespace"] != intent.namespace or row["text"] != text: + raise PersonTwinEvidenceError("PERSON_MEMORY_SOURCE_BINDING_MISMATCH") + _require_current_meta(row["meta"], context["at"]) + evidence = (TwinEvidence( + id=intent.memory_id, namespace=intent.namespace, text=text, + # Exact row selection marker, not a measured confidence score. + score=1.0, why="R2_PROVENANCE_AND_R4_READ_CURRENT", + ),) + return evidence, { + "intent_sha256": _canonical_hash(asdict(intent)), + "r4_receipt_hash": decision["receipt_hash"], + "evidence_sha256": _canonical_hash([asdict(row) for row in evidence]), + } + except PersonTwinEvidenceError: + raise + except Exception: + # R2/R4/provider/store errors can contain protected data or identifiers. + raise PersonTwinEvidenceError("PERSON_EVIDENCE_VALIDATION_FAILED") from None + + def prepare(self, query: str, store: MemoryReadStore) -> tuple[TwinEvidence, ...]: + evidence, _ = self._read(query, store) + return evidence + + def validate_for_prompt( + self, query: str, store: MemoryReadStore, evidence: Sequence[TwinEvidence], + ) -> dict[str, Any]: + if type(evidence) is not tuple or len(evidence) != 1 or type(evidence[0]) is not TwinEvidence: + raise PersonTwinEvidenceError("PERSON_PROMPT_EVIDENCE_SHAPE_INVALID") + current, commitments = self._read(query, store) + if evidence != current: + raise PersonTwinEvidenceError("PERSON_PROMPT_EVIDENCE_CHANGED") + receipt = { + "schema": "continuityos.person-twin.r21h-evidence/v1", + "decision": "ALLOW", "action": "READ", "evidence_count": 1, + "memory_refs": [f"mem:{self._intent.memory_id}"], + "prompt_boundary_revalidated": True, + "production_admission_status": "NOT_PRODUCTION_ADMITTED", + "execution_authority": "NONE", "can_execute": False, + "can_trade": False, "capital_permission": "DENY", + **commitments, + } + receipt["receipt_hash"] = _canonical_hash(receipt) + return receipt diff --git a/continuityos/sovereign_twin_runtime.py b/continuityos/sovereign_twin_runtime.py index e100ae6a..2c660532 100644 --- a/continuityos/sovereign_twin_runtime.py +++ b/continuityos/sovereign_twin_runtime.py @@ -15,9 +15,15 @@ """ from __future__ import annotations +from dataclasses import replace as _replace +from threading import get_ident as _get_ident from typing import Any from . import sovereign_twin_runtime_r21g as _r21g +from .person_twin_r21h_evidence import ( + PersonTwinEvidenceError, + PersonTwinR21HEvidenceAdapter, +) from .sovereign_twin_runtime_r21g import * # noqa: F401,F403 for _legacy_name, _legacy_value in vars(_r21g).items(): @@ -37,9 +43,92 @@ ) +class _PersonPromptBoundary: + """Keep historical class-static formatting, guard protected instance calls.""" + + def __get__(self, instance, owner): + if instance is None: + return _r21g.SovereignTwinRuntime._system_prompt + return instance._person_system_prompt + + +class _PersonDeepBoundary: + """Preserve inherited DEEP identity while denying protected instance entry.""" + + def __get__(self, instance, owner): + original = _r21g.SovereignTwinRuntime._ask_deep + if instance is None: + return original + if getattr(instance, "_person_evidence_adapter", None) is None: + return original.__get__(instance, owner) + return instance._person_deep_denied + + class SovereignTwinRuntime(_r21g.SovereignTwinRuntime): """R21G runtime plus fail-closed embedding-aware FAST startup/request ordering.""" + _system_prompt = _PersonPromptBoundary() + _ask_deep = _PersonDeepBoundary() + + def __init__( + self, memory_db: str, *, client: LmStudioClient | None = None, + recall_k: int = 8, profiles: Mapping[str, LocalModelProfile] | None = None, + embedding_model: str = DEFAULT_EMBEDDING_MODEL, + person_evidence_adapter: PersonTwinR21HEvidenceAdapter | None = None, + ): + if person_evidence_adapter is not None and type(person_evidence_adapter) is not PersonTwinR21HEvidenceAdapter: + raise PersonTwinEvidenceError("PERSON_ADAPTER_CONFIGURATION_INVALID") + super().__init__(memory_db, client=client, recall_k=recall_k, + profiles=profiles, embedding_model=embedding_model) + self._person_evidence_adapter = person_evidence_adapter + self._person_active_request = None + + def _person_request(self) -> dict[str, Any]: + active = getattr(self, "_person_active_request", None) + if active is None or active["owner_thread"] != _get_ident() or active["adapter"] is not getattr(self, "_person_evidence_adapter", None): + raise PersonTwinEvidenceError("PERSON_ACTIVE_ASK_REQUIRED") + return active + + def _person_deep_denied(self, *args: Any, **kwargs: Any) -> TwinAnswer: + raise PersonTwinEvidenceError("PERSON_PROTECTED_DEEP_UNSUPPORTED") + + def evidence(self, query: str) -> tuple[TwinEvidence, ...]: + adapter = getattr(self, "_person_evidence_adapter", None) + if adapter is None: + return super().evidence(query) + with self._model_lock: + active = self._person_request() + if query != active["query"]: + raise PersonTwinEvidenceError("PERSON_ACTIVE_QUERY_MISMATCH") + evidence = adapter.prepare(query, self.memory.store) + active["evidence"] = evidence + return evidence + + def _person_system_prompt(self, evidence: Sequence[TwinEvidence]) -> str: + adapter = getattr(self, "_person_evidence_adapter", None) + if adapter is None: + return super()._system_prompt(evidence) + with self._model_lock: + active = self._person_request() + if evidence is not active["evidence"]: + raise PersonTwinEvidenceError("PERSON_FOREIGN_PROMPT_EVIDENCE") + active["receipt"] = adapter.validate_for_prompt( + active["query"], self.memory.store, evidence, + ) + return super()._system_prompt(evidence) + + def _ask_fast_cold_streaming( + self, query: str, *, profile: LocalModelProfile, evidence: Any, + ) -> TwinAnswer: + if getattr(self, "_person_evidence_adapter", None) is None: + return super()._ask_fast_cold_streaming(query, profile=profile, evidence=evidence) + with self._model_lock: + active = self._person_request() + if query != active["query"] or evidence is not active["evidence"] or profile != self.profiles["fast"]: + raise PersonTwinEvidenceError("PERSON_FOREIGN_STREAM_REQUEST") + self._require_fast_streaming_transport() + return super()._ask_fast_cold_streaming(query, profile=profile, evidence=evidence) + def _require_fast_streaming_transport(self): stream_chat = getattr( self.client, @@ -73,6 +162,8 @@ def prewarm_fast_startup(self) -> dict[str, Any]: first prove the embedding endpoint with a fixed query that bypasses canonical memory, then make FAST the final startup acquisition. """ + if getattr(self, "_person_evidence_adapter", None) is not None: + raise PersonTwinEvidenceError("PERSON_PROTECTED_PREWARM_UNSUPPORTED") with self._model_lock: profile = self.profiles["fast"] @@ -168,7 +259,39 @@ def prewarm_fast_startup(self) -> dict[str, Any]: } def ask(self, query: str, *, mode: str = "fast") -> TwinAnswer: + adapter = getattr(self, "_person_evidence_adapter", None) + if adapter is None: + return self._ask_r21h(query, mode=mode) + if mode != "fast" or not callable(getattr(self.client, "chat_fast_streaming_jit_reconciled", None)): + raise PersonTwinEvidenceError("PERSON_PROTECTED_FAST_STREAMING_ONLY") + with self._model_lock: + if getattr(self, "_person_active_request", None) is not None: + raise PersonTwinEvidenceError("PERSON_NESTED_ASK_DENIED") + self._person_active_request = { + "adapter": adapter, "query": query, "owner_thread": _get_ident(), + "evidence": None, "receipt": None, "dispatcher_entered": False, + } + try: + answer = self._ask_r21h(query, mode=mode) + receipt = self._person_request()["receipt"] + if receipt is None: + raise PersonTwinEvidenceError("PERSON_PROMPT_REVALIDATION_MISSING") + return _replace(answer, stats={**answer.stats, "person_evidence": receipt}) + finally: + self._person_active_request = None + + def _ask_r21h(self, query: str, *, mode: str = "fast") -> TwinAnswer: """Answer FAST after evidence/reprobe, preserving old custom-client fallback.""" + if getattr(self, "_person_evidence_adapter", None) is not None: + with self._model_lock: + active = self._person_request() + if query != active["query"]: + raise PersonTwinEvidenceError("PERSON_ACTIVE_QUERY_MISMATCH") + if mode != "fast" or not callable(getattr(self.client, "chat_fast_streaming_jit_reconciled", None)): + raise PersonTwinEvidenceError("PERSON_PROTECTED_FAST_STREAMING_ONLY") + if active["dispatcher_entered"]: + raise PersonTwinEvidenceError("PERSON_NESTED_DISPATCH_DENIED") + active["dispatcher_entered"] = True if mode != "fast": return super().ask(query, mode=mode) diff --git a/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md new file mode 100644 index 00000000..0627f6f5 --- /dev/null +++ b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md @@ -0,0 +1,158 @@ +# Person evidence in R21H — R1 integration candidate + +Status: local synthetic integration candidate; `NOT_PRODUCTION_ADMITTED`. + +## Why this change exists + +R21H already answers from ContinuityOS memory. Its generic evidence path converts +ranked memory rows into `TwinEvidence`, without composing the Person admission, +privacy/provenance and authority/replay contracts. This candidate adds an explicit +controller-configured path for one admitted CURRENT context snapshot. + +The snapshot can contain a goal, current decision, constraint and next step. +It keeps a real SQLite `mem:` reference. The existing Memory store and R21H +model transport are reused; no second canonical store or new retrieval engine +is introduced. + +## Source composition + +- Protected master baseline: `719ed32a64301fc8083a5aad0c4ed7bed316040e`. +- Baseline tree: `0406a3020f86e405b7934ea50764d0f7e64ee63d`. +- The R4 module and its original tests are imported byte-for-byte from + PR #150 head `4515bea19978b522328607deb493a55234ef7d4c`. +- R1, R2 and R3 remain the existing master implementations. +- Historical R21B–R21G overlays remain unchanged. + +Importing R4 into this candidate does not merge PR #150 or grant pilot admission. + +## Request boundary + +An explicit `PersonReadIntent` pins the question hash, current-session hash, +authorization request and packet, memory ID and namespace, Person and source +record IDs, principal, scope, privacy class and purpose. A separate host-supplied +provider supplies the current R4 context. Neither the question nor memory metadata +is used to construct authority. + +The adapter composes R2 privacy/provenance filtering and the R4 READ decision +before fetching the exact SQLite row. It binds that row to the validated source +payload text and namespace. Metadata can veto a retired or malformed row; it +cannot grant access. Only the verified projection is converted to model evidence. + +Immediately before prompt construction, the protected runtime repeats the context +and row checks and compares the new projection with the pending evidence. A +missing binding, changed session, revoked consent, stale admission, mismatched +source text or changed evidence stops the request. Adapter validation errors use +fixed codes; raw R4 failure details and authority inputs are not copied into +model prompts. Existing model-transport error behavior is retained. + +Each successful answer carries a bounded decision receipt in +`answer.stats['person_evidence']`. This is an unsigned integration observation, +not a signed owner grant or an external trust attestation. + +## Deliberately limited runtime surface + +Protected mode covers R21H FAST with the streaming transport contract, including +warm chat and cold streaming acquisition. It rejects DEEP, legacy fallback, +startup prewarm and direct protected helper use outside the active request. +The request frame is scoped to the runtime lock and is cleared after completion +or failure. + +The generic runtime remains available when no adapter is configured. Its +compatibility behavior does not become Person-qualified by this change. +The class-level prompt formatter used by generic DeepLite is retained. +No CLI, HTTP, MCP or DeepLite option enables the new protected mode. + +## What the synthetic proof can establish + +The new tests use temporary synthetic SQLite records, the real R2/R3/R4 contracts +and recording fake model clients. Two fresh runtime instances receive the same +admitted snapshot through separate fixture sessions and authorization chains. +The test inspects the actual prompts and memory IDs delivered to those clients. +Old and unrelated records are excluded from that projection. + +Negative cases exercise identity, consent, scope, session, query, row-currentness +and final-check failures, plus attempts to enter an unsupported runtime path. +Existing FAST and generic runtime tests cover compatibility. + +This is a two-session transport/composition proof. It is not a comparison of two +live models, a scientific Person Twin benchmark, an end-user retention metric, +or a completed backup/restore or coding-tool interception demonstration. + +## Trust and release limits + +The host, intent issuer and context provider remain trusted integration inputs. +The adapter checks consistency and readback; it does not authenticate those +inputs independently. It does not establish trusted time, owner key custody, +model-sink egress consent or native same-binding broker identity. + +Repeated checks reduce the specific stale-evidence window exercised by the tests; +they are not an atomic transaction across admission, SQLite and the model sink. +R4 READ decisions do not reserve durable disclosure intent or prove delivery. +R4 still authorizes no execution, deployment or trading. + +H3 C0 remains `UNVERIFIED/HOLD`; native T11/T12 remain `UNEXECUTED`. +The successful G3 object-compilation workflow is separate evidence. +The twelve P3-P1 gate families and owner signoff remain separate acceptance work. + +Before a real personal pilot, a separately reviewed integration must supply +authenticated authority and currentness inputs, approved database custody and +egress, and coverage for every enabled route. This candidate does not enable +that pilot. + +## Validation record + +Observed 2026-10-12 19:03:36–19:03:38 UTC (2026-10-13 Asia/Bangkok): + +| Test group | Result | +|---|---:| +| New Person/R21H composition and negative cases | 61 passed | +| Unchanged R4 authority/replay tests | 50 passed | +| Existing R21G/R21H, generic runtime and DeepLite compatibility | 43 passed | +| Total | 154 passed; 0 failed; 0 skipped | + +Python 3.12.14; pytest 9.1.1; plugin autoload/cache/bytecode disabled. +All `socket.*` audit events were denied; zero were observed. Tested source hashes +were identical before and after the run. This was a scoped local run, not the +entire repository suite or Windows qualification. + +The independent source review found and the regression tests reproduced direct +and nested DEEP dispatcher entry before rejection, and nested FAST dispatch. +The revised runtime rejects these before the affected client operations and +preserves the outer request. The final reviewed source has no remaining blockers +within the trusted-host integration scope described above. + +| Tested file | SHA-256 | +|---|---| +| New adapter | `a918dde064962e78aaa60b9e5b8dd28d93034afd7a5e6a1a2be3a3984ddf89ec` | +| R21H runtime | `ca94b260b256125121e416d992504bf71bd0e8af8f2706084c5e3752c224cfa4` | +| New adapter tests | `fe7220f15bb5e58cd57cf40ec730465b03af50d3de970e8b9a968a3bd5b8e817` | + +The unsigned local run receipt has SHA-256 +`6a1f75c3c6bf18fac2745f479cda599ac0e9d22d9f628f6945744775277b3b85`. +It records the 154 test outcomes and exact inputs; it is not an authority grant. + +Reproduce the scoped run from the repository in an environment with pytest: + +```bash +PYTHONDONTWRITEBYTECODE=1 PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python - <<'PY' +import sys + +def deny_socket(event, args): + if event.startswith('socket.'): + raise RuntimeError('PERSON_R21H_VALIDATION_NETWORK_DENIED:' + event) + +sys.addaudithook(deny_socket) +import pytest +raise SystemExit(pytest.main([ + '-q', '-p', 'no:cacheprovider', + 'tests/test_person_twin_authority_replay.py', + 'tests/test_person_twin_r21h_evidence.py', + 'tests/test_sovereign_twin_runtime.py', + 'tests/test_sovereign_twin_r21g_fast_chat_coupled_streaming_jit.py', + 'tests/test_sovereign_twin_r21h_fast_startup_streaming_jit_prewarm.py', + 'tests/test_sovereign_twin_r21h_legacy_client_compatibility.py', + 'tests/test_sovereign_twin_r21h_cached_success_revalidation.py', + 'tests/test_sovereign_twin_deep_lite.py', +])) +PY +``` diff --git a/tests/test_person_twin_r21h_evidence.py b/tests/test_person_twin_r21h_evidence.py new file mode 100644 index 00000000..731c8ccb --- /dev/null +++ b/tests/test_person_twin_r21h_evidence.py @@ -0,0 +1,685 @@ +"""Synthetic two-session evidence transfer through the existing R21H runtime. + +SQLite and the R2/R3/R4 decisions are real; model calls only capture arguments. +The admission/session/controller inputs are synthetic fixtures, not live host +authority. These tests do not claim model understanding, full source lineage, +native H3 readiness, personal-data ingestion, or production admission. +""" +from __future__ import annotations + +import copy +from dataclasses import replace +import hashlib +import json +import sqlite3 +from threading import Event, Thread +import traceback +from types import SimpleNamespace + +import pytest + +from continuityos.memory import Memory +from continuityos.person_twin_admission_contracts import ( + build_consent_revocation_entry, + build_consent_revocation_ledger, +) +from continuityos.person_twin_admission_coordinator import PersonTwinAdmissionCoordinator +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, + PersonTwinEvidenceError, + PersonTwinR21HEvidenceAdapter, +) +from continuityos.sovereign_twin_runtime import ( + DEFAULT_PROFILES, + LocalChatResult, + SovereignTwinRuntime, + TwinEvidence, +) + +# Reuse the unchanged R4 fixture contract, including real R3 admission writes. +from test_person_twin_authority_replay import ( + Store as AdmissionStore, + _hash, + _inputs, + _policy, +) + + +QUERY = "Restore the current synthetic project context." +NAMESPACE = "synthetic-person-context" +CURRENT_TEXT = ( + "Goal: transfer the approved context to a second session.\n" + "Decision: use the read-only R21H evidence adapter.\n" + "Constraint: synthetic data only; no external effects.\n" + "Next step: inspect the captured mem reference and current snapshot." +) +OLD_TEXT = "OLD_DECISION_SENTINEL: use the retired prototype." +FOREIGN_TEXT = "FOREIGN_PRIVATE_SENTINEL: another person's context." +SECRET = "SYNTHETIC_PRIVATE_ERROR_MARKER" + + +def _copy_context(context): + # Independent store readbacks must remain live, not copied snapshots. + return { + key: value if key in {"admission_store", "replay_store"} else copy.deepcopy(value) + for key, value in context.items() + } + + +def _rebind_request(context, label): + """Create a different, internally valid synthetic request and packet.""" + request = context["authorization_request"] + preflight = context["authorization_preflight"] + request["proposal_id"] = f"proposal:r21h:{label}" + request["proposal_file_sha256"] = hashlib.sha256(label.encode()).hexdigest() + request["request_id"] = "omar-" + _hash({ + "proposal_id": request["proposal_id"], + "proposal_file_sha256": request["proposal_file_sha256"], + "project_id": request["project_id"], + "base": request["expected_base"], + "operation_count": request["operation_count"], + })[:40] + preflight["proposal_id"] = request["proposal_id"] + preflight["proposal_file_sha256"] = request["proposal_file_sha256"] + preflight["packet_id"] = "r4p-" + _hash({ + "request_id": request["request_id"], + "project_id": request["project_id"], + "proposal_id": request["proposal_id"], + "proposal_file_sha256": request["proposal_file_sha256"], + "expected_base": request["expected_base"], + "operation_count": request["operation_count"], + "authorization_file_sha256": preflight["authorization_file_sha256"], + "authorization": preflight["authorization"], + })[:40] + + +def _rebind_session(context, label): + session = context["current_session"] + session["challenge_id"] = f"challenge:r21h:{label}" + session["challenge_sha256"] = hashlib.sha256(f"challenge:{label}".encode()).hexdigest() + session["ack_sha256"] = hashlib.sha256(f"ack:{label}".encode()).hexdigest() + + +def _intent(context, memory_id, *, query=QUERY): + # These pins are constructed by the independent synthetic controller before + # the provider is allowed to drift, never inferred from the memory row. + return PersonReadIntent( + query_sha256=hashlib.sha256(query.encode()).hexdigest(), + current_session_sha256=_hash(context["current_session"]), + request_id=context["authorization_request"]["request_id"], + packet_id=context["authorization_preflight"]["packet_id"], + memory_id=memory_id, + namespace=NAMESPACE, + person_record_id=context["person_record"]["id"], + source_record_id=context["source_record"]["id"], + principal_id=context["principal_id"], + scope=context["requested_scope"], + privacy_class=context["requested_privacy_class"], + purpose=context["purpose"], + ) + + +class FreshProvider: + def __init__(self, context): + self.context = _copy_context(context) + self.reads = 0 + self.on_read = None + + def read_context(self): + self.reads += 1 + callback, self.on_read = self.on_read, None + if callback is not None: + callback() + return _copy_context(self.context) + + +class CapturingModel: + """No inference or network: record the real runtime's model arguments.""" + + def __init__(self, profile, *, cold=False): + self.profile = profile + self.instance_id = f"synthetic-instance:{profile.model}" + self.loaded = not cold + self.events = [] + self.calls = [] + self.on_models = None + + def models(self): + self.events.append("models") + callback, self.on_models = self.on_models, None + if callback is not None: + callback() + 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 embed(self, *args, **kwargs): + self.events.append("embed") + raise AssertionError("exact protected evidence must not use embedding retrieval") + + def load(self, **kwargs): + self.events.append("load") + raise AssertionError("protected evidence must not use legacy loading") + + def unload(self, *args, **kwargs): + self.events.append("unload") + raise AssertionError("protected FAST must not enter DEEP release") + + def load_fast_for_acquisition(self, **kwargs): + self.events.append("explicit_load") + raise AssertionError("official FAST must not be explicitly loaded") + + def chat(self, **kwargs): + self.events.append("chat") + self.calls.append(kwargs) + return LocalChatResult( + text="Synthetic captured answer.", model_instance_id=self.instance_id, + stats={}, reasoning=None, + ) + + def chat_fast_streaming_jit_reconciled(self, **kwargs): + self.events.append("stream") + self.calls.append(kwargs) + self.loaded = True + kwargs["on_model_load_end"](self.instance_id, 0.0) + return ( + LocalChatResult( + text="Synthetic captured streaming answer.", + model_instance_id=self.instance_id, stats={}, reasoning=None, + ), + { + "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"), + }, + ) + + +@pytest.fixture +def snapshot(tmp_path): + context = _inputs() + source = copy.deepcopy(context["source_record"]) + source["payload"] = {"title": "Synthetic current context", "text": CURRENT_TEXT} + source["content_hash"] = _hash(source["payload"]) + # This finite seam selects one admitted snapshot. It does not claim to prove + # semantic source lineage from a single R2 candidate. + assert source["supersedes"] is None + record = build_person_twin_provenance( + context["identity"], context["consent_receipt"], + context["current_revocation_ledger"], source, + admission_session_id="admission:r21h:synthetic", + purpose="person-memory", at="2026-08-25T11:00:00Z", + ) + admission_store = AdmissionStore() + coordinator = PersonTwinAdmissionCoordinator( + admission_store, session_id=record["admission_session_id"], + identity=context["identity"], consent_receipt=context["consent_receipt"], + source_record=source, candidate_record=record, purpose="person-memory", + ) + ledger = context["current_revocation_ledger"] + assert coordinator.validate(revocation_ledger=ledger, at="2026-08-25T11:10:00Z")["state"] == "VALIDATED" + assert coordinator.commit_not_current(current_revocation_ledger=ledger, at="2026-08-25T11:20:00Z")["state"] == "COMMITTED_NOT_CURRENT" + assert coordinator.postverify(current_revocation_ledger=ledger, at="2026-08-25T11:30:00Z")["state"] == "POSTVERIFY_PASS" + admission = coordinator.promote_current(current_revocation_ledger=ledger, at="2026-08-25T11:40:00Z") + assert admission["state"] == "CURRENT" + context.update( + source_record=source, person_record=record, admission_receipt=admission, + admission_store=admission_store, current_pointer=json.loads(admission_store.pointer), + ) + assert evaluate_person_twin_authority_replay(**context)["decision"] == "ALLOW" + + db = tmp_path / "synthetic-memory.db" + writer = Memory(str(db), embedder=lambda _: [1.0, 0.0, 0.0]) + try: + old_id = writer.remember(OLD_TEXT, namespace=NAMESPACE) + foreign_id = writer.remember(FOREIGN_TEXT, namespace="foreign-person") + current_id = writer.supersede(old_id, CURRENT_TEXT, namespace=NAMESPACE) + finally: + writer.store.con.close() + return SimpleNamespace( + context=context, db=db, current_id=current_id, + old_id=old_id, foreign_id=foreign_id, + ) + + +@pytest.fixture +def make_runtime(snapshot): + runtimes = [] + + def create(*, context=None, cold=False, label="A", intent=None): + provider = FreshProvider(snapshot.context if context is None else context) + pins = intent if intent is not None else _intent(provider.context, snapshot.current_id) + adapter = PersonTwinR21HEvidenceAdapter(pins, provider) + profile = replace(DEFAULT_PROFILES["fast"], model=f"synthetic-fast-{label}") + client = CapturingModel(profile, cold=cold) + runtime = SovereignTwinRuntime( + str(snapshot.db), client=client, + profiles={**DEFAULT_PROFILES, "fast": profile}, + person_evidence_adapter=adapter, + ) + runtimes.append(runtime) + return SimpleNamespace( + runtime=runtime, client=client, provider=provider, + adapter=adapter, intent=pins, profile=profile, + ) + + yield create + for runtime in runtimes: + runtime.close() + + +def _update_row(snapshot, **columns): + # Mutations are external synthetic fixture changes; runtime's own handle is + # genuinely SQLite read-only. Values are bound, never interpolated SQL. + assert set(columns) <= {"text", "namespace", "meta"} + with sqlite3.connect(snapshot.db) as connection: + assignments = ", ".join(f"{column} = ?" for column in columns) + connection.execute( + f"UPDATE items SET {assignments} WHERE id = ?", + (*columns.values(), snapshot.current_id), + ) + + +def _revoke_consent(context): + 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 R21H revocation", + ) + context["current_revocation_ledger"] = build_consent_revocation_ledger([entry]) + + +def _assert_no_generation(env): + assert env.client.calls == [] + assert not {"embed", "load", "unload", "explicit_load", "chat", "stream"}.intersection(env.client.events) + + +def _assert_current_prompt(answer, env, snapshot): + assert len(env.client.calls) == 1 + prompt = env.client.calls[0]["system_prompt"] + rows = json.loads(prompt.split("MEMORY_EVIDENCE_JSON:\n", 1)[1]) + assert [(row["ref"], row["namespace"], row["text"]) for row in rows] == [ + (f"mem:{snapshot.current_id}", NAMESPACE, CURRENT_TEXT), + ] + assert OLD_TEXT not in prompt and FOREIGN_TEXT not in prompt + assert env.client.calls[0]["input_text"] == QUERY + assert len(answer.evidence) == 1 + assert answer.evidence[0].id == snapshot.current_id + assert answer.evidence[0].text == CURRENT_TEXT + assert answer.execution_authority == "NONE" and answer.can_execute is False + receipt = answer.stats["person_evidence"] + assert receipt["decision"] == "ALLOW" and receipt["action"] == "READ" + assert receipt["memory_refs"] == [f"mem:{snapshot.current_id}"] + assert receipt["prompt_boundary_revalidated"] is True + assert receipt["production_admission_status"] == "NOT_PRODUCTION_ADMITTED" + assert receipt["execution_authority"] == "NONE" and receipt["can_execute"] is False + assert receipt["can_trade"] is False and receipt["capital_permission"] == "DENY" + public_receipt = json.dumps(receipt) + context = env.provider.context + for private in ( + CURRENT_TEXT, OLD_TEXT, FOREIGN_TEXT, + context["person_record"]["id"], context["source_record"]["id"], + context["authorization_request"]["request_id"], + context["authorization_preflight"]["packet_id"], + context["current_session"]["challenge_id"], + context["current_session"]["ack_sha256"], + context["consent_receipt"]["consent_receipt_id"], + ): + assert private not in public_receipt + return receipt + + +def test_two_fresh_sessions_transfer_same_current_snapshot_with_distinct_receipts(snapshot, make_runtime): + initial_db = hashlib.sha256(snapshot.db.read_bytes()).hexdigest() + store = snapshot.context["admission_store"] + replay_store = snapshot.context["replay_store"] + initial_authority = (store.pointer, dict(store.candidates), dict(replay_store.receipts)) + first = make_runtime(label="A") + second_context = _copy_context(snapshot.context) + _rebind_session(second_context, "B") + _rebind_request(second_context, "B") + # R4 is exercised independently; no ALLOW monkeypatch or fake admission. + first_r4 = evaluate_person_twin_authority_replay(**first.provider.context) + second_r4 = evaluate_person_twin_authority_replay(**second_context) + assert first_r4["decision"] == second_r4["decision"] == "ALLOW" + second = make_runtime(context=second_context, label="B", cold=True) + assert first.runtime is not second.runtime and first.client is not second.client + assert first.profile.model != second.profile.model + receipts = [] + for env in (first, second): + assert env.runtime.memory.store.con.execute("PRAGMA query_only").fetchone()[0] == 1 + answer = env.runtime.ask(QUERY) + receipts.append(_assert_current_prompt(answer, env, snapshot)) + assert env.provider.reads >= 2 + assert not {"embed", "load", "explicit_load"}.intersection(env.client.events) + assert "chat" in first.client.events and "stream" not in first.client.events + assert "stream" in second.client.events and "chat" not in second.client.events + assert receipts[0]["evidence_sha256"] == receipts[1]["evidence_sha256"] + for key in ("intent_sha256", "r4_receipt_hash", "receipt_hash"): + assert receipts[0][key] != receipts[1][key] + assert all(len(receipt[key]) == 64 for receipt in receipts) + assert receipts[0]["r4_receipt_hash"] == first_r4["receipt_hash"] + assert receipts[1]["r4_receipt_hash"] == second_r4["receipt_hash"] + assert (store.pointer, store.candidates, replay_store.receipts) == initial_authority + assert hashlib.sha256(snapshot.db.read_bytes()).hexdigest() == initial_db + + +def test_query_pin_rejects_different_query_before_context_or_model_read(make_runtime): + env = make_runtime() + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY + " Include extra context.") + assert env.provider.reads == 0 and env.client.events == [] + + +@pytest.mark.parametrize("drift", ["session", "request", "principal", "action"]) +def test_independently_valid_r4_input_cannot_replace_controller_pins(make_runtime, drift): + env = make_runtime() + context = env.provider.context + if drift == "session": + _rebind_session(context, "unapproved-session") + elif drift == "request": + _rebind_request(context, "unapproved-request") + elif drift == "principal": + context["principal_id"] = "human:owner" + else: + context["action"] = "PROPOSE" + assert evaluate_person_twin_authority_replay(**context)["decision"] == "ALLOW" + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + assert env.client.events == [] + + +@pytest.mark.parametrize("drift", ["scope", "privacy", "purpose", "consent", "expired", "policy", "candidate", "pointer"]) +def test_initial_authority_or_current_admission_drift_denies_before_model_read(make_runtime, drift): + env = make_runtime() + context = env.provider.context + if drift == "scope": + context["requested_scope"] = "person:foreign:private" + elif drift == "privacy": + context["requested_privacy_class"] = "COMPANY" + elif drift == "purpose": + context["purpose"] = "unapproved-purpose" + elif drift == "consent": + _revoke_consent(context) + elif drift == "expired": + context["at"] = "2026-08-27T12:50:00Z" + elif drift == "policy": + context["policy"] = _policy(context["identity"], revoked=True) + elif drift == "candidate": + context["admission_store"].candidates[context["person_record"]["id"]] = b"{}" + else: + context["admission_store"].pointer = b'{"state":"CURRENT","candidate_id":"foreign"}' + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + assert env.client.events == [] + + +@pytest.mark.parametrize("drift", ["text", "namespace", "missing"]) +def test_exact_sqlite_row_must_match_admitted_source(snapshot, make_runtime, drift): + env = make_runtime() + if drift == "missing": + with sqlite3.connect(snapshot.db) as connection: + connection.execute("DELETE FROM items WHERE id = ?", (snapshot.current_id,)) + elif drift == "text": + _update_row(snapshot, text=FOREIGN_TEXT) + else: + _update_row(snapshot, namespace="foreign-person") + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + assert env.client.events == [] + + +@pytest.mark.parametrize("raw_meta", [ + '{"superseded_by": 99}', '{"superseded_by": null}', + '{"valid_to": 0}', '{"valid_to": "invalid"}', + '[]', 'null', '{', '{"superseded_by": 9, "superseded_by": null}', + '{"valid_from": "yesterday"}', '{"valid_from": true}', + '{"valid_from": NaN}', '{"valid_from": 99999999999}', +], ids=[ + "superseded", "superseded-null", "expired", "invalid-expiry", + "array", "null", "malformed-json", "duplicate-key", + "invalid-start", "boolean-start", "nonfinite-start", "future-start", +]) +def test_unreadable_or_not_current_sqlite_metadata_vetoes_read(snapshot, make_runtime, raw_meta): + env = make_runtime() + _update_row(snapshot, meta=raw_meta) + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + assert env.client.events == [] + + +def test_metadata_cannot_supply_grants_or_inject_prompt_fields(snapshot, make_runtime): + env = make_runtime() + _update_row(snapshot, meta=json.dumps({ + "supersedes": snapshot.old_id, "valid_to": None, "valid_from": 0, + "decision": "ALLOW", "current_session": SECRET, + "execution_authority": "OWNER", "system_prompt": FOREIGN_TEXT, + })) + answer = env.runtime.ask(QUERY) + _assert_current_prompt(answer, env, snapshot) + assert SECRET not in env.client.calls[0]["system_prompt"] + env.provider.context["policy"] = _policy(env.provider.context["identity"], revoked=True) + env.client.calls.clear() + env.client.events.clear() + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + _assert_no_generation(env) + + +@pytest.mark.parametrize("cold, drift", [ + (False, "consent"), (True, "consent"), + (False, "policy"), (True, "candidate"), + (False, "text"), (True, "superseded"), + (False, "session"), (True, "request"), +]) +def test_fresh_prompt_boundary_blocks_drift_after_initial_allow(snapshot, make_runtime, cold, drift): + env = make_runtime(cold=cold) + context = env.provider.context + original_pointer = context["admission_store"].pointer + + def change_after_evidence(): + assert env.provider.reads >= 1 + if drift == "consent": + _revoke_consent(context) + elif drift == "policy": + context["policy"] = _policy(context["identity"], revoked=True) + elif drift == "candidate": + context["admission_store"].candidates[context["person_record"]["id"]] = b"{}" + elif drift == "text": + _update_row(snapshot, text=FOREIGN_TEXT) + elif drift == "superseded": + _update_row(snapshot, meta=json.dumps({"superseded_by": snapshot.foreign_id})) + elif drift == "session": + _rebind_session(context, "changed-after-prepare") + else: + _rebind_request(context, "changed-after-prepare") + + env.client.on_models = change_after_evidence + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + assert env.provider.reads >= 2 + _assert_no_generation(env) + # The current SQL row can expire even while R3's pointer has not moved. + assert context["admission_store"].pointer == original_pointer + + +@pytest.mark.parametrize("operation", ["deep", "legacy", "prewarm", "evidence", "prompt", "cold-helper"]) +def test_protected_paths_cannot_bypass_active_official_fast_ask(snapshot, make_runtime, operation): + env = make_runtime(cold=True) + forged = (TwinEvidence(snapshot.foreign_id, "foreign-person", FOREIGN_TEXT, 1.0, "forged"),) + if operation == "legacy": + env.client.chat_fast_streaming_jit_reconciled = None + with pytest.raises(PersonTwinEvidenceError): + if operation == "deep": + env.runtime.ask(QUERY, mode="deep") + elif operation == "legacy": + env.runtime.ask(QUERY) + elif operation == "prewarm": + env.runtime.prewarm_fast_startup() + elif operation == "evidence": + env.runtime.evidence(QUERY) + elif operation == "prompt": + env.runtime._system_prompt(forged) + else: + env.runtime._ask_fast_cold_streaming(QUERY, profile=env.profile, evidence=forged) + assert env.client.events == [] and env.provider.reads == 0 + + +@pytest.mark.parametrize("helper", ["r21h-dispatch", "deep-acquisition"]) +@pytest.mark.parametrize("inside_ask", [False, True], ids=["outside-ask", "nested-in-fast"]) +def test_direct_deep_helpers_deny_before_release_or_acquisition(snapshot, make_runtime, helper, inside_ask): + env = make_runtime() + + def forbidden_deep(): + with pytest.raises(PersonTwinEvidenceError): + if helper == "r21h-dispatch": + env.runtime._ask_r21h(QUERY, mode="deep") + else: + env.runtime._ask_deep( + QUERY, profile=env.runtime.profiles["deep"], evidence=(), + phase_timings_ms={}, request_started_at=0.0, + ) + assert env.client.events == [] + + if inside_ask: + env.provider.on_read = forbidden_deep + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + else: + forbidden_deep() + assert env.provider.reads == 0 + + +@pytest.mark.parametrize("inside_ask", [False, True], ids=["outside-ask", "nested-in-fast"]) +def test_direct_fast_dispatcher_cannot_reenter_the_active_ask(snapshot, make_runtime, inside_ask): + env = make_runtime() + + def forbidden_dispatch(): + with pytest.raises(PersonTwinEvidenceError): + env.runtime._ask_r21h(QUERY, mode="fast") + assert env.client.events == [] + + if inside_ask: + env.provider.on_read = forbidden_dispatch + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + else: + forbidden_dispatch() + assert env.provider.reads == 0 + + +@pytest.mark.parametrize("change", ["text", "id", "namespace", "append-foreign"]) +def test_replaced_evidence_payload_is_denied_inside_real_ask(snapshot, make_runtime, monkeypatch, change): + env = make_runtime() + original = env.runtime.evidence + + def forged_evidence(query): + evidence = original(query) + if change == "append-foreign": + return evidence + (TwinEvidence(snapshot.foreign_id, "foreign-person", FOREIGN_TEXT, 1.0, "forged"),) + value = {"text": FOREIGN_TEXT, "id": snapshot.foreign_id, "namespace": "foreign-person"}[change] + return (replace(evidence[0], **{change: value}),) + + monkeypatch.setattr(env.runtime, "evidence", forged_evidence) + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + _assert_no_generation(env) + + +@pytest.mark.parametrize("boundary", ["provider", "memory-store"]) +@pytest.mark.parametrize("error_class", [RuntimeError, PersonTwinEvidenceError]) +def test_read_boundary_exception_never_exposes_protected_text_or_anchor(make_runtime, monkeypatch, error_class, boundary): + env = make_runtime() + secret = SECRET + ":" + env.intent.request_id + + def fail(*args): + raise error_class(secret) + + if boundary == "provider": + env.provider.on_read = fail + else: + monkeypatch.setattr(env.runtime.memory.store, "get", fail) + with pytest.raises(PersonTwinEvidenceError) as caught: + env.runtime.ask(QUERY) + human_error = "".join(traceback.format_exception(caught.value)) + assert SECRET not in human_error and env.intent.request_id not in human_error + assert env.client.events == [] + + +def test_denied_request_releases_frame_and_cannot_reuse_previous_allow(snapshot, make_runtime): + env = make_runtime() + initial_policy = copy.deepcopy(env.provider.context["policy"]) + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + env.client.calls.clear() + env.client.events.clear() + env.provider.context["policy"] = _policy(env.provider.context["identity"], revoked=True) + with pytest.raises(PersonTwinEvidenceError): + env.runtime.ask(QUERY) + _assert_no_generation(env) + env.provider.context["policy"] = initial_policy + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + with pytest.raises(PersonTwinEvidenceError): + env.runtime._system_prompt(()) + + +def test_nested_ask_does_not_replace_or_clear_outer_frame(snapshot, make_runtime): + env = make_runtime() + nested_errors = [] + + def nested_request(): + with pytest.raises(PersonTwinEvidenceError) as caught: + env.runtime.ask(QUERY) + nested_errors.append(str(caught.value)) + + env.provider.on_read = nested_request + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + assert len(nested_errors) == 1 and env.provider.reads >= 2 + with pytest.raises(PersonTwinEvidenceError): + env.runtime.evidence(QUERY) + + +def test_foreign_thread_cannot_borrow_active_ask_for_protected_helpers(snapshot, make_runtime): + env = make_runtime() + started = Event() + results = [] + workers = [] + + def attempt_during_residency_probe(): + # Even a caller holding the legitimate tuple cannot borrow another + # thread's active ask. Lock release must not make old evidence usable. + evidence = env.runtime._person_active_request["evidence"] + + def worker(): + started.set() + for call in ( + lambda: env.runtime._system_prompt(evidence), + lambda: env.runtime.evidence(QUERY), + lambda: env.runtime._ask_fast_cold_streaming(QUERY, profile=env.profile, evidence=evidence), + ): + try: + value = call() + except BaseException as error: + results.append(error) + else: + results.append(value) + + thread = Thread(target=worker, daemon=True) + workers.append(thread) + thread.start() + assert started.wait(timeout=2) + assert results == [] + + env.client.on_models = attempt_during_residency_probe + try: + _assert_current_prompt(env.runtime.ask(QUERY), env, snapshot) + finally: + for worker in workers: + worker.join(timeout=2) + assert workers and all(not worker.is_alive() for worker in workers) + assert len(results) == 3 and all(isinstance(result, PersonTwinEvidenceError) for result in results) + assert env.provider.reads >= 2 and len(env.client.calls) == 1 From 742c7aaa12d3af5cbc79c119e9d75b7afaa94a87 Mon Sep 17 00:00:00 2001 From: bitmaster162 <115934939+bitmaster162@users.noreply.github.com> Date: Mon, 12 Oct 2026 12:32:59 -0700 Subject: [PATCH 3/4] test(person-twin): load R4 fixtures in isolated wheel suites Resolve the unchanged adjacent R4 fixture by file path instead of a bare module import so pytest importlib mode can collect the R21H adapter tests outside the source checkout. Preserve runtime, adapter, fixture and CI bytes. Document initial PR #211 wheel-collection failures and local verification: 154 scoped tests, 61 standalone importlib tests, and actual installed-wheel gate 2407 passed / 117 skipped / 870 subtests passed. Reviewed local candidate 81ccd745886a6a23b559029515b84f832426379c; exact tree 3ed09df888465c4cc3d61433fc30b1a6f6ecd5d4. Synthetic qualification only; no production/native admission. --- docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md | 64 +++++++++++++++++++- tests/test_person_twin_r21h_evidence.py | 25 +++++--- 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md index 0627f6f5..a6dbc200 100644 --- a/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md +++ b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md @@ -1,6 +1,6 @@ # Person evidence in R21H — R1 integration candidate -Status: local synthetic integration candidate; `NOT_PRODUCTION_ADMITTED`. +Status: synthetic integration under draft review; `NOT_PRODUCTION_ADMITTED`. ## Why this change exists @@ -22,6 +22,9 @@ is introduced. PR #150 head `4515bea19978b522328607deb493a55234ef7d4c`. - R1, R2 and R3 remain the existing master implementations. - Historical R21B–R21G overlays remain unchanged. +- The initial candidate was published as Draft PR #211 at + `4d8cf5d3028e11d4990b836e9e15ee881f446e6e`, with the approved tree + `1d97123f6b08fde74754117128eabb1dbb75ab7a`. Importing R4 into this candidate does not merge PR #150 or grant pilot admission. @@ -99,7 +102,10 @@ authenticated authority and currentness inputs, approved database custody and egress, and coverage for every enabled route. This candidate does not enable that pilot. -## Validation record +## Initial candidate validation record + +The following receipt describes the initial published tree above. Its test-file +hash predates the wheel-import followup documented below. Observed 2026-10-12 19:03:36–19:03:38 UTC (2026-10-13 Asia/Bangkok): @@ -156,3 +162,57 @@ raise SystemExit(pytest.main([ ])) PY ``` + +## Wheel-import followup qualification + +The initial PR #211 review-gates run +([38296546803](https://github.com/bitmaster162/continuityos/actions/runs/38296546803)) +passed its clean-source suites: Ubuntu recorded 2479 passed / 67 skipped, +and Windows recorded 2487 passed / 59 skipped; both recorded 870 passed subtests. +Both jobs then failed while collecting the external wheel suite: +`ModuleNotFoundError: No module named 'test_person_twin_authority_replay'`. +Those failed CI results remain the record for the initial published tree. + +The new test used a bare sibling-module import. The wheel gate copies the tests +to an external directory and uses `--import-mode=importlib`, where that short +module name is unavailable. A standalone importlib collection reproduced the +failure locally before the fix. The followup resolves the unchanged adjacent +R4 fixture with `Path(__file__).with_name(...)` and `importlib.util`, preserving +the four existing fixture aliases. It does not change `sys.path`, the runtime, +the R4 fixture, or the CI gate and its exclusions. + +The revised adapter test has SHA-256 +`661d8f96b42034c673fee872e157df68b03c6e779e857a064fce1b9f33437927`. +An independent review checked these exact bytes and found no remaining issue +with the fixture-loading change. + +Local verification on 2026-10-12, Python 3.12.14 / pytest 9.1.1: + +| Verification | Result | +|---|---:| +| Original scoped suite, default import mode, 19:21:10 UTC | 154 passed; 0 skipped | +| Adapter file selected alone, importlib mode, 19:21:35 UTC | 61 passed; 0 skipped | +| Actual installed-wheel suite, unmodified `tools/ci_review.py` wheel gate | 2407 passed; 117 skipped; 870 subtests passed | + +The two scoped runs denied all `socket.*` audit events and observed zero events; +their five source/test hashes were unchanged before and after. Their receipt +SHA-256 values are respectively +`8b2d32b9f3ed2ca9c82d94059732f5cd5f63ff8809511e5d7f1a647250ab6ccd` +and `8dae8b07732416ff4f90d82eb61ddfc434d1d94f3c1ea20bcf7944fc52602f87`. + +The wheel rehearsal built the declared package in a separate source copy, +installed it with pytest into a fresh isolated environment, and used exactly +the gate's four existing exclusions. The package imported from that environment's +`site-packages`; neither project checkout nor the build source appeared on the +test interpreter's import path. Package and distribution versions both read +`0.10.3`. Documentation was annotated after the source snapshot; the tested +runtime/R4 files and revised adapter test remained byte-identical. + +Wheel SHA-256: +`80ead6cc699a53efd01e0dbc0eba9c511ffd891f8010772651d26e32fc25d186`. +Installed-wheel gate receipt SHA-256: +`b4c597640abd016963c68accc5cc4ce6113901c32877f76e25e137dd6747cc68`. + +These are local Linux followup results. At recording time, the followup had not +been published or checked by GitHub's Ubuntu/Windows jobs. They do not replace +the initial failed CI record or change the integration's admission limits. diff --git a/tests/test_person_twin_r21h_evidence.py b/tests/test_person_twin_r21h_evidence.py index 731c8ccb..5f780e78 100644 --- a/tests/test_person_twin_r21h_evidence.py +++ b/tests/test_person_twin_r21h_evidence.py @@ -10,7 +10,9 @@ import copy from dataclasses import replace import hashlib +import importlib.util import json +from pathlib import Path import sqlite3 from threading import Event, Thread import traceback @@ -38,13 +40,22 @@ TwinEvidence, ) -# Reuse the unchanged R4 fixture contract, including real R3 admission writes. -from test_person_twin_authority_replay import ( - Store as AdmissionStore, - _hash, - _inputs, - _policy, -) +# Reuse the unchanged R4 fixture contract in source and isolated wheel suites. +def _load_r4_fixture_module(): + path = Path(__file__).with_name("test_person_twin_authority_replay.py") + spec = importlib.util.spec_from_file_location("_person_twin_r4_fixture_r21h", path) + if spec is None or spec.loader is None: + raise ImportError("Cannot load the adjacent R4 test fixture module") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +_R4_FIXTURE = _load_r4_fixture_module() +AdmissionStore = _R4_FIXTURE.Store +_hash = _R4_FIXTURE._hash +_inputs = _R4_FIXTURE._inputs +_policy = _R4_FIXTURE._policy QUERY = "Restore the current synthetic project context." From 28bfce3c1900d11bcc90fdd9428295ce18c68b14 Mon Sep 17 00:00:00 2001 From: bitmaster162 <115934939+bitmaster162@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:15:07 -0700 Subject: [PATCH 4/4] fix(person-twin): make protected FAST request phases single-use --- continuityos/sovereign_twin_runtime.py | 30 ++++++- docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md | 86 ++++++++++++++++++++ tests/test_person_twin_r21h_evidence.py | 66 +++++++++++++++ 3 files changed, 179 insertions(+), 3 deletions(-) diff --git a/continuityos/sovereign_twin_runtime.py b/continuityos/sovereign_twin_runtime.py index 2c660532..43d11de4 100644 --- a/continuityos/sovereign_twin_runtime.py +++ b/continuityos/sovereign_twin_runtime.py @@ -100,8 +100,12 @@ def evidence(self, query: str) -> tuple[TwinEvidence, ...]: active = self._person_request() if query != active["query"]: raise PersonTwinEvidenceError("PERSON_ACTIVE_QUERY_MISMATCH") + if active["phase"] != "EVIDENCE_ALLOWED": + raise PersonTwinEvidenceError("PERSON_EVIDENCE_PHASE_INVALID") + active["phase"] = "PREPARING_EVIDENCE" evidence = adapter.prepare(query, self.memory.store) active["evidence"] = evidence + active["phase"] = "EVIDENCE_READY" return evidence def _person_system_prompt(self, evidence: Sequence[TwinEvidence]) -> str: @@ -112,10 +116,16 @@ def _person_system_prompt(self, evidence: Sequence[TwinEvidence]) -> str: active = self._person_request() if evidence is not active["evidence"]: raise PersonTwinEvidenceError("PERSON_FOREIGN_PROMPT_EVIDENCE") + if active["phase"] != "PROMPT_ALLOWED": + raise PersonTwinEvidenceError("PERSON_PROMPT_PHASE_INVALID") + # Consume the boundary before provider callbacks can reenter it. + active["phase"] = "PROMPT_VALIDATING" active["receipt"] = adapter.validate_for_prompt( active["query"], self.memory.store, evidence, ) - return super()._system_prompt(evidence) + prompt = super()._system_prompt(evidence) + active["phase"] = "PROMPT_READY" + return prompt def _ask_fast_cold_streaming( self, query: str, *, profile: LocalModelProfile, evidence: Any, @@ -126,6 +136,11 @@ def _ask_fast_cold_streaming( active = self._person_request() if query != active["query"] or evidence is not active["evidence"] or profile != self.profiles["fast"]: raise PersonTwinEvidenceError("PERSON_FOREIGN_STREAM_REQUEST") + if active["phase"] != "COLD_DISPATCH_ALLOWED": + raise PersonTwinEvidenceError("PERSON_COLD_DISPATCH_PHASE_INVALID") + # Only the outer dispatcher issues this one-use permission, after + # its residency probe. An active frame alone cannot start a stream. + active["phase"] = "PROMPT_ALLOWED" self._require_fast_streaming_transport() return super()._ask_fast_cold_streaming(query, profile=profile, evidence=evidence) @@ -270,11 +285,13 @@ def ask(self, query: str, *, mode: str = "fast") -> TwinAnswer: self._person_active_request = { "adapter": adapter, "query": query, "owner_thread": _get_ident(), "evidence": None, "receipt": None, "dispatcher_entered": False, + "phase": "DISPATCH_PENDING", } try: answer = self._ask_r21h(query, mode=mode) - receipt = self._person_request()["receipt"] - if receipt is None: + active = self._person_request() + receipt = active["receipt"] + if receipt is None or active["phase"] != "PROMPT_READY": raise PersonTwinEvidenceError("PERSON_PROMPT_REVALIDATION_MISSING") return _replace(answer, stats={**answer.stats, "person_evidence": receipt}) finally: @@ -282,6 +299,7 @@ def ask(self, query: str, *, mode: str = "fast") -> TwinAnswer: def _ask_r21h(self, query: str, *, mode: str = "fast") -> TwinAnswer: """Answer FAST after evidence/reprobe, preserving old custom-client fallback.""" + active = None if getattr(self, "_person_evidence_adapter", None) is not None: with self._model_lock: active = self._person_request() @@ -303,17 +321,23 @@ def _ask_r21h(self, query: str, *, mode: str = "fast") -> TwinAnswer: raise ValueError(f"unknown Sovereign Twin mode: {mode}") profile = self.profiles[mode] + if active is not None: + active["phase"] = "EVIDENCE_ALLOWED" evidence = self.evidence(query) resident_id = self._probe_exact_fast_residency(profile) if resident_id is None: self._require_fast_streaming_transport() + if active is not None: + active["phase"] = "COLD_DISPATCH_ALLOWED" return self._ask_fast_cold_streaming( query, profile=profile, evidence=evidence, ) + if active is not None: + active["phase"] = "PROMPT_ALLOWED" result = self.client.chat( model=profile.model, system_prompt=self._system_prompt(evidence), diff --git a/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md index a6dbc200..bffaf711 100644 --- a/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md +++ b/docs/PERSON_TWIN_R21H_EVIDENCE_ADAPTER_R1.md @@ -60,6 +60,12 @@ startup prewarm and direct protected helper use outside the active request. The request frame is scoped to the runtime lock and is cleared after completion or failure. +The request-phase followup below makes evidence preparation, cold dispatch and +prompt construction one-use steps within that frame. The outer dispatcher +permits cold transport only after its residency probe has returned. Prompt +validation consumes its permission before calling the context provider. Nested +callbacks cannot borrow a pending or completed step to start another generation. + The generic runtime remains available when no adapter is configured. Its compatibility behavior does not become Person-qualified by this change. The class-level prompt formatter used by generic DeepLite is retained. @@ -216,3 +222,83 @@ Installed-wheel gate receipt SHA-256: These are local Linux followup results. At recording time, the followup had not been published or checked by GitHub's Ubuntu/Windows jobs. They do not replace the initial failed CI record or change the integration's admission limits. + +## Request-phase followup — post-CI review + +The wheel-import repair was published as HEAD +`742c7aaa12d3af5cbc79c119e9d75b7afaa94a87`, tree +`3ed09df888465c4cc3d61433fc30b1a6f6ecd5d4`. All five GitHub checks completed +successfully on that HEAD, including Ubuntu and Windows review-gates run +[38298448627](https://github.com/bitmaster162/continuityos/actions/runs/38298448627). +Those checks did not exercise the callback reentry described here. + +The subsequent source review found that the bound cold-streaming helper accepted +the same active query, profile and evidence tuple while the outer request was +still validating its prompt. A context-provider callback could therefore invoke +the helper again. An independently repeated local characterization of the +published source recorded two captured streaming calls with identical prompts, +three provider reads and an outer answer containing an ALLOW receipt. The +runtime methods and request fields were not replaced or modified by the probe. +All data and model clients were synthetic; no live inference was performed. + +The earlier integration review's no-blockers conclusion is superseded for this +request-reentry path. An active frame and a one-use top-level dispatcher did not +make the nested transport helper one-use. + +The fix adds internal request phases to the existing protected R21H path: + +- evidence permission is consumed before the initial provider read; +- cold-dispatch permission is issued only after the outer residency probe and + consumed before entering transport helpers; +- both warm and cold prompt construction consume the same prompt permission + before final provider validation; +- a successful answer requires a completed prompt phase and receipt; +- wrong-phase calls fail with fixed error codes, while the outer frame still + clears on success or failure. + +This is an in-process sequencing guard for the configured protected path. It +does not authenticate a hostile host, provide durable replay protection across +separate asks, or make admission, SQLite and model delivery atomic. + +Validation on 2026-10-12 UTC, Python 3.12.14 / pytest 9.1.1: + +| Observation | Result | +|---|---| +| Root characterization on the published source | Two model calls for one ask; defect reproduced | +| Twelve new regression cases on the published runtime | 12 failed as expected; characterization only | +| Corrected runtime with the same twelve cases and existing scoped suite | 166 passed; 0 failed; 0 skipped | +| Separate reviewer/process replay of the twelve cases in importlib mode | 12 passed; 0 failed; 0 skipped | + +The new cases cover cold-helper entry before dispatch on warm/cold paths, +during final provider reads, during model-load and final-residency callbacks, +plus nested prompt/evidence helpers. Each requires denial with no extra +generation, the unchanged successful outer prompt and a cleared request frame. +The complete scoped run contains 73 adapter/integration cases, 50 unchanged R4 +cases and 43 existing runtime/DeepLite cases. Socket events were denied and zero +were observed; all five selected code/test hashes matched before and after. + +| Corrected file | SHA-256 | +|---|---| +| R21H runtime | `8ce39204994fa7d8d622246cb4a766547980bf5ff5c89163eeb479dbbaafd935` | +| Adapter/integration tests | `eecf6b2b86974e2a7ebaaa9223e79b7a57cc929d5dc5fa9edbc7eaebdd83cc19` | + +Root characterization receipt SHA-256: +`c7180daccd3fd92da27ef113567258fd206350d0245d6b60bffd5a360d4b8b74`. +Expected-failure regression receipt SHA-256: +`9f4769b90d40de497b3c5efd05423f96eefc82a6340e2826e0df8c5809c49e4f`. +Corrected scoped-run receipt SHA-256: +`fea5f2bf344abeaa2ddf7c4ea17853a378001edd0f5a25bbdea399993557827e`. +Independent replay receipt SHA-256: +`c17a7ecd98769fdcf34ebd609cc56efe6547879caad7b2d999e54b8d1bbb912b`. +The replay used the exact expected-failure pytest arguments and node order in a +fresh process. Its five source/test hashes matched the corrected scoped run +before and after execution; no socket events were observed. This is a separate +verification role and process using the same source and fixture lineage. +These unsigned local receipts bind observed test inputs and outcomes; they do +not supply independent owner authority. The author review and root reproduction +use the same repository and synthetic fixture lineage. + +At recording time this request-phase followup is a local candidate. It has not +been published or checked by GitHub CI. The five successful checks above belong +to its predecessor. R4, the evidence adapter, fixture import, workflows and CI +tools are unchanged. Existing trust, personal-pilot and production limits remain. diff --git a/tests/test_person_twin_r21h_evidence.py b/tests/test_person_twin_r21h_evidence.py index 5f780e78..668719f2 100644 --- a/tests/test_person_twin_r21h_evidence.py +++ b/tests/test_person_twin_r21h_evidence.py @@ -694,3 +694,69 @@ def worker(): assert workers and all(not worker.is_alive() for worker in workers) assert len(results) == 3 and all(isinstance(result, PersonTwinEvidenceError) for result in results) assert env.provider.reads >= 2 and len(env.client.calls) == 1 + + +@pytest.mark.parametrize("helper, stage, cold", [ + pytest.param("cold-helper", "pre-dispatch", False, id="warm-cold-helper-pre-dispatch"), + pytest.param("cold-helper", "pre-dispatch", True, id="cold-cold-helper-pre-dispatch"), + pytest.param("cold-helper", "final-read", False, id="warm-cold-helper-final-read"), + pytest.param("cold-helper", "final-read", True, id="cold-cold-helper-final-read"), + pytest.param("cold-helper", "model-load", True, id="cold-cold-helper-model-load"), + pytest.param("cold-helper", "final-residency", True, id="cold-cold-helper-final-residency"), + pytest.param("prompt", "pre-dispatch", False, id="warm-prompt-pre-dispatch"), + pytest.param("prompt", "pre-dispatch", True, id="cold-prompt-pre-dispatch"), + pytest.param("prompt", "final-read", False, id="warm-prompt-final-read"), + pytest.param("prompt", "final-read", True, id="cold-prompt-final-read"), + pytest.param("evidence", "initial-read", False, id="evidence-initial-read"), + pytest.param("evidence", "final-read", False, id="evidence-final-read"), +]) +def test_request_phase_helpers_cannot_reenter_outer_fast(snapshot, make_runtime, helper, stage, cold): + env = make_runtime(cold=cold) + attempts = [] + + def attempt_nested_helper(): + calls_before = len(env.client.calls) + denied = False + try: + if helper == "evidence": + env.runtime.evidence(QUERY) + else: + # Reuse the legitimate pending tuple without modifying the frame + # or replacing runtime methods. Membership alone is insufficient. + evidence = env.runtime._person_active_request["evidence"] + if helper == "prompt": + env.runtime._system_prompt(evidence) + else: + env.runtime._ask_fast_cold_streaming( + QUERY, profile=env.profile, evidence=evidence, + ) + except PersonTwinEvidenceError: + denied = True + attempts.append({ + "denied": denied, + "extra_generations": len(env.client.calls) - calls_before, + }) + + if stage == "initial-read": + env.provider.on_read = attempt_nested_helper + elif stage == "pre-dispatch": + env.client.on_models = attempt_nested_helper + elif stage == "final-read": + env.client.on_models = lambda: setattr(env.provider, "on_read", attempt_nested_helper) + elif stage == "model-load": + env.client.on_models = lambda: setattr(env.client, "on_models", attempt_nested_helper) + else: + # Cold FAST probes before dispatch, at model_load.end, and once more + # after the stream returns. Arm this attempt for the final readback. + def arm_final_readback(): + env.client.on_models = attempt_nested_helper + + env.client.on_models = lambda: setattr(env.client, "on_models", arm_final_readback) + + answer = env.runtime.ask(QUERY) + assert attempts == [{"denied": True, "extra_generations": 0}], { + "attempts": attempts, + "generation_events": [event for event in env.client.events if event in {"chat", "stream"}], + } + _assert_current_prompt(answer, env, snapshot) + assert env.runtime._person_active_request is None