♻️ refactor(chat): unify client hetero executor on a shared mainAgentReducer (#15762)

* ✨ feat(hetero): add shared mainAgentCoordinator reducer

Pure, transactional main-agent run reducer mirroring subagentCoordinator.
Owns the asst→tool→asst chain rule (lastToolMsgIdEver) as the single source
of truth so client and server can converge on one processing flow. Not yet
wired into either interpreter.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* ♻️ refactor(chat): drive client hetero executor via shared mainAgentReducer

Replace the renderer's hand-written main-agent event state machine with the
shared reduceMainAgent + an applyIntent interpreter (main + delegated subagent
intents). The executor keeps its shell (persistQueue/IPC ordering, optimistic
intervention UI, op usage-metrics tray, notifications, resume fallback) and
still forwards raw events to the gateway handler for live UI; durable DB writes
now flow through the reducer's intents, so the asst→tool→asst parent chain
(incl. the lastToolMsgIdEver toolless-step rescue) is a single shared source of
truth with the server.

Tool/assistant message ids are now pre-allocated by the reducer (matching the
subagent path); updated the executor tests to honor caller-provided ids and
assert against captured ids instead of mock-minted ones.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* 📝 docs(chat): clarify why main-scope streamContent intent is a no-op

It's intentional, not dead code: main live token UI is driven by the raw
stream_chunk forward to the gateway handler; the intent only drives the
subagent thread bucket (whose events are dropped before that forward).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* 🐛 fix(chat): close two hetero executor races from reducer refactor

Two review-found bugs introduced by moving main-agent state into the queued
reduceAndApplyMain:

1. retryWithoutResume's hasStreamedState() read mainState, which is now only
   updated inside the queued reduce — so a recoverable resume error landing
   after partial output was queued (but before the queue drained) could start a
   second run and duplicate/interleave messages. Restore the old synchronous
   guarantee with a `sawStreamedEvent` flag set the moment a stream_chunk /
   tool_result arrives, before queueing.

2. A transient createMessage failure on a step-boundary assistant was
   best-effort (logged, not rethrown), so reduceAndApplyMain still committed
   currentAssistantId to a row that was never created — every later
   content/tool/result write then targeted a missing assistant and was lost.
   Rethrow so the commit is skipped and currentAssistantId stays valid, mirroring
   the subagent createMessage path.

Both guarded by regression tests that fail without the fix.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

packages/heterogeneous-agents/src/index.ts

@@ -11,6 +11,20 @@ export {
   REMOTE_HETEROGENEOUS_AGENT_CONFIGS,
 } from './config';
 export { HETEROGENEOUS_TYPE_LABELS } from './labels';
+export type {
+  CreateAssistantIntent,
+  MainAgentIntent,
+  MainAgentReduceCtx,
+  MainAgentRunState,
+  MainAgentTurnToolState,
+  MainPersistToolBatchIntent,
+  MainRecordUsageIntent,
+  MainResolveToolResultIntent,
+  MainStreamContentIntent,
+  PersistAssistantIntent,
+  SetErrorIntent,
+} from './mainAgentCoordinator';
+export { createMainAgentRunState, reduceMainAgent } from './mainAgentCoordinator';
 export { createAdapter, listAgentTypes } from './registry';
 export type {
   CreateMessageIntent,

packages/heterogeneous-agents/src/mainAgentCoordinator/index.ts

@@ -0,0 +1,15 @@
+export { reduce as reduceMainAgent } from './reducer';
+export type {
+  CreateAssistantIntent,
+  MainAgentIntent,
+  MainAgentReduceCtx,
+  MainAgentRunState,
+  MainAgentTurnToolState,
+  MainPersistToolBatchIntent,
+  MainRecordUsageIntent,
+  MainResolveToolResultIntent,
+  MainStreamContentIntent,
+  PersistAssistantIntent,
+  SetErrorIntent,
+} from './types';
+export { createMainAgentRunState } from './types';

packages/heterogeneous-agents/src/mainAgentCoordinator/reducer.test.ts

@@ -0,0 +1,333 @@
+import { describe, expect, it } from 'vitest';
+
+import type { SubagentIntent } from '../subagentCoordinator';
+import { createMainAgentRunState, reduceMainAgent } from './index';
+import type { MainAgentIntent, MainAgentReduceCtx, MainAgentRunState } from './types';
+
+type AnyIntent = MainAgentIntent | SubagentIntent;
+
+/** Deterministic id factory: `thd_1`, `msg_1`, `msg_2`, … per kind. */
+const makeCtx = (over: Partial<MainAgentReduceCtx> = {}): MainAgentReduceCtx => {
+  const counters = { message: 0, thread: 0 };
+  return {
+    agentId: 'agent-1',
+    newId: (kind) => {
+      counters[kind] += 1;
+      return `${kind === 'thread' ? 'thd' : 'msg'}_${counters[kind]}`;
+    },
+    topicId: 'topic-1',
+    ...over,
+  };
+};
+
+// ─── Event builders ───
+
+const initEvent = (model?: string, provider?: string) => ({
+  data: { model, newStep: false, provider },
+  type: 'stream_start',
+});
+
+const newStepEvent = (externalSignal?: any) => ({
+  data: { externalSignal, newStep: true },
+  type: 'stream_start',
+});
+
+const textEvent = (content: string, extra: Record<string, any> = {}) => ({
+  data: { chunkType: 'text', content, ...extra },
+  type: 'stream_chunk',
+});
+
+const reasoningEvent = (reasoning: string) => ({
+  data: { chunkType: 'reasoning', reasoning },
+  type: 'stream_chunk',
+});
+
+const tool = (id: string) => ({
+  apiName: 'Bash',
+  arguments: '{}',
+  id,
+  identifier: 'bash',
+  type: 'default',
+});
+
+const toolsEvent = (tools: any[]) => ({
+  data: { chunkType: 'tools_calling', toolsCalling: tools },
+  type: 'stream_chunk',
+});
+
+const toolResultEvent = (toolCallId: string, content: string, extra: Record<string, any> = {}) => ({
+  data: { content, toolCallId, ...extra },
+  type: 'tool_result',
+});
+
+const turnMetaEvent = (model?: string, provider?: string, usage?: any) => ({
+  data: { model, phase: 'turn_metadata', provider, usage },
+  type: 'step_complete',
+});
+
+const stdoutSignal = (seq: number) => ({
+  sequence: seq,
+  sourceToolCallId: 't1',
+  sourceToolName: 'Monitor',
+  type: 'tool-stdout',
+});
+
+const kinds = (intents: AnyIntent[]) => intents.map((i) => i.kind);
+const ofKind = <K extends AnyIntent['kind']>(intents: AnyIntent[], kind: K) =>
+  intents.filter((i) => i.kind === kind) as Extract<AnyIntent, { kind: K }>[];
+
+/** Apply events with commit-on-success; return final state + per-step intents. */
+const run = (events: { data?: any; type?: string }[], seed = 'A0', ctx = makeCtx()) => {
+  let state = createMainAgentRunState(seed);
+  const steps: AnyIntent[][] = [];
+  for (const e of events) {
+    const r = reduceMainAgent(state, e, ctx);
+    steps.push(r.intents);
+    state = r.state; // commit
+  }
+  return { state, steps };
+};
+
+describe('main agent reducer', () => {
+  it('stream_start init backfills model/provider onto the seed assistant', () => {
+    const { steps } = run([initEvent('claude-sonnet-4-6', 'claude-code')]);
+    expect(steps[0]).toEqual([
+      {
+        kind: 'persistAssistant',
+        messageId: 'A0',
+        model: 'claude-sonnet-4-6',
+        provider: 'claude-code',
+      },
+    ]);
+  });
+
+  it('accumulates text and emits a live streamContent each chunk', () => {
+    const { steps, state } = run([textEvent('Hello '), textEvent('world')]);
+    expect(steps[0]).toEqual([{ content: 'Hello ', kind: 'streamContent', messageId: 'A0' }]);
+    expect(steps[1]).toEqual([{ content: 'Hello world', kind: 'streamContent', messageId: 'A0' }]);
+    expect(state.accContent).toBe('Hello world');
+  });
+
+  it('replace-mode text snapshots replace, and stale sequences are dropped', () => {
+    const { steps, state } = run([
+      textEvent('v2', { snapshotMode: 'replace', snapshotSeq: 2 }),
+      textEvent('v1-late', { snapshotMode: 'replace', snapshotSeq: 1 }), // stale → ignored
+      textEvent('v3', { snapshotMode: 'replace', snapshotSeq: 3 }),
+    ]);
+    expect(steps[0]).toEqual([{ content: 'v2', kind: 'streamContent', messageId: 'A0' }]);
+    expect(steps[1]).toEqual([]); // stale snapshot, no intent
+    expect(steps[2]).toEqual([{ content: 'v3', kind: 'streamContent', messageId: 'A0' }]);
+    expect(state.accContent).toBe('v3');
+    expect(state.lastTextSnapshotSeq).toBe(3);
+  });
+
+  it('accumulates reasoning separately', () => {
+    const { steps, state } = run([reasoningEvent('think '), reasoningEvent('more')]);
+    expect(steps[1]).toEqual([{ kind: 'streamContent', messageId: 'A0', reasoning: 'think more' }]);
+    expect(state.accReasoning).toBe('think more');
+  });
+
+  it('persists a tool batch with pre-allocated ids and isNew flags', () => {
+    const { steps, state } = run([textEvent('let me run'), toolsEvent([tool('t1'), tool('t2')])]);
+    const batch = ofKind(steps[1], 'persistToolBatch')[0];
+    expect(batch.assistantMessageId).toBe('A0');
+    expect(batch.content).toBe('let me run');
+    expect(batch.tools).toEqual([
+      { isNew: true, payload: tool('t1'), toolMessageId: 'msg_1' },
+      { isNew: true, payload: tool('t2'), toolMessageId: 'msg_2' },
+    ]);
+    // per-turn pre-allocated map carries the interpreter's create ids
+    expect(state.toolState.toolMsgIdByCallId.get('t1')).toBe('msg_1');
+    expect(state.toolState.toolMsgIdByCallId.get('t2')).toBe('msg_2');
+    expect(state.lastToolMsgIdEver).toBe('msg_2');
+  });
+
+  it('de-dupes tools on retry (commit-on-success idempotency)', () => {
+    let state = createMainAgentRunState('A0');
+    const ctx = makeCtx();
+    const ev = toolsEvent([tool('t1')]);
+
+    const first = reduceMainAgent(state, ev, ctx);
+    expect(ofKind(first.intents, 'persistToolBatch')[0].tools[0].isNew).toBe(true);
+    state = first.state; // commit
+
+    // Replaying the SAME event must not create the tool again.
+    const second = reduceMainAgent(state, ev, ctx);
+    expect(ofKind(second.intents, 'persistToolBatch')[0].tools[0].isNew).toBe(false);
+  });
+
+  it('opens a new turn chained off the previous turn last tool', () => {
+    const { steps } = run([
+      textEvent('first'),
+      toolsEvent([tool('t1')]), // → msg_1
+      toolResultEvent('t1', 'ok'),
+      newStepEvent(), // → new assistant msg_2, parent = msg_1
+    ]);
+    const flush = ofKind(steps[3], 'persistAssistant')[0];
+    expect(flush).toMatchObject({ content: 'first', messageId: 'A0' });
+    const created = ofKind(steps[3], 'createAssistant')[0];
+    expect(created).toMatchObject({ messageId: 'msg_2', parentId: 'msg_1' });
+  });
+
+  // ─── The 断链 regression: toolless reactive (Monitor) turns must NOT fork ───
+  it('re-mounts toolless signal turns onto the source tool, not the prior assistant', () => {
+    const { steps, state } = run([
+      textEvent('watching build'),
+      toolsEvent([tool('t1')]), // Monitor → msg_1
+      newStepEvent(stdoutSignal(1)), // reactive toolless turn A → msg_2
+      textEvent('build started'),
+      newStepEvent(stdoutSignal(2)), // reactive toolless turn B → msg_3
+      textEvent('still compiling'),
+      newStepEvent(), // natural continuation back to main work → msg_4
+      toolsEvent([tool('t2')]), // → msg_5
+    ]);
+
+    const created = steps
+      .flatMap((s) => ofKind(s, 'createAssistant'))
+      .map((c) => ({ messageId: c.messageId, parentId: c.parentId, signalType: c.signal?.type }));
+
+    // EVERY turn after the Monitor tool hangs off msg_1 (the source tool),
+    // forming a flat fan-out — NOT a linear msg_2 → msg_3 → msg_4 chain that
+    // collectAssistantChain would sever at the first signal-tagged assistant.
+    expect(created).toEqual([
+      { messageId: 'msg_2', parentId: 'msg_1', signalType: 'tool-stdout' },
+      { messageId: 'msg_3', parentId: 'msg_1', signalType: 'tool-stdout' },
+      { messageId: 'msg_4', parentId: 'msg_1', signalType: undefined },
+    ]);
+    // The next real tool advances the chain fallback forward.
+    expect(state.lastToolMsgIdEver).toBe('msg_5');
+  });
+
+  it('falls back to the current assistant only before any tool exists', () => {
+    const { steps } = run([textEvent('hi'), newStepEvent()]); // no tool ever seen
+    expect(ofKind(steps[1], 'createAssistant')[0]).toMatchObject({
+      messageId: 'msg_1',
+      parentId: 'A0',
+    });
+  });
+
+  it('resolves a main tool_result via the global tool map', () => {
+    const { steps } = run([
+      toolsEvent([tool('t1')]),
+      toolResultEvent('t1', 'done', { isError: false, pluginState: { todos: [] } }),
+    ]);
+    expect(steps[1]).toEqual([
+      {
+        content: 'done',
+        isError: false,
+        kind: 'resolveToolResult',
+        pluginState: { todos: [] },
+        toolCallId: 't1',
+      },
+    ]);
+  });
+
+  it('records turn metadata and carries model/provider across turns', () => {
+    const { steps, state } = run([
+      turnMetaEvent('claude-opus-4-8', 'claude-code', { totalTokens: 9 }),
+    ]);
+    expect(steps[0]).toEqual([
+      {
+        kind: 'recordUsage',
+        messageId: 'A0',
+        model: 'claude-opus-4-8',
+        provider: 'claude-code',
+        usage: { totalTokens: 9 },
+      },
+    ]);
+    expect(state.turnModel).toBe('claude-opus-4-8');
+    expect(state.turnProvider).toBe('claude-code');
+  });
+
+  it('flushes the open turn on a terminal event', () => {
+    const { steps, state } = run([
+      textEvent('final answer'),
+      { data: {}, type: 'agent_runtime_end' },
+    ]);
+    expect(ofKind(steps[1], 'persistAssistant')[0]).toMatchObject({
+      content: 'final answer',
+      messageId: 'A0',
+    });
+    expect(state.ended).toBe(true);
+    expect(state.accContent).toBe(''); // reset → idempotent re-finalize
+  });
+
+  it('suppresses echoed content and stamps the error on AuthRequired terminal error', () => {
+    const stderr = 'invalid api key';
+    const { steps } = run([
+      textEvent(stderr), // CC echoed the stderr line into content
+      { data: { code: 'AuthRequired', message: 'auth failed', stderr }, type: 'error' },
+    ]);
+    const persisted = ofKind(steps[1], 'persistAssistant')[0];
+    expect(persisted.content).toBe(''); // cleared (echo)
+    const setError = ofKind(steps[1], 'setError')[0];
+    expect(setError).toMatchObject({ clearContent: true, messageId: 'A0' });
+  });
+
+  it('keeps content on a non-echo error', () => {
+    const { steps } = run([
+      textEvent('partial progress'),
+      { data: { code: 'AuthRequired', stderr: 'totally different' }, type: 'error' },
+    ]);
+    expect(ofKind(steps[1], 'persistAssistant')[0].content).toBe('partial progress');
+    expect(ofKind(steps[1], 'setError')[0].clearContent).toBe(false);
+  });
+
+  it('delegates subagent-tagged events to the subagent coordinator', () => {
+    const subEvent = {
+      data: {
+        chunkType: 'text',
+        content: 'investigating',
+        subagent: {
+          parentToolCallId: 'task-1',
+          spawnMetadata: { description: 'Find bug', prompt: 'go', subagentType: 'explorer' },
+          subagentMessageId: 'm1',
+        },
+      },
+      type: 'stream_chunk',
+    };
+    const { steps, state } = run([subEvent]);
+    // Subagent lazy-create intents come through unchanged.
+    expect(kinds(steps[0])).toEqual([
+      'createThread',
+      'createMessage',
+      'createMessage',
+      'streamContent',
+    ]);
+    // Main thread state is untouched; the run now owns one subagent run.
+    expect(state.currentAssistantId).toBe('A0');
+    expect(state.subagents.runs.size).toBe(1);
+  });
+
+  it('finalizes a subagent run on its parent-spawn tool_result (main-scoped)', () => {
+    // Seed a subagent run, then deliver the Task tool_result on the main thread.
+    const subEvent = {
+      data: {
+        chunkType: 'text',
+        content: 'done investigating',
+        subagent: { parentToolCallId: 'task-1', subagentMessageId: 'm1' },
+      },
+      type: 'stream_chunk',
+    };
+    const { steps } = run([subEvent, toolResultEvent('task-1', 'subagent summary')]);
+    // tool_result yields BOTH the main resolve and the subagent finalize.
+    expect(kinds(steps[1])).toContain('resolveToolResult'); // main tool message content
+    expect(kinds(steps[1])).toContain('finalizeThread'); // subagent run closed
+  });
+
+  it('never mutates the input state (pure reduce)', () => {
+    const state: MainAgentRunState = createMainAgentRunState('A0');
+    const before = JSON.stringify({
+      acc: state.accContent,
+      ever: state.lastToolMsgIdEver,
+      tools: [...state.toolState.toolMsgIdByCallId],
+    });
+    reduceMainAgent(state, toolsEvent([tool('t1')]), makeCtx());
+    const after = JSON.stringify({
+      acc: state.accContent,
+      ever: state.lastToolMsgIdEver,
+      tools: [...state.toolState.toolMsgIdByCallId],
+    });
+    expect(after).toBe(before);
+  });
+});

packages/heterogeneous-agents/src/mainAgentCoordinator/reducer.ts

@@ -0,0 +1,382 @@
+import type { SubagentIntent, SubagentReduceCtx } from '../subagentCoordinator';
+import { getEventScope, reduceSubagentRuns } from '../subagentCoordinator';
+import type { ToolCallPayload } from '../types';
+import type {
+  MainAgentIntent,
+  MainAgentReduceCtx,
+  MainAgentRunState,
+  MainAgentTurnToolState,
+} from './types';
+
+/**
+ * Pure, transactional main-agent run reducer. See `./types.ts` for the design.
+ *
+ * `reduce(state, event, ctx)` returns the NEXT state plus the intents to apply
+ * — it never mutates `state` in place. The caller commits the returned state
+ * ONLY after the intents are successfully applied (commit-on-success), so a
+ * throwing intent leaves the run un-advanced and a retry replays against the
+ * original state — the same resilience `reduceSubagentRuns` relies on.
+ *
+ * Subagent-scoped events (and the parent-spawn `tool_result` / terminal drain)
+ * are delegated to `reduceSubagentRuns`; its intents are merged into the
+ * returned list so a single call drives both the main thread and every nested
+ * subagent.
+ */
+
+type AnyIntent = MainAgentIntent | SubagentIntent;
+
+interface ReduceResult {
+  intents: AnyIntent[];
+  state: MainAgentRunState;
+}
+
+// ─── State copy helpers (structural sharing) ───
+
+const copyToolState = (s: MainAgentTurnToolState): MainAgentTurnToolState => ({
+  payloads: s.payloads.map((p) => ({ ...p })),
+  persistedIds: new Set(s.persistedIds),
+  toolMsgIdByCallId: new Map(s.toolMsgIdByCallId),
+});
+
+const emptyToolState = (): MainAgentTurnToolState => ({
+  payloads: [],
+  persistedIds: new Set(),
+  toolMsgIdByCallId: new Map(),
+});
+
+/** Deep-copy the parts of state a handler may mutate; subagents is swapped wholesale. */
+const copyState = (s: MainAgentRunState): MainAgentRunState => ({
+  ...s,
+  toolState: copyToolState(s.toolState),
+  turnMetadata: { ...s.turnMetadata },
+});
+
+// ─── Echo suppression (pure; mirrors both engines' shouldSuppressTerminalErrorEcho) ───
+
+const normalizeErrorText = (value?: string) => value?.replaceAll(/\s+/g, ' ').trim();
+
+/**
+ * CC sometimes streams the error string into `content` BEFORE emitting the
+ * structured error event (e.g. AuthRequired echoes the stderr line). Only
+ * suppress when the body is explicitly marked OR matches the AuthRequired code,
+ * AND the trimmed strings are equal. Anything else stays — accidental partial
+ * overlaps are not echo. Operates on the wire `error` event data
+ * (`HeterogeneousTerminalErrorData`), so the decision is pure.
+ */
+const shouldSuppressTerminalErrorEcho = (content: string, errorData: unknown): boolean => {
+  const body = errorData as
+    | { clearEchoedContent?: boolean; code?: string; message?: string; stderr?: string }
+    | undefined;
+  // Keep in sync with the interpreters' ECHO_TRIGGER_CODES.
+  if (!body?.clearEchoedContent && body?.code !== 'AuthRequired') return false;
+  const normalizedContent = normalizeErrorText(content);
+  const normalizedError = normalizeErrorText(body?.stderr || body?.message);
+  return !!normalizedContent && !!normalizedError && normalizedContent === normalizedError;
+};
+
+// ─── Subagent delegation ───
+
+const subagentCtx = (state: MainAgentRunState, ctx: MainAgentReduceCtx): SubagentReduceCtx => ({
+  agentId: ctx.agentId,
+  mainAssistantId: state.currentAssistantId,
+  newId: ctx.newId,
+  topicId: ctx.topicId,
+});
+
+/** Reduce an event through the nested subagent coordinator, folding its state back in. */
+const delegateSubagent = (
+  state: MainAgentRunState,
+  event: { data?: any; type?: string },
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  const { intents, state: subState } = reduceSubagentRuns(
+    state.subagents,
+    event,
+    subagentCtx(state, ctx),
+  );
+  return { intents, state: { ...state, subagents: subState } };
+};
+
+// ─── Chain rule ───
+
+/**
+ * Parent for the NEXT turn's assistant. Prefer the run-lifetime last tool
+ * message (`lastToolMsgIdEver`) so toolless reactive turns don't fork the wire;
+ * fall back to the current assistant only before any tool has been seen.
+ */
+const computeTurnParentId = (state: MainAgentRunState): string =>
+  state.lastToolMsgIdEver ?? state.currentAssistantId;
+
+// ─── Per-event handlers ───
+
+/** `stream_start { newStep: true }` — flush the prior turn, open a new assistant. */
+const openTurn = (state: MainAgentRunState, data: any, ctx: MainAgentReduceCtx): ReduceResult => {
+  const intents: AnyIntent[] = [];
+
+  // 1. Durably flush the prior turn's accumulators + model/provider.
+  const flush: Record<string, any> = {};
+  if (state.accContent) flush.content = state.accContent;
+  if (state.accReasoning) flush.reasoning = state.accReasoning;
+  if (state.turnModel) flush.model = state.turnModel;
+  if (state.turnProvider) flush.provider = state.turnProvider;
+  if (Object.keys(flush).length > 0) {
+    intents.push({ kind: 'persistAssistant', messageId: state.currentAssistantId, ...flush });
+  }
+
+  // 2. Open the new turn's assistant, chained off the last tool (chain rule).
+  const messageId = ctx.newId('message');
+  intents.push({
+    agentId: ctx.agentId,
+    kind: 'createAssistant',
+    messageId,
+    model: state.turnModel,
+    parentId: computeTurnParentId(state),
+    provider: state.turnProvider,
+    signal: data?.externalSignal,
+    topicId: ctx.topicId,
+  });
+
+  // 3. Advance: model/provider carry across (a fresh turn_metadata overwrites).
+  const next = copyState(state);
+  next.currentAssistantId = messageId;
+  next.accContent = '';
+  next.accReasoning = '';
+  next.lastTextSnapshotSeq = 0;
+  next.turnMetadata = {};
+  next.toolState = emptyToolState();
+  return { intents, state: next };
+};
+
+/** First `stream_start` (no newStep) carries the CLI's authoritative model/provider. */
+const streamInit = (state: MainAgentRunState, data: any): ReduceResult => {
+  const update: Record<string, any> = {};
+  if (data?.model) update.model = data.model;
+  if (data?.provider) update.provider = data.provider;
+  if (Object.keys(update).length === 0) return { intents: [], state };
+
+  const next = copyState(state);
+  if (data.model) next.turnModel = data.model;
+  if (data.provider) next.turnProvider = data.provider;
+  return {
+    intents: [{ kind: 'persistAssistant', messageId: state.currentAssistantId, ...update }],
+    state: next,
+  };
+};
+
+const reduceTextChunk = (state: MainAgentRunState, data: any): ReduceResult => {
+  const next = copyState(state);
+  const snapshotMode = data?.snapshotMode;
+  const snapshotSeq = typeof data?.snapshotSeq === 'number' ? data.snapshotSeq : undefined;
+
+  if (snapshotMode === 'replace' && snapshotSeq !== undefined) {
+    if (snapshotSeq <= state.lastTextSnapshotSeq) return { intents: [], state }; // stale snapshot
+    next.lastTextSnapshotSeq = snapshotSeq;
+    next.turnMetadata = { ...next.turnMetadata, heteroTextSnapshotSeq: snapshotSeq };
+    next.accContent = data.content;
+  } else {
+    if (!data?.content) return { intents: [], state };
+    next.accContent = state.accContent + data.content;
+  }
+
+  return {
+    intents: [
+      { content: next.accContent, kind: 'streamContent', messageId: next.currentAssistantId },
+    ],
+    state: next,
+  };
+};
+
+const reduceReasoningChunk = (state: MainAgentRunState, data: any): ReduceResult => {
+  if (!data?.reasoning) return { intents: [], state };
+  const next = copyState(state);
+  next.accReasoning = state.accReasoning + data.reasoning;
+  return {
+    intents: [
+      { kind: 'streamContent', messageId: next.currentAssistantId, reasoning: next.accReasoning },
+    ],
+    state: next,
+  };
+};
+
+const reduceToolsChunk = (
+  state: MainAgentRunState,
+  tools: ToolCallPayload[],
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  const next = copyState(state);
+  const newToolMsgIds: string[] = [];
+
+  for (const tool of tools) {
+    if (next.toolState.persistedIds.has(tool.id)) continue;
+    next.toolState.persistedIds.add(tool.id);
+    next.toolState.payloads.push({
+      apiName: tool.apiName,
+      arguments: tool.arguments,
+      id: tool.id,
+      identifier: tool.identifier,
+      type: tool.type,
+    });
+    const toolMessageId = ctx.newId('message');
+    next.toolState.toolMsgIdByCallId.set(tool.id, toolMessageId);
+    newToolMsgIds.push(toolMessageId);
+  }
+
+  const intents: AnyIntent[] = [
+    {
+      assistantMessageId: next.currentAssistantId,
+      content: next.accContent || undefined,
+      kind: 'persistToolBatch',
+      reasoning: next.accReasoning || undefined,
+      tools: next.toolState.payloads.map((p) => ({
+        isNew: newToolMsgIds.includes(next.toolState.toolMsgIdByCallId.get(p.id)!),
+        payload: { ...p },
+        toolMessageId: next.toolState.toolMsgIdByCallId.get(p.id)!,
+      })),
+    },
+  ];
+
+  // Advance the chain fallback to this turn's last tool message.
+  const lastToolMsgId = newToolMsgIds.at(-1);
+  if (lastToolMsgId) next.lastToolMsgIdEver = lastToolMsgId;
+
+  return { intents, state: next };
+};
+
+const reduceStreamChunk = (
+  state: MainAgentRunState,
+  data: any,
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  if (data?.chunkType === 'text' && typeof data.content === 'string') {
+    return reduceTextChunk(state, data);
+  }
+  if (data?.chunkType === 'reasoning' && typeof data.reasoning === 'string') {
+    return reduceReasoningChunk(state, data);
+  }
+  if (data?.chunkType === 'tools_calling') {
+    const tools = (data.toolsCalling as ToolCallPayload[] | undefined) ?? [];
+    if (tools.length === 0) return { intents: [], state };
+    return reduceToolsChunk(state, tools, ctx);
+  }
+  return { intents: [], state };
+};
+
+/** Main-agent (and parent-spawn) tool_result. Inner subagent results are delegated. */
+const reduceToolResult = (
+  state: MainAgentRunState,
+  event: { data?: any },
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  const data = event.data ?? {};
+  const toolCallId: string | undefined = data.toolCallId;
+  if (!toolCallId) return { intents: [], state };
+
+  // Resolve the (main-scoped) tool message content, then delegate so a
+  // parent-spawn tool_result finalizes its subagent run (no-op otherwise).
+  const main: AnyIntent = {
+    content: data.content ?? '',
+    isError: !!data.isError,
+    kind: 'resolveToolResult',
+    pluginState: data.pluginState,
+    toolCallId,
+  };
+  const delegated = delegateSubagent(state, event, ctx);
+  return { intents: [main, ...delegated.intents], state: delegated.state };
+};
+
+const reduceTurnMetadata = (state: MainAgentRunState, data: any): ReduceResult => {
+  const next = copyState(state);
+  if (data?.model) next.turnModel = data.model;
+  if (data?.provider) next.turnProvider = data.provider;
+  const usage = data?.usage;
+  if (usage) next.turnMetadata = { ...next.turnMetadata, usage };
+
+  if (!data?.model && !data?.provider && !usage) return { intents: [], state: next };
+  return {
+    intents: [
+      {
+        kind: 'recordUsage',
+        messageId: state.currentAssistantId,
+        model: data?.model,
+        provider: data?.provider,
+        usage,
+      },
+    ],
+    state: next,
+  };
+};
+
+const reduceTerminal = (
+  state: MainAgentRunState,
+  event: { data?: any; type?: string },
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  const isError = event.type === 'error';
+  const suppress = isError ? shouldSuppressTerminalErrorEcho(state.accContent, event.data) : false;
+
+  const intents: AnyIntent[] = [];
+  const flush: Record<string, any> = {};
+  if (suppress) flush.content = '';
+  else if (state.accContent) flush.content = state.accContent;
+  if (state.accReasoning) flush.reasoning = state.accReasoning;
+  if (state.turnModel) flush.model = state.turnModel;
+  if (state.turnProvider) flush.provider = state.turnProvider;
+  if (Object.keys(flush).length > 0) {
+    intents.push({ kind: 'persistAssistant', messageId: state.currentAssistantId, ...flush });
+  }
+  if (isError) {
+    intents.push({
+      clearContent: suppress,
+      errorData: event.data,
+      kind: 'setError',
+      messageId: state.currentAssistantId,
+    });
+  }
+
+  // Reset accumulators so a follow-up terminal/flush is an idempotent no-op,
+  // then drain any subagent runs that never saw their parent tool_result.
+  const drained = copyState(state);
+  drained.accContent = '';
+  drained.accReasoning = '';
+  drained.ended = true;
+  const delegated = delegateSubagent(drained, event, ctx);
+  return { intents: [...intents, ...delegated.intents], state: delegated.state };
+};
+
+/**
+ * Reduce a single stream event. Returns the next state and intents to apply.
+ * Subagent-scoped events are routed entirely through the subagent coordinator;
+ * `tool_result` / terminal events run the main path AND delegate.
+ */
+export const reduce = (
+  state: MainAgentRunState,
+  event: { data?: any; type?: string },
+  ctx: MainAgentReduceCtx,
+): ReduceResult => {
+  // Subagent-tagged chunks / turn_metadata belong wholly to the coordinator.
+  if (getEventScope(event).kind === 'subagent') return delegateSubagent(state, event, ctx);
+
+  const data = event.data ?? {};
+  switch (event.type) {
+    case 'stream_start': {
+      return data?.newStep ? openTurn(state, data, ctx) : streamInit(state, data);
+    }
+    case 'stream_chunk': {
+      return reduceStreamChunk(state, data, ctx);
+    }
+    case 'tool_result': {
+      return reduceToolResult(state, event, ctx);
+    }
+    case 'step_complete': {
+      if (data?.phase === 'turn_metadata') return reduceTurnMetadata(state, data);
+      return { intents: [], state };
+    }
+    case 'agent_runtime_end':
+    case 'error': {
+      return reduceTerminal(state, event, ctx);
+    }
+    default: {
+      return { intents: [], state };
+    }
+  }
+};

packages/heterogeneous-agents/src/mainAgentCoordinator/types.ts

@@ -0,0 +1,235 @@
+import type { PersistToolBatchEntry, SubagentRunsState } from '../subagentCoordinator';
+import { createSubagentRunsState } from '../subagentCoordinator';
+import type { ExternalSignalContext, ToolCallPayload } from '../types';
+
+/**
+ * Main-agent run coordinator — shared, side-effect-free state machine for the
+ * MAIN (non-subagent) thread of a heterogeneous-agent (Claude Code / Codex) run.
+ *
+ * Background: the renderer executor (`heterogeneousAgentExecutor.ts`) and the
+ * server persistence handler (`HeterogeneousPersistenceHandler.ts`) each
+ * hand-wrote the SAME main-agent state machine — content accumulation, step
+ * (turn) boundary, the `asst → tool → asst → tool` parent chain, 3-phase tool
+ * persist, tool_result resolution, terminal flush. That duplication is exactly
+ * how the two diverged: the renderer carries a run-lifetime `lastToolMsgIdEver`
+ * fallback that re-mounts toolless reactive turns (Monitor stdout pushes) onto
+ * the source tool so `MessageCollector.collectAssistantChain` keeps walking;
+ * the server lacked it and, on a cold serverless replica, fell back to chaining
+ * `asst → asst`, which forks the wire into disconnected bubbles (the remote
+ * "断链" bug).
+ *
+ * This module owns the "when to open a turn / persist / resolve / finalize"
+ * decisions in ONE pure reducer, mirroring `subagentCoordinator`. It performs
+ * no I/O: it returns a list of intents that each environment's interpreter
+ * executes against its own persistence + UI surfaces. The reducer pre-allocates
+ * every id (`ctx.newId`) so intents carry concrete `parentId` chains with no
+ * "create then backfill id" round-trip. It also OWNS the nested subagent runs
+ * by delegating subagent-scoped events to `reduceSubagentRuns`, so a single
+ * `reduce` call is the only entry point both engines need.
+ *
+ * The CHAIN RULE lives here and is authoritative for both engines:
+ *   the next turn's assistant parents off the last tool message of the most
+ *   recent tool-bearing turn (`lastToolMsgIdEver`), falling back to the current
+ *   assistant only when no tool has ever been seen. This keeps the rendered
+ *   chain a linear `asst → tool → asst → tool …` zigzag even across toolless
+ *   reactive (signal-tagged) turns.
+ */
+
+// ─── Reducer state ───
+
+/** Per-turn tool persistence state. Reset on every turn (step) boundary. */
+export interface MainAgentTurnToolState {
+  /**
+   * Cumulative `tools[]` payloads for the CURRENT turn's assistant. Carries NO
+   * `result_msg_id` — the interpreter backfills that from the pre-allocated
+   * tool-message id when it writes the assistant's `tools[]` (phase 3).
+   */
+  payloads: ToolCallPayload[];
+  /** tool_call ids already turned into tool messages this turn (de-dupe). */
+  persistedIds: Set<string>;
+  /** Pre-allocated tool-message id per tool_call id, for this turn's payloads. */
+  toolMsgIdByCallId: Map<string, string>;
+}
+
+/**
+ * Per-run main-agent state. Lifetime spans the whole CLI run. Designed to be
+ * fully RE-HYDRATABLE from the DB so a stateless server replica can project it
+ * and run the same pure reduce as the long-lived renderer process.
+ */
+export interface MainAgentRunState {
+  /** Accumulated text for the current turn's assistant. */
+  accContent: string;
+  /** Accumulated reasoning (thinking) for the current turn. */
+  accReasoning: string;
+  /** The main-agent assistant message currently being appended to. */
+  currentAssistantId: string;
+  /** Set once a terminal event has been reduced (idempotent finalize). */
+  ended: boolean;
+  /** Highest seen text snapshot sequence (replace-mode de-dup). */
+  lastTextSnapshotSeq: number;
+  /**
+   * Run-lifetime id of the most recent main-agent tool message. This is the
+   * chain-rule fallback: a new turn whose PRIOR turn produced no tools (e.g. a
+   * Monitor-stdout reactive reply) re-mounts onto this tool rather than onto
+   * the toolless assistant, so the rendered chain stays a linear zigzag.
+   * Only advances on tool batches; never reset across turns.
+   */
+  lastToolMsgIdEver: string | undefined;
+  /** Nested subagent runs — delegated to `reduceSubagentRuns`. */
+  subagents: SubagentRunsState;
+  /** Per-turn tool persistence state. */
+  toolState: MainAgentTurnToolState;
+  /** Accumulated metadata (usage, snapshot seq) for the current assistant. */
+  turnMetadata: Record<string, any>;
+  /** Latest model id for the run (carried across turns until overwritten). */
+  turnModel: string | undefined;
+  /** Latest provider for the run (carried across turns until overwritten). */
+  turnProvider: string | undefined;
+}
+
+/**
+ * Seed a fresh run state. `seedAssistantId` is the placeholder assistant the
+ * host already created (renderer: `conversationLifecycle`; server:
+ * `aiAgent.execAgent`) before the first stream event — the reducer never
+ * creates the first assistant, only subsequent turns.
+ */
+export const createMainAgentRunState = (seedAssistantId: string): MainAgentRunState => ({
+  accContent: '',
+  accReasoning: '',
+  currentAssistantId: seedAssistantId,
+  ended: false,
+  lastTextSnapshotSeq: 0,
+  lastToolMsgIdEver: undefined,
+  subagents: createSubagentRunsState(),
+  toolState: { payloads: [], persistedIds: new Set(), toolMsgIdByCallId: new Map() },
+  turnMetadata: {},
+  turnModel: undefined,
+  turnProvider: undefined,
+});
+
+// ─── Reduce context (per event) ───
+
+/**
+ * Per-event context the interpreter supplies. `mainAssistantId` is NOT here —
+ * it lives in the reducer state (`currentAssistantId`) and is forwarded to the
+ * subagent coordinator on delegation. `newId` pre-allocates a DB-compatible id;
+ * `'thread'` is forwarded to the subagent coordinator for thread creation.
+ */
+export interface MainAgentReduceCtx {
+  agentId?: string | null;
+  /** Allocate a prefixed id (`thd_…` / `msg_…`). Deterministic counter in tests. */
+  newId: (kind: 'message' | 'thread') => string;
+  topicId: string | null;
+}
+
+// ─── Intents ───
+
+/**
+ * Declarative "what happened" instructions for the MAIN agent. Each interpreter
+ * maps these to its own I/O. The vocabulary deliberately overlaps
+ * `SubagentIntent` (a `reduce` call returns a mix of both — main-scoped here
+ * plus any subagent intents from delegation) so one interpreter can serve both.
+ *
+ * The live (`streamContent`) vs durable (`persistAssistant` / `persistToolBatch`)
+ * split is intentional: the renderer applies `streamContent` to its live store
+ * for token-level UI and writes the DB only on the durable intents; the server
+ * no-ops `streamContent` (one DB write per token would be wasteful).
+ */
+export type MainAgentIntent =
+  | CreateAssistantIntent
+  | PersistAssistantIntent
+  | MainStreamContentIntent
+  | MainPersistToolBatchIntent
+  | MainResolveToolResultIntent
+  | MainRecordUsageIntent
+  | SetErrorIntent;
+
+/** Open a new turn's assistant message, chained off the computed `parentId`. */
+export interface CreateAssistantIntent {
+  agentId?: string | null;
+  kind: 'createAssistant';
+  messageId: string;
+  /** Last known model carried from the prior turn (real model lands via usage). */
+  model?: string;
+  parentId: string;
+  provider?: string;
+  /** External-signal context to stamp on `metadata.signal` (Monitor pushes etc.). */
+  signal?: ExternalSignalContext;
+  topicId: string | null;
+}
+
+/**
+ * Durable flush of an assistant's content/reasoning/model/provider/metadata.
+ * Used for the prior-turn flush at a boundary, the `stream_start` init
+ * model/provider backfill, and the terminal final flush.
+ */
+export interface PersistAssistantIntent {
+  content?: string;
+  kind: 'persistAssistant';
+  messageId: string;
+  metadata?: Record<string, any>;
+  model?: string;
+  provider?: string;
+  reasoning?: string;
+}
+
+/** Live in-memory content update (replace). Renderer applies; server no-ops. */
+export interface MainStreamContentIntent {
+  content?: string;
+  kind: 'streamContent';
+  messageId: string;
+  reasoning?: string;
+}
+
+/**
+ * Persist a batch of main-agent tool calls into the current assistant. Same
+ * 3-phase write as the subagent variant: (1) `assistant.tools[]` without
+ * result_msg_id, (2) create rows for `isNew` entries with their pre-allocated
+ * ids + populate the tool-message lookup, (3) re-write `assistant.tools[]` with
+ * `result_msg_id` backfilled from each entry's `toolMessageId`.
+ */
+export interface MainPersistToolBatchIntent {
+  assistantMessageId: string;
+  content?: string;
+  kind: 'persistToolBatch';
+  reasoning?: string;
+  tools: PersistToolBatchEntry[];
+}
+
+/**
+ * Resolve a main-agent tool_result. The interpreter looks up the tool-message
+ * id from its `toolCallId → messageId` map (the run-global one, DB-backed on
+ * the server so a cross-replica result still lands).
+ */
+export interface MainResolveToolResultIntent {
+  content: string;
+  isError: boolean;
+  kind: 'resolveToolResult';
+  pluginState?: Record<string, any>;
+  toolCallId: string;
+}
+
+/** Attach per-turn usage/model/provider to the current assistant. */
+export interface MainRecordUsageIntent {
+  kind: 'recordUsage';
+  messageId: string;
+  model?: string;
+  provider?: string;
+  usage: unknown;
+}
+
+/**
+ * Stamp a terminal error on the current assistant. The reducer decides
+ * `clearContent` (echo suppression) purely; the interpreter keeps ownership of
+ * provider-specific error CLASSIFICATION — it receives the raw wire `errorData`
+ * and runs its own classifier (`toHeterogeneousAgentMessageError` /
+ * `toChatMessageError`).
+ */
+export interface SetErrorIntent {
+  /** True when the streamed content echoed the error and should be cleared. */
+  clearContent: boolean;
+  /** Raw terminal error event data; interpreter classifies into ChatMessageError. */
+  errorData: unknown;
+  kind: 'setError';
+  messageId: string;
+}

src/store/chat/slices/aiChat/actions/__tests__/heterogeneousAgentExecutor.test.ts

@@ -435,8 +435,14 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       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: `created-${params.role}-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`,
+      id:
+        params.id ??
+        `created-${params.role}-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`,
     }));
     mockUpdateMessage.mockResolvedValue(undefined);
     mockUpdateMessageError.mockResolvedValue({ success: false });
