/**
 * Tests for heterogeneousAgentExecutor DB persistence layer.
 *
 * Verifies the critical path: CC stream events → messageService DB writes.
 * Covers:
 *   - Tool 3-phase persistence (pre-register → create → backfill)
 *   - Tool result content updates
 *   - Multi-step assistant message creation with correct parentId chain
 *   - Content/reasoning/model/usage final writes
 *   - Sync snapshot + reset to prevent cross-step content contamination
 */
import path from 'node:path';

import type * as LobeChatConst from '@lobechat/const';
import { HeterogeneousAgentSessionErrorCode } from '@lobechat/electron-client-ipc';
import type { AgentEventAdapter } from '@lobechat/heterogeneous-agents';
import { createAdapter } from '@lobechat/heterogeneous-agents';
import { ThreadStatus } from '@lobechat/types';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';

import { useChatStore } from '@/store/chat/store';

import { createGatewayEventHandler } from '../gatewayEventHandler';
import type { HeterogeneousAgentExecutorParams } from '../heterogeneousAgentExecutor';
import { executeHeterogeneousAgent } from '../heterogeneousAgentExecutor';

// ─── Mocks ───

// messageService — the DB layer under test
const mockCreateMessage = vi.fn();
const mockUpdateMessage = vi.fn();
const mockUpdateMessageError = vi.fn();
const mockUpdateToolMessage = vi.fn();
const mockGetMessages = vi.fn();

vi.mock('@/services/message', () => ({
  messageService: {
    createMessage: (...args: any[]) => mockCreateMessage(...args),
    getMessages: (...args: any[]) => mockGetMessages(...args),
    updateMessage: (...args: any[]) => mockUpdateMessage(...args),
    updateMessageError: (...args: any[]) => mockUpdateMessageError(...args),
    updateToolMessage: (...args: any[]) => mockUpdateToolMessage(...args),
  },
}));

// threadService — subagent Thread creation (CC `Task` tool_use)
const mockCreateThread = vi.fn();
const mockUpdateThread = vi.fn();
vi.mock('@/services/thread', () => ({
  threadService: {
    createThread: (...args: unknown[]) => mockCreateThread(...args),
    updateThread: (...args: unknown[]) => mockUpdateThread(...args),
  },
}));

// heterogeneousAgentService — IPC to Electron main
const mockStartSession = vi.fn();
const mockSendPrompt = vi.fn();
const mockStopSession = vi.fn();
const mockGetSessionInfo = vi.fn();

vi.mock('@/services/electron/heterogeneousAgent', () => ({
  heterogeneousAgentService: {
    getSessionInfo: (...args: any[]) => mockGetSessionInfo(...args),
    sendPrompt: (...args: any[]) => mockSendPrompt(...args),
    startSession: (...args: any[]) => mockStartSession(...args),
    stopSession: (...args: any[]) => mockStopSession(...args),
  },
}));

// Gateway event handler — we spy on it but let it run (it calls getMessages)
vi.mock('../gatewayEventHandler', () => ({
  createGatewayEventHandler: vi.fn(() => vi.fn()),
}));

// isDesktop — defaults to `false` (matching the real test env / __ELECTRON__
// undefined) so the existing suite is unaffected. The completion-notification
// characterization tests flip `desktopFlag.value` to `true` to exercise the
// desktop-only `notifyCompletion` branch, then restore it in afterEach.
// `vi.hoisted` so the flag exists when the hoisted `vi.mock` factory runs at
// module-evaluation time (which happens during collection, before any test).
const desktopFlag = vi.hoisted(() => ({ value: false }));
vi.mock('@lobechat/const', async (importOriginal) => {
  const actual = await importOriginal<typeof LobeChatConst>();
  return {
    ...actual,
    get isDesktop() {
      return desktopFlag.value;
    },
  };
});

// Desktop notification IPC — dynamically imported inside `notifyCompletion`.
const mockShowNotification = vi.fn(async (..._args: any[]) => {});
const mockSetBadgeCount = vi.fn(async (..._args: any[]) => {});
vi.mock('@/services/electron/desktopNotification', () => ({
  desktopNotificationService: {
    setBadgeCount: (...args: any[]) => mockSetBadgeCount(...args),
    showNotification: (...args: any[]) => mockShowNotification(...args),
  },
}));

// ─── Helpers ───

function setupIpcCapture() {
  // Mock window.electron.ipcRenderer
  const listeners = new Map<string, (...args: any[]) => void>();
  (globalThis as any).window = {
    electron: {
      ipcRenderer: {
        on: vi.fn((channel: string, handler: (...args: any[]) => void) => {
          listeners.set(channel, handler);
        }),
        removeListener: vi.fn(),
      },
    },
  };

  /**
   * Per-IPC-session adapter — mimics the desktop main pipeline:
   *   raw stdout JSON → adapter.adapt() → AgentStreamEvent → broadcast.
   * Test fixtures still feed raw CC/Codex events, so the existing ~2.8k lines
   * of stream-shape tests stay intact while the renderer's input boundary
   * becomes the new `heteroAgentEvent` channel.
   */
  const adapters = new Map<string, AgentEventAdapter>();
  /**
   * IPC-session → agent type. Defaults to `claude-code` so tests that don't
   * explicitly register codex still work; the multi-session resume test (and
   * any codex-only suite) registers explicitly via `setAgentType`.
   */
  const sessionAgentType = new Map<string, string>();

  const getAdapter = (sessionId: string) => {
    if (!adapters.has(sessionId)) {
      adapters.set(sessionId, createAdapter(sessionAgentType.get(sessionId) ?? 'claude-code'));
    }
    return adapters.get(sessionId)!;
  };

  return {
    getListeners: () => listeners,
    /** Register the agent type for an IPC session before emitting raw events. */
    setAgentType: (sessionId: string, type: string) => {
      sessionAgentType.set(sessionId, type);
    },
    /**
     * Look up the underlying adapter for an IPC session — used by the
     * `getSessionInfo` mock to mirror what main's `AgentStreamPipeline.sessionId`
     * returns to the renderer's post-prompt session-id sync.
     */
    getAdapterSessionId: (sessionId: string) => adapters.get(sessionId)?.sessionId,
    /**
     * Simulate the desktop main's per-stdout-line forwarding: feed `raw`
     * through the session's adapter, then broadcast each resulting
     * `AgentStreamEvent` over the `heteroAgentEvent` channel.
     */
    emitRawLine: (sessionId: string, raw: any) => {
      const handler = listeners.get('heteroAgentEvent');
      const adapter = getAdapter(sessionId);
      for (const event of adapter.adapt(raw)) {
        handler?.(null, {
          event: {
            data: event.data,
            operationId: defaultParams.operationId,
            stepIndex: event.stepIndex,
            timestamp: event.timestamp,
            type: event.type,
          },
          sessionId,
        });
      }
    },
    /** Simulate session completion */
    emitComplete: (sessionId: string) => {
      const handler = listeners.get('heteroAgentSessionComplete');
      handler?.(null, { sessionId });
    },
    /** Simulate session error */
    emitError: (sessionId: string, error: Record<string, unknown> | string) => {
      const handler = listeners.get('heteroAgentSessionError');
      handler?.(null, { error, sessionId });
    },
  };
}

function createMockStore(overrides: Record<string, any> = {}) {
  // Hand out a fresh AbortController + monotonically increasing sub-op id
  // for each subagent run, mirroring `startOperation`'s contract just
  // enough that the executor can build dispatchers + completion calls.
  let subOpCounter = 0;
  const store = {
    associateMessageWithOperation: vi.fn(),
    completeOperation: vi.fn(),
    drainQueuedMessages: vi.fn(() => []),
    internal_dispatchMessage: vi.fn(),
    internal_toggleToolCallingStreaming: vi.fn(),
    markTopicUnread: vi.fn(),
    operations: {
      'op-1': {
        context: { agentId: 'agent-1', scope: 'main', topicId: 'topic-1' },
        metadata: { startTime: 0 },
      },
    } as Record<string, any>,
    refreshMessages: vi.fn(async () => {}),
    refreshThreads: vi.fn(async () => {}),
    replaceMessages: vi.fn(),
    sendMessage: vi.fn(async () => {}),
    startOperation: vi.fn(() => {
      subOpCounter += 1;
      return {
        abortController: new AbortController(),
        operationId: `sub-op-${subOpCounter}`,
      };
    }),
    updateTopicMetadata: vi.fn().mockResolvedValue(undefined),
    ...overrides,
  } as any;

  if (!store.updateOperationMetadata) {
    store.updateOperationMetadata = vi.fn((operationId: string, metadata: Record<string, any>) => {
      const operation = store.operations[operationId];
      if (!operation) return;
      operation.metadata = {
        ...operation.metadata,
        ...metadata,
      };
    });
  }

  return store;
}

const defaultContext = {
  agentId: 'agent-1',
  scope: 'main' as const,
  topicId: 'topic-1',
};

const defaultParams: HeterogeneousAgentExecutorParams = {
  assistantMessageId: 'ast-initial',
  context: defaultContext,
  heterogeneousProvider: { command: 'claude', type: 'claude-code' as const },
  message: 'test prompt',
  operationId: 'op-1',
};

/** Flush async queues */
const flush = async () => {
  for (let i = 0; i < 10; i++) {
    await new Promise((r) => setTimeout(r, 10));
  }
};

// ─── CC stream-json event factories ───

const ccInit = (sessionId = 'cc-sess-1') => ({
  model: 'claude-sonnet-4-6',
  session_id: sessionId,
  subtype: 'init',
  type: 'system',
});

const ccAssistant = (msgId: string, content: any[], extra?: { model?: string; usage?: any }) => ({
  message: {
    content,
    id: msgId,
    model: extra?.model || 'claude-sonnet-4-6',
    role: 'assistant',
    usage: extra?.usage,
  },
  type: 'assistant',
});

const ccToolUse = (msgId: string, toolId: string, name: string, input: any = {}) =>
  ccAssistant(msgId, [{ id: toolId, input, name, type: 'tool_use' }]);

/**
 * CC subagent assistant event — carries `parent_tool_use_id` pointing back at
 * the outer `Task` tool_use. The adapter routes these through its subagent
 * handler which stamps `parentToolCallId` onto each tool payload.
 */
const ccSubagentToolUse = (
  msgId: string,
  parentToolUseId: string,
  toolId: string,
  name: string,
  input: any = {},
) => ({
  message: {
    content: [{ id: toolId, input, name, type: 'tool_use' }],
    id: msgId,
    role: 'assistant',
  },
  parent_tool_use_id: parentToolUseId,
  type: 'assistant',
});

/** Subagent assistant event with text content (closing summary-style turn). */
const ccSubagentText = (msgId: string, parentToolUseId: string, text: string) => ({
  message: {
    content: [{ text, type: 'text' }],
    id: msgId,
    role: 'assistant',
  },
  parent_tool_use_id: parentToolUseId,
  type: 'assistant',
});

/** The main-agent tool_result for a spawn tool_use (end of a subagent run). */
const ccSubagentSpawnResult = (spawnToolUseId: string, finalText: string) => ({
  message: {
    content: [{ content: finalText, tool_use_id: spawnToolUseId, type: 'tool_result' }],
    role: 'user',
  },
  type: 'user',
});

/**
 * Subagent INNER tool_result: a user event tagged with `parent_tool_use_id`,
 * which the adapter routes through its subagent path so the emitted
 * `tool_result` + `tool_end` events both carry the `subagent` peer field.
 */
const ccSubagentToolResult = (
  toolUseId: string,
  parentToolUseId: string,
  content: string,
  isError = false,
) => ({
  message: {
    content: [{ content, is_error: isError, tool_use_id: toolUseId, type: 'tool_result' }],
    role: 'user',
  },
  parent_tool_use_id: parentToolUseId,
  type: 'user',
});

const ccText = (msgId: string, text: string) => ccAssistant(msgId, [{ text, type: 'text' }]);

const ccThinking = (msgId: string, thinking: string) =>
  ccAssistant(msgId, [{ thinking, type: 'thinking' }]);

const ccToolResult = (toolUseId: string, content: string, isError = false) => ({
  message: {
    content: [{ content, is_error: isError, tool_use_id: toolUseId, type: 'tool_result' }],
    role: 'user',
  },
  type: 'user',
});

const ccResult = (isError = false, result = 'done') => ({
  is_error: isError,
  result,
  type: 'result',
});

/**
 * `stream_event: message_start` — primes adapter's in-flight message.id so a
 * following `message_delta` (which has no message.id of its own) can attach
 * its authoritative usage to the correct turn.
 */
const ccMessageStart = (msgId: string, model = 'claude-sonnet-4-6') => ({
  event: { message: { id: msgId, model }, type: 'message_start' },
  type: 'stream_event',
});

/**
 * `stream_event: message_delta` — the authoritative per-turn usage under
 * `--include-partial-messages` (CC's `assistant` events only echo a stale
 * message_start snapshot, so turn_metadata is driven off this event).
 */
const ccMessageDelta = (usage: {
  cache_creation_input_tokens?: number;
  cache_read_input_tokens?: number;
  input_tokens?: number;
  output_tokens?: number;
}) => ({
  event: { type: 'message_delta', usage },
  type: 'stream_event',
});

// ─── Codex JSONL event factories ───

const codexThreadStarted = (threadId = 'codex-thread-1') => ({
  thread_id: threadId,
  type: 'thread.started',
});

const codexSessionConfigured = (model = 'gpt-5.5') => ({
  model,
  type: 'session_configured',
});

const codexTurnStarted = () => ({
  type: 'turn.started',
});

const codexAgentMessage = (id: string, text: string) => ({
  item: {
    id,
    text,
    type: 'agent_message',
  },
  type: 'item.completed',
});

const codexCommandStarted = (id: string, command: string) => ({
  item: {
    aggregated_output: '',
    command,
    exit_code: null,
    id,
    status: 'in_progress',
    type: 'command_execution',
  },
  type: 'item.started',
});

const codexCommandCompleted = (id: string, command: string, aggregatedOutput: string) => ({
  item: {
    aggregated_output: aggregatedOutput,
    command,
    exit_code: 0,
    id,
    status: 'completed',
    type: 'command_execution',
  },
  type: 'item.completed',
});

const codexTurnCompleted = (usage?: {
  cached_input_tokens?: number;
  input_tokens?: number;
  output_tokens?: number;
}) => ({
  ...(usage ? { usage } : {}),
  type: 'turn.completed',
});

// ─── Tests ───

