diff --git a/docs/api-reference/veryfront/observability.md b/docs/api-reference/veryfront/observability.md index 71e89438e9..57844a835d 100644 --- a/docs/api-reference/veryfront/observability.md +++ b/docs/api-reference/veryfront/observability.md @@ -197,6 +197,7 @@ import { | `initializeOTLPWithApis` | Initialize OTLP tracing with explicit API adapters. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | | `injectContext` | Context for inject. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | | `isOTLPEnabled` | Check whether OTLP export is enabled. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | +| `markSpanFailed` | Marks a span as failed with a stable error code. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | | `setActiveSpanAttributes` | Sets active span attributes. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | | `setActiveSpanErrorStatus` | Marks the active span as failed. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | | `setSpanAttributes` | Sets span attributes. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/observability/tracing/otlp-setup.ts) | diff --git a/src/agent/hosted/agent-run-lifecycle.test.ts b/src/agent/hosted/agent-run-lifecycle.test.ts index 0602ad2923..c90fe576d3 100644 --- a/src/agent/hosted/agent-run-lifecycle.test.ts +++ b/src/agent/hosted/agent-run-lifecycle.test.ts @@ -20,6 +20,7 @@ class RecordingSpan implements HostedAgentRunSpan { attributes: Record = {}; finished = 0; withContextCalls = 0; + failedWith: string[] = []; setAttributes(attributes: Record): void { this.attributes = { ...this.attributes, ...attributes }; @@ -29,6 +30,10 @@ class RecordingSpan implements HostedAgentRunSpan { this.finished += 1; } + markFailed(errorCode: string): void { + this.failedWith.push(errorCode); + } + withContext(fn: () => T): T { this.withContextCalls += 1; return fn(); @@ -146,6 +151,7 @@ describe("hosted-agent-run-lifecycle", () => { controller.finalize({ status: "failed", terminalErrorCode: "LATE" }); assertEquals(span.finished, 1); + assertEquals(span.failedWith, []); assertEquals(span.attributes["message.id"], "message-2"); assertEquals(span.attributes["agent.run.final_status"], "completed"); assertEquals(span.attributes["gen_ai.provider.name"], "anthropic"); @@ -160,6 +166,61 @@ describe("hosted-agent-run-lifecycle", () => { assertEquals(span.attributes["gen_ai.usage.reasoning.output_tokens"], 1); }); + it("marks a failed hosted run span failed with its terminal error code before finishing", () => { + const span = new RecordingSpan(); + const finishedAfterFailure: number[] = []; + span.markFailed = (errorCode) => { + span.failedWith.push(errorCode); + finishedAfterFailure.push(span.finished); + }; + const controller = createHostedAgentRunSpanController({ + tracer: { startSpan: () => span }, + operationName: "invoke_agent", + projectId: "project-1", + userId: "user-1", + agentId: "agent-1", + }); + + controller.finalize({ status: "failed", terminalErrorCode: "insufficient-credits" }); + + assertEquals(span.failedWith, ["insufficient-credits"]); + assertEquals(finishedAfterFailure, [0]); + assertEquals(span.finished, 1); + assertEquals(span.attributes["error.type"], "insufficient-credits"); + }); + + it("marks a failed hosted run span without a stable terminal error code as STREAM_ERROR", () => { + const span = new RecordingSpan(); + const controller = createHostedAgentRunSpanController({ + tracer: { startSpan: () => span }, + operationName: "chat", + projectId: "project-1", + userId: "user-1", + agentId: "agent-1", + }); + + controller.finalize({ status: "failed", terminalErrorCode: "/srv/app/secret.env" }); + + assertEquals(span.failedWith, ["STREAM_ERROR"]); + assertEquals(span.attributes["error.type"], "STREAM_ERROR"); + }); + + it("leaves a cancelled hosted run span unmarked", () => { + const span = new RecordingSpan(); + const controller = createHostedAgentRunSpanController({ + tracer: { startSpan: () => span }, + operationName: "chat", + projectId: "project-1", + userId: "user-1", + agentId: "agent-1", + }); + + controller.finalize({ status: "cancelled" }); + + assertEquals(span.failedWith, []); + assertEquals(span.finished, 1); + }); + // veryfront/veryfront-issue-inbox#1500: the hosted run span reported token counts // and no spend, on every status. // diff --git a/src/agent/hosted/agent-run-lifecycle.ts b/src/agent/hosted/agent-run-lifecycle.ts index f2105b5027..439c16d5ac 100644 --- a/src/agent/hosted/agent-run-lifecycle.ts +++ b/src/agent/hosted/agent-run-lifecycle.ts @@ -11,11 +11,13 @@ import { type AgentTraceUsage, buildAgentRunTraceAttributes, buildFinalizedAgentRunTraceAttributes, + resolveAgentRunErrorType, } from "./trace-attributes.ts"; /** Public API contract for hosted agent run span. */ export interface HostedAgentRunSpan { setAttributes: (attributes: AgentTraceAttributes) => void; + markFailed?: (errorCode: string) => void; finish: () => void; withContext: (fn: () => T) => T; } @@ -126,6 +128,9 @@ export function createHostedAgentRunSpanController( finalized = true; span.setAttributes(buildFinalizedAgentRunTraceAttributes(finalState)); + if (finalState.status === "failed") { + span.markFailed?.(resolveAgentRunErrorType(finalState.terminalErrorCode)); + } span.finish(); }, }; diff --git a/src/agent/hosted/trace-attributes.test.ts b/src/agent/hosted/trace-attributes.test.ts index 06ddf506e6..bed6ec35f3 100644 --- a/src/agent/hosted/trace-attributes.test.ts +++ b/src/agent/hosted/trace-attributes.test.ts @@ -234,4 +234,21 @@ describe("agent/agent-trace-attributes", () => { }, ); }); + + it("keeps only classification-shaped terminal error codes as the failed run error type", () => { + assertEquals( + buildFinalizedAgentRunTraceAttributes({ + status: "failed", + terminalErrorCode: "insufficient-credits", + })["error.type"], + "insufficient-credits", + ); + assertEquals( + buildFinalizedAgentRunTraceAttributes({ + status: "failed", + terminalErrorCode: "postgres://app:secret@db.internal/prod", + })["error.type"], + "STREAM_ERROR", + ); + }); }); diff --git a/src/agent/hosted/trace-attributes.ts b/src/agent/hosted/trace-attributes.ts index bb05e17f36..6e90082fdb 100644 --- a/src/agent/hosted/trace-attributes.ts +++ b/src/agent/hosted/trace-attributes.ts @@ -287,6 +287,21 @@ export function buildInvokeAgentTraceAttributes(input: { }); } +const STABLE_RUN_ERROR_CODE_PATTERN = /^[A-Za-z][A-Za-z0-9_.-]{0,63}$/; + +/** + * Keeps a run error code only when it has the shape of a classification. A runtime + * can put any string there, and it becomes a span status message and a log field. + */ +export function toStableRunErrorCode(code: unknown): string | undefined { + return typeof code === "string" && STABLE_RUN_ERROR_CODE_PATTERN.test(code) ? code : undefined; +} + +/** Stable `error.type` of a failed agent run, also used as its span status message. */ +export function resolveAgentRunErrorType(terminalErrorCode?: string | null): string { + return toStableRunErrorCode(terminalErrorCode) ?? "STREAM_ERROR"; +} + /** Builds finalized agent run trace attributes. */ export function buildFinalizedAgentRunTraceAttributes(input: { status: "completed" | "failed" | "cancelled"; @@ -309,7 +324,7 @@ export function buildFinalizedAgentRunTraceAttributes(input: { ...(finishReason ? { "gen_ai.response.finish_reasons": [finishReason] } : {}), ...(input.status === "failed" ? { - "error.type": input.terminalErrorCode ?? "STREAM_ERROR", + "error.type": resolveAgentRunErrorType(input.terminalErrorCode), "error.message": input.terminalErrorMessage, } : {}), diff --git a/src/agent/runtime/agent-delegation.test.ts b/src/agent/runtime/agent-delegation.test.ts index 6ed1b9caf2..35ee0936d9 100644 --- a/src/agent/runtime/agent-delegation.test.ts +++ b/src/agent/runtime/agent-delegation.test.ts @@ -5,6 +5,7 @@ import { AGENT_DELEGATE_TOOL_PREFIX, buildAgentDelegateTools, createInvokeAgentTool, + isFrameworkChildRunTool, isProviderSafeDelegateId, } from "./agent-delegation.ts"; import type { Agent } from "../types.ts"; @@ -352,3 +353,18 @@ it("delegate agent execution preserves an explicit process-boundary restriction" assertEquals(observedPolicy, policy); assertEquals(observedDuringStreamConsumption, policy); }); + +it("isFrameworkChildRunTool recognizes framework invoke_agent and delegate tools only", () => { + const delegateTools = buildAgentDelegateTools({ + delegates: ["researcher"], + resolveAgent: () => undefined, + }); + + assertEquals(isFrameworkChildRunTool(createInvokeAgentTool()), true); + assertEquals(isFrameworkChildRunTool(delegateTools.agent_researcher), true); + assertEquals( + isFrameworkChildRunTool({ id: "agent_researcher", type: "function", execute: () => ({}) }), + false, + ); + assertEquals(isFrameworkChildRunTool(true), false); +}); diff --git a/src/agent/runtime/agent-delegation.ts b/src/agent/runtime/agent-delegation.ts index 67754ba75e..8413566359 100644 --- a/src/agent/runtime/agent-delegation.ts +++ b/src/agent/runtime/agent-delegation.ts @@ -13,12 +13,19 @@ export const INVOKE_AGENT_TOOL_ID = "invoke_agent"; const applyIntrinsic = Reflect.apply; const stringTrim = String.prototype.trim; const frameworkInvokeAgentTools = new WeakSet(); +const frameworkDelegateTools = new WeakSet(); /** Whether a tool is the framework-created invoke_agent from {@link createInvokeAgentTool}. */ export function isFrameworkInvokeAgentTool(value: unknown): boolean { return value !== null && typeof value === "object" && frameworkInvokeAgentTools.has(value); } +/** Whether a tool is a framework tool whose calls run a child agent (invoke_agent or `agent_{id}`). */ +export function isFrameworkChildRunTool(value: unknown): boolean { + return value !== null && typeof value === "object" && + (frameworkInvokeAgentTools.has(value) || frameworkDelegateTools.has(value)); +} + const getInvokeAgentInputSchema = defineSchema((v) => v.object({ agent_id: v.string() @@ -122,7 +129,7 @@ function createLazyDelegateTool( resolveAgent: DelegateAgentResolver, executeDelegate?: DelegateAgentExecutor, ): Tool { - return markRuntimeLocalTool({ + const tool = markRuntimeLocalTool({ id: `${AGENT_DELEGATE_TOOL_PREFIX}${delegateId}`, type: "function", description: `Delegate a self-contained subtask to the "${delegateId}" specialist agent, ` + @@ -150,6 +157,8 @@ function createLazyDelegateTool( return agentAsTool(target, `Delegate to ${delegateId}`).execute(input, context); }, }); + frameworkDelegateTools.add(tool); + return tool; } /** diff --git a/src/internal-agents/run-stream.test.ts b/src/internal-agents/run-stream.test.ts index 75fdc6261a..a3981aa79a 100644 --- a/src/internal-agents/run-stream.test.ts +++ b/src/internal-agents/run-stream.test.ts @@ -12,6 +12,7 @@ import { DEFAULT_RUNTIME_AGENT_CONTEXT_MARKER, } from "#veryfront/agent"; import { executeConfiguredTool, getAvailableTools } from "#veryfront/agent/runtime/tool-helpers.ts"; +import { buildAgentDelegateTools } from "#veryfront/agent/runtime/agent-delegation.ts"; import { flattenSystemInstructions } from "#veryfront/agent/runtime/tool-inventory.ts"; import { resolveAgentSystem } from "#veryfront/agent/runtime/effective-agent-system.ts"; import { createRuntimeAgentFromMarkdownDefinition } from "#veryfront/agent/runtime/agent-markdown-adapter.ts"; @@ -21,6 +22,7 @@ import { setGlobalTracerProvider, type Span, type SpanContext, + SpanStatusCode, type Tracer, } from "#veryfront/observability/tracing/api-shim.ts"; import type { @@ -4022,6 +4024,392 @@ describe("internal-agents/run-stream", () => { assertEquals(runSpan?.events.some((event) => event.name === "agent.run.completed"), false); }); + it("marks a run that ends on a terminal runtime error as an ERROR span with its error code", async () => { + const spans = installRecordingTracer(); + const logs = captureConsoleJsonLogs(); + const sessionManager = new AgentRunSessionManager(); + const agent = { + id: "credit-limited-agent", + config: { + id: "credit-limited-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_terminal_error_code", + parentRunId: "run_parent_of_terminal_error", + messages: [], + tools: [], + context: [], + } as Parameters[0]; + + try { + await withJsonDebugLogFormat(async () => { + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"error","code":"insufficient-credits","error":"AI credit limit exceeded"}\n\n', + ), + ); + controller.close(); + }, + }), + }), + }); + await response.text(); + }); + } finally { + logs.restore(); + } + + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.status?.code, SpanStatusCode.ERROR); + assertEquals(runSpan?.status?.message, "insufficient-credits"); + assertEquals(runSpan?.attributes["agent.run.final_status"], "failed"); + assertEquals(runSpan?.attributes["error.type"], "insufficient-credits"); + assertEquals(runSpan?.attributes["error.message"], undefined); + + const finalizedEntry = logs.getEntries().find((entry) => + entry.message === "Internal agent runtime stream finalized" + ); + assertEquals(finalizedEntry?.level, "warn"); + assertEquals(finalizedEntry?.context?.status, "failed"); + assertEquals(finalizedEntry?.context?.parentRunId, "run_parent_of_terminal_error"); + assertEquals(finalizedEntry?.context?.errorCode, "insufficient-credits"); + assertEquals(finalizedEntry?.context?.error, undefined); + assertEquals(JSON.stringify(finalizedEntry).includes("AI credit limit exceeded"), false); + }); + + it("replaces a terminal error code that is not a stable classification", async () => { + const spans = installRecordingTracer(); + const logs = captureConsoleJsonLogs(); + const sessionManager = new AgentRunSessionManager(); + const agent = { + id: "unsafe-code-agent", + config: { + id: "unsafe-code-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_unsafe_terminal_code", + messages: [], + tools: [], + context: [], + } as Parameters[0]; + + try { + await withJsonDebugLogFormat(async () => { + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"error","code":"postgres://app:secret@db.internal/prod","error":"boom"}\n\n', + ), + ); + controller.close(); + }, + }), + }), + }); + await response.text(); + }); + } finally { + logs.restore(); + } + + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.status?.message, "AgentRunTerminalError"); + assertEquals(runSpan?.attributes["error.type"], "AgentRunTerminalError"); + const finalizedEntry = logs.getEntries().find((entry) => + entry.message === "Internal agent runtime stream finalized" + ); + assertEquals(finalizedEntry?.context?.errorCode, "AgentRunTerminalError"); + assertEquals(JSON.stringify(finalizedEntry).includes("secret"), false); + }); + + it("marks a run whose runtime stream throws as an ERROR span with the run error code", async () => { + const spans = installRecordingTracer(); + const sessionManager = new AgentRunSessionManager(); + const agent = { + id: "throwing-agent", + config: { + id: "throwing-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_stream_throws", + messages: [], + tools: [], + context: [], + } as Parameters[0]; + + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + pull(controller) { + controller.error(new Error("socket hang up")); + }, + }), + }), + }); + const body = await response.text(); + + assertStringIncludes(body, "event: RunError"); + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.attributes["agent.run.final_status"], "failed"); + assertEquals(runSpan?.status?.code, SpanStatusCode.ERROR); + assertEquals(runSpan?.status?.message, "RUNTIME_ERROR"); + assertEquals(runSpan?.attributes["error.type"], "RUNTIME_ERROR"); + assertEquals(runSpan?.attributes["error.cause.type"], "Error"); + }); + + it("keeps a completed run that recovered from a tool error out of ERROR status", async () => { + const spans = installRecordingTracer(); + const sessionManager = new AgentRunSessionManager(); + const agent = { + id: "recovering-agent", + config: { + id: "recovering-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_recovered_tool_error", + messages: [], + tools: [], + context: [], + } as Parameters[0]; + + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + [ + 'data: {"type":"message-start","messageId":"assistant-1"}', + 'data: {"type":"tool-input-available","toolCallId":"tool-1","toolName":"lookup","input":{}}', + 'data: {"type":"tool-output-error","toolCallId":"tool-1","errorText":"lookup timed out"}', + 'data: {"type":"text-start","id":"text-1"}', + 'data: {"type":"text-delta","id":"text-1","delta":"done anyway"}', + 'data: {"type":"text-end","id":"text-1"}', + "", + "", + ].join("\n\n"), + ), + ); + controller.close(); + }, + }), + }), + }); + await response.text(); + + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.attributes["agent.run.final_status"], "completed"); + assertEquals(runSpan?.attributes["agent.run.tool_error_count"], 1); + assertEquals(runSpan?.status, undefined); + }); + + it("counts failed child agent runs apart from other tool errors on a recovered run", async () => { + const spans = installRecordingTracer(); + const logs = captureConsoleJsonLogs(); + const sessionManager = new AgentRunSessionManager(); + const agent = { + id: "delegating-agent", + config: { + id: "delegating-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + tools: buildAgentDelegateTools({ + delegates: ["researcher"], + resolveAgent: () => undefined, + }), + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_recovered_child_failure", + messages: [], + tools: [ + { name: "invoke_agent", parameters: { type: "object", properties: {} } }, + { name: "veryfront__invoke_agent", parameters: { type: "object", properties: {} } }, + ], + context: [], + } as Parameters[0]; + + try { + await withJsonDebugLogFormat(async () => { + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + [ + 'data: {"type":"message-start","messageId":"assistant-1"}', + 'data: {"type":"tool-input-start","toolCallId":"child-1","toolName":"invoke_agent"}', + 'data: {"type":"tool-input-available","toolCallId":"child-1","toolName":"invoke_agent","input":{}}', + 'data: {"type":"tool-output-error","toolCallId":"child-1","errorText":"child run failed"}', + 'data: {"type":"tool-input-start","toolCallId":"child-2","toolName":"veryfront__invoke_agent"}', + 'data: {"type":"tool-input-available","toolCallId":"child-2","toolName":"veryfront__invoke_agent","input":{}}', + 'data: {"type":"tool-output-available","toolCallId":"child-2","output":"ok"}', + 'data: {"type":"tool-input-start","toolCallId":"delegate-1","toolName":"agent_researcher"}', + 'data: {"type":"tool-input-available","toolCallId":"delegate-1","toolName":"agent_researcher","input":{}}', + 'data: {"type":"tool-output-error","toolCallId":"delegate-1","errorText":"delegate failed"}', + 'data: {"type":"tool-input-start","toolCallId":"fetch-1","toolName":"web_fetch"}', + 'data: {"type":"tool-input-available","toolCallId":"fetch-1","toolName":"web_fetch","input":{}}', + 'data: {"type":"tool-output-error","toolCallId":"fetch-1","errorText":"404"}', + 'data: {"type":"text-start","id":"text-1"}', + 'data: {"type":"text-delta","id":"text-1","delta":"handled"}', + 'data: {"type":"text-end","id":"text-1"}', + "", + "", + ].join("\n\n"), + ), + ); + controller.close(); + }, + }), + }), + }); + await response.text(); + }); + } finally { + logs.restore(); + } + + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.attributes["agent.run.final_status"], "completed"); + assertEquals(runSpan?.attributes["agent.run.tool_error_count"], 3); + assertEquals(runSpan?.attributes["agent.run.child_run_error_count"], 2); + assertEquals(runSpan?.status, undefined); + + const finalizedEntry = logs.getEntries().find((entry) => + entry.message === "Internal agent runtime stream finalized" + ); + assertEquals(finalizedEntry?.level, "info"); + assertEquals(finalizedEntry?.context?.toolErrorCount, 3); + assertEquals(finalizedEntry?.context?.childRunErrorCount, 2); + }); + + for ( + const { label, tools, registerSameNameTool } of [ + { + label: "an inline custom tool", + tools: { + invoke_agent: { + id: "invoke_agent", + type: "function", + description: "Project tool that happens to share the name", + inputSchema: { type: "object", properties: {} }, + execute: () => ({ ok: true }), + }, + }, + registerSameNameTool: false, + }, + { label: "a registry tool granted by tools: true", tools: true, registerSameNameTool: true }, + ] + ) { + it(`does not count ${label} that shares the invoke_agent name as a child run`, async () => { + const spans = installRecordingTracer(); + const sessionManager = new AgentRunSessionManager(); + if (registerSameNameTool) { + toolRegistryInternal.register("invoke_agent", { + id: "invoke_agent", + type: "function", + description: "Project tool that happens to share the name", + inputSchema: {} as never, + execute: () => ({ ok: true }), + } as unknown as Tool); + } + const agent = { + id: "custom-invoke-agent", + config: { + id: "custom-invoke-agent", + model: "anthropic/claude-opus-4-6", + system: "test", + tools, + }, + } as unknown as Agent; + const input = { + agentId: agent.id, + threadId: crypto.randomUUID(), + runId: "run_custom_invoke_agent_error", + messages: [], + tools: [], + context: [], + } as Parameters[0]; + + try { + const response = await createRuntimeAgentStreamResponse(input, agent, { + sessionManager, + createRuntime: () => ({ + stream: async () => + new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + [ + 'data: {"type":"message-start","messageId":"assistant-1"}', + 'data: {"type":"tool-input-start","toolCallId":"custom-1","toolName":"invoke_agent"}', + 'data: {"type":"tool-input-available","toolCallId":"custom-1","toolName":"invoke_agent","input":{}}', + 'data: {"type":"tool-output-error","toolCallId":"custom-1","errorText":"custom tool failed"}', + 'data: {"type":"text-start","id":"text-1"}', + 'data: {"type":"text-delta","id":"text-1","delta":"handled"}', + 'data: {"type":"text-end","id":"text-1"}', + "", + "", + ].join("\n\n"), + ), + ); + controller.close(); + }, + }), + }), + }); + await response.text(); + } finally { + toolRegistryInternal.clearAll(); + } + + const runSpan = spans.find((span) => span.name === "agent.run"); + assertEquals(runSpan?.attributes["agent.run.final_status"], "completed"); + assertEquals(runSpan?.attributes["agent.run.tool_error_count"], 1); + assertEquals(runSpan?.attributes["agent.run.child_run_error_count"], 0); + }); + } + it("records usage accumulated before a terminal runtime error on the agent.run span", async () => { const spans = installRecordingTracer(); const sessionManager = new AgentRunSessionManager(); @@ -4120,6 +4508,7 @@ describe("internal-agents/run-stream", () => { const runSpan = spans.find((span) => span.name === "agent.run"); assertEquals(runSpan?.attributes["agent.run.final_status"], "cancelled"); + assertEquals(runSpan?.status, undefined); assertEquals(runSpan?.attributes["agent.usage.cost_credits"], 8.5); assertEquals(runSpan?.attributes["gen_ai.usage.total_tokens"], 60); assertEquals(runSpan?.attributes["agent.run.usage_is_floor"], true); diff --git a/src/internal-agents/run-stream.ts b/src/internal-agents/run-stream.ts index bac1fd25ce..2e0494121d 100644 --- a/src/internal-agents/run-stream.ts +++ b/src/internal-agents/run-stream.ts @@ -34,6 +34,7 @@ import { getProviderNativeToolNames } from "#veryfront/agent/runtime/provider-na import { selectProviderCompatibleToolNames } from "#veryfront/agent/runtime/provider-tool-compat.ts"; import { INVOKE_AGENT_TOOL_ID, + isFrameworkChildRunTool, isFrameworkInvokeAgentTool, } from "#veryfront/agent/runtime/agent-delegation.ts"; import { @@ -52,6 +53,7 @@ import { SandboxShellToolsProviderName, } from "#veryfront/extensions/sandbox/index.ts"; import { resolveHostedRuntimeAllowedToolNames } from "#veryfront/agent/hosted/runtime-essential-tools.ts"; +import { toStableRunErrorCode } from "#veryfront/agent/hosted/trace-attributes.ts"; import { createToolsFromHostDefinitions, isToolVisibleTo, @@ -62,6 +64,7 @@ import { import { skillRegistry } from "#veryfront/skill/registry.ts"; import { addSpanEvent, + markSpanFailed, setSpanAttributes, withSpan, } from "#veryfront/observability/tracing/otlp-setup.ts"; @@ -226,6 +229,8 @@ function getRuntimeInferenceCredential(input: RuntimeRunAgentInput): string | un return runtimeInferenceCredentials.get(input); } +const controlPlaneInjectedTools = new WeakSet(); + function createInjectedStudioTool( runId: string, toolName: string, @@ -258,6 +263,7 @@ function createInjectedStudioTool( return waitResult.result; }, }; + controlPlaneInjectedTools.add(tool); return controlPlaneNames.some((name) => toolName === `veryfront__${name}`) ? markTrustedHostToolProvenance(tool) : tool; @@ -271,6 +277,33 @@ const controlPlaneNames = [ "studio_todo_write", ]; +const CHILD_RUN_CONTROL_PLANE_TOOL_NAMES = new Set([ + INVOKE_AGENT_TOOL_ID, + `veryfront__${INVOKE_AGENT_TOOL_ID}`, +]); + +/** + * Tool names whose calls run a child agent: control-plane delegation, or the + * framework's invoke_agent and `agent_{id}` delegate tools. A custom or registry + * tool that merely shares such a name is not one. + */ +function resolveChildRunToolNames(mergedTools: Agent["config"]["tools"]): Set { + const names = new Set(); + if (!mergedTools) return names; + const entries = mergedTools === true ? toolRegistry.getAll() : Object.entries(mergedTools); + for (const [toolName, entry] of entries) { + const tool = entry === true ? toolRegistry.get(toolName) : entry; + if ( + isFrameworkChildRunTool(tool) || + (CHILD_RUN_CONTROL_PLANE_TOOL_NAMES.has(toolName) && isRecord(tool) && + controlPlaneInjectedTools.has(tool as Tool)) + ) { + names.add(toolName); + } + } + return names; +} + function isExplicitlyDeniedToolName( agent: Agent, deniedToolNames: ReadonlySet, @@ -1059,6 +1092,7 @@ export async function createRuntimeAgentStreamResponse( const modelCallContextRelay = createModelCallContextRelay(timing); const providerReplayCheckpointRelay = createProviderReplayCheckpointRelay(); let shouldEmitProviderReplayCheckpoints = false; + let childRunToolNames = new Set(); try { const executionModel = getAgentExecutionConfig(agent.config).model ?? resolveConfiguredAgentModel(); @@ -1159,6 +1193,7 @@ export async function createRuntimeAgentStreamResponse( modelSupportedProviderToolNames.has(toolName) && !isExplicitlyDeniedToolName(agent, explicitlyDeniedToolNames, toolName, deps.localTools) ); + childRunToolNames = resolveChildRunToolNames(mergedTools); const mergedToolNames = mergedTools && mergedTools !== true ? Object.keys(mergedTools) : []; const allowedRemoteToolNameSet = new Set(allowedRemoteToolNames ?? []); const forwardedToolNames = (forwardedIntegrationToolDefs?.map((def) => def.name) ?? []) @@ -1366,6 +1401,31 @@ export async function createRuntimeAgentStreamResponse( "Internal agent runtime stream stopped before EOF", ); let readerCancellation: Promise | undefined; + let terminalRunErrorCode: string | undefined; + let toolErrorCount = 0; + let childRunErrorCount = 0; + const childRunToolCallIds = new Set(); + const observeRunOutcomeEvent = (event: string, payload: Record) => { + if ( + event === "ToolCallStart" && typeof payload.toolCallId === "string" && + typeof payload.toolCallName === "string" && + childRunToolNames.has(payload.toolCallName) + ) { + childRunToolCallIds.add(payload.toolCallId); + } + if (event === "ToolCallResult" && payload.isError === true) { + toolErrorCount++; + if ( + typeof payload.toolCallId === "string" && + childRunToolCallIds.has(payload.toolCallId) + ) { + childRunErrorCount++; + } + } + if (event === "RunError") { + terminalRunErrorCode ??= toStableRunErrorCode(payload.code); + } + }; let heartbeatTimer: ReturnType | undefined; stopHeartbeat = () => { if (heartbeatTimer) { @@ -1510,6 +1570,7 @@ export async function createRuntimeAgentStreamResponse( providerReplayStepOpen = false; } prepareToolResultIfNeeded(mappedEvent.event, mappedEvent.payload); + observeRunOutcomeEvent(mappedEvent.event, mappedEvent.payload); enqueueIfAttached(mappedEvent.event, mappedEvent.payload); }; heartbeatTimer = setInterval( @@ -1568,9 +1629,13 @@ export async function createRuntimeAgentStreamResponse( } for (const mappedEvent of finalizeRunEvents(state, completedResponse)) { + observeRunOutcomeEvent(mappedEvent.event, mappedEvent.payload); enqueueIfAttached(mappedEvent.event, mappedEvent.payload); } const finalStatus = state.sawTerminalError ? "failed" : "completed"; + const terminalErrorCode = state.sawTerminalError + ? terminalRunErrorCode ?? "AgentRunTerminalError" + : undefined; if (state.sawTerminalError) { deps.sessionManager.failRun(input.runId); } else { @@ -1586,7 +1651,11 @@ export async function createRuntimeAgentStreamResponse( "agent.run.final_status": finalStatus, "agent.run.saw_visible_output": state.sawVisibleOutput, "agent.run.saw_terminal_error": state.sawTerminalError, - ...(state.sawTerminalError ? { "error.type": "AgentRunTerminalError" } : {}), + "agent.run.tool_error_count": toolErrorCount, + "agent.run.child_run_error_count": childRunErrorCount, + // The RunError message can carry unclassified framework error text, so only + // the stable code leaves the process. + ...(terminalErrorCode ? { "error.type": terminalErrorCode } : {}), // Carries `agent.run.usage_is_floor` when the run never delivered a final // response, which is not the same question as whether it ended in error: // an empty assistant turn reaches here with an exact total and @@ -1600,15 +1669,28 @@ export async function createRuntimeAgentStreamResponse( runSpan, state.sawTerminalError ? "agent.run.failed" : "agent.run.completed", ); - logger.info("Internal agent runtime stream finalized", { + const finalizedLogContext = { runId: input.runId, threadId: input.threadId, + parentRunId: input.parentRunId, + projectId: deps.projectAgentSandbox?.projectId ?? undefined, agentId: agent.id, status: finalStatus, sawVisibleOutput: state.sawVisibleOutput, sawTerminalError: state.sawTerminalError, finishReason: state.metadata.finishReason, - }); + toolErrorCount, + childRunErrorCount, + }; + if (terminalErrorCode) { + markSpanFailed(runSpan, terminalErrorCode); + logger.warn("Internal agent runtime stream finalized", { + ...finalizedLogContext, + errorCode: terminalErrorCode, + }); + } else { + logger.info("Internal agent runtime stream finalized", finalizedLogContext); + } } catch (error) { readerExitReason = error; if (error instanceof AgentRunCancelledError) { @@ -1621,6 +1703,8 @@ export async function createRuntimeAgentStreamResponse( status: "cancelled", }), "agent.run.final_status": "cancelled", + "agent.run.tool_error_count": toolErrorCount, + "agent.run.child_run_error_count": childRunErrorCount, "error.type": "AgentRunCancelledError", "error.message": error.message, // The model call in flight at the abort may have been billed without @@ -1643,6 +1727,7 @@ export async function createRuntimeAgentStreamResponse( } else { deps.sessionManager.failRun(input.runId); const errorMessage = error instanceof Error ? error.message : String(error); + const runErrorCode = readProviderReplayTurnErrorCode(error) ?? "RUNTIME_ERROR"; setSpanAttributes(runSpan, { ...buildInternalAgentRunTraceAttributes({ runInput: input, @@ -1651,21 +1736,28 @@ export async function createRuntimeAgentStreamResponse( status: "failed", }), "agent.run.final_status": "failed", - "error.type": error instanceof Error ? error.name : "Error", + "agent.run.tool_error_count": toolErrorCount, + "agent.run.child_run_error_count": childRunErrorCount, + "error.type": runErrorCode, + "error.cause.type": error instanceof Error ? error.name : "Error", "error.message": errorMessage, // The model call in flight at the failure may have been billed without // ever reporting usage, so an accumulator total is marked a floor. ...resolveRunUsageAttributes(), }); addSpanEvent(runSpan, "agent.run.failed"); + markSpanFailed(runSpan, runErrorCode); logger.error("Internal agent runtime stream failed", { runId: input.runId, threadId: input.threadId, + parentRunId: input.parentRunId, + projectId: deps.projectAgentSandbox?.projectId ?? undefined, agentId: agent.id, + errorCode: runErrorCode, error: errorMessage, }); enqueueIfAttached("RunError", { - code: readProviderReplayTurnErrorCode(error) ?? "RUNTIME_ERROR", + code: runErrorCode, message: errorMessage, }); } diff --git a/src/observability/tracing/otlp-setup.test.ts b/src/observability/tracing/otlp-setup.test.ts index 32e30171a0..d675b85d0f 100644 --- a/src/observability/tracing/otlp-setup.test.ts +++ b/src/observability/tracing/otlp-setup.test.ts @@ -174,6 +174,58 @@ describe("observability/tracing/otlp-setup", () => { } }); + it("markSpanFailed sets ERROR status with the error code on a settled span", async () => { + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + setGlobalContextAccessor({ + active: () => createTestContext(), + with: (_context, fn) => fn(), + }); + setGlobalTracerProvider({ + getTracer(name, version) { + return provider.getTracer(name, version) as unknown as Tracer; + }, + }); + const { markSpanFailed, withSpan } = await import("./otlp-setup.ts"); + + try { + await withSpan("agent.run", async (span) => { + markSpanFailed(span, "insufficient-credits"); + }); + await provider.forceFlush(); + + const [finishedSpan] = exporter.getFinishedSpans(); + assertExists(finishedSpan); + assertEquals(finishedSpan.status.code, SpanStatusCode.ERROR); + assertEquals(finishedSpan.status.message, "insufficient-credits"); + assertEquals(finishedSpan.events[0]?.attributes?.["exception.stacktrace"], undefined); + } finally { + _resetShimForTests(); + await provider.shutdown(); + } + }); + + it("markSpanFailed ignores a missing span and survives a failing provider", async () => { + const { markSpanFailed } = await import("./otlp-setup.ts"); + const recordedMessages: string[] = []; + const span = createTestSpan({ + setStatus: () => { + throw new Error("telemetry status failed"); + }, + recordException: (exception) => { + recordedMessages.push((exception as Error).message); + }, + }); + + markSpanFailed(null, "RUNTIME_ERROR"); + assertEquals(recordedMessages, []); + + markSpanFailed(span, "RUNTIME_ERROR"); + assertEquals(recordedMessages, ["RUNTIME_ERROR"]); + }); + it("withSpan preserves callback outcomes when span completion fails", async () => { const { withSpan } = await import("./otlp-setup.ts"); const applicationError = new Error("application failed"); diff --git a/src/observability/tracing/otlp-setup.ts b/src/observability/tracing/otlp-setup.ts index 0a3c71433b..f49bca22a9 100644 --- a/src/observability/tracing/otlp-setup.ts +++ b/src/observability/tracing/otlp-setup.ts @@ -614,6 +614,19 @@ export function setActiveSpanErrorStatus(error: unknown): void { setSpanErrorStatus(span, error, "withoutStack"); } +/** + * Marks a span as failed with a stable error code. + * + * For work that settles its failure itself instead of throwing through the span, + * so `withSpan` never sees it. The code becomes the status message, so it must be + * a bounded classification rather than free text. + */ +export function markSpanFailed(span: unknown, errorCode: string): void { + if (!span) return; + + setSpanErrorStatus(unwrapPublicSpan(span as Span), new Error(errorCode), "withoutStack"); +} + /** Context for with. */ export async function withContext(spanContext: unknown, fn: () => Promise): Promise { return await runAsyncWithContextFallback( diff --git a/src/observability/tracing/service-tracer.test.ts b/src/observability/tracing/service-tracer.test.ts index dce913b1b0..296a98ec22 100644 --- a/src/observability/tracing/service-tracer.test.ts +++ b/src/observability/tracing/service-tracer.test.ts @@ -21,7 +21,7 @@ type FakeSpanOptions = { class FakeSpan { readonly context: { traceId: string; spanId: string }; readonly attributes: Record = {}; - status: { code: number } | null = null; + status: { code: number; message?: string } | null = null; exceptions: unknown[] = []; ended = false; throwOnSetAttribute = false; @@ -47,7 +47,7 @@ class FakeSpan { return this; } - setStatus(status: { code: number }): FakeSpan { + setStatus(status: { code: number; message?: string }): FakeSpan { this.status = status; return this; } @@ -569,6 +569,25 @@ describe("observability/tracing/service-tracer", () => { }); }); + it("marks a manual span failed with its error code as the status message", () => { + const harness = createHarness(); + const serviceTracer = createOpenTelemetryServiceTracer({ + serviceName: "test-service", + context: harness.contextApi, + trace: harness.traceApi, + errorStatusCode: 2, + }); + + const span = serviceTracer.tracer.startSpan("manual-operation"); + span.markFailed?.("STREAM_ERROR"); + + const otelSpan = harness.startedSpans[0]; + assertEquals(otelSpan?.status, { code: 2, message: "STREAM_ERROR" }); + assertEquals(otelSpan?.exceptions.length, 1); + assertEquals((otelSpan?.exceptions[0] as Error).message, "STREAM_ERROR"); + assertEquals((otelSpan?.exceptions[0] as Error).stack, undefined); + }); + it("starts a childOf span in its declared parent's context", () => { const harness = createHarness(); const serviceTracer = createOpenTelemetryServiceTracer({ @@ -782,8 +801,14 @@ describe("observability/tracing/service-tracer", () => { otelSpan.throwOnSetAttribute = true; otelSpan.throwOnEnd = true; + otelSpan.setStatus = () => { + throw new Error("telemetry status failure"); + }; + span.setTag("safe", "value"); span.setAttributes({ another: "value" }); + span.markFailed?.("STREAM_ERROR"); span.finish(); + assertEquals(otelSpan.exceptions.length, 1); }); }); diff --git a/src/observability/tracing/service-tracer.ts b/src/observability/tracing/service-tracer.ts index 3df3e6ca88..53d8b64bd9 100644 --- a/src/observability/tracing/service-tracer.ts +++ b/src/observability/tracing/service-tracer.ts @@ -9,6 +9,7 @@ import { sanitizeErrorForTelemetry, sanitizeTelemetryAttributeValue, sanitizeTelemetryText, + type TelemetryErrorDetail, } from "../telemetry-error.ts"; import { runSyncWithContextFallback } from "./context-callback.ts"; @@ -30,7 +31,7 @@ export type OpenTelemetrySpan = { ServiceTracerAttributePrimitive | readonly ServiceTracerAttributePrimitive[] >, ): unknown; - setStatus(status: { code: number }): unknown; + setStatus(status: { code: number; message?: string }): unknown; recordException(error: unknown): unknown; end(): unknown; spanContext(): OpenTelemetrySpanContext; @@ -80,6 +81,8 @@ export type ServiceTracerSpan< > = { setTag(key: string, value: ServiceTracerAttributeInput): TSpan; setAttributes(attributes: Record): TSpan; + /** Sets ERROR status with a stable error code, for work that settles its failure without throwing. */ + markFailed?(errorCode: string): void; finish(): void; withContext(fn: () => T): T; context(): ServiceTracerSpanContext | undefined; @@ -185,6 +188,7 @@ function createTracerSpan( contextApi: OpenTelemetryContextApi, span: TSpan, context: TContext, + errorStatusCode: number, ): ServiceTracerSpan { return { setTag: (key, value) => { @@ -202,6 +206,9 @@ function createTracerSpan( } return span; }, + markFailed: (errorCode) => { + setSpanErrorStatus(span, errorStatusCode, new Error(errorCode), "withoutStack", errorCode); + }, finish: () => { endSpan(span); }, @@ -230,14 +237,16 @@ function setSpanErrorStatus( span: TSpan, errorStatusCode: number, error: unknown, + detail: TelemetryErrorDetail = "withStack", + message?: string, ): void { try { - span.setStatus({ code: errorStatusCode }); + span.setStatus(message ? { code: errorStatusCode, message } : { code: errorStatusCode }); } catch (_) { /* expected: telemetry failures must not replace application failures */ } try { - span.recordException(sanitizeErrorForTelemetry(error)); + span.recordException(sanitizeErrorForTelemetry(error, detail)); } catch (_) { /* expected: telemetry failures must not replace application failures */ } @@ -540,7 +549,7 @@ export function createOpenTelemetryServiceTracer< const span = startSpan(name, startOptions, parentContext); const spanContext = setSpanOnContext(parentContext, span); - return createTracerSpan(options.context, span, spanContext); + return createTracerSpan(options.context, span, spanContext, options.errorStatusCode); }, scope: () => ({ active: () => { @@ -548,7 +557,12 @@ export function createOpenTelemetryServiceTracer< const activeSpan = getSpanFromContext(activeContext); if (!activeSpan) return null; - return createTracerSpan(options.context, activeSpan, activeContext); + return createTracerSpan( + options.context, + activeSpan, + activeContext, + options.errorStatusCode, + ); }, }), wrap,