diff --git a/mobile/src/mobile-web/mobile-web-capability-broker.ts b/mobile/src/mobile-web/mobile-web-capability-broker.ts index 812e29578b2..525aa1972c9 100644 --- a/mobile/src/mobile-web/mobile-web-capability-broker.ts +++ b/mobile/src/mobile-web/mobile-web-capability-broker.ts @@ -12,7 +12,7 @@ import { } from './mobile-web-broker-error' import { MobileWebOperationRateLimiter } from './mobile-web-operation-rate-limiter' import { MobileWebCommitMessageGeneration } from './mobile-web-commit-message-generation' -import { MobileWebCapabilitySubscriptions } from './mobile-web-capability-subscriptions' +import { MobileWebHostSubscriptions } from './mobile-web-host-subscriptions' import { MOBILE_WEB_PRODUCTION_GRANT_INDEX } from './mobile-web-production-grants' import { MobileWebTerminalStreams } from './mobile-web-terminal-streams' import { MOBILE_WEB_TERMINAL_CLIENT_CLOSURE } from './mobile-web-terminal-stream-retirement' @@ -41,7 +41,7 @@ type PendingRequest = { operationKey: string; subscriptionId?: string; cancelled export class MobileWebCapabilityBroker { private readonly pending = new Map() private readonly replay = new MobileWebBrokerReplayWindow() - private readonly subscriptions: MobileWebCapabilitySubscriptions + private readonly subscriptions: MobileWebHostSubscriptions private readonly terminalStreams: MobileWebTerminalStreams private readonly speechAuthority: MobileWebSpeechAuthority private readonly rateLimiter: MobileWebOperationRateLimiter @@ -60,7 +60,7 @@ export class MobileWebCapabilityBroker { postMessage: options.postMessage }) const posts = this.messages.subscriptionPosts() - this.subscriptions = new MobileWebCapabilitySubscriptions({ + this.subscriptions = new MobileWebHostSubscriptions({ ...posts, workspaceAuthority: this.authorities.workspace }) @@ -151,6 +151,8 @@ export class MobileWebCapabilityBroker { } const isHostRequest = mobileWebIsHostRequest(request) + const isHostForward = + isHostRequest || (request.capability === 'workspace' && request.operation === 'hostSubscribe') const grant = MOBILE_WEB_PRODUCTION_GRANT_INDEX.get(mobileWebOperationKey(request)) const expectsSubscription = mobileWebRequestExpectsSubscription(request) if (!grant || (request.mode === 'subscription') !== expectsSubscription) { @@ -161,7 +163,11 @@ export class MobileWebCapabilityBroker { await this.messages.error(request.requestId, 'invalid_request', false) return } - if (mobileWebEncodedByteLength(request.payload) > grant.limits.maxRequestBytes) { + // Generic forwarding checks the envelope after rewriting the workspace handle. + if ( + !isHostForward && + mobileWebEncodedByteLength(request.payload) > grant.limits.maxRequestBytes + ) { await this.messages.error(request.requestId, 'too_large', false) return } @@ -201,7 +207,7 @@ export class MobileWebCapabilityBroker { return } this.pending.delete(request.requestId) - await (mobileWebEncodedByteLength(payload) > grant.limits.maxResponseBytes + await (!isHostForward && mobileWebEncodedByteLength(payload) > grant.limits.maxResponseBytes ? this.messages.error(request.requestId, 'unavailable', false) : this.messages.success(request.requestId, payload)) } catch (error) { @@ -232,7 +238,7 @@ export class MobileWebCapabilityBroker { terminalClientId: this.options.terminalClientId, nativeAuthority: this.options.nativeAuthority, speechAuthority: this.speechAuthority, - hostSubscriptions: this.subscriptions.host, + hostSubscriptions: this.subscriptions, terminalStreams: this.terminalStreams, commitMessageGeneration: this.commitMessageGeneration, nativeChatAuthority: this.authorities.nativeChat, diff --git a/mobile/src/mobile-web/mobile-web-capability-schema-corpus.test.ts b/mobile/src/mobile-web/mobile-web-capability-schema-corpus.test.ts index 54682d663ce..1d6b82332cf 100644 --- a/mobile/src/mobile-web/mobile-web-capability-schema-corpus.test.ts +++ b/mobile/src/mobile-web/mobile-web-capability-schema-corpus.test.ts @@ -38,7 +38,15 @@ describe('mobile web capability schema corpus', () => { it('applies every production request-byte limit before host or native access', async () => { for (const [index, grant] of MOBILE_WEB_PRODUCTION_GRANTS.entries()) { const harness = createHarness() - const payload = { value: 'x'.repeat(grant.limits.maxRequestBytes + 1) } + const oversized = { value: 'x'.repeat(grant.limits.maxRequestBytes + 1) } + const payload = + grant.capability === 'workspace' && + (grant.operation === 'hostRequest' || grant.operation === 'hostSubscribe') + ? { + method: grant.operation === 'hostSubscribe' ? 'future.subscribe' : 'future.read', + params: oversized + } + : oversized await harness.broker.handle(requestFor(grant, payload, index)) diff --git a/mobile/src/mobile-web/mobile-web-capability-subscriptions.ts b/mobile/src/mobile-web/mobile-web-capability-subscriptions.ts deleted file mode 100644 index 193fb3fbd06..00000000000 --- a/mobile/src/mobile-web/mobile-web-capability-subscriptions.ts +++ /dev/null @@ -1,66 +0,0 @@ -import { MobileWebHostSubscriptions } from './mobile-web-host-subscriptions' -import type { MobileWebSubscriptionClosure } from './mobile-web-subscription-closure' -import type { - MobileWebSubscriptionLedgerConfig, - MobileWebSubscriptionLedgerHandle -} from './mobile-web-subscription-ledger' -import type { MobileWebWorkspaceAuthority } from './mobile-web-workspace-authority' - -export class MobileWebCapabilitySubscriptions { - readonly host: MobileWebHostSubscriptions - private readonly ledgers: MobileWebSubscriptionLedgerHandle[] - - constructor( - args: MobileWebSubscriptionLedgerConfig & { - workspaceAuthority: MobileWebWorkspaceAuthority - } - ) { - const shared = { - isActive: args.isActive, - postEvent: args.postEvent, - postClosed: args.postClosed - } - this.host = new MobileWebHostSubscriptions({ - ...shared, - workspaceAuthority: args.workspaceAuthority - }) - this.ledgers = [this.host] - } - - countForOperation(operationKey: string): number { - let count = 0 - for (const ledger of this.ledgers) { - count += ledger.countForOperation(operationKey) - } - return count - } - - cancel(subscriptionId: string): string | null { - for (const ledger of this.ledgers) { - const requestId = ledger.cancel(subscriptionId) - if (requestId !== null) { - return requestId - } - } - return null - } - - cancelByRequest(requestId: string): void { - for (const ledger of this.ledgers) { - ledger.cancelByRequest(requestId) - } - } - - /** Used when the page survives but its host feed does not, so every live entry learns it is over. */ - closeAll(closure: MobileWebSubscriptionClosure): void { - for (const ledger of this.ledgers) { - ledger.closeAll(closure) - } - } - - dispose(): void { - for (const ledger of this.ledgers) { - ledger.dispose() - } - } -} diff --git a/mobile/src/mobile-web/mobile-web-host-navigation-route.test.ts b/mobile/src/mobile-web/mobile-web-host-navigation-route.test.ts index 5abfb1b264e..1da536f853b 100644 --- a/mobile/src/mobile-web/mobile-web-host-navigation-route.test.ts +++ b/mobile/src/mobile-web/mobile-web-host-navigation-route.test.ts @@ -65,15 +65,6 @@ describe('mobile web host navigation route', () => { truncated: false }, 'unavailable' - ], - [ - 'malformed target', - { - worktrees: [{ worktreeId: HOST_WORKSPACE_ID }], - totalCount: 1, - truncated: false - }, - 'unavailable' ] ])('rejects a %s Desktop snapshot', async (_label, result, code) => { await expect( @@ -85,6 +76,19 @@ describe('mobile web host navigation route', () => { ).rejects.toMatchObject({ code }) }) + it('registers a workspace without repository metadata', async () => { + const route = await resolveMobileWebHostNavigationRoute( + HOST_WORKSPACE_ID, + hostClient({ worktrees: [{ worktreeId: HOST_WORKSPACE_ID }] }), + workspaceAuthority() + ) + expect(route).toEqual({ + kind: 'session', + workspaceId: `workspace_0_${'07'.repeat(16)}`, + workspaceName: 'Workspace' + }) + }) + it('rejects an oversized Desktop snapshot before registering its target', async () => { const result = { worktrees: [ diff --git a/mobile/src/mobile-web/mobile-web-host-navigation-route.ts b/mobile/src/mobile-web/mobile-web-host-navigation-route.ts index 766bd3b723b..91d3a18b10d 100644 --- a/mobile/src/mobile-web/mobile-web-host-navigation-route.ts +++ b/mobile/src/mobile-web/mobile-web-host-navigation-route.ts @@ -53,11 +53,7 @@ export async function resolveMobileWebHostNavigationRoute( } const match = matches[0] - const hostRepoId = boundedRequiredText(match.repoId, 512) - if (!hostRepoId) { - throw new MobileWebBrokerError('unavailable') - } - const pageWorkspaceId = authority.registerWorkspace(hostWorkspaceId, hostRepoId) + const pageWorkspaceId = authority.registerWorkspace(hostWorkspaceId) return { kind: 'session', workspaceId: pageWorkspaceId, diff --git a/mobile/src/mobile-web/mobile-web-host-subscriptions.ts b/mobile/src/mobile-web/mobile-web-host-subscriptions.ts index 136c5523ecf..b8cc4e5adcf 100644 --- a/mobile/src/mobile-web/mobile-web-host-subscriptions.ts +++ b/mobile/src/mobile-web/mobile-web-host-subscriptions.ts @@ -73,13 +73,14 @@ export class MobileWebHostSubscriptions extends MobileWebSubscriptionLedger< ) { return } - if (mobileWebHostPayloadByteLength(event) === undefined) { + const bytes = mobileWebHostPayloadByteLength(event) + if (bytes === undefined) { this.cancel(subscriptionId, { code: 'too_large', retryable: false }) return } const type = typeof event === 'object' && event !== null && 'type' in event ? event.type : undefined record.closing = type === 'end' || type === 'error' - this.enqueue(subscriptionId, record, event, record.closing) + this.enqueue(subscriptionId, record, event, record.closing, bytes) } } diff --git a/mobile/src/mobile-web/mobile-web-native-capability-operations.test.ts b/mobile/src/mobile-web/mobile-web-native-capability-operations.test.ts index 4f362756b88..7c170e83be7 100644 --- a/mobile/src/mobile-web/mobile-web-native-capability-operations.test.ts +++ b/mobile/src/mobile-web/mobile-web-native-capability-operations.test.ts @@ -114,7 +114,7 @@ describe('mobile web native capability operations', () => { it('resolves opaque workspace authority before reading or writing a shell draft', async () => { const harness = createHarness() const workspaceAuthority = new MobileWebWorkspaceAuthority((length) => new Uint8Array(length)) - const workspaceId = workspaceAuthority.registerWorkspace('host-workspace', 'host-repo') + const workspaceId = workspaceAuthority.registerWorkspace('host-workspace') const sessionChatDraftRead = vi.fn().mockResolvedValue('saved draft') const sessionChatDraftWrite = vi.fn().mockResolvedValue(undefined) harness.authority.sessionChatDraftRead = sessionChatDraftRead diff --git a/mobile/src/mobile-web/mobile-web-package-refresh.ts b/mobile/src/mobile-web/mobile-web-package-refresh.ts index e94f16b2056..e75373c15ac 100644 --- a/mobile/src/mobile-web/mobile-web-package-refresh.ts +++ b/mobile/src/mobile-web/mobile-web-package-refresh.ts @@ -60,6 +60,10 @@ export async function runMobileWebPackageRefresh(args: { downloaded.commit.buildId, MOBILE_WEB_BRIDGE_PROTOCOL_VERSION ) + if (!args.isCurrent()) { + await ExpoMobileWebShell.closeSession(session.sessionId).catch(() => {}) + return { kind: 'stale' } + } if (!(await args.publish(session, startedAt))) { return { kind: 'stale' } } diff --git a/mobile/src/mobile-web/mobile-web-production-grant-table.ts b/mobile/src/mobile-web/mobile-web-production-grant-table.ts index f05165ded35..2b8397fca26 100644 --- a/mobile/src/mobile-web/mobile-web-production-grant-table.ts +++ b/mobile/src/mobile-web/mobile-web-production-grant-table.ts @@ -44,8 +44,7 @@ export function capabilityGrants( })) } -/** Grants indexed by `capability.operation`. The broker resolves one per request, including every - * terminal keystroke, so a linear scan of all 226 entries is not an option. */ +// Indexed for every request, including terminal keystrokes. export function indexGrants( grants: readonly MobileWebOperationGrant[] ): ReadonlyMap { diff --git a/mobile/src/mobile-web/mobile-web-request-accounting.ts b/mobile/src/mobile-web/mobile-web-request-accounting.ts index 7a0d025e8df..6042d31b412 100644 --- a/mobile/src/mobile-web/mobile-web-request-accounting.ts +++ b/mobile/src/mobile-web/mobile-web-request-accounting.ts @@ -13,13 +13,7 @@ export function mobileWebRequestExpectsSubscription(request: { }): boolean { return ( (request.capability === 'workspace' && request.operation === 'hostSubscribe') || - ((request.capability === 'workspace' || - request.capability === 'account' || - request.capability === 'session' || - request.capability === 'sourceControl' || - request.capability === 'terminal' || - request.capability === 'nativeChat' || - request.capability === 'speech') && + ((request.capability === 'terminal' || request.capability === 'speech') && request.operation === 'subscribe') ) } diff --git a/mobile/src/mobile-web/mobile-web-subscription-ledger-lifecycle.test.ts b/mobile/src/mobile-web/mobile-web-subscription-ledger-lifecycle.test.ts index 02e5d2006d0..fb460c199c3 100644 --- a/mobile/src/mobile-web/mobile-web-subscription-ledger-lifecycle.test.ts +++ b/mobile/src/mobile-web/mobile-web-subscription-ledger-lifecycle.test.ts @@ -3,7 +3,6 @@ import type { MobileWebBridgeShellMessage } from '../../../src/shared/mobile-web import type { MobileWebSubscriptionClosure } from './mobile-web-subscription-closure' import type { RpcClient } from '../transport/rpc-client' import { MobileWebHostSubscriptions } from './mobile-web-host-subscriptions' -import { MobileWebCapabilitySubscriptions } from './mobile-web-capability-subscriptions' import { MobileWebBrokerMessageSender } from './mobile-web-broker-message-sender' import { MobileWebWorkspaceAuthority } from './mobile-web-workspace-authority' import { @@ -80,7 +79,7 @@ describe('subscription ledger teardown', () => { expect(ledger.countForOperation('workspace.hostSubscribe')).toBe(0) }) - it('fans closeAll out across every capability ledger', () => { + it('posts host-feed closures through the broker message sender', () => { const messages: MobileWebBridgeShellMessage[] = [] const sender = new MobileWebBrokerMessageSender({ context: MOBILE_WEB_BRIDGE_ROUNDTRIP_CONTEXT, @@ -91,12 +90,12 @@ describe('subscription ledger teardown', () => { }) const workspaceAuthority = new MobileWebWorkspaceAuthority(randomBytes) workspaceAuthority.synchronize(['host-workspace']) - const subscriptions = new MobileWebCapabilitySubscriptions({ + const subscriptions = new MobileWebHostSubscriptions({ ...sender.subscriptionPosts(), workspaceAuthority }) const client = stubClient(() => {}) - subscriptions.host.start({ + subscriptions.start({ requestId: 'r2', subscriptionId: 'host-1', payload: { method: 'mobileWeb.workspace.subscribe', params: {} }, diff --git a/mobile/src/mobile-web/mobile-web-subscription-ledger.ts b/mobile/src/mobile-web/mobile-web-subscription-ledger.ts index 5d33d5f05e6..cac4fbf5699 100644 --- a/mobile/src/mobile-web/mobile-web-subscription-ledger.ts +++ b/mobile/src/mobile-web/mobile-web-subscription-ledger.ts @@ -21,15 +21,6 @@ export type MobileWebSubscriptionLedgerOptions = { postClosed: MobileWebPostSubscriptionClosed } -/** The ledger surface callers share across every event and record type. */ -export type MobileWebSubscriptionLedgerHandle = { - cancel: (subscriptionId: string, closure?: MobileWebSubscriptionClosure) => string | null - cancelByRequest: (requestId: string) => void - countForOperation: (operationKey: string) => number - closeAll: (closure: MobileWebSubscriptionClosure) => void - dispose: () => void -} - /** What a concrete ledger supplies; the operation key is its own identity, not the caller's. */ export type MobileWebSubscriptionLedgerConfig = Omit< MobileWebSubscriptionLedgerOptions, @@ -55,7 +46,6 @@ export class MobileWebSubscriptionLedger< } record.active = false this.records.delete(subscriptionId) - this.retire(record) try { record.unsubscribe() } catch { @@ -138,7 +128,8 @@ export class MobileWebSubscriptionLedger< subscriptionId: string, record: TRecord, event: TEvent, - retireAfterDelivery = false + retireAfterDelivery = false, + retainedBytes = mobileWebEncodedByteLength(event) ): void { const sequence = record.sequence record.sequence += 1 @@ -151,7 +142,7 @@ export class MobileWebSubscriptionLedger< this.cancel(subscriptionId) } }, - mobileWebEncodedByteLength(event) + retainedBytes ) } @@ -186,7 +177,4 @@ export class MobileWebSubscriptionLedger< protected canDeliver(_subscriptionId: string, _record: TRecord): boolean { return true } - - /** Runs while the record is being removed, before the host handle is released. */ - protected retire(_record: TRecord): void {} } diff --git a/mobile/src/mobile-web/mobile-web-workspace-authority.test.ts b/mobile/src/mobile-web/mobile-web-workspace-authority.test.ts index 1c76833ba99..d1764dd30cd 100644 --- a/mobile/src/mobile-web/mobile-web-workspace-authority.test.ts +++ b/mobile/src/mobile-web/mobile-web-workspace-authority.test.ts @@ -25,19 +25,6 @@ describe('mobile web workspace authority', () => { expect(authority.pageWorkspaceId('repo::C:\\private\\second')).toBe(second) }) - it('revokes a repository handle only after workspace and catalog authority both omit it', () => { - const authority = new MobileWebWorkspaceAuthority((length) => new Uint8Array(length).fill(8)) - authority.synchronize(['folder-workspace']) - authority.synchronizeRepositories(['folder-repo']) - const repo = authority.pageRepoId('folder-repo') - - authority.synchronize([]) - expect(authority.pageRepoId('folder-repo')).toBe(repo) - - authority.synchronizeRepositories([]) - expect(() => authority.hostRepoId(repo)).toThrow('not_found') - }) - it('revokes every mapping when the shell session is cleared', () => { const authority = new MobileWebWorkspaceAuthority((length) => new Uint8Array(length)) authority.synchronize(['host-workspace']) @@ -46,6 +33,6 @@ describe('mobile web workspace authority', () => { authority.clear() expect(() => authority.hostWorkspaceId(pageWorkspaceId)).toThrow('not_found') - expect(() => authority.pageRepoId('host-repo')).toThrow('not_found') + expect(() => authority.pageWorkspaceId('host-workspace')).toThrow('not_found') }) }) diff --git a/mobile/src/mobile-web/mobile-web-workspace-authority.ts b/mobile/src/mobile-web/mobile-web-workspace-authority.ts index 19fd0959752..df33d3f9155 100644 --- a/mobile/src/mobile-web/mobile-web-workspace-authority.ts +++ b/mobile/src/mobile-web/mobile-web-workspace-authority.ts @@ -1,19 +1,10 @@ import { MobileWebBrokerError } from './mobile-web-broker-error' -import { - getRepoExecutionHostId, - parseExecutionHostId, - type ExecutionHostId -} from '../../../src/shared/execution-host' -import { getProjectIdentityKey } from '../../../src/shared/project-host-setup-projection' -import type { NewWorkspaceRepository } from '../worktree/host-workspace-creation-operations' declare const hostWorkspaceIdBrand: unique symbol -declare const hostRepoIdBrand: unique symbol /** Host-side identifiers the authority mints from an opaque page handle. Branding keeps a page * handle from being passed back in as if it were already resolved. */ export type MobileWebHostWorkspaceId = string & { readonly [hostWorkspaceIdBrand]: true } -export type MobileWebHostRepoId = string & { readonly [hostRepoIdBrand]: true } /** A value the host itself reported as a host identifier, so it never came from the page. Every * use is a boundary annotation, not a conversion of a page handle. */ @@ -21,20 +12,9 @@ export function mobileWebHostWorkspaceIdFromHost(value: string): MobileWebHostWo return value as MobileWebHostWorkspaceId } -export function mobileWebHostRepoIdFromHost(value: string): MobileWebHostRepoId { - return value as MobileWebHostRepoId -} - export class MobileWebWorkspaceAuthority { private readonly pageWorkspaceIdByHostId = new Map() private readonly hostWorkspaceIdByPageId = new Map() - private readonly hostRepoIdByHostWorkspaceId = new Map() - private readonly pageRepoIdByHostId = new Map() - private readonly hostRepoIdByPageId = new Map() - private catalogRepoIds = new Set() - private readonly hostConnectionIdByPageRepoId = new Map() - private readonly pageProjectIdByHostId = new Map() - private readonly pageExecutionHostIdByHostId = new Map() private nextHandle = 0 constructor(private readonly randomBytes: (length: number) => Uint8Array) {} @@ -50,11 +30,9 @@ export class MobileWebWorkspaceAuthority { } } } - this.hostRepoIdByHostWorkspaceId.clear() for (const hostWorkspaceId of hostWorkspaceIds) { this.rememberWorkspace(hostWorkspaceId) } - this.revokeUnreferencedRepos() } pageWorkspaceId(hostWorkspaceId: string): string { @@ -65,70 +43,6 @@ export class MobileWebWorkspaceAuthority { return pageWorkspaceId } - pageRepoId(hostRepoId: string): string { - const pageRepoId = this.pageRepoIdByHostId.get(hostRepoId) - if (!pageRepoId) { - throw new MobileWebBrokerError('not_found') - } - return pageRepoId - } - - hostRepoId(pageRepoId: string): MobileWebHostRepoId { - const hostRepoId = this.hostRepoIdByPageId.get(pageRepoId) - if (!hostRepoId) { - throw new MobileWebBrokerError('not_found') - } - return hostRepoId as MobileWebHostRepoId - } - - assertHostRepoBinding(pageRepoId: string, expectedHostRepoId: MobileWebHostRepoId): void { - if (this.hostRepoId(pageRepoId) !== expectedHostRepoId) { - throw new MobileWebBrokerError('conflict') - } - } - - synchronizeRepositories(hostRepoIds: readonly string[]): void { - this.catalogRepoIds = new Set(hostRepoIds) - hostRepoIds.forEach((hostRepoId) => this.rememberRepo(hostRepoId)) - this.revokeUnreferencedRepos() - } - - synchronizeCreationRepositories(repositories: readonly NewWorkspaceRepository[]): void { - this.synchronizeRepositories(repositories.map((repo) => repo.id)) - this.hostConnectionIdByPageRepoId.clear() - for (const repo of repositories) { - if (repo.connectionId) { - this.hostConnectionIdByPageRepoId.set(this.pageRepoId(repo.id), repo.connectionId) - } - this.rememberProject(getProjectIdentityKey(repo)) - this.rememberExecutionHost(getRepoExecutionHostId(repo)) - } - } - - pageProjectId(hostProjectId: string): string { - const pageProjectId = this.pageProjectIdByHostId.get(hostProjectId) - if (!pageProjectId) { - throw new MobileWebBrokerError('not_found') - } - return pageProjectId - } - - pageExecutionHostId(hostExecutionHostId: ExecutionHostId): ExecutionHostId { - const pageExecutionHostId = this.pageExecutionHostIdByHostId.get(hostExecutionHostId) - if (!pageExecutionHostId) { - throw new MobileWebBrokerError('not_found') - } - return pageExecutionHostId - } - - hostConnectionId(pageRepoId: string): string { - const connectionId = this.hostConnectionIdByPageRepoId.get(pageRepoId) - if (!connectionId) { - throw new MobileWebBrokerError('not_found') - } - return connectionId - } - hostWorkspaceId(pageWorkspaceId: string): MobileWebHostWorkspaceId { const hostWorkspaceId = this.hostWorkspaceIdByPageId.get(pageWorkspaceId) if (!hostWorkspaceId) { @@ -146,85 +60,33 @@ export class MobileWebWorkspaceAuthority { } } - registerWorkspace(hostWorkspaceId: string, hostRepoId?: string): string { + registerWorkspace(hostWorkspaceId: string): string { this.rememberWorkspace(hostWorkspaceId) - if (hostRepoId !== undefined) { - this.hostRepoIdByHostWorkspaceId.set(hostWorkspaceId, hostRepoId) - this.rememberRepo(hostRepoId) - } return this.pageWorkspaceId(hostWorkspaceId) } clear(): void { this.pageWorkspaceIdByHostId.clear() this.hostWorkspaceIdByPageId.clear() - this.hostRepoIdByHostWorkspaceId.clear() - this.pageRepoIdByHostId.clear() - this.hostRepoIdByPageId.clear() - this.catalogRepoIds.clear() - this.hostConnectionIdByPageRepoId.clear() - this.pageProjectIdByHostId.clear() - this.pageExecutionHostIdByHostId.clear() } private rememberWorkspace(hostWorkspaceId: string): void { if (this.pageWorkspaceIdByHostId.has(hostWorkspaceId)) { return } - const pageWorkspaceId = this.createHandle('workspace') + const pageWorkspaceId = this.createHandle() this.pageWorkspaceIdByHostId.set(hostWorkspaceId, pageWorkspaceId) this.hostWorkspaceIdByPageId.set(pageWorkspaceId, hostWorkspaceId) } - private rememberRepo(hostRepoId: string): void { - if (!this.pageRepoIdByHostId.has(hostRepoId)) { - const pageRepoId = this.createHandle('repo') - this.pageRepoIdByHostId.set(hostRepoId, pageRepoId) - this.hostRepoIdByPageId.set(pageRepoId, hostRepoId) - } - } - - private revokeUnreferencedRepos(): void { - const workspaceRepoIds = new Set(this.hostRepoIdByHostWorkspaceId.values()) - for (const [hostRepoId, pageRepoId] of this.pageRepoIdByHostId) { - if (workspaceRepoIds.has(hostRepoId) || this.catalogRepoIds.has(hostRepoId)) { - continue - } - this.pageRepoIdByHostId.delete(hostRepoId) - this.hostRepoIdByPageId.delete(pageRepoId) - this.hostConnectionIdByPageRepoId.delete(pageRepoId) - } - } - - private rememberProject(hostProjectId: string): void { - if (!this.pageProjectIdByHostId.has(hostProjectId)) { - this.pageProjectIdByHostId.set(hostProjectId, this.createHandle('project')) - } - } - - private rememberExecutionHost(hostExecutionHostId: ExecutionHostId): void { - if (this.pageExecutionHostIdByHostId.has(hostExecutionHostId)) { - return - } - const host = parseExecutionHostId(hostExecutionHostId) - if (!host || host.kind === 'local') { - this.pageExecutionHostIdByHostId.set(hostExecutionHostId, 'local') - return - } - this.pageExecutionHostIdByHostId.set( - hostExecutionHostId, - `${host.kind}:${this.createHandle('executionHost')}` - ) - } - - private createHandle(prefix: 'workspace' | 'repo' | 'project' | 'executionHost'): string { + private createHandle(): string { const bytes = this.randomBytes(16) if (bytes.byteLength !== 16) { throw new MobileWebBrokerError('internal') } const counter = this.nextHandle.toString(36) this.nextHandle += 1 - return `${prefix}_${counter}_${Array.from(bytes, byteToHex).join('')}` + return `workspace_${counter}_${Array.from(bytes, byteToHex).join('')}` } } diff --git a/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.test.ts b/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.test.ts index 6291788b14a..5c46de8c988 100644 --- a/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.test.ts +++ b/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.test.ts @@ -29,6 +29,27 @@ describe('mobile web workspace snapshot pager', () => { ).rejects.toMatchObject({ code: 'invalid_request' }) }) + it('retires an in-flight snapshot on cleanup without blocking the replacement client', async () => { + const authority = new MobileWebWorkspaceAuthority((length) => new Uint8Array(length)) + const pager = new MobileWebWorkspaceSnapshotPager((length) => new Uint8Array(length)) + const oldResponse = Promise.withResolvers>>() + const oldClient = { sendRequest: vi.fn(() => oldResponse.promise) } as unknown as RpcClient + const pending = pager.snapshot({ limit: 1 }, oldClient, authority) + const rejection = expect(pending).rejects.toMatchObject({ code: 'cancelled' }) + + pager.clear() + authority.clear() + const current = await pager.snapshot({ limit: 1 }, workspaceClient(2), authority) + oldResponse.resolve({ ok: true, result: { worktrees: [{ worktreeId: 'retired-workspace' }] } }) + await rejection + + expect(() => authority.pageWorkspaceId('retired-workspace')).toThrow('not_found') + expect(authority.hostWorkspaceId(current.workspaces[0]!.id)).toBe('host-workspace-0') + await expect( + pager.snapshot({ limit: 1, cursor: current.nextCursor! }, workspaceClient(0), authority) + ).resolves.toMatchObject({ workspaces: [{ name: 'Workspace 1' }], nextCursor: null }) + }) + it('revokes continuations on lifecycle cleanup and rejects oversized host lists', async () => { const authority = new MobileWebWorkspaceAuthority((length) => new Uint8Array(length)) const pager = new MobileWebWorkspaceSnapshotPager((length) => new Uint8Array(length)) diff --git a/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.ts b/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.ts index e872e010027..df5fc2c610d 100644 --- a/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.ts +++ b/mobile/src/mobile-web/mobile-web-workspace-snapshot-pager.ts @@ -20,7 +20,7 @@ type Continuation = { export class MobileWebWorkspaceSnapshotPager { private continuation: Continuation | null = null - private active = false + private active: object | null = null private nextCursorNumber = 0 constructor(private readonly randomBytes: (length: number) => Uint8Array) {} @@ -33,12 +33,16 @@ export class MobileWebWorkspaceSnapshotPager { if (this.active) { throw new MobileWebBrokerError('rate_limited') } - this.active = true + const request = {} + this.active = request try { const payload = MobileWebWorkspaceSnapshotPayloadSchema.parse(payloadValue) const continuation = payload.cursor ? this.consumeContinuation(payload.cursor) : await this.begin(client) + if (this.active !== request) { + throw new MobileWebBrokerError('cancelled') + } const page = mobileWebWorkspaceSnapshotPage( continuation.hostResult, payload.limit, @@ -52,16 +56,19 @@ export class MobileWebWorkspaceSnapshotPager { page.nextOffset === null ? null : this.retain(continuation, page.nextOffset) return MobileWebWorkspaceSnapshotResultSchema.parse({ ...page.snapshot, nextCursor }) } finally { - this.active = false + if (this.active === request) { + this.active = null + } } } clear(): void { + this.active = null this.continuation = null } private async begin(client: RpcClient): Promise { - this.clear() + this.continuation = null const response = await client.sendRequest('worktree.ps', { limit: MOBILE_WEB_WORKSPACE_LIST_LIMIT + 1 }) @@ -79,7 +86,7 @@ export class MobileWebWorkspaceSnapshotPager { private consumeContinuation(cursor: string): Continuation { const continuation = this.continuation - this.clear() + this.continuation = null if (!continuation || continuation.cursor !== cursor) { throw new MobileWebBrokerError('invalid_request') } diff --git a/mobile/src/mobile-web/use-mobile-web-package-session.test.ts b/mobile/src/mobile-web/use-mobile-web-package-session.test.ts index d68646c3d53..47e3710dd17 100644 --- a/mobile/src/mobile-web/use-mobile-web-package-session.test.ts +++ b/mobile/src/mobile-web/use-mobile-web-package-session.test.ts @@ -307,6 +307,53 @@ describe('useMobileWebPackageSession', () => { }) }) + it('allows recovery again after the desktop upgrades its build', async () => { + native.openSession.mockResolvedValue(SESSION_A) + downloadPackage.mockResolvedValue({ commit: { buildId: SESSION_A.buildId } }) + await mount('connected') + await act(async () => { + packageSession?.handleLoadFailure('mobile_web_generation_invalid') + await flushPromises() + }) + + await update('disconnected') + native.openSession.mockResolvedValue(SESSION_B) + downloadPackage.mockResolvedValue({ commit: { buildId: SESSION_B.buildId } }) + await update('connected') + expect(packageSession?.session).toEqual(SESSION_B) + removeHostCache.mockClear() + await act(async () => { + packageSession?.handleLoadFailure('mobile_web_generation_invalid') + await flushPromises() + }) + + expect(removeHostCache).toHaveBeenCalledWith(HOST.publicKeyB64) + }) + + it('does not reopen a new host when an old host cache removal finishes', async () => { + native.openSession.mockResolvedValue(SESSION_A) + await mount('disconnected') + const removal = deferred() + removeHostCache.mockReturnValue(removal.promise) + await act(async () => { + packageSession?.handleLoadFailure('mobile_web_generation_invalid') + await flushPromises() + }) + native.openSession.mockResolvedValue(SESSION_B) + await update('disconnected', HOST_B) + native.openSession.mockClear() + native.closeSession.mockClear() + + await act(async () => { + removal.resolve() + await flushPromises() + }) + + expect(packageSession?.session).toEqual(SESSION_B) + expect(native.openSession).not.toHaveBeenCalled() + expect(native.closeSession).not.toHaveBeenCalled() + }) + it('allows another drop after the host is selected again', async () => { native.openSession.mockResolvedValue(SESSION_A) downloadPackage.mockResolvedValue({ commit: { buildId: SESSION_A.buildId } }) @@ -376,6 +423,26 @@ describe('useMobileWebPackageSession', () => { expect(native.closeSession).toHaveBeenCalledWith(SESSION_A.sessionId) }) + it('closes a refreshed session opened after its connection was abandoned', async () => { + const refreshed = deferred() + native.openSession.mockImplementation((_host: string, buildId: string | null) => + buildId ? refreshed.promise : Promise.resolve(SESSION_A) + ) + downloadPackage.mockResolvedValue({ commit: { buildId: SESSION_B.buildId } }) + await mount('connected') + expect(packageSession?.session).toEqual(SESSION_A) + await update('disconnected') + + await act(async () => { + refreshed.resolve(SESSION_B) + await flushPromises() + }) + + expect(packageSession?.session).toEqual(SESSION_A) + expect(native.closeSession).toHaveBeenCalledWith(SESSION_B.sessionId) + expect(native.closeSession).not.toHaveBeenCalledWith(SESSION_A.sessionId) + }) + it('closes the previous host session when the selected host changes', async () => { native.openSession.mockResolvedValue(SESSION_A) await mount('disconnected') diff --git a/mobile/src/mobile-web/use-mobile-web-package-session.ts b/mobile/src/mobile-web/use-mobile-web-package-session.ts index 2f708f2c961..499a2d8659d 100644 --- a/mobile/src/mobile-web/use-mobile-web-package-session.ts +++ b/mobile/src/mobile-web/use-mobile-web-package-session.ts @@ -37,8 +37,8 @@ export function useMobileWebPackageSession({ const hostEpochRef = useRef(0) const ownedSessionRef = useRef(null) const cachedBuildRef = useRef>(Promise.resolve(null)) - // One re-download per host selection: a page that cannot load will not loop over the bundle. - const droppedGenerationRef = useRef(false) + // One re-download per build: a desktop upgrade gets its own recovery attempt. + const droppedGenerationRef = useRef(null) const droppedHostRef = useRef(undefined) const retryRef = useRef({ hostId: '', loadEpoch: -1, attempts: 0 }) const connectionId = currentConnectionId(client) @@ -105,7 +105,7 @@ export function useMobileWebPackageSession({ hostEpochRef.current = hostEpoch if (droppedHostRef.current !== host?.id) { droppedHostRef.current = host?.id - droppedGenerationRef.current = false + droppedGenerationRef.current = null } dispatch({ type: 'reopening', hasHost: Boolean(host) }) const closing = ownedSessionRef.current @@ -242,7 +242,7 @@ export function useMobileWebPackageSession({ return } mobileWebDiagnosticsStore.warning(hostId, reason ?? 'mobile_web_document_unavailable') - if (droppedGenerationRef.current || !owned) { + if (!owned || droppedGenerationRef.current === owned.buildId) { dispatch({ type: 'warning', warning: { message: 'Couldn’t open Orca.', code: reason } @@ -251,8 +251,8 @@ export function useMobileWebPackageSession({ } // The cached generation cannot render, so it is deleted and downloaded again; an unreachable // desktop leaves the shell in its offline state until the connection returns. - droppedGenerationRef.current = true - hostEpochRef.current += 1 + droppedGenerationRef.current = owned.buildId + const dropEpoch = ++hostEpochRef.current ownedSessionRef.current = null dispatch({ type: 'generation-dropping', @@ -261,7 +261,9 @@ export function useMobileWebPackageSession({ void (async () => { await ExpoMobileWebShell.closeSession(owned.sessionId).catch(() => {}) await removeMobileWebHostCache(host.publicKeyB64).catch(() => {}) - dispatch({ type: 'reload' }) + if (hostEpochRef.current === dropEpoch) { + dispatch({ type: 'reload' }) + } })() }, [host?.id, host?.publicKeyB64, host?.name] diff --git a/mobile/src/transport/mobile-relay-rpc-streams.ts b/mobile/src/transport/mobile-relay-rpc-streams.ts index 6188fa18ab8..e062cad3398 100644 --- a/mobile/src/transport/mobile-relay-rpc-streams.ts +++ b/mobile/src/transport/mobile-relay-rpc-streams.ts @@ -8,10 +8,7 @@ import { buildTerminalUnsubscribeParams, updateTerminalSubscriptionViewport } from './rpc-client-terminal-subscription' -import { - buildReadyStreamUnsubscribe, - buildServerSubscriptionUnsubscribe -} from './rpc-client-server-subscription' +import { buildServerSubscriptionUnsubscribe } from './rpc-client-server-subscription' import type { RpcClient } from './rpc-client' import { routeTerminalMultiplexFrame } from './rpc-client-terminal-multiplex' import type { RpcResponse, RpcSuccess } from './types' @@ -76,13 +73,6 @@ export class MobileRelayRpcStreams { listener: (result: unknown) => void, subscribeOptions?: Parameters[3] ): () => void { - if ( - subscribeOptions?.serverUnsubscribeMethod && - this.streams.size + this.cancelledSubscriptions.size >= 128 - ) { - listener({ type: 'error', message: 'Stream capacity reached' }) - return () => {} - } const id = this.options.nextId() const stream: StreamRecord = { method, @@ -131,7 +121,7 @@ export class MobileRelayRpcStreams { this.options.sendFrame({ id: this.options.nextId(), ...cancelled.unsubscribe }) } else if (typeof result.subscriptionId === 'string') { this.cancelledSubscriptions.delete(response.id) - const unsubscribe = buildReadyStreamUnsubscribe( + const unsubscribe = buildServerSubscriptionUnsubscribe( cancelled.method, result.subscriptionId, cancelled.serverUnsubscribeMethod @@ -229,7 +219,7 @@ export class MobileRelayRpcStreams { } } else { const unsubscribe = stream.subscriptionId - ? buildReadyStreamUnsubscribe( + ? buildServerSubscriptionUnsubscribe( stream.method, stream.subscriptionId, stream.serverUnsubscribeMethod diff --git a/mobile/src/transport/rpc-client-generic-stream-cleanup.test.ts b/mobile/src/transport/rpc-client-generic-stream-cleanup.test.ts index 64165973974..5d5b0179a45 100644 --- a/mobile/src/transport/rpc-client-generic-stream-cleanup.test.ts +++ b/mobile/src/transport/rpc-client-generic-stream-cleanup.test.ts @@ -68,18 +68,27 @@ describe('Desktop-advertised stream cleanup', () => { ) }) - it('bounds cancelled streams that have not supplied their cleanup token yet', () => { - const { registry } = fixture() - for (let index = 0; index < 128; index++) { - registry.subscribe('future.watch', {}, () => {}, { + it.each([true, false])( + 'removes ended streams before reconnect replay: streaming=%s', + (streaming) => { + const { registry, send, ready } = fixture() + const listener = vi.fn() + registry.subscribe('future.watch', {}, listener, { serverUnsubscribeMethod: 'future.release' - })() + }) + ready('lease') + registry.handleResponse({ + id: '1', + ok: true, + streaming, + result: { type: 'end' }, + _meta: { runtimeId: 'host' } + }) + expect(listener).toHaveBeenLastCalledWith({ type: 'end' }) + expect(registry.size()).toBe(0) + registry.markForReplay() + registry.replayAfterAuthentication() + expect(send).toHaveBeenCalledOnce() } - const listener = vi.fn() - registry.subscribe('future.watch', {}, listener, { serverUnsubscribeMethod: 'future.release' }) - expect(registry.size()).toBe(128) - expect(listener).toHaveBeenCalledWith( - expect.objectContaining({ type: 'error', error: { code: 'rate_limited' } }) - ) - }) + ) }) diff --git a/mobile/src/transport/rpc-client-server-subscription.ts b/mobile/src/transport/rpc-client-server-subscription.ts index 7d2fb65ac0b..a342702cd46 100644 --- a/mobile/src/transport/rpc-client-server-subscription.ts +++ b/mobile/src/transport/rpc-client-server-subscription.ts @@ -13,11 +13,3 @@ export function buildServerSubscriptionUnsubscribe( }[method] return unsubscribeMethod ? { method: unsubscribeMethod, params: { subscriptionId } } : null } - -export function buildReadyStreamUnsubscribe( - method: string, - subscriptionId: string, - cleanupMethod?: string -): { method: string; params: { subscriptionId: string } } | null { - return buildServerSubscriptionUnsubscribe(method, subscriptionId, cleanupMethod) -} diff --git a/mobile/src/transport/rpc-client-stream-registry.ts b/mobile/src/transport/rpc-client-stream-registry.ts index 2b80bc2eddc..e3569745553 100644 --- a/mobile/src/transport/rpc-client-stream-registry.ts +++ b/mobile/src/transport/rpc-client-stream-registry.ts @@ -7,7 +7,7 @@ import { buildTerminalUnsubscribeParams, updateTerminalSubscriptionViewport } from './rpc-client-terminal-subscription' -import { buildReadyStreamUnsubscribe } from './rpc-client-server-subscription' +import { buildServerSubscriptionUnsubscribe } from './rpc-client-server-subscription' import { isStreamingSubscriptionReadyResult, isTerminalSubscribedResult @@ -58,14 +58,6 @@ export class RpcClientStreamRegistry { listener: RpcStreamingListener, subscribeOptions?: RpcStreamSubscribeOptions ): () => void { - if (subscribeOptions?.serverUnsubscribeMethod && this.streams.size >= 128) { - listener({ - type: 'error', - error: { code: 'rate_limited' }, - message: 'Too many pending streams' - }) - return () => {} - } const id = this.options.nextId() const stream: StreamRequest = { method, @@ -127,10 +119,6 @@ export class RpcClientStreamRegistry { } handleResponse(response: RpcResponse): boolean { - if (response.ok && response.streaming === true) { - this.handleStreamingResponse(response) - return true - } const stream = this.streams.get(response.id) if (response.ok) { const result = (response as RpcSuccess).result as Record | null @@ -141,6 +129,10 @@ export class RpcClientStreamRegistry { this.remove(response.id) return true } + if (response.streaming === true) { + this.handleStreamingResponse(response) + return true + } if (stream && result?.type === 'scrollback') { stream.listener(result) return true @@ -256,12 +248,11 @@ export class RpcClientStreamRegistry { if (!stream.subscriptionId) { return } - const unsubscribe = stream.serverUnsubscribeMethod - ? { - method: stream.serverUnsubscribeMethod, - params: { subscriptionId: stream.subscriptionId } - } - : buildReadyStreamUnsubscribe(stream.method, stream.subscriptionId) + const unsubscribe = buildServerSubscriptionUnsubscribe( + stream.method, + stream.subscriptionId, + stream.serverUnsubscribeMethod + ) if (unsubscribe) { this.sendRpc(unsubscribe.method, unsubscribe.params) }