from __future__ import annotations from collections.abc import Mapping import hashlib import json import multiprocessing as mp import os import time from pathlib import Path import pytest import checkin_cli.diagnostic_isolation as diagnostic_isolation from checkin_cli.customer_coaching import ( CustomerRegistryError, load_diagnostic_runtime_customer_registry, ) from checkin_cli.diagnostic_isolation import ( AUDIT_PENDING_TEXT, DUPLICATE_TEXT, UNKNOWN_TEXT, DiagnosticIsolationError, DiagnosticAuditPendingNoSend, DiagnosticAuthorityError, ActivatedDiagnosticDelivery, DurableDiagnosticActivationLoader, DiagnosticDeliveryAuthority, DiagnosticDeliveryCandidate, DiagnosticDuplicateNoSend, DiagnosticIsolationSpecV1, DiagnosticRoleRoute, DiagnosticSession, DiagnosticSessionState, DiagnosticUnknownNoSend, VerifiedDiagnosticReservation, _DIAGNOSTIC_HOST_ADMISSION_TOKEN, ) def _spec() -> DiagnosticIsolationSpecV1: return DiagnosticIsolationSpecV1( owner_digest="1" * 64, test_bot_digest="2" * 64, customer_destination_digest="3" * 64, operator_destination_digest="5" * 64, profile_digest="6" * 64, max_provider_timeout_seconds=5, expires_at="2099-01-01T00:00:00Z", approved_by="owner", approved_at="2026-01-01T00:00:00Z", ) def _live_sources(spec: DiagnosticIsolationSpecV1): token = diagnostic_isolation._DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN owner = diagnostic_isolation.DiagnosticOwnerLiveLoader( None, spec, diagnostic_isolation.DiagnosticOwnerSnapshot("1" * 64, "active"), factory_token=token, ) registry = diagnostic_isolation.DiagnosticRegistryLiveLoader( None, spec, diagnostic_isolation.DiagnosticRegistrySnapshot( "7" * 64, "9" * 64, "f" * 64, "6" * 64, "enabled" ), factory_token=token, ) consent = diagnostic_isolation.DiagnosticConsentActivationLiveLoader( None, spec, diagnostic_isolation.DiagnosticConsentActivationSnapshot( "5" * 64, "a" * 64, "s", 1, "2099-01-01T00:00:00Z", "activated", False, ), factory_token=token, ) proposal = diagnostic_isolation.DiagnosticProposalLiveLoader( None, spec, diagnostic_isolation.DiagnosticProposalSnapshot( "8" * 64, 1, diagnostic_isolation._digest(1), hashlib.sha256(b"diagnostic only").hexdigest(), hashlib.sha256( json.dumps( ("customer-chat", "customer-topic"), separators=(",", ":"), ).encode("utf-8") ).hexdigest(), "approved", "2099-01-01T00:00:00Z", ), factory_token=token, ) config = diagnostic_isolation.DiagnosticConfigLiveLoader( None, spec, diagnostic_isolation.DiagnosticConfigSnapshot( "b" * 64, "4" * 64, spec.diagnostic_transport_binding_digest, "active", ), factory_token=token, ) artifacts = diagnostic_isolation.DiagnosticArtifactLiveLoader( None, spec, diagnostic_isolation.DiagnosticArtifactSnapshot( "c" * 64, "d" * 64, "e" * 64, "approved" ), factory_token=token, ) return diagnostic_isolation.DiagnosticLiveAuthoritySources.from_verified_loaders( authority=None, spec=spec, owner_loader=owner, registry_loader=registry, consent_activation_loader=consent, proposal_loader=proposal, config_loader=config, artifact_loader=artifacts, factory_token=token, ) def _authority(tmp_path: Path, *, session_id: str = "s"): spec = _spec() session = DiagnosticSession( session_id, DiagnosticSessionState.ACTIVE, 1, "boot", spec.spec_digest, spec.authority_digest, spec.diagnostic_transport_binding_digest, "2099-01-01T00:00:00Z", ) return DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ), spec, session def _write_live_record( profile_root: Path, *, owner_digest: str = "1" * 64, registry_digest: str = "9" * 64, proposal_digest: str = "8" * 64, config_digest: str = "b" * 64, policy_digest: str = "c" * 64, catalog_digest: str = "d" * 64, meal_constraints_digest: str = "e" * 64, revoked: bool = False, ) -> Path: body_digest = hashlib.sha256(b"diagnostic only").hexdigest() destination_digest = diagnostic_isolation._digest(("customer-chat", "customer-topic")) record = { "schema_version": diagnostic_isolation._LIVE_AUTHORITY_SCHEMA, "customer_key_digest": "7" * 64, "session_id": "s", "session_generation": 1, "owner_digest": owner_digest, "registry_digest": registry_digest, "consent_digest": "5" * 64, "activation_receipt_digest": "a" * 64, "proposal_digest": proposal_digest, "revision": 1, "revision_digest": diagnostic_isolation._digest(1), "rendered_body_digest": body_digest, "destination_digest": destination_digest, "config_digest": config_digest, "policy_digest": policy_digest, "catalog_digest": catalog_digest, "meal_constraints_digest": meal_constraints_digest, "source_digest": "f" * 64, "registration_digest": "6" * 64, "epoch_digest": "4" * 64, "diagnostic_transport_binding_digest": _spec().diagnostic_transport_binding_digest, "state": diagnostic_isolation._LIVE_AUTHORITY_STATE, "expires_at_kst": "2099-01-01T00:00:00Z", "revoked": revoked, } record_path = ( profile_root / "data" / "diagnostic-isolation" / "live-authority.json" ) record_path.parent.mkdir(parents=True) (profile_root / "data").chmod(0o700) record_path.parent.chmod(0o700) record_path.write_text(json.dumps(record, sort_keys=True), encoding="utf-8") record_path.chmod(0o600) return record_path def _prepared_session(spec: DiagnosticIsolationSpecV1) -> DiagnosticSession: return DiagnosticSession( "s", DiagnosticSessionState.PREPARED, 1, "boot", spec.spec_digest, spec.authority_digest, spec.diagnostic_transport_binding_digest, spec.expires_at, ) def test_profile_record_factory_is_dormant_and_does_not_send(tmp_path: Path): spec = _spec() _write_live_record(tmp_path) authority = DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(spec), boot_epoch="boot", profile_root=tmp_path, spec=spec, ) assert authority.session.state is DiagnosticSessionState.PREPARED assert authority.live_sources.record_path == ( tmp_path / "data" / "diagnostic-isolation" / "live-authority.json" ).absolute() assert not authority.activation_path.exists() assert authority.rows("never-sent") == () @pytest.mark.parametrize("record_state", ("missing", "symlink", "malformed")) def test_profile_record_factory_fails_closed_for_unreadable_record( tmp_path: Path, record_state: str, ): spec = _spec() record_path = ( tmp_path / "data" / "diagnostic-isolation" / "live-authority.json" ) if record_state == "symlink": target = tmp_path / "outside.json" target.write_text("{}", encoding="utf-8") record_path.parent.mkdir(parents=True) record_path.symlink_to(target) elif record_state == "malformed": record_path.parent.mkdir(parents=True) record_path.write_text("{}", encoding="utf-8") record_path.chmod(0o600) with pytest.raises(DiagnosticAuthorityError): DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(spec), boot_epoch="boot", profile_root=tmp_path, spec=spec, ) assert not (tmp_path / "diagnostic").exists() def test_profile_record_factory_rejects_ancestor_symlink(tmp_path: Path): profile_root = tmp_path / "profile" external_root = tmp_path / "external" profile_root.mkdir(mode=0o700) external_root.mkdir(mode=0o700) _write_live_record(external_root) (profile_root / "data").symlink_to(external_root / "data", target_is_directory=True) with pytest.raises(DiagnosticAuthorityError): DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(_spec()), boot_epoch="boot", profile_root=profile_root, spec=_spec(), ) def test_profile_record_factory_rejects_symlink_above_profile_root( tmp_path: Path, ): real_parent = tmp_path / "real-parent" real_parent.mkdir(mode=0o700) real_profile = real_parent / "profile" real_profile.mkdir(mode=0o700) _write_live_record(real_profile) alias = tmp_path / "profile-parent-alias" alias.symlink_to(real_parent, target_is_directory=True) spec = _spec() with pytest.raises(DiagnosticAuthorityError): DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(spec), boot_epoch="boot", profile_root=alias / "profile", spec=spec, ) def test_profile_record_reload_rejects_replaced_ancestor_before_reservation( tmp_path: Path, ): spec = _spec() record_path = _write_live_record(tmp_path) authority = DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(spec), boot_epoch="boot", profile_root=tmp_path, spec=spec, ) authority.activate_session(generation=1) external_root = tmp_path / "external" external_root.mkdir(mode=0o700) _write_live_record(external_root) original_parent = record_path.parent original_parent.rename(tmp_path / "original-live-authority") original_parent.symlink_to( external_root / "data" / "diagnostic-isolation", target_is_directory=True, ) candidate = _candidate(spec, key="ancestor-replaced") with pytest.raises(DiagnosticAuthorityError): _reserve(authority, candidate) assert authority.rows(candidate.dedupe_key) == () def test_authority_rejects_non_exact_or_mismatched_spec_without_creating_root( tmp_path: Path, ): spec = _spec() session = DiagnosticSession( "sealed", DiagnosticSessionState.ACTIVE, 1, "boot", spec.spec_digest, spec.authority_digest, spec.diagnostic_transport_binding_digest, "2099-01-01T00:00:00Z", ) class SpecSubclass(DiagnosticIsolationSpecV1): pass with pytest.raises(DiagnosticAuthorityError, match="required"): DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path / "subclass", spec=object.__new__(SpecSubclass), live_sources=object(), ) assert not (tmp_path / "subclass").exists() mismatched = DiagnosticIsolationSpecV1( owner_digest="f" * 64, test_bot_digest=spec.test_bot_digest, customer_destination_digest=spec.customer_destination_digest, operator_destination_digest=spec.operator_destination_digest, profile_digest=spec.profile_digest, max_provider_timeout_seconds=spec.max_provider_timeout_seconds, expires_at=spec.expires_at, approved_by=spec.approved_by, approved_at=spec.approved_at, ) with pytest.raises(DiagnosticAuthorityError, match="binding mismatch"): DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path / "mismatch", spec=mismatched, live_sources=_live_sources(mismatched), ) assert not (tmp_path / "mismatch").exists() def _candidate(spec, key="d", *, session_id: str = "s", generation: int = 1): body = b"diagnostic only" destination = ("customer-chat", "customer-topic") destination_digest = hashlib.sha256( json.dumps(destination, separators=(",", ":")).encode("utf-8") ).hexdigest() activated = ActivatedDiagnosticDelivery.from_persisted( schema_version="diagnostic_activated_delivery_v1", customer_key_digest="7" * 64, proposal_digest="8" * 64, revision=1, rendered_body=body, rendered_body_digest=hashlib.sha256(body).hexdigest(), destination=destination, destination_digest=destination_digest, session_id=session_id, session_generation=generation, diagnostic_transport_binding_digest=spec.diagnostic_transport_binding_digest, registry_digest="9" * 64, activation_receipt_digest="a" * 64, config_digest="b" * 64, policy_digest="c" * 64, catalog_digest="d" * 64, meal_constraints_digest="e" * 64, expires_at_kst="2099-01-01T00:00:00Z", ) from checkin_cli.diagnostic_isolation import _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN candidate = DiagnosticDeliveryCandidate.from_activated( activated, factory_token=_DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN, session_digest=spec.spec_digest, ) live_values = { "customer_key_digest": "7" * 64, "owner_digest": "1" * 64, "registry_digest": "9" * 64, "consent_digest": "5" * 64, "activation_receipt_digest": "a" * 64, "proposal_digest": "8" * 64, "revision_digest": diagnostic_isolation._digest(1), "rendered_body_digest": hashlib.sha256(body).hexdigest(), "destination_digest": destination_digest, "config_digest": "b" * 64, "policy_digest": "c" * 64, "catalog_digest": "d" * 64, "meal_constraints_digest": "e" * 64, "source_digest": "f" * 64, "registration_digest": "6" * 64, "epoch_digest": "4" * 64, "diagnostic_transport_binding_digest": spec.diagnostic_transport_binding_digest, } for name, value in live_values.items(): object.__setattr__(candidate, name, value) object.__setattr__(candidate, "revision", 1) return candidate def test_activation_subclass_is_rejected_before_reservation(tmp_path: Path): authority, spec, _ = _authority(tmp_path) class ActivationSubclass(ActivatedDiagnosticDelivery): pass from checkin_cli.diagnostic_isolation import _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN with pytest.raises(DiagnosticAuthorityError, match="activation"): DiagnosticDeliveryCandidate.from_activated( object.__new__(ActivationSubclass), factory_token=_DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN, session_digest=spec.spec_digest, ) assert authority.rows("never-reserved") == () def _reserve(authority, candidate): with authority.delivery_admission(_DIAGNOSTIC_HOST_ADMISSION_TOKEN) as admission: return authority.reserve_and_verify( candidate, lock_token=admission, ) @pytest.mark.parametrize( "field, value", ( ("owner_digest", "f" * 64), ("registry_digest", "0" * 64), ("proposal_digest", "f" * 64), ("config_digest", "f" * 64), ("policy_digest", "f" * 64), ("catalog_digest", "f" * 64), ("meal_constraints_digest", "f" * 64), ("revoked", True), ), ) def test_profile_record_live_changes_reject_before_reservation( tmp_path: Path, field: str, value: object, ): spec = _spec() record_path = _write_live_record(tmp_path) authority = DiagnosticDeliveryAuthority.from_profile_record( _prepared_session(spec), boot_epoch="boot", profile_root=tmp_path, spec=spec, ) authority.activate_session(generation=1) candidate = _candidate(spec, key=f"live-{field}") record = json.loads(record_path.read_text(encoding="utf-8")) record[field] = value record_path.write_text(json.dumps(record, sort_keys=True), encoding="utf-8") record_path.chmod(0o600) with pytest.raises(DiagnosticAuthorityError, match="diagnostic live"): _reserve(authority, candidate) assert authority.rows(candidate.dedupe_key) == () def _reserve_after_fence_worker( profile_root: str, session: DiagnosticSession, spec: DiagnosticIsolationSpecV1, result_queue, ) -> None: try: authority = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=Path(profile_root), spec=spec, live_sources=_live_sources(spec), ) candidate = _candidate(spec, key="post-fence") result = _reserve(authority, candidate) except Exception as exc: result_queue.put(("error", type(exc).__name__, str(exc))) else: result_queue.put(("reserved", type(result).__name__)) def _detach_worker( profile_root: str, session: DiagnosticSession, spec: DiagnosticIsolationSpecV1, started, done, result_queue, ) -> None: started.set() try: authority = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=Path(profile_root), spec=spec, live_sources=_live_sources(spec), ) authority.detach(generation=1, state="detaching") except Exception as exc: result_queue.put(("error", type(exc).__name__, str(exc))) else: result_queue.put(("detached",)) finally: done.set() def _provider_lease_worker( profile_root: str, session: DiagnosticSession, spec: DiagnosticIsolationSpecV1, entered, allow_provider, detach_done, result_queue, ) -> None: try: authority = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=Path(profile_root), spec=spec, live_sources=_live_sources(spec), ) authority.revalidate_after_restart(new_boot_epoch="provider") candidate = _candidate(spec, key="lease") activated = candidate.activated assert activated is not None loader = DurableDiagnosticActivationLoader(authority) with authority.delivery_admission(_DIAGNOSTIC_HOST_ADMISSION_TOKEN) as admission: verified = authority.reserve_and_verify(candidate, lock_token=admission) assert isinstance(verified, VerifiedDiagnosticReservation) entered.set() if not allow_provider.wait(timeout=10): raise RuntimeError("provider admission release timed out") authority.verify_provider_start( verified, deadline_monotonic=time.monotonic() + 5, lock_token=admission, activated=activated, activation_loader=loader, ) provider_after_detach = detach_done.is_set() authority.record_terminal( verified, receipt={"id": 1}, audited=True, lock_token=admission, ) result_queue.put(("provider", provider_after_detach)) except Exception as exc: result_queue.put(("error", type(exc).__name__, str(exc))) _DIGEST_STAGE_ORDER = ( "spec_core_digest", "diagnostic_transport_binding_digest", "spec_digest", "authority_digest", ) _SPEC_SCHEMA_VERSION = "diagnostic_isolation_spec_v1" _TRANSPORT_SCHEMA_VERSION = "diagnostic_transport_binding_v1" _TRANSPORT_ADAPTER_KIND = "telegram-test-bot" _SEALED_METHOD_VERSION = "send_diagnostic_customer_v1" _SPEC_INPUT_FIELDS = ( "schema_version", "owner_digest", "test_bot_digest", "customer_destination_digest", "operator_destination_digest", "profile_digest", "max_provider_timeout_seconds", "expires_at", "approved_by", "approved_at", "supersedes_digest", ) _SPEC_CORE_FIELDS = ( "schema_version", "owner_digest", "test_bot_digest", "customer_destination_digest", "operator_destination_digest", "profile_digest", "max_provider_timeout_seconds", "expires_at", ) _TRANSPORT_BINDING_FIELDS = ( "schema_version", "adapter_kind", "test_bot_digest", "customer_destination_digest", "max_provider_timeout_seconds", "method_version", "spec_core_digest", ) _SPEC_DIGEST_FIELDS = ( "spec_core_digest", "diagnostic_transport_binding_digest", ) _AUTHORITY_FIELDS = ( "spec_digest", "approved_by", "approved_at", "supersedes_digest", ) _GOLDEN_DIGESTS = { "spec_core_digest": "d0627479b4b7ef136577543602c4a6c7bf81b07627db3612f1664ef2fc22e6e3", "diagnostic_transport_binding_digest": "bfdf1f2d443a91735fc12f211eb452a846f629fa8b5edbc9d5e2d1737560fc6f", "spec_digest": "7ce683cfc7837f29dca096fb2902d97913229c2422d394c29fb1d3928924e4fe", "authority_digest": "547b81eede6a719983db56d900da4f7e98acabe345db5b30b45628970052ef64", } def _raw_spec(**overrides: object) -> dict[str, object]: raw = { "schema_version": _SPEC_SCHEMA_VERSION, "owner_digest": "1" * 64, "test_bot_digest": "2" * 64, "customer_destination_digest": "3" * 64, "operator_destination_digest": "5" * 64, "profile_digest": "6" * 64, "max_provider_timeout_seconds": 5, "expires_at": "2099-01-01T00:00:00Z", "approved_by": "owner", "approved_at": "2026-01-01T00:00:00Z", "supersedes_digest": "", } raw.update(overrides) return raw def _spec_from_raw(raw: Mapping[str, object]) -> DiagnosticIsolationSpecV1: return DiagnosticIsolationSpecV1( owner_digest=raw["owner_digest"], test_bot_digest=raw["test_bot_digest"], customer_destination_digest=raw["customer_destination_digest"], operator_destination_digest=raw["operator_destination_digest"], profile_digest=raw["profile_digest"], max_provider_timeout_seconds=raw["max_provider_timeout_seconds"], expires_at=raw["expires_at"], approved_by=raw["approved_by"], approved_at=raw["approved_at"], supersedes_digest=raw["supersedes_digest"], schema_version=raw["schema_version"], ) def _validate_digest_input(value: object, *, field: str, allow_empty: bool = False) -> None: if allow_empty and type(value) is str and value == "": return if ( type(value) is not str or len(value) != 64 or any(char not in "0123456789abcdef" for char in value) ): raise ValueError(f"raw spec field {field!r} is not a digest") def _validate_version(value: object, *, field: str, prefix: str) -> str: if type(value) is not str or not value.startswith(prefix): raise ValueError(f"{field} is not a supported version") suffix = value[len(prefix) :] if not suffix.isdigit() or int(suffix) < 1: raise ValueError(f"{field} is not a supported version") return value def _validate_adapter_kind(value: object) -> str: if type(value) is str and value == _TRANSPORT_ADAPTER_KIND: return value prefix = f"{_TRANSPORT_ADAPTER_KIND}-v" if ( type(value) is not str or not value.startswith(prefix) or not value[len(prefix) :].isdigit() or int(value[len(prefix) :]) < 1 ): raise ValueError("transport adapter contract version is invalid") return value def _validate_raw_spec(raw: Mapping[str, object]) -> None: if type(raw) is not dict: raise TypeError("raw spec must be a mapping") if set(raw) != set(_SPEC_INPUT_FIELDS): raise ValueError("raw spec has missing or extra fields") if type(raw["schema_version"]) is not str or raw["schema_version"] != _SPEC_SCHEMA_VERSION: raise ValueError("raw spec schema version is unsupported") for field in ( "owner_digest", "test_bot_digest", "customer_destination_digest", "operator_destination_digest", "profile_digest", ): _validate_digest_input(raw[field], field=field) _validate_digest_input( raw["supersedes_digest"], field="supersedes_digest", allow_empty=True ) timeout = raw["max_provider_timeout_seconds"] if type(timeout) is not int or not 1 <= timeout <= 30: raise ValueError("raw spec provider timeout is invalid") for field in ("expires_at", "approved_by", "approved_at"): if type(raw[field]) is not str or not raw[field]: raise ValueError(f"raw spec field {field!r} is invalid") def _stdlib_digest(value: object) -> str: canonical = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") return hashlib.sha256(canonical).hexdigest() def _independent_digest_vector( raw: Mapping[str, object], *, stage_order: tuple[str, ...] = _DIGEST_STAGE_ORDER, spec_schema_version: str | None = None, transport_schema_version: str | None = None, transport_adapter_kind: str | None = None, sealed_method_version: str | None = None, ) -> dict[str, str]: _validate_raw_spec(raw) if tuple(stage_order) != _DIGEST_STAGE_ORDER: raise ValueError("digest stages must be derived in the sealed order") core_schema = ( raw["schema_version"] if spec_schema_version is None else _validate_version( spec_schema_version, field="spec schema version", prefix="diagnostic_isolation_spec_v", ) ) binding_schema = ( _TRANSPORT_SCHEMA_VERSION if transport_schema_version is None else _validate_version( transport_schema_version, field="transport schema version", prefix="diagnostic_transport_binding_v", ) ) adapter_kind = ( _TRANSPORT_ADAPTER_KIND if transport_adapter_kind is None else _validate_adapter_kind(transport_adapter_kind) ) method_version = ( _SEALED_METHOD_VERSION if sealed_method_version is None else _validate_version( sealed_method_version, field="sealed method version", prefix="send_diagnostic_customer_v", ) ) core_preimage = { field: raw[field] for field in raw if field in _SPEC_CORE_FIELDS } if set(core_preimage) != set(_SPEC_CORE_FIELDS): raise ValueError("raw spec core fields are malformed") core_preimage["schema_version"] = core_schema spec_core_digest = _stdlib_digest(core_preimage) binding_values = { "schema_version": binding_schema, "adapter_kind": adapter_kind, "test_bot_digest": raw["test_bot_digest"], "customer_destination_digest": raw["customer_destination_digest"], "max_provider_timeout_seconds": raw["max_provider_timeout_seconds"], "method_version": method_version, "spec_core_digest": spec_core_digest, } binding_preimage = { field: binding_values[field] for field in _TRANSPORT_BINDING_FIELDS } diagnostic_transport_binding_digest = _stdlib_digest(binding_preimage) spec_values = { "spec_core_digest": spec_core_digest, "diagnostic_transport_binding_digest": diagnostic_transport_binding_digest, } spec_preimage = {field: spec_values[field] for field in _SPEC_DIGEST_FIELDS} spec_digest = _stdlib_digest(spec_preimage) authority_values = { "spec_digest": spec_digest, "approved_by": raw["approved_by"], "approved_at": raw["approved_at"], "supersedes_digest": raw["supersedes_digest"], } authority_preimage = { field: authority_values[field] for field in _AUTHORITY_FIELDS } authority_digest = _stdlib_digest(authority_preimage) return { "spec_core_digest": spec_core_digest, "diagnostic_transport_binding_digest": diagnostic_transport_binding_digest, "spec_digest": spec_digest, "authority_digest": authority_digest, } def test_digest_vector_matches_independent_raw_spec_golden(): assert _independent_digest_vector(_raw_spec()) == _GOLDEN_DIGESTS spec = _spec() assert spec.spec_core_digest == "d0627479b4b7ef136577543602c4a6c7bf81b07627db3612f1664ef2fc22e6e3" assert spec.diagnostic_transport_binding_digest == "bfdf1f2d443a91735fc12f211eb452a846f629fa8b5edbc9d5e2d1737560fc6f" assert spec.spec_digest == "7ce683cfc7837f29dca096fb2902d97913229c2422d394c29fb1d3928924e4fe" assert spec.authority_digest == "547b81eede6a719983db56d900da4f7e98acabe345db5b30b45628970052ef64" @pytest.mark.parametrize( ("field", "value", "changed_stages"), [ pytest.param("owner_digest", "a" * 64, _DIGEST_STAGE_ORDER, id="owner"), pytest.param("test_bot_digest", "a" * 64, _DIGEST_STAGE_ORDER, id="test-bot"), pytest.param( "customer_destination_digest", "a" * 64, _DIGEST_STAGE_ORDER, id="customer-destination" ), pytest.param( "operator_destination_digest", "a" * 64, _DIGEST_STAGE_ORDER, id="operator-destination" ), pytest.param("profile_digest", "a" * 64, _DIGEST_STAGE_ORDER, id="profile"), pytest.param( "max_provider_timeout_seconds", 6, _DIGEST_STAGE_ORDER, id="provider-timeout" ), pytest.param("expires_at", "2099-01-02T00:00:00Z", _DIGEST_STAGE_ORDER, id="expiry"), pytest.param("approved_by", "reviewer", ("authority_digest",), id="approved-by"), pytest.param( "approved_at", "2026-01-02T00:00:00Z", ("authority_digest",), id="approved-at" ), pytest.param("supersedes_digest", "a" * 64, ("authority_digest",), id="supersedes"), ], ) def test_digest_field_mutations_follow_explicit_stage_contract( field: str, value: object, changed_stages: tuple[str, ...] ): raw = _raw_spec() raw[field] = value expected = _independent_digest_vector(raw) actual = _spec_from_raw(raw) baseline = _spec() for stage in _DIGEST_STAGE_ORDER: assert getattr(actual, stage) == expected[stage] assert (getattr(actual, stage) != getattr(baseline, stage)) is ( stage in changed_stages ) @pytest.mark.parametrize( ("option", "value", "changed_stages"), [ pytest.param( "spec_schema_version", "diagnostic_isolation_spec_v2", _DIGEST_STAGE_ORDER, id="spec-schema-version", ), pytest.param( "transport_schema_version", "diagnostic_transport_binding_v2", _DIGEST_STAGE_ORDER[1:], id="transport-schema-version", ), pytest.param( "transport_adapter_kind", "telegram-test-bot-v2", _DIGEST_STAGE_ORDER[1:], id="transport-contract-version", ), pytest.param( "sealed_method_version", "send_diagnostic_customer_v2", _DIGEST_STAGE_ORDER[1:], id="sealed-method-version", ), ], ) def test_digest_schema_contract_and_method_versions_are_bound( option: str, value: str, changed_stages: tuple[str, ...] ): baseline = _independent_digest_vector(_raw_spec()) mutated = _independent_digest_vector(_raw_spec(), **{option: value}) for stage in _DIGEST_STAGE_ORDER: assert (mutated[stage] != baseline[stage]) is (stage in changed_stages) if option == "spec_schema_version": with pytest.raises(DiagnosticIsolationError, match="invalid"): _spec_from_raw(_raw_spec(schema_version=value)) def test_digest_vector_is_invariant_to_raw_input_key_order(): raw = _raw_spec() reordered = {field: raw[field] for field in reversed(tuple(raw))} assert tuple(raw) != tuple(reordered) assert _independent_digest_vector(reordered) == _GOLDEN_DIGESTS def test_independent_digest_vector_rejects_malformed_contract_inputs(): missing = _raw_spec() del missing["owner_digest"] extra = _raw_spec(unexpected="rejected") wrong_schema = _raw_spec(schema_version="diagnostic_isolation_spec_v0") for malformed in (missing, extra, wrong_schema): with pytest.raises((TypeError, ValueError)): _independent_digest_vector(malformed) with pytest.raises(ValueError, match="sealed order"): _independent_digest_vector( _raw_spec(), stage_order=tuple(reversed(_DIGEST_STAGE_ORDER)) ) for option, value in ( ("spec_schema_version", "diagnostic_isolation_spec_v0"), ("transport_schema_version", "diagnostic_transport_binding_v0"), ("transport_adapter_kind", "telegram-test-bot-v0"), ("sealed_method_version", "send_diagnostic_customer_v0"), ): with pytest.raises(ValueError, match="version"): _independent_digest_vector(_raw_spec(), **{option: value}) for field, value in ( ("owner_digest", 7), ("max_provider_timeout_seconds", True), ("max_provider_timeout_seconds", float("nan")), ("expires_at", None), ): malformed = _raw_spec() malformed[field] = value with pytest.raises((TypeError, ValueError)): _independent_digest_vector(malformed) def test_digest_algebra_is_deterministic_and_acyclic(): one, two = _spec(), _spec() assert one.spec_core_digest == two.spec_core_digest assert one.diagnostic_transport_binding_digest == two.diagnostic_transport_binding_digest assert one.spec_digest == two.spec_digest assert one.authority_digest == two.authority_digest assert len({one.spec_core_digest, one.diagnostic_transport_binding_digest, one.spec_digest, one.authority_digest}) == 4 def test_role_route_checks_full_triple_and_generation(): route = DiagnosticRoleRoute(1, -100, 59, "operator", 4) assert route.matches(user_id=1, chat_id=-100, topic_id=59, generation=4) assert not route.matches(user_id=2, chat_id=-100, topic_id=59, generation=4) @pytest.mark.parametrize( "values", [ (True, -100, 59, "operator", 4), (1, -100, 59, "operator", True), (1, -100, 59, "unknown", 4), (1, -100, 59, "operator", 0), ], ) def test_role_route_rejects_bool_and_closed_role_inputs(values): with pytest.raises(DiagnosticAuthorityError, match="role route"): DiagnosticRoleRoute(*values) def test_role_route_subclass_is_rejected(): class RouteSubclass(DiagnosticRoleRoute): pass with pytest.raises(DiagnosticAuthorityError, match="role route"): RouteSubclass(1, -100, 59, "operator", 4) @pytest.mark.parametrize( ("field", "value"), [ ("schema_version", "diagnostic_isolation_spec_v2"), ("owner_digest", "A" * 64), ("max_provider_timeout_seconds", True), ("approved_at", "2026-01-01T00:00:00"), ("expires_at", "2025-01-01T00:00:00Z"), ("approved_by", "x" * 129), ], ) def test_spec_rejects_strict_inputs_before_hash(field, value, monkeypatch): raw = _raw_spec() raw[field] = value def forbidden_hash(*args, **kwargs): raise AssertionError("digest/hash must not run for malformed spec") monkeypatch.setattr(diagnostic_isolation, "_digest", forbidden_hash) with pytest.raises(DiagnosticIsolationError): _spec_from_raw(raw) def test_authority_rejects_forged_spec_before_filesystem_initialization(tmp_path: Path): spec = _spec() session = DiagnosticSession( "forged", DiagnosticSessionState.ACTIVE, 1, "boot", spec.spec_digest, spec.authority_digest, spec.diagnostic_transport_binding_digest, spec.expires_at, ) forged = object.__new__(DiagnosticIsolationSpecV1) with pytest.raises(DiagnosticAuthorityError, match="required"): DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path / "forged", spec=forged, live_sources=object(), ) assert not (tmp_path / "forged").exists() def _activation_kwargs(activated: ActivatedDiagnosticDelivery) -> dict[str, object]: return { "schema_version": activated.schema_version, "customer_key_digest": activated.customer_key_digest, "proposal_digest": activated.proposal_digest, "revision": activated.revision, "rendered_body": activated.rendered_body, "rendered_body_digest": activated.rendered_body_digest, "destination": activated.destination, "destination_digest": activated.destination_digest, "session_id": activated.session_id, "session_generation": activated.session_generation, "diagnostic_transport_binding_digest": activated.diagnostic_transport_binding_digest, "registry_digest": activated.registry_digest, "activation_receipt_digest": activated.activation_receipt_digest, "config_digest": activated.config_digest, "policy_digest": activated.policy_digest, "catalog_digest": activated.catalog_digest, "meal_constraints_digest": activated.meal_constraints_digest, "expires_at_kst": activated.expires_at_kst, } @pytest.mark.parametrize( ("field", "value"), [ ("schema_version", "diagnostic_activated_delivery_v2"), ("customer_key_digest", "A" * 64), ("revision", True), ("destination_digest", "f" * 63), ("expires_at_kst", "2099-01-01T00:00:00"), ], ) def test_activation_rejects_strict_inputs_before_hash(field, value, monkeypatch, tmp_path: Path): _, spec, _ = _authority(tmp_path) activated = _candidate(spec).activated assert activated is not None values = _activation_kwargs(activated) values[field] = value def forbidden_hash(*args, **kwargs): raise AssertionError("digest/hash must not run for malformed activation") monkeypatch.setattr(diagnostic_isolation.hashlib, "sha256", forbidden_hash) with pytest.raises(DiagnosticAuthorityError): ActivatedDiagnosticDelivery.from_persisted(**values) def test_direct_candidate_construction_is_sealed(tmp_path: Path): _, spec, _ = _authority(tmp_path) body = b"diagnostic only" with pytest.raises(DiagnosticAuthorityError, match="factory"): DiagnosticDeliveryCandidate( "d", spec.spec_digest, 1, body, hashlib.sha256(body).hexdigest(), spec.customer_destination_digest, spec.diagnostic_transport_binding_digest, {"policy_digest": "a" * 64}, ) def test_direct_reserve_requires_private_host_admission_token(tmp_path: Path): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec) with pytest.raises(DiagnosticAuthorityError, match="host admission token"): authority.reserve_and_verify(candidate) with pytest.raises(DiagnosticAuthorityError, match="host admission token"): authority.reserve_and_verify(candidate, lock_token=object()) assert authority.rows(candidate.dedupe_key) == () def test_fresh_reservation_then_audited_repeat_is_duplicate(tmp_path: Path): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec) fresh = _reserve(authority, candidate) assert isinstance(fresh, VerifiedDiagnosticReservation) with authority.delivery_admission(_DIAGNOSTIC_HOST_ADMISSION_TOKEN) as admission: authority.record_terminal( fresh, receipt={"id": 1}, audited=True, lock_token=admission, ) repeated = _reserve(authority, candidate) assert isinstance(repeated, DiagnosticDuplicateNoSend) assert repeated.text == DUPLICATE_TEXT and not repeated.provider_authority def test_started_replay_terminalizes_unknown_without_provider_authority(tmp_path: Path): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec) assert isinstance(_reserve(authority, candidate), VerifiedDiagnosticReservation) repeated = _reserve(authority, candidate) assert isinstance(repeated, DiagnosticUnknownNoSend) assert repeated.text == UNKNOWN_TEXT and not repeated.reconciliation_available def test_restart_started_becomes_unknown_without_resend(tmp_path: Path): authority, spec, session = _authority(tmp_path) candidate = _candidate(spec) assert isinstance(_reserve(authority, candidate), VerifiedDiagnosticReservation) restarted = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ) with pytest.raises(Exception, match="restart requires revalidation"): _reserve(restarted, candidate) restarted.revalidate_after_restart(new_boot_epoch="boot-restarted") repeated = _reserve(restarted, candidate) assert isinstance(repeated, DiagnosticUnknownNoSend) assert not isinstance(repeated, VerifiedDiagnosticReservation) assert [row["status"] for row in restarted.rows(candidate.dedupe_key)] == [ "delivery_attempt_started", "delivery_unknown", ] def test_receipt_without_audit_is_stable_across_restart(tmp_path: Path): authority, spec, session = _authority(tmp_path) candidate = _candidate(spec) fresh = _reserve(authority, candidate) assert isinstance(fresh, VerifiedDiagnosticReservation) with authority.delivery_admission(_DIAGNOSTIC_HOST_ADMISSION_TOKEN) as admission: authority.record_terminal( fresh, receipt={"id": 1}, audited=False, lock_token=admission, ) restarted = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ) with pytest.raises(Exception, match="restart requires revalidation"): _reserve(restarted, candidate) restarted.revalidate_after_restart(new_boot_epoch="boot-restarted") pending = _reserve(restarted, candidate) assert isinstance(pending, DiagnosticAuditPendingNoSend) assert pending.text == AUDIT_PENDING_TEXT and pending.reconciliation_available with restarted.delivery_admission(_DIAGNOSTIC_HOST_ADMISSION_TOKEN) as admission: restarted.record_terminal( fresh, receipt={"id": 1}, audited=True, lock_token=admission, ) assert isinstance(_reserve(restarted, candidate), DiagnosticDuplicateNoSend) def test_dedupe_conflict_fails_closed(tmp_path: Path): authority, spec, _ = _authority(tmp_path) _reserve(authority, _candidate(spec)) other = _candidate(spec) object.__setattr__(other, "destination_digest", "f" * 64) with pytest.raises(DiagnosticAuthorityError): _reserve(authority, other) def test_tampered_or_reordered_journal_fails_closed(tmp_path: Path): authority, spec, _ = _authority(tmp_path) _reserve(authority, _candidate(spec)) journal = authority.journal_path rows = [json.loads(line) for line in journal.read_text(encoding="utf-8").splitlines()] rows[1]["body_digest"] = "f" * 64 journal.write_text("\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n", encoding="utf-8") with pytest.raises(Exception): DiagnosticDeliveryAuthority( _authority(tmp_path / "other")[2], boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ) def test_ordinary_registry_is_rejected_by_diagnostic_loader(tmp_path: Path): registry = tmp_path / "registry.json" registry.write_text( json.dumps( { "version": 1, "owner": {"user_id": "owner", "chat_id": "owner-chat", "topic_id": "owner-topic"}, "customers": [], } ), encoding="utf-8", ) with pytest.raises(CustomerRegistryError): load_diagnostic_runtime_customer_registry( registry, tmp_path, session_digest="a" * 64, ) def test_diagnostic_registry_requires_marked_session_binding(tmp_path: Path): registry = tmp_path / "registry.json" registry.write_text( json.dumps( { "version": 1, "registry_mode": "diagnostic_isolated_v1", "owner": {"user_id": "owner", "chat_id": "owner-chat", "topic_id": "owner-topic"}, "customers": [], } ), encoding="utf-8", ) with pytest.raises(CustomerRegistryError): load_diagnostic_runtime_customer_registry( registry, tmp_path, session_digest="a" * 64, ) def test_durable_detach_blocks_reconstructed_authority(tmp_path): authority, spec, session = _authority(tmp_path) authority.detach(generation=1, state="detaching") reconstructed = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ) candidate = _candidate(spec) with pytest.raises(Exception, match="durably detached"): _reserve(reconstructed, candidate) assert reconstructed.rows(candidate.dedupe_key) == () @pytest.mark.parametrize("state", ("detaching", "expired", "closed")) def test_detach_rejects_corrupt_existing_terminal_fence( tmp_path: Path, state: str, ): authority, _spec_obj, _session = _authority(tmp_path) authority._terminal_path.write_text( json.dumps({"session_id": authority.session.session_id}), encoding="utf-8", ) authority._terminal_path.chmod(0o600) before = authority._terminal_path.read_bytes() with pytest.raises(DiagnosticAuthorityError): authority.detach(generation=1, state=state) assert authority._terminal_path.read_bytes() == before def test_detach_rejects_symlinked_existing_terminal_fence(tmp_path: Path): authority, _spec_obj, _session = _authority(tmp_path) target = tmp_path / "outside-terminal.json" target.write_text("{}", encoding="utf-8") target.chmod(0o600) authority._terminal_path.symlink_to(target) with pytest.raises(DiagnosticAuthorityError): authority.detach(generation=1, state="detaching") assert target.read_text(encoding="utf-8") == "{}" def test_detach_does_not_overwrite_fence_created_during_commit( tmp_path: Path, monkeypatch, ): authority, _spec_obj, _session = _authority(tmp_path) original_link = os.link def race_link(source, destination, **kwargs): descriptor = os.open( destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=kwargs["dst_dir_fd"], ) try: os.write(descriptor, b"{}") os.fsync(descriptor) finally: os.close(descriptor) return original_link(source, destination, **kwargs) monkeypatch.setattr(os, "link", race_link) with pytest.raises(DiagnosticAuthorityError): authority.detach(generation=1, state="detaching") assert authority._terminal_path.read_bytes() == b"{}" def test_detach_zero_write_does_not_commit_terminal_fence( tmp_path: Path, monkeypatch, ): authority, _spec_obj, _session = _authority(tmp_path) monkeypatch.setattr(os, "write", lambda _descriptor, _payload: 0) with pytest.raises(DiagnosticAuthorityError, match="incomplete"): authority.detach(generation=1, state="detaching") assert not authority._terminal_path.exists() assert not tuple( authority._terminal_path.parent.glob( f".{authority._terminal_path.name}.*.tmp" ) ) def test_record_terminal_without_lease_cannot_claim_provider_success_after_fence( tmp_path: Path, ): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec, key="compat") verified = _reserve(authority, candidate) assert isinstance(verified, VerifiedDiagnosticReservation) activated = candidate.activated assert activated is not None authority.activation_path.write_text( json.dumps(activated.to_persisted_record(), ensure_ascii=False, sort_keys=True), encoding="utf-8", ) authority.activation_path.chmod(0o600) loader = DurableDiagnosticActivationLoader(authority) authority.detach(generation=1, state="detaching") with pytest.raises(DiagnosticAuthorityError, match="context"): authority.record_terminal( verified, receipt={"id": 1}, audited=True, lock_token=_DIAGNOSTIC_HOST_ADMISSION_TOKEN, ) assert [row["status"] for row in authority.rows(candidate.dedupe_key)] == [ "delivery_attempt_started", ] with pytest.raises(DiagnosticAuthorityError, match="context"): authority.verify_provider_start( verified, deadline_monotonic=time.monotonic() + 5, lock_token=_DIAGNOSTIC_HOST_ADMISSION_TOKEN, activated=activated, activation_loader=loader, ) def test_cross_process_fence_before_reserve_leaves_no_post_fence_row_or_tree_debris( tmp_path: Path, ): authority, spec, session = _authority(tmp_path) authority.detach(generation=1, state="detaching") context = mp.get_context("fork") result_queue = context.Queue() worker = context.Process( target=_reserve_after_fence_worker, args=(str(tmp_path), session, spec, result_queue), ) worker.start() worker.join(timeout=10) if worker.is_alive(): worker.terminate() worker.join() assert worker.exitcode == 0 result = result_queue.get(timeout=2) assert result[0] == "error" assert "durably detached" in result[2] assert authority.rows(_candidate(spec, key="post-fence").dedupe_key) == () assert { path.relative_to(tmp_path).as_posix() for path in tmp_path.rglob("*") } == { "diagnostic", "diagnostic/delivery-authority.lock", "diagnostic/delivery-authority.jsonl", "diagnostic/diagnostic-session-terminal.json", } terminal = authority.profile_root / "diagnostic" / "diagnostic-session-terminal.json" assert terminal.is_file() and not terminal.is_symlink() def test_cross_process_admission_lease_blocks_fence_between_reserve_and_provider( tmp_path: Path, ): authority, spec, session = _authority(tmp_path) activated = _candidate(spec, key="lease").activated assert activated is not None authority.activation_path.write_text( json.dumps(activated.to_persisted_record(), ensure_ascii=False, sort_keys=True), encoding="utf-8", ) authority.activation_path.chmod(0o600) context = mp.get_context("fork") entered = context.Event() allow_provider = context.Event() detach_started = context.Event() detach_done = context.Event() provider_results = context.Queue() detach_results = context.Queue() provider = context.Process( target=_provider_lease_worker, args=( str(tmp_path), session, spec, entered, allow_provider, detach_done, provider_results, ), ) provider.start() assert entered.wait(timeout=10) detacher = context.Process( target=_detach_worker, args=(str(tmp_path), session, spec, detach_started, detach_done, detach_results), ) detacher.start() assert detach_started.wait(timeout=10) allow_provider.set() provider.join(timeout=10) if provider.is_alive(): provider.terminate() provider.join() assert provider.exitcode == 0 provider_result = provider_results.get(timeout=2) assert provider_result == ("provider", False) detacher.join(timeout=10) if detacher.is_alive(): detacher.terminate() detacher.join() assert detacher.exitcode == 0 assert detach_results.get(timeout=2)[0] == "detached" rows = authority.rows(_candidate(spec, key="lease").dedupe_key) assert [row["status"] for row in rows] == [ "delivery_attempt_started", "provider_receipt", "sent_audited", ] terminal = authority.profile_root / "diagnostic" / "diagnostic-session-terminal.json" assert terminal.is_file() and not terminal.is_symlink() assert { path.relative_to(tmp_path).as_posix() for path in tmp_path.rglob("*") } == { "diagnostic", "diagnostic/activated-delivery.json", "diagnostic/delivery-authority.lock", "diagnostic/delivery-authority.jsonl", "diagnostic/diagnostic-session-terminal.json", } def test_durable_detach_symlink_replacement_keeps_reserve_and_provider_deltas_zero( tmp_path: Path, ): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec, key="pre-fence") verified = _reserve(authority, candidate) assert isinstance(verified, VerifiedDiagnosticReservation) activated = candidate.activated assert activated is not None authority.activation_path.write_text( json.dumps(activated.to_persisted_record(), ensure_ascii=False, sort_keys=True), encoding="utf-8", ) authority.activation_path.chmod(0o600) loader = DurableDiagnosticActivationLoader(authority) authority.detach(generation=1, state="detaching") terminal = authority.profile_root / "diagnostic" / "diagnostic-session-terminal.json" terminal.unlink() terminal.symlink_to(tmp_path / "missing-terminal-fence") before_provider = authority.rows(candidate.dedupe_key) with pytest.raises(DiagnosticAuthorityError): _reserve(authority, _candidate(spec, key="post-fence-reserve")) with pytest.raises(DiagnosticAuthorityError): authority.verify_provider_start( verified, deadline_monotonic=time.monotonic() + 5, lock_token=_DIAGNOSTIC_HOST_ADMISSION_TOKEN, activated=activated, activation_loader=loader, ) assert authority.rows("post-fence-reserve") == () assert authority.rows(candidate.dedupe_key) == before_provider def test_durable_detach_regular_inode_replacement_keeps_reserve_and_provider_deltas_zero( tmp_path: Path, monkeypatch, ): authority, spec, _ = _authority(tmp_path) candidate = _candidate(spec, key="pre-fence") verified = _reserve(authority, candidate) assert isinstance(verified, VerifiedDiagnosticReservation) activated = candidate.activated assert activated is not None authority.activation_path.write_text( json.dumps(activated.to_persisted_record(), ensure_ascii=False, sort_keys=True), encoding="utf-8", ) authority.activation_path.chmod(0o600) loader = DurableDiagnosticActivationLoader(authority) authority.detach(generation=1, state="detaching") terminal = authority.profile_root / "diagnostic" / "diagnostic-session-terminal.json" terminal_bytes = terminal.read_bytes() replacement = terminal.with_name("diagnostic-session-terminal.replacement") before_provider = authority.rows(candidate.dedupe_key) original_read = os.read swapped = False def read_and_replace(fd: int, size: int) -> bytes: nonlocal swapped chunk = original_read(fd, size) if not swapped: replacement.write_bytes(terminal_bytes) replacement.chmod(0o600) os.replace(replacement, terminal) swapped = True return chunk monkeypatch.setattr(os, "read", read_and_replace) with pytest.raises(DiagnosticAuthorityError): _reserve(authority, _candidate(spec, key="post-fence-reserve")) assert swapped with pytest.raises(DiagnosticAuthorityError): authority.verify_provider_start( verified, deadline_monotonic=time.monotonic() + 5, lock_token=_DIAGNOSTIC_HOST_ADMISSION_TOKEN, activated=activated, activation_loader=loader, ) assert authority.rows("post-fence-reserve") == () assert authority.rows(candidate.dedupe_key) == before_provider def test_durable_activation_loader_is_closed_and_pinned(tmp_path: Path): authority, spec, _ = _authority(tmp_path) activated = _candidate(spec).activated assert activated is not None authority.activation_path.write_text( json.dumps(activated.to_persisted_record(), ensure_ascii=False, sort_keys=True), encoding="utf-8", ) authority.activation_path.chmod(0o600) loader = DurableDiagnosticActivationLoader(authority) assert loader.load_activated_delivery("s") == activated replacement = dict(activated.to_persisted_record()) replacement["rendered_body"] = "changed" authority.activation_path.write_text( json.dumps(replacement, ensure_ascii=False, sort_keys=True), encoding="utf-8", ) with pytest.raises(DiagnosticAuthorityError): loader.load_activated_delivery("s") def test_dormant_session_starts_without_delivery_lifecycle(tmp_path: Path): spec = _spec() session = DiagnosticSession( "dormant", DiagnosticSessionState.PREPARED, 1, "boot", spec.spec_digest, spec.authority_digest, spec.diagnostic_transport_binding_digest, "2099-01-01T00:00:00Z", ) authority = DiagnosticDeliveryAuthority( session, boot_epoch="boot", profile_root=tmp_path, spec=spec, live_sources=_live_sources(spec), ) assert authority.session.state is DiagnosticSessionState.PREPARED assert authority.rows("never-reserved") == () assert not authority.activation_path.exists()