F21 (#815): persisted backend submit queue mirror + reload re-arm + bounded drain

app/harness/HarnessHost.tsx

@@ -25,6 +25,7 @@ import { ember, teal } from '../../lib/palette';
 import {
   createDefaultSessionStore,
   createEmptySession,
+  appendMessage,
   type SessionSnapshot,
   type SessionStore,
 } from '../../lib/sessionStore';
@@ -52,6 +53,14 @@ import {
   discardPendingModelChange,
 } from '../../lib/harnessHostModelPersist';
 import { paintQuotaAfterRebuild, tryLocalSave } from '../../lib/hostQuotaError';
+import {
+  TURN_QUEUE_DRAIN_MAX_ATTEMPTS,
+  queueAppend,
+  queueOf,
+  queueRestoreHead,
+  rearmQueueFromMirror,
+  removeQueuedText,
+} from '../../lib/turnQueue';
 import {
   AUTO_CONTINUE_PROMPT,
   migrateAutoContinueFlag,
@@ -223,6 +232,16 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
    * operator submit. Not persisted.
    */
   const didAutoContinueBySessionRef = useRef(new Map<string, boolean>());
+  /**
+   * backend-agents F21 (plan #815) — per-session failed-queue-start attempts.
+   * A persisted queue item whose POST /api/turns start failed (non-durable
+   * error: pre-header network/5xx/subscribe-fail — never a server-side run)
+   * re-promotes on a later poll tick; this in-memory counter bounds those
+   * host-side retries per session. Cleared when a queue item durably starts
+   * and on give-up (drop-with-paint resets the budget). A reload starts a
+   * fresh budget (the mirror re-arms with a fresh 5).
+   */
+  const drainAttemptsRef = useRef(new Map<string, number>());
   /**
    * Plan #813 (E19) — SSE frames **this JS heap** applied for the current
    * `turnRunId`. Null after F5 / adopt / switch (ring rebuilt from Blob).
@@ -274,6 +293,19 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
         windowStart,
       });
       ringWindowStartRef.current = start;
+      // ── backend-agents F21 (plan #815): reload hydration ──
+      // A default `hydrateMessages` clear wipes the Wasm submit FIFO; re-arm
+      // it from the persisted mirror (`session.queue`). Guards inside:
+      // skips when the Wasm queue is non-empty (live FIFO — never
+      // double-enqueues) and on any insert reject (fail-closed; the items
+      // stay in the mirror). Reverse-order `queuedInsertFront` rebuilds the
+      // original FIFO order. `inv_queued_count` parity holds: the mirror was
+      // exactly the set the previous heap drained from.
+      try {
+        rearmQueueFromMirror(bridge, session);
+      } catch {
+        /* torn-down bridge / stub without queue exports */
+      }
       return start;
     },
     [],
@@ -490,6 +522,19 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
       const attaching = attach != null;
       const sendWhileRunning =
         opts?.attach == null && attaching && (prompt ?? '').trim().length > 0;
