From 52245ab1c0bf2da5266d8abed9fe3038d42dc4ad Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 15:55:55 -0800 Subject: [PATCH 1/5] chore(openrouter): add dedicated Python SDK tracing Co-Authored-By: Claude Code --- .github/workflows/openrouter-python-tests.yml | 69 +++++ .release-please-manifest.json | 3 +- .../tests/test_openai_compatible_spans.py | 3 + .../openai/traceai_openai/_providers.py | 1 + python/frameworks/openrouter/CHANGELOG.md | 5 + python/frameworks/openrouter/README.md | 32 +++ python/frameworks/openrouter/pyproject.toml | 26 ++ .../openrouter/tests/test_instrumentation.py | 195 +++++++++++++ .../openrouter/traceai_openrouter/__init__.py | 264 ++++++++++++++++++ .../openrouter/traceai_openrouter/version.py | 1 + release-please-config.json | 8 + 11 files changed, 606 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/openrouter-python-tests.yml create mode 100644 python/frameworks/openrouter/CHANGELOG.md create mode 100644 python/frameworks/openrouter/README.md create mode 100644 python/frameworks/openrouter/pyproject.toml create mode 100644 python/frameworks/openrouter/tests/test_instrumentation.py create mode 100644 python/frameworks/openrouter/traceai_openrouter/__init__.py create mode 100644 python/frameworks/openrouter/traceai_openrouter/version.py diff --git a/.github/workflows/openrouter-python-tests.yml b/.github/workflows/openrouter-python-tests.yml new file mode 100644 index 000000000..5958a2815 --- /dev/null +++ b/.github/workflows/openrouter-python-tests.yml @@ -0,0 +1,69 @@ +name: OpenRouter Python SDK contracts + +on: + pull_request: + branches: [main, dev] + paths: + - 'python/frameworks/openrouter/**' + - 'python/frameworks/openai/**' + - 'python/fi_instrumentation/**' + - '.github/workflows/openrouter-python-tests.yml' + push: + branches: [dev] + paths: + - 'python/frameworks/openrouter/**' + - 'python/frameworks/openai/**' + - 'python/fi_instrumentation/**' + - '.github/workflows/openrouter-python-tests.yml' + +permissions: + contents: read + +jobs: + sdk-contract: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + - python: '3.10' + openrouter: '1.3.0' + artifact: wheel + - python: '3.13' + openrouter: '1.3.107' + artifact: sdist + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: ${{ matrix.python }} + - name: Build and install packages + shell: bash + env: + ARTIFACT: ${{ matrix.artifact }} + OPENROUTER_VERSION: ${{ matrix.openrouter }} + run: | + python -m venv .openrouter-test-venv + .openrouter-test-venv/bin/pip install build + .openrouter-test-venv/bin/python -m build "--$ARTIFACT" python/frameworks/openrouter + .openrouter-test-venv/bin/python -m build "--$ARTIFACT" python/frameworks/openai + if [ "$ARTIFACT" = wheel ]; then + packages=(python/frameworks/{openrouter,openai}/dist/*.whl) + else + packages=(python/frameworks/{openrouter,openai}/dist/*.tar.gz) + fi + .openrouter-test-venv/bin/pip install "${packages[@]}" \ + "openrouter==$OPENROUTER_VERSION" pytest pytest-asyncio + - name: Run SDK and attribution tests without external traffic + shell: bash + run: | + sudo unshare --net bash -c ' + set -e + ip link set lo up + "$1" -I -m pytest python/frameworks/openrouter/tests \ + python/frameworks/openai/tests/test_openai_compatible_spans.py \ + -q -p no:cacheprovider + ' _ "$PWD/.openrouter-test-venv/bin/python" diff --git a/.release-please-manifest.json b/.release-please-manifest.json index c87a9bd44..57ea875e6 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -129,5 +129,6 @@ "python": "1.2.2", "typescript/packages/traceai_elevenlabs": "0.1.0", "go/bedrock": "0.1.0", - "typescript/packages/traceai_huggingface_transformers": "0.1.0" + "typescript/packages/traceai_huggingface_transformers": "0.1.0", + "python/frameworks/openrouter": "0.1.0" } diff --git a/python/frameworks/openai/tests/test_openai_compatible_spans.py b/python/frameworks/openai/tests/test_openai_compatible_spans.py index 19880502a..1f4a37d6c 100644 --- a/python/frameworks/openai/tests/test_openai_compatible_spans.py +++ b/python/frameworks/openai/tests/test_openai_compatible_spans.py @@ -199,6 +199,9 @@ async def run(): @pytest.mark.parametrize("base_url,provider", [ ("https://api.openai.com/v1", "openai"), + ("https://openrouter.ai/api/v1", "openrouter"), + ("https://OPENROUTER.AI./api/v1", "openrouter"), + ("https://openrouter.ai.customer.example/api/v1", "openai"), ("https://my-resource.openai.azure.com/openai/v1/", "azure"), ("https://my-resource.services.ai.azure.com/openai/v1/", "azure"), ("https://generativelanguage.googleapis.com/v1beta/openai/", "google"), diff --git a/python/frameworks/openai/traceai_openai/_providers.py b/python/frameworks/openai/traceai_openai/_providers.py index 2c426e46f..ef090c2ea 100644 --- a/python/frameworks/openai/traceai_openai/_providers.py +++ b/python/frameworks/openai/traceai_openai/_providers.py @@ -15,6 +15,7 @@ # Exact API hosts. _HOSTS = { "api.openai.com": FiLLMProviderValues.OPENAI.value, + "openrouter.ai": "openrouter", "ollama.com": "ollama", "generativelanguage.googleapis.com": FiLLMProviderValues.GOOGLE.value, "api.anthropic.com": FiLLMProviderValues.ANTHROPIC.value, diff --git a/python/frameworks/openrouter/CHANGELOG.md b/python/frameworks/openrouter/CHANGELOG.md new file mode 100644 index 000000000..74487a06e --- /dev/null +++ b/python/frameworks/openrouter/CHANGELOG.md @@ -0,0 +1,5 @@ +# Changelog + +## 0.1.0 + +- Instrument the dedicated OpenRouter Python SDK's chat, embeddings, and Responses methods, including asynchronous requests and stream completion. diff --git a/python/frameworks/openrouter/README.md b/python/frameworks/openrouter/README.md new file mode 100644 index 000000000..b77dfa9e5 --- /dev/null +++ b/python/frameworks/openrouter/README.md @@ -0,0 +1,32 @@ +# OpenRouter Python SDK tracing + +Trace calls made through the dedicated [`openrouter` Python SDK](https://pypi.org/project/openrouter/) with the maintained `traceAI-openrouter` package. For the OpenAI-compatible client, use `traceAI-openai` instead; do not instrument the same request with both packages or HTTPX tracing. + +```bash +pip install traceAI-openrouter +``` + +Set `FI_API_KEY`, `FI_SECRET_KEY`, and `OPENROUTER_API_KEY` in your environment. Then: + +```python +import os + +from fi_instrumentation import register +from fi_instrumentation.fi_types import ProjectType +from openrouter import OpenRouter +from traceai_openrouter import OpenRouterInstrumentor + +provider = register(project_type=ProjectType.OBSERVE, project_name="openrouter-demo") +OpenRouterInstrumentor().instrument(tracer_provider=provider) +client = OpenRouter(api_key=os.environ["OPENROUTER_API_KEY"]) +answer = client.chat.send( + model="openai/gpt-4o-mini", + messages=[{"role": "user", "content": "What is the capital of Japan?"}], +) +print(answer.choices[0].message.content) +provider.force_flush() +``` + +Find the `openrouter.chat` span in your Future AGI project. The adapter covers synchronous and asynchronous Chat, Embeddings, Responses and deprecated `beta.responses` calls. Streams finish when fully consumed or closed; final usage is available only when the server sends a usage chunk. For early exits, close the stream, or use a context manager. + +The default TraceConfig captures textual chat inputs and outputs, capped at 32 KiB. Embedding vectors and input content, image payloads, exception messages and stacks, and authorization headers are never included. Pass `TraceConfig(hide_inputs=True, hide_outputs=True)` to `instrument(config=...)` to remove prompt/response content. `FI_HIDE_INPUTS` and `FI_HIDE_OUTPUTS` environment flags also apply. The package does not send requests to OpenRouter on its own. diff --git a/python/frameworks/openrouter/pyproject.toml b/python/frameworks/openrouter/pyproject.toml new file mode 100644 index 000000000..120a94753 --- /dev/null +++ b/python/frameworks/openrouter/pyproject.toml @@ -0,0 +1,26 @@ +[tool.poetry] +name = "traceAI-openrouter" +version = "0.1.0" +description = "OpenTelemetry instrumentation for the dedicated OpenRouter Python SDK" +authors = ["Future AGI "] +readme = "README.md" +license = "Apache-2.0" +packages = [{ include = "traceai_openrouter" }] +classifiers = ["Intended Audience :: Developers", "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Framework :: OpenTelemetry :: Instrumentations"] + +[tool.poetry.dependencies] +python = ">=3.10,<3.14" +fi-instrumentation-otel = ">=1.1.0" +openrouter = ">=1.3.0,<2" +wrapt = ">=1.15.0" + +[tool.poetry.group.dev.dependencies] +pytest = ">=7" +pytest-asyncio = ">=0.23" + +[tool.poetry.urls] +Repository = "https://github.com/future-agi/traceAI/tree/dev/python/frameworks/openrouter" + +[build-system] +requires = ["poetry-core"] +build-backend = "poetry.core.masonry.api" diff --git a/python/frameworks/openrouter/tests/test_instrumentation.py b/python/frameworks/openrouter/tests/test_instrumentation.py new file mode 100644 index 000000000..a84702e85 --- /dev/null +++ b/python/frameworks/openrouter/tests/test_instrumentation.py @@ -0,0 +1,195 @@ +"""Real OpenRouter SDK calls with synthetic responses and no external traffic.""" +import asyncio +import json + +import httpx +import pytest +from openrouter import OpenRouter +from fi_instrumentation import TraceConfig +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from traceai_openrouter import OpenRouterInstrumentor + + +@pytest.fixture +def spans(): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + instrumentor = OpenRouterInstrumentor() + instrumentor.instrument(tracer_provider=provider) + yield exporter + instrumentor.uninstrument() + provider.shutdown() + + +def sdk(handler): + transport = httpx.MockTransport(handler) + return OpenRouter(api_key="synthetic-key", client=httpx.Client(transport=transport), + async_client=httpx.AsyncClient(transport=transport)) + + +def chat_body(): + return {"id": "chat-1", "object": "chat.completion", "created": 1, + "model": "vendor/served", "system_fingerprint": None, "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": "Tokyo"}}], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}} + + +def attributes(spans): + finished = spans.get_finished_spans() + assert len(finished) == 1 + return dict(finished[0].attributes) + + +def test_chat_sync_records_model_usage_content_without_secret(spans): + seen = [] + def handle(req): + seen.append(req) + return httpx.Response(200, json=chat_body()) + client = sdk(handle) + result = client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + assert result.choices[0].message.content == "Tokyo" + a = attributes(spans) + assert len(seen) == 1 and seen[0].url.host == "openrouter.ai" + assert a["gen_ai.provider.name"] == "openrouter" + assert a["gen_ai.request.model"] == "vendor/request" + assert a["gen_ai.response.model"] == "vendor/served" + assert a["gen_ai.usage.input_tokens"] == 3 + assert a["gen_ai.usage.output_tokens"] == 2 + assert a["output.value"] == "Tokyo" + assert "synthetic-key" not in str(a) + + +def test_chat_async(spans): + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + result = asyncio.run(client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}])) + assert result.choices[0].message.content == "Tokyo" + assert attributes(spans)["gen_ai.usage.input_tokens"] == 3 + + +def test_error_does_not_export_response_or_key(spans): + client = sdk(lambda req: httpx.Response(401, json={"error": {"message": "synthetic-key is invalid"}})) + with pytest.raises(Exception): + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + a = attributes(spans) + assert a["error.type"] + assert "synthetic-key" not in str(a) + + +def test_chat_stream_finishes_after_consumption_with_final_usage(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}]} + for text in ("Tok", "yo")] + chunks.append({"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}}) + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + stream = client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + assert not spans.get_finished_spans() + assert list(stream) + a = attributes(spans) + assert a["output.value"] == "Tokyo" + assert a["gen_ai.usage.input_tokens"] == 3 + + +def test_embedding_has_usage_and_no_vector_content(spans): + client = sdk(lambda req: httpx.Response(200, json={ + "object": "list", "model": "vendor/embed", "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}], + "usage": {"prompt_tokens": 4, "total_tokens": 4}})) + answer = client.embeddings.generate(model="vendor/embed", input="secret input") + assert len(answer.data[0].embedding) == 2 + a = attributes(spans) + assert a["gen_ai.operation.name"] == "embeddings" + assert a["gen_ai.usage.input_tokens"] == 4 + assert "input.value" not in a and "output.value" not in a + + +def test_responses_and_beta_alias(spans): + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", + "output": [], "output_text": "Tokyo", "parallel_tool_calls": False, + "presence_penalty": None, "status": "completed", "temperature": None, + "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + client = sdk(lambda req: httpx.Response(200, json=body)) + assert client.responses.send(model="vendor/request", input="Japan?").output_text == "Tokyo" + assert client.beta.responses.send(model="vendor/request", input="Japan?").output_text == "Tokyo" + finished = spans.get_finished_spans() + assert len(finished) == 2 + assert all(s.attributes["gen_ai.provider.name"] == "openrouter" for s in finished) + assert all(s.attributes.get("gen_ai.usage.output_tokens") == 2 for s in finished), [dict(s.attributes) for s in finished] + + +def test_stream_early_close_marks_incomplete_without_claiming_final_text(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]}] + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + with client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) as stream: + next(stream) + a = attributes(spans) + assert a["openrouter.stream.incomplete"] is True + assert "output.value" not in a + + +def test_chat_async_stream_usage(spans): + chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tokyo"}, "finish_reason": None}]}, + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}}] + body = "".join("data: " + json.dumps(chunk) + "\n\n" for chunk in chunks) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + return [chunk async for chunk in stream] + assert len(asyncio.run(run())) == 2 + a = attributes(spans) + assert a["output.value"] == "Tokyo" + assert a["gen_ai.usage.input_tokens"] == 3 + + +def test_chat_async_close_ends_incomplete_span(spans): + chunk = {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]} + body = "data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())) + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}], stream=True) + await stream.__anext__() + await stream.close() + asyncio.run(run()) + assert attributes(spans)["openrouter.stream.incomplete"] is True + + +def test_hide_inputs_outputs(spans): + OpenRouterInstrumentor().uninstrument() + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + instrumentor = OpenRouterInstrumentor() + instrumentor.instrument(tracer_provider=provider, config=TraceConfig(hide_inputs=True, hide_outputs=True)) + try: + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "secret prompt"}]) + a = attributes(exporter) + assert "input.value" not in a and "output.value" not in a + assert a["gen_ai.usage.input_tokens"] == 3 + finally: + instrumentor.uninstrument() + provider.shutdown() + + +def test_uninstrument_restores_methods(spans): + instrumentor = OpenRouterInstrumentor() + instrumentor.uninstrument() + client = sdk(lambda req: httpx.Response(200, json=chat_body())) + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "Japan?"}]) + assert spans.get_finished_spans() == () diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py new file mode 100644 index 000000000..dacec1853 --- /dev/null +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -0,0 +1,264 @@ +"""Trace the dedicated Speakeasy-generated ``openrouter`` Python SDK. + +Instrument only inference endpoints. Do not instrument HTTPX or OpenAI alongside +this instrumentor for the same request; those paths would create duplicate spans. +""" +from __future__ import annotations + +import json +from importlib import import_module +from typing import Any, Collection + +from fi_instrumentation import FITracer, TraceConfig +from fi_instrumentation.fi_types import FiSpanKindValues, SpanAttributes as A +from opentelemetry import trace +from opentelemetry.instrumentation.instrumentor import BaseInstrumentor +from opentelemetry.trace import Status, StatusCode +from wrapt import wrap_function_wrapper + +from .version import __version__ + +_ENDPOINTS = ( + ("openrouter.chat", "Chat", "chat", "send"), + ("openrouter.embeddings", "Embeddings", "embeddings", "generate"), + ("openrouter.responses", "Responses", "responses", "send"), + ("openrouter.beta_responses", "BetaResponses", "responses", "send"), +) +_LIMIT = 32768 + + +def _field(value: Any, name: str) -> Any: + if isinstance(value, dict): + return value.get(name) + return getattr(value, name, None) + + +def _text(value: Any) -> str: + """Capture bounded textual output only, never embedding vectors or images.""" + choices = _field(value, "choices") + if choices: + first = choices[0] + msg = _field(first, "delta") or _field(first, "message") + content = _field(msg, "content") + return content[:_LIMIT] if isinstance(content, str) else "" + output_text = _field(value, "output_text") + if isinstance(output_text, str) and output_text: + return output_text[:_LIMIT] + output = _field(value, "output") + if output: + parts = [] + for item in output: + for block in _field(item, "content") or []: + text = _field(block, "text") + if isinstance(text, str): + parts.append(text) + return "".join(parts)[:_LIMIT] + return "" + + +def _usage(span: Any, value: Any) -> None: + usage = _field(value, "usage") + if not usage: + return + for attr, names in ( + (A.GEN_AI_USAGE_INPUT_TOKENS, ("prompt_tokens", "input_tokens")), + (A.GEN_AI_USAGE_OUTPUT_TOKENS, ("completion_tokens", "output_tokens")), + ): + for name in names: + number = _field(usage, name) + if isinstance(number, int) and not isinstance(number, bool) and number >= 0: + span.set_attribute(attr, number) + break + + +def _result(span: Any, value: Any, content: bool) -> None: + model = _field(value, "model") + if isinstance(model, str): + span.set_attribute(A.GEN_AI_RESPONSE_MODEL, model) + _usage(span, value) + if content: + text = _text(value) + if text: + span.set_attribute(A.OUTPUT_VALUE, text) + + +def _input(span: Any, operation: str, kwargs: dict[str, Any], content: bool) -> None: + if not content: + return + value = kwargs.get("messages") if operation == "chat" else kwargs.get("input") + if value is None or operation == "embeddings": + return # Embedding input can contain arbitrary binary/image payloads. + try: + serialized = json.dumps(value, default=lambda obj: obj.model_dump(exclude_none=True)) + except (TypeError, ValueError, AttributeError): + return + span.set_attribute(A.INPUT_VALUE, serialized[:_LIMIT]) + span.set_attribute(A.INPUT_MIME_TYPE, "application/json") + + +def _failure(span: Any, exc: BaseException) -> None: + # Exception text and stack traces may contain the API key, request body or URL. + span.set_attribute("error.type", type(exc).__name__) + span.set_status(Status(StatusCode.ERROR)) + span.end() + + +class _Stream: + def __init__(self, source: Any, span: Any, content: bool): + self._source, self._span, self._content = source, span, content + self._ended = False + self._pieces: list[str] = [] + self._length = 0 + + def __getattr__(self, name: str) -> Any: + return getattr(self._source, name) + + def _observe(self, chunk: Any) -> None: + _result(self._span, chunk, False) + if self._content and self._length < _LIMIT: + text = _text(chunk)[: _LIMIT - self._length] + self._pieces.append(text) + self._length += len(text) + + def _finish(self, complete: bool = True) -> None: + if not self._ended: + self._ended = True + if not complete: + self._span.set_attribute("openrouter.stream.incomplete", True) + if self._pieces and complete: + self._span.set_attribute(A.OUTPUT_VALUE, "".join(self._pieces)) + self._span.end() + + def __iter__(self): + return self + + def __next__(self): + try: + chunk = next(self._source) + except StopIteration: + self._finish() + raise + except BaseException as exc: + self._ended = True + _failure(self._span, exc) + raise + self._observe(chunk) + return chunk + + def __aiter__(self): + return self + + async def __anext__(self): + try: + chunk = await self._source.__anext__() + except StopAsyncIteration: + self._finish() + raise + except BaseException as exc: + self._ended = True + _failure(self._span, exc) + raise + self._observe(chunk) + return chunk + + def close(self): + try: + return self._source.close() + finally: + self._finish(complete=False) + + def __enter__(self): + self._source.__enter__() + return self + + def __exit__(self, *args): + try: + return self._source.__exit__(*args) + finally: + self._finish(complete=False) + + async def __aenter__(self): + await self._source.__aenter__() + return self + + async def __aexit__(self, *args): + try: + return await self._source.__aexit__(*args) + finally: + self._finish(complete=False) + + +class _AsyncStream(_Stream): + async def close(self): + try: + return await self._source.close() + finally: + self._finish(complete=False) + + +def _start(tracer: Any, operation: str, kwargs: dict[str, Any], config: TraceConfig): + kind = FiSpanKindValues.EMBEDDING if operation == "embeddings" else FiSpanKindValues.LLM + model = kwargs.get("model") + attrs = {A.GEN_AI_PROVIDER_NAME: "openrouter", A.GEN_AI_OPERATION_NAME: operation} + if isinstance(model, str): + attrs[A.GEN_AI_REQUEST_MODEL] = model + span = tracer.start_span("openrouter." + operation, attributes=attrs, fi_span_kind=kind) + _input(span, operation, kwargs, not (config.hide_inputs or config.hide_input_text)) + return span + + +def _wrap(tracer: Any, operation: str, config: TraceConfig, is_async: bool): + def complete(span, result): + if hasattr(result, "__anext__"): + return _AsyncStream(result, span, not (config.hide_outputs or config.hide_output_text)) + if hasattr(result, "__next__"): + return _Stream(result, span, not (config.hide_outputs or config.hide_output_text)) + _result(span, result, not (config.hide_outputs or config.hide_output_text)) + span.end() + return result + + if is_async: + async def wrapped(wrapped, instance, args, kwargs): + span = _start(tracer, operation, kwargs, config) + try: + return complete(span, await wrapped(*args, **kwargs)) + except BaseException as exc: + _failure(span, exc) + raise + else: + def wrapped(wrapped, instance, args, kwargs): + span = _start(tracer, operation, kwargs, config) + try: + return complete(span, wrapped(*args, **kwargs)) + except BaseException as exc: + _failure(span, exc) + raise + return wrapped + + +class OpenRouterInstrumentor(BaseInstrumentor): # type: ignore[misc] + """Instrument dedicated OpenRouter chat, embeddings and Responses calls.""" + + def instrumentation_dependencies(self) -> Collection[str]: + return ("openrouter >= 1.3.0, < 2",) + + def _instrument(self, **kwargs: Any) -> None: + config = kwargs.get("config") or TraceConfig() + if not isinstance(config, TraceConfig): + raise TypeError("config must be a TraceConfig") + tracer = FITracer(trace.get_tracer(__name__, __version__, kwargs.get("tracer_provider")), config) + self._originals = [] + for module, cls_name, operation, method in _ENDPOINTS: + cls = getattr(import_module(module), cls_name) + for name, is_async in ((method, False), (method + "_async", True)): + original = cls.__dict__[name] + wrap_function_wrapper(module, cls_name + "." + name, _wrap(tracer, operation, config, is_async)) + self._originals.append((cls, name, original)) + + def _uninstrument(self, **kwargs: Any) -> None: + for cls, name, original in getattr(self, "_originals", []): + setattr(cls, name, original) + self._originals = [] + + +__all__ = ["OpenRouterInstrumentor", "__version__"] diff --git a/python/frameworks/openrouter/traceai_openrouter/version.py b/python/frameworks/openrouter/traceai_openrouter/version.py new file mode 100644 index 000000000..75716eb9a --- /dev/null +++ b/python/frameworks/openrouter/traceai_openrouter/version.py @@ -0,0 +1 @@ +__version__ = "0.1.0" # x-release-please-version diff --git a/release-please-config.json b/release-please-config.json index 01c088890..17de55ed5 100644 --- a/release-please-config.json +++ b/release-please-config.json @@ -1100,6 +1100,14 @@ "extra-files": [ "src/version.ts" ] + }, + "python/frameworks/openrouter": { + "release-type": "python", + "component": "traceai-openrouter", + "package-name": "traceAI-openrouter", + "extra-files": [ + "traceai_openrouter/version.py" + ] } } } From 49aa0cd693e57cb892674da6cd2cf5d9904064b2 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 15:58:24 -0800 Subject: [PATCH 2/5] fix(openrouter): capture Responses streaming final usage Co-Authored-By: Claude Code --- .../openrouter/tests/test_instrumentation.py | 21 ++++++++++++++++++ .../openrouter/traceai_openrouter/__init__.py | 22 ++++++++++++++++--- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/python/frameworks/openrouter/tests/test_instrumentation.py b/python/frameworks/openrouter/tests/test_instrumentation.py index a84702e85..4ae8a74ea 100644 --- a/python/frameworks/openrouter/tests/test_instrumentation.py +++ b/python/frameworks/openrouter/tests/test_instrumentation.py @@ -127,6 +127,27 @@ def test_responses_and_beta_alias(spans): assert all(s.attributes.get("gen_ai.usage.output_tokens") == 2 for s in finished), [dict(s.attributes) for s in finished] +def test_responses_stream_final_event_usage(spans): + response = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", + "output": [], "output_text": "Tokyo", "parallel_tool_calls": False, + "presence_penalty": None, "status": "completed", "temperature": None, + "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + event = {"type": "response.completed", "sequence_number": 1, "response": response} + payload = "data: " + json.dumps(event) + "\n\ndata: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode())) + stream = client.responses.send(model="vendor/request", input="Japan?", stream=True) + assert list(stream) + a = attributes(spans) + assert a["gen_ai.response.model"] == "vendor/served" + assert a["gen_ai.usage.output_tokens"] == 2 + assert a["output.value"] == "Tokyo" + + def test_stream_early_close_marks_incomplete_without_claiming_final_text(spans): chunks = [{"id": "chat-1", "object": "chat.completion.chunk", "created": 1, "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]}] diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py index dacec1853..1acc6e5f4 100644 --- a/python/frameworks/openrouter/traceai_openrouter/__init__.py +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -114,11 +114,27 @@ def __getattr__(self, name: str) -> Any: return getattr(self._source, name) def _observe(self, chunk: Any) -> None: + # Responses SSE events carry final usage on response.completed, not on + # the event itself. The final response supersedes individual deltas. + event_type = _field(chunk, "type") + completed = _field(chunk, "response") + if event_type in ("response.completed", "response.failed") and completed: + _result(self._span, completed, False) + if event_type == "response.failed": + self._span.set_status(Status(StatusCode.ERROR)) + if self._content: + text = _text(completed) + if text: + self._pieces = [text[:_LIMIT]] + self._length = len(self._pieces[0]) + return _result(self._span, chunk, False) if self._content and self._length < _LIMIT: - text = _text(chunk)[: _LIMIT - self._length] - self._pieces.append(text) - self._length += len(text) + text = _field(chunk, "delta") if event_type == "response.output_text.delta" else _text(chunk) + if isinstance(text, str): + text = text[: _LIMIT - self._length] + self._pieces.append(text) + self._length += len(text) def _finish(self, complete: bool = True) -> None: if not self._ended: From c4c4522d8211327edd19cba6bf9969e2f389160a Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 16:22:14 -0800 Subject: [PATCH 3/5] fix(openrouter): honor privacy flags and failed SDK responses Co-Authored-By: Claude Code --- python/frameworks/openrouter/README.md | 2 +- .../tests/test_review_regressions.py | 156 ++++++++++++++++++ .../openrouter/traceai_openrouter/__init__.py | 65 ++++++-- 3 files changed, 205 insertions(+), 18 deletions(-) create mode 100644 python/frameworks/openrouter/tests/test_review_regressions.py diff --git a/python/frameworks/openrouter/README.md b/python/frameworks/openrouter/README.md index b77dfa9e5..a14274def 100644 --- a/python/frameworks/openrouter/README.md +++ b/python/frameworks/openrouter/README.md @@ -29,4 +29,4 @@ provider.force_flush() Find the `openrouter.chat` span in your Future AGI project. The adapter covers synchronous and asynchronous Chat, Embeddings, Responses and deprecated `beta.responses` calls. Streams finish when fully consumed or closed; final usage is available only when the server sends a usage chunk. For early exits, close the stream, or use a context manager. -The default TraceConfig captures textual chat inputs and outputs, capped at 32 KiB. Embedding vectors and input content, image payloads, exception messages and stacks, and authorization headers are never included. Pass `TraceConfig(hide_inputs=True, hide_outputs=True)` to `instrument(config=...)` to remove prompt/response content. `FI_HIDE_INPUTS` and `FI_HIDE_OUTPUTS` environment flags also apply. The package does not send requests to OpenRouter on its own. +The default TraceConfig captures plain-text chat messages, plain-text Responses inputs and textual outputs, capped at 32 KiB. Structured or multimodal input, embedding vectors and content, image payloads, exception messages and stacks, and authorization headers are never included. Pass `TraceConfig(hide_inputs=True, hide_outputs=True)` to `instrument(config=...)` to remove prompt/response content. Directional message, text and image privacy flags also suppress the corresponding generic input/output attributes; hiding input images suppresses the entire input value because a text string can contain an image reference. The package does not send requests to OpenRouter on its own. diff --git a/python/frameworks/openrouter/tests/test_review_regressions.py b/python/frameworks/openrouter/tests/test_review_regressions.py new file mode 100644 index 000000000..a8f669453 --- /dev/null +++ b/python/frameworks/openrouter/tests/test_review_regressions.py @@ -0,0 +1,156 @@ +"""Privacy, in-band failures, and nested request context using the real SDK.""" +import asyncio +import json + +import httpx +import pytest +from fi_instrumentation import TraceConfig +from openrouter import OpenRouter +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode + +from traceai_openrouter import OpenRouterInstrumentor +from test_instrumentation import chat_body, sdk + + +@pytest.fixture +def instrument(): + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + inst = OpenRouterInstrumentor() + def start(config=None): + inst.instrument(tracer_provider=provider, config=config or TraceConfig()) + return provider, exporter + yield start + inst.uninstrument() + provider.shutdown() + + +@pytest.mark.parametrize("flag,content", [ + ("hide_input_messages", "SYNTHETIC_PRIVATE_MESSAGE"), + ("hide_input_text", "SYNTHETIC_PRIVATE_MESSAGE"), + ("hide_input_images", "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"), +]) +def test_chat_input_privacy(instrument, flag, content): + _, exporter = instrument(TraceConfig(**{flag: True})) + messages = [{"role": "user", "content": [ + {"type": "text", "text": "SYNTHETIC_PRIVATE_MESSAGE"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"}}, + ]}] + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=messages) + attrs = dict(exporter.get_finished_spans()[0].attributes) + assert content not in str(attrs) + assert "input.value" not in attrs + + +@pytest.mark.parametrize("flag", ["hide_output_messages", "hide_output_text"]) +def test_chat_output_privacy(instrument, flag): + _, exporter = instrument(TraceConfig(**{flag: True})) + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}]) + attrs = dict(exporter.get_finished_spans()[0].attributes) + assert "Tokyo" not in str(attrs) + assert "output.value" not in attrs + + +def chat_error_stream(): + chunks = [ + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tok"}, "finish_reason": None}]}, + {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [], "error": {"code": 502, "message": "SYNTHETIC_PRIVATE_ERROR"}}, + ] + return httpx.Response(200, headers={"content-type": "text/event-stream"}, + content=("".join("data: " + json.dumps(c) + "\n\n" for c in chunks) + "data: [DONE]\n\n").encode()) + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_in_band_stream_error_drops_partial_answer(instrument, async_call): + _, exporter = instrument() + client = sdk(lambda req: chat_error_stream()) + if async_call: + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True) + return [part async for part in stream] + chunks = asyncio.run(run()) + else: + chunks = list(client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True)) + assert len(chunks) == 2 + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("state", ["failed", "incomplete"]) +def test_responses_failure_drops_partial_answer(instrument, state): + _, exporter = instrument() + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": 2, + "error": {"code": "server_error", "message": "SYNTHETIC_PRIVATE_ERROR"} if state == "failed" else None, + "frequency_penalty": None, "incomplete_details": None, "instructions": None, + "metadata": None, "model": "vendor/served", "output": [], "output_text": "partial", + "parallel_tool_calls": False, "presence_penalty": None, "status": state, + "temperature": None, "tool_choice": "auto", "tools": [], "top_p": None, + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, "output_tokens_details": {"reasoning_tokens": 0}}} + result = sdk(lambda req: httpx.Response(200, json=body)).responses.send(model="vendor/request", input="hi") + assert result.status == state + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_http_child_nested_under_llm(instrument, async_call): + provider, exporter = instrument() + tracer = provider.get_tracer("test") + def handler(req): + with tracer.start_as_current_span("http.client"): + return httpx.Response(200, json=chat_body()) + client = sdk(handler) + with tracer.start_as_current_span("caller"): + if async_call: + asyncio.run(client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}])) + else: + client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}]) + spans = {span.name: span for span in exporter.get_finished_spans()} + assert spans["openrouter.chat"].parent.span_id == spans["caller"].context.span_id + assert spans["http.client"].parent.span_id == spans["openrouter.chat"].context.span_id + + +def test_multimodal_input_never_serializes_image_by_default(instrument): + _, exporter = instrument() + messages = [{"role": "user", "content": [ + {"type": "text", "text": "hello"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC_IMAGE_MARKER"}}, + ]}] + sdk(lambda req: httpx.Response(200, json=chat_body())).chat.send(model="vendor/request", messages=messages) + assert "input.value" not in exporter.get_finished_spans()[0].attributes + + +@pytest.mark.parametrize("async_call", [False, True]) +def test_stream_http_child_nested_under_llm(instrument, async_call): + provider, exporter = instrument() + tracer = provider.get_tracer("test") + chunk = {"id": "chat-1", "object": "chat.completion.chunk", "created": 1, + "model": "vendor/served", "choices": [{"index": 0, "delta": {"content": "Tokyo"}, "finish_reason": None}]} + payload = "data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n" + def handler(req): + with tracer.start_as_current_span("http.client"): + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode()) + client = sdk(handler) + with tracer.start_as_current_span("caller"): + if async_call: + async def run(): + stream = await client.chat.send_async(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True) + return [part async for part in stream] + asyncio.run(run()) + else: + list(client.chat.send(model="vendor/request", messages=[{"role": "user", "content": "hi"}], stream=True)) + spans = {span.name: span for span in exporter.get_finished_spans()} + assert spans["http.client"].parent.span_id == spans["openrouter.chat"].context.span_id + assert spans["openrouter.chat"].parent.span_id == spans["caller"].context.span_id diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py index 1acc6e5f4..c8e123dae 100644 --- a/python/frameworks/openrouter/traceai_openrouter/__init__.py +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -76,6 +76,11 @@ def _result(span: Any, value: Any, content: bool) -> None: if isinstance(model, str): span.set_attribute(A.GEN_AI_RESPONSE_MODEL, model) _usage(span, value) + if _field(value, "status") in ("failed", "incomplete"): + # A 200 Responses body can still represent a failed/incomplete generation. + # Never export its error message or partial output. + span.set_status(Status(StatusCode.ERROR)) + return if content: text = _text(value) if text: @@ -88,10 +93,21 @@ def _input(span: Any, operation: str, kwargs: dict[str, Any], content: bool) -> value = kwargs.get("messages") if operation == "chat" else kwargs.get("input") if value is None or operation == "embeddings": return # Embedding input can contain arbitrary binary/image payloads. - try: - serialized = json.dumps(value, default=lambda obj: obj.model_dump(exclude_none=True)) - except (TypeError, ValueError, AttributeError): + # Never serialize multimodal objects wholesale into input.value: the generic + # attribute cannot distinguish text from an inline image, URL or tool data. + # Keep only plain chat messages and plain text Responses inputs. + if operation == "chat": + if not isinstance(value, (list, tuple)) or any( + not isinstance(msg, dict) or not isinstance(msg.get("content"), str) + for msg in value + ): + return + value = [{"role": msg.get("role"), "content": msg["content"]} for msg in value] + elif not isinstance(value, str): + return + if "data:image/" in str(value).lower(): return + serialized = json.dumps(value) span.set_attribute(A.INPUT_VALUE, serialized[:_LIMIT]) span.set_attribute(A.INPUT_MIME_TYPE, "application/json") @@ -107,6 +123,7 @@ class _Stream: def __init__(self, source: Any, span: Any, content: bool): self._source, self._span, self._content = source, span, content self._ended = False + self._failed = False self._pieces: list[str] = [] self._length = 0 @@ -118,18 +135,20 @@ def _observe(self, chunk: Any) -> None: # the event itself. The final response supersedes individual deltas. event_type = _field(chunk, "type") completed = _field(chunk, "response") - if event_type in ("response.completed", "response.failed") and completed: + if _field(chunk, "error") is not None or event_type in ("response.failed", "response.incomplete"): + self._failed = True + self._pieces.clear() + self._span.set_status(Status(StatusCode.ERROR)) + if event_type in ("response.completed", "response.failed", "response.incomplete") and completed: _result(self._span, completed, False) - if event_type == "response.failed": - self._span.set_status(Status(StatusCode.ERROR)) - if self._content: + if not self._failed and self._content: text = _text(completed) if text: self._pieces = [text[:_LIMIT]] self._length = len(self._pieces[0]) return _result(self._span, chunk, False) - if self._content and self._length < _LIMIT: + if not self._failed and self._content and self._length < _LIMIT: text = _field(chunk, "delta") if event_type == "response.output_text.delta" else _text(chunk) if isinstance(text, str): text = text[: _LIMIT - self._length] @@ -141,7 +160,7 @@ def _finish(self, complete: bool = True) -> None: self._ended = True if not complete: self._span.set_attribute("openrouter.stream.incomplete", True) - if self._pieces and complete: + if self._pieces and complete and not self._failed: self._span.set_attribute(A.OUTPUT_VALUE, "".join(self._pieces)) self._span.end() @@ -150,7 +169,8 @@ def __iter__(self): def __next__(self): try: - chunk = next(self._source) + with trace.use_span(self._span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + chunk = next(self._source) except StopIteration: self._finish() raise @@ -166,7 +186,8 @@ def __aiter__(self): async def __anext__(self): try: - chunk = await self._source.__anext__() + with trace.use_span(self._span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + chunk = await self._source.__anext__() except StopAsyncIteration: self._finish() raise @@ -219,17 +240,23 @@ def _start(tracer: Any, operation: str, kwargs: dict[str, Any], config: TraceCon if isinstance(model, str): attrs[A.GEN_AI_REQUEST_MODEL] = model span = tracer.start_span("openrouter." + operation, attributes=attrs, fi_span_kind=kind) - _input(span, operation, kwargs, not (config.hide_inputs or config.hide_input_text)) + # Generic input.value bypasses message/image redaction, so omit it for + # every directional privacy setting (including images embedded in strings). + capture_input = not ( + config.hide_inputs or config.hide_input_text or config.hide_input_messages or config.hide_input_images + ) + _input(span, operation, kwargs, capture_input) return span def _wrap(tracer: Any, operation: str, config: TraceConfig, is_async: bool): def complete(span, result): + capture_output = not (config.hide_outputs or config.hide_output_text or config.hide_output_messages) if hasattr(result, "__anext__"): - return _AsyncStream(result, span, not (config.hide_outputs or config.hide_output_text)) + return _AsyncStream(result, span, capture_output) if hasattr(result, "__next__"): - return _Stream(result, span, not (config.hide_outputs or config.hide_output_text)) - _result(span, result, not (config.hide_outputs or config.hide_output_text)) + return _Stream(result, span, capture_output) + _result(span, result, capture_output) span.end() return result @@ -237,7 +264,9 @@ def complete(span, result): async def wrapped(wrapped, instance, args, kwargs): span = _start(tracer, operation, kwargs, config) try: - return complete(span, await wrapped(*args, **kwargs)) + with trace.use_span(span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + result = await wrapped(*args, **kwargs) + return complete(span, result) except BaseException as exc: _failure(span, exc) raise @@ -245,7 +274,9 @@ async def wrapped(wrapped, instance, args, kwargs): def wrapped(wrapped, instance, args, kwargs): span = _start(tracer, operation, kwargs, config) try: - return complete(span, wrapped(*args, **kwargs)) + with trace.use_span(span, end_on_exit=False, record_exception=False, set_status_on_exception=False): + result = wrapped(*args, **kwargs) + return complete(span, result) except BaseException as exc: _failure(span, exc) raise From 197d66e2af3812fd601ad35ac1ff69e15ccefa36 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 16:23:25 -0800 Subject: [PATCH 4/5] fix(openrouter): skip unsupported prompt objects safely Co-Authored-By: Claude Code --- python/frameworks/openrouter/tests/test_review_regressions.py | 2 -- python/frameworks/openrouter/traceai_openrouter/__init__.py | 4 +++- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/python/frameworks/openrouter/tests/test_review_regressions.py b/python/frameworks/openrouter/tests/test_review_regressions.py index a8f669453..4144f2f80 100644 --- a/python/frameworks/openrouter/tests/test_review_regressions.py +++ b/python/frameworks/openrouter/tests/test_review_regressions.py @@ -5,8 +5,6 @@ import httpx import pytest from fi_instrumentation import TraceConfig -from openrouter import OpenRouter -from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py index c8e123dae..a19be4615 100644 --- a/python/frameworks/openrouter/traceai_openrouter/__init__.py +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -98,7 +98,9 @@ def _input(span: Any, operation: str, kwargs: dict[str, Any], content: bool) -> # Keep only plain chat messages and plain text Responses inputs. if operation == "chat": if not isinstance(value, (list, tuple)) or any( - not isinstance(msg, dict) or not isinstance(msg.get("content"), str) + not isinstance(msg, dict) + or not isinstance(msg.get("role"), str) + or not isinstance(msg.get("content"), str) for msg in value ): return From fbc491b58f9d16bef7d0677a43c5bca2318453d9 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Mon, 2 Nov 2026 16:40:16 -0800 Subject: [PATCH 5/5] fix(openrouter): reject Responses SSE errors and cancellation Co-Authored-By: Claude Code --- .../tests/test_review_regressions.py | 53 +++++++++++++++++++ .../openrouter/traceai_openrouter/__init__.py | 9 ++-- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/python/frameworks/openrouter/tests/test_review_regressions.py b/python/frameworks/openrouter/tests/test_review_regressions.py index 4144f2f80..0ffbb9c23 100644 --- a/python/frameworks/openrouter/tests/test_review_regressions.py +++ b/python/frameworks/openrouter/tests/test_review_regressions.py @@ -102,6 +102,59 @@ def test_responses_failure_drops_partial_answer(instrument, state): assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) +@pytest.mark.parametrize("async_call", [False, True]) +@pytest.mark.parametrize("beta", [False, True]) +def test_responses_stream_error_event_drops_partial_answer(instrument, async_call, beta): + _, exporter = instrument() + events = [ + {"type": "response.output_text.delta", "sequence_number": 1, "item_id": "item_1", + "output_index": 0, "content_index": 0, "logprobs": [], "delta": "partial"}, + {"type": "error", "sequence_number": 2, "code": "server_error", + "message": "SYNTHETIC_PRIVATE_ERROR", "param": None}, + ] + payload = "".join("data: " + json.dumps(event) + "\n\n" for event in events) + "data: [DONE]\n\n" + client = sdk(lambda req: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=payload.encode())) + responses = client.beta.responses if beta else client.responses + if async_call: + async def run(): + stream = await responses.send_async(model="vendor/request", input="hi", stream=True) + return [event async for event in stream] + chunks = asyncio.run(run()) + else: + chunks = list(responses.send(model="vendor/request", input="hi", stream=True)) + assert [chunk.type for chunk in chunks] == ["response.output_text.delta", "error"] + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert "output.value" not in span.attributes + assert "SYNTHETIC_PRIVATE_ERROR" not in str(dict(span.attributes)) + + +@pytest.mark.parametrize("async_call", [False, True]) +@pytest.mark.parametrize("beta", [False, True]) +def test_cancelled_responses_drops_partial_answer(instrument, async_call, beta): + _, exporter = instrument() + body = {"id": "resp_1", "object": "response", "created_at": 1, "completed_at": None, + "error": None, "frequency_penalty": None, "incomplete_details": None, + "instructions": None, "metadata": None, "model": "vendor/served", "output": [], + "output_text": "partial", "parallel_tool_calls": False, "presence_penalty": None, + "status": "cancelled", "temperature": None, "tool_choice": "auto", "tools": [], + "top_p": None, "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}}} + client = sdk(lambda req: httpx.Response(200, json=body)) + responses = client.beta.responses if beta else client.responses + if async_call: + result = asyncio.run(responses.send_async(model="vendor/request", input="hi")) + else: + result = responses.send(model="vendor/request", input="hi") + assert result.status == "cancelled" + span = exporter.get_finished_spans()[0] + assert span.status.status_code is StatusCode.ERROR + assert span.attributes["gen_ai.response.model"] == "vendor/served" + assert span.attributes["gen_ai.usage.input_tokens"] == 3 + assert "output.value" not in span.attributes + + @pytest.mark.parametrize("async_call", [False, True]) def test_http_child_nested_under_llm(instrument, async_call): provider, exporter = instrument() diff --git a/python/frameworks/openrouter/traceai_openrouter/__init__.py b/python/frameworks/openrouter/traceai_openrouter/__init__.py index a19be4615..ef653184a 100644 --- a/python/frameworks/openrouter/traceai_openrouter/__init__.py +++ b/python/frameworks/openrouter/traceai_openrouter/__init__.py @@ -76,8 +76,8 @@ def _result(span: Any, value: Any, content: bool) -> None: if isinstance(model, str): span.set_attribute(A.GEN_AI_RESPONSE_MODEL, model) _usage(span, value) - if _field(value, "status") in ("failed", "incomplete"): - # A 200 Responses body can still represent a failed/incomplete generation. + if _field(value, "status") in ("failed", "incomplete", "cancelled"): + # A 200 Responses body can still represent an aborted generation. # Never export its error message or partial output. span.set_status(Status(StatusCode.ERROR)) return @@ -137,12 +137,15 @@ def _observe(self, chunk: Any) -> None: # the event itself. The final response supersedes individual deltas. event_type = _field(chunk, "type") completed = _field(chunk, "response") - if _field(chunk, "error") is not None or event_type in ("response.failed", "response.incomplete"): + if _field(chunk, "error") is not None or event_type in ("error", "response.failed", "response.incomplete"): self._failed = True self._pieces.clear() self._span.set_status(Status(StatusCode.ERROR)) if event_type in ("response.completed", "response.failed", "response.incomplete") and completed: _result(self._span, completed, False) + if _field(completed, "status") in ("failed", "incomplete", "cancelled"): + self._failed = True + self._pieces.clear() if not self._failed and self._content: text = _text(completed) if text: