From 37b4ab89f067b29d715ddd60c92f1d082abd0a91 Mon Sep 17 00:00:00 2001 From: Chris Scott <> Date: Thu, 15 Oct 2026 13:00:08 +0000 Subject: [PATCH 1/3] fix(background-activity): reconcile shells for directories subscribed by the stream The background activity indicator went stale for sessions living in a directory that is not a registered repo. Running shells were only synced for repo directories derived from the repos query, so session directories subscribed over the event stream were never reconciled, leaving the indicator visible after the shell had exited. Report the directories actually subscribed (mid-connection updates and pending directories flushed after connect) from the event stream through onDirectoriesSubscribed, and reconcile running shells and pending permissions/forms for them in the event context. --- e2e/background-activity.spec.ts | 80 +++++++++++++++++++ frontend/src/contexts/EventContext.test.tsx | 65 ++++++++++++++- frontend/src/contexts/EventContext.tsx | 17 ++-- .../__tests__/openCodeEventStream.test.ts | 56 +++++++++++++ .../openCodeEventStream.ts | 43 ++++++---- 5 files changed, 232 insertions(+), 29 deletions(-) diff --git a/e2e/background-activity.spec.ts b/e2e/background-activity.spec.ts index ddd6d44a5..1e4f6be64 100644 --- a/e2e/background-activity.spec.ts +++ b/e2e/background-activity.spec.ts @@ -1,5 +1,9 @@ +import { execFileSync } from 'node:child_process' +import fs from 'node:fs' +import path from 'node:path' import type { APIRequestContext, Page } from '@playwright/test' import { test, expect } from './fixtures' +import { E2E_TMP } from './env' const SHELL_COMMAND = 'sleep 300' @@ -54,12 +58,38 @@ async function removeShell(request: APIRequestContext, directory: string, shellI } const createdShells: Array<{ directory: string; id: string }> = [] +const createdDirectories: string[] = [] +let directoryCounter = 0 + +function createUnregisteredDirectory(name: string): string { + directoryCounter += 1 + const directory = path.join(E2E_TMP, `background-${name}-${process.pid}-${directoryCounter}`) + fs.mkdirSync(directory, { recursive: true }) + fs.writeFileSync(path.join(directory, 'README.md'), '# background activity fixture\n') + + const git = (...args: string[]) => execFileSync('git', args, { cwd: directory, stdio: 'pipe' }) + git('-c', 'init.defaultBranch=main', 'init', '-q') + git('-c', 'user.name=e2e', '-c', 'user.email=e2e@example.com', 'add', '-A') + git('-c', 'user.name=e2e', '-c', 'user.email=e2e@example.com', 'commit', '-q', '-m', 'init') + + createdDirectories.push(directory) + return directory +} + +async function navigateWithinApp(page: Page, url: string): Promise { + await page.evaluate((target) => { + window.history.pushState({}, '', target) + window.dispatchEvent(new PopStateEvent('popstate')) + }, url) +} test.afterEach(async ({ request }) => { const shells = createdShells.splice(0) const results = await Promise.allSettled(shells.map(({ directory, id }) => removeShell(request, directory, id))) const failures = results.flatMap((result) => (result.status === 'rejected' ? [String(result.reason)] : [])) expect(failures, 'background shell cleanup failed').toEqual([]) + + createdDirectories.splice(0).forEach((directory) => fs.rmSync(directory, { recursive: true, force: true })) }) test.describe('session background activity', () => { @@ -138,4 +168,54 @@ test.describe('session background activity', () => { await expect(backgroundIndicator(page)).toHaveCount(0) await expect(backgroundStatus(page)).toHaveCount(0) }) + + test('shows the indicator for a shell started in an unregistered session directory before page load', async ({ + page, + request, + repo, + }) => { + const directory = createUnregisteredDirectory('preload') + const sessionId = await createSession(request, directory) + const shellId = await startBackgroundShell(request, directory, sessionId) + createdShells.push({ directory, id: shellId }) + + await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) + await expect(page.getByPlaceholder('Send a message...')).toBeVisible() + + await expect(backgroundIndicator(page)).toBeVisible() + + await removeShell(request, directory, shellId) + + await expect(backgroundIndicator(page)).toBeHidden() + }) + + test('clears the indicator for a shell that exits while an unregistered session directory is unsubscribed', async ({ + page, + request, + repo, + }) => { + const directory = createUnregisteredDirectory('stale') + const sessionId = await createSession(request, directory) + const subscribed = page.waitForResponse((response) => + response.url().includes('/api/sse/subscribe') && (response.request().postData() ?? '').includes(directory), + ) + + await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) + await expect(page.getByPlaceholder('Send a message...')).toBeVisible() + await subscribed + + const shellId = await startBackgroundShell(request, directory, sessionId) + createdShells.push({ directory, id: shellId }) + + await expect(backgroundIndicator(page)).toBeVisible() + + await navigateWithinApp(page, '/') + await page.waitForTimeout(3_000) + await removeShell(request, directory, shellId) + await page.waitForTimeout(3_000) + await navigateWithinApp(page, `/repos/${repo.id}/sessions/${sessionId}`) + + await expect(page.getByPlaceholder('Send a message...')).toBeVisible() + await expect(backgroundIndicator(page)).toBeHidden({ timeout: 20_000 }) + }) }) diff --git a/frontend/src/contexts/EventContext.test.tsx b/frontend/src/contexts/EventContext.test.tsx index e8b93176c..d3d2ee7a8 100644 --- a/frontend/src/contexts/EventContext.test.tsx +++ b/frontend/src/contexts/EventContext.test.tsx @@ -1865,11 +1865,18 @@ describe('EventProvider permissions and forms', () => { render(, { wrapper: createWrapper() }) - await waitFor(() => expect(mocks.listShells).toHaveBeenCalledWith('/repo')) + await waitFor(() => expect(mocks.subscribeGlobalMonitor).toHaveBeenCalled()) const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] + const onDirectoriesSubscribed = lastSubscribeCall[0].onDirectoriesSubscribed as (directories: string[]) => void const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void + act(() => { + onDirectoriesSubscribed(['/repo']) + }) + + await waitFor(() => expect(mocks.listShells).toHaveBeenCalledWith('/repo')) + act(() => { onEvent({ type: 'shell.exited', data: { id: 'shell-1', status: 'exited' }, directory: '/repo' }) }) @@ -1889,11 +1896,18 @@ describe('EventProvider permissions and forms', () => { render(, { wrapper: createWrapper() }) - await waitFor(() => expect(mocks.listShells).toHaveBeenCalledWith('/repo')) + await waitFor(() => expect(mocks.subscribeGlobalMonitor).toHaveBeenCalled()) const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] + const onDirectoriesSubscribed = lastSubscribeCall[0].onDirectoriesSubscribed as (directories: string[]) => void const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void + act(() => { + onDirectoriesSubscribed(['/repo']) + }) + + await waitFor(() => expect(mocks.listShells).toHaveBeenCalledWith('/repo')) + act(() => { onEvent({ type: 'shell.created', @@ -1913,7 +1927,7 @@ describe('EventProvider permissions and forms', () => { }) }) - it('resolves ancestry for a shell owner from the directory snapshot', async () => { + it('resolves ancestry for a shell owner from a newly subscribed directory snapshot', async () => { mocks.listRepos.mockResolvedValue([{ id: 123, fullPath: '/repo' }]) mocks.listShells.mockResolvedValue([ { id: 'shell-1', status: 'running', metadata: { sessionID: 'child-1' } }, @@ -1926,6 +1940,15 @@ describe('EventProvider permissions and forms', () => { render(, { wrapper: createWrapper() }) + await waitFor(() => expect(mocks.subscribeGlobalMonitor).toHaveBeenCalled()) + + const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] + const onDirectoriesSubscribed = lastSubscribeCall[0].onDirectoriesSubscribed as (directories: string[]) => void + + act(() => { + onDirectoriesSubscribed(['/repo']) + }) + await waitFor(() => { expect(useSessionStatus.getState().parents.get('child-1')).toBe('parent-1') }) @@ -1967,6 +1990,12 @@ describe('EventProvider permissions and forms', () => { it('syncs running shells for a directory whose repo arrives after connect', async () => { const reposDeferred = createDeferred>() mocks.listRepos.mockReturnValueOnce(reposDeferred.promise) + mocks.subscribeGlobalMonitor.mockImplementation((input: { onDirectoriesSubscribed?: (directories: string[]) => void }) => ({ + dispose: vi.fn(), + updateDirectories: (directories: string[]) => input.onDirectoriesSubscribed?.(directories), + reconnect: vi.fn(), + reportVisibility: vi.fn(), + })) render(, { wrapper: createWrapper() }) @@ -1988,4 +2017,34 @@ describe('EventProvider permissions and forms', () => { await waitFor(() => expect(mocks.listShells).toHaveBeenCalledWith('/repo')) }) + + it('reconciles newly subscribed directories reported by the stream', async () => { + mocks.listPendingPermissions.mockResolvedValue([pendingPermission]) + mocks.listPendingForms.mockResolvedValue([pendingForm]) + + render(, { wrapper: createWrapper() }) + + await waitFor(() => expect(mocks.subscribeGlobalMonitor).toHaveBeenCalled()) + + const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] + const onDirectoriesSubscribed = lastSubscribeCall[0].onDirectoriesSubscribed as (directories: string[]) => void + + act(() => { + useSessionStatus.getState().setShellRunning('shell-stale', 'session-1', '/session-dir') + }) + expect(useSessionStatus.getState().runningShells.has('shell-stale')).toBe(true) + + await act(async () => { + onDirectoriesSubscribed(['/session-dir']) + }) + + await waitFor(() => { + expect(mocks.listShells).toHaveBeenCalledWith('/session-dir') + expect(mocks.listPendingPermissions).toHaveBeenCalledWith('/session-dir') + expect(mocks.listPendingForms).toHaveBeenCalledWith('/session-dir') + }) + expect(useSessionStatus.getState().runningShells.has('shell-stale')).toBe(false) + expect(screen.getByTestId('permission-count').textContent).toBe('1') + expect(screen.getByTestId('count').textContent).toBe('1') + }) }) diff --git a/frontend/src/contexts/EventContext.tsx b/frontend/src/contexts/EventContext.tsx index 307451829..e8db0e53d 100644 --- a/frontend/src/contexts/EventContext.tsx +++ b/frontend/src/contexts/EventContext.tsx @@ -206,7 +206,6 @@ export function EventProvider({ children }: { children: React.ReactNode }) { const statusPollIntervalRef = useRef | null>(null) const subscriptionRef = useRef | null>(null) const reposRef = useRef(null) - const trackedRepoDirectoriesRef = useRef>(new Set()) const ancestryInFlightRef = useRef>(new Set()) const { data: repos } = useQuery({ @@ -725,6 +724,10 @@ export function EventProvider({ children }: { children: React.ReactNode }) { onStatusChange: handleStatusChange, onHealthChange: handleHealthChange, onResync: handleResync, + onDirectoriesSubscribed: (directories) => { + void reconcilePendingActionsForDirectories(directories) + void syncRunningShells(directories) + }, }) subscriptionRef.current = subscription @@ -739,16 +742,8 @@ export function EventProvider({ children }: { children: React.ReactNode }) { useEffect(() => { reposRef.current = repos const directories = [...new Set((repos ?? []).map(r => r.fullPath))] - const previousDirectories = trackedRepoDirectoriesRef.current - const addedDirectories = directories.filter((directory) => !previousDirectories.has(directory)) - trackedRepoDirectoriesRef.current = new Set(directories) - const sub = subscriptionRef.current - if (!sub) return - sub.updateDirectories(directories) - if (addedDirectories.length > 0) { - void syncRunningShells(addedDirectories) - } - }, [repos, syncRunningShells]) + subscriptionRef.current?.updateDirectories(directories) + }, [repos]) useEffect(() => { if (!repos || repos.length === 0) return diff --git a/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts b/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts index 5d41f0bfc..20255f7b0 100644 --- a/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts +++ b/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts @@ -2,6 +2,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { OpenCodeEventStream, TestEventStreamTransport } from '..' import type { EventStreamHealthState } from '..' +class RejectingSubscribeTransport extends TestEventStreamTransport { + override async post(path: string, body: unknown): Promise { + this.posts.push({ path, body }) + return false + } +} + describe('OpenCodeEventStream', () => { beforeEach(() => { vi.useFakeTimers() @@ -186,4 +193,53 @@ describe('OpenCodeEventStream', () => { body: { clientId: 'client-1', visible: true, activeSessionId: 'session-1' }, }) }) + + it('notifies the global monitor with directories subscribed mid-connection', async () => { + const transport = new TestEventStreamTransport() + const stream = new OpenCodeEventStream({ transport }) + const onDirectoriesSubscribed = vi.fn() + + const subscription = stream.subscribeGlobalMonitor({ + directories: ['/repo-a'], + onEvent: vi.fn(), + onDirectoriesSubscribed, + }) + transport.openConnection() + transport.connected('client-1') + await vi.advanceTimersByTimeAsync(0) + onDirectoriesSubscribed.mockClear() + + subscription.updateDirectories(['/repo-a', '/repo-b']) + await vi.advanceTimersByTimeAsync(0) + + expect(onDirectoriesSubscribed).toHaveBeenCalledWith(['/repo-b']) + }) + + it('notifies the global monitor for pending directories flushed after connect', async () => { + const transport = new TestEventStreamTransport() + const stream = new OpenCodeEventStream({ transport }) + const onDirectoriesSubscribed = vi.fn() + + stream.subscribeGlobalMonitor({ directories: ['/repo'], onEvent: vi.fn(), onDirectoriesSubscribed }) + expect(onDirectoriesSubscribed).not.toHaveBeenCalled() + + transport.openConnection() + transport.connected('client-1') + await vi.advanceTimersByTimeAsync(0) + + expect(onDirectoriesSubscribed).toHaveBeenCalledWith(['/repo']) + }) + + it('does not notify the global monitor when the subscribe request fails', async () => { + const transport = new RejectingSubscribeTransport() + const stream = new OpenCodeEventStream({ transport }) + const onDirectoriesSubscribed = vi.fn() + + stream.subscribeGlobalMonitor({ directories: ['/repo'], onEvent: vi.fn(), onDirectoriesSubscribed }) + transport.openConnection() + transport.connected('client-1') + await vi.advanceTimersByTimeAsync(0) + + expect(onDirectoriesSubscribed).not.toHaveBeenCalled() + }) }) diff --git a/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts b/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts index 715f058a7..8846f1df8 100644 --- a/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts +++ b/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts @@ -15,9 +15,19 @@ interface Subscriber { onStatusChange?: EventStreamStatusHandler onHealthChange?: (state: EventStreamHealthState) => void onResync?: () => void + onDirectoriesSubscribed?: (directories: string[]) => void directories: Set } +interface SubscriberInput { + onEvent: OpenCodeEventHandler + onStatusChange?: EventStreamStatusHandler + onHealthChange?: (state: EventStreamHealthState) => void + directories: string[] + onResync?: () => void + onDirectoriesSubscribed?: (directories: string[]) => void +} + interface OpenCodeEventStreamOptions { transport?: EventStreamTransport } @@ -45,14 +55,8 @@ export class OpenCodeEventStream { this.transport = options.transport ?? createBrowserEventStreamTransport() } - subscribeGlobalMonitor(input: { - directories: string[] - onEvent: OpenCodeEventHandler - onStatusChange?: EventStreamStatusHandler - onHealthChange?: (state: EventStreamHealthState) => void - onResync?: () => void - }): GlobalMonitorSubscription { - const id = this.addSubscriber(input.onEvent, input.onStatusChange, input.onHealthChange, input.directories, input.onResync) + subscribeGlobalMonitor(input: SubscriberInput): GlobalMonitorSubscription { + const id = this.addSubscriber(input) return { updateDirectories: (directories) => this.updateSubscriberDirectories(id, directories), @@ -66,13 +70,8 @@ export class OpenCodeEventStream { return this.buildHealth() } - private addSubscriber( - onEvent: OpenCodeEventHandler, - onStatusChange?: EventStreamStatusHandler, - onHealthChange?: (state: EventStreamHealthState) => void, - directories: string[] = [], - onResync?: () => void, - ): string { + private addSubscriber(input: SubscriberInput): string { + const { onEvent, onStatusChange, onHealthChange, directories, onResync, onDirectoriesSubscribed } = input const id = `sub_${++this.subscriberIdCounter}` const initialDirectories = new Set(directories.filter(Boolean)) @@ -96,6 +95,7 @@ export class OpenCodeEventStream { onStatusChange, onHealthChange, onResync, + onDirectoriesSubscribed, directories: initialDirectories, }) @@ -430,7 +430,20 @@ export class OpenCodeEventStream { if (!success) { directories.forEach((directory) => this.pendingDirectories.add(directory)) this.reconnect() + return } + + this.notifyDirectoriesSubscribed(directories) + } + + private notifyDirectoriesSubscribed(directories: string[]): void { + this.subscribers.forEach((subscriber) => { + try { + subscriber.onDirectoriesSubscribed?.(directories) + } catch { + void 0 + } + }) } private flushPendingDirectories(): void { From 74a2f6fb6ec3c91f176086832ae5119d54736620 Mon Sep 17 00:00:00 2001 From: Chris Scott <> Date: Thu, 15 Oct 2026 13:25:01 +0000 Subject: [PATCH 2/3] fix(session): follow a moved session's new directory without a reload session.moved is published on the old directory and carries the new one in data.location, so it reached the page. useSSE refreshed the cached session on session.renamed but had no session.moved case, and the event context only invalidated session lists, so the page kept using the old directory and never subscribed the new one. Shells, prompts and other directory-scoped events in the new directory were missed until a refocus, reconnect or reload. Handle session.moved like session.renamed. The refetch flips the page's sessionDirectory, it subscribes the new directory, and the stream's onDirectoriesSubscribed reconciliation loads that directory's running shells and pending prompts. --- e2e/background-activity.spec.ts | 39 +++++++++++++++++++++++++++--- frontend/src/hooks/useSSE.test.tsx | 34 ++++++++++++++++++++++++++ frontend/src/hooks/useSSE.ts | 1 + 3 files changed, 70 insertions(+), 4 deletions(-) diff --git a/e2e/background-activity.spec.ts b/e2e/background-activity.spec.ts index 1e4f6be64..aa5814c98 100644 --- a/e2e/background-activity.spec.ts +++ b/e2e/background-activity.spec.ts @@ -23,7 +23,7 @@ async function requireOk(response: Awaited if (!response.ok()) { throw new Error(`${label} failed: ${response.status()} ${await response.text()}`) } - return response.json() + return response.status() === 204 ? null : response.json() } async function createSession(request: APIRequestContext, directory: string, parentID?: string): Promise { @@ -76,6 +76,12 @@ function createUnregisteredDirectory(name: string): string { return directory } +function waitForDirectorySubscription(page: Page, directory: string) { + return page.waitForResponse((response) => + response.url().includes('/api/sse/subscribe') && (response.request().postData() ?? '').includes(directory), + ) +} + async function navigateWithinApp(page: Page, url: string): Promise { await page.evaluate((target) => { window.history.pushState({}, '', target) @@ -196,9 +202,7 @@ test.describe('session background activity', () => { }) => { const directory = createUnregisteredDirectory('stale') const sessionId = await createSession(request, directory) - const subscribed = page.waitForResponse((response) => - response.url().includes('/api/sse/subscribe') && (response.request().postData() ?? '').includes(directory), - ) + const subscribed = waitForDirectorySubscription(page, directory) await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) await expect(page.getByPlaceholder('Send a message...')).toBeVisible() @@ -218,4 +222,31 @@ test.describe('session background activity', () => { await expect(page.getByPlaceholder('Send a message...')).toBeVisible() await expect(backgroundIndicator(page)).toBeHidden({ timeout: 20_000 }) }) + + test('follows a session moved to another directory without a reload', async ({ page, request, repo }) => { + const source = createUnregisteredDirectory('move-source') + const target = createUnregisteredDirectory('move-target') + const sessionId = await createSession(request, source) + const subscribedToSource = waitForDirectorySubscription(page, source) + + await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) + await expect(page.getByPlaceholder('Send a message...')).toBeVisible() + await subscribedToSource + + const subscribedToTarget = waitForDirectorySubscription(page, target) + await requireOk( + await request.post(`/api/opencode/api/session/${sessionId}/move`, { data: { directory: target } }), + 'move session', + ) + await subscribedToTarget + + const shellId = await startBackgroundShell(request, target, sessionId) + createdShells.push({ directory: target, id: shellId }) + + await expect(backgroundIndicator(page)).toBeVisible() + + await removeShell(request, target, shellId) + + await expect(backgroundIndicator(page)).toBeHidden() + }) }) diff --git a/frontend/src/hooks/useSSE.test.tsx b/frontend/src/hooks/useSSE.test.tsx index d014d512a..a5997fa34 100644 --- a/frontend/src/hooks/useSSE.test.tsx +++ b/frontend/src/hooks/useSSE.test.tsx @@ -174,6 +174,40 @@ describe('useSSE', () => { unmount() }) + it('invalidates the cached session query on session.moved so the page follows the new directory', async () => { + const queryClient = new QueryClient({ + defaultOptions: { + queries: { retry: false }, + }, + }) + const invalidateQueries = vi.spyOn(queryClient, 'invalidateQueries') + queryClient.setQueryData(['opencode', 'session', 'session-1', '/repo'], { id: 'session-1' }) + + const { result, unmount } = renderHook( + () => useSSE('/repo', 'session-1'), + { wrapper: createWrapper(queryClient) } + ) + + await waitFor(() => expect(MockEventSource.instances).toHaveLength(1)) + await connect(0, 'client-1') + await waitFor(() => expect(result.current.isConnected).toBe(true)) + invalidateQueries.mockClear() + + act(() => { + MockEventSource.instances[0].emit('message', { + type: 'session.moved', + directory: '/repo', + data: { sessionID: 'session-1', projectID: 'proj-1', subpath: '', location: { directory: '/elsewhere' } }, + }) + }) + + await waitFor(() => { + expect(invalidateQueries).toHaveBeenCalledWith({ queryKey: ['opencode', 'session', 'session-1'] }) + }) + + unmount() + }) + it('invalidates the cached session query on session.renamed', async () => { const queryClient = new QueryClient({ defaultOptions: { diff --git a/frontend/src/hooks/useSSE.ts b/frontend/src/hooks/useSSE.ts index 5d9f18d88..2448b4635 100644 --- a/frontend/src/hooks/useSSE.ts +++ b/frontend/src/hooks/useSSE.ts @@ -80,6 +80,7 @@ export const useSSE = (directory?: string | string[], currentSessionId?: string) break case 'session.renamed': + case 'session.moved': invalidateSessionQueryIfCached(queryClient, event.data.sessionID) break From 1b2665bd2b501eebec77a2f4d5db2b9375e97d85 Mon Sep 17 00:00:00 2001 From: Chris Scott <99081550+chriswritescode-dev@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:18:06 -0400 Subject: [PATCH 3/3] fix(background-activity): address review on resync scope and stale subscribe responses A directory subscribed by the stream but neither registered as a repo nor remembered as a session directory was reconciled on subscribe but skipped on resync, so a missed shell exit could leave the indicator stale. Pass the stream's subscribed directories to onResync and reconcile their pending actions and running shells alongside the tracked ones. Subscribe responses that arrive after the connection was replaced or the directory was unsubscribed could also apply snapshots to state the client no longer owns. Ignore responses from a superseded connection and report only directories still subscribed. The stale-indicator e2e test now waits for the successful unsubscribe response and asserts the subscribe responses succeeded instead of sleeping. --- e2e/background-activity.spec.ts | 13 ++-- frontend/src/contexts/EventContext.test.tsx | 54 +++++++++++---- frontend/src/contexts/EventContext.tsx | 7 +- .../__tests__/openCodeEventStream.test.ts | 65 +++++++++++++++++++ .../openCodeEventStream.ts | 18 +++-- 5 files changed, 131 insertions(+), 26 deletions(-) diff --git a/e2e/background-activity.spec.ts b/e2e/background-activity.spec.ts index aa5814c98..8bdeaa9f6 100644 --- a/e2e/background-activity.spec.ts +++ b/e2e/background-activity.spec.ts @@ -76,9 +76,9 @@ function createUnregisteredDirectory(name: string): string { return directory } -function waitForDirectorySubscription(page: Page, directory: string) { +function waitForDirectorySubscription(page: Page, directory: string, endpoint = '/api/sse/subscribe') { return page.waitForResponse((response) => - response.url().includes('/api/sse/subscribe') && (response.request().postData() ?? '').includes(directory), + response.url().includes(endpoint) && (response.request().postData() ?? '').includes(directory), ) } @@ -206,15 +206,16 @@ test.describe('session background activity', () => { await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) await expect(page.getByPlaceholder('Send a message...')).toBeVisible() - await subscribed + expect((await subscribed).ok()).toBe(true) const shellId = await startBackgroundShell(request, directory, sessionId) createdShells.push({ directory, id: shellId }) await expect(backgroundIndicator(page)).toBeVisible() + const unsubscribed = waitForDirectorySubscription(page, directory, '/api/sse/unsubscribe') await navigateWithinApp(page, '/') - await page.waitForTimeout(3_000) + expect((await unsubscribed).ok()).toBe(true) await removeShell(request, directory, shellId) await page.waitForTimeout(3_000) await navigateWithinApp(page, `/repos/${repo.id}/sessions/${sessionId}`) @@ -231,14 +232,14 @@ test.describe('session background activity', () => { await page.goto(`/repos/${repo.id}/sessions/${sessionId}`) await expect(page.getByPlaceholder('Send a message...')).toBeVisible() - await subscribedToSource + expect((await subscribedToSource).ok()).toBe(true) const subscribedToTarget = waitForDirectorySubscription(page, target) await requireOk( await request.post(`/api/opencode/api/session/${sessionId}/move`, { data: { directory: target } }), 'move session', ) - await subscribedToTarget + expect((await subscribedToTarget).ok()).toBe(true) const shellId = await startBackgroundShell(request, target, sessionId) createdShells.push({ directory: target, id: shellId }) diff --git a/frontend/src/contexts/EventContext.test.tsx b/frontend/src/contexts/EventContext.test.tsx index d3d2ee7a8..1083097b0 100644 --- a/frontend/src/contexts/EventContext.test.tsx +++ b/frontend/src/contexts/EventContext.test.tsx @@ -457,7 +457,7 @@ describe('EventProvider permissions and forms', () => { const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { onEvent({ type: 'permission.asked', data: pendingPermission, directory: '/repo' }) @@ -470,7 +470,7 @@ describe('EventProvider permissions and forms', () => { }) act(() => { - onResync?.() + onResync?.([]) }) await waitFor(() => { @@ -492,7 +492,7 @@ describe('EventProvider permissions and forms', () => { const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { onEvent({ type: 'permission.asked', data: pendingPermission, directory: '/repo' }) @@ -505,7 +505,7 @@ describe('EventProvider permissions and forms', () => { }) act(() => { - onResync?.() + onResync?.([]) }) await waitFor(() => { @@ -531,7 +531,7 @@ describe('EventProvider permissions and forms', () => { const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { onEvent({ type: 'permission.asked', data: pendingPermission, directory: '/worktrees/wt-1' }) @@ -542,7 +542,7 @@ describe('EventProvider permissions and forms', () => { }) act(() => { - onResync?.() + onResync?.([]) }) await waitFor(() => { @@ -852,10 +852,10 @@ describe('EventProvider permissions and forms', () => { invalidateQueries.mockClear() const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { - onResync?.() + onResync?.([]) }) await waitFor(() => { @@ -939,10 +939,10 @@ describe('EventProvider permissions and forms', () => { mocks.listPendingForms.mockClear() const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { - onResync?.() + onResync?.([]) }) expect(mocks.listPendingPermissions).toHaveBeenCalledWith('/repo-a') @@ -1210,7 +1210,7 @@ describe('EventProvider permissions and forms', () => { const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] const onEvent = lastSubscribeCall[0].onEvent as (data: unknown) => void - const onResync = lastSubscribeCall[0].onResync as (() => void) | undefined + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined act(() => { onEvent({ type: 'session.execution.started', data: { sessionID: 'child-1' }, directory: '/repo' }) @@ -1218,7 +1218,7 @@ describe('EventProvider permissions and forms', () => { expect(useSessionStatus.getState().getStatus('child-1')).toEqual({ type: 'busy' }) act(() => { - onResync?.() + onResync?.([]) }) await waitFor(() => { @@ -2047,4 +2047,34 @@ describe('EventProvider permissions and forms', () => { expect(screen.getByTestId('permission-count').textContent).toBe('1') expect(screen.getByTestId('count').textContent).toBe('1') }) + + it('reconciles subscribed directories reported by a resync even when they are not tracked', async () => { + mocks.listPendingPermissions.mockResolvedValue([pendingPermission]) + mocks.listPendingForms.mockResolvedValue([pendingForm]) + + render(, { wrapper: createWrapper() }) + + await waitFor(() => expect(mocks.subscribeGlobalMonitor).toHaveBeenCalled()) + + const lastSubscribeCall = mocks.subscribeGlobalMonitor.mock.calls[mocks.subscribeGlobalMonitor.mock.calls.length - 1] + const onResync = lastSubscribeCall[0].onResync as ((directories: string[]) => void) | undefined + + act(() => { + useSessionStatus.getState().setShellRunning('shell-stale', 'session-1', '/unregistered-dir') + }) + expect(useSessionStatus.getState().runningShells.has('shell-stale')).toBe(true) + + await act(async () => { + onResync?.(['/unregistered-dir']) + }) + + await waitFor(() => { + expect(mocks.listShells).toHaveBeenCalledWith('/unregistered-dir') + expect(mocks.listPendingPermissions).toHaveBeenCalledWith('/unregistered-dir') + expect(mocks.listPendingForms).toHaveBeenCalledWith('/unregistered-dir') + }) + expect(useSessionStatus.getState().runningShells.has('shell-stale')).toBe(false) + expect(screen.getByTestId('permission-count').textContent).toBe('1') + expect(screen.getByTestId('count').textContent).toBe('1') + }) }) diff --git a/frontend/src/contexts/EventContext.tsx b/frontend/src/contexts/EventContext.tsx index e8db0e53d..af41c143d 100644 --- a/frontend/src/contexts/EventContext.tsx +++ b/frontend/src/contexts/EventContext.tsx @@ -709,10 +709,11 @@ export function EventProvider({ children }: { children: React.ReactNode }) { } } - const handleResync = () => { - void reconcilePendingActionsForDirectories(collectTrackedDirectories()) + const handleResync = (subscribedDirectories: string[]) => { + const directories = [...new Set([...collectTrackedDirectories(), ...subscribedDirectories])] + void reconcilePendingActionsForDirectories(directories) void fetchInitialSessionStatuses() - void syncRunningShells(collectTrackedDirectories()) + void syncRunningShells(directories) invalidateChildSessionCaches(queryClient) invalidateQueryKeysDebounced(queryClient, LOCATION_CATALOG_QUERY_KEYS) } diff --git a/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts b/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts index 20255f7b0..1e4fbb3a3 100644 --- a/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts +++ b/frontend/src/lib/opencode-event-stream/__tests__/openCodeEventStream.test.ts @@ -9,6 +9,23 @@ class RejectingSubscribeTransport extends TestEventStreamTransport { } } +class DeferredSubscribeTransport extends TestEventStreamTransport { + private resolvers: Array<(value: boolean) => void> = [] + + override post(path: string, body: unknown): Promise { + this.posts.push({ path, body }) + return new Promise((resolve) => { + this.resolvers.push(resolve) + }) + } + + resolvePosts(success = true): void { + const resolvers = this.resolvers + this.resolvers = [] + resolvers.forEach((resolve) => resolve(success)) + } +} + describe('OpenCodeEventStream', () => { beforeEach(() => { vi.useFakeTimers() @@ -176,6 +193,7 @@ describe('OpenCodeEventStream', () => { transport.resync() expect(onResync).toHaveBeenCalledTimes(1) + expect(onResync).toHaveBeenCalledWith(['/repo']) }) it('reports visibility through the transport adapter', async () => { @@ -242,4 +260,51 @@ describe('OpenCodeEventStream', () => { expect(onDirectoriesSubscribed).not.toHaveBeenCalled() }) + + it('ignores a subscribe response that resolves after the stream reconnected', async () => { + const transport = new DeferredSubscribeTransport() + const stream = new OpenCodeEventStream({ transport }) + const onDirectoriesSubscribed = vi.fn() + + stream.subscribeGlobalMonitor({ directories: ['/repo'], onEvent: vi.fn(), onDirectoriesSubscribed }) + transport.openConnection() + transport.connected('client-1') + await vi.advanceTimersByTimeAsync(0) + + expect(transport.posts).toContainEqual({ + path: '/api/sse/subscribe', + body: { clientId: 'client-1', directories: ['/repo'] }, + }) + + transport.fail() + transport.openConnection() + transport.connected('client-2') + + transport.resolvePosts() + await vi.advanceTimersByTimeAsync(0) + + expect(onDirectoriesSubscribed).not.toHaveBeenCalled() + }) + + it('does not notify a directory removed while its subscribe request is in flight', async () => { + const transport = new DeferredSubscribeTransport() + const stream = new OpenCodeEventStream({ transport }) + const onDirectoriesSubscribed = vi.fn() + + const subscription = stream.subscribeGlobalMonitor({ + directories: ['/repo'], + onEvent: vi.fn(), + onDirectoriesSubscribed, + }) + transport.openConnection() + transport.connected('client-1') + await vi.advanceTimersByTimeAsync(0) + + subscription.updateDirectories([]) + + transport.resolvePosts() + await vi.advanceTimersByTimeAsync(0) + + expect(onDirectoriesSubscribed).not.toHaveBeenCalled() + }) }) diff --git a/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts b/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts index 8846f1df8..32598c262 100644 --- a/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts +++ b/frontend/src/lib/opencode-event-stream/openCodeEventStream.ts @@ -14,7 +14,7 @@ interface Subscriber { onEvent: OpenCodeEventHandler onStatusChange?: EventStreamStatusHandler onHealthChange?: (state: EventStreamHealthState) => void - onResync?: () => void + onResync?: (subscribedDirectories: string[]) => void onDirectoriesSubscribed?: (directories: string[]) => void directories: Set } @@ -24,7 +24,7 @@ interface SubscriberInput { onStatusChange?: EventStreamStatusHandler onHealthChange?: (state: EventStreamHealthState) => void directories: string[] - onResync?: () => void + onResync?: (subscribedDirectories: string[]) => void onDirectoriesSubscribed?: (directories: string[]) => void } @@ -400,9 +400,10 @@ export class OpenCodeEventStream { private handleResync(): void { this.markActivity() + const directories = Array.from(this.directoryRefCounts.keys()) this.subscribers.forEach((subscriber) => { try { - subscriber.onResync?.() + subscriber.onResync?.(directories) } catch { void 0 } @@ -422,18 +423,25 @@ export class OpenCodeEventStream { private async subscribeToRemoteDirectories(directories: string[]): Promise { if (!this.clientId || directories.length === 0) return + const clientId = this.clientId + const success = await this.transport.post('/api/sse/subscribe', { - clientId: this.clientId, + clientId, directories, }) + if (this.clientId !== clientId) return + if (!success) { directories.forEach((directory) => this.pendingDirectories.add(directory)) this.reconnect() return } - this.notifyDirectoriesSubscribed(directories) + const subscribedDirectories = directories.filter((directory) => this.directoryRefCounts.has(directory)) + if (subscribedDirectories.length === 0) return + + this.notifyDirectoriesSubscribed(subscribedDirectories) } private notifyDirectoriesSubscribed(directories: string[]): void {