# `carlos run`, Plan 3: Box Pickup and Output Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Make the box that owns an instance pick up a queued run exactly once, run it through Plan 1's `RunCmd`, turn its two streams into chunks, and write `result.json` only after the last chunk, so a run is never dropped, never replayed and never reported complete before its output is.

**Architecture:** Reconcile hands every in-scope record that carries a run to a `RunSeen` hook; the box's runs worker notes them and, after the pass, acts on each: it picks a queued run for a host whose exec row it owns with a CAS from `queued` to `picked` (box name, agent boot id, the run's deadline), forwards a `cancel` to the run it is executing, and closes out a run it picked under an earlier boot as `interrupted: "the box's agent restarted"`. A boot pass at agent start reads every record once, after deleting every `.carlos-run` directory. One run's goroutine checks the incarnation, launches through `activator.Service.RunCmd` (whose `Prepare` re-reads the record and downloads the files), long-polls the unit, stops it on cancel or timeout, closes the `runs.Chunker`, writes `result.json`, CASes the run to `done`, deletes its directory and releases the activator hold.

**Tech Stack:** Go 1.25, the repo's `provision` (`SetRecordRun`, the run transitions, `ListAllRecords`, `ReconcileConfig`), `runs`, `activator` (`RunCmd`, `CmdHandle`, the errors), `instancefs`, `hostclaim.Incarnation`, `release.Store`.

