Feat: Supervise cortex-gateway as a sibling of cortex

cmd/agentop/cmd_uninstall.go

@@ -27,8 +27,8 @@ Usage:
 Unroutes Claude Code, OpenCode, IBM Bob, and the bob and pi shell functions and
 Pi's session extension where Cortex routed them; stops and removes the service, or the background proxy; removes the PATH
 lines setup added, unless other tools in ~/.local/bin still need them; and
-removes agentop, cortex and cortex-session-dump from ~/.local/bin. ~/.cortex
-stays unless --purge. What to remove is read from disk, not from a record of
+removes agentop, cortex, cortex-session-dump and cortex-gateway from
+~/.local/bin. ~/.cortex stays unless --purge. What to remove is read from disk, not from a record of
 the install. A removal that fails is reported with its fix and the rest still
 run, but ~/.cortex then stays; the end lists what was left behind. ~/.cortex also
 stays when a setting still points into it afterwards — one left in place, or one

cmd/agentop/cmd_uninstall_test.go

@@ -1129,6 +1129,25 @@ func TestUninstallRemovesOurPreRenameBinaries(t *testing.T) {
 	}
 }
 
+// cortex-gateway is Cortex's own: uninstall removes it with the rest, and it does
+// not count as another tool that needs the PATH lines.
+func TestUninstallRemovesTheGateway(t *testing.T) {
+	sc := newUninstallScene(t)
+	sc.setUp(t)
+	bin := filepath.Join(sc.home, ".local", "bin")
+	gw := filepath.Join(bin, "cortex-gateway")
+	writeExe(t, gw, "#!/bin/sh\n")
+	env := &setupEnv{binDir: bin}
+	if !cortexsOwn(env, "cortex-gateway") || slices.Contains(othersInBinDir(env), "cortex-gateway") {
+		t.Errorf("cortex-gateway is not Cortex's own: others = %q", othersInBinDir(env))
+	}
+	code, out := sc.uninstall(t, "--yes")
+	wantLines(t, code, 0, out, markLine("✓", "removed")+"agentop, cortex, cortex-gateway from ~/.local/bin\n")
+	if _, err := os.Lstat(gw); !os.IsNotExist(err) {
+		t.Errorf("cortex-gateway is still there (lstat: %v)", err)
+	}
+}
+
 // lockedWriter lets a test read what a spinner goroutine may still be writing.
 type lockedWriter struct {
 	mu sync.Mutex

cmd/agentop/setup_step_binaries.go

@@ -14,9 +14,10 @@ import (
 // installerOneLiner is the fix for "nothing is installed": the documented install.
 const installerOneLiner = "curl -fsSL https://www.rossoctl.dev/cortex/install.sh | sh"
 
-// setupBinaryNames are what the binaries step installs; cortex-session-dump only
-// when the installer staged it (its fetch is never fatal).
-var setupBinaryNames = []string{"agentop", proxyBinary, "cortex-session-dump"}
+// setupBinaryNames are what the binaries step installs; cortex-session-dump and
+// cortex-gateway only when the installer staged them (the dump's fetch is never
+// fatal, and no release has a gateway yet).
+var setupBinaryNames = []string{"agentop", proxyBinary, "cortex-session-dump", "cortex-gateway"}
 
 type binariesStep struct{}
 

cmd/agentop/setup_step_binaries_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"os"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"testing"
@@ -158,6 +159,19 @@ func TestBinariesNameWhatIsStaged(t *testing.T) {
 	}
 }
 
+// Setup installs cortex-gateway only when the installer staged it.
+func TestStagedBinariesNeedNoGateway(t *testing.T) {
+	env := newTestSetupEnv(t)
+	env.fromDir = stageDir(t, "#!/bin/sh\necho cortex v9.9.9\n")
+	if got := stagedBinaries(env); !slices.Equal(got, []string{"agentop", "cortex"}) {
+		t.Errorf("stagedBinaries = %q, want agentop and cortex only", got)
+	}
+	writeExe(t, filepath.Join(env.fromDir, "cortex-gateway"), "#!/bin/sh\n")
+	if got := stagedBinaries(env); !slices.Contains(got, "cortex-gateway") {
+		t.Errorf("stagedBinaries = %q, want cortex-gateway in it", got)
+	}
+}
+
 func TestBinariesUnchangedIsDone(t *testing.T) {
 	env := newTestSetupEnv(t)
 	env.fromDir = stageDir(t, "#!/bin/sh\necho cortex v9.9.9\n")

cmd/cortex/fallback.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"log/slog"
 	"net"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -56,9 +57,10 @@ func configPathFromArgs(args []string) string {
 	return path
 }
 
-// supervisorForwardAddr is the address the fallback listener binds: the forward proxy's
-// own, so a client routed at it reaches the fallback when no child holds it. Every miss
-// falls back rather than failing, for the reason on fallbackDefaultAddr.
+// supervisorConfig loads the config at path once, for the fallback address and for
+// praxis mode, or returns nil when there is no path or it will not load. Every miss
+// is direct mode with the default fallback address, for the reason on
+// fallbackDefaultAddr.
 //
 // The three steps below mirror the child's own sequence -- Load, coerce the mode,
 // ApplyPreset (main.go:491, :500, :501) -- because only all three yield the address the
@@ -75,10 +77,9 @@ func configPathFromArgs(args []string) string {
 // anything but proxy-sidecar (main.go:495) and so binds nothing at all. The fallback
 // still has somewhere useful to be: whatever the config pinned, else the laptop default,
 // which is where a client routed at this install points either way.
-func supervisorForwardAddr(args []string) string {
-	path := configPathFromArgs(args)
+func supervisorConfig(path string) *config.Config {
 	if path == "" {
-		return fallbackDefaultAddr
+		return nil
 	}
 	cfg, err := config.Load(path)
 	if err != nil {
@@ -87,15 +88,22 @@ func supervisorForwardAddr(args []string) string {
 		// fallback is listening somewhere other than where the config asked.
 		slog.Warn("supervisor: cannot read the config for the fallback address; using the default",
 			"config", path, "addr", fallbackDefaultAddr, "error", err)
-		return fallbackDefaultAddr
+		return nil
 	}
 	cfg.Mode = config.ModeProxySidecar
 	config.ApplyPreset(cfg)
+	return cfg
+}
+
+// supervisorForwardAddr is the address the fallback listener binds: the forward proxy's
+// own, so a client routed at it reaches the fallback when no child holds it. cfg is
+// supervisorConfig's, nil when it did not load.
+func supervisorForwardAddr(cfg *config.Config) string {
 	// Still reachable after the preset, which fills an address only for an ACTIVE role
 	// (presets.go:29): a reverse-only config leaves this empty, and the child then binds
 	// no forward proxy at all. Nothing is routed at the fallback in that case, so where
 	// it listens does not much matter -- but it has to listen somewhere.
-	if cfg.Listener.ForwardProxyAddr == "" {
+	if cfg == nil || cfg.Listener.ForwardProxyAddr == "" {
 		return fallbackDefaultAddr
 	}
 	return cfg.Listener.ForwardProxyAddr
@@ -290,3 +298,36 @@ func requestLineForLog(line string) string {
 	}
 	return line
 }
+
+// supervisorConfigPath is the config file the cortex child reads: the -config value,
+// else ~/.cortex/config.yaml when -local or its -demo alias is set, as main.go does.
+// main.go refuses -local together with -config, so -config is checked first.
+func supervisorConfigPath(args []string) string {
+	if path := configPathFromArgs(args); path != "" {
+		return path
+	}
+	// main.go reads the two bool flags separately and ORs them; last occurrence wins.
+	set := map[string]bool{}
+	for _, a := range args {
+		if !strings.HasPrefix(a, "-") {
+			continue
+		}
+		name, val, hasVal := strings.Cut(strings.TrimPrefix(a[1:], "-"), "=")
+		if name != "local" && name != "demo" {
+			continue
+		}
+		on := true
+		if hasVal {
+			on, _ = strconv.ParseBool(val) // flag.Parse rejects a bad value, so the child never runs
+		}
+		set[name] = on
+	}
+	if !set["local"] && !set["demo"] {
+		return ""
+	}
+	dir, err := defaultCortexDir()
+	if err != nil {
+		return "" // the child fails on the same error
+	}
+	return filepath.Join(dir, localConfigName)
+}

cmd/cortex/fallback_test.go

@@ -108,7 +108,7 @@ func TestSupervisorForwardAddr_ResolvesTheAddressTheChildWillBind(t *testing.T)
 		{"reverse-only leaves no forward address", []string{"--config", reverseOnly}, fallbackDefaultAddr},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := supervisorForwardAddr(tc.args); got != tc.want {
+			if got := supervisorForwardAddr(supervisorConfig(configPathFromArgs(tc.args))); got != tc.want {
 				t.Errorf("supervisorForwardAddr(%q) = %q, want %q", tc.args, got, tc.want)
 			}
 		})

cmd/cortex/gateway.go

@@ -0,0 +1,165 @@
+package main
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"net"
+	"net/netip"
+	"os"
+	"path/filepath"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/rossoctl/cortex/core/config"
+)
+
+// The gateway binary, and the two files the supervisor writes for it in the
+// directory of the --config file.
+const (
+	gatewayBinary    = "cortex-gateway"
+	gatewayTokenName = "gateway.token"
+	gatewayYAMLName  = "gateway.yaml"
+)
+
+// gatewayMaxRequestBytes is the forward proxy's maxRequestBodySize (32 MiB).
+const gatewayMaxRequestBytes = 32 << 20
+
+type gatewayFile struct {
+	Insecure     gatewayInsecure `yaml:"insecure_options"`
+	BodyLimits   gatewayLimits   `yaml:"body_limits"`
+	Listeners    []gatewayListen `yaml:"listeners"`
+	FilterChains []gatewayChain  `yaml:"filter_chains"`
+}
+
+type gatewayInsecure struct {
+	AllowUnboundedBody    bool `yaml:"allow_unbounded_body"`
+	AllowPrivateUpstreams bool `yaml:"allow_private_upstreams"`
+	AllowPrivateEndpoints bool `yaml:"allow_private_endpoints"`
+}
+
+type gatewayLimits struct {
+	MaxRequestBytes  int64  `yaml:"max_request_bytes"`
+	MaxResponseBytes *int64 `yaml:"max_response_bytes"` // nil writes null: no cap
+}
+
+type gatewayListen struct {
+	Name         string          `yaml:"name"`
+	Address      string          `yaml:"address"`
+	ForwardProxy map[string]bool `yaml:"forward_proxy"`
+	FilterChains []string        `yaml:"filter_chains"`
+}
+
+type gatewayChain struct {
+	Name    string          `yaml:"name"`
+	Filters []gatewayFilter `yaml:"filters"`
+}
+
+// gatewayFilter puts the filter name first; its options follow in key order.
+type gatewayFilter struct {
+	Filter string         `yaml:"filter"`
+	Opts   map[string]any `yaml:",inline"`
+}
+
+// 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.
+func gatewayYAML(cfg *config.Config, tokenPath string) ([]byte, error) {
+	ingest, err := ingestAddr(cfg.Listener.SessionAPIAddr)
+	if err != nil {
+		return nil, err
+	}
+	f := gatewayFile{
+		Insecure:   gatewayInsecure{true, true, true},
+		BodyLimits: gatewayLimits{MaxRequestBytes: gatewayMaxRequestBytes},
+		Listeners: []gatewayListen{{
+			Name:         "gateway",
+			Address:      cfg.Praxis.ListenAddr(),
+			ForwardProxy: map[string]bool{"enabled": true},
+			FilterChains: []string{"main"},
+		}},
+		FilterChains: []gatewayChain{{Name: "main", Filters: []gatewayFilter{
+			{Filter: "ip_acl", Opts: map[string]any{"allow": []string{"127.0.0.1/32", "::1/128"}}},
+			{Filter: "cortex_attribution", Opts: map[string]any{"token_file": tokenPath}},
+			{Filter: "cortex_observe", Opts: map[string]any{"token_file": tokenPath, "ingest": ingest}},
+			{Filter: "access_log"},
+			{Filter: "router", Opts: map[string]any{"routes": []map[string]string{{"path_prefix": "/", "cluster": "egress"}}}},
+			{Filter: "load_balancer", Opts: map[string]any{"clusters": []map[string]any{{"name": "egress", "endpoints": []string{cfg.Praxis.EgressAddr()}}}}},
+		}}},
+	}
+	var b bytes.Buffer
+	enc := yaml.NewEncoder(&b)
+	enc.SetIndent(2)
+	if err := enc.Encode(f); err != nil {
+		return nil, err
+	}
+	if err := enc.Close(); err != nil {
+		return nil, err
+	}
+	return b.Bytes(), nil
+}
+
+// ingestAddr returns the address the gateway posts rows to: the session API
+// address, with an empty or wildcard host made 127.0.0.1.
+func ingestAddr(sessionAPIAddr string) (string, error) {
+	host, port, err := net.SplitHostPort(sessionAPIAddr)
+	if err != nil {
+		return "", fmt.Errorf("listener.session_api_addr %q: %w", sessionAPIAddr, err)
+	}
+	if a, err := netip.ParseAddr(host); host == "" || (err == nil && a.IsUnspecified()) {
+		host = "127.0.0.1"
+	}
+	return net.JoinHostPort(host, port), nil
+}
+
+// writeGatewayFiles writes gateway.yaml into dir, and a random token when dir has
+// none. Each file has mode 0600 and is written by rename, so a reader never sees a
+// partial file. An existing token is kept: each child reads it once, at its start,
+// so a second supervisor on the same dir that wrote a new token would break the
+// pair of the first one at its next child restart.
+func writeGatewayFiles(dir string, cfg *config.Config) (tokenPath, yamlPath string, err error) {
+	tokenPath = filepath.Join(dir, gatewayTokenName)
+	yamlPath = filepath.Join(dir, gatewayYAMLName)
+	y, err := gatewayYAML(cfg, tokenPath)
+	if err != nil {
+		return "", "", err
+	}
+	if b, err := os.ReadFile(tokenPath); err != nil || len(bytes.TrimSpace(b)) == 0 {
+		raw := make([]byte, 32)
+		if _, err := rand.Read(raw); err != nil {
+			return "", "", err
+		}
+		if err := writeFileAtomic(tokenPath, []byte(hex.EncodeToString(raw))); err != nil {
+			return "", "", err
+		}
+	}
+	if err := writeFileAtomic(yamlPath, y); err != nil {
+		return "", "", err
+	}
+	return tokenPath, yamlPath, nil
+}
+
+// writeFileAtomic writes b to a 0600 temp file next to path, then renames it.
+func writeFileAtomic(path string, b []byte) error {
+	f, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".*")
+	if err != nil {
+		return err
+	}
+	tmp := f.Name()
+	_, err = f.Write(b)
+	if err == nil {
+		err = f.Sync()
+	}
+	if cerr := f.Close(); err == nil {
+		err = cerr
+	}
+	if err == nil {
+		err = os.Rename(tmp, path)
+	}
+	if err != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("write %s: %w", path, err)
+	}
+	return nil
+}