diff --git a/solstone/apps/settings/routes.py b/solstone/apps/settings/routes.py index ca10969cf..fd7336c4c 100644 --- a/solstone/apps/settings/routes.py +++ b/solstone/apps/settings/routes.py @@ -591,10 +591,12 @@ def get_transcribe() -> Any: try: from solstone.think.services import spp - confidential_lane_active = spp.confidential_provenance() is not None + # Routing display uses channel usability; dispatch refusal separately + # keys on bare confidential block presence to prevent egress. + confidential_channel_usable = spp.is_confidential_channel_usable(config) resource = transcribe_resource.get_transcribe_resource_payload( configured_backend=configured_backend, - confidential_lane_active=confidential_lane_active, + confidential_lane_active=confidential_channel_usable, confidential_audio=confidential_audio, ) except Exception: diff --git a/solstone/apps/settings/tests/test_transcribe_resource_payload.py b/solstone/apps/settings/tests/test_transcribe_resource_payload.py index 3cfc5497f..d7df35c6f 100644 --- a/solstone/apps/settings/tests/test_transcribe_resource_payload.py +++ b/solstone/apps/settings/tests/test_transcribe_resource_payload.py @@ -183,6 +183,11 @@ def test_transcribe_route_passes_confidential_lane_flag(settings_env, monkeypatc config.setdefault("services", {})["confidential"] = { "enabled_at": "2026-05-24T00:00:00Z" } + config.setdefault("providers", {})["local"] = { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + "credential": "confidential-credential", + } (journal_path / "config" / "journal.json").write_text( json.dumps(config, indent=2) + "\n", encoding="utf-8", diff --git a/solstone/apps/settings/transcribe_resource.py b/solstone/apps/settings/transcribe_resource.py index cc3186384..a285560d9 100644 --- a/solstone/apps/settings/transcribe_resource.py +++ b/solstone/apps/settings/transcribe_resource.py @@ -31,6 +31,7 @@ def get_transcribe_resource_payload( available_bytes = read_available_bytes() floor_bytes = stt_local_floor_bytes() local_backend = local_stt_backend() + # The legacy selector keyword is kept stable; callers pass channel usability. selected_backend = resolve_stt_backend_choice( configured_backend, available_bytes, diff --git a/solstone/observe/transcribe/__init__.py b/solstone/observe/transcribe/__init__.py index 0604d7304..26ecd2744 100644 --- a/solstone/observe/transcribe/__init__.py +++ b/solstone/observe/transcribe/__init__.py @@ -201,6 +201,8 @@ def transcribe( """ from solstone.think.services import spp + # Refusal keys on bare confidential block presence; routing uses channel + # usability earlier, but this gate must still prevent accidental egress. confidential_lane_active = spp.confidential_provenance() is not None if confidential_lane_active: meta = BACKEND_METADATA.get(backend) diff --git a/solstone/observe/transcribe/main.py b/solstone/observe/transcribe/main.py index 72fee9db5..a51207957 100644 --- a/solstone/observe/transcribe/main.py +++ b/solstone/observe/transcribe/main.py @@ -149,7 +149,52 @@ PYANNOTE_OVERLAP_MODEL_SHA256 = OVERLAP_DETECTOR_SHA256 _embedder_session: ort.InferenceSession | None = None -def resolve_default_backend(args: argparse.Namespace, transcribe_config: dict) -> str: +def _join_missing_fields(fields: list[str]) -> str: + if len(fields) == 1: + return fields[0] + if len(fields) == 2: + return f"{fields[0]} and {fields[1]}" + return f"{', '.join(fields[:-1])}, and {fields[-1]}" + + +def _confidential_backend_fallback_reason( + journal_config: dict, + *, + confidential_channel_usable: bool, + confidential_audio: bool, +) -> str: + if confidential_channel_usable and not confidential_audio: + return "confidential audio is disabled" + + from solstone.think.providers.local_endpoint import confidential_provenance_block + + if confidential_provenance_block(dict(journal_config)) is None: + return "confidential lane is inactive" + + providers = journal_config.get("providers") + local = providers.get("local", {}) if isinstance(providers, dict) else {} + if not isinstance(local, dict): + local = {} + missing: list[str] = [] + if not local.get("credential"): + missing.append("credential") + if not str(local.get("endpoint_url") or "").strip(): + missing.append("endpoint URL") + if not str(local.get("served_model_id") or "").strip(): + missing.append("served model ID") + if not missing: + return "confidential channel is incomplete" + return ( + f"confidential channel is incomplete: missing {_join_missing_fields(missing)}" + ) + + +def resolve_default_backend( + args: argparse.Namespace, + transcribe_config: dict, + *, + journal_config: dict | None = None, +) -> str: """Resolve the effective default STT backend once, from a single free-RAM read. Honors explicit CLI/config choices, warns on an explicit local choice below @@ -170,21 +215,25 @@ def resolve_default_backend(args: argparse.Namespace, transcribe_config: dict) - explicit_backend = None from solstone.think.services import spp - confidential_lane_active = spp.confidential_provenance() is not None + if journal_config is None: + journal_config = get_config() + # Routing uses channel usability; the dispatch refusal gate separately keys + # on bare confidential block presence to prevent accidental egress. + confidential_channel_usable = spp.is_confidential_channel_usable(journal_config) confidential_audio = confidential_audio_enabled(transcribe_config) backend = resolve_stt_backend_choice( explicit_backend, available_bytes, floor_bytes=floor_bytes, local_backend=local_backend, - confidential_lane_active=confidential_lane_active, + confidential_lane_active=confidential_channel_usable, confidential_audio_enabled=confidential_audio, ) if explicit_backend == "confidential" and backend != "confidential": - reason = ( - "confidential audio is disabled" - if confidential_lane_active - else "confidential lane is inactive" + reason = _confidential_backend_fallback_reason( + journal_config, + confidential_channel_usable=confidential_channel_usable, + confidential_audio=confidential_audio, ) logging.warning( "Configured STT backend 'confidential' cannot run because %s; using local STT placement", @@ -1452,7 +1501,11 @@ def main(): config = get_config() transcribe_config = config.get("transcribe", {}) - default_backend = resolve_default_backend(args, transcribe_config) + default_backend = resolve_default_backend( + args, + transcribe_config, + journal_config=config, + ) if args.all: processed = 0 diff --git a/solstone/think/services/spp.py b/solstone/think/services/spp.py index 9c0c32f05..117cd9c38 100644 --- a/solstone/think/services/spp.py +++ b/solstone/think/services/spp.py @@ -23,6 +23,7 @@ from solstone.think.journal_config import ( from solstone.think.providers.local_endpoint import ( confidential_provenance_block, normalize_local_endpoint_url, + resolve_local_endpoint_from_config, ) from solstone.think.services.spp_attest.cadence import AttestationSession @@ -305,3 +306,11 @@ def is_confidential_enabled(config: dict[str, Any] | None = None) -> bool: local = config.get("providers", {}).get("local", {}) credential = local.get("credential") if isinstance(local, dict) else None return isinstance(block, dict) and bool(credential) + + +def is_confidential_channel_usable(config: dict[str, Any] | None = None) -> bool: + """Return whether the confidential channel can carry a request now.""" + + config = read_journal_config() if config is None else config + endpoint = resolve_local_endpoint_from_config(config) + return is_confidential_enabled(config) and not endpoint.is_bundled diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index 1f181b940..753ce12fc 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -175,19 +175,21 @@ def linux_stt_uses_parakeet_cpp() -> bool: except RuntimeError: return False - from solstone.think.services import spp - - confidential = spp.confidential_provenance() is not None config = read_journal_config() transcribe = config.get("transcribe", {}) backend = transcribe.get("backend") if isinstance(transcribe, dict) else None + from solstone.think.services import spp + + # Routing uses channel usability; dispatch refusal still keys on bare + # confidential block presence to keep raw audio from accidental egress. + confidential_channel_usable = spp.is_confidential_channel_usable(config) selected = resolve_stt_backend_choice( backend if isinstance(backend, str) else None, read_available_bytes(), floor_bytes=stt_local_floor_bytes(), local_backend=local_stt_backend(), - confidential_lane_active=confidential, + confidential_lane_active=confidential_channel_usable, confidential_audio_enabled=confidential_audio_enabled(transcribe), ) return selected in {"parakeet", "parakeet-cpp"} @@ -2524,7 +2526,7 @@ def _parakeet_platform_can_host() -> bool: def _parakeet_stt_admission_input( - transcribe: dict[str, Any], confidential: bool + transcribe: dict[str, Any], confidential_channel_usable: bool ) -> dict[str, Any]: backend = transcribe.get("backend") if isinstance(transcribe, dict) else None return { @@ -2533,18 +2535,20 @@ def _parakeet_stt_admission_input( "backend": backend if isinstance(backend, str) else None, "local_backend": local_stt_backend(), "floor_bytes": stt_local_floor_bytes(), - "confidential_lane_active": confidential, + "confidential_lane_active": confidential_channel_usable, "confidential_audio_enabled": confidential_audio_enabled(transcribe), } def _parakeet_stt_admission_latch( transcribe: dict[str, Any], - confidential: bool, + confidential_channel_usable: bool, ) -> dict[str, Any]: global _parakeet_admission_retry_epoch - admission_input = _parakeet_stt_admission_input(transcribe, confidential) + admission_input = _parakeet_stt_admission_input( + transcribe, confidential_channel_usable + ) input_json, input_sha = _target_fingerprint_pair(admission_input) try: current = read_runtime_health("parakeet") @@ -2566,7 +2570,7 @@ def _parakeet_stt_admission_latch( available_bytes = ( None if explicit_backend in {"parakeet", "parakeet-cpp", "confidential"} - or confidential + or confidential_channel_usable else read_available_bytes() ) choice = resolve_stt_backend_choice( @@ -2574,14 +2578,14 @@ def _parakeet_stt_admission_latch( available_bytes, floor_bytes=floor_bytes if isinstance(floor_bytes, int) else None, local_backend=local_backend if isinstance(local_backend, str) else None, - confidential_lane_active=confidential, + confidential_lane_active=confidential_channel_usable, confidential_audio_enabled=bool(admission_input["confidential_audio_enabled"]), ) desired = choice in {"parakeet", "parakeet-cpp"} ram_blocked = ( choice == STT_SURFACE and explicit_backend is None - and not confidential + and not confidential_channel_usable and local_backend in {"parakeet", "parakeet-cpp"} and floor_bytes is not None ) @@ -2848,8 +2852,12 @@ def _observe_parakeet_provider_truth() -> ProviderTruthObservation: transcribe = {} from solstone.think.services import spp - confidential = spp.confidential_provenance() is not None - admission_latch = _parakeet_stt_admission_latch(transcribe, confidential) + # Routing uses channel usability; dispatch refusal still keys on bare + # confidential block presence to keep raw audio from accidental egress. + confidential_channel_usable = spp.is_confidential_channel_usable(config) + admission_latch = _parakeet_stt_admission_latch( + transcribe, confidential_channel_usable + ) if admission_latch["blocked"]: return ProviderTruthObservation( provider="parakeet", diff --git a/tests/services/test_spp_storage.py b/tests/services/test_spp_storage.py index 26e0aafb3..0d6933e1b 100644 --- a/tests/services/test_spp_storage.py +++ b/tests/services/test_spp_storage.py @@ -19,6 +19,7 @@ from solstone.think.services.spp import ( JournalNotInitializedError, confidential_provenance, disable_confidential, + is_confidential_channel_usable, is_confidential_enabled, provision_confidential_handoff, ) @@ -47,6 +48,65 @@ def _write_config(journal: Path, config: dict) -> None: seed_journal_config(config, journal) +@pytest.mark.parametrize( + ( + "config", + "expected_enabled", + "expected_usable", + ), + [ + ({}, False, False), + ( + {"services": {"confidential": {"enabled_at": "2026-05-24T00:00:00Z"}}}, + False, + False, + ), + ( + { + "services": {"confidential": {"enabled_at": "2026-05-24T00:00:00Z"}}, + "providers": { + "local": { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + } + }, + }, + False, + False, + ), + ( + { + "services": {"confidential": {"enabled_at": "2026-05-24T00:00:00Z"}}, + "providers": { + "local": { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + "credential": "confidential-credential", + } + }, + }, + True, + True, + ), + ( + { + "services": {"confidential": {"enabled_at": "2026-05-24T00:00:00Z"}}, + "providers": {"local": {"credential": "confidential-credential"}}, + }, + True, + False, + ), + ], +) +def test_confidential_channel_usable_is_provisioned_complete_endpoint( + config: dict, + expected_enabled: bool, + expected_usable: bool, +) -> None: + assert is_confidential_enabled(config) is expected_enabled + assert is_confidential_channel_usable(config) is expected_usable + + @pytest.fixture(autouse=True) def _canonical_active_profile(journal_copy: Path) -> None: config = _read_config(journal_copy) diff --git a/tests/test_no_implicit_cloud.py b/tests/test_no_implicit_cloud.py index df119bcca..83e18fc6a 100644 --- a/tests/test_no_implicit_cloud.py +++ b/tests/test_no_implicit_cloud.py @@ -930,6 +930,7 @@ def test_confidential_stt_posts_only_to_verified_forwarder(tmp_path, monkeypatch def test_confidential_stt_toggle_off_selection_is_immediate(tmp_path, monkeypatch): _empty_journal(tmp_path, monkeypatch) config = _confidential_config(provider_pins=False) + _add_local_endpoint(config) _seed_journal_config(tmp_path, config) transcribe_main = importlib.import_module("solstone.observe.transcribe.main") monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) diff --git a/tests/test_supervisor_parakeet.py b/tests/test_supervisor_parakeet.py index efa929f9d..5149cf8c4 100644 --- a/tests/test_supervisor_parakeet.py +++ b/tests/test_supervisor_parakeet.py @@ -23,11 +23,57 @@ from solstone.think.providers.parakeet_placement import ( PARAKEET_ATT_CONTEXT_ENV, PARAKEET_ATT_CONTEXT_FRAMES, ) +from tests.helpers.journal_config import seed_journal_config from tests.helpers.module_mocks import module_mock _LaunchRecord = dict[str, Any] +def _confidential_block() -> dict[str, Any]: + return { + "enabled_at": "2026-05-24T00:00:00Z", + "account_id": "acct-test", + "endpoint_url": "https://spp.example.test", + "served_model_id": "confidential-model", + "credential_fingerprint_sha256": "fingerprint", + } + + +def _stranded_confidential_stt_config() -> dict[str, Any]: + return { + "services": {"confidential": _confidential_block()}, + "providers": {"local": {}}, + "transcribe": {}, + } + + +def _usable_confidential_stt_config( + transcribe: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "services": {"confidential": _confidential_block()}, + "providers": { + "local": { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + "credential": "confidential-credential", + } + }, + "transcribe": transcribe or {}, + } + + +def _install_supervisor_config( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + config: dict[str, Any], +) -> None: + journal = tmp_path / "journal" + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + seed_journal_config(config, journal) + monkeypatch.setattr(supervisor, "read_journal_config", lambda: config) + + @pytest.fixture(autouse=True) def _isolate_supervisor_threading(monkeypatch): monkeypatch.setattr( @@ -596,20 +642,42 @@ def test_linux_stt_uses_parakeet_cpp_truth_table( transcribe_config["backend"] = backend if not confidential_audio: transcribe_config["confidential_audio"] = False - config = {"transcribe": transcribe_config} if transcribe_config else {} + if confidential: + config = _usable_confidential_stt_config(transcribe_config) + else: + config = {"transcribe": transcribe_config} if transcribe_config else {} monkeypatch.setattr(supervisor, "read_journal_config", lambda: config) monkeypatch.setattr(supervisor, "read_available_bytes", lambda: available_bytes) monkeypatch.setattr(supervisor, "stt_local_floor_bytes", lambda: 4 * 1024**3) monkeypatch.setattr(supervisor, "local_stt_backend", lambda: local_backend) - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"} if confidential else None, - ) monkeypatch.delenv("GOOGLE_API_KEY", raising=False) assert supervisor.linux_stt_uses_parakeet_cpp() is expected +@pytest.mark.parametrize( + ("available_bytes", "expected"), + [ + (2 * 1024**3, False), + (8 * 1024**3, True), + ], +) +def test_linux_stt_uses_parakeet_cpp_stranded_config_follows_local_resources( + monkeypatch, + available_bytes: int, + expected: bool, +) -> None: + config = _stranded_confidential_stt_config() + monkeypatch.setattr(supervisor.sys, "platform", "linux") + monkeypatch.setattr(supervisor.platform, "machine", lambda: "x86_64") + monkeypatch.setattr(supervisor, "read_journal_config", lambda: config) + monkeypatch.setattr(supervisor, "read_available_bytes", lambda: available_bytes) + monkeypatch.setattr(supervisor, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(supervisor, "local_stt_backend", lambda: "parakeet") + + assert supervisor.linux_stt_uses_parakeet_cpp() is expected + + def test_start_parakeet_server_early_returns_for_non_linux( monkeypatch, tmp_path ) -> None: @@ -639,11 +707,7 @@ def test_start_parakeet_server_early_returns_for_other_backend( monkeypatch.setattr( supervisor, "read_journal_config", - lambda: {"transcribe": {"backend": "confidential"}}, - ) - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"}, + lambda: _usable_confidential_stt_config({"backend": "confidential"}), ) observation = supervisor._observe_parakeet_provider_truth() @@ -653,6 +717,55 @@ def test_start_parakeet_server_early_returns_for_other_backend( assert parakeet_server.read_parakeet_placement() == "gpu" +def test_parakeet_truth_stranded_low_ram_reports_host_blocked( + monkeypatch, tmp_path +) -> None: + _install_supervisor_config( + tmp_path, + monkeypatch, + _stranded_confidential_stt_config(), + ) + monkeypatch.setattr(supervisor, "_parakeet_platform_can_host", lambda: True) + monkeypatch.setattr(supervisor, "read_available_bytes", lambda: 2 * 1024**3) + monkeypatch.setattr(supervisor, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(supervisor, "local_stt_backend", lambda: "parakeet") + + observation = supervisor._observe_parakeet_provider_truth() + + assert observation.phase == "host-blocked" + assert observation.reason_code == "host-admission-blocked" + assert observation.detail["stt_admission_latch"]["blocked"] is True + + +def test_parakeet_truth_stranded_adequate_ram_desires_parakeet( + monkeypatch, tmp_path +) -> None: + _install_supervisor_config( + tmp_path, + monkeypatch, + _stranded_confidential_stt_config(), + ) + monkeypatch.setattr(supervisor, "_parakeet_platform_can_host", lambda: True) + monkeypatch.setattr(supervisor, "read_available_bytes", lambda: 8 * 1024**3) + monkeypatch.setattr(supervisor, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(supervisor, "local_stt_backend", lambda: "parakeet") + monkeypatch.setattr( + parakeet_install, + "target_fingerprint", + lambda *, journal_path=None: {"provider": "parakeet"}, + ) + monkeypatch.setattr( + parakeet_install, + "inspect_readiness", + lambda journal_path=None: _parakeet_readiness(), + ) + + observation = supervisor._observe_parakeet_provider_truth() + + assert observation.phase == "artifact-not-ready" + assert observation.reason_code == "artifact-missing" + + def test_parakeet_truth_reports_artifact_not_ready_when_missing( monkeypatch, tmp_path, diff --git a/tests/test_transcribe_cli.py b/tests/test_transcribe_cli.py index c2648f2da..27774444f 100644 --- a/tests/test_transcribe_cli.py +++ b/tests/test_transcribe_cli.py @@ -13,12 +13,56 @@ from unittest.mock import MagicMock, patch import pytest from solstone.observe.vad import VadResult +from tests.helpers.journal_config import seed_journal_config def _args(backend: str | None = None) -> argparse.Namespace: return argparse.Namespace(backend=backend, cpu=False, model=None, redo=False) +def _confidential_block() -> dict: + return { + "enabled_at": "2026-05-24T00:00:00Z", + "account_id": "acct-test", + "endpoint_url": "https://spp.example.test", + "served_model_id": "confidential-model", + "credential_fingerprint_sha256": "fingerprint", + } + + +def _stranded_confidential_config(*, transcribe: dict | None = None) -> dict: + config = { + "services": {"confidential": _confidential_block()}, + "providers": {"local": {}}, + } + if transcribe is not None: + config["transcribe"] = transcribe + return config + + +def _healthy_confidential_config(*, transcribe: dict | None = None) -> dict: + config = { + "services": {"confidential": _confidential_block()}, + "providers": { + "local": { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + "credential": "confidential-credential", + } + }, + } + if transcribe is not None: + config["transcribe"] = transcribe + return config + + +def _seed_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config: dict) -> dict: + journal = tmp_path / "journal" + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + seed_journal_config(config, journal) + return config + + def test_main_accepts_journal_relative_path(tmp_path, monkeypatch): """main() resolves audio_path relative to journal when absolute path fails.""" seg_dir = tmp_path / "chronicle" / "20260201" / "default" / "090000_300" @@ -280,61 +324,220 @@ def test_main_google_key_decoy_below_floor_surfaces_local_requirement( assert "local transcription needs about 4 GB" in caplog.text -def test_resolve_default_backend_auto_selects_confidential_under_lane(monkeypatch): +def test_resolve_default_backend_stranded_low_ram_surfaces_requirement( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _stranded_confidential_config()) - monkeypatch.delenv("GOOGLE_API_KEY", raising=False) - monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 2 * 1024**3) monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"}, + + with pytest.raises(SystemExit) as exc_info: + transcribe_main.resolve_default_backend( + _args(), + config.get("transcribe", {}), + ) + + assert exc_info.value.code == 1 + + +def test_resolve_default_backend_stranded_adequate_ram_uses_local_backend( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + transcribe_dispatch = importlib.import_module("solstone.observe.transcribe") + config = _seed_config(tmp_path, monkeypatch, _stranded_confidential_config()) + + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 8 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + expected_backend = transcribe_main.local_stt_backend() + backend_module = MagicMock() + backend_module.transcribe.return_value = [{"text": "local"}] + get_backend = MagicMock(return_value=backend_module) + monkeypatch.setattr(transcribe_dispatch, "get_backend", get_backend) + + assert expected_backend is not None + resolved_backend = transcribe_main.resolve_default_backend( + _args(), config.get("transcribe", {}) + ) + assert resolved_backend == expected_backend + + assert transcribe_dispatch.transcribe(resolved_backend, [], 16000, {}) == [ + {"text": "local"} + ] + get_backend.assert_called_once_with(expected_backend) + backend_module.transcribe.assert_called_once_with([], 16000, {}) + + +def test_main_stranded_low_ram_preserves_audio_before_processing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from solstone.think.retention import resolve_segment_gate + + journal = tmp_path / "journal" + config = _stranded_confidential_config() + _seed_config(tmp_path, monkeypatch, config) + audio_file = ( + journal / "chronicle" / "20260722" / "_default" / "120000_0010" / "audio.wav" ) + audio_file.parent.mkdir(parents=True) + audio_file.write_bytes(b"not decoded on the surface path") + + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + monkeypatch.setattr("sys.argv", ["sol transcribe", str(audio_file)]) + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 2 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + process_one = MagicMock(return_value=None) + monkeypatch.setattr(transcribe_main, "_process_one", process_one) - assert transcribe_main.resolve_default_backend(_args(), {}) == "confidential" + with pytest.raises(SystemExit) as exc_info: + transcribe_main.main() + + assert exc_info.value.code == 1 + process_one.assert_not_called() + assert audio_file.exists() + assert not audio_file.with_suffix(".jsonl").exists() + assert resolve_segment_gate(audio_file.parent).verdict == "incomplete" + + +def test_resolve_default_backend_healthy_channel_without_brain_uses_confidential( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _healthy_confidential_config()) + + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + + assert ( + transcribe_main.resolve_default_backend( + _args(), + config.get("transcribe", {}), + journal_config=config, + ) + == "confidential" + ) -def test_resolve_default_backend_explicit_local_wins_under_lane(monkeypatch): +def test_resolve_default_backend_auto_selects_confidential_under_lane( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _healthy_confidential_config()) monkeypatch.delenv("GOOGLE_API_KEY", raising=False) monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"}, + + assert ( + transcribe_main.resolve_default_backend( + _args(), + config.get("transcribe", {}), + journal_config=config, + ) + == "confidential" ) + +def test_resolve_default_backend_explicit_local_wins_under_lane( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _healthy_confidential_config()) + + monkeypatch.delenv("GOOGLE_API_KEY", raising=False) + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + assert ( - transcribe_main.resolve_default_backend(_args(), {"backend": "parakeet"}) + transcribe_main.resolve_default_backend( + _args(), + {"backend": "parakeet"}, + journal_config=config, + ) == "parakeet" ) -def test_resolve_default_backend_confidential_fallback_never_cloud(monkeypatch, caplog): +def test_resolve_default_backend_confidential_fallback_never_cloud( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog +): transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _healthy_confidential_config()) monkeypatch.delenv("GOOGLE_API_KEY", raising=False) monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"}, - ) with caplog.at_level(logging.WARNING): backend = transcribe_main.resolve_default_backend( _args(), {"backend": "confidential", "confidential_audio": False}, + journal_config=config, ) assert backend == "parakeet" assert "confidential audio is disabled" in caplog.text +def test_resolve_default_backend_healthy_channel_audio_disabled_uses_local_backend( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config( + tmp_path, + monkeypatch, + _healthy_confidential_config(transcribe={"confidential_audio": False}), + ) + + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 1 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + expected_backend = transcribe_main.local_stt_backend() + + assert expected_backend is not None + assert ( + transcribe_main.resolve_default_backend( + _args(), + config.get("transcribe", {}), + journal_config=config, + ) + == expected_backend + ) + + +def test_resolve_default_backend_warns_when_confidential_channel_incomplete( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog +) -> None: + transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _stranded_confidential_config()) + + monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 8 * 1024**3) + monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) + monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") + + with caplog.at_level(logging.WARNING): + backend = transcribe_main.resolve_default_backend( + _args(), + {"backend": "confidential"}, + journal_config=config, + ) + + assert backend == "parakeet" + assert ( + "confidential channel is incomplete: missing credential, endpoint URL, " + "and served model ID" + ) in caplog.text + + def test_resolve_default_backend_surfaces_when_no_viable_backend(monkeypatch): transcribe_main = importlib.import_module("solstone.observe.transcribe.main") @@ -374,22 +577,21 @@ def test_resolve_default_backend_warns_but_honors_explicit_local(monkeypatch, ca def test_resolve_default_backend_stale_config_routes_to_confidential_under_lane( - monkeypatch, caplog + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog ): transcribe_main = importlib.import_module("solstone.observe.transcribe.main") + config = _seed_config(tmp_path, monkeypatch, _healthy_confidential_config()) monkeypatch.delenv("GOOGLE_API_KEY", raising=False) monkeypatch.setattr(transcribe_main, "read_available_bytes", lambda: 2 * 1024**3) monkeypatch.setattr(transcribe_main, "stt_local_floor_bytes", lambda: 4 * 1024**3) monkeypatch.setattr(transcribe_main, "local_stt_backend", lambda: "parakeet") - monkeypatch.setattr( - "solstone.think.services.spp.confidential_provenance", - lambda: {"enabled_at": "2026-05-24T00:00:00Z"}, - ) with caplog.at_level(logging.WARNING): backend = transcribe_main.resolve_default_backend( - _args(), {"backend": "removed-stt"} + _args(), + {"backend": "removed-stt"}, + journal_config=config, ) assert backend == "confidential" diff --git a/tests/test_transcribe_confidential.py b/tests/test_transcribe_confidential.py index 00673477a..809e9a705 100644 --- a/tests/test_transcribe_confidential.py +++ b/tests/test_transcribe_confidential.py @@ -14,8 +14,14 @@ import numpy as np import pytest import soundfile as sf +from solstone.observe import transcribe as transcribe_pkg from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED -from solstone.observe.transcribe import ConfidentialTranscribeDeferral, confidential +from solstone.observe.transcribe import ( + ConfidentialAudioEgressError, + ConfidentialTranscribeDeferral, + confidential, +) +from solstone.observe.transcribe import transcribe as dispatch_transcribe from solstone.observe.utils import SAMPLE_RATE from solstone.observe.vad import VadResult from solstone.think.models import AttestationFailedError, AttestationStaleError @@ -30,6 +36,26 @@ def _block() -> dict[str, str]: } +def _stranded_config() -> dict: + return { + "services": {"confidential": _block()}, + "providers": {"local": {}}, + } + + +def _healthy_config() -> dict: + return { + "services": {"confidential": _block()}, + "providers": { + "local": { + "endpoint_url": "https://spp.example.test/v1", + "served_model_id": "confidential-model", + "credential": "confidential-token", + } + }, + } + + def _local_endpoint() -> LocalEndpoint: return LocalEndpoint( base_url="https://configured-endpoint.example/v1", @@ -57,6 +83,40 @@ def _audio(seconds: float = 0.2) -> np.ndarray: return np.linspace(-0.5, 0.5, int(SAMPLE_RATE * seconds), dtype=np.float32) +@pytest.mark.parametrize("config", [_stranded_config(), _healthy_config()]) +def test_confidential_dispatch_gate_refuses_nonlocal_under_any_provenance_shape( + config: dict, + monkeypatch: pytest.MonkeyPatch, +) -> None: + block = config["services"]["confidential"] + monkeypatch.setattr(spp, "confidential_provenance", lambda: block) + monkeypatch.setitem( + transcribe_pkg.BACKEND_REGISTRY, + "remote-test", + "tests.unused_remote_backend", + ) + monkeypatch.setitem( + transcribe_pkg.BACKEND_METADATA, + "remote-test", + {"local": False}, + ) + backend = Mock() + backend.transcribe.return_value = [] + get_backend = Mock(return_value=backend) + monkeypatch.setattr(transcribe_pkg, "get_backend", get_backend) + + with pytest.raises(ConfidentialAudioEgressError): + dispatch_transcribe("missing-backend", _audio(), SAMPLE_RATE, {}) + get_backend.assert_not_called() + + with pytest.raises(ConfidentialAudioEgressError): + dispatch_transcribe("remote-test", _audio(), SAMPLE_RATE, {}) + get_backend.assert_not_called() + + assert dispatch_transcribe("parakeet", _audio(), SAMPLE_RATE, {}) == [] + get_backend.assert_called_once_with("parakeet") + + def test_confidential_transcribe_posts_canonical_wav_to_forwarder( monkeypatch: pytest.MonkeyPatch, ) -> None: