feat(harness): add bounded event store

2026-09-07 15:28:29 +08:00
parent e87c68f3ad
commit 7d1626dcde
6 changed files with 218 additions and 2 deletions
@@ -0,0 +1,85 @@
+import { describe, expect, it, vi } from "vitest";
+import * as harness from "../src/index.js";
+
+type Store = {
+  append(sessionId: string, input: Record<string, unknown>): { eventId: string; sequence: number; type: string };
+  listAfter(sessionId: string, cursor?: { eventId?: string; sequence?: number }): {
+    events: Array<{ eventId: string; sequence: number; type: string }>;
+    gap: boolean;
+    oldestSequence?: number;
+    latestSequence?: number;
+  };
+  subscribe(sessionId: string, listener: (event: unknown) => void): () => void;
+};
+
+function createStore(limit = 1000): Store {
+  const Constructor = Reflect.get(harness, "InMemoryEventStore") as (new (limit?: number) => Store) | undefined;
+  expect(Constructor, "InMemoryEventStore must be exported").toBeDefined();
+  return new Constructor!(limit);
+}
+
+const started = {
+  type: "session.started",
+  payload: { cwd: "workspace/sessions/ses_a" },
+};
+const runStarted = {
+  type: "run.started",
+  runId: "run_1",
+  payload: { modelId: "fake-model" },
+};
+const runCompleted = {
+  type: "run.completed",
+  runId: "run_1",
+  payload: { messageId: "msg_1" },
+};
+
+describe("InMemoryEventStore", () => {
+  it("allocates monotonic sequence independently per Session", () => {
+    const store = createStore();
+    expect(store.append("ses_a", started).sequence).toBe(1);
+    expect(store.append("ses_a", runStarted).sequence).toBe(2);
+    expect(store.append("ses_b", started).sequence).toBe(1);
+  });
+
+  it("bounds retained history and reports an old cursor gap", () => {
+    const store = createStore(2);
+    store.append("ses_a", started);
+    store.append("ses_a", runStarted);
+    store.append("ses_a", runCompleted);
+
+    const replay = store.listAfter("ses_a", { sequence: 0 });
+    expect(replay.events.map((event) => event.sequence)).toEqual([2, 3]);
+    expect(replay).toMatchObject({ gap: true, oldestSequence: 2, latestSequence: 3 });
+  });
+
+  it("uses eventId before sequence when both cursors are present", () => {
+    const store = createStore();
+    const first = store.append("ses_a", started);
+    const second = store.append("ses_a", runStarted);
+
+    expect(store.listAfter("ses_a", { eventId: first.eventId, sequence: 99 }).events)
+      .toEqual([second]);
+  });
+
+  it("notifies active subscribers after append and supports idempotent unsubscribe", () => {
+    const store = createStore();
+    const listener = vi.fn();
+    const unsubscribe = store.subscribe("ses_a", listener);
+
+    const event = store.append("ses_a", started);
+    expect(listener).toHaveBeenCalledWith(event);
+
+    unsubscribe();
+    unsubscribe();
+    store.append("ses_a", runStarted);
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+
+  it("rejects event payloads that do not match their type", () => {
+    const store = createStore();
+    expect(() => store.append("ses_a", {
+      type: "assistant.delta",
+      payload: { approvalId: "ap_1" },
+    })).toThrow();
+  });
+});