feat(gateway): expose pending remote events

Session state reads sample Gateway pending events; the Gateway contract now
provides getPendingRemoteEvents() for that read.
This commit is contained in:
imccyu
2026-10-25 18:14:17 +08:00
parent 7bc30c4cba
commit a1473ddec6
3 changed files with 25 additions and 1 deletions
+10
View File
@@ -36,6 +36,7 @@ import {
} from '@deepseek-ai/dsh-typert-protocol'
import type {
InvokeRemoteRequest,
PendingRemoteEventInfo,
TypertGateway,
TypertGatewayErrorCode,
TypertGatewayWireStream,
@@ -70,6 +71,7 @@ import {
export type {
InvokeRemoteRequest,
PendingRemoteEventInfo,
TypertGateway,
TypertGatewayErrorCode,
TypertGatewayWireStream,
@@ -333,6 +335,14 @@ export class TypertGatewayService extends Service implements TypertGateway {
return false
}
/**
* Inspect unsettled forwarded events without changing their lifetime.
* @returns detached event names and Agent identities.
*/
getPendingRemoteEvents(): readonly PendingRemoteEventInfo[] {
return [...this.pendingRemoteEvents.values()].map(({ frame }) => ({ event: frame.event, agentId: frame.agentId }))
}
/**
* Register the sole application-selected forwarded-event source.
* @param source - stream factory installed by the Remote assembly.
+10 -1
View File
@@ -5,7 +5,7 @@
import type { Context } from '@deepseek-ai/cordis'
import type { PeerScope, RemoteUplinkWindow, TypertResourceReader } from '@deepseek-ai/dsh-typert-protocol'
import type { RemoteEventHostInfo } from './stream-protocol.ts'
import type { RemoteEventAgentId, RemoteEventHostInfo } from './stream-protocol.ts'
/** One Remote method request after a carrier has decoded its envelope. */
export interface InvokeRemoteRequest {
@@ -144,6 +144,12 @@ export type TypertGatewayErrorCode =
| 'gateway/signature-invalid'
| 'gateway/uplink-overflow'
/** Public metadata of one unsettled forwarded Remote event. */
export interface PendingRemoteEventInfo {
readonly event: string
readonly agentId: RemoteEventAgentId
}
/** Host dispatcher consumed by Connection adapters. */
export interface TypertGateway {
/** Carrier adapter shared by WebSocket and in-process transports. */
@@ -155,6 +161,9 @@ export interface TypertGateway {
*/
hasLiveClient(): boolean
/** @returns detached metadata for all unsettled forwarded events in arrival order. */
getPendingRemoteEvents(): readonly PendingRemoteEventInfo[]
/**
* Register the application-selected forwarded-event source.
* @param source - stream factory installed by the Remote assembly.
@@ -730,6 +730,7 @@ describe('Gateway forwarded events for non-owner Peers', () => {
await expect(ownerEvents.next()).resolves.toMatchObject({ value: { type: 'ready', host: { home: '/fixture' } } })
expect(ctx.typertGateway.hasLiveClient()).toBe(true)
expect(ctx.typertGateway.getPendingRemoteEvents()).toEqual([])
const agentCtx = ctx.extend({})
const subject = {}
const settled = vi.fn()
@@ -745,6 +746,10 @@ describe('Gateway forwarded events for non-owner Peers', () => {
push({ event: 'fixture/shelf', args: ['s2'], resources })
push({ event: 'fixture/shelf', args: ['s1'], resources })
await expect(ownerEvents.next()).resolves.toMatchObject({ value: { type: 'waterfall', event: 'fixture/ask' } })
const pending = ctx.typertGateway.getPendingRemoteEvents()
expect(pending).toEqual([{ event: 'fixture/ask', agentId: 'agent-1' }])
expect(pending).not.toBe(ctx.typertGateway.getPendingRemoteEvents())
expect(pending[0]).not.toBe(ctx.typertGateway.getPendingRemoteEvents()[0])
await expect(ownerEvents.next()).resolves.toMatchObject({ value: { type: 'emit', event: 'fixture/notice' } })
await expect(ownerEvents.next()).resolves.toMatchObject({ value: { type: 'emit', event: 'fixture/shelf', args: ['s2'] } })
await expect(ownerEvents.next()).resolves.toMatchObject({ value: { type: 'emit', event: 'fixture/shelf', args: ['s1'] } })