fix(harness): report event log write failures

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<typeof appendFileSync>) => {
+      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[] = [];

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<RuntimeEventListener>();
   private unsubscribes: Array<() => void> = [];
+  private pendingLogWarnings = new Map<string, RuntimeEvent>();
+  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);