diff --git a/packages/context/time-context/README.i18n.yaml b/packages/context/time-context/README.i18n.yaml index 0c6f23cd4b..00bb63502d 100644 --- a/packages/context/time-context/README.i18n.yaml +++ b/packages/context/time-context/README.i18n.yaml @@ -36,8 +36,8 @@ en: 269d7aa1f64c4278 zh: 086018a60cc3a351 /deepseek-ai-dsh-time-context/understand-the-implementation/main-flow: - en: ae9f13873dde5271 - zh: 873f32e67f8b202b + en: f09994721526e830 + zh: 6f1a463355111513 /deepseek-ai-dsh-time-context/further-exploration: en: e1cd55f40edbb7ed zh: 8a2abf22ef859841 diff --git a/packages/context/time-context/README.md b/packages/context/time-context/README.md index 4d03796d30..cc5e40fda8 100644 --- a/packages/context/time-context/README.md +++ b/packages/context/time-context/README.md @@ -76,6 +76,8 @@ The plugin prepends an `agent/pre-step` listener that delegates first and append ### Main flow +The pre-step listener is installed before awaiting projection registration. Steps arriving during activation wait for the reader; registration failure or owner disposal stops the request, and cancellation while waiting produces no reading. + When an injection is due, the plugin samples the wall clock, derives the browser-zone policy from the open turn's `user-rpc` messages, resolves the display zone (request-local or fallback), and renders the three-line reading. Positive-interval scheduling scans raw durable session events for the latest plugin-attributed message — including one shadowed by compaction — so the schedule survives resume without a process-local cache. A reading records an entered step, not a completed or transmitted request; a later preparation failure can leave it in history. diff --git a/packages/context/time-context/README.zh.md b/packages/context/time-context/README.zh.md index b55a778d5c..f392be5216 100644 --- a/packages/context/time-context/README.zh.md +++ b/packages/context/time-context/README.zh.md @@ -76,6 +76,8 @@ kind: "package-reference" ### 主要流程 +pre-step 监听器在等待投影注册前安装。激活期间到达的步骤会等待读取句柄;注册失败或所属插件释放会终止请求,等待期间取消不会产生读数。 + 需要注入时,插件采样挂钟时间,从开放轮次的 `user-rpc` 消息派生浏览器时区策略,解析显示时区(请求本地或回退),并渲染三行读数。正数间隔调度会扫描原始持久会话事件,查找最新一条归因于插件的消息——包括被压缩(compaction)遮蔽的读数——因此调度无需进程本地缓存也能在恢复后存续。读数记录的是已进入的步骤,不是已完成或已传输的请求;后续准备失败时,该读数可能留在历史中。 diff --git a/packages/context/time-context/src/index.ts b/packages/context/time-context/src/index.ts index 69eb467fc6..03bc586d45 100644 --- a/packages/context/time-context/src/index.ts +++ b/packages/context/time-context/src/index.ts @@ -115,6 +115,8 @@ function validateRefreshInterval(refreshIntervalMs: number): void { /** * Register a prepended pre-step listener for the lifetime of `ctx`. + * Steps admitted during activation await the projection registration; a + * registration failure or owner disposal prevents the request from proceeding. * @param ctx - plugin context; the listener is disposed with it. * @param config - time zone and durable refresh scheduling configuration. * @throws when the refresh interval is invalid or the configured or process time zone cannot be resolved. @@ -145,7 +147,7 @@ export async function apply(ctx: Context, config: Config): Promise { return created } - const timeContextProjectionHandle = await ctx.sessionProjections.register({ + const projectionReady = ctx.sessionProjections.register({ key: 'timeContext', stateVersion: 2, stateSchema: timeContextStateSchema, @@ -177,6 +179,8 @@ export async function apply(ctx: Context, config: Config): Promise { { agent, turn, step, signal }, next, ): Promise => { + const timeContextProjectionHandle = await projectionReady + ctx.fiber.assertActive() const decision = await next() if (decision.kind === 'reject' || signal.aborted) return decision const now = Date.now() @@ -214,4 +218,5 @@ export async function apply(ctx: Context, config: Config): Promise { ], } }, { prepend: true }) + await projectionReady } diff --git a/packages/context/time-context/tests/initialization.spec.ts b/packages/context/time-context/tests/initialization.spec.ts new file mode 100644 index 0000000000..ab3d404ce4 --- /dev/null +++ b/packages/context/time-context/tests/initialization.spec.ts @@ -0,0 +1,68 @@ +/** Real Agent requests overlapping time-context projection registration. */ +import { Context } from '@deepseek-ai/cordis' +import AgentLoop from '@deepseek-ai/dsh-agent-loop' +import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' +import { createUserMessage } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' +import * as timeContext from '@deepseek-ai/dsh-time-context' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' + +describe('time-context activation', () => { + it.each(['ready', 'cancelled', 'disposed', 'failed'] as const)('handles a %s registration before the first request', async (outcome) => { + const ctx = new Context() + const registered = Promise.withResolvers() + const release = Promise.withResolvers() + onTestFinished(async () => { + release.resolve(undefined) + await ctx.fiber.dispose() + }) + await mountAgentLoopTestDependencies(ctx) + await ctx.plugin(AgentLoop, { agents: [] }) + const adapter = new MockAdapter([textResponse('ready')]) + ctx.llm.registerAdapter(['mock'], adapter) + const agent = await ctx.agentLoop.create(SessionId('initial-time'), { provider: 'mock', model: 'mock' }) + const register = ctx.sessionProjections.register.bind(ctx.sessionProjections) + const spy = vi.spyOn(ctx.sessionProjections, 'register').mockImplementation(async (definition) => { + const handle = await register(definition) + if (definition.key === 'timeContext') { + registered.resolve(undefined) + await release.promise + } + return handle + }) + onTestFinished(() => { spy.mockRestore() }) + const fiber = ctx.plugin(timeContext, { timeZone: 'UTC' }) + await registered.promise + const preparing = Promise.withResolvers() + ctx.on('agent/pre-step', (_payload, next) => { + preparing.resolve(undefined) + return next() + }, { prepend: true }) + const errors: unknown[] = [] + ctx.on('agent/error', ({ error }) => { errors.push(error) }) + agent.followup(createUserMessage({ content: [{ type: 'text', text: 'what time is it?' }], source: { kind: 'user' } })) + await preparing.promise + + if (outcome === 'cancelled') agent.cancel({ kind: 'user' }) + const disposal = outcome === 'disposed' ? fiber.dispose() : undefined + if (outcome === 'failed') { + const rejected = expect(Promise.resolve(fiber)).rejects.toThrow('time registration failed') + release.reject(new Error('time registration failed')) + await rejected + } else { + release.resolve(undefined) + await (disposal ?? fiber) + } + await agent.whenIdle() + + const readings = agent.session.snapshotEvents().filter(event => event.type === 'user/message' && event.data.source.kind === 'time-context') + expect(readings).toHaveLength(outcome === 'ready' ? 1 : 0) + expect(adapter.requests).toHaveLength(outcome === 'ready' ? 1 : 0) + if (outcome === 'ready') { + expect(adapter.requests[0]?.messages.flatMap(message => message.content) + .some(block => block.type === 'text' && block.text.startsWith('Time sampled while preparing turn 1, step 1:'))).toBe(true) + } + if (outcome === 'failed') expect(errors).toContainEqual(new Error('time registration failed')) + }) +})