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

app/harness/HarnessHost.tsx

@@ -52,6 +52,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 +231,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 +292,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 +521,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 +592,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 +630,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);
+              try {
+                bridge.pushMessage(
+                  MessageKind.Error,
+                  `Queued prompt dropped after ${attempts} failed starts: ${result.error}`,
+                );
+              } catch {
+                /* torn-down bridge */
+              }
+            } 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 +686,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 +700,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 +1271,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,50 @@ 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('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 { sanitizeQueue } from '../turnQueue';
 
 /** Worker-authored envelope clock source for the terminal B8 overlay (LWW). */
 export type OverlayClock = (storedUpdatedAt: number) => number;
@@ -105,13 +106,21 @@ 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);
+}
+
 /** 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 +142,12 @@ function buildThisRunChunk(opts: {
   };
   if (opts.prev) rec.prev = opts.prev;
   if (opts.depth !== undefined) rec.depth = opts.depth;
+  // F21 adversarial #901 Major L1: worker this-run chunks must copy-forward
+  // the submit-queue mirror. The field rides the transcript blob (not meta);
+  // dropping it here lets a cloud adopt wipe a localStorage re-arm.
+  const fromContent = queueFromBody(parsed);
+  const queue = fromContent ?? opts.priorQueue;
+  if (queue !== undefined && queue.length > 0) rec.queue = queue;
   return JSON.stringify(rec);
 }
 
@@ -231,6 +246,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 +304,7 @@ export function createTurnPersistSeam(
           }
           chunkPrev = pointer;
           chunkDepth = chainLen + 1;
+          priorQueue = queueFromBody(parsed);
         }
       }
 
@@ -297,6 +314,7 @@ export function createTurnPersistSeam(
         updatedAt,
         prev: chunkPrev,
         depth: chunkDepth,
+        priorQueue,
       });
       if (stampedRaw === null) {
         return await failWrite({

lib/sessionRepository.test.ts

@@ -1644,3 +1644,71 @@ describe('createHttpSessionRepository — envelope carrier (phase 0 #515)', () =
     expect(res).toEqual({ action: 'notfound' });
   });
 });
+
+describe('backend-agents F21 — persisted submit-queue mirror (plan #815)', () => {
+  const idA = '11111111-1111-4111-8111-111111111111';
+  const UPLOAD_URL = 'https://blob.example/upload';
+
+  it('trimForCloudPut folds the queue mirror into the record body; omits when absent', () => {
+    const out = trimForCloudPut({
+      id: 's',
+      updatedAt: 1,
+      messages: [],
+      queue: ['one', 'two'],
+    });
+    expect(out.queue).toEqual(['one', 'two']);
+    const bare = trimForCloudPut({ id: 's', updatedAt: 1, messages: [] });
+    expect('queue' in bare).toBe(false);
+  });
+
+  it('trimForCloudPut drops a poisoned mirror (never PUTs junk prompts)', () => {
+    const out = trimForCloudPut({
+      id: 's',
+      updatedAt: 1,
+      messages: [],
+      queue: 'not an array' as unknown as string[],
+    });
+    expect('queue' in out).toBe(false);
+  });
+
+  it('parseCloudSessionSnapshot restores the queue mirror; poison / empty stays unset', () => {
+    const out = parseCloudSessionSnapshot({
+      id: 'sess_x',
+      updatedAt: 1,
+      messages: [],
+      queue: [' alpha ', 42, ''],
+    });
+    expect(out?.queue).toEqual(['alpha']);
+
+    const empty = parseCloudSessionSnapshot({
+      id: 'sess_x',
+      updatedAt: 1,
+      messages: [],
+      queue: [],
+    });
+    expect('queue' in (empty ?? {})).toBe(false);
+
+    const junk = parseCloudSessionSnapshot({
+      id: 'sess_x',
+      updatedAt: 1,
+      messages: [],
+      queue: 'nope',
+    });
+    expect('queue' in (junk ?? {})).toBe(false);
+  });
+
+  it('overlayEnvelopeMeta leaves the queue mirror untouched (meta is not its carrier)', () => {
+    const transcript: SessionSnapshot = {
+      id: 's',
+      updatedAt: 1,
+      messages: [],
+      queue: ['p1'],
+    };
+    const over = overlayEnvelopeMeta(transcript, {
+      transcriptPointer: 'tx_1',
+      turnStatus: 'running',
+    });
+    expect(over.queue).toEqual(['p1']);
+  });
+});
+