# SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc from __future__ import annotations import hashlib import hmac import json from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import pytest from solstone.think.journal_io import atomic_replace from solstone.think.models import LOCAL_MODEL from solstone.think.providers import brain_state as brain_state_module from solstone.think.providers.brain_state import ( BRAIN_AGGREGATE_STATES, BRAIN_COMPONENT_STATUSES, BRAIN_EVIDENCE_REASON_CODES, BRAIN_LANES, BRAIN_PROJECTION_ONLY_REASON_CODES, BRAIN_REASON_CODES, BRAIN_REASON_TO_AGGREGATE, CHECKING_TTL, DEFAULT_READY_EVIDENCE_TTL, BrainProbeOutcome, BrainStateConflictError, BrainStateExpectedFingerprintStaleError, BrainStateValidationError, abandon_brain_prerequisite_renewal, abandon_brain_refresh, begin_brain_prerequisite_renewal, begin_brain_refresh, brain_fingerprint_key_path, brain_state_path, build_active_brain_fingerprint, finish_brain_prerequisite_renewal, finish_brain_refresh, inspect_brain_state, project_brain_state, record_brain_runtime_failure, runtime_phase_reason, validate_brain_state_record, ) from solstone.think.providers.runtime_health import runtime_health_path NOW = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc) BUNDLED_RUNTIME_FINGERPRINT = "b" * 64 def _write_config(journal: Path, config: dict[str, Any]) -> None: path = journal / "config" / "journal.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(config), encoding="utf-8") def _cloud_config(key: str = "config-secret", model: str = "gpt-5") -> dict[str, Any]: return { "providers": {"active": {"provider": "openai", "model": model}}, "env": {"OPENAI_API_KEY": key}, } def _local_endpoint_config() -> dict[str, Any]: return { "providers": { "active": {"provider": "local", "model": "local/custom"}, "local": { "endpoint_url": "https://brain.example.test/v1", "served_model_id": "served-model", "credential": "endpoint-secret", }, }, "env": {}, } def _spp_config( *, prior_model: str = "gemini-flash-latest", account_id: str = "acct-a", ) -> dict[str, Any]: credential = "endpoint-secret" endpoint_url = "https://brain.example.test/v1" served_model_id = "served-model" return { "providers": { "active": {"provider": "local", "model": LOCAL_MODEL}, "local": { "endpoint_url": endpoint_url, "served_model_id": served_model_id, "credential": credential, }, }, "services": { "confidential": { "enabled_at": "2026-01-02T03:04:05+00:00", "account_id": account_id, "endpoint_url": endpoint_url, "served_model_id": served_model_id, "credential_created_at": "2026-01-02T03:00:00+00:00", "credential_fingerprint_sha256": hashlib.sha256( credential.encode("utf-8") ).hexdigest(), "prior_active": {"provider": "google", "model": prior_model}, "prior_local_endpoint": { "endpoint_url": "https://old-local.example.test/v1", "served_model_id": "old-served-model", "credential": "old-secret", }, } }, "env": {}, } def _bundled_config() -> dict[str, Any]: return { "providers": {"active": {"provider": "local", "model": "local/bundled"}}, "env": {}, } def _runtime_inspection( *, status: str = "ok", phase: str = "ready", desired: str | None = BUNDLED_RUNTIME_FINGERPRINT, record_present: bool = True, ) -> dict[str, Any]: record = ( { "schema_version": 1, "provider": "local", "revision": 1, "phase": phase, "reason_code": None, "detail": {}, "desired_fingerprint_sha256": desired, "incarnation": None, "generation": 1, "attempt": 0, "process": None, "updated_at": NOW.isoformat(), "display_deadline_at": None, "owner": None, } if record_present else None ) return { "status": status, "provider": "local", "record_kind": "health", "path": "/tmp/runtime/local.json", "record": record, "reason_code": None, "error": None, } def _patch_bundled_runtime( monkeypatch: pytest.MonkeyPatch, desired: str = BUNDLED_RUNTIME_FINGERPRINT, ) -> None: monkeypatch.setattr( brain_state_module, "_bundled_runtime_fingerprint_sha", lambda: desired ) def _component( now: datetime = NOW, *, expires_at: datetime | None = None ) -> dict[str, Any]: return { "status": "ok", "observed_at": now.isoformat(), "expires_at": (expires_at or now + DEFAULT_READY_EVIDENCE_TTL).isoformat(), } def _ready_outcome(now: datetime = NOW) -> BrainProbeOutcome: return { "configuration": _component(now), "lane_prerequisites": _component(now), "generate": _component(now), "cogitate": _component(now), } def _deep_values(value: Any) -> list[Any]: found: list[Any] = [] if isinstance(value, dict): for key, child in value.items(): found.append(key) found.extend(_deep_values(child)) elif isinstance(value, list): for child in value: found.extend(_deep_values(child)) else: found.append(value) return found def _read_raw_record(journal: Path) -> dict[str, Any]: return json.loads( brain_state_path(journal_path=journal).read_text(encoding="utf-8") ) def _write_ready_record(journal: Path, config: dict[str, Any]) -> None: _write_config(journal, config) permit = begin_brain_refresh(NOW, journal_path=journal) assert permit is not None finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=journal) def _write_runtime_health_record( journal: Path, *, desired: str | None = BUNDLED_RUNTIME_FINGERPRINT, phase: str = "ready", ) -> None: runtime_health = _runtime_inspection(phase=phase, desired=desired) record = runtime_health["record"] assert record is not None path = runtime_health_path("local", journal_path=journal) path.parent.mkdir(parents=True, exist_ok=True) atomic_replace(path, json.dumps(record), mode=0o600) def _current_fingerprint(journal: Path, config: dict[str, Any]) -> str: key = brain_fingerprint_key_path(journal_path=journal).read_bytes() result = build_active_brain_fingerprint(config, hmac_key=key) assert result["fingerprint_sha256"] is not None return result["fingerprint_sha256"] def _health_snapshot(journal: Path) -> dict[str, tuple[int, bytes]]: root = journal / "health" if not root.exists(): return {} snapshot: dict[str, tuple[int, bytes]] = {} for path in sorted(root.rglob("*")): if path.is_file(): snapshot[path.relative_to(root).as_posix()] = ( path.stat().st_mtime_ns, path.read_bytes(), ) return snapshot def test_vocabularies_and_reason_mapping_are_closed() -> None: expected_reasons = { "brain_check_in_progress", "thinking_engine_not_chosen", "provider_key_missing", "endpoint_configuration_incomplete", "gpu_unavailable", "local_runtime_not_ready", "local_artifact_not_ready", "attestation_not_verified", "nvattest_install_in_progress", "nvattest_platform_unsupported", "nvattest_unavailable", "provider_key_invalid", "model_not_found", "provider_quota_exceeded", "provider_request_rejected", "provider_unavailable", "network_unreachable", "endpoint_unreachable", "endpoint_contract_failed", "chat_timeout", "provider_response_invalid", "cogitate_terminal_error", "attestation_rejected", "attestation_expired", "nvattest_install_failed", "nvattest_integrity_failed", "local_server_unhealthy", "configuration_invalid", "fingerprint_key_unavailable", "brain_record_missing", "brain_record_invalid", "brain_record_unavailable", "brain_record_stale", "brain_check_interrupted", "brain_config_changed", "brain_run_superseded", "probe_internal_error", "probe_output_starved", "local_runtime_state_invalid", "local_runtime_state_unavailable", "local_runtime_state_stale", "local_runtime_fingerprint_mismatch", } assert BRAIN_AGGREGATE_STATES == { "ready", "checking", "blocked", "unhealthy", "unknown", } assert BRAIN_COMPONENT_STATUSES == { "ok", "blocked", "failed", "unknown", "not_attempted", } assert BRAIN_LANES == {"none", "bundled", "spp", "byo-cloud", "byo-endpoint"} assert BRAIN_REASON_CODES == expected_reasons assert not (BRAIN_REASON_CODES & BRAIN_AGGREGATE_STATES) assert not (BRAIN_REASON_CODES & BRAIN_COMPONENT_STATUSES) assert set(BRAIN_REASON_TO_AGGREGATE) == BRAIN_REASON_CODES assert set(BRAIN_REASON_TO_AGGREGATE.values()) <= BRAIN_AGGREGATE_STATES evidence_reasons = frozenset().union(*BRAIN_EVIDENCE_REASON_CODES.values()) assert len(evidence_reasons) == 32 assert len(BRAIN_PROJECTION_ONLY_REASON_CODES) == 10 assert evidence_reasons | BRAIN_PROJECTION_ONLY_REASON_CODES == BRAIN_REASON_CODES assert not (evidence_reasons & BRAIN_PROJECTION_ONLY_REASON_CODES) def test_closed_schema_rejects_lanes_matrix(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) raw = json.loads( brain_state_path(journal_path=tmp_path).read_text(encoding="utf-8") ) raw["lanes"] = {} with pytest.raises(BrainStateValidationError): validate_brain_state_record(raw) raw = _read_raw_record(tmp_path) raw["active_lane"] = "unknown" with pytest.raises(BrainStateValidationError): validate_brain_state_record(raw) raw = _read_raw_record(tmp_path) raw["aggregate_state"] = "unhealthy" raw["reason_code"] = "runtime_failed" with pytest.raises(BrainStateValidationError): validate_brain_state_record(raw) def test_none_lane_is_blocked_thinking_engine_not_chosen(tmp_path: Path) -> None: _write_config(tmp_path, {"providers": {"active": {"provider": "none"}}, "env": {}}) assert begin_brain_refresh(NOW, journal_path=tmp_path) is None projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["active_lane"] == "none" assert projection["aggregate_state"] == "blocked" assert projection["reason_code"] == "thinking_engine_not_chosen" def test_endpoint_configuration_incomplete_is_evidence_only() -> None: evidence = { "configuration": { "status": "blocked", "observed_at": NOW.isoformat(), "reason_code": "endpoint_configuration_incomplete", }, "lane_prerequisites": None, "generate": None, "cogitate": None, } record = validate_brain_state_record( { "schema_version": 1, "revision": 1, "aggregate_state": "blocked", "reason_code": "endpoint_configuration_incomplete", "active_lane": "byo-endpoint", "active_provider": "local", "active_model": "local/custom", "fingerprint_sha256": "a" * 64, "checking": None, "evidence": evidence, "runtime_failure_marker": None, "diagnostic": {}, "updated_at": NOW.isoformat(), } ) assert record["reason_code"] == "endpoint_configuration_incomplete" def test_fingerprint_uses_config_env_not_process_env( monkeypatch: pytest.MonkeyPatch, ) -> None: key = b"k" * 32 config = _cloud_config(key="from-config") monkeypatch.setenv("OPENAI_API_KEY", "from-process") first = build_active_brain_fingerprint(config, hmac_key=key) monkeypatch.setenv("OPENAI_API_KEY", "changed-process") second = build_active_brain_fingerprint(config, hmac_key=key) changed_config = _cloud_config(key="changed-config") third = build_active_brain_fingerprint(changed_config, hmac_key=key) assert first["fingerprint_sha256"] == second["fingerprint_sha256"] assert first["fingerprint_sha256"] != third["fingerprint_sha256"] def test_ready_evidence_expires_at_boundary_is_not_ready(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None outcome = _ready_outcome(NOW) for component in outcome.values(): assert component is not None component["expires_at"] = NOW.isoformat() finish_brain_refresh(permit, outcome, NOW, journal_path=tmp_path) projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["aggregate_state"] == "unknown" assert projection["reason_code"] == "brain_record_stale" def test_ok_evidence_requires_expires_at() -> None: outcome = _ready_outcome() assert outcome["configuration"] is not None del outcome["configuration"]["expires_at"] with pytest.raises(BrainStateValidationError): validate_brain_state_record( { "schema_version": 1, "revision": 1, "aggregate_state": "ready", "reason_code": None, "active_lane": "byo-cloud", "active_provider": "openai", "active_model": "gpt-5", "fingerprint_sha256": "a" * 64, "checking": None, "evidence": outcome, "runtime_failure_marker": None, "diagnostic": {}, "updated_at": NOW.isoformat(), } ) def test_checking_cross_field_invariant_rejects_mismatches(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None try: raw = _read_raw_record(tmp_path) raw["checking"] = None with pytest.raises(BrainStateValidationError): validate_brain_state_record(raw) raw = _read_raw_record(tmp_path) raw["aggregate_state"] = "unknown" raw["reason_code"] = "local_runtime_not_ready" with pytest.raises(BrainStateValidationError): validate_brain_state_record(raw) finally: permit.release() def test_runtime_failure_ingress_rejects_checking_reason(tmp_path: Path) -> None: result = record_brain_runtime_failure( "brain_check_in_progress", NOW, expected_fingerprint_sha256="a" * 64, component="lane_prerequisites", journal_path=tmp_path, ) assert result["accepted"] is False assert result["rejected_reason"] == "reason_not_recordable" def test_diagnostic_string_must_be_declared_enum(tmp_path: Path) -> None: result = record_brain_runtime_failure( "local_server_unhealthy", NOW, expected_fingerprint_sha256="a" * 64, component="lane_prerequisites", diagnostic={"phase": "sk-secret-credential"}, journal_path=tmp_path, ) assert result["accepted"] is False assert result["rejected_reason"] == "reason_not_recordable" def test_bundled_runtime_health_ready_allows_ready_projection( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=_runtime_inspection(), ) assert projection["aggregate_state"] == "ready" assert projection["reason_code"] is None def test_inspect_brain_state_is_passive_host_probe_free_for_bundled_runtime( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) _write_runtime_health_record(tmp_path) before = _health_snapshot(tmp_path) def fail_probe(*_args: Any, **_kwargs: Any) -> None: raise AssertionError("passive bundled inspection must not probe host state") from solstone.think.providers import local_cuda, local_install monkeypatch.setattr( brain_state_module, "_bundled_runtime_fingerprint_sha", fail_probe ) monkeypatch.setattr(local_install, "target_fingerprint", fail_probe) monkeypatch.setattr(local_cuda, "resolve_local_backend", fail_probe) if brain_state_module.sys.platform == "darwin": from solstone.think.providers import mlx_install monkeypatch.setattr(mlx_install, "target_fingerprint", fail_probe) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["status"] == "ok" assert inspection["projection"]["aggregate_state"] == "ready" assert inspection["projection"]["reason_code"] is None assert _health_snapshot(tmp_path) == before def test_inspect_brain_state_uses_supplied_config_without_reread( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: config = _cloud_config() _write_ready_record(tmp_path, config) monkeypatch.setattr( brain_state_module, "read_journal_config", lambda *_args, **_kwargs: (_ for _ in ()).throw( AssertionError("config reread") ), ) inspection = inspect_brain_state( NOW, journal_path=tmp_path, config=config, ) assert inspection["projection"]["aggregate_state"] == "ready" assert inspection["projection"]["active_lane"] == "byo-cloud" def test_inspect_brain_state_faults_do_not_modify_health( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: cases: list[tuple[str, Any, str]] = [ ("config-oserror", OSError("config unavailable"), "configuration_invalid"), ( "config-corrupt", brain_state_module.CorruptConfigError(tmp_path / "config" / "journal.json"), "configuration_invalid", ), ] for _name, exc, expected_reason in cases: monkeypatch.setattr( brain_state_module, "read_journal_config", lambda _journal_path=None, exc=exc: (_ for _ in ()).throw(exc), ) before = _health_snapshot(tmp_path) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["aggregate_state"] == "unknown" assert inspection["projection"]["reason_code"] == expected_reason assert _health_snapshot(tmp_path) == before monkeypatch.undo() _write_ready_record(tmp_path, _cloud_config()) before = _health_snapshot(tmp_path) monkeypatch.setattr( brain_state_module, "_read_fingerprint_key", lambda _path: (_ for _ in ()).throw(OSError("key unavailable")), ) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["reason_code"] == "fingerprint_key_unavailable" assert _health_snapshot(tmp_path) == before monkeypatch.undo() _write_config(tmp_path, _cloud_config()) before = _health_snapshot(tmp_path) monkeypatch.setattr( brain_state_module, "read_json", lambda _path, **_kwargs: (_ for _ in ()).throw(OSError("record unavailable")), ) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["reason_code"] == "brain_record_unavailable" assert _health_snapshot(tmp_path) == before monkeypatch.undo() _write_config(tmp_path, _cloud_config()) brain_state_path(journal_path=tmp_path).parent.mkdir(parents=True, exist_ok=True) brain_state_path(journal_path=tmp_path).write_bytes(b"{") before = _health_snapshot(tmp_path) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["reason_code"] == "brain_record_invalid" assert _health_snapshot(tmp_path) == before brain_state_path(journal_path=tmp_path).unlink() _patch_bundled_runtime(monkeypatch) _write_ready_record(tmp_path, _bundled_config()) for runtime_status, expected_reason in ( ("corrupt", "local_runtime_state_invalid"), ("unavailable", "local_runtime_state_unavailable"), ): before = _health_snapshot(tmp_path) monkeypatch.setattr( brain_state_module, "inspect_runtime_health", lambda _provider, *, journal_path=None, runtime_status=runtime_status: ( _runtime_inspection( status=runtime_status, record_present=False, ) ), ) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["reason_code"] == expected_reason assert _health_snapshot(tmp_path) == before monkeypatch.undo() _patch_bundled_runtime(monkeypatch) _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None try: before = _health_snapshot(tmp_path) monkeypatch.setattr( brain_state_module, "probe_file_lease_held", lambda _path: (_ for _ in ()).throw(OSError("lease unavailable")), ) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["reason_code"] == "brain_check_interrupted" assert _health_snapshot(tmp_path) == before finally: permit.release() @pytest.mark.parametrize( ("config", "expected_lane"), [ (_cloud_config(), "byo-cloud"), (_local_endpoint_config(), "byo-endpoint"), ], ) def test_non_bundled_inspection_never_reads_runtime( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config: dict[str, Any], expected_lane: str, ) -> None: _write_ready_record(tmp_path, config) def fail_runtime(*_args: Any, **_kwargs: Any) -> None: raise AssertionError("runtime inspector should not be called") monkeypatch.setattr(brain_state_module, "inspect_runtime_health", fail_runtime) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["aggregate_state"] == "ready" assert inspection["projection"]["active_lane"] == expected_lane def test_bundled_runtime_health_phase_lattice( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() cases = { "not-desired": "local_runtime_not_ready", "observing": "local_runtime_not_ready", "artifact-not-ready": "local_artifact_not_ready", "host-blocked": "local_runtime_not_ready", "starting": "local_runtime_not_ready", "warming": "local_runtime_not_ready", "backoff": "local_runtime_not_ready", "retry-requested": "local_runtime_not_ready", "ready": None, "ready-proof-unavailable": "local_runtime_state_unavailable", "stop-deferred": "local_runtime_not_ready", "stopping": "local_runtime_not_ready", "stopped": "local_runtime_not_ready", "failed": "local_server_unhealthy", "cleanup-failed": "local_server_unhealthy", "state-corrupt": "local_runtime_state_invalid", "state-unavailable": "local_runtime_state_unavailable", } for phase, expected_reason in cases.items(): projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=_runtime_inspection(phase=phase), ) assert runtime_phase_reason(phase) == expected_reason assert projection["reason_code"] == expected_reason assert (projection["aggregate_state"] == "ready") is (expected_reason is None) def test_bundled_runtime_health_reason_overrides_and_disagreements( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() cases = [ ("host-blocked", "gpu-unavailable", "gpu_unavailable", "blocked", False), ( "host-blocked", "gpu-probe-failed", "local_runtime_state_unavailable", "unknown", False, ), ( "artifact-not-ready", "artifact-missing", "local_artifact_not_ready", "blocked", False, ), ( "artifact-not-ready", "install-in-progress", "local_runtime_not_ready", "blocked", True, ), ("ready", "artifact-missing", "local_runtime_state_invalid", "unknown", False), ("ready", None, None, "ready", False), ] for ( phase, runtime_reason, expected_reason, expected_state, expected_progressing, ) in cases: runtime_health = _runtime_inspection(phase=phase) assert runtime_health["record"] is not None runtime_health["record"]["reason_code"] = runtime_reason projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=runtime_health, ) assert projection["reason_code"] == expected_reason assert projection["aggregate_state"] == expected_state assert projection["runtime_transition_in_progress"] is expected_progressing def test_bundled_runtime_health_invalid_and_unavailable_records( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() cases = [ ( _runtime_inspection(status="corrupt", record_present=False), "local_runtime_state_invalid", ), ( _runtime_inspection(status="unavailable", record_present=False), "local_runtime_state_unavailable", ), ( _runtime_inspection(phase="ready", desired=None), "local_runtime_state_invalid", ), ] for runtime_health, expected_reason in cases: projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=runtime_health, ) assert projection["aggregate_state"] == "unknown" assert projection["reason_code"] == expected_reason def test_bundled_runtime_desired_change_invalidates_old_brain_evidence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=_runtime_inspection(desired="c" * 64), ) assert projection["aggregate_state"] == "unknown" assert projection["reason_code"] == "brain_config_changed" def test_runtime_transition_in_progress_is_closed( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _patch_bundled_runtime(monkeypatch) config = _bundled_config() _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() for phase in ("observing", "starting", "warming", "retry-requested"): projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=_runtime_inspection(phase=phase), ) assert projection["runtime_transition_in_progress"] is True runtime_health = _runtime_inspection(phase="artifact-not-ready") assert runtime_health["record"] is not None runtime_health["record"]["reason_code"] = "install-in-progress" projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=runtime_health, ) assert projection["runtime_transition_in_progress"] is True projection = project_brain_state( record, NOW, config=config, hmac_key=key, refresh_permit_active=False, runtime_health=_runtime_inspection(phase="backoff"), ) assert projection["runtime_transition_in_progress"] is False cloud_config = _cloud_config() _write_ready_record(tmp_path, cloud_config) cloud_record = validate_brain_state_record(_read_raw_record(tmp_path)) cloud_key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() projection = project_brain_state( cloud_record, NOW, config=cloud_config, hmac_key=cloud_key, refresh_permit_active=False, runtime_health=_runtime_inspection(phase="starting"), ) assert projection["runtime_transition_in_progress"] is False def test_config_change_invalidates_prior_ready_without_refresh(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config(key="first")) record = inspect_brain_state(NOW, journal_path=tmp_path)["record"] assert record is not None key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() projection = project_brain_state( record, NOW, config=_cloud_config(key="second"), hmac_key=key, refresh_permit_active=False, runtime_health=None, ) assert projection["aggregate_state"] == "unknown" assert projection["reason_code"] == "brain_config_changed" def test_spp_fingerprint_ignores_confidential_restore_snapshots( tmp_path: Path, ) -> None: config = _spp_config(prior_model="gemini-flash-latest") _write_ready_record(tmp_path, config) record = validate_brain_state_record(_read_raw_record(tmp_path)) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() before = build_active_brain_fingerprint(config, hmac_key=key) pinned = json.loads(json.dumps(config)) confidential = pinned["services"]["confidential"] confidential["prior_active"] = { "provider": "google", "model": "gemini-3.5-flash", } confidential["prior_local_endpoint"] = { "endpoint_url": "https://replacement.example.test/v1", "served_model_id": "replacement-served-model", "credential": "replacement-secret", } after = build_active_brain_fingerprint(pinned, hmac_key=key) assert before["active_lane"] == "spp" assert after["active_lane"] == "spp" assert before["fingerprint_sha256"] == after["fingerprint_sha256"] projection = project_brain_state( record, NOW, config=pinned, hmac_key=key, refresh_permit_active=False, runtime_health=None, ) assert projection["aggregate_state"] == "ready" def test_spp_fingerprint_tracks_confidential_active_provenance() -> None: key = b"k" * 32 before = build_active_brain_fingerprint( _spp_config(account_id="acct-a"), hmac_key=key ) after = build_active_brain_fingerprint( _spp_config(account_id="acct-b"), hmac_key=key ) assert before["active_lane"] == "spp" assert after["active_lane"] == "spp" assert before["fingerprint_sha256"] != after["fingerprint_sha256"] def test_active_model_changes_fingerprint_but_byo_memory_does_not() -> None: key = b"k" * 32 config = { "providers": { "active": {"provider": "google", "model": "gemini-3.5-flash"}, "byo_models": {"google": "gemini-flash-latest"}, }, "env": {"GOOGLE_API_KEY": "google-secret"}, } active_changed = json.loads(json.dumps(config)) active_changed["providers"]["active"]["model"] = "gemini-3.1-flash-lite" remembered_changed = json.loads(json.dumps(config)) remembered_changed["providers"]["byo_models"]["google"] = "gemini-3.5-flash" before = build_active_brain_fingerprint(config, hmac_key=key) active_after = build_active_brain_fingerprint(active_changed, hmac_key=key) remembered_after = build_active_brain_fingerprint(remembered_changed, hmac_key=key) assert before["fingerprint_sha256"] != active_after["fingerprint_sha256"] assert before["fingerprint_sha256"] == remembered_after["fingerprint_sha256"] def test_unresolved_lane_begin_refresh_does_not_persist(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) state_path = brain_state_path(journal_path=tmp_path) before_bytes = state_path.read_bytes() before_revision = _read_raw_record(tmp_path)["revision"] configs = [ {"providers": {"active": {"provider": "mystery", "model": "x"}}, "env": {}}, { "providers": { "active": {"provider": "local", "model": "local/custom"}, "local": {"endpoint_url": "https://brain.example.test/v1"}, }, "env": {}, }, { **_local_endpoint_config(), "services": {"confidential": {"prior_active": {"provider": "openai"}}}, }, ] for config in configs: _write_config(tmp_path, config) assert ( begin_brain_refresh(NOW + timedelta(seconds=1), journal_path=tmp_path) is None ) assert state_path.read_bytes() == before_bytes assert _read_raw_record(tmp_path)["revision"] == before_revision def test_runtime_failure_ingress_rejects_stale_fingerprint_without_write( tmp_path: Path, ) -> None: config_f1 = _cloud_config(key="first") _write_ready_record(tmp_path, config_f1) state_path = brain_state_path(journal_path=tmp_path) prior_bytes = state_path.read_bytes() prior_revision = _read_raw_record(tmp_path)["revision"] expected_f1 = _current_fingerprint(tmp_path, config_f1) _write_config(tmp_path, _cloud_config(key="second")) result = record_brain_runtime_failure( "provider_unavailable", NOW, expected_fingerprint_sha256=expected_f1, component="generate", journal_path=tmp_path, ) assert result["accepted"] is False assert result["rejected_reason"] == "fingerprint_mismatch" assert state_path.read_bytes() == prior_bytes assert _read_raw_record(tmp_path)["revision"] == prior_revision def test_begin_refresh_expected_active_fingerprint_stales_after_switch( tmp_path: Path, ) -> None: original = _spp_config(account_id="acct-a") _write_ready_record(tmp_path, original) expected = _current_fingerprint(tmp_path, original) before_record = brain_state_path(journal_path=tmp_path).read_bytes() before_key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() _write_config(tmp_path, _spp_config(account_id="acct-b")) with pytest.raises(BrainStateExpectedFingerprintStaleError): begin_brain_refresh( NOW + timedelta(seconds=1), expected_active_fingerprint_sha256=expected, journal_path=tmp_path, ) assert brain_state_path(journal_path=tmp_path).read_bytes() == before_record assert brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() == before_key def test_begin_refresh_expected_absent_fingerprint_bootstraps_key( tmp_path: Path, ) -> None: _write_config(tmp_path, _spp_config()) assert not brain_fingerprint_key_path(journal_path=tmp_path).exists() permit = begin_brain_refresh( NOW, expect_active_fingerprint_absent=True, journal_path=tmp_path, ) assert permit is not None assert brain_fingerprint_key_path(journal_path=tmp_path).exists() record = finish_brain_refresh( permit, _ready_outcome(NOW + timedelta(seconds=1)), NOW + timedelta(seconds=1), journal_path=tmp_path, ) assert record["aggregate_state"] == "ready" assert record["active_lane"] == "spp" def test_begin_refresh_expected_absent_fingerprint_stales_when_key_exists( tmp_path: Path, ) -> None: _write_ready_record(tmp_path, _spp_config()) before_record = brain_state_path(journal_path=tmp_path).read_bytes() before_key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() with pytest.raises(BrainStateExpectedFingerprintStaleError): begin_brain_refresh( NOW + timedelta(seconds=1), expect_active_fingerprint_absent=True, journal_path=tmp_path, ) assert brain_state_path(journal_path=tmp_path).read_bytes() == before_record assert brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() == before_key def test_prerequisite_renewal_preserves_same_fingerprint_model_evidence( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) before = _read_raw_record(tmp_path) expected = _current_fingerprint(tmp_path, config) begin = begin_brain_prerequisite_renewal( NOW + timedelta(minutes=1), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert begin["status"] == "started" checking = _read_raw_record(tmp_path) assert checking["aggregate_state"] == "checking" assert checking["evidence"]["generate"] == before["evidence"]["generate"] assert checking["evidence"]["cogitate"] == before["evidence"]["cogitate"] finish_now = NOW + timedelta(minutes=2) record = finish_brain_prerequisite_renewal( begin["permit"], _component(finish_now, expires_at=finish_now + timedelta(minutes=10)), finish_now, journal_path=tmp_path, ) assert record["aggregate_state"] == "ready" assert record["revision"] == before["revision"] + 2 assert record["evidence"]["configuration"] == before["evidence"]["configuration"] assert record["evidence"]["generate"] == before["evidence"]["generate"] assert record["evidence"]["cogitate"] == before["evidence"]["cogitate"] assert ( record["evidence"]["lane_prerequisites"]["observed_at"] == finish_now.isoformat() ) def test_prerequisite_renewal_refuses_expired_model_evidence(tmp_path: Path) -> None: config = _spp_config() _write_ready_record(tmp_path, config) raw = _read_raw_record(tmp_path) raw["evidence"]["generate"]["expires_at"] = (NOW - timedelta(seconds=1)).isoformat() atomic_replace(brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600) expected = _current_fingerprint(tmp_path, config) result = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert result["status"] == "unsafe" assert _read_raw_record(tmp_path)["revision"] == raw["revision"] def test_prerequisite_renewal_reports_busy_when_refresh_lease_is_held( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) holder = begin_brain_refresh(NOW + timedelta(seconds=1), journal_path=tmp_path) assert holder is not None try: result = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=2), expected_fingerprint_sha256=_current_fingerprint(tmp_path, config), journal_path=tmp_path, ) finally: holder.release() assert result["status"] == "busy" def test_prerequisite_renewal_refuses_non_spp_lane(tmp_path: Path) -> None: config = _cloud_config() _write_ready_record(tmp_path, config) before = brain_state_path(journal_path=tmp_path).read_bytes() result = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256=_current_fingerprint(tmp_path, config), journal_path=tmp_path, ) assert result["status"] == "unsafe" assert brain_state_path(journal_path=tmp_path).read_bytes() == before @pytest.mark.parametrize( "mutation", ( lambda raw: raw["evidence"].__setitem__("generate", None), lambda raw: raw["evidence"]["generate"].__setitem__("status", "failed"), lambda raw: raw["evidence"]["generate"].__setitem__("expires_at", "not-time"), ), ) def test_prerequisite_renewal_refuses_missing_malformed_or_non_ok_model_evidence( tmp_path: Path, mutation, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) raw = _read_raw_record(tmp_path) mutation(raw) atomic_replace(brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600) before = brain_state_path(journal_path=tmp_path).read_bytes() result = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256=_current_fingerprint(tmp_path, config), journal_path=tmp_path, ) assert result["status"] == "unsafe" assert brain_state_path(journal_path=tmp_path).read_bytes() == before def test_prerequisite_renewal_expected_fingerprint_mismatch_is_no_write( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) before = brain_state_path(journal_path=tmp_path).read_bytes() result = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256="0" * 64, journal_path=tmp_path, ) assert result["status"] == "unsafe" assert result["reason"] == "fingerprint_mismatch" assert brain_state_path(journal_path=tmp_path).read_bytes() == before def test_prerequisite_renewal_recovers_orphaned_checking_after_lease_released( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) first = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert first["status"] == "started" first["permit"].release() orphaned = _read_raw_record(tmp_path) assert orphaned["aggregate_state"] == "checking" second = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=2), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert second["status"] == "started" recovered = _read_raw_record(tmp_path) assert recovered["revision"] == orphaned["revision"] + 1 assert recovered["checking"]["run_id"] != orphaned["checking"]["run_id"] second["permit"].release() def test_prerequisite_renewal_expired_permit_conflicts_and_releases( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) begin = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert begin["status"] == "started" permit = begin["permit"] with pytest.raises(BrainStateConflictError): finish_brain_prerequisite_renewal( permit, _component( permit.expires_at, expires_at=permit.expires_at + timedelta(minutes=10) ), permit.expires_at, journal_path=tmp_path, ) retry = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert retry["status"] == "started" retry["permit"].release() def test_prerequisite_renewal_conflicts_on_revision_drift_and_releases( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) begin = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert begin["status"] == "started" raw = _read_raw_record(tmp_path) raw["revision"] += 1 raw["checking"]["checking_revision"] = raw["revision"] atomic_replace(brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600) with pytest.raises(BrainStateConflictError): finish_brain_prerequisite_renewal( begin["permit"], _component(NOW + timedelta(seconds=1)), NOW + timedelta(seconds=1), journal_path=tmp_path, ) retry = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=2), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert retry["status"] == "started" retry["permit"].release() def test_prerequisite_renewal_conflicts_on_runtime_marker_drift_and_releases( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) begin = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert begin["status"] == "started" raw = _read_raw_record(tmp_path) raw["checking"]["runtime_failure_marker_seen"] = "changed-marker" atomic_replace(brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600) with pytest.raises(BrainStateConflictError): finish_brain_prerequisite_renewal( begin["permit"], _component(NOW + timedelta(seconds=1)), NOW + timedelta(seconds=1), journal_path=tmp_path, ) retry = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=2), expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert retry["status"] == "started" retry["permit"].release() def test_prerequisite_renewal_conflicts_on_fingerprint_drift(tmp_path: Path) -> None: config = _spp_config(account_id="acct-a") _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) begin = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=expected, journal_path=tmp_path, ) assert begin["status"] == "started" _write_config(tmp_path, _spp_config(account_id="acct-b")) with pytest.raises(BrainStateConflictError): finish_brain_prerequisite_renewal( begin["permit"], _component(NOW, expires_at=NOW + timedelta(minutes=10)), NOW, journal_path=tmp_path, ) retry = begin_brain_prerequisite_renewal( NOW + timedelta(seconds=1), expected_fingerprint_sha256=_current_fingerprint( tmp_path, _spp_config(account_id="acct-b") ), journal_path=tmp_path, ) assert retry["status"] == "unsafe" def test_prerequisite_renewal_abandon_records_failure_without_losing_model_evidence( tmp_path: Path, ) -> None: config = _spp_config() _write_ready_record(tmp_path, config) before = _read_raw_record(tmp_path) begin = begin_brain_prerequisite_renewal( NOW, expected_fingerprint_sha256=_current_fingerprint(tmp_path, config), journal_path=tmp_path, ) assert begin["status"] == "started" record = abandon_brain_prerequisite_renewal( begin["permit"], "probe_internal_error", NOW + timedelta(seconds=1), journal_path=tmp_path, ) assert record["reason_code"] == "probe_internal_error" assert record["evidence"]["lane_prerequisites"]["reason_code"] == ( "probe_internal_error" ) assert record["evidence"]["generate"] == before["evidence"]["generate"] assert record["evidence"]["cogitate"] == before["evidence"]["cogitate"] def test_key_replacement_invalidates_prior_ready_record(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["aggregate_state"] == "ready" key_path = brain_fingerprint_key_path(journal_path=tmp_path) key_path.unlink() atomic_replace(key_path, b"z" * 32, mode=0o600) projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["aggregate_state"] == "unknown" assert projection["reason_code"] == "brain_config_changed" def test_mode_repair_uses_atomic_replace_for_brain_and_key(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) key_path = brain_fingerprint_key_path(journal_path=tmp_path) state_path = brain_state_path(journal_path=tmp_path) key_path.chmod(0o644) state_path.chmod(0o644) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None permit.release() assert (key_path.stat().st_mode & 0o777) == 0o600 assert (state_path.stat().st_mode & 0o777) == 0o600 def test_runtime_failure_survives_older_refresh_finalization(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None failure_result = record_brain_runtime_failure( "local_server_unhealthy", NOW, expected_fingerprint_sha256=permit.fingerprint_sha256, component="lane_prerequisites", diagnostic={"phase": "failed"}, journal_path=tmp_path, ) assert failure_result["accepted"] is True failure = failure_result["record"] assert failure is not None assert failure["aggregate_state"] == "unhealthy" with pytest.raises(BrainStateConflictError): finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=tmp_path) projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["aggregate_state"] == "unhealthy" assert projection["reason_code"] == "local_server_unhealthy" newer = begin_brain_refresh(NOW, journal_path=tmp_path) assert newer is not None finish_brain_refresh(newer, _ready_outcome(), NOW, journal_path=tmp_path) assert ( inspect_brain_state(NOW, journal_path=tmp_path)["projection"]["aggregate_state"] == "ready" ) def test_runtime_failure_preserves_other_same_fingerprint_evidence( tmp_path: Path, ) -> None: config = _cloud_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) generate_result = record_brain_runtime_failure( "provider_unavailable", NOW, expected_fingerprint_sha256=expected, component="generate", journal_path=tmp_path, ) assert generate_result["accepted"] is True generate_record = generate_result["record"] assert generate_record is not None assert generate_record["evidence"]["generate"] is not None assert generate_record["evidence"]["generate"]["status"] == "failed" assert generate_record["evidence"]["cogitate"] is not None assert generate_record["evidence"]["cogitate"]["status"] == "ok" _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) cogitate_result = record_brain_runtime_failure( "cogitate_terminal_error", NOW, expected_fingerprint_sha256=expected, component="cogitate", journal_path=tmp_path, ) assert cogitate_result["accepted"] is True cogitate_record = cogitate_result["record"] assert cogitate_record is not None assert cogitate_record["evidence"]["generate"] is not None assert cogitate_record["evidence"]["generate"]["status"] == "ok" assert cogitate_record["evidence"]["cogitate"] is not None assert cogitate_record["evidence"]["cogitate"]["status"] == "failed" def test_runtime_failure_provider_request_rejected_transitions_ready_to_unhealthy( tmp_path: Path, ) -> None: config = _cloud_config() _write_ready_record(tmp_path, config) assert ( inspect_brain_state(NOW, journal_path=tmp_path)["projection"]["aggregate_state"] == "ready" ) expected = _current_fingerprint(tmp_path, config) result = record_brain_runtime_failure( "provider_request_rejected", NOW, expected_fingerprint_sha256=expected, component="generate", journal_path=tmp_path, ) assert result["accepted"] is True record = result["record"] assert record is not None assert record["aggregate_state"] == "unhealthy" assert record["evidence"]["generate"] is not None assert record["evidence"]["generate"]["status"] == "failed" assert record["evidence"]["generate"]["reason_code"] == "provider_request_rejected" assert ( inspect_brain_state(NOW, journal_path=tmp_path)["projection"]["aggregate_state"] == "unhealthy" ) def test_runtime_failure_drops_evidence_from_prior_fingerprint(tmp_path: Path) -> None: config_f1 = _cloud_config(key="first") _write_ready_record(tmp_path, config_f1) config_f2 = _cloud_config(key="second") _write_config(tmp_path, config_f2) expected_f2 = _current_fingerprint(tmp_path, config_f2) result = record_brain_runtime_failure( "provider_unavailable", NOW, expected_fingerprint_sha256=expected_f2, component="generate", journal_path=tmp_path, ) assert result["accepted"] is True record = result["record"] assert record is not None assert record["fingerprint_sha256"] == expected_f2 assert record["evidence"]["configuration"] is None assert record["evidence"]["lane_prerequisites"] is None assert record["evidence"]["generate"] is not None assert record["evidence"]["cogitate"] is None def test_runtime_failure_creates_or_replaces_missing_prior_record( tmp_path: Path, ) -> None: config = _cloud_config() _write_ready_record(tmp_path, config) expected = _current_fingerprint(tmp_path, config) state_path = brain_state_path(journal_path=tmp_path) for prior_bytes in (None, b"{"): if prior_bytes is None: state_path.unlink(missing_ok=True) else: state_path.write_bytes(prior_bytes) result = record_brain_runtime_failure( "provider_unavailable", NOW, expected_fingerprint_sha256=expected, component="generate", journal_path=tmp_path, ) assert result["accepted"] is True record = result["record"] assert record is not None assert record["revision"] == 1 assert record["evidence"]["configuration"] is None assert record["evidence"]["generate"] is not None def test_accepted_mutations_increment_revision_exactly_once(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None record = _read_raw_record(tmp_path) assert record["revision"] == 1 finished = finish_brain_refresh( permit, _ready_outcome(), NOW, journal_path=tmp_path ) assert finished["revision"] == 2 expected = finished["fingerprint_sha256"] assert expected is not None failure_result = record_brain_runtime_failure( "local_server_unhealthy", NOW, expected_fingerprint_sha256=expected, component="lane_prerequisites", diagnostic={"phase": "failed"}, journal_path=tmp_path, ) assert failure_result["accepted"] is True failure = failure_result["record"] assert failure is not None assert failure["revision"] == 3 newer = begin_brain_refresh(NOW + timedelta(seconds=1), journal_path=tmp_path) assert newer is not None record = _read_raw_record(tmp_path) assert record["revision"] == 4 abandoned = abandon_brain_refresh( newer, "local_runtime_not_ready", NOW + timedelta(seconds=2), diagnostic={"phase": "starting"}, journal_path=tmp_path, ) assert abandoned["revision"] == 5 def test_rejected_finalization_does_not_increment_revision(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None before = _read_raw_record(tmp_path)["revision"] permit.run_id = "not-the-recorded-run" with pytest.raises(BrainStateConflictError): finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=tmp_path) assert _read_raw_record(tmp_path)["revision"] == before def test_finalization_rejects_mismatched_run_id(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None permit.run_id = "different-run" with pytest.raises(BrainStateConflictError): finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=tmp_path) def test_finalization_rejects_mismatched_revision(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None permit.checking_revision += 1 with pytest.raises(BrainStateConflictError): finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=tmp_path) def test_finalization_rejects_mismatched_fingerprint(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) permit = begin_brain_refresh(NOW, journal_path=tmp_path) assert permit is not None permit.fingerprint_sha256 = "f" * 64 with pytest.raises(BrainStateConflictError): finish_brain_refresh(permit, _ready_outcome(), NOW, journal_path=tmp_path) def test_run_identity_unique_and_expiry_is_exact_ten_minutes(tmp_path: Path) -> None: _write_config(tmp_path, _cloud_config()) first = begin_brain_refresh(NOW, journal_path=tmp_path) assert first is not None first.release() second = begin_brain_refresh(NOW + timedelta(seconds=1), journal_path=tmp_path) assert second is not None try: assert first.run_id != second.run_id assert first.expires_at == first.started_at + CHECKING_TTL assert second.expires_at == second.started_at + CHECKING_TTL finally: second.release() def test_brain_atomic_failure_preserves_prior_record_and_cleans_temp( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _write_ready_record(tmp_path, _cloud_config()) state_path = brain_state_path(journal_path=tmp_path) health_dir = state_path.parent prior_bytes = state_path.read_bytes() prior_revision = _read_raw_record(tmp_path)["revision"] real_replace = brain_state_module.atomic_replace.__globals__["os"].replace def fail_brain_replace(src: str, dst: str) -> None: if Path(dst).name == "brain.json": raise OSError("brain replace failed") real_replace(src, dst) monkeypatch.setattr( "solstone.think.journal_io.atomic.os.replace", fail_brain_replace ) expected = _read_raw_record(tmp_path)["fingerprint_sha256"] assert expected is not None result = record_brain_runtime_failure( "local_server_unhealthy", NOW, expected_fingerprint_sha256=expected, component="lane_prerequisites", diagnostic={"phase": "failed"}, journal_path=tmp_path, ) assert result["accepted"] is False assert result["rejected_reason"] == "state_unavailable" assert state_path.read_bytes() == prior_bytes assert _read_raw_record(tmp_path)["revision"] == prior_revision assert list(health_dir.glob(".tmp_*")) == [] def test_failed_checking_commit_performs_zero_provider_work( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _write_ready_record(tmp_path, _cloud_config()) real_replace = brain_state_module.atomic_replace.__globals__["os"].replace provider_calls: list[str] = [] def fail_brain_replace(src: str, dst: str) -> None: if Path(dst).name == "brain.json": raise OSError("brain replace failed") real_replace(src, dst) def provider_work() -> None: provider_calls.append("called") monkeypatch.setattr( "solstone.think.journal_io.atomic.os.replace", fail_brain_replace ) with pytest.raises(OSError): permit = begin_brain_refresh(NOW + timedelta(seconds=1), journal_path=tmp_path) if permit is not None: provider_work() permit.release() assert provider_calls == [] assert list((tmp_path / "health").glob(".tmp_*")) == [] def test_future_timestamp_never_projects_ready(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) raw = _read_raw_record(tmp_path) assert raw["evidence"]["configuration"] is not None raw["evidence"]["configuration"]["observed_at"] = ( NOW + timedelta(minutes=1) ).isoformat() atomic_replace( brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600, ) projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["aggregate_state"] != "ready" assert projection["reason_code"] == "brain_record_invalid" def test_naive_timestamp_never_projects_ready(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) raw = _read_raw_record(tmp_path) assert raw["evidence"]["configuration"] is not None raw["evidence"]["configuration"]["observed_at"] = "2026-01-02T03:04:05" atomic_replace( brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600, ) inspection = inspect_brain_state(NOW, journal_path=tmp_path) assert inspection["projection"]["aggregate_state"] != "ready" assert inspection["reason_code"] == "brain_record_invalid" def test_internally_inconsistent_timestamp_never_projects_ready( tmp_path: Path, ) -> None: _write_ready_record(tmp_path, _cloud_config()) raw = _read_raw_record(tmp_path) assert raw["evidence"]["configuration"] is not None raw["evidence"]["configuration"]["expires_at"] = ( NOW - timedelta(hours=1) ).isoformat() atomic_replace( brain_state_path(journal_path=tmp_path), json.dumps(raw), mode=0o600, ) projection = inspect_brain_state(NOW, journal_path=tmp_path)["projection"] assert projection["aggregate_state"] != "ready" assert projection["reason_code"] == "brain_record_invalid" def test_clock_rollback_never_projects_ready(tmp_path: Path) -> None: _write_ready_record(tmp_path, _cloud_config()) projection = inspect_brain_state(NOW - timedelta(seconds=1), journal_path=tmp_path)[ "projection" ] assert projection["aggregate_state"] != "ready" assert projection["reason_code"] == "brain_record_invalid" def test_secret_material_never_persists_in_brain_record(tmp_path: Path) -> None: config = _local_endpoint_config() _write_ready_record(tmp_path, config) key = brain_fingerprint_key_path(journal_path=tmp_path).read_bytes() credential = b"endpoint-secret" hmac_digest = hmac.digest(key, credential, "sha256").hex() raw = json.loads( brain_state_path(journal_path=tmp_path).read_text(encoding="utf-8") ) values = {str(value) for value in _deep_values(raw)} assert "endpoint-secret" not in values assert key.hex() not in values assert hmac_digest not in values assert "https://brain.example.test/v1" not in values assert "https://brain.example.test" not in values