Preserve retained adapter transport and adaptation policy

Per-operation reconstruction must honor the public router.lm() escape hatch for replacing transports. Sync and async regressions assert that only the replacement transport sends the request. This follows the managed snapshot patch without changing its authentication boundary.

Amp-Thread-ID: https://ampcode.com/threads/T-01a0c095-4917-7617-9dac-e99f1ef22d41
Co-authored-by: Isaac Miller <isaac@cmpnd.ai>

Author: ampagent

lm15/router.py

@@ -1170,8 +1170,9 @@ def _build_managed_lm(resolution: Resolution, config: RouterConfig, adapters: Ma
         inner = getattr(lm, "_inner", lm)
         if snapshot is None:
             def prepare():
+                execution_extra = dict(original_extra, transport=lm.transport, adaptations=lm.adaptations)
                 return _build_managed_lm(
-                    resolution, config, adapters, cls, original_definition, dict(original_extra), hosted,
+                    resolution, config, adapters, cls, original_definition, execution_extra, hosted,
                     original_url, snapshot=auth.request_auth(provider),
                 )
             lm._managed_prepare = prepare

tests/test_managed_dispatch.py

@@ -81,6 +81,20 @@ def test_retained_adapter_follows_whole_local_connection(asynchronous, operation
         wire_pair(sent, f"token-{label}", f"https://{label}.example/v1{path}")
 
 
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_retained_adapter_uses_replacement_transport(asynchronous):
+    auth = Auth.memory()
+    local(auth, "a")
+    original = transport_for("complete", asynchronous)
+    replacement = transport_for("complete", asynchronous)
+    lm = (AsyncLMRouter if asynchronous else LMRouter)(RouterConfig(auth=auth, transport=original)).lm("ollama:llama3")
+    lm.transport = replacement
+    invoke(lm, "complete", asynchronous)
+    assert original.requests == []
+    assert len(replacement.requests) == 1
+    wire_pair(replacement.requests[0], "token-a", "https://a.example/v1/chat/completions")
+
+
 @pytest.mark.parametrize("asynchronous", [False, True])
 def test_delayed_stream_never_sends_a_stale_snapshot(asynchronous):
     auth = Auth.memory()