fix(dictate): finish Muse on final transcript

services/dictate/src/pi_dictate_service/backend.py

@@ -422,8 +422,11 @@ class QueuedCloudTurn:
                         self._provider_event_bytes -= size
                     raise ProviderError(self._provider_prefix + "_event_overflow", "provider event queue is full") from error
         except ProviderError as error:
+            # Preserve a provider failure that races with successful
+            # finalization. Cancellation does not consume this error, but it
+            # avoids recursively closing a transport we are already closing.
+            self._receiver_error = error
             if not self._cancel_event.is_set():
-                self._receiver_error = error
                 self._abort_transport()
         except Exception:
             if not self._cancel_event.is_set():

services/dictate/src/pi_dictate_service/backends/meta.py

@@ -177,13 +177,10 @@ class MetaTurn(QueuedCloudTurn):
 
     def _finish_provider(self) -> str:
         deadline = time.monotonic() + self._timeout
-        transcript: str | None = None
         while time.monotonic() < deadline:
             event = self._next_provider_event(max(0.01, deadline - time.monotonic()))
             if event is None:
-                if transcript is None:
-                    raise ProviderError("meta_premature_close", "Meta closed without a final transcript")
-                return transcript
+                raise ProviderError("meta_premature_close", "Meta closed without a final transcript")
             _validate_meta_event(event)
             event_type = event["type"]
             if event_type == "error":
@@ -198,11 +195,9 @@ class MetaTurn(QueuedCloudTurn):
             if isinstance(value, Mapping):
                 value = value.get("text")
             try:
-                transcript = bounded_provider_text(value)
+                return bounded_provider_text(value)
             except ProviderError as error:
                 raise ProviderError("meta_malformed_event", "Meta sent an invalid transcript") from error
-            # Keep reading until the provider's normal close.  A subsequent
-            # error or abnormal close therefore cannot be mistaken for success.
         raise ProviderError("meta_timeout", "timed out waiting for Meta final transcript")
 
 

services/dictate/tests/test_backends.py

@@ -59,6 +59,44 @@ class FakeTransport:
         self.closed = True
 
 
+class PersistentTransport(FakeTransport):
+    """Keep the provider session open after all queued events are delivered."""
+
+    def __init__(self, events: list[str | bytes | None]) -> None:
+        super().__init__(events)
+        self.close_requested = threading.Event()
+
+    def recv(self, timeout: float | None = None) -> str | bytes | None:
+        if self.events:
+            return self.events.pop(0)
+        if self.close_requested.wait(timeout):
+            return None
+        raise TimeoutError
+
+    def close(self) -> None:
+        self.close_requested.set()
+        super().close()
+
+
+class ErrorOnCloseTransport(PersistentTransport):
+    """Release a provider error concurrently with client-initiated closure."""
+
+    def __init__(self, events: list[str | bytes | None], error_event: str) -> None:
+        super().__init__(events)
+        self.error_event = error_event
+        self.error_delivered = False
+
+    def recv(self, timeout: float | None = None) -> str | bytes | None:
+        if self.events:
+            return self.events.pop(0)
+        if not self.close_requested.wait(timeout):
+            raise TimeoutError
+        if not self.error_delivered:
+            self.error_delivered = True
+            return self.error_event
+        return None
+
+
 class BlockingCloseTransport(FakeTransport):
     def __init__(self, events: list[str | bytes | None]) -> None:
         super().__init__(events)
@@ -137,6 +175,47 @@ class BackendTests(unittest.TestCase):
         self.assertEqual(transport.sent[1], b"pcm")
         self.assertEqual(json.loads(transport.sent[2]), {"type": "endStream"})
 
+    def test_meta_final_transcript_does_not_wait_for_provider_close(self) -> None:
+        key, terms, temporary = self.make_files()
+        self.addCleanup(temporary.cleanup)
+        acknowledgement = (FIXTURE_ROOT / "meta" / "session_ack.json").read_text(encoding="utf-8").strip()
+        transport = PersistentTransport([acknowledgement, *fixture_events("meta")])
+        backend = MetaBackend(
+            MetaConfig("muse-voice-transcribe-1.0", key, terms),
+            transport_factory=lambda _endpoint: transport,
+            timeout=1,
+        )
+
+        turn = backend.start_turn()
+        turn.enqueue_pcm(b"pcm")
+        started_at = time.monotonic()
+        self.assertEqual(turn.finish(), "final fixture transcript")
+
+        self.assertLess(time.monotonic() - started_at, 0.5)
+        self.assertTrue(transport.closed)
+
+    def test_meta_provider_error_racing_with_finalization_is_authoritative(self) -> None:
+        key, terms, temporary = self.make_files()
+        self.addCleanup(temporary.cleanup)
+        acknowledgement = (FIXTURE_ROOT / "meta" / "session_ack.json").read_text(encoding="utf-8").strip()
+        transport = ErrorOnCloseTransport(
+            [acknowledgement, *fixture_events("meta")],
+            json.dumps({"type": "error", "code": "provider_failure"}),
+        )
+        backend = MetaBackend(
+            MetaConfig("muse-voice-transcribe-1.0", key, terms),
+            transport_factory=lambda _endpoint: transport,
+            timeout=1,
+        )
+
+        turn = backend.start_turn()
+        turn.enqueue_pcm(b"pcm")
+        with self.assertRaises(ProviderError) as error:
+            turn.finish()
+
+        self.assertEqual(error.exception.code, "meta_error")
+        self.assertTrue(transport.error_delivered)
+
     def test_openai_uses_ga_session_shape_and_correlates_committed_item(self) -> None:
         key, terms, temporary = self.make_files()
         self.addCleanup(temporary.cleanup)