Merge master (slash/input/session architecture) into web-session-model-selector

This commit is contained in:
imccyu
2026-07-27 10:23:51 +08:00
2673 changed files with 101832 additions and 39946 deletions

View File

@@ -0,0 +1,100 @@
/**
* Cold-session and degenerate-composition paths of the host ApiProxy:
* sessions.list merging persisted-but-unattached summaries (mtime source,
* createdAt fallbacks, lineage projection) and the resume error split when
* the composition has no persistence gate and no agent factory.
*/
import { mkdtempSync, writeFileSync, utimesSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import SessionStore from '@deepseek-ai/dsh-session'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
const sid = (id: string): SessionId => id as SessionId
let nextRpc = 1
function request<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId(`cold-${String(nextRpc++)}`), payload }
}
function header(id: string, createdAt: number, extra: Partial<SessionHeader> = {}): SessionHeader {
return { version: 0, id: sid(id), createdAt, cwd: '/proj', ...extra }
}
describe('sessions.list cold merge', () => {
it('summarizes unattached sessions: log mtime, locate-less and vanished-log createdAt fallbacks, lineage', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserInteractionService)
const root = mkdtempSync(join(tmpdir(), 'dsh-cold-'))
const logPath = join(root, 'a.log')
writeFileSync(logPath, 'log-bytes')
utimesSync(logPath, 5000, 5000) // mtime 5_000_000 ms — newer than every createdAt below
const metas = [
header('session-a', 1000),
header('session-b', 2000, { parentSession: sid('session-parent') }),
header('session-c', 1500),
]
// Structural fake of the persistence face list() consumes: list + locate.
// locate: a real per-session file (mtime wins), a backend without one
// (SQLite shape → createdAt), and a path whose file vanished (stat ENOENT
// → createdAt).
ctx.provide('sessionPersistence', {
list: () => Promise.resolve(metas),
locate: (meta: SessionHeader) => {
if (meta.id === sid('session-a')) return { kind: 'jsonl', path: logPath }
if (meta.id === sid('session-c')) return { kind: 'jsonl', path: join(root, 'vanished.log') }
return undefined
},
})
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const response = await api.sessions.list(request({}))
expect(response.result.ok).toBe(true)
if (!response.result.ok) throw new Error('unreachable')
const items = response.result.value.items
expect(items.map(item => item.sessionId)).toEqual(['session-a', 'session-b', 'session-c'])
const [a, b, c] = items
expect(a?.updatedAt).toBeCloseTo(5_000_000, -3)
expect(a?.running).toBe(false)
// Cold summaries are never blank: lazy persistence keeps never-appended
// sessions out of list(), so a listed session necessarily has events.
expect(items.every(item => !item.blank)).toBe(true)
expect(a?.cwd).toBe('/proj')
expect(a?.parentSessionId).toBeUndefined()
expect(b?.updatedAt).toBe(2000)
expect(b?.parentSessionId).toBe('session-parent')
expect(c?.updatedAt).toBe(1500)
})
})
describe('degenerate composition (no persistence, no factory)', () => {
it('list skips the cold merge and resume maps a non-not-found failure to internal', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserInteractionService)
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const listed = await api.sessions.list(request({}))
expect(listed.result.ok).toBe(true)
if (listed.result.ok) expect(listed.result.value.items).toEqual([])
// No persistence → the servable gate passes silently; the factory-less
// registry then rejects resume, which is NOT a SessionNotFound.
const response = await api.sessions.history(request({ sessionId: sid('session-ghost') }))
expect(response.result.ok).toBe(false)
if (!response.result.ok) {
expect(response.result.error.code).toBe('internal')
expect(response.result.error.message).toMatch(/resume failed for session "session-ghost"/)
}
})
})

View File

