fix(host-daemon,db): stop one undeliverable event from wedging every thread

apps/host-daemon/src/event-sink.test.ts

@@ -1,6 +1,21 @@
 import { threadScope } from "@bb/domain";
 import { describe, expect, it, vi } from "vitest";
 import { createEventSink, type CreateEventSinkOptions } from "./event-sink.js";
+import { ServerResponseError } from "./server-client.js";
+
+// The server rejects an event it can never store — e.g. a turn-scoped event
+// whose turn/started it never saw — with a non-retryable 409. Reposting the
+// identical batch always produces the identical rejection.
+function permanentRejection(bodyMessage: string): ServerResponseError {
+  return new ServerResponseError({
+    action: "post events",
+    bodyMessage,
+    code: "invalid_request",
+    retryable: false,
+    status: 409,
+    statusText: "Conflict",
+  });
+}
 
 function createLogger(): CreateEventSinkOptions["logger"] {
   return {
@@ -129,7 +144,7 @@ describe("event sink", () => {
     expect(postEvents).toHaveBeenCalledTimes(1);
   });
 
-  it("logs a debug tripwire once when the queue grows large while undelivered", () => {
+  it("warns once when the queue grows large while undelivered", () => {
     const logger = createLogger();
     const sink = createEventSink({
       isSessionOpen: () => false,
@@ -140,18 +155,198 @@ describe("event sink", () => {
     for (let index = 0; index < 511; index += 1) {
       sink.emit({ threadId: "thr_1", event: systemErrorEvent("thr_1") });
     }
-    expect(logger.debug).not.toHaveBeenCalled();
+    expect(logger.warn).not.toHaveBeenCalled();
 
     // Crossing the depth threshold fires the tripwire once...
     sink.emit({ threadId: "thr_1", event: systemErrorEvent("thr_1") });
     sink.emit({ threadId: "thr_1", event: systemErrorEvent("thr_1") });
-    expect(logger.debug).toHaveBeenCalledTimes(1);
-    expect(logger.debug).toHaveBeenCalledWith(
+    expect(logger.warn).toHaveBeenCalledTimes(1);
+    expect(logger.warn).toHaveBeenCalledWith(
       expect.objectContaining({ queueDepth: 512 }),
       expect.any(String),
     );
   });
 
+  it("drops a permanently rejected event instead of retrying it forever", async () => {
+    const logger = createLogger();
+    const postEvents = vi.fn<CreateEventSinkOptions["postEvents"]>(
+      async (events) => {
+        if (events.some((event) => event.threadId === "thr_poison")) {
+          throw permanentRejection(
+            "Cannot append provider/unhandled for turn auto-compact-1 before turn/started is stored",
+          );
+        }
+        return {
+          kind: "accepted",
+          acceptedEvents: events.map((event, eventIndex) => ({
+            eventIndex,
+            sequence: eventIndex + 1,
+            threadId: event.threadId,
+          })),
+          rejectedEvents: [],
+        };
+      },
+    );
+    const sink = createEventSink({
+      isSessionOpen: () => true,
+      logger,
+      postEvents,
+    });
+
+    sink.emit({
+      threadId: "thr_poison",
+      event: systemErrorEvent("thr_poison"),
+    });
+    await sink.flush();
+
+    // The poison event is gone, so a later flush has nothing left to send.
+    postEvents.mockClear();
+    await sink.flush();
+    expect(postEvents).not.toHaveBeenCalled();
+    expect(logger.error).toHaveBeenCalledTimes(1);
+  });
+
+  it("delivers events queued behind a permanently rejected event", async () => {
+    // The production wedge: one undeliverable event sat at the head of the
+    // single host-wide queue, so every other thread's events piled up behind it
+    // and never reached the server. Every thread showed as stuck until the app
+    // was restarted.
+    const delivered: string[] = [];
+    const postEvents = vi.fn<CreateEventSinkOptions["postEvents"]>(
+      async (events) => {
+        if (events.some((event) => event.threadId === "thr_poison")) {
+          throw permanentRejection(
+            "Cannot append provider/unhandled for turn auto-compact-1 before turn/started is stored",
+          );
+        }
+        delivered.push(...events.map((event) => event.threadId));
+        return {
+          kind: "accepted",
+          acceptedEvents: events.map((event, eventIndex) => ({
+            eventIndex,
+            sequence: eventIndex + 1,
+            threadId: event.threadId,
+          })),
+          rejectedEvents: [],
+        };
+      },
+    );
+    let sessionOpen = false;
+    const sink = createEventSink({
+      isSessionOpen: () => sessionOpen,
+      logger: createLogger(),
+      postEvents,
+    });
+
+    // One poison event, then healthy traffic from two other threads behind it,
+    // all accumulated while the session was closed — the same shape as the
+    // production queue at the moment the wedge began.
+    sink.emit({
+      threadId: "thr_poison",
+      event: systemErrorEvent("thr_poison"),
+    });
+    sink.emit({ threadId: "thr_a", event: systemErrorEvent("thr_a") });
+    sink.emit({ threadId: "thr_b", event: systemErrorEvent("thr_b") });
+    sink.emit({ threadId: "thr_a", event: systemErrorEvent("thr_a") });
+
+    await sink.flush();
+    expect(postEvents).not.toHaveBeenCalled();
+
+    sessionOpen = true;
+    await sink.flush();
+
+    expect(delivered).toEqual(["thr_a", "thr_b", "thr_a"]);
+
+    // Nothing is left behind: the queue fully drained.
+    postEvents.mockClear();
+    await sink.flush();
+    expect(postEvents).not.toHaveBeenCalled();
+  });
+
+  it("keeps retrying a batch that fails for a retryable reason", async () => {
+    const postEvents = vi
+      .fn<CreateEventSinkOptions["postEvents"]>()
+      .mockRejectedValueOnce(
+        new ServerResponseError({
+          action: "post events",
+          bodyMessage: "database is locked",
+          code: "internal_error",
+          retryable: true,
+          status: 500,
+          statusText: "Internal Server Error",
+        }),
+      )
+      .mockImplementation(async (events) => ({
+        kind: "accepted",
+        acceptedEvents: events.map((event, eventIndex) => ({
+          eventIndex,
+          sequence: eventIndex + 1,
+          threadId: event.threadId,
+        })),
+        rejectedEvents: [],
+      }));
+    const sink = createEventSink({
+      isSessionOpen: () => true,
+      logger: createLogger(),
+      postEvents,
+    });
+
+    sink.emit({ threadId: "thr_1", event: systemErrorEvent("thr_1") });
+    await sink.flush();
+    await sink.flush();
+
+    expect(postEvents).toHaveBeenCalledTimes(2);
+    expect(postEvents).toHaveBeenLastCalledWith([
+      { threadId: "thr_1", event: systemErrorEvent("thr_1") },
+    ]);
+  });
+
+  it("keeps events queued when the session, not the batch, is rejected", async () => {
+    // 401 inactive_session is a non-retryable 4xx that says nothing about the
+    // events — the daemon is about to reopen a session and deliver them. Only
+    // `invalid_request` means the payload itself is the problem, so these must
+    // survive rather than get bisected away one at a time.
+    const postEvents = vi
+      .fn<CreateEventSinkOptions["postEvents"]>()
+      .mockRejectedValueOnce(
+        new ServerResponseError({
+          action: "post events",
+          bodyMessage: "Session is not active",
+          code: "inactive_session",
+          retryable: false,
+          status: 401,
+          statusText: "Unauthorized",
+        }),
+      )
+      .mockImplementation(async (events) => ({
+        kind: "accepted",
+        acceptedEvents: events.map((event, eventIndex) => ({
+          eventIndex,
+          sequence: eventIndex + 1,
+          threadId: event.threadId,
+        })),
+        rejectedEvents: [],
+      }));
+    const sink = createEventSink({
+      isSessionOpen: () => true,
+      logger: createLogger(),
+      postEvents,
+    });
+
+    sink.emit({ threadId: "thr_1", event: systemErrorEvent("thr_1") });
+    sink.emit({ threadId: "thr_2", event: systemErrorEvent("thr_2") });
+    await sink.flush();
+
+    // Only the one failed attempt: no bisecting, nothing dropped.
+    expect(postEvents).toHaveBeenCalledTimes(1);
+
+    await sink.flush();
+    expect(postEvents).toHaveBeenLastCalledWith([
+      { threadId: "thr_1", event: systemErrorEvent("thr_1") },
+      { threadId: "thr_2", event: systemErrorEvent("thr_2") },
+    ]);
+  });
+
   it("never throws from emit regardless of how many events queue up", () => {
     const sink = createEventSink({
       isSessionOpen: () => false,