feat(iris): route actor RPC calls through proxy and decode responses (#4161)

## Summary
- `iris rpc actor call --actor-name <full-name> --method-name <method>`
now automatically sets the `x-iris-actor-endpoint` header, routing the
call through the controller's actor proxy to the correct actor server
- ActorResponse `serialized_value` is auto-unpickled so the CLI prints
human-readable JSON instead of opaque base64

### Example
```bash
uv run iris --config lib/iris/examples/marin-dev.yaml rpc actor call \
  --actor-name "/rav/my-job/zephyr-coord-0" \
  --method-name get_counters

# Output:
{
  "minhash/documents": 7849813,
  "minhash/buckets": 204095138
}
```

## Test plan
- [x] Tested live against a running zephyr dedup job on marin-dev
- [ ] Add unit test for `_format_actor_response`

🤖 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/client.py

@@ -113,7 +113,7 @@ def rpc_client(self) -> ActorServiceClientSync:
             len(result.endpoints),
         )
         endpoint = result.first()
-        logger.info("First endpoint: %s", endpoint)
+        logger.info("First endpoint: url=%s, actor_id=%s", endpoint.url, endpoint.actor_id)
         self._rpc_headers = dict(endpoint.metadata)
         self._rpc_client = ActorServiceClientSync(
             address=endpoint.url,

lib/iris/src/iris/cli/actor.py

@@ -0,0 +1,69 @@
+# Copyright The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""CLI for calling actor methods through the controller proxy."""
+
+import json
+
+import click
+
+from iris.actor import ActorClient, ProxyResolver
+from iris.actor.resolver import ResolveResult
+from iris.cli.main import require_controller_url
+from iris.rpc.auth import GcpAccessTokenProvider, TokenProvider
+
+
+class _AuthProxyResolver(ProxyResolver):
+    """ProxyResolver that injects a bearer token for dashboard auth."""
+
+    def __init__(self, controller_url: str, token_provider: TokenProvider):
+        super().__init__(controller_url)
+        self._token_provider = token_provider
+
+    def resolve(self, name: str) -> ResolveResult:
+        result = super().resolve(name)
+        token = self._token_provider.get_token()
+        if token:
+            for ep in result.endpoints:
+                ep.metadata["authorization"] = f"Bearer {token}"
+        return result
+
+
+@click.group()
+def actor():
+    """Interact with actors via the controller proxy."""
+    pass
+
+
+@actor.command()
+@click.argument("endpoint")
+@click.argument("method")
+@click.argument("kwargs", required=False, default=None)
+@click.option("--timeout", type=float, default=30.0, help="RPC timeout in seconds")
+@click.pass_context
+def call(ctx: click.Context, endpoint: str, method: str, kwargs: str | None, timeout: float):
+    """Call an actor method through the controller proxy.
+
+    ENDPOINT is the full actor name as registered in the endpoint registry
+    (e.g. /user/job/coordinator/actor-0).
+
+    METHOD is the method name to call (e.g. get_counters).
+
+    KWARGS is an optional JSON object of keyword arguments (e.g. '{"worker_id": "w-3"}').
+    """
+    controller_url = require_controller_url(ctx)
+    tp = ctx.obj.get("token_provider") if ctx.obj else None
+    if tp is None:
+        tp = GcpAccessTokenProvider()
+
+    resolver = _AuthProxyResolver(controller_url, tp)
+    client = ActorClient(resolver, endpoint, call_timeout=timeout)
+
+    parsed_kwargs = json.loads(kwargs) if kwargs else {}
+    rpc_method = getattr(client, method)
+    result = rpc_method(**parsed_kwargs)
+
+    try:
+        click.echo(json.dumps(result, indent=2, default=str))
+    except (TypeError, ValueError):
+        click.echo(repr(result))

lib/iris/src/iris/cli/main.py

@@ -337,11 +337,13 @@ def key_revoke(ctx, key_id: str):
 from iris.cli.build import build  # noqa: E402
 from iris.cli.cluster import cluster  # noqa: E402
 from iris.cli.job import job  # noqa: E402
+from iris.cli.actor import actor as actor_cmd  # noqa: E402
 from iris.cli.process_status import register_process_status_commands  # noqa: E402
 from iris.cli.query import query_cmd  # noqa: E402
 from iris.cli.rpc import register_rpc_commands  # noqa: E402
 from iris.cli.task import task  # noqa: E402
 
+iris.add_command(actor_cmd)
 iris.add_command(cluster)
 iris.add_command(build)
 iris.add_command(job)