describe('heterogeneousAgentExecutor DB persistence', () => {
  let ipc: ReturnType<typeof setupIpcCapture>;

  beforeEach(() => {
    vi.clearAllMocks();
    ipc = setupIpcCapture();
    // Register the IPC session's agent type from the params the executor
    // hands to startSession, so the helper picks the right adapter when the
    // test starts emitting raw events.
    mockStartSession.mockImplementation(async (params: any) => {
      ipc.setAgentType('ipc-sess-1', params.agentType ?? 'claude-code');
      return { sessionId: 'ipc-sess-1' };
    });
    mockSendPrompt.mockResolvedValue(undefined);
    mockStopSession.mockResolvedValue(undefined);
    // Mirror the desktop main: `getSessionInfo` returns whatever the producer
    // pipeline's adapter has extracted from the JSONL stream so far. Tests
    // that never emit an init / thread.started event get `agentSessionId:
    // undefined`, matching pre-Phase-0 behavior where the renderer-side
    // adapter never observed one either. The renderer service hands the raw
    // sessionId string straight to the mock — it's the underlying IPC handler
    // that wraps `{ sessionId }`, and that's stubbed here.
    mockGetSessionInfo.mockImplementation(async (sessionId: string) => ({
      agentSessionId: ipc.getAdapterSessionId(sessionId),
    }));
    mockGetMessages.mockResolvedValue([]);
    // Honor a caller-provided `id` like the real messageService does — the
    // main + subagent coordinators PRE-ALLOCATE message ids so their intents can
    // carry concrete parentId chains. A mock that minted its own id would break
    // the chain (the assistant's parentId would never match the tool row's id).
    mockCreateMessage.mockImplementation(async (params: any) => ({
      id:
        params.id ??
        `created-${params.role}-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`,
    }));
    mockUpdateMessage.mockResolvedValue(undefined);
    mockUpdateMessageError.mockResolvedValue({ success: false });
    mockUpdateToolMessage.mockResolvedValue(undefined);
    mockCreateThread.mockImplementation(async (params: any) => params.id || 'thread-generated');
  });

  afterEach(() => {
    delete (globalThis as any).window;
  });

  /**
   * Runs the executor in background, then feeds CC events and completes.
   * Returns a promise that resolves when the executor finishes.
   */
  async function runWithEvents(ccEvents: any[], opts?: { params?: Partial<typeof defaultParams> }) {
    const store = createMockStore();
    const get = vi.fn(() => store);

    // sendPrompt will resolve after we emit all events
    let resolveSendPrompt: () => void;
    mockSendPrompt.mockReturnValue(
      new Promise<void>((r) => {
        resolveSendPrompt = r;
      }),
    );

    const executorPromise = executeHeterogeneousAgent(get, {
      ...defaultParams,
      ...opts?.params,
    });

    // Wait for startSession + subscribeBroadcasts to complete
    await flush();

    // Feed CC events
    for (const event of ccEvents) {
      ipc.emitRawLine('ipc-sess-1', event);
    }

    // Signal completion
    ipc.emitComplete('ipc-sess-1');
    await flush();

    // Resolve sendPrompt to let executor continue
    resolveSendPrompt!();
    await flush();

    // Wait for executor to finish
    await executorPromise;
    await flush();

    return { get, store };
  }

  // ────────────────────────────────────────────────────
  // Tool 3-phase persistence
  // ────────────────────────────────────────────────────

  describe('tool 3-phase persistence', () => {
    it('should pre-register tools, create tool messages, then backfill result_msg_id', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_01', 'toolu_1', 'Read', { file_path: '/a.ts' }),
        ccToolResult('toolu_1', 'file content'),
        ccText('msg_02', 'Done'),
        ccResult(),
      ]);

      // Phase 1 + Phase 3: updateMessage called with tools[] on the assistant
      // Phase 1 has tools without result_msg_id, Phase 3 has tools with result_msg_id
      const toolUpdateCalls = mockUpdateMessage.mock.calls.filter(
        ([id, val]: any) => id === 'ast-initial' && val.tools?.length > 0,
      );
      // At least 2 calls: phase 1 (pre-register) + phase 3 (backfill)
      expect(toolUpdateCalls.length).toBeGreaterThanOrEqual(2);

      // Phase 2: createMessage called with role='tool'
      const toolCreateCalls = mockCreateMessage.mock.calls.filter(
        ([params]: any) => params.role === 'tool',
      );
      expect(toolCreateCalls.length).toBe(1);
      expect(toolCreateCalls[0][0]).toMatchObject({
        parentId: 'ast-initial',
        role: 'tool',
        tool_call_id: 'toolu_1',
        plugin: expect.objectContaining({ apiName: 'Read' }),
      });

      // Phase 3: the last tools[] write should backfill result_msg_id with the
      // tool message's (pre-allocated) id.
      const createdToolId = toolCreateCalls[0][0].id;
      const lastToolUpdate = toolUpdateCalls.at(-1)!;
      expect(lastToolUpdate[1].tools[0].result_msg_id).toBe(createdToolId);
    });

    it('should deduplicate tool calls (idempotent)', async () => {
      await runWithEvents([
        ccInit(),
        // Same tool_use id sent twice (CC can echo tool blocks)
        ccToolUse('msg_01', 'toolu_1', 'Bash', { command: 'ls' }),
        ccAssistant('msg_01', [
          { id: 'toolu_1', input: { command: 'ls' }, name: 'Bash', type: 'tool_use' },
        ]),
        ccToolResult('toolu_1', 'output'),
        ccResult(),
      ]);

      // Should only create ONE tool message despite two tool_use events with same id
      const toolCreates = mockCreateMessage.mock.calls.filter(([p]: any) => p.role === 'tool');
      expect(toolCreates.length).toBe(1);
    });
  });

  // ────────────────────────────────────────────────────
  // Tool result content persistence
  // ────────────────────────────────────────────────────

  describe('tool result persistence', () => {
    it('should update tool message content on tool_result', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_01', 'toolu_read', 'Read', { file_path: '/x.ts' }),
        ccToolResult('toolu_read', 'the file content here'),
        ccResult(),
      ]);

      const toolId = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_read',
      )![0].id;
      expect(mockUpdateToolMessage).toHaveBeenCalledWith(
        toolId,
        { content: 'the file content here', pluginError: undefined },
        { agentId: 'agent-1', topicId: 'topic-1' },
      );
    });

    it('should mark error tool results with pluginError', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_01', 'toolu_fail', 'Read', { file_path: '/nope' }),
        ccToolResult('toolu_fail', 'ENOENT: no such file', true),
        ccResult(),
      ]);

      const toolId = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_fail',
      )![0].id;
      expect(mockUpdateToolMessage).toHaveBeenCalledWith(
        toolId,
        { content: 'ENOENT: no such file', pluginError: { message: 'ENOENT: no such file' } },
        { agentId: 'agent-1', topicId: 'topic-1' },
      );
    });
  });

  // ────────────────────────────────────────────────────
  // Multi-step parentId chain
  // ────────────────────────────────────────────────────

  describe('multi-step parentId chain', () => {
    it('should chain step assistants along the spine, with tools inline', async () => {
      await runWithEvents([
        ccInit(),
        // Step 1: tool_use Read (message_start primes turn + model/provider
        // so the executor can stamp step 2's createMessage with them)
        ccMessageStart('msg_01'),
        ccToolUse('msg_01', 'toolu_1', 'Read', { file_path: '/a.ts' }),
        ccMessageDelta({ input_tokens: 10, output_tokens: 5 }),
        ccToolResult('toolu_1', 'content of a.ts'),
        // Step 2 (new message.id): tool_use Write
        ccMessageStart('msg_02'),
        ccToolUse('msg_02', 'toolu_2', 'Write', { file_path: '/b.ts', content: 'new' }),
        ccMessageDelta({ input_tokens: 20, output_tokens: 10 }),
        ccToolResult('toolu_2', 'file written'),
        // Step 3 (new message.id): final text
        ccMessageStart('msg_03'),
        ccText('msg_03', 'All done!'),
        ccMessageDelta({ input_tokens: 30, output_tokens: 15 }),
        ccResult(),
      ]);

      // Collect all createMessage calls with their parentId
      // Tool message for step 1 — parentId should be the initial assistant
      const tool1Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_1',
      );
      expect(tool1Create?.[0].parentId).toBe('ast-initial');

      // Assistant for step 2 — parentId should be the spine (the initial
      // assistant), NOT step 1's tool. Tools are inline children of their own
      // assistant; the next assistant chains off the most recent non-tool
      // main message (the spine).
      const step2Assistant = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.parentId !== undefined,
      );
      expect(step2Assistant).toBeDefined();
      expect(step2Assistant![0].parentId).toBe('ast-initial');
      // createMessage should carry the adapter provider so step 2's assistant
      // lands in DB with provider set from the start (no later backfill needed).
      expect(step2Assistant![0].provider).toBe('claude-code');
    });

    it('should fall back to assistant parentId when step has no tools', async () => {
      const ids: string[] = [];
      mockCreateMessage.mockImplementation(async (params: any) => {
        const id = `${params.role}-${ids.length}`;
        ids.push(id);
        return { id };
      });

      await runWithEvents([
        ccInit(),
        // Step 1: just text, no tools
        ccText('msg_01', 'Let me think...'),
        // Step 2: more text (new message.id, no tools in step 1)
        ccText('msg_02', 'Here is the answer.'),
        ccResult(),
      ]);

      // Step 2 assistant should have parentId = initial assistant (no tools to chain through)
      const step2 = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant');
      expect(step2?.[0].parentId).toBe('ast-initial');
    });
  });

  // ────────────────────────────────────────────────────
  // Final content + usage writes
  // ────────────────────────────────────────────────────

  describe('final content writes (onComplete)', () => {
    it('should write accumulated content + model + provider to the final assistant message', async () => {
      await runWithEvents([
        ccInit(),
        // message_start carries the model for this turn; individual assistant
        // content-block events echo the same model, so the final write should
        // stamp `claude-opus-4-6` (not the init-default sonnet).
        ccMessageStart('msg_01', 'claude-opus-4-6'),
        ccAssistant('msg_01', [{ text: 'Hello ', type: 'text' }], {
          model: 'claude-opus-4-6',
        }),
        ccAssistant('msg_01', [{ text: 'world!', type: 'text' }], {
          model: 'claude-opus-4-6',
        }),
        // message_delta fires the authoritative turn_metadata (with model from
        // the adapter's in-flight state)
        ccMessageDelta({ input_tokens: 100, output_tokens: 20 }),
        ccResult(),
      ]);

      const finalWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === 'Hello world!',
      );
      expect(finalWrite).toBeDefined();
      expect(finalWrite![1].model).toBe('claude-opus-4-6');
      // provider is emitted by the CC adapter on turn_metadata so it rides
      // along with the final content/model write.
      expect(finalWrite![1].provider).toBe('claude-code');
    });

    it('should write accumulated reasoning', async () => {
      await runWithEvents([
        ccInit(),
        ccThinking('msg_01', 'Let me think about this.'),
        ccText('msg_01', 'Answer.'),
        ccResult(),
      ]);

      const finalWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.reasoning,
      );
      expect(finalWrite).toBeDefined();
      expect(finalWrite![1].reasoning.content).toBe('Let me think about this.');
    });

    it('should persist per-step usage to each step assistant message, not accumulated', async () => {
      // Realistic CC partial-messages flow: message_start primes the turn,
      // assistant events echo a stale usage, message_delta carries the final.
      await runWithEvents([
        ccInit(),
        ccMessageStart('msg_01'),
        ccAssistant('msg_01', [{ text: 'a', type: 'text' }]),
        ccToolUse('msg_01', 'toolu_1', 'Bash', {}),
        ccMessageDelta({
          cache_creation_input_tokens: 50,
          cache_read_input_tokens: 200,
          input_tokens: 100,
          output_tokens: 50,
        }),
        ccToolResult('toolu_1', 'ok'),
        ccMessageStart('msg_02'),
        ccAssistant('msg_02', [{ text: 'b', type: 'text' }]),
        ccMessageDelta({ input_tokens: 300, output_tokens: 80 }),
        ccResult(),
      ]);

      const usageWrites = mockUpdateMessage.mock.calls.filter(
        ([, val]: any) => val.metadata?.usage?.totalTokens,
      );
      // One usage write per step (msg_01 → ast-initial, msg_02 → new step assistant)
      expect(usageWrites.length).toBe(2);
      // The step-2 assistant is the only newly-created assistant (pre-allocated id).
      const step2Id = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant')![0]
        .id;

      const step1 = usageWrites.find(([id]: any) => id === 'ast-initial');
      expect(step1).toBeDefined();
      const u1 = step1![1].metadata.usage;
      // msg_01: 100 input (miss) + 200 cached + 50 cache_create = 350; 50 output
      expect(u1.totalInputTokens).toBe(350);
      expect(u1.totalOutputTokens).toBe(50);
      expect(u1.totalTokens).toBe(400);
      expect(u1.inputCacheMissTokens).toBe(100);
      expect(u1.inputCachedTokens).toBe(200);
      expect(u1.inputWriteCacheTokens).toBe(50);

      const step2 = usageWrites.find(([id]: any) => id === step2Id);
      expect(step2).toBeDefined();
      const u2 = step2![1].metadata.usage;
      // msg_02: 300 input (miss, no cache); 80 output
      expect(u2.totalInputTokens).toBe(300);
      expect(u2.totalOutputTokens).toBe(80);
      expect(u2.totalTokens).toBe(380);
      expect(u2.inputCacheMissTokens).toBe(300);
      // No cache tokens for this turn — these fields should be absent
      expect(u2.inputCachedTokens).toBeUndefined();
      expect(u2.inputWriteCacheTokens).toBeUndefined();
    });

    it('should ignore stale usage on assistant events (from message_start echo)', async () => {
      // Regression for -style bug: under partial-messages mode, CC
      // echoes a stale message_start usage (e.g. output_tokens: 1) on every
      // content-block assistant event. If the adapter picked that up, the DB
      // would record output_tokens=1 instead of the real total. This verifies
      // the stale snapshot is ignored and only the message_delta total lands.
      await runWithEvents([
        ccInit(),
        ccMessageStart('msg_01'),
        // All assistant events below carry the STALE placeholder usage
        ccAssistant('msg_01', [{ text: 'hi', type: 'text' }], {
          usage: { input_tokens: 6, output_tokens: 1 }, // stale
        }),
        ccAssistant('msg_01', [{ id: 'tu', input: {}, name: 'Read', type: 'tool_use' }], {
          usage: { input_tokens: 6, output_tokens: 1 }, // stale echo
        }),
        // Authoritative final usage arrives on message_delta
        ccMessageDelta({ input_tokens: 6, output_tokens: 265 }),
        ccToolResult('tu', 'ok'),
        ccResult(),
      ]);

      const usageWrites = mockUpdateMessage.mock.calls.filter(
        ([, val]: any) => val.metadata?.usage?.totalTokens,
      );
      expect(usageWrites.length).toBe(1);
      expect(usageWrites[0][1].metadata.usage.totalOutputTokens).toBe(265); // not 1
      expect(usageWrites[0][1].metadata.usage.totalInputTokens).toBe(6);
    });
  });

  // ────────────────────────────────────────────────────
  // Sync snapshot prevents cross-step contamination
  // ────────────────────────────────────────────────────

  describe('sync snapshot on step boundary', () => {
    it('should NOT mix new-step content into old-step DB write', async () => {
      // This tests the race condition fix: when adapter produces
      // [stream_end, stream_start(newStep), stream_chunk(text)] from a single raw line,
      // the stream_chunk should go to the NEW step, not the old one.

      await runWithEvents([
        ccInit(),
        // Step 1: text
        ccText('msg_01', 'Step 1 content'),
        // Step 2: new message.id — adapter emits stream_end + stream_start(newStep) + chunks
        // in the SAME onRawLine call
        ccText('msg_02', 'Step 2 content'),
        ccResult(),
      ]);

      // The old step (ast-initial) should get "Step 1 content", NOT "Step 1 contentStep 2 content"
      const oldStepWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === 'Step 1 content',
      );
      expect(oldStepWrite).toBeDefined();

      // The new step's final write should have "Step 2 content"
      const newStepId = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant')?.[0]
        .id;
      expect(newStepId).toBeDefined();
      const newStepWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === newStepId && val.content === 'Step 2 content',
      );
      expect(newStepWrite).toBeDefined();
    });
  });

  // ────────────────────────────────────────────────────
  // Error handling
  // ────────────────────────────────────────────────────

  describe('error handling', () => {
    it('should persist accumulated content on error', async () => {
      const store = createMockStore();
      const get = vi.fn(() => store);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, defaultParams);
      await flush();

      // Feed some content, then error
      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', 'partial content'));
      ipc.emitError('ipc-sess-1', 'Connection lost');
      await flush();

      resolveSendPrompt!();
      await executorPromise.catch(() => {});
      await flush();

      // Should have written the partial content
      const contentWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === 'partial content',
      );
      expect(contentWrite).toBeDefined();
    });

    it('should not persist streamed auth error echoes as assistant content when the session errors', async () => {
      const rawAuthError =
        'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}';

      const store = createMockStore();
      const get = vi.fn(() => store);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, defaultParams);
      await flush();

      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', rawAuthError));
      ipc.emitError('ipc-sess-1', rawAuthError);
      await flush();

      resolveSendPrompt!();
      await executorPromise.catch(() => {});
      await flush();

      const contentWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === rawAuthError,
      );

      expect(contentWrite).toBeUndefined();
      expect(mockUpdateMessageError).toHaveBeenCalledWith(
        'ast-initial',
        {
          body: expect.objectContaining({
            agentType: 'claude-code',
            code: HeterogeneousAgentSessionErrorCode.AuthRequired,
            stderr: rawAuthError,
          }),
          message:
            'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
          type: 'AgentRuntimeError',
        },
        expect.any(Object),
      );
      expect(store.internal_dispatchMessage).toHaveBeenCalledWith(
        {
          id: 'ast-initial',
          type: 'updateMessage',
          value: {
            content: '',
            error: {
              body: expect.objectContaining({
                agentType: 'claude-code',
                code: HeterogeneousAgentSessionErrorCode.AuthRequired,
                stderr: rawAuthError,
              }),
              message:
                'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
              type: 'AgentRuntimeError',
            },
          },
        },
        { operationId: 'op-1' },
      );
    });

    it('should not keep streamed auth error echoes when the adapter ends with a result error', async () => {
      const rawAuthError =
        'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}';

      const { store } = await runWithEvents([
        ccInit(),
        ccText('msg_01', rawAuthError),
        ccResult(true, rawAuthError),
      ]);

      const contentWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === rawAuthError,
      );

      expect(contentWrite).toBeUndefined();
      expect(mockUpdateMessageError).toHaveBeenCalledWith(
        'ast-initial',
        {
          body: expect.objectContaining({
            agentType: 'claude-code',
            code: HeterogeneousAgentSessionErrorCode.AuthRequired,
            stderr: rawAuthError,
          }),
          message:
            'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
          type: 'AgentRuntimeError',
        },
        expect.any(Object),
      );
      expect(store.internal_dispatchMessage).toHaveBeenCalledWith(
        {
          id: 'ast-initial',
          type: 'updateMessage',
          value: {
            content: '',
            error: {
              body: expect.objectContaining({
                agentType: 'claude-code',
                code: HeterogeneousAgentSessionErrorCode.AuthRequired,
                stderr: rawAuthError,
              }),
              message:
                'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
              type: 'AgentRuntimeError',
            },
          },
        },
        { operationId: 'op-1' },
      );
    });

    it('should prefer deferred adapter auth errors over generic exit-code session errors', async () => {
      const rawAuthError =
        'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}';

      const store = createMockStore();
      const get = vi.fn(() => store);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, defaultParams);
      await flush();

      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', rawAuthError));
      ipc.emitRawLine('ipc-sess-1', ccResult(true, rawAuthError));
      ipc.emitError('ipc-sess-1', 'Agent exited with code 1');
      await flush();

      resolveSendPrompt!();
      await executorPromise.catch(() => {});
      await flush();

      const contentWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id === 'ast-initial' && val.content === rawAuthError,
      );

      expect(contentWrite).toBeUndefined();
      expect(mockUpdateMessageError).toHaveBeenCalledWith(
        'ast-initial',
        {
          body: expect.objectContaining({
            agentType: 'claude-code',
            code: HeterogeneousAgentSessionErrorCode.AuthRequired,
            stderr: rawAuthError,
          }),
          message:
            'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
          type: 'AgentRuntimeError',
        },
        expect.any(Object),
      );
      expect(store.internal_dispatchMessage).toHaveBeenCalledWith(
        {
          id: 'ast-initial',
          type: 'updateMessage',
          value: {
            content: '',
            error: {
              body: expect.objectContaining({
                agentType: 'claude-code',
                code: HeterogeneousAgentSessionErrorCode.AuthRequired,
                stderr: rawAuthError,
              }),
              message:
                'Claude Code could not authenticate. Sign in again or refresh its credentials, then retry.',
              type: 'AgentRuntimeError',
            },
          },
        },
        { operationId: 'op-1' },
      );
    });

    it('should prefer Codex JSONL terminal errors over stderr status session errors', async () => {
      const codexModelError =
        "The 'gpt-5.5' model requires a newer version of Codex. Please upgrade to the latest app or CLI and try again.";
      const rawCodexError = JSON.stringify({
        error: {
          message: codexModelError,
          type: 'invalid_request_error',
        },
        status: 400,
        type: 'error',
      });
      const store = createMockStore();
      const get = vi.fn(() => store);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, {
        ...defaultParams,
        heterogeneousProvider: { command: 'codex', type: 'codex' as const },
      });
      await flush();

      ipc.emitRawLine('ipc-sess-1', codexThreadStarted());
      ipc.emitRawLine('ipc-sess-1', codexTurnStarted());
      ipc.emitRawLine('ipc-sess-1', { message: rawCodexError, type: 'error' });
      ipc.emitRawLine('ipc-sess-1', {
        error: { message: rawCodexError },
        type: 'turn.failed',
      });
      ipc.emitError('ipc-sess-1', 'Agent exited with code 1');
      await flush();

      resolveSendPrompt!();
      await executorPromise.catch(() => {});
      await flush();

      expect(mockUpdateMessageError).toHaveBeenCalledWith(
        'ast-initial',
        {
          body: expect.objectContaining({
            agentType: 'codex',
            clearEchoedContent: true,
            message: codexModelError,
            stderr: rawCodexError,
          }),
          message: codexModelError,
          type: 'AgentRuntimeError',
        },
        expect.any(Object),
      );
      expect(mockUpdateMessageError).not.toHaveBeenCalledWith(
        'ast-initial',
        expect.objectContaining({ message: 'Reading prompt from stdin...' }),
        expect.any(Object),
      );
      expect(mockUpdateMessageError).not.toHaveBeenCalledWith(
        'ast-initial',
        expect.objectContaining({ message: 'Agent exited with code 1' }),
        expect.any(Object),
      );
    });

    it('should persist and dispatch structured cli-not-found errors when sendPrompt rejects', async () => {
      const store = createMockStore();
      const get = vi.fn(() => store);
      const cliError = {
        agentType: 'claude-code',
        code: HeterogeneousAgentSessionErrorCode.CliNotFound,
        docsUrl: 'https://docs.anthropic.com/en/docs/claude-code/setup',
        installCommands: ['curl -fsSL https://claude.ai/install.sh | bash'],
        message: 'Claude Code CLI was not found',
      };

      mockSendPrompt.mockRejectedValueOnce(cliError);

      await executeHeterogeneousAgent(get, defaultParams);
      await flush();

      expect(mockUpdateMessageError).toHaveBeenCalledWith(
        'ast-initial',
        {
          body: cliError,
          message: 'Claude Code CLI was not found',
          type: 'AgentRuntimeError',
        },
        {
          agentId: 'agent-1',
          groupId: undefined,
          threadId: undefined,
          topicId: 'topic-1',
        },
      );
      expect(store.refreshMessages).toHaveBeenCalled();
      expect(store.internal_dispatchMessage).toHaveBeenCalledWith(
        {
          id: 'ast-initial',
          type: 'updateMessage',
          value: {
            error: {
              body: cliError,
              message: 'Claude Code CLI was not found',
              type: 'AgentRuntimeError',
            },
          },
        },
        { operationId: 'op-1' },
      );
      expect(store.completeOperation).toHaveBeenCalledWith('op-1');
    });

    it('should forward imageList to heterogeneousAgentService.sendPrompt for Codex runs', async () => {
      const store = createMockStore();
      const get = vi.fn(() => store);
      setupIpcCapture();
      const imageList = [
        { id: 'image-1', url: 'https://example.com/screenshot-1.png' },
        { id: 'image-2', url: 'https://example.com/screenshot-2.png' },
      ];

      await executeHeterogeneousAgent(get, {
        ...defaultParams,
        heterogeneousProvider: { command: 'codex', type: 'codex' as const },
        imageList,
      });

      expect(mockSendPrompt).toHaveBeenCalledWith('ipc-sess-1', 'test prompt', 'op-1', imageList);
    });

    it('should clear stale resume metadata and retry once without resume for recoverable Codex errors', async () => {
      const store = createMockStore();
      const get = vi.fn(() => store);
      const sendPromptControllers = new Map<
        string,
        { reject: (reason?: unknown) => void; resolve: () => void }
      >();

      // Both spawned IPC sessions are codex; register so the helper's adapter
      // pipeline yields the right shape when the test emits raw codex events.
      ipc.setAgentType('ipc-sess-1', 'codex');
      ipc.setAgentType('ipc-sess-2', 'codex');
      let startCount = 0;
      mockStartSession.mockImplementation(async (params: any) => {
        startCount += 1;
        const sid = startCount === 1 ? 'ipc-sess-1' : 'ipc-sess-2';
        ipc.setAgentType(sid, params.agentType ?? 'claude-code');
        return { sessionId: sid };
      });
      mockSendPrompt.mockImplementation(
        (sessionId: string) =>
          new Promise<void>((resolve, reject) => {
            sendPromptControllers.set(sessionId, { reject, resolve });
          }),
      );

      const executorPromise = executeHeterogeneousAgent(get, {
        ...defaultParams,
        heterogeneousProvider: { command: 'codex', type: 'codex' as const },
        resumeSessionId: 'thread_stale_123',
        workingDirectory: '/Users/me/repo',
      });

      await flush();

      ipc.emitError('ipc-sess-1', {
        agentType: 'codex',
        code: HeterogeneousAgentSessionErrorCode.ResumeThreadNotFound,
        message: 'The saved Codex thread could not be found, so it can no longer be resumed.',
      });
      await flush();

      sendPromptControllers.get('ipc-sess-1')?.reject(new Error('resume failed'));
      await flush();

      expect(mockStartSession).toHaveBeenNthCalledWith(
        1,
        expect.objectContaining({
          agentType: 'codex',
          resumeSessionId: 'thread_stale_123',
        }),
      );
      expect(mockStartSession).toHaveBeenNthCalledWith(
        2,
        expect.objectContaining({
          agentType: 'codex',
          resumeSessionId: undefined,
        }),
      );
      expect(store.updateTopicMetadata).toHaveBeenCalledWith('topic-1', {
        heteroSessionId: undefined,
        workingDirectory: '/Users/me/repo',
      });

      ipc.emitRawLine('ipc-sess-2', { thread_id: 'thread_new_456', type: 'thread.started' });
      ipc.emitComplete('ipc-sess-2');
      await flush();

      sendPromptControllers.get('ipc-sess-2')?.resolve();
      await executorPromise;
      await flush();

      expect(store.updateTopicMetadata).toHaveBeenCalledWith('topic-1', {
        heteroSessionId: 'thread_new_456',
        workingDirectory: '/Users/me/repo',
      });
    });

    it('does NOT retry resume once partial output streamed, even before the persist queue drains', async () => {
      // Regression: content/tool/subagent state now lives in `mainState` and is
      // only updated inside the QUEUED reduceAndApplyMain. retryWithoutResume's
      // guard runs synchronously in onError BEFORE the queue drains, so it must
      // rely on a synchronous "saw streamed event" flag — not on mainState —
      // or it would start a second run and duplicate the partial output.
      const store = createMockStore();
      const get = vi.fn(() => store);
      let startCount = 0;
      mockStartSession.mockImplementation(async (params: any) => {
        startCount += 1;
        const sid = startCount === 1 ? 'ipc-sess-1' : 'ipc-sess-2';
        ipc.setAgentType(sid, params.agentType ?? 'claude-code');
        return { sessionId: sid };
      });
      // sendPrompt hangs so the run stays in-flight; onError drives the decision.
      mockSendPrompt.mockImplementation(() => new Promise<void>(() => {}));
      // Block the persist queue: updateMessage never resolves, so the queued
      // reduce can't commit into mainState — the exact window the old guard missed.
      let releaseUpdate: () => void = () => {};
      mockUpdateMessage.mockImplementation(
        () => new Promise<void>((resolve) => (releaseUpdate = resolve)),
      );

      void executeHeterogeneousAgent(get, {
        ...defaultParams,
        resumeSessionId: 'sess_stale',
        workingDirectory: '/repo',
      });
      await flush();

      // init → stream_start(model) → reduce blocks on the hung updateMessage.
      ipc.emitRawLine('ipc-sess-1', ccInit());
      // a text chunk = partial output: sets the sync flag, but its reduce sits
      // behind the blocked init reduce and never commits accContent.
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', 'partial answer so far'));
      await flush();

      // Recoverable resume error arrives while the queue is still blocked.
      ipc.emitError('ipc-sess-1', {
        agentType: 'claude-code',
        code: HeterogeneousAgentSessionErrorCode.ResumeThreadNotFound,
        message: 'resume gone',
      });
      await flush();

      // Output already streamed → no second run, no resume-metadata clear.
      expect(startCount).toBe(1);
      expect(store.updateTopicMetadata).not.toHaveBeenCalled();

      releaseUpdate();
    });

    it('does NOT advance currentAssistantId when a step-boundary assistant create fails', async () => {
      // Regression: a transient createMessage failure on the new-step assistant
      // must skip the reducer commit (like the subagent createMessage path),
      // NOT advance currentAssistantId to a row that was never created — else
      // every later content/tool write targets a missing assistant and is lost.
      mockCreateMessage.mockImplementation(async (p: any) => {
        if (p.role === 'assistant') throw new Error('transient create failure');
        return { id: p.id ?? `tool-${Date.now()}` };
      });

      await runWithEvents([
        ccInit(),
        // Turn 1 on the seed assistant (ast-initial): text + tool + result.
        ccText('msg_01', 'turn one'),
        ccToolUse('msg_01', 't1', 'Bash', { command: 'ls' }),
        ccToolResult('t1', 'ok'),
        // Turn 2 (new message.id → step boundary): the new-assistant create FAILS,
        // then a tool_use arrives for this turn.
        ccToolUse('msg_02', 't2', 'Read', { file_path: '/a' }),
        ccToolResult('t2', 'data'),
        ccResult(),
      ]);

      // Commit was skipped on the failed create → currentAssistantId stayed at
      // the seed, so turn-2's tool parents off ast-initial, never off the
      // assistant row that was never created.
      const t2Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 't2',
      );
      expect(t2Create).toBeDefined();
      expect(t2Create![0].parentId).toBe('ast-initial');
    });
  });

  describe('Codex multi-turn persistence', () => {
    it('should persist Codex host model metadata onto the current assistant message', async () => {
      await runWithEvents(
        [
          codexSessionConfigured('gpt-5.5'),
          codexThreadStarted(),
          codexTurnStarted(),
          codexAgentMessage('item_0', 'Done.'),
          codexTurnCompleted({ cached_input_tokens: 4, input_tokens: 6, output_tokens: 3 }),
        ],
        {
          params: {
            heterogeneousProvider: { command: 'codex', type: 'codex' as const },
          },
        },
      );

      const modelWrites = mockUpdateMessage.mock.calls.filter(
        ([id, value]: any) =>
          id === 'ast-initial' && value.model === 'gpt-5.5' && value.provider === 'codex',
      );
      expect(modelWrites.length).toBeGreaterThan(0);

      const usageWrite = modelWrites.find(([, value]: any) => value.metadata?.usage);
      expect(usageWrite?.[1]).toMatchObject({
        metadata: {
          usage: {
            inputCachedTokens: 4,
            inputCacheMissTokens: 6,
            totalInputTokens: 10,
            totalOutputTokens: 3,
            totalTokens: 13,
          },
        },
        model: 'gpt-5.5',
        provider: 'codex',
      });
    });

    it('should switch to a new assistant before persisting the next turn tool', async () => {
      const idCounter = { assistant: 0, tool: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool += 1;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }

        if (params.role === 'assistant') {
          idCounter.assistant += 1;
          return { id: params.id ?? `ast-new-${idCounter.assistant}` };
        }

        return { id: `created-${params.role}-${idCounter.assistant + idCounter.tool}` };
      });

      const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          toolsUpdates.push({
            assistantId: id,
            toolIds: val.tools.map((tool: any) => tool.id),
          });
        }
      });

      await runWithEvents(
        [
          codexThreadStarted(),
          codexTurnStarted(),
          codexAgentMessage('item_0', 'Running the first command.'),
          codexCommandStarted('item_1', '/bin/zsh -lc pwd'),
          codexCommandCompleted('item_1', '/bin/zsh -lc pwd', '/repo\n'),
          codexTurnCompleted({ input_tokens: 10, output_tokens: 3 }),
          codexTurnStarted(),
          codexAgentMessage('item_2', 'Running the second command.'),
          codexCommandStarted('item_3', "/bin/zsh -lc 'git status --short'"),
          codexCommandCompleted('item_3', "/bin/zsh -lc 'git status --short'", ' M src/file.ts\n'),
          codexTurnCompleted({ input_tokens: 12, output_tokens: 4 }),
        ],
        {
          params: {
            heterogeneousProvider: { command: 'codex', type: 'codex' as const },
          },
        },
      );

      // The second turn opens exactly one new assistant (pre-allocated id).
      const newAstId = mockCreateMessage.mock.calls.find(
        ([params]: any) => params.role === 'assistant',
      )![0].id;

      // The second-turn assistant chains off the spine (turn 1's assistant,
      // here the initial one), NOT turn 1's last tool.
      const secondTurnAssistantCreate = mockCreateMessage.mock.calls.find(
        ([params]: any) => params.role === 'assistant',
      );
      expect(secondTurnAssistantCreate?.[0]).toMatchObject({
        parentId: 'ast-initial',
        role: 'assistant',
      });

      const firstToolCreate = mockCreateMessage.mock.calls.find(
        ([params]: any) => params.role === 'tool' && params.tool_call_id === 'item_1',
      );
      expect(firstToolCreate?.[0]).toMatchObject({
        parentId: 'ast-initial',
        role: 'tool',
        tool_call_id: 'item_1',
      });

      const secondToolCreate = mockCreateMessage.mock.calls.find(
        ([params]: any) => params.role === 'tool' && params.tool_call_id === 'item_3',
      );
      expect(secondToolCreate?.[0]).toMatchObject({
        parentId: newAstId,
        role: 'tool',
        tool_call_id: 'item_3',
      });

      const firstTurnToolWrites = toolsUpdates.filter(
        (update) => update.assistantId === 'ast-initial' && update.toolIds.includes('item_1'),
      );
      expect(firstTurnToolWrites.length).toBeGreaterThanOrEqual(1);

      const secondTurnToolWrites = toolsUpdates.filter(
        (update) => update.assistantId === newAstId && update.toolIds.includes('item_3'),
      );
      expect(secondTurnToolWrites.length).toBeGreaterThanOrEqual(1);
    });

    it('should forward cumulative tools_calling chunks for multiple Codex tools in one step', async () => {
      await runWithEvents(
        [
          codexThreadStarted(),
          codexTurnStarted(),
          codexAgentMessage('item_0', 'Running the first checks.'),
          codexCommandStarted('item_1', '/bin/zsh -lc pwd'),
          codexCommandCompleted('item_1', '/bin/zsh -lc pwd', '/repo\n'),
          codexCommandStarted('item_2', "/bin/zsh -lc 'git status --short'"),
          codexCommandCompleted('item_2', "/bin/zsh -lc 'git status --short'", ' M src/file.ts\n'),
          codexAgentMessage('item_3', 'Now I will inspect the commit details.'),
          codexCommandStarted('item_4', "/bin/zsh -lc 'git show --stat --summary HEAD'"),
          codexCommandCompleted(
            'item_4',
            "/bin/zsh -lc 'git show --stat --summary HEAD'",
            ' src/file.ts | 1 +\n',
          ),
          codexTurnCompleted({ input_tokens: 10, output_tokens: 3 }),
        ],
        {
          params: {
            heterogeneousProvider: { command: 'codex', type: 'codex' as const },
          },
        },
      );

      const handlerSpy = vi.mocked(createGatewayEventHandler).mock.results[0]?.value as ReturnType<
        typeof vi.fn
      >;
      expect(handlerSpy).toBeDefined();

      const toolsCallingChunks = handlerSpy.mock.calls
        .map((call) => call[0])
        .filter(
          (event: any) =>
            event?.type === 'stream_chunk' && event.data?.chunkType === 'tools_calling',
        );

      expect(toolsCallingChunks).toHaveLength(3);
      expect(toolsCallingChunks[0]?.data.toolsCalling.map((tool: any) => tool.id)).toEqual([
        'item_1',
      ]);
      expect(toolsCallingChunks[1]?.data.toolsCalling.map((tool: any) => tool.id)).toEqual([
        'item_1',
        'item_2',
      ]);
      expect(toolsCallingChunks[2]?.data.toolsCalling.map((tool: any) => tool.id)).toEqual([
        'item_4',
      ]);
    });

    it('should cut new assistants for later agent_message items in the same Codex turn', async () => {
      const idCounter = { assistant: 0, tool: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool += 1;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }

        if (params.role === 'assistant') {
          idCounter.assistant += 1;
          return { id: params.id ?? `ast-new-${idCounter.assistant}` };
        }

        return { id: `created-${params.role}-${idCounter.assistant + idCounter.tool}` };
      });

      const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
      const contentUpdates: Array<{ assistantId: string; content: string }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          toolsUpdates.push({
            assistantId: id,
            toolIds: val.tools.map((tool: any) => tool.id),
          });
        }
        if (typeof val.content === 'string') {
          contentUpdates.push({ assistantId: id, content: val.content });
        }
      });

      await runWithEvents(
        [
          codexThreadStarted(),
          codexTurnStarted(),
          codexAgentMessage(
            'item_0',
            'Running the five read-only checks from the repo root exactly as requested.',
          ),
          codexCommandStarted('item_1', '/bin/zsh -lc pwd'),
          codexCommandCompleted('item_1', '/bin/zsh -lc pwd', '/repo\n'),
          codexCommandStarted('item_2', "/bin/zsh -lc 'git status --short'"),
          codexCommandCompleted('item_2', "/bin/zsh -lc 'git status --short'", ' M src/file.ts\n'),
          codexCommandStarted('item_3', "/bin/zsh -lc 'rg --files src | head -n 5'"),
          codexCommandCompleted(
            'item_3',
            "/bin/zsh -lc 'rg --files src | head -n 5'",
            'src/store/session/store.ts\n',
          ),
          codexAgentMessage(
            'item_4',
            'The workspace is dirty in a few files, but I am only collecting read-only outputs.',
          ),
          codexCommandStarted(
            'item_5',
            `/bin/zsh -lc 'rg -n "heterogeneousAgent" src apps packages | head -n 10'`,
          ),
          codexCommandCompleted(
            'item_5',
            `/bin/zsh -lc 'rg -n "heterogeneousAgent" src apps packages | head -n 10'`,
            'apps/desktop/src/main/controllers/HeterogeneousAgentCtr.ts:18:import ...\n',
          ),
          codexCommandStarted(
            'item_6',
            `/bin/zsh -lc 'rg -n "tool_call_id|tool_calls" src packages | head -n 10'`,
          ),
          codexCommandCompleted(
            'item_6',
            `/bin/zsh -lc 'rg -n "tool_call_id|tool_calls" src packages | head -n 10'`,
            'packages/agent-runtime/src/agents/GeneralChatAgent.ts:34:...\n',
          ),
          codexAgentMessage(
            'item_7',
            'Confirmed the repo root and the requested ripgrep checks returned matches.',
          ),
          codexTurnCompleted({
            cached_input_tokens: 92672,
            input_tokens: 107744,
            output_tokens: 996,
          }),
        ],
        {
          params: {
            heterogeneousProvider: { command: 'codex', type: 'codex' as const },
          },
        },
      );

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([params]: any) => params.role === 'assistant',
      );
      expect(assistantCreates).toHaveLength(2);
      const newAst1 = assistantCreates[0]![0].id;
      const newAst2 = assistantCreates[1]![0].id;
      // The cut assistants chain along the spine (each off the previous
      // non-tool main message), NOT off the preceding turn's last tool.
      expect(assistantCreates[0]?.[0]).toMatchObject({
        parentId: 'ast-initial',
        role: 'assistant',
      });
      expect(assistantCreates[1]?.[0]).toMatchObject({
        parentId: newAst1,
        role: 'assistant',
      });

      const firstStepToolWrites = toolsUpdates.filter(
        (update) =>
          update.assistantId === 'ast-initial' &&
          update.toolIds.includes('item_1') &&
          update.toolIds.includes('item_2') &&
          update.toolIds.includes('item_3'),
      );
      expect(firstStepToolWrites.length).toBeGreaterThanOrEqual(1);

      const secondStepToolWrites = toolsUpdates.filter(
        (update) =>
          update.assistantId === newAst1 &&
          update.toolIds.includes('item_5') &&
          update.toolIds.includes('item_6'),
      );
      expect(secondStepToolWrites.length).toBeGreaterThanOrEqual(1);

      const thirdStepToolWrites = toolsUpdates.filter(
        (update) => update.assistantId === newAst2 && update.toolIds.length > 0,
      );
      expect(thirdStepToolWrites).toHaveLength(0);

      expect(
        contentUpdates.findLast((update) => update.assistantId === 'ast-initial')?.content,
      ).toContain('Running the five read-only checks');
      expect(
        contentUpdates.findLast((update) => update.assistantId === newAst1)?.content,
      ).toContain('The workspace is dirty in a few files');
      expect(
        contentUpdates.findLast((update) => update.assistantId === newAst2)?.content,
      ).toContain('Confirmed the repo root');
    });
  });

  // ────────────────────────────────────────────────────
  // Full multi-step E2E
  // ────────────────────────────────────────────────────

  // ────────────────────────────────────────────────────
  // Orphan tool regression (img.png scenario)
  // ────────────────────────────────────────────────────

  describe('orphan tool regression', () => {
    /**
     * Reproduces the orphan tool scenario from img.png:
     *
     * Turn 1 (msg_01): text + Bash(git log)   → assistant1.tools should include git_log
     * tool_result for git log
     * Turn 2 (msg_02): Bash(git diff)          → assistant2.tools should include git_diff
     * tool_result for git diff
     * Turn 3 (msg_03): text summary
     *
     * The orphan happens when assistant2.tools[] does NOT contain
     * the git_diff entry, making the tool message appear orphaned in the UI.
     */
    it('should register tools on the correct assistant in multi-turn tool execution', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      // Track ALL updateMessage calls to inspect tools[] writes
      const toolsUpdates: Array<{ assistantId: string; tools: any[] }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          toolsUpdates.push({ assistantId: id, tools: val.tools });
        }
      });

      await runWithEvents([
        ccInit(),
        // Turn 1: text + Bash (git log) — same message.id
        ccAssistant('msg_01', [
          { text: '没有未提交的修改，看看已提交但未推送的变更：', type: 'text' },
        ]),
        ccToolUse('msg_01', 'toolu_gitlog', 'Bash', { command: 'git log canary..HEAD --oneline' }),
        ccToolResult('toolu_gitlog', 'abc123 feat: something\ndef456 fix: another'),
        // Turn 2: Bash (git diff) — NEW message.id → step boundary
        ccToolUse('msg_02', 'toolu_gitdiff', 'Bash', { command: 'git diff --stat' }),
        ccToolResult('toolu_gitdiff', ' file1.ts | 10 +\n file2.ts | 5 -'),
        // Turn 3: text summary — NEW message.id → step boundary
        ccText('msg_03', '当前分支有2个未推送的提交，修改了2个文件。'),
        ccResult(),
      ]);

      // ── Verify: Turn 1 tool registered on ast-initial ──
      const gitlogToolUpdates = toolsUpdates.filter(
        (u) => u.assistantId === 'ast-initial' && u.tools.some((t: any) => t.id === 'toolu_gitlog'),
      );
      expect(gitlogToolUpdates.length).toBeGreaterThanOrEqual(1);

      // Turn 2's assistant is the first newly-created assistant (pre-allocated id).
      const step2AstId = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant')![0]
        .id;

      // ── Verify: Turn 2 tool registered on the step-2 assistant ──
      // This is the critical assertion — if this fails, the tool becomes orphaned
      const gitdiffToolUpdates = toolsUpdates.filter(
        (u) => u.assistantId === step2AstId && u.tools.some((t: any) => t.id === 'toolu_gitdiff'),
      );
      expect(gitdiffToolUpdates.length).toBeGreaterThanOrEqual(1);

      // ── Verify: tool messages have correct parentId ──
      const gitlogToolCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_gitlog',
      );
      expect(gitlogToolCreate![0].parentId).toBe('ast-initial');

      const gitdiffToolCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_gitdiff',
      );
      expect(gitdiffToolCreate![0].parentId).toBe(step2AstId);
    });

    it('should register tools on correct assistant when turn has ONLY tool_use (no text)', async () => {
      // Edge case: turn 2 has only a tool_use, no text. The step transition creates
      // a new assistant, then the tool_use must be registered on it (not the old one).
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          toolsUpdates.push({
            assistantId: id,
            toolIds: val.tools.map((t: any) => t.id),
          });
        }
      });

      await runWithEvents([
        ccInit(),
        // Turn 1: just text, no tools
        ccText('msg_01', 'Let me check...'),
        // Turn 2: only tool_use (no text in this turn)
        ccToolUse('msg_02', 'toolu_bash', 'Bash', { command: 'ls -la' }),
        ccToolResult('toolu_bash', 'total 100\ndrwx...'),
        // Turn 3: final text
        ccText('msg_03', 'Done.'),
        ccResult(),
      ]);

      // The tool should be registered on the step-2 assistant, not ast-initial.
      const step2AstId = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant')![0]
        .id;
      const bashToolUpdates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_bash'));
      expect(bashToolUpdates.length).toBeGreaterThanOrEqual(1);
      // All of them should be on the step-2 assistant
      for (const u of bashToolUpdates) {
        expect(u.assistantId).toBe(step2AstId);
      }
    });
  });

  // ────────────────────────────────────────────────────
  // Real trace regression: multi-tool per turn (scenario)
  // ────────────────────────────────────────────────────

  describe('multi-tool per turn (real trace regression)', () => {
    /**
     * Reproduces the exact CC event pattern from the orphan trace.
     * Key pattern: a single turn (same message.id) has text + multiple tool_uses.
     * After step transition, the new turn also has multiple tool_uses with
     * out-of-order tool_results.
     */
    it('should register ALL tools on correct assistant when turn has text + multiple tool_uses', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          toolsUpdates.push({
            assistantId: id,
            toolIds: val.tools.map((t: any) => t.id),
          });
        }
      });

      await runWithEvents([
        ccInit(),
        // Turn 1 (msg_01): thinking + tool (Skill)
        ccThinking('msg_01', 'Let me check the issue'),
        ccToolUse('msg_01', 'toolu_skill', 'Skill', { skill: 'linear' }),
        ccToolResult('toolu_skill', 'Launching skill: linear'),

        // Turn 2 (msg_02): tool (ToolSearch) — step boundary
        ccToolUse('msg_02', 'toolu_search', 'ToolSearch', { query: 'select:get_issue' }),
        ccToolResult('toolu_search', 'tool loaded'),

        // Turn 3 (msg_03): tool (get_issue) — step boundary
        ccToolUse('msg_03', 'toolu_getissue', 'mcp__linear__get_issue', { id: '' }),
        ccToolResult('toolu_getissue', '{"title":"i18n"}'),

        // Turn 4 (msg_04): thinking + text + Grep + Grep — step boundary
        // This is the critical pattern: same message.id has text AND multiple tools
        ccThinking('msg_04', 'Let me understand the issue'),
        ccText('msg_04', '明白了，需要补充翻译'),
        ccToolUse('msg_04', 'toolu_grep1', 'Grep', { pattern: 'newClaudeCodeAgent' }),
        ccToolResult('toolu_grep1', 'found in chat.ts'),
        ccToolUse('msg_04', 'toolu_grep2', 'Grep', { pattern: 'agentProvider' }),
        ccToolResult('toolu_grep2', 'found in setting.ts'),

        // Turn 5 (msg_05): Grep + Glob + Glob — step boundary
        // Multiple tools, results may arrive out of order
        ccToolUse('msg_05', 'toolu_grep3', 'Grep', { pattern: 'agentProvider', path: 'locales' }),
        ccToolResult('toolu_grep3', 'locales content'),
        ccToolUse('msg_05', 'toolu_glob1', 'Glob', { pattern: 'zh-CN/chat.json' }),
        ccToolUse('msg_05', 'toolu_glob2', 'Glob', { pattern: 'en-US/chat.json' }),
        // Results arrive out of order: glob2 before glob1
        ccToolResult('toolu_glob2', 'locales/en-US/chat.json'),
        ccToolResult('toolu_glob1', 'locales/zh-CN/chat.json'),

        // Turn 6 (msg_06): text summary — step boundary
        ccText('msg_06', 'All translations updated.'),
        ccResult(),
      ]);

      // ── Verify Turn 1: Skill tool on ast-initial ──
      const skillUpdates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_skill'));
      expect(skillUpdates.length).toBeGreaterThanOrEqual(1);
      expect(skillUpdates.every((u) => u.assistantId === 'ast-initial')).toBe(true);

      // ── Verify Turn 4: BOTH Grep tools on same assistant (ast-new-3) ──
      const grep1Updates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_grep1'));
      const grep2Updates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_grep2'));
      expect(grep1Updates.length).toBeGreaterThanOrEqual(1);
      expect(grep2Updates.length).toBeGreaterThanOrEqual(1);

      // Both Grep tools must be registered on the SAME assistant
      const turn4AssistantId = grep1Updates[0].assistantId;
      expect(grep2Updates.some((u) => u.assistantId === turn4AssistantId)).toBe(true);

      // The final tools[] update for Turn 4's assistant should contain BOTH greps
      const turn4FinalUpdate = toolsUpdates.findLast((u) => u.assistantId === turn4AssistantId);
      expect(turn4FinalUpdate!.toolIds).toContain('toolu_grep1');
      expect(turn4FinalUpdate!.toolIds).toContain('toolu_grep2');

      // ── Verify Turn 5: all 3 tools (Grep + 2 Globs) on same assistant ──
      const grep3Updates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_grep3'));
      const glob1Updates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_glob1'));
      const glob2Updates = toolsUpdates.filter((u) => u.toolIds.includes('toolu_glob2'));
      expect(grep3Updates.length).toBeGreaterThanOrEqual(1);
      expect(glob1Updates.length).toBeGreaterThanOrEqual(1);
      expect(glob2Updates.length).toBeGreaterThanOrEqual(1);

      // All three must be on the SAME assistant (Turn 5's assistant)
      const turn5AssistantId = grep3Updates[0].assistantId;
      expect(turn5AssistantId).not.toBe(turn4AssistantId); // Different from Turn 4
      expect(glob1Updates.some((u) => u.assistantId === turn5AssistantId)).toBe(true);
      expect(glob2Updates.some((u) => u.assistantId === turn5AssistantId)).toBe(true);

      // Final tools[] for Turn 5's assistant should contain all 3
      const turn5FinalUpdate = toolsUpdates.findLast((u) => u.assistantId === turn5AssistantId);
      expect(turn5FinalUpdate!.toolIds).toContain('toolu_grep3');
      expect(turn5FinalUpdate!.toolIds).toContain('toolu_glob1');
      expect(turn5FinalUpdate!.toolIds).toContain('toolu_glob2');

      // ── Verify tool messages have correct parentId ──
      // Turn 4 tools should be children of Turn 4's assistant
      const grep1Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_grep1',
      );
      const grep2Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_grep2',
      );
      expect(grep1Create![0].parentId).toBe(turn4AssistantId);
      expect(grep2Create![0].parentId).toBe(turn4AssistantId);

      // Turn 5 tools should be children of Turn 5's assistant
      const grep3Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_grep3',
      );
      const glob1Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_glob1',
      );
      const glob2Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_glob2',
      );
      expect(grep3Create![0].parentId).toBe(turn5AssistantId);
      expect(glob1Create![0].parentId).toBe(turn5AssistantId);
      expect(glob2Create![0].parentId).toBe(turn5AssistantId);
    });

    /**
     * Regression: when a turn has text BEFORE tool_use under the same message.id,
     * the tools[] write must carry the accumulated content too. Otherwise the
     * gateway handler's `tool_end → fetchAndReplaceMessages` reads a tools-only
     * row and clobbers the in-memory streamed text in the UI.
     */
    it('should persist accumulated text alongside tools when turn has text + tool_use', async () => {
      const writes: Array<{ assistantId: string; content?: string; toolIds?: string[] }> = [];
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools) {
          writes.push({
            assistantId: id,
            content: val.content,
            toolIds: val.tools.map((t: any) => t.id),
          });
        }
      });

      await runWithEvents([
        ccInit(),
        // text streams first, then tool_use — same msg.id
        ccText('msg_01', 'Let me check the file...'),
        ccToolUse('msg_01', 'toolu_read', 'Read', { file_path: '/a.ts' }),
        ccToolResult('toolu_read', 'file content'),
        ccResult(),
      ]);

      const toolWrites = writes.filter((w) => w.toolIds?.includes('toolu_read'));
      expect(toolWrites.length).toBeGreaterThanOrEqual(1);
      // Every tools[] write for this assistant must carry the accumulated text
      for (const w of toolWrites) {
        expect(w.content).toBe('Let me check the file...');
      }
    });
  });

  // ────────────────────────────────────────────────────
  // Data-driven regression from real trace (regression.json)
  // ────────────────────────────────────────────────────

  describe('data-driven regression (133 events)', () => {
    it('should have no orphan tools when replaying real CC trace', async () => {
      // Load real trace data
      const fs = await import('node:fs');
      const tracePath = path.join(process.cwd(), 'regression.json');

      let traceData: any[];
      try {
        traceData = JSON.parse(fs.readFileSync(tracePath, 'utf8'));
      } catch {
        // Skip if file doesn't exist (CI)
        console.log('regression.json not found, skipping data-driven test');
        return;
      }

      // Track all createMessage and updateMessage calls
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-${idCounter.assistant}` };
      });

      // Collect tools[] writes per assistant
      const toolsRegistry = new Map<string, Set<string>>();
      mockUpdateMessage.mockImplementation(async (id: string, val: any) => {
        if (val.tools && Array.isArray(val.tools)) {
          if (!toolsRegistry.has(id)) toolsRegistry.set(id, new Set());
          const set = toolsRegistry.get(id)!;
          for (const t of val.tools) {
            if (t.id) set.add(t.id);
          }
        }
      });

      // Collect tool messages: { tool_call_id → parentId (assistant) }
      const toolMessages = new Map<string, string>();
      const origCreate = mockCreateMessage.getMockImplementation()!;
      mockCreateMessage.mockImplementation(async (params: any) => {
        const result = await origCreate(params);
        if (params.role === 'tool' && params.tool_call_id) {
          toolMessages.set(params.tool_call_id, params.parentId);
        }
        return result;
      });

      // Extract raw lines from trace
      const rawLines = traceData.map((entry: any) => entry.rawLine);

      await runWithEvents(rawLines);

      // ── Check for orphans ──
      // An orphan is a tool message whose tool_call_id doesn't appear in ANY
      // assistant's tools[] registry
      const allRegisteredToolIds = new Set<string>();
      for (const toolIds of toolsRegistry.values()) {
        for (const id of toolIds) allRegisteredToolIds.add(id);
      }

      const orphans: string[] = [];
      for (const [toolCallId, parentId] of toolMessages) {
        if (!allRegisteredToolIds.has(toolCallId)) {
          orphans.push(`tool_call_id=${toolCallId} parentId=${parentId}`);
        }
      }

      if (orphans.length > 0) {
        console.error('Orphan tools found:', orphans);
      }
      expect(orphans).toEqual([]);

      // ── Sanity checks ──
      // Should have created many tool messages (trace has ~60 tool calls)
      expect(toolMessages.size).toBeGreaterThan(20);
      // Should have many assistants
      expect(idCounter.assistant).toBeGreaterThan(10);
    });
  });

  // ────────────────────────────────────────────────────
  // reproduction: Skill → ToolSearch → MCP tool
  //
  // Mirrors the exact trace from the user-reported screenshot where
  // ToolSearch loads deferred MCP schemas before the MCP tool is called.
  // Verifies tool_result content is persisted for ALL three tools so the
  // UI stops showing "loading" after each tool completes.
  // ────────────────────────────────────────────────────

  describe('Skill → ToolSearch → MCP repro', () => {
    it('persists tool_result content for Skill, ToolSearch, and the deferred MCP tool', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      const schemaPayload =
        '<functions><function>{"description":"Get a Linear issue","name":"mcp__linear-server__get_issue","parameters":{}}</function></functions>';

      await runWithEvents([
        ccInit(),
        // Turn 1: Skill invocation
        ccToolUse('msg_01', 'toolu_skill', 'Skill', { skill: 'linear' }),
        ccToolResult('toolu_skill', 'Launching skill: linear'),
        // Turn 2: ToolSearch with select: prefix (deferred schema fetch)
        ccToolUse('msg_02', 'toolu_search', 'ToolSearch', {
          query: 'select:mcp__linear-server__get_issue,mcp__linear-server__save_issue',
          max_results: 3,
        }),
        ccToolResult('toolu_search', schemaPayload),
        // Turn 3: the deferred MCP tool now callable
        ccToolUse('msg_03', 'toolu_get_issue', 'mcp__linear-server__get_issue', {
          id: '',
        }),
        ccToolResult('toolu_get_issue', '{"title":"resume error on topic switch"}'),
        ccText('msg_04', 'done'),
        ccResult(),
      ]);

      const toolIdOf = (callId: string) =>
        mockCreateMessage.mock.calls.find(
          ([p]: any) => p.role === 'tool' && p.tool_call_id === callId,
        )![0].id;
      const astCreates = mockCreateMessage.mock.calls.filter(([p]: any) => p.role === 'assistant');
      const step2AstId = astCreates[0]![0].id;
      const step3AstId = astCreates[1]![0].id;

      // All three tool messages should have their content persisted.
      const skillResult = mockUpdateToolMessage.mock.calls.find(
        ([id]: any) => id === toolIdOf('toolu_skill'),
      );
      const searchResult = mockUpdateToolMessage.mock.calls.find(
        ([id]: any) => id === toolIdOf('toolu_search'),
      );
      const getIssueResult = mockUpdateToolMessage.mock.calls.find(
        ([id]: any) => id === toolIdOf('toolu_get_issue'),
      );

      expect(skillResult).toBeDefined();
      expect(skillResult![1]).toMatchObject({ content: 'Launching skill: linear' });

      expect(searchResult).toBeDefined();
      expect(searchResult![1]).toMatchObject({ content: schemaPayload });
      expect(searchResult![1].pluginError).toBeUndefined();

      expect(getIssueResult).toBeDefined();
      expect(getIssueResult![1]).toMatchObject({
        content: '{"title":"resume error on topic switch"}',
      });

      // tools[] registry on each step should contain the right tool id so the
      // UI can match tool messages to their assistant (no orphan warnings).
      const skillRegister = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) =>
          id === 'ast-initial' && val.tools?.some((t: any) => t.id === 'toolu_skill'),
      );
      expect(skillRegister).toBeDefined();

      const searchRegister = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) =>
          id === step2AstId && val.tools?.some((t: any) => t.id === 'toolu_search'),
      );
      expect(searchRegister).toBeDefined();

      const getIssueRegister = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) =>
          id === step3AstId && val.tools?.some((t: any) => t.id === 'toolu_get_issue'),
      );
      expect(getIssueRegister).toBeDefined();
    });
  });

  // ────────────────────────────────────────────────────
  // Full multi-step E2E
  // ────────────────────────────────────────────────────

  describe('full multi-step E2E', () => {
    it('should produce correct DB write sequence for Read → Write → text flow', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Turn 1: Read tool
        ccAssistant('msg_01', [{ thinking: 'Need to read the file', type: 'thinking' }]),
        ccToolUse('msg_01', 'toolu_read', 'Read', { file_path: '/src/app.ts' }),
        ccToolResult('toolu_read', 'export default function App() {}'),
        // Turn 2: Write tool (new message.id)
        ccToolUse('msg_02', 'toolu_write', 'Write', { file_path: '/src/app.ts', content: 'fixed' }),
        ccToolResult('toolu_write', 'File written'),
        // Turn 3: final summary (new message.id)
        ccText('msg_03', 'Fixed the bug in app.ts.'),
        ccResult(),
      ]);

      // --- Verify DB write sequence ---

      // 1. Tool message created for Read (parentId = initial assistant)
      const readToolCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_read',
      );
      expect(readToolCreate![0].parentId).toBe('ast-initial');
      expect(readToolCreate![0].plugin.apiName).toBe('Read');
      const readToolId = readToolCreate![0].id;

      // 2. Read tool result written (to the pre-allocated Read tool message id)
      expect(mockUpdateToolMessage).toHaveBeenCalledWith(
        readToolId,
        expect.objectContaining({ content: 'export default function App() {}' }),
        expect.any(Object),
      );

      // 3. Step 2 assistant created chained off the spine (the initial
      // assistant), with the Read tool inline under step 1.
      const step2Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.parentId === 'ast-initial',
      );
      expect(step2Create).toBeDefined();

      // 4. Write tool message created (parentId = step 2 assistant)
      const writeToolCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_write',
      );
      expect(writeToolCreate).toBeDefined();
      expect(writeToolCreate![0].parentId).toBe(step2Create![0].id);
      const writeToolId = writeToolCreate![0].id;

      // 5. Write tool result written
      expect(mockUpdateToolMessage).toHaveBeenCalledWith(
        writeToolId,
        expect.objectContaining({ content: 'File written' }),
        expect.any(Object),
      );

      // 6. Step 3 assistant created chained off the spine (step 2 assistant),
      // with the Write tool inline under step 2.
      const step3Create = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.parentId === step2Create![0].id,
      );
      expect(step3Create).toBeDefined();

      // 7. Final content written to the last assistant message
      const finalContentWrite = mockUpdateMessage.mock.calls.find(
        ([, val]: any) => val.content === 'Fixed the bug in app.ts.',
      );
      expect(finalContentWrite).toBeDefined();
    });
  });

  // ────────────────────────────────────────────────────
  // CC subagent thread-container model ()
  //
  // A subagent Thread is shaped as a nested conversation:
  //   user (prompt) → assistant#1 (tools[]) → tool → assistant#2 (tools[]) → tool → ...
  //
  // The executor creates the Thread lazily on the FIRST subagent event
  // (the adapter announces spawn metadata on that chunk), seeds it with
  // a `role:'user'` message from the Task prompt, then appends an
  // `assistant` message per subagent turn boundary (new `subagentMessageId`).
  //
  // Main assistant.tools[] only ever carries the outer Task tool_use —
  // subagent inner tools live on the in-thread assistants' tools[].
  // ────────────────────────────────────────────────────

  describe('CC subagent thread-container', () => {
    it('does NOT create a Thread on Task tool_use alone (lazy creation)', async () => {
      // Task tool_use without any subagent events should NOT trigger
      // Thread creation — we only know the spawn is real once the
      // adapter starts announcing subagent events.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'Find failing tests',
          prompt: 'run the suite',
          subagent_type: 'Explore',
        }),
        ccResult(),
      ]);

      expect(mockCreateThread).not.toHaveBeenCalled();
    });

    it('creates Thread + user + assistant messages on FIRST subagent event', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'Find failing tests',
          prompt: 'run the suite and list failures',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Bash'),
        ccResult(),
      ]);

      // Thread row seeded with adapter-supplied metadata.
      expect(mockCreateThread).toHaveBeenCalledTimes(1);
      expect(mockCreateThread).toHaveBeenCalledWith(
        expect.objectContaining({
          id: expect.stringMatching(/^thd_/),
          metadata: expect.objectContaining({
            sourceToolCallId: 'toolu_task',
            subagentType: 'Explore',
            startedAt: expect.any(String),
          }),
          sourceMessageId: 'ast-initial',
          title: 'Find failing tests',
          topicId: 'topic-1',
          type: 'isolation',
        }),
      );
      const threadId = mockCreateThread.mock.calls[0][0].id;

      // Thread gets a `role:'user'` message seeded with the Task prompt.
      const userMsg = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'user' && p.threadId === threadId,
      );
      expect(userMsg).toBeDefined();
      expect(userMsg![0]).toMatchObject({
        content: 'run the suite and list failures',
        parentId: 'ast-initial',
        threadId,
      });

      // Thread gets at least one `role:'assistant'` message scoped to
      // the subagent's first turn.
      const subAssistant = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.threadId === threadId,
      );
      expect(subAssistant).toBeDefined();
    });

    it('chains subagent inner tool messages to the in-thread assistant', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'inspect',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Bash', { command: 'ls' }),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const subAssistantMsg = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.threadId === threadId,
      );
      const subToolCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_child',
      );
      expect(subToolCreate).toBeDefined();
      expect(subToolCreate![0]).toMatchObject({
        role: 'tool',
        threadId,
        tool_call_id: 'toolu_child',
        plugin: expect.objectContaining({ apiName: 'Bash' }),
      });
      // Tool messages chain under the in-thread assistant (not the main one).
      expect(subToolCreate![0].parentId).not.toBe('ast-initial');
      // The in-thread assistant + tool messages share the same threadId.
      expect(subAssistantMsg![0]).toMatchObject({ threadId });
    });

    it('opens a NEW in-thread assistant when subagentMessageId changes (turn boundary)', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', { description: 'x', subagent_type: 'Plan' }),
        // Turn 1
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child_1', 'Read'),
        // Turn 2 — new message.id for subagent
        ccSubagentToolUse('msg_sub_2', 'toolu_task', 'toolu_child_2', 'Write'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const threadAssistants = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant' && p.threadId === threadId,
      );
      // One assistant per subagent turn — same shape as the main topic.
      expect(threadAssistants.length).toBeGreaterThanOrEqual(2);
    });

    it('routes delayed tool_result to thread bucket when it arrives after subagent turn has rolled over', async () => {
      // Regression: `findRunByInnerToolCallId` must resolve across ALL
      // turns of a subagent run, not just the current one. Previously
      // it only consulted `state.persistedIds`, which `ensureSubagentRun`
      // wipes on every turn advance — so a `tool_result` for a prior
      // turn's `tool_use` silently skipped `run.stream.update` and left
      // the in-thread tool bubble stuck on its loading spinner until the
      // user re-opened the Thread (main-topic `fetchAndReplaceMessages`
      // does not rehydrate thread buckets).
      const { store } = await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', { description: 'x', subagent_type: 'Plan' }),
        // Turn 1: lifetimeToolCallIds gets `toolu_child_1`.
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child_1', 'Read'),
        // Turn 2: ensureSubagentRun wipes state.persistedIds. The
        // run-lifetime set must still remember `toolu_child_1`.
        ccSubagentToolUse('msg_sub_2', 'toolu_task', 'toolu_child_2', 'Write'),
        // Delayed tool_result for the FIRST turn's tool_use.
        ccToolResult('toolu_child_1', 'turn 1 output'),
        ccResult(),
      ]);

      const startOpMock = (store.startOperation as ReturnType<typeof vi.fn>).mock;
      const subOpIdx = startOpMock.calls.findIndex(([p]: any) => p?.type === 'subagentThread');
      expect(subOpIdx).toBeGreaterThanOrEqual(0);
      const subOperationId = startOpMock.results[subOpIdx].value.operationId;

      const dispatches = (store.internal_dispatchMessage as ReturnType<typeof vi.fn>).mock.calls;
      const delayedResultDispatch = dispatches.find(
        ([payload, ctx]: any) =>
          ctx?.operationId === subOperationId &&
          payload.type === 'updateMessage' &&
          payload.value?.content === 'turn 1 output',
      );
      expect(delayedResultDispatch).toBeDefined();
    });

    it('records subagent tool_uses on IN-THREAD assistant tools[], not on main', async () => {
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', { description: 'x', subagent_type: 'Plan' }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Read'),
        ccResult(),
      ]);

      // Main assistant.tools[] only ever carries the outer Task tool_use.
      const mainAssistantToolWrites = mockUpdateMessage.mock.calls.filter(
        ([id, val]: any) => id === 'ast-initial' && val.tools?.length > 0,
      );
      for (const [, val] of mainAssistantToolWrites) {
        const ids = val.tools.map((t: any) => t.id);
        expect(ids).toContain('toolu_task');
        expect(ids).not.toContain('toolu_child');
      }

      // In-thread assistant should receive an updateMessage whose tools[]
      // includes the subagent's inner tool_use.
      const threadAssistantIds = new Set(
        mockCreateMessage.mock.calls
          .filter(
            ([p]: any) =>
              p.role === 'assistant' &&
              typeof p.threadId === 'string' &&
              p.threadId.startsWith('thd_'),
          )
          .map(([, returnValue]: any) => returnValue?.id),
      );
      // The mock returns a generated id — we can't match exactly, but we
      // can assert SOME update carried toolu_child on a non-main assistant.
      const subToolUpdateLanded = mockUpdateMessage.mock.calls.some(
        ([id, val]: any) =>
          id !== 'ast-initial' && val.tools?.some((t: any) => t.id === 'toolu_child'),
      );
      expect(subToolUpdateLanded).toBe(true);
      // (threadAssistantIds unused here but kept to document the intent.)
      expect(threadAssistantIds.size).toBeGreaterThan(0);
    });

    it('does NOT create a Thread when topicId is missing (non-topic-scoped run)', async () => {
      await runWithEvents(
        [
          ccInit(),
          ccToolUse('msg_main', 'toolu_task', 'Task', {
            description: 'x',
            subagent_type: 'Plan',
          }),
          ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Read'),
          ccResult(),
        ],
        {
          params: {
            context: { ...defaultContext, topicId: undefined as any },
          },
        },
      );

      expect(mockCreateThread).not.toHaveBeenCalled();
    });

    it('persists the subagent Thread user message content from spawnMetadata.prompt', async () => {
      // Real CC uses `Agent` for general-purpose subagents — the adapter
      // should still extract `prompt` from the input and seed the user
      // message content with it (earlier bug: user msg content empty).
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Agent', {
          description: 'lookup pwd',
          prompt: 'run pwd and summarize',
          subagent_type: 'general-purpose',
        }),
        ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Bash'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const threadUser = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'user' && p.threadId === threadId,
      );
      expect(threadUser).toBeDefined();
      expect(threadUser![0].content).toBe('run pwd and summarize');
    });

    it('accumulates subagent text into the in-thread assistant content', async () => {
      // Subagent emits a closing summary text turn after its tool work.
      // The thread should reflect it as the assistant's content so the
      // Thread view reads as a complete conversation.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Bash'),
        ccSubagentText('msg_sub_2', 'toolu_task', 'Here is the summary.'),
        ccSubagentSpawnResult('toolu_task', 'Final answer to main.'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      // At least one updateMessage on a subagent-thread assistant with
      // content = "Here is the summary." should have landed.
      const threadAssistantContentWrites = mockUpdateMessage.mock.calls.filter(
        ([id, val]: any) => id !== 'ast-initial' && val.content === 'Here is the summary.',
      );
      expect(threadAssistantContentWrites.length).toBeGreaterThan(0);
      expect(threadAssistantContentWrites[0][2]).toMatchObject({ topicId: 'topic-1' });
      // Sanity — the in-thread assistants exist under the right thread.
      const threadAssistants = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant' && p.threadId === threadId,
      );
      expect(threadAssistants.length).toBeGreaterThanOrEqual(1);
    });

    it('does NOT leak subagent text into main assistant accumulatedContent', async () => {
      await runWithEvents([
        ccInit(),
        ccText('msg_main_pre', 'I will delegate.'),
        ccToolUse('msg_main_pre', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentText('msg_sub', 'toolu_task', 'Subagent saying things'),
        ccSubagentSpawnResult('toolu_task', 'done'),
        ccResult(),
      ]);

      // Main assistant content writes should contain "I will delegate."
      // but NEVER "Subagent saying things".
      const mainContentWrites = mockUpdateMessage.mock.calls.filter(
        ([id, val]: any) => id === 'ast-initial' && typeof val.content === 'string',
      );
      for (const [, val] of mainContentWrites) {
        expect(val.content).not.toContain('Subagent saying things');
      }
    });

    it('finalizes subagent content when the spawn tool_result lands on main', async () => {
      // Subagent emits text but no subsequent event — normally content
      // only hits DB on the next persist or at onComplete. The spawn
      // tool_result arriving on main should trigger an explicit flush
      // so the final text lands in DB before fetchAndReplace.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentText('msg_sub', 'toolu_task', 'final summary text'),
        ccSubagentSpawnResult('toolu_task', 'returned to main'),
        ccResult(),
      ]);

      const finalizeWrite = mockUpdateMessage.mock.calls.find(
        ([id, val]: any) => id !== 'ast-initial' && val.content === 'final summary text',
      );
      expect(finalizeWrite).toBeDefined();
    });

    it('marks the subagent thread Active on finalize without denormalizing metrics', async () => {
      // Under read-time derivation the chip metrics (tool count / tokens /
      // model) are NOT written onto `thread.metadata` at finalize — they're
      // aggregated from the child messages on read. Finalize only flips the
      // thread status Processing → Active.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Bash', { command: 'ls' }),
        ccSubagentToolResult('toolu_child', 'toolu_task', 'file list'),
        {
          message: {
            content: [{ text: 'summary', type: 'text' }],
            id: 'msg_sub_2',
            model: 'claude-opus-4-8',
            role: 'assistant',
            usage: { input_tokens: 1000, output_tokens: 200 },
          },
          parent_tool_use_id: 'toolu_task',
          type: 'assistant',
        },
        ccSubagentSpawnResult('toolu_task', 'final answer'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const finalize = mockUpdateThread.mock.calls.find(([id]: any) => id === threadId);
      expect(finalize).toBeDefined();
      // Status-only — no metrics denormalized onto metadata.
      expect(finalize![1]).toEqual({ status: ThreadStatus.Active });
    });

    it('writes the subagent model onto the in-thread assistant for the live tooltip', async () => {
      // The chip tooltip derives the model from the child assistant's `model`
      // field live (before finalize). The turn_metadata branch must persist it.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        {
          message: {
            content: [{ text: 'summary', type: 'text' }],
            id: 'msg_sub',
            model: 'claude-opus-4-8',
            role: 'assistant',
            usage: { input_tokens: 1000, output_tokens: 200 },
          },
          parent_tool_use_id: 'toolu_task',
          type: 'assistant',
        },
        ccSubagentSpawnResult('toolu_task', 'final answer'),
        ccResult(),
      ]);

      const modelWrite = mockUpdateMessage.mock.calls.find(
        ([, val]: any) => val.model === 'claude-opus-4-8' && val.metadata?.usage,
      );
      expect(modelWrite).toBeDefined();
      expect(modelWrite![1].provider).toBe('claude-code');
    });

    it('retains subagent buffers + pinned target when the finalize flush fails', async () => {
      // Transient DB failures on the finalize-time flush used to silently
      // wipe the accumulators (buffer clear was outside the try/catch), so
      // the onComplete fallback had nothing left to retry. With buffers
      // preserved AND the flush target pinned, the retry writes the
      // leftover stream text to the ORIGINAL in-thread assistant — not to
      // the terminal message `resultContent` already advanced
      // `currentAssistantMsgId` onto.
      const idCounter = { tool: 0, assistant: 0, user: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        if (params.role === 'user') {
          idCounter.user++;
          return { id: `thread-user-${idCounter.user}` };
        }
        idCounter.assistant++;
        return { id: `thread-ast-${idCounter.assistant}` };
      });

      // Make the FIRST write targeting the in-thread streaming assistant
      // fail. The `content === 'streamed text'` check pins the rejection
      // to the finalize-time flush; the onComplete retry uses the same
      // update shape and must succeed.
      let failed = false;
      mockUpdateMessage.mockImplementation(async (_id: string, val: any) => {
        if (!failed && val.content === 'streamed text') {
          failed = true;
          throw new Error('transient DB failure');
        }
      });

      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentText('msg_sub', 'toolu_task', 'streamed text'),
        ccSubagentSpawnResult('toolu_task', 'terminal result'),
        ccResult(),
      ]);

      // Ids are now caller-pre-allocated (the coordinator assigns them and the
      // interpreter creates rows WITH that id), so we resolve the two relevant
      // in-thread assistants by their create payloads rather than a literal id:
      //   - the FIRST streaming-turn assistant (role assistant, empty content)
      //   - the terminal assistant (role assistant, the resultContent)
      const firstAssistantId = mockCreateMessage.mock.calls.find(
        ([p]: any) =>
          p.role === 'assistant' &&
          p.content === '' &&
          typeof p.threadId === 'string' &&
          p.threadId.startsWith('thd_'),
      )?.[0].id;
      const terminalCreate = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.content === 'terminal result',
      );
      expect(firstAssistantId).toBeDefined();
      // Terminal assistant carrying the authoritative `resultContent` was still
      // created as a fresh row (not overwritten by the retry), with its OWN id.
      expect(terminalCreate).toBeDefined();
      expect(terminalCreate![0].id).not.toBe(firstAssistantId);

      const streamedWrites = mockUpdateMessage.mock.calls.filter(
        ([, val]: any) => val.content === 'streamed text',
      );
      // At least the retry must have landed after the original failure.
      expect(streamedWrites.length).toBeGreaterThanOrEqual(2);
      // Every attempt — including the retry — must target the FIRST streaming
      // turn's assistant, NOT the terminal row, so the authoritative
      // `resultContent` is never clobbered by the leftover streamed buffer.
      for (const [id] of streamedWrites) {
        expect(id).toBe(firstAssistantId);
      }
    });

    it('retries the lazy thread create on the next event when a create intent fails (commit-on-success)', async () => {
      // A transient failure on createThread / createMessage must NOT advance
      // the coordinator state: committing would make the run look alive while
      // nothing landed, so later chunks could never recreate the thread and
      // the streamed content would be orphaned. With commit-on-success the
      // failed event is dropped and the NEXT subagent event re-runs the lazy
      // create end-to-end with fresh ids.
      mockCreateThread.mockRejectedValueOnce(new Error('transient IndexedDB failure'));

      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        // First subagent event: createThread throws → state must not commit.
        ccSubagentText('msg_sub', 'toolu_task', 'lost '),
        // Next subagent event retries the lazy create.
        ccSubagentText('msg_sub', 'toolu_task', 'kept'),
        ccSubagentSpawnResult('toolu_task', 'terminal result'),
        ccResult(),
      ]);

      // Lazy create attempted twice — first failed, retry landed with a fresh id.
      expect(mockCreateThread).toHaveBeenCalledTimes(2);
      const retryThreadId = mockCreateThread.mock.calls[1][0].id;
      expect(retryThreadId).not.toBe(mockCreateThread.mock.calls[0][0].id);

      // Every in-thread row (seed user, turn assistant, terminal assistant)
      // belongs to the retried thread — nothing was written against the
      // thread whose create failed.
      const threadCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => typeof p.threadId === 'string' && p.threadId.startsWith('thd_'),
      );
      expect(threadCreates.length).toBeGreaterThan(0);
      for (const [p] of threadCreates) {
        expect(p.threadId).toBe(retryThreadId);
      }

      // The run still finalized normally inside the retried thread.
      const terminal = mockCreateMessage.mock.calls.find(
        ([p]: any) => p.role === 'assistant' && p.content === 'terminal result',
      );
      expect(terminal).toBeDefined();
      expect(terminal![0].threadId).toBe(retryThreadId);
    });

    it('creates a terminal in-thread assistant with the main tool_result content', async () => {
      // CC never emits the subagent's final summary as a
      // `parent_tool_use_id`-tagged assistant event — the summary only
      // exists on the main side, as the `tool_result.content` of the
      // Agent spawn. Without an explicit terminal-assistant write, the
      // Thread ends mid-conversation (last message = a tool), and the
      // user has no visible result inside the subagent thread.
      //
      // This test covers the pure-tools subagent case (no inner text
      // event) to prove we still get a terminal summary message.
      const spawnResult = 'Here is what I found: ...';
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Bash', { command: 'ls' }),
        ccToolResult('toolu_child', 'ls output'),
        ccSubagentSpawnResult('toolu_task', spawnResult),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const terminalCreate = mockCreateMessage.mock.calls.find(
        ([payload]: any) =>
          payload.role === 'assistant' &&
          payload.threadId === threadId &&
          payload.content === spawnResult,
      );
      expect(terminalCreate).toBeDefined();

      // Terminal message chains off the in-thread assistant (the subagent
      // spine), with the child tool inline, so the transcript flows
      // user → asst(tools) → asst(result) — the subagent reducer keeps the
      // chain anchor on the assistant instead of advancing it to the tool.
      const toolCreate = mockCreateMessage.mock.calls.find(
        ([payload]: any) => payload.role === 'tool' && payload.tool_call_id === 'toolu_child',
      );
      expect(toolCreate).toBeDefined();
      const firstAssistantCreate = mockCreateMessage.mock.calls.find(
        ([payload]: any) => payload.role === 'assistant' && payload.threadId === threadId,
      );
      expect(terminalCreate![0].parentId).toBe(firstAssistantCreate![0].id);
    });

    it('streams the terminal assistant into the thread messagesMap bucket', async () => {
      // UI relies on internal_dispatchMessage to see the terminal
      // message arrive in the thread bucket — otherwise the Thread view
      // only picks it up on re-open / SWR refresh.
      const spawnResult = 'Final handoff text.';
      const { store } = await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Bash'),
        ccSubagentSpawnResult('toolu_task', spawnResult),
        ccResult(),
      ]);

      // The thread-scoped sub-op opened by `beginSubagentRun` is the
      // dispatchCtx.operationId for every Thread bucket dispatch — so
      // we identify it via the `startOperation` call with type
      // 'subagentThread' and filter dispatches against that id, instead
      // of inspecting threadId directly (the threadId override at the
      // dispatch boundary is gone — context flows through the standard
      // operation registry path now).
      const startOpCalls = (store.startOperation as ReturnType<typeof vi.fn>).mock;
      const subOpIdx = startOpCalls.calls.findIndex(([p]: any) => p?.type === 'subagentThread');
      expect(subOpIdx).toBeGreaterThanOrEqual(0);
      const subOperationId = startOpCalls.results[subOpIdx].value.operationId;

      const dispatches = (store.internal_dispatchMessage as ReturnType<typeof vi.fn>).mock.calls;
      const terminalDispatch = dispatches.find(
        ([payload, ctx]: any) =>
          ctx?.operationId === subOperationId &&
          payload.type === 'createMessage' &&
          payload.value.role === 'assistant' &&
          payload.value.content === spawnResult,
      );
      expect(terminalDispatch).toBeDefined();
    });

    it('does NOT create a terminal assistant when onComplete fires without a spawn tool_result', async () => {
      // CLI closed before the Agent's tool_result arrived (e.g. crash
      // mid-run). The fallback finalize in onComplete should only flush
      // any streamed content — NOT synthesize a terminal message out
      // of thin air, since no authoritative result exists yet.
      await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Bash'),
        ccToolResult('toolu_child', 'ls output'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      // Thread assistants created in this run: ONLY the seed assistant.
      // No terminal assistant should have been synthesized.
      const assistantCreatesInThread = mockCreateMessage.mock.calls.filter(
        ([payload]: any) => payload.role === 'assistant' && payload.threadId === threadId,
      );
      expect(assistantCreatesInThread.length).toBe(1);
    });

    it('invokes store.refreshThreads on lazy Thread creation (sidebar auto-refresh)', async () => {
      // Without this hook the new subagent Thread is only visible in the
      // sidebar after the user navigates topics / refreshes — an earlier
      // Electron E2E repro had the Thread land in DB but stay invisible
      // in the list until manual `refreshThreads()` call.
      const { store } = await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'x',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub', 'toolu_task', 'toolu_child', 'Bash'),
        ccResult(),
      ]);

      expect(store.refreshThreads).toHaveBeenCalledTimes(1);
    });

    it('does NOT call refreshThreads when no subagent events land', async () => {
      const { store } = await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_1', 'Read', { file_path: '/a.ts' }),
        ccToolResult('toolu_1', 'content'),
        ccResult(),
      ]);

      expect(store.refreshThreads).not.toHaveBeenCalled();
    });

    /**
     * Thread-scoped in-memory streaming. Per-spawn sub-operation is
     * opened via `startOperation({ type: 'subagentThread',
     * parentOperationId, context: { ..., threadId, scope: 'thread' } })`,
     * and every Thread bucket dispatch carries that sub-op's id —
     * `internal_getConversationContext` resolves the Thread context
     * through the standard operation registry path (no threadId-override
     * hack at the dispatch boundary). Without these dispatches the
     * Thread view would stay empty until SWR re-fetches on next open
     * (`fetchAndReplaceMessages` is main-topic scoped).
     */
    it('streams subagent create/update dispatches via a thread-scoped sub-operation', async () => {
      const { store } = await runWithEvents([
        ccInit(),
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'inspect',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_child', 'Bash', { command: 'ls' }),
        ccToolResult('toolu_child', 'ls output'),
        ccSubagentText('msg_sub_2', 'toolu_task', 'final summary'),
        ccSubagentSpawnResult('toolu_task', 'done'),
        ccResult(),
      ]);

      const threadId = mockCreateThread.mock.calls[0][0].id;
      const startOpMock = store.startOperation as ReturnType<typeof vi.fn>;

      // Sub-op opened with `subagentThread` type, parented to the main op,
      // carrying the Thread's ConversationContext (threadId + thread scope)
      // so dispatches resolve into the Thread bucket via the standard
      // `internal_getConversationContext` path.
      const subOpCallIdx = startOpMock.mock.calls.findIndex(
        ([p]: any) => p?.type === 'subagentThread',
      );
      expect(subOpCallIdx).toBeGreaterThanOrEqual(0);
      const subOpCallArg = startOpMock.mock.calls[subOpCallIdx][0];
      expect(subOpCallArg).toMatchObject({
        type: 'subagentThread',
        parentOperationId: 'op-1',
        context: { threadId, scope: 'thread' },
      });
      const subOperationId = startOpMock.mock.results[subOpCallIdx].value.operationId;

      const dispatches = (store.internal_dispatchMessage as ReturnType<typeof vi.fn>).mock.calls;
      const threadDispatches = dispatches.filter(
        ([, ctx]: any) => ctx?.operationId === subOperationId,
      );

      // Seed: user + first in-thread assistant + tool message must each
      // get a createMessage dispatch so the Thread renders the moment
      // the user opens it.
      const threadCreates = threadDispatches.filter(
        ([payload]: any) => payload.type === 'createMessage',
      );
      const threadCreateRoles = threadCreates.map(([p]: any) => (p.value as any).role);
      expect(threadCreateRoles).toContain('user');
      expect(threadCreateRoles).toContain('assistant');
      expect(threadCreateRoles).toContain('tool');

      // The tool createMessage carries the inner tool_use's tool_call_id
      // + apiName so the bubble renders with the right plugin shell.
      const toolCreate = threadCreates.find(([p]: any) => (p.value as any).role === 'tool');
      expect((toolCreate![0].value as any).tool_call_id).toBe('toolu_child');
      expect((toolCreate![0].value as any).plugin?.apiName).toBe('Bash');

      // Streaming: updateMessage dispatches must deliver the assistant's
      // tools[] (so the tool card animates) and the accumulated text
      // (so the closing summary streams).
      const threadUpdates = threadDispatches.filter(
        ([payload]: any) => payload.type === 'updateMessage',
      );
      const anyToolsUpdate = threadUpdates.some(([p]: any) =>
        Array.isArray((p.value as any).tools),
      );
      expect(anyToolsUpdate).toBe(true);
      const anyTextUpdate = threadUpdates.some(
        ([p]: any) =>
          typeof (p.value as any).content === 'string' && (p.value as any).content.length > 0,
      );
      expect(anyTextUpdate).toBe(true);

      // Tool result lands on the thread-scoped tool message id (the
      // DB-generated one captured by the createMessage dispatch above).
      const toolMsgId = toolCreate![0].id;
      const toolResultUpdate = threadUpdates.find(
        ([p]: any) => p.id === toolMsgId && (p.value as any).content === 'ls output',
      );
      expect(toolResultUpdate).toBeDefined();

      // Main bucket must NOT receive updates targeting the in-thread tool
      // message id — keep the main bubble clean of subagent bleed.
      const mainLeaks = dispatches.filter(
        ([payload, ctx]: any) =>
          ctx?.operationId !== subOperationId &&
          payload.type === 'updateMessage' &&
          payload.id === toolMsgId,
      );
      expect(mainLeaks).toHaveLength(0);

      // Sub-op is marked completed once the spawn's tool_result lands +
      // `finalizeSubagentRun` writes the terminal assistant. Cancel /
      // cleanup cascade then flow through the existing parent/child
      // operation linkage instead of any subagent-specific bookkeeping.
      expect(store.completeOperation).toHaveBeenCalledWith(subOperationId);
    });

    /**
     * Regression: parallel main tool_use + subagent inner tool_use rendered
     * Task/Agent as an orphan in the main bubble.
     *
     * The gateway handler is main-agent-only: its `stream_chunk`
     * case dispatches `updateMessage { tools }` to
     * `currentAssistantMessageId` (main). Forwarding a subagent-tagged
     * chunk would overwrite main.tools[] with the subagent's inner tools
     * in the in-memory store. The main's own Task / Agent tool_call_id
     * then has no matching entry in main.tools[], and every tool message
     * under it renders with the "orphan tool call" banner until the next
     * fetchAndReplaceMessages (or forever, if the last corrupting chunk
     * lands after the final fetch).
     *
     * DB persistence is separate (persistSubagent*Chunk writes to the
     * thread scope) and already correct — this guards only the forwarding
     * path.
     */
    it('does NOT forward subagent-tagged stream_chunks to the gateway handler', async () => {
      await runWithEvents([
        ccInit(),
        // Main emits Task + a parallel Read in the same message.id.
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'inspect',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccToolUse('msg_main', 'toolu_read', 'Read', { file_path: '/a.ts' }),
        // Subagent inner tools + closing text.
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_grep', 'Grep'),
        ccSubagentText('msg_sub_2', 'toolu_task', 'subagent summary'),
        ccToolResult('toolu_read', 'content'),
        ccSubagentSpawnResult('toolu_task', 'task done'),
        ccResult(),
      ]);

      const handlerSpy = vi.mocked(createGatewayEventHandler).mock.results[0]?.value as ReturnType<
        typeof vi.fn
      >;
      expect(handlerSpy).toBeDefined();

      // Collect every stream_chunk that reached the handler.
      const forwardedChunks = handlerSpy.mock.calls
        .map((call) => call[0])
        .filter((e: any) => e?.type === 'stream_chunk');

      // None of them may carry subagent context — those are already
      // persisted to the in-thread assistant and must not touch main.
      for (const chunk of forwardedChunks) {
        expect((chunk as any).data?.subagent).toBeUndefined();
      }

      // Sanity: main's parallel tool chunk (Task + Read) still reaches the
      // handler so the in-memory main.tools[] animation still fires.
      const mainToolsCallingChunks = forwardedChunks.filter(
        (e: any) => e.data?.chunkType === 'tools_calling',
      );
      const seenToolIds = new Set<string>();
      for (const c of mainToolsCallingChunks) {
        for (const t of (c as any).data.toolsCalling ?? []) seenToolIds.add(t.id);
      }
      expect(seenToolIds).toContain('toolu_task');
      expect(seenToolIds).toContain('toolu_read');
      expect(seenToolIds).not.toContain('toolu_grep');
    });

    /**
     * Regression for the subagent forwarding guard initially only
     * filtered `stream_chunk` events. `tool_start` / `tool_end` for subagent
     * inner tools still reached the main gateway handler, where:
     *   - `tool_start` would fire `dispatchOnBeforeCall` against the MAIN
     *     context for what is actually a subagent inner tool.
     *   - `tool_end`  would call `fetchAndReplaceMessages(main)` once per
     *     subagent inner tool result — wasted work AND a state-drift window
     *     that surfaced as the "orphan tool call" banner on the spawn's
     *     Task/Agent bubble in the main topic.
     *
     * The guard now covers ALL subagent-tagged events. Main-agent tool
     * lifecycle events (no `subagent` peer) must still reach the handler
     * so the main bubble's animation / onAfterCall hooks fire.
     */
    it('does NOT forward subagent-tagged tool_start / tool_end events to the gateway handler', async () => {
      await runWithEvents([
        ccInit(),
        // Main emits Task + a parallel Read in the same message.id.
        ccToolUse('msg_main', 'toolu_task', 'Task', {
          description: 'inspect',
          prompt: 'go',
          subagent_type: 'Explore',
        }),
        ccToolUse('msg_main', 'toolu_read', 'Read', { file_path: '/a.ts' }),
        // Subagent inner tool — adapter emits tool_start(subagent) here.
        ccSubagentToolUse('msg_sub_1', 'toolu_task', 'toolu_grep', 'Grep', {
          pattern: 'foo',
        }),
        // Subagent inner tool_result with parent_tool_use_id — adapter emits
        // tool_result(subagent) + tool_end(subagent) here. Without the
        // broadened guard, tool_end(subagent) would reach the main handler.
        ccSubagentToolResult('toolu_grep', 'toolu_task', 'grep output'),
        ccSubagentText('msg_sub_2', 'toolu_task', 'subagent summary'),
        // Main's own parallel tool result — emits tool_end (no subagent flag),
        // MUST reach the handler so dispatchOnAfterCall fires on main bucket.
        ccToolResult('toolu_read', 'read content'),
        // Spawn result closes the subagent run.
        ccSubagentSpawnResult('toolu_task', 'task done'),
        ccResult(),
      ]);

      const handlerSpy = vi.mocked(createGatewayEventHandler).mock.results[0]?.value as ReturnType<
        typeof vi.fn
      >;
      expect(handlerSpy).toBeDefined();

      const forwardedEvents = handlerSpy.mock.calls.map((call) => call[0]);

      // No forwarded event of ANY type may carry the subagent peer field —
      // they're all handled inline (tool_result) or routed through the
      // per-spawn thread-scoped dispatcher (stream_chunk, tool_start,
      // tool_end). The main gateway handler is main-agent-only.
      const leakedSubagentEvents = forwardedEvents.filter(
        (e: any) => e?.data?.subagent !== undefined,
      );
      expect(leakedSubagentEvents).toHaveLength(0);

      // Sanity: main-agent tool lifecycle for `toolu_read` still reaches the
      // handler — this is the path that drives the main bubble's tool card
      // animation + invalidates renderer caches via dispatchOnAfterCall.
      const mainToolStarts = forwardedEvents.filter(
        (e: any) => e?.type === 'tool_start' && e.data?.toolCalling?.id === 'toolu_read',
      );
      expect(mainToolStarts.length).toBeGreaterThan(0);

      const mainToolEnds = forwardedEvents.filter(
        (e: any) => e?.type === 'tool_end' && e.data?.toolCallId === 'toolu_read',
      );
      expect(mainToolEnds.length).toBeGreaterThan(0);

      // The subagent's inner Grep tool_start / tool_end specifically must
      // not appear in the forwarded set — guards against a future regression
      // that narrows the filter back to stream_chunk only.
      const grepToolStarts = forwardedEvents.filter(
        (e: any) => e?.type === 'tool_start' && e.data?.toolCalling?.id === 'toolu_grep',
      );
      expect(grepToolStarts).toHaveLength(0);

      const grepToolEnds = forwardedEvents.filter(
        (e: any) => e?.type === 'tool_end' && e.data?.toolCallId === 'toolu_grep',
      );
      expect(grepToolEnds).toHaveLength(0);
    });
  });

  // ────────────────────────────────────────────────────
  // Monitor parentId chain regression
  // ────────────────────────────────────────────────────

  describe('Monitor parentId chain', () => {
    /**
     * Monitor pattern: initial tool_use returns immediately ("Monitor started"),
     * then Monitor's stdout drives new CC assistant turns. These reactive steps
     * arrive as ordinary CC message steps (no `task_started`/`task_notification`
     * signal context — see the `external signal` describe below for the tagged
     * case), so under the spine rule each one parents off the most recent
     * non-tool main message: the conversation chains user → asst → asst with
     * tools inline, and the reactive callbacks render as plain spine assistants.
     */
    it('basic flow: each reactive step chains along the spine', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Step 0: Monitor starts a long-running task
        ccMessageStart('msg_01'),
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', {
          shell: 'until curl localhost/health; do sleep 5; done',
        }),
        ccMessageDelta({ input_tokens: 100, output_tokens: 20 }),
        ccToolResult('toolu_mon_0', 'Monitor started, task abc'),
        // Step 1 (new msg id): CC reacts to Monitor stdout with Bash + new Monitor
        ccMessageStart('msg_02'),
        ccToolUse('msg_02', 'toolu_bash_1', 'Bash', { command: 'echo ok' }),
        ccToolUse('msg_02', 'toolu_mon_1', 'Monitor', { shell: 'tail -f log' }),
        ccMessageDelta({ input_tokens: 150, output_tokens: 30 }),
        ccToolResult('toolu_bash_1', 'ok'),
        ccToolResult('toolu_mon_1', 'Monitor started, task def'),
        // Step 2 (new msg id)
        ccMessageStart('msg_03'),
        ccToolUse('msg_03', 'toolu_bash_2', 'Bash', { command: 'date' }),
        ccMessageDelta({ input_tokens: 200, output_tokens: 40 }),
        ccToolResult('toolu_bash_2', 'Mon Apr 21'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      // Two new assistants (step 1 + step 2); step 0 uses ast-initial
      expect(assistantCreates.length).toBe(2);

      // Step 1 parent = the spine (initial assistant); its tools are inline.
      expect(assistantCreates[0][0].parentId).toBe('ast-initial');
      // Step 2 parent = the spine advanced to step 1's assistant.
      expect(assistantCreates[1][0].parentId).toBe(assistantCreates[0][0].id);
    });

    /**
     * regression: a toolless step in the middle keeps the spine linear.
     * The next step chains off the toolless assistant (the most recent
     * non-tool main message), so no message is orphaned.
     */
    it('toolless middle step: spine stays linear through the toolless step', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Step 0: Monitor
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', { shell: 'watch -n1 date' }),
        ccToolResult('toolu_mon_0', 'Monitor started'),
        // Step 1 (new msg id): TEXT ONLY — no tool_use
        ccText('msg_02', 'Looks like it started. I will wait for output.'),
        // Step 2 (new msg id): Monitor emits a line, CC reacts with Bash
        ccToolUse('msg_03', 'toolu_bash_2', 'Bash', { command: 'echo reacting' }),
        ccToolResult('toolu_bash_2', 'reacting'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      expect(assistantCreates.length).toBe(2);

      // Step 1 (toolless) parent = the spine (initial assistant).
      expect(assistantCreates[0][0].parentId).toBe('ast-initial');

      // Step 2 parent = step 1's assistant: the spine advances on the toolless
      // step too, so the chain stays linear (no skip-back to a tool needed).
      expect(assistantCreates[1][0].parentId).toBe(assistantCreates[0][0].id);
    });

    /**
     * follow-up: N consecutive toolless steps (Monitor pushing
     * stdout line by line, each line triggering a new LLM call that only
     * answers with text). Each toolless assistant chains off the previous
     * one, forming a continuous spine — no message is dropped.
     */
    it('consecutive toolless steps: each chains off the previous spine assistant', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Step 0: Monitor kicks off the long-running task
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', { shell: 'tail -f log' }),
        ccToolResult('toolu_mon_0', 'Monitor started'),
        // Step 1, 2, 3: each Monitor stdout line drives a toolless reply
        ccText('msg_02', '等 list 完。'),
        ccText('msg_03', '84842 列完，开干。'),
        ccText('msg_04', '100/84842 全 skip…'),
        // Step 4: CC finally reacts with a Bash tool
        ccToolUse('msg_05', 'toolu_bash_1', 'Bash', { command: 'echo ack' }),
        ccToolResult('toolu_bash_1', 'ack'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      // 4 new assistants (steps 1–4); step 0 reuses ast-initial
      expect(assistantCreates.length).toBe(4);

      // Each step chains off the previous spine assistant; step 1 off the seed.
      expect(assistantCreates[0][0].parentId).toBe('ast-initial');
      expect(assistantCreates[1][0].parentId).toBe(assistantCreates[0][0].id);
      expect(assistantCreates[2][0].parentId).toBe(assistantCreates[1][0].id);
      expect(assistantCreates[3][0].parentId).toBe(assistantCreates[2][0].id);
    });

    /**
     * Hypothesis: Monitor's tool_result arrives AFTER the next message_start.
     * In CC's stream, tool_result comes from a `user` event AFTER the assistant
     * event that issued the tool_use, and BEFORE the next assistant turn.
     * But what if CC emits the next message_start BEFORE the tool_result lands?
     * (The adapter routes message_start via openMainMessage, which triggers
     * stream_start(newStep); stepParentId is computed in persistQueue —
     * this queue serializes with persistToolBatch but NOT with persistToolResult.)
     */
    it('delayed tool_result: Monitor tool_result arrives after next message_start', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Step 0: Monitor
        ccMessageStart('msg_01'),
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', { shell: '...' }),
        // Step 1 BEGINS before the Monitor tool_result arrives
        ccMessageStart('msg_02'),
        // NOW Monitor's tool_result arrives (interleaved)
        ccToolResult('toolu_mon_0', 'Monitor started'),
        ccToolUse('msg_02', 'toolu_bash_1', 'Bash', {}),
        ccToolResult('toolu_bash_1', 'ok'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      expect(assistantCreates.length).toBe(1);
      // Step 1 parent = the spine (initial assistant). The interleaved Monitor
      // tool_result does not affect the chain: the spine anchor is the most
      // recent non-tool main message, independent of tool timing.
      expect(assistantCreates[0][0].parentId).toBe('ast-initial');
    });
  });

  // ────────────────────────────────────────────────────
  // external signal stamping on Monitor-driven follow-up steps
  // ────────────────────────────────────────────────────

  describe('external signal (metadata.signal)', () => {
    const ccTaskStarted = (taskId: string, toolUseId: string) => ({
      session_id: 'cc-sess-1',
      subtype: 'task_started',
      task_id: taskId,
      tool_use_id: toolUseId,
      type: 'system',
    });
    const ccTaskNotification = (taskId: string) => ({
      session_id: 'cc-sess-1',
      subtype: 'task_notification',
      task_id: taskId,
      type: 'system',
    });

    it('stamps metadata.signal on assistant turns CC opens without user input while a task is active', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        // Step 0: LLM calls Monitor; CC registers it as a long-running
        // task; Monitor's initial "started" tool_result lands as a user
        // event — so the FOLLOW-UP turn is a natural confirmation, NOT
        // a signal callback.
        ccMessageStart('msg_01'),
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', { shell: 'every 1s' }),
        ccTaskStarted('task_a', 'toolu_mon_0'),
        ccToolResult('toolu_mon_0', 'Monitor started'),
        // Step 1: natural confirmation turn — NO signal tag.
        ccMessageStart('msg_02'),
        ccText('msg_02', 'Monitor 已启动。'),
        // Step 2: Monitor pushed stdout → CC re-invokes LLM. No new
        // user event was emitted between msg_02 end and msg_03 start.
        // This IS a signal callback.
        ccMessageStart('msg_03'),
        ccText('msg_03', '第 1 次：12:00:01'),
        // Step 3: another Monitor push → another signal callback.
        ccMessageStart('msg_04'),
        ccText('msg_04', '第 2 次：12:00:02'),
        // Task ends; Step 4 is a natural summary turn, NOT signal.
        ccTaskNotification('task_a'),
        ccMessageStart('msg_05'),
        ccText('msg_05', 'Monitor 任务已完成。'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      // Step 0 reuses ast-initial; steps 1..4 → 4 fresh creates.
      expect(assistantCreates.length).toBe(4);

      // Step 1 confirmation: no signal
      expect(assistantCreates[0][0].metadata?.signal).toBeUndefined();
      // Step 2 first signal callback
      expect(assistantCreates[1][0].metadata?.signal).toEqual({
        sequence: 1,
        sourceToolCallId: 'toolu_mon_0',
        sourceToolName: 'Monitor',
        type: 'tool-stdout',
      });
      // Step 3 second signal callback, sequence advances
      expect(assistantCreates[2][0].metadata?.signal).toEqual({
        sequence: 2,
        sourceToolCallId: 'toolu_mon_0',
        sourceToolName: 'Monitor',
        type: 'tool-stdout',
      });
      // Step 4 post-task summary: tagged with `task-completion` ()
      // so MessageCollector renders it inside the same AssistantGroup,
      // after the SignalCallbacks accordion.
      expect(assistantCreates[3][0].metadata?.signal).toEqual({
        sourceToolCallId: 'toolu_mon_0',
        sourceToolName: 'Monitor',
        type: 'task-completion',
      });
    });

    it('does NOT stamp metadata.signal on turns following a tool_result (main-chain follow-up)', async () => {
      const idCounter = { tool: 0, assistant: 0 };
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') {
          idCounter.tool++;
          return { id: params.id ?? `tool-${idCounter.tool}` };
        }
        idCounter.assistant++;
        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
      });

      await runWithEvents([
        ccInit(),
        ccMessageStart('msg_01'),
        ccToolUse('msg_01', 'toolu_mon_0', 'Monitor', {}),
        ccTaskStarted('task_a', 'toolu_mon_0'),
        ccToolResult('toolu_mon_0', 'Monitor started'),
        // Step 1: Monitor confirmation — main chain, no signal.
        ccMessageStart('msg_02'),
        ccText('msg_02', 'ok'),
        // Step 2: LLM emits Bash. Adapter can't know about tool_use at
        // stream_start time, so the signal tag IS stamped (Monitor is
        // active and no user input arrived). Reader-side
        // (`MessageCollector.getMessageSignal`) ignores the tag when
        // `tools.length > 0`, so the mismatch is benign.
        ccMessageStart('msg_03'),
        ccToolUse('msg_03', 'toolu_bash_0', 'Bash', { command: 'echo' }),
        ccToolResult('toolu_bash_0', 'echo result'),
        // Step 3: post-Bash continuation — adapter saw the tool_result,
        // so the next turn is a natural follow-up, no signal.
        ccMessageStart('msg_04'),
        ccText('msg_04', 'bash done'),
        ccResult(),
      ]);

      const assistantCreates = mockCreateMessage.mock.calls.filter(
        ([p]: any) => p.role === 'assistant',
      );
      expect(assistantCreates.length).toBe(3);
      // Step 1 confirmation — no signal
      expect(assistantCreates[0][0].metadata?.signal).toBeUndefined();
      // Step 2 with Bash tool — signal IS stamped at stream_start, but
      // collector defangs it (tools.length > 0).
      expect(assistantCreates[1][0].metadata?.signal?.sourceToolCallId).toBe('toolu_mon_0');
      // Step 3 post-Bash continuation — no signal.
      expect(assistantCreates[2][0].metadata?.signal).toBeUndefined();
    });
  });

  // ────────────────────────────────────────────────────
  // Parallel main tool batch: tool_end must not race ahead of persistQueue
  // ────────────────────────────────────────────────────

  describe('parallel-tools rollback regression', () => {
    /**
     * User-reported bug: when CC fires a large parallel tool batch (e.g. 7
     * Bash commands at once), the AssistantGroup tool count occasionally
     * "rolls back" — e.g. UI shows "7 次技能调用" then drops to 6.
     *
     * Root cause: the executor's `persistQueue` (DB writes) and the gateway
     * handler's `processingChain` (in-memory dispatch + fetchAndReplaceMessages
     * on tool_end) are two independent serial queues with no happens-before
     * between them. `tool_end` events are forwarded to the handler
     * SYNCHRONOUSLY at the bottom of `handleStreamEvent` (the
     * `pendingStepTransition` gate only fires on stream_start(newStep), not
     * inside a single message.id with parallel tool_use). So when a fast
     * tool_result lands before persistQueue has flushed the LAST
     * persistToolBatch's Phase 1/3 write, the handler runs
     * fetchAndReplaceMessages → reads `assistant.tools` with a partial array
     * → replaceMessages clobbers in-memory state from N → N-k.
     *
     * Observable invariant: by the time the handler is invoked with the FIRST
     * `tool_end` event (which is what triggers fetchAndReplaceMessages), the
     * most recent `mockUpdateMessage` call that wrote a `tools` array must
     * already carry the full cumulative tool list. Otherwise, in real life,
     * the DB read at that moment would return a shorter array and the UI
     * would visibly drop tools.
     *
     * The test slows down `mockUpdateMessage` whenever `val.tools` is present,
     * simulating the lag between Phase 1/3 writes and the rest of the stream.
     * `vi.fn().mock.invocationCallOrder` gives a total ordering across all
     * mocks so we can compare "when was the Nth tools-write CALLED" against
     * "when was the first tool_end forwarded to the handler".
     */
    it('handler must not receive tool_end before persistQueue flushes full tools[]', async () => {
      // Slow tools-bearing updateMessage writes — mirrors a real PG/lambda
      // round trip taking long enough that a fast Bash tool_result can land
      // before all 7 persistToolBatch operations finish their Phase 1/3.
      const TOOLS_WRITE_DELAY_MS = 12;
      mockUpdateMessage.mockImplementation(async (_id: string, val: any) => {
        if (val?.tools) {
          await new Promise((r) => setTimeout(r, TOOLS_WRITE_DELAY_MS));
        }
      });

      // Give each tool message a deterministic id so we can spot-check.
      let toolIdx = 0;
      mockCreateMessage.mockImplementation(async (params: any) => {
        if (params.role === 'tool') return { id: `tool-msg-${++toolIdx}` };
        return { id: `ast-${params.role}-${Date.now()}` };
      });

      const PARALLEL = 7;
      const toolIds = Array.from({ length: PARALLEL }, (_, i) => `toolu_par_${i + 1}`);

      // 7 parallel tool_use blocks in the SAME message.id (CC partial-messages
      // mode emits each tool_use in its own assistant event with the shared
      // msg id; adapter accumulates via toolCallsByMessageId), followed by
      // 7 tool_results arriving back-to-back.
      const events: any[] = [ccInit()];
      for (const id of toolIds) {
        events.push(ccToolUse('msg_par', id, 'Bash', { command: `echo ${id}` }));
      }
      for (const id of toolIds) {
        events.push(ccToolResult(id, `result of ${id}`));
      }
      events.push(ccResult());

      await runWithEvents(events);

      const handlerSpy = vi.mocked(createGatewayEventHandler).mock.results[0]?.value as ReturnType<
        typeof vi.fn
      >;
      expect(handlerSpy).toBeDefined();

      // Find the FIRST tool_end forwarded to the handler — this is the call
      // that would trigger `fetchAndReplaceMessages` in real life and read
      // assistant.tools[] from DB.
      const handlerCalls = handlerSpy.mock.calls.map((args, i) => ({
        event: args[0] as any,
        order: handlerSpy.mock.invocationCallOrder[i],
      }));
      const firstToolEnd = handlerCalls.find(({ event }) => event?.type === 'tool_end');
      expect(firstToolEnd).toBeDefined();

      // All `mockUpdateMessage(_, { tools })` calls — these are persistToolBatch
      // Phase 1 (pre-register) and Phase 3 (backfill) writes. Their invocation
      // order is the moment Phase 1/3 ENTERED `await messageService.updateMessage`.
      // Because persistToolBatch awaits each phase before moving on, the order
      // here is a faithful proxy for the DB-write timeline.
      const toolsWrites = mockUpdateMessage.mock.calls
        .map((args, i) => ({
          tools: (args[1] as any)?.tools,
          order: mockUpdateMessage.mock.invocationCallOrder[i],
        }))
        .filter(({ tools }) => Array.isArray(tools));

      // The latest tools[] write that started BEFORE the handler's first tool_end.
      // If the executor properly defers tool_end through persistQueue, this
      // should be the FINAL Phase 3 write carrying all 7 tools.
      const latestBeforeToolEnd = toolsWrites.findLast(({ order }) => order < firstToolEnd!.order);

      // The bug: without the deferral fix, persistQueue is still mid-flight
      // (or hasn't started) when tool_end is forwarded, so the latest tools[]
      // write seen at that point has fewer than PARALLEL entries — exactly
      // the "7 → 6" rollback the user sees in the UI.
      const writtenCount = latestBeforeToolEnd?.tools?.length ?? 0;
      const writtenIds = (latestBeforeToolEnd?.tools ?? []).map((t: any) => t.id);
      const handlerEventTrail = handlerCalls.map(({ event }) => event?.type).join(',');
      expect(
        writtenCount,
        `tool_end forwarded to handler at order ${firstToolEnd!.order} ` +
          `but the latest persistToolBatch tools[] write at that point had ` +
          `${writtenCount}/${PARALLEL} tools — fetchAndReplaceMessages would ` +
          `read partial assistant.tools[] and roll back the UI. ` +
          `Handler event trail: [${handlerEventTrail}]`,
      ).toBe(PARALLEL);
      // All 7 tool ids must be present in that write — guards against any
      // weird ordering where the last write happens to have 7 entries but
      // the wrong ones (e.g. dedupe bug repopulating from a stale set).
      for (const id of toolIds) expect(writtenIds).toContain(id);
    });
  });

  // ════════════════════════════════════════════════════════════════════════
  // Characterization tests locking the CURRENT hetero-completion lifecycle.
  //
  // These are a REGRESSION NET ahead of the send-message run-lifecycle refactor:
  // they pin what the code does NOW (notifyCompletion fan-out, metadata-save
  // isolation, queue-drain gating, the onError/abort skips), not idealized
  // behavior. If something here looks like a bug it is intentionally locked
  // as-is so the refactor surfaces any behavior change as a failing test.
  // ════════════════════════════════════════════════════════════════════════
  describe('hetero completion characterization (lifecycle refactor regression net)', () => {
    afterEach(() => {
      // Restore the global default so the rest of the suite keeps seeing the
      // non-desktop env that every other test assumes.
      desktopFlag.value = false;
    });

    /**
     * Drive a full run to a non-error terminal (`ccResult()` → agent_runtime_end)
     * with a caller-supplied store, mirroring the error-handling tests' manual
     * harness so per-test store overrides (abortController / drainQueuedMessages /
     * updateTopicStatus) are observable.
     */
    async function runToComplete(
      store: any,
      ccEvents: any[],
      paramOverrides: Partial<typeof defaultParams> = {},
    ) {
      const get = vi.fn(() => store);
      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, {
        ...defaultParams,
        ...paramOverrides,
      });
      await flush();

      for (const event of ccEvents) ipc.emitRawLine('ipc-sess-1', event);
      ipc.emitComplete('ipc-sess-1');
      await flush();

      resolveSendPrompt!();
      // Clean-completion helper: a successful run MUST resolve. Do NOT swallow the
      // rejection here — otherwise a regression that makes the happy path reject
      // (e.g. before the notification side effect runs) would spuriously satisfy the
      // negative assertions (e.g. isDesktop=false → notification NOT called). Let it
      // propagate so such a regression fails the test instead of passing silently.
      await executorPromise;
      await flush();

      return { get, store };
    }

    /**
     * Like `runToComplete` but ends the stream with a TRUE error terminal
     * (`ccResult(true)` → `deferredTerminalEvent.type === 'error'`). This is the
     * event-shape that drives the error branch in onComplete (isErrorTerminal)
     * AND gates the linear-flow queue drain off via `terminalEvent?.type !== 'error'`.
     */
    async function runToError(store: any, paramOverrides: Partial<typeof defaultParams> = {}) {
      const get = vi.fn(() => store);
      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, {
        ...defaultParams,
        ...paramOverrides,
      });
      await flush();

      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', 'partial content'));
      ipc.emitRawLine('ipc-sess-1', ccResult(true, 'the run failed'));
      ipc.emitComplete('ipc-sess-1');
      await flush();

      resolveSendPrompt!();
      // An error TERMINAL is still handled internally (onError / persistTerminalError):
      // the executor RESOLVES, it does not reject. Don't swallow a rejection — if a
      // regression makes the error path reject, the negative assertions (no notification,
      // no drain) must fail rather than be satisfied by an early bail-out.
      await executorPromise;
      await flush();

      return { get, store };
    }

    // ── 1. onComplete success → notifyCompletion (notification + dock badge) ──
    it('fires the desktop notification AND dock badge on a successful non-aborted completion', async () => {
      desktopFlag.value = true;
      const store = createMockStore();

      await runToComplete(store, [
        ccInit(),
        ccText('msg_01', 'All done with the task.'),
        ccResult(),
      ]);

      // notifyCompletion = Promise.allSettled([showNotification(...), setBadgeCount(1)]).
      expect(mockShowNotification).toHaveBeenCalledTimes(1);
      expect(mockShowNotification).toHaveBeenCalledWith(
        expect.objectContaining({
          // body = markdownToTxt(finalContent)
          body: expect.stringContaining('All done with the task'),
          // navigate path resolved from agentId + topicId
          navigate: { path: expect.any(String) },
          title: expect.any(String),
        }),
      );
      expect(mockSetBadgeCount).toHaveBeenCalledTimes(1);
      expect(mockSetBadgeCount).toHaveBeenCalledWith(1);
    });

    it('does NOT touch the desktop notification IPC when isDesktop is false (web/default env)', async () => {
      // desktopFlag.value stays false here — the early `if (!isDesktop) return` guard.
      const store = createMockStore();

      await runToComplete(store, [ccInit(), ccText('msg_01', 'done'), ccResult()]);

      expect(mockShowNotification).not.toHaveBeenCalled();
      expect(mockSetBadgeCount).not.toHaveBeenCalled();
    });

    // ── 2. metadata-save failure isolation (CURRENT behavior, locked as-is) ──
    it('swallows an updateTopicMetadata REJECTION without surfacing an error to the caller', async () => {
      // CHARACTERIZATION — locks ACTUAL current behavior, which DIFFERS from the
      // intended acceptance criterion ("metadata failure must not block the
      // drain"):
      //
      //   const sessionInfo = await getSessionInfo(...).catch(() => undefined);
      //   if (sessionInfo?.agentSessionId && context.topicId)
      //     await updateTopicMetadata?.(...);   // ← NO .catch here
      //   if (!isAborted() && terminalEvent?.type !== 'error') { ...drain... }
      //
      // Because the `updateTopicMetadata` await is UNGUARDED, a rejection throws
      // past the drain block into the function's outer try/catch. `completed` is
      // already `true` (onComplete ran), so `catch { if (!completed) ... }` is a
      // no-op: the executor resolves cleanly (no error escapes to the caller),
      // BUT the queue drain is SKIPPED. The completion notification still fired
      // (it runs inside onComplete, before this metadata await). Pinned here so
      // the lifecycle refactor surfaces any change to this latent edge.
      desktopFlag.value = true;
      const queued = [
        {
          content: 'follow-up please',
          editorData: undefined,
          files: [],
          id: 'q1',
          metadata: {},
        },
      ];
      const store = createMockStore({
        drainQueuedMessages: vi.fn(() => queued),
        updateTopicMetadata: vi.fn().mockRejectedValue(new Error('metadata save boom')),
      });
      const get = vi.fn(() => store);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      let threw = false;
      const executorPromise = executeHeterogeneousAgent(get, defaultParams).catch(() => {
        threw = true;
      });
      await flush();

      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', 'done'));
      ipc.emitRawLine('ipc-sess-1', ccResult());
      ipc.emitComplete('ipc-sess-1');
      await flush();

      resolveSendPrompt!();
      await executorPromise;
      await flush();

      // The metadata save was attempted (and rejected)...
      expect(store.updateTopicMetadata).toHaveBeenCalled();
      // ...and the rejection did NOT escape to the caller (executor resolved).
      expect(threw).toBe(false);
      // The completion notification still fired (runs in onComplete, BEFORE the
      // unguarded metadata await throws).
      expect(mockSetBadgeCount).toHaveBeenCalledWith(1);
      // Current behavior: the unguarded throw bypasses the drain → it is SKIPPED.
      expect(store.drainQueuedMessages).not.toHaveBeenCalled();
    });

    // ── 3. queue-drain gating ──
    it('(success + queued) drains, marks unread completed, and schedules a delayed sendMessage', async () => {
      const queued = [
        {
          content: 'next message',
          editorData: undefined,
          files: [],
          id: 'q1',
          metadata: {},
        },
      ];
      const store = createMockStore({
        drainQueuedMessages: vi.fn(() => queued),
      });
      const get = vi.fn(() => store);

      // The delayed drain dispatch goes through the SINGLETON store
      // (`useChatStore.getState().sendMessage`), NOT the per-call `get()` mock —
      // so spy on the real store's sendMessage to observe it.
      const realSendMessage = vi
        .spyOn(useChatStore.getState(), 'sendMessage')
        .mockResolvedValue(undefined as any);

      let resolveSendPrompt: () => void;
      mockSendPrompt.mockReturnValue(
        new Promise<void>((r) => {
          resolveSendPrompt = r;
        }),
      );

      const executorPromise = executeHeterogeneousAgent(get, defaultParams);
      await flush();

      ipc.emitRawLine('ipc-sess-1', ccInit());
      ipc.emitRawLine('ipc-sess-1', ccText('msg_01', 'done'));
      ipc.emitRawLine('ipc-sess-1', ccResult());
      ipc.emitComplete('ipc-sess-1');
      await flush();

      resolveSendPrompt!();
      await executorPromise;
      // Extra real-time wait so the drain's `setTimeout(() => sendMessage, 100)`
      // fires before we assert on it.
      await new Promise((r) => setTimeout(r, 200));
      await flush();

      expect(store.drainQueuedMessages).toHaveBeenCalled();
      expect(store.completeOperation).toHaveBeenCalledWith('op-1');
      // op-1 context carries agentId/topicId → markTopicUnread fires.
      expect(store.markTopicUnread).toHaveBeenCalledWith(
        expect.objectContaining({ agentId: 'agent-1', topicId: 'topic-1' }),
      );

      // The merged follow-up is dispatched via the setTimeout(100) sendMessage.
      expect(realSendMessage).toHaveBeenCalledTimes(1);
      expect(realSendMessage).toHaveBeenCalledWith(
        expect.objectContaining({ message: 'next message' }),
      );

      realSendMessage.mockRestore();
    });

    it('(terminal type "error") does NOT drain the queue', async () => {
      const store = createMockStore({
        drainQueuedMessages: vi.fn(() => [{ content: 'queued', id: 'q1' }]),
      });

      await runToError(store);

      // Gate is `terminalEvent?.type !== 'error'` — drainQueuedMessages is never
      // even consulted, so no merged follow-up can be dispatched.
      expect(store.drainQueuedMessages).not.toHaveBeenCalled();
    });

    it('(aborted) does NOT drain the queue even on an otherwise-clean stream end', async () => {
      const abortController = new AbortController();
      abortController.abort();
      const store = createMockStore({
        drainQueuedMessages: vi.fn(() => [{ content: 'queued', id: 'q1' }]),
        operations: {
          'op-1': {
            abortController,
            context: { agentId: 'agent-1', scope: 'main', topicId: 'topic-1' },
            metadata: { startTime: 0 },
          },
        },
      });

      await runToComplete(store, [ccInit(), ccText('msg_01', 'done'), ccResult()]);

      // Gate is `!isAborted()` — drainQueuedMessages is never consulted.
      expect(store.drainQueuedMessages).not.toHaveBeenCalled();
    });

    // ── 4. error terminal → no notification, no drain (the onError branch) ──
    it('an error terminal fires NO completion notification and NO queue drain', async () => {
      desktopFlag.value = true; // even on desktop, the error terminal must stay silent
      const store = createMockStore({
        drainQueuedMessages: vi.fn(() => [{ content: 'queued', id: 'q1' }]),
      });

      await runToError(store);

      // No completion signal on the error path (onComplete skips notifyCompletion
      // when isErrorTerminal).
      expect(mockShowNotification).not.toHaveBeenCalled();
      expect(mockSetBadgeCount).not.toHaveBeenCalled();
      // No queue drain on the error path (gated by terminalEvent?.type !== 'error').
      expect(store.drainQueuedMessages).not.toHaveBeenCalled();
      // The error itself is still persisted as a terminal error (persistTerminalError).
      expect(mockUpdateMessageError).toHaveBeenCalled();
    });

    // ── 5. abort path → no notification, no drain, topic status 'active' ──
    it('user abort fires NO notification, NO drain, and writes topic status "active"', async () => {
      desktopFlag.value = true;
      const updateTopicStatus = vi.fn();
      const abortController = new AbortController();
      abortController.abort();
      const store = createMockStore({
        // The user is viewing this topic, so the clean stream end clears the
        // running state back to 'active' (a background completion would instead
        // be left to markTopicUnread → status 'unread').
        activeTopicId: 'topic-1',
        drainQueuedMessages: vi.fn(() => [{ content: 'queued', id: 'q1' }]),
        operations: {
          'op-1': {
            abortController,
            context: { agentId: 'agent-1', scope: 'main', topicId: 'topic-1' },
            metadata: { startTime: 0 },
          },
        },
        updateTopicStatus,
      });

      await runToComplete(store, [ccInit(), ccText('msg_01', 'partial'), ccResult()]);

      expect(mockShowNotification).not.toHaveBeenCalled();
      expect(mockSetBadgeCount).not.toHaveBeenCalled();
      expect(store.drainQueuedMessages).not.toHaveBeenCalled();

      // Characterize the actual behavior: onComplete's non-error branch still
      // writes 'active' while the user is viewing (the abort gate only guards
      // the notification + drain, not the writeTopicStatus('active') call on a
      // clean stream end).
      expect(updateTopicStatus).toHaveBeenCalledWith(
        expect.objectContaining({
          status: 'active',
          topicId: 'topic-1',
        }),
      );
    });
  });
});