**Spec:** `docs/superpowers/specs/2026-10-08-carlos-run-design.md` rev 4: §5.2, §5.3 (the box half), §5.4 (the result's reading), §5.5, §5.7 (the box half), §5.8 "The box sweep", §9 "Pickup" and "Output"; dispositions codex 3, 4, 5, 6, R1-3, R2-2, R2-6. Index: `2026-10-08-carlos-run-0-index.md` (item 2: the boot pass). Copy: `2026-10-08-carlos-run-copy.md`.

**Build target:** `main` with Plans 1 and 2 merged.

## Global Constraints

- Every user-facing string (CLI output, errors, console messages, docs prose) is copy-reviewed by Paul before it is written into code. The deck is `docs/superpowers/plans/2026-10-08-carlos-run-copy.md` and its gate is `approved_run_copy_test.go`. Implementation tasks use only strings from the deck; a task that writes one adds its id to `runCopyBuilt` in the same commit. Copy voice: no em dashes; full stops and new sentences instead. (This plan writes no new string: the box writes reason constants that Plans 1 and 2 already carry.)
- CI cross-compiles darwin and windows: linux-only syscalls/units code needs `//go:build linux` plus a `!linux` stub.
- Local test commands: run targeted tests (`go test ./internal/<pkg>/ -run TestX`), not the full suite in a loop. Inside the sandbox the default Go build cache may be read-only; if so use `GOCACHE=$TMPDIR/gocache` re-exported per command. Socket tests need a short TMPDIR (`TMPDIR=/tmp/g`). The local gate may need `GOFLAGS=-buildvcs=false`.
- Heavy loops run under `nice`; no -race full suite until the end of a plan.
- Commits end with `Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>`.
- Only the box whose CAS from `queued` to `picked` lands launches; a box never acts on a queued run past its pickup deadline (that run is the expiry's), on a run for a host whose exec row it does not own, or on a run another box picked.
- A picked run is never executed again: a run this box picked under an earlier boot id ends `interrupted` (spec §5.2).
- `result.json` is written only after every chunk is durably written, and names the last seq (spec §5.5). Chunks: `runs/<id>/out/<seq>-<o|e>`, seq shared by both streams in read order, a chunk every second or every 256 KB, nothing past 64 MB in total (`truncated`; the command keeps running).
- The files land in `provision.InstanceDir(dataDir, host)/.carlos-run/<id>/` (codex 6), the directory 0700, each file 0600 created `O_EXCL` through `instancefs`, no symlink followed, each checked against `request.json`'s size and sha256, the names checked again (spec §7).
- Gate before the plan's last commit: `go build ./...`, `go vet ./...`, `gofmt -l .` clean, `make test`, and `GOOS=darwin go build ./...`, `GOOS=windows go build ./...`, `GOOS=linux GOARCH=arm64 go build ./...`.

## Review Focus

1. A run queued while the box was restarting is picked up by the boot pass at agent start, not a minute later when the first reconcile pass runs, because the CLI expires it at 60 seconds. Test in Task 4 (`TestBootPassPicksARunQueuedBeforeStart`).
2. A run cancelled after the box's pickup CAS but before the unit starts (the files are downloading) must end `cancelled` with no unit started, because `Prepare` re-reads the record right before launch. Test in Task 5 (`TestRunCancelledBeforeLaunchNeverStarts`).
3. An S3 put that fails twice and then succeeds must write the chunk once, under its own seq, with no gap the follower would wait on. Test in Task 1 (`TestChunkerRetriesAFailedPut`).
4. A command that interleaves stdout and stderr byte by byte must come back to the follower in the order the agent read it, even though every switch cuts a chunk. Test in Task 1 (`TestChunkerKeepsReadOrderAcrossStreams`).
5. A tenant that planted a file named like an upload in `.carlos-run/<id>/` before the box staged the run must make the staging fail, never be overwritten or followed. Test in Task 2 (`TestStageFilesRefusesAPlantedFile`).

---

## File structure

| path | responsibility |
|---|---|
| `internal/runs/chunker.go` | `Chunker`: the two streams into ordered, capped, durable chunks |
| `internal/runs/stage.go` | `StageFiles`, `RemoveRunDir`, `SweepRunDirs` |
| `internal/provision/reconcile.go` | `ReconcileConfig.RunSeen` and its one call |
| `internal/host/runs.go` | the runs worker: arming, `note`, `pass`, `bootPass`, `consider`, pickup, agent-restart recovery |
| `internal/host/runexec.go` | one run's life: incarnation, launch, poll, cancel, timeout, result, cleanup |
| `internal/host/agent.go` | arming, the reconcile hook, the passes |

---

### Task 1: `runs.Chunker`

**Files:**
- Create: `internal/runs/chunker.go`, `internal/runs/chunker_test.go`

**Interfaces:**
- Consumes: `runs.ChunkKey`, `runs.ChunkBytes`, `runs.ChunkEvery`, `runs.MaxOutputBytes`, `release.Store`, `release.ErrCASConflict`.
- Produces: `func NewChunker(ctx context.Context, s release.Store, acct, app, id string) *Chunker`; `func (c *Chunker) Stream(stream string) io.Writer` (`runs.StreamOut` or `runs.StreamErr`; writes never fail); `func (c *Chunker) Close() (lastSeq int64, truncated bool, err error)`. Unexported `newChunker(ctx, s, acct, app, id string, every time.Duration, max int64)` for tests.

- [ ] **Step 1: Write the failing tests**

`internal/runs/chunker_test.go`:

```go
package runs

import (
	"context"
	"errors"
	"io"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

const tid = "run_abcdefghijklmnopqrstuvwxyz"

func chunksOf(t *testing.T, s release.Store) []string {
	t.Helper()
	keys, err := ListChunks(context.Background(), s, "bab", "titogo", tid, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	var out []string
	for _, k := range keys {
		raw, _, err := s.Get(context.Background(), k)
		if err != nil {
			t.Fatal(err)
		}
		_, stream, _ := ParseChunkKey(k, OutPrefix("bab", "titogo", tid))
		out = append(out, stream+":"+string(raw))
	}
	return out
}

func TestChunkerKeepsReadOrderAcrossStreams(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
	io.WriteString(c.Stream(StreamOut), "a")
	io.WriteString(c.Stream(StreamErr), "b")
	io.WriteString(c.Stream(StreamOut), "c")
	io.WriteString(c.Stream(StreamOut), "d")
	last, truncated, err := c.Close()
	if err != nil || truncated || last != 3 {
		t.Fatalf("Close = %d %v %v", last, truncated, err)
	}
	if got := strings.Join(chunksOf(t, s), " "); got != "o:a e:b o:cd" {
		t.Fatalf("chunks %q", got)
	}
}

func TestChunkerFlushesOnTheTickAndAtTheSizeLimit(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	c := newChunker(context.Background(), s, "bab", "titogo", tid, 20*time.Millisecond, MaxOutputBytes)
	io.WriteString(c.Stream(StreamOut), "tick")
	deadline := time.Now().Add(2 * time.Second)
	for len(chunksOf(t, s)) == 0 && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	if got := chunksOf(t, s); len(got) != 1 || got[0] != "o:tick" {
		t.Fatalf("after a tick: %q", got)
	}
	c2 := newChunker(context.Background(), s, "bab", "titogo", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", time.Hour, MaxOutputBytes)
	io.WriteString(c2.Stream(StreamOut), strings.Repeat("x", ChunkBytes+1))
	keys, _ := ListChunks(context.Background(), s, "bab", "titogo", "run_bbbbbbbbbbbbbbbbbbbbbbbbbb", 0, 0)
	if len(keys) != 1 {
		t.Fatalf("a full chunk must flush at once: %q", keys)
	}
	if last, _, _ := c2.Close(); last != 2 {
		t.Fatalf("last %d, want 2 (one full chunk, one byte)", last)
	}
	c.Close()
}

func TestChunkerStopsAtTheCapAndSaysSo(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, 10)
	n, err := io.WriteString(c.Stream(StreamOut), "0123456789abcdef")
	if n != 16 || err != nil {
		t.Fatalf("a write past the cap must not fail the stream: %d %v", n, err)
	}
	io.WriteString(c.Stream(StreamErr), "more")
	_, truncated, err := c.Close()
	if !truncated || err != nil {
		t.Fatalf("truncated %v err %v", truncated, err)
	}
	if got := strings.Join(chunksOf(t, s), ""); got != "o:0123456789" {
		t.Fatalf("chunks %q", got)
	}
}

// failingStore fails the first n Puts.
type failingStore struct {
	release.Store
	mu   sync.Mutex
	fail int
}

func (f *failingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	f.mu.Lock()
	if f.fail > 0 {
		f.fail--
		f.mu.Unlock()
		return "", errors.New("slow down")
	}
	f.mu.Unlock()
	return f.Store.Put(ctx, key, body, ifMatch)
}

func TestChunkerRetriesAFailedPut(t *testing.T) {
	s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 2}
	c := newChunker(context.Background(), s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
	io.WriteString(c.Stream(StreamOut), "once")
	last, _, err := c.Close()
	if err != nil || last != 1 {
		t.Fatalf("Close = %d %v", last, err)
	}
	if got := chunksOf(t, s.Store); len(got) != 1 || got[0] != "o:once" {
		t.Fatalf("chunks %q", got)
	}
}

func TestChunkerGivesUpWhenItsContextEnds(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	s := &failingStore{Store: release.NewDirStore(t.TempDir()), fail: 1 << 30}
	c := newChunker(ctx, s, "bab", "titogo", tid, time.Hour, MaxOutputBytes)
	go func() { time.Sleep(50 * time.Millisecond); cancel() }()
	io.WriteString(c.Stream(StreamOut), strings.Repeat("x", ChunkBytes))
	_, truncated, err := c.Close()
	if err == nil || !truncated {
		t.Fatalf("a chunk that never lands: truncated %v err %v", truncated, err)
	}
}
```

- [ ] **Step 2: Run to verify they fail**

Run: `nice -n 10 go test ./internal/runs/ -run TestChunker -count=1`
Expected: FAIL, `undefined: newChunker`.

- [ ] **Step 3: Implement**

`internal/runs/chunker.go`:

```go
package runs

import (
	"context"
	"errors"
	"io"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

// Chunker turns a run's two streams into chunks (run spec §5.5):
// out/<seq>-<o|e>, one seq for both streams in the order the bytes were
// read. A chunk holds one stream's bytes: a write on the other stream cuts
// it, so the follower interleaves exactly as the agent read. A chunk is
// written every second, or as soon as it holds 256 KB. Past 64 MB in total
// nothing more is written and Close says truncated; writes never fail, so
// the command is never blocked by a refused pipe. Each chunk is a
// create-only Put retried until it lands or the context ends, under the
// lock, so seqs are written in order and a slow bucket slows the copy.
type Chunker struct {
	ctx           context.Context
	s             release.Store
	acct, app, id string
	every         time.Duration
	max           int64

	mu        sync.Mutex
	stream    string
	pending   []byte
	seq       int64
	total     int64
	truncated bool
	err       error

	stop      chan struct{}
	done      chan struct{}
	closeOnce sync.Once
}

// NewChunker starts a chunker for run id with the spec's cadence and cap.
func NewChunker(ctx context.Context, s release.Store, acct, app, id string) *Chunker {
	return newChunker(ctx, s, acct, app, id, ChunkEvery, MaxOutputBytes)
}

func newChunker(ctx context.Context, s release.Store, acct, app, id string, every time.Duration, max int64) *Chunker {
	c := &Chunker{ctx: ctx, s: s, acct: acct, app: app, id: id, every: every, max: max,
		stop: make(chan struct{}), done: make(chan struct{})}
	go c.tick()
	return c
}

func (c *Chunker) tick() {
	defer close(c.done)
	t := time.NewTicker(c.every)
	defer t.Stop()
	for {
		select {
		case <-c.stop:
			return
		case <-t.C:
			c.mu.Lock()
			c.flushLocked()
			c.mu.Unlock()
		}
	}
}

type chunkStream struct {
	c      *Chunker
	stream string
}

func (w chunkStream) Write(p []byte) (int, error) {
	w.c.write(w.stream, p)
	return len(p), nil
}

// Stream is the writer for one stream, StreamOut or StreamErr.
func (c *Chunker) Stream(stream string) io.Writer { return chunkStream{c: c, stream: stream} }

func (c *Chunker) write(stream string, p []byte) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err != nil {
		c.truncated = c.truncated || len(p) > 0
		return
	}
	if room := c.max - c.total; int64(len(p)) > room {
		p = p[:room]
		c.truncated = true
	}
	for len(p) > 0 {
		if c.stream != stream && len(c.pending) > 0 {
			c.flushLocked()
			if c.err != nil {
				return
			}
		}
		c.stream = stream
		n := min(len(p), ChunkBytes-len(c.pending))
		c.pending = append(c.pending, p[:n]...)
		c.total += int64(n)
		p = p[n:]
		if len(c.pending) >= ChunkBytes {
			c.flushLocked()
		}
	}
}

// flushLocked writes the pending chunk under the next seq. A Put that
// answers ErrCASConflict was written by an earlier attempt whose answer was
// lost; the seq is spent either way.
func (c *Chunker) flushLocked() {
	if len(c.pending) == 0 || c.err != nil {
		return
	}
	seq := c.seq + 1
	key := ChunkKey(c.acct, c.app, c.id, seq, c.stream)
	backoff := 100 * time.Millisecond
	for {
		_, err := c.s.Put(c.ctx, key, c.pending, "")
		if err == nil || errors.Is(err, release.ErrCASConflict) {
			break
		}
		select {
		case <-c.ctx.Done():
			c.err, c.truncated, c.pending = err, true, nil
			return
		case <-time.After(backoff):
		}
		backoff = min(backoff*2, 5*time.Second)
	}
	c.seq, c.pending = seq, nil
}

// Close stops the ticker and writes the last chunk. Call it after both
// streams reached their end (CmdHandle.Stop drains them). It returns the
// last seq written (0: no output), whether output was cut off, and the
// write failure that ended chunking, if one did.
func (c *Chunker) Close() (int64, bool, error) {
	c.closeOnce.Do(func() {
		close(c.stop)
		<-c.done
	})
	c.mu.Lock()
	defer c.mu.Unlock()
	c.flushLocked()
	return c.seq, c.truncated, c.err
}
```

- [ ] **Step 4: Run to verify they pass**

Run: `nice -n 10 go test ./internal/runs/ -count=1`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add internal/runs/chunker.go internal/runs/chunker_test.go
git commit -m "runs: the chunker: ordered, capped, durable output chunks (run spec §5.5, codex 5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 2: Staging the files, and the box sweep

**Files:**
- Create: `internal/runs/stage.go`, `internal/runs/stage_linux_test.go`

**Interfaces:**
- Consumes: `instancefs.EnsureDirBeneath(dir, name string, perm os.FileMode) (DirStat, error)`, `instancefs.OpenBeneath(dir, name string, flag int, perm os.FileMode) (*os.File, error)`, `instancefs.RemoveAllBeneath(dir, name string) error` (Plan 1 Task 5), `runs.DirName`, `runs.InputKey`, `runs.PlainName`, `runs.Request`.
- Produces: `func StageFiles(ctx context.Context, s release.Store, instanceDir string, req Request) (string, error)` (returns `<instanceDir>/.carlos-run/<id>`, always made, even with no files); `func RemoveRunDir(instanceDir, id string) error`; `func SweepRunDirs(dataDir string) error` (every `<dataDir>/instances/*/.carlos-run`).

- [ ] **Step 1: Write the failing tests**

`internal/runs/stage_linux_test.go`:

```go
//go:build linux

package runs

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"os"
	"path/filepath"
	"testing"

	"github.com/carlosframework/platform/internal/instancefs"
	"github.com/carlosframework/platform/internal/release"
)

func stagedRequest(t *testing.T, s release.Store, files map[string]string) Request {
	t.Helper()
	req := Request{ID: tid, Account: "bab", App: "titogo"}
	for name, body := range files {
		sum := sha256.Sum256([]byte(body))
		req.Files = append(req.Files, File{Name: name, Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])})
		if _, err := s.Put(context.Background(), InputKey("bab", "titogo", tid, name), []byte(body), ""); err != nil {
			t.Fatal(err)
		}
	}
	return req
}

func TestStageFilesWritesTheUploads(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	inst := t.TempDir()
	req := stagedRequest(t, s, map[string]string{"a.csv": "1,2", "b.json": "{}"})
	dir, err := StageFiles(context.Background(), s, inst, req)
	if err != nil {
		t.Fatal(err)
	}
	if dir != filepath.Join(inst, DirName, tid) {
		t.Fatalf("dir %q", dir)
	}
	st, err := os.Stat(dir)
	if err != nil || st.Mode().Perm() != 0o700 {
		t.Fatalf("run dir mode %v, %v", st.Mode(), err)
	}
	for name, want := range map[string]string{"a.csv": "1,2", "b.json": "{}"} {
		raw, err := os.ReadFile(filepath.Join(dir, name))
		fi, _ := os.Stat(filepath.Join(dir, name))
		if err != nil || string(raw) != want || fi.Mode().Perm() != 0o600 {
			t.Errorf("%s: %q %v mode %v", name, raw, err, fi.Mode())
		}
	}
}

func TestStageFilesMakesTheDirectoryWithNoFiles(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	inst := t.TempDir()
	dir, err := StageFiles(context.Background(), s, inst, Request{ID: tid, Account: "bab", App: "titogo"})
	if err != nil {
		t.Fatal(err)
	}
	if st, err := os.Stat(dir); err != nil || !st.IsDir() {
		t.Fatalf("CARLOS_RUN_DIR must always exist: %v", err)
	}
}

func TestStageFilesChecksSizeAndHash(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	inst := t.TempDir()
	req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
	req.Files[0].SHA256 = hex.EncodeToString(make([]byte, 32))
	if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
		t.Fatal("a hash mismatch was staged")
	}
	req = stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
	req.Files[0].Size = 2
	if err := RemoveRunDir(inst, tid); err != nil {
		t.Fatal(err)
	}
	if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
		t.Fatal("a size mismatch was staged")
	}
}

func TestStageFilesRefusesAPlantedFile(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	inst := t.TempDir()
	req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
	if err := os.MkdirAll(filepath.Join(inst, DirName, tid), 0o700); err != nil {
		t.Fatal(err)
	}
	outside := filepath.Join(t.TempDir(), "victim")
	if err := os.WriteFile(outside, []byte("keep"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(outside, filepath.Join(inst, DirName, tid, "a.csv")); err != nil {
		t.Fatal(err)
	}
	if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
		t.Fatal("staged over a planted symlink")
	}
	if raw, _ := os.ReadFile(outside); string(raw) != "keep" {
		t.Fatal("the planted symlink's target was written")
	}
}

func TestStageFilesRefusesASymlinkedRunRoot(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	inst, outside := t.TempDir(), t.TempDir()
	if err := os.Symlink(outside, filepath.Join(inst, DirName)); err != nil {
		t.Fatal(err)
	}
	_, err := StageFiles(context.Background(), s, inst, Request{ID: tid, Account: "bab", App: "titogo"})
	if !errors.Is(err, instancefs.ErrSymlink) {
		t.Fatalf("err %v, want ErrSymlink", err)
	}
}

func TestStageFilesRefusesANameThatIsNotPlain(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	req := Request{ID: tid, Account: "bab", App: "titogo", Files: []File{{Name: "../x", Size: 1}}}
	if _, err := StageFiles(context.Background(), s, t.TempDir(), req); err == nil {
		t.Fatal("a path name was staged (spec §7: checked again by the agent)")
	}
}

func TestSweepRunDirsRemovesEveryRunDirectory(t *testing.T) {
	data := t.TempDir()
	for _, host := range []string{"a.example.com", "b.example.com"} {
		if err := os.MkdirAll(filepath.Join(data, "instances", host, DirName, tid, "sub"), 0o700); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(filepath.Join(data, "instances", host, "app.db"), []byte("db"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	if err := SweepRunDirs(data); err != nil {
		t.Fatal(err)
	}
	for _, host := range []string{"a.example.com", "b.example.com"} {
		if _, err := os.Lstat(filepath.Join(data, "instances", host, DirName)); !os.IsNotExist(err) {
			t.Errorf("%s: .carlos-run survived", host)
		}
		if _, err := os.Stat(filepath.Join(data, "instances", host, "app.db")); err != nil {
			t.Errorf("%s: the sweep touched the database", host)
		}
	}
}
```

- [ ] **Step 2: Run to verify they fail**

Run: `nice -n 10 go test ./internal/runs/ -run 'TestStageFiles|TestSweepRunDirs' -count=1`
Expected: FAIL, `undefined: StageFiles`.

- [ ] **Step 3: Implement**

`internal/runs/stage.go` (no build tag: `instancefs` has `!linux` stubs, so this compiles everywhere and answers `errors.ErrUnsupported` off linux):

```go
package runs

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"

	"github.com/carlosframework/platform/internal/instancefs"
	"github.com/carlosframework/platform/internal/release"
)

// StageFiles makes <instanceDir>/.carlos-run/<id>/ and downloads the run's
// files into it (run spec §5.2 step 3). Both directories are created
// beneath the instance's own directory, 0700, without following a
// symlink; each file is created O_EXCL 0600 through instancefs, so a name
// the tenant planted first is a refusal, never a write through it; each is
// checked against request.json's size and sha256 and its name against
// PlainName. The directory exists even when there are no files: it is
// CARLOS_RUN_DIR.
func StageFiles(ctx context.Context, s release.Store, instanceDir string, req Request) (string, error) {
	if !ValidID(req.ID) {
		return "", fmt.Errorf("runs: %q is not a run id", req.ID)
	}
	root := filepath.Join(instanceDir, DirName)
	if _, err := instancefs.EnsureDirBeneath(instanceDir, DirName, 0o700); err != nil {
		return "", err
	}
	if _, err := instancefs.EnsureDirBeneath(root, req.ID, 0o700); err != nil {
		return "", err
	}
	dir := filepath.Join(root, req.ID)
	for _, f := range req.Files {
		if !PlainName(f.Name) {
			return "", fmt.Errorf("runs: %q is not a plain file name", f.Name)
		}
		raw, _, err := s.Get(ctx, InputKey(req.Account, req.App, req.ID, f.Name))
		if err != nil {
			return "", fmt.Errorf("runs: read %s: %w", f.Name, err)
		}
		sum := sha256.Sum256(raw)
		if int64(len(raw)) != f.Size || hex.EncodeToString(sum[:]) != f.SHA256 {
			return "", fmt.Errorf("runs: %s does not match its size and sha256", f.Name)
		}
		out, err := instancefs.OpenBeneath(dir, f.Name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
		if err != nil {
			return "", err
		}
		_, werr := io.Copy(out, bytes.NewReader(raw))
		serr := out.Sync()
		cerr := out.Close()
		if err := errors.Join(werr, serr, cerr); err != nil {
			return "", err
		}
	}
	return dir, nil
}

// RemoveRunDir deletes one run's directory, following no symlink.
func RemoveRunDir(instanceDir, id string) error {
	if !ValidID(id) {
		return fmt.Errorf("runs: %q is not a run id", id)
	}
	return instancefs.RemoveAllBeneath(filepath.Join(instanceDir, DirName), id)
}

// SweepRunDirs deletes every instance's .carlos-run directory (run spec
// §5.8, the box sweep). It runs at agent start, before any run is picked:
// the edge's stop took every cmd unit with it (BindsTo=carlos-edge.service),
// so no run on this box is open.
func SweepRunDirs(dataDir string) error {
	base := filepath.Join(dataDir, "instances")
	ents, err := os.ReadDir(base)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return err
	}
	var errs []error
	for _, e := range ents {
		if !e.IsDir() {
			continue
		}
		if err := instancefs.RemoveAllBeneath(filepath.Join(base, e.Name()), DirName); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}
```

- [ ] **Step 4: Run to verify they pass**

Run: `nice -n 10 go test ./internal/runs/ -count=1 && GOOS=darwin go build ./internal/runs/ && GOOS=windows go build ./internal/runs/`
Expected: PASS; both cross builds succeed.

- [ ] **Step 5: Commit**

```bash
git add internal/runs/stage.go internal/runs/stage_linux_test.go
git commit -m "runs: stage a run's files beneath its instance, and the boot sweep (run spec §5.2, §5.8, codex 6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 3: Reconcile hands runs to the box

**Files:**
- Modify: `internal/provision/reconcile.go:105-133` (`ReconcileConfig.RunSeen`), `:1072-1075` (the call)
- Test: `internal/provision/reconcile_test.go` (append)

**Interfaces:**
- Consumes: Reconcile's in-scope `recs` and the per-record loop at `reconcile.go:1072`.
- Produces: `ReconcileConfig.RunSeen func(rec Record)`: called once per pass for every in-scope record of an untrashed app that carries a run, before the pass acts on the record. The runs worker only notes it there.

- [ ] **Step 1: Write the failing test**

Append to `reconcile_test.go`:

```go
// Reconcile's own reads carry the runs to the box (run spec §5.2): every
// in-scope record with a run, including one whose row this very pass
// mints, and none out of scope.
func TestReconcileHandsRunsToRunSeen(t *testing.T) {
	s, sqid := setup(t)
	withRun := acmeRecord(sqid)
	withRun.ActiveRun = &RunState{ID: "run_abcdefghijklmnopqrstuvwxyz", State: RunQueued, Deadline: time.Now().Add(time.Minute)}
	plain := acmeRecord(sqid)
	plain.Host, plain.Slug = "plain.go.tito.io", "plain"
	away := acmeRecord(sqid)
	away.Host, away.Slug, away.Region = "away.go.tito.io", "away", "uk"
	away.ActiveRun = withRun.ActiveRun
	for _, rec := range []Record{withRun, plain, away} {
		if err := CreateRecord(ctx, s, rec); err != nil {
			t.Fatal(err)
		}
	}
	var seen []string
	cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", RunSeen: func(rec Record) {
		seen = append(seen, rec.Host+":"+rec.ActiveRun.ID)
	}}
	if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
		t.Fatal(err)
	}
	if len(seen) != 1 || seen[0] != "acme.go.tito.io:run_abcdefghijklmnopqrstuvwxyz" {
		t.Fatalf("RunSeen saw %q", seen)
	}
}
```

- [ ] **Step 2: Run to verify it fails**

Run: `nice -n 10 go test ./internal/provision/ -run TestReconcileHandsRunsToRunSeen -count=1`
Expected: FAIL, `unknown field RunSeen`.

- [ ] **Step 3: Implement**

In `ReconcileConfig`, after `HoldWritten`:

```go
	// RunSeen is called with every in-scope record of an untrashed app that
	// carries a run, once per pass, before the pass acts on the record (run
	// spec §5.2: pickup rides reconcile's own record reads, so an idle box
	// pays nothing extra). It must not block: the box's runs worker only
	// notes the record and acts after the pass, by which time a row this
	// pass minted is in the registry. Nil: no one is told.
	RunSeen func(rec Record)
