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:

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:

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

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

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

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:

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

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

		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

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:

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:

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:

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:

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

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:

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:

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

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:

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:

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

		armRunsPass(svc, reg, dd)

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

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

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

		runRunsPass(ctx)
		mark("runs")

instancesLoop, after runMovesPass(ctx, store):

		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

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

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.