fix(#922): always report a scenario run as finished, and make event delivery retry

javascript/src/events/__tests__/event-bus.test.ts

@@ -0,0 +1,160 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+import { EventBus } from "../event-bus";
+import {
+  ScenarioEventType,
+  type ScenarioEvent,
+  type ScenarioRunStartedEvent,
+  type ScenarioRunFinishedEvent,
+  ScenarioRunStatus,
+  Verdict,
+} from "../schema";
+
+vi.mock("../event-alert-message-logger", () => ({
+  EventAlertMessageLogger: vi.fn().mockImplementation(function (this: unknown) {
+    return { handleGreeting: vi.fn(), handleWatchMessage: vi.fn() };
+  }),
+}));
+
+function makeStartedEvent(runId = "run-1"): ScenarioRunStartedEvent {
+  return {
+    type: ScenarioEventType.RUN_STARTED,
+    batchRunId: "batch-1",
+    scenarioId: "scenario-1",
+    scenarioRunId: runId,
+    scenarioSetId: "default",
+    timestamp: Date.now(),
+    metadata: { name: "test-name", description: "test-description" },
+  };
+}
+
+function makeFinishedEvent(runId = "run-1"): ScenarioRunFinishedEvent {
+  return {
+    type: ScenarioEventType.RUN_FINISHED,
+    batchRunId: "batch-1",
+    scenarioId: "scenario-1",
+    scenarioRunId: runId,
+    scenarioSetId: "default",
+    timestamp: Date.now(),
+    status: ScenarioRunStatus.SUCCESS,
+    results: {
+      verdict: Verdict.SUCCESS,
+      metCriteria: [],
+      unmetCriteria: [],
+    },
+  };
+}
+
+function makeBus(postEvent: (event: ScenarioEvent) => Promise<{ setUrl?: string }>) {
+  const bus = new EventBus({
+    endpoint: "https://example.test",
+    apiKey: "test-key",
+  });
+  // Swap the private reporter for a controllable one.
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  (bus as any).eventReporter = { postEvent };
+  return bus;
+}
+
+/** Fails the test instead of hanging when drain regresses into a deadlock. */
+async function drainWithDeadline(bus: EventBus, timeoutMs = 5_000): Promise<void> {
+  await Promise.race([
+    bus.drain(),
+    new Promise<never>((_, reject) =>
+      setTimeout(() => reject(new Error("drain() did not resolve promptly")), timeoutMs)
+    ),
+  ]);
+}
+
+describe("EventBus", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("retries a transient failure and delivers on the third attempt", async () => {
+    let attempts = 0;
+    const bus = makeBus(async () => {
+      attempts += 1;
+      if (attempts < 3) throw new Error("transient failure");
+      return {};
+    });
+
+    bus.listen();
+    bus.publish(makeStartedEvent());
+    await drainWithDeadline(bus);
+
+    expect(attempts).toBe(3);
+  });
+
+  it("drops a permanently failing event without terminating the stream", async () => {
+    const delivered: ScenarioEvent[] = [];
+    const bus = makeBus(async (event) => {
+      if (event.type === ScenarioEventType.RUN_STARTED) {
+        throw new Error("endpoint is down");
+      }
+      delivered.push(event);
+      return {};
+    });
+
+    bus.listen();
+    bus.publish(makeStartedEvent());
+    bus.publish(makeFinishedEvent());
+    await drainWithDeadline(bus);
+
+    // The failing RUN_STARTED was dropped after retries; the stream stayed
+    // alive and still delivered the RUN_FINISHED that followed it.
+    expect(delivered.map((e) => e.type)).toEqual([ScenarioEventType.RUN_FINISHED]);
+  });
+
+  it("does not retry a permanent 4xx client error", async () => {
+    let attempts = 0;
+    const bus = makeBus(async () => {
+      attempts += 1;
+      throw Object.assign(new Error("bad request"), { status: 400 });
+    });
+
+    bus.listen();
+    bus.publish(makeStartedEvent());
+    await drainWithDeadline(bus);
+
+    expect(attempts).toBe(1);
+  });
+
+  it("retries 429 responses", async () => {
+    let attempts = 0;
+    const bus = makeBus(async () => {
+      attempts += 1;
+      if (attempts < 2) throw Object.assign(new Error("rate limited"), { status: 429 });
+      return {};
+    });
+
+    bus.listen();
+    bus.publish(makeStartedEvent());
+    await drainWithDeadline(bus);
+
+    expect(attempts).toBe(2);
+  });
+
+  it("drain resolves on stream completion even when RUN_FINISHED never arrives", async () => {
+    const bus = makeBus(async () => ({}));
+
+    bus.listen();
+    bus.publish(makeStartedEvent());
+
+    // No RUN_FINISHED published: drain must resolve when the stream
+    // completes instead of sitting on the 300s timeout.
+    await drainWithDeadline(bus);
+  });
+
+  it("removes the bus from the static registry after drain", async () => {
+    const bus = makeBus(async () => ({}));
+
+    expect(EventBus.getAllBuses().has(bus)).toBe(true);
+
+    bus.listen();
+    bus.publish(makeFinishedEvent());
+    await drainWithDeadline(bus);
+
+    expect(EventBus.getAllBuses().has(bus)).toBe(false);
+  });
+});

