From 18cf84d133a9220a03d095b5699cbccfc12af8ec Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Sun, 23 Aug 2026 12:06:22 +0800 Subject: [PATCH] fix(client): materialize Agent scopes before list baseline --- ...8-session-history-and-event-transport.zh.md | 6 +++--- packages/client/runtime/src/client/index.ts | 2 +- .../runtime/src/client/sessions/service.ts | 18 ++++++++++++++++++ .../runtime/tests/client-apply.client.spec.ts | 11 +++++++++++ .../tests/sessions-service.client.spec.ts | 16 ++++++++++++++++ 5 files changed, 49 insertions(+), 4 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-08-18-session-history-and-event-transport.zh.md b/.agents/notes/implemented/architecture/2026-08-18-session-history-and-event-transport.zh.md index 95d2712564..3d0b5d4ab7 100644 --- a/.agents/notes/implemented/architecture/2026-08-18-session-history-and-event-transport.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-18-session-history-and-event-transport.zh.md @@ -258,11 +258,11 @@ WebSocket JSON 与进程内 carrier 的入口都从 `unknown` 开始按 `type` Host 只投影 request 一级的 `agent` 与 `signal`：`agent` 变为 frame 的一级 `agentId`，`signal` 成为 delivery lifetime，其余字段必须整体为无损 JSON。 -Client 用 `agentId` 同步解析已存在的 Agent Context，把当前 delivery signal 放回 request 的直接 `signal` 字段，再在目标 Context 的私有 key 上调用 Cordis `waterfall()`。 +Client 用 `agentId` 同步解析或物化 Agent Context，把当前 delivery signal 放回 request 的直接 `signal` 字段，再在目标 Context 的私有 key 上调用 Cordis `waterfall()`。Session-backed adapter 在首个成功 Session 列表 baseline 到达前允许 transport 先物化 scope；baseline 到达后由列表生命周期接管 scope 存活判断。 系统不扫描任意深度对象，不传 path array 或 placeholder，不 deep clone／restore Context 和 AbortSignal，也不等待未来出现的 Agent Context。 -Client adapter 未注册、Agent Context 不存在或已经释放时，本 Client 立即返回 `next`。它不订阅 registry、不做 resolve 后竞态复查，也不为一次 delivery 创建临时 Fiber。 +Client adapter 未注册、resolver 未返回 Context 或解析抛错时，本 Client 立即返回 `next`。它不订阅 registry、不做 resolve 后竞态复查，也不为一次 delivery 创建临时 Fiber。 Gateway Host 为每个未完成 waterfall 保存 `eventId`、Host continuation 与已投递 Client generation。新 Client generation 会收到同一 pending event 的重放。 @@ -310,7 +310,7 @@ API Proxy 只承接自身拥有的独立业务 API，不是 Session、Workspace **把 Agent scope 做成任意深度对象投影。** 递归扫描 Context 与 AbortSignal 需要 path、placeholder、clone 和 restore 协议，并把偶然对象结构升级成 wire 约定；一级 `agent` 与 `signal` 足以覆盖当前 waterfall。 -**等待 Client Agent Context 或 adapter 后再分发。** registry waiter、竞态复查和临时 delivery Fiber 会为一个可直接委托的 Client 增加额外生命周期；目标不存在时立即 `next` 保持 Cordis waterfall 语义。 +**等待 Client Agent Context 或 adapter 后再分发。** registry waiter、竞态复查和临时 delivery Fiber 会为一个可同步解析或物化目标的 Client 增加额外生命周期；resolver 当下不能提供目标时立即 `next` 保持 Cordis waterfall 语义。 **给 Remote Event 使用独立物理 WebSocket 或 duplex stream。** Gateway mux 已提供认证升级、复用、取消、错误映射和重连；下行 `$events` 加上 HTTP `$events/result` 足以表达 request／response，不需要第三条连接。 diff --git a/packages/client/runtime/src/client/index.ts b/packages/client/runtime/src/client/index.ts index f7ebf2c916..593e076a96 100644 --- a/packages/client/runtime/src/client/index.ts +++ b/packages/client/runtime/src/client/index.ts @@ -225,7 +225,7 @@ export function apply(ctx: Context): void { sessionControl.start() ctx.typert.contexts.registerClient('agent', { identity: candidate => sessions.scopeOf(candidate), - resolve: sessionId => sessions.scope(sessionId), + resolve: sessionId => sessions.resolveAgentScope(sessionId), }) const workspaceModel = new ClientWorkspaceModel(ctx.remote.workspace) const workspaces = new WorkspaceRuntime(ctx, connection.api, workspaceModel, sessions) diff --git a/packages/client/runtime/src/client/sessions/service.ts b/packages/client/runtime/src/client/sessions/service.ts index 41feedd958..7f1dc908d2 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/client/runtime/src/client/sessions/service.ts @@ -572,6 +572,18 @@ export class SessionRuntime implements ISessions { return this.resolve(id)?.ctx } + /** + * Materialize the Agent scope named by a validated Host Remote Event. + * The first successful Session-list baseline becomes authoritative for its + * lifetime; until then, transport streams may address the scope in either + * arrival order. + * @param id - Host-projected Agent identity (the matching Session id). + * @returns the identity-stable Agent Context. + */ + resolveAgentScope(id: SessionId): AgentContext { + return (this.scopes.get(id) ?? this.materializeScope(id)).ctx + } + /** * Read the Agent scope tag off a context. Service-method boundary: fetch * bundles must reach scope resolution through ctx.sessions — a cross-bundle @@ -664,6 +676,11 @@ export class SessionRuntime implements ISessions { const existing = this.scopes.get(id) if (existing !== undefined) return existing if (!this.eligible(id)) return undefined + return this.materializeScope(id) + } + + /** Materialize one scope after its caller establishes that the id may be addressed. */ + private materializeScope(id: SessionId): ScopeRecord { const { fiber, ctx } = createScope(this.rootCtx, id) const session = this.manager.get(id) // The Session owns its scoped dispatch point (host Agent.loopCtx mirror); @@ -764,6 +781,7 @@ export class SessionRuntime implements ISessions { /** Tear down scope + instance for no-longer-eligible sessions off stage; the staged one defers until the stage moves. */ private pruneScopes(): void { + if (this.list.getSnapshot().phase === 'pending') return for (const [id, record] of this.scopes) { if (this.eligible(id)) continue if (id === this.watched) { diff --git a/packages/client/runtime/tests/client-apply.client.spec.ts b/packages/client/runtime/tests/client-apply.client.spec.ts index cfc6dff0a7..5846a3d454 100644 --- a/packages/client/runtime/tests/client-apply.client.spec.ts +++ b/packages/client/runtime/tests/client-apply.client.spec.ts @@ -10,6 +10,7 @@ import { SESSION_SEARCH_RESULT_LIMIT } from '@deepseek-ai/dsh-api-session-contro import TypertRegistry from '@deepseek-ai/dsh-typert-registry' import * as RuntimeClient from '../src/client/index.ts' import type { ConversationNodeDefinition } from '../src/client/contract/conversation.ts' +import { scopeOf } from '../src/client/agents/scope.ts' import { Session } from '../src/client/sessions/session.ts' import { SessionRuntime } from '../src/client/sessions/service.ts' import { FakeApiClient, fakeRemote, ok } from './fake-api.client.ts' @@ -68,6 +69,16 @@ async function flushMicrotasks(): Promise { } describe('runtime client apply', () => { + it('materializes Host-addressed Agent scopes before the Session list arrives', async () => { + const bench = await mount() + const adapter = bench.ctx.typert.contexts.getClient('agent') + const first = adapter?.resolve('s-early') + + expect(first).toBeDefined() + expect(scopeOf(first as Context)).toBe('s-early') + expect(adapter?.resolve('s-early')).toBe(first) + }) + it('refreshes Sessions on every Gateway connection generation', async () => { const refresh = vi.spyOn(SessionRuntime.prototype, 'handleConnected') const bench = await mount() diff --git a/packages/client/runtime/tests/sessions-service.client.spec.ts b/packages/client/runtime/tests/sessions-service.client.spec.ts index 75d1690fce..7285b25b0e 100644 --- a/packages/client/runtime/tests/sessions-service.client.spec.ts +++ b/packages/client/runtime/tests/sessions-service.client.spec.ts @@ -121,6 +121,22 @@ describe('search', () => { }) describe('scope tree', () => { + it('retains a Host-addressed scope until the first Session baseline owns pruning', async () => { + const b = bench() + const scoped = b.svc.resolveAgentScope(sid('s-early')) + expect(scopeOf(scoped)).toBe('s-early') + + b.svc.handleControlFrame({ + type: 'baseline', + value: { queues: {}, jobs: {}, projections: {} }, + }) + await Promise.resolve() + expect(b.svc.resolveAgentScope(sid('s-early'))).toBe(scoped) + + await feedList(b, []) + expect(b.svc.scope(sid('s-early'))).toBeUndefined() + }) + it('mints lazily on first resolution, tags the ctx, and keeps binding identity stable', async () => { const b = bench() await feedList(b, [{ id: 's1' }])