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

Merge remote-tracking branch 'origin/master' into worktree/dsbench-patch

Browse Source 
# Conflicts:
#	packages/bash/bash-local/README.md
#	packages/bash/bash-local/src/index.ts
#	packages/bash/bash-local/src/run.ts
#	packages/bash/bash-local/tests/run.spec.ts
#	packages/examples/agent-spine-demo/README.md
#	packages/examples/agent-spine-demo/src/index.ts
#	packages/ui/jsonrpc/README.md
#	packages/ui/jsonrpc/src/server.ts
This commit is contained in:
Yichen Jiang
2026-07-19 14:53:03 +08:00
parent a38ff125a7 7747aa450c
commit 3df3c6bf37
876 changed files with 48016 additions and 13119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
packages/ui/jsonrpc/README.md
View File

@@ -1,26 +1,26 @@
# @deepseek-ai/dsh-jsonrpc
Stdio JSON-RPC plugin for out-of-process SDK clients such as Python `deepseek_harness`. [`HarnessSdkServer`](src/server.ts) handles `initialize` → `session/prompt` → `shutdown` plus session and subagent notifications over [`JsonRpcLineTransport`](src/transport.ts). This package owns the protocol; [`jsonrpc-agent`](../../examples/jsonrpc-demo/README.md) boots the external `cordis.yml` that chooses the surrounding runtime. See the [single-executable RFC](../../../docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md) for the distribution design.
The `jsonrpc` plugin serves newline-delimited JSON-RPC over stdio so out-of-process SDK clients can drive harness agents. [`HarnessSdkServer`](src/server.ts) owns the protocol methods and notifications; [`jsonrpc-demo`](../../examples/jsonrpc-demo/README.md) supplies the surrounding `cordis.yml` application.
## Wiring
`inject: ['agents']`. The server gets or creates one agent per `sessionId` on `session/prompt` and demuxes `subagent/end` through the registry. If `initialize.model` lacks a registered adapter, it mounts `dsh-llm-deepseek` using `$DEEPSEEK_API_KEY` and `$DEEPSEEK_BASE_URL`; a config-registered adapter wins. Persistence, tools, and other adapters come from the surrounding `cordis.yml`.
`inject: ['agents']`. The server gets or creates one agent per `sessionId`. It forwards subagent completions only when the service-snapshotted lifecycle `local` flag is true; provider names, child ids, and durable lineage never establish locality. A registered adapter wins, an unowned `deepseek` route mounts `dsh-llm-deepseek`, and any other unowned provider fails initialization. Other capabilities come from the surrounding `cordis.yml`.
## Config
`maxTokensAsSuccess` defaults to `false`. Set it to `true` for evaluation hosts that distinguish an accepted, token-limited agent result from an infrastructure failure. `JsonRpcConfig.input`, `output`, and `exit` are test-only runtime seams; production uses process stdio and `process.exit`.
`maxTokensAsSuccess` defaults to `false`. Set it to `true` for evaluation hosts that distinguish an accepted, token-limited agent result from an infrastructure failure. `JsonRpcConfig.input`, `output`, and `exit` are runtime-only transport seams; production uses process stdio and `process.exit`.
## stdout is the protocol
stdout carries only JSON-RPC frames. The loading config must omit stdout loggers; diagnostics go to stderr.
Stdout carries only JSON-RPC frames. The deployment must not compose a stdout logger; diagnostics belong on stderr.
## Shutdown and exit semantics
A `shutdown` request flushes its response, disposes the plugin fiber, then exits 0. Disposal idempotently shuts down every SDK-created agent to quiescence, detaches subscriptions, and closes the transport. Bare fiber disposal only stops serving; it does not exit. The app bin owns root disposal for stdin EOF (0), SIGTERM (0), and SIGINT (130).
The plugin answers `shutdown`, disposes SDK-owned agents and subscriptions to quiescence, closes the transport, then exits with code 0. EOF and signal exits belong to the app bin, which disposes the root context. Unloading only this plugin stops serving without exiting the process.
## Wire notes
`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime`. Each session permits one in-flight prompt; overlap fails immediately, other sessions remain independent, and the session is reusable after settlement. Persistence roots and deployment persona remain in `cordis.yml`.
`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime`. A session accepts one in-flight prompt; overlap fails immediately, other sessions remain independent, and the session is reusable after settlement. Persistence roots and persona come from `cordis.yml`.
## Model Experience