```

At the top of the per-record loop (`reconcile.go:1072`), right after the `trashedFor` check:

```go
		if cfg.RunSeen != nil && rec.ActiveRun != nil {
			cfg.RunSeen(rec)
		}
```

- [ ] **Step 4: Run to verify it passes**

Run: `nice -n 10 go test ./internal/provision/ -count=1 && nice -n 10 go test . -run TestMoves -count=1`
Expected: PASS (`reconcile.go` is in `movesScanFiles`; the change adds no string).

- [ ] **Step 5: Commit**

```bash
git add internal/provision/reconcile.go internal/provision/reconcile_test.go
git commit -m "provision: reconcile hands records with a run to RunSeen (run spec §5.2, codex 3)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 4: The runs worker: pickup, agent-restart recovery, the boot pass

**Files:**
- Create: `internal/host/runs.go`, `internal/host/runs_test.go`

**Interfaces:**
- Consumes: `provision.SetRecordRun`, `provision.RunPick`, `provision.RunFinish`, `provision.ListAllRecords`, `provision.InstanceDir`, `runs.ReadRequest`, `runs.WriteResult`, `runs.LastChunkSeq`, `runs.RemoveRunDir`, `runs.SweepRunDirs`, `runs.InterruptedAgentRestart`, `activator.ActivatorManaged(rt registry.Route) bool`, `activator.CmdSpec`, `activator.CmdHandle`, `activator.CmdRunDirName`.
- Produces:
  - `type runLauncher interface { RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) }` (`*activator.Service` satisfies it).
  - `type runsWorker struct` with `store release.Store`, `routes func(host string) (*registry.Route, error)`, `launch runLauncher`, `dataDir, box, boot string`, `now func() time.Time`, `exec func(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle)`, `start func(func())` (a goroutine in the agent, inline in tests), and its own lock, `noted` and `active` maps.
  - `func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker` (sets `exec` to Task 5's `w.execute`).
  - `func (w *runsWorker) note(rec provision.Record)`, `func (w *runsWorker) pass(ctx context.Context)`, `func (w *runsWorker) bootPass(ctx context.Context)`, `func (w *runsWorker) consider(ctx context.Context, rec provision.Record)`.
  - `type runHandle struct` with `requestCancel()` and a `cancel <-chan struct{}`; `func newBootID() string`.

- [ ] **Step 1: Write the failing tests**

`internal/host/runs_test.go`:

```go
package host

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
)

const runHost = "acme.go.tito.io"

func runRecord() provision.Record {
	return provision.Record{Schema: 1, Host: runHost, Slug: "acme", App: "titogo", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
}

func ownedRoute(host string) func(string) (*registry.Route, error) {
	return func(h string) (*registry.Route, error) {
		if h != host {
			return nil, nil
		}
		return &registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: "/run/x.sock", Provisioned: true, App: "titogo", Account: "bab"}, nil
	}
}

func noRoutes(string) (*registry.Route, error) { return nil, nil }

// queueRun writes a record whose run is queued, with its request.json.
func queueRun(t *testing.T, s release.Store, deadline time.Time) (provision.Record, string) {
	t.Helper()
	rec := runRecord()
	if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
		t.Fatal(err)
	}
	id := "run_abcdefghijklmnopqrstuvwxyz"
	inc, _ := hostclaimIncarnation(rec)
	req := runs.Request{Schema: 1, ID: id, Account: rec.Account, App: rec.App, Host: rec.Host, Incarnation: inc, Argv: []string{"import"}, TimeoutS: 60, RequestedAt: time.Now()}
	putJSONObject(t, s, runs.RequestKey(rec.Account, rec.App, id), req)
	got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunAdmit(provision.RunAdmission{ID: id, By: "alice@keymail.dev", Deadline: deadline}))
	if err != nil {
		t.Fatal(err)
	}
	return *got, id
}

type execLog struct {
	mu  sync.Mutex
	ids []string
}

func (l *execLog) exec(_ context.Context, _ provision.Record, req runs.Request, _ registry.Route, _ *runHandle) {
	l.mu.Lock()
	l.ids = append(l.ids, req.ID)
	l.mu.Unlock()
}
func (l *execLog) count() int { l.mu.Lock(); defer l.mu.Unlock(); return len(l.ids) }

// testWorker execs inline, so a test counts launches as soon as consider
// returns.
func testWorker(s release.Store, routes func(string) (*registry.Route, error), box, boot string, log *execLog) *runsWorker {
	w := newRunsWorker(s, routes, nil, "/data", box, boot)
	w.exec = log.exec
	w.start = func(f func()) { f() }
	return w
}

func currentRun(t *testing.T, s release.Store) *provision.RunState {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), s, "bab", "titogo", runHost)
	if err != nil {
		t.Fatal(err)
	}
	return rec.ActiveRun
}

func TestRunsWorkerPicksOnlyRunsOnItsOwnRows(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	var other, mine execLog
	testWorker(s, noRoutes, "box-2", "boot-2", &other).consider(context.Background(), rec)
	if other.count() != 0 || currentRun(t, s).State != provision.RunQueued {
		t.Fatal("a box that does not own the row acted on the run")
	}
	testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &mine).consider(context.Background(), rec)
	r := currentRun(t, s)
	if mine.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.BootID != "boot-1" || r.ID != id {
		t.Fatalf("pickup: execs %d, run %+v", mine.count(), r)
	}
	if d := time.Until(r.Deadline); d < 50*time.Second || d > 61*time.Second {
		t.Errorf("the picked deadline is the run's timeout (60 s), got %s", d)
	}
}

func TestRunsWorkerPickupRaceLaunchesOnce(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
	var a, b execLog
	wa := testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &a)
	wb := testWorker(s, ownedRoute(runHost), "box-2", "boot-2", &b)
	var wg sync.WaitGroup
	for _, w := range []*runsWorker{wa, wb} {
		wg.Add(1)
		go func() { defer wg.Done(); w.consider(context.Background(), rec) }()
	}
	wg.Wait()
	if a.count()+b.count() != 1 {
		t.Fatalf("launched %d times, want once", a.count()+b.count())
	}
}

func TestRunsWorkerLeavesARunPastItsPickupDeadline(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(-time.Second))
	var l execLog
	testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
	if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
		t.Fatal("a box picked a run past its pickup deadline: that run is the expiry's")
	}
}

func TestRunsWorkerNeverPicksACancelledOrExpiredRun(t *testing.T) {
	for _, mut := range []func(string) func(*provision.Record) error{provision.RunCancelTransition, provision.RunExpire} {
		s := release.NewDirStore(t.TempDir())
		rec, id := queueRun(t, s, time.Now().Add(time.Minute))
		if _, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, mut(id)); err != nil {
			t.Fatal(err)
		}
		var l execLog
		// rec is the stale read the pass noted, still queued: the CAS refuses.
		testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
		if l.count() != 0 {
			t.Fatal("a cancelled or expired run was launched (codex R2-6)")
		}
	}
}

func TestRunsWorkerClosesOutARunPickedUnderAnEarlierBoot(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-1", "boot-old", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), runs.ChunkKey("bab", "titogo", id, 4, runs.StreamOut), []byte("x"), ""); err != nil {
		t.Fatal(err)
	}
	var l execLog
	testWorker(s, ownedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
	if l.count() != 0 {
		t.Fatal("a run picked before an agent restart was run again")
	}
	res, err := runs.ReadResult(context.Background(), s, "bab", "titogo", id)
	if err != nil || res == nil || res.Interrupted != runs.InterruptedAgentRestart || res.LastSeq != 4 || res.Box != "box-1" {
		t.Fatalf("result %+v, %v", res, err)
	}
	if r := currentRun(t, s); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
}

func TestRunsWorkerForwardsACancelToItsRun(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	_, id := queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &l)
	h := newRunHandle()
	w.active[id] = h
	got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	if got, err = provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunCancelTransition(id)); err != nil {
		t.Fatal(err)
	}
	w.consider(context.Background(), *got)
	select {
	case <-h.cancel:
	default:
		t.Fatal("the cancel did not reach the run")
	}
}

func TestBootPassPicksARunQueuedBeforeStart(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &l)
	w.dataDir = t.TempDir()
	w.bootPass(context.Background())
	if l.count() != 1 {
		t.Fatal("the boot pass did not pick a run queued before agent start")
	}
}

func TestRunsWorkerNotesAndPasses(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
	var l execLog
	w := testWorker(s, ownedRoute(runHost), "box-1", "boot-1", &l)
	w.note(rec)
	w.note(rec)
	w.pass(context.Background())
	w.pass(context.Background())
	if l.count() != 1 {
		t.Fatalf("execs %d, want 1: a noted record is acted on once", l.count())
	}
}

func TestRunDirNamesAgree(t *testing.T) {
	if runs.DirName != activator.CmdRunDirName {
		t.Fatalf("runs.DirName %q, activator.CmdRunDirName %q", runs.DirName, activator.CmdRunDirName)
	}
}
```

`hostclaimIncarnation` and `putJSONObject` are two small test helpers in the same file:

```go
func hostclaimIncarnation(rec provision.Record) (string, error) {
	return hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
}

func putJSONObject(t *testing.T, s release.Store, key string, v any) {
	t.Helper()
	raw, err := json.Marshal(v)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := s.Put(context.Background(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
}
```

(add `encoding/json` and `internal/hostclaim` to the imports).

- [ ] **Step 2: Run to verify they fail**

Run: `nice -n 10 go test ./internal/host/ -run 'TestRunsWorker|TestBootPass|TestRunDirNamesAgree' -count=1`
Expected: FAIL, `undefined: newRunsWorker`.

- [ ] **Step 3: Implement**

`internal/host/runs.go`:

```go
package host

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
)

// runLauncher is what the runs worker needs from the activator.
type runLauncher interface {
	RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error)
}

var _ runLauncher = (*activator.Service)(nil)

// runHandle is the worker's line to one executing run.
type runHandle struct {
	cancel chan struct{}
	once   sync.Once
}

func newRunHandle() *runHandle { return &runHandle{cancel: make(chan struct{})} }

func (h *runHandle) requestCancel() { h.once.Do(func() { close(h.cancel) }) }

// runsWorker is the box's half of carlos run (run spec §5.2). boot is this
// agent process's own id: a run picked under another boot of this box lost
// its unit when the edge stopped (BindsTo=carlos-edge.service).
type runsWorker struct {
	store   release.Store
	routes  func(host string) (*registry.Route, error)
	launch  runLauncher
	dataDir string
	box     string
	boot    string
	now     func() time.Time
	exec    func(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle)
	// start runs a picked run's exec: on its own goroutine in the agent,
	// inline in tests that count launches.
	start func(func())

	mu     sync.Mutex
	noted  map[string]provision.Record // host -> the record reconcile read
	active map[string]*runHandle       // run id -> the run this agent is executing
}

func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker {
	w := &runsWorker{store: store, routes: routes, launch: launch, dataDir: dataDir, box: box, boot: boot,
		now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{},
		start: func(f func()) { go f() }}
	w.exec = w.execute
	return w
}

// newBootID is a fresh id for this agent process.
func newBootID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		return time.Now().UTC().Format("20060102T150405.000000000")
	}
	return hex.EncodeToString(b[:])
}

// armedRuns is the worker the agent armed; nil until then (and on a box
// with no activator, which cannot run anything).
var armedRuns atomic.Pointer[runsWorker]

// note is ReconcileConfig.RunSeen: remember, never act, never block.
func (w *runsWorker) note(rec provision.Record) {
	w.mu.Lock()
	w.noted[rec.Host] = rec
	w.mu.Unlock()
}

// pass acts on every record noted since the last pass.
func (w *runsWorker) pass(ctx context.Context) {
	w.mu.Lock()
	noted := w.noted
	w.noted = map[string]provision.Record{}
	w.mu.Unlock()
	for _, rec := range noted {
		w.consider(ctx, rec)
	}
}

// bootPass runs once at agent start (index item 2: reconcile's first pass
// is a minute away, and the CLI expires a run at 60 s). It deletes every
// .carlos-run directory first (run spec §5.8: no run is open on the box
// yet), then reads every record once and acts on each run.
func (w *runsWorker) bootPass(ctx context.Context) {
	if err := runs.SweepRunDirs(w.dataDir); err != nil {
		log.Printf("agent: runs: boot sweep of .carlos-run: %v", err)
	}
	recs, err := provision.ListAllRecords(ctx, w.store)
	if err != nil {
		log.Printf("agent: runs: boot pass: %v", err)
		return
	}
	for _, rec := range recs {
		if rec.ActiveRun != nil {
			w.consider(ctx, rec)
		}
	}
}

// owned is the row this box runs the record's instance from, when it is
// this box's to act on: provisioned for this record's account and app,
// and one the activator manages.
func (w *runsWorker) owned(rec provision.Record) (registry.Route, bool) {
	rt, err := w.routes(rec.Host)
	if err != nil || rt == nil || !rt.Provisioned || rt.Account != rec.Account || rt.App != rec.App || !activator.ActivatorManaged(*rt) {
		return registry.Route{}, false
	}
	return *rt, true
}

// consider acts on one record's run (run spec §5.2).
func (w *runsWorker) consider(ctx context.Context, rec provision.Record) {
	r := rec.ActiveRun
	if r == nil {
		return
	}
	switch r.State {
	case provision.RunQueued:
		rt, ok := w.owned(rec)
		if !ok || w.now().After(r.Deadline) {
			return
		}
		w.pickup(ctx, rec, rt)
	case provision.RunPicked, provision.RunCancel:
		if r.Box != w.box {
			return
		}
		w.mu.Lock()
		h := w.active[r.ID]
		w.mu.Unlock()
		switch {
		case h != nil && r.State == provision.RunCancel:
			h.requestCancel()
		case h == nil && r.BootID != w.boot:
			w.closeInterrupted(ctx, rec)
		}
	}
}

// pickup takes a queued run with one CAS from queued to picked. Only the
// box whose CAS lands launches; any other outcome (another box, a cancel,
// an expiry) leaves the run alone.
func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt registry.Route) {
	id := rec.ActiveRun.ID
	req, err := runs.ReadRequest(ctx, w.store, rec.Account, rec.App, id)
	if err != nil {
		// The console writes request.json before the record names the run;
		// without it there is nothing to run, and the expiry ends it.
		log.Printf("agent: runs: %s %s: request: %v", rec.Host, id, err)
		return
	}
	h := newRunHandle()
	w.mu.Lock()
	if w.active[id] != nil {
		w.mu.Unlock()
		return
	}
	w.active[id] = h
	w.mu.Unlock()
	got, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host,
		provision.RunPick(id, w.box, w.boot, w.now().Add(req.Timeout())))
	if err != nil {
		w.forget(id)
		if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err)
		}
		return
	}
	log.Printf("agent: runs: %s: picked run %s", rec.Host, id)
	w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })
}

func (w *runsWorker) forget(id string) {
	w.mu.Lock()
	delete(w.active, id)
	w.mu.Unlock()
}

// closeInterrupted ends a run this box picked under an earlier boot (run
// spec §5.2): its unit went with the edge, so it is never executed again.
// The result names the last chunk the old boot wrote.
func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record) {
	id := rec.ActiveRun.ID
	last, err := runs.LastChunkSeq(ctx, w.store, rec.Account, rec.App, id)
	if err != nil {
		log.Printf("agent: runs: %s %s: list chunks: %v", rec.Host, id, err)
		return
	}
	res := runs.Result{Interrupted: runs.InterruptedAgentRestart, Box: w.box, Ended: w.now().UTC(), LastSeq: last}
	if err := runs.WriteResult(ctx, w.store, rec.Account, rec.App, id, res); err != nil {
		log.Printf("agent: runs: %s %s: result: %v", rec.Host, id, err)
		return
	}
	if _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, provision.RunFinish(id)); err != nil &&
		!errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
		log.Printf("agent: runs: %s %s: done: %v", rec.Host, id, err)
	}
	if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), id); err != nil && !errors.Is(err, os.ErrNotExist) {
		log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, id, err)
	}
	log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id)
}
```

Task 5 adds `execute`; until then add a temporary body at the bottom of `runs.go` so the package builds, and Task 5 replaces it:

```go
// execute runs one picked run; Task 5 gives it its body.
func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
	defer w.forget(req.ID)
}
```

- [ ] **Step 4: Run to verify they pass**

Run: `nice -n 10 go test ./internal/host/ -run 'TestRunsWorker|TestBootPass|TestRunDirNamesAgree' -count=1`
Expected: PASS.

- [ ] **Step 5: Commit**

```bash
git add internal/host/runs.go internal/host/runs_test.go
git commit -m "host: the runs worker: pickup by CAS, agent-restart close-out, the boot pass (run spec §5.2, codex 3, R2-6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 5: One run's life

**Files:**
- Create: `internal/host/runexec.go`, `internal/host/runexec_test.go`
- Modify: `internal/host/runs.go` (remove Task 4's temporary `execute`)

**Interfaces:**
- Consumes: `activator.CmdSpec`, `activator.CmdHandle`, `activator.ErrCmdClaim`, `activator.ErrCmdNotNested`, `*activator.CmdWakeError`, `*activator.CmdStoppedError`, `instancectl.CmdStatus`, `instancectl.CmdDone`, `instancectl.CmdGone`, `runs.NewChunker`, `runs.StageFiles`, `runs.RemoveRunDir`, `runs.WriteResult`, the `Refused*` and `Interrupted*` constants, `hostclaim.Incarnation`, `provision.GetRecord`, `provision.RunFinish`.
- Produces: `func (w *runsWorker) execute(ctx, rec, req, rt, h)`; `func runResultForLaunch(err error) runs.Result`; `func exitOf(st instancectl.CmdStatus) (exit int, signal string)`; `runPollRetry = time.Second`, `runRecheckEvery = 30 * time.Second` (vars, for tests).

The run's order, which the tests pin:

1. The incarnation in `request.json` matches the record the pass read (codex 3, R1-3), or `refused: "incarnation"` and no launch.
2. `RunCmd` with a `Prepare` that re-reads the record right before launch: the run must still be this box's under this boot and not `cancel` (else `cancelled`, no unit, codex R2-6), the record must not be an uncommitted import (else `refused: "import"`, codex R2-2), the incarnation must still match; then the files are staged.
3. Launch errors map to results: claim, flat, wake (with the member-safe detail), a stop before launch (`interrupted: <reason>`), a start failure (`interrupted: InterruptedStartFailed`).
4. Poll until `done` or `gone`, a `cancel`, or the run's timeout; on `cancel` or timeout the unit is stopped.
5. Always `handle.Stop` (the unit stopped and settled, both streams drained), then `Chunker.Close`, then `result.json` naming the last seq, then the CAS to `done`, then the run directory deleted, then `handle.Release`.

- [ ] **Step 1: Write the failing tests**

`internal/host/runexec_test.go`:

```go
package host

import (
	"context"
	"errors"
	"io"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
)

type fakeHandle struct {
	mu          sync.Mutex
	polls       chan instancectl.CmdStatus
	stops       int
	released    bool
	interrupted string
	afterStop   *instancectl.CmdStatus
}

func (h *fakeHandle) Invocation() string { return "inv1" }
func (h *fakeHandle) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
	h.mu.Lock()
	if h.stops > 0 {
		st := instancectl.CmdStatus{State: instancectl.CmdGone}
		if h.afterStop != nil {
			st = *h.afterStop
		}
		h.mu.Unlock()
		return st, nil
	}
	h.mu.Unlock()
	select {
	case st := <-h.polls:
		return st, nil
	case <-ctx.Done():
		return instancectl.CmdStatus{}, ctx.Err()
	}
}
func (h *fakeHandle) Interrupted() string { h.mu.Lock(); defer h.mu.Unlock(); return h.interrupted }
func (h *fakeHandle) Stop(context.Context) error {
	h.mu.Lock()
	h.stops++
	h.mu.Unlock()
	return nil
}
func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }

