feat(typert): propagate Remote cancellation

2026-08-06 18:13:15 +08:00
parent 9b63d72c94
commit 22bec5e63f
28 changed files with 280 additions and 66 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/host/api-gateway/README.md
-README.md: cc80bb19fec15414aa0857154a8a36fb4f642672
-README.zh.md: 6febb1cfe4fc7fa4c5a17e1e4f6a21e2ee03e295
+README.md: 9cb6e7e1c0a23789ab4ab2c999b5a6c2d4cd32f9
+README.zh.md: 609580ceb77649ba8df6103093a72092c9ccc8a1
@@ -12,11 +12,13 @@ Strict mode reads generated invocation descriptors from `ctx.typert.local`. Look

 The Host entry registers a trusted-host interceptor on Connection's shared `/api` FetchHandler. Connection passes this composite handler through its HTTP bridge; the handler dispatches claimed endpoints to Gateway and unclaimed endpoints to API Proxy. Direct `invoke()` calls preserve business errors; `TypertGatewayError` distinguishes failures owned by dispatch, binding, providers, lookup, Context, arguments, and codecs.

+A cancellation-aware Remote method declares `signal: AbortSignal` as its final Host parameter. The signal is descriptor metadata rather than a wire argument: Connection supplies it to the Gateway, and the Gateway injects it after decoded business parameters. SRC recognizes the reserved final name, while strict generation additionally requires the global `AbortSignal` type.
+
 ## Client service: `ClientApi` (ctx key: `api`)

 `ctx.api.mount()` validates and registers a generated Host-for-Client contribution, then installs concrete direct and scoped methods for the calling Cordis fiber. Duplicate endpoints, namespace collisions, and descriptors without strict generated codecs fail before methods become callable.

-Each call validates positional inputs, constructs the descriptor's exact named `args`, and sends it through `ctx.connection.rpc.call('/api', endpoint, ...)`. The returned value is validated before reaching application code. Withdrawing a contribution removes its descriptors and methods together, aborts in-flight calls, and makes retained method handles reject.
+Each call validates positional inputs, constructs the descriptor's exact named `args`, and sends it through `ctx.connection.rpc.call('/api', endpoint, ...)`. Generated cancellation-aware methods accept a final optional `AbortSignal`; the Client combines it with the contribution mount lifetime before calling Connection. The returned value is validated before reaching application code. Withdrawing a contribution removes its descriptors and methods together, aborts in-flight calls, and makes retained method handles reject.

 Generated declaration merges provide the TypeScript API. The Client entry contains no Host Service or Host Cordis interface merge, and method lookup and invocation use ordinary objects and functions rather than a JavaScript Proxy.

@@ -12,11 +12,13 @@

 Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandler 上注册 trusted-host interceptor。Connection 把这个复合 handler 交给 HTTP bridge；handler 将已认领 endpoint 分发给 Gateway，未认领 endpoint 则交给 API Proxy。直接调用 `invoke()` 会保留业务错误；`TypertGatewayError` 可区分分发、绑定、提供方、查找、Context、参数和编解码器各自负责的故障。

+支持取消的 Remote 方法会把 `signal: AbortSignal` 声明为最后一个 Host 参数。signal 是 descriptor 元数据，而不是 wire 参数：Connection 将它提供给 Gateway，Gateway 则在已解码的业务参数之后注入它。SRC 识别这个保留的末位参数名，严格生成还要求它具有全局 `AbortSignal` 类型。
+
 ## Client 服务：`ClientApi`（ctx key：`api`）

 `ctx.api.mount()` 会校验并注册生成的 Host-for-Client 贡献项，然后为发起调用的 Cordis fiber 安装具体的直接方法和作用域方法。重复端点、命名空间冲突，以及缺少生成的严格编解码器的描述符，都会在方法可调用前报错。