+      // ── backend-agents F21 (plan #815): submit while a run is live ──
+      // The prompt did NOT start a turn (it joins the Wasm band as a queued
+      // follow-up); persist it into the mirror so it survives a reload. The
+      // drain-start reconcile below removes it again when its own turn is
+      // accepted. Host-known items only (band-internal enqueues are not
+      // host-observable without a protocol bump — documented residual).
+      if (sendWhileRunning) {
+        const liveNow = sessionRef.current;
+        const p = (prompt ?? '').trim();
+        if (p && !(liveNow.queue ?? []).includes(p)) {
+          persist(queueAppend(liveNow, p));
+        }
+      }
       const modelId = bridge.getSelectedModel();
       if (!attaching && !modelId) {
         setHostNote('No model selected — catalog empty, failed to load, or not granted.');
@@ -548,6 +593,20 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
       setHostNote(null);
 
       try {
+        // ── backend-agents F21 (adversarial #901 Major L1) ──
+        // Strip the drained prompt from the mirror BEFORE runHarnessTurn so
+        // onTurnStarted → onSessionPatch cannot persist the in-flight prompt.
+        // A crash between accept and terminal would otherwise re-arm it and
+        // double-POST on the next Ready. Attach / send-while-running leaves
+        // the mirror alone (that call did not start this prompt's turn).
+        const pendingText = (prompt ?? '').trim();
+        const drainingQueued =
+          !attaching &&
+          pendingText.length > 0 &&
+          (queueOf(sessionRef.current) ?? []).includes(pendingText);
+        if (drainingQueued) {
+          persist(removeQueuedText(sessionRef.current, pendingText));
+        }
         const { result, session: next } = await runHarnessTurn(
           bridge,
           sessionRef.current,
@@ -572,8 +631,54 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
             ...(attach ? { attach } : {}),
           },
         );
+        // ── backend-agents F21 (plan #815): persisted submit-queue reconcile ──
+        // The drain-start strip above already dropped the prompt from the
+        // mirror (so mid-turn patches never carry it). This block only:
+        //   - durable start (result.ok OR blended x-workflow-run-id): reset
+        //     the failed-start budget; mirror already matches.
+        //   - failed start before any durable begin: restore the head
+        //     (queueRestoreHead persist + Wasm queuedInsertFront) and count
+        //     a failed attempt; give-up paints an Error (already stripped).
+        //   - attach / drain-attach: untouched (drainingQueued is false).
+        let reconciled = next;
+        if (!attaching) {
+          // AgentFailure/AgentSuccess blend `turnRunId` from the response
+          // header (post-headers aborts included, adversarial #844); the
+          // legacy chat result never carries one.
+          const resultRunId =
+            'turnRunId' in result && typeof result.turnRunId === 'string'
+              ? result.turnRunId
+              : undefined;
+          if (result.ok || resultRunId !== undefined) {
+            drainAttemptsRef.current.delete(reconciled.id);
+          } else if (drainingQueued) {
+            const attempts =
+              (drainAttemptsRef.current.get(reconciled.id) ?? 0) + 1;
+            if (attempts >= TURN_QUEUE_DRAIN_MAX_ATTEMPTS) {
+              // Give-up: already stripped at drain-start; paint, never silent.
+              drainAttemptsRef.current.delete(reconciled.id);
+              const dropLine = `Queued prompt dropped after ${attempts} failed starts: ${result.error}`;
+              try {
+                bridge.pushMessage(MessageKind.Error, dropLine);
+              } catch {
+                /* torn-down bridge */
+              }
+              // F21 adversarial #901 Minor: persist the Error so F5 is not silent.
+              reconciled = appendMessage(reconciled, 'error', dropLine);
+            } else {
+              // Defer: persist + re-arm the Wasm band head.
+              drainAttemptsRef.current.set(reconciled.id, attempts);
+              reconciled = queueRestoreHead(reconciled, pendingText);
+              try {
+                bridge.queuedInsertFront(pendingText);
+              } catch {
+                /* torn-down bridge — mirror is already restored */
+              }
+            }
+          }
+        }
         if (turnEpochRef.current !== epoch) {
-          persistTurn(next, next.turnStatus !== 'running');
+          persistTurn(reconciled, reconciled.turnStatus !== 'running');
           return;
         }
         // Plan #616 (source #610): fold the LIVE selection into the snapshot before
@@ -582,8 +687,8 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
         // just carries that same truth forward into the snapshot).
         const liveId = bridge.getSelectedModel();
         const folded: SessionSnapshot = liveId
-          ? { ...next, selectedModel: liveId }
-          : next;
+          ? { ...reconciled, selectedModel: liveId }
+          : reconciled;
         // Always persist — including user Stop/cancel (and late abort after a finished
         // stream). Dropping session on signal.aborted left SessionStore behind Wasm:
         // Load earlier / refresh could wipe the cancelled turn from the ring.
@@ -596,6 +701,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
         } else {
           heapAppliedRef.current = null;
         }
+
         // Adversarial #857: Send-while-running that finished the run (`done` /
         // 404 / post-start SSE error) re-POSTs the remapped prompt — C15 409
         // no longer applies. Wasm follow-up was stripped; pushUser paints it.
@@ -1166,6 +1272,8 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
     // Adversarial #844: mark discarded BEFORE remove so a late persistTurn
     // preserve PUT cannot LWW-upsert this row back into the picker.
     discardedSessionIdsRef.current.add(clearedId);
+    // F21: drop the cleared session's failed-drain budget (fresh session).
+    drainAttemptsRef.current.delete(clearedId);
     // INTENTIONAL ack-only (not flushPendingThenRestore). Clear deletes this
     // row. Fold-after-remove resurrects via a new-epoch PUT; fold-before-remove
     // is a wasted PUT then DELETE. New/switch flush; Clear acks. See

