Name the operation on mailbox change events

ARCHITECTURE.md

@@ -230,6 +230,30 @@ subscriber independently; one throwing listener does not stop the others. SSE
 connections serialize writes, bound the pending queue, and close on overflow or
 write failure rather than buffering forever.
 
+**The event names the operation that fired.** `publishMailboxEvent` takes a
+required `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`, `trash`,
+`archive`, `restore`, `enrich`, `assign`) and includes it on the published
+event. Every call site in this package passes one — the two delivery paths
+(`writeMailboxMessage`, `deliverInboxItems`) and the transport dual-write
+(`createMailboxPersist`) publish `create`; `mountMailbox`'s route table passes
+the mutation's own identifier, reusing `MailboxBulkAction`'s vocabulary for
+the single-message verbs so "read one" and "read fifty" report the same op.
+`op` stays *optional on `MailboxEventSchema`* even though it is required to
+publish — additive, not a reshape: a listener built against the original
+`{ type, id }` shape still validates, and a historical event replayed from
+before this field existed still passes. Requiring it on `publishMailboxEvent`
+is what keeps every call site *in this package* honest going forward; it
+cannot reach a caller outside the package, which is the other reason the
+schema field has to stay optional.
+
+`MailboxEventOp` deliberately keeps its own name and vocabulary rather than
+reusing `MailboxBulkAction`. It is a superset — `create`, `enrich`, and
+`assign` are not bulk actions, and never will be — so aliasing the two would
+claim an equivalence that does not hold. `mount.ts`'s route table is the one
+place that has to know both: it maps HTTP verbs to `MailboxBulkAction` values
+that also happen to be valid `MailboxEventOp` values, and a test in
+`bus.test.ts` keeps that overlap from drifting silently.
+
 **Triage enriches the message, not a task.** `priority`, `classification`,
 `status` and `assignee` are columns on the message's management row, not a
 spawned work item. Delegation is the `assignee` ref: the item stays in the
@@ -343,8 +367,13 @@ actual mail transport — this package neither sends nor receives SMTP.
 - **SSE events are non-durable nudges.** Publication is best-effort after
   commit, each connection's queue is bounded at `MAX_PENDING_SSE_EVENTS` (100),
   and a consumer that stops reading is disconnected rather than buffered for.
-  The client contract — reconnect and refetch the list and unread count on any
-  disconnect — is documented in the package README.
+  Events can be missed (dropped publish, overflow disconnect); duplicated,
+  but only when there is no stable dedupe key to prevent it — an inbox item
+  redelivered without one, or a broker-backed bus itself redelivering; or
+  arrive out of order (no cross-replica ordering guarantee). The client
+  contract — reconnect and refetch the list and unread count on any
+  disconnect, and never trust event arrival order over a refetch — is
+  documented in the package README.
 - **`sort=priority` pays a cross-table join** on top of a rank that was never
   index-servable; see the measurements above.
 - **List routes read inbound rows.** The `direction` column admits outbound

CHANGELOG.md

@@ -9,6 +9,26 @@ always called out under their own heading.
 
 ## [Unreleased]
 
