fix(flags): a2a streaming

src/cli/aws/__tests__/agentcore-a2a-bearer.test.ts

@@ -1,4 +1,4 @@
-import { invokeA2ARuntime } from '../agentcore.js';
+import { invokeA2ARuntime, invokeA2ARuntimeStreaming } from '../agentcore.js';
 import type { A2AInvokeOptions } from '../agentcore.js';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
@@ -95,3 +95,158 @@ describe('invokeA2ARuntime bearer-token auth path', () => {
     expect(fetchSpy).not.toHaveBeenCalled();
   });
 });
+
+// ---------------------------------------------------------------------------
+// Shared helpers for streaming tests
+// ---------------------------------------------------------------------------
+
+function makeReaderMock(frames: string[]): {
+  getReader: () => {
+    read: () => Promise<{ done: boolean; value: Uint8Array | undefined }>;
+    releaseLock: ReturnType<typeof vi.fn>;
+  };
+} {
+  const encoder = new TextEncoder();
+  let i = 0;
+  return {
+    getReader: () => ({
+      read: () => {
+        if (i < frames.length) {
+          return Promise.resolve({ done: false as const, value: encoder.encode(frames[i++]) });
+        }
+        return Promise.resolve({ done: true as const, value: undefined });
+      },
+      releaseLock: vi.fn(),
+    }),
+  };
+}
+
+describe('invokeA2ARuntimeStreaming SigV4', () => {
+  afterEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it('yields incremental chunks from status-update SSE events', async () => {
+    const frames = [
+      'data: {"kind":"status-update","status":{"state":"working","message":{"parts":[{"kind":"text","text":"Hello "}]}}}\n\n',
+      'data: {"kind":"status-update","status":{"state":"working","message":{"parts":[{"kind":"text","text":"world"}]}}}\n\n',
+    ];
+    const stream = makeReaderMock(frames);
+    mockSdkSend.mockResolvedValue({
+      runtimeSessionId: 'sigv4-stream-session',
+      response: { transformToWebStream: () => stream },
+    });
+
+    const result = await invokeA2ARuntimeStreaming(baseOpts, 'hello');
+    const text = await drain(result.stream);
+
+    expect(mockSdkSend).toHaveBeenCalledTimes(1);
+    expect(text).toBe('Hello world');
+    expect(result.sessionId).toBe('sigv4-stream-session');
+  });
+
+  it('yields artifact-update text with type:"text" parts (backward compat) when no status-update text', async () => {
+    const frames = [
+      'data: {"kind":"artifact-update","artifact":{"parts":[{"type":"text","text":"legacy output"}]}}\n\n',
+    ];
+    const stream = makeReaderMock(frames);
+    mockSdkSend.mockResolvedValue({
+      runtimeSessionId: 'sigv4-stream-session-2',
+      response: { transformToWebStream: () => stream },
+    });
+
+    const result = await invokeA2ARuntimeStreaming(baseOpts, 'hello');
+    const text = await drain(result.stream);
+
+    expect(text).toBe('legacy output');
+  });
+
+  it('suppresses artifact-update when status-update already emitted text', async () => {
+    const frames = [
+      'data: {"kind":"status-update","status":{"state":"working","message":{"parts":[{"kind":"text","text":"status text"}]}}}\n\n',
+      'data: {"kind":"artifact-update","artifact":{"parts":[{"kind":"text","text":"should be skipped"}]}}\n\n',
+    ];
+    const stream = makeReaderMock(frames);
+    mockSdkSend.mockResolvedValue({
+      runtimeSessionId: 'sigv4-stream-session-3',
+      response: { transformToWebStream: () => stream },
+    });
+
+    const result = await invokeA2ARuntimeStreaming(baseOpts, 'hello');
+    const text = await drain(result.stream);
+
+    expect(text).toBe('status text');
+  });
+
+  it('yields trailing SSE data line without final newline', async () => {
+    const frames = ['data: {"kind":"status-update","status":{"state":"working","message":{"parts":[{"kind":"text","text":"tail chunk"}]}}}'];
+    const stream = makeReaderMock(frames);
+    mockSdkSend.mockResolvedValue({
+      runtimeSessionId: 'sigv4-stream-session-4',
+      response: { transformToWebStream: () => stream },
+    });
+
+    const result = await invokeA2ARuntimeStreaming(baseOpts, 'hello');
+    const text = await drain(result.stream);
+
+    expect(text).toBe('tail chunk');
+  });
+});
+
+describe('invokeA2ARuntimeStreaming bearer-token', () => {
+  let fetchSpy: { mockRestore: () => void } | undefined;
+  let capturedRequests: { url: string; init: RequestInit }[];
+
+  afterEach(() => {
+    fetchSpy?.mockRestore();
+    vi.clearAllMocks();
+  });
+
+  function mockFetchStream(frames: string[], sessionId: string | null = 'bearer-stream-session'): void {
+    capturedRequests = [];
+    const stream = makeReaderMock(frames);
+    fetchSpy = vi.spyOn(globalThis, 'fetch').mockImplementation((input, init) => {
+      capturedRequests.push({ url: input as string, init: init! });
+      return Promise.resolve({
+        ok: true,
+        status: 200,
+        body: stream,
+        headers: {
+          get: (h: string) => (h === 'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id' ? sessionId : null),
+        },
+      } as unknown as Response);
+    });
+  }
+
+  it('uses message/stream method with Bearer token and yields SSE chunks', async () => {
+    mockFetchStream([
+      'data: {"kind":"status-update","status":{"state":"working","message":{"parts":[{"kind":"text","text":"streamed result"}]}}}\n\n',
+    ]);
+
+    const result = await invokeA2ARuntimeStreaming({ ...baseOpts, bearerToken: 'my-jwt' }, 'hello');
+    const text = await drain(result.stream);
+
+    expect(fetchSpy).toHaveBeenCalledTimes(1);
+    expect(mockSdkSend).not.toHaveBeenCalled();
+
+    const headers = capturedRequests[0]!.init.headers as Record<string, string>;
+    expect(headers.Authorization).toBe('Bearer my-jwt');
+
+    const reqBody = JSON.parse(capturedRequests[0]!.init.body as string);
+    expect(reqBody.method).toBe('message/stream');
+
+    expect(text).toBe('streamed result');
+    expect(result.sessionId).toBe('bearer-stream-session');
+  });
+
+  it('yields artifact-update text with type:"text" parts (backward compat)', async () => {
+    mockFetchStream([
+      'data: {"kind":"artifact-update","artifact":{"parts":[{"type":"text","text":"legacy bearer output"}]}}\n\n',
+    ]);
+
+    const result = await invokeA2ARuntimeStreaming({ ...baseOpts, bearerToken: 'my-jwt' }, 'hello');
+    const text = await drain(result.stream);
+
+    expect(text).toBe('legacy bearer output');
+  });
+});

