fix(mcpl): withdraw a batch-only turn whose batch is gone, rendering elsewhere, or queued for another agent

changelog.d/coalescing-batch-turn-content.changed.md

@@ -0,0 +1 @@
+- `getActiveTurnTrigger` returns the trigger of a turn that a coalesced batch woke without the host-owned `coalescingSubject`, `coalescingBatch` and `coalescingBatchSubjects` while that turn is retried: an in-turn retry, a provider hold's release, a requeue behind an auxiliary call, or the tool-result guard's restart. Those fields identify a wake for withdrawal while it is unconsumed, and a retry belongs to a turn that has already passed its batch gate. A turn whose merged wakes don't all share one subject shows no `coalescingSubject` either. Its other fields, such as `reason`, `source`, `channelId` and `counterparty`, are as before.

changelog.d/coalescing-batch-turn-content.fixed.md

@@ -0,0 +1,2 @@
+- A turn woken only by deferred coalescing batches is withdrawn before inference in more of the cases where it used to call the model with nothing new to read: its batch was retracted or replaced after the wake was taken (the replacement or the retraction's notice wakes agents as any push does, by its own `triggerInference`), its batch was being rendered for another agent's assembly (the turn waits for that render, which the other agent's request reads), or the render was placed in another agent's deferred queue because that agent's turn was running. In that last case that agent is now woken for the render, as the batch's own wake would have woken it; before, nothing woke it for the render. When several agents are woken for one batch, the first assembly consumes it, and the others' turns already taken from the queue are withdrawn, as their still-queued wakes already were. A turn woken by a batch that a provider hold kept after its call failed is now retried with its own trigger when the hold ends, even if a newer batch of its subject arrived meanwhile. Before, the newer batch withdrew it, and the agent's next turn came from the newer batch's wake or, with nothing left held, from a fresh compile that names no channel or author.
+- When one agent's queued wakes merge into one turn, the turn's trigger now carries a coalescing subject only when it is every merged wake's, so a withdrawal of that subject's wakes, such as a retraction's or a newer batch's, takes the merged wake only when that subject was its sole cause. Before, the merged trigger carried its first wake's subject, and such a withdrawal, landing while the merged wake was held or requeued before its turn started, also dropped the wake's other causes, such as an ordinary message.

src/framework.ts

@@ -11711,6 +11711,7 @@ export class AgentFramework {
       remove: (p) => this.removePlacement(p),
       deliver: (occ, materialized, assemblingFor) => this.deliverCoalesced(occ, materialized, assemblingFor),
       wakeForBatch: (occ) => this.wakeForCoalescedBatch(occ),
+      wakeToRead: (occ, agentName) => this.queueCoalescedWakes({ ...occ, deliverTo: agentName }, true),
       cancelWake: (subject) => this.cancelCoalescedWakes(subject),
       authorized: (occ) => this.coalescedAuthorized(occ),
       audience: (occ) => this.coalescedAudience(occ),
@@ -12098,8 +12099,16 @@ export class AgentFramework {
   }
 
   private async wakeForCoalescedBatch(occ: CoalescedOccurrence<CoalescedDelivery>): Promise<void> {
+    this.cancelCoalescedWakes(coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key));
+    await this.queueCoalescedWakes(occ);
+  }
+
+  /** Queue a wake for each agent of the batch's audience, when its latest
+   *  occurrence triggers inference. `plain`: an ordinary push wake, without
+   *  the subject a withdrawal of the batch's wakes goes by, for a render
+   *  placed in its agent's deferred queue (wakeToRead). */
+  private async queueCoalescedWakes(occ: CoalescedOccurrence<CoalescedDelivery>, plain = false): Promise<void> {
     const subject = coalescingSubjectKey(occ.serverId, occ.binding, occ.scope, occ.key);
-    this.cancelCoalescedWakes(subject);
     if (occ.event.lane !== 'push' || !occ.event.event.triggerInference) return;
     const origin = occ.event.event.origin;
     // The wake carries the batch's channel so the turn freezes its reply
@@ -12115,8 +12124,7 @@ export class AgentFramework {
         channelId,
         counterparty: authorId ? `${occ.serverId}:user:${authorId}` : undefined,
         addressed: isAddressedMessage(occ.tags, origin),
-        coalescingSubject: subject,
-        coalescingBatch: true,
+        ...(plain ? {} : { coalescingSubject: subject, coalescingBatch: true }),
       });
     }
   }
