Make CI structural match local check:structural (CL-7281)

.github/workflows/ci.yml

@@ -59,22 +59,7 @@ jobs:
         with:
           fetch-depth: 0
       - uses: ./.github/actions/setup-workbench
-      - name: Structural check self-tests
-        run: bun test scripts/checks/test
-      - run: bun run check:deletion
-      - run: bun run check:killdates
-      - run: bun run check:packages
-      - run: bun run check:licenses
-      - run: bun run check:db-gate
-      - run: bun run check:no-product-tenancy
-      - run: bun run check:browser-safe-subpaths
-      - run: bun run check:web-utilities
-      - run: bun run check:tailwind-source
-      - run: bun run check:ui-vocabulary
-      - run: bun run check:react-ui-drift
-      - run: bun run check:react-ui-pin
-      - run: bun run check:tool-package-pins
-      - run: bun run check:tool-package-freshness
+      - run: bun run check:structural
 
   e2e:
     runs-on: ubuntu-latest

AGENTS.md

@@ -51,7 +51,9 @@ behind human approval.
 ## Working conventions
 
 - `bun run check` (typecheck, lint, test, structural checks) must pass
-  before every commit.
+  before every commit. CI's `structural` job runs the same
+  `check:structural` list as local, including `check:report-error`, so
+  the two cannot drift.
 - Commit sequence per change: tests first ("Add tests for X"), then
   implementation ("X: what changed"), then docs ("Update docs: X"). One
   logical change per commit; commit messages are written for a public
@@ -130,7 +132,8 @@ check:*` script, so a violation fails CI rather than waiting for review.
   the current change's diff touches its line; a pre-existing catch is
   instead recorded in `scripts/checks/report-error-baseline.txt`, a debt
   ledger — not an allowlist — of 280 violations as of this check landing,
-  each one a real bug still to fix. Regenerate it with `bun run
+  each one a real bug still to fix. The same script runs in CI via
+  `check:structural`. Regenerate it with `bun run
 scripts/checks/report-error.ts --write-baseline` after fixing (or
   newly opting out) entries; a stale entry with no matching violation
   fails the check, so the ledger can only shrink. A violation already

apps/hub/src/inbox-unsnooze-sweep.ts

@@ -15,7 +15,6 @@
 // leaves the row for the next tick to retry rather than orphaning it —
 // the mail-then-claim lesson CL-7209 applied to the credential-expiry
 // sweep, applied here to the claim itself.
-import { getLogger } from "@intx/log";
 import { reportError } from "@corbits/error-sink";
 import {
   claimAndReopenSnooze,
@@ -49,7 +48,6 @@ export type InboxUnsnoozeSweepDeps = {
 };
 
 const POLL_INTERVAL_MS = 60 * 1000;
-const publishLog = getLogger(["hub", "inbox-unsnooze-sweep"]);
 
 function publishReopened(
   bus: Pick<MailboxEventBus, "publish">,
@@ -65,15 +63,11 @@ function publishReopened(
       { type: "mailbox", id: row.messageId, op: "enrich" },
     );
   } catch (error) {
-    publishLog.error(
-      "mailbox reopen event publish failed for {id} on tenant {tenantId}, principal {principalId}: {error}",
-      {
-        id: row.messageId,
-        tenantId: row.tenantId,
-        principalId: row.principalId,
-        error: error instanceof Error ? error.message : String(error),
-      },
-    );
+    reportError(error, {
+      operation: "inbox_unsnooze_sweep_publish",
+      tenantId: row.tenantId,
+      extra: { messageId: row.messageId, principalId: row.principalId },
+    });
   }
 }
 

apps/hub/test/inbox-unsnooze-sweep-report-error.test.ts

@@ -0,0 +1,65 @@
+// A bus publish failure after a committed reopen must still reach
+// reportError — the sweep's other catches already do; the publish helper
+// used to log only, so a flaky event bus left no refId.
+import { afterAll, beforeEach, describe, expect, mock, test } from "bun:test";
+import type { DueSnooze } from "@corbits/inbox";
+import type { MailboxEvent, MailboxEventScope } from "@corbits/mailbox";
+import type { InboxUnsnoozeSweepStore } from "../src/inbox-unsnooze-sweep";
+
+const reportErrorCalls: {
+  error: unknown;
+  context: Record<string, unknown>;
+}[] = [];
+
+mock.module("@corbits/error-sink", () => ({
+  reportError: (error: unknown, context: Record<string, unknown>) => {
+    reportErrorCalls.push({ error, context });
+    return "ref_test";
+  },
+  generateRefId: () => "ref_test",
+}));
+
+const { tickInboxUnsnoozeSweep } = await import("../src/inbox-unsnooze-sweep");
+
+function row(overrides: Partial<DueSnooze> = {}): DueSnooze {
+  return {
+    tenantId: "tnt_1",
+    principalId: "prn_1",
+    messageId: "msg_1",
+    ...overrides,
+  };
+}
+
+beforeEach(() => {
+  reportErrorCalls.length = 0;
+});
+afterAll(() => {
+  mock.restore();
+});
+
+describe("tickInboxUnsnoozeSweep reportError", () => {
+  test("a throwing mailbox publish reports through reportError and does not fail the tick", async () => {
+    const store: InboxUnsnoozeSweepStore = {
+      findDueSnoozes: async () => [row()],
+      claimAndReopen: async () => true,
+    };
+    const bus = {
+      publish(_scope: MailboxEventScope, _event: MailboxEvent) {
+        throw new Error("bus unavailable");
+      },
+    };
+
+    await tickInboxUnsnoozeSweep({ store, bus }, new Date());
+
+    expect(reportErrorCalls).toHaveLength(1);
+    const reported = reportErrorCalls[0]?.error;
+    expect(reported).toBeInstanceOf(Error);
+    if (!(reported instanceof Error)) throw new Error("expected Error");
+    expect(reported.message).toBe("bus unavailable");
+    expect(reportErrorCalls[0]?.context).toEqual({
+      operation: "inbox_unsnooze_sweep_publish",
+      tenantId: "tnt_1",
+      extra: { messageId: "msg_1", principalId: "prn_1" },
+    });
+  });
+});

