amadan

carlos / platform Public

isolated exec: plan 3, the activator's unit path

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YZ1ZvWj7naf73GjTX8MBv5
Paul Campbell pushed by paul@keymail.dev ac978215fc3033ee7bcbf1d837677f782e7f508f parent 08a0ed8
1 file changed, +2203 −0
  • docs/superpowers/plans/2026-09-11-isolated-exec-3-activator.md +2203 −0
diff --git a/docs/superpowers/plans/2026-09-11-isolated-exec-3-activator.md b/docs/superpowers/plans/2026-09-11-isolated-exec-3-activator.md
new file mode 100644
index 000000000..30fd9ca92
--- /dev/null
+++ b/docs/superpowers/plans/2026-09-11-isolated-exec-3-activator.md
@@ -0,0 +1,2203 @@
+# Isolated Exec, Plan 3 of 4: The Activator's Unit Path
+
+> **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:** Teach the exec backend a second spawn path: a `nested` host is launched as `carlos-instance@<iid>` through the control service, with a launch file, per-role log sockets, termination proven by unit state and cgroup emptiness, and the activator's lease, restore, seed, replication, hibernation, control socket and sidecar machinery unchanged. A `flat` host keeps today's `childProc` path byte for byte. Every consumer of the flat shape learns the nested one.
+
+**Architecture:** `ExecBackend` classifies each route (`flat`, `nested`, `fenced`, `conflict`, `foreign`; spec §8.1) and dispatches `Start`, `Seed`, `StartSidecar`, `HasSidecar`, `NextDue` and `instanceEnv` on it. The nested path is new code in `internal/activator` (`layout.go`, `paths.go`, `unitproc.go`, `loglistener.go`, `prepare.go`) plus a `Control` seam whose production implementation is `instancectl.Client`. `activator.go` changes in exactly three places (spec §6): the lease release on an unproven stop, the sidecar restart under the lock, and the route the wake runs on. Consumers of the shape (`syncexec`, the schedule token, `movespass`, the litestream glob) follow the classification. The agent sweeps orphan units and checks the mounts and `/ready` before any nested wake.
+
+**Tech Stack:** Go 1.26, `internal/instancectl` (plan 2: `Escape`, `UnitName`, `SliceName`, `Launch`, `LaunchFileName`, `ExitRecord`, `ExitRecordName`, `StartResponse`, `SeedResponse`, `SweepResponse`, the HTTP protocol of §5.3), `internal/instancefs`, `internal/provision` (plan 1 helpers), `internal/controlsock`, `internal/applog`, `internal/cgroup` tiers.
+
+**Spec:** `docs/superpowers/specs/2026-09-11-isolated-exec-design.md` (revision 12): §6 (all of it), §7, §8.1 (the classification table), §8.5 (the gate the agent honours), §9, §13.
+
+## Global Constraints
+
+- Gate before every commit: `go build ./...`, `go vet ./...`, `gofmt -l .` empty, `go test ./... -count=1`; env `GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off`; socket-binding packages (`internal/activator`, `internal/edge`, `internal/host`, root) need `TMPDIR=/tmp/g` outside the sandbox. Cross-compile gate: darwin, windows, linux-arm64 build; Linux-only syscalls live in `_linux.go` with `!linux` stubs.
+- **A flat host's behaviour does not change.** The existing `Start`/`Seed`/`StartSidecar` bodies become `startChild`/`seedChild`/`startSidecarChild` and are called for `flat` and `foreign` rows unchanged; every existing activator test keeps passing without edits to its assertions.
+- Classification (spec §8.1): `flat` = `DBPath == provision.DefaultDBPath(dataDir, host) && Socket == provision.DefaultSocket(host)`; `nested` = `DBPath == provision.NestedDBPath(dataDir, host) && Socket == provision.NestedSocket(host)`; `fenced` = a journal exists at `provision.LayoutJournalPath(dataDir, host)`; `conflict` = another activator-managed row on the box has the same `provision.InstanceID`; anything else `foreign`. `fenced` and `conflict` refuse to wake with a reason; `foreign` spawns flat.
+- Nested `Start` order (spec §6.1): prepare (dirs, admin token, env, control listener, log listener, launch file) → `POST /start` with the tier → `waitHealthyCtx` on `rt.Socket` → `0660` normalisation → return a `unitProc` with the invocation id. On health failure: `POST /stop`, wait for `done`, clean up.
+- `childEnv` produces one entry per key, last value wins (spec §6.2, codex R1-16); the launch file's `Env` is that list, `Argv` is `[/opt/carlos/live/<host>, --socket, rt.Socket, --db, dbPath]` for run; for seed the parsed `rt.Seed` argv plus `--db dbPath` exactly as `seedChild` builds it; for sidecar `[/opt/carlos/live/<host>, sidecar, run]`.
+- Termination (spec §6.5, codex R2-5): `done` closes only when the unit is terminal (`ActiveState` `inactive` or `failed` with the proc's own invocation, or `inactive` with an empty invocation) AND the unit's cgroup is absent or `populated 0`. Log EOF never means exit. The exit record is wording only. A foreign invocation is stopped and waited for.
+- `unitProc.Stop` returns `ErrStillRunning` when `ctx` expires before `done`; `stop()` then does not call `rep.Close` (spec §6.3).
+- Every control call carries a deadline: 30s for start/stop/reset, 150s for seed, 60s for sweep, 5s for ready (spec §5.3).
+- Before any nested wake the agent requires: all three mounts carry `nosymfollow` (`instancefs.MountNoSymfollow`), the control socket answers `GET /ready` with 204; otherwise the wake refuses with the reason (spec §8.5, §8.6).
+- Copy rules: event lines keep today's wording (`instance exited: ...`, `instance exited (exit 0)`, plus `instance exited (status unknown: no exit record)`); no em dashes; comments are full sentences explaining why.
+- Commit after every task with the session's trailer. Branch `worktree-isolated-exec-spec`.
+
+---
+
+## File Structure
+
+| File | Responsibility |
+|---|---|
+| `internal/provision/provision.go` (modify) | `LayoutJournalPath(dataDir, host)`; `InstanceRunDirOf`, `LogSocket` already exist |
+| `internal/instancefs/instancefs.go` + `_linux.go` + `_other.go` (modify) | `ReadBeneathLimit(dir, name, max)` |
+| `internal/instancectl/client.go` (create) | `Client`: the agent-side HTTP client over the control socket, one method per verb, with the deadlines |
+| `internal/instancectl/client_test.go` (create) | client against a `Server` with fakes |
+| `internal/activator/layout.go` (create) | `Layout`, `LayoutInputs`, `Classify` |
+| `internal/activator/paths.go` (create) | `instancePaths` and `pathsFor(rt)`: every per-host path by layout |
+| `internal/activator/control.go` (create) | the `Control` interface the backend calls; `UnitState` and `CgroupEvents` seams |
+| `internal/activator/loglistener.go` (create) | `logListener` |
+| `internal/activator/unitproc.go` (create) | `unitProc`, the exit observer, `ErrStillRunning` |
+| `internal/activator/prepare.go` (create) | `prepare(rt, role)` and the launch-file writer |
+| `internal/activator/backend_exec.go` (modify) | dispatch by layout; `startUnit`, `seedUnit`, `startSidecarUnit`; `childEnv` dedupe; path helpers replaced by `pathsFor` |
+| `internal/activator/activator.go` (modify) | the three changes of §6.3, §6.7, §6.8; `Routes` seam; `SweepOrphans` hook |
+| `internal/activator/*_test.go` (create/modify) | tests per task |
+| `internal/appconfig/syncexec.go` (modify) | nested write path; reap walks `instances/*/` |
+| `internal/schedule/runner.go`, `internal/host/schedule_wiring.go` (modify) | token path by layout |
+| `internal/host/movespass.go` (modify) | DB path from the row; refuse a fenced host |
+| `internal/ops/litestream.go` (modify) | glob gains `instances/*/*.db` |
+| `internal/host/agent.go` (modify) | wiring: `Control`, `LayoutInputs`, `CgroupRoot`, `Routes`, startup sweep, mount and ready gate |
+
+---
+
+### Task 1: Journal path, `ReadBeneathLimit`, and the control client
+
+**Files:**
+- Modify: `internal/provision/provision.go` (append after `LogSocket`)
+- Modify: `internal/instancefs/instancefs.go`, `instancefs_linux.go`, `instancefs_other.go`
+- Create: `internal/instancectl/client.go`, `internal/instancectl/client_test.go`
+- Test: `internal/provision/provision_test.go`, `internal/instancefs/instancefs_linux_test.go` (append)
+
+**Interfaces:**
+- Produces:
+ ```go
+ // provision
+ func LayoutJournalPath(dataDir, host string) string // <dataDir>/layout/<host>.json
+ // instancefs
+ var ErrTooLarge = errors.New("instancefs: file larger than the limit")
+ func ReadBeneathLimit(dir, name string, max int64) ([]byte, error)
+ // instancectl
+ type Client struct { Path string; Dial func(ctx context.Context, path string) (net.Conn, error) } // Dial nil = unixdial-free plain dial (the control socket is not an upstream)
+ func NewClient(path string) *Client
+ func (c *Client) Start(ctx context.Context, host string, role Role, memoryMax int64) (string, error) // invocation id
+ func (c *Client) Seed(ctx context.Context, host string, memoryMax int64) (SeedResponse, error)
+ func (c *Client) Stop(ctx context.Context, host string, role Role) error
+ func (c *Client) Reset(ctx context.Context, host string, role Role) error
+ func (c *Client) Sweep(ctx context.Context) (SweepResponse, error)
+ func (c *Client) Ready(ctx context.Context) error // nil on 204; *NotReadyError carrying the ProbeReport on 503
+ type NotReadyError struct{ Report ProbeReport }
+ const ( DeadlineControl = 30*time.Second; DeadlineSeed = 150*time.Second; DeadlineSweep = 60*time.Second; DeadlineReady = 5*time.Second )
+ ```
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/provision/provision_test.go`:
+
+```go
+func TestLayoutJournalPath(t *testing.T) {
+ if got := LayoutJournalPath("/data/", "h.example.com"); got != "/data/layout/h.example.com.json" {
+ t.Errorf("LayoutJournalPath = %q", got)
+ }
+}
+```
+
+Append to `internal/instancefs/instancefs_linux_test.go`:
+
+```go
+func TestReadBeneathLimitRefusesALargeFile(t *testing.T) {
+ dir := t.TempDir()
+ if err := os.WriteFile(filepath.Join(dir, "big"), make([]byte, 4096), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ReadBeneathLimit(dir, "big", 1024); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("err = %v, want ErrTooLarge", err)
+ }
+ got, err := ReadBeneathLimit(dir, "big", 4096)
+ if err != nil || len(got) != 4096 {
+ t.Fatalf("exact limit: %d bytes, %v", len(got), err)
+ }
+}
+```
+
+Create `internal/instancectl/client_test.go`:
+
+```go
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+// serveFake starts a Server on a temp unix socket with the given fake
+// systemctl and a probe report, and returns a Client for it.
+func serveFake(t *testing.T, ctl *fakeCtl, report *ProbeReport) *Client {
+ t.Helper()
+ root := t.TempDir()
+ s := &Server{
+ Systemctl: ctl.run, PeerOK: func(net.Conn) error { return nil },
+ CgroupRoot: filepath.Join(root, "cgroup"), ProbeOut: filepath.Join(root, "probe.out"),
+ FixtureOK: func() error { return nil }, ProbeEvery: time.Hour, Logf: t.Logf,
+ }
+ if report != nil {
+ b, _ := json.Marshal(report)
+ if err := os.WriteFile(s.ProbeOut, b, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ s.RunProbe(context.Background())
+ }
+ dir, err := os.MkdirTemp("", "icc")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ sock := filepath.Join(dir, "c.sock")
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ go s.Serve(ctx, ln) //nolint:errcheck
+ t.Cleanup(func() { ln.Close() })
+ return NewClient(sock)
+}
+
+func TestClientStartCarriesTierAndReturnsInvocation(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
+ c := serveFake(t, ctl, nil)
+ inv, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912)
+ if err != nil || inv != "inv1" {
+ t.Fatalf("Start = %q, %v", inv, err)
+ }
+ if !strings.Contains(ctl.joined(), "MemoryMax=536870912") {
+ t.Errorf("tier not sent: %s", ctl.joined())
+ }
+}
+
+func TestClientSeedStopSweepReady(t *testing.T) {
+ ctl := &fakeCtl{answers: map[string]string{
+ "show -p InvocationID,Result,ExecMainStatus,ExecMainCode": "InvocationID=s1\nResult=success\nExecMainStatus=0\nExecMainCode=exited\n",
+ "list-units": "", "list-jobs": "",
+ }}
+ c := serveFake(t, ctl, &ProbeReport{Ready: true})
+ res, err := c.Seed(context.Background(), "h.example.com", 536870912)
+ if err != nil || res.Result != "success" || res.InvocationID != "s1" {
+ t.Fatalf("Seed = %+v, %v", res, err)
+ }
+ if err := c.Stop(context.Background(), "h.example.com", RoleSidecar); err != nil {
+ t.Fatalf("Stop: %v", err)
+ }
+ if err := c.Reset(context.Background(), "h.example.com", RoleRun); err != nil {
+ t.Fatalf("Reset: %v", err)
+ }
+ sw, err := c.Sweep(context.Background())
+ if err != nil || len(sw.Units) != 0 {
+ t.Fatalf("Sweep = %+v, %v", sw, err)
+ }
+ if err := c.Ready(context.Background()); err != nil {
+ t.Fatalf("Ready: %v", err)
+ }
+}
+
+func TestClientReadyReturnsTheReport(t *testing.T) {
+ ctl := &fakeCtl{}
+ c := serveFake(t, ctl, &ProbeReport{Ready: false, Lines: []VerdictLine{{Check: "pid1", Verdict: "fail", Detail: "pid=7"}}})
+ err := c.Ready(context.Background())
+ var nr *NotReadyError
+ if !errors.As(err, &nr) || len(nr.Report.Lines) != 1 || nr.Report.Lines[0].Check != "pid1" {
+ t.Fatalf("Ready = %v", err)
+ }
+}
+
+func TestClientSurfacesA502AsAnError(t *testing.T) {
+ ctl := &fakeCtl{fail: map[string]error{"start": errors.New("boom")}} // keyed the way fakeCtl.run matches: by leading-argument prefix
+ c := serveFake(t, ctl, nil)
+ if _, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912); err == nil || !strings.Contains(err.Error(), "boom") {
+ t.Fatalf("Start error = %v, want the server's message", err)
+ }
+}
+
+func TestClientRefusesWhenTheSocketIsAbsent(t *testing.T) {
+ c := NewClient(filepath.Join(t.TempDir(), "absent.sock"))
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if err := c.Ready(ctx); err == nil || !strings.Contains(err.Error(), "instance control socket") {
+ t.Fatalf("absent socket: %v", err)
+ }
+}
+```
+
+`fakeCtl` already exists in `server_test.go` (same package).
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/provision ./internal/instancefs ./internal/instancectl -run 'TestLayoutJournalPath|TestReadBeneathLimit|TestClient' -count=1`
+Expected: build failures for the three undefined symbols.
+
+- [ ] **Step 3: Write the code**
+
+`internal/provision/provision.go`, after `LogSocket`:
+
+```go
+// LayoutJournalPath is the migration journal for a host in flight
+// (isolated-exec design §8.2): under <dataDir>/layout, outside every tenant
+// mount. Its presence is what makes a host `fenced` (§8.1); plan 4 writes
+// it, and every consumer that must not act on a half-migrated host checks
+// for it here.
+func LayoutJournalPath(dataDir, host string) string {
+ return filepath.Join(dataDir, "layout", host+".json")
+}
+```
+
+`internal/instancefs/instancefs.go`:
+
+```go
+// ErrTooLarge is returned by ReadBeneathLimit when the file exceeds max.
+var ErrTooLarge = errors.New("instancefs: file larger than the limit")
+
+// ReadBeneathLimit is ReadBeneath with a size cap. Files in an instance
+// directory are tenant-writable, and some are read by the agent outside the
+// tenant's memory wall (the exit record, the launch file), so no read there
+// is unbounded.
+func ReadBeneathLimit(dir, name string, max int64) ([]byte, error) {
+ f, err := OpenBeneath(dir, name, os.O_RDONLY, 0)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ b, err := io.ReadAll(io.LimitReader(f, max+1))
+ if err != nil {
+ return nil, err
+ }
+ if int64(len(b)) > max {
+ return nil, fmt.Errorf("%w: %s over %d bytes", ErrTooLarge, filepath.Join(dir, name), max)
+ }
+ return b, nil
+}
+```
+
+(add `"fmt"` to that file's imports; `OpenBeneath` on non-Linux returns `ErrUnsupported`, so no stub is needed.)
+
+`internal/instancectl/client.go`:
+
+```go
+package instancectl
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/url"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// Deadlines per verb (design §5.3). A start or stop that exceeds its
+// deadline is a failed wake or an unproven stop, never a hang: §6.7 puts
+// one such call on a path that must not stall the activator. The seed's is
+// long because a seed is allowed to be slow.
+const (
+ DeadlineControl = 30 * time.Second
+ DeadlineSeed = 150 * time.Second
+ DeadlineSweep = 60 * time.Second
+ DeadlineReady = 5 * time.Second
+)
+
+// Client is the agent's side of the control channel. It is plain HTTP over
+// the unix socket at Path; unixdial's checks are for UPSTREAMS the edge
+// proxies to, and this socket is the one path they refuse, so the client
+// dials it directly.
+type Client struct {
+ Path string
+ // Dial is the dialer; nil means net.Dialer. Tests do not need to override
+ // it, but a caller composing several clients may.
+ Dial func(ctx context.Context, path string) (net.Conn, error)
+}
+
+func NewClient(path string) *Client { return &Client{Path: path} }
+
+// NotReadyError carries the probe verdict a 503 /ready returned.
+type NotReadyError struct{ Report ProbeReport }
+
+func (e *NotReadyError) Error() string {
+ var fails []string
+ for _, l := range e.Report.Lines {
+ if l.Verdict != "ok" {
+ fails = append(fails, l.Check+" "+l.Verdict+" "+l.Detail)
+ }
+ }
+ if len(fails) == 0 {
+ return "instance control: box not ready (no verdict yet)"
+ }
+ return "instance control: box not ready: " + strings.Join(fails, "; ")
+}
+
+func (c *Client) httpClient() *http.Client {
+ dial := c.Dial
+ if dial == nil {
+ dial = func(ctx context.Context, path string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ }
+ }
+ return &http.Client{Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { return dial(ctx, c.Path) },
+ DisableKeepAlives: true,
+ }}
+}
+
+// do issues one verb with the given deadline and decodes a JSON body when
+// there is one. A non-2xx answer becomes an error carrying the server's
+// message; a dial failure names the socket so a forgotten enable reads as
+// what it is from carlos logs.
+func (c *Client) do(ctx context.Context, deadline time.Duration, method, path string, q url.Values, out any) (int, error) {
+ ctx, cancel := context.WithTimeout(ctx, deadline)
+ defer cancel()
+ u := "http://control" + path
+ if len(q) > 0 {
+ u += "?" + q.Encode()
+ }
+ req, err := http.NewRequestWithContext(ctx, method, u, nil)
+ if err != nil {
+ return 0, err
+ }
+ resp, err := c.httpClient().Do(req)
+ if err != nil {
+ return 0, fmt.Errorf("instance control socket %s: %w", c.Path, err)
+ }
+ defer resp.Body.Close()
+ body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+ if resp.StatusCode == http.StatusServiceUnavailable && path == "/ready" {
+ var r ProbeReport
+ _ = json.Unmarshal(body, &r)
+ return resp.StatusCode, &NotReadyError{Report: r}
+ }
+ if resp.StatusCode/100 != 2 {
+ return resp.StatusCode, fmt.Errorf("instance control %s: %d: %s", path, resp.StatusCode, strings.TrimSpace(string(body)))
+ }
+ if out != nil && len(body) > 0 {
+ if err := json.Unmarshal(body, out); err != nil {
+ return resp.StatusCode, fmt.Errorf("instance control %s: %v", path, err)
+ }
+ }
+ return resp.StatusCode, nil
+}
+
+func hostRole(host string, role Role) url.Values {
+ q := url.Values{"host": {host}}
+ if role != "" {
+ q.Set("role", string(role))
+ }
+ return q
+}
+
+// Start starts the run or sidecar unit for host with the memory tier and
+// returns the invocation id the server read after the start.
+func (c *Client) Start(ctx context.Context, host string, role Role, memoryMax int64) (string, error) {
+ q := hostRole(host, role)
+ q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
+ var out StartResponse
+ if _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/start", q, &out); err != nil {
+ return "", err
+ }
+ return out.InvocationID, nil
+}
+
+// Seed runs the seed unit to completion and returns its retained result.
+func (c *Client) Seed(ctx context.Context, host string, memoryMax int64) (SeedResponse, error) {
+ q := hostRole(host, "")
+ q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
+ var out SeedResponse
+ _, err := c.do(ctx, DeadlineSeed, http.MethodPost, "/seed", q, &out)
+ return out, err
+}
+
+func (c *Client) Stop(ctx context.Context, host string, role Role) error {
+ _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/stop", hostRole(host, role), nil)
+ return err
+}
+
+func (c *Client) Reset(ctx context.Context, host string, role Role) error {
+ _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/reset", hostRole(host, role), nil)
+ return err
+}
+
+// Sweep stops every instance unit on the box (design §6.7). A 502 carries
+// the units that were stopped and the ones that would not; both are
+// returned with the error.
+func (c *Client) Sweep(ctx context.Context) (SweepResponse, error) {
+ var out SweepResponse
+ status, err := c.do(ctx, DeadlineSweep, http.MethodPost, "/sweep", nil, &out)
+ if err != nil && status == http.StatusBadGateway {
+ return out, errors.Join(err, errors.New("some instance units would not stop"))
+ }
+ return out, err
+}
+
+// Ready is nil on 204 and a *NotReadyError on 503.
+func (c *Client) Ready(ctx context.Context) error {
+ _, err := c.do(ctx, DeadlineReady, http.MethodGet, "/ready", nil, nil)
+ return err
+}
+```
+
+`Sweep`'s 502 decoding: `do` returns before decoding on non-2xx; make `do` decode `out` for a 502 on `/sweep` too (add `|| (resp.StatusCode == http.StatusBadGateway && path == "/sweep")` to the decode condition, before the non-2xx return), so the caller sees the partial list.
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/provision ./internal/instancefs ./internal/instancectl -count=1` (outside the sandbox)
+Expected: PASS.
+
+- [ ] **Step 5: gofmt and commit**
+
+```bash
+gofmt -l internal/provision internal/instancefs internal/instancectl
+git add internal/provision internal/instancefs internal/instancectl
+git commit -m "instancectl client, LayoutJournalPath, ReadBeneathLimit (plan 3, task 1)"
+```
+
+---
+
+### Task 2: Layout classification and per-host paths
+
+**Files:**
+- Create: `internal/activator/layout.go`, `internal/activator/paths.go`
+- Create: `internal/activator/layout_test.go`
+
+**Interfaces:**
+- Produces:
+ ```go
+ type Layout string
+ const ( LayoutFlat Layout = "flat"; LayoutNested Layout = "nested"; LayoutFenced Layout = "fenced"; LayoutConflict Layout = "conflict"; LayoutForeign Layout = "foreign" )
+ type LayoutInputs struct {
+ DataDir string
+ Journal func(host string) bool // nil: never fenced
+ Peers func() ([]registry.Route, error) // nil: never conflict
+ }
+ func Classify(rt registry.Route, in LayoutInputs) (Layout, string) // layout and a reason for fenced/conflict/foreign
+ func ActivatorManaged(rt registry.Route) bool // Kind instance, Hibernate, Backing "", Tunnel "", Socket != ""
+ type instancePaths struct { layout Layout; dataDir, runDir, dbPath, envFile, sidecarEnv, configEnv string }
+ func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths
+ ```
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/activator/layout_test.go`:
+
+```go
+package activator
+
+import (
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func managed(host, db, sock string) registry.Route {
+ return registry.Route{Host: host, Kind: "instance", Hibernate: true, DBPath: db, Socket: sock}
+}
+
+func TestClassifyFlatAndNested(t *testing.T) {
+ in := LayoutInputs{DataDir: "/data"}
+ if l, _ := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFlat {
+ t.Errorf("flat row = %s", l)
+ }
+ if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), provision.NestedSocket("h.example.com")), in); l != LayoutNested {
+ t.Errorf("nested row = %s", l)
+ }
+ if l, r := Classify(managed("h.example.com", "/data/sub/store.db", "/run/carlos/h.example.com.sock"), in); l != LayoutForeign || !strings.Contains(r, "store.db") {
+ t.Errorf("custom row = %s %q", l, r)
+ }
+ // A nested DB with a flat socket is not nested: both columns must agree.
+ if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), "/run/carlos/h.example.com.sock"), in); l != LayoutForeign {
+ t.Errorf("half-migrated row = %s", l)
+ }
+}
+
+func TestClassifyFencedWinsOverEverything(t *testing.T) {
+ in := LayoutInputs{DataDir: "/data", Journal: func(h string) bool { return h == "h.example.com" }}
+ if l, r := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFenced || r == "" {
+ t.Errorf("fenced = %s %q", l, r)
+ }
+ if l, _ := Classify(managed("o.example.com", "/data/o.example.com.db", "/run/carlos/o.example.com.sock"), in); l != LayoutFlat {
+ t.Errorf("other host = %s", l)
+ }
+}
+
+func TestClassifyConflictRefusesBothHosts(t *testing.T) {
+ long := strings.Repeat("a", 60) + ".example.com"
+ twin := strings.Repeat("a", 60) + ".example.org"
+ if provision.InstanceID(long) == provision.InstanceID(twin) {
+ t.Skip("fixture hosts do not collide; pick two that do")
+ }
+ // Force a collision through the Peers seam: a peer whose InstanceID we
+ // fake by giving it the SAME host string is not a conflict (it is the
+ // same row); so build the collision with a peer that is a different
+ // host mapping to the same identifier, which needs a real hash
+ // collision. Instead test the rule directly: a nested row whose
+ // identifier is shared by another managed row classifies as conflict.
+ a := managed(long, provision.NestedDBPath("/data", long), provision.NestedSocket(long))
+ b := managed(twin, provision.NestedDBPath("/data", twin), provision.NestedSocket(twin))
+ in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, b}, nil }, SameID: func(x, y string) bool { return true }}
+ if l, r := Classify(a, in); l != LayoutConflict || !strings.Contains(r, twin) {
+ t.Errorf("conflict = %s %q", l, r)
+ }
+}
+
+func TestClassifyIgnoresUnmanagedPeers(t *testing.T) {
+ a := managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock")
+ ext := registry.Route{Host: "x.example.com", Kind: "instance", Hibernate: false, Socket: "/run/carlos/x.example.com.sock"}
+ in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, ext}, nil }, SameID: func(x, y string) bool { return true }}
+ if l, _ := Classify(a, in); l != LayoutFlat {
+ t.Errorf("an unmanaged peer must not conflict: %s", l)
+ }
+ if ActivatorManaged(ext) {
+ t.Error("unflagged socket route reported managed")
+ }
+}
+```
+
+`SameID` is a test seam (`func(a, b string) bool`, nil = `provision.InstanceID(a) == provision.InstanceID(b)`) so a collision can be exercised without a real hash collision.
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run TestClassify -count=1`
+Expected: build failure.
+
+- [ ] **Step 3: Write `layout.go` and `paths.go`**
+
+`internal/activator/layout.go`:
+
+```go
+package activator
+
+import (
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Layout is the classification of an activator-managed route's paths
+// (isolated-exec design §8.1). It decides the spawn path: flat and foreign
+// rows run today's exec child; nested rows run as units; fenced and
+// conflict rows refuse to wake.
+type Layout string
+
+const (
+ LayoutFlat Layout = "flat"
+ LayoutNested Layout = "nested"
+ LayoutFenced Layout = "fenced"
+ LayoutConflict Layout = "conflict"
+ LayoutForeign Layout = "foreign"
+)
+
+// LayoutInputs are the box facts Classify needs beyond the row: where the
+// data dir is, whether a migration journal exists for a host (plan 4
+// writes them), and the other rows on the box (for identifier conflicts).
+// Nil seams mean "never fenced" and "never conflict", which is what unit
+// tests without a box want.
+type LayoutInputs struct {
+ DataDir string
+ Journal func(host string) bool
+ Peers func() ([]registry.Route, error)
+ // SameID reports whether two hosts share an instance identifier. Nil
+ // means provision.InstanceID equality; tests inject a collision.
+ SameID func(a, b string) bool
+}
+
+// ActivatorManaged is the activator's own predicate for a route it owns
+// (activator.go's per-host gate and edge.go's proxy decision): an instance
+// route, hibernate-flagged, with a socket, not unit-backed, not
+// tunnel-served. An unflagged socket route is an externally supervised
+// process and is nobody's to migrate or isolate (design §8.2 step 1).
+func ActivatorManaged(rt registry.Route) bool {
+ return rt.Kind == "instance" && rt.Hibernate && rt.Socket != "" && rt.Backing == "" && rt.Tunnel == ""
+}
+
+// Classify returns the layout and, for anything but flat and nested, the
+// reason a member or operator sees.
+func Classify(rt registry.Route, in LayoutInputs) (Layout, string) {
+ if in.Journal != nil && in.Journal(rt.Host) {
+ return LayoutFenced, "layout migration in flight; see " + provision.LayoutJournalPath(in.DataDir, rt.Host)
+ }
+ if in.Peers != nil {
+ same := in.SameID
+ if same == nil {
+ same = func(a, b string) bool { return provision.InstanceID(a) == provision.InstanceID(b) }
+ }
+ peers, err := in.Peers()
+ if err == nil {
+ for _, p := range peers {
+ if p.Host != rt.Host && ActivatorManaged(p) && same(p.Host, rt.Host) {
+ return LayoutConflict, fmt.Sprintf("instance identifier collides with %s; neither host wakes until one is removed", p.Host)
+ }
+ }
+ }
+ }
+ switch {
+ case rt.DBPath == provision.DefaultDBPath(in.DataDir, rt.Host) && rt.Socket == provision.DefaultSocket(rt.Host):
+ return LayoutFlat, ""
+ case rt.DBPath == provision.NestedDBPath(in.DataDir, rt.Host) && rt.Socket == provision.NestedSocket(rt.Host):
+ return LayoutNested, ""
+ }
+ return LayoutForeign, fmt.Sprintf("db %q socket %q are neither the flat nor the nested defaults", rt.DBPath, rt.Socket)
+}
+```
+
+`internal/activator/paths.go`:
+
+```go
+package activator
+
+import (
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// instancePaths is every per-host path the exec backend touches, resolved
+// once per call by layout so that no method builds a path from DataDir and
+// the host on its own. Flat and foreign rows keep today's flat names; a
+// nested row's files live in its instance directory with the same names
+// (design §3.1).
+type instancePaths struct {
+ layout Layout
+ reason string
+ dataDir string // the directory that holds the four files
+ runDir string // the directory that holds app.sock and the control socket
+ dbPath string
+ envFile string
+ sidecarEnv string
+ configEnv string
+}
+
+func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
+ l, reason := Classify(rt, b.layoutInputs())
+ p := instancePaths{layout: l, reason: reason, dataDir: b.DataDir, runDir: provision.SocketDir}
+ if l == LayoutNested {
+ p.dataDir = provision.InstanceDir(b.DataDir, rt.Host)
+ p.runDir = provision.InstanceRunDir(rt.Host)
+ }
+ p.dbPath = rt.DBPath
+ if p.dbPath == "" {
+ p.dbPath = provision.DefaultDBPath(b.DataDir, rt.Host)
+ }
+ p.envFile = filepath.Join(p.dataDir, rt.Host+".env")
+ p.sidecarEnv = filepath.Join(p.dataDir, rt.Host+"-sidecar.env")
+ p.configEnv = filepath.Join(p.dataDir, rt.Host+".config.env")
+ return p
+}
+
+// layoutInputs is the backend's LayoutInputs with DataDir filled in.
+func (b *ExecBackend) layoutInputs() LayoutInputs {
+ in := b.Layout
+ in.DataDir = b.DataDir
+ return in
+}
+```
+
+Add to `ExecBackend` (backend_exec.go struct) the fields this task introduces, with comments:
+
+```go
+ // Layout supplies the box facts the classification of §8.1 needs beyond
+ // the row: journals and peers. Zero value: never fenced, never conflict.
+ Layout LayoutInputs
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestClassify' -count=1`
+Expected: PASS.
+
+- [ ] **Step 5: Replace the flat path builders with `pathsFor`**
+
+In `backend_exec.go`: `defaultDBPath` is deleted; every `filepath.Join(b.DataDir, host+".env")` (`Seed`, `Start`, `NextDue`, `instanceEnv`), `filepath.Join(b.DataDir, host+".config.env")` (`instanceEnv`) and `sidecarEnvPath` become reads of `b.pathsFor(rt)` fields. `instanceEnv(host string)` becomes `instanceEnv(rt registry.Route)`. For a flat row every resulting path is byte-identical to today's, which the existing tests prove.
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -count=1` (outside the sandbox)
+Expected: PASS with no test edits.
+
+- [ ] **Step 6: gofmt and commit**
+
+```bash
+gofmt -l internal/activator
+git add internal/activator
+git commit -m "activator: layout classification and per-host paths by layout (plan 3, task 2)"
+```
+
+---
+
+### Task 3: `childEnv` deduplication, the `Control` seam, the log listener
+
+**Files:**
+- Modify: `internal/activator/backend_exec.go` (`childEnv`)
+- Create: `internal/activator/control.go`, `internal/activator/loglistener.go`
+- Create: `internal/activator/loglistener_test.go`; modify `internal/activator/activator_test.go` (one new test for `childEnv`)
+
+**Interfaces:**
+- Produces:
+ ```go
+ // control.go
+ type Control interface {
+ Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
+ Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
+ Stop(ctx context.Context, host string, role instancectl.Role) error
+ Reset(ctx context.Context, host string, role instancectl.Role) error
+ Sweep(ctx context.Context) (instancectl.SweepResponse, error)
+ Ready(ctx context.Context) error
+ }
+ // UnitState answers systemctl show -p ActiveState,InvocationID for a unit; production execs systemctl (unprivileged).
+ type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)
+ // CgroupEvents reports whether the unit's cgroup exists and is populated, from <root>/carlos.slice/carlos-instances.slice/<slice>/<unit>/cgroup.events.
+ type CgroupEvents func(slice, unit string) (exists, populated bool, err error)
+ func systemctlUnitState(ctx context.Context, unit string) (string, string, error)
+ func cgroupEventsUnder(root string) CgroupEvents
+ // loglistener.go
+ type logListener struct{ ... }
+ func listenLogs(path string, emit func(line string)) (*logListener, error) // binds path 0600; accepts ONE connection per Reset; tees lines to os.Stdout and emit
+ func (l *logListener) Close() error
+ ```
+ `instancectl.Client` satisfies `Control` (compile-time assertion in `control.go`: `var _ Control = (*instancectl.Client)(nil)`).
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/activator/activator_test.go`:
+
+```go
+func TestChildEnvKeepsOneEntryPerKeyLastWins(t *testing.T) {
+ t.Setenv("PATH", "/usr/bin")
+ env := childEnv(map[string]string{"PATH": "/tenant/bin", "A": "1"})
+ seen := map[string]string{}
+ for _, kv := range env {
+ k, v, _ := strings.Cut(kv, "=")
+ if _, dup := seen[k]; dup {
+ t.Fatalf("duplicate key %s in %v", k, env)
+ }
+ seen[k] = v
+ }
+ if seen["PATH"] != "/tenant/bin" {
+ t.Errorf("PATH = %q, want the tenant's value to win over the inherited one", seen["PATH"])
+ }
+ if seen["A"] != "1" {
+ t.Errorf("A = %q", seen["A"])
+ }
+}
+```
+
+Create `internal/activator/loglistener_test.go`:
+
+```go
+package activator
+
+import (
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+func TestLogListenerTeesOneConnectionAndRefusesASecond(t *testing.T) {
+ dir, err := os.MkdirTemp("", "ll")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ path := filepath.Join(dir, "h.run.sock")
+ var mu sync.Mutex
+ var lines []string
+ l, err := listenLogs(path, func(s string) { mu.Lock(); lines = append(lines, s); mu.Unlock() })
+ if err != nil {
+ t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ defer l.Close()
+ c, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := c.Write([]byte("hello\nworld\n")); err != nil {
+ t.Fatal(err)
+ }
+ c.Close()
+ deadline := time.Now().Add(2 * time.Second)
+ for {
+ mu.Lock()
+ n := len(lines)
+ mu.Unlock()
+ if n == 2 || time.Now().After(deadline) {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ mu.Lock()
+ got := strings.Join(lines, "|")
+ mu.Unlock()
+ if got != "hello|world" {
+ t.Fatalf("lines = %q", got)
+ }
+ // A second connection during the same invocation is closed unread.
+ c2, err := net.Dial("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c2.SetReadDeadline(time.Now().Add(time.Second))
+ buf := make([]byte, 1)
+ if _, err := c2.Read(buf); err == nil {
+ t.Fatal("second connection was not closed")
+ }
+ c2.Close()
+ if l.Exited() {
+ t.Error("EOF on the log connection must not mark the process exited")
+ }
+}
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestChildEnvKeepsOneEntry|TestLogListener' -count=1`
+Expected: `TestChildEnvKeepsOneEntry` FAILS (duplicate PATH today); `TestLogListener` fails to build.
+
+- [ ] **Step 3: Write the code**
+
+`childEnv` in `backend_exec.go` becomes:
+
+```go
+// childEnv is the child's complete environment: the agent's own, minus
+// every CARLOS_* variable, with the instance's file values layered on top.
+// One entry per key, last value wins: os/exec deduplicates cmd.Env that
+// way, but the unit path hands this list to syscall.Exec, which passes
+// duplicates through, and a Go child then keeps the FIRST occurrence
+// (design §6.2, codex R1-16). Composing it deduplicated here keeps the
+// two spawn paths identical.
+func childEnv(fileEnv map[string]string) []string {
+ order := make([]string, 0, 32)
+ values := make(map[string]string, 32)
+ set := func(k, v string) {
+ if _, seen := values[k]; !seen {
+ order = append(order, k)
+ }
+ values[k] = v
+ }
+ for _, kv := range os.Environ() {
+ if strings.HasPrefix(kv, "CARLOS_") {
+ continue
+ }
+ k, v, ok := strings.Cut(kv, "=")
+ if !ok {
+ continue
+ }
+ set(k, v)
+ }
+ keys := make([]string, 0, len(fileEnv))
+ for k := range fileEnv {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ set(k, fileEnv[k])
+ }
+ env := make([]string, 0, len(order))
+ for _, k := range order {
+ env = append(env, k+"="+values[k])
+ }
+ return env
+}
+```
+
+(`sort` import; map iteration order for `fileEnv` is made deterministic by sorting keys, which also makes the launch file stable across writes.)
+
+`internal/activator/control.go`:
+
+```go
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+)
+
+// Control is what the exec backend needs from the root instance-control
+// channel (design §5.3). instancectl.Client is the production
+// implementation; tests use a fake.
+type Control interface {
+ Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
+ Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
+ Stop(ctx context.Context, host string, role instancectl.Role) error
+ Reset(ctx context.Context, host string, role instancectl.Role) error
+ Sweep(ctx context.Context) (instancectl.SweepResponse, error)
+ Ready(ctx context.Context) error
+}
+
+var _ Control = (*instancectl.Client)(nil)
+
+// UnitState answers `systemctl show -p ActiveState,InvocationID <unit>`,
+// which needs no privilege. An unloaded unit answers inactive with an
+// empty invocation.
+type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)
+
+func systemctlUnitState(ctx context.Context, unit string) (string, string, error) {
+ out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "ActiveState,InvocationID", unit).Output()
+ if err != nil {
+ return "", "", err
+ }
+ var active, inv string
+ for _, line := range strings.Split(string(out), "\n") {
+ if k, v, ok := strings.Cut(line, "="); ok {
+ switch k {
+ case "ActiveState":
+ active = strings.TrimSpace(v)
+ case "InvocationID":
+ inv = strings.TrimSpace(v)
+ }
+ }
+ }
+ return active, inv, nil
+}
+
+// CgroupEvents reads the unit's cgroup.events under the host slice. cgroup
+// v2 files are world-readable, so the agent reads them as carlos, as the
+// sampler does. An absent directory is the unit's cgroup already removed.
+type CgroupEvents func(slice, unit string) (exists, populated bool, err error)
+
+func cgroupEventsUnder(root string) CgroupEvents {
+ return func(slice, unit string) (bool, bool, error) {
+ b, err := os.ReadFile(filepath.Join(root, "carlos.slice", "carlos-instances.slice", slice, unit, "cgroup.events"))
+ if errors.Is(err, os.ErrNotExist) {
+ return false, false, nil
+ }
+ if err != nil {
+ return true, true, err
+ }
+ return true, !strings.Contains(string(b), "populated 0"), nil
+ }
+}
+```
+
+`internal/activator/loglistener.go`:
+
+```go
+package activator
+
+import (
+ "bufio"
+ "io"
+ "net"
+ "os"
+ "sync"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/edge"
+)
+
+// logListener is the agent's end of a unit's stdout (design §6.4): the
+// unit's StandardOutput=file: connects to this AF_UNIX socket once, at
+// exec, and every line is teed to the agent's own stdout (journald) and to
+// the app log sink, exactly as teeStd does for an exec child. One
+// connection per invocation is accepted; a second is closed unread. EOF
+// ends log collection and nothing else: it is never a sign the process is
+// gone (codex R1-4).
+type logListener struct {
+ ln net.Listener
+ emit func(string)
+ accepted atomic.Bool
+ closed atomic.Bool
+ wg sync.WaitGroup
+}
+
+func listenLogs(path string, emit func(string)) (*logListener, error) {
+ ln, err := edge.SecureUnixListen(path)
+ if err != nil {
+ return nil, err
+ }
+ l := &logListener{ln: ln, emit: emit}
+ l.wg.Add(1)
+ go l.accept()
+ return l, nil
+}
+
+func (l *logListener) accept() {
+ defer l.wg.Done()
+ for {
+ c, err := l.ln.Accept()
+ if err != nil {
+ return
+ }
+ if l.accepted.Swap(true) {
+ c.Close()
+ continue
+ }
+ go l.tee(c)
+ }
+}
+
+func (l *logListener) tee(c net.Conn) {
+ defer c.Close()
+ var sink io.Writer = os.Stdout
+ if l.emit != nil {
+ sink = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, l.emit))
+ }
+ r := bufio.NewReaderSize(c, execLogLineMax)
+ _, _ = io.Copy(sink, r)
+}
+
+// Exited always reports false: a log listener knows nothing about process
+// state, and this method exists so a caller cannot mistake it for one.
+func (l *logListener) Exited() bool { return false }
+
+func (l *logListener) Close() error {
+ if l.closed.Swap(true) {
+ return nil
+ }
+ err := l.ln.Close()
+ l.wg.Wait()
+ return err
+}
+```
+
+`edge.SecureUnixListen` does not exist exported; `secureUnixListen(socketPath string) (net.Listener, error)` is unexported in `internal/edge/purge_listen_unix.go` with a stub in `purge_listen_windows.go`. Export both as `SecureUnixListen` (rename, update its callers in `internal/edge`) in this task; it is the one owner-only unix listen the repo has and the log socket wants exactly its umask discipline. `teeStd` feeds `b.Logs.App(release.RouteAccount(rt), rt.App, rt.Host, line)`; `logEmitter` in Task 4 uses the same call.
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator ./internal/edge -count=1` (outside the sandbox)
+Expected: PASS.
+
+- [ ] **Step 5: gofmt and commit**
+
+```bash
+gofmt -l internal/activator internal/edge
+git add internal/activator internal/edge
+git commit -m "activator: deduplicated child env, the Control seam, the per-role log listener (plan 3, task 3)"
+```
+
+---
+
+### Task 4: `prepare`, the launch file, `unitProc` and the exit observer
+
+**Files:**
+- Create: `internal/activator/prepare.go`, `internal/activator/unitproc.go`
+- Create: `internal/activator/unitproc_test.go`
+- Modify: `internal/activator/backend_exec.go` (`ExecBackend` gains `Control`, `UnitState`, `CgroupEvents`, `CgroupRoot`; `Start`/`Seed`/`StartSidecar`/`HasSidecar` dispatch)
+
+**Interfaces:**
+- Produces:
+ ```go
+ var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
+ type unitProc struct { host string; role instancectl.Role; unit, slice, invocation string; ctl *edge.ControlListener; logs *logListener; exited atomic.Bool; done chan struct{}; stopRequested atomic.Bool }
+ func (p *unitProc) Stop(ctx context.Context) error
+ func (p *unitProc) Exited() bool
+ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role, extraEnv map[string]string) (paths instancePaths, ctl *edge.ControlListener, logs *logListener, err error)
+ func (b *ExecBackend) writeLaunch(paths instancePaths, role instancectl.Role, host string, argv, env []string) error
+ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error)
+ func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error
+ func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error)
+ func (b *ExecBackend) observe(p *unitProc, rt registry.Route) // the goroutine of §6.5
+ ```
+- `ExecBackend` new fields: `Control Control` (nil: nested wakes refuse with "instance control channel not configured"), `UnitState UnitState` (nil: `systemctlUnitState`), `CgroupEvents CgroupEvents` (nil: `cgroupEventsUnder("/sys/fs/cgroup")`), `NestedOK func() error` (nil: ok; production: the mounts and `/ready` check, Task 6), `ObservePoll time.Duration` (0: 1s).
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/activator/unitproc_test.go`:
+
+```go
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// fakeControl records verbs and lets a test drive the unit's state.
+type fakeControl struct {
+ mu sync.Mutex
+ calls []string
+ inv string
+ seed instancectl.SeedResponse
+ startErr error
+ stopHook func(host string, role instancectl.Role)
+ readyErr error
+}
+
+func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() }
+func (f *fakeControl) Start(_ context.Context, host string, role instancectl.Role, max int64) (string, error) {
+ f.rec("start " + host + " " + string(role))
+ return f.inv, f.startErr
+}
+func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancectl.SeedResponse, error) {
+ f.rec("seed " + host)
+ return f.seed, nil
+}
+func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error {
+ f.rec("stop " + host + " " + string(role))
+ if f.stopHook != nil {
+ f.stopHook(host, role)
+ }
+ return nil
+}
+func (f *fakeControl) Reset(_ context.Context, host string, role instancectl.Role) error { f.rec("reset"); return nil }
+func (f *fakeControl) Sweep(context.Context) (instancectl.SweepResponse, error) { f.rec("sweep"); return instancectl.SweepResponse{}, nil }
+func (f *fakeControl) Ready(context.Context) error { f.rec("ready"); return f.readyErr }
+func (f *fakeControl) joined() string { f.mu.Lock(); defer f.mu.Unlock(); return strings.Join(f.calls, "\n") }
+
+// fakeUnit is the unit's observable state: what `systemctl show` and
+// cgroup.events would say.
+type fakeUnit struct {
+ mu sync.Mutex
+ active string
+ inv string
+ exists bool
+ populated bool
+}
+
+func (u *fakeUnit) set(active, inv string, exists, populated bool) {
+ u.mu.Lock()
+ u.active, u.inv, u.exists, u.populated = active, inv, exists, populated
+ u.mu.Unlock()
+}
+func (u *fakeUnit) state(context.Context, string) (string, string, error) {
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ return u.active, u.inv, nil
+}
+func (u *fakeUnit) events(string, string) (bool, bool, error) {
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ return u.exists, u.populated, nil
+}
+
+// nestedBackend returns an ExecBackend wired for a nested host with fakes,
+// and the route. The data and run dirs are temp dirs standing in for
+// /data/instances/<iid> and /run/carlos/instances/<iid>; provision's
+// helpers are not used for the temp paths, so the route is built to
+// classify as nested by construction through the Layout seam.
+func nestedBackend(t *testing.T, ctl *fakeControl, unit *fakeUnit) (*ExecBackend, registry.Route) {
+ t.Helper()
+ root, err := os.MkdirTemp("", "nb")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(root) })
+ host := "h.example.com"
+ b := testExecBackend(t)
+ b.DataDir = filepath.Join(root, "data")
+ b.Control = ctl
+ b.UnitState = unit.state
+ b.CgroupEvents = unit.events
+ b.ObservePoll = 20 * time.Millisecond
+ b.RunDirOverride = filepath.Join(root, "run")
+ stageLive(t, b, host, "v1")
+ rt := registry.Route{Host: host, Kind: "instance", Hibernate: true,
+ DBPath: provision.NestedDBPath(b.DataDir, host), Socket: filepath.Join(b.RunDirOverride, "instances", host, "app.sock")}
+ return b, rt
+}
+```
+
+`RunDirOverride` is a test-only field: `pathsFor` uses it as the socket dir when non-empty (so a test can point `/run/carlos` at a temp dir); `Classify` compares `rt.Socket` against `provision.NestedSocket(host)` and would call this row foreign, so `pathsFor` passes a `LayoutInputs` whose `SocketDir` overrides `provision.SocketDir`: add `SocketDir string` to `LayoutInputs` (empty = `provision.SocketDir`) and use it in `Classify`'s two socket comparisons and in `pathsFor`. Update Task 2's `Classify` accordingly (flat: `filepath.Join(sockDir, host+".sock")`; nested: `filepath.Join(sockDir, "instances", provision.InstanceID(host), "app.sock")`); the Task 2 tests keep passing with an empty `SocketDir`.
+
+```go
+func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ // The fake control does not start anything, so health must be faked:
+ // bind a listener at rt.Socket that answers 200.
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, ver, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ _ = ver
+ up, ok := p.(*unitProc)
+ if !ok || up.invocation != "inv1" {
+ t.Fatalf("proc = %T %+v", p, p)
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.run.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l instancectl.Launch
+ if err := json.Unmarshal(raw, &l); err != nil {
+ t.Fatal(err)
+ }
+ if l.Schema != 1 || l.Host != rt.Host || len(l.Argv) != 5 || l.Argv[0] != filepath.Join(b.OptDir, "live", rt.Host) || l.Argv[2] != rt.Socket || l.Argv[4] != rt.DBPath {
+ t.Errorf("launch = %+v", l)
+ }
+ seenCtl := false
+ for _, kv := range l.Env {
+ if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") {
+ seenCtl = true
+ }
+ if strings.HasPrefix(kv, "CARLOS_ADMIN_TOKEN=") && len(kv) < 30 {
+ t.Errorf("token missing from env: %s", kv)
+ }
+ }
+ if !seenCtl {
+ t.Error("CARLOS_CONTROL_SOCKET not in the launch env")
+ }
+ if !strings.Contains(ctl.joined(), "start h.example.com run") {
+ t.Errorf("control calls: %s", ctl.joined())
+ }
+ if p.Exited() {
+ t.Error("exited before the unit went inactive")
+ }
+ // Termination: terminal state plus empty cgroup.
+ unit.set("inactive", "inv1", false, false)
+ waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if err := p.Stop(ctx); err != nil {
+ t.Errorf("Stop after exit: %v", err)
+ }
+}
+
+func TestUnitProcDoesNotExitOnTerminalStateWhileTheCgroupIsPopulated(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ unit.set("failed", "inv1", true, true) // a process survived SIGKILL
+ time.Sleep(150 * time.Millisecond)
+ if p.Exited() {
+ t.Fatal("exited while the cgroup is still populated (codex R2-5)")
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+ defer cancel()
+ if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) {
+ t.Fatalf("Stop = %v, want ErrStillRunning", err)
+ }
+ unit.set("failed", "inv1", false, false)
+ waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
+}
+
+func TestUnitProcStopsAForeignInvocation(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctl.stopHook = func(string, instancectl.Role) { unit.set("inactive", "", false, false) }
+ unit.set("active", "inv2", true, true) // someone restarted it outside the activator
+ waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
+ if !strings.Contains(ctl.joined(), "stop h.example.com run") {
+ t.Errorf("foreign invocation was not stopped: %s", ctl.joined())
+ }
+}
+
+func TestStartUnitRefusesWhenTheBoxIsNotReadyOrFenced(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ b.NestedOK = func() error { return errors.New("probe: pid1 fail") }
+ if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "pid1") {
+ t.Fatalf("not-ready wake = %v", err)
+ }
+ if strings.Contains(ctl.joined(), "start") {
+ t.Error("a start was issued on a box that is not ready")
+ }
+ b.NestedOK = nil
+ b.Layout.Journal = func(string) bool { return true }
+ if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "migration in flight") {
+ t.Fatalf("fenced wake = %v", err)
+ }
+}
+
+func TestSeedUnitCleansUpOnlyItsOwnDBOnFailure(t *testing.T) {
+ ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ rt.Seed = `["seed","--fast"]`
+ unit.set("inactive", "", false, false)
+ err := b.Seed(context.Background(), rt)
+ if err == nil || !strings.Contains(err.Error(), "exit 3") {
+ t.Fatalf("Seed = %v", err)
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.seed.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l instancectl.Launch
+ _ = json.Unmarshal(raw, &l)
+ if len(l.Argv) != 5 || l.Argv[1] != "seed" || l.Argv[2] != "--fast" || l.Argv[3] != "--db" || l.Argv[4] != rt.DBPath {
+ t.Errorf("seed argv = %v", l.Argv)
+ }
+ for _, kv := range l.Env {
+ if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") || strings.HasPrefix(kv, "CARLOS_PURGE=") {
+ t.Errorf("seed env carries %s", kv)
+ }
+ }
+ if _, err := os.Stat(rt.DBPath); !os.IsNotExist(err) {
+ t.Errorf("failed seed left its database: %v", err)
+ }
+}
+```
+
+Add helpers to the same file:
+
+```go
+// serveOK binds a unix socket at path and answers 200 to every request, so
+// waitHealthyCtx passes without a real unit.
+func serveOK(t *testing.T, path string) func() {
+ t.Helper()
+ if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
+ t.Fatal(err)
+ }
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })}
+ go srv.Serve(ln) //nolint:errcheck
+ return func() { srv.Close(); ln.Close() }
+}
+
+func waitFor(t *testing.T, cond func() bool, d time.Duration) {
+ t.Helper()
+ deadline := time.Now().Add(d)
+ for !cond() {
+ if time.Now().After(deadline) {
+ t.Fatal("condition not met in time")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+}
+```
+
+(imports `net`, `net/http` in the test file.)
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestStartUnit|TestUnitProc|TestSeedUnit' -count=1`
+Expected: build failure.
+
+- [ ] **Step 3: Write `prepare.go`**
+
+```go
+package activator
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// prepare is everything a nested role needs before its unit starts
+// (design §6.1): the two directories, the admin token and merged env (for
+// run and seed), the control listener (run only), the role's log listener,
+// and stale records cleared. It returns the listeners so the caller can
+// hand them to the proc or close them on a failed start.
+func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) {
+ p := b.pathsFor(rt)
+ if err := os.MkdirAll(p.dataDir, 0o700); err != nil {
+ return p, nil, nil, nil, err
+ }
+ if err := os.MkdirAll(p.runDir, 0o750|os.ModeSetgid); err != nil {
+ return p, nil, nil, nil, err
+ }
+ if err := os.MkdirAll(filepath.Dir(b.logSocket(rt.Host, role)), 0o700); err != nil {
+ return p, nil, nil, nil, err
+ }
+ var env map[string]string
+ if role == instancectl.RoleRun || role == instancectl.RoleSeed {
+ if err := EnsureAdminToken(p.envFile); err != nil {
+ return p, nil, nil, nil, fmt.Errorf("admin token for %s: %w", rt.Host, err)
+ }
+ var err error
+ env, err = b.instanceEnv(rt)
+ if err != nil {
+ return p, nil, nil, nil, err
+ }
+ } else {
+ var err error
+ env, err = readEnvFileBeneath(p.dataDir, filepath.Base(p.sidecarEnv))
+ if err != nil {
+ return p, nil, nil, nil, err
+ }
+ }
+ // Stale exit records from earlier invocations of this role.
+ if entries, err := os.ReadDir(p.runDir); err == nil {
+ for _, e := range entries {
+ if strings.HasPrefix(e.Name(), "exit."+string(role)+".") {
+ _ = os.Remove(filepath.Join(p.runDir, e.Name()))
+ }
+ }
+ }
+ var ctl *edge.ControlListener
+ if role == instancectl.RoleRun {
+ _ = os.Remove(rt.Socket) // nothing is running; a stale socket file would refuse the app's bind
+ ctl = b.listenControl(rt, p.envFile)
+ }
+ logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
+ if err != nil {
+ if ctl != nil {
+ ctl.Close()
+ }
+ return p, nil, nil, nil, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err)
+ }
+ return p, ctl, logs, env, nil
+}
+
+// listenControl binds the control socket exactly as Start does for an exec
+// child (the tokenReader and schedule mount are unchanged); a bind failure
+// is logged and the instance runs without its control verbs, as today.
+func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.ControlListener {
+ tokenReader := func() (string, error) {
+ env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(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(rt.Socket+".control", rt.Host, b.Cache, mount)
+ if err != nil {
+ log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
+ return nil
+ }
+ return ctl
+}
+
+// logSocket is the role's log socket path, honouring the test override of
+// the run directory.
+func (b *ExecBackend) logSocket(host string, role instancectl.Role) string {
+ if b.RunDirOverride != "" {
+ return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+string(role)+".sock")
+ }
+ return provision.LogSocket(host, string(role))
+}
+
+// logEmitter is the same sink teeStd feeds, or nil when Logs is unset.
+func (b *ExecBackend) logEmitter(rt registry.Route) func(string) {
+ if b.Logs == nil {
+ return nil
+ }
+ return func(line string) { b.Logs.App(release.RouteAccount(rt), rt.App, rt.Host, line) }
+}
+
+// writeLaunch writes launch.<role>.json atomically into the instance
+// directory (design §6.2). The env is childEnv's deduplicated list; the
+// trampoline execs it as is.
+func (b *ExecBackend) writeLaunch(p instancePaths, role instancectl.Role, host string, argv, env []string) error {
+ raw, err := json.MarshalIndent(instancectl.Launch{Schema: instancectl.LaunchSchema, Host: host, Argv: argv, Env: env}, "", " ")
+ if err != nil {
+ return err
+ }
+ return instancefs.WriteAtomic(p.dataDir, instancectl.LaunchFileName(role), append(raw, '\n'), 0o600)
+}
+
+// readEnvFileBeneath is readEnvFile through instancefs: a nested host's env
+// files are read without following a symlink (design §3.2). A missing file
+// is an empty env, as readEnvFile treats it.
+func readEnvFileBeneath(dir, name string) (map[string]string, error) {
+ b, err := instancefs.ReadBeneathLimit(dir, name, 1<<20)
+ if errors.Is(err, os.ErrNotExist) {
+ return map[string]string{}, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return parseEnvBytes(b), nil
+}
+```
+
+Check how `b.Logs` is called in `teeStd` (`b.Logs.App(...)` or via `emitEvent`) and mirror it exactly; the sweep says `Logs EventSink` with `Event(account, app, host, level, msg)` for events and `teeStd` feeding `Logs.App(acct, app, rt.Host, line)`. Read `teeStd` and use the same call. `readEnvFile`/`parseEnvBytes` exist in `backend_exec.go`/`admintoken.go`; for a flat host `instanceEnv` keeps calling `readEnvFile`; for a nested host it calls `readEnvFileBeneath` (dispatch inside `instanceEnv` on `p.layout`).
+
+- [ ] **Step 4: Write `unitproc.go`**
+
+```go
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "path/filepath"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// ErrStillRunning is Stop's answer when ctx expired before termination was
+// proven (design §6.3). The activator then keeps the lease and the entry
+// in stopping, and the sweeper retries; the lease is never released over a
+// writer that may still exist.
+var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
+
+// unitProc is a nested host's Proc (design §6.3): the unit the control
+// service started, the listeners the agent holds for it, and the observer
+// that proves termination.
+type unitProc struct {
+ b *ExecBackend
+ rt registry.Route
+ role instancectl.Role
+ unit, slice string
+ invocation string
+ runDir string
+ ctl *edge.ControlListener
+ logs *logListener
+ exited atomic.Bool
+ done chan struct{}
+ closeOnce atomic.Bool
+}
+
+func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc {
+ up := &unitProc{b: b, rt: rt, role: role, unit: instancectl.UnitName(role, rt.Host), slice: instancectl.SliceName(rt.Host),
+ invocation: inv, runDir: p.runDir, ctl: ctl, logs: logs, done: make(chan struct{})}
+ go b.observe(up)
+ return up
+}
+
+func (p *unitProc) Exited() bool { return p.exited.Load() }
+
+// Stop asks the control service to stop the unit and waits for termination
+// to be proven. The control listener and the log listener are closed only
+// after that, preserving childProc's ordering (signal first, close the
+// listener after the process is gone).
+func (p *unitProc) Stop(ctx context.Context) error {
+ if !p.Exited() {
+ sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl)
+ err := p.b.Control.Stop(sctx, p.rt.Host, p.role)
+ cancel()
+ if err != nil {
+ log.Printf("activator: stop %s %s: %v", p.rt.Host, p.role, err)
+ }
+ }
+ select {
+ case <-p.done:
+ p.closeListeners()
+ return nil
+ case <-ctx.Done():
+ return ErrStillRunning
+ }
+}
+
+func (p *unitProc) closeListeners() {
+ if p.closeOnce.Swap(true) {
+ return
+ }
+ if p.ctl != nil {
+ p.ctl.Close()
+ }
+ if p.logs != nil {
+ p.logs.Close()
+ }
+}
+
+// observe is the goroutine of design §6.5. It proves termination from two
+// facts the tenant cannot forge: systemd's unit state and cgroup emptiness.
+// The exit record decides the wording of the event line and nothing else.
+func (b *ExecBackend) observe(p *unitProc) {
+ poll := b.ObservePoll
+ if poll <= 0 {
+ poll = time.Second
+ }
+ state := b.UnitState
+ if state == nil {
+ state = systemctlUnitState
+ }
+ events := b.CgroupEvents
+ if events == nil {
+ events = cgroupEventsUnder(b.cgroupRoot())
+ }
+ foreignStopped := false
+ lastRemain := time.Time{}
+ for {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ active, inv, err := state(ctx, p.unit)
+ cancel()
+ if err != nil {
+ log.Printf("activator: %s: unit state: %v", p.rt.Host, err)
+ }
+ terminal := false
+ foreign := false
+ switch {
+ case err != nil:
+ case inv != "" && inv != p.invocation:
+ foreign = true
+ case active == "inactive" || active == "failed":
+ terminal = true
+ }
+ if foreign && !foreignStopped {
+ log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation)
+ sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
+ if err := b.Control.Stop(sctx, p.rt.Host, p.role); err != nil {
+ log.Printf("activator: %s: stop foreign invocation: %v", p.rt.Host, err)
+ }
+ scancel()
+ foreignStopped = true
+ }
+ if terminal || (foreign && foreignStopped) {
+ exists, populated, eerr := events(p.slice, p.unit)
+ if eerr != nil {
+ log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr)
+ }
+ if !exists || !populated {
+ if terminal {
+ emitEvent(b.Logs, p.rt, exitLevel(p), exitLine(p))
+ } else {
+ emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)")
+ }
+ p.exited.Store(true)
+ close(p.done)
+ if p.Exited() {
+ // Stop closes the listeners when it observes done; a
+ // crash nobody is stopping yet leaves them for the
+ // sweeper's stand-down, which calls Stop.
+ }
+ return
+ }
+ if time.Since(lastRemain) > 2*time.Second {
+ log.Printf("activator: %s: processes remain after %s; not releasing", p.rt.Host, active)
+ lastRemain = time.Now()
+ }
+ }
+ time.Sleep(poll)
+ }
+}
+
+// exitLine reads the advisory record for this invocation and words the
+// event line as childProc does: a non-zero status is a warn line, a clean
+// exit is info, and no record is unknown.
+func exitLine(p *unitProc) string {
+ rec, ok := p.readRecord()
+ switch {
+ case !ok:
+ return "instance exited (status unknown: no exit record)"
+ case rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0"):
+ return "instance exited (exit 0)"
+ case rec.Code == "killed" || rec.Code == "dumped":
+ return "instance exited: signal " + rec.Status
+ default:
+ return fmt.Sprintf("instance exited: exit %s (%s)", rec.Status, rec.Result)
+ }
+}
+
+func exitLevel(p *unitProc) string {
+ if rec, ok := p.readRecord(); ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) {
+ return "info"
+ }
+ return "warn"
+}
+
+func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) {
+ raw, err := instancefs.ReadBeneathLimit(p.runDir, instancectl.ExitRecordName(p.role, p.invocation), 4096)
+ if err != nil {
+ return instancectl.ExitRecord{}, false
+ }
+ var rec instancectl.ExitRecord
+ if err := json.Unmarshal(raw, &rec); err != nil {
+ return instancectl.ExitRecord{}, false
+ }
+ return rec, true
+}
+
+func (b *ExecBackend) cgroupRoot() string {
+ if b.CgroupRoot != "" {
+ return b.CgroupRoot
+ }
+ return "/sys/fs/cgroup"
+}
+```
+
+Remove the empty `if p.Exited() { ... }` block in `observe` before committing; it is shown only to carry its comment, which belongs above `close(p.done)`. Match `emitEvent`'s actual signature (`emitEvent(sink EventSink, rt registry.Route, level, msg string)`).
+
+- [ ] **Step 5: The dispatch in `backend_exec.go`**
+
+Rename the current `Start` body to `startChild`, `Seed`'s to `seedChild`, `StartSidecar`'s to `startSidecarChild` (signatures unchanged), and write the dispatchers:
+
+```go
+func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ p := b.pathsFor(rt)
+ switch p.layout {
+ case LayoutNested:
+ return b.startUnit(ctx, rt)
+ case LayoutFenced, LayoutConflict:
+ return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
+ }
+ return b.startChild(ctx, rt)
+}
+```
+
+(and the same shape for `Seed` and `StartSidecar`; `HasSidecar` stats `pathsFor(rt).sidecarEnv`.)
+
+`startUnit`:
+
+```go
+func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ if b.Control == nil {
+ return nil, "", fmt.Errorf("%s: instance control channel not configured", rt.Host)
+ }
+ if b.NestedOK != nil {
+ if err := b.NestedOK(); err != nil {
+ return nil, "", fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err)
+ }
+ }
+ if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
+ return nil, "", err
+ }
+ p, ctl, logs, fileEnv, err := b.prepare(rt, instancectl.RoleRun)
+ if err != nil {
+ return nil, "", err
+ }
+ fail := func(err error) (Proc, string, error) {
+ if ctl != nil {
+ ctl.Close()
+ }
+ logs.Close()
+ return nil, "", err
+ }
+ childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil) // the same withVar/withoutVars block Start uses today, extracted so both paths share it
+ argv := []string{b.liveLink(rt.Host), "--socket", rt.Socket, "--db", p.dbPath}
+ if err := b.writeLaunch(p, instancectl.RoleRun, rt.Host, argv, childEnv(childVars)); err != nil {
+ return fail(err)
+ }
+ spawnVer, _ := b.LiveVersion(rt.Host)
+ inv, err := b.Control.Start(ctx, rt.Host, instancectl.RoleRun, b.tier(rt).MemoryMax)
+ if err != nil {
+ return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
+ }
+ log.Printf("activator: %s: started %s memory.max=%d", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host), b.tier(rt).MemoryMax)
+ up := b.newUnitProc(rt, instancectl.RoleRun, p, inv, ctl, logs)
+ if b.Cache != nil {
+ if n := b.Cache.PurgeHost(rt.Host); n > 0 {
+ log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
+ }
+ }
+ if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = up.Stop(stopCtx)
+ return nil, "", err
+ }
+ b.normalizeSocketMode(rt, up) // the existing 0660 block, extracted
+ return up, spawnVer, nil
+}
+```
+
+Extract the `childVars` composition (the three `withVar`/`withoutVars` pairs) into `agentOwnedVars(rt, fileEnv, haveControl bool) map[string]string` and the post-health `0660` block into `normalizeSocketMode(rt, proc)`, and call both from `startChild` too, so the two paths cannot drift. `seedUnit`: prepare for `RoleSeed`; argv = the same `seedArgs` parse plus `--db p.dbPath` that `seedChild` builds (extract the parse into `seedArgv(rt, dbPath) ([]string, error)`); env `childEnv(withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE"))`; `seedOwnsDB` stat before as today; `res, err := b.Control.Seed(ctx, rt.Host, b.tier(rt).MemoryMax)`; then, to honour §6.2, wait for the seed unit's cgroup to be empty through `CgroupEvents` (up to `actStopTimeout`, polling `ObservePoll`) before deciding; a non-success `res.Result` returns `fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result)` after the same cleanup `seedChild` does when `seedOwnsDB`; close the seed's log listener afterwards. `startSidecarUnit`: prepare for `RoleSidecar` (env from the sidecar file only); argv `[liveLink, "sidecar", "run"]`; `Control.Start(..., RoleSidecar, ...)`; return `newUnitProc` with `ctl == nil`.
+
+- [ ] **Step 6: Run the tests to verify they pass**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -count=1` (outside the sandbox)
+Expected: PASS, including every pre-existing test unchanged.
+
+- [ ] **Step 7: gofmt and commit**
+
+```bash
+gofmt -l internal/activator
+git add internal/activator
+git commit -m "activator: the nested spawn path: prepare, launch file, unitProc and the exit observer (plan 3, task 4)"
+```
+
+---
+
+### Task 5: The three `activator.go` changes
+
+**Files:**
+- Modify: `internal/activator/activator.go` (`stop`, `sweepOnce`, `doWake`, `Service` gains `Routes`)
+- Test: `internal/activator/activator_test.go` (append three tests)
+
+**Interfaces:**
+- Produces:
+ ```go
+ // RouteLookup is the registry read doWake performs before Acquire (design §6.8); *registry.Registry satisfies it with Get.
+ type RouteLookup interface { Get(host string) (*registry.Route, error) }
+ // Service gains: Routes RouteLookup (nil: the wake runs on the copy the edge supplied, today's behaviour)
+ ```
+ `actEntry` gains `pendingStop bool`, `sidecarStarting bool`, `wakeGen uint64`.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/activator/activator_test.go`. The existing helpers are: `testActivator(t, idle) (*Service, *fakeBackend, *fakeClock, *registry.Registry)`, `testRoute(host) registry.Route`, `newTestServiceAwake(t, host)`, `fakeBackend` (fields `procs []*fakeProc`, `sidecars []*fakeProc`, `sidecarStaged bool`, `startGate chan struct{}`, `releases` counter via `releaseCount()`, `lastProc()`, `lastSidecar()`), `fakeProc`, `fakeClock`. A wake is driven the way the file's other tests do it: `a.beginRequest(ctx, rt)` then `release`; a stop by advancing the fake clock past `idle` and calling `a.sweepOnce()`, or `a.Bounce(host)`. Read the two nearest tests (an idle-hibernate test and a sidecar-restart test) and copy their driving code into the three tests below where a comment says `// drive`.
+
+```go
+// A Proc whose Stop reports ErrStillRunning must not have its lease
+// released: stop() keeps the entry in stopping and the sweeper finishes it
+// once the proc reports exited (design §6.3, codex R2-5).
+func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Minute)
+ b.procStuck = true // Start returns a fakeProc whose Stop reports ErrStillRunning until exit()
+ rt := testRoute("h.example.com")
+ // drive: wake rt and release the request, as the idle-hibernate test does
+ clk.advance(2 * time.Minute)
+ a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning
+ p := b.lastProc()
+ waitFor(t, func() bool { return p.stopCalls() >= 1 }, 2*time.Second)
+ if n := b.releaseCount(); n != 0 {
+ t.Fatalf("lease released %d times while the stop is unproven", n)
+ }
+ p.exit()
+ a.sweepOnce()
+ waitFor(t, func() bool { return b.releaseCount() == 1 }, 2*time.Second)
+}
+
+// The sidecar restart must not hold the activator mutex across the start
+// call (design §6.7, codex R2-6): a stalled sidecar start does not block
+// another host's wake.
+func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
+ a, b, clk, _ := testActivator(t, time.Hour)
+ b.sidecarStaged = true
+ release := make(chan struct{})
+ b.sidecarGate = release // StartSidecar blocks on this channel when non-nil
+ // drive: wake a.example.com with the sidecar gate open, then close the gate
+ b.lastSidecar().exit() // the sweeper now wants a restart
+ clk.advance(time.Minute)
+ go a.sweepOnce() // blocks inside StartSidecar until release is closed
+ time.Sleep(50 * time.Millisecond)
+ done := make(chan struct{})
+ go func() {
+ // drive: wake b.example.com
+ close(done)
+ }()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("a second host's wake blocked behind a stalled sidecar start")
+ }
+ close(release)
+}
+
+// doWake runs on the committed row, not the copy the edge supplied
+// (design §6.8, codex R8-1).
+func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ old := testRoute("h.example.com")
+ fresh := old
+ fresh.DBPath = "/data/instances/h.example.com/h.example.com.db"
+ fresh.Socket = "/run/carlos/instances/h.example.com/app.sock"
+ // insert fresh into reg the way the file's registry-backed tests do
+ // drive: wake with old (the edge's stale copy)
+ if got := b.lastStartedRoute().DBPath; got != fresh.DBPath {
+ t.Fatalf("Start saw DBPath %q, want the registry's %q", got, fresh.DBPath)
+ }
+}
+```
+
+Add to `fakeBackend`: `procStuck bool` (when set, `Start` returns a `fakeProc` with `stuck` true), `sidecarGate chan struct{}` (`StartSidecar` blocks on it when non-nil), `lastStartedRoute() registry.Route` (recorded in `Start`); to `fakeProc`: `stuck` handling in `Stop` (return `ErrStillRunning` while stuck and not exited, counting calls in `stopCalls()`), `exit()` if absent. `waitFor` is the helper Task 4 added in `unitproc_test.go` (same package). `fakeClock.advance` may be spelled differently; use the file's name.
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestStopDoesNotRelease|TestSidecarRestartDoesNotHold|TestDoWakeRefreshes' -count=1`
+Expected: FAIL (`Routes` undefined; the lease is released today; the lock is held today).
+
+- [ ] **Step 3: Make the three changes**
+
+In `stop(e *actEntry)`, after `err = e.proc.Stop(procCtx)`:
+
+```go
+ if errors.Is(err, ErrStillRunning) {
+ // The lease is never released over an unproven termination (design
+ // §6.3). Keep the entry in stopping; sweepOnce finishes the stop
+ // once the proc reports exited.
+ log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err)
+ a.mu.Lock()
+ e.pendingStop = true
+ a.mu.Unlock()
+ return
+ }
+```
+
+and in `sweepOnce`, under `a.mu`, before the idle pass:
+
+```go
+ for _, e := range a.entries {
+ if e.pendingStop && e.proc != nil && e.proc.Exited() {
+ e.pendingStop = false
+ go a.finishStop(e) // rep.Close, delete, close(e.stopped): the tail of stop()
+ }
+ }
+```
+
+with `finishStop` being the tail of today's `stop()` from `rep.Close` onward, extracted so `stop()` calls it too.
+
+In `sweepOnce`'s sidecar block, replace the synchronous `a.backend.StartSidecar(...)` with:
+
+```go
+ if !e.sidecarStarting {
+ e.sidecarStarting = true
+ gen := e.wakeGen
+ host := e.host
+ route := e.route
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
+ sc, err := a.backend.StartSidecar(ctx, route)
+ cancel()
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ if e == nil || e.state != actAwake || e.wakeGen != gen {
+ if sc != nil {
+ go sc.Stop(context.Background()) //nolint:errcheck
+ }
+ return
+ }
+ e.sidecarStarting = false
+ // install exactly as the synchronous code did
+ ...
+ }()
+ }
+```
+
+(`e.wakeGen` increments in `doWake` each time an entry becomes awake; the install branch is the existing `if err != nil { ... } else { e.sidecar = sc; ... }` block moved inside the goroutine.) The existing `stop()` waits for `sidecarStarting` to clear or stops the installed sidecar, whichever it finds: add, at the top of `stop()`'s sidecar section, a bounded wait (`sidecarStopTimeout`) on `sidecarStarting` becoming false.
+
+In `doWake`, at the top, before `Acquire`:
+
+```go
+ if a.Routes != nil {
+ fresh, err := a.Routes.Get(rt.Host)
+ if err != nil {
+ return nil, fmt.Errorf("%s: route lookup: %w", rt.Host, err)
+ }
+ if fresh == nil {
+ return nil, fmt.Errorf("%s: no route", rt.Host)
+ }
+ if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
+ log.Printf("activator: %s: route refreshed: db %q → %q socket %q → %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
+ }
+ rt = *fresh
+ a.mu.Lock()
+ if e := a.entries[rt.Host]; e != nil {
+ e.route = rt
+ }
+ a.mu.Unlock()
+ }
+```
+
+(Use a plain `->` in the log line, not an arrow glyph: the copy rule.)
+
+- [ ] **Step 4: Run the whole package**
+
+Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -race -count=1` (outside the sandbox)
+Expected: PASS, race-clean.
+
+- [ ] **Step 5: gofmt and commit**
+
+```bash
+gofmt -l internal/activator
+git add internal/activator
+git commit -m "activator: unproven stops keep the lease, sidecar restarts leave the lock, wakes run on the committed row (plan 3, task 5)"
+```
+
+---
+
+### Task 6: Consumers of the shape and the agent wiring
+
+**Files:**
+- Modify: `internal/appconfig/syncexec.go` (writer path; `reapConfigEnv`)
+- Modify: `internal/schedule/runner.go`, `internal/host/schedule_wiring.go` (token path)
+- Modify: `internal/host/movespass.go` (DB path from the row; fence)
+- Modify: `internal/ops/litestream.go` (glob)
+- Modify: `internal/host/agent.go` (wiring, startup sweep, gate)
+- Tests in each package
+
+**Interfaces:**
+- Consumes: `activator.Classify`, `activator.LayoutInputs`, `activator.ActivatorManaged`, `provision.InstanceDir`, `provision.LayoutJournalPath`, `instancectl.NewClient`, `controlsock.Path`, `instancefs.MountNoSymfollow`.
+- Produces: `appconfig.ConfigEnvPath(dataDir string, rt registry.Route) string` (nested for a nested row, flat otherwise); `schedule.TokenFromEnvFileFor(dataDir string, paths func(host string) string)` or the equivalent shape the runner already has (read it: the runner's `Token` seam takes a host; give it a `func(host) string` that resolves the env file through the same classification the backend uses, built in `schedule_wiring.go` from the registry).
+
+- [ ] **Step 1: Write the failing tests**
+
+`internal/appconfig/syncexec_test.go` (append):
+
+```go
+func TestSyncExecWritesANestedRouteIntoItsInstanceDir(t *testing.T) {
+ // Build the fixture exactly as TestSyncExecWritesConfigFileAndBouncesOnChange
+ // does, but with the route's DBPath and Socket at the nested defaults,
+ // and assert the file lands at <dataDir>/instances/<host>/<host>.config.env
+ // and NOT at <dataDir>/<host>.config.env.
+}
+
+func TestReapConfigEnvWalksInstanceDirs(t *testing.T) {
+ // A stale <dataDir>/instances/gone.example.com/gone.example.com.config.env
+ // with no route is removed; a live nested route's file is kept.
+}
+```
+
+Write both bodies by copying the closest existing test in that file and changing the route's two columns and the expected path.
+
+`internal/ops/ops_test.go` (append):
+
+```go
+func TestFleetDBsDiscoversNestedDatabases(t *testing.T) {
+ // A registry with a non-hibernating exec route whose DBPath is
+ // <dataDir>/instances/<host>/<host>.db and the file present: FleetDBs
+ // returns it. Copy the shape of the existing flat discovery test.
+}
+```
+
+`internal/host/movespass_test.go` (append): a cutting move for a host whose journal exists at `provision.LayoutJournalPath(cfg.DataDir, host)` does nothing and logs the fence; a cutting move for a nested row checkpoints `rt.DBPath`, not `<DataDir>/<host>.db`.
+
+`internal/host` (new test in `agent_isolated_test.go`): `nestedGate(mounts func(string) (bool, error), ready func(ctx) error) func() error` returns an error naming the first mount without `nosymfollow`, then the `/ready` error, else nil.
+
+- [ ] **Step 2: Run the new tests to verify they fail**
+
+Run each package's new tests; expected: FAIL.
+
+- [ ] **Step 3: Make the changes**
+
+`syncexec.go`: replace `path := filepath.Join(dataDir, rt.Host+configEnvSuffix)` (the writer and the addr warning path) with `path := ConfigEnvPath(dataDir, rt)`:
+
+```go
+// ConfigEnvPath is where a route's rendered bundle lands: inside the
+// instance directory for a nested row (isolated-exec design §7), flat for
+// everything else, including opted-in addr routes whose supervisor reads
+// the flat path.
+func ConfigEnvPath(dataDir string, rt registry.Route) string {
+ if rt.DBPath == provision.NestedDBPath(dataDir, rt.Host) && rt.Socket == provision.NestedSocket(rt.Host) {
+ return filepath.Join(provision.InstanceDir(dataDir, rt.Host), rt.Host+configEnvSuffix)
+ }
+ return filepath.Join(dataDir, rt.Host+configEnvSuffix)
+}
+```
+
+The writer's `os.WriteFile(tmp, ...)` + `os.Rename` becomes `instancefs.WriteAtomic(filepath.Dir(path), filepath.Base(path), rendered, 0o600)` for a nested path (the `O_EXCL` temp of §3.2) and stays as is for a flat one. `reapConfigEnv` additionally walks `filepath.Join(dataDir, "instances")` one level down, applying the same "host has no live route" rule to each `*.config.env` it finds there (the host is the file name minus the suffix); the flat directory's `x.config` ambiguity guard stays for the flat directory only.
+
+`schedule/runner.go`: add `TokenFromEnvPaths(envPath func(host string) string) func(host string) (string, error)` holding today's body of `TokenFromEnvFile` with the path taken from `envPath(host)`; `TokenFromEnvFile(dataDir)` becomes `TokenFromEnvPaths(func(host string) string { return filepath.Join(dataDir, host+".env") })`. `schedule_wiring.go`: `scheduleToken(dd, svc)` gains a `reg *registry.Registry` parameter and an `envPathFor(dd, reg) func(host string) string` that reads `reg.Get(host)`, classifies it with `activator.Classify(rt, activator.LayoutInputs{DataDir: dd})` and returns `filepath.Join(provision.InstanceDir(dd, host), host+".env")` for `LayoutNested`, else the flat path; both `schedule.TokenFromEnvPaths(envPathFor(...))` and the `EnsureAdminToken` prepare step use it. (A fenced host's token read keeps working: fenced is a refusal to wake, not to read.)
+
+`movespass.go`: `moveCfg` gains `Route func(host string) (*registry.Route, error)` (wired in `armMovesPass` as `reg.Get`) and the checkpoint block becomes:
+
+```go
+ if _, err := os.Stat(provision.LayoutJournalPath(cfg.DataDir, rec.Host)); err == nil {
+ log.Printf("moves: %s: layout migration in flight; leaving the move for the next pass", rec.Host)
+ continue // or return nil, matching how the loop skips a host today
+ }
+ dbPath := filepath.Join(cfg.DataDir, rec.Host+".db")
+ if cfg.Route != nil {
+ if rt, err := cfg.Route(rec.Host); err == nil && rt != nil && rt.DBPath != "" {
+ dbPath = rt.DBPath
+ }
+ }
+```
+
+`litestream.go:354`: `matches, err := filepath.Glob(filepath.Join(dataDir, "*.db"))` gains a second glob `filepath.Join(dataDir, "instances", "*", "*.db")` appended to `matches`.
+
+`agent.go`: after `eb := &activator.ExecBackend{...}`:
+
+```go
+ eb.Control = instancectl.NewClient(controlsock.Path)
+ eb.Layout = activator.LayoutInputs{
+ Journal: func(host string) bool { _, err := os.Stat(provision.LayoutJournalPath(dd, host)); return err == nil },
+ Peers: reg.List,
+ }
+ eb.NestedOK = nestedGate(instancefs.MountNoSymfollow, eb.Control.Ready)
+ act.Routes = reg
+```
+
+and before the activator's `Start()`:
+
+```go
+ if sw, err := eb.Control.Sweep(ctx); err != nil {
+ log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
+ } else if len(sw.Units) > 0 {
+ log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
+ }
+```
+
+`nestedGate`:
+
+```go
+// nestedGate is what an isolated wake requires of the box (design §8.5,
+// §8.6): the three nosymfollow mounts, and a control service that reports
+// ready. The answer is recomputed per wake (the mounts are cheap to read;
+// /ready is one short request), so a box that becomes ready needs no
+// restart and a box that stops being ready refuses the next wake.
+func nestedGate(mounts func(string) (bool, error), ready func(context.Context) error) func() error {
+ return func() error {
+ for _, m := range []string{"/data/instances", "/run/carlos/instances", "/run/carlos/logs"} {
+ ok, err := mounts(m)
+ if err != nil {
+ return fmt.Errorf("%s: %v", m, err)
+ }
+ if !ok {
+ return fmt.Errorf("%s is not a nosymfollow mount", m)
+ }
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineReady)
+ defer cancel()
+ return ready(ctx)
+ }
+}
+```
+
+(`reg.List` must match `func() ([]registry.Route, error)`; if the registry's list method has a different name or signature, wrap it.)
+
+- [ ] **Step 4: Full gate**
+
+```bash
+export GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off TMPDIR=/tmp/g
+go build ./... && go vet ./... && gofmt -l . && go test ./... -count=1
+GOOS=darwin go build ./... && GOOS=windows go build ./... && GOARCH=arm64 go build ./...
+```
+
+Expected: green.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/appconfig internal/schedule internal/host internal/ops
+git commit -m "isolated exec: consumers follow the layout; the agent wires the control client, the gate and the startup sweep (plan 3, task 6)"
+```
+
+---
+
+## Self-review
+
+**Spec coverage.** §6.1 prepare and Start order: Task 4. §6.2 launch file, dedupe, seed and sidecar argv: Tasks 3, 4. §6.3 `unitProc.Stop`, `ErrStillRunning`, `stop()` not releasing: Tasks 4, 5. §6.4 per-role log listeners, one connection, EOF is not exit: Task 3. §6.5 termination from unit state and cgroup, foreign invocation, record as wording, stale record removal: Task 4. §6.6 tier via `/start`: Task 4 (sampler roots landed in plan 1's fix wave; the edge key and per-host key are unchanged). §6.7 startup sweep and the sidecar restart: Tasks 5, 6. §6.8 route refresh: Task 5. §7 config delivery: Task 6. §8.1 classification and refusals: Task 2. §8.5/§8.6 the gate and the "channel absent" refusal: Tasks 4, 6. §9 consumers: Task 6 (syncexec, schedule token, movespass, litestream glob); status `layout` field and the console are plan 4.
+
+**Placeholder scan.** Task 6's three test stubs describe the fixture to copy and the assertion; the implementer writes the bodies from the named neighbouring tests. Everything else carries code. The activator.go edits (Task 5) name the exact existing functions and fields; where the current test file's helper names may differ, the task says to read and adapt the new tests, not the code.
+
+**Type consistency.** `Layout`, `LayoutInputs{DataDir, SocketDir, Journal, Peers, SameID}`, `Classify`, `ActivatorManaged`, `instancePaths`, `pathsFor`, `Control`, `UnitState`, `CgroupEvents`, `logListener`/`listenLogs`, `unitProc`, `ErrStillRunning`, `prepare`, `writeLaunch`, `startUnit`, `seedUnit`, `startSidecarUnit`, `observe`, `RouteLookup`, `Routes`, `nestedGate`, `ConfigEnvPath` are spelled identically across tasks. `ExecBackend` new fields: `Layout`, `Control`, `UnitState`, `CgroupEvents`, `CgroupRoot`, `NestedOK`, `ObservePoll`, `RunDirOverride`.