feat(harness): E19 attach handshake — Blob hydrate + stream attach

app/harness/HarnessHost.tsx

@@ -12,7 +12,8 @@ import {
 } from '../../lib/harnessChat';
 import { resetHarnessImageSession } from '../../lib/harnessImages';
 import { resetHarnessMathSession } from '../../lib/harnessMath';
-import { decideDetach, shouldAbortReader, abortReasonFor, decideDetachPersist, putPreservedTurn, shouldApplyMintBind, shouldSetHostTurnNote } from '../../lib/detachTurn';
+import { decideDetach, shouldAbortReader, abortReasonFor, decideDetachPersist, putPreservedTurn, shouldApplyMintBind, shouldSetHostTurnNote, isDetachAbort } from '../../lib/detachTurn';
+import { decideHotResume, decideSendAttach, shouldPaintAttachFollowUpNote, shouldRepostAttachFollowUp, shouldSkipAttachHotResume, ATTACH_FOLLOW_UP_NOTE, type HeapApplied } from '../../lib/turnAttach';
 import {
   HarnessBridge,
   HARNESS_PROTOCOL_VERSION,
@@ -60,6 +61,9 @@ import HarnessLoading from './HarnessLoading';
 
 type Phase = 'loading' | 'ready' | 'error';
 
+type RunPromptAttach = { runId: string; startIndex: number; dedup: boolean };
+type RunPromptOpts = { pushUser?: boolean; attach?: RunPromptAttach };
+
 type DvuiModule = {
   dvui: (
     canvas: string | HTMLCanvasElement,
@@ -199,6 +203,15 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
   const pollRef = useRef<number | null>(null);
   const abortRef = useRef<AbortController | null>(null);
   const inflightRef = useRef(false);
+  /**
+   * Plan #813 (E19) — SSE frames **this JS heap** applied for the current
+   * `turnRunId`. Null after F5 / adopt / switch (ring rebuilt from Blob).
+   * Hot resume reads this, never envelope `C`.
+   */
+  const heapAppliedRef = useRef<HeapApplied | null>(null);
+  const runPromptRef = useRef<(prompt: string, opts?: RunPromptOpts) => Promise<void>>(
+    async () => {},
+  );
   /** Bumped on detach so a late runPrompt persist cannot clobber a switched session. */
   const turnEpochRef = useRef(0);
   /**
@@ -309,6 +322,8 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
       if (bridge) {
         hydrateRingWindow(bridge, merged, latestRingStart(merged.messages.length));
       }
+      // Ring rebuilt from Blob/local — this heap has not applied the stream.
+      heapAppliedRef.current = null;
     },
     [writeLocalSession, hydrateRingWindow],
   );
@@ -348,14 +363,31 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
     });
   }, []);
 
+  /**
+   * Plan #813 — cold attach after the ring was rebuilt from Blob/local
+   * (boot / adopt / switch-back). Always `startIndex=0` + dedup. No-ops when
+   * not `running` or a turn is already inflight.
+   */
+  const kickColdAttach = useCallback(() => {
+    if (inflightRef.current) return;
+    const s = sessionRef.current;
+    if (s.turnStatus !== 'running' || !s.turnRunId) return;
+    heapAppliedRef.current = null;
+    void runPromptRef.current('', {
+      attach: { runId: s.turnRunId, startIndex: 0, dedup: true },
+    });
+  }, []);
+
   /** Activate a session (canonical id) on local state + Wasm ring + URL + picker. */
   const activateSession = useCallback(
     (next: SessionSnapshot) => {
       adoptCloudSession(next);
       setActiveSessionId(next.id);
       void refreshSessions();
+      // Plan #813: F5/login/new tab/switch-back rebuilt the ring — cold attach.
+      queueMicrotask(kickColdAttach);
     },
-    [adoptCloudSession, refreshSessions],
+    [adoptCloudSession, refreshSessions, kickColdAttach],
   );
 
   const persist = useCallback(
@@ -395,12 +427,35 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
   }, []);
 
   const runPrompt = useCallback(
-    async (prompt: string, opts?: { pushUser?: boolean }) => {
+    async (prompt: string, opts?: RunPromptOpts) => {
       const bridge = bridgeRef.current;
       if (!bridge || inflightRef.current) return;
 
+      // Adversarial #857: Send while a durable run is live (503 subscribe-fail,
+      // empty-EOF idle) must attach — never POST (C15 409 mixes Turn ended +
+      // Error with keep-running). Class follows this-heap applied frames, not a
+      // hard-coded cold-at-0 (count>0 → hot at C; else cold + dedup).
+      const live = sessionRef.current;
+      const sendAttach = decideSendAttach({
+        turnRunId: live.turnRunId,
+        turnStatus: live.turnStatus,
+        envelopeCursor: live.turnStreamCursor,
+        heapApplied: heapAppliedRef.current,
+      });
+      const attach: RunPromptAttach | undefined =
+        opts?.attach ??
+        (sendAttach.kind === 'none'
+          ? undefined
+          : {
+              runId: sendAttach.runId,
+              startIndex: sendAttach.startIndex,
+              dedup: sendAttach.dedup,
+            });
+      const attaching = attach != null;
+      const sendWhileRunning =
+        opts?.attach == null && attaching && (prompt ?? '').trim().length > 0;
       const modelId = bridge.getSelectedModel();
-      if (!modelId) {
+      if (!attaching && !modelId) {
         setHostNote('No model selected — catalog empty, failed to load, or not granted.');
         try {
           bridge.pushMessage(
@@ -465,14 +520,15 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
             signal: controller.signal,
             // Default false: Wasm already painted the user line in queueSubmitFromUi.
             // true when host snapped from a historical ring window before the turn.
-            pushUser: opts?.pushUser ?? false,
-            modelId,
+            pushUser: attaching ? false : opts?.pushUser ?? false,
+            ...(modelId ? { modelId } : {}),
             // Phase 2 (#627 / #625): persist every mid-turn session patch
             // (cwd change, sandbox switch) via the same persist callback the
             // turn-end path uses — local write + coalesced cloud PUT.
             // Adversarial #844: late patches after detach take decideDetachPersist
             // (never writeLocal onto a switched session; never PUT a Clear'd id).
             onSessionPatch: persistTurn,
+            ...(attach ? { attach } : {}),
           },
         );
         if (turnEpochRef.current !== epoch) {
@@ -491,8 +547,74 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
         // stream). Dropping session on signal.aborted left SessionStore behind Wasm:
         // Load earlier / refresh could wipe the cancelled turn from the ring.
         persistTurn(folded);
-        if (!result.ok && shouldSetHostTurnNote(folded.turnStatus)) {
+        if (folded.turnStatus === 'running' && folded.turnRunId) {
+          heapAppliedRef.current = {
+            runId: folded.turnRunId,
+            count: folded.turnStreamCursor ?? 0,
+          };
+        } 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.
+        if (
+          shouldRepostAttachFollowUp({
+            sendWhileRunning,
+            turnStatus: folded.turnStatus,
+          })
+        ) {
+          queueMicrotask(() => {
+            void runPromptRef.current(prompt, { pushUser: true });
+          });
+        } else if (!result.ok && shouldSetHostTurnNote(folded.turnStatus)) {
           setHostNote(result.error);
+        } else if (
+          shouldPaintAttachFollowUpNote({
+            sendWhileRunning,
+            resultOk: result.ok,
+            turnStatus: folded.turnStatus,
+          })
+        ) {
+          setHostNote(ATTACH_FOLLOW_UP_NOTE);
+          try {
+            bridge.pushMessage(MessageKind.System, ATTACH_FOLLOW_UP_NOTE);
+          } catch {
+            /* torn-down bridge */
+          }
+        }
+        // Plan #813: SSE drop while still mounted → hot resume at this-heap C.
+        // Empty-EOF GET (applied == startIndex) must not reconnect (spin).
+        // F5 is never this path (heapApplied was nulled; activateSession is cold).
+        // Operator Stop during attach: skip auto-resume this tick (D18 reader
+        // close, not G22 cancel — adversarial #857).
+        if (
+          folded.turnStatus === 'running' &&
+          folded.turnRunId &&
+          !shouldSkipAttachHotResume({
+            attaching,
+            aborted: controller.signal.aborted,
+            isDetachAbort: isDetachAbort(controller.signal),
+          })
+        ) {
+          const resume = decideHotResume({
+            turnRunId: folded.turnRunId,
+            turnStatus: folded.turnStatus,
+            envelopeCursor: folded.turnStreamCursor,
+            heapApplied: heapAppliedRef.current,
+            attachStart: attach?.startIndex,
+          });
+          if (resume.kind === 'hot') {
+            queueMicrotask(() => {
+              void runPromptRef.current('', {
+                attach: {
+                  runId: folded.turnRunId!,
+                  startIndex: resume.startIndex,
+                  dedup: false,
+                },
+              });
+            });
+          }
         }
       } finally {
         const detached = turnEpochRef.current !== epoch;
@@ -530,6 +652,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
     },
     [persist, setUrlSessionId, writeLocalSession],
   );
+  runPromptRef.current = runPrompt;
 
   useEffect(() => {
     let cancelled = false;
@@ -561,6 +684,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
             // re-checks getLocal, but this host guard is authoritative for the active id.
             if (snap.id !== sessionRef.current.id) return;
             adoptCloudSession(snap);
+            queueMicrotask(kickColdAttach);
           },
         });
         repoRef.current = repo;
@@ -710,6 +834,12 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
           // or refresh can recover — don't permanently strand local-only this page load.
           if (result.kind === 'local') setCloudEnabled(r.enabled);
           void refreshSessions();
+          // Plan #813: after Blob/local hydrate, cold-attach a still-running
+          // turn. activateSession also kicks; inflightRef de-dupes the pair.
+          // Do not auto-attach completed sessions (`turnStatus !== 'running'`).
+          if (!cancelled) {
+            queueMicrotask(kickColdAttach);
+          }
         })();
 
         const poll = () => {
@@ -816,6 +946,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
     setUrlSessionId,
     applySessionModel,
     foldPendingModelChange,
+    kickColdAttach,
   ]);
 
   /**
@@ -1152,7 +1283,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
           style={{
             margin: '0.5rem 1rem 0',
             fontSize: '0.75rem',
-            color: ember.muted,
+            color: hostNote === ATTACH_FOLLOW_UP_NOTE ? teal.muted : ember.muted,
             fontFamily: 'ui-monospace, SFMono-Regular, Menlo, monospace',
             flexShrink: 0,
           }}

docs/feature-divide.md

@@ -145,7 +145,7 @@ re-resolved each turn.
 | Theme | `native/harness/src/palette.zig` ↔ `lib/palette.ts` |
 | Export whitelist | `native/harness/build.zig` |
 
-Host `HARNESS_PROTOCOL_VERSION` must equal Wasm `PROTOCOL_VERSION` (currently **19** — 13 added the additive status-slot store; 14 the scalar turn-clock feed `inv_set_turn_elapsed`; 15 added the busy-tick `inv_set_busy_tick`; 16 added model-selection persistence `inv_set_selected_model` + pending-model-change; 17 added the session-rail catalog + pending switch; **18** adds `inv_queued_count` for the in-canvas submit queue; **19** adds `inv_set_queue_promote_allowed` — the host arms a one-shot per-terminal scalar so a Stop/Esc/error/timeout Ready never drains the queue, plan #760).
+Host `HARNESS_PROTOCOL_VERSION` must equal Wasm `PROTOCOL_VERSION` (currently **21** — 13 added the additive status-slot store; 14 the scalar turn-clock feed `inv_set_turn_elapsed`; 15 added the busy-tick `inv_set_busy_tick`; 16 added model-selection persistence `inv_set_selected_model` + pending-model-change; 17 added the session-rail catalog + pending switch; **18** adds `inv_queued_count` for the in-canvas submit queue; **19** adds `inv_set_queue_promote_allowed` — the host arms a one-shot per-terminal scalar so a Stop/Esc/error/timeout Ready never drains the queue, plan #760; **20** adds `inv_queued_insert_front` — turn-retry Continue-on-give-up (plan #759); **v21** adds `inv_clear_ring` — live-session ring replace that keeps the submit queue).
 Mismatch → load error; rebuild both sides. Image **bytes** enter only via bridge put; never dual DOM `<img>` product surface.
 
 ## Related

lib/harnessBridge.test.ts

@@ -1,4 +1,6 @@
 import { afterEach, describe, expect, it, vi } from 'vitest';
+import { readFileSync } from 'node:fs';
+import { resolve } from 'node:path';
 import {
   HARNESS_PROTOCOL_VERSION,
   HarnessBridge,
@@ -114,6 +116,9 @@ function makeMockExports(overrides?: Partial<HarnessBridgeExports>): HarnessBrid
       messages.length = 0;
       pending = null;
     },
+    inv_clear_ring: () => {
+      messages.length = 0;
+    },
     inv_echo: (ptr: number, len: number) => {
       echo = len === 0 ? '' : read(ptr, len);
       return echo.length;
@@ -332,6 +337,13 @@ describe('HarnessBridge', () => {
     vi.restoreAllMocks();
   });
 
+  it('Zig PROTOCOL_VERSION matches HARNESS_PROTOCOL_VERSION', () => {
+    const zig = readFileSync(resolve(process.cwd(), 'native/harness/src/bridge.zig'), 'utf8');
+    const m = zig.match(/pub const PROTOCOL_VERSION: u32 = (\d+);/);
+    expect(m, 'PROTOCOL_VERSION const in bridge.zig').toBeTruthy();
+    expect(Number(m![1])).toBe(HARNESS_PROTOCOL_VERSION);
+  });
+
   it('fromInstance succeeds with mock exports', () => {
     const exports = makeMockExports();
     const instance = { exports } as unknown as WebAssembly.Instance;
@@ -817,7 +829,7 @@ describe('skill_attached kind (protocol v12)', () => {
     // Distinct from the protocol version (13) — a hardcoded kind 13 would be an
     // unknown kind to the Wasm painter.
     expect(MessageKind.SkillAttached).not.toBe(HARNESS_PROTOCOL_VERSION);
-    expect(HARNESS_PROTOCOL_VERSION).toBe(20);
+    expect(HARNESS_PROTOCOL_VERSION).toBe(21);
   });
 
   it('push/readback round-trips a skill_attached row', () => {
@@ -847,7 +859,7 @@ describe('setTurnElapsed (protocol v14)', () => {
   });
 
   it('version bumped to 20 and the export is REQUIRED (fail-closed when missing)', () => {
-    expect(HARNESS_PROTOCOL_VERSION).toBe(20);
+    expect(HARNESS_PROTOCOL_VERSION).toBe(21);
     const exp = makeMockExports() as unknown as WebAssembly.Exports;
     expect(isHarnessBridgeExports(exp)).toBe(true);
     // A rebuilt Wasm that omits inv_set_turn_elapsed fails bridge-load closed,
@@ -916,7 +928,7 @@ describe('status-slot pack (protocol v13)', () => {
 
 describe('queuedCount (protocol v18)', () => {
   it('reads inv_queued_count and fails closed when the export is missing', () => {
-    expect(HARNESS_PROTOCOL_VERSION).toBe(20);
+    expect(HARNESS_PROTOCOL_VERSION).toBe(21);
     const exp = makeMockExports();
     const bridge = new HarnessBridge(exp);
     expect(bridge.queuedCount()).toBe(0);
@@ -926,6 +938,52 @@ describe('queuedCount (protocol v18)', () => {
   });
 });
 
+describe('clearRing / hydrateMessages preserveQueue (protocol v21, adversarial #857)', () => {
+  it('hydrateMessages({preserveQueue:true}) does not clear the mock FIFO', () => {
+    const exp = makeMockExports();
+    const queue: string[] = ['keep-me'];
+    const origClear = exp.inv_clear_messages;
+    exp.inv_clear_messages = () => {
+      origClear();
+      queue.length = 0;
+    };
+    exp.inv_clear_ring = () => {
+      exp.__messages.length = 0;
+    };
+    const bridge = new HarnessBridge(exp);
+    bridge.pushMessage(MessageKind.User, 'hello');
+    bridge.pushMessage(MessageKind.User, 'follow-up');
+    bridge.hydrateMessages([{ kind: MessageKind.User, text: 'hello' }], {
+      preserveQueue: true,
+    });
+    expect(exp.__messages.map((m) => m.text)).toEqual(['hello']);
+    expect(queue).toEqual(['keep-me']);
+  });
+
+  it('hydrateMessages() default still uses inv_clear_messages', () => {
+    const exp = makeMockExports();
+    const queue: string[] = ['stale'];
+    const origClear = exp.inv_clear_messages;
+    exp.inv_clear_messages = () => {
+      origClear();
+      queue.length = 0;
+    };
+    const bridge = new HarnessBridge(exp);
+    bridge.pushMessage(MessageKind.User, 'hello');
+    bridge.hydrateMessages([{ kind: MessageKind.User, text: 'hello' }]);
+    expect(queue).toEqual([]);
+  });
+
+  it('inv_clear_ring export is REQUIRED (fail-closed when missing)', () => {
+    expect(HARNESS_PROTOCOL_VERSION).toBe(21);
+    const exp = makeMockExports() as unknown as WebAssembly.Exports;
+    expect(isHarnessBridgeExports(exp)).toBe(true);
+    const record = exp as unknown as Record<string, unknown>;
+    delete record.inv_clear_ring;
+    expect(isHarnessBridgeExports(record as WebAssembly.Exports)).toBe(false);
+  });
+});
+
 describe('setQueuePromoteAllowed (protocol v19, plan #760)', () => {
   it('arms the one-shot scalar; default true mirrors legacy auto-promote', () => {
     const exp = makeMockExports();
@@ -938,7 +996,7 @@ describe('setQueuePromoteAllowed (protocol v19, plan #760)', () => {
   });
 
   it('export is REQUIRED (fail-closed when missing from the wasm)', () => {
-    expect(HARNESS_PROTOCOL_VERSION).toBe(20);
+    expect(HARNESS_PROTOCOL_VERSION).toBe(21);
     const exp = makeMockExports() as unknown as WebAssembly.Exports;
     expect(isHarnessBridgeExports(exp)).toBe(true);
     const record = exp as unknown as Record<string, unknown>;