diff --git a/packages/client/workbench/AGENTS.md b/packages/client/workbench/AGENTS.md index 6077282d6..30fe36273 100644 --- a/packages/client/workbench/AGENTS.md +++ b/packages/client/workbench/AGENTS.md @@ -28,8 +28,23 @@ app-specific entries (`apps/desktop`, `apps/webview`) and pure presentation (`pa the workbench **binding** — it pins the generic to `LinkCodeSdkClient`, promotes each generation into the ambient default tayori reads (`setDefaultClient`), and reports outcomes to product analytics. Behavior changes belong in client-core; only SDK/analytics wiring belongs here. SWR retains cached data across generations of the same - endpoint, starts a fresh cache after endpoint migration, and revalidates once after a generation - becomes protocol-ready; it does not own connection state. + endpoint, starts a fresh cache after endpoint migration, revalidates once after a generation + becomes protocol-ready, and revalidates the session and workspace list caches on every + `session.changed` push, coalesced through `coalesceRuns` (the daemon registers/freshens a + session's workspace *before* announcing the record on start and resume, so one frame covers both + lists there; an import of a brand-new cwd announces before the touch, and another client's + explicit `workspace.register` / rename / archive has no push at all, so both wait for the next + revalidation). Coalescing is not optional: one start emits several frames, a bulk import emits one + per entry, and SWR's key-filter `mutate` deletes its own dedupe markers, so an uncoalesced + subscription turns a burst into one forced round trip per frame per list. The effect's abort + signal prevents queued runs after generation teardown; a fresh controller snapshot gates + revalidation while a disposed generation remains mounted during recovery. This coalescer stays workbench-local: + SWR owns fetch errors here; client-core's direct refresh loop has different failure semantics. + It does not own connection state. +- `mock/` — the dev mock announces imports before touching the workspace, matching the engine's order, + but its synchronous touch cannot reproduce the engine's async import race. Mock tests prove + start/resume-driven revalidation only. The mock has no `session.delete` handler and therefore no + `session.changed` `removed` emission; deletion-driven revalidation needs separate coverage. - `surface/` — the workbench feature surface: the `Workbench` component, the `WorkbenchShell*` contract plus the default shell, and session orchestration hooks. - `terminal/` — the daemon-backed interactive terminal: the panel container, the key-scoped diff --git a/packages/client/workbench/src/mock/dev-mock-host.ts b/packages/client/workbench/src/mock/dev-mock-host.ts index dd9155e0c..7cf4b5861 100644 --- a/packages/client/workbench/src/mock/dev-mock-host.ts +++ b/packages/client/workbench/src/mock/dev-mock-host.ts @@ -908,9 +908,11 @@ export class DevMockHost { model, effort, }); - // Parity with the engine: starting a session registers/freshens its directory's workspace. + // Parity with the engine: starting a session registers/freshens its directory's workspace, + // then announces the record before answering the request. this.touchWorkspace(cwd, now); const { sessionId } = session; + this.send({ kind: 'session.changed', sessionId, reason: 'created' }); this.emit(sessionId, { type: 'status', status: 'starting' }); this.emit(sessionId, { type: 'current-mode-update', currentModeId: 'mock' }); this.emitDirectiveAdvertisement(sessionId); @@ -954,6 +956,10 @@ export class DevMockHost { updatedAt: now, origin, }); + // Engine order, deliberately: importRecord announces the record and only then touches the + // workspace, unlike start/resume which register it first. + this.send({ kind: 'session.changed', sessionId: session.sessionId, reason: 'created' }); + this.touchWorkspace(session.cwd, now); this.send({ kind: 'session.imported', replyTo, @@ -1084,6 +1090,9 @@ export class DevMockHost { return; } session.status = 'idle'; + this.touchWorkspace(session.cwd, Date.now()); + // Parity with the engine: a relaunch appends a run, which re-points the listed identity. + this.send({ kind: 'session.changed', sessionId, reason: 'updated' }); this.attachSession(sessionId); this.send({ kind: 'session.started', replyTo, sessionId }); } @@ -1239,7 +1248,10 @@ export class DevMockHost { content: ContentBlock[], ): Promise { const text = promptText(content); - if (text && !session.title) session.title = text.slice(0, 80); + if (text && !session.title) { + session.title = text.slice(0, 80); + this.send({ kind: 'session.changed', sessionId: session.sessionId, reason: 'updated' }); + } session.status = 'running'; this.emit(session.sessionId, { type: 'user-message', diff --git a/packages/client/workbench/src/runtime/__tests__/coalesce.test.ts b/packages/client/workbench/src/runtime/__tests__/coalesce.test.ts new file mode 100644 index 000000000..5df2fc451 --- /dev/null +++ b/packages/client/workbench/src/runtime/__tests__/coalesce.test.ts @@ -0,0 +1,79 @@ +import { wait } from 'foxts/wait'; +import { expect, it, vi } from 'vitest'; +import { coalesceRuns } from '../coalesce'; +import { deferred } from './connection-controller-test-helpers'; + +it('collapses a burst arriving mid-run into a single trailing run', async () => { + const gates = [deferred(), deferred()]; + let started = 0; + const trigger = coalesceRuns(() => { + const gate = gates[started] ?? deferred(); + started += 1; + return gate.promise; + }, new AbortController().signal); + + trigger(); + expect(started).toBe(1); + + // Three more frames while the first run is still in flight: they must collapse into one. + trigger(); + trigger(); + trigger(); + expect(started).toBe(1); + + gates[0].resolve(); + await vi.waitFor(() => expect(started).toBe(2)); + + gates[1].resolve(); + await wait(0); + expect(started).toBe(2); +}); + +it('runs again for a trigger that arrives after the previous run settled', async () => { + let started = 0; + const trigger = coalesceRuns(() => { + started += 1; + return Promise.resolve(); + }, new AbortController().signal); + + trigger(); + await wait(0); + trigger(); + await wait(0); + + expect(started).toBe(2); +}); + +it('keeps draining after a failed run', async () => { + let started = 0; + const trigger = coalesceRuns(() => { + started += 1; + return started === 1 ? Promise.reject(new Error('fetch failed')) : Promise.resolve(); + }, new AbortController().signal); + + trigger(); + trigger(); + await vi.waitFor(() => expect(started).toBe(2)); +}); + +it.each(['resolve', 'reject'] as const)( + 'drops queued work after abort when the in-flight run settles via %s', + async (outcome) => { + const controller = new AbortController(); + const gate = deferred(); + const run = vi.fn(() => gate.promise); + const trigger = coalesceRuns(run, controller.signal); + + trigger(); + trigger(); + expect(run).toHaveBeenCalledTimes(1); + controller.abort(); + if (outcome === 'resolve') gate.resolve(); + else gate.reject(new Error('client disposed')); + await wait(0); + expect(run).toHaveBeenCalledTimes(1); + + trigger(); + expect(run).toHaveBeenCalledTimes(1); + }, +); diff --git a/packages/client/workbench/src/runtime/coalesce.ts b/packages/client/workbench/src/runtime/coalesce.ts new file mode 100644 index 000000000..16d3b7a5a --- /dev/null +++ b/packages/client/workbench/src/runtime/coalesce.ts @@ -0,0 +1,39 @@ +import { noop } from 'foxts/noop'; + +/** + * Mid-run triggers need one trailing run; abort discards it without cancelling active work. + * The caller owns error reporting; a failed run still drains queued changes unless aborted. + */ +export function coalesceRuns(run: () => Promise, signal: AbortSignal): () => void { + let running = false; + let queued = false; + + // Read through a call, not `while (queued)`: the flag is only ever set from the closure below + // while a run is awaited, which narrowing cannot see. + const takeQueued = (): boolean => { + const wasQueued = queued; + queued = false; + return wasQueued; + }; + + const drain = async (): Promise => { + running = true; + try { + do { + // eslint-disable-next-line no-await-in-loop -- serializing is the point: one run at a time + await run().catch(noop); + } while (!signal.aborted && takeQueued()); + } finally { + running = false; + } + }; + + return () => { + if (signal.aborted) return; + if (running) { + queued = true; + return; + } + void drain().catch(noop); + }; +} diff --git a/packages/client/workbench/src/runtime/provider.tsx b/packages/client/workbench/src/runtime/provider.tsx index d539ba097..cc562ace2 100644 --- a/packages/client/workbench/src/runtime/provider.tsx +++ b/packages/client/workbench/src/runtime/provider.tsx @@ -1,5 +1,6 @@ import { LinkCodeProvider } from '@linkcode/client-core'; import type { LinkCodeSdkClient } from '@linkcode/sdk'; +import { listSessions, listWorkspaces } from '@linkcode/sdk'; import { ComposeContextProvider } from 'foxact/compose-context-provider'; import { nullthrow } from 'foxact/nullthrow'; import { useEffect } from 'foxact/use-abortable-effect'; @@ -10,6 +11,7 @@ import { wait } from 'foxts/wait'; import { createContext, useContext, useRef, useSyncExternalStore } from 'react'; import type { Cache, Middleware as SWRMiddleware } from 'swr'; import { SWRConfig, useSWRConfig } from 'swr'; +import { coalesceRuns } from './coalesce'; import type { WorkbenchConnectionGeneration, WorkbenchConnectionSource, @@ -168,14 +170,29 @@ function WorkbenchRuntimeGeneration({ , ]} > - + {children} - + ); } -function ReadyRevalidator({ +/** A `listSessions` / `listWorkspaces` cache entry, whichever surface owns it. Both tayori key + * forms land in the cache as the resolved `[sdkMethod, arg, cacheTags]` tuple, so this matches the + * tuple rather than tayori's brand — the lazy form brands its outer function, not the array. */ +function isHostListKey(key: unknown): boolean { + return Array.isArray(key) && (key[0] === listSessions || key[0] === listWorkspaces); +} + +/** + * Keeps SWR in step with the host: everything once a generation is protocol-ready, and the two + * list caches on each `session.changed` push. The daemon registers/freshens a session's workspace + * before it announces the record on start and resume, so one frame stands for both lists there; + * import announces first and touches after, so a brand-new imported cwd can need the next + * revalidation. Pushes are coalesced: a single start emits several frames, and a bulk import emits + * one per entry, while SWR's key-filter `mutate` deletes its own dedupe markers. + */ +function HostRevalidator({ children, controller, generation, @@ -197,6 +214,21 @@ function ReadyRevalidator({ void mutate(trueFn); }, [generation.id, mutate, status]); + const client = generation.client.raw; + useEffect( + (signal) => + client.subscribeSessionChanged( + coalesceRuns(async () => { + // Recovery retains disposed generations in React until a replacement is ready. + const snapshot = controller.getSnapshot(); + if (snapshot.status === 'ready' && snapshot.contextGeneration?.id === generation.id) { + await mutate(isHostListKey); + } + }, signal), + ), + [client, controller, generation.id, mutate], + ); + return children; } diff --git a/packages/client/workbench/src/workspace/hooks.ts b/packages/client/workbench/src/workspace/hooks.ts index 8897fcb7d..ec6ab21fb 100644 --- a/packages/client/workbench/src/workspace/hooks.ts +++ b/packages/client/workbench/src/workspace/hooks.ts @@ -2,9 +2,12 @@ import { listWorkspaces } from '@linkcode/sdk'; import { useData } from '../runtime/tayori'; /** - * Every registered workspace (directory), most recently used first. No push invalidation yet: - * after a workspace mutation the caller must call this hook's `mutate()` — the same convention - * `useWorkbenchSessions` follows for session mutations. + * Every registered workspace (directory), most recently used first. The runtime revalidates it on + * every `session.changed` push, which covers a session another client starts or resumes: the daemon + * registers that workspace before announcing the record. It does not cover an import of a + * brand-new cwd (announced before the touch) or another client's explicit register/rename/archive, + * which have no push at all. A workspace mutation this client issues itself still calls `mutate()`, + * the same convention `useWorkbenchSessions` follows for session mutations. */ export function useWorkspaces() { return useData(listWorkspaces, {}); diff --git a/packages/client/workbench/tests/integration/session-changed-revalidation.test.tsx b/packages/client/workbench/tests/integration/session-changed-revalidation.test.tsx new file mode 100644 index 000000000..623d9a2c8 --- /dev/null +++ b/packages/client/workbench/tests/integration/session-changed-revalidation.test.tsx @@ -0,0 +1,166 @@ +// @vitest-environment jsdom +import { LinkCodeClient, useLinkCodeClient } from '@linkcode/client-core'; +import { listSessions, listWorkspaces } from '@linkcode/sdk'; +import { cleanup, renderHook, waitFor } from '@testing-library/react'; +import { createFixedArray } from 'foxts/create-fixed-array'; +import { nullthrow } from 'foxts/guard'; +import { asyncNoop } from 'foxts/noop'; +import { wait } from 'foxts/wait'; +import { afterEach, expect, it, vi } from 'vitest'; +import { createDevMockTransport } from '../../src/mock/dev-mock-transport'; +import { + deferred, + TestTransport, +} from '../../src/runtime/__tests__/connection-controller-test-helpers'; +import { DebugProvider } from '../../src/runtime/debug'; +import { WorkbenchRuntimeProvider } from '../../src/runtime/provider'; +import { useData } from '../../src/runtime/tayori'; +import { useWorkspaces } from '../../src/workspace/hooks'; + +const connectionSource = { + resolve: () => ({ endpoint: 'mock://session-changed', transport: createDevMockTransport() }), +}; + +function Runtime({ children }: React.PropsWithChildren): React.ReactNode { + return ( + + + {children} + + + ); +} + +/** The sidebar's two inputs, read the way the workbench reads them — through the shared hooks, + * which never call `mutate()` themselves. */ +function useSidebarInputs() { + const { data: workspaces } = useWorkspaces(); + const { data: sessions } = useData(listSessions, {}); + return { client: useLinkCodeClient(), workspaces, sessions }; +} + +function useLazySidebarInputs() { + const { data: workspaces } = useData(listWorkspaces, () => ({})); + const { data: sessions } = useData(listSessions, () => ({})); + return { client: useLinkCodeClient(), workspaces, sessions }; +} + +afterEach(() => { + cleanup(); + vi.restoreAllMocks(); +}); + +// The mock host answers every control request after a scripted latency; each step here is one or +// more of those round trips. +const STEP_TIMEOUT = { timeout: 4000 }; + +it.each([ + { keyForm: 'object', useInputs: useSidebarInputs }, + { keyForm: 'lazy', useInputs: useLazySidebarInputs }, +])( + 'refreshes $keyForm keys when another client starts a session', + async ({ useInputs }) => { + const { result } = renderHook(useInputs, { wrapper: Runtime }); + await waitFor(() => expect(result.current.workspaces).toBeDefined(), STEP_TIMEOUT); + const cwd = '/mock/elsewhere/new-repo'; + expect(result.current.workspaces?.map((workspace) => workspace.cwd)).not.toContain(cwd); + + // Bypassing the workbench's own create path stands in for another client: this client only + // learns about the session from the host's pushed frames. + const sessionId = await result.current.client.startSession({ kind: 'claude-code', cwd }); + + await waitFor(() => { + expect(result.current.workspaces?.map((workspace) => workspace.cwd)).toContain(cwd); + expect(result.current.sessions?.map((session) => session.sessionId)).toContain(sessionId); + }, STEP_TIMEOUT); + }, + 15000, +); + +it('collapses a burst of pushes instead of one round trip per frame', async () => { + const listSpy = vi.spyOn(LinkCodeClient.prototype, 'listSessions'); + const workspaceSpy = vi.spyOn(LinkCodeClient.prototype, 'listWorkspaces'); + const { result } = renderHook(useSidebarInputs, { wrapper: Runtime }); + await waitFor(() => expect(result.current.workspaces).toBeDefined(), STEP_TIMEOUT); + + const starts = 6; + listSpy.mockClear(); + workspaceSpy.mockClear(); + const ids = await Promise.all( + createFixedArray(starts).map((index) => + result.current.client.startSession({ + kind: 'claude-code', + cwd: `/mock/elsewhere/burst-${index}`, + }), + ), + ); + + await waitFor(() => { + const listed = result.current.sessions?.map((session) => session.sessionId) ?? []; + const workspaces = result.current.workspaces?.map((workspace) => workspace.cwd) ?? []; + for (let i = 0, len = ids.length; i < len; i++) { + expect(listed).toContain(ids[i]); + expect(workspaces).toContain(`/mock/elsewhere/burst-${i}`); + } + }, STEP_TIMEOUT); + + // All starts complete within the mock's list latency: one in-flight fetch plus one trailing. + expect(listSpy.mock.calls.length).toBeLessThanOrEqual(2); + expect(workspaceSpy.mock.calls.length).toBeLessThanOrEqual(2); +}, 15000); + +it('restores an archived workspace before announcing a resumed session', async () => { + const { result } = renderHook(useSidebarInputs, { wrapper: Runtime }); + await waitFor(() => { + expect(result.current.sessions).toBeDefined(); + expect(result.current.workspaces).toBeDefined(); + }, STEP_TIMEOUT); + const { client } = result.current; + const session = nullthrow(result.current.sessions?.find((item) => item.status === 'stopped')); + const workspace = nullthrow(result.current.workspaces?.find((item) => item.cwd === session.cwd)); + await client.archiveWorkspace(workspace.workspaceId); + expect((await client.listWorkspaces()).map((item) => item.cwd)).not.toContain(session.cwd); + + await client.resumeSession(session.sessionId); + + await waitFor(() => { + const restored = result.current.workspaces?.find((item) => item.cwd === session.cwd); + expect(restored).toBeDefined(); + expect(restored?.lastUsedAt).toBeGreaterThan(workspace.lastUsedAt); + }, STEP_TIMEOUT); +}, 15000); + +it('does not drain queued refreshes through a disposed generation during recovery', async () => { + const transport = createDevMockTransport(); + vi.spyOn(connectionSource, 'resolve') + .mockReturnValueOnce({ endpoint: 'mock://session-changed', transport }) + .mockReturnValue({ + endpoint: 'mock://session-changed', + transport: new TestTransport(asyncNoop), + }); + const { result } = renderHook(useSidebarInputs, { wrapper: Runtime }); + await waitFor(() => expect(result.current.workspaces).toBeDefined(), STEP_TIMEOUT); + const { client } = result.current; + const pending = deferred(); + const pendingSessions = deferred(); + const workspaceSpy = vi + .spyOn(client, 'listWorkspaces') + .mockImplementationOnce(() => pending.promise.then(() => [])); + const sessionSpy = vi + .spyOn(client, 'listSessions') + .mockImplementationOnce(() => pendingSessions.promise.then(() => [])); + + await Promise.all([ + client.startSession({ kind: 'claude-code', cwd: '/mock/recovery-one' }), + client.startSession({ kind: 'claude-code', cwd: '/mock/recovery-two' }), + ]); + expect(workspaceSpy).toHaveBeenCalledTimes(1); + expect(sessionSpy).toHaveBeenCalledTimes(1); + transport.close(); + pending.resolve(); + pendingSessions.resolve(); + await wait(0); + + expect(workspaceSpy).toHaveBeenCalledTimes(1); + expect(sessionSpy).toHaveBeenCalledTimes(1); +}, 15000);