fix(deferral): a write the coalescing assembly makes lands after what was deferred for its agent

changelog.d/assembly-writes-in-arrival-order.fixed.md

@@ -0,0 +1,26 @@
+- What was deferred for an agent before a coalesced render or message the
+  agent's turn assembles now lands ahead of it, in the order it arrived. The
+  assembly's write used to be stored at once, ahead of a write still queued
+  for the agent, so a render could be read before the earlier render it
+  builds on: after a render landed in the agent's queue during its own
+  assembly, or after the turn's start couldn't store a queued render. When a
+  queued write can't be stored, the assembled one now waits behind it. What
+  is deferred for the agent while its assembly awaits a render, such as a
+  render another agent's assembly placed in its queue, now lands in that
+  turn's request, rather than at the turn's first tool boundary or its end.
+- A render's wake to read it, given when the render lands in an agent's
+  deferred queue, no longer starts a turn if, when the scheduler comes to
+  start it, the agent has already read the render or the render was
+  retracted.
+- A provider hold no longer keeps two wakes from the same millisecond as
+  one when one of them is a coalesced batch's wake or a wake to read a
+  render, so withdrawing the batch, or dropping the read render's wake,
+  no longer takes the other wake's cause with it.
+- A reply to a `[Continue]` turn that a coalesced batch's wake or a wake to
+  read a render started now keeps its `promptedBy` stamp, so later
+  requests render `[Continue]` before it as they do for other such replies.
+- A render that settles after its turn's assembly has failed, while the turn
+  is still alive, is now deferred like any other write during the turn,
+  rather than stored into the turn after its compile. While the host is
+  quiesced, the assembly's writes wait behind what is queued for the agent,
+  and land when it does.

src/mcpl/push-coalescer.ts

@@ -139,10 +139,11 @@ export interface AssemblyResult {
    *  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. */
+   *  include the agent, a render placed in a deferred queue (another
+   *  agent's, or this agent's own behind a write held there), 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>;
 }
 
