Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge origin/master into web-permission-sandbox

Browse Source 
39 conflicts resolved per the reattachment plan (missions worktree-projbiz
0728-1859): baseline wins for deleted packages (host/runtime, old ui/acp,
ui-sidebar Rows/tree) and retired specs; unions for wire-layer exports and
client summary fields; the approval takeover, waitingApprovals tracking, and
PendingApproval domain face carry over onto the master structure. The two
new host specs follow the runtime->apiproxy rename. Dead PR-side wiring
(ConversationInjected permissions/setPermission spread, InputBar controls
prop, boot.ts sandbox composition) resolves to master and its replacement
lands in follow-up commits.
This commit is contained in:
imccyu
2026-07-28 21:20:00 +08:00
parent f0410d592d cad0615261
commit a66d1e335f
3427 changed files with 156166 additions and 59003 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
packages/client/runtime/tests/client-apply.spec.ts
View File

@@ -1,5 +1,5 @@
/**
* Runtime plugin browser-half apply: slots + sessions mounting over the
* Runtime plugin browser-half apply: slots + object services mounting over the
* connection handle, stream-loop sink wiring into the object layer, and the
* fiber-scoped loop teardown.
*/
@@ -8,7 +8,9 @@ import { describe, expect, it } from 'vitest'
import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client'
import type { ConnectionSinks } from '@deepseek-ai/dsh-client-connection/client'
import * as RuntimeClient from '../src/client/index.ts'
import { FakeApiClient } from './fake-api.ts'
import type { SessionsService } from '../src/client/sessions/service.ts'
import type { WorkspacesService } from '../src/client/workspaces/service.ts'
import { FakeApiClient, ok } from './fake-api.ts'
interface Bench {
ctx: Context
@@ -33,29 +35,73 @@ async function mount(): Promise<Bench> {
return bench
}
async function flushMicrotasks(): Promise<void> {
for (let i = 0; i < 12; i++) await Promise.resolve()
}
describe('runtime client apply', () => {
it('mounts ctx.slots + ctx.sessions and wires the stream sinks into the manager', async () => {
it('mounts slots, Sessions, and Workspaces and fans host frames into both managers', async () => {
const bench = await mount()
expect(bench.ctx.get('slots') !== undefined).toBe(true)
// The built-in 'root' declaration ships with this package's SlotsService
// (the SlotMap 'root' merge lives here since the slot-parity rework).
expect(bench.ctx.slots.spec('root')).toEqual({ kind: 'single', scope: 'root' })
const sessions = bench.ctx.get('sessions')
const workspaces = bench.ctx.get('workspaces')
expect(sessions !== undefined).toBe(true)
expect(workspaces !== undefined).toBe(true)
if (workspaces === undefined) throw new Error('WorkspacesService missing after runtime apply')
expect(bench.sinks).toBeDefined()
// Frame sinks reach the object layer: a host session-added lands in the list store.
bench.sinks?.onHostEnvelope?.({
rpcId: 'r1' as never,
payload: { type: 'host/session-added', sessionId: 's-new' } as never,
payload: { type: 'host/session-added', blank: true, sessionId: 's-new' } as never,
})
await Promise.resolve()
expect((sessions as { list: { getSnapshot(): { ids: string[] } } }).list.getSnapshot().ids).toContain('s-new')
bench.sinks?.onHostEnvelope?.({
rpcId: 'r-workspace' as never,
payload: {
type: 'host/workspace-changed',
workspace: {
workspaceId: 'w-new', path: '/w/new', title: 'new', sessionIds: [],
createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z',
},
} as never,
})
await Promise.resolve()
expect(workspaces.list.getSnapshot().items[0]?.workspaceId).toBe('w-new')
// Mux sink and onConnected route without throwing (manager semantics own the behavior).
bench.sinks?.onMuxEnvelope?.({ rpcId: 'r2' as never, payload: { type: 'stream/error', message: 'x' } as never })
bench.sinks?.onConnected?.()
})
it('selects the recent Workspace once when the first baselines have no current session', async () => {
const bench = await mount()
bench.api.onWorkspaceList = () => Promise.resolve(ok({
items: [{
workspaceId: 'w-recent', path: '/w/recent', title: 'recent', sessionIds: [],
createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z',
}] as never[],
}))
bench.api.onList = () => Promise.resolve(ok({ items: [] }))
bench.sinks?.onConnected?.()
await flushMicrotasks()
const sessions = bench.ctx.get('sessions') as SessionsService
const workspaces = bench.ctx.get('workspaces') as WorkspacesService
expect(bench.api.callsOf('session.create')).toEqual([{ workspaceId: 'w-recent' }])
expect(sessions.list.getSnapshot().current).toBe('fk-new')
sessions.clear()
await workspaces.refresh()
await flushMicrotasks()
expect(sessions.list.getSnapshot().current).toBeUndefined()
expect(bench.api.callsOf('session.create')).toHaveLength(1)
})
it('stops the stream loop when the plugin fiber unloads', async () => {
const bench = await mount()
const fiber = [...bench.ctx.registry.values()].find(f => f.name?.includes('client'))

45
packages/client/runtime/tests/event-script.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage, createMessage, createToolResultMessage, CallId } from '@deepseek-ai/dsh-llm'
// Minimal SessionEvent builders for orchestration tests (shape mirrors what the
// host emits; only the fields the object layer reads).
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
@@ -13,7 +14,9 @@ export const ev = {
turnStart: (seq: number, turn: number): SessionEvent =>
at(seq, { type: 'turn/start', data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } }),
user: (seq: number, body: string): SessionEvent =>
at(seq, { type: 'user/message', surfaceOp: 'append', data: { content: text(body), source: { kind: 'user' } } }),
at(seq, { type: 'user/message', surfaceOp: 'append', data: createUserMessage({
content: text(body), source: { kind: 'user' },
}) }),
stepStart: (seq: number, turn: number, step = 0): SessionEvent =>
at(seq, { type: 'step/start', data: { turn, step } }),
chunkStart: (seq: number, turn: number, step = 0, index = 0): SessionEvent =>
@@ -21,15 +24,51 @@ export const ev = {
chunkText: (seq: number, 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: number, turn: number, body: string, step = 0): SessionEvent =>
at(seq, { type: 'assistant/message', surfaceOp: 'append', data: { turn, step, content: text(body), provenance: { provider: 'fake', model: 'fk-1' } } }),
at(seq, { type: 'assistant/message', surfaceOp: 'append', data: {
turn, step,
message: createMessage({
role: 'assistant',
content: text(body),
source: {
kind: 'model',
...{ provider: 'fake', model: 'fk-1' },
},
}),
} }),
toolCall: (seq: number, turn: number, callId: string, name: string, args: string, step = 0): SessionEvent =>
at(seq, { type: 'tool/call', data: { turn, step, callId, name, arguments: args } }),
toolResult: (seq: number, turn: number, callId: string, body: string, step = 0): SessionEvent =>
at(seq, { type: 'tool/result', surfaceOp: 'append', data: { turn, step, callId, content: text(body), isError: false } }),
at(seq, {
type: 'tool/result',
surfaceOp: 'append',
data: {
turn,
step,
message: createToolResultMessage({
callId: CallId(callId),
content: text(body),
isError: false,
}),
},
}),
codeDispatchStart: (seq: number, parentCallId: string, n: number, name: string, args: unknown): SessionEvent =>
at(seq, {
type: 'tool/code-dispatch-start',
data: { parentCallId, subCallId: `${parentCallId}:code:${n}`, name, arguments: args },
}),
codeDispatch: (seq: number, parentCallId: string, n: number, name: string, args: unknown, body: string, isError = false): SessionEvent =>
at(seq, {
type: 'tool/code-dispatch',
data: { parentCallId, subCallId: `${parentCallId}:code:${n}`, name, arguments: args, isError, content: text(body) },
}),
stepEnd: (seq: number, turn: number, step = 0): SessionEvent =>
at(seq, { type: 'step/end', data: { turn, step } }),
turnEnd: (seq: number, turn: number, reason: 'completed' | 'cancelled' = 'completed'): SessionEvent =>
at(seq, { type: 'turn/end', data: { turn, reason: { kind: reason } } }),
commandRun: (seq: number, commandId: string, name: string, args = ''): SessionEvent =>
at(seq, { type: 'command/run', data: { commandId, name, args, source: { kind: 'user' } } }),
commandDone: (seq: number, commandId: string, kind: 'success' | 'error' = 'success', text?: string): SessionEvent =>
at(seq, { type: 'command/done', data: { commandId, kind, ...text === undefined ? {} : { text } } }),
}
/** One complete plain turn (turn/start → user → step → assistant → turn/end), 6 events from startSeq. */

81
packages/client/runtime/tests/fake-api.ts
View File

@@ -1,11 +1,27 @@
// Test-local programmable IApiClient fake (NOT the fixture: fixture is a demo
// data source on a real clock; behavior tests need per-case responses and
// deferred-controlled timing). Streams are hand pumps: pushMux/pushHost.
import type { CommandId } from '@deepseek-ai/dsh-commands/brand'
import type {
ClientResponse, HostFrame, IApiClient, MuxFrame, RpcError, RpcReceipt, RpcRequest, RpcResponse, SessionId,
ClientResponse, CommandDescriptor, HostFrame, IApiClient, ModelTarget, MuxFrame,
RpcError, RpcReceipt, RpcRequest, RpcResponse, SessionId, SessionModels, SkillEntry,
WorkspaceId, WorkspaceView,
} from '@deepseek-ai/dsh-client-connection/client'
import { RpcId } from '@deepseek-ai/dsh-client-connection/client'
/** Programmable-default workspace row (branded id, ISO-ish times). */
function fakeWorkspace(id: string, over: Partial<WorkspaceView> = {}): WorkspaceView {
return {
workspaceId: id as WorkspaceId,
path: '/f/ws',
title: 'ws',
sessionIds: [],
createdAt: '2026-01-01T00:00:00.000Z',
updatedAt: '2026-01-01T00:00:00.000Z',
...over,
}
}
export interface Deferred<T> {
promise: Promise<T>
resolve(value: T): void
@@ -46,10 +62,23 @@ export class FakeApiClient implements IApiClient {
// Programmable slots (defaults answer OK-empty); reassign per case.
onList: (payload: unknown) => Promise<RpcResponse<{ items: never[] }>> = () => Promise.resolve(ok({ items: [] }))
onCreate: (payload: unknown) => Promise<RpcResponse<{ sessionId: SessionId }>> = () => Promise.resolve(ok({ sessionId: 'fk-new' as SessionId }))
readonly defaultModel: ModelTarget = { provider: 'deepseek', model: 'deepseek-v4-flash' }
onHistory: (payload: { sessionId: SessionId; beforeSeq?: number; maxMessages?: number })
=> Promise<RpcResponse<{ events: never[]; hasMore: boolean }>> =
() => Promise.resolve(ok({ events: [], hasMore: false }))
onModels: (payload: unknown) => Promise<RpcResponse<SessionModels>> = () => Promise.resolve(ok({
current: this.defaultModel,
groups: [{
id: 'deepseek',
name: 'DeepSeek',
models: [{ id: 'deepseek-v4-flash', name: 'DeepSeek V4 Flash' }],
}],
failures: [],
}))
onSelectModel: (payload: { provider: string; model: string }) =>
Promise<RpcResponse<{ selected: ModelTarget }>> =
payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } }))
onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onPermissions: (payload: unknown) =>
@@ -61,6 +90,10 @@ export class FakeApiClient implements IApiClient {
onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
() => Promise.resolve(ok({ version: '0-fake', cwd: '/f', attachedSessions: 0 }))
onPickDirectory: (payload: unknown) => Promise<RpcResponse<{ path: string | null }>> =
() => Promise.resolve(ok({ path: null }))
onOpenPath: (payload: unknown) => Promise<RpcResponse<{ opened: true }>> =
() => Promise.resolve(ok({ opened: true as const }))
private readonly muxConns: StreamConn<MuxFrame>[] = []
private readonly hostConns: StreamConn<HostFrame>[] = []
@@ -73,6 +106,9 @@ export class FakeApiClient implements IApiClient {
create: (payload: unknown) => this.record('session.create', payload, this.onCreate(payload)),
history: (payload: { sessionId: SessionId; beforeSeq?: number; maxMessages?: number }) =>
this.record('session.history', payload, this.onHistory(payload)),
models: (payload: unknown) => this.record('session.models', payload, this.onModels(payload)),
selectModel: (payload: { provider: string; model: string }) =>
this.record('session.selectModel', payload, this.onSelectModel(payload)),
prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
permissions: (payload: unknown) => this.record('session.permissions', payload, this.onPermissions(payload)),
@@ -81,6 +117,49 @@ export class FakeApiClient implements IApiClient {
readonly host: IApiClient['host'] = {
describe: (payload: unknown) => this.record('host.describe', payload, this.onDescribe(payload)),
pickDirectory: (payload: unknown) => this.record('host.pickDirectory', payload, this.onPickDirectory(payload)),
openPath: (payload: unknown) => this.record('host.openPath', payload, this.onOpenPath(payload)),
}
onWorkspaceList: (payload: unknown) => Promise<RpcResponse<{ items: never[] }>> = () => Promise.resolve(ok({ items: [] }))
onWorkspaceCreate: (payload: unknown) => Promise<RpcResponse<{ workspace: WorkspaceView; created: boolean }>> =
() => Promise.resolve(ok({ workspace: fakeWorkspace('fk-ws'), created: true }))
onWorkspaceRename: (payload: unknown) => Promise<RpcResponse<{ workspace: WorkspaceView }>> =
() => Promise.resolve(ok({ workspace: fakeWorkspace('fk-ws') }))
onWorkspaceDelete: (payload: unknown) => Promise<RpcResponse<{ deleted: true }>> =
() => Promise.resolve(ok({ deleted: true }))
onWorkspaceInsertSessionBefore: (payload: unknown) => Promise<RpcResponse<{ workspace: WorkspaceView }>> =
() => Promise.resolve(ok({ workspace: fakeWorkspace('fk-ws') }))
readonly workspace: IApiClient['workspace'] = {
list: (payload: unknown) => this.record('workspace.list', payload, this.onWorkspaceList(payload)),
create: (payload: unknown) => this.record('workspace.create', payload, this.onWorkspaceCreate(payload)),
rename: (payload: unknown) => this.record('workspace.rename', payload, this.onWorkspaceRename(payload)),
delete: (payload: unknown) => this.record('workspace.delete', payload, this.onWorkspaceDelete(payload)),
insertSessionBefore: (payload: unknown) =>
this.record('workspace.insertSessionBefore', payload, this.onWorkspaceInsertSessionBefore(payload)),
}
// Payloads stay `unknown` (lint-lane note above); response rows are the real
// wire shapes so cases can program requires-bearing catalogs and dual-address
// skill lists without casts.
onCommandList: (payload: unknown) => Promise<RpcResponse<{ commands: CommandDescriptor[] }>>
= () => Promise.resolve(ok({ commands: [] }))
onCommandExecute: (payload: unknown) => Promise<RpcResponse<{ matched: boolean; commandId?: CommandId }>>
= () => Promise.resolve(ok({ matched: false }))
onSkillList: (payload: unknown) => Promise<RpcResponse<{ skills: SkillEntry[] }>>
= () => Promise.resolve(ok({ skills: [] }))
readonly commands: IApiClient['commands'] = {
list: (payload: unknown) => this.record('command.list', payload, this.onCommandList(payload)),
execute: (payload: unknown) => this.record('command.execute', payload, this.onCommandExecute(payload)),
}
readonly skills: IApiClient['skills'] = {
list: (payload: unknown) => this.record('skill.list', payload, this.onSkillList(payload)),
}
/** When true, streams never fire onOpen (misbehaving-carrier material for the handshake timeout guard). */

114
packages/client/runtime/tests/fold-adapter.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage, CallId, createMessage, createToolResultMessage } from '@deepseek-ai/dsh-llm'
/**
* FoldAdapter over the real core SurfaceManager: padding sentinels for paged
* windows, incremental append with node-cache identity, six-variant
@@ -39,8 +40,16 @@ describe('FoldAdapter', () => {
const events = [
ev.user(0, '用户'),
ev.assistant(1, 0, '助手'),
at(2, { type: 'steering/message', surfaceOp: 'append', data: { turn: 0, content: [{ type: 'text', text: '插话' }], source: { kind: 'user' } } }),
at(3, { type: 'context/message', surfaceOp: 'append', data: { content: [{ type: 'text', text: '上下文' }], source: { kind: 'plugin', plugin: 'p' } } }),
at(2, { type: 'steering/message', surfaceOp: 'append', data: {
turn: 0,
message: createUserMessage({
content: [{ type: 'text', text: '插话' }],
source: { kind: 'user' },
}),
} }),
at(3, { type: 'user/message', surfaceOp: 'append', data: createUserMessage({
content: [{ type: 'text', text: '上下文' }], source: { kind: 'plugin', plugin: 'p' },
}) }),
ev.toolCall(4, 0, 'c1', 'echo', '{"x":1}'),
ev.toolResult(5, 0, 'c1', '结果'),
]
@@ -76,7 +85,17 @@ describe('FoldAdapter', () => {
// An invalid surfaceOp on a surface-eligible event deterministically throws in the core fold.
const window = [
ev.user(10, '正常'),
at(11, { type: 'assistant/message', surfaceOp: 'bogus-op', data: { turn: 0, step: 0, content: [{ type: 'text', text: '坏 op' }], provenance: { provider: 'x', model: 'y' } } }),
at(11, { type: 'assistant/message', surfaceOp: 'bogus-op', data: {
turn: 0, step: 0,
message: createMessage({
role: 'assistant',
content: [{ type: 'text', text: '坏 op' }],
source: {
kind: 'model',
...{ provider: 'x', model: 'y' },
},
}),
} }),
]
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
try {
@@ -98,7 +117,15 @@ describe('FoldAdapter', () => {
it('materializes a tool-result error field when present', () => {
const adapter = new FoldAdapter()
adapter.reset([
at(0, { type: 'tool/result', surfaceOp: 'append', data: { turn: 0, step: 0, callId: 'c1', content: [], isError: true, error: { name: 'Boom', code: 'boom' } } }),
at(0, { type: 'tool/result', surfaceOp: 'append', data: {
turn: 0, step: 0,
message: createToolResultMessage({
callId: CallId('c1'),
content: [],
isError: true,
}),
error: { name: 'Boom', code: 'boom' },
} }),
], 0)
expect(adapter.nodes().nodes[0]).toMatchObject({ kind: 'tool-result', isError: true, error: { code: 'boom' } })
})
@@ -142,4 +169,83 @@ describe('FoldAdapter', () => {
const node = adapter.nodes().nodes[0]
expect(node).toMatchObject({ kind: 'tool-result', call: null, callView: null, resultView: { title: '孤儿' } })
})
describe('command lifecycle nodes', () => {
it('folds a run/done pair into one settled node merged into flow order by seq', () => {
const adapter = new FoldAdapter()
adapter.reset([
ev.user(0, '先说话'),
ev.commandRun(1, 'cmd-1', 'plan'),
ev.commandDone(2, 'cmd-1', 'success', '已进入 plan mode'),
ev.assistant(3, 0, '然后回答'),
], 0)
const { nodes } = adapter.nodes()
expect(nodes.map(n => [n.kind, n.seq])).toEqual([['user', 0], ['command', 1], ['assistant', 3]])
expect(nodes[1]).toMatchObject({
kind: 'command', commandId: 'cmd-1', name: 'plan', args: '',
outcome: { kind: 'success', text: '已进入 plan mode' },
})
})
it('renders a run with no done as still executing (outcome null)', () => {
const adapter = new FoldAdapter()
adapter.reset([ev.commandRun(0, 'cmd-2', 'goal', ' ship it')], 0)
expect(adapter.nodes().nodes[0]).toMatchObject({
kind: 'command', name: 'goal', args: ' ship it', outcome: null,
})
})
it('soft-falls a done-only window into a node built from the done (cross-window cut)', () => {
const adapter = new FoldAdapter()
adapter.reset([ev.commandDone(80, 'cmd-3', 'error', '失败了')], 80)
expect(adapter.nodes().nodes[0]).toMatchObject({
kind: 'command', seq: 80, commandId: 'cmd-3', name: null, args: null,
outcome: { kind: 'error', text: '失败了' },
})
})
it('settles a live-appended done in place, keeping the node at the run seq', () => {
const adapter = new FoldAdapter()
adapter.reset(plainTurn(0, 0, 'q', 'a'), 0)
adapter.append(ev.commandRun(6, 'cmd-4', 'clear'))
const running = adapter.nodes().nodes.find(n => n.kind === 'command')
expect(running).toMatchObject({ outcome: null })
adapter.append(ev.commandDone(7, 'cmd-4'))
const settled = adapter.nodes().nodes.find(n => n.kind === 'command')
expect(settled).toMatchObject({ seq: 6, outcome: { kind: 'success' } })
// Settlement replaced the node object rather than mutating the published one.
expect(settled).not.toBe(running)
})
it('tails command nodes whose seq is past every surface node', () => {
const adapter = new FoldAdapter()
adapter.reset([ev.user(0, '问'), ev.commandRun(1, 'cmd-tail', 'plan')], 0)
expect(adapter.nodes().nodes.map(n => n.kind)).toEqual(['user', 'command'])
})
it('command nodes survive the degraded linear-scan branch', () => {
const adapter = new FoldAdapter()
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
try {
adapter.reset([
ev.commandRun(0, 'cmd-5', 'plan'),
ev.commandDone(1, 'cmd-5'),
at(2, { type: 'assistant/message', surfaceOp: 'bogus-op', data: {
turn: 0,
step: 0,
message: createMessage({
role: 'assistant',
content: [{ type: 'text', text: '坏 op' }],
source: { kind: 'model', provider: 'x', model: 'y' },
}),
} }),
], 0)
const { nodes, degraded } = adapter.nodes()
expect(degraded).toBe(true)
expect(nodes.some(n => n.kind === 'command')).toBe(true)
} finally {
errorSpy.mockRestore()
}
})
})
})

6
packages/client/runtime/tests/lineage.spec.ts
View File

@@ -8,12 +8,12 @@ import type { SessionId, SessionSummary } from '@deepseek-ai/dsh-client-connecti
import { flattenLineage } from '../src/client/sessions/lineage.ts'
const s = (id: string, updatedAt: number, parent?: string): SessionSummary => ({
sessionId: id as SessionId, updatedAt, running: false,
sessionId: id as SessionId, updatedAt, running: false, blank: false,
...(parent !== undefined ? { parentSessionId: parent as SessionId } : {}),
})
describe('flattenLineage', () => {
it('sorts roots by updatedAt desc and expands children DFS with depth, children sorted too', () => {
it('keeps established root and sibling order while expanding children DFS with depth', () => {
const out = flattenLineage([
s('old-root', 10),
s('new-root', 30),
@@ -22,7 +22,7 @@ describe('flattenLineage', () => {
s('grandkid', 5, 'kid-new'),
])
expect(out.map(e => [e.sessionId, e.depth])).toEqual([
['new-root', 0], ['kid-new', 1], ['grandkid', 2], ['kid-old', 1], ['old-root', 0],
['old-root', 0], ['new-root', 0], ['kid-old', 1], ['kid-new', 1], ['grandkid', 2],
])
})

173
packages/client/runtime/tests/manager.spec.ts
View File

@@ -12,8 +12,10 @@ import { entries, plainTurn } from './event-script.ts'
const S1 = 'fk-m1' as SessionId
const S2 = 'fk-m2' as SessionId
function summary(sessionId: SessionId, over: Partial<{ updatedAt: number; running: boolean; parentSessionId: SessionId }> = {}) {
return { sessionId, updatedAt: 100, running: false, ...over }
type SummaryOver = Partial<{ updatedAt: number; running: boolean; blank: boolean; parentSessionId: SessionId }>
function summary(sessionId: SessionId, over: SummaryOver = {}) {
return { sessionId, updatedAt: 100, running: false, blank: false, ...over }
}
describe('instances', () => {
@@ -57,7 +59,7 @@ describe('instances', () => {
})
describe('list lifecycle', () => {
it('single-flights refreshList and lands items sorted through lineage flattening', async () => {
it('single-flights refreshList and preserves the Host baseline order', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => gate.promise
@@ -65,12 +67,33 @@ describe('list lifecycle', () => {
const first = manager.refreshList()
const second = manager.refreshList()
expect(manager.getListSnapshot().state).toBe('loading')
gate.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
gate.resolve(ok({ items: [summary(S2, { updatedAt: 200 }), summary(S1)] as never[] }))
await Promise.all([first, second])
expect(api.callsOf('session.list')).toHaveLength(1)
const snapshot = manager.getListSnapshot()
expect(snapshot.state).toBe('idle')
expect(snapshot.items.map(i => i.sessionId)).toEqual([S2, S1]) // updatedAt desc
expect(snapshot.items.map(i => i.sessionId)).toEqual([S2, S1])
})
it('replays incremental frames over hydration and never batch-reorders established ids', async () => {
const api = new FakeApiClient()
const first = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => first.promise
const manager = new SessionManager(api)
const hydration = manager.refreshList()
manager.handleHostEnvelope({
rpcId: 'during-first' as never,
payload: { type: 'host/session-added', blank: true, sessionId: S2 },
})
first.resolve(ok({ items: [summary(S1)] as never[] }))
await hydration
expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
api.onList = () => Promise.resolve(ok({
items: [summary(S1, { updatedAt: 900 }), summary(S2, { updatedAt: 800 })] as never[],
}))
await manager.refreshList()
expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
})
it('keeps the error in the list snapshot on failure', async () => {
@@ -79,6 +102,26 @@ describe('list lifecycle', () => {
const manager = new SessionManager(api)
await manager.refreshList()
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal' } })
// A failed pull does not step the arrival phase: still pending.
expect(manager.getListSnapshot().phase).toBe('pending')
})
it('phase steps pending → ready on the first successful pull and never returns', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
expect(manager.getListSnapshot().phase).toBe('pending')
await manager.refreshList()
expect(manager.getListSnapshot().phase).toBe('ready')
// Sticky across later failures: the pull-activity axis reports the error,
// the arrival phase holds.
api.onList = () => Promise.resolve(err({ code: 'internal', message: 'down', details: {} }))
await manager.refreshList()
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', phase: 'ready' })
// And across an empty re-pull (empty-with-ready = truly no sessions).
api.onList = () => Promise.resolve(ok({ items: [] as never[] }))
await manager.refreshList()
expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', phase: 'ready' })
expect(manager.getListSnapshot().items).toEqual([])
})
it('merges create into the list immediately without waiting for a refresh', async () => {
@@ -90,21 +133,18 @@ describe('list lifecycle', () => {
expect(manager.getListSnapshot().items.map(i => i.sessionId)).toEqual([S2])
})
it('retains monotonic title snapshots before list arrival, merges recency, and clears them on removal', async () => {
it('retains title projections before list arrival, keeps last-wins by seq, and clears them on removal', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleMuxEnvelope({
rpcId: 'title-new' as never,
payload: { type: 'session/title', sessionId: S1, title: 'Newest', eventSeq: 4, updatedAt: 300 },
})
manager.handleMuxEnvelope({
rpcId: 'title-stale' as never,
payload: { type: 'session/title', sessionId: S1, title: 'Stale', eventSeq: 3, updatedAt: 900 },
})
manager.handleMuxEnvelope({
rpcId: 'title-equal' as never,
payload: { type: 'session/title', sessionId: S1, title: 'Equal', eventSeq: 4, updatedAt: 901 },
})
const titleFrame = (rpcId: string, title: string, seq: number) => {
manager.handleMuxEnvelope({
rpcId: rpcId as never,
payload: { type: 'session/projection', sessionId: S1, key: 'title', value: title, seq } as never,
})
}
titleFrame('title-new', 'Newest', 4)
titleFrame('title-stale', 'Stale', 3)
titleFrame('title-equal', 'Equal', 4)
api.onList = () => Promise.resolve(ok({
items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[],
}))
@@ -112,42 +152,35 @@ describe('list lifecycle', () => {
const titled = manager.getListSnapshot()
expect(titled.items.map(item => item.sessionId)).toEqual([S1, S2])
expect(titled.items[0]).toMatchObject({ title: 'Newest', updatedAt: 300 })
expect(titled.items[0]?.title).toBe('Newest')
expect(titled.items[1]?.title).toBeUndefined()
manager.handleHostEnvelope({ rpcId: 'removed' as never, payload: { type: 'host/session-removed', sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'readded' as never, payload: { type: 'host/session-added', sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'readded' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined()
})
it('drops a retained title beyond the subscription baseline before accepting its durable replay', async () => {
it('drops a projection row beyond the subscription baseline before accepting its durable replay', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
const manager = new SessionManager(api)
await manager.refreshList()
manager.handleMuxEnvelope({
rpcId: 'title-unflushed' as never,
payload: { type: 'session/title', sessionId: S1, title: 'Unflushed', eventSeq: 4, updatedAt: 400 },
})
const frame = (rpcId: string, payload: object) => {
manager.handleMuxEnvelope({ rpcId: rpcId as never, payload: payload as never })
}
frame('title-unflushed', { type: 'session/projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
manager.handleMuxEnvelope({
rpcId: 'subscribed-recovered' as never,
payload: { type: 'session/subscribed', sessionId: S1, lastSeq: 2 },
})
// The durable baseline says the host only knows up to seq 2: the phantom
// row rode lost state and must drop, or last-wins pins it forever.
frame('subscribed-recovered', { type: 'session/subscribed', sessionId: S1, lastSeq: 2 })
expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
expect(manager.getListSnapshot().items[0]?.updatedAt).toBe(100)
manager.handleMuxEnvelope({
rpcId: 'title-durable' as never,
payload: { type: 'session/title', sessionId: S1, title: 'Durable', eventSeq: 2, updatedAt: 200 },
})
expect(manager.getListSnapshot().items[0]).toMatchObject({ title: 'Durable', updatedAt: 200 })
frame('title-durable', { type: 'session/projection', sessionId: S1, key: 'title', value: 'Durable', seq: 2 })
expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
manager.handleMuxEnvelope({
rpcId: 'subscribed-current' as never,
payload: { type: 'session/subscribed', sessionId: S1, lastSeq: 2 },
})
expect(manager.getListSnapshot().items[0]).toMatchObject({ title: 'Durable', updatedAt: 200 })
// A baseline at or past the row's seq keeps it (nothing phantom to drop).
frame('subscribed-current', { type: 'session/subscribed', sessionId: S1, lastSeq: 2 })
expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
})
})
@@ -155,8 +188,8 @@ describe('host frame routing', () => {
it('adds/removes/flips sessions from host frames and keeps removed instances resident', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-added', sessionId: S1 } }) // dup: ignored
manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } }) // dup: ignored
expect(manager.getListSnapshot().items).toHaveLength(1)
const session = manager.get(S1)
@@ -192,14 +225,14 @@ describe('remaining branches', () => {
expect(session.getSnapshot().running).toBe(true)
})
it('create passes cwd through, folds transport throws, and skips the merge when already listed', async () => {
it('create passes cwd and a preallocated id, folds transport throws, and deduplicates the echo', async () => {
const api = new FakeApiClient()
api.onCreate = () => Promise.resolve(ok({ sessionId: S1 }))
const manager = new SessionManager(api)
await manager.create('/tmp/w')
expect(api.callsOf('session.create')).toEqual([{ cwd: '/tmp/w' }])
await manager.create({ cwd: '/tmp/w', sessionId: S1 })
expect(api.callsOf('session.create')).toEqual([{ cwd: '/tmp/w', sessionId: S1 }])
expect(manager.getListSnapshot().items[0]).toMatchObject({ sessionId: S1, cwd: '/tmp/w' })
await manager.create('/tmp/w') // same id returned: no duplicate row
await manager.create({ cwd: '/tmp/w' }) // same id returned: no duplicate row
expect(manager.getListSnapshot().items).toHaveLength(1)
api.onCreate = () => Promise.reject(new Error('create wire down'))
expect(await manager.create()).toMatchObject({ ok: false, error: { code: 'internal' } })
@@ -208,6 +241,42 @@ describe('remaining branches', () => {
expect(await manager.create()).toMatchObject({ ok: false })
})
it('publishes a real Ungrouped summary from workspace-attach-failed', async () => {
const api = new FakeApiClient()
api.onCreate = () => Promise.resolve(err({
code: 'workspace-attach-failed',
message: 'published but unattached',
details: { sessionId: S1, workspaceId: 'w1' },
} as never))
const manager = new SessionManager(api)
const result = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({ sessionId: S1 })])
expect(manager.getListSnapshot().items[0]).not.toHaveProperty('cwd')
})
it('reconciles a preallocated id after an ordinary transport failure', async () => {
const api = new FakeApiClient()
api.onCreate = () => Promise.reject(new Error('response lost'))
const manager = new SessionManager(api)
const failed = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
expect(failed).toMatchObject({ ok: false, error: { message: 'response lost' } })
expect(manager.getListSnapshot().items).toEqual([])
manager.handleHostEnvelope({
rpcId: 'published-later' as never,
payload: { type: 'host/session-added', blank: true, sessionId: S1, cwd: '/w/one' },
})
expect(manager.getListSnapshot().items).toEqual([
expect.objectContaining({ sessionId: S1, cwd: '/w/one' }),
])
manager.handleHostEnvelope({
rpcId: 'duplicate-frame' as never,
payload: { type: 'host/session-added', blank: true, sessionId: S1, cwd: '/w/one' },
})
expect(manager.getListSnapshot().items).toHaveLength(1)
})
it('subscribe notifies on list changes and stops after unsubscribe', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
@@ -218,7 +287,7 @@ describe('remaining branches', () => {
expect(notified).toBeGreaterThan(0)
const seen = notified
unsubscribe()
manager.handleHostEnvelope({ rpcId: 'h' as never, payload: { type: 'host/session-added', sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
await new Promise(resolve => setTimeout(resolve, 0))
expect(notified).toBe(seen)
})
@@ -257,8 +326,8 @@ describe('remaining branches', () => {
it('carries parentSessionId from host/session-added into the lineage row', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-added', sessionId: S2, parentSessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-added', blank: true, sessionId: S2, parentSessionId: S1 } })
const items = manager.getListSnapshot().items
expect(items.find(e => e.sessionId === S2)).toMatchObject({ parentSessionId: S1, depth: 1 })
})
@@ -267,7 +336,11 @@ describe('remaining branches', () => {
describe('connected generation', () => {
it('refreshes the list and resyncs only opened instances', async () => {
const api = new FakeApiClient()
api.onHistory = () => Promise.resolve(ok({ events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
api.onHistory = () => Promise.resolve(ok({
events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'deepseek-chat' },
}))
const manager = new SessionManager(api)
const openedSession = manager.get(S1)
await openedSession.open()

187
packages/client/runtime/tests/projection-store.spec.ts Normal file
View File

@@ -0,0 +1,187 @@
/**
* Projection value store (session-projection RFC, push model): the single
* higher-seq-wins rule on both paths (a stale baseline cannot overwrite a
* newer push frame; a replayed frame cannot regress), capability absence as
* undefined, generation truncation, and the Session/manager wiring (tail-page
* seeding, session/projection frame routing pre- and post-instantiation, the
* list rows' title projection).
*/
import { describe, expect, it } from 'vitest'
import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { ProjectionValueStore } from '../src/client/sessions/projection-store.ts'
import { Session } from '../src/client/sessions/session.ts'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient, ok } from './fake-api.ts'
import { entries, plainTurn } from './event-script.ts'
// Test-domain keys merged into the projection map (the interface package's
// pure-type outlet), the same way domain host plugins merge theirs.
declare module '@deepseek-ai/dsh-session-projection/types' {
interface SessionProjectionMap {
'test/marks': { marks: string[] }
}
}
const SID = 'fk-s1' as SessionId
describe('ProjectionValueStore semantics', () => {
it('reads undefined until a value lands (capability absence)', () => {
const store = new ProjectionValueStore()
expect(store.get('test/marks')).toBeUndefined()
expect(store.faceOf('test/marks').getSnapshot()).toBeUndefined()
})
it('applies frames last-wins by seq: replayed and stale frames drop', () => {
const store = new ProjectionValueStore()
store.apply('test/marks', { marks: ['a'] }, 5)
store.apply('test/marks', { marks: ['a', 'b'] }, 9)
expect(store.get('test/marks')).toEqual({ marks: ['a', 'b'] })
store.apply('test/marks', { marks: ['stale'] }, 5)
store.apply('test/marks', { marks: ['equal'] }, 9)
expect(store.get('test/marks')).toEqual({ marks: ['a', 'b'] })
})
it('a stale baseline can neither overwrite nor clear a newer frame; a fresh one reseeds and clears', () => {
const store = new ProjectionValueStore()
store.apply('test/marks', { marks: ['frame-20'] }, 20)
// Stale cut: carried key loses to the newer frame; omitted key survives.
store.seed({ asOfSeq: 10, values: { 'test/marks': { marks: ['baseline-10'] } } })
expect(store.get('test/marks')).toEqual({ marks: ['frame-20'] })
store.seed({ asOfSeq: 15, values: {} })
expect(store.get('test/marks')).toEqual({ marks: ['frame-20'] })
// Fresh cut: carried key reseeds…
store.seed({ asOfSeq: 30, values: { 'test/marks': { marks: ['baseline-30'] } } })
expect(store.get('test/marks')).toEqual({ marks: ['baseline-30'] })
// …and an omitting fresh cut clears (capability absent as of the cut).
store.seed({ asOfSeq: 40, values: {} })
expect(store.get('test/marks')).toBeUndefined()
})
it('truncate drops rows past the durable baseline and keeps the rest', () => {
const store = new ProjectionValueStore()
store.apply('test/marks', { marks: ['durable'] }, 5)
store.apply('other', 'phantom', 50)
store.truncate(10)
expect(store.get('test/marks')).toEqual({ marks: ['durable'] })
expect(store.get('other')).toBeUndefined()
})
it('notifies the key face on change (batched) and not on dropped applications', async () => {
const store = new ProjectionValueStore()
let keyTicks = 0
let anyTicks = 0
store.faceOf('test/marks').subscribe(() => { keyTicks += 1 })
store.subscribeAny(() => { anyTicks += 1 })
store.apply('test/marks', { marks: ['a'] }, 5)
await Promise.resolve()
expect(keyTicks).toBe(1)
expect(anyTicks).toBe(1)
store.apply('test/marks', { marks: ['replay'] }, 3)
await Promise.resolve()
expect(keyTicks).toBe(1)
expect(anyTicks).toBe(1)
})
it('faces are identity-stable per key (the React binding cache premise)', () => {
const store = new ProjectionValueStore()
expect(store.faceOf('test/marks')).toBe(store.faceOf('test/marks'))
})
})
describe('Session tail-page seeding', () => {
it('seeds the store from a history response carrying a projections block', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api)
api.onHistory = () => Promise.resolve(ok({
events: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false,
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['from-baseline'] } } },
} as never))
await session.open()
expect(session.projections.get('test/marks')).toEqual({ marks: ['from-baseline'] })
})
it('a resync serving a stale block keeps the newer pushed value (seq rule end to end)', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api)
api.onHistory = () => Promise.resolve(ok({
events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['baseline'] } } },
} as never))
await session.open()
session.projections.apply('test/marks', { marks: ['pushed-9'] }, 9)
await session.resync()
expect(session.projections.get('test/marks')).toEqual({ marks: ['pushed-9'] })
})
it('treats a blockless response as no reset: pushed values survive', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api)
api.onHistory = () => Promise.resolve(ok({ events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
await session.open()
session.projections.apply('test/marks', { marks: ['pushed'] }, 9)
await session.resync()
expect(session.projections.get('test/marks')).toEqual({ marks: ['pushed'] })
})
})
describe('manager frame routing', () => {
const sid = (s: string): SessionId => s as SessionId
it('lands session/projection frames before instantiation and the Session adopts the same store', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleMuxEnvelope({
rpcId: 'p1' as never,
payload: { type: 'session/projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['early'] }, seq: 7 } as never,
})
const session = manager.get(sid('s1'))
expect(session.projections.get('test/marks')).toEqual({ marks: ['early'] })
// Frames after instantiation land in the same store.
manager.handleMuxEnvelope({
rpcId: 'p2' as never,
payload: { type: 'session/projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['later'] }, seq: 9 } as never,
})
expect(session.projections.get('test/marks')).toEqual({ marks: ['later'] })
})
it('projects the title key into list rows and truncates phantom rows on the subscribed baseline', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
}) as never)
await manager.refreshList()
manager.handleMuxEnvelope({
rpcId: 't1' as never,
payload: { type: 'session/projection', sessionId: sid('s1'), key: 'title', value: 'Projected title', seq: 4 } as never,
})
await Promise.resolve()
expect(manager.getListSnapshot().items[0]?.title).toBe('Projected title')
// The durable baseline says the host only knows up to seq 2: the row rode
// lost state and must drop (the un-flushed title precedent).
manager.handleMuxEnvelope({
rpcId: 'sub' as never,
payload: { type: 'session/subscribed', sessionId: sid('s1'), lastSeq: 2 } as never,
})
await Promise.resolve()
expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
})
it('drops the projection store with the removed session', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
}) as never)
await manager.refreshList()
manager.handleMuxEnvelope({
rpcId: 't1' as never,
payload: { type: 'session/projection', sessionId: sid('s1'), key: 'title', value: 'Doomed', seq: 4 } as never,
})
manager.handleHostEnvelope({
rpcId: 'rm' as never,
payload: { type: 'host/session-removed', sessionId: sid('s1') } as never,
})
expect(manager.get(sid('s1')).projections.get('title')).toBeUndefined()
})
})

