addressing christian's feedback

packages/core/src/agents/agent.test.ts

@@ -0,0 +1,84 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { Agent } from './agent.js';
+import { AgentSession } from './session.js';
+import { makeFakeConfig } from '../test-utils/config.js';
+import { type AgentConfig } from './types.js';
+
+vi.mock('./session.js', () => ({
+  AgentSession: vi.fn().mockImplementation(() => ({
+    prompt: vi.fn().mockImplementation(async function* () {
+      yield { type: 'agent_start', value: { sessionId: 'test-session' } };
+      yield {
+        type: 'agent_finish',
+        value: { sessionId: 'test-session', totalTurns: 1 },
+      };
+    }),
+  })),
+}));
+
+describe('Agent', () => {
+  let mockConfig: ReturnType<typeof makeFakeConfig>;
+  const agentConfig: AgentConfig = {
+    name: 'TestAgent',
+    systemInstruction: 'You are a test agent.',
+  };
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockConfig = makeFakeConfig();
+    vi.spyOn(mockConfig, 'getSessionId').mockReturnValue('global-session-id');
+  });
+
+  it('should create an AgentSession', () => {
+    const agent = new Agent(agentConfig, mockConfig);
+    const session = agent.createSession('custom-session-id');
+
+    expect(session).toBeDefined();
+    expect(AgentSession).toHaveBeenCalledWith(
+      'custom-session-id',
+      agentConfig,
+      mockConfig,
+    );
+  });
+
+  it('should use global session ID if none provided to createSession', () => {
+    const agent = new Agent(agentConfig, mockConfig);
+    agent.createSession();
+
+    expect(AgentSession).toHaveBeenCalledWith(
+      'global-session-id',
+      agentConfig,
+      mockConfig,
+    );
+  });
+
+  it('should pass config and runtime to the session', () => {
+    const agent = new Agent(agentConfig, mockConfig);
+    agent.createSession('test-id');
+
+    expect(AgentSession).toHaveBeenCalledWith(
+      'test-id',
+      agentConfig,
+      mockConfig,
+    );
+  });
+
+  it('should prompt through a new session', async () => {
+    const agent = new Agent(agentConfig, mockConfig);
+    const events = [];
+    for await (const event of agent.prompt('Hello')) {
+      events.push(event);
+    }
+
+    expect(events).toHaveLength(2);
+    expect(events[0].type).toBe('agent_start');
+    expect(events[1].type).toBe('agent_finish');
+    expect(AgentSession).toHaveBeenCalled();
+  });
+});

packages/core/src/agents/agent.ts

@@ -0,0 +1,41 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { type Part } from '@google/genai';
+import { type Config } from '../config/config.js';
+import { type AgentEvent, type AgentConfig } from './types.js';
+import { AgentSession } from './session.js';
+
+/**
+ * The Agent class is a factory for creating stateful AgentSessions.
+ * This represents a configured agent template.
+ */
+export class Agent {
+  constructor(
+    private readonly config: AgentConfig,
+    private readonly runtime: Config,
+  ) {}
+
+  /**
+   * Creates a new stateful session for interacting with the agent.
+   */
+  createSession(sessionId?: string): AgentSession {
+    const id = sessionId ?? this.runtime.getSessionId();
+    return new AgentSession(id, this.config, this.runtime);
+  }
+
+  /**
+   * Helper to quickly run a single prompt and get the results.
+   */
+  prompt(
+    input: string | Part[],
+    sessionId?: string,
+    signal?: AbortSignal,
+  ): AsyncIterable<AgentEvent> {
+    const session = this.createSession(sessionId);
+    return session.prompt(input, signal);
+  }
+}

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

