refactor(packages): dissolve ui/ and rename sdk/ to scaffold/
git mv per the regrouping RFC: the five human-collaboration seams and tui join packages/interaction/, app-boot becomes packages/boot/, and jsonrpc joins the renamed scaffold/ (formerly sdk/) as its server half beside client/protocol/create-sdk/helper/scripts/telemetry, whose folders drop the legacy sdk- prefix. Three new group README triplets replace the ui/ and sdk/ ones; tsconfig references/paths/globs, knip keys, vitest globs, gate scripts, catalogs, docs, and the lockfile follow. Adds the four settled FIXME rename markers (dsh-sdk-server, dsh-sdk-telemetry, dsh-sdk-helper, dsh-sdk-scripts). The scaffold folders diverge from their npm names until those renames land, so tsconfig.base.json maps the three affected names explicitly beside the group wildcard. Also repairs two pre-existing stale-path classes the strengthened sweep surfaced: docs/web-styling.md's retired web-ui host package and type-model spec fixture-literal joins. app-boot's three Loader-composition specs time out at the default 5s under full-suite parallel load on this filesystem (pre-existing; pass isolated with --testTimeout=30000); interaction/scaffold/boot suites otherwise green (687 passed).
This commit is contained in:
Tianyi Cui
@@ -0,0 +1,122 @@
|
||||
/**
|
||||
* Built-artifact guard for the scope carrier shared by `dsh-subagent` and
|
||||
* `dsh-jsonrpc`. The carrier registry is module-local, so both bundles must
|
||||
* externalize `dsh-scope`; source-mode tests cannot expose an accidentally
|
||||
* inlined second registry. This test runs the real `lib/index.js` bundles in a
|
||||
* plain Node subprocess, disposes the child before settlement, and requires the
|
||||
* SDK completion notification to retain the delegating parent.
|
||||
*/
|
||||
|
||||
import { execFile } from 'node:child_process'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { promisify } from 'node:util'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
|
||||
const jsonrpcBundle = fileURLToPath(new URL('../lib/index.js', import.meta.url))
|
||||
const execFileAsync = promisify(execFile)
|
||||
|
||||
const builtRuntimeProbe = String.raw`
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
|
||||
const load = (path) => import(pathToFileURL(resolve(path)).href);
|
||||
const [
|
||||
{ Context },
|
||||
agentCore,
|
||||
{ default: SubagentService },
|
||||
{ default: SessionPersistenceJsonl },
|
||||
{ HarnessSdkServer },
|
||||
{ SessionId },
|
||||
] = await Promise.all([
|
||||
load("vendor/cordis/lib/index.js"),
|
||||
load("packages/examples/agent-spine-demo/lib/index.js"),
|
||||
load("packages/subagent/subagent/lib/index.js"),
|
||||
load("packages/session/session-persistence-jsonl/lib/index.js"),
|
||||
load("packages/scaffold/server/lib/index.js"),
|
||||
load("packages/core/session/lib/index.js"),
|
||||
]);
|
||||
|
||||
const storageRoot = await mkdtemp(join(tmpdir(), "jsonrpc-built-scope-"));
|
||||
const ctx = new Context();
|
||||
try {
|
||||
await ctx.plugin(agentCore, { workspaceContext: false });
|
||||
await ctx.plugin(SubagentService);
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: storageRoot });
|
||||
await new Promise((ready) => setTimeout(ready, 50));
|
||||
|
||||
const notifications = [];
|
||||
const server = new HarnessSdkServer(ctx, {
|
||||
request() { return Promise.reject(new Error("unexpected host request")); },
|
||||
notify(method, params) { notifications.push({ method, params }); },
|
||||
});
|
||||
const parent = await ctx.agents.create({
|
||||
sessionId: SessionId("built-parent"),
|
||||
meta: { cwd: storageRoot },
|
||||
agentOptions: { model: "test" },
|
||||
});
|
||||
const child = await parent.agent.ctx.agents.create({
|
||||
sessionId: SessionId("built-child"),
|
||||
meta: { cwd: storageRoot, parentSession: SessionId("built-parent") },
|
||||
agentOptions: { model: "test" },
|
||||
});
|
||||
const result = Promise.withResolvers();
|
||||
const unregister = ctx.subagents.registerProvider({
|
||||
name: "built-local",
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start() {
|
||||
return Promise.resolve({
|
||||
id: child.agent.id,
|
||||
localAgent: child.agent,
|
||||
result: result.promise,
|
||||
dispose() { return Promise.resolve(); },
|
||||
});
|
||||
},
|
||||
});
|
||||
const run = await ctx.subagents.start("built-local", {
|
||||
parent: parent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
});
|
||||
await child.dispose();
|
||||
result.resolve({ output: [], stopReason: "completed" });
|
||||
await run.result;
|
||||
await Promise.resolve();
|
||||
|
||||
console.log(JSON.stringify(notifications.filter(({ method }) => method === "subagent.finished")));
|
||||
await run.dispose();
|
||||
unregister();
|
||||
await parent.dispose();
|
||||
await server.shutdown();
|
||||
} finally {
|
||||
await ctx.fiber.dispose();
|
||||
await rm(storageRoot, { recursive: true, force: true });
|
||||
}
|
||||
`
|
||||
|
||||
describe.skipIf(!existsSync(jsonrpcBundle))('dsh-jsonrpc BUILT scope carrier', () => {
|
||||
it('preserves parent-scoped completion after child disposal', async () => {
|
||||
const { stdout, stderr } = await execFileAsync(process.execPath, ['--input-type=module', '-e', builtRuntimeProbe], {
|
||||
cwd: repoRoot,
|
||||
timeout: 15_000,
|
||||
})
|
||||
|
||||
expect(stderr).not.toContain('listener threw')
|
||||
expect(JSON.parse(stdout) as unknown).toEqual([{
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'built-local',
|
||||
agentId: 'built-child',
|
||||
parentSessionId: 'built-parent',
|
||||
childSessionId: 'built-child',
|
||||
status: 'ok',
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [],
|
||||
},
|
||||
}])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,305 @@
|
||||
import { createServer } from 'node:http'
|
||||
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { PassThrough, Writable } from 'node:stream'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import * as jsonrpc from '../src/index.ts'
|
||||
|
||||
/**
|
||||
* Mount the real namespace plugin with in-memory stdio and exit seams. Covers
|
||||
* the full transport/server path, response-before-exit shutdown exactly once,
|
||||
* and bare-fiber disposal without process exit.
|
||||
*/
|
||||
|
||||
/** One ordered frame, write completion, or exit observation. */
|
||||
type WireEvent =
|
||||
| { kind: 'frame'; frame: Record<string, unknown> }
|
||||
| { kind: 'write-complete'; ids: (string | number)[] }
|
||||
| { kind: 'root-disposed' }
|
||||
| { kind: 'exit'; code: number }
|
||||
|
||||
interface ApplyHarness {
|
||||
ctx: Context
|
||||
/** The plugin fiber used by the bare-dispose case. */
|
||||
fiber: Awaited<ReturnType<Context['plugin']>>
|
||||
/** Frames, write completions, and exits in observation order. */
|
||||
events: WireEvent[]
|
||||
outputErrors: Error[]
|
||||
send(frame: Record<string, unknown>): void
|
||||
sendRaw(text: string): void
|
||||
frames(): Record<string, unknown>[]
|
||||
exits(): number[]
|
||||
waitForFrame(predicate: (frame: Record<string, unknown>) => boolean, description: string): Promise<Record<string, unknown>>
|
||||
dispose(): Promise<void>
|
||||
}
|
||||
|
||||
/** Poll asynchronous output for up to five seconds. */
|
||||
async function waitFor<T>(get: () => T | undefined, description: string): Promise<T> {
|
||||
const deadline = Date.now() + 5000
|
||||
for (;;) {
|
||||
const value = get()
|
||||
if (value !== undefined) return value
|
||||
if (Date.now() > deadline) throw new Error(`timed out waiting for ${description}`)
|
||||
await new Promise(resolve => setTimeout(resolve, 5))
|
||||
}
|
||||
}
|
||||
|
||||
/** Drain asynchronous work before a negative assertion. */
|
||||
async function settle(): Promise<void> {
|
||||
await new Promise(resolve => setTimeout(resolve, 25))
|
||||
}
|
||||
|
||||
/** Mount the real plugin on a minimal harness with in-memory stdio and exit. */
|
||||
async function mountPlugin(
|
||||
storageDir: string,
|
||||
options: { writeDelayMs?: number; failFlush?: boolean } = {},
|
||||
): Promise<ApplyHarness> {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(agentCore, { workspaceContext: false })
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: storageDir })
|
||||
await new Promise(resolve => setTimeout(resolve, 50))
|
||||
|
||||
const input = new PassThrough()
|
||||
const events: WireEvent[] = []
|
||||
const outputErrors: Error[] = []
|
||||
let pendingOutput = ''
|
||||
// Record frame admission separately from write completion so delayed output
|
||||
// tests the flush barrier.
|
||||
const output = new Writable({
|
||||
write(chunk: Buffer, _encoding, callback) {
|
||||
const ids: (string | number)[] = []
|
||||
pendingOutput += chunk.toString('utf8')
|
||||
for (;;) {
|
||||
const newline = pendingOutput.indexOf('\n')
|
||||
if (newline < 0) break
|
||||
const line = pendingOutput.slice(0, newline).trim()
|
||||
pendingOutput = pendingOutput.slice(newline + 1)
|
||||
if (line) {
|
||||
const frame = JSON.parse(line) as Record<string, unknown>
|
||||
events.push({ kind: 'frame', frame })
|
||||
if (typeof frame.id === 'string' || typeof frame.id === 'number') ids.push(frame.id)
|
||||
}
|
||||
}
|
||||
const complete = (): void => {
|
||||
if (options.failFlush === true && chunk.length === 0) {
|
||||
callback(new Error('flush callback failed'))
|
||||
return
|
||||
}
|
||||
events.push({ kind: 'write-complete', ids })
|
||||
callback()
|
||||
}
|
||||
if ((options.writeDelayMs ?? 0) > 0) setTimeout(complete, options.writeDelayMs)
|
||||
else complete()
|
||||
},
|
||||
})
|
||||
output.on('error', (error: Error) => { outputErrors.push(error) })
|
||||
const exit = (code: number): void => { events.push({ kind: 'exit', code }) }
|
||||
|
||||
ctx.effect(() => () => { events.push({ kind: 'root-disposed' }) }, 'jsonrpc test root-disposal witness')
|
||||
const fiber = await ctx.plugin(jsonrpc, { input, output, exit })
|
||||
|
||||
const frames = (): Record<string, unknown>[] =>
|
||||
events.flatMap(event => event.kind === 'frame' ? [event.frame] : [])
|
||||
return {
|
||||
ctx,
|
||||
fiber,
|
||||
events,
|
||||
outputErrors,
|
||||
send: (frame) => { input.write(`${JSON.stringify(frame)}\n`) },
|
||||
sendRaw: (text) => { input.write(text) },
|
||||
frames,
|
||||
exits: () => events.flatMap(event => event.kind === 'exit' ? [event.code] : []),
|
||||
waitForFrame: (predicate, description) => waitFor(() => frames().find(predicate), description),
|
||||
dispose: async () => { await ctx.fiber.dispose() },
|
||||
}
|
||||
}
|
||||
|
||||
const servers: Server[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve))))
|
||||
vi.unstubAllEnvs()
|
||||
})
|
||||
|
||||
/** Keyless SSE endpoint for completing a prompt turn. */
|
||||
async function mockCompletionServer(): Promise<{ url: string; requests: unknown[] }> {
|
||||
const requests: unknown[] = []
|
||||
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
|
||||
let body = ''
|
||||
request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') })
|
||||
request.on('end', () => {
|
||||
requests.push(JSON.parse(body))
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
response.write('data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}\n\n')
|
||||
response.write('data: {"choices":[{"delta":{"content":"done"}}]}\n\n')
|
||||
response.write('data: {"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}\n\n')
|
||||
response.write('data: [DONE]\n\n')
|
||||
response.end()
|
||||
})
|
||||
})
|
||||
servers.push(server)
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') throw new Error('no port')
|
||||
return { url: `http://127.0.0.1:${address.port}`, requests }
|
||||
}
|
||||
|
||||
describe('dsh-jsonrpc plugin apply', () => {
|
||||
it('serves initialize over the injected stdio pair', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-init-'))
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
const harness = await mountPlugin(storageDir)
|
||||
try {
|
||||
harness.send({ jsonrpc: '2.0', id: 'init-1', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek-official', model: 'apply-model' } })
|
||||
|
||||
const response = await harness.waitForFrame(frame => frame.id === 'init-1', 'initialize response')
|
||||
expect(response).toEqual({
|
||||
jsonrpc: '2.0',
|
||||
id: 'init-1',
|
||||
result: { serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } },
|
||||
})
|
||||
expect(harness.exits()).toEqual([])
|
||||
} finally {
|
||||
await harness.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('drives a session/prompt turn end-to-end and forwards session notifications as output frames', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-prompt-'))
|
||||
const llmServer = await mockCompletionServer()
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url)
|
||||
const harness = await mountPlugin(storageDir)
|
||||
try {
|
||||
harness.send({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { cwd: storageDir, provider: 'deepseek-official', model: 'dsagent-model' } })
|
||||
await harness.waitForFrame(frame => frame.id === 1, 'initialize response')
|
||||
|
||||
harness.send({
|
||||
jsonrpc: '2.0',
|
||||
id: 2,
|
||||
method: 'session/prompt',
|
||||
params: { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }] },
|
||||
})
|
||||
const response = await harness.waitForFrame(frame => frame.id === 2, 'prompt response')
|
||||
expect((response.result as { messageId?: unknown }).messageId).toBeTypeOf('string')
|
||||
await harness.waitForFrame(
|
||||
frame => frame.method === 'session.status'
|
||||
&& (frame.params as { status?: string } | undefined)?.status === 'idle',
|
||||
'idle session status',
|
||||
)
|
||||
|
||||
expect(llmServer.requests).toHaveLength(1)
|
||||
const body = llmServer.requests[0] as { model: string; messages: { role: string }[] }
|
||||
expect(body.model).toBe('dsagent-model')
|
||||
expect(body.messages.at(-1)?.role).toBe('user')
|
||||
|
||||
// Notifications use the same transport and arrive as id-less frames.
|
||||
const notifications = harness.frames().filter(frame => frame.id === undefined)
|
||||
expect(notifications.some(frame => frame.method === 'session.event')).toBe(true)
|
||||
expect(notifications.findLast(frame => frame.method === 'session.status')).toMatchObject({
|
||||
jsonrpc: '2.0',
|
||||
params: { sessionId: 'main', status: 'idle' },
|
||||
})
|
||||
} finally {
|
||||
await harness.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('answers shutdown before exiting 0 exactly once, even against a racing second shutdown', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-shutdown-'))
|
||||
const harness = await mountPlugin(storageDir, { writeDelayMs: 10 })
|
||||
try {
|
||||
// One chunk makes the two deferred exit callbacks race.
|
||||
const first = { jsonrpc: '2.0', id: 'sd-1', method: 'shutdown' }
|
||||
const second = { jsonrpc: '2.0', id: 'sd-2', method: 'shutdown' }
|
||||
harness.sendRaw(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`)
|
||||
|
||||
await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit recorder call')
|
||||
expect(harness.exits()).toEqual([0])
|
||||
|
||||
// Both response writes and the flush barrier complete before exit.
|
||||
const exitIndex = harness.events.findIndex(event => event.kind === 'exit')
|
||||
const firstResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-1')
|
||||
const secondResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-2')
|
||||
const firstComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.includes('sd-1'))
|
||||
const secondComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.includes('sd-2'))
|
||||
const flushComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.length === 0)
|
||||
const rootDisposed = harness.events.findIndex(event => event.kind === 'root-disposed')
|
||||
expect(firstResponse).toBeGreaterThanOrEqual(0)
|
||||
expect(secondResponse).toBeGreaterThanOrEqual(0)
|
||||
expect(firstComplete).toBeGreaterThan(firstResponse)
|
||||
expect(secondComplete).toBeGreaterThan(secondResponse)
|
||||
expect(flushComplete).toBeGreaterThan(firstComplete)
|
||||
expect(flushComplete).toBeGreaterThan(secondComplete)
|
||||
expect(rootDisposed).toBeGreaterThan(flushComplete)
|
||||
expect(exitIndex).toBeGreaterThan(rootDisposed)
|
||||
|
||||
await settle()
|
||||
expect(harness.exits()).toEqual([0])
|
||||
expect(harness.events.filter(event => event.kind === 'root-disposed')).toHaveLength(1)
|
||||
|
||||
const before = harness.frames().length
|
||||
harness.send({ jsonrpc: '2.0', id: 'after-exit', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek-official', model: 'x' } })
|
||||
await settle()
|
||||
expect(harness.frames().length).toBe(before)
|
||||
} finally {
|
||||
await harness.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('still disposes and exits once when the flush callback fails', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-flush-failure-'))
|
||||
const harness = await mountPlugin(storageDir, { failFlush: true })
|
||||
try {
|
||||
harness.send({ jsonrpc: '2.0', id: 'sd-fail', method: 'shutdown' })
|
||||
|
||||
await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit after flush failure')
|
||||
await settle()
|
||||
expect(harness.exits()).toEqual([0])
|
||||
expect(harness.events.filter(event => event.kind === 'root-disposed')).toHaveLength(1)
|
||||
expect(harness.outputErrors.map(error => error.message)).toEqual(['flush callback failed'])
|
||||
|
||||
const before = harness.frames().length
|
||||
harness.send({ jsonrpc: '2.0', id: 'after-flush-failure', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek-official', model: 'x' } })
|
||||
await settle()
|
||||
expect(harness.frames().length).toBe(before)
|
||||
} finally {
|
||||
await harness.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('stops serving on a bare fiber dispose (HMR-style unload) without calling exit', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-dispose-'))
|
||||
const harness = await mountPlugin(storageDir)
|
||||
try {
|
||||
// Prove the handler-rejection path is live before disposal.
|
||||
harness.send({ jsonrpc: '2.0', id: 'probe-1', method: 'nope/unknown' })
|
||||
const error = await harness.waitForFrame(frame => frame.id === 'probe-1', 'error response for unknown method')
|
||||
expect(error.error).toMatchObject({
|
||||
code: -32603,
|
||||
message: 'unknown DeepSeek Harness SDK runtime method: nope/unknown',
|
||||
})
|
||||
|
||||
await harness.fiber.dispose()
|
||||
expect(harness.events.some(event => event.kind === 'root-disposed')).toBe(false)
|
||||
|
||||
const before = harness.frames().length
|
||||
harness.send({ jsonrpc: '2.0', id: 'probe-2', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek-official', model: 'x' } })
|
||||
await settle()
|
||||
expect(harness.frames().length).toBe(before)
|
||||
expect(harness.exits()).toEqual([])
|
||||
} finally {
|
||||
await harness.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,22 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import Loader from '@cordisjs/plugin-loader'
|
||||
import * as jsonrpc from '../src/index.ts'
|
||||
|
||||
/**
|
||||
* Run the real namespace export through `Loader.unwrapExports`; a stray
|
||||
* default would discard `name`, `inject`, `Config`, and `apply`.
|
||||
*/
|
||||
describe('dsh-jsonrpc plugin export shape', () => {
|
||||
it('has the namespace-plugin export shape (no stray default) so the Loader keeps name/inject/Config/apply', () => {
|
||||
expect('default' in jsonrpc).toBe(false)
|
||||
expect(typeof jsonrpc.apply).toBe('function')
|
||||
|
||||
const loader = Object.create(Loader.prototype) as Loader
|
||||
const unwrapped = loader.unwrapExports(jsonrpc) as Record<string, unknown>
|
||||
expect(unwrapped).toBe(jsonrpc)
|
||||
expect(unwrapped.name).toBe('jsonrpc')
|
||||
expect(unwrapped.inject).toEqual(['agents'])
|
||||
expect(unwrapped.Config).toBeDefined()
|
||||
expect(typeof unwrapped.apply).toBe('function')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,967 @@
|
||||
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
import { createServer } from 'node:http'
|
||||
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import AgentRegistry, { type Agent, type AgentHandle } from '@deepseek-ai/dsh-agent'
|
||||
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import SubagentService, { type SubagentResult, type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
|
||||
import type { JsonRpcTransportPeer } from '@deepseek-ai/dsh-sdk-protocol'
|
||||
import { HarnessSdkServer } from '../src/index.ts'
|
||||
|
||||
class FakeTransport implements JsonRpcTransportPeer {
|
||||
notifications: { method: string; params?: Record<string, unknown> }[] = []
|
||||
|
||||
async request(method: string, params: object): Promise<unknown> {
|
||||
throw new Error(`the SDK server should not call host JSON-RPC method ${method} with ${JSON.stringify(params)}`)
|
||||
}
|
||||
|
||||
notify(method: string, params?: object): void {
|
||||
this.notifications.push(params === undefined ? { method } : { method, params: params as Record<string, unknown> })
|
||||
}
|
||||
}
|
||||
|
||||
const servers: Server[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve))))
|
||||
vi.unstubAllEnvs()
|
||||
})
|
||||
|
||||
async function mockCompletionServer(): Promise<{ url: string; requests: unknown[]; headers: IncomingMessage['headers'][] }> {
|
||||
const requests: unknown[] = []
|
||||
const headers: IncomingMessage['headers'][] = []
|
||||
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
|
||||
let body = ''
|
||||
request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') })
|
||||
request.on('end', () => {
|
||||
requests.push(JSON.parse(body))
|
||||
headers.push(request.headers)
|
||||
response.writeHead(200, { 'content-type': 'text/event-stream' })
|
||||
response.write('data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}\n\n')
|
||||
response.write('data: {"choices":[{"delta":{"content":"done"}}]}\n\n')
|
||||
response.write('data: {"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}\n\n')
|
||||
response.write('data: [DONE]\n\n')
|
||||
response.end()
|
||||
})
|
||||
})
|
||||
servers.push(server)
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') throw new Error('no port')
|
||||
return { url: `http://127.0.0.1:${address.port}`, requests, headers }
|
||||
}
|
||||
|
||||
async function makeHarness(storageDir: string) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(agentCore, { workspaceContext: false })
|
||||
await ctx.plugin(SubagentService)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root: storageDir })
|
||||
await new Promise(resolve => setTimeout(resolve, 50))
|
||||
return ctx
|
||||
}
|
||||
|
||||
/** Drive the owning service so test lifecycle events carry the real parent scope. */
|
||||
async function settleSubagent(
|
||||
ctx: Context,
|
||||
parent: Agent,
|
||||
info: Omit<SubagentRunEndInfo, 'runId' | 'local'> & { localAgent: Agent | undefined },
|
||||
beforeSettle?: () => Promise<void>,
|
||||
): Promise<void> {
|
||||
const result = Promise.withResolvers<SubagentResult>()
|
||||
const disposeProvider = ctx.subagents.registerProvider({
|
||||
name: info.provider,
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: false,
|
||||
async start() {
|
||||
return {
|
||||
id: info.id,
|
||||
localAgent: info.localAgent,
|
||||
result: result.promise,
|
||||
dispose: () => Promise.resolve(),
|
||||
}
|
||||
},
|
||||
})
|
||||
try {
|
||||
const run = await ctx.subagents.start(info.provider, {
|
||||
parent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
await beforeSettle?.()
|
||||
if (info.lastAssistantMessage === undefined) {
|
||||
result.reject(new Error('synthetic infrastructure failure'))
|
||||
} else {
|
||||
result.resolve({ output: info.lastAssistantMessage, stopReason: info.stopReason })
|
||||
}
|
||||
await run.result.then(() => undefined, () => undefined)
|
||||
await run.dispose()
|
||||
} finally {
|
||||
disposeProvider()
|
||||
}
|
||||
}
|
||||
|
||||
describe('HarnessSdkServer', () => {
|
||||
it('creates a harness agent and calls the configured OpenAI-compatible endpoint', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-'))
|
||||
const llmServer = await mockCompletionServer()
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url)
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
|
||||
const init = await server.handleRequest('initialize', {
|
||||
cwd: storageDir,
|
||||
provider: 'deepseek-official',
|
||||
model: 'dsagent-model',
|
||||
maxTokens: 321,
|
||||
}) as { serverInfo: { name: string } }
|
||||
expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime')
|
||||
|
||||
const receipt = await server.handleRequest('session/prompt', {
|
||||
sessionId: 'main',
|
||||
contentBlocks: [{ type: 'text', text: 'fix it' }],
|
||||
})
|
||||
expect((receipt as { messageId?: unknown }).messageId).toBeTypeOf('string')
|
||||
|
||||
await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) })
|
||||
const body = llmServer.requests[0] as { model: string; messages: { role: string }[]; max_tokens?: number }
|
||||
expect(body.model).toBe('dsagent-model')
|
||||
expect(body.max_tokens).toBe(321)
|
||||
expect(body.messages[0]?.role).toBe('system')
|
||||
expect(body.messages.at(-1)?.role).toBe('user')
|
||||
expect(llmServer.headers[0]?.authorization).toBe('Bearer test-key')
|
||||
expect(transport.notifications.some(n => n.method === 'session.event')).toBe(true)
|
||||
await vi.waitFor(() => {
|
||||
expect(transport.notifications.findLast(n => n.method === 'session.status')).toEqual({
|
||||
method: 'session.status',
|
||||
params: { sessionId: 'main', status: 'idle' },
|
||||
})
|
||||
})
|
||||
|
||||
await server.handleRequest('session/prompt', {
|
||||
sessionId: 'main',
|
||||
contentBlocks: [{ type: 'text', text: 'again' }],
|
||||
})
|
||||
await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(2) })
|
||||
|
||||
const orphanHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('orphan-session'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'dsagent-model' },
|
||||
})
|
||||
orphanHandle.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'outside the sdk session map' }], source: { kind: 'user' } }))
|
||||
await orphanHandle.agent.whenIdle()
|
||||
await orphanHandle.dispose()
|
||||
expect(llmServer.requests).toHaveLength(3)
|
||||
|
||||
await server.handleRequest('shutdown', undefined)
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('queues overlapping prompts for one session without blocking other sessions', async () => {
|
||||
const mainFollowup = vi.fn<Agent['followup']>()
|
||||
const mainAgent = ({
|
||||
id: SessionId('main'),
|
||||
followup: mainFollowup,
|
||||
} satisfies Pick<Agent, 'id' | 'followup'>) as unknown as Agent
|
||||
const otherFollowup = vi.fn<Agent['followup']>()
|
||||
const otherAgent = ({
|
||||
id: SessionId('other'),
|
||||
followup: otherFollowup,
|
||||
} satisfies Pick<Agent, 'id' | 'followup'>) as unknown as Agent
|
||||
const mainHandle = { agent: mainAgent, dispose: vi.fn(() => Promise.resolve()) }
|
||||
const otherHandle = { agent: otherAgent, dispose: vi.fn(() => Promise.resolve()) }
|
||||
const create = vi.fn(async (options: { sessionId: SessionId }) =>
|
||||
String(options.sessionId) === 'main' ? mainHandle : otherHandle)
|
||||
const liveAgents = new Map<string, Agent>([['main', mainAgent], ['other', otherAgent]])
|
||||
const ctx = {
|
||||
on: vi.fn(() => () => undefined),
|
||||
agents: { create, get: (id: SessionId) => liveAgents.get(String(id)) },
|
||||
get: () => undefined,
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
const prompt = (sessionId: string, text: string) => server.prompt({
|
||||
sessionId,
|
||||
contentBlocks: [{ type: 'text', text }],
|
||||
})
|
||||
|
||||
expect((await prompt('main', 'first')).messageId).toBeTypeOf('string')
|
||||
expect((await prompt('main', 'overlap')).messageId).toBeTypeOf('string')
|
||||
expect((await prompt('other', 'independent')).messageId).toBeTypeOf('string')
|
||||
|
||||
expect(mainFollowup).toHaveBeenCalledTimes(2)
|
||||
expect(otherFollowup).toHaveBeenCalledOnce()
|
||||
await server.shutdown()
|
||||
expect(mainHandle.dispose).toHaveBeenCalledOnce()
|
||||
expect(otherHandle.dispose).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('rejects a prompt for a session whose agent was disposed outside the server', async () => {
|
||||
const followup = vi.fn<Agent['followup']>()
|
||||
const agent = ({
|
||||
id: SessionId('zombie'),
|
||||
followup,
|
||||
whenIdle: vi.fn(() => Promise.resolve()),
|
||||
} satisfies Pick<Agent, 'id' | 'followup' | 'whenIdle'>) as unknown as Agent
|
||||
const handle = { agent, dispose: vi.fn(() => Promise.resolve()) }
|
||||
// The registry drops the agent after creation, modelling an agent-loop-only
|
||||
// reload that leaves the server's SessionRecord pointing at a detached agent.
|
||||
let live = true
|
||||
const ctx = {
|
||||
on: vi.fn(() => () => undefined),
|
||||
agents: {
|
||||
create: vi.fn(async () => handle),
|
||||
get: (id: SessionId) => (live && String(id) === 'zombie' ? agent : undefined),
|
||||
},
|
||||
get: () => undefined,
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
const prompt = (text: string) => server.prompt({
|
||||
sessionId: 'zombie',
|
||||
contentBlocks: [{ type: 'text', text }],
|
||||
})
|
||||
|
||||
expect((await prompt('while live')).messageId).toBeTypeOf('string')
|
||||
live = false
|
||||
await expect(prompt('after detach')).rejects.toThrow('session agent was disposed outside the server: zombie')
|
||||
// The detached agent was never driven by the rejected prompt.
|
||||
expect(followup).toHaveBeenCalledOnce()
|
||||
await server.shutdown()
|
||||
})
|
||||
|
||||
it('forwards whole-agent status without attributing a turn outcome', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const session = ctx.sessions.create(SessionId('message-outcome'))
|
||||
const agent = ({
|
||||
id: SessionId('message-outcome'),
|
||||
session,
|
||||
} satisfies Pick<Agent, 'id' | 'session'>) as Agent
|
||||
|
||||
ctx.emit('agent/status', { agent, status: 'running' })
|
||||
ctx.emit('agent/status', { agent, status: 'idle' })
|
||||
|
||||
expect(transport.notifications.filter(notification => notification.method === 'session.status'))
|
||||
.toEqual([
|
||||
{ method: 'session.status', params: { sessionId: 'message-outcome', status: 'running' } },
|
||||
{ method: 'session.status', params: { sessionId: 'message-outcome', status: 'idle' } },
|
||||
])
|
||||
await server.shutdown()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('notifies the host when a child session is created with parent lineage', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
|
||||
ctx.sessions.create(SessionId('root-session'), {
|
||||
meta: { cwd: storageDir },
|
||||
})
|
||||
ctx.sessions.create(SessionId('child-session'), {
|
||||
meta: { cwd: storageDir, parentSession: SessionId('main') },
|
||||
})
|
||||
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.started',
|
||||
params: {
|
||||
parentSessionId: 'main',
|
||||
childSessionId: 'child-session',
|
||||
},
|
||||
})
|
||||
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('creates an SDK session without an optional system prompt', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-no-system-'))
|
||||
const llmServer = await mockCompletionServer()
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url)
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
|
||||
await server.initialize({ cwd: storageDir, provider: 'deepseek-official', model: 'plain-model' })
|
||||
await server.prompt({
|
||||
sessionId: 'plain',
|
||||
contentBlocks: [{ type: 'text', text: 'hello' }],
|
||||
})
|
||||
|
||||
await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) })
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('notifies the host when a subagent run settles', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-end-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
|
||||
const parentHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('main'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'deepseek-official' },
|
||||
})
|
||||
// A custom in-process provider may own its child at the provider/root
|
||||
// scope while preserving durable parent lineage.
|
||||
const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('child-session'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('main') },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'deepseek-official' },
|
||||
})
|
||||
expect(ctx.agents.roots()).toContain(handle.agent)
|
||||
const parentlessHandle = await parentHandle.agent.ctx.agents.create({
|
||||
sessionId: SessionId('parentless-child-session'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'spawn',
|
||||
id: SessionId('child-session'),
|
||||
localAgent: handle.agent,
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'child done' }],
|
||||
}, () => handle.dispose())
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'spawn',
|
||||
id: SessionId('parentless-child-session'),
|
||||
localAgent: parentlessHandle.agent,
|
||||
stopReason: 'error',
|
||||
}, () => parentlessHandle.dispose())
|
||||
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'spawn',
|
||||
agentId: 'child-session',
|
||||
parentSessionId: 'main',
|
||||
childSessionId: 'child-session',
|
||||
status: 'ok',
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'child done' }],
|
||||
},
|
||||
})
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'spawn',
|
||||
agentId: 'parentless-child-session',
|
||||
parentSessionId: 'main',
|
||||
childSessionId: 'parentless-child-session',
|
||||
status: 'error',
|
||||
stopReason: 'error',
|
||||
},
|
||||
})
|
||||
|
||||
await parentHandle.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('ignores a remote run id that collides with a local child of the same parent', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-remote-collision-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const parentHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('collision-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const collidingChild = await parentHandle.agent.ctx.agents.create({
|
||||
sessionId: SessionId('remote-run-id'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('collision-parent') },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'remote',
|
||||
id: SessionId('remote-run-id'),
|
||||
localAgent: undefined,
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [],
|
||||
})
|
||||
|
||||
expect(transport.notifications.some(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.agentId === 'remote-run-id',
|
||||
)).toBe(false)
|
||||
|
||||
await collidingChild.dispose()
|
||||
await parentHandle.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('retains locality across continuation runs on one live child', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-continuation-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const parentHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('continuation-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const childHandle = await parentHandle.agent.ctx.agents.create({
|
||||
sessionId: SessionId('continuation-child'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('continuation-parent') },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'continuation',
|
||||
id: SessionId('continuation-child'),
|
||||
localAgent: childHandle.agent,
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'first' }],
|
||||
})
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'continuation',
|
||||
id: SessionId('continuation-child'),
|
||||
localAgent: childHandle.agent,
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'second' }],
|
||||
}, () => childHandle.dispose())
|
||||
|
||||
expect(transport.notifications.filter(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.childSessionId === 'continuation-child',
|
||||
)).toHaveLength(2)
|
||||
|
||||
await parentHandle.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('correlates reused local ids by parent scope when runs settle out of order', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-reuse-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const oldParent = await ctx.agents.create({
|
||||
sessionId: SessionId('old-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const oldChild = await oldParent.agent.ctx.agents.create({
|
||||
sessionId: SessionId('reused-child'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('old-parent') },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const first = Promise.withResolvers<SubagentResult>()
|
||||
const sameLifetime = Promise.withResolvers<SubagentResult>()
|
||||
const replacement = Promise.withResolvers<SubagentResult>()
|
||||
const results = [first.promise, sameLifetime.promise, replacement.promise]
|
||||
let starts = 0
|
||||
let currentLocalAgent = oldChild.agent
|
||||
const disposeProvider = ctx.subagents.registerProvider({
|
||||
name: 'reused',
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start() {
|
||||
const result = results[starts]
|
||||
starts += 1
|
||||
if (result === undefined) throw new Error('unexpected fourth reused-id run')
|
||||
return Promise.resolve({ id: SessionId('reused-child'), localAgent: currentLocalAgent, result, dispose: () => Promise.resolve() })
|
||||
},
|
||||
})
|
||||
|
||||
const firstRun = await ctx.subagents.start('reused', {
|
||||
parent: oldParent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
const sameLifetimeRun = await ctx.subagents.start('reused', {
|
||||
parent: oldParent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
sameLifetime.resolve({ output: [{ type: 'text', text: 'same lifetime' }], stopReason: 'completed' })
|
||||
await sameLifetimeRun.result
|
||||
await oldChild.dispose()
|
||||
const newParent = await ctx.agents.create({
|
||||
sessionId: SessionId('new-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const newChild = await newParent.agent.ctx.agents.create({
|
||||
sessionId: SessionId('reused-child'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('new-parent') },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
currentLocalAgent = newChild.agent
|
||||
const secondRun = await ctx.subagents.start('reused', {
|
||||
parent: newParent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
|
||||
replacement.resolve({ output: [{ type: 'text', text: 'new lifetime' }], stopReason: 'completed' })
|
||||
await secondRun.result
|
||||
first.resolve({ output: [{ type: 'text', text: 'old lifetime' }], stopReason: 'completed' })
|
||||
await firstRun.result
|
||||
await Promise.resolve()
|
||||
|
||||
const finished = transport.notifications.filter(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.childSessionId === 'reused-child',
|
||||
)
|
||||
expect(finished.map(notification => notification.params?.lastAssistantMessage)).toEqual([
|
||||
[{ type: 'text', text: 'same lifetime' }],
|
||||
[{ type: 'text', text: 'new lifetime' }],
|
||||
[{ type: 'text', text: 'old lifetime' }],
|
||||
])
|
||||
expect(finished.map(notification => notification.params?.parentSessionId)).toEqual([
|
||||
'old-parent',
|
||||
'new-parent',
|
||||
'old-parent',
|
||||
])
|
||||
|
||||
await firstRun.dispose()
|
||||
await sameLifetimeRun.dispose()
|
||||
await secondRun.dispose()
|
||||
disposeProvider()
|
||||
await newChild.dispose()
|
||||
await oldParent.dispose()
|
||||
await newParent.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps locality bound to the accepted run across provider re-registration', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-provider-reuse-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const parent = await ctx.agents.create({
|
||||
sessionId: SessionId('provider-reuse-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const child = await parent.agent.ctx.agents.create({
|
||||
sessionId: SessionId('provider-reuse-child'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('provider-reuse-parent') },
|
||||
agentOptions: { model: 'deepseek-official' },
|
||||
})
|
||||
const localResult = Promise.withResolvers<SubagentResult>()
|
||||
const remoteResult = Promise.withResolvers<SubagentResult>()
|
||||
const unregisterLocal = ctx.subagents.registerProvider({
|
||||
name: 'reused-provider',
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => Promise.resolve({
|
||||
id: SessionId('provider-reuse-child'),
|
||||
localAgent: child.agent,
|
||||
result: localResult.promise,
|
||||
dispose: () => Promise.resolve(),
|
||||
}),
|
||||
})
|
||||
const localRun = await ctx.subagents.start('reused-provider', {
|
||||
parent: parent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
unregisterLocal()
|
||||
|
||||
const unregisterRemote = ctx.subagents.registerProvider({
|
||||
name: 'reused-provider',
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: false,
|
||||
start: () => Promise.resolve({
|
||||
id: SessionId('provider-reuse-child'),
|
||||
localAgent: undefined,
|
||||
result: remoteResult.promise,
|
||||
dispose: () => Promise.resolve(),
|
||||
}),
|
||||
})
|
||||
const remoteRun = await ctx.subagents.start('reused-provider', {
|
||||
parent: parent.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
|
||||
remoteResult.resolve({ output: [{ type: 'text', text: 'remote' }], stopReason: 'completed' })
|
||||
await remoteRun.result
|
||||
await Promise.resolve()
|
||||
expect(transport.notifications.some(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.lastAssistantMessage !== undefined,
|
||||
)).toBe(false)
|
||||
|
||||
await child.dispose()
|
||||
localResult.resolve({ output: [{ type: 'text', text: 'local' }], stopReason: 'completed' })
|
||||
await localRun.result
|
||||
await Promise.resolve()
|
||||
expect(transport.notifications.filter(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.childSessionId === 'provider-reuse-child',
|
||||
)).toEqual([{
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'reused-provider',
|
||||
agentId: 'provider-reuse-child',
|
||||
parentSessionId: 'provider-reuse-parent',
|
||||
childSessionId: 'provider-reuse-child',
|
||||
status: 'ok',
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'local' }],
|
||||
},
|
||||
}])
|
||||
|
||||
await localRun.dispose()
|
||||
await remoteRun.dispose()
|
||||
unregisterRemote()
|
||||
await parent.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('uses explicit local provenance when start was missed and ignores remote runs', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-fallback-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
let parentHandle: AgentHandle | undefined
|
||||
let handle: AgentHandle | undefined
|
||||
let failedHandle: AgentHandle | undefined
|
||||
try {
|
||||
parentHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('fallback-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'deepseek-official' },
|
||||
})
|
||||
handle = await parentHandle.agent.ctx.agents.create({
|
||||
sessionId: SessionId('fallback-child-session'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'deepseek-official' },
|
||||
})
|
||||
const fallbackChild = handle.agent
|
||||
failedHandle = await parentHandle.agent.ctx.agents.create({
|
||||
sessionId: SessionId('failed-child-session'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { provider: 'deepseek-official', model: 'deepseek-official' },
|
||||
})
|
||||
const missedStartResult = Promise.withResolvers<SubagentResult>()
|
||||
const disposeMissedStartProvider = ctx.subagents.registerProvider({
|
||||
name: 'fork',
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
inheritsParentContext: true,
|
||||
start: () => Promise.resolve({
|
||||
id: SessionId('fallback-child-session'),
|
||||
localAgent: fallbackChild,
|
||||
result: missedStartResult.promise,
|
||||
dispose: () => Promise.resolve(),
|
||||
}),
|
||||
})
|
||||
// Start before the server subscribes. The terminal payload still carries
|
||||
// this run's exact local child without reconstructing it from ids.
|
||||
const missedStartRun = await ctx.subagents.start('fork', {
|
||||
parent: parentHandle.agent,
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport, { maxTokensAsSuccess: true })
|
||||
|
||||
missedStartResult.resolve({ output: [], stopReason: 'max-tokens' })
|
||||
await missedStartRun.result
|
||||
await Promise.resolve()
|
||||
await missedStartRun.dispose()
|
||||
disposeMissedStartProvider()
|
||||
// The server also missed this agent's creation but sees the exact child
|
||||
// on the run lifecycle payload.
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'fork-live-fallback',
|
||||
id: SessionId('fallback-child-session'),
|
||||
localAgent: fallbackChild,
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [],
|
||||
})
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'fork',
|
||||
id: SessionId('failed-child-session'),
|
||||
localAgent: failedHandle.agent,
|
||||
stopReason: 'error',
|
||||
})
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'fork',
|
||||
id: SessionId('missing-child-agent'),
|
||||
localAgent: undefined,
|
||||
stopReason: 'error',
|
||||
})
|
||||
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'fork',
|
||||
agentId: 'fallback-child-session',
|
||||
parentSessionId: 'fallback-parent',
|
||||
childSessionId: 'fallback-child-session',
|
||||
status: 'ok',
|
||||
stopReason: 'max-tokens',
|
||||
lastAssistantMessage: [],
|
||||
},
|
||||
})
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.finished',
|
||||
params: {
|
||||
provider: 'fork',
|
||||
agentId: 'failed-child-session',
|
||||
parentSessionId: 'fallback-parent',
|
||||
childSessionId: 'failed-child-session',
|
||||
status: 'error',
|
||||
stopReason: 'error',
|
||||
},
|
||||
})
|
||||
expect(transport.notifications.some(n =>
|
||||
n.method === 'subagent.finished'
|
||||
&& n.params?.agentId === 'missing-child-agent',
|
||||
)).toBe(false)
|
||||
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await handle?.dispose()
|
||||
await failedHandle?.dispose()
|
||||
await parentHandle?.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('does not re-register an LLM adapter whose provider already has an owner', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-existing-llm-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
await ctx.plugin(LlmDeepSeek)
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
const inspect = server as unknown as { hasAdapterFor(provider: string): boolean }
|
||||
|
||||
expect(inspect.hasAdapterFor('deepseek-official')).toBe(true)
|
||||
expect(inspect.hasAdapterFor('missing-provider')).toBe(false)
|
||||
await server.initialize({ cwd: storageDir, provider: 'deepseek-official', model: 'preinstalled-model' })
|
||||
|
||||
expect(ctx.get('llm')?.listProviders().filter(provider => provider.id === 'deepseek-official')).toEqual([{ id: 'deepseek-official', name: 'DeepSeek' }])
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects a missing non-DeepSeek provider when an LLM service already exists', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-new-llm-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
|
||||
await ctx.plugin(LlmDeepSeek)
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
|
||||
await expect(server.initialize({ cwd: storageDir, provider: 'private', model: 'new-model' }))
|
||||
.rejects.toThrow('no adapter registered for provider "private"')
|
||||
|
||||
expect(ctx.get('llm')?.listProviders()).toEqual([{ id: 'deepseek-official', name: 'DeepSeek' }])
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it.each([0, -1, 1.5, Number.NaN, Number.MAX_SAFE_INTEGER + 1])(
|
||||
'rejects invalid initialize maxTokens %s at the wire boundary',
|
||||
async (maxTokens) => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-invalid-max-tokens-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
await expect(server.initialize({
|
||||
cwd: storageDir,
|
||||
provider: 'deepseek-official',
|
||||
model: 'model',
|
||||
maxTokens,
|
||||
})).rejects.toThrow('initialize maxTokens must be a positive safe integer')
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
it('reports no adapter when the LLM service is absent', async () => {
|
||||
const ctx = new Context()
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
|
||||
hasAdapterFor(model: string): boolean
|
||||
shutdown(): Promise<Record<string, never>>
|
||||
}
|
||||
|
||||
expect(server.hasAdapterFor('missing-model')).toBe(false)
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects unknown JSON-RPC runtime methods', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-unknown-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
try {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
|
||||
await expect(server.handleRequest('does/not/exist', {}))
|
||||
.rejects
|
||||
.toThrow('unknown DeepSeek Harness SDK runtime method: does/not/exist')
|
||||
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('coalesces concurrent session creation and retries a failed creation', async () => {
|
||||
let resolveShared: ((handle: AgentHandle) => void) | undefined
|
||||
const sharedCreation = new Promise<AgentHandle>((resolve) => { resolveShared = resolve })
|
||||
const sharedHandle = { agent: {} as Agent, dispose: vi.fn(() => Promise.resolve()) }
|
||||
const retryHandle = { agent: {} as Agent, dispose: vi.fn(() => Promise.resolve()) }
|
||||
const create = vi.fn<(options: unknown) => Promise<AgentHandle>>()
|
||||
.mockReturnValueOnce(sharedCreation)
|
||||
.mockRejectedValueOnce(new Error('creation failed'))
|
||||
.mockResolvedValueOnce(retryHandle)
|
||||
const ctx = {
|
||||
on: vi.fn(() => () => undefined),
|
||||
agents: { create, get: () => undefined },
|
||||
get: () => undefined,
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
|
||||
getOrCreateSession(sessionId: string): Promise<{ handle: AgentHandle }>
|
||||
shutdown(): Promise<Record<string, never>>
|
||||
}
|
||||
|
||||
const first = server.getOrCreateSession('shared')
|
||||
const second = server.getOrCreateSession('shared')
|
||||
expect(create).toHaveBeenCalledTimes(1)
|
||||
resolveShared?.(sharedHandle)
|
||||
const [firstRecord, secondRecord] = await Promise.all([first, second])
|
||||
expect(firstRecord).toBe(secondRecord)
|
||||
|
||||
await expect(server.getOrCreateSession('retry')).rejects.toThrow('creation failed')
|
||||
await expect(server.getOrCreateSession('retry')).resolves.toMatchObject({ handle: retryHandle })
|
||||
expect(create).toHaveBeenCalledTimes(3)
|
||||
|
||||
await server.shutdown()
|
||||
expect(sharedHandle.dispose).toHaveBeenCalledOnce()
|
||||
expect(retryHandle.dispose).toHaveBeenCalledOnce()
|
||||
await expect(server.getOrCreateSession('after-shutdown')).rejects.toThrow('SDK server is shutting down')
|
||||
})
|
||||
|
||||
it('resolves a relative cwd before creating the session', async () => {
|
||||
const create = vi.fn<(options: unknown) => Promise<AgentHandle>>()
|
||||
.mockResolvedValue({ agent: {} as Agent, dispose: () => Promise.resolve() })
|
||||
const ctx = {
|
||||
on: vi.fn(() => () => undefined),
|
||||
agents: { create, get: () => undefined },
|
||||
get: () => ({ listProviders: () => [{ id: 'mock', name: 'Mock' }] }),
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
|
||||
initialize(params: { cwd: string; provider: string; model: string; maxTokens?: number }): Promise<unknown>
|
||||
getOrCreateSession(sessionId: string): Promise<unknown>
|
||||
shutdown(): Promise<Record<string, never>>
|
||||
}
|
||||
|
||||
await server.initialize({ cwd: '.', provider: 'mock', model: 'model', maxTokens: 123 })
|
||||
await server.getOrCreateSession('relative')
|
||||
|
||||
expect(create).toHaveBeenCalledWith(expect.objectContaining({
|
||||
meta: { cwd: process.cwd() },
|
||||
agentOptions: { provider: 'mock', model: 'model', maxTokens: 123 },
|
||||
}))
|
||||
await server.shutdown()
|
||||
})
|
||||
|
||||
it('settles every teardown and aggregates multiple failures', async () => {
|
||||
const firstDispose = vi.fn(() => { throw new Error('first teardown failed') })
|
||||
const secondDispose = vi.fn(() => Promise.reject(new Error('second teardown failed')))
|
||||
const ctx = {
|
||||
on: vi.fn(() => () => undefined),
|
||||
agents: { create: vi.fn(), get: () => undefined },
|
||||
get: () => undefined,
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
|
||||
sessions: Map<string, { handle: AgentHandle; lastTurnEnd: undefined; activePrompt: boolean }>
|
||||
shutdown(): Promise<Record<string, never>>
|
||||
}
|
||||
server.sessions.set('first', { handle: { agent: {} as Agent, dispose: firstDispose }, lastTurnEnd: undefined, activePrompt: false })
|
||||
server.sessions.set('second', { handle: { agent: {} as Agent, dispose: secondDispose }, lastTurnEnd: undefined, activePrompt: false })
|
||||
|
||||
await expect(server.shutdown()).rejects.toThrow('SDK server teardown failed')
|
||||
expect(firstDispose).toHaveBeenCalledOnce()
|
||||
expect(secondDispose).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('continues teardown after a subscription disposer fails', async () => {
|
||||
let subscription = 0
|
||||
const listenerFailure = new Error('listener teardown failed')
|
||||
const on = vi.fn(() => {
|
||||
subscription += 1
|
||||
return subscription === 1 ? () => { throw listenerFailure } : () => undefined
|
||||
})
|
||||
const ctx = {
|
||||
on,
|
||||
agents: { create: vi.fn(), get: () => undefined },
|
||||
get: () => undefined,
|
||||
} as unknown as Context
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
|
||||
await expect(server.shutdown()).rejects.toBe(listenerFailure)
|
||||
expect(on).toHaveBeenCalledTimes(4)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user