feat(weave): trace Claude Agent SDK subagents

AGENTS.md

@@ -452,6 +452,10 @@ pnpm exec tsx examples/claudeAgents.ts
 - `Turn.messages` stores input messages, while `Turn.output_messages` stores
   the terminal agent response. `Turn.record(messages=..., output_messages=...)`
   replaces the two lists independently.
+- `SubAgent` owns its input/output messages and originating tool-call
+  arguments/result. Integrations populate them through `SubAgent.record()` so
+  `_build_attrs()` applies `include_content` and PII redaction; do not write
+  those content attributes directly with `set_attributes()`.
 - Mypy requires explicit `return None` paths in functions annotated with
   `T | None`; bare `return` and implicit fallthrough trigger return-value
   errors.
@@ -478,13 +482,26 @@ pnpm exec tsx examples/claudeAgents.ts
 - Regression coverage must exercise both the calls-based and OTel integrations
   with nonzero cache-read and cache-creation counts.
 - The OTel-selected Claude Agent SDK path composes the Python GenAI
-  `Conversation`, `Turn`, `LLM`, and `Tool` handles; it must not create raw
-  OTel spans or call the low-level GenAI attribute builders itself. Use
-  `set_attributes()` only for semantic fields the typed handles do not expose.
-  The legacy calls-based path remains separate.
+  `Conversation`, `Turn`, `LLM`, `Tool`, and `SubAgent` handles; it must not
+  create raw OTel spans or call the low-level GenAI attribute builders itself.
+  Use `set_attributes()` only for semantic fields the typed handles do not
+  expose. The legacy calls-based path remains separate.
 - Tap Python `AsyncIterable[dict]` prompts without consuming or cloning them.
   Map Claude text and base64/URL image blocks to GenAI text, blob, and URI
   parts; media payloads remain data for `content_refs`, not chat prose.
+- Treat `Agent` and legacy `Task` tool calls as subagents keyed by tool-use ID.
+  Route nested assistant messages and tools through `parent_tool_use_id`.
+  Synchronous calls close on their matching tool result; background launch
+  acknowledgements stay open until `task_notification`. Dispatch task events
+  through `SystemMessage.subtype` and `data`, map `task_id` to tool-use ID from
+  `task_started` / `task_progress`, and do not import typed task messages that
+  are absent from older supported Claude Agent SDK versions.
+- Start those subagents with `SubAgent.start(set_current=False)`, never
+  `__enter__`. Parallel delegations close in completion order, and `end()`
+  detaches through `ContextVar.reset`, so an out-of-LIFO close leaves the
+  ambient OTel context pointing at an ended span — user code running between
+  streamed messages would nest under it. Children nest either way, because
+  `start_llm` / `start_tool` / `start_subagent` thread an explicit parent.
 - Stream adapters can create child handles when work starts, then enter them
   with a normal `with` when the completion message arrives. Preserve logical
   timing with `LLM.started_at` and an explicit `Tool.started_at` instead of

tests/conversation/test_conversation_otel.py

@@ -2318,13 +2318,25 @@ def test_sets_all_fields(self) -> None:
             name="research-bot",
             model="gpt-4o-mini",
             system_instructions=["sys"],
+            input_messages=[Message.user("question")],
+            output_messages=[Message.assistant("answer")],
+            tool_name="Agent",
+            tool_call_id="toolu_1",
+            tool_call_arguments='{"prompt": "question"}',
+            tool_call_result="answer",
             agent_id="id-1",
             agent_description="desc",
             agent_version="v1",
         )
         assert sa.name == "research-bot"
         assert sa.model == "gpt-4o-mini"
         assert sa.system_instructions == ["sys"]
+        assert sa.input_messages == [Message.user("question")]
+        assert sa.output_messages == [Message.assistant("answer")]
+        assert sa.tool_name == "Agent"
+        assert sa.tool_call_id == "toolu_1"
+        assert sa.tool_call_arguments == '{"prompt": "question"}'
+        assert sa.tool_call_result == "answer"
         assert sa.agent_id == "id-1"
         assert sa.agent_description == "desc"
         assert sa.agent_version == "v1"