type fakeLauncher struct {
	mu     sync.Mutex
	calls  int
	err    error
	handle *fakeHandle
	out    string
	errOut string
}

func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
	l.mu.Lock()
	l.calls++
	l.mu.Unlock()
	if l.err != nil {
		return nil, l.err
	}
	if spec.Prepare != nil {
		if err := spec.Prepare(ctx); err != nil {
			return nil, err
		}
	}
	io.WriteString(spec.Stdout, l.out)
	io.WriteString(spec.Stderr, l.errOut)
	return l.handle, nil
}

// pickedRun queues and picks a run for box-1/boot-1 and returns what
// execute is called with.
func pickedRun(t *testing.T, s release.Store, timeoutS int64) (provision.Record, runs.Request) {
	t.Helper()
	rec, id := queueRun(t, s, time.Now().Add(time.Minute))
	req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
	if err != nil {
		t.Fatal(err)
	}
	req.TimeoutS = timeoutS
	got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err != nil {
		t.Fatal(err)
	}
	return *got, *req
}

func runWith(t *testing.T, s release.Store, l *fakeLauncher, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
	t.Helper()
	w := newRunsWorker(s, ownedRoute(runHost), l, t.TempDir(), "box-1", "boot-1")
	rt, _ := ownedRoute(runHost)(runHost)
	w.active[req.ID] = h
	w.execute(context.Background(), rec, req, *rt, h)
	res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID)
	if err != nil || res == nil {
		t.Fatalf("no result: %v", err)
	}
	if w.active[req.ID] != nil {
		t.Error("the run was not forgotten")
	}
	return res
}