-每次调用都会校验位置参数，构造与描述符完全匹配的具名 `args`，再通过 `ctx.connection.rpc.call('/api', endpoint, ...)` 发送。返回值经过校验后才会交给应用代码。撤回贡献项会同时移除其描述符和方法、中止正在进行的调用，并使外部仍持有的方法句柄在调用时返回拒绝。
+每次调用都会校验位置参数，构造与描述符完全匹配的具名 `args`，再通过 `ctx.connection.rpc.call('/api', endpoint, ...)` 发送。生成的支持取消的方法接受最后一个可选 `AbortSignal`；Client 会在调用 Connection 前将它与贡献项的挂载生命周期合并。返回值经过校验后才会交给应用代码。撤回贡献项会同时移除其描述符和方法、中止正在进行的调用，并使外部仍持有的方法句柄在调用时返回拒绝。

 生成的声明合并提供 TypeScript API。Client 入口不包含 Host 服务或 Host Cordis 接口合并；方法查找和调用使用普通对象与函数，而不使用 JavaScript Proxy。

@@ -223,9 +223,13 @@ class ClientApiService extends Service implements ClientApi {
    const endpoint = endpointOf(descriptor)
    if (!token.active) throw new Error(`client api: Remote method ${endpoint} is no longer mounted`)
    const expected = descriptor.parameters.length - (projection?.parameterIndex === undefined ? 0 : 1)
-    if (values.length !== expected) {
+    const hasCallerSignal = descriptor.cancellation !== undefined && values.length === expected + 1
+    if (values.length !== expected && !hasCallerSignal) {
+      const contract = descriptor.cancellation === undefined
+        ? `${String(expected)} argument(s)`
+        : `${String(expected)} business argument(s) plus an optional AbortSignal`
      throw new Error(
-        `client api: ${endpoint} expected ${String(expected)} argument(s), got ${String(values.length)}`,
+        `client api: ${endpoint} expected ${contract}, got ${String(values.length)}`,
      )
    }
    const args: Record<string, unknown> = {}
@@ -248,7 +252,11 @@ class ClientApiService extends Service implements ClientApi {
    })
    const connection = this.ownerCtx.get('connection') as ConnectionHandle | undefined
    if (connection === undefined) throw new Error(`client api: ${endpoint} has no active Connection`)
-    const result = await connection.rpc.call('/api', endpoint, { args }, token.abort.signal)
+    const callerSignal = hasCallerSignal ? values[expected] as AbortSignal | undefined : undefined
+    const signal = callerSignal === undefined
+      ? token.abort.signal
+      : AbortSignal.any([token.abort.signal, callerSignal])
+    const result = await connection.rpc.call('/api', endpoint, { args }, signal)
    if (!mountActive(token)) throw new Error(`client api: Remote method ${endpoint} was withdrawn during invocation`)
    if (!result.ok) throw remoteFailure(endpoint, result.error)
    return parse(descriptor.result, result.value, endpoint, 'result')
@@ -36,6 +36,7 @@ interface ResolvedBinding {
 }

 type ConnectionRpcResult = Awaited<ReturnType<ConnectionRpcHandler>>
+const NEVER_ABORTED_SIGNAL = new AbortController().signal

 /** Dispatch failure produced outside the invoked business method. */
 export class TypertGatewayError extends Error {
@@ -129,6 +130,7 @@ export class TypertGatewayService extends Service implements TypertGateway {
    }
    validateBinding(receiver, descriptor.service, descriptor.namespace, endpoint)
    const args = descriptor.parameters.map(parameter => this.resolveParameter(parameter, request.args, endpoint))
+    if (descriptor.cancellation !== undefined) args.push(request.signal ?? NEVER_ABORTED_SIGNAL)
    const implementation = descriptor.implementation ?? descriptor.method
    const method = Reflect.get(receiver, implementation) as unknown
    if (typeof method !== 'function') {
@@ -146,13 +148,12 @@ export class TypertGatewayService extends Service implements TypertGateway {
  private async dispatchRpc(
    endpoint: string,
    payload: unknown,
-    _signal: AbortSignal,
+    signal: AbortSignal,
  ): Promise<ConnectionRpcResult> {
-    // Remote methods have no cancellation parameter yet, so disconnects do not cancel business work.
-    return this.invokeRpc(endpoint, payload)
+    return this.invokeRpc(endpoint, payload, signal)
  }

-  private async invokeRpc(endpoint: string, payload: unknown): Promise<ConnectionRpcResult> {
+  private async invokeRpc(endpoint: string, payload: unknown, signal: AbortSignal): Promise<ConnectionRpcResult> {
    try {
      const segments = endpoint.split('/')
      if (segments.length !== 2 || segments[0] === '' || segments[1] === '') {
@@ -171,6 +172,7 @@ export class TypertGatewayService extends Service implements TypertGateway {
        namespace,
        method,
        args: payload.args,
+        signal,
      })
      return { ok: true, value }
    } catch (error) {
@@ -226,9 +228,22 @@ export class TypertGatewayService extends Service implements TypertGateway {
    endpoint: string,
  ): InvocationDescriptor {
    const names = methodParameterNames(binding.service, marker.method, endpoint)
+    const signalIndex = names.indexOf('signal')
+    if (signalIndex >= 0 && signalIndex !== names.length - 1) {
+      throw new TypertGatewayError(
+        'signature-invalid',
+        endpoint,
+        'SRC cancellation parameter signal must be the final parameter',
+        { field: 'signal' },
+      )
+    }
+    const cancellation = signalIndex >= 0
+      ? { parameter: 'signal' as const }
+      : undefined
+    const businessNames = cancellation === undefined ? names : names.slice(0, -1)
    const parameters: InvocationParameterDescriptor[] = []
    const wires = new Set<string>()
-    for (const name of names) {
+    for (const name of businessNames) {
      const matches = this.ctx.typert.lookups.definitions()
        .filter(definition => definition.parameter === name)
      if (matches.length > 1) {
@@ -295,6 +310,7 @@ export class TypertGatewayService extends Service implements TypertGateway {
      ...(marker.method === method ? {} : { implementation: marker.method }),
      invocation: receiver,
      parameters,
+      ...(cancellation === undefined ? {} : { cancellation }),
      result: { mode: 'src-json' },
    }
  }
@@ -11,6 +11,8 @@ export interface InvokeRemoteRequest {
  readonly method: string
  /** Named wire values; fields must exactly match the descriptor. */
  readonly args: Readonly<Record<string, unknown>>
+  /** Carrier or direct-caller cancellation injected only into cancellation-aware methods. */
+  readonly signal?: AbortSignal
 }

 /** Stable infrastructure and boundary failures emitted before or after business execution. */
@@ -17,11 +17,18 @@ declare module '@deepseek-ai/dsh-type-meta' {
  }

  interface TypeRTRemoteMap {
-    'goals/create': (agentId: string, request: { readonly objective: string }) => Promise<{ readonly ref: string }>
+    'goals/create': (
+      agentId: string,
+      request: { readonly objective: string },
+      signal?: AbortSignal,
+    ) => Promise<{ readonly ref: string }>
  }

  interface TypeRTRemoteContextMap {
-    'fixture:goals/create': (request: { readonly objective: string }) => Promise<{ readonly ref: string }>
+    'fixture:goals/create': (
+      request: { readonly objective: string },
+      signal?: AbortSignal,
+    ) => Promise<{ readonly ref: string }>
    'fixture:goals/rename': (request: { readonly objective: string }) => Promise<{ readonly renamed: boolean }>
  }

@@ -58,6 +65,7 @@ function directDescriptor(): InvocationDescriptor {
      source: 'json',
      codec: { mode: 'strict', typeSymbol: '@fixture#CreateRequest', schema: requestSchema },
    }],
+    cancellation: { parameter: 'signal' },
    result: { mode: 'strict', typeSymbol: '@fixture#CreateResult', schema: createResultSchema },
  }
 }
@@ -114,6 +122,19 @@ describe('Client TypeRT API', () => {
      { args: { agentId: 'agent-1', request: { objective: 'ship' } } },
      expect.any(AbortSignal),
    )
+    const callerAbort = new AbortController()
+    await expect(ctx.api.goals.create(
+      'agent-1',
+      { objective: 'cancel me' },
+      callerAbort.signal,
+    )).resolves.toEqual({ ref: 'goal-1' })
+    const combinedSignal = call.mock.calls.at(-1)?.[3]
+    expect(combinedSignal).toBeInstanceOf(AbortSignal)
+    expect(combinedSignal).not.toBe(callerAbort.signal)
+    const cancellation = new Error('caller cancelled')
+    callerAbort.abort(cancellation)
+    expect(combinedSignal?.aborted).toBe(true)
+    expect(combinedSignal?.reason).toBe(cancellation)
    await expect(ctx.api.goals.create('', { objective: 'ship' })).rejects.toThrow('rejected "agentId"')

    call.mockResolvedValueOnce({ ok: true, value: { ref: 1 } })
@@ -299,10 +320,18 @@ describe('Client TypeRT API', () => {
      .mockResolvedValue({ ok: true, value: { ref: 'goal-1' } })
    const ctx = await bench(call)
    const descriptor = directDescriptor()
-    const dispose = ctx.api.mount({ package: '@fixture/goals', descriptors: [descriptor] })
+    const dispose = ctx.api.mount({
+      package: '@fixture/goals',
+      descriptors: [descriptor, contextDescriptor()],
+    })
    const create = ctx.api.goals.create as unknown as (...args: unknown[]) => Promise<unknown>
+    const goals = (ctx as FixtureContext).goals
+    const rename = goals.rename as unknown as (...args: unknown[]) => Promise<unknown>

-    await expect(create('agent-1')).rejects.toThrow('expected 2 argument(s), got 1')
+    await expect(create('agent-1')).rejects.toThrow('expected 2 business argument(s) plus an optional AbortSignal, got 1')
+    await expect(create('agent-1', { objective: 'ship' }, undefined, 'extra'))
+      .rejects.toThrow('got 4')
+    await expect(rename.call(goals)).rejects.toThrow('expected 1 argument(s), got 0')
    await expect((ctx as FixtureContext).goals.create({ objective: 'ship' }))
      .rejects.toThrow('no Client Context binder')

@@ -45,6 +45,7 @@ const emptyModel: TypertContribution['model'] = {
 class GoalService extends Service {
  readonly typertGateway = bindTypeRTGateway(this, 'goals')
  readonly calls: string[] = []
+  lastSignal: AbortSignal | undefined
  nextResult: unknown = undefined
  businessError: Error | undefined

@@ -53,8 +54,9 @@ class GoalService extends Service {
  }

  @Remote
-  create(agent: FixtureAgent, request: { readonly title: string }): unknown {
+  create(agent: FixtureAgent, request: { readonly title: string }, signal: AbortSignal): unknown {
    this.calls.push('create')
+    this.lastSignal = signal
    return {
      agentId: agent.id,
      title: request.title,
@@ -224,6 +226,19 @@ class RestParameterService extends Service {
  }
 }

+class NonFinalSignalService extends Service {
+  readonly typertGateway = bindTypeRTGateway(this, 'nonFinalSignal', { namespace: 'invalid-signal' })
+
+  constructor(ctx: Context) {
+    super(ctx, 'nonFinalSignal')
+  }
+
+  @Remote
+  run(signal: AbortSignal, value: string): string {
+    return signal.aborted ? '' : value
+  }
+}
+
 class WrongBindingService extends Service {
  readonly typertGateway = bindTypeRTGateway(this, 'notWrongBinding', { namespace: 'wrong-binding' })

@@ -334,13 +349,24 @@ describe('TypertGatewayService', () => {
    registerAgentLookup(ctx, agent)
    registerStrict(ctx, [createDescriptor()])
    const caller = ctx.extend({ fixtureScope: 'direct-caller' })
+    const abort = new AbortController()

    await expect(caller.typertGateway.invoke({
      namespace: 'goals',
      method: 'create',
      args: { agentId: 'agent-1', request: { title: '  ship  ' } },
+      signal: abort.signal,
    })).resolves.toEqual({ agentId: 'agent-1', title: 'ship', scope: 'direct-caller' })
    expect(service.calls).toEqual(['create'])
+    expect(service.lastSignal).toBe(abort.signal)
+
+    await expect(caller.typertGateway.invoke({
+      namespace: 'goals',
+      method: 'create',
+      args: { agentId: 'agent-1', request: { title: 'again' } },
+    })).resolves.toEqual({ agentId: 'agent-1', title: 'again', scope: 'direct-caller' })
+    expect(service.lastSignal).toBeInstanceOf(AbortSignal)
+    expect(service.lastSignal?.aborted).toBe(false)
  })

  it('resolves strict Remote Context identity without adding a business argument', async () => {
@@ -358,16 +384,19 @@ describe('TypertGatewayService', () => {
  })

  it('derives SRC direct lookup and JSON parameters from marker and parameter names', async () => {
-    const { ctx } = await setup()
+    const { ctx, service } = await setup()
    const agent = { id: 'agent-1' }
    registerAgentLookup(ctx, agent)
    const caller = ctx.extend({ fixtureScope: 'direct-src' })
+    const abort = new AbortController()

    await expect(caller.typertGateway.invoke({
      namespace: 'goals',
      method: 'create',
      args: { agentId: 'agent-1', request: { title: 'ship' } },
+      signal: abort.signal,
    })).resolves.toEqual({ agentId: 'agent-1', title: 'ship', scope: 'direct-src' })
+    expect(service.lastSignal).toBe(abort.signal)
  })

  it('does not downgrade an observed SRC lookup after its provider unloads', async () => {
@@ -605,6 +634,7 @@ describe('TypertGatewayService', () => {
      { plugin: DefaultParameterService, namespace: 'invalid-default', args: { value: 'x' } },
      { plugin: DestructuredParameterService, namespace: 'invalid-destructure', args: { value: { value: 'x' } } },
      { plugin: RestParameterService, namespace: 'invalid-rest', args: { values: ['x'] } },
+      { plugin: NonFinalSignalService, namespace: 'invalid-signal', args: { value: 'x' } },
    ] as const
    for (const testCase of cases) {
      const ctx = await setupGateway()
@@ -874,7 +904,8 @@ describe('TypertGatewayService', () => {
    expect(connection.matches?.('goals')).toBe(false)
    expect(connection.matches?.('goals/missing')).toBe(false)
    expect(connection.matches?.('legacy/list')).toBe(false)
-    const signal = new AbortController().signal
+    const abort = new AbortController()
+    const signal = abort.signal
    const handler = connection.handler
    if (handler === undefined) throw new Error('fixture Connection did not retain the /api interceptor')
    await expect(handler('goals/create', {
@@ -883,6 +914,10 @@ describe('TypertGatewayService', () => {
      ok: true,
      value: { agentId: 'agent-1', title: 'ship', scope: 'rpc-caller' },
    })
+    const service = rawGoalService(ctx)
+    expect(service.lastSignal).toBe(signal)
+    abort.abort(new Error('client disconnected'))
+    expect(service.lastSignal?.aborted).toBe(true)
    const invalid = await handler('goals/create', { invalid: true }, signal)
    expect(invalid).toMatchObject({
      ok: false,
@@ -904,7 +939,6 @@ describe('TypertGatewayService', () => {
      expect(result.error.message).toContain('plain-object args field')
    }

-    const service = rawGoalService(ctx)
    service.businessError = 'non-error failure' as unknown as Error
    await expect(handler('goals/fail', { args: { request: null } }, signal)).resolves.toEqual({
      ok: false,
@@ -1099,6 +1133,7 @@ function createDescriptor(): InvocationDescriptor {
        })),
      },
    ],
+    cancellation: { parameter: 'signal' },
    result: strictCodec('@fixture/gateway#CreateResult', z.object({
      agentId: z.string(),
      title: z.string(),