fix(iris): handle scheme in actor proxy upstream URL (#4157)

## Summary
- The iris backend registers actor endpoints with a full URL including
scheme (e.g. `http://10.164.0.11:63143`), but the actor proxy
unconditionally prepended `http://`, producing a malformed
`http://http://...` URL that fails with "Name or service not known"
- Use the address as-is when it already contains a scheme; prepend
`http://` only for bare `host:port` addresses

## Test plan
- [x] Existing `test_actor_proxy.py` tests pass (bare `host:port` path
still works)
- [ ] Verify proxy works end-to-end with a running zephyr job after
controller redeploy

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Rafal Wojdyla <ravwojdyla@gmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

lib/iris/src/iris/actor/resolver.py

@@ -85,24 +85,24 @@ class ProxyResolver:
     URL so all RPCs go through the proxy. The proxy uses the
     ``X-Iris-Actor-Endpoint`` header to resolve the actual actor endpoint.
 
+    The caller passes the full actor name as registered in the endpoint registry
+    (e.g. ``/user/job/coordinator/actor-0``).
+
     Args:
         controller_url: Controller URL (e.g., ``http://localhost:8080``)
-        namespace: Namespace prefix for endpoint resolution
     """
 
-    def __init__(self, controller_url: str, namespace: str):
+    def __init__(self, controller_url: str):
         self._controller_url = controller_url.rstrip("/")
-        self._namespace = namespace
 
     def resolve(self, name: str) -> ResolveResult:
-        endpoint_name = f"{self._namespace}/{name}"
         return ResolveResult(
             name=name,
             endpoints=[
                 ResolvedEndpoint(
                     url=self._controller_url,
-                    actor_id=f"proxy-{endpoint_name}",
-                    metadata={ACTOR_ENDPOINT_HEADER: endpoint_name},
+                    actor_id=f"proxy-{name}",
+                    metadata={ACTOR_ENDPOINT_HEADER: name},
                 )
             ],
         )

lib/iris/src/iris/cluster/controller/actor_proxy.py

@@ -11,7 +11,7 @@
 Route pattern::
 
     POST /iris.actor.ActorService/{method}
-    X-Iris-Actor-Endpoint: namespace/actor-name
+    X-Iris-Actor-Endpoint: /user/job/actor-name
 """
 
 import logging
@@ -72,7 +72,8 @@ async def handle(self, request: Request) -> Response:
                 status_code=404,
             )
 
-        upstream_url = f"http://{address}/iris.actor.ActorService/{method}"
+        base = address if "://" in address else f"http://{address}"
+        upstream_url = f"{base}/iris.actor.ActorService/{method}"
         body = await request.body()
         forward_headers = {k: v for k, v in request.headers.items() if k.lower() not in _HOP_BY_HOP_HEADERS}
 

lib/iris/tests/actor/test_actor_proxy.py

@@ -33,23 +33,10 @@ def echo(self, message: str) -> str:
         return f"echo: {message}"
 
 
-class FakeEndpointDB:
-    """Minimal fake that satisfies ActorProxy._resolve_endpoint via the same interface as ControllerDB."""
-
-    def __init__(self):
-        self._endpoints: dict[str, str] = {}
-
-    def register(self, name: str, address: str) -> None:
-        self._endpoints[name] = address
-
-    def resolve(self, name: str) -> str | None:
-        return self._endpoints.get(name)
-
-
 class StandaloneActorProxy:
     """A standalone proxy for testing without a full controller.
 
-    Wraps ActorProxy's forwarding logic but uses a simple dict-based
+    Mirrors ActorProxy's forwarding logic but uses a simple dict-based
     endpoint registry instead of ControllerDB.
     """
 
@@ -81,7 +68,8 @@ async def handle(self, request):
                 status_code=404,
             )
 
-        upstream_url = f"http://{address}/iris.actor.ActorService/{method}"
+        base = address if "://" in address else f"http://{address}"
+        upstream_url = f"{base}/iris.actor.ActorService/{method}"
         body = await request.body()
         # Forward all headers except hop-by-hop and routing headers.
         skip = frozenset({"host", "transfer-encoding", "connection", "keep-alive", ACTOR_ENDPOINT_HEADER})
@@ -125,17 +113,19 @@ def _start_proxy_server(proxy: StandaloneActorProxy, threads: ThreadContainer) -
 def test_proxy_round_trip():
     """Full round-trip: client → proxy → actor server → response."""
     threads = ThreadContainer()
+
+    full_name = "test-ns/status"
     actor_server = ActorServer(host="127.0.0.1", threads=threads)
-    actor_server.register("status", StatusActor())
+    actor_server.register(full_name, StatusActor())
     actor_port = actor_server.serve_background()
 
     proxy = StandaloneActorProxy()
-    proxy.register("test-ns/status", f"127.0.0.1:{actor_port}")
+    proxy.register(full_name, f"127.0.0.1:{actor_port}")
     proxy_port = _start_proxy_server(proxy, threads)
 
     try:
-        resolver = ProxyResolver(f"http://127.0.0.1:{proxy_port}", namespace="test-ns")
-        client = ActorClient(resolver, "status")
+        resolver = ProxyResolver(f"http://127.0.0.1:{proxy_port}")
+        client = ActorClient(resolver, full_name)
 
         result = client.get_status()
         assert result["documents_processed"] == 1
@@ -151,6 +141,33 @@ def test_proxy_round_trip():
         threads.stop()
 
 
+def test_proxy_namespaced_actor():
+    """Proxy forwards to an actor registered under a full namespaced path.
+
+    This mirrors real iris backend behavior where actors are registered as
+    /user/job/coordinator/actor-0 and the address includes the http:// scheme.
+    """
+    threads = ThreadContainer()
+
+    full_name = "/user/my-job/coordinator/status-0"
+    actor_server = ActorServer(host="127.0.0.1", threads=threads)
+    actor_server.register(full_name, StatusActor())
+    actor_port = actor_server.serve_background()
+
+    proxy = StandaloneActorProxy()
+    proxy.register(full_name, f"http://127.0.0.1:{actor_port}")
+    proxy_port = _start_proxy_server(proxy, threads)
+
+    try:
+        resolver = ProxyResolver(f"http://127.0.0.1:{proxy_port}")
+        client = ActorClient(resolver, full_name)
+
+        result = client.get_status()
+        assert result["documents_processed"] == 1
+    finally:
+        threads.stop()
+
+
 def test_proxy_missing_endpoint_header():
     """Proxy returns 400 when the endpoint header is missing."""
     threads = ThreadContainer()