Feat: Join gateway rows to tunnels and turn on the hop in praxis mode

cmd/cortex/gateway.go

@@ -63,8 +63,8 @@ type gatewayFilter struct {
 }
 
 // gatewayYAML returns the cortex-gateway config for cfg (praxis mode spec,
-// Section 4). Until the egress listener lands (#1721), one catch-all route
-// sends every request to the egress cluster.
+// Section 4). One catch-all route sends every request to the egress cluster:
+// Cortex's egress listener, which dials the origin that X-Cortex-Dest names.
 func gatewayYAML(cfg *config.Config, tokenPath string) ([]byte, error) {
 	ingest, err := ingestAddr(cfg.Listener.SessionAPIAddr)
 	if err != nil {

cmd/cortex/ingest.go

@@ -13,23 +13,28 @@ import (
 // POST /v1/ingest. The supervisor creates the file; Cortex only reads it.
 const gatewayTokenEnv = "CORTEX_GATEWAY_TOKEN_FILE"
 
-// ingestOptions turns on the ingestion endpoint in praxis mode. In direct mode it returns
-// nil and reads neither the environment nor a file.
-func ingestOptions(cfg *config.Config, rates pricing.Resolver, getenv func(string) string, readFile func(string) ([]byte, error)) ([]sessionapi.Option, error) {
-	if !cfg.PraxisEnabled() {
-		return nil, nil
-	}
+// readGatewayToken reads the token file that CORTEX_GATEWAY_TOKEN_FILE names.
+func readGatewayToken(getenv func(string) string, readFile func(string) ([]byte, error)) (string, error) {
 	path := getenv(gatewayTokenEnv)
 	if path == "" {
-		return nil, fmt.Errorf("praxis.enabled needs %s to name the gateway token file", gatewayTokenEnv)
+		return "", fmt.Errorf("praxis.enabled needs %s to name the gateway token file", gatewayTokenEnv)
 	}
 	b, err := readFile(path)
 	if err != nil {
-		return nil, fmt.Errorf("read the gateway token file (%s): %w", gatewayTokenEnv, err)
+		return "", fmt.Errorf("read the gateway token file (%s): %w", gatewayTokenEnv, err)
 	}
 	token := strings.TrimSpace(string(b))
 	if token == "" {
-		return nil, fmt.Errorf("the gateway token file %s (%s) is empty", path, gatewayTokenEnv)
+		return "", fmt.Errorf("the gateway token file %s (%s) is empty", path, gatewayTokenEnv)
+	}
+	return token, nil
+}
+
+// ingestOptions turns on the ingestion endpoint in praxis mode, with the hop's joins
+// (nil when no forward proxy runs). In direct mode it returns nil.
+func ingestOptions(cfg *config.Config, rates pricing.Resolver, token string, joins sessionapi.HopJoins) []sessionapi.Option {
+	if !cfg.PraxisEnabled() {
+		return nil
 	}
-	return []sessionapi.Option{sessionapi.WithIngest(sessionapi.Ingest{Token: token, Rates: rates})}, nil
+	return []sessionapi.Option{sessionapi.WithIngest(sessionapi.Ingest{Token: token, Rates: rates, Joins: joins})}
 }

cmd/cortex/ingest_test.go

@@ -21,37 +21,37 @@ func praxisCfg(enabled bool) *config.Config {
 }
 
 func TestIngestOptions_DirectModeNone(t *testing.T) {
-	getenv := func(string) string { t.Fatal("getenv called in direct mode"); return "" }
-	readFile := func(string) ([]byte, error) { t.Fatal("readFile called in direct mode"); return nil, nil }
-	opts, err := ingestOptions(praxisCfg(false), nil, getenv, readFile)
-	if err != nil || opts != nil {
-		t.Fatalf("got %v, %v; want nil, nil", opts, err)
+	if opts := ingestOptions(praxisCfg(false), nil, "tok", nil); opts != nil {
+		t.Fatalf("got %v; want nil", opts)
 	}
 }
 
-func TestIngestOptions_NeedsTokenFile(t *testing.T) {
-	_, err := ingestOptions(praxisCfg(true), nil, func(string) string { return "" },
+func TestReadGatewayToken_NeedsTokenFile(t *testing.T) {
+	_, err := readGatewayToken(func(string) string { return "" },
 		func(string) ([]byte, error) { return nil, nil })
 	if err == nil || !strings.Contains(err.Error(), "CORTEX_GATEWAY_TOKEN_FILE") {
 		t.Fatalf("err = %v, want one that names CORTEX_GATEWAY_TOKEN_FILE", err)
 	}
-	_, err = ingestOptions(praxisCfg(true), nil, func(string) string { return "/x" },
+}
+
+func TestReadGatewayToken_Unreadable(t *testing.T) {
+	_, err := readGatewayToken(func(string) string { return "/x" },
 		func(string) ([]byte, error) { return nil, errors.New("no such file") })
-	if err == nil {
-		t.Fatal("an unreadable token file must be an error")
+	if err == nil || !strings.Contains(err.Error(), "read the gateway token file") {
+		t.Fatalf("err = %v, want an unreadable-file error", err)
 	}
 }
 
-func TestIngestOptions_EmptyToken(t *testing.T) {
-	_, err := ingestOptions(praxisCfg(true), nil, func(string) string { return "/x" },
+func TestReadGatewayToken_EmptyToken(t *testing.T) {
+	_, err := readGatewayToken(func(string) string { return "/x" },
 		func(string) ([]byte, error) { return []byte(" \n"), nil })
-	if err == nil {
-		t.Fatal("an empty token must be an error")
+	if err == nil || !strings.Contains(err.Error(), "is empty") {
+		t.Fatalf("err = %v, want an empty-token error", err)
 	}
 }
 
 func TestIngestOptions_ReadsToken(t *testing.T) {
-	opts, err := ingestOptions(praxisCfg(true), nil,
+	token, err := readGatewayToken(
 		func(k string) string {
 			if k != "CORTEX_GATEWAY_TOKEN_FILE" {
 				t.Fatalf("getenv(%q)", k)
@@ -64,8 +64,12 @@ func TestIngestOptions_ReadsToken(t *testing.T) {
 			}
 			return []byte("abc\n"), nil
 		})
-	if err != nil || len(opts) != 1 {
-		t.Fatalf("got %d options, %v; want 1, nil", len(opts), err)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts := ingestOptions(praxisCfg(true), nil, token, nil)
+	if len(opts) != 1 {
+		t.Fatalf("got %d options; want 1", len(opts))
 	}
 	srv := sessionapi.New(":0", session.New(0, 0, 0), opts...)
 	ts := httptest.NewServer(srv.Server().Handler)

cmd/cortex/main.go

@@ -579,7 +579,7 @@ func main() {
 		if err != nil {
 			return nil, nil, nil, fmt.Errorf("inbound: %w", err)
 		}
-		out, err := plugins.BuildWithDeps(c.Pipeline.Outbound.Plugins, outDeps)
+		out, err := plugins.BuildWithDeps(outboundEntries(c, slog.Default()), outDeps)
 		if err != nil {
 			return nil, nil, nil, fmt.Errorf("outbound: %w", err)
 		}
@@ -874,6 +874,14 @@ func main() {
 	// TLS so the outbound pipeline sees decrypted HTTPS. Constructed
 	// here and set on fpSrv below (mirroring fpSrv.SkipHosts / fpSrv.Shared).
 	// A nil *Engine leaves today's blind-tunnel behavior intact.
+	// The bridge's re-origination client, and the praxis egress listener's, dial
+	// through the egress guard when one is armed, so a bridged request is checked
+	// exactly like a plaintext one. No option when the guard is off: nil
+	// DialContext is the transport's default dialer.
+	var upOpts []tlsbridge.UpstreamOption
+	if egress != nil {
+		upOpts = append(upOpts, tlsbridge.WithDialContext(egress.DialContext))
+	}
 	var bridge *tlsbridge.Engine
 	if cfg.TLSBridge != nil && cfg.TLSBridge.Mode == "enabled" {
 		// CA is normally the operator-mounted cert-manager Secret (tls.crt/tls.key
@@ -957,14 +965,6 @@ func main() {
 				fatalf("tls-bridge upstream_ca_bundle read failed: %v", err)
 			}
 		}
-		// The bridge's re-origination client dials through the egress guard
-		// when one is armed, so a bridged request is checked exactly like a
-		// plaintext one. No option when the guard is off: nil DialContext is
-		// the transport's default dialer.
-		var upOpts []tlsbridge.UpstreamOption
-		if egress != nil {
-			upOpts = append(upOpts, tlsbridge.WithDialContext(egress.DialContext))
-		}
 		up, uerr := tlsbridge.NewUpstreamClient(extra, cfg.TLSBridge.UpstreamInsecure, upOpts...)
 		if uerr != nil {
 			fatalf("tls-bridge upstream client failed: %v", uerr)
@@ -1051,6 +1051,17 @@ func main() {
 	if ctrlErr != nil {
 		fatalf("vm control: %v", ctrlErr)
 	}
+	// Praxis mode: one read of the gateway token, for the hop, the egress listener and
+	// POST /v1/ingest. hopJoins stays nil when no forward proxy runs.
+	var gatewayToken string
+	var hopJoins sessionapi.HopJoins
+	if cfg.PraxisEnabled() {
+		t, terr := readGatewayToken(os.Getenv, os.ReadFile)
+		if terr != nil {
+			fatalf("praxis: %v", terr)
+		}
+		gatewayToken = t
+	}
 	if roles[config.RoleForward] {
 		fpSrv, ferr := forwardproxy.NewServer(outboundH, sessions, fpMTLS, forwardproxy.WithEgressGuard(egress))
 		if ferr != nil {
@@ -1147,9 +1158,33 @@ func main() {
 		if ctrlCfg.Socket != "" {
 			vmCtrl = startVMControl(ctrlCfg, fpSrv)
 		}
+		// Praxis mode, also before the forward proxy serves: startPraxis sets fpSrv.Hop
+		// and fpSrv.ResponsesWebSockets. The egress listener uses the bridge's upstream
+		// options (Decision 11).
+		var egressSrv *http.Server
+		if cfg.PraxisEnabled() {
+			var upstream http.RoundTripper
+			if bridge != nil {
+				upstream = bridge.Upstream.Transport
+			} else {
+				up, uerr := tlsbridge.NewUpstreamClient(nil, false, upOpts...)
+				if uerr != nil {
+					fatalf("praxis egress upstream client failed: %v", uerr)
+				}
+				upstream = up.Transport
+			}
+			var hop *forwardproxy.Hop
+			hop, egressSrv = startPraxis(ctx, cfg, fpSrv, gatewayToken, upstream, listeners.take("praxis-egress"))
+			hopJoins = hop
+		}
 
 		httpServers = append(httpServers, bootstrap.ServeHTTPServer("forward-proxy", fpSrv.Handler(),
 			listeners.take("forward-proxy"), bootstrap.WithConnContext(fpSrv.ConnContext)))
+		// After the forward proxy: the drain shuts servers down in slice order, and a
+		// request still on the forward proxy reaches the gateway, which dials this listener.
+		if egressSrv != nil {
+			httpServers = append(httpServers, egressSrv)
+		}
 
 		// Outbound transparent listener (enforce-redirect mode). It shares the
 		// forward proxy's outbound pipeline via HandleTransparentConn, so explicit
@@ -1192,10 +1227,7 @@ func main() {
 	var sessionAPISrv *sessionapi.Server
 	if cfg.Listener.SessionAPIAddr != "" && sessions != nil {
 		// POST /v1/ingest, in praxis mode only: nil options in direct mode.
-		ingestOpts, err := ingestOptions(cfg, pricingRegistry, os.Getenv, os.ReadFile)
-		if err != nil {
-			fatalf("praxis ingestion: %v", err)
-		}
+		ingestOpts := ingestOptions(cfg, pricingRegistry, gatewayToken, hopJoins)
 		sessionAPISrv = sessionapi.New(
 			cfg.Listener.SessionAPIAddr,
 			sessions,

cmd/cortex/praxis.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/rossoctl/cortex/core/bootstrap"
+	"github.com/rossoctl/cortex/core/config"
+	"github.com/rossoctl/cortex/core/listener/forwardproxy"
+	"github.com/rossoctl/cortex/core/sessionapi"
+)
+
+var _ sessionapi.HopJoins = (*forwardproxy.Hop)(nil)
+
+// praxisProbeEvery is how often the hop probes the gateway until one probe passes (Decision 3).
+const praxisProbeEvery = time.Second
+
+// outboundEntries is the outbound plugin list to build: none in praxis mode, and it logs the
+// skipped names at Warn: no policy plugin runs in praxis mode, because the gateway does not
+// run them and a WebSocket upgrade does not go through the gateway. The config keeps its list.
+func outboundEntries(c *config.Config, log *slog.Logger) []config.PluginEntry {
+	if !c.PraxisEnabled() {
+		return c.Pipeline.Outbound.Plugins
+	}
+	if len(c.Pipeline.Outbound.Plugins) > 0 {
+		names := make([]string, len(c.Pipeline.Outbound.Plugins))
+		for i, e := range c.Pipeline.Outbound.Plugins {
+			names[i] = e.Name
+		}
+		log.Warn("praxis mode: outbound plugins are off; cortex-gateway does not run them, and WebSocket upgrades go direct", "plugins", names)
+	}
+	return nil
+}
+
+// startPraxis sets fpSrv.Hop and decline mode, serves the egress listener on ln with upstream,
+// starts the probe, and returns the hop for Ingest.Joins and the egress server for shutdown.
+// Call it before the forward proxy serves: nothing locks the fields it sets.
+func startPraxis(ctx context.Context, cfg *config.Config, fpSrv *forwardproxy.Server, token string, upstream http.RoundTripper, ln net.Listener) (*forwardproxy.Hop, *http.Server) {
+	hop := forwardproxy.NewHop(cfg.Praxis.ListenAddr(), token)
+	fpSrv.Hop = hop
+	// The gateway does not carry a Responses API WebSocket, so Codex must use HTTPS.
+	fpSrv.ResponsesWebSockets = config.ResponsesWebSocketsDecline
+	srv := bootstrap.ServeHTTPServer("praxis-egress", forwardproxy.NewEgressHandler(token, upstream), ln)
+	go hop.Probe(ctx, praxisProbeEvery)
+	return hop, srv
+}

cmd/cortex/praxis_test.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rossoctl/cortex/core/config"
+	"github.com/rossoctl/cortex/core/listener/forwardproxy"
+)
+
+func outboundCfg(praxis bool) *config.Config {
+	cfg := praxisCfg(praxis)
+	cfg.Pipeline.Outbound.Plugins = []config.PluginEntry{{Name: "inference-parser"}, {Name: "tool-prune"}}
+	return cfg
+}
+
+func TestOutboundEntries_PraxisSkipsPlugins(t *testing.T) {
+	var buf bytes.Buffer
+	log := slog.New(slog.NewTextHandler(&buf, nil))
+	cfg := outboundCfg(true)
+	if got := outboundEntries(cfg, log); len(got) != 0 {
+		t.Fatalf("praxis mode built %v, want none", got)
+	}
+	if !strings.Contains(buf.String(), "level=WARN") {
+		t.Errorf("log %q is not at Warn", buf.String())
+	}
+	for _, name := range []string{"inference-parser", "tool-prune"} {
+		if !strings.Contains(buf.String(), name) {
+			t.Errorf("log %q does not name %s", buf.String(), name)
+		}
+	}
+	if len(cfg.Pipeline.Outbound.Plugins) != 2 {
+		t.Errorf("the config's list changed: %v", cfg.Pipeline.Outbound.Plugins)
+	}
+
+	buf.Reset()
+	got := outboundEntries(outboundCfg(false), log)
+	if len(got) != 2 || !slices.Equal([]string{got[0].Name, got[1].Name}, []string{"inference-parser", "tool-prune"}) {
+		t.Fatalf("direct mode got %v", got)
+	}
+	if buf.Len() != 0 {
+		t.Errorf("direct mode logged %q", buf.String())
+	}
+}
+
+func TestOutboundEntries_PraxisEmptyListLogsNothing(t *testing.T) {
+	var buf bytes.Buffer
+	if got := outboundEntries(praxisCfg(true), slog.New(slog.NewTextHandler(&buf, nil))); len(got) != 0 {
+		t.Fatalf("got %v", got)
+	}
+	if buf.Len() != 0 {
+		t.Errorf("logged %q", buf.String())
+	}
+}
+
+func listen(t *testing.T) net.Listener {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ln
+}
+
+func TestStartPraxis_DeclinesResponsesWebSocket(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	fp := &forwardproxy.Server{ResponsesWebSockets: config.ResponsesWebSocketsParse}
+	hop, srv := startPraxis(ctx, praxisCfg(true), fp, "tok", http.DefaultTransport, listen(t))
+	defer srv.Close()
+	if fp.ResponsesWebSockets != config.ResponsesWebSocketsDecline {
+		t.Errorf("ResponsesWebSockets = %q, want decline", fp.ResponsesWebSockets)
+	}
+	if fp.Hop == nil || fp.Hop != hop {
+		t.Errorf("fp.Hop = %v, want the returned hop", fp.Hop)
+	}
+}
+
+func TestStartPraxis_EgressServesProbe(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ln := listen(t)
+	// A fake gateway: it forwards everything, headers and all, to the egress listener.
+	gw := httptest.NewServer(httputil.NewSingleHostReverseProxy(&url.URL{Scheme: "http", Host: ln.Addr().String()}))
+	defer gw.Close()
+	cfg := praxisCfg(true)
+	cfg.Praxis.Listen = gw.Listener.Addr().String()
+	hop, srv := startPraxis(ctx, cfg, &forwardproxy.Server{}, "tok", http.DefaultTransport, ln)
+	defer srv.Close()
+	deadline := time.Now().Add(3 * time.Second)
+	for !hop.Ready() {
+		if time.Now().After(deadline) {
+			t.Fatal("the hop is not ready after 3s")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}

core/attribution/testdata/wire.golden

@@ -0,0 +1 @@
+eyJhZ2VudCI6ImNsYXVkZS1jb2RlIiwicHJvYyI6InAxIiwiY2xpZW50UHJvYyI6InAyIiwic2Vzc2lvbiI6InMxIiwic291cmNlIjoiaGVhZGVyIiwic2Vzc2lvbkhlYWRlciI6IlgtQ2xhdWRlLUNvZGUtU2Vzc2lvbi1JZCIsImNoYWluU2Vzc2lvbiI6InMwIiwicmVjb3JkU2Vzc2lvbiI6InMxIn0