From f5ec776ae76348e4d611584abbe3c0cd1fed5af5 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 00:30:22 -0500 Subject: [PATCH 1/6] test: Run the praxis-mode end-to-end checks through cortex-gateway Add cmd/cortex/praxis_e2e_test.go (build tag praxis_e2e). Each test starts the real cortex-gateway binary that CORTEX_GATEWAY_BIN names, and wires praxis mode with startPraxis, as main does. A test dialer sends each origin address to one mock origin. A direct-mode twin with inference-parser gives the rows to compare. The tests cover the W1 items 1, 3, 5, 7, 8 and 9, the Codex mislabeled SSE endpoint, Accept-Encoding, the hop headers, a request with no token to praxis.listen, and the rows of both modes. The rows can differ only in invocations, which praxis/README.md (gateway decision 6) gives, and the test pins that the gateway rows have none. The 11 MiB response streams as SSE. A buffered response over 10 MiB is a 502 in direct mode too. The Praxis workflow builds cortex-gateway and runs these tests after cargo test, with the same blocking rule as the other steps. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .github/workflows/praxis.yaml | 24 + cmd/cortex/praxis_e2e_test.go | 803 ++++++++++++++++++++++++++++++++++ 2 files changed, 827 insertions(+) create mode 100644 cmd/cortex/praxis_e2e_test.go diff --git a/.github/workflows/praxis.yaml b/.github/workflows/praxis.yaml index 8acb30a4d..e758ae1e9 100644 --- a/.github/workflows/praxis.yaml +++ b/.github/workflows/praxis.yaml @@ -31,6 +31,10 @@ on: - "core/redact/**" - "core/pipeline/subagent.go" - "core/listener/forwardproxy/ingest_replay_test.go" + - "core/listener/forwardproxy/hop*.go" + - "core/listener/forwardproxy/egress*.go" + - "cmd/cortex/praxis*.go" + - "cmd/cortex/gateway.go" push: branches: - main @@ -54,6 +58,10 @@ on: - "core/redact/**" - "core/pipeline/subagent.go" - "core/listener/forwardproxy/ingest_replay_test.go" + - "core/listener/forwardproxy/hop*.go" + - "core/listener/forwardproxy/egress*.go" + - "cmd/cortex/praxis*.go" + - "cmd/cortex/gateway.go" permissions: contents: read @@ -116,6 +124,20 @@ jobs: continue-on-error: ${{ steps.changes.outputs.blocking != 'true' }} run: cargo test --locked + - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: cmd/cortex/go.mod + + # Runs from the repo root: the command names praxis/ and cmd/cortex. + - name: End-to-end (praxis mode) + id: e2e + continue-on-error: ${{ steps.changes.outputs.blocking != 'true' }} + run: | + set -euo pipefail + cargo build --locked -p cortex-gateway --manifest-path praxis/Cargo.toml + TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) + CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E . + # A report-only run passes, so a warning on the PR shows the parity gap. - name: Report failures that do not block if: always() && steps.changes.outputs.blocking != 'true' @@ -123,10 +145,12 @@ jobs: FMT: ${{ steps.fmt.outcome }} CLIPPY: ${{ steps.clippy.outcome }} TEST: ${{ steps.test.outcome }} + E2E: ${{ steps.e2e.outcome }} run: | [ "$FMT" != failure ] || echo "::warning::cargo fmt failed (does not block: no praxis/ changes)" [ "$CLIPPY" != failure ] || echo "::warning::cargo clippy failed (does not block: no praxis/ changes)" [ "$TEST" != failure ] || echo "::warning::cargo test failed (does not block: no praxis/ changes)" + [ "$E2E" != failure ] || echo "::warning::praxis end-to-end tests failed (does not block: no praxis/ changes)" # Size cap: the cache post-step must not save a target/ over 3 GiB. - name: Cap the target/ cache size diff --git a/cmd/cortex/praxis_e2e_test.go b/cmd/cortex/praxis_e2e_test.go new file mode 100644 index 000000000..933e77b31 --- /dev/null +++ b/cmd/cortex/praxis_e2e_test.go @@ -0,0 +1,803 @@ +//go:build praxis_e2e + +package main + +// End-to-end tests of praxis mode through the real cortex-gateway binary (#1721, plan +// Task 8). Each test starts the binary that CORTEX_GATEWAY_BIN names. A test dialer +// sends every origin address to one mock origin, so names such as chatgpt.com never +// reach DNS. + +import ( + "bufio" + "bytes" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "os/exec" + "reflect" + "strings" + "sync" + "testing" + "time" + + "github.com/rossoctl/cortex/core/config" + "github.com/rossoctl/cortex/core/cost/pricing" + "github.com/rossoctl/cortex/core/listener/forwardproxy" + "github.com/rossoctl/cortex/core/model" + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/plugins" + "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/sessionapi" + "github.com/rossoctl/cortex/core/tlsbridge" +) + +const e2eSession = "00000000-0000-4000-8000-0000000000e2" + +// allHopHeaders are the twelve hop headers (praxis/filters/src/record/attribution.rs HOP_HEADERS). +var allHopHeaders = []string{ + "X-Cortex-Agent", "X-Cortex-Proc", "X-Cortex-Client-Proc", "X-Cortex-Session", "X-Cortex-Source", + "X-Cortex-Session-Header", "X-Cortex-Chain-Session", "X-Cortex-Attribution", "X-Cortex-Request-Id", + "X-Cortex-Probe", "X-Cortex-Dest", "X-Cortex-Token", +} + +// mockOrigin serves the handler a test sets and keeps each request's headers. +type mockOrigin struct { + *httptest.Server + mu sync.Mutex + handler http.HandlerFunc + headers []http.Header +} + +func newMockOrigin(t *testing.T) *mockOrigin { + o := &mockOrigin{} + o.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + o.mu.Lock() + o.headers = append(o.headers, r.Header.Clone()) + h := o.handler + o.mu.Unlock() + if h == nil { + w.WriteHeader(http.StatusOK) + return + } + h(w, r) + })) + t.Cleanup(o.Close) + return o +} + +func (o *mockOrigin) serve(h http.HandlerFunc) { + o.mu.Lock() + o.handler = h + o.mu.Unlock() +} + +func (o *mockOrigin) lastHeader(t *testing.T) http.Header { + t.Helper() + o.mu.Lock() + defer o.mu.Unlock() + if len(o.headers) == 0 { + t.Fatal("the origin got no request") + } + return o.headers[len(o.headers)-1] +} + +// dial sends every address to the origin (Decision 15). +func (o *mockOrigin) dial(ctx context.Context, network, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, o.Listener.Addr().String()) +} + +// e2eRates prices every host and model at one rate, for both chains. +func e2eRates(t *testing.T) *pricing.Registry { + t.Helper() + var r pricing.Rates + for tier, perToken := range map[pricing.Tier]float64{ + pricing.TierInput: 7e-6, pricing.TierCacheWrite: 11e-6, pricing.TierCacheRead: 3e-6, pricing.TierOutput: 23e-6, + } { + r.Base[tier], r.Set[tier] = perToken, true + } + tab, err := pricing.NewTable([]pricing.Entry{{Host: "*", Model: "*", Rates: r, Prov: pricing.ProvConfigured}}) + if err != nil { + t.Fatal(err) + } + return pricing.NewRegistry(tab) +} + +// chain is a forward proxy and its session store, in praxis mode or in direct mode. +type chain struct { + origin *mockOrigin + store *session.Store + proxy string // the forward proxy, host:port + client *http.Client + cfg *config.Config + hop *forwardproxy.Hop // nil in direct mode + // startGateway starts cortex-gateway. newPraxisChain calls it unless noGateway is set. + startGateway func() +} + +type chainOpts struct { + noGateway bool +} + +// e2eConfig has inference-parser on the outbound chain. In praxis mode outboundEntries +// builds none of it, as in main. +func e2eConfig(praxis bool) *config.Config { + cfg := &config.Config{} + cfg.Pipeline.Outbound.Plugins = []config.PluginEntry{{Name: "inference-parser"}} + if praxis { + cfg.Praxis = &config.PraxisConfig{Enabled: true} + } + return cfg +} + +// newForwardProxy builds the forward proxy as main does, for cfg. +func newForwardProxy(t *testing.T, cfg *config.Config, store *session.Store, rates *pricing.Registry) *forwardproxy.Server { + t.Helper() + p, err := plugins.BuildWithDeps(outboundEntries(cfg, slog.Default()), + plugins.Deps{Pricing: rates, Listener: forwardproxy.Support(false), LocalInstall: true}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { p.Stop(context.Background()) }) + fp, err := forwardproxy.NewServer(pipeline.NewHolder(p), store, nil) + if err != nil { + t.Fatal(err) + } + fp.ResponsesWebSockets = cfg.Listener.ResponsesWebSocketsMode() + fp.SessionIDHeaders = cfg.Session.SessionIDHeadersFor(true) + fp.ClientAffinity = cfg.Session.ClientAffinityEnabled() + return fp +} + +func (c *chain) serveProxy(t *testing.T, fp *forwardproxy.Server) { + t.Helper() + ts := httptest.NewUnstartedServer(fp.Handler()) + ts.Config.ConnContext = fp.ConnContext + ts.Start() + t.Cleanup(ts.Close) + c.proxy = ts.Listener.Addr().String() + c.client = &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(&url.URL{Scheme: "http", Host: c.proxy})}} +} + +func newStore(t *testing.T) *session.Store { + store := session.New(time.Hour, 0, 0) + t.Cleanup(store.Close) + return store +} + +// newDirectChain is the direct-mode twin: the same origin and dialer, praxis off. +func newDirectChain(t *testing.T, origin *mockOrigin) *chain { + t.Helper() + c := &chain{origin: origin, store: newStore(t), cfg: e2eConfig(false)} + fp := newForwardProxy(t, c.cfg, c.store, e2eRates(t)) + fp.Client.Transport.(*http.Transport).DialContext = origin.dial + c.serveProxy(t, fp) + return c +} + +// newPraxisChain wires praxis mode with startPraxis, as main does, and starts the gateway. +func newPraxisChain(t *testing.T, origin *mockOrigin, opts chainOpts) *chain { + t.Helper() + bin := os.Getenv("CORTEX_GATEWAY_BIN") + if st, err := os.Stat(bin); bin == "" || err != nil || st.IsDir() { + t.Skip("CORTEX_GATEWAY_BIN does not name the cortex-gateway binary") + } + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + c := &chain{origin: origin, store: newStore(t), cfg: e2eConfig(true)} + apiLn, egressLn := listen(t), listen(t) + c.cfg.Listener.SessionAPIAddr = apiLn.Addr().String() + c.cfg.Praxis.Listen = freeAddr(t) + c.cfg.Praxis.Egress = egressLn.Addr().String() + + dir := t.TempDir() + tokenPath, yamlPath, err := writeGatewayFiles(dir, c.cfg) + if err != nil { + t.Fatal(err) + } + token, err := readGatewayToken(func(string) string { return tokenPath }, os.ReadFile) + if err != nil { + t.Fatal(err) + } + + rates := e2eRates(t) + fp := newForwardProxy(t, c.cfg, c.store, rates) + up, err := tlsbridge.NewUpstreamClient(nil, false, tlsbridge.WithDialContext(origin.dial)) + if err != nil { + t.Fatal(err) + } + hop, egressSrv := startPraxis(ctx, c.cfg, fp, token, up.Transport, egressLn) + t.Cleanup(func() { egressSrv.Close() }) + c.hop = hop + + api := sessionapi.New("", c.store, ingestOptions(c.cfg, rates, token, hop)...) + go func() { _ = api.Server().Serve(apiLn) }() + t.Cleanup(func() { api.Server().Close() }) + c.serveProxy(t, fp) + + c.startGateway = func() { + cmd := exec.Command(bin, "-c", yamlPath, "--exit-on-stdin-eof") //nolint:gosec // the test's own binary + cmd.Env = []string{} + var out bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &out + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + stdin.Close() + done := make(chan struct{}) + go func() { _ = cmd.Wait(); close(done) }() + select { + case <-done: + case <-time.After(10 * time.Second): + _ = cmd.Process.Kill() + <-done + } + if t.Failed() { + t.Logf("cortex-gateway output:\n%s", out.String()) + } + }) + } + if !opts.noGateway { + c.startGateway() + c.waitReady(t, 20*time.Second) + } + return c +} + +func (c *chain) waitReady(t *testing.T, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for !c.hop.Ready() { + if time.Now().After(deadline) { + t.Fatalf("the hop is not ready after %s", d) + } + time.Sleep(20 * time.Millisecond) + } +} + +// request makes a request with the e2e session header. +func request(t *testing.T, ctx context.Context, method, rawURL string, body []byte) *http.Request { + t.Helper() + req, err := http.NewRequestWithContext(ctx, method, rawURL, bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set(session.ClaudeCodeSessionHeader, e2eSession) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + return req +} + +func (c *chain) do(t *testing.T, req *http.Request) (int, []byte, http.Header) { + t.Helper() + resp, err := c.client.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + b, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return resp.StatusCode, b, resp.Header +} + +func responseCount(evs []model.SessionEvent) int { + n := 0 + for _, ev := range evs { + if ev.Phase == pipeline.SessionResponse { + n++ + } + } + return n +} + +// waitResponses waits until the session holds n response rows, and returns its rows. +func (c *chain) waitResponses(t *testing.T, n int, d time.Duration) []model.SessionEvent { + t.Helper() + deadline := time.Now().Add(d) + for { + var evs []model.SessionEvent + if v := c.store.View(e2eSession); v != nil { + evs = v.Events + } + if responseCount(evs) >= n { + return evs + } + if time.Now().After(deadline) { + t.Fatalf("the session has %d response rows after %s, want %d: %+v", responseCount(evs), d, n, evs) + } + time.Sleep(20 * time.Millisecond) + } +} + +// lastResponse is the newest response row. +func lastResponse(t *testing.T, evs []model.SessionEvent) model.SessionEvent { + t.Helper() + for i := len(evs) - 1; i >= 0; i-- { + if evs[i].Phase == pipeline.SessionResponse { + return evs[i] + } + } + t.Fatal("no response row") + return model.SessionEvent{} +} + +const chatURL = "http://api.openai.com/v1/chat/completions" + +var chatRequest = []byte(`{"model":"gpt-4o","messages":[{"role":"user","content":"hello"}]}`) + +const chatJSON = `{"id":"c1","object":"chat.completion","model":"gpt-4o","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":11,"completion_tokens":5,"total_tokens":16}}` + +var chatStreamRequest = []byte(`{"model":"gpt-4o","stream":true,"stream_options":{"include_usage":true},"messages":[{"role":"user","content":"hello"}]}`) + +var chatFrames = []string{ + `data: {"id":"c2","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{"role":"assistant","content":"hi"}}]}` + "\n\n", + `data: {"id":"c2","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}` + "\n\n", + `data: {"id":"c2","object":"chat.completion.chunk","model":"gpt-4o","choices":[],"usage":{"prompt_tokens":11,"completion_tokens":5,"total_tokens":16}}` + "\n\n", + "data: [DONE]\n\n", +} + +func serveJSON(body string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, body) + } +} + +// serveFrames writes frames[0], then waits for release (when not nil), then the rest. +// gz sends them gzip-encoded, with a flush after each frame. +func serveFrames(contentType string, gz bool, frames []string, release <-chan struct{}, pause time.Duration) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", contentType) + var out io.Writer = w + var zw *gzip.Writer + if gz { + w.Header().Set("Content-Encoding", "gzip") + zw = gzip.NewWriter(w) + out = zw + } + w.WriteHeader(http.StatusOK) + flush := func() { + if zw != nil { + _ = zw.Flush() + } + w.(http.Flusher).Flush() + } + for i, f := range frames { + if i == 1 { + if release != nil { + select { + case <-release: + case <-r.Context().Done(): + return + } + } + time.Sleep(pause) + } + _, _ = io.WriteString(out, f) + flush() + } + if zw != nil { + _ = zw.Close() + } + } +} + +// readFrame reads one SSE frame (to the blank line), or fails after 10 s. +func readFrame(t *testing.T, br *bufio.Reader) string { + t.Helper() + got := make(chan string, 1) + go func() { + var b strings.Builder + for { + line, err := br.ReadString('\n') + b.WriteString(line) + if err != nil || line == "\n" { + got <- b.String() + return + } + } + }() + select { + case s := <-got: + return s + case <-time.After(10 * time.Second): + t.Fatal("no frame in 10s: the response is buffered") + return "" + } +} + +// W1 1: the response row's host and cost equal the direct run's. +func TestPraxisE2E_HostAndCostMatchDirect(t *testing.T) { + origin := newMockOrigin(t) + origin.serve(serveJSON(chatJSON)) + var rows [2]model.SessionEvent + for i, c := range []*chain{newDirectChain(t, origin), newPraxisChain(t, origin, chainOpts{})} { + if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatRequest)); st != http.StatusOK { + t.Fatalf("status %d: %s", st, b) + } + rows[i] = lastResponse(t, c.waitResponses(t, 1, 10*time.Second)) + } + direct, praxis := rows[0], rows[1] + if praxis.Host != direct.Host || praxis.Host == "" { + t.Errorf("host = %q, want the direct run's %q", praxis.Host, direct.Host) + } + if direct.Plugins["cost"] == nil { + t.Fatalf("the direct run has no cost: %+v", direct.Plugins) + } + if !jsonEqual(praxis.Plugins["cost"], direct.Plugins["cost"]) { + t.Errorf("cost = %s, want the direct run's %s", praxis.Plugins["cost"], direct.Plugins["cost"]) + } +} + +func jsonEqual(a, b []byte) bool { + var x, y any + if json.Unmarshal(a, &x) != nil || json.Unmarshal(b, &y) != nil { + return false + } + return reflect.DeepEqual(x, y) +} + +// streamFirstFrame sends req, reads frame 1 before it releases the origin, and returns +// the whole body. +func streamFirstFrame(t *testing.T, c *chain, req *http.Request, release chan struct{}) string { + t.Helper() + resp, err := c.client.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + b, _ := io.ReadAll(resp.Body) + t.Fatalf("status %d: %s", resp.StatusCode, b) + } + br := bufio.NewReader(resp.Body) + first := readFrame(t, br) + close(release) + rest, err := io.ReadAll(br) + if err != nil { + t.Fatal(err) + } + return first + string(rest) +} + +// W1 3: a gzip SSE response arrives frame by frame. +func TestPraxisE2E_GzipSSEArrivesAsFrames(t *testing.T) { + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + release := make(chan struct{}) + origin.serve(serveFrames("text/event-stream", true, chatFrames, release, 0)) + got := streamFirstFrame(t, c, request(t, t.Context(), http.MethodPost, chatURL, chatStreamRequest), release) + if want := strings.Join(chatFrames, ""); got != want { + t.Errorf("body = %q, want %q", got, want) + } +} + +// W1 5: a client that hangs up mid-stream gives client_canceled, and the origin sees it. +func TestPraxisE2E_HangUpIsClientCanceled(t *testing.T) { + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + originDone := make(chan struct{}) + origin.serve(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, chatFrames[0]) + w.(http.Flusher).Flush() + <-r.Context().Done() + close(originDone) + }) + ctx, cancel := context.WithCancel(t.Context()) + resp, err := c.client.Do(request(t, ctx, http.MethodPost, chatURL, chatStreamRequest)) + if err != nil { + t.Fatal(err) + } + readFrame(t, bufio.NewReader(resp.Body)) + cancel() + resp.Body.Close() + select { + case <-originDone: + case <-time.After(5 * time.Second): + t.Fatal("the origin's request context did not end within 5s") + } + row := lastResponse(t, c.waitResponses(t, 1, 10*time.Second)) + if row.Error == nil || row.Error.Kind != "client_canceled" { + t.Errorf("error = %+v, want client_canceled", row.Error) + } +} + +// W1 7: with no gateway, a request gets 503 gateway_unavailable. Once the gateway runs, 200. +func TestPraxisE2E_FailClosedAtStartup(t *testing.T) { + origin := newMockOrigin(t) + origin.serve(serveJSON(chatJSON)) + c := newPraxisChain(t, origin, chainOpts{noGateway: true}) + if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatRequest)); st != http.StatusServiceUnavailable { + t.Fatalf("status = %d (%s), want 503", st, b) + } + row := lastResponse(t, c.waitResponses(t, 1, 5*time.Second)) + if row.Error == nil || row.Error.Kind != model.ErrorKindGatewayUnavailable { + t.Errorf("error = %+v, want %s", row.Error, model.ErrorKindGatewayUnavailable) + } + c.startGateway() + deadline := time.Now().Add(10 * time.Second) + for { + st, _, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatRequest)) + if st == http.StatusOK { + return + } + if time.Now().After(deadline) { + t.Fatalf("status = %d 10s after the gateway started, want 200", st) + } + time.Sleep(200 * time.Millisecond) + } +} + +// W1 8: 65 s of silence mid-stream does not cut the stream. +func TestPraxisE2E_SilenceMidStream(t *testing.T) { + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + origin.serve(serveFrames("text/event-stream", false, chatFrames, nil, 65*time.Second)) + st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatStreamRequest)) + if want := strings.Join(chatFrames, ""); st != http.StatusOK || string(b) != want { + t.Fatalf("got %d %q, want 200 %q", st, b, want) + } + if row := lastResponse(t, c.waitResponses(t, 1, 10*time.Second)); row.Error != nil { + t.Errorf("error = %+v, want none", row.Error) + } +} + +// W1 9: an 11 MiB body each way passes, and the row counts both. The response streams +// as SSE frames of 64 KiB: a buffered response over 10 MiB is a 502 in direct mode too. +func TestPraxisE2E_OverTenMiBEachWay(t *testing.T) { + const size = 11 << 20 + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + got := make(chan int64, 1) + origin.serve(func(w http.ResponseWriter, r *http.Request) { + n, _ := io.Copy(io.Discard, r.Body) + got <- n + w.Header().Set("Content-Type", "text/event-stream") + chunk := []byte("data: " + strings.Repeat("x", 64<<10-8) + "\n\n") + for sent := 0; sent < size; sent += len(chunk) { + _, _ = w.Write(chunk) + w.(http.Flusher).Flush() + } + }) + req := request(t, t.Context(), http.MethodPost, "http://files.example.com/upload", bytes.Repeat([]byte("u"), size)) + req.Header.Set("Content-Type", "application/octet-stream") + st, b, _ := c.do(t, req) + if st != http.StatusOK || len(b) != size { + t.Fatalf("got %d with %d bytes, want 200 with %d", st, len(b), size) + } + if n := <-got; n != size { + t.Errorf("the origin read %d bytes, want %d", n, size) + } + // One direction per row (docs/session-api.md): bytesUp on the request row, bytesDown on the response row. + evs := c.waitResponses(t, 1, 10*time.Second) + var up int64 + for _, ev := range evs { + if ev.Phase == pipeline.SessionRequest { + up = ev.BytesUp + } + } + if down := lastResponse(t, evs).BytesDown; up != size || down != size { + t.Errorf("bytesUp, bytesDown = %d, %d, want %d each", up, down, size) + } +} + +const codexURL = "http://chatgpt.com/backend-api/codex/responses" + +var codexRequest = []byte(`{"model":"gpt-5","stream":true,"input":[{"role":"user","content":"hello"}]}`) + +// Codex's endpoint labels its SSE as application/json: it still streams. A 405 keeps its body. +func TestPraxisE2E_CodexMislabeledSSE(t *testing.T) { + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + frames := []string{ + `data: {"type":"response.created","response":{"id":"r1","model":"gpt-5"}}` + "\n\n", + `data: {"type":"response.completed","response":{"id":"r1","model":"gpt-5","usage":{"input_tokens":11,"output_tokens":5,"total_tokens":16}}}` + "\n\n", + } + release := make(chan struct{}) + origin.serve(serveFrames("application/json", false, frames, release, 0)) + if got := streamFirstFrame(t, c, request(t, t.Context(), http.MethodPost, codexURL, codexRequest), release); got != strings.Join(frames, "") { + t.Errorf("body = %q", got) + } + c.waitResponses(t, 1, 10*time.Second) + + const detail = `{"detail":"Method Not Allowed"}` + origin.serve(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusMethodNotAllowed) + _, _ = io.WriteString(w, detail) + }) + st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, codexURL, codexRequest)) + if st != http.StatusMethodNotAllowed || string(b) != detail { + t.Fatalf("got %d %q, want 405 %q", st, b, detail) + } + row := lastResponse(t, c.waitResponses(t, 2, 10*time.Second)) + if row.StatusCode != http.StatusMethodNotAllowed || row.Error == nil || row.Error.Kind != "backend_error" || row.Error.Code != "405" { + t.Errorf("row = %d %+v, want 405 backend_error/405", row.StatusCode, row.Error) + } +} + +// Cortex owns the encoding: the origin sees only gzip, and the client gets plain JSON. +func TestPraxisE2E_AcceptEncodingDecoded(t *testing.T) { + origin := newMockOrigin(t) + c := newPraxisChain(t, origin, chainOpts{}) + origin.serve(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Encoding", "gzip") + zw := gzip.NewWriter(w) + _, _ = io.WriteString(zw, chatJSON) + _ = zw.Close() + }) + req := request(t, t.Context(), http.MethodPost, chatURL, chatRequest) + req.Header.Set("Accept-Encoding", "gzip, deflate, br") + st, b, h := c.do(t, req) + if st != http.StatusOK || string(b) != chatJSON || h.Get("Content-Encoding") != "" { + t.Fatalf("got %d %q (Content-Encoding %q), want 200 and the plain JSON", st, b, h.Get("Content-Encoding")) + } + if ae := origin.lastHeader(t).Get("Accept-Encoding"); ae != "gzip" { + t.Errorf("the origin saw Accept-Encoding %q, want gzip", ae) + } + row := lastResponse(t, c.waitResponses(t, 1, 10*time.Second)) + if row.Inference == nil || row.Inference.InputTokens != 11 || row.Inference.OutputTokens != 5 { + t.Errorf("inference = %+v, want 11 input and 5 output tokens", row.Inference) + } +} + +// No hop header reaches the origin, whoever sent it. Other X-Cortex-* headers pass. +func TestPraxisE2E_NoHopHeaderReachesOrigin(t *testing.T) { + origin := newMockOrigin(t) + origin.serve(serveJSON(chatJSON)) + c := newPraxisChain(t, origin, chainOpts{}) + req := request(t, t.Context(), http.MethodPost, chatURL, chatRequest) + for _, k := range []string{"X-Cortex-Attribution", "X-Cortex-Dest", "X-Cortex-Token", "X-Cortex-Request-Id", "X-Cortex-Probe"} { + req.Header.Set(k, "from-the-client") + } + req.Header.Set("X-Cortex-Custom", "keep") + if st, b, _ := c.do(t, req); st != http.StatusOK { + t.Fatalf("status %d: %s", st, b) + } + h := origin.lastHeader(t) + for _, k := range allHopHeaders { + if v := h.Values(k); len(v) > 0 { + t.Errorf("the origin saw %s: %q", k, v) + } + } + if h.Get("X-Cortex-Custom") != "keep" { + t.Errorf("the origin did not see X-Cortex-Custom: %v", h) + } +} + +// A request straight to praxis.listen with no token gets 403 and records nothing. +func TestPraxisE2E_TokenlessRequestTo47605(t *testing.T) { + origin := newMockOrigin(t) + origin.serve(serveJSON(chatJSON)) + c := newPraxisChain(t, origin, chainOpts{}) + before := len(c.store.ListSessions()) + req := request(t, t.Context(), http.MethodPost, "http://"+c.cfg.Praxis.ListenAddr()+"/v1/chat/completions", chatRequest) + req.Host = "api.openai.com" + resp, err := http.DefaultTransport.RoundTrip(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + t.Errorf("status = %d, want 403", resp.StatusCode) + } + time.Sleep(time.Second) // a row, if one were sent, lands well within this + if after := len(c.store.ListSessions()); after != before || c.store.View(e2eSession) != nil { + t.Errorf("sessions went from %d to %d; e2e session = %+v", before, after, c.store.View(e2eSession)) + } +} + +// eventsJSON reads the session's rows from the session API's JSON form. +func eventsJSON(t *testing.T, c *chain) []map[string]any { + t.Helper() + srv := httptest.NewServer(sessionapi.New("", c.store).Server().Handler) + defer srv.Close() + resp, err := http.Get(srv.URL + "/v1/sessions/" + e2eSession) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var v struct { + Events []map[string]any `json:"events"` + } + if err := json.NewDecoder(resp.Body).Decode(&v); err != nil { + t.Fatal(err) + } + return v.Events +} + +// The JSON and SSE requests record the same rows in both modes. +func TestPraxisE2E_SameEventsAsDirect(t *testing.T) { + origin := newMockOrigin(t) + var got [2][]map[string]any + for i, c := range []*chain{newDirectChain(t, origin), newPraxisChain(t, origin, chainOpts{})} { + origin.serve(serveJSON(chatJSON)) + if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatRequest)); st != http.StatusOK { + t.Fatalf("status %d: %s", st, b) + } + c.waitResponses(t, 1, 10*time.Second) + origin.serve(serveFrames("text/event-stream", false, chatFrames, nil, 0)) + if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatStreamRequest)); st != http.StatusOK { + t.Fatalf("status %d: %s", st, b) + } + c.waitResponses(t, 2, 10*time.Second) + got[i] = eventsJSON(t, c) + } + direct, praxis := got[0], got[1] + if len(direct) != len(praxis) { + t.Fatalf("direct has %d rows, praxis %d:\ndirect %v\npraxis %v", len(direct), len(praxis), direct, praxis) + } + for i := range direct { + d, p := direct[i], praxis[i] + for _, k := range []string{"seq", "at", "sessionId", "requestId"} { + delete(d, k) + delete(p, k) + } + for _, k := range []string{"ttfbMs", "durationMs"} { + dv, _ := d[k].(float64) + pv, _ := p[k].(float64) + if dv-pv > 50 || pv-dv > 50 { + t.Errorf("row %d: %s = %v, direct %v (over 50 ms apart)", i, k, pv, dv) + } + delete(d, k) + delete(p, k) + } + // Known difference, praxis/README.md "Gateway decisions" 6: the gateway runs no + // Cortex plugin, so its rows have no invocations. + if p["invocations"] != nil { + t.Errorf("row %d: praxis invocations = %s, want none", i, compact(p["invocations"])) + } + delete(d, "invocations") + for _, k := range unionKeys(d, p) { + if !reflect.DeepEqual(d[k], p[k]) { + t.Errorf("row %d (%v): %s = %s, direct %s", i, d["phase"], k, compact(p[k]), compact(d[k])) + } + } + } +} + +func unionKeys(a, b map[string]any) []string { + var keys []string + for k := range a { + keys = append(keys, k) + } + for k := range b { + if _, ok := a[k]; !ok { + keys = append(keys, k) + } + } + return keys +} + +func compact(v any) string { + if v == nil { + return "" + } + b, err := json.Marshal(v) + if err != nil { + return fmt.Sprint(v) + } + return string(b) +} From bf2a33c4f1aaa0c6fd7131bdb9cb038016382d9d Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 00:34:55 -0500 Subject: [PATCH 2/6] fix: Use the praxis toolchain in the end-to-end CI step The step runs from the repo root. rustup reads rust-toolchain.toml only below praxis/, so the step used the default toolchain, which can be older than the 1.96 that the crates need. The step now reads the channel from praxis/rust-toolchain.toml and sets RUSTUP_TOOLCHAIN. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .github/workflows/praxis.yaml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/praxis.yaml b/.github/workflows/praxis.yaml index e758ae1e9..95ea2b95c 100644 --- a/.github/workflows/praxis.yaml +++ b/.github/workflows/praxis.yaml @@ -128,12 +128,15 @@ jobs: with: go-version-file: cmd/cortex/go.mod - # Runs from the repo root: the command names praxis/ and cmd/cortex. + # Runs from the repo root: the command names praxis/ and cmd/cortex. rustup reads + # rust-toolchain.toml only below praxis/, so the step names that toolchain itself. - name: End-to-end (praxis mode) id: e2e continue-on-error: ${{ steps.changes.outputs.blocking != 'true' }} run: | set -euo pipefail + RUSTUP_TOOLCHAIN=$(sed -n 's/^channel = "\(.*\)"$/\1/p' praxis/rust-toolchain.toml) + export RUSTUP_TOOLCHAIN cargo build --locked -p cortex-gateway --manifest-path praxis/Cargo.toml TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E . From 50acfde2c3f28610a5d68eb554f78aab83436813 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 00:46:59 -0500 Subject: [PATCH 3/6] docs: Fix the praxis plan gate and the Codex 405 row text cargo fmt --check with --manifest-path stops with "Failed to find targets" on the praxis workspace. The gate now uses --all. The Codex 405 row keeps only a provider error type, not the detail text, so the plan now says what the row holds. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .../plans/2026-11-09-praxis-mode-4-hop-and-egress.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md b/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md index 4cdb8d9c6..e5ccb11c4 100644 --- a/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md +++ b/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md @@ -405,7 +405,7 @@ Depends on: 1, 7. Estimated lines: 850. - `TestPraxisE2E_FailClosedAtStartup` (W1 7): start the chain without the gateway. A request gets 503 and a `gateway_unavailable` row. Start the gateway. Within 10 s a request gets 200. - `TestPraxisE2E_SilenceMidStream` (W1 8): the origin sends one frame, is silent for 65 s, then sends the rest. The client gets every frame and the row has no error. - `TestPraxisE2E_OverTenMiBEachWay` (W1 9): an 11 MiB request body (not inference) and an 11 MiB streamed response. `bytesUp` and `bytesDown` are the sizes. - - `TestPraxisE2E_CodexMislabeledSSE`: host `chatgpt.com`, path `/backend-api/codex/responses`, `Content-Type: application/json`, 2xx SSE body. The client reads frame 1 before frame 2 is released. A 405 with `{"detail":"Method Not Allowed"}`: the client gets that body, and the row carries the error detail. + - `TestPraxisE2E_CodexMislabeledSSE`: host `chatgpt.com`, path `/backend-api/codex/responses`, `Content-Type: application/json`, 2xx SSE body. The client reads frame 1 before frame 2 is released. A 405 with `{"detail":"Method Not Allowed"}`: the client gets that body, and the row is `backend_error` with code `405`. The row's `message` stays empty: `upstreamErrorKind` keeps only a provider error type, and this body has none. - `TestPraxisE2E_AcceptEncodingDecoded`: the client sends `Accept-Encoding: gzip, deflate, br`. The origin sees only `gzip` and answers gzip JSON. The client gets plain JSON, and the row has the tokens. - `TestPraxisE2E_NoHopHeaderReachesOrigin`: the client sends the five hop headers and `X-Cortex-Custom`. The origin sees none of the twelve hop headers and sees `X-Cortex-Custom`. - `TestPraxisE2E_TokenlessRequestTo47605`: a request straight to `praxis.listen` with no token gets 403. `/v1/sessions` shows no new row. @@ -423,7 +423,7 @@ Depends on: 1, 7. Estimated lines: 850. From the worktree root: ```sh -test -z "$(gofmt -l core cmd/cortex)" && GOWORK=off go -C core vet ./... && GOWORK=off go -C core test -race ./attribution/... ./listener/... ./sessionapi/... ./config/... ./model/... ./pipeline/... && TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) && GOWORK=off go -C cmd/cortex vet -tags "$TAGS" ./... && GOWORK=off go -C cmd/cortex test -tags "$TAGS" ./... && cargo fmt --check --manifest-path praxis/Cargo.toml && cargo clippy --manifest-path praxis/Cargo.toml --all-targets --locked -- -D warnings && cargo test --manifest-path praxis/Cargo.toml --locked && cargo build --locked -p cortex-gateway --manifest-path praxis/Cargo.toml && CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E . +test -z "$(gofmt -l core cmd/cortex)" && GOWORK=off go -C core vet ./... && GOWORK=off go -C core test -race ./attribution/... ./listener/... ./sessionapi/... ./config/... ./model/... ./pipeline/... && TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) && GOWORK=off go -C cmd/cortex vet -tags "$TAGS" ./... && GOWORK=off go -C cmd/cortex test -tags "$TAGS" ./... && cargo fmt --all --check --manifest-path praxis/Cargo.toml && cargo clippy --manifest-path praxis/Cargo.toml --all-targets --locked -- -D warnings && cargo test --manifest-path praxis/Cargo.toml --locked && cargo build --locked -p cortex-gateway --manifest-path praxis/Cargo.toml && CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E . ``` Before P1 and P2 there is no `praxis_e2e` file, so the last step runs no test and passes. From 33bfcc9eeb279527db4b922aa20b0b25db2a7799 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 07:14:49 -0500 Subject: [PATCH 4/6] fix: Make the praxis end-to-end checks block and compile on Go changes - ci.yaml: vet cmd/cortex with the praxis_e2e tag in the local-profile step. Before, only praxis.yaml compiled the end-to-end file. A rename in a helper that the file calls (for example readGatewayToken) did not fail any check. - praxis.yaml: add cmd/cortex/ingest.go to the triggers. A failure now blocks the PR also when the PR changes the Go side of praxis mode: the hop, the egress listener, and the praxis, gateway and ingest files in cmd/cortex. Before, a PR that broke the hop got only a warning. - SameEventsAsDirect: fail when ttfbMs or durationMs is on one side only. Before, a missing key read as 0 and passed the 50 ms check. The origin now waits 5 ms, because both fields are omitempty and a local mock answers in less than 1 ms. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .github/workflows/ci.yaml | 1 + .github/workflows/praxis.yaml | 21 +++++++++++++-------- cmd/cortex/praxis_e2e_test.go | 18 +++++++++++++----- 3 files changed, 27 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index e453a97cb..b650f05d4 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -106,6 +106,7 @@ jobs: run: | TAGS=$(go -C ../../scripts/profile-tags run . local) go build -tags "${TAGS}" ./... + go vet -tags "praxis_e2e,${TAGS}" . go test -race -tags "${TAGS}" ./... - name: Build and test cmd/agentop diff --git a/.github/workflows/praxis.yaml b/.github/workflows/praxis.yaml index 95ea2b95c..db81c2fe9 100644 --- a/.github/workflows/praxis.yaml +++ b/.github/workflows/praxis.yaml @@ -1,8 +1,9 @@ # Praxis - fmt, clippy and tests for the Rust workspace in praxis/ # # Runs when praxis/ changes, or when the Go code that praxis/ ports changes. -# Only a change to praxis/ (or to this file) makes a failure block the PR. -# A Go-only change runs the steps too, so drift shows, but does not block. +# A failure blocks the PR only for a change to praxis/, to this file, or to the Go side +# of praxis mode (the hop, the egress listener, cmd/cortex's praxis wiring). Any other +# Go change runs the steps too, so drift shows, but does not block. # name: Praxis @@ -35,6 +36,7 @@ on: - "core/listener/forwardproxy/egress*.go" - "cmd/cortex/praxis*.go" - "cmd/cortex/gateway.go" + - "cmd/cortex/ingest.go" push: branches: - main @@ -62,6 +64,7 @@ on: - "core/listener/forwardproxy/egress*.go" - "cmd/cortex/praxis*.go" - "cmd/cortex/gateway.go" + - "cmd/cortex/ingest.go" permissions: contents: read @@ -88,12 +91,14 @@ jobs: echo "blocking=true" >> "$GITHUB_OUTPUT" exit 0 fi - pattern='^(praxis/|\.github/workflows/praxis\.yaml$)' + # The hop, the egress listener and the gateway wiring are praxis mode's Go side, + # not ported code: a change there that breaks the end-to-end tests must block. + pattern='^(praxis/|\.github/workflows/praxis\.yaml$|core/listener/forwardproxy/(hop|egress)[^/]*\.go$|cmd/cortex/(praxis[^/]*|gateway|ingest)\.go$)' if git diff --name-only "$BASE_SHA" HEAD | grep -qE "$pattern"; then echo "blocking=true" >> "$GITHUB_OUTPUT" else echo "blocking=false" >> "$GITHUB_OUTPUT" - echo "No praxis/ changes; failures do not block." + echo "No praxis-mode changes. Failures do not block." fi # Installs the toolchain that praxis/rust-toolchain.toml names. @@ -150,10 +155,10 @@ jobs: TEST: ${{ steps.test.outcome }} E2E: ${{ steps.e2e.outcome }} run: | - [ "$FMT" != failure ] || echo "::warning::cargo fmt failed (does not block: no praxis/ changes)" - [ "$CLIPPY" != failure ] || echo "::warning::cargo clippy failed (does not block: no praxis/ changes)" - [ "$TEST" != failure ] || echo "::warning::cargo test failed (does not block: no praxis/ changes)" - [ "$E2E" != failure ] || echo "::warning::praxis end-to-end tests failed (does not block: no praxis/ changes)" + [ "$FMT" != failure ] || echo "::warning::cargo fmt failed (does not block: no praxis-mode changes)" + [ "$CLIPPY" != failure ] || echo "::warning::cargo clippy failed (does not block: no praxis-mode changes)" + [ "$TEST" != failure ] || echo "::warning::cargo test failed (does not block: no praxis-mode changes)" + [ "$E2E" != failure ] || echo "::warning::praxis end-to-end tests failed (does not block: no praxis-mode changes)" # Size cap: the cache post-step must not save a target/ over 3 GiB. - name: Cap the target/ cache size diff --git a/cmd/cortex/praxis_e2e_test.go b/cmd/cortex/praxis_e2e_test.go index 933e77b31..a0c4f77c2 100644 --- a/cmd/cortex/praxis_e2e_test.go +++ b/cmd/cortex/praxis_e2e_test.go @@ -731,14 +731,19 @@ func eventsJSON(t *testing.T, c *chain) []map[string]any { // The JSON and SSE requests record the same rows in both modes. func TestPraxisE2E_SameEventsAsDirect(t *testing.T) { origin := newMockOrigin(t) + // ttfbMs and durationMs are omitempty. The delay keeps both at 5 ms or more, so a + // missing key means the field is missing, not a 0 ms value. + slow := func(h http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { time.Sleep(5 * time.Millisecond); h(w, r) } + } var got [2][]map[string]any for i, c := range []*chain{newDirectChain(t, origin), newPraxisChain(t, origin, chainOpts{})} { - origin.serve(serveJSON(chatJSON)) + origin.serve(slow(serveJSON(chatJSON))) if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatRequest)); st != http.StatusOK { t.Fatalf("status %d: %s", st, b) } c.waitResponses(t, 1, 10*time.Second) - origin.serve(serveFrames("text/event-stream", false, chatFrames, nil, 0)) + origin.serve(slow(serveFrames("text/event-stream", false, chatFrames, nil, 0))) if st, b, _ := c.do(t, request(t, t.Context(), http.MethodPost, chatURL, chatStreamRequest)); st != http.StatusOK { t.Fatalf("status %d: %s", st, b) } @@ -756,9 +761,12 @@ func TestPraxisE2E_SameEventsAsDirect(t *testing.T) { delete(p, k) } for _, k := range []string{"ttfbMs", "durationMs"} { - dv, _ := d[k].(float64) - pv, _ := p[k].(float64) - if dv-pv > 50 || pv-dv > 50 { + dv, dok := d[k].(float64) + pv, pok := p[k].(float64) + // A missing key reads as 0, which is within 50 ms of the other side's value. + if dok != pok { + t.Errorf("row %d: %s = %s, direct %s (on one side only)", i, k, compact(p[k]), compact(d[k])) + } else if dv-pv > 50 || pv-dv > 50 { t.Errorf("row %d: %s = %v, direct %v (over 50 ms apart)", i, k, pv, dv) } delete(d, k) From 3185cfb1efb4d939b1a421fbfbc23e3a7fcdeaa8 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 07:46:36 -0500 Subject: [PATCH 5/6] test: Fail the praxis end-to-end tests on a wrong gateway path - newPraxisChain skips only when CORTEX_GATEWAY_BIN is not set. A set path that names no file now fails the test. Before, a wrong path in CI skipped all 11 tests, and the blocking step passed with no test run. - CodexMislabeledSSE: the 405 row must have no error message. The plan says that the provider's text does not go into the store, and now a test checks it. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- cmd/cortex/praxis_e2e_test.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/cmd/cortex/praxis_e2e_test.go b/cmd/cortex/praxis_e2e_test.go index a0c4f77c2..a3cc0dfd2 100644 --- a/cmd/cortex/praxis_e2e_test.go +++ b/cmd/cortex/praxis_e2e_test.go @@ -186,8 +186,12 @@ func newDirectChain(t *testing.T, origin *mockOrigin) *chain { func newPraxisChain(t *testing.T, origin *mockOrigin, opts chainOpts) *chain { t.Helper() bin := os.Getenv("CORTEX_GATEWAY_BIN") - if st, err := os.Stat(bin); bin == "" || err != nil || st.IsDir() { - t.Skip("CORTEX_GATEWAY_BIN does not name the cortex-gateway binary") + if bin == "" { + t.Skip("CORTEX_GATEWAY_BIN is not set") + } + // A set but wrong path fails: a skip there would make CI's blocking step pass with no test run. + if st, err := os.Stat(bin); err != nil || st.IsDir() { + t.Fatalf("CORTEX_GATEWAY_BIN=%q does not name the cortex-gateway binary: %v", bin, err) } ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) @@ -631,8 +635,9 @@ func TestPraxisE2E_CodexMislabeledSSE(t *testing.T) { t.Fatalf("got %d %q, want 405 %q", st, b, detail) } row := lastResponse(t, c.waitResponses(t, 2, 10*time.Second)) - if row.StatusCode != http.StatusMethodNotAllowed || row.Error == nil || row.Error.Kind != "backend_error" || row.Error.Code != "405" { - t.Errorf("row = %d %+v, want 405 backend_error/405", row.StatusCode, row.Error) + // The provider's text must not go into the store, which has no authentication. + if row.StatusCode != http.StatusMethodNotAllowed || row.Error == nil || row.Error.Kind != "backend_error" || row.Error.Code != "405" || row.Error.Message != "" { + t.Errorf("row = %d %+v, want 405 backend_error/405 with no message", row.StatusCode, row.Error) } } From 9c933eba638562d3c73740e240c8ae1b73c98e9d Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Tue, 10 Nov 2026 09:02:54 -0500 Subject: [PATCH 6/6] ci: Run the praxis end-to-end checks for record.go and tunnel.go and cache Go modules Fixes review: record.go and tunnel.go did not start the workflow Fixes review: the setup-go step had no cache-dependency-path Files: - .github/workflows/praxis.yaml Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- .github/workflows/praxis.yaml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/praxis.yaml b/.github/workflows/praxis.yaml index d254b5bd7..3d3a4b4a3 100644 --- a/.github/workflows/praxis.yaml +++ b/.github/workflows/praxis.yaml @@ -36,6 +36,8 @@ on: - "core/listener/forwardproxy/ingest_replay_test.go" - "core/listener/forwardproxy/hop*.go" - "core/listener/forwardproxy/egress*.go" + - "core/listener/forwardproxy/record.go" + - "core/listener/forwardproxy/tunnel.go" - "cmd/cortex/praxis*.go" - "cmd/cortex/gateway.go" - "cmd/cortex/ingest.go" @@ -66,6 +68,8 @@ on: - "core/listener/forwardproxy/ingest_replay_test.go" - "core/listener/forwardproxy/hop*.go" - "core/listener/forwardproxy/egress*.go" + - "core/listener/forwardproxy/record.go" + - "core/listener/forwardproxy/tunnel.go" - "cmd/cortex/praxis*.go" - "cmd/cortex/gateway.go" - "cmd/cortex/ingest.go" @@ -136,6 +140,7 @@ jobs: - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 with: go-version-file: cmd/cortex/go.mod + cache-dependency-path: cmd/cortex/go.sum # Runs from the repo root: the command names praxis/ and cmd/cortex. rustup reads # rust-toolchain.toml only below praxis/, so the step names that toolchain itself.