diff --git a/src/web-ui/src/features/dispatch/AGENTS.md b/src/web-ui/src/features/dispatch/AGENTS.md new file mode 100644 index 0000000000..ed5eefb91f --- /dev/null +++ b/src/web-ui/src/features/dispatch/AGENTS.md @@ -0,0 +1,19 @@ +# Dispatch observer verification + +For observer scheduling, event replay, approval delivery, or persisted job cache +changes, run from the repository root: + +```bash +pnpm --dir src/web-ui run test:run src/features/dispatch/DispatchJobObserver.test.ts src/features/dispatch/dispatchJobStore.test.ts src/features/dispatch/dispatchApi.test.ts src/features/dispatch/dispatch.contract.test.ts +``` + +Keep fast polling for unfinished jobs and terminal jobs whose events are not yet +drained, including hidden windows. Idle discovery must remain bounded and active +so records submitted by another process are found without a renderer event. +Explicit refreshes and local submissions/follow-ups must bypass the idle delay. +Retain single-flight reconciliation, observer teardown, per-observer cursors, +legacy record recovery, and controller polling suppression in Peer Device Mode. + +These tests use simulated clocks and host/target APIs. They verify scheduling and +projection contracts, not native CPU usage, OS notification display, or live SSH, +account-device, remote-control, and Detached Dispatch integration. diff --git a/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts b/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts index 8bc03e580d..1597cd5302 100644 --- a/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts +++ b/src/web-ui/src/features/dispatch/DispatchJobObserver.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DISPATCH_JOB_POLL_INTERVAL_MS, + DISPATCH_IDLE_DISCOVERY_INTERVAL_MS, dispatchEventId, installDispatchJobObserver, projectDispatchAgentEvent, @@ -30,6 +31,7 @@ import { } from '@/flow_chat/store/runtimeStatusStore'; const mocks = vi.hoisted(() => ({ + peerModeActive: false, listJobs: vi.fn(), status: vi.fn(), loadTranscript: vi.fn(), @@ -49,7 +51,7 @@ vi.mock('./dispatchApi', () => ({ })); vi.mock('@/infrastructure/peer-device/peerModeFlag', () => ({ - isPeerDeviceModeActive: () => false, + isPeerDeviceModeActive: () => mocks.peerModeActive, })); vi.mock('@/flow_chat/services/AgenticEventListener', () => ({ @@ -330,6 +332,7 @@ function createDeferred() { describe('DispatchJobObserver', () => { beforeEach(() => { vi.useFakeTimers(); + mocks.peerModeActive = false; dispatchJobStore.getState().clear(); useRuntimeStatusStore.getState().reset(); flowChatStore.setState(() => ({ @@ -360,6 +363,248 @@ describe('DispatchJobObserver', () => { vi.useRealTimers(); }); + describe('poll scheduling', () => { + let cleanup: (() => void) | undefined; + + afterEach(() => { + cleanup?.(); + cleanup = undefined; + }); + + it.each(['visible', 'hidden'])('discovers idle jobs at a bounded low rate while %s', async visibility => { + Object.defineProperty(document, 'visibilityState', { configurable: true, value: visibility }); + cleanup = installDispatchJobObserver(createContext()); + + await vi.advanceTimersByTimeAsync(60_000); + + // One initial discovery and two 30-second checks, previously 34 reads. + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + expect(mocks.status).not.toHaveBeenCalled(); + cleanup(); + registerRunningJob(); + await vi.advanceTimersByTimeAsync(60_000); + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + }); + + it.each(['queued', 'running', 'submission_unknown'] as const)('keeps the fast cadence for %s jobs', async state => { + registerRunningJob(); + dispatchJobStore.getState().registerJob({ ...dispatchJobStore.getState().jobs['job-1'], state }); + mocks.listJobs.mockResolvedValue([{ ...runningOutboundRecord(), lastState: state }]); + mocks.status.mockResolvedValue(status({ state })); + cleanup = installDispatchJobObserver(createContext()); + + await vi.advanceTimersByTimeAsync(3 * DISPATCH_JOB_POLL_INTERVAL_MS); + + expect(mocks.listJobs).toHaveBeenCalledTimes(4); + expect(mocks.status).toHaveBeenCalledTimes(4); + }); + + it.each(['visible', 'hidden'])('finds another instance\'s durable job without a renderer event while %s', async visibility => { + Object.defineProperty(document, 'visibilityState', { configurable: true, value: visibility }); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + + // Only the host response changes: no store mutation, storage event or + // explicit refresh from the submitting instance reaches this renderer. + mocks.listJobs.mockResolvedValue([{ ...runningOutboundRecord(), lastCursor: 99 }]); + mocks.status.mockResolvedValue(status()); + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS - 1); + expect(mocks.listJobs).toHaveBeenCalledTimes(1); + expect(mocks.status).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(mocks.status).toHaveBeenCalledWith('job-1', 0); + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(mocks.status).toHaveBeenCalledTimes(2); + }); + + it('wakes immediately when this renderer registers a job during idle discovery', async () => { + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + registerRunningJob(); + mocks.status.mockResolvedValue(status()); + + await vi.advanceTimersByTimeAsync(0); + expect(mocks.status).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(mocks.status).toHaveBeenCalledTimes(2); + }); + + it('waits for terminal event draining before reducing the polling rate', async () => { + registerRunningJob(); + mocks.status.mockResolvedValueOnce(status({ + state: 'succeeded', cursor: 12, + events: [{ type: 'jobState', state: 'succeeded', timestamp: '2026-07-28T00:00:00Z' }], + })).mockResolvedValue(status({ state: 'succeeded', cursor: 12 })); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + expect(dispatchJobStore.getState().jobs['job-1'].terminalDrained).toBe(false); + + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(mocks.status).toHaveBeenCalledTimes(2); + expect(dispatchJobStore.getState().jobs['job-1'].terminalDrained).toBe(true); + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS - 1); + expect(mocks.listJobs).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + expect(mocks.status).toHaveBeenCalledTimes(2); + + // A local follow-up must wake a terminal job without waiting 30 seconds. + dispatchJobStore.getState().markFollowUpAccepted('job-1', 'queued'); + mocks.listJobs.mockResolvedValue([{ ...runningOutboundRecord(), lastState: 'queued' }]); + mocks.status.mockResolvedValue(status({ state: 'running', cursor: 12 })); + await vi.advanceTimersByTimeAsync(0); + expect(mocks.status).toHaveBeenCalledTimes(3); + }); + + it('preserves background progress, approval delivery and completion notifications', async () => { + registerRunningJob(); + Object.defineProperty(document, 'visibilityState', { configurable: true, value: 'hidden' }); + const context = createContext(); + const permission = { requestId: 'approval-1', type: 'tool' }; + context.flowChatStore.getState().sessions.get('session-1').dialogTurns = []; + mocks.status.mockResolvedValue(status()); + cleanup = installDispatchJobObserver(context); + await vi.advanceTimersByTimeAsync(0); + mocks.status.mockResolvedValueOnce(status({ + cursor: 10, + events: [{ + type: 'agentEvent', timestamp: '2026-07-28T00:00:00Z', + event: { id: 'progress-1', frontendEventName: 'agentic://text-chunk', + frontendPayload: { sessionId: 'session-1', turnId: 'turn-1', roundId: 'round-1', text: 'working' } }, + }], + pendingPermissions: [permission], + })).mockResolvedValue(status({ cursor: 10, pendingPermissions: [permission] })); + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(mocks.dispatchExternal).toHaveBeenCalledWith('agentic://text-chunk', expect.objectContaining({ text: 'working' })); + expect(dispatchJobStore.getState().jobs['job-1'].pendingPermissions).toEqual([permission]); + expect(mocks.sendSystemNotification).toHaveBeenCalledTimes(1); + + // Approval acknowledgement requests a refresh immediately, not on the timer. + mocks.status.mockResolvedValue(status({ cursor: 10 })); + requestDispatchJobRefresh('job-1'); + await vi.advanceTimersByTimeAsync(0); + expect(dispatchJobStore.getState().jobs['job-1'].pendingPermissions).toEqual([]); + mocks.status.mockResolvedValue(status({ state: 'succeeded', cursor: 10 })); + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(context.flowChatStore.markSessionUnreadCompletion).toHaveBeenCalledWith('session-1', 'completed'); + expect(mocks.sendSystemNotification).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.sendSystemNotification).toHaveBeenCalledTimes(2); + }); + + it('retains full refresh requests across a slow read without overlapping polls', async () => { + const discovery = createDeferred[]>(); + mocks.listJobs.mockReturnValueOnce(discovery.promise); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + requestDispatchJobRefresh('job-1'); + requestDispatchJobRefresh(); + await vi.advanceTimersByTimeAsync(2 * DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(1); + discovery.resolve([]); + await vi.advanceTimersByTimeAsync(0); + expect(mocks.listJobs).toHaveBeenCalledTimes(2); + }); + + it('coalesces refreshes for different jobs during a slow target read without losing either', async () => { + registerRunningJob(); + dispatchJobStore.getState().registerJob({ + ...dispatchJobStore.getState().jobs['job-1'], jobId: 'job-2', sessionId: 'session-2', + }); + mocks.listJobs.mockResolvedValue([ + runningOutboundRecord(), + { ...runningOutboundRecord(), jobId: 'job-2', sessionId: 'session-2' }, + ]); + const context = createContext(); + const sessions = context.flowChatStore.getState().sessions; + sessions.set('session-2', { + ...sessions.get('session-1'), sessionId: 'session-2', + config: { ...sessions.get('session-1').config, dispatchJobId: 'job-2' }, + }); + mocks.status.mockResolvedValue(status()); + cleanup = installDispatchJobObserver(context); + await vi.advanceTimersByTimeAsync(0); + mocks.status.mockClear(); + mocks.listJobs.mockClear(); + + const targetRead = createDeferred(); + mocks.status.mockReturnValueOnce(targetRead.promise); + requestDispatchJobRefresh('job-1'); + await vi.advanceTimersByTimeAsync(0); + requestDispatchJobRefresh('job-1'); + requestDispatchJobRefresh('job-2'); + await vi.advanceTimersByTimeAsync(2 * DISPATCH_JOB_POLL_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(1); + expect(mocks.status).toHaveBeenCalledTimes(1); + + targetRead.resolve(status()); + await vi.advanceTimersByTimeAsync(0); + expect(mocks.listJobs).toHaveBeenCalledTimes(2); + expect(mocks.status.mock.calls).toEqual([ + ['job-1', 0], ['job-1', 0], ['job-2', 0], + ]); + }); + + it('replays a job submitted and completed by another instance between idle discoveries', async () => { + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + mocks.listJobs.mockResolvedValue([{ + ...runningOutboundRecord(), lastState: 'succeeded', lastCursor: 99, + }]); + mocks.status.mockResolvedValueOnce(status({ + state: 'succeeded', cursor: 99, + events: [{ type: 'jobState', state: 'succeeded', timestamp: '2026-07-28T00:00:00Z' }], + })).mockResolvedValue(status({ state: 'succeeded', cursor: 99 })); + + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.status).toHaveBeenCalledWith('job-1', 0); + await vi.advanceTimersByTimeAsync(DISPATCH_JOB_POLL_INTERVAL_MS); + expect(dispatchJobStore.getState().jobs['job-1']).toMatchObject({ + cursor: 99, state: 'succeeded', terminalDrained: true, + }); + }); + + it('does not revive a disposed observer when its discovery or queued refresh finishes', async () => { + const discovery = createDeferred[]>(); + mocks.listJobs.mockReturnValueOnce(discovery.promise); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + requestDispatchJobRefresh(); + cleanup(); + discovery.resolve([runningOutboundRecord()]); + + await vi.advanceTimersByTimeAsync(2 * DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(1); + expect(mocks.status).not.toHaveBeenCalled(); + expect(dispatchJobStore.getState().jobs).toEqual({}); + }); + + it('refreshes on visibility recovery and retries failed idle discovery', async () => { + mocks.listJobs.mockRejectedValueOnce(new Error('temporary index failure')); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(2); + document.dispatchEvent(new Event('visibilitychange')); + await vi.advanceTimersByTimeAsync(0); + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + }); + + it('keeps controller polling suppressed in peer mode and replaces observers cleanly', async () => { + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(0); + cleanup = installDispatchJobObserver(createContext()); + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + mocks.peerModeActive = true; + requestDispatchJobRefresh(); + await vi.advanceTimersByTimeAsync(2 * DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(3); + mocks.peerModeActive = false; + await vi.advanceTimersByTimeAsync(DISPATCH_IDLE_DISCOVERY_INTERVAL_MS); + expect(mocks.listJobs).toHaveBeenCalledTimes(4); + }); + }); + it.each(['ssh', 'device'] as const)('persists %s generated titles and restores them after projection recreation', async (kind) => { registerRunningJob(); if (kind === 'device') { diff --git a/src/web-ui/src/features/dispatch/DispatchJobObserver.ts b/src/web-ui/src/features/dispatch/DispatchJobObserver.ts index bf4f45864e..b31761bda2 100644 --- a/src/web-ui/src/features/dispatch/DispatchJobObserver.ts +++ b/src/web-ui/src/features/dispatch/DispatchJobObserver.ts @@ -37,6 +37,7 @@ import { const log = createLogger('DispatchJobObserver'); export const DISPATCH_JOB_POLL_INTERVAL_MS = 1800; +export const DISPATCH_IDLE_DISCOVERY_INTERVAL_MS = 30_000; const BASELINE_WORKTREE_CHECK_INTERVAL_MS = 30_000; const baselineWorktreeChecks = new Map(); @@ -1049,9 +1050,13 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void let disposed = false; let inFlight = false; - let queuedJobId: string | undefined; + // A full refresh (jobId undefined) is still a request. Coalesce different + // targeted requests into a full refresh instead of losing an approval wakeup. + let pendingRefresh: { jobId?: string } | null = null; let immediateTimer: ReturnType | null = null; let interval: ReturnType | null = null; + let pollIntervalMs = 0; + let unsubscribeJobs: (() => void) | null = null; let handleVisibilityChanged: (() => void) | null = null; const lease: DispatchObserverLease = { requestRefresh: schedule, @@ -1063,11 +1068,7 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void ); async function run(requestedJobId?: string): Promise { - if (!ownsLease() || isPeerDeviceModeActive()) return; - if (inFlight) { - queuedJobId = requestedJobId; - return; - } + if (!ownsLease() || isPeerDeviceModeActive() || inFlight) return; inFlight = true; try { @@ -1112,22 +1113,46 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void log.warn('Failed to reconcile outbound dispatch jobs', { error }); } finally { inFlight = false; - if (queuedJobId !== undefined && ownsLease()) { - const next = queuedJobId; - queuedJobId = undefined; - schedule(next); + if (ownsLease()) { + updatePollingInterval(); + if (pendingRefresh) schedule(pendingRefresh.jobId); } } } + function updatePollingInterval(): void { + if (!ownsLease()) return; + const hasUnsettledJobs = Object.values(dispatchJobStore.getState().jobs).some(job => ( + !isDispatchJobTerminal(job.state) || !job.terminalDrained + )); + const nextIntervalMs = hasUnsettledJobs + ? DISPATCH_JOB_POLL_INTERVAL_MS + : DISPATCH_IDLE_DISCOVERY_INTERVAL_MS; + if (interval !== null && pollIntervalMs === nextIntervalMs) return; + if (interval !== null) clearInterval(interval); + pollIntervalMs = nextIntervalMs; + // Idle discovery remains active while hidden: another controller process + // can submit a job without updating this renderer's store or sending events. + interval = setInterval(() => { void run(); }, nextIntervalMs); + } + function schedule(jobId?: string): void { if (!ownsLease()) return; + pendingRefresh = pendingRefresh === null || pendingRefresh.jobId === jobId + ? { jobId } + : {}; + // Keep explicit refreshes queued across a slow read. Periodic ticks never + // overwrite this queue or start a second in-flight reconciliation. + if (inFlight) return; if (immediateTimer !== null) { clearTimeout(immediateTimer); } immediateTimer = setTimeout(() => { immediateTimer = null; - void run(jobId); + if (inFlight) return; + const requestedJobId = pendingRefresh?.jobId; + pendingRefresh = null; + void run(requestedJobId); }, 0); } @@ -1147,6 +1172,9 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void clearInterval(interval); interval = null; } + pendingRefresh = null; + unsubscribeJobs?.(); + unsubscribeJobs = null; if (typeof document !== 'undefined' && handleVisibilityChanged) { document.removeEventListener('visibilitychange', handleVisibilityChanged); } @@ -1157,9 +1185,15 @@ export function installDispatchJobObserver(context: FlowChatContext): () => void } observerGlobal.__openbitfunDispatchJobObserverLease__ = lease; - interval = setInterval(() => { - void run(); - }, DISPATCH_JOB_POLL_INTERVAL_MS); + updatePollingInterval(); + unsubscribeJobs = dispatchJobStore.subscribe(() => { + // run() updates the store repeatedly while replaying events. Re-evaluate + // once in finally, rather than treating its own writes as new work. + if (inFlight || !ownsLease()) return; + const wasIdle = pollIntervalMs === DISPATCH_IDLE_DISCOVERY_INTERVAL_MS; + updatePollingInterval(); + if (wasIdle && pollIntervalMs === DISPATCH_JOB_POLL_INTERVAL_MS) schedule(); + }); handleVisibilityChanged = () => { if (typeof document === 'undefined' || document.visibilityState === 'visible') { schedule();