234
packages/client/runtime/tests/queue-store.spec.ts Normal file
View File

@@ -0,0 +1,234 @@
/**
* Queue mirror semantics (web input-triggers queue cut 1): session/queued
* intake, host-rule retirement (message turn/start claims oldest non-steering;
* steering/message drains by source), leave-running sweep, reconnect reset,
* pre-instantiation buffering, and snapshot reference stability.
*/
import { describe, expect, it } from 'vitest'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { MuxFrame, RpcId, SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { Session } from '../src/client/sessions/session.ts'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient } from './fake-api.ts'
import { ev } from './event-script.ts'
const SID = 'fk-q1' as SessionId
const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }]
const rid = (id: string): RpcId => id as RpcId
/** session/queued frame with the wire-sourced rpcId key (the host prompt path). */
function queuedFrame(body: string, rpcId: string, steering = false): MuxFrame {
return {
type: 'session/queued',
sessionId: SID,
message: createUserMessage({
content: text(body),
source: { kind: 'user', rpcId: rid(rpcId) } as never,
}),
steering,
}
}
function makeSession(): Session {
return new Session(SID, new FakeApiClient())
}
describe('queue intake', () => {
it('lands a queued frame as a row keyed by the source rpcId with a flat preview', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-1'), queuedFrame('第一条 排队\n消息', 'p-1'))
const queue = session.getSnapshot().queue
expect(queue).toEqual([{ key: 'p-1', preview: '第一条 排队 消息' }])
})
it('falls back to the envelope rpcId when the source carries none, and tags non-text blocks', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-2'), {
type: 'session/queued',
sessionId: SID,
message: createUserMessage({
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
source: { kind: 'plugin', plugin: 'loop' },
}),
steering: false,
})
expect(session.getSnapshot().queue).toEqual([{ key: 'f:env-2', preview: 'hi [image]' }])
})
it('caps the preview at 200 code points with an ellipsis', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-3'), queuedFrame('长'.repeat(201), 'p-cap'))
const preview = session.getSnapshot().queue[0]?.preview ?? ''
expect(Array.from(preview)).toHaveLength(201) // 200 + …
expect(preview.endsWith('…')).toBe(true)
})
it('keeps the queue array reference stable across unrelated snapshot swaps', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-4'), queuedFrame('稳定', 'p-s'))
const before = session.getSnapshot().queue
session.handleAgentError('unrelated') // dirties the snapshot without touching the queue
expect(session.getSnapshot().queue).toBe(before)
})
})
describe('queue retirement (host queuedMirror rules)', () => {
it('a message-triggered turn/start claims the oldest non-steering row', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('先', 'p-1'))
session.handleMuxEnvelope(rid('e2'), queuedFrame('后', 'p-2'))
session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: ev.turnStart(0, 0) })
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-2'])
})
it('an injection-triggered turn/start claims nothing', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('留', 'p-1'))
const injection = {
...ev.turnStart(0, 0),
data: { turn: 0, trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'x' } } },
} as never
session.handleMuxEnvelope(rid('e2'), { type: 'session/event', sessionId: SID, event: injection })
expect(session.getSnapshot().queue).toHaveLength(1)
})
it('steering/message drains the source-matched steering row only', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('普通', 'p-1')) // idle → non-steering
session.handleMuxEnvelope(rid('e3'), queuedFrame('插话', 'p-2', true))
// Loop-authored steering (different source) must not consume the user entry.
const foreignSteering = {
seq: 0, time: 1,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 0,
message: createUserMessage({
content: text('loop'),
source: { kind: 'plugin', plugin: 'loop' },
}),
},
} as never
session.handleMuxEnvelope(rid('e4'), { type: 'session/event', sessionId: SID, event: foreignSteering })
expect(session.getSnapshot().queue).toHaveLength(2)
const matchedSteering = {
seq: 1, time: 2,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 0,
message: createUserMessage({
content: text('插话'),
source: { kind: 'user', rpcId: rid('p-2') },
}),
},
} as never
session.handleMuxEnvelope(rid('e5'), { type: 'session/event', sessionId: SID, event: matchedSteering })
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-1'])
})
it('a leave-running flip sweeps the whole mirror (cancel/terminal-drop cover)', () => {
const session = makeSession()
session.handleRunning(true)
session.handleMuxEnvelope(rid('e1'), queuedFrame('一', 'p-1'))
session.handleMuxEnvelope(rid('e2'), queuedFrame('二', 'p-2'))
session.handleRunning(false)
expect(session.getSnapshot().queue).toEqual([])
})
it('a stale not-running relay on an idle session still sweeps replayed rows', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('孤儿', 'p-1'))
session.handleRunning(false) // running already false: equality path must not skip the sweep
expect(session.getSnapshot().queue).toEqual([])
})
})
describe('queue reconnect semantics', () => {
it('session/subscribed re-baselines the mirror: stale rows drop, the following snapshot lands fresh', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('旧连接', 'p-old'))
// New mux generation: subscribed arrives first on the same stream...
session.handleMuxEnvelope(rid('e2'), { type: 'session/subscribed', sessionId: SID, lastSeq: 10 })
expect(session.getSnapshot().queue).toEqual([])
// ...then the queue snapshot replays the live inbox.
session.handleMuxEnvelope(rid('e3'), queuedFrame('新基线', 'p-new'))
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-new'])
})
it('resync must NOT clear the mirror (regression: onConnected races the mux baseline)', async () => {
const session = makeSession()
// Reconnect ordering that broke: mux opened first and already delivered
// the fresh generation's baseline; host stream (and with it onConnected →
// resync) lands after. The host never resends — clearing here left the
// dock empty until the next enqueue.
session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
session.handleMuxEnvelope(rid('e2'), queuedFrame('新基线', 'p-fresh'))
await session.resync()
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-fresh'])
})
it('replayed steering retires without a replayed turn/start', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
session.handleMuxEnvelope(rid('e2'), queuedFrame('重连插话', 'p-steer', true))
const committed = {
seq: 6, time: 2,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 1,
message: createUserMessage({
content: text('重连插话'),
source: { kind: 'user', rpcId: rid('p-steer') },
}),
},
} as never
session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: committed })
expect(session.getSnapshot().queue).toEqual([])
})
})
describe('manager buffering of queued frames', () => {
it('buffers session/queued for uninstantiated sessions and replays before the running sync', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleMuxEnvelope({ rpcId: rid('b1'), payload: queuedFrame('预热', 'p-b1') })
// Instantiation replays the buffer; no summary exists, so no running sweep runs.
const session = manager.get(SID)
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-b1'])
// The buffer is consumed: a second get must not double-replay.
expect(manager.get(SID).getSnapshot().queue).toHaveLength(1)
})
it('a not-running list summary sweeps replayed rows at instantiation', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok([{ sessionId: SID, updatedAt: 1, running: false }]))
const manager = new SessionManager(api)
await manager.refreshList()
manager.handleMuxEnvelope({ rpcId: rid('b2'), payload: queuedFrame('该扫掉', 'p-b2') })
expect(manager.get(SID).getSnapshot().queue).toEqual([])
})
it('subscribed re-baselines the uninstantiated buffer: stale queued frames drop, non-queue frames survive (regression: reconnect duplication)', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
// Generation 1 baseline lands while the session is uninstantiated, along
// with a pending approval (never re-derivable from history).
manager.handleMuxEnvelope({ rpcId: rid('g1a'), payload: queuedFrame('第一代', 'p-g1') })
manager.handleMuxEnvelope({
rpcId: rid('g1b'),
payload: { type: 'approval/requested', sessionId: SID, approvalId: 'ap-1' as never, toolName: 'bash' },
})
// Reconnect: generation 2 replays subscribed + the SAME live queue entry.
manager.handleMuxEnvelope({ rpcId: rid('g2a'), payload: { type: 'session/subscribed', sessionId: SID, lastSeq: 3 } })
manager.handleMuxEnvelope({ rpcId: rid('g2b'), payload: queuedFrame('第一代', 'p-g1') })
const snapshot = manager.get(SID).getSnapshot()
// One queue row (no duplicate batch); the approval survived the re-baseline.
expect(snapshot.queue.map(r => r.key)).toEqual(['p-g1'])
expect(snapshot.pending.map(p => p.kind)).toEqual(['approval'])
})
})
/** ok wrapper with a typed items payload (the shared helper pins value to never[]). */
function ok(items: { sessionId: SessionId; updatedAt: number; running: boolean }[]) {
return { rpcId: rid(`ok-${items.length}`), result: { ok: true as const, value: { items: items as never[] } } }
}

