mcp: serialize JSON-RPC send+read with per-connection mutex
ober
d64174b02709683a948c4ba3ce6e45d979531874
--- a/src/jcode/mcp/client.ss +++ b/src/jcode/mcp/client.ss @@ -20,13 +20,23 @@ :jcode/core/config :jcode/tool/registry :jerboa/core - :jerboa/runtime) + :jerboa/runtime + ;; Chez's raw mutex so `with-mutex` (a Chez macro) works on it. + ;; The prelude shadows make-mutex with a Gerbil wrapper that + ;; with-mutex cannot operate on. + (rename (only (chezscheme) make-mutex) + (make-mutex chez-make-mutex))) (def logger (make-logger "mcp")) ;; --- MCP server state --- -(defstruct mcp-conn (name to-stdin from-stdout from-stderr pid next-id)) +;; lock serializes mcp-send! send+read pairs on this connection. +;; Without it, parallel tool dispatch (agent-loop spawns one thread +;; per tool call) has multiple threads writing to the same stdin and +;; reading from the same stdout — JSON-RPC framing corrupts and both +;; threads block forever in get-line. +(defstruct mcp-conn (name to-stdin from-stdout from-stderr pid next-id lock)) (def *mcp-servers* '()) ;; Cached tool count per server name. Populated by register-mcp-tools at init. @@ -42,7 +52,8 @@ (let ((cmd-str (string-join (cons command args) " "))) (let-values (((to-stdin from-stdout from-stderr pid) (open-process-ports cmd-str 'block (make-transcoder (utf-8-codec))))) - (let ((conn (make-mcp-conn name to-stdin from-stdout from-stderr pid 1))) + (let ((conn (make-mcp-conn name to-stdin from-stdout from-stderr pid 1 + (chez-make-mutex)))) (set! *mcp-servers* (cons conn *mcp-servers*)) conn)))) @@ -82,30 +93,34 @@ id)) (def (mcp-send! conn method params) - "Send a JSON-RPC request and return the result." - (let* ((id (mcp-next-id! conn)) - (msg (make-hash-table))) - (hash-put! msg "jsonrpc" "2.0") - (hash-put! msg "id" id) - (hash-put! msg "method" method) - (hash-put! msg "params" (or params (make-hash-table))) - (let ((json-str (json-object->string msg))) - (log-debug logger "send" `((method . ,method) (id . ,id))) - (display json-str (mcp-conn-to-stdin conn)) - (newline (mcp-conn-to-stdin conn)) - (flush-output-port (mcp-conn-to-stdin conn)) - (mcp-read-response conn id)))) + "Send a JSON-RPC request and return the result. + Holds the connection lock across send+read so parallel tool dispatch + doesn't interleave on the shared stdio pipes." + (with-mutex (mcp-conn-lock conn) + (let* ((id (mcp-next-id! conn)) + (msg (make-hash-table))) + (hash-put! msg "jsonrpc" "2.0") + (hash-put! msg "id" id) + (hash-put! msg "method" method) + (hash-put! msg "params" (or params (make-hash-table))) + (let ((json-str (json-object->string msg))) + (log-debug logger "send" `((method . ,method) (id . ,id))) + (display json-str (mcp-conn-to-stdin conn)) + (newline (mcp-conn-to-stdin conn)) + (flush-output-port (mcp-conn-to-stdin conn)) + (mcp-read-response conn id))))) (def (mcp-notify! conn method params) "Send a JSON-RPC notification (no id, no response expected)." - (let ((msg (make-hash-table))) - (hash-put! msg "jsonrpc" "2.0") - (hash-put! msg "method" method) - (hash-put! msg "params" (or params (make-hash-table))) - (let ((json-str (json-object->string msg))) - (display json-str (mcp-conn-to-stdin conn)) - (newline (mcp-conn-to-stdin conn)) - (flush-output-port (mcp-conn-to-stdin conn))))) + (with-mutex (mcp-conn-lock conn) + (let ((msg (make-hash-table))) + (hash-put! msg "jsonrpc" "2.0") + (hash-put! msg "method" method) + (hash-put! msg "params" (or params (make-hash-table))) + (let ((json-str (json-object->string msg))) + (display json-str (mcp-conn-to-stdin conn)) + (newline (mcp-conn-to-stdin conn)) + (flush-output-port (mcp-conn-to-stdin conn)))))) (def (mcp-read-response conn expected-id) "Read lines from stdout until we get a response matching expected-id."