carlos / platform Public

Download this file

1	# Isolated Exec, Plan 3 of 4: The Activator's Unit Path
2
3	> **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.
4
5	**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.
6
7	**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.
8
9	**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.
10
11	**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.
12
13	## Global Constraints
14
15	- 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.
16	- **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.
17	- 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.
18	- 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.
19	- `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]`.
20	- 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.
21	- `unitProc.Stop` returns `ErrStillRunning` when `ctx` expires before `done`; `stop()` then does not call `rep.Close` (spec §6.3).
22	- Every control call carries a deadline: 30s for start/stop/reset, 150s for seed, 60s for sweep, 5s for ready (spec §5.3).
23	- 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).
24	- 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.
25	- Commit after every task with the session's trailer. Branch `worktree-isolated-exec-spec`.
26
27	---
28
29	## File Structure
30
31	| File | Responsibility |
32	|---|---|
33	| `internal/provision/provision.go` (modify) | `LayoutJournalPath(dataDir, host)`; `InstanceRunDirOf`, `LogSocket` already exist |
34	| `internal/instancefs/instancefs.go` + `_linux.go` + `_other.go` (modify) | `ReadBeneathLimit(dir, name, max)` |
35	| `internal/instancectl/client.go` (create) | `Client`: the agent-side HTTP client over the control socket, one method per verb, with the deadlines |
36	| `internal/instancectl/client_test.go` (create) | client against a `Server` with fakes |
37	| `internal/activator/layout.go` (create) | `Layout`, `LayoutInputs`, `Classify` |
38	| `internal/activator/paths.go` (create) | `instancePaths` and `pathsFor(rt)`: every per-host path by layout |
39	| `internal/activator/control.go` (create) | the `Control` interface the backend calls; `UnitState` and `CgroupEvents` seams |
40	| `internal/activator/loglistener.go` (create) | `logListener` |
41	| `internal/activator/unitproc.go` (create) | `unitProc`, the exit observer, `ErrStillRunning` |
42	| `internal/activator/prepare.go` (create) | `prepare(rt, role)` and the launch-file writer |
43	| `internal/activator/backend_exec.go` (modify) | dispatch by layout; `startUnit`, `seedUnit`, `startSidecarUnit`; `childEnv` dedupe; path helpers replaced by `pathsFor` |
44	| `internal/activator/activator.go` (modify) | the three changes of §6.3, §6.7, §6.8; `Routes` seam; `SweepOrphans` hook |
45	| `internal/activator/*_test.go` (create/modify) | tests per task |
46	| `internal/appconfig/syncexec.go` (modify) | nested write path; reap walks `instances/*/` |
47	| `internal/schedule/runner.go`, `internal/host/schedule_wiring.go` (modify) | token path by layout |
48	| `internal/host/movespass.go` (modify) | DB path from the row; refuse a fenced host |
49	| `internal/ops/litestream.go` (modify) | glob gains `instances/*/*.db` |
50	| `internal/host/agent.go` (modify) | wiring: `Control`, `LayoutInputs`, `CgroupRoot`, `Routes`, startup sweep, mount and ready gate |
51
52	---
53
54	### Task 1: Journal path, `ReadBeneathLimit`, and the control client
55
56	**Files:**
57	- Modify: `internal/provision/provision.go` (append after `LogSocket`)
58	- Modify: `internal/instancefs/instancefs.go`, `instancefs_linux.go`, `instancefs_other.go`
59	- Create: `internal/instancectl/client.go`, `internal/instancectl/client_test.go`
60	- Test: `internal/provision/provision_test.go`, `internal/instancefs/instancefs_linux_test.go` (append)
61
62	**Interfaces:**
63	- Produces:
64	```go
65	// provision
66	func LayoutJournalPath(dataDir, host string) string // <dataDir>/layout/<host>.json
67	// instancefs
68	var ErrTooLarge = errors.New("instancefs: file larger than the limit")
69	func ReadBeneathLimit(dir, name string, max int64) ([]byte, error)
70	// instancectl
71	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)
72	func NewClient(path string) *Client
73	func (c *Client) Start(ctx context.Context, host string, role Role, memoryMax int64) (string, error) // invocation id
74	func (c *Client) Seed(ctx context.Context, host string, memoryMax int64) (SeedResponse, error)
75	func (c *Client) Stop(ctx context.Context, host string, role Role) error
76	func (c *Client) Reset(ctx context.Context, host string, role Role) error
77	func (c *Client) Sweep(ctx context.Context) (SweepResponse, error)
78	func (c *Client) Ready(ctx context.Context) error // nil on 204; *NotReadyError carrying the ProbeReport on 503
79	type NotReadyError struct{ Report ProbeReport }
80	const ( DeadlineControl = 30*time.Second; DeadlineSeed = 150*time.Second; DeadlineSweep = 60*time.Second; DeadlineReady = 5*time.Second )
81	```
82
83	- [ ] **Step 1: Write the failing tests**
84
85	Append to `internal/provision/provision_test.go`:
86
87	```go
88	func TestLayoutJournalPath(t *testing.T) {
89	if got := LayoutJournalPath("/data/", "h.example.com"); got != "/data/layout/h.example.com.json" {
90	t.Errorf("LayoutJournalPath = %q", got)
91	}
92	}
93	```
94
95	Append to `internal/instancefs/instancefs_linux_test.go`:
96
97	```go
98	func TestReadBeneathLimitRefusesALargeFile(t *testing.T) {
99	dir := t.TempDir()
100	if err := os.WriteFile(filepath.Join(dir, "big"), make([]byte, 4096), 0o600); err != nil {
101	t.Fatal(err)
102	}
103	if _, err := ReadBeneathLimit(dir, "big", 1024); !errors.Is(err, ErrTooLarge) {
104	t.Fatalf("err = %v, want ErrTooLarge", err)
105	}
106	got, err := ReadBeneathLimit(dir, "big", 4096)
107	if err != nil || len(got) != 4096 {
108	t.Fatalf("exact limit: %d bytes, %v", len(got), err)
109	}
110	}
111	```
112
113	Create `internal/instancectl/client_test.go`:
114
115	```go
116	package instancectl
117
118	import (
119	"context"
120	"encoding/json"
121	"errors"
122	"net"
123	"os"
124	"path/filepath"
125	"strings"
126	"testing"
127	"time"
128	)
129
130	// serveFake starts a Server on a temp unix socket with the given fake
131	// systemctl and a probe report, and returns a Client for it.
132	func serveFake(t *testing.T, ctl *fakeCtl, report *ProbeReport) *Client {
133	t.Helper()
134	root := t.TempDir()
135	s := &Server{
136	Systemctl: ctl.run, PeerOK: func(net.Conn) error { return nil },
137	CgroupRoot: filepath.Join(root, "cgroup"), ProbeOut: filepath.Join(root, "probe.out"),
138	FixtureOK: func() error { return nil }, ProbeEvery: time.Hour, Logf: t.Logf,
139	}
140	if report != nil {
141	b, _ := json.Marshal(report)
142	if err := os.WriteFile(s.ProbeOut, b, 0o644); err != nil {
143	t.Fatal(err)
144	}
145	s.RunProbe(context.Background())
146	}
147	dir, err := os.MkdirTemp("", "icc")
148	if err != nil {
149	t.Fatal(err)
150	}
151	t.Cleanup(func() { os.RemoveAll(dir) })
152	sock := filepath.Join(dir, "c.sock")
153	ln, err := net.Listen("unix", sock)
154	if err != nil {
155	t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
156	}
157	ctx, cancel := context.WithCancel(context.Background())
158	t.Cleanup(cancel)
159	go s.Serve(ctx, ln) //nolint:errcheck
160	t.Cleanup(func() { ln.Close() })
161	return NewClient(sock)
162	}
163
164	func TestClientStartCarriesTierAndReturnsInvocation(t *testing.T) {
165	ctl := &fakeCtl{answers: map[string]string{"show -p InvocationID --value": "inv1\n"}}
166	c := serveFake(t, ctl, nil)
167	inv, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912)
168	if err != nil || inv != "inv1" {
169	t.Fatalf("Start = %q, %v", inv, err)
170	}
171	if !strings.Contains(ctl.joined(), "MemoryMax=536870912") {
172	t.Errorf("tier not sent: %s", ctl.joined())
173	}
174	}
175
176	func TestClientSeedStopSweepReady(t *testing.T) {
177	ctl := &fakeCtl{answers: map[string]string{
178	"show -p InvocationID,Result,ExecMainStatus,ExecMainCode": "InvocationID=s1\nResult=success\nExecMainStatus=0\nExecMainCode=exited\n",
179	"list-units": "", "list-jobs": "",
180	}}
181	c := serveFake(t, ctl, &ProbeReport{Ready: true})
182	res, err := c.Seed(context.Background(), "h.example.com", 536870912)
183	if err != nil || res.Result != "success" || res.InvocationID != "s1" {
184	t.Fatalf("Seed = %+v, %v", res, err)
185	}
186	if err := c.Stop(context.Background(), "h.example.com", RoleSidecar); err != nil {
187	t.Fatalf("Stop: %v", err)
188	}
189	if err := c.Reset(context.Background(), "h.example.com", RoleRun); err != nil {
190	t.Fatalf("Reset: %v", err)
191	}
192	sw, err := c.Sweep(context.Background())
193	if err != nil || len(sw.Units) != 0 {
194	t.Fatalf("Sweep = %+v, %v", sw, err)
195	}
196	if err := c.Ready(context.Background()); err != nil {
197	t.Fatalf("Ready: %v", err)
198	}
199	}
200
201	func TestClientReadyReturnsTheReport(t *testing.T) {
202	ctl := &fakeCtl{}
203	c := serveFake(t, ctl, &ProbeReport{Ready: false, Lines: []VerdictLine{{Check: "pid1", Verdict: "fail", Detail: "pid=7"}}})
204	err := c.Ready(context.Background())
205	var nr *NotReadyError
206	if !errors.As(err, &nr) || len(nr.Report.Lines) != 1 || nr.Report.Lines[0].Check != "pid1" {
207	t.Fatalf("Ready = %v", err)
208	}
209	}
210
211	func TestClientSurfacesA502AsAnError(t *testing.T) {
212	ctl := &fakeCtl{fail: map[string]error{"start": errors.New("boom")}} // keyed the way fakeCtl.run matches: by leading-argument prefix
213	c := serveFake(t, ctl, nil)
214	if _, err := c.Start(context.Background(), "h.example.com", RoleRun, 536870912); err == nil || !strings.Contains(err.Error(), "boom") {
215	t.Fatalf("Start error = %v, want the server's message", err)
216	}
217	}
218
219	func TestClientRefusesWhenTheSocketIsAbsent(t *testing.T) {
220	c := NewClient(filepath.Join(t.TempDir(), "absent.sock"))
221	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
222	defer cancel()
223	if err := c.Ready(ctx); err == nil || !strings.Contains(err.Error(), "instance control socket") {
224	t.Fatalf("absent socket: %v", err)
225	}
226	}
227	```
228
229	`fakeCtl` already exists in `server_test.go` (same package).
230
231	- [ ] **Step 2: Run the tests to verify they fail**
232
233	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`
234	Expected: build failures for the three undefined symbols.
235
236	- [ ] **Step 3: Write the code**
237
238	`internal/provision/provision.go`, after `LogSocket`:
239
240	```go
241	// LayoutJournalPath is the migration journal for a host in flight
242	// (isolated-exec design §8.2): under <dataDir>/layout, outside every tenant
243	// mount. Its presence is what makes a host `fenced` (§8.1); plan 4 writes
244	// it, and every consumer that must not act on a half-migrated host checks
245	// for it here.
246	func LayoutJournalPath(dataDir, host string) string {
247	return filepath.Join(dataDir, "layout", host+".json")
248	}
249	```
250
251	`internal/instancefs/instancefs.go`:
252
253	```go
254	// ErrTooLarge is returned by ReadBeneathLimit when the file exceeds max.
255	var ErrTooLarge = errors.New("instancefs: file larger than the limit")
256
257	// ReadBeneathLimit is ReadBeneath with a size cap. Files in an instance
258	// directory are tenant-writable, and some are read by the agent outside the
259	// tenant's memory wall (the exit record, the launch file), so no read there
260	// is unbounded.
261	func ReadBeneathLimit(dir, name string, max int64) ([]byte, error) {
262	f, err := OpenBeneath(dir, name, os.O_RDONLY, 0)
263	if err != nil {
264	return nil, err
265	}
266	defer f.Close()
267	b, err := io.ReadAll(io.LimitReader(f, max+1))
268	if err != nil {
269	return nil, err
270	}
271	if int64(len(b)) > max {
272	return nil, fmt.Errorf("%w: %s over %d bytes", ErrTooLarge, filepath.Join(dir, name), max)
273	}
274	return b, nil
275	}
276	```
277
278	(add `"fmt"` to that file's imports; `OpenBeneath` on non-Linux returns `ErrUnsupported`, so no stub is needed.)
279
280	`internal/instancectl/client.go`:
281
282	```go
283	package instancectl
284
285	import (
286	"context"
287	"encoding/json"
288	"errors"
289	"fmt"
290	"io"
291	"net"
292	"net/http"
293	"net/url"
294	"strconv"
295	"strings"
296	"time"
297	)
298
299	// Deadlines per verb (design §5.3). A start or stop that exceeds its
300	// deadline is a failed wake or an unproven stop, never a hang: §6.7 puts
301	// one such call on a path that must not stall the activator. The seed's is
302	// long because a seed is allowed to be slow.
303	const (
304	DeadlineControl = 30 * time.Second
305	DeadlineSeed = 150 * time.Second
306	DeadlineSweep = 60 * time.Second
307	DeadlineReady = 5 * time.Second
308	)
309
310	// Client is the agent's side of the control channel. It is plain HTTP over
311	// the unix socket at Path; unixdial's checks are for UPSTREAMS the edge
312	// proxies to, and this socket is the one path they refuse, so the client
313	// dials it directly.
314	type Client struct {
315	Path string
316	// Dial is the dialer; nil means net.Dialer. Tests do not need to override
317	// it, but a caller composing several clients may.
318	Dial func(ctx context.Context, path string) (net.Conn, error)
319	}
320
321	func NewClient(path string) *Client { return &Client{Path: path} }
322
323	// NotReadyError carries the probe verdict a 503 /ready returned.
324	type NotReadyError struct{ Report ProbeReport }
325
326	func (e *NotReadyError) Error() string {
327	var fails []string
328	for _, l := range e.Report.Lines {
329	if l.Verdict != "ok" {
330	fails = append(fails, l.Check+" "+l.Verdict+" "+l.Detail)
331	}
332	}
333	if len(fails) == 0 {
334	return "instance control: box not ready (no verdict yet)"
335	}
336	return "instance control: box not ready: " + strings.Join(fails, "; ")
337	}
338
339	func (c *Client) httpClient() *http.Client {
340	dial := c.Dial
341	if dial == nil {
342	dial = func(ctx context.Context, path string) (net.Conn, error) {
343	return (&net.Dialer{}).DialContext(ctx, "unix", path)
344	}
345	}
346	return &http.Client{Transport: &http.Transport{
347	DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { return dial(ctx, c.Path) },
348	DisableKeepAlives: true,
349	}}
350	}
351
352	// do issues one verb with the given deadline and decodes a JSON body when
353	// there is one. A non-2xx answer becomes an error carrying the server's
354	// message; a dial failure names the socket so a forgotten enable reads as
355	// what it is from carlos logs.
356	func (c *Client) do(ctx context.Context, deadline time.Duration, method, path string, q url.Values, out any) (int, error) {
357	ctx, cancel := context.WithTimeout(ctx, deadline)
358	defer cancel()
359	u := "http://control" + path
360	if len(q) > 0 {
361	u += "?" + q.Encode()
362	}
363	req, err := http.NewRequestWithContext(ctx, method, u, nil)
364	if err != nil {
365	return 0, err
366	}
367	resp, err := c.httpClient().Do(req)
368	if err != nil {
369	return 0, fmt.Errorf("instance control socket %s: %w", c.Path, err)
370	}
371	defer resp.Body.Close()
372	body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
373	if resp.StatusCode == http.StatusServiceUnavailable && path == "/ready" {
374	var r ProbeReport
375	_ = json.Unmarshal(body, &r)
376	return resp.StatusCode, &NotReadyError{Report: r}
377	}
378	if resp.StatusCode/100 != 2 {
379	return resp.StatusCode, fmt.Errorf("instance control %s: %d: %s", path, resp.StatusCode, strings.TrimSpace(string(body)))
380	}
381	if out != nil && len(body) > 0 {
382	if err := json.Unmarshal(body, out); err != nil {
383	return resp.StatusCode, fmt.Errorf("instance control %s: %v", path, err)
384	}
385	}
386	return resp.StatusCode, nil
387	}
388
389	func hostRole(host string, role Role) url.Values {
390	q := url.Values{"host": {host}}
391	if role != "" {
392	q.Set("role", string(role))
393	}
394	return q
395	}
396
397	// Start starts the run or sidecar unit for host with the memory tier and
398	// returns the invocation id the server read after the start.
399	func (c *Client) Start(ctx context.Context, host string, role Role, memoryMax int64) (string, error) {
400	q := hostRole(host, role)
401	q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
402	var out StartResponse
403	if _, err := c.do(ctx, DeadlineControl, http.MethodPost, "/start", q, &out); err != nil {
404	return "", err
405	}
406	return out.InvocationID, nil
407	}
408
409	// Seed runs the seed unit to completion and returns its retained result.
410	func (c *Client) Seed(ctx context.Context, host string, memoryMax int64) (SeedResponse, error) {
411	q := hostRole(host, "")
412	q.Set("memory_max", strconv.FormatInt(memoryMax, 10))
413	var out SeedResponse
414	_, err := c.do(ctx, DeadlineSeed, http.MethodPost, "/seed", q, &out)
415	return out, err
416	}
417
418	func (c *Client) Stop(ctx context.Context, host string, role Role) error {
419	_, err := c.do(ctx, DeadlineControl, http.MethodPost, "/stop", hostRole(host, role), nil)
420	return err
421	}
422
423	func (c *Client) Reset(ctx context.Context, host string, role Role) error {
424	_, err := c.do(ctx, DeadlineControl, http.MethodPost, "/reset", hostRole(host, role), nil)
425	return err
426	}
427
428	// Sweep stops every instance unit on the box (design §6.7). A 502 carries
429	// the units that were stopped and the ones that would not; both are
430	// returned with the error.
431	func (c *Client) Sweep(ctx context.Context) (SweepResponse, error) {
432	var out SweepResponse
433	status, err := c.do(ctx, DeadlineSweep, http.MethodPost, "/sweep", nil, &out)
434	if err != nil && status == http.StatusBadGateway {
435	return out, errors.Join(err, errors.New("some instance units would not stop"))
436	}
437	return out, err
438	}
439
440	// Ready is nil on 204 and a *NotReadyError on 503.
441	func (c *Client) Ready(ctx context.Context) error {
442	_, err := c.do(ctx, DeadlineReady, http.MethodGet, "/ready", nil, nil)
443	return err
444	}
445	```
446
447	`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.
448
449	- [ ] **Step 4: Run the tests to verify they pass**
450
451	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)
452	Expected: PASS.
453
454	- [ ] **Step 5: gofmt and commit**
455
456	```bash
457	gofmt -l internal/provision internal/instancefs internal/instancectl
458	git add internal/provision internal/instancefs internal/instancectl
459	git commit -m "instancectl client, LayoutJournalPath, ReadBeneathLimit (plan 3, task 1)"
460	```
461
462	---
463
464	### Task 2: Layout classification and per-host paths
465
466	**Files:**
467	- Create: `internal/activator/layout.go`, `internal/activator/paths.go`
468	- Create: `internal/activator/layout_test.go`
469
470	**Interfaces:**
471	- Produces:
472	```go
473	type Layout string
474	const ( LayoutFlat Layout = "flat"; LayoutNested Layout = "nested"; LayoutFenced Layout = "fenced"; LayoutConflict Layout = "conflict"; LayoutForeign Layout = "foreign" )
475	type LayoutInputs struct {
476	DataDir string
477	Journal func(host string) bool // nil: never fenced
478	Peers func() ([]registry.Route, error) // nil: never conflict
479	}
480	func Classify(rt registry.Route, in LayoutInputs) (Layout, string) // layout and a reason for fenced/conflict/foreign
481	func ActivatorManaged(rt registry.Route) bool // Kind instance, Hibernate, Backing "", Tunnel "", Socket != ""
482	type instancePaths struct { layout Layout; dataDir, runDir, dbPath, envFile, sidecarEnv, configEnv string }
483	func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths
484	```
485
486	- [ ] **Step 1: Write the failing tests**
487
488	Create `internal/activator/layout_test.go`:
489
490	```go
491	package activator
492
493	import (
494	"strings"
495	"testing"
496
497	"github.com/carlosframework/platform/internal/provision"
498	"github.com/carlosframework/platform/internal/registry"
499	)
500
501	func managed(host, db, sock string) registry.Route {
502	return registry.Route{Host: host, Kind: "instance", Hibernate: true, DBPath: db, Socket: sock}
503	}
504
505	func TestClassifyFlatAndNested(t *testing.T) {
506	in := LayoutInputs{DataDir: "/data"}
507	if l, _ := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFlat {
508	t.Errorf("flat row = %s", l)
509	}
510	if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), provision.NestedSocket("h.example.com")), in); l != LayoutNested {
511	t.Errorf("nested row = %s", l)
512	}
513	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") {
514	t.Errorf("custom row = %s %q", l, r)
515	}
516	// A nested DB with a flat socket is not nested: both columns must agree.
517	if l, _ := Classify(managed("h.example.com", provision.NestedDBPath("/data", "h.example.com"), "/run/carlos/h.example.com.sock"), in); l != LayoutForeign {
518	t.Errorf("half-migrated row = %s", l)
519	}
520	}
521
522	func TestClassifyFencedWinsOverEverything(t *testing.T) {
523	in := LayoutInputs{DataDir: "/data", Journal: func(h string) bool { return h == "h.example.com" }}
524	if l, r := Classify(managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock"), in); l != LayoutFenced || r == "" {
525	t.Errorf("fenced = %s %q", l, r)
526	}
527	if l, _ := Classify(managed("o.example.com", "/data/o.example.com.db", "/run/carlos/o.example.com.sock"), in); l != LayoutFlat {
528	t.Errorf("other host = %s", l)
529	}
530	}
531
532	func TestClassifyConflictRefusesBothHosts(t *testing.T) {
533	long := strings.Repeat("a", 60) + ".example.com"
534	twin := strings.Repeat("a", 60) + ".example.org"
535	if provision.InstanceID(long) == provision.InstanceID(twin) {
536	t.Skip("fixture hosts do not collide; pick two that do")
537	}
538	// Force a collision through the Peers seam: a peer whose InstanceID we
539	// fake by giving it the SAME host string is not a conflict (it is the
540	// same row); so build the collision with a peer that is a different
541	// host mapping to the same identifier, which needs a real hash
542	// collision. Instead test the rule directly: a nested row whose
543	// identifier is shared by another managed row classifies as conflict.
544	a := managed(long, provision.NestedDBPath("/data", long), provision.NestedSocket(long))
545	b := managed(twin, provision.NestedDBPath("/data", twin), provision.NestedSocket(twin))
546	in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, b}, nil }, SameID: func(x, y string) bool { return true }}
547	if l, r := Classify(a, in); l != LayoutConflict || !strings.Contains(r, twin) {
548	t.Errorf("conflict = %s %q", l, r)
549	}
550	}
551
552	func TestClassifyIgnoresUnmanagedPeers(t *testing.T) {
553	a := managed("h.example.com", "/data/h.example.com.db", "/run/carlos/h.example.com.sock")
554	ext := registry.Route{Host: "x.example.com", Kind: "instance", Hibernate: false, Socket: "/run/carlos/x.example.com.sock"}
555	in := LayoutInputs{DataDir: "/data", Peers: func() ([]registry.Route, error) { return []registry.Route{a, ext}, nil }, SameID: func(x, y string) bool { return true }}
556	if l, _ := Classify(a, in); l != LayoutFlat {
557	t.Errorf("an unmanaged peer must not conflict: %s", l)
558	}
559	if ActivatorManaged(ext) {
560	t.Error("unflagged socket route reported managed")
561	}
562	}
563	```
564
565	`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.
566
567	- [ ] **Step 2: Run the tests to verify they fail**
568
569	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run TestClassify -count=1`
570	Expected: build failure.
571
572	- [ ] **Step 3: Write `layout.go` and `paths.go`**
573
574	`internal/activator/layout.go`:
575
576	```go
577	package activator
578
579	import (
580	"fmt"
581
582	"github.com/carlosframework/platform/internal/provision"
583	"github.com/carlosframework/platform/internal/registry"
584	)
585
586	// Layout is the classification of an activator-managed route's paths
587	// (isolated-exec design §8.1). It decides the spawn path: flat and foreign
588	// rows run today's exec child; nested rows run as units; fenced and
589	// conflict rows refuse to wake.
590	type Layout string
591
592	const (
593	LayoutFlat Layout = "flat"
594	LayoutNested Layout = "nested"
595	LayoutFenced Layout = "fenced"
596	LayoutConflict Layout = "conflict"
597	LayoutForeign Layout = "foreign"
598	)
599
600	// LayoutInputs are the box facts Classify needs beyond the row: where the
601	// data dir is, whether a migration journal exists for a host (plan 4
602	// writes them), and the other rows on the box (for identifier conflicts).
603	// Nil seams mean "never fenced" and "never conflict", which is what unit
604	// tests without a box want.
605	type LayoutInputs struct {
606	DataDir string
607	// SocketDir is where flat sockets live and under which the nested
608	// instances directory sits. Empty means provision.SocketDir; a test
609	// points it at a temp dir so a route can classify as nested without
610	// writing to /run/carlos.
611	SocketDir string
612	Journal func(host string) bool
613	Peers func() ([]registry.Route, error)
614	// SameID reports whether two hosts share an instance identifier. Nil
615	// means provision.InstanceID equality; tests inject a collision.
616	SameID func(a, b string) bool
617	}
618
619	// sockets returns the flat and nested socket paths for host under the
620	// configured socket directory.
621	func (in LayoutInputs) sockets(host string) (flat, nested string) {
622	dir := in.SocketDir
623	if dir == "" {
624	dir = provision.SocketDir
625	}
626	return filepath.Join(dir, host+".sock"), filepath.Join(dir, "instances", provision.InstanceID(host), "app.sock")
627	}
628
629	// ActivatorManaged is the activator's own predicate for a route it owns
630	// (activator.go's per-host gate and edge.go's proxy decision): an instance
631	// route, hibernate-flagged, with a socket, not unit-backed, not
632	// tunnel-served. An unflagged socket route is an externally supervised
633	// process and is nobody's to migrate or isolate (design §8.2 step 1).
634	func ActivatorManaged(rt registry.Route) bool {
635	return rt.Kind == "instance" && rt.Hibernate && rt.Socket != "" && rt.Backing == "" && rt.Tunnel == ""
636	}
637
638	// Classify returns the layout and, for anything but flat and nested, the
639	// reason a member or operator sees.
640	func Classify(rt registry.Route, in LayoutInputs) (Layout, string) {
641	if in.Journal != nil && in.Journal(rt.Host) {
642	return LayoutFenced, "layout migration in flight; see " + provision.LayoutJournalPath(in.DataDir, rt.Host)
643	}
644	if in.Peers != nil {
645	same := in.SameID
646	if same == nil {
647	same = func(a, b string) bool { return provision.InstanceID(a) == provision.InstanceID(b) }
648	}
649	peers, err := in.Peers()
650	if err == nil {
651	for _, p := range peers {
652	if p.Host != rt.Host && ActivatorManaged(p) && same(p.Host, rt.Host) {
653	return LayoutConflict, fmt.Sprintf("instance identifier collides with %s; neither host wakes until one is removed", p.Host)
654	}
655	}
656	}
657	}
658	flatSock, nestedSock := in.sockets(rt.Host)
659	switch {
660	case rt.DBPath == provision.DefaultDBPath(in.DataDir, rt.Host) && rt.Socket == flatSock:
661	return LayoutFlat, ""
662	case rt.DBPath == provision.NestedDBPath(in.DataDir, rt.Host) && rt.Socket == nestedSock:
663	return LayoutNested, ""
664	}
665	return LayoutForeign, fmt.Sprintf("db %q socket %q are neither the flat nor the nested defaults", rt.DBPath, rt.Socket)
666	}
667	```
668
669	`internal/activator/paths.go`:
670
671	```go
672	package activator
673
674	import (
675	"path/filepath"
676
677	"github.com/carlosframework/platform/internal/provision"
678	"github.com/carlosframework/platform/internal/registry"
679	)
680
681	// instancePaths is every per-host path the exec backend touches, resolved
682	// once per call by layout so that no method builds a path from DataDir and
683	// the host on its own. Flat and foreign rows keep today's flat names; a
684	// nested row's files live in its instance directory with the same names
685	// (design §3.1).
686	type instancePaths struct {
687	layout Layout
688	reason string
689	dataDir string // the directory that holds the four files
690	runDir string // the directory that holds app.sock and the control socket
691	dbPath string
692	envFile string
693	sidecarEnv string
694	configEnv string
695	}
696
697	func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
698	l, reason := Classify(rt, b.layoutInputs())
699	p := instancePaths{layout: l, reason: reason, dataDir: b.DataDir, runDir: provision.SocketDir}
700	if l == LayoutNested {
701	p.dataDir = provision.InstanceDir(b.DataDir, rt.Host)
702	p.runDir = provision.InstanceRunDir(rt.Host)
703	}
704	p.dbPath = rt.DBPath
705	if p.dbPath == "" {
706	p.dbPath = provision.DefaultDBPath(b.DataDir, rt.Host)
707	}
708	p.envFile = filepath.Join(p.dataDir, rt.Host+".env")
709	p.sidecarEnv = filepath.Join(p.dataDir, rt.Host+"-sidecar.env")
710	p.configEnv = filepath.Join(p.dataDir, rt.Host+".config.env")
711	return p
712	}
713
714	// layoutInputs is the backend's LayoutInputs with DataDir filled in.
715	func (b *ExecBackend) layoutInputs() LayoutInputs {
716	in := b.Layout
717	in.DataDir = b.DataDir
718	return in
719	}
720	```
721
722	Add to `ExecBackend` (backend_exec.go struct) the fields this task introduces, with comments:
723
724	```go
725	// Layout supplies the box facts the classification of §8.1 needs beyond
726	// the row: journals and peers. Zero value: never fenced, never conflict.
727	Layout LayoutInputs
728	```
729
730	- [ ] **Step 4: Run the tests to verify they pass**
731
732	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestClassify' -count=1`
733	Expected: PASS.
734
735	- [ ] **Step 5: Replace the flat path builders with `pathsFor`**
736
737	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.
738
739	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -count=1` (outside the sandbox)
740	Expected: PASS with no test edits.
741
742	- [ ] **Step 6: gofmt and commit**
743
744	```bash
745	gofmt -l internal/activator
746	git add internal/activator
747	git commit -m "activator: layout classification and per-host paths by layout (plan 3, task 2)"
748	```
749
750	---
751
752	### Task 3: `childEnv` deduplication, the `Control` seam, the log listener
753
754	**Files:**
755	- Modify: `internal/activator/backend_exec.go` (`childEnv`)
756	- Create: `internal/activator/control.go`, `internal/activator/loglistener.go`
757	- Create: `internal/activator/loglistener_test.go`; modify `internal/activator/activator_test.go` (one new test for `childEnv`)
758
759	**Interfaces:**
760	- Produces:
761	```go
762	// control.go
763	type Control interface {
764	Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
765	Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
766	Stop(ctx context.Context, host string, role instancectl.Role) error
767	Reset(ctx context.Context, host string, role instancectl.Role) error
768	Sweep(ctx context.Context) (instancectl.SweepResponse, error)
769	Ready(ctx context.Context) error
770	}
771	// UnitState answers systemctl show -p ActiveState,InvocationID for a unit; production execs systemctl (unprivileged).
772	type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)
773	// CgroupEvents reports whether the unit's cgroup exists and is populated, from <root>/carlos.slice/carlos-instances.slice/<slice>/<unit>/cgroup.events.
774	type CgroupEvents func(slice, unit string) (exists, populated bool, err error)
775	func systemctlUnitState(ctx context.Context, unit string) (string, string, error)
776	func cgroupEventsUnder(root string) CgroupEvents
777	// loglistener.go
778	type logListener struct{ ... }
779	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
780	func (l *logListener) Close() error
781	```
782	`instancectl.Client` satisfies `Control` (compile-time assertion in `control.go`: `var _ Control = (*instancectl.Client)(nil)`).
783
784	- [ ] **Step 1: Write the failing tests**
785
786	Append to `internal/activator/activator_test.go`:
787
788	```go
789	func TestChildEnvKeepsOneEntryPerKeyLastWins(t *testing.T) {
790	t.Setenv("PATH", "/usr/bin")
791	env := childEnv(map[string]string{"PATH": "/tenant/bin", "A": "1"})
792	seen := map[string]string{}
793	for _, kv := range env {
794	k, v, _ := strings.Cut(kv, "=")
795	if _, dup := seen[k]; dup {
796	t.Fatalf("duplicate key %s in %v", k, env)
797	}
798	seen[k] = v
799	}
800	if seen["PATH"] != "/tenant/bin" {
801	t.Errorf("PATH = %q, want the tenant's value to win over the inherited one", seen["PATH"])
802	}
803	if seen["A"] != "1" {
804	t.Errorf("A = %q", seen["A"])
805	}
806	}
807	```
808
809	Create `internal/activator/loglistener_test.go`:
810
811	```go
812	package activator
813
814	import (
815	"net"
816	"os"
817	"path/filepath"
818	"strings"
819	"sync"
820	"testing"
821	"time"
822	)
823
824	func TestLogListenerTeesOneConnectionAndRefusesASecond(t *testing.T) {
825	dir, err := os.MkdirTemp("", "ll")
826	if err != nil {
827	t.Fatal(err)
828	}
829	t.Cleanup(func() { os.RemoveAll(dir) })
830	path := filepath.Join(dir, "h.run.sock")
831	var mu sync.Mutex
832	var lines []string
833	l, err := listenLogs(path, func(s string) { mu.Lock(); lines = append(lines, s); mu.Unlock() })
834	if err != nil {
835	t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
836	}
837	defer l.Close()
838	c, err := net.Dial("unix", path)
839	if err != nil {
840	t.Fatal(err)
841	}
842	if _, err := c.Write([]byte("hello\nworld\n")); err != nil {
843	t.Fatal(err)
844	}
845	c.Close()
846	deadline := time.Now().Add(2 * time.Second)
847	for {
848	mu.Lock()
849	n := len(lines)
850	mu.Unlock()
851	if n == 2 || time.Now().After(deadline) {
852	break
853	}
854	time.Sleep(10 * time.Millisecond)
855	}
856	mu.Lock()
857	got := strings.Join(lines, "|")
858	mu.Unlock()
859	if got != "hello|world" {
860	t.Fatalf("lines = %q", got)
861	}
862	// A second connection during the same invocation is closed unread.
863	c2, err := net.Dial("unix", path)
864	if err != nil {
865	t.Fatal(err)
866	}
867	c2.SetReadDeadline(time.Now().Add(time.Second))
868	buf := make([]byte, 1)
869	if _, err := c2.Read(buf); err == nil {
870	t.Fatal("second connection was not closed")
871	}
872	c2.Close()
873	if l.Exited() {
874	t.Error("EOF on the log connection must not mark the process exited")
875	}
876	}
877	```
878
879	- [ ] **Step 2: Run the tests to verify they fail**
880
881	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestChildEnvKeepsOneEntry|TestLogListener' -count=1`
882	Expected: `TestChildEnvKeepsOneEntry` FAILS (duplicate PATH today); `TestLogListener` fails to build.
883
884	- [ ] **Step 3: Write the code**
885
886	`childEnv` in `backend_exec.go` becomes:
887
888	```go
889	// childEnv is the child's complete environment: the agent's own, minus
890	// every CARLOS_* variable, with the instance's file values layered on top.
891	// One entry per key, last value wins: os/exec deduplicates cmd.Env that
892	// way, but the unit path hands this list to syscall.Exec, which passes
893	// duplicates through, and a Go child then keeps the FIRST occurrence
894	// (design §6.2, codex R1-16). Composing it deduplicated here keeps the
895	// two spawn paths identical.
896	func childEnv(fileEnv map[string]string) []string {
897	order := make([]string, 0, 32)
898	values := make(map[string]string, 32)
899	set := func(k, v string) {
900	if _, seen := values[k]; !seen {
901	order = append(order, k)
902	}
903	values[k] = v
904	}
905	for _, kv := range os.Environ() {
906	if strings.HasPrefix(kv, "CARLOS_") {
907	continue
908	}
909	k, v, ok := strings.Cut(kv, "=")
910	if !ok {
911	continue
912	}
913	set(k, v)
914	}
915	keys := make([]string, 0, len(fileEnv))
916	for k := range fileEnv {
917	keys = append(keys, k)
918	}
919	sort.Strings(keys)
920	for _, k := range keys {
921	set(k, fileEnv[k])
922	}
923	env := make([]string, 0, len(order))
924	for _, k := range order {
925	env = append(env, k+"="+values[k])
926	}
927	return env
928	}
929	```
930
931	(`sort` import; map iteration order for `fileEnv` is made deterministic by sorting keys, which also makes the launch file stable across writes.)
932
933	`internal/activator/control.go`:
934
935	```go
936	package activator
937
938	import (
939	"context"
940	"errors"
941	"os"
942	"os/exec"
943	"path/filepath"
944	"strings"
945
946	"github.com/carlosframework/platform/internal/instancectl"
947	)
948
949	// Control is what the exec backend needs from the root instance-control
950	// channel (design §5.3). instancectl.Client is the production
951	// implementation; tests use a fake.
952	type Control interface {
953	Start(ctx context.Context, host string, role instancectl.Role, memoryMax int64) (string, error)
954	Seed(ctx context.Context, host string, memoryMax int64) (instancectl.SeedResponse, error)
955	Stop(ctx context.Context, host string, role instancectl.Role) error
956	Reset(ctx context.Context, host string, role instancectl.Role) error
957	Sweep(ctx context.Context) (instancectl.SweepResponse, error)
958	Ready(ctx context.Context) error
959	}
960
961	var _ Control = (*instancectl.Client)(nil)
962
963	// UnitState answers `systemctl show -p ActiveState,InvocationID <unit>`,
964	// which needs no privilege. An unloaded unit answers inactive with an
965	// empty invocation.
966	type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)
967
968	func systemctlUnitState(ctx context.Context, unit string) (string, string, error) {
969	out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "ActiveState,InvocationID", unit).Output()
970	if err != nil {
971	return "", "", err
972	}
973	var active, inv string
974	for _, line := range strings.Split(string(out), "\n") {
975	if k, v, ok := strings.Cut(line, "="); ok {
976	switch k {
977	case "ActiveState":
978	active = strings.TrimSpace(v)
979	case "InvocationID":
980	inv = strings.TrimSpace(v)
981	}
982	}
983	}
984	return active, inv, nil
985	}
986
987	// CgroupEvents reads the unit's cgroup.events under the host slice. cgroup
988	// v2 files are world-readable, so the agent reads them as carlos, as the
989	// sampler does. An absent directory is the unit's cgroup already removed.
990	type CgroupEvents func(slice, unit string) (exists, populated bool, err error)
991
992	func cgroupEventsUnder(root string) CgroupEvents {
993	return func(slice, unit string) (bool, bool, error) {
994	b, err := os.ReadFile(filepath.Join(root, "carlos.slice", "carlos-instances.slice", slice, unit, "cgroup.events"))
995	if errors.Is(err, os.ErrNotExist) {
996	return false, false, nil
997	}
998	if err != nil {
999	return true, true, err
1000	}
1001	return true, !strings.Contains(string(b), "populated 0"), nil
1002	}
1003	}
1004	```
1005
1006	`internal/activator/loglistener.go`:
1007
1008	```go
1009	package activator
1010
1011	import (
1012	"bufio"
1013	"io"
1014	"net"
1015	"os"
1016	"sync"
1017	"sync/atomic"
1018
1019	"github.com/carlosframework/platform/internal/applog"
1020	"github.com/carlosframework/platform/internal/edge"
1021	)
1022
1023	// logListener is the agent's end of a unit's stdout (design §6.4): the
1024	// unit's StandardOutput=file: connects to this AF_UNIX socket once, at
1025	// exec, and every line is teed to the agent's own stdout (journald) and to
1026	// the app log sink, exactly as teeStd does for an exec child. One
1027	// connection per invocation is accepted; a second is closed unread. EOF
1028	// ends log collection and nothing else: it is never a sign the process is
1029	// gone (codex R1-4).
1030	type logListener struct {
1031	ln net.Listener
1032	emit func(string)
1033	accepted atomic.Bool
1034	closed atomic.Bool
1035	wg sync.WaitGroup
1036	}
1037
1038	func listenLogs(path string, emit func(string)) (*logListener, error) {
1039	ln, err := edge.SecureUnixListen(path)
1040	if err != nil {
1041	return nil, err
1042	}
1043	l := &logListener{ln: ln, emit: emit}
1044	l.wg.Add(1)
1045	go l.accept()
1046	return l, nil
1047	}
1048
1049	func (l *logListener) accept() {
1050	defer l.wg.Done()
1051	for {
1052	c, err := l.ln.Accept()
1053	if err != nil {
1054	return
1055	}
1056	if l.accepted.Swap(true) {
1057	c.Close()
1058	continue
1059	}
1060	go l.tee(c)
1061	}
1062	}
1063
1064	func (l *logListener) tee(c net.Conn) {
1065	defer c.Close()
1066	var sink io.Writer = os.Stdout
1067	if l.emit != nil {
1068	sink = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, l.emit))
1069	}
1070	r := bufio.NewReaderSize(c, execLogLineMax)
1071	_, _ = io.Copy(sink, r)
1072	}
1073
1074	// Exited always reports false: a log listener knows nothing about process
1075	// state, and this method exists so a caller cannot mistake it for one.
1076	func (l *logListener) Exited() bool { return false }
1077
1078	func (l *logListener) Close() error {
1079	if l.closed.Swap(true) {
1080	return nil
1081	}
1082	err := l.ln.Close()
1083	l.wg.Wait()
1084	return err
1085	}
1086	```
1087
1088	`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.
1089
1090	- [ ] **Step 4: Run the tests to verify they pass**
1091
1092	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)
1093	Expected: PASS.
1094
1095	- [ ] **Step 5: gofmt and commit**
1096
1097	```bash
1098	gofmt -l internal/activator internal/edge
1099	git add internal/activator internal/edge
1100	git commit -m "activator: deduplicated child env, the Control seam, the per-role log listener (plan 3, task 3)"
1101	```
1102
1103	---
1104
1105	### Task 4: `prepare`, the launch file, `unitProc` and the exit observer
1106
1107	**Files:**
1108	- Create: `internal/activator/prepare.go`, `internal/activator/unitproc.go`
1109	- Create: `internal/activator/unitproc_test.go`
1110	- Modify: `internal/activator/backend_exec.go` (`ExecBackend` gains `Control`, `UnitState`, `CgroupEvents`, `CgroupRoot`; `Start`/`Seed`/`StartSidecar`/`HasSidecar` dispatch)
1111
1112	**Interfaces:**
1113	- Produces:
1114	```go
1115	var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
1116	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 }
1117	func (p *unitProc) Stop(ctx context.Context) error
1118	func (p *unitProc) Exited() bool
1119	func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (paths instancePaths, ctl *edge.ControlListener, logs *logListener, fileEnv map[string]string, err error)
1120	func (b *ExecBackend) writeLaunch(paths instancePaths, role instancectl.Role, host string, argv, env []string) error
1121	func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error)
1122	func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error
1123	func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error)
1124	func (b *ExecBackend) observe(p *unitProc, rt registry.Route) // the goroutine of §6.5
1125	```
1126	- `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).
1127
1128	- [ ] **Step 1: Write the failing tests**
1129
1130	Create `internal/activator/unitproc_test.go`:
1131
1132	```go
1133	package activator
1134
1135	import (
1136	"context"
1137	"encoding/json"
1138	"errors"
1139	"os"
1140	"path/filepath"
1141	"strings"
1142	"sync"
1143	"testing"
1144	"time"
1145
1146	"github.com/carlosframework/platform/internal/instancectl"
1147	"github.com/carlosframework/platform/internal/provision"
1148	"github.com/carlosframework/platform/internal/registry"
1149	)
1150
1151	// fakeControl records verbs and lets a test drive the unit's state.
1152	type fakeControl struct {
1153	mu sync.Mutex
1154	calls []string
1155	inv string
1156	seed instancectl.SeedResponse
1157	startErr error
1158	stopHook func(host string, role instancectl.Role)
1159	readyErr error
1160	}
1161
1162	func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() }
1163	func (f *fakeControl) Start(_ context.Context, host string, role instancectl.Role, max int64) (string, error) {
1164	f.rec("start " + host + " " + string(role))
1165	return f.inv, f.startErr
1166	}
1167	func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancectl.SeedResponse, error) {
1168	f.rec("seed " + host)
1169	return f.seed, nil
1170	}
1171	func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error {
1172	f.rec("stop " + host + " " + string(role))
1173	if f.stopHook != nil {
1174	f.stopHook(host, role)
1175	}
1176	return nil
1177	}
1178	func (f *fakeControl) Reset(_ context.Context, host string, role instancectl.Role) error { f.rec("reset"); return nil }
1179	func (f *fakeControl) Sweep(context.Context) (instancectl.SweepResponse, error) { f.rec("sweep"); return instancectl.SweepResponse{}, nil }
1180	func (f *fakeControl) Ready(context.Context) error { f.rec("ready"); return f.readyErr }
1181	func (f *fakeControl) joined() string { f.mu.Lock(); defer f.mu.Unlock(); return strings.Join(f.calls, "\n") }
1182
1183	// fakeUnit is the unit's observable state: what `systemctl show` and
1184	// cgroup.events would say.
1185	type fakeUnit struct {
1186	mu sync.Mutex
1187	active string
1188	inv string
1189	exists bool
1190	populated bool
1191	}
1192
1193	func (u *fakeUnit) set(active, inv string, exists, populated bool) {
1194	u.mu.Lock()
1195	u.active, u.inv, u.exists, u.populated = active, inv, exists, populated
1196	u.mu.Unlock()
1197	}
1198	func (u *fakeUnit) state(context.Context, string) (string, string, error) {
1199	u.mu.Lock()
1200	defer u.mu.Unlock()
1201	return u.active, u.inv, nil
1202	}
1203	func (u *fakeUnit) events(string, string) (bool, bool, error) {
1204	u.mu.Lock()
1205	defer u.mu.Unlock()
1206	return u.exists, u.populated, nil
1207	}
1208
1209	// nestedBackend returns an ExecBackend wired for a nested host with fakes,
1210	// and the route. The data and run dirs are temp dirs standing in for
1211	// /data/instances/<iid> and /run/carlos/instances/<iid>; provision's
1212	// helpers are not used for the temp paths, so the route is built to
1213	// classify as nested by construction through the Layout seam.
1214	func nestedBackend(t *testing.T, ctl *fakeControl, unit *fakeUnit) (*ExecBackend, registry.Route) {
1215	t.Helper()
1216	root, err := os.MkdirTemp("", "nb")
1217	if err != nil {
1218	t.Fatal(err)
1219	}
1220	t.Cleanup(func() { os.RemoveAll(root) })
1221	host := "h.example.com"
1222	b := testExecBackend(t)
1223	b.DataDir = filepath.Join(root, "data")
1224	b.Control = ctl
1225	b.UnitState = unit.state
1226	b.CgroupEvents = unit.events
1227	b.ObservePoll = 20 * time.Millisecond
1228	b.RunDirOverride = filepath.Join(root, "run")
1229	stageLive(t, b, host, "v1")
1230	rt := registry.Route{Host: host, Kind: "instance", Hibernate: true,
1231	DBPath: provision.NestedDBPath(b.DataDir, host), Socket: filepath.Join(b.RunDirOverride, "instances", host, "app.sock")}
1232	return b, rt
1233	}
1234	```
1235
1236	`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.
1237
1238	```go
1239	func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) {
1240	ctl := &fakeControl{inv: "inv1"}
1241	unit := &fakeUnit{}
1242	b, rt := nestedBackend(t, ctl, unit)
1243	// The fake control does not start anything, so health must be faked:
1244	// bind a listener at rt.Socket that answers 200.
1245	stopServing := serveOK(t, rt.Socket)
1246	defer stopServing()
1247	unit.set("active", "inv1", true, true)
1248	p, ver, err := b.Start(context.Background(), rt)
1249	if err != nil {
1250	t.Fatalf("Start: %v", err)
1251	}
1252	_ = ver
1253	up, ok := p.(*unitProc)
1254	if !ok || up.invocation != "inv1" {
1255	t.Fatalf("proc = %T %+v", p, p)
1256	}
1257	raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.run.json"))
1258	if err != nil {
1259	t.Fatal(err)
1260	}
1261	var l instancectl.Launch
1262	if err := json.Unmarshal(raw, &l); err != nil {
1263	t.Fatal(err)
1264	}
1265	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 {
1266	t.Errorf("launch = %+v", l)
1267	}
1268	seenCtl := false
1269	for _, kv := range l.Env {
1270	if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") {
1271	seenCtl = true
1272	}
1273	if strings.HasPrefix(kv, "CARLOS_ADMIN_TOKEN=") && len(kv) < 30 {
1274	t.Errorf("token missing from env: %s", kv)
1275	}
1276	}
1277	if !seenCtl {
1278	t.Error("CARLOS_CONTROL_SOCKET not in the launch env")
1279	}
1280	if !strings.Contains(ctl.joined(), "start h.example.com run") {
1281	t.Errorf("control calls: %s", ctl.joined())
1282	}
1283	if p.Exited() {
1284	t.Error("exited before the unit went inactive")
1285	}
1286	// Termination: terminal state plus empty cgroup.
1287	unit.set("inactive", "inv1", false, false)
1288	waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
1289	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
1290	defer cancel()
1291	if err := p.Stop(ctx); err != nil {
1292	t.Errorf("Stop after exit: %v", err)
1293	}
1294	}
1295
1296	func TestUnitProcDoesNotExitOnTerminalStateWhileTheCgroupIsPopulated(t *testing.T) {
1297	ctl := &fakeControl{inv: "inv1"}
1298	unit := &fakeUnit{}
1299	b, rt := nestedBackend(t, ctl, unit)
1300	stopServing := serveOK(t, rt.Socket)
1301	defer stopServing()
1302	unit.set("active", "inv1", true, true)
1303	p, _, err := b.Start(context.Background(), rt)
1304	if err != nil {
1305	t.Fatal(err)
1306	}
1307	unit.set("failed", "inv1", true, true) // a process survived SIGKILL
1308	time.Sleep(150 * time.Millisecond)
1309	if p.Exited() {
1310	t.Fatal("exited while the cgroup is still populated (codex R2-5)")
1311	}
1312	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
1313	defer cancel()
1314	if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) {
1315	t.Fatalf("Stop = %v, want ErrStillRunning", err)
1316	}
1317	unit.set("failed", "inv1", false, false)
1318	waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
1319	}
1320
1321	func TestUnitProcStopsAForeignInvocation(t *testing.T) {
1322	ctl := &fakeControl{inv: "inv1"}
1323	unit := &fakeUnit{}
1324	b, rt := nestedBackend(t, ctl, unit)
1325	stopServing := serveOK(t, rt.Socket)
1326	defer stopServing()
1327	unit.set("active", "inv1", true, true)
1328	p, _, err := b.Start(context.Background(), rt)
1329	if err != nil {
1330	t.Fatal(err)
1331	}
1332	ctl.stopHook = func(string, instancectl.Role) { unit.set("inactive", "", false, false) }
1333	unit.set("active", "inv2", true, true) // someone restarted it outside the activator
1334	waitFor(t, func() bool { return p.Exited() }, 2*time.Second)
1335	if !strings.Contains(ctl.joined(), "stop h.example.com run") {
1336	t.Errorf("foreign invocation was not stopped: %s", ctl.joined())
1337	}
1338	}
1339
1340	func TestStartUnitRefusesWhenTheBoxIsNotReadyOrFenced(t *testing.T) {
1341	ctl := &fakeControl{inv: "inv1"}
1342	unit := &fakeUnit{}
1343	b, rt := nestedBackend(t, ctl, unit)
1344	b.NestedOK = func() error { return errors.New("probe: pid1 fail") }
1345	if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "pid1") {
1346	t.Fatalf("not-ready wake = %v", err)
1347	}
1348	if strings.Contains(ctl.joined(), "start") {
1349	t.Error("a start was issued on a box that is not ready")
1350	}
1351	b.NestedOK = nil
1352	b.Layout.Journal = func(string) bool { return true }
1353	if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "migration in flight") {
1354	t.Fatalf("fenced wake = %v", err)
1355	}
1356	}
1357
1358	func TestSeedUnitCleansUpOnlyItsOwnDBOnFailure(t *testing.T) {
1359	ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}}
1360	unit := &fakeUnit{}
1361	b, rt := nestedBackend(t, ctl, unit)
1362	rt.Seed = `["seed","--fast"]`
1363	unit.set("inactive", "", false, false)
1364	err := b.Seed(context.Background(), rt)
1365	if err == nil || !strings.Contains(err.Error(), "exit 3") {
1366	t.Fatalf("Seed = %v", err)
1367	}
1368	raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.seed.json"))
1369	if err != nil {
1370	t.Fatal(err)
1371	}
1372	var l instancectl.Launch
1373	_ = json.Unmarshal(raw, &l)
1374	if len(l.Argv) != 5 || l.Argv[1] != "seed" || l.Argv[2] != "--fast" || l.Argv[3] != "--db" || l.Argv[4] != rt.DBPath {
1375	t.Errorf("seed argv = %v", l.Argv)
1376	}
1377	for _, kv := range l.Env {
1378	if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") || strings.HasPrefix(kv, "CARLOS_PURGE=") {
1379	t.Errorf("seed env carries %s", kv)
1380	}
1381	}
1382	if _, err := os.Stat(rt.DBPath); !os.IsNotExist(err) {
1383	t.Errorf("failed seed left its database: %v", err)
1384	}
1385	}
1386	```
1387
1388	Add helpers to the same file:
1389
1390	```go
1391	// serveOK binds a unix socket at path and answers 200 to every request, so
1392	// waitHealthyCtx passes without a real unit.
1393	func serveOK(t *testing.T, path string) func() {
1394	t.Helper()
1395	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
1396	t.Fatal(err)
1397	}
1398	ln, err := net.Listen("unix", path)
1399	if err != nil {
1400	t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
1401	}
1402	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })}
1403	go srv.Serve(ln) //nolint:errcheck
1404	return func() { srv.Close(); ln.Close() }
1405	}
1406
1407	func waitFor(t *testing.T, cond func() bool, d time.Duration) {
1408	t.Helper()
1409	deadline := time.Now().Add(d)
1410	for !cond() {
1411	if time.Now().After(deadline) {
1412	t.Fatal("condition not met in time")
1413	}
1414	time.Sleep(10 * time.Millisecond)
1415	}
1416	}
1417	```
1418
1419	(imports `net`, `net/http` in the test file.)
1420
1421	- [ ] **Step 2: Run the tests to verify they fail**
1422
1423	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestStartUnit|TestUnitProc|TestSeedUnit' -count=1`
1424	Expected: build failure.
1425
1426	- [ ] **Step 3: Write `prepare.go`**
1427
1428	```go
1429	package activator
1430
1431	import (
1432	"encoding/json"
1433	"fmt"
1434	"os"
1435	"path/filepath"
1436
1437	"github.com/carlosframework/platform/internal/edge"
1438	"github.com/carlosframework/platform/internal/instancectl"
1439	"github.com/carlosframework/platform/internal/instancefs"
1440	"github.com/carlosframework/platform/internal/provision"
1441	"github.com/carlosframework/platform/internal/registry"
1442	"github.com/carlosframework/platform/internal/schedule"
1443	)
1444
1445	// prepare is everything a nested role needs before its unit starts
1446	// (design §6.1): the two directories, the admin token and merged env (for
1447	// run and seed), the control listener (run only), the role's log listener,
1448	// and stale records cleared. It returns the listeners so the caller can
1449	// hand them to the proc or close them on a failed start.
1450	func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) {
1451	p := b.pathsFor(rt)
1452	if err := os.MkdirAll(p.dataDir, 0o700); err != nil {
1453	return p, nil, nil, nil, err
1454	}
1455	if err := os.MkdirAll(p.runDir, 0o750|os.ModeSetgid); err != nil {
1456	return p, nil, nil, nil, err
1457	}
1458	if err := os.MkdirAll(filepath.Dir(b.logSocket(rt.Host, role)), 0o700); err != nil {
1459	return p, nil, nil, nil, err
1460	}
1461	var env map[string]string
1462	if role == instancectl.RoleRun || role == instancectl.RoleSeed {
1463	if err := EnsureAdminToken(p.envFile); err != nil {
1464	return p, nil, nil, nil, fmt.Errorf("admin token for %s: %w", rt.Host, err)
1465	}
1466	var err error
1467	env, err = b.instanceEnv(rt)
1468	if err != nil {
1469	return p, nil, nil, nil, err
1470	}
1471	} else {
1472	var err error
1473	env, err = readEnvFileBeneath(p.dataDir, filepath.Base(p.sidecarEnv))
1474	if err != nil {
1475	return p, nil, nil, nil, err
1476	}
1477	}
1478	// Stale exit records from earlier invocations of this role.
1479	if entries, err := os.ReadDir(p.runDir); err == nil {
1480	for _, e := range entries {
1481	if strings.HasPrefix(e.Name(), "exit."+string(role)+".") {
1482	_ = os.Remove(filepath.Join(p.runDir, e.Name()))
1483	}
1484	}
1485	}
1486	var ctl *edge.ControlListener
1487	if role == instancectl.RoleRun {
1488	_ = os.Remove(rt.Socket) // nothing is running; a stale socket file would refuse the app's bind
1489	ctl = b.listenControl(rt, p.envFile)
1490	}
1491	logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
1492	if err != nil {
1493	if ctl != nil {
1494	ctl.Close()
1495	}
1496	return p, nil, nil, nil, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err)
1497	}
1498	return p, ctl, logs, env, nil
1499	}
1500
1501	// listenControl binds the control socket exactly as Start does for an exec
1502	// child (the tokenReader and schedule mount are unchanged); a bind failure
1503	// is logged and the instance runs without its control verbs, as today.
1504	func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.ControlListener {
1505	tokenReader := func() (string, error) {
1506	env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(envPath))
1507	if err != nil {
1508	return "", err
1509	}
1510	return env["CARLOS_ADMIN_TOKEN"], nil
1511	}
1512	var mount func(*http.ServeMux)
1513	if b.Timers != nil {
1514	mount = func(mux *http.ServeMux) { schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now) }
1515	}
1516	ctl, err := edge.ListenControl(rt.Socket+".control", rt.Host, b.Cache, mount)
1517	if err != nil {
1518	log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
1519	return nil
1520	}
1521	return ctl
1522	}
1523
1524	// logSocket is the role's log socket path, honouring the test override of
1525	// the run directory.
1526	func (b *ExecBackend) logSocket(host string, role instancectl.Role) string {
1527	if b.RunDirOverride != "" {
1528	return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+string(role)+".sock")
1529	}
1530	return provision.LogSocket(host, string(role))
1531	}
1532
1533	// logEmitter is the same sink teeStd feeds, or nil when Logs is unset.
1534	func (b *ExecBackend) logEmitter(rt registry.Route) func(string) {
1535	if b.Logs == nil {
1536	return nil
1537	}
1538	return func(line string) { b.Logs.App(release.RouteAccount(rt), rt.App, rt.Host, line) }
1539	}
1540
1541	// writeLaunch writes launch.<role>.json atomically into the instance
1542	// directory (design §6.2). The env is childEnv's deduplicated list; the
1543	// trampoline execs it as is.
1544	func (b *ExecBackend) writeLaunch(p instancePaths, role instancectl.Role, host string, argv, env []string) error {
1545	raw, err := json.MarshalIndent(instancectl.Launch{Schema: instancectl.LaunchSchema, Host: host, Argv: argv, Env: env}, "", " ")
1546	if err != nil {
1547	return err
1548	}
1549	return instancefs.WriteAtomic(p.dataDir, instancectl.LaunchFileName(role), append(raw, '\n'), 0o600)
1550	}
1551
1552	// readEnvFileBeneath is readEnvFile through instancefs: a nested host's env
1553	// files are read without following a symlink (design §3.2). A missing file
1554	// is an empty env, as readEnvFile treats it.
1555	func readEnvFileBeneath(dir, name string) (map[string]string, error) {
1556	b, err := instancefs.ReadBeneathLimit(dir, name, 1<<20)
1557	if errors.Is(err, os.ErrNotExist) {
1558	return map[string]string{}, nil
1559	}
1560	if err != nil {
1561	return nil, err
1562	}
1563	return parseEnvBytes(b), nil
1564	}
1565	```
1566
1567	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`).
1568
1569	- [ ] **Step 4: Write `unitproc.go`**
1570
1571	```go
1572	package activator
1573
1574	import (
1575	"context"
1576	"encoding/json"
1577	"errors"
1578	"fmt"
1579	"log"
1580	"path/filepath"
1581	"sync/atomic"
1582	"time"
1583
1584	"github.com/carlosframework/platform/internal/edge"
1585	"github.com/carlosframework/platform/internal/instancectl"
1586	"github.com/carlosframework/platform/internal/instancefs"
1587	"github.com/carlosframework/platform/internal/registry"
1588	)
1589
1590	// ErrStillRunning is Stop's answer when ctx expired before termination was
1591	// proven (design §6.3). The activator then keeps the lease and the entry
1592	// in stopping, and the sweeper retries; the lease is never released over a
1593	// writer that may still exist.
1594	var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
1595
1596	// unitProc is a nested host's Proc (design §6.3): the unit the control
1597	// service started, the listeners the agent holds for it, and the observer
1598	// that proves termination.
1599	type unitProc struct {
1600	b *ExecBackend
1601	rt registry.Route
1602	role instancectl.Role
1603	unit, slice string
1604	invocation string
1605	runDir string
1606	ctl *edge.ControlListener
1607	logs *logListener
1608	exited atomic.Bool
1609	done chan struct{}
1610	closeOnce atomic.Bool
1611	}
1612
1613	func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc {
1614	up := &unitProc{b: b, rt: rt, role: role, unit: instancectl.UnitName(role, rt.Host), slice: instancectl.SliceName(rt.Host),
1615	invocation: inv, runDir: p.runDir, ctl: ctl, logs: logs, done: make(chan struct{})}
1616	go b.observe(up)
1617	return up
1618	}
1619
1620	func (p *unitProc) Exited() bool { return p.exited.Load() }
1621
1622	// Stop asks the control service to stop the unit and waits for termination
1623	// to be proven. The control listener and the log listener are closed only
1624	// after that, preserving childProc's ordering (signal first, close the
1625	// listener after the process is gone).
1626	func (p *unitProc) Stop(ctx context.Context) error {
1627	if !p.Exited() {
1628	sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl)
1629	err := p.b.Control.Stop(sctx, p.rt.Host, p.role)
1630	cancel()
1631	if err != nil {
1632	log.Printf("activator: stop %s %s: %v", p.rt.Host, p.role, err)
1633	}
1634	}
1635	select {
1636	case <-p.done:
1637	p.closeListeners()
1638	return nil
1639	case <-ctx.Done():
1640	return ErrStillRunning
1641	}
1642	}
1643
1644	func (p *unitProc) closeListeners() {
1645	if p.closeOnce.Swap(true) {
1646	return
1647	}
1648	if p.ctl != nil {
1649	p.ctl.Close()
1650	}
1651	if p.logs != nil {
1652	p.logs.Close()
1653	}
1654	}
1655
1656	// observe is the goroutine of design §6.5. It proves termination from two
1657	// facts the tenant cannot forge: systemd's unit state and cgroup emptiness.
1658	// The exit record decides the wording of the event line and nothing else.
1659	func (b *ExecBackend) observe(p *unitProc) {
1660	poll := b.ObservePoll
1661	if poll <= 0 {
1662	poll = time.Second
1663	}
1664	state := b.UnitState
1665	if state == nil {
1666	state = systemctlUnitState
1667	}
1668	events := b.CgroupEvents
1669	if events == nil {
1670	events = cgroupEventsUnder(b.cgroupRoot())
1671	}
1672	foreignStopped := false
1673	lastRemain := time.Time{}
1674	for {
1675	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
1676	active, inv, err := state(ctx, p.unit)
1677	cancel()
1678	if err != nil {
1679	log.Printf("activator: %s: unit state: %v", p.rt.Host, err)
1680	}
1681	terminal := false
1682	foreign := false
1683	switch {
1684	case err != nil:
1685	case inv != "" && inv != p.invocation:
1686	foreign = true
1687	case active == "inactive" || active == "failed":
1688	terminal = true
1689	}
1690	if foreign && !foreignStopped {
1691	log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation)
1692	sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
1693	if err := b.Control.Stop(sctx, p.rt.Host, p.role); err != nil {
1694	log.Printf("activator: %s: stop foreign invocation: %v", p.rt.Host, err)
1695	}
1696	scancel()
1697	foreignStopped = true
1698	}
1699	if terminal || (foreign && foreignStopped) {
1700	exists, populated, eerr := events(p.slice, p.unit)
1701	if eerr != nil {
1702	log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr)
1703	}
1704	if !exists || !populated {
1705	if terminal {
1706	emitEvent(b.Logs, p.rt, exitLevel(p), exitLine(p))
1707	} else {
1708	emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)")
1709	}
1710	p.exited.Store(true)
1711	close(p.done)
1712	if p.Exited() {
1713	// Stop closes the listeners when it observes done; a
1714	// crash nobody is stopping yet leaves them for the
1715	// sweeper's stand-down, which calls Stop.
1716	}
1717	return
1718	}
1719	if time.Since(lastRemain) > 2*time.Second {
1720	log.Printf("activator: %s: processes remain after %s; not releasing", p.rt.Host, active)
1721	lastRemain = time.Now()
1722	}
1723	}
1724	time.Sleep(poll)
1725	}
1726	}
1727
1728	// exitLine reads the advisory record for this invocation and words the
1729	// event line as childProc does: a non-zero status is a warn line, a clean
1730	// exit is info, and no record is unknown. The record is tenant-writable
1731	// text (design §6.5, final review of plan 2), so every field that reaches
1732	// the line passes through recordToken first.
1733	func exitLine(p *unitProc) string {
1734	rec, ok := p.readRecord()
1735	switch {
1736	case !ok:
1737	return "instance exited (status unknown: no exit record)"
1738	case rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0"):
1739	return "instance exited (exit 0)"
1740	case rec.Code == "killed" || rec.Code == "dumped":
1741	return "instance exited: signal " + recordToken(rec.Status)
1742	default:
1743	return fmt.Sprintf("instance exited: exit %s (%s)", recordToken(rec.Status), recordToken(rec.Result))
1744	}
1745	}
1746
1747	// recordToken bounds a field of the exit record to what systemd itself
1748	// would write: at most 32 bytes of [A-Za-z0-9_-]. Anything else becomes
1749	// "?" so a tenant cannot put arbitrary text or control characters into
1750	// the platform's event stream.
1751	func recordToken(s string) string {
1752	if len(s) == 0 || len(s) > 32 {
1753	return "?"
1754	}
1755	for _, c := range s {
1756	if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-') {
1757	return "?"
1758	}
1759	}
1760	return s
1761	}
1762
1763	func exitLevel(p *unitProc) string {
1764	if rec, ok := p.readRecord(); ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) {
1765	return "info"
1766	}
1767	return "warn"
1768	}
1769
1770	func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) {
1771	raw, err := instancefs.ReadBeneathLimit(p.runDir, instancectl.ExitRecordName(p.role, p.invocation), 4096)
1772	if err != nil {
1773	return instancectl.ExitRecord{}, false
1774	}
1775	var rec instancectl.ExitRecord
1776	if err := json.Unmarshal(raw, &rec); err != nil {
1777	return instancectl.ExitRecord{}, false
1778	}
1779	return rec, true
1780	}
1781
1782	func (b *ExecBackend) cgroupRoot() string {
1783	if b.CgroupRoot != "" {
1784	return b.CgroupRoot
1785	}
1786	return "/sys/fs/cgroup"
1787	}
1788	```
1789
1790	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)`).
1791
1792	- [ ] **Step 5: The dispatch in `backend_exec.go`**
1793
1794	Rename the current `Start` body to `startChild`, `Seed`'s to `seedChild`, `StartSidecar`'s to `startSidecarChild` (signatures unchanged), and write the dispatchers:
1795
1796	```go
1797	func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
1798	p := b.pathsFor(rt)
1799	switch p.layout {
1800	case LayoutNested:
1801	return b.startUnit(ctx, rt)
1802	case LayoutFenced, LayoutConflict:
1803	return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
1804	}
1805	return b.startChild(ctx, rt)
1806	}
1807	```
1808
1809	(and the same shape for `Seed` and `StartSidecar`; `HasSidecar` stats `pathsFor(rt).sidecarEnv`.)
1810
1811	`startUnit`:
1812
1813	```go
1814	func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) {
1815	if b.Control == nil {
1816	return nil, "", fmt.Errorf("%s: instance control channel not configured", rt.Host)
1817	}
1818	if b.NestedOK != nil {
1819	if err := b.NestedOK(); err != nil {
1820	return nil, "", fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err)
1821	}
1822	}
1823	if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
1824	return nil, "", err
1825	}
1826	p, ctl, logs, fileEnv, err := b.prepare(rt, instancectl.RoleRun)
1827	if err != nil {
1828	return nil, "", err
1829	}
1830	fail := func(err error) (Proc, string, error) {
1831	if ctl != nil {
1832	ctl.Close()
1833	}
1834	logs.Close()
1835	return nil, "", err
1836	}
1837	childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil) // the same withVar/withoutVars block Start uses today, extracted so both paths share it
1838	argv := []string{b.liveLink(rt.Host), "--socket", rt.Socket, "--db", p.dbPath}
1839	if err := b.writeLaunch(p, instancectl.RoleRun, rt.Host, argv, childEnv(childVars)); err != nil {
1840	return fail(err)
1841	}
1842	spawnVer, _ := b.LiveVersion(rt.Host)
1843	inv, err := b.Control.Start(ctx, rt.Host, instancectl.RoleRun, b.tier(rt).MemoryMax)
1844	if err != nil {
1845	return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
1846	}
1847	if inv == "" {
1848	// The server read the invocation after the start and found none:
1849	// the unit already exited (plan 2 final review). Nothing is
1850	// running, so there is nothing to stop; the wake failed.
1851	return fail(fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host)))
1852	}
1853	log.Printf("activator: %s: started %s memory.max=%d", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host), b.tier(rt).MemoryMax)
1854	up := b.newUnitProc(rt, instancectl.RoleRun, p, inv, ctl, logs)
1855	if b.Cache != nil {
1856	if n := b.Cache.PurgeHost(rt.Host); n > 0 {
1857	log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
1858	}
1859	}
1860	if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
1861	stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
1862	defer cancel()
1863	_ = up.Stop(stopCtx)
1864	return nil, "", err
1865	}
1866	b.normalizeSocketMode(rt, up) // the existing 0660 block, extracted
1867	return up, spawnVer, nil
1868	}
1869	```
1870
1871	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`.
1872
1873	- [ ] **Step 6: Run the tests to verify they pass**
1874
1875	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -count=1` (outside the sandbox)
1876	Expected: PASS, including every pre-existing test unchanged.
1877
1878	- [ ] **Step 7: gofmt and commit**
1879
1880	```bash
1881	gofmt -l internal/activator
1882	git add internal/activator
1883	git commit -m "activator: the nested spawn path: prepare, launch file, unitProc and the exit observer (plan 3, task 4)"
1884	```
1885
1886	---
1887
1888	### Task 5: The three `activator.go` changes
1889
1890	**Files:**
1891	- Modify: `internal/activator/activator.go` (`stop`, `sweepOnce`, `doWake`, `Service` gains `Routes`)
1892	- Test: `internal/activator/activator_test.go` (append three tests)
1893
1894	**Interfaces:**
1895	- Produces:
1896	```go
1897	// RouteLookup is the registry read doWake performs before Acquire (design §6.8); *registry.Registry satisfies it with Get.
1898	type RouteLookup interface { Get(host string) (*registry.Route, error) }
1899	// Service gains: Routes RouteLookup (nil: the wake runs on the copy the edge supplied, today's behaviour)
1900	```
1901	`actEntry` gains `pendingStop bool`, `sidecarStarting bool`, `wakeGen uint64`.
1902
1903	- [ ] **Step 1: Write the failing tests**
1904
1905	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`.
1906
1907	```go
1908	// A Proc whose Stop reports ErrStillRunning must not have its lease
1909	// released: stop() keeps the entry in stopping and the sweeper finishes it
1910	// once the proc reports exited (design §6.3, codex R2-5).
1911	func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
1912	a, b, clk, _ := testActivator(t, time.Minute)
1913	b.procStuck = true // Start returns a fakeProc whose Stop reports ErrStillRunning until exit()
1914	rt := testRoute("h.example.com")
1915	// drive: wake rt and release the request, as the idle-hibernate test does
1916	clk.advance(2 * time.Minute)
1917	a.sweepOnce() // idle: stop() runs, Stop reports ErrStillRunning
1918	p := b.lastProc()
1919	waitFor(t, func() bool { return p.stopCalls() >= 1 }, 2*time.Second)
1920	if n := b.releaseCount(); n != 0 {
1921	t.Fatalf("lease released %d times while the stop is unproven", n)
1922	}
1923	p.exit()
1924	a.sweepOnce()
1925	waitFor(t, func() bool { return b.releaseCount() == 1 }, 2*time.Second)
1926	}
1927
1928	// The sidecar restart must not hold the activator mutex across the start
1929	// call (design §6.7, codex R2-6): a stalled sidecar start does not block
1930	// another host's wake.
1931	func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
1932	a, b, clk, _ := testActivator(t, time.Hour)
1933	b.sidecarStaged = true
1934	release := make(chan struct{})
1935	b.sidecarGate = release // StartSidecar blocks on this channel when non-nil
1936	// drive: wake a.example.com with the sidecar gate open, then close the gate
1937	b.lastSidecar().exit() // the sweeper now wants a restart
1938	clk.advance(time.Minute)
1939	go a.sweepOnce() // blocks inside StartSidecar until release is closed
1940	time.Sleep(50 * time.Millisecond)
1941	done := make(chan struct{})
1942	go func() {
1943	// drive: wake b.example.com
1944	close(done)
1945	}()
1946	select {
1947	case <-done:
1948	case <-time.After(2 * time.Second):
1949	t.Fatal("a second host's wake blocked behind a stalled sidecar start")
1950	}
1951	close(release)
1952	}
1953
1954	// doWake runs on the committed row, not the copy the edge supplied
1955	// (design §6.8, codex R8-1).
1956	func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
1957	a, b, _, reg := testActivator(t, time.Hour)
1958	a.Routes = reg
1959	old := testRoute("h.example.com")
1960	fresh := old
1961	fresh.DBPath = "/data/instances/h.example.com/h.example.com.db"
1962	fresh.Socket = "/run/carlos/instances/h.example.com/app.sock"
1963	// insert fresh into reg the way the file's registry-backed tests do
1964	// drive: wake with old (the edge's stale copy)
1965	if got := b.lastStartedRoute().DBPath; got != fresh.DBPath {
1966	t.Fatalf("Start saw DBPath %q, want the registry's %q", got, fresh.DBPath)
1967	}
1968	}
1969	```
1970
1971	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.
1972
1973	- [ ] **Step 2: Run the tests to verify they fail**
1974
1975	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -run 'TestStopDoesNotRelease|TestSidecarRestartDoesNotHold|TestDoWakeRefreshes' -count=1`
1976	Expected: FAIL (`Routes` undefined; the lease is released today; the lock is held today).
1977
1978	- [ ] **Step 3: Make the three changes**
1979
1980	In `stop(e *actEntry)`, after `err = e.proc.Stop(procCtx)`:
1981
1982	```go
1983	if errors.Is(err, ErrStillRunning) {
1984	// The lease is never released over an unproven termination (design
1985	// §6.3). Keep the entry in stopping; sweepOnce finishes the stop
1986	// once the proc reports exited.
1987	log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err)
1988	a.mu.Lock()
1989	e.pendingStop = true
1990	a.mu.Unlock()
1991	return
1992	}
1993	```
1994
1995	and in `sweepOnce`, under `a.mu`, before the idle pass:
1996
1997	```go
1998	for _, e := range a.entries {
1999	if e.pendingStop && e.proc != nil && e.proc.Exited() {
2000	e.pendingStop = false
2001	go a.finishStop(e) // rep.Close, delete, close(e.stopped): the tail of stop()
2002	}
2003	}
2004	```
2005
2006	with `finishStop` being the tail of today's `stop()` from `rep.Close` onward, extracted so `stop()` calls it too.
2007
2008	In `sweepOnce`'s sidecar block, replace the synchronous `a.backend.StartSidecar(...)` with:
2009
2010	```go
2011	if !e.sidecarStarting {
2012	e.sidecarStarting = true
2013	gen := e.wakeGen
2014	host := e.host
2015	route := e.route
2016	go func() {
2017	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
2018	sc, err := a.backend.StartSidecar(ctx, route)
2019	cancel()
2020	a.mu.Lock()
2021	defer a.mu.Unlock()
2022	e := a.entries[host]
2023	if e == nil || e.state != actAwake || e.wakeGen != gen {
2024	if sc != nil {
2025	go sc.Stop(context.Background()) //nolint:errcheck
2026	}
2027	return
2028	}
2029	e.sidecarStarting = false
2030	// install exactly as the synchronous code did
2031	...
2032	}()
2033	}
2034	```
2035
2036	(`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.
2037
2038	In `doWake`, at the top, before `Acquire`:
2039
2040	```go
2041	if a.Routes != nil {
2042	fresh, err := a.Routes.Get(rt.Host)
2043	if err != nil {
2044	return nil, fmt.Errorf("%s: route lookup: %w", rt.Host, err)
2045	}
2046	if fresh == nil {
2047	return nil, fmt.Errorf("%s: no route", rt.Host)
2048	}
2049	if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
2050	log.Printf("activator: %s: route refreshed: db %q → %q socket %q → %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
2051	}
2052	rt = *fresh
2053	a.mu.Lock()
2054	if e := a.entries[rt.Host]; e != nil {
2055	e.route = rt
2056	}
2057	a.mu.Unlock()
2058	}
2059	```
2060
2061	(Use a plain `->` in the log line, not an arrow glyph: the copy rule.)
2062
2063	- [ ] **Step 4: Run the whole package**
2064
2065	Run: `TMPDIR=/tmp/g GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off go test ./internal/activator -race -count=1` (outside the sandbox)
2066	Expected: PASS, race-clean.
2067
2068	- [ ] **Step 5: gofmt and commit**
2069
2070	```bash
2071	gofmt -l internal/activator
2072	git add internal/activator
2073	git commit -m "activator: unproven stops keep the lease, sidecar restarts leave the lock, wakes run on the committed row (plan 3, task 5)"
2074	```
2075
2076	---
2077
2078	### Task 6: Consumers of the shape and the agent wiring
2079
2080	**Files:**
2081	- Modify: `internal/appconfig/syncexec.go` (writer path; `reapConfigEnv`)
2082	- Modify: `internal/schedule/runner.go`, `internal/host/schedule_wiring.go` (token path)
2083	- Modify: `internal/host/movespass.go` (DB path from the row; fence)
2084	- Modify: `internal/ops/litestream.go` (glob)
2085	- Modify: `internal/host/agent.go` (wiring, startup sweep, gate)
2086	- Tests in each package
2087
2088	**Interfaces:**
2089	- Consumes: `activator.Classify`, `activator.LayoutInputs`, `activator.ActivatorManaged`, `provision.InstanceDir`, `provision.LayoutJournalPath`, `instancectl.NewClient`, `controlsock.Path`, `instancefs.MountNoSymfollow`.
2090	- 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).
2091
2092	- [ ] **Step 1: Write the failing tests**
2093
2094	`internal/appconfig/syncexec_test.go` (append):
2095
2096	```go
2097	func TestSyncExecWritesANestedRouteIntoItsInstanceDir(t *testing.T) {
2098	// Build the fixture exactly as TestSyncExecWritesConfigFileAndBouncesOnChange
2099	// does, but with the route's DBPath and Socket at the nested defaults,
2100	// and assert the file lands at <dataDir>/instances/<host>/<host>.config.env
2101	// and NOT at <dataDir>/<host>.config.env.
2102	}
2103
2104	func TestReapConfigEnvWalksInstanceDirs(t *testing.T) {
2105	// A stale <dataDir>/instances/gone.example.com/gone.example.com.config.env
2106	// with no route is removed; a live nested route's file is kept.
2107	}
2108	```
2109
2110	Write both bodies by copying the closest existing test in that file and changing the route's two columns and the expected path.
2111
2112	`internal/ops/ops_test.go` (append):
2113
2114	```go
2115	func TestFleetDBsDiscoversNestedDatabases(t *testing.T) {
2116	// A registry with a non-hibernating exec route whose DBPath is
2117	// <dataDir>/instances/<host>/<host>.db and the file present: FleetDBs
2118	// returns it. Copy the shape of the existing flat discovery test.
2119	}
2120	```
2121
2122	`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`.
2123
2124	`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.
2125
2126	- [ ] **Step 2: Run the new tests to verify they fail**
2127
2128	Run each package's new tests; expected: FAIL.
2129
2130	- [ ] **Step 3: Make the changes**
2131
2132	`syncexec.go`: replace `path := filepath.Join(dataDir, rt.Host+configEnvSuffix)` (the writer and the addr warning path) with `path := ConfigEnvPath(dataDir, rt)`:
2133
2134	```go
2135	// ConfigEnvPath is where a route's rendered bundle lands: inside the
2136	// instance directory for a nested row (isolated-exec design §7), flat for
2137	// everything else, including opted-in addr routes whose supervisor reads
2138	// the flat path.
2139	func ConfigEnvPath(dataDir string, rt registry.Route) string {
2140	if rt.DBPath == provision.NestedDBPath(dataDir, rt.Host) && rt.Socket == provision.NestedSocket(rt.Host) {
2141	return filepath.Join(provision.InstanceDir(dataDir, rt.Host), rt.Host+configEnvSuffix)
2142	}
2143	return filepath.Join(dataDir, rt.Host+configEnvSuffix)
2144	}
2145	```
2146
2147	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.
2148
2149	`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.)
2150
2151	`movespass.go`: `moveCfg` gains `Route func(host string) (*registry.Route, error)` (wired in `armMovesPass` as `reg.Get`) and the checkpoint block becomes:
2152
2153	```go
2154	if _, err := os.Stat(provision.LayoutJournalPath(cfg.DataDir, rec.Host)); err == nil {
2155	log.Printf("moves: %s: layout migration in flight; leaving the move for the next pass", rec.Host)
2156	continue // or return nil, matching how the loop skips a host today
2157	}
2158	dbPath := filepath.Join(cfg.DataDir, rec.Host+".db")
2159	if cfg.Route != nil {
2160	if rt, err := cfg.Route(rec.Host); err == nil && rt != nil && rt.DBPath != "" {
2161	dbPath = rt.DBPath
2162	}
2163	}
2164	```
2165
2166	`litestream.go:354`: `matches, err := filepath.Glob(filepath.Join(dataDir, "*.db"))` gains a second glob `filepath.Join(dataDir, "instances", "*", "*.db")` appended to `matches`.
2167
2168	`agent.go`: after `eb := &activator.ExecBackend{...}`:
2169
2170	```go
2171	eb.Control = instancectl.NewClient(controlsock.Path)
2172	eb.Layout = activator.LayoutInputs{
2173	Journal: func(host string) bool { _, err := os.Stat(provision.LayoutJournalPath(dd, host)); return err == nil },
2174	Peers: reg.List,
2175	}
2176	eb.NestedOK = nestedGate(instancefs.MountNoSymfollow, eb.Control.Ready)
2177	act.Routes = reg
2178	```
2179
2180	and before the activator's `Start()`:
2181
2182	```go
2183	if sw, err := eb.Control.Sweep(ctx); err != nil {
2184	log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
2185	} else if len(sw.Units) > 0 {
2186	log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
2187	}
2188	```
2189
2190	`nestedGate`:
2191
2192	```go
2193	// nestedGate is what an isolated wake requires of the box (design §8.5,
2194	// §8.6): the three nosymfollow mounts, and a control service that reports
2195	// ready. The answer is recomputed per wake (the mounts are cheap to read;
2196	// /ready is one short request), so a box that becomes ready needs no
2197	// restart and a box that stops being ready refuses the next wake.
2198	func nestedGate(mounts func(string) (bool, error), ready func(context.Context) error) func() error {
2199	return func() error {
2200	for _, m := range []string{"/data/instances", "/run/carlos/instances", "/run/carlos/logs"} {
2201	ok, err := mounts(m)
2202	if err != nil {
2203	return fmt.Errorf("%s: %v", m, err)
2204	}
2205	if !ok {
2206	return fmt.Errorf("%s is not a nosymfollow mount", m)
2207	}
2208	}
2209	ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineReady)
2210	defer cancel()
2211	return ready(ctx)
2212	}
2213	}
2214	```
2215
2216	(`reg.List` must match `func() ([]registry.Route, error)`; if the registry's list method has a different name or signature, wrap it.)
2217
2218	- [ ] **Step 4: Full gate**
2219
2220	```bash
2221	export GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off TMPDIR=/tmp/g
2222	go build ./... && go vet ./... && gofmt -l . && go test ./... -count=1
2223	GOOS=darwin go build ./... && GOOS=windows go build ./... && GOARCH=arm64 go build ./...
2224	```
2225
2226	Expected: green.
2227
2228	- [ ] **Step 5: Commit**
2229
2230	```bash
2231	git add internal/appconfig internal/schedule internal/host internal/ops
2232	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)"
2233	```
2234
2235	---
2236
2237	## Self-review
2238
2239	**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.
2240
2241	**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.
2242
2243	**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`.
2244