packages/chat/src/turn-queue-report-error.test.ts

@@ -0,0 +1,155 @@
+// The turn queue's three catch paths (rejecting dispatch, claim-store
+// throw mid-drain, reclaim throw on enqueue) must call reportError in
+// the catch itself so a failure always has a refId — helpers that wrap
+// reportError are invisible to check:report-error.
+import { afterAll, beforeEach, describe, expect, mock, test } from "bun:test";
+import { createInMemoryTurnClaimStore } from "./turn-claims";
+import type { TurnClaimStore } from "./turn-claims";
+import type { QueuedTurn } from "./turn-queue";
+
+const reportErrorCalls: {
+  error: unknown;
+  context: Record<string, unknown>;
+}[] = [];
+
+mock.module("@corbits/error-sink", () => ({
+  reportError: (error: unknown, context: Record<string, unknown>) => {
+    reportErrorCalls.push({ error, context });
+    return "ref_test";
+  },
+  generateRefId: () => "ref_test",
+}));
+
+const { createWorkbenchTurnQueue } = await import("./turn-queue");
+
+function turn(messageId: string, text: string): QueuedTurn {
+  return {
+    messageId,
+    principalId: "prn_1",
+    recipients: ["ins_echo1@acme.example"],
+    parts: [{ kind: "text", text }],
+  };
+}
+
+beforeEach(() => {
+  reportErrorCalls.length = 0;
+});
+afterAll(() => {
+  mock.restore();
+});
+
+describe("createWorkbenchTurnQueue reportError", () => {
+  test("a rejecting dispatch reports through reportError and does not reject run()", async () => {
+    const queue = createWorkbenchTurnQueue({
+      claims: createInMemoryTurnClaimStore({ ttlMs: 60_000 }),
+      publish: () => undefined,
+    });
+    const boom = new Error("dispatch blew up");
+
+    await queue.run("wb_1", turn("msg_1", "one"), async () => {
+      throw boom;
+    });
+
+    expect(reportErrorCalls).toHaveLength(1);
+    expect(reportErrorCalls[0]?.error).toBe(boom);
+    expect(reportErrorCalls[0]?.context).toEqual({
+      operation: "chat.turnQueue.dispatch",
+      roomId: "wb_1",
+      extra: { messageIds: ["msg_1"] },
+    });
+  });
+
+  test("a claim-store throw mid-drain reports through reportError", async () => {
+    const holders = new Map<string, string>();
+    let n = 0;
+    const boom = new Error("claim store connection reset");
+    const claims: TurnClaimStore = {
+      async tryClaim(c) {
+        if (holders.has(c.workbenchId)) return false;
+        const t = String(++n);
+        holders.set(c.workbenchId, t);
+        return t;
+      },
+      async release(c, t) {
+        if (holders.get(c.workbenchId) !== t) return false;
+        holders.delete(c.workbenchId);
+        return true;
+      },
+      async holds() {
+        throw boom;
+      },
+    };
+    const queue = createWorkbenchTurnQueue({
+      claims,
+      publish: () => undefined,
+    });
+
+    await queue.run("wb", turn("m1", "one"), async () => undefined);
+
+    expect(reportErrorCalls).toHaveLength(1);
+    expect(reportErrorCalls[0]?.error).toBe(boom);
+    expect(reportErrorCalls[0]?.context).toEqual({
+      operation: "chat.turnQueue.drain",
+      roomId: "wb",
+    });
+  });
+
+  test("a throwing tryClaim on enqueue reclaim reports through reportError", async () => {
+    const holders = new Map<string, string>();
+    let n = 0;
+    let tryClaimCalls = 0;
+    const boom = new Error("reclaim connection reset");
+    const claims: TurnClaimStore = {
+      async tryClaim(c) {
+        tryClaimCalls += 1;
+        // First run wins. The second run's opening tryClaim sees the holder
+        // and returns false (enqueue). Its follow-up reclaim tryClaim throws.
+        if (tryClaimCalls === 1) {
+          const t = String(++n);
+          holders.set(c.workbenchId, t);
+          return t;
+        }
+        if (tryClaimCalls === 2) return false;
+        throw boom;
+      },
+      async release(c, t) {
+        if (holders.get(c.workbenchId) !== t) return false;
+        holders.delete(c.workbenchId);
+        return true;
+      },
+      async holds(c, t) {
+        return holders.get(c.workbenchId) === t;
+      },
+    };
+    const queue = createWorkbenchTurnQueue({
+      claims,
+      publish: () => undefined,
+    });
+    let resolveFirst: (() => void) | undefined;
+    let holdFirstDispatch = true;
+    const first = queue.run("wb", turn("m1", "one"), () => {
+      if (!holdFirstDispatch) return Promise.resolve();
+      holdFirstDispatch = false;
+      return new Promise<void>((resolve) => {
+        resolveFirst = resolve;
+      });
+    });
+    for (let i = 0; i < 20 && resolveFirst === undefined; i++) {
+      await Promise.resolve();
+    }
+    expect(resolveFirst).toBeDefined();
+
+    // In-flight claim: this run enqueues, then its reclaim tryClaim throws.
+    await queue.run("wb", turn("m2", "two"), async () => undefined);
+
+    expect(reportErrorCalls).toHaveLength(1);
+    expect(reportErrorCalls[0]?.error).toBe(boom);
+    expect(reportErrorCalls[0]?.context).toEqual({
+      operation: "chat.turnQueue.enqueueReclaim",
+      roomId: "wb",
+    });
+
+    resolveFirst?.();
+    await first;
+  });
+});

