diff --git a/packages/harness/package.json b/packages/harness/package.json index 2656ec3..c2a074b 100644 --- a/packages/harness/package.json +++ b/packages/harness/package.json @@ -12,6 +12,7 @@ "scripts": { "build": "tsc -p tsconfig.build.json", "lint": "eslint 'src/**/*.ts' 'tests/**/*.ts' --no-error-on-unmatched-pattern --max-warnings 0", + "pretest": "pnpm --filter @agent-studio/shared build", "typecheck": "tsc -p tsconfig.json", "test": "vitest run --passWithNoTests" }, diff --git a/packages/harness/src/events/event-store.ts b/packages/harness/src/events/event-store.ts new file mode 100644 index 0000000..17070b9 --- /dev/null +++ b/packages/harness/src/events/event-store.ts @@ -0,0 +1,27 @@ +import type { + EventInput, + HarnessEvent, + HarnessEventOf, + HarnessEventType, +} from "@agent-studio/shared"; + +export type EventCursor = { + eventId?: string; + sequence?: number; +}; + +export type EventReplay = { + events: HarnessEvent[]; + gap: boolean; + oldestSequence?: number; + latestSequence?: number; +}; + +export interface EventStore { + append( + sessionId: string, + event: EventInput, + ): HarnessEventOf; + listAfter(sessionId: string, cursor?: EventCursor): EventReplay; + subscribe(sessionId: string, listener: (event: HarnessEvent) => void): () => void; +} diff --git a/packages/harness/src/events/in-memory-event-store.ts b/packages/harness/src/events/in-memory-event-store.ts new file mode 100644 index 0000000..f65753f --- /dev/null +++ b/packages/harness/src/events/in-memory-event-store.ts @@ -0,0 +1,102 @@ +import { randomUUID } from "node:crypto"; +import { + HarnessEventSchema, + type EventInput, + type HarnessEvent, + type HarnessEventOf, + type HarnessEventType, +} from "@agent-studio/shared"; +import type { EventCursor, EventReplay, EventStore } from "./event-store.js"; + +type SessionEvents = { + nextSequence: number; + events: HarnessEvent[]; + listeners: Set<(event: HarnessEvent) => void>; +}; + +export class InMemoryEventStore implements EventStore { + readonly #sessions = new Map(); + + public constructor(private readonly maxEventsPerSession = 1000) { + if (!Number.isSafeInteger(maxEventsPerSession) || maxEventsPerSession < 1) { + throw new RangeError("maxEventsPerSession must be a positive integer"); + } + } + + public append( + sessionId: string, + input: EventInput, + ): HarnessEventOf { + const state = this.#state(sessionId); + const event = HarnessEventSchema.parse({ + ...input, + eventId: `evt_${randomUUID()}`, + sessionId, + sequence: state.nextSequence, + timestamp: new Date().toISOString(), + }) as HarnessEventOf; + + state.nextSequence += 1; + state.events.push(event); + if (state.events.length > this.maxEventsPerSession) { + state.events.splice(0, state.events.length - this.maxEventsPerSession); + } + + for (const listener of [...state.listeners]) { + listener(event); + } + + return event; + } + + public listAfter(sessionId: string, cursor?: EventCursor): EventReplay { + const state = this.#sessions.get(sessionId); + if (state === undefined || state.events.length === 0) { + return { events: [], gap: false }; + } + + const oldestSequence = state.events[0]!.sequence; + const latestSequence = state.events.at(-1)!.sequence; + let events: HarnessEvent[]; + let gap = false; + + if (cursor?.eventId !== undefined) { + const index = state.events.findIndex((event) => event.eventId === cursor.eventId); + gap = index < 0; + events = index < 0 ? [...state.events] : state.events.slice(index + 1); + } else if (cursor?.sequence !== undefined) { + const sequence = cursor.sequence; + gap = sequence < oldestSequence - 1; + events = state.events.filter((event) => event.sequence > sequence); + } else { + events = [...state.events]; + } + + return { events, gap, oldestSequence, latestSequence }; + } + + public subscribe(sessionId: string, listener: (event: HarnessEvent) => void): () => void { + const state = this.#state(sessionId); + state.listeners.add(listener); + let active = true; + + return () => { + if (!active) return; + active = false; + state.listeners.delete(listener); + }; + } + + #state(sessionId: string): SessionEvents { + const existing = this.#sessions.get(sessionId); + if (existing !== undefined) return existing; + + const created: SessionEvents = { + nextSequence: 1, + events: [], + listeners: new Set(), + }; + this.#sessions.set(sessionId, created); + return created; + } +} diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index cb0ff5c..93bfccf 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -1 +1,2 @@ -export {}; +export * from "./events/event-store.js"; +export * from "./events/in-memory-event-store.js"; diff --git a/packages/harness/tests/event-store.test.ts b/packages/harness/tests/event-store.test.ts new file mode 100644 index 0000000..9808261 --- /dev/null +++ b/packages/harness/tests/event-store.test.ts @@ -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): { 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(); + }); +}); diff --git a/packages/shared/package.json b/packages/shared/package.json index 65be3e3..75ae856 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -5,7 +5,7 @@ "type": "module", "exports": { ".": { - "types": "./dist/index.d.ts", + "types": "./src/index.ts", "default": "./dist/index.js" } },