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..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; @@ -150,12 +148,18 @@ public List getPendingToolCalls() { } /** - * 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. */ - /** 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")); + Set.of("method", "evaluatorType", "expression", "ctx", "workerTag", "agentConfig"); + /** 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"); @@ -182,7 +186,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..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,11 +13,15 @@ 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; import org.conductoross.conductor.ai.exceptions.WorkerStallError; import org.conductoross.conductor.ai.internal.ServerLivenessMonitor; import org.slf4j.Logger; @@ -346,50 +350,93 @@ 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. - // 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<>(); + // 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 { - 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()); + // 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 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. */ + 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<>(); } /** - * 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. + * 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", + "HTTP", + "CALL_MCP_TOOL", + "SUB_WORKFLOW", + "HUMAN", + "GENERATE_IMAGE", + "GENERATE_AUDIO", + "GENERATE_VIDEO", + "GENERATE_PDF", + "LLM_INDEX_TEXT", + "LLM_SEARCH_INDEX", + "PULL_WORKFLOW_MESSAGES"); + + /** 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"; + + /** 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)}. */ 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) { - 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 +444,29 @@ 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, tagging)) continue; + // 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()); + // 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; } + + 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(toolCallEvent(name, args, executionId)); + out.events.add(toolResultEvent(name, result, executionId)); } if (sawTokens) { out.tokenUsage = new TokenUsage(promptT, completionT, totalT); @@ -430,14 +474,135 @@ private static TaskExtract extractFromTasks(Workflow workflow) { return out; } + /** + * 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 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; + // Tagged every kind or none, so untagged means not a tool. + if (tagging.tagged) return false; + + // Older server: fall back to what it forked dynamically. + if (refName == null || !tagging.dynamicallyForked.contains(withoutIteration(refName))) return false; + return isToolTaskType(task); + } + + /** + * 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(); + if (taskType == null) return false; + return TOOL_TASK_TYPES.contains(taskType) || taskType.equals(task.getTaskDefName()); + } + + /** + * How to tell a workflow's tool tasks apart, decided once for the task list. + * + *

{@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 { + /** True when the server tagged any task. */ + final boolean tagged; + + /** Reference names the server reports having forked. */ + 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 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(withoutIteration(name.toString())); + } + } + } + return new ToolTagging(tagged, forked); + } + } + + /** + * 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(); + 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(); + } + // 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; + } + + /** + * 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()) { + 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". + */ + 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}. * - *

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 and tool-call 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} @@ -465,8 +630,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..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 @@ -66,6 +66,14 @@ public List> getToolCalls() { return toolCalls; } + /** + * The events of the run, oldest first, ending in {@code done} or {@code error}. + * + *

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 new file mode 100644 index 000000000..74654e3b5 --- /dev/null +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/model/AgentHandleToolExtractionTest.java @@ -0,0 +1,433 @@ +/* + * 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, through + * {@link AgentHandle#fromWorkflow(Workflow)} — the seam {@code waitForResult()} shares. + * + *

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 { + + 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, 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); + 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 arguments are folded into what + * that task type takes, so 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 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 + 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 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() { + 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 {@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() { + 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()); + } + + // ── Servers that set no _agent_tool_name ───────────────────────────────── + // 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) { + 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 system-task tool from a pre-tag server: only the task's own name identifies it. */ + 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. 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()); + } + + /** A guardrail worker: a SIMPLE task like a worker tool, but static, so never forked. */ + private static Task staticGuardrailWorker() { + return task( + "tone_guardrail", + "agent_ext_guardrail_tone", + Map.of("content", "hello", "input", "hello", "iteration", 1), + Map.of("result", "pass")); + } + + /** + * 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() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + forkTask("toolu_01_0", "toolu_02_0"), + 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 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() { + 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"), + untaggedWorkerTool("toolu_01_0__1", "get_weather", Map.of("city", "SF")))); + + assertEquals(List.of("get_weather"), names(result)); + } + + /** + * 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() { + AgentResult result = AgentHandle.fromWorkflow(workflow( + llmTask(10, 5), + 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() { + 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` |