harden #5979 follow-up per Codex gate: non-blocking disk-only warm + profile scope

Codex gate found two real issues in the send-path warm:

1. Blocking lock wait (CORE): get_available_models(prefer_cache=True) still
   acquires _available_models_cache_lock and can wait up to ~60s (unbounded in
   synchronous rebuild mode) on an in-flight rebuild — unacceptable on the send
   hot path. Rewrote warm_models_catalog_provenance_if_cold as a genuinely
   non-blocking, disk-only publish: try the cache lock NON-BLOCKING (return
   immediately if busy — a concurrent build will publish provenance itself),
   read ONLY the on-disk cache via _load_models_cache_from_disk (no network, no
   rebuild), publish snapshot+fingerprint + _sync_models_cache_provenance().
   Verified: returns in ~3.5ms while the lock is held (was a 60s block).

2. Profile scope (CORE): the streaming worker is a separate thread that does NOT
   inherit the HTTP handler's request-profile TLS. Both the warm and the resolve
   read profile-keyed config (cache path + source fingerprint via
   get_active_profile_name), so a cold send from a NAMED profile could resolve
   against the DEFAULT profile's config and route to the wrong provider/base_url
   (get_available_models reloads config on path mismatch). Wrapped warm + resolve
   in profile_scope_for_detached_worker(_resolved_profile_name) so both see the
   owning session's profile (no-op for default/root).

Tests: rewrote never-live-rebuilds (now asserts warm NEVER calls
get_available_models — reads disk directly) and added a non-blocking regression
(warm returns <2s while the cache lock is held). 65 resolver+sprint40 tests pass;
b3nw end-to-end + non-blocking + disk-warm all verified.

api/config.py

@@ -8172,7 +8172,7 @@ def _models_cache_file_age_seconds(cache_path: Path, now: float) -> float | None
 
 
 def warm_models_catalog_provenance_if_cold() -> None:
-    """Best-effort, network-free publish of catalog provenance when memory is cold.
+    """Best-effort, NON-BLOCKING, disk-only publish of catalog provenance.
 
     The send path (``api/streaming.py``) resolves the wire model via
     ``resolve_model_provider`` without ever building the models catalog, and the
@@ -8182,28 +8182,50 @@ def warm_models_catalog_provenance_if_cold() -> None:
     invalidation, not a rare race. In that state the custom-proxy provenance
     signal (``_endpoint_advertised_model_ids``) is ``None`` and resolution falls
     to the cold-preserve default; this helper restores the endpoint-advertised
-    signal from the durable disk cache so the #433 bare-only-strip stays exact
-    and #5979 full-id preserve is backed by real provenance rather than the
-    fail-safe default.
+    signal from the durable disk cache so the #433 bare-only-strip stays exact.
 
