Method runChatSubscription

src/node/acp/agent.ts:1529–1706

(
    sessionId: string,
    workspaceId: string,
    onConnected: () => void,
    onChatMode: OnChatMode,
    subscriptionToken: symbol
  )

  }

  private async runChatSubscription(
    sessionId: string,
    workspaceId: string,
    onConnected: () => void,
    onChatMode: OnChatMode,
    subscriptionToken: symbol
  ): Promise<void> {
    const chatStream = await this.server.client.workspace.onChat({
      workspaceId,
      mode: onChatMode,
    });
    onConnected();
    this.touchSession(sessionId);

    // Decouple turn resolution from sessionUpdate writes.
    // handleStreamEvent must fire for every event regardless of stdout
    // backpressure. Without this decoupling, stream-end can stall behind
    // a blocked sessionUpdate write, preventing resolveTurn from firing
    // and causing prompt() to hang indefinitely.
    //
    // Keep memory bounded without stalling terminal-event observation.
    //
    // If stdout backpressure saturates the forwarding queue and we block the
    // drain loop, stream-end/stream-abort can be delayed behind thousands of
    // queued deltas, making prompt() appear hung. Instead, once saturated we
    // drop only high-volume chunk/replay events while preserving lifecycle/
    // control events (caught-up, tool-call-end, etc.) so turn-completion and
    // translator state remain correct.
    const { push, iterate, end } = createAsyncMessageQueue<WorkspaceChatMessage>();
    let queuedEventCount = 0;
    let droppedNonTerminalEventCount = 0;
    let hasCaughtUp = onChatMode.type !== "full";

    const isTerminalStreamEvent = (event: WorkspaceChatMessage): boolean =>
      event.type === "stream-end" ||
      event.type === "stream-abort" ||
      event.type === "stream-error" ||
      event.type === "error";

    const isDroppableUnderBackpressure = (event: WorkspaceChatMessage): boolean => {
      const isReplayMessageEvent =
        event.type === "message" &&
        ((event as { replay?: boolean }).replay === true || !hasCaughtUp);

      return (
        event.type === "stream-delta" ||
        event.type === "reasoning-delta" ||
        event.type === "tool-call-delta" ||
        event.type === "usage-delta" ||
        event.type === "session-usage-delta" ||
        event.type === "advisor-output" ||
        event.type === "advisor-reasoning-output" ||
        event.type === "bash-output" ||
        event.type === "init-output" ||
        // Drop replay history messages under saturation, but keep live message
        // events so ACP clients do not miss real-time conversation updates.
        isReplayMessageEvent
      );