openclaw/src/infra/session-cost-usage.stream-errors.test.ts

// Regression test: session-cost readline stream errors are swallowed instead of
// crashing the caller's async iteration.
import nodeFs from "node:fs";
import fs from "node:fs/promises";
import path from "node:path";
import { PassThrough } from "node:stream";
import { afterEach, describe, expect, it, vi } from "vitest";
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
import { withEnvAsync } from "../test-utils/env.js";
import { readSessionCostUsageRollupRows } from "./session-cost-usage-cache.sqlite.js";
import { loadCostUsageSummaryFromCache, loadSessionLogs } from "./session-cost-usage.js";

const tempDirs = useAutoCleanupTempDirTracker(afterEach);

describe("session cost usage stream errors", () => {
  afterEach(() => {
    vi.restoreAllMocks();
  });

  it("does not crash when the transcript stream emits an error mid-read", async () => {
    const tempDir = tempDirs.make("openclaw-session-cost-stream-");
    const sessionsDir = path.join(tempDir, "agents", "main", "sessions");
    await fs.mkdir(sessionsDir, { recursive: true });
    const sessionFile = path.join(sessionsDir, "sess-stream-error.jsonl");
    await fs.writeFile(
      sessionFile,
      [
        JSON.stringify({ type: "session", version: 1, id: "sess-stream-error" }),
        JSON.stringify({
          type: "message",
          timestamp: new Date().toISOString(),
          message: { role: "user", content: "hello" },
        }),
        "",
      ].join("\n"),
      "utf-8",
    );

    vi.spyOn(nodeFs, "createReadStream").mockImplementationOnce(() => {
      const stream = new PassThrough();
      stream.write(`${JSON.stringify({ type: "session", version: 1, id: "sess-stream-error" })}\n`);
      process.nextTick(() => {
        stream.destroy(new Error("stream read failed"));
      });
      return stream as unknown as nodeFs.ReadStream;
    });

    const logs = await loadSessionLogs({ sessionFile });

    expect(logs).toEqual([]);
  });

  it("does not persist a partial durable cache entry after a background stream error", async () => {
    const tempDir = tempDirs.make("openclaw-session-cost-cache-stream-");
    const sessionsDir = path.join(tempDir, "agents", "main", "sessions");
    await fs.mkdir(sessionsDir, { recursive: true });
    const sessionFile = path.join(sessionsDir, "sess-cache-stream-error.jsonl");
    const usageEntry = (timestamp: string, input: number) =>
      JSON.stringify({
        type: "message",
        timestamp,
        message: {
          role: "assistant",
          usage: { input, output: 0, totalTokens: input, cost: { total: input / 1000 } },
        },
      });
    await fs.writeFile(sessionFile, `${usageEntry("2026-07-06T12:00:00.000Z", 10)}\n`, "utf-8");

    await withEnvAsync({ OPENCLAW_STATE_DIR: tempDir }, async () => {
      const range = {
        startMs: Date.UTC(2026, 6, 6),
        endMs: Date.UTC(2026, 6, 7),
      };
      await loadCostUsageSummaryFromCache({
        ...range,
        refreshMode: "sync-when-empty",
      });
      const rollupsBefore = readSessionCostUsageRollupRows();

      const appendedEntry = `${usageEntry("2026-07-06T12:01:00.000Z", 20)}\n`;
      await fs.appendFile(sessionFile, appendedEntry, "utf-8");
      vi.spyOn(nodeFs, "createReadStream").mockImplementationOnce(() => {
        const stream = new PassThrough();
        stream.write(appendedEntry);
        process.nextTick(() => {
          stream.destroy(new Error("stream read failed"));
        });
        return stream as unknown as nodeFs.ReadStream;
      });

      await loadCostUsageSummaryFromCache(range);
      let summary = await loadCostUsageSummaryFromCache({ ...range, requestRefresh: false });
      await vi.waitFor(
        async () => {
          summary = await loadCostUsageSummaryFromCache({ ...range, requestRefresh: false });
          expect(summary.cacheStatus?.status).toBe("partial");
        },
        { interval: 5, timeout: 1_000 },
      );

      expect(readSessionCostUsageRollupRows()).toEqual(rollupsBefore);
      expect(summary.totals.totalTokens).toBe(10);
      expect(summary.cacheStatus?.pendingFiles).toBe(1);
    });
  });
});