refactor(connection): carry Host facts with generations

This commit is contained in:
imccyu
2026-08-27 22:26:28 +08:00
parent 17c03bbbcc
commit e036aae7c0
24 changed files with 298 additions and 106 deletions
@@ -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<void> {
private async pumpEvents(
signal: AbortSignal,
ready: (host: ConnectionHostInfo) => void,
): Promise<void> {
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. */
@@ -48,7 +48,7 @@ export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>>
* @param options - domain stream opener, end classification, and diagnostics.
*/
constructor(
private readonly connection: Pick<ConnectionHandle, 'hostDescription'>,
private readonly connection: Pick<ConnectionHandle, 'generation'>,
private readonly options: RemoteStreamOptions<Item>,
) {}
@@ -157,13 +157,13 @@ export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>>
}
async function waitForRemoteStreamRetry(
connection: Pick<ConnectionHandle, 'hostDescription'>,
connection: Pick<ConnectionHandle, 'generation'>,
error: RemoteStreamCarrierError,
attempt: number,
signal: AbortSignal,
): Promise<void> {
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 })
+10 -3
View File
@@ -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<void>
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<void> {
registerRemoteEvents(
source: TypertRemoteEventSource,
host: RemoteEventHostInfo,
): () => Promise<void> {
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)
@@ -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. */
+6 -1
View File
@@ -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<void>
registerRemoteEvents(
source: TypertRemoteEventSource,
host: RemoteEventHostInfo,
): () => Promise<void>
/**
* Invoke one live Remote method without assuming a carrier or response envelope.
@@ -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<ConnectionHandle, 'hostDescription'>
connection: Pick<ConnectionHandle, 'generation'>
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<Item>(generations: Generation<Item>[], opened?: () => void) {
}
function supervisor<Item>(
connection: Pick<ConnectionHandle, 'hostDescription'>,
connection: Pick<ConnectionHandle, 'generation'>,
generations: Generation<Item>[],
carrierFailed?: (error: RemoteStreamCarrierError) => void,
): RemoteStream<Item> {
@@ -122,8 +116,8 @@ describe('RemoteStream', () => {
let listener: (() => void) | undefined
const subscribed = Promise.withResolvers<undefined>()
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<string> } = {}
let subscriptions = 0
const connection = {
hostDescription: {
generation: {
getSnapshot: () => undefined,
subscribe: () => {
subscriptions++
@@ -36,6 +36,7 @@ vi.mock('node:crypto', async (importOriginal) => {
const randomUuid = vi.mocked(randomUUID)
const browserCookies = new WeakMap<Context, string>()
const REMOTE_HOST = { home: '/home/fixture' } as const
type AgentWireId = TypertContextWire<TypertContextMap['agent']>
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<typeof randomUUID>
const secondId = '00000000-0000-4000-8000-000000000012' as ReturnType<typeof randomUUID>
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<never> {})()
})
}, REMOTE_HOST)
const invalidPayloads: readonly unknown[] = [
null,
[],
@@ -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<ConnectionHandle['rpc']['open']> = () => (async function *() {
@@ -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')
@@ -42,10 +42,8 @@ interface Generation {
type PageSource = Page | Promise<Page> | ((signal: AbortSignal) => Promise<Page>)
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: () => () => {},
},
}
+2 -1
View File
@@ -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',
)
}
@@ -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<void>
registerRemoteEvents(
source: TypertRemoteEventSource,
host: RemoteEventHostInfo,
): () => Promise<void>
}
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()
@@ -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),
@@ -64,6 +64,24 @@ async function mount(initialHost?: HostDescription): Promise<Bench> {
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')),
},
@@ -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: () => () => {},
},
}
@@ -28,10 +28,8 @@ type SessionTransportRemote = Pick<SessionRemote, 'control' | 'follow' | 'page'>
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: () => () => {},
},
}
@@ -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<ConnectionHandle, 'hostDescription'> = AVAILABLE_CONNECTION,
connection: Pick<ConnectionHandle, 'generation'> = 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')),
},
@@ -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<void>
/**
@@ -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<void>((resolve, reject) => {
const ready = new Promise<ConnectionHostInfo>((resolve, reject) => {
resolveReady = resolve
rejectReady = reject
})
const sourceLost = new Promise<never>((_resolve, reject) => {
rejectSourceLost = reject
})
const reportReady = (): void => {
const reportReady = (host: ConnectionHostInfo): void => {
if (sourceReady) return
sourceReady = true
resolveReady()
resolveReady(host)
}
const failed = new Promise<void>((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<void>, timeoutMs: number, signal: AbortSignal): Promise<void> {
return new Promise<void>((resolve, reject) => {
function waitForReady<T>(ready: Promise<T>, timeoutMs: number, signal: AbortSignal): Promise<T> {
return new Promise<T>((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 })
},
)
})
@@ -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)
+50 -5
View File
@@ -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)
},
@@ -37,7 +37,7 @@ class GenerationProbe {
}
this.active.add(finish)
signal.addEventListener('abort', finish, { once: true })
ready()
ready({ home: '/h' })
if (signal.aborted) finish()
})
@@ -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<void>((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
@@ -95,7 +95,10 @@ export class FakeApiClient implements IApiClient {
return response
}
private async openGeneration(signal: AbortSignal, onOpen: () => void): Promise<void> {
private async openGeneration(
signal: AbortSignal,
onOpen: (host: { readonly home: string }) => void,
): Promise<void> {
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) {
@@ -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<Context>()
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<ConnectionHandle> {
;(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<void>((resolve) => {
if (signal.aborted) resolve()
else signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}
connection.registerGenerationSource(source)
const seen: Array<string | undefined> = []
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()
})
})