func TestRunExecutesAndWritesTheResultAfterTheLastChunk(t *testing.T) {
	s := &putOrder{Store: release.NewDirStore(t.TempDir())}
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "3", Code: "1"}
	l := &fakeLauncher{handle: h, out: "hello\n", errOut: "oops\n"}
	res := runWith(t, s, l, rec, req, newRunHandle())
	if res.Exit != 3 || res.Signal != "" || res.LastSeq != 2 || res.Box != "box-1" || res.Started.IsZero() || res.Ended.IsZero() {
		t.Fatalf("result %+v", res)
	}
	if h.stops == 0 || !h.released {
		t.Fatal("the unit was not stopped after done, or the hold not released")
	}
	if r := currentRun(t, s.Store); r.State != provision.RunDone {
		t.Fatalf("run %+v, want done", r)
	}
	if !s.resultAfterChunks() {
		t.Fatalf("puts %q: result.json must come after every chunk", s.keys)
	}
}

// putOrder records the order of Puts.
type putOrder struct {
	release.Store
	mu   sync.Mutex
	keys []string
}

func (p *putOrder) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	p.mu.Lock()
	p.keys = append(p.keys, key)
	p.mu.Unlock()
	return p.Store.Put(ctx, key, body, ifMatch)
}

func (p *putOrder) resultAfterChunks() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	result := -1
	lastChunk := -1
	for i, k := range p.keys {
		if strings.HasSuffix(k, "/result.json") {
			result = i
		}
		if strings.Contains(k, "/out/") {
			lastChunk = i
		}
	}
	return result > lastChunk
}

