mirror of
https://github.com/stablyai/orca.git
synced 2026-10-24 00:04:01 +00:00
Fix mobile capability recovery after reconnect
This commit is contained in:
@@ -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<typeof useNewWorktreeRuntimeCapabilities>
|
||||
|
||||
class CapabilityClient implements RpcClient {
|
||||
private readonly listeners = new Set<(state: ConnectionState) => void>()
|
||||
private state: ConnectionState = 'connected'
|
||||
readonly statusRequests = vi.fn<() => Promise<RpcResponse>>()
|
||||
readonly createRequests: Record<string, unknown>[] = []
|
||||
|
||||
constructor(statusOutcomes: Array<RpcResponse | Error>) {
|
||||
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<RpcResponse> {
|
||||
if (method === 'status.get') {
|
||||
return this.statusRequests()
|
||||
}
|
||||
this.createRequests.push((params ?? {}) as Record<string, unknown>)
|
||||
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()
|
||||
})
|
||||
})
|
||||
@@ -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<CapabilityProbeResult>
|
||||
}
|
||||
|
||||
type ConnectionRevision = {
|
||||
client: RpcClient | null
|
||||
revision: number
|
||||
state: ReturnType<RpcClient['getState']> | 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<NewWorktreeRuntimeCapabilities> {
|
||||
return (await probeNewWorktreeRuntimeCapabilities(client)).capabilities
|
||||
}
|
||||
|
||||
async function probeNewWorktreeRuntimeCapabilities(
|
||||
client: RpcClient
|
||||
): Promise<CapabilityProbeResult> {
|
||||
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<NodeJS.Platform | null>(null)
|
||||
const capabilityProbeRef = useRef<{
|
||||
client: RpcClient | null
|
||||
promise: Promise<NewWorktreeRuntimeCapabilities>
|
||||
} | null>(null)
|
||||
const getCapabilities = useCallback((): Promise<NewWorktreeRuntimeCapabilities> => {
|
||||
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<CapabilityProbeRecord | null>(null)
|
||||
const connectionRevisionRef = useRef<ConnectionRevision>(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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user