mengzhx/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness.git synced 2026-08-29 04:26:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(subagent): cover Remote control migration

Browse Source
This commit is contained in:
imccyu
2026-08-26 14:52:53 +08:00
parent cbe5d76e5c
commit 91fea67745
22 changed files with 502 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/web/tests/subagent-conversation.e2e.ts
+3 -3
View File
@@ -267,7 +267,7 @@ describe('web e2e: persisted subagent conversation and human continuation', () =
it('keeps known descendants reachable across a stale empty catalog response', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-stale-catalog'))
const pattern = '**/api/subagent.list'
const pattern = '**/api/subagents/list'
let firstClaimed = false
let emptyDelivered = false
let trailingRequested = false
@@ -373,7 +373,7 @@ describe('web e2e: persisted subagent conversation and human continuation', () =
it('keeps a restored child neutral until its parent availability arrives', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-restore'))
const pattern = '**/api/subagent.list'
const pattern = '**/api/subagents/list'
let requested = false
let releaseCatalog = (): void => {}
const catalogHeld = new Promise<void>((resolve) => { releaseCatalog = resolve })
@@ -546,7 +546,7 @@ describe('web e2e: persisted subagent conversation and human continuation', () =
const input = page.locator('[data-composer-input][contenteditable="true"]').first()
await input.waitFor()
const promptResponse = page.waitForResponse(response =>
new URL(response.url()).pathname === '/api/subagent.prompt')
new URL(response.url()).pathname === '/api/subagents/prompt')
await input.fill(POST_FORK_FOLLOWUP)
await input.press('Enter')
const promptReceipt = await (await promptResponse).json() as {
apps/web/tests/subagent-interrupt-ui.e2e.ts
+6 -6
View File
@@ -1,7 +1,7 @@
// Web e2e scenario: the composer's independent Stop interrupts a running
// continuable child. The child holds its model turn open through a replay
// hang entry; the browser proves Send and Stop coexist, the parent-offline
// disabled-Send-with-Stop composer, the subagent.interrupt
// disabled-Send-with-Stop composer, the subagents/interruptByParent
// (never session.cancel) transport, the parked follow-up, and the FIFO resume
// on a waking send.
//
@@ -187,7 +187,7 @@ describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running co
// parentAvailable: false while the child Activation stays live (the
// interrupt RPC itself needs no live parent — covered host-side by
// subagent-interrupt.e2e.ts).
const pattern = '**/api/subagent.list'
const pattern = '**/api/subagents/list'
await page.route(pattern, async (route) => {
const response = await route.fetch()
const body = await response.json() as {
@@ -228,7 +228,7 @@ describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running co
)
const aborted = waitForAbortedTurn(scaffold, childId)
const interruptResponse = page.waitForResponse(response =>
new URL(response.url()).pathname === '/api/subagent.interrupt')
new URL(response.url()).pathname === '/api/subagents/interruptByParent')
await stop.click()
expect(((await (await interruptResponse).json()) as {
result: { ok: boolean; value?: { accepted: boolean } }
@@ -252,7 +252,7 @@ describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running co
}
}, 60_000)
it('interrupts through subagent.interrupt, parks the follow-up, and resumes it FIFO', async () => {
it('interrupts through subagents/interruptByParent, parks the follow-up, and resumes it FIFO', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-interrupt-flow'))
// Reselect the child with the truthful catalog: parent available again.
await page.getByRole('navigation', { name: 'Session hierarchy' })
@@ -265,7 +265,7 @@ describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running co
// Queue a follow-up through Send while independent Stop remains available.
const promptResponse = page.waitForResponse(response =>
new URL(response.url()).pathname === '/api/subagent.prompt')
new URL(response.url()).pathname === '/api/subagents/prompt')
await input.fill(FOLLOWUP)
await page.getByRole('button', { name: 'Send message' }).click()
expect(((await (await promptResponse).json()) as { result: { ok: boolean } }).result)
@@ -275,7 +275,7 @@ describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running co
const stop = page.getByRole('button', { name: 'Stop generating' })
expect(await stop.count()).toBe(1)
const interruptResponse = page.waitForResponse(response =>
new URL(response.url()).pathname === '/api/subagent.interrupt')
new URL(response.url()).pathname === '/api/subagents/interruptByParent')
await stop.click()
expect(((await (await interruptResponse).json()) as {
result: { ok: boolean; value?: { accepted: boolean } }
apps/web/tests/subagent-interrupt.e2e.ts
+32 -34
View File
@@ -1,4 +1,4 @@
// Web e2e scenario (browserless): the subagent.interrupt RPC against the real
// Web e2e scenario (browserless): the subagents interrupt Remote against the real
// composition. A live continuable child holds its model turn open through a
// replay hang entry; plain HTTP queues a follow-up, interrupts the turn, and
// proves from the real session state that the turn aborted, the follow-up
@@ -22,25 +22,12 @@ const WAKING = 'And add one concrete example.'
type RpcResult<T> = { ok: true; value: T } | { ok: false; error: { code: string; message: string } }
/** POST one API Proxy unary RPC through the real HTTP carrier and unwrap its result. */
async function rpc<T>(scaffold: WebScaffold, method: string, payload: unknown): Promise<RpcResult<T>> {
const response = await scaffold.hostFetch(`/api/${method}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({
type: 'client-request',
rpcId: `interrupt-e2e-${method}-${randomUUID()}`,
method,
payload,
}),
})
if (!response.ok) throw new Error(`${method} failed over HTTP ${response.status}: ${await response.text()}`)
return (await response.json() as { result: RpcResult<T> }).result
}
/** POST one generated Session Remote unary through the API Gateway carrier. */
async function sessionRemote<T>(scaffold: WebScaffold, method: string, request: unknown): Promise<RpcResult<T>> {
const endpoint = `session/${method}`
/** POST one generated Remote unary through the API Gateway carrier. */
async function remote<T>(
scaffold: WebScaffold,
endpoint: string,
args: Readonly<Record<string, unknown>>,
): Promise<RpcResult<T>> {
const response = await scaffold.hostFetch(`/api/${endpoint}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
@@ -48,13 +35,18 @@ async function sessionRemote<T>(scaffold: WebScaffold, method: string, request:
type: 'client-request',
rpcId: `interrupt-e2e-${endpoint}-${randomUUID()}`,
method: endpoint,
payload: { args: { request } },
payload: { args },
}),
})
if (!response.ok) throw new Error(`${endpoint} failed over HTTP ${response.status}: ${await response.text()}`)
return (await response.json() as { result: RpcResult<T> }).result
}
/** POST one generated Session Remote unary through the API Gateway carrier. */
function sessionRemote<T>(scaffold: WebScaffold, method: string, request: unknown): Promise<RpcResult<T>> {
return remote<T>(scaffold, `session/${method}`, { request })
}
/** Poll a synchronous condition (hook-safe; expect.poll is test-body only). */
async function waitFor(predicate: () => boolean, what: string, timeoutMs = 30_000): Promise<void> {
const deadline = Date.now() + timeoutMs
@@ -78,7 +70,7 @@ function textCompletion(text: string): object {
}
}
describe.skipIf(MODE === 'record')('web e2e: subagent.interrupt over the real composition', () => {
describe.skipIf(MODE === 'record')('web e2e: subagents/interruptByParent over the real composition', () => {
let scaffold: WebScaffold
let sidecarRoot: string
let readyFile: string
@@ -138,18 +130,21 @@ describe.skipIf(MODE === 'record')('web e2e: subagent.interrupt over the real co
it('parks a queued follow-up on interrupt and resumes it FIFO on a waking send', async () => {
// Queue the follow-up while the turn is still open, then interrupt.
const queued = await rpc<{ messageId: string }>(scaffold, 'subagent.prompt', {
parentSessionId: parentId,
childSessionId: childId,
mode: 'continuable',
content: [{ type: 'text', text: FOLLOWUP }],
const queued = await remote<{ messageId: string }>(scaffold, 'subagents/prompt', {
request: {
requestId: randomUUID(),
parentSessionId: parentId,
childSessionId: childId,
mode: 'continuable',
content: [{ type: 'text', text: FOLLOWUP }],
},
})
expect(queued).toMatchObject({ ok: true })
const settled = scaffold.whenTurnSettled()
const interrupted = await rpc<{ accepted: true }>(scaffold, 'subagent.interrupt', {
parentSessionId: parentId,
const interrupted = await remote<{ accepted: true }>(scaffold, 'subagents/interruptByParent', {
childSessionId: childId,
parentSessionId: parentId,
mode: 'continuable',
})
expect(interrupted).toMatchObject({ ok: true, value: { accepted: true } })
@@ -169,11 +164,14 @@ describe.skipIf(MODE === 'record')('web e2e: subagent.interrupt over the real co
// Only an explicit waking send resumes the parked queue, FIFO, then the
// child runs both turns to completion and settles.
const waking = await rpc<{ messageId: string }>(scaffold, 'subagent.prompt', {
parentSessionId: parentId,
childSessionId: childId,
mode: 'continuable',
content: [{ type: 'text', text: WAKING }],
const waking = await remote<{ messageId: string }>(scaffold, 'subagents/prompt', {
request: {
requestId: randomUUID(),
parentSessionId: parentId,
childSessionId: childId,
mode: 'continuable',
content: [{ type: 'text', text: WAKING }],
},
})
expect(waking).toMatchObject({ ok: true })
await expect.poll(() => scaffold.ctx.agents.get(childId), { timeout: 60_000 }).toBeUndefined()
packages/api/session-controller/tests/client-apply.client.spec.ts
+1
View File
@@ -85,6 +85,7 @@ async function mount(initialHost?: HostDescription): Promise<Bench> {
})
ctx.reflect.provide('remote.commands', remote.commands)
ctx.reflect.provide('remote.session', remote.session)
ctx.reflect.provide('remote.subagents', remote.subagents)
const fiber = ctx.plugin(SessionClient)
await fiber
const sessions = ctx.sessions as ClientSessions
packages/api/session-controller/tests/fake-api.client.ts
+33 -15
View File
@@ -2,8 +2,9 @@
// data source on a real clock; behavior tests need per-case responses and
// deferred-controlled timing). Session streams are hand pumps: pushFollow/pushControl.
import type {
IApiClient,
IApiClient, MessageId,
RpcError, RpcResponse, SessionId, SessionSearchItem, SkillEntry,
SubagentCatalog, SubagentInterruptReceipt, SubagentPromptReceipt,
WorkspaceId, WorkspaceView,
} from '@deepseek-ai/dsh-api-remotes/client'
import type {
@@ -20,7 +21,7 @@ import type {
} from '@deepseek-ai/dsh-api-session-controller/types'
import type { WorkspaceRemote } from '@deepseek-ai/dsh-api-workspace-controller/client'
import type { WorkspaceFollowFrame } from '@deepseek-ai/dsh-api-workspace-controller/types'
import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
import type { RemoteFailure, RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
import {
RemoteStream,
RemoteStreamError,
@@ -84,10 +85,20 @@ export function err<T>(error: RpcError): RpcResponse<T> {
}
/** Successful generated Remote result for programmable domain fakes. */
function remoteOk<T>(value: T): RemoteResult<T> {
export function remoteOk<T>(value: T): RemoteResult<T> {
return { ok: true, value }
}
/**
* Failed generated Remote result carrying an owner's own failure vocabulary,
* which the carrier's closed RPC code set does not contain.
* @param error - the owner-declared failure.
* @returns the failure branch of a Remote result.
*/
export function remoteErr<T>(error: RemoteFailure): RemoteResult<T> {
return { ok: false, error }
}
type ValueStreamItem<F> =
| { kind: 'frame'; value: F; delivered?: () => void }
| { kind: 'end' }
@@ -189,19 +200,13 @@ export class FakeApiClient implements IApiClient {
}
lastSearchSignal: AbortSignal | undefined
onSubagentList: (payload: unknown) => Promise<RpcResponse<{ entries: never[]; parentAvailable: boolean }>>
= () => Promise.resolve(ok({ entries: [], parentAvailable: true }))
onSubagentPrompt: (payload: unknown) => Promise<RpcResponse<{ messageId: never }>>
= () => Promise.resolve(ok({ messageId: 'fake-message' as never }))
onSubagentList: (payload: unknown) => Promise<RemoteResult<SubagentCatalog>>
= () => Promise.resolve(remoteOk({ entries: [], parentAvailable: true }))
onSubagentPrompt: (payload: unknown) => Promise<RemoteResult<SubagentPromptReceipt>>
= () => Promise.resolve(remoteOk({ messageId: 'fake-message' as MessageId }))
onSubagentInterrupt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>>
= () => Promise.resolve(ok({ accepted: true as const }))
readonly subagents: IApiClient['subagents'] = {
list: (payload: unknown) => this.record('subagent.list', payload, this.onSubagentList(payload)),
prompt: (payload: unknown) => this.record('subagent.prompt', payload, this.onSubagentPrompt(payload)),
interrupt: (payload: unknown) => this.record('subagent.interrupt', payload, this.onSubagentInterrupt(payload)),
}
onSubagentInterrupt: (payload: unknown) => Promise<RemoteResult<SubagentInterruptReceipt>>
= () => Promise.resolve(remoteOk({ accepted: true as const }))
readonly host: IApiClient['host'] = {
describe: (payload: unknown) => this.record('host.describe', payload, this.onDescribe(payload)),
@@ -301,6 +306,19 @@ export class FakeApiClient implements IApiClient {
follow: (request, signal) => this.openFollow(request, signal),
control: signal => this.openControl(signal),
},
subagents: {
list: parentSessionId => this.record(
'subagents.list',
parentSessionId,
this.onSubagentList(parentSessionId),
),
prompt: request => this.record('subagents.prompt', request, this.onSubagentPrompt(request)),
interruptByParent: (childSessionId, parentSessionId, mode) => this.record(
'subagents.interruptByParent',
{ childSessionId, parentSessionId, mode },
this.onSubagentInterrupt({ childSessionId, parentSessionId, mode }),
),
},
workspace: {
create: payload => this.record('workspace.create', payload, this.onWorkspaceCreate(payload)),
rename: payload => this.record('workspace.rename', payload, this.onWorkspaceRename(payload)),
packages/api/session-controller/tests/manager.client.spec.ts
+73 -77
View File
@@ -8,7 +8,7 @@ import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types'
import type {} from '@deepseek-ai/dsh-session-title/client'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts'
import { FakeApiClient, deferred, err, fakeRemote, ok, remoteErr, remoteOk } from './fake-api.client.ts'
import { entries, plainTurn } from './event-script.client.ts'
const S1 = 'fk-m1' as SessionId
@@ -29,14 +29,14 @@ function summary(sessionId: SessionId, over: SummaryOver = {}) {
function makeManager(): SessionManager {
const api = new FakeApiClient()
return new SessionManager(api, fakeRemote(api))
return new SessionManager(fakeRemote(api))
}
describe('SessionManager instances', () => {
it('lazily builds one resident instance per id and syncs the running bit from the list', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
const session = manager.get(S1)
expect(manager.get(S1)).toBe(session) // resident: same instance forever
@@ -50,7 +50,7 @@ describe('list lifecycle', () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => gate.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const first = manager.refreshList()
const second = manager.refreshList()
expect(manager.getListSnapshot().state).toBe('loading')
@@ -66,7 +66,7 @@ describe('list lifecycle', () => {
const api = new FakeApiClient()
const first = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => first.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const hydration = manager.refreshList()
manager.handleSessionAdded(summary(S2, { blank: true }))
first.resolve(ok({ items: [summary(S1)] as never[] }))
@@ -83,7 +83,7 @@ describe('list lifecycle', () => {
it('advances list activity from the filtered Host notification', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
manager.handleSessionActivity(S1, 500)
@@ -93,7 +93,7 @@ describe('list lifecycle', () => {
it('keeps the error in the list snapshot on failure', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(err({ code: 'internal', message: 'boom', details: {} }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal' } })
// A failed pull does not step the arrival phase: still pending.
@@ -102,7 +102,7 @@ describe('list lifecycle', () => {
it('phase steps pending → ready on the first successful pull and never returns', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
expect(manager.getListSnapshot().phase).toBe('pending')
await manager.refreshList()
expect(manager.getListSnapshot().phase).toBe('ready')
@@ -121,7 +121,7 @@ describe('list lifecycle', () => {
it('merges create into the list immediately without waiting for a refresh', async () => {
const api = new FakeApiClient()
api.onCreate = () => Promise.resolve(ok({ sessionId: S2 }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const result = await manager.create()
expect(result).toMatchObject({ ok: true, value: { sessionId: S2 } })
expect(manager.getListSnapshot().items.map(i => i.sessionId)).toEqual([S2])
@@ -129,7 +129,7 @@ describe('list lifecycle', () => {
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, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const titleFrame = (title: string, seq: number) => {
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: title, seq })
}
@@ -153,7 +153,7 @@ describe('list lifecycle', () => {
it('seeds cold titles from the list rows\' projections block under higher-seq-wins', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
// A push frame landed before the list (S2's title is newer than the block's cut).
manager.handleControlFrame({
type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9,
@@ -175,7 +175,7 @@ describe('list lifecycle', () => {
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, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
const frame = (payload: SessionControlFrame) => { manager.handleControlFrame(payload) }
frame({ type: 'projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
@@ -213,7 +213,7 @@ describe('search', () => {
items: [{ sessionId: S1, snippet: 'matching excerpt' }],
hasMore: true,
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const signal = new AbortController().signal
await expect(manager.search('exact phrase', signal)).resolves.toEqual({
@@ -229,7 +229,7 @@ describe('search', () => {
it('preserves business errors and folds transport failures', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
api.onSearch = () => Promise.resolve(err({
code: 'internal',
message: 'index unavailable',
@@ -252,7 +252,7 @@ describe('search', () => {
describe('Host Remote event routing', () => {
it('adds/removes/flips sessions and keeps removed instances resident', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
manager.handleSessionAdded(summary(S1, { blank: true }))
manager.handleSessionAdded(summary(S1, { blank: true })) // dup: ignored
expect(manager.getListSnapshot().items).toHaveLength(1)
@@ -279,14 +279,14 @@ describe('subagent catalogs', () => {
summary(S1),
summary(S2, { parentSessionId: S1, origin: 'subagent' }),
] as never[] }))
api.onSubagentList = () => Promise.resolve(ok({
api.onSubagentList = () => Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: S2, mode: 'continuable', label: 'worker',
activity: 'running', hasChildren: false,
}] as never[],
parentAvailable: true,
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
await manager.refreshSubagents(S1)
manager.selectSubagent({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
@@ -319,21 +319,23 @@ describe('subagent catalogs', () => {
},
])
expect(api.callsOf('subagent.history')).toEqual([])
expect(api.callsOf('subagent.prompt')).toEqual([
expect(api.callsOf('subagents.prompt')).toEqual([
{
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
requestId: expect.any(String) as unknown as string,
parentSessionId: S1, childSessionId: S2,
mode: 'continuable',
content: [{ type: 'text', text: 'continue' }],
clientTimeZone: new Intl.DateTimeFormat().resolvedOptions().timeZone,
},
])
expect(api.callsOf('session.history')).toEqual([])
expect(api.callsOf('session.prompt')).toEqual([])
const listCalls = api.callsOf('subagent.list').length
const listCalls = api.callsOf('subagents.list').length
manager.handleSessionStatus(S2, false)
expect(manager.getListSnapshot().subagentsByParent[S1]?.entries[0]).toMatchObject({
kind: 'child', id: S2, activity: 'inactive',
})
expect(api.callsOf('subagent.list')).toHaveLength(listCalls)
expect(api.callsOf('subagents.list')).toHaveLength(listCalls)
manager.handleSessionRemoved(S2)
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toMatchObject({
@@ -351,20 +353,20 @@ describe('subagent catalogs', () => {
vi.useFakeTimers()
try {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshSubagents(S1)
manager.setSubagentCatalogOpen(S1, true)
await Promise.resolve()
const baseline = api.callsOf('subagent.list').length
const baseline = api.callsOf('subagents.list').length
manager.handleSessionAdded(summary(S2, { parentSessionId: S1 }))
manager.handleSessionAdded(summary('fk-m3' as SessionId, { parentSessionId: S1 }))
await vi.advanceTimersByTimeAsync(50)
expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
expect(api.callsOf('subagents.list')).toHaveLength(baseline + 1)
manager.setSubagentCatalogOpen(S1, false)
manager.handleSessionAdded(summary('fk-m4' as SessionId, { parentSessionId: S1 }))
await vi.advanceTimersByTimeAsync(50)
expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
expect(api.callsOf('subagents.list')).toHaveLength(baseline + 1)
} finally {
vi.useRealTimers()
}
@@ -373,7 +375,7 @@ describe('subagent catalogs', () => {
it('marks a loaded parent row expandable only for a direct subagent publication', async () => {
const api = new FakeApiClient()
const root = 'fk-root' as SessionId
api.onSubagentList = () => Promise.resolve(ok({
api.onSubagentList = () => Promise.resolve(remoteOk({
entries: [
{
kind: 'child', id: S1, mode: 'continuable', label: 'parent',
@@ -386,7 +388,7 @@ describe('subagent catalogs', () => {
] as never[],
parentAvailable: true,
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshSubagents(root)
manager.handleSessionAdded(summary('fk-grandchild' as SessionId, {
@@ -405,13 +407,13 @@ describe('subagent catalogs', () => {
const root = 'fk-root' as SessionId
const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => response.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshSubagents(root)
manager.handleSessionAdded(summary('fk-grandchild' as SessionId, {
parentSessionId: S1, origin: 'subagent',
}))
response.resolve(ok({
response.resolve(remoteOk({
entries: [{
kind: 'child', id: S1, mode: 'continuable', label: 'parent',
activity: 'inactive', hasChildren: false,
@@ -424,7 +426,7 @@ describe('subagent catalogs', () => {
{ kind: 'child', id: S1, hasChildren: true },
])
api.onSubagentList = () => Promise.resolve(ok({
api.onSubagentList = () => Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: S1, mode: 'continuable', label: 'parent',
activity: 'inactive', hasChildren: false,
@@ -442,12 +444,12 @@ describe('subagent catalogs', () => {
const root = 'fk-root' as SessionId
const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => response.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshSubagents(root)
manager.handleSessionStatus(S1, false)
manager.handleSessionStatus(S2, true)
response.resolve(ok({
response.resolve(remoteOk({
entries: [
{
kind: 'child', id: S1, mode: 'continuable', label: 'stopped',
@@ -470,14 +472,14 @@ describe('subagent catalogs', () => {
it('marks a detached catalog child inactive without requiring a selected address', async () => {
const api = new FakeApiClient()
api.onSubagentList = () => Promise.resolve(ok({
api.onSubagentList = () => Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: S2, mode: 'continuable', label: 'worker',
activity: 'running', hasChildren: false,
}] as never[],
parentAvailable: true,
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshSubagents(S1)
manager.handleSessionRemoved(S2)
@@ -492,15 +494,15 @@ describe('subagent catalogs', () => {
const root = 'fk-root' as SessionId
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => first.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshSubagents(root)
expect(manager.refreshSubagents(root)).toBe(refresh)
api.onSubagentList = () => Promise.resolve(ok({ entries: [], parentAvailable: true }))
first.resolve(ok({ entries: [], parentAvailable: true }))
api.onSubagentList = () => Promise.resolve(remoteOk({ entries: [], parentAvailable: true }))
first.resolve(remoteOk({ entries: [], parentAvailable: true }))
await refresh
expect(api.callsOf('subagent.list')).toHaveLength(1)
expect(api.callsOf('subagents.list')).toHaveLength(1)
})
it('runs one trailing catalog refresh for a membership change coalesced into an in-flight pull', async () => {
@@ -511,7 +513,7 @@ describe('subagent catalogs', () => {
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
const second = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => first.promise
const manager = new SessionManager(api, fakeRemote(api), root)
const manager = new SessionManager(fakeRemote(api), root)
const refresh = manager.refreshSubagents(root)
manager.setSubagentCatalogOpen(root, true)
@@ -522,7 +524,7 @@ describe('subagent catalogs', () => {
manager.handleSessionAdded(summary(S2, { parentSessionId: root }))
await vi.advanceTimersByTimeAsync(50)
api.onSubagentList = () => second.promise
first.resolve(ok({
first.resolve(remoteOk({
entries: [{
kind: 'child', id: S1, mode: 'continuable', label: 'older',
activity: 'inactive', hasChildren: false,
@@ -531,7 +533,7 @@ describe('subagent catalogs', () => {
}))
await refresh
// The trailing pull is already in flight (kicked synchronously in finally).
second.resolve(ok({
second.resolve(remoteOk({
entries: [
{
kind: 'child', id: S1, mode: 'continuable', label: 'older',
@@ -545,8 +547,10 @@ describe('subagent catalogs', () => {
parentAvailable: true,
}))
await second.promise
// The Remote face resolves one microtask after the response settles.
await vi.advanceTimersByTimeAsync(0)
expect(api.callsOf('subagent.list')).toHaveLength(2)
expect(api.callsOf('subagents.list')).toHaveLength(2)
expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
{ kind: 'child', id: S1, label: 'older' },
{ kind: 'child', id: S2, label: 'new child' },
@@ -565,9 +569,9 @@ describe('subagent catalogs', () => {
})
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => first.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshSubagents(root)
first.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
first.resolve(remoteOk({ entries: [child()] as never[], parentAvailable: true }))
await refresh
manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
@@ -579,12 +583,12 @@ describe('subagent catalogs', () => {
manager.handleSessionRemoved(root)
const trailing = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = () => trailing.promise
mid.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
mid.resolve(remoteOk({ entries: [child()] as never[], parentAvailable: true }))
await midRefresh
expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
trailing.resolve(err({ code: 'internal', message: 'trailing pull failed', details: {} }))
trailing.resolve(remoteErr({ code: 'internal', message: 'trailing pull failed', details: {} }))
await vi.waitFor(() => {
expect(manager.getListSnapshot().subagentsByParent[root]).toMatchObject({
state: 'error',
@@ -592,8 +596,7 @@ describe('subagent catalogs', () => {
})
})
const rootCalls = api.callsOf('subagent.list')
.filter(call => (call as { parentSessionId: SessionId }).parentSessionId === root)
const rootCalls = api.callsOf('subagents.list').filter(call => call === root)
expect(rootCalls).toHaveLength(3)
expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
@@ -602,14 +605,14 @@ describe('subagent catalogs', () => {
it('invalidates catalog availability when the owning parent is removed', async () => {
const api = new FakeApiClient()
const root = 'fk-root' as SessionId
api.onSubagentList = () => Promise.resolve(ok({
api.onSubagentList = () => Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: S2, mode: 'continuable', label: 'worker',
activity: 'inactive', hasChildren: false,
}] as never[],
parentAvailable: true,
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshSubagents(root)
manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: true })
@@ -625,14 +628,14 @@ describe('remaining branches', () => {
it('refreshList folds a transport throw into the error state', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.reject(new Error('list wire down'))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal', message: 'list wire down' } })
})
it('refreshList pushes running bits down to already-instantiated sessions', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const session = manager.get(S1)
api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
await manager.refreshList()
@@ -642,7 +645,7 @@ describe('remaining branches', () => {
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, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
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' })
@@ -662,7 +665,7 @@ describe('remaining branches', () => {
message: 'published but unattached',
details: { sessionId: S1, workspaceId: 'w1' },
} as never))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(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 })])
@@ -676,7 +679,7 @@ describe('remaining branches', () => {
message: 'forked but unattached',
details: { sessionId: S2, workspaceId: 'w1' },
} as never))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const result = await manager.fork({ sessionId: S1 })
expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({
@@ -689,7 +692,7 @@ describe('remaining branches', () => {
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, fakeRemote(api))
const manager = new SessionManager(fakeRemote(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([])
@@ -704,7 +707,7 @@ describe('remaining branches', () => {
it('subscribe notifies on list changes and stops after unsubscribe', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
let notified = 0
const unsubscribe = manager.subscribe(() => { notified++ })
await manager.refreshList()
@@ -719,7 +722,7 @@ describe('remaining branches', () => {
it('ignores Host status and error events for sessions without an instance', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
manager.handleSessionStatus(S2, true)
manager.handleSessionError(S2, '无实例')
})
@@ -727,7 +730,7 @@ describe('remaining branches', () => {
it('keeps list-entry identity for unchanged rows across an unrelated list change', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
const before = manager.getListSnapshot()
manager.handleSessionStatus(S2, true)
@@ -743,7 +746,7 @@ describe('remaining branches', () => {
it('carries parentSessionId from the added event into the lineage row', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
manager.handleSessionAdded(summary(S1, { blank: true }))
manager.handleSessionAdded(summary(S2, {
blank: true, parentSessionId: S1, origin: 'subagent',
@@ -763,7 +766,7 @@ describe('connected generation', () => {
hasMore: false,
modelSelection: { provider: 'deepseek-official', model: 'deepseek-chat' },
}))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const openedSession = manager.get(S1)
await openedSession.open()
manager.get(S2) // instantiated but never opened
@@ -782,26 +785,19 @@ describe('connected generation', () => {
}
const parent = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
const child = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
api.onSubagentList = payload => (
(payload as { parentSessionId: SessionId }).parentSessionId === S1
? parent.promise
: child.promise
)
const manager = new SessionManager(api, fakeRemote(api), S2, address)
api.onSubagentList = payload => (payload === S1 ? parent.promise : child.promise)
const manager = new SessionManager(fakeRemote(api), S2, address)
manager.handleConnected()
expect(manager.get(S2).getSnapshot().subagent).toEqual({ address })
parent.resolve(ok({ entries: [], parentAvailable: true }))
child.resolve(ok({ entries: [], parentAvailable: true }))
parent.resolve(remoteOk({ entries: [], parentAvailable: true }))
child.resolve(remoteOk({ entries: [], parentAvailable: true }))
await vi.waitFor(() => {
expect(api.callsOf('session.list')).toHaveLength(1)
})
await vi.waitFor(() => {
expect(api.callsOf('subagent.list')).toEqual([
{ parentSessionId: S1 },
{ parentSessionId: S2 },
])
expect(api.callsOf('subagents.list')).toEqual([S1, S2])
})
expect(manager.get(S2).getSnapshot().subagent).toEqual({
address,
@@ -882,7 +878,7 @@ describe('completed reminder', () => {
it('a list refresh carrying the running→idle transition arms the reminder', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: true })] as never[] }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
manager.select(S1)
expect(entry(manager, S2)?.completed).toBe(false)
@@ -894,7 +890,7 @@ describe('completed reminder', () => {
it('never arms for sessions already idle at first observation', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
await manager.refreshList()
manager.select(S1)
expect(entry(manager, S2)?.completed).toBe(false)
@@ -907,7 +903,7 @@ describe('completed reminder', () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => gate.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshList()
// The session finishes while the first pull is still in flight; the pull
// response recorded it as running at pull time.
@@ -921,7 +917,7 @@ describe('completed reminder', () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
api.onList = () => gate.promise
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
const refresh = manager.refreshList()
// The unknown session starts and finishes while the first pull is in
// flight; the pull-time baseline recorded it idle, so the running→idle
packages/api/session-controller/tests/projection-store.client.spec.ts
+7 -7
View File
@@ -103,7 +103,7 @@ describe('Session projection value semantics', () => {
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, fakeRemote(api))
const session = new Session(SID, fakeRemote(api))
api.onHistory = () => Promise.resolve(ok({
records: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false,
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['from-baseline'] } } },
@@ -114,7 +114,7 @@ describe('Session tail-page seeding', () => {
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, fakeRemote(api))
const session = new Session(SID, fakeRemote(api))
api.onHistory = () => Promise.resolve(ok({
records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['baseline'] } } },
@@ -127,7 +127,7 @@ describe('Session tail-page seeding', () => {
it('treats a blockless response as no reset: pushed values survive', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api, fakeRemote(api))
const session = new Session(SID, fakeRemote(api))
api.onHistory = () => Promise.resolve(ok({ records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
await session.open()
session.projections.apply('test/marks', { marks: ['pushed'] }, 9)
@@ -141,7 +141,7 @@ describe('manager frame routing', () => {
it('lands projection frames before instantiation and the Session adopts the same store', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
manager.handleControlFrame({
type: 'projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['early'] }, seq: 7,
})
@@ -156,7 +156,7 @@ describe('manager frame routing', () => {
it('projects the title key into list rows and truncates phantom rows on the control baseline', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
}) as never)
@@ -181,7 +181,7 @@ describe('manager frame routing', () => {
it('projects every retained value into list rows with stable snapshot identity', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
api.onList = () => Promise.resolve(ok({
items: [{
sessionId: sid('s1'), updatedAt: 1, running: false, blank: false,
@@ -208,7 +208,7 @@ describe('manager frame routing', () => {
it('drops the projection store with the removed session', async () => {
const api = new FakeApiClient()
const manager = new SessionManager(api, fakeRemote(api))
const manager = new SessionManager(fakeRemote(api))
api.onList = () => Promise.resolve(ok({
items: [{ sessionId: sid('s1'), updatedAt: 1, running: false, blank: false }],
}) as never)
packages/api/session-controller/tests/queue-store.client.spec.ts
+3 -3
View File
@@ -48,12 +48,12 @@ function makeSession(): Session {
function makeBench(): { api: FakeApiClient; session: Session } {
const api = new FakeApiClient()
return { api, session: new Session(SID, api, fakeRemote(api)) }
return { api, session: new Session(SID, fakeRemote(api)) }
}
function makeManager(): SessionManager {
const api = new FakeApiClient()
return new SessionManager(api, fakeRemote(api))
return new SessionManager(fakeRemote(api))
}
describe('Session queue snapshot intake', () => {
@@ -210,7 +210,7 @@ describe('Session queue snapshot intake', () => {
describe('queue operation transport', () => {
it('addresses the session.updateQueue RPC without optimistic local mutation', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api, fakeRemote(api))
const session = new Session(SID, fakeRemote(api))
session.handleControlFrame(queueFrame([{ id: 'q-op', body: 'pending' }]))
const before = session.getSnapshot().queue
packages/api/session-controller/tests/session.client.spec.ts
+15 -13
View File
@@ -5,7 +5,7 @@ import { RemoteStreamError } from '@deepseek-ai/dsh-api-gateway/client'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
import { Session, type SessionOptions } from '../src/client/sessions/session.ts'
import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts'
import { FakeApiClient, deferred, err, fakeRemote, ok, remoteErr } from './fake-api.client.ts'
import { entries, ev, historyValue, plainTurn } from './event-script.client.ts'
const SID = 'fk-s1' as SessionId
@@ -19,7 +19,7 @@ function makeSession(
api = new FakeApiClient(),
options: SessionOptions = {},
): { api: FakeApiClient; session: Session } {
return { api, session: new Session(SID, api, fakeRemote(api), options) }
return { api, session: new Session(SID, fakeRemote(api), options) }
}
function follow(
@@ -239,7 +239,7 @@ describe('paging', () => {
describe('prompt and cancel errors', () => {
it('routes an addressed child through non-activating history, continuation prompt, and interrupt only', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api, fakeRemote(api), {
const session = new Session(SID, fakeRemote(api), {
address: { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable' },
parentAvailable: true,
})
@@ -258,15 +258,17 @@ describe('prompt and cancel errors', () => {
},
])
expect(api.callsOf('subagent.history')).toEqual([])
expect(api.callsOf('subagent.prompt')).toEqual([
expect(api.callsOf('subagents.prompt')).toEqual([
{
parentSessionId: PARENT, childSessionId: SID, mode: 'continuable',
requestId: expect.any(String) as unknown as string,
parentSessionId: PARENT, childSessionId: SID,
mode: 'continuable',
content: [{ type: 'text', text: '继续' }],
clientTimeZone: new Intl.DateTimeFormat().resolvedOptions().timeZone,
},
])
expect(api.callsOf('subagent.interrupt')).toEqual([
{ parentSessionId: PARENT, childSessionId: SID, mode: 'continuable' },
expect(api.callsOf('subagents.interruptByParent')).toEqual([
{ childSessionId: SID, parentSessionId: PARENT, mode: 'continuable' },
])
expect(api.callsOf('session.history')).toEqual([])
expect(api.callsOf('session.prompt')).toEqual([])
@@ -281,10 +283,10 @@ describe('prompt and cancel errors', () => {
it('lands an interrupt business failure in promptError with op=stop', async () => {
const api = new FakeApiClient()
api.onSubagentInterrupt = () => Promise.resolve(err({
api.onSubagentInterrupt = () => Promise.resolve(remoteErr({
code: 'subagent-unauthorized', message: 'nope', details: { childSessionId: SID },
}) as never)
const session = new Session(SID, api, fakeRemote(api), {
}))
const session = new Session(SID, fakeRemote(api), {
address: { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable' },
parentAvailable: true,
})
@@ -298,7 +300,7 @@ describe('prompt and cancel errors', () => {
it('keeps one-shot history readable without exposing prompt or cancel transport', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api, fakeRemote(api), {
const session = new Session(SID, fakeRemote(api), {
address: { parentSessionId: PARENT, childSessionId: SID, mode: 'one-shot' },
})
await session.open()
@@ -316,8 +318,8 @@ describe('prompt and cancel errors', () => {
},
])
expect(api.callsOf('subagent.history')).toEqual([])
expect(api.callsOf('subagent.prompt')).toEqual([])
expect(api.callsOf('subagent.interrupt')).toEqual([])
expect(api.callsOf('subagents.prompt')).toEqual([])
expect(api.callsOf('subagents.interruptByParent')).toEqual([])
expect(api.callsOf('session.cancel')).toEqual([])
})
packages/api/session-controller/tests/sessions-service.client.spec.ts
+10 -9
View File
@@ -18,6 +18,7 @@ import {
err,
fakeRemote,
ok,
remoteOk,
type RuntimeRemotes,
} from './fake-api.client.ts'
@@ -33,7 +34,7 @@ function bench(configureRemote?: (remote: RuntimeRemotes) => RuntimeRemotes): Be
const ctx = new Context()
const api = new FakeApiClient()
const remote = fakeRemote(api)
const svc = new ClientSessions(ctx, api, configureRemote?.(remote) ?? remote)
const svc = new ClientSessions(ctx, configureRemote?.(remote) ?? remote)
return { ctx, api, svc }
}
@@ -522,9 +523,9 @@ describe('catalog-addressed navigation', () => {
it('uses catalog labels for a listed addressed route', async () => {
const b = bench()
b.api.onSubagentList = (payload) => {
const { parentSessionId } = payload as { parentSessionId: SessionId }
const parentSessionId = payload as SessionId
if (parentSessionId === sid('root')) {
return Promise.resolve(ok({
return Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: sid('child'), mode: 'continuable', label: 'Child',
activity: 'inactive', hasChildren: true,
@@ -533,7 +534,7 @@ describe('catalog-addressed navigation', () => {
}))
}
if (parentSessionId === sid('child')) {
return Promise.resolve(ok({
return Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: sid('grandchild'), mode: 'continuable', label: 'Grandchild',
activity: 'inactive', hasChildren: false,
@@ -541,7 +542,7 @@ describe('catalog-addressed navigation', () => {
parentAvailable: false,
}))
}
return Promise.resolve(ok({ entries: [], parentAvailable: false }))
return Promise.resolve(remoteOk({ entries: [], parentAvailable: false }))
}
await feedList(b, [
{ id: 'root' },
@@ -561,9 +562,9 @@ describe('catalog-addressed navigation', () => {
it('projects a directly opened descendant route without retaining ancestor scopes or addresses', async () => {
const b = bench()
b.api.onSubagentList = (payload) => {
const { parentSessionId } = payload as { parentSessionId: SessionId }
const parentSessionId = payload as SessionId
if (parentSessionId === sid('root')) {
return Promise.resolve(ok({
return Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: sid('child'), mode: 'continuable', label: 'Child',
activity: 'inactive', hasChildren: true,
@@ -572,7 +573,7 @@ describe('catalog-addressed navigation', () => {
}))
}
if (parentSessionId === sid('child')) {
return Promise.resolve(ok({
return Promise.resolve(remoteOk({
entries: [{
kind: 'child', id: sid('grandchild'), mode: 'continuable', label: 'Grandchild',
activity: 'inactive', hasChildren: false,
@@ -580,7 +581,7 @@ describe('catalog-addressed navigation', () => {
parentAvailable: false,
}))
}
return Promise.resolve(ok({ entries: [], parentAvailable: false }))
return Promise.resolve(remoteOk({ entries: [], parentAvailable: false }))
}
await feedList(b, [{ id: 'root' }])
await b.svc.refreshSubagents(sid('root'))
packages/client/connection/src/client/fixture.ts
+11 -10
View File
@@ -3260,13 +3260,6 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
}
const api: ApiProxy = {
subagents: {
list: request => ok(request, { entries: [], parentAvailable: true }),
prompt: request => Promise.resolve(ok(request, {
messageId: `fixture-message-${request.payload.childSessionId}` as never,
})),
interrupt: request => Promise.resolve(ok(request, { accepted: true as const })),
},
host: {
describe: request => ok(request, {
version: '0.0.0-fixture', cwd: '/tmp/fixture', attachedSessions, home: FIXTURE_HOME, canOpenPath: true,
@@ -3469,6 +3462,17 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
case 'agentPresets/read': return Promise.resolve(presetRemotes.read(args.agentPreset as string))
case 'agentPresets/copy': return Promise.resolve(presetRemotes.copy(args.from as string, args.id as string))
case 'agentPresets/deletePreset': return Promise.resolve(presetRemotes.deletePreset(args.id as string))
case 'subagents/list': return Promise.resolve({
ok: true,
value: { entries: [], parentAvailable: true },
})
case 'subagents/prompt': return Promise.resolve({
ok: true,
value: {
messageId: `fixture-message-${(request as { childSessionId: SessionId }).childSessionId}`,
},
})
case 'subagents/interruptByParent': return Promise.resolve({ ok: true, value: { accepted: true } })
case 'session/list': return sessionApi.list(
args._request as Parameters<FixtureSessionApi['list']>[0],
)
@@ -3591,9 +3595,6 @@ export class FixtureApiClient extends AbstractApiClient {
signal: AbortSignal,
): Promise<RpcResponse<unknown>> {
switch (method) {
case 'subagent.list': return this.api.subagents.list(request)
case 'subagent.prompt': return this.api.subagents.prompt(request, signal)
case 'subagent.interrupt': return this.api.subagents.interrupt(request)
case 'host.describe': return this.api.host.describe(request)
case 'host.pickDirectory': return this.api.host.pickDirectory(request, new AbortController().signal)
case 'host.listDirectory': return this.api.host.listDirectory(request, new AbortController().signal)
packages/client/connection/tests/fake-api.client.ts
-13
View File
@@ -69,19 +69,6 @@ export class FakeApiClient implements IApiClient {
private readonly generationConns: StreamConn[] = []
readonly subagents: IApiClient['subagents'] = {
list: (payload: unknown) => this.record('subagent.list', payload, Promise.resolve(ok({
entries: [],
parentAvailable: true,
}))),
prompt: (payload: unknown) => this.record('subagent.prompt', payload, Promise.resolve(ok({
messageId: 'fake-message' as never,
}))),
interrupt: (payload: unknown) => this.record('subagent.interrupt', payload, Promise.resolve(ok({
accepted: true as const,
}))),
}
readonly host: IApiClient['host'] = {
describe: payload => this.record('host.describe', payload, this.onDescribe(payload)),
pickDirectory: payload => this.record('host.pickDirectory', payload, this.onPickDirectory(payload)),
packages/client/ui-subagent/tests/browser-plugin.client.spec.ts
+1 -1
View File
@@ -5,7 +5,7 @@ import { describe, expect, it } from 'vitest'
import type {
SessionListState, SessionSnapshot, SessionSummary,
} from '@deepseek-ai/dsh-api-session-controller/client'
import type { SubagentAddress } from '@deepseek-ai/dsh-client-connection/client'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
import { SlotRegistry } from '@deepseek-ai/dsh-client-ui-renderer/client'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import type { ComposerChainProps } from '@deepseek-ai/dsh-client-ui-conversation/client'
packages/client/ui-workspace/tests/workspaces-service.client.spec.ts
-1
View File
@@ -185,7 +185,6 @@ class FakeApiClient implements IApiClient {
onCreateDirectory: IApiClient['host']['createDirectory'] = () => Promise.resolve(ok({ path: '/home/u/new' }))
onOpenPath: IApiClient['host']['openPath'] = () => Promise.resolve(ok({ opened: true }))
declare readonly subagents: IApiClient['subagents']
declare readonly skills: IApiClient['skills']
declare readonly agentPresets: IApiClient['agentPresets']
declare readonly settings: IApiClient['settings']
packages/host/apiproxy/tests/api-proxy-subagents.spec.ts
-264
View File
@@ -1,264 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import type { SessionId } from '@deepseek-ai/dsh-session'
import { SubagentError } from '@deepseek-ai/dsh-subagent'
import { RpcId } from '../src/api/rpc.ts'
import type { RpcRequest } from '../src/api/rpc.ts'
import { createApiProxy } from '../src/api-proxy.ts'
const sid = (value: string): SessionId => value as SessionId
const PARENT = sid('parent')
const CHILD = sid('child')
function request<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId('subagent-rpc'), payload }
}
function bench(options: {
parentLive?: boolean
childStatus?: 'idle' | 'running'
entries?: object[]
followupError?: Error
interruptError?: Error
listError?: Error
} = {}) {
const parent = { id: PARENT }
const child = options.childStatus === undefined
? undefined
: { id: CHILD, status: options.childStatus }
const getAgent = vi.fn((id: SessionId) => {
if (options.parentLive !== false && id === PARENT) return parent
if (id === CHILD) return child
return undefined
})
const listChildren = vi.fn(() => options.listError === undefined
? Promise.resolve(options.entries ?? [
{
kind: 'child', id: CHILD, mode: 'continuable', label: 'worker',
activity: 'inactive', hasChildren: false,
},
])
: Promise.reject(options.listError))
const followup = vi.fn((
_parent: unknown,
_childId: SessionId,
_content: unknown,
_delivery: {
source: { kind: string; rpcId: RpcId; clientTimeZone?: string }
signal: AbortSignal
},
) => options.followupError === undefined
? Promise.resolve('message-1')
: Promise.reject(options.followupError))
const interrupt = vi.fn((
_targetSessionId: SessionId,
_authority: { kind: 'user'; parentSessionId: SessionId },
) => {
if (options.interruptError !== undefined) throw options.interruptError
})
const ctx = new Context()
ctx.provide('agents', { get: getAgent })
ctx.provide('subagents', { listChildren, followup, interrupt })
const api = createApiProxy(ctx, {
defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp',
})
return { api, getAgent, listChildren, followup, interrupt, parent }
}
describe('subagent gateway', () => {
it('lists the complete catalog and reports exact live-parent availability', async () => {
const { api, listChildren } = bench({ parentLive: false, entries: [
{
kind: 'child', id: CHILD, mode: 'continuable', label: 'worker',
activity: 'inactive', hasChildren: true,
},
{
kind: 'child', id: sid('one-shot'), mode: 'one-shot',
activity: 'inactive', hasChildren: false,
},
{ kind: 'diagnostic', id: sid('bad'), reason: 'corrupt' },
] })
const response = await api.subagents.list(request({ parentSessionId: PARENT }))
expect(response.rpcId).toBe('subagent-rpc')
expect(response.result).toMatchObject({
ok: true,
value: {
parentAvailable: false,
entries: [
{ kind: 'child', mode: 'continuable' },
{ kind: 'child', mode: 'one-shot' },
{ kind: 'diagnostic' },
],
},
})
expect(listChildren).toHaveBeenCalledWith(PARENT, undefined)
})
it('derives catalog activity from the live child Agent rather than Session residency', async () => {
const residentIdle = bench({ childStatus: 'idle', entries: [{
kind: 'child', id: CHILD, mode: 'continuable', label: 'worker',
activity: 'running', hasChildren: false,
}] })
expect((await residentIdle.api.subagents.list(request({ parentSessionId: PARENT }))).result)
.toMatchObject({ ok: true, value: { entries: [{ activity: 'inactive' }] } })
const running = bench({ childStatus: 'running' })
expect((await running.api.subagents.list(request({ parentSessionId: PARENT }))).result)
.toMatchObject({ ok: true, value: { entries: [{ activity: 'running' }] } })
})
it('maps missing catalog projections on list without preflighting prompt delivery', async () => {
const listError = () => new SubagentError(
'listing subagents requires the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE',
)
const expected = {
code: 'internal',
message: 'subagent catalog is unavailable: this deployment does not mount the sessionProjections registry (load @deepseek-ai/dsh-session-projection)',
}
const list = bench({ listError: listError() })
expect((await list.api.subagents.list(request({ parentSessionId: PARENT }))).result)
.toMatchObject({ ok: false, error: expected })
const prompt = bench({ listError: listError() })
expect((await prompt.api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content: [],
}), new AbortController().signal)).result).toMatchObject({ ok: true })
expect(prompt.listChildren).not.toHaveBeenCalled()
expect(prompt.followup).toHaveBeenCalledOnce()
})
it('routes human content through the exact live parent with rpc attribution', async () => {
const { api, parent, followup } = bench()
const content = [{ type: 'text' as const, text: '继续' }]
const signal = new AbortController().signal
const response = await api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content,
}), signal)
expect(response.result).toMatchObject({
ok: true, value: { messageId: 'message-1' },
})
expect(followup).toHaveBeenCalledWith(
parent,
CHILD,
content,
{ source: { kind: 'user', rpcId: RpcId('subagent-rpc') }, signal },
)
})
it('canonicalizes browser-zone provenance before delivering a child prompt', async () => {
const { api, parent, followup } = bench()
const alias = 'US/Pacific'
const canonical = new Intl.DateTimeFormat('en-US', { timeZone: alias })
.resolvedOptions().timeZone
const content = [{ type: 'text' as const, text: 'continue locally' }]
const signal = new AbortController().signal
await expect(api.subagents.prompt(request({
parentSessionId: PARENT,
childSessionId: CHILD,
mode: 'continuable',
content,
clientTimeZone: alias,
}), signal)).resolves.toMatchObject({ result: { ok: true } })
expect(followup).toHaveBeenCalledWith(parent, CHILD, content, {
source: { kind: 'user', rpcId: RpcId('subagent-rpc'), clientTimeZone: canonical },
signal,
})
const invalid = await api.subagents.prompt(request({
parentSessionId: PARENT,
childSessionId: CHILD,
mode: 'continuable',
content,
clientTimeZone: 'Not/A_Real_Zone',
}), signal)
expect(invalid.result).toEqual({
ok: false,
error: {
code: 'invalid-time-zone',
message: 'clientTimeZone must be UTC or a valid IANA Area/Location name',
details: { value: 'Not/A_Real_Zone' },
},
})
expect(followup).toHaveBeenCalledOnce()
})
it('fails before delivery when the parent is absent and maps continuation failures', async () => {
const absent = bench({ parentLive: false })
expect((await absent.api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content: [],
}), new AbortController().signal)).result).toMatchObject({
ok: false, error: { code: 'subagent-parent-unavailable' },
})
expect(absent.listChildren).not.toHaveBeenCalled()
const failed = bench({ followupError: new SubagentError('draining', 'DRAINING') })
expect((await failed.api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content: [],
}), new AbortController().signal)).result).toMatchObject({
ok: false, error: { code: 'subagent-delivery-unavailable' },
})
})
it('hides unexpected backend details', async () => {
const catalog = bench({ listError: new Error('secret descriptor') })
expect((await catalog.api.subagents.list(request({
parentSessionId: PARENT,
}))).result).toMatchObject({
ok: false,
error: { code: 'internal', message: 'subagent catalog read failed' },
})
const prompt = bench({ followupError: new Error('secret provider') })
expect((await prompt.api.subagents.prompt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable', content: [],
}), new AbortController().signal)).result).toMatchObject({
ok: false,
error: { code: 'internal', message: 'subagent prompt failed' },
})
})
it('interrupts through the core primitive alone while the parent Agent is offline', async () => {
const { api, interrupt, getAgent, listChildren } = bench({ parentLive: false })
const response = await api.subagents.interrupt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const,
}))
expect(response.rpcId).toBe('subagent-rpc')
expect(response.result).toEqual({ ok: true, value: { accepted: true } })
expect(interrupt).toHaveBeenCalledExactlyOnceWith(CHILD, { kind: 'user', parentSessionId: PARENT })
// No parent-registry or catalog dependency: this is what keeps a
// live child interruptible after its parent Agent went offline.
expect(getAgent).not.toHaveBeenCalled()
expect(listChildren).not.toHaveBeenCalled()
})
it('maps interrupt authorization rejection without touching other services', async () => {
const { api, listChildren } = bench({
interruptError: new SubagentError('secret lineage', 'UNAUTHORIZED'),
})
const response = await api.subagents.interrupt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const,
}))
expect(response.result).toEqual({
ok: false,
error: {
code: 'subagent-unauthorized',
message: 'subagent does not belong to this parent',
details: { childSessionId: CHILD },
},
})
expect(listChildren).not.toHaveBeenCalled()
})
it('hides unexpected interrupt failures behind the internal code', async () => {
const { api } = bench({ interruptError: new Error('secret activation state') })
const response = await api.subagents.interrupt(request({
parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const,
}))
expect(response.result).toEqual({
ok: false,
error: { code: 'internal', message: 'subagent interrupt failed', details: {} },
})
})
})
packages/host/apiproxy/tests/client-handler.spec.ts
-36
View File
@@ -6,19 +6,15 @@
*/
import { describe, expect, it, vi } from 'vitest'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { ApiProxy, RpcMessage, RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy'
import { InProcessApiClient, RpcId, toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
const sid = (id: string): SessionId => id as SessionId
function ok<T>(request: RpcRequest<unknown>, value: T): Promise<RpcResponse<T>> {
return Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value } })
}
/** Scripted impl: every method resolves an empty-ish OK unless a case overrides it. */
function scriptedApi(overrides: {
subagents?: Partial<ApiProxy['subagents']>
host?: Partial<ApiProxy['host']>
skills?: Partial<ApiProxy['skills']>
agentPresets?: Partial<ApiProxy['agentPresets']>
@@ -29,12 +25,6 @@ function scriptedApi(overrides: {
const err = <T>(r: RpcRequest<unknown>): Promise<RpcResponse<T>> =>
Promise.resolve({ rpcId: r.rpcId, result: { ok: false, error: { code: 'internal' as const, message: 'stub', details: {} } } })
return {
subagents: {
list: r => ok(r, { entries: [], parentAvailable: false }),
prompt: r => ok(r, { messageId: 'message-1' as never }),
interrupt: r => ok(r, { accepted: true as const }),
...overrides.subagents,
},
host: {
describe: r => ok(r, {
version: '0-test', cwd: '/t', attachedSessions: 0, home: '/h', canOpenPath: true,
@@ -140,32 +130,6 @@ describe('unary round trip', () => {
await expect(client(api).host.describe({})).rejects.toThrow(/rpcId mismatch/)
})
it('round-trips subagent.interrupt and rejects a one-shot or incomplete address', async () => {
const interrupt = vi.fn((r: RpcRequest<unknown>) => ok(r, { accepted: true as const }))
const api = scriptedApi({ subagents: { interrupt } })
const c = client(api)
const accepted = await c.subagents.interrupt({
parentSessionId: sid('parent'), childSessionId: sid('child'), mode: 'continuable',
})
expect(accepted.result).toEqual({ ok: true, value: { accepted: true } })
expect(interrupt).toHaveBeenCalledTimes(1)
// The wire schema owns the mode fence: a one-shot address never reaches the impl.
const oneShot = await c.subagents.interrupt({
parentSessionId: sid('parent'), childSessionId: sid('child'), mode: 'one-shot',
} as never)
expect(oneShot.result.ok).toBe(false)
if (!oneShot.result.ok) expect(oneShot.result.error.code).toBe('bad-request')
const incomplete = await c.subagents.interrupt({
parentSessionId: sid('parent'), mode: 'continuable',
} as never)
expect(incomplete.result.ok).toBe(false)
if (!incomplete.result.ok) expect(incomplete.result.error.code).toBe('bad-request')
expect(interrupt).toHaveBeenCalledTimes(1)
})
it('rejects a method/path mismatch as bad-request', async () => {
const handler = toFetchHandler(scriptedApi())
const body = { type: 'client-request', rpcId: 'r1', method: 'host.describe', payload: {} }
packages/host/apiproxy/tests/fetch-carrier.spec.ts
-70
View File
@@ -7,31 +7,6 @@ import { AbstractApiClient, InProcessApiClient } from '../src/fetch/client.ts'
/** Minimal in-memory ApiProxy that echoes rpcIds. */
function fakeApi(overrides: Partial<{ crashOn: string }> = {}): ApiProxy {
return {
subagents: {
async list(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { entries: [], parentAvailable: false } } }
},
async prompt(request, signal) {
if (request.payload.content.some(block => block.type === 'text' && block.text === 'hang')) {
if (!signal.aborted) {
await new Promise<void>((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}
return {
rpcId: request.rpcId,
result: { ok: false, error: { code: 'cancelled' as const, message: 'aborted', details: {} } },
}
}
return {
rpcId: request.rpcId,
result: { ok: true, value: { messageId: 'message-1' as never } },
}
},
async interrupt(request) {
return { rpcId: request.rpcId, result: { ok: true, value: { accepted: true as const } } }
},
},
host: {
async describe(request) {
if (overrides.crashOn === 'host.describe') throw new Error('impl crashed')
@@ -221,23 +196,6 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
}
})
it('round-trips the subagent domain through the wire form', async () => {
const c = client()
expect((await c.subagents.list({ parentSessionId: 'parent' as never })).result)
.toEqual({ ok: true, value: { entries: [], parentAvailable: false } })
expect((await c.subagents.prompt({
parentSessionId: 'parent' as never,
childSessionId: 'child' as never,
mode: 'continuable',
content: [],
})).result).toEqual({ ok: true, value: { messageId: 'message-1' } })
expect((await c.subagents.interrupt({
parentSessionId: 'parent' as never,
childSessionId: 'child' as never,
mode: 'continuable',
})).result).toEqual({ ok: true, value: { accepted: true } })
})
it('keeps caller and connection aborts on a deadline-exempt unary', async () => {
const api = fakeApi()
const started = Promise.withResolvers<AbortSignal>()
@@ -263,34 +221,6 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
expect(handlerSignal.aborted).toBe(true)
})
it('propagates the carrier Request signal into subagent.prompt', async () => {
const handler = toFetchHandler(fakeApi())
const controller = new AbortController()
const body = JSON.stringify({
type: 'client-request',
rpcId: 'r-subagent-sig',
method: 'subagent.prompt',
payload: {
parentSessionId: 'parent',
childSessionId: 'child',
mode: 'continuable',
content: [{ type: 'text', text: 'hang' }],
},
})
const pending = handler.fetch(new Request(
'http://x/api/subagent.prompt',
{ method: 'POST', headers: { 'content-type': 'application/json' }, body, signal: controller.signal },
))
controller.abort()
const response = await pending
const parsed = await response.json() as {
rpcId: string
result: { error?: { code: string } }
}
expect(parsed.rpcId).toBe('r-subagent-sig')
expect(parsed.result.error?.code).toBe('cancelled')
})
it('propagates the carrier Request signal into host.pickDirectory', async () => {
const api = fakeApi()
api.host.pickDirectory = async (request, signal) => {
packages/host/apiproxy/tests/rpc-schemas.spec.ts
-25
View File
@@ -12,7 +12,6 @@ import {
} from '../src/api/host.schema.ts'
import { skillEntrySchema, skillListRequestSchema, skillListValueSchema } from '../src/api/skills.schema.ts'
import { agentPresetOpenDocumentValueSchema } from '../src/api/agent-presets.schema.ts'
import { subagentPromptRequestSchema } from '../src/api/subagents.schema.ts'
describe('RpcId', () => {
it('brands a raw string at zero runtime cost', () => {
@@ -51,12 +50,6 @@ describe('rpcErrorSchema', () => {
// The credentials producer still emits this code, so the branch has to stay.
expect(rpcErrorSchema.parse({ code: 'credential-rejected', message: 'm', details: { ref: 'r' } }).code).toBe('credential-rejected')
expect(rpcErrorSchema.parse({ code: 'model-discovery-failed', message: 'm', details: { settingsNs: 'n' } }).code).toBe('model-discovery-failed')
expect(rpcErrorSchema.parse({ code: 'subagent-parent-unavailable', message: 'm', details: { parentSessionId: 'p' } }).code).toBe('subagent-parent-unavailable')
expect(rpcErrorSchema.parse({ code: 'subagent-not-found', message: 'm', details: { parentSessionId: 'p', childSessionId: 'c' } }).code).toBe('subagent-not-found')
expect(rpcErrorSchema.parse({ code: 'subagent-catalog-diagnostic', message: 'm', details: { parentSessionId: 'p', childSessionId: 'c', reason: 'corrupt' } }).code).toBe('subagent-catalog-diagnostic')
expect(rpcErrorSchema.parse({ code: 'subagent-not-resumable', message: 'm', details: { childSessionId: 'c' } }).code).toBe('subagent-not-resumable')
expect(rpcErrorSchema.parse({ code: 'subagent-unauthorized', message: 'm', details: { childSessionId: 'c' } }).code).toBe('subagent-unauthorized')
expect(rpcErrorSchema.parse({ code: 'subagent-delivery-unavailable', message: 'm', details: { childSessionId: 'c' } }).code).toBe('subagent-delivery-unavailable')
expect(rpcErrorSchema.parse({ code: 'internal', message: 'm', details: {} }).code).toBe('internal')
})
@@ -99,24 +92,6 @@ describe('wire full-form schemas', () => {
})
})
describe('subagent domain schemas', () => {
it('carries optional request-local browser-zone provenance on prompts', () => {
expect(subagentPromptRequestSchema.parse({
parentSessionId: 'parent',
childSessionId: 'child',
mode: 'continuable',
content: [{ type: 'text', text: 'continue' }],
clientTimeZone: 'Asia/Shanghai',
}).clientTimeZone).toBe('Asia/Shanghai')
expect(subagentPromptRequestSchema.parse({
parentSessionId: 'parent',
childSessionId: 'child',
mode: 'continuable',
content: [],
}).clientTimeZone).toBeUndefined()
})
})
describe('host domain schemas', () => {
it('validates describe request/value', () => {
expect(hostDescribeRequestSchema.parse({})).toEqual({})
packages/subagent/subagent/tests/control.spec.ts
+301
View File
@@ -0,0 +1,301 @@
// The browser-facing control surface: catalog assembly against the live Agent
// registry, prompt admission, and the stable failure codes each answers with.
// The durable listing, continuation, and interrupt primitives they wrap have
// their own specs, so each case scripts them.
import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import type { MessageId } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import SubagentRuntime, {
SubagentError,
type SubagentListEntry,
type SubagentPromptRequestId,
} from '@deepseek-ai/dsh-subagent'
const PARENT = SessionId('parent')
const CHILD = SessionId('child')
const OTHER = SessionId('other')
const BROKEN = SessionId('broken')
const REQUEST_ID = 'req-1' as SubagentPromptRequestId
const signal = new AbortController().signal
/** The runtime plus a programmable live-Agent registry, omitted to compose none. */
async function bench(live?: Record<string, { status: 'running' | 'idle' }>) {
const ctx = new Context()
await ctx.plugin(SubagentRuntime)
if (live !== undefined) {
ctx.provide('agents', { get: (id: SessionId) => live[id] } as never)
}
return { ctx, subagents: ctx.subagents }
}
function childRow(id: SessionId, activity: 'running' | 'inactive'): SubagentListEntry {
return { kind: 'child', id, mode: 'continuable', label: 'worker', activity, hasChildren: false }
}
function promptRequest(clientTimeZone?: string) {
return {
requestId: REQUEST_ID,
parentSessionId: PARENT,
childSessionId: CHILD,
mode: 'continuable' as const,
content: [{ type: 'text' as const, text: 'continue' }],
...clientTimeZone === undefined ? {} : { clientTimeZone },
}
}
function emptyIdFailure(method: string, field: string) {
return {
code: 'bad-request',
message: `invalid payload for ${method}`,
details: {
issues: [{
origin: 'string',
code: 'too_small',
minimum: 1,
inclusive: true,
path: [field],
message: 'Too small: expected string to have >=1 characters',
}],
},
}
}
describe('subagent catalog Remote', () => {
it('rejects an empty parent id before listing', async () => {
const { subagents } = await bench()
const listChildren = vi.spyOn(subagents, 'listChildren')
await expect(subagents.remoteExportList(SessionId(''), signal))
.rejects.toMatchObject({ failure: emptyIdFailure('subagent.list', 'parentSessionId') })
expect(listChildren).not.toHaveBeenCalled()
})
it('samples row activity from the live Agent driver and reports parent availability', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' }, [CHILD]: { status: 'running' } })
vi.spyOn(subagents, 'listChildren').mockResolvedValue([
// The durable listing reports store presence; the browser row reports the driver.
childRow(CHILD, 'inactive'),
childRow(OTHER, 'running'),
{ kind: 'diagnostic', id: BROKEN, reason: 'corrupt' },
])
await expect(subagents.remoteExportList(PARENT, signal)).resolves.toEqual({
entries: [
childRow(CHILD, 'running'),
childRow(OTHER, 'inactive'),
{ kind: 'diagnostic', id: BROKEN, reason: 'corrupt' },
],
parentAvailable: true,
})
})
it('reports every row inactive and the parent unavailable without an Agent registry', async () => {
const { subagents } = await bench()
vi.spyOn(subagents, 'listChildren').mockResolvedValue([childRow(CHILD, 'running')])
await expect(subagents.remoteExportList(PARENT, signal)).resolves.toEqual({
entries: [childRow(CHILD, 'inactive')],
parentAvailable: false,
})
})
it('reports an unknown parent as unavailable while the registry serves other sessions', async () => {
const { subagents } = await bench({ [CHILD]: { status: 'running' } })
vi.spyOn(subagents, 'listChildren').mockResolvedValue([])
await expect(subagents.remoteExportList(PARENT, signal))
.resolves.toEqual({ entries: [], parentAvailable: false })
})
it('separates cancellation, the missing projections capability, and an unexplained read failure', async () => {
const { subagents } = await bench()
const listChildren = vi.spyOn(subagents, 'listChildren')
const aborted = new AbortController()
aborted.abort()
listChildren.mockRejectedValue(new Error('read stopped'))
await expect(subagents.remoteExportList(PARENT, aborted.signal))
.rejects.toMatchObject({ failure: { code: 'cancelled' } })
listChildren.mockRejectedValue(new SubagentError('cancelled', 'CANCELLED'))
await expect(subagents.remoteExportList(PARENT, signal))
.rejects.toMatchObject({ failure: { code: 'cancelled' } })
listChildren.mockRejectedValue(
new SubagentError('no registry', 'SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE'),
)
await expect(subagents.remoteExportList(PARENT, signal)).rejects.toMatchObject({
failure: {
code: 'subagent-projections-unavailable',
message: expect.stringContaining('sessionProjections') as unknown as string,
},
})
listChildren.mockRejectedValue(new Error('disk gone'))
await expect(subagents.remoteExportList(PARENT, signal))
.rejects.toMatchObject({ failure: { code: 'internal', message: 'subagent catalog read failed' } })
})
})
describe('subagent prompt Remote', () => {
it('rejects empty parent and child ids before delivery', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const followup = vi.spyOn(subagents, 'followup')
const cases: readonly {
readonly field: 'parentSessionId' | 'childSessionId'
readonly request: ReturnType<typeof promptRequest>
}[] = [
{ field: 'parentSessionId', request: { ...promptRequest(), parentSessionId: SessionId('') } },
{ field: 'childSessionId', request: { ...promptRequest(), childSessionId: SessionId('') } },
]
for (const { field, request } of cases) {
await expect(subagents.prompt(request, signal))
.rejects.toMatchObject({ failure: emptyIdFailure('subagent.prompt', field) })
}
expect(followup).not.toHaveBeenCalled()
})
it('forwards non-text content blocks without narrowing them', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const followup = vi.spyOn(subagents, 'followup').mockResolvedValue('m-content' as MessageId)
const content = [{ type: 'reasoning' as const, text: 'retain this block' }]
await expect(subagents.prompt({ ...promptRequest(), content }, signal))
.resolves.toEqual({ messageId: 'm-content' })
expect(followup.mock.calls[0]?.[2]).toEqual(content)
})
it('delivers the content under the caller-minted identity and canonical browser zone', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const followup = vi.spyOn(subagents, 'followup').mockResolvedValue('m-1' as MessageId)
await expect(subagents.prompt(promptRequest('Asia/Shanghai'), signal))
.resolves.toEqual({ messageId: 'm-1' })
expect(followup).toHaveBeenCalledWith(
{ status: 'idle' },
CHILD,
[{ type: 'text', text: 'continue' }],
{
source: { kind: 'user', rpcId: REQUEST_ID, clientTimeZone: 'Asia/Shanghai' },
signal,
},
)
})
it('omits the zone from the durable source when the browser reported none', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const followup = vi.spyOn(subagents, 'followup').mockResolvedValue('m-2' as MessageId)
await expect(subagents.prompt(promptRequest(), signal)).resolves.toEqual({ messageId: 'm-2' })
expect(followup.mock.calls[0]?.[3].source).toEqual({ kind: 'user', rpcId: REQUEST_ID })
})
it('accepts UTC and rejects an empty, untrimmed, malformed, or unknown zone', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
vi.spyOn(subagents, 'followup').mockResolvedValue('m-3' as MessageId)
await expect(subagents.prompt(promptRequest('UTC'), signal)).resolves.toEqual({ messageId: 'm-3' })
for (const zone of ['', ' UTC', 'Shanghai', 'Nowhere/Nowhere']) {
await expect(subagents.prompt(promptRequest(zone), signal)).rejects.toMatchObject({
failure: { code: 'invalid-time-zone', details: { value: zone } },
})
}
})
it('refuses delivery when the exact parent Agent is not live', async () => {
const { subagents } = await bench()
const followup = vi.spyOn(subagents, 'followup')
await expect(subagents.prompt(promptRequest(), signal)).rejects.toMatchObject({
failure: { code: 'subagent-parent-unavailable', details: { parentSessionId: PARENT } },
})
expect(followup).not.toHaveBeenCalled()
})
it('maps each admission failure onto its stable code and hides the rest', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const followup = vi.spyOn(subagents, 'followup')
const cases: readonly [string, string][] = [
['NOT_RESUMABLE', 'subagent-not-resumable'],
['UNAUTHORIZED', 'subagent-unauthorized'],
['DRAINING', 'subagent-delivery-unavailable'],
['ACTIVATION_CLOSING', 'subagent-delivery-unavailable'],
['NO_PROVIDER', 'internal'],
]
for (const [thrown, code] of cases) {
followup.mockRejectedValue(new SubagentError('refused', thrown))
await expect(subagents.prompt(promptRequest(), signal))
.rejects.toMatchObject({ failure: { code } })
}
followup.mockRejectedValue(new Error('inbox exploded'))
await expect(subagents.prompt(promptRequest(), signal))
.rejects.toMatchObject({ failure: { code: 'internal', message: 'subagent prompt failed' } })
})
it('answers a caller-cancelled delivery as cancelled rather than a failure', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
const aborted = new AbortController()
vi.spyOn(subagents, 'followup').mockImplementation(() => {
aborted.abort()
return Promise.reject(new SubagentError('gone', 'NOT_RESUMABLE'))
})
await expect(subagents.prompt(promptRequest(), aborted.signal))
.rejects.toMatchObject({ failure: { code: 'cancelled' } })
})
it('preserves a cancellation reported by the continuation operation', async () => {
const { subagents } = await bench({ [PARENT]: { status: 'idle' } })
vi.spyOn(subagents, 'followup')
.mockRejectedValue(new SubagentError('stopped', 'CANCELLED'))
await expect(subagents.prompt(promptRequest(), signal))
.rejects.toMatchObject({ failure: { code: 'cancelled' } })
})
})
describe('subagent interrupt Remote', () => {
it('rejects empty child and parent ids before interrupting', async () => {
const { subagents } = await bench()
const interrupt = vi.spyOn(subagents, 'interrupt')
for (const [childSessionId, parentSessionId] of [
[SessionId(''), PARENT],
[CHILD, SessionId('')],
] as const) {
const field = childSessionId.length === 0 ? 'childSessionId' : 'parentSessionId'
expect(() => subagents.interruptByParent(childSessionId, parentSessionId, 'continuable'))
.toThrow(expect.objectContaining({
failure: emptyIdFailure('subagent.interrupt', field),
}))
}
expect(interrupt).not.toHaveBeenCalled()
})
it('admits the parent-addressed interrupt and acknowledges it', async () => {
const { subagents } = await bench()
const interrupt = vi.spyOn(subagents, 'interrupt').mockReturnValue()
expect(subagents.interruptByParent(CHILD, PARENT, 'continuable')).toEqual({ accepted: true })
expect(interrupt).toHaveBeenCalledWith(CHILD, { kind: 'user', parentSessionId: PARENT })
})
it('answers a foreign address as unauthorized and everything else as internal', async () => {
const { subagents } = await bench()
const interrupt = vi.spyOn(subagents, 'interrupt')
interrupt.mockImplementation(() => { throw new SubagentError('not yours', 'UNAUTHORIZED') })
expect(() => subagents.interruptByParent(CHILD, PARENT, 'continuable')).toThrow(
expect.objectContaining({ failure: { code: 'subagent-unauthorized', message: expect.any(String) as unknown as string, details: { childSessionId: CHILD } } }),
)
interrupt.mockImplementation(() => { throw new Error('boom') })
expect(() => subagents.interruptByParent(CHILD, PARENT, 'continuable')).toThrow(
expect.objectContaining({ failure: { code: 'internal', message: 'subagent interrupt failed', details: {} } }),
)
})
})
packages/test-support/client-runtime/package.json
+2
View File
@@ -45,6 +45,7 @@
"@deepseek-ai/dsh-client-ui-slots": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/cordis": "workspace:^",
"react": "^18.2.0",
"react-dom": "^18.2.0"
@@ -63,6 +64,7 @@
"@deepseek-ai/dsh-client-ui-slots": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",
"@types/react": "~18.3.1",
"@types/react-dom": "~18.3.0",
"@deepseek-ai/cordis": "workspace:^",
packages/test-support/client-runtime/src/sessions.ts
+1 -1
View File
@@ -9,7 +9,7 @@ import type {
SessionEventLikeEntry, SessionLiveEventEntry, SessionSearchResultItem,
SessionSnapshot, SessionSummary,
} from '@deepseek-ai/dsh-api-session-controller/client'
import type { SubagentAddress } from '@deepseek-ai/dsh-client-connection/client'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
import { createSnapshotStore } from '@deepseek-ai/dsh-client-store'
import type { ObservableSnapshot, SnapshotStore } from '@deepseek-ai/dsh-client-store'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
packages/test-support/client-runtime/tsconfig.json
+3
View File
@@ -47,6 +47,9 @@
{
"path": "../../client/ui-settings"
},
{
"path": "../../subagent/subagent"
},
{
"path": "../../runtime-diagnostics/invariants"
}