-    Contract: ``prefer_cache=True`` resolves fresh-memory → valid-disk-cache
-    (which publishes provenance) → minimal static catalog (which deliberately
-    does NOT publish provenance, so a config-derived catalog can never
-    masquerade as endpoint evidence). It NEVER triggers a live provider rebuild,
-    so there is no network latency or flaky-network stall. Costs one global read
-    when already warm and a single ~sub-millisecond disk load once per cold
-    window. MUST be called from a request/worker context that does NOT hold
-    ``_cfg_lock`` (``get_available_models`` takes ``_available_models_cache_lock``
-    and may ``reload_config()`` → ``_cfg_lock``; calling it under ``_cfg_lock``
-    would invert the lock order). The send path satisfies this — it runs on the
-    detached worker thread with no config lock held.
+    Deliberately does NOT call ``get_available_models(prefer_cache=True)``: even
+    in prefer-cache mode that acquires ``_available_models_cache_lock`` and can
+    block up to ~60s waiting on an in-flight rebuild (unbounded in synchronous
+    rebuild mode) — unacceptable on the send hot path. Instead this:
+      * tries the cache lock NON-BLOCKING and returns immediately if it's busy
+        (a concurrent rebuild will publish provenance itself);
+      * reads ONLY the on-disk cache (no network, no live probe, no rebuild);
+      * publishes the snapshot + source fingerprint via the same globals the
+        real publish sites use, then ``_sync_models_cache_provenance()``.
+    Publishing the fingerprint from the CURRENT runtime is correct: the disk
+    cache is validated by schema/version/source-fingerprint on load
+    (``_is_loadable_disk_cache``), so a load success means it belongs to this
+    profile. Callers must not hold ``_cfg_lock`` (this reads config for the
+    fingerprint); the send worker satisfies that.
     """
+    global _available_models_cache, _available_models_cache_ts
+    global _available_models_cache_source_fingerprint
     if _models_cache_provenance is not None:
         return  # already warm — one global read, no work
+    got = _available_models_cache_lock.acquire(blocking=False)
+    if not got:
+        return  # a concurrent build/publish holds the lock; it will publish
     try:
-        get_available_models(prefer_cache=True)
+        if _models_cache_provenance is not None:
+            return  # published while we waited for the lock
+        try:
+            disk_groups = _load_models_cache_from_disk()
+        except Exception:
+            disk_groups = None
+        if disk_groups is None:
+            return  # no durable cache → stay cold, resolver preserves verbatim
+        _available_models_cache = disk_groups
+        _available_models_cache_ts = time.monotonic()
+        try:
+            _available_models_cache_source_fingerprint = _models_cache_source_fingerprint()
+        except Exception:
+            _available_models_cache_source_fingerprint = None
+        _sync_models_cache_provenance()
     except Exception:
         logger.debug("models catalog provenance warm failed", exc_info=True)
+    finally:
+        _available_models_cache_lock.release()
 
 
 def get_available_models_for_session_visit() -> dict:

api/streaming.py

@@ -7893,14 +7893,23 @@ def _run_agent_streaming(
             _session_db = _build_session_db_for_stream(_state_db_path)
             # #5979: publish catalog provenance from the durable disk cache when
             # memory is cold, so the custom-proxy resolver below sees the
-            # endpoint-advertised model ids (network-free, never live-rebuilds,
-            # no _cfg_lock held on this worker thread). Without this the resolver
-            # falls to the cold-preserve default; with it, #433 bare-only-strip
-            # stays exact and #5979 full-id preserve is provenance-backed.
-            warm_models_catalog_provenance_if_cold()
-            resolved_model, resolved_provider, resolved_base_url = resolve_model_provider(
-                model_with_provider_context(model, provider_context)
-            )
+            # endpoint-advertised model ids (non-blocking, disk-only, never
+            # live-rebuilds). Both the warm and the resolve read profile-keyed
+            # config (cache path + source fingerprint via get_active_profile_name),
+            # but this streaming worker is a separate thread that does NOT inherit
+            # the HTTP handler's request-profile TLS — without binding it, a cold
+            # send from a NAMED profile would resolve against the DEFAULT profile's
+            # config and route to the wrong provider/base_url. Bind the captured
+            # owning-session profile across warm + resolve so both see the right
+            # profile (no-op for the default/root profile).
+            from api import profiles as profiles_api
+            with profiles_api.profile_scope_for_detached_worker(
+                _resolved_profile_name, "model resolution", logger_override=logger
+            ):
+                warm_models_catalog_provenance_if_cold()
+                resolved_model, resolved_provider, resolved_base_url = resolve_model_provider(
+                    model_with_provider_context(model, provider_context)
+                )
             configured_base_url = resolved_base_url
 
             # Resolve API key via Hermes runtime provider (matches gateway behaviour).

tests/test_model_resolver.py

@@ -597,23 +597,57 @@ def test_warm_models_catalog_provenance_noop_when_already_warm_5979():
 
 
 def test_warm_models_catalog_provenance_never_live_rebuilds_5979(monkeypatch):
-    """The warm helper must call get_available_models with prefer_cache=True so
-    it can never trigger a live provider probe on the send path.
+    """The warm helper must NEVER call get_available_models (which can acquire
+    the cache lock and block on / trigger a rebuild). It reads the disk cache
+    directly, so a live provider probe is structurally impossible on the send
+    path.
     """
     old_prov = config._models_cache_provenance
     config._models_cache_provenance = None
-    captured = {}
-    def _fake_get_available_models(*, prefer_cache=False, force_refresh=False):
-        captured['prefer_cache'] = prefer_cache
-        captured['force_refresh'] = force_refresh
-        return {'groups': []}
-    monkeypatch.setattr(config, 'get_available_models', _fake_get_available_models)
+    called = {'get_available_models': False}
+    def _boom(*a, **k):
+        called['get_available_models'] = True
+        raise AssertionError("warm must not call get_available_models")
+    monkeypatch.setattr(config, 'get_available_models', _boom)
     try:
-        config.warm_models_catalog_provenance_if_cold()
+        config.warm_models_catalog_provenance_if_cold()  # must not raise / call the above
     finally:
         config._models_cache_provenance = old_prov
-    assert captured.get('prefer_cache') is True, "warm must use prefer_cache=True (no live rebuild)"
-    assert captured.get('force_refresh') is False, "warm must never force_refresh"
+    assert called['get_available_models'] is False, (
+        "warm helper must read disk directly, never via get_available_models"
+    )
+
+
+def test_warm_models_catalog_provenance_nonblocking_when_lock_held_5979():
+    """The warm helper must NOT block on the models-cache lock. It tries the
+    lock non-blocking and returns immediately when a concurrent build/publish
+    holds it, so a send can never hang up to 60s behind a rebuild.
+    """
+    import threading
+    import time as _time
+    old_prov = config._models_cache_provenance
+    config._models_cache_provenance = None
+    got = config._available_models_cache_lock.acquire(blocking=False)
+    assert got, "precondition: could not take cache lock"
+    try:
+        result = {}
+        def _worker():
+            t0 = _time.time()
+            config.warm_models_catalog_provenance_if_cold()
+            result['elapsed'] = _time.time() - t0
+        th = threading.Thread(target=_worker)
+        th.start()
+        th.join(timeout=5)
+        blocked = th.is_alive()
+    finally:
+        config._available_models_cache_lock.release()
+        config._models_cache_provenance = old_prov
+    if blocked:
+        th.join(timeout=5)
+    assert not blocked, "warm helper blocked on a held cache lock (must be non-blocking)"
+    assert result.get('elapsed', 99) < 2.0, (
+        f"warm helper should return promptly when lock held, took {result.get('elapsed')!r}s"
+    )
 
 
 def test_custom_remote_preserves_unknown_prefix_548():