@@ -498,16 +504,6 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
 
   describe('tool 3-phase persistence', () => {
     it('should pre-register tools, create tool messages, then backfill result_msg_id', async () => {
-      // Track createMessage call order and IDs
-      let toolMsgCounter = 0;
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        if (params.role === 'tool') {
-          toolMsgCounter++;
-          return { id: `tool-msg-${toolMsgCounter}` };
-        }
-        return { id: `msg-${params.role}-${Date.now()}` };
-      });
-
       await runWithEvents([
         ccInit(),
         ccToolUse('msg_01', 'toolu_1', 'Read', { file_path: '/a.ts' }),
@@ -536,9 +532,11 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         plugin: expect.objectContaining({ apiName: 'Read' }),
       });
 
-      // Phase 3: the last tools[] write should have result_msg_id backfilled
+      // 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('tool-msg-1');
+      expect(lastToolUpdate[1].tools[0].result_msg_id).toBe(createdToolId);
     });
 
     it('should deduplicate tool calls (idempotent)', async () => {
@@ -565,11 +563,6 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
 
   describe('tool result persistence', () => {
     it('should update tool message content on tool_result', async () => {
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        if (params.role === 'tool') return { id: 'tool-msg-read' };
-        return { id: `msg-${Date.now()}` };
-      });
-
       await runWithEvents([
         ccInit(),
         ccToolUse('msg_01', 'toolu_read', 'Read', { file_path: '/x.ts' }),
@@ -577,19 +570,17 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ccResult(),
       ]);
 
+      const toolId = mockCreateMessage.mock.calls.find(
+        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_read',
+      )![0].id;
       expect(mockUpdateToolMessage).toHaveBeenCalledWith(
-        'tool-msg-read',
+        toolId,
         { content: 'the file content here', pluginError: undefined },
         { agentId: 'agent-1', topicId: 'topic-1' },
       );
     });
 
     it('should mark error tool results with pluginError', async () => {
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        if (params.role === 'tool') return { id: 'tool-msg-err' };
-        return { id: `msg-${Date.now()}` };
-      });
-
       await runWithEvents([
         ccInit(),
         ccToolUse('msg_01', 'toolu_fail', 'Read', { file_path: '/nope' }),
@@ -597,8 +588,11 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ccResult(),
       ]);
 
+      const toolId = mockCreateMessage.mock.calls.find(
+        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_fail',
+      )![0].id;
       expect(mockUpdateToolMessage).toHaveBeenCalledWith(
-        'tool-msg-err',
+        toolId,
         { content: 'ENOENT: no such file', pluginError: { message: 'ENOENT: no such file' } },
         { agentId: 'agent-1', topicId: 'topic-1' },
       );
@@ -611,14 +605,6 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
 
   describe('multi-step parentId chain', () => {
     it('should create assistant messages chained: assistant → tool → assistant', async () => {
-      const createdIds: string[] = [];
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        const id =
-          params.role === 'tool' ? `tool-${createdIds.length}` : `ast-step-${createdIds.length}`;
-        createdIds.push(id);
-        return { id };
-      });
-
       await runWithEvents([
         ccInit(),
         // Step 1: tool_use Read (message_start primes turn + model/provider
@@ -651,8 +637,8 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ([p]: any) => p.role === 'assistant' && p.parentId !== undefined,
       );
       expect(step2Assistant).toBeDefined();
-      // The parentId should be the tool message ID from step 1
+      // The parentId should be the (pre-allocated) tool message ID from step 1
-      const tool1Id = createdIds.find((id) => id.startsWith('tool-'));
+      const tool1Id = tool1Create![0].id;
       expect(step2Assistant![0].parentId).toBe(tool1Id);
       // createMessage should carry the adapter provider so step 2's assistant
       // lands in DB with provider set from the start (no later backfill needed).
@@ -732,16 +718,6 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
     });
 
     it('should persist per-step usage to each step assistant message, not accumulated', async () => {
-      // Deterministic ids for new-step assistant messages so we can assert per-message usage.
-      let astStepCounter = 0;
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        if (params.role === 'assistant') {
-          astStepCounter++;
-          return { id: `ast-step-${astStepCounter}` };
-        }
-        return { id: `tool-${Date.now()}` };
-      });
-
       // Realistic CC partial-messages flow: message_start primes the turn,
       // assistant events echo a stale usage, message_delta carries the final.
       const { store } = await runWithEvents([
@@ -765,8 +741,11 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       const usageWrites = mockUpdateMessage.mock.calls.filter(
         ([, val]: any) => val.metadata?.usage?.totalTokens,
       );
-      // One usage write per step (msg_01 → ast-initial, msg_02 → ast-step-1)
+      // 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();
@@ -779,7 +758,7 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       expect(u1.inputCachedTokens).toBe(200);
       expect(u1.inputWriteCacheTokens).toBe(50);
 
-      const step2 = usageWrites.find(([id]: any) => id === 'ast-step-1');
+      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
@@ -840,13 +819,6 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       // [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.
 
-      const createdIds: string[] = [];
-      mockCreateMessage.mockImplementation(async (params: any) => {
-        const id = `${params.role}-${createdIds.length}`;
-        createdIds.push(id);
-        return { id };
-      });
-
       await runWithEvents([
         ccInit(),
         // Step 1: text
@@ -864,13 +836,13 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       expect(oldStepWrite).toBeDefined();
 
       // The new step's final write should have "Step 2 content"
-      const newStepId = createdIds.find((id) => id.startsWith('assistant-'));
+      const newStepId = mockCreateMessage.mock.calls.find(([p]: any) => p.role === 'assistant')?.[0]
-      if (newStepId) {
+        .id;
-        const newStepWrite = mockUpdateMessage.mock.calls.find(
+      expect(newStepId).toBeDefined();
-          ([id, val]: any) => id === newStepId && val.content === 'Step 2 content',
+      const newStepWrite = mockUpdateMessage.mock.calls.find(
-        );
+        ([id, val]: any) => id === newStepId && val.content === 'Step 2 content',
-        expect(newStepWrite).toBeDefined();
+      );
-      }
+      expect(newStepWrite).toBeDefined();
     });
   });
 
@@ -1306,6 +1278,92 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         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', () => {
@@ -1352,12 +1410,12 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool += 1;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
 
         if (params.role === 'assistant') {
           idCounter.assistant += 1;
-          return { id: `ast-new-${idCounter.assistant}` };
+          return { id: params.id ?? `ast-new-${idCounter.assistant}` };
         }
 
         return { id: `created-${params.role}-${idCounter.assistant + idCounter.tool}` };
@@ -1394,11 +1452,19 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         },
       );
 
+      // The second turn opens exactly one new assistant (pre-allocated id).
+      const newAstId = mockCreateMessage.mock.calls.find(
+        ([params]: any) => params.role === 'assistant',
+      )![0].id;
+      const item1ToolId = mockCreateMessage.mock.calls.find(
+        ([params]: any) => params.role === 'tool' && params.tool_call_id === 'item_1',
+      )![0].id;
+
       const secondTurnAssistantCreate = mockCreateMessage.mock.calls.find(
         ([params]: any) => params.role === 'assistant',
       );
       expect(secondTurnAssistantCreate?.[0]).toMatchObject({
-        parentId: 'tool-1',
+        parentId: item1ToolId,
         role: 'assistant',
       });
 
@@ -1415,7 +1481,7 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ([params]: any) => params.role === 'tool' && params.tool_call_id === 'item_3',
       );
       expect(secondToolCreate?.[0]).toMatchObject({
-        parentId: 'ast-new-1',
+        parentId: newAstId,
         role: 'tool',
         tool_call_id: 'item_3',
       });
