From 120b2882c0753f342080021e0464f0bbcb87eaba Mon Sep 17 00:00:00 2001 From: pku-xht Date: Sat, 8 Aug 2026 00:11:39 +0800 Subject: [PATCH] fix(time-context): validate restored readings --- apps/web/tests/scaffold.ts | 11 +- apps/web/tests/schedule-after.e2e.ts | 113 +++++++++--------- .../context/time-context/src/invariant.ts | 1 + .../time-context/tests/invariant.spec.ts | 32 +++++ 4 files changed, 101 insertions(+), 56 deletions(-) diff --git a/apps/web/tests/scaffold.ts b/apps/web/tests/scaffold.ts index 31a638c12e..0cd3c34df2 100644 --- a/apps/web/tests/scaffold.ts +++ b/apps/web/tests/scaffold.ts @@ -196,6 +196,8 @@ export interface LaunchOptions { paceMs?: number /** Synthetic model capacity for UI scenarios whose seeded history must remain uncompacted. */ replayContextWindow?: number + /** Caller-owned keyless adapter for a fixture that must derive its response at stream time. */ + fixtureAdapter?: LlmAdapter /** * Tool presentation mode patched onto the shipped `tools` row (`code` * collapses the wire to run_code + the SDK prompt section). Omit for the @@ -264,6 +266,11 @@ export async function launchWebScaffold(options: LaunchOptions = {}): Promise ctx.llm.registerAdapter( replayProviders(options.replayContextWindow).map(provider => provider.id), - new RouteOnlyAdapter(replayProviders(options.replayContextWindow)), - ), 'web e2e scaffold: route-only adapter') + options.fixtureAdapter ?? new RouteOnlyAdapter(replayProviders(options.replayContextWindow)), + ), 'web e2e scaffold: fixture adapter') } } catch (error) { if (process.cwd() !== originalCwd) process.chdir(originalCwd) diff --git a/apps/web/tests/schedule-after.e2e.ts b/apps/web/tests/schedule-after.e2e.ts index 95b891a141..95ea416171 100644 --- a/apps/web/tests/schedule-after.e2e.ts +++ b/apps/web/tests/schedule-after.e2e.ts @@ -3,7 +3,7 @@ // one-second owner path queues a best-effort followup, commits dispatch, and // renders the Host's durability-gated reminder sidecar. A separate browser // scenario drives local at through the real zone wire and model tool call. -import { mkdtemp, realpath, rm, writeFile } from 'node:fs/promises' +import { mkdtemp, realpath, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { fileURLToPath } from 'node:url' @@ -11,8 +11,8 @@ import type { Browser, Page } from 'playwright' import { chromium } from 'playwright' import { afterAll, beforeAll, describe, expect, it, onTestFailed } from 'vitest' import type { AgentHandle } from '@deepseek-ai/dsh-agent' -import { CallId, createUserMessage } from '@deepseek-ai/dsh-llm' -import type { ReplayEntry } from '@deepseek-ai/dsh-llm-replay' +import { CallId, createUserMessage, LlmAdapter } from '@deepseek-ai/dsh-llm' +import type { GenerateOptions, LlmResolvedModelInfo, StreamChunk } from '@deepseek-ai/dsh-llm' import { SessionId } from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' @@ -44,6 +44,50 @@ interface CreatedScheduleView { deliveryMode: 'session-local' } +/** Deterministic model boundary that selects local at relative to its actual first request. */ +class BrowserZoneAtAdapter extends LlmAdapter { + readonly requests: GenerateOptions[] = [] + scheduledAt: string | undefined + + override resolveModel(provider: string, model: string): Promise { + return Promise.resolve({ provider, id: model, name: model, contextWindow: 128_000 }) + } + + override async * stream(options: GenerateOptions): AsyncIterable { + this.requests.push(options) + if (this.requests.length === 1) { + const target = Math.ceil((Date.now() + 10_000) / 1_000) * 1_000 + const scheduledAt = new Date(target).toISOString() + this.scheduledAt = scheduledAt + const args = JSON.stringify({ + prompt: AT_PROMPT, + at: { date: scheduledAt.slice(0, 10), time: scheduledAt.slice(11, 19) }, + }) + const callId = CallId('schedule-at-wire-call') + yield { type: 'block-start', index: 0, blockType: 'tool-call' } + yield { + type: 'tool-call-delta', index: 0, id: callId, + name: 'schedule_create', argumentsDelta: args, + } + yield { + type: 'block-end', index: 0, + block: { type: 'tool-call', id: callId, name: 'schedule_create', arguments: args }, + } + yield { type: 'usage', usage: { inputTokens: 256, outputTokens: 32 } } + yield { type: 'finish', reason: { kind: 'tool-calls' } } + return + } + const text = this.requests.length === 2 + ? 'The zone-aware reminder is scheduled.' + : 'The zone-aware reminder is due.' + yield { type: 'block-start', index: 0, blockType: 'text' } + yield { type: 'text-delta', index: 0, text } + yield { type: 'block-end', index: 0, block: { type: 'text', text } } + yield { type: 'usage', usage: { inputTokens: 128, outputTokens: 16 } } + yield { type: 'finish', reason: { kind: 'stop' } } + } +} + /** Wait for one in-process lifecycle fact without using test-scoped expect.poll in beforeAll. */ async function waitForFact(read: () => boolean, timeoutMs: number): Promise { const deadline = Date.now() + timeoutMs @@ -187,59 +231,12 @@ describe.skipIf(MODE === 'record')('web e2e: browser-zone local at reminder', () let browser: Browser let page: Page let tripwire: ReturnType - let replayDir: string - let scheduledAt: string + const adapter = new BrowserZoneAtAdapter() beforeAll(async () => { - replayDir = await mkdtemp(join(tmpdir(), 'dsh-schedule-at-wire-replay-')) - const replayOverride = join(replayDir, 'replay.override.json') - const target = Math.ceil((Date.now() + 30_000) / 1_000) * 1_000 - scheduledAt = new Date(target).toISOString() - const args = JSON.stringify({ - prompt: AT_PROMPT, - at: { date: scheduledAt.slice(0, 10), time: scheduledAt.slice(11, 19) }, - }) - const callId = CallId('schedule-at-wire-call') - const toolCall: ReplayEntry = { - kind: 'chunks', - chunks: [ - { type: 'block-start', index: 0, blockType: 'tool-call' }, - { - type: 'tool-call-delta', - index: 0, - id: callId, - name: 'schedule_create', - argumentsDelta: args, - }, - { - type: 'block-end', - index: 0, - block: { type: 'tool-call', id: callId, name: 'schedule_create', arguments: args }, - }, - { type: 'usage', usage: { inputTokens: 256, outputTokens: 32 } }, - { type: 'finish', reason: { kind: 'tool-calls' } }, - ], - } - const textReply = (text: string): ReplayEntry => ({ - kind: 'chunks', - chunks: [ - { type: 'block-start', index: 0, blockType: 'text' }, - { type: 'text-delta', index: 0, text }, - { type: 'block-end', index: 0, block: { type: 'text', text } }, - { type: 'usage', usage: { inputTokens: 128, outputTokens: 16 } }, - { type: 'finish', reason: { kind: 'stop' } }, - ], - }) - await writeFile(replayOverride, JSON.stringify([ - toolCall, - textReply('The zone-aware reminder is scheduled.'), - textReply('The zone-aware reminder is due.'), - ] satisfies ReplayEntry[])) scaffold = await launchWebScaffold({ extraOverlayPath: OVERLAY, - replayFixture: join(replayDir, 'override-only.jsonl'), - replayOverride, - replayContextWindow: 128_000, + fixtureAdapter: adapter, }) browser = await chromium.launch() page = await browser.newPage({ @@ -258,7 +255,6 @@ describe.skipIf(MODE === 'record')('web e2e: browser-zone local at reminder', () const failures: unknown[] = [] await browser?.close().catch((error: unknown) => failures.push(error)) await scaffold?.close().catch((error: unknown) => failures.push(error)) - await rm(replayDir, { recursive: true, force: true }).catch((error: unknown) => failures.push(error)) if (failures.length === 1) throw failures[0] if (failures.length > 1) throw new AggregateError(failures, 'Schedule at wire evidence teardown failed') }) @@ -300,6 +296,14 @@ describe.skipIf(MODE === 'record')('web e2e: browser-zone local at reminder', () expect(timeContextIndex).toBeGreaterThanOrEqual(0) expect(toolCallIndex).toBeGreaterThan(timeContextIndex) + const firstRequest = adapter.requests[0] + if (firstRequest === undefined) throw new Error('model did not receive the browser prompt') + expect(JSON.stringify(firstRequest.messages)).toContain('Session time zone: UTC.') + expect(JSON.stringify(firstRequest.messages)).toContain('Client time zone for this request: UTC.') + expect(firstRequest.tools?.some(tool => tool.name === 'schedule_create')).toBe(true) + + const scheduledAt = adapter.scheduledAt + if (scheduledAt === undefined) throw new Error('model did not choose a local at target') const created = agent.session.events.find(event => event.type === 'schedule/change' && event.data.operation === 'create' @@ -312,8 +316,9 @@ describe.skipIf(MODE === 'record')('web e2e: browser-zone local at reminder', () await waitForFact(() => agent.session.events.some(event => event.type === 'schedule/change' && event.data.operation === 'dispatch' - && event.data.id === scheduleId), 45_000) + && event.data.id === scheduleId), 20_000) await agent.whenIdle() + expect(adapter.requests).toHaveLength(3) await expect(scaffold.ctx.sessions.flush(agent.session)).resolves.toBe(true) const history = await scaffold.ctx.apiProxy.sessions.history({ diff --git a/packages/context/time-context/src/invariant.ts b/packages/context/time-context/src/invariant.ts index 49007a8eec..be42a0cbe1 100644 --- a/packages/context/time-context/src/invariant.ts +++ b/packages/context/time-context/src/invariant.ts @@ -173,6 +173,7 @@ function validateSession(session: Session, fail: InvariantFailure): void { /** Install validation for loaded and newly appended context readings. */ const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { for (const session of ctx.sessions.list()) validateSession(session, fail) + ctx.on('session/created', (session) => { validateSession(session, fail) }, { global: true }) ctx.on('internal/dispatch', (_mode, eventName, args) => { if (eventName !== 'session/event') return const [session, event] = args as [Session, SessionEvent] diff --git a/packages/context/time-context/tests/invariant.spec.ts b/packages/context/time-context/tests/invariant.spec.ts index ea73f3ea27..9affaf780c 100644 --- a/packages/context/time-context/tests/invariant.spec.ts +++ b/packages/context/time-context/tests/invariant.spec.ts @@ -189,6 +189,38 @@ describe('time-context invariants', () => { }).toThrow(/Session time zone cannot format its durable timestamp/) }) + it('rejects a malformed reading seeded after companion setup', async () => { + const ctx = await setup() + const id = SessionId('time-invariant-future-seed') + const text = reading( + '1', + '1', + 'model-visible message', + '2026-07-14T00:00:00+00:00[UTC]', + 'Asia/Shanghai', + 'Asia/Shanghai', + ) + expect(() => ctx.sessions.create(id, { + meta: { timeZone: 'Asia/Shanghai' }, + seed: [ + { type: 'turn/start', seq: 0, time: SECOND, data: { turn: 1 } }, + { + type: 'user/message', + seq: 1, + time: SECOND, + surfaceOp: 'append', + data: createUserMessage({ + content: [{ type: 'text', text: 'seeded request' }], + source: { kind: 'user', rpcId: 'seeded-request', clientTimeZone: 'Asia/Shanghai' } as never, + }), + }, + { type: 'step/start', seq: 2, time: SECOND, data: { turn: 1, step: 1 } }, + { ...event(text), seq: 3, surfaceOp: 'append' }, + ], + })).toThrow(/rendered timestamp does not match the Session time zone/) + expect(ctx.sessions.get(id)).toBeUndefined() + }) + it('rejects a time-context source that duplicates request authority', async () => { const ctx = await setup() const base = event(reading())