Method appendPartAndEmit

src/node/services/streamManager.ts:1195–1231

(
    workspaceId: WorkspaceId,
    streamInfo: WorkspaceStreamInfo,
    part: CompletedMessagePart,
    schedulePartialWrite = false
  )

  }

  private async appendPartAndEmit(
    workspaceId: WorkspaceId,
    streamInfo: WorkspaceStreamInfo,
    part: CompletedMessagePart,
    schedulePartialWrite = false
  ): Promise<void> {
    // Emit BEFORE adding to streamInfo.parts.
    //
    // On reconnect, we call replayStream() which snapshots streamInfo.parts. If we push a part to
    // streamInfo.parts and then await tokenization/emit, replay can include the "in-flight" part
    // and then the live emit still happens, causing duplicate deltas in the renderer.
    try {
      await this.emitPartAsEvent(workspaceId, streamInfo.messageId, part);
    } finally {
      // Always persist the part in-memory (and to partial.json, if enabled), even if emit fails.
      let partToPersist = part;
      let pendingAttachment: WorkflowRunToolAttachment | undefined;
      if (part.type === "dynamic-tool") {
        pendingAttachment = this.takePendingWorkflowRunAttachment(streamInfo, part.toolCallId);
        if (pendingAttachment != null) {
          partToPersist = { ...part, workflowRun: pendingAttachment };
        }
      }
      streamInfo.parts.push(partToPersist);
      if (pendingAttachment != null && part.type === "dynamic-tool") {
        await this.flushPartialWrite(workspaceId, streamInfo);
        this.emitWorkflowRunAttachedFromAttachment({
          workspaceId,
          messageId: streamInfo.messageId,
          toolCallId: part.toolCallId,
          attachment: pendingAttachment,
        });
      } else if (schedulePartialWrite) {
        void this.schedulePartialWrite(workspaceId, streamInfo);
      }
    }
  }

  private async cancelStreamSafely(
    workspaceId: WorkspaceId,