fix(mcpl): withdraw a turn whose deferred batches give it nothing to read

changelog.d/coalescing-declined-freeze.fixed.md

@@ -0,0 +1 @@
+- A turn woken only by deferred coalescing batches is no longer started when none of them has anything for it to read. It used to run, calling the model although no new input had arrived for it, when the assembly found a batch's authority revoked (before or after its render-start commit), when the render-start commit failed, or when it found that the agent had left the batch's audience since the batch woke it. Such a turn is now withdrawn before inference, as a turn whose batches all rendered nothing already was, and traces `turn-withdrawn`. A batch whose render-start commit failed is still kept, and its recovery wake delivers its fallback once storage recovers. The stderr line for a withdrawn turn now reads `[coalescing] <agent>: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`. It said `deferred render produced nothing`, which is untrue where no render ran.

src/framework.ts

@@ -12990,19 +12990,21 @@ export class AgentFramework {
       }
     }
 
-    // A turn whose only causes were deferred batches, all of which rendered
-    // nothing (empty or blank render, §5.2; cancelled; revoked), 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.
+    // 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;
     if (assembly && batchSubjects?.length && assembly.materialized.size === 0
       && batchSubjects.every((s) => assembly.settled.has(s))) {
-      console.error(`[coalescing] ${agent.name}: deferred render produced nothing; turn not started (no wake cause left)`);
+      console.error(`[coalescing] ${agent.name}: its deferred batches gave this turn nothing to read; turn not started (no wake cause left)`);
       this.emitTrace({ type: 'mcpl:coalescing', kind: 'turn-withdrawn', agentName: agent.name, subjects: batchSubjects });
       // Release the turn and land anything deferred while assembly awaited
       // the render, as a torn-down turn would.

src/mcpl/push-coalescer.ts

@@ -129,11 +129,15 @@ interface Rendering<E> {
 
 /** What an assembly did, per subject, for the turn being assembled. */
 export interface AssemblyResult {
-  /** Subjects whose render this assembly waited on to completion. */
+  /** Subjects this assembly finished with: it waited on their render to
+   *  completion, or freeze() declined them for this agent. */
   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, or revoked at response. */
+   *  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. */
   materialized: Set<string>;
 }
 
@@ -674,7 +678,8 @@ export class PushCoalescer<E = unknown> {
       // 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));
-      if (started) {
+      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);
@@ -690,27 +695,42 @@ 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) or undefined when nothing was started.
+   * 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.
    */
-  private async freeze(subject: string, agentName: string): Promise<{ done: Promise<void>; rendering: Rendering<E> } | undefined> {
+  private async freeze(subject: string, agentName: string): Promise<{ done: Promise<void>; rendering: Rendering<E> } | 'declined' | undefined> {
     const state = this.subjects.get(subject);
     if (!state || this.suspended) return undefined;
     if (state.rendering) {
-      // Another assembly froze this batch; share its outcome (vector 25).
+      // Another assembly froze this batch; share its outcome (vector 25),
+      // unless this agent isn't in its audience, as below.
       const rendering = state.rendering;
-      return (await this.host.audience(rendering.batch.latest)).includes(agentName) && state.rendering === rendering ? { done: rendering.done, rendering } : undefined;
+      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 };
     }
     const batch = state.batch;
     if (!batch) return undefined;
     // The audience lookup may spawn a fork. Under the lock no admission can
     // move the subject on meanwhile; the identity check guards the host.
     const audience = await this.host.audience(batch.latest);
-    if (!audience.includes(agentName) || state.batch !== batch || state.rendering) return undefined;
+    if (state.batch !== batch || state.rendering) return undefined;
+    // Not this agent's batch, or no longer: an agent woken for it can have left
+    // its audience since. This turn has nothing of it to show. The batch stays
+    // pending; nothing here wakes its current audience.
+    if (!audience.includes(agentName)) return 'declined';
     state.batch = undefined;
     this.host.cancelWake(subject);
     if (!this.host.authorized(batch.latest)) {
       this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId });
-      return undefined;
+      return 'declined';
     }
     const rendering: Rendering<E> = { batch, cancelled: false, done: Promise.resolve() };
     state.rendering = rendering;
@@ -728,7 +748,7 @@ export class PushCoalescer<E = unknown> {
       // unrelated traffic happens by. Re-wake with backoff, so a persistent
       // storage outage does not become a tight loop of model turns.
       this.scheduleRecovery(subject, state);
-      return undefined;
+      return 'declined';
     }
     this.clearRecovery(subject);
     if (this.suspended) { state.rendering = undefined; state.batch = batch; return undefined; }
@@ -738,7 +758,7 @@ export class PushCoalescer<E = unknown> {
     if (!this.host.authorized(batch.latest)) {
       this.host.audit({ kind: 'revoked', subject, eventId: batch.latest.eventId });
       state.rendering = undefined;
-      return undefined;
+      return 'declined';
     }
     rendering.done = this.render(subject, state, rendering, agentName);
     return { done: rendering.done, rendering };

test/helpers/coalescing-fixture.ts

@@ -18,7 +18,8 @@ export const TS = '2026-09-30T00:00:00Z';
 
 /**
  * How long the fixture's waits run before they fail: send() for a request's
- * reply, answered() for the host's answers, eventually() for its condition.
+ * reply, answered() for the host's answers, eventually() for its condition,
+ * settles() for the work it's given.
  * Each is a hang guard: it turns a wait that never ends into a failure
  * instead of a hung run, and it isn't a check on how fast the host works. So
  * it's set wide, and the same for every wait: a loaded host's stall needn't
@@ -145,6 +146,21 @@ export async function eventually(predicate: () => boolean, what = 'condition'):
   while (!predicate()) { assert(performance.now() < deadline, `${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`); await new Promise((r) => setTimeout(r, 10)); }
 }
 
+/**
+ * Awaits `work`, failing with `what` if it hasn't settled within
+ * REPLY_DEADLINE_MS. CI runs without a test timeout, so a wait that never
+ * ends would hold the whole run, and this turns it into a failure that names
+ * what was awaited. When it fires, the work it waited on keeps running, and
+ * the test's close then runs under it.
+ */
+export async function settles<T>(work: Promise<T>, what: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const late = new Promise<never>((_, reject) => {
+    timer = setTimeout(() => reject(new Error(`${what} did not settle within ${REPLY_DEADLINE_MS / 1000} s`)), REPLY_DEADLINE_MS);
+  });
+  try { return await Promise.race([work, late]); } finally { clearTimeout(timer); }
+}
+
 
 /** Simulate a crash: stop without the snapshot flush a clean stop performs. */
 export async function crash(f: Awaited<ReturnType<typeof fixture>>): Promise<void> {