From 979bb6d69d948ddbf72a171d09aa47c1c66d4aaf Mon Sep 17 00:00:00 2001 From: rethinger Date: Wed, 14 Oct 2026 22:28:33 +0300 Subject: [PATCH] feat(serve,proxy): bridge the caller's prompt and tools into the client; rotate exits on the free-tier refusal serve: park each request (system, history, tool schemas) at /bridge/.json next to the session it opened, rebuild the model call from it in the client-side plugin (examples/lanvello-bridge.ts, installed in ~/.config/opencode/plugins/), block the caller's tool calls inside the client and answer with an openai tool_calls turn instead of executing them there. a tool-less api request replaces the client's tools with nothing. proxy: the free tier's 403 FreeTierError arrives brotli-encoded, so decode the peek (br/gzip/plain) before matching it; a refusal is per exit, like a 429, so park it, rotate the lane and replay the request (and rotate only on that refusal, not on every 403), wait for the fresh exit instead of answering with the refusal, and honour the far end's whole retry-after in lanes.NoteLimited (capped at 6h) instead of clamping it to 10 minutes. README: what crosses the boundary now, the client daemon's env (NO_PROXY is not optional), and the refusal in the reverse-engineering notes. --- README.md | 142 +++++++++- cmd/lanvello/main.go | 134 ++++++++- examples/README.md | 14 + examples/lanvello-bridge.ts | 150 ++++++++++ go.mod | 2 + go.sum | 4 + internal/bridge/bridge.go | 150 ++++++++++ internal/clientapi/client.go | 265 ++++++++++++++++++ internal/lanes/lanes.go | 17 +- internal/lanes/lanes_test.go | 21 ++ internal/proxy/ca.go | 166 ++++++++++++ internal/proxy/proxy.go | 498 ++++++++++++++++++++++++++++++++++ internal/proxy/proxy_test.go | 318 ++++++++++++++++++++++ internal/server/parse_test.go | 107 ++++++++ internal/server/server.go | 355 +++++++++++++++++++++++- 15 files changed, 2327 insertions(+), 16 deletions(-) create mode 100644 examples/lanvello-bridge.ts create mode 100644 internal/bridge/bridge.go create mode 100644 internal/clientapi/client.go create mode 100644 internal/proxy/ca.go create mode 100644 internal/proxy/proxy.go create mode 100644 internal/proxy/proxy_test.go create mode 100644 internal/server/parse_test.go diff --git a/README.md b/README.md index 2d0d467..2409117 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,50 @@ # lanvello -one local port serving opencode free models over an openai-compatible api. -no login, no api keys from you: lanvello replays the official client -fingerprint (`opencode/*` user-agent, `x-opencode-*` session headers, -forced streaming, bash/glob/grep/read tool quartet) against -`https://opencode.ai/zen/v1` with `Bearer public`, exactly like the cli -does for its free tier. tor lanes rotate exits when an ip is throttled. +one binary, two ways at opencode's free models, egress always through tor lanes. + +**`lanvello proxy` — the mode that works.** a local http proxy the real opencode +client is pointed at. opencode leases a free-tier session to the *running client* +and to nobody else: replaying the fingerprint (user-agent, `x-opencode-*`, tool +set) gets `403 FreeTierError`, and so does a session id copied from a live run, +even seconds later on the same machine. so the client keeps making its own calls +and the lanes move underneath it — a 429 on a spent exit parks that exit for the +reset window the far end named and replays the very same request on another one. + +**`lanvello serve` — the openai-compatible api.** it serves free models again, +but not by replaying the client's fingerprint: the free tier leases its session +to the running opencode client, so `serve` asks that client to make the call and +translates the answer back into openai shapes (streamed chunks included). + +``` +./bin/lanvello serve --listen 127.0.0.1:11434 --lanes 5 +inference: opencode client at http://127.0.0.1:49374 (the free-tier session lives there) +``` + +- `/v1/models` lists what the client itself can drive (its own list, current). +- `/v1/chat/completions` runs the caller's messages through a client session, + created per request and dropped again, so the client's session list stays + clean. `stream:false` aggregates; `stream:true` relays deltas as they arrive. +- one request per client session means requests are concurrent, and different + models can be asked for in parallel (each session is pinned to its model). +- the caller's own system prompt, history and tool definitions cross the + boundary through a spool: `serve` parks the request at + `/bridge/.json` next to the session it created, and a + plugin inside the client — `examples/lanvello-bridge.ts`, installed in + `~/.config/opencode/plugins/` — rebuilds the model call from it, so the + answer comes from the caller's prompt and the caller's tools, not the + client's. without the plugin the bridge still answers, with the client's own + agent and its own tools. +- tools are the api caller's to run. the plugin injects the caller's schemas + into the client (replacing the client's own, so a tool-less api request + cannot run anything in the client's directory), blocks the call the model + makes, records it, and `serve` returns it as an openai `tool_calls` answer + with `finish_reason: "tool_calls"` — including when the caller asked for + `stream:true`. the caller appends the result as a `role: "tool"` message and + the next request carries it back into the client. +- `--client` makes a missing client fatal; `--no-client` keeps the old replay + path, which the free tier answers with `403 FreeTierError` (see the notes at + the bottom). without a client, `serve` says so on startup instead of serving + errors. ## install @@ -31,6 +70,7 @@ go build -o bin/lanvello ./cmd/lanvello ./bin/lanvello key add --name aider # prints sk-lanv-... once + baseURL hint ./bin/lanvello lanes status ./bin/lanvello tui +cp examples/lanvello-bridge.ts ~/.config/opencode/plugins/ # serve: caller's prompt + tools ``` any harness that accepts an openai-compatible provider then uses: @@ -41,6 +81,62 @@ apiKey: sk-lanv-... model: opencode/muse-spark-1.3-contributor-free ``` +## proxy + +``` +./bin/lanvello proxy --listen 127.0.0.1:8888 --lanes 5 +./bin/lanvello proxy --print-ca # the CA path, for NODE_EXTRA_CA_CERTS +``` + +point the client at it (the proxy prints both lines on start): + +``` +HTTPS_PROXY=http://[IP_ADDRESS]:8888 \ +NODE_EXTRA_CA_CERTS=~/.local/state/lanvello/ca.pem \ +opencode +``` + +the long-running client daemon — `opencode serve --service`, the process +`lanvello serve` drives — needs the same variables where *it* is started, +plus loopback kept out of the proxy: + +``` +NO_PROXY=127.0.0.1,localhost,::1 \ +HTTPS_PROXY=http://[IP_ADDRESS]:8888 \ +NODE_EXTRA_CA_CERTS=~/.local/state/lanvello/ca.pem \ +opencode serve --service +``` + +a daemon launched without the proxy answers `403 FreeTierError` to every +session it runs, and that is the error `serve` hands back. a daemon launched +*with* the proxy but without `NO_PROXY` looks like something else again: the +client calls its own api on `127.0.0.1`, those calls go out through a lane +too, land on the exit's own loopback, and the session dies before it starts — +the proxy log shows `tls handshake opencode.ai: EOF` and the client reports +`getaddrinfo ETIMEOUT`. + +what happens on the wire: + +``` +opencode --CONNECT opencode.ai--> lanvello proxy + | tls terminated with a ca minted in the data dir + | request replayed per lane + +--> lane 2 {de} --socks--> opencode.ai 200 + +--> 429 retry-after 8760 -> lane 2 parked 146m, replay lane 3 {se} 200 +``` + +- `--print-ca` prints `ca.pem`; the client only ever sees a leaf minted for the + host it asked for, and the CA never leaves the data dir. +- 429, 502, 503 and 504 park the exit for as long as the far end's + `retry-after` says (up to 6h) and replay the same request on the next lane + *before* a byte reaches the client: a spent exit costs a retry, not a failure. + `--quiet` drops the per-request lane log. +- `socks: ["127.0.0.1:52001", ...]` in the config reuses lanes someone else + spawned (a running `lanvello serve`, say); without it the proxy spawns its own + tor, exactly like serve. +- loopback targets bypass the lanes: the client's own local server is not + the internet, and tor would refuse it anyway. + ## endpoints - `GET /v1/models` — the live free catalog, fetched upstream through a lane @@ -184,3 +280,37 @@ docker run --rm -p 127.0.0.1:11434:11434 \ see `deploy/lanvello.service` for systemd. override target with `LANVELLO_BASE_URL` if zen ever moves. + +## reverse-engineering notes (2026-10-13) + +what the official client does today, read off the running binary and the live +wire — the next person to touch the free tier needs this: + +- the provider moved: `https://models.opencode.ai/api.json` now serves + `"api": "https://opencode.ai/inference/v1"` for provider `opencode`. `/zen/v1` + still answers but it is the older path; the client uses `/inference/v1`. +- request shape: user agent `opencode///` (today + `opencode/dev/0.0.0-dev-20455/cli`), `authorization: Bearer public`, + `x-opencode-client`, `x-opencode-project`, `x-opencode-session` (the root + session id), `x-opencode-session-id`, `x-session-affinity`, `x-session-id`, + `b3` and `traceparent`. +- session ids are minted locally: `ses_` + `hex48(~(Date.now()*4096+counter))` + + 14 base62 characters from random bytes. there is no signature in them; the + timestamp decodes back to the moment the session was made (counter 1). +- the model transport is a websocket (`@opencode/SessionModelTransport`, + 55 minute rotation, 15s connect budget, plain http after 5 stream failures); + the free models are served over plain http today. +- the gate is a server-side lease on the session, not a fingerprint: bytes + replayed from another process — same headers in the same order and case, same + body, the client's own TLS stack, same machine — answer `403 FreeTierError`, + while the same bytes carrying the client's own session answer `200`. no + header, tool set, body, protocol, exit or address moves that line, and the + lease does not survive the client's request (a session id taken off the wire + is refused seconds later). this is why `proxy` exists and why `serve` cannot + come back. +- the free list moved with it: `fledge-alpha-free` answers RegionError, + `mimo-v2.5-free` and `exo-free` are deprecated (410); current free ids include + `mimo-v2.6-flash-free`, `step-5-preview-free`, `cat-jam-free`, + `muse-spark-1.3-contributor-free`, `nemotron-3-ultra-free`, + `nemotron-3.5-lightning-free`, `space-bunny-free`, `ling-3.0/3.1-flash`. + \ No newline at end of file diff --git a/cmd/lanvello/main.go b/cmd/lanvello/main.go index e7c1098..96f663c 100644 --- a/cmd/lanvello/main.go +++ b/cmd/lanvello/main.go @@ -2,16 +2,21 @@ package main import ( + "context" "flag" "fmt" + "net" "net/http" "os" "strings" "time" + "lanvello/internal/bridge" + "lanvello/internal/clientapi" "lanvello/internal/config" "lanvello/internal/keys" "lanvello/internal/lanes" + "lanvello/internal/proxy" "lanvello/internal/server" "lanvello/internal/tor" "lanvello/internal/upstream" @@ -28,6 +33,8 @@ func main() { switch os.Args[1] { case "serve": cmdServe(os.Args[2:]) + case "proxy": + cmdProxy(os.Args[2:]) case "key": cmdKey(os.Args[2:]) case "lanes": @@ -48,6 +55,7 @@ func usage() { fmt.Println(`lanvello - opencode free models over one local openai-compatible port usage: lanvello serve [--listen 127.0.0.1:11434] [--lanes 5] [--config file.json] + lanvello proxy [--listen 127.0.0.1:8888] [--lanes 5] [--print-ca] [--quiet] lanvello key add --name aider lanvello key list lanvello key revoke --name aider @@ -110,14 +118,49 @@ func healthySocks(m *lanes.Manager) int { return n } -func cmdServe(args []string) { - cfg := baseConfig(args) - ks, m := openDeps(cfg) - // tor-only: direct egress is gone. without tor lanes there is no service. +// defaultProxyListen is where the client is pointed: a local port, nothing +// outside the machine ever sees it. +const defaultProxyListen = "127.0.0.1:8888" + +// cmdProxy runs the lane-backed MITM proxy. The free tier leases a session to +// the running client only, so the client stays the one making the calls and the +// lanes move underneath it: 429 retires the exit until retry-after and the +// request is replayed on the next one. +func cmdProxy(args []string) { + fs := flag.NewFlagSet("proxy", flag.ContinueOnError) + listen := fs.String("listen", defaultProxyListen, "listen addr of the local proxy") + printCA := fs.Bool("print-ca", false, "print the CA certificate path and exit") + quiet := fs.Bool("quiet", false, "quiet: no lane log on stderr") + ln := fs.Int("lanes", 0, "lane count") + cfgPath := fs.String("config", "", "config json path") + dataDir := fs.String("data-dir", "", "state dir") + _ = fs.Parse(args) + + cfg := config.Defaults() + if *ln > 0 { + cfg.Lanes = *ln + } + if *dataDir != "" { + cfg.DataDir = *dataDir + } + var err error + cfg, err = config.LoadFile(*cfgPath, cfg) + if err != nil { + fmt.Fprintln(os.Stderr, "config error: "+err.Error()) + os.Exit(1) + } + _, m := openDeps(cfg) + ca, err := proxy.OpenCA(cfg.DataDir) + if err != nil { + fmt.Fprintln(os.Stderr, "ca: "+err.Error()) + os.Exit(1) + } + if *printCA { + fmt.Println(ca.Path()) + return + } if len(cfg.Socks) == 0 { tor.Ensure(m, cfg.DataDir, 52001, 52301, 90*time.Second) - // A 429 rotates the lane: respawn its tor in the background so the - // next request lands on a fresh exit. m.OnRotate = func(l *lanes.Lane) { tor.Respawn(m, l, cfg.DataDir, 52001, 52301, 90*time.Second) } @@ -126,6 +169,67 @@ func cmdServe(args []string) { fmt.Fprintln(os.Stderr, "no tor lanes up (tor binary missing or bootstrap failed) -- refusing to serve direct") os.Exit(1) } + lnr, err := net.Listen("tcp", *listen) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + p := proxy.New(m, ca) + if !*quiet { + p.Logf = func(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } + } + fmt.Printf("lanvello %s proxy on %s lanes=%d, egress only through lanes\n", version, *listen, len(m.Lanes())) + fmt.Printf(" HTTPS_PROXY=http://%s NODE_EXTRA_CA_CERTS=%s opencode\n", *listen, ca.Path()) + if err := p.Serve(lnr); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func cmdServe(args []string) { + cfg := baseConfig(args) + ks, m := openDeps(cfg) + // Inference has to be made by the client: the free tier leases its session + // to the running client and answers our own requests with 403 FreeTierError. + // So when a client server is up, that is the upstream. Lanes are then only + // needed for the model catalog, and the client brings its own egress. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + ca, caErr := clientapi.Discover() + useClient := caErr == nil && ca.Ready(ctx) == nil + if hasFlag(args, "--no-client") { + useClient = false + } + switch { + case useClient: + fmt.Printf("inference: opencode client at %s (the free-tier session lives there)\n", ca.BaseURL) + case hasFlag(args, "--client"): + fmt.Fprintln(os.Stderr, "no opencode client server to serve free models:", caErr) + os.Exit(1) + case hasFlag(args, "--no-client"): + fmt.Fprintln(os.Stderr, "inference: replay upstream only (--no-client)") + default: + fmt.Fprintf(os.Stderr, "inference: replay upstream only -- the free tier answers it with 403 (%v)\n", caErr) + } + + // tor-only: the replay path never dials direct. without lanes there is no + // service to serve, and no live catalog to advertise. + if !useClient { + if len(cfg.Socks) == 0 { + tor.Ensure(m, cfg.DataDir, 52001, 52301, 90*time.Second) + // A 429 rotates the lane: respawn its tor in the background so the + // next request lands on a fresh exit. + m.OnRotate = func(l *lanes.Lane) { + tor.Respawn(m, l, cfg.DataDir, 52001, 52301, 90*time.Second) + } + } + if n := healthySocks(m); n == 0 { + fmt.Fprintln(os.Stderr, "no tor lanes up (tor binary missing or bootstrap failed) -- refusing to serve direct") + os.Exit(1) + } + } else if len(cfg.Socks) == 0 { + fmt.Fprintln(os.Stderr, "note: no lanes configured -- /v1/models will use the embedded list") + } up := upstream.NewClient(m) up.BaseURL = cfg.BaseURL up.WaitBudget = time.Duration(cfg.WaitBudgetS * float64(time.Second)) @@ -138,6 +242,13 @@ func cmdServe(args []string) { } srv := server.New(ks, m, up, cfg.DataDir) srv.RequireKey = cfg.RequireKey + if useClient { + srv.ClientAPI = ca + // The plugin reads a spool directory, so a moved data dir moves both + // ends: no env var to forget. + srv.BridgeDir = bridge.Dir(cfg.DataDir) + bridge.Sweep(srv.BridgeDir, time.Hour) + } fmt.Printf("lanvello %s listening on http://%s lanes=%d tor-only, free, no login\n", version, cfg.Listen, cfg.Lanes) if err := http.ListenAndServe(cfg.Listen, srv.Handler()); err != nil { fmt.Fprintln(os.Stderr, err) @@ -145,6 +256,17 @@ func cmdServe(args []string) { } } +// hasFlag reports whether a bare switch (--client, --no-client) is present: +// they carry no value, so they are read straight off argv. +func hasFlag(args []string, name string) bool { + for _, a := range args { + if a == name || strings.HasPrefix(a, name+"=") { + return true + } + } + return false +} + // baseFlags lifts the flags baseConfig understands (--config, // --data-dir, --listen, --lanes) out of a subcommand's argument // list, dropping flags that belong to the subcommand itself diff --git a/examples/README.md b/examples/README.md index 4d730bb..06808e7 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,5 +1,19 @@ # examples +## lanvello-bridge.ts + +the client-side half of the `serve` bridge: copy it into +`~/.config/opencode/plugins/` and the caller's own system prompt, history and +tool schemas travel into the client's session, from the spool `serve` writes +under `/bridge/`. the caller's tools are injected into the client's +tool set — replacing the client's own, so a tool-less api request runs nothing +in the client's directory — and a call the model makes is recorded for the +gateway and blocked inside the client instead of being executed there: the api +caller gets it back as an openai `tool_calls` turn and runs it itself. + +the spool directory is `$LANVELLO_BRIDGE_DIR`, defaulting to the gateway's own +`/bridge`, so a moved data dir moves both ends. + ## lanvello-ext.mjs pi-family provider extension (omp, prime-agent) pointing at a local diff --git a/examples/lanvello-bridge.ts b/examples/lanvello-bridge.ts new file mode 100644 index 0000000..149f9a0 --- /dev/null +++ b/examples/lanvello-bridge.ts @@ -0,0 +1,150 @@ +// lanvello-bridge — hand an api caller's request to this client. +// +// The free tier leases a session to the running client and to nobody else, so +// lanvello's api asks this client to make the call. The caller's own system +// prompt, history and tool schemas arrive in a spool file the gateway writes +// next to the session it opened; this plugin rebuilds the model request from +// it, and refuses to run the caller's tools — their results come back from the +// api caller, not from inside the client. +// +// Install by copying this file into ~/.config/opencode/plugins/ (or +// .opencode/plugins/ inside a project). The spool directory is +// $LANVELLO_BRIDGE_DIR, defaulting to /lanvello/bridge, which is where +// `lanvello serve` parks requests by default. Without this plugin the bridge +// still answers, but with the client's own agent prompt and its own tools. + +const fs = require("fs"); +const path = require("path"); + +const DIR = + process.env.LANVELLO_BRIDGE_DIR || + path.join( + process.env.XDG_STATE_HOME || path.join(process.env.HOME || ".", ".local", "state"), + "lanvello", + "bridge", + ); + +type ParkedMessage = { + role: string; + content?: string; + tool_calls?: Array<{ id?: string; name: string; arguments?: unknown }>; + tool_call_id?: string; +}; + +type ParkedTool = { name: string; description?: string; parameters?: unknown }; + +type Parked = { + sessionId?: string; + system?: string; + messages?: ParkedMessage[]; + tools?: ParkedTool[]; + model?: string; +}; + +type ContextHook = { + sessionID?: string; + system?: unknown[]; + messages?: unknown[]; + tools?: Record; +}; + +type ExecPayload = { id?: string; sessionID: string; tool: string; input?: unknown }; + +type PluginCtx = { + session?: { hook?: (name: string, fn: (input: ContextHook) => void) => unknown }; + tool?: { hook?: (name: string, fn: (payload: unknown) => void) => unknown }; +}; + +function parked(sessionID: string | undefined): Parked | null { + if (!sessionID) return null; + try { + const raw: unknown = JSON.parse(fs.readFileSync(path.join(DIR, sessionID + ".json"), "utf8")); + if (!raw || typeof raw !== "object") return null; + return raw as Parked; + } catch (e) { + return null; + } +} + +// asExec narrows the untrusted hook payload down to the call it describes. +function asExec(payload: unknown): ExecPayload | null { + if (!payload || typeof payload !== "object") return null; + const p = payload as Record; + if (typeof p.sessionID !== "string" || typeof p.tool !== "string") return null; + return { + id: typeof p.id === "string" ? p.id : undefined, + sessionID: p.sessionID, + tool: p.tool, + input: p.input, + }; +} + +// buildMessage turns one caller turn into the client's message shape: text for +// text, and the tool traffic spelled out, so the model still sees what it asked +// for and what came back. +function buildMessage(m: ParkedMessage): { role: string; content: unknown[]; metadata: object } | null { + const parts: Array<{ type: string; text: string }> = []; + if (m.content && m.role === "tool") { + parts.push({ type: "text", text: `[tool result${m.tool_call_id ? " for " + m.tool_call_id : ""}] ${m.content}` }); + } else if (m.content) { + parts.push({ type: "text", text: m.content }); + } + for (const c of m.tool_calls || []) { + parts.push({ type: "text", text: `[tool call ${c.name}(${JSON.stringify(c.arguments ?? {})})]` }); + } + if (parts.length === 0) return null; + return { role: m.role === "tool" ? "user" : m.role, content: parts, metadata: {} }; +} + +export default { + id: "lanvello-bridge", + setup: async (ctx: PluginCtx) => { + if (ctx.session?.hook) { + await ctx.session.hook("context", (input: ContextHook) => { + const req = parked(input.sessionID); + if (!req) return; + + // the caller's system prompt replaces the client's own agent prompt + input.system = req.system ? [{ type: "text", text: req.system }] : []; + + if (Array.isArray(req.messages)) { + const built = req.messages.map(buildMessage).filter((m) => m !== null); + if (built.length > 0) input.messages = built; + } + + // and its tools replace the client's — including replacing them with + // nothing, so a tool-less api request cannot run bash in this directory + const tools: Record = {}; + for (const t of req.tools || []) { + if (!t || !t.name) continue; + tools[t.name] = { + description: t.description || "", + input: t.parameters || { type: "object", properties: {} }, + }; + } + input.tools = tools; + }); + } + + if (ctx.tool?.hook) { + await ctx.tool.hook("execute.before", (payload: unknown) => { + const call = asExec(payload); + if (!call) return; + const req = parked(call.sessionID); + if (!req) return; + if (!(req.tools || []).some((t) => t && t.name === call.tool)) return; + // this call is the api caller's to run: record it for the gateway and + // stop the client from executing it here + try { + fs.appendFileSync( + path.join(DIR, call.sessionID + ".calls.jsonl"), + JSON.stringify({ id: call.id, name: call.tool, arguments: call.input ?? {} }) + "\n", + ); + } catch (e) {} + throw new Error( + `lanvello-bridge: ${call.tool} belongs to the api caller; its result arrives in the next request`, + ); + }); + } + }, +}; diff --git a/go.mod b/go.mod index eae8b9e..9da142e 100644 --- a/go.mod +++ b/go.mod @@ -3,3 +3,5 @@ module lanvello go 1.24 require golang.org/x/net v0.33.0 + +require github.com/andybalholm/brotli v1.2.6 diff --git a/go.sum b/go.sum index 16660ab..499b718 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,6 @@ +github.com/andybalholm/brotli v1.2.6 h1:ftYnfj6usCp+UGV5kSJ3+chpMQgU+gJf/AxsUQ52REI= +github.com/andybalholm/brotli v1.2.6/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= diff --git a/internal/bridge/bridge.go b/internal/bridge/bridge.go new file mode 100644 index 0000000..263f9ac --- /dev/null +++ b/internal/bridge/bridge.go @@ -0,0 +1,150 @@ +// Package bridge carries one api request across the client boundary. +// +// The gateway cannot hand a caller's system prompt, history and tool schemas to +// the free tier itself: only the running client gets a session, and its own http +// api takes nothing but text. So the request is parked in a spool directory the +// client-side plugin reads, keyed by the session the gateway just opened: +// +// /.json what the caller asked for (system, messages, tools) +// /.calls.jsonl what the model called, written by the plugin +// +// The plugin rewrites the model request from the .json (its `session.hook +// ("context")` hands it system, messages and tools), blocks execution of the +// caller's tools and appends each call to the .jsonl instead. The gateway reads +// that back and answers the api caller with an openai tool_calls turn. +package bridge + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// Message is one turn of the caller's conversation, flattened enough for the +// plugin to rebuild it: text, plus the tool call/result that travelled with it. +type Message struct { + Role string `json:"role"` + Content string `json:"content,omitempty"` + ToolCalls []ToolCall `json:"tool_calls,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` +} + +// ToolCall is an openai tool call, either from the caller's history or from the +// model (the plugin fills id/name/arguments). +type ToolCall struct { + ID string `json:"id,omitempty"` + Name string `json:"name"` + Arguments json.RawMessage `json:"arguments,omitempty"` + Input json.RawMessage `json:"input,omitempty"` +} + +// Tool is one tool the caller declared, in the shape the plugin expects +// (the client's own tool entries are {description, input}). +type Tool struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + Parameters json.RawMessage `json:"parameters,omitempty"` +} + +// Request is what the plugin reads. +type Request struct { + SessionID string `json:"sessionId"` + System string `json:"system,omitempty"` + Messages []Message `json:"messages"` + Tools []Tool `json:"tools,omitempty"` + Model string `json:"model"` + Created time.Time `json:"created"` +} + +// Dir is where the spool lives. The plugin reads the same path (it honours the +// same env var), so a moved data dir moves both ends. +func Dir(dataDir string) string { + if v := os.Getenv("LANVELLO_BRIDGE_DIR"); v != "" { + return v + } + return filepath.Join(dataDir, "bridge") +} + +func requestPath(dir, sessionID string) string { return filepath.Join(dir, sessionID+".json") } +func callsPath(dir, sessionID string) string { return filepath.Join(dir, sessionID+".calls.jsonl") } + +// Write parks one request for the plugin. +func Write(dir string, r Request) error { + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + r.Created = time.Now() + b, err := json.Marshal(r) + if err != nil { + return err + } + tmp := requestPath(dir, r.SessionID) + ".tmp" + if err := os.WriteFile(tmp, b, 0o600); err != nil { + return err + } + return os.Rename(tmp, requestPath(dir, r.SessionID)) +} + +// Remove drops a finished request together with whatever calls it produced: a +// later request reuses the session id space and must not inherit old calls. +func Remove(dir, sessionID string) { + _ = os.Remove(requestPath(dir, sessionID)) + _ = os.Remove(callsPath(dir, sessionID)) +} + +// Calls reads the tool calls the plugin recorded so far. +func Calls(dir, sessionID string) []ToolCall { + f, err := os.Open(callsPath(dir, sessionID)) + if err != nil { + return nil + } + defer f.Close() + var out []ToolCall + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 4<<20) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var c ToolCall + if err := json.Unmarshal([]byte(line), &c); err != nil { + continue + } + if c.Name == "" { + continue + } + if len(c.Arguments) == 0 { + c.Arguments = c.Input + } + if len(c.Arguments) == 0 { + c.Arguments = json.RawMessage("{}") + } + if c.ID == "" { + c.ID = fmt.Sprintf("call_%d_%s", len(out), c.Name) + } + out = append(out, c) + } + return out +} + +// Sweep removes spool entries older than ttl: a crashed gateway must not leave +// the plugin rewriting requests that will never be read again. +func Sweep(dir string, ttl time.Duration) { + ents, err := os.ReadDir(dir) + if err != nil { + return + } + cut := time.Now().Add(-ttl) + for _, e := range ents { + info, err := e.Info() + if err != nil || info.ModTime().After(cut) { + continue + } + _ = os.Remove(filepath.Join(dir, e.Name())) + } +} diff --git a/internal/clientapi/client.go b/internal/clientapi/client.go new file mode 100644 index 0000000..24e1293 --- /dev/null +++ b/internal/clientapi/client.go @@ -0,0 +1,265 @@ +// Package clientapi drives the running opencode client as the model upstream. +// +// It exists because the free tier leases a session to the client process only: +// a replayed request (our own headers, our own session id) is answered with 403 +// FreeTierError, while the same bytes carrying a session the client just used +// are answered normally. So the client makes the call, and this package is the +// thin, honest contract with it: +// +// POST /api/session create a session pinned to a model +// POST /api/session/{id}/prompt send the text +// GET /api/session/{id}/message read the assistant text (and the idle part) +// DELETE /api/session/{id} drop it again +// +// Auth is HTTP Basic with the username "opencode" and the password the client +// writes into its own service.json. +package clientapi + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "time" +) + +// Client talks to one opencode client server. +type Client struct { + BaseURL string + Password string + HTTP *http.Client +} + +// Usage is what the client reports for a finished session. +type Usage struct { + Input int `json:"input"` + Output int `json:"output"` + Reasoning int `json:"reasoning"` + CacheRead int `json:"cache_read"` + CacheWrite int `json:"cache_write"` +} + +type serviceFile struct { + URL string `json:"url"` + Password string `json:"password"` + PID int `json:"pid"` + Version string `json:"version"` +} + +// ServicePath is where the client records the server it started. +func ServicePath() string { + if v := os.Getenv("OPENCODE_SERVICE_FILE"); v != "" { + return v + } + state := os.Getenv("XDG_STATE_HOME") + if state == "" { + home, _ := os.UserHomeDir() + state = filepath.Join(home, ".local", "state") + } + return filepath.Join(state, "opencode", "service.json") +} + +// Discover returns a client for the running opencode server, or an error when +// there is none: without the client there is no free tier, and pretending +// otherwise is how a gateway ends up serving 403s. +func Discover() (*Client, error) { + path := ServicePath() + b, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("no opencode client server (%s): %w", path, err) + } + var sf serviceFile + if err := json.Unmarshal(b, &sf); err != nil { + return nil, fmt.Errorf("bad %s: %w", path, err) + } + if sf.URL == "" || sf.Password == "" { + return nil, fmt.Errorf("%s has no url/password", path) + } + return &Client{ + BaseURL: strings.TrimRight(sf.URL, "/"), + Password: sf.Password, + HTTP: &http.Client{Timeout: 30 * time.Second}, + }, nil +} + +func (c *Client) do(ctx context.Context, method, path string, body any, out any) error { + var rdr io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + return err + } + rdr = bytes.NewReader(b) + } + req, err := http.NewRequestWithContext(ctx, method, c.BaseURL+path, rdr) + if err != nil { + return err + } + token := base64.StdEncoding.EncodeToString([]byte("opencode:" + c.Password)) + req.Header.Set("Authorization", "Basic "+token) + req.Header.Set("Content-Type", "application/json") + resp, err := c.HTTP.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return err + } + if resp.StatusCode != 200 { + return fmt.Errorf("client %s %s: HTTP %d %s", method, path, resp.StatusCode, strings.TrimSpace(string(raw[:min(len(raw), 200)]))) + } + if out == nil { + return nil + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("client %s %s: %w", method, path, err) + } + return nil +} + +// Ready reports whether the client is answering, which is what decides if free +// models can be served at all. +func (c *Client) Ready(ctx context.Context) error { + var v map[string]any + return c.do(ctx, "GET", "/api/info", nil, &v) +} + +// Model is one entry of the client's own model list. +type Model struct { + ID string `json:"id"` + ProviderID string `json:"providerID"` + Name string `json:"name"` +} + +// Models returns what the client can drive right now. This is the honest +// catalog for a client-backed gateway: the client's list is the one its own +// requests are built from, retired ids included. +func (c *Client) Models(ctx context.Context) ([]Model, error) { + var out struct { + Data []Model `json:"data"` + } + if err := c.do(ctx, "GET", "/api/model", nil, &out); err != nil { + return nil, err + } + return out.Data, nil +} + +// Create opens a session pinned to a model. +func (c *Client) Create(ctx context.Context, provider, model, variant string) (string, error) { + ref := map[string]any{"providerID": provider, "id": model} + if variant != "" && variant != "default" { + ref["variant"] = variant + } + var out struct { + Data struct { + ID string `json:"id"` + } `json:"data"` + } + if err := c.do(ctx, "POST", "/api/session", map[string]any{"model": ref}, &out); err != nil { + return "", err + } + if out.Data.ID == "" { + return "", fmt.Errorf("client returned no session id") + } + return out.Data.ID, nil +} + +// Drop deletes a session: a gateway call should not litter the client's list. +func (c *Client) Drop(ctx context.Context, sessionID string) { + _ = c.do(ctx, "DELETE", "/api/session/"+sessionID, nil, nil) +} + +// Prompt sends the text that will be answered. +func (c *Client) Prompt(ctx context.Context, sessionID, text string) error { + return c.do(ctx, "POST", "/api/session/"+sessionID+"/prompt", map[string]any{"text": text}, nil) +} + +type part struct { + ID string `json:"id"` + Type string `json:"type"` + Outcome string `json:"outcome"` + Text string `json:"text"` + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"content"` +} + +type messagePage struct { + Data []part `json:"data"` +} + +func (p part) answer() string { + var b strings.Builder + for _, c := range p.Content { + if c.Type == "text" { + b.WriteString(c.Text) + } + } + return b.String() +} + +// Snapshot is the state of one session's answer right now: the text so far, +// whether the turn is over, and — once it is — what it cost. +type Snapshot struct { + Text string + Idle bool + Outcome string + Usage Usage +} + +// Snapshot reads a session's answer. The caller decides when to stop asking +// (its own deadline, an idle part, or a tool call it wants to answer itself). +func (c *Client) Snapshot(ctx context.Context, sessionID string) (Snapshot, error) { + var page messagePage + if err := c.do(ctx, "GET", "/api/session/"+sessionID+"/message", nil, &page); err != nil { + return Snapshot{}, err + } + var out Snapshot + for _, p := range page.Data { + switch p.Type { + case "assistant": + out.Text = p.answer() + case "idle": + out.Idle = true + out.Outcome = p.Outcome + } + } + if out.Idle { + c.fillUsage(ctx, sessionID, &out) + } + return out, nil +} + +func (c *Client) fillUsage(ctx context.Context, sessionID string, snap *Snapshot) { + var out struct { + Data struct { + Tokens struct { + Input int `json:"input"` + Output int `json:"output"` + Reasoning int `json:"reasoning"` + Cache struct { + Read int `json:"read"` + Write int `json:"write"` + } `json:"cache"` + } `json:"tokens"` + Outcome string `json:"outcome"` + } `json:"data"` + } + if err := c.do(ctx, "GET", "/api/session/"+sessionID, nil, &out); err != nil { + return + } + t := out.Data.Tokens + snap.Usage = Usage{Input: t.Input, Output: t.Output, Reasoning: t.Reasoning, CacheRead: t.Cache.Read, CacheWrite: t.Cache.Write} + if out.Data.Outcome != "" { + snap.Outcome = out.Data.Outcome + } +} diff --git a/internal/lanes/lanes.go b/internal/lanes/lanes.go index 601b67f..35ca3d3 100644 --- a/internal/lanes/lanes.go +++ b/internal/lanes/lanes.go @@ -16,6 +16,11 @@ import ( "time" ) +// maxRetire caps how long a spent exit is parked: long enough to respect the +// far end's own reset window (they are often measured in hours), short enough +// that a lane cannot be parked forever by a bogus header. +const maxRetire = 6 * time.Hour + type Lane struct { Index int Country string @@ -261,9 +266,15 @@ func (m *Manager) NoteResult(country string, status int) { func (m *Manager) NoteLimited(l *Lane, retryAfter time.Duration) time.Time { m.mu.Lock() defer m.mu.Unlock() - span := 10 * time.Minute - if retryAfter > 0 && retryAfter < span { - span = retryAfter + // The far end knows its own window and it is often hours wide: retiring + // the exit for less than it asked hands the same spent address straight + // back to the next request, which is how a free tier turns into a wall. + span := retryAfter + if span <= 0 { + span = 10 * time.Minute + } + if span > maxRetire { + span = maxRetire } l.LimitedTill = time.Now().Add(span) l.Score-- diff --git a/internal/lanes/lanes_test.go b/internal/lanes/lanes_test.go index 6e313ba..d6b2669 100644 --- a/internal/lanes/lanes_test.go +++ b/internal/lanes/lanes_test.go @@ -90,3 +90,24 @@ func TestRotateHookAndDone(t *testing.T) { t.Fatalf("RotateDone not applied: %+v", ls[0]) } } + +// A 429 carries the far end's own reset window and it is often hours wide: +// retiring an exit for less than that hands the same spent address straight +// back to the next request. +func TestNoteLimitedHonorsRetryAfter(t *testing.T) { + m := lanes.New(t.TempDir(), []string{"de"}, nil, nil, true, 1) + l := m.Lanes()[0] + + until := m.NoteLimited(l, 146*time.Minute) + if d := time.Until(until); d < 145*time.Minute || d > 147*time.Minute { + t.Fatalf("146m retry-after retired the exit for %s", d) + } + until = m.NoteLimited(l, 0) + if d := time.Until(until); d < 9*time.Minute || d > 11*time.Minute { + t.Fatalf("missing retry-after fell back to %s, want about 10m", d) + } + until = m.NoteLimited(l, 400*time.Hour) + if d := time.Until(until); d > 6*time.Hour+time.Minute { + t.Fatalf("a bogus retry-after parked the exit for %s, want at most 6h", d) + } +} diff --git a/internal/proxy/ca.go b/internal/proxy/ca.go new file mode 100644 index 0000000..db9fa62 --- /dev/null +++ b/internal/proxy/ca.go @@ -0,0 +1,166 @@ +// Package proxy serves a local HTTP(S) proxy whose egress is the lane pool. +// It exists because the free tier now leases a session to the running opencode +// client only: no third party can mint one, so the client itself has to make +// the calls, and the lanes have to sit under it instead of behind a gateway. +package proxy + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "fmt" + "math/big" + "net" + "os" + "path/filepath" + "sync" + "time" +) + +// CA mints the throwaway certificate authority that opencode is told to trust +// (NODE_EXTRA_CA_CERTS). It never leaves the data dir. +type CA struct { + path string + cert *x509.Certificate + key *ecdsa.PrivateKey + raw tls.Certificate + + mu sync.Mutex + cache map[string]*tls.Certificate +} + +// OpenCA loads the CA from dir, creating it on first use. The PEM file it +// returns is what NODE_EXTRA_CA_CERTS must point at. +func OpenCA(dir string) (*CA, error) { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + pemPath := filepath.Join(dir, "ca.pem") + keyPath := filepath.Join(dir, "ca-key.pem") + c := &CA{path: pemPath, cache: map[string]*tls.Certificate{}} + + certPEM, cerr := os.ReadFile(pemPath) + keyPEM, kerr := os.ReadFile(keyPath) + if cerr == nil && kerr == nil { + cert, key, err := parseCA(certPEM, keyPEM) + if err == nil { + c.cert, c.key = cert, key + c.raw = tls.Certificate{Certificate: [][]byte{cert.Raw}, PrivateKey: key} + return c, nil + } + } + cert, key, err := newCA() + if err != nil { + return nil, err + } + c.cert, c.key = cert, key + c.raw = tls.Certificate{Certificate: [][]byte{cert.Raw}, PrivateKey: key} + der, err := x509.MarshalECPrivateKey(key) + if err != nil { + return nil, err + } + if err := os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: der}), 0o600); err != nil { + return nil, err + } + if err := os.WriteFile(pemPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert.Raw}), 0o644); err != nil { + return nil, err + } + return c, nil +} + +// Path is the PEM file to hand to the client as NODE_EXTRA_CA_CERTS. +func (c *CA) Path() string { return c.path } + +func newCA() (*x509.Certificate, *ecdsa.PrivateKey, error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, err + } + serial, err := randSerial() + if err != nil { + return nil, nil, err + } + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: "lanvello local CA", Organization: []string{"lanvello"}}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().AddDate(5, 0, 0), + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + IsCA: true, + MaxPathLenZero: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + return nil, nil, err + } + cert, err := x509.ParseCertificate(der) + if err != nil { + return nil, nil, err + } + return cert, key, nil +} + +func parseCA(certPEM, keyPEM []byte) (*x509.Certificate, *ecdsa.PrivateKey, error) { + cb, _ := pem.Decode(certPEM) + kb, _ := pem.Decode(keyPEM) + if cb == nil || kb == nil { + return nil, nil, fmt.Errorf("ca: bad pem") + } + cert, err := x509.ParseCertificate(cb.Bytes) + if err != nil { + return nil, nil, err + } + key, err := x509.ParseECPrivateKey(kb.Bytes) + if err != nil { + return nil, nil, err + } + return cert, key, nil +} + +func randSerial() (*big.Int, error) { + limit := new(big.Int).Lsh(big.NewInt(1), 128) + return rand.Int(rand.Reader, limit) +} + +// Cert returns a certificate for host, minted on demand and cached for the +// process lifetime: one CA, one leaf per host the client is talking to. +func (c *CA) Cert(host string) (tls.Certificate, error) { + c.mu.Lock() + defer c.mu.Unlock() + if got, ok := c.cache[host]; ok { + return *got, nil + } + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return tls.Certificate{}, err + } + serial, err := randSerial() + if err != nil { + return tls.Certificate{}, err + } + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: host}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().AddDate(2, 0, 0), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + if ip := net.ParseIP(host); ip != nil { + tmpl.IPAddresses = []net.IP{ip} + } else { + tmpl.DNSNames = []string{host} + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, c.cert, &key.PublicKey, c.key) + if err != nil { + return tls.Certificate{}, err + } + out := tls.Certificate{Certificate: [][]byte{der, c.cert.Raw}, PrivateKey: key} + c.cache[host] = &out + return out, nil +} diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go new file mode 100644 index 0000000..1e30c64 --- /dev/null +++ b/internal/proxy/proxy.go @@ -0,0 +1,498 @@ +package proxy + +import ( + "bufio" + "bytes" + "compress/gzip" + "context" + "crypto/tls" + "fmt" + "io" + "net" + "net/http" + "strconv" + "strings" + "time" + + "github.com/andybalholm/brotli" + "golang.org/x/net/proxy" + + "lanvello/internal/lanes" +) + +// Proxy is a local HTTP(S) proxy whose every byte leaves through a lane. +// +// It terminates the client's TLS with its own CA, so a 429 from the free tier +// is visible as a status instead of an opaque stream: the exit is retired until +// retry-after and the very same request is replayed on the next lane. That is +// the whole trick — the client keeps its session lease, the throttle moves. +type Proxy struct { + Lanes *lanes.Manager + CA *CA + // Attempts bounds the lane rotation for one client request. + Attempts int + // Logf reports lane decisions (429 and free-tier rotations among them). + Logf func(string, ...any) + + // refusalPark is how long an exit the free tier refuses stays out of + // rotation. The far end names no window for that refusal, so pick one. + refusalPark time.Duration + // laneWait bounds how long a request waits for a rotating lane to come back + // with a fresh exit before it settles for the answer it already has. + laneWait time.Duration + + // dial bounds the connect+TLS+headers phase; the body is not on a clock + // because a thinking model is not a broken socket. + dial time.Duration +} + +func New(m *lanes.Manager, ca *CA) *Proxy { + return &Proxy{ + Lanes: m, CA: ca, Attempts: 4, + dial: 3 * time.Minute, refusalPark: 15 * time.Minute, laneWait: 30 * time.Second, + } +} + +const maxBody = 64 << 20 + +func (p *Proxy) logf(format string, a ...any) { + if p.Logf != nil { + p.Logf(format, a...) + } +} + +// Serve accepts client connections until the listener dies. +func (p *Proxy) Serve(l net.Listener) error { + for { + c, err := l.Accept() + if err != nil { + return err + } + go p.conn(c) + } +} + +func (p *Proxy) conn(c net.Conn) { + defer c.Close() + br := bufio.NewReader(c) + for { + req, err := http.ReadRequest(br) + if err != nil { + return + } + if req.Method == http.MethodConnect { + p.connect(c, req) + return + } + if !p.relay(c, req, "http", req.Host) { + return + } + if req.Close { + return + } + } +} + +// connect upgrades the client to TLS with a minted certificate for the target +// host and then serves whatever rides on that tunnel — normally opencode's +// whole session, one request after another. +func (p *Proxy) connect(client net.Conn, req *http.Request) { + host, port := splitHostPort(req.Host, "443") + if _, err := io.WriteString(client, "HTTP/1.1 200 Connection established\r\n\r\n"); err != nil { + return + } + cert, err := p.CA.Cert(host) + if err != nil { + p.logf("cert %s: %v", host, err) + return + } + tc := tls.Server(client, &tls.Config{Certificates: []tls.Certificate{cert}}) + if err := tc.Handshake(); err != nil { + p.logf("tls handshake %s: %v", host, err) + return + } + defer tc.Close() + tbr := bufio.NewReader(tc) + for { + creq, err := http.ReadRequest(tbr) + if err != nil { + return + } + if !p.relay(tc, creq, "https", net.JoinHostPort(host, port)) { + return + } + if creq.Close { + return + } + } +} + +// relay performs one client request against the lane pool and writes the +// answer back. It returns false when the client connection is done for. +func (p *Proxy) relay(w io.Writer, req *http.Request, scheme, addr string) bool { + host, port := splitHostPort(addr, defaultPort(scheme)) + body, err := readBody(req) + if err != nil { + writeStatus(w, http.StatusRequestEntityTooLarge, "request body too large") + return false + } + resp, lane, err := p.roundtrip(scheme, host, port, req, body) + if err != nil { + p.logf("%s %s%s: %v", req.Method, host, req.URL.Path, err) + writeStatus(w, http.StatusBadGateway, err.Error()) + return false + } + defer resp.Body.Close() + if err := p.writeResponse(w, resp); err != nil { + return false + } + p.logf("%d %s %s%s lane %s{%s}", resp.StatusCode, req.Method, host, req.URL.Path, laneTag(lane), laneCountry(lane)) + return true +} + +// roundtrip sends the request on a lane and rotates on a throttled exit. The +// client sees one answer: retries happen before a single byte goes downstream. +func (p *Proxy) roundtrip(scheme, host, port string, req *http.Request, body []byte) (*http.Response, *lanes.Lane, error) { + tried := map[int]bool{} + var last *http.Response + var lastErr error + deadline := time.Now().Add(p.laneWait) + for attempts := 0; attempts < p.Attempts; { + l := p.Lanes.Pick(tried) + if l == nil { + // Every lane is rotating or parked — and a rotation is exactly what + // this request is waiting for, since the fresh exit is the answer. + // Wait for one, then give the new exits another look. + if time.Now().After(deadline) { + break + } + time.Sleep(200 * time.Millisecond) + tried = map[int]bool{} + continue + } + attempts++ + tried[l.Index] = true + started := time.Now() + resp, err := p.attempt(l, scheme, host, port, req, body) + if err != nil { + lastErr = err + p.Lanes.NoteResult(l.Country, 502) + p.Lanes.Release(l) + continue + } + p.Lanes.Emit(lanes.Proof{T: time.Now(), Lane: l.Index, Country: l.Country, IP: l.ExitIP, + Model: host + req.URL.Path, Status: resp.StatusCode, Bytes: resp.ContentLength, Secs: time.Since(started).Seconds()}) + if freeTierRefused(resp) { + // The far end has had enough of this exit: the same answer as a + // 429, and the same remedy — park it, replay elsewhere, and let the + // next request start on an exit that still passes. + p.Lanes.NoteLimited(l, p.refusalPark) + country := p.Lanes.Rotate(l) + p.logf("%s %s%s: lane %d{%s} refused the free tier -- replaying on %s", + req.Method, host, req.URL.Path, l.Index, l.Country, country) + drain(resp) + p.Lanes.Release(l) + last = &http.Response{StatusCode: resp.StatusCode, Header: resp.Header.Clone(), Body: http.NoBody, Request: req} + lastErr = fmt.Errorf("lane %d{%s} refused the free tier", l.Index, l.Country) + continue + } + if !throttled(resp.StatusCode) { + p.Lanes.NoteResult(l.Country, resp.StatusCode) + p.Lanes.Release(l) + return resp, l, nil + } + // The exit is spent: park it for as long as the far end said, ask for a + // fresh one in the best country and put this request elsewhere. + d := retryAfter(resp) + p.Lanes.NoteLimited(l, d) + country := p.Lanes.Rotate(l) + p.logf("%s %s%s: lane %d{%s} %d exit limited %s -- replaying on %s", + req.Method, host, req.URL.Path, l.Index, l.Country, resp.StatusCode, d, country) + drain(resp) + p.Lanes.Release(l) + last = &http.Response{StatusCode: resp.StatusCode, Header: resp.Header.Clone(), Body: http.NoBody, Request: req} + lastErr = fmt.Errorf("lane %d{%s} answered %d", l.Index, l.Country, resp.StatusCode) + } + if last != nil { + return last, nil, nil + } + if lastErr == nil { + lastErr = fmt.Errorf("no lane answered") + } + return nil, nil, lastErr +} + +// attempt dials one lane, speaks TLS to the far end and returns the response +// headers with the body still streaming. +func (p *Proxy) attempt(l *lanes.Lane, scheme, host, port string, req *http.Request, body []byte) (*http.Response, error) { + conn, err := p.dialLane(l, host, port) + if err != nil { + return nil, err + } + deadline := time.Now().Add(p.dial) + _ = conn.SetDeadline(deadline) + + var rw io.ReadWriter = conn + if scheme == "https" { + tc := tls.Client(conn, &tls.Config{ServerName: host}) + if err := tc.Handshake(); err != nil { + conn.Close() + return nil, err + } + rw = tc + } + + out := req.Clone(context.Background()) + out.URL.Scheme, out.URL.Host = scheme, net.JoinHostPort(host, port) + out.RequestURI = "" + out.Body = io.NopCloser(bytes.NewReader(body)) + out.ContentLength = int64(len(body)) + out.Close = true + if req.Host != "" { + out.Host = req.Host + } + cleanHeaders(out.Header) + if err := out.Write(rw); err != nil { + conn.Close() + return nil, err + } + br := bufio.NewReader(rw) + resp, err := http.ReadResponse(br, out) + if err != nil { + conn.Close() + return nil, err + } + // Streaming starts now: a model that thinks for minutes must not be cut. + _ = conn.SetDeadline(time.Time{}) + resp.Body = &laneBody{ReadCloser: resp.Body, conn: conn} + return resp, nil +} + +func (p *Proxy) dialLane(l *lanes.Lane, host, port string) (net.Conn, error) { + addr := net.JoinHostPort(host, port) + if l == nil || l.SocksAddr == "" || isLoopback(host) { + // The client's own local server is not lane material. + return net.DialTimeout("tcp", addr, 30*time.Second) + } + d, err := proxy.SOCKS5("tcp", l.SocksAddr, nil, proxy.Direct) + if err != nil { + return nil, err + } + if cd, ok := d.(proxy.ContextDialer); ok { + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + return cd.DialContext(ctx, "tcp", addr) + } + return d.Dial("tcp", addr) +} + +// writeResponse frames the answer for the client without giving up keep-alive: +// the pool may close the upstream connection between requests, the client must +// not have to reconnect its session for that. +func (p *Proxy) writeResponse(w io.Writer, resp *http.Response) error { + h := resp.Header.Clone() + h.Del("Connection") + h.Del("Keep-Alive") + h.Del("Proxy-Connection") + h.Del("Transfer-Encoding") + h.Del("Upgrade") + h.Del("Content-Length") + + var head bytes.Buffer + fmt.Fprintf(&head, "HTTP/1.1 %d %s\r\n", resp.StatusCode, statusText(resp.StatusCode)) + if resp.ContentLength >= 0 { + h.Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10)) + } else { + h.Set("Transfer-Encoding", "chunked") + } + for k, vs := range h { + for _, v := range vs { + fmt.Fprintf(&head, "%s: %s\r\n", k, v) + } + } + head.WriteString("\r\n") + if _, err := w.Write(head.Bytes()); err != nil { + return err + } + if resp.ContentLength >= 0 { + _, err := io.Copy(w, resp.Body) + return err + } + buf := make([]byte, 32*1024) + for { + n, rerr := resp.Body.Read(buf) + if n > 0 { + if _, werr := fmt.Fprintf(w, "%x\r\n", n); werr != nil { + return werr + } + if _, werr := w.Write(buf[:n]); werr != nil { + return werr + } + if _, werr := io.WriteString(w, "\r\n"); werr != nil { + return werr + } + } + if rerr != nil { + if rerr == io.EOF { + _, werr := io.WriteString(w, "0\r\n\r\n") + return werr + } + return rerr + } + } +} + +// laneBody closes the upstream connection once the answer has been read, which +// keeps one request per lane connection without leaking sockets. +type laneBody struct { + io.ReadCloser + conn net.Conn +} + +func (b *laneBody) Close() error { + err := b.ReadCloser.Close() + b.conn.Close() + return err +} + +// drain reads a bounded slice of an answer we are throwing away so the lane +// connection closes cleanly instead of resetting. +func drain(resp *http.Response) { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20)) + _ = resp.Body.Close() +} + +// refusalPeek is how much of a 403 body is read to tell a free-tier refusal +// from any other 403. The far end's error page is far shorter than this. +const refusalPeek = 8 << 10 + +// peekedBody keeps the unread rest of a body readable and still closes the +// original: what was peeked is consumed, what follows is not. +type peekedBody struct { + io.Reader + io.Closer +} + +// freeTierRefused reports whether this 403 is the free tier refusing the exit +// that carried it ("OpenCode's free tier can only be used from within +// OpenCode"), leaving the body readable either way. The far end states the +// refusal in a compressed body — br, as the client asks for — so the peek is +// decoded before it is read. +func freeTierRefused(resp *http.Response) bool { + if resp.StatusCode != http.StatusForbidden { + return false + } + head, err := io.ReadAll(io.LimitReader(resp.Body, refusalPeek)) + resp.Body = peekedBody{Reader: io.MultiReader(bytes.NewReader(head), resp.Body), Closer: resp.Body} + if len(head) == 0 && err != nil { + return false + } + plain := head + switch strings.ToLower(strings.TrimSpace(resp.Header.Get("Content-Encoding"))) { + case "br": + if b, err := io.ReadAll(io.LimitReader(brotli.NewReader(bytes.NewReader(head)), refusalPeek)); err == nil { + plain = b + } + case "gzip": + if zr, err := gzip.NewReader(bytes.NewReader(head)); err == nil { + if b, err := io.ReadAll(io.LimitReader(zr, refusalPeek)); err == nil { + plain = b + } + } + } + return bytes.Contains(plain, []byte("FreeTierError")) || bytes.Contains(plain, []byte("free tier can only be used")) +} + +func throttled(status int) bool { + switch status { + case 429, 502, 503, 504: + return true + } + return false +} + +// retryAfter reads the far end's own reset window; 10 minutes is the floor when +// it stays silent, because a spent exit is worth seconds, not another request. +func retryAfter(resp *http.Response) time.Duration { + if v := strings.TrimSpace(resp.Header.Get("Retry-After")); v != "" { + if secs, err := strconv.Atoi(v); err == nil && secs > 0 { + return time.Duration(secs) * time.Second + } + if t, err := http.ParseTime(v); err == nil { + if d := time.Until(t); d > 0 { + return d + } + } + } + return 10 * time.Minute +} + +func readBody(req *http.Request) ([]byte, error) { + if req.Body == nil { + return nil, nil + } + defer req.Body.Close() + lr := io.LimitReader(req.Body, maxBody+1) + b, err := io.ReadAll(lr) + if err != nil { + return nil, err + } + if len(b) > maxBody { + return nil, fmt.Errorf("body over %d bytes", maxBody) + } + return b, nil +} + +func cleanHeaders(h http.Header) { + for _, k := range []string{"Connection", "Keep-Alive", "Proxy-Connection", "Proxy-Authorization", "Te", "Trailer", "Transfer-Encoding", "Upgrade"} { + h.Del(k) + } +} + +func writeStatus(w io.Writer, code int, msg string) { + body := msg + "\n" + fmt.Fprintf(w, "HTTP/1.1 %d %s\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s", + code, statusText(code), len(body), body) +} + +func statusText(code int) string { + if t := http.StatusText(code); t != "" { + return t + } + return "Status" +} + +func splitHostPort(hostport, defPort string) (string, string) { + if h, p, err := net.SplitHostPort(hostport); err == nil { + return h, p + } + return hostport, defPort +} + +func defaultPort(scheme string) string { + if scheme == "https" { + return "443" + } + return "80" +} + +func isLoopback(host string) bool { + if ip := net.ParseIP(host); ip != nil { + return ip.IsLoopback() + } + return host == "localhost" +} + +func laneTag(l *lanes.Lane) string { + if l == nil { + return "-" + } + return strconv.Itoa(l.Index) +} + +func laneCountry(l *lanes.Lane) string { + if l == nil { + return "-" + } + return l.Country +} diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go new file mode 100644 index 0000000..a392da3 --- /dev/null +++ b/internal/proxy/proxy_test.go @@ -0,0 +1,318 @@ +package proxy + +import ( + "bufio" + "bytes" + "compress/gzip" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/andybalholm/brotli" + + "lanvello/internal/lanes" +) + +// A throttled exit must cost the client nothing: the request is replayed on the +// next lane and the client sees one answer. Two lanes, first one spent. +func TestThrottledExitIsRotatedAndReplayed(t *testing.T) { + var hits atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := hits.Add(1) + if n == 1 { + w.Header().Set("Retry-After", "600") + w.WriteHeader(http.StatusTooManyRequests) + fmt.Fprint(w, `{"error":"exit spent"}`) + return + } + fmt.Fprintf(w, `{"hit":%d}`, n) + })) + defer upstream.Close() + + dir := t.TempDir() + ca, err := OpenCA(dir) + if err != nil { + t.Fatalf("ca: %v", err) + } + // noTor lanes have no socks address: the dial goes direct, which is what a + // test upstream on loopback needs. Picking and rotation stay the real code. + m := lanes.New(dir, []string{"de"}, nil, nil, true, 2) + p := New(m, ca) + p.Attempts = 3 + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + go p.Serve(ln) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + req := fmt.Sprintf("GET %s HTTP/1.1\r\nHost: %s\r\nAccept: */*\r\n\r\n", upstream.URL, upstream.Listener.Addr().String()) + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatalf("write: %v", err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("client saw %d, want 200: the throttle was not rotated away", resp.StatusCode) + } + if got := hits.Load(); got != 2 { + t.Fatalf("upstream saw %d requests, want 2 (429 then the replay)", got) + } + parked := 0 + for _, l := range m.Lanes() { + if l.LimitedTill.After(time.Now()) { + parked++ + } + } + if parked != 1 { + t.Fatalf("lanes parked for a spent exit = %d, want 1", parked) + } +} + +// A 403 that says "this exit is not OpenCode" is a spent exit like any other: +// the request is replayed on the next lane and the client never sees it. The +// free tier hands that 403 to exits it has had enough of, so rotating on it is +// what keeps a fleet of them usable. +func TestFreeTierRefusalIsRotatedAndReplayed(t *testing.T) { + var hits atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := hits.Add(1) + if n == 1 { + w.WriteHeader(http.StatusForbidden) + fmt.Fprint(w, `{"error":{"type":"FreeTierError","message":"OpenCode's free tier can only be used from within OpenCode. Download it at https://opencode.ai and keep building."}}`) + return + } + fmt.Fprintf(w, `{"hit":%d}`, n) + })) + defer upstream.Close() + + dir := t.TempDir() + ca, err := OpenCA(dir) + if err != nil { + t.Fatalf("ca: %v", err) + } + m := lanes.New(dir, []string{"de"}, nil, nil, true, 2) + p := New(m, ca) + p.Attempts = 3 + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + go p.Serve(ln) + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + req := fmt.Sprintf("GET %s HTTP/1.1\r\nHost: %s\r\nAccept: */*\r\n\r\n", upstream.URL, upstream.Listener.Addr().String()) + if _, err := conn.Write([]byte(req)); err != nil { + t.Fatalf("write: %v", err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("client saw %d, want 200: the refusal was not rotated away", resp.StatusCode) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if string(body) != `{"hit":2}` { + t.Fatalf("client body = %q, want the replayed answer", body) + } + if got := hits.Load(); got != 2 { + t.Fatalf("upstream saw %d requests, want 2 (refusal then the replay)", got) + } + parked := 0 + for _, l := range m.Lanes() { + if l.LimitedTill.After(time.Now()) { + parked++ + } + } + if parked != 1 { + t.Fatalf("lanes parked for a refused exit = %d, want 1", parked) + } +} + +// Only the free tier's own refusal earns a rotation: every other 403 is an +// answer the client asked for, and its body must reach the client untouched — +// byte for byte, still compressed, because the client decodes it. +func TestFreeTierRefusedOnlyOnTheRefusal(t *testing.T) { + refusal := `{"error":{"type":"FreeTierError","message":"OpenCode's free tier can only be used from within OpenCode. Download it at https://opencode.ai and keep building."}}` + cases := []struct { + name string + status int + enc string + body string + want bool + }{ + {name: "the refusal as it arrives", status: http.StatusForbidden, enc: "br", body: refusal, want: true}, + {name: "the refusal, gzipped", status: http.StatusForbidden, enc: "gzip", body: refusal, want: true}, + {name: "the refusal, plain", status: http.StatusForbidden, body: refusal, want: true}, + {name: "any other 403", status: http.StatusForbidden, body: `{"error":"forbidden"}`, want: false}, + {name: "an empty 403", status: http.StatusForbidden, body: "", want: false}, + {name: "a 200 that mentions the tier", status: http.StatusOK, body: "FreeTierError", want: false}, + } + for _, c := range cases { + var wire []byte + switch c.enc { + case "br": + var buf bytes.Buffer + bw := brotli.NewWriter(&buf) + if _, err := bw.Write([]byte(c.body)); err != nil { + t.Fatalf("%s: brotli: %v", c.name, err) + } + if err := bw.Close(); err != nil { + t.Fatalf("%s: brotli close: %v", c.name, err) + } + wire = buf.Bytes() + case "gzip": + var buf bytes.Buffer + zw := gzip.NewWriter(&buf) + if _, err := zw.Write([]byte(c.body)); err != nil { + t.Fatalf("%s: gzip: %v", c.name, err) + } + if err := zw.Close(); err != nil { + t.Fatalf("%s: gzip close: %v", c.name, err) + } + wire = buf.Bytes() + default: + wire = []byte(c.body) + } + hdr := http.Header{} + if c.enc != "" { + hdr.Set("Content-Encoding", c.enc) + } + resp := &http.Response{StatusCode: c.status, Header: hdr, Body: io.NopCloser(bytes.NewReader(wire))} + if got := freeTierRefused(resp); got != c.want { + t.Errorf("%s: freeTierRefused = %v, want %v", c.name, got, c.want) + } + kept, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("%s: body unreadable after the check: %v", c.name, err) + } + if !bytes.Equal(kept, wire) { + t.Errorf("%s: body changed: %d bytes after the check, %d before", c.name, len(kept), len(wire)) + } + if err := resp.Body.Close(); err != nil { + t.Errorf("%s: close: %v", c.name, err) + } + } +} + +func TestRetryAfterIsHonoredInFull(t *testing.T) { + cases := []struct { + name string + header string + want time.Duration + }{ + {"seconds", "7200", 2 * time.Hour}, + {"the 146 minute wall opencode reports", "8760", 146 * time.Minute}, + {"http date", time.Now().Add(90 * time.Second).UTC().Format(http.TimeFormat), 90 * time.Second}, + {"missing falls back", "", 10 * time.Minute}, + {"garbage falls back", "soon", 10 * time.Minute}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + resp := &http.Response{Header: http.Header{}} + if c.header != "" { + resp.Header.Set("Retry-After", c.header) + } + got := retryAfter(resp) + if c.name == "http date" { + // parsed from a round-tripped second boundary + if got > c.want || got < c.want-time.Second { + t.Fatalf("retryAfter(%q) = %s, want ~%s", c.header, got, c.want) + } + return + } + if got != c.want { + t.Fatalf("retryAfter(%q) = %s, want %s", c.header, got, c.want) + } + }) + } +} + +func TestSplitHostPortDefaults(t *testing.T) { + if h, p := splitHostPort("example.com", defaultPort("https")); h != "example.com" || p != "443" { + t.Fatalf("https default = %s/%s", h, p) + } + if h, p := splitHostPort("example.com:8443", defaultPort("https")); h != "example.com" || p != "8443" { + t.Fatalf("explicit port = %s/%s", h, p) + } + if _, p := splitHostPort("example.com", defaultPort("http")); p != "80" { + t.Fatalf("http default port = %s", p) + } +} + +func TestLoopbackIsNotLaned(t *testing.T) { + if !isLoopback("127.0.0.1") || !isLoopback("localhost") || !isLoopback("::1") { + t.Fatal("loopback must bypass the lanes: the client's own server is not lane material") + } + if isLoopback("opencode.ai") { + t.Fatal("opencode.ai must go through a lane") + } +} + +func TestChunkedAnswerReachesTheClient(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fl := w.(http.Flusher) + fmt.Fprint(w, "data: one\n\n") + fl.Flush() + fmt.Fprint(w, "data: two\n\n") + })) + defer upstream.Close() + + dir := t.TempDir() + ca, _ := OpenCA(dir) + m := lanes.New(dir, []string{"de"}, nil, nil, true, 1) + p := New(m, ca) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + go p.Serve(ln) + + conn, _ := net.Dial("tcp", ln.Addr().String()) + defer conn.Close() + req := fmt.Sprintf("GET %s HTTP/1.1\r\nHost: %s\r\n\r\n", upstream.URL, upstream.Listener.Addr().String()) + conn.Write([]byte(req)) + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read: %v", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("stream read: %v", err) + } + // The client must get the events framed so it can parse them as they + // arrive: no length was known when the headers went out. + if got := string(body); got != "data: one\n\ndata: two\n\n" { + t.Fatalf("streamed body = %q", got) + } + if len(resp.TransferEncoding) == 0 && resp.ContentLength < 0 { + t.Fatal("framing lost: neither chunked nor a length was announced") + } +} diff --git a/internal/server/parse_test.go b/internal/server/parse_test.go new file mode 100644 index 0000000..e5f71c1 --- /dev/null +++ b/internal/server/parse_test.go @@ -0,0 +1,107 @@ +package server + +import ( + "encoding/json" + "testing" +) + +// The parser is the boundary between an api caller's json and what the +// client-side plugin rebuilds the model call from: a mistake here is invisible +// in the answer and shows up as the client running its own agent. +func TestParseChatRequest(t *testing.T) { + chat := map[string]any{ + "reasoning_effort": "high", + "messages": []any{ + map[string]any{"role": "system", "content": "be brief"}, + map[string]any{"role": "system", "content": "answer in french"}, + map[string]any{"role": "user", "content": "weather in paris?"}, + map[string]any{"role": "assistant", "tool_calls": []any{ + map[string]any{ + "id": "call_1", + "type": "function", + "function": map[string]any{ + "name": "get_weather", + "arguments": `{"city":"paris"}`, + }, + }, + }}, + map[string]any{"role": "tool", "tool_call_id": "call_1", "content": "18C"}, + }, + "tools": []any{ + map[string]any{ + "type": "function", + "function": map[string]any{ + "name": "get_weather", + "description": "current weather", + "parameters": map[string]any{ + "type": "object", + "properties": map[string]any{"city": map[string]any{"type": "string"}}, + }, + }, + }, + map[string]any{"name": "ping", "description": "flat shape"}, + }, + } + + got := parseChatRequest(chat) + + if got.System != "be brief\n\nanswer in french" { + t.Errorf("system = %q, want both system messages joined", got.System) + } + if got.Variant != "high" { + t.Errorf("variant = %q, want high", got.Variant) + } + // the prompt is the trigger the client runs: the last user text, not the + // tool result that follows it + if got.Prompt != "weather in paris?" { + t.Errorf("prompt = %q, want the last user text", got.Prompt) + } + if len(got.Messages) != 3 { + t.Fatalf("messages = %d, want 3 (user, assistant, tool)", len(got.Messages)) + } + am := got.Messages[1] + if am.Role != "assistant" || len(am.ToolCalls) != 1 { + t.Fatalf("assistant %+v, want one tool call", am) + } + if tc := am.ToolCalls[0]; tc.ID != "call_1" || tc.Name != "get_weather" || string(tc.Arguments) != `{"city":"paris"}` { + t.Errorf("tool call %+v, want the caller's id, name and arguments", tc) + } + if tm := got.Messages[2]; tm.Role != "tool" || tm.ToolCallID != "call_1" || tm.Content != "18C" { + t.Errorf("tool message %+v, want the result tied to call_1", tm) + } + + if len(got.Tools) != 2 { + t.Fatalf("tools = %d, want 2 (wrapped and flat)", len(got.Tools)) + } + if w := got.Tools[0]; w.Name != "get_weather" || w.Description != "current weather" { + t.Errorf("tool %+v, want the wrapped function unpacked", w) + } + var schema map[string]any + if err := json.Unmarshal(got.Tools[0].Parameters, &schema); err != nil { + t.Fatalf("parameters are not json: %v", err) + } + if schema["type"] != "object" { + t.Errorf("parameters = %s, want the caller's schema through", got.Tools[0].Parameters) + } + // a tool with no schema still has to carry a valid object schema: the + // client refuses a definition without one + if p := string(got.Tools[1].Parameters); p != `{"type":"object","properties":{}}` { + t.Errorf("parameters of a schema-less tool = %s, want an empty object schema", p) + } +} + +// An api caller that sends no tools must not inherit the client's own. +func TestParseChatRequestWithoutTools(t *testing.T) { + got := parseChatRequest(map[string]any{ + "messages": []any{map[string]any{"role": "user", "content": "hi"}}, + }) + if len(got.Tools) != 0 { + t.Errorf("tools = %d, want none", len(got.Tools)) + } + if got.System != "" { + t.Errorf("system = %q, want empty", got.System) + } + if got.Prompt != "hi" { + t.Errorf("prompt = %q, want hi", got.Prompt) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 73a14e4..baa26d4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -4,14 +4,20 @@ package server import ( "bytes" + "context" + "crypto/rand" + "encoding/hex" "encoding/json" + "fmt" "io" "net/http" "strings" "sync" "time" + "lanvello/internal/bridge" "lanvello/internal/caps" + "lanvello/internal/clientapi" "lanvello/internal/keys" "lanvello/internal/lanes" "lanvello/internal/translate" @@ -27,6 +33,17 @@ type Server struct { Lane *lanes.Manager Up *upstream.Client Caps *caps.Store + // ClientAPI, when set, is where inference goes: the running opencode + // client makes the call (it holds the free-tier session), we translate. + // The replay path (Up) cannot serve the free tier at all any more, so + // this is not an optimisation, it is the only way in. + ClientAPI *clientapi.Client + // BridgeDir is where a caller's request is parked for the client-side + // plugin (system, history, tools). Empty disables the bridge: the client + // then answers with its own agent and its own tools. + BridgeDir string + // pollGap is how often the client's session is read while answering. + pollGap time.Duration // RequireKey refuses keyless /v1 requests even before the first key is // issued (config `requireKey` / LANVELLO_REQUIRE_KEY): a public // deployment must never fall back to open mode. @@ -46,7 +63,7 @@ type modelEnt struct { } func New(ks *keys.Store, lm *lanes.Manager, up *upstream.Client, dataDir string) *Server { - s := &Server{Keys: ks, Lane: lm, Up: up, Caps: caps.Open(dataDir, up, lm)} + s := &Server{Keys: ks, Lane: lm, Up: up, Caps: caps.Open(dataDir, up, lm), pollGap: 250 * time.Millisecond} // warm the capability table in the background; never blocks // startup and never dials directly. go s.Caps.MaybeRefresh() @@ -150,6 +167,25 @@ func defaultEntries() []modelEnt { } func (s *Server) fetchCatalog() []modelEnt { + // Client-backed: the client is the authority on what it can drive, and its + // list needs no lane (the catalog mirror is not gated, but the client has + // already fetched it). + if s.ClientAPI != nil { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if mods, err := s.ClientAPI.Models(ctx); err == nil { + var out []modelEnt + for _, m := range mods { + if m.ProviderID != "opencode" || m.ID == "" { + continue + } + out = append(out, modelEnt{ID: "opencode/" + m.ID}) + } + if len(out) > 0 { + return out + } + } + } status, body := s.Up.GetJSON("/zen/v1/models") if status != 200 || len(body) == 0 { return nil @@ -306,6 +342,13 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { wantStream, _ := chat["stream"].(bool) id := s.identity(r) + // The free tier leases a session to the running client, so when one is + // wired up the client is the upstream: our own request cannot get in. + if s.ClientAPI != nil { + s.chatViaClient(w, r, chat, model, wantStream) + return + } + if !upstream.IsResponsesModel(model) { fwd := s.fingerprintChat(chat) if wantStream { @@ -344,6 +387,316 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(translate.AggregateResponsesConv(body, "opencode/"+model)) } +// chatViaClient answers one chat request through the running opencode client. +// +// The request is parked in the bridge spool (when the plugin is installed), so +// the client-side plugin can rebuild the model call from the caller's own +// system prompt, history and tool schemas — the client's http api takes nothing +// but text. Tool calls the model makes are recorded by the plugin instead of +// being executed, and come back here as an openai tool_calls turn. +func (s *Server) chatViaClient(w http.ResponseWriter, r *http.Request, chat map[string]any, model string, wantStream bool) { + req := parseChatRequest(chat) + if req.Prompt == "" { + http.Error(w, "chat requires messages", 400) + return + } + ctx, cancel := context.WithTimeout(r.Context(), 15*time.Minute) + defer cancel() + + sessionID, err := s.ClientAPI.Create(ctx, "opencode", model, req.Variant) + if err != nil { + clientError(w, err) + return + } + defer s.ClientAPI.Drop(context.WithoutCancel(ctx), sessionID) + if s.BridgeDir != "" { + parked := bridge.Request{SessionID: sessionID, System: req.System, Messages: req.Messages, Tools: req.Tools, Model: model} + if err := bridge.Write(s.BridgeDir, parked); err != nil { + clientError(w, err) + return + } + defer bridge.Remove(s.BridgeDir, sessionID) + } + if err := s.ClientAPI.Prompt(ctx, sessionID, req.Prompt); err != nil { + clientError(w, err) + return + } + + id := "chatcmpl-" + randHex(16) + created := time.Now().Unix() + name := "opencode/" + model + var out *sseWriter + if wantStream { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(200) + out = &sseWriter{w: w} + out.chunk(openaiChunk(id, created, name, map[string]any{"role": "assistant", "content": ""}, nil)) + } + + // Poll the client: text grows into the answer, and a tool call the plugin + // recorded ends the turn — the caller runs that tool, not the client. + var last string + var calls []bridge.ToolCall + var snap clientapi.Snapshot + quiet := time.Now() + for { + select { + case <-ctx.Done(): + err = ctx.Err() + case <-time.After(s.pollGap): + } + if err != nil { + break + } + snap, err = s.ClientAPI.Snapshot(ctx, sessionID) + if err != nil { + break + } + if len(snap.Text) > len(last) { + if out != nil { + out.chunk(openaiChunk(id, created, name, map[string]any{"content": snap.Text[len(last):]}, nil)) + } + last = snap.Text + quiet = time.Now() + } + if s.BridgeDir != "" { + if found := bridge.Calls(s.BridgeDir, sessionID); len(found) > 0 { + calls = found + break + } + } + if snap.Idle || (last != "" && time.Since(quiet) > idleGrace) { + break + } + } + if snap.Idle { + // the answer is over; the usage the client counted rides along + } else if last == "" && len(calls) == 0 { + if err == nil { + err = fmt.Errorf("client stopped without an answer") + } + clientError(w, err) + return + } + + finish := finishReason(snap.Outcome, len(calls)) + if wantStream { + if len(calls) > 0 { + for i, c := range calls { + out.chunk(openaiChunk(id, created, name, map[string]any{ + "tool_calls": []any{map[string]any{"index": i, "id": c.ID, "type": "function", + "function": map[string]any{"name": c.Name, "arguments": ""}}}}, nil)) + out.chunk(openaiChunk(id, created, name, map[string]any{ + "tool_calls": []any{map[string]any{"index": i, "function": map[string]any{"arguments": string(c.Arguments)}}}}, nil)) + } + } + out.chunk(openaiChunk(id, created, name, map[string]any{}, &finish)) + _, _ = io.WriteString(w, "data: [DONE]\n\n") + out.flush() + return + } + msg := map[string]any{"role": "assistant", "content": last} + if len(calls) > 0 { + msg["tool_calls"] = openAIToolCalls(calls) + if last == "" { + msg["content"] = nil + } + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(200) + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": id, "object": "chat.completion", "created": created, "model": name, + "choices": []any{map[string]any{"index": 0, "message": msg, "finish_reason": finish}}, + "usage": openAIUsage(snap.Usage), + }) +} + +const idleGrace = 3 * time.Second + +func openaiChunk(id string, created int64, model string, delta map[string]any, finish *string) map[string]any { + var fr any + if finish != nil { + fr = *finish + } + return map[string]any{ + "id": id, "object": "chat.completion.chunk", "created": created, "model": model, + "choices": []any{map[string]any{"index": 0, "delta": delta, "finish_reason": fr}}, + } +} + +func openAIToolCalls(calls []bridge.ToolCall) []any { + out := make([]any, 0, len(calls)) + for _, c := range calls { + args := string(c.Arguments) + if args == "" { + args = "{}" + } + out = append(out, map[string]any{ + "id": c.ID, "type": "function", + "function": map[string]any{"name": c.Name, "arguments": args}, + }) + } + return out +} + +// chatRequest is the caller's request, flattened into what the plugin needs. +type chatRequest struct { + Prompt string + System string + Messages []bridge.Message + Tools []bridge.Tool + Variant string +} + +// parseChatRequest reads an openai chat body: the last user text becomes the +// prompt that wakes the client, everything else travels through the spool. +func parseChatRequest(chat map[string]any) chatRequest { + var out chatRequest + out.Variant, _ = chat["reasoning_effort"].(string) + msgs, _ := chat["messages"].([]any) + for _, m := range msgs { + mm, _ := m.(map[string]any) + role, _ := mm["role"].(string) + text := messageText(mm["content"]) + switch role { + case "system", "developer": + if out.System != "" { + out.System += "\n\n" + } + out.System += text + case "assistant": + am := bridge.Message{Role: "assistant", Content: text} + if tcs, ok := mm["tool_calls"].([]any); ok { + for _, tc := range tcs { + t, _ := tc.(map[string]any) + fn, _ := t["function"].(map[string]any) + name, _ := fn["name"].(string) + args, _ := fn["arguments"].(string) + id, _ := t["id"].(string) + am.ToolCalls = append(am.ToolCalls, bridge.ToolCall{ID: id, Name: name, Arguments: json.RawMessage(args)}) + } + } + out.Messages = append(out.Messages, am) + if text != "" { + out.Prompt = text + } + case "tool": + id, _ := mm["tool_call_id"].(string) + out.Messages = append(out.Messages, bridge.Message{Role: "tool", Content: text, ToolCallID: id}) + default: + out.Messages = append(out.Messages, bridge.Message{Role: "user", Content: text}) + if text != "" { + out.Prompt = text + } + } + } + if tools, ok := chat["tools"].([]any); ok { + for _, t := range tools { + tm, _ := t.(map[string]any) + fn, _ := tm["function"].(map[string]any) + if fn == nil { + fn = tm + } + name, _ := fn["name"].(string) + if name == "" { + continue + } + desc, _ := fn["description"].(string) + params, _ := json.Marshal(fn["parameters"]) + if string(params) == "null" { + params = json.RawMessage(`{"type":"object","properties":{}}`) + } + out.Tools = append(out.Tools, bridge.Tool{Name: name, Description: desc, Parameters: params}) + } + } + return out +} + +// finishReason maps the client's outcome onto an openai finish reason. +func finishReason(outcome string, calls int) string { + if calls > 0 { + return "tool_calls" + } + if outcome != "" && outcome != "succeeded" { + return "error" + } + return "stop" +} + +func clientError(w http.ResponseWriter, err error) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadGateway) + _, _ = io.WriteString(w, `{"error":{"type":"client_upstream_error","message":`+quote(err.Error())+`}}`) +} + +func quote(s string) string { + b, _ := json.Marshal(s) + return string(b) +} + +func openAIUsage(u clientapi.Usage) map[string]any { + return map[string]any{ + "prompt_tokens": u.Input, + "completion_tokens": u.Output, + "total_tokens": u.Input + u.Output, + "prompt_tokens_details": map[string]any{"cached_tokens": u.CacheRead}, + "completion_tokens_details": map[string]any{"reasoning_tokens": u.Reasoning}, + } +} + +func messageText(v any) string { + switch t := v.(type) { + case string: + return t + case []any: + var b strings.Builder + for _, p := range t { + pm, _ := p.(map[string]any) + if s, _ := pm["text"].(string); s != "" { + b.WriteString(s) + } + } + return b.String() + } + return "" +} + +// sseWriter frames one openai chunk per event and flushes it, so tokens reach +// the caller while the model is still talking. +type sseWriter struct { + w http.ResponseWriter + fl http.Flusher +} + +func (s *sseWriter) chunk(v any) { + b, err := json.Marshal(v) + if err != nil { + return + } + _, _ = fmt.Fprintf(s.w, "data: %s\n\n", b) + s.flush() +} + +func (s *sseWriter) flush() { + if s.fl == nil { + s.fl, _ = s.w.(http.Flusher) + } + if s.fl != nil { + s.fl.Flush() + } +} + +func randHex(n int) string { + var b [32]byte + if _, err := rand.Read(b[:]); err != nil { + return strings.Repeat("0", n) + } + return hex.EncodeToString(b[:])[:n] +} + // fingerprintChat prepares a native chat body for the free gate. func (s *Server) fingerprintChat(chat map[string]any) []byte { chat["stream"] = true