feat(mcpl): RFC-006 event coalescing, kept in context

README.md

@@ -239,6 +239,30 @@ Optional host-side implementation of the MCP Live protocol. External servers (ga
 }
 ```
 
+#### Event coalescing (RFC-006)
+
+The host advertises `eventCoalescing` (both lanes, deferred rendering,
+channel-scoped pushes, a one-hour retry window). A server may add
+`coalesce: { key }` to a `push/event` or a `channels/incoming` message so that a
+later occurrence of the same subject **replaces** the earlier one while it is
+still unread, and **appends** once a model has seen it; `retract: true`
+withdraws unread content and appends the supplied deletion notice only when
+some version was read (or history is unknown). `initial: true` on a create lets
+the host know the subject's history is complete, so a create → edit → delete
+that nobody read leaves no trace. `deferred: true` (push only) holds a batch of
+notices outside context and asks the server for the content with `push/render`
+when a turn is assembled.
+
+Plain content stays in context and is delivered on arrival through the ordinary
+channel/push path; the host only remembers where it landed. "Unread" means:
+above the agent's consumed watermark (advanced at every compile of a model
+request), on the current branch, not a sharded message, and not folded into a
+summary by compression. On any doubt — restart, branch switch, eviction — the
+host appends, as it does today. There is no per-subject cap: a busy channel's
+unread backlog is simply its unread backlog. Receipts and subject history are
+kept in the `mcpl/coalescing` state (retry window 1 h); audit lines are
+`mcpl:coalescing` trace events.
+
 ### Streaming Lifecycle
 
 1. **Start**: Framework calls `agent.startStreamWithInjections()` → `YieldingStream`

changelog.d/mcpl-event-coalescing.added.md

@@ -0,0 +1 @@
+- MCPL event coalescing (RFC-006, mcpl PR #5 revision 7): `coalesce` on `push/event` and `channels/incoming` — replace-if-unread, atomic retraction with a conditional deletion notice, `initial` birth marker, deferred batches rendered through `push/render` at assembly, channel-scoped pushes delivered as messages of their channel. Content stays in context; "unread" is above the agent's consumed watermark and not folded by compression. Advertised as `eventCoalescing` with a one-hour retry window.

src/agent.ts

@@ -522,8 +522,50 @@ export class Agent {
     this.updateHotContextSettings({ preparedWindowTokens: null });
   }
 
+  /**
+   * RFC-006 consumed watermark: the newest stored message at the last compile
+   * of a model request, per branch. Everything above it has never been in a
+   * request and may still be replaced or withdrawn in place by its sender.
+   */
+  private consumedWatermark: { branch: string; sequence: number } | null = null;
+
+  getConsumedWatermark(): { branch: string; sequence: number } | null {
+    return this.consumedWatermark;
+  }
+
+  /** Mark everything currently stored as consumed (boot, branch switch). */
+  markContextConsumed(): void {
+    // Tolerates partial context-manager doubles (tests build Agents over
+    // stubs): without a watermark nothing is ever treated as unread.
+    this.consumedWatermark = this.pendingWatermark();
+  }
+
+  /** The watermark a successful compile will establish (taken before, applied after). */
+  private pendingWatermark(): { branch: string; sequence: number } | null {
+    const cm = this.contextManager as Partial<ContextManager>;
+    try {
+      if (typeof cm.currentBranch !== 'function' || typeof cm.getAllMessages !== 'function') return null;
+      // The chronicle head, not this slot's last message: a compile reads
+      // everything stored on the branch so far, including messages merged in
+      // from another slot (the subconscious reads the residents' shared slot).
+      const store = typeof cm.getStore === 'function' ? cm.getStore() as { currentSequence?: () => number } : undefined;
+      let sequence = typeof store?.currentSequence === 'function' ? store.currentSequence() : undefined;
+      if (sequence === undefined) {
+        const count = typeof cm.getMessageCount === 'function' ? cm.getMessageCount() : cm.getAllMessages().length;
+        sequence = count > 0 ? cm.getAllMessages().at(-1)?.sequence ?? 0 : 0;
+      }
+      return { branch: cm.currentBranch().name, sequence };
+    } catch {
+      return null;
+    }
+  }
+
   async compileContext(budget?: TokenBudget): Promise<CompileResult> {
+    // RFC-006 §3.3: consumption is determined at assembly — a compile that
+    // fails assembled nothing, so the watermark moves only on success.
+    const watermark = this.pendingWatermark();
     const result = await this.contextManager.compile(this.resolveBudget(budget));
+    if (watermark) this.consumedWatermark = watermark;
     if (!budget) this.settleRuntimeSettingsTransition();
     return result;
   }
@@ -538,9 +580,11 @@ export class Agent {
     injections?: ContextInjection[],
     opts?: { kvUnifiedImmutablePrefixHash?: string },
   ): Promise<CompileResult> {
+    const watermark = this.pendingWatermark();
     const result = await this.contextManager.compile(
       this.resolveBudget(budget), injections, opts as never,
     );
+    if (watermark) this.consumedWatermark = watermark;
     if (!budget) this.settleRuntimeSettingsTransition();
     return result;
   }

src/api/types.ts

@@ -369,6 +369,8 @@ export type ApiEventType =
   | 'module:added'
   | 'module:removed'
   | 'message:added'
+  | 'message:removed'
+  | 'mcpl:coalescing'
   // Server-specific events
   | 'connected'
   | 'speech'