fix(openai): mark interrupted agents and truncated chat streams as errors

.github/workflows/parallel-python-tests.yml

@@ -37,7 +37,7 @@ jobs:
             distribution: wheel
           - python: '3.13'
             parallel: '1.3.7'
-            openai: '3.42.0'
+            openai: '3.42.1'
             distribution: sdist
     env:
       FI_BASE_URL: http://127.0.0.1:9
@@ -65,7 +65,8 @@ jobs:
             packages=(python/frameworks/{parallel,openai}/dist/*.tar.gz)
           fi
           .parallel-test-venv/bin/pip install "${packages[@]}" \
-            "parallel-web==$PARALLEL_VERSION" "openai==$OPENAI_VERSION" pytest pytest-asyncio
+            "parallel-web==$PARALLEL_VERSION" "openai==$OPENAI_VERSION" \
+            'fi-instrumentation-otel>=1.2.2' pytest pytest-asyncio
       - name: Test with only loopback networking
         shell: bash
         run: |

python/examples/perplexity/tests/test_perplexity_recipe.py

@@ -37,6 +37,19 @@
         == Path(CORE_DISTRIBUTION.locate_file("fi_instrumentation/__init__.py")).resolve()
     )
 
+# The recipe CI deliberately retains the old published SDK as a compatibility
+# control. Only that exact installed distribution exports echoed instructions;
+# current source must not reintroduce them to satisfy legacy expectations.
+try:
+    OPENAI_DISTRIBUTION = distribution("traceAI-openai")
+except PackageNotFoundError:
+    LEGACY_OUTPUT = False
+else:
+    LEGACY_OUTPUT = OPENAI_DISTRIBUTION.version == "0.1.10" and (
+        Path(traceai_openai.__file__).resolve()
+        == Path(OPENAI_DISTRIBUTION.locate_file("traceai_openai/__init__.py")).resolve()
+    )
+
 TEST_DIR = Path(__file__).resolve().parent
 RECIPE_DIR = TEST_DIR.parent
 VENDOR_KEY = "placeholder-perplexity-key"
@@ -254,7 +267,10 @@ def test_responses_at_documented_host(tracing):
             assert raw["usage"]["input_tokens_details"][key] == value
         assert raw["usage"]["output_tokens_details"] == fake.USAGE["output_tokens_details"]
         assert raw["usage"]["tool_calls_details"] == fake.USAGE["tool_calls_details"]
-        assert fake.INSTRUCTIONS_MARKER in raw["instructions"]
+        if LEGACY_OUTPUT:
+            assert fake.INSTRUCTIONS_MARKER in raw["instructions"]
+        else:
+            assert "instructions" not in raw
         assert fake.SEARCH_URL in attrs["output.value"]
         assert fake.SEARCH_SNIPPET in attrs["output.value"]
 
@@ -349,9 +365,11 @@ def handler(request):
             assert _call(handler, prompt=marker) == fake.ANSWER
             span, _ = _assert_export(receiver, provider, project, hide_inputs=hidden)
             assert (marker in json.dumps(span)) is (not hidden)
-            # Output echoes remain visible even when the input itself is hidden.
-            for output_marker in (fake.QUERY_MARKER, fake.INSTRUCTIONS_MARKER):
-                assert output_marker in _attributes(span)["output.value"]
+            # Search queries are response-owned output; instructions are input.
+            output = _attributes(span)["output.value"]
+            assert fake.QUERY_MARKER in output
+            assert (fake.INSTRUCTIONS_MARKER in output) is LEGACY_OUTPUT
+            assert fake.ANSWER in output
 
 
 @pytest.mark.parametrize("stream", [False, True], ids=["nonstream", "stream"])
@@ -363,8 +381,11 @@ def test_hide_outputs_with_control(tracing, monkeypatch, stream):
             span, attrs = _assert_export(receiver, provider, project)
             _assert_usage(attrs)
             assert (fake.ANSWER in json.dumps(span)) is (not hidden)
-            for marker in (fake.SEARCH_URL, fake.SEARCH_SNIPPET, fake.QUERY_MARKER, fake.INSTRUCTIONS_MARKER):
+            for marker in (fake.SEARCH_URL, fake.SEARCH_SNIPPET, fake.QUERY_MARKER):
                 assert (marker in json.dumps(span)) is (not hidden and not stream)
+            assert (fake.INSTRUCTIONS_MARKER in json.dumps(span)) is (
+                LEGACY_OUTPUT and not hidden and not stream
+            )
 
 
 @pytest.mark.parametrize("hidden", [False, True], ids=["visible", "hidden"])
@@ -381,8 +402,9 @@ def test_empty_text_stream_raw_fallback(tracing, monkeypatch, hidden):
             list(client.responses.create(model=fake.REQUEST_MODEL, input="empty stream", stream=True))
         span, attrs = _assert_export(receiver, provider, project)
         _assert_usage(attrs)
-        for marker in (fake.SEARCH_URL, fake.QUERY_MARKER, fake.INSTRUCTIONS_MARKER):
+        for marker in (fake.SEARCH_URL, fake.QUERY_MARKER):
             assert (marker in json.dumps(span)) is (not hidden)
+        assert (fake.INSTRUCTIONS_MARKER in json.dumps(span)) is (LEGACY_OUTPUT and not hidden)
 
 
 def test_cache_details_require_output_details(tracing):
@@ -457,7 +479,7 @@ def test_subprocess_app(tmp_path, stream, cli_model):
         assert (fake.SEARCH_URL in json.dumps(attrs)) is (not stream)
         assert (fake.SEARCH_SNIPPET in json.dumps(attrs)) is (not stream)
         assert (fake.QUERY_MARKER in json.dumps(attrs)) is (not stream)
-        assert (fake.INSTRUCTIONS_MARKER in json.dumps(attrs)) is (not stream)
+        assert (fake.INSTRUCTIONS_MARKER in json.dumps(attrs)) is (LEGACY_OUTPUT and not stream)
         [export] = receiver.requests()
         assert export["path"] == "/tracer/v1/traces"
         assert export["headers"]["x-api-key"] == FI_KEY

python/fi_instrumentation/instrumentation/_tracers.py

@@ -203,7 +203,21 @@ def start_as_current_span(
             record_exception=record_exception,
             set_status_on_exception=set_status_on_exception,
         ) as current_span:
-            yield cast(FiSpan, current_span)
+            try:
+                yield cast(FiSpan, current_span)
+            except BaseException as error:
+                # OTel use_span handles Exception, but deliberately excludes
+                # cancellation, KeyboardInterrupt and other local aborts. Record
+                # those before use_span ends the span; leave ordinary errors to
+                # OTel so they are not recorded twice. FiSpan applies privacy.
+                if not isinstance(error, Exception) and span.is_recording():
+                    if record_exception:
+                        span.record_exception(error, escaped=True)
+                    if set_status_on_exception:
+                        span.set_status(Status(
+                            StatusCode.ERROR, f"{type(error).__name__}: {error}"
+                        ))
+                raise
 
     def start_span(
         self,

python/frameworks/openai/pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
 
 [tool.poetry.dependencies]
 python = ">3.9,<3.14"
-fi-instrumentation-otel = ">=0.1.11"
+fi-instrumentation-otel = ">=1.2.2"
 openai = ">=1.69.0"
 
 [tool.poetry.urls]

python/frameworks/openai/tests/test_chat_stream_completion.py

@@ -0,0 +1,181 @@
+"""HTTP EOF is not evidence that every requested chat choice completed."""
+import asyncio
+import json
+
+import httpx
+import openai
+import pytest
+from fi_instrumentation import TraceConfig
+from opentelemetry.trace import StatusCode
+
+from .test_openai_compatible_spans import (
+    REQUEST_MODEL, async_client, client, make_tracer, only_span,
+)
+
+
+def chunk(index=0, finish_reason=None, content=None):
+    return {
+        "id": "synthetic", "object": "chat.completion.chunk", "created": 1,
+        "model": REQUEST_MODEL,
+        "choices": [{"index": index, "delta": {"content": content} if content else {},
+                     "finish_reason": finish_reason}],
+    }
+
+
+def stream_response(chunks, done):
+    body = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks)
+    if done:
+        body += "data: [DONE]\n\n"
+    return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("done", [False, True])
+@pytest.mark.parametrize("case", [
+    "empty", "partial", "usage_only", "stop", "length", "tool_calls", "content_filter",
+    "function_call", "two_complete", "second_partial", "second_missing", "usage_tail",
+])
+def test_chat_completion_requires_terminal_choices(make_tracer, asynchronous, done, case):
+    exporter = make_tracer(TraceConfig(hide_inputs=True, hide_outputs=True, hide_error_details=True))
+    chunks = [chunk(content="synthetic-private-output")]
+    n = 1
+    complete = case in {"stop", "length", "tool_calls", "content_filter", "function_call", "two_complete", "usage_tail"}
+    usage = {**chunk(), "choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5}}
+    if case == "empty":
+        chunks = []
+    elif case == "usage_only":
+        chunks = [usage]
+    elif case in {"two_complete", "second_partial", "second_missing"}:
+        n = 2
+        chunks.append(chunk(finish_reason="stop"))
+        if case != "second_missing":
+            chunks.append(chunk(index=1, content="second"))
+        if case == "two_complete":
+            chunks.append(chunk(index=1, finish_reason="stop"))
+    elif case == "usage_tail":
+        chunks.extend([chunk(finish_reason="stop"), usage])
+    elif complete:
+        chunks.append(chunk(finish_reason=case))
+    params = dict(model=REQUEST_MODEL, messages=[{"role": "user", "content": "synthetic-input"}], stream=True, n=n)
+
+    async def run_async():
+        async with async_client(lambda _: stream_response(chunks, done)) as sdk:
+            stream = await sdk.chat.completions.create(**params)
+            actual = [c async for c in stream]
+            assert len(actual) == len(chunks)
+            await stream.close()
+
+    if asynchronous:
+        asyncio.run(run_async())
+    else:
+        with client(lambda _: stream_response(chunks, done)) as sdk:
+            stream = sdk.chat.completions.create(**params)
+            actual = list(stream)
+            assert len(actual) == len(chunks)
+            stream.close()
+    span, attrs = only_span(exporter)
+    assert span.status.status_code == (StatusCode.OK if complete else StatusCode.ERROR)
+    assert "synthetic-private-output" not in str(attrs)
+    if case == "usage_tail":
+        assert attrs["gen_ai.usage.total_tokens"] == 5
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("declared_n,override_n,completed_choices,expected_status", [
+    pytest.param(2, 1, 1, StatusCode.OK, id="override-down"),
+    pytest.param(1, 2, 1, StatusCode.ERROR, id="override-up-incomplete"),
+    pytest.param(1, 2, 2, StatusCode.OK, id="override-up-complete"),
+    pytest.param(2, openai.Omit(), 1, StatusCode.OK, id="remove-n"),
+    pytest.param(2, None, 1, StatusCode.OK, id="null-n"),
+    pytest.param(openai.NOT_GIVEN, 2, 1, StatusCode.ERROR, id="extra-body-only-n"),
+])
+def test_completion_count_uses_effective_body(
+    make_tracer, asynchronous, declared_n, override_n, completed_choices, expected_status,
+):
+    exporter = make_tracer()
+    chunks = [chunk(index=i, finish_reason="stop") for i in range(completed_choices)]
+    wire_bodies = []
+
+    def handler(request):
+        body = json.loads(request.content)
+        wire_bodies.append(body)
+        if isinstance(override_n, openai.Omit):
+            assert "n" not in body
+        else:
+            assert body["n"] == override_n
+        return stream_response(chunks, done=True)
+
+    extra_body = {"n": override_n, "synthetic_extra": "not-exported"}
+    params = dict(
+        model=REQUEST_MODEL, messages=[{"role": "user", "content": "synthetic-input"}],
+        stream=True, n=declared_n, extra_body=extra_body,
+    )
+
+    async def run_async():
+        async with async_client(handler) as sdk:
+            stream = await sdk.chat.completions.create(**params)
+            assert len([c async for c in stream]) == completed_choices
+            await stream.close()
+
+    if asynchronous:
+        asyncio.run(run_async())
+    else:
+        with client(handler) as sdk:
+            stream = sdk.chat.completions.create(**params)
+            assert len(list(stream)) == completed_choices
+            stream.close()
+
+    assert len(wire_bodies) == 1
+    assert extra_body == {"n": override_n, "synthetic_extra": "not-exported"}
+    span, attrs = only_span(exporter)
+    assert span.status.status_code == expected_status
+    # This fix must not expand exported metadata to arbitrary extra_body fields.
+    parameters = json.loads(attrs["gen_ai.request.parameters"])
+    if declared_n is openai.NOT_GIVEN:
+        assert "n" not in parameters
+    else:
+        assert parameters["n"] == declared_n
+    assert "synthetic_extra" not in str(attrs)
+    assert "not-exported" not in str(attrs)
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("terminal", [False, True])
+@pytest.mark.parametrize("consumer_error", [False, True])
+def test_context_exit_uses_observed_terminal_state(make_tracer, asynchronous, terminal, consumer_error):
+    exporter = make_tracer()
+    chunks = [chunk(content="partial", finish_reason="stop" if terminal else None)]
+    params = dict(model=REQUEST_MODEL, messages=[{"role": "user", "content": "hi"}], stream=True)
+    error = ValueError("synthetic consumer error")
+
+    async def run_async():
+        async with async_client(lambda _: stream_response(chunks, True)) as sdk:
+            stream = await sdk.chat.completions.create(**params)
+            async with stream:
+                await stream.__anext__()
+                if consumer_error:
+                    raise error
+
+    def run_sync():
+        with client(lambda _: stream_response(chunks, True)) as sdk:
+            stream = sdk.chat.completions.create(**params)
+            with stream:
+                next(stream)
+                if consumer_error:
+                    raise error
+
+    def run():
+        if asynchronous:
+            asyncio.run(run_async())
+        else:
+            run_sync()
+
+    if consumer_error:
+        with pytest.raises(ValueError) as caught:
+            run()
+        assert caught.value is error
+    else:
+        run()
+    span, _ = only_span(exporter)
+    assert span.status.status_code == (StatusCode.OK if terminal and not consumer_error else StatusCode.ERROR)
+    assert len([e for e in span.events if e.name == "exception"]) == int(consumer_error)

python/frameworks/openai/tests/test_core_dependency_privacy.py

@@ -0,0 +1,42 @@
+"""Synthetic fixture to verify declared core compatibility and hidden error details."""
+import asyncio
+import httpx
+import openai
+import pytest
+from fi_instrumentation import TraceConfig, register
+from fi_instrumentation.fi_types import ProjectType
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from traceai_openai import OpenAIInstrumentor
+
+MESSAGE = 'synthetic-private-upstream-error-483129'
+
+@pytest.mark.parametrize('async_', [False, True])
+def test_hidden_failed_response(async_, monkeypatch):
+    monkeypatch.setenv('FI_HIDE_ERROR_DETAILS', 'true')
+    exporter = InMemorySpanExporter()
+    provider = register(project_name='core-floor-probe', project_type=ProjectType.OBSERVE,
+                        span_exporter=exporter, batch=False, verbose=False)
+    instrumentor = OpenAIInstrumentor()
+    instrumentor.instrument(tracer_provider=provider, config=TraceConfig(hide_inputs=True))
+    body = {'id': 'resp_probe', 'object': 'response', 'created_at': 1, 'model': 'synthetic',
+            'status': 'failed', 'output': [], 'error': {'code': 'server_error', 'message': MESSAGE},
+            'usage': {'input_tokens': 1, 'output_tokens': 0, 'total_tokens': 1}}
+    def handler(req): return httpx.Response(200, json=body)
+    def make_client():
+        http = (httpx.AsyncClient if async_ else httpx.Client)(transport=httpx.MockTransport(handler), trust_env=False)
+        return (openai.AsyncOpenAI if async_ else openai.OpenAI)(api_key='synthetic-key', base_url='https://api.parallel.ai/v1', max_retries=0, http_client=http)
+    try:
+        if async_:
+            async def run():
+                async with make_client() as client:
+                    assert (await client.responses.create(model='synthetic', input='private input')).status == 'failed'
+            asyncio.run(run())
+        else:
+            with make_client() as client:
+                assert client.responses.create(model='synthetic', input='private input').status == 'failed'
+        spans = exporter.get_finished_spans()
+        assert len(spans) == 1
+        assert MESSAGE not in spans[0].to_json()
+    finally:
+        instrumentor.uninstrument()
+        provider.shutdown()