@@ -0,0 +1,314 @@
/**
* Command/skill RPC handlers and the two new frames over createApiProxy:
* command.list serves the addressed agent's effective catalog (missing
* registry = loud internal error), command.execute dispatches through the
* registry with the carrier signal, skill.list resolves cwd from the session
* header (never via the Agent registry), the host stream broadcasts
* commands-changed, and the mux stream carries live queued frames plus the
* open-time queue snapshot.
*/
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentMessage } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import type { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import CommandService from '@deepseek-ai/dsh-commands'
import SkillService from '@deepseek-ai/dsh-skill'
import type { HostFrame, MuxFrame } from '../src/api/index.ts'
import type { RpcRequest, RpcResponse } from '../src/api/rpc.ts'
import { RpcId } from '../src/api/rpc.ts'
import { createApiProxy } from '../src/api-proxy.ts'
const DEFAULTS = { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }
function request<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId(`req-${String(nextRpc++)}`), payload }
}
let nextRpc = 1
function expectOk<T>(response: RpcResponse<T>): T {
expect(response.result.ok).toBe(true)
if (!response.result.ok) throw new Error('unreachable')
return response.result.value
}
function expectErr<T>(response: RpcResponse<T>): { code: string; message: string } {
expect(response.result.ok).toBe(false)
if (response.result.ok) throw new Error('unreachable')
return response.result.error
}
/** Composition floor for the command/skill paths (no LLM, no persistence). */
async function harness(options: { commands?: boolean; skills?: boolean } = {}): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRegistry)
await ctx.plugin(UserInteractionService)
await ctx.plugin(AgentRegistry)
if (options.skills !== false) await ctx.plugin(SkillService, {})
if (options.commands !== false) await ctx.plugin(CommandService)
// Host-stream opener reads the committed-workspace baseline; the stub
// suffices here — the real workspace composition is api-proxy-workspace.spec's.
ctx.provide('workspace', { list: () => [] } as never)
return ctx
}
/** Register a live structural agent stub (api-proxy-view precedent: only id/session/status/ctx are read). */
function stubAgent(ctx: Context, sessionId?: SessionId): Agent {
const session = ctx.sessions.create(sessionId)
const agent = { id: session.id, session, status: 'idle', ctx } as Agent
ctx.agents.register(agent)
return agent
}
/** Drain `count` frames from a stream, then abort it. */
async function collect<F>(iterable: AsyncIterable<RpcRequest<F>>, count: number, abort: AbortController): Promise<F[]> {
const frames: F[] = []
for await (const frame of iterable) {
frames.push(frame.payload)
if (frames.length >= count) abort.abort()
}
return frames
}
describe('command.list', () => {
it('serves the addressed agent\'s name-sorted catalog', async () => {
const ctx = await harness()
ctx.commands.register({ name: 'zeta', description: 'z', handler: () => ({ kind: 'success' }) })
ctx.commands.register({ name: 'alpha', description: 'a', input: { hint: '<x>' }, handler: () => ({ kind: 'success' }) })
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const value = expectOk(await api.commands.list(request({ sessionId: agent.id })))
expect(value.commands).toEqual([
{ name: 'alpha', description: 'a', input: { hint: '<x>' } },
{ name: 'zeta', description: 'z' },
])
})
it('fails loud with internal when the command registry is not mounted', async () => {
const ctx = await harness({ commands: false })
const api = createApiProxy(ctx, DEFAULTS)
const error = expectErr(await api.commands.list(request({ sessionId: 's' as SessionId })))
expect(error.code).toBe('internal')
expect(error.message).toContain('command registry')
})
})
describe('command.execute', () => {
it('executes a known command against the addressed agent and detaches the result', async () => {
const ctx = await harness()
let received: string | undefined
ctx.commands.register({
name: 'goal',
description: 'set goal',
handler: (invocation) => {
received = invocation.rawInput
return { kind: 'success', text: `goal:${invocation.agent.id}` }
},
})
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const value = expectOk(await api.commands.execute(request({ sessionId: agent.id, line: '/goal ship it' }), new AbortController().signal))
expect(value).toEqual({ matched: true, result: { kind: 'success', text: `goal:${agent.id}` } })
expect(received).toBe(' ship it')
})
it('returns matched:false when syntax or name does not resolve', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const signal = new AbortController().signal
expect(expectOk(await api.commands.execute(request({ sessionId: agent.id, line: '/unknown' }), signal))).toEqual({ matched: false })
expect(expectOk(await api.commands.execute(request({ sessionId: agent.id, line: 'not a command' }), signal))).toEqual({ matched: false })
})
it('maps a session miss to session-not-found and a registry gap to internal', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const missing = expectErr(await api.commands.execute(
request({ sessionId: 'session-nope' as SessionId, line: '/x' }), new AbortController().signal))
expect(missing.code).toBe('internal') // no persistence configured: resume fails loud past the gate
const bare = await harness({ commands: false })
const bareApi = createApiProxy(bare, DEFAULTS)
expect(expectErr(await bareApi.commands.execute(
request({ sessionId: 's' as SessionId, line: '/x' }), new AbortController().signal)).code).toBe('internal')
})
it('reports an aborted handler as cancelled and a throwing handler as internal', async () => {
const ctx = await harness()
ctx.commands.register({
name: 'hang',
description: 'never settles on its own',
handler: () => new Promise(() => { /* settled only by abort */ }),
})
ctx.commands.register({
name: 'boom',
description: 'throws',
handler: () => { throw new Error('kaboom') },
})
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const controller = new AbortController()
const pending = api.commands.execute(request({ sessionId: agent.id, line: '/hang' }), controller.signal)
controller.abort()
expect(expectErr(await pending).code).toBe('cancelled')
const thrown = expectErr(await api.commands.execute(request({ sessionId: agent.id, line: '/boom' }), new AbortController().signal))
expect(thrown.code).toBe('internal')
expect(thrown.message).toContain('kaboom')
})
})
describe('skill.list', () => {
it('lists skills for the session cwd taken from the header', async () => {
const ctx = await harness()
const seenCwds: (string | undefined)[] = []
ctx.skills.registerProvider({
name: 'probe',
list: (options) => {
seenCwds.push(options.cwd)
return Promise.resolve([{
name: 'commit-helper', description: 'Git commits', whenToUse: 'when committing',
source: 'custom', provider: 'probe', rank: 0, locator: null,
}])
},
get: () => Promise.resolve(undefined),
})
const api = createApiProxy(ctx, DEFAULTS)
// No agent is registered for this session: header resolution must not
// touch (or resume through) the Agent registry.
const session = ctx.sessions.create(undefined, { meta: { cwd: '/proj' } })
const value = expectOk(await api.skills.list(request({ sessionId: session.id })))
expect(value.skills).toEqual([{ name: 'commit-helper', description: 'Git commits', whenToUse: 'when committing' }])
expect(seenCwds).toEqual(['/proj'])
expect(ctx.agents.get(session.id)).toBeUndefined()
})
it('fails loud on an unattached session id (business error, no resume attempt)', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const error = expectErr(await api.skills.list(request({ sessionId: 'session-cold' as SessionId })))
expect(error.code).toBe('session-not-found')
})
it('fails loud with internal when the skill registry is not mounted', async () => {
const ctx = await harness({ skills: false })
const api = createApiProxy(ctx, DEFAULTS)
const session = ctx.sessions.create(undefined, { meta: { cwd: '/proj' } })
const error = expectErr(await api.skills.list(request({ sessionId: session.id })))
expect(error.code).toBe('internal')
expect(error.message).toContain('skill registry is absent')
})
it('folds a provider failure into internal', async () => {
const ctx = await harness()
ctx.skills.registerProvider({
name: 'broken',
list: () => Promise.reject(new Error('directory exploded')),
get: () => Promise.resolve(undefined),
})
const api = createApiProxy(ctx, DEFAULTS)
const session = ctx.sessions.create(undefined, { meta: { cwd: '/proj' } })
const response = await api.skills.list(request({ sessionId: session.id }))
// dsh-skill contains one provider's failure (logs and serves the rest), so
// this surfaces as an empty ok catalog rather than an error.
const value = expectOk(response)
expect(value.skills).toEqual([])
})
})
describe('host/commands-changed frame', () => {
it('broadcasts on registry change', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const abort = new AbortController()
const stream = api.events.host({ rpcId: RpcId('t-host'), payload: {} }, abort.signal)
const collected = collect<HostFrame>(stream, 1, abort)
ctx.commands.register({ name: 'late', description: 'l', handler: () => ({ kind: 'success' }) })
expect(await collected).toEqual([{ type: 'host/commands-changed' }])
})
})
/** Build one frozen inbox message for the live `agent/inbox/*` events. */
function inboxMessage(id: string, text: string, steering: boolean, rpcId?: string): AgentMessage {
return Object.freeze({
id: AgentMessageId(id),
content: [{ type: 'text' as const, text }],
source: rpcId === undefined ? { kind: 'user' as const } : { kind: 'user' as const, rpcId: RpcId(rpcId) },
contexts: [],
steering,
wakeup: true,
})
}
describe('session/queued frames', () => {
it('forwards live enqueue events and replays the snapshot on a later mux open', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const live = new AbortController()
const liveStream = api.events.mux({ rpcId: RpcId('t-mux-live'), payload: {} }, live.signal)
// subscribed baseline + 2 queued frames
const liveCollected = collect<MuxFrame>(liveStream, 3, live)
const queued = inboxMessage('m-1', 'queued prompt', false)
const steering = inboxMessage('m-2', 'queued prompt', true)
ctx.emit('agent/inbox/enqueue', agent, queued)
ctx.emit('agent/inbox/enqueue', agent, steering)
const liveFrames = (await liveCollected).filter(f => f.type === 'session/queued')
expect(liveFrames).toEqual([
{ type: 'session/queued', sessionId: agent.id, content: queued.content, source: { kind: 'user' }, steering: false },
{ type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' }, steering: true },
])
// A fresh mux connection replays the still-pending entries as its baseline.
const replay = new AbortController()
const replayFrames = await collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-replay'), payload: {} }, replay.signal), 3, replay)
expect(replayFrames.filter(f => f.type === 'session/queued')).toHaveLength(2)
})
it('retires mirror entries on their terminal dequeue', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const queued = inboxMessage('m-3', 'x', false)
const steering = inboxMessage('m-4', 'x', true, 'r-1')
ctx.emit('agent/inbox/enqueue', agent, queued)
ctx.emit('agent/inbox/enqueue', agent, steering)
ctx.emit('agent/inbox/dequeue', agent, queued)
ctx.emit('agent/inbox/dequeue', agent, steering)
const abort = new AbortController()
const frames = await collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-after'), payload: {} }, abort.signal), 1, abort)
expect(frames.filter(f => f.type === 'session/queued')).toHaveLength(0)
})
it('retires mirror entries on a batch discard (cancel path)', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const doomed = inboxMessage('m-5', 'doomed', false)
const survivor = inboxMessage('m-6', 'survivor', false)
ctx.emit('agent/inbox/enqueue', agent, doomed)
ctx.emit('agent/inbox/enqueue', agent, survivor)
ctx.emit('agent/inbox/discard', agent, [doomed])
const abort = new AbortController()
const frames = await collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-swept'), payload: {} }, abort.signal), 2, abort)
const remaining = frames.filter(f => f.type === 'session/queued')
expect(remaining).toHaveLength(1)
expect(remaining[0]).toMatchObject({ content: survivor.content })
})
})

