test(tracing): cover privacy and suspension edges

Co-Authored-By: Claude <noreply@anthropic.com>

Author: 2 people

tests/tracing/test_content_recording.py

@@ -197,6 +197,31 @@ async def test_chat_span_carries_system_input_and_raw_payload(self):
         assert resp["id"] == "faux-1"
         assert resp["role"] == "assistant"
 
+    async def test_provider_exception_details_require_content_opt_in(self):
+        secret = "Authorization: Bearer OPTED-IN-ERROR"
+        agent, provider, exporter, tracer = await _build(record_content=True)
+
+        def explode(*_args):
+            raise RuntimeError(secret)
+
+        provider.append_responses([explode])
+
+        await agent.prompt("private prompt")
+        await agent.wait_for_idle()
+        await tracer.shutdown()
+
+        chat = next(span for span in exporter.spans if span.name.startswith("chat "))
+        exception = next(
+            event
+            for event in chat.events
+            if event.name == "gen_ai.client.operation.exception"
+        )
+        attrs = dict(exception.attributes or {})
+        assert attrs["exception.type"] == "RuntimeError"
+        assert attrs["exception.message"] == secret
+        assert secret in attrs["exception.stacktrace"]
+        assert chat.status.description == secret
+
 
 class TestToolContent:
     async def test_execute_tool_carries_arguments_and_result(self):

tests/tracing/test_hitl_suspension.py

@@ -91,6 +91,34 @@ async def _detach_tracing(detach: Any, tracer: Tracer) -> None:
     await tracer.shutdown()
 
 
+async def test_suspended_outcome_without_snapshot_emits_no_event() -> None:
+    agent, _channel, _exporter, tracer = _build_suspending_agent("thread-no-snapshot")
+    observed: list[str] = []
+
+    async def observe(event, signal=None):  # noqa: ANN001, ANN202
+        del signal
+        observed.append(event.type)
+
+    unsubscribe = agent.subscribe(observe)
+    await agent._emit_suspended_event("suspended")
+    unsubscribe()
+    await tracer.shutdown()
+
+    assert observed == []
+
+
+async def test_recorder_suspension_without_active_run_is_noop() -> None:
+    from cubepi.tracing.recorder import Recorder
+
+    tracer = Tracer(service_name="test", exporters=[])
+    recorder = Recorder(tracer)
+
+    recorder._on_agent_suspended()
+    await tracer.shutdown()
+
+    assert recorder._run is None
+
+
 async def test_suspended_run_exports_suspended_trace_without_abort_or_leaks() -> None:
     agent, channel, exporter, tracer = _build_suspending_agent("thread-1")
     provider_stack_baseline = len(mcp_tracing._provider_stack)
@@ -130,6 +158,105 @@ async def test_suspended_run_exports_suspended_trace_without_abort_or_leaks() ->
     assert len(mcp_tracing._active_entries) == active_entries_baseline
 
 