src/cli/aws/agentcore.ts

@@ -960,7 +960,6 @@ export async function invokeA2ARuntime(options: A2AInvokeOptions, message: strin
   if (options.bearerToken) {
     const url = buildInvokeUrl(options.region, options.runtimeArn);
     const headers = buildBearerInvokeHeaders(options, 'application/json, text/event-stream');
-
     const res = await fetch(url, { method: 'POST', headers, body: JSON.stringify(body) });
     if (!res.ok) {
       const errBody = await res.text().catch(() => '');
@@ -1006,6 +1005,147 @@ export async function invokeA2ARuntime(options: A2AInvokeOptions, message: strin
   };
 }
 
+/**
+ * Invoke a deployed A2A agent via InvokeAgentRuntime with JSON-RPC message/stream.
+ * Yields text parts incrementally from SSE events.
+ */
+export async function invokeA2ARuntimeStreaming(
+  options: A2AInvokeOptions,
+  message: string
+): Promise<StreamingInvokeResult> {
+  const body = {
+    jsonrpc: '2.0',
+    id: a2aRequestId++,
+    method: 'message/stream',
+    params: {
+      message: {
+        role: 'user',
+        parts: [{ kind: 'text', text: message }],
+        messageId: `msg-${Date.now()}`,
+      },
+    },
+  };
+
+  options.logger?.logSSEEvent(`A2A streaming request: ${JSON.stringify(body)}`);
+
+  async function* a2aStreamGenerator(
+    reader: ReadableStreamDefaultReader<Uint8Array>
+  ): AsyncGenerator<string, void, unknown> {
+    const decoder = new TextDecoder();
+    let buffer = '';
+    let streamedFromStatus = false;
+
+    function extractData(line: string): string {
+      if (!line.startsWith('data: ')) return '';
+      return line.slice(6).trim();
+    }
+
+    function collectTextParts(
+      parts?: { kind?: string; type?: string; text?: string }[]
+    ): string {
+      if (!parts) return '';
+      return parts
+        .filter(p => (p.kind === 'text' || p.type === 'text') && p.text)
+        .map(p => p.text!)
+        .join('');
+    }
+
+    function* parseDataLine(line: string): Generator<string, void, unknown> {
+      const data = extractData(line);
+      if (!data) return;
+
+      options.logger?.logSSEEvent(line);
+
+      try {
+        const event = JSON.parse(data) as Record<string, unknown>;
+        // Unwrap JSON-RPC result envelope if present
+        const target = (event.result as Record<string, unknown>) ?? event;
+        const kind = target.kind as string | undefined;
+
+        if (kind === 'status-update') {
+          const status = target.status as
+            | { state?: string; message?: { parts?: { kind?: string; type?: string; text?: string }[] } }
+            | undefined;
+          const text = collectTextParts(status?.message?.parts);
+          if (text) {
+            streamedFromStatus = true;
+            yield text;
+          }
+        } else if (kind === 'artifact-update' && !streamedFromStatus) {
+          const artifact = target.artifact as
+            | { parts?: { kind?: string; type?: string; text?: string }[] }
+            | undefined;
+          const text = collectTextParts(artifact?.parts);
+          if (text) yield text;
+        }
+      } catch {
+        // Non-JSON SSE line, skip
+      }
+    }
+
+    try {
+      while (true) {
+        const result = await reader.read();
+        if (result.done) break;
+
+        buffer += decoder.decode(result.value, { stream: true });
+        const lines = buffer.split('\n');
+        buffer = lines.pop() ?? '';
+
+        for (const line of lines) {
+          yield* parseDataLine(line);
+        }
+      }
+
+      // Flush any buffered decoder state so split multi-byte characters are preserved.
+      const trailingDecoded = decoder.decode();
+      if (trailingDecoded) {
+        buffer += trailingDecoded;
+      }
+
+      if (buffer) {
+        yield* parseDataLine(buffer);
+      }
+    } finally {
+      reader.releaseLock();
+    }
+  }
+
+  if (options.bearerToken) {
+    const url = buildInvokeUrl(options.region, options.runtimeArn);
+    const streamHeaders = buildBearerInvokeHeaders(options, 'text/event-stream');
+    const res = await fetch(url, { method: 'POST', headers: streamHeaders, body: JSON.stringify(body) });
+    if (!res.ok) {
+      const errBody = await res.text().catch(() => '');
+      throw new Error(`Invoke failed (${res.status}): ${errBody || res.statusText}`);
+    }
+    if (!res.body) throw new Error('No response body for A2A streaming');
+    const sessionId = res.headers.get('X-Amzn-Bedrock-AgentCore-Runtime-Session-Id') ?? undefined;
+    return { stream: a2aStreamGenerator(res.body.getReader()), sessionId };
+  }
+
+  const client = createAgentCoreClient(options.region, options.headers);
+
+  const streamCommand = new InvokeAgentRuntimeCommand({
+    agentRuntimeArn: options.runtimeArn,
+    payload: new TextEncoder().encode(JSON.stringify(body)),
+    contentType: 'application/json',
+    accept: 'text/event-stream',
+    runtimeUserId: options.userId ?? DEFAULT_RUNTIME_USER_ID,
+    ...(options.sessionId && { runtimeSessionId: options.sessionId }),
+  });
+
+  const streamResponse = await client.send(streamCommand);
+  const sessionId = streamResponse.runtimeSessionId;
+
+  if (!streamResponse.response) {
+    throw new Error('No response from AgentCore Runtime');
+  }
+
+  const reader = streamResponse.response.transformToWebStream().getReader();
+  return { stream: a2aStreamGenerator(reader), sessionId };
+}
+
 /** Wrap a single string value as an AsyncGenerator for StreamingInvokeResult compatibility. */
 // eslint-disable-next-line @typescript-eslint/require-await
 async function* singleValueStream(value: string): AsyncGenerator<string, void, unknown> {

src/cli/aws/index.ts

@@ -55,6 +55,7 @@ export {
   DEFAULT_RUNTIME_USER_ID,
   executeBashCommand,
   invokeA2ARuntime,
+  invokeA2ARuntimeStreaming,
   invokeAgentRuntime,
   invokeAgentRuntimeStreaming,
   mcpInitSession,