| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283 |
- /**
- * SDK-facing JSON-RPC plugin over stdio. An external `cordis.yml` decides
- * whether to load it; see the single-executable RFC and package README.
- * Stdout is reserved for protocol frames, so the tree must not load a stdout logger.
- * This plugin answers `shutdown`, disposes its own fiber, and exits 0; the app bin
- * owns EOF and signal exits. Keep named plugin exports with no default export so
- * Loader `unwrapExports` preserves `name`, `inject`, `Config`, and `apply`.
- *
- * @module @deepseek-ai/dsh-jsonrpc
- */
- import type { Context } from 'cordis'
- import type { Readable, Writable } from 'node:stream'
- import Schema from 'schemastery'
- 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().
- export const inject = ['agents']
- /** Runtime-only test seams; no field is configurable from `cordis.yml`. */
- export interface JsonRpcConfig {
- /** Transport input override; production uses `process.stdin`. */
- input?: Readable
- /** Transport output override; production uses `process.stdout`. */
- output?: Writable
- /** Process-exit override; production uses `process.exit`. */
- exit?: (code: number) => void
- }
- export const Config: Schema<JsonRpcConfig> = Schema.object({})
- /**
- * Serve SDK requests over the configured streams. Effect disposal shuts down
- * SDK-created agents and closes the transport. A `shutdown` response is flushed
- * before this plugin's fiber is disposed and the process exits 0; the app bin
- * owns root-context disposal for EOF and signals.
- */
- export function apply(ctx: Context, config: JsonRpcConfig): void {
- // The later transport callback must dispose this plugin's fiber, not its ambient context.
- const fiber = ctx.fiber
- /* v8 ignore next -- production stdio wiring; tests always inject the runtime seams */
- const input = config.input ?? process.stdin
- /* v8 ignore next -- production stdio wiring; tests always inject the runtime seams */
- const output = config.output ?? process.stdout
- /* v8 ignore next -- production exit wiring; tests always inject the runtime seams */
- const exit = config.exit ?? ((code: number): void => { process.exit(code) })
- const transport = new JsonRpcLineTransport(input, output)
- const server = new HarnessSdkServer(ctx, transport)
- // Share one exit task and attempt flush and disposal independently before exiting.
- let exitTask: Promise<void> | undefined
- const disposeAndExit = (): Promise<void> => {
- exitTask ??= (async () => {
- await Promise.allSettled([Promise.resolve().then(() => transport.flush())])
- await Promise.allSettled([Promise.resolve().then(() => fiber.dispose())])
- exit(0)
- })()
- return exitTask
- }
- transport.onRequest(async (method, params) => {
- const result = await server.handleRequest(method, params)
- if (method === 'shutdown') {
- // Run after the handler result is written; the task then flushes, disposes, and exits.
- setImmediate(() => { void disposeAndExit() })
- }
- return result
- })
- ctx.effect(() => {
- transport.start()
- return async () => {
- await server.shutdown()
- transport.close()
- }
- }, 'jsonrpc.serve')
- }
|