View File

@@ -0,0 +1,197 @@
/**
* Tool-card view computation over the mux live path: three standard card types
* arrive on the frame, a presenterless tool ships no view field, a call-only
* presenter keeps raw result content out of the view payload, and a throwing
* presenter soft-falls to no view (the event still ships). Result pairing
* works both through the live open-call table and the backscan fallback after
* turn/end cleared it.
*/
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { ToolDefinition } from '@deepseek-ai/dsh-tools'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
const reply = (text: string): Promise<ContentBlock[]> => Promise.resolve([{ type: 'text', text }])
function tool(name: string, presenters: Pick<ToolDefinition, 'presentCall' | 'presentResult'>): ToolDefinition {
return defineContentToolFixture({
name,
description: `tool ${name}`,
parameters: {},
execute: () => reply(`ran:${name}`),
...presenters,
})
}
async function harness(): Promise<{ ctx: Context }> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRegistry)
await ctx.plugin(UserInteractionService)
await ctx.plugin(AgentRegistry)
ctx.tools.register(tool('gen', {
presentCall: () => ({ card: 'generic', title: 'gen call' }),
presentResult: (_args, result) => ({ card: 'generic', title: result.isError ? 'gen failed' : 'gen done' }),
}))
ctx.tools.register(tool('term', {
presentCall: args => ({ card: 'terminal', title: (args as { cmd?: string }).cmd ?? '' }),
presentResult: () => ({ card: 'terminal', output: 'done' }),
}))
ctx.tools.register(tool('diffy', {
presentCall: () => ({ card: 'diff', title: 'Write f.txt', diffs: [{ path: 'f.txt', oldText: null, newText: 'x' }] }),
}))
ctx.tools.register(tool('call-only', {
presentCall: () => ({ card: 'generic', title: 'program', kind: 'execute', rawInput: 'return value' }),
}))
ctx.tools.register(tool('plain', {}))
ctx.tools.register(tool('boom', {
presentCall: () => { throw new Error('presenter exploded') },
}))
return { ctx }
}
/** Drain frames from an open mux stream until `count` session/event frames arrived. */
async function collect(iterable: AsyncIterable<RpcRequest<MuxFrame>>, count: number, abort: AbortController): Promise<MuxFrame[]> {
const frames: MuxFrame[] = []
for await (const frame of iterable) {
frames.push(frame.payload)
if (frames.filter(f => f.type === 'session/event').length >= count) abort.abort()
}
return frames
}
describe('mux live view computation', () => {
it('attaches the three standard card views, omits view without a presenter, soft-falls on throw', async () => {
const { ctx } = await harness()
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const abort = new AbortController()
const stream = api.events.mux({ rpcId: RpcId('t-mux'), payload: {} }, abort.signal)
const collected = collect(stream, 9, abort)
const rawResult = `RAW_RESULT:${'x'.repeat(64 * 1024)}`
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-gen'), name: 'gen', arguments: '{}' })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-term'), name: 'term', arguments: '{"cmd":"echo hi"}' })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-diff'), name: 'diffy', arguments: '{}' })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-call-only'), name: 'call-only', arguments: '{}' })
session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-call-only'), content: [{ type: 'text', text: rawResult }], isError: false }, { surfaceOp: 'append' })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-plain'), name: 'plain', arguments: '{}' })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-boom'), name: 'boom', arguments: '{}' })
session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-gen'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' })
const frames = await collected
const events = frames.filter(f => f.type === 'session/event')
const byCall = new Map(events
.filter(f => f.event.type === 'tool/call' || f.event.type === 'tool/result')
.map(f => [`${f.event.type}:${(f.event.data as { callId: string }).callId}`, f]))
expect(byCall.get('tool/call:c-gen')?.view).toEqual({ for: 'call', view: { card: 'generic', title: 'gen call' } })
expect(byCall.get('tool/call:c-term')?.view).toEqual({ for: 'call', view: { card: 'terminal', title: 'echo hi' } })
expect(byCall.get('tool/call:c-diff')?.view?.view.card).toBe('diff')
expect(byCall.get('tool/call:c-call-only')?.view).toEqual({
for: 'call',
view: { card: 'generic', title: 'program', kind: 'execute', rawInput: 'return value' },
})
const callOnlyResult = byCall.get('tool/result:c-call-only')
expect('view' in (callOnlyResult ?? {})).toBe(false)
const serializedResult = JSON.stringify(callOnlyResult)
expect(serializedResult.indexOf(rawResult)).toBeGreaterThanOrEqual(0)
expect(serializedResult.indexOf(rawResult)).toBe(serializedResult.lastIndexOf(rawResult))
// No presenter → the frame carries no view property at all.
expect('view' in (byCall.get('tool/call:c-plain') ?? {})).toBe(false)
// Throwing presenter → soft-fall: event ships, no view.
expect(byCall.get('tool/call:c-boom')).toBeDefined()
expect('view' in (byCall.get('tool/call:c-boom') ?? {})).toBe(false)
// Result pairing through the live table: presentResult saw the call's args.
expect(byCall.get('tool/result:c-gen')?.view).toEqual({ for: 'result', view: { card: 'generic', title: 'gen done' } })
})
it('serves history entries with call/result views, backscan pairing, and soft-falls', async () => {
const { ctx } = await harness()
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const session = ctx.sessions.create()
// history resolves the agent first; a live structural stub is enough (only
// .session is read on this path).
ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent)
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-term'), name: 'term', arguments: '{"cmd":"ls"}' })
// meta rides through to presentResult's ToolResult (the spread arm).
session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-term'), content: [{ type: 'text', text: 'ok' }], isError: false, meta: { n: 1 } }, { surfaceOp: 'append' })
// Unpaired result: no tool/call with this id anywhere in the page.
session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-orphan'), content: [{ type: 'text', text: 'x' }], isError: false }, { surfaceOp: 'append' })
// Paired, but the call's stored arguments do not parse: backscan soft-falls.
session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-bad'), name: 'term', arguments: '{broken' })
session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-bad'), content: [{ type: 'text', text: 'y' }], isError: false }, { surfaceOp: 'append' })
// Presenterless tool: pairing succeeds but presentResult is absent.
session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-plain'), name: 'plain', arguments: '{}' })
session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-plain'), content: [{ type: 'text', text: 'z' }], isError: false }, { surfaceOp: 'append' })
const response = await api.sessions.history({ rpcId: RpcId('t-hist'), payload: { sessionId: session.id } })
expect(response.result.ok).toBe(true)
if (!response.result.ok) throw new Error('unreachable')
const entries = response.result.value.events
const byKey = new Map(entries
.filter(entry => entry.event.type === 'tool/call' || entry.event.type === 'tool/result')
.map(entry => [`${entry.event.type}:${(entry.event.data as { callId: string }).callId}`, entry]))
expect(byKey.get('tool/call:h-term')?.view).toEqual({ for: 'call', view: { card: 'terminal', title: 'ls' } })
expect(byKey.get('tool/result:h-term')?.view).toEqual({ for: 'result', view: { card: 'terminal', output: 'done' } })
expect('view' in (byKey.get('tool/result:h-orphan') ?? {})).toBe(false)
expect('view' in (byKey.get('tool/result:h-bad') ?? {})).toBe(false)
expect('view' in (byKey.get('tool/result:h-plain') ?? {})).toBe(false)
})
it('drops a disposed session from the live open-call table (result after dispose gets no view)', async () => {
const { ctx } = await harness()
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const abort = new AbortController()
const stream = api.events.mux({ rpcId: RpcId('t-mux3'), payload: {} }, abort.signal)
let session: Session | undefined
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create('session-doomed' as SessionId)
}, { inject: ['sessions'] }))
session?.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session?.append('tool/call', { turn: 1, step: 1, callId: CallId('c-doomed'), name: 'term', arguments: '{"cmd":"x"}' })
// Disposing the owning fiber detaches the session mid-stream; the
// session/disposed listener must clear its open-call table entry.
await fiber.dispose()
const frames = await collect(stream, 2, abort)
const call = frames.find(f => f.type === 'session/event' && f.event.type === 'tool/call')
expect(call?.type === 'session/event' && call.view?.for).toBe('call')
})
it('pairs a result after turn/end via the in-memory backscan fallback', async () => {
const { ctx } = await harness()
const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
const abort = new AbortController()
const stream = api.events.mux({ rpcId: RpcId('t-mux2'), payload: {} }, abort.signal)
const collected = collect(stream, 4, abort)
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-late'), name: 'term', arguments: '{"cmd":"tail"}' })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
// The turn/end above cleared the live table; pairing must fall back to
// scanning the session's in-memory events.
session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-late'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' })
const frames = await collected
const result = frames.find(f => f.type === 'session/event' && f.event.type === 'tool/result')
expect(result?.type === 'session/event' && result.view).toEqual({ for: 'result', view: { card: 'terminal', output: 'done' } })
})
})