func TestRunKilledBySignal(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "signal", Status: "9", Code: "2"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Signal != "9" || res.Exit != 0 {
		t.Fatalf("result %+v", res)
	}
}

func TestRunCancelledAfterLaunch(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	done := instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "143", Code: "1"}
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus), afterStop: &done}
	rh := newRunHandle()
	go func() { time.Sleep(50 * time.Millisecond); rh.requestCancel() }()
	res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, rh)
	if !res.Cancelled || res.Exit != 143 {
		t.Fatalf("result %+v", res)
	}
}

func TestRunTimesOut(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 1)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
	start := time.Now()
	res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
	if !res.TimedOut || time.Since(start) < time.Second {
		t.Fatalf("result %+v after %s", res, time.Since(start))
	}
}

func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	if _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID)); err != nil {
		t.Fatal(err)
	}
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
	res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
	if !res.Cancelled || h.stops != 0 {
		t.Fatalf("result %+v, unit stops %d", res, h.stops)
	}
}

func TestRunRefusesAnIncarnationThatDoesNotMatch(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	req.Incarnation = strings.Repeat("0", 64)
	l := &fakeLauncher{handle: &fakeHandle{}}
	if res := runWith(t, s, l, rec, req, newRunHandle()); res.Refused != runs.RefusedIncarnation || l.calls != 0 {
		t.Fatalf("result %+v, launches %d", res, l.calls)
	}
}

func TestRunRefusesAnImportUncommittedAtLaunch(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	cur, etag, err := provision.GetRecordETag(context.Background(), s, rec.Account, rec.App, rec.Host)
	if err != nil {
		t.Fatal(err)
	}
	cur.Seed = &provision.SeedSource{Kind: provision.SeedKindImport, State: "placed"}
	if err := provision.UpdateRecord(context.Background(), s, *cur, etag); err != nil {
		t.Fatal(err)
	}
	if res := runWith(t, s, &fakeLauncher{handle: &fakeHandle{}}, rec, req, newRunHandle()); res.Refused != runs.RefusedImport {
		t.Fatalf("result %+v (codex R2-2: checked again at launch)", res)
	}
}