@@ -1426,7 +1492,7 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       expect(firstTurnToolWrites.length).toBeGreaterThanOrEqual(1);
 
       const secondTurnToolWrites = toolsUpdates.filter(
-        (update) => update.assistantId === 'ast-new-1' && update.toolIds.includes('item_3'),
+        (update) => update.assistantId === newAstId && update.toolIds.includes('item_3'),
       );
       expect(secondTurnToolWrites.length).toBeGreaterThanOrEqual(1);
     });
@@ -1487,12 +1553,12 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool += 1;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
 
         if (params.role === 'assistant') {
           idCounter.assistant += 1;
-          return { id: `ast-new-${idCounter.assistant}` };
+          return { id: params.id ?? `ast-new-${idCounter.assistant}` };
         }
 
         return { id: `created-${params.role}-${idCounter.assistant + idCounter.tool}` };
@@ -1573,12 +1639,18 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ([params]: any) => params.role === 'assistant',
       );
       expect(assistantCreates).toHaveLength(2);
+      const toolIdOf = (callId: string) =>
+        mockCreateMessage.mock.calls.find(
+          ([p]: any) => p.role === 'tool' && p.tool_call_id === callId,
+        )![0].id;
+      const newAst1 = assistantCreates[0]![0].id;
+      const newAst2 = assistantCreates[1]![0].id;
       expect(assistantCreates[0]?.[0]).toMatchObject({
-        parentId: 'tool-3',
+        parentId: toolIdOf('item_3'),
         role: 'assistant',
       });
       expect(assistantCreates[1]?.[0]).toMatchObject({
-        parentId: 'tool-5',
+        parentId: toolIdOf('item_6'),
         role: 'assistant',
       });
 