tests/conversation/test_conversation_settings.py

@@ -19,6 +19,7 @@
 
 from __future__ import annotations
 
+import json
 import platform
 import sys
 from collections.abc import Callable
@@ -34,6 +35,7 @@
     LLM,
     Message,
     Reasoning,
+    SubAgent,
     Tool,
     Turn,
     log_conversation,
@@ -289,6 +291,92 @@ def test_redact_pii_applied(
         assert _REDACTED_EMAIL in attrs[key], key
 
 
+def _streaming_subagent_with_content(include_content: bool) -> None:
+    with start_conversation(
+        conversation_id="convo-subagent-content",
+        include_content=include_content,
+    ) as session:
+        with session.start_turn() as turn:
+            with turn.start_subagent(name="researcher") as subagent:
+                subagent.record(
+                    input_messages=[Message.user(f"Research {_PII_EMAIL}")],
+                    output_messages=[Message.assistant(f"Found {_PII_EMAIL}")],
+                    tool_name="Agent",
+                    tool_call_id="toolu_01",
+                    tool_call_arguments=json.dumps({"email": _PII_EMAIL}),
+                    tool_call_result=f"Found {_PII_EMAIL}",
+                )
+
+
+def _batch_subagent_with_content(include_content: bool) -> None:
+    log_turn(
+        conversation_id="convo-subagent-content",
+        agent_name="orchestrator",
+        include_content=include_content,
+        spans=[
+            SubAgent(
+                name="researcher",
+                input_messages=[Message.user(f"Research {_PII_EMAIL}")],
+                output_messages=[Message.assistant(f"Found {_PII_EMAIL}")],
+                tool_name="Agent",
+                tool_call_id="toolu_01",
+                tool_call_arguments=json.dumps({"email": _PII_EMAIL}),
+                tool_call_result=f"Found {_PII_EMAIL}",
+            )
+        ],
+    )
+
+
+@pytest.mark.parametrize(
+    "emit_subagent",
+    [
+        pytest.param(_streaming_subagent_with_content, id="streaming"),
+        pytest.param(_batch_subagent_with_content, id="batch"),
+    ],
+)
+def test_subagent_content_is_pii_redacted(
+    emit_subagent: Callable[[bool], None],
+    otel_spans: InMemorySpanExporter,
+    fake_presidio: None,
+) -> None:
+    with override_settings(
+        redact_pii=True,
+        capture_client_info=False,
+        capture_system_info=False,
+    ):
+        emit_subagent(True)
+
+    spans = _spans_with_prefix(otel_spans, "invoke_agent researcher")
+    assert len(spans) == 1
+    assert dict(spans[0].attributes) == {
+        "gen_ai.operation.name": "invoke_agent",
+        "gen_ai.agent.name": "researcher",
+        "gen_ai.conversation.id": "convo-subagent-content",
+        "gen_ai.input.messages": json.dumps(
+            [
+                {
+                    "role": "user",
+                    "parts": [
+                        {"type": "text", "content": f"Research {_REDACTED_EMAIL}"}
+                    ],
+                }
+            ]
+        ),
+        "gen_ai.output.messages": json.dumps(
+            [
+                {
+                    "role": "assistant",
+                    "parts": [{"type": "text", "content": f"Found {_REDACTED_EMAIL}"}],
+                }
+            ]
+        ),
+        "gen_ai.tool.name": "Agent",
+        "gen_ai.tool.call.id": "toolu_01",
+        "gen_ai.tool.call.arguments": json.dumps({"email": _REDACTED_EMAIL}),
+        "gen_ai.tool.call.result": f"Found {_REDACTED_EMAIL}",
+    }
+
+
 # ---------------------------------------------------------------------------
 # include_content=False — content dropped at source, Presidio never loaded
 # ---------------------------------------------------------------------------
@@ -358,6 +446,37 @@ def test_include_content_false_skips_presidio(
     mock_get_engines.assert_not_called()
 
 
+@pytest.mark.parametrize(
+    "emit_subagent",
+    [
+        pytest.param(_streaming_subagent_with_content, id="streaming"),
+        pytest.param(_batch_subagent_with_content, id="batch"),
+    ],
+)
+def test_subagent_include_content_false_drops_content_before_redaction(
+    emit_subagent: Callable[[bool], None],
+    otel_spans: InMemorySpanExporter,
+) -> None:
+    with override_settings(
+        redact_pii=True,
+        capture_client_info=False,
+        capture_system_info=False,
+    ):
+        with patch("weave.utils.pii_redaction._get_engines") as mock_get_engines:
+            emit_subagent(False)
+
+    mock_get_engines.assert_not_called()
+    spans = _spans_with_prefix(otel_spans, "invoke_agent researcher")
+    assert len(spans) == 1
+    assert dict(spans[0].attributes) == {
+        "gen_ai.operation.name": "invoke_agent",
+        "gen_ai.agent.name": "researcher",
+        "gen_ai.conversation.id": "convo-subagent-content",
+        "gen_ai.tool.name": "Agent",
+        "gen_ai.tool.call.id": "toolu_01",
+    }
+
+
 def _reasoning_content_off_streaming() -> None:
     with start_conversation(
         conversation_id="convo-reasoning-off-streaming", include_content=False

tests/conversation/test_subagent_nesting.py

@@ -18,6 +18,7 @@
 import threading
 
 import pytest
+from opentelemetry import trace as otel_trace
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
 from weave.conversation.conversation import (
@@ -333,3 +334,59 @@ def worker() -> None:
         llm_span = _by_prefix(spans, "chat")
         assert llm_span.parent is not None
         assert llm_span.parent.span_id == sa_span.context.span_id
+
+
+# ---------------------------------------------------------------------------
+# Concurrent sub-agents: SubAgent.start(set_current=False)
+# ---------------------------------------------------------------------------
+
+
+class TestConcurrentSubagents:
+    """``end()`` detaches via ``ContextVar.reset``, so overlapping sub-agents
+    that finish out of LIFO order corrupt the ambient OTel context. Adapters
+    that run sub-agents concurrently start them with ``set_current=False``.
+    """
+
+    @pytest.mark.parametrize("close_order", [("a", "b"), ("b", "a")])
+    def test_ambient_context_survives_any_close_order(
+        self, otel_spans: InMemorySpanExporter, close_order: tuple[str, str]
+    ) -> None:
+        with Conversation(conversation_id="s"), Turn(agent_name="bot") as turn:
+            subagents = {
+                "a": turn.start_subagent(name="alpha").start(set_current=False),
+                "b": turn.start_subagent(name="beta").start(set_current=False),
+            }
+            for key in close_order:
+                subagents[key].end()
+            assert (
+                otel_trace.get_current_span().get_span_context().span_id
+                == turn._otel_span.get_span_context().span_id
+            )
+
+        spans = otel_spans.get_finished_spans()
+        turn_span = _by_agent_name(spans, "bot")
+        for name in ("alpha", "beta"):
+            assert _by_agent_name(spans, name).parent.span_id == (
+                turn_span.context.span_id
+            )
+
+    def test_children_still_nest_under_a_non_current_subagent(
+        self, otel_spans: InMemorySpanExporter
+    ) -> None:
+        with Conversation(conversation_id="s"), Turn(agent_name="bot") as turn:
+            first = turn.start_subagent(name="alpha").start(set_current=False)
+            second = turn.start_subagent(name="beta").start(set_current=False)
+            with first.start_llm(model="gpt-4o"):
+                pass
+            with second.start_tool(name="grep"):
+                pass
+            first.end()
+            second.end()
+
+        spans = otel_spans.get_finished_spans()
+        assert _by_prefix(spans, "chat").parent.span_id == (
+            _by_agent_name(spans, "alpha").context.span_id
+        )
+        assert _by_prefix(spans, "execute_tool").parent.span_id == (
+            _by_agent_name(spans, "beta").context.span_id
+        )