fix(deferral): a deferred write with no target names its agent, boot numbers the restored queue 1..N, and a sharded body is injected joined (#286's follow-up)

changelog.d/deferred-writes-286-follow-up.fixed.md

@@ -0,0 +1,20 @@
+- On a host with no primary (`agents: []`), a write with no target goes to the
+  first registered agent, such as an ephemeral run. Deferred, it now names
+  that agent in its queue entry, so it is that agent's as a targeted write is:
+  the agent's own flushes store it, and so does its window's release when it
+  leaves. Before, the entry named no agent, so neither the run's flushes nor
+  the run's end took it: after the run it was still queued, unstored, and in
+  the recovery file when the queue was persisted.
+- Boot recovery numbers the restored deferred-write queue 1, 2, 3, … in the
+  order it restores it, and later deferrals are numbered on from there.
+  Before, it kept each entry's `seq` and numbered only the entries without a
+  valid one, after the largest, so the counter went on from the largest valid
+  `seq`: one at `Number.MAX_SAFE_INTEGER`, which boot accepts, gave every
+  later deferral the same `seq`, 2^53. A recovery file whose `seq`s are all
+  valid isn't rewritten at boot for the new numbers; the queue's next write
+  carries them.
+- At a tool boundary, a deferred message that the strategy stores as more than
+  one message (a sharded body) is now injected as later compiles render it,
+  its shards joined. Before, it was injected as given, so a body of more than
+  one text block, or with a non-text block after a text block, reached the
+  stream in a different shape from later requests.

src/framework.ts

@@ -582,6 +582,8 @@ interface DeferredWrite {
   participant: string;
   content: ContentBlock[];
   metadata?: MessageMetadata;
+  /** The agent the entry is for. Absent means the primary: addMessage
+   *  records any other agent it defers a write for. */
   forAgent?: string;
 }
 const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq;
@@ -595,10 +597,12 @@ type InboundResponder = {
 
 /** A drained entry as writeDrainedInOrder wrote it: the id its store gave it
  *  (empty when addMessage deferred it again; a sharded body's first shard)
- *  and what the store holds of it. For an entry whose write threw after
- *  storing it, found in the store and counted as written, the content is
- *  what landed, as later compiles render it, which for a sharded body can be
- *  its first shards alone. */
+ *  and the content to carry for it: as given, or for a sharded body read
+ *  back from its store, its shards joined as later compiles render them.
+ *  For an entry whose write threw after storing it, found in the store and
+ *  counted as written, it is what landed, joined the same way, which for a
+ *  sharded body can be its first shards alone. A store that can't be read
+ *  back (storedCopiesOf logs it) leaves a sharded body as given. */
 interface DrainedWrite {
   id: MessageId;
   content: ContentBlock[];
@@ -2167,7 +2171,9 @@ export class AgentFramework {
   private unackedDeferredWrites: DeferredWrite[] = [];
   /** Monotonic order stamp for deferred writes: the durable queue is always
    *  written, restored, drained and flushed in `seq` order, whatever the
-   *  pending/un-acked split — a re-deferred entry keeps its place. */
+   *  pending/un-acked split — a re-deferred entry keeps its place. Boot
+   *  recovery numbers the restored queue 1..N in that order, and the stamp
+   *  goes on from N (restorePersistedDeferredWrites). */
   private deferredSeq = 0;
   /**
    * Per target agent: the store's message count when the OLDEST currently
@@ -2177,8 +2183,9 @@ export class AgentFramework {
    */
   private deferredScanFrom = new Map<string, number>();
   private deferredMessages: Array<{
-    /** Durable identity for per-message flush acknowledgement: a flush
-     *  interrupted by a crash replays only the messages not yet acked. */
+    /** Durable identity, stamped on the stored message as its
+     *  deferredWriteId: a flush interrupted by a crash replays its batch at
+     *  the next boot, which skips each entry that already landed by it. */
     id: string;
     /** Order stamp (see deferredSeq). */
     seq: number;
@@ -2844,9 +2851,11 @@ export class AgentFramework {
       const hostMode = framework.readHostMode();
       // Deferred writes are recovered REGARDLESS of the mode flag: a crash
       // between a resume's flag clear and the end of its flush must not
-      // strand accepted messages. Per-message acks mean only the remainder
-      // replays. Serving boot with a remainder → flush it right now (no turn
-      // is alive yet, so this is a safe boundary).
+      // strand accepted messages. A flush acknowledges its batch once, after
+      // the chronicle syncs, and boot skips each entry that already landed
+      // (by its deferredWriteId), so only the remainder replays (see
+      // flushDeferredWrites). Serving boot with a remainder → flush it right
+      // now (no turn is alive yet, so this is a safe boundary).
       const restoredWrites = framework.restorePersistedDeferredWrites();
       if (!hostMode?.quiesced && restoredWrites > 0) {
         console.error(
@@ -5800,11 +5809,11 @@ export class AgentFramework {
       // The windows of forks and runs deregistered during the window are
       // released first, each landing its queued writes and passing on its
       // unread failed replies, as at any release. Then flush context writes
-      // deferred by the quiesce window
-      // (module events, api message.send) — with no turn alive there is no
-      // other flush point. Per-message try/catch and per-message durable ack
-      // (see flushDeferredWrites); nothing here may reach the plane reopen
-      // below.
+      // deferred by the quiesce window (module events, api message.send) —
+      // with no turn alive there is no other flush point. The batch is
+      // stored in queue order and acknowledged once, after the chronicle
+      // syncs (see flushDeferredWrites); nothing here may reach the plane
+      // reopen below.
       this.runReleasesAfterResume();
       await this.flushDeferredWrites('resume');
       this.persistHostMode(null);
@@ -6450,41 +6459,35 @@ export class AgentFramework {
       seen.add(id);
       kept.push({ ...m, id });
     }
-    // A seq is a positive whole number no larger than
-    // Number.MAX_SAFE_INTEGER, the entry's place in arrival order.
-    // An entry without one (a file that predates seq, which is in write
-    // order already) or with any other value (damage or a hand edit:
-    // JSON's 1e400 reads as Infinity) is numbered after every valid seq, in
-    // file order, so the counter stays finite and the rest keep their order.
+    // A seq says only where its entry came in arrival order. The queue is
+    // restored in the order the file gives: entries whose seq is a positive
+    // whole number up to Number.MAX_SAFE_INTEGER by seq, then, in file
+    // order, any without one (a file that predates seq, which is in write
+    // order already) or with another value (damage or a hand edit: JSON's
+    // 1e400 reads as Infinity). Then it is numbered 1..N in that order, so
+    // the counter starts again from N at every boot, wherever the file's
+    // seqs had got to.
     const validSeq = (seq: unknown): seq is number => Number.isSafeInteger(seq) && (seq as number) > 0;
-    let lastSeq = 0;
-    for (const m of kept) if (validSeq(m.seq)) lastSeq = Math.max(lastSeq, m.seq);
-    let renumbered = 0;
-    const candidates: DeferredWrite[] = kept
-      .map((m) => {
-        if (validSeq(m.seq)) return { ...m, seq: m.seq };
-        renumbered++;
-        if (m.seq !== undefined) {
-          // A value that isn't a number is named by its type: it could be
-          // any size.
-          const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`;
-          console.error(
-            `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` +
-            `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`,
-          );
-        }
-        return { ...m, seq: ++lastSeq };
-      })
-      .sort(bySeq);
-    this.deferredSeq = Math.max(this.deferredSeq, lastSeq);
+    const unnumbered = kept.filter((m) => !validSeq(m.seq));
+    for (const m of unnumbered) {
+      if (m.seq === undefined) continue;
+      // A value that isn't a number is named by its type: it could be
+      // any size.
+      const seq = typeof m.seq === 'number' ? String(m.seq) : m.seq === null ? 'null' : `of type ${typeof m.seq}`;
+      console.error(
+        `[host-mode] a recovered deferred context write (${m.id}) has seq ${seq}, not a positive ` +
+        `whole number up to Number.MAX_SAFE_INTEGER; it is queued after the others`,
+      );
+    }
+    const ordered = [...kept.filter((m) => validSeq(m.seq)).sort((a, b) => (a.seq as number) - (b.seq as number)), ...unnumbered];
     // Exactly-once: a message that landed before its ack was written is
     // already in the store under its deferredWriteId — do not replay it.
     // The scan covers every slot the interrupted batch could occupy (from
     // the receipt's per-agent position), or the whole store when the
     // receipt predates the position field.
     let landed: Set<string>;
     try {
-      landed = this.landedDeferredWriteIds(new Set(candidates.map((m) => m.id)), scanFrom);
+      landed = this.landedDeferredWriteIds(new Set(ordered.map((m) => m.id)), scanFrom);
     } catch (error) {
       throw new Error(
         `the deferred context writes (${file ?? DEFERRED_WRITES_ID}) can't be checked for writes that already landed: ` +
@@ -6494,7 +6497,8 @@ export class AgentFramework {
         { cause: error },
       );
     }
-    const restored = candidates.filter((m) => !landed.has(m.id));
+    const restored: DeferredWrite[] = ordered.filter((m) => !landed.has(m.id)).map((m, i) => ({ ...m, seq: i + 1 }));
+    this.deferredSeq = Math.max(this.deferredSeq, restored.length);
     if (landed.size > 0) {
       console.error(
         `[host-mode] ${landed.size} deferred context write(s) had already landed before their ack ` +
@@ -6505,8 +6509,12 @@ export class AgentFramework {
     this.deferredWritesPersisted = true;
     // Rewrite the durable queue without the already-landed entries so a
     // second interrupted boot does not re-scan them, without the malformed
-    // and repeated ones, and with the seqs given here.
-    if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || renumbered > 0) this.persistDeferredWrites();
+    // and repeated ones, and with a seq for each entry that came without a
+    // valid one, so the next boot doesn't name it again. The numbers given
+    // here aren't a reason to rewrite it: every write of the queue writes
+    // it whole, with them, and a boot before that numbers the same entries
+    // the same way.
+    if (landed.size > 0 || malformed.length > 0 || duplicates > 0 || unnumbered.length > 0) this.persistDeferredWrites();
     return restored.length;
   }
 
@@ -10588,9 +10596,10 @@ export class AgentFramework {
             // Stored in queue order, cut at the first that can't be stored: it
             // and what follows wait for the turn's end, and only what was
             // stored is injected (the round's results go on either way). A
-            // write that threw after storing counts as stored, and what landed
-            // of it is what's injected, so the request carries what the store
-            // holds (writeDrainedInOrder).
+            // sharded body read back from its store is injected as later
+            // compiles render it, its shards joined, and a write that threw
+            // after storing, which counts as stored, as what landed of it
+            // (writeDrainedInOrder).
             const stored = this.writeDrainedInOrder(deferred, `tool boundary for ${agent.name}`, (msg) =>
               agent.getContextManager().addMessage(msg.participant, msg.content, withDeferredWriteId(msg.metadata, msg.id)));
             for (const [msg, { id, content }] of stored) {
@@ -19360,7 +19369,13 @@ export class AgentFramework {
       if (opts?.fromQueue) {
         this.unackedDeferredWrites = this.unackedDeferredWrites.filter((m) => m.id !== id);
       }
-      this.deferredMessages.push({ id, seq: opts?.fromQueue?.seq ?? ++this.deferredSeq, participant, content, metadata, forAgent: opts?.forAgent });
+      // An entry with no target is the primary's. A write with no target that
+      // went to another agent (on a host with no primary, the first one
+      // registered, such as an ephemeral run) names that agent, so the
+      // agent's flushes and its window's release take the entry as they take
+      // a targeted one.
+      const forAgent = opts?.forAgent ?? (agent.name === this.primaryAgentName ? undefined : agent.name);
+      this.deferredMessages.push({ id, seq: opts?.fromQueue?.seq ?? ++this.deferredSeq, participant, content, metadata, forAgent });
       if (this.quiesced || this.deferredWritesPersisted) this.persistDeferredWrites();
       if (opts?.placement) { opts.placement.agent = agent.name; opts.placement.deferredId = id; }
       return '' as MessageId; // Deferred — flushed at the target's next boundary
@@ -19481,13 +19496,20 @@ export class AgentFramework {
    * of the body isn't stored. (Requeued, the
    * landed shards would be stored twice, and only the rest could not be
    * appended later without splitting the body.) Boot's scan for landed
-   * writes counts a landed shard the same way.
+   * writes counts a landed shard the same way. When the store can't be read
+   * back (storedCopiesOf logs it), nothing is found, and the entry is put
+   * back as one that couldn't be stored: if it had landed, a later flush
+   * stores it again.
    *
    * Returns what was written (stored, or deferred again by addMessage, whose
    * id is then empty), in queue order: for one target, a prefix of its
-   * entries. `onLanded` hears an entry whose write threw after storing it,
-   * with the id it landed under, as soon as it's found: before the writes
-   * after it, so whatever it records keeps their order.
+   * entries. Each comes with its content: as given, unless the store took it
+   * as more than one message (a sharded body) and can be read back, when its
+   * copies are returned joined, as later compiles render them (asCompiled),
+   * or its write threw after storing it (above). `onLanded` hears an entry
+   * whose write threw after storing it, with the id it landed under, as soon
+   * as it's found: before the writes after it, so whatever it records keeps
+   * their order.
    */
   private writeDrainedInOrder(
     entries: DeferredWrite[],
@@ -19507,7 +19529,15 @@ export class AgentFramework {
       const agent = this.deliveryAgent(msg.forAgent);
       const before = agent ? this.storedCountOf(agent) : undefined;
       try {
-        written.set(msg, { id: write(msg), content: msg.content });
+        const id = write(msg);
+        // A body the store took as more than one message (a sharded body) is
+        // returned as later compiles render it: its copies, joined
+        // (asCompiled). The store is read for them only when it grew by more
+        // than one message.
+        const copies = agent && before !== undefined && (this.storedCountOf(agent) ?? before) - before > 1
+          ? this.storedCopiesOf(agent, before, msg.id)
+          : [];
+        written.set(msg, { id, content: copies.length > 1 ? asCompiled(copies) : msg.content });
       } catch (error) {
         const landed = agent && before !== undefined ? this.storedCopiesOf(agent, before, msg.id) : [];
         if (landed.length > 0) {
@@ -19545,7 +19575,7 @@ export class AgentFramework {
         .filter((m) => m.metadata?.deferredWriteId === deferredWriteId)
         .map((m) => ({ id: m.id, content: m.content, ...(m.bodyGroupId !== undefined ? { bodyGroupId: m.bodyGroupId } : {}) }));
     } catch (error) {
-      console.error(`[deferral] could not read ${agent.name}'s store for a deferred write that threw:`, error);
+      console.error(`[deferral] could not read back what deferred write ${deferredWriteId} stored in ${agent.name}'s store:`, error);
       return [];
     }
   }