84
packages/client/runtime/tests/scope.spec.ts Normal file
View File

@@ -0,0 +1,84 @@
/**
* Agent-scope primitive spec: the actx minted by createScope carries the
* tag and the dispatch filter itself, so plain cordis dispatch with the actx
* as subject routes by agent — same-agent tagged listeners receive,
* foreign-agent ones are filtered out, untagged listeners hear everything,
* and a subject-less root dispatch stays unfiltered. Scope-owned listeners
* dispose with the fiber.
*/
import { Context } from 'cordis'
import { describe, expect, it } from 'vitest'
import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { createScope, scopeOf } from '../src/client/agents/scope.ts'
const sid = (k: string): SessionId => k as SessionId
declare module 'cordis' {
interface Events {
/**
* Test-only routed probe event.
* @param payload - marker payload.
* @mode bail
*/
'test/scope-probe'(payload: { from: string }): true | undefined
}
}
function bench() {
const root = new Context()
const a = createScope(root, sid('a'))
const b = createScope(root, sid('b'))
const seen: string[] = []
const listen = (label: string, ctx: Context, answer?: true) => {
ctx.on('test/scope-probe', (payload) => {
seen.push(`${label}:${payload.from}`)
return answer
})
}
return { root, a, b, seen, listen }
}
describe('createScope', () => {
it('tags the ctx (scopeOf) and leaves the root untagged', () => {
const { root, a } = bench()
expect(scopeOf(a.ctx)).toBe(sid('a'))
expect(scopeOf(root)).toBeUndefined()
})
it('scoped dispatch reaches same-session and untagged listeners, never a foreign session', () => {
const { root, a, b, seen, listen } = bench()
listen('a', a.ctx)
listen('b', b.ctx)
listen('root', root)
a.ctx.bail(a.ctx, 'test/scope-probe', { from: 'a' })
expect(seen).toEqual(['a:a', 'root:a'])
seen.length = 0
b.ctx.emit(b.ctx, 'test/scope-probe', { from: 'b' })
expect(seen).toEqual(['b:b', 'root:b'])
})
it('bail answers the first same-scope listener and skips filtered foreign ones', () => {
const { a, b, listen } = bench()
listen('b', b.ctx, true) // registered first, but foreign → filtered out
expect(a.ctx.bail(a.ctx, 'test/scope-probe', { from: 'a' })).toBeUndefined()
listen('a', a.ctx, true)
expect(a.ctx.bail(a.ctx, 'test/scope-probe', { from: 'a' })).toBe(true)
})
it('a subject-less root dispatch is unfiltered (every listener hears it)', () => {
const { root, a, b, seen, listen } = bench()
listen('a', a.ctx)
listen('b', b.ctx)
listen('root', root)
root.emit('test/scope-probe', { from: 'root' })
expect(seen).toEqual(['a:root', 'b:root', 'root:root'])
})
it('fiber disposal removes scope-owned listeners', async () => {
const { a, seen, listen } = bench()
listen('a', a.ctx)
await a.fiber.dispose()
a.ctx.emit(a.ctx, 'test/scope-probe', { from: 'late' })
expect(seen).toEqual([])
})
})

