From 9447ace2c3ca62eeffb1fc3621e5cec727d8f0ec Mon Sep 17 00:00:00 2001
From: Palash Debnath <4178343+debpalash@users.noreply.github.com>
Date: Sun, 30 Aug 2026 22:21:18 +0530
Subject: [PATCH] fix: invalidate unloaded ASR evidence

---
 backend/services/asr_backend.py         | 34 +++++++++++++---------
 tests/test_engine_execution_evidence.py | 38 +++++++++++++++++++++++++
 2 files changed, 58 insertions(+), 14 deletions(-)

diff --git a/backend/services/asr_backend.py b/backend/services/asr_backend.py
index 74ec138e..116f2b79 100644
--- a/backend/services/asr_backend.py
+++ b/backend/services/asr_backend.py
@@ -30,6 +30,7 @@ import re
 import contextlib
 import threading
 import time
+import weakref
 from utils.containment import contain_system_exit
 
 from abc import ABC, abstractmethod
@@ -2369,6 +2370,7 @@ _LAST_ERRORS: dict[str, str] = {}
 # reinstall / ``uv sync --reinstall`` and an app restart anyway.
 _DEEP_IMPORT_BROKEN: dict[str, str] = {}
 _RUNTIME_EVIDENCE: dict[str, dict] = {}
+_RUNTIME_INSTANCES: weakref.WeakValueDictionary[str, "ASRBackend"] = weakref.WeakValueDictionary()
 
 
 def _deep_import_reason(cls: type["ASRBackend"], exc: ImportError) -> str:
@@ -2425,20 +2427,23 @@ def list_backends() -> list[dict]:
         isolation = "subprocess" if getattr(cls, "_is_subprocess_isolated", False) else "in-process"
         gpu_compat = getattr(cls, "gpu_compat", ("cpu",))
         routing = routing_fields(gpu_compat, caps)
-        execution_evidence = _RUNTIME_EVIDENCE.get(bid)
-        if isolation == "subprocess":
-            # Cached load-time facts are valid only while this exact child is
-            # still alive. Recompute lifecycle state so shutdown/reaping cannot
-            # leave a ghost "loaded" engine in diagnostics.
-            execution_evidence = execution_snapshot(
-                engine_id=bid,
-                engine_cls=cls,
-                instance=_ISOLATED_INSTANCES.get(bid),
-                routing=routing,
-                caps=caps,
-            )
-            if execution_evidence["evidence_state"] == "not_loaded":
-                _RUNTIME_EVIDENCE.pop(bid, None)
+        # Cached load-time facts are valid only while their exact backend still
+        # owns live model state. Recompute from that instance so unload/reaping
+        # cannot leave ghost GPU/provider evidence in diagnostics.
+        instance = (
+            _ISOLATED_INSTANCES.get(bid)
+            if isolation == "subprocess"
+            else _RUNTIME_INSTANCES.get(bid)
+        )
+        execution_evidence = execution_snapshot(
+            engine_id=bid,
+            engine_cls=cls,
+            instance=instance,
+            routing=routing,
+            caps=caps,
+        )
+        if execution_evidence["evidence_state"] == "not_loaded":
+            _RUNTIME_EVIDENCE.pop(bid, None)
         out.append({
             "id": bid,
             "display_name": cls.display_name,
@@ -2712,6 +2717,7 @@ def load_active_asr_backend(*, asr_pipe=None) -> ASRBackend:
                 routing=routing,
                 caps=caps,
             )
+            _RUNTIME_INSTANCES[bid] = backend
             return backend
         except ImportError as e:
             # ModuleNotFoundError and its ImportError parent ("cannot import
diff --git a/tests/test_engine_execution_evidence.py b/tests/test_engine_execution_evidence.py
index 2f373dfb..29599840 100644
--- a/tests/test_engine_execution_evidence.py
+++ b/tests/test_engine_execution_evidence.py
@@ -232,3 +232,41 @@ def test_stopped_asr_sidecar_invalidates_cached_loaded_evidence(monkeypatch):
     row = asr_backend.list_backends()[0]
     assert row["execution_evidence"]["evidence_state"] == "not_loaded"
     assert "stopped" not in asr_backend._RUNTIME_EVIDENCE
+
+
+def test_unloaded_in_process_asr_invalidates_cached_loaded_evidence(monkeypatch):
+    from services import asr_backend
+
+    class _Backend:
+        id = "released"
+        display_name = "Released backend"
+        gpu_compat = ("cuda", "cpu")
+
+        def __init__(self):
+            self._model = object()
+
+        @classmethod
+        def is_available(cls):
+            return True, "ready"
+
+        def execution_evidence_loaded(self):
+            return self._model is not None
+
+        def unload(self):
+            self._model = None
+
+    instance = _Backend()
+    monkeypatch.setattr(asr_backend, "_REGISTRY", {"released": _Backend})
+    monkeypatch.setattr(asr_backend, "_RUNTIME_INSTANCES", {"released": instance})
+    monkeypatch.setattr(
+        asr_backend,
+        "_RUNTIME_EVIDENCE",
+        {"released": {"evidence_state": "loaded", "actual_execution_device": "cuda:0"}},
+    )
+
+    instance.unload()
+    row = asr_backend.list_backends()[0]
+
+    assert row["execution_evidence"]["evidence_state"] == "not_loaded"
+    assert row["execution_evidence"]["actual_execution_device"] is None
+    assert "released" not in asr_backend._RUNTIME_EVIDENCE