packages/chat/src/turn-queue.ts

@@ -87,40 +87,6 @@ export type WorkbenchTurnQueue = {
   ): Promise<void>;
 };
 
-function reportDispatchFailure(
-  workbenchId: string,
-  batch: readonly QueuedTurn[],
-  err: unknown,
-): void {
-  const messageIds = batch.map((t) => t.messageId);
-  const refId = reportError(err, {
-    operation: "chat.turnQueue.dispatch",
-    roomId: workbenchId,
-    extra: { messageIds },
-  });
-  // `dispatch` documents "must never reject" (see `DispatchTurnBatch`);
-  // reaching here means some caller broke that contract.
-  log.error(
-    'turn queue: dispatch rejected for workbench {workbenchId}, message(s) {messageIds} (ref {refId}), violating its "never reject" contract: {err}',
-    { workbenchId, messageIds, refId, err },
-  );
-}
-
-function reportClaimStoreFailure(
-  operation: string,
-  workbenchId: string,
-  err: unknown,
-): void {
-  const refId = reportError(err, {
-    operation,
-    roomId: workbenchId,
-  });
-  log.error(
-    "turn queue: claim store call failed for workbench {workbenchId} (ref {refId}): {err}",
-    { workbenchId, refId, err },
-  );
-}
-
 export function createWorkbenchTurnQueue(
   deps: WorkbenchTurnQueueDeps,
 ): WorkbenchTurnQueue {
@@ -169,7 +135,18 @@ export function createWorkbenchTurnQueue(
         try {
           await dispatch(batch);
         } catch (err) {
-          reportDispatchFailure(workbenchId, batch, err);
+          const messageIds = batch.map((t) => t.messageId);
+          const refId = reportError(err, {
+            operation: "chat.turnQueue.dispatch",
+            roomId: workbenchId,
+            extra: { messageIds },
+          });
+          // `dispatch` documents "must never reject" (see `DispatchTurnBatch`);
+          // reaching here means some caller broke that contract.
+          log.error(
+            'turn queue: dispatch rejected for workbench {workbenchId}, message(s) {messageIds} (ref {refId}), violating its "never reject" contract: {err}',
+            { workbenchId, messageIds, refId, err },
+          );
         }
 
         if (!(await deps.claims.holds({ workbenchId }, token))) {
@@ -215,7 +192,14 @@ export function createWorkbenchTurnQueue(
         batch = afterReclaim;
       }
     } catch (err) {
-      reportClaimStoreFailure("chat.turnQueue.drain", workbenchId, err);
+      const refId = reportError(err, {
+        operation: "chat.turnQueue.drain",
+        roomId: workbenchId,
+      });
+      log.error(
+        "turn queue: claim store call failed for workbench {workbenchId} (ref {refId}): {err}",
+        { workbenchId, refId, err },
+      );
       // Best-effort: `token` may already be released (this call then
       // no-ops per the store's contract) or may be the one this loop
       // never got to release because the failure happened first either
@@ -265,10 +249,13 @@ export function createWorkbenchTurnQueue(
         }
         await drain(workbenchId, reclaimed, batch, dispatch);
       } catch (err) {
-        reportClaimStoreFailure(
-          "chat.turnQueue.enqueueReclaim",
-          workbenchId,
-          err,
+        const refId = reportError(err, {
+          operation: "chat.turnQueue.enqueueReclaim",
+          roomId: workbenchId,
+        });
+        log.error(
+          "turn queue: claim store call failed for workbench {workbenchId} (ref {refId}): {err}",
+          { workbenchId, refId, err },
         );
       }
     },