Something went wrong. Try again.
forked niri
Something went wrong. Try again.
7.0 kB · 191 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192import http from "node:http"import { timingSafeEqual } from "node:crypto"import type { AddressInfo } from "node:net"import type { ClientToolHost } from "@nano/harness-core"import { isClientToolName, parseToolInvocation } from "@nano/harness-protocol"
export type ToolClientHttpServerOptions = { host: ClientToolHost listenHost?: string port?: number bodyLimitBytes?: number /** * Optional shared bearer token. When set, every request must present * `Authorization: Bearer <token>` or it is rejected with 401. This is the * auth boundary for cross-machine control-plane -> executor calls. */ authToken?: string}
export type ToolClientHttpServerAddress = { host: string port: number url: string}
function sendJson(response: http.ServerResponse, statusCode: number, body: unknown): void { const data = JSON.stringify(body) response.writeHead(statusCode, { "content-type": "application/json", "content-length": Buffer.byteLength(data), }) response.end(data)}
function beginOutputStream(response: http.ServerResponse): void { if (response.headersSent) return response.writeHead(200, { "content-type": "application/x-ndjson; charset=utf-8", "cache-control": "no-store", })}
function writeOutputStreamRecord(response: http.ServerResponse, body: unknown): void { if (response.destroyed || response.writableEnded) return beginOutputStream(response) response.write(`${JSON.stringify(body)}\n`)}
async function readJson(request: http.IncomingMessage, limit: number): Promise<unknown> { const chunks: Buffer[] = [] let bytes = 0 for await (const chunk of request) { const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk) bytes += buffer.length if (bytes > limit) throw new Error(`request body exceeds ${limit} bytes`) chunks.push(buffer) } const text = Buffer.concat(chunks).toString("utf8") return text.trim() ? JSON.parse(text) : null}
export class ToolClientHttpServer { private readonly toolHost: ClientToolHost private readonly listenHost: string private readonly port: number private readonly bodyLimitBytes: number private readonly authToken: string | null private readonly server: http.Server private address: ToolClientHttpServerAddress | null = null
constructor(options: ToolClientHttpServerOptions) { this.toolHost = options.host this.listenHost = options.listenHost?.trim() || "0.0.0.0" this.port = options.port ?? 3002 this.bodyLimitBytes = Math.max(1_024, options.bodyLimitBytes ?? 2_000_000) this.authToken = options.authToken?.trim() || null this.server = http.createServer((request, response) => { void this.handle(request, response).catch((error) => { if (!response.headersSent) { try { sendJson(response, 500, { error: error instanceof Error ? error.message : String(error) }) } catch {} } else { response.destroy(error instanceof Error ? error : new Error(String(error))) } }) }) }
async start(): Promise<ToolClientHttpServerAddress> { if (this.address) return this.address await this.toolHost.start() try { await new Promise<void>((resolve, reject) => { const onError = (error: Error) => reject(error) this.server.once("error", onError) this.server.listen(this.port, this.listenHost, () => { this.server.off("error", onError) resolve() }) }) } catch (error) { await this.toolHost.stop().catch(() => {}) throw error } const address = this.server.address() as AddressInfo const publicHost = this.listenHost === "0.0.0.0" || this.listenHost === "::" ? "127.0.0.1" : this.listenHost this.address = { host: this.listenHost, port: address.port, url: `http://${publicHost}:${address.port}` } return this.address }
async stop(): Promise<void> { await this.toolHost.stop() this.server.closeAllConnections() if (this.server.listening) { await new Promise<void>((resolve, reject) => this.server.close((error) => error ? reject(error) : resolve())) } this.address = null }
private authorized(request: http.IncomingMessage): boolean { if (!this.authToken) return true const header = request.headers.authorization ?? "" const prefix = "Bearer " if (!header.startsWith(prefix)) return false const presented = Buffer.from(header.slice(prefix.length)) const expected = Buffer.from(this.authToken) // timingSafeEqual throws on length mismatch; guard first. return presented.length === expected.length && timingSafeEqual(presented, expected) }
private async handle(request: http.IncomingMessage, response: http.ServerResponse): Promise<void> { try { if (!this.authorized(request)) { sendJson(response, 401, { error: "unauthorized" }) return } const url = new URL(request.url ?? "/", "http://client") if (request.method === "GET" && url.pathname === "/health") { sendJson(response, 200, { ok: true, capabilities: this.toolHost.getCapabilities(), workspace: this.toolHost.getWorkspace(), }) return }
const match = /^\/tools\/([^/]+)$/.exec(url.pathname) const tool = match ? decodeURIComponent(match[1] ?? "") : "" if (request.method !== "POST" || !isClientToolName(tool)) { sendJson(response, 404, { error: "not found" }) return }
const invocation = parseToolInvocation(await readJson(request, this.bodyLimitBytes)) if (!invocation || invocation.tool !== tool) { sendJson(response, 400, { error: "invalid tool invocation" }) return } const controller = new AbortController() const streamOutput = request.headers.accept?.includes("application/x-ndjson") === true const abort = () => controller.abort(new Error("tool client disconnected")) const abortOnClosedResponse = () => { if (!response.writableEnded) abort() } request.once("aborted", abort) response.once("close", abortOnClosedResponse) try { const result = await this.toolHost.execute( invocation, controller.signal, streamOutput ? (chunk) => writeOutputStreamRecord(response, { type: "tool.output", output: chunk }) : undefined, ) if (!response.destroyed) { if (streamOutput) { writeOutputStreamRecord(response, { type: "tool.result", result }) response.end() } else { sendJson(response, 200, result) } } } finally { request.removeListener("aborted", abort) response.removeListener("close", abortOnClosedResponse) } } catch (error) { const message = error instanceof Error ? error.message : String(error) sendJson(response, message.includes("request body exceeds") ? 413 : 400, { error: message }) } }}