Merge remote-tracking branch 'origin/master' into dshw/pr-2250

packages/client/connection/tests/client-apply.spec.ts

@@ -87,10 +87,87 @@ describe('connection client apply', () => {
   it('start() hands out one loop, rejects a second consumer, and stop() aborts the streams', async () => {
     ;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
     const handle = await mount()
-    // config omitted: the `config ?? {}` default arm is part of the API.
-    const loop = handle.start({})
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
+    const descriptions: Array<boolean | undefined> = []
+    const stopThrowing = handle.hostDescription.subscribe(() => { throw new Error('subscriber bug') })
+    const stopDescription = handle.hostDescription.subscribe(() => {
+      descriptions.push(handle.hostDescription.getSnapshot()?.canOpenPath)
+    })
+    expect(handle.hostDescription.getSnapshot()).toBeUndefined()
+    // config omitted: the `config ?? {}` default arm is part of the surface.
+    let connected = 0
+    const loop = handle.start({ onConnected: () => { connected++ } })
     expect(() => handle.start({})).toThrow(/already owned by another consumer/)
+    await vi.waitFor(() => {
+      expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
+    })
     loop.stop() // teardown must not throw; the fixture streams abort quietly
+    expect(handle.hostDescription.getSnapshot()).toBeUndefined()
+    expect(descriptions).toEqual([true, undefined])
+    expect(connected).toBe(1)
+    expect(errorSpy).toHaveBeenCalledTimes(2)
+    stopThrowing()
+    stopDescription()
+    errorSpy.mockRestore()
+  })
+
+  it('does not announce a generation synchronously stopped by a description subscriber', async () => {
+    ;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
+    const handle = await mount()
+    const owner: { loop?: ReturnType<ConnectionHandle['start']> } = {}
+    let sawDescription = false
+    const stopDescription = handle.hostDescription.subscribe(() => {
+      if (handle.hostDescription.getSnapshot() === undefined) return
+      sawDescription = true
+      owner.loop?.stop()
+    })
+    const connected = vi.fn()
+    const loop = handle.start({ onConnected: connected })
+    owner.loop = loop
+    try {
+      await vi.waitFor(() => { expect(sawDescription).toBe(true) })
+      expect(handle.hostDescription.getSnapshot()).toBeUndefined()
+      expect(connected).not.toHaveBeenCalled()
+    } finally {
+      stopDescription()
+      loop.stop()
+    }
+  })
+
+  it('retracts the host description while reconnecting and republishes the next generation', async () => {
+    ;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
+    const handle = await mount()
+    const descriptions: Array<boolean | undefined> = []
+    const reconnectSnapshots: Array<boolean | undefined> = []
+    const stopDescription = handle.hostDescription.subscribe(() => {
+      descriptions.push(handle.hostDescription.getSnapshot()?.canOpenPath)
+    })
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const loop = handle.start({
+      onStateChange: (state) => {
+        if (state === 'reconnecting') {
+          reconnectSnapshots.push(handle.hostDescription.getSnapshot()?.canOpenPath)
+        }
+      },
+    }, { backoffBaseMs: 10, backoffFactor: 1, backoffMaxMs: 10, streamOpenTimeoutMs: 500 })
+    try {
+      await vi.waitFor(() => {
+        expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
+      })
+      const timing = (globalThis as Record<string, unknown>).__fxTiming as
+        | { breakStreams(): void }
+        | undefined
+      if (timing === undefined) throw new Error('fixture timing hooks missing')
+      timing.breakStreams()
+
+      await vi.waitFor(() => { expect(reconnectSnapshots).toEqual([undefined]) })
+      await vi.waitFor(() => { expect(descriptions).toEqual([true, undefined, true]) })
+      expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
+    } finally {
+      stopDescription()
+      loop.stop()
+      warnSpy.mockRestore()
+    }
   })
 
   it('WebApiClient keeps unary calls and respond on globalThis.fetch', async () => {

packages/client/connection/tests/connection.spec.ts

@@ -23,10 +23,14 @@ describe('connection lifecycle', () => {
   it('announces connected after describe + both streams open, then pumps frames to sinks', async () => {
     const api = new FakeApiClient()
     const muxSeen: string[] = []
+    const descriptions: boolean[] = []
     let connected = 0
     const controller = new ConnectionController(api, {
       onMuxEnvelope: envelope => muxSeen.push(envelope.payload.type),
-      onConnected: () => { connected++ },
+      onConnected: (description) => {
+        connected++
+        descriptions.push(description.canOpenPath)
+      },
     }, FAST)
     controller.start()
     try {
@@ -34,6 +38,7 @@ describe('connection lifecycle', () => {
       api.pushMux(subscribedFrame())
       await vi.waitFor(() => { expect(muxSeen).toEqual(['session/subscribed']) })
       expect(api.callsOf('host.describe')).toHaveLength(1)
+      expect(descriptions).toEqual([true])
     } finally {
       controller.stop()
     }
@@ -75,7 +80,7 @@ describe('connection lifecycle', () => {
     try {
       await vi.waitFor(() => { expect(describeCalls).toBe(2) }) // retried after backoff
       expect(connected).toBe(0) // never announced during the failed generation
-      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
+      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, canOpenPath: true }))
       await vi.waitFor(() => { expect(connected).toBe(1) })
     } finally {
       controller.stop()
@@ -97,7 +102,7 @@ describe('connection lifecycle', () => {
           },
         })
       }
-      return Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
+      return Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, canOpenPath: true }))
     }
     let connected = 0
     const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
@@ -175,6 +180,38 @@ describe('connection lifecycle', () => {
     }
   })
 
+  it('rejects a generation whose streams end during readiness and retries', async () => {
+    const api = new FakeApiClient()
+    const firstDescribe = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
+    let describeCalls = 0
+    api.onDescribe = () => {
+      describeCalls++
+      return describeCalls === 1
+        ? firstDescribe.promise
+        : Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, canOpenPath: true }))
+    }
+    const states: ConnectionState[] = []
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, {
+      onConnected: () => { connected++ },
+      onStateChange: state => states.push(state),
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(api.openMuxCount).toBe(1) })
+      api.endStreams()
+      firstDescribe.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, canOpenPath: true }))
+
+      await vi.waitFor(() => { expect(describeCalls).toBe(2) })
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      expect(states).toEqual(['reconnecting', 'connected'])
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+  })
+
   it('proceeds as connected via the timeout guard when a carrier never fires onOpen', async () => {
     const api = new FakeApiClient()
     api.suppressStreamOpen = true // misbehaving carrier: streams open but onOpen never fires
@@ -210,6 +247,24 @@ describe('connection lifecycle', () => {
     }
   })
 
+  it('does not announce a generation stopped synchronously by its connected state sink', async () => {
+    const api = new FakeApiClient()
+    const states: ConnectionState[] = []
+    let connected = 0
+    const controller = new ConnectionController(api, {
+      onConnected: () => { connected++ },
+      onStateChange: (state) => {
+        states.push(state)
+        if (state === 'connected') controller.stop()
+      },
+    }, FAST)
+
+    controller.start()
+    await vi.waitFor(() => { expect(states).toEqual(['connected']) })
+    await vi.waitFor(() => { expect(api.openMuxCount).toBe(0) })
+    expect(connected).toBe(0)
+  })
+
   it('deduplicates consecutive reconnecting emissions across two straight failures', async () => {
     const api = new FakeApiClient()
     const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
@@ -228,7 +283,7 @@ describe('connection lifecycle', () => {
     controller.start()
     try {
       await vi.waitFor(() => { expect(describeCalls).toBe(3) })
-      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
+      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, canOpenPath: true }))
       await vi.waitFor(() => { expect(connected).toBe(1) })
       expect(states).toEqual(['reconnecting', 'connected']) // two failures, one reconnecting emission
     } finally {

packages/client/connection/tests/fake-api.ts

@@ -71,8 +71,15 @@ export class FakeApiClient implements IApiClient {
     () => Promise.resolve(ok({ attachment: { attachmentId: 'a' as never, mediaType: 'image/png', bytes: 1, width: 1, height: 1 }, data: 'AA==' }))
   onUpdateQueue: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
   onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
-  onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
-    () => Promise.resolve(ok({ version: '0-fake', cwd: '/f', attachedSessions: 0 }))
+  onDescribe: (payload: unknown) => Promise<RpcResponse<{
+    version: string
+    cwd: string
+    attachedSessions: number
+    canOpenPath: boolean
+  }>> =
+    () => Promise.resolve(ok({
+      version: '0-fake', cwd: '/f', attachedSessions: 0, canOpenPath: true,
+    }))
   onPickDirectory: (payload: unknown) => Promise<RpcResponse<{ path: string | null }>> =
     () => Promise.resolve(ok({ path: null }))
   onOpenPath: (payload: unknown) => Promise<RpcResponse<{ opened: true }>> =