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