diff options
| author | Adam Malczewski <[email protected]> | 2026-06-11 21:12:03 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-11 21:12:03 +0900 |
| commit | e7eada4802ceebd86c83bcd6e3eca70152e7f331 (patch) | |
| tree | 447095fd60b43980358d1565506f3ae2430e5f29 /packages/lsp/src/rpc.ts | |
| parent | 35937cee7f838e414eb8147c67205e01d85a4da0 (diff) | |
| download | dispatch-e7eada4802ceebd86c83bcd6e3eca70152e7f331.tar.gz dispatch-e7eada4802ceebd86c83bcd6e3eca70152e7f331.zip | |
feat(lsp,cwd): LSP integration + per-conversation cwd; fix cache-warming cache bust
LSP + per-conversation CWD feature:
- new bundled `lsp` extension: hand-rolled JSON-RPC codec (framing/rpc), lazy
one-server-per-(serverID,root), per-cwd config resolution, on-demand `lsp` tool
- `conversation-store`: getCwd/setCwd (cwdKey); `session-orchestrator` defaults a
turn's cwd from the store
- `transport-http`: cwd + lsp status endpoints; wire types in transport-contract
- host-bin: register lsp; config wiring
Cache-warming fix (the warm read 0% on the first reheat after a message):
- warm assembled tools under a different cwd than the real turn (a reheat sends no
cwd, and the warm service had no store fallback). The skills filter rewrites the
cwd-sensitive `load_skill` description, so the tools block — the first bytes of
the prompt-cache prefix — diverged and the cache missed entirely. Warm now
resolves cwd as opts.cwd ?? conversationStore.getCwd(), mirroring handleMessage.
- capture warm sends as `provider.request` spans flagged `warm:true` (thread a
child logger into providerOpts) so warm vs real bodies are diffable (obs §3.1).
- kernel logger: span-close now merges child-bound attrs like span-open, so a
`warm:true` query finds the closed span (with usage/status), not just the open.
Tests: warm forwards a warm-flagged logger; warm falls back to stored cwd; logger
open/close attr consistency. Full suite green (873).
Diffstat (limited to 'packages/lsp/src/rpc.ts')
| -rw-r--r-- | packages/lsp/src/rpc.ts | 132 |
1 files changed, 132 insertions, 0 deletions
diff --git a/packages/lsp/src/rpc.ts b/packages/lsp/src/rpc.ts new file mode 100644 index 0000000..45adf42 --- /dev/null +++ b/packages/lsp/src/rpc.ts @@ -0,0 +1,132 @@ +/** + * JSON-RPC connection over an injected write function. + * + * Provides sendRequest (correlated by id), sendNotification, onRequest, + * and onNotification. The caller feeds decoded JSON messages via `handleMessage`. + */ + +import { encode } from "./framing.js"; + +export type WriteFn = (bytes: Uint8Array) => void; + +export interface PendingRequest { + readonly resolve: (value: unknown) => void; + readonly reject: (reason: unknown) => void; +} + +export type RequestHandler = (params: unknown) => unknown | Promise<unknown>; +export type NotificationHandler = (params: unknown) => void; + +export interface JsonRpcMessage { + readonly jsonrpc: "2.0"; + readonly id?: number | string | undefined; + readonly method?: string | undefined; + readonly params?: unknown; + readonly result?: unknown; + readonly error?: + | { readonly code: number; readonly message: string; readonly data?: unknown } + | undefined; +} + +export class JsonRpcConnection { + private nextId = 1; + private pending = new Map<number | string, PendingRequest>(); + private requestHandlers = new Map<string, RequestHandler>(); + private notificationHandlers = new Map<string, NotificationHandler>(); + private write: WriteFn; + + constructor(write: WriteFn) { + this.write = write; + } + + sendRequest(method: string, params?: unknown): Promise<unknown> { + const id = this.nextId++; + const msg: JsonRpcMessage = { jsonrpc: "2.0", id, method, params }; + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }); + this.sendMessage(msg); + }); + } + + sendNotification(method: string, params?: unknown): void { + const msg: JsonRpcMessage = { jsonrpc: "2.0", method, params }; + this.sendMessage(msg); + } + + onRequest(method: string, handler: RequestHandler): void { + this.requestHandlers.set(method, handler); + } + + onNotification(method: string, handler: NotificationHandler): void { + this.notificationHandlers.set(method, handler); + } + + async handleMessage(json: string): Promise<void> { + const msg = JSON.parse(json) as JsonRpcMessage; + const { id, method } = msg; + + if (id !== undefined && method !== undefined) { + await this.handleIncomingRequest(id, method, msg.params); + } else if (method !== undefined) { + this.handleIncomingNotification(method, msg.params); + } else if (id !== undefined) { + this.handleResponse(id, msg); + } + } + + private sendMessage(msg: JsonRpcMessage): void { + this.write(encode(JSON.stringify(msg))); + } + + private handleResponse(id: number | string, msg: JsonRpcMessage): void { + const entry = this.pending.get(id); + if (!entry) return; + this.pending.delete(id); + if (msg.error) { + entry.reject(new Error(msg.error.message)); + } else { + entry.resolve(msg.result); + } + } + + private async handleIncomingRequest( + id: number | string, + method: string, + params: unknown, + ): Promise<void> { + const handler = this.requestHandlers.get(method); + if (!handler) { + this.sendMessage({ + jsonrpc: "2.0", + id, + error: { code: -32601, message: `Method not found: ${method}` }, + }); + return; + } + try { + const result = await handler(params); + this.sendMessage({ jsonrpc: "2.0", id, result }); + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); + this.sendMessage({ + jsonrpc: "2.0", + id, + error: { code: -32603, message }, + }); + } + } + + private handleIncomingNotification(method: string, params: unknown): void { + const handler = this.notificationHandlers.get(method); + if (handler) { + handler(params); + } + } + + dispose(): void { + for (const entry of this.pending.values()) { + entry.reject(new Error("Connection closed")); + } + this.pending.clear(); + } +} |
