mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 00:03:20 +00:00
Fix mobile cutover activation and usage refresh loops
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -197,6 +197,10 @@ import {
|
||||
isDictationSetupRequiredError
|
||||
} from '../../../../src/dictation/mobile-dictation-setup'
|
||||
import { TerminalPaneView } from '../../../../src/session/TerminalPaneView'
|
||||
import {
|
||||
activateMobileSessionTab,
|
||||
focusMobileTerminal
|
||||
} from '../../../../src/session/mobile-session-tab-activation'
|
||||
import {
|
||||
getRepoIdFromMobileWorktreeId,
|
||||
isFileExistsErrorMessage,
|
||||
@@ -2931,15 +2935,13 @@ export default function SessionScreen() {
|
||||
}
|
||||
subscribeToTerminal(handle)
|
||||
if (client) {
|
||||
void client.sendRequest('terminal.focus', { terminal: handle }).catch(() => {})
|
||||
void focusMobileTerminal(client, handle).catch(() => {})
|
||||
if (matchingTab) {
|
||||
void client
|
||||
.sendRequest('session.tabs.activate', {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: matchingTab.id,
|
||||
notifyClients: false
|
||||
})
|
||||
.catch(() => {})
|
||||
void activateMobileSessionTab(client, {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: matchingTab.id,
|
||||
notifyClients: false
|
||||
}).catch(() => {})
|
||||
}
|
||||
}
|
||||
},
|
||||
@@ -2973,13 +2975,11 @@ export default function SessionScreen() {
|
||||
activeHandleRef.current = null
|
||||
setActiveHandle(null)
|
||||
if (client) {
|
||||
void client
|
||||
.sendRequest('session.tabs.activate', {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: tab.id,
|
||||
notifyClients: false
|
||||
})
|
||||
.catch(() => {})
|
||||
void activateMobileSessionTab(client, {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: tab.id,
|
||||
notifyClients: false
|
||||
}).catch(() => {})
|
||||
}
|
||||
return
|
||||
}
|
||||
@@ -2997,13 +2997,11 @@ export default function SessionScreen() {
|
||||
activeHandleRef.current = null
|
||||
setActiveHandle(null)
|
||||
if (client) {
|
||||
void client
|
||||
.sendRequest('session.tabs.activate', {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: tab.id,
|
||||
notifyClients: false
|
||||
})
|
||||
.catch(() => {})
|
||||
void activateMobileSessionTab(client, {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: tab.id,
|
||||
notifyClients: false
|
||||
}).catch(() => {})
|
||||
}
|
||||
if (tab.type === 'browser') {
|
||||
return
|
||||
@@ -4302,13 +4300,12 @@ export default function SessionScreen() {
|
||||
// Why: a hydrated headless/server-owned tab can already be active but still
|
||||
// pending; activation is the RPC that materializes or focuses its PTY handle.
|
||||
pendingTerminalActivationAttemptRef.current = activationKey
|
||||
void client
|
||||
.sendRequest('session.tabs.activate', {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: activePendingTerminalTab.id,
|
||||
leafId: activePendingTerminalTab.leafId,
|
||||
notifyClients: false
|
||||
})
|
||||
void activateMobileSessionTab(client, {
|
||||
worktree: `id:${worktreeId}`,
|
||||
tabId: activePendingTerminalTab.id,
|
||||
leafId: activePendingTerminalTab.leafId,
|
||||
notifyClients: false
|
||||
})
|
||||
.then((response) => {
|
||||
if (!response.ok) {
|
||||
if (pendingTerminalActivationAttemptRef.current === activationKey) {
|
||||
|
||||
@@ -58,7 +58,7 @@ describe('mobile session startup', () => {
|
||||
expect(pendingActivationEffect).toContain(
|
||||
'pendingTerminalActivationAttemptRef.current === activationKey'
|
||||
)
|
||||
expect(pendingActivationEffect).toContain("sendRequest('session.tabs.activate'")
|
||||
expect(pendingActivationEffect).toContain('activateMobileSessionTab(client,')
|
||||
expect(pendingActivationEffect).toContain('tabId: activePendingTerminalTab.id')
|
||||
expect(pendingActivationEffect).toContain('leafId: activePendingTerminalTab.leafId')
|
||||
expect(pendingActivationEffect).toContain('notifyClients: false')
|
||||
@@ -69,7 +69,7 @@ describe('mobile session startup', () => {
|
||||
})
|
||||
|
||||
it('keeps mobile session tab activation local to the phone', () => {
|
||||
const activationRequests = source.split("sendRequest('session.tabs.activate'").slice(1)
|
||||
const activationRequests = source.split('activateMobileSessionTab(client,').slice(1)
|
||||
|
||||
expect(activationRequests).toHaveLength(4)
|
||||
for (const request of activationRequests) {
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { LogicalClientCutoverError } from '../transport/stable-logical-rpc-client'
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
import type { RpcResponse } from '../transport/types'
|
||||
import { activateMobileSessionTab, focusMobileTerminal } from './mobile-session-tab-activation'
|
||||
|
||||
function success(): RpcResponse {
|
||||
return { id: 'rpc-1', ok: true, result: {}, _meta: { runtimeId: 'runtime-1' } }
|
||||
}
|
||||
|
||||
function clientWith(sendRequest: RpcClient['sendRequest']): Pick<RpcClient, 'sendRequest'> {
|
||||
return { sendRequest }
|
||||
}
|
||||
|
||||
describe('mobile session tab activation', () => {
|
||||
it('retries terminal focus once on the authenticated replacement after cutover', async () => {
|
||||
const sendRequest = vi
|
||||
.fn<RpcClient['sendRequest']>()
|
||||
.mockRejectedValueOnce(new LogicalClientCutoverError())
|
||||
.mockResolvedValueOnce(success())
|
||||
|
||||
await expect(focusMobileTerminal(clientWith(sendRequest), 'terminal-1')).resolves.toMatchObject(
|
||||
{
|
||||
ok: true
|
||||
}
|
||||
)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
expect(sendRequest).toHaveBeenNthCalledWith(1, 'terminal.focus', { terminal: 'terminal-1' })
|
||||
expect(sendRequest).toHaveBeenNthCalledWith(2, 'terminal.focus', { terminal: 'terminal-1' })
|
||||
})
|
||||
|
||||
it('retries session-tab activation with the same target after cutover', async () => {
|
||||
const sendRequest = vi
|
||||
.fn<RpcClient['sendRequest']>()
|
||||
.mockRejectedValueOnce(new LogicalClientCutoverError())
|
||||
.mockResolvedValueOnce(success())
|
||||
const params = {
|
||||
worktree: 'id:worktree-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'leaf-1',
|
||||
notifyClients: false as const
|
||||
}
|
||||
|
||||
await expect(activateMobileSessionTab(clientWith(sendRequest), params)).resolves.toMatchObject({
|
||||
ok: true
|
||||
})
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
expect(sendRequest).toHaveBeenNthCalledWith(1, 'session.tabs.activate', params)
|
||||
expect(sendRequest).toHaveBeenNthCalledWith(2, 'session.tabs.activate', params)
|
||||
})
|
||||
|
||||
it('does not retry unrelated transport failures', async () => {
|
||||
const sendRequest = vi.fn<RpcClient['sendRequest']>().mockRejectedValue(new Error('offline'))
|
||||
|
||||
await expect(focusMobileTerminal(clientWith(sendRequest), 'terminal-1')).rejects.toThrow(
|
||||
'offline'
|
||||
)
|
||||
expect(sendRequest).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('retries at most once when consecutive cutovers interrupt activation', async () => {
|
||||
const sendRequest = vi
|
||||
.fn<RpcClient['sendRequest']>()
|
||||
.mockRejectedValue(new LogicalClientCutoverError())
|
||||
|
||||
await expect(
|
||||
activateMobileSessionTab(clientWith(sendRequest), {
|
||||
worktree: 'id:worktree-1',
|
||||
tabId: 'tab-1',
|
||||
notifyClients: false
|
||||
})
|
||||
).rejects.toBeInstanceOf(LogicalClientCutoverError)
|
||||
expect(sendRequest).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,45 @@
|
||||
import type { RpcClient } from '../transport/rpc-client'
|
||||
import { LogicalClientCutoverError } from '../transport/stable-logical-rpc-client'
|
||||
import type { RpcResponse } from '../transport/types'
|
||||
|
||||
type ActivationClient = Pick<RpcClient, 'sendRequest'>
|
||||
|
||||
type MobileSessionTabActivationParams = {
|
||||
worktree: string
|
||||
tabId: string
|
||||
leafId?: string
|
||||
notifyClients: false
|
||||
}
|
||||
|
||||
async function retryIdempotentActivationAfterCutover(
|
||||
request: () => Promise<RpcResponse>
|
||||
): Promise<RpcResponse> {
|
||||
try {
|
||||
return await request()
|
||||
} catch (error) {
|
||||
if (!(error instanceof LogicalClientCutoverError)) {
|
||||
throw error
|
||||
}
|
||||
// Why: cutover rejects ambiguous in-flight work after the replacement is
|
||||
// active; these state-setting requests are idempotent and safe to repeat once.
|
||||
return request()
|
||||
}
|
||||
}
|
||||
|
||||
export function focusMobileTerminal(
|
||||
client: ActivationClient,
|
||||
terminal: string
|
||||
): Promise<RpcResponse> {
|
||||
return retryIdempotentActivationAfterCutover(() =>
|
||||
client.sendRequest('terminal.focus', { terminal })
|
||||
)
|
||||
}
|
||||
|
||||
export function activateMobileSessionTab(
|
||||
client: ActivationClient,
|
||||
params: MobileSessionTabActivationParams
|
||||
): Promise<RpcResponse> {
|
||||
return retryIdempotentActivationAfterCutover(() =>
|
||||
client.sendRequest('session.tabs.activate', params)
|
||||
)
|
||||
}
|
||||
@@ -764,6 +764,39 @@ describe('RateLimitService', () => {
|
||||
expect(fetchCodexRateLimits).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('does not refetch fresh provider data for replayed mobile subscriptions', async () => {
|
||||
const service = new RateLimitService()
|
||||
vi.mocked(fetchClaudeRateLimits).mockResolvedValue(okProvider('claude', 10))
|
||||
vi.mocked(fetchCodexRateLimits).mockResolvedValue(okProvider('codex', 20))
|
||||
|
||||
await service.refreshIfStale()
|
||||
await service.refreshIfStale()
|
||||
await service.refreshIfStale()
|
||||
|
||||
expect(fetchClaudeRateLimits).toHaveBeenCalledOnce()
|
||||
expect(fetchCodexRateLimits).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('does not queue a follow-up fetch when a mobile subscription replays mid-fetch', async () => {
|
||||
const service = new RateLimitService()
|
||||
const claude = deferred<ProviderRateLimits>()
|
||||
const codex = deferred<ProviderRateLimits>()
|
||||
vi.mocked(fetchClaudeRateLimits).mockReturnValue(claude.promise)
|
||||
vi.mocked(fetchCodexRateLimits).mockReturnValue(codex.promise)
|
||||
|
||||
const firstRefresh = service.refreshIfStale()
|
||||
await Promise.resolve()
|
||||
const replayedRefresh = service.refreshIfStale()
|
||||
|
||||
claude.resolve(okProvider('claude', 10))
|
||||
codex.resolve(okProvider('codex', 20))
|
||||
await firstRefresh
|
||||
await replayedRefresh
|
||||
|
||||
expect(fetchClaudeRateLimits).toHaveBeenCalledOnce()
|
||||
expect(fetchCodexRateLimits).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('waits for a queued explicit refresh when another fetch is already in flight', async () => {
|
||||
const service = new RateLimitService()
|
||||
const firstClaude = deferred<ProviderRateLimits>()
|
||||
@@ -1285,6 +1318,24 @@ describe('RateLimitService', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('does not start overlapping inactive Claude preview fetches', async () => {
|
||||
const service = new RateLimitService()
|
||||
const accountFetch = deferred<ProviderRateLimits>()
|
||||
service.setInactiveClaudeAccountsResolver(() => [
|
||||
{ id: 'account-1', managedAuthPath: '/tmp/account-1/auth' }
|
||||
])
|
||||
vi.mocked(fetchManagedAccountUsage).mockReturnValueOnce(accountFetch.promise)
|
||||
|
||||
const firstFetch = service.fetchInactiveClaudeAccountsOnOpen()
|
||||
await Promise.resolve()
|
||||
await service.fetchInactiveClaudeAccountsOnOpen()
|
||||
|
||||
expect(fetchManagedAccountUsage).toHaveBeenCalledTimes(1)
|
||||
|
||||
accountFetch.resolve(okProvider('claude', 50, Date.now()))
|
||||
await firstFetch
|
||||
})
|
||||
|
||||
it('does not start overlapping inactive Codex preview fetches', async () => {
|
||||
const service = new RateLimitService()
|
||||
const accountFetch = deferred<ProviderRateLimits>()
|
||||
|
||||
@@ -347,6 +347,14 @@ export class RateLimitService {
|
||||
return this.getState()
|
||||
}
|
||||
|
||||
async refreshIfStale(): Promise<RateLimitState> {
|
||||
// Why: reconnecting mobile subscribers need fresh backgrounded-desktop data,
|
||||
// but replaying a subscription must not queue another forced provider fetch.
|
||||
const plan = this.getActiveWindowRefreshPlan(Date.now())
|
||||
await this.runActiveWindowRefreshPlan(plan)
|
||||
return this.getState()
|
||||
}
|
||||
|
||||
async refreshGrok(): Promise<RateLimitState> {
|
||||
await this.fetchGrokOnly({ force: true })
|
||||
return this.getState()
|
||||
@@ -477,6 +485,9 @@ export class RateLimitService {
|
||||
return
|
||||
}
|
||||
this.pruneInactiveClaudeState()
|
||||
if (this.inactiveClaudeFetching.size > 0) {
|
||||
return
|
||||
}
|
||||
const accounts = this.inactiveClaudeAccountsResolver?.() ?? []
|
||||
if (accounts.length === 0) {
|
||||
return
|
||||
|
||||
@@ -8061,6 +8061,17 @@ export class OrcaRuntimeService {
|
||||
])
|
||||
}
|
||||
|
||||
// Why: connection migration replays subscriptions; use the stale-aware lane
|
||||
// so a reconnect cannot turn one mobile viewer into continuous forced fetches.
|
||||
async refreshAccountsForMobileSubscriber(): Promise<void> {
|
||||
const { rateLimits } = this.requireAccountServices()
|
||||
await Promise.allSettled([
|
||||
rateLimits.refreshIfStale(),
|
||||
rateLimits.fetchInactiveClaudeAccountsOnOpen(),
|
||||
rateLimits.fetchInactiveCodexAccountsOnOpen()
|
||||
])
|
||||
}
|
||||
|
||||
selectClaudeAccount(accountId: string | null): Promise<ClaudeRateLimitAccountsState> {
|
||||
return this.requireAccountServices().claudeAccounts.selectAccount(accountId)
|
||||
}
|
||||
|
||||
@@ -337,6 +337,11 @@ describe('remote runtime request connection integration', () => {
|
||||
listener({ claude: null, codex: null })
|
||||
}
|
||||
},
|
||||
refreshAccountsForMobileSubscriber: async () => {
|
||||
for (const listener of accountsListeners) {
|
||||
listener({ claude: null, codex: null })
|
||||
}
|
||||
},
|
||||
onAccountsChanged: (listener: (snapshot: unknown) => void) => {
|
||||
accountsListeners.add(listener)
|
||||
return () => accountsListeners.delete(listener)
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import { isStreamingMethod } from '../core'
|
||||
import { ACCOUNT_METHODS } from './accounts'
|
||||
|
||||
function method(name: string) {
|
||||
const found = ACCOUNT_METHODS.find((candidate) => candidate.name === name)
|
||||
if (!found) {
|
||||
throw new Error(`Missing method ${name}`)
|
||||
}
|
||||
return found
|
||||
}
|
||||
|
||||
describe('account RPC methods', () => {
|
||||
it('keeps explicit account-list refreshes on the forced refresh lane', async () => {
|
||||
const snapshot = { claude: null, codex: null }
|
||||
const runtime = {
|
||||
refreshAccountsForMobile: vi.fn().mockResolvedValue(undefined),
|
||||
getAccountsSnapshot: vi.fn(() => snapshot)
|
||||
} as unknown as OrcaRuntimeService
|
||||
const list = method('accounts.list')
|
||||
if (isStreamingMethod(list)) {
|
||||
throw new Error('accounts.list must be a request method')
|
||||
}
|
||||
|
||||
await expect(list.handler(undefined, { runtime })).resolves.toBe(snapshot)
|
||||
expect(runtime.refreshAccountsForMobile).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('uses a stale-aware refresh when a connection replays the subscription', async () => {
|
||||
const snapshot = { claude: null, codex: null }
|
||||
let cleanup: (() => void) | undefined
|
||||
const runtime = {
|
||||
getAccountsSnapshot: vi.fn(() => snapshot),
|
||||
onAccountsChanged: vi.fn(() => vi.fn()),
|
||||
registerSubscriptionCleanup: vi.fn((_id: string, nextCleanup: () => void) => {
|
||||
cleanup = nextCleanup
|
||||
}),
|
||||
refreshAccountsForMobile: vi.fn().mockResolvedValue(undefined),
|
||||
refreshAccountsForMobileSubscriber: vi.fn().mockResolvedValue(undefined)
|
||||
} as unknown as OrcaRuntimeService
|
||||
const subscribe = method('accounts.subscribe')
|
||||
if (!isStreamingMethod(subscribe)) {
|
||||
throw new Error('accounts.subscribe must be a streaming method')
|
||||
}
|
||||
const emit = vi.fn()
|
||||
|
||||
const running = subscribe.handler(undefined, { runtime, connectionId: 'connection-1' }, emit)
|
||||
await vi.waitFor(() => {
|
||||
expect(runtime.refreshAccountsForMobileSubscriber).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
expect(runtime.refreshAccountsForMobile).not.toHaveBeenCalled()
|
||||
expect(emit).toHaveBeenCalledWith(expect.objectContaining({ type: 'ready', snapshot }))
|
||||
cleanup?.()
|
||||
await running
|
||||
})
|
||||
})
|
||||
@@ -92,11 +92,11 @@ export const ACCOUNT_METHODS: readonly RpcAnyMethod[] = [
|
||||
)
|
||||
|
||||
// Why: emit the current snapshot synchronously so the phone has
|
||||
// something to render immediately, then kick a forced refresh that
|
||||
// will broadcast a fresh snapshot through the listener once each
|
||||
// provider fetch completes.
|
||||
// something to render immediately, then refresh only stale data.
|
||||
// Connection cutovers replay this subscription and must not turn the
|
||||
// manual-force lane into an unbounded provider-fetch loop.
|
||||
emit({ type: 'ready', subscriptionId, snapshot: runtime.getAccountsSnapshot() })
|
||||
void runtime.refreshAccountsForMobile()
|
||||
void runtime.refreshAccountsForMobileSubscriber()
|
||||
})
|
||||
}
|
||||
}),
|
||||
|
||||
Reference in New Issue
Block a user