@@ -1593,14 +1665,14 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
 
       const secondStepToolWrites = toolsUpdates.filter(
         (update) =>
-          update.assistantId === 'ast-new-1' &&
+          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 === 'ast-new-2' && update.toolIds.length > 0,
+        (update) => update.assistantId === newAst2 && update.toolIds.length > 0,
       );
       expect(thirdStepToolWrites).toHaveLength(0);
 
@@ -1608,10 +1680,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         contentUpdates.findLast((update) => update.assistantId === 'ast-initial')?.content,
       ).toContain('Running the five read-only checks');
       expect(
-        contentUpdates.findLast((update) => update.assistantId === 'ast-new-1')?.content,
+        contentUpdates.findLast((update) => update.assistantId === newAst1)?.content,
       ).toContain('The workspace is dirty in a few files');
       expect(
-        contentUpdates.findLast((update) => update.assistantId === 'ast-new-2')?.content,
+        contentUpdates.findLast((update) => update.assistantId === newAst2)?.content,
       ).toContain('Confirmed the repo root');
     });
   });
@@ -1642,10 +1714,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       // Track ALL updateMessage calls to inspect tools[] writes
@@ -1678,10 +1750,14 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       );
       expect(gitlogToolUpdates.length).toBeGreaterThanOrEqual(1);
 
