Add tool streaming

agents-run-api/src/__tests__/routes/chat/dataChat.test.ts

@@ -1,5 +1,7 @@
 import { generateId } from '@inkeep/agents-core';
 import { describe, expect, it, vi } from 'vitest';
+import { pendingToolApprovalManager } from '../../../services/PendingToolApprovalManager';
+import { toolApprovalUiBus } from '../../../services/ToolApprovalUiBus';
 
 // Logger mock is now in setup.ts globally
 
@@ -13,6 +15,23 @@ vi.mock('../../../handlers/executionHandler', () => {
           await args.sseHelper.writeRole();
           await args.sseHelper.writeContent('[{"type":"text", "text":"Test response from agent"}]');
         }
+
+        // Allow tests to simulate delegated approval UI propagation by publishing to the bus.
+        if (args.userMessage === '__trigger_approval_ui_bus__') {
+          await toolApprovalUiBus.publish(args.requestId, {
+            type: 'approval-needed',
+            toolCallId: 'call_bus_1',
+            toolName: 'delete_file',
+            input: { filePath: 'user/readme.md' },
+            providerMetadata: { openai: { itemId: 'fc_test' } },
+            approvalId: 'aitxt-call_bus_1',
+          });
+          await toolApprovalUiBus.publish(args.requestId, {
+            type: 'approval-resolved',
+            toolCallId: 'call_bus_1',
+            approved: true,
+          });
+        }
         return { success: true, iterations: 1 };
       }),
     })),
@@ -118,6 +137,7 @@ vi.mock('@inkeep/agents-core', async (importOriginal) => {
       })
     ),
     setActiveAgentForConversation: vi.fn().mockReturnValue(vi.fn().mockResolvedValue(undefined)),
+    getConversation: vi.fn().mockReturnValue(vi.fn().mockResolvedValue({ id: 'conv-123' })),
   };
 });
 
@@ -187,4 +207,118 @@ describe('Chat Data Stream Route', () => {
     expect(text).toMatch(/Test/);
     expect(text).toMatch(/response/);
   });
+
+  it('should stream approval UI events published to ToolApprovalUiBus (simulating delegated agent approval)', async () => {
+    // Ensure deterministic requestId inside route subscription (chatds-${Date.now()})
+    const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(12345);
+
+    const body = {
+      conversationId: 'conv-123',
+      messages: [
+        {
+          role: 'user',
+          content: '__trigger_approval_ui_bus__',
+        },
+      ],
+    };
+
+    const res = await makeRequest('/api/chat', {
+      method: 'POST',
+      body: JSON.stringify(body),
+    });
+
+    expect(res.status).toBe(200);
+    expect(res.headers.get('x-vercel-ai-data-stream')).toBe('v2');
+
+    const text = await res.text();
+    expect(text).toMatch(/"type":"tool-input-start"/);
+    expect(text).toMatch(/"type":"tool-approval-request"/);
+    expect(text).toMatch(/"type":"tool-output-available"/);
+
+    nowSpy.mockRestore();
+  });
+
+  it('should accept approval responded tool part via the same /api/chat endpoint and return JSON ack', async () => {
+    const toolCallId = 'call_test_approval_1';
+    const conversationId = 'conv-123';
+
+    // Create a pending approval first
+    const approvalPromise = pendingToolApprovalManager.waitForApproval(
+      toolCallId,
+      'delete_file',
+      { filePath: 'user/readme.md' },
+      conversationId,
+      'test-agent'
+    );
+
+    const body = {
+      conversationId,
+      messages: [
+        {
+          role: 'assistant',
+          content: null,
+          parts: [
+            { type: 'step-start' },
+            {
+              type: 'tool-delete_file',
+              toolCallId,
+              state: 'approval-responded',
+              input: { filePath: '/tmp/test.txt' },
+              callProviderMetadata: { openai: { itemId: 'fc_test' } },
+              approval: { id: `aitxt-${toolCallId}`, approved: true },
+            },
+          ],
+        },
+      ],
+    };
+
+    const res = await makeRequest('/api/chat', {
+      method: 'POST',
+      body: JSON.stringify(body),
+    });
+
+    expect(res.status).toBe(200);
+    expect(res.headers.get('content-type') || '').toMatch(/application\/json/);
+    const json = await res.json();
+    expect(json).toMatchObject({ success: true, toolCallId, approved: true });
+
+    await expect(approvalPromise).resolves.toMatchObject({ approved: true });
+  });
+
+  it('should treat approval responded tool part for unknown toolCallId as alreadyProcessed (idempotent 200)', async () => {
+    const toolCallId = 'call_test_approval_missing';
+    const conversationId = 'conv-123';
+
+    const body = {
+      conversationId,
+      messages: [
+        {
+          role: 'assistant',
+          content: null,
+          parts: [
+            {
+              type: 'tool-delete_file',
+              toolCallId,
+              state: 'approval-responded',
+              approval: { id: `aitxt-${toolCallId}`, approved: true },
+            },
+          ],
+        },
+      ],
+    };
+
+    const res = await makeRequest('/api/chat', {
+      method: 'POST',
+      body: JSON.stringify(body),
+    });
+
+    expect(res.status).toBe(200);
+    const json = await res.json();
+    expect(json).toMatchObject({
+      success: true,
+      toolCallId,
+      approved: true,
+      alreadyProcessed: true,
+    });
+  });
 });

