feat(bridge): JSON-RPC pipe + SSE writer (forgejo-mcp-broker-am1)
Adds internal/bridge: connects HTTP-side MCP clients to a stdio-side
child via JSON-RPC framing. Decoupled from internal/supervisor — takes
io.Writer + *bufio.Reader + done channel directly so it tests cleanly
with io.Pipe pairs and could later wrap something other than a child
process.
Routing model: one reader goroutine consumes child stdout line-by-line.
Each line is parsed only enough to extract the JSON-RPC `id` field
(string/number/null kept as raw JSON, so `1` and `"1"` don't collide).
HTTP requests register a per-id waiter channel before forwarding their
body to the child; the reader delivers the response to whichever waiter
matches. Concurrent in-flight requests are safe; a duplicate id while
the first is still pending returns 409.
HandleSSE response shapes:
- request with id + child reply → 200 text/event-stream, one
`event: message` SSE event carrying the JSON-RPC response
- request without id (notification) → 204 No Content (no waiter
needed; MCP notifications are fire-and-forget)
- empty body → 400
- duplicate in-flight id → 409
- send-to-child fails → 502
- client disconnect mid-wait → bridge cleans up its waiter; child
keeps running, other in-flight requests unaffected
- child exits before reply → SSE `error` event with reason=child_exited
Tests cover all of the above plus stale unsolicited replies, malformed
lines from the child, and reader robustness across both. 90.0%
coverage. The remaining gap is splitLines' empty-data branch (only
reachable if the child sends a literal `\n` line).
Closes forgejo-mcp-broker-am1.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
7be7f5e199
commit
bd68d7ed06
3 changed files with 703 additions and 2 deletions
261
internal/bridge/bridge.go
Normal file
261
internal/bridge/bridge.go
Normal file
|
|
@ -0,0 +1,261 @@
|
|||
// Package bridge connects an HTTP-side MCP client (e.g. Claude.ai over
|
||||
// streamable HTTP) to a stdio-side MCP server (e.g. forgejo-mcp managed
|
||||
// by supervisor) by piping JSON-RPC messages.
|
||||
//
|
||||
// The bridge is protocol-opaque: it parses each line only enough to extract
|
||||
// the JSON-RPC `id` field for response routing. Everything else passes
|
||||
// through verbatim. This is the design.md §5.3 "raw bytes through"
|
||||
// approach — MCP is JSON-RPC 2.0 on both transports, so we can pipe
|
||||
// without understanding semantics.
|
||||
//
|
||||
// One Bridge wraps one child. Multiple HTTP requests can be in flight
|
||||
// concurrently; responses are routed back to the right HTTP handler by
|
||||
// matching their `id` field against pending waiters.
|
||||
package bridge
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// Bridge owns the message routing for a single supervised child.
|
||||
type Bridge struct {
|
||||
stdin io.Writer
|
||||
stdout *bufio.Reader
|
||||
done <-chan struct{}
|
||||
log *slog.Logger
|
||||
|
||||
writeMu sync.Mutex // serializes writes to stdin
|
||||
pending sync.Map // string(id JSON) -> chan []byte
|
||||
started atomic.Bool
|
||||
closed atomic.Bool
|
||||
}
|
||||
|
||||
// New constructs a Bridge over the supplied pipes. The caller is
|
||||
// responsible for keeping stdin/stdout alive for the bridge's lifetime
|
||||
// (typically by holding the supervisor.Child).
|
||||
func New(stdin io.Writer, stdout *bufio.Reader, done <-chan struct{}, log *slog.Logger) *Bridge {
|
||||
if log == nil {
|
||||
log = slog.New(slog.DiscardHandler)
|
||||
}
|
||||
return &Bridge{stdin: stdin, stdout: stdout, done: done, log: log}
|
||||
}
|
||||
|
||||
// Start kicks off the reader goroutine. Must be called exactly once
|
||||
// before HandleSSE is used; calling twice is a programmer error and is
|
||||
// silently ignored.
|
||||
func (b *Bridge) Start() {
|
||||
if !b.started.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
go b.readLoop()
|
||||
}
|
||||
|
||||
// HandleSSE forwards the request body to the child as one newline-framed
|
||||
// JSON-RPC line, then streams the matching response back as a single SSE
|
||||
// event. Returns when the response is delivered, the client disconnects,
|
||||
// or the child exits.
|
||||
//
|
||||
// Errors before SSE headers go out: written as plain HTTP errors with the
|
||||
// appropriate status. Errors after headers: best-effort write of an SSE
|
||||
// `error` event, then return.
|
||||
func (b *Bridge) HandleSSE(w http.ResponseWriter, r *http.Request) {
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if len(body) == 0 {
|
||||
http.Error(w, "empty body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
idKey, ok := extractIDKey(body)
|
||||
if !ok {
|
||||
// No id → notification. Forward without waiting; reply 204.
|
||||
// (MCP notifications don't expect a response.)
|
||||
if err := b.send(body); err != nil {
|
||||
http.Error(w, "send: "+err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
|
||||
// Register a waiter before sending so a fast child can't reply before
|
||||
// we're ready to receive.
|
||||
respCh := make(chan []byte, 1)
|
||||
if _, loaded := b.pending.LoadOrStore(idKey, respCh); loaded {
|
||||
http.Error(w, "duplicate in-flight id", http.StatusConflict)
|
||||
return
|
||||
}
|
||||
defer b.pending.Delete(idKey)
|
||||
|
||||
if err := b.send(body); err != nil {
|
||||
http.Error(w, "send: "+err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
// SSE headers must go out before the first event.
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
flusher, _ := w.(http.Flusher)
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
|
||||
select {
|
||||
case resp := <-respCh:
|
||||
writeSSEEvent(w, "message", resp)
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
case <-r.Context().Done():
|
||||
// Client disconnected. Just return — child keeps running, future
|
||||
// requests on the same bridge are unaffected.
|
||||
b.log.Debug("client disconnected mid-stream", slog.String("id", idKey))
|
||||
case <-b.done:
|
||||
writeSSEEvent(w, "error", []byte(`{"reason":"child_exited"}`))
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// send writes one newline-framed message to the child. Concurrent writes
|
||||
// to the underlying pipe are not safe (interleaved bytes), so we serialize
|
||||
// here.
|
||||
func (b *Bridge) send(msg []byte) error {
|
||||
b.writeMu.Lock()
|
||||
defer b.writeMu.Unlock()
|
||||
if _, err := b.stdin.Write(msg); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(msg) == 0 || msg[len(msg)-1] != '\n' {
|
||||
if _, err := b.stdin.Write([]byte{'\n'}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// readLoop reads one JSON-RPC message per line from the child's stdout
|
||||
// and routes each to the waiting handler (if any). Lines without an id
|
||||
// (server-initiated notifications) are dropped on the floor for now —
|
||||
// phase 5 introduces the GET-stream channel for those.
|
||||
func (b *Bridge) readLoop() {
|
||||
for {
|
||||
line, err := b.stdout.ReadString('\n')
|
||||
if line != "" {
|
||||
b.routeResponse(line)
|
||||
}
|
||||
if err != nil {
|
||||
if errors.Is(err, io.EOF) {
|
||||
b.log.Debug("child stdout closed; bridge reader exiting")
|
||||
} else {
|
||||
b.log.Warn("bridge reader error", slog.String("err", err.Error()))
|
||||
}
|
||||
b.closed.Store(true)
|
||||
// Wake any pending waiters so handlers don't hang forever after
|
||||
// the child dies.
|
||||
b.pending.Range(func(k, v any) bool {
|
||||
if ch, ok := v.(chan []byte); ok {
|
||||
select {
|
||||
case <-ch:
|
||||
default:
|
||||
close(ch)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Bridge) routeResponse(line string) {
|
||||
idKey, ok := extractIDKey([]byte(line))
|
||||
if !ok {
|
||||
// Server-initiated notification or malformed message — log and skip.
|
||||
b.log.Debug("bridge: dropping un-routable line", slog.Int("len", len(line)))
|
||||
return
|
||||
}
|
||||
v, ok := b.pending.LoadAndDelete(idKey)
|
||||
if !ok {
|
||||
// No waiter — request was likely abandoned (client disconnected).
|
||||
// Logging at debug; this is normal.
|
||||
b.log.Debug("bridge: response with no waiter", slog.String("id", idKey))
|
||||
return
|
||||
}
|
||||
ch, _ := v.(chan []byte)
|
||||
if ch == nil {
|
||||
return
|
||||
}
|
||||
// Non-blocking send: the caller created the channel with buffer 1, so
|
||||
// this is guaranteed to succeed unless we're racing a Close.
|
||||
select {
|
||||
case ch <- []byte(line):
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// extractIDKey parses the JSON-RPC `id` field and returns its raw JSON
|
||||
// representation as a routing key. Returns ok=false when the message has
|
||||
// no id (notification) or fails to parse.
|
||||
//
|
||||
// JSON-RPC ids are string, number, or null. Comparing the raw JSON token
|
||||
// avoids quirks like distinguishing 1 from "1".
|
||||
func extractIDKey(data []byte) (string, bool) {
|
||||
var probe struct {
|
||||
ID json.RawMessage `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &probe); err != nil {
|
||||
return "", false
|
||||
}
|
||||
if len(probe.ID) == 0 || string(probe.ID) == "null" {
|
||||
return "", false
|
||||
}
|
||||
return string(probe.ID), true
|
||||
}
|
||||
|
||||
// writeSSEEvent writes one Server-Sent Event to w. The data may be
|
||||
// multi-line — we re-frame it per the SSE spec (each \n becomes a
|
||||
// separate `data:` line). For our typical single-line JSON, this is a
|
||||
// no-op.
|
||||
func writeSSEEvent(w io.Writer, event string, data []byte) {
|
||||
fmt.Fprintf(w, "event: %s\n", event)
|
||||
for _, line := range splitLines(data) {
|
||||
fmt.Fprintf(w, "data: %s\n", line)
|
||||
}
|
||||
_, _ = w.Write([]byte("\n"))
|
||||
}
|
||||
|
||||
func splitLines(data []byte) []string {
|
||||
// Strip a single trailing newline (typical for child output).
|
||||
if n := len(data); n > 0 && data[n-1] == '\n' {
|
||||
data = data[:n-1]
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return []string{""}
|
||||
}
|
||||
out := []string{}
|
||||
start := 0
|
||||
for i, b := range data {
|
||||
if b == '\n' {
|
||||
out = append(out, string(data[start:i]))
|
||||
start = i + 1
|
||||
}
|
||||
}
|
||||
out = append(out, string(data[start:]))
|
||||
return out
|
||||
}
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue