Merge newer master into skill catalog hot refresh
This commit is contained in:
Tianyi Cui
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 9297147b7a53739c46871b527ce51868eac1e244
|
||||
README.zh.md: 62596197b95729215408dfc1496c129ace6cbad4
|
||||
README.md: 96dae46c9c6ecce6643bb408a5e57c2db2275a83
|
||||
README.zh.md: 2c257b13a15ad3c6c1bf0c4dd44a04e308e8f0b0
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
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.
|
||||
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; the transport and the named wire types live in [`dsh-sdk-protocol`](../../sdk/sdk-protocol/README.md), shared with the client SDKs; [`jsonrpc-demo`](../../examples/jsonrpc-demo/README.md) supplies the surrounding `cordis.yml` application.
|
||||
|
||||
## Wiring
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
`jsonrpc` 插件通过 stdio 提供以换行符分隔的 JSON-RPC，使进程外 SDK 客户端能够驱动 harness agent（智能体）。[`HarnessSdkServer`](src/server.ts) 持有协议方法和通知；[`jsonrpc-demo`](../../examples/jsonrpc-demo/README.md) 提供外围的 `cordis.yml` 应用。
|
||||
`jsonrpc` 插件通过 stdio 提供以换行符分隔的 JSON-RPC，使进程外 SDK 客户端能够驱动 harness agent（智能体）。[`HarnessSdkServer`](src/server.ts) 持有协议方法和通知；传输与具名线类型位于 [`dsh-sdk-protocol`](../../sdk/sdk-protocol/README.md)，与客户端 SDK 共享；[`jsonrpc-demo`](../../examples/jsonrpc-demo/README.md) 提供外围的 `cordis.yml` 应用。
|
||||
|
||||
## 组装
|
||||
|
||||
|
||||
@@ -35,6 +35,7 @@
|
||||
"@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-sdk-protocol": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-subagent": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
@@ -47,6 +48,7 @@
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm-deepseek": "workspace:^",
|
||||
"@deepseek-ai/dsh-scope": "workspace:^",
|
||||
"@deepseek-ai/dsh-sdk-protocol": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
|
||||
"@deepseek-ai/dsh-subagent": "workspace:^",
|
||||
|
||||
@@ -12,11 +12,10 @@
|
||||
import type { Context } from 'cordis'
|
||||
import type { Readable, Writable } from 'node:stream'
|
||||
import Schema from 'schemastery'
|
||||
import { JsonRpcLineTransport } from '@deepseek-ai/dsh-sdk-protocol'
|
||||
import { HarnessSdkServer } from './server.ts'
|
||||
import { JsonRpcLineTransport } from './transport.ts'
|
||||
|
||||
export * from './server.ts'
|
||||
export * from './transport.ts'
|
||||
|
||||
export const name = 'jsonrpc'
|
||||
// Only the agent factory is required; initialize reads the optional LLM seam with ctx.get().
|
||||
|
||||
@@ -7,44 +7,23 @@
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { resolve } from 'node:path'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { Agent, AgentHandle } from '@deepseek-ai/dsh-agent'
|
||||
import { carrierKeyOf, type Scoped } from '@deepseek-ai/dsh-scope'
|
||||
import { findLastMessageTurnEnd, 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'
|
||||
|
||||
/** 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
|
||||
}
|
||||
|
||||
/** 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 }
|
||||
}
|
||||
|
||||
/** 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
|
||||
/** The prompt content blocks, sent verbatim as the user message. */
|
||||
contentBlocks: ContentBlock[]
|
||||
}
|
||||
|
||||
/** 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
|
||||
}
|
||||
import type {
|
||||
InitializeParams,
|
||||
InitializeResult,
|
||||
JsonRpcTransportPeer,
|
||||
SessionEventNotification,
|
||||
SessionFinishedNotification,
|
||||
SessionPromptParams,
|
||||
SessionPromptResult,
|
||||
SubagentFinishedNotification,
|
||||
SubagentStartedNotification,
|
||||
} from '@deepseek-ai/dsh-sdk-protocol'
|
||||
|
||||
interface SessionRecord {
|
||||
handle: AgentHandle
|
||||
@@ -97,15 +76,17 @@ export class HarnessSdkServer {
|
||||
rec.lastTurnEnd = event.data.reason
|
||||
}
|
||||
}
|
||||
this.transport.notify('session.event', { sessionId: String(session.id), event })
|
||||
const payload: SessionEventNotification = { sessionId: String(session.id), event }
|
||||
this.transport.notify('session.event', payload)
|
||||
}))
|
||||
this.disposers.push(ctx.on('session/created', (session) => {
|
||||
const parentSession = session.header.parentSession
|
||||
if (parentSession === undefined) return
|
||||
this.transport.notify('subagent.started', {
|
||||
const payload: SubagentStartedNotification = {
|
||||
parentSessionId: String(parentSession),
|
||||
childSessionId: String(session.id),
|
||||
})
|
||||
}
|
||||
this.transport.notify('subagent.started', payload)
|
||||
}))
|
||||
this.disposers.push(ctx.on('subagent/end', function (this: Scoped<SubagentService>, info: SubagentRunEndInfo) {
|
||||
const parent = subagentParentOf(this)
|
||||
@@ -113,7 +94,7 @@ export class HarnessSdkServer {
|
||||
// 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', {
|
||||
const payload: SubagentFinishedNotification = {
|
||||
provider: info.provider,
|
||||
agentId: String(info.id),
|
||||
parentSessionId: String(parent.session.id),
|
||||
@@ -121,7 +102,8 @@ export class HarnessSdkServer {
|
||||
status: successStatus(info.stopReason, serverOptions),
|
||||
stopReason: info.stopReason,
|
||||
...(info.lastAssistantMessage === undefined ? {} : { lastAssistantMessage: info.lastAssistantMessage }),
|
||||
})
|
||||
}
|
||||
transport.notify('subagent.finished', payload)
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -160,12 +142,12 @@ export class HarnessSdkServer {
|
||||
rec.lastTurnEnd = undefined
|
||||
rec.handle.agent.followup({ content: params.contentBlocks, source: { kind: 'user' } })
|
||||
await rec.handle.agent.whenIdle()
|
||||
const status = this.finishedStatus(rec.lastTurnEnd)
|
||||
this.transport.notify('session.finished', {
|
||||
const payload: SessionFinishedNotification = {
|
||||
sessionId: params.sessionId,
|
||||
status,
|
||||
status: this.finishedStatus(rec.lastTurnEnd),
|
||||
reason: rec.lastTurnEnd,
|
||||
})
|
||||
}
|
||||
this.transport.notify('session.finished', payload)
|
||||
return { accepted: true }
|
||||
} finally {
|
||||
rec.activePrompt = false
|
||||
|
||||
@@ -1,223 +0,0 @@
|
||||
/**
|
||||
* Newline-delimited JSON-RPC 2.0 over byte streams. Frames with `id` and
|
||||
* `method` are requests, `id` alone is a response, and `method` alone is a
|
||||
* notification. Malformed lines are ignored; handler failures become error frames.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-jsonrpc/transport
|
||||
*/
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Readable, Writable } from 'node:stream'
|
||||
import { StringDecoder } from 'node:string_decoder'
|
||||
|
||||
type JsonRpcId = string | number
|
||||
type RequestHandler = (method: string, params: Record<string, unknown>) => Promise<unknown>
|
||||
type NotificationHandler = (method: string, params: Record<string, unknown>) => void
|
||||
|
||||
/**
|
||||
* Outbound request and notification surface used by {@link HarnessSdkServer}.
|
||||
*/
|
||||
export interface JsonRpcTransportPeer {
|
||||
/**
|
||||
* Send a request and await its response.
|
||||
* @param method - the JSON-RPC method name.
|
||||
* @param params - the request parameters object.
|
||||
* @returns the result; rejects on an error response, write failure, or closure.
|
||||
*/
|
||||
request(method: string, params: Record<string, unknown>): Promise<unknown>
|
||||
/**
|
||||
* Send a notification; omitted params produce no `params` member.
|
||||
* @param method - the JSON-RPC method name.
|
||||
* @param params - the optional notification parameters object.
|
||||
*/
|
||||
notify(method: string, params?: Record<string, unknown>): void
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: unknown) => void
|
||||
reject: (error: Error) => void
|
||||
}
|
||||
|
||||
/**
|
||||
* Line-delimited endpoint over caller-owned streams. {@link start} attaches
|
||||
* listeners; {@link close} detaches them and rejects pending requests without
|
||||
* destroying the streams. Missing request handlers return `-32601`; handler
|
||||
* failures return `-32603`. Notifications without a handler are dropped.
|
||||
*/
|
||||
export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
private buffer = ''
|
||||
private readonly decoder = new StringDecoder('utf8')
|
||||
private started = false
|
||||
private requestHandler: RequestHandler | undefined
|
||||
private notificationHandler: NotificationHandler | undefined
|
||||
private readonly pending = new Map<JsonRpcId, PendingRequest>()
|
||||
|
||||
constructor(
|
||||
private readonly input: Readable,
|
||||
private readonly output: Writable,
|
||||
) {}
|
||||
|
||||
/** Attach the input listeners and begin reading frames. Idempotent. */
|
||||
start(): void {
|
||||
if (this.started) return
|
||||
this.started = true
|
||||
this.input.on('data', this.onData)
|
||||
this.input.on('error', this.onInputError)
|
||||
this.input.on('end', this.onInputEnd)
|
||||
}
|
||||
|
||||
/**
|
||||
* Detach listeners and reject pending requests. Safe before {@link start}.
|
||||
*/
|
||||
close(): void {
|
||||
this.input.off('data', this.onData)
|
||||
this.input.off('error', this.onInputError)
|
||||
this.input.off('end', this.onInputEnd)
|
||||
this.failPending(new Error('JSON-RPC transport closed'))
|
||||
}
|
||||
|
||||
/**
|
||||
* Install the request handler, replacing any prior handler.
|
||||
* @param handler - resolves to the response `result`; a rejection becomes a
|
||||
* `-32603` error response carrying the message.
|
||||
*/
|
||||
onRequest(handler: RequestHandler): void {
|
||||
this.requestHandler = handler
|
||||
}
|
||||
|
||||
/**
|
||||
* Install the notification handler, replacing any prior handler.
|
||||
* @param handler - invoked per notification with the method and normalized
|
||||
* params object.
|
||||
*/
|
||||
onNotification(handler: NotificationHandler): void {
|
||||
this.notificationHandler = handler
|
||||
}
|
||||
|
||||
request(method: string, params: Record<string, unknown>): Promise<unknown> {
|
||||
const id = `req_${randomUUID().replaceAll('-', '')}`
|
||||
const message = { jsonrpc: '2.0', id, method, params }
|
||||
return new Promise((resolve, reject) => {
|
||||
this.pending.set(id, { resolve, reject })
|
||||
try {
|
||||
this.write(message)
|
||||
} catch (error) {
|
||||
this.pending.delete(id)
|
||||
reject(error instanceof Error ? error : new Error(String(error)))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
notify(method: string, params?: Record<string, unknown>): void {
|
||||
this.write(params === undefined ? { jsonrpc: '2.0', method } : { jsonrpc: '2.0', method, params })
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for prior frame write callbacks. The empty barrier emits no bytes.
|
||||
* @returns a promise that settles with the output write callback.
|
||||
*/
|
||||
flush(): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
this.output.write('', (error) => {
|
||||
if (error) reject(error)
|
||||
else resolve()
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
private readonly onData = (chunk: Buffer | string): void => {
|
||||
this.buffer += typeof chunk === 'string' ? chunk : this.decoder.write(chunk)
|
||||
this.drainLines()
|
||||
}
|
||||
|
||||
private drainLines(): void {
|
||||
for (;;) {
|
||||
const newline = this.buffer.indexOf('\n')
|
||||
if (newline < 0) break
|
||||
const line = this.buffer.slice(0, newline).trim()
|
||||
this.buffer = this.buffer.slice(newline + 1)
|
||||
if (!line) continue
|
||||
void this.handleLine(line)
|
||||
}
|
||||
}
|
||||
|
||||
private readonly onInputError = (error: Error): void => {
|
||||
this.failPending(error)
|
||||
}
|
||||
|
||||
private readonly onInputEnd = (): void => {
|
||||
this.buffer += this.decoder.end()
|
||||
this.drainLines()
|
||||
this.failPending(new Error('JSON-RPC input closed'))
|
||||
}
|
||||
|
||||
private async handleLine(line: string): Promise<void> {
|
||||
let message: unknown
|
||||
try {
|
||||
message = JSON.parse(line)
|
||||
} catch {
|
||||
// Only JSON syntax errors reach this catch; malformed peer lines are ignored.
|
||||
return
|
||||
}
|
||||
if (!message || typeof message !== 'object') return
|
||||
const frame = message as Record<string, unknown>
|
||||
const id = frame.id
|
||||
const method = frame.method
|
||||
if ((typeof id === 'string' || typeof id === 'number') && typeof method === 'string') {
|
||||
await this.handleIncomingRequest(id, method, objectParams(frame.params))
|
||||
return
|
||||
}
|
||||
if (typeof id === 'string' || typeof id === 'number') {
|
||||
this.handleIncomingResponse(id, frame)
|
||||
return
|
||||
}
|
||||
if (typeof method === 'string') {
|
||||
this.notificationHandler?.(method, objectParams(frame.params))
|
||||
}
|
||||
}
|
||||
|
||||
private async handleIncomingRequest(id: JsonRpcId, method: string, params: Record<string, unknown>): Promise<void> {
|
||||
const handler = this.requestHandler
|
||||
if (!handler) {
|
||||
this.writeError(id, -32601, `method not found: ${method}`)
|
||||
return
|
||||
}
|
||||
try {
|
||||
const result = await handler(method, params)
|
||||
this.write({ jsonrpc: '2.0', id, result })
|
||||
} catch (error) {
|
||||
this.writeError(id, -32603, error instanceof Error ? error.message : String(error))
|
||||
}
|
||||
}
|
||||
|
||||
private handleIncomingResponse(id: JsonRpcId, frame: Record<string, unknown>): void {
|
||||
const pending = this.pending.get(id)
|
||||
if (!pending) return
|
||||
this.pending.delete(id)
|
||||
if (frame.error && typeof frame.error === 'object') {
|
||||
const error = frame.error as Record<string, unknown>
|
||||
pending.reject(new Error(typeof error.message === 'string' ? error.message : 'JSON-RPC error'))
|
||||
return
|
||||
}
|
||||
pending.resolve(frame.result)
|
||||
}
|
||||
|
||||
private writeError(id: JsonRpcId, code: number, message: string): void {
|
||||
this.write({ jsonrpc: '2.0', id, error: { code, message } })
|
||||
}
|
||||
|
||||
private write(message: Record<string, unknown>): void {
|
||||
this.output.write(`${JSON.stringify(message)}\n`)
|
||||
}
|
||||
|
||||
private failPending(error: Error): void {
|
||||
const pending = [...this.pending.values()]
|
||||
this.pending.clear()
|
||||
for (const waiter of pending) waiter.reject(error)
|
||||
}
|
||||
}
|
||||
|
||||
/** Normalize JSON-RPC `params` to a plain object (arrays and scalars collapse to `{}`). */
|
||||
function objectParams(params: unknown): Record<string, unknown> {
|
||||
return params && typeof params === 'object' && !Array.isArray(params) ? params as Record<string, unknown> : {}
|
||||
}
|
||||
@@ -12,17 +12,18 @@ 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 { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts'
|
||||
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: Record<string, unknown>): Promise<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?: Record<string, unknown>): void {
|
||||
this.notifications.push(params === undefined ? { method } : { method, params })
|
||||
notify(method: string, params?: object): void {
|
||||
this.notifications.push(params === undefined ? { method } : { method, params: params as Record<string, unknown> })
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,260 +0,0 @@
|
||||
import { once } from 'node:events'
|
||||
import { PassThrough, Writable } from 'node:stream'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { JsonRpcLineTransport } from '../src/index.ts'
|
||||
|
||||
function transportPair() {
|
||||
const aToB = new PassThrough()
|
||||
const bToA = new PassThrough()
|
||||
const a = new JsonRpcLineTransport(bToA, aToB)
|
||||
const b = new JsonRpcLineTransport(aToB, bToA)
|
||||
return { a, b, aToB, bToA }
|
||||
}
|
||||
|
||||
describe('JsonRpcLineTransport', () => {
|
||||
it('supports bidirectional requests and notifications over newline-delimited JSON-RPC', async () => {
|
||||
const { a, b } = transportPair()
|
||||
const notifications: Record<string, unknown>[] = []
|
||||
|
||||
a.onRequest(async (method, params) => {
|
||||
expect(method).toBe('echo')
|
||||
return { echoed: params }
|
||||
})
|
||||
b.onNotification((method, params) => {
|
||||
notifications.push({ method, params })
|
||||
})
|
||||
a.start()
|
||||
b.start()
|
||||
|
||||
const response = await b.request('echo', { value: 42 })
|
||||
expect(response).toEqual({ echoed: { value: 42 } })
|
||||
|
||||
a.notify('session.finished', { sessionId: 'main', status: 'ok' })
|
||||
a.notify('heartbeat')
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
expect(notifications).toEqual([
|
||||
{ method: 'session.finished', params: { sessionId: 'main', status: 'ok' } },
|
||||
{ method: 'heartbeat', params: {} },
|
||||
])
|
||||
|
||||
a.close()
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('reports JSON-RPC request errors from the remote peer', async () => {
|
||||
const { a, b } = transportPair()
|
||||
a.onRequest(async () => {
|
||||
throw new Error('handler boom')
|
||||
})
|
||||
a.start()
|
||||
b.start()
|
||||
|
||||
await expect(b.request('explode', {})).rejects.toThrow('handler boom')
|
||||
|
||||
a.close()
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('stringifies non-Error request handler failures', async () => {
|
||||
const { a, b } = transportPair()
|
||||
a.onRequest(async () => {
|
||||
throw 'string boom'
|
||||
})
|
||||
a.start()
|
||||
b.start()
|
||||
|
||||
await expect(b.request('explode-string', {})).rejects.toThrow('string boom')
|
||||
|
||||
a.close()
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('reports method-not-found when no request handler is installed', async () => {
|
||||
const { a, b } = transportPair()
|
||||
a.start()
|
||||
b.start()
|
||||
|
||||
await expect(b.request('missing', {})).rejects.toThrow('method not found: missing')
|
||||
|
||||
a.close()
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('normalizes non-object request params and ignores notifications without a handler', async () => {
|
||||
const { aToB, bToA, b } = transportPair()
|
||||
const seen: Record<string, unknown>[] = []
|
||||
b.onRequest(async (method, params) => {
|
||||
seen.push({ method, params })
|
||||
return { ok: true }
|
||||
})
|
||||
b.start()
|
||||
|
||||
aToB.write('{"jsonrpc":"2.0","method":"ignored"}\n')
|
||||
aToB.write('{"jsonrpc":"2.0","id":7,"method":"array-params","params":[]}\n')
|
||||
const chunk = (await once(bToA, 'data'))[0] as Buffer | string
|
||||
|
||||
expect(seen).toEqual([{ method: 'array-params', params: {} }])
|
||||
expect(JSON.parse(String(chunk))).toEqual({ jsonrpc: '2.0', id: 7, result: { ok: true } })
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('ignores malformed frames and accepts notifications without params', async () => {
|
||||
const { aToB, b } = transportPair()
|
||||
const notifications: Record<string, unknown>[] = []
|
||||
b.onNotification((method, params) => {
|
||||
notifications.push({ method, params })
|
||||
})
|
||||
b.start()
|
||||
b.start()
|
||||
|
||||
aToB.write('not json\n')
|
||||
aToB.write('\n')
|
||||
aToB.write('null\n')
|
||||
aToB.write('{"jsonrpc":"2.0","params":{}}\n')
|
||||
aToB.write('{"jsonrpc":"2.0","method":"tick"}\n')
|
||||
aToB.emit('data', '{"jsonrpc":"2.0","method":"string-chunk"}\n')
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
|
||||
expect(notifications).toEqual([
|
||||
{ method: 'tick', params: {} },
|
||||
{ method: 'string-chunk', params: {} },
|
||||
])
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('preserves multibyte UTF-8 characters split across Buffer chunks', async () => {
|
||||
const input = new PassThrough()
|
||||
const output = new PassThrough()
|
||||
const transport = new JsonRpcLineTransport(input, output)
|
||||
const notifications: Record<string, unknown>[] = []
|
||||
transport.onNotification((method, params) => { notifications.push({ method, params }) })
|
||||
transport.start()
|
||||
|
||||
const frame = Buffer.from(`${JSON.stringify({ jsonrpc: '2.0', method: 'message', params: { text: '你好' } })}\n`)
|
||||
const character = Buffer.from('你')
|
||||
const characterStart = frame.indexOf(character)
|
||||
expect(characterStart).toBeGreaterThanOrEqual(0)
|
||||
input.write(frame.subarray(0, characterStart + 1))
|
||||
input.write(frame.subarray(characterStart + 1))
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
|
||||
expect(notifications).toEqual([{ method: 'message', params: { text: '你好' } }])
|
||||
transport.close()
|
||||
})
|
||||
|
||||
it('flush waits for all earlier output writes', async () => {
|
||||
const events: string[] = []
|
||||
const output = new Writable({
|
||||
write(chunk: Buffer, _encoding, callback) {
|
||||
const label = chunk.length === 0 ? 'barrier' : 'frame'
|
||||
events.push(`start:${label}`)
|
||||
setTimeout(() => {
|
||||
events.push(`finish:${label}`)
|
||||
callback()
|
||||
}, 5)
|
||||
},
|
||||
})
|
||||
const transport = new JsonRpcLineTransport(new PassThrough(), output)
|
||||
|
||||
transport.notify('tick')
|
||||
await transport.flush()
|
||||
|
||||
expect(events).toEqual([
|
||||
'start:frame',
|
||||
'finish:frame',
|
||||
'start:barrier',
|
||||
'finish:barrier',
|
||||
])
|
||||
transport.close()
|
||||
})
|
||||
|
||||
it('reports an output callback failure from flush', async () => {
|
||||
const output = {
|
||||
write(_chunk: string, callback?: (error?: Error) => void) {
|
||||
callback?.(new Error('flush failed'))
|
||||
return true
|
||||
},
|
||||
}
|
||||
const transport = new JsonRpcLineTransport(new PassThrough(), output as never)
|
||||
|
||||
await expect(transport.flush()).rejects.toThrow('flush failed')
|
||||
})
|
||||
|
||||
it('rejects pending requests when the input closes', async () => {
|
||||
const { aToB, b } = transportPair()
|
||||
b.start()
|
||||
|
||||
const pending = b.request('never-replies', {})
|
||||
aToB.end()
|
||||
|
||||
await expect(pending).rejects.toThrow('JSON-RPC input closed')
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('rejects pending requests when the input errors', async () => {
|
||||
const { aToB, b } = transportPair()
|
||||
b.start()
|
||||
|
||||
const pending = b.request('never-replies', {})
|
||||
aToB.emit('error', new Error('input broke'))
|
||||
|
||||
await expect(pending).rejects.toThrow('input broke')
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('rejects pending requests when the transport closes', async () => {
|
||||
const { b } = transportPair()
|
||||
|
||||
const pending = b.request('never-replies', {})
|
||||
b.close()
|
||||
|
||||
await expect(pending).rejects.toThrow('JSON-RPC transport closed')
|
||||
})
|
||||
|
||||
it('rejects a request when writing the frame throws', async () => {
|
||||
const input = new PassThrough()
|
||||
const output = {
|
||||
write() {
|
||||
throw new Error('write exploded')
|
||||
},
|
||||
}
|
||||
const transport = new JsonRpcLineTransport(input, output as never)
|
||||
|
||||
await expect(transport.request('write-fails', {})).rejects.toThrow('write exploded')
|
||||
})
|
||||
|
||||
it('stringifies non-Error write failures', async () => {
|
||||
const input = new PassThrough()
|
||||
const output = {
|
||||
write() {
|
||||
throw 'write string'
|
||||
},
|
||||
}
|
||||
const transport = new JsonRpcLineTransport(input, output as never)
|
||||
|
||||
await expect(transport.request('write-fails', {})).rejects.toThrow('write string')
|
||||
})
|
||||
|
||||
it('uses a fallback message for malformed JSON-RPC error responses', async () => {
|
||||
const { aToB, bToA, b } = transportPair()
|
||||
b.start()
|
||||
|
||||
const pending = b.request('remote-error', {})
|
||||
const requestChunk = (await once(bToA, 'data'))[0] as Buffer | string
|
||||
const request = JSON.parse(String(requestChunk)) as { id: string }
|
||||
aToB.write(`${JSON.stringify({ jsonrpc: '2.0', id: request.id, error: {} })}\n`)
|
||||
|
||||
await expect(pending).rejects.toThrow('JSON-RPC error')
|
||||
b.close()
|
||||
})
|
||||
|
||||
it('ignores responses that do not match a pending request', async () => {
|
||||
const { aToB, b } = transportPair()
|
||||
b.start()
|
||||
|
||||
aToB.write('{"jsonrpc":"2.0","id":"unknown","result":{"ignored":true}}\n')
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
|
||||
b.close()
|
||||
})
|
||||
})
|
||||
@@ -26,6 +26,9 @@
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../sdk/sdk-protocol"
|
||||
},
|
||||
{
|
||||
"path": "../../subagent/subagent"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user