feat(tracing): task spans say the declared output format and what came out, agent spans carry the prompt and answer, tool spans say whether the cache answered (#7597)

* feat(tracing): record the task's declared output format, the agent's prompt and answer, and the tool cache flag on their spans

A reader of a run's OTel spans could see a task's raw output but not the
format it declared, nor whether a Pydantic object or a JSON dict actually
came out of it; could see an agent's goal, backstory and model but not the
prompt it was handed or the answer it gave; and could see a tool's result
but not whether the tool ran or the cache answered.

execute task: crewai.task.output_format (json / pydantic / raw; from the
declaration on start and failure, from the TaskOutput on completion),
crewai.task.output_pydantic_produced, crewai.task.output_json_produced.

execute agent: gen_ai.input.messages carries the task prompt and
gen_ai.output.messages the answer, the spec shape the task span already
uses for its own text, under the existing per-attribute byte cap with the
.truncated / .original_size_bytes markers when cut.

call tool: crewai.tool.from_cache.

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

* test(tracing): the agent's prompt and answer leave under the two standard message keys and no other

Pins the review decision on #7597: the text travels as
gen_ai.input.messages / gen_ai.output.messages — the keys the call llm
span already exports its messages under — so a rule an exporter or a
redaction processor applies to LLM content by key name applies to the
agent span unchanged. A copy under a crewai.agent.* key would fail this.

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

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

lib/crewai/src/crewai/telemetry/tracing/handlers.py

@@ -111,6 +111,7 @@ from crewai.events.types.skill_events import (
     SkillLoadedEvent,
     SkillUsedEvent,
 )
+from crewai.tasks.output_format import OutputFormat
 from crewai.telemetry.tracing import semantic_conventions
 from crewai.telemetry.tracing.context import (
     PendingSpanEnd,
@@ -569,6 +570,24 @@ def _serialize(value: Any) -> str:
     return json.dumps(to_serializable(value))
 
 
+def _task_output_format(task: Any, output: Any = None) -> str:
+    """The output format ``task`` declared, as the ``OutputFormat`` value string.
+
+    A finished task's ``TaskOutput`` records the format it was produced under,
+    so that is preferred when given. Before completion, and on failure, the
+    format is read off the declaration (``output_json`` / ``output_pydantic``)
+    with the precedence ``Task._get_output_format`` applies.
+    """
+    declared = getattr(output, "output_format", None)
+    if declared is not None:
+        return str(getattr(declared, "value", declared))
+    if getattr(task, "output_json", None) is not None:
+        return OutputFormat.JSON.value
+    if getattr(task, "output_pydantic", None) is not None:
+        return OutputFormat.PYDANTIC.value
+    return OutputFormat.RAW.value
+
+
 def _set_span_attributes(span: Span, attributes: dict[str, Any]) -> None:
     for key, value in attributes.items():
         if value is not None:
@@ -904,6 +923,7 @@ def handle_task_started(
             name=task.name,
             description=task.description,
             expected_output=task.expected_output,
+            output_format=_task_output_format(task),
         ),
         **semantic_conventions.gen_ai(
             operation_name=semantic_conventions.GEN_AI_OP_EXECUTE_TASK,
@@ -936,6 +956,7 @@ def handle_task_completed(
     span = (
         ctx.active_spans.get(event.started_event_id) if event.started_event_id else None
     )
+    output = event.output
 
     attrs: dict[str, Any] = {
         **semantic_conventions.crewai_span(
@@ -948,11 +969,14 @@ def handle_task_completed(
             key=task.key,
             id=str(task.id),
             name=task.name,
-            output=task.output.raw if task.output else None,
-        ),
-        **semantic_conventions.gen_ai_io(
-            output_value=task.output.raw if task.output else None
+            output=output.raw,
+            output_format=_task_output_format(task, output),
+            # The declaration says what was asked for; these say what the run
+            # actually yielded, so a consumer can tell the two apart.
+            output_pydantic_produced=output.pydantic is not None,
+            output_json_produced=output.json_dict is not None,
         ),
+        **semantic_conventions.gen_ai_io(output_value=output.raw),
     }
 
     providers.emit_log(
@@ -989,6 +1013,7 @@ def handle_task_failed(
             name=task.name,
             description=task.description,
             expected_output=task.expected_output,
+            output_format=_task_output_format(task),
         ),
     }
 
@@ -1090,6 +1115,7 @@ def handle_agent_execution_started(
             tool_definitions=agent.tools,
             system_instructions=agent.backstory,
         ),
+        **semantic_conventions.gen_ai_io(input_value=event.task_prompt),
         **semantic_conventions.crewai_span(event_name=event.type, subject=agent.role),
         **semantic_conventions.crewai_agent(role=agent.role),
     }
@@ -1141,6 +1167,7 @@ def handle_agent_execution_completed(
             tool_definitions=agent.tools,
             system_instructions=agent.backstory,
         ),
+        **semantic_conventions.gen_ai_io(output_value=event.output),
         **semantic_conventions.crewai_span(event_name=event.type, subject=agent.role),
         **semantic_conventions.crewai_agent(role=agent.role),
     }
@@ -1398,6 +1425,7 @@ def handle_tool_usage_finished(
             event_name=event.type, subject=event.tool_name
         ),
         **semantic_conventions.crewai_agent(key=event.agent_key, role=event.agent_role),
+        **semantic_conventions.crewai_tool(from_cache=event.from_cache),
         **_tool_failure_attrs(failure),
     }
 

lib/crewai/src/crewai/telemetry/tracing/semantic_conventions.py

@@ -23,6 +23,7 @@ Custom conventions (crewai.* namespace):
     crewai_policy()     crewai.policy.*
     crewai_memory()     crewai.memory.*
     crewai_knowledge()  crewai.knowledge.*
+    crewai_tool()       crewai.tool.*
     crewai_llm()        crewai.llm.*
     crewai_mcp()        crewai.mcp.*
     crewai_a2a()        crewai.a2a.*
@@ -58,6 +59,17 @@ MCP_TOOL_EXECUTION_DURATION_MS = "crewai.mcp.tool_execution_duration_ms"
 HUMAN_FEEDBACK_WAIT_DURATION_MS = "crewai.human_feedback.wait_duration_ms"
 HUMAN_FEEDBACK_REQUEST_ID = "crewai.human_feedback.request_id"
 
+# Attribute keys that let a consumer of the spans reason about a run without
+# reconstructing it from raw text.
+TASK_OUTPUT_FORMAT = "crewai.task.output_format"
+"""The output format the task declared: ``json``, ``pydantic`` or ``raw``."""
+TASK_OUTPUT_PYDANTIC_PRODUCED = "crewai.task.output_pydantic_produced"
+"""Whether the task's output actually carries a parsed Pydantic object."""
+TASK_OUTPUT_JSON_PRODUCED = "crewai.task.output_json_produced"
+"""Whether the task's output actually carries a parsed JSON dict."""
+TOOL_FROM_CACHE = "crewai.tool.from_cache"
+"""Whether the tool result came from the tool cache rather than a live run."""
+
 GEN_AI_OP_INVOKE_WORKFLOW = "invoke_workflow"
 GEN_AI_OP_EXECUTE_METHOD = "execute_method"
 GEN_AI_OP_EXECUTE_TASK = "execute_task"
@@ -360,6 +372,9 @@ def crewai_task(
     description: str | None = None,
     expected_output: str | None = None,
     output: str | None = None,
+    output_format: str | None = None,
+    output_pydantic_produced: bool | None = None,
+    output_json_produced: bool | None = None,
 ) -> dict[str, Any]:
     return _filter_none(
         {
@@ -369,6 +384,9 @@ def crewai_task(
             "crewai.task.description": description,
             "crewai.task.expected_output": expected_output,
             "crewai.task.output": output,
+            TASK_OUTPUT_FORMAT: output_format,
+            TASK_OUTPUT_PYDANTIC_PRODUCED: output_pydantic_produced,
+            TASK_OUTPUT_JSON_PRODUCED: output_json_produced,
         }
     )
 
@@ -537,6 +555,23 @@ def crewai_knowledge(
     )
 
 
+def crewai_tool(
+    *,
+    from_cache: bool | None = None,
+) -> dict[str, Any]:
+    """Attributes of a tool call that completed, whatever it returned.
+
+    ``from_cache`` says whether the result was served from the tool cache; a
+    cached call never ran the tool, so its duration and result mean something
+    different to a consumer than a live call's.
+    """
+    return _filter_none(
+        {
+            TOOL_FROM_CACHE: from_cache,
+        }
+    )
+
+
 def crewai_tool_failure(
     *,
     message: str | None = None,

lib/crewai/tests/telemetry/test_span_task_agent_tool_attributes.py

@@ -0,0 +1,297 @@
+"""The execute task, execute agent and call tool spans carry what a reader needs.
+
+A consumer of a run's spans could see a task's raw text but not the format it
+declared, nor whether a Pydantic object or a JSON dict actually came out of it;
+it could see an agent's goal, backstory and model but not the prompt the agent
+was handed or the answer it gave; and it could see a tool's result but not
+whether the tool ran or the cache answered. Each had to be reconstructed from
+other spans, or could not be known at all.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+
+from crewai import Agent, Crew, Task
+from crewai.events.types.agent_events import (
+    AgentExecutionCompletedEvent,
+    AgentExecutionStartedEvent,
+)
+from crewai.events.types.task_events import (
+    TaskCompletedEvent,
+    TaskFailedEvent,
+    TaskStartedEvent,
+)
+from crewai.events.types.tool_usage_events import (
+    ToolUsageFinishedEvent,
+    ToolUsageStartedEvent,
+)
+from crewai.tasks.output_format import OutputFormat
+from crewai.tasks.task_output import TaskOutput
+from crewai.telemetry.tracing import handlers
+from crewai.telemetry.tracing.context import TelemetryExecutionContext
+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 pydantic import BaseModel
+import pytest
+
+
+# Long and non-repeating, so a truncated or elided copy cannot compare equal.
+# Under the default 32 KiB attribute cap, so it must arrive whole.
+LONG_TEXT = "".join(
+    f"paragraph {i}: the quick brown fox jumps over the lazy dog\n" for i in range(400)
+)
+assert 20_000 < len(LONG_TEXT.encode("utf-8")) < 32 * 1024
+
+
+@pytest.fixture(autouse=True)
+def enable_otel_sdk(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The suite otherwise runs with OTEL_SDK_DISABLED, which makes every
+    assertion here pass vacuously against non-recording spans."""
+    monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False)
+    monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False)
+
+
+class _Providers:
+    """All a handler asks of its providers: a tracer and somewhere to log."""
+
+    def __init__(self, tracer) -> None:
+        self._tracer = tracer
+
+    def get_tracer(self, name: str | None = None):
+        return self._tracer
+
+    def emit_log(self, *args, **kwargs) -> None:
+        pass
+
+
+@pytest.fixture
+def pipeline():
+    """Handler inputs whose spans land in memory, exported as each one ends."""
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    tracer = provider.get_tracer("test")
+    ctx = TelemetryExecutionContext(
+        kickoff_id="kickoff", automation_name="test", tracer=tracer
+    )
+    # yield, not return: the frame keeps `provider` alive, and a collected
+    # provider shuts down its processor and silently loses spans.
+    yield _Providers(tracer), ctx, exporter
+
+
+def _only_span(exporter: InMemorySpanExporter, name: str):
+    """The single exported span with ``name``, failing loudly if not unique."""
+    matches = [s for s in exporter.get_finished_spans() if s.name == name]
+    assert len(matches) == 1, [s.name for s in exporter.get_finished_spans()]
+    return matches[0]
+
+
+class Finding(BaseModel):
+    title: str
+
+
+def _crewed_task(**declaration) -> Task:
+    """A task whose agent belongs to a crew, as the task handlers require."""
+    agent = Agent(role="Researcher", goal="g", backstory="b")
+    task = Task(description="d", expected_output="e", agent=agent, **declaration)
+    crew = Crew(agents=[agent], tasks=[task])
+    # A kickoff attaches the crew to its agents; these events skip the kickoff.
+    agent.crew = crew
+    return task
+
+
+@pytest.mark.parametrize(
+    ("declaration", "produced", "expected"),
+    [
+        (
+            {"output_pydantic": Finding},
+            {"pydantic": Finding(title="t"), "output_format": OutputFormat.PYDANTIC},
+            ("pydantic", True, False),
+        ),
+        (
+            {"output_json": Finding},
+            {"json_dict": {"title": "t"}, "output_format": OutputFormat.JSON},
+            ("json", False, True),
+        ),
+        (
+            {},
+            {"output_format": OutputFormat.RAW},
+            ("raw", False, False),
+        ),
+    ],
+    ids=["pydantic", "json", "raw"],
+)
+def test_execute_task_span_says_what_was_declared_and_what_came_out(
+    pipeline, declaration, produced, expected
+) -> None:
+    providers, ctx, exporter = pipeline
+    task = _crewed_task(**declaration)
+    output = TaskOutput(
+        description="d", raw='{"title": "t"}', agent="Researcher", **produced
+    )
+
+    started = TaskStartedEvent(context=None, task=task)
+    handlers.handle_task_started(providers, ctx, task, started)
+    handlers.handle_task_completed(
+        providers,
+        ctx,
+        task,
+        TaskCompletedEvent(output=output, task=task, started_event_id=started.event_id),
+    )
+
+    span = _only_span(exporter, "execute task")
+    output_format, pydantic_produced, json_produced = expected
+    assert span.attributes["crewai.task.output_format"] == output_format
+    assert span.attributes["crewai.task.output_pydantic_produced"] is pydantic_produced
+    assert span.attributes["crewai.task.output_json_produced"] is json_produced
+    # The raw text is still there, as before.
+    assert span.attributes["crewai.task.output"] == '{"title": "t"}'
+
+
+def test_a_declared_pydantic_task_that_produced_none_is_told_apart(pipeline) -> None:
+    """The gap the two flags close: declared is not the same as produced."""
+    providers, ctx, exporter = pipeline
+    task = _crewed_task(output_pydantic=Finding)
+    # The producer could not convert, so the output carries only raw text.
+    output = TaskOutput(
+        description="d",
+        raw="not json at all",
+        agent="Researcher",
+        output_format=OutputFormat.PYDANTIC,
+    )
+
+    started = TaskStartedEvent(context=None, task=task)
+    handlers.handle_task_started(providers, ctx, task, started)
+    handlers.handle_task_completed(
+        providers,
+        ctx,
+        task,
+        TaskCompletedEvent(output=output, task=task, started_event_id=started.event_id),
+    )
+
+    span = _only_span(exporter, "execute task")
+    assert span.attributes["crewai.task.output_format"] == "pydantic"
+    assert span.attributes["crewai.task.output_pydantic_produced"] is False
+    assert span.attributes["crewai.task.output_json_produced"] is False
+
+
+def test_a_failed_task_still_says_what_it_declared(pipeline) -> None:
+    """There is no output to inspect, so the format comes from the declaration."""
+    providers, ctx, exporter = pipeline
+    task = _crewed_task(output_pydantic=Finding)
+
+    started = TaskStartedEvent(context=None, task=task)
+    handlers.handle_task_started(providers, ctx, task, started)
+    handlers.handle_task_failed(
+        providers,
+        ctx,
+        task,
+        TaskFailedEvent(error="boom", task=task, started_event_id=started.event_id),
+    )
+
+    span = _only_span(exporter, "execute task")
+    assert span.attributes["crewai.task.output_format"] == "pydantic"
+    assert "crewai.task.output_pydantic_produced" not in span.attributes
+    assert "crewai.task.output_json_produced" not in span.attributes
+
+
+def _agent_span(pipeline, *, prompt: str, output: str):
+    providers, ctx, exporter = pipeline
+    agent = Agent(role="Researcher", goal="g", backstory="b")
+
+    started = AgentExecutionStartedEvent(
+        agent=agent, task=None, tools=None, task_prompt=prompt
+    )
+    handlers.handle_agent_execution_started(providers, ctx, agent, started)
+    # The completed handler waits (up to 5 s) for the agent's LLM-call count to
+    # settle; one recorded call is what a real run would have signalled.
+    handlers._record_agent_llm_call(ctx, str(agent.id))
+    handlers.handle_agent_execution_completed(
+        providers,
+        ctx,
+        agent,
+        AgentExecutionCompletedEvent(
+            agent=agent, task=None, output=output, started_event_id=started.event_id
+        ),
+    )
+    return _only_span(exporter, "execute agent")
+
+
+def test_execute_agent_span_carries_the_exact_prompt_and_answer(pipeline) -> None:
+    span = _agent_span(pipeline, prompt=LONG_TEXT, output="The fox is quick.")
+
+    # The same spec shape the task span already uses for its own text.
+    prompt = json.loads(span.attributes["gen_ai.input.messages"])
+    assert prompt[0]["parts"][0]["content"] == LONG_TEXT
+    assert "gen_ai.input.messages.truncated" not in span.attributes
+    answer = json.loads(span.attributes["gen_ai.output.messages"])
+    assert answer[0]["parts"][0]["content"] == "The fox is quick."
+    assert "gen_ai.output.messages.truncated" not in span.attributes
+
+
+def test_the_agent_text_travels_under_the_standard_keys_only(pipeline) -> None:
+    """The prompt and the answer leave under the two keys the ``call llm`` span
+    already uses for its messages, and under no other. Whatever rule an
+    exporter or a redaction processor applies to LLM message content applies
+    to these unchanged; a copy under a ``crewai.agent.*`` key would escape a
+    rule that selects by key name."""
+    span = _agent_span(pipeline, prompt="PROMPT-7597", output="ANSWER-7597")
+
+    carrying = {
+        key
+        for key, value in span.attributes.items()
+        if "PROMPT-7597" in str(value) or "ANSWER-7597" in str(value)
+    }
+    assert carrying == {"gen_ai.input.messages", "gen_ai.output.messages"}
+
+
+def test_an_over_cap_prompt_is_declared_truncated_never_silently_cut(
+    pipeline, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    cap = 2048
+    monkeypatch.setenv("CREWAI_OTEL_MAX_ATTR_BYTES", str(cap))
+
+    span = _agent_span(pipeline, prompt=LONG_TEXT, output="short")
+
+    payload = span.attributes["gen_ai.input.messages"]
+    assert span.attributes["gen_ai.input.messages.truncated"] is True
+    assert span.attributes["gen_ai.input.messages.original_size_bytes"] > cap
+    assert len(payload.encode("utf-8")) <= cap
+    # The answer fit, so it arrives whole and unmarked.
+    answer = json.loads(span.attributes["gen_ai.output.messages"])
+    assert answer[0]["parts"][0]["content"] == "short"
+    assert "gen_ai.output.messages.truncated" not in span.attributes
+
+
+@pytest.mark.parametrize("from_cache", [True, False], ids=["cached", "ran"])
+def test_call_tool_span_says_whether_the_cache_answered(pipeline, from_cache) -> None:
+    providers, ctx, exporter = pipeline
+    now = datetime.now(timezone.utc)
+
+    started = ToolUsageStartedEvent(
+        tool_name="search", tool_args={"q": "fox"}, agent_key="k", agent_role="r"
+    )
+    handlers.handle_tool_usage_started(providers, ctx, None, started)
+    handlers.handle_tool_usage_finished(
+        providers,
+        ctx,
+        None,
+        ToolUsageFinishedEvent(
+            tool_name="search",
+            tool_args={"q": "fox"},
+            agent_key="k",
+            agent_role="r",
+            started_at=now,
+            finished_at=now,
+            output="found",
+            from_cache=from_cache,
+            started_event_id=started.event_id,
+        ),
+    )
+
+    span = _only_span(exporter, "call tool")
+    assert span.attributes["crewai.tool.from_cache"] is from_cache