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 d719089ce..3d3a4b4a3 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 @@ -33,6 +34,13 @@ on: - "core/redact/**" - "core/pipeline/subagent.go" - "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" push: branches: - main @@ -58,6 +66,13 @@ on: - "core/redact/**" - "core/pipeline/subagent.go" - "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" permissions: contents: read @@ -84,12 +99,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. @@ -120,6 +137,24 @@ 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 + 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. + - 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 . + # 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' @@ -127,10 +162,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)" + [ "$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 new file mode 100644 index 000000000..a3cc0dfd2 --- /dev/null +++ b/cmd/cortex/praxis_e2e_test.go @@ -0,0 +1,816 @@ +//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 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) + 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)) + // 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) + } +} + +// 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) + // 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(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(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) + } + 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, 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) + 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) +} 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 773a5cc1a..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 @@ -92,10 +92,10 @@ Depends on: none. Estimated lines: 80. - [ ] **Step 1: Change the tests first.** - In `hop_headers_removed_custom_kept`: assert that `x-cortex-dest` and `x-cortex-token` now reach the upstream with the values sent, and that the other ten hop headers do not. `X-Cortex-Custom` still reaches it. The request capture in the ingested line still has none of the twelve. - In `row.rs` tests: `egress_kind("destination_private") == "destination_private"`. -- [ ] **Step 2:** `cd praxis && cargo test --locked` fails on both. +- [ ] **Step 2:** `cargo test --locked --manifest-path praxis/Cargo.toml` fails on both. - [ ] **Step 3:** Implement. Doc comment on `UPSTREAM_STRIP_HEADERS`: the interim upstream is always the egress listener, which removes Dest and Token. At phase B both go on this list. - [ ] **Step 4:** Update README decision 20: the gateway keeps Dest and Token for its upstream, the egress listener (#1721), and removes them before any other upstream. -- [ ] **Step 5:** `cd praxis && cargo fmt --all --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked` passes. +- [ ] **Step 5:** `cd praxis && cargo fmt --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked` passes. - [ ] **Step 6:** Commit `feat: Keep X-Cortex-Dest and X-Cortex-Token for the egress listener`. ### Task 2: `attribution.Wire`, the Go encoder of `X-Cortex-Attribution` @@ -405,13 +405,13 @@ 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. - `TestPraxisE2E_SameEventsAsDirect`: the JSON and SSE requests through both chains. Compare `/v1/sessions/{id}` events field by field after removing `seq`, `at`, `sessionId`, `requestId`. `ttfbMs` and `durationMs` agree within 50 ms. Any other difference fails the test. Do not widen the list in the test: fix the cause, or add it to the README's known differences with its reason and a pinned assertion. - [ ] **Step 2:** Build the gateway and run: - `(cd praxis && cargo build --locked -p cortex-gateway) && 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 .` + `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 .` Fix what fails in the task that owns the code, in this PR. - [ ] **Step 3: CI.** In `praxis.yaml`, after `cargo test`, add `actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0` (`go-version-file: cmd/cortex/go.mod`) and a step `End-to-end (praxis mode)` with the Step 2 command and `continue-on-error: ${{ steps.changes.outputs.blocking != 'true' }}`. Give the step `id: e2e` and no `working-directory`, so it runs from the repo root as the command expects. Add its outcome to the report step as the other three are (`E2E: ${{ steps.e2e.outcome }}` in `env`, and a `::warning::` line). Add to both `paths` lists: `core/listener/forwardproxy/hop*.go`, `core/listener/forwardproxy/egress*.go`, `cmd/cortex/praxis*.go`, `cmd/cortex/gateway.go`. - [ ] **Step 4:** Commit `test: Run the praxis-mode end-to-end checks through cortex-gateway`. @@ -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" ./... && (cd praxis && cargo fmt --all --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked && cargo build --locked -p cortex-gateway) && 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.