View File

@@ -0,0 +1,247 @@
import { existsSync, mkdirSync, mkdtempSync, realpathSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentFactory } from '@deepseek-ai/dsh-agent'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import Storage from '@deepseek-ai/dsh-storage'
import { DomainFacility } from '@deepseek-ai/dsh-storage-domain'
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
import WorkspaceRegistry from '@deepseek-ai/dsh-workspace'
import type { HostFrame, WorkspaceId } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
import { MemoryStorageBackend } from '../../../storage/storage-domain/tests/helpers/memory-backend.ts'
let nextRpc = 1
function request<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId(`workspace-${String(nextRpc++)}`), payload }
}
function expectOk<T>(response: RpcResponse<T>): T {
expect(response.result.ok).toBe(true)
if (!response.result.ok) throw new Error('unreachable')
return response.result.value
}
async function nextHostFrame(
stream: AsyncIterator<RpcRequest<HostFrame>>,
): Promise<RpcRequest<HostFrame>> {
const next = await stream.next()
if (next.done === true) throw new Error('Host stream ended before the expected increment')
return next.value
}
function stubAgent(session: Session): Agent {
return {
id: session.id,
options: {},
session,
status: 'idle',
ctx: new Context(),
followup: () => AgentMessageId('stub'),
queue: () => AgentMessageId('stub'),
steer: () => AgentMessageId('stub'),
inject: () => AgentMessageId('stub'),
send: () => AgentMessageId('stub'),
cancel() {},
whenIdle: () => Promise.resolve(),
}
}
/** Compose the API over real Session, Agent, Storage, Domain, and Workspace services. */
async function harness(
workspaceRoot = realpathSync(mkdtempSync(join(tmpdir(), 'dsh-apiproxy-workspace-'))),
) {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserInteractionService)
await ctx.plugin(Storage)
ctx.storage.backend.register('memory', new MemoryStorageBackend())
const storageDomain = new DomainFacility(ctx, { backend: 'memory', routes: {} })
ctx.storage.mount('domain', storageDomain)
ctx.provide('storageDomain', storageDomain)
ctx.provide('sessionPersistence', { list: () => Promise.resolve([]) } as never)
await ctx.plugin(WorkspaceRegistry)
const factory: AgentFactory = {
async createAgent(_ownerCtx, options) {
const session = ctx.sessions.create(
options.sessionId,
options.meta === undefined ? {} : { meta: options.meta },
)
const agent = stubAgent(session)
const unregister = ctx.agents.register(agent)
return {
agent,
dispose: () => {
unregister()
return Promise.resolve()
},
}
},
async resume() {
throw new Error('test harness has no persisted sessions')
},
}
ctx.agents.setFactory(factory)
const api = createApiProxy(ctx, {
provider: 'test',
model: 'test-model',
cwd: workspaceRoot,
workspaceRoot,
})
return { api, ctx, storageDomain, workspaceRoot }
}
describe('workspace.create', () => {
it('serializes concurrent names and rejects the duplicate', async () => {
const { api, workspaceRoot } = await harness()
const responses = await Promise.all([
api.workspace.create(request({ name: 'alpha' })),
api.workspace.create(request({ name: 'alpha' })),
])
const created = responses.find(response => response.result.ok)
const duplicate = responses.find(response => !response.result.ok)
expect(created).toBeDefined()
expect(expectOk(created!)).toMatchObject({
created: true,
workspace: { path: join(workspaceRoot, 'alpha'), title: 'alpha' },
})
expect(duplicate?.result).toMatchObject({
ok: false,
error: { code: 'workspace-name-conflict', details: { name: 'alpha' } },
})
expect(existsSync(join(workspaceRoot, 'alpha'))).toBe(true)
})
it('adopts only existing directories and rejects unsafe names', async () => {
const { api, workspaceRoot } = await harness()
const existing = join(workspaceRoot, 'existing')
mkdirSync(existing)
const first = expectOk(await api.workspace.create(request({ path: existing })))
const repeated = expectOk(await api.workspace.create(request({ path: existing })))
expect(first).toMatchObject({ created: true, workspace: { path: existing, title: 'existing' } })
expect(repeated).toMatchObject({ created: false, workspace: { workspaceId: first.workspace.workspaceId } })
const missing = join(workspaceRoot, 'missing')
const missingResult = await api.workspace.create(request({ path: missing }))
expect(missingResult.result).toMatchObject({ ok: false, error: { code: 'workspace-invalid-path' } })
expect(existsSync(missing)).toBe(false)
for (const name of ['', '.', '..', 'a/b', 'a\\b']) {
const invalid = await api.workspace.create(request({ name }))
expect(invalid.result).toMatchObject({ ok: false, error: { code: 'workspace-invalid-path' } })
}
})
})
describe('session creation and Workspace membership', () => {
it('attaches a preallocated idempotent session while cwd-only sessions stay ungrouped', async () => {
const { api, ctx } = await harness()
const workspace = expectOk(await api.workspace.create(request({ name: 'project' }))).workspace
const sessionId = SessionId('session-workspace-preallocated')
expectOk(await api.sessions.create(request({ workspaceId: workspace.workspaceId, sessionId })))
expectOk(await api.sessions.create(request({ workspaceId: workspace.workspaceId, sessionId })))
expect(expectOk(await api.workspace.list(request({}))).items[0]?.sessionIds).toEqual([sessionId])
expect(ctx.agents.list().filter(agent => agent.id === sessionId)).toHaveLength(1)
const ungrouped = SessionId('session-cwd-only')
expectOk(await api.sessions.create(request({ cwd: workspace.path, sessionId: ungrouped })))
expect(expectOk(await api.workspace.list(request({}))).items[0]?.sessionIds).toEqual([sessionId])
expect(expectOk(await api.sessions.list(request({}))).items.map(item => item.sessionId)).toContain(ungrouped)
const conflict = await api.sessions.create(request({ cwd: join(workspace.path, 'other'), sessionId }))
expect(conflict.result).toMatchObject({
ok: false,
error: { code: 'session-conflict', details: { sessionId, existingCwd: workspace.path } },
})
const missing = await api.sessions.create(request({
workspaceId: 'missing-workspace' as WorkspaceId,
sessionId: SessionId('session-missing-workspace'),
}))
expect(missing.result).toMatchObject({ ok: false, error: { code: 'workspace-not-found' } })
})
it('retains a published session when attachment fails and repairs it on retry', async () => {
const { api, ctx } = await harness()
const created = expectOk(await api.workspace.create(request({ name: 'project' }))).workspace
const workspace = ctx.workspace.list()[0]
if (workspace === undefined) throw new Error('workspace missing from registry')
vi.spyOn(workspace, 'attachSession').mockRejectedValueOnce(new Error('simulated write failure'))
const sessionId = SessionId('session-attach-retry')
const failed = await api.sessions.create(request({ workspaceId: created.workspaceId, sessionId }))
expect(failed.result).toMatchObject({
ok: false,
error: { code: 'workspace-attach-failed', details: { sessionId, workspaceId: created.workspaceId } },
})
expect(ctx.agents.get(sessionId)).toBeDefined()
expectOk(await api.sessions.create(request({ workspaceId: created.workspaceId, sessionId })))
expect(expectOk(await api.workspace.list(request({}))).items[0]?.sessionIds).toEqual([sessionId])
})
})
describe('Host Workspace increments', () => {
it('streams committed Workspace and Session increments after empty baselines', async () => {
const { api } = await harness()
expect(expectOk(await api.workspace.list(request({}))).items).toEqual([])
expect(expectOk(await api.sessions.list(request({}))).items).toEqual([])
const abort = new AbortController()
const stream: AsyncIterator<RpcRequest<HostFrame>> =
api.events.host(request({}), abort.signal)[Symbol.asyncIterator]()
const workspaceIncrement = nextHostFrame(stream)
const workspace = expectOk(await api.workspace.create(request({ name: 'project' }))).workspace
expect(await workspaceIncrement).toMatchObject({
payload: { type: 'host/workspace-changed', workspace: { workspaceId: workspace.workspaceId } },
})
const sessionId = SessionId('session-streamed-workspace')
const pending = nextHostFrame(stream)
expectOk(await api.sessions.create(request({ workspaceId: workspace.workspaceId, sessionId })))
const increments: HostFrame[] = []
increments.push((await pending).payload)
while (increments.length < 2) {
const next = await stream.next()
if (next.done === true) throw new Error('Host stream ended before both increments')
increments.push(next.value.payload)
}
expect(increments.find(increment => increment.type === 'host/session-added')).toMatchObject({
// A just-created session has no events: the frame constantly carries blank:true.
type: 'host/session-added', sessionId, blank: true, cwd: workspace.path,
})
const workspaceChanged = increments.find(
(increment): increment is Extract<HostFrame, { type: 'host/workspace-changed' }> =>
increment.type === 'host/workspace-changed',
)
expect(workspaceChanged?.workspace.sessionIds).toEqual([sessionId])
abort.abort()
})
it('does not publish a Workspace whose registry-order commit fails', async () => {
const { api, storageDomain } = await harness()
const domain = storageDomain.get('workspace')
if (domain === undefined) throw new Error('workspace domain is not open')
vi.spyOn(domain.global, 'set').mockRejectedValueOnce(new Error('simulated registry order failure'))
const abort = new AbortController()
const stream: AsyncIterator<RpcRequest<HostFrame>> =
api.events.host(request({}), abort.signal)[Symbol.asyncIterator]()
const next = stream.next()
const failed = await api.workspace.create(request({ name: 'ghost' }))
expect(failed.result.ok).toBe(false)
expect(expectOk(await api.workspace.list(request({}))).items).toEqual([])
abort.abort()
expect(await next).toMatchObject({ done: true })
})
})

