From a1473ddec67900612a34b8384056631d8fd2d01f Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Sun, 25 Oct 2026 18:14:17 +0800 Subject: [PATCH] feat(gateway): expose pending remote events Session state reads sample Gateway pending events; the Gateway contract now provides getPendingRemoteEvents() for that read. --- packages/api/gateway/src/index.ts | 10 ++++++++++ packages/api/gateway/src/types.ts | 11 ++++++++++- .../api/gateway/tests/gateway-access.host.spec.ts | 5 +++++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/packages/api/gateway/src/index.ts b/packages/api/gateway/src/index.ts index 878c77cf5e..4fe2652671 100644 --- a/packages/api/gateway/src/index.ts +++ b/packages/api/gateway/src/index.ts @@ -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. diff --git a/packages/api/gateway/src/types.ts b/packages/api/gateway/src/types.ts index 83ae7f0730..b802aba061 100644 --- a/packages/api/gateway/src/types.ts +++ b/packages/api/gateway/src/types.ts @@ -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. diff --git a/packages/api/gateway/tests/gateway-access.host.spec.ts b/packages/api/gateway/tests/gateway-access.host.spec.ts index 53727804ee..346f50a8be 100644 --- a/packages/api/gateway/tests/gateway-access.host.spec.ts +++ b/packages/api/gateway/tests/gateway-access.host.spec.ts @@ -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'] } })