import importlib from pathlib import Path import pytest from figment import config as config_module from figment.config import FIGMENT_CANNED_MODEL_ID, FigmentConfig, NVIDIA_OMNI_API_MODEL_ID from figment.validators import validate_audio_ready, validate_navigator_output CONFIG_ENV_KEYS = ( "FIGMENT_MODE", "MODEL_STACK", "MODEL_BACKEND", "AUDIO_BACKEND", "ENABLE_AUDIO_INTAKE", "ALLOW_LOCAL_ASR", "ALLOW_SELF_HOSTED_OMNI", "ALLOW_STRETCH_STACK", "HF_MODEL_ID", "NVIDIA_MODEL_ID", "NVIDIA_BASE_URL", "NVIDIA_API_KEY", "LOCAL_MODEL_ID", "HF_ENDPOINT_URL", "OMNI_ENDPOINT_URL", "HF_TOKEN", "LLAMA_BASE_URL", "FIGMENT_TRACE_DIR", ) def _clear_config_env(monkeypatch: pytest.MonkeyPatch) -> None: for key in CONFIG_ENV_KEYS: monkeypatch.delenv(key, raising=False) def test_config_defaults_to_omni_with_canned_fallback_without_hosted_secret( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) config = FigmentConfig.from_env() assert config.model_stack == "omni_native" assert config.model_backend == "canned" assert config.active_model_id == FIGMENT_CANNED_MODEL_ID def test_config_uses_hosted_omni_api_model_id_when_secret_is_present( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) monkeypatch.setenv("NVIDIA_API_KEY", "test-nvidia-key") config = FigmentConfig.from_env() assert config.model_stack == "omni_native" assert config.model_backend == "hosted_omni" assert config.nvidia_model_id == NVIDIA_OMNI_API_MODEL_ID assert config.active_model_id == NVIDIA_OMNI_API_MODEL_ID def test_hf_token_without_endpoint_does_not_select_hosted_omni( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) monkeypatch.setenv("HF_TOKEN", "hf-test-token") config = FigmentConfig.from_env() assert config.model_backend == "canned" assert config.active_model_id == FIGMENT_CANNED_MODEL_ID def test_hf_endpoint_selects_hosted_omni_with_hf_token( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) monkeypatch.setenv("HF_TOKEN", "hf-test-token") monkeypatch.setenv("HF_ENDPOINT_URL", "https://hf.example.test/v1") config = FigmentConfig.from_env() assert config.model_backend == "hosted_omni" assert config.hf_endpoint_url == "https://hf.example.test/v1" def test_explicit_hosted_omni_nvidia_endpoint_requires_key() -> None: with pytest.raises(ValueError, match="NVIDIA_API_KEY"): FigmentConfig(model_backend="hosted_omni", nvidia_api_key="").validated() def test_demo_audio_examples_cover_committed_audio_assets() -> None: app = importlib.import_module("app") examples = app._demo_audio_examples() assert len(examples) == 3 assert {Path(example[0]).name for example in examples} == { "case_1_dictated_intake.wav", "case_2_dictated_intake.wav", "case_3_dictated_intake.wav", } assert all(Path(example[0]).exists() for example in examples) assert all(example[1] == "" for example in examples) def test_config_gates_local_4b_parakeet_stack_with_local_asr( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: local_4b_model_id = getattr(config_module, "NVIDIA_NEMOTRON_3_NANO_4B_BF16_MODEL_ID", None) parakeet_model_id = getattr(config_module, "PARAKEET_ASR_MODEL_ID", None) assert local_4b_model_id == "nvidia/NVIDIA-Nemotron-3-Nano-4B-BF16" assert parakeet_model_id == "nvidia/parakeet-rnnt-1.1b" config = FigmentConfig(model_stack="local_4b_parakeet", model_backend="llama_cpp").validated() assert config.local_model_id == local_4b_model_id assert config.active_model_id == local_4b_model_id with pytest.raises(ValueError, match="parakeet_nemo requires ALLOW_LOCAL_ASR"): FigmentConfig(audio_backend="parakeet_nemo").validated() with pytest.raises(ValueError, match="MODEL_STACK=local_4b_parakeet"): FigmentConfig( model_stack="omni_native", audio_backend="parakeet_nemo", allow_local_asr=True, ).validated() monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) monkeypatch.setenv("FIGMENT_MODE", "local") monkeypatch.setenv("MODEL_BACKEND", "llama_cpp") monkeypatch.setenv("MODEL_STACK", "local_4b_parakeet") monkeypatch.setenv("AUDIO_BACKEND", "parakeet_nemo") monkeypatch.setenv("ALLOW_LOCAL_ASR", "true") local_config = FigmentConfig.from_env() assert local_config.allow_local_asr is True assert local_config.audio_model_id == parakeet_model_id assert local_config.active_model_id == local_4b_model_id def test_legacy_split_stack_contract_is_rejected(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None: with pytest.raises(ValueError, match="base_nano_parakeet.*local_4b_parakeet"): FigmentConfig(model_stack="base_nano_parakeet").validated() monkeypatch.chdir(tmp_path) _clear_config_env(monkeypatch) monkeypatch.setenv("ALLOW_STRETCH_STACK", "true") with pytest.raises(ValueError, match="ALLOW_STRETCH_STACK.*ALLOW_LOCAL_ASR"): FigmentConfig.from_env() def test_red_flag_rules_run_only_on_confirmed_intake() -> None: app = importlib.import_module("app") unconfirmed = app.collect_intake( setting="shelter clinic", patient_age="52", pregnancy_status="not_applicable", chief_concern="Chest pain and shortness of breath", symptoms="crushing chest pain", vitals="HR 118", allergies="unknown", medications="unknown", available_supplies="oxygen, AED", responder_note="Adult reports chest pain after cleanup work.", ) assert app.evaluate_red_flags(unconfirmed) == [] confirmed = app.confirm_intake(unconfirmed) rules = app.evaluate_red_flags(confirmed) assert {rule["rule_id"] for rule in rules} >= {"red_flag_chest_pain"} assert app.urgency_floor_from_rules(rules) == "emergency" def test_validators_reject_downgraded_or_unsafe_navigator_output() -> None: output = { "protocol_urgency": "routine", "candidate_protocol_pathways": [ {"card_id": "CHEST-PAIN-v1", "reason_relevant": "Chest pain was reported."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Diagnose the cause and give 5 mg medication."], "do_not_do": [], "source_cards": ["CHEST-PAIN-v1"], "handoff_note_sbar": { "situation": "Chest pain", "background": "Synthetic case", "assessment_observations_only": "Pain reported", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "", } result = validate_navigator_output(output, {"CHEST-PAIN-v1"}, urgency_floor="emergency") assert not result.passed assert any("below deterministic floor" in item for item in result.failures) assert any("forbidden clinical language" in item for item in result.failures) def test_validator_allows_negated_safety_boundary_language() -> None: output = { "protocol_urgency": "emergency", "candidate_protocol_pathways": [ {"card_id": "CHEST-PAIN-v1", "reason_relevant": "Chest pain was reported."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": ["Do not diagnose the cause."], "source_cards": ["CHEST-PAIN-v1"], "handoff_note_sbar": { "situation": "Chest pain", "background": "Synthetic case", "assessment_observations_only": "Pain reported", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "This is not a diagnosis or prescription and cannot diagnose, prescribe, dose, or override local protocol.", } result = validate_navigator_output(output, {"CHEST-PAIN-v1"}, urgency_floor="emergency") assert result.passed def test_validator_allows_does_not_diagnose_or_prescribe_boundary_language() -> None: output = { "protocol_urgency": "routine", "candidate_protocol_pathways": [ {"card_id": "SAFETY-BOUNDARIES-v1", "reason_relevant": "Safety boundary review."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": [], "source_cards": ["SAFETY-BOUNDARIES-v1"], "handoff_note_sbar": { "situation": "Mild symptoms", "background": "Synthetic case", "assessment_observations_only": "Temperature normal.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "This output does not diagnose or prescribe and does not replace local protocol.", } result = validate_navigator_output(output, {"SAFETY-BOUNDARIES-v1"}) assert result.passed def test_validator_allows_negated_do_not_start_or_prescribe_list() -> None: output = { "protocol_urgency": "urgent", "candidate_protocol_pathways": [ {"card_id": "WOUND-INFECTION-ESCALATION-v1", "reason_relevant": "Wound cue."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": [ "Do not start antibiotics, prescribe medication, or provide dosing instructions unless local protocol explicitly provides them." ], "source_cards": ["WOUND-INFECTION-ESCALATION-v1"], "handoff_note_sbar": { "situation": "Wound concern", "background": "Synthetic case", "assessment_observations_only": "Spreading redness reported.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } result = validate_navigator_output(output, {"WOUND-INFECTION-ESCALATION-v1"}, urgency_floor="urgent") assert result.passed def test_validator_allows_negated_diagnosis_safety_instruction_with_condition_context() -> None: output = { "protocol_urgency": "emergency", "candidate_protocol_pathways": [ {"card_id": "PED-DEHYD-RED-FLAGS-v1", "reason_relevant": "Pediatric danger signs were reported."} ], "missing_info_to_collect": ["repeat vitals"], "next_observations_to_collect": ["urine output"], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": ["Do not diagnose dehydration severity as a final clinical diagnosis."], "source_cards": ["PED-DEHYD-RED-FLAGS-v1"], "handoff_note_sbar": { "situation": "Pediatric dehydration concern", "background": "Synthetic case", "assessment_observations_only": "No urine since morning.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } result = validate_navigator_output(output, {"PED-DEHYD-RED-FLAGS-v1"}, urgency_floor="emergency") assert result.passed def test_validator_rejects_prescribing_after_unrelated_negated_instruction() -> None: output = { "protocol_urgency": "routine", "candidate_protocol_pathways": [ {"card_id": "SAFETY-BOUNDARIES-v1", "reason_relevant": "Safety boundary review."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Do not wait, prescribe opioid now."], "do_not_do": [], "source_cards": ["SAFETY-BOUNDARIES-v1"], "handoff_note_sbar": { "situation": "Mild symptoms", "background": "Synthetic case", "assessment_observations_only": "Temperature normal.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } result = validate_navigator_output(output, {"SAFETY-BOUNDARIES-v1"}) assert not result.passed assert any("forbidden clinical language: prescribe" in item for item in result.failures) def test_validator_rejects_incomplete_handoff_schema() -> None: output = { "protocol_urgency": "urgent", "candidate_protocol_pathways": [ {"card_id": "CHEST-PAIN-v1", "reason_relevant": "Chest pain was reported."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": ["Do not diagnose the cause."], "source_cards": ["CHEST-PAIN-v1"], "handoff_note_sbar": { "situation": "Chest pain", "background": "Synthetic case", "assessment_observations_only": "Pain reported", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } result = validate_navigator_output(output, {"CHEST-PAIN-v1"}, urgency_floor="urgent") assert not result.passed assert any("handoff_note_sbar is missing" in item for item in result.failures) def test_validator_rejects_forbidden_language_after_unrelated_not() -> None: output = { "protocol_urgency": "routine", "candidate_protocol_pathways": [ {"card_id": "SAFETY-BOUNDARIES-v1", "reason_relevant": "Safety boundary review."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": [], "source_cards": ["SAFETY-BOUNDARIES-v1"], "handoff_note_sbar": { "situation": "Mild symptoms", "background": "Synthetic case", "assessment_observations_only": "Patient not stable, prescribe opioid now.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } result = validate_navigator_output(output, {"SAFETY-BOUNDARIES-v1"}) assert not result.passed assert any("forbidden clinical language: prescribe" in item for item in result.failures) def test_validator_rejects_ungrounded_sbar_when_intake_context_is_available() -> None: output = { "protocol_urgency": "routine", "candidate_protocol_pathways": [ {"card_id": "SAFETY-BOUNDARIES-v1", "reason_relevant": "Safety boundary review."} ], "missing_info_to_collect": [], "next_observations_to_collect": [], "conflicts_or_uncertainties": [], "responder_checklist": ["Escalate per cited local protocol."], "do_not_do": ["Do not diagnose."], "source_cards": ["SAFETY-BOUNDARIES-v1"], "handoff_note_sbar": { "situation": "Head injury with loss of consciousness", "background": "Fall from ladder at home", "assessment_observations_only": "Blood pressure 220/140 and skull fracture observed.", "handoff_request": "Escalate per protocol", }, "responder_plain_language_script": "", "safety_boundary": "Prototype protocol navigation only.", } intake = { "setting": "shelter clinic", "patient_age": "40", "pregnancy_status": "not_applicable", "chief_concern": "mild cough", "symptoms": "speaking normally, no distress cues", "vitals": "temperature normal", "allergies": "unknown", "medications": "unknown", "available_supplies": "radio", "responder_note": "No injury reported.", "confirmed": True, } result = validate_navigator_output( output, {"SAFETY-BOUNDARIES-v1"}, confirmed_intake=intake, rule_results=[], ) assert not result.passed assert any("handoff_note_sbar" in item and "not grounded" in item for item in result.failures) def test_audio_drafts_are_non_authoritative_until_confirmed() -> None: app = importlib.import_module("app") disabled_config = FigmentConfig(enable_audio_intake=False, audio_backend="none").validated() disabled_draft = app.draft_audio_intake( transcript="Patient says chest pain, but audio is disabled.", config=disabled_config, ) assert disabled_draft["audio_runtime"] == "none" assert disabled_draft["confirmed_intake_required"] is False assert disabled_draft["provisional_red_flag_mentions"] == [] assert validate_audio_ready(disabled_draft).passed enabled_config = FigmentConfig( model_backend="canned", enable_audio_intake=True, audio_backend="omni_native", ).validated() audio_draft = app.draft_audio_intake( transcript="Adult with chest pain and trouble breathing.", config=enabled_config, ) assert audio_draft["confirmation_status"] == "unconfirmed" assert "chest pain" in " ".join(audio_draft["provisional_red_flag_mentions"]).lower() assert not validate_audio_ready(audio_draft).passed intake = app.collect_intake( setting="mobile clinic", patient_age="adult", pregnancy_status="not_applicable", chief_concern="", symptoms="", vitals="", allergies="", medications="", available_supplies="basic kit", responder_note="", ) with pytest.raises(ValueError, match="audio-derived intake must be confirmed"): app.confirm_intake(intake, audio_draft=audio_draft) updated_intake, confirmed_audio = app.confirm_audio_draft( intake, audio_draft, accept=True, ) confirmed_intake = app.confirm_intake(updated_intake, audio_draft=confirmed_audio) assert confirmed_audio["confirmation_status"] == "confirmed" assert confirmed_intake["confirmed"] is True assert app.evaluate_red_flags(confirmed_intake) def test_run_case_uses_environment_config_when_not_supplied(monkeypatch: pytest.MonkeyPatch) -> None: app = importlib.import_module("app") _clear_config_env(monkeypatch) monkeypatch.setenv("PYTHON_DOTENV_DISABLED", "true") monkeypatch.setenv("MODEL_BACKEND", "hosted_omni") monkeypatch.setenv("NVIDIA_API_KEY", "test-nvidia-key") seen = {} class FakeTrace: validator_result = {"passed": True, "failures": []} def to_dict(self) -> dict[str, object]: return {"model_route": {"model_backend": seen["config"].model_backend}} def fake_run_navigation(intake, rules, *, audio_draft=None, config=None, **_kwargs): seen["config"] = config return {"protocol_urgency": "routine", "source_cards": []}, FakeTrace() monkeypatch.setattr(app, "run_navigation", fake_run_navigation) monkeypatch.setattr(app, "render_sbar", lambda *_args, **_kwargs: "handoff") intake = app.collect_intake( setting="mobile clinic", patient_age="52", pregnancy_status="not_applicable", chief_concern="wound concern", symptoms="spreading redness", vitals="unknown", allergies="unknown", medications="unknown", available_supplies="dressings", responder_note="Cut from debris.", ) result = app.run_case(intake) assert seen["config"].model_backend == "hosted_omni" assert result["trace"]["model_route"]["model_backend"] == "hosted_omni" def test_app_import_and_server_smoke() -> None: app = importlib.import_module("app") config = FigmentConfig(model_backend="canned", audio_backend="none").validated() demo = app.build_app(config=config) assert demo.__class__.__name__ == "Server" assert hasattr(demo, "api") assert hasattr(demo, "launch") assert {route.path for route in demo.routes} >= {"/", "/health"} assert app.TAB_TITLES == [ "Intake", "Risk Check", "Protocol Guidance", "Navigator Output + Handoff", "Trace", ]