View File

@@ -20,6 +20,8 @@ function ok<T>(request: RpcRequest<unknown>, value: T): Promise<RpcResponse<T>>
function scriptedApi(overrides: {
sessions?: Partial<ApiProxy['sessions']>
host?: Partial<ApiProxy['host']>
commands?: Partial<ApiProxy['commands']>
skills?: Partial<ApiProxy['skills']>
events?: Partial<ApiProxy['events']>
respond?: ApiProxy['respond']
} = {}): ApiProxy {
@@ -46,6 +48,18 @@ function scriptedApi(overrides: {
...overrides.sessions,
},
host: { describe: r => ok(r, { version: '0-test', cwd: '/t', attachedSessions: 0 }), ...overrides.host },
workspace: {
list: r => ok(r, { items: [] }),
create: r => ok(r, { workspace: { workspaceId: 'w1' as never, path: '/t', title: 't', sessionIds: [], createdAt: '0', updatedAt: '0' }, created: true }),
rename: r => ok(r, { workspace: { workspaceId: 'w1' as never, path: '/t', title: 't', sessionIds: [], createdAt: '0', updatedAt: '0' } }),
insertSessionBefore: r => ok(r, { workspace: { workspaceId: 'w1' as never, path: '/t', title: 't', sessionIds: [], createdAt: '0', updatedAt: '0' } }),
},
commands: {
list: r => ok(r, { commands: [] }),
execute: r => ok(r, { matched: false }),
...overrides.commands,
},
skills: { list: r => ok(r, { skills: [] }), ...overrides.skills },
events: { mux: () => empty<MuxFrame>(), host: () => empty<HostFrame>(), ...overrides.events },
respond: overrides.respond ?? (() => Promise.resolve({ accepted: false as const, reason: 'not-pending' as const })),
}
@@ -62,7 +76,7 @@ describe('unary round trip', () => {
sessions: {
list: (r) => {
seen = r
return ok(r, { items: [{ sessionId: sid('s1'), updatedAt: 7, running: false }] })
return ok(r, { items: [{ sessionId: sid('s1'), updatedAt: 7, running: false, blank: false }] })
},
},
})
@@ -71,7 +85,20 @@ describe('unary round trip', () => {
expect(seen?.payload).toEqual({ cursor: 'c1' })
expect(seen?.rpcId).toBeTruthy()
expect(response.rpcId).toBe(seen?.rpcId)
expect(response.result).toEqual({ ok: true, value: { items: [{ sessionId: 's1', updatedAt: 7, running: false }] } })
expect(response.result).toEqual({ ok: true, value: { items: [{ sessionId: 's1', updatedAt: 7, running: false, blank: false }] } })
})
it('routes workspace rename and insertSessionBefore through the wire', async () => {
const api = scriptedApi()
const c = client(api)
const renamed = await c.workspace.rename({ workspaceId: 'w1' as never, title: 'next' })
expect(renamed.result.ok).toBe(true)
const blankTitle = await c.workspace.rename({ workspaceId: 'w1' as never, title: ' ' })
expect(blankTitle.result).toMatchObject({ ok: false, error: { code: 'bad-request' } })
const anchored = await c.workspace.insertSessionBefore({ workspaceId: 'w1' as never, sessionId: sid('s1'), beforeSessionId: sid('s2') })
expect(anchored.result.ok).toBe(true)
const appended = await c.workspace.insertSessionBefore({ workspaceId: 'w1' as never, sessionId: sid('s1') })
expect(appended.result.ok).toBe(true)
})
it('passes business errors through as 200 + err result, not a throw', async () => {
@@ -202,6 +229,23 @@ describe('unary round trip', () => {
})
})
describe('workspace domain round trip', () => {
it('routes both workspace methods through their handler rows and value schemas', async () => {
const c = client(scriptedApi())
const list = await c.workspace.list({})
expect(list.result).toEqual({ ok: true, value: { items: [] } })
const created = await c.workspace.create({ path: '/t' })
expect(created.result.ok).toBe(true)
if (created.result.ok) expect(created.result.value.created).toBe(true)
})
it('rejects a create payload violating the exactly-one refine at the handler', async () => {
const response = await client(scriptedApi()).workspace.create({})
expect(response.result.ok).toBe(false)
if (!response.result.ok) expect(response.result.error.code).toBe('bad-request')
})
})
describe('SSE stream path', () => {
it('yields frames in order and skips the comment preamble', async () => {
const frames: MuxFrame[] = [
@@ -253,7 +297,7 @@ describe('SSE stream path', () => {
const api = scriptedApi({
events: {
async *host(request): AsyncGenerator<RpcRequest<HostFrame>> {
yield { rpcId: RpcId(`p-${request.rpcId}`), payload: { type: 'host/session-added', sessionId: sid('s1') } }
yield { rpcId: RpcId(`p-${request.rpcId}`), payload: { type: 'host/session-added', sessionId: sid('s1'), blank: true } }
throw new Error('impl died mid-stream')
},
},

View File

@@ -64,6 +64,53 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
return { rpcId: request.rpcId, result: { ok: true, value: { version: 'v', cwd: '/w', attachedSessions: 0 } } }
},
},
workspace: {
async list(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { items: [] } } }
},
async create(request) {
return {
rpcId: request.rpcId,
result: { ok: true, value: { workspace: { workspaceId: 'w1' as never, path: '/w', title: 'w', sessionIds: [], createdAt: 't', updatedAt: 't' }, created: true } },
}
},
async rename(request) {
return {
rpcId: request.rpcId,
result: { ok: true, value: { workspace: { workspaceId: 'w1' as never, path: '/w', title: 'w', sessionIds: [], createdAt: 't', updatedAt: 't' } } },
}
},
async insertSessionBefore(request) {
return {
rpcId: request.rpcId,
result: { ok: true, value: { workspace: { workspaceId: 'w1' as never, path: '/w', title: 'w', sessionIds: [], createdAt: 't', updatedAt: 't' } } },
}
},
},
commands: {
async list(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { commands: [{ name: 'plan', description: 'Toggle plan mode', input: { hint: 'on|off' } }] } } }
},
async execute(request, signal) {
if (request.payload.line === '/hang') {
// Cooperative hang: settles only through the carrier signal (sticky
// abort checked first — listeners never fire retroactively).
if (!signal.aborted) {
await new Promise<void>((resolve) => { signal.addEventListener('abort', () => { resolve() }, { once: true }) })
}
return { rpcId: request.rpcId, result: { ok: false, error: { code: 'cancelled', message: 'aborted', details: {} } } }
}
if (request.payload.line.startsWith('/plan')) {
return { rpcId: request.rpcId, result: { ok: true, value: { matched: true, result: { kind: 'success' as const, text: 'plan set' } } } }
}
return { rpcId: request.rpcId, result: { ok: true, value: { matched: false } } }
},
},
skills: {
async list(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { skills: [{ name: 'commit-helper', description: 'Git commits' }] } } }
},
},
events: {
mux: (_request, signal) => stream(muxFrames, signal),
host: (_request, signal) => stream(hostFrames, signal),
@@ -110,6 +157,32 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
expect((await c.sessions.cancel({ sessionId: 's' as never })).result.ok).toBe(true)
expect((await c.host.describe({})).result.ok).toBe(true)
})
it('round-trips command.list / command.execute / skill.list through the wire form', async () => {
const c = client()
const list = await c.commands.list({ sessionId: 's' as never })
expect(list.result).toEqual({ ok: true, value: { commands: [{ name: 'plan', description: 'Toggle plan mode', input: { hint: 'on|off' } }] } })
const hit = await c.commands.execute({ sessionId: 's' as never, line: '/plan off' })
expect(hit.result).toEqual({ ok: true, value: { matched: true, result: { kind: 'success', text: 'plan set' } } })
const miss = await c.commands.execute({ sessionId: 's' as never, line: '/nope' })
expect(miss.result).toEqual({ ok: true, value: { matched: false } })
const skills = await c.skills.list({ sessionId: 's' as never })
expect(skills.result).toEqual({ ok: true, value: { skills: [{ name: 'commit-helper', description: 'Git commits' }] } })
})
it('propagates the carrier Request signal into command.execute', async () => {
const handler = toFetchHandler(fakeApi())
const controller = new AbortController()
const body = JSON.stringify({ type: 'client-request', rpcId: 'r-sig', method: 'command.execute', payload: { sessionId: 's', line: '/hang' } })
// The fake's /hang settles only when the invoke-level signal aborts: a
// completed response with the cancelled error proves req.signal reached it.
const pending = handler.fetch(new Request('http://x/api/command.execute', { method: 'POST', body, signal: controller.signal }))
controller.abort()
const response = await pending
const parsed = await response.json() as { rpcId: string; result: { ok: boolean; error?: { code: string } } }
expect(parsed.rpcId).toBe('r-sig')
expect(parsed.result.error?.code).toBe('cancelled')
})
})
describe('handler carrier-layer statuses', () => {

View File

@@ -1,5 +1,5 @@
import { describe, expect, it } from 'vitest'
import { RpcId } from '../src/api/rpc.ts'
import { RpcId, transportError } from '../src/api/rpc.ts'
import {
clientRequestSchema, clientResponseSchema, rpcErrorSchema, rpcIdSchema, rpcMessageSchema,
rpcReceiptSchema, rpcResultSchema, serverRequestSchema, serverResponseSchema,
@@ -8,11 +8,21 @@ import { z } from 'zod'
import {
contentBlockSchema, sessionCancelRequestSchema, sessionCancelValueSchema, sessionCreateRequestSchema,
sessionCreateValueSchema, sessionEventSchema, sessionHistoryRequestSchema, sessionHistoryValueSchema,
sessionIdSchema, sessionListRequestSchema, sessionListValueSchema, sessionModelsRequestSchema,
sessionModelsValueSchema, sessionPromptRequestSchema, sessionPromptValueSchema,
sessionSelectModelRequestSchema, sessionSelectModelValueSchema, sessionSummarySchema,
sessionIdSchema, sessionListRequestSchema, sessionListValueSchema, sessionPromptRequestSchema,
sessionPromptValueSchema, sessionSummarySchema,
} from '../src/api/sessions.schema.ts'
import { hostDescribeRequestSchema, hostDescribeValueSchema } from '../src/api/host.schema.ts'
import {
workspaceCreateRequestSchema, workspaceCreateValueSchema, workspaceIdSchema,
workspaceInsertSessionBeforeRequestSchema, workspaceInsertSessionBeforeValueSchema,
workspaceListRequestSchema, workspaceListValueSchema,
workspaceRenameRequestSchema, workspaceRenameValueSchema, workspaceViewSchema,
} from '../src/api/workspace.schema.ts'
import {
commandDescriptorSchema, commandExecuteRequestSchema, commandExecuteValueSchema,
commandListRequestSchema, commandListValueSchema,
} from '../src/api/commands.schema.ts'
import { skillEntrySchema, skillListRequestSchema, skillListValueSchema } from '../src/api/skills.schema.ts'
import { hostFrameSchema, muxFrameSchema, askUserQuestionItemSchema } from '../src/api/events.schema.ts'
import { approvalRequestIdSchema, approvalResponsePayloadSchema } from '../src/api/approvals.schema.ts'
import { askUserQuestionAnswerSchema, questionResponsePayloadSchema } from '../src/api/questions.schema.ts'
@@ -27,16 +37,24 @@ describe('RpcId', () => {
})
})
describe('transportError', () => {
it('folds Error and non-Error throws into the internal error branch', () => {
expect(transportError(new Error('wire down'))).toEqual({ ok: false, error: { code: 'internal', message: 'wire down', details: {} } })
expect(transportError('raw')).toMatchObject({ ok: false, error: { code: 'internal', message: 'raw' } })
})
})
describe('rpcErrorSchema', () => {
it('accepts every code branch with its required details', () => {
expect(rpcErrorSchema.parse({ code: 'bad-request', message: 'm', details: { issues: [] } }).code).toBe('bad-request')
expect(rpcErrorSchema.parse({ code: 'cancelled', message: 'm', details: {} }).code).toBe('cancelled')
expect(rpcErrorSchema.parse({ code: 'session-not-found', message: 'm', details: { sessionId: 's' } }).code).toBe('session-not-found')
expect(rpcErrorSchema.parse({
code: 'model-unavailable',
message: 'm',
details: { provider: 'p', model: 'm' },
}).code).toBe('model-unavailable')
expect(rpcErrorSchema.parse({ code: 'session-conflict', message: 'm', details: { sessionId: 's', requestedCwd: '/a', existingCwd: '/b' } }).code).toBe('session-conflict')
expect(rpcErrorSchema.parse({ code: 'workspace-attach-failed', message: 'm', details: { sessionId: 's', workspaceId: 'w' } }).code).toBe('workspace-attach-failed')
expect(rpcErrorSchema.parse({ code: 'workspace-not-found', message: 'm', details: { workspaceId: 'w' } }).code).toBe('workspace-not-found')
expect(rpcErrorSchema.parse({ code: 'workspace-invalid-path', message: 'm', details: { path: '/x' } }).code).toBe('workspace-invalid-path')
expect(rpcErrorSchema.parse({ code: 'workspace-name-conflict', message: 'm', details: { name: 'x' } }).code).toBe('workspace-name-conflict')
expect(rpcErrorSchema.parse({ code: 'workspace-move-invalid', message: 'm', details: { workspaceId: 'w', sessionId: 's' } }).code).toBe('workspace-move-invalid')
expect(rpcErrorSchema.parse({ code: 'agent-busy', message: 'm', details: { reason: 'r' } }).code).toBe('agent-busy')
expect(rpcErrorSchema.parse({ code: 'internal', message: 'm', details: {} }).code).toBe('internal')
})
@@ -90,8 +108,10 @@ describe('sessions domain schemas', () => {
it('validates ids, summaries, and the event passthrough envelope', () => {
expect(sessionIdSchema.parse('s1')).toBe('s1')
expect(() => sessionIdSchema.parse('')).toThrow()
expect(sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: false })).toMatchObject({ sessionId: 's1' })
expect(sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: true, parentSessionId: 'p', cwd: '/x' }).cwd).toBe('/x')
expect(sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: false, blank: true })).toMatchObject({ sessionId: 's1', blank: true })
expect(sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: true, blank: false, parentSessionId: 'p', cwd: '/x' }).cwd).toBe('/x')
// blank is mandatory: a summary without it fails the parse.
expect(() => sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: false })).toThrow()
const event = sessionEventSchema.parse({ type: 'user/message', seq: 0, time: 1, data: { any: true } })
expect(event).toMatchObject({ type: 'user/message' })
expect(() => sessionEventSchema.parse({ type: 'user/message', seq: -1, time: 1, data: {} })).toThrow()
@@ -102,42 +122,13 @@ describe('sessions domain schemas', () => {
expect(sessionListRequestSchema.parse({ cursor: 'c' }).cursor).toBe('c')
expect(sessionListValueSchema.parse({ items: [] }).items).toEqual([])
expect(sessionCreateRequestSchema.parse({ cwd: '/w' }).cwd).toBe('/w')
// The refine's both-sides branch: workspaceId alone passes, workspaceId+cwd rejects.
expect(sessionCreateRequestSchema.parse({ workspaceId: 'w1', sessionId: 's1' }).sessionId).toBe('s1')
expect(() => sessionCreateRequestSchema.parse({ workspaceId: 'w1', cwd: '/w' })).toThrow(/not both/)
expect(sessionCreateValueSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
expect(sessionHistoryRequestSchema.parse({ sessionId: 's1', beforeSeq: 3, maxMessages: 5 }).beforeSeq).toBe(3)
expect(() => sessionHistoryRequestSchema.parse({ sessionId: 's1', maxMessages: 0 })).toThrow()
expect(sessionHistoryValueSchema.parse({
events: [],
hasMore: false,
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
}).hasMore).toBe(false)
expect(sessionModelsRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
expect(sessionModelsValueSchema.parse({
current: { provider: 'deepseek', model: 'deepseek-v4-flash' },
groups: [{
id: 'deepseek',
name: 'DeepSeek',
models: [{
id: 'deepseek-v4-flash',
name: 'DeepSeek V4 Flash',
description: 'fast',
unlisted: true,
}],
}],
failures: [{ id: 'broken', name: 'Broken', message: 'offline' }],
}).groups[0]?.models[0]?.id).toBe('deepseek-v4-flash')
expect(sessionSelectModelRequestSchema.parse({
sessionId: 's1',
provider: 'deepseek',
model: 'deepseek-v4-pro',
}).model).toBe('deepseek-v4-pro')
expect(sessionSelectModelValueSchema.parse({
selected: { provider: 'deepseek', model: 'deepseek-v4-pro' },
}).selected.model).toBe('deepseek-v4-pro')
expect(() => sessionSelectModelRequestSchema.parse({
sessionId: 's1',
provider: '',
model: 'm',
})).toThrow()
expect(sessionHistoryValueSchema.parse({ events: [], hasMore: false }).hasMore).toBe(false)
const prompt = sessionPromptRequestSchema.parse({ sessionId: 's1', mode: 'queue', content: [{ type: 'text', text: 'hi' }] })
expect(prompt.mode).toBe('queue')
expect(() => sessionPromptRequestSchema.parse({ sessionId: 's1', mode: 'inject', content: [] })).toThrow()
@@ -157,6 +148,88 @@ describe('host domain schemas', () => {
})
})
describe('workspace domain schemas', () => {
const view = {
workspaceId: 'w1', path: '/p', title: 'p', sessionIds: ['s1'],
createdAt: '2026-07-25T00:00:00.000Z', updatedAt: '2026-07-25T00:00:00.000Z',
}
it('validates ids, the view row, and list request/value', () => {
expect(workspaceIdSchema.parse('w1')).toBe('w1')
expect(() => workspaceIdSchema.parse('')).toThrow()
expect(workspaceViewSchema.parse(view).sessionIds).toEqual(['s1'])
expect(() => workspaceViewSchema.parse({ ...view, sessionIds: 's1' })).toThrow()
expect(workspaceListRequestSchema.parse({})).toEqual({})
expect(workspaceListValueSchema.parse({ items: [view] }).items).toHaveLength(1)
})
it('create requires exactly one of path/name (both refine arms)', () => {
expect(workspaceCreateRequestSchema.parse({ path: '/p' }).path).toBe('/p')
expect(workspaceCreateRequestSchema.parse({ name: 'n' }).name).toBe('n')
expect(() => workspaceCreateRequestSchema.parse({})).toThrow(/exactly one/)
expect(() => workspaceCreateRequestSchema.parse({ path: '/p', name: 'n' })).toThrow(/exactly one/)
expect(workspaceCreateValueSchema.parse({ workspace: view, created: false }).created).toBe(false)
})
it('rename requires a non-blank title (both refine arms)', () => {
expect(workspaceRenameRequestSchema.parse({ workspaceId: 'w1', title: 'new' }).title).toBe('new')
expect(() => workspaceRenameRequestSchema.parse({ workspaceId: 'w1', title: ' ' })).toThrow(/non-blank/)
expect(workspaceRenameValueSchema.parse({ workspace: view }).workspace.workspaceId).toBe('w1')
})
it('insertSessionBefore accepts an anchored and an anchorless move', () => {
expect(workspaceInsertSessionBeforeRequestSchema.parse({ workspaceId: 'w1', sessionId: 's1', beforeSessionId: 's2' }).beforeSessionId).toBe('s2')
expect(workspaceInsertSessionBeforeRequestSchema.parse({ workspaceId: 'w1', sessionId: 's1' }).beforeSessionId).toBeUndefined()
expect(() => workspaceInsertSessionBeforeRequestSchema.parse({ workspaceId: 'w1' })).toThrow()
expect(workspaceInsertSessionBeforeValueSchema.parse({ workspace: view }).workspace.workspaceId).toBe('w1')
})
})
describe('commands domain schemas', () => {
it('validates the catalog request/value pair', () => {
expect(commandListRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
// The wire is session-addressed only: a sessionId-less payload fails.
expect(() => commandListRequestSchema.parse({})).toThrow()
expect(commandListValueSchema.parse({ commands: [] }).commands).toEqual([])
const value = commandListValueSchema.parse({ commands: [
{ name: 'plan', description: 'Toggle plan mode' },
{ name: 'goal', description: 'Set the goal', input: { hint: '<goal>' } },
] })
expect(value.commands[1]?.input?.hint).toBe('<goal>')
expect(commandDescriptorSchema.parse({ name: 'x', description: 'd' }).input).toBeUndefined()
expect(() => commandDescriptorSchema.parse({ name: '', description: 'd' })).toThrow()
expect(() => commandDescriptorSchema.parse({ name: 'x', description: 'd', input: {} })).toThrow()
})
it('validates the execute request/value pair with both matched branches', () => {
expect(commandExecuteRequestSchema.parse({ sessionId: 's1', line: '/plan off' }).line).toBe('/plan off')
// Both members are mandatory: dropping either fails the parse.
expect(() => commandExecuteRequestSchema.parse({ line: '/compact' })).toThrow()
expect(() => commandExecuteRequestSchema.parse({ sessionId: 's1' })).toThrow()
expect(commandExecuteValueSchema.parse({ matched: false })).toEqual({ matched: false })
const matched = commandExecuteValueSchema.parse({ matched: true, result: { kind: 'success', text: 'done' } })
expect(matched.result?.kind).toBe('success')
expect(commandExecuteValueSchema.parse({ matched: true, result: { kind: 'error', text: 'bad' } }).result?.kind).toBe('error')
expect(() => commandExecuteValueSchema.parse({ matched: true, result: { kind: 'other' } })).toThrow()
})
})
describe('skills domain schemas', () => {
it('validates the list request/value pair', () => {
expect(skillListRequestSchema.parse({ sessionId: 's1' })).toEqual({ sessionId: 's1' })
// The wire is session-addressed only: a sessionId-less payload fails.
expect(() => skillListRequestSchema.parse({})).toThrow()
expect(skillListValueSchema.parse({ skills: [] }).skills).toEqual([])
const value = skillListValueSchema.parse({ skills: [
{ name: 'commit-helper', description: 'Git commits', whenToUse: 'when committing' },
{ name: 'bare', description: 'No guidance' },
] })
expect(value.skills[0]?.whenToUse).toBe('when committing')
expect(value.skills[1]?.whenToUse).toBeUndefined()
expect(() => skillEntrySchema.parse({ name: '', description: 'd' })).toThrow()
})
})
describe('events frame schemas', () => {
it('accepts every mux frame branch', () => {
const frames = [
@@ -167,6 +240,8 @@ describe('events frame schemas', () => {
{ type: 'approval/resolved', sessionId: 's', approvalId: 'a', outcome: 'allowed-once' },
{ type: 'question/requested', sessionId: 's', questions: [{ id: 'q', question: 'Q?', options: [{ label: 'L' }], multiSelect: true }] },
{ type: 'question/resolved', sessionId: 's', questionRpcId: 'r', outcome: 'answered' },
{ type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' }, steering: false },
{ type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' }, steering: true },
{ type: 'stream/error', error: { code: 'internal', message: 'm', details: {} } },
]
for (const frame of frames) expect(muxFrameSchema.parse(frame)).toMatchObject({ type: frame.type })
@@ -185,13 +260,20 @@ describe('events frame schemas', () => {
expect(() => muxFrameSchema.parse({ type: 'question/requested', sessionId: 's', questions: [] })).toThrow()
})
it('rejects a queued frame missing its members', () => {
expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [{ type: 'text' }], source: { kind: 'user' } })).toThrow()
expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: 'x', source: { kind: 'user' }, steering: false })).toThrow()
expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: {}, steering: false })).toThrow()
})
it('accepts every host frame branch', () => {
const frames = [
{ type: 'host/session-added', sessionId: 's', parentSessionId: 'p' },
{ type: 'host/session-added', sessionId: 's' },
{ type: 'host/session-added', sessionId: 's', blank: true, parentSessionId: 'p' },
{ type: 'host/session-added', sessionId: 's', blank: true },
{ type: 'host/session-removed', sessionId: 's' },
{ type: 'host/session-status', sessionId: 's', running: true },
{ type: 'host/agent-error', sessionId: 's', message: 'boom' },
{ type: 'host/commands-changed' },
{ type: 'stream/error', error: { code: 'internal', message: 'm', details: {} } },
]
for (const frame of frames) expect(hostFrameSchema.parse(frame)).toMatchObject({ type: frame.type })