diff --git a/cmd/agentop/cmd_uninstall.go b/cmd/agentop/cmd_uninstall.go index ebe0375b6..c20d34e3e 100644 --- a/cmd/agentop/cmd_uninstall.go +++ b/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 diff --git a/cmd/agentop/cmd_uninstall_test.go b/cmd/agentop/cmd_uninstall_test.go index b9a844993..743bdb4f7 100644 --- a/cmd/agentop/cmd_uninstall_test.go +++ b/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 diff --git a/cmd/agentop/setup_step_binaries.go b/cmd/agentop/setup_step_binaries.go index 1f2d3137b..f0fdd9919 100644 --- a/cmd/agentop/setup_step_binaries.go +++ b/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{} diff --git a/cmd/agentop/setup_step_binaries_test.go b/cmd/agentop/setup_step_binaries_test.go index 7118d6575..1deffd303 100644 --- a/cmd/agentop/setup_step_binaries_test.go +++ b/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") diff --git a/cmd/cortex/fallback.go b/cmd/cortex/fallback.go index 0918ad154..011bc97a6 100644 --- a/cmd/cortex/fallback.go +++ b/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) +} diff --git a/cmd/cortex/fallback_test.go b/cmd/cortex/fallback_test.go index cad9573cc..07c965915 100644 --- a/cmd/cortex/fallback_test.go +++ b/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) } }) diff --git a/cmd/cortex/gateway.go b/cmd/cortex/gateway.go new file mode 100644 index 000000000..f84f39163 --- /dev/null +++ b/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 +} diff --git a/cmd/cortex/gateway_test.go b/cmd/cortex/gateway_test.go new file mode 100644 index 000000000..503505280 --- /dev/null +++ b/cmd/cortex/gateway_test.go @@ -0,0 +1,225 @@ +package main + +import ( + "bytes" + "flag" + "os" + "path/filepath" + "regexp" + "testing" + + "gopkg.in/yaml.v3" + + "github.com/rossoctl/cortex/core/config" +) + +var updateGolden = flag.Bool("update", false, "rewrite testdata goldens") + +func praxisConfig() *config.Config { + return &config.Config{ + Mode: config.ModeProxySidecar, + Listener: config.ListenerConfig{SessionAPIAddr: "127.0.0.1:47601", BindLoopbackOnly: true, ReverseProxyBackend: "127.0.0.1:8000"}, + Praxis: &config.PraxisConfig{Enabled: true}, + } +} + +func TestGatewayYAML_Golden(t *testing.T) { + got, err := gatewayYAML(praxisConfig(), "/CORTEX_DIR/gateway.token") + if err != nil { + t.Fatal(err) + } + golden := filepath.Join("testdata", "gateway.yaml.golden") + if *updateGolden { + if err := os.WriteFile(golden, got, 0o644); err != nil { + t.Fatal(err) + } + } + want, err := os.ReadFile(golden) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, want) { + t.Errorf("gatewayYAML differs from %s (run with -update):\n%s", golden, got) + } +} + +func TestGatewayYAML_Fields(t *testing.T) { + out, err := gatewayYAML(praxisConfig(), "/d/gateway.token") + if err != nil { + t.Fatal(err) + } + var g struct { + Insecure map[string]bool `yaml:"insecure_options"` + BodyLimits map[string]any `yaml:"body_limits"` + Listeners []struct { + Name string `yaml:"name"` + Address string `yaml:"address"` + ForwardProxy map[string]bool `yaml:"forward_proxy"` + FilterChains []string `yaml:"filter_chains"` + } `yaml:"listeners"` + FilterChains []struct { + Name string `yaml:"name"` + Filters []map[string]any `yaml:"filters"` + } `yaml:"filter_chains"` + } + if err := yaml.Unmarshal(out, &g); err != nil { + t.Fatal(err) + } + for _, k := range []string{"allow_unbounded_body", "allow_private_upstreams", "allow_private_endpoints"} { + if !g.Insecure[k] { + t.Errorf("insecure_options.%s = false", k) + } + } + if v, ok := g.BodyLimits["max_request_bytes"]; !ok || v != 33554432 { + t.Errorf("max_request_bytes = %v", v) + } + if v, ok := g.BodyLimits["max_response_bytes"]; !ok || v != nil { + t.Errorf("max_response_bytes = %v (present %v), want null", v, ok) + } + if len(g.Listeners) != 1 { + t.Fatalf("listeners = %d", len(g.Listeners)) + } + l := g.Listeners[0] + if l.Name != "gateway" || l.Address != config.DefaultPraxisListen || !l.ForwardProxy["enabled"] || len(l.FilterChains) != 1 || l.FilterChains[0] != "main" { + t.Errorf("listener = %+v", l) + } + if len(g.FilterChains) != 1 || g.FilterChains[0].Name != "main" { + t.Fatalf("filter_chains = %+v", g.FilterChains) + } + fs := g.FilterChains[0].Filters + var order []any + for _, f := range fs { + order = append(order, f["filter"]) + } + wantOrder := []any{"ip_acl", "cortex_attribution", "cortex_observe", "access_log", "router", "load_balancer"} + if len(order) != len(wantOrder) { + t.Fatalf("filter order = %v", order) + } + for i := range wantOrder { + if order[i] != wantOrder[i] { + t.Fatalf("filter order = %v, want %v", order, wantOrder) + } + } + acl := fs[0] + if allow, _ := acl["allow"].([]any); len(allow) != 2 || allow[0] != "127.0.0.1/32" || allow[1] != "::1/128" { + t.Errorf("ip_acl.allow = %v", acl["allow"]) + } + if _, ok := acl["deny"]; ok { + t.Error("ip_acl has deny") + } + if _, ok := acl["conditions"]; ok { + t.Error("ip_acl has conditions") + } + if fs[1]["token_file"] != "/d/gateway.token" || fs[2]["token_file"] != "/d/gateway.token" { + t.Errorf("token_file = %v, %v", fs[1]["token_file"], fs[2]["token_file"]) + } + if fs[2]["ingest"] != "127.0.0.1:47601" { + t.Errorf("ingest = %v", fs[2]["ingest"]) + } + routes, _ := fs[4]["routes"].([]any) + if len(routes) != 1 { + t.Fatalf("routes = %v", fs[4]["routes"]) + } + if r, _ := routes[0].(map[string]any); r["path_prefix"] != "/" || r["cluster"] != "egress" { + t.Errorf("route = %v", routes[0]) + } + clusters, _ := fs[5]["clusters"].([]any) + if len(clusters) != 1 { + t.Fatalf("clusters = %v", fs[5]["clusters"]) + } + c, _ := clusters[0].(map[string]any) + eps, _ := c["endpoints"].([]any) + if c["name"] != "egress" || len(eps) != 1 || eps[0] != config.DefaultPraxisEgress { + t.Errorf("cluster = %v", clusters[0]) + } +} + +func TestGatewayYAML_BadSessionAddr(t *testing.T) { + cfg := praxisConfig() + cfg.Listener.SessionAPIAddr = "bad" + if _, err := gatewayYAML(cfg, "/t"); err == nil { + t.Error("want an error for a session API address that does not split") + } +} + +func TestIngestAddr(t *testing.T) { + for in, want := range map[string]string{ + ":47601": "127.0.0.1:47601", + "0.0.0.0:1": "127.0.0.1:1", + "[::]:1": "127.0.0.1:1", + "127.0.0.1:1": "127.0.0.1:1", + } { + got, err := ingestAddr(in) + if err != nil || got != want { + t.Errorf("ingestAddr(%q) = %q, %v; want %q", in, got, err, want) + } + } + if _, err := ingestAddr("bad"); err == nil { + t.Error(`ingestAddr("bad") gave no error`) + } +} + +// A second call keeps the token, so a second supervisor on the same dir cannot +// break the first one's pair. An empty token file gets a new token. +func TestWriteGatewayFiles_TokenIs0600AndKept(t *testing.T) { + dir := t.TempDir() + hex64 := regexp.MustCompile(`^[0-9a-f]{64}$`) + var tokens []string + for range 2 { + tok, yml, err := writeGatewayFiles(dir, praxisConfig()) + if err != nil { + t.Fatal(err) + } + if tok != filepath.Join(dir, gatewayTokenName) || yml != filepath.Join(dir, gatewayYAMLName) { + t.Fatalf("paths = %s, %s", tok, yml) + } + for _, p := range []string{tok, yml} { + fi, err := os.Stat(p) + if err != nil { + t.Fatal(err) + } + if m := fi.Mode().Perm(); m != 0o600 { + t.Errorf("%s mode = %o, want 600", p, m) + } + } + b, err := os.ReadFile(tok) + if err != nil { + t.Fatal(err) + } + if !hex64.Match(b) { + t.Errorf("token = %q, want 64 hex chars", b) + } + tokens = append(tokens, string(b)) + y, err := os.ReadFile(yml) + if err != nil { + t.Fatal(err) + } + if !bytes.Contains(y, []byte(tok)) { + t.Errorf("gateway.yaml does not name %s", tok) + } + } + if tokens[0] != tokens[1] { + t.Errorf("the second call changed the token: %q, then %q", tokens[0], tokens[1]) + } + tok := filepath.Join(dir, gatewayTokenName) + if err := os.WriteFile(tok, []byte("\n"), 0o600); err != nil { + t.Fatal(err) + } + if _, _, err := writeGatewayFiles(dir, praxisConfig()); err != nil { + t.Fatal(err) + } + if b, _ := os.ReadFile(tok); !hex64.Match(b) { + t.Errorf("after an empty token file, token = %q, want 64 hex chars", b) + } + ents, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + if len(ents) != 2 { + var names []string + for _, e := range ents { + names = append(names, e.Name()) + } + t.Errorf("dir holds %v, want only the two files", names) + } +} diff --git a/cmd/cortex/go.mod b/cmd/cortex/go.mod index 7ff204783..641e69834 100644 --- a/cmd/cortex/go.mod +++ b/cmd/cortex/go.mod @@ -5,6 +5,7 @@ go 1.26.10 require ( github.com/rossoctl/cortex/core v0.0.0 github.com/rossoctl/cortex/core/storage/redis v0.0.0 + gopkg.in/yaml.v3 v3.0.1 ) require ( @@ -125,7 +126,6 @@ require ( google.golang.org/grpc v1.85.0 // indirect google.golang.org/protobuf v1.36.12 // indirect gopkg.in/ini.v1 v1.67.3 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect oras.land/oras-go/v2 v2.6.2 // indirect ) diff --git a/cmd/cortex/portconflict.go b/cmd/cortex/portconflict.go index 712271d6a..e8d699b0c 100644 --- a/cmd/cortex/portconflict.go +++ b/cmd/cortex/portconflict.go @@ -42,11 +42,12 @@ import ( // crash worth restarting at speed. Nothing outside this package depends on the number. const exitPortInUse = 3 -// cortexBinaries are the basenames this proxy has shipped under. The Cortex remedy — stop +// cortexBinaries are the basenames this proxy has shipped under, and its praxis-mode +// gateway, which holds praxis.listen. The Cortex remedy — stop // the other install — is only right for these: a sidecar that cannot bind :8080 in a pod // has met something entirely unrelated, and must not be sent looking for a second laptop // install it does not have. -var cortexBinaries = map[string]bool{"cortex": true, "authbridge-proxy": true} +var cortexBinaries = map[string]bool{"cortex": true, "authbridge-proxy": true, "cortex-gateway": true} // listenError is a socket that could not be bound: which one, and at what address. The // address is the part a caller cannot otherwise recover — it lives only inside diff --git a/cmd/cortex/portconflict_test.go b/cmd/cortex/portconflict_test.go index 38be02565..46801ab80 100644 --- a/cmd/cortex/portconflict_test.go +++ b/cmd/cortex/portconflict_test.go @@ -168,3 +168,15 @@ func TestExitPortInUse_IsNotAnOrdinaryFailure(t *testing.T) { t.Errorf("exitPortInUse = %d, which a crash or a usage error also leaves", exitPortInUse) } } + +// A holder by any name this proxy or its gateway ships under gets the Cortex remedy. +func TestCortexBinaries(t *testing.T) { + for name, want := range map[string]bool{ + "cortex": true, "authbridge-proxy": true, "cortex-gateway": true, + "cortex-envoy": false, "nginx": false, + } { + if got := cortexBinaries[name]; got != want { + t.Errorf("cortexBinaries[%q] = %v, want %v", name, got, want) + } + } +} diff --git a/cmd/cortex/restart_test.go b/cmd/cortex/restart_test.go index 0ad88b314..29c635404 100644 --- a/cmd/cortex/restart_test.go +++ b/cmd/cortex/restart_test.go @@ -2,7 +2,9 @@ package main import ( "bytes" + "errors" "fmt" + "io" "net" "net/http" "net/url" @@ -12,11 +14,16 @@ import ( "path/filepath" "regexp" "slices" + "strconv" "strings" "sync" "syscall" "testing" "time" + + "gopkg.in/yaml.v3" + + "github.com/rossoctl/cortex/core/config" ) // runMainEnv makes a re-executed copy of this test binary run something other than the @@ -25,6 +32,10 @@ import ( const runMainEnv = "CORTEX_TEST_RUN" func TestMain(m *testing.M) { + // The supervisor gives the gateway an empty env, so its copy cannot read runMainEnv. + if filepath.Base(os.Args[0]) == gatewayBinary { + fakeGateway(os.Args[1:]) + } switch os.Getenv(runMainEnv) { case "main": main() @@ -40,10 +51,63 @@ func TestMain(m *testing.M) { fmt.Fprintln(os.Stderr, "stuck child: ignoring SIGTERM") time.Sleep(time.Minute) os.Exit(0) + case "idle-child": + // A proxy that starts at once and stops on SIGTERM, for the tests about the + // gateway beside it. + if slices.Contains(os.Args[1:], "-supervise") { + main() + os.Exit(0) + } + fmt.Fprintf(os.Stderr, "idle child pid %d\n", os.Getpid()) + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGTERM) + <-sig + os.Exit(0) } os.Exit(m.Run()) } +// fakeGateway stands in for cortex-gateway: it listens on the gateway listener of +// its -c file and, with --exit-on-stdin-eof, exits when stdin closes. +func fakeGateway(args []string) { + var path string + for i, a := range args { + if a == "-c" && i+1 < len(args) { + path = args[i+1] + } + } + b, err := os.ReadFile(path) + var f gatewayFile + if err == nil { + err = yaml.Unmarshal(b, &f) + } + if err == nil && len(f.Listeners) == 0 { + err = errors.New("no listener") + } + var ln net.Listener + if err == nil { + ln, err = net.Listen("tcp", f.Listeners[0].Address) + } + if err != nil { + fmt.Fprintf(os.Stderr, "fake gateway: %v\n", err) + os.Exit(1) + } + defer ln.Close() + fmt.Fprintf(os.Stderr, "fake gateway pid %d\n", os.Getpid()) + if slices.Contains(args, "--exit-on-stdin-eof") { + _, _ = io.Copy(io.Discard, os.Stdin) + os.Exit(0) + } + // Block in Accept, not select {}: with no other goroutine, select {} is a fatal deadlock. + for { + c, err := ln.Accept() + if err != nil { + os.Exit(1) + } + _ = c.Close() + } +} + // lockedBuffer collects a child's output while the test reads it. type lockedBuffer struct { mu sync.Mutex @@ -145,9 +209,15 @@ type child struct { } func startChild(t *testing.T, home, run string, args ...string) *child { + t.Helper() + return startChildAt(t, os.Args[0], home, run, args...) +} + +// startChildAt is startChild with a copy of this test binary at path. +func startChildAt(t *testing.T, path, home, run string, args ...string) *child { t.Helper() c := &child{out: &lockedBuffer{}, done: make(chan error, 1)} - c.cmd = exec.Command(os.Args[0], args...) //nolint:gosec // this test binary + c.cmd = exec.Command(path, args...) //nolint:gosec // this test binary c.cmd.Env = childEnv(home, run) c.cmd.Stdout, c.cmd.Stderr = c.out, c.out // Its own process group, as under launchd, so cleanup can end a supervisor's child too: @@ -343,3 +413,188 @@ func TestSupervisorEndsAStuckProxyInsideLaunchdsWindow(t *testing.T) { t.Errorf("supervisor exited without killing the stuck proxy:\n%s", sup.out.String()) } } + +// praxisRig is a bin dir with this test binary copied in as cortex and as +// cortex-gateway, and a praxis-mode config. Copies, not links: os.Executable +// resolves a symlink on Linux, and the supervisor finds the gateway next to itself. +type praxisRig struct { + home, bin, config, listen string +} + +func newPraxisRig(t *testing.T, praxis bool) praxisRig { + t.Helper() + r := praxisRig{home: t.TempDir(), listen: freeAddr(t)} + r.bin = filepath.Join(r.home, "bin") + dir := filepath.Join(r.home, cortexDirName) + for _, d := range []string{r.bin, dir} { + if err := os.MkdirAll(d, 0o700); err != nil { + t.Fatal(err) + } + } + self, err := os.ReadFile(os.Args[0]) + if err != nil { + t.Fatal(err) + } + for _, n := range []string{"cortex", gatewayBinary} { + if err := os.WriteFile(filepath.Join(r.bin, n), self, 0o700); err != nil { + t.Fatal(err) + } + } + r.config = filepath.Join(dir, "config.yaml") + writeProxyConfig(t, r.config, anyPort) + praxisBlock := fmt.Sprintf("praxis:\n enabled: %v\n listen: %s\n", praxis, r.listen) + b, _ := os.ReadFile(r.config) + if err := os.WriteFile(r.config, append(b, praxisBlock...), 0o600); err != nil { + t.Fatal(err) + } + if _, err := config.Load(r.config); err != nil { + t.Fatalf("the fixture config does not load: %v", err) + } + return r +} + +func (r praxisRig) supervise(t *testing.T) *child { + t.Helper() + return startChildAt(t, filepath.Join(r.bin, "cortex"), r.home, "idle-child", "-supervise", "-config", r.config) +} + +var ( + idleChildPid = regexp.MustCompile(`idle child pid (\d+)`) + fakeGatewayPid = regexp.MustCompile(`fake gateway pid (\d+)`) +) + +// pids waits until out holds at least n lines that re matches, and returns their pids. +func (c *child) pids(t *testing.T, re *regexp.Regexp, n int) []int { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for { + var pids []int + for _, m := range re.FindAllStringSubmatch(c.out.String(), -1) { + p, _ := strconv.Atoi(m[1]) + pids = append(pids, p) + } + if len(pids) >= n { + return pids + } + if time.Now().After(deadline) { + t.Fatalf("waited for %d lines matching %s:\n%s", n, re, c.out.String()) + } + time.Sleep(20 * time.Millisecond) + } +} + +// alive reports whether pid runs. A zombie has exited: in a container whose PID 1 +// does not reap (CI's job container), an orphan stays a zombie, and kill(pid, 0) +// still succeeds on it. +func alive(pid int) bool { + if b, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); err == nil { + if i := bytes.LastIndexByte(b, ')'); i >= 0 && i+2 < len(b) { + return b[i+2] != 'Z' + } + } + return syscall.Kill(pid, 0) == nil +} + +func waitGone(t *testing.T, pid int, within time.Duration) { + t.Helper() + deadline := time.Now().Add(within) + for alive(pid) { + if time.Now().After(deadline) { + t.Fatalf("pid %d still runs after %s", pid, within) + } + time.Sleep(20 * time.Millisecond) + } +} + +// A cortex crash restarts cortex only: the gateway beside it keeps its pid. +func TestSupervisor_CortexRestartKeepsGateway(t *testing.T) { + r := newPraxisRig(t, true) + sup := r.supervise(t) + cortex := sup.pids(t, idleChildPid, 1)[0] + gw := sup.pids(t, fakeGatewayPid, 1)[0] + + if err := syscall.Kill(cortex, syscall.SIGKILL); err != nil { + t.Fatal(err) + } + if again := sup.pids(t, idleChildPid, 2); again[1] == cortex { + t.Fatalf("cortex restarted with the same pid %d", cortex) + } + if gws := sup.pids(t, fakeGatewayPid, 1); len(gws) != 1 || !alive(gw) { + t.Fatalf("the gateway changed after a cortex crash: pids %v, %d alive = %v", gws, gw, alive(gw)) + } + + sent := time.Now() + if err := sup.cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatal(err) + } + sup.waitExit(t, sent, 5*time.Second) + newCortex := sup.pids(t, idleChildPid, 2)[1] + waitGone(t, newCortex, 5*time.Second-time.Since(sent)) + waitGone(t, gw, 5*time.Second-time.Since(sent)) +} + +// A SIGKILL of the supervisor closes the gateway's stdin, so the gateway exits. +func TestSupervisor_KillLeavesNoGateway(t *testing.T) { + r := newPraxisRig(t, true) + sup := r.supervise(t) + sup.pids(t, idleChildPid, 1) + gw := sup.pids(t, fakeGatewayPid, 1)[0] + if err := sup.cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + // The idle cortex child is left behind; startChild's cleanup kills its group. + waitGone(t, gw, 5*time.Second) +} + +// Direct mode starts no gateway and writes no gateway file, even with no +// cortex-gateway binary next to cortex. +func TestSupervisor_DirectModeStartsNoGateway(t *testing.T) { + r := newPraxisRig(t, false) + if err := os.Remove(filepath.Join(r.bin, gatewayBinary)); err != nil { + t.Fatal(err) + } + sup := r.supervise(t) + sup.pids(t, idleChildPid, 1) + time.Sleep(500 * time.Millisecond) + if out := sup.out.String(); strings.Contains(out, gatewayBinary) { + t.Errorf("direct mode logged the gateway:\n%s", out) + } + for _, n := range []string{gatewayTokenName, gatewayYAMLName} { + if _, err := os.Stat(filepath.Join(filepath.Dir(r.config), n)); !errors.Is(err, os.ErrNotExist) { + t.Errorf("%s: err = %v, want not exist", n, err) + } + } +} + +// A gateway that cannot start (here, a missing binary) does not stop cortex: the +// supervisor says so at ERROR one time, retries with backoff, and still stops on SIGTERM. +func TestSupervisor_GatewayStartFailureKeepsCortex(t *testing.T) { + r := newPraxisRig(t, true) + gwPath := filepath.Join(r.bin, gatewayBinary) + if err := os.Remove(gwPath); err != nil { + t.Fatal(err) + } + sup := r.supervise(t) + cortex := sup.pids(t, idleChildPid, 1)[0] + // Past the first retry (superviseRestartDelay), so a second failure has happened. + time.Sleep(superviseRestartDelay + time.Second) + if !alive(cortex) { + t.Fatalf("cortex pid %d stopped after the gateway failed to start:\n%s", cortex, sup.out.String()) + } + var errLines []string + for _, l := range strings.Split(sup.out.String(), "\n") { + if strings.Contains(l, "cannot start the child") && strings.Contains(l, "ERROR") { + errLines = append(errLines, l) + } + } + if len(errLines) != 1 || !strings.Contains(errLines[0], gwPath) { + t.Fatalf("want one ERROR line that names %s, got %q:\n%s", gwPath, errLines, sup.out.String()) + } + + sent := time.Now() + if err := sup.cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatal(err) + } + sup.waitExit(t, sent, 5*time.Second) + waitGone(t, cortex, 5*time.Second-time.Since(sent)) +} diff --git a/cmd/cortex/supervise.go b/cmd/cortex/supervise.go index d21543497..d25f8c325 100644 --- a/cmd/cortex/supervise.go +++ b/cmd/cortex/supervise.go @@ -3,13 +3,18 @@ package main import ( "errors" "fmt" + "io" "log/slog" "os" "os/exec" "os/signal" + "path/filepath" "slices" + "sync" "syscall" "time" + + "github.com/rossoctl/cortex/core/config" ) // Supervision exists because launchd cannot be relied on for it. @@ -157,8 +162,59 @@ func portInUseExit(werr error) bool { return errors.As(werr, &ee) && ee.ExitCode() == exitPortInUse } +// cortexChild names the proxy child in the log and in its childSpec. +const cortexChild = "cortex" + +// childSpec is one child the supervisor runs, each in its own restart loop, so an +// exit of one child does not touch the other. +type childSpec struct { + name string + path string + args []string + env []string // nil = inherit + stdinPipe bool // the supervisor keeps the write end; closing it stops the child + fallback *fallbackServer + fallbackAddr string // cortex only, with fallback +} + +// supervisorChildren returns the children to run. Direct mode is cortex alone, with +// today's args and env, and touches no file. Praxis mode first writes the gateway +// files next to the config, then adds cortex-gateway, found next to this binary. The +// gateway gets an empty env, so no proxy variable can reach it. +// +// When the gateway cannot run, cortex runs alone: it then refuses the praxis config +// at each start and logs why, and the fallback keeps the forward address. Ending the +// supervisor instead would leave nothing until the next login. +func supervisorChildren(cfg *config.Config, configPath, self string, args []string, + write func(dir string, cfg *config.Config) (string, string, error)) []childSpec { + cortex := childSpec{name: cortexChild, path: self, args: args} + if cfg == nil || !cfg.PraxisEnabled() { + return []childSpec{cortex} + } + // The cortex child refuses this config at each start. Do not run a gateway for it. + if err := config.Validate(cfg); err != nil { + slog.Error("supervisor: the config is not valid, so cortex-gateway does not start", "error", err) + return []childSpec{cortex} + } + tokenPath, yamlPath, err := write(filepath.Dir(configPath), cfg) + if err != nil { + slog.Error("supervisor: cannot write the gateway files, so cortex-gateway does not start", "error", err) + return []childSpec{cortex} + } + cortex.env = append(os.Environ(), gatewayTokenEnv+"="+tokenPath) + gateway := childSpec{ + name: gatewayBinary, + path: filepath.Join(filepath.Dir(self), gatewayBinary), + args: []string{"-c", yamlPath, "--exit-on-stdin-eof"}, + env: []string{}, + stdinPipe: true, + } + return []childSpec{cortex, gateway} +} + // runSupervisor re-executes this binary without the supervise flag and restarts it -// whenever it exits, until the supervisor is asked to stop. +// whenever it exits, until the supervisor is asked to stop. In praxis mode it runs +// cortex-gateway beside it, in a loop of its own. func runSupervisor(flagName string) error { self, err := os.Executable() if err != nil { @@ -176,35 +232,98 @@ func runSupervisor(flagName string) error { return errors.New("internal: supervise flag survived argument filtering") } - // Forward termination to the child so `launchctl bootout` / `systemctl stop` + // Forward termination to the children so `launchctl bootout` / `systemctl stop` // takes the whole tree down; otherwise the proxy is orphaned and keeps the ports, // which would look exactly like the "stop that does not stop" this replaces. stopping := make(chan os.Signal, 1) signal.Notify(stopping, syscall.SIGTERM, syscall.SIGINT) + configPath := supervisorConfigPath(args) + cfg := supervisorConfig(configPath) + // The fallback holds the forward address whenever no child does, so a routed agent // gets a 502 that names Cortex instead of a refused connection it cannot trace. It // is opened here because nothing is serving yet, and closed before every start. See // fallback.go. It deliberately does not cover `agentop service stop`: that stops // this supervisor too, and that path already warns, having been asked for. fallback := &fallbackServer{} - fallbackAddr := supervisorForwardAddr(args) + fallbackAddr := supervisorForwardAddr(cfg) fallback.open(fallbackAddr) defer fallback.close() + specs := supervisorChildren(cfg, configPath, self, args, writeGatewayFiles) + specs[0].fallback, specs[0].fallbackAddr = fallback, fallbackAddr + + // Each child stops at the same time, so the whole stop stays inside launchd's 5s. + stop := make(chan struct{}) + errs := make(chan error, len(specs)) + var wg sync.WaitGroup + for _, spec := range specs { + wg.Add(1) + go func() { + defer wg.Done() + errs <- superviseChild(spec, stop) + }() + } + select { + case sig := <-stopping: + slog.Info("supervisor: stopping", "signal", sig.String()) + case err = <-errs: // only a cortex start failure ends a loop by itself + } + close(stop) + wg.Wait() + return err +} + +// superviseChild runs one child and restarts it whenever it exits, until stop closes. +// A start failure of cortex ends the supervisor, as before praxis mode; a start +// failure of the gateway (a missing binary, say) is retried with the normal backoff, +// and cortex keeps running. +func superviseChild(spec childSpec, stop <-chan struct{}) error { + log := slog.With("child", spec.name) policy := &restartPolicy{delay: superviseRestartDelay} + startFailing := false for { // Hand the address to the child. Nothing else frees it, so this is what lets the // child bind at all. - fallback.close() + if spec.fallback != nil { + spec.fallback.close() + } - cmd := exec.Command(self, args...) //nolint:gosec // our own binary, filtered args + cmd := exec.Command(spec.path, spec.args...) //nolint:gosec // our own binaries, filtered args + cmd.Env = spec.env cmd.Stdout, cmd.Stderr, cmd.Stdin = os.Stdout, os.Stderr, nil + // The write end is close-on-exec, so only this process holds it, and the kernel + // closes it however this process exits, SIGKILL included. + var stdin io.Closer + var err error + if spec.stdinPipe { + stdin, err = cmd.StdinPipe() + } + if err == nil { + err = cmd.Start() + } start := time.Now() - if err := cmd.Start(); err != nil { - return fmt.Errorf("starting the proxy: %w", err) + if err != nil { + if spec.name == cortexChild { + return fmt.Errorf("starting the proxy: %w", err) + } + note := policy.exited(false, false) + failed := log.Debug + if !startFailing { + failed = log.Error + } + startFailing = true + failed("supervisor: cannot start the child; retrying", "path", spec.path, "err", err, "restart_in", note.delay) + select { + case <-stop: + return nil + case <-time.After(note.delay): + } + continue } - slog.Info("supervisor: proxy started", "pid", cmd.Process.Pid) + startFailing = false + log.Info("supervisor: child started", "pid", cmd.Process.Pid) done := make(chan error, 1) go func() { done <- cmd.Wait() }() @@ -221,14 +340,19 @@ func runSupervisor(flagName string) error { wait: for { select { - case sig := <-stopping: + case <-stop: healthy.Stop() - slog.Info("supervisor: stopping, forwarding signal to the proxy", "signal", sig.String()) - _ = cmd.Process.Signal(syscall.SIGTERM) //nolint:errcheck + if stdin != nil { + log.Info("supervisor: stopping, closing the child's stdin") + _ = stdin.Close() + } else { + log.Info("supervisor: stopping, forwarding SIGTERM to the child") + _ = cmd.Process.Signal(syscall.SIGTERM) //nolint:errcheck + } select { case <-done: case <-time.After(superviseStopGrace): - slog.Warn("supervisor: proxy did not exit in time; killing it", "waited", superviseStopGrace) + log.Warn("supervisor: child did not exit in time; killing it", "waited", superviseStopGrace) _ = cmd.Process.Kill() //nolint:errcheck <-done } @@ -237,7 +361,7 @@ func runSupervisor(flagName string) error { case <-healthy.C: wasHealthy = true if policy.healthy() { - slog.Info("supervisor: the proxy is up; the address it could not bind is free again") + log.Info("supervisor: the proxy is up; the address it could not bind is free again") } case werr = <-done: @@ -248,17 +372,20 @@ func runSupervisor(flagName string) error { } // The child is gone, so the address is free and nothing would answer on it. - fallback.open(fallbackAddr) + if spec.fallback != nil { + spec.fallback.open(spec.fallbackAddr) + } - note := policy.exited(portInUseExit(werr), wasHealthy) + // Only cortex uses exitPortInUse; the gateway's status gets the normal backoff. + note := policy.exited(spec.name == cortexChild && portInUseExit(werr), wasHealthy) if note.explain { - slog.Error("supervisor: the proxy cannot bind an address it needs, so it exits as soon as it starts", + log.Error("supervisor: the proxy cannot bind an address it needs, so it exits as soon as it starts", "detail", "the error the proxy logged just above names the address and the process holding it", "doing", "retrying, so the proxy starts by itself once that address is free", "not_doing", "giving up — the holder may stop at any moment, and nothing restarts this supervisor before the next login") } if note.slowed { - slog.Warn("supervisor: the address is still held; retrying less often from here", + log.Warn("supervisor: the address is still held; retrying less often from here", "attempts", note.attempts, "retry_every_at_most", superviseBlockedMaxDelay) } // Once the condition has been stated and nothing has changed, the per-restart @@ -266,16 +393,16 @@ func runSupervisor(flagName string) error { // day, which is what buried the one line that explained anything. The child still // logs its own ERROR on every attempt, so the log tail `agentop service status` // prints always carries the diagnosis. - restarting := slog.Warn + restarting := log.Warn if note.quiet { - restarting = slog.Debug + restarting = log.Debug } - restarting("supervisor: proxy exited; restarting", + restarting("supervisor: child exited; restarting", "ran", ran.Round(time.Millisecond), "err", werr, "restart_in", note.delay) select { - case sig := <-stopping: - slog.Info("supervisor: stopping instead of restarting", "signal", sig.String()) + case <-stop: + log.Info("supervisor: stopping instead of restarting") return nil case <-time.After(note.delay): } diff --git a/cmd/cortex/supervise_test.go b/cmd/cortex/supervise_test.go index 95527e02e..f1e647557 100644 --- a/cmd/cortex/supervise_test.go +++ b/cmd/cortex/supervise_test.go @@ -2,10 +2,15 @@ package main import ( "errors" + "os" "os/exec" + "path/filepath" + "slices" "strconv" "testing" "time" + + "github.com/rossoctl/cortex/core/config" ) // The policy is the whole of #1122's behaviour change, and it is deliberately separate @@ -197,3 +202,129 @@ func TestSuperviseDelays_AreOrdered(t *testing.T) { t.Error("a child counts as healthy sooner than the wait before starting it") } } + +// Direct mode is the default, and it must not change: one child, today's args and +// env, no gateway file and no lookup of the gateway binary. A praxis config that +// does not validate also runs cortex alone. +func TestSupervisorChildren_DirectModeHasOnlyCortex(t *testing.T) { + args := []string{"--config", "/x/config.yaml", "-local"} + noWrite := func(string, *config.Config) (string, string, error) { + t.Error("direct mode wrote the gateway files") + return "", "", nil + } + // Praxis on, but cortex refuses the config: a gateway would serve on 0.0.0.0. + notValid := praxisConfig() + notValid.Praxis.Listen = "0.0.0.0:47605" + for name, cfg := range map[string]*config.Config{ + "praxis off": {Praxis: &config.PraxisConfig{Enabled: false}}, + "no config": nil, + "praxis on, not valid": notValid, + } { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + specs := supervisorChildren(cfg, filepath.Join(dir, "config.yaml"), "/bin/cortex", args, noWrite) + if len(specs) != 1 || specs[0].name != "cortex" { + t.Fatalf("specs = %+v, want only cortex", specs) + } + s := specs[0] + if s.path != "/bin/cortex" || !slices.Equal(s.args, args) || s.env != nil || s.stdinPipe { + t.Errorf("cortex spec = %+v, want today's path, args and inherited env", s) + } + if ents, _ := os.ReadDir(dir); len(ents) != 0 { + t.Errorf("files in the config dir: %v", ents) + } + }) + } +} + +// Gateway files that cannot be written leave cortex alone, with today's env. It must +// not end the supervisor: nothing would then restart anything before the next login. +func TestSupervisorChildren_WriteFailureRunsCortexAlone(t *testing.T) { + failWrite := func(string, *config.Config) (string, string, error) { + return "", "", errors.New("read-only dir") + } + specs := supervisorChildren(praxisConfig(), "/x/config.yaml", "/bin/cortex", nil, failWrite) + if len(specs) != 1 || specs[0].name != "cortex" || specs[0].env != nil { + t.Fatalf("specs = %+v, want only cortex with the inherited env", specs) + } +} + +func TestSupervisorChildren_PraxisAddsGateway(t *testing.T) { + dir := t.TempDir() + cfg := praxisConfig() + specs := supervisorChildren(cfg, filepath.Join(dir, "config.yaml"), "/opt/bin/cortex", []string{"--config", "c"}, writeGatewayFiles) + if len(specs) != 2 { + t.Fatalf("specs = %+v, want cortex and the gateway", specs) + } + c, g := specs[0], specs[1] + tokenEnv := gatewayTokenEnv + "=" + filepath.Join(dir, gatewayTokenName) + if c.name != "cortex" || len(c.env) == 0 || c.env[len(c.env)-1] != tokenEnv { + t.Errorf("cortex spec = %+v, want env ending in %s", c, tokenEnv) + } + wantArgs := []string{"-c", filepath.Join(dir, gatewayYAMLName), "--exit-on-stdin-eof"} + if g.name != gatewayBinary || g.path != "/opt/bin/cortex-gateway" || !slices.Equal(g.args, wantArgs) { + t.Errorf("gateway spec = %+v, want path /opt/bin/cortex-gateway and args %q", g, wantArgs) + } + if g.env == nil || len(g.env) != 0 || !g.stdinPipe { + t.Errorf("gateway env = %#v, stdinPipe = %v; want an empty non-nil env and a stdin pipe", g.env, g.stdinPipe) + } + for _, n := range []string{gatewayTokenName, gatewayYAMLName} { + if _, err := os.Stat(filepath.Join(dir, n)); err != nil { + t.Errorf("%s: %v", n, err) + } + } +} + +// The supervisor must read the config file the cortex child reads. With --local (or its +// --demo alias) and no --config, that is ~/.cortex/config.yaml. +func TestSupervisorConfigPath_LocalReadsTheChildsConfig(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + local := filepath.Join(home, cortexDirName, localConfigName) + for _, tc := range []struct { + args []string + want string + }{ + {[]string{"-local"}, local}, + {[]string{"--local"}, local}, + {[]string{"--local=true"}, local}, + {[]string{"-local=1"}, local}, + {[]string{"--demo"}, local}, + {[]string{"-demo=true"}, local}, + {[]string{"--local=false"}, ""}, + {[]string{"--local", "--local=false"}, ""}, + {[]string{"--local=false", "-local"}, local}, + {[]string{"--config", "/etc/c.yaml"}, "/etc/c.yaml"}, + {[]string{"-ca-dir", "/d"}, ""}, + {[]string{"local"}, ""}, + {nil, ""}, + } { + if got := supervisorConfigPath(tc.args); got != tc.want { + t.Errorf("supervisorConfigPath(%q) = %q, want %q", tc.args, got, tc.want) + } + } +} + +// The reviewer's repro: `cortex --local --supervise` with praxis.enabled in +// ~/.cortex/config.yaml runs the gateway beside cortex. +func TestSupervisorChildren_LocalPraxisAddsGateway(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + dir := filepath.Join(home, cortexDirName) + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + yaml := "mode: proxy-sidecar\nlistener:\n session_api_addr: 127.0.0.1:47601\n bind_loopback_only: true\n reverse_proxy_backend: 127.0.0.1:8000\npraxis:\n enabled: true\n" + if err := os.WriteFile(filepath.Join(dir, localConfigName), []byte(yaml), 0o600); err != nil { + t.Fatal(err) + } + args := []string{"--local"} + path := supervisorConfigPath(args) + specs := supervisorChildren(supervisorConfig(path), path, "/opt/bin/cortex", args, writeGatewayFiles) + if len(specs) != 2 || specs[1].name != gatewayBinary { + t.Fatalf("specs = %+v, want cortex and the gateway", specs) + } + if _, err := os.Stat(filepath.Join(dir, gatewayTokenName)); err != nil { + t.Errorf("gateway token not in ~/.cortex: %v", err) + } +} diff --git a/cmd/cortex/testdata/gateway.yaml.golden b/cmd/cortex/testdata/gateway.yaml.golden new file mode 100644 index 000000000..853e281a9 --- /dev/null +++ b/cmd/cortex/testdata/gateway.yaml.golden @@ -0,0 +1,36 @@ +insecure_options: + allow_unbounded_body: true + allow_private_upstreams: true + allow_private_endpoints: true +body_limits: + max_request_bytes: 33554432 + max_response_bytes: null +listeners: + - name: gateway + address: 127.0.0.1:47605 + forward_proxy: + enabled: true + filter_chains: + - main +filter_chains: + - name: main + filters: + - filter: ip_acl + allow: + - 127.0.0.1/32 + - ::1/128 + - filter: cortex_attribution + token_file: /CORTEX_DIR/gateway.token + - filter: cortex_observe + ingest: 127.0.0.1:47601 + token_file: /CORTEX_DIR/gateway.token + - filter: access_log + - filter: router + routes: + - cluster: egress + path_prefix: / + - filter: load_balancer + clusters: + - endpoints: + - 127.0.0.1:47606 + name: egress diff --git a/core/config/config.go b/core/config/config.go index 38cab768f..1a3123693 100644 --- a/core/config/config.go +++ b/core/config/config.go @@ -335,10 +335,37 @@ func (b *TLSBridgeConfig) Validate() error { return nil } -// PraxisConfig is the top-level `praxis:` block. listen and egress land with -// their callers (#1720, #1721). +// Default loopback addresses of the praxis-mode listeners. +const ( + DefaultPraxisListen = "127.0.0.1:47605" + DefaultPraxisEgress = "127.0.0.1:47606" +) + +// PraxisConfig is the top-level `praxis:` block. Listen is cortex-gateway's +// listener. Egress is the address gateway.yaml routes to; the egress listener +// itself lands with #1721. type PraxisConfig struct { - Enabled bool `yaml:"enabled" json:"enabled"` + Enabled bool `yaml:"enabled" json:"enabled"` + Listen string `yaml:"listen,omitempty" json:"listen,omitempty"` + Egress string `yaml:"egress,omitempty" json:"egress,omitempty"` +} + +// ListenAddr returns praxis.listen, or DefaultPraxisListen when it is empty or +// the block is nil. +func (p *PraxisConfig) ListenAddr() string { + if p == nil || p.Listen == "" { + return DefaultPraxisListen + } + return p.Listen +} + +// EgressAddr returns praxis.egress, or DefaultPraxisEgress when it is empty or +// the block is nil. +func (p *PraxisConfig) EgressAddr() string { + if p == nil || p.Egress == "" { + return DefaultPraxisEgress + } + return p.Egress } // PraxisEnabled reports whether praxis mode is on. False on a nil block. diff --git a/core/config/praxis_test.go b/core/config/praxis_test.go index 961af9ffd..c6e07b115 100644 --- a/core/config/praxis_test.go +++ b/core/config/praxis_test.go @@ -63,3 +63,56 @@ func praxisConfig() *Config { c.Listener.ReverseProxyBackend = "127.0.0.1:8000" return c } + +func TestPraxis_AddrDefaults(t *testing.T) { + var nilBlock *PraxisConfig + for _, p := range []*PraxisConfig{nilBlock, {}} { + if got := p.ListenAddr(); got != DefaultPraxisListen { + t.Errorf("ListenAddr() = %q, want %q", got, DefaultPraxisListen) + } + if got := p.EgressAddr(); got != DefaultPraxisEgress { + t.Errorf("EgressAddr() = %q, want %q", got, DefaultPraxisEgress) + } + } + p := &PraxisConfig{Listen: "127.0.0.1:1", Egress: "127.0.0.1:2"} + if p.ListenAddr() != "127.0.0.1:1" || p.EgressAddr() != "127.0.0.1:2" { + t.Errorf("set fields not returned: %q %q", p.ListenAddr(), p.EgressAddr()) + } +} + +func TestPraxis_AddrMustBeLoopback(t *testing.T) { + for _, tc := range []struct { + addr string + ok bool + }{ + {"0.0.0.0:47605", false}, + {"localhost:47605", false}, + {"127.0.0.1", false}, + {"127.0.0.1:0", false}, + {"10.0.0.1:1", false}, + {"127.0.0.1:1", true}, + {"[::1]:2", true}, + } { + for _, field := range []string{"praxis.listen", "praxis.egress"} { + c := praxisConfig() + c.Listener.BindLoopbackOnly = true + if field == "praxis.listen" { + c.Praxis.Listen = tc.addr + } else { + c.Praxis.Egress = tc.addr + } + err := Validate(c) + if tc.ok && err != nil { + t.Errorf("%s %q: Validate = %v, want nil", field, tc.addr, err) + } + if !tc.ok && (err == nil || !strings.Contains(err.Error(), field)) { + t.Errorf("%s %q: Validate = %v, want an error naming %s", field, tc.addr, err, field) + } + // The checks run only in praxis mode. + c.Praxis.Enabled = false + if err := Validate(c); err != nil { + t.Errorf("%s %q with praxis off: Validate = %v, want nil", field, tc.addr, err) + } + } + } +} diff --git a/core/config/validate.go b/core/config/validate.go index ad33cdc19..7f5dc8da3 100644 --- a/core/config/validate.go +++ b/core/config/validate.go @@ -3,6 +3,7 @@ package config import ( "fmt" "log/slog" + "net/netip" "github.com/rossoctl/cortex/core/cost/pricing" ) @@ -93,6 +94,22 @@ func validatePraxis(cfg *Config) error { if !cfg.Session.SessionEnabled() { return fmt.Errorf("praxis.enabled: true cannot be used with session.enabled: false: Cortex ingests the gateway's rows into the session store") } + if err := validatePraxisAddr("praxis.listen", cfg.Praxis.Listen); err != nil { + return err + } + return validatePraxisAddr("praxis.egress", cfg.Praxis.Egress) +} + +// validatePraxisAddr accepts an empty addr (the default applies) or a literal +// loopback host:port with a non-zero port. +func validatePraxisAddr(field, addr string) error { + if addr == "" { + return nil + } + ap, err := netip.ParseAddrPort(addr) + if err != nil || !ap.Addr().IsLoopback() || ap.Port() == 0 { + return fmt.Errorf("%s: %q must be a loopback IP and a non-zero port, such as %q", field, addr, "127.0.0.1:47605") + } return nil } diff --git a/docs/laptop-service.md b/docs/laptop-service.md index 76b8e1798..2e615da82 100644 --- a/docs/laptop-service.md +++ b/docs/laptop-service.md @@ -37,6 +37,20 @@ agents added mid-session — verified across `KeepAlive`, `StartInterval` and `RunAtLoad` — so the supervisor is what makes crash recovery work. On Linux there is one process; systemd handles it. +In praxis mode (`praxis.enabled: true`), the supervisor has a second child, +`cortex-gateway`, so you see three processes. At each start the supervisor writes +`~/.cortex/gateway.token` and `~/.cortex/gateway.yaml`. A restart of the proxy does +not restart the gateway. When the supervisor exits, the gateway exits too. The +supervisor reads `praxis.*` only when it starts, so after you change `praxis.*`, run +`agentop service restart`. Until then, the next restart of the proxy fails at each +try. + +**On Linux, do not set `praxis.enabled` for the service yet.** The systemd unit does +not use `--supervise`, so nothing starts the gateway or writes `gateway.token`. Then +`cortex` refuses to start (`praxis.enabled needs CORTEX_GATEWAY_TOKEN_FILE …`) and +systemd stops after five tries. A later step (#1722) makes the unit supervise in +praxis mode. + ## Linux: staying up after you log out On Linux, Cortex is a systemd **user** unit, which runs under your own systemd @@ -1005,7 +1019,7 @@ It unroutes Claude Code, OpenCode, IBM Bob and the bob shell function where Cort routed them, restarting OpenCode's background service when it finds it running. It stops and removes the service, or the background proxy. It 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`. A step that fails is reported +`cortex`, `cortex-session-dump` and `cortex-gateway` from `~/.local/bin`. A step that fails is reported with its fix, the rest still run, and the end lists what was left behind. `--purge` also deletes `~/.cortex`: config, CA, logs, cost history, session history, @@ -1125,5 +1139,5 @@ variables. Remove the ones that point at Cortex with the `opencode` CLI, as [Undoing it by hand](agents/opencode.md#undoing-it-by-hand) shows, also before `rm -rf ~/.cortex`. -Last, the files setup installed: `rm -f ~/.local/bin/cortex ~/.local/bin/cortex-session-dump`, +Last, the files setup installed: `rm -f ~/.local/bin/cortex ~/.local/bin/cortex-session-dump ~/.local/bin/cortex-gateway`, and the two lines it marked in your shell profile, the first starting `# added by Cortex`. diff --git a/docs/superpowers/plans/2026-11-09-praxis-mode-3-supervise-gateway.md b/docs/superpowers/plans/2026-11-09-praxis-mode-3-supervise-gateway.md new file mode 100644 index 000000000..9e2fc051e --- /dev/null +++ b/docs/superpowers/plans/2026-11-09-praxis-mode-3-supervise-gateway.md @@ -0,0 +1,241 @@ +# Praxis mode 3/8: Supervise cortex-gateway as a sibling of cortex — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** In praxis mode, `cortex -supervise` runs `cortex` and `cortex-gateway` as two siblings. Each child has its own restart loop. The supervisor writes the token file and `gateway.yaml` at each start. The gateway gets a clean env and exits when the supervisor goes away. In direct mode nothing changes. + +**Architecture:** `cmd/cortex/gateway.go` turns a loaded `*config.Config` into the two gateway files (pure functions, golden-tested). `cmd/cortex/supervise.go` gets a list of child specs. The cortex spec is the loop of today (fallback server, port-in-use backoff). The gateway spec has no fallback. Each spec runs its own loop with its own `restartPolicy`, so a cortex exit does not touch the gateway. The gateway's stdin is a pipe from the supervisor. `cortex-gateway --exit-on-stdin-eof` exits when the pipe closes, which happens on every supervisor exit, SIGKILL included, on Linux and macOS. + +**Tech Stack:** Go 1.26.10, Rust 1.96.0, `praxis-proxy =0.8.0`, POSIX sh. + +**Spec:** `docs/superpowers/specs/2026-11-07-praxis-mode-design.md`, Section 4 ("Supervision", "`gateway.yaml`, in the end-state shape") and Section 6 step 3. Issue: #1720. Tracker: #1714. Previous step: #1719 (plan `docs/superpowers/plans/2026-11-09-praxis-mode-2-gateway-and-ingest.md`, merged as PR #1758). + +## Decisions this plan takes + +Each one closes a question that the spec or the issue leaves open. + +1. **One mechanism for "no orphans": a stdin pipe, on Linux and macOS.** The supervisor sets the gateway's stdin to a pipe and keeps the write end. The kernel closes the write end when the supervisor exits for any reason, SIGKILL included. `cortex-gateway --exit-on-stdin-eof` starts a thread that reads stdin to EOF and then calls `std::process::exit(0)`. No `Pdeathsig` on Linux: it fires on the exit of the parent *thread*, not the process, which is a known trap with Go's thread pool, and one mechanism is testable on macOS. The write end is close-on-exec (Go's default), so the `cortex` child does not hold it. +2. **The flag is opt-in** (`--exit-on-stdin-eof`). Without it, a gateway that is started by hand with `/gateway.token` and `/gateway.yaml`. The service always runs `cortex --supervise --config ~/.cortex/config.yaml`, so this is `~/.cortex/`. Without `--config` the supervisor has no config, so praxis mode is off. +5. **A new token at each supervisor start.** 32 random bytes, hex, written with mode 0600 by write-to-temp-then-rename in the same directory. Both children read the same file for the whole supervisor lifetime, so a restart of one child keeps the token valid for the other. +6. **The `cortex` child gets `CORTEX_GATEWAY_TOKEN_FILE=/gateway.token`** added to the supervisor's env. Only in praxis mode. In direct mode its env is the same as today. +7. **The gateway env is empty** (`cmd.Env = []string{}`, not nil). "Clean env" in the spec means no proxy variables. An empty env is the simplest form that cannot leak one. The gateway needs no variable: the config path comes from `-c`. +8. **`praxis.listen` and `praxis.egress` land in this PR**, because `gateway.yaml` names both (the listener and the interim catch-all route to the egress cluster). Defaults `127.0.0.1:47605` and `127.0.0.1:47606`. When set, each must be a literal loopback `host:port` with a non-zero port, else `Validate` refuses. The egress listener itself is #1721. +9. **The ingest address** in `gateway.yaml` is `listener.session_api_addr` after `ApplyPreset`. An empty or wildcard host becomes `127.0.0.1`. +10. **`gateway.yaml` content** (spec Section 4): `insecure_options` (`allow_unbounded_body`, `allow_private_upstreams`, `allow_private_endpoints`, all true); `body_limits` (`max_request_bytes: 33554432`, `max_response_bytes: null`); one listener `gateway` on `praxis.listen` with `forward_proxy: {enabled: true}`; one chain `main`: `ip_acl` (allow `127.0.0.1/32` and `::1/128` only, no `conditions`), `cortex_attribution`, `cortex_observe`, `access_log`, `router` (catch-all `path_prefix: "/"` to cluster `egress`), `load_balancer` (cluster `egress` on `praxis.egress`). The spec puts `access_log` last among the Cortex filters; Praxis needs `router` and `load_balancer` after it, as in `praxis/server/tests/testdata/gateway.yaml.tmpl`. The 0.8.0 names (checked in `praxis-proxy-filter-0.8.0/src/builtins/http/security/ip_acl.rs` and `praxis-proxy-core-0.8.0/src/config/listener.rs`): the filter is `- filter: ip_acl` with `allow: ["127.0.0.1/32", "::1/128"]`. Do not set `deny`: `allow` and `deny` are mutually exclusive, and the config has `deny_unknown_fields`. The listener key is `forward_proxy: {enabled: true}` on the listener entry. The `cortex_attribution` and `cortex_observe` keys (`token_file`, `ingest`) are as in `gateway.yaml.tmpl`. +11. **The admin metrics port is not in `gateway.yaml` in this PR.** The issue does not list it, and no caller reads it yet. +12. **The startup probe and `gateway_unavailable` are #1721.** The spec lists "the probe" in step 3, but the probe goes through the hop, and the hop is #1721. The issue does not list the probe. +13. **The gateway path is `filepath.Join(filepath.Dir(self), "cortex-gateway")`**, where `self` is `os.Executable()`. A missing or non-executable gateway is a start failure of that child: it logs ERROR once per run of failures and retries with the normal backoff. The supervisor never gives up (see `supervise.go`), and the `cortex` child keeps running. +14. **The stale-gateway stop runs once, at supervisor start, in praxis mode only.** It uses `holderOf(praxis.listen)` (`portconflict.go`). When the holder's exe basename is `cortex-gateway`, it sends SIGTERM, waits up to `superviseStopGrace` for the port to be free, then sends SIGKILL. Any other holder is left alone: the gateway then fails to bind and backs off, and the log names the holder. +15. **The gateway's exit status gets the normal backoff.** It does not use `exitPortInUse`, so `restartPolicy.exited(false, …)` applies. +16. **The version check of the gateway is not in this PR.** The spec ties it to `agentop update` and a released gateway. No release has a gateway before step 7 (#1723). +17. **`cortexPorts` in agentop does not change in this PR.** agentop probes ports only on a fresh install (`env.configFresh`), which has no config, so it cannot be in praxis mode. The only way to ask for praxis mode on a fresh install is `install.sh --praxis` / `agentop setup` with praxis, which step 5 (#1722) adds. The probe of 47605 and 47606 lands there, with its caller. Write this in the PR body as a deviation from the issue text. +18. **`cortex-gateway` is added to agentop's `setupBinaryNames`.** The list has three uses, all presence-based: `planRemoveBinaries` (uninstall, removes only names that `Lstat` finds), `cortexsOwn` (ownership by name), and `stagedBinaries` (installs a name only when the installer staged it, as for `cortex-session-dump`). So setup does not need the gateway, and no other code changes. +19. **`install.sh --stop`** gets `stop_gateway`: after the service and the pidfile proxy are stopped, it finds the process that listens on 47605 (`lsof -t -nP -iTCP@127.0.0.1:47605 -sTCP:LISTEN`), and sends it SIGTERM only when `pid_exe_path` is `${BIN_DIR}/cortex-gateway`. The port check after the stop also lists 47605 and 47606. +20. **The systemd unit does not change in this PR, so a Linux service does not supervise the gateway.** The spec (Section 4, its diagram) says "launchd / systemd -> cortex -supervise", but the unit runs `cortex` without `--supervise` (`cmd/agentop/cmd_service_platform.go`): it relies on `Restart=on-failure` and StartLimit, which a nested supervisor would hide. On a Linux service with `praxis.enabled: true`, nothing writes `gateway.token` and `CORTEX_GATEWAY_TOKEN_FILE` is not set, so `ingestOptions` refuses the boot and `cortex` fails at each start. This is not reachable by a supported path: as decision 17 says, the only way to ask for praxis mode is `install.sh --praxis` / `agentop setup` with praxis, which step 5 (#1722) adds. Step 5 makes the unit use `--supervise` in praxis mode, with its caller. `docs/laptop-service.md` says that a Linux service must not set `praxis.enabled` until then. Write this in the PR body as a deviation from the spec. + +## Global Constraints + +- **Direct mode does not change.** With `praxis.enabled: false` (the default): the supervisor starts only `cortex`; it does not stat, start or need `cortex-gateway`; it writes no token file and no `gateway.yaml`; the `cortex` child's env and args are the same as today. Every existing Go test passes with no change to its expectations. +- Exact names: binary `cortex-gateway`; flag `--exit-on-stdin-eof`; files `gateway.token`, `gateway.yaml`; env `CORTEX_GATEWAY_TOKEN_FILE` (`gatewayTokenEnv`, `cmd/cortex/ingest.go`); defaults `127.0.0.1:47605`, `127.0.0.1:47606`. +- The gateway command line is exactly `cortex-gateway -c /gateway.yaml --exit-on-stdin-eof`. +- Caps in `gateway.yaml`: `max_request_bytes: 33554432` (`32 << 20`, `maxRequestBodySize`), `max_response_bytes: null`. +- Log with `log/slog`, prefix `supervisor:` as today. A child's log lines say which child (`"child", "cortex"` or `"child", "cortex-gateway"`). +- Rust lint gate, from `praxis/`: `cargo fmt --check`, `cargo clippy --all-targets --locked -- -D warnings`, `cargo test --locked`. Use the rustup `cargo` (`$HOME/.cargo/bin`). +- Go: `GOWORK=off go vet ./...` and `gofmt -l ` in each module touched (`core`, `cmd/cortex`, `cmd/agentop`). +- Commits: `git commit -s`, message in ASD-STE100, ending with `Assisted-By: Claude (Anthropic AI) `. No `Co-Authored-By`. PR title `Feat: Supervise cortex-gateway as a sibling of cortex`, body with `Fixes #1720` and decisions 11, 12, 16, 17 and 20 as deviations from the issue or spec. + +## Review Focus + +1. **A cortex crash in praxis mode.** Expected: the gateway pid does not change. Test: Task 5 `TestSupervisor_CortexRestartKeepsGateway`. +2. **A SIGKILL of the supervisor.** Expected: the gateway exits within 5 s. Test: Task 5 `TestSupervisor_KillLeavesNoGateway`; Task 2 `exits_on_stdin_eof`. +3. **Direct mode.** Expected: one child, no gateway lookup, no files. Test: Task 5 `TestSupervisorChildren_DirectModeHasOnlyCortex`. +4. **A `gateway.yaml` that Praxis 0.8.0 refuses.** Expected: the real binary loads the Go-written file and serves a request through it. Test: Task 4 `supervisor_gateway_yaml_serves`. +5. **The token file mode.** Expected: 0600, new at each start. Test: Task 3 `TestWriteGatewayFiles_TokenIs0600AndNew`. + +## File Structure + +| File | Task | What | +|---|---|---| +| `core/config/config.go`, `core/config/validate.go` | 1 | `praxis.listen`, `praxis.egress`, defaults, validation | +| `core/config/praxis_test.go` | 1 | tests | +| `praxis/server/src/main.rs` | 2 | `--exit-on-stdin-eof` | +| `praxis/server/tests/gateway.rs` | 2, 4 | stdin-EOF test; Go-golden test | +| `cmd/cortex/gateway.go` | 3 | `gatewayYAML`, `writeGatewayFiles` | +| `cmd/cortex/gateway_test.go`, `cmd/cortex/testdata/gateway.yaml.golden` | 3 | tests, golden (`-update`) | +| `cmd/cortex/supervise.go` | 5 | child specs, two loops, stale stop, stdin pipe | +| `cmd/cortex/supervise_test.go`, `cmd/cortex/restart_test.go` | 5 | unit and process tests | +| `cmd/cortex/portconflict.go` | 6 | `cortexBinaries` | +| `cmd/agentop/setup_step_binaries.go` (+ tests) | 6 | `setupBinaryNames` | +| `scripts/install.sh`, `scripts/install_test.sh` | 6 | `stop_gateway`, port list | +| `praxis/README.md`, `docs/laptop-service.md` | 7 | docs | + +--- + +### Task 1: `praxis.listen` and `praxis.egress` (Go) + +Depends on: none + +**Files:** +- Modify: `core/config/config.go` (`PraxisConfig`; update its doc comment: `egress` is read by `gateway.yaml` now, the egress listener lands with #1721), `core/config/validate.go` (`validatePraxis`) +- Test: `core/config/praxis_test.go` + +**Interfaces:** +- Produces: fields `Listen string \`yaml:"listen,omitempty" json:"listen,omitempty"\`` and `Egress string \`yaml:"egress,omitempty" json:"egress,omitempty"\`` on `PraxisConfig`; consts `DefaultPraxisListen = "127.0.0.1:47605"`, `DefaultPraxisEgress = "127.0.0.1:47606"`; methods `func (p *PraxisConfig) ListenAddr() string` and `func (p *PraxisConfig) EgressAddr() string` (the field, or the default when empty; the default on a nil receiver). + +- [ ] **Step 1: Write the failing tests:** `TestPraxis_AddrDefaults` (nil block and empty fields give the defaults); `TestPraxis_AddrMustBeLoopback` (table: `0.0.0.0:47605`, `localhost:47605`, `127.0.0.1`, `127.0.0.1:0`, `10.0.0.1:1` refused, with the field name in the error; `127.0.0.1:1`, `[::1]:2` accepted). The checks run only when `praxis.enabled` is true. +- [ ] **Step 2: Run** `GOWORK=off go test ./config/ -run Praxis` from `core` → FAIL. +- [ ] **Step 3: Implement** with `netip.ParseAddrPort` and `IsLoopback()`. +- [ ] **Step 4: Run** → PASS; `GOWORK=off go vet ./...`, `gofmt -l config` clean. +- [ ] **Step 5: Commit.** "Feat: Add the praxis listen and egress addresses". + +Estimated lines: 90. + +### Task 2: `cortex-gateway --exit-on-stdin-eof` (Rust) + +Depends on: none + +**Files:** +- Modify: `praxis/server/src/main.rs` +- Test: `praxis/server/src/main.rs` (unit, flag parse), `praxis/server/tests/gateway.rs` (end to end) + +**Interfaces:** +- Produces: the flag `--exit-on-stdin-eof`. With it, before `run_server_with_registry`, spawn a thread that reads `std::io::stdin()` to EOF (or error) and then calls `std::process::exit(0)`. + +- [ ] **Step 1: Write the failing tests:** unit `exit_flag_is_read` (a helper `has_flag(args, "--exit-on-stdin-eof")`, also when it comes before or after `-c path`); e2e `exits_on_stdin_eof`: start the binary with the test template config, `Stdio::piped()` stdin, wait until the listener accepts, drop the stdin handle, assert the child exits within 5 s. Also `keeps_running_without_flag`: same, without the flag, assert it still runs after 1 s (then kill it). +- [ ] **Step 2: Run** `cargo test --locked -p cortex-gateway` from `praxis` → FAIL. +- [ ] **Step 3: Implement.** `config_arg` must skip the new flag (it is not a value). +- [ ] **Step 4: Run** → PASS; `cargo fmt --check`, `cargo clippy --all-targets --locked -- -D warnings` clean. +- [ ] **Step 5: Commit.** "Feat: Let cortex-gateway exit when its stdin closes". + +Estimated lines: 90. + +### Task 3: The gateway files (Go) + +Depends on: Task 1 + +**Files:** +- Create: `cmd/cortex/gateway.go`, `cmd/cortex/gateway_test.go`, `cmd/cortex/testdata/gateway.yaml.golden` + +**Interfaces:** +- Consumes: `config.PraxisConfig.ListenAddr`, `EgressAddr` (Task 1); `cfg.Listener.SessionAPIAddr`. +- Produces: + - `const gatewayBinary = "cortex-gateway"`, `gatewayTokenName = "gateway.token"`, `gatewayYAMLName = "gateway.yaml"`. + - `func gatewayYAML(cfg *config.Config, tokenPath string) ([]byte, error)`: the file of decision 10. Build it as a Go struct or `map` marshalled with `gopkg.in/yaml.v3` (the library `core/config` uses; already in the module graph; it moves from indirect to direct in `cmd/cortex/go.mod`, so run `GOWORK=off go mod tidy` there). The error is for a session API address that does not split. + - `func ingestAddr(sessionAPIAddr string) (string, error)`: decision 9. + - `func writeGatewayFiles(dir string, cfg *config.Config) (tokenPath, yamlPath string, err error)`: decision 5 for the token, then `gateway.yaml` (mode 0600, write-to-temp-then-rename). + +- [ ] **Step 1: Write the failing tests:** + - `TestGatewayYAML_Golden`: a config with `praxis.enabled`, default addresses, `session_api_addr: 127.0.0.1:47601`, token path `/CORTEX_DIR/gateway.token` → equals `testdata/gateway.yaml.golden`. `-update` rewrites the golden. No test in `cmd/cortex` declares an `update` flag today, so declare `var updateGolden = flag.Bool("update", false, "rewrite testdata goldens")` in `gateway_test.go`. + - `TestGatewayYAML_Fields`: unmarshal the output and check each value of decision 10 (listener address, `forward_proxy.enabled`, filter order, `ip_acl` allows only loopback, `max_request_bytes` 33554432, `max_response_bytes` null, the three insecure options, the egress endpoint, the ingest address). + - `TestIngestAddr`: `:47601` → `127.0.0.1:47601`; `0.0.0.0:1` → `127.0.0.1:1`; `[::]:1` → `127.0.0.1:1`; `127.0.0.1:1` unchanged; `bad` → error. + - `TestWriteGatewayFiles_TokenIs0600AndNew`: two calls in one temp dir give two different 64-hex-char tokens; both files have mode 0600; no temp file is left in the dir. +- [ ] **Step 2: Run** `GOWORK=off go test ./ -run 'Gateway|IngestAddr'` from `cmd/cortex` → FAIL. +- [ ] **Step 3: Implement.** Run with `-update` once, then read the golden by eye against decision 10. +- [ ] **Step 4: Run** → PASS; vet and gofmt clean. +- [ ] **Step 5: Commit.** "Feat: Write the gateway token and gateway.yaml". + +Estimated lines: 260 (the golden does not count). + +### Task 4: The real gateway loads the Go-written `gateway.yaml` (Rust) + +Depends on: Task 2, Task 3 (Task 2 edits the same `gateway.rs`) + +**Files:** +- Modify: `praxis/server/tests/gateway.rs` + +**Interfaces:** +- Consumes: `cmd/cortex/testdata/gateway.yaml.golden` (Task 3), read with `include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../cmd/cortex/testdata/gateway.yaml.golden"))`. + +- [ ] **Step 1: Write the test** `supervisor_gateway_yaml_serves`: take the golden, replace `/CORTEX_DIR/gateway.token` with a temp token file, `127.0.0.1:47605` with a free listen port, `127.0.0.1:47606` with the mock origin's port, `127.0.0.1:47601` with the mock ingest port. Start the binary with it. Send one request with a valid `X-Cortex-Token` from loopback. Assert the origin gets it and the ingest mock gets a request line and a response line. Reuse the file's existing helpers. +- [ ] **Step 2: Run** `cargo test --locked -p cortex-gateway supervisor_gateway_yaml` from `praxis`. When Praxis refuses the file, fix `gatewayYAML` in Go (Task 3's file) and the golden, not the test. Go is the source of truth for the shape; Praxis is the judge of validity. +- [ ] **Step 3: Run** the full Rust gate → PASS. +- [ ] **Step 4: Commit.** "Test: Serve a request through the supervisor's gateway.yaml". + +Estimated lines: 110. + +### Task 5: Two siblings under `cortex -supervise` (Go) + +Depends on: Task 1, Task 3 + +**Files:** +- Modify: `cmd/cortex/supervise.go`, `cmd/cortex/fallback.go` (load the config once; keep `supervisorForwardAddr`'s behavior and its WARN line), `cmd/cortex/restart_test.go` (`TestMain` modes) +- Test: `cmd/cortex/supervise_test.go`, `cmd/cortex/restart_test.go` + +**Interfaces:** +- Consumes: `writeGatewayFiles`, `gatewayBinary` (Task 3); `holderOf` (`portconflict.go`); `gatewayTokenEnv` (`ingest.go`); `restartPolicy`, `portInUseExit`, `fallbackServer` (as today). +- Produces: + - `type childSpec struct { name string; path string; args []string; env []string /* nil = inherit */; stdinPipe bool; fallback *fallbackServer /* cortex only */ }`. + - `func supervisorChildren(cfg *config.Config, configPath, self string, args []string, write func(dir string, cfg *config.Config) (string, string, error)) ([]childSpec, error)`. Direct mode (`cfg == nil` or `!cfg.PraxisEnabled()`): one spec, cortex, same args and env as today; `write` is not called; nothing is stat'ed. Praxis mode: calls `write(filepath.Dir(configPath), cfg)`, adds `CORTEX_GATEWAY_TOKEN_FILE` to the cortex env (`append(os.Environ(), …)`), and adds the gateway spec (decisions 7, 13; command line in Global Constraints). + - `func stopStaleGateway(addr string, lookup func(string) (holder, error), signal func(pid int, sig syscall.Signal) error, free func(string) bool)`: decision 14. + - `runSupervisor` runs one goroutine per spec (`superviseChild(spec, stop <-chan struct{}) `), each with its own `restartPolicy`. The fallback open/close stays in the cortex loop only. On SIGTERM/SIGINT the supervisor closes `stop`; each loop stops its child (cortex: SIGTERM then SIGKILL after `superviseStopGrace`, as today; gateway: close the stdin pipe, then SIGKILL after `superviseStopGrace`) and the supervisor waits for both. A missing gateway binary is a start failure, not a return from `runSupervisor` (decision 13). The cortex child's `cmd.Start` failure keeps today's behavior. + +- [ ] **Step 1: Write the failing unit tests** (`supervise_test.go`): + - `TestSupervisorChildren_DirectModeHasOnlyCortex`: with `praxis.enabled: false`, and with `cfg == nil`: one spec named `cortex`, args unchanged, `env` nil; the `write` func fails the test if called; no file appears in the temp config dir. (The issue's "A test checks this".) + - `TestSupervisorChildren_PraxisAddsGateway`: two specs; gateway path `/cortex-gateway`, args `-c /gateway.yaml --exit-on-stdin-eof`, env empty and non-nil, `stdinPipe` true; cortex env has `CORTEX_GATEWAY_TOKEN_FILE=/gateway.token`. + - `TestStopStaleGateway`: table with fake funcs: holder exe `/x/cortex-gateway` → SIGTERM, port frees → no SIGKILL; port stays held → SIGKILL after the grace; holder `/x/other` → no signal; lookup error → no signal. +- [ ] **Step 2: Write the failing process tests** (`restart_test.go`). Copy `os.Args[0]` into a temp bin dir twice, as `cortex` and as `cortex-gateway` (a copy, not a link: `os.Executable` resolves links on Linux). In `TestMain`, when `filepath.Base(os.Args[0]) == "cortex-gateway"`, run a fake gateway (its env is empty, so it cannot read `CORTEX_TEST_RUN`): print `fake gateway pid ` to stderr, read the `gateway` listener address from the `-c` file and listen on it, and with `--exit-on-stdin-eof` exit on stdin EOF. Add a `CORTEX_TEST_RUN=idle-child` mode: with `-supervise` run `main()`; else print `idle child pid ` and block until SIGTERM. The test config has `praxis.enabled: true`, `listener.bind_loopback_only: true`, a free `praxis.listen`, and passes the existing config validation. + - `TestSupervisor_CortexRestartKeepsGateway`: start the copied `cortex -supervise --config `; read both pids from the output; SIGKILL the cortex child; wait for a new `idle child pid`; assert the gateway pid is the same and alive. Then SIGTERM the supervisor; assert both children exit within 5 s. + - `TestSupervisor_KillLeavesNoGateway`: start it; read the gateway pid; SIGKILL the supervisor; assert the gateway pid is gone within 5 s (`syscall.Kill(pid, 0)` returns `ESRCH`). Kill the orphaned idle cortex child in `t.Cleanup`. + - `TestSupervisor_StopsStaleGateway`: start a fake gateway by hand (from the copied `cortex-gateway`, without the flag) on the listen address; start the supervisor; assert the old pid exits and a new `fake gateway pid` line comes. + - `TestSupervisor_DirectModeStartsNoGateway`: `praxis.enabled: false`, the copied `cortex-gateway` removed: the idle child starts, no `gateway` line is logged, no `gateway.token` or `gateway.yaml` is in the config dir. +- [ ] **Step 3: Run** `GOWORK=off go test -race ./ -run 'Supervisor|StopStale'` from `cmd/cortex` → FAIL. +- [ ] **Step 4: Implement.** Keep every existing comment that still holds; move a comment with its code. Keep `TestSupervisorEndsAStuckProxyInsideLaunchdsWindow` passing unchanged. +- [ ] **Step 5: Run** `GOWORK=off go test -race ./...` from `cmd/cortex` → PASS; vet, gofmt clean. +- [ ] **Step 6: Commit.** "Feat: Supervise cortex-gateway as a sibling of cortex". + +Estimated lines: 650 (the supervise.go refactor and four process tests). + +### Task 6: Know the gateway by name: port conflict, uninstall, `install.sh --stop` + +Depends on: none + +**Files:** +- Modify: `cmd/cortex/portconflict.go` (add `"cortex-gateway": true` to `cortexBinaries`), `cmd/agentop/setup_step_binaries.go` (add `"cortex-gateway"` to `setupBinaryNames`; decision 18), `scripts/install.sh` (`DEMO_GATEWAY_PORT=47605`, `DEMO_EGRESS_PORT=47606`, `stop_gateway`, the port list in `stop_cortex`, the `restage` binary list), `scripts/install_test.sh` +- Test: `cmd/cortex/portconflict_test.go`, `cmd/agentop/cmd_uninstall_test.go`, `cmd/agentop/setup_step_binaries_test.go` (create it if it does not exist), `scripts/install_test.sh` + +- [ ] **Step 1: Write the failing tests:** + - `cmd/cortex`: a `cortexBinaries` case for `cortex-gateway` in the existing table. + - `cmd/agentop`: `planRemoveBinaries` removes a `cortex-gateway` in the bin dir; `cortexsOwn` is true for it; `othersInBinDir` does not list it. Add one `stagedBinaries` test: a staging dir without `cortex-gateway` gives no gateway entry (decision 18). + - `install_test.sh`: `stop_gateway` with mocked `lsof`, `pid_exe_path` and `kill`: holder exe `${BIN_DIR}/cortex-gateway` → TERM; a holder from another directory → no signal; no holder → no signal. Add `stop_gateway` to every `_fn` list that extracts `stop_cortex`, and the two new port variables to every block that prints the `DEMO_*_PORT` lines. +- [ ] **Step 2: Run** the three test commands → FAIL. +- [ ] **Step 3: Implement.** `stop_gateway` follows `pre_rename_proxy_running`'s path compare (`readlink -f` fallback). It is a no-op without `lsof`. +- [ ] **Step 4: Run** → PASS; `shellcheck --severity=error scripts/install.sh`; vet and gofmt clean. +- [ ] **Step 5: Commit.** "Feat: Know cortex-gateway in the port check, uninstall and install.sh --stop". + +Estimated lines: 140. + +### Task 7: Docs + +Depends on: Task 2, Task 5 + +**Files:** +- Modify: `praxis/README.md` ("Run the gateway": the supervisor now writes `gateway.yaml`; the flag `--exit-on-stdin-eof`; add the decisions of this plan that change the gateway, as a new `## Supervision decisions (#1720)` section after `## Accepted edge cases (#1719)`), `docs/laptop-service.md` (one paragraph after the "two `cortex` processes" paragraph near the top: in praxis mode the supervisor has a third child, `cortex-gateway`; it writes `~/.cortex/gateway.token` and `~/.cortex/gateway.yaml` at each start; a cortex restart keeps the gateway) + +- [ ] **Step 1: Write** the text in ASD-STE100. No CLAUDE.md change. +- [ ] **Step 2: Run** `pre-commit run --files `. +- [ ] **Step 3: Commit.** "Docs: Describe the supervised gateway". + +Estimated lines: 50. + +## PRs + +| PR | Title | Parent | Tasks | estLines | +|---|---|---|---|---| +| pr1 | Feat: Supervise cortex-gateway as a sibling of cortex | (none) | 1, 2, 3, 4, 5, 6, 7 | 1390 | + +One PR: the total is under 2,000 counted lines, and the issue is one delivery step (spec Section 6: "Each step is one PR"). + +## Gate + +From the worktree root: + +```sh +export PATH="$HOME/.cargo/bin:$PATH" && (cd praxis && cargo fmt --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked) && (cd core && GOWORK=off go vet ./... && GOWORK=off go test ./config/... ./reloader/... ./audits/...) && (cd cmd/cortex && GOWORK=off go vet ./... && GOWORK=off go mod tidy -diff && GOWORK=off go test -race ./...) && (cd cmd/agentop && GOWORK=off go vet ./...) && scripts/dev/go-test-shards.sh -short cmd/agentop && sh scripts/install_test.sh && shellcheck --severity=error scripts/install.sh && test -z "$(gofmt -l core cmd/cortex cmd/agentop)" +``` + +## Self-review notes + +- Issue scope → tasks: two siblings and backoff → 5; token file and `gateway.yaml` → 3 (with 1 for the addresses); clean env, exit with the supervisor, stale gateway → 2, 5; names in `cortexBinaries`, `install.sh --stop`, `agentop uninstall` → 6; `cortexPorts` → deferred (decision 17). +- Done-when → tests: cortex restart keeps the gateway and a supervisor exit stops it → `TestSupervisor_CortexRestartKeepsGateway`; no orphan after SIGKILL → `TestSupervisor_KillLeavesNoGateway`, `exits_on_stdin_eof`; direct mode → `TestSupervisorChildren_DirectModeHasOnlyCortex`, `TestSupervisor_DirectModeStartsNoGateway`. diff --git a/praxis/README.md b/praxis/README.md index 481b75ff8..eac8e854f 100644 --- a/praxis/README.md +++ b/praxis/README.md @@ -31,8 +31,13 @@ cargo run --locked -p cortex-gateway -- -c gateway.yaml ``` The config path comes from `-c `, then `PRAXIS_CONFIG`, then -`praxis.yaml` in the current directory. The supervisor (#1720) writes -`gateway.yaml`. Until then, write it by hand. +`praxis.yaml` in the current directory. In praxis mode, `cortex -supervise` +writes `gateway.yaml` and starts the gateway (#1720). To run the gateway by +hand, write the file yourself. + +`--exit-on-stdin-eof` makes the gateway exit when its stdin closes. The +supervisor uses it. Without the flag, a gateway that you start with +`/gateway.yaml --exit-on-stdin-eof`. `` is the directory of the `--config` file, `~/.cortex/` for the service. The binary is in the same directory as `cortex`. +3. The gateway's stdin is a pipe from the supervisor. When the supervisor exits for any reason, SIGKILL included, the pipe closes and the gateway exits. This is the only orphan guard, on Linux and macOS. No `Pdeathsig`. +4. To stop the gateway, the supervisor closes the pipe. After `superviseStopGrace` it sends SIGKILL. +5. At each start, the supervisor writes a new `gateway.token` (32 random bytes, hex, mode 0600) and `gateway.yaml`. It writes each file to a temp file and then renames it. The `cortex` child gets `CORTEX_GATEWAY_TOKEN_FILE`. +6. The gateway gets an empty env, so no proxy variable can get to it. +7. `gateway.yaml` listens on `praxis.listen` (default `127.0.0.1:47605`). A catch-all route sends all traffic to `praxis.egress` (default `127.0.0.1:47606`). The egress listener is #1721. `ip_acl` allows only `127.0.0.1/32` and `::1/128`. +8. A missing gateway binary, or a gateway that exits, gets the normal backoff. The supervisor does not stop, and `cortex` keeps running. +9. The supervisor does not stop a gateway from an earlier run. The stdin pipe ends that gateway when its supervisor exits, so a new gateway that cannot bind `praxis.listen` binds it on the next try. A config that `config.Validate` refuses starts no gateway. +10. Not in this step: the admin metrics port, the startup probe and `gateway_unavailable` (#1721), and the version check of the gateway (#1723). +11. Direct mode does not change. The supervisor starts only `cortex` and writes no gateway file. + ## The parity rule Go is the source of truth. A Go parser change regenerates `expected.json`, diff --git a/praxis/server/src/main.rs b/praxis/server/src/main.rs index fb80d640f..a71c178ca 100644 --- a/praxis/server/src/main.rs +++ b/praxis/server/src/main.rs @@ -10,6 +10,10 @@ praxis_filter::register_filters! { http "cortex_observe" => CortexObserve::from_config, } +/// Exit when stdin closes. The supervisor holds the write end of a pipe, so +/// the gateway exits when the supervisor exits for any reason, SIGKILL included. +const EXIT_ON_STDIN_EOF: &str = "--exit-on-stdin-eof"; + fn main() { // Before anything that can build a TLS config. praxis::install_crypto_provider(); @@ -25,6 +29,13 @@ fn main() { let tracing_guard = praxis::init_tracing(&config).unwrap_or_else(|e| praxis::fatal(&e)); let log_level = Some(tracing_guard.log_level_state()); let _tracing_guard = tracing_guard; + if has_flag(std::env::args().skip(1), EXIT_ON_STDIN_EOF) { + std::thread::spawn(|| { + // EOF or a read error: either way the supervisor is gone. + let _ = std::io::copy(&mut std::io::stdin().lock(), &mut std::io::sink()); + std::process::exit(0); + }); + } praxis::run_server_with_registry(config, custom_registry(), config_path, log_level) } @@ -38,9 +49,14 @@ fn config_arg(mut args: impl Iterator) -> Option { None } +/// True when `flag` is one of the arguments. +fn has_flag(mut args: impl Iterator, flag: &str) -> bool { + args.any(|a| a == flag) +} + #[cfg(test)] mod tests { - use super::config_arg; + use super::{EXIT_ON_STDIN_EOF, config_arg, has_flag}; fn args(list: &[&str]) -> impl Iterator { list.iter() @@ -62,4 +78,15 @@ mod tests { assert_eq!(config_arg(args(&[])), None); assert_eq!(config_arg(args(&["-c"])), None); } + + #[test] + fn exit_flag_is_read() { + let before = ["--exit-on-stdin-eof", "-c", "a.yaml"]; + let after = ["-c", "a.yaml", "--exit-on-stdin-eof"]; + for list in [before, after] { + assert!(has_flag(args(&list), EXIT_ON_STDIN_EOF)); + assert_eq!(config_arg(args(&list)).as_deref(), Some("a.yaml")); + } + assert!(!has_flag(args(&["-c", "a.yaml"]), EXIT_ON_STDIN_EOF)); + } } diff --git a/praxis/server/tests/gateway.rs b/praxis/server/tests/gateway.rs index db82eecd1..d483a780c 100644 --- a/praxis/server/tests/gateway.rs +++ b/praxis/server/tests/gateway.rs @@ -19,6 +19,11 @@ const FIXTURE: &str = concat!( "/../../core/plugins/inferenceparser/testdata/claude-code/main-stream-thinking-tool-use" ); const TEMPLATE: &str = include_str!("testdata/gateway.yaml.tmpl"); +/// The `gateway.yaml` that `cortex -supervise` writes (Go is its source). +const SUPERVISOR_YAML: &str = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../cmd/cortex/testdata/gateway.yaml.golden" +)); const ACCESS_LOG: &str = " - filter: access_log"; const WAIT: Duration = Duration::from_secs(10); const QUIET: Duration = Duration::from_secs(1); @@ -324,6 +329,25 @@ fn free_port() -> u16 { } fn gateway(origin: &Origin, ingest: &Ingest, access_log: bool) -> Gateway { + gateway_with(origin, ingest, access_log, &[]) +} + +/// Starts the gateway with `flags` added. Its stdin is a pipe that the +/// returned `Gateway` holds. +fn gateway_with(origin: &Origin, ingest: &Ingest, access_log: bool, flags: &[&str]) -> Gateway { + start(flags, |token_file, port| { + TEMPLATE + .replace("{{LISTEN}}", &port.to_string()) + .replace("{{TOKEN_FILE}}", token_file) + .replace("{{INGEST}}", &ingest.port.to_string()) + .replace("{{ORIGIN}}", &origin.port.to_string()) + .replace("{{ACCESS_LOG}}", if access_log { ACCESS_LOG } else { "" }) + }) +} + +/// Starts the gateway on the config that `config(token_file, listen_port)` +/// gives, with `flags` added. +fn start(flags: &[&str], config: impl FnOnce(&str, u16) -> String) -> Gateway { static N: AtomicUsize = AtomicUsize::new(0); let dir = std::env::temp_dir().join(format!( "cortex-gateway-test-{}-{}", @@ -334,18 +358,17 @@ fn gateway(origin: &Origin, ingest: &Ingest, access_log: bool) -> Gateway { let token_file = dir.join("token"); std::fs::write(&token_file, format!("{TOKEN}\n")).expect("token file"); let port = free_port(); - let yaml = TEMPLATE - .replace("{{LISTEN}}", &port.to_string()) - .replace("{{TOKEN_FILE}}", token_file.to_str().expect("path")) - .replace("{{INGEST}}", &ingest.port.to_string()) - .replace("{{ORIGIN}}", &origin.port.to_string()) - .replace("{{ACCESS_LOG}}", if access_log { ACCESS_LOG } else { "" }); + let yaml = config(token_file.to_str().expect("path"), port); let config = dir.join("gateway.yaml"); std::fs::write(&config, yaml).expect("config"); let log = std::fs::File::create(dir.join("gateway.log")).expect("log"); let mut child = Command::new(env!("CARGO_BIN_EXE_cortex-gateway")) + // The supervisor gives the gateway an empty env. Test it the same way. + .env_clear() .arg("-c") .arg(&config) + .args(flags) + .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(log) .spawn() @@ -951,3 +974,64 @@ fn client_close_mid_sse_is_client_canceled() { assert_eq!(resp["event"]["statusCode"], 499, "{resp}"); assert_eq!(resp["event"]["error"]["kind"], "client_canceled"); } + +#[test] +fn exits_on_stdin_eof() { + let (o, i) = (ok_origin(), ingest()); + let mut g = gateway_with(&o, &i, true, &["--exit-on-stdin-eof"]); + drop(g.child.stdin.take()); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(st) = g.child.try_wait().expect("try_wait") { + assert!(st.success(), "cortex-gateway exited with {st}"); + return; + } + assert!( + Instant::now() < deadline, + "cortex-gateway still runs 5s after stdin closed" + ); + std::thread::sleep(Duration::from_millis(20)); + } +} + +#[test] +fn keeps_running_without_flag() { + let (o, i) = (ok_origin(), ingest()); + let mut g = gateway(&o, &i, true); + drop(g.child.stdin.take()); + std::thread::sleep(QUIET); + assert!( + g.child.try_wait().expect("try_wait").is_none(), + "cortex-gateway exited without --exit-on-stdin-eof" + ); +} + +#[test] +fn supervisor_gateway_yaml_serves() { + // Every default port in the golden is one this test replaces. Checked on the + // golden, not the result: the mocks' ephemeral ports can be in 47600-47609. + let rest = SUPERVISOR_YAML + .replace("127.0.0.1:47605", "") + .replace("127.0.0.1:47606", "") + .replace("127.0.0.1:47601", ""); + assert!(!rest.contains(":4760"), "{SUPERVISOR_YAML}"); + let (o, i) = (ok_origin(), ingest()); + let g = start(&[], |token_file, port| { + let yaml = SUPERVISOR_YAML + .replace("/CORTEX_DIR/gateway.token", token_file) + .replace("127.0.0.1:47605", &format!("127.0.0.1:{port}")) + .replace("127.0.0.1:47606", &format!("127.0.0.1:{}", o.port)) + .replace("127.0.0.1:47601", &format!("127.0.0.1:{}", i.port)); + assert!(!yaml.contains("CORTEX_DIR"), "{yaml}"); + yaml + }); + let (h, body) = send(&g, "GET", "/v1/models", &[("X-Cortex-Token", TOKEN)], b""); + assert_eq!(status(&h), 200); + assert_eq!(body, b"ok"); + o.seen + .recv_timeout(WAIT) + .expect("the origin saw the request"); + let lines = i.take(2); + assert_eq!(lines[0]["event"]["phase"], "request"); + assert_eq!(lines[1]["event"]["phase"], "response"); +} diff --git a/scripts/install.sh b/scripts/install.sh index d2b888346..4e010d01b 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -611,6 +611,9 @@ DEMO_FORWARD_PORT=47600 DEMO_SESSION_PORT=47601 DEMO_STATS_PORT=47602 DEMO_HEALTH_PORT=47604 +# praxis mode: cortex-gateway's listener (praxis.listen) and the egress (praxis.egress). +DEMO_GATEWAY_PORT=47605 +DEMO_EGRESS_PORT=47606 # port_in_use exits 0 if something is already listening on the given loopback port. # Best-effort across platforms: lsof (macOS + many Linux), then ss (iproute2, the @@ -780,7 +783,7 @@ stop_cortex() { # we stopped what we know about means a foreign proxy this script did not start — # worth naming, since in a sandbox it cannot be found through ps. _busy="" - for _p in "${DEMO_FORWARD_PORT}" "${DEMO_SESSION_PORT}" "${DEMO_STATS_PORT}" "${DEMO_HEALTH_PORT}"; do + for _p in "${DEMO_FORWARD_PORT}" "${DEMO_SESSION_PORT}" "${DEMO_STATS_PORT}" "${DEMO_HEALTH_PORT}" "${DEMO_GATEWAY_PORT}" "${DEMO_EGRESS_PORT}"; do port_in_use "${_p}" && _busy="${_busy} ${_p}" done if [ -n "${_stopped}" ]; then @@ -836,7 +839,7 @@ no_setup() { restage() { mkdir -p "${CORTEX_DIR}/tmp" || return 1 _rs=$(mktemp -d "${CORTEX_DIR}/tmp/stage.XXXXXX") || return 1 - for _rs_b in agentop cortex cortex-session-dump; do + for _rs_b in agentop cortex cortex-session-dump cortex-gateway; do [ ! -f "$1/${_rs_b}" ] || cp -p "$1/${_rs_b}" "${_rs}/${_rs_b}" || { rm -rf "${_rs}" return 1 diff --git a/scripts/install_test.sh b/scripts/install_test.sh index edc1d31fe..9c2114b40 100644 --- a/scripts/install_test.sh +++ b/scripts/install_test.sh @@ -737,6 +737,7 @@ with_pre_rename() { # exe-of-4242(__NONE__) bin_dir ps-comm(fail) { printf 'PROXY_PIDFILE=%s\nBIN_DIR=%s\nCORTEX_DIR=%s\nSUPERVISOR_NAME=x\n' "${_pf}" "$2" "${TMP}" printf 'DEMO_FORWARD_PORT=47600\nDEMO_SESSION_PORT=47601\nDEMO_STATS_PORT=47603\nDEMO_HEALTH_PORT=47604\n' + printf 'DEMO_GATEWAY_PORT=47605\nDEMO_EGRESS_PORT=47606\n' printf 'info() { :; }\nwarn() { :; }\nsleep() { :; }\n' printf 'port_in_use() { return 1; }\n' printf 'kill() {\n' @@ -782,6 +783,12 @@ check "pre-rename: --stop matches a BIN_DIR reached through a symlink" \ "old:stopped pidfile:none" \ "$(with_pre_rename "$(cd "${TMP}/prr_real" && pwd -P)/authbridge-proxy" "${TMP}/prr_link" authbridge-prox)" +# --- stop_cortex and restage know the praxis-mode gateway --- +check "stop_cortex checks the gateway and egress ports too" "1" \ + "$(grep -c 'for _p in .*DEMO_GATEWAY_PORT.*DEMO_EGRESS_PORT' "${INSTALL_SH}")" +check "restage copies cortex-gateway" "1" \ + "$(grep -c 'for _rs_b in .*cortex-gateway' "${INSTALL_SH}")" + # --- ensure_tmpdir: exports TMPDIR on BOTH paths (regression: unset-TMPDIR abort) --- # # With `set -u`, a later "${TMPDIR}" reference (the svc_err mktemp on the supervised