mirror of
https://github.com/stablyai/orca.git
synced 2026-11-14 01:47:24 +00:00
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>
This commit is contained in:
Brennan Benson
Neil
+94
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string>()
|
||||
private readonly observations = new Map<string, Map<ClientHostedPageObservation, Set<string>>>()
|
||||
|
||||
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<string> | 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<ClientHostedPageObservation, Set<string>>()
|
||||
const held = observations.get(observer) ?? new Set<string>()
|
||||
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<T extends { clientHostedPagesUnreconciled?: true }>(
|
||||
holdFor<T extends { worktree: string; clientHostedPagesUnreconciled?: true }>(
|
||||
result: T,
|
||||
pairedDeviceId: string | undefined,
|
||||
now: number
|
||||
|
||||
+80
@@ -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<void> }[] = []
|
||||
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))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
})
|
||||
@@ -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<typeof openRecoveryTabsStream> | 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()
|
||||
}
|
||||
})
|
||||
@@ -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()
|
||||
}
|
||||
})
|
||||
+275
@@ -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()
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -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<void>()
|
||||
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() }
|
||||
}
|
||||
@@ -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<MobileSessionTabSubscription>()
|
||||
|
||||
protected pendingMobileSessionTabsChangeSequenceByWorktree = new Map<string, number>()
|
||||
|
||||
|
||||
@@ -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<typeof projectSessionTabAgentStatus>[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
|
||||
)
|
||||
}
|
||||
+54
-1
@@ -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()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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<SessionTabsInventory> {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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<typeof projectSessionTabAgentStatus>[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<SessionTabsInventory> {
|
||||
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)
|
||||
|
||||
@@ -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<string, () => 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'])
|
||||
}
|
||||
)
|
||||
}
|
||||
)
|
||||
@@ -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<typeof currentSnapshot>,
|
||||
capabilities = capabilitySource?.current ?? clientCapabilities
|
||||
) => projectSessionTabsForClient(snapshot, clientKind, capabilities)
|
||||
const emitSnapshot = (
|
||||
projected: NonNullable<typeof currentSnapshot>,
|
||||
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.
|
||||
|
||||
@@ -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'
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<SessionTabsProjection['toMobileSessionTabsResult']>
|
||||
): ReturnType<SessionTabsProjection['toMobileSessionTabsResult']>
|
||||
|
||||
protected abstract hasAgentSenderListeners(): boolean
|
||||
protected abstract selectMobileSessionTabsListeners(
|
||||
worktree?: string,
|
||||
clientNavigationId?: string,
|
||||
candidates?: Iterable<MobileSessionTabSubscription>
|
||||
): MobileSessionTabSubscription[]
|
||||
|
||||
abstract getTerminalWorktreeIdForPaneKey(
|
||||
...args: Parameters<GetTerminalInteractiveWaitMethods['getTerminalWorktreeIdForPaneKey']>
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ export abstract class OrcaRuntimeWithStoredMobileSnapshotHasStalePreservedTab ex
|
||||
protected abstract getListedBrowserTabsByPageId(
|
||||
...args: Parameters<ScheduleMobileSessionTabsChangedMethods['getListedBrowserTabsByPageId']>
|
||||
): ReturnType<ScheduleMobileSessionTabsChangedMethods['getListedBrowserTabsByPageId']>
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<MobileSessionTabSubscription, 'observation' | 'acceptsWorktree'>
|
||||
): () => 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<MobileSessionTabSubscription> = 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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')
|
||||
}
|
||||
})
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user