-      // ── Verify: Turn 2 tool registered on ast-new-1 (step 2 assistant) ──
+      // 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 === 'ast-new-1' && u.tools.some((t: any) => t.id === 'toolu_gitdiff'),
+        (u) => u.assistantId === step2AstId && u.tools.some((t: any) => t.id === 'toolu_gitdiff'),
       );
       expect(gitdiffToolUpdates.length).toBeGreaterThanOrEqual(1);
 
@@ -1694,7 +1770,7 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       const gitdiffToolCreate = mockCreateMessage.mock.calls.find(
         ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_gitdiff',
       );
-      expect(gitdiffToolCreate![0].parentId).toBe('ast-new-1');
+      expect(gitdiffToolCreate![0].parentId).toBe(step2AstId);
     });
 
     it('should register tools on correct assistant when turn has ONLY tool_use (no text)', async () => {
@@ -1704,10 +1780,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
@@ -1732,12 +1808,14 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ccResult(),
       ]);
 
-      // The tool should be registered on ast-new-1 (step 2 assistant), not ast-initial
+      // 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 ast-new-1
+      // All of them should be on the step-2 assistant
       for (const u of bashToolUpdates) {
-        expect(u.assistantId).toBe('ast-new-1');
+        expect(u.assistantId).toBe(step2AstId);
       }
     });
   });