func TestRunLaunchErrorsBecomeResults(t *testing.T) {
	cases := []struct {
		err  error
		want func(runs.Result) bool
	}{
		{activator.ErrCmdClaim, func(r runs.Result) bool { return r.Refused == runs.RefusedClaim }},
		{activator.ErrCmdNotNested, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
		{&activator.CmdWakeError{Reason: "wake failed: restore", Err: errors.New("x")}, func(r runs.Result) bool {
			return r.Refused == runs.RefusedWake && r.Detail == "wake failed: restore"
		}},
		{&activator.CmdStoppedError{Reason: activator.StopRestart}, func(r runs.Result) bool { return r.Interrupted == activator.StopRestart }},
		{errors.New("start cmd unit: 502"), func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
	}
	for _, c := range cases {
		s := release.NewDirStore(t.TempDir())
		rec, req := pickedRun(t, s, 60)
		if res := runWith(t, s, &fakeLauncher{err: c.err}, rec, req, newRunHandle()); !c.want(*res) {
			t.Errorf("%v: result %+v", c.err, res)
		}
		if r := currentRun(t, s); r.State != provision.RunDone {
			t.Errorf("%v: run %+v, want done", c.err, r)
		}
	}
}

func TestRunInterruptedByAnInstanceStop(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "143", Code: "1"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != activator.StopRestart {
		t.Fatalf("result %+v", res)
	}
}

func TestRunWhoseUnitWentAway(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	rec, req := pickedRun(t, s, 60)
	h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
	h.polls <- instancectl.CmdStatus{State: instancectl.CmdGone, InvocationID: "inv2"}
	if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != runs.InterruptedGone {
		t.Fatalf("result %+v", res)
	}
}

func TestExitOfReadsNumbersAndWords(t *testing.T) {
	for _, c := range []struct {
		code, status string
		exit         int
		signal       string
	}{
		{"1", "3", 3, ""}, {"exited", "3", 3, ""}, {"2", "9", 0, "9"}, {"killed", "15", 0, "15"}, {"3", "6", 0, "6"}, {"dumped", "6", 0, "6"},
	} {
		exit, signal := exitOf(instancectl.CmdStatus{Code: c.code, Status: c.status})
		if exit != c.exit || signal != c.signal {
			t.Errorf("%s/%s: %d %q", c.code, c.status, exit, signal)
		}
	}
}
```

`TestRunCancelledBeforeLaunchNeverStarts` and `TestRunRefusesAnImportUncommittedAtLaunch` rely on `Prepare`: the fake launcher calls it as `RunCmd` does, before any unit exists. `runWith` uses `ownedRoute`, `queueRun` and `currentRun` from `runs_test.go`.

- [ ] **Step 2: Run to verify they fail**

Run: `nice -n 10 go test ./internal/host/ -run 'TestRun|TestExitOf' -count=1`
Expected: FAIL, `undefined: exitOf`.

- [ ] **Step 3: Implement**

Delete Task 4's temporary `execute` from `runs.go`. `internal/host/runexec.go`:

```go
package host

import (
	"context"
	"errors"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/runs"
)

// runPollRetry paces a failed long-poll; runRecheckEvery re-reads the
// record while a run executes, a backstop for a cancel whose serial bump
// this box missed.
var (
	runPollRetry    = time.Second
	runRecheckEvery = 30 * time.Second
)

var (
	errRunCancelledBeforeLaunch = errors.New("host: the run was cancelled before launch")
	errRunImportAtLaunch        = errors.New("host: the instance is an import that is not committed")
	errRunIncarnation           = errors.New("host: the record is another incarnation")
	errRunNotOurs               = errors.New("host: the record no longer names this run as this box's")
)

// exitOf reads systemd's ExecMainCode and ExecMainStatus. `systemctl show`
// prints the code as a number (1 exited, 2 killed, 3 dumped); the words
// are accepted too. A signal leaves exit 0 and names the signal.
func exitOf(st instancectl.CmdStatus) (int, string) {
	n, _ := strconv.Atoi(st.Status)
	switch st.Code {
	case "2", "3", "killed", "dumped":
		return 0, strconv.Itoa(n)
	}
	return n, ""
}

// runResultForLaunch maps RunCmd's refusals to the result the member reads.
func runResultForLaunch(err error) runs.Result {
	var we *activator.CmdWakeError
	var se *activator.CmdStoppedError
	switch {
	case errors.Is(err, activator.ErrCmdClaim):
		return runs.Result{Refused: runs.RefusedClaim}
	case errors.Is(err, activator.ErrCmdNotNested):
		return runs.Result{Refused: runs.RefusedFlat}
	case errors.As(err, &we):
		return runs.Result{Refused: runs.RefusedWake, Detail: we.Reason}
	case errors.As(err, &se):
		return runs.Result{Interrupted: se.Reason}
	case errors.Is(err, errRunCancelledBeforeLaunch):
		return runs.Result{Cancelled: true}
	case errors.Is(err, errRunImportAtLaunch):
		return runs.Result{Refused: runs.RefusedImport}
	case errors.Is(err, errRunIncarnation):
		return runs.Result{Refused: runs.RefusedIncarnation}
	case errors.Is(err, errRunNotOurs):
		return runs.Result{Interrupted: runs.InterruptedGone}
	}
	return runs.Result{Interrupted: runs.InterruptedStartFailed}
}

// execute is one picked run's life (run spec §5.2 to §5.7).
func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
	defer w.forget(req.ID)
	instDir := provision.InstanceDir(w.dataDir, rec.Host)
	var handle activator.CmdHandle
	finish := func(res runs.Result) {
		res.Box, res.Ended = w.box, w.now().UTC()
		if err := runs.WriteResult(ctx, w.store, rec.Account, rec.App, req.ID, res); err != nil {
			log.Printf("agent: runs: %s %s: result: %v", rec.Host, req.ID, err)
		}
		if _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, provision.RunFinish(req.ID)); err != nil &&
			!errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
			log.Printf("agent: runs: %s %s: done: %v", rec.Host, req.ID, err)
		}
		if err := runs.RemoveRunDir(instDir, req.ID); err != nil && !errors.Is(err, os.ErrNotExist) {
			log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, req.ID, err)
		}
		if handle != nil {
			handle.Release()
		}
	}
	incarnation := func(r provision.Record) string {
		inc, _ := hostclaim.Incarnation(r.Account, r.App, r.Host, r.CreatedAt)
		return inc
	}
	if req.Host != rec.Host || req.Incarnation == "" || req.Incarnation != incarnation(rec) {
		finish(runs.Result{Refused: runs.RefusedIncarnation})
		return
	}
	chunks := runs.NewChunker(ctx, w.store, rec.Account, rec.App, req.ID)
	spec := activator.CmdSpec{
		RunID:  req.ID,
		Args:   req.Argv,
		RunDir: filepath.Join(instDir, runs.DirName, req.ID),
		Stdout: chunks.Stream(runs.StreamOut),
		Stderr: chunks.Stream(runs.StreamErr),
		// Right before launch (run spec §5.2): the run is still this box's,
		// not cancelled, not an uncommitted import, the same incarnation.
		// Then the files.
		Prepare: func(ctx context.Context) error {
			cur, err := provision.GetRecord(ctx, w.store, rec.Account, rec.App, rec.Host)
			if err != nil {
				return errRunNotOurs
			}
			r := cur.ActiveRun
			switch {
			case r == nil || r.ID != req.ID || r.Box != w.box || r.BootID != w.boot:
				return errRunNotOurs
			case r.State == provision.RunCancel:
				return errRunCancelledBeforeLaunch
			case cur.ImportUncommitted():
				return errRunImportAtLaunch
			case incarnation(*cur) != req.Incarnation:
				return errRunIncarnation
			}
			select {
			case <-h.cancel:
				return errRunCancelledBeforeLaunch
			default:
			}
			_, err = runs.StageFiles(ctx, w.store, instDir, req)
			return err
		},
	}
	handle, err := w.launch.RunCmd(ctx, rt, spec)
	if err != nil {
		last, truncated, _ := chunks.Close()
		res := runResultForLaunch(err)
		res.LastSeq, res.Truncated = last, truncated
		if res.Interrupted == runs.InterruptedStartFailed {
			log.Printf("agent: runs: %s %s: start: %v", rec.Host, req.ID, err)
		}
		finish(res)
		return
	}
	log.Printf("agent: runs: %s: run %s started (%s)", rec.Host, req.ID, handle.Invocation())
	res := runs.Result{Started: w.now().UTC()}
	status := w.follow(ctx, rec, req, handle, h, &res)
	// The unit is stopped and settled after every ending, done included
	// (run spec §5.4), so its streams have reached their end before the
	// chunker closes and the result names the last chunk.
	sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Minute)
	if err := handle.Stop(sctx); err != nil {
		log.Printf("agent: runs: %s %s: stop: %v", rec.Host, req.ID, err)
	}
	cancel()
	if status == nil && (res.Cancelled || res.TimedOut) {
		pctx, pcancel := context.WithTimeout(ctx, instancectl.DeadlineCmdPoll)
		if st, err := handle.Poll(pctx); err == nil && st.State == instancectl.CmdDone {
			status = &st
		}
		pcancel()
	}
	if status != nil && status.State == instancectl.CmdDone {
		res.Exit, res.Signal = exitOf(*status)
	}
	if reason := handle.Interrupted(); reason != "" {
		res.Interrupted = reason
	} else if status != nil && status.State == instancectl.CmdGone && !res.Cancelled && !res.TimedOut {
		res.Interrupted = runs.InterruptedGone
	}
	last, truncated, cerr := chunks.Close()
	if cerr != nil {
		log.Printf("agent: runs: %s %s: output: %v", rec.Host, req.ID, cerr)
	}
	res.LastSeq, res.Truncated = last, truncated
	finish(res)
}