@@ -222,9 +223,9 @@ export interface CoalescerHost<E> {
   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>;
+   *  no request consumed, at `deferredId`. It carries no subject, so the
+   *  subject's other wakes, and later withdrawals of them, leave it. */
+  wakeToRead(occurrence: CoalescedOccurrence<E>, agent: string, deferredId?: 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. */
@@ -917,13 +918,17 @@ export class PushCoalescer<E = unknown> {
         const run = state.occupant?.rendered ? state.occupant : undefined;
         const placement = await this.host.deliver({ ...occurrence, timestamp, content }, content, assemblingFor);
         if (placement?.deferredId) {
-          // Landed in another agent's deferred queue (its turn is alive): still
-          // 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.
+          // Landed in a deferred queue: another agent's, whose turn is alive,
+          // or the assembling agent's own, behind a write held there (the
+          // host's bypassDeferralFor), or after its assembly ended, as any
+          // write during its turn. Still 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 (the wake starts no turn
+          // once the agent has read it, or it's removed).
           // The run's renders keep their own records, whatever else an
           // occupant carries.
           let before: Occupant['before'];
@@ -937,7 +942,7 @@ export class PushCoalescer<E = unknown> {
             await this.host.wakeForBatch(state.batch.latest);
           }
           state.audienceAgent = placement.agent;
-          await this.host.wakeToRead(occurrence, placement.agent);
+          await this.host.wakeToRead(occurrence, placement.agent, placement.deferredId);
           return;
         }
         // The materialized occurrence is consumed by the request being

src/mcpl/push-handler.ts

@@ -48,7 +48,9 @@ export interface McplPushEvent {
   targetAgents?: string[];
   /** RFC-006: set when the coalescer delivers; rides the wake it queues. */
   coalescingSubject?: string;
-  /** RFC-006 assembly: materialized for this agent's turn (store directly). */
+  /** RFC-006 assembly: materialized for this agent's turn, stored directly
+   *  while its assembly runs, after what is queued for it (the framework's
+   *  addMessage, bypassDeferralFor). */
   assemblingFor?: string;
   /** Host acceptance time (epoch ms), stamped where the push is admitted. */
   acceptedAt?: number;

src/types/agent.ts

@@ -371,6 +371,10 @@ export interface InferenceRequest {
    *  all of them without materializing content, the turn has no cause left
    *  and does not run. */
   coalescingBatchSubjects?: string[];
+  /** Host-owned: this wake's sole cause is reading this deferred write, a
+   *  coalesced render placed in its agent's queue (wakeToRead). Once the
+   *  agent has read the write, or it was removed, the wake starts no turn. */
+  readsDeferredWrite?: string;
   agentName: string;
   reason: string;
   source: string;

test/deferred-wake-integrity.test.ts

@@ -16,7 +16,10 @@
  *     it (turn-start flush runs before the locus announcement and compile).
  *  2. addMessage defers on turn-alive (activeTurnTokens), not merely
  *     stream-alive — the compile window between dequeue and stream
- *     registration is closed to cross-turn writers.
+ *     registration is closed to cross-turn writers: what they write is
+ *     deferred, and what is deferred before the turn's checkpoint lands as
+ *     the turn's input through its own flushes (flushForStart), unless the
+ *     host is quiesced by then, when it waits for resume().
  *  3. endTurn edge: a message stored at an endTurn tool boundary (collected
  *     for injection but never delivered — the stream is cancelled) is seen by
  *     the follow-up turn its own wake starts, positioned AFTER the turn that

test/deferred-window-only.test.ts

@@ -1,7 +1,10 @@
 /**
  * What the window stores mid-turn is what the live stream carried.
  *
- * A message deferred during a turn is stored at the agent's next tool
+ * A message deferred during a turn's start, until its coalescing assembly
+ * ends, is stored before the turn's checkpoint, in that request, as the
+ * turn-start flush stores what came before. One deferred after that is
+ * stored at the agent's next tool
  * boundary and injected into the resumed stream (hear-while-acting), so the
  * next compile's prefix is the request the turn's later replies were minted
  * under. Two kinds of message can't be injected: one carrying tool blocks
@@ -331,11 +334,18 @@ describe('mid-turn deliveries keep the window what the stream carried', () => {
       pendingAssistantBlocks: Map<string, ContentBlock[]>;
       activeTurnTokens: Map<string, number>;
       activeStreams: Map<string, unknown>;
-      deferredMessages: Array<{ id: string; seq: number; participant: string; forAgent?: string }>;
+      deferredMessages: Array<{ id: string; seq: number; participant: string; forAgent?: string; content: ContentBlock[] }>;
+      unackedDeferredWrites: Array<{ id: string }>;
       deferredWritesPersisted: boolean;
+      quiesced: boolean;
+      assembling: Set<string>;
       drainDeferredFor: (name: string, stopAt?: unknown) => unknown[];
       readdDrained: (entries: unknown[], where: string) => unknown[];
-      addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string; fromQueue?: { id: string; seq: number } }): string;
+      ackDeferredWrites: () => void;
+      addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: {
+        forAgent?: string; fromQueue?: { id: string; seq: number }; bypassDeferralFor?: string;
+        placement?: { agent?: string; messageId?: string; deferredId?: string };
+      }): string;
     };
   }
   const otherCycle = (): ContentBlock[] => [{ type: 'tool_use', id: 'b1', name: 'robot--move', input: {} } as ContentBlock];
@@ -1087,4 +1097,164 @@ describe('mid-turn deliveries keep the window what the stream carried', () => {
     }
     await framework.stop();
   });
+
+  // The coalescing assembly writes past its agent's deferral, in the window
+  // the turn-start flush writes in (bypassDeferralFor). What was deferred
+  // for the agent before such a write came first, so it lands first.
+  const textsOf = (entries: Array<{ content: ContentBlock[] }>): string[] =>
+    entries.map((m) => m.content.map((b) => (b as { text?: string }).text ?? `<${b.type}>`).join(' '));
+
+  it("a write assembled for a turn's start lands after what was deferred for its agent before it, and acknowledges it", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    internals.deferredWritesPersisted = true;
+    internals.activeTurnTokens.set('assistant', 1);
+    internals.assembling.add('assistant');
+    try {
+      assert.equal(internals.addMessage('Antra', text('first')), '', 'deferred: the turn is alive');
+      // Another target's entry: it neither flushes nor holds this agent's write.
+      internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') });
+      const placement: { agent?: string; messageId?: string; deferredId?: string } = {};
+      const id = internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement });
+      assert.notEqual(id, '', 'the assembled write is stored');
+      assert.equal(placement.deferredId, undefined);
+      assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second'], 'what was deferred first lands first');
+      assert.deepEqual(textsOf(internals.deferredMessages), ['elsewhere'], "only the other target's entry is still queued");
+      assert.deepEqual(internals.unackedDeferredWrites, [], 'what the flush stored is acknowledged');
+    } finally {
+      internals.assembling.delete('assistant');
+      internals.activeTurnTokens.delete('assistant');
+      internals.deferredMessages.splice(0);
+    }
+    await framework.stop();
+  });
+
+  it("a write naming its agent's assembly after that assembly has ended is deferred like any write during the turn", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    internals.activeTurnTokens.set('assistant', 1);
+    try {
+      internals.addMessage('Antra', text('first'));
+      // An assembly that throws can leave a render running, which delivers
+      // after the window has closed.
+      const placement: { agent?: string; messageId?: string; deferredId?: string } = {};
+      assert.equal(internals.addMessage('Antra', text('late'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred');
+      assert.ok(placement.deferredId);
+      assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'late'], 'nothing was flushed into the live turn');
+      assert.ok(!windowOf(framework).some((line) => line === 'Antra: first' || line === 'Antra: late'));
+    } finally {
+      internals.activeTurnTokens.delete('assistant');
+    }
+    await framework.stop();
+  });
+
+  it("a write assembled for a turn's start waits behind a deferred write that can't be stored, and lands after it", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    const cm = framework.getAgent('assistant')!.getContextManager();
+    const add = cm.addMessage.bind(cm);
+    let failing = true;
+    (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => {
+      if (failing && (args[1] as Array<{ text?: string }>).some((b) => b.text === 'first')) throw new Error('disk full');
+      return add(...args);
+    };
+    const errors: string[] = [];
+    const error = console.error;
+    console.error = (...args: unknown[]) => { errors.push(args.map(String).join(' ')); };
+    internals.activeTurnTokens.set('assistant', 1);
+    internals.assembling.add('assistant');
+    try {
+      internals.addMessage('Antra', text('first'));
+      const placement: { agent?: string; messageId?: string; deferredId?: string } = {};
+      assert.equal(internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred');
+      assert.ok(placement.deferredId, 'its placement names the queued entry');
+      assert.ok(errors.some((e) => e.includes('assembly flush for assistant') && e.includes('disk full')), errors.join('\n'));
+      assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'second'], 'it waits behind the held write');
+      assert.ok(!windowOf(framework).includes('Antra: second'));
+    } finally {
+      internals.assembling.delete('assistant');
+      internals.activeTurnTokens.delete('assistant');
+      console.error = error;
+    }
+    failing = false;
+    internals.readdDrained(internals.drainDeferredFor('assistant'), 'the turn ends');
+    internals.ackDeferredWrites();
+    assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second'], 'both land, in the order they arrived');
+    await framework.stop();
+  });
+
+  it("a write assembled for a turn's start waits behind what is queued for its agent while the host is quiesced", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    internals.activeTurnTokens.set('assistant', 1);
+    internals.assembling.add('assistant');
+    try {
+      internals.addMessage('Antra', text('first'));
+      // A quiesce that begins while the turn's assembly awaits a render lets
+      // the turn drain: what is queued waits for resume(), and so does what
+      // the assembly writes after it.
+      internals.quiesced = true;
+      const placement: { agent?: string; messageId?: string; deferredId?: string } = {};
+      assert.equal(internals.addMessage('Antra', text('second'), undefined, { bypassDeferralFor: 'assistant', placement }), '', 'deferred');
+      assert.ok(placement.deferredId);
+      assert.deepEqual(textsOf([...internals.deferredMessages].sort((a, b) => a.seq - b.seq)), ['first', 'second']);
+      assert.ok(!windowOf(framework).some((line) => line === 'Antra: first' || line === 'Antra: second'), 'nothing queued is written while quiesced');
+    } finally {
+      internals.quiesced = false;
+      internals.assembling.delete('assistant');
+      internals.activeTurnTokens.delete('assistant');
+    }
+    internals.readdDrained(internals.drainDeferredFor('assistant'), 'resume');
+    internals.ackDeferredWrites();
+    assert.deepEqual(windowOf(framework).slice(-2), ['Antra: first', 'Antra: second']);
+    await framework.stop();
+  });
+
+  it("a write assembled for a turn's start is stored while the host is quiesced when nothing is queued for its agent", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    internals.activeTurnTokens.set('assistant', 1);
+    internals.assembling.add('assistant');
+    internals.quiesced = true;
+    try {
+      // Another target's entry doesn't count as queued for this agent.
+      internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') });
+      assert.notEqual(internals.addMessage('Antra', text('assembled'), undefined, { bypassDeferralFor: 'assistant' }), '', 'stored, as the turn\'s own input');
+      assert.deepEqual(windowOf(framework).slice(-1), ['Antra: assembled']);
+      assert.deepEqual(textsOf(internals.deferredMessages), ['elsewhere'], 'and nothing of the queue was written');
+    } finally {
+      internals.quiesced = false;
+      internals.assembling.delete('assistant');
+      internals.activeTurnTokens.delete('assistant');
+      internals.deferredMessages.splice(0);
+    }
+    await framework.stop();
+  });
+
+  it("stop()'s drain, which writes past each agent's deferral, keeps its entries un-acked until it has stored them", async () => {
+    const framework = await createFramework();
+    const internals = internalsOf(framework);
+    internals.deferredWritesPersisted = true;
+    internals.activeTurnTokens.set('assistant', 1);
+    internals.addMessage('Antra', text('first'));
+    internals.addMessage('Antra', text('second'));
+    internals.activeTurnTokens.delete('assistant');
+    // An entry for a target that isn't registered stays queued through the
+    // drain, so the queue isn't empty while the drain writes.
+    internals.deferredMessages.push({ id: 'ghost-1', seq: 0, participant: 'Antra', forAgent: 'ghost', content: text('elsewhere') });
+    const cm = framework.getAgent('assistant')!.getContextManager();
+    const add = cm.addMessage.bind(cm);
+    const unackedAt: Record<string, string[]> = {};
+    (cm as unknown as { addMessage: typeof add }).addMessage = (...args) => {
+      const written = (args[1] as Array<{ text?: string }>).map((b) => b.text).join(' ');
+      unackedAt[written] = internals.unackedDeferredWrites.map((m) => m.id);
+      return add(...args);
+    };
+    const secondId = internals.deferredMessages.find((m) => textsOf([m])[0] === 'second')!.id;
+    const error = console.error;
+    console.error = () => {};
+    try { await framework.stop(); } finally { console.error = error; }
+    assert.ok(unackedAt.first?.includes(secondId), `'second' is still un-acked while 'first' is stored: ${JSON.stringify(unackedAt)}`);
+    assert.ok(unackedAt.second?.includes(secondId), `and while 'second' itself is stored: ${JSON.stringify(unackedAt)}`);
+  });
 });

test/mcpl-coalescer-render-run.test.ts

@@ -18,7 +18,8 @@ interface Options {
  *  'other', an agent whose turn is alive, until it's removed, read, or gone,
  *  whichever agent assembled it. CoalescerHost leaves placement to the host,
  *  so this is one it allows; the framework stores the assembling agent's own
- *  render directly, which mcpl-coalescing's tests run. Its removal, like the
+ *  render directly while its assembly runs, after what is queued for it,
+ *  which mcpl-coalescing's tests run. Its removal, like the
  *  framework's, doesn't refuse a message a request has read. Each render says
  *  which notices it covers, as a notice-only server's does. */
 function makeHost(options: Options = {}) {