Add ASR interim results support in Assistant model and API

- Introduced `asr_interim_enabled` field in the Assistant model to control interim ASR results.
- Updated AssistantBase and AssistantUpdate schemas to include the new field.
- Modified the database schema to add the `asr_interim_enabled` column.
- Enhanced runtime metadata to reflect interim ASR settings.
- Updated API endpoints and tests to validate the new functionality.
- Adjusted documentation to include details about interim ASR results configuration.

engine/tests/test_asr_factory_modes.py

@@ -32,6 +32,7 @@ def test_create_asr_service_openai_compatible_returns_offline_provider():
     )
     assert isinstance(service, OpenAICompatibleASRService)
     assert service.mode == "offline"
+    assert service.enable_interim is False
 
 
 def test_create_asr_service_fallback_buffered_for_unsupported_provider():

engine/tests/test_backend_adapters.py

@@ -282,7 +282,7 @@ async def test_local_yaml_adapter_rejects_path_traversal_like_assistant_id(tmp_p
 
 
 @pytest.mark.asyncio
-async def test_local_yaml_translates_agent_schema_to_runtime_services(tmp_path):
+async def test_local_yaml_translates_agent_schema_with_asr_interim_flag(tmp_path):
     config_dir = tmp_path / "assistants"
     config_dir.mkdir(parents=True, exist_ok=True)
     (config_dir / "default.yaml").write_text(
@@ -305,6 +305,7 @@ async def test_local_yaml_translates_agent_schema_to_runtime_services(tmp_path):
                 "    model: asr-model",
                 "    api_key: sk-asr",
                 "    api_url: https://asr.example.com/v1/audio/transcriptions",
+                "    enable_interim: false",
                 "  duplex:",
                 "    system_prompt: You are test assistant",
             ]
@@ -321,4 +322,5 @@ async def test_local_yaml_translates_agent_schema_to_runtime_services(tmp_path):
     assert services.get("llm", {}).get("apiKey") == "sk-llm"
     assert services.get("tts", {}).get("apiKey") == "sk-tts"
     assert services.get("asr", {}).get("apiKey") == "sk-asr"
+    assert services.get("asr", {}).get("enableInterim") is False
     assert assistant.get("systemPrompt") == "You are test assistant"

engine/tests/test_duplex_asr_modes.py

@@ -145,10 +145,11 @@ async def test_start_asr_capture_uses_streaming_begin(monkeypatch):
 
 
 @pytest.mark.asyncio
-async def test_start_asr_capture_uses_offline_interim_control(monkeypatch):
+async def test_start_asr_capture_uses_offline_interim_control_when_enabled(monkeypatch):
     asr = _FakeOfflineASR()
     pipeline = _build_pipeline(monkeypatch, asr)
     pipeline._asr_mode = "offline"
+    pipeline._runtime_asr["enableInterim"] = True
     pipeline._pending_speech_audio = b"\x00" * 320
     pipeline._pre_speech_buffer = b"\x00" * 640
 
@@ -159,6 +160,69 @@ async def test_start_asr_capture_uses_offline_interim_control(monkeypatch):
     assert pipeline._asr_capture_active is True
 
 
+@pytest.mark.asyncio
+async def test_start_asr_capture_skips_offline_interim_control_when_disabled(monkeypatch):
+    asr = _FakeOfflineASR()
+    pipeline = _build_pipeline(monkeypatch, asr)
+    pipeline._asr_mode = "offline"
+    pipeline._runtime_asr["enableInterim"] = False
+    pipeline._pending_speech_audio = b"\x00" * 320
+    pipeline._pre_speech_buffer = b"\x00" * 640
+
+    await pipeline._start_asr_capture()
+
+    assert asr.start_interim_calls == 0
+    assert asr.sent_audio
+    assert pipeline._asr_capture_active is True
+
+
+@pytest.mark.asyncio
+async def test_offline_interim_callback_ignored_when_disabled(monkeypatch):
+    asr = _FakeOfflineASR()
+    pipeline = _build_pipeline(monkeypatch, asr)
+    pipeline._asr_mode = "offline"
+    pipeline._runtime_asr["enableInterim"] = False
+
+    captured_events = []
+    captured_deltas = []
+
+    async def _capture_event(event: Dict[str, Any], priority: int = 20):
+        _ = priority
+        captured_events.append(event)
+
+    async def _capture_delta(text: str):
+        captured_deltas.append(text)
+
+    monkeypatch.setattr(pipeline, "_send_event", _capture_event)
+    monkeypatch.setattr(pipeline, "_emit_transcript_delta", _capture_delta)
+
+    await pipeline._on_transcript_callback("ignored interim", is_final=False)
+
+    assert captured_events == []
+    assert captured_deltas == []
+    assert pipeline._latest_asr_interim_text == ""
+
+
+@pytest.mark.asyncio
+async def test_offline_final_callback_emits_when_interim_disabled(monkeypatch):
+    asr = _FakeOfflineASR()
+    pipeline = _build_pipeline(monkeypatch, asr)
+    pipeline._asr_mode = "offline"
+    pipeline._runtime_asr["enableInterim"] = False
+
+    captured_events = []
+
+    async def _capture_event(event: Dict[str, Any], priority: int = 20):
+        _ = priority
+        captured_events.append(event)
+
+    monkeypatch.setattr(pipeline, "_send_event", _capture_event)
+
+    await pipeline._on_transcript_callback("final only", is_final=True)
+
+    assert any(event.get("type") == "transcript.final" for event in captured_events)
+
+
 @pytest.mark.asyncio
 async def test_streaming_eou_falls_back_to_latest_interim(monkeypatch):
     asr = _FakeStreamingASR()