diff --git a/packages/api/gateway/src/client/remote-events.ts b/packages/api/gateway/src/client/remote-events.ts index 341c1f3f4d..9552478162 100644 --- a/packages/api/gateway/src/client/remote-events.ts +++ b/packages/api/gateway/src/client/remote-events.ts @@ -3,6 +3,7 @@ import type { Context } from '@deepseek-ai/cordis' import type { ConnectionGenerationSource, + ConnectionHostInfo, ConnectionHandle, } from '@deepseek-ai/dsh-client-connection/client' import type { @@ -118,7 +119,10 @@ export class ClientRemoteEvents { } /** Run one Connection generation over the forwarded-event logical stream. */ - private async pumpEvents(signal: AbortSignal, ready: () => void): Promise { + private async pumpEvents( + signal: AbortSignal, + ready: (host: ConnectionHostInfo) => void, + ): Promise { let clientId: RemoteEventClientId | undefined const failed = new AbortController() const generationSignal = AbortSignal.any([signal, failed.signal]) @@ -134,8 +138,9 @@ export class ClientRemoteEvents { try { for await (const value of source) { if (clientId === undefined) { - clientId = parseRemoteEventReady(value) - ready() + const opening = parseRemoteEventReady(value) + clientId = opening.clientId + ready(opening.host) continue } const frame = parseRemoteEventFrame(value) @@ -251,15 +256,21 @@ export class ClientRemoteEvents { } } -/** Validate and return the Client identity from one generation's opening item. */ -function parseRemoteEventReady(value: unknown): RemoteEventClientId { +/** Validate and return one generation's Client identity and Host facts. */ +function parseRemoteEventReady(value: unknown): { + readonly clientId: RemoteEventClientId + readonly host: ConnectionHostInfo +} { if (!isRemoteEventRecord(value) - || !hasExactRemoteEventKeys(value, ['type', 'clientId']) + || !hasExactRemoteEventKeys(value, ['type', 'clientId', 'host']) || value.type !== 'ready' - || !isRemoteEventClientId(value.clientId)) { + || !isRemoteEventClientId(value.clientId) + || !isRemoteEventRecord(value.host) + || !hasExactRemoteEventKeys(value.host, ['home']) + || typeof value.host.home !== 'string') { throw new TypeError('client api: forwarded Remote event stream did not begin with ready') } - return value.clientId + return { clientId: value.clientId, host: { home: value.host.home } } } /** Validate one untrusted value from the Gateway-internal forwarded-event stream. */ diff --git a/packages/api/gateway/src/client/remote-stream.ts b/packages/api/gateway/src/client/remote-stream.ts index 799a371ef5..71f1f546ae 100644 --- a/packages/api/gateway/src/client/remote-stream.ts +++ b/packages/api/gateway/src/client/remote-stream.ts @@ -48,7 +48,7 @@ export class RemoteStream implements AsyncIterable> * @param options - domain stream opener, end classification, and diagnostics. */ constructor( - private readonly connection: Pick, + private readonly connection: Pick, private readonly options: RemoteStreamOptions, ) {} @@ -157,13 +157,13 @@ export class RemoteStream implements AsyncIterable> } async function waitForRemoteStreamRetry( - connection: Pick, + connection: Pick, error: RemoteStreamCarrierError, attempt: number, signal: AbortSignal, ): Promise { signal.throwIfAborted() - if (connection.hostDescription.getSnapshot() !== undefined) { + if (connection.generation.getSnapshot() !== undefined) { if (attempt === 1) return throw error } @@ -181,12 +181,12 @@ async function waitForRemoteStreamRetry( else reject(failure) } const inspect = (): void => { - if (connection.hostDescription.getSnapshot() !== undefined) finish() + if (connection.generation.getSnapshot() !== undefined) finish() } const aborted = (): void => { finish(new Error('Remote stream retry aborted', { cause: signal.reason })) } - const dispose = connection.hostDescription.subscribe(inspect) + const dispose = connection.generation.subscribe(inspect) subscription.dispose = dispose if (subscription.finished) dispose() signal.addEventListener('abort', aborted, { once: true }) diff --git a/packages/api/gateway/src/index.ts b/packages/api/gateway/src/index.ts index ec4eb1eaef..1d754b05a7 100644 --- a/packages/api/gateway/src/index.ts +++ b/packages/api/gateway/src/index.ts @@ -48,6 +48,7 @@ import { type RemoteEventCancellationFrame, type RemoteEventClientId, type RemoteEventEmitFrame, + type RemoteEventHostInfo, type RemoteEventId, type RemoteEventInvocationFrame, type RemoteEventReadyFrame, @@ -66,6 +67,7 @@ export type { TypertRemoteEventOutcome, TypertRemoteEventSource, } from './types.ts' +export type { RemoteEventHostInfo } from './stream-protocol.ts' interface GatewayErrorOptions { readonly cause?: unknown @@ -88,6 +90,7 @@ interface PreparedInvocation { interface RegisteredRemoteEventSource { readonly lifetime: AbortController readonly done: Promise + readonly host: RemoteEventHostInfo } interface RemoteEventClient { @@ -233,9 +236,13 @@ export class TypertGatewayService extends Service implements TypertGateway { /** * Register the sole application-selected forwarded-event source. * @param source - stream factory installed by the Remote assembly. + * @param host - stable Host facts included in each Client generation's opening frame. * @returns disposer removing this source and cancelling its active streams. */ - registerRemoteEvents(source: TypertRemoteEventSource): () => Promise { + registerRemoteEvents( + source: TypertRemoteEventSource, + host: RemoteEventHostInfo, + ): () => Promise { if (this.remoteEvents !== undefined) { throw new Error('typert gateway: forwarded Remote event source is already registered') } @@ -247,7 +254,7 @@ export class TypertGatewayService extends Service implements TypertGateway { this.remoteEvents = undefined lifetime.abort(error) }) - const registration: RegisteredRemoteEventSource = { lifetime, done } + const registration: RegisteredRemoteEventSource = { lifetime, done, host: { home: host.home } } this.remoteEvents = registration return async () => { if (this.remoteEvents === registration) { @@ -417,7 +424,7 @@ export class TypertGatewayService extends Service implements TypertGateway { this.remoteEventClients.set(clientId, client) for (const pending of this.pendingRemoteEvents.values()) this.deliverRemoteEvent(pending, client) try { - yield { ...REMOTE_EVENT_STREAM_READY, clientId } + yield { ...REMOTE_EVENT_STREAM_READY, clientId, host: registration.host } yield* client.queue.iterate(lifetime) } finally { this.removeRemoteEventClient(client) diff --git a/packages/api/gateway/src/stream-protocol.ts b/packages/api/gateway/src/stream-protocol.ts index 2c6e8ebf70..142598863e 100644 --- a/packages/api/gateway/src/stream-protocol.ts +++ b/packages/api/gateway/src/stream-protocol.ts @@ -23,10 +23,18 @@ export type RemoteEventClientId = Branded<'RemoteEventClientId'> /** Opaque correlation id for one pending Host-to-Client Remote Event. */ export type RemoteEventId = Branded<'RemoteEventId'> +/** Stable Host facts published with every established Client event generation. */ +export interface RemoteEventHostInfo { + /** Host account home used only to abbreviate displayed filesystem paths. */ + readonly home: string +} + /** Opening item that binds later HTTP results to this active event stream. */ export interface RemoteEventReadyFrame { readonly type: 'ready' readonly clientId: RemoteEventClientId + /** Stable Host facts attached to this connection generation. */ + readonly host: RemoteEventHostInfo } /** Opaque Agent identity carried by one scoped Remote Event. */ diff --git a/packages/api/gateway/src/types.ts b/packages/api/gateway/src/types.ts index b41f35e905..9b4475cb9a 100644 --- a/packages/api/gateway/src/types.ts +++ b/packages/api/gateway/src/types.ts @@ -4,6 +4,7 @@ */ import type { Context } from '@deepseek-ai/cordis' +import type { RemoteEventHostInfo } from './stream-protocol.ts' /** One Remote method request after a carrier has decoded its envelope. */ export interface InvokeRemoteRequest { @@ -124,9 +125,13 @@ export interface TypertGateway { /** * Register the application-selected forwarded-event source. * @param source - stream factory installed by the Remote assembly. + * @param host - stable Host facts included in each Client generation's opening frame. * @returns disposer removing this exact source and cancelling its active streams. */ - registerRemoteEvents(source: TypertRemoteEventSource): () => Promise + registerRemoteEvents( + source: TypertRemoteEventSource, + host: RemoteEventHostInfo, + ): () => Promise /** * Invoke one live Remote method without assuming a carrier or response envelope. diff --git a/packages/api/gateway/tests/control-retry.client.spec.ts b/packages/api/gateway/tests/control-retry.client.spec.ts index a278cc6acd..234251babb 100644 --- a/packages/api/gateway/tests/control-retry.client.spec.ts +++ b/packages/api/gateway/tests/control-retry.client.spec.ts @@ -5,23 +5,17 @@ import { RemoteStream, } from '../src/client/index.ts' -const DESCRIPTION = { - version: 'fixture', - cwd: '/fixture', - attachedSessions: 0, - home: '/home/fixture', - canOpenPath: true, -} +const GENERATION = { id: 1, host: { home: '/home/fixture' } } function hostSource(initiallyAvailable: boolean): { - connection: Pick + connection: Pick publish(available: boolean): void } { - let current = initiallyAvailable ? DESCRIPTION : undefined + let current = initiallyAvailable ? GENERATION : undefined const listeners = new Set<() => void>() return { connection: { - hostDescription: { + generation: { getSnapshot: () => current, subscribe: (listener) => { listeners.add(listener) @@ -30,7 +24,7 @@ function hostSource(initiallyAvailable: boolean): { }, }, publish: (available) => { - current = available ? DESCRIPTION : undefined + current = available ? GENERATION : undefined for (const listener of listeners) listener() }, } @@ -67,7 +61,7 @@ function scripted(generations: Generation[], opened?: () => void) { } function supervisor( - connection: Pick, + connection: Pick, generations: Generation[], carrierFailed?: (error: RemoteStreamCarrierError) => void, ): RemoteStream { @@ -122,8 +116,8 @@ describe('RemoteStream', () => { let listener: (() => void) | undefined const subscribed = Promise.withResolvers() const connection = { - hostDescription: { - getSnapshot: () => available ? DESCRIPTION : undefined, + generation: { + getSnapshot: () => available ? GENERATION : undefined, subscribe: (value: () => void) => { listener = value subscribed.resolve(undefined) @@ -177,8 +171,8 @@ describe('RemoteStream', () => { let reads = 0 let disposed = 0 const connection = { - hostDescription: { - getSnapshot: () => reads++ === 0 ? undefined : DESCRIPTION, + generation: { + getSnapshot: () => reads++ === 0 ? undefined : GENERATION, subscribe: (listener: () => void) => { listener() return () => { disposed++ } @@ -261,7 +255,7 @@ describe('RemoteStream', () => { const holder: { stream?: RemoteStream } = {} let subscriptions = 0 const connection = { - hostDescription: { + generation: { getSnapshot: () => undefined, subscribe: () => { subscriptions++ diff --git a/packages/api/gateway/tests/gateway-stream.host.spec.ts b/packages/api/gateway/tests/gateway-stream.host.spec.ts index debb6763d5..c769192a4b 100644 --- a/packages/api/gateway/tests/gateway-stream.host.spec.ts +++ b/packages/api/gateway/tests/gateway-stream.host.spec.ts @@ -36,6 +36,7 @@ vi.mock('node:crypto', async (importOriginal) => { const randomUuid = vi.mocked(randomUUID) const browserCookies = new WeakMap() +const REMOTE_HOST = { home: '/home/fixture' } as const type AgentWireId = TypertContextWire const agentId = (value: string): AgentWireId => value as AgentWireId @@ -386,8 +387,8 @@ describe('Typert Remote streams', () => { } })() } - const unregister = ctx.typertGateway.registerRemoteEvents(source) - expect(() => { ctx.typertGateway.registerRemoteEvents(source) }) + const unregister = ctx.typertGateway.registerRemoteEvents(source, REMOTE_HOST) + expect(() => { ctx.typertGateway.registerRemoteEvents(source, REMOTE_HOST) }) .toThrow('forwarded Remote event source is already registered') const socket = new WebSocket(`ws://127.0.0.1:${String(ctx.webServer.port)}/api/remote.mux`, { @@ -402,7 +403,7 @@ describe('Typert Remote streams', () => { const eventFrames = frames.filter(frame => frame.streamId === 'events') expect(eventFrames).toHaveLength(1) expect(eventFrames[0]).toMatchObject({ - type: 'item', streamId: 'events', value: { type: 'ready' }, + type: 'item', streamId: 'events', value: { type: 'ready', host: REMOTE_HOST }, }) expect(typeof Reflect.get(eventFrames[0]!.value as object, 'clientId')).toBe('string') }) @@ -411,7 +412,7 @@ describe('Typert Remote streams', () => { const eventFrames = frames.filter(frame => frame.streamId === 'events').slice(0, 2) expect(eventFrames).toHaveLength(2) expect(eventFrames[0]).toMatchObject({ - type: 'item', streamId: 'events', value: { type: 'ready' }, + type: 'item', streamId: 'events', value: { type: 'ready', host: REMOTE_HOST }, }) expect(typeof Reflect.get(eventFrames[0]!.value as object, 'clientId')).toBe('string') expect(eventFrames[1]).toEqual({ @@ -429,9 +430,9 @@ describe('Typert Remote streams', () => { expect(frames).toContainEqual({ type: 'end', streamId: 'events' }) }) - const unregisterReplacement = ctx.typertGateway.registerRemoteEvents(source) + const unregisterReplacement = ctx.typertGateway.registerRemoteEvents(source, REMOTE_HOST) await unregister() - expect(() => { ctx.typertGateway.registerRemoteEvents(source) }) + expect(() => { ctx.typertGateway.registerRemoteEvents(source, REMOTE_HOST) }) .toThrow('forwarded Remote event source is already registered') await unregisterReplacement() socket.close() @@ -446,7 +447,7 @@ describe('Typert Remote streams', () => { await publish.promise yield pending.dispatch })() - const unregister = ctx.typertGateway.registerRemoteEvents(source) + const unregister = ctx.typertGateway.registerRemoteEvents(source, REMOTE_HOST) const rejected = expect(pending.outcome).rejects.toThrow( 'forwarded Remote event source was removed', ) @@ -479,7 +480,7 @@ describe('Typert Remote streams', () => { else signal.addEventListener('abort', () => { resolve() }, { once: true }) }) throw new Error('fixture source rejected during removal') - })()) + })(), REMOTE_HOST) const client = await openEventClient(ctx, 'events-removal') await vi.waitFor(() => { expect(deliveredInvocation(client)).toBeDefined() }) @@ -496,7 +497,7 @@ describe('Typert Remote streams', () => { it('delegates unavailable Contexts and rejects malformed scoped invocations', async () => { const { ctx } = await setup(false) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) for (const event of [42, ''] as const) { const invalidName = pendingInvocation(ctx) @@ -579,7 +580,7 @@ describe('Typert Remote streams', () => { return (async function* () { yield frame as unknown as TypertRemoteEventDispatch })() - }) + }, REMOTE_HOST) await vi.waitFor(() => { expect(sourceSignal?.aborted).toBe(true) }) const reason: unknown = sourceSignal?.reason if (!(reason instanceof Error)) throw new Error('Remote event source did not fail with an Error') @@ -591,7 +592,7 @@ describe('Typert Remote streams', () => { it('retries a colliding Remote event id before publishing the second waterfall', async () => { const { ctx } = await setup(false) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -626,7 +627,7 @@ describe('Typert Remote streams', () => { it('retries a colliding Remote event Client id before opening the second generation', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const firstId = '00000000-0000-4000-8000-000000000011' as ReturnType const secondId = '00000000-0000-4000-8000-000000000012' as ReturnType randomUuid.mockReturnValueOnce(firstId).mockReturnValueOnce(firstId).mockReturnValueOnce(secondId) @@ -645,7 +646,7 @@ describe('Typert Remote streams', () => { it('fans one scoped waterfall out and accepts the first Client result', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -699,7 +700,7 @@ describe('Typert Remote streams', () => { it('rejects the Host waterfall with the first Client listener rejection', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -739,7 +740,7 @@ describe('Typert Remote streams', () => { it('delegates to the Host only after every active Client returns next', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -772,7 +773,7 @@ describe('Typert Remote streams', () => { it('delivers a pending waterfall to the first Client that connects', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -805,7 +806,7 @@ describe('Typert Remote streams', () => { it('replays a pending event id to a replacement Client generation', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const agent = ctx.extend() ctx.typert.contexts.registerHost('agent', { wire: 'agentId', @@ -839,7 +840,7 @@ describe('Typert Remote streams', () => { it('cancels pending deliveries when the Host signal or Context ends', async () => { const { ctx } = await setup(true) const source = new RemoteEventSourceProbe() - const unregister = ctx.typertGateway.registerRemoteEvents(source.source) + const unregister = ctx.typertGateway.registerRemoteEvents(source.source, REMOTE_HOST) const signalAgent = ctx.extend() const contextFiber = ctx.plugin(() => {}) await contextFiber @@ -929,7 +930,7 @@ describe('Typert Remote streams', () => { const unregister = ctx.typertGateway.registerRemoteEvents(() => { sourceCalls += 1 return (async function *(): AsyncIterable {})() - }) + }, REMOTE_HOST) const invalidPayloads: readonly unknown[] = [ null, [], diff --git a/packages/api/gateway/tests/gateway.client.spec.ts b/packages/api/gateway/tests/gateway.client.spec.ts index 92eeecb5e0..a77e4a22af 100644 --- a/packages/api/gateway/tests/gateway.client.spec.ts +++ b/packages/api/gateway/tests/gateway.client.spec.ts @@ -483,7 +483,7 @@ class RemoteEventCarrier { const abort = (): void => { connection.wake?.() } signal.addEventListener('abort', abort, { once: true }) try { - yield { type: 'ready', clientId } + yield { type: 'ready', clientId, host: { home: '/home/fixture' } } while (!signal.aborted) { while (connection.items.length > 0) { const item = connection.items.shift() as EventStreamItem @@ -1769,7 +1769,7 @@ describe('Client Typert API', () => { socket.receive({ type: 'item', streamId: opened.streamId, - value: { type: 'ready', clientId: 'browser-client' }, + value: { type: 'ready', clientId: 'browser-client', host: { home: '/home/browser' } }, }) await run.ready socket.receive({ @@ -1825,6 +1825,7 @@ describe('Client Typert API', () => { await vi.waitFor(() => { expect(connection.hostDescription.getSnapshot()?.home).toBe('/home/fixture') + expect(connection.generation.getSnapshot()?.host.home).toBe('/home/fixture') }) } finally { await ctx.fiber.dispose() @@ -1841,6 +1842,10 @@ describe('Client Typert API', () => { { type: 'ready' }, { type: 'ready', clientId: '' }, { type: 'ready', clientId: 'client', extra: true }, + { type: 'ready', clientId: 'client', host: null }, + { type: 'ready', clientId: 'client', host: {} }, + { type: 'ready', clientId: 'client', host: { home: 1 } }, + { type: 'ready', clientId: 'client', host: { home: '/home', extra: true } }, { type: 'emit', event: 'fixture/changed', args: ['too early'] }, ])('rejects malformed forwarded-event readiness item %#', async (opening) => { const open: NonNullable = () => (async function *() { diff --git a/packages/api/gateway/tests/gateway.host.spec.ts b/packages/api/gateway/tests/gateway.host.spec.ts index c0455bba10..ab1ee9d6b9 100644 --- a/packages/api/gateway/tests/gateway.host.spec.ts +++ b/packages/api/gateway/tests/gateway.host.spec.ts @@ -1084,11 +1084,14 @@ describe('TypertGatewayService', () => { if (signal.aborted) resolve() else signal.addEventListener('abort', () => { resolve() }, { once: true }) }) - })()) + })(), { home: '/home/fixture' }) const carrier = new AbortController() const events = rawGatewayEventHarness(ctx).openRemoteEvents({ args: {} }, carrier.signal) const opening = await events.next() - expect(opening).toMatchObject({ done: false, value: { type: 'ready' } }) + expect(opening).toMatchObject({ + done: false, + value: { type: 'ready', host: { home: '/home/fixture' } }, + }) if (opening.done) throw new Error('Remote event stream ended before ready') const clientId: unknown = Reflect.get(opening.value as object, 'clientId') if (typeof clientId !== 'string') throw new Error('Remote event stream omitted its Client id') diff --git a/packages/api/gateway/tests/journal-stream.client.spec.ts b/packages/api/gateway/tests/journal-stream.client.spec.ts index 0d134f3d99..e5e02b2d28 100644 --- a/packages/api/gateway/tests/journal-stream.client.spec.ts +++ b/packages/api/gateway/tests/journal-stream.client.spec.ts @@ -42,10 +42,8 @@ interface Generation { type PageSource = Page | Promise | ((signal: AbortSignal) => Promise) const AVAILABLE_CONNECTION = { - hostDescription: { - getSnapshot: () => ({ - version: 'fixture', cwd: '/fixture', attachedSessions: 0, home: '/home/fixture', canOpenPath: true, - }), + generation: { + getSnapshot: () => ({ id: 1, host: { home: '/home/fixture' } }), subscribe: () => () => {}, }, } diff --git a/packages/api/remotes/src/index.ts b/packages/api/remotes/src/index.ts index 4d0256e162..e775d3a5ae 100644 --- a/packages/api/remotes/src/index.ts +++ b/packages/api/remotes/src/index.ts @@ -1,5 +1,6 @@ /** Host BFF entry and Loader shell for the Remote contribution assembly. */ +import { homedir } from 'node:os' import type { Context } from '@deepseek-ai/cordis' import type { TypertRemoteEventDispatch, @@ -35,7 +36,7 @@ export const inject = ['typertGateway'] /** Host plugin body registering this application's selected Cordis event source. */ export function apply(ctx: Context): void { ctx.effect( - () => ctx.typertGateway.registerRemoteEvents(remoteEventSource(ctx)), + () => ctx.typertGateway.registerRemoteEvents(remoteEventSource(ctx), { home: homedir() }), 'api-remotes: forwarded Cordis event source', ) } diff --git a/packages/api/remotes/tests/remote-events.host.spec.ts b/packages/api/remotes/tests/remote-events.host.spec.ts index eefe64e664..1d2e8c235d 100644 --- a/packages/api/remotes/tests/remote-events.host.spec.ts +++ b/packages/api/remotes/tests/remote-events.host.spec.ts @@ -1,6 +1,7 @@ import { Context } from '@deepseek-ai/cordis' import type { Fiber } from '@deepseek-ai/cordis' import type { + RemoteEventHostInfo, TypertRemoteEventInvocation, TypertRemoteEventSource, } from '@deepseek-ai/dsh-api-gateway' @@ -10,8 +11,12 @@ import { apply, inject } from '../src/index.ts' interface GatewayProbe { source: TypertRemoteEventSource | undefined + host: RemoteEventHostInfo | undefined removals: number - registerRemoteEvents(source: TypertRemoteEventSource): () => Promise + registerRemoteEvents( + source: TypertRemoteEventSource, + host: RemoteEventHostInfo, + ): () => Promise } async function setup(): Promise<{ @@ -22,12 +27,15 @@ async function setup(): Promise<{ const ctx = new Context() const gateway: GatewayProbe = { source: undefined, + host: undefined, removals: 0, - registerRemoteEvents(source) { + registerRemoteEvents(source, host) { gateway.source = source + gateway.host = host return async () => { if (gateway.source !== source) return gateway.source = undefined + gateway.host = undefined gateway.removals += 1 } }, @@ -71,6 +79,14 @@ function invocationOf(value: unknown): TypertRemoteEventInvocation { } describe('Remote event Host source', () => { + it('registers the Host home used by Client connection generations', async () => { + const { gateway, fiber } = await setup() + expect(gateway.host?.home).toBeTypeOf('string') + expect(gateway.host?.home.length).toBeGreaterThan(0) + await fiber.dispose() + expect(gateway.host).toBeUndefined() + }) + it('gives each Client stream an independent allowlisted event queue', async () => { const { ctx, gateway, fiber } = await setup() const firstAbort = new AbortController() diff --git a/packages/api/session-controller/src/client/index.ts b/packages/api/session-controller/src/client/index.ts index bca9f9d3d0..04b7cbb553 100644 --- a/packages/api/session-controller/src/client/index.ts +++ b/packages/api/session-controller/src/client/index.ts @@ -111,7 +111,7 @@ export function apply(ctx: Context): void { }) control.start() ctx.on('connection/reset', () => { sessions.handleConnected() }) - if (connection.hostDescription.getSnapshot() !== undefined) sessions.handleConnected() + if (connection.generation.getSnapshot() !== undefined) sessions.handleConnected() ctx.typert.contexts.registerClient('agent', { identity: candidate => sessions.scopeOf(candidate), resolve: sessionId => sessions.resolveAgentScope(sessionId), diff --git a/packages/api/session-controller/tests/client-apply.client.spec.ts b/packages/api/session-controller/tests/client-apply.client.spec.ts index 876786a759..b6c7e40972 100644 --- a/packages/api/session-controller/tests/client-apply.client.spec.ts +++ b/packages/api/session-controller/tests/client-apply.client.spec.ts @@ -64,6 +64,24 @@ async function mount(initialHost?: HostDescription): Promise { return () => { hostListeners.delete(listener) } }, }, + generation: { + getSnapshot: () => host === undefined + ? undefined + : { id: 1, host: { home: host.home } }, + subscribe: (listener) => { + hostListeners.add(listener) + return () => { hostListeners.delete(listener) } + }, + }, + generation: { + getSnapshot: () => host === undefined + ? undefined + : { id: 1, host: { home: host.home } }, + subscribe: (listener) => { + hostListeners.add(listener) + return () => { hostListeners.delete(listener) } + }, + }, rpc: { call: () => Promise.reject(new Error('unexpected generic RPC call')), }, diff --git a/packages/api/session-controller/tests/fake-api.client.ts b/packages/api/session-controller/tests/fake-api.client.ts index f1369e4c8c..851ce01030 100644 --- a/packages/api/session-controller/tests/fake-api.client.ts +++ b/packages/api/session-controller/tests/fake-api.client.ts @@ -32,10 +32,8 @@ import type { SessionRemotes } from '../src/client/sessions/remotes.ts' import { historyRecordLastSeq } from '../src/client/sessions/history-records.ts' const AVAILABLE_STREAM_CONNECTION = { - hostDescription: { - getSnapshot: () => ({ - version: 'fixture', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true, - }), + generation: { + getSnapshot: () => ({ id: 1, host: { home: '/h' } }), subscribe: () => () => {}, }, } diff --git a/packages/api/session-controller/tests/transport.client.spec.ts b/packages/api/session-controller/tests/transport.client.spec.ts index 1ad1d4e858..36afa47632 100644 --- a/packages/api/session-controller/tests/transport.client.spec.ts +++ b/packages/api/session-controller/tests/transport.client.spec.ts @@ -28,10 +28,8 @@ type SessionTransportRemote = Pick const ADDRESS: SessionAddress = { kind: 'session', sessionId: 'session-1' as never } const AVAILABLE_CONNECTION = { - hostDescription: { - getSnapshot: () => ({ - version: 'fixture', cwd: '/fixture', attachedSessions: 0, home: '/home/fixture', canOpenPath: true, - }), + generation: { + getSnapshot: () => ({ id: 1, host: { home: '/home/fixture' } }), subscribe: () => () => {}, }, } diff --git a/packages/api/workspace-controller/tests/transport.client.spec.ts b/packages/api/workspace-controller/tests/transport.client.spec.ts index d71cfba44e..14cf3a02db 100644 --- a/packages/api/workspace-controller/tests/transport.client.spec.ts +++ b/packages/api/workspace-controller/tests/transport.client.spec.ts @@ -36,17 +36,15 @@ import type { } from '../src/types.ts' const AVAILABLE_CONNECTION = { - hostDescription: { - getSnapshot: () => ({ - version: 'fixture', cwd: '/fixture', attachedSessions: 0, home: '/home/fixture', canOpenPath: true, - }), + generation: { + getSnapshot: () => ({ id: 1, host: { home: '/home/fixture' } }), subscribe: () => () => {}, }, } function workspaceClient( remote: WorkspaceRemote, - connection: Pick = AVAILABLE_CONNECTION, + connection: Pick = AVAILABLE_CONNECTION, ) { return { workspace: remote, @@ -211,6 +209,7 @@ function provideClientServices(ctx: Context, remote: WorkspaceRemote): void { }), subscribe: () => () => {}, }, + generation: AVAILABLE_CONNECTION.generation, rpc: { call: () => Promise.reject(new Error('unexpected generic RPC call')), }, diff --git a/packages/client/connection/src/client/connection.ts b/packages/client/connection/src/client/connection.ts index cad2de394d..ebcac74ea7 100644 --- a/packages/client/connection/src/client/connection.ts +++ b/packages/client/connection/src/client/connection.ts @@ -1,5 +1,19 @@ import type { HostDescription, IApiClient } from './api.ts' +/** Stable Host facts delivered by one established Remote event generation. */ +export interface ConnectionHostInfo { + /** Host account home used only to abbreviate displayed filesystem paths. */ + readonly home: string +} + +/** One successfully established Host generation. */ +export interface ConnectionGeneration { + /** Monotone generation number within this Client runtime. */ + readonly id: number + /** Host facts carried by this generation's opening frame. */ + readonly host: ConnectionHostInfo +} + /** Reconnect/backoff tunables (deployment-varying — no hardcoded tunables; these become the * future `ctx.connection` plugin's Config). All fields optional; defaults below. */ export interface ConnectionConfig { @@ -39,7 +53,7 @@ export type ConnectionState = 'connected' | 'reconnecting' /** Connection-generation callbacks owned by API Gateway. */ export interface ConnectionSinks { /** After the generation source is ready and host.describe succeeds, first connect included. */ - onConnected?: (description: HostDescription) => void + onConnected?: (description: HostDescription, host: ConnectionHostInfo) => void /** Coarse state transitions (deduplicated: fires only on change). The initial pre-connect * span reports nothing — the UI treats "no state yet" as connecting, not as an outage. */ onStateChange?: (state: ConnectionState) => void @@ -55,7 +69,7 @@ export interface ConnectionSinks { */ export type ConnectionGenerationSource = ( signal: AbortSignal, - ready: () => void, + ready: (host: ConnectionHostInfo) => void, ) => Promise /** @@ -117,19 +131,20 @@ export class ConnectionController { this.current = ac let sourceReady = false - let resolveReady!: () => void + let resolveReady!: (host: ConnectionHostInfo) => void let rejectReady!: (error: Error) => void let rejectSourceLost!: (error: Error) => void - const ready = new Promise((resolve, reject) => { + const ready = new Promise((resolve, reject) => { resolveReady = resolve rejectReady = reject }) const sourceLost = new Promise((_resolve, reject) => { rejectSourceLost = reject }) - const reportReady = (): void => { + const reportReady = (host: ConnectionHostInfo): void => { + if (sourceReady) return sourceReady = true - resolveReady() + resolveReady(host) } const failed = new Promise((resolve) => { @@ -161,7 +176,7 @@ export class ConnectionController { // The source reports ready only after its incremental listeners exist; // describe may complete in parallel, but consumers see neither result // until both sides of the baseline-plus-increment handshake are ready. - const [description] = await Promise.race([ + const [description, host] = await Promise.race([ Promise.all([ this.api.host.describe({}, ac.signal), waitForReady(ready, this.config.generationReadyTimeoutMs, ac.signal), @@ -178,7 +193,7 @@ export class ConnectionController { // A state sink may synchronously stop this controller. Do not publish // a description for a generation that no longer exists afterward. if (this.isGenerationActive(ac)) { - this.callSink(() => { this.sinks.onConnected?.(descriptionResult.value) }) + this.callSink(() => { this.sinks.onConnected?.(descriptionResult.value, host) }) } } catch { // Transport failure: treat as generation failure, fall through to the shared backoff. @@ -213,28 +228,28 @@ export class ConnectionController { } /** Await source readiness without letting a stalled carrier wedge startup forever. */ -function waitForReady(ready: Promise, timeoutMs: number, signal: AbortSignal): Promise { - return new Promise((resolve, reject) => { +function waitForReady(ready: Promise, timeoutMs: number, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { let settled = false const timeout = setTimeout(() => { - finish(new Error(`connection generation was not ready within ${String(timeoutMs)}ms`)) + finish({ error: new Error(`connection generation was not ready within ${String(timeoutMs)}ms`) }) }, timeoutMs) const aborted = (): void => { - finish(new Error('connection generation aborted', { cause: signal.reason })) + finish({ error: new Error('connection generation aborted', { cause: signal.reason }) }) } - const finish = (error?: Error): void => { + const finish = (outcome: { readonly value: T } | { readonly error: Error }): void => { if (settled) return settled = true clearTimeout(timeout) signal.removeEventListener('abort', aborted) - if (error === undefined) resolve() - else reject(error) + if ('error' in outcome) reject(outcome.error) + else resolve(outcome.value) } signal.addEventListener('abort', aborted, { once: true }) void ready.then( - () => { finish() }, + (value) => { finish({ value }) }, (error: unknown) => { - finish(error as Error) + finish({ error: error as Error }) }, ) }) diff --git a/packages/client/connection/src/client/fixture.ts b/packages/client/connection/src/client/fixture.ts index a2a7617a7e..6a5a1017c5 100644 --- a/packages/client/connection/src/client/fixture.ts +++ b/packages/client/connection/src/client/fixture.ts @@ -176,6 +176,7 @@ interface FixtureRemoteEventResult { interface FixtureRemoteEventReadyFrame { readonly type: 'ready' readonly clientId: string + readonly host: { readonly home: string } } interface FixtureProjectionFrame { @@ -3161,7 +3162,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld { if (gamma !== undefined) setRunning(gamma.sessionId, !gamma.running) }, 5000) try { - yield { type: 'ready', clientId } + yield { type: 'ready', clientId, host: { home: FIXTURE_HOME } } if (approvalPending) yield approvalInvocation() if (questionPending) yield questionInvocation() yield* conn.drain(signal) diff --git a/packages/client/connection/src/client/index.ts b/packages/client/connection/src/client/index.ts index a24d014496..a03ab1e5d7 100644 --- a/packages/client/connection/src/client/index.ts +++ b/packages/client/connection/src/client/index.ts @@ -7,9 +7,9 @@ import type { HostDescription, IApiClient } from './api.ts' import { ConnectionController, type ConnectionConfig, + type ConnectionGeneration, type ConnectionGenerationSource, type ConnectionSinks, - type ConnectionState, } from './connection.ts' import { FixtureApiClient } from './fixture.ts' import { WebApiClient } from './web-api-client.ts' @@ -45,7 +45,14 @@ export { // Connection loop types are public through ConnectionHandle.start; the // controller remains package-internal. -export type { ConnectionConfig, ConnectionGenerationSource, ConnectionSinks, ConnectionState } +export type { + ConnectionConfig, + ConnectionGeneration, + ConnectionGenerationSource, + ConnectionHostInfo, + ConnectionSinks, + ConnectionState, +} from './connection.ts' export type { ClientConnectionRpc, ConnectionRpcFailure, ConnectionRpcResult, } from '../rpc.ts' @@ -59,6 +66,14 @@ export interface HostDescriptionSource { subscribe(listener: () => void): () => void } +/** Observable identity and Host facts for the active connection generation. */ +export interface ConnectionGenerationState { + /** Active generation, or undefined before readiness and while reconnecting. */ + getSnapshot(): ConnectionGeneration | undefined + /** Subscribe to generation establishment, replacement, and loss. */ + subscribe(listener: () => void): () => void +} + /** Required services (none — this is the wire root). */ export const inject: string[] = [] @@ -113,6 +128,8 @@ export interface ConnectionHandle { readonly isLoopback: boolean /** Generation-scoped Host facts, including the account home and native path-open capability. */ readonly hostDescription: HostDescriptionSource + /** Current Remote event generation and the Host facts carried by its opening frame. */ + readonly generation: ConnectionGenerationState /** Generic logical RPC channels over the same Connection transport. */ readonly rpc: ClientConnectionRpc /** @@ -151,6 +168,9 @@ export function apply(ctx: Context): void { const rpc = fixtureClient?.rpc ?? createWebConnectionRpc(transport?.fetch, transport?.openStream) let generationSource: ConnectionGenerationSource | undefined let owner: ConnectionOwner | undefined + let generationId = 0 + let generation: ConnectionGeneration | undefined + const generationListeners = new Set<() => void>() let description: HostDescription | undefined const descriptionListeners = new Set<() => void>() const publishDescription = (next: HostDescription | undefined): void => { @@ -164,10 +184,22 @@ export function apply(ctx: Context): void { } } } + const publishGeneration = (next: ConnectionGeneration | undefined): void => { + if (Object.is(generation, next)) return + generation = next + for (const listener of [...generationListeners]) { + try { + listener() + } catch (error) { + console.error('[connection] generation listener threw:', error) + } + } + } const releaseOwner = (current: ConnectionOwner): void => { if (owner !== current) return owner = undefined current.controller.stop() + publishGeneration(undefined) publishDescription(undefined) } const handle: ConnectionHandle = { @@ -180,6 +212,13 @@ export function apply(ctx: Context): void { return () => { descriptionListeners.delete(listener) } }, }, + generation: { + getSnapshot: () => generation, + subscribe: (listener) => { + generationListeners.add(listener) + return () => { generationListeners.delete(listener) } + }, + }, rpc, registerGenerationSource(source) { if (generationSource !== undefined) { @@ -201,17 +240,23 @@ export function apply(ctx: Context): void { const ownsGeneration = (): boolean => owner?.token === token const controller = new ConnectionController(api, source, { ...sinks, - onConnected: (next) => { + onConnected: (next, host) => { + const nextGeneration = { id: ++generationId, host } + publishGeneration(nextGeneration) + if (!ownsGeneration() || !Object.is(generation, nextGeneration)) return publishDescription(next) // A description subscriber may synchronously stop the loop. In that // case publishDescription(undefined) has already retracted this // generation, so do not leak its stale connected notification to // the consumer sink afterward. if (!ownsGeneration() || !Object.is(description, next)) return - sinks.onConnected?.(next) + sinks.onConnected?.(next, host) }, onStateChange: (state) => { - if (state === 'reconnecting') publishDescription(undefined) + if (state === 'reconnecting') { + publishGeneration(undefined) + publishDescription(undefined) + } if (!ownsGeneration()) return sinks.onStateChange?.(state) }, diff --git a/packages/client/connection/tests/client-apply.client.spec.ts b/packages/client/connection/tests/client-apply.client.spec.ts index 443d398eba..fb5257e355 100644 --- a/packages/client/connection/tests/client-apply.client.spec.ts +++ b/packages/client/connection/tests/client-apply.client.spec.ts @@ -37,7 +37,7 @@ class GenerationProbe { } this.active.add(finish) signal.addEventListener('abort', finish, { once: true }) - ready() + ready({ home: '/h' }) if (signal.aborted) finish() }) diff --git a/packages/client/connection/tests/connection.client.spec.ts b/packages/client/connection/tests/connection.client.spec.ts index 03f57f9273..40f9a5ec6c 100644 --- a/packages/client/connection/tests/connection.client.spec.ts +++ b/packages/client/connection/tests/connection.client.spec.ts @@ -192,7 +192,7 @@ describe('connection lifecycle', () => { const controller = new ConnectionController(api, (signal, ready) => { sourceCalls++ if (sourceCalls === 1) return fail() - ready() + ready({ home: '/h' }) return new Promise((resolve) => { signal.addEventListener('abort', () => { resolve() }, { once: true }) }) diff --git a/packages/client/connection/tests/fake-api.client.ts b/packages/client/connection/tests/fake-api.client.ts index bc546ad185..639f52f2b9 100644 --- a/packages/client/connection/tests/fake-api.client.ts +++ b/packages/client/connection/tests/fake-api.client.ts @@ -95,7 +95,10 @@ export class FakeApiClient implements IApiClient { return response } - private async openGeneration(signal: AbortSignal, onOpen: () => void): Promise { + private async openGeneration( + signal: AbortSignal, + onOpen: (host: { readonly home: string }) => void, + ): Promise { const inbox: StreamItem[] = [] let wake: (() => void) | null = null const conn: StreamConn = { @@ -105,8 +108,9 @@ export class FakeApiClient implements IApiClient { }, } this.generationConns.push(conn) - if (this.holdGenerationReady) this.heldOpens.push(onOpen) - else if (!this.suppressGenerationReady) onOpen() + const ready = (): void => { onOpen({ home: '/h' }) } + if (this.holdGenerationReady) this.heldOpens.push(ready) + else if (!this.suppressGenerationReady) ready() try { while (!signal.aborted) { while (inbox.length > 0) { diff --git a/packages/client/connection/tests/generation.client.spec.ts b/packages/client/connection/tests/generation.client.spec.ts new file mode 100644 index 0000000000..0deace5dba --- /dev/null +++ b/packages/client/connection/tests/generation.client.spec.ts @@ -0,0 +1,65 @@ +import { Context } from '@deepseek-ai/cordis' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + apply, + type ConnectionGenerationSource, + type ConnectionHandle, +} from '../src/client/index.ts' + +type BrowserGlobal = { + location?: { hostname: string; search: string } +} + +const contexts = new Set() + +afterEach(async () => { + vi.restoreAllMocks() + delete (globalThis as BrowserGlobal).location + await Promise.all([...contexts].map(async ctx => ctx.fiber.dispose())) + contexts.clear() +}) + +async function mount(): Promise { + ;(globalThis as BrowserGlobal).location = { hostname: 'localhost', search: '?fixture' } + const ctx = new Context() + contexts.add(ctx) + await ctx.plugin({ apply, inject: [] }) + const connection = ctx.get('connection') as ConnectionHandle | undefined + if (connection === undefined) throw new Error('fixture did not provide Connection') + return connection +} + +describe('Connection generation facts', () => { + it('publishes ready-frame Host facts and retracts them when the loop stops', async () => { + const connection = await mount() + const source: ConnectionGenerationSource = (signal, ready) => { + ready({ home: '/home/from-ready' }) + return new Promise((resolve) => { + if (signal.aborted) resolve() + else signal.addEventListener('abort', () => { resolve() }, { once: true }) + }) + } + connection.registerGenerationSource(source) + const seen: Array = [] + const stopListening = connection.generation.subscribe(() => { + seen.push(connection.generation.getSnapshot()?.host.home) + }) + const loop = connection.start({}, { + backoffBaseMs: 1, + backoffFactor: 1, + backoffMaxMs: 1, + generationReadyTimeoutMs: 100, + }) + + await vi.waitFor(() => { + expect(connection.generation.getSnapshot()).toEqual({ + id: 1, + host: { home: '/home/from-ready' }, + }) + }) + loop.stop() + expect(connection.generation.getSnapshot()).toBeUndefined() + expect(seen).toEqual(['/home/from-ready', undefined]) + stopListening() + }) +})