javascript/src/events/__tests__/event-reporter.test.ts

@@ -56,6 +56,19 @@ function makeAudioSnapshotEvent(): ScenarioEvent {
   } as unknown as ScenarioEvent;
 }
 
+/** A base64 run long enough to be recognised as an audio payload. */
+const LONG_BASE64_AUDIO = "QUJDRA".repeat(60);
+
+/** The same snapshot shape, with an audio payload of realistic size. */
+function makeLongAudioSnapshotEvent(): ScenarioEvent {
+  const event = makeAudioSnapshotEvent() as unknown as {
+    messages: { content: { input_audio?: { data: string } }[] }[];
+  };
+  const audioPart = event.messages[0]!.content[1]!;
+  audioPart.input_audio!.data = LONG_BASE64_AUDIO;
+  return event as unknown as ScenarioEvent;
+}
+
 function makeEvent(): ScenarioRunStartedEvent {
   return {
     type: ScenarioEventType.RUN_STARTED,
@@ -219,6 +232,96 @@ describe("EventReporter", () => {
     expect(result).toEqual({});
   });
 
+  it("throws on a non-2xx response so the bus can retry, carrying the status", async () => {
+    const fetchMock = vi.fn().mockResolvedValue(
+      new Response("boom", { status: 500 })
+    );
+    vi.stubGlobal("fetch", fetchMock);
+    const reporter = new EventReporter({
+      endpoint: "https://app.langwatch.ai",
+      apiKey: "test-api-key",
+    });
+
+    await expect(reporter.postEvent(makeEvent())).rejects.toMatchObject({
+      status: 500,
+    });
+  });
+
+  it("throws on a network failure so the bus can retry", async () => {
+    const fetchMock = vi.fn().mockRejectedValue(new Error("connection refused"));
+    vi.stubGlobal("fetch", fetchMock);
+    const reporter = new EventReporter({
+      endpoint: "https://app.langwatch.ai",
+      apiKey: "test-api-key",
+    });
+
+    await expect(reporter.postEvent(makeEvent())).rejects.toThrow(
+      "connection refused"
+    );
+  });
+
+  it("succeeds after the bus retries a fetch that failed twice", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockRejectedValueOnce(new Error("transient failure"))
+      .mockRejectedValueOnce(new Error("transient failure"))
+      .mockResolvedValue(
+        new Response(JSON.stringify({ url: "https://app.langwatch.ai/s/run-1" }), {
+          status: 200,
+          headers: { "Content-Type": "application/json" },
+        })
+      );
+    vi.stubGlobal("fetch", fetchMock);
+    const reporter = new EventReporter({
+      endpoint: "https://app.langwatch.ai",
+      apiKey: "test-api-key",
+    });
+
+    await expect(reporter.postEvent(makeEvent())).rejects.toThrow("transient failure");
+    await expect(reporter.postEvent(makeEvent())).rejects.toThrow("transient failure");
+    const result = await reporter.postEvent(makeEvent());
+
+    expect(result.setUrl).toBe("https://app.langwatch.ai/s/run-1");
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+  });
+
+  it("treats a 2xx response with a body that is not JSON as delivered", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response("", { status: 202 }));
+    vi.stubGlobal("fetch", fetchMock);
+    const reporter = new EventReporter({
+      endpoint: "https://app.langwatch.ai",
+      apiKey: "test-api-key",
+    });
+
+    const result = await reporter.postEvent(makeEvent());
+
+    expect(result.setUrl).toBeUndefined();
+    expect(fetchMock).toHaveBeenCalledOnce();
+  });
+
+  it("keeps base64 audio out of the failure log", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValue(new Response("boom", { status: 500 }));
+    vi.stubGlobal("fetch", fetchMock);
+    const errorLog = vi.spyOn(console, "error").mockImplementation(() => {});
+    const reporter = new EventReporter({
+      endpoint: "https://app.langwatch.ai",
+      apiKey: "test-api-key",
+    });
+
+    await expect(
+      reporter.postEvent(makeLongAudioSnapshotEvent()),
+    ).rejects.toMatchObject({ status: 500 });
+
+    const logged = JSON.stringify(errorLog.mock.calls);
+    expect(logged).not.toContain(LONG_BASE64_AUDIO);
+    expect(logged).toContain("b64 chars elided");
+    errorLog.mockRestore();
+  });
+
   it("returns setUrl from a successful response", async () => {
     mockOkFetch();
     const reporter = new EventReporter({