2
packages/ui/jsonrpc/package.json
View File

@@ -28,6 +28,7 @@
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-llm-deepseek": "^0.0.1",
"@deepseek-ai/dsh-scope": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-subagent": "^0.0.1",
"cordis": "^4.0.0-rc.7"
@@ -38,6 +39,7 @@
"@deepseek-ai/dsh-agent-spine-demo": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-llm-deepseek": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-subagent": "workspace:^",

115
packages/ui/jsonrpc/src/server.ts
View File

@@ -1,8 +1,6 @@
/**
* JSON-RPC methods and notifications for SDK clients. Requests are
* `initialize`, repeated `session/prompt`, then `shutdown`; notifications carry
* durable session events, settled turns, and subagent lineage/outcomes. The
* external `cordis.yml` owns plugins, persistence, and the adapter set.
* JSON-RPC method and notification surface for out-of-process harness SDKs.
* The surrounding context owns plugins, persistence, and configured adapters.
*
* @module @deepseek-ai/dsh-jsonrpc/server
*/
@@ -10,31 +8,31 @@
import type { Context } from 'cordis'
import { resolve } from 'node:path'
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { AgentHandle } from '@deepseek-ai/dsh-agent'
import { AgentId } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentHandle } from '@deepseek-ai/dsh-agent'
import { carrierKeyOf, type Scoped } from '@deepseek-ai/dsh-scope'
import { SessionId, type TurnEndReason } from '@deepseek-ai/dsh-session'
import type SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import type { JsonRpcTransportPeer } from './transport.ts'
/** One-time SDK initialization parameters. */
/** Parameters for the process-wide SDK handshake. */
export interface InitializeParams {
/** Working directory recorded on every SDK-created session's header. */
cwd: string
/** Provider route every SDK-created agent runs on. */
provider: string
/** Model name every SDK-created agent runs on (see {@link HarnessSdkServer.initialize} for adapter fallback). */
model: string
}
/** SDK handshake result. */
/** Wire-stable server identity returned by initialization. */
export interface InitializeResult {
/** Wire-stable server identity (`deepseek-harness-sdk-runtime`) and version. */
serverInfo: { name: string; version: string }
}
/**
* Parameters of a `session/prompt` request: one user turn on one SDK session,
* with at most one in flight per session.
*/
/** One user turn on one SDK session. */
export interface SessionPromptParams {
/** The SDK-side session id; an unknown id lazily creates the agent+session pair. */
sessionId: string
@@ -42,7 +40,7 @@ export interface SessionPromptParams {
contentBlocks: ContentBlock[]
}
/** Accepted prompt result; the outcome is reported by `session.finished`. */
/** Prompt acceptance after turn settlement; outcome rides on `session.finished`. */
export interface SessionPromptResult {
/** Always `true`; the turn outcome is the paired `session.finished` notification. */
accepted: true
@@ -54,9 +52,9 @@ interface SessionRecord {
activePrompt: boolean
}
interface SubagentRecord {
childSessionId: string
parentSessionId: string | undefined
/** Recover the delegating parent from the service-owned scoped carrier. */
function subagentParentOf(carrier: Scoped<SubagentService>): Agent {
return carrierKeyOf(carrier) as Agent
}
/** Deployment-specific status mapping for SDK turn and subagent outcomes. */
@@ -65,6 +63,11 @@ export interface HarnessSdkServerOptions {
maxTokensAsSuccess?: boolean
}
function successStatus(reason: string, options: HarnessSdkServerOptions): 'ok' | 'error' {
if (reason === 'completed') return 'ok'
return reason === 'max-tokens' && options.maxTokensAsSuccess === true ? 'ok' : 'error'
}
/**
* SDK server over one booted harness context and transport peer. Construction
* subscribes to session, agent, and subagent lifecycle events until shutdown;
@@ -72,11 +75,11 @@ export interface HarnessSdkServerOptions {
*/
export class HarnessSdkServer {
private cwd = process.cwd()
private provider = 'deepseek'
private model = 'deepseek'
private llmFiber: { dispose(): Promise<void> } | undefined
private readonly sessions = new Map<string, SessionRecord>()
private readonly sessionCreations = new Map<string, Promise<SessionRecord>>()
private readonly subagentSessions = new Map<string, SubagentRecord>()
private readonly disposers: (() => void)[] = []
private shutdownTask: Promise<Record<string, never>> | undefined
private shuttingDown = false
@@ -86,6 +89,7 @@ export class HarnessSdkServer {
private readonly transport: JsonRpcTransportPeer,
private readonly options: HarnessSdkServerOptions = {},
) {
const serverOptions = this.options
this.disposers.push(ctx.on('session/event', (session, event) => {
if (event.type === 'turn/end') {
const rec = this.sessions.get(String(session.id))
@@ -101,29 +105,18 @@ export class HarnessSdkServer {
childSessionId: String(session.id),
})
}))
// Cache lineage before child disposal removes the agent from the registry.
this.disposers.push(ctx.on('agent/created', (agent) => {
this.subagentSessions.set(String(agent.id), {
childSessionId: String(agent.session.id),
parentSessionId: agent.session.header.parentSession === undefined
? undefined
: String(agent.session.header.parentSession),
})
}))
this.disposers.push(ctx.on('subagent/end', (info: SubagentRunEndInfo) => {
const rec = this.subagentSessions.get(String(info.id))
const agent = this.ctx.agents.get(info.id)
const childSessionId = rec?.childSessionId ?? (agent === undefined ? undefined : String(agent.session.id))
const parentSessionId = rec?.parentSessionId ?? (
agent?.session.header.parentSession === undefined ? undefined : String(agent.session.header.parentSession)
)
if (childSessionId === undefined) return
this.transport.notify('subagent.finished', {
this.disposers.push(ctx.on('subagent/end', function (this: Scoped<SubagentService>, info: SubagentRunEndInfo) {
const parent = subagentParentOf(this)
// This protocol reports only in-process child sessions. The service
// snapshots the provider's exact run provenance through child disposal;
// matching ids or parent lineage alone never establishes locality.
if (!info.local) return
transport.notify('subagent.finished', {
provider: info.provider,
agentId: String(info.id),
...(parentSessionId === undefined ? {} : { parentSessionId }),
childSessionId,
status: this.successStatus(info.stopReason),
parentSessionId: String(parent.session.id),
childSessionId: String(info.id),
status: successStatus(info.stopReason, serverOptions),
stopReason: info.stopReason,
...(info.lastAssistantMessage === undefined ? {} : { lastAssistantMessage: info.lastAssistantMessage }),
})
@@ -131,26 +124,25 @@ export class HarnessSdkServer {
}
/**
* Record cwd and model, mounting the DeepSeek adapter only when the config
* registered no adapter for that model.
* @param params - the SDK handshake parameters.
* @returns the server identity for the handshake.
* Configure the SDK route, mounting the DeepSeek fallback only when unowned.
* @param params - SDK handshake parameters.
* @returns server identity for the handshake.
*/
async initialize(params: InitializeParams): Promise<InitializeResult> {
this.cwd = resolve(params.cwd)
this.provider = params.provider
this.model = params.model
if (!this.llmFiber && !this.hasAdapterFor(this.model)) {
this.llmFiber = await this.ctx.plugin(LlmDeepSeek, { models: [this.model] })
if (!this.hasAdapterFor(this.provider)) {
if (this.provider !== 'deepseek') throw new Error(`no adapter registered for provider "${this.provider}"`)
this.llmFiber = await this.ctx.plugin(LlmDeepSeek, {})
}
return { serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } }
}
/**
* Get or create the session agent, send the prompt, await quiescence, then
* notify `session.finished`. A session accepts one prompt at a time; other
* sessions remain independent.
* @param params - the target session id and prompt content.
* @returns `{ accepted: true }` after the turn settled.
* Run one prompt to settlement; overlap on the same session fails.
* @param params - target session and user content.
* @returns acceptance after the turn settled.
*/
async prompt(params: SessionPromptParams): Promise<SessionPromptResult> {
const rec = await this.getOrCreateSession(params.sessionId)
@@ -173,9 +165,9 @@ export class HarnessSdkServer {
}
/**
* Dispose SDK-created agents to quiescence, unmount the server-mounted adapter,
* and detach subscriptions. The surrounding context remains running.
* @returns an empty object (the JSON-RPC result).
* Dispose server-owned agents, adapter, and subscriptions to quiescence.
* The surrounding context remains running.
* @returns empty JSON-RPC result.
*/
shutdown(): Promise<Record<string, never>> {
this.shutdownTask ??= this.performShutdown()
@@ -189,7 +181,6 @@ export class HarnessSdkServer {
this.sessionCreations.clear()
const records = [...this.sessions.values()]
this.sessions.clear()
this.subagentSessions.clear()
const failures: unknown[] = []
while (this.disposers.length > 0) {
try {
@@ -212,8 +203,8 @@ export class HarnessSdkServer {
}
/**
* Dispatch an incoming request; unknown methods throw for transport conversion
* to a JSON-RPC error response.
* Dispatch one incoming JSON-RPC request to its typed handler. Throws (→ a
* JSON-RPC error response) on an unknown method.
* @param method - the JSON-RPC method name.
* @param params - the raw params object from the wire.
* @returns the handler's result, to be serialized as the response.
@@ -248,10 +239,9 @@ export class HarnessSdkServer {
private async createSession(sessionId: string): Promise<SessionRecord> {
const handle = await this.ctx.agents.create({
agentId: AgentId(sessionId),
sessionId: SessionId(sessionId),
meta: { cwd: this.cwd },
agentOptions: { model: this.model },
agentOptions: { provider: this.provider, model: this.model },
})
const rec: SessionRecord = { handle, lastTurnEnd: undefined, activePrompt: false }
this.sessions.set(sessionId, rec)
@@ -260,15 +250,10 @@ export class HarnessSdkServer {
private finishedStatus(reason: TurnEndReason | undefined): 'ok' | 'error' {
if (!reason) return 'error'
return this.successStatus(reason.kind)
return successStatus(reason.kind, this.options)
}
private successStatus(reason: string): 'ok' | 'error' {
if (reason === 'completed') return 'ok'
return reason === 'max-tokens' && this.options.maxTokensAsSuccess === true ? 'ok' : 'error'
}
private hasAdapterFor(model: string): boolean {
return this.ctx.get('llm')?.models().includes(model) ?? false
private hasAdapterFor(provider: string): boolean {
return this.ctx.get('llm')?.listProviders().some(entry => entry.id === provider) ?? false
}
}

122
packages/ui/jsonrpc/tests/built-scope-carrier.e2e.ts Normal file
View File

@@ -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-persistence/session-persistence-jsonl/lib/index.js"),
load("packages/ui/jsonrpc/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: [],
},
}])
})
})

10
packages/ui/jsonrpc/tests/plugin-apply.spec.ts
View File

@@ -153,7 +153,7 @@ describe('dsh-jsonrpc plugin apply', () => {
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, model: 'apply-model' } })
harness.send({ jsonrpc: '2.0', id: 'init-1', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek', model: 'apply-model' } })
const response = await harness.waitForFrame(frame => frame.id === 'init-1', 'initialize response')
expect(response).toEqual({
@@ -175,7 +175,7 @@ describe('dsh-jsonrpc plugin apply', () => {
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, model: 'dsagent-model' } })
harness.send({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { cwd: storageDir, provider: 'deepseek', model: 'dsagent-model' } })
await harness.waitForFrame(frame => frame.id === 1, 'initialize response')
harness.send({
@@ -236,7 +236,7 @@ describe('dsh-jsonrpc plugin apply', () => {
expect(harness.exits()).toEqual([0])
const before = harness.frames().length
harness.send({ jsonrpc: '2.0', id: 'after-exit', method: 'initialize', params: { cwd: storageDir, model: 'x' } })
harness.send({ jsonrpc: '2.0', id: 'after-exit', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek', model: 'x' } })
await settle()
expect(harness.frames().length).toBe(before)
} finally {
@@ -257,7 +257,7 @@ describe('dsh-jsonrpc plugin apply', () => {
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, model: 'x' } })
harness.send({ jsonrpc: '2.0', id: 'after-flush-failure', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek', model: 'x' } })
await settle()
expect(harness.frames().length).toBe(before)
} finally {
@@ -281,7 +281,7 @@ describe('dsh-jsonrpc plugin apply', () => {
await harness.fiber.dispose()
const before = harness.frames().length
harness.send({ jsonrpc: '2.0', id: 'probe-2', method: 'initialize', params: { cwd: storageDir, model: 'x' } })
harness.send({ jsonrpc: '2.0', id: 'probe-2', method: 'initialize', params: { cwd: storageDir, provider: 'deepseek', model: 'x' } })
await settle()
expect(harness.frames().length).toBe(before)
expect(harness.exits()).toEqual([])

437
packages/ui/jsonrpc/tests/server.spec.ts
View File

@@ -5,12 +5,13 @@ import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { AgentId, type Agent, type AgentHandle } from '@deepseek-ai/dsh-agent'
import { type Agent, type AgentHandle } from '@deepseek-ai/dsh-agent'
import { 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 SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
import SubagentService, { type SubagentResult, type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
import { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts'
class FakeTransport implements JsonRpcTransportPeer {
@@ -66,7 +67,13 @@ async function makeHarness(storageDir: string) {
}
/** Drive the owning service so test lifecycle events carry the real parent scope. */
async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndInfo): Promise<void> {
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 },
@@ -74,9 +81,8 @@ async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndI
async start() {
return {
id: info.id,
result: info.lastAssistantMessage === undefined
? Promise.reject(new Error('synthetic infrastructure failure'))
: Promise.resolve({ output: info.lastAssistantMessage, stopReason: info.stopReason }),
localAgent: info.localAgent,
result: result.promise,
dispose: () => Promise.resolve(),
}
},
@@ -87,6 +93,12 @@ async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndI
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 {
@@ -107,6 +119,7 @@ describe('HarnessSdkServer', () => {
const init = await server.handleRequest('initialize', {
cwd: storageDir,
provider: 'deepseek',
model: 'dsagent-model',
}) as { serverInfo: { name: string } }
expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime')
@@ -135,10 +148,9 @@ describe('HarnessSdkServer', () => {
expect(llmServer.requests).toHaveLength(2)
const orphanHandle = await ctx.agents.create({
agentId: AgentId('orphan-agent'),
sessionId: SessionId('orphan-session'),
meta: { cwd: storageDir },
agentOptions: { model: 'dsagent-model' },
agentOptions: { provider: 'deepseek', model: 'dsagent-model' },
})
orphanHandle.agent.send([{ type: 'text', text: 'outside the sdk session map' }])
await orphanHandle.agent.whenIdle()
@@ -170,8 +182,8 @@ describe('HarnessSdkServer', () => {
} 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: { agentId: AgentId }) =>
String(options.agentId) === 'main' ? mainHandle : otherHandle)
const create = vi.fn(async (options: { sessionId: SessionId }) =>
String(options.sessionId) === 'main' ? mainHandle : otherHandle)
const ctx = {
on: vi.fn(() => () => undefined),
agents: { create, get: () => undefined },
@@ -241,7 +253,7 @@ describe('HarnessSdkServer', () => {
try {
const server = new HarnessSdkServer(ctx, new FakeTransport())
await server.initialize({ cwd: storageDir, model: 'plain-model' })
await server.initialize({ cwd: storageDir, provider: 'deepseek', model: 'plain-model' })
await server.prompt({
sessionId: 'plain',
contentBlocks: [{ type: 'text', text: 'hello' }],
@@ -263,29 +275,42 @@ describe('HarnessSdkServer', () => {
const server = new HarnessSdkServer(ctx, transport)
const parentHandle = await ctx.agents.create({
agentId: AgentId('parent-agent'),
sessionId: SessionId('main'),
meta: { cwd: storageDir },
agentOptions: { model: 'deepseek' },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
// 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({
agentId: AgentId('child-agent'),
sessionId: SessionId('child-session'),
meta: { cwd: storageDir, parentSession: SessionId('main') },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
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' },
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'spawn',
id: AgentId('child-agent'),
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-agent',
agentId: 'child-session',
parentSessionId: 'main',
childSessionId: 'child-session',
status: 'ok',
@@ -293,8 +318,18 @@ describe('HarnessSdkServer', () => {
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 handle.dispose()
await parentHandle.dispose()
await server.shutdown()
} finally {
@@ -303,7 +338,282 @@ describe('HarnessSdkServer', () => {
}
})
it('falls back to live agent lineage for uncached subagent end events', async () => {
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' },
})
const collidingChild = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('remote-run-id'),
meta: { cwd: storageDir, parentSession: SessionId('collision-parent') },
agentOptions: { model: 'deepseek' },
})
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' },
})
const childHandle = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('continuation-child'),
meta: { cwd: storageDir, parentSession: SessionId('continuation-parent') },
agentOptions: { model: 'deepseek' },
})
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' },
})
const oldChild = await oldParent.agent.ctx.agents.create({
sessionId: SessionId('reused-child'),
meta: { cwd: storageDir, parentSession: SessionId('old-parent') },
agentOptions: { model: 'deepseek' },
})
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' },
})
const newChild = await newParent.agent.ctx.agents.create({
sessionId: SessionId('reused-child'),
meta: { cwd: storageDir, parentSession: SessionId('new-parent') },
agentOptions: { model: 'deepseek' },
})
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' },
})
const child = await parent.agent.ctx.agents.create({
sessionId: SessionId('provider-reuse-child'),
meta: { cwd: storageDir, parentSession: SessionId('provider-reuse-parent') },
agentOptions: { model: 'deepseek' },
})
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
@@ -311,40 +621,67 @@ describe('HarnessSdkServer', () => {
let failedHandle: AgentHandle | undefined
try {
parentHandle = await ctx.agents.create({
agentId: AgentId('fallback-parent-agent'),
sessionId: SessionId('fallback-parent'),
meta: { cwd: storageDir },
agentOptions: { model: 'deepseek' },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
handle = await ctx.agents.create({
agentId: AgentId('fallback-child-agent'),
handle = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('fallback-child-session'),
meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
agentOptions: { model: 'deepseek' },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
failedHandle = await ctx.agents.create({
agentId: AgentId('failed-child-agent'),
const fallbackChild = handle.agent
failedHandle = await parentHandle.agent.ctx.agents.create({
sessionId: SessionId('failed-child-session'),
meta: { cwd: storageDir },
agentOptions: { model: 'deepseek' },
agentOptions: { provider: 'deepseek', model: 'deepseek' },
})
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',
id: AgentId('fallback-child-agent'),
stopReason: 'max-tokens',
provider: 'fork-live-fallback',
id: SessionId('fallback-child-session'),
localAgent: fallbackChild,
stopReason: 'completed',
lastAssistantMessage: [],
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'fork',
id: AgentId('failed-child-agent'),
id: SessionId('failed-child-session'),
localAgent: failedHandle.agent,
stopReason: 'error',
})
await settleSubagent(ctx, parentHandle.agent, {
provider: 'fork',
id: AgentId('missing-child-agent'),
id: SessionId('missing-child-agent'),
localAgent: undefined,
stopReason: 'error',
})
@@ -352,7 +689,7 @@ describe('HarnessSdkServer', () => {
method: 'subagent.finished',
params: {
provider: 'fork',
agentId: 'fallback-child-agent',
agentId: 'fallback-child-session',
parentSessionId: 'fallback-parent',
childSessionId: 'fallback-child-session',
status: 'ok',
@@ -364,7 +701,8 @@ describe('HarnessSdkServer', () => {
method: 'subagent.finished',
params: {
provider: 'fork',
agentId: 'failed-child-agent',
agentId: 'failed-child-session',
parentSessionId: 'fallback-parent',
childSessionId: 'failed-child-session',
status: 'error',
stopReason: 'error',
@@ -385,20 +723,20 @@ describe('HarnessSdkServer', () => {
}
})
it('does not re-register an LLM adapter that already exists', async () => {
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, { models: ['preinstalled-model'] })
await ctx.plugin(LlmDeepSeek)
try {
const server = new HarnessSdkServer(ctx, new FakeTransport())
const inspect = server as unknown as { hasAdapterFor(model: string): boolean }
const inspect = server as unknown as { hasAdapterFor(provider: string): boolean }
expect(inspect.hasAdapterFor('preinstalled-model')).toBe(true)
expect(inspect.hasAdapterFor('missing-model')).toBe(false)
await server.initialize({ cwd: storageDir, model: 'preinstalled-model' })
expect(inspect.hasAdapterFor('deepseek')).toBe(true)
expect(inspect.hasAdapterFor('missing-provider')).toBe(false)
await server.initialize({ cwd: storageDir, provider: 'deepseek', model: 'preinstalled-model' })
expect(ctx.get('llm')?.models().filter(model => model === 'preinstalled-model')).toEqual(['preinstalled-model'])
expect(ctx.get('llm')?.listProviders().filter(provider => provider.id === 'deepseek')).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
await server.shutdown()
} finally {
await ctx.fiber.dispose()
@@ -406,17 +744,18 @@ describe('HarnessSdkServer', () => {
}
})
it('registers a missing model when an LLM service already exists', async () => {
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, { models: ['other-model'] })
await ctx.plugin(LlmDeepSeek)
try {
const server = new HarnessSdkServer(ctx, new FakeTransport())
await server.initialize({ cwd: storageDir, model: 'new-model' })
await expect(server.initialize({ cwd: storageDir, provider: 'private', model: 'new-model' }))
.rejects.toThrow('no adapter registered for provider "private"')
expect(ctx.get('llm')?.models()).toEqual(expect.arrayContaining(['other-model', 'new-model']))
expect(ctx.get('llm')?.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
await server.shutdown()
} finally {
await ctx.fiber.dispose()
@@ -535,15 +874,15 @@ describe('HarnessSdkServer', () => {
const ctx = {
on: vi.fn(() => () => undefined),
agents: { create, get: () => undefined },
get: () => ({ models: () => ['model'] }),
get: () => ({ listProviders: () => [{ id: 'mock', name: 'Mock' }] }),
} as unknown as Context
const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
initialize(params: { cwd: string; model: string }): Promise<unknown>
initialize(params: { cwd: string; provider: string; model: string }): Promise<unknown>
getOrCreateSession(sessionId: string): Promise<unknown>
shutdown(): Promise<Record<string, never>>
}
await server.initialize({ cwd: '.', model: 'model' })
await server.initialize({ cwd: '.', provider: 'mock', model: 'model' })
await server.getOrCreateSession('relative')
expect(create).toHaveBeenCalledWith(expect.objectContaining({ meta: { cwd: process.cwd() } }))
@@ -585,6 +924,6 @@ describe('HarnessSdkServer', () => {
const server = new HarnessSdkServer(ctx, new FakeTransport())
await expect(server.shutdown()).rejects.toBe(listenerFailure)
expect(on).toHaveBeenCalledTimes(4)
expect(on).toHaveBeenCalledTimes(3)
})
})