diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index 1338ec0..8b91a6a 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -42,6 +42,9 @@ Flags: --extract-model/-provider/-auth/-base still override the file's transport. --upstream forward ALL requests here (e.g. an eval gateway); default routes by provider (api.anthropic.com / api.openai.com) + --mode initial reduction mode: on|off|deterministic (default on). Settable at + runtime without a restart: POST /labcx/mode {"mode":"off"} (GET reads it). + off = transparent passthrough; deterministic is reserved (behaves like on). --max-body-bytes cap request body size in bytes (default 33554432 = 32 MiB); 0 means no cap --upstream-timeout max total time per upstream request (e.g. 30s; default 0s = no @@ -77,6 +80,7 @@ func runProxy(args []string) { preset := fs.String("preset", "balanced", "settings preset") configPath := fs.String("config", "", "path to a YAML config file (overrides --preset; see configs/lab-cx.yaml)") upstream := fs.String("upstream", "", "forward all requests to this base URL") + mode := fs.String("mode", "on", "initial reduction mode: on|off|deterministic (runtime-settable via POST /labcx/mode)") extractModel := fs.String("extract-model", "", "enable cheap-model extraction with this model (e.g. claude-haiku-4-5)") extractProvider := fs.String("extract-provider", "anthropic", "extraction model provider: anthropic|openai") extractBase := fs.String("extract-base", "", "base URL for the extraction model (default per provider)") @@ -96,6 +100,12 @@ func runProxy(args []string) { os.Exit(2) } + initialMode, ok := proxyhttp.ParseMode(*mode) + if !ok { + fmt.Fprintf(os.Stderr, "lab-cx: invalid --mode %q (want on|off|deterministic)\n", *mode) + os.Exit(2) + } + if os.Getenv("LABCX_DISABLE") == "1" { fmt.Fprintln(os.Stderr, "lab-cx: LABCX_DISABLE=1 — running as a transparent passthrough") } @@ -195,6 +205,7 @@ func runProxy(args []string) { Aggregator: agg, MaxBodyBytes: *maxBodyBytes, UpstreamTimeout: upstreamTimeoutDur, + InitialMode: initialMode, } if useIncoming { // A compactor reuses the proxied request's own model + credentials: build a diff --git a/internal/proxyhttp/proxy.go b/internal/proxyhttp/proxy.go index 6d18404..3eed989 100644 --- a/internal/proxyhttp/proxy.go +++ b/internal/proxyhttp/proxy.go @@ -13,6 +13,7 @@ import ( "io" "net/http" "strings" + "sync/atomic" "time" "github.com/kagenti/lab-context-engineering/engine" @@ -20,6 +21,32 @@ import ( "github.com/kagenti/lab-context-engineering/surfaces" ) +// Mode is the proxy's process-wide runtime reduction mode, settable without a +// restart via POST /labcx/mode. It is the new-stack equivalent of the old +// CE-Manager activation: a KV-pressure monitor flips it on/off live. +type Mode string + +const ( + ModeOn Mode = "on" // reduce with the startup-configured method + ModeOff Mode = "off" // bypass: forward the original request untouched + ModeDeterministic Mode = "deterministic" // reserved; currently behaves like ModeOn +) + +// ParseMode validates a mode string (case/space-insensitive). ok=false for anything +// outside the enum, so callers can reject it with HTTP 400. +func ParseMode(s string) (Mode, bool) { + switch Mode(strings.ToLower(strings.TrimSpace(s))) { + case ModeOn: + return ModeOn, true + case ModeOff: + return ModeOff, true + case ModeDeterministic: + return ModeDeterministic, true + default: + return "", false + } +} + // Config configures the proxy handler. type Config struct { Engine *engine.Engine @@ -47,10 +74,15 @@ type Config struct { // "no per-request model" (those compactors then no-op). The host (proxy binary) // supplies this so cheapmodel construction stays out of this package. BuildRequestModel func(surfaceName, model, base string, h http.Header) engine.Model + // InitialMode is the reduction mode at startup (on|off|deterministic). Empty or + // invalid defaults to ModeOn, preserving the always-reduce default. Settable at + // runtime via POST /labcx/mode. + InitialMode Mode } type proxy struct { - cfg Config + cfg Config + mode atomic.Value // holds Mode; runtime-settable via POST /labcx/mode } // New returns the proxy http.Handler. Routing is by path SUFFIX so it works both for @@ -74,7 +106,13 @@ func New(cfg Config) http.Handler { if cfg.Emitter == nil { cfg.Emitter = observability.Nop{} } - return &proxy{cfg: cfg} + p := &proxy{cfg: cfg} + initial := cfg.InitialMode + if _, ok := ParseMode(string(initial)); !ok { + initial = ModeOn + } + p.mode.Store(initial) + return p } func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -84,6 +122,8 @@ func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { p.ok(w, r) case path == "/labcx/expand": p.expand(w, r) + case path == "/labcx/mode": + p.modeControl(w, r) case path == "/stats": p.stats(w, r) case strings.HasSuffix(path, "/v1/messages"): @@ -119,6 +159,46 @@ func (p *proxy) expand(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, original) } +// currentMode returns the live reduction mode (defaults to ModeOn if unset). +func (p *proxy) currentMode() Mode { + if m, ok := p.mode.Load().(Mode); ok { + return m + } + return ModeOn +} + +// modeControl gets or sets the process-wide reduction mode at runtime — no restart. +// GET → {"mode":""}. POST/PUT reads {"mode":"on|off|deterministic"} (or the +// ?mode= query) and stores it; an unknown value is rejected with HTTP 400. This is the +// switch a KV-pressure monitor flips to activate/deactivate context engineering live. +func (p *proxy) modeControl(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + // fall through and report the current mode + case http.MethodPost, http.MethodPut: + want := r.URL.Query().Get("mode") + if want == "" { + var body struct { + Mode string `json:"mode"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&body); err == nil { + want = body.Mode + } + } + m, ok := ParseMode(want) + if !ok { + http.Error(w, "invalid mode (want on|off|deterministic)", http.StatusBadRequest) + return + } + p.mode.Store(m) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"mode": string(p.currentMode())}) +} + // upstreamBase returns the base URL to forward to: the configured single upstream, // or the provider default. func (p *proxy) upstreamBase(providerDefault string) string { @@ -135,9 +215,10 @@ func (p *proxy) model(surface surfaces.Surface, providerDefault string) http.Han if err != nil { return // readBody already wrote the response } - // Reduce unless bypassed or the surface can't map this format. + // Reduce unless bypassed (per-request header or process-wide mode=off) or the + // surface can't map this format. outBody := body - if !bypassed(r) { + if !bypassed(r) && p.currentMode() != ModeOff { start := time.Now() ctx := r.Context() if p.cfg.BuildRequestModel != nil { diff --git a/internal/proxyhttp/proxy_test.go b/internal/proxyhttp/proxy_test.go index 03304b6..d3a390c 100644 --- a/internal/proxyhttp/proxy_test.go +++ b/internal/proxyhttp/proxy_test.go @@ -80,6 +80,93 @@ func TestBypassForwardsOriginal(t *testing.T) { } } +// TestRuntimeModeToggle: POST /labcx/mode flips reduction on/off without a restart. +// mode=off → forward the original untouched; mode=on → reduce again. Bad value → 400. +func TestRuntimeModeToggle(t *testing.T) { + var upstreamGot string + up := captureUpstream(t, &upstreamGot) + defer up.Close() + h := newProxy(t, up.URL) + + big := strings.Repeat("a.go long file content line for the read result here\\n", 12) + body := `{"messages":[ + {"role":"assistant","content":[{"type":"tool_use","id":"u1","name":"read","input":{"file_path":"a.go"}}]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"u1","content":"` + big + `"}]}, + {"role":"assistant","content":[{"type":"tool_use","id":"u2","name":"edit","input":{"file_path":"a.go"}}]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"u2","content":"success: edited"}]}, + {"role":"user","content":[{"type":"text","text":"thanks"}]} + ]}` + + post := func(payload string) string { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/labcx/mode", strings.NewReader(payload)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("POST /labcx/mode %s = %d", payload, rec.Code) + } + var got struct{ Mode string } + _ = json.Unmarshal(rec.Body.Bytes(), &got) + return got.Mode + } + + // Default is "on": a reduce-eligible request gets reduced. + { + req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if len(upstreamGot) >= len(body) { + t.Fatalf("default mode should reduce: got %d, original %d", len(upstreamGot), len(body)) + } + } + + // Flip to off → the next request is forwarded verbatim. + if m := post(`{"mode":"off"}`); m != "off" { + t.Fatalf("mode after off = %q", m) + } + upstreamGot = "" + { + req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if upstreamGot != body { + t.Fatalf("mode=off should forward original verbatim:\n got: %s\nwant: %s", upstreamGot, body) + } + } + + // Flip back to on → reduction resumes. + if m := post(`{"mode":"on"}`); m != "on" { + t.Fatalf("mode after on = %q", m) + } + upstreamGot = "" + { + req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(body)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if len(upstreamGot) >= len(body) { + t.Fatalf("mode=on should reduce again: got %d, original %d", len(upstreamGot), len(body)) + } + } + + // Unknown value → 400, mode unchanged. + req := httptest.NewRequest(http.MethodPost, "/labcx/mode", strings.NewReader(`{"mode":"bogus"}`)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("bad mode should be 400, got %d", rec.Code) + } + + // GET reports the current mode. + greq := httptest.NewRequest(http.MethodGet, "/labcx/mode", nil) + grec := httptest.NewRecorder() + h.ServeHTTP(grec, greq) + var got struct{ Mode string } + _ = json.Unmarshal(grec.Body.Bytes(), &got) + if got.Mode != "on" { + t.Fatalf("GET mode = %q, want on", got.Mode) + } +} + func TestPassthroughForwardsUnknownPaths(t *testing.T) { var upstreamGot string up := captureUpstream(t, &upstreamGot)