diff --git a/mobile/src/tasks/worktree-create-capability-reconnect.test.tsx b/mobile/src/tasks/worktree-create-capability-reconnect.test.tsx new file mode 100644 index 00000000000..2792db5910c --- /dev/null +++ b/mobile/src/tasks/worktree-create-capability-reconnect.test.tsx @@ -0,0 +1,176 @@ +import { createElement } from 'react' +import { act, create, type ReactTestRenderer } from 'react-test-renderer' +import { describe, expect, it, vi } from 'vitest' +import type { RpcClient } from '../transport/rpc-client' +import type { ConnectionState, RpcResponse } from '../transport/types' +import { useNewWorktreeRuntimeCapabilities } from './worktree-create-capability' +import { createWorktreeWithNameRetry } from './worktree-create-retry' + +type CapabilityHook = ReturnType + +class CapabilityClient implements RpcClient { + private readonly listeners = new Set<(state: ConnectionState) => void>() + private state: ConnectionState = 'connected' + readonly statusRequests = vi.fn<() => Promise>() + readonly createRequests: Record[] = [] + + constructor(statusOutcomes: Array) { + let statusCall = 0 + this.statusRequests.mockImplementation(async () => { + const outcome = statusOutcomes[Math.min(statusCall, statusOutcomes.length - 1)]! + statusCall += 1 + if (outcome instanceof Error) { + throw outcome + } + return outcome + }) + } + + async sendRequest(method: string, params?: unknown): Promise { + if (method === 'status.get') { + return this.statusRequests() + } + this.createRequests.push((params ?? {}) as Record) + return success({ worktree: { id: 'wt-1' } }) + } + + subscribe(): () => void { + return () => {} + } + + updateTerminalSubscriptionViewport(): void {} + getState(): ConnectionState { + return this.state + } + getReconnectAttempt(): number { + return 0 + } + getLastConnectedAt(): number | null { + return null + } + onStateChange(listener: (state: ConnectionState) => void): () => void { + this.listeners.add(listener) + return () => this.listeners.delete(listener) + } + notifyForeground(): void {} + close(): void {} + + emitState(state: ConnectionState): void { + this.state = state + for (const listener of this.listeners) { + listener(state) + } + } +} + +function success(result: unknown): RpcResponse { + return { id: '1', ok: true, result, _meta: { runtimeId: 'r' } } +} + +function unsupported(): RpcResponse { + return { + id: '1', + ok: false, + error: { code: 'unsupported', message: 'unsupported' }, + _meta: { runtimeId: 'r' } + } +} + +function supported(): RpcResponse { + return success({ + capabilities: ['mobile.tasks.v1', 'worktree.create-idempotency.v1'], + hostPlatform: 'darwin' + }) +} + +async function mountCapabilities(client: RpcClient): Promise<{ + readonly current: CapabilityHook + unmount(): void +}> { + let current: CapabilityHook | null = null + let renderer: ReactTestRenderer | null = null + function Probe(): null { + current = useNewWorktreeRuntimeCapabilities(client, true) + return null + } + await act(async () => { + renderer = create(createElement(Probe)) + await Promise.resolve() + }) + return { + get current(): CapabilityHook { + if (!current) { + throw new Error('capability probe did not render') + } + return current + }, + unmount: () => act(() => renderer?.unmount()) + } +} + +describe('useNewWorktreeRuntimeCapabilities reconnect recovery', () => { + it('recovers a transient probe and enables idempotent create after reconnect', async () => { + const client = new CapabilityClient([new Error('offline'), supported()]) + const capabilities = await mountCapabilities(client) + + await act(async () => { + client.emitState('reconnecting') + client.emitState('connected') + await Promise.resolve() + }) + await createWorktreeWithNameRetry({ + client, + baseName: 'otter', + buildParams: (name) => ({ repo: 'id:r', name }), + supportsIdempotentCutoverRetry: capabilities.current.getWorktreeCreateCutoverSupport(), + mintMutationId: () => 'mutation-after-reconnect' + }) + + expect({ + statusRequests: client.statusRequests.mock.calls.length, + tasksSupported: capabilities.current.tasksSupported, + clientMutationId: client.createRequests[0]?.clientMutationId + }).toEqual({ + statusRequests: 2, + tasksSupported: true, + clientMutationId: 'mutation-after-reconnect' + }) + capabilities.unmount() + }) + + it('keeps an authoritative unsupported response cached across reconnect', async () => { + const client = new CapabilityClient([unsupported(), supported()]) + const capabilities = await mountCapabilities(client) + + await act(async () => { + client.emitState('reconnecting') + client.emitState('connected') + await Promise.resolve() + }) + + expect(client.statusRequests).toHaveBeenCalledOnce() + expect(capabilities.current.tasksSupported).toBe(false) + capabilities.unmount() + }) + + it('starts at most one replacement probe for a flapping reconnect', async () => { + const client = new CapabilityClient([new Error('offline'), supported()]) + const capabilities = await mountCapabilities(client) + + await act(async () => { + client.emitState('reconnecting') + client.emitState('disconnected') + client.emitState('reconnecting') + client.emitState('connected') + client.emitState('connected') + await Promise.all([ + capabilities.current.getWorktreeCreateCutoverSupport(), + capabilities.current.getWorktreeCreateCutoverSupport() + ]) + }) + + expect(client.statusRequests).toHaveBeenCalledTimes(2) + expect(capabilities.current.tasksSupported).toBe(true) + capabilities.unmount() + }) +}) diff --git a/mobile/src/tasks/worktree-create-capability.ts b/mobile/src/tasks/worktree-create-capability.ts index c4ca8170312..c78befb8803 100644 --- a/mobile/src/tasks/worktree-create-capability.ts +++ b/mobile/src/tasks/worktree-create-capability.ts @@ -29,16 +29,42 @@ const UNSUPPORTED_CAPABILITIES: NewWorktreeRuntimeCapabilities = { hostPlatform: null } +type CapabilityProbeResult = { + capabilities: NewWorktreeRuntimeCapabilities + cache: 'authoritative' | 'transient' +} + +type CapabilityProbeRecord = { + client: RpcClient | null + connectionRevision: number + cache: 'pending' | CapabilityProbeResult['cache'] + promise: Promise +} + +type ConnectionRevision = { + client: RpcClient | null + revision: number + state: ReturnType | null + generation: number | null + connectedAt: number | null +} + // Why: status.get is safe to replay and must settle before create, independently // of slower provider probes, so ambiguous cutover retries are gated correctly. export async function readNewWorktreeRuntimeCapabilities( client: RpcClient ): Promise { + return (await probeNewWorktreeRuntimeCapabilities(client)).capabilities +} + +async function probeNewWorktreeRuntimeCapabilities( + client: RpcClient +): Promise { for (let migrationRetry = 0; ; migrationRetry += 1) { try { const response = await client.sendRequest('status.get') if (!response.ok) { - return UNSUPPORTED_CAPABILITIES + return { capabilities: UNSUPPORTED_CAPABILITIES, cache: 'authoritative' } } const result = (response as RpcSuccess).result as { capabilities?: string[] @@ -50,23 +76,26 @@ export async function readNewWorktreeRuntimeCapabilities( ) const advertisedIdempotency = result.worktreeCreateIdempotency return { - tasksSupported: capabilities.includes(MOBILE_TASKS_CAPABILITY), - worktreeCreateIdempotency: supportsIdempotency - ? advertisedIdempotency === undefined - ? { dedupeTtlMs: WORKTREE_CREATE_DEDUPE_TTL_LEGACY_HOST_MS } - : advertisedIdempotency !== null && - typeof advertisedIdempotency === 'object' && - !Array.isArray(advertisedIdempotency) - ? resolveWorktreeCreateIdempotencySupport( - (advertisedIdempotency as { dedupeTtlMs?: unknown }).dedupeTtlMs - ) - : { dedupeTtlMs: 0 } - : false, - hostPlatform: readMobileRuntimeHostPlatform(result) + capabilities: { + tasksSupported: capabilities.includes(MOBILE_TASKS_CAPABILITY), + worktreeCreateIdempotency: supportsIdempotency + ? advertisedIdempotency === undefined + ? { dedupeTtlMs: WORKTREE_CREATE_DEDUPE_TTL_LEGACY_HOST_MS } + : advertisedIdempotency !== null && + typeof advertisedIdempotency === 'object' && + !Array.isArray(advertisedIdempotency) + ? resolveWorktreeCreateIdempotencySupport( + (advertisedIdempotency as { dedupeTtlMs?: unknown }).dedupeTtlMs + ) + : { dedupeTtlMs: 0 } + : false, + hostPlatform: readMobileRuntimeHostPlatform(result) + }, + cache: 'authoritative' } } catch (error) { if (!isLogicalClientCutoverError(error) || migrationRetry >= STATUS_CUTOVER_MAX_RETRIES) { - return UNSUPPORTED_CAPABILITIES + return { capabilities: UNSUPPORTED_CAPABILITIES, cache: 'transient' } } } } @@ -82,22 +111,40 @@ export function useNewWorktreeRuntimeCapabilities( } { const [tasksSupported, setTasksSupported] = useState(false) const [hostPlatform, setHostPlatform] = useState(null) - const capabilityProbeRef = useRef<{ - client: RpcClient | null - promise: Promise - } | null>(null) - const getCapabilities = useCallback((): Promise => { - if (!capabilityProbeRef.current || capabilityProbeRef.current.client !== client) { - // Why: a queued tap can reach Create before passive effects run; lazily - // starting one shared probe keeps that path from failing open. - capabilityProbeRef.current = { - client, - promise: client - ? readNewWorktreeRuntimeCapabilities(client) - : Promise.resolve(UNSUPPORTED_CAPABILITIES) - } + const capabilityProbeRef = useRef(null) + const connectionRevisionRef = useRef(makeConnectionRevision(client)) + if (connectionRevisionRef.current.client !== client) { + connectionRevisionRef.current = makeConnectionRevision(client) + capabilityProbeRef.current = null + } + + const getCapabilityProbe = useCallback((): CapabilityProbeRecord => { + updateConnectionRevision(connectionRevisionRef.current) + const current = capabilityProbeRef.current + // Why: a transient answer gets one shared retry per recovered connection, + // while a host's authoritative answer remains durable across reconnects. + if ( + current?.client === client && + (current.cache === 'authoritative' || + current.connectionRevision === connectionRevisionRef.current.revision) + ) { + return current } - return capabilityProbeRef.current.promise + const record: CapabilityProbeRecord = { + client, + connectionRevision: connectionRevisionRef.current.revision, + cache: 'pending', + promise: client + ? probeNewWorktreeRuntimeCapabilities(client) + : Promise.resolve({ capabilities: UNSUPPORTED_CAPABILITIES, cache: 'authoritative' }) + } + capabilityProbeRef.current = record + void record.promise.then((result) => { + if (capabilityProbeRef.current === record) { + record.cache = result.cache + } + }) + return record }, [client]) useEffect(() => { @@ -105,20 +152,73 @@ export function useNewWorktreeRuntimeCapabilities( return } let stale = false - void getCapabilities().then((capabilities) => { - if (!stale) { - setTasksSupported(capabilities.tasksSupported) - setHostPlatform(capabilities.hostPlatform) + const refresh = (): void => { + const record = getCapabilityProbe() + void record.promise.then(({ capabilities }) => { + if (!stale && capabilityProbeRef.current === record) { + setTasksSupported(capabilities.tasksSupported) + setHostPlatform(capabilities.hostPlatform) + } + }) + } + const unsubscribe = client.onStateChange(() => { + const previousRevision = connectionRevisionRef.current.revision + updateConnectionRevision(connectionRevisionRef.current) + if ( + connectionRevisionRef.current.revision !== previousRevision && + capabilityProbeRef.current?.cache !== 'authoritative' + ) { + refresh() } }) + refresh() return () => { stale = true + unsubscribe() } - }, [client, enabled, getCapabilities, setTasksSupported]) + }, [client, enabled, getCapabilityProbe, setTasksSupported]) const getWorktreeCreateCutoverSupport = useCallback( - () => getCapabilities().then((capabilities) => capabilities.worktreeCreateIdempotency), - [getCapabilities] + () => + getCapabilityProbe().promise.then( + ({ capabilities }) => capabilities.worktreeCreateIdempotency + ), + [getCapabilityProbe] ) return { tasksSupported, hostPlatform, getWorktreeCreateCutoverSupport } } + +function makeConnectionRevision(client: RpcClient | null): ConnectionRevision { + return { + client, + revision: 0, + state: client?.getState() ?? null, + generation: readLogicalClientGeneration(client), + connectedAt: client?.getLastConnectedAt() ?? null + } +} + +function updateConnectionRevision(connection: ConnectionRevision): void { + const { client } = connection + if (!client) { + return + } + const state = client.getState() + const generation = readLogicalClientGeneration(client) + const connectedAt = client.getLastConnectedAt() + const reconnected = state === 'connected' && connection.state !== 'connected' + const migrated = generation !== null && generation !== connection.generation + const connectionChanged = connectedAt !== null && connectedAt !== connection.connectedAt + connection.state = state + connection.generation = generation + connection.connectedAt = connectedAt + if (reconnected || migrated || connectionChanged) { + connection.revision += 1 + } +} + +function readLogicalClientGeneration(client: RpcClient | null): number | null { + const getGeneration = (client as (RpcClient & { getGeneration?: () => number }) | null) + ?.getGeneration + return getGeneration?.() ?? null +}