@@ -12680,6 +12688,13 @@ export class AgentFramework {
         coalescingBatchSubjects: batchOnly
           ? [...new Set(requests.flatMap((r) => r.coalescingBatchSubjects ?? [r.coalescingSubject!]))]
           : undefined,
+        // A withdrawal of a subject's wakes takes a wake whose sole cause is
+        // that subject (§4.2), so the merged trigger carries a subject only
+        // when it is every merged request's: withdrawing it then never takes
+        // the merge's other causes, such as an ordinary wake, another
+        // subject's batch, or a retry past its gate.
+        coalescingSubject: requests.every((r) => r.coalescingSubject === trigger?.coalescingSubject)
+          ? trigger?.coalescingSubject : undefined,
         suppressProse: silentOnly ? trigger?.suppressProse : undefined,
         ephemeralSystemPrompt: silentOnly ? trigger?.ephemeralSystemPrompt : undefined,
         silentHeartbeat: silentOnly ? trigger?.silentHeartbeat : undefined,
@@ -14747,10 +14762,16 @@ export class AgentFramework {
     // boundary — after the deferred flush (same window: turn alive, compile
     // not yet run) and before the checkpoint (they are the turn's inputs).
     // Not on a context-budget restart: that continues the same logical turn.
+    // A turn whose only causes are deferred batches assembles even when no
+    // batch is pending: its batch may have been withdrawn after its wake was
+    // dispatched or be rendering for another agent's assembly, and only the
+    // assembly can say whether the turn has anything to read.
+    const batchSubjects = trigger?.coalescingBatchSubjects;
     let assembly: AssemblyResult | undefined;
-    if (attempt === 0 && !resumesTurn && this.pushCoalescer?.pendingBatches()) {
+    if (attempt === 0 && !resumesTurn && this.pushCoalescer
+      && (batchSubjects?.length || this.pushCoalescer.pendingBatches())) {
       try {
-        assembly = await this.pushCoalescer.assemble(agent.name);
+        assembly = await this.pushCoalescer.assemble(agent.name, batchSubjects);
       } catch (err) {
         console.error(`[coalescing] assembly for ${agent.name} failed:`, err);
       }
@@ -14759,15 +14780,16 @@ export class AgentFramework {
     // A turn whose only causes were deferred batches, none of which gave it
     // anything to read (an empty or blank render, §5.2; cancelled; revoked; a
     // render-start that could not be made durable, whose batch then waits for
-    // its recovery wake; a batch whose audience doesn't include the agent),
-    // has nothing new to show: running it would wake the model to
-    // `[Continue]` or an older message, an uncaused wake. Stop before the
-    // checkpoint, locus, typing and compile. Any batch this assembly
-    // materialized keeps the turn, including one that was not among its
-    // causes (a batch whose wake the freeze withdrew, or one that never
-    // qualified for a wake). A cause this assembly did not settle (rendered
-    // elsewhere, not yet frozen) keeps the turn, as before.
-    const batchSubjects = trigger?.coalescingBatchSubjects;
+    // its recovery wake; a batch whose audience doesn't include the agent; a
+    // render placed in another agent's deferred queue; a batch another
+    // assembly froze, or one withdrawn, displaced or consumed by another
+    // request before this assembly), has nothing new to show: running it would
+    // wake the model to `[Continue]` or an older message, an uncaused wake.
+    // Stop before the checkpoint, locus, typing and compile. Any batch this
+    // assembly materialized keeps the turn, including one that was not among
+    // its causes (a batch whose wake the freeze withdrew, or one that never
+    // qualified for a wake). A cause this assembly did not settle (the host
+    // stopping) keeps the turn, as before.
     if (assembly && batchSubjects?.length && assembly.materialized.size === 0
       && batchSubjects.every((s) => assembly.settled.has(s))) {
       console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`);
@@ -14785,6 +14807,20 @@ export class AgentFramework {
       }
       return false;
     }
+    // The turn has passed its gate, so from here it goes on with a copy of
+    // its trigger without a batch wake's marks. A retry of it then comes back
+    // as an ordinary wake by whatever door it takes (a provider hold's
+    // release, the auxiliary park's requeue, the tool-result guard's
+    // restart): this gate doesn't withdraw it for finding its batches
+    // consumed, and no withdrawal of the subject's wakes reaches it (§4.2
+    // withdraws a wake whose sole cause is the subject, and a merge carries
+    // a subject only when it is every merged request's). This frame's
+    // activeTurnTriggers entry keeps the trigger as it came; a retry's frame
+    // records the copy it carries.
+    if (trigger?.coalescingSubject !== undefined || trigger?.coalescingBatch) {
+      const { coalescingSubject: _subject, coalescingBatch: _batch, coalescingBatchSubjects: _subjects, ...retry } = trigger;
+      trigger = retry;
+    }
 
     // Record turn checkpoint before inference (only on first attempt, not retries)
     if (attempt === 0 && trigger?.reason !== 'tool_result_guard_retry') {

src/mcpl/push-coalescer.ts

@@ -130,14 +130,19 @@ interface Rendering<E> {
 /** What an assembly did, per subject, for the turn being assembled. */
 export interface AssemblyResult {
   /** Subjects this assembly finished with: it waited on their render to
-   *  completion, or freeze() declined them for this agent. */
+   *  completion, freeze() declined them for this agent, or they were among
+   *  the turn's causes and their batch was gone or another assembly's. */
   settled: Set<string>;
-  /** Of those, the subjects whose render or fallback was delivered as content.
-   *  A settled subject that is not here gave the turn nothing to read: empty
-   *  or blank render (§5.2), cancelled, revoked before the render or at its
-   *  response, a render-start that could not be made durable (the batch then
-   *  waits for its recovery wake), or a batch whose audience doesn't include
-   *  the agent. */
+  /** Of those, the subjects whose render or fallback this assembly delivered
+   *  where its compile reads it, as content the turn reads as its own. A
+   *  settled subject that is not here gave the turn nothing new of its own to
+   *  read: empty or blank render (§5.2), cancelled, revoked before the render
+   *  or at its response, a render-start that could not be made durable (the
+   *  batch then waits for its recovery wake), a batch whose audience doesn't
+   *  include the agent, a render placed in another agent's deferred queue, a
+   *  render another assembly froze (its compile may show it, as that
+   *  request's), or a cause whose batch was withdrawn, displaced or consumed
+   *  by another request. */
   materialized: Set<string>;
 }
 
@@ -193,6 +198,11 @@ export interface CoalescerHost<E> {
   deliver(occurrence: CoalescedOccurrence<E>, materialized?: McplContentBlock[], assemblingFor?: string): Promise<CoalescingPlacement | undefined>;
   /** Queue a wake for a batch that has no model-visible content yet. */
   wakeForBatch(occurrence: CoalescedOccurrence<E>): Promise<void>;
+  /** Queue an ordinary push wake for `agent`, under the batch's own wake
+   *  policy, to read a render of the batch placed in its deferred queue that
+   *  no request consumed. It carries no subject, so the subject's other wakes,
+   *  and later withdrawals of them, leave it. */
+  wakeToRead(occurrence: CoalescedOccurrence<E>, agent: string): Promise<void>;
   /** Withdraw unstarted wakes whose sole cause is this subject. */
   cancelWake(subject: string): void;
   /** The batch's audience still holds the authority it was admitted under. */
@@ -675,23 +685,32 @@ export class PushCoalescer<E = unknown> {
    * `agentName`. Called by the host at a turn's assembly boundary, before the
    * compile. Bounded by the host's render timeout; a failed or late render
    * materializes the admitted fallback (§5.3). Reports which subjects
-   * settled and which of them produced content, so the host can tell a turn
-   * whose only cause rendered nothing.
+   * settled and which of them gave the turn content, so the host can tell a
+   * turn whose only causes gave it nothing to read.
+   *
+   * `causes` are the subjects whose batch wakes started the turn. A batch has
+   * one consumer, the request it was frozen for (§5.2: the materialized
+   * occurrence is consumed by that request). So a cause whose batch is gone
+   * by now settles and gives the turn nothing, and an assembly that joins
+   * another's render waits for it, so its compile shows the content, without
+   * taking it as its own.
    */
-  async assemble(agentName: string): Promise<AssemblyResult> {
+  async assemble(agentName: string, causes: readonly string[] = []): Promise<AssemblyResult> {
     const result: AssemblyResult = { settled: new Set(), materialized: new Set() };
     if (this.suspended) return result;
     const work: Promise<void>[] = [];
-    for (const subject of [...this.subjects.keys()]) {
+    const wanted = new Set(causes);
+    for (const subject of new Set([...this.subjects.keys(), ...causes])) {
       // The handle is wrapped: a bare promise returned through `locked` would
       // be flattened by `then`, holding the lock until the render settled —
       // and the settlement itself needs the lock.
-      const started = await this.locked(() => this.freeze(subject, agentName));
+      const started = await this.locked(() => this.freeze(subject, agentName, wanted.has(subject)));
       if (started === 'declined') result.settled.add(subject);
       else if (started) {
         work.push(started.done.then(() => {
           result.settled.add(subject);
-          if (started.rendering.materialized) result.materialized.add(subject);
+          // A render another assembly froze is that request's to read.
+          if (started.rendering.materialized && !started.joined) result.materialized.add(subject);
         }));
       }
     }
@@ -703,27 +722,40 @@ export class PushCoalescer<E = unknown> {
   /**
    * Under the lock: decide whether `subject`'s batch renders for `agentName`,
    * freeze it, make the render-start durable, re-check authority, and start
-   * the RPC. Returns the render's completion (shared by every assembly that
-   * waits on this batch, vector 25). Returns 'declined' when the turn being
-   * assembled gets nothing of the batch: the agent isn't in its audience,
-   * whether or not another assembly is rendering it (the batch is left as it
-   * is); authority was revoked before the render could start (the batch is
-   * dropped); or the render-start could not be made durable (the batch is
-   * kept for its recovery wake). Returns undefined when the batch was not
-   * this assembly's to decide on: there is none, the subject moved on during
-   * the audience lookup, or the host is stopping.
+   * the RPC. Returns the render's completion, which an assembly that joins
+   * the render waits for too (`joined`: vector 25, one render). Returns
+   * 'declined' when the turn being assembled gets nothing of the batch: the
+   * agent isn't in its audience, whether or not another assembly is
+   * rendering it (the batch is left as it is); authority was revoked before
+   * the render could start (the batch is dropped); the render-start could not
+   * be made durable (the batch is kept for its recovery wake); or, for one of
+   * the turn's causes (`cause`), the batch is gone. Returns undefined when
+   * the batch was not this assembly's to decide on: there is none and it
+   * wasn't a cause, the subject moved on during the audience lookup, or the
+   * host is stopping.
    */
-  private async freeze(subject: string, agentName: string): Promise<{ done: Promise<void>; rendering: Rendering<E> } | 'declined' | undefined> {
+  private async freeze(subject: string, agentName: string, cause = false): Promise<{ done: Promise<void>; rendering: Rendering<E>; joined?: true } | 'declined' | undefined> {
+    if (this.suspended) return undefined;
     const state = this.subjects.get(subject);
-    if (!state || this.suspended) return undefined;
+    // The batch whose wake started this turn is gone: a retraction or a
+    // displacement withdrew it after the wake was dispatched, or another
+    // request consumed it, the batch's one consumption, as the freeze's
+    // withdrawal of the other queued wakes already makes it. Whatever replaced
+    // it is read through its own delivery and wake. (A retry of a turn that
+    // read its batch comes back without the marks of a batch wake, so it
+    // never reaches here as a cause: beginAgentTurn.)
+    if (!state || (!state.batch && !state.rendering)) return cause ? 'declined' : undefined;
     if (state.rendering) {
-      // Another assembly froze this batch; share its outcome (vector 25),
-      // unless this agent isn't in its audience, as below.
+      // Another assembly froze this batch, and the request it assembles
+      // consumes it. An assembly in its audience waits for that render, so
+      // its compile shows the content (vector 25), but the content isn't its
+      // own to read, cause or not; unless this agent isn't in its audience,
+      // as below.
       const rendering = state.rendering;
       const audience = await this.host.audience(rendering.batch.latest);
       if (state.rendering !== rendering) return undefined;
       if (!audience.includes(agentName)) return 'declined';
-      return { done: rendering.done, rendering };
+      return { done: rendering.done, rendering, joined: true };
     }
     const batch = state.batch;
     if (!batch) return undefined;
@@ -824,15 +856,21 @@ export class PushCoalescer<E = unknown> {
         this.host.audit({ kind: 'rendered', subject, eventId: occurrence.eventId, source, empty });
         if (empty) return; // §5.2: nothing happened (blank text included)
         const placement = await this.host.deliver({ ...occurrence, timestamp, content }, content, assemblingFor);
-        rendering.materialized = true;
         if (placement?.deferredId) {
           // Landed in another agent's deferred queue (its turn is alive): still
-          // unread there, so it stays replaceable and withdrawable.
+          // unread there, so it stays replaceable and withdrawable. The turn
+          // being assembled doesn't read it, so it isn't materialized for that
+          // turn, and no request has consumed it. The freeze withdrew the
+          // queue's agent's wake for this batch; give it an ordinary wake, as
+          // any delivery during its turn has, beside any wakes a newer batch
+          // has, so that its next turn reads the content.
           state.occupant = { eventId: occurrence.eventId, placement };
           state.audienceAgent = placement.agent;
+          await this.host.wakeToRead(occurrence, placement.agent);
           return;
         }
         // The materialized occurrence is consumed by the request being assembled.
+        rendering.materialized = true;
         state.history = 'some';
         state.consumedEventId = occurrence.eventId;
         state.occupant = undefined;