From 04899806310788bdbeb57d28fe9197320a7ad5da Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Tue, 8 Sep 2026 15:23:37 -0700 Subject: [PATCH 1/3] Added converse-stream functionality for bedrock --- canyonos_core/llm_proxy/README.md | 7 +- canyonos_core/llm_proxy/core.py | 17 ++- canyonos_core/llm_proxy/hooks.py | 44 ++++-- canyonos_core/llm_proxy/providers/base.py | 9 +- canyonos_core/llm_proxy/providers/bedrock.py | 111 ++++++++++++++- canyonos_core/llm_proxy/stub.py | 34 ++++- tests/test_llm_proxy_bedrock_streaming.py | 142 +++++++++++++++++++ tests/test_llm_proxy_streaming_e2e.py | 60 ++++++++ 8 files changed, 397 insertions(+), 27 deletions(-) create mode 100644 tests/test_llm_proxy_bedrock_streaming.py create mode 100644 tests/test_llm_proxy_streaming_e2e.py diff --git a/canyonos_core/llm_proxy/README.md b/canyonos_core/llm_proxy/README.md index 8de3d8e..1677549 100644 --- a/canyonos_core/llm_proxy/README.md +++ b/canyonos_core/llm_proxy/README.md @@ -21,7 +21,8 @@ your app (unchanged) localhost:8080 real upstream - **OpenAI / Anthropic** — straight HTTP reverse-proxy: rewrite host, swap in the real key, forward with `requests`, return the response. - **Bedrock** — re-issued through the proxy's own `boto3` client (handles SigV4 - signing + URL-encoding correctly). Only `invoke` is wired up. + signing + URL-encoding correctly). `invoke`, `converse`, and `converse-stream` + are wired up; `invoke-with-response-stream` is not. ## Run @@ -104,7 +105,9 @@ automatically inject headers or write telemetry. ## Limitations -- **No streaming.** `stream=True` / `invoke-with-response-stream` are not handled. +- **Bedrock `converse-stream` only.** OpenAI/Anthropic `stream=True` and + Bedrock `invoke-with-response-stream` are still not handled — both remain + fully buffered / unimplemented, respectively. - **Bedrock error bodies are reconstructed**, not passed through byte-for-byte (boto3 raises on 4xx/5xx; we rebuild a JSON body with the real status + message). OpenAI/Anthropic errors pass through unchanged. diff --git a/canyonos_core/llm_proxy/core.py b/canyonos_core/llm_proxy/core.py index 38558af..3795b0a 100644 --- a/canyonos_core/llm_proxy/core.py +++ b/canyonos_core/llm_proxy/core.py @@ -6,7 +6,7 @@ import time from typing import Optional -from flask import Response +from flask import Response, stream_with_context from canyonos_core.llm_proxy.hooks import Ctx @@ -53,5 +53,20 @@ def proxy_request(provider, subpath, flask_request): else: pr = provider.forward(flask_request, subpath, body) + if pr.stream is not None: + # Streamed responses: relay chunks as they arrive, and only take telemetry when the whole response is done. + def relay(): + try: + yield from pr.stream + finally: + hooks.on_response(ctx, pr) + + return Response( + stream_with_context(relay()), + status=pr.status, + headers=pr.headers, + direct_passthrough=True, + ) + hooks.on_response(ctx, pr) return Response(pr.content, status=pr.status, headers=pr.headers) diff --git a/canyonos_core/llm_proxy/hooks.py b/canyonos_core/llm_proxy/hooks.py index 89182a5..5a655fe 100644 --- a/canyonos_core/llm_proxy/hooks.py +++ b/canyonos_core/llm_proxy/hooks.py @@ -76,15 +76,23 @@ def on_request(self, ctx: Ctx) -> None: ) def on_response(self, ctx: Ctx, resp: Any) -> None: - # Extract tokens for Bedrock + # Extract tokens for Bedrock; streamed calls carry usage on resp.stream_usage instead of the JSON body. usage = None + is_stream = getattr(resp, "stream", None) is not None if ctx.provider == "bedrock": - usage = self._extract_bedrock_tokens(resp) - + if is_stream: + usage = self._usage_from_dict(getattr(resp, "stream_usage", None)) + else: + usage = self._extract_bedrock_tokens(resp) + + status = getattr(resp, "status", "?") + if is_stream and getattr(resp, "stream_error", False): + status = f"{status} (stream-error)" + log.info( "← %s %s /%s -> %s in %.0fms | %s", ctx.provider, ctx.method, ctx.subpath, - getattr(resp, "status", "?"), ctx.elapsed_ms(), + status, ctx.elapsed_ms(), usage or "no usage" ) @@ -98,8 +106,10 @@ def on_response(self, ctx: Ctx, resp: Any) -> None: # Extract model ID model_id = self._extract_model_id(ctx) - is_error = resp.status >= 400 - + is_error = resp.status >= 400 or ( + is_stream and getattr(resp, "stream_error", False) + ) + # Build telemetry data data = { "model": model_id, @@ -135,6 +145,18 @@ def _extract_model_id(self, ctx: Ctx) -> str: return "unknown" + @staticmethod + def _usage_from_dict(usage: Optional[Dict[str, Any]]) -> Optional[TokenUsage]: + if not usage: + return None + return TokenUsage( + input_tokens=usage.get("inputTokens", 0), + output_tokens=usage.get("outputTokens", 0), + total_tokens=usage.get("totalTokens", 0), + input_cache_tokens=usage.get("cacheReadInputTokens", 0), + input_cache_write_tokens=usage.get("cacheCreationInputTokens", 0), + ) + def _extract_bedrock_tokens(self, resp: Any) -> Optional[TokenUsage]: """Extract token usage from Bedrock response. It requires diff logic from OpenAI/Anthropic""" if resp.status != 200: @@ -142,15 +164,7 @@ def _extract_bedrock_tokens(self, resp: Any) -> Optional[TokenUsage]: try: data = json.loads(resp.content.decode("utf-8")) - usage = data.get("usage", {}) - if usage: - return TokenUsage( - input_tokens=usage.get("inputTokens", 0), - output_tokens=usage.get("outputTokens", 0), - total_tokens=usage.get("totalTokens", 0), - input_cache_tokens=usage.get("cacheReadInputTokens", 0), - input_cache_write_tokens=usage.get("cacheCreationInputTokens", 0), - ) + return self._usage_from_dict(data.get("usage")) except: pass return None diff --git a/canyonos_core/llm_proxy/providers/base.py b/canyonos_core/llm_proxy/providers/base.py index ef5db41..76a2fe5 100644 --- a/canyonos_core/llm_proxy/providers/base.py +++ b/canyonos_core/llm_proxy/providers/base.py @@ -10,7 +10,7 @@ import json from dataclasses import dataclass, field -from typing import Dict, Iterable, List, Tuple +from typing import Dict, Iterable, Iterator, List, Optional, Tuple import requests @@ -43,7 +43,12 @@ class UpstreamRequest: class ProxyResponse: status: int headers: List[Tuple[str, str]] - content: bytes + content: bytes = b"" + # Set instead of ``content`` for streamed responses; core.proxy_request streams these chunks through directly. + stream: Optional[Iterator[bytes]] = None + # Filled in by the stream generator once usage is known (only available after the trailing "metadata" event). + stream_usage: Optional[dict] = None + stream_error: bool = False def json(self): return json.loads(self.content.decode("utf-8")) diff --git a/canyonos_core/llm_proxy/providers/bedrock.py b/canyonos_core/llm_proxy/providers/bedrock.py index a3f3fb6..43e9ec4 100644 --- a/canyonos_core/llm_proxy/providers/bedrock.py +++ b/canyonos_core/llm_proxy/providers/bedrock.py @@ -1,25 +1,83 @@ """Bedrock adapter. +TLDR: User code calls boto3 which requires certain format, but sends requests to llm-proxy, which has its own boto3 that makes/recieves requests. But being a middleman, we need to decrypt the messages to get contents, and then re-encrypt so the users boto3 call receives the correct format. + Rather than re-sign the caller's SigV4 request (fiddly once model IDs contain ``:`` and ``/``), we re-issue the call through the proxy's own boto3 client, -which handles signing and URL-encoding correctly by construction. This is clean -for request/response; streaming (``invoke-with-response-stream``) is out of scope -for now. +which handles signing and URL-encoding correctly by construction. + +``converse-stream`` is supported: boto3's ``converse_stream`` already decodes +the upstream AWS event-stream response into plain dicts, so we re-encode those +back into the same ``application/vnd.amazon.eventstream`` wire format so the +caller's own boto3 client (pointed at us via +``AWS_ENDPOINT_URL_BEDROCK_RUNTIME``) can decode it exactly as if it had hit +Bedrock directly. ``invoke-with-response-stream`` (raw per-model streaming, as +opposed to the unified Converse API) remains out of scope. """ from __future__ import annotations import json +import struct +import zlib import boto3 from botocore.exceptions import ClientError from canyonos_core.llm_proxy.providers.base import Provider, ProxyResponse -# bedrock-runtime operations that can appear as the last path segment; only the -# non-streaming "invoke" is wired up for now. +# bedrock-runtime operations that can appear as the last path segment ("invoke-with-response-stream" remains out of scope). _SUPPORTED_OPS = {"invoke", "invoke-with-response-stream", "converse", "converse-stream"} +# Header value type ID for "string" from the AWS event-stream binary format spec (the only type Bedrock's headers use). +_HEADER_TYPE_STRING = 7 + + +def _encode_event_headers(headers: dict) -> bytes: + """Pack event-stream headers: [1B name len][name][1B type][2B value + len][value], repeated. Mirrors what botocore.eventstream decodes.""" + buf = bytearray() + for name, value in headers.items(): + name_bytes = name.encode("utf-8") + value_bytes = value.encode("utf-8") + buf.append(len(name_bytes)) + buf.extend(name_bytes) + buf.append(_HEADER_TYPE_STRING) + buf.extend(struct.pack(">H", len(value_bytes))) + buf.extend(value_bytes) + return bytes(buf) + + +def _encode_event(headers: dict, payload: bytes) -> bytes: + """Encode one AWS event-stream frame (botocore only decodes this format, never encodes it).""" + header_bytes = _encode_event_headers(headers) + total_length = 8 + 4 + len(header_bytes) + len(payload) + 4 + prelude = struct.pack(">II", total_length, len(header_bytes)) + prelude_crc = struct.pack(">I", zlib.crc32(prelude) & 0xFFFFFFFF) + message = prelude + prelude_crc + header_bytes + payload + message_crc = struct.pack(">I", zlib.crc32(message) & 0xFFFFFFFF) + return message + message_crc + + +def _event_frame(event_type: str, body: dict) -> bytes: + """Encode a normal Bedrock ConverseStream event (e.g. messageStart, contentBlockDelta) as a frame.""" + headers = { + ":event-type": event_type, + ":content-type": "application/json", + ":message-type": "event", + } + return _encode_event(headers, json.dumps(body).encode("utf-8")) + + +def _exception_frame(exception_type: str, message: str) -> bytes: + """Encode a mid-stream error as a Bedrock ConverseStream exception frame.""" + headers = { + ":exception-type": exception_type, + ":content-type": "application/json", + ":message-type": "exception", + } + return _encode_event(headers, json.dumps({"message": message}).encode("utf-8")) + class BedrockProvider(Provider): name = "bedrock" @@ -74,9 +132,21 @@ def forward(self, req, subpath, body): headers=[("Content-Type", "application/json")], content=payload ) + + elif op == "converse-stream": + params = json.loads(body) + params["modelId"] = model_id + resp = self._client.converse_stream(**params) + + pr = ProxyResponse( + status=resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 200), + headers=[("Content-Type", "application/vnd.amazon.eventstream")], + ) + pr.stream = self._encode_converse_stream(resp["stream"], pr) + return pr else: raise NotImplementedError( - f"bedrock op '{op}' not supported (only invoke and converse)" + f"bedrock op '{op}' not supported (only invoke, converse, and converse-stream)" ) except ClientError as exc: @@ -90,6 +160,35 @@ def forward(self, req, subpath, body): + @staticmethod + def _encode_converse_stream(events, pr: ProxyResponse): + """Re-frame boto3's already-decoded ConverseStream events + (``{"messageStart": {...}}``, ``{"contentBlockDelta": {...}}``, ..., + finally ``{"metadata": {"usage": {...}}}``) back into the AWS + event-stream wire format the caller's own boto3 client expects. + + Also captures usage off the trailing "metadata" event onto ``pr`` (read + by hooks.on_response only after this generator is exhausted, since + usage isn't known until then) and turns any mid-stream failure into a + single exception frame instead of dropping the connection. + """ + try: + for event in events: + event_type, event_body = next(iter(event.items())) + if event_type == "metadata": + pr.stream_usage = event_body.get("usage") + yield _event_frame(event_type, event_body) + except ClientError as exc: + pr.stream_error = True + err = exc.response.get("Error", {}) + yield _exception_frame( + err.get("Code", "InternalServerException"), + err.get("Message", str(exc)), + ) + except Exception as exc: # noqa: BLE001 - surface any mid-stream failure as an exception frame instead of truncating silently + pr.stream_error = True + yield _exception_frame(type(exc).__name__, str(exc)) + @staticmethod def _parse(subpath): # subpath looks like "model//"; the modelId may itself diff --git a/canyonos_core/llm_proxy/stub.py b/canyonos_core/llm_proxy/stub.py index c919e1e..f656bd0 100644 --- a/canyonos_core/llm_proxy/stub.py +++ b/canyonos_core/llm_proxy/stub.py @@ -36,11 +36,43 @@ def _json_response(obj, status=200): ) +def _bedrock_converse_stream_response(text): + """A minimal, validly-framed ConverseStream event sequence so + ``CANYONOS_LLM_STUB_TEXT`` exercises the exact same wire format + (``application/vnd.amazon.eventstream``) a real Bedrock call would, + without needing AWS credentials.""" + # Imported lazily so non-Bedrock stubs don't need boto3/botocore. + from canyonos_core.llm_proxy.providers.bedrock import _event_frame + from canyonos_core.llm_proxy.providers.base import ProxyResponse + + events = [ + ("messageStart", {"role": "assistant"}), + ("contentBlockDelta", {"contentBlockIndex": 0, "delta": {"text": text}}), + ("contentBlockStop", {"contentBlockIndex": 0}), + ("messageStop", {"stopReason": "end_turn"}), + ("metadata", {"usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}}), + ] + + def gen(): + for event_type, body in events: + yield _event_frame(event_type, body) + + pr = ProxyResponse( + status=200, + headers=[("Content-Type", "application/vnd.amazon.eventstream")], + ) + pr.stream = gen() + pr.stream_usage = {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2} + return pr + + def build_stub(provider_name, subpath, text): """Build a provider-appropriate canned response carrying ``text``.""" if provider_name == "bedrock": op = subpath.rsplit("/", 1)[-1] if subpath else "" - if op in ("converse", "converse-stream"): + if op == "converse-stream": + return _bedrock_converse_stream_response(text) + if op == "converse": return _json_response({ "output": {"message": {"role": "assistant", "content": [{"text": text}]}}, diff --git a/tests/test_llm_proxy_bedrock_streaming.py b/tests/test_llm_proxy_bedrock_streaming.py new file mode 100644 index 0000000..86f9c9b --- /dev/null +++ b/tests/test_llm_proxy_bedrock_streaming.py @@ -0,0 +1,142 @@ +import json +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from botocore.eventstream import EventStreamBuffer + +from canyonos_core.llm_proxy.providers import bedrock as bedrock_module +from canyonos_core.llm_proxy.providers.bedrock import ( + BedrockProvider, + _encode_event, + _event_frame, + _exception_frame, +) + + +def _decode_frames(raw: bytes): + """Feed raw bytes through botocore's real decoder and return the parsed + (headers, payload) pairs -- proves our encoder is wire-compatible with + the same parser a caller's boto3 client would use.""" + buf = EventStreamBuffer() + buf.add_data(raw) + return [(msg.headers, msg.payload) for msg in buf] + + +class EncodeEventTests(unittest.TestCase): + def test_round_trips_through_botocore_decoder(self): + headers = { + ":event-type": "contentBlockDelta", + ":content-type": "application/json", + ":message-type": "event", + } + payload = json.dumps({"delta": {"text": "hi"}}).encode("utf-8") + raw = _encode_event(headers, payload) + + [(decoded_headers, decoded_payload)] = _decode_frames(raw) + self.assertEqual(decoded_headers, headers) + self.assertEqual(decoded_payload, payload) + + def test_event_frame_helper(self): + raw = _event_frame("messageStart", {"role": "assistant"}) + [(headers, payload)] = _decode_frames(raw) + self.assertEqual(headers[":event-type"], "messageStart") + self.assertEqual(headers[":message-type"], "event") + self.assertEqual(json.loads(payload), {"role": "assistant"}) + + def test_exception_frame_helper(self): + raw = _exception_frame("ThrottlingException", "slow down") + [(headers, payload)] = _decode_frames(raw) + self.assertEqual(headers[":exception-type"], "ThrottlingException") + self.assertEqual(headers[":message-type"], "exception") + self.assertEqual(json.loads(payload), {"message": "slow down"}) + + def test_multiple_events_concatenate_and_decode_in_order(self): + raw = ( + _event_frame("messageStart", {"role": "assistant"}) + + _event_frame("contentBlockDelta", {"delta": {"text": "hi"}}) + + _event_frame("messageStop", {"stopReason": "end_turn"}) + ) + decoded = _decode_frames(raw) + self.assertEqual([h[":event-type"] for h, _ in decoded], + ["messageStart", "contentBlockDelta", "messageStop"]) + + +class _FakeCfg: + bedrock_region = "us-east-1" + bedrock_upstream_host = "bedrock-runtime.us-east-1.amazonaws.com" + + +class BedrockProviderConverseStreamTests(unittest.TestCase): + def _make_provider(self): + with patch.object(bedrock_module.boto3, "client") as mock_client_factory: + self.mock_client = MagicMock() + mock_client_factory.return_value = self.mock_client + return BedrockProvider(_FakeCfg()) + + def test_converse_stream_encodes_events_and_captures_usage(self): + provider = self._make_provider() + events = [ + {"messageStart": {"role": "assistant"}}, + {"contentBlockDelta": {"delta": {"text": "hi"}}}, + {"messageStop": {"stopReason": "end_turn"}}, + {"metadata": {"usage": {"inputTokens": 3, "outputTokens": 5, + "totalTokens": 8}}}, + ] + self.mock_client.converse_stream.return_value = { + "ResponseMetadata": {"HTTPStatusCode": 200}, + "stream": iter(events), + } + + req = MagicMock() + body = json.dumps({"messages": [{"role": "user", "content": [{"text": "hi"}]}]}).encode() + pr = provider.forward(req, "model/anthropic.claude-3/converse-stream", body) + + self.assertIsNotNone(pr.stream) + self.assertEqual(pr.headers, [("Content-Type", "application/vnd.amazon.eventstream")]) + + raw = b"".join(pr.stream) + decoded = _decode_frames(raw) + self.assertEqual( + [h[":event-type"] for h, _ in decoded], + ["messageStart", "contentBlockDelta", "messageStop", "metadata"], + ) + self.assertEqual(json.loads(decoded[1][1]), {"delta": {"text": "hi"}}) + + # Usage is only populated once the generator has actually been drained. + self.assertEqual(pr.stream_usage, {"inputTokens": 3, "outputTokens": 5, + "totalTokens": 8}) + self.assertFalse(pr.stream_error) + + self.mock_client.converse_stream.assert_called_once() + called_kwargs = self.mock_client.converse_stream.call_args.kwargs + self.assertEqual(called_kwargs["modelId"], "anthropic.claude-3") + + def test_mid_stream_error_yields_exception_frame_instead_of_raising(self): + provider = self._make_provider() + + def failing_events(): + yield {"messageStart": {"role": "assistant"}} + raise RuntimeError("boom") + + self.mock_client.converse_stream.return_value = { + "ResponseMetadata": {"HTTPStatusCode": 200}, + "stream": failing_events(), + } + + req = MagicMock() + body = json.dumps({"messages": []}).encode() + pr = provider.forward(req, "model/anthropic.claude-3/converse-stream", body) + + raw = b"".join(pr.stream) # must not raise + decoded = _decode_frames(raw) + self.assertEqual(decoded[0][0][":event-type"], "messageStart") + self.assertEqual(decoded[1][0][":message-type"], "exception") + self.assertTrue(pr.stream_error) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_llm_proxy_streaming_e2e.py b/tests/test_llm_proxy_streaming_e2e.py new file mode 100644 index 0000000..e8ed5bc --- /dev/null +++ b/tests/test_llm_proxy_streaming_e2e.py @@ -0,0 +1,60 @@ +"""End-to-end: Flask test client -> core.proxy_request -> stub converse-stream +response, decoded the same way a real boto3 caller would.""" + +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from botocore.eventstream import EventStreamBuffer + +from canyonos_core.llm_proxy.app import create_app +from canyonos_core.llm_proxy.config import Config, ProviderConfig + + +def _decode_frames(raw: bytes): + buf = EventStreamBuffer() + buf.add_data(raw) + return [(msg.headers, msg.payload) for msg in buf] + + +class ConverseStreamStubE2ETests(unittest.TestCase): + def setUp(self): + cfg = Config( + host="127.0.0.1", port=0, connect_timeout=1, read_timeout=1, + openai=ProviderConfig(upstream_base="https://api.openai.com"), + anthropic=ProviderConfig(upstream_base="https://api.anthropic.com"), + bedrock_region="us-east-1", + bedrock_upstream_host="bedrock-runtime.us-east-1.amazonaws.com", + redis_host="localhost", redis_port=6379, + ) + self.app = create_app(cfg) + self.client = self.app.test_client() + + @patch.dict(os.environ, {"CANYONOS_LLM_STUB_TEXT": "hello from stub"}) + def test_converse_stream_returns_valid_eventstream_body(self): + resp = self.client.post( + "/bedrock/model/anthropic.claude-3-5-sonnet-20240620-v1:0/converse-stream", + data=b'{"messages": [{"role": "user", "content": [{"text": "hi"}]}]}', + content_type="application/json", + ) + self.assertEqual(resp.status_code, 200) + self.assertEqual(resp.headers["Content-Type"], "application/vnd.amazon.eventstream") + + decoded = _decode_frames(resp.data) + event_types = [h[":event-type"] for h, _ in decoded] + self.assertEqual( + event_types, + ["messageStart", "contentBlockDelta", "contentBlockStop", + "messageStop", "metadata"], + ) + + import json + delta_payload = json.loads(decoded[1][1]) + self.assertEqual(delta_payload["delta"]["text"], "hello from stub") + + +if __name__ == "__main__": + unittest.main() From 9765affea9160128017e91ab816d7f6c6216be91 Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Tue, 8 Sep 2026 22:16:27 -0700 Subject: [PATCH 2/3] Added streaming to bedrock + more --- canyonos_core/llm_proxy/README.md | 48 +++++---- canyonos_core/llm_proxy/hooks.py | 104 +++++++++++++------ canyonos_core/llm_proxy/providers/bedrock.py | 75 ++++++++----- canyonos_core/llm_proxy/stub.py | 44 +++++--- tests/test_llm_proxy_bedrock_streaming.py | 43 +++++++- tests/test_llm_proxy_streaming_e2e.py | 19 ++++ tests/test_llm_proxy_usage_extraction.py | 78 ++++++++++++++ 7 files changed, 323 insertions(+), 88 deletions(-) create mode 100644 tests/test_llm_proxy_usage_extraction.py diff --git a/canyonos_core/llm_proxy/README.md b/canyonos_core/llm_proxy/README.md index 1677549..2784286 100644 --- a/canyonos_core/llm_proxy/README.md +++ b/canyonos_core/llm_proxy/README.md @@ -21,8 +21,8 @@ your app (unchanged) localhost:8080 real upstream - **OpenAI / Anthropic** — straight HTTP reverse-proxy: rewrite host, swap in the real key, forward with `requests`, return the response. - **Bedrock** — re-issued through the proxy's own `boto3` client (handles SigV4 - signing + URL-encoding correctly). `invoke`, `converse`, and `converse-stream` - are wired up; `invoke-with-response-stream` is not. + signing + URL-encoding correctly). `invoke`, `converse`, `converse-stream`, + and `invoke-with-response-stream` are all wired up. ## Run @@ -79,35 +79,47 @@ boto3.client("bedrock-runtime").invoke_model( ## Telemetry & Metrics -**Automatic telemetry is currently Bedrock-only.** The proxy captures: +The proxy captures, per response: - Model ID - Input/output/total token counts -- Cache tokens (read & write) +- Cache tokens (read & write, where the provider reports them) - Error status -Telemetry is automatically written to Redis under `future:` keys. +Telemetry is written to Redis under `future:` keys, keyed off an +`X-Canyonos-Future-ID` request header. -### How it works (Bedrock only) +### Usage extraction coverage (`hooks.py::Hooks._extract_usage`) -1. **Auto-injection:** boto3 hook (`proxy.py`) injects `X-Canyonos-Future-ID` header from thread-local context -2. **Token extraction:** `hooks.py` parses response `usage` field -3. **Redis write:** All metrics written to `future:` hash +| Provider / op | Usage schema used | Status | +| --- | --- | --- | +| Bedrock `converse` / `converse-stream` | Bedrock's own camelCase (`inputTokens`, ...) | works for any model | +| Bedrock `invoke` / `invoke-with-response-stream`, `anthropic.*` model | Anthropic's native (`input_tokens`, ...) | works | +| Bedrock `invoke` / `invoke-with-response-stream`, other model families | model-specific, unknown | no usage (schema not implemented yet) | +| Direct Anthropic API (`/anthropic/...`) | Anthropic's native | works | +| Direct OpenAI API (`/openai/...`) | OpenAI's native (`prompt_tokens`, ...) | works | -### Why Bedrock-only? +### How it works + +1. **Auto-injection:** boto3 hook (`proxy.py`) injects `X-Canyonos-Future-ID` header from thread-local context -- **Bedrock (boto3) only**; the OpenAI/Anthropic SDKs don't fire this hook, so callers using those SDKs directly won't get the header auto-attached. +2. **Usage extraction:** `hooks.py`'s `_extract_usage` parses the response `usage` field with the schema matching that provider/op/model (table above) -- this part works for all three providers whenever the header is present. +3. **Redis write:** All metrics written to `future:` hash. + +So in practice, automatic end-to-end telemetry (header injection + extraction) is Bedrock-only for now; OpenAI/Anthropic usage extraction works, but nothing auto-attaches `X-Canyonos-Future-ID` for those SDKs yet. + +### Why is header auto-injection Bedrock-only? OpenAI and Anthropic use their own Python SDKs (`openai`, `anthropic`), not boto3. -The boto3 event hook doesn't fire for non-AWS SDKs. To add telemetry for those: +The boto3 event hook doesn't fire for non-AWS SDKs. To add auto-injection for those: - Would need separate hooks in each SDK's HTTP client -- Or callers would need to use proxy directly (not through SDKs) - -The proxy *forwards* OpenAI/Anthropic requests and *can* extract tokens, but doesn't -automatically inject headers or write telemetry. +- Or callers would need to attach `X-Canyonos-Future-ID` themselves ## Limitations -- **Bedrock `converse-stream` only.** OpenAI/Anthropic `stream=True` and - Bedrock `invoke-with-response-stream` are still not handled — both remain - fully buffered / unimplemented, respectively. +- **Bedrock `converse-stream` and `invoke-with-response-stream` only.** + OpenAI/Anthropic `stream=True` is still not handled and remains fully + buffered. `invoke-with-response-stream` also has no usage/token telemetry + regardless of model (unlike `converse-stream`, it has no metadata event to + read usage from -- see the usage extraction coverage table above). - **Bedrock error bodies are reconstructed**, not passed through byte-for-byte (boto3 raises on 4xx/5xx; we rebuild a JSON body with the real status + message). OpenAI/Anthropic errors pass through unchanged. diff --git a/canyonos_core/llm_proxy/hooks.py b/canyonos_core/llm_proxy/hooks.py index 5a655fe..4b4932c 100644 --- a/canyonos_core/llm_proxy/hooks.py +++ b/canyonos_core/llm_proxy/hooks.py @@ -1,9 +1,10 @@ """The metrics seam. -Every proxied call passes through ``on_request`` / ``on_response``. Today these -only log. Token accounting lands here later: because the whole response is -buffered, usage extraction is a one-liner, e.g. ``resp.json().get("usage")`` for -OpenAI/Anthropic (Bedrock's usage lives in its per-model response body). +Every proxied call passes through ``on_request`` / ``on_response``, which logs +and (when Redis is configured) extracts token usage per provider/op/model -- +see ``Hooks._extract_usage``. Bedrock invoke's usage schema is only known for +anthropic.* models today; other model families and OpenAI/Anthropic streaming +remain unhandled. """ from __future__ import annotations @@ -76,14 +77,8 @@ def on_request(self, ctx: Ctx) -> None: ) def on_response(self, ctx: Ctx, resp: Any) -> None: - # Extract tokens for Bedrock; streamed calls carry usage on resp.stream_usage instead of the JSON body. - usage = None + usage = self._extract_usage(ctx, resp) is_stream = getattr(resp, "stream", None) is not None - if ctx.provider == "bedrock": - if is_stream: - usage = self._usage_from_dict(getattr(resp, "stream_usage", None)) - else: - usage = self._extract_bedrock_tokens(resp) status = getattr(resp, "status", "?") if is_stream and getattr(resp, "stream_error", False): @@ -135,18 +130,55 @@ def _extract_model_id(self, ctx: Ctx) -> str: """Extract model ID from context or subpath.""" if ctx.model: return ctx.model - - # For Bedrock: subpath is "model//operation" - # Use rpartition to peel operation off the right (same as provider logic) - if ctx.provider == "bedrock" and ctx.subpath.startswith("model/"): - model_id, sep, op = ctx.subpath[len("model/"):].rpartition("/") - if sep: # Found a separator + if ctx.provider == "bedrock": + model_id, _op = self._bedrock_model_and_op(ctx.subpath) + if model_id: return model_id - return "unknown" - + + @staticmethod + def _bedrock_model_and_op(subpath: str): + """Split a Bedrock subpath ("model//") into (model_id, op); (None, None) if unrecognized.""" + if not subpath.startswith("model/"): + return None, None + model_id, sep, op = subpath[len("model/"):].rpartition("/") + return (model_id, op) if sep else (None, None) + + def _extract_usage(self, ctx: Ctx, resp: Any) -> Optional[TokenUsage]: + """Dispatch to the right usage schema for this provider/op/model.""" + is_stream = getattr(resp, "stream", None) is not None + + if ctx.provider == "bedrock": + if is_stream: + return self._usage_from_dict(getattr(resp, "stream_usage", None)) + model_id, op = self._bedrock_model_and_op(ctx.subpath) + # invoke's body is model-native; only anthropic.*'s schema is known so far. + if op == "invoke" and (model_id or "").startswith("anthropic."): + return self._extract_json_usage(resp, self._usage_from_anthropic_dict) + return self._extract_json_usage(resp, self._usage_from_dict) + + if ctx.provider == "anthropic": + return self._extract_json_usage(resp, self._usage_from_anthropic_dict) + + if ctx.provider == "openai": + return self._extract_json_usage(resp, self._usage_from_openai_dict) + + return None + + @staticmethod + def _extract_json_usage(resp: Any, parser) -> Optional[TokenUsage]: + """Parse resp.content as JSON and hand its "usage" key to `parser`.""" + if getattr(resp, "status", None) != 200: + return None + try: + data = json.loads(resp.content.decode("utf-8")) + except Exception: + return None + return parser(data.get("usage")) + @staticmethod def _usage_from_dict(usage: Optional[Dict[str, Any]]) -> Optional[TokenUsage]: + """Bedrock Converse's usage schema (camelCase), used for both converse and converse-stream.""" if not usage: return None return TokenUsage( @@ -157,17 +189,31 @@ def _usage_from_dict(usage: Optional[Dict[str, Any]]) -> Optional[TokenUsage]: input_cache_write_tokens=usage.get("cacheCreationInputTokens", 0), ) - def _extract_bedrock_tokens(self, resp: Any) -> Optional[TokenUsage]: - """Extract token usage from Bedrock response. It requires diff logic from OpenAI/Anthropic""" - if resp.status != 200: + @staticmethod + def _usage_from_anthropic_dict(usage: Optional[Dict[str, Any]]) -> Optional[TokenUsage]: + """Anthropic's native usage schema (snake_case, no total field); shared by direct Anthropic API calls and Bedrock invoke for anthropic.* models, since Bedrock returns Anthropic's own response body unchanged for that op.""" + if not usage: return None - - try: - data = json.loads(resp.content.decode("utf-8")) - return self._usage_from_dict(data.get("usage")) - except: - pass - return None + input_tokens = usage.get("input_tokens", 0) + output_tokens = usage.get("output_tokens", 0) + return TokenUsage( + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=input_tokens + output_tokens, + input_cache_tokens=usage.get("cache_read_input_tokens", 0), + input_cache_write_tokens=usage.get("cache_creation_input_tokens", 0), + ) + + @staticmethod + def _usage_from_openai_dict(usage: Optional[Dict[str, Any]]) -> Optional[TokenUsage]: + """OpenAI's native usage schema.""" + if not usage: + return None + return TokenUsage( + input_tokens=usage.get("prompt_tokens", 0), + output_tokens=usage.get("completion_tokens", 0), + total_tokens=usage.get("total_tokens", 0), + ) hooks = Hooks() diff --git a/canyonos_core/llm_proxy/providers/bedrock.py b/canyonos_core/llm_proxy/providers/bedrock.py index 43e9ec4..e7c8b54 100644 --- a/canyonos_core/llm_proxy/providers/bedrock.py +++ b/canyonos_core/llm_proxy/providers/bedrock.py @@ -6,17 +6,17 @@ ``:`` and ``/``), we re-issue the call through the proxy's own boto3 client, which handles signing and URL-encoding correctly by construction. -``converse-stream`` is supported: boto3's ``converse_stream`` already decodes -the upstream AWS event-stream response into plain dicts, so we re-encode those -back into the same ``application/vnd.amazon.eventstream`` wire format so the -caller's own boto3 client (pointed at us via -``AWS_ENDPOINT_URL_BEDROCK_RUNTIME``) can decode it exactly as if it had hit -Bedrock directly. ``invoke-with-response-stream`` (raw per-model streaming, as -opposed to the unified Converse API) remains out of scope. +``converse-stream`` and ``invoke-with-response-stream`` are both supported: +boto3 already decodes the upstream AWS event-stream response into plain +dicts for either, so we re-encode those back into the same +``application/vnd.amazon.eventstream`` wire format so the caller's own boto3 +client (pointed at us via ``AWS_ENDPOINT_URL_BEDROCK_RUNTIME``) can decode it +exactly as if it had hit Bedrock directly. """ from __future__ import annotations +import base64 import json import struct import zlib @@ -26,7 +26,6 @@ from canyonos_core.llm_proxy.providers.base import Provider, ProxyResponse -# bedrock-runtime operations that can appear as the last path segment ("invoke-with-response-stream" remains out of scope). _SUPPORTED_OPS = {"invoke", "invoke-with-response-stream", "converse", "converse-stream"} # Header value type ID for "string" from the AWS event-stream binary format spec (the only type Bedrock's headers use). @@ -59,20 +58,29 @@ def _encode_event(headers: dict, payload: bytes) -> bytes: return message + message_crc +def _jsonify_blobs(body: dict) -> dict: + """Base64-encode any raw ``bytes`` values (e.g. InvokeModelWithResponseStream's chunk payload) so the body is JSON-serializable, matching how AWS's blob type is represented on the wire.""" + return { + k: base64.b64encode(v).decode("ascii") if isinstance(v, (bytes, bytearray)) else v + for k, v in body.items() + } + + def _event_frame(event_type: str, body: dict) -> bytes: - """Encode a normal Bedrock ConverseStream event (e.g. messageStart, contentBlockDelta) as a frame.""" + """Encode a normal Bedrock stream event (e.g. messageStart, contentBlockDelta, chunk) as a frame.""" headers = { ":event-type": event_type, ":content-type": "application/json", ":message-type": "event", } - return _encode_event(headers, json.dumps(body).encode("utf-8")) + return _encode_event(headers, json.dumps(_jsonify_blobs(body)).encode("utf-8")) -def _exception_frame(exception_type: str, message: str) -> bytes: - """Encode a mid-stream error as a Bedrock ConverseStream exception frame.""" +def _exception_frame(error_code: str, message: str) -> bytes: + """Encode a mid-stream error frame using the generic error-code/error-message headers (botocore falls back to these unless the code exactly matches one of the operation's named exception shapes, e.g. "validationException").""" headers = { - ":exception-type": exception_type, + ":error-code": error_code, + ":error-message": message, ":content-type": "application/json", ":message-type": "exception", } @@ -142,11 +150,26 @@ def forward(self, req, subpath, body): status=resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 200), headers=[("Content-Type", "application/vnd.amazon.eventstream")], ) - pr.stream = self._encode_converse_stream(resp["stream"], pr) + pr.stream = self._encode_event_stream(resp["stream"], pr) + return pr + + elif op == "invoke-with-response-stream": + resp = self._client.invoke_model_with_response_stream( + modelId=model_id, + body=body, + contentType=req.headers.get("Content-Type", "application/json"), + accept=req.headers.get("Accept", "application/json"), + ) + + pr = ProxyResponse( + status=resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 200), + headers=[("Content-Type", "application/vnd.amazon.eventstream")], + ) + pr.stream = self._encode_event_stream(resp["body"], pr) return pr else: raise NotImplementedError( - f"bedrock op '{op}' not supported (only invoke, converse, and converse-stream)" + f"bedrock op '{op}' not supported (only invoke, converse, converse-stream, and invoke-with-response-stream)" ) except ClientError as exc: @@ -161,16 +184,18 @@ def forward(self, req, subpath, body): @staticmethod - def _encode_converse_stream(events, pr: ProxyResponse): - """Re-frame boto3's already-decoded ConverseStream events - (``{"messageStart": {...}}``, ``{"contentBlockDelta": {...}}``, ..., - finally ``{"metadata": {"usage": {...}}}``) back into the AWS - event-stream wire format the caller's own boto3 client expects. - - Also captures usage off the trailing "metadata" event onto ``pr`` (read - by hooks.on_response only after this generator is exhausted, since - usage isn't known until then) and turns any mid-stream failure into a - single exception frame instead of dropping the connection. + def _encode_event_stream(events, pr: ProxyResponse): + """Re-frame boto3's already-decoded events (ConverseStream's + ``{"messageStart": {...}}``, ..., ``{"metadata": {"usage": {...}}}``, + or InvokeModelWithResponseStream's ``{"chunk": {"bytes": ...}}``) back + into the AWS event-stream wire format the caller's own boto3 client + expects. Shared by both streaming ops since neither's framing depends + on which operation produced the events. + + Also captures usage off a trailing ConverseStream "metadata" event + onto ``pr`` (a no-op for InvokeModelWithResponseStream, which has no + such event) and turns any mid-stream failure into a single exception + frame instead of dropping the connection. """ try: for event in events: diff --git a/canyonos_core/llm_proxy/stub.py b/canyonos_core/llm_proxy/stub.py index f656bd0..ae75457 100644 --- a/canyonos_core/llm_proxy/stub.py +++ b/canyonos_core/llm_proxy/stub.py @@ -36,22 +36,10 @@ def _json_response(obj, status=200): ) -def _bedrock_converse_stream_response(text): - """A minimal, validly-framed ConverseStream event sequence so - ``CANYONOS_LLM_STUB_TEXT`` exercises the exact same wire format - (``application/vnd.amazon.eventstream``) a real Bedrock call would, - without needing AWS credentials.""" +def _stream_response(events, stream_usage=None): + """Build a validly-framed event-stream ``ProxyResponse`` from a list of (event_type, body) pairs, shared by both Bedrock streaming stubs.""" # Imported lazily so non-Bedrock stubs don't need boto3/botocore. from canyonos_core.llm_proxy.providers.bedrock import _event_frame - from canyonos_core.llm_proxy.providers.base import ProxyResponse - - events = [ - ("messageStart", {"role": "assistant"}), - ("contentBlockDelta", {"contentBlockIndex": 0, "delta": {"text": text}}), - ("contentBlockStop", {"contentBlockIndex": 0}), - ("messageStop", {"stopReason": "end_turn"}), - ("metadata", {"usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}}), - ] def gen(): for event_type, body in events: @@ -62,16 +50,42 @@ def gen(): headers=[("Content-Type", "application/vnd.amazon.eventstream")], ) pr.stream = gen() - pr.stream_usage = {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2} + pr.stream_usage = stream_usage return pr +def _bedrock_converse_stream_response(text): + """A minimal ConverseStream event sequence so ``CANYONOS_LLM_STUB_TEXT`` + exercises the exact same wire format a real Bedrock call would, without + needing AWS credentials.""" + usage = {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2} + events = [ + ("messageStart", {"role": "assistant"}), + ("contentBlockDelta", {"contentBlockIndex": 0, "delta": {"text": text}}), + ("contentBlockStop", {"contentBlockIndex": 0}), + ("messageStop", {"stopReason": "end_turn"}), + ("metadata", {"usage": usage}), + ] + return _stream_response(events, stream_usage=usage) + + +def _bedrock_invoke_stream_response(text): + """A minimal InvokeModelWithResponseStream ``chunk`` event sequence; unlike + ConverseStream there's no unified usage metadata event, so ``stream_usage`` + stays unset here (matches real Bedrock behavior for this op).""" + chunk_body = json.dumps({"outputText": text, "generation": text}).encode("utf-8") + events = [("chunk", {"bytes": chunk_body})] + return _stream_response(events) + + def build_stub(provider_name, subpath, text): """Build a provider-appropriate canned response carrying ``text``.""" if provider_name == "bedrock": op = subpath.rsplit("/", 1)[-1] if subpath else "" if op == "converse-stream": return _bedrock_converse_stream_response(text) + if op == "invoke-with-response-stream": + return _bedrock_invoke_stream_response(text) if op == "converse": return _json_response({ "output": {"message": {"role": "assistant", diff --git a/tests/test_llm_proxy_bedrock_streaming.py b/tests/test_llm_proxy_bedrock_streaming.py index 86f9c9b..f4b4197 100644 --- a/tests/test_llm_proxy_bedrock_streaming.py +++ b/tests/test_llm_proxy_bedrock_streaming.py @@ -50,7 +50,8 @@ def test_event_frame_helper(self): def test_exception_frame_helper(self): raw = _exception_frame("ThrottlingException", "slow down") [(headers, payload)] = _decode_frames(raw) - self.assertEqual(headers[":exception-type"], "ThrottlingException") + self.assertEqual(headers[":error-code"], "ThrottlingException") + self.assertEqual(headers[":error-message"], "slow down") self.assertEqual(headers[":message-type"], "exception") self.assertEqual(json.loads(payload), {"message": "slow down"}) @@ -138,5 +139,45 @@ def failing_events(): self.assertTrue(pr.stream_error) +class BedrockProviderInvokeStreamTests(unittest.TestCase): + def _make_provider(self): + with patch.object(bedrock_module.boto3, "client") as mock_client_factory: + self.mock_client = MagicMock() + mock_client_factory.return_value = self.mock_client + return BedrockProvider(_FakeCfg()) + + def test_invoke_stream_encodes_chunks(self): + provider = self._make_provider() + chunk_bytes = [ + json.dumps({"generation": "hi"}).encode("utf-8"), + json.dumps({"generation": " there"}).encode("utf-8"), + ] + events = [{"chunk": {"bytes": b}} for b in chunk_bytes] + self.mock_client.invoke_model_with_response_stream.return_value = { + "ResponseMetadata": {"HTTPStatusCode": 200}, + "body": iter(events), + } + + req = MagicMock() + req.headers = {} + body = json.dumps({"prompt": "hi"}).encode() + pr = provider.forward(req, "model/meta.llama3-8b/invoke-with-response-stream", body) + + self.assertIsNotNone(pr.stream) + raw = b"".join(pr.stream) + decoded = _decode_frames(raw) + self.assertEqual([h[":event-type"] for h, _ in decoded], ["chunk", "chunk"]) + + import base64 + first_payload = json.loads(decoded[0][1]) + self.assertEqual(base64.b64decode(first_payload["bytes"]), chunk_bytes[0]) + self.assertIsNone(pr.stream_usage) # no usage metadata event for this op + self.assertFalse(pr.stream_error) + + self.mock_client.invoke_model_with_response_stream.assert_called_once() + called_kwargs = self.mock_client.invoke_model_with_response_stream.call_args.kwargs + self.assertEqual(called_kwargs["modelId"], "meta.llama3-8b") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_llm_proxy_streaming_e2e.py b/tests/test_llm_proxy_streaming_e2e.py index e8ed5bc..1481224 100644 --- a/tests/test_llm_proxy_streaming_e2e.py +++ b/tests/test_llm_proxy_streaming_e2e.py @@ -55,6 +55,25 @@ def test_converse_stream_returns_valid_eventstream_body(self): delta_payload = json.loads(decoded[1][1]) self.assertEqual(delta_payload["delta"]["text"], "hello from stub") + @patch.dict(os.environ, {"CANYONOS_LLM_STUB_TEXT": "hello from stub"}) + def test_invoke_with_response_stream_returns_valid_eventstream_body(self): + resp = self.client.post( + "/bedrock/model/meta.llama3-8b-instruct-v1:0/invoke-with-response-stream", + data=b'{"prompt": "hi"}', + content_type="application/json", + ) + self.assertEqual(resp.status_code, 200) + self.assertEqual(resp.headers["Content-Type"], "application/vnd.amazon.eventstream") + + decoded = _decode_frames(resp.data) + self.assertEqual([h[":event-type"] for h, _ in decoded], ["chunk"]) + + import base64 + import json + chunk_payload = json.loads(decoded[0][1]) + chunk_bytes = base64.b64decode(chunk_payload["bytes"]) + self.assertEqual(json.loads(chunk_bytes)["generation"], "hello from stub") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_llm_proxy_usage_extraction.py b/tests/test_llm_proxy_usage_extraction.py new file mode 100644 index 0000000..89d9140 --- /dev/null +++ b/tests/test_llm_proxy_usage_extraction.py @@ -0,0 +1,78 @@ +"""Unit tests for Hooks._extract_usage's per-provider/op/model dispatch.""" + +import json +import os +import sys +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from canyonos_core.llm_proxy.hooks import Ctx, Hooks +from canyonos_core.llm_proxy.providers.base import ProxyResponse + + +def _ctx(provider, subpath): + return Ctx(provider=provider, method="POST", subpath=subpath, body=b"", + headers={}, t0=0.0) + + +def _json_response(obj, status=200): + return ProxyResponse(status=status, headers=[], content=json.dumps(obj).encode()) + + +class ExtractUsageTests(unittest.TestCase): + def setUp(self): + self.hooks = Hooks() + + def test_bedrock_converse_uses_camel_case_schema(self): + resp = _json_response({"usage": {"inputTokens": 3, "outputTokens": 5, "totalTokens": 8}}) + usage = self.hooks._extract_usage(_ctx("bedrock", "model/anthropic.claude-3/converse"), resp) + self.assertEqual((usage.input_tokens, usage.output_tokens, usage.total_tokens), (3, 5, 8)) + + def test_bedrock_converse_stream_reads_stream_usage_not_content(self): + resp = ProxyResponse(status=200, headers=[]) + resp.stream = iter([b""]) + resp.stream_usage = {"inputTokens": 1, "outputTokens": 2, "totalTokens": 3} + usage = self.hooks._extract_usage(_ctx("bedrock", "model/anthropic.claude-3/converse-stream"), resp) + self.assertEqual((usage.input_tokens, usage.output_tokens), (1, 2)) + + def test_bedrock_invoke_anthropic_model_uses_snake_case_schema(self): + resp = _json_response({"usage": {"input_tokens": 10, "output_tokens": 20}}) + usage = self.hooks._extract_usage( + _ctx("bedrock", "model/anthropic.claude-3-5-sonnet-20240620-v1:0/invoke"), resp + ) + self.assertEqual((usage.input_tokens, usage.output_tokens, usage.total_tokens), (10, 20, 30)) + + def test_bedrock_invoke_non_anthropic_model_yields_no_usage(self): + # Llama's native invoke body has no "usage" key at all -- unsupported for now. + resp = _json_response({"generation": "hi", "prompt_token_count": 5, "generation_token_count": 7}) + usage = self.hooks._extract_usage(_ctx("bedrock", "model/meta.llama3-8b-instruct-v1:0/invoke"), resp) + self.assertIsNone(usage) + + def test_bedrock_invoke_with_response_stream_yields_no_usage(self): + resp = ProxyResponse(status=200, headers=[]) + resp.stream = iter([b""]) + resp.stream_usage = None + usage = self.hooks._extract_usage( + _ctx("bedrock", "model/meta.llama3-8b-instruct-v1:0/invoke-with-response-stream"), resp + ) + self.assertIsNone(usage) + + def test_direct_anthropic_api_uses_snake_case_schema(self): + resp = _json_response({"usage": {"input_tokens": 4, "output_tokens": 6}}) + usage = self.hooks._extract_usage(_ctx("anthropic", "v1/messages"), resp) + self.assertEqual((usage.input_tokens, usage.output_tokens, usage.total_tokens), (4, 6, 10)) + + def test_direct_openai_api_uses_openai_schema(self): + resp = _json_response({"usage": {"prompt_tokens": 7, "completion_tokens": 9, "total_tokens": 16}}) + usage = self.hooks._extract_usage(_ctx("openai", "v1/chat/completions"), resp) + self.assertEqual((usage.input_tokens, usage.output_tokens, usage.total_tokens), (7, 9, 16)) + + def test_non_200_status_yields_no_usage(self): + resp = _json_response({"usage": {"input_tokens": 1, "output_tokens": 1}}, status=400) + usage = self.hooks._extract_usage(_ctx("anthropic", "v1/messages"), resp) + self.assertIsNone(usage) + + +if __name__ == "__main__": + unittest.main() From cb13743ff490ffdfc1e3028043456ab154953a1b Mon Sep 17 00:00:00 2001 From: Saaketh Sodanapalli Date: Wed, 9 Sep 2026 14:59:33 -0700 Subject: [PATCH 3/3] Remove stale ventis/ leftovers from the CanyonOS rename - ventis/llm_proxy/: an orphaned pre-rename copy of the LLM proxy package left behind by PR #76's merge. Verified nothing imports from it -- every real reference in the repo (local_controller.py, stub_generator.py, and this branch's own new tests) already points at canyonos_core.llm_proxy, which has this branch's actual streaming work. Dead weight, not wired to anything. - VENTIS_TO_CANYONOS_RENAME.md: the pre-rename migration analysis/plan doc (dated 2026-09-05, explicitly 'no runtime code has been renamed yet'). The rename it planned has since been executed and merged (PR #76); the plan document is now stale. Co-Authored-By: Claude Sonnet 5 --- VENTIS_TO_CANYONOS_RENAME.md | 691 ------------------------ ventis/llm_proxy/README.md | 112 ---- ventis/llm_proxy/__init__.py | 14 - ventis/llm_proxy/__main__.py | 29 - ventis/llm_proxy/app.py | 46 -- ventis/llm_proxy/config.py | 66 --- ventis/llm_proxy/core.py | 46 -- ventis/llm_proxy/hooks.py | 159 ------ ventis/llm_proxy/providers/__init__.py | 14 - ventis/llm_proxy/providers/anthropic.py | 19 - ventis/llm_proxy/providers/base.py | 96 ---- ventis/llm_proxy/providers/bedrock.py | 119 ---- ventis/llm_proxy/providers/openai.py | 18 - ventis/llm_proxy/proxy.py | 58 -- ventis/llm_proxy/requirements.txt | 3 - 15 files changed, 1490 deletions(-) delete mode 100644 VENTIS_TO_CANYONOS_RENAME.md delete mode 100644 ventis/llm_proxy/README.md delete mode 100644 ventis/llm_proxy/__init__.py delete mode 100644 ventis/llm_proxy/__main__.py delete mode 100644 ventis/llm_proxy/app.py delete mode 100644 ventis/llm_proxy/config.py delete mode 100644 ventis/llm_proxy/core.py delete mode 100644 ventis/llm_proxy/hooks.py delete mode 100644 ventis/llm_proxy/providers/__init__.py delete mode 100644 ventis/llm_proxy/providers/anthropic.py delete mode 100644 ventis/llm_proxy/providers/base.py delete mode 100644 ventis/llm_proxy/providers/bedrock.py delete mode 100644 ventis/llm_proxy/providers/openai.py delete mode 100644 ventis/llm_proxy/proxy.py delete mode 100644 ventis/llm_proxy/requirements.txt diff --git a/VENTIS_TO_CANYONOS_RENAME.md b/VENTIS_TO_CANYONOS_RENAME.md deleted file mode 100644 index 43c2d4e..0000000 --- a/VENTIS_TO_CANYONOS_RENAME.md +++ /dev/null @@ -1,691 +0,0 @@ -# Ventis → CanyonOS Rename: Verified Migration Plan - -> **Audit status (2026-09-05):** re-checked against the complete tracked tree, -> hidden files, ignored/generated state, current tests, and Canyon Code company -> memory. Seven independent Luna agents audited runtime, packaging, infrastructure, -> tests, documentation, edge cases, and inventory. This file is analysis and an -> execution plan only; no runtime code has been renamed yet. - -## Executive verdict - -Do **not** implement this as a global search-and-replace. The current tree has one -blocking architecture decision and several versioned contracts that require an -additive compatibility phase. - -1. **Blocking package-name collision.** The root distribution/package is currently - `ventis` (`pyproject.toml:2,24,31`), while `cli/` already owns the distribution, - import package, and executable name `canyonos` (`cli/pyproject.toml:2,14,21`). - Renaming the root package and distribution to `canyonos` would make two editable - projects provide the same distribution and top-level package. A temporary - reproduction fails `uv lock --offline` with conflicting URLs for `canyonos`. -2. **The repository deliberately documents old names as compatibility protocol.** - `.claude/skills/porting-to-canyonos-core/SKILL.md:4-12` and - `references/runtime-contract.md:1-5` currently require the `ventis` Python/CLI, - `VENTIS_*` variables, and `ventis-*` Docker names. That policy must be replaced - or deprecated; simply changing code makes the skill teach users the wrong API. -3. **Generated artifacts are part of the runtime contract.** The build generator - creates a nested `ventis.llm_proxy` package, copies `ventis_context.py`, writes - `VENTIS_AGENT_*`, and the generated controller launches `python -m - ventis.llm_proxy`. Updating only the source package will produce images that - build but fail at runtime. -4. **The host CLI participates in the old protocol.** Contrary to the previous - draft, `cli/canyonos/init.py:156` writes `VENTIS_REDIS_HOST` into the Global - Controller container. It also consumes old Docker prefixes and the validator's - `capabilities.ventis` JSON field. -5. **The test baseline is not green.** A clean `uv run pytest -q` currently stops - with nine collection errors because `local_controler_pb2` is not generated. - With both protos generated into a temporary `PYTHONPATH`, the baseline is - **212 passed, 11 failed, 3 subtests passed**. Those 11 failures are pre-existing - behavior/test drift, not rename regressions. - -The safe route is: decide package ownership, introduce CanyonOS names alongside -legacy readers/aliases, switch every producer and consumer together, validate -fresh and upgrade deployments, then remove compatibility names in a later major -release. A literal zero-match tree is the **end state**, not a safe first commit. - -> **NOTE:** The operator has since chosen a **hard cutover with no compatibility -> shims** (see "Locked decisions" below). That decision **supersedes** every -> "additive/compat/fallback/dual-read/shim" recommendation in this document. The -> *contract coupling* (which producers and consumers must change together) still -> fully applies — only the transitional fallbacks are dropped. Where a section -> below proposes old+new fallbacks, read it as "change producer and consumer to the -> new name in the same commit; delete the old name outright." - -## Locked decisions (2026-09-05, operator-approved) - -Hard cutover. **No** legacy env-var dual-read, **no** legacy import alias, **no** -old-header fallback, **no** compatibility release window. Existing running -deployments must be torn down and rebuilt; this is an accepted breaking change. - -| Surface | Decision | -|---|---| -| Core import package / dir | `canyonos_core` (dir `ventis/` → `canyonos_core/`); all imports `from canyonos_core…` | -| Core distribution name | `canyonos-core` | -| `ventis_context` module + alias | `canyonos_context.py` / alias `canyonos_context` | -| Env vars | `VENTIS_*` → `CANYONOS_*` (⚠ collision note below) | -| Resource prefix (network/image/container/redis/tags/ids) | `canyonos-*` (`canyonos-local`, `canyonos-`, `canyonos-redis-*`, `canyonos-ec2-*`) | -| Future-ID HTTP header | `X-Canyonos-Future-ID` (`Canyonos` cased, `ID` all-caps) — injector **and** reader standardized to this exact spelling | -| Published GC image | **unchanged** — stays `saakeths/canyonos:latest` | -| Core executable | **removed** — core is import-only; the standalone `canyonos` CLI is the sole console script | -| Root `pyproject.toml` | **NOT deleted — rewritten import-only** (deleting breaks the container build; see below) | -| `uv.lock` | regenerate **after** the rename + pyproject rewrite land; never hand-edit | -| SSH key default | **unchanged** — stays `~/.ssh/ventis_ec2` for now | -| Logo asset + README clone URL | **unchanged** for now (`images/ventis-logo.png`, git URL) | - -**Root `pyproject.toml` is critical — do not delete.** `ventis/Dockerfile:5-6` -runs `COPY . /ventis` + `RUN pip install /ventis`, which requires the root -`pyproject.toml`. It also supplies the entire runtime dependency set (boto3, -grpcio(-tools), redis, sqlalchemy, psycopg, flask, opentelemetry-\*) and the -`[tool.setuptools.package-data]` that ships `controller/proto/*.proto` and -`controller/utils/aws_pricing_chart.db` into the installed package. Deleting it -makes the container build fail and the runtime lose its bundled data. **Rewrite it -instead:** -- `name = "canyonos-core"`, `version` kept; -- **drop** `[project.scripts]` entirely (import-only); -- **drop** `[dependency-groups] dev` `canyonos` entry **and** `[tool.uv.sources] - canyonos = { path = "cli", editable = true }` — this self-dependency is exactly - what caused the `uv lock` collision noted in company memory; removing it is what - makes the two distributions coexist; -- `[tool.setuptools.packages.find] include = ["canyonos_core*"]`; -- `[tool.setuptools.package-data]` key `ventis` → `canyonos_core` (and drop the - stale `templates/**/*` entry — that dir no longer exists); -- Ty `[tool.ty.*]` include/exclude/allowed-unresolved-import paths - (`ventis` → `canyonos_core`, `ventis_context` → `canyonos_context`). - -**⚠ `CANYONOS_*` collision watch.** `VENTIS_REDIS_HOST`/`VENTIS_REDIS_PORT` become -`CANYONOS_REDIS_HOST`/`CANYONOS_REDIS_PORT`, which are **also** the names the -user-side dashboard stack already writes (`cli/canyonos/dashboard_stack.py:227-228`). -They live in different process/compose scopes (core GC/agent containers vs. the -dashboard compose), so there is no runtime clash today — but the names are now -semantically overloaded. Verify no single process reads both; if that ever changes, -the core vars would need a `CANYONOS_CORE_*` namespace. - -## Verified inventory - -The canonical count uses the tracked `HEAD` tree (before this currently untracked -analysis file is added) and case-insensitive token matches. All future recounts -must exclude this document so it does not count its own inventory: - -| Scope | Matching files | Occurrences | Matching lines | -|---|---:|---:|---:| -| Source tree, excluding this document and `uv.lock` | 93 | 628 | 578 | -| `uv.lock` | 1 | 1 | 1 | -| Source tree including `uv.lock`, excluding this document | 94 | 629 | 579 | - -Exact-case totals excluding `uv.lock` are 458 `ventis`, 62 `Ventis`, and 108 -`VENTIS`. There are 173 tracked files total. The 94 matching files break down as: - -| Area | Files with content matches | -|---|---:| -| Root runtime directory | 36 | -| Tests | 25 | -| Examples | 15 | -| Vendored porting skill | 7 | -| Standalone CLI | 7 | -| Root metadata/docs | 4 | - -There are 55 tracked paths containing the old name: all 53 files under `ventis/`, -plus `images/ventis-logo.png` and `tests/test_ventis_context.py`. The migration -document's own filename/content must be excluded while work is in progress and -renamed or archived at final cleanup. - -Reproduce the audit with: - -```bash -git grep -I -i -l ventis -- . ':!uv.lock' ':!VENTIS_TO_CANYONOS_RENAME.md' | wc -l -git grep -I -i -o ventis -- . ':!uv.lock' ':!VENTIS_TO_CANYONOS_RENAME.md' | wc -l -git grep -I -i -n ventis -- . ':!VENTIS_TO_CANYONOS_RENAME.md' -git ls-files | rg -i 'ventis' -find . -path './.git' -prune -o -iname '*ventis*' -print -rg --hidden --no-ignore -i ventis -g '!.git/**' -``` - -`git grep` is the tracked-source authority; the final `rg` catches stale virtual -environments, editable-install metadata, generated `.car/` output, and other -ignored files that can mask a bad migration. - -## Decision gate 1: package and executable ownership - -This must be resolved before moving `ventis/`. **RESOLVED** — see Locked decisions. - -### Topology (locked) - -| Surface | Owner/name | -|---|---| -| User-facing distribution | existing `cli/` distribution: `canyonos` | -| User-facing import package | existing `cli/canyonos/` | -| User-facing executable | existing `canyonos` console script (the **only** one) | -| Core runtime distribution | `canyonos-core` | -| Core runtime import package | `canyonos_core` | -| In-container entrypoints | `python -m canyonos_core.server`, `.cli`, `.llm_proxy` | -| Legacy package/executable | **none** — hard cutover, no shim; core console script removed | - -This keeps the already-thin host CLI independent and avoids two wheels overwriting -one `canyonos/` directory. It also lets the root runtime be versioned independently -inside `saakeths/canyonos:`. - -The valid alternative is to merge the root runtime into the existing CLI -distribution under one intentionally owned `canyonos` tree. That is a larger -packaging refactor and must include dependency, image, and release ownership. - -**Invalid topology:** changing root `name`, package include, directory, and console -script to `canyonos` while leaving `cli/pyproject.toml` unchanged. It breaks lock -resolution, editable installs, module resolution from the repo root, and script -ownership. - -Also decide whether the old root command remains temporarily available. The old -runtime command exposes `new-project`, `deploy`, and `clean`; the standalone -`canyonos` CLI exposes `new-app`, agent-driven `build`, HTTP-driven `deploy`, and -other commands. `tests/run_tests.sh` therefore cannot be fixed by replacing the -word in place—the desired command semantics must be mapped explicitly. - -## Decision gate 2: compatibility policy - -**RESOLVED — hard cutover, declared breaking.** No compatibility window, no -dual-read, no fallback. A full teardown/rebuild is required; existing deployments -do **not** survive the switch. Every producer and its consumer(s) change to the new -name in the same commit, and the old name is deleted outright. The per-item -"legacy fallback" bullets below are **void** and retained only to enumerate the -producer/consumer pairs that must move together: - -- Env variable: producer + consumer switch to `CANYONOS_*` together; no `VENTIS_*` reader remains. -- HTTP header: injector + reader switch to `X-Canyonos-Future-ID` together. -- Validator capability key: emitter (`validate.py`) + consumer (`cli/canyonos/verify.py`) + tests switch together. -- SSH default: **unchanged** (`~/.ssh/ventis_ec2`) per Locked decisions. -- Docker/resource prefixes: generators + `cli/canyonos/verify.py` switch to `canyonos-*` together; no old-prefix recognition. - -## Contract map: changes that must move together - -### 1. Distribution, Python imports, and process entrypoints — critical - -Current package metadata in `pyproject.toml` contains: - -- distribution `name = "ventis"`; -- console script `ventis = "ventis.cli:main"`; -- package discovery `include = ["ventis*"]`; -- package-data ownership under `ventis`; -- Ty include/exclude paths rooted at `ventis`; -- Ty's allowed flat import `ventis_context`. - -After choosing the topology, update all of those together and regenerate -`uv.lock`; do not hand-edit the lock. `uv.lock:57` already contains the CLI -`canyonos` package and `uv.lock:1225` contains the root `ventis` package, which is -direct evidence of the collision. - -Runtime imports span `server.py`, `cli.py`, `stub_generator.py`, all controller -modules/providers/utilities, `OTLP_Exporter`, and `llm_proxy`. The easy-to-miss -process boundaries are: - -- `ventis/Dockerfile:5-17`: `/ventis` build root, protoc paths, and - `python -m ventis.server`; -- `ventis/server.py:9-10,56`: imports runtime helpers and spawns - `python -m ventis.cli deploy`; -- `ventis/controller/local_controller.py:143-147`: passes runtime env and spawns - `python -m ventis.llm_proxy`; -- `ventis/controller/instance_manager.py:14,232`: imports both provider runtimes; -- `.claude/skills/porting-to-canyonos-core/validate.py:122-141`: imports the runtime - and probes its env-file module paths. - -The bare fallbacks (`import ventis_context`, `import deploy`, generated gRPC -modules, etc.) exist because source files are copied flat into generated images. -Do not mechanically convert those to package-qualified imports without testing -both installed-source and generated-flat layouts. - -### 2. Generated agent/workflow build contexts — critical - -`ventis/stub_generator.py` is effectively a template engine even though it does -not use template files: - -- lines 310-324 copy `llm_proxy` to `/ventis/llm_proxy` and copy the - package `__init__.py`; -- lines 395-414 and 521 copy `controller/ventis_context.py` as the flat file - `ventis_context.py`; -- lines 443-461 emit `ENV VENTIS_AGENT_NAME` and `VENTIS_AGENT_FILE`; -- copied `local_controller.py` launches `python -m ventis.llm_proxy`; -- the collision list in `validate.py:41-46` explicitly reserves - `ventis_context.py`. - -Source, generated destination, fallback import names, validator collision rules, -and generated Dockerfile env names must change in one slice. During a compatibility -release, generated contexts may carry a small legacy import shim and both env-name -read paths. Tests must inspect the generated files and boot them; a successful -host-side import is insufficient. - -### 3. Host CLI ↔ Global Controller contracts — critical - -The host CLI communicates with the image over stable HTTP endpoints (`/deploy`, -`/clean`, `/status`, `/endpoints`); those endpoint paths contain no old brand and -should not be renamed. - -Brand-bearing coupling that does require coordination: - -- `cli/canyonos/init.py:156` injects `VENTIS_REDIS_HOST`; `ventis/server.py:85-95` - and the controller read it. During a mixed-image transition the CLI should pass - both names, and the new runtime should dual-read with `CANYONOS_*` precedence. -- `cli/canyonos/verify.py:43,260-262` looks for `ventis-local-*` containers and - `ventis-*` images. It must recognize both during upgrade and switch its emitted - guidance to CanyonOS. -- Validator JSON is a wire-like contract: `validate.py:120-126` emits - `capabilities["ventis"]`; `cli/canyonos/verify.py:88-109` consumes it; tests in - `tests/test_canyonos_test.py:44-50,150-173` encode it. Prefer a new neutral or - runtime-specific key while accepting the old key for one compatibility release. -- CLI docstrings/help in `cli/cli.py`, `cli/canyonos/deploy.py`, `gc.py`, - `verify.py`, `dashboard.compose.yml`, and `cli/ARCHITECTURE.md` still describe - the old runtime and must follow the functional cutover. - -The deploy progress parser matches message substrings rather than logger prefixes, -so renaming `logging.getLogger("ventis")` should not break its current parser. -However, `tests/test_deploy_progress.py` hardcodes many complete -`INFO:ventis...` lines and must be updated. - -### 4. Environment variables — critical external API - -Distinct current variables: - -```text -VENTIS_AGENT_FILE -VENTIS_AGENT_HOST -VENTIS_AGENT_NAME -VENTIS_AGENT_PORT -VENTIS_DATABASE_URL -VENTIS_DEMO_SERVER_COST_MULTIPLIER -VENTIS_DEMO_TOKEN_COST_MULTIPLIER -VENTIS_DOCKER_PLATFORM -VENTIS_LC_HOST -VENTIS_LC_PORT -VENTIS_MAX_AGENT_INSTANCES -VENTIS_OTEL_DESTINATIONS -VENTIS_POLL_INTERVAL -VENTIS_PROJECT_ID -VENTIS_REDIS_HOST -VENTIS_REDIS_PORT -``` - -Producers include both provider runtimes, `stub_generator.py`, and the standalone -CLI's `init.py`. Consumers include deploy/future/global/local controllers, -controller frontend, server, session/telemetry logging, LLM proxy config, and root -CLI. Tests heavily patch only the legacy names today. - -For a no-break transition: - -1. Add a single helper for `CANYONOS_*` first / `VENTIS_*` fallback and warn once. -2. Update producers to emit new names; where old images may consume them, emit - both temporarily. -3. Add precedence, fallback, and warning tests for every externally configurable - variable class—not just a blind test-string rename. -4. Update docs/skill only after the new readers are released. -5. Remove the legacy branch only at the announced compatibility boundary. - -`VENTIS_OTEL_DESTINATIONS` appears in docs/tests but current runtime configuration -has moved to the Redis key `otel:destinations`; confirm whether the env name is -already obsolete before adding a new alias. - -### 5. Future-ID HTTP header — telemetry correctness - -`ventis/llm_proxy/proxy.py:30-46` injects `X-Ventis-Future-ID`; the proxy reads it -at `ventis/llm_proxy/hooks.py:94` using different casing. HTTP header names are -case-insensitive, so the casing difference itself is safe. - -**Locked:** switch injector **and** reader to the exact spelling -`X-Canyonos-Future-ID` in the same commit (no legacy header accepted). Fix the -existing casing inconsistency at the same time so both sides use `X-Canyonos-Future-ID`. -Add the currently missing producer→consumer regression test; otherwise attribution -can silently disappear while requests still succeed. - -Do not rename the existing `gen_ai.*`, `project_id`, or `canyon.project.id` OTEL -attributes merely for branding. Those are separate telemetry schemas and no -`ventis`-prefixed OTEL attribute exists. - -### 6. Docker, EC2, Redis, and filesystem resource names — upgrade risk - -Name generation is distributed, not confined to `global_controller.py`: - -- Local provider (`Local/_runtime.py:19,42-56`): network `ventis-local`, Redis - host/container, runtime IDs, image names; -- EC2 provider (`EC2/_runtime.py:86-108,211,241-242`): AWS `Name` tags, - `ventis-ec2-*` runtime IDs, Redis containers, images, and containers; -- Global controller (`global_controller.py:153-179,396`): stale-resource cleanup - and Redis containers; -- root build (`ventis/cli.py:400`): image tags; -- CLI verification (`cli/canyonos/verify.py:43,260-262`): expected image/container - names; -- Redis probe (`controller/utils/redis_utils.py:10`): exact key - `__ventis_redis_healthcheck__`; -- remote secret copy (`controller/utils/env_file.py:61`): - `/tmp/ventis-env-`; -- Flask/logger identifiers (`server.py:12`, `controller/deploy.py:103`, - `ventis/cli.py:22`) and user-facing controller description - (`global_controller.py:922`). - -There is also a pre-existing cleanup mismatch: global cleanup expects -`ventis--` while the Local provider launches -`ventis-local--`. Fix or explicitly account for that before using -cleanup behavior as proof of a successful rename. - -A compatible upgrade must: - -- stop the active deployment before switching image versions; -- discover and remove both old and new container prefixes during the transition; -- account for both network names and avoid orphaning the old network; -- make verification recognize old resources but label them as legacy; -- rebuild every agent/workflow image so generated code and the GC agree; -- update EC2 tag expectations and any operational filters; -- clean both old and new remote env-file patterns best-effort; -- test rollback using a pinned previous GC image, not mutable `latest` alone. - -Runtime routing Redis keys such as `routing_table:*`, `agent:*`, `future:*`, and -`request:*` are brand-neutral and should remain unchanged. Runtime IDs stored in -those records do contain old Docker names, so an in-place Redis deployment must -not straddle versions; prefer a controlled teardown and fresh deploy. - -### 7. Files, defaults, and persistent data - -- SSH defaults exist in `EC2/_runtime.py:33` and - `global_controller.py:783` as `~/.ssh/ventis_ec2`. **Locked: leave unchanged for - now** — both stay `~/.ssh/ventis_ec2`. (These two lines are an intentional - exception to the zero-`ventis` end state until a later pass.) -- `examples/helloworld/config/global_controller.yaml:42` uses - `sqlite:///ventis_runtime.db`. Updating the sample does not migrate user-owned - databases. Existing config paths should remain valid; document an optional - user-controlled file move. -- `ventis/OTLP_Exporter/otel_queue.db` is tracked beneath the package directory. - Preserve it across the directory move and verify whether packaging/runtime - writes beside installed code are intentional before changing its location. -- `images/ventis-logo.png` and its `README.md` reference (plus the README clone URL) - are **locked as unchanged for now** — deferred to a later branding pass. -- `.gitignore:30,51-52` includes old comments and `.ventis-tests/`. -- `controller/utils/env_file.py` remote temp names can leave old files after a - crash; cleanup should understand both patterns, without broad `/tmp` deletion. - -### 8. Porting skill and remote delivery - -The entire vendored `.claude/skills/porting-to-canyonos-core/` tree teaches the -legacy compatibility contract. Functional changes are required in `validate.py`, -not just prose: - -- import/module probes at lines 120-141; -- flat-name collision list at line 45; -- dependency-name exception at line 1002; -- capability JSON/report handling at lines 1171-1172; -- messages and command examples throughout. - -The standalone CLI does not necessarily use this working-tree copy. -`cli/canyonos/build.py` downloads a skill from `SKILL_REF` and `SKILL_PATH` in the -GitHub repository. Update/publish that referenced branch/path first (or repoint it -to the merged source), then test a fresh cache. Existing local/global skill caches -can otherwise continue generating legacy scaffolding after this repo appears clean. - -### 9. Tests, examples, docs, and assets - -Functional test updates cover 24 test source/script files plus `tests/README.md`: - -- package imports/patch targets/loggers: `test_cli.py`, `test_deploy.py`, - `test_error_propagation.py`, `test_future.py`, controller tests, exporter tests, - Redis/runtime tests, session/telemetry tests, and `test_ventis_context.py`; -- Docker/resource contracts: `test_canyonos_test.py`, - `test_instance_manager_runtime.py`, `test_global_controller_redis_reuse.py`, - and `test_runtime_ec2.py`; -- log fixtures: `test_deploy_progress.py`; -- path injection: `test_future.py`, `test_error_propagation.py`, - `test_local_controller_metrics.py`, and `test_otel_exporter_fanout.py`; -- integration command semantics and temp/project paths: `tests/run_tests.sh`. - -All four example projects contain old prose, commands, source comments, or config -defaults. Documentation cleanup includes root `README.md`, `ventis/README.md`, -`FUTURE_SCHEMA.md` by directory move, exporter/proxy/EC2 docs, -`examples/helloworld/README.md`, `tests/README.md`, CLI docs, and the complete -porting-skill tree. - -Do docs/comments last. Several apparent prose strings are actually executable -examples or validator guidance and should be covered by command/import checks. - -## Pre-existing blockers to establish before rename work - -Record or fix these on a baseline commit so the migration has trustworthy gates: - -1. `tests/run_tests.sh` invokes pytest before generating protobuf modules. Three - tests also insert the absent `ventis/templates/grpc_stubs` path. Generate stubs - into a deterministic test location or isolate imports with fixtures. -2. After temporary proto generation, the current suite reports 212 passed and 11 - failed. Capture the exact expected baseline or fix those failures separately. -3. Root `ventis new-project` expects a `ventis/templates` directory that no longer - exists, while the standalone CLI uses the different `new-app` workflow. -4. Root `pyproject.toml` still has stale `templates/**/*` package-data and Ty - exclude entries. Confirm removal versus restoration instead of carrying them - through mechanically. -5. CI runs Ruff and Ty but not pytest, wheel-install tests, generated-context - tests, or image builds. Passing CI currently does not prove rename safety. -6. Ignored `.venv/`, `ventis.egg-info/`, `.pytest_cache/`, generated `.car/`, and - Docker state can preserve old entrypoints/imports. Verification must start from - clean generated state. - -## Ordered implementation plan - -### Phase 0 — freeze and baseline - -- Choose the package topology and compatibility window. -- Pin the current GC image by digest/tag for rollback. -- Fix or record baseline tests and deterministic proto generation. -- Add contract tests for env fallback/precedence, header fallback, validator JSON, - generated contexts, and old/new resource discovery. - -**Gate:** reproducible baseline in a clean environment, with known failures -explicitly separated from rename work. - -### Phase 1 — introduce the new runtime identity ✅ DONE (2026-09-05) - -**Executed (hard cutover, package identity only):** -- `git mv ventis/ → canyonos_core/`; `controller/ventis_context.py → canyonos_context.py`; - `tests/test_ventis_context.py → test_canyonos_context.py`. -- All `from ventis…/import ventis…` and module-path strings (test mocks, `-m` spawns, - logger names) → `canyonos_core`; `ventis_context` alias → `canyonos_context`. -- Generated flat-copy identity in `stub_generator.py` (`ventis/llm_proxy` → - `canyonos_core/llm_proxy`, flat `canyonos_context.py`) + fallback imports in - `local_controller.py`/`proxy.py` so agent containers import `canyonos_core`. -- Package logger `getLogger("ventis")` → `"canyonos_core"` (+ `test_deploy_progress`, - `test_cli` expectations); argparse `prog` → `canyonos_core`. -- `Dockerfile`: `COPY . /src`, `pip install /src`, protoc `-I/src/canyonos_core/...`, - `ENTRYPOINT python -m canyonos_core.server`. Published image name kept `saakeths/canyonos`. -- Root `pyproject.toml` rewritten import-only: `name = canyonos-core`, no - `[project.scripts]`, `find.include = [canyonos_core*]`, package-data key + Ty paths - updated, stale `templates/**` dropped. **Kept** the `canyonos` (cli) editable dev-dep - + `[tool.uv.sources]` — no longer collides now that root is `canyonos-core`, and the - root suite imports the CLI. `uv.lock` regenerated (`ventis` gone, `canyonos-core` in). -- **Verification:** `py_compile` all tracked `.py` OK; `canyonos_core` + entrypoints - import OK; **`uv run pytest -q` = 223 passed, 3 subtests passed, 0 failed.** - -### Environment-variable phase ✅ DONE (2026-09-05) - -**Executed (hard cutover, `VENTIS_*` → `CANYONOS_*`, producers + consumers together):** -- All core env reads/writes renamed across `canyonos_core/**` (controllers, both - provider `_runtime.py`, `future.py`, `server.py`, `deploy.py`, `llm_proxy/config.py`, - session/telemetry logging incl. `CANYONOS_DEMO_*` multipliers) and the generated - agent Dockerfile `ENV` in `stub_generator.py` (`CANYONOS_AGENT_NAME/FILE`). -- **Cross-boundary producer:** `cli/canyonos/init.py` now injects `CANYONOS_REDIS_HOST` - into the GC container, matching the core reader. -- Test expectations updated (`test_deploy`, `test_instance_manager_runtime`, - `test_global_controller_identity`, `test_session/telemetry_logging`, etc.). -- `VENTIS_OTEL_DESTINATIONS` confirmed **dead in code** (replaced by Redis key - `otel:destinations`) — no code rename needed; only stale in docs. -- **Collision watch confirmed benign:** `CANYONOS_REDIS_HOST/PORT` is also written by - `cli/canyonos/dashboard_stack.py`, but that targets the dashboard compose while - `init.py` targets the GC container — different processes, no single reader of both. -- **Verification:** `uv run pytest -q` = **223 passed, 3 subtests passed, 0 failed.** -- **Still `VENTIS_*` on purpose:** only the porting-skill docs (`SKILL.md`, - `runtime-contract.md`, `validate.py` message strings) and `OTLP_Exporter/DESIGN.md` - — deferred to the skill/docs phase. - -### Resource-name + header + validator-key + cosmetic phases ✅ DONE (2026-09-05) - -**Executed (hard cutover; every producer + consumer moved together):** -- **Resource prefixes `ventis-*` → `canyonos-*`:** core generators (both provider - `_runtime.py`, `global_controller.py` network/redis/container names, `cli.py` image - tags, `deploy.py`/`server.py` Flask app names, `env_file.py` `/tmp/canyonos-env-`, - `redis_utils.py` `__canyonos_redis_healthcheck__`) **and** the user-CLI consumer - `cli/canyonos/verify.py` (`RUNTIME_PREFIX`, image name) + all resource-name tests. -- **Future-ID header:** injector (`proxy.py`) and reader (`hooks.py`) both standardized - to exactly `X-Canyonos-Future-ID` (fixed the old `-ID`/`-Id` casing split), plus - `_inject_canyonos_headers`. -- **Validator capability key + framework-import check:** `caps["canyonos_core"]` / - `capabilities.canyonos_core` / `name == "canyonos_core"` aligned across - `validate.py` (emitter), `cli/canyonos/verify.py` (consumer), and - `test_canyonos_test.py`. -- **Docs/prose/cosmetic:** brand sweep `Ventis`→`CanyonOS`, `ventis`→`canyonos` across - READMEs, porting-skill docs (incl. remaining `VENTIS_*`→`CANYONOS_*`), `DESIGN.md`, - example configs/comments, `run_tests.sh` (`canyonos_test`), `.gitignore` - (`.canyonos-tests/`), and code comments/log strings; stale `ventis_context.py` doc - ref → `canyonos_context.py`; `VentisContextTests` → `CanyonosContextTests`. -- **Verification:** `py_compile` all tracked `.py` OK; header injector/reader agree; - `verify.py` prefix agrees with core generators; capability key aligned; `uv.lock` - has zero `ventis`; **`uv run pytest -q` = 223 passed, 3 subtests passed, 0 failed.** - -**Intentionally still `ventis` (operator decision):** only the EC2 SSH key default -`~/.ssh/ventis_ec2` (2 code lines + EC2 README + example config). Logo/URL were -changed by the operator directly. **No other `ventis` token remains anywhere in the -tracked tree.** - -#### Porting-skill semantic caveat -The token sweep updated the skill's identifiers, but `SKILL.md` / -`runtime-contract.md` still *describe* the old names as a "compatibility protocol that -remains" — which is no longer true under the hard cutover. A follow-up semantic pass -should rewrite that framing (out of scope for a pure rename). - ---- - -#### Original Phase 1 intent (for reference) - -- Create the chosen distinct runtime distribution/import package. -- Update packaging, Ty paths, internal imports, process module paths, and root - Dockerfile; regenerate `uv.lock`. -- If backward compatibility is promised, ship a minimal old import/command shim - that delegates to the new runtime and warns. -- Do not let root and `cli/` both own `canyonos`. - -**Gate:** isolated wheel installs prove the CLI and runtime packages can coexist; -the `canyonos` executable resolves to the standalone CLI; both new and promised -legacy imports behave as specified. - -### Phase 2 — migrate generated runtime artifacts - -- Update generator source/destination paths, flat context module, copied proxy - package, local-controller process invocation, Dockerfile env, and validator - collision rules as one unit. -- Rebuild all generated contexts from scratch; never reuse old output. - -**Gate:** generated agent and workflow contexts contain the intended package/env -names, import their controller/proxy, load an example agent, and boot in Docker. - -### Phase 3 — migrate protocol identifiers compatibly - -- Add new-first/old-fallback env reads and header reads. -- Change producers, including CLI `init.py` and generated Dockerfiles. -- Version the validator capability JSON transition and update CLI verification. -- Publish the updated remote porting skill and test an empty cache. - -**Gate:** old CLI/new image and new CLI/old image combinations either work within -the declared matrix or fail early with a precise version error; telemetry -attribution remains intact. - -### Phase 4 — migrate operational resource names - -- Change Local/EC2 image, container, network, runtime ID, Redis container, AWS tag, - healthcheck, and remote env-file names. -- Update GC cleanup and CLI verification together, recognizing both generations - for the compatibility release. -- Resolve the existing Local stale-cleanup mismatch. - -**Gate:** fresh local deploy, EC2 mocked/probe tests, upgrade teardown, verify, -clean, and rollback all leave no unexpected containers/networks/temp files. - -### Phase 5 — publish and switch - -- Build and inspect both wheels/sdists in clean environments. -- Build the GC image from the renamed runtime, pin a versioned tag/digest, smoke - `/status`, then update `GC_IMAGE`/release metadata. -- Run a representative end-to-end workflow through build, deploy, request, status, - telemetry, verify, stop, and quit. - -**Gate:** the published artifacts—not editable source installs—pass the full -matrix on a clean machine or clean VM. - -### Phase 6 — cosmetic cleanup and later compatibility removal - -- Update prose, help, examples, comments, ASCII/logo assets, and test names. -- After the announced compatibility period, remove shims/fallbacks and legacy - resource discovery in a major release. -- Rename/archive this migration document, recreate all generated state, and run - the final forbidden-token scan. - -## Acceptance matrix - -| Layer | Required proof | -|---|---| -| Static tree | No old token/path outside an explicit temporary compatibility allowlist | -| Lock/metadata | `uv lock` succeeds; wheel metadata has distinct owners/names | -| Clean installs | CLI and runtime wheels coexist; import paths and script owner are exact | -| Type/lint | Ruff and Ty pass with renamed include/exclude/unresolved-import paths | -| Unit tests | Protos generated deterministically; rename does not add failures | -| Generated output | Context contains new proxy/context/env names and no accidental stale package | -| Header telemetry | New header attributes correctly; legacy fallback works during transition | -| Env contract | New-wins precedence and every promised legacy fallback are tested | -| Local runtime | Image/network/Redis/container names agree with `canyonos verify` | -| EC2 runtime | Tags, image/container names, SSH fallback, and remote temp cleanup agree | -| Fresh deploy | Build → deploy → request → poll → telemetry → verify → teardown succeeds | -| Upgrade deploy | Old resources are detected/removed; no mixed-version silent failure | -| Rollback | Previous pinned image can be restored without deleting user DBs/keys/config | -| Remote skill | Fresh download/cache teaches and validates the new contract | -| Published image | New Docker entrypoint imports and `/status` answers from the released tag | - -Suggested final scans (the migration file and explicitly approved compatibility -shim are the only temporary exceptions): - -```bash -git grep -I -i -n ventis -- . ':!VENTIS_TO_CANYONOS_RENAME.md' -git ls-files | rg -i 'ventis' -rg --hidden --no-ignore -i ventis \ - -g '!.git/**' -g '!VENTIS_TO_CANYONOS_RENAME.md' -find . -path './.git' -prune -o -iname '*ventis*' -print -``` - -## Rollback rules - -- Never make `latest` the only rollback reference; retain the previous image - digest and compatibility matrix. -- Stop the deploy before changing resource prefixes. Do not run old and new GCs - against one Redis state concurrently. -- Preserve user config, `.env`, SSH keys, SQLite databases, and OTEL data. Rename - or copy user-owned files only on explicit user action. -- Keep cleanup exact and prefix-scoped; never broadly delete Docker or `/tmp` - state. -- If the new image fails, tear down only resources created by that attempt, - restore the previous pinned image, and use legacy env/header/resource support - until the failure is fixed. - -## Complete matching-file coverage - -The scan includes all old-name matches in these groups: - -- **Runtime (36):** the package Dockerfile; package README; exporter design/source; - package/controller initializers; root runtime CLI/server/stub generator; Local and - EC2 runtime/readme; deploy/future/global/instance/local controllers; env, - process-supervisor, Redis, session, and telemetry utilities; the LLM proxy README, - entrypoint, app/config/core/hooks/proxy, and all provider modules. -- **Tests (25):** `tests/README.md`, `run_tests.sh`, `test_canyonos_test.py`, - `test_cli.py`, `test_deploy.py`, `test_deploy_progress.py`, - `test_error_propagation.py`, `test_future.py`, every `test_global_controller_*`, - `test_gpu_metrics.py`, `test_instance_manager_runtime.py`, both - `test_local_controller_*`, both exporter tests, Redis/EC2/session/stub/telemetry - tests, and `test_ventis_context.py`. -- **Examples (15):** `examples/helloworld/README.md`; finance agent/config/workflow; - helloworld config/workflow; portfolio advisor/intent/metrics agents plus - config/workflow; text2sql generator/vLLM agents plus config/workflow. -- **Porting skill (7):** `SKILL.md`, `validate.py`, and the EC2, LLM proxy, - packaging, runtime-contract, and troubleshooting references. -- **Standalone CLI (7):** `cli/ARCHITECTURE.md`, `cli/cli.py`, and CanyonOS - dashboard compose, deploy, GC, init, and verify modules. -- **Root (4):** `.gitignore`, `README.md`, `pyproject.toml`, and `uv.lock`. - -No additional `setup.py`, `setup.cfg`, package manifest, Dockerfile, or compose -file contains the old token. `requirements.txt` has no project-name match. Binary -inspection found no embedded old token in the tracked SQLite/JPEG/PNG assets; the -PNG still requires a filename/reference rename because its basename is branded. diff --git a/ventis/llm_proxy/README.md b/ventis/llm_proxy/README.md deleted file mode 100644 index 873144d..0000000 --- a/ventis/llm_proxy/README.md +++ /dev/null @@ -1,112 +0,0 @@ -# llm_proxy - -A local, single-machine pass-through proxy for **OpenAI**, **Anthropic**, and -**Bedrock**. Callers keep their exact SDK calling convention — the only change is -one base-URL env var per provider. Every call flows through one function -(`core.proxy_request`) where token/metrics hooks fire. - -**Scope:** request/response ("call and return") only. Streaming is intentionally -not implemented yet. - -## How it works - -``` -your app (unchanged) localhost:8080 real upstream - openai SDK ─/openai/... ─┐ - anthropic SDK ─/anthropic/ ─┼─▶ proxy_request(ctx) ─▶ provider ─▶ api.openai.com - boto3 bedrock ─/bedrock/... ┘ (metrics hooks) adapter api.anthropic.com - bedrock-runtime..amazonaws.com -``` - -- **OpenAI / Anthropic** — straight HTTP reverse-proxy: rewrite host, swap in the - real key, forward with `requests`, return the response. -- **Bedrock** — re-issued through the proxy's own `boto3` client (handles SigV4 - signing + URL-encoding correctly). Only `invoke` is wired up. - -## Run - -```bash -pip install -r llm_proxy/requirements.txt - -# real upstream credentials live here; callers can use dummy keys -export OPENAI_API_KEY=sk-... -export ANTHROPIC_API_KEY=sk-ant-... -export AWS_REGION=us-east-1 # + normal AWS creds (env / ~/.aws / role) - -python -m llm_proxy # listens on 127.0.0.1:8080 -``` - -## Point your SDKs at it - -No code changes — just env vars: - -```bash -export OPENAI_BASE_URL=http://localhost:8080/openai/v1 -export ANTHROPIC_BASE_URL=http://localhost:8080/anthropic -export AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://localhost:8080/bedrock -``` - -Then your existing code works unchanged: - -```python -from openai import OpenAI -OpenAI().chat.completions.create(model="gpt-4o-mini", - messages=[{"role": "user", "content": "hi"}]) - -from anthropic import Anthropic -Anthropic().messages.create(model="claude-3-5-sonnet-20241022", max_tokens=64, - messages=[{"role": "user", "content": "hi"}]) - -import boto3, json -boto3.client("bedrock-runtime").invoke_model( - modelId="anthropic.claude-3-5-sonnet-20240620-v1:0", - body=json.dumps({"anthropic_version": "bedrock-2023-05-31", - "max_tokens": 64, - "messages": [{"role": "user", "content": "hi"}]})) -``` - -## Configuration (env vars) - -| Var | Default | Purpose | -|---|---|---| -| `PROXY_HOST` / `PROXY_PORT` | `127.0.0.1` / `8080` | where the proxy listens | -| `PROXY_CONNECT_TIMEOUT` / `PROXY_READ_TIMEOUT` | `10` / `600` | upstream timeouts (s) | -| `OPENAI_API_KEY` / `ANTHROPIC_API_KEY` | — | real upstream keys the proxy injects | -| `OPENAI_UPSTREAM_BASE` / `ANTHROPIC_UPSTREAM_BASE` | official APIs | override upstream (e.g. Azure/gateway) | -| `BEDROCK_REGION` (or `AWS_REGION`) | `us-east-1` | Bedrock region | -| `BEDROCK_UPSTREAM_HOST` | `bedrock-runtime..amazonaws.com` | override Bedrock host | - -## Telemetry & Metrics - -**Automatic telemetry is currently Bedrock-only.** The proxy captures: -- Model ID -- Input/output/total token counts -- Cache tokens (read & write) -- Error status - -Telemetry is automatically written to Redis under `future:` keys. - -### How it works (Bedrock only) - -1. **Auto-injection:** boto3 hook (`proxy.py`) injects `X-Ventis-Future-Id` header from thread-local context -2. **Token extraction:** `hooks.py` parses response `usage` field -3. **Redis write:** All metrics written to `future:` hash - -### Why Bedrock-only? - -OpenAI and Anthropic use their own Python SDKs (`openai`, `anthropic`), not boto3. -The boto3 event hook doesn't fire for non-AWS SDKs. To add telemetry for those: -- Would need separate hooks in each SDK's HTTP client -- Or callers would need to use proxy directly (not through SDKs) - -The proxy *forwards* OpenAI/Anthropic requests and *can* extract tokens, but doesn't -automatically inject headers or write telemetry. - -## Limitations - -- **No streaming.** `stream=True` / `invoke-with-response-stream` are not handled. -- **Bedrock error bodies are reconstructed**, not passed through byte-for-byte - (boto3 raises on 4xx/5xx; we rebuild a JSON body with the real status + - message). OpenAI/Anthropic errors pass through unchanged. -- **Dev server.** Runs on Flask's built-in server — fine for a local proxy, not - meant for production traffic. diff --git a/ventis/llm_proxy/__init__.py b/ventis/llm_proxy/__init__.py deleted file mode 100644 index 827377e..0000000 --- a/ventis/llm_proxy/__init__.py +++ /dev/null @@ -1,14 +0,0 @@ -"""Local LLM proxy. - -A transparent, single-machine pass-through for OpenAI, Anthropic, and Bedrock. -Point each provider's SDK at this service via its base-URL env var and calls flow -through one choke point (``llm_proxy.core.proxy_request``) where request/response -metrics hooks fire. - -Scope: request/response ("call and return") only. Streaming is intentionally -not implemented yet. -""" - -__all__ = ["__version__"] - -__version__ = "0.1.0" diff --git a/ventis/llm_proxy/__main__.py b/ventis/llm_proxy/__main__.py deleted file mode 100644 index c0af2e9..0000000 --- a/ventis/llm_proxy/__main__.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Entry point: ``python -m llm_proxy``.""" - -from __future__ import annotations - -import logging - -from ventis.llm_proxy.app import create_app -from ventis.llm_proxy.config import Config - - -def main() -> None: - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s", - ) - cfg = Config.from_env() - app = create_app(cfg) - logging.getLogger("llm_proxy").info( - "llm_proxy on http://%s:%d (openai=%s, anthropic=%s, bedrock=%s [%s])", - cfg.host, cfg.port, cfg.openai.upstream_base, cfg.anthropic.upstream_base, - cfg.bedrock_upstream_host, cfg.bedrock_region, - ) - # threaded so concurrent callers don't serialize; dev server is fine for a - # local proxy. - app.run(host=cfg.host, port=cfg.port, threaded=True) - - -if __name__ == "__main__": - main() diff --git a/ventis/llm_proxy/app.py b/ventis/llm_proxy/app.py deleted file mode 100644 index b2e3b09..0000000 --- a/ventis/llm_proxy/app.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Flask app: one catch-all route per provider prefix, all funneled through -``proxy_request``.""" - -from __future__ import annotations - -import logging - -from flask import Flask, jsonify, request - -from ventis.llm_proxy.config import Config -from ventis.llm_proxy.core import proxy_request -from ventis.llm_proxy.providers import build_registry - -log = logging.getLogger("llm_proxy") - -ALL_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE"] - - -def create_app(cfg: Config = None) -> Flask: - cfg = cfg or Config.from_env() - app = Flask(__name__) - registry = build_registry(cfg) - - # Initialize hooks with config for Redis - from ventis.llm_proxy import hooks as hooks_module - hooks_module.hooks = hooks_module.Hooks(cfg) - - @app.route("/healthz", methods=["GET"]) - def healthz(): - return jsonify(status="ok", providers=sorted(registry.keys())) - - @app.route("//", methods=ALL_METHODS) - def dispatch(provider, subpath): - prov = registry.get(provider) - if prov is None: - return ( - jsonify(error=f"unknown provider '{provider}'", known=sorted(registry.keys())), - 404, - ) - try: - return proxy_request(prov, subpath, request) - except Exception as exc: # surface upstream/adapter errors as 502 - log.exception("proxy error for %s/%s", provider, subpath) - return jsonify(error="proxy_error", detail=str(exc)), 502 - - return app diff --git a/ventis/llm_proxy/config.py b/ventis/llm_proxy/config.py deleted file mode 100644 index 9e85cfa..0000000 --- a/ventis/llm_proxy/config.py +++ /dev/null @@ -1,66 +0,0 @@ -"""Configuration, read once from the environment at startup. - -The proxy holds the *real* upstream credentials; callers can send dummy keys. -""" - -from __future__ import annotations - -import os -from dataclasses import dataclass -from typing import Optional - - -@dataclass -class ProviderConfig: - upstream_base: str - api_key: Optional[str] = None - - -@dataclass -class Config: - host: str - port: int - connect_timeout: float - read_timeout: float - - openai: ProviderConfig - anthropic: ProviderConfig - - bedrock_region: str - bedrock_upstream_host: str - - redis_host: str - redis_port: int - - @classmethod - def from_env(cls) -> "Config": - region = ( - os.getenv("BEDROCK_REGION") - or os.getenv("AWS_REGION") - or os.getenv("AWS_DEFAULT_REGION") - or "us-east-1" - ) - return cls( - host=os.getenv("PROXY_HOST", "127.0.0.1"), - port=int(os.getenv("PROXY_PORT", "8080")), - connect_timeout=float(os.getenv("PROXY_CONNECT_TIMEOUT", "10")), - read_timeout=float(os.getenv("PROXY_READ_TIMEOUT", "600")), - openai=ProviderConfig( - upstream_base=os.getenv( - "OPENAI_UPSTREAM_BASE", "https://api.openai.com" - ).rstrip("/"), - api_key=os.getenv("OPENAI_API_KEY"), - ), - anthropic=ProviderConfig( - upstream_base=os.getenv( - "ANTHROPIC_UPSTREAM_BASE", "https://api.anthropic.com" - ).rstrip("/"), - api_key=os.getenv("ANTHROPIC_API_KEY"), - ), - bedrock_region=region, - bedrock_upstream_host=os.getenv( - "BEDROCK_UPSTREAM_HOST", f"bedrock-runtime.{region}.amazonaws.com" - ), - redis_host=os.getenv("VENTIS_REDIS_HOST", "localhost"), - redis_port=int(os.getenv("VENTIS_REDIS_PORT", "6379")), - ) diff --git a/ventis/llm_proxy/core.py b/ventis/llm_proxy/core.py deleted file mode 100644 index 6284e00..0000000 --- a/ventis/llm_proxy/core.py +++ /dev/null @@ -1,46 +0,0 @@ -"""The single choke point every proxied call flows through.""" - -from __future__ import annotations - -import json -import time -from typing import Optional - -from flask import Response - -from ventis.llm_proxy.hooks import Ctx - - -def _guess_model(body: bytes) -> Optional[str]: - """Best-effort model name from the JSON body, for logging/metrics. - - Never raises. Returns None for requests whose model isn't in the body - (e.g. Bedrock, where it's in the path and already shown via the subpath). - """ - try: - model = json.loads(body).get("model") - return model if isinstance(model, str) else None - except Exception: - return None - - -def proxy_request(provider, subpath, flask_request): - # Import hooks here to get the instance created by create_app - from ventis.llm_proxy.hooks import hooks - - body = flask_request.get_data() - ctx = Ctx( - provider=provider.name, - method=flask_request.method, - subpath=subpath, - body=body, - headers=dict(flask_request.headers), - t0=time.monotonic(), - model=_guess_model(body), - ) - hooks.on_request(ctx) - - pr = provider.forward(flask_request, subpath, body) - - hooks.on_response(ctx, pr) - return Response(pr.content, status=pr.status, headers=pr.headers) diff --git a/ventis/llm_proxy/hooks.py b/ventis/llm_proxy/hooks.py deleted file mode 100644 index 7cc9957..0000000 --- a/ventis/llm_proxy/hooks.py +++ /dev/null @@ -1,159 +0,0 @@ -"""The metrics seam. - -Every proxied call passes through ``on_request`` / ``on_response``. Today these -only log. Token accounting lands here later: because the whole response is -buffered, usage extraction is a one-liner, e.g. ``resp.json().get("usage")`` for -OpenAI/Anthropic (Bedrock's usage lives in its per-model response body). -""" - -from __future__ import annotations - -import json -import logging -import time -from dataclasses import dataclass -from typing import Any, Dict, Optional - -log = logging.getLogger("llm_proxy") - - -@dataclass -class TokenUsage: - """Token usage extracted from LLM responses.""" - input_tokens: int = 0 - output_tokens: int = 0 - total_tokens: int = 0 - input_cache_tokens: int = 0 - input_cache_write_tokens: int = 0 - - def __repr__(self): - parts = [f"in={self.input_tokens}", f"out={self.output_tokens}"] - if self.input_cache_tokens: - parts.append(f"cache_read={self.input_cache_tokens}") - if self.input_cache_write_tokens: - parts.append(f"cache_write={self.input_cache_write_tokens}") - return f"TokenUsage({', '.join(parts)})" - - -@dataclass -class Ctx: - provider: str - method: str - subpath: str - body: bytes - headers: Dict[str, str] - t0: float - model: Optional[str] = None - - def elapsed_ms(self) -> float: - return (time.monotonic() - self.t0) * 1000.0 - - -class Hooks: - def __init__(self, config=None): - self.config = config - self._redis = None - - if config: - try: - try: - from ventis.controller.utils.redis_client import RedisClient - except ImportError: - # In-container the framework files are copied flat to /app. - from redis_client import RedisClient - self._redis = RedisClient( - host=config.redis_host, - port=config.redis_port, - ) - log.info("Redis telemetry enabled: %s:%s", config.redis_host, config.redis_port) - except Exception as e: - log.warning("Redis not available: %s", e) - - def on_request(self, ctx: Ctx) -> None: - log.info( - "→ %s %s /%s model=%s (%d bytes)", - ctx.provider, ctx.method, ctx.subpath, ctx.model, len(ctx.body), - ) - - def on_response(self, ctx: Ctx, resp: Any) -> None: - # Extract tokens for Bedrock - usage = None - if ctx.provider == "bedrock": - usage = self._extract_bedrock_tokens(resp) - - log.info( - "← %s %s /%s -> %s in %.0fms | %s", - ctx.provider, ctx.method, ctx.subpath, - getattr(resp, "status", "?"), ctx.elapsed_ms(), - usage or "no usage" - ) - - # Write to Redis if we have context - log.info("Checking telemetry write: redis=%s", "yes" if self._redis else "no") - if self._redis: - future_id = ctx.headers.get("X-Ventis-Future-Id") - log.info("Future ID from headers: %s", future_id) - if future_id: - try: - # Extract model ID - model_id = self._extract_model_id(ctx) - - is_error = resp.status >= 400 - - # Build telemetry data - data = { - "model": model_id, - "errors": "1" if is_error else "0", - } - - # Add token data if available - if usage: - data.update({ - "input_token_count": str(usage.input_tokens), - "output_token_count": str(usage.output_tokens), - "token_count": str(usage.total_tokens), - "input_cache_tokens": str(usage.input_cache_tokens), - "input_cache_write_tokens": str(usage.input_cache_write_tokens), - }) - - self._redis.hset_multiple(f"future:{future_id}", data) - log.info("Wrote telemetry to future:%s with data: %s", future_id, data) - except Exception as e: - log.error("Failed to write telemetry: %s", e) - - def _extract_model_id(self, ctx: Ctx) -> str: - """Extract model ID from context or subpath.""" - if ctx.model: - return ctx.model - - # For Bedrock: subpath is "model//operation" - # Use rpartition to peel operation off the right (same as provider logic) - if ctx.provider == "bedrock" and ctx.subpath.startswith("model/"): - model_id, sep, op = ctx.subpath[len("model/"):].rpartition("/") - if sep: # Found a separator - return model_id - - return "unknown" - - def _extract_bedrock_tokens(self, resp: Any) -> Optional[TokenUsage]: - """Extract token usage from Bedrock response. It requires diff logic from OpenAI/Anthropic""" - if resp.status != 200: - return None - - try: - data = json.loads(resp.content.decode("utf-8")) - usage = data.get("usage", {}) - if usage: - return TokenUsage( - input_tokens=usage.get("inputTokens", 0), - output_tokens=usage.get("outputTokens", 0), - total_tokens=usage.get("totalTokens", 0), - input_cache_tokens=usage.get("cacheReadInputTokens", 0), - input_cache_write_tokens=usage.get("cacheCreationInputTokens", 0), - ) - except: - pass - return None - - -hooks = Hooks() diff --git a/ventis/llm_proxy/providers/__init__.py b/ventis/llm_proxy/providers/__init__.py deleted file mode 100644 index d9ff021..0000000 --- a/ventis/llm_proxy/providers/__init__.py +++ /dev/null @@ -1,14 +0,0 @@ -from __future__ import annotations - -from ventis.llm_proxy.providers.anthropic import AnthropicProvider -from ventis.llm_proxy.providers.bedrock import BedrockProvider -from ventis.llm_proxy.providers.openai import OpenAIProvider - - -def build_registry(cfg): - """Map the URL prefix -> provider instance.""" - return { - "openai": OpenAIProvider(cfg), - "anthropic": AnthropicProvider(cfg), - "bedrock": BedrockProvider(cfg), - } diff --git a/ventis/llm_proxy/providers/anthropic.py b/ventis/llm_proxy/providers/anthropic.py deleted file mode 100644 index 33e14aa..0000000 --- a/ventis/llm_proxy/providers/anthropic.py +++ /dev/null @@ -1,19 +0,0 @@ -from __future__ import annotations - -from ventis.llm_proxy.providers.base import HttpProvider, UpstreamRequest, client_headers - - -class AnthropicProvider(HttpProvider): - name = "anthropic" - - def target(self, req, subpath, body): - headers = client_headers(req, drop=["x-api-key", "authorization"]) - if self.cfg.anthropic.api_key: - headers["x-api-key"] = self.cfg.anthropic.api_key - # `anthropic-version` is supplied by the SDK and passes through untouched. - return UpstreamRequest( - method=req.method, - url=f"{self.cfg.anthropic.upstream_base}/{subpath}", - headers=headers, - params=req.args.to_dict(flat=True), - ) diff --git a/ventis/llm_proxy/providers/base.py b/ventis/llm_proxy/providers/base.py deleted file mode 100644 index ef5db41..0000000 --- a/ventis/llm_proxy/providers/base.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Provider abstraction and shared HTTP plumbing. - -A provider's only job is to take the incoming request and produce a -``ProxyResponse``. Straight HTTP reverse-proxy providers (OpenAI, Anthropic) -subclass ``HttpProvider`` and just describe the upstream target; Bedrock owns -its own ``forward`` because it re-issues through boto3. -""" - -from __future__ import annotations - -import json -from dataclasses import dataclass, field -from typing import Dict, Iterable, List, Tuple - -import requests - -# Request headers we never forward: hop-by-hop (RFC 7230), ones we rewrite, and -# accept-encoding (we let the HTTP client negotiate + decode, then re-frame the -# response ourselves). -DROP_REQUEST_HEADERS = { - "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", - "te", "trailers", "transfer-encoding", "upgrade", - "host", "content-length", "accept-encoding", -} - -# Response headers we drop: we return already-decoded content and let the WSGI -# layer recompute framing headers. -DROP_RESPONSE_HEADERS = { - "content-encoding", "content-length", "transfer-encoding", - "connection", "keep-alive", -} - - -@dataclass -class UpstreamRequest: - method: str - url: str - headers: Dict[str, str] - params: Dict[str, str] = field(default_factory=dict) - - -@dataclass -class ProxyResponse: - status: int - headers: List[Tuple[str, str]] - content: bytes - - def json(self): - return json.loads(self.content.decode("utf-8")) - - -def client_headers(incoming, drop: Iterable[str] = ()) -> Dict[str, str]: - """Copy the caller's headers minus the ones we must not forward.""" - extra = {d.lower() for d in drop} - return { - k: v - for k, v in incoming.headers.items() - if k.lower() not in DROP_REQUEST_HEADERS and k.lower() not in extra - } - - -def filter_response_headers(headers) -> List[Tuple[str, str]]: - return [(k, v) for k, v in headers.items() if k.lower() not in DROP_RESPONSE_HEADERS] - - -class Provider: - name = "base" - - def __init__(self, cfg): - self.cfg = cfg - - def forward(self, req, subpath: str, body: bytes) -> ProxyResponse: - raise NotImplementedError - - -class HttpProvider(Provider): - """Providers that are a straight HTTP reverse-proxy (OpenAI, Anthropic).""" - - def target(self, req, subpath: str, body: bytes) -> UpstreamRequest: - raise NotImplementedError - - def forward(self, req, subpath, body): - up = self.target(req, subpath, body) - resp = requests.request( - up.method, - up.url, - headers=up.headers, - params=up.params, - data=body, - timeout=(self.cfg.connect_timeout, self.cfg.read_timeout), - ) - return ProxyResponse( - status=resp.status_code, - headers=filter_response_headers(resp.headers), - content=resp.content, - ) diff --git a/ventis/llm_proxy/providers/bedrock.py b/ventis/llm_proxy/providers/bedrock.py deleted file mode 100644 index f0efddc..0000000 --- a/ventis/llm_proxy/providers/bedrock.py +++ /dev/null @@ -1,119 +0,0 @@ -"""Bedrock adapter. - -Rather than re-sign the caller's SigV4 request (fiddly once model IDs contain -``:`` and ``/``), we re-issue the call through the proxy's own boto3 client, -which handles signing and URL-encoding correctly by construction. This is clean -for request/response; streaming (``invoke-with-response-stream``) is out of scope -for now. -""" - -from __future__ import annotations - -import json - -import boto3 -from botocore.exceptions import ClientError - -from ventis.llm_proxy.providers.base import Provider, ProxyResponse - -# bedrock-runtime operations that can appear as the last path segment; only the -# non-streaming "invoke" is wired up for now. -_SUPPORTED_OPS = {"invoke", "invoke-with-response-stream", "converse", "converse-stream"} - - -class BedrockProvider(Provider): - name = "bedrock" - - def __init__(self, cfg): - super().__init__(cfg) - # Explicitly set endpoint_url to bypass AWS_ENDPOINT_URL_BEDROCK_RUNTIME - # environment variable that points to this proxy (would create infinite loop) - self._client = boto3.client( - "bedrock-runtime", - region_name=cfg.bedrock_region, - endpoint_url=f"https://{cfg.bedrock_upstream_host}" - ) - - def forward(self, req, subpath, body): - model_id, op = self._parse(subpath) - - try: - if op == "invoke": - resp = self._client.invoke_model( - modelId=model_id, - body=body, - contentType=req.headers.get("Content-Type", "application/json"), - accept=req.headers.get("Accept", "application/json"), - ) - # For invoke, return raw response body - payload = resp["body"].read() - status = resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 200) - headers = [("Content-Type", resp.get("contentType", "application/json"))] - return ProxyResponse(status=status, headers=headers, content=payload) - - elif op == "converse": - params = json.loads(body) - params["modelId"] = model_id - resp = self._client.converse(**params) - - # Return response as JSON - response_data = { - "output": resp.get("output", {}), - "stopReason": resp.get("stopReason"), - "usage": resp.get("usage", {}), - } - # Include optional fields if present - for field in ["metrics", "trace", "additionalModelResponseFields"]: - if field in resp: - response_data[field] = resp[field] - - payload = json.dumps(response_data).encode("utf-8") - status = resp.get("ResponseMetadata", {}).get("HTTPStatusCode", 200) - return ProxyResponse( - status=status, - headers=[("Content-Type", "application/json")], - content=payload - ) - else: - raise NotImplementedError( - f"bedrock op '{op}' not supported (only invoke and converse)" - ) - - except ClientError as exc: - return self._error_response(exc) - except (json.JSONDecodeError, KeyError) as exc: - return ProxyResponse( - status=400, - headers=[("Content-Type", "application/json")], - content=json.dumps({"message": f"Invalid request: {exc}"}).encode(), - ) - - - - @staticmethod - def _parse(subpath): - # subpath looks like "model//"; the modelId may itself - # contain "/" (inference-profile ARNs), so peel the op off the right. - if not subpath.startswith("model/"): - raise ValueError(f"unrecognized bedrock path: /{subpath}") - model_id, sep, op = subpath[len("model/"):].rpartition("/") - if not sep or op not in _SUPPORTED_OPS: - raise ValueError(f"unrecognized bedrock path: /{subpath}") - return model_id, op - - @staticmethod - def _error_response(exc: ClientError) -> ProxyResponse: - # boto3 raises on 4xx/5xx; reconstruct a JSON error body carrying the - # real status + message. (Byte-for-byte error passthrough is a property - # only the HTTP providers have; this is the cost of re-issuing via boto3.) - meta = exc.response.get("ResponseMetadata", {}) - err = exc.response.get("Error", {}) - status = meta.get("HTTPStatusCode", 500) - body = json.dumps( - {"message": err.get("Message", str(exc)), "code": err.get("Code")} - ).encode("utf-8") - return ProxyResponse( - status=status, - headers=[("Content-Type", "application/json")], - content=body, - ) diff --git a/ventis/llm_proxy/providers/openai.py b/ventis/llm_proxy/providers/openai.py deleted file mode 100644 index 67457eb..0000000 --- a/ventis/llm_proxy/providers/openai.py +++ /dev/null @@ -1,18 +0,0 @@ -from __future__ import annotations - -from ventis.llm_proxy.providers.base import HttpProvider, UpstreamRequest, client_headers - - -class OpenAIProvider(HttpProvider): - name = "openai" - - def target(self, req, subpath, body): - headers = client_headers(req, drop=["authorization"]) - if self.cfg.openai.api_key: - headers["Authorization"] = f"Bearer {self.cfg.openai.api_key}" - return UpstreamRequest( - method=req.method, - url=f"{self.cfg.openai.upstream_base}/{subpath}", - headers=headers, - params=req.args.to_dict(flat=True), - ) diff --git a/ventis/llm_proxy/proxy.py b/ventis/llm_proxy/proxy.py deleted file mode 100644 index ec0956d..0000000 --- a/ventis/llm_proxy/proxy.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Auto-inject Ventis headers into ALL boto3 Bedrock calls. - -Import this module once and all subsequent boto3.client("bedrock-runtime") calls -will automatically include the X-Ventis-Future-ID header. - -Usage: - import ventis.llm_proxy_auto # Just import once - import boto3 - - # Now this automatically includes the header! - client = boto3.client("bedrock-runtime") - response = client.converse(...) -""" - -import boto3 -import logging - -try: - import ventis.controller.ventis_context as ventis_context -except ImportError: - # In-container the framework files are copied flat to /app. - try: - import ventis_context - except ImportError: - ventis_context = None - -log = logging.getLogger(__name__) - - -def _inject_ventis_headers(params=None, **kwargs): - """Inject X-Ventis-Future-ID into the outgoing Bedrock HTTP request. - - Registered on boto3's ``before-call.bedrock-runtime`` event, whose handlers - receive the prepared-request ``params`` dict (with a mutable ``headers``). - The ``request`` object only exists on the later ``before-send`` event, so - reading it here would always be None and silently drop the header. - """ - if not ventis_context or params is None: - return - - # Get current future_id from thread-local context - try: - future_id = ventis_context.get_current_future_id() - if future_id: - params.setdefault("headers", {})["X-Ventis-Future-ID"] = future_id - log.debug("Injected X-Ventis-Future-ID: %s", future_id) - except Exception as e: - log.debug("Could not inject future_id: %s", e) - - -# Register the hook globally on the default session -_session = boto3.Session() -_session.events.register_first('before-call.bedrock-runtime', _inject_ventis_headers) - -# Also patch the default session used by boto3.client() -boto3.DEFAULT_SESSION = _session - -log.info("Ventis boto3 hook registered - all Bedrock calls will include future_id header") diff --git a/ventis/llm_proxy/requirements.txt b/ventis/llm_proxy/requirements.txt deleted file mode 100644 index 2f7091c..0000000 --- a/ventis/llm_proxy/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -flask>=2.0 -requests>=2.28 -boto3>=1.28