fix(time-context): await projection readiness for initial requests
This commit is contained in:
1 parent
05f1066f8c
commit
2f30c684ac
5 files changed
+80
-3
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
</details>
|
||||
|
||||
@@ -76,6 +76,8 @@ kind: "package-reference"
|
||||
|
||||
### 主要流程
|
||||
|
||||
pre-step 监听器在等待投影注册前安装。激活期间到达的步骤会等待读取句柄;注册失败或所属插件释放会终止请求,等待期间取消不会产生读数。
|
||||
|
||||
需要注入时,插件采样挂钟时间,从开放轮次的 `user-rpc` 消息派生浏览器时区策略,解析显示时区(请求本地或回退),并渲染三行读数。正数间隔调度会扫描原始持久会话事件,查找最新一条归因于插件的消息——包括被压缩(compaction)遮蔽的读数——因此调度无需进程本地缓存也能在恢复后存续。读数记录的是已进入的步骤,不是已完成或已传输的请求;后续准备失败时,该读数可能留在历史中。
|
||||
|
||||
</details>
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
{ agent, turn, step, signal },
|
||||
next,
|
||||
): Promise<PreStepDecision> => {
|
||||
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<void> {
|
||||
],
|
||||
}
|
||||
}, { prepend: true })
|
||||
await projectionReady
|
||||
}
|
||||
@@ -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<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
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<undefined>()
|
||||
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'))
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user