diff --git a/server/harness/bus.test.ts b/server/harness/bus.test.ts index f3fdb010c..f67e2539d 100644 --- a/server/harness/bus.test.ts +++ b/server/harness/bus.test.ts @@ -1,9 +1,9 @@ // The bus is the seam every client depends on: events must arrive // stamped with their instanceId, cross-driver leaks must be dropped, and // neither logging nor a broken listener may take down the stream. -import { existsSync, readFileSync, rmSync } from "node:fs"; +import { appendFileSync, existsSync, readFileSync, rmSync } from "node:fs"; import { join } from "node:path"; -import { beforeEach, describe, expect, it } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { EVENTS_DIR, ensureDirs } from "../config.ts"; import type { RuntimeEvent } from "../contracts.ts"; @@ -87,17 +87,48 @@ describe("EventBus", () => { expect(logged).toContain("«redacted"); }); - it("still delivers when the NDJSON log cannot be written", () => { + it("reports an incomplete log once while continuing live delivery", () => { rmSync(EVENTS_DIR, { recursive: true, force: true }); const bus = new EventBus(); const seen: RuntimeEvent[] = []; bus.subscribe((e) => seen.push(e)); bus.publish(testEvent()); - expect(seen).toHaveLength(1); + bus.publish(testEvent({ eventId: "ev-2", type: "turn.completed", ok: true })); + + expect(seen).toHaveLength(3); + expect(seen[0]).toMatchObject({ + type: "runtime.error", + threadId: "thread-1", + message: expect.stringContaining("event history is incomplete"), + }); + expect(seen.slice(1).map((event) => event.eventId)).toEqual(["ev-1", "ev-2"]); expect(existsSync(EVENTS_DIR)).toBe(false); }); + it("writes the incomplete marker before the first event after logging recovers", () => { + let failing = true; + const writes: string[] = []; + const append: typeof appendFileSync = vi.fn((...args: Parameters) => { + if (failing) throw new Error("disk full"); + writes.push(String(args[1])); + }); + const bus = new EventBus(append); + const seen: RuntimeEvent[] = []; + bus.subscribe((event) => seen.push(event)); + + bus.publish(testEvent()); + failing = false; + bus.publish(testEvent({ eventId: "ev-2", type: "turn.completed", ok: true })); + bus.publish(testEvent({ eventId: "ev-3" })); + + const recovered = writes[0].trim().split("\n").map((line) => JSON.parse(line)); + expect(recovered.map((event) => event.type)).toEqual(["runtime.error", "turn.completed"]); + expect(recovered[0].message).toContain("event history is incomplete"); + expect(writes[1].trim()).toContain('"eventId":"ev-3"'); + expect(seen.filter((event) => event.type === "runtime.error")).toHaveLength(1); + }); + it("a throwing listener does not starve the others", () => { const bus = new EventBus(); const seen: RuntimeEvent[] = []; diff --git a/server/harness/bus.ts b/server/harness/bus.ts index 1e3809c46..40f698d61 100644 --- a/server/harness/bus.ts +++ b/server/harness/bus.ts @@ -9,11 +9,20 @@ import { join } from "node:path"; import { EVENTS_DIR } from "../config.ts"; import { redactSecrets } from "../redact.ts"; -import type { ProviderInstance, RuntimeEvent, RuntimeEventListener } from "../contracts.ts"; +import { newId, type ProviderInstance, type RuntimeEvent, type RuntimeEventListener } from "../contracts.ts"; + +const INCOMPLETE_LOG_MESSAGE = + "Canonical event history is incomplete: OpenMausBot could not write one or more events to disk. Live updates will continue."; export class EventBus { private listeners = new Set(); private unsubscribes: Array<() => void> = []; + private pendingLogWarnings = new Map(); + private readonly appendLog: typeof appendFileSync; + + constructor(appendLog: typeof appendFileSync = appendFileSync) { + this.appendLog = appendLog; + } attach(instances: ProviderInstance[]) { for (const instance of instances) { @@ -31,18 +40,42 @@ export class EventBus { } publish(event: RuntimeEvent) { + const pendingWarning = this.pendingLogWarnings.get(event.threadId); + const persistedEvents = pendingWarning ? [pendingWarning, redactSecrets(event)] : [redactSecrets(event)]; try { // the canonical log is a file people paste into bug reports; scrub // credential-shaped content (tool titles, request summaries, reply // text) the same way the native tee does - appendFileSync( + this.appendLog( join(EVENTS_DIR, `${event.threadId}.ndjson`), - JSON.stringify(redactSecrets(event)) + "\n", + persistedEvents.map((entry) => JSON.stringify(entry)).join("\n") + "\n", { mode: 0o600 }, ); - } catch { - /* logging must never take down the stream */ + if (pendingWarning) this.pendingLogWarnings.delete(event.threadId); + } catch (error) { + // Never feed this warning back through publish(): that would retry the + // same failed write and recurse. Deliver it once for this outage, then + // persist the same marker before the first event written after recovery. + if (!pendingWarning) { + const warning: RuntimeEvent = { + eventId: newId(), + provider: event.provider, + providerInstanceId: event.providerInstanceId, + threadId: event.threadId, + createdAt: new Date().toISOString(), + turnId: event.turnId, + type: "runtime.error", + message: INCOMPLETE_LOG_MESSAGE, + }; + this.pendingLogWarnings.set(event.threadId, warning); + console.error("bus: canonical event log write failed", error); + this.deliver(warning); + } } + this.deliver(event); + } + + private deliver(event: RuntimeEvent) { for (const listener of [...this.listeners]) { try { listener(event);