agents-run-api/src/__tests__/utils/tool-streaming.test.ts

@@ -0,0 +1,138 @@
+import { describe, expect, it, vi } from 'vitest';
+import { createSSEStreamHelper, createVercelStreamHelper } from '../../utils/stream-helpers';
+
+describe('tool streaming', () => {
+  describe('SSEStreamHelper (chat completions)', () => {
+    it('streams tool_calls deltas and tool output envelopes', async () => {
+      const sent: string[] = [];
+
+      const stream = {
+        writeSSE: vi.fn(async ({ data }: { data: string }) => {
+          sent.push(data);
+        }),
+        sleep: vi.fn(async () => {}),
+      };
+
+      const helper = createSSEStreamHelper(stream as any, 'req_123', Math.floor(Date.now() / 1000));
+
+      await helper.writeToolInputStart({ toolCallId: 'call_1234xyz', toolName: 'get_weather' });
+      expect(sent).toHaveLength(1);
+      const first = JSON.parse(sent[0]);
+      expect(first.object).toBe('chat.completion.chunk');
+      expect(first.choices[0].delta.tool_calls).toEqual([
+        {
+          index: 0,
+          id: 'call_1234xyz',
+          type: 'function',
+          function: { name: 'get_weather', arguments: '' },
+        },
+      ]);
+
+      await helper.writeToolInputDelta({ toolCallId: 'call_1234xyz', inputTextDelta: '{"' });
+      const second = JSON.parse(sent[1]);
+      expect(second.choices[0].delta.tool_calls).toEqual([
+        {
+          index: 0,
+          id: null,
+          type: null,
+          function: { name: null, arguments: '{"' },
+        },
+      ]);
+
+      await helper.writeToolOutputAvailable({ toolCallId: 'call_1234xyz', output: { ok: true } });
+      const third = JSON.parse(sent[2]);
+      expect(third.choices[0].delta.content).toBeTypeOf('string');
+      expect(JSON.parse(third.choices[0].delta.content)).toEqual({
+        type: 'tool-output-available',
+        toolCallId: 'call_1234xyz',
+        output: { ok: true },
+      });
+
+      await helper.writeToolOutputError({ toolCallId: 'call_1234xyz', error: 'boom' });
+      const fourth = JSON.parse(sent[3]);
+      expect(JSON.parse(fourth.choices[0].delta.content)).toEqual({
+        type: 'tool-output-error',
+        toolCallId: 'call_1234xyz',
+        error: 'boom',
+        output: null,
+      });
+
+      await helper.writeToolApprovalRequest({ approvalId: 'aitxt-call_1234xyz', toolCallId: 'call_1234xyz' });
+      const fifth = JSON.parse(sent[4]);
+      expect(JSON.parse(fifth.choices[0].delta.content)).toEqual({
+        type: 'tool-approval-request',
+        approvalId: 'aitxt-call_1234xyz',
+        toolCallId: 'call_1234xyz',
+      });
+
+      await helper.writeToolOutputDenied({ toolCallId: 'call_1234xyz' });
+      const sixth = JSON.parse(sent[5]);
+      expect(JSON.parse(sixth.choices[0].delta.content)).toEqual({
+        type: 'tool-output-denied',
+        toolCallId: 'call_1234xyz',
+      });
+    });
+  });
+
+  describe('VercelDataStreamHelper (AI SDK UI stream)', () => {
+    it('writes tool-input/tool-output parts as top-level chunks', async () => {
+      const writer = {
+        write: vi.fn(),
+        merge: vi.fn(),
+        onError: vi.fn(),
+      };
+
+      const helper = createVercelStreamHelper(writer as any);
+
+      await helper.writeToolInputStart({ toolCallId: 'call_1', toolName: 'delete_file' });
+      await helper.writeToolInputDelta({ toolCallId: 'call_1', inputTextDelta: '{"' });
+      await helper.writeToolInputAvailable({
+        toolCallId: 'call_1',
+        toolName: 'delete_file',
+        input: { filePath: 'user/none.md' },
+      });
+      await helper.writeToolOutputAvailable({ toolCallId: 'call_1', output: { success: true } });
+      await helper.writeToolOutputError({ toolCallId: 'call_1', error: 'nope' });
+      await helper.writeToolApprovalRequest({ approvalId: 'aitxt-call_1', toolCallId: 'call_1' });
+      await helper.writeToolOutputDenied({ toolCallId: 'call_1' });
+
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-input-start',
+        toolCallId: 'call_1',
+        toolName: 'delete_file',
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-input-delta',
+        toolCallId: 'call_1',
+        inputTextDelta: '{"',
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-input-available',
+        toolCallId: 'call_1',
+        toolName: 'delete_file',
+        input: { filePath: 'user/none.md' },
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-output-available',
+        toolCallId: 'call_1',
+        output: { success: true },
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-output-error',
+        toolCallId: 'call_1',
+        error: 'nope',
+        output: null,
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-approval-request',
+        approvalId: 'aitxt-call_1',
+        toolCallId: 'call_1',
+      });
+      expect(writer.write).toHaveBeenCalledWith({
+        type: 'tool-output-denied',
+        toolCallId: 'call_1',
+      });
+    });
+  });
+});
+