From b3f8a5ac8d13d75736d7639f2f83e189ff2e4b39 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 12 Nov 2026 11:37:12 -0800 Subject: [PATCH] Deliver browser recovery completion to the clients that need it (#41489) * Deliver client browser reconciliation completion to observed workspaces * Supply the workspace snapshot in empty adoption coverage * Skip recovery completion output after its recipient unsubscribes * Check activity work in the departed recipient regression * Select accepting session listeners before recovery delivery * Preserve scoped removal and recovery clearing delivery * Specify reconciliation listener selection type directly * Use inherited reconciliation listener selection contract * Type reconciliation observations by their actual owners * Update recovery test row-order import after main integration * fix(tests): import GitHub image capability from its owner (#41625) --------- Co-authored-by: Neil <4138956+nwparker@users.noreply.github.com> --- ...hosted-browser-persistence-test-harness.ts | 94 ++++++ ...lient-hosted-page-reconciliation-window.ts | 76 ++++- ...ge-reconciliation-delivery-test-harness.ts | 80 +++++ ...lient-page-reconciliation-delivery.test.ts | 129 ++++++++ .../client-page-reconciliation-scope.test.ts | 252 ++++++++++++++++ ...ent-page-recovery-hold-publication.test.ts | 177 +++++++++++ ...t-page-recovery-stream-observation.test.ts | 275 ++++++++++++++++++ ...lient-page-recovery-stream-test-harness.ts | 49 ++++ .../construction/runtime-identity-state.ts | 20 +- .../session-tabs-client-projection.ts | 24 ++ ...session-tabs-inventory-census-race.test.ts | 55 +++- .../session-tabs-inventory-collection.ts | 33 +++ .../session-tabs/session-tabs-inventory.ts | 119 ++++---- .../session-tabs-tab-delta-stream.test.ts | 193 ++++++++++++ .../rpc/methods/session-tabs/session-tabs.ts | 86 +++--- .../session-graph/session-graph-authority.ts | 25 +- .../session-tabs/mobile-session-membership.ts | 16 +- ...mobile-session-publication-preservation.ts | 7 +- .../mobile-session-tab-subscriptions.ts | 73 ++++- ...ca-runtime-session-tabs-path-alias.test.ts | 32 ++ .../session-tabs-change-publication.ts | 32 +- 21 files changed, 1718 insertions(+), 129 deletions(-) create mode 100644 src/main/runtime/browser/client-pages/client-hosted-browser-persistence-test-harness.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-reconciliation-delivery-test-harness.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-reconciliation-delivery.test.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-reconciliation-scope.test.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-recovery-hold-publication.test.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-recovery-stream-observation.test.ts create mode 100644 src/main/runtime/browser/client-pages/client-page-recovery-stream-test-harness.ts create mode 100644 src/main/runtime/rpc/methods/session-tabs/session-tabs-client-projection.ts create mode 100644 src/main/runtime/rpc/methods/session-tabs/session-tabs-tab-delta-stream.test.ts diff --git a/src/main/runtime/browser/client-pages/client-hosted-browser-persistence-test-harness.ts b/src/main/runtime/browser/client-pages/client-hosted-browser-persistence-test-harness.ts new file mode 100644 index 00000000000..7aeefc0ac94 --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-hosted-browser-persistence-test-harness.ts @@ -0,0 +1,94 @@ +import type { WorkspaceSessionState } from '../../../../shared/workspace-session/workspace-session-state-types' +import { orderTerminalRowsByTabGroups } from '../../../../shared/workspace-layout/terminal-topology/terminal-row-order' +const WT = 'repo-1::/tmp/worktree-a' + +const storeBase = { + getRepo: () => ({ + id: 'repo-1', + path: '/tmp/repo', + displayName: 'repo', + badgeColor: 'blue', + addedAt: 1 + }), + getRepos: () => [storeBase.getRepo()], + addRepo: () => {}, + updateRepo: () => { + throw new Error('unused fixture write') + }, + getAllWorktreeMeta: () => ({}), + getWorktreeMeta: () => undefined, + getGitHubCache: () => ({ pr: {}, issue: {} }), + setWorktreeMeta: () => { + throw new Error('unused fixture write') + }, + removeWorktreeMeta: () => {}, + getRetiredWorktreeNameRegistry: () => ({ exhaustedTiers: 0, names: [] }), + addRetiredWorktreeName: () => {}, + mergeRetiredWorktreeNames: () => false, + getSettings: () => ({ + workspaceDir: '/tmp/workspaces', + nestWorkspaces: false, + refreshLocalBaseRefOnWorktreeCreate: false, + branchPrefix: 'none', + branchPrefixCustom: '' + }) +} + +export function createProfile(worktreeCount = 0) { + const work = { writes: 0, normalizedWorktrees: 0 } + let measuringWrite = false + const tabsByWorktree: WorkspaceSessionState['tabsByWorktree'] = {} + for (let index = 0; index < worktreeCount; index += 1) { + const id = index === 0 ? WT : `repo-1::/tmp/worktree-${index}` + const rows = [ + { + id: `terminal-${index}`, + ptyId: null, + worktreeId: id, + title: 'Terminal', + customTitle: null, + color: null, + sortOrder: 0, + createdAt: 1 + } + ] + Object.defineProperty(tabsByWorktree, id, { + enumerable: true, + configurable: true, + get: () => { + if (measuringWrite) { + work.normalizedWorktrees += 1 + } + return rows + } + }) + } + const state: { session: WorkspaceSessionState } = { + session: { + activeRepoId: 'repo-1', + activeWorktreeId: WT, + activeTabId: null, + tabsByWorktree, + terminalLayoutsByTabId: {} + } satisfies WorkspaceSessionState + } + return { + work, + get session() { + return state.session + }, + store: { + ...storeBase, + getWorkspaceSession: () => state.session, + setWorkspaceSession: (next: WorkspaceSessionState) => { + work.writes += 1 + measuringWrite = true + try { + state.session = orderTerminalRowsByTabGroups(next) + } finally { + measuringWrite = false + } + } + } + } +} diff --git a/src/main/runtime/browser/client-pages/client-hosted-page-reconciliation-window.ts b/src/main/runtime/browser/client-pages/client-hosted-page-reconciliation-window.ts index 35656f4273d..ba2c3fac021 100644 --- a/src/main/runtime/browser/client-pages/client-hosted-page-reconciliation-window.ts +++ b/src/main/runtime/browser/client-pages/client-hosted-page-reconciliation-window.ts @@ -1,20 +1,90 @@ +import type { ClientHostedPageObservation } from '../../construction/runtime-identity-state' export const DEFAULT_CLIENT_HOSTED_RECONCILIATION_WINDOW_MS = 45_000 /** A client with no paired-device identity, which no host attach can ever speak for. */ const ANONYMOUS_CLIENT = '\0anonymous' +const READ_REPLY = Symbol('read-reply') /** Restart snapshots precede host inventories; hold browser-row culling until each client's host reconciles. * One client's attach proves nothing about another; bound never-attached clients by the deadline. */ export class ClientHostedPageReconciliationWindow { private readonly reconciledClients = new Set() + private readonly observations = new Map>>() constructor( private readonly openedAt: number, private readonly windowMs: number = DEFAULT_CLIENT_HOSTED_RECONCILIATION_WINDOW_MS ) {} - markReconciled(pairedDeviceId: string): void { - this.reconciledClients.add(clientKey(pairedDeviceId)) + markReconciled(pairedDeviceId: string): ReadonlySet | undefined { + const key = clientKey(pairedDeviceId) + this.reconciledClients.add(key) + const observations = this.observations.get(key) + this.observations.delete(key) + return observations && new Set([...observations.values()].flatMap((held) => [...held])) + } + + forgetClient(pairedDeviceId: string): void { + this.observations.delete(clientKey(pairedDeviceId)) + } + + observe( + frames: readonly { worktree: string; clientHostedPagesUnreconciled?: true; removed?: true }[], + pairedDeviceId?: string, + observer: ClientHostedPageObservation = READ_REPLY + ): void { + const key = clientKey(pairedDeviceId) + const observations = + this.observations.get(key) ?? new Map>() + const held = observations.get(observer) ?? new Set() + for (const frame of frames) { + if (frame.clientHostedPagesUnreconciled && !frame.removed) { + held.add(frame.worktree) + } else { + held.delete(frame.worktree) + } + } + if (held.size) { + observations.set(observer, held) + } else { + observations.delete(observer) + } + if (observations.size) { + this.observations.set(key, observations) + } else { + this.observations.delete(key) + } + } + + forgetObservation( + pairedDeviceId: string | undefined, + observer: ClientHostedPageObservation + ): void { + const key = clientKey(pairedDeviceId) + const observations = this.observations.get(key) + observations?.delete(observer) + if (observations?.size === 0) { + this.observations.delete(key) + } + } + + forgetWorktree(worktree: string): void { + for (const [key, observations] of this.observations) { + for (const [observer, held] of observations) { + held.delete(worktree) + if (!held.size) { + observations.delete(observer) + } + } + if (!observations.size) { + this.observations.delete(key) + } + } + } + + dispose(): void { + this.observations.clear() + this.reconciledClients.clear() } isUnreconciled(pairedDeviceId: string | undefined, now: number): boolean { @@ -24,7 +94,7 @@ export class ClientHostedPageReconciliationWindow { } /** Set and clear per recipient: shared fan-out frames must not inherit another client's hold. */ - holdFor( + holdFor( result: T, pairedDeviceId: string | undefined, now: number diff --git a/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery-test-harness.ts b/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery-test-harness.ts new file mode 100644 index 00000000000..46b071a176f --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery-test-harness.ts @@ -0,0 +1,80 @@ +import { OrcaRuntimeService } from '../../orca-runtime' +import type { RuntimeMobileSessionTabsResult } from '../../../../shared/runtime-session-contracts' +import { createProfile } from './client-hosted-browser-persistence-test-harness' +import { RpcDispatcher } from '../../rpc/dispatcher' +import { BROWSER_CLIENT_HOST_METHODS } from '../../rpc/methods/browser/browser-client-host' +import { BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY } from '../../../../shared/browser/runtime/browser-runtime-capabilities' + +class ReconciliationDeliveryRuntime extends OrcaRuntimeService { + projectedFrames = 0 + sessionTabsSequence() { + return this.mobileSessionTabsChangeSequence + } + protected override projectMobileSessionTabsForClient( + result: RuntimeMobileSessionTabsResult, + clientId?: string + ) { + this.projectedFrames += 1 + return super.projectMobileSessionTabsForClient(result, clientId) + } + peekSnapshotForClient(worktree: string, clientId: string) { + return this.getMobileSessionTabsForWorktree(worktree, clientId, { + includeAgentSenders: false + }) + } + snapshotForClient(worktree: string, clientId: string) { + const result = this.peekSnapshotForClient(worktree, clientId) + this.observeClientHostedPageSnapshots([result], clientId) + return result + } + override getRuntimeId(): string { + return 'runtime-a' + } +} + +export function createReconciliationDeliveryRuntime() { + const profile = createProfile() + const hostRuntime = new ReconciliationDeliveryRuntime(profile.store, undefined, {}) + const rpc = new RpcDispatcher({ runtime: hostRuntime, methods: BROWSER_CLIENT_HOST_METHODS }) + const streams: { stopped: AbortController; completion: Promise }[] = [] + function attach() { + const stopped = new AbortController() + const completion = rpc.dispatchStreaming( + { + id: 'empty-adoption', + authToken: 'test', + method: 'browser.clientHost.attach', + params: { + authorityRuntimeId: 'runtime-a', + browserHostClientId: 'client-a', + hostCapabilities: ['webview'], + pageCommandProtocolVersion: 1, + pageReconciliationProtocolVersion: 1, + pageInventoryProtocolVersion: 1, + pageInventory: [], + leaseReconnectProtocolVersion: 1 + } + }, + () => {}, + { + connectionId: 'empty-adoption', + pairedDeviceId: 'device-a', + clientKind: 'runtime', + clientCapabilities: [BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY], + signal: stopped.signal + } + ) + streams.push({ stopped, completion }) + return { abort: () => stopped.abort(), completion } + } + return { + hostRuntime, + attach, + async close() { + for (const stream of streams) { + stream.stopped.abort() + } + await Promise.all(streams.map((stream) => stream.completion)) + } + } +} diff --git a/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery.test.ts b/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery.test.ts new file mode 100644 index 00000000000..3c5bafe39df --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-reconciliation-delivery.test.ts @@ -0,0 +1,129 @@ +import { expect, it, vi } from 'vitest' +import { createReconciliationDeliveryRuntime } from './client-page-reconciliation-delivery-test-harness' +import { RpcDispatcher } from '../../rpc/dispatcher' +import { SESSION_TAB_METHODS } from '../../rpc/methods/session-tabs/session-tabs' + +import { openRecoveryTabsStream } from './client-page-recovery-stream-test-harness' + +const WORKSPACE = 'repo-1::/tmp/worktree-a' + +it.each(['session.tabs.subscribe', 'session.tabs.subscribeAll'])( + 'forgets %s initial hold after the same stream receives an unheld frame', + async (method) => { + vi.useFakeTimers({ toFake: ['Date'] }) + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const stopped = new AbortController() + const replies: string[] = [] + try { + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree: WORKSPACE, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const stream = rpc.dispatchStreaming( + { + id: 'held-stream', + authToken: 'test', + method, + params: method === 'session.tabs.subscribe' ? { worktree: `id:${WORKSPACE}` } : {} + }, + (reply) => replies.push(reply), + { + connectionId: 'held-connection', + pairedDeviceId: 'device-a', + clientKind: 'mobile', + signal: stopped.signal + } + ) + await vi.waitFor(() => + expect( + replies.some((reply) => + ['snapshots', 'snapshot'].includes(JSON.parse(reply).result?.type) + ) + ).toBe(true) + ) + expect( + replies.some((reply) => + (JSON.parse(reply).result?.snapshots ?? [JSON.parse(reply).result]).some( + (frame: { worktree: string; clientHostedPagesUnreconciled?: true }) => + frame.worktree === WORKSPACE && frame.clientHostedPagesUnreconciled + ) + ) + ).toBe(true) + vi.setSystemTime(Date.now() + 46_000) + runtime.notifyMobileSessionTabsChanged(WORKSPACE) + expect( + replies.some( + (reply) => + JSON.parse(reply).result?.type === 'updated' && + JSON.parse(reply).result?.clientHostedPagesUnreconciled === undefined + ) + ).toBe(true) + runtime.projectedFrames = 0 + const delivered = replies.length + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(0) + expect(replies).toHaveLength(delivered) + stopped.abort() + await stream + } finally { + stopped.abort() + await h.close() + vi.useRealTimers() + } + } +) + +it('delivers a cleared snapshot when authenticated empty adoption completes without another workspace change', async () => { + const h = createReconciliationDeliveryRuntime() + h.hostRuntime.attachWindow(1) + h.hostRuntime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree: WORKSPACE, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + const stream = openRecoveryTabsStream(h.hostRuntime, 'empty-adoption-consumer') + try { + await stream.initial + expect( + stream.frames.findLast((frame) => frame.worktree === WORKSPACE)?.clientHostedPagesUnreconciled + ).toBe(true) + const attached = h.attach() + await vi.waitFor(() => + expect( + stream.frames.findLast((frame) => frame.worktree === WORKSPACE) + ?.clientHostedPagesUnreconciled + ).toBeUndefined() + ) + expect(stream.errors).toEqual([]) + attached.abort() + await attached.completion + } finally { + stream.stop() + await stream.completion + await h.close() + } +}) diff --git a/src/main/runtime/browser/client-pages/client-page-reconciliation-scope.test.ts b/src/main/runtime/browser/client-pages/client-page-reconciliation-scope.test.ts new file mode 100644 index 00000000000..f48d8c48b88 --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-reconciliation-scope.test.ts @@ -0,0 +1,252 @@ +import { expect, it, vi } from 'vitest' +import { RuntimeAgentSenderAssociations } from '../../agent-senders/runtime-agent-sender-associations' +import { RpcDispatcher } from '../../rpc/dispatcher' +import { SESSION_TAB_METHODS } from '../../rpc/methods/session-tabs/session-tabs' +import { createReconciliationDeliveryRuntime } from './client-page-reconciliation-delivery-test-harness' +import { openRecoveryTabsStream } from './client-page-recovery-stream-test-harness' + +vi.mock('../../../../shared/sidebar-inputs-switch', () => ({ ORCA_SIDEBAR_INPUTS_IS_MAIN: true })) + +function snapshots(count: number) { + return Array.from({ length: count }, (_, index) => ({ + worktree: `repo-1::/tmp/covered-${index}`, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + })) +} + +it.each([750, 7500])( + 'clears only the live workspace channel among %i held read replies', + async (count) => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const corpus = snapshots(count) + const workspace = corpus[0]!.worktree + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: corpus }) + const global = openRecoveryTabsStream(runtime, 'departing-global') + const active = openRecoveryTabsStream( + runtime, + 'surviving-active', + `id:${workspace}`, + 'device-a', + 'runtime' + ) + const refresh = vi.spyOn(runtime.sidebarFacts!.workspaceActivityFacts, 'refreshWorkspace') + const senderWorkspaces: string[] = [] + const build = RuntimeAgentSenderAssociations.prototype.buildForFramePass + const senderPass = vi + .spyOn(RuntimeAgentSenderAssociations.prototype, 'buildForFramePass') + .mockImplementation(function (this: RuntimeAgentSenderAssociations) { + const scope = build.call(this) + return (worktree) => { + senderWorkspaces.push(worktree) + return scope(worktree) + } + }) + try { + await Promise.all([global.initial, active.initial]) + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const reads: string[] = [] + await rpc.dispatchStreaming( + { id: 'held-census', authToken: 'test', method: 'session.tabs.listAll', params: {} }, + (reply) => reads.push(reply), + { connectionId: 'held-census', pairedDeviceId: 'device-a', clientKind: 'runtime' } + ) + const supplied = JSON.parse(reads[0]!).result.snapshots + expect(supplied).toHaveLength(count) + expect( + supplied.every( + (frame: { clientHostedPagesUnreconciled?: true }) => + frame.clientHostedPagesUnreconciled === true + ) + ).toBe(true) + expect(active.frames.at(-1)?.clientHostedPagesUnreconciled).toBe(true) + global.stop() + await global.completion + runtime.projectedFrames = 0 + senderWorkspaces.length = 0 + refresh.mockClear() + const sequence = runtime.sessionTabsSequence() + const delivered = active.frames.length + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(1) + expect(runtime.sessionTabsSequence()).toBe(sequence + 1) + expect(active.frames).toHaveLength(delivered + 1) + expect(active.frames.at(-1)).toMatchObject({ worktree: workspace }) + expect(active.frames.at(-1)?.clientHostedPagesUnreconciled).toBeUndefined() + expect(senderWorkspaces).toEqual([workspace]) + expect(refresh).not.toHaveBeenCalled() + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(1) + expect(runtime.sessionTabsSequence()).toBe(sequence + 1) + runtime.projectedFrames = 0 + senderWorkspaces.length = 0 + runtime.notifyMobileSessionTabsChanged(corpus[1]!.worktree) + runtime.onMobileSessionTabsChanged(() => {}, 'flush-device', 'mobile')() + expect(runtime.projectedFrames).toBe(0) + expect(senderWorkspaces).toEqual([]) + expect(active.frames).toHaveLength(delivered + 1) + const later = openRecoveryTabsStream(runtime, 'later-uncovered', `id:${corpus[1]!.worktree}`) + await later.initial + expect(later.frames.at(-1)?.clientHostedPagesUnreconciled).toBeUndefined() + later.stop() + await later.completion + } finally { + global.stop() + active.stop() + await Promise.all([global.completion, active.completion]) + await h.close() + refresh.mockRestore() + senderPass.mockRestore() + } + } +) + +it('stops projecting remaining owed workspaces when clearing output releases the stream', async () => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: snapshots(2) }) + const stopped = new AbortController() + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const cleared: string[] = [] + try { + await rpc.dispatchStreaming( + { + id: 'release-on-clear', + authToken: 'test', + method: 'session.tabs.subscribeAll', + params: {} + }, + (reply) => { + const frame = JSON.parse(reply).result + if (frame?.type === 'updated') { + cleared.push(frame.worktree) + stopped.abort() + } + }, + { + connectionId: 'release-on-clear', + pairedDeviceId: 'device-a', + clientKind: 'mobile', + signal: stopped.signal + } + ) + runtime.projectedFrames = 0 + const sequence = runtime.sessionTabsSequence() + runtime.markClientHostedPagesReconciled('device-a') + expect(cleared).toEqual(['repo-1::/tmp/covered-0']) + expect(runtime.projectedFrames).toBe(1) + expect(runtime.sessionTabsSequence()).toBe(sequence + 1) + } finally { + stopped.abort() + await h.close() + } +}) + +it('refreshes sender scopes only through accepting desktop channels and settles after broad delivery', async () => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const corpus = snapshots(2) + const activeWorkspace = corpus[0]!.worktree + const otherWorkspace = corpus[1]!.worktree + let peerKey = 'sender-peer-before' + runtime.setAgentSenderPeerKey(() => peerKey) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: corpus }) + const initial = openRecoveryTabsStream( + runtime, + 'initial-sender-census', + undefined, + 'device-a', + 'runtime' + ) + const active = openRecoveryTabsStream( + runtime, + 'active-sender-scope', + `id:${activeWorkspace}`, + 'device-a', + 'runtime' + ) + const phone = openRecoveryTabsStream( + runtime, + 'phone-sender-witness', + `id:${otherWorkspace}`, + 'phone' + ) + let broad: ReturnType | undefined + try { + await Promise.all([initial.initial, active.initial, phone.initial]) + initial.stop() + await initial.completion + const before = phone.frames.at(-1)!.snapshotVersion + const beforeCount = phone.frames.length + peerKey = 'sender-peer-after' + runtime.refreshAgentSenderAssociations() + runtime.onMobileSessionTabsChanged(() => {}, 'flush', 'mobile')() + runtime.refreshAgentSenderAssociations() + runtime.onMobileSessionTabsChanged(() => {}, 'flush', 'mobile')() + expect(phone.frames.at(-1)!.snapshotVersion).toBe(before) + expect(phone.frames).toHaveLength(beforeCount) + broad = openRecoveryTabsStream( + runtime, + 'returning-sender-census', + undefined, + 'device-a', + 'runtime' + ) + await broad.initial + const refreshed = broad.frames.findLast((frame) => frame.worktree === otherWorkspace) + expect(refreshed?.agentSenderAssociations?.peerFingerprint).toBe( + 'pXI0wdMoVIozNMf8PhOCNlB9sTAAEaBF6DfNP9doYqk' + ) + runtime.refreshAgentSenderAssociations() + runtime.onMobileSessionTabsChanged(() => {}, 'flush', 'mobile')() + const settled = phone.frames.length + const version = phone.frames.at(-1)!.snapshotVersion + runtime.refreshAgentSenderAssociations() + runtime.onMobileSessionTabsChanged(() => {}, 'flush', 'mobile')() + expect(phone.frames).toHaveLength(settled) + expect(phone.frames.at(-1)!.snapshotVersion).toBe(version) + } finally { + initial.stop() + active.stop() + phone.stop() + broad?.stop() + await Promise.all([initial.completion, active.completion, phone.completion, broad?.completion]) + await h.close() + } +}) + +it('routes removal only to the matching workspace and broad RPC streams', async () => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const corpus = snapshots(2) + const keptWorkspace = corpus[0]!.worktree + const removedWorkspace = corpus[1]!.worktree + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: corpus }) + const kept = openRecoveryTabsStream(runtime, 'kept-workspace', `id:${keptWorkspace}`) + const removed = openRecoveryTabsStream(runtime, 'removed-workspace', `id:${removedWorkspace}`) + const all = openRecoveryTabsStream(runtime, 'all-workspaces') + try { + await Promise.all([kept.initial, removed.initial, all.initial]) + const delivered = kept.frames.length + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: [corpus[0]!] }) + runtime.onMobileSessionTabsChanged(() => {}, 'flush', 'mobile')() + expect(kept.frames).toHaveLength(delivered) + expect(removed.frames.at(-1)).toMatchObject({ worktree: removedWorkspace, removed: true }) + expect(all.frames.at(-1)).toMatchObject({ worktree: removedWorkspace, removed: true }) + } finally { + kept.stop() + removed.stop() + all.stop() + await Promise.all([kept.completion, removed.completion, all.completion]) + await h.close() + } +}) diff --git a/src/main/runtime/browser/client-pages/client-page-recovery-hold-publication.test.ts b/src/main/runtime/browser/client-pages/client-page-recovery-hold-publication.test.ts new file mode 100644 index 00000000000..7587dbaf817 --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-recovery-hold-publication.test.ts @@ -0,0 +1,177 @@ +import { expect, it, vi } from 'vitest' +import { createReconciliationDeliveryRuntime } from './client-page-reconciliation-delivery-test-harness' +import type { RuntimeMobileSessionTabsResult } from '../../../../shared/runtime-session-contracts' + +vi.mock('../../../../shared/sidebar-inputs-switch', () => ({ ORCA_SIDEBAR_INPUTS_IS_MAIN: true })) + +it.each([750, 7500])( + 'only clears previously held frames in a %i-workspace runtime', + async (count) => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const snapshots = Array.from({ length: count }, (_, index) => ({ + worktree: `repo-1::/tmp/worktree-${index}`, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + })) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: snapshots }) + runtime.markClientHostedPagesReconciled('device-a') + const frames: RuntimeMobileSessionTabsResult[] = [] + const otherFrames: RuntimeMobileSessionTabsResult[] = [] + const unsubscribe = runtime.onMobileSessionTabsChanged( + (frame) => frames.push(frame), + 'device-a', + 'mobile' + ) + const unsubscribeHeld = runtime.onMobileSessionTabsChanged( + (frame) => otherFrames.push(frame), + 'device-b', + 'mobile' + ) + const facts = runtime.sidebarFacts + if (!facts) { + throw new Error('sidebar facts fixture is required') + } + const refresh = vi.spyOn(facts.workspaceActivityFacts, 'refreshWorkspace') + try { + runtime.projectedFrames = 0 + const attached = h.attach() + expect(runtime.projectedFrames).toBe(0) + expect(frames).toEqual([]) + expect(otherFrames).toEqual([]) + expect(refresh).not.toHaveBeenCalled() + attached.abort() + await attached.completion + const heldWorktrees = [snapshots[0]!.worktree, snapshots[count - 1]!.worktree] + for (const worktree of heldWorktrees) { + expect(runtime.snapshotForClient(worktree, 'device-b').clientHostedPagesUnreconciled).toBe( + true + ) + } + runtime.projectedFrames = 0 + runtime.markClientHostedPagesReconciled('device-b') + expect(runtime.projectedFrames).toBe(2) + expect(otherFrames.map((frame) => frame.worktree)).toEqual(heldWorktrees) + expect(otherFrames.every((frame) => frame.clientHostedPagesUnreconciled === undefined)).toBe( + true + ) + expect(frames).toEqual([]) + expect(refresh).not.toHaveBeenCalled() + otherFrames.length = 0 + runtime.projectedFrames = 0 + runtime.markClientHostedPagesReconciled('device-b') + expect(runtime.projectedFrames).toBe(0) + expect(otherFrames).toEqual([]) + expect(refresh).not.toHaveBeenCalled() + } finally { + refresh.mockRestore() + unsubscribe() + unsubscribeHeld() + await h.close() + } + } +) + +it('does not owe delivery for speculative projections and forgets removed workspaces', async () => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const worktree = 'repo-1::/tmp/worktree-observed' + const frames: RuntimeMobileSessionTabsResult[] = [] + const unsubscribe = runtime.onMobileSessionTabsChanged( + (frame) => frames.push(frame), + 'device-a', + 'mobile' + ) + const unsubscribeOther = runtime.onMobileSessionTabsChanged( + (frame) => frames.push(frame), + 'device-b', + 'mobile' + ) + try { + expect(runtime.peekSnapshotForClient(worktree, 'device-a').clientHostedPagesUnreconciled).toBe( + true + ) + runtime.markClientHostedPagesReconciled('device-a') + expect(frames).toEqual([]) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + expect(runtime.snapshotForClient(worktree, 'device-b').clientHostedPagesUnreconciled).toBe(true) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: [] }) + frames.length = 0 + runtime.markClientHostedPagesReconciled('device-b') + expect(frames).toEqual([]) + } finally { + unsubscribe() + unsubscribeOther() + await h.close() + } +}) + +it('keeps each subscription last held frame independently of later read replies', async () => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const worktree = 'repo-1::/tmp/worktree-observed' + const frames: RuntimeMobileSessionTabsResult[] = [] + const unsubscribe = runtime.onMobileSessionTabsChanged( + (frame) => frames.push(frame), + 'device-a', + 'mobile' + ) + try { + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + // A subscription flush is a production delivery boundary. + runtime.onMobileSessionTabsChanged(() => {}, 'other-device', 'mobile')() + expect(frames.some((frame) => frame.clientHostedPagesUnreconciled === true)).toBe(true) + frames.length = 0 + const now = Date.now() + vi.spyOn(Date, 'now').mockReturnValue(now + 45_001) + expect( + runtime.snapshotForClient(worktree, 'device-a').clientHostedPagesUnreconciled + ).toBeUndefined() + runtime.markClientHostedPagesReconciled('device-a') + expect(frames.map((frame) => [frame.worktree, frame.clientHostedPagesUnreconciled])).toEqual([ + [worktree, undefined] + ]) + frames.length = 0 + runtime.markClientHostedPagesReconciled('device-a') + expect(frames).toEqual([]) + } finally { + vi.restoreAllMocks() + unsubscribe() + await h.close() + } +}) diff --git a/src/main/runtime/browser/client-pages/client-page-recovery-stream-observation.test.ts b/src/main/runtime/browser/client-pages/client-page-recovery-stream-observation.test.ts new file mode 100644 index 00000000000..97208ff8302 --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-recovery-stream-observation.test.ts @@ -0,0 +1,275 @@ +import { RpcDispatcher } from '../../rpc/dispatcher' +import { SESSION_TAB_METHODS } from '../../rpc/methods/session-tabs/session-tabs' +import { expect, it, vi } from 'vitest' +import { createReconciliationDeliveryRuntime } from './client-page-reconciliation-delivery-test-harness' +import { openRecoveryTabsStream } from './client-page-recovery-stream-test-harness' + +const WORKSPACE = 'repo-1::/tmp/worktree-a' + +vi.mock('../../../../shared/sidebar-inputs-switch', () => ({ ORCA_SIDEBAR_INPUTS_IS_MAIN: true })) + +it.each([750, 7500])( + 'owes no completion for %i workspace updates filtered by a real RPC stream', + async (count) => { + vi.useFakeTimers({ toFake: ['Date'] }) + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const snapshots = [ + WORKSPACE, + ...Array.from({ length: count }, (_, index) => `repo-1::/tmp/filtered-${index}`) + ].map((worktree) => ({ + worktree, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + })) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: snapshots }) + const stream = openRecoveryTabsStream(runtime, 'filtered-stream', `id:${WORKSPACE}`) + const facts = runtime.sidebarFacts! + const refresh = vi.spyOn(facts.workspaceActivityFacts, 'refreshWorkspace') + try { + await stream.initial + for (const snapshot of snapshots.slice(1)) { + runtime.notifyMobileSessionTabsChanged(snapshot.worktree) + } + runtime.onMobileSessionTabsChanged(() => {}, 'flush-device', 'mobile')() + expect(stream.frames.every((frame) => frame.worktree === WORKSPACE)).toBe(true) + vi.setSystemTime(Date.now() + 46_000) + runtime.notifyMobileSessionTabsChanged(WORKSPACE) + runtime.onMobileSessionTabsChanged(() => {}, 'flush-device', 'mobile')() + expect(stream.frames.at(-1)?.clientHostedPagesUnreconciled).toBeUndefined() + runtime.projectedFrames = 0 + refresh.mockClear() + const sequence = runtime.sessionTabsSequence() + const delivered = stream.frames.length + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(0) + expect(runtime.sessionTabsSequence()).toBe(sequence) + expect(stream.frames).toHaveLength(delivered) + expect(refresh).not.toHaveBeenCalled() + } finally { + stream.stop() + await h.close() + refresh.mockRestore() + vi.useRealTimers() + } + } +) + +it.each(['workspace', 'all'])( + 'keeps an independent held read after %s stream progression, then forgets disposed streams and removed workspaces', + async (kind) => { + vi.useFakeTimers({ toFake: ['Date'] }) + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree: WORKSPACE, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + const stream = openRecoveryTabsStream( + runtime, + 'progressing-stream', + kind === 'workspace' ? `id:${WORKSPACE}` : undefined + ) + const independent = openRecoveryTabsStream(runtime, 'independent-stream') + try { + await Promise.all([stream.initial, independent.initial]) + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const reads: string[] = [] + await rpc.dispatchStreaming( + { id: 'independent-read', authToken: 'test', method: 'session.tabs.listAll', params: {} }, + (reply) => reads.push(reply), + { connectionId: 'independent-read', pairedDeviceId: 'device-a', clientKind: 'mobile' } + ) + expect(JSON.parse(reads[0]!).result.snapshots[0].clientHostedPagesUnreconciled).toBe(true) + independent.stop() + await independent.completion + vi.setSystemTime(Date.now() + 46_000) + runtime.notifyMobileSessionTabsChanged(WORKSPACE) + runtime.onMobileSessionTabsChanged(() => {}, 'flush-device', 'mobile')() + expect(stream.frames.at(-1)?.clientHostedPagesUnreconciled).toBeUndefined() + runtime.projectedFrames = 0 + const delivered = stream.frames.length + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(1) + expect(stream.frames).toHaveLength(delivered + 1) + expect(stream.frames.at(-1)?.clientHostedPagesUnreconciled).toBeUndefined() + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(1) + expect(stream.frames).toHaveLength(delivered + 1) + stream.stop() + await stream.completion + vi.setSystemTime(Date.now() - 46_000) + const removed = openRecoveryTabsStream(runtime, 'removed-stream', undefined, 'device-b') + await removed.initial + expect( + removed.frames.findLast((frame) => frame.worktree === WORKSPACE) + ?.clientHostedPagesUnreconciled + ).toBe(true) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: [] }) + runtime.projectedFrames = 0 + runtime.markClientHostedPagesReconciled('device-b') + expect(runtime.projectedFrames).toBe(0) + removed.stop() + await removed.completion + } finally { + stream.stop() + independent.stop() + await h.close() + vi.useRealTimers() + } + } +) + +it('cannot recreate a raw observation when its listener unsubscribes during delivery', async () => { + const h = createReconciliationDeliveryRuntime() + let unsubscribe = () => {} + try { + unsubscribe = h.hostRuntime.onMobileSessionTabsChanged( + () => unsubscribe(), + 'device-a', + 'mobile' + ) + h.hostRuntime.notifyMobileSessionTabsChanged(WORKSPACE) + h.hostRuntime.onMobileSessionTabsChanged(() => {}, 'flush-device', 'mobile')() + h.hostRuntime.projectedFrames = 0 + h.hostRuntime.markClientHostedPagesReconciled('device-a') + expect(h.hostRuntime.projectedFrames).toBe(0) + } finally { + unsubscribe() + await h.close() + } +}) + +it.each(['session.tabs.subscribe', 'session.tabs.subscribeAll'])( + 'cannot recreate %s observations when initial output releases the stream', + async (method) => { + const h = createReconciliationDeliveryRuntime() + h.hostRuntime.attachWindow(1) + h.hostRuntime.syncWindowGraph(1, { + tabs: [], + leaves: [], + mobileSessionTabs: [ + { + worktree: WORKSPACE, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + } + ] + }) + const stopped = new AbortController() + const rpc = new RpcDispatcher({ runtime: h.hostRuntime, methods: SESSION_TAB_METHODS }) + try { + let supplied = false + await rpc.dispatchStreaming( + { + id: 'release-during-output', + authToken: 'test', + method, + params: method === 'session.tabs.subscribe' ? { worktree: `id:${WORKSPACE}` } : {} + }, + (reply) => { + if (['snapshot', 'snapshots'].includes(JSON.parse(reply).result?.type)) { + supplied = true + stopped.abort() + } + }, + { + connectionId: 'release-during-output', + pairedDeviceId: 'device-a', + clientKind: 'mobile', + signal: stopped.signal + } + ) + expect(supplied).toBe(true) + h.hostRuntime.projectedFrames = 0 + h.hostRuntime.markClientHostedPagesReconciled('device-a') + expect(h.hostRuntime.projectedFrames).toBe(0) + } finally { + stopped.abort() + await h.close() + } + } +) + +it.each([750, 7500])( + 'does no completion projection for %i held read replies after that device unsubscribes', + async (count) => { + const h = createReconciliationDeliveryRuntime() + const runtime = h.hostRuntime + const snapshots = Array.from({ length: count }, (_, index) => ({ + worktree: `repo-1::/tmp/unsubscribed-${index}`, + publicationEpoch: 'renderer', + snapshotVersion: 1, + activeGroupId: null, + activeTabId: null, + activeTabType: null, + tabs: [] + })) + runtime.attachWindow(1) + runtime.syncWindowGraph(1, { tabs: [], leaves: [], mobileSessionTabs: snapshots }) + const departed = openRecoveryTabsStream(runtime, 'departing-device-stream') + const other = openRecoveryTabsStream( + runtime, + 'remaining-device-stream', + `id:${snapshots[0]!.worktree}`, + 'device-b' + ) + const refresh = vi.spyOn(runtime.sidebarFacts!.workspaceActivityFacts, 'refreshWorkspace') + try { + await Promise.all([departed.initial, other.initial]) + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const reads: string[] = [] + await rpc.dispatchStreaming( + { id: 'held-read', authToken: 'test', method: 'session.tabs.listAll', params: {} }, + (reply) => reads.push(reply), + { connectionId: 'held-read', pairedDeviceId: 'device-a', clientKind: 'runtime' } + ) + const supplied = JSON.parse(reads[0]!).result.snapshots + expect(supplied).toHaveLength(count) + expect( + supplied.every( + (frame: { clientHostedPagesUnreconciled?: true }) => + frame.clientHostedPagesUnreconciled === true + ) + ).toBe(true) + departed.stop() + await departed.completion + runtime.projectedFrames = 0 + refresh.mockClear() + const sequence = runtime.sessionTabsSequence() + const delivered = other.frames.length + runtime.markClientHostedPagesReconciled('device-a') + expect(runtime.projectedFrames).toBe(0) + expect(runtime.sessionTabsSequence()).toBe(sequence) + expect(other.frames).toHaveLength(delivered) + expect(refresh).not.toHaveBeenCalled() + } finally { + departed.stop() + other.stop() + await Promise.all([departed.completion, other.completion]) + await h.close() + refresh.mockRestore() + } + } +) diff --git a/src/main/runtime/browser/client-pages/client-page-recovery-stream-test-harness.ts b/src/main/runtime/browser/client-pages/client-page-recovery-stream-test-harness.ts new file mode 100644 index 00000000000..1ad26127583 --- /dev/null +++ b/src/main/runtime/browser/client-pages/client-page-recovery-stream-test-harness.ts @@ -0,0 +1,49 @@ +import type { RuntimeMobileSessionTabsResult } from '../../../../shared/runtime-session-contracts' +import type { OrcaRuntimeService } from '../../orca-runtime' +import { RpcDispatcher } from '../../rpc/dispatcher' +import { SESSION_TAB_METHODS } from '../../rpc/methods/session-tabs/session-tabs' + +export function openRecoveryTabsStream( + runtime: OrcaRuntimeService, + id: string, + worktree?: string, + deviceId = 'device-a', + clientKind: 'mobile' | 'runtime' = 'mobile' +) { + const stopped = new AbortController() + const frames: RuntimeMobileSessionTabsResult[] = [] + const errors: unknown[] = [] + const initial = Promise.withResolvers() + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const completion = rpc.dispatchStreaming( + { + id, + authToken: 'test', + method: worktree ? 'session.tabs.subscribe' : 'session.tabs.subscribeAll', + params: worktree ? { worktree } : {} + }, + (reply) => { + const message: { + ok?: boolean + error?: unknown + result?: + | { type: 'snapshots'; snapshots: RuntimeMobileSessionTabsResult[] } + | ({ type: 'snapshot' | 'updated' } & RuntimeMobileSessionTabsResult) + } = JSON.parse(reply) + if (message.ok === false) { + errors.push(message.error) + initial.reject(new Error('session-stream-failed')) + } + const result = message.result + if (result?.type === 'snapshots') { + frames.push(...result.snapshots) + initial.resolve() + } else if (result?.type === 'snapshot' || result?.type === 'updated') { + frames.push(result) + initial.resolve() + } + }, + { connectionId: id, pairedDeviceId: deviceId, clientKind, signal: stopped.signal } + ) + return { frames, errors, initial: initial.promise, completion, stop: () => stopped.abort() } +} diff --git a/src/main/runtime/construction/runtime-identity-state.ts b/src/main/runtime/construction/runtime-identity-state.ts index cd404f04b8d..ea4a9ce5aa7 100644 --- a/src/main/runtime/construction/runtime-identity-state.ts +++ b/src/main/runtime/construction/runtime-identity-state.ts @@ -71,6 +71,20 @@ import type { TerminalHandleAccess } from '../agent-senders/runtime-agent-sender import { RuntimeConstructionServices } from './runtime-construction-services' +export type ClientHostedPageObservation = MobileSessionTabSubscription | symbol + +export type MobileSessionTabSubscription = { + listener: ( + snapshot: RuntimeMobileSessionTabsResult, + changeSequence: number, + delivery?: 'reconciliation' + ) => void + clientNavigationId?: string + clientKind?: 'mobile' | 'runtime' + observation?: 'caller' + acceptsWorktree?: (worktree: string) => boolean +} + export abstract class OrcaRuntimeWithRuntimeId extends RuntimeConstructionServices { protected getLeafKey(tabId: string, leafId: string): string { return `${tabId}::${leafId}` @@ -217,11 +231,7 @@ export abstract class OrcaRuntimeWithRuntimeId extends RuntimeConstructionServic } >() - protected mobileSessionTabListeners = new Set<{ - listener: (snapshot: RuntimeMobileSessionTabsResult, changeSequence: number) => void - clientNavigationId?: string - clientKind?: 'mobile' | 'runtime' - }>() + protected mobileSessionTabListeners = new Set() protected pendingMobileSessionTabsChangeSequenceByWorktree = new Map() diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs-client-projection.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs-client-projection.ts new file mode 100644 index 00000000000..ff38de8ad47 --- /dev/null +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs-client-projection.ts @@ -0,0 +1,24 @@ +import type { RuntimeMobileSessionTabsResult } from '../../../../../shared/runtime-session-contracts' +import { projectSessionTabAgentStatus } from './session-tab-agent-status-projection' +import { projectSessionTabBrowserPlacements } from './session-tab-browser-placement-projection' +import { projectSessionTabFileHosts } from './session-tab-file-host-projection' + +export function projectSessionTabsForClient( + snapshot: RuntimeMobileSessionTabsResult, + clientKind: 'mobile' | 'runtime' | undefined, + clientCapabilities: Parameters[2] +): RuntimeMobileSessionTabsResult { + // Phones draw no agent message cards, so they never pay for sender identity. + const { agentSenderAssociations: _desktopOnly, ...phone } = snapshot + return projectSessionTabBrowserPlacements( + projectSessionTabFileHosts( + projectSessionTabAgentStatus( + clientKind === 'mobile' ? phone : snapshot, + clientKind, + clientCapabilities + ), + clientKind + ), + clientCapabilities + ) +} diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-census-race.test.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-census-race.test.ts index 375be7af149..901868c9e57 100644 --- a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-census-race.test.ts +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-census-race.test.ts @@ -118,8 +118,19 @@ function deferredPtyInventory(): { } } +class CensusRuntime extends OrcaRuntimeService { + projectedFrames = 0 + protected override projectMobileSessionTabsForClient( + snapshot: RuntimeMobileSessionTabsResult, + clientId?: string + ) { + this.projectedFrames += 1 + return super.projectMobileSessionTabsForClient(snapshot, clientId) + } +} + function createRuntimeHarness(initialSnapshots: RuntimeMobileSessionTabsSnapshot[] = []) { - const runtime = new OrcaRuntimeService() + const runtime = new CensusRuntime() runtime.syncWindowGraph(0, { tabs: [], leaves: [], mobileSessionTabs: initialSnapshots }) const census = deferredPtyInventory() const internals = runtime as unknown as RuntimeInventoryInternals @@ -648,3 +659,45 @@ describe('session tabs inventory census boundary', () => { expect(emit).not.toHaveBeenCalled() }) }) + +it.each(['subsumed', 'invalidated', 'aborted'] as const)( + 'does not owe completion for %s held census candidates never emitted', + async (caseName) => { + vi.useFakeTimers({ toFake: ['Date'] }) + const h = createRuntimeHarness() + try { + const created = runtimeSnapshot('wt-census-race', 1) + h.publish([created]) + if (caseName === 'invalidated') { + for (let index = 0; index < 300; index += 1) { + h.publish(index % 2 ? [created] : []) + } + } + if (caseName === 'aborted') { + h.runtime.cleanupSubscription( + 'session.tabs:conn-runtime-census-race:*:req-runtime-census-race' + ) + } + vi.setSystemTime(Date.now() + 46_000) + h.census.resolve(completePtyInventory()) + if (caseName === 'aborted') { + await expect(h.pending).rejects.toThrow('client_disconnected') + expect(h.emit).not.toHaveBeenCalled() + } else { + await h.pending + expect(h.emit.mock.calls).toHaveLength(1) + expect(h.emit.mock.calls[0]?.[0]).toMatchObject({ type: 'snapshots' }) + } + h.runtime.projectedFrames = 0 + const outputs = h.emit.mock.calls.length + h.runtime.markClientHostedPagesReconciled('paired-runtime-census-race') + expect(h.runtime.projectedFrames).toBe(0) + expect(h.emit.mock.calls).toHaveLength(outputs) + } finally { + h.runtime.cleanupSubscription( + 'session.tabs:conn-runtime-census-race:*:req-runtime-census-race' + ) + vi.useRealTimers() + } + } +) diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-collection.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-collection.ts index baddcc5d1bd..895f8f44558 100644 --- a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-collection.ts +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory-collection.ts @@ -1,3 +1,5 @@ +import { SESSION_TABS_AUTHORITATIVE_INVENTORY_RUNTIME_CAPABILITY } from '../../../../../shared/session-tabs-runtime-capabilities' +import { projectSessionTabsForClient } from './session-tabs-client-projection' import type { RuntimeMobileSessionTabsResult } from '../../../../../shared/runtime-session-contracts' import type { RpcContext } from '../../core' @@ -34,3 +36,34 @@ export async function collectSessionTabsInventory( changeSequence: collected.changeSequence } } + +export function clientUnderstandsAuthoritativeInventory(context: RpcContext): boolean { + return ( + context.clientCapabilities?.includes( + SESSION_TABS_AUTHORITATIVE_INVENTORY_RUNTIME_CAPABILITY + ) === true + ) +} + +function projectInventory( + inventory: SessionTabsInventory, + context: RpcContext +): SessionTabsInventory { + return { + snapshots: inventory.snapshots.map((snapshot) => + projectSessionTabsForClient(snapshot, context.clientKind, context.clientCapabilities) + ), + ...(inventory.authoritative && clientUnderstandsAuthoritativeInventory(context) + ? { authoritative: true as const } + : {}) + } +} + +export async function listSessionTabsInventory(context: RpcContext): Promise { + const result = projectInventory((await collectSessionTabsInventory(context)).inventory, context) + if (context.signal?.aborted) { + throw new Error('client_disconnected') + } + context.runtime.observeClientHostedPageSnapshots?.(result.snapshots, context.pairedDeviceId) + return result +} diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory.ts index 7fd80f1dce5..b095b026b07 100644 --- a/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory.ts +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs-inventory.ts @@ -1,15 +1,14 @@ +import { projectSessionTabsForClient } from './session-tabs-client-projection' +export { projectSessionTabsForClient } from './session-tabs-client-projection' import { collectSessionTabsInventory, - type SessionTabsInventory + clientUnderstandsAuthoritativeInventory } from './session-tabs-inventory-collection' +export { listSessionTabsInventory } from './session-tabs-inventory-collection' import { isDeepStrictEqual } from 'node:util' import '../../../../../shared/protocol-version' -import { SESSION_TABS_AUTHORITATIVE_INVENTORY_RUNTIME_CAPABILITY } from '../../../../../shared/session-tabs-runtime-capabilities' import type { RuntimeMobileSessionTabsResult } from '../../../../../shared/runtime-session-contracts' import type { RpcContext } from '../../core' -import { projectSessionTabAgentStatus } from './session-tab-agent-status-projection' -import { projectSessionTabBrowserPlacements } from './session-tab-browser-placement-projection' -import { projectSessionTabFileHosts } from './session-tab-file-host-projection' import { createSessionTabsStreamDelta } from './session-tabs-sender-delta' import { prepareStructuredTabsIfSupported } from './structured-session-tab-restore' import { createSessionTabsTabDeltaEncoder } from '../../../../../shared/session-tabs-tab-delta' @@ -19,52 +18,6 @@ type SessionTabsChange = RuntimeMobileSessionTabsResult & { removed?: true } const MAX_BUFFERED_CENSUS_CHANGES = 256 const MAX_CENSUS_COLLECTION_ATTEMPTS = 3 -function clientUnderstandsAuthoritativeInventory(context: RpcContext): boolean { - return ( - context.clientCapabilities?.includes( - SESSION_TABS_AUTHORITATIVE_INVENTORY_RUNTIME_CAPABILITY - ) === true - ) -} - -export function projectSessionTabsForClient( - snapshot: RuntimeMobileSessionTabsResult, - clientKind: 'mobile' | 'runtime' | undefined, - clientCapabilities: Parameters[2] -): RuntimeMobileSessionTabsResult { - // Phones draw no agent message cards, so they never pay for sender identity. - const { agentSenderAssociations: _desktopOnly, ...phone } = snapshot - return projectSessionTabBrowserPlacements( - projectSessionTabFileHosts( - projectSessionTabAgentStatus( - clientKind === 'mobile' ? phone : snapshot, - clientKind, - clientCapabilities - ), - clientKind - ), - clientCapabilities - ) -} - -function projectInventory( - inventory: SessionTabsInventory, - context: RpcContext -): SessionTabsInventory { - return { - snapshots: inventory.snapshots.map((snapshot) => - projectSessionTabsForClient(snapshot, context.clientKind, context.clientCapabilities) - ), - ...(inventory.authoritative && clientUnderstandsAuthoritativeInventory(context) - ? { authoritative: true as const } - : {}) - } -} - -export async function listSessionTabsInventory(context: RpcContext): Promise { - return projectInventory((await collectSessionTabsInventory(context)).inventory, context) -} - export async function subscribeSessionTabsInventory( context: RpcContext, emit: (result: unknown) => void @@ -75,6 +28,7 @@ export async function subscribeSessionTabsInventory( } const cleanupPrefix = `session.tabs:${connectionId ?? 'local'}:*` const subscriptionId = requestId ? `${cleanupPrefix}:${requestId}` : cleanupPrefix + const observation = Symbol('session-tabs-stream') const inventoryController = new AbortController() // For the stream's whole life, not just the census: a desktop unsubscribe arrives as this abort. const onTransportAbort = (): void => runtime.cleanupSubscription(subscriptionId) @@ -163,7 +117,26 @@ export async function subscribeSessionTabsInventory( changeSequence }) } - const publishChange = (snapshot: SessionTabsChange, changeSequence: number): void => { + const emitChange = (projected: SessionTabsChange): void => { + if (closed) { + return + } + emit({ type: 'updated', ...tabDelta.update(withProofDelta(projected)) }) + if (!closed) { + runtime.observeClientHostedPageSnapshots?.( + [ + projected.removed ? { ...projected, clientHostedPagesUnreconciled: undefined } : projected + ], + pairedDeviceId, + observation + ) + } + } + const publishChange = ( + snapshot: SessionTabsChange, + changeSequence: number, + delivery?: 'reconciliation' + ): void => { const { navigationIntent: _intent, ...canonical } = snapshot if (snapshot.removed) { canonicalSnapshotsByWorktree.delete(snapshot.worktree) @@ -173,14 +146,18 @@ export async function subscribeSessionTabsInventory( const projected = projectChange(snapshot) const state = withoutNavigationIntent(projected) const published = publishedSnapshotsByWorktree.get(snapshot.worktree) - if (projected.navigationIntent === undefined && isDeepStrictEqual(published, state)) { + if ( + delivery !== 'reconciliation' && + projected.navigationIntent === undefined && + isDeepStrictEqual(published, state) + ) { deliveredChangeSequenceByWorktree.set(snapshot.worktree, changeSequence) return } - emit({ - type: 'updated', - ...tabDelta.update(withProofDelta(projected)) - }) + emitChange(projected) + if (closed) { + return + } if (projected.removed === true) { publishedSnapshotsByWorktree.delete(snapshot.worktree) deliveredChangeSequenceByWorktree.delete(snapshot.worktree) @@ -190,7 +167,10 @@ export async function subscribeSessionTabsInventory( } } const unsubscribe = runtime.onMobileSessionTabsChanged( - (runtimeSnapshot, changeSequence) => { + (runtimeSnapshot, changeSequence, delivery) => { + if (closed) { + return + } const snapshot = runtimeSnapshot as SessionTabsChange if (!initialized) { bufferChange(snapshot, changeSequence) @@ -202,10 +182,11 @@ export async function subscribeSessionTabsInventory( ) { return } - publishChange(snapshot, changeSequence) + publishChange(snapshot, changeSequence, delivery) }, pairedDeviceId, - context.clientKind + context.clientKind, + { observation: 'caller', acceptsWorktree: () => !closed } ) const stopCapabilities = context.capabilitySource?.observe(() => { if (!initialized || closed) { @@ -216,17 +197,24 @@ export async function subscribeSessionTabsInventory( if (isDeepStrictEqual(publishedSnapshotsByWorktree.get(snapshot.worktree), projected)) { continue } - emit({ type: 'updated', ...tabDelta.update(withProofDelta(projected)) }) + emitChange(projected) + if (closed) { + return + } publishedSnapshotsByWorktree.set(snapshot.worktree, projected) } }) runtime.registerSubscriptionCleanup( subscriptionId, () => { + if (closed) { + return + } closed = true context.signal?.removeEventListener('abort', onTransportAbort) inventoryController.abort() unsubscribe() + runtime.forgetClientHostedPageObservation?.(pairedDeviceId, observation) stopCapabilities?.() canonicalSnapshotsByWorktree.clear() clearBufferedChanges() @@ -278,15 +266,18 @@ export async function subscribeSessionTabsInventory( } const { inventory, changeSequence } = collected censusChangeSequence = changeSequence + const projectedSnapshots = inventory.snapshots.map(projectChange) emit({ type: 'snapshots', ...(inventory.authoritative && clientUnderstandsAuthoritativeInventory(context) ? { authoritative: true as const } : {}), - snapshots: inventory.snapshots.map((snapshot) => - tabDelta.full(withProofDelta(projectChange(snapshot))) - ) + snapshots: projectedSnapshots.map((snapshot) => tabDelta.full(withProofDelta(snapshot))) }) + if (closed) { + return + } + runtime.observeClientHostedPageSnapshots?.(projectedSnapshots, pairedDeviceId, observation) for (const snapshot of inventory.snapshots) { const canonical = withoutNavigationIntent(snapshot) canonicalSnapshotsByWorktree.set(snapshot.worktree, canonical) diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs-tab-delta-stream.test.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs-tab-delta-stream.test.ts new file mode 100644 index 00000000000..54cabd9027b --- /dev/null +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs-tab-delta-stream.test.ts @@ -0,0 +1,193 @@ +import { RuntimeClientCapabilities } from '../../callers/runtime-client-capabilities' +import { RUNTIME_CLIENT_CAPABILITY_METHODS } from '../client-events/runtime-client-capabilities' +import { visibleSnapshot } from './session-tabs-snapshot.test-fixture' +import { describe, expect, it, vi } from 'vitest' +import type { OrcaRuntimeService } from '../../../orca-runtime' +import '../../../../../shared/protocol-version' +import { + SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY, + SESSION_TABS_TAB_DELTA_RUNTIME_CAPABILITY +} from '../../../../../shared/session-tabs-runtime-capabilities' +import type { RuntimeMobileSessionTabsResult } from '../../../../../shared/runtime-session-contracts' +import { createSessionTabsTabDeltaDecoder } from '../../../../../shared/session-tabs-tab-delta' +import { RpcDispatcher } from '../../dispatcher' +import { SESSION_TAB_METHODS } from './session-tabs' + +describe.each(['session.tabs.subscribe', 'session.tabs.subscribeAll'] as const)( + '%s capability recovery', + (method) => { + it.each([false, true])( + 'reprojects on the same stream with opening delta support=%s', + async (delta) => { + const source = new RuntimeClientCapabilities() + source.replace( + delta + ? [ + SESSION_TABS_TAB_DELTA_RUNTIME_CAPABILITY, + SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY + ] + : [] + ) + const changed = vi.fn<(value: RuntimeMobileSessionTabsResult, sequence: number) => void>() + let listener: typeof changed = changed + const cleanups = new Map void>() + const initial: RuntimeMobileSessionTabsResult = { + ...visibleSnapshot(), + retiredTerminalSurfaces: [ + { + parentTabId: 'retired', + leafId: 'leaf', + ptyId: 'pty', + terminal: 'term_1', + incarnationId: 'inc' + } + ], + tabs: [ + { + type: 'agent-session', + id: 'claude', + sessionId: 'claude-session', + agent: 'claude', + title: 'Claude chat', + isActive: true + }, + { + type: 'agent-session', + id: 'codex', + sessionId: 'codex-session', + agent: 'codex', + title: 'Codex chat', + isActive: false + } + ] + } + const observedFrames: RuntimeMobileSessionTabsResult[][] = [] + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: these real RPC handlers reach only the runtime members stubbed here. + const runtime = { + observeClientHostedPageSnapshots: (frames: RuntimeMobileSessionTabsResult[]) => + observedFrames.push(frames), + getRuntimeId: () => 'runtime', + getClientSettings: () => ({}), + listMobileSessionTabs: async () => initial, + supportsAuthoritativeSessionTabsInventory: () => false, + listAllMobileSessionTabsWithChangeSequence: async () => ({ + snapshots: [initial], + changeSequence: 1 + }), + registerSubscriptionCleanup: (id: string, cleanup: () => void) => + cleanups.set(id, cleanup), + cleanupSubscription: (id: string) => cleanups.get(id)?.(), + onMobileSessionTabsChanged: (callback: typeof listener) => { + listener = callback + return () => {} + } + } as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ + runtime, + methods: [...SESSION_TAB_METHODS, ...RUNTIME_CLIENT_CAPABILITY_METHODS] + }) + const messages: string[] = [] + const options = { + clientKind: 'mobile' as const, + connectionId: 'phone', + clientCapabilities: source.current, + capabilitySource: source, + updateClientCapabilities: (next: readonly string[]) => source.replace(next) + } + await dispatcher.dispatchStreaming( + { + id: 'stream', + authToken: 'tok', + method, + params: method === 'session.tabs.subscribe' ? { worktree: 'id:wt-1' } : undefined + }, + (raw) => messages.push(raw), + options + ) + const primaryListener = listener + const siblingSource = new RuntimeClientCapabilities() + const siblingMessages: string[] = [] + await dispatcher.dispatchStreaming( + { + id: 'sibling', + authToken: 'tok', + method, + params: method === 'session.tabs.subscribe' ? { worktree: 'id:wt-1' } : undefined + }, + (raw) => siblingMessages.push(raw), + { ...options, connectionId: 'other-phone', capabilitySource: siblingSource } + ) + const decoder = createSessionTabsTabDeltaDecoder() + const decode = (raw: string) => { + const result = decoder.decode(JSON.parse(raw).result) + if (!result.ok) { + throw new Error('invalid stream baseline') + } + return result.result + } + expect(JSON.stringify(decode(messages[0]!))).toContain('Open on desktop') + expect(messages).toHaveLength(1) + const update = async (caps: string[]) => { + await dispatcher.dispatchStreaming( + { + id: 'upgrade', + authToken: 'tok', + method: 'runtime.clientCapabilities.update', + params: { clientCapabilities: caps } + }, + () => {}, + options + ) + } + const structured = [ + 'agent-session.structured.v1', + 'agent-session.structured.claude.v1', + SESSION_TABS_TAB_DELTA_RUNTIME_CAPABILITY, + SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY + ] + // A capability the projection ignores sends no frame; the next real change still decodes. + await update([...source.current, 'unrelated.test.v1']) + expect(messages).toHaveLength(1) + expect(observedFrames).toHaveLength(2) + await update(structured) + expect(siblingMessages).toHaveLength(1) + expect(messages).toHaveLength(2) + expect(observedFrames).toHaveLength(3) + expect(observedFrames.at(-1)?.[0]?.tabs[0]).toMatchObject({ + type: 'agent-session', + title: 'Claude chat' + }) + const recovered = decode(messages[1]!) + expect(JSON.stringify(recovered)).toContain('Claude chat') + const proofState = { retiredTerminalSurfaces: delta ? [] : initial.retiredTerminalSurfaces } + expect(recovered).toMatchObject(proofState) + if (!delta) { + expect(JSON.parse(messages[1]!).result).not.toHaveProperty('tabsRevision') + } + const renamed = { + ...initial, + snapshotVersion: 2, + tabs: initial.tabs.map((tab) => ({ ...tab, title: `${tab.title} renamed` })) + } + primaryListener(renamed, 2) + expect(JSON.stringify(decode(messages[2]!))).toContain('Claude chat renamed') + primaryListener(renamed, 3) + if (method === 'session.tabs.subscribeAll') { + expect(messages).toHaveLength(3) + } else { + decode(messages[3]!) + } + const beforeLoss = messages.length + await update([]) + expect(JSON.stringify(decode(messages[beforeLoss]!))).toContain('Open on desktop') + expect(siblingMessages).toHaveLength(1) + const count = messages.length + for (const cleanup of cleanups.values()) { + cleanup() + } + await update(['agent-session.structured.v1']) + expect(messages.slice(count).map((raw) => JSON.parse(raw).result.type)).toEqual(['end']) + } + ) + } +) diff --git a/src/main/runtime/rpc/methods/session-tabs/session-tabs.ts b/src/main/runtime/rpc/methods/session-tabs/session-tabs.ts index 4143840057c..4da2eca8a52 100644 --- a/src/main/runtime/rpc/methods/session-tabs/session-tabs.ts +++ b/src/main/runtime/rpc/methods/session-tabs/session-tabs.ts @@ -36,7 +36,7 @@ export const SESSION_TAB_METHODS = [ params: WorktreeTabSelector, handler: async (params, { runtime, pairedDeviceId, clientKind, clientCapabilities }) => { await prepareStructuredTabsIfSupported({ runtime, clientKind, clientCapabilities }) - return projectSessionTabsForClient( + const result = projectSessionTabsForClient( await runtime.listMobileSessionTabs(params.worktree, pairedDeviceId, { includeInitialSurfaceLaunch: true, includeAgentSenders: clientKind !== 'mobile' @@ -44,6 +44,8 @@ export const SESSION_TAB_METHODS = [ clientKind, clientCapabilities ) + runtime.observeClientHostedPageSnapshots?.([result], pairedDeviceId) + return result } }), defineMethod({ @@ -129,6 +131,7 @@ export const SESSION_TAB_METHODS = [ }, emit ) => { + const observation = Symbol('session-tabs-stream') let subscriptionId: string | null = null let released = false let endOnRelease = true @@ -164,6 +167,7 @@ export const SESSION_TAB_METHODS = [ closePhoneSession?.() signal?.removeEventListener('abort', release) stopListening() + runtime.forgetClientHostedPageObservation?.(pairedDeviceId, observation) stopCapabilities() if (endOnRelease) { emit({ type: 'end' }) @@ -184,13 +188,30 @@ export const SESSION_TAB_METHODS = [ snapshot: NonNullable, capabilities = capabilitySource?.current ?? clientCapabilities ) => projectSessionTabsForClient(snapshot, clientKind, capabilities) + const emitSnapshot = ( + projected: NonNullable, + type: 'snapshot' | 'updated' + ): void => { + if (released) { + return + } + emit({ + type, + ...(type === 'snapshot' + ? tabDelta.full(withProofDelta(projected)) + : tabDelta.update(withProofDelta(projected))) + }) + if (!released) { + runtime.observeClientHostedPageSnapshots?.([projected], pairedDeviceId, observation) + } + } stopCapabilities = capabilitySource?.observe((previous) => { if (!released && currentSnapshot) { const projected = project(currentSnapshot) // The last frame was projected with the previous set; a change it ignores sends nothing. if (!isDeepStrictEqual(projected, project(currentSnapshot, previous))) { - emit({ type: 'updated', ...tabDelta.update(withProofDelta(projected)) }) + emitSnapshot(projected, 'updated') } } }) ?? (() => {}) @@ -223,45 +244,44 @@ export const SESSION_TAB_METHODS = [ } let subscribedWorktree = initial.worktree currentSnapshot = initial - emit({ - type: 'snapshot', - ...tabDelta.full(withProofDelta(project(initial))) - }) + emitSnapshot(project(initial), 'snapshot') if (released) { return } + const acceptsWorktree = (worktree: string): boolean => { + if (released) { + return false + } + if (worktree !== subscribedWorktree && worktreeIdsEqual(worktree, subscribedWorktree)) { + // Subscribe can precede publication of the host's stored spelling. + try { + subscribedWorktree = + runtime.getExplicitSessionWorktreeIdSelector(`id:${initial.worktree}`) ?? + subscribedWorktree + } catch { + // An ambiguous alias cannot authorize switching to either session. + return false + } + } + return worktree === subscribedWorktree + } stopListening = runtime.onMobileSessionTabsChanged( (snapshot) => { - if ( - snapshot.worktree !== subscribedWorktree && - worktreeIdsEqual(snapshot.worktree, subscribedWorktree) - ) { - // Subscribe can precede publication of the host's stored spelling. - try { - subscribedWorktree = - runtime.getExplicitSessionWorktreeIdSelector(`id:${initial.worktree}`) ?? - subscribedWorktree - } catch { - // An ambiguous alias cannot authorize switching to either session. - return - } - } - if (snapshot.worktree === subscribedWorktree) { - // Keep one stream spelling because clients index frames by worktree. - const streamSnapshot = - snapshot.worktree === initial.worktree - ? snapshot - : { ...snapshot, worktree: initial.worktree } - const { navigationIntent: _intent, ...retained } = streamSnapshot - currentSnapshot = retained - emit({ - type: 'updated', - ...tabDelta.update(withProofDelta(project(streamSnapshot))) - }) + if (released) { + return } + // Keep one stream spelling because clients index frames by worktree. + const streamSnapshot = + snapshot.worktree === initial.worktree + ? snapshot + : { ...snapshot, worktree: initial.worktree } + const { navigationIntent: _intent, ...retained } = streamSnapshot + currentSnapshot = retained + emitSnapshot(project(streamSnapshot), 'updated') }, pairedDeviceId, - clientKind + clientKind, + { observation: 'caller', acceptsWorktree } ) } catch (error) { // A stream already ended by its release must not also report an error. diff --git a/src/main/runtime/session-graph/session-graph-authority.ts b/src/main/runtime/session-graph/session-graph-authority.ts index ea078c04082..3aaf7892048 100644 --- a/src/main/runtime/session-graph/session-graph-authority.ts +++ b/src/main/runtime/session-graph/session-graph-authority.ts @@ -252,6 +252,29 @@ export abstract class OrcaRuntimeWithTransitionGraphReloadToTerminalState extend /** Reconcile client-hosted pages per paired device; one attached client cannot account for another's pages. */ markClientHostedPagesReconciled(pairedDeviceId: string): void { - this.clientHostedPageReconciliation.markReconciled(pairedDeviceId) + const held = this.clientHostedPageReconciliation.markReconciled(pairedDeviceId) + if (!held?.size) { + return + } + const candidates = this.selectMobileSessionTabsListeners(undefined, pairedDeviceId) + if (candidates.length === 0) { + return + } + const deliver = this.createMobileSessionTabsDelivery() + for (const worktree of held) { + const selected = this.selectMobileSessionTabsListeners(worktree, pairedDeviceId, candidates) + if (selected.length === 0) { + continue + } + deliver( + this.getMobileSessionTabsForWorktree(worktree, pairedDeviceId, { + includeAgentSenders: false + }), + ++this.mobileSessionTabsChangeSequence, + pairedDeviceId, + selected, + 'reconciliation' + ) + } } } diff --git a/src/main/runtime/session-tabs/mobile-session-membership.ts b/src/main/runtime/session-tabs/mobile-session-membership.ts index d9fe6ec16ac..881d9d14ff4 100644 --- a/src/main/runtime/session-tabs/mobile-session-membership.ts +++ b/src/main/runtime/session-tabs/mobile-session-membership.ts @@ -1,3 +1,4 @@ +import type { MobileSessionTabSubscription } from '../construction/runtime-identity-state' import type { OrcaRuntimeWithScheduleMobileSessionTabsChanged as SessionTabsPublication } from './session-tabs-change-publication' import type { OrcaRuntimeWithPruneMobileSessionTabGroupLayout as SessionTabsProjection } from '../session-lifecycle/projection/prune-mobile-session-tab-group-layout' import type { OrcaRuntimeWithGetTerminalInteractiveWait as GetTerminalInteractiveWaitMethods } from '../terminal/wait/terminal-interactive-wait' @@ -22,7 +23,11 @@ export abstract class OrcaRuntimeWithTouchMobileSessionTabsForWorktree extends O ...args: Parameters ): ReturnType - protected abstract hasAgentSenderListeners(): boolean + protected abstract selectMobileSessionTabsListeners( + worktree?: string, + clientNavigationId?: string, + candidates?: Iterable + ): MobileSessionTabSubscription[] abstract getTerminalWorktreeIdForPaneKey( ...args: Parameters @@ -35,14 +40,19 @@ export abstract class OrcaRuntimeWithTouchMobileSessionTabsForWorktree extends O /** Republish only worktrees whose sender scope differs from the one their last frame carried. */ refreshAgentSenderAssociations(): void { - if (!this.hasAgentSenderListeners()) { + const listeners = this.selectMobileSessionTabsListeners().filter( + (subscription) => subscription.clientKind !== 'mobile' + ) + if (listeners.length === 0) { return } const changed = this.agentProjections.senders.changedScopes( this.mobileSessionTabsByWorktree.keys() ) for (const worktreeId of changed) { - this.touchMobileSessionTabsForWorktree(worktreeId) + if (this.selectMobileSessionTabsListeners(worktreeId, undefined, listeners).length > 0) { + this.touchMobileSessionTabsForWorktree(worktreeId) + } } } diff --git a/src/main/runtime/session-tabs/mobile-session-publication-preservation.ts b/src/main/runtime/session-tabs/mobile-session-publication-preservation.ts index c4aa4873899..ec48166dddd 100644 --- a/src/main/runtime/session-tabs/mobile-session-publication-preservation.ts +++ b/src/main/runtime/session-tabs/mobile-session-publication-preservation.ts @@ -18,6 +18,7 @@ export abstract class OrcaRuntimeWithStoredMobileSnapshotHasStalePreservedTab ex protected abstract getListedBrowserTabsByPageId( ...args: Parameters ): ReturnType + protected abstract notifyMobileSessionTabsChangedNow( ...args: Parameters< ScheduleMobileSessionTabsChangedMethods['notifyMobileSessionTabsChangedNow'] @@ -208,12 +209,16 @@ export abstract class OrcaRuntimeWithStoredMobileSnapshotHasStalePreservedTab ex tabs: [] } const changeSequence = ++this.mobileSessionTabsChangeSequence - for (const subscription of this.mobileSessionTabListeners) { + for (const subscription of this.selectMobileSessionTabsListeners(worktreeId)) { + if (!this.mobileSessionTabListeners.has(subscription)) { + continue + } subscription.listener( this.clientSessionTabSelections.project(removed, subscription.clientNavigationId), changeSequence ) } + this.clientHostedPageReconciliation.forgetWorktree(worktreeId) this.clientSessionTabSelections.forgetWorktree(worktreeId) } diff --git a/src/main/runtime/session-tabs/mobile-session-tab-subscriptions.ts b/src/main/runtime/session-tabs/mobile-session-tab-subscriptions.ts index 8b01ffdba50..44455726f30 100644 --- a/src/main/runtime/session-tabs/mobile-session-tab-subscriptions.ts +++ b/src/main/runtime/session-tabs/mobile-session-tab-subscriptions.ts @@ -1,19 +1,31 @@ +import type { + ClientHostedPageObservation, + MobileSessionTabSubscription +} from '../construction/runtime-identity-state' import type { RuntimeMobileSessionTabsResult } from '../../../shared/runtime-session-contracts' import { RuntimeWorkspaceFileTargets } from '../workspaces/workspace-file-session-targets' export abstract class RuntimeMobileSessionTabSubscriptions extends RuntimeWorkspaceFileTargets { onMobileSessionTabsChanged( - listener: (snapshot: RuntimeMobileSessionTabsResult, changeSequence: number) => void, + listener: MobileSessionTabSubscription['listener'], clientNavigationId?: string, - clientKind?: 'mobile' | 'runtime' + clientKind?: 'mobile' | 'runtime', + options?: Pick ): () => void { // Drain coalesced notifications before subscribing: the initial snapshot already includes them. this.mobileSessionTabsNotifyCoalescer.flushAll() - const subscription = { listener, clientNavigationId, clientKind } + const subscription = { + listener, + clientNavigationId, + clientKind, + observation: options?.observation, + acceptsWorktree: options?.acceptsWorktree + } this.mobileSessionTabListeners.add(subscription) return () => { // Flush before removal so this listener receives the latest settled state. this.mobileSessionTabsNotifyCoalescer.flushAll() const removed = this.mobileSessionTabListeners.delete(subscription) + this.clientHostedPageReconciliation.forgetObservation(clientNavigationId, subscription) if (removed && clientKind !== 'mobile') { this.releaseAgentSenderHistoryIfUnobserved() } @@ -23,6 +35,60 @@ export abstract class RuntimeMobileSessionTabSubscriptions extends RuntimeWorksp } } + protected deliverMobileSessionTabsFrame( + subscription: MobileSessionTabSubscription, + frame: RuntimeMobileSessionTabsResult, + sequence: number, + delivery?: 'reconciliation' + ): void { + subscription.listener(frame, sequence, delivery) + if ( + subscription.observation === 'caller' || + !this.mobileSessionTabListeners.has(subscription) + ) { + return + } + this.clientHostedPageReconciliation.observe( + [frame], + subscription.clientNavigationId, + subscription + ) + } + + observeClientHostedPageSnapshots( + frames: readonly RuntimeMobileSessionTabsResult[], + clientNavigationId?: string, + observer?: ClientHostedPageObservation + ): void { + this.clientHostedPageReconciliation.observe(frames, clientNavigationId, observer) + } + + forgetClientHostedPageObservation( + clientNavigationId: string | undefined, + observer: ClientHostedPageObservation + ): void { + this.clientHostedPageReconciliation.forgetObservation(clientNavigationId, observer) + } + + protected selectMobileSessionTabsListeners( + worktree?: string, + clientNavigationId?: string, + candidates: Iterable = this.mobileSessionTabListeners + ): MobileSessionTabSubscription[] { + const selected: MobileSessionTabSubscription[] = [] + for (const subscription of candidates) { + if ( + this.mobileSessionTabListeners.has(subscription) && + (clientNavigationId === undefined || + subscription.clientNavigationId === clientNavigationId) && + (worktree === undefined || subscription.acceptsWorktree?.(worktree) !== false) + ) { + selected.push(subscription) + } + } + return selected + } + protected hasAgentSenderListeners(): boolean { for (const { clientKind } of this.mobileSessionTabListeners) { if (clientKind !== 'mobile') { @@ -40,5 +106,6 @@ export abstract class RuntimeMobileSessionTabSubscriptions extends RuntimeWorksp forgetClientNavigationState(clientNavigationId: string): void { this.clientSessionTabSelections.forgetClient(clientNavigationId) + this.clientHostedPageReconciliation.forgetClient(clientNavigationId) } } diff --git a/src/main/runtime/session-tabs/orca-runtime-session-tabs-path-alias.test.ts b/src/main/runtime/session-tabs/orca-runtime-session-tabs-path-alias.test.ts index 3dde336c70a..363ca55c4ab 100644 --- a/src/main/runtime/session-tabs/orca-runtime-session-tabs-path-alias.test.ts +++ b/src/main/runtime/session-tabs/orca-runtime-session-tabs-path-alias.test.ts @@ -158,3 +158,35 @@ describe('session tab workspace path aliases', () => { runtime.cleanupSubscriptionsForConnection('alias-connection') }) }) + +it('rejects ambiguous later aliases while retaining an exact host spelling', async () => { + const runtime = new OrcaRuntimeService() + runtime.attachWindow(1) + const rpc = new RpcDispatcher({ runtime, methods: SESSION_TAB_METHODS }) + const frames: { type?: string; worktree?: string }[] = [] + try { + await rpc.dispatchStreaming( + { + id: 'ambiguous-alias', + authToken: 'test', + method: 'session.tabs.subscribe', + params: { worktree: `id:${ALIAS}` } + }, + (reply) => frames.push(JSON.parse(reply).result), + { connectionId: 'ambiguous-alias' } + ) + publishGraph(runtime, WORKTREE, [snapshot('repo::g:/git/example', 'pty-other')]) + runtime.onMobileSessionTabsChanged(() => {}, undefined, 'mobile')() + expect(frames.filter((frame) => frame.type === 'updated')).toEqual([]) + publishGraph(runtime, ALIAS, [ + snapshot(WORKTREE), + snapshot('repo::g:/git/example', 'pty-other') + ]) + runtime.onMobileSessionTabsChanged(() => {}, undefined, 'mobile')() + expect( + frames.filter((frame) => frame.type === 'updated').map((frame) => frame.worktree) + ).toEqual([ALIAS]) + } finally { + runtime.cleanupSubscriptionsForConnection('ambiguous-alias') + } +}) diff --git a/src/main/runtime/session-tabs/session-tabs-change-publication.ts b/src/main/runtime/session-tabs/session-tabs-change-publication.ts index 95832e5f5e6..dd780267785 100644 --- a/src/main/runtime/session-tabs/session-tabs-change-publication.ts +++ b/src/main/runtime/session-tabs/session-tabs-change-publication.ts @@ -1,3 +1,4 @@ +import type { MobileSessionTabSubscription } from '../construction/runtime-identity-state' import { OrcaRuntimeWithStoredMobileSnapshotHasStalePreservedTab } from './mobile-session-publication-preservation' import type { BrowserTabInfo } from '../../../shared/browser/runtime/runtime-browser-contracts' import type { @@ -67,7 +68,9 @@ export abstract class OrcaRuntimeWithScheduleMobileSessionTabsChanged extends Or return ( result: RuntimeMobileSessionTabsResult, changeSequence: number, - clientNavigationId?: string + clientNavigationId?: string, + candidates?: readonly MobileSessionTabSubscription[], + delivery?: 'reconciliation' ): void => { const { agentSenderAssociations: _senders, ...tabsOnly } = result // A frame for one client is already its projection; projecting again stacks the epoch suffix and revision, so the client fences it. @@ -75,32 +78,31 @@ export abstract class OrcaRuntimeWithScheduleMobileSessionTabsChanged extends Or clientNavigationId === undefined ? this.projectMobileSessionTabsForClient(frame, id) : frame let desktopFrame: RuntimeMobileSessionTabsResult | undefined try { - for (const subscription of this.mobileSessionTabListeners) { - if ( - clientNavigationId !== undefined && - subscription.clientNavigationId !== clientNavigationId - ) { + for (const subscription of this.selectMobileSessionTabsListeners( + result.worktree, + clientNavigationId, + candidates + )) { + if (!this.mobileSessionTabListeners.has(subscription)) { continue } let frame: RuntimeMobileSessionTabsResult = tabsOnly if (subscription.clientKind !== 'mobile') { if (!desktopFrame) { - if (result.agentSenderAssociations) { - desktopFrame = result - } else { + desktopFrame = result + if (!result.agentSenderAssociations) { senderScope ??= this.agentProjections.senders.buildForFramePass() desktopFrame = { ...result, agentSenderAssociations: senderScope(result.worktree) } } } frame = desktopFrame } - subscription.listener(project(frame, subscription.clientNavigationId), changeSequence) + const supplied = project(frame, subscription.clientNavigationId) + this.deliverMobileSessionTabsFrame(subscription, supplied, changeSequence, delivery) } - if (clientNavigationId === undefined && desktopFrame?.agentSenderAssociations) { - this.agentProjections.senders.acknowledgeFrame( - result.worktree, - desktopFrame.agentSenderAssociations - ) + const senders = desktopFrame?.agentSenderAssociations + if (clientNavigationId === undefined && senders) { + this.agentProjections.senders.acknowledgeFrame(result.worktree, senders) } } finally { if (senderScope) {