Feat: runtime reduction mode API (on/off/deterministic)

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

internal/proxyhttp/proxy.go

@@ -13,13 +13,40 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/kagenti/lab-context-engineering/engine"
 	"github.com/kagenti/lab-context-engineering/observability"
 	"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":"<current>"}. 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 {

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)