diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md
index 29964ab..073efac 100644
--- a/ARCHITECTURE.md
+++ b/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
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0aeb85a..fd1ff3a 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/examples/reference-host/test/acceptance.test.ts b/examples/reference-host/test/acceptance.test.ts
index f6ed867..a9ab222 100644
--- a/examples/reference-host/test/acceptance.test.ts
+++ b/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" }),
     );
   });
 
diff --git a/packages/mailbox/README.md b/packages/mailbox/README.md
index 2eaa133..7543abf 100644
--- a/packages/mailbox/README.md
+++ b/packages/mailbox/README.md
@@ -114,10 +114,18 @@ 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
@@ -125,6 +133,20 @@ reading is **disconnected**, not buffered for. So the contract for any client:
 - 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:
 
diff --git a/packages/mailbox/src/bus.test.ts b/packages/mailbox/src/bus.test.ts
new file mode 100644
index 0000000..1d8885b
--- /dev/null
+++ b/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" }]);
+  });
+});
diff --git a/packages/mailbox/src/bus.ts b/packages/mailbox/src/bus.ts
index 21f8fd0..019cc72 100644
--- a/packages/mailbox/src/bus.ts
+++ b/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,
diff --git a/packages/mailbox/src/index.ts b/packages/mailbox/src/index.ts
index 2f3aafb..1ac31fc 100644
--- a/packages/mailbox/src/index.ts
+++ b/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 {
diff --git a/packages/mailbox/src/mount-event-op.test.ts b/packages/mailbox/src/mount-event-op.test.ts
new file mode 100644
index 0000000..ddb7733
--- /dev/null
+++ b/packages/mailbox/src/mount-event-op.test.ts
@@ -0,0 +1,153 @@
+// CL-5018: the live event must name the operation that fired so a listener
+// can react without re-fetching and diffing against remembered state.
+import { beforeEach, describe, expect, test } from "bun:test";
+import { Hono } from "hono";
+import { mountMailbox } from "./mount.js";
+import {
+  createInMemoryMailboxEventBus,
+  type MailboxEvent,
+  type MailboxEventOp,
+} from "./bus.js";
+import { writeMailboxMessage } from "./write.js";
+import { withTestDb, seedScope, TEST_VOCABULARY } from "./test-helpers.js";
+import type { MailboxDb } from "./db.js";
+
+let db: MailboxDb;
+const P = { tenantId: "t1", principalId: "p1" };
+
+beforeEach(async () => {
+  db = await withTestDb();
+  await seedScope(db, P.tenantId, P.principalId);
+});
+
+function buildApp(bus: ReturnType<typeof createInMemoryMailboxEventBus>) {
+  const app = new Hono();
+  mountMailbox(app, {
+    db,
+    bus,
+    resolvePrincipal: () => P,
+    vocabulary: TEST_VOCABULARY,
+  });
+  return app;
+}
+
+async function seedMessage(messageKey: string): Promise<string> {
+  const written = await writeMailboxMessage(db, {
+    ...P,
+    address: "p1@t1.example",
+    fromAddress: "a@t1.example",
+    subject: "Hi",
+    body: "Body",
+    messageKey,
+  });
+  return written!.id;
+}
+
+describe("single-message mutations publish their op", () => {
+  const cases: Array<{ verb: string; op: MailboxEventOp; seedKey: string }> = [
+    { verb: "read", op: "mark_read", seedKey: "op-read" },
+    { verb: "unread", op: "mark_unread", seedKey: "op-unread" },
+    { verb: "trash", op: "trash", seedKey: "op-trash" },
+    { verb: "archive", op: "archive", seedKey: "op-archive" },
+    { verb: "restore", op: "restore", seedKey: "op-restore" },
+  ];
+
+  for (const { verb, op, seedKey } of cases) {
+    test(`POST .../${verb} publishes op "${op}"`, async () => {
+      const id = await seedMessage(seedKey);
+      const bus = createInMemoryMailboxEventBus();
+      const received: MailboxEvent[] = [];
+      bus.subscribe(P, (e) => received.push(e));
+      const app = buildApp(bus);
+
+      const res = await app.request(`/me/inbox/${id}/${verb}`, {
+        method: "POST",
+      });
+
+      expect(res.status).toBe(200);
+      expect(received).toEqual([{ type: "mailbox", id, op }]);
+    });
+  }
+});
+
+describe("enrich and assign publish their own op", () => {
+  test("enrich publishes op 'enrich'", async () => {
+    const id = await seedMessage("op-enrich");
+    const bus = createInMemoryMailboxEventBus();
+    const received: MailboxEvent[] = [];
+    bus.subscribe(P, (e) => received.push(e));
+    const app = buildApp(bus);
+
+    const res = await app.request(`/me/inbox/${id}/enrich`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ priority: "urgent" }),
+    });
+
+    expect(res.status).toBe(200);
+    expect(received).toEqual([{ type: "mailbox", id, op: "enrich" }]);
+  });
+
+  test("assign publishes op 'assign'", async () => {
+    const id = await seedMessage("op-assign");
+    const bus = createInMemoryMailboxEventBus();
+    const received: MailboxEvent[] = [];
+    bus.subscribe(P, (e) => received.push(e));
+    const app = buildApp(bus);
+
+    const res = await app.request(`/me/inbox/${id}/assign`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ assignee: "teammate-1" }),
+    });
+
+    expect(res.status).toBe(200);
+    expect(received).toEqual([{ type: "mailbox", id, op: "assign" }]);
+  });
+});
+
+describe("bulk mutation publishes the requested action as op", () => {
+  test("bulk trash publishes op 'trash' for every updated id", async () => {
+    const idA = await seedMessage("op-bulk-a");
+    const idB = await seedMessage("op-bulk-b");
+    const bus = createInMemoryMailboxEventBus();
+    const received: MailboxEvent[] = [];
+    bus.subscribe(P, (e) => received.push(e));
+    const app = buildApp(bus);
+
+    const res = await app.request("/me/inbox/bulk", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ action: "trash", ids: [idA, idB] }),
+    });
+
+    expect(res.status).toBe(200);
+    expect(received).toHaveLength(2);
+    expect(received.every((e) => e.op === "trash")).toBe(true);
+    expect(new Set(received.map((e) => e.id))).toEqual(new Set([idA, idB]));
+  });
+});
+
+describe("delivery publishes op 'create'", () => {
+  test("writeMailboxMessage against the same bus a mount would use", async () => {
+    const bus = createInMemoryMailboxEventBus();
+    const received: MailboxEvent[] = [];
+    bus.subscribe(P, (e) => received.push(e));
+
+    await writeMailboxMessage(
+      db,
+      {
+        ...P,
+        address: "p1@t1.example",
+        fromAddress: "a@t1.example",
+        subject: "New",
+        body: "Body",
+        messageKey: "op-create-write",
+      },
+      bus,
+    );
+
+    expect(received).toHaveLength(1);
+    expect(received[0]?.op).toBe("create");
+  });
+});
diff --git a/packages/mailbox/src/mount.ts b/packages/mailbox/src/mount.ts
index fdfe5b7..60a364a 100644
--- a/packages/mailbox/src/mount.ts
+++ b/packages/mailbox/src/mount.ts
@@ -8,6 +8,7 @@ import {
   publishMailboxEvent,
   type MailboxEvent,
   type MailboxEventBus,
+  type MailboxEventOp,
 } from "./bus.js";
 import {
   countUnreadActiveMailbox,
@@ -210,24 +211,40 @@ const ID_PARAM = {
 };
 
 // The five single-message mutations differ only in verb and handler, so they
-// are registered from one table instead of five near-identical blocks.
+// are registered from one table instead of five near-identical blocks. `op`
+// is the event op published on success — same identifiers `applyMailboxBulkAction`
+// takes as its `action`, so a listener sees the same op for "read one" and
+// "read fifty".
 const SINGLE_MUTATIONS = [
-  { verb: "read", summary: "Mark a message read", run: markMailboxMessageRead },
+  {
+    verb: "read",
+    summary: "Mark a message read",
+    run: markMailboxMessageRead,
+    op: "mark_read",
+  },
   {
     verb: "unread",
     summary: "Mark an active message unread",
     run: markMailboxMessageUnread,
+    op: "mark_unread",
+  },
+  {
+    verb: "trash",
+    summary: "Trash a message",
+    run: trashMailboxMessage,
+    op: "trash",
   },
-  { verb: "trash", summary: "Trash a message", run: trashMailboxMessage },
   {
     verb: "archive",
     summary: "Archive a message (refused once trashed)",
     run: archiveMailboxMessage,
+    op: "archive",
   },
   {
     verb: "restore",
     summary: "Restore a message out of archive or trash",
     run: restoreMailboxMessage,
+    op: "restore",
   },
 ] as const;
 
@@ -262,8 +279,12 @@ export function mountMailbox<E extends Env>(
     );
   }
 
