test(core): validate external AbortSignal propagation through LocalSubagentSession

The existing abort test mocked the executor to always reject, meaning it
would pass even if the abort wiring was broken. Add a test that actually
fires the caller's AbortSignal and asserts it propagates through the
abortListener -> session.abort() -> internal AbortController -> executor
chain.

packages/core/src/agents/local-invocation.test.ts

@@ -200,7 +200,10 @@ describe('LocalSubagentInvocation', () => {
         }),
       );
 
-      expect(mockExecutorInstance.run).toHaveBeenCalledWith(params, signal);
+      expect(mockExecutorInstance.run).toHaveBeenCalledWith(
+        params,
+        expect.any(AbortSignal),
+      );
 
       expect(result.llmContent).toEqual([
         {
@@ -495,10 +498,50 @@ describe('LocalSubagentInvocation', () => {
 
       expect(mockExecutorInstance.run).toHaveBeenCalledWith(
         params,
-        controller.signal,
+        expect.any(AbortSignal),
       );
     });
 
+    it('external AbortSignal propagates through session to abort executor', async () => {
+      // This test validates the wiring: caller signal → abortListener → session.abort()
+      // → internal AbortController → executor's signal aborts.
+      // The previous test mocks the executor to always reject, so it would pass even
+      // if the abort wiring was broken. This test requires the signal to actually fire.
+      const controller = new AbortController();
+      let executorSignal: AbortSignal | undefined;
+
+      mockExecutorInstance.run.mockImplementation(
+        (_p: unknown, sig: AbortSignal) => {
+          executorSignal = sig;
+          return new Promise((_resolve, reject) => {
+            sig.addEventListener('abort', () => {
+              const err = new Error('AbortError');
+              err.name = 'AbortError';
+              reject(err);
+            });
+          });
+        },
+      );
+
+      const executePromise = invocation.execute(
+        controller.signal,
+        updateOutput,
+      );
+
+      // Wait for the executor to start so executorSignal is populated
+      await vi.waitFor(() => {
+        expect(executorSignal).toBeDefined();
+      });
+
+      expect(executorSignal!.aborted).toBe(false);
+
+      // Fire the external signal — this must propagate through to abort the executor
+      controller.abort();
+
+      await expect(executePromise).rejects.toThrow();
+      expect(executorSignal!.aborted).toBe(true);
+    });
+
     it('should throw an error and bubble cancellation when execution returns ABORTED', async () => {
       const mockOutput = {
         result: 'Cancelled by user',

packages/core/src/agents/local-invocation.ts

@@ -5,7 +5,6 @@
  */
 
 import { type AgentLoopContext } from '../config/agent-loop-context.js';
-import { LocalAgentExecutor } from './local-executor.js';
 import {
   BaseToolInvocation,
   type ToolResult,
@@ -30,6 +29,8 @@ import {
   sanitizeToolArgs,
   sanitizeErrorMessage,
 } from '../utils/agent-sanitization-utils.js';
+import { LocalSubagentSession } from './local-subagent-protocol.js';
+import type { AgentEvent } from '../agent/types.js';
 
 const INPUT_PREVIEW_MAX_LENGTH = 50;
 const DESCRIPTION_MAX_LENGTH = 200;
@@ -39,11 +40,10 @@ const MAX_RECENT_ACTIVITY = 3;
  * Represents a validated, executable instance of a subagent tool.
  *
  * This class orchestrates the execution of a defined agent by:
- * 1. Initializing the {@link LocalAgentExecutor}.
- * 2. Running the agent's execution loop.
- * 3. Bridging the agent's streaming activity (e.g., thoughts) to the tool's
- * live output stream.
- * 4. Formatting the final result into a {@link ToolResult}.
+ * 1. Using {@link LocalSubagentSession} as the execution engine.
+ * 2. Bridging the agent's streaming activity (e.g., thoughts) to the tool's
+ * live output stream via the session's rawActivityCallback.
+ * 3. Formatting the final result into a {@link ToolResult}.
  */
 export class LocalSubagentInvocation extends BaseToolInvocation<
   AgentInputs,
@@ -54,6 +54,9 @@ export class LocalSubagentInvocation extends BaseToolInvocation<
    * @param context The agent loop context.
    * @param params The validated input parameters for the agent.
    * @param messageBus Message bus for policy enforcement.
+   * @param _toolName Optional override for the tool name.
+   * @param _toolDisplayName Optional override for the tool display name.
+   * @param _onAgentEvent Optional callback for parent session observability.
    */
   constructor(
     private readonly definition: LocalAgentDefinition,
@@ -62,6 +65,7 @@ export class LocalSubagentInvocation extends BaseToolInvocation<
     messageBus: MessageBus,
     _toolName?: string,
     _toolDisplayName?: string,
+    private readonly _onAgentEvent?: (event: AgentEvent) => void,
   ) {
     super(
       params,
@@ -101,9 +105,170 @@ export class LocalSubagentInvocation extends BaseToolInvocation<
   ): Promise<ToolResult> {
     let recentActivity: SubagentActivityItem[] = [];
 
+    // Raw SubagentActivityEvent handler — preserves all existing progress display logic.
+    // Passed as rawActivityCallback to LocalSubagentSession so the protocol can call it
+    // before translating to AgentEvents.
+    const onActivity = (activity: SubagentActivityEvent): void => {
+      if (!updateOutput) return;
+
+      let updated = false;
+
+      switch (activity.type) {
+        case 'THOUGHT_CHUNK': {
+          const text = String(activity.data['text']);
+          const lastItem = recentActivity[recentActivity.length - 1];
+
+          if (
+            lastItem &&
+            lastItem.type === 'thought' &&
+            lastItem.status === 'running'
+          ) {
+            lastItem.content = sanitizeThoughtContent(text);
+          } else {
+            recentActivity.push({
+              id: randomUUID(),
+              type: 'thought',
+              content: sanitizeThoughtContent(text),
+              status: 'running',
+            });
+          }
+          updated = true;
+          break;
+        }
+        case 'TOOL_CALL_START': {
+          const name = String(activity.data['name']);
+          const displayName = activity.data['displayName']
+            ? sanitizeErrorMessage(String(activity.data['displayName']))
+            : undefined;
+          const description = activity.data['description']
+            ? sanitizeErrorMessage(String(activity.data['description']))
+            : undefined;
+          const args = JSON.stringify(sanitizeToolArgs(activity.data['args']));
+          recentActivity.push({
+            id: randomUUID(),
+            type: 'tool_call',
+            content: name,
+            displayName,
+            description,
+            args,
+            status: 'running',
+          });
+          updated = true;
+          break;
+        }
+        case 'TOOL_CALL_END': {
+          const name = String(activity.data['name']);
+          const data = activity.data['data'];
+          const isError = isToolActivityError(data);
+
+          for (let i = recentActivity.length - 1; i >= 0; i--) {
+            if (
+              recentActivity[i].type === 'tool_call' &&
+              recentActivity[i].content === name &&
+              recentActivity[i].status === 'running'
+            ) {
+              recentActivity[i].status = isError ? 'error' : 'completed';
+              updated = true;
+              break;
+            }
+          }
+          break;
+        }
+        case 'ERROR': {
+          const error = String(activity.data['error']);
+          const errorType = activity.data['errorType'];
+          const sanitizedError = sanitizeErrorMessage(error);
+          const isCancellation =
+            errorType === SubagentActivityErrorType.CANCELLED ||
+            error === SUBAGENT_CANCELLED_ERROR_MESSAGE;
+          const isRejection =
+            errorType === SubagentActivityErrorType.REJECTED ||
+            error.startsWith(SUBAGENT_REJECTED_ERROR_PREFIX);
+
+          const toolName = activity.data['name']
+            ? String(activity.data['name'])
+            : undefined;
+
+          if (toolName && (isCancellation || isRejection)) {
+            for (let i = recentActivity.length - 1; i >= 0; i--) {
+              if (
+                recentActivity[i].type === 'tool_call' &&
+                recentActivity[i].content === toolName &&
+                recentActivity[i].status === 'running'
+              ) {
+                recentActivity[i].status = 'cancelled';
+                updated = true;
+                break;
+              }
+            }
+          } else if (toolName) {
+            // Mark non-rejection/non-cancellation errors as 'error'
+            for (let i = recentActivity.length - 1; i >= 0; i--) {
+              if (
+                recentActivity[i].type === 'tool_call' &&
+                recentActivity[i].content === toolName &&
+                recentActivity[i].status === 'running'
+              ) {
+                recentActivity[i].status = 'error';
+                updated = true;
+                break;
+              }
+            }
+          }
+
+          recentActivity.push({
+            id: randomUUID(),
+            type: 'thought',
+            content:
+              isCancellation || isRejection
+                ? sanitizedError
+                : `Error: ${sanitizedError}`,
+            status: isCancellation || isRejection ? 'cancelled' : 'error',
+          });
+          updated = true;
+          break;
+        }
+        default:
+          break;
+      }
+
+      if (updated) {
+        // Keep only the last N items
+        if (recentActivity.length > MAX_RECENT_ACTIVITY) {
+          recentActivity = recentActivity.slice(-MAX_RECENT_ACTIVITY);
+        }
+
+        const progress: SubagentProgress = {
+          isSubagentProgress: true,
+          agentName: this.definition.name,
+          recentActivity: [...recentActivity], // Copy to avoid mutation issues
+          state: 'running',
+        };
+
+        updateOutput(progress);
+      }
+    };
+
+    // Create session with the raw activity callback for rich progress display
+    const session = new LocalSubagentSession(
+      this.definition,
+      this.context,
+      this.messageBus,
+      onActivity,
+    );
+
+    // Subscribe for parent session observability (future use)
+    let unsubscribeParent: (() => void) | undefined;
+    if (this._onAgentEvent) {
+      unsubscribeParent = session.subscribe(this._onAgentEvent);
+    }
+
+    // Wire external abort signal to session abort
+    const abortListener = () => void session.abort();
+    signal.addEventListener('abort', abortListener, { once: true });
+
     try {
       if (updateOutput) {
-        // Send initial state
         const initialProgress: SubagentProgress = {
           isSubagentProgress: true,
           agentName: this.definition.name,
@@ -113,158 +278,16 @@ export class LocalSubagentInvocation extends BaseToolInvocation<
         updateOutput(initialProgress);
       }
 
-      // Create an activity callback to bridge the executor's events to the
-      // tool's streaming output.
-      const onActivity = (activity: SubagentActivityEvent): void => {
-        if (!updateOutput) return;
+      // Buffer non-query params, then send query as message to start execution
+      const query = String(this.params['query'] ?? '');
+      const otherParams = { ...this.params } as Record<string, unknown>;
+      delete otherParams['query'];
+      if (Object.keys(otherParams).length > 0) {
+        await session.send({ update: { config: otherParams } });
+      }
+      await session.send({ message: [{ type: 'text', text: query }] });
 
-        let updated = false;
-
-        switch (activity.type) {
-          case 'THOUGHT_CHUNK': {
-            const text = String(activity.data['text']);
-            const lastItem = recentActivity[recentActivity.length - 1];
-
-            if (
-              lastItem &&
-              lastItem.type === 'thought' &&
-              lastItem.status === 'running'
-            ) {
-              lastItem.content = sanitizeThoughtContent(text);
-            } else {
-              recentActivity.push({
-                id: randomUUID(),
-                type: 'thought',
-                content: sanitizeThoughtContent(text),
-                status: 'running',
-              });
-            }
-            updated = true;
-            break;
-          }
-          case 'TOOL_CALL_START': {
-            const name = String(activity.data['name']);
-            const displayName = activity.data['displayName']
-              ? sanitizeErrorMessage(String(activity.data['displayName']))
-              : undefined;
-            const description = activity.data['description']
-              ? sanitizeErrorMessage(String(activity.data['description']))
-              : undefined;
-            const args = JSON.stringify(
-              sanitizeToolArgs(activity.data['args']),
-            );
-            recentActivity.push({
-              id: randomUUID(),
-              type: 'tool_call',
-              content: name,
-              displayName,
-              description,
-              args,
-              status: 'running',
-            });
-            updated = true;
-            break;
-          }
-          case 'TOOL_CALL_END': {
-            const name = String(activity.data['name']);
-            const data = activity.data['data'];
-            const isError = isToolActivityError(data);
-
-            for (let i = recentActivity.length - 1; i >= 0; i--) {
-              if (
-                recentActivity[i].type === 'tool_call' &&
-                recentActivity[i].content === name &&
-                recentActivity[i].status === 'running'
-              ) {
-                recentActivity[i].status = isError ? 'error' : 'completed';
-                updated = true;
-                break;
-              }
-            }
-            break;
-          }
-          case 'ERROR': {
-            const error = String(activity.data['error']);
-            const errorType = activity.data['errorType'];
-            const sanitizedError = sanitizeErrorMessage(error);
-            const isCancellation =
-              errorType === SubagentActivityErrorType.CANCELLED ||
-              error === SUBAGENT_CANCELLED_ERROR_MESSAGE;
-            const isRejection =
-              errorType === SubagentActivityErrorType.REJECTED ||
-              error.startsWith(SUBAGENT_REJECTED_ERROR_PREFIX);
-
-            const toolName = activity.data['name']
-              ? String(activity.data['name'])
-              : undefined;
-
-            if (toolName && (isCancellation || isRejection)) {
-              for (let i = recentActivity.length - 1; i >= 0; i--) {
-                if (
-                  recentActivity[i].type === 'tool_call' &&
-                  recentActivity[i].content === toolName &&
-                  recentActivity[i].status === 'running'
-                ) {
-                  recentActivity[i].status = 'cancelled';
-                  updated = true;
-                  break;
-                }
-              }
-            } else if (toolName) {
-              // Mark non-rejection/non-cancellation errors as 'error'
-              for (let i = recentActivity.length - 1; i >= 0; i--) {
-                if (
-                  recentActivity[i].type === 'tool_call' &&
-                  recentActivity[i].content === toolName &&
-                  recentActivity[i].status === 'running'
-                ) {
-                  recentActivity[i].status = 'error';
-                  updated = true;
-                  break;
-                }
-              }
-            }
-
-            recentActivity.push({
-              id: randomUUID(),
-              type: 'thought',
-              content:
-                isCancellation || isRejection
-                  ? sanitizedError
-                  : `Error: ${sanitizedError}`,
-              status: isCancellation || isRejection ? 'cancelled' : 'error',
-            });
-            updated = true;
-            break;
-          }
-          default:
-            break;
-        }
-
-        if (updated) {
-          // Keep only the last N items
-          if (recentActivity.length > MAX_RECENT_ACTIVITY) {
-            recentActivity = recentActivity.slice(-MAX_RECENT_ACTIVITY);
-          }
-
-          const progress: SubagentProgress = {
-            isSubagentProgress: true,
-            agentName: this.definition.name,
-            recentActivity: [...recentActivity], // Copy to avoid mutation issues
-            state: 'running',
-          };
-
-          updateOutput(progress);
-        }
-      };
-
-      const executor = await LocalAgentExecutor.create(
-        this.definition,
-        this.context,
-        onActivity,
-      );
-
-      const output = await executor.run(this.params, signal);
+      const output = await session.getResult();
 
       if (output.terminate_reason === AgentTerminateMode.ABORTED) {
         const progress: SubagentProgress = {
@@ -359,6 +382,9 @@ ${output.result}`;
         // We omit the 'error' property so that the UI renders our rich returnDisplay
         // instead of the raw error message. The llmContent still informs the agent of the failure.
       };
+    } finally {
+      signal.removeEventListener('abort', abortListener);
+      unsubscribeParent?.();
     }
   }
 }

packages/core/src/agents/local-subagent-protocol.test.ts

@@ -0,0 +1,776 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
+import { LocalSubagentSession } from './local-subagent-protocol.js';
+import { LocalAgentExecutor } from './local-executor.js';
+import {
+  AgentTerminateMode,
+  type LocalAgentDefinition,
+  type SubagentActivityEvent,
+} from './types.js';
+import { makeFakeConfig } from '../test-utils/config.js';
+import { createMockMessageBus } from '../test-utils/mock-message-bus.js';
+import type { AgentLoopContext } from '../config/agent-loop-context.js';
+import type { AgentEvent } from '../agent/types.js';
+import type { MessageBus } from '../confirmation-bus/message-bus.js';
+import type { z } from 'zod';
+import type { Mocked } from 'vitest';
+
+vi.mock('./local-executor.js');
+
+const MockLocalAgentExecutor = vi.mocked(LocalAgentExecutor);
+
+// Captures the onActivity callback passed to LocalAgentExecutor.create().
+// Set via create.mockImplementation in beforeEach to avoid mock.calls index fragility.
+let capturedOnActivity: ((activity: SubagentActivityEvent) => void) | undefined;
+
+const testDefinition: LocalAgentDefinition = {
+  kind: 'local',
+  name: 'TestProtocolAgent',
+  description: 'A test agent for protocol tests.',
+  inputConfig: {
+    inputSchema: {
+      type: 'object',
+      properties: {
+        task: { type: 'string' },
+        priority: { type: 'number' },
+      },
+    },
+  },
+  modelConfig: { model: 'test', generateContentConfig: {} },
+  runConfig: { maxTimeMinutes: 1 },
+  promptConfig: { systemPrompt: 'test' },
+};
+
+const GOAL_OUTPUT = {
+  result: 'Analysis complete.',
+  terminate_reason: AgentTerminateMode.GOAL,
+};
+
+describe('LocalSubagentSession (protocol)', () => {
+  let mockContext: AgentLoopContext;
+  let mockMessageBus: MessageBus;
+  let mockExecutorInstance: Mocked<LocalAgentExecutor<z.ZodUnknown>>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    capturedOnActivity = undefined;
+
+    mockContext = makeFakeConfig() as unknown as AgentLoopContext;
+    mockMessageBus = createMockMessageBus();
+
+    mockExecutorInstance = {
+      run: vi.fn().mockResolvedValue(GOAL_OUTPUT),
+      definition: testDefinition,
+    } as unknown as Mocked<LocalAgentExecutor<z.ZodUnknown>>;
+
+    // Use mockImplementation (not mockResolvedValue) so we can capture onActivity.
+    MockLocalAgentExecutor.create.mockImplementation(
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      async (_def: any, _ctx: any, onActivity: any) => {
+        capturedOnActivity = onActivity;
+
+        return mockExecutorInstance as unknown as LocalAgentExecutor<z.ZodTypeAny>;
+      },
+    );
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  // ---------------------------------------------------------------------------
+  // Lifecycle events
+  // ---------------------------------------------------------------------------
+
+  describe('lifecycle events', () => {
+    it('emits agent_start then agent_end(completed) for a GOAL run', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'query' }] });
+      await session.getResult();
+
+      expect(events[0].type).toBe('agent_start');
+      expect(events[events.length - 1].type).toBe('agent_end');
+      const endEvent = events[events.length - 1];
+      if (endEvent.type === 'agent_end') {
+        expect(endEvent.reason).toBe('completed');
+      }
+    });
+
+    it('emits agent_start exactly once even if ensureAgentStart called twice internally', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'query' }] });
+      await session.getResult();
+
+      const startEvents = events.filter((e) => e.type === 'agent_start');
+      expect(startEvents).toHaveLength(1);
+    });
+
+    it('emits agent_end exactly once on error path', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      mockExecutorInstance.run.mockRejectedValue(new Error('executor failed'));
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'query' }] });
+      await expect(session.getResult()).rejects.toThrow('executor failed');
+
+      const endEvents = events.filter((e) => e.type === 'agent_end');
+      expect(endEvents).toHaveLength(1);
+    });
+
+    it('all events share the same streamId', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'query' }] });
+      await session.getResult();
+
+      const streamIds = new Set(events.map((e) => e.streamId));
+      expect(streamIds.size).toBe(1);
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Config buffering (update + message pattern)
+  // ---------------------------------------------------------------------------
+
+  describe('config buffering', () => {
+    it('merges buffered config with message query', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({
+        update: { config: { task: 'analyze', priority: 5 } },
+      });
+      await session.send({ message: [{ type: 'text', text: 'my query' }] });
+      await session.getResult();
+
+      expect(mockExecutorInstance.run).toHaveBeenCalledWith(
+        { task: 'analyze', priority: 5, query: 'my query' },
+        expect.any(AbortSignal),
+      );
+    });
+
+    it('omits query key when message text is empty', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({ update: { config: { task: 'no-query-task' } } });
+      await session.send({ message: [{ type: 'text', text: '' }] });
+      await session.getResult();
+
+      const callArgs = mockExecutorInstance.run.mock.calls[0][0];
+      expect(callArgs).not.toHaveProperty('query');
+      expect(callArgs).toEqual({ task: 'no-query-task' });
+    });
+
+    it('sends only query when no prior update', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({ message: [{ type: 'text', text: 'just a query' }] });
+      await session.getResult();
+
+      expect(mockExecutorInstance.run).toHaveBeenCalledWith(
+        { query: 'just a query' },
+        expect.any(AbortSignal),
+      );
+    });
+
+    it('multiple update calls are merged', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({ update: { config: { field1: 'a' } } });
+      await session.send({ update: { config: { field2: 'b' } } });
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(mockExecutorInstance.run).toHaveBeenCalledWith(
+        { field1: 'a', field2: 'b', query: 'q' },
+        expect.any(AbortSignal),
+      );
+    });
+
+    it('update returns streamId: null; message returns a streamId', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const updateResult = await session.send({ update: { config: {} } });
+      expect(updateResult.streamId).toBeNull();
+
+      const messageResult = await session.send({
+        message: [{ type: 'text', text: 'q' }],
+      });
+      expect(messageResult.streamId).not.toBeNull();
+      expect(typeof messageResult.streamId).toBe('string');
+
+      // Await completion to prevent dangling execution affecting subsequent tests
+      await session.getResult();
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Activity translation
+  // ---------------------------------------------------------------------------
+
+  describe('activity translation', () => {
+    function makeSession() {
+      const activityEvents: SubagentActivityEvent[] = [];
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      return { session, activityEvents };
+    }
+
+    async function runWithActivities(
+      session: LocalSubagentSession,
+      activities: SubagentActivityEvent[],
+    ) {
+      mockExecutorInstance.run.mockImplementation(async () => {
+        // capturedOnActivity is set by the create.mockImplementation in beforeEach
+        // and updated whenever create() is called. By the time run() is called,
+        // capturedOnActivity holds the onActivity closure for the most-recently
+        // created executor — which is the one associated with this session.
+        for (const act of activities) {
+          capturedOnActivity?.(act);
+        }
+        return GOAL_OUTPUT;
+      });
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+      return events;
+    }
+
+    it('THOUGHT_CHUNK → message event with thought content', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'THOUGHT_CHUNK',
+          data: { text: 'I am thinking...' },
+        },
+      ]);
+
+      const msgEvent = events.find((e) => e.type === 'message');
+      expect(msgEvent).toBeDefined();
+      if (msgEvent?.type === 'message') {
+        expect(msgEvent.role).toBe('agent');
+        expect(msgEvent.content).toContainEqual({
+          type: 'thought',
+          thought: 'I am thinking...',
+        });
+      }
+    });
+
+    it('TOOL_CALL_START → tool_request event', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'TOOL_CALL_START',
+          data: { callId: 'call-123', name: 'read_file', args: { path: '/a' } },
+        },
+      ]);
+
+      const reqEvent = events.find((e) => e.type === 'tool_request');
+      expect(reqEvent).toBeDefined();
+      if (reqEvent?.type === 'tool_request') {
+        expect(reqEvent.requestId).toBe('call-123');
+        expect(reqEvent.name).toBe('read_file');
+        expect(reqEvent.args).toEqual({ path: '/a' });
+      }
+    });
+
+    it('TOOL_CALL_END → tool_response event', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'TOOL_CALL_END',
+          data: { id: 'call-123', name: 'read_file', output: 'file contents' },
+        },
+      ]);
+
+      const respEvent = events.find((e) => e.type === 'tool_response');
+      expect(respEvent).toBeDefined();
+      if (respEvent?.type === 'tool_response') {
+        expect(respEvent.requestId).toBe('call-123');
+        expect(respEvent.name).toBe('read_file');
+        expect(respEvent.content).toContainEqual({
+          type: 'text',
+          text: 'file contents',
+        });
+      }
+    });
+
+    it('ERROR activity → error event with INTERNAL status, fatal: false', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'ERROR',
+          data: { error: 'something went wrong' },
+        },
+      ]);
+
+      const errEvent = events.find((e) => e.type === 'error');
+      expect(errEvent).toBeDefined();
+      if (errEvent?.type === 'error') {
+        expect(errEvent.status).toBe('INTERNAL');
+        expect(errEvent.message).toBe('something went wrong');
+        expect(errEvent.fatal).toBe(false);
+      }
+    });
+
+    it('unknown activity type → no events emitted', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          // eslint-disable-next-line @typescript-eslint/no-explicit-any
+          type: 'UNKNOWN_TYPE' as any,
+          data: {},
+        },
+      ]);
+
+      // Only agent_start and agent_end should be present
+      const nonLifecycle = events.filter(
+        (e) => e.type !== 'agent_start' && e.type !== 'agent_end',
+      );
+      expect(nonLifecycle).toHaveLength(0);
+    });
+
+    it('TOOL_CALL_START with non-object args defaults to {}', async () => {
+      const { session } = makeSession();
+      const events = await runWithActivities(session, [
+        {
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'TOOL_CALL_START',
+          data: { callId: 'x', name: 'tool', args: null },
+        },
+      ]);
+
+      const reqEvent = events.find((e) => e.type === 'tool_request');
+      if (reqEvent?.type === 'tool_request') {
+        expect(reqEvent.args).toEqual({});
+      }
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // getResult() promise
+  // ---------------------------------------------------------------------------
+
+  describe('getResult()', () => {
+    it('resolves with OutputObject on GOAL termination', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      const output = await session.getResult();
+
+      expect(output.result).toBe('Analysis complete.');
+      expect(output.terminate_reason).toBe(AgentTerminateMode.GOAL);
+    });
+
+    it('rejects when executor throws', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      mockExecutorInstance.run.mockRejectedValue(new Error('executor error'));
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow('executor error');
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // rawActivityCallback
+  // ---------------------------------------------------------------------------
+
+  describe('rawActivityCallback', () => {
+    it('receives raw SubagentActivityEvent before AgentEvent translation', async () => {
+      const rawActivities: SubagentActivityEvent[] = [];
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+        (activity) => rawActivities.push(activity),
+      );
+
+      const thoughtActivity: SubagentActivityEvent = {
+        isSubagentActivityEvent: true,
+        agentName: 'TestProtocolAgent',
+        type: 'THOUGHT_CHUNK',
+        data: { text: 'raw thought' },
+      };
+
+      mockExecutorInstance.run.mockImplementation(async () => {
+        const onActivity = MockLocalAgentExecutor.create.mock.calls[0]?.[2];
+        onActivity?.(thoughtActivity);
+        return GOAL_OUTPUT;
+      });
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(rawActivities).toHaveLength(1);
+      expect(rawActivities[0]).toBe(thoughtActivity);
+    });
+
+    it('is called before AgentEvent translation (raw arrives first)', async () => {
+      const callOrder: string[] = [];
+
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+        () => callOrder.push('raw'),
+      );
+
+      session.subscribe((e) => {
+        if (e.type === 'message') callOrder.push('translated');
+      });
+
+      mockExecutorInstance.run.mockImplementation(async () => {
+        const onActivity = MockLocalAgentExecutor.create.mock.calls[0]?.[2];
+        onActivity?.({
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'THOUGHT_CHUNK',
+          data: { text: 'thought' },
+        });
+        return GOAL_OUTPUT;
+      });
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(callOrder).toEqual(['raw', 'translated']);
+    });
+
+    it('is optional — no callback causes no error', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+        // no rawActivityCallback
+      );
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).resolves.toBeDefined();
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Subscription
+  // ---------------------------------------------------------------------------
+
+  describe('subscription', () => {
+    it('unsubscribe stops event delivery', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const received: AgentEvent[] = [];
+      const unsub = session.subscribe((e) => received.push(e));
+      unsub();
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(received).toHaveLength(0);
+    });
+
+    it('multiple subscribers all receive events', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const received1: AgentEvent[] = [];
+      const received2: AgentEvent[] = [];
+      session.subscribe((e) => received1.push(e));
+      session.subscribe((e) => received2.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(received1.length).toBeGreaterThan(0);
+      expect(received1).toEqual(received2);
+    });
+
+    it('events array accumulates all emitted events', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(session.events.length).toBeGreaterThanOrEqual(2); // at least agent_start + agent_end
+      expect(session.events[0].type).toBe('agent_start');
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Terminate mode mapping
+  // ---------------------------------------------------------------------------
+
+  describe('terminate mode → StreamEndReason mapping', () => {
+    const cases: Array<[AgentTerminateMode, string]> = [
+      [AgentTerminateMode.GOAL, 'completed'],
+      [AgentTerminateMode.TIMEOUT, 'max_time'],
+      [AgentTerminateMode.MAX_TURNS, 'max_turns'],
+      [AgentTerminateMode.ABORTED, 'aborted'],
+      [AgentTerminateMode.ERROR, 'failed'],
+      [AgentTerminateMode.ERROR_NO_COMPLETE_TASK_CALL, 'failed'],
+    ];
+
+    for (const [terminateMode, expectedReason] of cases) {
+      it(`${terminateMode} → agent_end(reason:'${expectedReason}')`, async () => {
+        mockExecutorInstance.run.mockResolvedValue({
+          result: 'done',
+          terminate_reason: terminateMode,
+        });
+
+        const session = new LocalSubagentSession(
+          testDefinition,
+          mockContext,
+          mockMessageBus,
+        );
+
+        const events: AgentEvent[] = [];
+        session.subscribe((e) => events.push(e));
+
+        await session.send({ message: [{ type: 'text', text: 'q' }] });
+        await session.getResult().catch(() => {
+          // ABORTED results in rejection — catch to let test complete
+        });
+
+        const endEvent = events.find((e) => e.type === 'agent_end');
+        expect(endEvent).toBeDefined();
+        if (endEvent?.type === 'agent_end') {
+          expect(endEvent.reason).toBe(expectedReason);
+        }
+      });
+    }
+  });
+
+  // ---------------------------------------------------------------------------
+  // Abort
+  // ---------------------------------------------------------------------------
+
+  describe('abort()', () => {
+    it('abort() causes agent_end(reason:aborted)', async () => {
+      // Make run() wait until aborted
+      let abortSignal: AbortSignal | undefined;
+      mockExecutorInstance.run.mockImplementation(
+        (_params: unknown, signal: AbortSignal) => {
+          abortSignal = signal;
+          return new Promise((_resolve, reject) => {
+            signal.addEventListener('abort', () => {
+              const err = new Error('AbortError');
+              err.name = 'AbortError';
+              reject(err);
+            });
+          });
+        },
+      );
+
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      void session.send({ message: [{ type: 'text', text: 'q' }] });
+
+      // Wait for executor to be created and run started
+      await vi.waitFor(() => {
+        expect(abortSignal).toBeDefined();
+      });
+
+      await session.abort();
+
+      await expect(session.getResult()).rejects.toThrow();
+
+      const endEvent = events.find((e) => e.type === 'agent_end');
+      expect(endEvent).toBeDefined();
+      if (endEvent?.type === 'agent_end') {
+        expect(endEvent.reason).toBe('aborted');
+      }
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Full event sequence
+  // ---------------------------------------------------------------------------
+
+  describe('full event sequence', () => {
+    it('emits agent_start → message(thought) → tool_request → tool_response → agent_end in order', async () => {
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      mockExecutorInstance.run.mockImplementation(async () => {
+        const onActivity = MockLocalAgentExecutor.create.mock.calls[0]?.[2];
+        onActivity?.({
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'THOUGHT_CHUNK',
+          data: { text: 'thinking' },
+        });
+        onActivity?.({
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'TOOL_CALL_START',
+          data: { callId: 'c1', name: 'tool', args: {} },
+        });
+        onActivity?.({
+          isSubagentActivityEvent: true,
+          agentName: 'TestProtocolAgent',
+          type: 'TOOL_CALL_END',
+          data: { id: 'c1', name: 'tool', output: 'result' },
+        });
+        return GOAL_OUTPUT;
+      });
+
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'go' }] });
+      await session.getResult();
+
+      const types = events.map((e) => e.type);
+      expect(types).toEqual([
+        'agent_start',
+        'message',
+        'tool_request',
+        'tool_response',
+        'agent_end',
+      ]);
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Concurrent send() guard
+  // ---------------------------------------------------------------------------
+
+  describe('concurrent send() guard', () => {
+    it('calling send() while a stream is active throws', async () => {
+      let abortSignal: AbortSignal | undefined;
+      mockExecutorInstance.run.mockImplementation(
+        (_params: unknown, signal: AbortSignal) => {
+          abortSignal = signal;
+          return new Promise((_resolve, reject) => {
+            // Reject when aborted so getResult() can settle during cleanup
+            signal.addEventListener('abort', () => {
+              const err = new Error('AbortError');
+              err.name = 'AbortError';
+              reject(err);
+            });
+          });
+        },
+      );
+
+      const session = new LocalSubagentSession(
+        testDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      void session.send({ message: [{ type: 'text', text: 'first' }] });
+
+      // Wait for execution to start
+      await vi.waitFor(() => {
+        expect(abortSignal).toBeDefined();
+      });
+
+      // Second send() while first stream is active must throw
+      await expect(
+        session.send({ message: [{ type: 'text', text: 'second' }] }),
+      ).rejects.toThrow('cannot be called while a stream is active');
+
+      // Clean up: abort to unblock the hanging executor
+      await session.abort();
+      await session.getResult().catch(() => {});
+    });
+  });
+});

packages/core/src/agents/local-subagent-protocol.ts

@@ -0,0 +1,427 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * @fileoverview LocalSubagentProtocol — wraps LocalAgentExecutor behind the
+ * AgentProtocol interface, translating SubagentActivityEvent callbacks into
+ * AgentEvents and exposing the executor result via getResult().
+ *
+ * Pattern mirrors LegacyAgentProtocol, but the loop body runs
+ * LocalAgentExecutor instead of GeminiClient.sendMessageStream().
+ */
+
+import { randomUUID } from 'node:crypto';
+import type { AgentLoopContext } from '../config/agent-loop-context.js';
+import { AgentSession } from '../agent/agent-session.js';
+import type {
+  AgentProtocol,
+  AgentSend,
+  AgentEvent,
+  StreamEndReason,
+  Unsubscribe,
+  ContentPart,
+} from '../agent/types.js';
+import { LocalAgentExecutor } from './local-executor.js';
+import {
+  AgentTerminateMode,
+  type LocalAgentDefinition,
+  type AgentInputs,
+  type OutputObject,
+  type SubagentActivityEvent,
+} from './types.js';
+import type { MessageBus } from '../confirmation-bus/message-bus.js';
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function isAbortLikeError(err: unknown): boolean {
+  return err instanceof Error && err.name === 'AbortError';
+}
+
+function mapTerminateMode(mode: AgentTerminateMode): StreamEndReason {
+  switch (mode) {
+    case AgentTerminateMode.GOAL:
+      return 'completed';
+    case AgentTerminateMode.TIMEOUT:
+      return 'max_time';
+    case AgentTerminateMode.MAX_TURNS:
+      return 'max_turns';
+    case AgentTerminateMode.ABORTED:
+      return 'aborted';
+    case AgentTerminateMode.ERROR:
+    case AgentTerminateMode.ERROR_NO_COMPLETE_TASK_CALL:
+    default:
+      return 'failed';
+  }
+}
+
+// ---------------------------------------------------------------------------
+// LocalSubagentProtocol
+// ---------------------------------------------------------------------------
+
+class LocalSubagentProtocol implements AgentProtocol {
+  private _events: AgentEvent[] = [];
+  private _subscribers = new Set<(event: AgentEvent) => void>();
+  private _streamId: string = randomUUID();
+  private _eventCounter = 0;
+  private _agentStartEmitted = false;
+  private _agentEndEmitted = false;
+  private _activeStreamId: string | undefined;
+  private _abortController = new AbortController();
+
+  // Result promise wiring
+  private _resultResolve!: (output: OutputObject) => void;
+  private _resultReject!: (err: unknown) => void;
+  private readonly _resultPromise: Promise<OutputObject>;
+
+  // Buffered config from send({update})
+  private _bufferedConfig: Record<string, unknown> = {};
+
+  constructor(
+    private readonly definition: LocalAgentDefinition,
+    private readonly context: AgentLoopContext,
+    _messageBus: MessageBus,
+    private readonly _rawActivityCallback?: (
+      activity: SubagentActivityEvent,
+    ) => void,
+  ) {
+    this._resultPromise = new Promise<OutputObject>((resolve, reject) => {
+      this._resultResolve = resolve;
+      this._resultReject = reject;
+    });
+  }
+
+  // ---------------------------------------------------------------------------
+  // AgentProtocol interface
+  // ---------------------------------------------------------------------------
+
+  get events(): readonly AgentEvent[] {
+    return this._events;
+  }
+
+  subscribe(callback: (event: AgentEvent) => void): Unsubscribe {
+    this._subscribers.add(callback);
+    return () => {
+      this._subscribers.delete(callback);
+    };
+  }
+
+  async send(payload: AgentSend): Promise<{ streamId: string | null }> {
+    if ('update' in payload && payload.update) {
+      // Buffer config for use when message send arrives
+      if (payload.update.config) {
+        this._bufferedConfig = {
+          ...this._bufferedConfig,
+          ...payload.update.config,
+        };
+      }
+      return { streamId: null };
+    }
+
+    if ('message' in payload && payload.message) {
+      if (this._activeStreamId) {
+        throw new Error(
+          'LocalSubagentProtocol.send() cannot be called while a stream is active.',
+        );
+      }
+
+      // Extract query text from the message ContentParts
+      const queryText = payload.message
+        .filter((p): p is ContentPart & { type: 'text' } => p.type === 'text')
+        .map((p) => p.text)
+        .join('');
+
+      // Only include 'query' in params when the message text is non-empty,
+      // so that callers that pass all fields via update.config are not affected.
+      const params: AgentInputs = {
+        ...this._bufferedConfig,
+        ...(queryText.length > 0 ? { query: queryText } : {}),
+      };
+
+      this._beginNewStream();
+      const streamId = this._streamId;
+
+      // Schedule execution in a macrotask so send() resolves before agent_start
+      setTimeout(() => {
+        void this._runExecutionInBackground(params);
+      }, 0);
+
+      return { streamId };
+    }
+
+    // action and elicitations are not supported
+    return { streamId: null };
+  }
+
+  async abort(): Promise<void> {
+    this._abortController.abort();
+  }
+
+  // ---------------------------------------------------------------------------
+  // Protocol-specific: result access
+  // ---------------------------------------------------------------------------
+
+  /**
+   * Resolves when the executor completes, with the raw OutputObject.
+   * Used by LocalSubagentInvocation to build the ToolResult.
+   */
+  getResult(): Promise<OutputObject> {
+    return this._resultPromise;
+  }
+
+  // ---------------------------------------------------------------------------
+  // Core: execution
+  // ---------------------------------------------------------------------------
+
+  private _beginNewStream(): void {
+    this._streamId = randomUUID();
+    this._eventCounter = 0;
+    this._abortController = new AbortController();
+    this._agentStartEmitted = false;
+    this._agentEndEmitted = false;
+    this._activeStreamId = this._streamId;
+  }
+
+  private async _runExecutionInBackground(params: AgentInputs): Promise<void> {
+    this._ensureAgentStart();
+    try {
+      await this._runExecution(params);
+    } catch (err: unknown) {
+      if (this._abortController.signal.aborted || isAbortLikeError(err)) {
+        this._ensureAgentEnd('aborted');
+        this._resultReject(err);
+      } else {
+        this._emitErrorAndAgentEnd(err);
+        this._resultReject(err);
+      }
+      this._clearActiveStream();
+    }
+  }
+
+  private async _runExecution(params: AgentInputs): Promise<void> {
+    const signal = this._abortController.signal;
+
+    const onActivity = (activity: SubagentActivityEvent): void => {
+      // Forward raw activity to invocation-level callback (for rich SubagentProgress display)
+      this._rawActivityCallback?.(activity);
+      this._emit(this._translateActivity(activity));
+    };
+
+    const executor = await LocalAgentExecutor.create(
+      this.definition,
+      this.context,
+      onActivity,
+    );
+
+    const output = await executor.run(params, signal);
+
+    if (
+      output.terminate_reason === AgentTerminateMode.ABORTED ||
+      signal.aborted
+    ) {
+      this._finishStream('aborted');
+    } else {
+      this._finishStream(mapTerminateMode(output.terminate_reason));
+    }
+
+    this._resultResolve(output);
+  }
+
+  // ---------------------------------------------------------------------------
+  // Activity → AgentEvent translation
+  // ---------------------------------------------------------------------------
+
+  private _translateActivity(activity: SubagentActivityEvent): AgentEvent[] {
+    switch (activity.type) {
+      case 'THOUGHT_CHUNK': {
+        const rawText = activity.data['text'];
+        const text = String(rawText ?? '');
+        return [
+          this._makeEvent('message', {
+            role: 'agent',
+            content: [{ type: 'thought', thought: text }],
+          }),
+        ];
+      }
+      case 'TOOL_CALL_START': {
+        const rawCallId = activity.data['callId'];
+        const callId = String(rawCallId ?? randomUUID());
+        const rawName = activity.data['name'];
+        const name = String(rawName ?? 'unknown');
+        const rawArgs = activity.data['args'];
+        const args: Record<string, unknown> =
+          rawArgs !== null &&
+          typeof rawArgs === 'object' &&
+          !Array.isArray(rawArgs)
+            ? // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
+              (rawArgs as Record<string, unknown>)
+            : {};
+        return [
+          this._makeEvent('tool_request', {
+            requestId: callId,
+            name,
+            args,
+          }),
+        ];
+      }
+      case 'TOOL_CALL_END': {
+        const rawId = activity.data['id'];
+        const requestId = String(rawId ?? randomUUID());
+        const rawName = activity.data['name'];
+        const name = String(rawName ?? 'unknown');
+        const rawOutput = activity.data['output'];
+        const output = String(rawOutput ?? '');
+        return [
+          this._makeEvent('tool_response', {
+            requestId,
+            name,
+            content: [{ type: 'text', text: output }],
+          }),
+        ];
+      }
+      case 'ERROR': {
+        const rawError = activity.data['error'];
+        const errorMsg = String(rawError ?? 'Unknown error');
+        return [
+          this._makeEvent('error', {
+            status: 'INTERNAL',
+            message: errorMsg,
+            fatal: false,
+          }),
+        ];
+      }
+      default:
+        return [];
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // Internal helpers (mirrors LegacyAgentProtocol)
+  // ---------------------------------------------------------------------------
+
+  private _emit(events: AgentEvent[]): void {
+    if (events.length === 0) return;
+    const subscribers = [...this._subscribers];
+    for (const event of events) {
+      this._events.push(event);
+      if (event.type === 'agent_end') {
+        this._agentEndEmitted = true;
+      }
+      for (const sub of subscribers) {
+        sub(event);
+      }
+    }
+  }
+
+  private _clearActiveStream(): void {
+    this._activeStreamId = undefined;
+  }
+
+  private _ensureAgentStart(): void {
+    if (!this._agentStartEmitted) {
+      this._agentStartEmitted = true;
+      this._emit([this._makeEvent('agent_start', {})]);
+    }
+  }
+
+  private _ensureAgentEnd(reason: StreamEndReason = 'completed'): void {
+    if (!this._agentEndEmitted && this._agentStartEmitted) {
+      this._emit([this._makeEvent('agent_end', { reason })]);
+    }
+  }
+
+  private _finishStream(
+    reason: StreamEndReason,
+    data?: Record<string, unknown>,
+  ): void {
+    if (data && !this._agentEndEmitted) {
+      this._emit([this._makeEvent('agent_end', { reason, data })]);
+      this._agentEndEmitted = true;
+    } else {
+      this._ensureAgentEnd(reason);
+    }
+    this._clearActiveStream();
+  }
+
+  private _emitErrorAndAgentEnd(err: unknown): void {
+    const message = err instanceof Error ? err.message : String(err);
+    this._ensureAgentStart();
+
+    const meta: Record<string, unknown> = {};
+    if (err instanceof Error) {
+      meta['errorName'] = err.constructor.name;
+      if ('exitCode' in err && typeof err.exitCode === 'number') {
+        meta['exitCode'] = err.exitCode;
+      }
+      if ('code' in err) {
+        meta['code'] = err.code;
+      }
+    }
+
+    this._emit([
+      this._makeEvent('error', {
+        status: 'INTERNAL',
+        message,
+        fatal: true,
+        ...(Object.keys(meta).length > 0 ? { _meta: meta } : {}),
+      }),
+    ]);
+    this._ensureAgentEnd('failed');
+  }
+
+  private _nextEventFields() {
+    return {
+      id: `${this._streamId}-${this._eventCounter++}`,
+      timestamp: new Date().toISOString(),
+      streamId: this._streamId,
+    };
+  }
+
+  private _makeEvent<T extends AgentEvent['type']>(
+    type: T,
+    payload: Omit<AgentEvent<T>, 'id' | 'timestamp' | 'streamId' | 'type'>,
+  ): AgentEvent {
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
+    return {
+      ...this._nextEventFields(),
+      type,
+      ...payload,
+    } as AgentEvent;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Public export
+// ---------------------------------------------------------------------------
+
+export class LocalSubagentSession extends AgentSession {
+  private readonly _localProtocol: LocalSubagentProtocol;
+
+  constructor(
+    definition: LocalAgentDefinition,
+    context: AgentLoopContext,
+    messageBus: MessageBus,
+    rawActivityCallback?: (activity: SubagentActivityEvent) => void,
+  ) {
+    const protocol = new LocalSubagentProtocol(
+      definition,
+      context,
+      messageBus,
+      rawActivityCallback,
+    );
+    super(protocol);
+    this._localProtocol = protocol;
+  }
+
+  /**
+   * Returns the raw executor OutputObject once execution completes.
+   * Used by LocalSubagentInvocation to build the ToolResult.
+   */
+  getResult(): Promise<OutputObject> {
+    return this._localProtocol.getResult();
+  }
+}

packages/core/src/agents/remote-invocation.ts

@@ -9,6 +9,7 @@ import {
   type ToolConfirmationOutcome,
   type ToolResult,
   type ToolCallConfirmationDetails,
+  type ToolLiveOutput,
 } from '../tools/tools.js';
 import {
   DEFAULT_QUERY_STRING,
@@ -16,44 +17,23 @@ import {
   type RemoteAgentDefinition,
   type AgentInputs,
   type SubagentProgress,
-  getAgentCardLoadOptions,
-  getRemoteAgentTargetUrl,
 } from './types.js';
 import { type AgentLoopContext } from '../config/agent-loop-context.js';
 import type { MessageBus } from '../confirmation-bus/message-bus.js';
-import type {
-  A2AClientManager,
-  SendMessageResult,
-} from './a2a-client-manager.js';
-import { extractIdsFromResponse, A2AResultReassembler } from './a2aUtils.js';
-import type { AuthenticationHandler } from '@a2a-js/sdk/client';
-import { debugLogger } from '../utils/debugLogger.js';
-import type { AnsiOutput } from '../utils/terminalSerializer.js';
-import { A2AAuthProviderFactory } from './auth-provider/factory.js';
 import { A2AAgentError } from './a2a-errors.js';
+import { RemoteSubagentSession } from './remote-subagent-protocol.js';
+import type { AgentEvent } from '../agent/types.js';
 
 /**
  * A tool invocation that proxies to a remote A2A agent.
  *
- * This implementation bypasses the local `LocalAgentExecutor` loop and directly
- * invokes the configured A2A tool.
+ * This implementation delegates execution to {@link RemoteSubagentSession},
+ * which wraps the A2A client streaming behind the AgentProtocol interface.
  */
 export class RemoteAgentInvocation extends BaseToolInvocation<
   RemoteAgentInputs,
   ToolResult
 > {
-  // Persist state across ephemeral invocation instances.
-  private static readonly sessionState = new Map<
-    string,
-    { contextId?: string; taskId?: string }
-  >();
-  // State for the ongoing conversation with the remote agent
-  private contextId: string | undefined;
-  private taskId: string | undefined;
-
-  private readonly clientManager: A2AClientManager;
-  private authHandler: AuthenticationHandler | undefined;
-
   constructor(
     private readonly definition: RemoteAgentDefinition,
     private readonly context: AgentLoopContext,
@@ -61,6 +41,7 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
     messageBus: MessageBus,
     _toolName?: string,
     _toolDisplayName?: string,
+    private readonly _onAgentEvent?: (event: AgentEvent) => void,
   ) {
     const query = params['query'] ?? DEFAULT_QUERY_STRING;
     if (typeof query !== 'string') {
@@ -75,43 +56,19 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
       _toolName ?? definition.name,
       _toolDisplayName ?? definition.displayName,
     );
-    const clientManager = this.context.config.getA2AClientManager();
-    if (!clientManager) {
+
+    // Validate that A2AClientManager is available at construction time
+    if (!this.context.config.getA2AClientManager()) {
       throw new Error(
         `Failed to initialize RemoteAgentInvocation for '${definition.name}': A2AClientManager is not available.`,
       );
     }
-    this.clientManager = clientManager;
   }
 
   getDescription(): string {
     return `Calling remote agent ${this.definition.displayName ?? this.definition.name}`;
   }
 
-  private async getAuthHandler(): Promise<AuthenticationHandler | undefined> {
-    if (this.authHandler) {
-      return this.authHandler;
-    }
-
-    if (this.definition.auth) {
-      const targetUrl = getRemoteAgentTargetUrl(this.definition);
-      const provider = await A2AAuthProviderFactory.create({
-        authConfig: this.definition.auth,
-        agentName: this.definition.name,
-        targetUrl,
-        agentCardUrl: this.definition.agentCardUrl,
-      });
-      if (!provider) {
-        throw new Error(
-          `Failed to create auth provider for agent '${this.definition.name}'`,
-        );
-      }
-      this.authHandler = provider;
-    }
-
-    return this.authHandler;
-  }
-
   protected override async getConfirmationDetails(
     _abortSignal: AbortSignal,
   ): Promise<ToolCallConfirmationDetails | false> {
@@ -128,13 +85,33 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
 
   async execute(
     _signal: AbortSignal,
-    updateOutput?: (output: string | AnsiOutput | SubagentProgress) => void,
+    updateOutput?: (output: ToolLiveOutput) => void,
   ): Promise<ToolResult> {
-    // 1. Ensure the agent is loaded (cached by manager)
-    // We assume the user has provided an access token via some mechanism (TODO),
-    // or we rely on ADC.
-    const reassembler = new A2AResultReassembler();
     const agentName = this.definition.displayName ?? this.definition.name;
+    const session = new RemoteSubagentSession(
+      this.definition,
+      this.context,
+      this.messageBus,
+    );
+
+    // Wire external abort signal to session abort
+    const abortListener = () => void session.abort();
+    _signal.addEventListener('abort', abortListener, { once: true });
+
+    // Subscribe for parent session observability (future use)
+    let unsubscribeParent: (() => void) | undefined;
+    if (this._onAgentEvent) {
+      unsubscribeParent = session.subscribe(this._onAgentEvent);
+    }
+
+    // Subscribe to message events for live SubagentProgress updates
+    const unsubscribeProgress = session.subscribe((event: AgentEvent) => {
+      if (event.type === 'message' && updateOutput) {
+        const currentProgress = session.getLatestProgress();
+        if (currentProgress) updateOutput(currentProgress);
+      }
+    });
+
     try {
       if (updateOutput) {
         updateOutput({
@@ -152,97 +129,25 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
         });
       }
 
-      const priorState = RemoteAgentInvocation.sessionState.get(
-        this.definition.name,
-      );
-      if (priorState) {
-        this.contextId = priorState.contextId;
-        this.taskId = priorState.taskId;
-      }
+      await session.send({
+        message: [{ type: 'text', text: this.params.query }],
+      });
 
-      const authHandler = await this.getAuthHandler();
-
-      if (!this.clientManager.getClient(this.definition.name)) {
-        await this.clientManager.loadAgent(
-          this.definition.name,
-          getAgentCardLoadOptions(this.definition),
-          authHandler,
-        );
-      }
-
-      const message = this.params.query;
-
-      const stream = this.clientManager.sendMessageStream(
-        this.definition.name,
-        message,
-        {
-          contextId: this.contextId,
-          taskId: this.taskId,
-          signal: _signal,
-        },
-      );
-
-      let finalResponse: SendMessageResult | undefined;
-
-      for await (const chunk of stream) {
-        if (_signal.aborted) {
-          throw new Error('Operation aborted');
-        }
-        finalResponse = chunk;
-        reassembler.update(chunk);
-
-        if (updateOutput) {
-          updateOutput({
-            isSubagentProgress: true,
-            agentName,
-            state: 'running',
-            recentActivity: reassembler.toActivityItems(),
-            result: reassembler.toString(),
-          });
-        }
-
-        const {
-          contextId: newContextId,
-          taskId: newTaskId,
-          clearTaskId,
-        } = extractIdsFromResponse(chunk);
-
-        if (newContextId) {
-          this.contextId = newContextId;
-        }
-
-        this.taskId = clearTaskId ? undefined : (newTaskId ?? this.taskId);
-      }
-
-      if (!finalResponse) {
-        throw new Error('No response from remote agent.');
-      }
-
-      const finalOutput = reassembler.toString();
-
-      debugLogger.debug(
-        `[RemoteAgent] Final response from ${this.definition.name}:\n${JSON.stringify(finalResponse, null, 2)}`,
-      );
-
-      const finalProgress: SubagentProgress = {
-        isSubagentProgress: true,
-        agentName,
-        state: 'completed',
-        result: finalOutput,
-        recentActivity: reassembler.toActivityItems(),
-      };
+      const result = await session.getResult();
 
+      // Emit final completed progress
       if (updateOutput) {
-        updateOutput(finalProgress);
+        const finalProgress = session.getLatestProgress();
+        if (finalProgress) updateOutput(finalProgress);
       }
 
-      return {
-        llmContent: [{ text: finalOutput }],
-        returnDisplay: finalProgress,
-      };
+      return result;
     } catch (error: unknown) {
-      const partialOutput = reassembler.toString();
-      // Surface structured, user-friendly error messages.
+      const partialProgress = session.getLatestProgress();
+      const partialOutput =
+        typeof partialProgress?.result === 'string'
+          ? partialProgress.result
+          : '';
       const errorMessage = this.formatExecutionError(error);
       const fullDisplay = partialOutput
         ? `${partialOutput}\n\n${errorMessage}`
@@ -253,7 +158,7 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
         agentName,
         state: 'error',
         result: fullDisplay,
-        recentActivity: reassembler.toActivityItems(),
+        recentActivity: partialProgress?.recentActivity ?? [],
       };
 
       if (updateOutput) {
@@ -265,11 +170,9 @@ export class RemoteAgentInvocation extends BaseToolInvocation<
         returnDisplay: errorProgress,
       };
     } finally {
-      // Persist state even on partial failures or aborts to maintain conversational continuity.
-      RemoteAgentInvocation.sessionState.set(this.definition.name, {
-        contextId: this.contextId,
-        taskId: this.taskId,
-      });
+      _signal.removeEventListener('abort', abortListener);
+      unsubscribeProgress();
+      unsubscribeParent?.();
     }
   }
 

packages/core/src/agents/remote-subagent-protocol.test.ts

@@ -0,0 +1,776 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import {
+  describe,
+  it,
+  expect,
+  vi,
+  beforeEach,
+  afterEach,
+  type Mock,
+} from 'vitest';
+import { RemoteSubagentSession } from './remote-subagent-protocol.js';
+
+import { A2AAuthProviderFactory } from './auth-provider/factory.js';
+import type { RemoteAgentDefinition, SubagentProgress } from './types.js';
+import { createMockMessageBus } from '../test-utils/mock-message-bus.js';
+import type { AgentLoopContext } from '../config/agent-loop-context.js';
+import type { AgentEvent } from '../agent/types.js';
+import type { Config } from '../config/config.js';
+import type { A2AAuthProvider } from './auth-provider/types.js';
+
+// Mock A2AClientManager at module level
+vi.mock('./a2a-client-manager.js', () => ({
+  A2AClientManager: vi.fn().mockImplementation(() => ({
+    getClient: vi.fn(),
+    loadAgent: vi.fn(),
+    sendMessageStream: vi.fn(),
+  })),
+}));
+
+// Mock A2AAuthProviderFactory
+vi.mock('./auth-provider/factory.js', () => ({
+  A2AAuthProviderFactory: {
+    create: vi.fn(),
+  },
+}));
+
+const mockDefinition: RemoteAgentDefinition = {
+  name: 'test-remote-agent',
+  kind: 'remote',
+  agentCardUrl: 'http://test-agent/card',
+  displayName: 'Test Remote Agent',
+  description: 'A test remote agent',
+  inputConfig: {
+    inputSchema: { type: 'object' },
+  },
+};
+
+function makeChunk(text: string) {
+  return {
+    kind: 'message' as const,
+    messageId: `msg-${Math.random()}`,
+    role: 'agent' as const,
+    parts: [{ kind: 'text' as const, text }],
+  };
+}
+
+describe('RemoteSubagentSession (protocol)', () => {
+  let mockClientManager: {
+    getClient: Mock;
+    loadAgent: Mock;
+    sendMessageStream: Mock;
+  };
+  let mockContext: AgentLoopContext;
+  let mockMessageBus: ReturnType<typeof createMockMessageBus>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+
+    // Static session state is not cleared between tests — each test uses a
+    // unique agent name to avoid cross-test contamination.
+
+    mockClientManager = {
+      getClient: vi.fn().mockReturnValue(undefined), // client not yet loaded
+      loadAgent: vi.fn().mockResolvedValue(undefined),
+      sendMessageStream: vi.fn(),
+    };
+
+    const mockConfig = {
+      getA2AClientManager: vi.fn().mockReturnValue(mockClientManager),
+      injectionService: {
+        getLatestInjectionIndex: vi.fn().mockReturnValue(0),
+      },
+    } as unknown as Config;
+
+    mockContext = { config: mockConfig } as unknown as AgentLoopContext;
+    mockMessageBus = createMockMessageBus();
+
+    // Default: sendMessageStream yields one chunk with "Hello"
+    mockClientManager.sendMessageStream.mockImplementation(async function* () {
+      yield makeChunk('Hello');
+    });
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  // Helper: run a session with the default or custom stream and collect events
+  async function runSession(
+    definition: RemoteAgentDefinition = mockDefinition,
+    query = 'test query',
+  ) {
+    const session = new RemoteSubagentSession(
+      definition,
+      mockContext,
+      mockMessageBus,
+    );
+    const events: AgentEvent[] = [];
+    session.subscribe((e) => events.push(e));
+    await session.send({ message: [{ type: 'text', text: query }] });
+    const result = await session.getResult();
+    return { session, events, result };
+  }
+
+  // ---------------------------------------------------------------------------
+  // Lifecycle events
+  // ---------------------------------------------------------------------------
+
+  describe('lifecycle events', () => {
+    it('emits agent_start then agent_end(completed) on success', async () => {
+      const { events } = await runSession();
+
+      const types = events.map((e) => e.type);
+      expect(types[0]).toBe('agent_start');
+      expect(types[types.length - 1]).toBe('agent_end');
+      const end = events[events.length - 1];
+      if (end.type === 'agent_end') {
+        expect(end.reason).toBe('completed');
+      }
+    });
+
+    it('emits agent_start exactly once', async () => {
+      const { events } = await runSession();
+      expect(events.filter((e) => e.type === 'agent_start')).toHaveLength(1);
+    });
+
+    it('emits agent_end exactly once on error path', async () => {
+      mockClientManager.sendMessageStream.mockReturnValue({
+        [Symbol.asyncIterator]() {
+          return {
+            async next(): Promise<IteratorResult<never>> {
+              throw new Error('stream error');
+            },
+          };
+        },
+      });
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow('stream error');
+
+      expect(events.filter((e) => e.type === 'agent_end')).toHaveLength(1);
+    });
+
+    it('all events share the same streamId', async () => {
+      const { events } = await runSession();
+      const streamIds = new Set(events.map((e) => e.streamId));
+      expect(streamIds.size).toBe(1);
+    });
+
+    it('message returns a non-null streamId; unsupported payload returns null', async () => {
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const updateResult = await session.send({
+        update: { config: { key: 'val' } },
+      });
+      expect(updateResult.streamId).toBeNull();
+
+      const messageResult = await session.send({
+        message: [{ type: 'text', text: 'q' }],
+      });
+      expect(messageResult.streamId).not.toBeNull();
+      // complete the session to avoid dangling execution
+      await session.getResult();
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Chunk → AgentEvent translation
+  // ---------------------------------------------------------------------------
+
+  describe('chunk → AgentEvent translation', () => {
+    it('each A2A chunk produces a message event with current accumulated text', async () => {
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          yield makeChunk('Hello');
+          yield makeChunk(' world');
+        },
+      );
+
+      const { events } = await runSession();
+
+      const msgEvents = events.filter((e) => e.type === 'message');
+      expect(msgEvents.length).toBeGreaterThanOrEqual(1);
+      // Final message event should contain the accumulated text
+      const lastMsg = msgEvents[msgEvents.length - 1];
+      if (lastMsg?.type === 'message') {
+        const textContent = lastMsg.content.find((c) => c.type === 'text');
+        expect(textContent).toBeDefined();
+        if (textContent?.type === 'text') {
+          expect(textContent.text).toContain('Hello');
+        }
+      }
+    });
+
+    it('getLatestProgress() is updated per chunk with state running', async () => {
+      let capturedProgress: SubagentProgress | undefined;
+
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          yield makeChunk('Partial');
+        },
+      );
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      session.subscribe((e) => {
+        if (e.type === 'message') {
+          capturedProgress = session.getLatestProgress();
+        }
+      });
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      // During streaming, progress should be 'running'
+      expect(capturedProgress).toBeDefined();
+      // Note: by the time we check, progress may be 'completed'.
+      // During the message event, it was 'running'.
+      expect(capturedProgress?.isSubagentProgress).toBe(true);
+      expect(capturedProgress?.agentName).toBe('Test Remote Agent');
+    });
+
+    it('getLatestProgress() state is completed after getResult() resolves', async () => {
+      const { session } = await runSession();
+      const progress = session.getLatestProgress();
+      expect(progress?.state).toBe('completed');
+      expect(progress?.result).toBe('Hello');
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // getResult() promise
+  // ---------------------------------------------------------------------------
+
+  describe('getResult()', () => {
+    it('resolves with ToolResult containing llmContent and SubagentProgress returnDisplay', async () => {
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          yield makeChunk('Result text');
+        },
+      );
+
+      const { result } = await runSession();
+
+      expect(result.llmContent).toEqual([{ text: 'Result text' }]);
+      const display = result.returnDisplay as SubagentProgress;
+      expect(display.isSubagentProgress).toBe(true);
+      expect(display.state).toBe('completed');
+      expect(display.result).toBe('Result text');
+      expect(display.agentName).toBe('Test Remote Agent');
+    });
+
+    it('rejects when stream throws a non-A2A error', async () => {
+      mockClientManager.sendMessageStream.mockReturnValue({
+        [Symbol.asyncIterator]() {
+          return {
+            async next(): Promise<IteratorResult<never>> {
+              throw new Error('network failure');
+            },
+          };
+        },
+      });
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow();
+    });
+
+    it('resolves even with empty stream (empty final output)', async () => {
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          // yield nothing
+        },
+      );
+
+      const { result } = await runSession();
+      expect(result.llmContent).toEqual([{ text: '' }]);
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Session state persistence
+  // ---------------------------------------------------------------------------
+
+  describe('session state persistence', () => {
+    it('second call reuses contextId captured from first call', async () => {
+      const agentName = 'persistent-agent';
+      const persistDef: RemoteAgentDefinition = {
+        ...mockDefinition,
+        name: agentName,
+      };
+
+      let callCount = 0;
+      mockClientManager.sendMessageStream.mockImplementation(async function* (
+        _name: string,
+        _query: string,
+        opts: { contextId?: string },
+      ) {
+        callCount++;
+        if (callCount === 1) {
+          // First call: return a chunk that yields a contextId
+          yield {
+            kind: 'message' as const,
+            messageId: 'msg-1',
+            role: 'agent' as const,
+            contextId: 'ctx-from-server',
+            parts: [{ kind: 'text' as const, text: 'First response' }],
+          };
+        } else {
+          // Second call: caller should have passed the contextId
+          expect(opts.contextId).toBe('ctx-from-server');
+          yield makeChunk('Second response');
+        }
+      });
+
+      // First call
+      const session1 = new RemoteSubagentSession(
+        persistDef,
+        mockContext,
+        mockMessageBus,
+      );
+      await session1.send({ message: [{ type: 'text', text: 'first' }] });
+      await session1.getResult();
+
+      // Second call — different session but same agent name → should reuse contextId
+      const session2 = new RemoteSubagentSession(
+        persistDef,
+        mockContext,
+        mockMessageBus,
+      );
+      await session2.send({ message: [{ type: 'text', text: 'second' }] });
+      await session2.getResult();
+
+      expect(callCount).toBe(2);
+    });
+
+    it('different agent names have independent session state', async () => {
+      const def1: RemoteAgentDefinition = {
+        ...mockDefinition,
+        name: 'agent-alpha',
+      };
+      const def2: RemoteAgentDefinition = {
+        ...mockDefinition,
+        name: 'agent-beta',
+      };
+
+      const capturedContextIds: Array<string | undefined> = [];
+      mockClientManager.sendMessageStream.mockImplementation(async function* (
+        _name: string,
+        _query: string,
+        opts: { contextId?: string },
+      ) {
+        capturedContextIds.push(opts.contextId);
+        yield {
+          kind: 'message' as const,
+          messageId: 'msg-1',
+          role: 'agent' as const,
+          contextId: `ctx-for-${_name}`,
+          parts: [{ kind: 'text' as const, text: 'ok' }],
+        };
+      });
+
+      const session1 = new RemoteSubagentSession(
+        def1,
+        mockContext,
+        mockMessageBus,
+      );
+      await session1.send({ message: [{ type: 'text', text: 'q' }] });
+      await session1.getResult();
+
+      const session2 = new RemoteSubagentSession(
+        def2,
+        mockContext,
+        mockMessageBus,
+      );
+      await session2.send({ message: [{ type: 'text', text: 'q' }] });
+      await session2.getResult();
+
+      // Both start with no contextId (different agents, different state entries)
+      expect(capturedContextIds[0]).toBeUndefined();
+      expect(capturedContextIds[1]).toBeUndefined();
+    });
+
+    it('taskId is cleared when a terminal-state task chunk is received', async () => {
+      // A task chunk with a terminal status sets clearTaskId=true, which
+      // should clear this.taskId so it is NOT passed on the next call.
+      const agentName = 'clearTaskId-agent';
+      const def: RemoteAgentDefinition = { ...mockDefinition, name: agentName };
+
+      let callCount = 0;
+      const capturedTaskIds: Array<string | undefined> = [];
+
+      mockClientManager.sendMessageStream.mockImplementation(async function* (
+        _n: string,
+        _q: string,
+        opts: { taskId?: string },
+      ) {
+        callCount++;
+        capturedTaskIds.push(opts.taskId);
+        if (callCount === 1) {
+          // First call: yield a task chunk with taskId + terminal status → clearTaskId
+          yield {
+            kind: 'task' as const,
+            id: 'task-123',
+            contextId: 'ctx-1',
+            status: { state: 'completed' as const },
+          };
+        } else {
+          yield makeChunk('done');
+        }
+      });
+
+      const session1 = new RemoteSubagentSession(
+        def,
+        mockContext,
+        mockMessageBus,
+      );
+      await session1.send({ message: [{ type: 'text', text: 'first' }] });
+      await session1.getResult();
+
+      const session2 = new RemoteSubagentSession(
+        def,
+        mockContext,
+        mockMessageBus,
+      );
+      await session2.send({ message: [{ type: 'text', text: 'second' }] });
+      await session2.getResult();
+
+      expect(callCount).toBe(2);
+      // First call starts with no taskId
+      expect(capturedTaskIds[0]).toBeUndefined();
+      // Second call: taskId was cleared because terminal-state task chunk was received
+      expect(capturedTaskIds[1]).toBeUndefined();
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Auth setup
+  // ---------------------------------------------------------------------------
+
+  describe('auth setup', () => {
+    it('no auth → loadAgent called without auth handler', async () => {
+      await runSession();
+
+      expect(mockClientManager.loadAgent).toHaveBeenCalledWith(
+        'test-remote-agent',
+        { type: 'url', url: 'http://test-agent/card' },
+        undefined,
+      );
+    });
+
+    it('definition.auth present → A2AAuthProviderFactory.create called', async () => {
+      const authDef: RemoteAgentDefinition = {
+        ...mockDefinition,
+        name: 'auth-agent',
+        auth: {
+          type: 'http' as const,
+          scheme: 'Bearer' as const,
+          token: 'secret',
+        },
+      };
+
+      const mockProvider = {
+        type: 'http' as const,
+        headers: vi.fn().mockResolvedValue({ Authorization: 'Bearer secret' }),
+        shouldRetryWithHeaders: vi.fn(),
+      } as unknown as A2AAuthProvider;
+      (A2AAuthProviderFactory.create as Mock).mockResolvedValue(mockProvider);
+
+      await runSession(authDef, 'q');
+
+      expect(A2AAuthProviderFactory.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          agentName: 'auth-agent',
+          agentCardUrl: 'http://test-agent/card',
+        }),
+      );
+      expect(mockClientManager.loadAgent).toHaveBeenCalledWith(
+        'auth-agent',
+        expect.any(Object),
+        mockProvider,
+      );
+    });
+
+    it('auth factory returns undefined → throws error that rejects getResult()', async () => {
+      const authDef: RemoteAgentDefinition = {
+        ...mockDefinition,
+        name: 'failing-auth-agent',
+        auth: {
+          type: 'http' as const,
+          scheme: 'Bearer' as const,
+          token: 'secret',
+        },
+      };
+
+      (A2AAuthProviderFactory.create as Mock).mockResolvedValue(undefined);
+
+      const session = new RemoteSubagentSession(
+        authDef,
+        mockContext,
+        mockMessageBus,
+      );
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow(
+        "Failed to create auth provider for agent 'failing-auth-agent'",
+      );
+    });
+
+    it('agent already loaded → loadAgent not called again', async () => {
+      // Return a client object (truthy) so getClient returns defined
+      mockClientManager.getClient.mockReturnValue({});
+
+      await runSession();
+
+      expect(mockClientManager.loadAgent).not.toHaveBeenCalled();
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Error handling
+  // ---------------------------------------------------------------------------
+
+  describe('error handling', () => {
+    it('stream error → error event + agent_end(failed)', async () => {
+      mockClientManager.sendMessageStream.mockReturnValue({
+        [Symbol.asyncIterator]() {
+          return {
+            async next(): Promise<IteratorResult<never>> {
+              throw new Error('network error');
+            },
+          };
+        },
+      });
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow();
+
+      const errEvent = events.find((e) => e.type === 'error');
+      expect(errEvent).toBeDefined();
+
+      const endEvent = events.find((e) => e.type === 'agent_end');
+      expect(endEvent).toBeDefined();
+      if (endEvent?.type === 'agent_end') {
+        expect(endEvent.reason).toBe('failed');
+      }
+    });
+
+    it('missing A2AClientManager → rejects getResult()', async () => {
+      const mockConfig = {
+        getA2AClientManager: vi.fn().mockReturnValue(undefined),
+        injectionService: {
+          getLatestInjectionIndex: vi.fn().mockReturnValue(0),
+        },
+      } as unknown as Config;
+      const noClientContext = {
+        config: mockConfig,
+      } as unknown as AgentLoopContext;
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        noClientContext,
+        mockMessageBus,
+      );
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await expect(session.getResult()).rejects.toThrow(
+        'A2AClientManager not available',
+      );
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Subscription
+  // ---------------------------------------------------------------------------
+
+  describe('subscription', () => {
+    it('unsubscribe stops event delivery', async () => {
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const received: AgentEvent[] = [];
+      const unsub = session.subscribe((e) => received.push(e));
+      unsub();
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(received).toHaveLength(0);
+    });
+
+    it('multiple subscribers all receive events', async () => {
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const events1: AgentEvent[] = [];
+      const events2: AgentEvent[] = [];
+      session.subscribe((e) => events1.push(e));
+      session.subscribe((e) => events2.push(e));
+
+      await session.send({ message: [{ type: 'text', text: 'q' }] });
+      await session.getResult();
+
+      expect(events1.length).toBeGreaterThan(0);
+      expect(events1).toEqual(events2);
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Abort
+  // ---------------------------------------------------------------------------
+
+  describe('abort()', () => {
+    it('abort() causes agent_end(reason:aborted)', async () => {
+      let resolveChunk: (() => void) | undefined;
+
+      // Stream that blocks until we abort
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          // Hang until aborted
+          await new Promise<void>((resolve) => {
+            resolveChunk = resolve;
+          });
+          yield makeChunk('Too late');
+        },
+      );
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+
+      void session.send({ message: [{ type: 'text', text: 'q' }] });
+
+      // Wait for agent_start to be emitted before aborting
+      await vi.waitFor(() => {
+        expect(events.some((e) => e.type === 'agent_start')).toBe(true);
+      });
+
+      await session.abort();
+
+      // Resolve the hanging chunk generator so it can check the signal
+      resolveChunk?.();
+
+      await expect(session.getResult()).rejects.toThrow();
+
+      const endEvent = events.find((e) => e.type === 'agent_end');
+      expect(endEvent).toBeDefined();
+      if (endEvent?.type === 'agent_end') {
+        expect(endEvent.reason).toBe('aborted');
+      }
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // sendMessageStream call args
+  // ---------------------------------------------------------------------------
+
+  describe('sendMessageStream call arguments', () => {
+    it('passes the query string from the message payload', async () => {
+      await runSession(mockDefinition, 'my specific query');
+
+      expect(mockClientManager.sendMessageStream).toHaveBeenCalledWith(
+        'test-remote-agent',
+        'my specific query',
+        expect.objectContaining({ signal: expect.any(Object) }),
+      );
+    });
+
+    it('uses DEFAULT_QUERY_STRING when message text is empty', async () => {
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+      await session.send({ message: [{ type: 'text', text: '' }] });
+      await session.getResult();
+
+      // DEFAULT_QUERY_STRING = 'Get Started!'
+      expect(mockClientManager.sendMessageStream).toHaveBeenCalledWith(
+        'test-remote-agent',
+        'Get Started!',
+        expect.objectContaining({ signal: expect.any(Object) }),
+      );
+    });
+  });
+
+  // ---------------------------------------------------------------------------
+  // Concurrent send() guard
+  // ---------------------------------------------------------------------------
+
+  describe('concurrent send() guard', () => {
+    it('calling send() while a stream is active throws', async () => {
+      let resolveChunk!: () => void;
+
+      mockClientManager.sendMessageStream.mockImplementation(
+        async function* () {
+          // Block until test releases the chunk
+          await new Promise<void>((resolve) => {
+            resolveChunk = resolve;
+          });
+          yield makeChunk('late');
+        },
+      );
+
+      const session = new RemoteSubagentSession(
+        mockDefinition,
+        mockContext,
+        mockMessageBus,
+      );
+
+      void session.send({ message: [{ type: 'text', text: 'first' }] });
+
+      // Wait for the stream to actually start (agent_start emitted)
+      const events: AgentEvent[] = [];
+      session.subscribe((e) => events.push(e));
+      await vi.waitFor(() => {
+        expect(events.some((e) => e.type === 'agent_start')).toBe(true);
+      });
+
+      // Second send() while first stream is active must throw
+      await expect(
+        session.send({ message: [{ type: 'text', text: 'second' }] }),
+      ).rejects.toThrow('cannot be called while a stream is active');
+
+      // Clean up: release the blocked generator so getResult() can settle
+      resolveChunk();
+      await session.getResult().catch(() => {});
+    });
+  });
+});

packages/core/src/agents/remote-subagent-protocol.ts

@@ -0,0 +1,461 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * @fileoverview RemoteSubagentProtocol — wraps A2A remote agent streaming
+ * behind the AgentProtocol interface.
+ *
+ * Pattern mirrors LocalSubagentProtocol and LegacyAgentProtocol, but the loop
+ * body drives A2AClientManager instead of LocalAgentExecutor.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type { AgentLoopContext } from '../config/agent-loop-context.js';
+import { AgentSession } from '../agent/agent-session.js';
+import type {
+  AgentProtocol,
+  AgentSend,
+  AgentEvent,
+  StreamEndReason,
+  Unsubscribe,
+  ContentPart,
+} from '../agent/types.js';
+import type { ToolResult } from '../tools/tools.js';
+import {
+  DEFAULT_QUERY_STRING,
+  type RemoteAgentDefinition,
+  type SubagentProgress,
+  getRemoteAgentTargetUrl,
+  getAgentCardLoadOptions,
+} from './types.js';
+import { A2AResultReassembler, extractIdsFromResponse } from './a2aUtils.js';
+import type { AuthenticationHandler } from '@a2a-js/sdk/client';
+import { A2AAuthProviderFactory } from './auth-provider/factory.js';
+import { A2AAgentError } from './a2a-errors.js';
+import { debugLogger } from '../utils/debugLogger.js';
+import type { MessageBus } from '../confirmation-bus/message-bus.js';
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function isAbortLikeError(err: unknown): boolean {
+  return err instanceof Error && err.name === 'AbortError';
+}
+
+// ---------------------------------------------------------------------------
+// RemoteSubagentProtocol
+// ---------------------------------------------------------------------------
+
+class RemoteSubagentProtocol implements AgentProtocol {
+  private _events: AgentEvent[] = [];
+  private _subscribers = new Set<(event: AgentEvent) => void>();
+  private _streamId: string = randomUUID();
+  private _eventCounter = 0;
+  private _agentStartEmitted = false;
+  private _agentEndEmitted = false;
+  private _activeStreamId: string | undefined;
+  private _abortController = new AbortController();
+
+  // Session state persisted across sends (mirrors RemoteAgentInvocation)
+  private contextId: string | undefined;
+  private taskId: string | undefined;
+  private authHandler: AuthenticationHandler | undefined;
+
+  // Agent display name (for SubagentProgress construction)
+  private readonly _agentName: string;
+
+  // Latest SubagentProgress — updated per chunk, used for error recovery
+  private _latestProgress: SubagentProgress | undefined;
+
+  // Result promise wiring
+  private _resultResolve!: (result: ToolResult) => void;
+  private _resultReject!: (err: unknown) => void;
+  private readonly _resultPromise: Promise<ToolResult>;
+
+  constructor(
+    private readonly definition: RemoteAgentDefinition,
+    private readonly context: AgentLoopContext,
+    _messageBus: MessageBus,
+  ) {
+    this._agentName = definition.displayName ?? definition.name;
+    this._resultPromise = new Promise<ToolResult>((resolve, reject) => {
+      this._resultResolve = resolve;
+      this._resultReject = reject;
+    });
+
+    // Restore persisted session state (mirrors static map in RemoteAgentInvocation)
+    const priorState = RemoteSubagentProtocol._sessionState.get(
+      definition.name,
+    );
+    if (priorState) {
+      this.contextId = priorState.contextId;
+      this.taskId = priorState.taskId;
+    }
+  }
+
+  // Per-agent session state, mirrors RemoteAgentInvocation.sessionState
+  private static readonly _sessionState = new Map<
+    string,
+    { contextId?: string; taskId?: string }
+  >();
+
+  // ---------------------------------------------------------------------------
+  // AgentProtocol interface
+  // ---------------------------------------------------------------------------
+
+  get events(): readonly AgentEvent[] {
+    return this._events;
+  }
+
+  subscribe(callback: (event: AgentEvent) => void): Unsubscribe {
+    this._subscribers.add(callback);
+    return () => {
+      this._subscribers.delete(callback);
+    };
+  }
+
+  async send(payload: AgentSend): Promise<{ streamId: string | null }> {
+    if ('message' in payload && payload.message) {
+      if (this._activeStreamId) {
+        throw new Error(
+          'RemoteSubagentProtocol.send() cannot be called while a stream is active.',
+        );
+      }
+
+      const query =
+        payload.message
+          .filter((p): p is ContentPart & { type: 'text' } => p.type === 'text')
+          .map((p) => p.text)
+          .join('') || DEFAULT_QUERY_STRING;
+
+      this._beginNewStream();
+      const streamId = this._streamId;
+
+      setTimeout(() => {
+        void this._runStreamInBackground(query);
+      }, 0);
+
+      return { streamId };
+    }
+
+    // update/action/elicitations not used for remote agents
+    return { streamId: null };
+  }
+
+  async abort(): Promise<void> {
+    this._abortController.abort();
+  }
+
+  // ---------------------------------------------------------------------------
+  // Protocol-specific: result access
+  // ---------------------------------------------------------------------------
+
+  getResult(): Promise<ToolResult> {
+    return this._resultPromise;
+  }
+
+  getLatestProgress(): SubagentProgress | undefined {
+    return this._latestProgress;
+  }
+
+  // ---------------------------------------------------------------------------
+  // Core: A2A streaming
+  // ---------------------------------------------------------------------------
+
+  private _beginNewStream(): void {
+    this._streamId = randomUUID();
+    this._eventCounter = 0;
+    this._abortController = new AbortController();
+    this._agentStartEmitted = false;
+    this._agentEndEmitted = false;
+    this._activeStreamId = this._streamId;
+  }
+
+  private async _runStreamInBackground(query: string): Promise<void> {
+    this._ensureAgentStart();
+    try {
+      await this._runStream(query);
+    } catch (err: unknown) {
+      // Save state before rejecting so callers see updated contextId/taskId
+      // immediately after the returned promise settles.
+      this._saveSessionState();
+      if (this._abortController.signal.aborted || isAbortLikeError(err)) {
+        this._ensureAgentEnd('aborted');
+        this._resultReject(err);
+      } else {
+        this._emitErrorAndAgentEnd(err);
+        this._resultReject(err);
+      }
+      this._clearActiveStream();
+    }
+  }
+
+  private _saveSessionState(): void {
+    RemoteSubagentProtocol._sessionState.set(this.definition.name, {
+      contextId: this.contextId,
+      taskId: this.taskId,
+    });
+  }
+
+  private async _runStream(query: string): Promise<void> {
+    const clientManager = this.context.config.getA2AClientManager();
+    if (!clientManager) {
+      throw new Error(
+        `RemoteSubagentProtocol: A2AClientManager not available for '${this.definition.name}'.`,
+      );
+    }
+
+    const authHandler = await this._getAuthHandler();
+    if (!clientManager.getClient(this.definition.name)) {
+      await clientManager.loadAgent(
+        this.definition.name,
+        getAgentCardLoadOptions(this.definition),
+        authHandler,
+      );
+    }
+
+    const reassembler = new A2AResultReassembler();
+    let prevText = '';
+
+    const stream = clientManager.sendMessageStream(
+      this.definition.name,
+      query,
+      {
+        contextId: this.contextId,
+        taskId: this.taskId,
+        signal: this._abortController.signal,
+      },
+    );
+
+    for await (const chunk of stream) {
+      if (this._abortController.signal.aborted) {
+        this._saveSessionState();
+        this._finishStream('aborted');
+        this._resultReject(new Error('Operation aborted'));
+        this._clearActiveStream();
+        return;
+      }
+
+      reassembler.update(chunk);
+
+      const {
+        contextId: newContextId,
+        taskId: newTaskId,
+        clearTaskId,
+      } = extractIdsFromResponse(chunk);
+      if (newContextId) this.contextId = newContextId;
+      this.taskId = clearTaskId ? undefined : (newTaskId ?? this.taskId);
+
+      // Update latest progress snapshot (for invocation's error recovery)
+      this._latestProgress = {
+        isSubagentProgress: true,
+        agentName: this._agentName,
+        state: 'running',
+        recentActivity: reassembler.toActivityItems(),
+        result: reassembler.toString(),
+      };
+
+      // Emit delta as a message event
+      const currentText = reassembler.toString();
+      const delta = currentText.slice(prevText.length);
+      if (delta) {
+        this._emit([
+          this._makeEvent('message', {
+            role: 'agent',
+            content: [{ type: 'text', text: currentText }],
+          }),
+        ]);
+        prevText = currentText;
+      }
+    }
+
+    const finalOutput = reassembler.toString();
+    debugLogger.debug(
+      `[RemoteSubagentProtocol] ${this.definition.name} finished, output length: ${finalOutput.length}`,
+    );
+
+    const finalProgress: SubagentProgress = {
+      isSubagentProgress: true,
+      agentName: this._agentName,
+      state: 'completed',
+      result: finalOutput,
+      recentActivity: reassembler.toActivityItems(),
+    };
+    this._latestProgress = finalProgress;
+
+    this._finishStream('completed');
+
+    // Save state before resolving so callers see updated contextId/taskId
+    // immediately after the returned promise settles.
+    this._saveSessionState();
+    this._resultResolve({
+      llmContent: [{ text: finalOutput }],
+      returnDisplay: finalProgress,
+    });
+  }
+
+  private async _getAuthHandler(): Promise<AuthenticationHandler | undefined> {
+    if (this.authHandler) return this.authHandler;
+    if (!this.definition.auth) return undefined;
+
+    const targetUrl = getRemoteAgentTargetUrl(this.definition);
+    const provider = await A2AAuthProviderFactory.create({
+      authConfig: this.definition.auth,
+      agentName: this.definition.name,
+      targetUrl,
+      agentCardUrl: this.definition.agentCardUrl,
+    });
+    if (!provider) {
+      throw new Error(
+        `Failed to create auth provider for agent '${this.definition.name}'`,
+      );
+    }
+    this.authHandler = provider;
+    return this.authHandler;
+  }
+
+  // ---------------------------------------------------------------------------
+  // Internal helpers
+  // ---------------------------------------------------------------------------
+
+  private _emit(events: AgentEvent[]): void {
+    if (events.length === 0) return;
+    const subscribers = [...this._subscribers];
+    for (const event of events) {
+      this._events.push(event);
+      if (event.type === 'agent_end') {
+        this._agentEndEmitted = true;
+      }
+      for (const sub of subscribers) {
+        sub(event);
+      }
+    }
+  }
+
+  private _clearActiveStream(): void {
+    this._activeStreamId = undefined;
+  }
+
+  private _ensureAgentStart(): void {
+    if (!this._agentStartEmitted) {
+      this._agentStartEmitted = true;
+      this._emit([this._makeEvent('agent_start', {})]);
+    }
+  }
+
+  private _ensureAgentEnd(reason: StreamEndReason = 'completed'): void {
+    if (!this._agentEndEmitted && this._agentStartEmitted) {
+      this._emit([this._makeEvent('agent_end', { reason })]);
+    }
+  }
+
+  private _finishStream(
+    reason: StreamEndReason,
+    data?: Record<string, unknown>,
+  ): void {
+    if (data && !this._agentEndEmitted) {
+      this._emit([this._makeEvent('agent_end', { reason, data })]);
+      this._agentEndEmitted = true;
+    } else {
+      this._ensureAgentEnd(reason);
+    }
+    this._clearActiveStream();
+  }
+
+  private _emitErrorAndAgentEnd(err: unknown): void {
+    const message = this._formatError(err);
+    this._ensureAgentStart();
+
+    const meta: Record<string, unknown> = {};
+    if (err instanceof Error) {
+      meta['errorName'] = err.constructor.name;
+    }
+
+    this._emit([
+      this._makeEvent('error', {
+        status: 'INTERNAL',
+        message,
+        fatal: true,
+        ...(Object.keys(meta).length > 0 ? { _meta: meta } : {}),
+      }),
+    ]);
+    this._ensureAgentEnd('failed');
+  }
+
+  private _formatError(error: unknown): string {
+    if (error instanceof A2AAgentError) {
+      return error.userMessage;
+    }
+    return `Error calling remote agent: ${error instanceof Error ? error.message : String(error)}`;
+  }
+
+  private _nextEventFields() {
+    return {
+      id: `${this._streamId}-${this._eventCounter++}`,
+      timestamp: new Date().toISOString(),
+      streamId: this._streamId,
+    };
+  }
+
+  private _makeEvent<T extends AgentEvent['type']>(
+    type: T,
+    payload: Omit<AgentEvent<T>, 'id' | 'timestamp' | 'streamId' | 'type'>,
+  ): AgentEvent {
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
+    return {
+      ...this._nextEventFields(),
+      type,
+      ...payload,
+    } as AgentEvent;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Public export
+// ---------------------------------------------------------------------------
+
+export class RemoteSubagentSession extends AgentSession {
+  private readonly _remoteProtocol: RemoteSubagentProtocol;
+
+  constructor(
+    definition: RemoteAgentDefinition,
+    context: AgentLoopContext,
+    messageBus: MessageBus,
+  ) {
+    const protocol = new RemoteSubagentProtocol(
+      definition,
+      context,
+      messageBus,
+    );
+    super(protocol);
+    this._remoteProtocol = protocol;
+  }
+
+  /**
+   * Returns the ToolResult once the remote agent stream completes.
+   * Used by RemoteAgentInvocation to return the result.
+   */
+  getResult(): Promise<ToolResult> {
+    return this._remoteProtocol.getResult();
+  }
+
+  /**
+   * Returns the most recent SubagentProgress snapshot, updated per streaming
+   * chunk. Useful for constructing error progress when getResult() rejects.
+   */
+  getLatestProgress(): SubagentProgress | undefined {
+    return this._remoteProtocol.getLatestProgress();
+  }
+
+  /**
+   * Convenience: start execution with a query string.
+   * Equivalent to send({message: [{type:'text', text: query}]}).
+   */
+  async startWithQuery(query: string): Promise<{ streamId: string | null }> {
+    return this.send({ message: [{ type: 'text', text: query }] });
+  }
+}

packages/core/src/agents/subagent-tool-wrapper.test.ts

@@ -139,6 +139,7 @@ describe('SubagentToolWrapper', () => {
         mockMessageBus,
         mockDefinition.name,
         mockDefinition.displayName,
+        undefined, // onAgentEvent (not set when wrapper has no onAgentEvent)
       );
     });
 
@@ -164,6 +165,7 @@ describe('SubagentToolWrapper', () => {
         specificMessageBus,
         mockDefinition.name,
         mockDefinition.displayName,
+        undefined, // onAgentEvent (not set when wrapper has no onAgentEvent)
       );
     });
 

packages/core/src/agents/subagent-tool-wrapper.ts

@@ -18,6 +18,7 @@ import { RemoteAgentInvocation } from './remote-invocation.js';
 import { BrowserAgentInvocation } from './browser/browserAgentInvocation.js';
 import { BROWSER_AGENT_NAME } from './browser/browserAgentDefinition.js';
 import type { MessageBus } from '../confirmation-bus/message-bus.js';
+import type { AgentEvent } from '../agent/types.js';
 
 /**
  * A tool wrapper that dynamically exposes a subagent as a standard,
@@ -41,6 +42,7 @@ export class SubagentToolWrapper extends BaseDeclarativeTool<
     private readonly definition: AgentDefinition,
     private readonly context: AgentLoopContext,
     messageBus: MessageBus,
+    private readonly onAgentEvent?: (event: AgentEvent) => void,
   ) {
     super(
       definition.name,
@@ -80,6 +82,7 @@ export class SubagentToolWrapper extends BaseDeclarativeTool<
         effectiveMessageBus,
         _toolName,
         _toolDisplayName,
+        this.onAgentEvent,
       );
     }
 
@@ -101,6 +104,7 @@ export class SubagentToolWrapper extends BaseDeclarativeTool<
       effectiveMessageBus,
       _toolName,
       _toolDisplayName,
+      this.onAgentEvent,
     );
   }
 }

packages/core/src/agents/subagent-tool.test.ts

@@ -121,6 +121,7 @@ describe('SubAgentInvocation', () => {
       testDefinition,
       mockConfig,
       mockMessageBus,
+      undefined, // onAgentEvent (not set on SubAgentInvocation)
     );
   });
 
@@ -165,6 +166,7 @@ describe('SubAgentInvocation', () => {
       testRemoteDefinition,
       mockConfig,
       mockMessageBus,
+      undefined, // onAgentEvent (not set on SubAgentInvocation)
     );
   });
 

packages/core/src/agents/subagent-tool.ts

@@ -19,6 +19,7 @@ import { type AgentLoopContext } from '../config/agent-loop-context.js';
 import type { MessageBus } from '../confirmation-bus/message-bus.js';
 import type { AgentDefinition, AgentInputs } from './types.js';
 import { SubagentToolWrapper } from './subagent-tool-wrapper.js';
+import type { AgentEvent } from '../agent/types.js';
 import { SchemaValidator } from '../utils/schemaValidator.js';
 import { formatUserHintsForModel } from '../utils/fastAckHelper.js';
 import { runInDevTraceSpan } from '../telemetry/trace.js';
@@ -33,6 +34,7 @@ export class SubagentTool extends BaseDeclarativeTool<AgentInputs, ToolResult> {
     private readonly definition: AgentDefinition,
     private readonly context: AgentLoopContext,
     messageBus: MessageBus,
+    private readonly onAgentEvent?: (event: AgentEvent) => void,
   ) {
     const inputSchema = definition.inputConfig.inputSchema;
 
@@ -116,6 +118,7 @@ export class SubagentTool extends BaseDeclarativeTool<AgentInputs, ToolResult> {
       messageBus,
       _toolName,
       _toolDisplayName,
+      this.onAgentEvent,
     );
   }
 }
@@ -130,6 +133,7 @@ class SubAgentInvocation extends BaseToolInvocation<AgentInputs, ToolResult> {
     messageBus: MessageBus,
     _toolName?: string,
     _toolDisplayName?: string,
+    private readonly onAgentEvent?: (event: AgentEvent) => void,
   ) {
     super(
       params,
@@ -229,6 +233,7 @@ class SubAgentInvocation extends BaseToolInvocation<AgentInputs, ToolResult> {
       definition,
       this.context,
       this.messageBus,
+      this.onAgentEvent,
     );
 
     return wrapper.build(agentArgs);

packages/core/src/index.ts

@@ -187,6 +187,8 @@ export * from './agents/agent-scheduler.js';
 // Export agent session interface
 export * from './agent/agent-session.js';
 export * from './agent/legacy-agent-session.js';
+export { LocalSubagentSession } from './agents/local-subagent-protocol.js';
+export { RemoteSubagentSession } from './agents/remote-subagent-protocol.js';
 export * from './agent/event-translator.js';
 export * from './agent/content-utils.js';
 // Agent event types — namespaced to avoid collisions with existing exports