+### Added
+
+- **Live events name the operation that fired.** `MailboxEvent` gains an
+  optional `op` (`MailboxEventOp`: `create`, `mark_read`, `mark_unread`,
+  `trash`, `archive`, `restore`, `enrich`, `assign`) alongside the existing
+  `id` — a listener can react to a specific kind of change without
+  re-fetching and diffing the whole message. `op` is additive on the wire:
+  it is optional on `MailboxEventSchema`, so an existing listener reading
+  only `id` is unaffected, and a historical event replayed from before this
+  field existed still validates. `publishMailboxEvent` requires `op` — every
+  call site in this package always knew the operation, and the parameter now
+  enforces that a future call site can't silently regress to an op-less
+  event. `MAILBOX_EVENT_OPS` and `MailboxEventOp` are now exported.
+- **Delivery semantics are documented.** Events can be missed (best-effort
+  publish, bounded SSE queue with overflow disconnect); duplicated, but only
+  when there is no stable dedupe key to prevent it — an undeduped inbox
+  redelivery, or a broker-backed bus a host supplies redelivering itself; or
+  arrive out of order (no cross-replica ordering guarantee) — see the
+  README's SSE client contract and `MailboxEventBus`'s doc comment.
+
 ### Security
 
 - Require `drizzle-orm` `>= 0.45.2` (peer and dev pins, plus a root

examples/reference-host/test/acceptance.test.ts

@@ -355,8 +355,9 @@ describe("reference host", () => {
     expect(text).toContain("event: mailbox");
     // The id in the frame is the row that was just written, not merely "some"
     // event — a stream echoing the wrong id would pass a substring check.
+    // `op` names the operation that produced it — a new message is a `create`.
     expect(text).toContain(
-      JSON.stringify({ type: "mailbox", id: written!.id }),
+      JSON.stringify({ type: "mailbox", id: written!.id, op: "create" }),
     );
   });
 

packages/mailbox/README.md

@@ -114,17 +114,39 @@ All routes carry `describeRoute`, so they appear in the host's OpenAPI document.
 
 ### The SSE client contract
 
-**Events are non-durable nudges, not data.** Each event carries only an id to
-refetch; Postgres holds the truth. The server queues at most
-`MAX_PENDING_SSE_EVENTS` (100) events per connection — a consumer that stops
-reading is **disconnected**, not buffered for. So the contract for any client:
+**Events are non-durable nudges, not data.** Each event carries an id to
+refetch and, when the publisher knows it, `op` — which operation fired
+(`create`, `mark_read`, `mark_unread`, `trash`, `archive`, `restore`,
+`enrich`, `assign`). `op` is **additive**: it is optional on the wire, so a
+client that only reads `id` (the original shape) keeps working unchanged,
+and a client that wants to react to a specific kind of change (e.g. badge a
+new arrival differently from a read receipt) can switch on it instead of
+re-fetching and diffing every message. Postgres remains the source of truth
+either way — `op` narrows what changed, it does not replace a refetch when
+you need the new state. The server queues at most `MAX_PENDING_SSE_EVENTS`
+(100) events per connection — a consumer that stops reading is
+**disconnected**, not buffered for. So the contract for any client:
 
 - On **any** disconnect — network drop, server restart, or an overflow close —
   reconnect and **refetch from the API**: the list and the unread count. Never
   assume the stream told you everything that happened while you were away.
 - Do not treat the stream as a change log. It may drop events (publish is
   best-effort after commit) and the server may close a stream whose consumer
   stops draining it.
+- **Events can be missed, duplicated, or arrive out of order** — this is a
+  best-effort nudge channel, not a durable log:
+  - *Missed*: publish failures are logged and swallowed, never retried; an
+    overflowing connection is disconnected, not buffered for.
+  - *Duplicated*: redelivery is deduped by default — an inbox item with a
+    stable external identifier inserts once, on conflict-do-nothing. Only a
+    redelivered item with no such identifier inserts a second row and
+    publishes a second `create`; a broker-backed bus a host supplies for
+    multi-replica fan-out may also redeliver on its own. Treat a repeat of
+    an already-applied `op` for the same `id` as a no-op.
+  - *Out of order*: the default in-memory bus preserves publish order within
+    one process for one mailbox, but a broker-backed bus, or multiple
+    replicas publishing concurrently, gives no such guarantee. Never infer
+    "later event = later state" from arrival order.
 
 Four more behaviors to know before wiring a UI:
 

packages/mailbox/src/bus.test.ts

@@ -0,0 +1,66 @@
+import { describe, test, expect } from "bun:test";
+import { type } from "arktype";
+import {
+  MAILBOX_EVENT_OPS,
+  MailboxEventSchema,
+  publishMailboxEvent,
+  type MailboxEvent,
+} from "./bus.js";
+import { MAILBOX_BULK_ACTIONS } from "./mutations.js";
+
+const SCOPE = { tenantId: "t1", principalId: "p1" };
+const noopLogger = { error: () => {} };
+
+describe("MailboxEventSchema", () => {
+  test("accepts the pre-existing shape: type and id, no op", () => {
+    // The additive contract: a listener (or a stored/replayed event) built
+    // before `op` existed must still validate.
+    const result = MailboxEventSchema({ type: "mailbox", id: "row-1" });
+    expect(result instanceof type.errors).toBe(false);
+  });
+
+  test("accepts a known op", () => {
+    const result = MailboxEventSchema({
+      type: "mailbox",
+      id: "row-1",
+      op: "mark_read",
+    });
+    expect(result instanceof type.errors).toBe(false);
+  });
+
+  test("rejects an op outside the known vocabulary", () => {
+    const result = MailboxEventSchema({
+      type: "mailbox",
+      id: "row-1",
+      op: "delete_everything",
+    });
+    expect(result instanceof type.errors).toBe(true);
+  });
+});
+
+describe("MAILBOX_EVENT_OPS vs MAILBOX_BULK_ACTIONS", () => {
+  // MAILBOX_EVENT_OPS is duplicated from MAILBOX_BULK_ACTIONS rather than
+  // importing it (to avoid a bus.ts -> mutations.ts -> write.ts -> bus.ts
+  // cycle), and nothing at runtime enforces that the copy stays in sync.
+  // This is that enforcement: a bulk action added to mutations.ts without a
+  // matching entry here fails this test instead of silently losing its op.
+  test("every bulk action has a matching event op", () => {
+    for (const action of MAILBOX_BULK_ACTIONS) {
+      expect(MAILBOX_EVENT_OPS).toContain(action);
+    }
+  });
+});
+
+describe("publishMailboxEvent", () => {
+  test("publishes op on the event", () => {
+    const seen: MailboxEvent[] = [];
+    const bus = {
+      publish: (_scope: typeof SCOPE, event: MailboxEvent) => {
+        seen.push(event);
+      },
+      subscribe: () => () => {},
+    };
+    publishMailboxEvent(bus, SCOPE, "row-1", noopLogger, "archive");
+    expect(seen).toEqual([{ type: "mailbox", id: "row-1", op: "archive" }]);
+  });
+});

packages/mailbox/src/bus.ts

@@ -1,8 +1,58 @@
 import { type } from "arktype";
 
+/**
+ * The operation that produced an event, when the publisher knows it. Mirrors
+ * `MailboxBulkAction` in mutations.ts (`mark_read`, `mark_unread`, `trash`,
+ * `archive`, `restore`) plus the three operations mutations.ts does not own:
+ * `create` (a new message landed, from `writeMailboxMessage`,
+ * `deliverInboxItems`, or `createMailboxPersist`), `enrich` (triage stamp),
+ * `assign` (delegation). Duplicated here rather than imported from
+ * mutations.ts to avoid a bus.ts -> mutations.ts -> write.ts -> bus.ts import
+ * cycle; mount.ts's route table keeps the two lists in sync, and
+ * `bus.test.ts` asserts every `MailboxBulkAction` value is a member of this
+ * list.
+ *
+ * This is a deliberately different name from its two siblings, not an
+ * accident: mount.ts's HTTP route table calls the same five shared values a
+ * "verb" (the path segment), mutations.ts calls them an "action", and this
+ * is an "op". The three names share five values because a single-message
+ * mutation and its bulk equivalent report the same op, but `MailboxEventOp`
+ * is a strict superset — `create`, `enrich`, `assign` are not bulk actions
+ * and never will be — so this stays its own vocabulary rather than
+ * importing/aliasing `MailboxBulkAction`, which would claim an equivalence
+ * the two sets don't have.
+ */
+export const MAILBOX_EVENT_OPS = [
+  "create",
+  "mark_read",
+  "mark_unread",
+  "trash",
+  "archive",
+  "restore",
+  "enrich",
+  "assign",
+] as const;
+export type MailboxEventOp = (typeof MAILBOX_EVENT_OPS)[number];
+
+/**
+ * `op` is optional and additive on the wire: a listener that only reads `id`
+ * (the original shape) keeps working unchanged. A listener that wants to
+ * react to a specific kind of change without re-fetching and diffing can
+ * switch on `op` when present, and still fall back to a refetch when it is
+ * absent.
+ *
+ * It stays optional here — not because a caller *inside this package* might
+ * reasonably omit it (every call site names one) — but because two things
+ * outside this package's control cannot: a caller outside this package that
+ * predates `op` and has no reason to know about it, and a historical event
+ * replayed from before this field existed. `publishMailboxEvent` below makes
+ * `op` a required parameter precisely so no future call site here can
+ * silently produce one of the events this schema still has to tolerate.
+ */
 export const MailboxEventSchema = type({
   type: "'mailbox'",
   id: "string",
+  "op?": type.enumerated(...MAILBOX_EVENT_OPS),
 });
 export type MailboxEvent = typeof MailboxEventSchema.infer;
 
@@ -29,6 +79,27 @@ export type MailboxEventScope = { tenantId: string; principalId: string };
  * subscribers for that scope from receiving the event. Events are
  * best-effort nudges, so a per-listener failure is swallowable; starving
  * healthy connections is not.
+ *
+ * Delivery semantics — the same for every bus, in-memory or host-supplied:
+ *
+ * - **Events can be missed.** Publish is best-effort after commit (a
+ *   publish failure is logged and swallowed, never retried), and the SSE
+ *   route disconnects a consumer whose queue exceeds `MAX_PENDING_SSE_EVENTS`
+ *   rather than buffering for it. A listener must treat `id`/`op` as a hint
+ *   to refetch, not as a complete change log.
+ * - **Events can be duplicated, but only when there is no stable dedupe key
+ *   to prevent it.** Redelivery is deduped by default — an inbox item with a
+ *   stable external identifier inserts once, on conflict-do-nothing. Only a
+ *   redelivered item that carries no such identifier inserts a second row
+ *   and publishes a second `create`; a broker-backed bus a host supplies may
+ *   also redeliver on its own. Handling a repeat of an already-applied `op`
+ *   for the same `id` must be a no-op, not an error.
+ * - **Events can arrive out of order.** The in-memory default preserves
+ *   publish order within one process for one mailbox (a (tenant, principal)
+ *   pair — same scope as the fan-out guarantee above); a broker-backed bus,
+ *   or multiple replicas publishing concurrently, offers no such guarantee.
+ *   A listener must not infer "later event = later state" — refetch the
+ *   specific message (or the list) rather than trusting event arrival order.
  */
 export interface MailboxEventBus {
   publish(scope: MailboxEventScope, event: MailboxEvent): void;
@@ -40,15 +111,22 @@ export interface MailboxEventBus {
  * may be broker-backed and therefore may throw, and a publish failure must
  * never turn a committed write into a caller-visible error the client will
  * retry forever. The failure is logged and swallowed.
+ *
+ * `op` is required here even though it is optional on `MailboxEventSchema`:
+ * every call site in this package knows what it just did, and requiring it
+ * is what keeps a future call site from silently regressing to an
+ * op-less event. The schema field stays optional for the callers this
+ * function's signature cannot reach — see the comment above it.
  */
 export function publishMailboxEvent(
   bus: MailboxEventBus,
   scope: MailboxEventScope,
   id: string,
   logger: { error: (message: string, data?: Record<string, unknown>) => void },
+  op: MailboxEventOp,
 ): void {
   try {
-    bus.publish(scope, { type: "mailbox", id });
+    bus.publish(scope, { type: "mailbox", id, op });
   } catch (err) {
     logger.error("mailbox event publish failed for {rowId}", {
       rowId: id,

packages/mailbox/src/index.ts

@@ -55,11 +55,16 @@ export type { MailboxScopeIds } from "./write.js";
 
 export { purgeTenantMailbox, purgePrincipalMailbox } from "./purge.js";
 
-export { createInMemoryMailboxEventBus, MailboxEventSchema } from "./bus.js";
+export {
+  createInMemoryMailboxEventBus,
+  MailboxEventSchema,
+  MAILBOX_EVENT_OPS,
+} from "./bus.js";
 export type {
   MailboxEventBus,
   MailboxEvent,
   MailboxEventScope,
+  MailboxEventOp,
 } from "./bus.js";
 
 export {