amadan

carlos / platform Public

activator: mint CARLOS_ADMIN_TOKEN; control socket always on, schedule verbs mounted

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 3f9c658c68e9d1624eba1fc7fa7445dcd05bd6fc parent 9fa030f
6 files changed, +506 −94
  • internal/activator/activator_test.go +201 −18
  • internal/activator/admintoken.go +79 −0
  • internal/activator/admintoken_test.go +47 −0
  • internal/activator/backend_exec.go +110 −51
  • internal/edge/edge.go +6 −0
  • internal/edge/purge.go +63 −25
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 62bad5c92..1fbd9335f 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -2412,8 +2412,9 @@ func waitForFile(t *testing.T, path string, within time.Duration) string {
}
// Start hands the spawned instance its control socket through the ENVIRONMENT
-// (CARLOS_CONTROL_SOCKET), never as a CLI flag, and starts a purge listener
-// bound to it whenever ExecBackend.Cache is set.
+// (CARLOS_CONTROL_SOCKET), never as a CLI flag, and binds a listener on it.
+// This is the WITH-CACHE case, where that listener also serves the purge
+// verb; the nil-cache sibling below covers the socket existing anyway.
//
// The argv assertion is the regression guard: a --control-socket flag makes
// every app binary that predates this feature exit 2 at startup (Go's flag
@@ -2486,31 +2487,208 @@ func TestExecBackendStartPassesControlSocketInEnvAndStartsPurgeListener(t *testi
}
}
-// A nil ExecBackend.Cache disables the purge listener entirely — no control
-// socket, no CARLOS_CONTROL_SOCKET in the child's environment (there is
-// nothing for it to talk to), no error.
-func TestExecBackendStartWithNilCacheSkipsPurgeListener(t *testing.T) {
+// The control channel is no longer the purge channel. It is bound for EVERY
+// exec-backed child, cache or no cache (spec 2026-08-23-scheduled-work-design
+// §7.2 — the scheduled-work verbs are served on it, and a box with no edge
+// cache still runs scheduled work), so a nil ExecBackend.Cache no longer means
+// "no control socket". What it does still mean is no CARLOS_PURGE: nothing on
+// this box can purge anything, and the rule that a variable naming a channel
+// must not be set where that channel does nothing is exactly what makes the
+// two names diverge here.
+//
+// The same spawn now also mints the instance's CARLOS_ADMIN_TOKEN before it
+// reads the env, so the child is handed the same secret the agent presents
+// back to it (NextDue, the tick) and the control socket checks bearers
+// against — nothing minted it before this, which is why the next-due probe
+// silently failed for every console-created instance.
+func TestExecBackendStartWithNilCacheStillBindsControlSocket(t *testing.T) {
b := testExecBackend(t) // b.Cache left nil
- _, envPath := stageRecordingLive(t, b, "no-cache.example.com", "v1")
+ host := "no-cache.example.com"
+ _, envPath := stageRecordingLiveFullEnv(t, b, host, "v1")
rt := registry.Route{
- Host: "no-cache.example.com",
+ Host: host,
Kind: "instance",
Socket: filepath.Join(t.TempDir(), "instance.sock"),
}
+ wantControlSocket := rt.Socket + ".control"
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
- proc, _, _ := b.Start(ctx, rt)
+
+ // Same shape as the cached case above: poll for the socket WHILE Start's
+ // health-wait is still running, because the fake binary never binds
+ // rt.Socket and Start's failure path Stops the child, which unlinks the
+ // control socket before Start returns.
+ done := make(chan struct{})
+ var proc Proc
+ go func() {
+ proc, _, _ = b.Start(ctx, rt) // health-wait error from the fake binary is expected here
+ close(done)
+ }()
+
+ found := false
+ deadline := time.Now().Add(1500 * time.Millisecond)
+ for time.Now().Before(deadline) {
+ if _, err := os.Stat(wantControlSocket); err == nil {
+ found = true
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ child := map[string]string{}
+ for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") {
+ if k, v, ok := strings.Cut(line, "="); ok {
+ child[k] = v
+ }
+ }
+
+ <-done
if proc != nil {
defer proc.Stop(context.Background())
}
- if _, err := os.Stat(rt.Socket + ".control"); err == nil {
- t.Fatal("expected no control socket when ExecBackend.Cache is nil")
+ if !found {
+ t.Fatalf("expected control socket %s while Start's health-wait ran — it is bound for every exec child now, cache or not", wantControlSocket)
+ }
+ if got := child["CARLOS_CONTROL_SOCKET"]; got != wantControlSocket {
+ t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q", got, wantControlSocket)
+ }
+ if got, ok := child["CARLOS_PURGE"]; ok {
+ t.Fatalf("child CARLOS_PURGE = %q, want it unset when there is no cache to purge", got)
+ }
+
+ minted, err := readEnvFile(filepath.Join(b.DataDir, host+".env"))
+ if err != nil {
+ t.Fatalf("read the instance env file the spawn should have minted into: %v", err)
+ }
+ if len(minted["CARLOS_ADMIN_TOKEN"]) != 64 {
+ t.Fatalf("minted CARLOS_ADMIN_TOKEN = %q, want 64 hex chars", minted["CARLOS_ADMIN_TOKEN"])
}
- if got := strings.TrimSpace(waitForFile(t, envPath, time.Second)); got != "<unset>" {
- t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want it unset when there is no cache", got)
+ if got := child["CARLOS_ADMIN_TOKEN"]; got != minted["CARLOS_ADMIN_TOKEN"] {
+ t.Fatalf("child CARLOS_ADMIN_TOKEN = %q, want the minted %q — the child must hold the same secret the agent presents back to it", got, minted["CARLOS_ADMIN_TOKEN"])
+ }
+}
+
+// fakeTimerStore is the registry subset schedule.Mount needs, stubbed: these
+// tests are about the WIRING (are the verbs mounted, is the bearer the
+// instance's minted token), not about timer semantics — internal/schedule
+// owns those.
+type fakeTimerStore struct{ rows []registry.Timer }
+
+func (f *fakeTimerStore) UpsertTimer(t registry.Timer) error { f.rows = append(f.rows, t); return nil }
+func (f *fakeTimerStore) DeleteTimer(host, name string) error { return nil }
+func (f *fakeTimerStore) GetTimer(host, name string) (*registry.Timer, error) {
+ return nil, os.ErrNotExist
+}
+func (f *fakeTimerStore) TimersFor(host string) ([]registry.Timer, error) { return f.rows, nil }
+func (f *fakeTimerStore) CountTimers(host, kind string) (int, error) { return len(f.rows), nil }
+
+// startAndReachControlSocket spawns host through the real Start path and hands
+// the test the control socket's path plus the token that spawn minted, while
+// Start's health-wait is STILL RUNNING — the fake binary never binds rt.Socket,
+// so once Start returns its failure path has already Stopped the child and
+// unlinked the control socket. Returns a stop func the caller must defer.
+func startAndReachControlSocket(t *testing.T, b *ExecBackend, host string) (controlSocket, token string, stop func()) {
+ t.Helper()
+ stageRecordingLiveFullEnv(t, b, host, "v1")
+ rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "instance.sock")}
+ controlSocket = rt.Socket + ".control"
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ done := make(chan struct{})
+ var proc Proc
+ go func() {
+ proc, _, _ = b.Start(ctx, rt)
+ close(done)
+ }()
+
+ deadline := time.Now().Add(3 * time.Second)
+ for time.Now().Before(deadline) {
+ if _, err := os.Stat(controlSocket); err == nil {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ env, err := readEnvFile(filepath.Join(b.DataDir, host+".env"))
+ if err != nil {
+ cancel()
+ <-done
+ t.Fatalf("read minted env file: %v", err)
+ }
+ return controlSocket, env["CARLOS_ADMIN_TOKEN"], func() {
+ cancel()
+ <-done
+ if proc != nil {
+ proc.Stop(context.Background())
+ }
+ }
+}
+
+// The whole point of the generalised control channel: with ExecBackend.Timers
+// set, the scheduled-work verbs are served on the spawned instance's control
+// socket, gated on the token THIS spawn minted into <host>.env. The wrong
+// bearer is a 401, which is what closes the same-uid-sibling gap purge.go
+// documents — a sibling can compute this socket's path, but it cannot read
+// another instance's 0600 env file to learn its token.
+func TestExecBackendStartMountsScheduleVerbsOnControlSocket(t *testing.T) {
+ b := testExecBackend(t) // b.Cache left nil — the schedule verbs do not need one
+ b.Timers = &fakeTimerStore{}
+ ctl, token, stop := startAndReachControlSocket(t, b, "timers.example.com")
+ defer stop()
+
+ if len(token) != 64 {
+ t.Fatalf("minted CARLOS_ADMIN_TOKEN = %q, want 64 hex chars", token)
+ }
+
+ get := func(bearer string) int {
+ req, err := http.NewRequest("GET", "http://carlos/schedule", nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if bearer != "" {
+ req.Header.Set("Authorization", "Bearer "+bearer)
+ }
+ resp, err := socketClient(ctl).Do(req)
+ if err != nil {
+ t.Fatalf("dial control socket %s: %v", ctl, err)
+ }
+ defer resp.Body.Close()
+ return resp.StatusCode
+ }
+
+ if got := get(token); got != http.StatusOK {
+ t.Fatalf("GET /schedule with the minted token = %d, want 200 — the verbs are not mounted", got)
+ }
+ if got := get("not-the-token"); got != http.StatusUnauthorized {
+ t.Fatalf("GET /schedule with a wrong bearer = %d, want 401 — a same-uid sibling could reach this socket", got)
+ }
+ if got := get(""); got != http.StatusUnauthorized {
+ t.Fatalf("GET /schedule with no bearer = %d, want 401", got)
+ }
+}
+
+// Timers is nil-tolerant, and honestly so: the socket is still bound (the
+// channel is per-child now, not per-feature) but nothing answers /schedule.
+// A 404 is the truth for an agent that predates the feature; a socket that
+// wasn't there at all would make the child's CARLOS_CONTROL_SOCKET a lie.
+func TestExecBackendStartWithNilTimersMountsNoScheduleVerbs(t *testing.T) {
+ b := testExecBackend(t) // b.Timers and b.Cache both nil
+ ctl, token, stop := startAndReachControlSocket(t, b, "no-timers.example.com")
+ defer stop()
+
+ req, err := http.NewRequest("GET", "http://carlos/schedule", nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Authorization", "Bearer "+token)
+ resp, err := socketClient(ctl).Do(req)
+ if err != nil {
+ t.Fatalf("dial control socket %s: %v — it must exist even with no verbs mounted", ctl, err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("GET /schedule with nil Timers = %d, want 404", resp.StatusCode)
}
}
@@ -2656,10 +2834,14 @@ func TestExecBackendStartWithNilCacheSetsNoPurgeVar(t *testing.T) {
// gating only on "does the agent SET the var" left a tenant free to set it
// itself. On a box with no cache the child would then be told a purge channel
// exists when none does, and an app that trusts CARLOS_PURGE would skip its
-// own invalidation and serve stale content indefinitely.
+// own invalidation and serve stale content indefinitely. CARLOS_PURGE is
+// therefore REMOVED here.
//
-// CARLOS_CONTROL_SOCKET is the same hole and a worse one: a tenant-chosen
-// socket path is a name the agent never vouched for. Both are removed.
+// CARLOS_CONTROL_SOCKET is the same hole and a worse one — a tenant-chosen
+// socket path is a name the agent never vouched for — but its channel now
+// exists on this box (bound for every exec child, §7.2), so the fix is
+// OVERRIDE rather than removal: the child must be pointed at the socket the
+// agent actually bound, never at /tmp/attacker.sock.
func TestExecBackendStartTenantCannotClaimAbsentPurgeChannel(t *testing.T) {
b := testExecBackend(t) // b.Cache left nil — no purge channel on this box
host := "liar.example.com"
@@ -2693,8 +2875,9 @@ func TestExecBackendStartTenantCannotClaimAbsentPurgeChannel(t *testing.T) {
if got, ok := child["CARLOS_PURGE"]; ok {
t.Fatalf("tenant-supplied CARLOS_PURGE=%q survived onto a box with no cache", got)
}
- if got, ok := child["CARLOS_CONTROL_SOCKET"]; ok {
- t.Fatalf("tenant-supplied CARLOS_CONTROL_SOCKET=%q survived onto a box with no cache", got)
+ if want := rt.Socket + ".control"; child["CARLOS_CONTROL_SOCKET"] != want {
+ t.Fatalf("child CARLOS_CONTROL_SOCKET = %q, want %q — a tenant must not redirect the control socket, cache or no cache",
+ child["CARLOS_CONTROL_SOCKET"], want)
}
// Only the agent-owned names are stripped; ordinary tenant config is
// untouched.
diff --git a/internal/activator/admintoken.go b/internal/activator/admintoken.go
new file mode 100644
index 000000000..13ed62fb6
--- /dev/null
+++ b/internal/activator/admintoken.go
@@ -0,0 +1,79 @@
+package activator
+
+// The instance secret. CARLOS_ADMIN_TOKEN is what the agent presents to an
+// instance for the things only the platform may ask (GET /api/next-due, and
+// now the scheduled-work tick — spec 2026-08-23-scheduled-work-design.md
+// §6.4) and what the instance presents back on its control socket, where it
+// is the bearer that closes the same-uid-sibling gap internal/edge/purge.go
+// documents. Until this file nothing on the platform minted it: Eleven's own
+// tooling wrote it by hand, console-created instances never had one at all,
+// and the next-due probe failed silently for every one of them. Minted here,
+// once, on the first spawn (or seed) that finds it missing; never rotated by
+// the platform, because the running instance already holds the value and a
+// rotation behind its back would lock the agent out of its own instance.
+
+import (
+ "crypto/rand"
+ "encoding/hex"
+ "fmt"
+ "os"
+ "strings"
+)
+
+// ensureAdminToken makes sure path (an instance's own <host>.env) carries a
+// CARLOS_ADMIN_TOKEN, minting one if it does not. The file is the instance's,
+// not the agent's: every other line in it is preserved untouched and an
+// existing token is never rewritten. An absent file is created 0600 — the
+// same mode the env file carries everywhere else, and the only thing standing
+// between another UNIX user on the box and the instance's secret.
+func ensureAdminToken(path string) error {
+ // Read the raw bytes once: they answer both questions this needs — is a
+ // token already there, and does the file end on a line boundary.
+ raw, err := os.ReadFile(path)
+ if err != nil && !os.IsNotExist(err) {
+ return fmt.Errorf("read %s: %w", path, err)
+ }
+ if parseEnvBytes(raw)["CARLOS_ADMIN_TOKEN"] != "" {
+ return nil
+ }
+ secret := make([]byte, 32)
+ if _, err := rand.Read(secret); err != nil {
+ return fmt.Errorf("mint admin token for %s: %w", path, err)
+ }
+ // Append rather than rewrite: the instance owns this file, and a
+ // read-modify-write would race anything else editing it.
+ f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0o600)
+ if err != nil {
+ return fmt.Errorf("open %s: %w", path, err)
+ }
+ defer f.Close()
+ // Land on a fresh line: a file whose last line has no trailing newline
+ // would otherwise swallow our variable into it.
+ prefix := ""
+ if len(raw) > 0 && raw[len(raw)-1] != '\n' {
+ prefix = "\n"
+ }
+ if _, err := fmt.Fprintf(f, "%sCARLOS_ADMIN_TOKEN=%s\n", prefix, hex.EncodeToString(secret)); err != nil {
+ return fmt.Errorf("write %s: %w", path, err)
+ }
+ return nil
+}
+
+// parseEnvBytes parses an EnvironmentFile already in hand: KEY=VALUE lines,
+// blank lines and # comments skipped. It is readEnvFile's whole body (that
+// one adds the read), split out so ensureAdminToken can answer "is the token
+// set" and "does the file end in a newline" from ONE read of the file — two
+// reads could disagree, and the second one decides where the token lands.
+func parseEnvBytes(raw []byte) map[string]string {
+ m := map[string]string{}
+ for _, line := range strings.Split(string(raw), "\n") {
+ line = strings.TrimSpace(line)
+ if line == "" || strings.HasPrefix(line, "#") {
+ continue
+ }
+ if k, v, ok := strings.Cut(line, "="); ok {
+ m[strings.TrimSpace(k)] = strings.TrimSpace(v)
+ }
+ }
+ return m
+}
diff --git a/internal/activator/admintoken_test.go b/internal/activator/admintoken_test.go
new file mode 100644
index 000000000..0a5c2a898
--- /dev/null
+++ b/internal/activator/admintoken_test.go
@@ -0,0 +1,47 @@
+package activator
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func TestEnsureAdminTokenMintsOnce(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "h.env")
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ st, _ := os.Stat(p)
+ if st.Mode().Perm() != 0o600 {
+ t.Fatalf("mode = %o", st.Mode().Perm())
+ }
+ env, _ := readEnvFile(p)
+ tok := env["CARLOS_ADMIN_TOKEN"]
+ if len(tok) != 64 {
+ t.Fatalf("token = %q", tok)
+ }
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ env, _ = readEnvFile(p)
+ if env["CARLOS_ADMIN_TOKEN"] != tok {
+ t.Fatal("token rotated on second call")
+ }
+}
+
+func TestEnsureAdminTokenPreservesOtherVars(t *testing.T) {
+ p := filepath.Join(t.TempDir(), "h.env")
+ os.WriteFile(p, []byte("DATABASE_URL=sqlite\n# comment\n"), 0o600)
+ if err := ensureAdminToken(p); err != nil {
+ t.Fatal(err)
+ }
+ raw, _ := os.ReadFile(p)
+ if !strings.HasPrefix(string(raw), "DATABASE_URL=sqlite\n# comment\n") || !strings.Contains(string(raw), "CARLOS_ADMIN_TOKEN=") {
+ t.Fatalf("file = %q", raw)
+ }
+ env, _ := readEnvFile(p)
+ if env["DATABASE_URL"] != "sqlite" {
+ t.Fatal("other var lost")
+ }
+}
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index e0679120e..08cfcedf4 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -3,7 +3,6 @@
package activator
import (
- "bufio"
"context"
"encoding/json"
"errors"
@@ -26,6 +25,7 @@ import (
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
)
// spawnHealthTimeout bounds how long a spawned instance may take to bind its
@@ -60,9 +60,11 @@ type ExecBackend struct {
// but a carlos app is a separate binary from the release store — there is
// nothing sensible to invent here, and the adopt pass owns the symlinks.
Materialize func(ctx context.Context, host string) error
- // Cache, when non-nil, gets a purge listener started for each spawned
- // instance so it can invalidate its own cached responses right after a
- // write. Nil disables this — no purge listener, no error.
+ // Cache, when non-nil, adds the purge verb to each spawned instance's
+ // control socket so it can invalidate its own cached responses right
+ // after a write. Nil disables that ONE verb, not the socket: the control
+ // channel is bound for every exec child either way (§7.2), because the
+ // scheduled-work verbs need it on boxes with no edge cache too.
Cache *edge.Cache
// Cgroups, when non-nil, places every spawned child — instance, sidecar,
// seed — in a per-host cgroup under the agent's delegated subtree so the
@@ -74,6 +76,13 @@ type ExecBackend struct {
// byte it got before. Nil (tests, a box without the shipping loop)
// spawns exactly as before: fd passthrough, no pipes.
Logs EventSink
+ // Timers, when non-nil, mounts the scheduled-work verbs on each spawned
+ // instance's control socket so an app can set and clear its own one-shot
+ // timers (spec 2026-08-23-scheduled-work-design.md §7.2). The box wiring
+ // passes the registry. Nil — tests, and any box whose agent predates the
+ // feature — binds the control socket without those verbs, so an app that
+ // asks gets 404 rather than a socket that isn't there.
+ Timers schedule.TimerStore
}
// execLogLineMax caps one teed log line (spec §5): longer is truncated with
@@ -271,6 +280,13 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
if err := b.ensureLiveLink(ctx, host); err != nil {
return err
}
+ // Mint before the env is read, exactly as Start does — a seed is the FIRST
+ // thing a fresh instance runs, so this is usually where the token comes
+ // into existence, and instanceEnv's contract is that a seed sees the same
+ // environment the instance will.
+ if err := ensureAdminToken(filepath.Join(b.DataDir, host+".env")); err != nil {
+ return fmt.Errorf("admin token for %s: %w", host, err)
+ }
fileEnv, err := b.instanceEnv(host)
if err != nil {
return err
@@ -297,13 +313,15 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
// start's sweep) removes it.
cmd, err := b.Cgroups.Start(host, func() *exec.Cmd {
cmd := exec.CommandContext(ctx, b.liveLink(host), argv...)
- // The purge-channel vars are agent-owned in BOTH directions here too
- // (Start's discipline, see withoutVars): fileEnv is tenant-supplied, so
- // without this a tenant's own env file could hand its seed a
- // CARLOS_CONTROL_SOCKET the agent never opened and a CARLOS_PURGE for a
- // channel that does not exist — and the seed has no serving cache to
- // purge in any case. This is what makes the contract above true rather
- // than merely intended.
+ // The control-channel vars are agent-owned in BOTH directions here too
+ // (Start's discipline, see withoutVars), and here they are always
+ // STRIPPED: the seed gets no control listener of its own, so fileEnv is
+ // tenant-supplied and without this a tenant's own env file could hand
+ // its seed a CARLOS_CONTROL_SOCKET the agent never opened and a
+ // CARLOS_PURGE for a channel that does not exist — and the seed has no
+ // serving cache to purge, nor any business setting timers, in any case.
+ // This is what makes the contract above true rather than merely
+ // intended.
cmd.Env = childEnv(withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE"))
// Like the instance child, the seed logs through the agent's own streams
// (journald under systemd) — and its stderr is ALSO teed into a bounded
@@ -488,6 +506,17 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
return nil, "", err
}
+ // Mint the instance's admin token BEFORE the env is read for the spawn, so
+ // the child is handed the same secret the agent will present back to it
+ // (NextDue, the scheduled-work tick) and the same one its control socket
+ // checks bearers against. A failure here fails the wake on purpose: an
+ // instance with no token can neither answer /api/next-due nor accept a
+ // tick, and a wake that quietly skipped it would report a healthy instance
+ // that silently never runs its scheduled work — the #110 kind of lie.
+ if err := ensureAdminToken(filepath.Join(b.DataDir, host+".env")); err != nil {
+ return nil, "", fmt.Errorf("admin token for %s: %w", host, err)
+ }
+
fileEnv, err := b.instanceEnv(host)
if err != nil {
return nil, "", err
@@ -499,14 +528,46 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// passing --control-socket would refuse to start every app that hasn't
// also shipped support for it, a fleet-wide outage on agent upgrade. An
// environment variable a binary doesn't know about is simply ignored.
- // Only set when there's actually a purge listener to talk to, so the
- // child's environment stays honest about what exists.
controlSocket := rt.Socket + ".control"
- // Both purge-channel vars are AGENT-OWNED in BOTH directions: set when the
+ // Bind the control channel BEFORE the spawn, so CARLOS_CONTROL_SOCKET can
+ // follow the OUTCOME rather than the intention. It used to be set from
+ // "b.Cache != nil" and the listener started afterwards; now that the
+ // channel exists for every exec child, the only thing that can make the
+ // variable a lie is a bind that failed (windows has no secureUnixListen at
+ // all, and a stale path can refuse), and binding first is what lets the
+ // else-branch below strip the name in exactly that case.
+ //
+ // A bind failure is still NOT fatal to the wake — same reasoning as before,
+ // the instance serves requests either way; it just goes without its control
+ // verbs until the next restart, a stuck-stale cache self-heals via the
+ // safety-net TTL, and a missed tick is retried by the runner.
+ envPath := filepath.Join(b.DataDir, host+".env")
+ // Read on EVERY call, never captured: the env file can be rewritten under a
+ // running instance (`carlos env`, a re-mint on a later spawn), and a token
+ // cached at bind time would keep authorising the old secret.
+ tokenReader := func() (string, error) {
+ env, err := readEnvFile(envPath)
+ if err != nil {
+ return "", err
+ }
+ return env["CARLOS_ADMIN_TOKEN"], nil
+ }
+ var mount func(*http.ServeMux)
+ if b.Timers != nil {
+ mount = func(mux *http.ServeMux) {
+ schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now)
+ }
+ }
+ ctl, err := edge.ListenControl(controlSocket, rt.Host, b.Cache, mount)
+ if err != nil {
+ log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
+ ctl = nil
+ }
+ // Both control-channel vars are AGENT-OWNED in BOTH directions: set when the
// channel exists, and REMOVED when it does not. The removal half is not
// decoration. fileEnv is tenant-supplied — <host>.config.env comes from the
// bucket via `carlos env`, <host>.env is the instance's own file — so
- // without the else branch a tenant could put CARLOS_PURGE=header (or worse,
+ // without the else branches a tenant could put CARLOS_PURGE=header (or worse,
// CARLOS_CONTROL_SOCKET=/some/path) in its own config and have it survive
// into the child on a box with NO cache. The app would then be told a purge
// channel exists when none does: an app that trusts the var skips its own
@@ -514,7 +575,14 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// chose is a name the agent never vouched for.
//
// "A variable naming a channel must not be set where that channel does
- // nothing" only holds if the agent is the sole author of those names.
+ // nothing" still holds — the two names just answer to different channels
+ // now. CARLOS_CONTROL_SOCKET names the SOCKET, which exists for every exec
+ // child once the bind above succeeded (the purge verb is only one of the
+ // things served on it; the scheduled-work verbs are served there on a box
+ // with no cache at all). CARLOS_PURGE names the PURGE channel, which still
+ // exists only where there is a cache to purge. Both are set from what was
+ // actually created, and stripped otherwise; neither is set from a tenant's
+ // say-so.
//
// CARLOS_PURGE is ADVISORY: it tells the app that this box's edge reads the
// Carlos-Purge response header (internal/edge/purge_header.go), so the app
@@ -525,11 +593,15 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// both are live for exec-backed instances, and converging them is deferred
// (spec §10.3).
childVars := fileEnv
- if b.Cache != nil {
+ if ctl != nil {
childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", controlSocket)
+ } else {
+ childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET")
+ }
+ if b.Cache != nil {
childVars = withVar(childVars, "CARLOS_PURGE", "header")
} else {
- childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE")
+ childVars = withoutVars(childVars, "CARLOS_PURGE")
}
// Record the build we're about to exec — read the symlink immediately before
// the start so it names the version the kernel will resolve. Reading it here,
@@ -559,10 +631,16 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
return cmd
})
if err != nil {
+ // The listener was bound before the spawn; nothing will ever answer on
+ // it now, so unbind it rather than leaving a live socket for a process
+ // that does not exist.
+ if ctl != nil {
+ ctl.Close()
+ }
return nil, "", err
}
- p := &childProc{cmd: cmd, done: make(chan struct{})}
+ p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl}
if b.Cache != nil {
// A fresh process for this host is exactly the moment its cached
// content becomes suspect: a deploy/redeploy means new code and
@@ -572,15 +650,6 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
if n := b.Cache.PurgeHost(rt.Host); n > 0 {
log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
}
- pl, err := edge.ListenPurge(controlSocket, rt.Host, b.Cache)
- if err != nil {
- // Not fatal to the wake — the instance still serves requests, it
- // just can't purge its own cache until the next restart. A
- // stuck-stale cache self-heals via the safety-net TTL anyway.
- log.Printf("activator: purge listener for %s: %v (continuing without it)", rt.Host, err)
- } else {
- p.purge = pl
- }
}
go func() {
p.err = cmd.Wait()
@@ -641,8 +710,10 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// see mergeEnv. Shared by Start and Seed precisely so a seed sees the SAME
// environment the instance will: a seed that read different credentials, or a
// different database URL, than the process it initializes would be a trap.
-// (Start layers its agent-owned purge vars on top of this; those belong to a
-// serving process, not to a one-shot init.)
+// (Start layers its agent-owned control-channel vars on top of this; those
+// belong to a serving process, not to a one-shot init. Both callers mint the
+// instance's CARLOS_ADMIN_TOKEN into <host>.env first, so "the same
+// environment" includes the same secret.)
func (b *ExecBackend) instanceEnv(host string) (map[string]string, error) {
cfgEnv, err := readEnvFile(filepath.Join(b.DataDir, host+".config.env"))
if err != nil && !os.IsNotExist(err) {
@@ -692,7 +763,7 @@ func withVar(env map[string]string, key, value string) map[string]string {
// default. Same copy discipline as withVar: the caller's map is never mutated.
//
// This package keeps its own agent-owned list inline (the two names above,
-// at the b.Cache != nil branch) rather than sharing one with
+// at Start's control-channel branches) rather than sharing one with
// internal/appconfig's agentOwned (agentvars.go), which the unit-tenant sync
// path uses for the same purpose — the two lists must not drift apart.
func withoutVars(env map[string]string, keys ...string) map[string]string {
@@ -825,32 +896,20 @@ func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) e
// readEnvFile parses a KEY=value EnvironmentFile (blank lines and #comments
// ignored).
func readEnvFile(path string) (map[string]string, error) {
- f, err := os.Open(path)
+ raw, err := os.ReadFile(path)
if err != nil {
return nil, err
}
- defer f.Close()
- m := map[string]string{}
- sc := bufio.NewScanner(f)
- for sc.Scan() {
- line := strings.TrimSpace(sc.Text())
- if line == "" || strings.HasPrefix(line, "#") {
- continue
- }
- if k, v, ok := strings.Cut(line, "="); ok {
- m[strings.TrimSpace(k)] = strings.TrimSpace(v)
- }
- }
- return m, sc.Err()
+ return parseEnvBytes(raw), nil
}
// childProc is a spawned child: SIGTERM starts the instance's own graceful
// drain; SIGKILL only if it overstays.
type childProc struct {
- cmd *exec.Cmd
- done chan struct{}
- err error
- purge *edge.PurgeListener // nil if b.Cache was nil or ListenPurge failed
+ cmd *exec.Cmd
+ done chan struct{}
+ err error
+ ctl *edge.ControlListener // nil only if the pre-spawn bind failed
}
func (p *childProc) Exited() bool {
@@ -863,8 +922,8 @@ func (p *childProc) Exited() bool {
}
func (p *childProc) Stop(ctx context.Context) error {
- if p.purge != nil {
- p.purge.Close()
+ if p.ctl != nil {
+ p.ctl.Close()
}
if p.Exited() {
return nil
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 19a565b38..579f868a3 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -1454,6 +1454,12 @@ func versionHeader(ver string, next http.Handler) http.Handler {
})
}
+// UnixProxy is unixProxy for callers outside this package — the scheduled-work
+// runner in internal/host, which delivers a tick to a route's own socket and
+// must NOT take it through the edge's cache, compression or usage layers: a
+// tick is platform traffic, not a visitor's request.
+func UnixProxy(socketPath string) http.Handler { return unixProxy(socketPath) }
+
// unixProxy reverse-proxies to a backend listening on a unix socket,
// preserving the client's Host header (so the backend sees the real hostname)
// and passing WebSocket upgrades straight through.
diff --git a/internal/edge/purge.go b/internal/edge/purge.go
index b5ef0d77a..cada89768 100644
--- a/internal/edge/purge.go
+++ b/internal/edge/purge.go
@@ -6,33 +6,49 @@ import (
"net/http"
)
-// PurgeListener is a per-instance, local-only channel for invalidating
-// that instance's cache entries right after a write — see
-// docs/superpowers/specs/2026-08-02-edge-cache-before-wake-design.md §3-4.
+// ControlListener is an instance's local-only control channel: the socket
+// the agent binds beside the instance's own, through which the instance
+// talks back to the platform. It began as the purge channel alone — see
+// docs/superpowers/specs/2026-08-02-edge-cache-before-wake-design.md §3-4 —
+// and generalized to carry the scheduled-work verbs too
+// (2026-08-23-scheduled-work-design.md §7.2), which is why it now exists
+// for every exec-backed instance rather than only for boxes with a cache.
+//
// It is deliberately NOT reachable over the network: an externally
// reachable purge endpoint would let anyone manufacture wake storms by
// force-purging popular pages, exactly what this feature exists to
-// prevent.
-type PurgeListener struct {
+// prevent — and an externally reachable schedule endpoint would hand the
+// same anyone a timer that wakes an instance on demand.
+type ControlListener struct {
ln net.Listener
srv *http.Server
}
+// PurgeListener is what this was called when purge was the only verb.
+// Retained as an alias so callers and tests that predate the generalisation
+// keep compiling.
+type PurgeListener = ControlListener
+
type purgeRequest struct {
Path string `json:"path"`
Query string `json:"query"`
}
-// ListenPurge binds socketPath (removing any stale file left behind by a
-// crashed previous instance) and serves purge requests scoped to host.
+// ListenControl binds socketPath (removing any stale file left behind by a
+// crashed previous instance) and serves this host's control verbs on it.
+// POST /purge is registered only when cache is non-nil — a box with no edge
+// cache has nothing to purge, and a route that answered /purge with no cache
+// behind it would be a lie. mount, when non-nil, adds the caller's own verbs
+// to the same mux (the scheduled-work ones, schedule.Mount).
//
-// What that scoping actually guarantees: host is fixed here, at
+// What the host scoping actually guarantees: host is fixed here, at
// construction, and never taken from the request — purgeRequest carries
// only Path and Query, never a host — so no request body, however
// malicious or malformed, can make this listener purge any host other than
// the one it was built for. That closes a confused-deputy purge: an
// attacker (or a bug) cannot use one instance's control channel to reach
-// into another host's cache entries THROUGH this listener.
+// into another host's cache entries THROUGH this listener. schedule.Mount
+// takes host by the same discipline, for the same reason.
//
// What it does NOT guarantee: that only the instance this listener was
// created for can ever reach it. Every exec-backed instance on a box runs
@@ -52,6 +68,15 @@ type purgeRequest struct {
// response body to another, since each listener only ever touches the one
// host's cache keys it was built for.
//
+// Purge therefore stays UNAUTHENTICATED, deliberately: every app purging
+// today does it with a bare POST, an auth requirement would break all of
+// them at once, and the blast radius above is bounded to a re-fetch. The
+// verbs mount adds do NOT inherit that: schedule.Mount bearer-gates every
+// one of them on the host's own CARLOS_ADMIN_TOKEN, which a sibling that
+// can compute this path cannot read out of another instance's 0600 env
+// file. The sibling gap is closed for the new verbs and knowingly left
+// open for the old one.
+//
// Unit-backed tenants (carlos-app@<host>.service, backend_unit.go) do NOT
// share this gap, or any control socket at all, because they don't share a
// uid with anything: DynamicUser=yes mints a fresh, unpredictable uid on
@@ -59,28 +84,34 @@ type purgeRequest struct {
// never another tenant's. That's the opposite problem from the one above:
// a DynamicUser tenant can't reach a same-uid sibling's control socket
// because it can't reach ANY control socket, including one built for
-// itself. Unit-backed tenants have no purge listener wired up yet as a
+// itself. Unit-backed tenants have no control listener wired up yet as a
// result (see backend_unit.go); their cached routes rely on the TTL
-// ceiling alone until that gets its own design.
-func ListenPurge(socketPath, host string, cache *Cache) (*PurgeListener, error) {
+// ceiling alone, and their once-timers go through the console rather than
+// the socket, until that gets its own design.
+func ListenControl(socketPath, host string, cache *Cache, mount func(*http.ServeMux)) (*ControlListener, error) {
ln, err := secureUnixListen(socketPath)
if err != nil {
return nil, err
}
mux := http.NewServeMux()
- mux.HandleFunc("POST /purge", func(w http.ResponseWriter, r *http.Request) {
- var req purgeRequest
- if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
- http.Error(w, "bad request", http.StatusBadRequest)
- return
- }
- cache.PurgeURL(host, req.Path, req.Query)
- w.WriteHeader(http.StatusOK)
- })
+ if cache != nil {
+ mux.HandleFunc("POST /purge", func(w http.ResponseWriter, r *http.Request) {
+ var req purgeRequest
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ http.Error(w, "bad request", http.StatusBadRequest)
+ return
+ }
+ cache.PurgeURL(host, req.Path, req.Query)
+ w.WriteHeader(http.StatusOK)
+ })
+ }
+ if mount != nil {
+ mount(mux)
+ }
srv := &http.Server{Handler: mux}
- p := &PurgeListener{ln: ln, srv: srv}
+ p := &ControlListener{ln: ln, srv: srv}
// Close (below) closes p.ln directly, not through srv — so the error
// Serve returns here is whatever a closed-listener Accept produces, not
// necessarily http.ErrServerClosed. Either way it's expected and unused.
@@ -88,12 +119,19 @@ func ListenPurge(socketPath, host string, cache *Cache) (*PurgeListener, error)
return p, nil
}
-// Close stops accepting purge requests, removes the socket file, and tears
+// ListenPurge is ListenControl with nothing but the purge verb — the shape
+// this had before the channel generalized, kept for callers that only want
+// that one verb.
+func ListenPurge(socketPath, host string, cache *Cache) (*PurgeListener, error) {
+ return ListenControl(socketPath, host, cache, nil)
+}
+
+// Close stops accepting control requests, removes the socket file, and tears
// down any connection already accepted before this call — in that order,
// and both steps matter independently:
//
// 1. p.ln.Close() first, deterministically unlinking the socket file and
-// ending Accept regardless of whether the Serve goroutine ListenPurge
+// ending Accept regardless of whether the Serve goroutine ListenControl
// started has ever been scheduled. p.srv.Close() alone can't be trusted
// for this: net/http's Server only closes listeners its own Serve call
// has already registered (in an internal map populated by Serve itself,
@@ -118,7 +156,7 @@ func ListenPurge(socketPath, host string, cache *Cache) (*PurgeListener, error)
// depend on; srv.Close()'s error, if any, is dropped. Calling Close a
// second time (nobody does today) returns "use of closed network
// connection" from the second p.ln.Close().
-func (p *PurgeListener) Close() error {
+func (p *ControlListener) Close() error {
err := p.ln.Close()
p.srv.Close()
return err