postprocess: strip_actual_usage; bodylog keeps actual_usage; rename rate_limit to load_rate_limit

.github/workflows/test.yml

@@ -1,4 +1,5 @@
-# Unit tests for the Go components.
+# Unit tests: the Go components (bodylog-listener) and the Lua engine modules
+# (test/utest_*.lua, run with the resty CLI by test/run_lua_unit_tests.sh).
 #
 # Separate from release.yml so a pull request is tested without going anywhere
 # near the publish path. release.yml runs the same tests as a gate before it
@@ -7,13 +8,17 @@ name: test
 
 on:
   pull_request:
-    paths: &go_paths
+    paths: &test_paths
       - 'bodylog-listener-go/**'
       - 'Dockerfile.bodylog'
+      - 'lua/**'
+      - 'test/utest_*.lua'
+      - 'test/run_lua_unit_tests.sh'
+      - 'Dockerfile'
       - '.github/workflows/test.yml'
   push:
     branches: [main]
-    paths: *go_paths
+    paths: *test_paths
   workflow_dispatch:
 
 permissions:
@@ -37,3 +42,23 @@ jobs:
         env:
           CGO_ENABLED: 1
         run: go test ./...
+
+  lua:
+    # jammy, so the OpenResty pinned in the Dockerfile (OPENRESTY_VER) installs as is
+    runs-on: ubuntu-22.04
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: install openresty + resty CLI (the Dockerfile's version)
+        run: |
+          set -eux
+          ver=$(sed -n 's/^ARG OPENRESTY_VER=//p' Dockerfile)
+          test -n "$ver"
+          wget -qO - https://openresty.org/package/pubkey.gpg | sudo gpg --dearmor -o /usr/share/keyrings/openresty.gpg
+          echo "deb [arch=amd64 signed-by=/usr/share/keyrings/openresty.gpg] http://openresty.org/package/ubuntu jammy main" \
+            | sudo tee /etc/apt/sources.list.d/openresty.list
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends "openresty=$ver" "openresty-resty=$ver"
+
+      - name: lua unit tests
+        run: bash test/run_lua_unit_tests.sh /usr/local/openresty/bin/resty

README.md

@@ -238,6 +238,42 @@ pairs. The fix for a small pool is to *lower* `session_prio_r_new`, or supply a
 also that under adaptive concurrency the limit itself is a float that moves over
 time, so the effective number of distinct ceilings moves with it.
 
+## Response post-processing
+
+A route can rewrite upstream responses before they reach the client, to hide backend
+quirks. It is off unless the route names a handler (`lua/postprocess.lua`):
+
+```yaml
+# ModelRoute
+spec:
+  nginx:
+    values:
+      postprocess: strip_actual_usage
+```
+
+(in a route conf: `postprocess = "strip_actual_usage"`; `_G.POSTPROCESS_DEFAULT` in
+`session_base.conf` sets one for every route that names none.)
+
+| handler | what it does |
+|---|---|
+| `strip_actual_usage` | removes the top-level `actual_usage` some sglang builds return next to `usage` (a second token count, e.g. with different `cached_tokens`). Nothing else is touched. |
+| `vllm_format` | normalizes vLLM's response shape: drops vLLM's placeholder fields when they are null (`prompt_token_ids`, `prompt_logprobs`, `logprobs`, `token_ids`, `refusal`, `annotations`, ...), always drops `stop_reason`, `service_tier`, `system_fingerprint`, `kv_transfer_params`, `ec_transfer_params` and `actual_usage`, and renames `reasoning` to `reasoning_content` and `created_cache_tokens` to `cache_write_tokens`. |
+
+How it behaves:
+
+- One handler per route.
+- Only 2xx responses, and only `text/event-stream` (each `data:` frame) and
+  `application/json` bodies; anything else passes through.
+- A frame or body with nothing to change is sent byte-identical. A changed one is
+  re-encoded, so its key order and `/` escaping can differ from the backend's.
+- `Content-Length` is dropped on the route's 2xx responses (they are sent chunked),
+  and a non-stream JSON body is held until it is complete (up to 8 MiB; a larger
+  one passes through unchanged). Streams are not held: frames go out as they arrive.
+- Fail-open: a frame that does not parse, or a handler error, leaves the original bytes.
+- The bodylog sees the raw upstream response (it runs first). The listener keeps
+  `usage` and `actual_usage` in the record's `resp_meta`, so a stripped
+  `actual_usage` is still in the bodylog.
+
 ## Authentication
 
 Keys are read from a file, default `/etc/openresty/api-keys/keys`, overridable with
@@ -305,15 +341,17 @@ never added to the real 429 counters. Content rules (`reject_rules`) have their
 switch and are not covered. Three layers, highest first:
 - the runtime override `POST /<route>/_rate_limit_toggle?on=0|1` (127.0.0.1 only) takes effect without a reload
   and survives one; `?reset=1` removes it;
-- the route option `rate_limit` (route conf, or `nginx.values` on a ModelRoute);
-- the global `_G.RATE_LIMIT_ENABLED` in `session_base.conf`.
+- the route option `load_rate_limit` (route conf, or `nginx.values` on a ModelRoute). Named so,
+  not `rate_limit`: on a ModelRoute that key is a video route's download speed limit, and the CRD
+  rejects it on llm routes;
+- the global `_G.LOAD_RATE_LIMIT_ENABLED` in `session_base.conf`.
 
 Things to watch when switching it off at runtime. The override lives in this openresty
 instance's shared memory: with several replicas (e.g. two openresty pods) send it to every one of
 them, and a restarted instance comes back without it. It survives reloads and lasts until
 `?reset=1` or a full restart, so remove it once the incident is over (`/_rate_limit_status` shows
 `source: runtime` while it is in place). To keep limiting off for longer than an incident, set the
-route option (`rate_limit: false` in the ModelRoute's `nginx.values`) instead: it is part of the
+route option (`load_rate_limit: false` in the ModelRoute's `nginx.values`) instead: it is part of the
 deployed configuration and applies to every instance. And the adaptive limit keeps moving on traffic it no
 longer caps -- down under overload, up under pressure -- so switching back on can bring a burst of
 429s until it settles. Likewise `session_prio`'s turn-1 hold flag is neither set nor released while
@@ -387,12 +425,16 @@ cd test
 ./test_lazy_init_timing.sh      # lazy route registration across workers
 ```
 
-Pure-Lua unit tests need only `resty`:
+Pure-Lua unit tests (`test/utest_*.lua`) need only `resty`. Run them all from the
+repository root, as CI does on every pull request that touches `lua/` or a unit test:
 
 ```bash
-resty utest_surgical_normalize.lua ../lua/reqtransform.lua
+bash test/run_lua_unit_tests.sh [path/to/resty]
 ```
 
+It exits non-zero if any test fails, or if it finds none. A test that needs a special
+invocation (a shared dict, another working directory) is listed in the script.
+
 A note on how these are written: several assertions exist specifically to fail when
 the thing under test is removed. A check that only confirms "no error appeared" also
 passes when the feature never ran, so the harnesses pair each such check with a

bodylog-listener-go/main.go

@@ -272,6 +272,7 @@ type respFields struct {
 	finishReason string
 	model        string
 	usage        any
+	actualUsage  any // some sglang builds: a second token count next to usage (openresty may strip it from the client response)
 	toolCalls    []any
 	stopReason   string // Anthropic
 	errorObj     any
@@ -365,6 +366,9 @@ func extractFromObj(obj map[string]any, rf *respFields) {
 	if u, ok := obj["usage"].(map[string]any); ok && u != nil {
 		rf.usage = u
 	}
+	if u, ok := obj["actual_usage"].(map[string]any); ok && u != nil {
+		rf.actualUsage = u
+	}
 	if m, ok := obj["model"].(string); ok && m != "" {
 		rf.model = m
 	}
@@ -490,6 +494,9 @@ func buildRespMeta(rf respFields) map[string]any {
 	if rf.usage != nil {
 		m["usage"] = rf.usage
 	}
+	if rf.actualUsage != nil {
+		m["actual_usage"] = rf.actualUsage
+	}
 	if len(rf.toolCalls) > 0 {
 		m["tool_calls"] = reassembleToolCalls(rf.toolCalls)
 	}

bodylog-listener-go/resp_meta_actual_usage_test.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+// Some sglang builds return a top-level actual_usage next to usage. openresty can strip it
+// from the client response (postprocess strip_actual_usage), so the bodylog is the only place
+// it survives: resp_meta must carry it, as it carries usage.
+func TestRespMetaKeepsActualUsage(t *testing.T) {
+	usage := `{"prompt_tokens":4096,"completion_tokens":128,"total_tokens":4224,"prompt_tokens_details":{"cached_tokens":2048,"cache_write_tokens":1792}}`
+	actual := `{"prompt_tokens":4096,"completion_tokens":128,"total_tokens":4224,"prompt_tokens_details":{"cached_tokens":3072}}`
+	cases := map[string]string{
+		"stream": `data: {"id":"c1","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":null}]}` + "\n\n" +
+			`data: {"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}` + "\n\n" +
+			`data: {"id":"c1","choices":[],"usage":` + usage + `,"actual_usage":` + actual + `}` + "\n\n" +
+			"data: [DONE]\n\n",
+		"json": `{"id":"c2","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":` +
+			usage + `,"actual_usage":` + actual + `}`,
+	}
+	for name, body := range cases {
+		meta := buildRespMeta(extractResp(body))
+		if meta == nil {
+			t.Fatalf("%s: no resp_meta", name)
+		}
+		b, _ := json.Marshal(meta)
+		var got struct {
+			Usage struct {
+				Details struct {
+					Cached int `json:"cached_tokens"`
+				} `json:"prompt_tokens_details"`
+			} `json:"usage"`
+			Actual *struct {
+				Prompt  int `json:"prompt_tokens"`
+				Details struct {
+					Cached int `json:"cached_tokens"`
+				} `json:"prompt_tokens_details"`
+			} `json:"actual_usage"`
+		}
+		if err := json.Unmarshal(b, &got); err != nil {
+			t.Fatalf("%s: %v", name, err)
+		}
+		if got.Usage.Details.Cached != 2048 {
+			t.Errorf("%s: usage.cached_tokens = %d, want 2048 (%s)", name, got.Usage.Details.Cached, b)
+		}
+		if got.Actual == nil {
+			t.Fatalf("%s: resp_meta has no actual_usage: %s", name, b)
+		}
+		if got.Actual.Details.Cached != 3072 || got.Actual.Prompt != 4096 {
+			t.Errorf("%s: actual_usage = %+v, want cached 3072 / prompt 4096", name, *got.Actual)
+		}
+	}
+
+	// absent upstream -> absent in resp_meta (no empty key)
+	meta := buildRespMeta(extractResp(`{"choices":[{"message":{"content":"x"},"finish_reason":"stop"}],"usage":` + usage + `}`))
+	if _, ok := meta["actual_usage"]; ok {
+		t.Errorf("actual_usage present though upstream sent none: %v", meta)
+	}
+}

