abridge: SessionForward + Convert∘Session composition (PR #122 stage A)

agentix/sio.py

@@ -56,10 +56,15 @@ async def fetch_remote(self, payload):
 class RemoteSioError(RuntimeError):
     """Raised by `Namespace.request()` when the reply carries an `:error`."""
 
-    def __init__(self, type_: str, message: str) -> None:
+    def __init__(self, type_: str, message: str, status_code: int | None = None) -> None:
         super().__init__(f"{type_}: {message}")
         self.type = type_
         self.message = message
+        # Upstream HTTP status the host handler chose (e.g. abridge's
+        # AbridgeError 429/400); None when the error envelope carried no
+        # status. Lets a protocol layer reply with the real status instead
+        # of collapsing every remote failure to one blanket code.
+        self.status_code = status_code
 
 
 # ── module-level bridge state ──────────────────────────────────────
@@ -241,10 +246,12 @@ async def _on_reply_error(self, payload: Any) -> None:
         fut = self._pending_requests.get(req_id) if isinstance(req_id, str) else None
         if fut is not None and not fut.done():
             err = payload.get("error") or {"type": "Unknown", "message": ""}
+            status_code = err.get("status_code")
             fut.set_exception(
                 RemoteSioError(
                     err.get("type", "Unknown"),
                     err.get("message", ""),
+                    status_code if isinstance(status_code, int) else None,
                 )
             )
 

plugins/abridge/README.md

@@ -49,7 +49,7 @@ from agentix.bridge.clients import AnthropicFromOpenAIClient
 client = AnthropicFromOpenAIClient(
     base_url="https://api.openai.com/v1",   # OpenAI / OpenRouter / vLLM / your gateway
     api_key="sk-...",
-    upstream_model="gpt-4o",                # the agent keeps sending claude-* model ids
+    model="gpt-4o",                         # the agent keeps sending claude-* model ids
 )
 proxy = Proxy(client)
 
@@ -75,12 +75,14 @@ client = OpenAIClient(base_url=..., api_key=..., model="gpt-4o")
 proxy = Proxy(client)
 
 async with proxy.session(sandbox) as handle:
-    await sandbox.remote(agent, base_url=f"{handle.url}/v1", api_key=PLACEHOLDER_API_KEY)
+    await sandbox.remote(agent, env=client.environ(handle))
 ```
 
-`OpenAIClient` doesn't ship an `environ(handle)` helper — most OpenAI
-SDK callers construct the client with explicit `base_url=`/`api_key=`
-arguments rather than reading env vars.
+`client.environ(handle)` returns
+`{"OPENAI_BASE_URL": handle.url + "/v1", "OPENAI_API_KEY": "<placeholder>"}` —
+the `/v1` suffix the OpenAI SDK expects is baked in so a caller can't
+drop it. Agents that construct their SDK client explicitly can still
+pass `base_url=f"{handle.url}/v1", api_key=PLACEHOLDER_API_KEY` instead.
 
 ### Anthropic agent → native Anthropic upstream
 
@@ -252,6 +254,7 @@ agentix/bridge/
     ├── openai.py                  # OpenAIClient (openai SDK) + PLACEHOLDER_API_KEY
     ├── anthropic.py               # AnthropicClient (anthropic SDK) + environ() + PLACEHOLDER_API_KEY
     ├── anthropic_from_openai.py   # AnthropicFromOpenAIClient (openai SDK + translation) + environ()
+    ├── anthropic_to_openai.py     # AnthropicToOpenAI (SDK-free translation over any Handler) + environ()
     ├── _genai_span.py             # populate_openai_span / populate_anthropic_span
     └── _anthropic_transforms.py   # pure Anthropic↔OpenAI converters
 ```

plugins/abridge/agentix/bridge/__init__.py

@@ -25,20 +25,21 @@
 via Python multiple inheritance (mixins) or by passing several to
 `Proxy(*handlers)`. Two handlers must not register the same path.
 
-`agentix.bridge.clients` ships the three standard handlers
-(`OpenAIClient`, `AnthropicClient`, `AnthropicFromOpenAIClient`), each
-built on the corresponding provider SDK. Pull from there as building
-blocks; abridge core stays shape-blind.
+`agentix.bridge.clients` ships the standard handlers (`OpenAIClient`,
+`AnthropicClient`, `AnthropicFromOpenAIClient`, and the transport-blind
+`AnthropicToOpenAI`). Pull from there as building blocks; abridge core
+stays shape-blind.
 """
 
 from __future__ import annotations
 
-from .forward import Forward
+from .forward import Forward, SessionForward
 from .proxy import (
     NAMESPACE,
     AbridgeError,
     Client,
     ClientResponse,
+    DynamicRoutes,
     Handler,
     Proxy,
     Request,
@@ -54,11 +55,13 @@
     "Client",
     "ClientResponse",
     "Command",
+    "DynamicRoutes",
     "Forward",
     "Handler",
     "NAMESPACE",
     "Proxy",
     "Request",
+    "SessionForward",
     "Sidecar",
     "SidecarError",
     "TunnelHandle",

plugins/abridge/agentix/bridge/clients/__init__.py

@@ -1,24 +1,32 @@
 """Bundled handler clients for abridge.
 
-Three out-of-the-box implementations, all built around the official
-provider SDKs (`openai`, `anthropic`). Each is a plain class with
-`@on(path)`-decorated methods — pass an instance to `Proxy(...)` or
-mixin-compose multiple in one user-defined class.
+Four out-of-the-box implementations. Each is a plain class with handler
+methods — pass an instance to `Proxy(...)` or mixin-compose multiple in
+one user-defined class. The first three own their upstream transport via
+the official provider SDKs (`openai`, `anthropic`); the fourth is
+SDK-free and transport-blind.
 
   * `OpenAIClient` — agent speaks OpenAI Chat Completions, upstream is
     OpenAI-compatible. One `@on("/v1/chat/completions")`.
   * `AnthropicClient` — agent speaks Anthropic Messages, upstream is
     native Anthropic. `@on("/v1/messages")` + `@on("/v1/messages/count_tokens")`.
   * `AnthropicFromOpenAIClient` — agent speaks Anthropic, upstream is
-    OpenAI-compatible (translation lives here). Same path set as
-    `AnthropicClient`.
+    OpenAI-compatible (translation lives here, transport owned by the
+    `openai` SDK). Same path set as `AnthropicClient`.
+  * `AnthropicToOpenAI` — the same translation direction, but over ANY
+    abridge `Handler` downstream (`Forward(...).handler()`,
+    `SessionForward(...).handler()`, …): the downstream owns HTTP,
+    sessions, and recording. Pick this one to compose with a
+    session-scoped recorder; pick `AnthropicFromOpenAIClient` when a
+    plain SDK-managed upstream is all you need.
 
-The two Anthropic-side classes also expose `environ(handle)` (instance
-method) — the env-var bundle (`ANTHROPIC_BASE_URL` + placeholder
-`ANTHROPIC_API_KEY`) an in-sandbox Anthropic SDK needs to route through
-the tunnel. The OpenAI client doesn't ship an `environ`; agents that
-use the OpenAI SDK typically construct the client with
-`base_url=handle.url + "/v1"` directly.
+All four classes expose `environ(handle)` (instance method) — the env-var
+bundle an in-sandbox SDK needs to route through the tunnel, so the wiring step
+is `env=client.environ(handle)` uniformly. `OpenAIClient` returns
+`{OPENAI_BASE_URL: handle.url + "/v1", OPENAI_API_KEY: placeholder}` — the `/v1`
+suffix the OpenAI SDK expects is baked in so a caller can't drop it; the two
+Anthropic-side classes return `{ANTHROPIC_BASE_URL: handle.url, ANTHROPIC_API_KEY:
+placeholder}` (no `/v1` — the Anthropic SDK appends it itself).
 
 The two `populate_*_span` helpers are exposed at this level so user-
 written clients can stamp the same OTel GenAI attrs the bundled
@@ -31,13 +39,15 @@
 from .anthropic import PLACEHOLDER_API_KEY as ANTHROPIC_PLACEHOLDER_API_KEY
 from .anthropic import AnthropicClient
 from .anthropic_from_openai import AnthropicFromOpenAIClient
+from .anthropic_to_openai import AnthropicToOpenAI
 from .openai import PLACEHOLDER_API_KEY as OPENAI_PLACEHOLDER_API_KEY
 from .openai import OpenAIClient
 
 __all__ = [
     "ANTHROPIC_PLACEHOLDER_API_KEY",
     "AnthropicClient",
     "AnthropicFromOpenAIClient",
+    "AnthropicToOpenAI",
     "OPENAI_PLACEHOLDER_API_KEY",
     "OpenAIClient",
     "populate_anthropic_span",

plugins/abridge/agentix/bridge/clients/anthropic_to_openai.py

@@ -0,0 +1,166 @@
+"""`AnthropicToOpenAI` — the Convert capability: an Anthropic-Messages agent over
+ANY OpenAI-compatible downstream.
+
+This is one orthogonal capability — protocol translation — and nothing else. It
+translates the agent's Anthropic `/v1/messages` to an OpenAI chat-completions body,
+hands that body to a `downstream` (any abridge `Handler`), then translates the
+OpenAI completion back to Anthropic. It is deliberately transport-blind: the
+downstream owns the HTTP, the session, and any recording.
+
+Compose it with a transport/routing capability:
+
+    # plain OpenAI gateway (no session)
+    Proxy(AnthropicToOpenAI(Forward(base_url, paths=["/v1/chat/completions"]).handler()))
+
+    # session-scoped recorder (the TITO gateway) — session stays transparent
+    tito  = SessionForward(gateway_url, paths=["/v1/chat/completions"])
+    proxy = Proxy(AnthropicToOpenAI(tito.handler(), model="qwen3-4b"))
+    ...                          # the agent only ever speaks Anthropic
+    harvest(tito.session_id)     # the session lives in the SessionForward, not here
+
+`AnthropicToOpenAI` knows nothing about sessions; `SessionForward` knows nothing
+about Anthropic. They compose because the seam between them is just abridge's
+`Handler` (an OpenAI chat body in, a `ClientResponse` out). No OpenAI SDK
+dependency — the upstream hop is whatever `Handler` you pass.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from agentix.utils import trace
+
+from ..proxy import ClientResponse, Handler, Request, TunnelHandle, _AsyncCloseable, on
+from ._anthropic_transforms import (
+    anthropic_messages_to_openai,
+    anthropic_sse,
+    count_anthropic_tokens,
+    openai_to_anthropic_messages,
+)
+from ._genai_span import populate_anthropic_span
+from .anthropic import PLACEHOLDER_API_KEY
+
+logger = logging.getLogger(__name__)
+
+
+class AnthropicToOpenAI:
+    """Anthropic Messages agent → any OpenAI-compatible `downstream` Handler.
+
+    `downstream` is the transport/routing capability this converter sits on top of
+    — a `Forward(...).handler()` for a direct OpenAI gateway, or a
+    `SessionForward(...).handler()` for a session-scoped recorder like the TITO
+    gateway. `model`, when set, overrides the agent's model id in the OpenAI body.
+    `count_tokens` is answered locally (character estimate, no downstream call).
+
+    `aclose()` delegates to a closeable downstream (a `Forward.handler()` carries
+    one), so `Proxy.stop` reaps the transport's HTTP pool through the composition.
+    Everything else about the downstream stays yours: hold the forwarder to read
+    `.session_id` / call `delete_session()`.
+    """
+
+    def __init__(self, downstream: Handler, *, model: str | None = None) -> None:
+        self._downstream = downstream
+        self._model = model
+        # Memory of the EXACT assistant message the downstream returned, one
+        # entry per tool-calling turn. The Anthropic round-trip is lossy (it
+        # drops reasoning_content and the per-tool-call `index`), so a
+        # reconstructed assistant won't match what a session-recording backend
+        # (TITO) stored byte-for-byte. On the next turn we replay the remembered
+        # original verbatim, keyed by the (id, name, canonical-args) triple of
+        # each tool call — all of which survive the round-trip. Ids alone are
+        # NOT a safe key: some servers reuse ids across turns (TGI emits "0"),
+        # and an id-only key would replay the latest turn into every matching
+        # history slot. Entries are never evicted — scope one converter
+        # instance per rollout/conversation.
+        self._assistant_by_calls: dict[
+            tuple[tuple[str, str, str], ...], dict[str, Any]
+        ] = {}
+
+    @on("/v1/messages")
+    async def messages(self, request: Request) -> ClientResponse:
+        openai_body = anthropic_messages_to_openai(request.body, upstream_model=self._model)
+        # The downstream produces a non-streaming OpenAI completion (a TITO
+        # recorder needs output_token_logprobs); we re-render SSE locally below if
+        # the agent asked for streaming.
+        openai_body["stream"] = False
+        self._replay_remembered_assistants(openai_body)
+        with trace.span(f"anthropic messages {request.body.get('model') or ''}"):
+            resp = await self._downstream(Request(path="/v1/messages", body=openai_body))
+            if resp.status_code != 200:
+                # Pass the downstream's error (4xx/5xx) straight through with its
+                # status; the agent sees a non-200, not a malformed body.
+                return resp
+            openai_resp = json.loads(resp.body)
+            self._remember_assistant(openai_resp)
+            anthropic_resp = openai_to_anthropic_messages(
+                openai_resp, response_model=str(request.body.get("model") or "")
+            )
+            populate_anthropic_span(request=request.body, response=anthropic_resp)
+            if request.body.get("stream"):
+                return ClientResponse.sse(anthropic_sse(anthropic_resp))
+            return ClientResponse.json(anthropic_resp)
+
+    @staticmethod
+    def _call_key(message: dict[str, Any]) -> tuple[tuple[str, str, str], ...]:
+        """One (id, name, canonical-args) triple per tool call. Arguments are
+        re-serialized with sorted keys so the downstream's original JSON string
+        and the round-trip's `json.dumps` rendering compare equal regardless of
+        spacing or key order."""
+        key: list[tuple[str, str, str]] = []
+        for tc in message.get("tool_calls") or []:
+            if not isinstance(tc, dict) or not tc.get("id"):
+                continue
+            function = tc.get("function") or {}
+            raw_args = function.get("arguments") or "{}"
+            try:
+                args = json.dumps(json.loads(raw_args), sort_keys=True)
+            except ValueError:
+                args = raw_args
+            key.append((str(tc["id"]), str(function.get("name") or ""), args))
+        return tuple(key)
+
+    def _remember_assistant(self, openai_resp: dict[str, Any]) -> None:
+        choice = (openai_resp.get("choices") or [{}])[0]
+        message = choice.get("message") or {}
+        calls = self._call_key(message)
+        if calls:
+            self._assistant_by_calls[calls] = message
+
+    def _replay_remembered_assistants(self, openai_body: dict[str, Any]) -> None:
+        messages = openai_body.get("messages")
+        if not isinstance(messages, list):
+            return
+        for i, msg in enumerate(messages):
+            if not isinstance(msg, dict) or msg.get("role") != "assistant":
+                continue
+            remembered = self._assistant_by_calls.get(self._call_key(msg))
+            if remembered is not None:
+                messages[i] = remembered
+
+    @on("/v1/messages/count_tokens")
+    async def count_tokens(self, request: Request) -> ClientResponse:
+        return ClientResponse.json(
+            {"input_tokens": count_anthropic_tokens(request.body).input_tokens}
+        )
+
+    async def aclose(self) -> None:
+        """Close the downstream if it is closeable; a bare function is a no-op.
+
+        Lets `Proxy.stop` reach the transport's HTTP pool through this
+        converter — otherwise `Proxy(AnthropicToOpenAI(fwd.handler()))` leaks
+        the forwarder's pool past the proxy lifecycle."""
+        if isinstance(self._downstream, _AsyncCloseable):
+            await self._downstream.aclose()
+
+    def environ(self, handle: TunnelHandle) -> dict[str, str]:
+        """Anthropic env-var bundle — from the agent's POV the wire is Anthropic,
+        regardless of the OpenAI downstream."""
+        return {
+            "ANTHROPIC_BASE_URL": handle.url,
+            "ANTHROPIC_API_KEY": PLACEHOLDER_API_KEY,
+        }
+
+
+__all__ = ["AnthropicToOpenAI"]

plugins/abridge/agentix/bridge/clients/openai.py

@@ -7,9 +7,10 @@
 at our tunnel.
 
 The SDK accepts the OpenAI Chat Completions request shape via typed
-kwargs. Agents that send non-standard fields the SDK doesn't accept
-will see an `UpstreamError`; for arbitrary-shape forwarding write your
-own `@on("/v1/chat/completions")` handler with raw httpx.
+kwargs. Upstream failures surface as `AbridgeError` carrying the
+upstream HTTP status (so the agent sees the real 429/400/... instead of
+a blanket 502); for arbitrary-shape forwarding write your own
+`@on("/v1/chat/completions")` handler with raw httpx.
 """
 
 from __future__ import annotations
@@ -20,7 +21,7 @@
 
 from agentix.utils import trace
 
-from ..proxy import AbridgeError, ClientResponse, Request, on
+from ..proxy import AbridgeError, ClientResponse, Request, TunnelHandle, on
 from ._genai_span import populate_openai_span
 
 if TYPE_CHECKING:
@@ -113,5 +114,17 @@ async def chat(self, request: Request) -> ClientResponse:
             populate_openai_span(request=request.body, response=response_dict)
             return ClientResponse.json(response_dict)
 
+    def environ(self, handle: TunnelHandle) -> dict[str, str]:
+        """Env-var bundle an in-sandbox OpenAI SDK needs to route through `handle`,
+        mirroring `AnthropicClient.environ`. `OPENAI_BASE_URL` carries the `/v1`
+        suffix the OpenAI SDK expects (it appends only `/chat/completions`), so the
+        load-bearing suffix is baked in here instead of left to the caller.
+        `OPENAI_API_KEY` is a non-secret placeholder whose shape passes the SDK's
+        local format check — the real upstream key lives on the host (this client)."""
+        return {
+            "OPENAI_BASE_URL": handle.url + "/v1",
+            "OPENAI_API_KEY": PLACEHOLDER_API_KEY,
+        }
+
 
 __all__ = ["PLACEHOLDER_API_KEY", "OpenAIClient"]