-  function publish(scope: ResolvedPrincipal, id: string): void {
-    publishMailboxEvent(bus, scope, id, logger);
+  function publish(
+    scope: ResolvedPrincipal,
+    id: string,
+    op: MailboxEventOp,
+  ): void {
+    publishMailboxEvent(bus, scope, id, logger, op);
   }
 
   app.get(
@@ -432,7 +453,9 @@ export function mountMailbox<E extends Env>(
       tags: TAGS,
       summary: "Server-sent stream of mailbox events for the caller",
       description:
-        "Emits a `mailbox` event per affected message id, plus a heartbeat comment every 25s.",
+        "Emits a `mailbox` event per affected message id, plus a heartbeat comment every 25s. " +
+        "Each event also carries `op` (create/mark_read/mark_unread/trash/archive/restore/enrich/assign) " +
+        "when the publisher knows it — optional and additive, so a listener reading only `id` still works.",
       responses: {
         200: { description: "text/event-stream" },
         403: { description: "No resolvable principalId" },
@@ -554,6 +577,7 @@ export function mountMailbox<E extends Env>(
       principalId: string;
       id: string;
     }) => Promise<boolean>,
+    op: MailboxEventOp,
   ) {
     if (!isUuid(id)) {
       return c.json({ error: "Message id must be a UUID" }, 400);
@@ -564,11 +588,11 @@ export function mountMailbox<E extends Env>(
     }
     const ok = await run({ ...resolved, id });
     if (!ok) return c.json({ error: "Message not found" }, 404);
-    publish(resolved, id);
+    publish(resolved, id, op);
     return c.json({ id, ok: true as const });
   }
 
-  for (const { verb, summary, run } of SINGLE_MUTATIONS) {
+  for (const { verb, summary, run, op } of SINGLE_MUTATIONS) {
     app.post(
       `/me/inbox/:id/${verb}`,
       describeRoute({
@@ -582,7 +606,8 @@ export function mountMailbox<E extends Env>(
           404: { description: "No message in scope for this action" },
         },
       }),
-      (c) => singleMutation(c, c.req.param("id"), (scope) => run(db, scope)),
+      (c) =>
+        singleMutation(c, c.req.param("id"), (scope) => run(db, scope), op),
     );
   }
 
@@ -638,7 +663,7 @@ export function mountMailbox<E extends Env>(
         throw err;
       }
       if (!ok) return c.json({ error: "Message not found" }, 404);
-      publish(resolved, id);
+      publish(resolved, id, "enrich");
       return c.json({ id, ok: true as const });
     },
   );
@@ -683,7 +708,7 @@ export function mountMailbox<E extends Env>(
         body.assignee,
       );
       if (!ok) return c.json({ error: "Message not found" }, 404);
-      publish(resolved, id);
+      publish(resolved, id, "assign");
       return c.json({ id, ok: true as const });
     },
   );
