diff --git a/src/browser/features/Messages/ToolMessage.tsx b/src/browser/features/Messages/ToolMessage.tsx index 27a28c3164..ae8692bd68 100644 --- a/src/browser/features/Messages/ToolMessage.tsx +++ b/src/browser/features/Messages/ToolMessage.tsx @@ -63,8 +63,14 @@ export const ToolMessage: React.FC = ({ toolCallId={toolCallId} // Workflow-specific workflowRunHint={message.workflowRun} - // Bash-specific - startedAt={message.timestamp} + // Elapsed timers (bash/advisor/task_await): start when execute() actually + // began running, not when the model emitted the call — parallel tool calls + // run sequentially, so queued calls must not accumulate elapsed time. + startedAt={message.executionStartedAt} + // Freshness lower bound (task/workflow discovery): when the model emitted the + // call. Unlike executionStartedAt this survives history replay of parts that + // predate execution-start tracking. + toolCallTimestamp={message.timestamp} // FileEdit-specific onReviewNote={onReviewNote} // ProposePlan-specific diff --git a/src/browser/features/Tools/Shared/ElapsedTimeDisplay.tsx b/src/browser/features/Tools/Shared/ElapsedTimeDisplay.tsx index d758c1d85b..1bdf0532ed 100644 --- a/src/browser/features/Tools/Shared/ElapsedTimeDisplay.tsx +++ b/src/browser/features/Tools/Shared/ElapsedTimeDisplay.tsx @@ -10,6 +10,10 @@ interface ElapsedTimeDisplayProps { /** * Shared elapsed time display for tool headers. * Keeps requestAnimationFrame + per-second updates at the leaf so parent tool calls do not re-render. + * + * Renders nothing until `startedAt` is known: for tool calls, that is when execute() + * actually begins running. Parallel tool calls run sequentially, so a queued call has + * no start time yet and must not show a ticking timer (it could exceed its own timeout). */ export const ElapsedTimeDisplay: React.FC = ({ startedAt, @@ -23,7 +27,7 @@ export const ElapsedTimeDisplay: React.FC = ({ const baseStart = useRef(startedAt ?? Date.now()); useEffect(() => { - if (!isActive) { + if (!isActive || startedAt === undefined) { elapsedRef.current = 0; if (frameRef.current !== null) { cancelAnimationFrame(frameRef.current); @@ -32,7 +36,7 @@ export const ElapsedTimeDisplay: React.FC = ({ return; } - baseStart.current = startedAt ?? Date.now(); + baseStart.current = startedAt; let lastSecond = -1; const tick = () => { @@ -60,7 +64,7 @@ export const ElapsedTimeDisplay: React.FC = ({ }; }, [isActive, startedAt]); - if (!isActive || elapsedRef.current === 0) { + if (!isActive || startedAt === undefined || elapsedRef.current === 0) { return null; } diff --git a/src/browser/features/Tools/TaskToolCall.tsx b/src/browser/features/Tools/TaskToolCall.tsx index 2f49d86006..63f5a88993 100644 --- a/src/browser/features/Tools/TaskToolCall.tsx +++ b/src/browser/features/Tools/TaskToolCall.tsx @@ -437,6 +437,8 @@ interface TaskToolCallProps { workspaceId?: string; toolCallId?: string; startedAt?: number; + /** When the model emitted the call; freshness fallback when startedAt is unknown. */ + toolCallTimestamp?: number; } interface TaskToolDisplayEntry { @@ -836,6 +838,7 @@ export const TaskToolCall: React.FC = ({ taskReportLinking, toolCallId, startedAt, + toolCallTimestamp, }) => { const errorResult = isToolErrorResult(result) ? result : null; const successResult: TaskToolSuccessResult | null = @@ -869,7 +872,10 @@ export const TaskToolCall: React.FC = ({ requestedCandidateCount: requestedTaskGroupCount, requestedGroupKind: taskGroupKind, knownTaskIds: [...resultTaskIds, ...liveTaskIds, ...recoveredTaskIdsRef.current], - toolStartedAt: startedAt, + // Prefer the true execution start; fall back to the model-emission timestamp for + // parts without execution-start tracking (history replay). Both are valid lower + // bounds on when this call could have created child workspaces. + toolStartedAt: startedAt ?? toolCallTimestamp, workspaceMetadata, }); if (recoveredWorkspaceEntries.length > 0) { diff --git a/src/browser/features/Tools/WorkflowRunToolCall.tsx b/src/browser/features/Tools/WorkflowRunToolCall.tsx index e4545c972b..698fa7e39d 100644 --- a/src/browser/features/Tools/WorkflowRunToolCall.tsx +++ b/src/browser/features/Tools/WorkflowRunToolCall.tsx @@ -79,6 +79,8 @@ interface WorkflowRunToolCallProps { workspaceId?: string; toolCallId?: string; startedAt?: number; + /** When the model emitted the call; freshness fallback when startedAt is unknown. */ + toolCallTimestamp?: number; /** Which tool rendered this card; controls the header icon. */ toolName?: "workflow_run" | "workflow_resume"; /** @@ -1159,10 +1161,14 @@ export const WorkflowRunToolCall: React.FC = ({ workspaceId, toolCallId, startedAt, + toolCallTimestamp, toolName = "workflow_run", knownRunId, workflowRunHint: explicitWorkflowRunHint, }) => { + // Freshness bound for run discovery: prefer the true execution start, but fall back to + // the model-emission timestamp for parts without execution-start tracking (history replay). + const discoveryFreshnessBound = startedAt ?? toolCallTimestamp; const apiState = useContext(APIContext); const commandRegistry = useOptionalCommandRegistry(); const registerCommandSource = commandRegistry?.registerSource; @@ -1395,7 +1401,11 @@ export const WorkflowRunToolCall: React.FC = ({ const discover = async () => { try { const runs = await apiState.api.workflows.listRuns({ workspaceId }); - const foregroundRun = findForegroundWorkflowRun({ runs, args, startedAt }); + const foregroundRun = findForegroundWorkflowRun({ + runs, + args, + startedAt: discoveryFreshnessBound, + }); if (!ignore && foregroundRun != null) { setRefreshedRun((current) => getNewestWorkflowRunSnapshot(current, foregroundRun)); } @@ -1412,7 +1422,7 @@ export const WorkflowRunToolCall: React.FC = ({ ignore = true; window.clearInterval(interval); }; - }, [apiState?.api, args, runId, startedAt, status, workspaceId]); + }, [apiState?.api, args, runId, discoveryFreshnessBound, status, workspaceId]); const exactDiscoveryRunId = knownRunId ?? workflowRunHint?.runId; diff --git a/src/browser/stores/WorkspaceStore.ts b/src/browser/stores/WorkspaceStore.ts index b303720103..c98f60d99b 100644 --- a/src/browser/stores/WorkspaceStore.ts +++ b/src/browser/stores/WorkspaceStore.ts @@ -897,6 +897,10 @@ export class WorkspaceStore { applyWorkspaceChatEventToAggregator(aggregator, data); this.states.bump(workspaceId); }, + "tool-call-execution-start": (workspaceId, aggregator, data) => { + applyWorkspaceChatEventToAggregator(aggregator, data); + this.states.bump(workspaceId); + }, "tool-call-delta": (workspaceId, aggregator, data) => { applyWorkspaceChatEventToAggregator(aggregator, data); this.scheduleStreamingStateBump(workspaceId); diff --git a/src/browser/utils/messages/StreamingMessageAggregator.test.ts b/src/browser/utils/messages/StreamingMessageAggregator.test.ts index 775154da2c..f39478bb7a 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.test.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.test.ts @@ -1903,6 +1903,201 @@ describe("StreamingMessageAggregator", () => { }); }); + test("tool-call-execution-start stamps executionStartedAt onto the displayed tool row", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + startTestStream(aggregator, { messageId: "msg-1" }); + startToolCall(aggregator, { + toolCallId: "tool-queued", + toolName: "bash", + args: { command: "echo hi" }, + timestamp: 1_000, + }); + + // Queued behind a sibling: no execution start yet. + const queuedRow = aggregator + .getDisplayedMessages() + .find((m) => m.type === "tool" && m.toolCallId === "tool-queued"); + expect(queuedRow?.type).toBe("tool"); + expect(queuedRow?.type === "tool" ? queuedRow.executionStartedAt : null).toBeUndefined(); + + aggregator.handleToolCallExecutionStart({ + type: "tool-call-execution-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-queued", + timestamp: 5_000, + }); + + const executingRow = aggregator + .getDisplayedMessages() + .find((m) => m.type === "tool" && m.toolCallId === "tool-queued"); + expect(executingRow?.type === "tool" ? executingRow.executionStartedAt : undefined).toBe(5_000); + }); + + test("tool execution stats measure from execution start, not queue entry", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + startTestStream(aggregator, { messageId: "msg-1" }); + // Two parallel tool calls emitted back-to-back; they execute sequentially. + startToolCall(aggregator, { + toolCallId: "tool-a", + toolName: "bash", + args: {}, + timestamp: 1_000, + }); + startToolCall(aggregator, { + toolCallId: "tool-b", + toolName: "bash", + args: {}, + timestamp: 1_001, + }); + + aggregator.handleToolCallExecutionStart({ + type: "tool-call-execution-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-a", + timestamp: 1_002, + }); + endToolCall(aggregator, { + toolCallId: "tool-a", + toolName: "bash", + result: {}, + timestamp: 5_000, + }); + aggregator.handleToolCallExecutionStart({ + type: "tool-call-execution-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-b", + timestamp: 5_001, + }); + endToolCall(aggregator, { + toolCallId: "tool-b", + toolName: "bash", + result: {}, + timestamp: 9_000, + }); + + const stats = aggregator.getActiveStreamTimingStats(); + // A: 5000-1002, B: 9000-5001. Without execution-start re-anchoring, B would + // count from emission (9000-1001) and double-count A's run time. + expect(stats?.toolExecutionMs).toBe(3_998 + 3_999); + }); + + test("replayed tool-call-start merges executionStartedAt into an existing tool part", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + startTestStream(aggregator, { messageId: "msg-1" }); + // Client saw the queued tool-call-start live (no execution start yet)... + startToolCall(aggregator, { + toolCallId: "tool-reconnect", + toolName: "bash", + args: { command: "echo hi" }, + timestamp: 1_000, + }); + + // ...disconnected, execute() began, and the `since` replay re-sends the enriched start. + aggregator.handleToolCallStart({ + type: "tool-call-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-reconnect", + toolName: "bash", + args: { command: "echo hi" }, + tokens: 0, + timestamp: 1_000, + executionStartedAt: 6_000, + replay: true, + }); + + const row = aggregator + .getDisplayedMessages() + .find((m) => m.type === "tool" && m.toolCallId === "tool-reconnect"); + expect(row?.type === "tool" ? row.executionStartedAt : undefined).toBe(6_000); + + // The merge must also re-anchor timing stats: the tool's duration counts from + // execution start (6000), not the pre-disconnect emission seed (1000). + endToolCall(aggregator, { + toolCallId: "tool-reconnect", + toolName: "bash", + result: {}, + timestamp: 9_000, + }); + expect(aggregator.getActiveStreamTimingStats()?.toolExecutionMs).toBe(3_000); + }); + + test("stashes execution starts that arrive before their tool part exists", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + startTestStream(aggregator, { messageId: "msg-1" }); + // Reconnect replay in progress: the live execution-start (not replay-buffered) + // arrives before the replayed tool-call-start creates the part. + aggregator.handleToolCallExecutionStart({ + type: "tool-call-execution-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-race", + timestamp: 6_000, + }); + + // Replayed start without executionStartedAt (snapshot predates execute()). + aggregator.handleToolCallStart({ + type: "tool-call-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-race", + toolName: "bash", + args: { command: "echo hi" }, + tokens: 0, + timestamp: 1_000, + replay: true, + }); + + const row = aggregator + .getDisplayedMessages() + .find((m) => m.type === "tool" && m.toolCallId === "tool-race"); + expect(row?.type === "tool" ? row.executionStartedAt : undefined).toBe(6_000); + + // Timing stats must also anchor at execution start, not the replayed emission time. + endToolCall(aggregator, { + toolCallId: "tool-race", + toolName: "bash", + result: {}, + timestamp: 9_000, + }); + expect(aggregator.getActiveStreamTimingStats()?.toolExecutionMs).toBe(3_000); + }); + + test("fresh replayed tool-call-start seeds timing stats from executionStartedAt", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + startTestStream(aggregator, { messageId: "msg-1" }); + // Full replay after reconnect: the client never saw the live start, and the + // replayed event already carries the true execution start. + aggregator.handleToolCallStart({ + type: "tool-call-start", + workspaceId: TEST_WORKSPACE_ID, + messageId: "msg-1", + toolCallId: "tool-fresh-replay", + toolName: "bash", + args: { command: "echo hi" }, + tokens: 0, + timestamp: 1_000, + executionStartedAt: 6_000, + replay: true, + }); + endToolCall(aggregator, { + toolCallId: "tool-fresh-replay", + toolName: "bash", + result: {}, + timestamp: 9_000, + }); + + expect(aggregator.getActiveStreamTimingStats()?.toolExecutionMs).toBe(3_000); + }); + test("keeps richer in-memory parts when append replay sends a stale duplicate", () => { const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); diff --git a/src/browser/utils/messages/StreamingMessageAggregator.ts b/src/browser/utils/messages/StreamingMessageAggregator.ts index 22f38c6a07..cf5bc4ec39 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.ts @@ -17,6 +17,7 @@ import { type StreamEndEvent, type StreamAbortEvent, type StreamAbortReasonSnapshot, + type ToolCallExecutionStartEvent, type ToolCallStartEvent, type ToolCallDeltaEvent, type ToolCallEndEvent, @@ -614,6 +615,13 @@ export class StreamingMessageAggregator { // Shows "interrupting..." in StreamingBarrier until real stream-abort arrives private interruptingMessageId: string | null = null; + // Execution starts that arrived before their tool part exists (messageId → toolCallId → + // timestamp). During reconnect replay, live tool-call-execution-start events are not + // replay-buffered and can beat the replayed tool-call-start that creates the part; + // dropping them would leave the running tool's elapsed timer hidden until the next + // reconnect. Consumed in handleToolCallStart, cleared in cleanupStreamState. + private stashedToolExecutionStarts = new Map>(); + // Session-level timing stats: model -> stats (totals computed on-the-fly) private sessionTimingStats: Record< string, @@ -923,6 +931,9 @@ export class StreamingMessageAggregator { this.interruptingMessageId = null; } + // Drop unconsumed execution starts (e.g. the tool part never materialized). + this.stashedToolExecutionStarts.delete(messageId); + // Capture timing stats before removing the stream context const context = this.activeStreams.get(messageId); if (context) { @@ -2366,16 +2377,36 @@ export class StreamingMessageAggregator { part.type === "dynamic-tool" && part.toolCallId === data.toolCallId ); + // A live tool-call-execution-start may have arrived before this (replayed) start + // created the part; fold the stashed timestamp in as if the event carried it. + const executionStartedAt = + data.executionStartedAt ?? + this.takeStashedToolExecutionStart(data.messageId, data.toolCallId); + if (existingToolPart) { + // A `since` replay re-sends tool-call-start when a queued tool's execute() began + // after the reconnect cursor. Merge the enriched execution start into the existing + // row (otherwise the elapsed timer stays hidden) instead of dropping the event. + if (executionStartedAt !== undefined && existingToolPart.executionStartedAt === undefined) { + existingToolPart.executionStartedAt = executionStartedAt; + // Also re-anchor the timing-stats start (seeded from the live tool-call-start's + // emission timestamp before disconnect), mirroring handleToolCallExecutionStart — + // otherwise tool-call-end still counts queue wait as execution time. + this.reanchorPendingToolStart(data.messageId, data.toolCallId, executionStartedAt); + this.markMessageDirty(data.messageId); + return; + } console.warn(`Tool call ${data.toolCallId} already exists, skipping duplicate`); return; } - // Track tool start time for execution duration calculation + // Track tool start time for execution duration calculation. + // Replayed starts may already carry the true execution start; prefer it so + // toolExecutionMs never counts time spent queued behind serialized siblings. const context = this.activeStreams.get(data.messageId); if (context) { this.updateStreamClock(context, data.timestamp); - context.pendingToolStarts.set(data.toolCallId, data.timestamp); + context.pendingToolStarts.set(data.toolCallId, executionStartedAt ?? data.timestamp); } // Add tool part to maintain temporal order @@ -2386,6 +2417,8 @@ export class StreamingMessageAggregator { state: "input-available", input: data.args, timestamp: data.timestamp, + // Present on replay when execute() had already begun before reconnect. + ...(executionStartedAt !== undefined ? { executionStartedAt } : {}), }; message.parts.push(toolPart); @@ -2401,6 +2434,69 @@ export class StreamingMessageAggregator { // Tool deltas are for display - args are in dynamic-tool part } + /** + * Re-anchor the timing-stats start for a tool whose execute() actually began running. + * toolExecutionMs sums per-tool durations at tool-call-end; without this, each + * serialized sibling's duration would include the time it spent queued behind earlier + * siblings (double-counting tool time and deflating derived streaming/tok-s stats). + * Provider-executed tools never report an execution start and keep their + * tool-call-start anchor. + */ + private reanchorPendingToolStart( + messageId: string, + toolCallId: string, + executionStartedAt: number + ): void { + const context = this.activeStreams.get(messageId); + if (context) { + this.updateStreamClock(context, executionStartedAt); + if (context.pendingToolStarts.has(toolCallId)) { + context.pendingToolStarts.set(toolCallId, executionStartedAt); + } + } + } + + /** + * Mark when a tool call's execute() actually began running. Parallel tool calls are + * serialized in the backend, so this arrives once the call reaches the front of the + * queue — elapsed timers start here instead of at tool-call-start. + */ + handleToolCallExecutionStart(data: ToolCallExecutionStartEvent): void { + this.reanchorPendingToolStart(data.messageId, data.toolCallId, data.timestamp); + + const message = this.messages.get(data.messageId); + const toolPart = message?.parts.find( + (part): part is DynamicToolPart => + part.type === "dynamic-tool" && part.toolCallId === data.toolCallId + ); + if (!toolPart) { + // Live event won the race against the replayed tool-call-start that creates the + // part (execution-start events are not replay-buffered). Stash it for + // handleToolCallStart to apply. + const stash = + this.stashedToolExecutionStarts.get(data.messageId) ?? new Map(); + stash.set(data.toolCallId, data.timestamp); + this.stashedToolExecutionStarts.set(data.messageId, stash); + return; + } + + toolPart.executionStartedAt = data.timestamp; + this.markMessageDirty(data.messageId); + } + + /** Consume a stashed execution start that arrived before its tool part existed. */ + private takeStashedToolExecutionStart(messageId: string, toolCallId: string): number | undefined { + const stash = this.stashedToolExecutionStarts.get(messageId); + const timestamp = stash?.get(toolCallId); + if (stash && timestamp !== undefined) { + stash.delete(toolCallId); + if (stash.size === 0) { + this.stashedToolExecutionStarts.delete(messageId); + } + } + return timestamp; + } + private trackLoadedSkill(skill: LoadedSkill): void { const existing = this.loadedSkills.get(skill.name); if ( diff --git a/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.test.ts b/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.test.ts index 078c85939f..9877f5f4b7 100644 --- a/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.test.ts +++ b/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.test.ts @@ -12,6 +12,7 @@ import type { StreamStartEvent, ToolCallDeltaEvent, ToolCallEndEvent, + ToolCallExecutionStartEvent, ToolCallStartEvent, UsageDeltaEvent, } from "@/common/types/stream"; @@ -51,6 +52,10 @@ class StubAggregator implements WorkspaceChatEventAggregator { this.calls.push(`handleToolCallStart:${data.toolCallId}`); } + handleToolCallExecutionStart(data: ToolCallExecutionStartEvent): void { + this.calls.push(`handleToolCallExecutionStart:${data.toolCallId}`); + } + handleToolCallDelta(data: ToolCallDeltaEvent): void { this.calls.push(`handleToolCallDelta:${data.toolCallId}`); } diff --git a/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.ts b/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.ts index d8cb10f1b6..f846950aba 100644 --- a/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.ts +++ b/src/browser/utils/messages/applyWorkspaceChatEventToAggregator.ts @@ -24,6 +24,7 @@ import { isStreamStart, isToolCallDelta, isToolCallEnd, + isToolCallExecutionStart, isToolCallStart, isUsageDelta, } from "@/common/orpc/types"; @@ -36,6 +37,7 @@ import type { StreamStartEvent, ToolCallDeltaEvent, ToolCallEndEvent, + ToolCallExecutionStartEvent, ToolCallStartEvent, UsageDeltaEvent, RuntimeStatusEvent, @@ -67,6 +69,7 @@ export interface WorkspaceChatEventAggregator { handleStreamError(data: StreamErrorMessage): void; handleToolCallStart(data: ToolCallStartEvent): void; + handleToolCallExecutionStart(data: ToolCallExecutionStartEvent): void; handleToolCallDelta(data: ToolCallDeltaEvent): void; handleToolCallEnd(data: ToolCallEndEvent): void; @@ -167,6 +170,11 @@ export function applyWorkspaceChatEventToAggregator( return "immediate"; } + if (isToolCallExecutionStart(event)) { + aggregator.handleToolCallExecutionStart(event); + return "immediate"; + } + if (isToolCallDelta(event)) { aggregator.handleToolCallDelta(event); return "throttled"; diff --git a/src/browser/utils/messages/displayedMessageBuilder.ts b/src/browser/utils/messages/displayedMessageBuilder.ts index 8a16b26a29..53b8ec3270 100644 --- a/src/browser/utils/messages/displayedMessageBuilder.ts +++ b/src/browser/utils/messages/displayedMessageBuilder.ts @@ -524,6 +524,7 @@ function appendToolRows( isLastPartOfMessage: options.isLastPartOfMessage, ...(part.workflowRun != null ? { workflowRun: part.workflowRun } : {}), timestamp: part.timestamp ?? options.baseTimestamp, + ...(part.executionStartedAt != null ? { executionStartedAt: part.executionStartedAt } : {}), nestedCalls, }); } diff --git a/src/common/orpc/schemas.ts b/src/common/orpc/schemas.ts index 1597178de0..7d164ba622 100644 --- a/src/common/orpc/schemas.ts +++ b/src/common/orpc/schemas.ts @@ -258,6 +258,7 @@ export { StreamStartEventSchema, ToolCallDeltaEventSchema, ToolCallEndEventSchema, + ToolCallExecutionStartEventSchema, ToolCallStartEventSchema, BashOutputEventSchema, TaskCreatedEventSchema, diff --git a/src/common/orpc/schemas/message.ts b/src/common/orpc/schemas/message.ts index 2d5afd47ce..81193f9b02 100644 --- a/src/common/orpc/schemas/message.ts +++ b/src/common/orpc/schemas/message.ts @@ -39,6 +39,11 @@ const MuxToolPartBase = z.object({ toolName: z.string(), input: z.unknown(), timestamp: z.number().optional(), + // When the tool's execute() actually began running. Parallel tool calls are + // serialized (withSequentialExecution), so this can be much later than + // `timestamp` (when the model emitted the call). UI elapsed timers must use + // this so queued-but-not-yet-executing tools don't appear to run. + executionStartedAt: z.number().optional(), workflowRun: WorkflowRunToolAttachmentSchema.optional(), }); diff --git a/src/common/orpc/schemas/stream.ts b/src/common/orpc/schemas/stream.ts index 2d376f0225..6ede6f5603 100644 --- a/src/common/orpc/schemas/stream.ts +++ b/src/common/orpc/schemas/stream.ts @@ -335,9 +335,28 @@ export const ToolCallStartEventSchema = z.object({ args: z.unknown(), tokens: z.number().meta({ description: "Token count for tool input" }), timestamp: z.number().meta({ description: "When tool call started (Date.now())" }), + executionStartedAt: z.number().optional().meta({ + description: + "When the tool's execute() began running, if it already had by the time this event was built (replay); live streams deliver it via tool-call-execution-start instead", + }), parentToolCallId: z.string().optional().meta({ description: "Set for nested PTC calls" }), }); +/** + * Emitted when a tool call's execute() actually begins running. + * + * Parallel tool calls are serialized by withSequentialExecution, so this can fire + * long after tool-call-start (which marks when the model emitted the call). + * The UI uses this to start the elapsed timer only once real execution begins. + */ +export const ToolCallExecutionStartEventSchema = z.object({ + type: z.literal("tool-call-execution-start"), + workspaceId: z.string(), + messageId: z.string(), + toolCallId: z.string(), + timestamp: z.number().meta({ description: "When execute() began running (Date.now())" }), +}); + export const ToolCallDeltaEventSchema = z.object({ type: z.literal("tool-call-delta"), workspaceId: z.string(), @@ -637,6 +656,7 @@ export const WorkspaceChatMessageSchema = z.discriminatedUnion("type", [ StreamAbortEventSchema, // Tool events ToolCallStartEventSchema, + ToolCallExecutionStartEventSchema, ToolCallDeltaEventSchema, ToolCallEndEventSchema, BashOutputEventSchema, diff --git a/src/common/orpc/types.ts b/src/common/orpc/types.ts index 3cc70492f7..3b877d47ba 100644 --- a/src/common/orpc/types.ts +++ b/src/common/orpc/types.ts @@ -8,6 +8,7 @@ import type { StreamEndEvent, StreamAbortEvent, ToolCallStartEvent, + ToolCallExecutionStartEvent, ToolCallDeltaEvent, ToolCallEndEvent, AdvisorOutputEvent, @@ -99,6 +100,12 @@ export function isToolCallStart(msg: WorkspaceChatMessage): msg is ToolCallStart return (msg as { type?: string }).type === "tool-call-start"; } +export function isToolCallExecutionStart( + msg: WorkspaceChatMessage +): msg is ToolCallExecutionStartEvent { + return (msg as { type?: string }).type === "tool-call-execution-start"; +} + export function isToolCallDelta(msg: WorkspaceChatMessage): msg is ToolCallDeltaEvent { return (msg as { type?: string }).type === "tool-call-delta"; } diff --git a/src/common/types/message.ts b/src/common/types/message.ts index a1b3e2121d..33c466e525 100644 --- a/src/common/types/message.ts +++ b/src/common/types/message.ts @@ -817,6 +817,12 @@ export type DisplayedMessage = streamSequence?: number; // Local ordering within this assistant message isLastPartOfMessage?: boolean; // True if this is the last part of a multi-part message timestamp?: number; + /** + * When the tool's execute() actually began running. Parallel tool calls are + * serialized, so this can be much later than `timestamp`; elapsed timers must + * start here (and stay hidden while the call is still queued). + */ + executionStartedAt?: number; /** Durable workflow run attachment recovered from partial history. */ workflowRun?: MuxToolPart["workflowRun"]; // Nested tool calls for code_execution (from PTC streaming or reconstructed from result) diff --git a/src/common/types/stream.ts b/src/common/types/stream.ts index 0e599453f4..c1979b0bd4 100644 --- a/src/common/types/stream.ts +++ b/src/common/types/stream.ts @@ -23,6 +23,7 @@ import type { StreamStartEventSchema, ToolCallDeltaEventSchema, ToolCallEndEventSchema, + ToolCallExecutionStartEventSchema, ToolCallStartEventSchema, BashOutputEventSchema, AdvisorOutputEventSchema, @@ -73,6 +74,7 @@ export type TaskCreatedEvent = z.infer; export type WorkflowRunAttachedEvent = z.infer; export type AdvisorPhaseEvent = z.infer; export type ToolCallStartEvent = z.infer; +export type ToolCallExecutionStartEvent = z.infer; export type ToolCallDeltaEvent = z.infer; export type ToolCallEndEvent = z.infer; diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 34f11d5612..c6657c22f6 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -4550,6 +4550,9 @@ export class AgentSession { this.markActiveStreamHadAnyOutput(); this.emitChatEvent(payload); }); + forward("tool-call-execution-start", (payload) => { + this.emitChatEvent(payload); + }); forward("bash-output", (payload) => { this.markActiveStreamHadAnyOutput(); this.emitChatEvent(payload); diff --git a/src/node/services/aiService.ts b/src/node/services/aiService.ts index f585bd4c9e..2d915016db 100644 --- a/src/node/services/aiService.ts +++ b/src/node/services/aiService.ts @@ -646,6 +646,7 @@ export class AIService extends EventEmitter { "stream-start", "stream-delta", "tool-call-start", + "tool-call-execution-start", "tool-call-delta", "tool-call-end", "reasoning-delta", diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 74195b87f5..8a7af7d8d4 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -6,7 +6,11 @@ import * as path from "node:path"; import { KNOWN_MODELS } from "@/common/constants/knownModels"; import type { ProvidersConfigMap } from "@/common/orpc/types"; import { StreamEndEventSchema } from "@/common/orpc/schemas/stream"; -import type { CompletedMessagePart, WorkflowRunAttachedEvent } from "@/common/types/stream"; +import type { + CompletedMessagePart, + ToolCallExecutionStartEvent, + WorkflowRunAttachedEvent, +} from "@/common/types/stream"; import { Ok, Err } from "@/common/types/result"; import type { ToolPolicy } from "@/common/utils/tools/toolPolicy"; import type { ToolSearchStreamState } from "@/common/utils/tools/toolCatalog"; @@ -203,6 +207,7 @@ function createStreamInfoForTests( lastPartTimestamp: now, toolCompletionTimestamps: new Map(), pendingWorkflowRunAttachments: new Map(), + pendingToolExecutionStarts: new Map(), model, metadataModel: overrides.metadataModel ?? model, historySequence: 1, @@ -348,6 +353,107 @@ describe("StreamManager - workflow run attachments", () => { }); }); +describe("StreamManager - tool execution start timing", () => { + test("stamps executionStartedAt and emits tool-call-execution-start when the part already exists", () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "execution-start-workspace"; + const messageId = "execution-start-message"; + // Seed the monotonic stream clock ahead of wall time: a raw Date.now() execution + // start would be <= the tool-call timestamp and reconnect replay's + // `executionStartedAt > cursor` repair predicate would never fire. + const timestamp = Date.now() + 60_000; + const streamInfo = createStreamInfoForTests({ + messageId, + lastPartTimestamp: timestamp, + parts: [ + { + type: "dynamic-tool", + toolCallId: "tool-call-1", + toolName: "bash", + input: { command: "echo hi" }, + state: "input-available", + timestamp, + }, + ], + }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + const events: ToolCallExecutionStartEvent[] = []; + streamManager.on("tool-call-execution-start", (event: ToolCallExecutionStartEvent) => { + events.push(event); + }); + + const handleToolExecutionStart = getPrivateMethodForTests< + (workspaceId: string, messageId: string, toolCallId: string) => void + >(streamManager, "handleToolExecutionStart"); + handleToolExecutionStart.call(streamManager, workspaceId, messageId, "tool-call-1"); + + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + type: "tool-call-execution-start", + workspaceId, + messageId, + toolCallId: "tool-call-1", + }); + // Cursor-monotonic: strictly after the tool-call part timestamp even when the wall + // clock has not advanced past it. + expect(events[0].timestamp).toBeGreaterThan(timestamp); + + const parts = streamInfo.parts as Array>; + expect(parts[0].executionStartedAt).toBe(events[0].timestamp); + }); + + test("applies execution starts that arrive before the tool part lands", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "execution-start-race-workspace"; + const messageId = "execution-start-race-message"; + const streamInfo = createStreamInfoForTests({ messageId, parts: [] }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + const events: ToolCallExecutionStartEvent[] = []; + streamManager.on("tool-call-execution-start", (event: ToolCallExecutionStartEvent) => { + events.push(event); + }); + + // execute() wins the race: no part yet, so the start is parked as pending. + const handleToolExecutionStart = getPrivateMethodForTests< + (workspaceId: string, messageId: string, toolCallId: string) => void + >(streamManager, "handleToolExecutionStart"); + handleToolExecutionStart.call(streamManager, workspaceId, messageId, "tool-call-race"); + expect(events).toHaveLength(0); + + const appendPartAndEmit = getPrivateMethodForTests< + ( + workspaceId: string, + streamInfo: Record, + part: CompletedMessagePart, + schedulePartialWrite?: boolean + ) => Promise + >(streamManager, "appendPartAndEmit"); + await appendPartAndEmit.call( + streamManager, + workspaceId, + streamInfo, + { + type: "dynamic-tool", + toolCallId: "tool-call-race", + toolName: "bash", + input: { command: "echo hi" }, + state: "input-available", + timestamp: Date.now(), + }, + false + ); + + expect(events).toHaveLength(1); + expect(events[0].toolCallId).toBe("tool-call-race"); + + const parts = streamInfo.parts as Array>; + expect(parts[0].executionStartedAt).toBe(events[0].timestamp); + expect((streamInfo.pendingToolExecutionStarts as Map).size).toBe(0); + }); +}); + describe("StreamManager - createTempDirForStream", () => { test("creates ~/.mux-tmp/ under the runtime's home", async () => { using home = new DisposableTempDir("stream-home"); @@ -4341,6 +4447,70 @@ describe("StreamManager - replayStream", () => { expect(replayedToolEnds).toEqual(["tool-new"]); }); + + test("replayStream replays queued tool parts whose execute() began after the cursor", async () => { + const streamManager = createReplayStreamManager(); + + const workspaceId = "ws-replay-exec-start"; + + const replayedToolStarts: Array<{ toolCallId: string; executionStartedAt?: number }> = []; + streamManager.on( + "tool-call-start", + (event: { + replay?: boolean | undefined; + toolCallId: string; + executionStartedAt?: number; + }) => { + expect(event.replay).toBe(true); + replayedToolStarts.push({ + toolCallId: event.toolCallId, + executionStartedAt: event.executionStartedAt, + }); + } + ); + + const streamInfo = { + state: "streaming", + messageId: "msg-exec-start", + model: "claude-sonnet-4", + historySequence: 1, + startTime: 123, + initialMetadata: {}, + toolCompletionTimestamps: new Map(), + parts: [ + { + // Queued before the cursor, still waiting for the execution lock: not replayed. + type: "dynamic-tool", + toolCallId: "tool-still-queued", + toolName: "bash", + input: {}, + state: "input-available", + timestamp: 10, + }, + { + // Queued before the cursor, execute() began after it: replayed with the + // enriched executionStartedAt so the renderer can start the elapsed timer. + type: "dynamic-tool", + toolCallId: "tool-started-after-cursor", + toolName: "bash", + input: {}, + state: "input-available", + timestamp: 12, + executionStartedAt: 25, + }, + ], + }; + + setReplayStreamInfo(streamManager, workspaceId, streamInfo); + stubReplayTokenTracker(streamManager); + + await streamManager.replayStream(workspaceId, { afterTimestamp: 20 }); + + expect(replayedToolStarts).toEqual([ + { toolCallId: "tool-started-after-cursor", executionStartedAt: 25 }, + ]); + }); + test("replayStream emits replay usage-delta from tracked step/cumulative usage", async () => { const streamManager = createReplayStreamManager(); diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 1ab630a5f2..331421cabb 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -15,6 +15,7 @@ import { } from "ai"; import type { LanguageModelV2Usage } from "@ai-sdk/provider"; import type { Result } from "@/common/types/result"; +import assert from "@/common/utils/assert"; import { Ok, Err } from "@/common/types/result"; import { log, type Logger } from "./log"; import type { @@ -23,6 +24,7 @@ import type { StreamAbortReason, UsageDeltaEvent, ToolCallEndEvent, + ToolCallExecutionStartEvent, CompletedMessagePart, WorkflowRunAttachedEvent, } from "@/common/types/stream"; @@ -477,6 +479,11 @@ interface WorkspaceStreamInfo { // attachment and apply it as soon as the matching dynamic-tool part lands. pendingWorkflowRunAttachments: Map; + // execute() can begin (lock acquired in withSequentialExecution) before the fullStream + // consumer has stored the matching dynamic-tool part. Keep the execution-start timestamp + // and apply it as soon as the part lands. + pendingToolExecutionStarts: Map; + model: string; /** Metadata model resolved from provider mapping for cost/token metadata lookups. */ metadataModel: string; @@ -712,6 +719,65 @@ export class StreamManager extends EventEmitter { return true; } + /** + * Record on the dynamic-tool part when its execute() actually began running and notify + * the UI. Returns false when the part has not landed in streamInfo.parts yet. + */ + private applyToolExecutionStart( + workspaceId: WorkspaceId, + streamInfo: WorkspaceStreamInfo, + toolCallId: string, + timestamp: number + ): boolean { + const partIndex = streamInfo.parts.findIndex( + (part) => part.type === "dynamic-tool" && part.toolCallId === toolCallId + ); + if (partIndex === -1) { + return false; + } + + const part = streamInfo.parts[partIndex]; + assert(part.type === "dynamic-tool", "applyToolExecutionStart matched a non-tool part"); + streamInfo.parts[partIndex] = { ...part, executionStartedAt: timestamp }; + + this.emit("tool-call-execution-start", { + type: "tool-call-execution-start", + workspaceId: workspaceId as string, + messageId: streamInfo.messageId, + toolCallId, + timestamp, + } satisfies ToolCallExecutionStartEvent); + return true; + } + + /** + * Called from withSequentialExecution the moment a tool call's execute() acquires the + * execution lock. Parallel tool calls run sequentially, so this is the honest start of + * execution — the part's own `timestamp` only marks when the model emitted the call. + */ + private handleToolExecutionStart( + workspaceId: WorkspaceId, + messageId: string, + toolCallId: string + ): void { + const streamInfo = this.workspaceStreams.get(workspaceId); + if (streamInfo?.messageId !== messageId) { + return; + } + + // Use the stream's monotonic clock, not raw Date.now(): the tool-call part timestamp + // was monotonicized by nextPartTimestamp(), so a same-millisecond raw reading could be + // <= it. Reconnect replay repairs missed execution starts only when + // executionStartedAt > cursor, and the cursor sits at the tool-call timestamp when the + // client disconnected right after tool-call-start. + const timestamp = nextPartTimestamp(streamInfo); + if (!this.applyToolExecutionStart(workspaceId, streamInfo, toolCallId, timestamp)) { + // execute() won the race against the fullStream consumer; the "tool-call" case + // consumes this entry right after storing the part. + (streamInfo.pendingToolExecutionStarts ??= new Map()).set(toolCallId, timestamp); + } + } + /** * Write the current partial message to disk (throttled by mtime) * Ensures writes happen during rapid streaming (crash-resilient) @@ -1145,6 +1211,11 @@ export class StreamManager extends EventEmitter { args: part.input, tokens, timestamp, + // Replays rebuild parts from scratch; carry the real execution start so + // elapsed timers don't restart from the model-emission timestamp. + ...(part.executionStartedAt !== undefined + ? { executionStartedAt: part.executionStartedAt } + : {}), }); if (part.workflowRun != null) { @@ -1193,13 +1264,35 @@ export class StreamManager extends EventEmitter { // Always persist the part in-memory (and to partial.json, if enabled), even if emit fails. let partToPersist = part; let pendingAttachment: WorkflowRunToolAttachment | undefined; + let pendingExecutionStart: number | undefined; if (part.type === "dynamic-tool") { pendingAttachment = this.takePendingWorkflowRunAttachment(streamInfo, part.toolCallId); - if (pendingAttachment != null) { - partToPersist = { ...part, workflowRun: pendingAttachment }; + // execute() may have started (lock acquired) before the fullStream consumer + // stored this part; carry the real execution start onto the persisted part. + pendingExecutionStart = streamInfo.pendingToolExecutionStarts?.get(part.toolCallId); + if (pendingExecutionStart !== undefined) { + streamInfo.pendingToolExecutionStarts.delete(part.toolCallId); + } + if (pendingAttachment != null || pendingExecutionStart !== undefined) { + partToPersist = { + ...part, + ...(pendingAttachment != null ? { workflowRun: pendingAttachment } : {}), + ...(pendingExecutionStart !== undefined + ? { executionStartedAt: pendingExecutionStart } + : {}), + }; } } streamInfo.parts.push(partToPersist); + if (pendingExecutionStart !== undefined && part.type === "dynamic-tool") { + this.emit("tool-call-execution-start", { + type: "tool-call-execution-start", + workspaceId: workspaceId as string, + messageId: streamInfo.messageId, + toolCallId: part.toolCallId, + timestamp: pendingExecutionStart, + } satisfies ToolCallExecutionStartEvent); + } if (pendingAttachment != null && part.type === "dynamic-tool") { await this.flushPartialWrite(workspaceId, streamInfo); this.emitWorkflowRunAttachedFromAttachment({ @@ -1479,7 +1572,8 @@ export class StreamManager extends EventEmitter { anthropicCacheTtlOverride?: AnthropicCacheTtl, onChunk?: StreamTextOnChunk, onStepMessages?: (messages: ModelMessage[]) => void, - toolSearchState?: ToolSearchStreamState + toolSearchState?: ToolSearchStreamState, + onToolExecutionStart?: (toolCallId: string) => void ): StreamRequestConfig { const finalProviderOptions = providerOptions; @@ -1543,7 +1637,7 @@ export class StreamManager extends EventEmitter { system: finalSystem, // Keep provider-level parallel tool planning enabled, but serialize sibling // execute() handlers inside this stream so shared mutable state cannot race. - tools: withSequentialExecution(finalTools), + tools: withSequentialExecution(finalTools, onToolExecutionStart), providerOptions: finalProviderOptions, headers, maxOutputTokens: effectiveMaxOutputTokens, @@ -1711,7 +1805,8 @@ export class StreamManager extends EventEmitter { anthropicCacheTtlOverride, onChunk, onStepMessages, - toolSearchState + toolSearchState, + (toolCallId) => this.handleToolExecutionStart(workspaceId, messageId, toolCallId) ); // Start streaming - this can throw immediately if API key is missing @@ -1737,6 +1832,7 @@ export class StreamManager extends EventEmitter { lastPartTimestamp: startTime, toolCompletionTimestamps: new Map(), pendingWorkflowRunAttachments: new Map(), + pendingToolExecutionStarts: new Map(), model: modelString, metadataModel, thinkingLevel, @@ -2374,7 +2470,8 @@ export class StreamManager extends EventEmitter { streamInfo.request.onStepMessages, // Same state object: aiService's fallback prepare() rebuilt it in place // against the fallback toolset, so prepareStep keeps reading live state. - streamInfo.request.toolSearchState + streamInfo.request.toolSearchState, + (toolCallId) => this.handleToolExecutionStart(workspaceId, streamInfo.messageId, toolCallId) ); // createStreamResult may eagerly prepare the first fallback step and update // latestMessages. Clear stale source-step messages before starting it so a @@ -4269,6 +4366,17 @@ export class StreamManager extends EventEmitter { return completionTimestamp > afterTimestamp; } + + // A queued tool's execute() can begin after the reconnect cursor while the + // part's own timestamp (model emission) is older. Replay the part so the + // enriched tool-call-start carries executionStartedAt and the renderer can + // start the elapsed timer (the aggregator merges it into the existing row). + if ( + part.executionStartedAt !== undefined && + part.executionStartedAt > afterTimestamp + ) { + return true; + } } return false; diff --git a/src/node/services/tools/withSequentialExecution.test.ts b/src/node/services/tools/withSequentialExecution.test.ts index 6e1c8e52d0..12cefcf018 100644 --- a/src/node/services/tools/withSequentialExecution.test.ts +++ b/src/node/services/tools/withSequentialExecution.test.ts @@ -123,6 +123,51 @@ describe("withSequentialExecution", () => { expect(executionLog).toEqual(["start A", "end A", "start B", "end B", "start C", "end C"]); }); + test("reports execution start only after the lock is acquired", async () => { + const events: string[] = []; + const startedA = createDeferred(); + const releaseA = createDeferred(); + + const tools = { + a: tool({ + description: "Tool A", + inputSchema: z.object({}), + execute: async () => { + events.push("run A"); + startedA.resolve(); + await releaseA.promise; + return { tool: "A" }; + }, + }), + b: tool({ + description: "Tool B", + inputSchema: z.object({}), + execute: () => { + events.push("run B"); + return Promise.resolve({ tool: "B" }); + }, + }), + }; + + const wrappedTools = withSequentialExecution(tools, (toolCallId) => { + events.push(`execution-start ${toolCallId}`); + }); + + const resultsPromise = Promise.all([ + callWrappedExecute(wrappedTools!.a as Record, {}, { toolCallId: "call-a" }), + callWrappedExecute(wrappedTools!.b as Record, {}, { toolCallId: "call-b" }), + ]); + + await startedA.promise; + await Promise.resolve(); + // B is queued behind A: its execution start must not have been reported yet. + expect(events).toEqual(["execution-start call-a", "run A"]); + + releaseA.resolve(); + await resultsPromise; + expect(events).toEqual(["execution-start call-a", "run A", "execution-start call-b", "run B"]); + }); + test("does not execute queued siblings after stream abort", async () => { const executionLog: string[] = []; const startedA = createDeferred(); diff --git a/src/node/services/tools/withSequentialExecution.ts b/src/node/services/tools/withSequentialExecution.ts index 00f03c64c0..d6bea6df83 100644 --- a/src/node/services/tools/withSequentialExecution.ts +++ b/src/node/services/tools/withSequentialExecution.ts @@ -14,6 +14,7 @@ interface ParallelTaskArgs { interface ToolExecutionContext { abortSignal?: AbortSignal; + toolCallId?: string; } function getAbortSignal(options: unknown): AbortSignal | undefined { @@ -25,6 +26,15 @@ function getAbortSignal(options: unknown): AbortSignal | undefined { return context.abortSignal; } +function getToolCallId(options: unknown): string | undefined { + if (typeof options !== "object" || options === null) { + return undefined; + } + + const { toolCallId } = options as ToolExecutionContext; + return typeof toolCallId === "string" ? toolCallId : undefined; +} + function getRequestedAgentId(args: unknown): string | undefined { if (typeof args !== "object" || args === null) { return undefined; @@ -198,9 +208,14 @@ class SharedExecutionLock { * provider's parallel-tool-call planning behavior. Built-in forked explore * tasks share the read side so they can overlap with each other, while every * other tool call stays exclusive. + * + * `onExecutionStart` fires right after the execution lock is acquired (i.e. + * when the tool actually starts running, not when the model emitted the call), + * so queued siblings don't count wait time as execution time. */ export function withSequentialExecution( - tools: Record | undefined + tools: Record | undefined, + onExecutionStart?: (toolCallId: string) => void ): Record | undefined { if (!tools) { return tools; @@ -231,6 +246,10 @@ export function withSequentialExecution( await using _lock = canRunWithSiblingExploreTasks(baseTool, args) ? await executionLock.acquireRead(abortSignal) : await executionLock.acquireWrite(abortSignal); + const toolCallId = getToolCallId(options); + if (onExecutionStart && toolCallId !== undefined) { + onExecutionStart(toolCallId); + } return await executeFn.call(baseTool, args, options); };