Function finalizeStreamAndReadMessage

src/node/services/streamManager.test.ts:3233–3341

(params: {
    workspaceId: string;
    messageId: string;
    historySequence: number;
    startTime: number;
    parts: unknown[];
    initialMetadata?: Record<string, unknown>;
    emitStartEvent?: boolean;
    onStreamStart?: (event: Record<string, unknown>) => void;
    onStreamEnd?: (event: { metadata?: Record<string, unknown> }) => void;
    usage?: {
      inputTokens: number;
      outputTokens: number;
      totalTokens: number;
      reasoningTokens?: number;
    };
    model?: string;
    metadataModel?: string;
    streamManager?: StreamManager;
    beforeProcess?: (params: {
      streamManager: StreamManager;
      workspaceId: string;
      messageId: string;
    }) => Promise<void> | void;
  })

  }

  async function finalizeStreamAndReadMessage(params: {
    workspaceId: string;
    messageId: string;
    historySequence: number;
    startTime: number;
    parts: unknown[];
    initialMetadata?: Record<string, unknown>;
    emitStartEvent?: boolean;
    onStreamStart?: (event: Record<string, unknown>) => void;
    onStreamEnd?: (event: { metadata?: Record<string, unknown> }) => void;
    usage?: {
      inputTokens: number;
      outputTokens: number;
      totalTokens: number;
      reasoningTokens?: number;
    };
    model?: string;
    metadataModel?: string;
    streamManager?: StreamManager;
    beforeProcess?: (params: {
      streamManager: StreamManager;
      workspaceId: string;
      messageId: string;
    }) => Promise<void> | void;
  }) {
    const streamManager = params.streamManager ?? new StreamManager(historyService);
    // Suppress error events from bubbling up as uncaught exceptions during tests
    streamManager.on("error", () => undefined);

    if (params.onStreamStart) {
      streamManager.on("stream-start", params.onStreamStart);
    }
    if (params.onStreamEnd) {
      streamManager.on("stream-end", params.onStreamEnd);
    }

    const replaceTokenTrackerResult = Reflect.set(streamManager, "tokenTracker", {
      setModel: () => Promise.resolve(undefined),
      countTokens: () => Promise.resolve(0),
    });
    if (!replaceTokenTrackerResult) {
      throw new Error("Failed to mock StreamManager.tokenTracker");
    }

    await appendPartialAssistantForTests(
      params.workspaceId,
      params.messageId,
      params.historySequence
    );

    const processStreamWithCleanup = getProcessStreamWithCleanupForTests(streamManager);
    const usage = params.usage ?? { inputTokens: 4, outputTokens: 6, totalTokens: 10 };
    const streamInfo = createStreamInfoForTests({
      streamResult: createStreamResultForTests(
        (async function* () {
          // Tests pre-populate parts but still need the provider's terminal proof of completion.
          await Promise.resolve();
          yield { type: "finish", finishReason: "stop" };