From 0505f56f76dfdbef0d68bf53983b4ca821e2904f Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 22 Jul 2026 16:09:05 -0600 Subject: [PATCH] fix(transcribe): route STT by confidential usability - Bug: a stranded journal with services.confidential present and providers.local emptied of its credential made STT callers treat the confidential lane as active from bare block presence. They selected the confidential backend, then deferred forever in confidential.py::_headers; transcription silently stopped while the journal looked healthy. - Add spp.is_confidential_channel_usable(config): is_confidential_enabled(config) and not resolve_local_endpoint_from_config(config).is_bundled. It is config-pure: no network and no attestation probe. - Do not adopt is_confidential_enabled alone. It returns True for a credential-only config with no endpoint_url or served_model_id; that shape fails the forwarder probe with endpoint_invalid and would strand STT identically. - Keep the asymmetry explicit: routing follows channel usability. The egress refusal at observe/transcribe/__init__.py:204 stays on bare block presence because it answers whether raw audio may go remote, and it remains fail-closed and broad. - Changed observe/transcribe/main.py::resolve_default_backend; think/supervisor.py::linux_stt_uses_parakeet_cpp; and think/supervisor.py::_observe_parakeet_provider_truth. - In _parakeet_stt_admission_latch, all four uses now carry usability: payload/fingerprint key value, RAM-read short-circuit, selector call, and ram_blocked. - Changed apps/settings/routes.py to pass usability into apps/settings/transcribe_resource.py, the direct selector caller the original scope map missed. - Add three-way warning copy in main.py so explicit confidential fallback distinguishes disabled confidential audio, no confidential lane, and an incomplete channel that names missing fields. - Widen fixtures without weakening assertions. test_no_implicit_cloud.py uses _add_local_endpoint; both assertions are verbatim. - settings test_transcribe_resource_payload seeds endpoint_url, served_model_id, and credential; assertions are unchanged. - test_supervisor_parakeet.py and test_transcribe_cli.py replace monkeypatch-provenance-alone usable-channel setups with complete usable provider config where that was the intent. - Deliberate AC5 behavior change: stranded low-RAM now exits via SystemExit(1) from resolve_default_backend instead of EXIT_PROVIDER_BLOCKED (69) from deferral. - Data-safety finding: input audio remains on disk, no JSONL is written, _process_one is never reached, resolve_segment_gate reports incomplete, and sense's non-69 path does not delete, move, quarantine, or mark-failed the audio. - No config migration: confidential_lane_active is part of _parakeet_stt_admission_input, so old latch records hash to a different input_sha256 and are recomputed on the first tick. - Red-first evidence: AC1 and AC5 failed with "DID NOT RAISE SystemExit"; AC2 failed selecting confidential instead of parakeet; AC7 failed with not-desired instead of host-blocked and artifact-not-ready. AC6 was green on main by construction. - Reported, not fixed: linux_stt_uses_parakeet_cpp is dead in production; apps/settings/routes.py:209 backend coercion makes the settings needs_setup/notice path unreachable; routes.py:590 uses bool(...) instead of confidential_audio_enabled's true-when-unset default; the confidential_lane_active selector keyword and latch payload key remain a deferred rename. - Validated before commit: focused STT/settings/retention suites passed, then make ci passed with 14738 passed and 16 skipped. Co-Authored-By: Claude Opus 4.8 (1M context) --- solstone/apps/settings/routes.py | 6 +- .../tests/test_transcribe_resource_payload.py | 5 + solstone/apps/settings/transcribe_resource.py | 1 + solstone/observe/transcribe/__init__.py | 2 + solstone/observe/transcribe/main.py | 69 ++++- solstone/think/services/spp.py | 9 + solstone/think/supervisor.py | 34 ++- tests/services/test_spp_storage.py | 60 +++++ tests/test_no_implicit_cloud.py | 1 + tests/test_supervisor_parakeet.py | 133 +++++++++- tests/test_transcribe_cli.py | 248 ++++++++++++++++-- tests/test_transcribe_confidential.py | 62 ++++- 12 files changed, 573 insertions(+), 57 deletions(-) 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: -- 2.51.2