⏱️ feat: Expose Tool Preparation and Dispatch Timing

src/__tests__/stream.eagerEventExecution.test.ts

@@ -180,12 +180,20 @@ describe('ChatModelStreamHandler eager event tool execution', () => {
   it('prestarts a complete event-driven tool call from the stream', async () => {
     const graph = createGraph();
     const toolExecuteCalls: t.ToolExecuteBatchRequest[] = [];
+    const dispatches: t.ToolCallsDispatchedEvent[] = [];
+    const order: string[] = [];
     jest
       .spyOn(events, 'safeDispatchCustomEvent')
       .mockImplementation(async (event, data): Promise<void> => {
+        if (event === GraphEvents.ON_TOOL_CALLS_DISPATCHED) {
+          dispatches.push(data as t.ToolCallsDispatchedEvent);
+          order.push('dispatched');
+          return;
+        }
         if (event !== GraphEvents.ON_TOOL_EXECUTE) {
           return;
         }
+        order.push('host');
         const batch = data as t.ToolExecuteBatchRequest;
         toolExecuteCalls.push(batch);
         batch.resolve([
@@ -224,6 +232,17 @@ describe('ChatModelStreamHandler eager event tool execution', () => {
       stepId: expect.stringMatching(/^step_/),
       turn: 0,
     });
+    expect(order).toEqual(['dispatched', 'host']);
+    expect(dispatches).toEqual([
+      expect.objectContaining({
+        dispatched_at: expect.any(Number),
+        toolCalls: [{
+          id: 'call_weather',
+          name: 'weather',
+          stepId: expect.stringMatching(/^step_/),
+        }],
+      }),
+    ]);
     expect(toolExecuteCalls[0].callerCapabilityProjection).toEqual({
       version: 1,
       directToolNames: ['weather'],
@@ -239,6 +258,126 @@ describe('ChatModelStreamHandler eager event tool execution', () => {
     expect(graph.toolCallStepIds.has('call_weather')).toBe(true);
   });
 
+  it('keeps a fragmented tool call in preparation until host dispatch', async () => {
+    let now = 1_000;
+    jest.spyOn(Date, 'now').mockImplementation(() => now);
+    const graph = createGraph();
+    const starts: t.ToolCallsDispatchedEvent[] = [];
+    const completions: t.ToolCompleteEvent[] = [];
+    jest.spyOn(events, 'safeDispatchCustomEvent').mockImplementation(
+      async (event, data): Promise<void> => {
+        if (event === GraphEvents.ON_TOOL_CALLS_DISPATCHED) {
+          starts.push(data as t.ToolCallsDispatchedEvent);
+        }
+        if (event === GraphEvents.ON_TOOL_EXECUTE) {
+          now = 4_240;
+          (data as t.ToolExecuteBatchRequest).resolve([
+            { toolCallId: 'call_weather', content: 'sunny', status: 'success' },
+          ]);
+        }
+        if (event === GraphEvents.ON_RUN_STEP_COMPLETED) {
+          completions.push((data as { result: t.ToolCompleteEvent }).result);
+        }
+      }
+    );
+    const handler = new ChatModelStreamHandler();
+    const metadata = { langgraph_node: 'agent' };
+
+    await handler.handle(
+      GraphEvents.CHAT_MODEL_STREAM,
+      {
+        chunk: {
+          content: '',
+          tool_call_chunks: [{ id: 'call_weather', name: 'weather', args: '{"city":', index: 0 }],
+        } as unknown as t.StreamChunk,
+      },
+      metadata,
+      graph
+    );
+    const stepId = graph.toolCallStepIds.get('call_weather');
+    expect(stepId).toBeDefined();
+    expect(graph.getRunStep(stepId!)?.type).toBe(StepTypes.TOOL_CALLS);
+    expect(graph.dispatchRunStepDelta).toHaveBeenCalledWith(
+      stepId!,
+      expect.objectContaining({ type: StepTypes.TOOL_CALLS }),
+      metadata,
+      1_000
+    );
+    expect(starts).toHaveLength(0);
+
+    now = 4_200;
+    await handler.handle(
+      GraphEvents.CHAT_MODEL_STREAM,
+      {
+        chunk: {
+          content: '',
+          tool_calls: [{ id: 'call_weather', name: 'weather', args: { city: 'NYC' } }],
+          response_metadata: finalToolCallResponseMetadata,
+        } as unknown as t.StreamChunk,
+      },
+      metadata,
+      graph
+    );
+    await graph.eagerEventToolExecutions.get('call_weather')?.promise;
+
+    expect(starts).toEqual([{
+      dispatched_at: 4_200,
+      toolCalls: [{ id: 'call_weather', name: 'weather', stepId }],
+    }]);
+    expect(completions).toEqual([
+      expect.objectContaining({
+        completed_at: 4_240,
+        tool_call: expect.objectContaining({ id: 'call_weather' }),
+      }),
+    ]);
+    const firstFragmentAt = (graph.dispatchRunStepDelta as jest.Mock).mock.calls[0][3] as number;
+    expect(starts[0].dispatched_at - firstFragmentAt).toBe(3_200);
+    expect((completions[0].completed_at ?? 0) - starts[0].dispatched_at).toBe(40);
+  });
+
+  it('timestamps a coalesced fragment before an awaited step and eager host dispatch', async () => {
+    let now = 1_000;
+    jest.spyOn(Date, 'now').mockImplementation(() => now);
+    const graph = createGraph();
+    const dispatchRunStep = graph.dispatchRunStep.bind(graph);
+    graph.dispatchRunStep = jest.fn<StandardGraph['dispatchRunStep']>(async (key, details, meta) => {
+      now = 1_900;
+      return dispatchRunStep(key, details, meta);
+    });
+    const dispatches: t.ToolCallsDispatchedEvent[] = [];
+    jest.spyOn(events, 'safeDispatchCustomEvent').mockImplementation(
+      async (event, data): Promise<void> => {
+        if (event === GraphEvents.ON_TOOL_CALLS_DISPATCHED) {
+          dispatches.push(data as t.ToolCallsDispatchedEvent);
+        }
+        if (event === GraphEvents.ON_TOOL_EXECUTE) {
+          (data as t.ToolExecuteBatchRequest).resolve([
+            { toolCallId: 'call_weather', status: 'success', content: 'sunny' },
+          ]);
+        }
+      }
+    );
+    await new ChatModelStreamHandler().handle(
+      GraphEvents.CHAT_MODEL_STREAM,
+      { chunk: {
+        content: '',
+        tool_calls: [{ id: 'call_weather', name: 'weather', args: { city: 'NYC' } }],
+        tool_call_chunks: [{ id: 'call_weather', name: 'weather', args: '{"city":"NYC"}', index: 0 }],
+        response_metadata: finalToolCallResponseMetadata,
+      } as unknown as t.StreamChunk },
+      { langgraph_node: 'agent' },
+      graph
+    );
+    expect(dispatches).toHaveLength(1);
+    expect(dispatches[0].dispatched_at).toBe(1_900);
+    expect(graph.dispatchRunStepDelta).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.objectContaining({ type: StepTypes.TOOL_CALLS }),
+      { langgraph_node: 'agent' },
+      1_000
+    );
+  });
+
   it('captures normalized code-session identity before eager dispatch', async () => {
     const graph = createGraph({
       sessions: new Map([

src/common/enum.ts

@@ -26,6 +26,8 @@ export enum GraphEvents {
   ON_MESSAGE_DELTA = 'on_message_delta',
   /** [Custom] Reasoning Delta events for messages */
   ON_REASONING_DELTA = 'on_reasoning_delta',
+  /** [Custom] Per-call handoff boundary between streamed tool preparation and execution */
+  ON_TOOL_CALLS_DISPATCHED = 'on_tool_calls_dispatched',
   /** [Custom] Request to execute tools - dispatched by ToolNode, handled by host */
   ON_TOOL_EXECUTE = 'on_tool_execute',
   /** [Custom] Emitted when the summarize node begins generating a summary */

src/graphs/Graph.ts

@@ -766,7 +766,8 @@ export abstract class Graph<
   abstract dispatchRunStepDelta(
     id: string,
     delta: t.ToolCallDelta,
-    metadata?: Record<string, unknown>
+    metadata?: Record<string, unknown>,
+    observedAt?: number
   ): Promise<void>;
   abstract dispatchMessageDelta(
     id: string,
@@ -6130,7 +6131,8 @@ export class StandardGraph extends Graph<t.BaseGraphState, t.GraphNode> {
   async dispatchRunStepDelta(
     id: string,
     delta: t.ToolCallDelta,
-    metadata?: Record<string, unknown>
+    metadata?: Record<string, unknown>,
+    observedAt?: number
   ): Promise<void> {
     if (!this.config) {
       throw new Error('No config provided');
@@ -6140,6 +6142,7 @@ export class StandardGraph extends Graph<t.BaseGraphState, t.GraphNode> {
     const runStepDelta: t.RunStepDeltaEvent = {
       id,
       delta,
+      ...(observedAt != null && { observed_at: observedAt }),
     };
     const handler = this.handlerRegistry?.getHandler(
       GraphEvents.ON_RUN_STEP_DELTA

src/graphs/__tests__/Graph.closeRunStep.test.ts

@@ -1,6 +1,6 @@
 // src/graphs/__tests__/Graph.closeRunStep.test.ts
-import { BaseCallbackHandler } from '@langchain/core/callbacks/base';
 import { CallbackManager } from '@langchain/core/callbacks/manager';
+import { BaseCallbackHandler } from '@langchain/core/callbacks/base';
 import type * as t from '@/types';
 import { GraphEvents, StepTypes, Providers } from '@/common';
 import { HandlerRegistry } from '@/events';
@@ -59,6 +59,29 @@ function seedStep(
   return step;
 }
 
+describe('StandardGraph.dispatchRunStepDelta', () => {
+  it('delivers the producer receipt time rather than timing a slow consumer', async () => {
+    const { graph } = createGraph();
+    const received: t.RunStepDeltaEvent[] = [];
+    graph.config = { configurable: { run_id: 'run_1' } };
+    graph.handlerRegistry?.register(GraphEvents.ON_RUN_STEP_DELTA, {
+      handle: async (_event, data): Promise<void> => {
+        await new Promise((resolve) => setTimeout(resolve, 5));
+        received.push(data as t.RunStepDeltaEvent);
+      },
+    });
+    await graph.dispatchRunStepDelta(
+      'step_tool',
+      { type: StepTypes.TOOL_CALLS, tool_calls: [{ index: 0, args: '{', type: 'tool_call_chunk' }] },
+      undefined,
+      1_000
+    );
+    expect(received).toEqual([
+      { id: 'step_tool', delta: expect.objectContaining({ type: StepTypes.TOOL_CALLS }), observed_at: 1_000 },
+    ]);
+  });
+});
+
 describe('StandardGraph.closeRunStep', () => {
   it('stamps the terminal status + timestamp and emits ON_RUN_STEP_CLOSED once', async () => {
     const { graph, closed } = createGraph();

src/run.ts

@@ -37,13 +37,6 @@ import {
   DEFAULT_MAX_STOP_CONTINUATIONS,
   DEFAULT_RECURSION_LIMIT,
 } from '@/common';
-import {
-  ACTIVITY_PHASE_LABEL_PROMPT,
-  ACTIVITY_LABEL_PROMPT,
-  buildActivityLabelPrompt,
-  buildActivityPhaseLabelPrompt,
-  normalizeActivityPhaseLabel,
-} from '@/prompts/activityLabel';
 import {
   createLangfuseTraceMetadata,
   createLangfuseHandler,
@@ -53,6 +46,13 @@ import {
   isLangfuseCallbackHandler,
   withLangfuseAttributes,
 } from '@/langfuse';
+import {
+  ACTIVITY_PHASE_LABEL_PROMPT,
+  ACTIVITY_LABEL_PROMPT,
+  buildActivityLabelPrompt,
+  buildActivityPhaseLabelPrompt,
+  normalizeActivityPhaseLabel,
+} from '@/prompts/activityLabel';
 import {
   requireValidSubagentResumeManifest,
   SUBAGENT_RESUME_ATTEMPT_CONFIG_KEY,
@@ -204,6 +204,7 @@ const CUSTOM_GRAPH_EVENTS = new Set<string>([
   GraphEvents.ON_RUN_STEP_DELTA,
   GraphEvents.ON_RUN_STEP_COMPLETED,
   GraphEvents.ON_RUN_STEP_CLOSED,
+  GraphEvents.ON_TOOL_CALLS_DISPATCHED,
   GraphEvents.ON_MESSAGE_DELTA,
   GraphEvents.ON_REASONING_DELTA,
   GraphEvents.ON_TOOL_EXECUTE,
@@ -1095,6 +1096,7 @@ export class Run<_T extends t.BaseGraphState> {
               | t.RunStep
               | t.RunStepDeltaEvent
               | t.RunStepClosedEvent
+              | t.ToolCallsDispatchedEvent
               | t.MessageDeltaEvent
               | t.ReasoningDeltaEvent
               | { result: t.ToolEndEvent },

src/stream.ts

@@ -55,13 +55,13 @@ import {
   calculateMaxToolResultChars,
   truncateToolResultContent,
 } from '@/utils/truncation';
+import { createToolCallsDispatchedEvent, safeDispatchCustomEvent } from '@/utils/events';
 import { resolveToolOutcome, outcomeFieldsFromResult } from '@/tools/intentArg';
-import { formatToolErrorContent } from '@/tools/toolErrorContent';
 import { snapshotValidatedModelChunk } from '@/graphs/acceptedModelResponse';
 import { TOOL_OUTPUT_REF_PATTERN } from '@/tools/toolOutputReferences';
+import { formatToolErrorContent } from '@/tools/toolErrorContent';
 import { PreparedSubagentError } from '@/tools/preparedSubagents';
 import { isReasoningContentBlock } from '@/messages/core';
-import { safeDispatchCustomEvent } from '@/utils/events';
 import { composeAbortSignals } from '@/utils/misc';
 import { isGoogleLike } from '@/utils/llm';
 import { getMessageId } from '@/messages';
@@ -871,6 +871,13 @@ function startEagerToolExecutions(args: {
       reject,
     };
 
+    if (graph.config != null) {
+      void safeDispatchCustomEvent(
+        GraphEvents.ON_TOOL_CALLS_DISPATCHED,
+        createToolCallsDispatchedEvent(graph.config, batchRequest.toolCalls),
+        graph.config
+      );
+    }
     void safeDispatchCustomEvent(
       GraphEvents.ON_TOOL_EXECUTE,
       batchRequest,
@@ -980,7 +987,7 @@ async function dispatchEagerToolCompletions(args: {
               progress: 1,
               ...(outcome != null && { outcome }),
             } as t.ProcessedToolCall,
-            completed_at: Date.now(),
+            completed_at: result.received_at ?? Date.now(),
           },
         },
         graph.config
@@ -1701,6 +1708,7 @@ export class ChatModelStreamHandler implements t.EventHandler {
     // Callback delivery can beat the producer's iterator. Validate before
     // accounting, run steps, or eager dispatch reads raw tool descriptors.
     chunk = snapshotValidatedModelChunk(chunk as AIMessageChunk);
+    const observedAt = Date.now();
 
     /**
      * Enforced before every content-specific early return below
@@ -1923,6 +1931,7 @@ export class ChatModelStreamHandler implements t.EventHandler {
         stepKey,
         toolCallChunks: chunk.tool_call_chunks,
         metadata,
+        observedAt,
       });
       if (canStreamEager) {
         if (runScopeInvalidated()) {