diff --git a/plugins/abridge/ARCHITECTURE.md b/plugins/abridge/ARCHITECTURE.md index 7e5161a..634d66a 100644 --- a/plugins/abridge/ARCHITECTURE.md +++ b/plugins/abridge/ARCHITECTURE.md @@ -62,17 +62,21 @@ Solid = code present in this branch. Dashed = planned (see below). - **`Sidecar(command=..., health_path=...)`** — owns a local sidecar process's lifecycle (spawn → health → URL → teardown). abridge-managed by default; pass an external URL straight to `Forward` to opt out. +- **`cc_convert_sidecar(...)`** *(experimental)* — configures an externally + installed `cc_convert_sidecar` executable through its environment contract. + abridge does not install, build, or publish that executable. ## Status -- **Implemented in this branch:** `Forward` and generic `Sidecar` process - supervision. Tests cover a local HTTP sidecar, the complete in-process - tunnel/namespace/Proxy/Forward path, process cleanup, noisy output, and - auto-port bind retries. +- **Implemented in this branch:** `Forward`, generic `Sidecar` process + supervision, and an experimental `cc_convert_sidecar` configuration preset. + Required tests cover the generic path and preset environment wiring. An + optional binary-backed test exercises translated JSON and a completed SSE + payload only when a caller supplies the external executable. - **Not implemented yet:** incremental SSE delivery. `Forward` buffers the complete sidecar response before the tunnel sends it to the agent. -- **Planned:** pinned adapters for specific gateway binaries, a deployed - runtime/SIO/sidecar integration check, the `tito` pretokenize/record - sidecar, a first-class `Session`/`Trajectory` model bridged onto `/trace`, +- **Planned:** a reproducibly pinned `cc_convert_sidecar` binary and required + full tunnel/SIO/sidecar integration check; the `tito` pretokenize/record + sidecar; a first-class `Session`/`Trajectory` model bridged onto `/trace`; and an open/chunk/end streaming primitive. Existing in-process clients remain supported until a separate migration removes them. diff --git a/plugins/abridge/README.md b/plugins/abridge/README.md index 795aa24..c03f082 100644 --- a/plugins/abridge/README.md +++ b/plugins/abridge/README.md @@ -205,6 +205,7 @@ agentix/bridge/ ├── proxy.py # Proxy + @on + sandbox tunnel + wire types ├── forward.py # JSON POST forwarding to a host-side service ├── sidecar.py # local process lifecycle + health supervision +├── sidecars.py # presets for external sidecar binaries └── clients/ # bundled handler implementations ├── openai.py # OpenAIClient (openai SDK) + PLACEHOLDER_API_KEY ├── anthropic.py # AnthropicClient (anthropic SDK) + environ() + PLACEHOLDER_API_KEY diff --git a/plugins/abridge/agentix/bridge/sidecars.py b/plugins/abridge/agentix/bridge/sidecars.py new file mode 100644 index 0000000..9be22af --- /dev/null +++ b/plugins/abridge/agentix/bridge/sidecars.py @@ -0,0 +1,65 @@ +"""Experimental presets for externally installed gateway binaries. + +These are thin convenience builders: translation and pretokenization live +inside the external sidecar process while abridge forwards decoded JSON +objects without understanding their provider-specific schema. +Pair the returned `Sidecar` with a `Forward` pointed at its URL: + + async with cc_convert_sidecar(binary="cc_convert_sidecar", + upstream_url="https://api.openai.com/v1/chat/completions", + upstream_key="sk-...") as url: + proxy = Proxy(Forward(url, paths=["/v1/messages"])) + async with proxy.session(sandbox) as handle: + await sandbox.remote(agent, base_url=handle.url) + +The `cc_convert_sidecar` preset only supplies the executable's expected +environment contract. abridge does not install, build, publish, or otherwise +verify that external executable. If it produces SSE, the current abridge +`Forward` path buffers the complete payload before returning it to the +sandbox. +""" + +from __future__ import annotations + +from .sidecar import Sidecar + + +def cc_convert_sidecar( + *, + upstream_url: str, + upstream_key: str | None = None, + binary: str = "cc_convert_sidecar", + host: str = "127.0.0.1", + port: int = 0, + litellm_compat: bool = False, + ready_timeout: float = 30.0, +) -> Sidecar: + """Build a `Sidecar` running the cc_convert translation binary. + + `upstream_url` is the OpenAI-compatible `/v1/chat/completions` URL the + sidecar forwards to; `upstream_key` is the bearer token it sends (kept + on the host — never in the sandbox). `litellm_compat` switches the + external executable's LiteLLM compatibility mode. + """ + env = { + "CC_CONVERT_UPSTREAM_URL": upstream_url, + } + if upstream_key: + env["CC_CONVERT_UPSTREAM_API_KEY"] = upstream_key + if litellm_compat: + env["CC_CONVERT_LITELLM_COMPAT"] = "1" + + def sidecar_env(bound_host: str, bound_port: int) -> dict[str, str]: + return {**env, "CC_CONVERT_LISTEN_ADDR": f"{bound_host}:{bound_port}"} + + return Sidecar( + command=[binary], + host=host, + port=port, + env=sidecar_env, + health_path="/healthz", + ready_timeout=ready_timeout, + ) + + +__all__ = ["cc_convert_sidecar"] diff --git a/plugins/abridge/tests/test_cc_convert_sidecar.py b/plugins/abridge/tests/test_cc_convert_sidecar.py new file mode 100644 index 0000000..2525ea5 --- /dev/null +++ b/plugins/abridge/tests/test_cc_convert_sidecar.py @@ -0,0 +1,185 @@ +"""Compatibility tests for the external ``cc_convert_sidecar`` preset. + +The required test suite uses a small executable probe to verify the preset's +process and environment wiring without downloading an unversioned external +executable. The Anthropic↔OpenAI compatibility test uses that executable +when a caller supplies one through ``CC_CONVERT_SIDECAR_BIN`` (or ``PATH``). +It remains optional until that binary has a published source and pinned +revision that CI can build reproducibly. +""" + +from __future__ import annotations + +import json +import os +import shutil +import sys + +import httpx +import pytest +from agentix.bridge import Forward, Request, Sidecar +from agentix.bridge.sidecars import cc_convert_sidecar + +BIN = os.environ.get("CC_CONVERT_SIDECAR_BIN") or shutil.which("cc_convert_sidecar") + +REQUIRES_REAL_CC_CONVERT = pytest.mark.skipif( + not BIN, + reason="external cc_convert_sidecar binary not available (set CC_CONVERT_SIDECAR_BIN)", +) + +PRESET_PROBE = r""" +import json +import os +from http.server import BaseHTTPRequestHandler, HTTPServer + +CONFIG = { + "listen_addr": os.environ["CC_CONVERT_LISTEN_ADDR"], + "upstream_url": os.environ["CC_CONVERT_UPSTREAM_URL"], + "upstream_key": os.environ.get("CC_CONVERT_UPSTREAM_API_KEY"), + "litellm_compat": os.environ.get("CC_CONVERT_LITELLM_COMPAT"), +} + +class H(BaseHTTPRequestHandler): + def do_GET(self): + payload = json.dumps(CONFIG).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *a): + pass + +host, port = CONFIG["listen_addr"].rsplit(":", 1) +HTTPServer((host, int(port)), H).serve_forever() +""" + +# Mock OpenAI Chat Completions upstream: 200 on GET (health), an OpenAI +# completion on POST — streamed SSE when the request asks for it. +MOCK_OPENAI = r""" +import sys, json +from http.server import BaseHTTPRequestHandler, HTTPServer + +NONSTREAM = { + "id": "chatcmpl-mock", "object": "chat.completion", "created": 0, "model": "mock", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": "hello from upstream"}}], + "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}, +} +CHUNKS = [ + {"id": "chatcmpl-mock", "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hello"}, "finish_reason": None}]}, + {"id": "chatcmpl-mock", "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {"content": " world"}, "finish_reason": None}]}, + {"id": "chatcmpl-mock", "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, +] + +class H(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200); self.end_headers(); self.wfile.write(b"ok") + + def do_POST(self): + n = int(self.headers.get("content-length", 0)) + body = json.loads(self.rfile.read(n) or b"{}") + if body.get("stream"): + self.send_response(200) + self.send_header("content-type", "text/event-stream") + self.end_headers() + for c in CHUNKS: + self.wfile.write(b"data: " + json.dumps(c).encode() + b"\n\n"); self.wfile.flush() + self.wfile.write(b"data: [DONE]\n\n"); self.wfile.flush() + else: + payload = json.dumps(NONSTREAM).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *a): + pass + +HTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever() +""" + + +async def test_cc_convert_sidecar_preset_wires_process_environment(tmp_path) -> None: + """The preset must pass the dynamically selected port and credentials.""" + probe = tmp_path / "cc_convert_sidecar_probe" + probe.write_text(f"#!{sys.executable}\n{PRESET_PROBE}") + probe.chmod(0o755) + + async with cc_convert_sidecar( + binary=str(probe), + upstream_url="https://openai.example/v1/chat/completions", + upstream_key="test-key", + litellm_compat=True, + ready_timeout=5.0, + ) as side_url: + async with httpx.AsyncClient() as client: + response = await client.get(side_url + "/config") + response.raise_for_status() + config = response.json() + + assert config == { + "listen_addr": side_url.removeprefix("http://"), + "upstream_url": "https://openai.example/v1/chat/completions", + "upstream_key": "test-key", + "litellm_compat": "1", + } + + +@REQUIRES_REAL_CC_CONVERT +async def test_anthropic_through_cc_convert_sidecar(tmp_path) -> None: + assert BIN is not None + mock = tmp_path / "mock_openai.py" + mock.write_text(MOCK_OPENAI) + + async with Sidecar(command=[sys.executable, str(mock), "{port}"]) as mock_url: + async with cc_convert_sidecar( + binary=BIN, + upstream_url=mock_url + "/v1/chat/completions", + ) as side_url: + fwd = Forward(side_url, paths=["/v1/messages"]) + handler = fwd.abridge_routes()["/v1/messages"] + try: + # ── non-streaming: OpenAI completion → Anthropic message ── + resp = await handler( + Request( + "/v1/messages", + { + "model": "claude-3-haiku", + "max_tokens": 50, + "messages": [{"role": "user", "content": "hi"}], + }, + ) + ) + assert resp.media_type == "application/json" + body = json.loads(resp.body) + assert body["type"] == "message" + assert body["role"] == "assistant" + assert body["content"][0]["type"] == "text" + assert body["content"][0]["text"] == "hello from upstream" + assert body["stop_reason"] == "end_turn" + assert body["usage"]["input_tokens"] == 5 + assert body["usage"]["output_tokens"] == 3 + + # ── buffered SSE compatibility ── + sresp = await handler( + Request( + "/v1/messages", + { + "model": "claude-3-haiku", + "max_tokens": 50, + "stream": True, + "messages": [{"role": "user", "content": "hi"}], + }, + ) + ) + assert sresp.media_type == "text/event-stream" + assert b"event: message_start" in sresp.body + assert b"event: message_stop" in sresp.body + assert b"hello" in sresp.body + finally: + await fwd.aclose() diff --git a/plugins/abridge/tests/test_sidecar_reliability.py b/plugins/abridge/tests/test_sidecar_reliability.py index c3a46eb..f45b3ee 100644 --- a/plugins/abridge/tests/test_sidecar_reliability.py +++ b/plugins/abridge/tests/test_sidecar_reliability.py @@ -11,6 +11,7 @@ import httpx import pytest from agentix.bridge import Sidecar, SidecarError +from agentix.bridge.sidecars import cc_convert_sidecar NOISY_SERVER = r""" import os @@ -71,6 +72,7 @@ def log_message(self, *args): HTTPServer((host, int(raw_port)), H).serve_forever() """ +PRESET_SERVER = ENV_FACTORY_SERVER.replace("SIDECAR_LISTEN_ADDR", "CC_CONVERT_LISTEN_ADDR") def _available_port() -> int: @@ -208,6 +210,23 @@ def env(host: str, port: int) -> dict[str, str]: assert (await client.get(url + "/healthz")).status_code == 200 +async def test_cc_convert_preset_allocates_port_and_env_on_entry(tmp_path: Path) -> None: + binary = tmp_path / "fake_cc_convert_sidecar" + binary.write_text(PRESET_SERVER) + binary.chmod(0o755) + sidecar = cc_convert_sidecar( + binary=str(binary), + upstream_url="http://upstream.invalid/v1/chat/completions", + ready_timeout=5.0, + ) + + assert sidecar.url == "http://127.0.0.1:0" + async with sidecar as url: + assert not url.endswith(":0") + async with httpx.AsyncClient(timeout=2.0) as client: + assert (await client.get(url + "/healthz")).status_code == 200 + + async def test_teardown_cancels_and_awaits_stuck_drain_tasks(monkeypatch: pytest.MonkeyPatch) -> None: sidecar = Sidecar(command=[sys.executable, "-c", "pass"]) stuck = asyncio.create_task(asyncio.Event().wait())