fix(daemon): drain transaction events to quiescence

packages/daemon/src/domain/event-bus.ts

@@ -1,10 +1,38 @@
 import type Database from "better-sqlite3";
+import { createHash } from "node:crypto";
 import type { RigEvent, PersistedEvent } from "./types.js";
 
 type Subscriber = (event: PersistedEvent) => void;
 
+declare const notifyTokenBrand: unique symbol;
+export type NotifyToken = PersistedEvent & { readonly [notifyTokenBrand]: true };
+export type NotifyRegister = (token: PersistedEvent) => void;
+
+export interface NotifyDrainStatus {
+  state: "healthy" | "unparseable";
+  watermark: number;
+  lastPoison: { seq: number; error: string; payloadSha: string } | null;
+}
+
+export class NotifyEnvelopeError extends Error {
+  readonly code: "notify_registration_mismatch" | "nested_notify_envelope";
+
+  constructor(code: NotifyEnvelopeError["code"], message: string) {
+    super(message);
+    this.code = code;
+  }
+}
+
+interface ActiveEnvelope {
+  persisted: Set<PersistedEvent>;
+}
+
 export class EventBus {
   private subscribers = new Set<Subscriber>();
+  private activeEnvelope: ActiveEnvelope | null = null;
+  private drainingNotifyRows = false;
+  private notifyEnvelopeRuns = 0;
+  private notifyDrainStatus: NotifyDrainStatus;
   readonly db: Database.Database;
 
   get subscriberCount(): number {
@@ -13,6 +41,11 @@ export class EventBus {
 
   constructor(db: Database.Database) {
     this.db = db;
+    this.notifyDrainStatus = {
+      state: "healthy",
+      watermark: this.maxSeq(),
+      lastPoison: null,
+    };
   }
 
   /**
@@ -29,9 +62,10 @@ export class EventBus {
    * Insert an event row into the events table and return a PersistedEvent.
    * Call this inside a caller-managed db.transaction() so the event insert
    * is atomic with other writes (e.g., session + binding + event in one txn).
-   * Does NOT notify subscribers — call notifySubscribers() after commit.
+   * Does NOT notify subscribers. A notify envelope drains after commit;
+   * legacy single-event callers retain their explicit post-commit notify.
    */
-  persistWithinTransaction(event: RigEvent): PersistedEvent {
+  persistWithinTransaction(event: RigEvent): NotifyToken {
     const rigId = "rigId" in event ? (event as { rigId: string }).rigId : null;
     const nodeId = "nodeId" in event ? event.nodeId : null;
 
@@ -47,11 +81,70 @@ export class EventBus {
       .prepare("SELECT created_at FROM events WHERE seq = ?")
       .get(seq) as { created_at: string };
 
-    return {
+    const token = {
       ...event,
       seq,
       createdAt: row.created_at,
+    } as NotifyToken;
+    this.activeEnvelope?.persisted.add(token);
+    return token;
+  }
+
+  /**
+   * Run one caller-owned synchronous transaction under the W2b notification
+   * envelope. Every token persisted through EventBus during the callback must
+   * be registered before it returns. The exact object-identity sets are
+   * compared before commit; committed rows are then delivered from the log.
+   */
+  withNotifyEnvelope<T>(callback: (register: NotifyRegister) => T): T {
+    if (this.activeEnvelope) {
+      throw new NotifyEnvelopeError(
+        "nested_notify_envelope",
+        "notify envelope cannot be nested",
+      );
+    }
+
+    const envelope: ActiveEnvelope = { persisted: new Set() };
+    const registered = new Set<PersistedEvent>();
+    const register: NotifyRegister = (token) => {
+      registered.add(token);
     };
+    this.notifyEnvelopeRuns += 1;
+
+    const txn = this.db.transaction(() => {
+      this.activeEnvelope = envelope;
+      try {
+        const result = callback(register);
+        if (!setsEqual(envelope.persisted, registered)) {
+          throw new NotifyEnvelopeError(
+            "notify_registration_mismatch",
+            `notify envelope persisted ${envelope.persisted.size} token(s) but registered ${registered.size} exact token(s)`,
+          );
+        }
+        return result;
+      } finally {
+        this.activeEnvelope = null;
+      }
+    });
+
+    const result = txn();
+    this.drainNotifyRowsToQuiescence(this.notifyDrainStatus.watermark);
+    return result;
+  }
+
+  getNotifyDrainStatus(): NotifyDrainStatus {
+    return {
+      ...this.notifyDrainStatus,
+      lastPoison: this.notifyDrainStatus.lastPoison
+        ? { ...this.notifyDrainStatus.lastPoison }
+        : null,
+    };
+  }
+
+  assertNotifyEnvelopeExercised(): void {
+    if (this.notifyEnvelopeRuns === 0) {
+      throw new Error("zero notify envelope transactions examined");
+    }
   }
 
   /**
@@ -59,6 +152,7 @@ export class EventBus {
    * Does NOT insert into DB. Subscriber errors are isolated.
    */
   notifySubscribers(event: PersistedEvent): void {
+    const nestedRowsStartAfter = this.maxSeq();
     for (const subscriber of this.subscribers) {
       try {
         subscriber(event);
@@ -66,6 +160,10 @@ export class EventBus {
         console.error("EventBus subscriber error:", err);
       }
     }
+    this.notifyDrainStatus.watermark = Math.max(this.notifyDrainStatus.watermark, event.seq);
+    if (!this.drainingNotifyRows) {
+      this.drainNotifyRowsToQuiescence(nestedRowsStartAfter);
+    }
   }
 
   subscribe(cb: Subscriber): () => void {
@@ -86,17 +184,78 @@ export class EventBus {
   }
 
   replayAll(seq: number): PersistedEvent[] {
-    const rows = this.db
-      .prepare(
-        "SELECT seq, rig_id, node_id, type, payload, created_at FROM events WHERE seq > ? ORDER BY seq"
-      )
-      .all(seq) as EventRow[];
+    const rows = this.rowsAfter(seq);
 
     return rows.map((row) => this.rowToPersistedEvent(row));
   }
 
+  private drainNotifyRowsToQuiescence(startAfter: number): void {
+    if (this.drainingNotifyRows) return;
+    this.drainingNotifyRows = true;
+    let cursor = startAfter;
+    try {
+      for (;;) {
+        const rows = this.rowsAfter(cursor);
+        if (rows.length === 0) return;
+        for (const row of rows) {
+          let event: PersistedEvent;
+          try {
+            event = this.rowToPersistedEvent(row);
+          } catch (caught) {
+            const error = caught instanceof SyntaxError
+              ? "invalid event payload JSON"
+              : "invalid event payload shape";
+            const payloadSha = createHash("sha256").update(row.payload).digest("hex");
+            this.notifyDrainStatus = {
+              state: "unparseable",
+              watermark: row.seq,
+              lastPoison: { seq: row.seq, error, payloadSha },
+            };
+            this.persistWithinTransaction({
+              type: "event.delivery_poisoned",
+              poisonedSeq: row.seq,
+              error,
+              payloadSha,
+            });
+            cursor = row.seq;
+            continue;
+          }
+          this.notifySubscribers(event);
+          cursor = row.seq;
+        }
+      }
+    } finally {
+      this.drainingNotifyRows = false;
+    }
+  }
+
+  private maxSeq(): number {
+    const row = this.db.prepare("SELECT MAX(seq) AS seq FROM events").get() as {
+      seq: number | null;
+    };
+    return row.seq ?? 0;
+  }
+
+  private rowsAfter(seq: number): EventRow[] {
+    return this.db
+      .prepare(
+        "SELECT seq, rig_id, node_id, type, payload, created_at FROM events WHERE seq > ? ORDER BY seq",
+      )
+      .all(seq) as EventRow[];
+  }
+
   private rowToPersistedEvent(row: EventRow): PersistedEvent {
-    const event = JSON.parse(row.payload) as RigEvent;
+    const parsed = JSON.parse(row.payload) as unknown;
+    if (
+      typeof parsed !== "object" ||
+      parsed === null ||
+      Array.isArray(parsed) ||
+      typeof (parsed as { type?: unknown }).type !== "string" ||
+      (parsed as { type: string }).type.length === 0
+    ) {
+      throw new Error("invalid event payload shape");
+    }
+    const event = parsed as RigEvent;
     return {
       ...event,
       seq: row.seq,
@@ -107,9 +266,17 @@ export class EventBus {
 
 interface EventRow {
   seq: number;
-  rig_id: string;
+  rig_id: string | null;
   node_id: string | null;
   type: string;
   payload: string;
   created_at: string;
 }
+
+function setsEqual<T>(left: ReadonlySet<T>, right: ReadonlySet<T>): boolean {
+  if (left.size !== right.size) return false;
+  for (const value of left) {
+    if (!right.has(value)) return false;
+  }
+  return true;
+}

packages/daemon/src/domain/mission-control/mission-control-write-contract.ts

@@ -18,10 +18,10 @@
 //      after state snapshots.
 //   5. Persist the mission_control.action_executed event.
 //
-// Post-commit (outside the transaction): notifySubscribers + opt-in
-// best-effort transport notify. Notify failure does NOT roll back
-// durable mutations (PRD invariant: "notify failure does NOT roll
-// back durable mutations").
+// Post-commit (outside the transaction): the event envelope drains to
+// subscribers, then opt-in best-effort transport notify runs. Notify failure
+// does NOT roll back durable mutations (PRD invariant: "notify failure does
+// NOT roll back durable mutations").
 //
 // Verb mappings:
 //   approve   → state="done",        closure_reason="no-follow-on"
@@ -153,99 +153,95 @@ export class MissionControlWriteContract {
     let createdDestination: string | undefined;
     let createdNudge: boolean | undefined;
     let actionEntry: ReturnType<MissionControlActionLog["record"]> | null = null;
-    const persistedEvents: PersistedEvent[] = [];
-
-    const txn = this.db.transaction(() => {
-      // 1. Close/transition the source via Phase A's queue closure primitive.
-      const closeResult = this.queueRepo.updateWithinTransaction({
-        qitemId: input.qitemId,
-        actorSession: input.actorSession,
-        state: closure.state,
-        closureReason: closure.closureReason,
-        closureTarget: closure.closureTarget ?? undefined,
-        handedOffTo: closure.handedOffTo,
-        blockedOn: closure.blockedOn,
-        transitionNote: `mission-control:${input.verb}${input.reason ? ` (${input.reason})` : ""}`,
-      });
-      persistedEvents.push(closeResult.persistedEvent);
-
-      // 2. For route/handoff: create the destination packet in same txn.
-      if ((input.verb === "route" || input.verb === "handoff") && input.destinationSession) {
-        const created = this.queueRepo.createWithinTransaction({
-          sourceSession: input.actorSession,
-          destinationSession: input.destinationSession,
-          body: input.body ?? source.body,
-          priority: source.priority,
-          tier: source.tier ?? undefined,
-          summary: source.summary,
-          evidenceRef: source.evidenceRef,
-          tags: source.tags
-            ? [...source.tags, `mission-control:${input.verb}`]
-            : [`mission-control:${input.verb}`],
-          chainOfRecord: [...(source.chainOfRecord ?? []), input.qitemId],
-          // Default nudge handled post-commit per Phase D pattern.
-          nudge: input.notify,
+    try {
+      this.eventBus.withNotifyEnvelope((register) => {
+        // 1. Close/transition the source via Phase A's queue closure primitive.
+        const closeResult = this.queueRepo.updateWithinTransaction({
+          qitemId: input.qitemId,
+          actorSession: input.actorSession,
+          state: closure.state,
+          closureReason: closure.closureReason,
+          closureTarget: closure.closureTarget ?? undefined,
+          handedOffTo: closure.handedOffTo,
+          blockedOn: closure.blockedOn,
+          transitionNote: `mission-control:${input.verb}${input.reason ? ` (${input.reason})` : ""}`,
         });
-        createdQitemId = created.qitemId;
-        createdDestination = created.destinationSession;
-        createdNudge = created.nudge;
-        persistedEvents.push(created.persistedEvent);
-        // P34: stage the successor's WAKE INTENT inside this transaction, so the
-        // close and the durable wake commit as ONE act or NONE. The pane write
-        // itself stays post-commit (reversed-never).
-        this.queueRepo.stageWakeIntent(
-          created.qitemId,
-          input.actorSession,
-          created.destinationSession,
-          input.identityProvenance ?? null,
-          created.nudge,
-        );
-      }
+        register(closeResult.persistedEvent);
 
-      // 3. Append the audit record. Snapshot the closed qitem state.
-      const closedQitem = this.queueRepo.getById(input.qitemId);
-      const afterSnapshot = closedQitem ? snapshotQitem(closedQitem) : null;
-      actionEntry = this.actionLog.record({
-        actionVerb: input.verb,
-        qitemId: input.qitemId,
-        actorSession: input.actorSession,
-        actedAt: evaluatedAt,
-        beforeState: beforeSnapshot,
-        afterState: afterSnapshot,
-        reason: input.reason ?? null,
-        annotation: input.annotation ?? null,
-        notifyAttempted: false,
-        notifyResult: null,
-        auditNotes: input.auditNotes ?? null,
-        identityProvenance: input.identityProvenance ?? null, // P21 era-stamp on the audit row
-      });
+        // 2. For route/handoff: create the destination packet in same txn.
+        if ((input.verb === "route" || input.verb === "handoff") && input.destinationSession) {
+          const created = this.queueRepo.createWithinTransaction({
+            sourceSession: input.actorSession,
+            destinationSession: input.destinationSession,
+            body: input.body ?? source.body,
+            priority: source.priority,
+            tier: source.tier ?? undefined,
+            summary: source.summary,
+            evidenceRef: source.evidenceRef,
+            tags: source.tags
+              ? [...source.tags, `mission-control:${input.verb}`]
+              : [`mission-control:${input.verb}`],
+            chainOfRecord: [...(source.chainOfRecord ?? []), input.qitemId],
+            // Default nudge handled post-commit per Phase D pattern.
+            nudge: input.notify,
+          });
+          createdQitemId = created.qitemId;
+          createdDestination = created.destinationSession;
+          createdNudge = created.nudge;
+          register(created.persistedEvent);
+          // P34: stage the successor's WAKE INTENT inside this transaction, so the
+          // close and the durable wake commit as ONE act or NONE. The pane write
+          // itself stays post-commit (reversed-never).
+          this.queueRepo.stageWakeIntent(
+            created.qitemId,
+            input.actorSession,
+            created.destinationSession,
+            input.identityProvenance ?? null,
+            created.nudge,
+          );
+        }
 
-      // 4. Persist the mission_control.action_executed event in same txn.
-      persistedEvents.push(
-        this.eventBus.persistWithinTransaction({
-          type: "mission_control.action_executed",
-          actionId: actionEntry.actionId,
+        // 3. Append the audit record. Snapshot the closed qitem state.
+        const closedQitem = this.queueRepo.getById(input.qitemId);
+        const afterSnapshot = closedQitem ? snapshotQitem(closedQitem) : null;
+        actionEntry = this.actionLog.record({
           actionVerb: input.verb,
           qitemId: input.qitemId,
           actorSession: input.actorSession,
-        }),
-      );
+          actedAt: evaluatedAt,
+          beforeState: beforeSnapshot,
+          afterState: afterSnapshot,
+          reason: input.reason ?? null,
+          annotation: input.annotation ?? null,
+          notifyAttempted: false,
+          notifyResult: null,
+          auditNotes: input.auditNotes ?? null,
+          identityProvenance: input.identityProvenance ?? null, // P21 era-stamp on the audit row
+        });
 
-      // P34: the W1 seam, run as the LAST statement of this transaction. If this
-      // act terminally closed the source AND created a successor, the successor's
-      // wake intent must be durable in the SAME transaction — otherwise the whole
-      // act rolls back here rather than committing an executed-but-unwoken item.
-      // A non-terminal transition (the `hold` park) returns early inside the
-      // primitive itself (queue-repository.ts, isTerminalState), so park callers
-      // are untouched by construction; a terminal close with NO successor has
-      // nothing to wake and is not paired here.
-      if (createdQitemId && createdDestination) {
-        this.queueRepo.assertTerminalClosureHasIntent(input.qitemId, createdQitemId, createdNudge);
-      }
-    });
+        // 4. Persist the mission_control.action_executed event in same txn.
+        register(
+          this.eventBus.persistWithinTransaction({
+            type: "mission_control.action_executed",
+            actionId: actionEntry.actionId,
+            actionVerb: input.verb,
+            qitemId: input.qitemId,
+            actorSession: input.actorSession,
+          }),
+        );
 
-    try {
-      txn();
+        // P34: the W1 seam, run as the LAST statement of this transaction. If this
+        // act terminally closed the source AND created a successor, the successor's
+        // wake intent must be durable in the SAME transaction — otherwise the whole
+        // act rolls back here rather than committing an executed-but-unwoken item.
+        // A non-terminal transition (the `hold` park) returns early inside the
+        // primitive itself (queue-repository.ts, isTerminalState), so park callers
+        // are untouched by construction; a terminal close with NO successor has
+        // nothing to wake and is not paired here.
+        if (createdQitemId && createdDestination) {
+          this.queueRepo.assertTerminalClosureHasIntent(input.qitemId, createdQitemId, createdNudge);
+        }
+      });
     } catch (err) {
       if (err instanceof MissionControlActionLogError) {
         throw new MissionControlWriteContractError(err.code, err.message, err.details);
@@ -253,9 +249,6 @@ export class MissionControlWriteContract {
       throw err;
     }
 
-    // Post-commit: fan out events.
-    for (const e of persistedEvents) this.eventBus.notifySubscribers(e);
-
     // Post-commit best-effort notify on handoff/route. Default true per
     // PL-004 R1 pattern; failure does NOT roll back durable mutations.
     let notifyAttempted = false;
@@ -403,49 +396,45 @@ export class MissionControlWriteContract {
     const evaluatedAt = this.now().toISOString();
     const beforeSnapshot = snapshotQitem(source);
     let actionEntry: ReturnType<MissionControlActionLog["record"]> | null = null;
-    const persistedEvents: PersistedEvent[] = [];
+    try {
+      this.eventBus.withNotifyEnvelope((register) => {
+        // Rail 2+3: blocked → in-progress; decision text = transition_note;
+        // actor_session = the resolving human/relay session. Emits
+        // queue.updated (the P2 refresh-after-action contract). blocked_on is
+        // deliberately retained as provenance of whom it was parked on — the
+        // attention query keys on state='blocked', so the resolved item drops
+        // out of attention regardless.
+        const updateResult = this.queueRepo.updateWithinTransaction({
+          qitemId: input.qitemId,
+          actorSession: input.actorSession,
+          state: "in-progress",
+          transitionNote: decision,
+        });
+        register(updateResult.persistedEvent);
 
-    const txn = this.db.transaction(() => {
-      // Rail 2+3: blocked → in-progress; decision text = transition_note;
-      // actor_session = the resolving human/relay session. Emits
-      // queue.updated (the P2 refresh-after-action contract). blocked_on is
-      // deliberately retained as provenance of whom it was parked on — the
-      // attention query keys on state='blocked', so the resolved item drops
-      // out of attention regardless.
-      const updateResult = this.queueRepo.updateWithinTransaction({
-        qitemId: input.qitemId,
-        actorSession: input.actorSession,
-        state: "in-progress",
-        transitionNote: decision,
-      });
-      persistedEvents.push(updateResult.persistedEvent);
-
-      const resolvedQitem = this.queueRepo.getById(input.qitemId);
-      actionEntry = this.actionLog.record({
-        actionVerb: "resolve",
-        qitemId: input.qitemId,
-        actorSession: input.actorSession,
-        actedAt: evaluatedAt,
-        beforeState: beforeSnapshot,
-        afterState: resolvedQitem ? snapshotQitem(resolvedQitem) : null,
-        reason: decision,
-        auditNotes: input.auditNotes ?? null,
-        identityProvenance: input.identityProvenance ?? null, // P21 era-stamp on the audit row
-      });
-
-      persistedEvents.push(
-        this.eventBus.persistWithinTransaction({
-          type: "mission_control.action_executed",
-          actionId: actionEntry.actionId,
+        const resolvedQitem = this.queueRepo.getById(input.qitemId);
+        actionEntry = this.actionLog.record({
           actionVerb: "resolve",
           qitemId: input.qitemId,
           actorSession: input.actorSession,
-        }),
-      );
-    });
+          actedAt: evaluatedAt,
+          beforeState: beforeSnapshot,
+          afterState: resolvedQitem ? snapshotQitem(resolvedQitem) : null,
+          reason: decision,
+          auditNotes: input.auditNotes ?? null,
+          identityProvenance: input.identityProvenance ?? null, // P21 era-stamp on the audit row
+        });
 
-    try {
-      txn();
+        register(
+          this.eventBus.persistWithinTransaction({
+            type: "mission_control.action_executed",
+            actionId: actionEntry.actionId,
+            actionVerb: "resolve",
+            qitemId: input.qitemId,
+            actorSession: input.actorSession,
+          }),
+        );
+      });
     } catch (err) {
       if (err instanceof MissionControlActionLogError) {
         throw new MissionControlWriteContractError(err.code, err.message, err.details);
@@ -453,8 +442,6 @@ export class MissionControlWriteContract {
       throw err;
     }
 
-    for (const e of persistedEvents) this.eventBus.notifySubscribers(e);
-
     // Rail 3 + BR-8: best-effort nudge to the PARKED OWNER carrying the
     // decision text, after the commit — nudge outcome recorded via the
     // existing last_nudge_* mechanics; a failed nudge never unwinds the

packages/daemon/src/domain/types.ts

@@ -101,6 +101,7 @@ export interface Session {
 // -- Event types --
 
 export type RigEvent =
+  | { type: "event.delivery_poisoned"; poisonedSeq: number; error: string; payloadSha: string }
   | { type: "rig.created"; rigId: string }
   | { type: "rig.deleted"; rigId: string }
   | { type: "node.added"; rigId: string; nodeId: string; logicalId: string }

packages/daemon/src/domain/view-projector.ts

@@ -327,12 +327,11 @@ export class ViewProjector {
    * subscribe to event-bus events and trigger this notify.
    */
   notifyViewChanged(viewName: string, cause: string): void {
-    const persisted = this.eventBus.persistWithinTransaction({
+    this.eventBus.persistWithinTransaction({
       type: "view.changed",
       viewName,
       cause,
     });
-    this.eventBus.notifySubscribers(persisted);
   }
 
   private getCustomViewByIdOrThrow(viewId: string): CustomView {

packages/daemon/src/domain/workflow-projector.ts

@@ -355,8 +355,11 @@ export class WorkflowProjector {
     // (delivery is best-effort; the durable item is the guarantee).
     let exceptionItemPostCommit: (PostCommitNudge & { qitemId: string }) | null = null;
     const persistedEvents: PersistedEvent[] = [];
-
-    const txn = this.db.transaction(() => {
+    this.eventBus.withNotifyEnvelope((register) => {
+      const registerEvent = (event: PersistedEvent): void => {
+        persistedEvents.push(event);
+        register(event);
+      };
       // 1. Resolve next owner BEFORE closing prior so closure_target is set.
       // OPR.0.4.6.WF2: owner resolution runs for ANY route (structural
       // handoff or branch-mapped exit). A gated target step compiles to
@@ -413,7 +416,7 @@ export class WorkflowProjector {
         blockedOn: closure.blockedOn,
         transitionNote: closure.transitionNote,
       });
-      persistedEvents.push(queueUpdate.persistedEvent);
+      registerEvent(queueUpdate.persistedEvent);
 
       // 3. If this close routes (structural handoff OR branch-mapped
       // exit): create the next-step qitem in the same txn. A gated
@@ -520,7 +523,7 @@ export class WorkflowProjector {
             blockedOn: gateCompile.parkOn,
             transitionNote: `workflow gate: parked on ${gateCompile.parkOn} pending sign-off`,
           });
-          persistedEvents.push(parked.persistedEvent);
+          registerEvent(parked.persistedEvent);
         }
       }
 
@@ -659,7 +662,7 @@ export class WorkflowProjector {
       }
 
       // 6. Persist workflow events within the same txn.
-      persistedEvents.push(
+      registerEvent(
         this.eventBus.persistWithinTransaction({
           type: "workflow.step_closed",
           instanceId: instance.instanceId,
@@ -670,8 +673,8 @@ export class WorkflowProjector {
         }),
       );
       if (createdNext && nextStep && resolvedNextOwner) {
-        persistedEvents.push(createdNext.persistedEvent);
-        persistedEvents.push(
+        registerEvent(createdNext.persistedEvent);
+        registerEvent(
           this.eventBus.persistWithinTransaction({
             type: "workflow.next_qitem_projected",
             instanceId: instance.instanceId,
@@ -682,7 +685,7 @@ export class WorkflowProjector {
         );
       }
       if (nextStatus === "completed") {
-        persistedEvents.push(
+        registerEvent(
           this.eventBus.persistWithinTransaction({
             type: "workflow.completed",
             instanceId: instance.instanceId,
@@ -690,7 +693,7 @@ export class WorkflowProjector {
           }),
         );
       } else if (nextStatus === "failed") {
-        persistedEvents.push(
+        registerEvent(
           this.eventBus.persistWithinTransaction({
             type: "workflow.failed",
             instanceId: instance.instanceId,
@@ -765,7 +768,7 @@ export class WorkflowProjector {
               "human-gate",
             );
           }
-          persistedEvents.push(createdException.persistedEvent);
+          registerEvent(createdException.persistedEvent);
           exceptionItemPostCommit = {
             qitemId: createdException.qitemId,
             destinationSession: createdException.destinationSession,
@@ -808,12 +811,6 @@ export class WorkflowProjector {
       }
     });
 
-    txn();
-
-    // Post-commit fan-out: notify subscribers + nudge next owner.
-    for (const e of persistedEvents) {
-      this.eventBus.notifySubscribers(e);
-    }
     // OPR.0.4.6.WF5 FR-2: best-effort nudge for the exception item (the
     // shipped non-fatal delivery pattern — the durable item is the
     // guarantee, the nudge is a courtesy).

packages/daemon/src/domain/workflow-runtime.ts

@@ -12,7 +12,6 @@
 import type Database from "better-sqlite3";
 import type { EventBus } from "./event-bus.js";
 import type { QueueRepository } from "./queue-repository.js";
-import type { PersistedEvent } from "./types.js";
 import {
   type CreateWorkflowInstanceInput,
   WorkflowInstanceStore,
@@ -535,9 +534,7 @@ export class WorkflowRuntime {
     let entryQitemDestinationSession: string | undefined;
     let entryQitemNudge: boolean | undefined;
     let instanceId: string | undefined;
-    const persistedEvents: PersistedEvent[] = [];
-
-    const txn = this.db.transaction(() => {
+    this.eventBus.withNotifyEnvelope((register) => {
       const instance = this.instanceStore.create({
         workflowName: specRow.name,
         workflowVersion: specRow.version,
@@ -598,7 +595,7 @@ export class WorkflowRuntime {
       entryQitemId = created.qitemId;
       entryQitemDestinationSession = created.destinationSession;
       entryQitemNudge = created.nudge;
-      persistedEvents.push(created.persistedEvent);
+      register(created.persistedEvent);
 
       // OPR.0.4.6.WF2 FR-5 (guard blocker 1): a HUMAN-gated ENTRY parks
       // in the same txn — the leg-1 blocked_on human-seat shape the
@@ -614,7 +611,7 @@ export class WorkflowRuntime {
           blockedOn: entryGate.parkOn,
           transitionNote: `workflow gate: parked on ${entryGate.parkOn} pending sign-off`,
         });
-        persistedEvents.push(parked.persistedEvent);
+        register(parked.persistedEvent);
       }
 
       this.instanceStore.updateFrontier(instance.instanceId, [created.qitemId], entryGate ? "waiting" : "active", {
@@ -634,7 +631,7 @@ export class WorkflowRuntime {
         });
       }
 
-      persistedEvents.push(
+      register(
         this.eventBus.persistWithinTransaction({
           type: "workflow.instantiated",
           instanceId: instance.instanceId,
@@ -644,9 +641,6 @@ export class WorkflowRuntime {
         }),
       );
     });
-    txn();
-
-    for (const e of persistedEvents) this.eventBus.notifySubscribers(e);
     if (entryQitemId && entryQitemDestinationSession) {
       await this.queueRepo.maybeNudge(entryQitemId, entryQitemDestinationSession, entryQitemNudge);
     }
@@ -714,7 +708,6 @@ export class WorkflowRuntime {
     resumeCount: number;
     exceptionItemsClosed: number;
   }> {
-    const persistedEvents: PersistedEvent[] = [];
     let result!: {
       instanceId: string;
       stepId: string;
@@ -725,7 +718,7 @@ export class WorkflowRuntime {
     };
     let nudgeTo: { qitemId: string; session: string; nudge: boolean | undefined } | null = null;
 
-    const txn = this.db.transaction(() => {
+    this.eventBus.withNotifyEnvelope((register) => {
       const instance = this.instanceStore.getByIdOrThrow(input.instanceId);
       if (instance.status !== "failed") {
         throw new WorkflowProjectorError(
@@ -809,7 +802,7 @@ export class WorkflowRuntime {
         ],
         chainOfRecord: failedPacketId ? [failedPacketId] : undefined,
       });
-      persistedEvents.push(created.persistedEvent);
+      register(created.persistedEvent);
       nudgeTo = { qitemId: created.qitemId, session: created.destinationSession, nudge: created.nudge };
       // P34: stage the redrive packet's wake intent inside this transaction.
       this.queueRepo.stageWakeIntent(
@@ -843,7 +836,7 @@ export class WorkflowRuntime {
             closureReason: "no-follow-on",
             transitionNote: `workflow resume: occurrence resolved by ${input.actorSession} redriving step ${step.id}${input.decision ? ` — ${input.decision}` : ""}`,
           });
-          persistedEvents.push(closedItem.persistedEvent);
+          register(closedItem.persistedEvent);
           exceptionItemsClosed += 1;
         }
       }
@@ -870,7 +863,7 @@ export class WorkflowRuntime {
         });
       }
 
-      persistedEvents.push(
+      register(
         this.eventBus.persistWithinTransaction({
           type: "workflow.resumed",
           instanceId: instance.instanceId,
@@ -911,11 +904,6 @@ export class WorkflowRuntime {
       // the seam to roll back against, and inventing one would buy a green check
       // that asserts nothing.
     });
-    txn();
-
-    for (const e of persistedEvents) {
-      this.eventBus.notifySubscribers(e);
-    }
     // Closure-assignment cast (the shipped post-commit idiom): TS cannot
     // track the txn-closure write, so narrow via the cast.
     const resumeNudge = nudgeTo as { qitemId: string; session: string; nudge: boolean | undefined } | null;
@@ -947,7 +935,6 @@ export class WorkflowRuntime {
     toSession: string;
     instanceStatus: WorkflowInstance["status"];
   }> {
-    const persistedEvents: PersistedEvent[] = [];
     let result!: {
       instanceId: string;
       stepId: string | null;
@@ -959,7 +946,7 @@ export class WorkflowRuntime {
     };
     let nudgeTo: { qitemId: string; session: string; nudge: boolean | undefined } | null = null;
 
-    const txn = this.db.transaction(() => {
+    this.eventBus.withNotifyEnvelope((register) => {
       const instance = this.instanceStore.getByIdOrThrow(input.instanceId);
       if (instance.status !== "active" && instance.status !== "waiting") {
         throw new WorkflowProjectorError(
@@ -1012,7 +999,7 @@ export class WorkflowRuntime {
         handedOffTo: input.toSession,
         transitionNote: `workflow route: ${input.actorSession} re-routed step ${instance.currentStepId ?? "?"} from ${fromSession} to ${input.toSession}${input.reason ? ` — ${input.reason}` : ""}`,
       });
-      persistedEvents.push(closed.persistedEvent);
+      register(closed.persistedEvent);
 
       // Recreate the SAME step for the new owner (step identity is the
       // work's continuity — the qitem id is a storage artifact of the
@@ -1053,7 +1040,7 @@ export class WorkflowRuntime {
         evidenceRef: oldPacket.evidenceRef ?? undefined,
         targetRepo: oldPacket.targetRepo ?? undefined,
       });
-      persistedEvents.push(created.persistedEvent);
+      register(created.persistedEvent);
       nudgeTo = { qitemId: created.qitemId, session: created.destinationSession, nudge: created.nudge };
       // P34 (site :992): stage the re-routed packet's wake intent in the same
       // transaction that closed the old frontier packet.
@@ -1085,7 +1072,7 @@ export class WorkflowRuntime {
           blockedOn: oldPacket.blockedOn,
           transitionNote: `workflow route: park preserved (${oldPacket.blockedOn})`,
         });
-        persistedEvents.push(reparked.persistedEvent);
+        register(reparked.persistedEvent);
       }
 
       // (2)+(5)+(7) frontier REBIND: same step, new packet, version
@@ -1111,7 +1098,7 @@ export class WorkflowRuntime {
       }
 
       // (6) the shipped event shape {rigName, cause} extended ADDITIVELY.
-      persistedEvents.push(
+      register(
         this.eventBus.persistWithinTransaction({
           type: "workflow.routing_table_changed",
           // OPR.0.4.6.FAC1 (display-only): the instance's actual bound
@@ -1140,9 +1127,6 @@ export class WorkflowRuntime {
       // creates its successor, so the pairing here is unconditional.
       this.queueRepo.assertTerminalClosureHasIntent(oldPacketId, created.qitemId, created.nudge);
     });
-    txn();
-
-    for (const e of persistedEvents) this.eventBus.notifySubscribers(e);
     if (nudgeTo) {
       const n = nudgeTo as { qitemId: string; session: string; nudge: boolean | undefined };
       // P34: shared staged-intent delivery (see the resume path above).

packages/daemon/test/event-notify-envelope.test.ts

@@ -0,0 +1,266 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import type Database from "better-sqlite3";
+import { createFullTestDb } from "./helpers/test-app.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import type { PersistedEvent } from "../src/domain/types.js";
+
+describe("W2b exact-set notify envelope", () => {
+  let db: Database.Database;
+  let bus: EventBus;
+
+  beforeEach(() => {
+    db = createFullTestDb();
+    bus = new EventBus(db);
+  });
+
+  afterEach(() => db.close());
+
+  function persisted(label: string) {
+    return bus.persistWithinTransaction({
+      type: "view.changed",
+      viewName: label,
+      cause: "w2b-test",
+    });
+  }
+
+  it("registers inertly while the transaction is open, then delivers after commit in seq order", () => {
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => {
+      expect(db.inTransaction).toBe(false);
+      received.push(event);
+    });
+
+    bus.withNotifyEnvelope((register) => {
+      const first = persisted("first");
+      register(first);
+      expect(received).toEqual([]);
+      const second = persisted("second");
+      register(second);
+      expect(received).toEqual([]);
+    });
+
+    expect(received.map((event) => event.seq)).toEqual([
+      received[0]!.seq,
+      received[1]!.seq,
+    ]);
+    expect(received[0]!.seq).toBeLessThan(received[1]!.seq);
+    expect(received.map((event) => event.type)).toEqual(["view.changed", "view.changed"]);
+  });
+
+  it.each([
+    [
+      "overwritten result",
+      () => {
+        let token = persisted("overwritten-a");
+        token = persisted("overwritten-b");
+        return [token];
+      },
+    ],
+    [
+      "switch-gated collection",
+      () => {
+        const token = persisted("switch");
+        const registered: ReturnType<typeof persisted>[] = [];
+        switch ("other") {
+          case "register":
+            registered.push(token);
+            break;
+        }
+        return registered;
+      },
+    ],
+    [
+      "loop continue bypass",
+      () => {
+        const registered: ReturnType<typeof persisted>[] = [];
+        for (const label of ["skip", "keep"]) {
+          const token = persisted(label);
+          if (label === "skip") continue;
+          registered.push(token);
+        }
+        return registered;
+      },
+    ],
+    [
+      "wider initializer transform",
+      () => {
+        const registered = [persisted("initializer-a"), persisted("initializer-b")]
+          .filter((token) => token.type === "view.changed")
+          .slice(1);
+        return registered;
+      },
+    ],
+  ])("rejects the %s discriminator by effect", (_name, build) => {
+    expect(() =>
+      bus.withNotifyEnvelope((register) => {
+        for (const token of build()) register(token);
+      }),
+    ).toThrow(/notify envelope.*persisted.*registered/i);
+
+    expect(db.prepare("SELECT COUNT(*) AS count FROM events").get()).toEqual({ count: 0 });
+  });
+
+  it("compares exact token identity, not cardinality", () => {
+    expect(() =>
+      bus.withNotifyEnvelope((register) => {
+        const dropped = persisted("dropped");
+        const duplicated = persisted("duplicated");
+        void dropped;
+        register(duplicated);
+        register(duplicated);
+      }),
+    ).toThrow(/notify envelope.*persisted.*registered/i);
+
+    expect(db.prepare("SELECT COUNT(*) AS count FROM events").get()).toEqual({ count: 0 });
+  });
+
+  it("preserves the current miss as a rollback-producing validation failure", async () => {
+    const queueRepo = new QueueRepository(db, bus);
+    const item = await queueRepo.create({
+      sourceSession: "source@w2b-rig",
+      destinationSession: "owner@w2b-rig",
+      body: "current miss",
+      nudge: false,
+    });
+
+    expect(() =>
+      bus.withNotifyEnvelope(() => {
+        queueRepo.updateWithinTransaction({
+          qitemId: item.qitemId,
+          actorSession: "owner@w2b-rig",
+          state: "in-progress",
+          transitionNote: "intentionally omitted registration",
+        });
+      }),
+    ).toThrow(/notify envelope.*persisted.*registered/i);
+
+    expect(queueRepo.getById(item.qitemId)?.state).toBe("pending");
+  });
+
+  it("accepts a canonical registration loop and a zero-event transaction", () => {
+    expect(() => bus.withNotifyEnvelope(() => undefined)).not.toThrow();
+
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => received.push(event));
+    bus.withNotifyEnvelope((register) => {
+      const tokens = [persisted("a"), persisted("b")];
+      for (const token of tokens) register(token);
+    });
+
+    expect(received.map((event) => (event as { viewName?: string }).viewName)).toEqual(["a", "b"]);
+  });
+
+  it("fails a validation that examined zero envelope transactions", () => {
+    expect(() => bus.assertNotifyEnvelopeExercised()).toThrow(/zero notify envelope transactions/i);
+    bus.withNotifyEnvelope(() => undefined);
+    expect(() => bus.assertNotifyEnvelopeExercised()).not.toThrow();
+  });
+
+  it("records malformed rows durably, advances the watermark, and reports unparseable", () => {
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => received.push(event));
+    let malformedSeq = 0;
+
+    bus.withNotifyEnvelope(() => {
+      malformedSeq = Number(
+        db.prepare("INSERT INTO events (rig_id, node_id, type, payload) VALUES (?, ?, ?, ?)")
+          .run(null, null, "malformed.fixture", "{not-json")
+          .lastInsertRowid,
+      );
+    });
+
+    const status = bus.getNotifyDrainStatus();
+    expect(status.state).toBe("unparseable");
+    expect(status.watermark).toBeGreaterThanOrEqual(malformedSeq);
+    expect(status.lastPoison).toMatchObject({ seq: malformedSeq });
+    expect(status.lastPoison?.payloadSha).toMatch(/^[a-f0-9]{64}$/);
+    expect(status.lastPoison?.error).toBe("invalid event payload JSON");
+    expect(received.some((event) => event.type === "malformed.fixture")).toBe(false);
+    expect(received.filter((event) => event.type === "event.delivery_poisoned")).toHaveLength(1);
+
+    const poison = db.prepare("SELECT payload FROM events WHERE type = 'event.delivery_poisoned'").get() as {
+      payload: string;
+    };
+    expect(JSON.parse(poison.payload)).toMatchObject({
+      type: "event.delivery_poisoned",
+      poisonedSeq: malformedSeq,
+      payloadSha: status.lastPoison?.payloadSha,
+    });
+  });
+
+  it.each([
+    ["null", "null"],
+    ["array", "[]"],
+    ["missing type", "{}"],
+    ["non-string type", '{"type":17}'],
+  ])("treats parseable-but-malformed %s payload as the unparseable third state", (_label, payload) => {
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => received.push(event));
+
+    bus.withNotifyEnvelope(() => {
+      db.prepare("INSERT INTO events (rig_id, node_id, type, payload) VALUES (?, ?, ?, ?)")
+        .run(null, null, "malformed.fixture", payload);
+    });
+
+    expect(bus.getNotifyDrainStatus()).toMatchObject({
+      state: "unparseable",
+      lastPoison: { error: "invalid event payload shape" },
+    });
+    expect(received.filter((event) => event.type === "event.delivery_poisoned")).toHaveLength(1);
+    expect(received.some((event) => (event as { type?: unknown }).type === 17)).toBe(false);
+  });
+
+  it("starts at MAX(seq) so legacy malformed rows remain replay-only", () => {
+    db.prepare("INSERT INTO events (rig_id, node_id, type, payload) VALUES (?, ?, ?, ?)")
+      .run(null, null, "legacy.malformed", "{not-json");
+    const freshBus = new EventBus(db);
+    const received: PersistedEvent[] = [];
+    freshBus.subscribe((event) => received.push(event));
+
+    freshBus.withNotifyEnvelope((register) => {
+      register(freshBus.persistWithinTransaction({
+        type: "view.changed",
+        viewName: "fresh",
+        cause: "post-boot",
+      }));
+    });
+
+    expect(freshBus.getNotifyDrainStatus().state).toBe("healthy");
+    expect(received.map((event) => event.type)).toEqual(["view.changed"]);
+  });
+
+  it("delivers the row-derived value so live and replay payloads are identical", () => {
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => received.push(event));
+
+    bus.withNotifyEnvelope((register) => {
+      register(bus.persistWithinTransaction({
+        type: "view.changed",
+        viewName: "parity",
+        cause: "row-derived",
+        transient: undefined,
+      } as Parameters<EventBus["persistWithinTransaction"]>[0]));
+    });
+
+    expect(received).toEqual(bus.replayAll(0));
+    expect(received[0]).not.toHaveProperty("transient");
+  });
+
+  it("delivers an orphan row normally without consulting live topology", () => {
+    const received: PersistedEvent[] = [];
+    bus.subscribe((event) => received.push(event));
+
+    bus.withNotifyEnvelope((register) => {
+      register(
+        bus.persistWithinTransaction({
+          type: "rig.deleted",
+          rigId: "rig-that-does-not-exist",
+        }),
+      );
+    });
+
+    expect(received).toHaveLength(1);
+    expect(received[0]).toMatchObject({ type: "rig.deleted", rigId: "rig-that-does-not-exist" });
+  });
+});

packages/daemon/test/view-event-bridge.test.ts

@@ -1,4 +1,4 @@
-import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 import type Database from "better-sqlite3";
 import { createDb } from "../src/db/connection.js";
 import { migrate } from "../src/db/migrate.js";
@@ -59,7 +59,10 @@ describe("view-event-bridge (PL-004 Phase B R1; BLOCKER 2 fix)", () => {
     wireViewEventBridge(bus, projector);
   });
 
-  afterEach(() => db.close());
+  afterEach(() => {
+    vi.restoreAllMocks();
+    db.close();
+  });
 
   function viewChangedEvents(): Array<{ viewName: string; cause: string }> {
     return captured
@@ -217,15 +220,77 @@ describe("view-event-bridge (PL-004 Phase B R1; BLOCKER 2 fix)", () => {
     expect(events.filter((e) => e.cause === "classifier.lease_acquired")).toHaveLength(0);
   });
 
-  it("view.changed itself does NOT trigger view.changed (no feedback loop)", () => {
+  it("notifyViewChanged persists only and view.changed does not echo through the bridge", () => {
     projector.notifyViewChanged("recently-active", "manual-test");
     captured.length = 0;
     projector.notifyViewChanged("recently-active", "manual-test-2");
-    // The bridge should not re-emit view.changed in response to view.changed.
-    const newViewChangedEvents = captured.filter((e) => e.type === "view.changed");
-    // Only the direct notifyViewChanged call's event; no echo.
-    expect(newViewChangedEvents).toHaveLength(1);
-    expect((newViewChangedEvents[0] as { cause: string }).cause).toBe("manual-test-2");
+    expect(captured).toEqual([]);
+    const persisted = bus.replayAll(0).filter((event) => event.type === "view.changed");
+    expect(persisted).toHaveLength(2);
+    expect(persisted.map((event) => event.cause)).toEqual(["manual-test", "manual-test-2"]);
+  });
+
+  it("W2b drains outer events before re-entrant view.changed rows and delivers each exactly once", () => {
+    captured.length = 0;
+
+    bus.withNotifyEnvelope((register) => {
+      register(bus.persistWithinTransaction({
+        type: "queue.handed_off",
+        qitemId: "q-w2b-handoff",
+        fromSession: "alice@rig",
+        toSession: "bob@rig",
+        closureReason: "handed_off_to",
+        summary: null,
+      }));
+      register(bus.persistWithinTransaction({
+        type: "inbox.denied",
+        inboxId: "inbox-w2b-denied",
+        destinationSession: "bob@rig",
+        senderSession: "alice@rig",
+        reason: "ordering discriminator",
+      }));
+    });
+
+    expect(captured.slice(0, 2).map((event) => event.type)).toEqual([
+      "queue.handed_off",
+      "inbox.denied",
+    ]);
+    const replayed = bus.replayAll(0);
+    expect(captured.map((event) => event.seq)).toEqual([1, 2, 3, 4, 5, 6]);
+    expect(captured.map((event) => event.seq)).toEqual(replayed.map((event) => event.seq));
+    expect(new Set(captured.map((event) => event.seq)).size).toBe(captured.length);
+    expect(
+      captured
+        .filter((event) => event.type === "view.changed")
+        .map((event) => `${event.viewName}:${event.cause}`),
+    ).toEqual([
+      "recently-active:queue.handed_off",
+      "pod-load:queue.handed_off",
+      "activity:queue.handed_off",
+      "activity:inbox.denied",
+    ]);
+  });
+
+  it("W2b keeps drain-time delivery failures outside the bridge best-effort persist catch", () => {
+    const originalNotify = bus.notifySubscribers.bind(bus);
+    vi.spyOn(bus, "notifySubscribers").mockImplementation((event) => {
+      if (event.type === "view.changed") throw new Error("drain delivery failed");
+      originalNotify(event);
+    });
+
+    expect(() =>
+      bus.withNotifyEnvelope((register) => {
+        register(bus.persistWithinTransaction({
+          type: "queue.created",
+          qitemId: "q-w2b-delivery-error",
+          sourceSession: "alice@rig",
+          destinationSession: "bob@rig",
+          priority: "routine",
+          tier: null,
+          summary: null,
+        }));
+      }),
+    ).toThrow("drain delivery failed");
   });
 
   it("unsubscribe stops the bridge", async () => {

packages/daemon/test/view-projector.test.ts

@@ -197,10 +197,15 @@ describe("ViewProjector (PL-004 Phase B; L5 read-only projections)", () => {
     expect(result.custom[0]!.viewName).toBe("my-view");
   });
 
-  it("notifyViewChanged emits view.changed event", () => {
+  it("notifyViewChanged persists view.changed for the enclosing drain without direct delivery", () => {
     const captured: unknown[] = [];
     bus.subscribe((e) => captured.push(e));
     projector.notifyViewChanged("recently-active", "queue.created");
-    expect(captured.some((e) => (e as { type: string }).type === "view.changed")).toBe(true);
+    expect(captured).toEqual([]);
+    expect(bus.replayAll(0).at(-1)).toMatchObject({
+      type: "view.changed",
+      viewName: "recently-active",
+      cause: "queue.created",
+    });
   });
 });