From 5049280bb6addfd59bd96f51870f7b654296451c Mon Sep 17 00:00:00 2001 From: flyingbug2 Date: Fri, 23 Oct 2026 19:49:49 +0800 Subject: [PATCH] fix(bridge): speak for an in-process owner Peer on dsh 0.2.1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dsh 0.2.1 makes the Peer a hard requirement on every Gateway entry: `openWireStream`, `invoke`/`prepareInvocation` and `$events/result` all read `peer.signal`, so an undefined Peer throws `TypeError: Cannot read properties of undefined (reading 'signal')`. The bridge then closes the extension socket with 1011 right after `hello.ok` (the side panel reconnects forever), and every RPC — `workspace.create`, `session.create/list/page` — fails with `gateway/internal`. Mint one in-process owner Peer from the Host Connection service (`createPeer`, exposed from 0.2.1) and use it for wire streams, `invoke` requests and the `$events/result` fetch, so the answering Peer is the very object the stream was opened with. Kernels without `createPeer` (0.2.0 and earlier) keep the previous behaviour. Refs: browser-context V4 source kind is already handled on main. --- .../bridge-browser/src/remote-host-api.ts | 70 ++++++++++++-- .../tests/remote-host-api.spec.ts | 91 +++++++++++++++++++ 2 files changed, 151 insertions(+), 10 deletions(-) diff --git a/packages/browser/bridge-browser/src/remote-host-api.ts b/packages/browser/bridge-browser/src/remote-host-api.ts index e620d517..ac8da592 100644 --- a/packages/browser/bridge-browser/src/remote-host-api.ts +++ b/packages/browser/bridge-browser/src/remote-host-api.ts @@ -44,9 +44,44 @@ export interface TypertGatewayLike { readonly method: string readonly args: Readonly> readonly signal?: AbortSignal + /** Peer the invocation speaks for; dsh 0.2.1 requires it on every Gateway entry. */ + readonly peer?: unknown }): Promise } +/** + * An in-process owner Peer minted by the Host Connection service. + * + * dsh 0.2.1 makes the Peer a hard requirement on every Gateway entry: + * `openWireStream`, `invoke`/`prepareInvocation`, and `$events/result` all read + * `peer.signal`, so an undefined Peer throws + * `TypeError: Cannot read properties of undefined (reading 'signal')` — the + * bridge then closes the extension socket with 1011 right after `hello.ok`. + * `$events` additionally requires the Peer that posts `$events/result` to be + * the very object the stream was opened with, so one Peer serves both + * directions. Earlier kernels expose no `createPeer` and need none. + */ +interface PeerMint { + readonly principal: { readonly kind: 'owner' } + readonly interactions?: 'answer' | 'none' +} + +/** + * Mint the in-process Peer this bridge speaks for, or `undefined` when the + * active kernel predates peer-aware Gateway admission. + * @param connection - Host Connection service read from the plugin Context. + * @returns the Peer to bind, or `undefined` for pre-0.2.1 kernels. + */ +function createBridgePeer(connection: HostConnectionLike): unknown { + const createPeer = connection.createPeer + if (typeof createPeer !== 'function') return undefined + try { + return createPeer.call(connection, { principal: { kind: 'owner' }, interactions: 'answer' }) + } catch { + return undefined + } +} + /** * Empty Client→Host uplink for in-process Host wireStream.open calls. * dsh 0.2 requires the uplink slot; Gateway-owned endpoints ($events) discard @@ -65,19 +100,24 @@ const EMPTY_WIRE_UPLINK: AsyncIterable = { * these as three-arg maps AbortSignal onto uplink and leaves signal * undefined (hello.ok → stream-failed → WS 1011). */ -function openWireStream(gateway: TypertGatewayLike, endpoint: string, payload: unknown, signal: AbortSignal): Promise> { +function openWireStream(gateway: TypertGatewayLike, endpoint: string, payload: unknown, signal: AbortSignal, peer?: unknown): Promise> { const open = gateway.wireStream.open if (open.length === 3) { return open(endpoint, payload, signal) } - return open(endpoint, payload, EMPTY_WIRE_UPLINK, undefined, signal) + return open(endpoint, payload, EMPTY_WIRE_UPLINK, peer, signal) } /** Structural subset of dsh 0.2's Host Connection service. */ export interface HostConnectionLike { createSharedFetchHandler(channel: '/api'): { - fetch(request: Request): Promise + fetch(request: Request, peer?: unknown): Promise } + /** + * Mint one in-process Peer owned by the Context that read the service. + * Absent on kernels that predate peer-aware Gateway admission. + */ + createPeer?(options: PeerMint): unknown } interface InvokeTarget { @@ -112,6 +152,8 @@ export function createRemoteHostApi( class RemoteHostApi implements BrowserHostApi { private readonly fetchHandler: ReturnType + /** Peer every Gateway call and wire stream speaks for; `$events/result` must answer as it. */ + private readonly peer: unknown private readonly extensionSessions = new ExtensionSessionRegistry() /** Last known session/follow tip per Session; drives session/page throughSeq. */ private readonly historyCursors = new Map() @@ -121,6 +163,7 @@ class RemoteHostApi implements BrowserHostApi { private readonly gateway: TypertGatewayLike, connection: HostConnectionLike, ) { + this.peer = createBridgePeer(connection) this.fetchHandler = connection.createSharedFetchHandler('/api') } @@ -145,6 +188,7 @@ class RemoteHostApi implements BrowserHostApi { method: target.method, args: target.args, signal: call.signal, + peer: this.peer, }) // Only claim ownership after a successful create/prompt. A failed prompt // against a Desktop session must not steal later ask_user_question away @@ -167,6 +211,7 @@ class RemoteHostApi implements BrowserHostApi { this.extensionSessions, this.noteHistoryCursor.bind(this), signal, + this.peer, ) const previous = this.activeEvents this.activeEvents = generation @@ -212,12 +257,13 @@ class RemoteHostApi implements BrowserHostApi { }, }, signal: call.signal, + peer: this.peer, }) return { ok: true, value: historyPageValue(page) } } const snapshot = this.activeEvents === undefined - ? await oneShotSessionSnapshot(this.gateway, sessionId, call.signal, maxMessages) + ? await oneShotSessionSnapshot(this.gateway, sessionId, call.signal, maxMessages, this.peer) : await this.activeEvents.openSessionHistory(sessionId, call.signal, maxMessages) this.noteHistoryCursor(sessionId, snapshot.cursor) return { ok: true, value: historyValue(snapshot) } @@ -232,11 +278,11 @@ class RemoteHostApi implements BrowserHostApi { const sessionId = sessionIdOf(call.payload) try { const catalog = await this.gateway.invoke({ - namespace: 'session', method: 'modelCatalog', args: {}, signal: call.signal, + namespace: 'session', method: 'modelCatalog', args: {}, signal: call.signal, peer: this.peer, }) // Provisional Sessions request only the catalog and overlay their pending choice. const projections = sessionId === undefined ? undefined : await this.gateway.invoke({ - namespace: 'session', method: 'projections', args: { request: { sessionId } }, signal: call.signal, + namespace: 'session', method: 'projections', args: { request: { sessionId } }, signal: call.signal, peer: this.peer, }) const values = isRecord(projections) ? projections.values : undefined const modelSelection = isRecord(values) ? values.modelSelection : undefined @@ -255,7 +301,7 @@ class RemoteHostApi implements BrowserHostApi { const cached = this.historyCursors.get(sessionId) if (cached !== undefined) return cached const snapshot = this.activeEvents === undefined - ? await oneShotSessionSnapshot(this.gateway, sessionId, signal) + ? await oneShotSessionSnapshot(this.gateway, sessionId, signal, undefined, this.peer) : await this.activeEvents.openSessionHistory(sessionId, signal) this.noteHistoryCursor(sessionId, snapshot.cursor) const throughSeq = this.historyCursors.get(sessionId) @@ -277,7 +323,7 @@ class RemoteHostApi implements BrowserHostApi { try { const controller = new AbortController() const signal = AbortSignal.any([call.signal, controller.signal]) - const source = await openWireStream(this.gateway, 'workspace/follow', { args: {} }, signal) + const source = await openWireStream(this.gateway, 'workspace/follow', { args: {} }, signal, this.peer) const iterator = source[Symbol.asyncIterator]() try { const first = await iterator.next() @@ -320,7 +366,7 @@ class RemoteHostApi implements BrowserHostApi { }), signal, }) - const response = await this.fetchHandler.fetch(request) + const response = await this.fetchHandler.fetch(request, this.peer) if (!response.ok) { throw new Error(`$events/result transport failed with HTTP ${String(response.status)}: ${await response.text()}`) } @@ -380,6 +426,7 @@ class EventGeneration { private readonly extensionSessions: ExtensionSessionRegistry, private readonly onHistoryCursor: (sessionId: string, cursor: number) => void, outerSignal: AbortSignal, + private readonly peer?: unknown, ) { this.signal = AbortSignal.any([outerSignal, this.lifetime.signal]) } @@ -454,6 +501,7 @@ class EventGeneration { }, }, signal, + this.peer, ) const iterator = source[Symbol.asyncIterator]() const first = await iterator.next() @@ -549,7 +597,7 @@ class EventGeneration { private async pumpRemoteEvents(): Promise { try { - const source = await openWireStream(this.gateway, '$events', { args: {} }, this.signal) + const source = await openWireStream(this.gateway, '$events', { args: {} }, this.signal, this.peer) let ready = false for await (const value of source) { if (!ready) { @@ -739,6 +787,7 @@ async function oneShotSessionSnapshot( sessionId: string, outerSignal: AbortSignal, maxMessages?: number, + peer?: unknown, ): Promise { const controller = new AbortController() const signal = AbortSignal.any([outerSignal, controller.signal]) @@ -754,6 +803,7 @@ async function oneShotSessionSnapshot( }, }, signal, + peer, ) const iterator = source[Symbol.asyncIterator]() try { diff --git a/packages/browser/bridge-browser/tests/remote-host-api.spec.ts b/packages/browser/bridge-browser/tests/remote-host-api.spec.ts index 471e1c82..67608db2 100644 --- a/packages/browser/bridge-browser/tests/remote-host-api.spec.ts +++ b/packages/browser/bridge-browser/tests/remote-host-api.spec.ts @@ -19,6 +19,7 @@ function harness(options: { invoke?: TypertGatewayLike['invoke'] open?: TypertGatewayLike['wireStream']['open'] fetch?: (request: Request) => Promise + createPeer?: HostConnectionLike['createPeer'] } = {}) { const invoke = vi.fn(options.invoke ?? (async () => ({ accepted: true }))) const open = vi.fn(options.open ?? (async (_endpoint, _payload, _uplink, _peer, signal) => ({ @@ -46,10 +47,100 @@ function harness(options: { } const connection: HostConnectionLike = { createSharedFetchHandler: () => ({ fetch }), + ...(options.createPeer === undefined ? {} : { createPeer: options.createPeer }), } return { api: createRemoteHostApi(gateway, connection), invoke, open, fetch, requestBodies } } +describe('dsh 0.2.1 in-process owner Peer', () => { + const peer = { + id: 'peer-owner', + interactions: 'answer', + principal: { kind: 'owner' }, + signal: new AbortController().signal, + } + + it('mints one owner Peer for wire streams, invoke, and $events/result', async () => { + let releaseCancel: (() => void) | undefined + const { api, open, invoke, fetch } = harness({ + createPeer: () => peer, + open: async (endpoint, _payload, _uplink, _peer, signal) => { + if (endpoint === 'workspace/follow') { + return { + async *[Symbol.asyncIterator]() { + yield { type: 'baseline', value: { items: [] } } + await abortWait(signal) + }, + } + } + if (endpoint !== '$events') throw new Error(endpoint) + return { + async *[Symbol.asyncIterator]() { + yield { type: 'ready', clientId: 'client-peer', host: { home: '/home/test' } } + yield { + type: 'waterfall', + event: 'user-questions/request', + eventId: 'question-peer', + agentId: 'session-1', + request: { questions: [{ id: 'db', question: 'Database?' }] }, + } + await new Promise((resolve) => { releaseCancel = resolve }) + await abortWait(signal) + }, + } + }, + }) + + await api.call(call('workspace.list', {})) + expect(open.mock.calls[0]?.[3]).toBe(peer) + + await api.call(call('session.models', {})) + expect(invoke.mock.calls[0]?.[0]).toMatchObject({ peer }) + + await expect(api.call(call('session.prompt', { + sessionId: 'session-1', + mode: 'queue', + content: [], + }, 'prompt-peer'))).resolves.toEqual({ ok: true, value: { accepted: true } }) + + const abort = new AbortController() + const events = api.events(abort.signal)[Symbol.asyncIterator]() + await expect(events.next()).resolves.toMatchObject({ + value: { rpcId: 'question-peer', method: 'question/requested' }, + }) + expect(open).toHaveBeenCalledWith('$events', { args: {} }, expect.anything(), peer, expect.any(AbortSignal)) + + await api.respond('question-peer', { + ok: true, + value: { + sessionId: 'session-1', + answer: { answers: [{ id: 'db', selected: ['SQLite'] }] }, + }, + }, abort.signal) + expect(fetch.mock.calls[0]?.[1]).toBe(peer) + + releaseCancel?.() + abort.abort() + await events.return?.() + }) + + it('leaves the Peer slot undefined on kernels that expose no createPeer', async () => { + const { api, open, invoke } = harness({ + open: async (_endpoint, _payload, _uplink, _peer, _signal) => ({ + async *[Symbol.asyncIterator]() { + yield { type: 'baseline', value: { items: [] } } + }, + }), + }) + + await api.call(call('workspace.list', {})) + expect(open.mock.calls[0]?.[3]).toBeUndefined() + + await api.call(call('session.models', {})) + expect(invoke.mock.calls[0]?.[0]).toMatchObject({ peer: undefined }) + }) +}) + describe('dsh 0.2 Remote Host adapter', () => { it.each([ { next: { provider: 'session-provider', model: 'saved-model', reasoningEffort: 'high' }, routable: true },