From b8d4e004ed3def78546b931d9234ad2ef55cc803 Mon Sep 17 00:00:00 2001 From: pku-xht Date: Fri, 7 Aug 2026 16:59:04 +0800 Subject: [PATCH] test(schedule): reject mismatched receipt time zones --- .../tests/api-proxy-schedule-view.spec.ts | 298 ++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 packages/host/apiproxy/tests/api-proxy-schedule-view.spec.ts diff --git a/packages/host/apiproxy/tests/api-proxy-schedule-view.spec.ts b/packages/host/apiproxy/tests/api-proxy-schedule-view.spec.ts new file mode 100644 index 0000000000..70467734f7 --- /dev/null +++ b/packages/host/apiproxy/tests/api-proxy-schedule-view.spec.ts @@ -0,0 +1,298 @@ +/** + * Schedule reminder views cross the Host only after persistence proves their + * dispatch prefix. Live append sends raw events; session/flushed replays the + * identical dispatch with a generic sidecar. History independently gates the + * same projection on an identity-matching stored prefix. + */ + +import { Context } from 'cordis' +import { describe, expect, it } from 'vitest' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import UserInteractionService from '@deepseek-ai/dsh-user-interaction' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' +import SystemPrompt from '@deepseek-ai/dsh-system-prompt' +import ToolRegistry from '@deepseek-ai/dsh-tools' +import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api' +import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' +import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy' +import { ScheduleId } from '@deepseek-ai/dsh-tool-schedule' + +interface FlushControl { + handler: () => true | Promise +} + +function reminderCreateData(id: string, prompt: string) { + return { + version: 1 as const, + operation: 'create' as const, + schedule: { + id: ScheduleId(id), + kind: 'after' as const, + prompt, + afterSeconds: 1, + scheduledAt: '2026-08-05T12:00:01.000Z', + }, + } +} + +async function harness(control?: FlushControl): Promise { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt, { persona: '' }) + await ctx.plugin(ToolRegistry) + await ctx.plugin(UserInteractionService) + await ctx.plugin(AgentRegistry) + if (control !== undefined) ctx.on('session/flush', () => control.handler()) + return ctx +} + +function appendReminder( + session: Session, + id: string, + prompt: string, +): { create: SessionEvent; dispatch: SessionEvent } { + const scheduleId = ScheduleId(id) + const create = session.append('schedule/change', reminderCreateData(id, prompt)) + const dispatch = session.append('schedule/change', { + version: 1, + operation: 'dispatch', + id: scheduleId, + }) + return { create, dispatch } +} + +async function collectEvents( + iterable: AsyncIterable>, + count: number, + abort: AbortController, +): Promise[]> { + const events: Extract[] = [] + for await (const envelope of iterable) { + if (envelope.payload.type !== 'session/event') continue + events.push(envelope.payload) + if (events.length >= count) abort.abort() + } + return events +} + +describe('commit-aware Schedule live views', () => { + it('takes the max of reverse flush completion and replays each dispatch once', async () => { + const first = Promise.withResolvers() + let calls = 0 + const ctx = await harness({ + handler: () => ++calls === 1 ? first.promise : true, + }) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + const abort = new AbortController() + const collected = collectEvents( + api.events.mux({ rpcId: RpcId('schedule-live'), payload: {} }, abort.signal), + 6, + abort, + ) + const session = ctx.sessions.create(SessionId('schedule-live')) + const firstPair = appendReminder(session, 'schedule-1', 'first') + const slow = ctx.sessions.flush(session) + const secondPair = appendReminder(session, 'schedule-2', 'second') + await expect(ctx.sessions.flush(session)).resolves.toBe(true) + first.resolve(true) + await expect(slow).resolves.toBe(true) + + const frames = await collected + const raw = frames.filter(frame => frame.view === undefined) + const presented = frames.filter(frame => frame.view?.for === 'event') + expect(raw.map(frame => frame.event.seq)).toEqual([0, 1, 2, 3]) + expect(presented.map(frame => frame.event.seq)).toEqual([1, 3]) + expect(presented[0]?.event).toBe(firstPair.dispatch) + expect(presented[1]?.event).toBe(secondPair.dispatch) + expect(presented.map(frame => frame.view)).toEqual([ + { + for: 'event', + view: { + scheduleId: 'schedule-1', prompt: 'first', + occurrenceAt: '2026-08-05T12:00:01.000Z', + }, + }, + { + for: 'event', + view: { + scheduleId: 'schedule-2', prompt: 'second', + occurrenceAt: '2026-08-05T12:00:01.000Z', + }, + }, + ]) + expect(firstPair.create.seq).toBe(0) + await ctx.fiber.dispose() + }) + + it('withholds a view after rejection and publishes it on the next successful checkpoint', async () => { + let calls = 0 + const ctx = await harness({ + handler: () => ++calls === 1 ? Promise.reject(new Error('disk unavailable')) : true, + }) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + const abort = new AbortController() + const collected = collectEvents( + api.events.mux({ rpcId: RpcId('schedule-retry'), payload: {} }, abort.signal), + 3, + abort, + ) + const session = ctx.sessions.create(SessionId('schedule-retry')) + appendReminder(session, 'schedule-1', 'retry me') + await expect(ctx.sessions.flush(session)).rejects.toThrow('disk unavailable') + await expect(ctx.sessions.flush(session)).resolves.toBe(true) + + const frames = await collected + expect(frames.filter(frame => frame.view?.for === 'event')).toHaveLength(1) + expect(frames.at(-1)?.view).toMatchObject({ + for: 'event', + }) + await ctx.fiber.dispose() + }) +}) + +describe('Schedule history views', () => { + it('presents a resumed ancestor dispatch copied into a fork seed', async () => { + const ctx = await harness() + const scheduleId = ScheduleId('resumed-reminder') + const resumed = ctx.sessions.create(SessionId('schedule-resumed'), { + seed: [{ + type: 'schedule/change', + seq: 0, + time: 1, + data: reminderCreateData('resumed-reminder', 'after restart'), + }], + meta: { cwd: '/tmp' }, + }) + const dispatch = resumed.append('schedule/change', { + version: 1, + operation: 'dispatch', + id: scheduleId, + }) + const child = ctx.sessions.fork(resumed, undefined, SessionId('schedule-fork')) + ctx.provide('sessionPersistence', { + readFrom: () => Promise.resolve({ meta: child.header, events: [...child.events] }), + } as never) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + + const response = await api.sessions.history({ + rpcId: RpcId('schedule-resumed-fork'), payload: { sessionId: child.id }, + }) + if (!response.result.ok) throw new Error(response.result.error.message) + expect(response.result.value.events.find(entry => entry.event.seq === dispatch.seq)?.view).toEqual({ + for: 'event', + view: { + scheduleId, + prompt: 'after restart', + occurrenceAt: '2026-08-05T12:00:01.000Z', + }, + }) + await ctx.fiber.dispose() + }) + + it('uses only the attached identity-matching stored prefix and fails soft to raw history', async () => { + const ctx = await harness() + const parent = ctx.sessions.create(SessionId('schedule-parent'), { meta: { cwd: '/tmp' } }) + appendReminder(parent, 'parent-reminder', 'from parent') + const session = ctx.sessions.create(SessionId('schedule-attached'), { + seed: [...parent.events], + meta: { cwd: '/tmp', parentSession: parent.id, seedLength: 2 }, + }) + let readFrom = (): Promise<{ meta: SessionHeader; events: SessionEvent[] }> => Promise.resolve({ + meta: session.header, + events: [...session.events.slice(0, 1)], + }) + ctx.provide('sessionPersistence', { + readFrom: () => readFrom(), + } as never) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + const history = async () => { + const response = await api.sessions.history({ + rpcId: RpcId('schedule-history'), payload: { sessionId: session.id }, + }) + if (!response.result.ok) throw new Error(response.result.error.message) + return response.result.value.events + } + + expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined() + readFrom = () => Promise.resolve({ + meta: { ...session.header, delegationDepth: 0 }, + events: [...session.events.slice(0, 2)], + }) + expect((await history()).find(entry => entry.event.seq === 1)?.view).toMatchObject({ + for: 'event', + }) + readFrom = () => Promise.resolve({ + meta: { ...session.header, cwd: '/different', delegationDepth: 0 }, + events: [...session.events.slice(0, 2)], + }) + expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined() + readFrom = () => Promise.resolve({ + meta: { ...session.header, timeZone: 'UTC', delegationDepth: 0 }, + events: [...session.events.slice(0, 2)], + }) + expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined() + readFrom = () => Promise.reject(new Error('physical read unavailable')) + expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('presents every dispatch in detached persisted history', async () => { + const ctx = await harness() + let source: Session | undefined + const owner = await ctx.plugin(Object.assign((inner: Context) => { + source = inner.sessions.create(SessionId('schedule-source'), { meta: { cwd: '/tmp' } }) + }, { inject: ['sessions'] })) + if (source === undefined) throw new Error('session owner did not publish its session') + appendReminder(source, 'schedule-1', 'cold reminder') + const meta = source.header + const events = [...source.events] + await owner.dispose() + ctx.provide('sessionPersistence', { + list: () => Promise.resolve([meta]), + inspect: () => Promise.resolve({ meta, events }), + readFrom: () => Promise.resolve({ meta, events }), + } as never) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + const response = await api.sessions.history({ + rpcId: RpcId('schedule-cold'), payload: { sessionId: meta.id }, + }) + if (!response.result.ok) throw new Error(response.result.error.message) + expect(response.result.value.events.find(entry => entry.event.seq === 1)?.view).toMatchObject({ + for: 'event', + }) + await ctx.fiber.dispose() + }) + + it('withholds a detached view that exists only in a logical inspection', async () => { + const ctx = await harness() + let source: Session | undefined + const owner = await ctx.plugin(Object.assign((inner: Context) => { + source = inner.sessions.create(SessionId('schedule-logical-only'), { meta: { cwd: '/tmp' } }) + }, { inject: ['sessions'] })) + if (source === undefined) throw new Error('session owner did not publish its session') + appendReminder(source, 'schedule-logical', 'not physically committed') + const meta = source.header + const events = [...source.events] + await owner.dispose() + let physicalEvents = events.slice(0, 1) + ctx.provide('sessionPersistence', { + list: () => Promise.resolve([meta]), + inspect: () => Promise.resolve({ meta, events }), + readFrom: () => Promise.resolve({ meta, events: physicalEvents }), + } as never) + const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp' }) + const history = async () => { + const response = await api.sessions.history({ + rpcId: RpcId('schedule-logical-only-history'), payload: { sessionId: meta.id }, + }) + if (!response.result.ok) throw new Error(response.result.error.message) + return response.result.value.events + } + + expect((await history()).find(entry => entry.event.seq === 1)?.view).toBeUndefined() + physicalEvents = events + expect((await history()).find(entry => entry.event.seq === 1)?.view).toMatchObject({ for: 'event' }) + await ctx.fiber.dispose() + }) +})