// follow long-polls the unit until it is done or gone, the run is
// cancelled, or its timeout passes. It returns the final status when the
// unit reported one, and marks a cancel or a timeout on res.
func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) *instancectl.CmdStatus {
	pctx, stopPolling := context.WithCancel(ctx)
	defer stopPolling()
	statuses := make(chan instancectl.CmdStatus, 1)
	go func() {
		for {
			st, err := handle.Poll(pctx)
			if pctx.Err() != nil {
				return
			}
			if err != nil {
				time.Sleep(runPollRetry)
				continue
			}
			if st.State != instancectl.CmdRunning {
				statuses <- st
				return
			}
		}
	}()
	timeout := time.NewTimer(req.Timeout())
	defer timeout.Stop()
	recheck := time.NewTicker(runRecheckEvery)
	defer recheck.Stop()
	for {
		select {
		case st := <-statuses:
			return &st
		case <-h.cancel:
			res.Cancelled = true
			return nil
		case <-timeout.C:
			res.TimedOut = true
			return nil
		case <-recheck.C:
			if cur, err := provision.GetRecord(ctx, w.store, rec.Account, rec.App, rec.Host); err == nil &&
				cur.ActiveRun != nil && cur.ActiveRun.ID == req.ID && cur.ActiveRun.State == provision.RunCancel {
				h.requestCancel()
			}
		}
	}
}
```

- [ ] **Step 4: Run to verify they pass**

Run: `nice -n 10 go test ./internal/host/ -run 'TestRun|TestExitOf|TestBootPass' -count=1`
Expected: PASS. `StageFiles` runs for real inside `Prepare` (an empty file list makes the directory under the test's temp data dir; this needs linux, where these tests run).

- [ ] **Step 5: Commit**

```bash
git add internal/host/runexec.go internal/host/runexec_test.go internal/host/runs.go
git commit -m "host: one run's life: launch, follow, cancel, timeout, result after the last chunk (run spec §5.2 to §5.7, codex 3, 4, 5, R2-2, R2-6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 6: Wire the worker into the agent

**Files:**
- Modify: `internal/host/runs.go` (arming and the two pass entry points), `internal/host/agent.go:410` (arm), `:1131-1150` (`reconcileSerialized`), `:2211` (the boot pass at `fastConvergeLoop`'s arming, after `refreshMovesHolds`), `:2251-2252` (`fullPass`), `:2825` (`instancesLoop`)
- Test: `internal/host/runs_test.go` (append)

**Interfaces:**
- Consumes: `openDeployStore(ctx) (release.Store, error)`, `BoxLabel() string` (`metrics_loop.go:744`), `*registry.Registry.Get`, `*activator.Service`, `dataDir()`.
- Produces: `func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string)`; `func withRunsHook(cfg provision.ReconcileConfig) provision.ReconcileConfig`; `func runRunsPass(ctx context.Context)`; `func runRunsBootPass(ctx context.Context)`. Every one is a no-op until armed.

- [ ] **Step 1: Write the failing test**

Append to `runs_test.go`:

```go
func TestRunsHookIsOnlyThereWhenArmed(t *testing.T) {
	armedRuns.Store(nil)
	if cfg := withRunsHook(provision.ReconcileConfig{}); cfg.RunSeen != nil {
		t.Fatal("an unarmed box must not hook reconcile")
	}
	runRunsPass(context.Background())     // no-op, no panic
	runRunsBootPass(context.Background()) // ditto
	var l execLog
	w := testWorker(release.NewDirStore(t.TempDir()), noRoutes, "box-1", "boot-1", &l)
	armedRuns.Store(w)
	t.Cleanup(func() { armedRuns.Store(nil) })
	cfg := withRunsHook(provision.ReconcileConfig{})
	if cfg.RunSeen == nil {
		t.Fatal("an armed box hooks reconcile")
	}
	cfg.RunSeen(runRecord())
	if len(w.noted) != 1 {
		t.Fatal("the hook does not reach the worker")
	}
}
```

- [ ] **Step 2: Run to verify it fails**

Run: `nice -n 10 go test ./internal/host/ -run TestRunsHookIsOnlyThereWhenArmed -count=1`
Expected: FAIL, `undefined: withRunsHook`.

- [ ] **Step 3: Implement**

Append to `runs.go`:

```go
// armRunsPass arms the runs worker on a box with an activator. Thin, like
// armMovesPass: the behaviour is the worker's, tested there.
func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string) {
	if svc == nil || reg == nil {
		return
	}
	store, err := openDeployStore(context.Background())
	if err != nil {
		log.Printf("agent: runs disabled: %v", err)
		return
	}
	w := newRunsWorker(store, reg.Get, svc, dataDir, BoxLabel(), newBootID())
	armedRuns.Store(w)
	log.Printf("agent: runs armed (box %s, boot %s)", w.box, w.boot)
}

// withRunsHook gives reconcile the worker's note (run spec §5.2).
func withRunsHook(cfg provision.ReconcileConfig) provision.ReconcileConfig {
	if w := armedRuns.Load(); w != nil && cfg.RunSeen == nil {
		cfg.RunSeen = w.note
	}
	return cfg
}

// runRunsPass acts on what the last reconcile pass noted.
func runRunsPass(ctx context.Context) {
	if w := armedRuns.Load(); w != nil {
		w.pass(ctx)
	}
}

// runRunsBootPass is the worker's one pass at agent start.
func runRunsBootPass(ctx context.Context) {
	if w := armedRuns.Load(); w != nil {
		w.bootPass(ctx)
	}
}
```

`agent.go:410`, right after `armMovesPass(...)`:

```go
		armRunsPass(svc, reg, dd)
```

`reconcileSerialized` (`agent.go:1146`), beside `cfg = withMovesHolds(cfg)`:

```go
	// Runs ride this pass's record reads (run spec §5.2): the worker notes
	// every in-scope record with a run, and acts after the pass.
	cfg = withRunsHook(cfg)
```

`fastConvergeLoop`, right after `refreshMovesHolds(ctx, store)`:

```go
	// carlos run's boot pass (run spec §5.2, index item 2): reconcile's
	// first pass is a minute away and the CLI expires a run at 60 s.
	runRunsBootPass(ctx)
```

`fullPass`, after `mark("moves")`:

```go
		runRunsPass(ctx)
		mark("runs")
```

`instancesLoop`, after `runMovesPass(ctx, store)`:

```go
		runRunsPass(ctx)
```

- [ ] **Step 4: Run to verify it passes**

Run: `nice -n 10 go test ./internal/host/ -run 'TestRuns|TestRun|TestBootPass' -count=1 && go vet ./internal/host/ && GOOS=darwin go build ./... && GOOS=windows go build ./...`
Expected: PASS; both cross builds succeed (`internal/host` is linked into the CLI).

- [ ] **Step 5: Commit**

```bash
git add internal/host/runs.go internal/host/runs_test.go internal/host/agent.go
git commit -m "host: arm the runs worker; reconcile notes runs, the passes act, a boot pass at start

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>"
```

---

### Task 7: The plan's gate

- [ ] **Step 1: Run the gate**

```bash
export GOFLAGS=-buildvcs=false
go build ./... && go vet ./... && test -z "$(gofmt -l .)"
TMPDIR=/tmp/g nice -n 10 make test
GOOS=darwin go build ./... && GOOS=windows go build ./... && GOOS=linux GOARCH=arm64 go build ./...
nice -n 10 go test -race ./internal/runs/ ./internal/host/ -run 'Run|Chunker|Stage|BootPass' -count=1
```

Expected: every command exits 0 (prefix `GOCACHE=$TMPDIR/gocache` if the cache is read-only).

- [ ] **Step 2: Commit anything the gate changed**

`git status --short` is clean, or one `run: gofmt` commit with the trailer.

---

## Self-review notes

- §5.2: when the box looks (Task 3's hook, Task 6's passes, the boot pass for "at agent start"); which runs it takes (Task 4 `owned`); taking a run (one CAS, Task 4; re-check before launch, Task 5 `Prepare`); a picked run never run again (Task 4 `closeInterrupted`); the layout, wake, files and launch (Plan 1's `RunCmd`, Task 5's mapping and `Prepare`, Task 2's staging). "No pickup" is the CLI's and the console's (Plans 2, 4); the box never picks past the deadline.
- §5.4: done covers active (exited) and failed (Plan 1); here the result is read, then the unit is stopped and settled (`handle.Stop`), and `gone` ends `interrupted`.
- §5.5: the chunker (Task 1) and the order of `result.json` (Task 5 `TestRunExecutesAndWritesTheResultAfterTheLastChunk`); the run directory is deleted and the hold released after the result.
- §5.7: cancel before launch never launches; cancel and timeout stop the unit; the second Ctrl-C is the CLI's.
- §5.8 the box sweep: `SweepRunDirs` at the boot pass.
- §9 Pickup: owning box only, queued before agent start, queued before the box owned the host (Task 3's hook sees a row minted in the same pass, and the pass acts after reconcile), the two-box race, expiry against late pickup (here and Plan 2 Task 1), cancel before pickup never launching, the incarnation, the import at launch, the agent restart. "Run, move and import abort racing on the ETag" is Plan 2 Task 1's for run and move; the import abort is the seeded plan's.