lib/agent/turnPersistSeam.test.ts

@@ -502,6 +502,123 @@ describe('createTurnPersistSeam — real B7/B8/B6 persist (backend-agents B13)',
     ]);
   });
 
+  it('copy-forwards session.queue onto worker this-run chunks (F21 adversarial #901)', async () => {
+    const { seam, blobStore } = await makeSeam();
+    const first = await seam.persist({
+      turnRunId: realRunId,
+      deltas: [{ d: 1 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [
+          { id: 'cp_0', role: 'user', text: 'turn-1 user', at: 1 },
+          { id: 'cp_1', role: 'assistant', text: 'turn-1 assistant', at: 2 },
+        ],
+        queue: ['follow-up B', 'follow-up C'],
+      }),
+    });
+    expect(first.ok).toBe(true);
+    if (!first.ok) return;
+    const firstBody = JSON.parse((await blobStore.read(first.objectId!)) ?? 'null') as {
+      queue?: string[];
+    };
+    expect(firstBody.queue).toEqual(['follow-up B', 'follow-up C']);
+
+    const second = await seam.persist({
+      turnRunId: 'wr_0000_2a3b4c5d6e7f',
+      deltas: [{ d: 2 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [
+          { id: 'cp_0', role: 'user', text: 'turn-2 user', at: 1 },
+          { id: 'cp_1', role: 'assistant', text: 'turn-2 assistant', at: 2 },
+        ],
+      }),
+    });
+    expect(second.ok).toBe(true);
+    if (!second.ok) return;
+    const chunk = JSON.parse((await blobStore.read(second.objectId!)) ?? 'null') as {
+      queue?: string[];
+      prev?: string;
+    };
+    expect(chunk.queue).toEqual(['follow-up B', 'follow-up C']);
+    expect(typeof chunk.prev).toBe('string');
+    const parsed = parseCloudSessionSnapshot(chunk, scope.sessionId);
+    expect(parsed?.queue).toEqual(['follow-up B', 'follow-up C']);
+  });
+
+  it('copy-forward drops this-run user prompt from the queue (F21 adversarial #901 HEAD)', async () => {
+    const { seam, blobStore } = await makeSeam();
+    const first = await seam.persist({
+      turnRunId: realRunId,
+      deltas: [{ d: 1 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [
+          { id: 'cp_0', role: 'user', text: 'turn-1 user', at: 1 },
+          { id: 'cp_1', role: 'assistant', text: 'turn-1 assistant', at: 2 },
+        ],
+        queue: ['follow-up B', 'follow-up C'],
+      }),
+    });
+    expect(first.ok).toBe(true);
+    if (!first.ok) return;
+
+    const second = await seam.persist({
+      turnRunId: 'wr_0000_2a3b4c5d6e7f',
+      deltas: [{ d: 2 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [
+          { id: 'cp_0', role: 'user', text: 'follow-up B', at: 1 },
+          { id: 'cp_1', role: 'assistant', text: 'turn-2 assistant', at: 2 },
+        ],
+      }),
+    });
+    expect(second.ok).toBe(true);
+    if (!second.ok) return;
+    const chunk = JSON.parse((await blobStore.read(second.objectId!)) ?? 'null') as {
+      queue?: string[];
+    };
+    expect(chunk.queue).toEqual(['follow-up C']);
+    const parsed = parseCloudSessionSnapshot(chunk, scope.sessionId);
+    expect(parsed?.queue).toEqual(['follow-up C']);
+  });
+
+  it('copy-forward unsets the carrier when this-run user was the last queued item', async () => {
+    const { seam, blobStore } = await makeSeam();
+    const first = await seam.persist({
+      turnRunId: realRunId,
+      deltas: [{ d: 1 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [
+          { id: 'cp_0', role: 'user', text: 'turn-1 user', at: 1 },
+          { id: 'cp_1', role: 'assistant', text: 'turn-1 assistant', at: 2 },
+        ],
+        queue: ['follow-up B'],
+      }),
+    });
+    expect(first.ok).toBe(true);
+    if (!first.ok) return;
+
+    const second = await seam.persist({
+      turnRunId: 'wr_0000_2a3b4c5d6e7f',
+      deltas: [{ d: 2 }],
+      content: JSON.stringify({
+        id: scope.sessionId,
+        messages: [{ id: 'cp_0', role: 'user', text: 'follow-up B', at: 1 }],
+      }),
+    });
+    expect(second.ok).toBe(true);
+    if (!second.ok) return;
+    const chunk = JSON.parse((await blobStore.read(second.objectId!)) ?? 'null') as {
+      queue?: string[];
+    };
+    expect('queue' in chunk).toBe(false);
+    const parsed = parseCloudSessionSnapshot(chunk, scope.sessionId);
+    expect('queue' in (parsed ?? {})).toBe(false);
+  });
+
   it('host-shaped prior that already ends with this turn is not duplicated', async () => {
     const blobStore = new MemoryBlobTranscriptStore();
     const envelopeStore = new MemorySessionStore();

lib/agent/turnPersistSeam.ts

@@ -72,6 +72,7 @@ import type {
   PersistStepSeam,
 } from '../workflows/persistStep';
 import { persistOverlayStatus, stampSnapshotUpdatedAt } from '../workflows/persistStep';
+import { queueWithoutText, sanitizeQueue } from '../turnQueue';
 
 /** Worker-authored envelope clock source for the terminal B8 overlay (LWW). */
 export type OverlayClock = (storedUpdatedAt: number) => number;
@@ -105,13 +106,33 @@ export interface TurnPersistSeamDeps {
 const toMessage = (err: unknown): string =>
   err instanceof Error ? err.message : String(err);
 
+/** F21 queue mirror on a snapshot-shaped body; undefined = no/poisoned carrier. */
+function queueFromBody(body: unknown): string[] | undefined {
+  if (body === null || typeof body !== 'object' || Array.isArray(body)) return undefined;
+  return sanitizeQueue((body as Record<string, unknown>).queue);
+}
+
+/** First non-blank user text on a this-run snapshot (persistStep checkpoint). */
+function firstUserText(
+  messages: Array<{ role: string; text: string }>,
+): string | undefined {
+  for (const m of messages) {
+    if (m.role !== 'user') continue;
+    const t = m.text.trim();
+    if (t) return t;
+  }
+  return undefined;
+}
+
 /** This-run snapshot + optional `prev`/`depth`; non-snapshot test bodies keep stamp-only. */
 function buildThisRunChunk(opts: {
   content: string;
   sessionId: string;
   updatedAt: number;
   prev: string | undefined;
   depth: number | undefined;
+  /** Prior blob's sanitized `queue` (copy-forward when this-run content omits it). */
+  priorQueue?: string[];
 }): string | null {
   let parsed: unknown;
   try {
@@ -133,6 +154,18 @@ function buildThisRunChunk(opts: {
   };
   if (opts.prev) rec.prev = opts.prev;
   if (opts.depth !== undefined) rec.depth = opts.depth;
+  // F21 adversarial #901: worker this-run chunks must copy-forward the
+  // submit-queue mirror (field rides the transcript blob, not meta). Dropping
+  // it lets a cloud adopt wipe a localStorage re-arm. Copy-forward of the
+  // *in-flight* prompt (HEAD Major): persistStep content is `{id, messages}`
+  // so fromContent is always unset; a coalesced host strip PUT cannot beat
+  // B7. Strip this-run's first user text (removeQueuedText semantics) so a
+  // drain that has durably started cannot re-arm itself on F5.
+  const fromContent = queueFromBody(parsed);
+  let queue = fromContent ?? opts.priorQueue;
+  const started = firstUserText(incoming);
+  if (started) queue = queueWithoutText(queue, started);
+  if (queue !== undefined && queue.length > 0) rec.queue = queue;
   return JSON.stringify(rec);
 }
 
@@ -231,6 +264,7 @@ export function createTurnPersistSeam(
       // walking (reconstruct fail-closes the whole blob at 256).
       let chunkPrev: string | undefined;
       let chunkDepth: number | undefined;
+      let priorQueue: string[] | undefined;
       const pointer = stored?.meta?.transcriptPointer;
       if (typeof pointer === 'string' && isObjectIdBoundTo(pointer, scope)) {
         let raw: string | null;
@@ -288,6 +322,7 @@ export function createTurnPersistSeam(
           }
           chunkPrev = pointer;
           chunkDepth = chainLen + 1;
+          priorQueue = queueFromBody(parsed);
         }
       }
 
@@ -297,6 +332,7 @@ export function createTurnPersistSeam(
         updatedAt,
         prev: chunkPrev,
         depth: chunkDepth,
+        priorQueue,
       });
       if (stampedRaw === null) {
         return await failWrite({