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:

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:

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:

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:

// 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:

// 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:

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

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:

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:

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
	// SocketDir is where flat sockets live and under which the nested
	// instances directory sits. Empty means provision.SocketDir; a test
	// points it at a temp dir so a route can classify as nested without
	// writing to /run/carlos.
	SocketDir 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
}

// sockets returns the flat and nested socket paths for host under the
// configured socket directory.
func (in LayoutInputs) sockets(host string) (flat, nested string) {
	dir := in.SocketDir
	if dir == "" {
		dir = provision.SocketDir
	}
	return filepath.Join(dir, host+".sock"), filepath.Join(dir, "instances", provision.InstanceID(host), "app.sock")
}

// 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)
				}
			}
		}
	}
	flatSock, nestedSock := in.sockets(rt.Host)
	switch {
	case rt.DBPath == provision.DefaultDBPath(in.DataDir, rt.Host) && rt.Socket == flatSock:
		return LayoutFlat, ""
	case rt.DBPath == provision.NestedDBPath(in.DataDir, rt.Host) && rt.Socket == nestedSock:
		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:

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:

	// 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

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:

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:

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:

// 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:

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:

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

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) (paths instancePaths, ctl *edge.ControlListener, logs *logListener, fileEnv map[string]string, 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:

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 on ExecBackend: when non-empty, layoutInputs() sets LayoutInputs.SocketDir to it and pathsFor uses it in place of provision.SocketDir for runDir (nested: <RunDirOverride>/instances/<iid>), so a test can point /run/carlos at a temp dir and the row still classifies as nested through Task 2's sockets helper.

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:

// 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

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

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. The record is tenant-writable
// text (design §6.5, final review of plan 2), so every field that reaches
// the line passes through recordToken first.
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 " + recordToken(rec.Status)
	default:
		return fmt.Sprintf("instance exited: exit %s (%s)", recordToken(rec.Status), recordToken(rec.Result))
	}
}

// recordToken bounds a field of the exit record to what systemd itself
// would write: at most 32 bytes of [A-Za-z0-9_-]. Anything else becomes
// "?" so a tenant cannot put arbitrary text or control characters into
// the platform's event stream.
func recordToken(s string) string {
	if len(s) == 0 || len(s) > 32 {
		return "?"
	}
	for _, c := range s {
		if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-') {
			return "?"
		}
	}
	return s
}

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:

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:

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))
	}
	if inv == "" {
		// The server read the invocation after the start and found none:
		// the unit already exited (plan 2 final review). Nothing is
		// running, so there is nothing to stop; the wake failed.
		return fail(fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host)))
	}
	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

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.

// 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):

	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:

	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:

			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:

	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

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):

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):

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):

// 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:

		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{...}:

	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():

	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:

// 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

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

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.