@@ -1758,10 +1836,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       const toolsUpdates: Array<{ assistantId: string; toolIds: string[] }> = [];
@@ -1939,10 +2017,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-${idCounter.assistant}` };
       });
 
       // Collect tools[] writes per assistant
@@ -2016,10 +2094,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       const schemaPayload =
@@ -2045,10 +2123,24 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         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 === 'tool-1');
+      const skillResult = mockUpdateToolMessage.mock.calls.find(
-      const searchResult = mockUpdateToolMessage.mock.calls.find(([id]: any) => id === 'tool-2');
+        ([id]: any) => id === toolIdOf('toolu_skill'),
-      const getIssueResult = mockUpdateToolMessage.mock.calls.find(([id]: any) => id === 'tool-3');
+      );
+      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' });
@@ -2072,13 +2164,13 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
 
       const searchRegister = mockUpdateMessage.mock.calls.find(
         ([id, val]: any) =>
-          id === 'ast-new-1' && val.tools?.some((t: any) => t.id === 'toolu_search'),
+          id === step2AstId && val.tools?.some((t: any) => t.id === 'toolu_search'),
       );
       expect(searchRegister).toBeDefined();
 
       const getIssueRegister = mockUpdateMessage.mock.calls.find(
         ([id, val]: any) =>
-          id === 'ast-new-2' && val.tools?.some((t: any) => t.id === 'toolu_get_issue'),
+          id === step3AstId && val.tools?.some((t: any) => t.id === 'toolu_get_issue'),
       );
       expect(getIssueRegister).toBeDefined();
     });
@@ -2094,10 +2186,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -2122,17 +2214,18 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       );
       expect(readToolCreate![0].parentId).toBe('ast-initial');
       expect(readToolCreate![0].plugin.apiName).toBe('Read');
+      const readToolId = readToolCreate![0].id;
 
-      // 2. Read tool result written
+      // 2. Read tool result written (to the pre-allocated Read tool message id)
       expect(mockUpdateToolMessage).toHaveBeenCalledWith(
-        'tool-1',
+        readToolId,
         expect.objectContaining({ content: 'export default function App() {}' }),
         expect.any(Object),
       );
 
-      // 3. Step 2 assistant created with parentId = tool-1 (Read tool message)
+      // 3. Step 2 assistant created chained off the Read tool message
       const step2Create = mockCreateMessage.mock.calls.find(
-        ([p]: any) => p.role === 'assistant' && p.parentId === 'tool-1',
+        ([p]: any) => p.role === 'assistant' && p.parentId === readToolId,
       );
       expect(step2Create).toBeDefined();
 
@@ -2141,18 +2234,19 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_write',
       );
       expect(writeToolCreate).toBeDefined();
-      expect(writeToolCreate![0].parentId).toBe('ast-new-1');
+      expect(writeToolCreate![0].parentId).toBe(step2Create![0].id);
+      const writeToolId = writeToolCreate![0].id;
 
       // 5. Write tool result written
       expect(mockUpdateToolMessage).toHaveBeenCalledWith(
-        'tool-2',
+        writeToolId,
         expect.objectContaining({ content: 'File written' }),
         expect.any(Object),
       );
 
-      // 6. Step 3 assistant created with parentId = tool-2 (Write tool message)
+      // 6. Step 3 assistant created chained off the Write tool message
       const step3Create = mockCreateMessage.mock.calls.find(
-        ([p]: any) => p.role === 'assistant' && p.parentId === 'tool-2',
+        ([p]: any) => p.role === 'assistant' && p.parentId === writeToolId,
       );
       expect(step3Create).toBeDefined();
 
@@ -2577,7 +2671,7 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         if (params.role === 'user') {
           idCounter.user++;
@@ -3114,10 +3208,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -3150,10 +3244,14 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       // Two new assistants (step 1 + step 2); step 0 uses ast-initial
       expect(assistantCreates.length).toBe(2);
 
-      // Step 1 parent = Monitor tool from step 0 (tool-1)
+      const toolId = (callId: string) =>
-      expect(assistantCreates[0][0].parentId).toBe('tool-1');
+        mockCreateMessage.mock.calls.find(
-      // Step 2 parent = LAST tool from step 1 = Monitor_1 (tool-3)
+          ([p]: any) => p.role === 'tool' && p.tool_call_id === callId,
-      expect(assistantCreates[1][0].parentId).toBe('tool-3');
+        )![0].id;
+      // Step 1 parent = Monitor tool from step 0
+      expect(assistantCreates[0][0].parentId).toBe(toolId('toolu_mon_0'));
+      // Step 2 parent = LAST tool from step 1 = Monitor_1
+      expect(assistantCreates[1][0].parentId).toBe(toolId('toolu_mon_1'));
     });
 
     /**
@@ -3166,10 +3264,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -3190,13 +3288,16 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       );
       expect(assistantCreates.length).toBe(2);
 
-      // Step 1 parent = Monitor tool from step 0 (tool-1)
+      const monitorToolId = mockCreateMessage.mock.calls.find(
-      expect(assistantCreates[0][0].parentId).toBe('tool-1');
+        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_mon_0',
+      )![0].id;
+      // Step 1 parent = Monitor tool from step 0
+      expect(assistantCreates[0][0].parentId).toBe(monitorToolId);
 
       // Step 2 parent: step 1 was toolless, but the chain must skip back to
-      // step 0's Monitor (tool-1) so MessageCollector's assistant → tool →
+      // step 0's Monitor so MessageCollector's assistant → tool → assistant
-      // assistant walk keeps every assistant in the same group.
+      // walk keeps every assistant in the same group.
-      expect(assistantCreates[1][0].parentId).toBe('tool-1');
+      expect(assistantCreates[1][0].parentId).toBe(monitorToolId);
     });
 
     /**
@@ -3211,10 +3312,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -3238,13 +3339,16 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       // 4 new assistants (steps 1–4); step 0 reuses ast-initial
       expect(assistantCreates.length).toBe(4);
 
+      const monitorToolId = mockCreateMessage.mock.calls.find(
+        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_mon_0',
+      )![0].id;
       // All toolless steps chain back to the Monitor tool from step 0
-      expect(assistantCreates[0][0].parentId).toBe('tool-1');
+      expect(assistantCreates[0][0].parentId).toBe(monitorToolId);
-      expect(assistantCreates[1][0].parentId).toBe('tool-1');
+      expect(assistantCreates[1][0].parentId).toBe(monitorToolId);
-      expect(assistantCreates[2][0].parentId).toBe('tool-1');
+      expect(assistantCreates[2][0].parentId).toBe(monitorToolId);
-      // Step 4 also chains to tool-1 — its own step had no tools yet at
+      // Step 4 also chains to the Monitor tool — its own step had no tools yet
-      // step_start, the Bash tool only persists after stream_start fires.
+      // at step_start, the Bash tool only persists after stream_start fires.
-      expect(assistantCreates[3][0].parentId).toBe('tool-1');
+      expect(assistantCreates[3][0].parentId).toBe(monitorToolId);
     });
 
     /**
@@ -3261,10 +3365,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -3285,10 +3389,13 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
         ([p]: any) => p.role === 'assistant',
       );
       expect(assistantCreates.length).toBe(1);
-      // Step 1 parent should be the Monitor tool from step 0
+      // Step 1 parent should be the Monitor tool from step 0. The tool message
-      // result_msg_id is set by persistToolBatch Phase 2 (queued on persistQueue
+      // id is pre-allocated by the reducer (carried in the persistToolBatch
-      // BEFORE the step_boundary persistQueue.then), so this should work.
+      // intent), so the chain resolves even though the tool_result interleaves.
-      expect(assistantCreates[0][0].parentId).toBe('tool-1');
+      const monitorToolId = mockCreateMessage.mock.calls.find(
+        ([p]: any) => p.role === 'tool' && p.tool_call_id === 'toolu_mon_0',
+      )![0].id;
+      expect(assistantCreates[0][0].parentId).toBe(monitorToolId);
     });
   });
 
@@ -3316,10 +3423,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([
@@ -3387,10 +3494,10 @@ describe('heterogeneousAgentExecutor DB persistence', () => {
       mockCreateMessage.mockImplementation(async (params: any) => {
         if (params.role === 'tool') {
           idCounter.tool++;
-          return { id: `tool-${idCounter.tool}` };
+          return { id: params.id ?? `tool-${idCounter.tool}` };
         }
         idCounter.assistant++;
-        return { id: `ast-new-${idCounter.assistant}` };
+        return { id: params.id ?? `ast-new-${idCounter.assistant}` };
       });
 
       await runWithEvents([

src/store/chat/slices/aiChat/actions/heterogeneousAgentExecutor.ts

@@ -11,13 +11,12 @@ import {
   HeterogeneousAgentSessionErrorCode,
 } from '@lobechat/electron-client-ipc';
 import {
-  createSubagentRunsState,
+  createMainAgentRunState,
-  reduceSubagentRuns,
+  type MainAgentIntent,
-  type SubagentEventContext,
+  type MainAgentReduceCtx,
+  type MainAgentRunState,
+  reduceMainAgent,
   type SubagentIntent,
-  type SubagentReduceCtx,
-  type SubagentRunsState,
-  type ToolCallPayload,
 } from '@lobechat/heterogeneous-agents';
 import type {
   ChatMessageError,
@@ -284,21 +283,6 @@ const subscribeBroadcasts = (
   };
 };
 
-/**
- * Per-assistant-message persistence state — covers ONE assistant row's
- * `tools[]` JSONB and the de-dupe set for its tool_uses. Main-agent
- * and subagent-thread assistants each have their own instance; the
- * `tool_use.id → tool message DB id` lookup is SHARED globally across
- * all scopes (see `toolMsgIdByCallId` in `executeHeterogeneousAgent`)
- * because `tool_result` events identify the target by id alone.
- */
-interface ToolPersistenceState {
-  /** Ordered list of ChatToolPayload[] written to this assistant's tools JSONB */
-  payloads: ChatToolPayload[];
-  /** Set of tool_use.id that have been persisted (de-dupe guard) */
-  persistedIds: Set<string>;
-}
-
 /**
  * Thread-scoped in-memory dispatcher for a single subagent run. The
  * caller binds it to a per-spawn sub-operation whose
@@ -323,122 +307,6 @@ interface SubagentStoreDispatcher {
   update: (id: string, value: Partial<UIChatMessage>) => void;
 }
 
-/**
- * Runs the 3-phase tool persistence flow for ONE assistant message —
- * either the main-agent assistant or a subagent-thread-scoped assistant.
- * Same ordering guarantee in both scopes:
- *
- *   1. Pre-register tools[] on the assistant (no result_msg_id yet), so
- *      LobeHub's conversation-flow parser finds matching ids the moment
- *      tool messages land in DB — no orphan window.
- *   2. Create `role:'tool'` messages, one per fresh tool_use. `threadId`
- *      is only set for subagent scope (so the tool messages stay inside
- *      the subagent Thread and don't leak into the main topic).
- *   3. Re-write assistant.tools[] with the backfilled `result_msg_id`
- *      so the UI can hydrate tool results.
- *
- * Carries the latest accumulated text/reasoning into Phases 1+3 so DB
- * stays in sync with streamed content. Without this, the gateway
- * handler's `tool_end → fetchAndReplaceMessages` would read a
- * tools-only row and clobber in-memory streamed text in the UI.
- *
- * Idempotent against re-processing: tool_use ids already in
- * `state.persistedIds` are skipped.
- */
-const persistToolBatch = async (
-  incoming: ToolCallPayload[],
-  state: ToolPersistenceState,
-  assistantMessageId: string,
-  context: ConversationContext,
-  snapshot: { content: string; reasoning: string },
-  /**
-   * Global `tool_use.id → tool message DB id` map, populated by every
-   * call (main + every subagent run) so a later `tool_result` lookup
-   * finds its row without needing to know which scope created it.
-   */
-  toolMsgIdByCallId: Map<string, string>,
-  /**
-   * When set, tool messages are scoped to this thread (subagent mode) and
-   * Phase 1 / 3 target the subagent-thread assistant. Undefined = main
-   * agent scope (tools live under the main topic, threadId stays null).
-   */
-  threadId?: string,
-  /**
-   * Invoked immediately after each fresh tool's `role:'tool'` DB row is
-   * created, with the DB-generated id + the payload. Subagent callers
-   * use this to seed the thread's messagesMap bucket so the UI shows
-   * the tool bubble in sync with the DB row; main-agent callers leave
-   * it undefined (fetchAndReplaceMessages hydrates the main bucket).
-   */
-  onToolCreated?: (args: {
-    assistantMessageId: string;
-    toolMessageId: string;
-    tool: ToolCallPayload;
-  }) => void,
-) => {
-  const freshTools = incoming.filter((t) => !state.persistedIds.has(t.id));
-  if (freshTools.length === 0) return;
-
-  // Mark all fresh tools as persisted up front, so re-entrant calls (from
-  // Claude Code echoing tool_use blocks) are safely deduped.
-  for (const tool of freshTools) state.persistedIds.add(tool.id);
-
-  const buildUpdate = (): Record<string, any> => {
-    const update: Record<string, any> = { tools: state.payloads };
-    if (snapshot.content) update.content = snapshot.content;
-    if (snapshot.reasoning) update.reasoning = { content: snapshot.reasoning };
-    return update;
-  };
-
-  // ─── PHASE 1: pre-register tools[] on the assistant row ───
-  for (const tool of freshTools) state.payloads.push({ ...tool } as ChatToolPayload);
-  try {
-    await messageService.updateMessage(assistantMessageId, buildUpdate(), {
-      agentId: context.agentId,
-      topicId: context.topicId,
-    });
-  } catch (err) {
-    console.error('[HeterogeneousAgent] Failed to pre-register assistant tools:', err);
-  }
-
-  // ─── PHASE 2: create the tool messages ───
-  for (const tool of freshTools) {
-    try {
-      const result = await messageService.createMessage({
-        agentId: context.agentId,
-        content: '',
-        parentId: assistantMessageId,
-        plugin: {
-          apiName: tool.apiName,
-          arguments: tool.arguments,
-          identifier: tool.identifier,
-          type: tool.type as ChatToolPayload['type'],
-        },
-        role: 'tool',
-        threadId,
-        tool_call_id: tool.id,
-        topicId: context.topicId ?? undefined,
-      });
-      toolMsgIdByCallId.set(tool.id, result.id);
-      const entry = state.payloads.find((p) => p.id === tool.id);
-      if (entry) entry.result_msg_id = result.id;
-      onToolCreated?.({ assistantMessageId, toolMessageId: result.id, tool });
-    } catch (err) {
-      console.error('[HeterogeneousAgent] Failed to create tool message:', err);
-    }
-  }
-
-  // ─── PHASE 3: backfill result_msg_id on assistant.tools[] ───
-  try {
-    await messageService.updateMessage(assistantMessageId, buildUpdate(), {
-      agentId: context.agentId,
-      topicId: context.topicId,
-    });
-  } catch (err) {
-    console.error('[HeterogeneousAgent] Failed to finalize assistant tools:', err);
-  }
-};
-
 /**
  * Update a tool message's content in DB when tool_result arrives.
  *
@@ -518,13 +386,13 @@ export const executeHeterogeneousAgent = async (
     options?: { clearContent?: boolean },
   ) => {
     writeTopicStatus('failed');
-    get().internal_toggleToolCallingStreaming(currentAssistantMessageId, undefined);
+    get().internal_toggleToolCallingStreaming(mainState.currentAssistantId, undefined);
     get().completeOperation(operationId);
 
     if (options?.clearContent) {
       await messageService
         .updateMessage(
-          currentAssistantMessageId,
+          mainState.currentAssistantId,
           { content: '' },
           {
             agentId: context.agentId,
@@ -535,7 +403,7 @@ export const executeHeterogeneousAgent = async (
     }
 
     const updateResult = await messageService
-      .updateMessageError(currentAssistantMessageId, messageError, {
+      .updateMessageError(mainState.currentAssistantId, messageError, {
         agentId: context.agentId,
         groupId: context.groupId,
         threadId: context.threadId,
@@ -551,7 +419,7 @@ export const executeHeterogeneousAgent = async (
 
     get().internal_dispatchMessage(
       {
-        id: currentAssistantMessageId,
+        id: mainState.currentAssistantId,
         type: 'updateMessage',
         value: {
           ...(options?.clearContent ? { content: '' } : {}),
@@ -568,29 +436,26 @@ export const executeHeterogeneousAgent = async (
   let fallbackPromise: Promise<void> | undefined;
   let resumeFallbackTriggered = false;
 
-  // Track state for DB persistence (main-agent scope)
-  const toolState: ToolPersistenceState = {
-    payloads: [],
-    persistedIds: new Set(),
-  };
   /**
    * Global `tool_use.id → tool message DB id` lookup, shared across the
    * main agent and every subagent run. `tool_result` events identify
    * the target row by `toolCallId` alone (no scope context needed), so
-   * one flat map keeps the lookup trivial. Populated by every
+   * one flat map keeps the lookup trivial. Interpreter-owned (NOT reducer
-   * `persistToolBatch` call.
+   * state — it maps to DB row ids the reducer pre-allocates): populated when
+   * `applyMainIntent` / `applySubagentIntent` create tool messages, read by
+   * `persistToolResult` and the intervention handlers.
    */
   const toolMsgIdByCallId: Map<string, string> = new Map();
   /**
-   * Shared subagent run coordinator state (the pure reducer in
+   * Shared main-agent run coordinator state — the pure reducer in
-   * `@lobechat/heterogeneous-agents`). Holds the run map keyed by the
+   * `@lobechat/heterogeneous-agents`. Owns the main turn/step state machine
-   * main-agent Task tool_use id; the renderer interpreter
+   * (content accumulation, the `asst → tool → asst` parent chain incl. the
-   * (`applySubagentIntent`) maps the reducer's intents onto DB writes +
+   * `lastToolMsgIdEver` toolless-step rescue) AND the nested subagent runs
-   * live thread-bucket dispatch. Reassigned (commit-on-success) by
+   * (delegated to `reduceSubagentRuns`). The renderer interpreters
-   * `reduceAndApplySubagent`. Lives at executor scope because a subagent
+   * (`applyMainIntent` / `applySubagentIntent`) map its intents onto DB writes
-   * spawn can emit events before and after a main-agent step cut.
+   * + live UI. Reassigned (commit-on-success) by `reduceAndApplyMain`.
    */
-  let subagentState: SubagentRunsState = createSubagentRunsState();
+  let mainState: MainAgentRunState = createMainAgentRunState(assistantMessageId);
   /**
    * Per-thread UI handles the reducer doesn't model: the thread-scoped store
    * dispatcher + its sub-operation id. Created on the `createThread` intent,
@@ -614,32 +479,6 @@ export const executeHeterogeneousAgent = async (
   >();
   /** Serializes async persist operations so ordering is stable. */
   let persistQueue: Promise<void> = Promise.resolve();
-  /** Tracks the current assistant message being written to (switches on new steps) */
-  let currentAssistantMessageId = assistantMessageId;
-  /** Content accumulators — reset on each new step */
-  let accumulatedContent = '';
-  let accumulatedReasoning = '';
-  /** Latest model string — updated per turn, written alongside content on step boundaries. */
-  let lastModel: string | undefined;
-  /** Adapter/CLI provider (e.g. `claude-code`) — carried on every turn_metadata. */
-  let lastProvider: string | undefined;
-  /**
-   * Most recent tool `result_msg_id` seen across step boundaries — survives the
-   * `toolState.payloads` reset that happens on every new step.
-   *
-   * Required for the **toolless middle step** case (): when a step
-   * produces only text (e.g. Monitor stdout drives Claude to reply "等一下…"
-   * without invoking a tool), `toolState.payloads` is empty at the next step
-   * boundary. Without this tracker, `stepParentId` would fall back to
-   * `currentAssistantMessageId` (= the toolless assistant), forming an
-   * `assistant → assistant` link. `MessageCollector.collectAssistantChain`
-   * only walks the `assistant → tool → assistant` zigzag, so the UI splits
-   * into one bubble per Monitor stdout line.
-   *
-   * Scope: executor lifetime (one user run). A new user message spawns a
-   * new executor, so this resets implicitly at run boundaries.
-   */
-  let lastToolMsgIdEver: string | undefined;
   /**
    * Deferred terminal event (agent_runtime_end or error). We don't forward
    * these to the gateway handler immediately because handler triggers
@@ -654,6 +493,17 @@ export const executeHeterogeneousAgent = async (
    * Without this, tools_calling gets dispatched to the OLD assistant → orphan.
    */
   let pendingStepTransition = false;
+  /**
+   * Set synchronously the moment any output-bearing stream event (stream_chunk /
+   * tool_result) ARRIVES, before it's queued onto `persistQueue`. The reducer
+   * now accumulates content/tools/subagent state only INSIDE the queued
+   * `reduceAndApplyMain`, so `hasStreamedState()` (which reads `mainState`) is
+   * blind to events that arrived but haven't drained yet. `retryWithoutResume`
+   * runs its guard synchronously in `onError` BEFORE awaiting the queue, so
+   * without this flag a recoverable resume error landing after partial output
+   * was queued could start a second run and duplicate/interleave messages.
+   */
+  let sawStreamedEvent = false;
 
   // Subscribe to the operation's abort signal so we can drop late events and
   // stop writing to DB the moment the user clicks Stop. If the op is gone
@@ -662,11 +512,12 @@ export const executeHeterogeneousAgent = async (
   const isAborted = () => !!abortSignal?.aborted;
   const updateTopicMetadata = get().updateTopicMetadata;
   const hasStreamedState = () =>
-    !!accumulatedContent ||
+    sawStreamedEvent ||
-    !!accumulatedReasoning ||
+    !!mainState.accContent ||
-    toolState.payloads.length > 0 ||
+    !!mainState.accReasoning ||
+    mainState.toolState.payloads.length > 0 ||
     toolMsgIdByCallId.size > 0 ||
-    subagentState.runs.size > 0;
+    mainState.subagents.runs.size > 0;
   const clearStaleResumeMetadata = async () => {
     if (!context.topicId || !updateTopicMetadata) return;
 
@@ -805,7 +656,7 @@ export const executeHeterogeneousAgent = async (
             type: ThreadType.Isolation,
           });
         } catch (err) {
-          // Rethrow so `reduceAndApplySubagent` skips the state commit — the
+          // Rethrow so `reduceAndApplyMain` skips the state commit — the
           // run stays absent and the next chunk retries the lazy create.
           console.error('[HeterogeneousAgent] Failed to create subagent thread:', err);
           throw err;
@@ -831,7 +682,7 @@ export const executeHeterogeneousAgent = async (
         try {
           await messageService.createMessage(msg);
         } catch (err) {
-          // Rethrow so `reduceAndApplySubagent` skips the state commit — the
+          // Rethrow so `reduceAndApplyMain` skips the state commit — the
           // run keeps its pre-create shape and the next event re-emits the
           // turn-boundary / lazy-create with fresh ids.
           console.error('[HeterogeneousAgent] Failed to create subagent message:', err);
@@ -1002,40 +853,209 @@ export const executeHeterogeneousAgent = async (
     }
   };
 
+  // ─── Main-agent run coordinator (shared reducer) interpreter ─────────────
+
   /**
-   * Reduce one event through the shared coordinator and apply its intents.
+   * Apply ONE main-scoped coordinator intent against the renderer's DB
-   * `mainAssistantId` is snapshotted at event-arrival time (the spawning main
+   * surfaces. Best-effort (errors logged, never thrown) — mirroring the prior
-   * assistant) and threaded in as the thread/seed parent. Commit-on-success:
+   * inline persist helpers, so the run state always advances regardless of a
-   * `subagentState` advances only after all intents land — a throwing create
+   * transient DB failure (the next event / terminal flush re-persists). Live
-   * intent (createThread / createMessage) skips the commit so the next event
+   * UI is NOT driven here: the executor still forwards raw stream events to the
-   * re-emits the lazy create / turn boundary, while flush failures are pinned
+   * gateway `eventHandler` for token-level streaming, so `streamContent` is a
-   * in `pendingSubagentFlush` for the onComplete replay (subsumes the old
+   * no-op (the server no-ops it too).
-   * `pendingFlushTarget`). Always invoked inside `persistQueue` so reduce reads
-   * the latest committed state and ordering matches arrival.
    */
-  const reduceAndApplySubagent = async (event: AgentStreamEvent, mainAssistantId: string) => {
+  const applyMainIntent = async (intent: MainAgentIntent) => {
-    // Without a topicId we can't scope a Thread — drop subagent routing
+    switch (intent.kind) {
-    // silently (non-topic-scoped run / test harness), matching the old guard.
+      case 'createAssistant': {
-    if (!context.topicId) return;
+        try {
-    const ctx: SubagentReduceCtx = {
+          await messageService.createMessage({
+            agentId: intent.agentId ?? context.agentId,
+            content: '',
+            id: intent.messageId,
+            ...(intent.signal ? { metadata: { signal: intent.signal } } : {}),
+            model: intent.model,
+            parentId: intent.parentId,
+            provider: intent.provider,
+            role: 'assistant',
+            topicId: intent.topicId ?? context.topicId ?? undefined,
+          } as any);
+        } catch (err) {
+          // Rethrow so `reduceAndApplyMain` skips the state commit — DO NOT
+          // advance `currentAssistantId` to a row that was never created, or
+          // every later content/tool/result write (and the gateway handler's
+          // switch) would target a missing assistant and be lost. Keeping the
+          // prior state lets the next event re-derive against the still-valid
+          // current assistant. Mirrors the subagent createMessage path.
+          console.error('[HeterogeneousAgent] Failed to create step assistant:', err);
+          throw err;
+        }
+        // Associate so cancellation / cleanup tracks the new step's message.
+        get().associateMessageWithOperation(intent.messageId, operationId);
+        return;
+      }
+
+      // Durable flush of content/reasoning/model/provider/metadata.
+      case 'persistAssistant': {
+        const update: Record<string, any> = {};
+        if (intent.content !== undefined) update.content = intent.content;
+        if (intent.reasoning !== undefined) update.reasoning = { content: intent.reasoning };
+        if (intent.model) update.model = intent.model;
+        if (intent.provider) update.provider = intent.provider;
+        if (intent.metadata) update.metadata = intent.metadata;
+        if (Object.keys(update).length === 0) return;
+        await messageService
+          .updateMessage(intent.messageId, update, {
+            agentId: context.agentId,
+            topicId: context.topicId,
+          })
+          .catch((err) =>
+            console.error('[HeterogeneousAgent] Failed to flush main assistant:', err),
+          );
+        return;
+      }
+
+      // No-op ON PURPOSE — not dead code. Main-agent live token UI is already
+      // driven by forwarding the RAW stream_chunk to the gateway `eventHandler`
+      // (see handleStreamEvent: it dispatches into `messagesMap` for live
+      // display). Applying streamContent here too would be a redundant double
+      // write. (Contrast the SUBAGENT interpreter, whose `streamContent` DOES
+      // update the thread bucket — subagent events are dropped before the
+      // gateway forward, so the intent is their only live-UI path.)
+      // Verified: in-memory content streams 3→…→N while the op runs; the durable
+      // write still lands via persistAssistant / persistToolBatch.
+      case 'streamContent': {
+        return;
+      }
+
+      case 'persistToolBatch': {
+        const buildUpdate = (withResult: boolean): Record<string, any> => {
+          const update: Record<string, any> = {
+            tools: intent.tools.map((x) =>
+              withResult ? { ...x.payload, result_msg_id: x.toolMessageId } : { ...x.payload },
+            ),
+          };
+          if (intent.content) update.content = intent.content;
+          if (intent.reasoning) update.reasoning = { content: intent.reasoning };
+          return update;
+        };
+
+        // Phase 1: pre-register assistant.tools[] (no result_msg_id yet) so the
+        // conversation-flow parser finds matching ids the moment tool rows land.
+        await messageService
+          .updateMessage(intent.assistantMessageId, buildUpdate(false), {
+            agentId: context.agentId,
+            topicId: context.topicId,
+          })
+          .catch((err) =>
+            console.error('[HeterogeneousAgent] Failed to pre-register main tools:', err),
+          );
+
+        // Phase 2: create rows for new tools with their pre-allocated ids and
+        // register the global lookup so a later tool_result resolves.
+        for (const x of intent.tools) {
+          if (!x.isNew) continue;
+          try {
+            await messageService.createMessage({
+              agentId: context.agentId,
+              content: '',
+              id: x.toolMessageId,
+              parentId: intent.assistantMessageId,
+              plugin: {
+                apiName: x.payload.apiName,
+                arguments: x.payload.arguments,
+                identifier: x.payload.identifier,
+                type: x.payload.type as ChatToolPayload['type'],
+              },
+              role: 'tool',
+              tool_call_id: x.payload.id,
+              topicId: context.topicId ?? undefined,
+            } as any);
+          } catch (err) {
+            console.error('[HeterogeneousAgent] Failed to create main tool message:', err);
+            continue;
+          }
+          toolMsgIdByCallId.set(x.payload.id, x.toolMessageId);
+        }
+
+        // Phase 3: backfill result_msg_id on assistant.tools[].
+        await messageService
+          .updateMessage(intent.assistantMessageId, buildUpdate(true), {
+            agentId: context.agentId,
+            topicId: context.topicId,
+          })
+          .catch((err) =>
+            console.error('[HeterogeneousAgent] Failed to finalize main tools:', err),
+          );
+        return;
+      }
+
+      case 'resolveToolResult': {
+        await persistToolResult(
+          intent.toolCallId,
+          intent.content,
+          intent.isError,
+          toolMsgIdByCallId,
+          context,
+          intent.pluginState,
+        );
+        return;
+      }
+
+      case 'recordUsage': {
+        const update = {
+          metadata: { usage: intent.usage as any },
+          ...(intent.model && { model: intent.model }),
+          ...(intent.provider && { provider: intent.provider }),
+        };
+        await messageService
+          .updateMessage(intent.messageId, update, {
+            agentId: context.agentId,
+            topicId: context.topicId,
+          })
+          .catch((err) => console.error('[HeterogeneousAgent] Failed to record main usage:', err));
+        return;
+      }
+
+      // Terminal error: classify the raw wire data here (adapterType lives in
+      // the interpreter) and route through the full error-UI routine.
+      case 'setError': {
+        const messageError = toHeterogeneousAgentMessageError(intent.errorData, adapterType);
+        await persistTerminalError(messageError, { clearContent: intent.clearContent });
+        return;
+      }
+    }
+  };
+
+  /**
+   * Reduce one event through the shared main-agent coordinator and apply its
+   * intents. The reducer returns a mix of main-scoped and (delegated)
+   * subagent-scoped intents; route each by whether it carries a `threadId`
+   * (subagent) so the existing `applySubagentIntent` stays untouched.
+   *
+   * Commit-on-success: `mainState` advances only after every intent lands. A
+   * throwing intent (only subagent createThread / createMessage rethrow; main
+   * intents are best-effort) skips the commit so the next event re-derives —
+   * the same resilience `reduceSubagentRuns` relies on. Always invoked inside
+   * `persistQueue` so reduce reads the latest committed state and ordering
+   * matches arrival.
+   */
+  const reduceAndApplyMain = async (event: AgentStreamEvent) => {
+    const ctx: MainAgentReduceCtx = {
       agentId: context.agentId,
-      mainAssistantId,
       newId: (kind) => (kind === 'thread' ? generateThreadId() : `msg_${createNanoId(18)()}`),
       topicId: context.topicId ?? null,
     };
-    const { state: next, intents } = reduceSubagentRuns(subagentState, event, ctx);
+    const { intents, state: next } = reduceMainAgent(mainState, event, ctx);
     try {
-      for (const intent of intents) await applySubagentIntent(intent);
+      for (const intent of intents) {
+        if ('threadId' in intent) await applySubagentIntent(intent as SubagentIntent);
+        else await applyMainIntent(intent as MainAgentIntent);
+      }
     } catch (err) {
-      // An intent failed to land (e.g. transient IndexedDB / message-service
+      console.error('[HeterogeneousAgent] Intent failed, run state not advanced:', err);
-      // error on createThread / createMessage). Do NOT commit `next`: keeping
-      // the prior state lets the next event re-emit the create / flush, and
-      // keeps the run visible to the onComplete orphan drain. Swallow here so
-      // the rejection doesn't poison the shared persistQueue chain.
-      console.error('[HeterogeneousAgent] Subagent intent failed, run state not advanced:', err);
       return;
     }
-    subagentState = next;
+    mainState = next;
   };
 
   try {
@@ -1161,195 +1181,52 @@ export const executeHeterogeneousAgent = async (
         return;
       }
 
-      // ─── tool_result: update tool message content in DB (ACP-only) ───
+      // ─── tool_result: reducer writes the tool content + finalizes spawns ───
+      // For a main tool (incl. a subagent's parent Task tool, which is
+      // main-scoped) the reducer emits `resolveToolResult` → DB content write
+      // via the global `toolMsgIdByCallId` map, AND delegates so a parent-spawn
+      // tool_result finalizes its run (terminal assistant + thread Active). An
+      // inner subagent tool_result resolves into its thread. Not forwarded —
+      // the following tool_end triggers fetchAndReplaceMessages.
       if (event.type === 'tool_result') {
-        const { content, isError, pluginState, subagent, toolCallId } = event.data as {
+        sawStreamedEvent = true; // sync: partial output exists even before the queue drains
-          content: string;
+        persistQueue = persistQueue.then(() => reduceAndApplyMain(event));
-          isError?: boolean;
-          pluginState?: Record<string, any>;
-          subagent?: SubagentEventContext;
-          toolCallId: string;
-        };
-
-        // Main tools (including a subagent's parent Task tool, which is
-        // main-scoped) get their DB content written here via the global
-        // `toolMsgIdByCallId` map. Subagent INNER tool_results are skipped —
-        // the coordinator's `resolveToolResult` intent owns their DB write +
-        // thread-bucket update (avoids a double write).
-        if (!subagent) {
-          persistQueue = persistQueue.then(() =>
-            persistToolResult(
-              toolCallId,
-              content,
-              !!isError,
-              toolMsgIdByCallId,
-              context,
-              pluginState,
-            ),
-          );
-        }
-
-        // Route through the coordinator: an inner subagent tool_result →
-        // resolveToolResult (DB + live thread bucket); a parent-spawn
-        // tool_result → finalize (terminal assistant + thread Active); a plain
-        // main tool_result → no intents. Queued so earlier subagent chunks in
-        // the same batch have registered the run before the parent finalize
-        // checks for it.
-        const mainAsstId = currentAssistantMessageId;
-        persistQueue = persistQueue.then(() => reduceAndApplySubagent(event, mainAsstId));
-
-        // Don't forward — the tool_end that follows triggers fetchAndReplaceMessages
-        // which reads the updated content from DB.
         return;
       }
 
-      // ─── step_complete with turn_metadata: persist per-step usage ───
+      // ─── step_complete with turn_metadata: per-step usage + model/provider ───
-      // `turn_metadata.usage` is the per-turn delta (deduped by adapter per
+      // The reducer writes usage/model/provider onto the current step's
-      // message.id) and already normalized to the MessageMetadata.usage
+      // assistant (main) or the subagent's in-thread assistant (delegated).
-      // shape — write it straight through to the current step's assistant
+      // `result_usage` (grand total) is ignored by the reducer. Main usage also
-      // message. Queue the write so it lands after any in-flight
+      // feeds the operation's usage-metrics tray — a renderer-only side effect
-      // stream_start(newStep) that may still be swapping
+      // the reducer doesn't model, so it stays here. Not forwarded (bookkeeping).
-      // `currentAssistantMessageId` to the new step's message.
-      //
-      // `result_usage` (grand total across all turns) is intentionally
-      // ignored — applying it would overwrite the last step with the sum
-      // of all prior steps. Sum of turn_metadata equals result_usage for
-      // a healthy run.
       if (event.type === 'step_complete' && event.data?.phase === 'turn_metadata') {
-        const turnUsage = event.data.usage;
+        if (!event.data.subagent && event.data.usage) {
-
-        // Subagent-tagged usage routes through the coordinator (RecordUsage
-        // intent → written onto the subagent's in-thread assistant + thread
-        // bucket). It must NOT touch the MAIN agent's `lastModel` /
-        // `lastProvider`, which carry main-agent step state.
-        if (event.data.subagent) {
-          const mainAsstId = currentAssistantMessageId;
-          persistQueue = persistQueue.then(() => reduceAndApplySubagent(event, mainAsstId));
-          return;
-        }
-
-        if (turnUsage) {
           const operation = get().operations[operationId];
           get().updateOperationMetadata(operationId, {
-            usageMetrics: addUsageToOperationMetrics(operation?.metadata?.usageMetrics, turnUsage),
+            usageMetrics: addUsageToOperationMetrics(
+              operation?.metadata?.usageMetrics,
+              event.data.usage,
+            ),
           });
         }
-
+        persistQueue = persistQueue.then(() => reduceAndApplyMain(event));
-        if (event.data.model) lastModel = event.data.model;
-        if (event.data.provider) lastProvider = event.data.provider;
-        const updateValue: Record<string, any> = {};
-        if (turnUsage) updateValue.metadata = { usage: turnUsage };
-        if (event.data.model) updateValue.model = event.data.model;
-        if (event.data.provider) updateValue.provider = event.data.provider;
-
-        if (Object.keys(updateValue).length > 0) {
-          persistQueue = persistQueue.then(async () => {
-            await messageService
-              .updateMessage(currentAssistantMessageId, updateValue, {
-                agentId: context.agentId,
-                topicId: context.topicId,
-              })
-              .catch(console.error);
-          });
-        }
-        // Don't forward turn metadata — it's internal bookkeeping
         return;
       }
 
-      // ─── stream_start with newStep: new LLM turn, create new assistant message ───
+      // ─── stream_start with newStep: new LLM turn ───
+      // The reducer flushes the prior turn's content/reasoning/model and opens
+      // a new assistant chained off the last tool message (the shared chain rule
+      // incl. the `lastToolMsgIdEver` toolless-step rescue — the 断链 fix). We
+      // then forward the event (carrying the new assistant id) for live UI.
       if (event.type === 'stream_start' && event.data?.newStep) {
-        // ⚠️ Snapshot CONTENT accumulators synchronously — stream_chunk events for
+        // Defer same-batch stream_chunk / tool events through persistQueue so the
-        // the new step arrive in the same stream batch and would contaminate.
+        // handler receives stream_start FIRST — otherwise it dispatches tools to
-        // Tool state (toolMsgIdByCallId) is populated ASYNC by persistQueue, so
+        // the OLD assistant (orphan tool bug).
-        // it must be read inside the queue where previous persists have completed.
-        const prevContent = accumulatedContent;
-        const prevReasoning = accumulatedReasoning;
-        const prevModel = lastModel;
-        const prevProvider = lastProvider;
-        // External-signal context (): set when the adapter
-        // detected a repeated tool_result on the same tool_use.id
-        // (Monitor stdout push, etc.). Stamp on the new message's
-        // `metadata.signal` so MessageCollector can route toolless
-        // signal-tagged assistants into a SignalCallbacksNode.
-        //
-        // Phase 1 lives in metadata; Phase 2 () promotes to a
-        // dedicated `messages.signal` column — to migrate, change THIS
-        // assignment and the `getMessageSignal()` helper in
-        // conversation-flow, nothing else.
-        const externalSignal = event.data.externalSignal;
-
-        // Reset content accumulators synchronously so new-step chunks go to fresh state
-        accumulatedContent = '';
-        accumulatedReasoning = '';
-
-        // Mark that we're in a step transition. Events from the same stream
-        // batch (stream_chunk, tool_start, etc.) must be deferred through
-        // persistQueue so the handler receives stream_start FIRST — otherwise
-        // it dispatches tools to the OLD assistant (orphan tool bug).
         pendingStepTransition = true;
-
+        persistQueue = persistQueue.then(() => reduceAndApplyMain(event));
-        persistQueue = persistQueue.then(async () => {
-          // Persist previous step's content to its assistant message
-          const prevUpdate: Record<string, any> = {};
-          if (prevContent) prevUpdate.content = prevContent;
-          if (prevReasoning) prevUpdate.reasoning = { content: prevReasoning };
-          if (prevModel) prevUpdate.model = prevModel;
-          if (prevProvider) prevUpdate.provider = prevProvider;
-          if (Object.keys(prevUpdate).length > 0) {
-            await messageService
-              .updateMessage(currentAssistantMessageId, prevUpdate, {
-                agentId: context.agentId,
-                topicId: context.topicId,
-              })
-              .catch(console.error);
-          }
-
-          // Create new assistant message for this step.
-          // parentId should point to the last tool message from the previous step
-          // (if any), forming the chain: assistant → tool → assistant → tool → ...
-          // If no tool was used, fall back to the previous assistant message.
-          //
-          // Read from `toolState.payloads` (not the global
-          // `toolMsgIdByCallId`) so we only pick up MAIN-agent tools —
-          // the global map also holds subagent tool msg ids which
-          // would break the main-agent step chain.
-          const lastToolMsgId = [...toolState.payloads]
-            .reverse()
-            .find((p) => !!p.result_msg_id)?.result_msg_id;
-          if (lastToolMsgId) lastToolMsgIdEver = lastToolMsgId;
-          // Prefer this step's last tool, then the most recent tool ever seen
-          // in the run (rescues toolless middle steps — see ), then
-          // the previous assistant as a last resort.
-          const stepParentId = lastToolMsgId ?? lastToolMsgIdEver ?? currentAssistantMessageId;
-
-          const newMsg = await messageService.createMessage({
-            agentId: context.agentId,
-            content: '',
-            ...(externalSignal ? { metadata: { signal: externalSignal } } : {}),
-            model: lastModel,
-            parentId: stepParentId,
-            provider: lastProvider,
-            role: 'assistant',
-            topicId: context.topicId ?? undefined,
-          });
-          currentAssistantMessageId = newMsg.id;
-
-          // Associate the new message with the operation
-          get().associateMessageWithOperation(currentAssistantMessageId, operationId);
-
-          // Reset tool state AFTER reading — new-step tool persists are queued
-          // AFTER this handler, so they'll write to the clean state.
-          toolState.payloads = [];
-          toolState.persistedIds.clear();
-          // toolMsgIdByCallId is NOT cleared — it's the global
-          // id→row lookup and subagent tool_results from a previous
-          // step may still land after the step boundary.
-        });
-
-        // Update the stream_start event to carry the new message ID
-        // so the gateway handler can switch to it
         persistQueue = persistQueue.then(() => {
-          event.data.assistantMessage = { id: currentAssistantMessageId };
+          event.data.assistantMessage = { id: mainState.currentAssistantId };
           eventHandler(event);
           // Step transition complete — handler has the new assistant ID now
           pendingStepTransition = false;
@@ -1365,49 +1242,20 @@ export const executeHeterogeneousAgent = async (
         return;
       }
 
-      // ─── stream_chunk: accumulate content + persist tool_use ───
+      // ─── stream_chunk / stream_start(init): drive the reducer for DB ───
-      if (event.type === 'stream_chunk') {
+      // text/reasoning accumulation, main tool-batch persistence, subagent
-        const chunk = event.data;
+      // delegation (thread create / turn boundary / tool persist / live thread
-
+      // bucket), and the init model/provider backfill all live in the reducer.
-        // Subagent-scoped chunks (text / reasoning / tools_calling) route
+      // Ordering is preserved by the single FIFO persistQueue. Live MAIN-scope
-        // through the shared coordinator — it owns thread create, turn
+      // UI is still driven by the raw-event forward below (subagent events are
-        // boundaries, tool persistence, and live thread-bucket streaming. Kept
+      // dropped from that forward).
-        // off the main path so main-agent snapshot logic stays untouched.
+      if (event.type === 'stream_chunk' || event.type === 'stream_start') {
-        if (chunk?.subagent) {
+        // A stream_chunk = partial output (text / reasoning / tools / subagent
-          const mainAsstId = currentAssistantMessageId;
+        // activity). Flag it synchronously so a resume-error retry can't fire
-          persistQueue = persistQueue.then(() => reduceAndApplySubagent(event, mainAsstId));
+        // before the queued reduce records it. (stream_start init carries only
-        } else {
+        // model/provider — no output — so it doesn't count.)
-          if (chunk?.chunkType === 'text' && chunk.content) {
+        if (event.type === 'stream_chunk') sawStreamedEvent = true;
-            accumulatedContent += chunk.content;
+        persistQueue = persistQueue.then(() => reduceAndApplyMain(event));
-          }
-          if (chunk?.chunkType === 'reasoning' && chunk.reasoning) {
-            accumulatedReasoning += chunk.reasoning;
-          }
-          if (chunk?.chunkType === 'tools_calling') {
-            const tools = chunk.toolsCalling as ToolCallPayload[];
-            if (tools?.length) {
-              // Snapshot accumulators sync — must travel with the same step's
-              // assistantMessageId. A late-bound getter would read the NEW
-              // step's content if a step transition lands between scheduling
-              // and execution, while assistantMessageId would still be the OLD
-              // one (also captured sync) → cross-step contamination.
-              const snapshot = {
-                content: accumulatedContent,
-                reasoning: accumulatedReasoning,
-              };
-              persistQueue = persistQueue.then(() =>
-                persistToolBatch(
-                  tools,
-                  toolState,
-                  currentAssistantMessageId,
-                  context,
-                  snapshot,
-                  toolMsgIdByCallId,
-                ),
-              );
-            }
-          }
-        }
       }
 
       // ALL subagent-tagged events are handled inline (tool_result, line
@@ -1473,21 +1321,28 @@ export const executeHeterogeneousAgent = async (
         // Wait for all tool persistence to finish before writing final state
         await persistQueue.catch(console.error);
 
-        // Drain any subagent runs that didn't see their parent's tool_result
+        const isErrorTerminal = deferredTerminalEvent?.type === 'error';
-        // (e.g. CLI crashed mid-subagent, or CC emitted the spawn's
+        // Snapshot the final content BEFORE the terminal reduce resets the
-        // tool_result after the stream closed). The coordinator flushes each
+        // accumulator — used for the completion notification body below.
-        // run's trailing content and marks the thread Active. Drive it with a
+        const finalContent = mainState.accContent;
-        // synthetic terminal event so the reducer's orphan-drain path runs.
+        const terminalEvent: AgentStreamEvent = deferredTerminalEvent ?? {
-        await reduceAndApplySubagent(
+          data: {},
-          deferredTerminalEvent ?? {
+          operationId,
-            data: {},
+          stepIndex: 0,
-            operationId,
+          timestamp: Date.now(),
-            stepIndex: 0,
+          type: 'agent_runtime_end',
-            timestamp: Date.now(),
+        };
-            type: 'agent_runtime_end',
+
-          },
+        // Reduce the terminal event through the shared coordinator: it flushes
-          currentAssistantMessageId,
+        // the last step's content/reasoning/model (with echo suppression), and
-        ).catch(console.error);
+        // — for an error terminal — emits `setError` → `persistTerminalError`
+        // (full error UI). It also drains any subagent run that never saw its
+        // parent tool_result (CLI crashed mid-subagent, or the spawn's
+        // tool_result arrived after the stream closed), flushing each run's
+        // trailing content and marking the thread Active. `completeOperation`
+        // does not cascade to child sub-ops, so the main-error path running
+        // before the drain still lets each subagent op finalize.
+        await reduceAndApplyMain(terminalEvent);
 
         // Replay any subagent flush that failed transiently mid-stream, pinned
         // to its original in-thread assistant (NOT the terminal row).
@@ -1508,56 +1363,18 @@ export const executeHeterogeneousAgent = async (
         }
         pendingSubagentFlush.clear();
 
-        // Persist final content + reasoning + model for the last step BEFORE the
+        if (!isErrorTerminal) {
-        // terminal event triggers fetchAndReplaceMessages. Usage for this step
-        // was already written per-turn via the turn_metadata branch.
-        const terminalMessageError =
-          deferredTerminalEvent?.type === 'error'
-            ? toHeterogeneousAgentMessageError(deferredTerminalEvent.data, adapterType)
-            : undefined;
-        const shouldClearTerminalErrorContent =
-          !!terminalMessageError &&
-          shouldSuppressTerminalErrorEcho(accumulatedContent, terminalMessageError);
-        const updateValue: Record<string, any> = {};
-        if (accumulatedContent && !shouldClearTerminalErrorContent) {
-          updateValue.content = accumulatedContent;
-        }
-        if (accumulatedReasoning) updateValue.reasoning = { content: accumulatedReasoning };
-        if (lastModel) updateValue.model = lastModel;
-        if (lastProvider) updateValue.provider = lastProvider;
-
-        if (Object.keys(updateValue).length > 0) {
-          await messageService
-            .updateMessage(currentAssistantMessageId, updateValue, {
-              agentId: context.agentId,
-              topicId: context.topicId,
-            })
-            .catch(console.error);
-        }
-
-        if (terminalMessageError) {
-          await persistTerminalError(terminalMessageError, {
-            clearContent: shouldClearTerminalErrorContent,
-          });
-        } else {
           writeTopicStatus('active');
-          // NOW forward the deferred terminal event — handler will fetchAndReplaceMessages
+          // NOW forward the deferred terminal event — handler will
-          // and pick up the final persisted state.
+          // fetchAndReplaceMessages and pick up the final persisted state.
-          const terminal: AgentStreamEvent = deferredTerminalEvent ?? {
+          eventHandler(terminalEvent);
-            data: {},
-            operationId,
-            stepIndex: 0,
-            timestamp: Date.now(),
-            type: 'agent_runtime_end',
-          };
-          eventHandler(terminal);
         }
 
         // Signal completion to the user — dock badge + (window-hidden) notification.
         // Skip for aborted runs and for error terminations.
-        if (!isAborted() && deferredTerminalEvent?.type !== 'error') {
+        if (!isAborted() && !isErrorTerminal) {
-          const body = accumulatedContent
+          const body = finalContent
-            ? markdownToTxt(accumulatedContent)
+            ? markdownToTxt(finalContent)
             : t('notification.finishChatGeneration', { ns: 'electron' });
           notifyCompletion(
             t('notification.finishChatGeneration', { ns: 'electron' }),
@@ -1581,15 +1398,15 @@ export const executeHeterogeneousAgent = async (
         const messageError =
           deferredMessageError || toHeterogeneousAgentMessageError(error, adapterType);
         const shouldClearTerminalErrorContent = shouldSuppressTerminalErrorEcho(
-          accumulatedContent,
+          mainState.accContent,
           messageError,
         );
 
-        if (accumulatedContent && !shouldClearTerminalErrorContent) {
+        if (mainState.accContent && !shouldClearTerminalErrorContent) {
           await messageService
             .updateMessage(
-              currentAssistantMessageId,
+              mainState.currentAssistantId,
-              { content: accumulatedContent },
+              { content: mainState.accContent },
               {
                 agentId: context.agentId,
                 topicId: context.topicId,
@@ -1699,7 +1516,7 @@ export const executeHeterogeneousAgent = async (
       }
       const messageError = toHeterogeneousAgentMessageError(error, adapterType);
       await persistTerminalError(messageError, {
-        clearContent: shouldSuppressTerminalErrorEcho(accumulatedContent, messageError),
+        clearContent: shouldSuppressTerminalErrorEcho(mainState.accContent, messageError),
       });
     }
   } finally {