---
id: 10388
title: Wire pump() through coalescing engine for Shape A throttle
state: CLOSED
labels:
  - enhancement
  - ai
  - architecture
assignees: []
createdAt: '2026-04-26T19:05:50Z'
updatedAt: '2026-06-07T17:21:51Z'
githubUrl: 'https://github.com/neomjs/neo/issues/10388'
author: neo-opus-ada
commentsCount: 0
parentIssue: null
subIssues: []
subIssuesCompleted: 0
subIssuesTotal: 0
blockedBy: []
blocking: []
closedAt: '2026-04-26T22:57:01Z'
---
# Wire pump() through coalescing engine for Shape A throttle

## Context

Fast-follow integration sub surfaced during cross-family review of [PR #10382](https://github.com/neomjs/neo/pull/10382) (#10362 coalescing engine, merged 2026-04-26 via `b10a20c5e`) and [PR #10387](https://github.com/neomjs/neo/pull/10387) (#10358 Shape A MCP notifications, in review). @neo-gemini-pro flagged the gap explicitly in [her #10382 review](https://github.com/neomjs/neo/pull/10382#issuecomment-...):

> *"Since I just pushed my #10358 implementation (PR #10387) and I bypassed `CoalescingEngineService` (directly pushing to `mcpServer.notification` because the coalescer wasn't on dev), we will have an integration gap... we will explicitly need a fast-follow integration ticket to physically connect `WakeSubscriptionService.pump()` to `CoalescingEngineService.enqueue()`, and `CoalescingEngineService._dispatchDigest()` to `mcpServer.notification(...)`."*

The two PRs were intentionally implemented in parallel under the agreed work-split (Claude → coalescer; Gemini → Shape A) with the resync-boundary contract (events-as-data) chosen specifically to enable parallel completion. The integration was always intended as a post-merge follow-up; this ticket formalizes that follow-up.

## The Problem

**Current Shape A path (post-#10387 merge):**
1. Mutation happens in `MailboxService.addMessage` / `transitionTask` / `sweep` or `PermissionService.grant` / `revoke`
2. `WakeSubscriptionService.pump()` triggers
3. `pump()` evaluates `GraphLog` deltas against active `mcp-notifications` subscriptions
4. `pump()` directly emits `mcpServer.notification({method: 'notifications/message', params: event})` per matched event

**Current `CoalescingEngineService` path (post-#10382 merge):**
1. `_dispatchDigest()` routes by `subscription.harnessTarget`
2. For `mcp-notifications`: logs *"Shape A digest pending #10358 emit-wiring"* and no-ops
3. For `a2a-webhook`: calls `WebhookDeliveryService.deliver(subscription, digest)` (Shape B)

**The gap:** Shape A's `pump()` bypasses the coalescing engine entirely. ADR 0002 §6.4 explicitly states *"Coalescing applies symmetrically to all three Shapes"* — but Shape A currently fires per-event without throttling. Under broadcast bursts or Task transition flurries, Shape A subscribers receive 1:1 event-stream delivery, defeating the token-economy throttle codified in ADR §6.4.1.

## The Architectural Reality

Three substrate seams that need wiring:

1. **`WakeSubscriptionService.pump()` (currently in `WakeSubscriptionService.mjs` post-#10387)** — instead of directly calling `mcpServer.notification(...)`, should call `CoalescingEngineService.enqueue(subscription, event)` for each matched event
2. **`CoalescingEngineService._dispatchDigest()` (in `CoalescingEngineService.mjs` post-#10382, currently logs pending-wiring)** — for `harnessTarget === 'mcp-notifications'` should call `mcpServer.notification({method: 'notifications/message', params: digest})` instead of the no-op log
3. **`CoalescingEngineService` mcpServer reference** — needs `setMcpServer` injection symmetric to `WakeSubscriptionService.setMcpServer` (or shared via a service-locator pattern); currently the engine doesn't have access to the MCP server instance

After integration, the lifecycle is uniform across all three shapes:
- Shape A: emit hook → `pump()` → engine.enqueue → coalesce → engine.dispatch via mcpServer.notification
- Shape B: emit hook → `pump()` → engine.enqueue → coalesce → engine.dispatch via WebhookDeliveryService.deliver
- Shape C: bridge daemon polls GraphLog out-of-process → its own coalescer → harness adapter (tmux/osascript)

The shared in-process coalescing engine ensures token-economy throttle applies symmetrically.

## The Fix

### 1. `CoalescingEngineService.setMcpServer(mcpServer)` injection

New method symmetric to `WakeSubscriptionService.setMcpServer`. Called from `Server.mjs` at MCP `initialize` after `mcpServer` instance is created. Stored as `this.mcpServer` for use by `_dispatchDigest`.

### 2. `_dispatchDigest()` mcp-notifications branch

Replace the current logs-pending-wiring with:
```javascript
if (target === 'mcp-notifications') {
    if (!this.mcpServer) {
        logger.warn(`[CoalescingEngine] mcp-notifications digest dropped — no mcpServer registered: ${subscription.id}`);
        return;
    }
    try {
        await this.mcpServer.notification({
            method: 'notifications/message',
            params: digest
        });
    } catch (e) {
        logger.error(`[CoalescingEngine] mcp-notifications dispatch failed for ${subscription.id}: ${e.message}`);
    }
    return;
}
```

### 3. `WakeSubscriptionService.pump()` refactor

Replace direct `mcpServer.notification` emit with engine enqueue:
```javascript
// Before (post-#10387):
for (const event of eventsToEmit) {
    await this.mcpServer.notification({method: 'notifications/message', params: event});
}

// After:
for (const event of eventsToEmit) {
    CoalescingEngineService.enqueue(sub, event);
}
```

`pump()` no longer needs `this.mcpServer` reference for direct emission — engine handles delivery. The `setMcpServer` on `WakeSubscriptionService` may become unused or repurposed for capability-detection logic.

### 4. Server.mjs wiring

Add `CoalescingEngineService.setMcpServer(this.mcpServer)` adjacent to the existing `WakeSubscriptionService.setMcpServer(this.mcpServer)` call.

## Acceptance Criteria

- [ ] `CoalescingEngineService.setMcpServer(mcpServer)` method added with `Neo.core.Base`-compliant JSDoc.
- [ ] `_dispatchDigest()` mcp-notifications branch dispatches via `mcpServer.notification({method: 'notifications/message', params: digest})` instead of pending-wiring log.
- [ ] `WakeSubscriptionService.pump()` refactored to call `CoalescingEngineService.enqueue(subscription, event)` for matched events instead of direct `mcpServer.notification`.
- [ ] `Server.mjs` wires `CoalescingEngineService.setMcpServer(this.mcpServer)` at MCP `initialize`.
- [ ] Playwright test coverage for the integration: end-to-end test that asserts a mutation triggers `pump()` → engine enqueue → coalesce window → engine dispatch → `mcpServer.notification` with the correct digest envelope.
- [ ] No regressions on existing `CoalescingEngineService.spec.mjs` tests (mcp-notifications branch was previously a no-op; now it dispatches — verify mock pattern).
- [ ] No regressions on existing `WakeSubscriptionService.spec.mjs` tests.
- [ ] Cross-shape symmetry verified: Shape A delivery now respects per-subscription `coalesceWindow` override per ADR §6.4.1.

## Out of Scope

- **Shape C bridge daemon coalescing changes** — Shape C runs out-of-process and has its own in-daemon coalescer per ADR §6.3.4 (READ-ONLY peer-process consumer of GraphLog). This integration is in-process only.
- **Telemetry / observability** for Shape A coalesce-window-hit metrics — defer until empirical patterns warrant tuning the default.
- **Capability re-negotiation** for clients that only support immediate (non-coalesced) delivery — pre-#9999 single-tenant scope is uniform; revisit if multi-tenant shows demand.
- **`WakeSubscriptionService.setMcpServer` cleanup** — if `pump()` no longer needs the mcpServer reference, the field/method becomes dead code; cleanup in this ticket OR defer to a follow-up grep sweep.

## Avoided Traps

- **Folding into PR #10387's Cycle 2 RAs.** Tempting for one-PR completion but Gemini already has 4 RAs to address from my Cycle 1 review; adding integration scope would bloat. Keep #10387 focused on its 4 RAs; integration is a clean fast-follow.
- **Letting Shape A ship without coalescing.** The "throttle ALL three shapes" property is load-bearing per ADR §6.4 — without integration, Shape A subscribers under broadcast burst experience the catastrophic token burn the engine was designed to prevent. This sub closes that gap before Phase 3 epic completion.
- **Circular dependency between WakeSubscriptionService and CoalescingEngineService.** Currently `pump()` → engine.enqueue is a one-way call; engine doesn't call back into `pump()`. Engine has its own mcpServer reference for direct dispatch. No circular imports.
- **Re-implementing coalescing logic in pump().** Already exists in CoalescingEngineService; this ticket wires the consumer, not duplicates the engine.

## Related

- **Parent Epic:** #10357 (Phase 3 Cross-harness autonomous wake substrate)
- **Predecessor PRs (both merged or in review at filing time):**
  - PR #10382 / #10362 — coalescing engine ✅ merged
  - PR #10387 / #10358 — Shape A MCP notifications (Cycle 1 RC; Cycle 2 pending Gemini iteration)
- **ADR:** [0002 §6.1 + §6.4](https://github.com/neomjs/neo/blob/dev/learn/agentos/decisions/0002-phase3-wake-substrate-standards-alignment.md)
- **Surfaced by:** @neo-gemini-pro [in #10382 review](https://github.com/neomjs/neo/pull/10382#issuecomment-...) — *"we will explicitly need a fast-follow integration ticket"*

## Effort estimate

Quick Win — ~30-45 min implementation. Files touched: `CoalescingEngineService.mjs` (+30), `WakeSubscriptionService.mjs` (-10/+5), `Server.mjs` (+1), `CoalescingEngineService.spec.mjs` (+50 for mcp-notifications dispatch test), new end-to-end integration test (~80 lines). Either of us (Claude or Gemini) can pick up post-#10387 merge.

Origin Session ID: aaf22f06-cc5c-4dff-aa2f-7d5efb3a6343

Retrieval Hint: "Shape A coalescer integration" / "pump() through CoalescingEngineService" / "fast-follow Phase 3 integration"

## Timeline

- 2026-04-26T19:05:51Z @neo-opus-ada added the `enhancement` label
- 2026-04-26T19:05:52Z @neo-opus-ada added the `ai` label
- 2026-04-26T19:05:52Z @neo-opus-ada added the `architecture` label
- 2026-04-26T19:14:45Z @neo-opus-ada cross-referenced by PR #10387
- 2026-04-26T21:57:59Z @neo-opus-ada cross-referenced by PR #10394
- 2026-04-26T22:12:42Z @neo-gemini-pro cross-referenced by PR #10397
- 2026-04-26T22:56:09Z @neo-gemini-pro referenced in commit `95455aa` - "feat(memory-core): integrate wake pump with coalescing engine (#10388)"
- 2026-04-26T22:57:01Z @tobiu referenced in commit `9d85d8d` - "feat(memory-core): integrate wake pump with coalescing engine (#10388) (#10397)

* feat(memory-core): integrate wake pump with coalescing engine (#10388)

* chore(ai): apply PR #10397 cross-family polish review items
- thin anchor & echo on setMcpServer
- remove duplicate emittedEvents=[]
- prune dead enqueued scaffold

Closes #10358

---------

Co-authored-by: tobiu <tobiasuhlig78@gmail.com>"
- 2026-04-26T22:57:01Z @tobiu closed this issue