lua/debug_endpoints.lua

@@ -502,8 +502,8 @@ local function rate_limit_view(opts)
         enabled          = enabled,
         source           = source,                       -- runtime | route | global
         runtime_override = override,
-        route_value      = opts.rate_limit,              -- route conf, or the global when unset
-        global_default   = _G.RATE_LIMIT_ENABLED ~= false,
+        route_value      = opts.load_rate_limit,              -- route conf, or the global when unset
+        global_default   = _G.LOAD_RATE_LIMIT_ENABLED ~= false,
         covers           = route.RATE_LIMIT_REASONS,
         -- While the switch is off: per gate, the requests that tripped it and were let through.
         -- A trip, not a refusal: a probe slot or a warm turn in session_prio's soft band might have

lua/directives.lua

@@ -62,7 +62,7 @@ M.DIRECTIVES = {
 
   { key = "default_max", label = "Default peer maximum", group = "Capacity & failover", kind = "number",
     detail = "In-flight requests allowed per peer when the peer entry names no maximum of its own. The static pool maximum -- what the adaptive limit is bounded by -- is the largest per-tier sum of peer maximums." },
-  { key = "rate_limit", label = "Rate limiting", group = "Capacity & failover", kind = "boolean",
+  { key = "load_rate_limit", label = "Load-based rate limiting", group = "Capacity & failover", kind = "boolean",
     detail = "Master switch for the load-based 429s: concurrency (the static pool limit or the adaptive one, session_prio's per-turn shares included), TTFT (session_prio's graded gate included) and decode rate. Off lets those requests through while everything behind them -- the measurements, the adaptive limit, health bans and the 503 for an all-banned pool -- keeps running, and /_rate_limit_status counts what would have been refused. Content rules (reject_rules) are not covered. A runtime override from /_rate_limit_toggle wins over this value. Accepts true/false (also yes/no, on/off, 1/0); anything else is logged and the default stands." },
   { hidden = true, key = "rt_limit_factor", label = "Runtime limit factor", group = "Capacity & failover", kind = "number",
     detail = "Scales the pool limit at which requests are turned away for concurrency: a request is refused once in-flight requests reach limit x factor. Must be over 0; anything else is logged and the global default stands." },