From 4b42a4b3e09c527d09b5459fa8bd6cb7bd8e0add Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Fri, 28 Aug 2026 18:32:09 -0700 Subject: [PATCH 1/4] fix: name every tool kind correctly and populate events when polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit waitForResult() reported an HTTP tool as "HTTP", an MCP tool as "CALL_MCP_TOOL", an agent-as-tool as "SUB_WORKFLOW", a human tool as "HUMAN" and an image tool as "GENERATE_IMAGE" — the tool name came from getTaskType(), which is only the tool's own name for a worker, because Conductor rewrites an executed SIMPLE task's type to the task name. It also missed tool calls entirely on non-OpenAI providers: selection keyed on the reference name starting "call_", which is the provider's tool-call id format, so an Anthropic-backed run (toolu_) recorded none. And getEvents() was hard-coded to null on this path, so a run that called three tools looked like one that did nothing. - identify a tool task by task type, allowlisting off the server's ToolCompiler.TYPE_MAP plus _agent_tool_name, never the reference name - resolve the name from inputData._agent_tool_name, then inputData.method, then getTaskDefName() - synthesize tool_call/tool_result events per tool task and close with a terminal done/error event, so both paths report the same call the same way - skip a tool task that has neither finished nor produced output, and report the whole output map for a tool whose output isn't wrapped in "result" - strip every _-prefixed key from a streamed event's args, as the polled path already did extractFromTasks had no coverage at either level; AgentHandleToolExtractionTest covers it through the fromWorkflow seam. --- .../conductor/ai/model/AgentEvent.java | 17 +- .../conductor/ai/model/AgentHandle.java | 209 ++++++++--- .../conductor/ai/model/AgentResult.java | 11 + .../model/AgentHandleToolExtractionTest.java | 332 ++++++++++++++++++ docs/agents/reference/runtime.md | 4 +- 5 files changed, 521 insertions(+), 52 deletions(-) create mode 100644 conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java index 1ac971dbf..db60b3321 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java @@ -152,9 +152,18 @@ public List getPendingToolCalls() { /** * Create an AgentEvent from a raw map (as parsed from SSE JSON). */ - /** Internal keys injected by the server that should not be shown as tool arguments. */ - private static final Set INTERNAL_KEYS = - new HashSet<>(Arrays.asList("_agent_state", "method")); + /** + * Internal keys injected by the server that should not be shown as tool + * arguments, alongside every {@code _}-prefixed key — {@code _agent_state}, + * {@code _agent_tool_name}, {@code _allowed_commands} and whatever the server + * adds next. The polled path strips the same set, so both report one call's + * arguments identically. + */ + private static final Set INTERNAL_KEYS = new HashSet<>(Arrays.asList("method")); + + private static boolean isInternalKey(String key) { + return key != null && (key.startsWith("_") || INTERNAL_KEYS.contains(key)); + } @SuppressWarnings("unchecked") public static AgentEvent fromMap(Map data) { @@ -182,7 +191,7 @@ public static AgentEvent fromMap(Map data) { if (rawArgs != null) { cleanArgs = new LinkedHashMap<>(); for (Map.Entry entry : rawArgs.entrySet()) { - if (!INTERNAL_KEYS.contains(entry.getKey())) { + if (!isInternalKey(entry.getKey())) { cleanArgs.put(entry.getKey(), entry.getValue()); } } diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java index 9e631278e..a10cc3647 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java @@ -16,8 +16,10 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import org.conductoross.conductor.ai.enums.AgentStatus; +import org.conductoross.conductor.ai.enums.EventType; import org.conductoross.conductor.ai.exceptions.WorkerStallError; import org.conductoross.conductor.ai.internal.ServerLivenessMonitor; import org.slf4j.Logger; @@ -346,50 +348,89 @@ private AgentResult buildResult(AgentStatusResponse statusResponse, String workf output = java.util.Collections.singletonMap("result", output); } - // Token usage + tool calls: the server doesn't aggregate either on the - // workflow status response, but every LLM_CHAT_COMPLETE task carries - // tokenUsed/promptTokens/completionTokens in its outputData, and every - // tool-worker SIMPLE task in the workflow corresponds to one LLM tool - // call. Walk the workflow tasks once and aggregate both. + // Token usage, tool calls and events: the server aggregates none of them + // on the workflow status response, but every LLM_CHAT_COMPLETE task + // carries tokenUsed/promptTokens/completionTokens in its outputData and + // every tool task in the workflow is one LLM tool call. Walk the + // workflow tasks once and aggregate all three. // WorkflowClient is the standard Conductor client for /api/workflow/* — // no need to go through AgentClient for this standard endpoint. - TokenUsage tokenUsage = null; - List> toolCalls = new ArrayList<>(); + TaskExtract extract = new TaskExtract(); try { - Workflow workflow = workflowClient.getWorkflow(executionId, true); - TaskExtract extract = extractFromTasks(workflow); - tokenUsage = extract.tokenUsage; - toolCalls = extract.toolCalls; + extract = extractFromTasks(workflowClient.getWorkflow(executionId, true)); } catch (Exception e) { logger.debug("Could not extract tokens/toolCalls for {}: {}", executionId, e.getMessage()); } - return new AgentResult(output, executionId, status, toolCalls, null, tokenUsage, error); + return toResult(extract, output, executionId, status, error); + } + + /** + * Assemble the {@link AgentResult} both non-streaming paths return: what the + * task walk found, closed with the terminal event. + */ + private static AgentResult toResult( + TaskExtract extract, Object output, String executionId, AgentStatus status, String error) { + extract.events.add(terminalEvent(executionId, status, output, error)); + return new AgentResult( + output, executionId, status, extract.toolCalls, extract.events, extract.tokenUsage, error); } - /** Bundles the token usage + tool calls walked out of a workflow's tasks. */ + /** Bundles the token usage, tool calls and events walked out of a workflow's tasks. */ private static final class TaskExtract { TokenUsage tokenUsage; List> toolCalls = new ArrayList<>(); + List events = new ArrayList<>(); } + /** + * Conductor task types the server compiles agent tools to — the values of the + * server's {@code ToolCompiler.TYPE_MAP}, plus {@code GENERATE_PDF}, which + * that map leaves to the upper-cased tool type. + * + *

A tool task is never recognised by its reference name. The server seeds + * that from the provider's own tool-call id, so only OpenAI's happens to + * start {@code call_} — an Anthropic-backed agent records {@code toolu_}, and + * the next provider picks its own format again. + */ + private static final Set TOOL_TASK_TYPES = Set.of( + "SIMPLE", + "HTTP", + "CALL_MCP_TOOL", + "SUB_WORKFLOW", + "HUMAN", + "GENERATE_IMAGE", + "GENERATE_AUDIO", + "GENERATE_VIDEO", + "GENERATE_PDF", + "LLM_INDEX_TEXT", + "LLM_SEARCH_INDEX", + "PULL_WORKFLOW_MESSAGES"); + + /** The one input key the server sets on every tool kind it dispatches. */ + private static final String TOOL_NAME_KEY = "_agent_tool_name"; + + /** The tool name a server-compiled tool task carries, predating {@link #TOOL_NAME_KEY}. */ + private static final String TOOL_METHOD_KEY = "method"; + /** * Walk a workflow's tasks once and aggregate token usage (from - * {@code LLM_CHAT_COMPLETE} tasks) and tool calls (from {@code call_*} - * worker tasks). Shared by both {@link #buildResult} and - * {@link #fromWorkflow(Workflow)} so the extraction lives in one place. + * {@code LLM_CHAT_COMPLETE} tasks) plus the tool calls and their + * {@code tool_call}/{@code tool_result} events (from tool tasks). Shared by + * both {@link #buildResult} and {@link #fromWorkflow(Workflow)} so the + * extraction lives in one place. */ private static TaskExtract extractFromTasks(Workflow workflow) { TaskExtract out = new TaskExtract(); List tasks = workflow != null && workflow.getTasks() != null ? workflow.getTasks() : List.of(); + String executionId = workflow != null && workflow.getWorkflowId() != null ? workflow.getWorkflowId() : ""; int promptT = 0, completionT = 0, totalT = 0; boolean sawTokens = false; for (Task task : tasks) { - String taskType = task.getTaskType(); Map outputData = task.getOutputData(); // LLM task — aggregate tokens - if ("LLM_CHAT_COMPLETE".equals(taskType) && outputData != null) { + if ("LLM_CHAT_COMPLETE".equals(task.getTaskType()) && outputData != null) { promptT += toInt(outputData.get("promptTokens")); completionT += toInt(outputData.get("completionTokens")); totalT += toInt(outputData.get("tokenUsed")); @@ -397,32 +438,34 @@ private static TaskExtract extractFromTasks(Workflow workflow) { continue; } - // Tool worker task — capture name, input args (stripping - // internal runtime fields), and output result. - // referenceTaskName starts with "call_" for LLM-dispatched tool calls. - String refName = task.getReferenceTaskName(); - if (refName != null && refName.startsWith("call_") && outputData != null) { - Map tc = new LinkedHashMap<>(); - tc.put("name", taskType); - Map inputData = task.getInputData(); - if (inputData != null) { - Map cleaned = new LinkedHashMap<>(); - for (Map.Entry e : inputData.entrySet()) { - String k = e.getKey(); - if (k.startsWith("_") - || "method".equals(k) - || "evaluatorType".equals(k) - || "expression".equals(k) - || "ctx".equals(k) - || "workerTag".equals(k) - || "agentConfig".equals(k)) continue; - cleaned.put(k, e.getValue()); - } - tc.put("args", cleaned); - } - tc.put("result", outputData.get("result")); - out.toolCalls.add(tc); + if (!isToolTask(task)) continue; + // A tool task that has neither finished nor produced anything is a + // call the agent has not made yet — a HUMAN tool still waiting on + // its assignee, say. Reporting it would claim a call that has not + // happened. + boolean produced = outputData != null && !outputData.isEmpty(); + if (!produced && (task.getStatus() == null || !task.getStatus().isTerminal())) continue; + + String name = resolveToolName(task); + Map args = toolArgs(task.getInputData()); + // A tool whose output isn't wrapped in "result" — HTTP, MCP — reports + // the whole output map, as the server's own event listener does. Keyed + // on the key being there, so a tool that answers null keeps its null. + Object result = null; + if (produced) { + result = outputData.containsKey("result") ? outputData.get("result") : outputData; } + + Map tc = new LinkedHashMap<>(); + tc.put("name", name); + if (args != null) tc.put("args", args); + tc.put("result", result); + out.toolCalls.add(tc); + + out.events.add( + new AgentEvent(EventType.TOOL_CALL, null, name, args, null, null, executionId, null, null)); + out.events.add( + new AgentEvent(EventType.TOOL_RESULT, null, name, null, result, null, executionId, null, null)); } if (sawTokens) { out.tokenUsage = new TokenUsage(promptT, completionT, totalT); @@ -430,6 +473,81 @@ private static TaskExtract extractFromTasks(Workflow workflow) { return out; } + /** + * Whether a task is one of the agent's tool invocations, as opposed to the + * LLM call, a control-flow task, a guardrail or the approval gate. + */ + private static boolean isToolTask(Task task) { + // Framework passthrough wrappers restate a tool task that is already + // in the list on its own. + String refName = task.getReferenceTaskName(); + if (refName != null && refName.startsWith("_fw_")) return false; + + Map inputData = task.getInputData(); + if (inputData != null && inputData.get(TOOL_NAME_KEY) != null) return true; + + if (!TOOL_TASK_TYPES.contains(task.getTaskType())) return false; + // Typed like a tool but untagged, so it only counts if it names one. + // That keeps out the approval gate's HUMAN task and guardrail workers, + // which share their task types with real tools. + return inputData != null && inputData.get(TOOL_METHOD_KEY) != null; + } + + /** + * The tool's own name, which the server puts in the task's input. Never the + * task type: Conductor overwrites an executed SIMPLE task's type with the + * task's own name, so that reads correctly for a worker tool and reports + * every other kind under its system task type — an HTTP tool as + * {@code "HTTP"}, an MCP tool as {@code "CALL_MCP_TOOL"}. + */ + private static String resolveToolName(Task task) { + Map inputData = task.getInputData(); + if (inputData != null) { + Object toolName = inputData.get(TOOL_NAME_KEY); + if (toolName != null && !toolName.toString().isEmpty()) return toolName.toString(); + Object method = inputData.get(TOOL_METHOD_KEY); + if (method != null && !method.toString().isEmpty()) return method.toString(); + } + // Last resort, and only reachable when the server set one of the keys + // above to an empty string: getTaskDefName() itself falls back to the + // task type, so it is right for a worker tool and no better than the + // old behaviour for anything else. + String taskDefName = task.getTaskDefName(); + return taskDefName != null && !taskDefName.isEmpty() ? taskDefName : null; + } + + /** A tool task's input with the server's internal runtime keys stripped. */ + private static Map toolArgs(Map inputData) { + if (inputData == null) return null; + Map cleaned = new LinkedHashMap<>(); + for (Map.Entry e : inputData.entrySet()) { + String k = e.getKey(); + if (k.startsWith("_") + || TOOL_METHOD_KEY.equals(k) + || "evaluatorType".equals(k) + || "expression".equals(k) + || "ctx".equals(k) + || "workerTag".equals(k) + || "agentConfig".equals(k)) continue; + cleaned.put(k, e.getValue()); + } + return cleaned; + } + + /** + * The event the stream would have ended on. Both non-streaming paths append + * one so that an events list is never empty for a run that happened — + * without it, "no events were collected" and "no events occurred" are the + * same empty list. + */ + private static AgentEvent terminalEvent(String executionId, AgentStatus status, Object output, String error) { + String id = executionId != null ? executionId : ""; + if (status == AgentStatus.COMPLETED) { + return new AgentEvent(EventType.DONE, null, null, null, null, output, id, null, null); + } + return new AgentEvent(EventType.ERROR, error, null, null, null, output, id, null, null); + } + /** * Build an {@link AgentResult} from a terminal {@link Workflow}. * @@ -437,7 +555,7 @@ private static TaskExtract extractFromTasks(Workflow workflow) { * already hold a completed {@link Workflow}. Maps the workflow status to an {@link AgentStatus}, * normalizes the output map, surfaces {@code reasonForIncompletion} as the * error for non-completed runs, and reuses {@link #extractFromTasks} for the - * token-usage and tool-call aggregation. + * token-usage, tool-call and event aggregation. * * @param workflow a finished (or at least populated) workflow; may be null * @return the equivalent {@link AgentResult} @@ -465,8 +583,7 @@ public static AgentResult fromWorkflow(Workflow workflow) { output = java.util.Collections.singletonMap("result", output); } - TaskExtract extract = extractFromTasks(workflow); - return new AgentResult(output, executionId, status, extract.toolCalls, null, extract.tokenUsage, error); + return toResult(extractFromTasks(workflow), output, executionId, status, error); } private static int toInt(Object value) { diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java index 3755ce1fe..1a4101973 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java @@ -66,6 +66,17 @@ public List> getToolCalls() { return toolCalls; } + /** + * The events of the run, oldest first. + * + *

Streaming runs report what the server emitted, which ends in a + * {@code done} or {@code error} event unless the stream was cut short. + * Polled runs reconstruct a {@code tool_call}/{@code tool_result} pair per + * tool task from the workflow record and always close with a terminal + * event — so both paths name the same call the same way, but only the + * streamed list carries the incremental {@code thinking} and + * {@code message} events, which the workflow record does not keep. + */ public List getEvents() { return events; } diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java new file mode 100644 index 000000000..e28366d26 --- /dev/null +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java @@ -0,0 +1,332 @@ +/* + * Copyright 2025 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.conductoross.conductor.ai.model; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import org.conductoross.conductor.ai.enums.EventType; +import org.junit.jupiter.api.Test; + +import com.netflix.conductor.common.metadata.tasks.Task; +import com.netflix.conductor.common.run.Workflow; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Tool-call and event extraction on the non-streaming path, exercised through + * {@link AgentHandle#fromWorkflow(Workflow)} — the seam {@code waitForResult()} + * shares with it. + * + *

The fixtures reproduce what Conductor actually stores: an executed SIMPLE + * task carries the task's own name as its {@code taskType}, every other tool + * kind carries its system task type, the server tags every kind's input with + * {@code _agent_tool_name}, and the reference name is seeded from the LLM + * provider's tool-call id. + */ +class AgentHandleToolExtractionTest { + + private static Task task(String taskType, String refName, Map input, Map output) { + Task t = new Task(); + t.setTaskType(taskType); + t.setReferenceTaskName(refName); + t.setInputData(input); + t.setOutputData(output); + t.setStatus(Task.Status.COMPLETED); + return t; + } + + /** + * A worker tool as the server compiles it: the LLM's arguments sit at the top + * level beside the runtime keys, and Conductor has rewritten {@code taskType} + * to the task's own name by the time the task is executed. + */ + private static Task workerToolTask(String refName, String toolName, Map args) { + Map input = new LinkedHashMap<>(); + input.put("_agent_tool_name", toolName); + input.put("_agent_state", Map.of()); + input.put("method", toolName); + input.putAll(args); + return task(toolName, refName, input, Map.of("result", toolName + "-output")); + } + + /** + * A tool the server compiles to a system task. The LLM's arguments have been + * folded into whatever that task type takes — an HTTP tool's into + * {@code http_request} — and only {@code _agent_tool_name} still names the + * tool. + */ + private static Task systemToolTask( + String taskType, String refName, String toolName, Map compiledInput) { + Map input = new LinkedHashMap<>(); + input.put("_agent_tool_name", toolName); + input.putAll(compiledInput); + return task(taskType, refName, input, Map.of("result", toolName + "-output")); + } + + private static Task llmTask(int prompt, int completion) { + Map out = new LinkedHashMap<>(); + out.put("promptTokens", prompt); + out.put("completionTokens", completion); + out.put("tokenUsed", prompt + completion); + return task("LLM_CHAT_COMPLETE", "chat_0", Map.of(), out); + } + + private static Workflow workflow(Task... tasks) { + Workflow wf = new Workflow(); + wf.setWorkflowId("exec-1"); + wf.setStatus(Workflow.WorkflowStatus.COMPLETED); + wf.setOutput(Map.of("result", "done")); + wf.setTasks(new ArrayList<>(List.of(tasks))); + return wf; + } + + private static List names(AgentResult result) { + return result.getToolCalls().stream() + .map(tc -> (String) tc.get("name")) + .collect(Collectors.toList()); + } + + /** + * Every tool kind is reported under the tool's own name, not the system task + * type it compiles to. COUNTERFACTUAL (pre-fix): the names came from + * {@code getTaskType()}, so this read + * {@code [get_weather, HTTP, CALL_MCP_TOOL, SUB_WORKFLOW, HUMAN, GENERATE_IMAGE]}. + */ + @Test + void namesEveryToolKindAfterTheToolNotItsTaskType() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")), + systemToolTask( + "HTTP", + "call_b_0__1", + "fetch_page", + Map.of("http_request", Map.of("uri", "https://example.com", "method", "GET"))), + systemToolTask("CALL_MCP_TOOL", "call_c_0__1", "search_docs", Map.of("toolInput", Map.of("q", "e"))), + systemToolTask("SUB_WORKFLOW", "call_d_0__1", "billing_agent", Map.of("request", "invoice?")), + systemToolTask("HUMAN", "call_e_0__1", "ask_manager", Map.of("prompt", "approve?")), + systemToolTask("GENERATE_IMAGE", "call_f_0__1", "draw_chart", Map.of("prompt", "a bar chart")))); + + assertEquals( + List.of("get_weather", "fetch_page", "search_docs", "billing_agent", "ask_manager", "draw_chart"), + names(result)); + } + + /** + * Detection does not key on the provider's tool-call id format. COUNTERFACTUAL + * (pre-fix): selection was {@code refName.startsWith("call_")}, so an + * Anthropic-backed run — whose ids start {@code toolu_} — reported no tool + * calls at all. + */ + @Test + void detectsToolCallsWhateverTheProviderIdFormat() { + AgentResult anthropic = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), workerToolTask("toolu_01ABCdef_0__1", "get_weather", Map.of("city", "SF")))); + AgentResult uuidFallback = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + workerToolTask("3f2b1c9e-0d4a-4c7b-9f11-2a6d8e5b7c30_0__1", "get_weather", Map.of()))); + + assertEquals(List.of("get_weather"), names(anthropic)); + assertEquals(List.of("get_weather"), names(uuidFallback)); + } + + /** Internal runtime keys the server injects are not reported as tool arguments. */ + @Test + @SuppressWarnings("unchecked") + void stripsInternalKeysFromArgs() { + AgentResult result = AgentHandle.fromWorkflow( + workflow(workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")))); + + Map args = (Map) result.getToolCalls().get(0).get("args"); + assertEquals(Map.of("city", "SF"), args); + } + + /** + * A tool whose output isn't wrapped in {@code result} reports the whole output + * map, and its args are the compiled request — the server folds an HTTP tool's + * LLM arguments into {@code http_request}, so that is all the workflow record + * keeps of them. + */ + @Test + @SuppressWarnings("unchecked") + void reportsUnwrappedOutputForSystemTaskTools() { + Map httpRequest = Map.of("uri", "https://example.com?q=evals", "method", "GET"); + Map httpOutput = Map.of("response", Map.of("body", "hello"), "statusCode", 200); + Task http = task( + "HTTP", + "call_b_0__1", + Map.of("_agent_tool_name", "fetch_page", "http_request", httpRequest), + httpOutput); + + AgentResult result = AgentHandle.fromWorkflow(workflow(http)); + + Map call = result.getToolCalls().get(0); + assertEquals("fetch_page", call.get("name")); + assertEquals(Map.of("http_request", httpRequest), (Map) call.get("args")); + assertEquals(httpOutput, call.get("result")); + } + + /** + * The last rung of the name resolution. Defensive rather than observed: the + * server tags every tool it dispatches, so this only fires if that tag ever + * arrives blank. + */ + @Test + void fallsBackToTheTaskDefNameWhenTheServerTagIsBlank() { + Task human = task( + "HUMAN", "call_e_0__1", Map.of("_agent_tool_name", ""), Map.of("result", "approved")); + human.setTaskDefName("ask_manager"); + + AgentResult result = AgentHandle.fromWorkflow(workflow(human)); + + assertEquals(List.of("ask_manager"), names(result)); + } + + /** A tool that answers {@code null} keeps its null rather than reporting its own input. */ + @Test + void keepsANullResult() { + Map output = new java.util.HashMap<>(); + output.put("result", null); + Task worker = task("get_weather", "call_a_0__1", Map.of("_agent_tool_name", "get_weather"), output); + + AgentResult result = AgentHandle.fromWorkflow(workflow(worker)); + + assertEquals(1, result.getToolCalls().size()); + assertNull(result.getToolCalls().get(0).get("result")); + } + + /** + * The LLM call, control flow, the approval gate and guardrail workers share + * task types with real tools, and none of them is a tool call. + */ + @Test + void ignoresNonToolTasks() { + Task approvalGate = task( + "HUMAN", + "weather_agent_approval_human", + Map.of("__humanTaskDefinition", Map.of("displayName", "Approve")), + Map.of("approved", true)); + Task guardrail = task("toxicity_guardrail", "weather_agent_guardrail_0", Map.of(), Map.of("passed", true)); + Task frameworkWrapper = task("SIMPLE", "_fw_task", Map.of("_agent_tool_name", "get_weather"), Map.of()); + + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + task("SWITCH", "tool_switch", Map.of(), Map.of()), + task("INLINE", "enrich_tools", Map.of(), Map.of("result", Map.of())), + task("JOIN", "tool_join", Map.of(), Map.of()), + approvalGate, + guardrail, + frameworkWrapper)); + + assertTrue(result.getToolCalls().isEmpty(), "expected no tool calls, got " + names(result)); + } + + /** A tool task that has neither finished nor produced output is not yet a call. */ + @Test + void ignoresAToolTaskStillInFlight() { + Task pendingHuman = new Task(); + pendingHuman.setTaskType("HUMAN"); + pendingHuman.setReferenceTaskName("call_e_0__1"); + pendingHuman.setInputData(Map.of("_agent_tool_name", "ask_manager")); + pendingHuman.setOutputData(Map.of()); + pendingHuman.setStatus(Task.Status.IN_PROGRESS); + + AgentResult result = AgentHandle.fromWorkflow(workflow(pendingHuman)); + + assertTrue(result.getToolCalls().isEmpty()); + } + + /** Token usage still aggregates across LLM tasks. */ + @Test + void aggregatesTokenUsage() { + AgentResult result = AgentHandle.fromWorkflow(workflow(llmTask(10, 5), llmTask(7, 3))); + + assertNotNull(result.getTokenUsage()); + assertEquals(17, result.getTokenUsage().getPromptTokens()); + assertEquals(8, result.getTokenUsage().getCompletionTokens()); + assertEquals(25, result.getTokenUsage().getTotalTokens()); + } + + /** + * COUNTERFACTUAL (pre-fix): {@code events} was hard-coded to {@code null} on + * this path and normalized to an empty list, so a run that called three tools + * was indistinguishable from one that did nothing. + */ + @Test + void populatesEventsOnTheNonStreamingPath() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")), + systemToolTask("HTTP", "call_b_0__1", "fetch_page", Map.of("http_request", Map.of("uri", "u"))))); + + List events = result.getEvents(); + assertEquals( + List.of( + EventType.TOOL_CALL, + EventType.TOOL_RESULT, + EventType.TOOL_CALL, + EventType.TOOL_RESULT, + EventType.DONE), + events.stream().map(AgentEvent::getType).collect(Collectors.toList())); + assertEquals("get_weather", events.get(0).getToolName()); + assertEquals(Map.of("city", "SF"), events.get(0).getArgs()); + assertEquals("get_weather-output", events.get(1).getResult()); + assertEquals("fetch_page", events.get(2).getToolName()); + assertEquals("exec-1", events.get(0).getExecutionId()); + assertEquals(Map.of("result", "done"), events.get(4).getOutput()); + } + + /** A run with no tools still records that it ran, rather than an empty list. */ + @Test + void alwaysRecordsATerminalEvent() { + AgentResult completed = AgentHandle.fromWorkflow(workflow(llmTask(10, 5))); + assertEquals(1, completed.getEvents().size()); + assertEquals(EventType.DONE, completed.getEvents().get(0).getType()); + + Workflow failed = workflow(llmTask(10, 5)); + failed.setStatus(Workflow.WorkflowStatus.FAILED); + failed.setReasonForIncompletion("model unavailable"); + + AgentResult result = AgentHandle.fromWorkflow(failed); + AgentEvent last = result.getEvents().get(result.getEvents().size() - 1); + assertEquals(EventType.ERROR, last.getType()); + assertEquals("model unavailable", last.getContent()); + } + + /** Both paths report the same call under the same name. */ + @Test + void agreesWithTheStreamingPathOnToolNames() { + AgentResult polled = AgentHandle.fromWorkflow( + workflow(workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")))); + + // What the server emits on the stream for the same call: the tool's own + // name, and the task input minus the keys AgentEvent strips. + AgentEvent streamed = AgentEvent.fromMap(Map.of( + "type", "tool_call", + "toolName", "get_weather", + "args", + Map.of( + "city", "SF", + "method", "get_weather", + "_agent_tool_name", "get_weather", + "_agent_state", Map.of()), + "executionId", "exec-1")); + + assertEquals(streamed.getToolName(), polled.getToolCalls().get(0).get("name")); + assertEquals(streamed.getArgs(), polled.getToolCalls().get(0).get("args")); + } +} diff --git a/docs/agents/reference/runtime.md b/docs/agents/reference/runtime.md index 779080c68..aae4e2bb2 100644 --- a/docs/agents/reference/runtime.md +++ b/docs/agents/reference/runtime.md @@ -140,7 +140,7 @@ null, empty, and whitespace-only values are omitted. 1. Workers for the agent's tools are registered with the Conductor task runner. 2. `POST /api/agent/start` — server compiles, registers, and starts the workflow. 3. Polls `GET /api/agent/{id}/status` every 2 seconds until terminal. -4. On completion, calls `GET /api/workflow/{id}` once to aggregate token usage and tool calls into the `AgentResult`. +4. On completion, calls `GET /api/workflow/{id}` once to aggregate token usage, tool calls and events into the `AgentResult`. **Returns `AgentResult`:** @@ -151,7 +151,7 @@ null, empty, and whitespace-only values are omitted. | `getExecutionId()` | `String` | Conductor workflow ID | | `getTokenUsage()` | `TokenUsage` | Aggregated `promptTokens`, `completionTokens`, `totalTokens` | | `getToolCalls()` | `List>` | All tool invocations: `{name, args, result}` | -| `getEvents()` | `List` | Full event log (populated by streaming paths) | +| `getEvents()` | `List` | Event log, oldest first, always ending in `done` or `error`. Streaming paths report what the server emitted; polled runs reconstruct a `tool_call`/`tool_result` pair per tool task, without the incremental `thinking` and `message` events | | `getError()` | `String` | Failure/termination reason when `status != COMPLETED` | | `isSuccess()` | `boolean` | `true` when `status == COMPLETED` | From 668fdcc92497ca4d680217856c1723b2273996f3 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Tue, 8 Sep 2026 16:37:06 -0700 Subject: [PATCH 2/4] fix: find tool calls on a server that sets no tool-name tag Detection rested entirely on the server tagging a tool task's input with _agent_tool_name. Against a server predating that tag, every tool kind whose per-kind compile step replaces the task input carries nothing that names it, so isToolTask matched none of them and the calls vanished from getToolCalls() rather than arriving under a wrong name. Two tool calls became zero. TOOL_TASK_TYPES could not cover for it: an executed worker tool reports its own name as its taskType, so "SIMPLE" in that set never matches a task that ran, and the type-and-method branch only ever reached the system-task kinds. - fall back to the dynamic fork when no task carries the tag: an agent dispatches tool calls through FORK_JOIN_DYNAMIC, and Conductor records the reference names it forked on the fork task's own input - recognise an executed worker tool by its type agreeing with its task-def name - keep the tag authoritative where the server sets it, because an agent can fan out dynamically for reasons of its own, so the fork list is a fallback rather than a second source of truth Behaviour against a tagging server is unchanged. --- .../conductor/ai/model/AgentHandle.java | 94 +++++++++++++-- .../model/AgentHandleToolExtractionTest.java | 109 ++++++++++++++++++ 2 files changed, 196 insertions(+), 7 deletions(-) diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java index a10cc3647..0bcb13041 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java @@ -392,6 +392,10 @@ private static final class TaskExtract { * that from the provider's own tool-call id, so only OpenAI's happens to * start {@code call_} — an Anthropic-backed agent records {@code toolu_}, and * the next provider picks its own format again. + * + *

{@code SIMPLE} is here for a worker tool that was scheduled and never + * ran. One that ran reports its own name as its type instead, which is what + * {@link #isToolTaskType} handles. */ private static final Set TOOL_TASK_TYPES = Set.of( "SIMPLE", @@ -413,6 +417,12 @@ private static final class TaskExtract { /** The tool name a server-compiled tool task carries, predating {@link #TOOL_NAME_KEY}. */ private static final String TOOL_METHOD_KEY = "method"; + /** + * The reference names Conductor forked dynamically, which it records on the + * {@code FORK_JOIN_DYNAMIC} task's own input. + */ + private static final String FORKED_TASKS_KEY = "forkedTasks"; + /** * Walk a workflow's tasks once and aggregate token usage (from * {@code LLM_CHAT_COMPLETE} tasks) plus the tool calls and their @@ -424,6 +434,7 @@ private static TaskExtract extractFromTasks(Workflow workflow) { TaskExtract out = new TaskExtract(); List tasks = workflow != null && workflow.getTasks() != null ? workflow.getTasks() : List.of(); String executionId = workflow != null && workflow.getWorkflowId() != null ? workflow.getWorkflowId() : ""; + ToolTagging tagging = ToolTagging.of(tasks); int promptT = 0, completionT = 0, totalT = 0; boolean sawTokens = false; for (Task task : tasks) { @@ -438,7 +449,7 @@ private static TaskExtract extractFromTasks(Workflow workflow) { continue; } - if (!isToolTask(task)) continue; + if (!isToolTask(task, tagging)) continue; // A tool task that has neither finished nor produced anything is a // call the agent has not made yet — a HUMAN tool still waiting on // its assignee, say. Reporting it would claim a call that has not @@ -477,7 +488,7 @@ private static TaskExtract extractFromTasks(Workflow workflow) { * Whether a task is one of the agent's tool invocations, as opposed to the * LLM call, a control-flow task, a guardrail or the approval gate. */ - private static boolean isToolTask(Task task) { + private static boolean isToolTask(Task task, ToolTagging tagging) { // Framework passthrough wrappers restate a tool task that is already // in the list on its own. String refName = task.getReferenceTaskName(); @@ -485,12 +496,81 @@ private static boolean isToolTask(Task task) { Map inputData = task.getInputData(); if (inputData != null && inputData.get(TOOL_NAME_KEY) != null) return true; + // Where the server tags at all it tags every tool kind, so an untagged + // task is not a tool call and guessing past that only invents calls. + if (tagging.tagged) return false; + + // Untagged server: the tool calls are the tasks it forked dynamically. + if (refName == null || !tagging.dynamicallyForked.contains(refName)) return false; + return isToolTaskType(task); + } - if (!TOOL_TASK_TYPES.contains(task.getTaskType())) return false; - // Typed like a tool but untagged, so it only counts if it names one. - // That keeps out the approval gate's HUMAN task and guardrail workers, - // which share their task types with real tools. - return inputData != null && inputData.get(TOOL_METHOD_KEY) != null; + /** + * Whether a task's type is one an agent tool compiles to. + * + *

{@link #TOOL_TASK_TYPES} alone cannot answer this for a worker tool, + * because by the time the task has run Conductor has rewritten its type from + * {@code SIMPLE} to the task's own name. A task whose type and task-def name + * agree is that case. + */ + private static boolean isToolTaskType(Task task) { + String taskType = task.getTaskType(); + if (taskType == null) return false; + return TOOL_TASK_TYPES.contains(taskType) || taskType.equals(task.getTaskDefName()); + } + + /** + * How one workflow's tool tasks can be told apart, decided once for the + * whole task list. + * + *

A server that tags tool tasks with {@link #TOOL_NAME_KEY} is + * authoritative, and that is the only signal worth having: it names the tool + * as well as identifying it. The tag is set on every tool kind, after the + * per-kind branch that rewrites the task's input, so on such a server tagged + * and tool are the same set. + * + *

Servers predating the tag set it on nothing, and there the fallback is + * the dynamic fork. An agent dispatches its tool calls through + * {@code FORK_JOIN_DYNAMIC}, and Conductor records the reference names it + * forked on the fork task's own input, so a tool task is one the server + * itself reports as dynamically created. Everything the agent compiles for + * its own use is static by comparison, including the guardrail workers, the + * approval gate and a handoff's sub-workflow, which share their task types + * with real tools and would otherwise be counted as tool calls. + * + *

The fallback is a best effort, not a second authority. An agent can + * dynamically fan out for reasons of its own, and on an untagged server such + * a task is indistinguishable from a tool call; the tag is what removes the + * ambiguity, which is why it is preferred whenever the server sets it. + */ + private static final class ToolTagging { + /** Whether the server tagged any task at all with {@link #TOOL_NAME_KEY}. */ + final boolean tagged; + + /** Reference names the server reports having forked dynamically. */ + final Set dynamicallyForked; + + private ToolTagging(boolean tagged, Set dynamicallyForked) { + this.tagged = tagged; + this.dynamicallyForked = dynamicallyForked; + } + + static ToolTagging of(List tasks) { + boolean tagged = false; + Set forked = new java.util.HashSet<>(); + for (Task task : tasks) { + Map inputData = task.getInputData(); + if (inputData == null) continue; + if (inputData.get(TOOL_NAME_KEY) != null) tagged = true; + Object names = inputData.get(FORKED_TASKS_KEY); + if (names instanceof List) { + for (Object name : (List) names) { + if (name != null) forked.add(name.toString()); + } + } + } + return new ToolTagging(tagged, forked); + } } /** diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java index e28366d26..a3dd7cb11 100644 --- a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java @@ -307,6 +307,115 @@ void alwaysRecordsATerminalEvent() { assertEquals("model unavailable", last.getContent()); } + // ── Servers that set no _agent_tool_name ───────────────────────────────── + // + // The tag is set on every tool kind or on none. On a server predating it, + // the per-kind compile step has already replaced the task's input, so an + // HTTP tool carries only its http_request and nothing names the tool. What + // still identifies it is the dynamic fork: Conductor records the reference + // names it forked on the fork task's own input. + + /** A tool task as a pre-tag server dispatches it: no {@code _agent_tool_name}. */ + private static Task untaggedWorkerTool(String refName, String toolName, Map args) { + Map input = new LinkedHashMap<>(); + input.put("_agent_state", Map.of()); + input.put("method", toolName); + input.putAll(args); + return task(toolName, refName, input, Map.of("result", toolName + "-output")); + } + + /** + * A tool the server compiles to a system task, dispatched by a pre-tag + * server: the per-kind step has replaced the input, so the task's own name + * is all that is left of the tool's identity. + */ + private static Task untaggedSystemTool( + String taskType, String refName, String toolName, Map compiledInput) { + Task t = task(taskType, refName, compiledInput, Map.of("result", toolName + "-output")); + t.setTaskDefName(toolName); + return t; + } + + /** The synthetic FORK task Conductor writes when it fans out dynamically. */ + private static Task forkTask(String... forkedRefNames) { + return task("FORK", "agent_fork", Map.of("forkedTasks", List.of(forkedRefNames)), Map.of()); + } + + /** + * A guardrail worker, which is a SIMPLE task like a worker tool and whose + * type is rewritten the same way, but which the agent compiles statically + * and so never appears in the fork list. + */ + private static Task staticGuardrailWorker() { + return task( + "tone_guardrail", + "agent_ext_guardrail_tone", + Map.of("content", "hello", "input", "hello", "iteration", 1), + Map.of("result", "pass")); + } + + /** + * Every tool kind is still found when the server tags none of them, by + * falling back to the reference names it reports having forked. + * COUNTERFACTUAL: without the fork-list fallback, detection rests on the tag + * alone and this reads {@code []} — the tool calls vanish rather than + * arriving under a wrong name. + */ + @Test + void detectsToolCallsWhenTheServerSetsNoToolNameTag() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + forkTask("toolu_01_0__1", "toolu_02_0__1"), + untaggedWorkerTool("toolu_01_0__1", "get_weather", Map.of("city", "SF")), + untaggedSystemTool( + "HTTP", + "toolu_02_0__1", + "fetch_page", + Map.of("http_request", Map.of("uri", "https://example.com"))))); + + assertEquals(List.of("get_weather", "fetch_page"), names(result)); + } + + /** + * The agent's own scaffolding is compiled statically, so it is absent from + * the fork list even when it shares a task type with a real tool. A + * guardrail worker is the awkward case: SIMPLE, and its type rewritten to + * its own name exactly as a worker tool's is. + */ + @Test + void ignoresStaticTasksWhenTheServerSetsNoToolNameTag() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + staticGuardrailWorker(), + task("HUMAN", "agent_guardrail_human", Map.of("__humanTaskDefinition", Map.of()), Map.of()), + task( + "SUB_WORKFLOW", + "agent_transfer_refunds", + Map.of("prompt", "refund please", "session_id", "s1"), + Map.of("result", "handled")), + forkTask("toolu_01_0__1"), + untaggedWorkerTool("toolu_01_0__1", "get_weather", Map.of("city", "SF")))); + + assertEquals(List.of("get_weather"), names(result)); + } + + /** + * Where the server tags, the tag is the whole answer: an untagged task is + * not a tool call even if it was forked dynamically, because an agent can + * fan out for reasons of its own. The fork list is a fallback, not a second + * source of truth. + */ + @Test + void prefersTheTagOverTheForkListWhereTheServerTags() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + forkTask("call_a_0__1", "fanout_0__1"), + workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")), + task("SUB_WORKFLOW", "fanout_0__1", Map.of("prompt", "sub-task"), Map.of("result", "done")))); + + assertEquals(List.of("get_weather"), names(result)); + } + /** Both paths report the same call under the same name. */ @Test void agreesWithTheStreamingPathOnToolNames() { From 2f3af7375d86d5f4f6ae28dea250c68bdc422650 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Tue, 8 Sep 2026 21:38:39 -0700 Subject: [PATCH 3/4] docs: trim tool-extraction comments to the surrounding style --- .../conductor/ai/model/AgentEvent.java | 7 +- .../conductor/ai/model/AgentHandle.java | 137 ++++++------------ .../conductor/ai/model/AgentResult.java | 13 +- .../model/AgentHandleToolExtractionTest.java | 83 ++++------- 4 files changed, 77 insertions(+), 163 deletions(-) diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java index db60b3321..4fc417c9a 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java @@ -153,11 +153,8 @@ public List getPendingToolCalls() { * Create an AgentEvent from a raw map (as parsed from SSE JSON). */ /** - * Internal keys injected by the server that should not be shown as tool - * arguments, alongside every {@code _}-prefixed key — {@code _agent_state}, - * {@code _agent_tool_name}, {@code _allowed_commands} and whatever the server - * adds next. The polled path strips the same set, so both report one call's - * arguments identically. + * Internal keys the server injects, which are not tool arguments. Every + * {@code _}-prefixed key is stripped too, as the polled path does. */ private static final Set INTERNAL_KEYS = new HashSet<>(Arrays.asList("method")); diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java index 0bcb13041..509bf90bd 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java @@ -348,13 +348,9 @@ private AgentResult buildResult(AgentStatusResponse statusResponse, String workf output = java.util.Collections.singletonMap("result", output); } - // Token usage, tool calls and events: the server aggregates none of them - // on the workflow status response, but every LLM_CHAT_COMPLETE task - // carries tokenUsed/promptTokens/completionTokens in its outputData and - // every tool task in the workflow is one LLM tool call. Walk the - // workflow tasks once and aggregate all three. - // WorkflowClient is the standard Conductor client for /api/workflow/* — - // no need to go through AgentClient for this standard endpoint. + // The status response aggregates none of this, so walk the tasks once: + // LLM_CHAT_COMPLETE tasks carry the token counts, tool tasks the calls. + // WorkflowClient is the standard client for /api/workflow/*. TaskExtract extract = new TaskExtract(); try { extract = extractFromTasks(workflowClient.getWorkflow(executionId, true)); @@ -365,10 +361,7 @@ private AgentResult buildResult(AgentStatusResponse statusResponse, String workf return toResult(extract, output, executionId, status, error); } - /** - * Assemble the {@link AgentResult} both non-streaming paths return: what the - * task walk found, closed with the terminal event. - */ + /** Assemble the {@link AgentResult} both non-streaming paths return. */ private static AgentResult toResult( TaskExtract extract, Object output, String executionId, AgentStatus status, String error) { extract.events.add(terminalEvent(executionId, status, output, error)); @@ -384,18 +377,9 @@ private static final class TaskExtract { } /** - * Conductor task types the server compiles agent tools to — the values of the - * server's {@code ToolCompiler.TYPE_MAP}, plus {@code GENERATE_PDF}, which - * that map leaves to the upper-cased tool type. - * - *

A tool task is never recognised by its reference name. The server seeds - * that from the provider's own tool-call id, so only OpenAI's happens to - * start {@code call_} — an Anthropic-backed agent records {@code toolu_}, and - * the next provider picks its own format again. - * - *

{@code SIMPLE} is here for a worker tool that was scheduled and never - * ran. One that ran reports its own name as its type instead, which is what - * {@link #isToolTaskType} handles. + * Task types the server compiles agent tools to: its {@code ToolCompiler.TYPE_MAP} + * values plus {@code GENERATE_PDF}. {@code SIMPLE} only matches a worker tool that + * never ran; see {@link #isToolTaskType}. */ private static final Set TOOL_TASK_TYPES = Set.of( "SIMPLE", @@ -411,24 +395,18 @@ private static final class TaskExtract { "LLM_SEARCH_INDEX", "PULL_WORKFLOW_MESSAGES"); - /** The one input key the server sets on every tool kind it dispatches. */ + /** Tool-name tag the server sets on every tool task's input. Absent on older servers. */ private static final String TOOL_NAME_KEY = "_agent_tool_name"; /** The tool name a server-compiled tool task carries, predating {@link #TOOL_NAME_KEY}. */ private static final String TOOL_METHOD_KEY = "method"; - /** - * The reference names Conductor forked dynamically, which it records on the - * {@code FORK_JOIN_DYNAMIC} task's own input. - */ + /** Reference names Conductor forked, recorded on the {@code FORK_JOIN_DYNAMIC} task's input. */ private static final String FORKED_TASKS_KEY = "forkedTasks"; /** - * Walk a workflow's tasks once and aggregate token usage (from - * {@code LLM_CHAT_COMPLETE} tasks) plus the tool calls and their - * {@code tool_call}/{@code tool_result} events (from tool tasks). Shared by - * both {@link #buildResult} and {@link #fromWorkflow(Workflow)} so the - * extraction lives in one place. + * Walk a workflow's tasks once and aggregate token usage, tool calls and their + * events. Shared by {@link #buildResult} and {@link #fromWorkflow(Workflow)}. */ private static TaskExtract extractFromTasks(Workflow workflow) { TaskExtract out = new TaskExtract(); @@ -450,18 +428,15 @@ private static TaskExtract extractFromTasks(Workflow workflow) { } if (!isToolTask(task, tagging)) continue; - // A tool task that has neither finished nor produced anything is a - // call the agent has not made yet — a HUMAN tool still waiting on - // its assignee, say. Reporting it would claim a call that has not - // happened. + // Unfinished and empty means the call hasn't happened yet, e.g. a + // HUMAN tool still awaiting its assignee. boolean produced = outputData != null && !outputData.isEmpty(); if (!produced && (task.getStatus() == null || !task.getStatus().isTerminal())) continue; String name = resolveToolName(task); Map args = toolArgs(task.getInputData()); - // A tool whose output isn't wrapped in "result" — HTTP, MCP — reports - // the whole output map, as the server's own event listener does. Keyed - // on the key being there, so a tool that answers null keeps its null. + // HTTP and MCP don't wrap their output in "result". Test for the key + // so a tool that returned null keeps its null. Object result = null; if (produced) { result = outputData.containsKey("result") ? outputData.get("result") : outputData; @@ -485,33 +460,29 @@ private static TaskExtract extractFromTasks(Workflow workflow) { } /** - * Whether a task is one of the agent's tool invocations, as opposed to the - * LLM call, a control-flow task, a guardrail or the approval gate. + * Whether a task is a tool invocation, as opposed to the LLM call, control + * flow, a guardrail or the approval gate. + * + *

Never keyed on the reference name, which carries the provider's tool-call id. */ private static boolean isToolTask(Task task, ToolTagging tagging) { - // Framework passthrough wrappers restate a tool task that is already - // in the list on its own. + // Framework wrappers restate a tool task already in the list. String refName = task.getReferenceTaskName(); if (refName != null && refName.startsWith("_fw_")) return false; Map inputData = task.getInputData(); if (inputData != null && inputData.get(TOOL_NAME_KEY) != null) return true; - // Where the server tags at all it tags every tool kind, so an untagged - // task is not a tool call and guessing past that only invents calls. + // Tagged every kind or none, so untagged means not a tool. if (tagging.tagged) return false; - // Untagged server: the tool calls are the tasks it forked dynamically. + // Older server: fall back to what it forked dynamically. if (refName == null || !tagging.dynamicallyForked.contains(refName)) return false; return isToolTaskType(task); } /** - * Whether a task's type is one an agent tool compiles to. - * - *

{@link #TOOL_TASK_TYPES} alone cannot answer this for a worker tool, - * because by the time the task has run Conductor has rewritten its type from - * {@code SIMPLE} to the task's own name. A task whose type and task-def name - * agree is that case. + * Whether a task's type is one a tool compiles to. An executed worker tool + * reports its own name as its type, so no type list alone can match it. */ private static boolean isToolTaskType(Task task) { String taskType = task.getTaskType(); @@ -520,34 +491,19 @@ private static boolean isToolTaskType(Task task) { } /** - * How one workflow's tool tasks can be told apart, decided once for the - * whole task list. - * - *

A server that tags tool tasks with {@link #TOOL_NAME_KEY} is - * authoritative, and that is the only signal worth having: it names the tool - * as well as identifying it. The tag is set on every tool kind, after the - * per-kind branch that rewrites the task's input, so on such a server tagged - * and tool are the same set. - * - *

Servers predating the tag set it on nothing, and there the fallback is - * the dynamic fork. An agent dispatches its tool calls through - * {@code FORK_JOIN_DYNAMIC}, and Conductor records the reference names it - * forked on the fork task's own input, so a tool task is one the server - * itself reports as dynamically created. Everything the agent compiles for - * its own use is static by comparison, including the guardrail workers, the - * approval gate and a handoff's sub-workflow, which share their task types - * with real tools and would otherwise be counted as tool calls. + * How to tell a workflow's tool tasks apart, decided once for the task list. * - *

The fallback is a best effort, not a second authority. An agent can - * dynamically fan out for reasons of its own, and on an untagged server such - * a task is indistinguishable from a tool call; the tag is what removes the - * ambiguity, which is why it is preferred whenever the server sets it. + *

{@link #TOOL_NAME_KEY} is authoritative where the server sets it. Where it + * doesn't, fall back to the dynamic fork: tool calls are forked, while the + * guardrails, approval gate and handoff sub-workflow are static and would + * otherwise be counted as tool calls. Best-effort only, since an agent can also + * fan out for reasons of its own. */ private static final class ToolTagging { - /** Whether the server tagged any task at all with {@link #TOOL_NAME_KEY}. */ + /** True when the server tagged any task. */ final boolean tagged; - /** Reference names the server reports having forked dynamically. */ + /** Reference names the server reports having forked. */ final Set dynamicallyForked; private ToolTagging(boolean tagged, Set dynamicallyForked) { @@ -574,11 +530,9 @@ static ToolTagging of(List tasks) { } /** - * The tool's own name, which the server puts in the task's input. Never the - * task type: Conductor overwrites an executed SIMPLE task's type with the - * task's own name, so that reads correctly for a worker tool and reports - * every other kind under its system task type — an HTTP tool as - * {@code "HTTP"}, an MCP tool as {@code "CALL_MCP_TOOL"}. + * The tool's own name, from the task's input. Never the task type: Conductor + * rewrites an executed SIMPLE task's type to the task name, which is right for + * a worker and names every other kind after its system task type. */ private static String resolveToolName(Task task) { Map inputData = task.getInputData(); @@ -588,10 +542,8 @@ private static String resolveToolName(Task task) { Object method = inputData.get(TOOL_METHOD_KEY); if (method != null && !method.toString().isEmpty()) return method.toString(); } - // Last resort, and only reachable when the server set one of the keys - // above to an empty string: getTaskDefName() itself falls back to the - // task type, so it is right for a worker tool and no better than the - // old behaviour for anything else. + // Only reached if the server set one of the keys above to an empty string. + // getTaskDefName() itself falls back to the task type. String taskDefName = task.getTaskDefName(); return taskDefName != null && !taskDefName.isEmpty() ? taskDefName : null; } @@ -615,10 +567,8 @@ private static Map toolArgs(Map inputData) { } /** - * The event the stream would have ended on. Both non-streaming paths append - * one so that an events list is never empty for a run that happened — - * without it, "no events were collected" and "no events occurred" are the - * same empty list. + * The event the stream would have ended on. Appended so an empty event list + * means "nothing happened" rather than "nothing was collected". */ private static AgentEvent terminalEvent(String executionId, AgentStatus status, Object output, String error) { String id = executionId != null ? executionId : ""; @@ -631,11 +581,10 @@ private static AgentEvent terminalEvent(String executionId, AgentStatus status, /** * Build an {@link AgentResult} from a terminal {@link Workflow}. * - *

Shared workflow → {@link AgentResult} extraction used by callers that - * already hold a completed {@link Workflow}. Maps the workflow status to an {@link AgentStatus}, - * normalizes the output map, surfaces {@code reasonForIncompletion} as the - * error for non-completed runs, and reuses {@link #extractFromTasks} for the - * token-usage, tool-call and event aggregation. + *

For callers that already hold a completed {@link Workflow}. Maps the status + * to an {@link AgentStatus}, normalizes the output map, surfaces + * {@code reasonForIncompletion} as the error, and reuses + * {@link #extractFromTasks} for the aggregation. * * @param workflow a finished (or at least populated) workflow; may be null * @return the equivalent {@link AgentResult} diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java index 1a4101973..94056d194 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentResult.java @@ -67,15 +67,12 @@ public List> getToolCalls() { } /** - * The events of the run, oldest first. + * The events of the run, oldest first, ending in {@code done} or {@code error}. * - *

Streaming runs report what the server emitted, which ends in a - * {@code done} or {@code error} event unless the stream was cut short. - * Polled runs reconstruct a {@code tool_call}/{@code tool_result} pair per - * tool task from the workflow record and always close with a terminal - * event — so both paths name the same call the same way, but only the - * streamed list carries the incremental {@code thinking} and - * {@code message} events, which the workflow record does not keep. + *

Streaming runs report what the server emitted. Polled runs reconstruct a + * {@code tool_call}/{@code tool_result} pair per tool task, without the + * incremental {@code thinking} and {@code message} events that the workflow + * record does not keep. */ public List getEvents() { return events; diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java index a3dd7cb11..53bdc95a7 100644 --- a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java @@ -27,15 +27,12 @@ import static org.junit.jupiter.api.Assertions.*; /** - * Tool-call and event extraction on the non-streaming path, exercised through - * {@link AgentHandle#fromWorkflow(Workflow)} — the seam {@code waitForResult()} - * shares with it. + * Tool-call and event extraction on the non-streaming path, through + * {@link AgentHandle#fromWorkflow(Workflow)} — the seam {@code waitForResult()} shares. * - *

The fixtures reproduce what Conductor actually stores: an executed SIMPLE - * task carries the task's own name as its {@code taskType}, every other tool - * kind carries its system task type, the server tags every kind's input with - * {@code _agent_tool_name}, and the reference name is seeded from the LLM - * provider's tool-call id. + *

Fixtures reproduce what Conductor stores: an executed SIMPLE task carries the + * task's own name as its {@code taskType}, other kinds carry their system task type, + * and the reference name is seeded from the provider's tool-call id. */ class AgentHandleToolExtractionTest { @@ -49,11 +46,7 @@ private static Task task(String taskType, String refName, Map in return t; } - /** - * A worker tool as the server compiles it: the LLM's arguments sit at the top - * level beside the runtime keys, and Conductor has rewritten {@code taskType} - * to the task's own name by the time the task is executed. - */ + /** A worker tool as the server compiles it, with {@code taskType} already rewritten. */ private static Task workerToolTask(String refName, String toolName, Map args) { Map input = new LinkedHashMap<>(); input.put("_agent_tool_name", toolName); @@ -64,10 +57,8 @@ private static Task workerToolTask(String refName, String toolName, Map compiledInput) { @@ -101,9 +92,8 @@ private static List names(AgentResult result) { } /** - * Every tool kind is reported under the tool's own name, not the system task - * type it compiles to. COUNTERFACTUAL (pre-fix): the names came from - * {@code getTaskType()}, so this read + * Every kind reports the tool's own name, not the system task type. + * COUNTERFACTUAL (pre-fix): names came from {@code getTaskType()}, reading * {@code [get_weather, HTTP, CALL_MCP_TOOL, SUB_WORKFLOW, HUMAN, GENERATE_IMAGE]}. */ @Test @@ -127,10 +117,9 @@ void namesEveryToolKindAfterTheToolNotItsTaskType() { } /** - * Detection does not key on the provider's tool-call id format. COUNTERFACTUAL - * (pre-fix): selection was {@code refName.startsWith("call_")}, so an - * Anthropic-backed run — whose ids start {@code toolu_} — reported no tool - * calls at all. + * Detection ignores the provider's tool-call id format. COUNTERFACTUAL (pre-fix): + * selection was {@code refName.startsWith("call_")}, so an Anthropic-backed run + * reported no tool calls at all. */ @Test void detectsToolCallsWhateverTheProviderIdFormat() { @@ -262,9 +251,8 @@ void aggregatesTokenUsage() { } /** - * COUNTERFACTUAL (pre-fix): {@code events} was hard-coded to {@code null} on - * this path and normalized to an empty list, so a run that called three tools - * was indistinguishable from one that did nothing. + * COUNTERFACTUAL (pre-fix): {@code events} was {@code null} here and normalized to + * an empty list, so a run that called three tools looked like one that did nothing. */ @Test void populatesEventsOnTheNonStreamingPath() { @@ -308,12 +296,8 @@ void alwaysRecordsATerminalEvent() { } // ── Servers that set no _agent_tool_name ───────────────────────────────── - // - // The tag is set on every tool kind or on none. On a server predating it, - // the per-kind compile step has already replaced the task's input, so an - // HTTP tool carries only its http_request and nothing names the tool. What - // still identifies it is the dynamic fork: Conductor records the reference - // names it forked on the fork task's own input. + // The per-kind compile step has replaced the task's input, so nothing names + // the tool. The dynamic fork is what still identifies it. /** A tool task as a pre-tag server dispatches it: no {@code _agent_tool_name}. */ private static Task untaggedWorkerTool(String refName, String toolName, Map args) { @@ -324,11 +308,7 @@ private static Task untaggedWorkerTool(String refName, String toolName, Map compiledInput) { Task t = task(taskType, refName, compiledInput, Map.of("result", toolName + "-output")); @@ -341,11 +321,7 @@ private static Task forkTask(String... forkedRefNames) { return task("FORK", "agent_fork", Map.of("forkedTasks", List.of(forkedRefNames)), Map.of()); } - /** - * A guardrail worker, which is a SIMPLE task like a worker tool and whose - * type is rewritten the same way, but which the agent compiles statically - * and so never appears in the fork list. - */ + /** A guardrail worker: a SIMPLE task like a worker tool, but static, so never forked. */ private static Task staticGuardrailWorker() { return task( "tone_guardrail", @@ -355,11 +331,9 @@ private static Task staticGuardrailWorker() { } /** - * Every tool kind is still found when the server tags none of them, by - * falling back to the reference names it reports having forked. - * COUNTERFACTUAL: without the fork-list fallback, detection rests on the tag - * alone and this reads {@code []} — the tool calls vanish rather than - * arriving under a wrong name. + * Tools are still found when the server tags none of them. + * COUNTERFACTUAL: on the tag alone this reads {@code []}, losing the calls + * entirely rather than naming them wrongly. */ @Test void detectsToolCallsWhenTheServerSetsNoToolNameTag() { @@ -377,10 +351,9 @@ void detectsToolCallsWhenTheServerSetsNoToolNameTag() { } /** - * The agent's own scaffolding is compiled statically, so it is absent from - * the fork list even when it shares a task type with a real tool. A - * guardrail worker is the awkward case: SIMPLE, and its type rewritten to - * its own name exactly as a worker tool's is. + * The agent's own scaffolding is static, so it is absent from the fork list even + * when it shares a task type with a real tool. A guardrail worker is the awkward + * case: SIMPLE, with its type rewritten just as a worker tool's is. */ @Test void ignoresStaticTasksWhenTheServerSetsNoToolNameTag() { @@ -400,10 +373,8 @@ void ignoresStaticTasksWhenTheServerSetsNoToolNameTag() { } /** - * Where the server tags, the tag is the whole answer: an untagged task is - * not a tool call even if it was forked dynamically, because an agent can - * fan out for reasons of its own. The fork list is a fallback, not a second - * source of truth. + * Where the server tags, an untagged task is not a tool call even when forked, + * since an agent can fan out for reasons of its own. */ @Test void prefersTheTagOverTheForkListWhereTheServerTags() { From d4494ddf08df22f9eb4fefbfc309730ec76e9498 Mon Sep 17 00:00:00 2001 From: francisco-orkes Date: Tue, 8 Sep 2026 23:15:06 -0700 Subject: [PATCH 4/4] fix: match the fork list against the loop-suffixed reference name The untagged-server fallback compared a task's reference name to the fork list verbatim, and the two never agree for a real agent. Conductor records forkedTasks when it maps the FORK_JOIN_DYNAMIC, before the enclosing DO_WHILE appends __ to each task it schedules, so the list holds toolu_ab_0 while the tool task is toolu_ab_0__1. Every tool call inside the ReAct loop failed the lookup, which is all of them. Verified against a live server: a dynamic fork inside a DO_WHILE records forkedTasks ["toolu_99_0"] and schedules toolu_99_0__1. The three earlier tests agreed with the defect because their fork lists carried the iteration suffix, which the server never does. - normalize both sides on the __ suffix, matched only at the end of the name rather than TaskUtils' split on the first "__" - fix the fixtures to record fork lists as the server does, and cover a call forked inside the loop - share one internal-key predicate with AgentEvent so both paths strip the same set structurally, rather than by a test asserting they agree - name the tool-call and tool-result event builders - warn rather than debug when the task walk fails, since the empty result it leaves is indistinguishable from a run that used no tools --- .../conductor/ai/model/AgentEvent.java | 14 +++-- .../conductor/ai/model/AgentHandle.java | 54 ++++++++++++------- .../model/AgentHandleToolExtractionTest.java | 29 ++++++++-- 3 files changed, 67 insertions(+), 30 deletions(-) diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java index 4fc417c9a..7bab2717e 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentEvent.java @@ -13,9 +13,7 @@ package org.conductoross.conductor.ai.model; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; -import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -149,19 +147,19 @@ public List getPendingToolCalls() { return calls; } - /** - * Create an AgentEvent from a raw map (as parsed from SSE JSON). - */ /** * Internal keys the server injects, which are not tool arguments. Every - * {@code _}-prefixed key is stripped too, as the polled path does. + * {@code _}-prefixed key is stripped too. */ - private static final Set INTERNAL_KEYS = new HashSet<>(Arrays.asList("method")); + private static final Set INTERNAL_KEYS = + Set.of("method", "evaluatorType", "expression", "ctx", "workerTag", "agentConfig"); - private static boolean isInternalKey(String key) { + /** Shared with {@code AgentHandle} so both paths report one call's arguments identically. */ + static boolean isInternalKey(String key) { return key != null && (key.startsWith("_") || INTERNAL_KEYS.contains(key)); } + /** Create an AgentEvent from a raw map (as parsed from SSE JSON). */ @SuppressWarnings("unchecked") public static AgentEvent fromMap(Map data) { String typeStr = (String) data.get("type"); diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java index 509bf90bd..42b6fa2ef 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/model/AgentHandle.java @@ -13,10 +13,12 @@ package org.conductoross.conductor.ai.model; import java.util.ArrayList; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.regex.Pattern; import org.conductoross.conductor.ai.enums.AgentStatus; import org.conductoross.conductor.ai.enums.EventType; @@ -355,7 +357,8 @@ private AgentResult buildResult(AgentStatusResponse statusResponse, String workf try { extract = extractFromTasks(workflowClient.getWorkflow(executionId, true)); } catch (Exception e) { - logger.debug("Could not extract tokens/toolCalls for {}: {}", executionId, e.getMessage()); + // Nothing distinguishes this from a tool-free run in the result, so say so here. + logger.warn("Could not extract tokens/toolCalls/events for {}: {}", executionId, e.getMessage()); } return toResult(extract, output, executionId, status, error); @@ -404,6 +407,20 @@ private static final class TaskExtract { /** Reference names Conductor forked, recorded on the {@code FORK_JOIN_DYNAMIC} task's input. */ private static final String FORKED_TASKS_KEY = "forkedTasks"; + /** + * The {@code __} suffix Conductor appends to a task's reference name inside a + * {@code DO_WHILE}. The fork list is recorded before that happens, so both sides of the + * comparison in {@link ToolTagging} have to be normalized. Matched only at the end, unlike + * {@code TaskUtils.removeIterationFromTaskRefName}, which splits on the first {@code __} + * and would truncate a reference name that already contains one. + */ + private static final Pattern ITERATION_SUFFIX = Pattern.compile("__\\d+$"); + + /** A reference name without the loop-iteration suffix. */ + private static String withoutIteration(String refName) { + return refName == null ? null : ITERATION_SUFFIX.matcher(refName).replaceFirst(""); + } + /** * Walk a workflow's tasks once and aggregate token usage, tool calls and their * events. Shared by {@link #buildResult} and {@link #fromWorkflow(Workflow)}. @@ -448,10 +465,8 @@ private static TaskExtract extractFromTasks(Workflow workflow) { tc.put("result", result); out.toolCalls.add(tc); - out.events.add( - new AgentEvent(EventType.TOOL_CALL, null, name, args, null, null, executionId, null, null)); - out.events.add( - new AgentEvent(EventType.TOOL_RESULT, null, name, null, result, null, executionId, null, null)); + out.events.add(toolCallEvent(name, args, executionId)); + out.events.add(toolResultEvent(name, result, executionId)); } if (sawTokens) { out.tokenUsage = new TokenUsage(promptT, completionT, totalT); @@ -476,7 +491,7 @@ private static boolean isToolTask(Task task, ToolTagging tagging) { if (tagging.tagged) return false; // Older server: fall back to what it forked dynamically. - if (refName == null || !tagging.dynamicallyForked.contains(refName)) return false; + if (refName == null || !tagging.dynamicallyForked.contains(withoutIteration(refName))) return false; return isToolTaskType(task); } @@ -513,7 +528,7 @@ private ToolTagging(boolean tagged, Set dynamicallyForked) { static ToolTagging of(List tasks) { boolean tagged = false; - Set forked = new java.util.HashSet<>(); + Set forked = new HashSet<>(); for (Task task : tasks) { Map inputData = task.getInputData(); if (inputData == null) continue; @@ -521,7 +536,7 @@ static ToolTagging of(List tasks) { Object names = inputData.get(FORKED_TASKS_KEY); if (names instanceof List) { for (Object name : (List) names) { - if (name != null) forked.add(name.toString()); + if (name != null) forked.add(withoutIteration(name.toString())); } } } @@ -548,24 +563,27 @@ private static String resolveToolName(Task task) { return taskDefName != null && !taskDefName.isEmpty() ? taskDefName : null; } - /** A tool task's input with the server's internal runtime keys stripped. */ + /** + * A tool task's input with the server's internal runtime keys stripped, using the same + * predicate as the streaming path so both report one call's arguments identically. + */ private static Map toolArgs(Map inputData) { if (inputData == null) return null; Map cleaned = new LinkedHashMap<>(); for (Map.Entry e : inputData.entrySet()) { - String k = e.getKey(); - if (k.startsWith("_") - || TOOL_METHOD_KEY.equals(k) - || "evaluatorType".equals(k) - || "expression".equals(k) - || "ctx".equals(k) - || "workerTag".equals(k) - || "agentConfig".equals(k)) continue; - cleaned.put(k, e.getValue()); + if (!AgentEvent.isInternalKey(e.getKey())) cleaned.put(e.getKey(), e.getValue()); } return cleaned; } + private static AgentEvent toolCallEvent(String name, Map args, String executionId) { + return new AgentEvent(EventType.TOOL_CALL, null, name, args, null, null, executionId, null, null); + } + + private static AgentEvent toolResultEvent(String name, Object result, String executionId) { + return new AgentEvent(EventType.TOOL_RESULT, null, name, null, result, null, executionId, null, null); + } + /** * The event the stream would have ended on. Appended so an empty event list * means "nothing happened" rather than "nothing was collected". diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java index 53bdc95a7..74654e3b5 100644 --- a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java @@ -316,7 +316,12 @@ private static Task untaggedSystemTool( return t; } - /** The synthetic FORK task Conductor writes when it fans out dynamically. */ + /** + * The synthetic FORK task Conductor writes when it fans out dynamically. The names it + * records carry no {@code __} suffix: the fork is mapped before the + * {@code DO_WHILE} appends one to the tasks it schedules, so the two differ by that + * suffix for every tool call in the loop. + */ private static Task forkTask(String... forkedRefNames) { return task("FORK", "agent_fork", Map.of("forkedTasks", List.of(forkedRefNames)), Map.of()); } @@ -339,7 +344,7 @@ private static Task staticGuardrailWorker() { void detectsToolCallsWhenTheServerSetsNoToolNameTag() { AgentResult result = AgentHandle.fromWorkflow(workflow( llmTask(10, 5), - forkTask("toolu_01_0__1", "toolu_02_0__1"), + forkTask("toolu_01_0", "toolu_02_0"), untaggedWorkerTool("toolu_01_0__1", "get_weather", Map.of("city", "SF")), untaggedSystemTool( "HTTP", @@ -366,7 +371,7 @@ void ignoresStaticTasksWhenTheServerSetsNoToolNameTag() { "agent_transfer_refunds", Map.of("prompt", "refund please", "session_id", "s1"), Map.of("result", "handled")), - forkTask("toolu_01_0__1"), + forkTask("toolu_01_0"), untaggedWorkerTool("toolu_01_0__1", "get_weather", Map.of("city", "SF")))); assertEquals(List.of("get_weather"), names(result)); @@ -380,13 +385,29 @@ void ignoresStaticTasksWhenTheServerSetsNoToolNameTag() { void prefersTheTagOverTheForkListWhereTheServerTags() { AgentResult result = AgentHandle.fromWorkflow(workflow( llmTask(10, 5), - forkTask("call_a_0__1", "fanout_0__1"), + forkTask("call_a_0", "fanout_0"), workerToolTask("call_a_0__1", "get_weather", Map.of("city", "SF")), task("SUB_WORKFLOW", "fanout_0__1", Map.of("prompt", "sub-task"), Map.of("result", "done")))); assertEquals(List.of("get_weather"), names(result)); } + /** + * A tool call inside the agent's loop is still found, though the fork list records it + * without the iteration suffix Conductor gives the scheduled task. COUNTERFACTUAL: + * comparing the reference names verbatim reads {@code []} for every real agent, since + * the tool fork always sits inside the {@code DO_WHILE}. + */ + @Test + void detectsToolCallsForkedInsideTheAgentLoop() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + forkTask("toolu_ab_0"), + untaggedWorkerTool("toolu_ab_0__3", "get_weather", Map.of("city", "SF")))); + + assertEquals(List.of("get_weather"), names(result)); + } + /** Both paths report the same call under the same name. */ @Test void agreesWithTheStreamingPathOnToolNames() {