import importlib import json import wave import pytest from figment.config import FigmentConfig, OMNI_MODEL_ID def test_hosted_omni_provider_payload_becomes_unconfirmed_audio_draft() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", nvidia_api_key="test-nvidia-key", enable_audio_intake=True, audio_backend="omni_native", ).validated() provider_payload = { "transcript": "Adult reports severe headache and visual spots after flood cleanup.", "suggested_fields": [ { "field": "symptoms", "draft_value": "severe headache and visual spots", "source_snippet": "severe headache and visual spots", "source_timecode": "00:01-00:05", } ], "missing_or_unclear_fields": ["blood_pressure", "pregnancy_status"], "provisional_red_flag_mentions": ["severe headache", "visual spots"], } draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) assert draft["audio_intake_path"] == "omni_native" assert draft["audio_runtime"] == "omni_native" assert draft["audio_model_id"] == OMNI_MODEL_ID assert draft["field_fill_model_id"] is None assert draft["transcript_source"] == "hosted_omni_provider" assert draft["transcript"] == provider_payload["transcript"] assert draft["missing_or_unclear_fields"] == ["blood_pressure", "pregnancy_status"] assert draft["provisional_red_flag_mentions"] == ["severe headache", "visual spots"] assert draft["confirmed_intake_required"] is True assert draft["confirmation_status"] == "unconfirmed" assert draft["raw_audio_stored"] is False assert draft["suggested_fields"] == [ { "field": "symptoms", "draft_value": "severe headache and visual spots", "source_snippet": "severe headache and visual spots", "source_timecode": "00:01-00:05", "status": "audio_draft", "needs_confirmation": True, } ] def test_hosted_omni_provider_payload_with_invalid_field_names_falls_back_to_transcript_suggestions() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", nvidia_api_key="test-nvidia-key", enable_audio_intake=True, audio_backend="omni_native", ).validated() provider_payload = { "transcript": ( "Seven-year-old at a shelter clinic after flood cleanup. Child cannot keep fluids down, " "is lethargic, has a very dry mouth, and has no urine since morning." ), "suggested_fields": [ { "field": "chief_concern|symptoms|vitals|patient_age|pregnancy_status|allergies|medications|available_supplies|responder_note", "draft_value": "lethargic, dry mouth, no urine", "source_snippet": "lethargic, dry mouth, no urine", } ], "provisional_red_flag_mentions": ["lethargic", "no urine"], } draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) fields = {suggestion["field"] for suggestion in draft["suggested_fields"]} assert "chief_concern|symptoms|vitals|patient_age|pregnancy_status|allergies|medications|available_supplies|responder_note" not in fields assert {"responder_note", "patient_age", "chief_concern", "symptoms"}.issubset(fields) assert next(item for item in draft["suggested_fields"] if item["field"] == "patient_age")["draft_value"] == "7 years" def test_hosted_omni_provider_payload_without_transcript_or_valid_suggestions_fails_closed() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", nvidia_api_key="test-nvidia-key", enable_audio_intake=True, audio_backend="omni_native", ).validated() provider_payload = { "suggested_fields": [ { "field": "chief_concern|symptoms", "draft_value": "chest pain", "source_snippet": "chest pain", } ] } draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" assert draft["audio_runtime"] == "unprocessed_audio" assert draft["transcript"] == "" assert draft["suggested_fields"] == [] assert draft["confirmed_intake_required"] is False assert draft["confirmation_status"] == "confirmed" assert draft["missing_or_unclear_fields"] == ["transcript_or_valid_provider_suggestions"] assert "valid field suggestions" in draft["processing_status"].lower() def test_hosted_omni_provider_payload_drops_non_informative_field_values_and_preserves_red_flag_mentions() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", nvidia_api_key="test-nvidia-key", enable_audio_intake=True, audio_backend="omni_native", ).validated() provider_payload = { "transcript": ( "Seven-year-old at a shelter clinic after flood cleanup. Child cannot keep fluids down, " "is lethargic, has a very dry mouth, and has no urine since morning." ), "suggested_fields": [ { "field": "symptoms", "draft_value": "cannot keep fluids down, lethargic, very dry mouth, no urine since morning", "source_snippet": "cannot keep fluids down, lethargic", }, { "field": "allergies", "draft_value": "Not answerable", "source_snippet": "", }, { "field": "medications", "draft_value": "not provided", "source_snippet": "", }, { "field": "responder_note", "draft_value": "Provisional intake draft for trained responder", "source_snippet": "", }, ], "missing_or_unclear_fields": [], "provisional_red_flag_mentions": [], } draft = app.draft_audio_intake(config=config, provider_payload=provider_payload) assert {item["field"] for item in draft["suggested_fields"]} == {"symptoms"} assert set(draft["missing_or_unclear_fields"]) >= {"allergies", "medications"} assert set(draft["provisional_red_flag_mentions"]) >= {"lethargic", "no urine"} def test_manual_edit_wins_over_hosted_omni_audio_draft_on_confirmation() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", nvidia_api_key="test-nvidia-key", enable_audio_intake=True, audio_backend="omni_native", ).validated() audio_draft = app.draft_audio_intake( config=config, provider_payload={ "transcript": "Audio says chest pain, but the medic corrects it.", "suggested_fields": [ { "field": "chief_concern", "draft_value": "chest pain", "source_snippet": "chest pain", } ], }, ) confirmed, _, confirmed_audio = app._confirm_ui_intake( "mobile clinic", "52", "not_applicable", "manual correction: wound concern", "", "", "unknown", "unknown", "basic kit", "Typed note remains source of truth.", audio_draft, ) assert confirmed["chief_concern"] == "manual correction: wound concern" assert confirmed_audio["suggested_fields"][0]["draft_value"] == "manual correction: wound concern" assert confirmed_audio["suggested_fields"][0]["status"] == "edited" assert confirmed_audio["suggested_fields"][0]["needs_confirmation"] is False def test_uploaded_audio_uses_hosted_omni_provider_when_enabled(tmp_path, monkeypatch) -> None: app = importlib.import_module("app") audio_path = tmp_path / "field-note.wav" audio_path.write_bytes(b"fake wav bytes") config = FigmentConfig( model_backend="hosted_omni", enable_audio_intake=True, audio_backend="omni_native", nvidia_api_key="test-nvidia-key", ).validated() seen = {} class FakeModelClient: def __init__(self, config_arg): seen["config"] = config_arg def generate_audio_draft(self, audio_file): seen["audio_file"] = audio_file return { "transcript": "Adult reports trouble breathing.", "suggested_fields": [ { "field": "symptoms", "draft_value": "trouble breathing", "source_snippet": "trouble breathing", } ], "missing_or_unclear_fields": ["vitals"], "provisional_red_flag_mentions": ["trouble breathing"], } monkeypatch.setattr(app, "ModelClient", FakeModelClient) draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) assert seen["config"] is config assert seen["audio_file"] == str(audio_path) assert draft["audio_intake_path"] == "omni_native" assert draft["audio_runtime"] == "omni_native" assert draft["audio_model_id"] == OMNI_MODEL_ID assert draft["transcript"] == "Adult reports trouble breathing." assert draft["suggested_fields"][0]["field"] == "symptoms" assert draft["confirmation_status"] == "unconfirmed" assert draft["audio_filename"] == "field-note.wav" assert draft["raw_audio_stored"] is False assert "not written to figment traces" in draft["audio_retention_note"].lower() def test_hosted_omni_audio_failure_fails_closed(tmp_path, monkeypatch) -> None: app = importlib.import_module("app") audio_path = tmp_path / "field-note.wav" audio_path.write_bytes(b"fake wav bytes") config = FigmentConfig( model_backend="hosted_omni", enable_audio_intake=True, audio_backend="omni_native", nvidia_api_key="test-nvidia-key", ).validated() class FailingModelClient: def __init__(self, *_args, **_kwargs): pass def generate_audio_draft(self, _audio_file): raise app.ModelClientError("endpoint unavailable") monkeypatch.setattr(app, "ModelClient", FailingModelClient) draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" assert draft["audio_runtime"] == "unprocessed_audio" assert draft["suggested_fields"] == [] assert "hosted omni audio draft failed" in draft["processing_status"].lower() assert draft["audio_filename"] == "field-note.wav" assert draft["raw_audio_stored"] is False assert "not written to figment traces" in draft["audio_retention_note"].lower() def test_local_parakeet_provider_payload_uses_local_4b_runtime_label_when_gated() -> None: app = importlib.import_module("app") config = FigmentConfig( model_stack="local_4b_parakeet", model_backend="llama_cpp", enable_audio_intake=True, audio_backend="parakeet_nemo", allow_local_asr=True, ).validated() draft = app.draft_audio_intake( provider_payload={ "transcript": "Local Parakeet transcript says the patient has trouble breathing.", "suggested_fields": [ { "field": "symptoms", "draft_value": "trouble breathing", "source_snippet": "trouble breathing", } ], }, config=config, ) assert draft["audio_intake_path"] == "parakeet_rnnt_plus_text_nemotron" assert draft["audio_runtime"] == "local_4b_parakeet" assert draft["audio_model_id"] == "nvidia/parakeet-rnnt-1.1b" assert draft["field_fill_model_id"] == "nvidia/NVIDIA-Nemotron-3-Nano-4B-BF16" assert draft["transcript_source"] == "local_parakeet_asr_provider" assert draft["confirmation_status"] == "unconfirmed" def test_typed_transcript_under_parakeet_config_stays_heuristic_not_asr() -> None: app = importlib.import_module("app") config = FigmentConfig( model_stack="local_4b_parakeet", model_backend="llama_cpp", enable_audio_intake=True, audio_backend="parakeet_nemo", allow_local_asr=True, ).validated() draft = app.draft_audio_intake( transcript="Typed note says the patient has trouble breathing.", config=config, ) assert draft["audio_intake_path"] == "typed_transcript_heuristic" assert draft["audio_runtime"] == "typed_transcript_heuristic" assert draft["audio_model_id"] is None assert draft["field_fill_model_id"] is None assert draft["draft_source"] == "typed_transcript_heuristic" assert draft["transcript_source"] == "typed_transcript" def test_confirm_intake_does_not_silently_bulk_accept_audio_suggestions() -> None: app = importlib.import_module("app") audio_draft = { "audio_intake_path": "omni_native", "audio_runtime": "omni_native", "transcript": "Adult with chest pain.", "suggested_fields": [ { "field": "chief_concern", "draft_value": "chest pain", "source_snippet": "chest pain", "status": "audio_draft", "needs_confirmation": True, } ], "confirmed_intake_required": True, "confirmation_status": "unconfirmed", "raw_audio_stored": False, } confirmed, _, confirmed_audio = app._confirm_ui_intake( "mobile clinic", "52", "not_applicable", "", "", "", "unknown", "unknown", "basic kit", "Responder typed note remains source of truth.", audio_draft, ) assert confirmed["chief_concern"] == "" assert confirmed["responder_note"] == "Responder typed note remains source of truth." assert confirmed_audio["suggested_fields"][0]["status"] == "rejected" assert confirmed_audio["suggested_fields"][0]["needs_confirmation"] is False def test_rejected_audio_transcript_is_not_passed_to_navigator_prompt(monkeypatch) -> None: app = importlib.import_module("app") import figment.navigator as navigator audio_draft = { "audio_intake_path": "omni_native", "audio_runtime": "omni_native", "transcript": "Adult says chest pain from audio only.", "suggested_fields": [ { "field": "chief_concern", "draft_value": "chest pain", "source_snippet": "chest pain", "status": "audio_draft", "needs_confirmation": True, } ], "confirmed_intake_required": True, "confirmation_status": "unconfirmed", "raw_audio_stored": False, } confirmed, _, confirmed_audio = app._confirm_ui_intake( "mobile clinic", "52", "not_applicable", "", "", "", "unknown", "unknown", "basic kit", "Typed note remains the source of truth.", audio_draft, ) captured: dict[str, str] = {} class CapturePromptModelClient: def __init__(self, *_args, **_kwargs) -> None: pass def generate_json(self, prompt, context): captured["prompt"] = prompt return navigator.canned_navigator_output( context["intake"], context["rule_results"], context["retrieved_cards"], context["urgency_floor"], ) monkeypatch.setattr(navigator, "ModelClient", CapturePromptModelClient) navigator.run_navigation( confirmed, [], audio_draft=confirmed_audio, config=FigmentConfig(model_backend="hosted_omni", nvidia_api_key="unused"), retrieved_cards=[ { "card_id": "SAFETY-BOUNDARIES-v1", "card": {"card_id": "SAFETY-BOUNDARIES-v1", "title": "Safety boundaries"}, } ], ) assert "Adult says chest pain from audio only." not in captured["prompt"] assert "chest pain" not in captured["prompt"].lower() def test_apply_audio_draft_prefills_empty_fields_without_confirming() -> None: app = importlib.import_module("app") audio_draft = { "audio_intake_path": "omni_native", "audio_runtime": "omni_native", "transcript": "Adult with chest pain.", "suggested_fields": [ { "field": "chief_concern", "draft_value": "chest pain", "source_snippet": "chest pain", "status": "audio_draft", "needs_confirmation": True, } ], "confirmed_intake_required": True, "confirmation_status": "unconfirmed", "raw_audio_stored": False, } *fields, audio_json, audio_state = app._apply_audio_draft_ui( "mobile clinic", "52", "not_applicable", "", "", "", "unknown", "unknown", "basic kit", "", audio_draft, ) assert fields[3] == "chest pain" assert audio_json["confirmation_status"] == "unconfirmed" assert audio_json["suggested_fields"][0]["status"] == "applied_unreviewed" assert audio_json["suggested_fields"][0]["needs_confirmation"] is True assert audio_state is audio_json intake = app.collect_intake( fields[0], fields[1], fields[2], fields[3], fields[4], fields[5], fields[6], fields[7], fields[8], fields[9], ) with pytest.raises(ValueError, match="audio-derived intake must be confirmed"): app.confirm_intake(intake, audio_draft=audio_state) def test_loading_demo_case_resets_audio_and_downstream_state() -> None: app = importlib.import_module("app") values = app._load_demo_case_and_reset("Rural clinic: pregnancy danger sign") assert values[:10] == app._load_demo_case("Rural clinic: pregnancy danger sign") assert values[10] is None assert values[11] == "" assert values[12] is None assert values[13] is None assert values[14] == {"red_flags": [], "protocol_urgency": "routine"} assert "PROTOCOL_URGENCY" in values[15] assert values[16] == [] assert values[17] == "" assert "Selected Protocol Card" in values[18] assert values[19] == {} assert values[20] == "" assert "Protocol Urgency" in values[21] assert values[22] == {} assert values[23] is None assert "Run the navigator" in values[24] assert values[25] == {} assert values[26] is None assert values[27] == {} def test_source_and_audio_changes_clear_stale_pipeline_outputs() -> None: app = importlib.import_module("app") assert app._clear_source_outputs() == [ None, {"red_flags": [], "protocol_urgency": "routine"}, app._risk_summary_html({"red_flags": [], "protocol_urgency": "routine"}), [], "", app._protocol_results_html([]), {}, "", app._navigator_summary_html({}), {}, None, app._trace_audit_html({}), {}, {}, ] assert app._clear_audio_outputs() == [ None, None, {"red_flags": [], "protocol_urgency": "routine"}, app._risk_summary_html({"red_flags": [], "protocol_urgency": "routine"}), [], "", app._protocol_results_html([]), {}, "", app._navigator_summary_html({}), {}, None, app._trace_audit_html({}), {}, None, {}, ] def test_uploaded_audio_without_transcript_or_payload_is_not_fabricated_as_omni() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="canned", enable_audio_intake=True, audio_backend="omni_native", ).validated() draft = app.draft_audio_intake( transcript="", config=config, audio_file="/tmp/field-note.wav", provider_payload={}, ) assert draft["audio_file_received"] is True assert draft["audio_filename"] == "field-note.wav" assert draft["transcript"] == "" assert draft["suggested_fields"] == [] assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" assert draft["audio_runtime"] == "unprocessed_audio" assert "needs transcript/model support" in draft["processing_status"].lower() def test_canned_audio_backend_is_labeled_as_canned_demo_runtime() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="canned", enable_audio_intake=True, audio_backend="canned", ).validated() draft = app.draft_audio_intake(transcript="", config=config) assert draft["audio_runtime"] == "canned" assert draft["audio_intake_path"] == "canned_audio_demo" assert draft["audio_model_id"] is None assert draft["transcript_source"] == "synthetic_demo_manifest" assert draft["transcript"] assert draft["suggested_fields"] def test_canned_demo_transcript_matches_committed_manifest_pediatric_age() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="canned", enable_audio_intake=True, audio_backend="canned", ).validated() manifest = json.loads((app.PROJECT_ROOT / "data" / "demo_audio" / "manifest.json").read_text(encoding="utf-8")) pediatric_case = next(item for item in manifest["cases"] if item["slug"] == "pediatric_dehydration") draft = app.draft_audio_intake(transcript="", config=config, provider_payload=None) assert draft["transcript"] == pediatric_case["source_script"] assert "Seven year old" in draft["transcript"] assert "Four-year-old" not in draft["transcript"] assert next(item for item in draft["suggested_fields"] if item["field"] == "patient_age")["draft_value"] == "7 years" def test_hosted_audio_ui_discloses_endpoint_and_deidentified_limit() -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", enable_audio_intake=True, audio_backend="omni_native", nvidia_api_key="test-nvidia-key", ).validated() subtitle = app._audio_section_subtitle(config).lower() chips = app._audio_runtime_chips_html(config).lower() assert "sent to the configured hosted endpoint" in subtitle assert "synthetic or de-identified" in subtitle assert "10 mb" in subtitle assert "60 seconds" in subtitle assert "hosted endpoint" in chips def test_hosted_audio_oversize_file_fails_closed_before_provider_call(tmp_path, monkeypatch) -> None: app = importlib.import_module("app") import figment.model_client as model_client audio_path = tmp_path / "too-large.wav" audio_path.write_bytes(b"0123456789") monkeypatch.setattr(model_client, "HOSTED_AUDIO_MAX_BYTES", 8) config = FigmentConfig( model_backend="hosted_omni", enable_audio_intake=True, audio_backend="omni_native", nvidia_api_key="test-nvidia-key", ).validated() class FailIfCalledModelClient: def __init__(self, *_args, **_kwargs): pass def generate_audio_draft(self, _audio_file): raise AssertionError("oversize audio should be rejected before provider call") monkeypatch.setattr(app, "ModelClient", FailIfCalledModelClient) draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" assert "exceeds hosted audio size limit" in draft["processing_status"].lower() assert draft["audio_filename"] == "too-large.wav" def test_hosted_audio_long_wav_fails_closed_before_provider_call(tmp_path, monkeypatch) -> None: app = importlib.import_module("app") import figment.model_client as model_client audio_path = tmp_path / "too-long.wav" with wave.open(str(audio_path), "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(8000) wav_file.writeframes(b"\0\0" * 8000) monkeypatch.setattr(model_client, "HOSTED_AUDIO_MAX_SECONDS", 0.01) config = FigmentConfig( model_backend="hosted_omni", enable_audio_intake=True, audio_backend="omni_native", nvidia_api_key="test-nvidia-key", ).validated() class FailIfCalledModelClient: def __init__(self, *_args, **_kwargs): pass def generate_audio_draft(self, _audio_file): raise AssertionError("long audio should be rejected before provider call") monkeypatch.setattr(app, "ModelClient", FailIfCalledModelClient) draft = app.draft_audio_intake(audio_file=str(audio_path), config=config) assert draft["audio_intake_path"] == "audio_received_needs_transcript_or_model" assert "exceeds hosted audio duration limit" in draft["processing_status"].lower() assert draft["audio_filename"] == "too-long.wav" def test_navigator_ui_uses_supplied_runtime_config(monkeypatch) -> None: app = importlib.import_module("app") config = FigmentConfig( model_backend="hosted_omni", audio_backend="omni_native", enable_audio_intake=True, omni_endpoint_url="http://omni.example.test/v1", ).validated() intake = { "confirmed": True, "setting": "mobile clinic", "chief_concern": "typed chest pain concern", "responder_note": "typed source of truth", } seen = {} def fake_run_case(intake_arg, config_arg=None, audio_draft=None): seen["config"] = config_arg return { "navigator_output": {"protocol_urgency": "routine"}, "sbar": "handoff", "trace": {"model_route": {"model_backend": config_arg.model_backend}}, } monkeypatch.setattr(app, "run_case", fake_run_case) output, sbar, trace, state = app._navigate_ui(intake, None, config=config) assert seen["config"] is config assert output["protocol_urgency"] == "routine" assert sbar == "handoff" assert trace == state