@@ -546,11 +546,7 @@ export class LocalAgentExecutor<TOutput extends z.ZodTypeAny> {
       // === UNIFIED RECOVERY BLOCK ===
       // Only attempt recovery if it's a known recoverable reason.
       // We don't recover from GOAL (already done) or ABORTED (user cancelled).
-      if (
-        terminateReason !== AgentTerminateMode.ERROR &&
-        terminateReason !== AgentTerminateMode.ABORTED &&
-        terminateReason !== AgentTerminateMode.GOAL
-      ) {
+      if (this.isRecoverableReason(terminateReason)) {
         const recoveryResult = await this.executeFinalWarningTurn(
           chat,
           turnCounter, // Use current turnCounter for the recovery attempt
@@ -1256,6 +1252,23 @@ Important Rules:
     return null;
   }
 
+  /**
+   * Returns true if the agent should attempt a recovery turn for the given reason.
+   */
+  private isRecoverableReason(
+    reason: AgentTerminateMode,
+  ): reason is
+    | AgentTerminateMode.TIMEOUT
+    | AgentTerminateMode.MAX_TURNS
+    | AgentTerminateMode.ERROR_NO_COMPLETE_TASK_CALL {
+    return (
+      reason !== AgentTerminateMode.ERROR &&
+      reason !== AgentTerminateMode.ABORTED &&
+      reason !== AgentTerminateMode.GOAL &&
+      reason !== AgentTerminateMode.LOOP
+    );
+  }
+
   /** Emits an activity event to the configured callback. */
   private emitActivity(
     type: SubagentActivityEvent['type'],

packages/core/src/agents/session.test.ts

@@ -0,0 +1,524 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { AgentSession } from './session.js';
+import { makeFakeConfig } from '../test-utils/config.js';
+import {
+  type AgentConfig,
+  AgentTerminateMode,
+  type AgentEvent,
+  type AgentFinishEvent,
+  type ToolSuiteStartEvent,
+  type ToolCallFinishEvent,
+} from './types.js';
+import { Scheduler } from '../scheduler/scheduler.js';
+import { GeminiEventType, CompressionStatus } from '../core/turn.js';
+import { ChatCompressionService } from '../services/chatCompressionService.js';
+import {
+  MessageBusType,
+  type ToolCallsUpdateMessage,
+} from '../confirmation-bus/types.js';
+import {
+  CoreToolCallStatus,
+  type ToolCallRequestInfo,
+} from '../scheduler/types.js';
+import { type ResumedSessionData } from '../services/chatRecordingService.js';
+import { type GeminiClient } from '../core/client.js';
+
+vi.mock('../core/client.js');
+vi.mock('../scheduler/scheduler.js');
+vi.mock('../services/chatCompressionService.js');
+
+describe('AgentSession', () => {
+  let mockConfig: ReturnType<typeof makeFakeConfig>;
+  let mockClient: {
+    sendMessageStream: ReturnType<typeof vi.fn>;
+    isInitialized: ReturnType<typeof vi.fn>;
+    initialize: ReturnType<typeof vi.fn>;
+    getChat: ReturnType<typeof vi.fn>;
+    getCurrentSequenceModel: ReturnType<typeof vi.fn>;
+    getHistory: ReturnType<typeof vi.fn>;
+    resumeChat: ReturnType<typeof vi.fn>;
+  };
+  let mockScheduler: {
+    schedule: ReturnType<typeof vi.fn>;
+  };
+  let mockCompressionService: {
+    compress: ReturnType<typeof vi.fn>;
+  };
+  let session: AgentSession;
+  const agentConfig: AgentConfig = {
+    name: 'TestAgent',
+    systemInstruction: 'You are a test agent.',
+    capabilities: { compression: true },
+  };
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockConfig = makeFakeConfig();
+
+    mockClient = {
+      sendMessageStream: vi.fn(),
+      isInitialized: vi.fn().mockReturnValue(false),
+      initialize: vi.fn().mockResolvedValue(undefined),
+      getChat: vi.fn().mockReturnValue({
+        recordCompletedToolCalls: vi.fn(),
+        setHistory: vi.fn(),
+        getHistory: vi.fn().mockReturnValue([]),
+        setSystemInstruction: vi.fn(),
+      }),
+      getCurrentSequenceModel: vi.fn().mockReturnValue('test-model'),
+      getHistory: vi.fn().mockReturnValue([]),
+      resumeChat: vi.fn(),
+    };
+
+    mockScheduler = {
+      schedule: vi.fn(),
+    };
+
+    mockCompressionService = {
+      compress: vi.fn().mockResolvedValue({
+        newHistory: null,
+        info: { compressionStatus: CompressionStatus.NOOP },
+      }),
+    };
+
+    vi.spyOn(mockConfig, 'getGeminiClient').mockReturnValue(
+      mockClient as unknown as GeminiClient,
+    );
+    vi.mocked(Scheduler).mockImplementation(
+      (options) =>
+        ({
+          ...mockScheduler,
+          schedulerId: (options as { schedulerId: string }).schedulerId,
+        }) as unknown as Scheduler,
+    );
+    vi.mocked(ChatCompressionService).mockImplementation(
+      () => mockCompressionService as unknown as ChatCompressionService,
+    );
+
+    session = new AgentSession('test-session', agentConfig, mockConfig);
+  });
+
+  it('should emit agent_start and agent_finish', async () => {
+    mockClient.sendMessageStream.mockImplementation(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Hello' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    const events = [];
+    for await (const event of session.prompt('Hi')) {
+      events.push(event);
+    }
+
+    const finishEvent = events[events.length - 1] as AgentFinishEvent;
+    expect(events[0].type).toBe('agent_start');
+    expect(finishEvent.type).toBe('agent_finish');
+    expect(finishEvent.value.reason).toBe(AgentTerminateMode.GOAL);
+    expect(mockClient.sendMessageStream).toHaveBeenCalled();
+  });
+
+  it('should handle tool calls and execute them via MessageBus updates', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'call1', name: 'test_tool', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Tool executed' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    const toolResponse = {
+      response: {
+        callId: 'call1',
+        responseParts: [
+          {
+            functionResponse: {
+              name: 'test_tool',
+              response: { ok: true },
+              id: 'call1',
+            },
+          },
+        ],
+      },
+    };
+
+    mockScheduler.schedule.mockImplementation(async () => {
+      const bus = mockConfig.getMessageBus();
+      const schedulerId = (session as unknown as { schedulerId: string })
+        .schedulerId;
+
+      await bus.publish({
+        type: MessageBusType.TOOL_CALLS_UPDATE,
+        schedulerId,
+        toolCalls: [
+          {
+            request: { callId: 'call1', name: 'test_tool', args: {} },
+            status: CoreToolCallStatus.Executing,
+            schedulerId,
+          } as unknown as ToolCallsUpdateMessage['toolCalls'][number],
+        ],
+      });
+
+      await bus.publish({
+        type: MessageBusType.TOOL_CALLS_UPDATE,
+        schedulerId,
+        toolCalls: [
+          {
+            request: { callId: 'call1', name: 'test_tool', args: {} },
+            status: CoreToolCallStatus.Success,
+            response: toolResponse.response,
+            schedulerId,
+          } as unknown as ToolCallsUpdateMessage['toolCalls'][number],
+        ],
+      });
+
+      return [toolResponse];
+    });
+
+    const events = [];
+    for await (const event of session.prompt('Run tool')) {
+      events.push(event);
+    }
+
+    expect(mockClient.sendMessageStream).toHaveBeenCalledTimes(2);
+    expect(mockScheduler.schedule).toHaveBeenCalledTimes(1);
+
+    const callStart = events.find((e) => e.type === 'tool_call_start');
+    const callFinish = events.find((e) => e.type === 'tool_call_finish');
+    expect(callStart).toBeDefined();
+    expect(callFinish).toBeDefined();
+    expect((callFinish as ToolCallFinishEvent).value.callId).toBe('call1');
+  });
+
+  it('should handle multiple consecutive ReAct turns', async () => {
+    // Turn 1: tool1
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'c1', name: 'tool1', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+    // Turn 2: tool2
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'c2', name: 'tool2', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+    // Turn 3: final content
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.Content, value: 'All done' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    mockScheduler.schedule.mockImplementation(async (calls) =>
+      (calls as ToolCallRequestInfo[]).map((c) => ({
+        response: { callId: c.callId, responseParts: [] },
+      })),
+    );
+
+    const events = [];
+    for await (const event of session.prompt('Start multistep')) {
+      events.push(event);
+    }
+
+    expect(mockClient.sendMessageStream).toHaveBeenCalledTimes(3);
+    expect(mockScheduler.schedule).toHaveBeenCalledTimes(2);
+    expect(events.filter((e) => e.type === 'tool_suite_start')).toHaveLength(2);
+  });
+
+  it('should handle parallel tool calls in a single turn', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'p1', name: 'toolA', args: {} },
+      };
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'p2', name: 'toolB', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Parallel done' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    mockScheduler.schedule.mockImplementation(async (calls) =>
+      (calls as ToolCallRequestInfo[]).map((c) => ({
+        response: { callId: c.callId, responseParts: [] },
+      })),
+    );
+
+    const events = [];
+    for await (const event of session.prompt('Parallel')) {
+      events.push(event);
+    }
+
+    const suiteStart = events.find(
+      (e) => e.type === 'tool_suite_start',
+    ) as ToolSuiteStartEvent;
+    expect(suiteStart.value.count).toBe(2);
+    expect(mockScheduler.schedule).toHaveBeenCalledTimes(1);
+    expect(mockScheduler.schedule).toHaveBeenCalledWith(
+      expect.arrayContaining([
+        expect.objectContaining({ callId: 'p1' }),
+        expect.objectContaining({ callId: 'p2' }),
+      ]),
+      expect.anything(),
+    );
+  });
+
+  it('should resume from session data', async () => {
+    const resumeData = {
+      conversation: {
+        messages: [{ type: 'user', content: 'Hello' }],
+      },
+    } as unknown as ResumedSessionData;
+
+    await session.resume(resumeData);
+
+    expect(mockClient.resumeChat).toHaveBeenCalledWith(
+      expect.arrayContaining([
+        expect.objectContaining({ role: 'user', parts: [{ text: 'Hello' }] }),
+      ]),
+      resumeData,
+    );
+  });
+
+  it('should handle model stream errors gracefully', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield* []; // satisfy require-yield
+      throw new Error('Model connection lost');
+    });
+
+    const events: AgentEvent[] = [];
+    try {
+      for await (const event of session.prompt('Error test')) {
+        events.push(event);
+      }
+    } catch (_e) {
+      // Expected error
+    }
+
+    const finishEvent = events.find(
+      (e) => e.type === 'agent_finish',
+    ) as AgentFinishEvent;
+    expect(finishEvent).toBeDefined();
+    expect(finishEvent.value.reason).toBe(AgentTerminateMode.ERROR);
+    expect(finishEvent.value.message).toBe('Model connection lost');
+  });
+
+  it('should ignore MessageBus updates from other schedulers', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'call1', name: 'test_tool', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Done' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    mockScheduler.schedule.mockImplementation(async () => {
+      const bus = mockConfig.getMessageBus();
+
+      // Update from ANOTHER scheduler
+      await bus.publish({
+        type: MessageBusType.TOOL_CALLS_UPDATE,
+        schedulerId: 'different-scheduler',
+        toolCalls: [
+          {
+            request: { callId: 'call1', name: 'test_tool', args: {} },
+            status: CoreToolCallStatus.Executing,
+            schedulerId: 'different-scheduler',
+          } as unknown as ToolCallsUpdateMessage['toolCalls'][number],
+        ],
+      });
+
+      return [{ response: { callId: 'call1', responseParts: [] } }];
+    });
+
+    const events = [];
+    for await (const event of session.prompt('Isolation test')) {
+      events.push(event);
+    }
+
+    // Should NOT see tool_call_start because it was from a different schedulerId
+    expect(events.find((e) => e.type === 'tool_call_start')).toBeUndefined();
+  });
+
+  it('should terminate with LOOP when loop is detected by model', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.LoopDetected };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    const events = [];
+    for await (const event of session.prompt('Loop')) {
+      events.push(event);
+    }
+    const finishEvent = events.find(
+      (e) => e.type === 'agent_finish',
+    ) as AgentFinishEvent;
+    expect(finishEvent.value.reason).toBe(AgentTerminateMode.LOOP);
+    expect(finishEvent.value.message).toContain('Loop detected');
+  });
+
+  it('should handle Part[] input correctly', async () => {
+    mockClient.sendMessageStream.mockImplementationOnce(async function* () {
+      yield { type: GeminiEventType.Content, value: 'I see parts' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    const input = [{ text: 'Hello' }, { text: 'World' }];
+    for await (const _ of session.prompt(input)) {
+      // consume
+    }
+
+    expect(mockClient.sendMessageStream).toHaveBeenCalledWith(
+      input,
+      expect.anything(),
+      expect.anything(),
+      undefined,
+      false,
+      input,
+    );
+  });
+
+  it('should trigger compression if enabled', async () => {
+    mockClient.sendMessageStream.mockImplementation(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Done' };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    for await (const _ of session.prompt('Compress me')) {
+      // consume stream to trigger compression
+    }
+
+    expect(mockCompressionService.compress).toHaveBeenCalled();
+  });
+
+  it('should respect abort signal', async () => {
+    const controller = new AbortController();
+    mockClient.sendMessageStream.mockImplementation(async function* () {
+      yield { type: GeminiEventType.Content, value: 'Thinking...' };
+      controller.abort();
+      yield { type: GeminiEventType.Content, value: 'Still thinking...' };
+    });
+
+    const events = [];
+    for await (const event of session.prompt('Long task', controller.signal)) {
+      events.push(event);
+    }
+
+    const finishEvent = events[events.length - 1] as AgentFinishEvent;
+    expect(finishEvent.type).toBe('agent_finish');
+    expect(finishEvent.value.reason).toBe(AgentTerminateMode.ABORTED);
+  });
+
+  it('should apply systemInstruction from AgentConfig', async () => {
+    const customConfig: AgentConfig = {
+      ...agentConfig,
+      systemInstruction: 'You are a helpful assistant.',
+    };
+
+    // Mock isInitialized to true so constructor can set it
+    mockClient.isInitialized.mockReturnValue(true);
+    const mockChat = { setSystemInstruction: vi.fn() };
+    mockClient.getChat.mockReturnValue(mockChat);
+
+    // Re-create to trigger constructor logic
+    new AgentSession('test-session-3', customConfig, mockConfig);
+    expect(mockChat.setSystemInstruction).toHaveBeenCalledWith(
+      'You are a helpful assistant.',
+    );
+  });
+
+  it('should abort internal operations if caller stops iterating', async () => {
+    let internalSignal: AbortSignal | undefined;
+
+    mockClient.sendMessageStream.mockImplementation(async function* (
+      _parts: unknown,
+      signal: AbortSignal,
+    ) {
+      internalSignal = signal;
+      yield { type: GeminiEventType.Content, value: 'Part 1' };
+      yield { type: GeminiEventType.Content, value: 'Part 2' };
+    });
+
+    const promptStream = session.prompt('Test cancellation');
+    const iterator = promptStream[Symbol.asyncIterator]();
+
+    const firstEvent = await iterator.next();
+    expect(firstEvent.value?.type).toBe('agent_start');
+
+    const secondEvent = await iterator.next(); // content Part 1
+    expect(secondEvent.value?.type).toBe(GeminiEventType.Content);
+
+    // Caller stops here and closes the generator
+    if (iterator.return) {
+      await iterator.return();
+    }
+
+    expect(internalSignal?.aborted).toBe(true);
+  });
+
+  it('should respect maxTurns from config', async () => {
+    const customSession = new AgentSession(
+      'test-session-2',
+      { ...agentConfig, maxTurns: 2 },
+      mockConfig,
+    );
+
+    mockClient.sendMessageStream.mockImplementation(async function* () {
+      yield {
+        type: GeminiEventType.ToolCallRequest,
+        value: { callId: 'call', name: 'test_tool', args: {} },
+      };
+      yield { type: GeminiEventType.Finished, value: { reason: 'STOP' } };
+    });
+
+    mockScheduler.schedule.mockResolvedValue([
+      {
+        response: {
+          callId: 'call',
+          responseParts: [
+            {
+              functionResponse: {
+                name: 'test_tool',
+                response: { ok: true },
+                id: 'call',
+              },
+            },
+          ],
+        },
+      },
+    ]);
+
+    const events = [];
+    for await (const event of customSession.prompt('Start loop')) {
+      events.push(event);
+    }
+
+    expect(mockScheduler.schedule).toHaveBeenCalledTimes(2);
+
+    const finishEvent = events[events.length - 1] as AgentFinishEvent;
+    expect(finishEvent.type).toBe('agent_finish');
+    expect(finishEvent.value.totalTurns).toBe(2);
+    expect(finishEvent.value.reason).toBe(AgentTerminateMode.MAX_TURNS);
+  });
+});

packages/core/src/agents/session.ts

@@ -0,0 +1,469 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { type Part } from '@google/genai';
+import { type Config } from '../config/config.js';
+import { type GeminiClient } from '../core/client.js';
+import { type AgentEvent, type AgentConfig } from './types.js';
+import { Scheduler } from '../scheduler/scheduler.js';
+import {
+  type ToolCallRequestInfo,
+  type ToolCallResponseInfo,
+  CoreToolCallStatus,
+  type CompletedToolCall,
+} from '../scheduler/types.js';
+import { GeminiEventType, CompressionStatus } from '../core/turn.js';
+import { recordToolCallInteractions } from '../code_assist/telemetry.js';
+import { debugLogger } from '../utils/debugLogger.js';
+import { ToolErrorType } from '../tools/tool-error.js';
+import { ChatCompressionService } from '../services/chatCompressionService.js';
+import { AgentTerminateMode } from './types.js';
+import type { ResumedSessionData } from '../services/chatRecordingService.js';
+import { convertSessionToClientHistory } from '../utils/sessionUtils.js';
+import {
+  MessageBusType,
+  type ToolCallsUpdateMessage,
+} from '../confirmation-bus/types.js';
+
+/** Result of a single model turn in the ReAct loop. */
+export interface ModelTurnResult {
+  /** The specific tool calls requested by the model. */
+  toolCalls: ToolCallRequestInfo[];
+  /** The unified event stream from this model turn. */
+  events: AsyncIterable<AgentEvent>;
+  /** Whether an infinite tool loop was detected. */
+  loopDetected: boolean;
+}
+
+/** Result of executing a batch of tool calls. */
+export interface ToolExecutionResult {
+  /** The response parts from the tool execution to be sent back to the model. */
+  nextParts: Part[];
+  /** Whether execution should stop immediately (e.g. on fatal tool error). */
+  stopExecution: boolean;
+  /** Optional details if execution was stopped. */
+  stopExecutionInfo: ToolCallResponseInfo | undefined;
+}
+
+/**
+ * AgentSession manages the state of a conversation and orchestrates the agent
+ * loop.
+ */
+export class AgentSession {
+  readonly sessionId: string;
+  private readonly client: GeminiClient;
+  private readonly scheduler: Scheduler;
+  private readonly schedulerId: string;
+  private readonly compressionService: ChatCompressionService;
+  private totalTurns = 0;
+  private hasFailedCompressionAttempt = false;
+
+  constructor(
+    sessionId: string,
+    private readonly config: AgentConfig,
+    private readonly runtime: Config,
+  ) {
+    this.sessionId = sessionId;
+    this.client = this.runtime.getGeminiClient();
+    this.schedulerId = `agent-scheduler-${this.sessionId}-${Math.random().toString(36).substring(2, 9)}`;
+    this.scheduler = new Scheduler({
+      config: this.runtime,
+      messageBus: this.runtime.getMessageBus(),
+      getPreferredEditor: () => undefined,
+      schedulerId: this.schedulerId,
+    });
+    this.compressionService = new ChatCompressionService();
+
+    // Ensure system instruction is set from AgentConfig
+    if (this.config.systemInstruction) {
+      if (this.client.isInitialized()) {
+        this.client
+          .getChat()
+          .setSystemInstruction(this.config.systemInstruction);
+      }
+    }
+  }
+
+  /**
+   * Resumes the agent session from persistent storage data.
+   * Hydrates the internal language model client with the previously saved trajectory.
+   *
+   * @param resumedSessionData The raw payload of a previously saved session.
+   */
+  async resume(resumedSessionData: ResumedSessionData): Promise<void> {
+    const clientHistory = convertSessionToClientHistory(
+      resumedSessionData.conversation.messages,
+    );
+    await this.client.resumeChat(clientHistory, resumedSessionData);
+
+    // Re-apply system instruction after resume since resume re-creates the chat
+    if (this.config.systemInstruction) {
+      this.client.getChat().setSystemInstruction(this.config.systemInstruction);
+    }
+  }
+
+  /**
+   * Executes the ReAct loop for a given user input.
+   * Returns an AsyncIterable of events occurring during the session.
+   */
+  async *prompt(
+    input: string | Part[],
+    signal?: AbortSignal,
+  ): AsyncIterable<AgentEvent> {
+    const internalController = new AbortController();
+    const combinedSignal = signal
+      ? AbortSignal.any([signal, internalController.signal])
+      : internalController.signal;
+
+    yield {
+      type: 'agent_start',
+      value: { sessionId: this.sessionId },
+    };
+
+    let terminationReason = AgentTerminateMode.GOAL;
+    let terminationMessage: string | undefined = undefined;
+    let terminationError: unknown | undefined = undefined;
+
+    try {
+      const loop = this._runLoop(input, combinedSignal);
+      for await (const event of loop) {
+        if (event.type === GeminiEventType.LoopDetected) {
+          terminationReason = AgentTerminateMode.LOOP;
+          terminationMessage = 'Loop detected, stopping execution';
+        }
+        yield event;
+      }
+
+      if (combinedSignal.aborted) {
+        terminationReason = AgentTerminateMode.ABORTED;
+      } else if (
+        terminationReason === AgentTerminateMode.GOAL &&
+        this.config.maxTurns &&
+        this.config.maxTurns !== -1 &&
+        this.totalTurns >= this.config.maxTurns
+      ) {
+        // Only set MAX_TURNS if we haven't already hit another reason (like LOOP)
+        // and we are actually at or above the turn limit.
+        terminationReason = AgentTerminateMode.MAX_TURNS;
+        terminationMessage = 'Maximum session turns exceeded.';
+      }
+    } catch (e) {
+      terminationReason = AgentTerminateMode.ERROR;
+      terminationMessage = e instanceof Error ? e.message : String(e);
+      terminationError = e;
+      throw e;
+    } finally {
+      internalController.abort();
+      yield {
+        type: 'agent_finish',
+        value: {
+          sessionId: this.sessionId,
+          totalTurns: this.totalTurns,
+          reason: terminationReason,
+          message: terminationMessage,
+          error: terminationError,
+        },
+      };
+    }
+  }
+
+  /**
+   * Internal generator managing the turn-by-turn ReAct loop.
+   */
+  private async *_runLoop(
+    input: string | Part[],
+    signal: AbortSignal,
+  ): AsyncIterable<AgentEvent> {
+    let currentInput = input;
+    let isContinuation = false;
+    const maxTurns = this.config.maxTurns ?? -1;
+
+    while (maxTurns === -1 || this.totalTurns < maxTurns) {
+      if (signal.aborted) return;
+
+      this.totalTurns++;
+      const promptId = `${this.sessionId}#${this.totalTurns}`;
+
+      if (this.config.capabilities?.compression) {
+        await this.tryCompressChat(promptId);
+      }
+
+      const results = await this.runModelTurn(
+        currentInput,
+        promptId,
+        isContinuation ? undefined : input,
+        signal,
+      );
+
+      for await (const event of results.events) {
+        yield event;
+      }
+
+      if (results.loopDetected) return;
+      if (signal.aborted) return;
+
+      if (results.toolCalls.length > 0) {
+        const toolRun = this._handleToolCalls(results.toolCalls, signal);
+        let toolResults: ToolExecutionResult;
+        while (true) {
+          const { value, done } = await toolRun.next();
+          if (done) {
+            toolResults = value;
+            break;
+          }
+          yield value;
+        }
+
+        if (toolResults.stopExecution || signal.aborted) {
+          if (toolResults.stopExecution && toolResults.stopExecutionInfo) {
+            throw (
+              toolResults.stopExecutionInfo.error ??
+              new Error('Tool execution stopped')
+            );
+          }
+          return;
+        }
+
+        if (maxTurns !== -1 && this.totalTurns >= maxTurns) {
+          return;
+        }
+
+        currentInput = toolResults.nextParts;
+        isContinuation = true;
+      } else {
+        return;
+      }
+    }
+  }
+
+  /**
+   * Orchestrates tool execution turn and yields events.
+   */
+  private async *_handleToolCalls(
+    toolCalls: ToolCallRequestInfo[],
+    signal: AbortSignal,
+  ): AsyncGenerator<AgentEvent, ToolExecutionResult> {
+    const toolRun = this.executeTools(toolCalls, signal);
+    while (true) {
+      const { value, done } = await toolRun.next();
+      if (done) return value;
+      yield value;
+    }
+  }
+
+  /**
+   * Calls the model and yields the event stream.
+   * Collects tool call requests for the next phase.
+   */
+  private async runModelTurn(
+    input: string | Part[],
+    promptId: string,
+    displayContent?: string | Part[],
+    signal?: AbortSignal,
+  ): Promise<ModelTurnResult> {
+    const parts = Array.isArray(input) ? input : [{ text: input }];
+    const toolCalls: ToolCallRequestInfo[] = [];
+    let loopDetected = false;
+
+    // Ensure client is initialized before sending message
+    if (!this.client.isInitialized()) {
+      await this.client.initialize();
+      // Re-apply system instruction after initialization
+      if (this.config.systemInstruction) {
+        this.client
+          .getChat()
+          .setSystemInstruction(this.config.systemInstruction);
+      }
+    }
+
+    const stream = this.client.sendMessageStream(
+      parts,
+      signal ?? new AbortController().signal,
+      promptId,
+      undefined, // maxTurns (client handles its own)
+      false, // isInvalidStreamRetry
+      displayContent,
+    );
+
+    const eventGenerator = async function* (): AsyncIterable<AgentEvent> {
+      for await (const event of stream) {
+        if (event.type === GeminiEventType.ToolCallRequest) {
+          toolCalls.push(event.value);
+        } else if (event.type === GeminiEventType.LoopDetected) {
+          loopDetected = true;
+        }
+        yield event;
+      }
+    };
+
+    const events = eventGenerator();
+
+    return {
+      toolCalls,
+      events,
+      get loopDetected() {
+        return loopDetected;
+      },
+    };
+  }
+
+  /**
+   * Executes a batch of tool calls via the Scheduler.
+   */
+  private async *executeTools(
+    toolCalls: ToolCallRequestInfo[],
+    signal?: AbortSignal,
+  ): AsyncGenerator<AgentEvent, ToolExecutionResult> {
+    yield {
+      type: 'tool_suite_start',
+      value: { count: toolCalls.length },
+    };
+
+    const eventQueue: AgentEvent[] = [];
+    let resolveNext: (() => void) | undefined;
+    let isFinished = false;
+
+    const onToolUpdate = this._createToolUpdateHandler(eventQueue, () =>
+      resolveNext?.(),
+    );
+
+    const messageBus = this.runtime.getMessageBus();
+    messageBus.subscribe(MessageBusType.TOOL_CALLS_UPDATE, onToolUpdate);
+
+    const schedulePromise = this.scheduler.schedule(
+      toolCalls,
+      signal ?? new AbortController().signal,
+    );
+
+    try {
+      while (!isFinished || eventQueue.length > 0) {
+        if (eventQueue.length > 0) {
+          const event = eventQueue.shift();
+          if (event) yield event;
+        } else {
+          const waitNext = new Promise<void>((resolve) => {
+            resolveNext = resolve;
+          });
+          await Promise.race([
+            waitNext,
+            schedulePromise.then(() => {
+              isFinished = true;
+              resolveNext?.();
+            }),
+          ]);
+        }
+      }
+
+      const completedCalls = await schedulePromise;
+
+      yield {
+        type: 'tool_suite_finish',
+        value: { responses: completedCalls.map((c) => c.response) },
+      };
+
+      await this._recordTelemetry(completedCalls);
+
+      const nextParts = completedCalls.flatMap((c) => c.response.responseParts);
+      const stopExecutionInfo = completedCalls.find(
+        (c) => c.response.errorType === ToolErrorType.STOP_EXECUTION,
+      )?.response;
+
+      return {
+        nextParts,
+        stopExecution: !!stopExecutionInfo,
+        stopExecutionInfo,
+      };
+    } finally {
+      messageBus.unsubscribe(MessageBusType.TOOL_CALLS_UPDATE, onToolUpdate);
+    }
+  }
+
+  /**
+   * Creates a handler for MessageBus tool update events.
+   */
+  private _createToolUpdateHandler(
+    eventQueue: AgentEvent[],
+    onNewEvents: () => void,
+  ) {
+    const seenStatuses = new Map<string, CoreToolCallStatus>();
+
+    return (message: ToolCallsUpdateMessage) => {
+      if (message.schedulerId !== this.schedulerId) return;
+
+      for (const call of message.toolCalls) {
+        const prevStatus = seenStatuses.get(call.request.callId);
+        if (prevStatus === call.status) continue;
+
+        if (call.status === CoreToolCallStatus.Executing) {
+          eventQueue.push({ type: 'tool_call_start', value: call.request });
+        } else if (
+          call.status === CoreToolCallStatus.Success ||
+          call.status === CoreToolCallStatus.Error ||
+          call.status === CoreToolCallStatus.Cancelled
+        ) {
+          eventQueue.push({
+            type: 'tool_call_finish',
+            value: call.response,
+          });
+        }
+        seenStatuses.set(call.request.callId, call.status);
+      }
+      onNewEvents();
+    };
+  }
+
+  /**
+   * Records tool interaction telemetry and persistence data.
+   */
+  private async _recordTelemetry(completedCalls: CompletedToolCall[]) {
+    try {
+      const currentModel =
+        this.client.getCurrentSequenceModel() ?? this.runtime.getModel();
+      this.client
+        .getChat()
+        .recordCompletedToolCalls(currentModel, completedCalls);
+      await recordToolCallInteractions(this.runtime, completedCalls);
+    } catch (e) {
+      debugLogger.warn(`Error recording tool call information: ${e}`);
+    }
+  }
+
+  /**
+   * Attempts to compress the chat history if thresholds are exceeded.
+   */
+  private async tryCompressChat(promptId: string): Promise<void> {
+    const chat = this.client.getChat();
+    const model = this.config.model ?? this.runtime.getModel();
+
+    const { newHistory, info } = await this.compressionService.compress(
+      chat,
+      promptId,
+      false,
+      model,
+      this.runtime,
+      this.hasFailedCompressionAttempt,
+    );
+
+    if (
+      info.compressionStatus ===
+      CompressionStatus.COMPRESSION_FAILED_INFLATED_TOKEN_COUNT
+    ) {
+      this.hasFailedCompressionAttempt = true;
+    } else if (info.compressionStatus === CompressionStatus.COMPRESSED) {
+      if (newHistory) {
+        chat.setHistory(newHistory);
+        this.hasFailedCompressionAttempt = false;
+      }
+    }
+  }
+
+  /**
+   * Returns the current message history for this session.
+   */
+  getHistory() {
+    return this.client.getHistory();
+  }
+}

packages/core/src/agents/types.ts

@@ -4,16 +4,111 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-/**
- * @fileoverview Defines the core configuration interfaces and types for the agent architecture.
- */
-
 import type { Content, FunctionDeclaration } from '@google/genai';
 import type { AnyDeclarativeTool } from '../tools/tools.js';
 import { type z } from 'zod';
 import type { ModelConfig } from '../services/modelConfigService.js';
 import type { AnySchema } from 'ajv';
 import type { A2AAuthConfig } from './auth-provider/types.js';
+import { type ServerGeminiStreamEvent } from '../core/turn.js';
+import {
+  type ToolCallResponseInfo,
+  type ToolCallRequestInfo,
+} from '../scheduler/types.js';
+
+/** Emitted when an agent session begins execution. */
+export interface AgentStartEvent {
+  type: 'agent_start';
+  value: { sessionId: string };
+}
+
+/** Emitted when an agent session completes, providing termination details. */
+export interface AgentFinishEvent {
+  type: 'agent_finish';
+  value: {
+    sessionId: string;
+    totalTurns: number;
+    reason: AgentTerminateMode;
+    message?: string;
+    error?: unknown;
+  };
+}
+
+/** Emitted when a group of tool calls is about to be executed. */
+export interface ToolSuiteStartEvent {
+  type: 'tool_suite_start';
+  value: { count: number };
+}
+
+/** Emitted when a group of tool calls has finished executing. */
+export interface ToolSuiteFinishEvent {
+  type: 'tool_suite_finish';
+  value: { responses: ToolCallResponseInfo[] };
+}
+
+/** Emitted when an individual tool call begins execution. */
+export interface ToolCallStartEvent {
+  type: 'tool_call_start';
+  value: ToolCallRequestInfo;
+}
+
+/** Emitted when an individual tool call has finished execution. */
+export interface ToolCallFinishEvent {
+  type: 'tool_call_finish';
+  value: ToolCallResponseInfo;
+}
+
+/** Emitted when the model generates internal reasoning or "thought" content. */
+export interface ThoughtEvent {
+  type: 'thought';
+  value: string;
+}
+
+/** Emitted when an infinite loop is detected in the model's tool calling patterns. */
+export interface LoopDetectedEvent {
+  type: 'loop_detected';
+  value: { sessionId: string };
+}
+
+/**
+ * Unified event type for the Agent loop.
+ * This extends the base Gemini stream events with higher-level agent lifecycle events.
+ */
+export type AgentEvent =
+  | ServerGeminiStreamEvent
+  | AgentStartEvent
+  | AgentFinishEvent
+  | ToolSuiteStartEvent
+  | ToolSuiteFinishEvent
+  | ToolCallStartEvent
+  | ToolCallFinishEvent
+  | ThoughtEvent
+  | LoopDetectedEvent;
+
+/**
+ * Configuration for an Agent.
+ */
+export interface AgentConfig {
+  /** The name of the agent. */
+  name: string;
+  /** The system instruction (personality/rules) for the agent. */
+  systemInstruction: string;
+  /** Optional override for the model to use. */
+  model?: string;
+  /**
+   * Optional maximum number of conversational turns.
+   * Set to -1 for no limit, defaults to -1 if not specified.
+   */
+  maxTurns?: number;
+  /**
+   * Optional capabilities to enable for this agent.
+   */
+  capabilities?: {
+    compression?: boolean;
+    loopDetection?: boolean;
+    ideContext?: boolean;
+  };
+}
 
 /**
  * Describes the possible termination modes for an agent.
@@ -24,6 +119,7 @@ export enum AgentTerminateMode {
   GOAL = 'GOAL',
   MAX_TURNS = 'MAX_TURNS',
   ABORTED = 'ABORTED',
+  LOOP = 'LOOP',
   ERROR_NO_COMPLETE_TASK_CALL = 'ERROR_NO_COMPLETE_TASK_CALL',
 }