From fea3bcdbf43a90a0f0dd5380bee34a747fe4bf07 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:05:38 +0800 Subject: [PATCH] feat(typert): support default resource resolvers --- ...-session-history-stream-and-snapshot.zh.md | 110 ++++++++++++++++++ packages/api/gateway/src/access.ts | 31 +++-- .../gateway/tests/gateway-access.host.spec.ts | 29 +++++ packages/api/session-controller/src/index.ts | 19 +-- .../tests/remote-access.host.spec.ts | 7 +- packages/typert/protocol/src/types.ts | 17 +-- packages/typert/registry/src/service.ts | 6 +- 7 files changed, 181 insertions(+), 38 deletions(-) create mode 100644 .agents/notes/proposed/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md diff --git a/.agents/notes/proposed/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md b/.agents/notes/proposed/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md new file mode 100644 index 0000000000..f6565f9dec --- /dev/null +++ b/.agents/notes/proposed/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md @@ -0,0 +1,110 @@ +# Agent Note: Session 历史快照、纯监听与 Web Follow 聚合 + +Status: proposed + +## Problem + +历史的一次读取、实时事件订阅和 Web 开场聚合需要不同的生命周期。若纯监听也读取 Observation 来计算起点，订阅就引入不必要的历史读取；若聚合层另建历史读取模块，又会让标准接口与 Web 各自拥有一条读取路径。需要明确资源持有和接续顺序，使历史快照与实时事件能组合而不重复读取、不遗漏开场期间的事件。 + +## Proposal + +在 session-controller 包内保留并行的 `NewSessionController`（`newSession`）与 `NewWebController`（`newWeb`）。旧 `SessionController`、旧客户端与旧 API 不变；新接口仅交付 History Record，不包含 header、projection、assistant stream 或 activate/resume。 + +### 职责与接口 + +| 所有者 | 职责 | 不承担的工作 | +|---|---|---| +| `NewSessionController.history` | 取得 Observation，执行 `paginate`、`pageRecords`，返回 `{ cursor, records, hasMore }`，随后释放租约 | 不监听事件、不激活 | +| `NewSessionController.observeHistory` | 创建 Listener，发出无 cursor 的 `ready`，转发 `listener.events()` | 不读取历史、不取得 Observation | +| `NewWebController.follow` | 先创建 Listener，再在 Host 内直接调用 `ctx.newSessionController.history`，发 snapshot 后转发 `eventsAfter(cursor)` | 不另建 `historyRead` 模块、不经 RPC 调用标准方法 | +| `HistoryListener` | 注册 `session/event` 和 `session/created` 适配，缓冲、去重、检查连续性，以 `entryFor` 转为事件帧并输出，关闭时释放监听 | 不取得或持有 Observation、不补读快照、不激活 | + +`history` 接收 `address`、可选的 `throughSeq`、`beforeSeq` 与分页预算。不传 `throughSeq` 时，返回的 cursor 是本次 Observation 的末尾；传入时必须是该 Observation 可读取的有效点位。`beforeSeq` 只选择更早的页面，不替代基准 cursor；返回 cursor 不等于旧页最后一条记录的位置。`follow` 只接收 address 和开场分页预算，不接受旧 `throughSeq` 或 `beforeSeq`，开场读取当前点位的最新页。 + +复用已有 [SessionObservation](../../../../packages/session-query/session-query/src/observation.ts) 定义及 [Observation 决策](../../implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md)，不引入来自 #4107 的新概念。`acquireObservation` 仅表示现有 `sessionQuery.observeSession` 加地址校验与错误映射；租约只在 `history` 方法生成返回值期间持有，不转交给 Listener，也不绑定流的整个生命周期。 + +### 三个入口 + +以下是职责示意，不作为可编译声明。`resolveHistoryRequest` 表示入口的参数校验和显式默认值处理；请求取消在读取完成后、返回结果前再次检查。 + +```text +// NewSessionController +async history(request, signal) { + const spec = resolveHistoryRequest(request) + using observation = await acquireObservation(this.ctx, spec.address, signal) + signal.throwIfAborted() + const cursor = spec.throughSeq ?? observation.cursor + validateHistoryCut(cursor, observation.cursor) + const page = paginate( + observation.events, spec.beforeSeq, spec.maxMessages, cursor, spec.turnWindow, + ) + return { cursor, records: pageRecords(page.events), hasMore: page.hasMore } +} +``` + +```text +// NewSessionController +async *observeHistory(request, signal) { + using listener = new HistoryListener(this.ctx, request.address, signal) + yield { type: 'ready' } + yield* listener.events() +} +``` + +```text +// NewWebController +async *follow(request, signal) { + using listener = new HistoryListener(this.ctx, request.address, signal) + const opening = await this.ctx.newSessionController.history(request, signal) + yield { type: 'snapshot', ...opening } + yield* listener.eventsAfter(opening.cursor) +} +``` + +### Listener 与快照接续 + +Listener 构造完成时已同步挂好事件监听并开始缓冲。`events()` 和 `eventsAfter(C)` 是同一个缓冲的两种单消费者选择，一次 Listener 生命周期只能选择并消费其中一种；不增加 `align` 或第二次初始化。两者返回异步事件帧序列，生成器延迟到首次迭代才执行，不影响构造时已开始的监听与缓冲。 + +- `events()` 输出订阅生效后收集的事件，以首个可输出事件建立序列起点，之后去重并检查连续性；不承诺交付订阅前的历史。 +- `eventsAfter(C)` 丢弃 `seq ≤ C` 的事件，第一条必须是 `C + 1`，随后保持连续；尚未收到下一条时等待，遇到缺号则明确结束并报告错误，不静默跳过或补读。 +- `session/created` 适配负责把构造期未逐条广播的种子事件纳入同一输入。这不调用 Observation，也不补读持久化快照；实现必须覆盖生成器尚未开始消费时发生创建的情况，不能依赖提前执行生成器来避免丢失。 + +例如 Listener 首先收到 15，读取 History 期间已缓冲 15～20，而 History 返回 `cursor = 20`。Follow 先发覆盖到 20 的 snapshot，再由 Listener 丢弃缓冲中重复的 15～20，从 21 开始输出增量。丢弃的是监听缓冲中的重叠，不是快照里的记录；不把快照故意截回 14。快照按页交付，更早记录仍通过 History 分页读取。 + +### Ready、授权与释放 + +`ready` 只确认该流的监听已安装，不包含 cursor，不证明会话存在、子会话身份已经验证或 Agent 就绪。独立组合接口的调用方持续消费并缓冲流，收到 ready 后再请求 History，应用历史页后按返回 cursor 丢弃重复增量；这保证远端请求顺序不会把读取排在监听生效之前。 + +Remote 方法继续声明现有 `resources` 会话授权；Host 内部直接调用不代替对 `follow` 入口本身的授权。纯流如何在不调用 Observation 的情况下完成子会话身份验证，是实现前必须核实的安全项，本提案不指定替代校验机制；不能把 ready 当成授权证据，也不能在必要身份校验未完成时输出事件数据。 + +History 成功、失败或取消都释放已取得的 Observation。流结束、调用方取消、插件卸载、历史读取失败及事件处理错误都必须撤销监听、释放缓冲并唤醒等待者；部分监听注册失败也必须清理已注册部分。Listener 不因没有新事件而阻止取消完成。 + +### 与既有决策及后续范围的关系 + +[历史与 Remote 事件传输决策](../../implemented/architecture/2026-08-18-session-history-and-event-transport.zh.md)继续拥有旧 API、传输生命周期与客户端恢复行为；[Observation 决策](../../implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md)继续拥有精确读取和 projection 状态。两篇仍有独立作用，本提案不替代或归档它们。 + +History-only `newWeb.follow` 不是旧完整 Web 协议的直接替换；旧客户端迁移不在本阶段。Projection 下一阶段先讨论独立接口，再讨论 Follow 多数据聚合与同游标问题；本提案不预设共享 Observation、projection revision 或其他对齐方案。 + +## Alternatives considered + +**先读历史，再开 tail 补读缺口。** 不采用；先安装监听就能覆盖读取期间的增量，不需要额外历史读取或任意旧游标续传能力。 + +**纯监听取得 Observation 并返回 ready.cursor。** 不采用；监听只需要确认订阅生效，历史点位由独立 History 返回。 + +**另建 historyRead 模块或让 Listener 持有 Observation。** 不采用；标准 History 方法已经拥有一次读取，Listener 仅按结果 cursor 消费缓冲，二者无需共享资源所有权。 + +## Acceptance criteria + +- History 一次调用取得一份 Observation，返回与该读取一致的 cursor、records、hasMore；覆盖显式点位、最新点位、空历史、非法点位及旧页分页。 +- 纯监听不调用 Observation 或激活路径；ready 不含 cursor，始终先于事件帧，事件去重并保持已建立起点之后的连续性。 +- Follow 的监听注册先于 History 调用，且仅调用一次标准 History；开场第一帧是 snapshot，重叠例子中第一条增量为 21，生成器延迟消费期间的事件不丢失。 +- 覆盖监听期间 Session 创建、种子事件与普通通知重复、缺号、取消、消费者结束、读取失败及插件卸载；断言租约、监听与等待者全部释放。 +- 实现前核实纯流子会话身份验证；拒绝错误父子关系或身份不符的地址，不能以 ready 或 child id 过滤替代授权。Host 集成测试覆盖真实资源授权与直接方法调用路径。 +- 新增无密钥快照用例固定 history-only 的开场与增量输出；旧 API 与旧客户端的协议、激活行为保持不变。测试结果与类型检查在实现阶段提供，本提案不代表现有示例代码已通过验证。 + +## Risks + +- 纯流不提供历史起点或存在性结论；调用方必须正确理解 ready，并在需要历史时按先订阅后读取的顺序组合。 +- 开场读取慢或消费者停止拉取会积累缓冲。实现前需核实项目既有背压与容量策略；不能通过静默丢事件维持连接，本提案不新增容量配置。 +- 子会话身份校验和 `session/created` 在消费开始前的适配必须完成验证，不能因为新入口更短而削弱数据访问控制或事件完整性。 +- 新旧接口并行保留有暂时维护成本；命名相近不表示新 Follow 已提供旧 Web 的全部数据或副作用。 diff --git a/packages/api/gateway/src/access.ts b/packages/api/gateway/src/access.ts index d85ebb3e65..b9c5ae42be 100644 --- a/packages/api/gateway/src/access.ts +++ b/packages/api/gateway/src/access.ts @@ -46,25 +46,24 @@ export function callResources( ): readonly RemoteAccessResource[] | undefined { // Each later source replaces an earlier one for its kind and keeps the kind's position. const sources = new Map ResourceIds>() - for (const { kind, field } of typert.resources.entries()) { - sources.set(kind, () => { - // A default field names exactly one resource id; only a reader may return several. - const id = defaultFieldId(descriptor, args, field) + const positionalArgs: readonly unknown[] = [ + ...descriptor.parameters.map(parameter => args[parameter.wire]), + ...descriptor.cancellation === undefined ? [] : [signal], + ] + for (const declaration of typert.resources.entries()) { + sources.set(declaration.kind, () => { + if ('resolve' in declaration) { + return read(() => declaration.resolve(...positionalArgs), []) + } + const id = defaultFieldId(descriptor, args, declaration.field) return Array.isArray(id) ? 'invalid' : id === undefined ? [] : [id] }) } for (const [kind, ids] of lookupResources(typert, descriptor, args)) { sources.set(kind, () => ids) } - let positional: readonly unknown[] | undefined for (const [kind, reader] of Object.entries(readers ?? {})) { - sources.set(kind, () => { - positional ??= [ - ...descriptor.parameters.map(parameter => args[parameter.wire]), - ...descriptor.cancellation === undefined ? [] : [signal], - ] - return read(reader, positional, []) - }) + sources.set(kind, () => read(() => reader(...positionalArgs), [])) } const resources: RemoteAccessResource[] = [] for (const [kind, ids] of sources) { @@ -91,7 +90,7 @@ export function eventResources( if (declared.length === 0) return GLOBAL const resources: RemoteAccessResource[] = [] for (const [kind, reader] of declared) { - if (!collect(resources, kind, read(reader, args, 'invalid'))) return undefined + if (!collect(resources, kind, read(() => reader(...args), 'invalid'))) return undefined } return resources } @@ -129,11 +128,11 @@ function lookupResources( return ids } -/** Run one reader with spread positional arguments: `undefined` yields `absent`, a list names several and must not be empty. */ -function read(reader: TypertResourceReader, args: readonly unknown[], absent: ResourceIds): ResourceIds { +/** Read resource identities: `undefined` yields `absent`, a list names several and must not be empty. */ +function read(reader: () => unknown, absent: ResourceIds): ResourceIds { let value: unknown try { - value = reader(...args) + value = reader() } catch { // Any reader error means it cannot name its resources, so they are invalid. return 'invalid' diff --git a/packages/api/gateway/tests/gateway-access.host.spec.ts b/packages/api/gateway/tests/gateway-access.host.spec.ts index 960405939f..374fcf0cfa 100644 --- a/packages/api/gateway/tests/gateway-access.host.spec.ts +++ b/packages/api/gateway/tests/gateway-access.host.spec.ts @@ -868,6 +868,35 @@ describe('resource resolution', () => { expect(callResources(typert, undefined, method(), {}, NEVER)).toEqual([GLOBAL]) }) + it('resolves default identities through the registered function and preserves explicit overrides', async () => { + const ctx = new Context() + roots.push(ctx) + await ctx.plugin(TypertRegistry) + const resolve = vi.fn((request?: string | { address?: { id: string } }) => { + return typeof request === 'string' ? request : request?.address?.id + }) + const dispose = ctx.typert.resources.register('test-shelf', { resolve }) + expect(callResources(ctx.typert, undefined, method('roomId'), { roomId: 'r1' }, NEVER)).toEqual([ROOM]) + expect(callResources(ctx.typert, undefined, method('request'), { request: { address: { id: 'r1' } } }, NEVER)).toEqual([ROOM]) + expect(callResources(ctx.typert, undefined, method('request'), { request: {} }, NEVER)).toEqual([GLOBAL]) + resolve.mockClear() + const args = { other: {}, later: { address: { id: 'r1' } } } + expect(callResources(ctx.typert, undefined, method('other', 'later'), args, NEVER)).toEqual([GLOBAL]) + expect(resolve).toHaveBeenCalledExactlyOnceWith(args.other, args.later) + resolve.mockClear() + expect(callResources(ctx.typert, undefined, { + ...method('renamed'), cancellation: { parameter: 'signal' }, + }, { renamed: { address: { id: 'r1' } } }, NEVER)).toEqual([ROOM]) + expect(resolve).toHaveBeenCalledExactlyOnceWith({ address: { id: 'r1' } }, NEVER) + expect(callResources(ctx.typert, { 'test-shelf': () => 'override' }, method('request'), { + request: { address: { id: 'r1' } }, + }, NEVER)).toEqual([shelf('override')]) + await dispose() + expect(ctx.typert.resources.entries()).toEqual([]) + ctx.typert.resources.register('test-shelf', { resolve: () => { throw new Error('cannot resolve') } }) + expect(callResources(ctx.typert, undefined, method('request'), { request: {} }, NEVER)).toBeUndefined() + }) + it('lets a reader replace the defaults for its kind, called with the positional wire arguments', async () => { const typert = await registry() const readers = { 'test-shelf': (ids: unknown) => ids } diff --git a/packages/api/session-controller/src/index.ts b/packages/api/session-controller/src/index.ts index 85770b8e21..e6d0acb551 100644 --- a/packages/api/session-controller/src/index.ts +++ b/packages/api/session-controller/src/index.ts @@ -78,13 +78,14 @@ declare module '@deepseek-ai/cordis' { } /** - * Read the Session resource a history address names: the Session itself, or - * the child Session whose transcript a subagent address reads. - * @param request - history request carrying the durable address. - * @returns the Session whose access the read requires. + * Extract the Session resource from an ID or a history address. + * @param request - first wire argument carrying a Session ID or address. + * @returns the target Session identity when present. */ -function addressedSession(request: { readonly address: SessionAddress }): SessionId { - return request.address.kind === 'session' ? request.address.sessionId : request.address.childSessionId +function resolveSessionResource(request?: { readonly sessionId?: SessionId; readonly address?: SessionAddress }): SessionId | undefined { + const address = request?.address + return request?.sessionId ?? (address === undefined ? undefined + : address.kind === 'session' ? address.sessionId : address.childSessionId) } /** Session Controller deployment policy. */ @@ -165,7 +166,7 @@ export class SessionController extends TypertRemoteService { protected async [Service.init](): Promise { const ctx = this.ctx - ctx.typert.resources.register('session', { field: 'sessionId' }) + ctx.typert.resources.register('session', { resolve: resolveSessionResource }) this.agents = new ApiSessionAgentController(ctx) this.agents.initialize() this.commands = new SessionCommandController(ctx, this.agents, process.cwd()) @@ -493,7 +494,7 @@ export class SessionController extends TypertRemoteService { * @param signal - cancellation for persistence reads. * @returns one chronological page. */ - @Remote({ name: 'page', resources: { session: addressedSession } }) + @Remote('page') page(request: SessionPageRequest, signal: AbortSignal): Promise { return this.history.page(request, signal) } @@ -505,7 +506,7 @@ export class SessionController extends TypertRemoteService { * @returns a complete opening snapshot followed by gap-free durable event * frames and optional cursorless assistant-stream frames. */ - @Remote({ mode: 'stream', resources: { session: addressedSession } }) + @Remote({ mode: 'stream' }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable { return this.history.follow(request, signal) } diff --git a/packages/api/session-controller/tests/remote-access.host.spec.ts b/packages/api/session-controller/tests/remote-access.host.spec.ts index 42b78e5190..824eb5f702 100644 --- a/packages/api/session-controller/tests/remote-access.host.spec.ts +++ b/packages/api/session-controller/tests/remote-access.host.spec.ts @@ -107,15 +107,18 @@ function message(text: string) { } describe('Session Controller access', () => { - it('owns the Session resource kind and its default sessionId field', async () => { + it('owns the Session resource kind and its default resolver', async () => { const { ctx } = await harness() expect(() => ctx.typert.resources.register('session', { field: 'sessionId' })).toThrow(/already registered/) }) it('reads the Session resource of page and follow requests from the addressed Session', async () => { - const { controller } = await harness() + const { ctx, controller } = await harness() + const resource = ctx.typert.resources.get('session') + if (resource === undefined || !('resolve' in resource)) throw new Error('Session resource resolver is missing') for (const method of ['page', 'follow']) { const reader = remoteMethods(controller).find(marker => marker.method === method)?.resources?.['session'] + ?? resource.resolve expect(reader?.({ address: { kind: 'session', sessionId: SHARED } })).toBe(SHARED) // A subagent address reads the child transcript; whether the parent's access covers it is the policy's rule. expect(reader?.({ address: { kind: 'subagent', parentSessionId: SHARED, childSessionId: PRIVATE } })).toBe(PRIVATE) diff --git a/packages/typert/protocol/src/types.ts b/packages/typert/protocol/src/types.ts index 650b2b3556..6b1d658b50 100644 --- a/packages/typert/protocol/src/types.ts +++ b/packages/typert/protocol/src/types.ts @@ -878,15 +878,16 @@ export interface TypertContextRegistry { subscribe(listener: TypertRegistryListener): TypertDisposer } -/** Default wire field of one resource kind. */ -export interface TypertResourceDeclaration { - /** Top-level or first-object wire field holding one identity of the kind. */ - readonly field: string -} +/** Default identity extraction for one resource kind. */ +export type TypertResourceDeclaration = + | { readonly field: string } + | { + /** Called once with spread positional wire arguments. Undefined means no matching resource. */ + resolve(...args: readonly unknown[]): string | undefined + } -/** One registered resource kind with its default wire field. */ -export interface TypertResourceEntry extends TypertResourceDeclaration { - /** Registered resource kind. */ +/** One registered resource kind with its default identity extraction. */ +export type TypertResourceEntry = TypertResourceDeclaration & { readonly kind: string } diff --git a/packages/typert/registry/src/service.ts b/packages/typert/registry/src/service.ts index c64817c819..15a5e46273 100644 --- a/packages/typert/registry/src/service.ts +++ b/packages/typert/registry/src/service.ts @@ -461,7 +461,7 @@ class ResourceStore { return { register: (kind, declaration) => this.register(ctx, kind, declaration), get: kind => this.declarations.get(kind)?.provider, - entries: () => [...this.declarations].map(([kind, entry]) => Object.freeze({ kind, field: entry.provider.field })), + entries: () => [...this.declarations].map(([kind, entry]) => Object.freeze({ kind, ...entry.provider })), subscribe: listener => this.changes.subscribe(ctx, listener), } } @@ -471,9 +471,9 @@ class ResourceStore { if (kind === GLOBAL_RESOURCE_KIND) { throw new Error(`typert: resource kind "${GLOBAL_RESOURCE_KIND}" is reserved for the built-in global resource`) } - validateWireName('resource field', declaration.field) + if ('field' in declaration) validateWireName('resource field', declaration.field) if (this.declarations.has(kind)) throw new Error(`typert: resource kind "${kind}" is already registered`) - const entry: ProviderEntry = { provider: { field: declaration.field }, owner: {} } + const entry: ProviderEntry = { provider: { ...declaration }, owner: {} } const { declarations, changes } = this return ctx.effect(function* () { declarations.set(kind, entry)