Merge remote-tracking branch 'origin/master' into xtr/explicit-agent-context

# Conflicts:
#	packages/subagent/subagent/src/continuation.ts
This commit is contained in:
_Kerman
2026-09-07 12:09:06 +08:00
4365 changed files with 70780 additions and 23613 deletions
@@ -1,11 +1,11 @@
import { mkdtempSync, writeFileSync } from 'node:fs'
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-presets'
import SessionStore, { SessionLogOffset, SessionId } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionLogOffset, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
@@ -22,8 +22,12 @@ import { installSessionReadTestServices, testSessionPersistence } from './test-r
const roots: Context[] = []
/** Session cwd roots created per test, removed after their context settles. */
const tempDirs: string[] = []
afterEach(async () => {
await Promise.all(roots.splice(0).map(ctx => ctx.fiber.dispose()))
for (const dir of tempDirs.splice(0)) rmSync(dir, { recursive: true, force: true })
})
async function harness(): Promise<{ ctx: Context; agents: ApiSessionAgentController }> {
@@ -44,7 +48,7 @@ async function harness(): Promise<{ ctx: Context; agents: ApiSessionAgentControl
function header(id: string, cwd: string | null = '/workspace'): SessionHeader {
return {
version: 0,
version: SESSION_FORMAT_VERSION,
id: SessionId(id),
createdAt: 1,
isSeeded: false,
@@ -297,6 +301,7 @@ describe('ApiSession create or adoption', () => {
it('shares one in-flight creation between concurrent callers', async () => {
const { ctx, agents } = await harness()
const cwd = mkdtempSync(join(tmpdir(), 'dsh-session-controller-concurrent-'))
tempDirs.push(cwd)
const meta = header('concurrent-create', cwd)
const created = unpublishedAgent(ctx, meta)
let release!: () => void
@@ -317,6 +322,7 @@ describe('ApiSession create or adoption', () => {
it('accepts a raced ordinary creation and rejects a raced attached child', async () => {
const ordinary = await harness()
const cwd = mkdtempSync(join(tmpdir(), 'dsh-session-controller-create-'))
tempDirs.push(cwd)
const ordinaryMeta = header('create-race', cwd)
const winner = agent(ordinary.ctx, ordinaryMeta)
vi.spyOn(ordinary.ctx.agents, 'create').mockImplementation(async () => {
@@ -328,6 +334,7 @@ describe('ApiSession create or adoption', () => {
const child = await harness()
const childCwd = mkdtempSync(join(tmpdir(), 'dsh-session-controller-child-'))
tempDirs.push(childCwd)
const childId = SessionId('create-child-race')
vi.spyOn(child.ctx.agents, 'create').mockImplementation(async () => {
child.ctx.sessions.create(childId, {
@@ -342,6 +349,7 @@ describe('ApiSession create or adoption', () => {
it('validates ownership and cwd on the Agent returned by creation', async () => {
const child = await harness()
const childCwd = mkdtempSync(join(tmpdir(), 'dsh-session-controller-returned-child-'))
tempDirs.push(childCwd)
const childMeta = {
...header('returned-child', childCwd),
parentSession: SessionId('parent'),
@@ -357,6 +365,7 @@ describe('ApiSession create or adoption', () => {
const wrong = await harness()
const requestedCwd = mkdtempSync(join(tmpdir(), 'dsh-session-controller-wrong-cwd-'))
tempDirs.push(requestedCwd)
const wrongAgent = unpublishedAgent(wrong.ctx, header('wrong-returned-cwd', '/other'))
vi.spyOn(wrong.ctx.agents, 'create').mockResolvedValue({
agent: wrongAgent,
@@ -437,6 +446,7 @@ describe('ApiSession create or adoption', () => {
it('surfaces directory creation failure', async () => {
const { agents } = await harness()
const parent = mkdtempSync(join(tmpdir(), 'dsh-session-controller-file-'))
tempDirs.push(parent)
const file = join(parent, 'file')
writeFileSync(file, 'not a directory')
await expect(agents.ensureSession(SessionId('mkdir-failure'), join(file, 'child'), false))
@@ -0,0 +1,211 @@
import { describe, expect, it } from 'vitest'
import { LlmAttemptId, createAssistantMessage } from '@deepseek-ai/dsh-llm'
import { SessionSeq, type SessionEvent } from '@deepseek-ai/dsh-session'
import type {
SessionAssistantStreamBaseline,
SessionAssistantStreamFrame,
} from '../src/types.ts'
import { ClientAssistantStream } from '../src/client/sessions/assistant-stream.ts'
import type { SessionLiveEventEntry } from '../src/client/contract/events.ts'
const ATTEMPT = LlmAttemptId('session:1')
function entry(event: SessionEvent): SessionLiveEventEntry {
return { type: 'event', event }
}
function ordinary(seq: number): SessionLiveEventEntry {
return entry({ type: 'turn/start', seq: SessionSeq(seq), time: seq, data: { turn: 1 } })
}
function attemptEvent(seq: number, turn = 1, step = 1): SessionLiveEventEntry {
return entry({
type: 'assistant/attempt',
seq: SessionSeq(seq),
time: seq,
data: { turn, step, stream: [] },
})
}
function messageEvent(
seq: number,
turn = 1,
step = 1,
surfaceOp: 'append' | { readonly op: 'replace'; readonly start: number; readonly end: number } = 'append',
): SessionLiveEventEntry {
return entry({
type: 'assistant/message',
seq: SessionSeq(seq),
time: seq,
data: {
turn,
step,
message: createAssistantMessage({
content: [{ type: 'text', text: 'done' }],
source: { provider: 'mock', model: 'mock' },
}),
stream: [],
},
surfaceOp: surfaceOp === 'append'
? surfaceOp
: { ...surfaceOp, start: SessionSeq(surfaceOp.start), end: SessionSeq(surfaceOp.end) },
})
}
function start(
attemptId = ATTEMPT,
startedAfterSeq = -1,
): SessionAssistantStreamFrame {
return {
type: 'start', attemptId, revision: 1,
startedAfterSeq: startedAfterSeq === -1 ? -1 : SessionSeq(startedAfterSeq),
turn: 1, step: 1,
}
}
function chunkFrame(
index: number,
attemptId = ATTEMPT,
): SessionAssistantStreamFrame {
return {
type: 'chunk', attemptId, revision: index + 2, index, time: 20 + index,
chunk: { type: 'text-delta', index: 0, text: `chunk-${index}` },
}
}
function end(
index: number,
outcome: Extract<SessionAssistantStreamFrame, { type: 'end' }>['outcome'],
attemptId = ATTEMPT,
): SessionAssistantStreamFrame {
return { type: 'end', attemptId, revision: index + 2, index, outcome }
}
function baseline(nextIndex = 1): SessionAssistantStreamBaseline {
return {
revision: nextIndex + 1,
activeAttempt: {
attemptId: ATTEMPT,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex,
stream: [
{ type: 'chunk', time: 20, chunk: { type: 'text-delta', index: 0, text: 'first' } },
{ type: 'chunk', time: 21, chunk: { type: 'text-delta', index: 0, text: 'second' } },
],
},
}
}
function opened(): ClientAssistantStream {
const stream = new ClientAssistantStream()
expect(stream.acceptFrame(start())).toBeUndefined()
return stream
}
describe('ClientAssistantStream', () => {
it('replaces the durable window and reconstructs only the baseline prefix', () => {
const stream = new ClientAssistantStream()
const durable = ordinary(4)
const visible = stream.replace([durable], baseline(1))
expect(visible[0]).toBe(durable)
expect(visible).toHaveLength(2)
const reconstructed = visible[1]
if (reconstructed?.type !== 'transient') throw new Error('expected reconstructed transient chunk')
expect(reconstructed.event.type).toBe('assistant/live-chunk')
expect(reconstructed.event.seq).toBe(4.5)
expect(reconstructed.event.time).toBe(20)
expect(stream.replace([], baseline(3))).toHaveLength(2)
expect(stream.replace([])).toEqual([])
})
it('passes through durable events not owned by the active attempt', () => {
const stream = new ClientAssistantStream()
stream.acceptFrame(start(ATTEMPT, 1))
for (const durable of [
ordinary(1),
messageEvent(2, 1, 1, { op: 'replace', start: 0, end: 0 }),
attemptEvent(0),
attemptEvent(3, 2, 1),
attemptEvent(4, 1, 2),
]) {
expect(stream.acceptDurable(durable)).toEqual({ type: 'publish', entry: durable })
}
})
it('stages one owned settlement and releases it from the matching end frame', () => {
const stream = opened()
const durable = messageEvent(2)
expect(stream.acceptDurable(durable)).toBeUndefined()
expect(stream.acceptFrame(chunkFrame(0))).toEqual(expect.objectContaining({ type: 'transient' }))
expect(stream.acceptFrame(end(1, {
kind: 'committed', eventType: 'assistant/message', seq: 2,
}))).toEqual({ type: 'settlement', attemptId: String(ATTEMPT), entry: durable })
})
it('rebaselines duplicate durable settlements or starts', () => {
const duplicate = opened()
const durable = attemptEvent(2)
expect(duplicate.acceptDurable(durable)).toBeUndefined()
expect(duplicate.acceptDurable(durable)).toEqual({ type: 'rebaseline' })
expect(duplicate.acceptFrame(start(LlmAttemptId('session:2')))).toEqual({ type: 'rebaseline' })
const clean = new ClientAssistantStream()
expect(clean.acceptFrame(start())).toBeUndefined()
})
it('falls back to durable settlement for frames from an unknown attempt', () => {
const stream = new ClientAssistantStream()
const unknown = LlmAttemptId('session:unknown')
expect(stream.acceptFrame(chunkFrame(0, unknown))).toBeUndefined()
expect(stream.acceptFrame(end(0, { kind: 'abandoned' }, unknown))).toBeUndefined()
const durable = attemptEvent(2)
expect(stream.acceptDurable(durable)).toEqual({ type: 'publish', entry: durable })
const known = opened()
expect(known.acceptFrame(chunkFrame(0, unknown))).toBeUndefined()
expect(known.acceptFrame(end(0, { kind: 'abandoned' }, unknown))).toBeUndefined()
})
it('rebaselines known attempts on chunk or terminal index mismatch', () => {
const chunkMismatch = opened()
expect(chunkMismatch.acceptFrame(chunkFrame(1))).toEqual({ type: 'rebaseline' })
const endMismatch = opened()
expect(endMismatch.acceptFrame(end(1, { kind: 'abandoned' }))).toEqual({ type: 'rebaseline' })
})
it('settles abandonment only when no durable settlement remains pending', () => {
const empty = opened()
expect(empty.acceptFrame(end(0, { kind: 'abandoned' }))).toEqual({
type: 'abandonment',
attemptId: String(ATTEMPT),
})
const pending = opened()
expect(pending.acceptDurable(attemptEvent(2))).toBeUndefined()
expect(pending.acceptFrame(end(0, { kind: 'abandoned' }))).toEqual({ type: 'rebaseline' })
})
it('rebaselines committed outcomes without one exact staged settlement', () => {
const published = new ClientAssistantStream()
published.replace([attemptEvent(2)], baseline(0))
expect(published.acceptFrame(end(0, {
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
}))).toBeUndefined()
const missing = opened()
expect(missing.acceptFrame(end(0, {
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
}))).toEqual({ type: 'rebaseline' })
const wrongType = opened()
expect(wrongType.acceptDurable(messageEvent(2))).toBeUndefined()
expect(wrongType.acceptFrame(end(0, {
kind: 'committed', eventType: 'assistant/attempt', seq: 2,
}))).toEqual({ type: 'rebaseline' })
})
})
@@ -0,0 +1,56 @@
import { describe, expect, it } from 'vitest'
import { LlmAttemptId } from '@deepseek-ai/dsh-llm'
import { SessionSeq } from '@deepseek-ai/dsh-session'
import { SessionAssistantStreamAccumulator } from '../src/assistant-stream.ts'
describe('SessionAssistantStreamAccumulator', () => {
it('replaces stale lifecycles, rejects frame gaps, and caches each baseline', () => {
const accumulator = new SessionAssistantStreamAccumulator()
const empty = accumulator.snapshot()
expect(accumulator.snapshot()).toBe(empty)
accumulator.accept({
type: 'start', attemptId: LlmAttemptId('stale'), revision: 2, turn: 1, step: 1,
}, -1)
expect(accumulator.snapshot()).toEqual({ revision: 2 })
accumulator.accept({
type: 'start', attemptId: LlmAttemptId('current'), revision: 1, turn: 2, step: 3,
}, SessionSeq(5))
expect(accumulator.snapshot()).toMatchObject({
revision: 1,
activeAttempt: {
attemptId: 'current', startedAfterSeq: 5, turn: 2, step: 3, nextIndex: 0, stream: [],
},
})
accumulator.accept({
type: 'chunk', attemptId: LlmAttemptId('other'), revision: 2, index: 0,
time: 4, chunk: { type: 'text-delta', index: 0, text: 'lost' },
}, SessionSeq(5))
expect(accumulator.snapshot()).toEqual({ revision: 2 })
accumulator.accept({
type: 'start', attemptId: LlmAttemptId('settled'), revision: 3, turn: 2, step: 4,
}, SessionSeq(8))
accumulator.accept({
type: 'chunk', attemptId: LlmAttemptId('settled'), revision: 4, index: 0,
time: 5, chunk: { type: 'text-delta', index: 0, text: 'ok' },
}, SessionSeq(8))
const active = accumulator.snapshot()
expect(active).toMatchObject({
revision: 4,
activeAttempt: {
attemptId: 'settled', startedAfterSeq: 8, nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 5, index: 0, dt: [], texts: ['ok'] }],
},
})
expect(accumulator.snapshot()).toBe(active)
accumulator.accept({
type: 'end', attemptId: LlmAttemptId('settled'), revision: 5, index: 1,
outcome: { kind: 'abandoned' },
}, SessionSeq(9))
expect(accumulator.snapshot()).toEqual({ revision: 5 })
})
})
@@ -65,6 +65,11 @@ async function mount(initialGeneration?: ConnectionGeneration): Promise<Bench> {
registerGenerationSource: () => () => {},
start: () => ({ stop: () => {} }),
}
ctx.reflect.provide('connection', connection)
ctx.reflect.provide('fileUpload', {
available: true,
post: () => Promise.reject(new Error('unexpected file upload')),
})
ctx.reflect.provide('remote', {
...remote,
$stream: <Item>(options: RemoteStreamOptions<Item>) => (
@@ -2,8 +2,10 @@ import { describe, expect, expectTypeOf, it, vi } from 'vitest'
import type { PromptContentPart as AttachmentPromptContentPart } from '@deepseek-ai/dsh-attachment/types'
import { SessionSeq, type SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
import {
MutableSessionEventSource, type SessionLiveEventEntry,
MutableSessionEventSource, type SessionAssistantSettlementEntry,
type SessionLiveEventEntry, type SessionTransientEventEntry,
} from '../src/client/contract/events.ts'
import { LlmAttemptId } from '@deepseek-ai/dsh-llm/brand'
import type { ISession } from '../src/client/contract/session.ts'
import type { ProjectionsBaseline } from '../src/client/sessions/projection-store.ts'
import { ProjectionValueStore } from '../src/client/sessions/projection-store.ts'
@@ -24,6 +26,35 @@ function entry(seq: SessionSeq): SessionLiveEventEntry {
}
}
function transient(attemptId: string, seq = 1.5): SessionTransientEventEntry {
return {
type: 'transient',
event: {
type: 'assistant/live-chunk',
seq,
time: 1,
data: {
attemptId: LlmAttemptId(attemptId),
turn: 1,
step: 1,
chunk: { type: 'text-delta', index: 0, text: 'live' },
},
},
}
}
function assistantSettlement(seq: SessionSeq): SessionAssistantSettlementEntry {
return {
type: 'event',
event: {
type: 'assistant/attempt',
seq,
time: seq,
data: { turn: 1, step: 1, stream: [] },
},
}
}
describe('Client Session contracts', () => {
it('requires branded Session positions at internal event fixture boundaries', () => {
expectTypeOf(entry).parameter(0).toEqualTypeOf<SessionSeq>()
@@ -42,8 +73,9 @@ describe('Client Session contracts', () => {
expectTypeOf<SessionPageRequest['throughSeq']>().toEqualTypeOf<number>()
})
it('keeps its catalog-visible prompt parts identical to attachment intake', () => {
expectTypeOf<SessionPromptContentPart>().toEqualTypeOf<AttachmentPromptContentPart>()
it('keeps its text and image prompt parts identical to attachment intake', () => {
expectTypeOf<Exclude<SessionPromptContentPart, { type: 'file' }>>()
.toEqualTypeOf<AttachmentPromptContentPart>()
})
it('publishes exact replace, prepend, and append event-window changes', () => {
@@ -105,4 +137,34 @@ describe('Client Session contracts', () => {
expect(iterate).toHaveBeenCalledOnce()
})
it('publishes one exact Assistant settlement delta after retiring its transient rows', () => {
const feed = new MutableSessionEventSource()
const opening = entry(SessionSeq(1))
const live = transient('attempt-one')
const settlement = assistantSettlement(SessionSeq(2))
const later = entry(SessionSeq(3))
feed.replace([opening, live, later], false)
feed.settleAssistant(LlmAttemptId('attempt-one'), settlement)
expect(feed.getSnapshot()).toEqual({
entries: [opening, settlement, later],
hasMore: false,
revision: 2,
change: {
kind: 'settle-assistant',
attemptId: 'attempt-one',
entry: settlement,
},
})
feed.append(transient('attempt-two', 3.5))
feed.settleAssistant(LlmAttemptId('attempt-two'))
expect(feed.getSnapshot().entries).toEqual([opening, settlement, later])
expect(feed.getSnapshot().change).toEqual({
kind: 'settle-assistant',
attemptId: 'attempt-two',
})
})
})
@@ -4,7 +4,7 @@ import type { Agent, ModelSelectionRef } from '@deepseek-ai/dsh-agent'
import { AttachmentError, AttachmentId } from '@deepseek-ai/dsh-attachment'
import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
import { createAssistantMessage, createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import { describe, expect, it, vi } from 'vitest'
import { ApiSessionAgentController } from '../src/agent.ts'
@@ -112,6 +112,23 @@ describe('Session queue commands', () => {
})).toEqual({ accepted: true })
expect(steer).toHaveBeenCalledWith(steered)
const queuedFile = createUserMessage({
content: [{
type: 'file',
attachment: { attachmentId: AttachmentId('file-queued'), name: 'queued.txt', bytes: 6 },
}],
source: { kind: 'user', rpcId: 'file-rpc' as never },
})
inbox.append('next-turn', queuedFile)
expect(controller.updateQueue({
sessionId: agent.id, itemId: queuedFile.id, action: { kind: 'steer' },
})).toEqual({ accepted: true })
expect(steer).toHaveBeenLastCalledWith(queuedFile)
expect(queuedFile).toMatchObject({
source: { kind: 'user', rpcId: 'file-rpc' },
content: [{ type: 'file', attachment: { name: 'queued.txt', bytes: 6 } }],
})
await expectFailure(Promise.resolve().then(() => controller.cancel({
sessionId: SessionId('missing'),
})), 'session/not-found')
@@ -143,7 +160,7 @@ async function persistedController(
await ctx.plugin(SessionStore)
const sessionId = SessionId('cold-attachment')
const meta: SessionHeader = {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -178,6 +195,7 @@ describe('Session attachment authorization', () => {
{ ...event('assistant/message', SessionSeq(1), {
turn: 1,
step: 1,
stream: [],
message: createAssistantMessage({
content: [{ type: 'image', attachment: message }],
source: { provider: 'fixture', model: 'fixture' },
@@ -191,10 +209,30 @@ describe('Session attachment authorization', () => {
source: { kind: 'user' },
})],
}),
event('assistant/chunk', SessionSeq(3), {
event('assistant/attempt', SessionSeq(3), {
turn: 1,
step: 1,
chunk: { type: 'block-end', index: 0, block: { type: 'image', attachment: streamed } },
stream: [
{
type: 'chunk',
time: 3,
chunk: { type: 'block-start', index: 0, blockType: 'text' },
},
{
type: 'chunk',
time: 3,
chunk: { type: 'block-end', index: 0, block: { type: 'text', text: '' } },
},
],
}),
event('assistant/attempt', SessionSeq(4), {
turn: 1,
step: 1,
stream: [{
type: 'chunk',
time: 4,
chunk: { type: 'block-end', index: 0, block: { type: 'image', attachment: streamed } },
}],
}),
]
const readImage = vi.fn((ref: ImageAttachmentRef) => Promise.resolve({ ref, data: Uint8Array.of(1) }))
@@ -0,0 +1,471 @@
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
import type { Agent, ModelSelectionRef } from '@deepseek-ai/dsh-agent'
import AttachmentStore, { AttachmentId } from '@deepseek-ai/dsh-attachment'
import type {
FileAttachmentRef, ImageAttachmentRef, SaveFileAttachment, SaveFileStreamAttachment,
} from '@deepseek-ai/dsh-attachment'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { UserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import CommandRuntime from '@deepseek-ai/dsh-commands'
import { createScope } from '@deepseek-ai/dsh-scope'
import FileUploads from '@deepseek-ai/dsh-client-file-upload'
import type { FileUploadReceiptId } from '@deepseek-ai/dsh-client-file-upload/types'
import { describe, expect, it, vi } from 'vitest'
import type { ApiSessionAgentController } from '../src/agent.ts'
import { SessionCommandController } from '../src/commands.ts'
import type { SessionRequestId } from '../src/types.ts'
const SESSION = SessionId('upload-session')
async function uploadHarness(origin?: 'subagent'): Promise<{
ctx: Context
controller: SessionCommandController
uploads: FileUploads
agent: Agent
followup: ReturnType<typeof vi.fn>
saveFile: ReturnType<typeof vi.fn>
saveFileStream: ReturnType<typeof vi.fn>
saveImages: ReturnType<typeof vi.fn>
disposeAgent: () => void
uploadRoute: (request: Request) => Promise<Response>
}> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(CommandRuntime)
const session = ctx.sessions.create(SESSION, {
meta: { cwd: '/workspace', ...(origin === undefined ? {} : { origin }) },
})
const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
const followup = vi.fn()
const agent = {
id: session.id,
session,
inbox,
status: 'idle',
ctx: undefined,
steer: vi.fn(),
followup,
cancel: vi.fn(),
} as unknown as Agent
;(agent as { ctx: Context }).ctx = createScope(ctx, agent).ctx
const disposeAgent = ctx.agents.register(agent)
const saveFile = vi.fn((input: SaveFileAttachment): Promise<FileAttachmentRef> => Promise.resolve({
attachmentId: AttachmentId(`sha256:${'cd'.repeat(32)}`),
name: input.name ?? 'file',
bytes: input.data.byteLength,
}))
const saveFileStream = vi.fn(async (input: SaveFileStreamAttachment): Promise<FileAttachmentRef> => {
let bytes = 0
for await (const chunk of input.data) bytes += chunk.byteLength
return {
attachmentId: AttachmentId(`sha256:${'ef'.repeat(32)}`),
name: input.name ?? 'file',
bytes,
}
})
const saveImages = vi.fn((): Promise<readonly ImageAttachmentRef[]> =>
Promise.reject(new Error('fixture did not expect image persistence')))
ctx.provide('attachments', Object.setPrototypeOf(
{ saveFile, saveFileStream, saveImages },
AttachmentStore.prototype,
) as never)
let uploadRoute: ((request: Request) => Promise<Response>) | undefined
ctx.provide('connection', {
fetch: {
register: (route: { readonly fetch: (request: Request) => Promise<Response> }) => {
uploadRoute = route.fetch
return () => {}
},
},
} as never)
ctx.provide('llm', {
listProviders: () => [{ id: 'fixture', name: 'Fixture' }],
resolveModelInfo: () => Promise.resolve({ provider: 'fixture', id: 'fixture-model', name: 'Fixture' }),
} as never)
const selection: ModelSelectionRef = {
current: { provider: 'fixture', model: 'fixture-model' },
assembled: undefined,
}
const agents = {
resolveAgent: () => Promise.resolve({ agent }),
selectionFor: () => selection,
serializeImageAdmission: <Value>(_agent: Agent, operation: () => Promise<Value>) => operation(),
} as unknown as ApiSessionAgentController
const uploads = new FileUploads(ctx)
if (uploadRoute === undefined) throw new Error('file upload route was not registered')
return {
ctx,
controller: new SessionCommandController(ctx, agents, '/workspace'),
uploads,
agent,
followup,
saveFile,
saveFileStream,
saveImages,
disposeAgent,
uploadRoute,
}
}
function promptRequest(content: Parameters<SessionCommandController['prompt']>[0]['content']) {
return {
requestId: 'req-1' as SessionRequestId,
sessionId: SESSION,
mode: 'queue' as const,
content,
}
}
describe('Session file uploads', () => {
it('registers an HTTP route bound to the upload service', async () => {
const { uploadRoute } = await uploadHarness()
await expect(uploadRoute(new Request('http://host/upload')))
.resolves.toMatchObject({ status: 405 })
})
it('stages one verbatim upload and preserves its order with an admitted image', async () => {
const { ctx, controller, uploads, agent, followup, saveFile, saveImages } = await uploadHarness()
const receipt = await uploads.upload(agent, { data: 'AAAA', name: 'notes.pdf' }, new AbortController().signal)
expect(saveFile).toHaveBeenCalledTimes(1)
expect(receipt.file.name).toBe('notes.pdf')
expect(receipt.file.bytes).toBe(3)
expect(uploads.resolve(agent, receipt.receiptId)).toEqual(receipt.file)
expect(uploads.resolve(agent, 'missing' as FileUploadReceiptId)).toBeUndefined()
const commandHandler = vi.fn((_invocation: unknown) => ({ kind: 'success' as const }))
ctx.commands.register({
name: 'files', description: 'Use staged files', input: { hint: '<task>', attachments: true },
handler: commandHandler,
})
await ctx.commands.execute(
agent,
'/files inspect',
[{ type: 'file', receiptId: receipt.receiptId }],
new AbortController().signal,
)
expect(commandHandler.mock.calls[0]?.[0]).toMatchObject({
attachments: [{ type: 'file', attachment: receipt.file }],
})
const image: ImageAttachmentRef = {
attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`),
mediaType: 'image/png',
bytes: 3,
width: 1,
height: 1,
}
saveImages.mockResolvedValueOnce([image])
await controller.prompt(promptRequest([
{ type: 'file', receiptId: receipt.receiptId },
{ type: 'image', mediaType: 'image/png', data: 'AAAA' },
{ type: 'text', text: 'read it' },
]))
expect(followup).toHaveBeenCalledTimes(1)
const message = followup.mock.calls[0]?.[0] as UserMessage
expect(message.content).toEqual([
{ type: 'file', attachment: receipt.file },
{ type: 'image', attachment: image },
{ type: 'text', text: 'read it' },
])
})
it('stages a bounded byte stream and forwards cancellation to storage', async () => {
const { controller, uploads, followup, saveFileStream } = await uploadHarness()
const abort = new AbortController()
const receipt = await uploads.uploadStream({
sessionId: SESSION,
data: (async function* (): AsyncIterable<Uint8Array> {
yield Uint8Array.of(1, 2)
yield Uint8Array.of(3, 4, 5)
})(),
signal: abort.signal,
name: 'huge.bin',
})
expect(saveFileStream).toHaveBeenCalledWith(expect.objectContaining({
signal: abort.signal,
name: 'huge.bin',
}))
expect(receipt.file).toMatchObject({ name: 'huge.bin', bytes: 5 })
await controller.prompt(promptRequest([{ type: 'file', receiptId: receipt.receiptId }]))
expect((followup.mock.calls[0]?.[0] as UserMessage).content).toEqual([
{ type: 'file', attachment: receipt.file },
])
})
it('keeps the stream name optional and maps storage failures through the same error vocabulary', async () => {
const { uploads, saveFileStream } = await uploadHarness()
const stream = (async function* (): AsyncIterable<Uint8Array> {})()
await expect(uploads.uploadStream({ sessionId: SESSION, data: stream }))
.resolves.toMatchObject({ file: { name: 'file', bytes: 0 } })
expect(saveFileStream).toHaveBeenLastCalledWith({ data: stream })
saveFileStream.mockRejectedValueOnce('disk offline')
await expect(uploads.uploadStream({
sessionId: SESSION,
data: (async function* (): AsyncIterable<Uint8Array> { yield Uint8Array.of(1) })(),
}))
.rejects.toMatchObject({
code: 'gateway/internal', message: 'failed to store file upload: disk offline',
})
})
it('rejects a prompt citing a file that was never staged for the session', async () => {
const { controller, followup, saveImages } = await uploadHarness()
await expect(controller.prompt(promptRequest([
{ type: 'image', mediaType: 'image/png', data: 'AAAA' },
{ type: 'file', receiptId: 'missing-receipt' as FileUploadReceiptId },
]))).rejects.toMatchObject({ code: 'session/attachment-invalid', details: { reason: 'FILE_NOT_STAGED' } })
expect(followup).not.toHaveBeenCalled()
expect(saveImages).not.toHaveBeenCalled()
})
it('publishes no receipt when its exact Agent is disposed during storage', async () => {
const { uploads, agent, saveFile, disposeAgent } = await uploadHarness()
const saved = Promise.withResolvers<FileAttachmentRef>()
saveFile.mockReturnValueOnce(saved.promise)
const uploading = uploads.upload(agent, { data: 'AAAA', name: 'late.bin' }, new AbortController().signal)
await vi.waitFor(() => { expect(saveFile).toHaveBeenCalledOnce() })
disposeAgent()
saved.resolve({
attachmentId: AttachmentId(`sha256:${'ab'.repeat(32)}`), name: 'late.bin', bytes: 3,
})
await expect(uploading).rejects.toMatchObject({ code: 'session/not-found' })
})
it('resolves a cold ordinary Agent and releases the resolver registration', async () => {
const { ctx, uploads, agent, disposeAgent } = await uploadHarness()
disposeAgent()
const resolveAgent = vi.fn(async () => {
ctx.agents.register(agent)
return agent
})
const disposeResolver = uploads.registerAgentResolver(resolveAgent)
expect(() => { uploads.registerAgentResolver(resolveAgent) }).toThrow('already registered')
await expect(uploads.uploadStream({
sessionId: SESSION,
data: (async function* (): AsyncIterable<Uint8Array> { yield Uint8Array.of(1) })(),
})).resolves.toMatchObject({ file: { bytes: 1 } })
expect(resolveAgent).toHaveBeenCalledWith(SESSION)
disposeResolver()
const replacement = vi.fn(async () => agent)
const disposeReplacement = uploads.registerAgentResolver(replacement)
disposeResolver()
expect(() => { uploads.registerAgentResolver(replacement) }).toThrow('already registered')
expect(disposeReplacement).toBeTypeOf('function')
disposeReplacement()
})
it('rejects a cold upload when no Agent resolver is registered', async () => {
const { uploads, disposeAgent } = await uploadHarness()
disposeAgent()
await expect(uploads.uploadStream({
sessionId: SESSION,
data: (async function* (): AsyncIterable<Uint8Array> {})(),
})).rejects.toMatchObject({ code: 'session/not-found' })
})
it('rejects subagent uploads and access outside the receiving Agent scope', async () => {
const child = await uploadHarness('subagent')
await expect(child.uploads.upload(
child.agent,
{ data: 'AAAA' },
new AbortController().signal,
)).rejects.toMatchObject({
code: 'subagent/attachment-invalid',
details: { reason: 'SUBAGENT_FILE_UNSUPPORTED' },
})
expect(child.saveFile).not.toHaveBeenCalled()
const ordinary = await uploadHarness()
const receipt = await ordinary.uploads.upload(
ordinary.agent,
{ data: 'AAAA' },
new AbortController().signal,
)
const foreignScope = { ...ordinary.agent, ctx: ordinary.ctx } as Agent
expect(() => ordinary.uploads.resolve(foreignScope, receipt.receiptId))
.toThrow('operation requires the Agent\'s own scope')
})
it('retires accepted receipts after their rpcId becomes observable', async () => {
const { controller, uploads, agent } = await uploadHarness()
uploads.retirePrompt(agent, 'not-staged')
const receipt = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal)
await controller.prompt(promptRequest([{ type: 'file', receiptId: receipt.receiptId }]))
expect(uploads.resolve(agent, receipt.receiptId)).toEqual(receipt.file)
uploads.retirePrompt(agent, 'other-request')
expect(uploads.resolve(agent, receipt.receiptId)).toEqual(receipt.file)
uploads.retirePrompt(agent, 'req-1')
expect(uploads.resolve(agent, receipt.receiptId)).toBeUndefined()
})
it('retires observed prompt receipts and drops staged state with the Session', async () => {
const first = await uploadHarness()
const observed = await first.uploads.upload(
first.agent,
{ data: 'AAAA' },
new AbortController().signal,
)
await first.controller.prompt(promptRequest([{ type: 'file', receiptId: observed.receiptId }]))
first.ctx.emit('session/event', first.agent.session, {
type: 'user/message',
data: createUserMessage({
content: [{ type: 'text', text: 'extension event' }],
source: { kind: 'user', rpcId: 1 } as never,
}),
} as never)
expect(first.uploads.resolve(first.agent, observed.receiptId)).toEqual(observed.file)
first.agent.session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'observed' }],
source: { kind: 'user', rpcId: 'req-1' as SessionRequestId },
}), { surfaceOp: 'append' })
expect(first.uploads.resolve(first.agent, observed.receiptId)).toBeUndefined()
const second = await uploadHarness()
const abandoned = await second.uploads.upload(
second.agent,
{ data: 'AAAA' },
new AbortController().signal,
)
second.ctx.emit('session/disposed', second.agent.session)
expect(second.uploads.resolve(second.agent, abandoned.receiptId)).toBeUndefined()
})
it('deduplicates a retried rpcId already present in the Agent inbox', async () => {
const { controller, agent, followup } = await uploadHarness()
const request = promptRequest([{ type: 'text', text: 'once' }])
await controller.prompt(request)
agent.inbox.append('next-turn', followup.mock.calls[0]?.[0] as UserMessage)
await expect(controller.prompt(request)).resolves.toEqual({ accepted: true })
expect(followup).toHaveBeenCalledOnce()
})
it('deduplicates a retried rpcId already present in the durable log', async () => {
const { controller, agent, followup } = await uploadHarness()
const request = promptRequest([{ type: 'text', text: 'once' }])
agent.session.append('turn/start', { turn: 1 })
agent.session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'unidentified' }],
source: { kind: 'user' },
}), { surfaceOp: 'append' })
agent.session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'accepted' }],
source: { kind: 'user', rpcId: request.requestId },
}), { surfaceOp: 'append' })
await expect(controller.prompt(request)).resolves.toEqual({ accepted: true })
expect(followup).not.toHaveBeenCalled()
})
it('rejects when the Agent disappears during prompt admission', async () => {
const { controller, saveImages, disposeAgent } = await uploadHarness()
const admitted = Promise.withResolvers<readonly ImageAttachmentRef[]>()
saveImages.mockReturnValueOnce(admitted.promise)
const prompting = controller.prompt(promptRequest([
{ type: 'image', mediaType: 'image/png', data: 'AAAA' },
]))
await vi.waitFor(() => { expect(saveImages).toHaveBeenCalledOnce() })
disposeAgent()
admitted.resolve([{
attachmentId: AttachmentId('admitted-image'), mediaType: 'image/png', bytes: 3, width: 1, height: 1,
}])
await expect(prompting).rejects.toMatchObject({ code: 'session/not-found' })
})
it('rejects when a previously bound receipt retires during image admission', async () => {
const { controller, uploads, agent, saveImages } = await uploadHarness()
const receipt = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal)
await controller.prompt(promptRequest([{ type: 'file', receiptId: receipt.receiptId }]))
const admitted = Promise.withResolvers<readonly ImageAttachmentRef[]>()
saveImages.mockReturnValueOnce(admitted.promise)
const prompting = controller.prompt({
...promptRequest([
{ type: 'file', receiptId: receipt.receiptId },
{ type: 'image', mediaType: 'image/png', data: 'AAAA' },
]),
requestId: 'req-2' as SessionRequestId,
})
await vi.waitFor(() => { expect(saveImages).toHaveBeenCalledOnce() })
uploads.retirePrompt(agent, 'req-1')
admitted.resolve([{
attachmentId: AttachmentId('admitted-image'), mediaType: 'image/png', bytes: 3, width: 1, height: 1,
}])
await expect(prompting).rejects.toMatchObject({
code: 'session/attachment-invalid', details: { reason: 'FILE_NOT_STAGED' },
})
})
it('keeps the prior receipt binding when a later prompt attempt fails', async () => {
const { controller, uploads, agent, followup } = await uploadHarness()
const receipt = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal)
await controller.prompt(promptRequest([{ type: 'file', receiptId: receipt.receiptId }]))
followup.mockImplementationOnce(() => { throw new Error('busy') })
await expect(controller.prompt({
...promptRequest([{ type: 'file', receiptId: receipt.receiptId }]),
requestId: 'req-2' as SessionRequestId,
})).rejects.toMatchObject({ code: 'session/agent-busy' })
uploads.retirePrompt(agent, 'req-1')
expect(uploads.resolve(agent, receipt.receiptId)).toBeUndefined()
})
it('restores an unbound receipt after prompt delivery fails', async () => {
const { controller, uploads, agent, followup } = await uploadHarness()
const receipt = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal)
followup.mockImplementationOnce(() => { throw new Error('busy') })
await expect(controller.prompt(promptRequest([
{ type: 'file', receiptId: receipt.receiptId },
]))).rejects.toMatchObject({ code: 'session/agent-busy' })
uploads.retirePrompt(agent, 'req-1')
expect(uploads.resolve(agent, receipt.receiptId)).toEqual(receipt.file)
})
it('retires a prompt-bound receipt when its queued occurrence is removed', async () => {
const { controller, uploads, agent, followup } = await uploadHarness()
const receipt = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal)
await controller.prompt(promptRequest([{ type: 'file', receiptId: receipt.receiptId }]))
const queued = followup.mock.calls[0]?.[0] as UserMessage
agent.inbox.append('next-turn', queued)
expect(controller.updateQueue({
sessionId: SESSION,
itemId: queued.id,
action: { kind: 'remove' },
})).toEqual({ accepted: true })
expect(uploads.resolve(agent, receipt.receiptId)).toBeUndefined()
})
it('keeps separate names for identical bytes uploaded more than once', async () => {
const { controller, uploads, agent, followup } = await uploadHarness()
const first = await uploads.upload(agent, { data: 'AAAA', name: 'first.txt' }, new AbortController().signal)
const second = await uploads.upload(agent, { data: 'AAAA', name: 'second.txt' }, new AbortController().signal)
expect(first.file.attachmentId).toBe(second.file.attachmentId)
expect(first.receiptId).not.toBe(second.receiptId)
await controller.prompt(promptRequest([
{ type: 'file', receiptId: first.receiptId },
{ type: 'file', receiptId: second.receiptId },
]))
const message = followup.mock.calls[0]?.[0] as UserMessage
expect(message.content).toEqual([
{ type: 'file', attachment: first.file },
{ type: 'file', attachment: second.file },
])
})
it('maps a non-canonical payload to the wire attachment error', async () => {
const { uploads, agent, saveFile } = await uploadHarness()
await expect(uploads.upload(agent, { data: 'not base64!!' }, new AbortController().signal))
.rejects.toMatchObject({ code: 'session/attachment-invalid', details: { reason: 'INVALID_FILE_BASE64' } })
expect(saveFile).not.toHaveBeenCalled()
})
it('maps an unexpected storage failure to the internal wire error', async () => {
const { uploads, agent, saveFile } = await uploadHarness()
saveFile.mockRejectedValueOnce(new Error('disk unavailable'))
await expect(uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal))
.rejects.toMatchObject({
code: 'gateway/internal',
message: 'failed to store file upload: Error: disk unavailable',
})
})
})
@@ -3,7 +3,7 @@ import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import type { JobOutcome } from '@deepseek-ai/dsh-jobs'
import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import { describe, expect, it } from 'vitest'
@@ -181,7 +181,7 @@ describe('Session control jobs updates', () => {
const coldId = SessionId('session-cold-tasks')
let loaded = false
ctx.provide('sessionPersistence', {
list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
list: async () => [{ version: SESSION_FORMAT_VERSION, id: coldId, createdAt: 5, cwd: '/tmp' }],
locate: () => undefined,
load: () => { loaded = true; throw new Error('job projection must not load a cold log') },
} as never)
@@ -2,7 +2,7 @@ import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { describe, expect, it, vi } from 'vitest'
@@ -26,7 +26,7 @@ describe('SessionController facade', () => {
await ctx.plugin(AgentRegistry)
const sessionId = SessionId('controller-session')
const header: SessionHeader = {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -42,6 +42,16 @@ describe('SessionController facade', () => {
list: () => Promise.resolve([header]),
inspect,
}) as never)
let uploadResolver: ((sessionId: SessionId) => Promise<Agent>) | undefined
ctx.provide('fileUploads', {
registerAgentResolver: (resolve: (sessionId: SessionId) => Promise<Agent>) => {
uploadResolver = resolve
return () => {}
},
resolve: () => undefined,
bindPrompt: () => ({ commit: () => {}, [Symbol.dispose]: () => {} }),
retirePrompt: () => {},
} as never)
const controller = createSessionTestController(ctx, defaults)
const status = vi.fn()
const failure = vi.fn()
@@ -65,6 +75,17 @@ describe('SessionController facade', () => {
ctx,
} as Agent
ctx.agents.register(agent)
const resolveUploadAgent = (id: SessionId): Promise<Agent> => {
if (uploadResolver === undefined) throw new Error('file upload resolver was not registered')
return uploadResolver(id)
}
await expect(resolveUploadAgent(sessionId)).resolves.toBe(agent)
const activationError = new RemoteError('session/not-found', 'missing upload session', { sessionId })
vi.spyOn(
(controller as unknown as { agents: ApiSessionAgentController }).agents,
'resolveAgent',
).mockResolvedValueOnce({ error: activationError })
await expect(resolveUploadAgent(sessionId)).rejects.toBe(activationError)
const consumeSelection = vi.spyOn(
(controller as unknown as { agents: ApiSessionAgentController }).agents,
'consumeSelection',
@@ -83,6 +104,10 @@ describe('SessionController facade', () => {
content: [{ type: 'text', text: 'hello' }],
source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'browser prompt' }],
source: { kind: 'user', rpcId: 'controller-rpc' as never },
}), { surfaceOp: 'append' })
expect(status).toHaveBeenCalledWith(sessionId, true)
expect(failure).toHaveBeenCalledWith(sessionId, expect.stringContaining('fixture failure'))
expect(activity).toHaveBeenCalledWith(sessionId, expect.any(Number))
@@ -108,7 +133,7 @@ describe('SessionController facade', () => {
}, abort.signal)[Symbol.asyncIterator]()
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: { type: 'snapshot', cursor: 1 },
value: { type: 'snapshot', cursor: 2 },
})
abort.abort()
await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined })
@@ -122,7 +147,7 @@ describe('SessionController facade', () => {
await ctx.plugin(AgentRegistry)
const sessionId = SessionId(`background-${outcome}`)
const header: SessionHeader = {
version: 0, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
}
ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
list: () => Promise.resolve([header]),
@@ -178,7 +203,7 @@ describe('SessionController facade', () => {
await ctx.plugin(AgentRegistry)
const sessionId = SessionId('background-disposal')
const header: SessionHeader = {
version: 0, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
}
ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
list: () => Promise.resolve([header]),
@@ -27,13 +27,18 @@ export const ev = {
}) }),
stepStart: (seq: SessionSeq, turn: number, step = 0): SessionEvent =>
at(seq, { type: 'step/start', data: { turn, step } }),
chunkStart: (seq: SessionSeq, turn: number, step = 0, index = 0): SessionEvent =>
at(seq, { type: 'assistant/chunk', data: { turn, step, chunk: { type: 'block-start', index, blockType: 'text' } } }),
chunkText: (seq: SessionSeq, turn: number, piece: string, step = 0, index = 0): SessionEvent =>
at(seq, { type: 'assistant/chunk', data: { turn, step, chunk: { type: 'text-delta', index, text: piece } } }),
assistant: (seq: SessionSeq, turn: number, body: string, step = 0): SessionEvent =>
at(seq, { type: 'assistant/message', surfaceOp: 'append', data: {
turn, step,
stream: [
{ type: 'chunk', time: 1_700_000_000_000 + seq, chunk: { type: 'block-start', index: 0, blockType: 'text' } },
{ type: 'text-chunks', time0: 1_700_000_000_000 + seq, index: 0, dt: [], texts: [body] },
{
type: 'chunk', time: 1_700_000_000_000 + seq,
chunk: { type: 'block-end', index: 0, block: { type: 'text', text: body } },
},
{ type: 'chunk', time: 1_700_000_000_000 + seq, chunk: { type: 'finish', reason: { kind: 'stop' } } },
],
message: createMessage({
role: 'assistant',
content: text(body),
@@ -9,6 +9,7 @@ import type {
} from '@deepseek-ai/dsh-api-remotes/client'
import type {
SessionAddress,
SessionAssistantStreamBaseline,
SessionControlBaseline,
SessionControlFrame,
SessionFollowFrame,
@@ -27,6 +28,7 @@ import {
type RemoteStreamOptions,
} from '@deepseek-ai/dsh-api-gateway/client'
import type { SessionRemotes } from '../src/client/sessions/remotes.ts'
import { SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session/types'
import { historyRecordLastSeq } from '../src/client/sessions/history-records.ts'
const AVAILABLE_STREAM_CONNECTION = {
@@ -158,6 +160,9 @@ export class FakeApiClient {
jobs: {},
projections: {},
}
assistantStreamBaseline: SessionAssistantStreamBaseline = {
revision: 0,
}
workspaceBaseline: Extract<WorkspaceFollowFrame, { type: 'baseline' }>['value'] = {
items: [],
archivedSessionIds: [],
@@ -276,7 +281,7 @@ export class FakeApiClient {
/** Push one live Session event to every follower of that Session. */
async pushFollow(
sessionId: SessionId,
frame: Extract<SessionFollowFrame, { type: 'event' }>,
frame: Exclude<SessionFollowFrame, { type: 'snapshot' }>,
): Promise<void> {
await Promise.all([...(this.followConns.get(sessionId) ?? [])].map(conn => new Promise<void>((resolve) => {
conn.feed({ kind: 'frame', value: frame, delivered: resolve })
@@ -388,9 +393,10 @@ export class FakeApiClient {
yield {
type: 'snapshot',
header: {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 0,
isSeeded: false,
...(request.address.kind === 'subagent'
? { origin: 'subagent' as const, parentSession: request.address.parentSessionId }
: {}),
@@ -399,6 +405,9 @@ export class FakeApiClient {
records: page.records.filter(record => historyRecordLastSeq(record) <= cursor),
hasMore: page.hasMore,
projections: page.projections ?? { asOfSeq: cursor, values: {} },
...request.assistantStream === true
? { assistantStream: this.assistantStreamBaseline }
: {},
}
yield* stream.values
} finally {
@@ -1,4 +1,4 @@
/** Packed history records become one event-shaped Client value per wire record. */
/** V2 history records become one event-shaped Client value per wire record. */
import { describe, expect, it } from 'vitest'
import { ToolCallId } from '@deepseek-ai/dsh-llm/brand'
@@ -26,53 +26,69 @@ describe('Session history record projection', () => {
expect(historyRecordLastSeq(ordinary)).toBe(7)
})
it('retains one packed text row without copying or reshaping it', () => {
const packed: SessionHistoryRecord = {
type: 'chunks',
it('retains one message with an embedded compact text stream', () => {
const message: SessionHistoryRecord = {
type: 'event',
event: {
type: 'chunkrow/text-chunks',
type: 'assistant/message',
seq: 11,
time: 20,
data: { turn: 1, step: 2, index: 0, dt: [1, 2, 3], texts: ['a', 'b', 'c', 'd'] },
data: {
turn: 1,
step: 2,
message: {
id: 'message-1',
role: 'assistant',
content: [{ type: 'text', text: 'abcd' }],
source: { kind: 'model', provider: 'fixture', model: 'fixture-v2' },
},
stream: [{ type: 'text-chunks', time0: 20, index: 0, dt: [1, 2, 3], texts: ['a', 'b', 'c', 'd'] }],
},
},
}
const [entry] = historyEntries([packed])
if (entry?.type !== 'chunks') throw new Error('expected packed history entry')
const [entry] = historyEntries([message])
if (entry?.type !== 'event') throw new Error('expected v2 history entry')
const { event } = entry
expect(entry).toBe(packed)
expect(event).toBe(packed.event)
expect(historyRecordFirstSeq(packed)).toBe(11)
expect(entry).toBe(message)
expect(event).toBe(message.event)
expect(historyRecordFirstSeq(message)).toBe(11)
expect(event.time).toBe(20)
expect(historyRecordLastSeq(packed)).toBe(14)
expect(historyRecordLastSeq(message)).toBe(11)
})
it('preserves a packed tool-call row and optional-name absence', () => {
const packed: SessionHistoryRecord = {
type: 'chunks',
it('preserves an attempt stream and optional tool-name absence', () => {
const attempt: SessionHistoryRecord = {
type: 'event',
event: {
type: 'chunkrow/tool-call-chunks',
type: 'assistant/attempt',
seq: 20,
time: 200,
data: {
turn: 2,
step: 4,
index: 1,
id: ToolCallId('call-1'),
dt: [2, 3],
args: ['', '{"x":', '1}'],
stream: [{
type: 'tool-call-chunks',
time0: 200,
index: 1,
id: ToolCallId('call-1'),
dt: [2, 3],
args: ['', '{"x":', '1}'],
}],
},
},
}
const [entry] = historyEntries([packed])
if (entry?.type !== 'chunks') throw new Error('expected packed history entry')
const [entry] = historyEntries([attempt])
if (entry?.type !== 'event') throw new Error('expected v2 history entry')
const { event } = entry
if (event.type !== 'chunkrow/tool-call-chunks') throw new Error('expected packed history event')
expect(event).toBe(packed.event)
expect(Object.hasOwn(event.data, 'name')).toBe(false)
expect(historyRecordLastSeq(packed)).toBe(22)
if (event.type !== 'assistant/attempt') throw new Error('expected Assistant attempt event')
expect(event).toBe(attempt.event)
const [record] = event.data.stream
expect(record).toMatchObject({ type: 'tool-call-chunks', id: 'call-1' })
expect(Object.hasOwn(record as object, 'name')).toBe(false)
expect(historyRecordLastSeq(attempt)).toBe(20)
})
})
@@ -310,6 +310,7 @@ describe('subagent catalogs', () => {
address: {
kind: 'subagent', parentSessionId: S1, childSessionId: S2, mode: 'continuable',
},
assistantStream: true,
maxMessages: 50,
},
])
@@ -73,21 +73,29 @@ describe('Session queue snapshot intake', () => {
])
})
it('marks mixed-content messages non-editable and keeps image blocks out of the text preview', () => {
it('marks mixed-content messages non-editable and keeps attachment blocks out of the text preview', () => {
const session = makeSession()
session.handleControlFrame(queueFrame([{
id: 'q-image',
body: '',
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
content: [
{ type: 'text', text: 'hi' },
{ type: 'image', data: 'x' } as never,
{ type: 'file', attachment: { attachmentId: 'file-1', name: 'notes.txt', bytes: 5 } } as never,
],
}]))
const queue = session.getSnapshot().queue
expect(typeof queue[0]?.messageId).toBe('string')
expect(queue).toMatchObject([
{
id: 'q-image', placement: 'queued',
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' }],
// Image blocks render as thumbnails from `content`, so the preview
// carries only the text; non-image foreign blocks keep their marker.
content: [
{ type: 'text', text: 'hi' },
{ type: 'image', data: 'x' },
{ type: 'file', attachment: { attachmentId: 'file-1', name: 'notes.txt', bytes: 5 } },
],
// Attachment blocks render from `content`, so the preview carries
// only text; other foreign blocks keep their marker.
preview: 'hi', text: null,
},
])
@@ -4,11 +4,11 @@
* isolation, and prompt failure mapping.
*/
import { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import SessionStore from '@deepseek-ai/dsh-session'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/src/projection.ts'
import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
@@ -21,10 +21,8 @@ import {
SessionPersistenceRevision,
type SessionPersistenceSnapshot,
} from '@deepseek-ai/dsh-session-persistence'
import { ApiSessionList } from '../src/list.ts'
import {
createSessionTestRemote,
installSessionReadTestServices,
testSessionPersistence,
} from './test-remote.ts'
@@ -44,8 +42,12 @@ function promptRequest(
}
}
function inboxFor(session: Session): Inbox {
return new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
}
function header(id: string, createdAt: number, extra: Partial<SessionHeader> = {}): SessionHeader {
return { version: 0, id: sid(id), createdAt, isSeeded: false, cwd: '/proj', ...extra }
return { version: SESSION_FORMAT_VERSION, id: sid(id), createdAt, isSeeded: false, cwd: '/proj', ...extra }
}
function providePersistence(ctx: Context, persistence: Record<string, unknown>): () => void {
@@ -59,11 +61,6 @@ function statSnapshot(
return { header: meta, revision: SessionPersistenceRevision(`test:${meta.id}:stat`), ...metrics }
}
/** A stored log whose only event is the seed boundary: still blank. */
function blankEvents(): SessionEvent[] {
return [{ type: 'session/end-seed', seq: SessionSeq(0), time: 700, data: {} }] as SessionEvent[]
}
/** A stored log with one human prompt at time 1200: proven non-blank. */
function conversationEvents(): SessionEvent[] {
return [
@@ -77,15 +74,65 @@ function conversationEvents(): SessionEvent[] {
}
describe('sessions.list cold merge', () => {
it('serves cold rows from cached projections when stat offers no size metadata', async () => {
it('uses a predecessor title hint with zero cold stat or body reads', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const metas = [
const metas = [header('legacy-title', 100), header('uncached', 200)]
const stat = vi.fn(async (id: SessionId) => statSnapshot(
metas.find(meta => meta.id === id)!,
{ sizeBytes: 1 },
))
const inspect = vi.fn(async (id: SessionId) => ({
meta: metas.find(meta => meta.id === id)!,
events: conversationEvents(),
}))
providePersistence(ctx, {
list: () => Promise.resolve(metas),
stat,
inspect,
})
ctx.provide('sessionProjectionCache', {
cachedSnapshot: () => undefined,
cachedPredecessorTitle: (meta: SessionHeader) => meta.id === sid('legacy-title')
? { asOfSeq: -1, values: { title: 'Cached predecessor title' } }
: undefined,
} as never)
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
cwd: '/tmp',
})
const observe = vi.spyOn(ctx.sessionQuery, 'observeSession')
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({
sessionId: sid('uncached'),
blank: false,
updatedAt: 200,
}),
expect.objectContaining({
sessionId: sid('legacy-title'),
blank: false,
updatedAt: 100,
projections: { asOfSeq: -1, values: { title: 'Cached predecessor title' } },
}),
])
expect(stat).not.toHaveBeenCalled()
expect(inspect).not.toHaveBeenCalled()
expect(observe).not.toHaveBeenCalled()
})
it('serves cold rows from current cached projections without body access', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const metas: SessionHeader[] = [
header('cached-blank', 100),
header('cached-conversation', 200),
header('uncached', 300, { parentSession: sid('session-parent'), origin: 'subagent' }),
header('seeded-cold', 450, { isSeeded: true }),
{ version: 0, id: sid('missing-cwd'), createdAt: 800, isSeeded: false },
{ version: SESSION_FORMAT_VERSION, id: sid('missing-cwd'), createdAt: 800, isSeeded: false },
]
const inspect = vi.fn()
providePersistence(ctx, {
@@ -104,6 +151,7 @@ describe('sessions.list cold merge', () => {
}
return undefined
},
cachedPredecessorTitle: () => undefined,
} as never)
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
@@ -113,7 +161,7 @@ describe('sessions.list cold merge', () => {
const byId = Object.fromEntries(response.value.items.map(item => [item.sessionId, item]))
expect(byId['cached-blank']).toMatchObject({ blank: true, updatedAt: 100, running: false })
expect(byId['cached-conversation']).toMatchObject({ blank: false, updatedAt: 1000 })
// A cache miss with a metadata-less stat leaves blankness unknown; the row stays visible.
// A cache miss leaves blankness unknown; the row stays visible without a body read.
expect(byId['uncached']).toMatchObject({
blank: false,
updatedAt: 300,
@@ -128,362 +176,6 @@ describe('sessions.list cold merge', () => {
expect(inspect).not.toHaveBeenCalled()
})
it('fully observes only small possibly-blank logs gated by stat eventCount', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const metas = [
header('small-blank', 100),
header('small-conversation', 200),
header('large-unknown', 300),
header('cached-nonblank', 400),
header('vanished', 600),
{ version: 0, id: sid('missing-cwd'), createdAt: 800, isSeeded: false },
]
const inspect = vi.fn(async (id: SessionId) => {
if (id === sid('small-blank')) return { meta: metas[0]!, events: blankEvents() }
if (id === sid('small-conversation')) return { meta: metas[1]!, events: conversationEvents() }
throw new Error(`unexpected cold read: ${id}`)
})
const stat = vi.fn(async (id: SessionId) => {
if (id === sid('small-blank')) return statSnapshot(metas[0]!, { eventCount: 1 })
if (id === sid('small-conversation')) return statSnapshot(metas[1]!, { eventCount: 2 })
if (id === sid('large-unknown')) return statSnapshot(metas[2]!, { eventCount: 17 })
if (id === sid('vanished')) return undefined
throw new Error(`unexpected stat: ${id}`)
})
providePersistence(ctx, {
list: () => Promise.resolve(metas),
stat,
inspect,
})
ctx.provide('sessionProjectionCache', {
cachedSnapshot: (meta: SessionHeader) => {
if (meta.id === sid('small-blank')) {
return { asOfSeq: 0, values: { sessionListMetadata: { blank: true, lastPromptAt: null } } }
}
if (meta.id === sid('small-conversation')) {
return { asOfSeq: 0, values: { sessionListMetadata: { blank: true, lastPromptAt: 900 } } }
}
if (meta.id === sid('cached-nonblank')) {
return { asOfSeq: 1, values: { sessionListMetadata: { blank: false, lastPromptAt: 1000 } } }
}
return undefined
},
hydratePrepared: (session: Session, events: readonly SessionEvent[]) =>
ctx.sessionProjections.hydrate(session, {}, events, SessionLogOffset(0)),
} as never)
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const response = await remote.list(request({}))
expect(response.ok).toBe(true)
if (!response.ok) throw new Error('unreachable')
const byId = Object.fromEntries(response.value.items.map(item => [item.sessionId, item]))
expect(byId['small-blank']).toMatchObject({ blank: true, updatedAt: 100, running: false })
expect(byId['small-conversation']).toMatchObject({ blank: false, updatedAt: 1200 })
expect(byId['large-unknown']).toMatchObject({ blank: false, updatedAt: 300 })
expect(byId['cached-nonblank']).toMatchObject({ blank: false, updatedAt: 1000 })
expect(byId['vanished']).toMatchObject({ blank: false, updatedAt: 600 })
expect(byId['missing-cwd']).toBeUndefined()
// A cache row proving blank:false is never re-probed.
expect(stat.mock.calls.map(([id]) => id)).not.toContain(sid('cached-nonblank'))
expect(inspect).toHaveBeenCalledTimes(2)
expect(inspect.mock.calls.map(([id]) => id)).toEqual(expect.arrayContaining([
sid('small-blank'),
sid('small-conversation'),
]))
})
it('falls back to the stat sizeBytes gate when no eventCount is offered', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const metas = [header('small-jsonl', 100), header('large-jsonl', 200)]
const inspect = vi.fn(async (id: SessionId) => {
if (id === sid('small-jsonl')) return { meta: metas[0]!, events: conversationEvents() }
throw new Error(`unexpected cold read: ${id}`)
})
providePersistence(ctx, {
list: () => Promise.resolve(metas),
stat: (id: SessionId) => Promise.resolve(id === sid('small-jsonl')
? statSnapshot(metas[0]!, { sizeBytes: 1024 })
: statSnapshot(metas[1]!, { sizeBytes: 1025 })),
inspect,
})
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
const byId = Object.fromEntries(response.value.items.map(item => [item.sessionId, item]))
expect(byId['small-jsonl']).toMatchObject({ blank: false, updatedAt: 1200 })
expect(byId['large-jsonl']).toMatchObject({ blank: false, updatedAt: 200 })
expect(inspect).toHaveBeenCalledTimes(1)
expect(inspect).toHaveBeenCalledWith(sid('small-jsonl'), expect.anything())
})
it('skips the observation when stat offers neither eventCount nor sizeBytes', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('no-metrics', 100)
const inspect = vi.fn()
const stat = vi.fn(async () => statSnapshot(meta))
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat,
inspect,
})
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false, updatedAt: meta.createdAt }),
])
expect(stat).toHaveBeenCalledOnce()
expect(inspect).not.toHaveBeenCalled()
})
it('can disable both probe gates without hiding cold Sessions or calling stat', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('probe-disabled', 100)
const inspect = vi.fn()
const stat = vi.fn()
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat,
inspect,
})
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
cwd: '/tmp',
coldBlankProbeMaxEvents: 0,
coldBlankProbeMaxBytes: 0,
})
const response = await remote.list(request({}))
if (!response.ok) throw new Error('unreachable')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false, updatedAt: meta.createdAt }),
])
expect(stat).not.toHaveBeenCalled()
expect(inspect).not.toHaveBeenCalled()
})
it('treats zero as disabling one gate without falling back to the other metric', async () => {
const bench = async (
metrics: Partial<Pick<SessionPersistenceSnapshot, 'eventCount' | 'sizeBytes'>>,
thresholds: { coldBlankProbeMaxEvents?: number; coldBlankProbeMaxBytes?: number },
) => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('gate-off', 100)
const inspect = vi.fn()
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat: () => Promise.resolve(statSnapshot(meta, metrics)),
inspect,
})
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
cwd: '/tmp',
...thresholds,
})
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false }),
])
expect(inspect).not.toHaveBeenCalled()
}
// An offered eventCount never falls through to the byte gate, even disabled.
await bench({ eventCount: 1, sizeBytes: 10 }, { coldBlankProbeMaxEvents: 0 })
await bench({ sizeBytes: 10 }, { coldBlankProbeMaxBytes: 0 })
})
it('prefers a Session that attaches during its bounded cold observation', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
const meta = header('attached-during-probe', 100)
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat: () => {
const session = ctx.sessions.create(meta.id, {
meta,
seed: [{ type: 'turn/start', seq: SessionSeq(0), time: 200, data: { turn: 1 } }],
})
ctx.agents.register({ id: session.id, session, status: 'running', ctx } as Agent)
return Promise.resolve(statSnapshot(meta, { eventCount: 1 }))
},
})
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp',
})
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, running: true, blank: false }),
])
await ctx.fiber.dispose()
})
it('serves a session whose cold stat fails as visible instead of failing the list', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('broken-stat', 100)
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat: () => Promise.reject(new Error('stat failed')),
})
const warned = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp',
})
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, running: false }),
])
expect(warned.mock.calls.join('\n')).toContain('cold stat for "broken-stat" failed')
warned.mockRestore()
await ctx.fiber.dispose()
})
it('a stat rejection after cancellation propagates instead of degrading', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('stat-abort', 100)
const controller = new AbortController()
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat: () => {
controller.abort(new Error('caller left'))
return Promise.reject(new Error('stat failed'))
},
})
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp',
})
await expect(remote.list(request({}), controller.signal)).resolves.toMatchObject({
ok: false,
error: { code: 'gateway/cancelled' },
})
await ctx.fiber.dispose()
})
it('serves a small cold Session as visible when its observation fails', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const meta = header('read-failure', 700)
const inspect = vi.fn(async () => { throw new Error('simulated read failure') })
providePersistence(ctx, {
list: () => Promise.resolve([meta]),
stat: () => Promise.resolve(statSnapshot(meta, { eventCount: 1 })),
inspect,
})
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp',
})
const response = await remote.list(request({}))
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false, updatedAt: 700 }),
])
expect(inspect).toHaveBeenCalledOnce()
})
it('supports an unsignalled probe whose observation has no projection block', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
installSessionReadTestServices(ctx)
const meta = header('unprojected-small', 100)
ctx.provide('sessionPersistence', {
list: () => Promise.resolve([meta]),
stat: () => Promise.resolve(statSnapshot(meta, { eventCount: 0 })),
} as never)
vi.spyOn(ctx.sessionQuery, 'listSessions').mockResolvedValue([{
header: meta, live: false, persisted: true,
}])
vi.spyOn(ctx.sessionQuery, 'observeSession').mockResolvedValue({
source: 'prepared', header: meta, inheritedEventCount: SessionLogOffset(0), events: [], cursor: -1,
retain: vi.fn(), [Symbol.dispose]: vi.fn(),
})
const list = new ApiSessionList(ctx, { coldBlankProbeMaxEvents: 16, coldBlankProbeMaxBytes: 1024 })
await expect(list.list()).resolves.toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false }),
])
await ctx.fiber.dispose()
})
it('serves a cold row visible when no persistence service can stat it', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
installSessionReadTestServices(ctx)
const meta = header('service-less', 100)
vi.spyOn(ctx.sessionQuery, 'listSessions').mockResolvedValue([{
header: meta, live: false, persisted: true,
}])
const list = new ApiSessionList(ctx, { coldBlankProbeMaxEvents: 16, coldBlankProbeMaxBytes: 1024 })
await expect(list.list()).resolves.toEqual([
expect.objectContaining({ sessionId: meta.id, blank: false }),
])
await ctx.fiber.dispose()
})
it('prefers a live row attached during the query without folding its seed', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
const meta = header('attached-during-list', 100)
const started = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
providePersistence(ctx, {
list: async () => {
started.resolve(undefined)
await release.promise
return [meta]
},
})
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const listing = remote.list(request({}))
await started.promise
const session = ctx.sessions.create(meta.id, {
seed: [
{ type: 'turn/start', seq: SessionSeq(0), time: 200, data: { turn: 1 } },
{
type: 'user/message', seq: SessionSeq(1), time: 300,
data: createUserMessage({ content: [{ type: 'text', text: 'live' }], source: { kind: 'user' } }),
surfaceOp: 'append',
},
],
meta: {
...meta.cwd === undefined ? {} : { cwd: meta.cwd },
createdAt: meta.createdAt,
},
})
ctx.agents.register({ id: session.id, session, status: 'running', ctx } as Agent)
release.resolve(undefined)
const response = await listing
if (!response.ok) throw new Error('list failed')
expect(response.value.items).toEqual([
expect.objectContaining({
sessionId: meta.id,
blank: false,
running: true,
updatedAt: 100,
}),
])
})
})
describe('attached updatedAt tracks human prompts', () => {
@@ -852,7 +544,9 @@ describe('subagent ownership fence', () => {
inheritedEventCount: SessionLogOffset(1),
})
const followup = vi.fn()
const agent = { id: session.id, session, status: 'idle', ctx, followup } as unknown as Agent
const agent = {
id: session.id, session, inbox: inboxFor(session), status: 'idle', ctx, followup,
} as unknown as Agent
ctx.agents.register(agent)
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
@@ -871,7 +565,9 @@ describe('subagent ownership fence', () => {
await ctx.plugin(AgentRegistry)
const session = ctx.sessions.create(sid('session-browser-zone'), { meta: { cwd: '/proj' } })
const followup = vi.fn()
const agent = { id: session.id, session, status: 'idle', ctx, followup } as unknown as Agent
const agent = {
id: session.id, session, inbox: inboxFor(session), status: 'idle', ctx, followup,
} as unknown as Agent
ctx.agents.register(agent)
const remote = createSessionTestRemote(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
@@ -991,6 +687,7 @@ describe('sessions.prompt synchronous rejection', () => {
ctx.agents.register({
id: session.id,
session,
inbox: inboxFor(session),
status: 'idle',
ctx,
followup: () => { throw new Error('agent "session-throwing" lifecycle disposed') },
@@ -6,7 +6,7 @@ import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentHandle, CreateAgentOptions } from '@deepseek-ai/dsh-agent'
import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import type { Workspace } from '@deepseek-ai/dsh-workspace'
@@ -151,7 +151,7 @@ describe('sessions.fork', () => {
const sourceId = sid('session-cold-subagent')
const parentId = sid('session-cold-parent')
const header: SessionHeader = {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sourceId,
createdAt: 1,
cwd: '/proj',
@@ -2,18 +2,12 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import SessionStore, { SessionSeq } from '@deepseek-ai/dsh-session'
import { decodeStorageRecord, type ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
import { ToolCallId, createMessage, createToolResultMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
import AgentRegistry, { type Agent, type AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import { LlmAttemptId, ToolCallId, createMessage, createToolResultMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
import type {
ChunkRowEvent,
SessionFollowFrame,
SessionPage,
SessionWireEvent,
} from '@deepseek-ai/dsh-api-session-controller/types'
import type { SessionFollowFrame, SessionPage, SessionWireEvent } from '@deepseek-ai/dsh-api-session-controller/types'
import { createSessionTestRemote, installSessionReadTestServices } from './test-remote.ts'
/** Append a production-shaped human prompt to the session surface. */
@@ -33,6 +27,7 @@ function appendAssistantText(session: Session, text: string, step: number): Sess
content: [{ type: 'text', text }],
source: { kind: 'model', provider: 'p', model: 'm' },
}),
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: [text] }],
}, { surfaceOp: 'append' })
}
@@ -83,25 +78,571 @@ async function openFollow(
return { [Symbol.asyncIterator]: () => iterator }
}
/** Expand packed page records for assertions over the logical journal. */
function pageEvents(page: SessionPage): SessionWireEvent[] {
return page.records.flatMap(record => record.type === 'event'
? [record.event]
: decodeStorageRecord(chunkRow(record.event)).map(event => event as unknown as SessionWireEvent))
/** Abort one follow and await both its iterator and owning Context teardown. */
async function disposeFollow(
ctx: Context,
iterator: AsyncIterator<SessionFollowFrame>,
abort: AbortController,
): Promise<void> {
abort.abort()
await iterator.return?.()
await ctx.fiber.dispose()
}
function chunkRow(event: ChunkRowEvent): ChunkRow {
switch (event.type) {
case 'chunkrow/text-chunks':
return { type: 'text-chunks', seq0: SessionSeq(event.seq), time0: event.time, data: event.data }
case 'chunkrow/reasoning-chunks':
return { type: 'reasoning-chunks', seq0: SessionSeq(event.seq), time0: event.time, data: event.data }
case 'chunkrow/tool-call-chunks':
return { type: 'tool-call-chunks', seq0: SessionSeq(event.seq), time0: event.time, data: event.data }
}
/** Read scalar v2 page records for assertions over the logical journal. */
function pageEvents(page: SessionPage): SessionWireEvent[] {
return page.records.map(record => record.event)
}
describe('Session history raw journal', () => {
it('opens an empty opted-in Assistant baseline before any live attempt exists', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: { type: 'snapshot', assistantStream: { revision: 0 } },
})
abort.abort()
await iterator.next()
await ctx.fiber.dispose()
})
it('filters foreign and opening-baseline frames buffered during the source observation', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
const observe = vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation(async (...args) => {
entered.resolve(undefined)
await release.promise
return originalObserve(...args)
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
const opening = iterator.next()
await entered.promise
const attemptId = LlmAttemptId('buffered-opening-attempt')
ctx.emit('agent/assistant-stream', {
agent,
frame: { type: 'start', attemptId, revision: 1, turn: 1, step: 1 },
})
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'chunk', attemptId, revision: 2, index: 0,
time: 2, chunk: { type: 'text-delta', index: 0, text: 'buffered' },
},
})
const foreign = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
ctx.emit('agent/assistant-stream', {
agent: { id: foreign.id, session: foreign, status: 'running', ctx } as Agent,
frame: {
type: 'start', attemptId: LlmAttemptId('foreign-attempt'), revision: 1, turn: 1, step: 1,
},
})
release.resolve(undefined)
await expect(opening).resolves.toMatchObject({
done: false,
value: { type: 'snapshot', assistantStream: { revision: 2 } },
})
const next = iterator.next()
const durable = session.append('turn/start', { turn: 1 })
await expect(next).resolves.toEqual({ done: false, value: { type: 'event', event: durable } })
abort.abort()
await iterator.next()
observe.mockRestore()
await ctx.fiber.dispose()
})
it('opens an opted-in assistant baseline and preserves mixed live FIFO order', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId('live-follow-attempt')
const emit = (frame: AssistantStreamFrame): void => {
ctx.emit('agent/assistant-stream', { agent, frame })
}
emit({
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
})
const firstChunk = { type: 'text-delta', index: 0, text: 'a' } as const
emit({
type: 'chunk', attemptId, revision: 2, index: 0,
time: 1, chunk: firstChunk,
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: ['a'] }],
},
},
},
})
const nextFrame: AssistantStreamFrame = {
type: 'chunk', attemptId, revision: 3, index: 1,
time: 2, chunk: { type: 'text-delta', index: 0, text: 'b' },
}
emit(nextFrame)
const message = appendAssistantText(session, 'ab', 1)
const endFrame: AssistantStreamFrame = {
type: 'end', attemptId, revision: 4, index: 2,
outcome: { kind: 'committed', eventType: 'assistant/message', seq: message.seq },
}
emit(endFrame)
await expect(iterator.next()).resolves.toEqual({
done: false, value: { type: 'assistant-stream', frame: nextFrame },
})
await expect(iterator.next()).resolves.toEqual({
done: false, value: { type: 'event', event: message },
})
await expect(iterator.next()).resolves.toEqual({
done: false, value: { type: 'assistant-stream', frame: endFrame },
})
abort.abort()
await iterator.next()
await ctx.fiber.dispose()
})
it('forwards revision one when the attached Agent lifecycle restarts after opening', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId(`${session.id}:1`)
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const oldChunk = { type: 'text-delta', index: 0, text: 'old' } as const
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'chunk', attemptId, revision: 2, index: 0,
time: 101, chunk: oldChunk,
},
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 101, index: 0, dt: [], texts: ['old'] }],
},
},
},
})
ctx.emit('agent/disposed', { agent })
const replacementAgent = { id: session.id, session, status: 'running', ctx } as Agent
const replacement: AssistantStreamFrame = {
type: 'start', attemptId, revision: 1,
turn: 2, step: 1,
}
ctx.emit('agent/assistant-stream', { agent: replacementAgent, frame: replacement })
await expect(iterator.next()).resolves.toEqual({
done: false,
value: { type: 'assistant-stream', frame: { ...replacement, startedAfterSeq: -1 } },
})
} finally {
await disposeFollow(ctx, iterator, abort)
}
})
it('publishes an empty replacement baseline after an Agent frame revision gap', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId('revision-gap-attempt')
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const chunk = { type: 'text-delta', index: 0, text: 'after gap' } as const
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'chunk', attemptId, revision: 3, index: 0,
time: 101, chunk,
},
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: { revision: 3 },
},
})
} finally {
await disposeFollow(ctx, iterator, abort)
}
})
it('drops active attempts when an Agent chunk index is not dense', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId('dense-index-attempt')
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const chunk = { type: 'text-delta', index: 0, text: 'out of order' } as const
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'chunk', attemptId, revision: 2, index: 1,
time: 101, chunk,
},
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: { revision: 2 },
},
})
} finally {
await disposeFollow(ctx, iterator, abort)
}
})
it('reuses an unchanged Assistant baseline across follow openings', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId('cached-baseline-attempt')
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const firstAbort = new AbortController()
const firstIterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, firstAbort.signal)[Symbol.asyncIterator]()
const secondAbort = new AbortController()
const secondIterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, secondAbort.signal)[Symbol.asyncIterator]()
try {
const first = await firstIterator.next()
if (first.done || first.value.type !== 'snapshot') throw new Error('first follow did not open')
const baseline = first.value.assistantStream
expect(baseline).toMatchObject({ revision: 1, activeAttempt: { attemptId } })
const second = await secondIterator.next()
if (second.done || second.value.type !== 'snapshot') throw new Error('second follow did not open')
expect(second.value.assistantStream).toEqual(baseline)
} finally {
firstAbort.abort()
secondAbort.abort()
await firstIterator.return?.()
await secondIterator.return?.()
await ctx.fiber.dispose()
}
})
it('opens an empty Assistant baseline before the target Agent emits frames', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: { revision: 0 },
},
})
} finally {
await disposeFollow(ctx, iterator, abort)
}
})
it('filters Assistant frames from another Session out of the target follow', async () => {
const { ctx } = await harness()
const target = ctx.sessions.create(undefined, { meta: { cwd: '/target' } })
const other = ctx.sessions.create(undefined, { meta: { cwd: '/other' } })
const otherAgent = { id: other.id, session: other, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: target.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
await expect(iterator.next()).resolves.toMatchObject({
done: false,
value: { type: 'snapshot' },
})
ctx.emit('agent/assistant-stream', {
agent: otherAgent,
frame: {
type: 'start', attemptId: LlmAttemptId('other-session-attempt'),
revision: 1, turn: 1, step: 1,
},
})
const targetEvent = target.append('turn/start', { turn: 1 })
await expect(iterator.next()).resolves.toEqual({
done: false,
value: { type: 'event', event: targetEvent },
})
} finally {
await disposeFollow(ctx, iterator, abort)
}
})
it('does not replay a buffered Assistant frame already represented by the opening baseline', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const observationStarted = Promise.withResolvers<undefined>()
const releaseObservation = Promise.withResolvers<undefined>()
const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
const observe = vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation(async (sessionId, options) => {
observationStarted.resolve(undefined)
await releaseObservation.promise
return await originalObserve(sessionId, options)
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
const opening = iterator.next()
await observationStarted.promise
const frame: AssistantStreamFrame = {
type: 'start', attemptId: LlmAttemptId('opening-cut-attempt'),
revision: 1, turn: 1, step: 1,
}
ctx.emit('agent/assistant-stream', { agent, frame })
releaseObservation.resolve(undefined)
await expect(opening).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: { revision: 1, activeAttempt: { attemptId: frame.attemptId } },
},
})
const durable = session.append('turn/start', { turn: 1 })
await expect(iterator.next()).resolves.toEqual({
done: false,
value: { type: 'event', event: durable },
})
} finally {
releaseObservation.resolve(undefined)
observe.mockRestore()
abort.abort()
await iterator.return?.()
await ctx.fiber.dispose()
}
})
it('does not release an old-lifecycle frame after the opening baseline resets to revision one', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const attemptId = LlmAttemptId(`${session.id}:1`)
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const observationStarted = Promise.withResolvers<undefined>()
const releaseObservation = Promise.withResolvers<undefined>()
const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
const observe = vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation(async (sessionId, options) => {
observationStarted.resolve(undefined)
await releaseObservation.promise
return await originalObserve(sessionId, options)
})
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
assistantStream: true,
}, abort.signal)[Symbol.asyncIterator]()
try {
const opening = iterator.next()
await observationStarted.promise
const oldChunk = { type: 'text-delta', index: 0, text: 'old lifecycle' } as const
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'chunk', attemptId, revision: 2, index: 0,
time: 101, chunk: oldChunk,
},
})
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 2, step: 1,
},
})
releaseObservation.resolve(undefined)
await expect(opening).resolves.toMatchObject({
done: false,
value: {
type: 'snapshot',
assistantStream: {
revision: 1,
activeAttempt: {
attemptId, startedAfterSeq: -1,
turn: 2, step: 1, nextIndex: 0, stream: [],
},
},
},
})
const durable = session.append('turn/start', { turn: 2 })
await expect(iterator.next()).resolves.toEqual({
done: false,
value: { type: 'event', event: durable },
})
} finally {
releaseObservation.resolve(undefined)
observe.mockRestore()
abort.abort()
await iterator.return?.()
await ctx.fiber.dispose()
}
})
it('keeps assistant frames out of a durable-only follower', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const agent = { id: session.id, session, status: 'running', ctx } as Agent
const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
const abort = new AbortController()
const iterator = history.follow({
address: { kind: 'session', sessionId: session.id },
}, abort.signal)[Symbol.asyncIterator]()
const opening = await iterator.next()
expect(opening.value).not.toHaveProperty('assistantStream')
const attemptId = LlmAttemptId('durable-only-attempt')
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'start', attemptId, revision: 1,
turn: 1, step: 1,
},
})
const durable = session.append('turn/start', { turn: 1 })
ctx.emit('agent/assistant-stream', {
agent,
frame: {
type: 'end', attemptId, revision: 2, index: 0, outcome: { kind: 'abandoned' },
},
})
const next = session.append('turn/end', {
turn: 1, reason: { kind: 'completed' },
})
await expect(iterator.next()).resolves.toEqual({
done: false, value: { type: 'event', event: durable },
})
await expect(iterator.next()).resolves.toEqual({
done: false, value: { type: 'event', event: next },
})
abort.abort()
await iterator.next()
await ctx.fiber.dispose()
})
it('follows raw tool events and preserves result metadata without a Tools service', async () => {
const { ctx } = await harness()
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
@@ -254,25 +795,23 @@ describe('Session history raw journal', () => {
expect(page.map(event => event.seq)).toEqual(page.map((_event, index) => third.seq + index))
})
it('paginates a message with many provenance sources without variadic argument expansion', async () => {
it('paginates a message with a large embedded stream without expanding physical records', async () => {
const { ctx } = await harness()
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
session.append('turn/start', { turn: 1 })
const sources = Array.from({ length: 128 }, () => session.append('assistant/chunk', {
turn: 1,
step: 1,
chunk: { type: 'text-delta', index: 0, text: 'x' },
}).seq)
session.append('step/start', { turn: 1, step: 1 })
const texts = Array.from({ length: 128 }, () => 'x')
const message = session.append('assistant/message', {
turn: 1,
step: 1,
message: createMessage({
role: 'assistant',
content: [{ type: 'text', text: 'x'.repeat(sources.length) }],
content: [{ type: 'text', text: 'x'.repeat(texts.length) }],
source: { kind: 'model', provider: 'p', model: 'm' },
}),
}, { surfaceOp: 'append', sourceEventSeqs: sources })
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: texts.slice(1).map(() => 0), texts }],
}, { surfaceOp: 'append' })
const scalarMin = Math.min
const min = vi.spyOn(Math, 'min').mockImplementation((...values) => {
@@ -286,68 +825,55 @@ describe('Session history raw journal', () => {
maxMessages: 1,
})
if (!response.ok) throw new Error('unreachable')
expect(pageEvents(response.value).map(event => event.seq)).toEqual([...sources, message.seq])
expect(response.value.records.filter(record => record.type === 'chunks')).toHaveLength(1)
expect(pageEvents(response.value).map(event => event.seq)).toEqual([message.seq])
expect(response.value.records).toEqual([{ type: 'event', event: message }])
expect(response.value.hasMore).toBe(true)
} finally {
min.mockRestore()
}
})
it('encodes reasoning and tool-call runs as aligned chunk events', async () => {
it('keeps an earlier declared source on the same message-aligned page', async () => {
const { ctx } = await harness()
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const source = session.append('request/context', { provider: 'p', model: 'm' })
const laterSource = session.append('request/context', { provider: 'p', model: 'm' })
const message = session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'with source' }], source: { kind: 'user' },
}), { sourceEventSeqs: [source.seq, laterSource.seq], surfaceOp: 'append' })
const response = await remote.page({
address: { kind: 'session', sessionId: session.id },
throughSeq: message.seq,
maxMessages: 1,
})
if (!response.ok) throw new Error('unreachable')
expect(pageEvents(response.value).map(event => event.seq)).toEqual([source.seq, laterSource.seq, message.seq])
expect(response.value.hasMore).toBe(false)
await ctx.fiber.dispose()
})
it('keeps compact reasoning and tool-call runs nested in one attempt event', async () => {
const { ctx } = await harness()
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
const reasoning = [0, 1, 2].map(index => session.append('assistant/chunk', {
turn: 1,
step: 1,
chunk: { type: 'reasoning-delta', index: 0, text: `r${String(index)}` },
}))
const callId = ToolCallId('packed-call')
const toolCall = [0, 1, 2].map(index => session.append('assistant/chunk', {
const attempt = session.append('assistant/attempt', {
turn: 1,
step: 1,
chunk: { type: 'tool-call-delta', index: 1, id: callId, argumentsDelta: `a${String(index)}` },
}))
stream: [
{ type: 'reasoning-chunks', time0: 1, index: 0, dt: [1, 1], texts: ['r0', 'r1', 'r2'] },
{ type: 'tool-call-chunks', time0: 4, index: 1, id: callId, dt: [1, 1], args: ['a0', 'a1', 'a2'] },
],
})
const response = await remote.page({
address: { kind: 'session', sessionId: session.id },
throughSeq: session.seq - 1,
})
if (!response.ok) throw new Error('unreachable')
expect(response.value.records).toEqual([
{
type: 'chunks',
event: {
type: 'chunkrow/reasoning-chunks',
seq: reasoning[0]?.seq,
time: reasoning[0]?.time,
data: {
turn: 1,
step: 1,
index: 0,
dt: reasoning.slice(1).map((event, index) => event.time - (reasoning[index]?.time ?? 0)),
texts: ['r0', 'r1', 'r2'],
},
},
},
{
type: 'chunks',
event: {
type: 'chunkrow/tool-call-chunks',
seq: toolCall[0]?.seq,
time: toolCall[0]?.time,
data: {
turn: 1,
step: 1,
index: 1,
id: callId,
dt: toolCall.slice(1).map((event, index) => event.time - (toolCall[index]?.time ?? 0)),
args: ['a0', 'a1', 'a2'],
},
},
},
])
expect(response.value.records).toEqual([{ type: 'event', event: attempt }])
await ctx.fiber.dispose()
})
@@ -665,7 +665,7 @@ describe('Web session model selection', () => {
const savedRef = {
attachmentId: 'saved-image', mediaType: 'image/png' as const, bytes: 1, width: 1, height: 1,
}
ctx.provide('attachments', {
ctx.provide('attachments', Object.setPrototypeOf({
saveImages: () => {
if (saveMode === 'error') return Promise.reject(new Error('image store offline'))
if (saveMode === 'remote') {
@@ -673,7 +673,7 @@ describe('Web session model selection', () => {
}
return Promise.resolve([savedRef])
},
} as never)
}, AttachmentStore.prototype) as never)
const followup = vi.fn()
Object.assign(agent, { followup })
const remote = createSessionTestRemote(ctx, {
@@ -2,7 +2,7 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
import type { FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
import { SessionSeq, type SessionEvent, type SessionId } from '@deepseek-ai/dsh-session/types'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { Session } from '../src/client/sessions/session.ts'
@@ -31,8 +31,20 @@ function imageRef(id: string): ImageAttachmentRef {
} as unknown as ImageAttachmentRef
}
function fileRef(id: string, name = 'notes.txt'): FileAttachmentRef {
return { attachmentId: id, name, bytes: 3 } as unknown as FileAttachmentRef
}
type AttachmentRef = ImageAttachmentRef | FileAttachmentRef
function attachmentBlock(attachment: AttachmentRef) {
return 'mediaType' in attachment
? { type: 'image' as const, attachment }
: { type: 'file' as const, attachment }
}
/** A durable browser-prompt user/message whose source echoes `rpcId`. */
function promptEvent(seq: SessionSeq, rpcId: SessionRequestId, refs: readonly ImageAttachmentRef[] = []): SessionEvent {
function promptEvent(seq: SessionSeq, rpcId: SessionRequestId, refs: readonly AttachmentRef[] = []): SessionEvent {
return {
seq,
time: 1_700_000_000_000 + seq,
@@ -40,7 +52,7 @@ function promptEvent(seq: SessionSeq, rpcId: SessionRequestId, refs: readonly Im
surfaceOp: 'append',
data: createUserMessage({
content: [
...refs.map(attachment => ({ type: 'image' as const, attachment })),
...refs.map(attachmentBlock),
{ type: 'text' as const, text: '发送' },
],
source: { kind: 'user', rpcId },
@@ -48,14 +60,14 @@ function promptEvent(seq: SessionSeq, rpcId: SessionRequestId, refs: readonly Im
} as unknown as SessionEvent
}
function queuedItem(rpcId: SessionRequestId, refs: readonly ImageAttachmentRef[] = []): SessionQueuedItem {
function queuedItem(rpcId: SessionRequestId, refs: readonly AttachmentRef[] = []): SessionQueuedItem {
return {
id: 'm-queued' as SessionQueuedItem['id'],
placement: 'queued',
rpcId,
message: {
id: 'm-queued' as SessionQueuedItem['id'],
content: refs.map(attachment => ({ type: 'image', attachment })) as unknown as SessionQueuedItem['message']['content'],
content: refs.map(attachmentBlock) as unknown as SessionQueuedItem['message']['content'],
},
}
}
@@ -72,23 +84,27 @@ describe('beginSubmission', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '你好',
images: [{ previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 }],
attachments: [{
type: 'image', value: { previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 },
}],
})
expect(session.getSnapshot().promptAttempted).toBe(true)
expect(session.getSnapshot().pendingSubmissions).toMatchObject([{
requestId: handle.requestId,
placement: 'transcript',
text: '你好',
images: [{ previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 }],
attachments: [{
type: 'image', value: { previewUrl: 'blob:p1', name: 'a.png', width: 4, height: 3 },
}],
}])
})
it('derives and captures the echo placement from running state and delivery mode', () => {
const { session } = makeSession()
session.beginSubmission({ mode: 'queue', text: '空闲', images: [] })
session.beginSubmission({ mode: 'queue', text: '空闲', attachments: [] })
session.handleRunning(true)
session.beginSubmission({ mode: 'queue', text: '排队', images: [] })
session.beginSubmission({ mode: 'steer', text: '纠偏', images: [] })
session.beginSubmission({ mode: 'queue', text: '排队', attachments: [] })
session.beginSubmission({ mode: 'steer', text: '纠偏', attachments: [] })
session.handleRunning(false)
expect(session.getSnapshot().pendingSubmissions.map(({ text, placement }) => ({ text, placement }))).toEqual([
{ text: '空闲', placement: 'transcript' },
@@ -103,7 +119,7 @@ describe('beginSubmission', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '放弃',
images: [],
attachments: [],
onRetire: retirement => retirements.push(retirement),
})
handle.abandon()
@@ -121,7 +137,7 @@ describe('prompt-coupled retirement', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '失败的',
images: [],
attachments: [],
onRetire: retirement => retirements.push(retirement),
})
const result = await session.prompt([{ type: 'text', text: '失败的' }], 'queue', undefined, handle.requestId)
@@ -133,7 +149,7 @@ describe('prompt-coupled retirement', () => {
it('sends the echo identity as the prompt requestId', async () => {
const { api, session } = makeSession()
const handle = session.beginSubmission({ mode: 'queue', text: '带 id', images: [] })
const handle = session.beginSubmission({ mode: 'queue', text: '带 id', attachments: [] })
await session.prompt([{ type: 'text', text: '带 id' }], 'queue', undefined, handle.requestId)
expect(api.callsOf('session.prompt')).toMatchObject([{ requestId: handle.requestId }])
})
@@ -141,7 +157,7 @@ describe('prompt-coupled retirement', () => {
it('an unidentified prompt failure leaves registered echoes alone', async () => {
const { api, session } = makeSession()
api.onPrompt = () => Promise.resolve(err(new RemoteError('session/agent-busy', '忙', { reason: 'busy' })))
session.beginSubmission({ mode: 'queue', text: '还在', images: [] })
session.beginSubmission({ mode: 'queue', text: '还在', attachments: [] })
await session.prompt([{ type: 'text', text: '另一个' }], 'queue')
expect(session.getSnapshot().pendingSubmissions).toHaveLength(1)
})
@@ -156,7 +172,7 @@ describe('observed retirement', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '发送',
images: [{ previewUrl: 'blob:p1' }],
attachments: [{ type: 'image', value: { previewUrl: 'blob:p1' } }],
onRetire: retirement => retirements.push(retirement),
})
const refs = [imageRef('att-1')]
@@ -176,7 +192,7 @@ describe('observed retirement', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '排队',
images: [{ previewUrl: 'blob:p1' }],
attachments: [{ type: 'image', value: { previewUrl: 'blob:p1' } }],
onRetire: retirement => retirements.push(retirement),
})
const refs = [imageRef('att-q')]
@@ -188,9 +204,31 @@ describe('observed retirement', () => {
expect(session.getSnapshot().queue).toMatchObject([{ rpcId: handle.requestId }])
})
it('retires a mixed echo with durable references in original selection order', async () => {
const { api, session } = makeSession()
api.onHistory = () => Promise.resolve(ok(historyValue([])))
await session.open()
const retirements: PendingSubmissionRetirement[] = []
const file = fileRef('file-1')
const handle = session.beginSubmission({
mode: 'queue',
text: 'mixed',
attachments: [
{ type: 'image', value: { previewUrl: 'blob:first' } },
{ type: 'file', value: file },
{ type: 'image', value: { previewUrl: 'blob:last' } },
],
onRetire: retirement => retirements.push(retirement),
})
const refs = [imageRef('image-1'), file, imageRef('image-2')]
await api.pushFollow(SID, { type: 'event', event: promptEvent(SessionSeq(0), handle.requestId, refs) as never })
await settleFrames()
expect(retirements).toEqual([{ reason: 'observed', attachments: refs }])
})
it('a full-window install (reconnect resync) retires echoes observed in the window', async () => {
const { api, session } = makeSession()
const handle = session.beginSubmission({ mode: 'queue', text: '重连', images: [] })
const handle = session.beginSubmission({ mode: 'queue', text: '重连', attachments: [] })
api.onHistory = () => Promise.resolve(ok(historyValue([promptEvent(SessionSeq(12), handle.requestId)])))
await session.open()
await settleFrames()
@@ -205,7 +243,7 @@ describe('observed retirement', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '先观察',
images: [],
attachments: [],
onRetire: retirement => retirements.push(retirement),
})
await api.pushFollow(SID, { type: 'event', event: promptEvent(SessionSeq(0), handle.requestId) as never })
@@ -222,7 +260,7 @@ describe('observed retirement', () => {
const handle = session.beginSubmission({
mode: 'queue',
text: '同一请求',
images: [],
attachments: [],
onRetire: retirement => retirements.push(retirement),
})
session.handleControlFrame({
@@ -245,7 +283,7 @@ describe('observed retirement', () => {
const { api, session } = makeSession()
api.onHistory = () => Promise.resolve(ok(historyValue([])))
await session.open()
const handle = session.beginSubmission({ mode: 'queue', text: '帧', images: [] })
const handle = session.beginSubmission({ mode: 'queue', text: '帧', attachments: [] })
await api.pushFollow(SID, { type: 'event', event: promptEvent(SessionSeq(0), handle.requestId) as never })
expect(session.getSnapshot().pendingSubmissions).toHaveLength(1)
expect(frames).toHaveLength(1)
@@ -263,13 +301,13 @@ describe('disposal', () => {
const observed = session.beginSubmission({
mode: 'queue',
text: '已观察',
images: [],
attachments: [],
onRetire: retirement => retirements.push({ text: '已观察', retirement }),
})
session.beginSubmission({
mode: 'queue',
text: '未settle',
images: [],
attachments: [],
onRetire: retirement => retirements.push({ text: '未settle', retirement }),
})
await api.pushFollow(SID, { type: 'event', event: promptEvent(SessionSeq(0), observed.requestId) as never })
@@ -1,6 +1,6 @@
/** Session creation and adoption rules for Agent preset identity. */
import { mkdtempSync, realpathSync } from 'node:fs'
import { mkdtempSync, realpathSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Context } from '@deepseek-ai/cordis'
@@ -10,9 +10,17 @@ import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-presets'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { describe, expect, it } from 'vitest'
import { afterEach, describe, expect, it } from 'vitest'
import { createSessionTestRemote } from './test-remote.ts'
/** Booted contexts and their temp roots, torn down after each test. */
const contexts: Context[] = []
const tempDirs: string[] = []
afterEach(async () => {
await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose()))
for (const dir of tempDirs.splice(0)) rmSync(dir, { recursive: true, force: true })
})
function stubAgent(session: Session): Agent {
return { id: session.id, session, status: 'idle' } as unknown as Agent
}
@@ -42,7 +50,9 @@ function roster(ids: readonly string[]): unknown {
async function harness(presets?: readonly string[]) {
const cwd = realpathSync(mkdtempSync(join(tmpdir(), 'dsh-session-preset-')))
tempDirs.push(cwd)
const ctx = new Context()
contexts.push(ctx)
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
if (presets !== undefined) {
@@ -18,7 +18,7 @@ import { AttachmentStore } from '@deepseek-ai/dsh-attachment'
import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-presets'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
@@ -154,14 +154,14 @@ describe('session.history projections block', () => {
const snapshot = await opening(remote(ctx), child.id)
expect(snapshot.header).toEqual({
version: 0,
version: SESSION_FORMAT_VERSION,
id: child.id,
createdAt: child.header.createdAt,
cwd: '/workspace',
parentSession: parent.id,
seedLength: inheritedEventCount,
isSeeded: true,
})
expect(snapshot.header).not.toHaveProperty('isSeeded')
expect(snapshot.header).not.toHaveProperty('seedLength')
})
it('tracks pending and used model selections across repeated request headers', async () => {
@@ -435,7 +435,7 @@ describe('session.list projections column', () => {
const coldId = SessionId('session-cold-listing')
const load = () => { throw new Error('list must not load event logs') }
ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
list: async () => [{ version: 0, id: coldId, createdAt: 5, isSeeded: false, cwd: '/tmp' }],
list: async () => [{ version: SESSION_FORMAT_VERSION, id: coldId, createdAt: 5, isSeeded: false, cwd: '/tmp' }],
inspect: load,
open: load,
}) as never)
@@ -524,7 +524,7 @@ describe('session.list projections column', () => {
const { ctx } = await harness(true)
const coldId = SessionId('session-cold-uncached')
ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
list: async () => [{ version: 0, id: coldId, createdAt: 5, isSeeded: false, cwd: '/tmp' }],
list: async () => [{ version: SESSION_FORMAT_VERSION, id: coldId, createdAt: 5, isSeeded: false, cwd: '/tmp' }],
}) as never)
const response = await remote(ctx).list(request({}))
if (!response.ok) throw new Error('unreachable')
@@ -8,7 +8,7 @@ import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionSeq } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionSeq } from '@deepseek-ai/dsh-session'
import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import {
@@ -29,7 +29,7 @@ function request(query: string): { query: string } {
function header(id: string, cwd: string | null = '/project'): SessionHeader {
return {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sid(id),
createdAt: 100,
isSeeded: false,
@@ -96,7 +96,7 @@ function installSearchQuery(
describe('session.search', () => {
it('rejects search when the query service is absent', async () => {
const ctx = await baseContext()
const list = new ApiSessionList(ctx, { coldBlankProbeMaxEvents: 16, coldBlankProbeMaxBytes: 1024 })
const list = new ApiSessionList(ctx)
await expect(list.search('query', new AbortController().signal)).rejects.toMatchObject({
code: 'gateway/internal',
@@ -1,7 +1,7 @@
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore, { SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type {} from '@deepseek-ai/dsh-skill'
import { describe, expect, it, vi } from 'vitest'
@@ -15,7 +15,7 @@ function observation(
const lease = (): SessionObservation => ({
source: 'live',
header: {
version: 0,
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
isSeeded: false,
@@ -15,7 +15,6 @@ const PARENT = 'fk-parent' as SessionId
afterEach(() => {
vi.unstubAllGlobals()
})
function makeSession(
api = new FakeApiClient(),
options: SessionOptions = {},
@@ -99,29 +98,6 @@ describe('Session open', () => {
expect(api.followStarts).toHaveLength(2)
})
it('lands a packed live record in openState=error instead of crashing the stream loop', async () => {
const { api, session } = makeSession()
api.onHistory = () => histResponse(plainTurn(SessionSeq(0), 0, 'a', 'b'))
await session.open()
expect(session.getSnapshot().openState).toBe('open')
// The live tail may carry only events; a packed record breaks that contract.
await api.pushFollow(SID, {
type: 'chunks',
event: {
type: 'chunkrow/text-chunks',
seq: 6,
time: 6,
data: { turn: 1, step: 1, index: 0, texts: ['a'], dt: [] },
},
} as never)
await vi.waitFor(() => { expect(session.getSnapshot().openState).toBe('error') })
expect(session.getSnapshot().openError).toMatchObject({
code: 'gateway/internal', message: 'session live stream emitted a packed history record',
})
})
it('lands a Gateway-marked stream failure in openState=error', async () => {
const { api, session } = makeSession()
api.onHistory = () => Promise.reject(new Error('socket died'))
@@ -433,6 +409,7 @@ describe('prompt and cancel errors', () => {
address: {
kind: 'subagent', parentSessionId: PARENT, childSessionId: SID, mode: 'continuable',
},
assistantStream: true,
maxMessages: 50,
},
])
@@ -501,6 +478,29 @@ describe('prompt and cancel errors', () => {
})
})
it('rejects staged files instead of dropping them from subagent continuations', async () => {
const api = new FakeApiClient()
const session = new Session(SID, fakeRemote(api), {
address: { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable' },
parentAvailable: true,
})
await session.open()
const prompted = await session.prompt([
{ type: 'file', receiptId: 'receipt' as never },
{ type: 'text', text: '继续' },
], 'queue')
expect(prompted).toMatchObject({
ok: false,
error: {
code: 'subagent/attachment-invalid',
details: { reason: 'SUBAGENT_FILE_UNSUPPORTED' },
},
})
expect(api.callsOf('subagents.prompt')).toEqual([])
})
it('sends a one-shot address to the Host under the continuable marker', async () => {
const api = new FakeApiClient()
api.onSubagentPrompt = () => Promise.resolve(err(new RemoteError(
@@ -527,6 +527,7 @@ describe('prompt and cancel errors', () => {
address: {
kind: 'subagent', parentSessionId: PARENT, childSessionId: SID, mode: 'one-shot',
},
assistantStream: true,
maxMessages: 50,
},
])
@@ -10,6 +10,9 @@ import { Context } from '@deepseek-ai/cordis'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { LlmAttemptId } from '@deepseek-ai/dsh-llm'
import { RemoteStreamCarrierError } from '@deepseek-ai/dsh-api-gateway/client'
import { SESSION_FORMAT_VERSION, SessionSeq } from '@deepseek-ai/dsh-session/types'
import { ClientSessions, SessionCreateError } from '../src/client/sessions/service.ts'
import { scopeOf } from '../src/client/scope.ts'
import type { SessionFollowFrame } from '../src/types.ts'
@@ -132,6 +135,282 @@ describe('search', () => {
})
describe('scope tree', () => {
it('publishes transient Assistant chunks and the named durable v2 settlement through one event source', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }])
b.svc.open(sid('s1'))
const binding = b.svc.binding(sid('s1'))
if (binding === undefined) throw new Error('expected Session binding')
await vi.waitFor(() => {
expect(binding.session.getSnapshot().openState).toBe('open')
})
const attemptId = LlmAttemptId('web-live-attempt')
const durableMessage = {
type: 'event' as const,
event: {
type: 'assistant/message', seq: 0, time: 2,
data: {
turn: 1,
step: 1,
message: {
role: 'assistant',
content: [{ type: 'text', text: 'live' }],
source: { kind: 'model', provider: 'p', model: 'm' },
id: 'message-1',
},
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: ['live'] }],
},
surfaceOp: 'append' as const,
},
}
const publications: string[][] = []
const dispose = binding.eventSource.subscribe(() => {
publications.push(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
})
await b.api.pushFollow(sid('s1'), {
type: 'assistant-stream',
frame: {
type: 'start', attemptId, revision: 1, startedAfterSeq: -1,
turn: 1, step: 1,
},
})
await b.api.pushFollow(sid('s1'), {
type: 'assistant-stream',
frame: {
type: 'chunk', attemptId, revision: 2, index: 0,
time: 1,
chunk: { type: 'text-delta', index: 0, text: 'live' },
},
})
await vi.waitFor(() => {
expect(binding.eventSource.getSnapshot().entries).toHaveLength(1)
})
await b.api.pushFollow(sid('s1'), durableMessage)
await Promise.resolve()
expect(binding.eventSource.getSnapshot().entries).toHaveLength(1)
await b.api.pushFollow(sid('s1'), {
type: 'assistant-stream',
frame: {
type: 'end', attemptId, revision: 3, index: 1,
outcome: { kind: 'committed', eventType: 'assistant/message', seq: 0 },
},
})
await vi.waitFor(() => {
expect(binding.eventSource.getSnapshot().entries).toHaveLength(1)
})
expect(publications).toEqual([
['assistant/live-chunk'],
['assistant/message'],
])
dispose()
})
it('replaces an active assistant baseline on reconnect without duplicate chunks', async () => {
const b = bench()
const attemptId = LlmAttemptId('reconnect-attempt')
let records: never[] = []
b.api.onHistory = () => Promise.resolve(ok({ records, hasMore: false }))
b.api.assistantStreamBaseline = {
revision: 2,
activeAttempt: {
attemptId, startedAfterSeq: -1, turn: 1, step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: ['a'] }],
},
}
await feedList(b, [{ id: 's1' }])
b.svc.open(sid('s1'))
const binding = b.svc.binding(sid('s1'))
if (binding === undefined) throw new Error('expected Session binding')
await vi.waitFor(() => {
expect(binding.eventSource.getSnapshot().entries).toHaveLength(1)
})
records = []
b.api.assistantStreamBaseline = {
revision: 3,
activeAttempt: {
attemptId, startedAfterSeq: -1, turn: 1, step: 1,
nextIndex: 2,
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [1], texts: ['a', 'b'] }],
},
}
b.api.failStreams(new RemoteStreamCarrierError('lost'))
await vi.waitFor(() => {
expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(2)
expect(binding.eventSource.getSnapshot().entries).toHaveLength(2)
})
expect(binding.eventSource.getSnapshot().entries.map(entry => (
entry.event.type === 'assistant/live-chunk' && entry.event.data.chunk.type === 'text-delta'
? entry.event.data.chunk.text
: undefined
))).toEqual(['a', 'b'])
})
it('stages a post-opening assistant settlement behind its exact active attempt', async () => {
const b = bench()
const attemptId = LlmAttemptId('reconnect-settlement-attempt')
const priorMessage = {
type: 'event' as const,
event: {
type: 'assistant/message', seq: 0, time: 30,
data: {
turn: 1,
step: 1,
message: {
role: 'assistant',
content: [{ type: 'text', text: 'retry ' }],
source: { kind: 'model', provider: 'p', model: 'm' },
id: 'prior-attempt-message',
},
stream: [{ type: 'text-chunks', time0: 10, index: 0, dt: [], texts: ['retry '] }],
},
surfaceOp: 'append' as const,
},
}
const currentMessage = {
type: 'event' as const,
event: {
type: 'assistant/message', seq: 1, time: 19,
data: {
turn: 1,
step: 1,
message: {
role: 'assistant',
content: [{ type: 'text', text: 'settled' }],
source: { kind: 'model', provider: 'p', model: 'm' },
id: 'current-attempt-message',
},
stream: [{ type: 'text-chunks', time0: 20, index: 0, dt: [], texts: ['settled'] }],
},
surfaceOp: 'append' as const,
},
}
b.api.onHistory = () => Promise.resolve(ok({
records: [priorMessage] as never[],
hasMore: false,
}))
b.api.assistantStreamBaseline = {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: SessionSeq(0),
turn: 1,
step: 1,
nextIndex: 1,
stream: currentMessage.event.data.stream,
},
}
await feedList(b, [{ id: 's1' }])
b.svc.open(sid('s1'))
const binding = b.svc.binding(sid('s1'))
if (binding === undefined) throw new Error('expected Session binding')
await vi.waitFor(() => {
expect(binding.session.getSnapshot().openState).toBe('open')
})
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['assistant/message', 'assistant/live-chunk'])
expect(binding.eventSource.getSnapshot().entries[0]?.event).toBe(priorMessage.event)
await b.api.pushFollow(sid('s1'), currentMessage)
await Promise.resolve()
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['assistant/message', 'assistant/live-chunk'])
await b.api.pushFollow(sid('s1'), {
type: 'assistant-stream',
frame: {
type: 'end', attemptId, revision: 3, index: 1,
outcome: { kind: 'committed', eventType: 'assistant/message', seq: 1 },
},
})
await vi.waitFor(() => {
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['assistant/message', 'assistant/message'])
})
expect(binding.eventSource.getSnapshot().change).toEqual({
kind: 'settle-assistant', attemptId: String(attemptId), entry: currentMessage,
})
})
it('replaces an invalid settlement with the authoritative post-end baseline', async () => {
const b = bench()
const attemptId = LlmAttemptId('reconnect-end-index-attempt')
const prior = {
type: 'event' as const,
event: { type: 'turn/start', seq: 0, time: 19, data: { turn: 1 } },
}
const message = {
type: 'event' as const,
event: {
type: 'assistant/message', seq: 1, time: 21,
data: {
turn: 1,
step: 1,
message: {
role: 'assistant',
content: [{ type: 'text', text: 'settled' }],
source: { kind: 'model', provider: 'p', model: 'm' },
id: 'current-attempt-message',
},
stream: [{ type: 'text-chunks', time0: 20, index: 0, dt: [], texts: ['settled'] }],
},
surfaceOp: 'append' as const,
},
}
let records = [prior] as never[]
b.api.onHistory = () => Promise.resolve(ok({
records,
hasMore: false,
}))
b.api.assistantStreamBaseline = {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: SessionSeq(0),
turn: 1,
step: 1,
nextIndex: 1,
stream: message.event.data.stream,
},
}
await feedList(b, [{ id: 's1' }])
b.svc.open(sid('s1'))
const binding = b.svc.binding(sid('s1'))
if (binding === undefined) throw new Error('expected Session binding')
await vi.waitFor(() => {
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['turn/start', 'assistant/live-chunk'])
})
const openingRevision = binding.eventSource.getSnapshot().revision
await b.api.pushFollow(sid('s1'), message)
await Promise.resolve()
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['turn/start', 'assistant/live-chunk'])
records = [prior, message] as never[]
b.api.assistantStreamBaseline = { revision: 3 }
await b.api.pushFollow(sid('s1'), {
type: 'assistant-stream',
frame: {
type: 'end', attemptId, revision: 3, index: 0,
outcome: { kind: 'committed', eventType: 'assistant/message', seq: 0 },
},
})
await vi.waitFor(() => {
expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(2)
expect(b.api.activeFollows(sid('s1'))).toBe(1)
expect(binding.eventSource.getSnapshot().revision).toBeGreaterThan(openingRevision)
expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type))
.toEqual(['turn/start', 'assistant/message'])
})
})
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'))
@@ -268,16 +547,18 @@ describe('Agent scope disposal lifecycle', () => {
value: {
type: 'snapshot',
header: {
version: 0,
version: SESSION_FORMAT_VERSION,
id: request.address.kind === 'session'
? request.address.sessionId
: request.address.childSessionId,
createdAt: 0,
isSeeded: false,
},
cursor: -1,
records: [],
hasMore: false,
projections: { asOfSeq: -1, values: {} },
assistantStream: { revision: 0 },
} as const,
})
}
@@ -342,11 +623,12 @@ describe('Agent scope disposal lifecycle', () => {
done: false,
value: {
type: 'snapshot',
header: { version: 0, id: sessionId, createdAt: 0 },
header: { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 0, isSeeded: false },
cursor: -1,
records: [],
hasMore: false,
projections: { asOfSeq: -1, values: {} },
assistantStream: { revision: 0 },
} as const,
})
}
@@ -3,6 +3,11 @@
import { SessionLogOffset } from '@deepseek-ai/dsh-session'
import type { Context } from '@deepseek-ai/cordis'
import type { ModelSelection as AgentModelSelection } from '@deepseek-ai/dsh-agent'
import type {
AdmittedPromptContentPart,
AttachmentAdmissionPart,
ImageAttachmentLimits,
} from '@deepseek-ai/dsh-attachment'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import {
SessionPersistenceNotFoundError,
@@ -83,8 +88,6 @@ export interface TestSessionRemote {
export interface TestSessionRemoteDefaults {
readonly defaultModelSelection: () => AgentModelSelection
readonly cwd: string
readonly coldBlankProbeMaxEvents?: number
readonly coldBlankProbeMaxBytes?: number
readonly nativeOpen?: boolean
readonly saveDefaultModelSelection?: (selection: AgentModelSelection) => void | Promise<void>
readonly openPath?: (path: string, signal: AbortSignal) => Promise<void>
@@ -93,6 +96,15 @@ export interface TestSessionRemoteDefaults {
const installed = new WeakMap<Context, SessionController>()
const TEST_IMAGE_LIMITS: ImageAttachmentLimits = Object.freeze({
maxImageBytes: 5 * 1024 * 1024,
maxImagesPerMessage: 20,
maxMessageImageBytes: 100 * 1024 * 1024,
maxImagePixels: 40_000_000,
maxImageDimension: 2000,
mediaTypes: Object.freeze(['image/png'] as const),
})
/** Compact header-and-events point read a persistence double declares per session. */
interface TestSessionInspection {
readonly meta: SessionHeader
@@ -129,7 +141,10 @@ function testReadHandle(
access: 'read',
read: (offset = 0, length?: number, options?: SessionHandleReadOptions) => {
options?.signal?.throwIfAborted()
return Promise.resolve(events.slice(offset, length === undefined ? undefined : offset + length))
return Promise.resolve({
eventState: 'detached',
events: structuredClone(events.slice(offset, length === undefined ? undefined : offset + length)),
} as const)
},
append: () => Promise.reject(new SessionReadOnlyError(sessionId, 'append')),
flush: () => Promise.reject(new SessionReadOnlyError(sessionId, 'flush')),
@@ -141,8 +156,7 @@ function testReadHandle(
/**
* Adapt a compact header/inspect persistence double onto the handle-based
* abstract the production readers consume: `list` snapshots wrap the double's
* headers, `stat` derives a metadata-less snapshot from the listing (so the
* cold-blank probe skips unless the double declares its own `stat`), and
* headers, `stat` derives a metadata-less snapshot from the listing, and
* `open` serves immutable read handles over the double's `inspect` result.
*/
export function testSessionPersistence(
@@ -236,6 +250,29 @@ function installControllers(
},
} as never)
}
if (ctx.get('attachments') === undefined) {
ctx.provide('attachments', {
imageLimits: TEST_IMAGE_LIMITS,
admitPromptContent: async (
content: readonly AttachmentAdmissionPart[],
): Promise<AdmittedPromptContentPart[]> => {
const admitted: AdmittedPromptContentPart[] = []
for (const part of content) {
if (part.type === 'image') throw new Error('test did not configure image persistence')
admitted.push(part)
}
return admitted
},
} as never)
}
if (ctx.get('fileUploads') === undefined) {
ctx.provide('fileUploads', {
registerAgentResolver: () => () => {},
resolve: () => undefined,
bindPrompt: () => ({ commit: () => {}, [Symbol.dispose]: () => {} }),
retirePrompt: () => {},
} as never)
}
installSessionReadTestServices(ctx)
const cwd = vi.spyOn(process, 'cwd').mockReturnValue(defaults.cwd)
let controller: SessionController
@@ -243,12 +280,6 @@ function installControllers(
controller = new SessionController(
ctx,
{
...defaults.coldBlankProbeMaxEvents === undefined
? {}
: { coldBlankProbeMaxEvents: defaults.coldBlankProbeMaxEvents },
...defaults.coldBlankProbeMaxBytes === undefined
? {}
: { coldBlankProbeMaxBytes: defaults.coldBlankProbeMaxBytes },
...defaults.nativeOpen === undefined ? {} : { nativeOpen: defaults.nativeOpen },
},
{
@@ -1,11 +1,12 @@
import { describe, expect, it, vi } from 'vitest'
import {
isRemoteFailure,
RemoteStream,
RemoteStreamCarrierError,
type RemoteStreamOptions,
} from '@deepseek-ai/dsh-api-gateway/client'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import { LlmAttemptId } from '@deepseek-ai/dsh-llm'
import { SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session/types'
import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
import {
createSessionControlStream,
@@ -16,6 +17,8 @@ import {
import type { SessionRemotes } from '../src/client/sessions/remotes.ts'
import type {
SessionAddress,
SessionAssistantStreamBaseline,
SessionAssistantStreamFrame,
SessionControlFrame,
SessionEventEntry,
SessionFollowFrame,
@@ -39,18 +42,6 @@ function entry(seq: number): SessionEventEntry {
return { type: 'event', event: { type: 'turn/start', seq, time: seq, data: { turn: seq } } }
}
function chunks(seq0: number): SessionHistoryRecord {
return {
type: 'chunks',
event: {
type: 'chunkrow/text-chunks',
seq: seq0,
time: seq0,
data: { turn: 1, step: 1, index: 0, texts: ['a', 'b', 'c'], dt: [1, 1] },
},
}
}
function page(records: readonly SessionHistoryRecord[], hasMore = false): SessionPage {
return { records, hasMore }
}
@@ -59,21 +50,28 @@ function snapshot(
cursor: number,
records: readonly SessionHistoryRecord[],
hasMore = false,
assistantStream: SessionAssistantStreamBaseline = { revision: 0 },
): SessionFollowFrame {
return {
type: 'snapshot',
header: {
version: 0,
version: SESSION_FORMAT_VERSION,
id: ADDRESS.kind === 'session' ? ADDRESS.sessionId : ADDRESS.childSessionId,
createdAt: 0,
isSeeded: false,
},
cursor,
records,
hasMore,
projections: { asOfSeq: cursor, values: {} },
assistantStream,
}
}
function assistantFrame(frame: SessionAssistantStreamFrame): SessionFollowFrame {
return { type: 'assistant-stream', frame }
}
function sessionClient(remote: SessionTransportRemote): SessionRemotes {
return {
session: remote as SessionRemote,
@@ -141,10 +139,209 @@ class ScriptedSessionRemote implements SessionTransportRemote {
}
describe('Session Client stream adapters', () => {
it('validates a packed logical range before publishing one compact Client entry', async () => {
const row = chunks(1)
it('opts into assistant notifications and publishes the reconnect baseline plus live frame', async () => {
const attemptId = LlmAttemptId('transport-attempt')
const baseline: SessionAssistantStreamBaseline = {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 0, index: 0, dt: [], texts: ['a'] }],
},
}
const frame: SessionAssistantStreamFrame = {
type: 'chunk', attemptId, revision: 3, index: 1,
time: 1, chunk: { type: 'text-delta', index: 0, text: 'b' },
}
const remote = new ScriptedSessionRemote(
[{ frames: [snapshot(4, [entry(0), row, entry(4)]), entry(5)], hold: true }],
[{ frames: [snapshot(0, [entry(0)], false, baseline), assistantFrame(frame)], hold: true }],
[],
)
const changes: SessionJournalChange[] = []
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: (change) => { changes.push(change) },
failed: vi.fn(),
})
await stream.open({})
await vi.waitFor(() => { expect(changes).toHaveLength(2) })
expect(remote.followRequests).toEqual([{ address: ADDRESS, assistantStream: true }])
expect(changes).toMatchObject([
{ type: 'replace', page: { assistantStream: baseline } },
{ type: 'assistant-stream', frame },
])
await stream.dispose()
})
it('rejects an opted-in opening that omits its Assistant baseline', async () => {
const remote = new ScriptedSessionRemote([{
frames: [{
type: 'snapshot',
header: {
version: SESSION_FORMAT_VERSION,
id: ADDRESS.sessionId,
createdAt: 0,
isSeeded: false,
},
cursor: -1,
records: [],
hasMore: false,
projections: { asOfSeq: -1, values: {} },
}],
}], [])
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: vi.fn(),
failed: vi.fn(),
})
try {
await expect(stream.open({})).rejects.toMatchObject({
code: 'gateway/internal',
message: 'session assistant stream omitted its opted-in opening baseline',
})
} finally {
await stream.dispose()
}
})
it('rejects an Assistant frame that arrives before the opening baseline', async () => {
const remote = new ScriptedSessionRemote([{
frames: [assistantFrame({
type: 'start', attemptId: LlmAttemptId('pre-opening-attempt'),
revision: 1, startedAfterSeq: -1, turn: 1, step: 1,
})],
}], [])
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: vi.fn(),
failed: vi.fn(),
})
try {
await expect(stream.open({})).rejects.toMatchObject({
code: 'gateway/internal',
message: 'session event stream emitted an entry before its opening cursor',
})
expect(remote.followRequests).toEqual([{ address: ADDRESS, assistantStream: true }])
} finally {
await stream.dispose()
}
})
it('rebaselines after a transient assistant revision gap without advancing the durable cursor', async () => {
const attemptId = LlmAttemptId('gapped-attempt')
const start: SessionAssistantStreamFrame = {
type: 'start', attemptId, revision: 1, startedAfterSeq: -1,
turn: 1, step: 1,
}
const gap: SessionAssistantStreamFrame = {
type: 'chunk', attemptId, revision: 3, index: 0,
time: 1, chunk: { type: 'text-delta', index: 0, text: 'lost predecessor' },
}
const replacement: SessionAssistantStreamBaseline = {
revision: 3,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: ['lost predecessor'] }],
},
}
const remote = new ScriptedSessionRemote([
{
frames: [snapshot(0, [entry(0)]), assistantFrame(start), assistantFrame(gap)],
},
{ frames: [snapshot(0, [entry(0)], false, replacement)], hold: true },
], [])
const changes: SessionJournalChange[] = []
const carrierFailed = vi.fn()
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: (change) => { changes.push(change) },
carrierFailed,
failed: vi.fn(),
})
await stream.open({})
await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
expect(changes.map(change => change.type)).toEqual([
'replace', 'assistant-stream', 'replace',
])
expect(changes.at(-1)).toMatchObject({
type: 'replace', page: { assistantStream: replacement },
})
expect(remote.pageRequests).toEqual([])
expect(carrierFailed).toHaveBeenCalledWith(expect.objectContaining({
message: 'session assistant stream skipped revision 2',
}))
await stream.dispose()
})
it('rebaselines when a replacement Agent lifecycle restarts at revision one', async () => {
const attemptId = LlmAttemptId('replacement-lifecycle-attempt')
const previous: SessionAssistantStreamBaseline = {
revision: 2,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 1,
step: 1,
nextIndex: 1,
stream: [{ type: 'text-chunks', time0: 1, index: 0, dt: [], texts: ['old'] }],
},
}
const replacementStart: SessionAssistantStreamFrame = {
type: 'start', attemptId, revision: 1, startedAfterSeq: -1,
turn: 2, step: 1,
}
const replacement: SessionAssistantStreamBaseline = {
revision: 1,
activeAttempt: {
attemptId,
startedAfterSeq: -1,
turn: 2,
step: 1,
nextIndex: 0,
stream: [],
},
}
const remote = new ScriptedSessionRemote([
{
frames: [snapshot(0, [entry(0)], false, previous), assistantFrame(replacementStart)],
},
{ frames: [snapshot(0, [entry(0)], false, replacement)], hold: true },
], [])
const changes: SessionJournalChange[] = []
const carrierFailed = vi.fn()
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: (change) => { changes.push(change) },
carrierFailed,
failed: vi.fn(),
})
try {
await stream.open({})
await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
expect(changes).toMatchObject([
{ type: 'replace', page: { assistantStream: previous } },
{ type: 'replace', page: { assistantStream: replacement } },
])
expect(carrierFailed).toHaveBeenCalledWith(expect.objectContaining({
message: 'session assistant stream skipped revision 3',
}))
} finally {
await stream.dispose()
}
})
it('validates one scalar current-event range before publishing Client entries', async () => {
const remote = new ScriptedSessionRemote(
[{ frames: [snapshot(2, [entry(0), entry(1), entry(2)]), entry(3)], hold: true }],
[],
)
const changes: SessionJournalChange[] = []
@@ -160,34 +357,11 @@ describe('Session Client stream adapters', () => {
type: 'replace',
entries: [
entry(0),
row,
entry(4),
entry(1),
entry(2),
],
})
expect(changes[0]?.type === 'replace' ? changes[0].entries[1] : undefined).toBe(row)
expect(changes[1]).toEqual({ type: 'append', entry: entry(5) })
await stream.dispose()
})
it('rejects a packed record emitted by the live follow path', async () => {
const failed = vi.fn()
const remote = new ScriptedSessionRemote(
[{ frames: [snapshot(-1, []), chunks(0) as SessionFollowFrame], hold: true }],
[],
)
const stream = new SessionEventStream(sessionClient(remote), ADDRESS, {
publish: vi.fn(),
failed,
})
await stream.open({})
await vi.waitFor(() => { expect(failed).toHaveBeenCalledOnce() })
const violation: unknown = failed.mock.calls[0]?.[0]
expect(isRemoteFailure(violation)).toBe(true)
expect(violation).toMatchObject({
code: 'gateway/internal',
message: 'session live stream emitted a packed history record',
})
expect(changes[1]).toEqual({ type: 'append', entry: entry(3) })
await stream.dispose()
})
@@ -215,7 +389,9 @@ describe('Session Client stream adapters', () => {
await vi.waitFor(() => { expect(changes).toHaveLength(2) })
await stream.prepend({ beforeSeq: 2, maxMessages: 50 })
expect(remote.followRequests).toEqual([{ address: ADDRESS, maxMessages: 50 }])
expect(remote.followRequests).toEqual([{
address: ADDRESS, assistantStream: true, maxMessages: 50,
}])
expect(remote.pageRequests).toEqual([
{ address: ADDRESS, throughSeq: 4, beforeSeq: 2, maxMessages: 50 },
])
@@ -252,8 +428,8 @@ describe('Session Client stream adapters', () => {
await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
expect(remote.followRequests).toEqual([
{ address: ADDRESS, maxMessages: 50 },
{ address: ADDRESS, maxMessages: 50 },
{ address: ADDRESS, assistantStream: true, maxMessages: 50 },
{ address: ADDRESS, assistantStream: true, maxMessages: 50 },
])
expect(remote.pageRequests).toEqual([])
expect(changes.map(change => change.type)).toEqual(['replace', 'append', 'replace'])
@@ -282,7 +458,10 @@ describe('Session Client stream adapters', () => {
await stream.open({})
finish.resolve(undefined)
await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
expect(remote.followRequests).toEqual([{ address: ADDRESS }, { address: ADDRESS }])
expect(remote.followRequests).toEqual([
{ address: ADDRESS, assistantStream: true },
{ address: ADDRESS, assistantStream: true },
])
expect(remote.pageRequests).toEqual([])
await stream.dispose()
})
@@ -1,6 +1,6 @@
import { Context } from '@deepseek-ai/cordis'
import { createScope } from '@deepseek-ai/dsh-scope'
import SessionStore, { SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader, SurfaceIntent } from '@deepseek-ai/dsh-session'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import { snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
@@ -170,8 +170,8 @@ describe('SessionHistoryController', () => {
it('subscribes before a cold read and ignores unrelated and replayed buffered events', async () => {
const { ctx, transport } = await setup()
const sessionId = SessionId('cold-race')
const header = {
version: 0,
const header: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -211,8 +211,8 @@ describe('SessionHistoryController', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const sessionId = SessionId('created-during-observation')
const header = {
version: 0,
const header: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -269,8 +269,8 @@ describe('SessionHistoryController', () => {
{ inject: ['sessions'] },
))
const sessionId = SessionId('cold-attach')
const header = {
version: 0,
const header: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -317,8 +317,8 @@ describe('SessionHistoryController', () => {
it('rejects gaps in replayed and live event sequences', async () => {
const replay = await setup()
const replayId = SessionId('replay-gap')
const replayHeader = {
version: 0,
const replayHeader: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: replayId,
createdAt: 1,
cwd: '/workspace',
@@ -366,8 +366,8 @@ describe('SessionHistoryController', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const sessionId = SessionId('projectionless-follow')
const meta = {
version: 0,
const meta: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: sessionId,
createdAt: 1,
cwd: '/workspace',
@@ -400,7 +400,7 @@ describe('SessionHistoryController', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const sessionId = SessionId('promotion-failure')
const meta = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' }
const meta = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace' }
const disposePromotion = vi.fn()
const promotion = {
source: 'prepared', header: meta, events: [], cursor: -1,
@@ -469,7 +469,7 @@ describe('SessionHistoryController', () => {
const { ctx, transport } = await setup()
const sessionId = SessionId('corrupt-cold')
const failure = new Error('cold log is corrupt')
const header = { version: 0, id: sessionId, createdAt: 1, isSeeded: false, cwd: '/workspace' }
const header: SessionHeader = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, isSeeded: false, cwd: '/workspace' }
ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
list: () => Promise.resolve([header]),
inspect: () => Promise.reject(failure),
@@ -507,7 +507,7 @@ describe('SessionHistoryController', () => {
const corruptId = SessionId('missing-through-seq')
cold(
corrupt.ctx,
{ version: 0, id: corruptId, createdAt: 1, cwd: '/workspace', isSeeded: false },
{ version: SESSION_FORMAT_VERSION, id: corruptId, createdAt: 1, cwd: '/workspace', isSeeded: false },
[event('fixture/start', SessionSeq(0)), event('fixture/gap', SessionSeq(2))],
)
await expect(corrupt.transport.page({
@@ -552,7 +552,7 @@ describe('SessionHistoryController', () => {
const first = await setup()
const sessionId = SessionId('incomplete')
const address = { kind: 'session' as const, sessionId }
const firstHeader = { version: 0, id: sessionId, createdAt: 1, isSeeded: false }
const firstHeader: SessionHeader = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, isSeeded: false }
first.ctx.provide('sessionPersistence', testSessionPersistence(first.ctx, {
list: () => Promise.resolve([firstHeader]),
inspect: () => Promise.resolve({
@@ -565,8 +565,8 @@ describe('SessionHistoryController', () => {
.rejects.toMatchObject({ code: 'session/not-found' })
const second = await setup()
const listed = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false }
const inspected = { version: 0, id: sessionId, createdAt: 1, isSeeded: false }
const listed: SessionHeader = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false }
const inspected: SessionHeader = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, isSeeded: false }
second.ctx.provide('sessionPersistence', testSessionPersistence(second.ctx, {
list: () => Promise.resolve([listed]),
inspect: () => Promise.resolve({
@@ -582,8 +582,8 @@ describe('SessionHistoryController', () => {
it('serves cold ordinary history and validates every durable subagent descriptor state', async () => {
const ordinaryBench = await setup()
const ordinaryId = SessionId('cold-ordinary')
const ordinaryHeader = {
version: 0,
const ordinaryHeader: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: ordinaryId,
createdAt: 1,
cwd: '/workspace',
@@ -599,8 +599,8 @@ describe('SessionHistoryController', () => {
const parentSessionId = SessionId('cold-parent')
const childSessionId = SessionId('cold-child')
const childHeader = {
version: 0,
const childHeader: SessionHeader = {
version: SESSION_FORMAT_VERSION,
id: childSessionId,
createdAt: 1,
cwd: '/workspace',
@@ -637,7 +637,7 @@ describe('SessionHistoryController', () => {
const parentSessionId = SessionId('missing-projection-parent')
const childSessionId = SessionId('missing-projection-child')
const meta: SessionHeader = {
version: 0,
version: SESSION_FORMAT_VERSION,
id: childSessionId,
createdAt: 1,
cwd: '/workspace',