Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md

Large diffs are not rendered by default.

115 changes: 109 additions & 6 deletions app/harness/HarnessHost.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import {
import { resetHarnessImageSession } from '../../lib/harnessImages';
import { resetHarnessMathSession } from '../../lib/harnessMath';
import { decideDetach, shouldAbortReader, abortReasonFor, decideDetachPersist, putPreservedTurn, shouldApplyMintBind, shouldSetHostTurnNote } from '../../lib/detachTurn';
import { decideHotResume, decideSendAttach, ATTACH_FOLLOW_UP_NOTE, type HeapApplied } from '../../lib/turnAttach';
import {
HarnessBridge,
HARNESS_PROTOCOL_VERSION,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);
/**
Expand Down Expand Up @@ -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],
);
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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) {
Expand All @@ -491,8 +547,46 @@ 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 (folded.turnStatus === 'running' && folded.turnRunId) {
heapAppliedRef.current = {
runId: folded.turnRunId,
count: folded.turnStreamCursor ?? 0,
};
} else {
heapAppliedRef.current = null;
}
if (!result.ok && shouldSetHostTurnNote(folded.turnStatus)) {
setHostNote(result.error);
} else if (sendWhileRunning) {
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).
if (folded.turnStatus === 'running' && folded.turnRunId) {
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;
Expand Down Expand Up @@ -530,6 +624,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
},
[persist, setUrlSessionId, writeLocalSession],
);
runPromptRef.current = runPrompt;

useEffect(() => {
let cancelled = false;
Expand Down Expand Up @@ -561,6 +656,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;
Expand Down Expand Up @@ -710,6 +806,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 = () => {
Expand Down Expand Up @@ -816,6 +918,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) {
setUrlSessionId,
applySessionModel,
foldPendingModelChange,
kickColdAttach,
]);

/**
Expand Down Expand Up @@ -1152,7 +1255,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,
}}
Expand Down
2 changes: 1 addition & 1 deletion docs/feature-divide.md
Original file line number Diff line number Diff line change
Expand Up @@ -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; **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
Expand Down
66 changes: 62 additions & 4 deletions lib/harnessBridge.test.ts
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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', () => {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand All @@ -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();
Expand All @@ -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>;
Expand Down
Loading
Loading