From 2a3e64deb769f4f850baeceb3f62ba26f908bd2d Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Fri, 28 Aug 2026 19:11:26 -0700 Subject: [PATCH 1/2] Fix tool-call extraction and populate events on the run() path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tool tasks were identified by a `call_` reference-name prefix. That prefix is the LLM provider's tool-call id format, not something the server adds, so an Anthropic-backed agent (`toolu_…`) recorded no tool calls at all. Five tool kinds were also filtered out as system tasks and five more were named from the task type, and `run()` never passed `events=` at all. Identify a tool by the task type it compiled to, per the server's `ToolCompiler.TYPE_MAP`, and resolve its name from `_agent_tool_name` — which the tool-dispatch script sets on every tool kind — falling back to `method` and the task definition. Nothing reads the reference name for identity, so nothing depends on provider-controlled data. Names are no longer case-folded, so `getWeather` stops arriving as `getweather`. `agent_tool` and a strategy handoff both compile to SUB_WORKFLOW; only the tool carries `_agent_tool_name`, which is what tells them apart. The agent's own statically-compiled workers — guardrails, callbacks, gates, routing — compile to SIMPLE tasks like a worker tool does, so they are excluded by name; that check is consulted only after the dispatch key has failed to settle the question. `run()` and `run_async()` now derive `events` from the execution they already fetch for `tool_calls`, with no extra call. Eight assertions in `conductor.ai.agents.testing` read `events`, and four of them check for absence — including `assert_no_errors`, which the eval runner runs on every case — so they were passing without evidence. `expect_handoff_to`, used by the eval runner's own documented example, was failing every time. The task-to-event mapping was copy-pasted across the sync and async polling streams; both now share `_task_events`, which is also what `run()` uses, so an assertion reads the same whichever way the agent was run. Verified against a live server: before, an Anthropic-backed agent with one tool returned `tool_calls: []` and `events: []`; after, `[('getWeather', {'city': 'Tokyo'})]` and a full event list. The OpenAI path is unchanged apart from the name no longer being lowercased. --- CHANGELOG.md | 3 + src/conductor/ai/agents/result.py | 23 +- src/conductor/ai/agents/runtime/runtime.py | 657 ++++++++++++--------- tests/unit/ai/test_runtime.py | 2 +- tests/unit/ai/test_tool_extraction.py | 500 ++++++++++++++++ 5 files changed, 891 insertions(+), 294 deletions(-) create mode 100644 tests/unit/ai/test_tool_extraction.py diff --git a/CHANGELOG.md b/CHANGELOG.md index adcb49dd..dc1f4e8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,3 +29,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Worker processes killed by a signal now log a diagnostic hint (signal number, `PYTHONFAULTHANDLER=1` guidance) instead of restarting silently - Per-schedule pause/resume now work on both Conductor server families: the client sends `PUT` (the OSS Conductor dialect — the spec-generated `GET` failed there) and transparently falls back to `GET` on a 405 for Orkes servers - `Agent(model="claude-code/...")` registration no longer crashes with `SpawnSafetyError`/`PicklingError` under the `spawn` start method (top-level or nested as a sub-agent); the tool worker is now built through the same picklable passthrough path already used by other frameworks +- `AgentResult.tool_calls` no longer drops or misnames tool calls. A tool task was identified by a `call_` reference-name prefix, which is the LLM provider's tool-call id format — so an Anthropic-backed agent (`toolu_…`) recorded no tool calls at all. Tools are now identified by the task type they compiled to, and named from `_agent_tool_name` (set by the server on every tool kind), `method`, or the task definition. Consequences: `http`, `api`, `mcp`, `agent_tool`, `human` and `pull_workflow_messages` tools now appear at all; media and RAG tools report their own name instead of `generate_image`/`llm_index_text`; and a camelCase worker name is no longer lowercased, so `assert_tool_used(result, "getWeather")` matches +- `AgentResult.events` is now populated by `run()` and `run_async()` for natively-compiled agents, derived from the same execution already fetched for `tool_calls` rather than from an extra call. It was previously only filled in when an `on_event` callback or `stream()` was used, so the eight `conductor.ai.agents.testing` assertions that read events — `assert_no_errors`, `assert_handoff_to`, `assert_events_contain`, `assert_max_turns` and the rest — saw an empty list. Four of them check for *absence* and so passed vacuously, `assert_no_errors` on every eval case (`expect_no_errors` defaults to `True`); `expect_handoff_to`, used by the eval runner's own documented example, failed every time. Foreign-framework agents (LangChain, LangGraph, OpenAI Agents, Claude Agent SDK) are unchanged: their events come from the stream, since the whole agent runs inside one passthrough task that polling cannot decompose +- Tool arguments reported on `AgentEvent.args` and in `AgentResult.tool_calls` now agree, and both strip every key Conductor injects — previously `_agent_tool_name`, `_allowed_commands`, `__humanTaskDefinition` and `__conductor_agent_ctx__` leaked into one surface or the other diff --git a/src/conductor/ai/agents/result.py b/src/conductor/ai/agents/result.py index a06930bd..99fe004e 100644 --- a/src/conductor/ai/agents/result.py +++ b/src/conductor/ai/agents/result.py @@ -106,6 +106,8 @@ class AgentResult: error: Human-readable error message when the agent failed. token_usage: Aggregated token usage across all LLM calls. metadata: Extra data from the workflow execution. + events: The execution's event history (:class:`AgentEvent`), whether the + agent was streamed or run to completion. sub_results: Per-agent outputs for multi-agent strategies (parallel). Empty dict for single-agent runs. Keyed by agent name. """ @@ -691,6 +693,22 @@ def __repr__(self) -> str: # ── AgentEvent (yielded by stream()) ─────────────────────────────────── +#: Keys Conductor injects into a tool task's input that are not arguments the +#: LLM chose. ``method`` and ``_agent_tool_name`` carry the tool's *name*; +#: the rest are agent-loop plumbing. Shared by :class:`AgentEvent` and the +#: runtime's ``tool_calls`` extraction so both report the same arguments. +INTERNAL_ARG_KEYS = frozenset( + { + "_agent_state", + "_agent_tool_name", + "_allowed_commands", + "method", + "__humanTaskDefinition", + "__conductor_agent_ctx__", + } +) + + class EventType(str, Enum): """Types of events emitted during agent execution.""" @@ -723,9 +741,6 @@ class AgentEvent: guardrail_name: Guardrail name (for ``guardrail_pass``, ``guardrail_fail``). """ - # Keys injected by Conductor that should not appear in user-facing args. - _INTERNAL_ARG_KEYS = frozenset({"_agent_state", "method"}) - type: str content: Optional[str] = None tool_name: Optional[str] = None @@ -738,7 +753,7 @@ class AgentEvent: def __post_init__(self): if self.args and isinstance(self.args, dict): - cleaned = {k: v for k, v in self.args.items() if k not in self._INTERNAL_ARG_KEYS} + cleaned = {k: v for k, v in self.args.items() if k not in INTERNAL_ARG_KEYS} object.__setattr__(self, "args", cleaned if cleaned else None) diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index e4a1709c..0fc0d9a6 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -34,6 +34,7 @@ AsyncAgentStream, DeploymentInfo, EventType, + INTERNAL_ARG_KEYS, FinishReason, TokenUsage, ) @@ -264,6 +265,314 @@ def _normalize_handoff_target(task_ref: str) -> str: return name +# ── Tool-task identification ─────────────────────────────────────────── +# +# A tool is identified by what the server compiled it *into*, never by the +# task's reference name. Reference names carry the LLM provider's tool-call id +# (``call_…`` for OpenAI, ``toolu_…`` for Anthropic, a UUID otherwise), so +# matching on one silently drops every other provider's tool calls. + +#: Conductor task types a tool compiles to, per the server's +#: ``ToolCompiler.TYPE_MAP``. Two kinds are absent. ``worker`` compiles to +#: SIMPLE, whose type Conductor rewrites to the task's own name on execution, so +#: those are matched by :func:`_is_tool_task`'s task-definition fallback. +#: ``agent_tool`` compiles to SUB_WORKFLOW, which is also how a strategy handoff +#: is compiled, so those are told apart by the tool-name key alone. +_TOOL_TASK_TYPES = frozenset( + { + "HTTP", # http and api tools + "CALL_MCP_TOOL", + "HUMAN", + "PULL_WORKFLOW_MESSAGES", + "GENERATE_IMAGE", + "GENERATE_AUDIO", + "GENERATE_VIDEO", + "LLM_INDEX_TEXT", # rag_index + "LLM_SEARCH_INDEX", # rag_search + } +) + +#: Task types the agent compiler emits that are never a tool invocation. +_NON_TOOL_TASK_TYPES = frozenset( + { + "LLM_CHAT_COMPLETE", + "SWITCH", + "DO_WHILE", + "INLINE", + "SET_VARIABLE", + "FORK", + "FORK_JOIN", + "FORK_JOIN_DYNAMIC", + "JOIN", + "EXCLUSIVE_JOIN", + "TERMINATE", + "LIST_MCP_TOOLS", + "WAIT", + "EVENT", + "DECISION", + "START_WORKFLOW", + "JSON_JQ_TRANSFORM", + } +) + +#: Reference-name prefix of the framework passthrough wrapper task. It wraps a +#: whole foreign-framework agent, which emits its own fine-grained events. +_FRAMEWORK_TASK_REF_PREFIX = "_fw_" + +#: Input key the server's tool-dispatch script sets on every tool task, +#: whatever kind it compiled to. +_TOOL_NAME_KEY = "_agent_tool_name" + +#: Name suffixes of the workers this runtime registers for an agent's own +#: machinery — callbacks, termination conditions, gates, routing. They compile +#: to SIMPLE tasks exactly as a worker tool does, but the LLM never chose to +#: call one, so they are not tool calls. Kept in step with the +#: ``AgentRuntime._register_*_worker`` methods, which are where these names are +#: minted; swarm ``{agent}_transfer_to_{sub}`` workers are deliberately absent, +#: because the LLM does call those. +_AGENT_INTERNAL_TASK_SUFFIXES = ( + "_stop_when", + "_gate", + "_termination", + "_check_transfer", + "_transfer_check", + "_router", + "_router_fn", + "_handoff_check", + "_process_selection", + "_guardrail", + "_before_agent", + "_after_agent", + "_before_model", + "_after_model", + "_before_tool", + "_after_tool", +) + +#: A custom guardrail's worker is named after the user's guardrail rather than +#: after the agent, so it is matched on the reference name the guardrail +#: compiler builds — ``{agent}_{kind}_guardrail_{name}``, optionally +#: ``_worker``-suffixed. Matched with both underscores so a tool the user +#: called ``guardrail_lookup`` is still a tool. +_GUARDRAIL_TASK_MARKER = "_guardrail_" + + +def _is_guardrail_task(ref: str) -> bool: + """Whether a reference name is one the guardrail compiler built.""" + lowered = ref.lower() + return _GUARDRAIL_TASK_MARKER in lowered or lowered.endswith("_guardrail") + + +def _is_agent_internal_task(ref: str, task_def_name: Optional[str]) -> bool: + """Whether a task is the agent's own machinery rather than a tool call. + + Name-based, and deliberately so — these tasks are compiled statically and + carry nothing else to tell them apart from a worker tool. It is the one + place a name is read for identity, and it is safe because a dispatched tool + is settled by :data:`_TOOL_NAME_KEY` before this is ever consulted. + """ + if _is_guardrail_task(ref): + return True + return any( + name.lower().endswith(_AGENT_INTERNAL_TASK_SUFFIXES) + for name in (ref, task_def_name or "") + ) + + +def _is_tool_task(task: Any) -> bool: + """Whether an execution task is a tool invocation.""" + ref = str(getattr(task, "reference_task_name", "") or "") + if ref.startswith(_FRAMEWORK_TASK_REF_PREFIX): + return False + + # The server's tool-dispatch script stamps the tool-name key on every tool + # it dispatches and on nothing else, so its presence settles the question + # outright — including for a tool whose own name happens to end like one of + # the agent-internal suffixes below. + if _TOOL_NAME_KEY in (getattr(task, "input_data", None) or {}): + return True + + task_def_name = getattr(task, "task_def_name", None) + if _is_agent_internal_task(ref, task_def_name if isinstance(task_def_name, str) else None): + return False + + task_type = str(getattr(task, "task_type", "") or "").upper() + if task_type in _NON_TOOL_TASK_TYPES: + return False + + # A SUB_WORKFLOW without the key above is a strategy handoff, not an + # ``agent_tool``. + if task_type == "SUB_WORKFLOW": + return False + + if task_type in _TOOL_TASK_TYPES: + return True + + # Worker tools: Conductor rewrites an executed SIMPLE task's type to the + # task's own name, so a worker tool's type is unenumerable and anything left + # with a task definition behind it is one. That makes an unrecognised task + # type read as a tool rather than vanish — the safer way round, because a + # missing tool call is what makes an assertion pass without evidence. + return task_type == "SIMPLE" or task_def_name is not None + + +def _tool_name(task: Any) -> str: + """Resolve a tool task's name from a field that actually carries it. + + Never case-folds: ``getWeather`` is a different tool from ``getweather`` to + every assertion that compares names. + """ + input_data = getattr(task, "input_data", None) or {} + for key in (_TOOL_NAME_KEY, "method"): + value = input_data.get(key) + if isinstance(value, str) and value: + return value + + task_def_name = getattr(task, "task_def_name", None) + if isinstance(task_def_name, str) and task_def_name: + return task_def_name + + return str(getattr(task, "task_type", "") or "") + + +def _tool_args(task: Any) -> Dict[str, Any]: + """A tool task's input with Conductor's own injected keys removed.""" + input_data = getattr(task, "input_data", None) or {} + if not isinstance(input_data, dict): + return {} + return {k: v for k, v in input_data.items() if k not in INTERNAL_ARG_KEYS} + + +# ── Task-to-event mapping ────────────────────────────────────────────── + + +def _task_events(task: Any, execution_id: str) -> Iterator[AgentEvent]: + """Yield the events a single execution task represents. + + The one place that knows how a Conductor task maps onto an + :class:`AgentEvent`, shared by the polling streams and by + :meth:`AgentRuntime._extract_events`. + """ + task_type = str(getattr(task, "task_type", "") or "").upper() + task_ref = str(getattr(task, "reference_task_name", "") or "") + task_status = str(getattr(task, "status", "") or "").upper() + output_data = getattr(task, "output_data", None) or {} + + # LLM task -> THINKING + if "LLM_CHAT_COMPLETE" in task_type: + yield AgentEvent( + type=EventType.THINKING, + content=f"LLM processing ({task_ref})", + execution_id=execution_id, + ) + return + + # Dispatch task with function -> TOOL_CALL + TOOL_RESULT (local compile) + if "dispatch" in task_ref.lower() and task_status == "COMPLETED": + fn_name = output_data.get("function") + if fn_name: + yield AgentEvent( + type=EventType.TOOL_CALL, + tool_name=fn_name, + args=output_data.get("parameters"), + execution_id=execution_id, + ) + yield AgentEvent( + type=EventType.TOOL_RESULT, + tool_name=fn_name, + result=output_data.get("result"), + execution_id=execution_id, + ) + return + + # Guardrail task -> GUARDRAIL_PASS or GUARDRAIL_FAIL + if _is_guardrail_task(task_ref) and task_status == "COMPLETED": + passed = output_data.get("passed") + if passed is not None: + guardrail_name = output_data.get("guardrail_name", task_ref) + if passed: + yield AgentEvent( + type=EventType.GUARDRAIL_PASS, + guardrail_name=guardrail_name, + execution_id=execution_id, + ) + else: + yield AgentEvent( + type=EventType.GUARDRAIL_FAIL, + guardrail_name=guardrail_name, + content=output_data.get("message", ""), + execution_id=execution_id, + ) + return + + # Tool task -> TOOL_CALL + TOOL_RESULT. Only once the task has completed, + # because a TOOL_RESULT is half of what this pair means; ``tool_calls`` + # deliberately differs and records a tool in any status, so that a failed + # tool still answers ``assert_tool_used``. + is_tool = _is_tool_task(task) + if is_tool and task_status == "COMPLETED": + fn_name = _tool_name(task) + yield AgentEvent( + type=EventType.TOOL_CALL, + tool_name=fn_name, + args=_tool_args(task), + execution_id=execution_id, + ) + yield AgentEvent( + type=EventType.TOOL_RESULT, + tool_name=fn_name, + result=output_data, + execution_id=execution_id, + ) + return + + # Sub-workflow that is not an agent_tool -> HANDOFF. Guarded on ``is_tool`` + # rather than on falling through the branch above, so an agent_tool still + # running does not read as a handoff. + if task_type == "SUB_WORKFLOW" and not is_tool: + yield AgentEvent( + type=EventType.HANDOFF, + target=_normalize_handoff_target(task_ref), + execution_id=execution_id, + ) + return + + # Failed task -> ERROR + if task_status == "FAILED": + reason = output_data.get("reason", "Task failed") + yield AgentEvent( + type=EventType.ERROR, + content=f"Task '{task_ref}' failed: {reason}", + execution_id=execution_id, + ) + + +def _terminal_event(workflow_run: Any, execution_id: str) -> Optional[AgentEvent]: + """The DONE or ERROR event closing a finished execution, if it has finished.""" + raw_status = str(getattr(workflow_run, "status", "") or "").upper() + if raw_status not in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"): + return None + + output = None + raw_output = getattr(workflow_run, "output", None) + if raw_output: + if isinstance(raw_output, dict): + output = raw_output.get("result", raw_output) + else: + output = raw_output + + if raw_status == "COMPLETED": + return AgentEvent(type=EventType.DONE, output=output, execution_id=execution_id) + + reason = getattr(workflow_run, "reason", None) + return AgentEvent( + type=EventType.ERROR, + content=reason if isinstance(reason, str) and reason else f"Execution {raw_status}", + output=output, + execution_id=execution_id, + ) + + # Backward compat alias — SSEUnavailableError is now in conductor.client.agent_client _SSEUnavailableError = SSEUnavailableError @@ -2520,6 +2829,7 @@ def run( # and token_usage — these are not available from the status endpoint. tool_calls: List[Dict[str, Any]] = [] messages: List[Dict[str, Any]] = [] + events: List[AgentEvent] = [] token_usage: Optional[TokenUsage] = None task_failure_reason: Optional[str] = None try: @@ -2529,6 +2839,7 @@ def run( ) tool_calls = self._extract_tool_calls(wf) messages = self._extract_messages(wf) + events = self._extract_events(wf, execution_id) token_usage = self._extract_token_usage(execution_id) if raw_status == "FAILED": task_failure_reason = self._extract_failed_task_reason(wf) @@ -2552,6 +2863,7 @@ def run( tool_calls=tool_calls, messages=messages, token_usage=token_usage, + events=events, sub_results=self._extract_sub_results(output), ) @@ -2612,12 +2924,14 @@ def _run_by_name( tool_calls: List[Dict[str, Any]] = [] messages: List[Dict[str, Any]] = [] + events: List[AgentEvent] = [] token_usage: Optional[TokenUsage] = None task_failure_reason: Optional[str] = None try: wf = self._workflow_client.get_workflow(execution_id, include_tasks=True) tool_calls = self._extract_tool_calls(wf) messages = self._extract_messages(wf) + events = self._extract_events(wf, execution_id) token_usage = self._extract_token_usage(execution_id) if status.status == "FAILED": task_failure_reason = self._extract_failed_task_reason(wf) @@ -2638,6 +2952,7 @@ def _run_by_name( tool_calls=tool_calls, messages=messages, token_usage=token_usage, + events=events, ) def _start_by_name( @@ -2722,6 +3037,7 @@ async def _run_by_name_async( tool_calls: List[Dict[str, Any]] = [] messages: List[Dict[str, Any]] = [] + events: List[AgentEvent] = [] token_usage: Optional[TokenUsage] = None try: wf = await loop.run_in_executor( @@ -2730,6 +3046,7 @@ async def _run_by_name_async( ) tool_calls = self._extract_tool_calls(wf) messages = self._extract_messages(wf) + events = self._extract_events(wf, execution_id) token_usage = self._extract_token_usage(execution_id) except Exception as exc: logger.debug("Could not fetch execution details: %s", exc) @@ -2744,6 +3061,7 @@ async def _run_by_name_async( tool_calls=tool_calls, messages=messages, token_usage=token_usage, + events=events, ) async def _start_by_name_async( @@ -3657,105 +3975,11 @@ def _stream_polling(self, execution_id: str) -> Iterator[AgentEvent]: raw_status = getattr(wf, "status", "UNKNOWN") # Process new/updated tasks - if hasattr(wf, "tasks") and wf.tasks: - for task in wf.tasks: - task_id = getattr(task, "task_id", None) - if task_id and task_id not in seen_task_ids: - seen_task_ids.add(task_id) - task_type = str(getattr(task, "task_type", "")).upper() - task_ref = getattr(task, "reference_task_name", "") - task_status = str(getattr(task, "status", "")).upper() - output_data = getattr(task, "output_data", {}) or {} - - # Built-in Conductor task types (not tool workers) - # LLM task -> THINKING - if "LLM_CHAT_COMPLETE" in task_type: - yield AgentEvent( - type=EventType.THINKING, - content=f"LLM processing ({task_ref})", - execution_id=execution_id, - ) - - # Dispatch task with function -> TOOL_CALL (local compile) - elif "dispatch" in task_ref.lower() and task_status == "COMPLETED": - fn_name = output_data.get("function") - if fn_name: - yield AgentEvent( - type=EventType.TOOL_CALL, - tool_name=fn_name, - args=output_data.get("parameters"), - execution_id=execution_id, - ) - yield AgentEvent( - type=EventType.TOOL_RESULT, - tool_name=fn_name, - result=output_data.get("result"), - execution_id=execution_id, - ) - - # Worker/tool task -> TOOL_CALL + TOOL_RESULT (server compile) - # Server-compiled workflows use the tool function name as - # the task type (e.g. "get_weather") with a "call_" ref. - elif ( - task_ref.startswith("call_") - and task_type not in self._SYSTEM_TASK_TYPES - and task_status == "COMPLETED" - ): - fn_name = task_type.lower() - raw_args = getattr(task, "input_data", None) or {} - clean_args = { - k: v for k, v in raw_args.items() if k != "__conductor_agent_ctx__" - } - yield AgentEvent( - type=EventType.TOOL_CALL, - tool_name=fn_name, - args=clean_args, - execution_id=execution_id, - ) - yield AgentEvent( - type=EventType.TOOL_RESULT, - tool_name=fn_name, - result=output_data, - execution_id=execution_id, - ) - - # Guardrail task -> GUARDRAIL_PASS or GUARDRAIL_FAIL - elif "guardrail" in task_ref.lower() and task_status == "COMPLETED": - passed = output_data.get("passed") - if passed is not None: - g_name = output_data.get("guardrail_name", task_ref) - g_message = output_data.get("message", "") - if passed: - yield AgentEvent( - type=EventType.GUARDRAIL_PASS, - guardrail_name=g_name, - execution_id=execution_id, - ) - else: - yield AgentEvent( - type=EventType.GUARDRAIL_FAIL, - guardrail_name=g_name, - content=g_message, - execution_id=execution_id, - ) - - # SubWorkflow -> HANDOFF - elif "SUB_WORKFLOW" in task_type: - target = _normalize_handoff_target(task_ref) - yield AgentEvent( - type=EventType.HANDOFF, - target=target, - execution_id=execution_id, - ) - - # Failed task -> ERROR - elif task_status == "FAILED": - reason = output_data.get("reason", "Task failed") - yield AgentEvent( - type=EventType.ERROR, - content=f"Task '{task_ref}' failed: {reason}", - execution_id=execution_id, - ) + for task in getattr(wf, "tasks", None) or []: + task_id = getattr(task, "task_id", None) + if task_id and task_id not in seen_task_ids: + seen_task_ids.add(task_id) + yield from _task_events(task, execution_id) # Detect HUMAN and PULL_WORKFLOW_MESSAGES tasks waiting for input has_waiting_human = False @@ -3793,32 +4017,9 @@ def _stream_polling(self, execution_id: str) -> Iterator[AgentEvent]: execution_id=execution_id, ) - if raw_status in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"): - output = None - if hasattr(wf, "output") and wf.output: - output_data = wf.output - if isinstance(output_data, dict): - output = output_data.get("result", output_data) - else: - output = output_data - - if raw_status == "COMPLETED": - yield AgentEvent( - type=EventType.DONE, - output=output, - execution_id=execution_id, - ) - else: - reason = getattr(wf, "reason", None) - error_msg = ( - reason if isinstance(reason, str) and reason else f"Execution {raw_status}" - ) - yield AgentEvent( - type=EventType.ERROR, - content=error_msg, - output=output, - execution_id=execution_id, - ) + terminal = _terminal_event(wf, execution_id) + if terminal is not None: + yield terminal break # Don't busy-poll while waiting for human input @@ -3977,6 +4178,7 @@ async def run_async( # and token_usage — these are not available from the status endpoint. tool_calls: List[Dict[str, Any]] = [] messages: List[Dict[str, Any]] = [] + events: List[AgentEvent] = [] token_usage: Optional[TokenUsage] = None try: loop = asyncio.get_event_loop() @@ -3989,6 +4191,7 @@ async def run_async( ) tool_calls = self._extract_tool_calls(wf) messages = self._extract_messages(wf) + events = self._extract_events(wf, execution_id) token_usage = self._extract_token_usage(execution_id) except Exception as exc: logger.debug("Could not fetch execution details for %s: %s", execution_id, exc) @@ -4004,6 +4207,7 @@ async def run_async( tool_calls=tool_calls, messages=messages, token_usage=token_usage, + events=events, sub_results=self._extract_sub_results(output), ) @@ -4194,91 +4398,12 @@ async def _stream_polling_async(self, execution_id: str) -> AsyncIterator[AgentE raw_status = getattr(wf, "status", "UNKNOWN") # Process new/updated tasks - if hasattr(wf, "tasks") and wf.tasks: - for task in wf.tasks: - task_id = getattr(task, "task_id", None) - if task_id and task_id not in seen_task_ids: - seen_task_ids.add(task_id) - task_type = str(getattr(task, "task_type", "")).upper() - task_ref = getattr(task, "reference_task_name", "") - task_status = str(getattr(task, "status", "")).upper() - output_data = getattr(task, "output_data", {}) or {} - - if "LLM_CHAT_COMPLETE" in task_type: - yield AgentEvent( - type=EventType.THINKING, - content=f"LLM processing ({task_ref})", - execution_id=execution_id, - ) - elif "dispatch" in task_ref.lower() and task_status == "COMPLETED": - fn_name = output_data.get("function") - if fn_name: - yield AgentEvent( - type=EventType.TOOL_CALL, - tool_name=fn_name, - args=output_data.get("parameters"), - execution_id=execution_id, - ) - yield AgentEvent( - type=EventType.TOOL_RESULT, - tool_name=fn_name, - result=output_data.get("result"), - execution_id=execution_id, - ) - elif ( - task_ref.startswith("call_") - and task_type not in self._SYSTEM_TASK_TYPES - and task_status == "COMPLETED" - ): - fn_name = task_type.lower() - raw_args = getattr(task, "input_data", None) or {} - clean_args = { - k: v for k, v in raw_args.items() if k != "__conductor_agent_ctx__" - } - yield AgentEvent( - type=EventType.TOOL_CALL, - tool_name=fn_name, - args=clean_args, - execution_id=execution_id, - ) - yield AgentEvent( - type=EventType.TOOL_RESULT, - tool_name=fn_name, - result=output_data, - execution_id=execution_id, - ) - elif "guardrail" in task_ref.lower() and task_status == "COMPLETED": - passed = output_data.get("passed") - if passed is not None: - g_name = output_data.get("guardrail_name", task_ref) - g_message = output_data.get("message", "") - if passed: - yield AgentEvent( - type=EventType.GUARDRAIL_PASS, - guardrail_name=g_name, - execution_id=execution_id, - ) - else: - yield AgentEvent( - type=EventType.GUARDRAIL_FAIL, - guardrail_name=g_name, - content=g_message, - execution_id=execution_id, - ) - elif "SUB_WORKFLOW" in task_type: - target = _normalize_handoff_target(task_ref) - yield AgentEvent( - type=EventType.HANDOFF, - target=target, - execution_id=execution_id, - ) - elif task_status == "FAILED": - reason = output_data.get("reason", "Task failed") - yield AgentEvent( - type=EventType.ERROR, - content=f"Task '{task_ref}' failed: {reason}", - execution_id=execution_id, - ) + for task in getattr(wf, "tasks", None) or []: + task_id = getattr(task, "task_id", None) + if task_id and task_id not in seen_task_ids: + seen_task_ids.add(task_id) + for event in _task_events(task, execution_id): + yield event # Detect HUMAN and PULL_WORKFLOW_MESSAGES tasks waiting for input has_waiting_human = False @@ -4315,32 +4440,9 @@ async def _stream_polling_async(self, execution_id: str) -> AsyncIterator[AgentE execution_id=execution_id, ) - if raw_status in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"): - output = None - if hasattr(wf, "output") and wf.output: - output_data = wf.output - if isinstance(output_data, dict): - output = output_data.get("result", output_data) - else: - output = output_data - - if raw_status == "COMPLETED": - yield AgentEvent( - type=EventType.DONE, - output=output, - execution_id=execution_id, - ) - else: - reason = getattr(wf, "reason", None) - error_msg = ( - reason if isinstance(reason, str) and reason else f"Execution {raw_status}" - ) - yield AgentEvent( - type=EventType.ERROR, - content=error_msg, - output=output, - execution_id=execution_id, - ) + terminal = _terminal_event(wf, execution_id) + if terminal is not None: + yield terminal break if has_waiting_human: @@ -5022,6 +5124,7 @@ def _build_result_from_workflow( tool_calls: List[Dict[str, Any]] = [] messages: List[Dict[str, Any]] = [] + events: List[AgentEvent] = [] token_usage: Optional[TokenUsage] = None task_failure_reason: Optional[str] = None try: @@ -5032,6 +5135,7 @@ def _build_result_from_workflow( ) tool_calls = self._extract_tool_calls(full) messages = self._extract_messages(full) + events = self._extract_events(full, execution_id) token_usage = self._extract_token_usage(execution_id) if raw_status == "FAILED": task_failure_reason = self._extract_failed_task_reason(full) @@ -5052,6 +5156,7 @@ def _build_result_from_workflow( tool_calls=tool_calls, messages=messages, token_usage=token_usage, + events=events, sub_results=self._extract_sub_results(output), ) @@ -5139,66 +5244,40 @@ def _extract_messages(self, workflow_run: Any) -> List[Dict[str, Any]]: last_llm_msgs = msgs return last_llm_msgs - # System task types that are never user-defined tool calls - _SYSTEM_TASK_TYPES = frozenset( - { - "LLM_CHAT_COMPLETE", - "SWITCH", - "DO_WHILE", - "INLINE", - "SET_VARIABLE", - "FORK", - "FORK_JOIN_DYNAMIC", - "JOIN", - "SUB_WORKFLOW", - "HUMAN", - "PULL_WORKFLOW_MESSAGES", - "TERMINATE", - "HTTP", - "CALL_MCP_TOOL", - "LIST_MCP_TOOLS", - "WAIT", - "EVENT", - "DECISION", - } - ) - def _extract_tool_calls(self, workflow_run: Any) -> List[Dict[str, Any]]: """Extract tool call history from execution tasks. - Tool tasks are identified by their reference task name starting with - ``call_`` (the pattern the compiler uses for all tool invocations). + Tool tasks are identified by the task type they compiled to and named + from the input key the server sets on every one — never from the + reference name, which carries the LLM provider's tool-call id. """ - tool_calls: List[Dict[str, Any]] = [] - if not (hasattr(workflow_run, "tasks") and workflow_run.tasks): - return tool_calls - - for task in workflow_run.tasks: - task_type = str(getattr(task, "task_type", "")).upper() - ref = str(getattr(task, "reference_task_name", "")) - - # Skip known system tasks - if task_type in self._SYSTEM_TASK_TYPES: - continue + tasks = getattr(workflow_run, "tasks", None) or [] + return [ + { + "name": _tool_name(task), + "args": _tool_args(task), + "result": getattr(task, "output_data", {}), + } + for task in tasks + if _is_tool_task(task) + ] - # Tool invocation refs follow the pattern call___ - if not ref.startswith("call_"): - continue + def _extract_events(self, workflow_run: Any, execution_id: str) -> List[AgentEvent]: + """Rebuild an execution's event history from its finished task list. - input_data = dict(getattr(task, "input_data", {}) or {}) - # Strip internal Conductor keys from the displayed args - for k in ("_agent_state", "method", "__humanTaskDefinition"): - input_data.pop(k, None) - - tool_calls.append( - { - "name": task_type.lower(), - "args": input_data, - "result": getattr(task, "output_data", {}), - } - ) + ``run()`` polls rather than streams, so its events are derived here from + the same execution it already fetched for ``tool_calls``. Same mapping + as the polling stream, so an assertion reads identically whichever way + the agent was run. + """ + events: List[AgentEvent] = [] + for task in getattr(workflow_run, "tasks", None) or []: + events.extend(_task_events(task, execution_id)) - return tool_calls + terminal = _terminal_event(workflow_run, execution_id) + if terminal is not None: + events.append(terminal) + return events def _fetch_agent_workflow(self, execution_id: str) -> Optional[dict]: """Fetch an execution with its full task list from GET /api/agent/execution/{id}.""" diff --git a/tests/unit/ai/test_runtime.py b/tests/unit/ai/test_runtime.py index c43532da..2b5ace45 100644 --- a/tests/unit/ai/test_runtime.py +++ b/tests/unit/ai/test_runtime.py @@ -835,7 +835,7 @@ def test_empty_tasks(self, runtime): def test_non_tool_tasks_ignored(self, runtime): task = MagicMock() - task.task_type = "SIMPLE" + task.task_type = "INLINE" wf_run = MockWorkflowRun(tasks=[task]) assert runtime._extract_tool_calls(wf_run) == [] diff --git a/tests/unit/ai/test_tool_extraction.py b/tests/unit/ai/test_tool_extraction.py new file mode 100644 index 00000000..7d58d4dc --- /dev/null +++ b/tests/unit/ai/test_tool_extraction.py @@ -0,0 +1,500 @@ +"""Unit tests for tool-call and event extraction from an execution's tasks. + +Covers the two surfaces the ``testing`` assertions read: +:meth:`AgentRuntime._extract_tool_calls` and :meth:`AgentRuntime._extract_events`. + +Tasks are built as plain objects rather than ``MagicMock`` on purpose — a mock +answers every attribute with a truthy stub, which is how a fixture ends up +agreeing with a detection bug instead of catching it. +""" + +from typing import Any, Dict, Optional +from unittest.mock import patch + +import pytest + +from conductor.ai.agents.agent import Agent +from conductor.ai.agents.result import AgentStatus, EventType +from conductor.ai.agents.testing.assertions import assert_max_turns, assert_no_errors + + +class Task: + """A minimal stand-in for ``conductor.client.http.models.Task``.""" + + def __init__( + self, + *, + task_type: str, + reference_task_name: str = "ref", + task_def_name: Optional[str] = None, + input_data: Optional[Dict[str, Any]] = None, + output_data: Optional[Dict[str, Any]] = None, + status: str = "COMPLETED", + task_id: Optional[str] = None, + ): + self.task_type = task_type + self.reference_task_name = reference_task_name + self.task_def_name = task_def_name + self.input_data = input_data or {} + self.output_data = output_data or {} + self.status = status + self.task_id = task_id or reference_task_name + + +class Workflow: + """A minimal stand-in for a workflow execution with tasks.""" + + def __init__(self, tasks=None, status="COMPLETED", output=None, reason=None): + self.tasks = tasks or [] + self.status = status + self.output = output + self.reason = reason + + +@pytest.fixture() +def runtime(): + with patch("conductor.client.orkes_clients.OrkesClients"): + with patch("conductor.ai.agents.runtime.worker_manager.TaskHandler", create=True): + from conductor.ai.agents.runtime.config import AgentConfig + from conductor.ai.agents.runtime.runtime import AgentRuntime + + return AgentRuntime(settings=AgentConfig(auto_start_workers=False)) + + +# ── Tool identity does not depend on the provider's reference name ────── + + +class TestToolCallDetection: + def test_detects_worker_tool_under_an_anthropic_call_id(self, runtime): + """The reference name carries the provider's tool-call id, not ours.""" + task = Task( + task_type="SIMPLE", + reference_task_name="toolu_01PJDP6YvZbhFp3wBnQeC2D3", + task_def_name="Read", + input_data={"file_path": "/tmp/x"}, + output_data={"content": "hello"}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert len(calls) == 1 + assert calls[0]["name"] == "Read" + assert calls[0]["args"] == {"file_path": "/tmp/x"} + assert calls[0]["result"] == {"content": "hello"} + + def test_detects_worker_tool_under_an_openai_call_id(self, runtime): + task = Task( + task_type="get_weather", + reference_task_name="call_PMnNIdOPvm9EQ8e6tn2kbxPY_0__1", + task_def_name="get_weather", + input_data={"city": "NYC"}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == ["get_weather"] + + def test_preserves_camel_case_worker_names(self, runtime): + task = Task( + task_type="getWeather", + reference_task_name="call_abc__0", + task_def_name="getWeather", + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert calls[0]["name"] == "getWeather" + + def test_skips_framework_passthrough_wrapper(self, runtime): + task = Task( + task_type="my_agent_worker", + reference_task_name="_fw_task", + task_def_name="my_agent_worker", + ) + + assert runtime._extract_tool_calls(Workflow([task])) == [] + + @pytest.mark.parametrize( + "task_type", + ["LLM_CHAT_COMPLETE", "SWITCH", "DO_WHILE", "INLINE", "SET_VARIABLE", "JOIN", "TERMINATE"], + ) + def test_skips_system_tasks(self, runtime, task_type): + task = Task(task_type=task_type, task_def_name=task_type.lower()) + + assert runtime._extract_tool_calls(Workflow([task])) == [] + + @pytest.mark.parametrize( + "worker_name", + [ + "support_stop_when", + "support_gate", + "support_termination", + "support_check_transfer", + "support_router_fn", + "support_handoff_check", + "support_process_selection", + "support_before_model", + "support_after_tool", + "support_output_guardrail", + ], + ) + def test_skips_the_agents_own_machinery(self, runtime, worker_name): + """Callbacks, guardrails and routing compile to SIMPLE tasks too.""" + task = Task( + task_type=worker_name, + reference_task_name=worker_name, + task_def_name=worker_name, + ) + + assert runtime._extract_tool_calls(Workflow([task])) == [] + + def test_skips_a_custom_guardrail_worker_named_by_the_user(self, runtime): + task = Task( + task_type="no_profanity", + reference_task_name="support_output_guardrail_no_profanity_worker", + task_def_name="no_profanity", + ) + + assert runtime._extract_tool_calls(Workflow([task])) == [] + + def test_keeps_a_user_tool_whose_name_merely_mentions_guardrails(self, runtime): + task = Task( + task_type="guardrail_lookup", + reference_task_name="toolu_01LOOKUP", + task_def_name="guardrail_lookup", + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == ["guardrail_lookup"] + + def test_dispatched_tool_wins_over_an_internal_name_suffix(self, runtime): + """A user tool may legitimately be called ``open_gate``.""" + task = Task( + task_type="open_gate", + reference_task_name="toolu_01GATE", + task_def_name="open_gate", + input_data={"_agent_tool_name": "open_gate", "door": "front"}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == ["open_gate"] + assert calls[0]["args"] == {"door": "front"} + + def test_keeps_swarm_transfer_tools(self, runtime): + """``transfer_to_x`` is a tool the LLM chose to call, not machinery.""" + task = Task( + task_type="support_transfer_to_billing", + reference_task_name="call_abc__0", + task_def_name="support_transfer_to_billing", + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == ["support_transfer_to_billing"] + + +class TestToolKindsAndNaming: + """Every row of the server's ``ToolCompiler.TYPE_MAP`` is a tool call.""" + + @pytest.mark.parametrize( + ("task_type", "tool_name"), + [ + ("HTTP", "fetch_quote"), + ("CALL_MCP_TOOL", "list_files"), + ("HUMAN", "ask_question"), + ("PULL_WORKFLOW_MESSAGES", "await_message"), + ("GENERATE_IMAGE", "draw_logo"), + ("GENERATE_AUDIO", "narrate"), + ("GENERATE_VIDEO", "animate"), + ("LLM_INDEX_TEXT", "index_docs"), + ("LLM_SEARCH_INDEX", "search_docs"), + ], + ) + def test_non_worker_tool_kinds_are_detected_and_named(self, runtime, task_type, tool_name): + task = Task( + task_type=task_type, + reference_task_name="whatever_0", + task_def_name=task_type.lower(), + input_data={"_agent_tool_name": tool_name, "query": "q"}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == [tool_name] + assert calls[0]["args"] == {"query": "q"} + + def test_agent_tool_sub_workflow_is_a_tool_call(self, runtime): + task = Task( + task_type="SUB_WORKFLOW", + reference_task_name="call_x__0", + input_data={"_agent_tool_name": "research_agent", "prompt": "find it"}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert [c["name"] for c in calls] == ["research_agent"] + + def test_handoff_sub_workflow_is_not_a_tool_call(self, runtime): + """A strategy handoff has no ``_agent_tool_name``; it is not a tool.""" + task = Task(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") + + assert runtime._extract_tool_calls(Workflow([task])) == [] + + def test_mcp_tool_name_falls_back_to_method(self, runtime): + task = Task( + task_type="CALL_MCP_TOOL", + task_def_name="call_mcp_tool", + input_data={"method": "read_file", "arguments": {"path": "/tmp/x"}}, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert calls[0]["name"] == "read_file" + + def test_internal_keys_are_stripped_from_args(self, runtime): + task = Task( + task_type="get_weather", + task_def_name="get_weather", + input_data={ + "city": "NYC", + "_agent_tool_name": "get_weather", + "_agent_state": {"turn": 1}, + "_allowed_commands": ["ls"], + "method": "get_weather", + "__humanTaskDefinition": {}, + "__conductor_agent_ctx__": {}, + }, + ) + + calls = runtime._extract_tool_calls(Workflow([task])) + + assert calls[0]["args"] == {"city": "NYC"} + + +# ── Events on the default (non-streaming) path ────────────────────────── + + +class TestExtractEvents: + def test_tool_task_yields_call_and_result(self, runtime): + task = Task( + task_type="SIMPLE", + reference_task_name="toolu_01ABC", + task_def_name="Read", + input_data={"file_path": "/tmp/x"}, + output_data={"content": "hi"}, + ) + + events = runtime._extract_events(Workflow([task]), "wf-1") + by_type = [e.type for e in events] + + assert EventType.TOOL_CALL in by_type + assert EventType.TOOL_RESULT in by_type + call = next(e for e in events if e.type == EventType.TOOL_CALL) + assert call.tool_name == "Read" + assert call.args == {"file_path": "/tmp/x"} + assert call.execution_id == "wf-1" + + def test_llm_task_yields_thinking(self, runtime): + task = Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0") + + events = runtime._extract_events(Workflow([task]), "wf-1") + + assert [e.type for e in events] == [EventType.THINKING, EventType.DONE] + + def test_handoff_sub_workflow_yields_handoff(self, runtime): + task = Task(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") + + events = runtime._extract_events(Workflow([task]), "wf-1") + handoffs = [e for e in events if e.type == EventType.HANDOFF] + + assert [e.target for e in handoffs] == ["billing"] + + def test_running_agent_tool_is_not_reported_as_a_handoff(self, runtime): + """An ``agent_tool`` is a SUB_WORKFLOW; only a strategy handoff is one.""" + task = Task( + task_type="SUB_WORKFLOW", + reference_task_name="toolu_01SUB", + status="IN_PROGRESS", + input_data={"_agent_tool_name": "research_agent"}, + ) + + events = runtime._extract_events(Workflow([task], status="RUNNING"), "wf-1") + + assert [e.type for e in events if e.type == EventType.HANDOFF] == [] + + def test_guardrail_task_yields_pass_and_fail(self, runtime): + ok = Task( + task_type="SIMPLE", + reference_task_name="support_regex_guardrail_pii", + output_data={"passed": True, "guardrail_name": "pii"}, + ) + bad = Task( + task_type="SIMPLE", + reference_task_name="support_llm_guardrail_tone", + output_data={"passed": False, "guardrail_name": "tone", "message": "rude"}, + ) + + events = runtime._extract_events(Workflow([ok, bad]), "wf-1") + + assert [e.guardrail_name for e in events if e.type == EventType.GUARDRAIL_PASS] == ["pii"] + assert [e.guardrail_name for e in events if e.type == EventType.GUARDRAIL_FAIL] == ["tone"] + + def test_completed_workflow_ends_with_done(self, runtime): + wf = Workflow([], status="COMPLETED", output={"result": "42"}) + + events = runtime._extract_events(wf, "wf-1") + + assert events[-1].type == EventType.DONE + assert events[-1].output == "42" + + def test_failed_task_yields_error(self, runtime): + task = Task( + task_type="get_weather", + task_def_name="get_weather", + status="FAILED", + output_data={"reason": "boom"}, + ) + + events = runtime._extract_events(Workflow([task], status="FAILED"), "wf-1") + errors = [e for e in events if e.type == EventType.ERROR] + + assert "boom" in errors[0].content + assert errors[-1].content == "Execution FAILED" + + +# ── run() wires the events through ────────────────────────────────────── + + +class TestRunPopulatesEvents: + def test_run_populates_events_without_an_on_event_callback(self, runtime): + agent = Agent(name="test", model="openai/gpt-4o") + runtime._prepare_workers = lambda *a, **k: None + runtime._start_via_server = lambda *a, **k: ("wf-events", None, []) + runtime._poll_status_until_complete = lambda *a, **k: AgentStatus( + execution_id="wf-events", + is_complete=True, + output={"result": "sunny", "finishReason": "STOP"}, + status="COMPLETED", + ) + + wf = Workflow( + [ + Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), + Task( + task_type="SIMPLE", + reference_task_name="toolu_01ABC", + task_def_name="get_weather", + input_data={"city": "NYC"}, + output_data={"temp": 72}, + ), + ], + output={"result": "sunny"}, + ) + wf.variables = {"messages": []} + + calls = [] + + def get_workflow(execution_id, include_tasks=False): + calls.append(execution_id) + return wf + + runtime._workflow_client.get_workflow = get_workflow + + with patch.object(runtime, "_fetch_agent_workflow", return_value=None): + result = runtime.run(agent, "What's the weather?") + + assert [e.type for e in result.events] == [ + EventType.THINKING, + EventType.TOOL_CALL, + EventType.TOOL_RESULT, + EventType.DONE, + ] + assert [tc["name"] for tc in result.tool_calls] == ["get_weather"] + assert calls == ["wf-events"], "events must reuse the execution already fetched" + + assert_no_errors(result) + assert_max_turns(result, 2) + + +# ── The eval runner's documented example ──────────────────────────────── + + +class TestEvalRunnerAgainstAPolledRun: + """The runner calls ``run()`` with no ``on_event``; its checks read events.""" + + @pytest.fixture() + def support_run(self, runtime): + """A handoff run: one tool call, then a handoff to ``billing``.""" + wf = Workflow( + [ + Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), + Task( + task_type="SIMPLE", + reference_task_name="toolu_01LOOKUP", + task_def_name="lookup_order", + input_data={"order_id": "123"}, + output_data={"status": "shipped"}, + ), + Task( + task_type="SUB_WORKFLOW", + reference_task_name="support_handoff_0_billing", + ), + ], + output={"result": "Your refund is on its way."}, + ) + wf.variables = {"messages": []} + + runtime._prepare_workers = lambda *a, **k: None + runtime._start_via_server = lambda *a, **k: ("wf-support", None, []) + runtime._poll_status_until_complete = lambda *a, **k: AgentStatus( + execution_id="wf-support", + is_complete=True, + output={"result": "Your refund is on its way.", "finishReason": "STOP"}, + status="COMPLETED", + ) + runtime._workflow_client.get_workflow = lambda *a, **k: wf + return runtime + + def test_documented_eval_case_passes(self, support_run): + from conductor.ai.agents.testing.eval_runner import CorrectnessEval, EvalCase + + agent = Agent(name="support", model="openai/gpt-4o") + with patch.object(support_run, "_fetch_agent_workflow", return_value=None): + suite = CorrectnessEval(support_run).run( + [ + EvalCase( + name="billing_routes_correctly", + agent=agent, + prompt="I need a refund for order #123", + expect_tools=["lookup_order"], + expect_handoff_to="billing", + expect_output_contains=["refund"], + ) + ] + ) + + failures = [c for case in suite.cases for c in case.checks if not c.passed] + assert failures == [] + + def test_absence_checks_can_still_fail(self, support_run): + """The four assertions that used to pass vacuously now see real evidence.""" + from conductor.ai.agents.testing.eval_runner import CorrectnessEval, EvalCase + + agent = Agent(name="support", model="openai/gpt-4o") + with patch.object(support_run, "_fetch_agent_workflow", return_value=None): + suite = CorrectnessEval(support_run).run( + [ + EvalCase( + name="wrongly_expects_no_billing", + agent=agent, + prompt="I need a refund for order #123", + expect_tools_not_used=["lookup_order"], + expect_no_handoff_to=["billing"], + ) + ] + ) + + failed = {c.check for case in suite.cases for c in case.checks if not c.passed} + assert failed == {"tool_not_used:lookup_order", "no_handoff_to:billing"} From 09cd94544c4bf206f0eb7e47bab43c617082beeb Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Tue, 8 Sep 2026 23:17:12 -0700 Subject: [PATCH 2/2] Populate tool_calls and events for framework agents, tighten internal-task detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Claude Agent SDK injects one task per tool call into the running execution (POST /agent/{id}/tasks) using the Anthropic tool-use id as the reference name, so those tool calls sit in the polled task list. But run() routes every framework agent to _run_framework, which built its result with neither tool_calls nor events, so the very case #501 captured its payloads from still came back empty. All four framework result paths now derive both fields from the execution, as the native path does. LangChain and LangGraph publish their steps to the event stream rather than as tasks, so their per-step detail is still stream-only. _is_agent_internal_task matched its suffixes against the reference name or the task definition name, which dropped an injected tool of the user's own named, say, refresh_gate. It now requires both names to look internal, and strips Conductor's __N turn counter from the reference first — without which an internal worker inside the agent loop was read as a tool call, since only the task definition name saved it. Drops _transfer_check, a suffix nothing mints. Renames the new test doubles to FakeTask/FakeWorkflowRun, so Workflow no longer stands for an execution. Comments on the new code rewritten to the module's own style. --- CHANGELOG.md | 2 +- src/conductor/ai/agents/result.py | 6 +- src/conductor/ai/agents/runtime/runtime.py | 142 +++++++------- tests/unit/ai/test_tool_extraction.py | 209 +++++++++++++++------ 4 files changed, 228 insertions(+), 131 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dc1f4e8f..6a98406b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,5 +30,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Per-schedule pause/resume now work on both Conductor server families: the client sends `PUT` (the OSS Conductor dialect — the spec-generated `GET` failed there) and transparently falls back to `GET` on a 405 for Orkes servers - `Agent(model="claude-code/...")` registration no longer crashes with `SpawnSafetyError`/`PicklingError` under the `spawn` start method (top-level or nested as a sub-agent); the tool worker is now built through the same picklable passthrough path already used by other frameworks - `AgentResult.tool_calls` no longer drops or misnames tool calls. A tool task was identified by a `call_` reference-name prefix, which is the LLM provider's tool-call id format — so an Anthropic-backed agent (`toolu_…`) recorded no tool calls at all. Tools are now identified by the task type they compiled to, and named from `_agent_tool_name` (set by the server on every tool kind), `method`, or the task definition. Consequences: `http`, `api`, `mcp`, `agent_tool`, `human` and `pull_workflow_messages` tools now appear at all; media and RAG tools report their own name instead of `generate_image`/`llm_index_text`; and a camelCase worker name is no longer lowercased, so `assert_tool_used(result, "getWeather")` matches -- `AgentResult.events` is now populated by `run()` and `run_async()` for natively-compiled agents, derived from the same execution already fetched for `tool_calls` rather than from an extra call. It was previously only filled in when an `on_event` callback or `stream()` was used, so the eight `conductor.ai.agents.testing` assertions that read events — `assert_no_errors`, `assert_handoff_to`, `assert_events_contain`, `assert_max_turns` and the rest — saw an empty list. Four of them check for *absence* and so passed vacuously, `assert_no_errors` on every eval case (`expect_no_errors` defaults to `True`); `expect_handoff_to`, used by the eval runner's own documented example, failed every time. Foreign-framework agents (LangChain, LangGraph, OpenAI Agents, Claude Agent SDK) are unchanged: their events come from the stream, since the whole agent runs inside one passthrough task that polling cannot decompose +- `AgentResult.events` is now populated by `run()` and `run_async()` for natively-compiled agents, derived from the same execution already fetched for `tool_calls` rather than from an extra call. It was previously only filled in when an `on_event` callback or `stream()` was used, so the eight `conductor.ai.agents.testing` assertions that read events — `assert_no_errors`, `assert_handoff_to`, `assert_events_contain`, `assert_max_turns` and the rest — saw an empty list. Four of them check for *absence* and so passed vacuously, `assert_no_errors` on every eval case (`expect_no_errors` defaults to `True`); `expect_handoff_to`, used by the eval runner's own documented example, failed every time. Framework agents (LangChain, LangGraph, OpenAI Agents, Claude Agent SDK) also derive both fields from their execution's task list now: the whole agent runs inside one passthrough task, but the Claude Agent SDK injects a task per tool call, so those tool calls appear. LangChain and LangGraph publish their steps to the event stream rather than as tasks, so that detail still reaches `on_event` and `stream()` consumers only - Tool arguments reported on `AgentEvent.args` and in `AgentResult.tool_calls` now agree, and both strip every key Conductor injects — previously `_agent_tool_name`, `_allowed_commands`, `__humanTaskDefinition` and `__conductor_agent_ctx__` leaked into one surface or the other diff --git a/src/conductor/ai/agents/result.py b/src/conductor/ai/agents/result.py index 99fe004e..29ccc92f 100644 --- a/src/conductor/ai/agents/result.py +++ b/src/conductor/ai/agents/result.py @@ -693,10 +693,8 @@ def __repr__(self) -> str: # ── AgentEvent (yielded by stream()) ─────────────────────────────────── -#: Keys Conductor injects into a tool task's input that are not arguments the -#: LLM chose. ``method`` and ``_agent_tool_name`` carry the tool's *name*; -#: the rest are agent-loop plumbing. Shared by :class:`AgentEvent` and the -#: runtime's ``tool_calls`` extraction so both report the same arguments. +# Keys Conductor injects into a tool task's input, not arguments the LLM chose. +# Shared with the runtime's tool_calls extraction so both report the same args. INTERNAL_ARG_KEYS = frozenset( { "_agent_state", diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index 0fc0d9a6..e53b921a 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -17,7 +17,7 @@ import threading import time import uuid -from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, List, Optional, Union +from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, List, Optional, Tuple, Union if TYPE_CHECKING: from conductor.ai.agents.runtime.config import AgentConfig @@ -267,17 +267,13 @@ def _normalize_handoff_target(task_ref: str) -> str: # ── Tool-task identification ─────────────────────────────────────────── # -# A tool is identified by what the server compiled it *into*, never by the -# task's reference name. Reference names carry the LLM provider's tool-call id -# (``call_…`` for OpenAI, ``toolu_…`` for Anthropic, a UUID otherwise), so -# matching on one silently drops every other provider's tool calls. - -#: Conductor task types a tool compiles to, per the server's -#: ``ToolCompiler.TYPE_MAP``. Two kinds are absent. ``worker`` compiles to -#: SIMPLE, whose type Conductor rewrites to the task's own name on execution, so -#: those are matched by :func:`_is_tool_task`'s task-definition fallback. -#: ``agent_tool`` compiles to SUB_WORKFLOW, which is also how a strategy handoff -#: is compiled, so those are told apart by the tool-name key alone. +# Tools are identified by the task type they compiled to, not by reference name: +# reference names carry the provider's tool-call id (`call_…` on OpenAI, `toolu_…` +# on Anthropic), so matching on one drops every other provider's tool calls. + +# Task types a tool compiles to, per the server's ToolCompiler.TYPE_MAP. Worker +# tools (SIMPLE) and agent_tool (SUB_WORKFLOW) share a type with non-tool tasks, +# so _is_tool_task tells those apart by other means. _TOOL_TASK_TYPES = frozenset( { "HTTP", # http and api tools @@ -292,7 +288,7 @@ def _normalize_handoff_target(task_ref: str) -> str: } ) -#: Task types the agent compiler emits that are never a tool invocation. +# Task types the agent compiler emits that are never a tool invocation. _NON_TOOL_TASK_TYPES = frozenset( { "LLM_CHAT_COMPLETE", @@ -315,27 +311,20 @@ def _normalize_handoff_target(task_ref: str) -> str: } ) -#: Reference-name prefix of the framework passthrough wrapper task. It wraps a -#: whole foreign-framework agent, which emits its own fine-grained events. +# Wrapper task around a whole foreign-framework agent, which emits its own events. _FRAMEWORK_TASK_REF_PREFIX = "_fw_" -#: Input key the server's tool-dispatch script sets on every tool task, -#: whatever kind it compiled to. +# Input key the server's tool-dispatch script sets on every tool task. _TOOL_NAME_KEY = "_agent_tool_name" -#: Name suffixes of the workers this runtime registers for an agent's own -#: machinery — callbacks, termination conditions, gates, routing. They compile -#: to SIMPLE tasks exactly as a worker tool does, but the LLM never chose to -#: call one, so they are not tool calls. Kept in step with the -#: ``AgentRuntime._register_*_worker`` methods, which are where these names are -#: minted; swarm ``{agent}_transfer_to_{sub}`` workers are deliberately absent, -#: because the LLM does call those. +# Workers for the agent's own machinery (callbacks, termination, gates, routing): +# they compile to SIMPLE like a worker tool, but the LLM never called them. Keep in +# step with _register_*_worker. Swarm transfer workers belong to the LLM. _AGENT_INTERNAL_TASK_SUFFIXES = ( "_stop_when", "_gate", "_termination", "_check_transfer", - "_transfer_check", "_router", "_router_fn", "_handoff_check", @@ -349,11 +338,9 @@ def _normalize_handoff_target(task_ref: str) -> str: "_after_tool", ) -#: A custom guardrail's worker is named after the user's guardrail rather than -#: after the agent, so it is matched on the reference name the guardrail -#: compiler builds — ``{agent}_{kind}_guardrail_{name}``, optionally -#: ``_worker``-suffixed. Matched with both underscores so a tool the user -#: called ``guardrail_lookup`` is still a tool. +# Custom guardrail workers are named after the guardrail rather than the agent, so +# they are matched on the reference name. Both underscores, so a user's tool named +# guardrail_lookup stays a tool. _GUARDRAIL_TASK_MARKER = "_guardrail_" @@ -366,17 +353,18 @@ def _is_guardrail_task(ref: str) -> bool: def _is_agent_internal_task(ref: str, task_def_name: Optional[str]) -> bool: """Whether a task is the agent's own machinery rather than a tool call. - Name-based, and deliberately so — these tasks are compiled statically and - carry nothing else to tell them apart from a worker tool. It is the one - place a name is read for identity, and it is safe because a dispatched tool - is settled by :data:`_TOOL_NAME_KEY` before this is ever consulted. + Name-based because these tasks are compiled statically and carry nothing + else to tell them apart from a worker tool. Both names have to look + internal: a framework injects tool tasks under a provider-id reference name + (`toolu_…`), so a user tool of its own called ``refresh_gate`` keeps its + reference name to vouch for it. """ if _is_guardrail_task(ref): return True - return any( - name.lower().endswith(_AGENT_INTERNAL_TASK_SUFFIXES) - for name in (ref, task_def_name or "") - ) + compiled_ref = re.sub(r"(__\d+)?(_worker)?$", "", ref.lower()) + if not compiled_ref.endswith(_AGENT_INTERNAL_TASK_SUFFIXES): + return False + return (task_def_name or "").lower().endswith(_AGENT_INTERNAL_TASK_SUFFIXES) def _is_tool_task(task: Any) -> bool: @@ -385,10 +373,8 @@ def _is_tool_task(task: Any) -> bool: if ref.startswith(_FRAMEWORK_TASK_REF_PREFIX): return False - # The server's tool-dispatch script stamps the tool-name key on every tool - # it dispatches and on nothing else, so its presence settles the question - # outright — including for a tool whose own name happens to end like one of - # the agent-internal suffixes below. + # The dispatch script sets this key on tools and nothing else, so it settles + # the question before the name checks below. if _TOOL_NAME_KEY in (getattr(task, "input_data", None) or {}): return True @@ -400,26 +386,23 @@ def _is_tool_task(task: Any) -> bool: if task_type in _NON_TOOL_TASK_TYPES: return False - # A SUB_WORKFLOW without the key above is a strategy handoff, not an - # ``agent_tool``. + # Without that key, a SUB_WORKFLOW is a strategy handoff, not an agent_tool. if task_type == "SUB_WORKFLOW": return False if task_type in _TOOL_TASK_TYPES: return True - # Worker tools: Conductor rewrites an executed SIMPLE task's type to the - # task's own name, so a worker tool's type is unenumerable and anything left - # with a task definition behind it is one. That makes an unrecognised task - # type read as a tool rather than vanish — the safer way round, because a - # missing tool call is what makes an assertion pass without evidence. + # Conductor rewrites an executed SIMPLE task's type to the task's own name, so + # a worker tool's type is unenumerable: anything left with a task definition is + # one. A dropped tool call passes an assertion, so err towards calling it one. return task_type == "SIMPLE" or task_def_name is not None def _tool_name(task: Any) -> str: - """Resolve a tool task's name from a field that actually carries it. + """Resolve a tool task's name from a field that carries it. - Never case-folds: ``getWeather`` is a different tool from ``getweather`` to + Never case-folds: ``getWeather`` and ``getweather`` are different tools to every assertion that compares names. """ input_data = getattr(task, "input_data", None) or {} @@ -449,9 +432,8 @@ def _tool_args(task: Any) -> Dict[str, Any]: def _task_events(task: Any, execution_id: str) -> Iterator[AgentEvent]: """Yield the events a single execution task represents. - The one place that knows how a Conductor task maps onto an - :class:`AgentEvent`, shared by the polling streams and by - :meth:`AgentRuntime._extract_events`. + Shared by the polling streams and :meth:`AgentRuntime._extract_events`, so an + assertion reads the same events whichever way the agent was run. """ task_type = str(getattr(task, "task_type", "") or "").upper() task_ref = str(getattr(task, "reference_task_name", "") or "") @@ -505,10 +487,9 @@ def _task_events(task: Any, execution_id: str) -> Iterator[AgentEvent]: ) return - # Tool task -> TOOL_CALL + TOOL_RESULT. Only once the task has completed, - # because a TOOL_RESULT is half of what this pair means; ``tool_calls`` - # deliberately differs and records a tool in any status, so that a failed - # tool still answers ``assert_tool_used``. + # Tool task -> TOOL_CALL + TOOL_RESULT, only once completed, since the pair + # includes a result. tool_calls records a tool in any status, so a failed tool + # still answers assert_tool_used. is_tool = _is_tool_task(task) if is_tool and task_status == "COMPLETED": fn_name = _tool_name(task) @@ -526,9 +507,8 @@ def _task_events(task: Any, execution_id: str) -> Iterator[AgentEvent]: ) return - # Sub-workflow that is not an agent_tool -> HANDOFF. Guarded on ``is_tool`` - # rather than on falling through the branch above, so an agent_tool still - # running does not read as a handoff. + # Sub-workflow that is not an agent_tool -> HANDOFF. Guarded on is_tool so a + # running agent_tool does not read as a handoff. if task_type == "SUB_WORKFLOW" and not is_tool: yield AgentEvent( type=EventType.HANDOFF, @@ -3196,6 +3176,7 @@ def _run_framework( "Framework agent '%s' completed (execution_id=%s)", agent_name, execution_id ) token_usage = self._extract_token_usage(execution_id) + tool_calls, events = self._extract_tool_calls_and_events(execution_id) return AgentResult( output=output, execution_id=execution_id, @@ -3204,6 +3185,8 @@ def _run_framework( finish_reason=self._derive_finish_reason(raw_status, status.output), error=status.reason if raw_status in ("FAILED", "TERMINATED") else None, token_usage=token_usage, + tool_calls=tool_calls, + events=events, sub_results=self._extract_sub_results(output), ) finally: @@ -3536,6 +3519,7 @@ def _run_framework_with_events( status = self._poll_status_until_complete(execution_id, timeout=timeout) output = self._normalize_output(status.output, status.status, status.reason) token_usage = self._extract_token_usage(execution_id) + tool_calls, _ = self._extract_tool_calls_and_events(execution_id) return AgentResult( output=output, execution_id=execution_id, @@ -3544,6 +3528,7 @@ def _run_framework_with_events( finish_reason=self._derive_finish_reason(status.status, status.output), error=status.reason if status.status in ("FAILED", "TERMINATED") else None, token_usage=token_usage, + tool_calls=tool_calls, events=events, sub_results=self._extract_sub_results(output), ) @@ -4523,6 +4508,10 @@ async def _run_framework_async( output = status.reason output = self._normalize_output(output, status.status, status.reason) token_usage = self._extract_token_usage(execution_id) + loop = asyncio.get_event_loop() + tool_calls, _ = await loop.run_in_executor( + None, lambda: self._extract_tool_calls_and_events(execution_id) + ) return AgentResult( output=output, execution_id=execution_id, @@ -4531,6 +4520,7 @@ async def _run_framework_async( finish_reason=self._derive_finish_reason(status.status, status.output), error=status.reason if status.status in ("FAILED", "TERMINATED") else None, token_usage=token_usage, + tool_calls=tool_calls, events=captured_events, sub_results=self._extract_sub_results(output), ) @@ -4553,6 +4543,10 @@ async def _run_framework_async( "Framework agent '%s' completed (execution_id=%s)", agent_name, execution_id ) token_usage = self._extract_token_usage(execution_id) + loop = asyncio.get_event_loop() + tool_calls, events = await loop.run_in_executor( + None, lambda: self._extract_tool_calls_and_events(execution_id) + ) return AgentResult( output=output, execution_id=execution_id, @@ -4561,6 +4555,8 @@ async def _run_framework_async( finish_reason=self._derive_finish_reason(raw_status, status.output), error=status.reason if raw_status in ("FAILED", "TERMINATED") else None, token_usage=token_usage, + tool_calls=tool_calls, + events=events, sub_results=self._extract_sub_results(output), ) finally: @@ -5263,12 +5259,10 @@ def _extract_tool_calls(self, workflow_run: Any) -> List[Dict[str, Any]]: ] def _extract_events(self, workflow_run: Any, execution_id: str) -> List[AgentEvent]: - """Rebuild an execution's event history from its finished task list. + """Rebuild an execution's event history from its task list. - ``run()`` polls rather than streams, so its events are derived here from - the same execution it already fetched for ``tool_calls``. Same mapping - as the polling stream, so an assertion reads identically whichever way - the agent was run. + ``run()`` polls rather than streams, so events are derived from the + execution it already fetched for ``tool_calls``. """ events: List[AgentEvent] = [] for task in getattr(workflow_run, "tasks", None) or []: @@ -5279,6 +5273,22 @@ def _extract_events(self, workflow_run: Any, execution_id: str) -> List[AgentEve events.append(terminal) return events + def _extract_tool_calls_and_events( + self, execution_id: str + ) -> Tuple[List[Dict[str, Any]], List[AgentEvent]]: + """Tool calls and events derivable from an execution's task list. + + A framework agent runs inside a single passthrough task, but the Claude + Agent SDK injects one task per tool it calls, so the task list can still + carry the tool calls even though this runtime never compiled them. + """ + try: + wf = self._workflow_client.get_workflow(execution_id, include_tasks=True) + except Exception as exc: + logger.debug("Could not fetch execution details for %s: %s", execution_id, exc) + return [], [] + return self._extract_tool_calls(wf), self._extract_events(wf, execution_id) + def _fetch_agent_workflow(self, execution_id: str) -> Optional[dict]: """Fetch an execution with its full task list from GET /api/agent/execution/{id}.""" try: diff --git a/tests/unit/ai/test_tool_extraction.py b/tests/unit/ai/test_tool_extraction.py index 7d58d4dc..80b73588 100644 --- a/tests/unit/ai/test_tool_extraction.py +++ b/tests/unit/ai/test_tool_extraction.py @@ -1,15 +1,12 @@ """Unit tests for tool-call and event extraction from an execution's tasks. -Covers the two surfaces the ``testing`` assertions read: -:meth:`AgentRuntime._extract_tool_calls` and :meth:`AgentRuntime._extract_events`. - -Tasks are built as plain objects rather than ``MagicMock`` on purpose — a mock -answers every attribute with a truthy stub, which is how a fixture ends up -agreeing with a detection bug instead of catching it. +Covers what the testing assertions read: _extract_tool_calls and +_extract_events. Tasks are plain objects rather than MagicMock, which answers +every attribute with a truthy stub and would agree with a detection bug. """ from typing import Any, Dict, Optional -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest @@ -18,7 +15,7 @@ from conductor.ai.agents.testing.assertions import assert_max_turns, assert_no_errors -class Task: +class FakeTask: """A minimal stand-in for ``conductor.client.http.models.Task``.""" def __init__( @@ -41,7 +38,7 @@ def __init__( self.task_id = task_id or reference_task_name -class Workflow: +class FakeWorkflowRun: """A minimal stand-in for a workflow execution with tasks.""" def __init__(self, tasks=None, status="COMPLETED", output=None, reason=None): @@ -67,7 +64,7 @@ def runtime(): class TestToolCallDetection: def test_detects_worker_tool_under_an_anthropic_call_id(self, runtime): """The reference name carries the provider's tool-call id, not ours.""" - task = Task( + task = FakeTask( task_type="SIMPLE", reference_task_name="toolu_01PJDP6YvZbhFp3wBnQeC2D3", task_def_name="Read", @@ -75,7 +72,7 @@ def test_detects_worker_tool_under_an_anthropic_call_id(self, runtime): output_data={"content": "hello"}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert len(calls) == 1 assert calls[0]["name"] == "Read" @@ -83,45 +80,45 @@ def test_detects_worker_tool_under_an_anthropic_call_id(self, runtime): assert calls[0]["result"] == {"content": "hello"} def test_detects_worker_tool_under_an_openai_call_id(self, runtime): - task = Task( + task = FakeTask( task_type="get_weather", reference_task_name="call_PMnNIdOPvm9EQ8e6tn2kbxPY_0__1", task_def_name="get_weather", input_data={"city": "NYC"}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == ["get_weather"] def test_preserves_camel_case_worker_names(self, runtime): - task = Task( + task = FakeTask( task_type="getWeather", reference_task_name="call_abc__0", task_def_name="getWeather", ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert calls[0]["name"] == "getWeather" def test_skips_framework_passthrough_wrapper(self, runtime): - task = Task( + task = FakeTask( task_type="my_agent_worker", reference_task_name="_fw_task", task_def_name="my_agent_worker", ) - assert runtime._extract_tool_calls(Workflow([task])) == [] + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] @pytest.mark.parametrize( "task_type", ["LLM_CHAT_COMPLETE", "SWITCH", "DO_WHILE", "INLINE", "SET_VARIABLE", "JOIN", "TERMINATE"], ) def test_skips_system_tasks(self, runtime, task_type): - task = Task(task_type=task_type, task_def_name=task_type.lower()) + task = FakeTask(task_type=task_type, task_def_name=task_type.lower()) - assert runtime._extract_tool_calls(Workflow([task])) == [] + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] @pytest.mark.parametrize( "worker_name", @@ -140,57 +137,57 @@ def test_skips_system_tasks(self, runtime, task_type): ) def test_skips_the_agents_own_machinery(self, runtime, worker_name): """Callbacks, guardrails and routing compile to SIMPLE tasks too.""" - task = Task( + task = FakeTask( task_type=worker_name, reference_task_name=worker_name, task_def_name=worker_name, ) - assert runtime._extract_tool_calls(Workflow([task])) == [] + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] def test_skips_a_custom_guardrail_worker_named_by_the_user(self, runtime): - task = Task( + task = FakeTask( task_type="no_profanity", reference_task_name="support_output_guardrail_no_profanity_worker", task_def_name="no_profanity", ) - assert runtime._extract_tool_calls(Workflow([task])) == [] + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] def test_keeps_a_user_tool_whose_name_merely_mentions_guardrails(self, runtime): - task = Task( + task = FakeTask( task_type="guardrail_lookup", reference_task_name="toolu_01LOOKUP", task_def_name="guardrail_lookup", ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == ["guardrail_lookup"] def test_dispatched_tool_wins_over_an_internal_name_suffix(self, runtime): """A user tool may legitimately be called ``open_gate``.""" - task = Task( + task = FakeTask( task_type="open_gate", reference_task_name="toolu_01GATE", task_def_name="open_gate", input_data={"_agent_tool_name": "open_gate", "door": "front"}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == ["open_gate"] assert calls[0]["args"] == {"door": "front"} def test_keeps_swarm_transfer_tools(self, runtime): """``transfer_to_x`` is a tool the LLM chose to call, not machinery.""" - task = Task( + task = FakeTask( task_type="support_transfer_to_billing", reference_task_name="call_abc__0", task_def_name="support_transfer_to_billing", ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == ["support_transfer_to_billing"] @@ -213,48 +210,48 @@ class TestToolKindsAndNaming: ], ) def test_non_worker_tool_kinds_are_detected_and_named(self, runtime, task_type, tool_name): - task = Task( + task = FakeTask( task_type=task_type, reference_task_name="whatever_0", task_def_name=task_type.lower(), input_data={"_agent_tool_name": tool_name, "query": "q"}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == [tool_name] assert calls[0]["args"] == {"query": "q"} def test_agent_tool_sub_workflow_is_a_tool_call(self, runtime): - task = Task( + task = FakeTask( task_type="SUB_WORKFLOW", reference_task_name="call_x__0", input_data={"_agent_tool_name": "research_agent", "prompt": "find it"}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert [c["name"] for c in calls] == ["research_agent"] def test_handoff_sub_workflow_is_not_a_tool_call(self, runtime): """A strategy handoff has no ``_agent_tool_name``; it is not a tool.""" - task = Task(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") + task = FakeTask(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") - assert runtime._extract_tool_calls(Workflow([task])) == [] + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] def test_mcp_tool_name_falls_back_to_method(self, runtime): - task = Task( + task = FakeTask( task_type="CALL_MCP_TOOL", task_def_name="call_mcp_tool", input_data={"method": "read_file", "arguments": {"path": "/tmp/x"}}, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert calls[0]["name"] == "read_file" def test_internal_keys_are_stripped_from_args(self, runtime): - task = Task( + task = FakeTask( task_type="get_weather", task_def_name="get_weather", input_data={ @@ -268,7 +265,7 @@ def test_internal_keys_are_stripped_from_args(self, runtime): }, ) - calls = runtime._extract_tool_calls(Workflow([task])) + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) assert calls[0]["args"] == {"city": "NYC"} @@ -278,7 +275,7 @@ def test_internal_keys_are_stripped_from_args(self, runtime): class TestExtractEvents: def test_tool_task_yields_call_and_result(self, runtime): - task = Task( + task = FakeTask( task_type="SIMPLE", reference_task_name="toolu_01ABC", task_def_name="Read", @@ -286,7 +283,7 @@ def test_tool_task_yields_call_and_result(self, runtime): output_data={"content": "hi"}, ) - events = runtime._extract_events(Workflow([task]), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([task]), "wf-1") by_type = [e.type for e in events] assert EventType.TOOL_CALL in by_type @@ -297,52 +294,52 @@ def test_tool_task_yields_call_and_result(self, runtime): assert call.execution_id == "wf-1" def test_llm_task_yields_thinking(self, runtime): - task = Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0") + task = FakeTask(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0") - events = runtime._extract_events(Workflow([task]), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([task]), "wf-1") assert [e.type for e in events] == [EventType.THINKING, EventType.DONE] def test_handoff_sub_workflow_yields_handoff(self, runtime): - task = Task(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") + task = FakeTask(task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing") - events = runtime._extract_events(Workflow([task]), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([task]), "wf-1") handoffs = [e for e in events if e.type == EventType.HANDOFF] assert [e.target for e in handoffs] == ["billing"] def test_running_agent_tool_is_not_reported_as_a_handoff(self, runtime): """An ``agent_tool`` is a SUB_WORKFLOW; only a strategy handoff is one.""" - task = Task( + task = FakeTask( task_type="SUB_WORKFLOW", reference_task_name="toolu_01SUB", status="IN_PROGRESS", input_data={"_agent_tool_name": "research_agent"}, ) - events = runtime._extract_events(Workflow([task], status="RUNNING"), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([task], status="RUNNING"), "wf-1") assert [e.type for e in events if e.type == EventType.HANDOFF] == [] def test_guardrail_task_yields_pass_and_fail(self, runtime): - ok = Task( + ok = FakeTask( task_type="SIMPLE", reference_task_name="support_regex_guardrail_pii", output_data={"passed": True, "guardrail_name": "pii"}, ) - bad = Task( + bad = FakeTask( task_type="SIMPLE", reference_task_name="support_llm_guardrail_tone", output_data={"passed": False, "guardrail_name": "tone", "message": "rude"}, ) - events = runtime._extract_events(Workflow([ok, bad]), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([ok, bad]), "wf-1") assert [e.guardrail_name for e in events if e.type == EventType.GUARDRAIL_PASS] == ["pii"] assert [e.guardrail_name for e in events if e.type == EventType.GUARDRAIL_FAIL] == ["tone"] def test_completed_workflow_ends_with_done(self, runtime): - wf = Workflow([], status="COMPLETED", output={"result": "42"}) + wf = FakeWorkflowRun([], status="COMPLETED", output={"result": "42"}) events = runtime._extract_events(wf, "wf-1") @@ -350,14 +347,14 @@ def test_completed_workflow_ends_with_done(self, runtime): assert events[-1].output == "42" def test_failed_task_yields_error(self, runtime): - task = Task( + task = FakeTask( task_type="get_weather", task_def_name="get_weather", status="FAILED", output_data={"reason": "boom"}, ) - events = runtime._extract_events(Workflow([task], status="FAILED"), "wf-1") + events = runtime._extract_events(FakeWorkflowRun([task], status="FAILED"), "wf-1") errors = [e for e in events if e.type == EventType.ERROR] assert "boom" in errors[0].content @@ -379,10 +376,10 @@ def test_run_populates_events_without_an_on_event_callback(self, runtime): status="COMPLETED", ) - wf = Workflow( + wf = FakeWorkflowRun( [ - Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), - Task( + FakeTask(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), + FakeTask( task_type="SIMPLE", reference_task_name="toolu_01ABC", task_def_name="get_weather", @@ -427,17 +424,17 @@ class TestEvalRunnerAgainstAPolledRun: @pytest.fixture() def support_run(self, runtime): """A handoff run: one tool call, then a handoff to ``billing``.""" - wf = Workflow( + wf = FakeWorkflowRun( [ - Task(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), - Task( + FakeTask(task_type="LLM_CHAT_COMPLETE", reference_task_name="llm_0"), + FakeTask( task_type="SIMPLE", reference_task_name="toolu_01LOOKUP", task_def_name="lookup_order", input_data={"order_id": "123"}, output_data={"status": "shipped"}, ), - Task( + FakeTask( task_type="SUB_WORKFLOW", reference_task_name="support_handoff_0_billing", ), @@ -498,3 +495,95 @@ def test_absence_checks_can_still_fail(self, support_run): failed = {c.check for case in suite.cases for c in case.checks if not c.passed} assert failed == {"tool_not_used:lookup_order", "no_handoff_to:billing"} + +# ── Agent-internal workers, and the tools that look like them ─────────── + + +class TestAgentInternalTasks: + def test_skips_internal_worker_with_a_turn_counter(self, runtime): + """Conductor appends __N inside the agent loop; the name still matches.""" + task = FakeTask( + task_type="support_gate", + reference_task_name="support_gate__1", + task_def_name="support_gate", + ) + + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] + + def test_skips_internal_worker_with_a_worker_suffix(self, runtime): + task = FakeTask( + task_type="support_router", + reference_task_name="support_router_worker", + task_def_name="support_router", + ) + + assert runtime._extract_tool_calls(FakeWorkflowRun([task])) == [] + + def test_keeps_an_injected_tool_whose_name_ends_like_internal_machinery(self, runtime): + """A framework injects tools under a provider id, so the ref vouches for them.""" + task = FakeTask( + task_type="SIMPLE", + reference_task_name="toolu_01PJDP6YvZbhFp3wBnQeC2D3", + task_def_name="refresh_gate", + input_data={"scope": "session"}, + ) + + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) + + assert [c["name"] for c in calls] == ["refresh_gate"] + + def test_keeps_a_dispatched_tool_whose_name_ends_like_internal_machinery(self, runtime): + """The dispatch key settles it before any name is read.""" + task = FakeTask( + task_type="my_router", + reference_task_name="my_router", + task_def_name="my_router", + input_data={"_agent_tool_name": "my_router", "q": "x"}, + ) + + calls = runtime._extract_tool_calls(FakeWorkflowRun([task])) + + assert [c["name"] for c in calls] == ["my_router"] + + def test_bare_simple_task_counts_as_a_tool(self, runtime): + """The server's own isToolTask does the same: a dropped call passes assertions.""" + task = FakeTask(task_type="SIMPLE", reference_task_name="anything") + + assert len(runtime._extract_tool_calls(FakeWorkflowRun([task]))) == 1 + + +# ── Framework agents: tool tasks injected into the execution ──────────── + + +class TestFrameworkExecutionExtraction: + def test_extracts_injected_tool_tasks(self, runtime): + """The Claude Agent SDK injects one task per tool call it makes.""" + wf = FakeWorkflowRun( + [ + FakeTask( + task_type="my_agent_worker", + reference_task_name="_fw_my_agent", + task_def_name="my_agent_worker", + ), + FakeTask( + task_type="SIMPLE", + reference_task_name="toolu_01PJDP6YvZbhFp3wBnQeC2D3", + task_def_name="Read", + input_data={"file_path": "/tmp/x"}, + output_data={"content": "hello"}, + ), + ] + ) + runtime._workflow_client = MagicMock() + runtime._workflow_client.get_workflow.return_value = wf + + tool_calls, events = runtime._extract_tool_calls_and_events("exec-1") + + assert [c["name"] for c in tool_calls] == ["Read"] + assert EventType.TOOL_CALL in [e.type for e in events] + + def test_survives_an_unreachable_execution(self, runtime): + runtime._workflow_client = MagicMock() + runtime._workflow_client.get_workflow.side_effect = RuntimeError("boom") + + assert runtime._extract_tool_calls_and_events("exec-1") == ([], [])