163
packages/client/runtime/tests/session.spec.ts
View File

@@ -75,7 +75,11 @@ describe('open', () => {
const page = plainTurn(10, 0, '早', '安')
session.handleMuxEnvelope('r1' as never, { type: 'session/event', sessionId: SID, event: ev.turnStart(15, 1) })
session.handleMuxEnvelope('r2' as never, { type: 'session/event', sessionId: SID, event: ev.user(16, '插进来的') })
gate.resolve(ok({ events: entries(page) as never[], hasMore: false }))
gate.resolve(ok({
events: entries(page) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
}))
await opening
const seqs = session.getSnapshot().nodes.map(n => n.seq)
// Overlapping seq-15 frame (== page tail turn/end) was dropped; 16 appended once.
@@ -83,6 +87,7 @@ describe('open', () => {
})
})
describe('live event path', () => {
async function opened(events: SessionEvent[] = plainTurn(0, 0, 'a', 'b')) {
const { api, session } = makeSession()
@@ -99,6 +104,28 @@ describe('live event path', () => {
expect(session.getSnapshot().nodes).toEqual(before.nodes)
})
it('materializes a command node from live lifecycle frames and reproduces it from a history window', async () => {
// Live path: run mints an executing node, done settles it in the flow.
const { session } = await opened()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.commandRun(6, 'cmd-live', 'plan'))
let command = session.getSnapshot().nodes.at(-1)
expect(command).toMatchObject({ kind: 'command', name: 'plan', args: '', outcome: null })
feed(ev.commandDone(7, 'cmd-live', 'success', '已进入 plan mode'))
command = session.getSnapshot().nodes.at(-1)
expect(command).toMatchObject({ kind: 'command', seq: 6, outcome: { kind: 'success', text: '已进入 plan mode' } })
// Replay path (refresh): the same pair inside the history window folds identically.
const replayed = await opened([
...plainTurn(0, 0, 'a', 'b'),
ev.commandRun(6, 'cmd-live', 'plan'),
ev.commandDone(7, 'cmd-live', 'success', '已进入 plan mode'),
])
expect(replayed.session.getSnapshot().nodes.at(-1)).toMatchObject({
kind: 'command', seq: 6, name: 'plan', outcome: { kind: 'success', text: '已进入 plan mode' },
})
})
it('accumulates chunks into partial, then finalize swaps partial out as the node lands', async () => {
const { session } = await opened()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
@@ -210,26 +237,44 @@ describe('paging', () => {
api.onHistory = () => gate.promise
const first = session.loadOlder()
const second = session.loadOlder()
gate.resolve(ok({ events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
gate.resolve(ok({
events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
}))
await Promise.all([first, second])
expect(api.callsOf('session.history')).toHaveLength(2) // open + one page, not two
})
})
describe('prompt and cancel errors', () => {
it('sends content through session.prompt with the mode passed through', async () => {
it('sends content through session.prompt; composerPhase steps blank → engaging synchronously at send entry', async () => {
const { api, session } = makeSession()
const result = await session.prompt([{ type: 'text', text: '要发的' }], 'queue')
// The blank → engaging edge fires before the RPC settles: the first-send
// flow reads the phase on the session area's first frame to keep the
// guidance hero from flashing back in.
expect(session.getSnapshot().composerPhase).toBe('blank')
const inFlight = session.prompt([{ type: 'text', text: '要发的' }], 'queue')
expect(session.getSnapshot().composerPhase).toBe('engaging')
const result = await inFlight
expect(result.ok).toBe(true)
// Monotone: settlement alone does not step the phase anywhere.
expect(session.getSnapshot().composerPhase).toBe('engaging')
expect(api.callsOf('session.prompt')).toMatchObject([{ sessionId: SID, mode: 'queue', content: [{ type: 'text', text: '要发的' }] }])
// First content lands (running turn): engaging → active.
session.handleRunning(true)
expect(session.getSnapshot().composerPhase).toBe('active')
})
it('business failure lands in promptError with op=send', async () => {
it('business failure lands in promptError with op=send; the phase stays engaging (retry, no hero bounce)', async () => {
const { api, session } = makeSession()
api.onPrompt = () => Promise.resolve(err({ code: 'agent-busy', message: 'busy', details: { reason: 'x' } }))
const result = await session.prompt([{ type: 'text', text: '失败的' }], 'queue')
expect(result.ok).toBe(false)
expect(session.getSnapshot().promptError).toMatchObject({ op: 'send', error: { code: 'agent-busy' } })
// Failed first prompt: composer + error strip is the retry surface —
// blank is unreachable once a send was initiated.
expect(session.getSnapshot().composerPhase).toBe('engaging')
})
it('lands cancel failures in promptError with op=stop', async () => {
@@ -482,7 +527,11 @@ describe('remaining branches', () => {
const opening = session.open()
api.onHistory = () => histResponse(plainTurn(6, 1, '新', '代'))
const resynced = session.resync()
stale.resolve(ok({ events: entries(plainTurn(0, 0, '旧', '代')) as never[], hasMore: false })) // success, but its generation is gone
stale.resolve(ok({
events: entries(plainTurn(0, 0, '旧', '代')) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'stale' },
})) // success, but its generation is gone
await Promise.all([opening, resynced])
expect(session.getSnapshot().nodes.map(n => n.seq)).toEqual([7, 9]) // only the fresh generation's window
})
@@ -501,7 +550,11 @@ describe('remaining branches', () => {
const opening = session.open() // triggers the second pull, which parks
await vi.waitFor(() => { expect(call).toBe(2) })
const resynced = session.resync()
secondPull.resolve(ok({ events: entries([...plainTurn(0, 0, 'a', 'b'), ...plainTurn(6, 1, 'c', 'd')]) as never[], hasMore: false }))
secondPull.resolve(ok({
events: entries([...plainTurn(0, 0, 'a', 'b'), ...plainTurn(6, 1, 'c', 'd')]) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'stale' },
}))
await Promise.all([opening, resynced])
expect(session.getSnapshot().openState).toBe('open')
})
@@ -515,7 +568,11 @@ describe('remaining branches', () => {
session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event: ev.user(9, '洞') }) // starts repairGap
api.onHistory = () => histResponse(plainTurn(6, 1, 'c', 'd'))
const resynced = session.resync() // bumps the generation
repairPull.resolve(ok({ events: entries(plainTurn(0, 0, '旧', '页')) as never[], hasMore: false })) // repair result: stale, dropped
repairPull.resolve(ok({
events: entries(plainTurn(0, 0, '旧', '页')) as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'stale' },
})) // repair result: stale, dropped
await resynced
expect(session.getSnapshot().nodes.map(n => n.seq)).toEqual([7, 9])
})
@@ -559,6 +616,7 @@ describe('remaining branches', () => {
{ event: ev.toolResult(7, 1, 'h1', 'done'), view: { for: 'result', view: { card: 'generic', title: '历史果' } } },
] as never[],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
}))
await session.open()
expect(session.getSnapshot().nodes.at(-1)).toMatchObject({
@@ -630,6 +688,95 @@ describe('resync', () => {
})
})
describe('run_code sub-dispatch indexing', () => {
it('a start event lands as a running-shaped sub-call and its settle replaces it in place', async () => {
const { api, session } = makeSession()
api.onHistory = () => histResponse(plainTurn(0, 0, '问', '答'))
await session.open()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(ev.toolCall(7, 1, 'p1', 'run_code', '{"code":"1","description":"d"}'))
feed(ev.codeDispatchStart(8, 'p1', 1, 'bash', { command: 'sleep' }))
feed(ev.codeDispatchStart(9, 'p1', 2, 'read', { path: 'a.txt' }))
const live = session.getSnapshot().codeDispatches.get('p1')
expect(live).toHaveLength(2)
// Running shape (no 'kind'): the exact RunningToolCall form native rows use.
expect(live?.[0]).toMatchObject({ callId: 'p1:code:1', name: 'bash', argsRaw: '{"command":"sleep"}' })
expect(live?.[0] !== undefined && 'kind' in live[0]).toBe(false)
// Settle out of order (parallel run): #2 first — replaces in place, keeping start order.
feed(ev.codeDispatch(10, 'p1', 2, 'read', { path: 'a.txt' }, 'alpha'))
const mixed = session.getSnapshot().codeDispatches.get('p1')
expect(mixed?.map(sub => 'kind' in sub)).toEqual([false, true])
expect(mixed?.[1]).toMatchObject({ callId: 'p1:code:2', content: [{ type: 'text', text: 'alpha' }] })
// The settle carries the paired start's time as callTime (duration source).
feed(ev.codeDispatch(11, 'p1', 1, 'bash', { command: 'sleep' }, 'done'))
const settled = session.getSnapshot().codeDispatches.get('p1')
expect(settled?.map(sub => 'kind' in sub)).toEqual([true, true])
expect(settled?.[0]).toMatchObject({ callId: 'p1:code:1', callTime: 1_700_000_000_008 })
})
it('indexes live tool/code-dispatch events under their parent as native-shaped result nodes', async () => {
const { api, session } = makeSession()
api.onHistory = () => histResponse(plainTurn(0, 0, '问', '答'))
await session.open()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(ev.toolCall(7, 1, 'p1', 'run_code', '{"code":"return 1","description":"跑一个程序"}'))
feed(ev.codeDispatch(8, 'p1', 1, 'bash', { command: 'ls', description: '列目录' }, 'demo.txt'))
feed(ev.codeDispatch(9, 'p1', 2, 'read', { path: 'a.txt' }, 'Error: ENOENT', true))
const subs = session.getSnapshot().codeDispatches.get('p1')
expect(subs).toHaveLength(2)
expect(subs?.[0]).toMatchObject({
kind: 'tool-result', callId: 'p1:code:1',
call: { name: 'bash', argsRaw: '{"command":"ls","description":"列目录"}' },
// The settle event carries no start time: callTime stays null (never a
// fabricated zero-duration).
callTime: null,
isError: false, content: [{ type: 'text', text: 'demo.txt' }],
})
expect(subs?.[1]).toMatchObject({ callId: 'p1:code:2', isError: true })
// No paired start in the window: duration is UNKNOWN (null), never a
// fabricated zero-duration span.
expect(subs?.[0]).toMatchObject({ callTime: null })
// Sub-dispatches never join the surface flow.
expect(session.getSnapshot().nodes.some(n => n.kind === 'tool-result' && n.callId.includes(':code:'))).toBe(false)
})
it('rebuilds the same index from a history window (replay parity)', async () => {
const { api, session } = makeSession()
api.onHistory = () => histResponse([
...plainTurn(0, 0, '问', '答'),
ev.turnStart(6, 1),
ev.toolCall(7, 1, 'p1', 'run_code', '{"code":"return 1","description":"跑一个程序"}'),
ev.codeDispatch(8, 'p1', 1, 'bash', { command: 'ls' }, 'demo.txt'),
ev.toolResult(9, 1, 'p1', '{"done":true}'),
ev.turnEnd(10, 1),
])
await session.open()
const subs = session.getSnapshot().codeDispatches.get('p1')
expect(subs).toHaveLength(1)
expect(subs?.[0]).toMatchObject({ callId: 'p1:code:1', call: { name: 'bash' } })
})
it('keeps the dispatch map reference across unrelated changes and swaps it on a new dispatch', async () => {
const { api, session } = makeSession()
api.onHistory = () => histResponse(plainTurn(0, 0, '稳', '定'))
await session.open()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(ev.toolCall(7, 1, 'p1', 'run_code', '{"code":"1","description":"d"}'))
feed(ev.codeDispatch(8, 'p1', 1, 'bash', { command: 'ls' }, 'x'))
const before = session.getSnapshot()
feed(ev.chunkStart(9, 1))
feed(ev.chunkText(10, 1, '流式'))
const after = session.getSnapshot()
expect(after.codeDispatches).toBe(before.codeDispatches)
feed(ev.codeDispatch(11, 'p1', 2, 'read', { path: 'a' }, 'y'))
expect(session.getSnapshot().codeDispatches).not.toBe(after.codeDispatches)
expect(session.getSnapshot().codeDispatches.get('p1')).toHaveLength(2)
})
})
describe('reference stability (the memo contract)', () => {
it('keeps unchanged node references across an append and swaps the snapshot object', async () => {
const { api, session } = makeSession()

256
packages/client/runtime/tests/sessions-service.spec.ts
View File

@@ -9,8 +9,8 @@
import { Context } from 'cordis'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { SessionsService, scopeOf } from '../src/client/sessions/service.ts'
import { FakeApiClient, ok } from './fake-api.ts'
import { SessionCreateError, SessionsService, scopeOf } from '../src/client/sessions/service.ts'
import { FakeApiClient, deferred, ok } from './fake-api.ts'
const sid = (s: string): SessionId => s as SessionId
@@ -28,24 +28,26 @@ function bench(): Bench {
}
/** Refresh the manager list from programmable rows and flush the microtask batch. */
async function feedList(b: Bench, rows: { id: string; cwd?: string; parentId?: string; running?: boolean }[]): Promise<void> {
type FeedRow = { id: string; cwd?: string; parentId?: string; running?: boolean; blank?: boolean }
async function feedList(b: Bench, rows: FeedRow[]): Promise<void> {
b.api.onList = () => Promise.resolve(ok({
items: rows.map(r => ({
sessionId: sid(r.id), updatedAt: 1, running: r.running ?? false,
sessionId: sid(r.id), updatedAt: 1, running: r.running ?? false, blank: r.blank ?? false,
...(r.cwd !== undefined ? { cwd: r.cwd } : {}),
...(r.parentId !== undefined ? { parentSessionId: sid(r.parentId) } : {}),
})),
}) as never)
await b.svc.manager.refreshList()
await b.svc.refresh()
await Promise.resolve() // manager notifier flush
}
describe('list store projection', () => {
it('projects durable titles separately from cwd/id display fallbacks and parent links', async () => {
const b = bench()
b.svc.manager.handleMuxEnvelope({
b.svc.handleMuxEnvelope({
rpcId: 'title' as never,
payload: { type: 'session/title', sessionId: sid('s1'), title: 'Durable title', eventSeq: 2, updatedAt: 3 },
payload: { type: 'session/projection', sessionId: sid('s1'), key: 'title', value: 'Durable title', seq: 2 } as never,
})
await feedList(b, [
{ id: 's1', cwd: '/home/u/proj-a/' },
@@ -61,7 +63,7 @@ describe('list store projection', () => {
it('reflects live increments (host stream via manager) into the store', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }])
b.svc.manager.handleHostEnvelope({ rpcId: 'r1' as never, payload: { type: 'host/session-added', sessionId: sid('s2') } as never })
b.svc.handleHostEnvelope({ rpcId: 'r1' as never, payload: { type: 'host/session-added', blank: true, sessionId: sid('s2') } as never })
await Promise.resolve()
expect(b.svc.list.getSnapshot().ids).toContain('s2')
})
@@ -77,7 +79,8 @@ describe('scope tree', () => {
expect(scopeOf(scoped as Context)).toBe('s1')
expect(scopeOf(b.ctx)).toBeUndefined()
const binding = b.svc.binding(sid('s1'))
expect(binding?.session).toBe(b.svc.manager.get(sid('s1')))
b.svc.open(sid('s1'))
expect(binding?.session).toBe(b.svc.currentProvideInfo.getSnapshot().hooks['session'])
expect(b.svc.binding(sid('s1'))).toBe(binding)
expect(binding?.ctx).toBe(scoped)
})
@@ -181,25 +184,82 @@ describe('current selection (migrated from ui-layout, arbitrated into the list s
})
describe('cell (render-layer session kit)', () => {
it('resolves an identity-stable {sessionId, session} cell; unknown ids yield undefined', async () => {
it('resolves an identity-stable {sessionId, session} cell through the current projection', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }])
const cell = b.svc.cell('s1')
expect(cell).toBeDefined()
expect(cell?.sessionId).toBe('s1')
// Bare-source form (store migration): the cell carries the Session
// observable itself; hook binding happens in the React machinery.
expect(cell?.session).toBe(b.svc.manager.get(sid('s1')))
expect(b.svc.cell('s1')).toBe(cell)
expect(b.svc.cell('ghost')).toBeUndefined()
b.svc.open(sid('s1'))
const info = b.svc.currentProvideInfo.getSnapshot()
expect(info.sessionId).toBe('s1')
// The bundle carries bare observables; hook binding happens in React.
expect(info.hooks['session']).toBe(b.svc.binding(sid('s1'))?.session)
// Re-staging the same id republishes nothing: identity holds.
b.svc.open(sid('s1'))
expect(b.svc.currentProvideInfo.getSnapshot()).toBe(info)
})
it('cell()/binding() are pure resolution: no staging, no deferred sweep', async () => {
it('currentProvideInfo follows selection: absent projection ↔ definite bundle, notified on each move', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }, { id: 's2' }])
const absent = b.svc.currentProvideInfo.getSnapshot()
expect(absent.sessionId).toBeUndefined()
expect(Object.hasOwn(absent.hooks, 'session')).toBe(true)
const notified = vi.fn()
b.svc.currentProvideInfo.subscribe(notified)
b.svc.open(sid('s1'))
const s1Bundle = b.svc.currentProvideInfo.getSnapshot()
expect(s1Bundle.sessionId).toBe('s1')
expect(s1Bundle.hooks['session']).toBe(b.svc.binding(sid('s1'))?.session)
expect(notified).toHaveBeenCalledTimes(1)
b.svc.open(sid('s2'))
const s2Bundle = b.svc.currentProvideInfo.getSnapshot()
expect(s2Bundle.sessionId).toBe('s2')
expect(s2Bundle).not.toBe(s1Bundle)
expect(notified).toHaveBeenCalledTimes(2)
b.svc.clear()
await Promise.resolve() // clearSelection projects through the manager notifier
expect(b.svc.currentProvideInfo.getSnapshot().sessionId).toBeUndefined()
})
it('a provider roster change under a stable current id republishes the bundle', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }])
b.svc.open(sid('s1'))
const before = b.svc.currentProvideInfo.getSnapshot()
const notified = vi.fn()
b.svc.currentProvideInfo.subscribe(notified)
const source = { getSnapshot: () => 'live', subscribe: () => () => {} }
const dispose = b.svc.provide({
hooks: ['extra'],
props: ['marker'],
resolve: () => ({ hooks: { extra: source }, props: { marker: 7 } }),
})
const added = b.svc.currentProvideInfo.getSnapshot()
expect(added).not.toBe(before)
expect(added).toMatchObject({ sessionId: 's1', props: { marker: 7 } })
expect(added.hooks['extra']).toBe(source)
expect(notified).toHaveBeenCalledTimes(1)
dispose()
const removed = b.svc.currentProvideInfo.getSnapshot()
expect(removed).not.toBe(added)
expect(Object.hasOwn(removed.hooks, 'extra')).toBe(false)
expect(notified).toHaveBeenCalledTimes(2)
})
it('an unsubscribed currentProvideInfo listener stops receiving notifications', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }])
const notified = vi.fn()
const off = b.svc.currentProvideInfo.subscribe(notified)
off()
b.svc.open(sid('s1'))
expect(notified).not.toHaveBeenCalled()
})
it('binding() is pure resolution: no staging, no deferred sweep', async () => {
const b = bench()
await feedList(b, [{ id: 's1' }, { id: 's2' }])
b.svc.open(sid('s1')) // staged
b.svc.cell('s2') // resolution only — must NOT move the stage
b.svc.binding(sid('s2'))
b.svc.binding(sid('s2')) // resolution only — must NOT move the stage
await feedList(b, [{ id: 's2' }]) // s1 removed: still staged → deferred, scope survives
expect(b.svc.scope(sid('s1'))).toBeDefined()
})
@@ -210,7 +270,6 @@ describe('cell (render-layer session kit)', () => {
const historyCalls = () => b.api.calls.filter(c => c.method === 'session.history')
// Resolution is addressing, not staging: no window pull.
b.svc.scope(sid('s1'))
b.svc.cell('s1')
b.svc.binding(sid('s1'))
expect(historyCalls()).toHaveLength(0)
b.svc.open(sid('s1'))
@@ -285,36 +344,157 @@ describe('ancestry', () => {
})
describe('create', () => {
it('returns the new id on ok and throws a coded error on failure', async () => {
it('passes a preallocated id and preserves it on ordinary failure', async () => {
const b = bench()
b.api.onCreate = () => Promise.resolve(ok({ sessionId: sid('fresh') }))
await expect(b.svc.create({ cwd: '/w' })).resolves.toBe('fresh')
await expect(b.svc.create({ cwd: '/w', sessionId: sid('fresh') })).resolves.toBe('fresh')
expect(b.api.callsOf('session.create')).toEqual([{ cwd: '/w', sessionId: 'fresh' }])
b.api.onCreate = () => Promise.resolve({
rpcId: 'e' as never,
result: { ok: false as const, error: { code: 'internal' as const, message: '爆了', details: {} } },
} as never)
await expect(b.svc.create()).rejects.toThrow(/internal: 爆了/)
const failure = await b.svc.create({ sessionId: sid('candidate') }).catch((error: unknown) => error)
expect(failure).toBeInstanceOf(SessionCreateError)
expect(failure).toMatchObject({
requestedSessionId: 'candidate',
rpcError: { code: 'internal', message: '爆了' },
})
})
it('resolves with the session already listed and binding-resolvable (no flush wait)', async () => {
const b = bench()
b.api.onCreate = () => Promise.resolve(ok({ sessionId: sid('born') }))
const born = await b.svc.create({ workspaceId: 'ws' as never })
// Synchronously after resolution — the draft hand-off contract: the
// create echo IS the entity entering the client's view (blank row +
// resolvable scope/binding), no notifier flush in between.
expect(b.svc.list.getSnapshot().byId[born]).toMatchObject({ id: 'born', blank: true })
expect(b.svc.binding(born)).toBeDefined()
expect(b.svc.scope(born)).toBeDefined()
})
it('lists the published id after Workspace attachment fails (publication precedes attachment)', async () => {
const b = bench()
b.api.onCreate = () => Promise.resolve({
rpcId: 'attach' as never,
result: {
ok: false,
error: {
code: 'workspace-attach-failed', message: 'ledger unavailable',
details: { sessionId: sid('published'), workspaceId: 'ws' },
},
},
} as never)
const failure = await b.svc.create({
workspaceId: 'ws' as never,
sessionId: sid('published'),
}).catch((error: unknown) => error)
await Promise.resolve()
expect(failure).toBeInstanceOf(SessionCreateError)
expect(failure).toMatchObject({
requestedSessionId: 'published',
rpcError: { code: 'workspace-attach-failed' },
})
expect(b.svc.list.getSnapshot().byId[sid('published')]).toMatchObject({ id: 'published', blank: true })
})
})
describe('createWorkspace', () => {
it('joins host.describe cwd with the name and creates there', async () => {
describe('scope lifecycle rides the list mirror (entity parity: no client-side pre-birth)', () => {
it('a session-added frame births the row (blank) and makes the scope resolvable; removal prunes it', async () => {
const b = bench()
b.api.onDescribe = () => Promise.resolve(ok({ version: '0', cwd: '/host/root', attachedSessions: 0 }))
b.api.onCreate = () => Promise.resolve(ok({ sessionId: sid('ws') }))
await expect(b.svc.createWorkspace('My Proj')).resolves.toBe('ws')
expect(b.api.callsOf('session.create')).toEqual([{ cwd: '/host/root/My Proj' }])
await feedList(b, [])
expect(b.svc.scope(sid('s-new'))).toBeUndefined() // not in view: no scope, no exceptions
b.svc.handleHostEnvelope({
rpcId: 'add' as never,
payload: { type: 'host/session-added', sessionId: sid('s-new'), blank: true, cwd: '/w/a' } as never,
})
await Promise.resolve()
const scoped = b.svc.scope(sid('s-new'))
expect(scoped).toBeDefined()
expect(scopeOf(scoped as Context)).toBe('s-new')
b.svc.handleHostEnvelope({
rpcId: 'rm' as never,
payload: { type: 'host/session-removed', sessionId: sid('s-new') },
})
await Promise.resolve()
expect(b.svc.scope(sid('s-new'))).toBeUndefined()
})
})
describe('blank mirror', () => {
it('flips blank=false from the running:true status frame (cross-client conversion)', async () => {
const b = bench()
await feedList(b, [{ id: 's1', blank: true }])
expect(b.svc.list.getSnapshot().byId[sid('s1')]).toMatchObject({ blank: true })
b.svc.handleHostEnvelope({
rpcId: 'st' as never,
payload: { type: 'host/session-status', sessionId: sid('s1'), running: true },
})
await Promise.resolve()
expect(b.svc.list.getSnapshot().byId[sid('s1')]).toMatchObject({ blank: false, running: true })
// The instantiated Session mirrors the same flip.
expect(b.svc.binding(sid('s1'))?.session.getSnapshot().blank).toBe(false)
})
it('rejects empty names and path separators; surfaces describe failures', async () => {
it('flips blank=false on prompt ACCEPTANCE, not on the attempt', async () => {
const b = bench()
await expect(b.svc.createWorkspace(' ')).rejects.toThrow(/name is required/)
await expect(b.svc.createWorkspace('a/b')).rejects.toThrow(/path separators/)
b.api.onDescribe = () => Promise.resolve({
rpcId: 'e' as never,
result: { ok: false as const, error: { code: 'internal' as const, message: 'down', details: {} } },
await feedList(b, [{ id: 's1', blank: true, cwd: '/w/a' }])
const session = b.svc.binding(sid('s1'))!.session
expect(session.getSnapshot().blank).toBe(true)
const gate = deferred<Awaited<ReturnType<FakeApiClient['onPrompt']>>>()
b.api.onPrompt = () => gate.promise
const send = session.prompt([{ type: 'text', text: 'hi' }], 'queue')
// In flight: still blank (the flip point is the success response, which
// proves the user message reached the host log).
expect(session.getSnapshot().blank).toBe(true)
gate.resolve(ok({ accepted: true as const }))
await send
expect(session.getSnapshot().blank).toBe(false)
await Promise.resolve()
expect(b.svc.list.getSnapshot().byId[sid('s1')]).toMatchObject({ blank: false })
})
it('keeps a rejected first prompt blank: hidden and still reusable', async () => {
const b = bench()
await feedList(b, [{ id: 's1', blank: true, cwd: '/w/a' }])
const session = b.svc.binding(sid('s1'))!.session
b.api.onPrompt = () => Promise.resolve({
rpcId: 'busy' as never,
result: { ok: false as const, error: { code: 'internal' as const, message: 'agent busy', details: {} } },
} as never)
await expect(b.svc.createWorkspace('ok')).rejects.toThrow(/host.describe failed/)
const result = await session.prompt([{ type: 'text', text: 'hi' }], 'queue')
expect(result.ok).toBe(false)
// No flip on failure: local stays aligned with the host authority
// (events.length still 0), so the session stays hidden and reusable.
expect(session.getSnapshot().blank).toBe(true)
await Promise.resolve()
expect(b.svc.list.getSnapshot().byId[sid('s1')]).toMatchObject({ blank: true })
})
it('takes session-added blank=true as the hidden birth and list blank as reconnect authority', async () => {
const b = bench()
await feedList(b, [])
b.svc.handleHostEnvelope({
rpcId: 'add' as never,
payload: { type: 'host/session-added', sessionId: sid('s-new'), blank: true, cwd: '/w/a' } as never,
})
await Promise.resolve()
expect(b.svc.list.getSnapshot().byId[sid('s-new')]).toMatchObject({ blank: true })
// Reconnect re-pull: the summary's blank=false wins (authoritative alignment).
await feedList(b, [{ id: 's-new', blank: false, cwd: '/w/a' }])
expect(b.svc.list.getSnapshot().byId[sid('s-new')]).toMatchObject({ blank: false })
})
it('never re-blanks: a stale blank=true summary cannot hide an engaged session', async () => {
const b = bench()
await feedList(b, [{ id: 's1', blank: true }])
const session = b.svc.binding(sid('s1'))!.session
await session.prompt([{ type: 'text', text: 'hi' }], 'queue')
await Promise.resolve()
expect(b.svc.list.getSnapshot().byId[sid('s1')]).toMatchObject({ blank: false })
// The next list pull still claims blank (host hasn't logged the message yet).
await feedList(b, [{ id: 's1', blank: true }])
expect(b.svc.binding(sid('s1'))?.session.getSnapshot().blank).toBe(false)
})
})

36
packages/client/runtime/tests/slots-service.spec.ts
View File

@@ -85,19 +85,25 @@ function captureHost(bench: Bench, children?: object): SlotRendererHost {
})
bench.erased.register({ name: 'root', ...(children !== undefined ? { children } : {}) }, C)
bench.ctx.reflect.provide('sessions', fakeSessions())
bench.ctx.reflect.provide('workspaces', fakeWorkspaces())
bench.erased.renderSlot('root', {})
if (host === undefined) throw new Error('renderer never received the host')
return host
}
/** Minimal sessions face for the host seam (list observable + cell). */
/** Minimal independent Workspace list source for the renderer host seam. */
function fakeWorkspaces() {
const state = { items: [], phase: 'ready' as const }
return { list: { getSnapshot: () => state, subscribe: () => () => undefined } }
}
/** Minimal sessions face for the host seam (list observable + current provide projection). */
function fakeSessions() {
const state = { ids: [], byId: {}, current: undefined as string | undefined }
const absentInfo = { sessionId: undefined, hooks: { session: undefined }, props: {} }
return {
list: { getSnapshot: () => state, subscribe: () => () => undefined },
cell: (id: string) => (id === 'known'
? { sessionId: id, session: { getSnapshot: () => undefined, subscribe: () => () => undefined } }
: undefined),
currentProvideInfo: { getSnapshot: () => absentInfo, subscribe: () => () => undefined },
}
}
@@ -190,9 +196,18 @@ describe('renderer install seam', () => {
bench.erased.install({ renderRoot })
bench.erased.register({ name: 'root' }, C)
bench.ctx.reflect.provide('sessions', fakeSessions())
bench.ctx.reflect.provide('workspaces', fakeWorkspaces())
expect(bench.erased.renderSlot('root', {})).toBe('tree')
expect(renderRoot).toHaveBeenCalledTimes(1)
})
it('fails before rendering when the Workspace object layer is absent', async () => {
const bench = await boot()
bench.erased.install({ renderRoot: () => null })
bench.erased.register({ name: 'root' }, C)
bench.ctx.reflect.provide('sessions', fakeSessions())
expect(() => bench.erased.renderSlot('root', {})).toThrow(/workspaces service mounted/)
})
})
describe('host face', () => {
@@ -212,13 +227,17 @@ describe('host face', () => {
expect(host.entriesOf('t.host')).toHaveLength(0)
})
it('exposes sessions list/current/cell (current riding the list snapshot)', async () => {
it('exposes the session list and the atomic current provide projection', async () => {
const bench = await boot()
const host = captureHost(bench)
expect(host.sessions.list.getSnapshot()).toMatchObject({ ids: [] })
expect(host.sessions.current.getSnapshot()).toBeUndefined()
expect(host.sessions.cell('known')).toMatchObject({ sessionId: 'known' })
expect(host.sessions.cell('ghost')).toBeUndefined()
expect(host.sessions.provideInfo.getSnapshot()).toMatchObject({ sessionId: undefined })
})
it('exposes the independent Workspace list source', async () => {
const bench = await boot()
const host = captureHost(bench)
expect(host.workspaces.list.getSnapshot()).toEqual({ items: [], phase: 'ready' })
})
})
@@ -315,6 +334,7 @@ describe('entry-unload cascade', () => {
renderRoot: (h: SlotRendererHost) => { host = h; return 'rendered' },
})
bench.ctx.reflect.provide('sessions', fakeSessions())
bench.ctx.reflect.provide('workspaces', fakeWorkspaces())
// The declarer here is NOT the root occupant: root stays occupied by a
// separate entry so disposing the declarer only kills its children.
const disposeRoot = bench.erased.register({ name: 'root' }, C)

55
packages/client/runtime/tests/wire-events.spec.ts Normal file
View File

@@ -0,0 +1,55 @@
/**
* Wire-to-typed-event bridge (web input-triggers cut 1): host/commands-changed
* → ctx 'commands/changed'; each established connection generation →
* ctx 'connection/reset' (the forced cache-invalidation broadcast).
*/
import { Context } from 'cordis'
import { describe, expect, it } from 'vitest'
import type { ConnectionHandle, ConnectionSinks } from '@deepseek-ai/dsh-client-connection/client'
import * as RuntimeClient from '../src/client/index.ts'
import { FakeApiClient } from './fake-api.ts'
interface Bench {
ctx: Context
sinks: ConnectionSinks | undefined
}
async function mount(): Promise<Bench> {
const ctx = new Context()
const api = new FakeApiClient()
const bench: Bench = { ctx, sinks: undefined }
const handle: ConnectionHandle = {
api,
start: (sinks) => {
bench.sinks = sinks
return { stop: () => {} }
},
}
ctx.reflect.provide('connection', handle)
await ctx.plugin(RuntimeClient).await()
return bench
}
describe('wire event bridge', () => {
it('broadcasts commands/changed on a host/commands-changed frame, not on other host frames', async () => {
const bench = await mount()
let changed = 0
bench.ctx.on('commands/changed', () => { changed++ })
bench.sinks?.onHostEnvelope?.({ rpcId: 'r1' as never, payload: { type: 'host/commands-changed' } })
expect(changed).toBe(1)
bench.sinks?.onHostEnvelope?.({
rpcId: 'r2' as never,
payload: { type: 'host/session-status', sessionId: 's1' as never, running: true },
})
expect(changed).toBe(1)
})
it('broadcasts connection/reset on every established generation (reconnect invalidation)', async () => {
const bench = await mount()
let resets = 0
bench.ctx.on('connection/reset', () => { resets++ })
bench.sinks?.onConnected?.()
bench.sinks?.onConnected?.() // second generation after a reconnect
expect(resets).toBe(2)
})
})

265
packages/client/runtime/tests/workspaces-service.spec.ts Normal file
View File

@@ -0,0 +1,265 @@
import { Context } from 'cordis'
import { describe, expect, it } from 'vitest'
import type { SessionId, WorkspaceId, WorkspaceView } from '@deepseek-ai/dsh-client-connection/client'
import { SessionsService } from '../src/client/sessions/service.ts'
import { WorkspaceManager } from '../src/client/workspaces/manager.ts'
import { WorkspaceCreateError, WorkspacesService } from '../src/client/workspaces/service.ts'
import { FakeApiClient, deferred, err, ok } from './fake-api.ts'
const sid = (id: string): SessionId => id as SessionId
const wid = (id: string): WorkspaceId => id as WorkspaceId
function workspace(id: string, sessionIds: SessionId[] = [], createdAt = '2026-01-01T00:00:00.000Z'): WorkspaceView {
return {
workspaceId: wid(id), path: `/w/${id}`, title: id, sessionIds,
createdAt, updatedAt: createdAt,
}
}
describe('WorkspaceManager', () => {
it('replays changed frames over hydration and keeps established order on refresh', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onWorkspaceList']>>>()
api.onWorkspaceList = () => gate.promise
const manager = new WorkspaceManager(api)
const hydration = manager.refresh()
manager.handleHostEnvelope({
rpcId: 'changed' as never,
payload: { type: 'host/workspace-changed', workspace: workspace('new') },
})
gate.resolve(ok({ items: [workspace('old')] as never[] }))
await hydration
expect(manager.getSnapshot()).toMatchObject({ phase: 'ready', state: 'idle' })
expect(manager.getSnapshot().items.map(item => item.workspaceId)).toEqual(['new', 'old'])
api.onWorkspaceList = () => Promise.resolve(ok({
items: [workspace('old'), workspace('new')] as never[],
}))
await manager.refresh()
expect(manager.getSnapshot().items.map(item => item.workspaceId)).toEqual(['new', 'old'])
})
it('single-flights refreshes and exposes result and transport failures independently of readiness', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onWorkspaceList']>>>()
api.onWorkspaceList = () => gate.promise
const manager = new WorkspaceManager(api)
const first = manager.refresh()
const second = manager.refresh()
expect(manager.getSnapshot().state).toBe('loading')
gate.resolve(ok({ items: [] }))
await Promise.all([first, second])
expect(api.callsOf('workspace.list')).toHaveLength(1)
api.onWorkspaceList = () => Promise.resolve(err({ code: 'internal', message: 'down', details: {} }))
await manager.refresh()
expect(manager.getSnapshot()).toMatchObject({ phase: 'ready', state: 'error', error: { message: 'down' } })
api.onWorkspaceList = () => Promise.reject(new Error('wire down'))
await manager.refresh()
expect(manager.getSnapshot()).toMatchObject({ phase: 'ready', state: 'error', error: { message: 'wire down' } })
})
it('creates by name/path, prepends a new row, and folds failures', async () => {
const api = new FakeApiClient()
const manager = new WorkspaceManager(api)
api.onWorkspaceCreate = payload => Promise.resolve(ok({
workspace: workspace('created', [], '2026-02-01T00:00:00.000Z'),
created: true,
payload,
} as never))
await expect(manager.create({ name: 'created' })).resolves.toMatchObject({ ok: true })
expect(api.callsOf('workspace.create')).toEqual([{ name: 'created' }])
expect(manager.getSnapshot().items[0]?.workspaceId).toBe('created')
api.onWorkspaceCreate = () => Promise.reject(new Error('create transport'))
await expect(manager.create({ path: '/w/existing' })).resolves.toMatchObject({
ok: false, error: { code: 'internal', message: 'create transport' },
})
})
it('replays removal over an in-flight baseline and ignores duplicate or late updates', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onWorkspaceList']>>>()
api.onWorkspaceList = () => gate.promise
const manager = new WorkspaceManager(api)
const hydration = manager.refresh()
manager.handleHostEnvelope({
rpcId: 'removed' as never,
payload: { type: 'host/workspace-removed', workspaceId: wid('gone') },
})
gate.resolve(ok({ items: [workspace('gone'), workspace('kept')] as never[] }))
await hydration
expect(manager.getSnapshot().items.map(item => item.workspaceId)).toEqual(['kept'])
manager.handleHostEnvelope({
rpcId: 'late-change' as never,
payload: { type: 'host/workspace-changed', workspace: workspace('gone') },
})
manager.handleHostEnvelope({
rpcId: 'duplicate-remove' as never,
payload: { type: 'host/workspace-removed', workspaceId: wid('gone') },
})
expect(manager.getSnapshot().items.map(item => item.workspaceId)).toEqual(['kept'])
})
it('removes from the unary delete echo while a refresh is in flight', async () => {
const api = new FakeApiClient()
api.onWorkspaceList = () => Promise.resolve(ok({ items: [workspace('gone')] as never[] }))
const manager = new WorkspaceManager(api)
await manager.refresh()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onWorkspaceList']>>>()
api.onWorkspaceList = () => gate.promise
const refresh = manager.refresh()
await expect(manager.delete(wid('gone'))).resolves.toMatchObject({ ok: true })
expect(api.callsOf('workspace.delete')).toEqual([{ workspaceId: 'gone' }])
expect(manager.getSnapshot().items).toEqual([])
gate.resolve(ok({ items: [workspace('gone')] as never[] }))
await refresh
expect(manager.getSnapshot().items).toEqual([])
})
})
describe('WorkspacesService', () => {
it('feeds readiness and recent-Workspace targeting without changing Host order', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onWorkspaceList = () => Promise.resolve(ok({
items: [
workspace('stable-first', [], '2026-01-03T00:00:00.000Z'),
workspace('active', [sid('s-active')], '2026-01-01T00:00:00.000Z'),
] as never[],
}))
await workspaces.refresh()
await Promise.resolve()
expect(workspaces.list.getSnapshot()).toMatchObject({ baselinesReady: false, recentWorkspaceId: undefined })
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s-active'), updatedAt: Date.parse('2026-02-01'), running: false, blank: false }] as never[],
}))
await sessions.refresh()
await Promise.resolve()
await Promise.resolve()
expect(workspaces.list.getSnapshot()).toMatchObject({
baselinesReady: true,
recentWorkspaceId: 'active',
})
expect(workspaces.list.getSnapshot().items.map(item => item.workspaceId)).toEqual(['stable-first', 'active'])
})
it('connectWorkspace reuses the workspace-matched blank session and creates otherwise', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onWorkspaceList = () => Promise.resolve(ok({
items: [workspace('alpha'), workspace('beta')] as never[],
}))
api.onList = () => Promise.resolve(ok({
items: [
// Blank session already parked in alpha (cwd == workspace path canon).
{ sessionId: sid('s-blank'), updatedAt: 2, running: false, blank: true, cwd: '/w/alpha' },
// Non-blank sibling in beta must never be reused.
{ sessionId: sid('s-active'), updatedAt: 3, running: false, blank: false, cwd: '/w/beta' },
] as never[],
}))
await Promise.all([workspaces.refresh(), sessions.refresh()])
await Promise.resolve()
// Hit: same workspace → the parked blank session comes back, no create RPC.
await expect(workspaces.connectWorkspace(wid('alpha'))).resolves.toBe('s-blank')
expect(api.callsOf('session.create')).toEqual([])
// Resolution guarantee: the id is binding-resolvable synchronously.
expect(sessions.binding(sid('s-blank'))).toBeDefined()
// Miss: beta has only a non-blank session → host create with workspaceId.
api.onCreate = () => Promise.resolve(ok({ sessionId: sid('s-fresh') }))
await expect(workspaces.connectWorkspace(wid('beta'))).resolves.toBe('s-fresh')
expect(api.callsOf('session.create')).toEqual([{ workspaceId: 'beta' }])
// Same guarantee on the create arm (draft hand-off writes the machine pre-open).
expect(sessions.binding(sid('s-fresh'))).toBeDefined()
// Unknown workspace fails loud instead of silently creating in nowhere.
await expect(workspaces.connectWorkspace(wid('ghost'))).rejects.toThrow(/unknown workspace ghost/)
})
it('a rejected first prompt keeps the blank session eligible for connectWorkspace reuse', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onWorkspaceList = () => Promise.resolve(ok({ items: [workspace('alpha')] as never[] }))
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s-blank'), updatedAt: 2, running: false, blank: true, cwd: '/w/alpha' }] as never[],
}))
await Promise.all([workspaces.refresh(), sessions.refresh()])
await Promise.resolve()
const session = sessions.binding(sid('s-blank'))!.session
api.onPrompt = () => Promise.resolve(err({ code: 'internal', message: 'agent busy', details: {} }) as never)
await session.prompt([{ type: 'text', text: 'hi' }], 'queue')
await Promise.resolve()
// Failure leaves blank intact, so the same session is still the reuse hit.
await expect(workspaces.connectWorkspace(wid('alpha'))).resolves.toBe('s-blank')
expect(api.callsOf('session.create')).toEqual([])
})
it('returns created Workspaces and preserves Host business errors', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onWorkspaceCreate = () => Promise.resolve(ok({
workspace: { ...workspace('picked'), path: '/w/alpha', title: 'alpha' }, created: true,
}))
await expect(workspaces.create({ path: '/w/alpha' })).resolves.toMatchObject({ workspaceId: 'picked' })
expect(workspaces.list.getSnapshot().items[0]).toMatchObject({ path: '/w/alpha', title: 'alpha' })
expect(api.callsOf('workspace.create')).toEqual([{ path: '/w/alpha' }])
api.onWorkspaceCreate = () => Promise.resolve(err({
code: 'workspace-invalid-path', message: 'missing', details: { path: '/missing' },
}))
const rejected = workspaces.create({ path: '/missing' })
await expect(rejected).rejects.toThrow(/workspace-invalid-path: missing/)
await expect(rejected).rejects.toBeInstanceOf(WorkspaceCreateError)
})
it('passes native directory selection and cancellation through without local state', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onPickDirectory = () => Promise.resolve(ok({ path: '/w/alpha' }))
await expect(workspaces.pickDirectory()).resolves.toBe('/w/alpha')
api.onPickDirectory = () => Promise.resolve(ok({ path: null }))
await expect(workspaces.pickDirectory()).resolves.toBeNull()
expect(api.callsOf('host.pickDirectory')).toEqual([{}, {}])
})
it('opens a filesystem path through the host without local state', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
await expect(workspaces.openPath('/w/alpha/a.ts')).resolves.toBeUndefined()
expect(api.callsOf('host.openPath')).toEqual([{ path: '/w/alpha/a.ts' }])
api.onOpenPath = () => Promise.resolve(err({ code: 'internal', message: 'boom', details: {} }))
await expect(workspaces.openPath('/missing')).rejects.toThrow(/path open failed/)
})
it('deletes a Workspace or preserves it when the Host rejects deletion', async () => {
const ctx = new Context()
const api = new FakeApiClient()
const sessions = new SessionsService(ctx, api)
const workspaces = new WorkspacesService(ctx, api, sessions)
api.onWorkspaceList = () => Promise.resolve(ok({ items: [workspace('alpha')] as never[] }))
await workspaces.refresh()
await expect(workspaces.delete(wid('alpha'))).resolves.toBeUndefined()
expect(workspaces.list.getSnapshot().items).toEqual([])
api.onWorkspaceDelete = () => Promise.resolve(err({
code: 'workspace-not-found', message: 'gone', details: { workspaceId: 'ghost' },
}))
await expect(workspaces.delete(wid('ghost'))).rejects.toThrow(/workspace-not-found: gone/)
})
})