@@ -734,7 +759,7 @@ export function mountMailbox<E extends Env>(
         throw err;
       }
       for (const r of results) {
-        if (r.ok) publish(resolved, r.id);
+        if (r.ok) publish(resolved, r.id, body.action);
       }
       return c.json({
         updated: results.filter((r) => r.ok).length,
diff --git a/packages/mailbox/src/persist.test.ts b/packages/mailbox/src/persist.test.ts
index 0e4c91e..dfb3f3b 100644
--- a/packages/mailbox/src/persist.test.ts
+++ b/packages/mailbox/src/persist.test.ts
@@ -247,13 +247,15 @@ describe("dual-write independence", () => {
 });
 
 describe("announcements", () => {
-  test("each inserted row is published to its own principalId", async () => {
+  test("each inserted row is published to its own principalId, as a `create`", async () => {
     const bus = createInMemoryMailboxEventBus();
     const forUser1: string[] = [];
     const forUser2: string[] = [];
-    bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) =>
-      forUser1.push(e.id),
-    );
+    const opsUser1: Array<string | undefined> = [];
+    bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) => {
+      forUser1.push(e.id);
+      opsUser1.push(e.op);
+    });
     bus.subscribe({ tenantId: "acme", principalId: "user-2" }, (e) =>
       forUser2.push(e.id),
     );
@@ -273,6 +275,9 @@ describe("announcements", () => {
     expect(forUser1).toHaveLength(1);
     expect(forUser2).toHaveLength(1);
     expect(forUser1[0]).not.toBe(forUser2[0]);
+    // Dual-write inserts are always new mail, never a mutation of an
+    // existing row, so the op is unconditionally `create`.
+    expect(opsUser1).toEqual(["create"]);
   });
 
   test("onRow reports the row id, principalId and sender", async () => {
diff --git a/packages/mailbox/src/persist.ts b/packages/mailbox/src/persist.ts
index eab4144..a3051ef 100644
--- a/packages/mailbox/src/persist.ts
+++ b/packages/mailbox/src/persist.ts
@@ -126,6 +126,7 @@ export function createMailboxPersist<R>(
         { tenantId: row.tenantId, principalId: row.principalId },
         row.id,
         logger,
+        "create",
       );
     }
     if (!opts.onRow) return;
diff --git a/packages/mailbox/src/write.test.ts b/packages/mailbox/src/write.test.ts
index a4291d7..83e8abf 100644
--- a/packages/mailbox/src/write.test.ts
+++ b/packages/mailbox/src/write.test.ts
@@ -190,9 +190,9 @@ describe("writeMailboxMessage", () => {
 
   test("successful write publishes to the bus", async () => {
     const bus = createInMemoryMailboxEventBus();
-    const received: string[] = [];
+    const received: Array<{ id: string; op?: string }> = [];
     bus.subscribe({ tenantId: "t1", principalId: "p1" }, (event) =>
-      received.push(event.id),
+      received.push(event),
     );
     await writeMailboxMessage(
       db,
@@ -208,6 +208,9 @@ describe("writeMailboxMessage", () => {
       bus,
     );
     expect(received.length).toBe(1);
+    // A new message is a `create` — a listener can tell delivery apart from
+    // a mutation without re-fetching and diffing.
+    expect(received[0]?.op).toBe("create");
   });
 });
 
@@ -495,6 +498,32 @@ describe("deliverInboxItems", () => {
     expect(enqueued).toEqual([]);
   });
 
+  test("a newly delivered item publishes a `create` event", async () => {
+    const bus = createInMemoryMailboxEventBus();
+    const received: Array<{ id: string; op?: string }> = [];
+    bus.subscribe({ tenantId: "t1", principalId: "p1" }, (event) =>
+      received.push(event),
+    );
+    await deliverInboxItems(
+      db,
+      [
+        {
+          tenantId: "t1",
+          principalId: "p1",
+          address: "p1@t1.example",
+          fromAddress: "sender@ext.example",
+          subject: "Delivered",
+          body: "Body",
+          source: "gmail",
+          externalId: "op-create",
+        },
+      ],
+      { bus },
+    );
+    expect(received).toHaveLength(1);
+    expect(received[0]?.op).toBe("create");
+  });
+
   test("bus publish and enqueue run only after a successful batch commit", async () => {
     const bus = createInMemoryMailboxEventBus();
     const received: string[] = [];
diff --git a/packages/mailbox/src/write.ts b/packages/mailbox/src/write.ts
index 63bd248..3f98e4f 100644
--- a/packages/mailbox/src/write.ts
+++ b/packages/mailbox/src/write.ts
@@ -271,6 +271,7 @@ export async function writeMailboxMessage(
       { tenantId: args.tenantId, principalId: args.principalId },
       row.id,
       logger,
+      "create",
     );
   }
   return { id: row.id };
@@ -413,6 +414,7 @@ export async function deliverInboxItems(
         { tenantId: item.tenantId, principalId: item.principalId },
         id,
         logger,
+        "create",
       );
     }
     if (!opts?.enqueue) continue;