fix(bridge): keep the extension connected on dsh 0.2.1 (Peer is mandatory)

packages/browser/bridge-browser/src/remote-host-api.ts

@@ -44,6 +44,8 @@ export interface TypertGatewayLike {
     readonly method: string
     readonly args: Readonly<Record<string, unknown>>
     readonly signal?: AbortSignal
+    /** dsh 0.2.1+: the Peer the call speaks for (mandatory since 0.2.1). */
+    readonly peer?: unknown
   }): Promise<unknown>
 }
 
@@ -65,19 +67,44 @@ const EMPTY_WIRE_UPLINK: AsyncIterable<unknown> = {
  *   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<AsyncIterable<unknown>> {
+function openWireStream(
+  gateway: TypertGatewayLike,
+  endpoint: string,
+  payload: unknown,
+  signal: AbortSignal,
+  peer?: unknown,
+): Promise<AsyncIterable<unknown>> {
   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<Response>
   }
+  /** dsh 0.2.1+: owner Peer admitted for browser-originated traffic. */
+  readonly browserPeer?: unknown
+  /** dsh 0.2.1+: in-process Peer factory used when `browserPeer` is absent. */
+  createPeer?(options: { principal: { kind: 'owner' } }): unknown
+}
+
+/**
+ * Resolve the Peer dsh 0.2.1 requires on every Gateway call.
+ * dsh 0.2.0 tolerates `undefined`; 0.2.1 dereferences `peer.signal`
+ * (`assertPeerLive`) and fails the stream with "Cannot read properties of
+ * undefined (reading 'signal')" without it.
+ */
+function resolvePeer(connection: HostConnectionLike): unknown {
+  if (connection.browserPeer !== undefined) return connection.browserPeer
+  try {
+    return connection.createPeer?.({ principal: { kind: 'owner' } })
+  } catch {
+    return undefined
+  }
 }
 
 interface InvokeTarget {
@@ -116,12 +143,15 @@ class RemoteHostApi implements BrowserHostApi {
   /** Last known session/follow tip per Session; drives session/page throughSeq. */
   private readonly historyCursors = new Map<string, number>()
   private activeEvents: EventGeneration | undefined
+  /** Peer every Gateway call speaks for; dsh 0.2.1 requires it explicitly. */
+  private readonly peer: unknown
 
   constructor(
     private readonly gateway: TypertGatewayLike,
     connection: HostConnectionLike,
   ) {
     this.fetchHandler = connection.createSharedFetchHandler('/api')
+    this.peer = resolvePeer(connection)
   }
 
   async call(call: HostRpcCall): Promise<HostRpcResult> {
@@ -145,6 +175,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 +198,7 @@ class RemoteHostApi implements BrowserHostApi {
       this.extensionSessions,
       this.noteHistoryCursor.bind(this),
       signal,
+      this.peer,
     )
     const previous = this.activeEvents
     this.activeEvents = generation
@@ -212,12 +244,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 +265,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 +288,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 +310,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()
@@ -380,6 +413,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 +488,7 @@ class EventGeneration {
           },
         },
         signal,
+        this.peer,
       )
       const iterator = source[Symbol.asyncIterator]()
       const first = await iterator.next()
@@ -549,7 +584,7 @@ class EventGeneration {
 
   private async pumpRemoteEvents(): Promise<void> {
     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 +774,7 @@ async function oneShotSessionSnapshot(
   sessionId: string,
   outerSignal: AbortSignal,
   maxMessages?: number,
+  peer?: unknown,
 ): Promise<SessionSnapshot> {
   const controller = new AbortController()
   const signal = AbortSignal.any([outerSignal, controller.signal])
@@ -754,6 +790,7 @@ async function oneShotSessionSnapshot(
       },
     },
     signal,
+    peer,
   )
   const iterator = source[Symbol.asyncIterator]()
   try {

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<Response>
+  connection?: Partial<HostConnectionLike>
 } = {}) {
   const invoke = vi.fn(options.invoke ?? (async () => ({ accepted: true })))
   const open = vi.fn(options.open ?? (async (_endpoint, _payload, _uplink, _peer, signal) => ({
@@ -46,6 +47,7 @@ function harness(options: {
   }
   const connection: HostConnectionLike = {
     createSharedFetchHandler: () => ({ fetch }),
+    ...options.connection,
   }
   return { api: createRemoteHostApi(gateway, connection), invoke, open, fetch, requestBodies }
 }
@@ -781,3 +783,48 @@ describe('dsh 0.2 Remote Host adapter', () => {
     await expect(pending).resolves.toEqual({ done: true, value: undefined })
   })
 })
+
+describe('dsh 0.2.1 Peer scope', () => {
+  const browserPeer = { id: 'peer-browser', principal: { kind: 'owner' }, signal: new AbortController().signal }
+
+  it('names the Connection browser Peer on every Gateway call', async () => {
+    const open = vi.fn(async (_endpoint: string, _payload: unknown, _uplink: unknown, _peer: unknown, signal: AbortSignal) => ({
+      async *[Symbol.asyncIterator]() {
+        yield { type: 'ready', clientId: 'test', host: {} }
+        await abortWait(signal)
+      },
+    }))
+    const invoke = vi.fn(async () => ({ default: null, routableProviders: [], groups: [], failures: [] }))
+    const { api } = harness({ connection: { browserPeer }, open, invoke })
+    const controller = new AbortController()
+    const pending = api.events(controller.signal)[Symbol.asyncIterator]().next()
+    await vi.waitFor(() => expect(open).toHaveBeenCalled())
+    await api.call(call('session.models', {}))
+    controller.abort()
+    await pending.catch(() => undefined)
+
+    // dsh 0.2.1 made `peer` mandatory: Gateway dereferences peer.signal on open.
+    expect(open.mock.calls[0]?.[3]).toBe(browserPeer)
+    expect(invoke.mock.calls[0]?.[0]).toMatchObject({ peer: browserPeer })
+  })
+
+  it('mints an owner Peer when the Connection exposes only createPeer', async () => {
+    const created = { id: 'peer-created', principal: { kind: 'owner' }, signal: new AbortController().signal }
+    const createPeer = vi.fn(() => created)
+    const invoke = vi.fn(async () => ({ default: null, routableProviders: [], groups: [], failures: [] }))
+    const { api } = harness({ connection: { createPeer }, invoke })
+    await api.call(call('session.models', {}))
+
+    expect(createPeer).toHaveBeenCalledWith({ principal: { kind: 'owner' } })
+    expect(invoke.mock.calls[0]?.[0]).toMatchObject({ peer: created })
+  })
+
+  it('leaves the Peer slot undefined on runtimes that never ask for one', async () => {
+    const invoke = vi.fn(async () => ({ default: null, routableProviders: [], groups: [], failures: [] }))
+    const { api } = harness({ invoke })
+    await api.call(call('session.models', {}))
+
+    // dsh 0.2.0 tolerates an absent Peer; keep omitting it there.
+    expect(invoke.mock.calls[0]?.[0].peer).toBeUndefined()
+  })
+})