+async def test_recorder_suspension_finalizes_all_open_resources() -> None:
+    from cubepi.tracing.recorder import Recorder, _RunState
+
+    class _Span:
+        def __init__(self) -> None:
+            self.attributes: dict[str, Any] = {}
+            self.ended = False
+
+        def set_attribute(self, key: str, value: Any) -> None:
+            self.attributes[key] = value
+
+        def end(self) -> None:
+            self.ended = True
+
+    class _Stream:
+        def __init__(self) -> None:
+            self.closed = False
+
+        def close(self) -> None:
+            self.closed = True
+
+    tracer = Tracer(service_name="test", exporters=[])
+    recorder = Recorder(tracer)
+    agent_span = _Span()
+    turn_span = _Span()
+    chat_span = _Span()
+    tool_span = _Span()
+    stream = _Stream()
+    run = _RunState(
+        run_id="run-open-resources",
+        agent_span=agent_span,
+        turn_span=turn_span,
+        chat_span=chat_span,
+        tool_spans={"tool-1": tool_span},
+        stream_file=stream,
+    )
+    recorder._run = run
+
+    recorder._on_agent_suspended()
+    await tracer.shutdown()
+
+    for span in (agent_span, turn_span, chat_span, tool_span):
+        assert span.attributes["cubepi.run.outcome"] == "suspended"
+        assert span.ended is True
+    assert stream.closed is True
+    assert run.tool_spans == {}
+    assert run.chat_span is None
+    assert run.turn_span is None
+    assert run.stream_file is None
+    assert recorder._run is None
+
+
+async def test_recorder_suspension_cleanup_continues_after_resource_errors() -> None:
+    from cubepi.tracing.recorder import Recorder, _RunState
+
+    class _Span:
+        def __init__(self) -> None:
+            self.attributes: dict[str, Any] = {}
+            self.ended = False
+
+        def set_attribute(self, key: str, value: Any) -> None:
+            self.attributes[key] = value
+
+        def end(self) -> None:
+            self.ended = True
+
+    class _BoomSpan(_Span):
+        def end(self) -> None:
+            raise RuntimeError("span end failed")
+
+    class _BoomStream:
+        def close(self) -> None:
+            raise OSError("stream close failed")
+
+    tracer = Tracer(service_name="test", exporters=[])
+    recorder = Recorder(tracer)
+    agent_span = _Span()
+    run = _RunState(
+        run_id="run-resource-errors",
+        agent_span=agent_span,
+        turn_span=_BoomSpan(),
+        chat_span=_BoomSpan(),
+        tool_spans={"tool-1": _BoomSpan()},
+        stream_file=_BoomStream(),
+    )
+    recorder._run = run
+
+    recorder._on_agent_suspended()
+    await tracer.shutdown()
+
+    assert agent_span.attributes["cubepi.run.outcome"] == "suspended"
+    assert agent_span.ended is True
+    assert run.tool_spans == {}
+    assert run.chat_span is None
+    assert run.turn_span is None
+    assert run.stream_file is None
+    assert recorder._run is None
+
+
 async def test_suspension_event_observes_committed_runtime_state() -> None:
     agent, channel, _exporter, tracer = _build_suspending_agent("thread-2")
     detach_tracing = tracer.attach(agent)

tests/tracing/test_recorder_coverage.py

@@ -547,6 +547,60 @@ async def noop(tool_call_id: str, params: P, *, signal=None, on_update=None):
         assert end_ev["args_chars"] > 0
         assert "args_preview" not in end_ev
 
+    async def test_stream_toolcall_previews_require_content_opt_in(self, tmp_path):
+        from pydantic import BaseModel
+
+        secret = "Bearer OPTED-IN-STREAM"
+
+        class P(BaseModel):
+            authorization: str
+
+        async def noop(tool_call_id: str, params: P, *, signal=None, on_update=None):
+            return AgentToolResult(content=[TextContent(text="done")])
+
+        tool = AgentTool(name="noop", description="d", parameters=P, execute=noop)
+        provider = FauxProvider(provider_id="faux")
+        provider.append_responses(
+            [
+                faux_assistant_message(
+                    [
+                        ToolCall(
+                            id="tc1",
+                            name="noop",
+                            arguments={"authorization": secret},
+                        )
+                    ],
+                    stop_reason="tool_use",
+                ),
+                faux_assistant_message("all done"),
+            ]
+        )
+        agent = Agent(model=provider.model(MODEL.id), system_prompt="s", tools=[tool])
+        tracer = Tracer(
+            service_name="t",
+            agent_name="a",
+            exporters=[],
+            record_content=True,
+            record_stream=True,
+            stream_dir=tmp_path,
+        )
+        tracer.attach(agent)
+
+        await agent.prompt("go")
+        await agent.wait_for_idle()
+        await tracer.shutdown()
+
+        stream_file = next(tmp_path.glob("*.stream.jsonl"))
+        events = [
+            json.loads(line) for line in stream_file.read_text().splitlines() if line
+        ]
+        delta_previews = "".join(
+            event["preview"] for event in events if event["type"] == "toolcall_delta"
+        )
+        end_ev = next(event for event in events if event["type"] == "toolcall_end")
+        assert "OPTED-IN-STREAM" in delta_previews
+        assert "OPTED-IN-STREAM" in end_ev["args_preview"]
+
     async def test_no_stream_file_without_record_stream(self, tmp_path):
         """Default (record_stream=False) must not create any stream file."""
         provider = FauxProvider(provider_id="faux")