From 5f239661f0534651eaab7376271866663cde1193 Mon Sep 17 00:00:00 2001 From: Barack Sokullu Date: Tue, 13 Oct 2026 03:10:00 +0300 Subject: [PATCH] Preserve browser handoff ownership through reload and cancellation --- docs/cloud-browser-handoff.md | 24 ++ src/chrome/src/agent/cloud-handoff.js | 6 +- src/chrome/src/cloud-runs.js | 169 ++++++++++++-- src/firefox/src/agent/cloud-handoff.js | 6 +- src/firefox/src/cloud-runs.js | 169 ++++++++++++-- test/cloud-handoff.mjs | 310 ++++++++++++++++++++++++- 6 files changed, 634 insertions(+), 50 deletions(-) create mode 100644 docs/cloud-browser-handoff.md diff --git a/docs/cloud-browser-handoff.md b/docs/cloud-browser-handoff.md new file mode 100644 index 000000000..d8dbc9a0d --- /dev/null +++ b/docs/cloud-browser-handoff.md @@ -0,0 +1,24 @@ +# Cloud browser handoff + +Cloud runtimes advertise browser handoff through `run_handoff_v1`. + +`cloud_pause` requests manual browser control for the existing run. The run stays +reserved while it is `pausing`; a writable `paused` acknowledgement follows only +after the current model/tool boundary settles and the handoff guard is saved. +Pausing a run does not suspend or power off its browser session. + +`cloud_resume` returns control with an optional note. WebBrain reads the current +page before further work. The note reports state and grants no new task authority. +An existing clarification remains pending and requires its original explicit answer. + +The durable `webbrainCloudHandoffGuards` record contains only run ID, tab ID, and +timestamps. Provider credentials, task content, notes, memory, and results are not +stored in this guard. A service worker restart, extension reload/update, or browser +restart restores manual ownership as interrupted and nonresumable. It never replays +the old task. Inspect the browser and explicitly abort the interrupted run before +starting further work. Other agents, scheduled runs, and Cloud tasks remain blocked +on that profile until reconciliation completes. + +The guard clears only after a fresh handback observation is saved or explicit +cancellation confirms the native handler has settled, including recording cleanup. +Storage failures keep the browser read-only and preserve its reservation. diff --git a/src/chrome/src/agent/cloud-handoff.js b/src/chrome/src/agent/cloud-handoff.js index 0442021db..89a777bb6 100644 --- a/src/chrome/src/agent/cloud-handoff.js +++ b/src/chrome/src/agent/cloud-handoff.js @@ -19,10 +19,12 @@ export class CloudHandoffGate { async acknowledge() { if (this.closed || this.state !== 'pausing') return; - this.state = 'paused'; // A durable acknowledgement is required before the viewer becomes writable. await this.onState('paused'); - if (!this.closed) this.stopKeepAlive = this.keepAlive(); + if (!this.closed && this.state === 'pausing') { + this.state = 'paused'; + this.stopKeepAlive = this.keepAlive(); + } } async boundary() { diff --git a/src/chrome/src/cloud-runs.js b/src/chrome/src/cloud-runs.js index 5a72f0729..768cc974a 100644 --- a/src/chrome/src/cloud-runs.js +++ b/src/chrome/src/cloud-runs.js @@ -12,6 +12,7 @@ const PRIVATE_RESULT_PLACEHOLDER = '[Private final answer encrypted for this wor const DEFAULT_CLOUD_BRIDGE_URL = 'ws://127.0.0.1:17374/extension'; const CLOUD_RUN_STORAGE_KEY = 'webbrainCloudRunSnapshots'; +const CLOUD_HANDOFF_GUARD_KEY = 'webbrainCloudHandoffGuards'; const CLOUD_UPDATE_LIMIT = 200; const CLOUD_RUN_LIMIT = 50; const CLOUD_STRING_LIMIT = 16 * 1024; @@ -713,6 +714,21 @@ export function buildCloudPersistenceRows(runs) { return rows; } +// Session snapshots disappear on extension reload and browser restart. Keep +// only the identity needed to reserve manual ownership in durable storage. +function normalizeCloudHandoffGuard(row) { + const timestamp = value => typeof value === 'string' && /^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$/.test(value) + && Number.isFinite(Date.parse(value)) && new Date(value).toISOString() === value; + // Cloud callers may supply correlation IDs containing punctuation or + // Unicode. Match the existing start API rather than changing that contract. + if (!row || typeof row.runId !== 'string' || !row.runId.trim() || row.runId.length > CLOUD_STRING_LIMIT + || !Number.isSafeInteger(row.tabId) || row.tabId < 0 || row.tabId > 2147483647 + || !timestamp(row.createdAt) || !timestamp(row.updatedAt) + || row.pausedAt != null && !timestamp(row.pausedAt)) throw new Error('Browser handoff guard could not be reconciled.'); + return { runId: row.runId, tabId: row.tabId, createdAt: row.createdAt, + updatedAt: row.updatedAt, pausedAt: row.pausedAt || null }; +} + function cloudSnapshot(run, { includeUpdates = true } = {}) { if (!run) return null; return { @@ -770,6 +786,7 @@ export function createCloudRunController({ const startingRunIds = new Set(); let pendingStarts = 0; const handoffGates = new Map(); + const activeExecutions = new Set(); const startingConversations = new Set(); const runAbortControllers = new Map(); // Private envelopes live only for the active handle, outside snapshots and @@ -778,9 +795,24 @@ export function createCloudRunController({ let hydratePromise = null; let persistQueue = Promise.resolve(); let persistTimer = null; + const durableHandoffGuards = new Map(); + let storedHandoffGuard = false; const isoNow = () => now().toISOString(); + function snapshot(run, options) { + const result = cloudSnapshot(run, options); + // Handler errors and cancellation can assign a terminal status before + // asynchronous cleanup finishes. A retained guard remains authoritative + // for pollers and streamers until confirmed removal releases ownership. + if (result && durableHandoffGuards.has(run.runId) && TERMINAL_STATUSES.has(result.status)) { + result.status = run.handoff ? 'paused' : 'aborting'; + result.handoff = run.handoff ? { ...run.handoff, state: 'paused', resumable: false, interrupted: true } : null; + result.completedAt = null; + } + return result; + } + // MV3 service workers suspend after ~30s idle; a bare setTimeout does not // reset that timer. Ping getPlatformInfo on a 20s interval for the duration // of an in-handle resume wait so Chrome keeps the worker alive. Best-effort: @@ -804,12 +836,39 @@ export function createCloudRunController({ } } - async function persist() { - if (!api.storage?.session?.set) return; - const rows = buildCloudPersistenceRows(runs); + async function persist({ clearHandoffRunId = null } = {}) { persistQueue = persistQueue .catch(() => {}) - .then(() => api.storage.session.set({ [CLOUD_RUN_STORAGE_KEY]: rows })); + .then(async () => { + // Build inside the queue so an older ordinary checkpoint cannot put + // back a reservation already cleared by confirmed reconciliation. + const cancellationPending = id => activeExecutions.has(id) && runAbortControllers.get(id)?.signal.aborted; + const clearing = [...durableHandoffGuards.entries()].filter(([id]) => { + const run = runs.get(id); + return !cancellationPending(id) && (id === clearHandoffRunId + || (run && TERMINAL_STATUSES.has(run.status) && !activeExecutions.has(id))); + }); + const clearedIds = new Set(clearing.map(([id]) => id)); + const guards = [...durableHandoffGuards.values()].filter(row => !clearedIds.has(row.runId)); + // A failed session write must not remove the durable reservation. + await api.storage?.session?.set?.({ [CLOUD_RUN_STORAGE_KEY]: buildCloudPersistenceRows(runs) }); + if (guards.length || storedHandoffGuard || clearing.length) { + if (!api.storage?.local?.set) throw new Error('Durable browser handoff storage is unavailable.'); + await api.storage.local.set({ [CLOUD_HANDOFF_GUARD_KEY]: guards }); + // Cancellation can arrive during an awaited handback clear. Keep + // the in-memory reservation and restore its durable row before this + // checkpoint can acknowledge anything or allow another actor in. + const retained = clearing.filter(([id]) => cancellationPending(id)); + if (retained.length) { + await api.storage.local.set({ [CLOUD_HANDOFF_GUARD_KEY]: [ + ...guards, ...retained.map(([id, row]) => durableHandoffGuards.get(id) || row), + ] }); + } + storedHandoffGuard = guards.length > 0 || retained.length > 0; + const retainedIds = new Set(retained.map(([id]) => id)); + for (const [id, row] of clearing) if (!retainedIds.has(id) && durableHandoffGuards.get(id) === row) durableHandoffGuards.delete(id); + } + }); await persistQueue; } @@ -824,9 +883,26 @@ export function createCloudRunController({ async function hydrate() { if (hydratePromise) return hydratePromise; hydratePromise = (async () => { - if (!api.storage?.session?.get) return; - const stored = await api.storage.session.get(CLOUD_RUN_STORAGE_KEY); + const local = await api.storage?.local?.get?.(CLOUD_HANDOFF_GUARD_KEY) || {}; + const guardRows = local[CLOUD_HANDOFF_GUARD_KEY] === undefined ? [] : local[CLOUD_HANDOFF_GUARD_KEY]; + if (!Array.isArray(guardRows) || guardRows.length > CLOUD_RUN_LIMIT) throw new Error('Browser handoff guard could not be reconciled.'); + for (const row of guardRows) { + const guard = normalizeCloudHandoffGuard(row); + if (durableHandoffGuards.has(guard.runId)) throw new Error('Browser handoff guard could not be reconciled.'); + durableHandoffGuards.set(guard.runId, guard); + } + storedHandoffGuard = durableHandoffGuards.size > 0; + // Preserve main's fail-closed session read. Read failures are not proof + // that no run owns this profile. + const stored = await api.storage?.session?.get?.(CLOUD_RUN_STORAGE_KEY) || {}; const rows = Array.isArray(stored?.[CLOUD_RUN_STORAGE_KEY]) ? stored[CLOUD_RUN_STORAGE_KEY] : []; + for (const guard of durableHandoffGuards.values()) { + const old = rows.findIndex(row => row?.runId === guard.runId); + const restored = { ...(old < 0 ? { updates: [] } : rows[old]), ...guard, status: 'paused', + handoff: { state: 'paused', resumable: false, interrupted: true, + ...(guard.pausedAt ? { pausedAt: guard.pausedAt } : {}) }, completedAt: null }; + if (old < 0) rows.push(restored); else rows[old] = restored; + } let changed = false; for (const row of rows) { if (!row?.runId) continue; @@ -847,6 +923,11 @@ export function createCloudRunController({ // Keep the browser reserved for its owner; never replay this run. restored.status = 'paused'; restored.handoff = { ...restored.handoff, state: 'paused', resumable: false, interrupted: true }; + durableHandoffGuards.set(restored.runId, normalizeCloudHandoffGuard({ + runId: restored.runId, tabId: restored.tabId, + createdAt: durableHandoffGuards.get(restored.runId)?.createdAt || isoNow(), + updatedAt: isoNow(), pausedAt: restored.handoff.pausedAt || null, + })); restored.error = 'WebBrain restarted during browser handoff. Inspect the page and explicitly stop this interrupted run before starting further work.'; startingTabs.add(restored.tabId); pushUpdate(restored, 'handoff', restored.handoff); @@ -1142,7 +1223,7 @@ export function createCloudRunController({ } function assertNoHandoff() { - if (activeHandoff()) { + if (durableHandoffGuards.size || activeHandoff()) { throw cloudRunError('Browser control belongs to an active handoff. Return control or explicitly reconcile the interrupted run before starting another task.', 409); } } @@ -1192,6 +1273,7 @@ export function createCloudRunController({ ? null : await importPrivateResultRecipient(msg.private_result_public_key); const suppliedRunId = msg.runId ?? msg.run_id; const requestedRunId = suppliedRunId == null ? '' : String(suppliedRunId).trim(); + if (requestedRunId.length > CLOUD_STRING_LIMIT) throw cloudRunError('Cloud run ID exceeds the supported length.', 400); const parentRunId = String(msg.parentRunId || msg.parent_run_id || '').trim() || null; const savedParent = parentRunId ? runs.get(parentRunId) : null; if (parentRunId) assertMemoryContinuation(savedParent, memoryOwner); @@ -1330,17 +1412,41 @@ export function createCloudRunController({ }; const handoffGate = new CloudHandoffGate({ async onState(state) { + const checkpoint = async options => { + try { await persist(options); } catch (error) { + // An abort may close the gate while this write is pending. Keep + // cancellation pending until cleanup; never resurrect its pause. + if (!handoffGate.closed && !cancelled()) { + run.status = 'paused'; + run.handoff = { state: 'paused', resumable: false, interrupted: true }; + run.error = 'Browser handoff could not be saved safely. Keep the browser read-only and explicitly stop this run.'; + pushUpdate(run, 'handoff', run.handoff); + } + throw error; + } + }; if (state === 'running') { + await checkpoint({ clearHandoffRunId: run.runId }); + if (handoffGate.closed || cancelled()) return; run.handoff = null; run.status = run.pendingInput ? 'needs_user_input' : 'running'; } else { + if (handoffGate.closed || cancelled()) return; + // Guard ownership from the pause request onwards. Neither polling + // nor streaming may see writable paused state before this write. + durableHandoffGuards.set(run.runId, normalizeCloudHandoffGuard({ + runId: run.runId, tabId: run.tabId, createdAt: run.createdAt, updatedAt: isoNow(), + pausedAt: state === 'paused' ? isoNow() : run.handoff?.pausedAt || null, + })); + await checkpoint(); + if (handoffGate.closed || cancelled()) return; run.status = state; run.handoff = { state, resumable: true, interrupted: false, ...(state === 'paused' ? { pausedAt: isoNow() } : run.handoff?.pausedAt ? { pausedAt: run.handoff.pausedAt } : {}) }; } run.updatedAt = isoNow(); pushUpdate(run, 'handoff', run.handoff || { state: null }); - await persist(); + await checkpoint(); }, keepAlive() { const ping = () => { try { const p = api.runtime?.getPlatformInfo?.(() => { void api.runtime.lastError; }); p?.catch?.(() => {}); } catch {} }; @@ -1378,6 +1484,7 @@ export function createCloudRunController({ throw error; } + activeExecutions.add(run.runId); (async () => { let recordingId = null; const strictSecretMode = !!recipient || agent.strictSecretMode === true; @@ -1658,6 +1765,23 @@ export function createCloudRunController({ strictSecretOverflowRuns.delete(run.runId); strictTraceOverflowRuns.delete(run.runId); memoryConfigurations.delete(run.runId); + activeExecutions.delete(run.runId); + if (run.status === 'aborted' && durableHandoffGuards.has(run.runId)) { + // The handler and recording cleanup have settled. Clear the durable + // reservation before publicly acknowledging terminal cancellation. + run.status = 'aborting'; + run.completedAt = null; + try { + await persist({ clearHandoffRunId: run.runId }); + run.status = 'aborted'; + run.completedAt = isoNow(); + } catch { + run.status = 'paused'; + run.handoff = { state: 'paused', resumable: false, interrupted: true }; + run.error = 'Browser handoff cancellation could not be saved. Keep the browser read-only and explicitly stop this run again.'; + pushUpdate(run, 'handoff', run.handoff); + } + } sendIndicator(tabId, 'WB_HIDE_AGENT_INDICATORS'); // Recording flush is still owned by this run. Keep the tab reserved // through all cleanup so aborting a nonterminal capture cannot cancel @@ -1667,7 +1791,7 @@ export function createCloudRunController({ } })(); - return cloudSnapshot(run, { includeUpdates: false }); + return snapshot(run, { includeUpdates: false }); } async function startWorkflowRun(msg = {}) { @@ -1744,17 +1868,18 @@ export function createCloudRunController({ // has finished writing; a poll cannot race a pending storage write. await persistQueue; const runId = msg.runId || msg.run_id; - if (!runId) return { runs: [...runs.values()].map(run => cloudSnapshot(run, { includeUpdates: false })) }; + if (!runId) return { runs: [...runs.values()].map(run => snapshot(run, { includeUpdates: false })) }; const run = runs.get(runId); if (!run) throw new Error('Unknown cloud run.'); - return cloudSnapshot(run); + return snapshot(run); } async function abort(msg = {}) { await hydrate(); const run = runs.get(msg.runId || msg.run_id); if (!run) throw new Error('Unknown cloud run.'); - if (run.handoff?.interrupted) { + if (run.handoff?.interrupted && !activeExecutions.has(run.runId)) { + await persist({ clearHandoffRunId: run.runId }); run.status = 'aborted'; run.handoff = null; pushUpdate(run, 'handoff', { state: null }); @@ -1774,19 +1899,20 @@ export function createCloudRunController({ runAbortControllers.get(run.runId)?.abort(); agent.abort(run.tabId); await persist(); - } - return cloudSnapshot(run); + } else if (run.status === 'aborted') await persist(); + return snapshot(run); } async function pause(msg = {}) { await hydrate(); + await persistQueue; const run = runs.get(msg.runId || msg.run_id); if (!run) throw cloudRunError('Unknown cloud run.', 404); if (run.workflowId) throw cloudRunError('Saved workflow replay does not support browser handoff.', 409); if (run.handoff) { // A duplicate pause must not expose a checkpoint still being written. await persistQueue; - return cloudSnapshot(run); + return snapshot(run); } if (!['running', 'needs_user_input'].includes(run.status)) throw cloudRunError('Only a running browser task can be paused.', 409); if (pendingStarts || [...runs.values()].some(other => other !== run && !TERMINAL_STATUSES.has(other.status)) @@ -1811,21 +1937,22 @@ export function createCloudRunController({ } await gate.acknowledge(); } - return cloudSnapshot(run); + return snapshot(run); } async function resume(msg = {}) { await hydrate(); + await persistQueue; const run = runs.get(msg.runId || msg.run_id); if (!run) throw cloudRunError('Unknown cloud run.', 404); - if (!run.handoff?.resumable || run.handoff.interrupted) { + if (run.status !== 'paused' || run.handoff?.state !== 'paused' || !run.handoff.resumable || run.handoff.interrupted) { throw cloudRunError('This handoff was interrupted. Inspect the browser and explicitly stop the interrupted run; it cannot safely resume.', 409); } if (typeof (msg.note ?? '') !== 'string' || (msg.note || '').length > 2000) throw cloudRunError('The handoff note must be text of at most 2000 characters.', 400); const gate = handoffGates.get(run.runId); if (!gate) throw cloudRunError('The browser handoff is no longer resumable.', 409); await gate.resume((msg.note || '').trim()); - return cloudSnapshot(run); + return snapshot(run); } async function respond(msg = {}) { @@ -1851,7 +1978,7 @@ export function createCloudRunController({ run.error = ''; pushUpdate(run, 'clarify_response', { clarifyId, source: 'cloud_api' }); await persist(); - return cloudSnapshot(run); + return snapshot(run); } // Persistent bridge identity (chrome.storage.local). The token is the Cloud @@ -1921,11 +2048,11 @@ export function createCloudRunController({ const owner = activeHandoff(); const settlingOwner = kind === 'cloud' && owner?.tabId === tabId && owner.handoff?.state === 'pausing' && !owner.handoff.interrupted; - if (owner && !settlingOwner) { + if ((durableHandoffGuards.size || owner) && !settlingOwner) { throw cloudRunError('The browser is reserved for manual control. Return control or reconcile the interrupted handoff before starting an agent.', 409); } }, - isRunning: (tabId) => startingTabs.has(tabId), + isRunning: (tabId) => startingTabs.has(tabId) || [...durableHandoffGuards.values()].some(row => row.tabId === tabId), startRun, startWorkflowRun, compileWorkflow, diff --git a/src/firefox/src/agent/cloud-handoff.js b/src/firefox/src/agent/cloud-handoff.js index 0442021db..89a777bb6 100644 --- a/src/firefox/src/agent/cloud-handoff.js +++ b/src/firefox/src/agent/cloud-handoff.js @@ -19,10 +19,12 @@ export class CloudHandoffGate { async acknowledge() { if (this.closed || this.state !== 'pausing') return; - this.state = 'paused'; // A durable acknowledgement is required before the viewer becomes writable. await this.onState('paused'); - if (!this.closed) this.stopKeepAlive = this.keepAlive(); + if (!this.closed && this.state === 'pausing') { + this.state = 'paused'; + this.stopKeepAlive = this.keepAlive(); + } } async boundary() { diff --git a/src/firefox/src/cloud-runs.js b/src/firefox/src/cloud-runs.js index c2716079d..b57873589 100644 --- a/src/firefox/src/cloud-runs.js +++ b/src/firefox/src/cloud-runs.js @@ -12,6 +12,7 @@ const PRIVATE_RESULT_PLACEHOLDER = '[Private final answer encrypted for this wor const DEFAULT_CLOUD_BRIDGE_URL = 'ws://127.0.0.1:17374/extension'; const CLOUD_RUN_STORAGE_KEY = 'webbrainCloudRunSnapshots'; +const CLOUD_HANDOFF_GUARD_KEY = 'webbrainCloudHandoffGuards'; const CLOUD_UPDATE_LIMIT = 200; const CLOUD_RUN_LIMIT = 50; const CLOUD_STRING_LIMIT = 16 * 1024; @@ -713,6 +714,21 @@ export function buildCloudPersistenceRows(runs) { return rows; } +// Session snapshots disappear on extension reload and browser restart. Keep +// only the identity needed to reserve manual ownership in durable storage. +function normalizeCloudHandoffGuard(row) { + const timestamp = value => typeof value === 'string' && /^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$/.test(value) + && Number.isFinite(Date.parse(value)) && new Date(value).toISOString() === value; + // Cloud callers may supply correlation IDs containing punctuation or + // Unicode. Match the existing start API rather than changing that contract. + if (!row || typeof row.runId !== 'string' || !row.runId.trim() || row.runId.length > CLOUD_STRING_LIMIT + || !Number.isSafeInteger(row.tabId) || row.tabId < 0 || row.tabId > 2147483647 + || !timestamp(row.createdAt) || !timestamp(row.updatedAt) + || row.pausedAt != null && !timestamp(row.pausedAt)) throw new Error('Browser handoff guard could not be reconciled.'); + return { runId: row.runId, tabId: row.tabId, createdAt: row.createdAt, + updatedAt: row.updatedAt, pausedAt: row.pausedAt || null }; +} + function cloudSnapshot(run, { includeUpdates = true } = {}) { if (!run) return null; return { @@ -772,6 +788,7 @@ export function createCloudRunController({ const startingRunIds = new Set(); let pendingStarts = 0; const handoffGates = new Map(); + const activeExecutions = new Set(); const startingConversations = new Set(); const runAbortControllers = new Map(); // Private envelopes live only for the active handle, outside snapshots and @@ -780,9 +797,24 @@ export function createCloudRunController({ let hydratePromise = null; let persistQueue = Promise.resolve(); let persistTimer = null; + const durableHandoffGuards = new Map(); + let storedHandoffGuard = false; const isoNow = () => now().toISOString(); + function snapshot(run, options) { + const result = cloudSnapshot(run, options); + // Handler errors and cancellation can assign a terminal status before + // asynchronous cleanup finishes. A retained guard remains authoritative + // for pollers and streamers until confirmed removal releases ownership. + if (result && durableHandoffGuards.has(run.runId) && TERMINAL_STATUSES.has(result.status)) { + result.status = run.handoff ? 'paused' : 'aborting'; + result.handoff = run.handoff ? { ...run.handoff, state: 'paused', resumable: false, interrupted: true } : null; + result.completedAt = null; + } + return result; + } + // Firefox uses a persistent background page, but the same ping is harmless // and keeps the Chrome MV3 service worker alive during resume waits. A bare // setTimeout does not reset the MV3 idle timer. @@ -805,12 +837,39 @@ export function createCloudRunController({ } } - async function persist() { - if (!api.storage?.session?.set) return; - const rows = buildCloudPersistenceRows(runs); + async function persist({ clearHandoffRunId = null } = {}) { persistQueue = persistQueue .catch(() => {}) - .then(() => api.storage.session.set({ [CLOUD_RUN_STORAGE_KEY]: rows })); + .then(async () => { + // Build inside the queue so an older ordinary checkpoint cannot put + // back a reservation already cleared by confirmed reconciliation. + const cancellationPending = id => activeExecutions.has(id) && runAbortControllers.get(id)?.signal.aborted; + const clearing = [...durableHandoffGuards.entries()].filter(([id]) => { + const run = runs.get(id); + return !cancellationPending(id) && (id === clearHandoffRunId + || (run && TERMINAL_STATUSES.has(run.status) && !activeExecutions.has(id))); + }); + const clearedIds = new Set(clearing.map(([id]) => id)); + const guards = [...durableHandoffGuards.values()].filter(row => !clearedIds.has(row.runId)); + // A failed session write must not remove the durable reservation. + await api.storage?.session?.set?.({ [CLOUD_RUN_STORAGE_KEY]: buildCloudPersistenceRows(runs) }); + if (guards.length || storedHandoffGuard || clearing.length) { + if (!api.storage?.local?.set) throw new Error('Durable browser handoff storage is unavailable.'); + await api.storage.local.set({ [CLOUD_HANDOFF_GUARD_KEY]: guards }); + // Cancellation can arrive during an awaited handback clear. Keep + // the in-memory reservation and restore its durable row before this + // checkpoint can acknowledge anything or allow another actor in. + const retained = clearing.filter(([id]) => cancellationPending(id)); + if (retained.length) { + await api.storage.local.set({ [CLOUD_HANDOFF_GUARD_KEY]: [ + ...guards, ...retained.map(([id, row]) => durableHandoffGuards.get(id) || row), + ] }); + } + storedHandoffGuard = guards.length > 0 || retained.length > 0; + const retainedIds = new Set(retained.map(([id]) => id)); + for (const [id, row] of clearing) if (!retainedIds.has(id) && durableHandoffGuards.get(id) === row) durableHandoffGuards.delete(id); + } + }); await persistQueue; } @@ -825,9 +884,26 @@ export function createCloudRunController({ async function hydrate() { if (hydratePromise) return hydratePromise; hydratePromise = (async () => { - if (!api.storage?.session?.get) return; - const stored = await api.storage.session.get(CLOUD_RUN_STORAGE_KEY); + const local = await api.storage?.local?.get?.(CLOUD_HANDOFF_GUARD_KEY) || {}; + const guardRows = local[CLOUD_HANDOFF_GUARD_KEY] === undefined ? [] : local[CLOUD_HANDOFF_GUARD_KEY]; + if (!Array.isArray(guardRows) || guardRows.length > CLOUD_RUN_LIMIT) throw new Error('Browser handoff guard could not be reconciled.'); + for (const row of guardRows) { + const guard = normalizeCloudHandoffGuard(row); + if (durableHandoffGuards.has(guard.runId)) throw new Error('Browser handoff guard could not be reconciled.'); + durableHandoffGuards.set(guard.runId, guard); + } + storedHandoffGuard = durableHandoffGuards.size > 0; + // Preserve main's fail-closed session read. Read failures are not proof + // that no run owns this profile. + const stored = await api.storage?.session?.get?.(CLOUD_RUN_STORAGE_KEY) || {}; const rows = Array.isArray(stored?.[CLOUD_RUN_STORAGE_KEY]) ? stored[CLOUD_RUN_STORAGE_KEY] : []; + for (const guard of durableHandoffGuards.values()) { + const old = rows.findIndex(row => row?.runId === guard.runId); + const restored = { ...(old < 0 ? { updates: [] } : rows[old]), ...guard, status: 'paused', + handoff: { state: 'paused', resumable: false, interrupted: true, + ...(guard.pausedAt ? { pausedAt: guard.pausedAt } : {}) }, completedAt: null }; + if (old < 0) rows.push(restored); else rows[old] = restored; + } let changed = false; for (const row of rows) { if (!row?.runId) continue; @@ -848,6 +924,11 @@ export function createCloudRunController({ // Keep the browser reserved for its owner; never replay this run. restored.status = 'paused'; restored.handoff = { ...restored.handoff, state: 'paused', resumable: false, interrupted: true }; + durableHandoffGuards.set(restored.runId, normalizeCloudHandoffGuard({ + runId: restored.runId, tabId: restored.tabId, + createdAt: durableHandoffGuards.get(restored.runId)?.createdAt || isoNow(), + updatedAt: isoNow(), pausedAt: restored.handoff.pausedAt || null, + })); restored.error = 'WebBrain restarted during browser handoff. Inspect the page and explicitly stop this interrupted run before starting further work.'; startingTabs.add(restored.tabId); pushUpdate(restored, 'handoff', restored.handoff); @@ -1143,7 +1224,7 @@ export function createCloudRunController({ } function assertNoHandoff() { - if (activeHandoff()) { + if (durableHandoffGuards.size || activeHandoff()) { throw cloudRunError('Browser control belongs to an active handoff. Return control or explicitly reconcile the interrupted run before starting another task.', 409); } } @@ -1193,6 +1274,7 @@ export function createCloudRunController({ ? null : await importPrivateResultRecipient(msg.private_result_public_key); const suppliedRunId = msg.runId ?? msg.run_id; const requestedRunId = suppliedRunId == null ? '' : String(suppliedRunId).trim(); + if (requestedRunId.length > CLOUD_STRING_LIMIT) throw cloudRunError('Cloud run ID exceeds the supported length.', 400); const parentRunId = String(msg.parentRunId || msg.parent_run_id || '').trim() || null; const savedParent = parentRunId ? runs.get(parentRunId) : null; if (parentRunId) assertMemoryContinuation(savedParent, memoryOwner); @@ -1331,17 +1413,41 @@ export function createCloudRunController({ }; const handoffGate = new CloudHandoffGate({ async onState(state) { + const checkpoint = async options => { + try { await persist(options); } catch (error) { + // An abort may close the gate while this write is pending. Keep + // cancellation pending until cleanup; never resurrect its pause. + if (!handoffGate.closed && !cancelled()) { + run.status = 'paused'; + run.handoff = { state: 'paused', resumable: false, interrupted: true }; + run.error = 'Browser handoff could not be saved safely. Keep the browser read-only and explicitly stop this run.'; + pushUpdate(run, 'handoff', run.handoff); + } + throw error; + } + }; if (state === 'running') { + await checkpoint({ clearHandoffRunId: run.runId }); + if (handoffGate.closed || cancelled()) return; run.handoff = null; run.status = run.pendingInput ? 'needs_user_input' : 'running'; } else { + if (handoffGate.closed || cancelled()) return; + // Guard ownership from the pause request onwards. Neither polling + // nor streaming may see writable paused state before this write. + durableHandoffGuards.set(run.runId, normalizeCloudHandoffGuard({ + runId: run.runId, tabId: run.tabId, createdAt: run.createdAt, updatedAt: isoNow(), + pausedAt: state === 'paused' ? isoNow() : run.handoff?.pausedAt || null, + })); + await checkpoint(); + if (handoffGate.closed || cancelled()) return; run.status = state; run.handoff = { state, resumable: true, interrupted: false, ...(state === 'paused' ? { pausedAt: isoNow() } : run.handoff?.pausedAt ? { pausedAt: run.handoff.pausedAt } : {}) }; } run.updatedAt = isoNow(); pushUpdate(run, 'handoff', run.handoff || { state: null }); - await persist(); + await checkpoint(); }, keepAlive() { const ping = () => { try { const p = api.runtime?.getPlatformInfo?.(() => { void api.runtime.lastError; }); p?.catch?.(() => {}); } catch {} }; @@ -1379,6 +1485,7 @@ export function createCloudRunController({ throw error; } + activeExecutions.add(run.runId); (async () => { let recordingId = null; const strictSecretMode = !!recipient || agent.strictSecretMode === true; @@ -1659,6 +1766,23 @@ export function createCloudRunController({ strictSecretOverflowRuns.delete(run.runId); strictTraceOverflowRuns.delete(run.runId); memoryConfigurations.delete(run.runId); + activeExecutions.delete(run.runId); + if (run.status === 'aborted' && durableHandoffGuards.has(run.runId)) { + // The handler and recording cleanup have settled. Clear the durable + // reservation before publicly acknowledging terminal cancellation. + run.status = 'aborting'; + run.completedAt = null; + try { + await persist({ clearHandoffRunId: run.runId }); + run.status = 'aborted'; + run.completedAt = isoNow(); + } catch { + run.status = 'paused'; + run.handoff = { state: 'paused', resumable: false, interrupted: true }; + run.error = 'Browser handoff cancellation could not be saved. Keep the browser read-only and explicitly stop this run again.'; + pushUpdate(run, 'handoff', run.handoff); + } + } sendIndicator(tabId, 'WB_HIDE_AGENT_INDICATORS'); // Recording flush is still owned by this run. Keep the tab reserved // through all cleanup so aborting a nonterminal capture cannot cancel @@ -1668,7 +1792,7 @@ export function createCloudRunController({ } })(); - return cloudSnapshot(run, { includeUpdates: false }); + return snapshot(run, { includeUpdates: false }); } async function startWorkflowRun(msg = {}) { @@ -1745,17 +1869,18 @@ export function createCloudRunController({ // has finished writing; a poll cannot race a pending storage write. await persistQueue; const runId = msg.runId || msg.run_id; - if (!runId) return { runs: [...runs.values()].map(run => cloudSnapshot(run, { includeUpdates: false })) }; + if (!runId) return { runs: [...runs.values()].map(run => snapshot(run, { includeUpdates: false })) }; const run = runs.get(runId); if (!run) throw new Error('Unknown cloud run.'); - return cloudSnapshot(run); + return snapshot(run); } async function abort(msg = {}) { await hydrate(); const run = runs.get(msg.runId || msg.run_id); if (!run) throw new Error('Unknown cloud run.'); - if (run.handoff?.interrupted) { + if (run.handoff?.interrupted && !activeExecutions.has(run.runId)) { + await persist({ clearHandoffRunId: run.runId }); run.status = 'aborted'; run.handoff = null; pushUpdate(run, 'handoff', { state: null }); @@ -1775,19 +1900,20 @@ export function createCloudRunController({ runAbortControllers.get(run.runId)?.abort(); agent.abort(run.tabId); await persist(); - } - return cloudSnapshot(run); + } else if (run.status === 'aborted') await persist(); + return snapshot(run); } async function pause(msg = {}) { await hydrate(); + await persistQueue; const run = runs.get(msg.runId || msg.run_id); if (!run) throw cloudRunError('Unknown cloud run.', 404); if (run.workflowId) throw cloudRunError('Saved workflow replay does not support browser handoff.', 409); if (run.handoff) { // A duplicate pause must not expose a checkpoint still being written. await persistQueue; - return cloudSnapshot(run); + return snapshot(run); } if (!['running', 'needs_user_input'].includes(run.status)) throw cloudRunError('Only a running browser task can be paused.', 409); if (pendingStarts || [...runs.values()].some(other => other !== run && !TERMINAL_STATUSES.has(other.status)) @@ -1812,21 +1938,22 @@ export function createCloudRunController({ } await gate.acknowledge(); } - return cloudSnapshot(run); + return snapshot(run); } async function resume(msg = {}) { await hydrate(); + await persistQueue; const run = runs.get(msg.runId || msg.run_id); if (!run) throw cloudRunError('Unknown cloud run.', 404); - if (!run.handoff?.resumable || run.handoff.interrupted) { + if (run.status !== 'paused' || run.handoff?.state !== 'paused' || !run.handoff.resumable || run.handoff.interrupted) { throw cloudRunError('This handoff was interrupted. Inspect the browser and explicitly stop the interrupted run; it cannot safely resume.', 409); } if (typeof (msg.note ?? '') !== 'string' || (msg.note || '').length > 2000) throw cloudRunError('The handoff note must be text of at most 2000 characters.', 400); const gate = handoffGates.get(run.runId); if (!gate) throw cloudRunError('The browser handoff is no longer resumable.', 409); await gate.resume((msg.note || '').trim()); - return cloudSnapshot(run); + return snapshot(run); } async function respond(msg = {}) { @@ -1852,7 +1979,7 @@ export function createCloudRunController({ run.error = ''; pushUpdate(run, 'clarify_response', { clarifyId, source: 'cloud_api' }); await persist(); - return cloudSnapshot(run); + return snapshot(run); } // Persistent bridge identity (chrome.storage.local). The token is the Cloud @@ -1919,11 +2046,11 @@ export function createCloudRunController({ const owner = activeHandoff(); const settlingOwner = kind === 'cloud' && owner?.tabId === tabId && owner.handoff?.state === 'pausing' && !owner.handoff.interrupted; - if (owner && !settlingOwner) { + if ((durableHandoffGuards.size || owner) && !settlingOwner) { throw cloudRunError('The browser is reserved for manual control. Return control or reconcile the interrupted handoff before starting an agent.', 409); } }, - isRunning: (tabId) => startingTabs.has(tabId), + isRunning: (tabId) => startingTabs.has(tabId) || [...durableHandoffGuards.values()].some(row => row.tabId === tabId), startRun, startWorkflowRun, compileWorkflow, diff --git a/test/cloud-handoff.mjs b/test/cloud-handoff.mjs index a68c49fe2..c6eaf0bf0 100644 --- a/test/cloud-handoff.mjs +++ b/test/cloud-handoff.mjs @@ -19,17 +19,18 @@ for (const browser of ['chrome', 'firefox']) { const { createCloudRunController, buildCloudPersistenceRows } = await import(`../src/${browser}/src/cloud-runs.js`); const { Agent } = await import(`../src/${browser}/src/agent/agent.js`); - function harness(processMessage) { + function harness(processMessage, options = {}) { const stored = {}; - const api = { storage: { session: { get: async key => ({ [key]: structuredClone(stored[key]) }), set: async data => Object.assign(stored, structuredClone(data)) } }, + const local = {}; + const api = { storage: { local: { get: async key => ({ [key]: structuredClone(local[key]) }), set: async data => Object.assign(local, structuredClone(data)) }, session: { get: async key => ({ [key]: structuredClone(stored[key]) }), set: async data => Object.assign(stored, structuredClone(data)) } }, tabs: { get: async id => ({ id, url: 'https://example.com/' }), query: async () => [{ id: 7, url: 'https://example.com/' }], update: async () => {} } }; let clarificationResolve; const agent = { isRunning: () => false, hasOtherRunningTabs: () => false, canPauseAtClarification: () => true, strictSecretMode: true, processMessage, abort() {}, _persistNow: async () => ({ ok: true }), submitClarifyResponse(_tab, _id, answer) { clarificationResolve?.(answer); return true; } }; let seq = 0; - const controller = createCloudRunController({ chromeApi: api, agent, makeRunId: () => `run_handoff_${++seq}`, ensureOffscreen: async () => {} }); - return { controller, stored, agent, api, setClarificationResolver(value) { clarificationResolve = value; } }; + const controller = createCloudRunController({ chromeApi: api, agent, makeRunId: () => `run_handoff_${++seq}`, ensureOffscreen: async () => {}, ...options }); + return { controller, stored, local, agent, api, setClarificationResolver(value) { clarificationResolve = value; } }; } test(`${browser}: pausing acknowledges only a settled boundary and keeps its tab reservation`, async () => { @@ -313,4 +314,305 @@ for (const browser of ['chrome', 'firefox']) { assert.equal(agent.hasOtherRunningTabs(7), true); assert.equal(agent.hasOtherRunningTabs(8), false); }); + for (const restart of ['extension reload', 'browser startup without session storage']) { + test(`${browser}: ${restart} restores only a durable manual guard and never replays`, async () => { + const ready = deferred(); let calls = 0; + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { calls++; await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Private task with secret-provider-token' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); + await until(() => h.controller.runs.get(run.runId).status === 'paused'); + await h.controller.status({ runId: run.runId }); + const guard = h.local.webbrainCloudHandoffGuards[0]; + assert.deepEqual(Object.keys(guard).sort(), ['createdAt', 'pausedAt', 'runId', 'tabId', 'updatedAt']); + assert.equal(JSON.stringify(h.local).includes('secret-provider-token'), false); + delete h.stored.webbrainCloudRunSnapshots; + if (restart.includes('without')) delete h.api.storage.session; + const restarted = createCloudRunController({ chromeApi: h.api, agent: h.agent, ensureOffscreen: async () => {} }); + const restored = await restarted.status({ runId: run.runId }); + assert.equal(restored.status, 'paused'); assert.equal(restored.handoff.interrupted, true); assert.equal(restored.handoff.resumable, false); + assert.equal(restored.task, undefined); assert.equal(restored.result, undefined); assert.equal(calls, 1); + await assert.rejects(restarted.resume({ runId: run.runId }), error => error.status === 409); + await assert.rejects(restarted.startRun({ task: 'Other bot', tabId: 8 }), error => error.status === 409); + await assert.rejects(restarted.guardRunStart(8, { kind: 'scheduled' }), error => error.status === 409); + assert.equal((await restarted.abort({ runId: run.runId })).status, 'aborted'); + assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await restarted.guardRunStart(8, { kind: 'interactive' }); + await h.controller.abort({ runId: run.runId }); + }); + } + + test(`${browser}: a durable guard overrides a matching stale terminal snapshot`, async () => { + const ready = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); + await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + h.stored.webbrainCloudRunSnapshots[0].status = 'aborted'; h.stored.webbrainCloudRunSnapshots[0].handoff = null; + const restarted = createCloudRunController({ chromeApi: h.api, agent: h.agent, ensureOffscreen: async () => {} }); + const restored = await restarted.status({ runId: run.runId }); + assert.equal(restored.status, 'paused'); assert.equal(restored.handoff.interrupted, true); + await restarted.abort({ runId: run.runId }); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await h.controller.abort({ runId: run.runId }); + }); + + test(`${browser}: durable local acknowledgement blocks publication and direct resume while writing`, async () => { + const ready = deferred(), writing = deferred(), written = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async data => { if (data.webbrainCloudHandoffGuards?.[0]?.pausedAt) { writing.resolve(); await written.promise; } return write(data); }; + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); await writing.promise; + assert.equal(h.controller.runs.get(run.runId).status, 'pausing'); + let exposed = false, resumed = false; + const status = h.controller.status({ runId: run.runId }).then(value => { exposed = true; return value; }); + const returning = h.controller.resume({ runId: run.runId }).then(() => { resumed = true; }, error => { assert.equal(error.status, 409); }); + await tick(); assert.equal(exposed, false); assert.equal(resumed, false); + written.resolve(); await status; await returning; + await h.controller.abort({ runId: run.runId }); + }); + + test(`${browser}: failed guard writes retain read-only ownership and abort waits for native quiescence`, async () => { + const ready = deferred(), finished = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); await finished.promise; options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Read' }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async () => { throw Error('Fixture guard write failure'); }; + await assert.rejects(h.controller.pause({ runId: run.runId }), /guard write failure/); + assert.equal(h.controller.runs.get(run.runId).handoff.resumable, false); + await assert.rejects(h.controller.guardRunStart(8, { kind: 'scheduled' }), error => error.status === 409); + h.api.storage.local.set = write; + const aborting = await h.controller.abort({ runId: run.runId }); assert.equal(aborting.status, 'aborting'); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1, 'a cancellation request must retain the durable guard'); + ready.resolve(); await tick(); assert.equal(h.local.webbrainCloudHandoffGuards.length, 1); + finished.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'aborted'); + await h.controller.status({ runId: run.runId }); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + }); + + test(`${browser}: guard removal failures preserve the lock until explicit cancellation retries`, async () => { + const ready = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + const restarted = createCloudRunController({ chromeApi: h.api, agent: h.agent, ensureOffscreen: async () => {} }); + await restarted.status({ runId: run.runId }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async data => { if (!data.webbrainCloudHandoffGuards.length) throw Error('Fixture clear failure'); return write(data); }; + await assert.rejects(restarted.abort({ runId: run.runId }), /clear failure/); + assert.equal(restarted.runs.get(run.runId).status, 'paused'); + assert.equal(restarted.runs.get(run.runId).handoff.interrupted, true); + assert.equal(restarted.runs.get(run.runId).updates.at(-1).data.state, 'paused', 'failed cancellation cannot publish a null handoff'); + assert.equal(restarted.isRunning(7), true); + await assert.rejects(restarted.guardRunStart(8, { kind: 'interactive' }), error => error.status === 409); + h.api.storage.local.set = write; + assert.equal((await restarted.abort({ runId: run.runId })).status, 'aborted'); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await h.controller.abort({ runId: run.runId }); + }); + + test(`${browser}: handback guard-clear failure stops before any further action`, async () => { + const ready = deferred(); let observations = 0, actions = 0; + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); observations++; + await options.handoffObserved(); actions++; options.onRunFinished('done'); return 'Read'; + }); + const run = await h.controller.startRun({ task: 'Read only' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async data => { if (!data.webbrainCloudHandoffGuards.length) throw Error('Fixture observed-clear failure'); return write(data); }; + await h.controller.resume({ runId: run.runId }); await until(() => h.controller.runs.get(run.runId).handoff?.interrupted); + const stopped = await h.controller.status({ runId: run.runId }); + assert.equal(stopped.status, 'paused'); assert.equal(stopped.handoff.resumable, false); assert.equal(observations, 1); assert.equal(actions, 0); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1); + await assert.rejects(h.controller.startRun({ task: 'Another bot', tabId: 8 }), error => error.status === 409); + h.api.storage.local.set = write; await h.controller.abort({ runId: run.runId }); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + }); + + test(`${browser}: queued persistence cannot resurrect a confirmed cleared guard`, async () => { + const ready = deferred(), clearing = deferred(), cleared = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + const restarted = createCloudRunController({ chromeApi: h.api, agent: h.agent, ensureOffscreen: async () => {} }); await restarted.status({ runId: run.runId }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async data => { if (!data.webbrainCloudHandoffGuards.length) { clearing.resolve(); await cleared.promise; } return write(data); }; + const first = restarted.abort({ runId: run.runId }); await clearing.promise; + const queued = restarted.abort({ runId: run.runId }); await tick(); cleared.resolve(); + await first; await queued; await restarted.status({ runId: run.runId }); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await h.controller.abort({ runId: run.runId }); + }); + + test(`${browser}: failed or malformed durable guard reads prevent run startup`, async () => { + const valid = { runId: 'owned_1', tabId: 7, createdAt: '2026-10-12T23:00:00.000Z', updatedAt: '2026-10-12T23:00:00.000Z', pausedAt: null }; + const readFailure = Symbol('read failure'); + for (const value of [readFailure, null, false, 0, '', [{ runId: '', tabId: 7 }], [{ ...valid, runId: 123 }], [{ ...valid, runId: ' ' }], + [valid, valid], [{ ...valid, tabId: '7' }], [{ ...valid, createdAt: 'yesterday' }]]) { + let calls = 0; + const h = harness(async () => { calls++; }); + h.api.storage.local.get = async () => { if (value === readFailure) throw Error('Fixture read failure'); return { webbrainCloudHandoffGuards: value }; }; + await assert.rejects(h.controller.startRun({ task: 'Read' })); assert.equal(calls, 0); + } + }); + + test(`${browser}: failed live cancellation clear stays interrupted until owner retries`, async () => { + const ready = deferred(); + const h = harness(async (_tab, _task, update, _mode, _attachments, options) => { await ready.promise; await options.handoffBoundary(); update('clarify', { clarifyId: 'late', question: 'Late question' }); options.onRunFinished('done'); return 'Read'; }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + const write = h.api.storage.local.set; + h.api.storage.local.set = async data => { if (!data.webbrainCloudHandoffGuards.length) throw Error('Fixture live clear failure'); return write(data); }; + assert.equal((await h.controller.abort({ runId: run.runId })).status, 'aborting'); + await until(() => h.controller.runs.get(run.runId).handoff?.interrupted); + const stopped = await h.controller.status({ runId: run.runId }); + assert.equal(stopped.status, 'paused'); assert.equal(stopped.handoff.resumable, false); assert.equal(stopped.pendingInput, null); assert.equal(stopped.completedAt, null); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1); + await assert.rejects(h.controller.guardRunStart(8, { kind: 'scheduled' }), error => error.status === 409); + h.api.storage.local.set = write; assert.equal((await h.controller.abort({ runId: run.runId })).status, 'aborted'); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + }); + + test(`${browser}: valid caller correlation IDs remain usable for handoff and reload`, async () => { + for (const runId of ['request:17', 'tenant/run/1', 'browser-ü-任务']) { + const ready = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; + }); + await h.controller.startRun({ task: 'Read only', runId }); + await h.controller.pause({ runId }); ready.resolve(); + await until(() => h.controller.runs.get(runId).status === 'paused'); await h.controller.status({ runId }); + delete h.stored.webbrainCloudRunSnapshots; + const restarted = createCloudRunController({ chromeApi: h.api, agent: h.agent }); + assert.equal((await restarted.status({ runId })).handoff.interrupted, true); + await restarted.abort({ runId }); await h.controller.abort({ runId }); + } + const h = harness(async () => assert.fail('Oversized ID must not dispatch')); + await assert.rejects(h.controller.startRun({ task: 'Read', runId: 'x'.repeat(16 * 1024 + 1) }), error => error.status === 400); + assert.equal(h.controller.runs.size, 0); + }); + + test(`${browser}: cancellation during a pending pause guard write cannot resurrect manual control`, async () => { + for (const fail of [false, true]) { + const ready = deferred(), writing = deferred(), written = deferred(), finished = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); await finished.promise; options.onRunFinished('done'); return 'Read'; + }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); + const set = h.api.storage.local.set; + let failed = false; + h.api.storage.local.set = async data => { + if (data.webbrainCloudHandoffGuards?.[0]?.pausedAt) { + writing.resolve(); await written.promise; + if (fail && !failed) { failed = true; throw Error('Fixture pending pause write failure'); } + } + return set(data); + }; + ready.resolve(); await writing.promise; + const aborting = h.controller.abort({ runId: run.runId }); + await tick(); assert.equal(h.controller.runs.get(run.runId).status, 'aborting'); + await assert.rejects(h.controller.guardRunStart(8, { kind: 'scheduled' }), error => error.status === 409); + written.resolve(); await aborting; + assert.equal(h.controller.runs.get(run.runId).handoff, null, 'a delayed pause callback cannot restore writable state'); + finished.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'aborted'); + assert.equal((await h.controller.status({ runId: run.runId })).status, 'aborted'); + assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + } + }); + + test(`${browser}: cancellation stays nonterminal through native and recording cleanup`, async () => { + const ready = deferred(), settled = deferred(), flushing = deferred(), flushed = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); await settled.promise; options.onRunFinished('done'); return 'Read'; + }, { + startRecording: async () => ({ ok: true, state: { recordingId: 'capture-1' } }), + stopRecording: async () => { flushing.resolve(); await flushed.promise; return { ok: true }; }, + }); + const run = await h.controller.startRun({ task: 'Read', capture: 'video' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); + await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + assert.equal((await h.controller.abort({ runId: run.runId })).status, 'aborting'); + for (const kind of ['cloud', 'scheduled', 'interactive']) await assert.rejects(h.controller.guardRunStart(8, { kind }), error => error.status === 409); + await assert.rejects(h.controller.startRun({ task: 'Another bot', tabId: 8 }), error => error.status === 409); + settled.resolve(); await flushing.promise; + const pending = await h.controller.status({ runId: run.runId }); + assert.equal(pending.completedAt, null); assert.notEqual(pending.status, 'aborted'); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1, 'recording still owns the profile'); + flushed.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'aborted'); + await h.controller.status({ runId: run.runId }); assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await h.controller.guardRunStart(8, { kind: 'interactive' }); + }); + + test(`${browser}: a failed public paused checkpoint becomes nonresumable`, async () => { + const ready = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); options.onRunFinished('done'); return 'Read'; + }); + const run = await h.controller.startRun({ task: 'Read' }); + await h.controller.pause({ runId: run.runId }); + const set = h.api.storage.session.set; + h.api.storage.session.set = async data => { + if (data.webbrainCloudRunSnapshots?.[0]?.handoff?.resumable && data.webbrainCloudRunSnapshots[0].status === 'paused') throw Error('Fixture public checkpoint failure'); + return set(data); + }; + ready.resolve(); await until(() => h.controller.runs.get(run.runId).handoff?.interrupted); + assert.equal(h.controller.runs.get(run.runId).handoff.resumable, false); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1); + await assert.rejects(h.controller.resume({ runId: run.runId })); + h.api.storage.session.set = set; + await until(() => !h.controller.runs.get(run.runId).pendingInput); + await h.controller.abort({ runId: run.runId }); await until(() => h.controller.runs.get(run.runId).status === 'aborted'); + }); + + test(`${browser}: abort racing a fresh-observation guard clear retains ownership through recording cleanup`, async () => { + const ready = deferred(), clearing = deferred(), cleared = deferred(), continued = deferred(), flushing = deferred(), flushed = deferred(); + const h = harness(async (_tab, _task, _update, _mode, _attachments, options) => { + await ready.promise; await options.handoffBoundary(); await options.handoffObserved(); + await continued.promise; options.onRunFinished('done'); return 'Read'; + }, { + startRecording: async () => ({ ok: true, state: { recordingId: 'capture-2' } }), + stopRecording: async () => { flushing.resolve(); await flushed.promise; return { ok: true }; }, + }); + const run = await h.controller.startRun({ task: 'Read', capture: 'video' }); + await h.controller.pause({ runId: run.runId }); ready.resolve(); + await until(() => h.controller.runs.get(run.runId).status === 'paused'); await h.controller.status({ runId: run.runId }); + const set = h.api.storage.local.set; + let held = false; + h.api.storage.local.set = async data => { + if (!held && !data.webbrainCloudHandoffGuards.length) { + held = true; clearing.resolve(); await cleared.promise; + } + return set(data); + }; + await h.controller.resume({ runId: run.runId }); await clearing.promise; + const aborting = h.controller.abort({ runId: run.runId }); + await tick(); + await assert.rejects(h.controller.guardRunStart(8, { kind: 'scheduled' }), error => error.status === 409); + cleared.resolve(); + assert.equal((await aborting).status, 'aborting'); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1, 'cancellation during clear restores the durable reservation'); + continued.resolve(); await flushing.promise; + for (const kind of ['cloud', 'scheduled', 'interactive']) await assert.rejects(h.controller.guardRunStart(8, { kind }), error => error.status === 409); + await assert.rejects(h.controller.startRun({ task: 'Another bot', tabId: 8 }), error => error.status === 409); + const pending = await h.controller.status({ runId: run.runId }); + assert.equal(pending.completedAt, null); assert.notEqual(pending.status, 'aborted'); + assert.equal(h.local.webbrainCloudHandoffGuards.length, 1); + flushed.resolve(); await until(() => h.controller.runs.get(run.runId).status === 'aborted'); + assert.equal((await h.controller.status({ runId: run.runId })).status, 'aborted'); + assert.deepEqual(h.local.webbrainCloudHandoffGuards, []); + await h.controller.guardRunStart(8, { kind: 'interactive' }); + }); + + if (browser === 'firefox') test('Firefox keeps its in-process Cloud bridge transport', async () => { + const calls = []; + const h = harness(async () => assert.fail('Bridge controls must not start browser work'), { + bridge: { + start: async options => { calls.push(['start', options.url]); return { connected: true }; }, + stop: async () => { calls.push(['stop']); return { connected: false }; }, + status: async () => { calls.push(['status']); return { connected: true }; }, + }, + ensureOffscreen: () => assert.fail('Firefox has no offscreen document'), + }); + h.api.runtime = { sendMessage: () => assert.fail('Firefox uses its injected background bridge') }; + assert.equal((await h.controller.startBridge('ws://127.0.0.1:17374/extension')).connected, true); + assert.equal((await h.controller.bridgeStatus()).connected, true); + assert.equal((await h.controller.stopBridge()).connected, false); + assert.deepEqual(calls, [['start', 'ws://127.0.0.1:17374/extension'], ['status'], ['stop']]); + }); }