diff --git a/README.md b/README.md
index e9217a4..1ebc3e9 100644
--- a/README.md
+++ b/README.md
@@ -3,9 +3,9 @@
 Core of [Opifica](https://github.com/opifica), an open-source CI/CD system built in the EU
 where every job runs in a fresh virtual machine.
 
-> **Status:** early development. Working and tested: the pipeline language, the worker kit's warm pool,
-> workers joining and staying connected to the controller, and scheduling runs onto them. The job runner and
-> agent come next; see [docs/ROADMAP.md](docs/ROADMAP.md).
+> **Status:** early development. A pipeline runs end to end: the controller schedules its jobs, a worker checks
+> out the exact commit and runs each job's commands through the agent, and the output is stored. Real VMs,
+> secrets and Git providers come next; see [docs/ROADMAP.md](docs/ROADMAP.md).
 > See the [organisation page](https://github.com/opifica) for the architecture.
 
 ## What lives here
@@ -16,12 +16,14 @@ where every job runs in a fresh virtual machine.
 | `schema/` | JSON Schema for pipeline files, for editor autocompletion | Working |
 | `cmd/opifica` | CLI: `validate`, `plan`, `schema` | Working |
 | `proto/` | API definitions: worker ↔ controller, worker ↔ guest agent (Protobuf, ConnectRPC) | Draft |
-| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, image verification, controller client | Working |
+| `pkg/worker` | Shared worker kit: `Hypervisor` interface, warm VM pool, controller client, Git mirrors, job runner, `worker.Run` | Working |
 | `pkg/worker/workertest` | Fake hypervisor for testing worker code without VMs | Working |
 | `cmd/controller` | Join tokens, worker joining over TLS, the Connect stream, worker revocation | Working |
 | `internal/controller/scheduler` | Runs and jobs, matching jobs to workers, leases, retries, cancellation | Working |
-| | Logs, secrets, Git providers, admin API | Not started |
-| `cmd/agent` | Runs inside each VM, executes steps, streams logs and results | Skeleton |
+| `internal/controller/logs` | Job output on the controller's disk | Working |
+| | Secrets, Git providers, admin API | Not started |
+| `cmd/agent`, `internal/agent` | Runs inside each VM: checkout, running steps, masking secrets in output, killing process trees on cancel and timeout | Working |
+| `cmd/devworker` | A worker without VMs for trying Opifica on one machine (no isolation) | Working |
 | `cmd/signer` | Signing service for release artifacts | Not started |
 
 The workers live in their own repositories, so each one contains only its own OS-specific code:
@@ -123,18 +125,25 @@ Database tests need PostgreSQL. Either set `OPIFICA_TEST_DATABASE_URL` to a serv
 on, or install the PostgreSQL server binaries (e.g. `apt install postgresql-17`): the tests then start their own
 throwaway server. Without either, those tests are skipped.
 
-### Trying the controller
+### Running a pipeline on one machine
+
+`devworker` runs jobs without VMs, directly on your machine, with **no isolation**. Use it only with code you trust.
 
 ```sh
 export OPIFICA_DATABASE_URL=postgres://localhost/opifica
-go run ./cmd/controller token create -pool trusted   # prints a join token and the CA fingerprint
-go run ./cmd/controller serve                        # TLS on :8443
-go run ./cmd/controller worker list
+go build -o bin/ ./cmd/...
+
+bin/controller token create -pool trusted        # prints a join token and the CA fingerprint
+bin/controller serve -addr 127.0.0.1:8443 &       # TLS; keep it running
+
+bin/devworker join -controller https://127.0.0.1:8443 -token opj_... -ca-fingerprint sha256:...
+bin/devworker run  -controller https://127.0.0.1:8443 &
 
 # Until Git providers exist, start runs by hand:
-go run ./cmd/controller run create -repo acme/app -pool trusted -sha $(git rev-parse HEAD) -file .opifica/pipeline.yaml
-go run ./cmd/controller run list
-go run ./cmd/controller run show <run-id>
+bin/controller run create -repo acme/app -clone-url "$PWD" -pool trusted \
+    -sha "$(git rev-parse HEAD)" -file .opifica/pipeline.yaml
+bin/controller run show <run-id>
+bin/controller run logs <run-id>
 ```
 
 ## Stack
diff --git a/cmd/agent/main.go b/cmd/agent/main.go
index f934a2a..2204144 100644
--- a/cmd/agent/main.go
+++ b/cmd/agent/main.go
@@ -2,41 +2,59 @@
 // worker on the host: checkout, running steps, streaming output and
 // collecting files. It never talks to the controller.
 //
-// This is a skeleton: every method returns "unimplemented". In production the
-// agent listens on vsock or virtio-serial; -listen is for development.
+// In a VM the agent listens on vsock (not implemented yet; see the worker
+// repositories). -listen with a TCP address or unix:/path is for development.
 package main
 
 import (
 	"context"
 	"flag"
 	"log/slog"
-	"net/http"
+	"net"
 	"os"
 	"os/signal"
+	"path/filepath"
+	"strings"
 	"syscall"
 
-	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/agent"
 	"github.com/opifica/opifica/internal/server"
 	"github.com/opifica/opifica/internal/version"
 )
 
 func main() {
-	listen := flag.String("listen", "127.0.0.1:7070", "address to listen on (development only)")
+	listen := flag.String("listen", "127.0.0.1:7070", "TCP address or unix:/path/to/socket (development)")
+	root := flag.String("root", defaultRoot(), "work directory; every request path stays inside it")
 	flag.Parse()
 	log := slog.New(slog.NewTextHandler(os.Stderr, nil)).With("service", "agent")
-	log.Info("starting", "version", version.Version)
+	log.Info("starting", "version", version.Version, "root", *root)
+
+	if err := os.MkdirAll(*root, 0o755); err != nil {
+		log.Error("creating work directory", "err", err)
+		os.Exit(1)
+	}
+	network, addr := "tcp", *listen
+	if p, ok := strings.CutPrefix(*listen, "unix:"); ok {
+		network, addr = "unix", p
+		os.Remove(p)
+	}
+	ln, err := net.Listen(network, addr)
+	if err != nil {
+		log.Error("listening", "err", err)
+		os.Exit(1)
+	}
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
-	mux := http.NewServeMux()
-	mux.HandleFunc("GET /healthz", server.Health)
-	mux.Handle(opificav1connect.NewAgentServiceHandler(&agentService{}))
-	if err := server.Run(ctx, *listen, mux, log); err != nil {
+	if err := server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: *root, Log: log}), log); err != nil {
 		log.Error("server stopped", "err", err)
 		os.Exit(1)
 	}
 }
 
-type agentService struct {
-	opificav1connect.UnimplementedAgentServiceHandler
+func defaultRoot() string {
+	if d, err := os.UserHomeDir(); err == nil {
+		return filepath.Join(d, "opifica-work")
+	}
+	return "opifica-work"
 }
diff --git a/cmd/controller/main.go b/cmd/controller/main.go
index d845eeb..752e676 100644
--- a/cmd/controller/main.go
+++ b/cmd/controller/main.go
@@ -2,8 +2,8 @@
 // integration, pipeline engine, scheduler and secrets broker.
 //
 // Implemented so far: the database, the built-in certificate authority, join
-// tokens, Join, the Connect stream, and the scheduler (runs, jobs, leases,
-// retries, cancellation). Logs, secrets and Git providers come next.
+// tokens, Join, the Connect stream, the scheduler (runs, jobs, leases,
+// retries, cancellation) and job logs. Secrets and Git providers come next.
 package main
 
 import (
@@ -15,6 +15,7 @@ import (
 	"log/slog"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"strings"
 	"syscall"
 	"text/tabwriter"
@@ -24,6 +25,7 @@ import (
 
 	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/workers"
@@ -42,6 +44,7 @@ Commands:
   run list       List recent runs
   run show       Show a run and its jobs
   run cancel     Cancel a run
+  run logs       Print the output of a run's jobs
   version        Print the version
 
 Common options (or environment variables):
@@ -139,7 +142,9 @@ func serve(args []string, stderr io.Writer) int {
 		return 1
 	}
 	log.Info("certificate authority ready", "fingerprint", svc.CA.Fingerprint())
-	sched := scheduler.New(svc.Store, svc.Sessions(), log.With("component", "scheduler"), scheduler.Config{})
+	sched := scheduler.New(svc.Store, svc.Sessions(), log.With("component", "scheduler"), scheduler.Config{
+		Logs: &logs.Store{Dir: filepath.Join(c.dataDir, "logs")},
+	})
 	svc.Messages = sched
 	go sched.Run(ctx)
 	if err := server.RunTLS(ctx, *addr, api.NewHandler(svc), tlsCfg, log); err != nil {
diff --git a/cmd/controller/runs.go b/cmd/controller/runs.go
index 058a245..c5b3709 100644
--- a/cmd/controller/runs.go
+++ b/cmd/controller/runs.go
@@ -2,17 +2,20 @@ package main
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"log/slog"
 	"os"
+	"path/filepath"
 	"strings"
 	"text/tabwriter"
 	"time"
 
 	"github.com/google/uuid"
 
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store/db"
 	"github.com/opifica/opifica/internal/pipeline"
@@ -28,6 +31,8 @@ func runCommand(sub string, args []string, stdout, stderr io.Writer) int {
 		return runShow(args, stdout, stderr)
 	case "cancel":
 		return runCancel(args, stdout, stderr)
+	case "logs":
+		return runLogs(args, stdout, stderr)
 	}
 	fmt.Fprintf(stderr, "controller: unknown command \"run %s\"\n\n%s", sub, usage)
 	return 2
@@ -233,3 +238,47 @@ func short(sha string) string {
 	}
 	return sha
 }
+
+func runLogs(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run logs", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	only := fs.String("job", "", "show only this job (its key, e.g. test[os=linux], or its name)")
+	id, ok := runIDArg(fs, args, stderr, "logs")
+	if !ok {
+		return 2
+	}
+	store := &logs.Store{Dir: filepath.Join(c.dataDir, "logs")}
+	return withScheduler(c, stderr, func(ctx context.Context, _ *scheduler.Scheduler, q *db.Queries) error {
+		jobs, err := q.ListRunJobs(ctx, id)
+		if err != nil {
+			return err
+		}
+		shown := 0
+		for _, j := range jobs {
+			if *only != "" && j.Key != *only && j.Name != *only {
+				continue
+			}
+			shown++
+			fmt.Fprintf(stdout, "---- %s (%s, %s) ----\n", j.Name, j.Os, j.State)
+			r, err := store.Open(j.ID)
+			if errors.Is(err, logs.ErrNoLog) {
+				fmt.Fprintln(stdout, "(no output)")
+				continue
+			}
+			if err != nil {
+				return err
+			}
+			_, err = io.Copy(stdout, r)
+			r.Close()
+			if err != nil {
+				return err
+			}
+		}
+		if *only != "" && shown == 0 {
+			return fmt.Errorf("run %s has no job %q", id, *only)
+		}
+		return nil
+	})
+}
diff --git a/cmd/devworker/main.go b/cmd/devworker/main.go
new file mode 100644
index 0000000..ae6e740
--- /dev/null
+++ b/cmd/devworker/main.go
@@ -0,0 +1,129 @@
+// Command devworker is a worker for trying Opifica on one machine, without
+// VMs. Every job gets a fresh directory and its own agent, but runs directly
+// on this machine as the current user.
+//
+// THERE IS NO ISOLATION. Run only code you trust. Real workers live in the
+// worker-linux, worker-windows and worker-macos repositories.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"github.com/opifica/opifica/internal/version"
+	"github.com/opifica/opifica/pkg/worker"
+	"github.com/opifica/opifica/pkg/worker/workertest"
+)
+
+const usage = `Usage: devworker <command> [options]
+
+A worker WITHOUT VMs, for development: jobs run on this machine, unisolated.
+
+Commands:
+  join   Join a controller with a token from "controller token create"
+  run    Run jobs
+`
+
+func main() {
+	os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
+}
+
+func run(args []string, stdout, stderr io.Writer) int {
+	if len(args) == 0 {
+		fmt.Fprint(stderr, usage)
+		return 2
+	}
+	switch args[0] {
+	case "join":
+		return join(args[1:], stdout, stderr)
+	case "run":
+		return runJobs(args[1:], stderr)
+	case "version":
+		fmt.Fprintln(stdout, version.Version)
+		return 0
+	}
+	fmt.Fprint(stderr, usage)
+	return 2
+}
+
+type common struct {
+	controller string
+	stateDir   string
+	capacity   int
+}
+
+func (c *common) register(fs *flag.FlagSet) {
+	fs.StringVar(&c.controller, "controller", "https://localhost:8443", "controller URL")
+	home, _ := os.UserHomeDir()
+	fs.StringVar(&c.stateDir, "state-dir", filepath.Join(home, ".opifica-devworker"), "identity and repository mirrors")
+	fs.IntVar(&c.capacity, "capacity", 2, "jobs to run at once")
+}
+
+func (c *common) pool() worker.PoolConfig {
+	return worker.PoolConfig{
+		Capacity: c.capacity,
+		Images:   map[string]worker.Image{"default": {Name: "default"}},
+		Warm:     map[string]int{"default": 1},
+	}
+}
+
+func join(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("join", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	token := fs.String("token", "", "join token (required)")
+	fp := fs.String("ca-fingerprint", "", "CA fingerprint printed with the token (required)")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if *token == "" || *fp == "" {
+		fmt.Fprintln(stderr, "devworker: -token and -ca-fingerprint are required")
+		return 2
+	}
+	id, err := worker.Join(context.Background(), worker.JoinConfig{
+		ControllerURL: c.controller, Token: *token, CAFingerprint: *fp, StateDir: c.stateDir,
+		Info: worker.Info(c.pool(), nil),
+	})
+	if err != nil {
+		fmt.Fprintf(stderr, "devworker: %v\n", err)
+		return 1
+	}
+	fmt.Fprintf(stdout, "Joined as worker %s. Identity saved in %s.\n", id.WorkerID, c.stateDir)
+	return 0
+}
+
+func runJobs(args []string, stderr io.Writer) int {
+	fs := flag.NewFlagSet("run", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	log := slog.New(slog.NewTextHandler(stderr, nil)).With("service", "devworker")
+	log.Warn("jobs run on this machine WITHOUT isolation; use only with code you trust")
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	err := worker.Run(ctx, worker.Config{
+		ControllerURL:   c.controller,
+		StateDir:        c.stateDir,
+		Hypervisor:      &workertest.Hypervisor{Agent: true},
+		Pool:            c.pool(),
+		AllowLocalRepos: true,
+		Logger:          log,
+	})
+	if err != nil && ctx.Err() == nil {
+		log.Error("worker stopped", "err", err)
+		return 1
+	}
+	return 0
+}
diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md
index 242f0b0..6d10433 100644
--- a/docs/ROADMAP.md
+++ b/docs/ROADMAP.md
@@ -36,7 +36,7 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real
 - [x] **Scheduler:** match planned jobs to workers by OS, architecture, image, labels and pool; respect `needs`;
   safe with several controllers (`FOR UPDATE SKIP LOCKED`)
 - [x] **Runs:** create a run from a pipeline file and an event; track job and run state; cancel
-- [ ] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk
+- [x] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk (`controller run logs`)
 - [ ] **Secrets:** encrypted store; requests checked against the job token and policy
 - [ ] **Admin config:** static file for pools, routing (event → pool) and secret policy
 - [x] **TLS** on the worker API, with worker client certificates accepted
@@ -48,16 +48,18 @@ result are stored. Tested in ordinary CI, using the fake hypervisor with a real
 ### Worker kit and agent
 
 - [x] Controller client: join (saving the identity to disk), connect, reconnect with backoff, heartbeats
-- [ ] Heartbeats filled from the warm pool and job runner
-- [ ] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → collect reports
-  and artifacts → destroy the VM
-- [ ] Local Git mirror per repository
-- [ ] Agent: `Ready`, `Checkout`, `Exec`, `Collect`
+- [x] Heartbeats filled from the warm pool and job runner
+- [x] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → destroy the VM;
+  updates survive reconnects through an outbox
+- [ ] Collect and store test reports and artifacts (steps are skipped with a note for now)
+- [x] Local Git mirror per repository
+- [x] Agent: `Ready`, `Checkout`, `Exec`, `Collect`; secrets masked in output; process trees killed on cancel and timeout
+- [x] `devworker`: try everything on one machine without VMs (no isolation)
 
 ### Testing
 
-- [ ] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI
-- [ ] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret
+- [x] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI
+- [x] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret
 
 ## M2 · Triggered by Git
 
diff --git a/internal/agent/agent.go b/internal/agent/agent.go
new file mode 100644
index 0000000..e1b0689
--- /dev/null
+++ b/internal/agent/agent.go
@@ -0,0 +1,321 @@
+// Package agent is the program that runs inside every job VM. It serves the
+// AgentService to the worker on the host: checking out code, running steps
+// and collecting files. It never talks to the controller, and it only ever
+// sees the secrets of the step it is running.
+package agent
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"log/slog"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strings"
+	"sync"
+	"time"
+
+	"connectrpc.com/connect"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/mask"
+	"github.com/opifica/opifica/internal/version"
+)
+
+// TimeoutExitCode is reported when a command is stopped because it ran out
+// of time, matching the timeout(1) convention.
+const TimeoutExitCode = 124
+
+const (
+	chunkSize    = 64 << 10
+	maxCollected = 2 << 30 // per Collect call
+)
+
+// Server implements opificav1connect.AgentServiceHandler.
+type Server struct {
+	// Root is the work directory. Every path in a request is relative to it
+	// and may not leave it.
+	Root string
+	Log  *slog.Logger
+}
+
+var _ opificav1connect.AgentServiceHandler = (*Server)(nil)
+
+func (s *Server) log() *slog.Logger {
+	if s.Log != nil {
+		return s.Log
+	}
+	return slog.Default()
+}
+
+// resolve turns a request path into a path under Root.
+func (s *Server) resolve(p string) (string, error) {
+	if p == "" {
+		return s.Root, nil
+	}
+	if filepath.IsAbs(p) || strings.HasPrefix(p, "/") || strings.HasPrefix(p, `\`) {
+		return "", fmt.Errorf("path %q must be relative to the work directory", p)
+	}
+	full := filepath.Join(s.Root, p)
+	rel, err := filepath.Rel(s.Root, full)
+	if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("path %q leaves the work directory", p)
+	}
+	return full, nil
+}
+
+func invalid(err error) error { return connect.NewError(connect.CodeInvalidArgument, err) }
+
+// Ready implements AgentServiceHandler.
+func (s *Server) Ready(context.Context, *connect.Request[opificav1.ReadyRequest]) (*connect.Response[opificav1.ReadyResponse], error) {
+	return connect.NewResponse(&opificav1.ReadyResponse{
+		AgentVersion: version.Version, Os: runtime.GOOS, Arch: runtime.GOARCH,
+	}), nil
+}
+
+var shaRe = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
+
+// Checkout implements AgentServiceHandler.
+func (s *Server) Checkout(ctx context.Context, req *connect.Request[opificav1.CheckoutRequest]) (*connect.Response[opificav1.CheckoutResponse], error) {
+	m := req.Msg
+	if !shaRe.MatchString(m.GetCommitSha()) {
+		return nil, invalid(fmt.Errorf("commit_sha must be a full hex commit id, got %q", m.GetCommitSha()))
+	}
+	if m.GetMirrorUrl() == "" {
+		return nil, invalid(errors.New("mirror_url is required"))
+	}
+	dir, err := s.resolve(m.GetDirectory())
+	if err != nil {
+		return nil, invalid(err)
+	}
+	if entries, err := os.ReadDir(dir); err == nil && len(entries) > 0 {
+		return nil, invalid(fmt.Errorf("checkout directory %q is not empty", m.GetDirectory()))
+	}
+
+	steps := [][]string{
+		{"clone", "--quiet", "--no-checkout", "--", m.GetMirrorUrl(), dir},
+		{"-C", dir, "checkout", "--quiet", "--detach", m.GetCommitSha()},
+		{"-C", dir, "rev-parse", "HEAD"},
+	}
+	var head string
+	for _, args := range steps {
+		out, err := git(ctx, args...)
+		if err != nil {
+			return nil, connect.NewError(connect.CodeFailedPrecondition, err)
+		}
+		head = strings.TrimSpace(out)
+	}
+	if head != m.GetCommitSha() {
+		return nil, connect.NewError(connect.CodeFailedPrecondition, fmt.Errorf("checked out %s, wanted %s", head, m.GetCommitSha()))
+	}
+	return connect.NewResponse(&opificav1.CheckoutResponse{HeadSha: head}), nil
+}
+
+func git(ctx context.Context, args ...string) (string, error) {
+	cmd := exec.CommandContext(ctx, "git", args...)
+	// Never prompt, never read the host's or user's git config.
+	cmd.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL="+os.DevNull)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &stdout, &stderr
+	if err := cmd.Run(); err != nil {
+		return "", fmt.Errorf("git %s: %v: %s", args[0], err, strings.TrimSpace(stderr.String()))
+	}
+	return stdout.String(), nil
+}
+
+// shellCommand returns the program and arguments that run script.
+func shellCommand(shell, script string) ([]string, error) {
+	if shell == "" {
+		if runtime.GOOS == "windows" {
+			shell = "pwsh"
+			if _, err := exec.LookPath("pwsh"); err != nil {
+				shell = "powershell"
+			}
+		} else {
+			shell = "bash"
+			if _, err := exec.LookPath("bash"); err != nil {
+				shell = "sh"
+			}
+		}
+	}
+	switch shell {
+	case "bash":
+		// Stop at the first failing command, including inside pipelines.
+		return []string{"bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", script}, nil
+	case "sh":
+		return []string{"sh", "-e", "-c", script}, nil
+	case "pwsh", "powershell":
+		return []string{shell, "-NoProfile", "-NonInteractive", "-Command",
+			"$ErrorActionPreference = 'Stop'\n" + script + "\nif ($LASTEXITCODE) { exit $LASTEXITCODE }"}, nil
+	case "cmd":
+		return []string{"cmd", "/D", "/S", "/C", script}, nil
+	}
+	return nil, fmt.Errorf("unknown shell %q (bash, sh, pwsh, powershell or cmd)", shell)
+}
+
+// streamWriter sends output as ExecResponse chunks. stdout and stderr write
+// from different goroutines, so sends are serialized.
+type streamWriter struct {
+	mu     *sync.Mutex
+	stream *connect.ServerStream[opificav1.ExecResponse]
+	stderr bool
+}
+
+func (w *streamWriter) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	data := bytes.Clone(p)
+	msg := &opificav1.ExecResponse{Event: &opificav1.ExecResponse_Stdout{Stdout: data}}
+	if w.stderr {
+		msg.Event = &opificav1.ExecResponse_Stderr{Stderr: data}
+	}
+	if err := w.stream.Send(msg); err != nil {
+		return 0, err
+	}
+	return len(p), nil
+}
+
+// Exec implements AgentServiceHandler. Output is masked before it leaves the
+// agent. The command and everything it started are killed when the step
+// times out or the worker cancels the call.
+func (s *Server) Exec(ctx context.Context, req *connect.Request[opificav1.ExecRequest], stream *connect.ServerStream[opificav1.ExecResponse]) error {
+	m := req.Msg
+	if strings.TrimSpace(m.GetCommand()) == "" {
+		return invalid(errors.New("command is required"))
+	}
+	argv, err := shellCommand(m.GetShell(), m.GetCommand())
+	if err != nil {
+		return invalid(err)
+	}
+	dir, err := s.resolve(m.GetWorkingDirectory())
+	if err != nil {
+		return invalid(err)
+	}
+	if info, err := os.Stat(dir); err != nil || !info.IsDir() {
+		return invalid(fmt.Errorf("working directory %q does not exist", m.GetWorkingDirectory()))
+	}
+
+	runCtx := ctx
+	var cancelTimeout context.CancelFunc = func() {}
+	if t := m.GetTimeoutSeconds(); t > 0 {
+		runCtx, cancelTimeout = context.WithTimeout(ctx, time.Duration(t)*time.Second)
+	}
+	defer cancelTimeout()
+
+	cmd := exec.CommandContext(runCtx, argv[0], argv[1:]...)
+	cmd.Dir = dir
+	cmd.Env = os.Environ()
+	for k, v := range m.GetEnv() {
+		cmd.Env = append(cmd.Env, k+"="+v)
+	}
+	killTree(cmd)
+	// If a background process keeps the output open, don't wait forever.
+	cmd.WaitDelay = 5 * time.Second
+
+	var mu sync.Mutex
+	stdout := mask.NewWriter(&streamWriter{mu: &mu, stream: stream}, m.GetMask())
+	stderr := mask.NewWriter(&streamWriter{mu: &mu, stream: stream, stderr: true}, m.GetMask())
+	cmd.Stdout, cmd.Stderr = stdout, stderr
+
+	runErr := cmd.Run()
+	stdout.Flush()
+	stderr.Flush()
+
+	if ctx.Err() != nil {
+		return connect.NewError(connect.CodeCanceled, errors.New("cancelled by the worker"))
+	}
+	code := 0
+	var exitErr *exec.ExitError
+	switch {
+	case runErr == nil:
+	case errors.Is(runCtx.Err(), context.DeadlineExceeded):
+		code = TimeoutExitCode
+		fmt.Fprintf(stderr, "\nopifica: step timed out after %ds\n", m.GetTimeoutSeconds())
+		stderr.Flush()
+	case errors.As(runErr, &exitErr):
+		code = exitErr.ExitCode()
+		if code < 0 {
+			code = 1 // killed by a signal
+		}
+	default:
+		// The command couldn't start, e.g. the shell is missing.
+		return connect.NewError(connect.CodeFailedPrecondition, runErr)
+	}
+	return stream.Send(&opificav1.ExecResponse{Event: &opificav1.ExecResponse_ExitCode{ExitCode: int32(code)}})
+}
+
+// Collect implements AgentServiceHandler. Patterns are globs relative to
+// base_directory. Symbolic links are skipped, so a job can't make the agent
+// read files outside the work directory.
+func (s *Server) Collect(ctx context.Context, req *connect.Request[opificav1.CollectRequest], stream *connect.ServerStream[opificav1.CollectResponse]) error {
+	base, err := s.resolve(req.Msg.GetBaseDirectory())
+	if err != nil {
+		return invalid(err)
+	}
+	seen := map[string]bool{}
+	var total int64
+	for _, pattern := range req.Msg.GetPatterns() {
+		if filepath.IsAbs(pattern) || strings.Contains(filepath.ToSlash(pattern), "..") {
+			return invalid(fmt.Errorf("pattern %q must stay inside the base directory", pattern))
+		}
+		matches, err := filepath.Glob(filepath.Join(base, pattern))
+		if err != nil {
+			return invalid(fmt.Errorf("pattern %q: %v", pattern, err))
+		}
+		for _, path := range matches {
+			if seen[path] {
+				continue
+			}
+			seen[path] = true
+			info, err := os.Lstat(path)
+			if err != nil || !info.Mode().IsRegular() {
+				continue // directories, symlinks, devices
+			}
+			if total += info.Size(); total > maxCollected {
+				return connect.NewError(connect.CodeResourceExhausted, fmt.Errorf("collected files exceed %d bytes", maxCollected))
+			}
+			rel, _ := filepath.Rel(base, path)
+			if err := sendFile(ctx, stream, path, filepath.ToSlash(rel)); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func sendFile(ctx context.Context, stream *connect.ServerStream[opificav1.CollectResponse], path, rel string) error {
+	f, err := os.Open(path)
+	if err != nil {
+		if errors.Is(err, fs.ErrNotExist) {
+			return nil
+		}
+		return err
+	}
+	defer f.Close()
+	buf := make([]byte, chunkSize)
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		n, err := f.Read(buf)
+		eof := errors.Is(err, io.EOF)
+		if err != nil && !eof {
+			return err
+		}
+		if n > 0 || eof {
+			if err := stream.Send(&opificav1.CollectResponse{Path: rel, Data: bytes.Clone(buf[:n]), Eof: eof}); err != nil {
+				return err
+			}
+		}
+		if eof {
+			return nil
+		}
+	}
+}
diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go
new file mode 100644
index 0000000..2a82a95
--- /dev/null
+++ b/internal/agent/agent_test.go
@@ -0,0 +1,337 @@
+package agent_test
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"connectrpc.com/connect"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/agent"
+	"github.com/opifica/opifica/internal/server"
+)
+
+func start(t *testing.T) (opificav1connect.AgentServiceClient, string) {
+	t.Helper()
+	root := t.TempDir()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
+	done := make(chan struct{})
+	go func() {
+		server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: root, Log: log}), log)
+		close(done)
+	}()
+	t.Cleanup(func() { cancel(); <-done })
+	addr := ln.Addr().String()
+	client := agent.NewClient(func(ctx context.Context) (net.Conn, error) {
+		var d net.Dialer
+		return d.DialContext(ctx, "tcp", addr)
+	})
+	return client, root
+}
+
+type result struct {
+	stdout, stderr string
+	exit           int32
+	err            error
+}
+
+func run(t *testing.T, c opificav1connect.AgentServiceClient, req *opificav1.ExecRequest) result {
+	t.Helper()
+	return runCtx(context.Background(), t, c, req)
+}
+
+func runCtx(ctx context.Context, t *testing.T, c opificav1connect.AgentServiceClient, req *opificav1.ExecRequest) result {
+	t.Helper()
+	stream, err := c.Exec(ctx, connect.NewRequest(req))
+	if err != nil {
+		return result{err: err}
+	}
+	var r result
+	r.exit = -1
+	for stream.Receive() {
+		switch e := stream.Msg().GetEvent().(type) {
+		case *opificav1.ExecResponse_Stdout:
+			r.stdout += string(e.Stdout)
+		case *opificav1.ExecResponse_Stderr:
+			r.stderr += string(e.Stderr)
+		case *opificav1.ExecResponse_ExitCode:
+			r.exit = e.ExitCode
+		}
+	}
+	r.err = stream.Err()
+	return r
+}
+
+func unixOnly(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("uses a POSIX shell")
+	}
+}
+
+func TestReady(t *testing.T) {
+	c, _ := start(t)
+	resp, err := c.Ready(context.Background(), connect.NewRequest(&opificav1.ReadyRequest{}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.Msg.GetOs() != runtime.GOOS || resp.Msg.GetArch() != runtime.GOARCH {
+		t.Errorf("ready = %v", resp.Msg)
+	}
+}
+
+func TestExec(t *testing.T) {
+	unixOnly(t)
+	c, root := start(t)
+	os.MkdirAll(filepath.Join(root, "src", "sub"), 0o755)
+
+	r := run(t, c, &opificav1.ExecRequest{
+		Command:          `echo "out $GREETING"; echo err >&2; pwd`,
+		WorkingDirectory: "src/sub",
+		Env:              map[string]string{"GREETING": "hello"},
+	})
+	if r.err != nil || r.exit != 0 {
+		t.Fatalf("exit %d, err %v", r.exit, r.err)
+	}
+	if !strings.HasPrefix(r.stdout, "out hello\n") || r.stderr != "err\n" {
+		t.Errorf("stdout %q, stderr %q", r.stdout, r.stderr)
+	}
+	if !strings.HasSuffix(strings.TrimSpace(r.stdout), filepath.Join("src", "sub")) {
+		t.Errorf("ran in the wrong directory: %q", r.stdout)
+	}
+
+	if r := run(t, c, &opificav1.ExecRequest{Command: "exit 3"}); r.exit != 3 || r.err != nil {
+		t.Errorf("exit 3: got exit %d, err %v", r.exit, r.err)
+	}
+}
+
+func TestBashStopsAtFirstFailure(t *testing.T) {
+	unixOnly(t)
+	if _, err := exec.LookPath("bash"); err != nil {
+		t.Skip("no bash")
+	}
+	c, _ := start(t)
+	r := run(t, c, &opificav1.ExecRequest{Command: "false | true\necho should-not-run"})
+	if r.exit == 0 || strings.Contains(r.stdout, "should-not-run") {
+		t.Errorf("a failing pipeline didn't stop the step: exit %d, stdout %q", r.exit, r.stdout)
+	}
+	r = run(t, c, &opificav1.ExecRequest{Command: "echo ok", Shell: "sh"})
+	if r.exit != 0 || r.stdout != "ok\n" {
+		t.Errorf("sh: exit %d, stdout %q", r.exit, r.stdout)
+	}
+}
+
+func TestSecretsAreMasked(t *testing.T) {
+	unixOnly(t)
+	c, _ := start(t)
+	// printf in pieces so the secret reaches the agent split across writes.
+	r := run(t, c, &opificav1.ExecRequest{
+		Command: `printf 'token=s3cr'; sleep 0.05; printf '3t-value!\n'; echo "$TOKEN" >&2`,
+		Env:     map[string]string{"TOKEN": "s3cr3t-value"},
+		Mask:    []string{"s3cr3t-value"},
+	})
+	if strings.Contains(r.stdout+r.stderr, "s3cr3t") {
+		t.Errorf("secret leaked: stdout %q, stderr %q", r.stdout, r.stderr)
+	}
+	if r.stdout != "token=***!\n" || r.stderr != "***\n" {
+		t.Errorf("stdout %q, stderr %q", r.stdout, r.stderr)
+	}
+}
+
+func TestTimeoutKillsTheWholeTree(t *testing.T) {
+	unixOnly(t)
+	c, root := start(t)
+	pidFile := filepath.Join(root, "child.pid")
+	start := time.Now()
+	r := run(t, c, &opificav1.ExecRequest{
+		Command:        `sleep 60 & echo $! > child.pid; wait`,
+		TimeoutSeconds: 1,
+	})
+	if r.exit != agent.TimeoutExitCode || !strings.Contains(r.stderr, "timed out after 1s") {
+		t.Errorf("exit %d, stderr %q", r.exit, r.stderr)
+	}
+	if d := time.Since(start); d > 10*time.Second {
+		t.Errorf("took %v", d)
+	}
+	assertDead(t, pidFile)
+}
+
+func TestCancelKillsTheWholeTree(t *testing.T) {
+	unixOnly(t)
+	c, root := start(t)
+	pidFile := filepath.Join(root, "child.pid")
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		for {
+			if _, err := os.Stat(pidFile); err == nil {
+				cancel()
+				return
+			}
+			time.Sleep(10 * time.Millisecond)
+		}
+	}()
+	r := runCtx(ctx, t, c, &opificav1.ExecRequest{Command: `sleep 60 & echo $! > child.pid; wait`})
+	if r.err == nil {
+		t.Error("cancelled exec returned no error")
+	}
+	assertDead(t, pidFile)
+}
+
+// assertDead waits for the process whose PID is in file to be gone.
+func assertDead(t *testing.T, file string) {
+	t.Helper()
+	data, err := os.ReadFile(file)
+	if err != nil {
+		t.Fatalf("reading pid: %v", err)
+	}
+	pid, _ := strconv.Atoi(strings.TrimSpace(string(data)))
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		if err := syscall.Kill(pid, 0); errors.Is(err, syscall.ESRCH) {
+			return
+		}
+		if time.Now().After(deadline) {
+			syscall.Kill(pid, syscall.SIGKILL)
+			t.Fatalf("background process %d survived", pid)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}
+
+func TestExecRejects(t *testing.T) {
+	c, _ := start(t)
+	for name, req := range map[string]*opificav1.ExecRequest{
+		"empty":         {Command: " "},
+		"unknown shell": {Command: "x", Shell: "fish"},
+		"absolute dir":  {Command: "x", WorkingDirectory: "/etc"},
+		"escaping dir":  {Command: "x", WorkingDirectory: "../.."},
+		"missing dir":   {Command: "x", WorkingDirectory: "nope"},
+	} {
+		if r := run(t, c, req); connect.CodeOf(r.err) != connect.CodeInvalidArgument {
+			t.Errorf("%s: err = %v, want invalid argument", name, r.err)
+		}
+	}
+}
+
+// newRepo creates a git repository with one commit and returns its path and
+// the commit id.
+func newRepo(t *testing.T) (string, string) {
+	t.Helper()
+	dir := t.TempDir()
+	gitIn := func(args ...string) string {
+		cmd := exec.Command("git", args...)
+		cmd.Dir = dir
+		cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.com",
+			"GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.com", "GIT_CONFIG_GLOBAL="+os.DevNull)
+		out, err := cmd.CombinedOutput()
+		if err != nil {
+			t.Fatalf("git %v: %v: %s", args, err, out)
+		}
+		return strings.TrimSpace(string(out))
+	}
+	gitIn("init", "-q", "-b", "main")
+	os.WriteFile(filepath.Join(dir, "README"), []byte("hello\n"), 0o644)
+	gitIn("add", ".")
+	gitIn("commit", "-q", "-m", "first")
+	first := gitIn("rev-parse", "HEAD")
+	os.WriteFile(filepath.Join(dir, "README"), []byte("changed\n"), 0o644)
+	gitIn("commit", "-q", "-am", "second")
+	return dir, first
+}
+
+func TestCheckout(t *testing.T) {
+	if _, err := exec.LookPath("git"); err != nil {
+		t.Skip("no git")
+	}
+	c, root := start(t)
+	repo, first := newRepo(t)
+	ctx := context.Background()
+
+	// An older commit, not the branch head: the exact SHA is what's built.
+	resp, err := c.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{MirrorUrl: repo, CommitSha: first, Directory: "src"}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.Msg.GetHeadSha() != first {
+		t.Errorf("head = %s, want %s", resp.Msg.GetHeadSha(), first)
+	}
+	if data, _ := os.ReadFile(filepath.Join(root, "src", "README")); string(data) != "hello\n" {
+		t.Errorf("README = %q, want the first commit's", data)
+	}
+
+	for name, req := range map[string]*opificav1.CheckoutRequest{
+		"not empty":    {MirrorUrl: repo, CommitSha: first, Directory: "src"},
+		"short sha":    {MirrorUrl: repo, CommitSha: first[:8], Directory: "b"},
+		"escaping dir": {MirrorUrl: repo, CommitSha: first, Directory: "../x"},
+		"no mirror":    {CommitSha: first, Directory: "c"},
+	} {
+		if _, err := c.Checkout(ctx, connect.NewRequest(req)); connect.CodeOf(err) != connect.CodeInvalidArgument {
+			t.Errorf("%s: err = %v, want invalid argument", name, err)
+		}
+	}
+	missing := strings.Repeat("ab", 20)
+	if _, err := c.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{MirrorUrl: repo, CommitSha: missing, Directory: "d"})); connect.CodeOf(err) != connect.CodeFailedPrecondition {
+		t.Errorf("unknown commit: err = %v, want failed precondition", err)
+	}
+}
+
+func TestCollect(t *testing.T) {
+	c, root := start(t)
+	os.MkdirAll(filepath.Join(root, "src", "build"), 0o755)
+	big := strings.Repeat("x", 200<<10) // several chunks
+	os.WriteFile(filepath.Join(root, "src", "build", "a.xml"), []byte("<a/>"), 0o644)
+	os.WriteFile(filepath.Join(root, "src", "build", "b.xml"), []byte(big), 0o644)
+	os.WriteFile(filepath.Join(root, "src", "build", "c.txt"), []byte("no"), 0o644)
+	secret := filepath.Join(t.TempDir(), "secret")
+	os.WriteFile(secret, []byte("outside"), 0o644)
+	os.Symlink(secret, filepath.Join(root, "src", "build", "link.xml"))
+
+	stream, err := c.Collect(context.Background(), connect.NewRequest(&opificav1.CollectRequest{
+		BaseDirectory: "src", Patterns: []string{"build/*.xml", "build/a.xml"},
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	files := map[string]string{}
+	for stream.Receive() {
+		files[stream.Msg().GetPath()] += string(stream.Msg().GetData())
+	}
+	if err := stream.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if len(files) != 2 || files["build/a.xml"] != "<a/>" || files["build/b.xml"] != big {
+		t.Errorf("collected %d files: %v", len(files), keys(files))
+	}
+
+	bad, _ := c.Collect(context.Background(), connect.NewRequest(&opificav1.CollectRequest{Patterns: []string{"../../etc/passwd"}}))
+	for bad.Receive() {
+	}
+	if connect.CodeOf(bad.Err()) != connect.CodeInvalidArgument {
+		t.Errorf("escaping pattern: err = %v", bad.Err())
+	}
+}
+
+func keys(m map[string]string) []string {
+	var ks []string
+	for k := range m {
+		ks = append(ks, k)
+	}
+	return ks
+}
diff --git a/internal/agent/client.go b/internal/agent/client.go
new file mode 100644
index 0000000..830ed04
--- /dev/null
+++ b/internal/agent/client.go
@@ -0,0 +1,31 @@
+package agent
+
+import (
+	"context"
+	"net"
+	"net/http"
+
+	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+)
+
+// NewClient returns an AgentService client that reaches the agent through
+// dial, e.g. a VM's vsock connection. The channel is private to the host and
+// the guest, so it is plain HTTP/2 without TLS.
+func NewClient(dial func(ctx context.Context) (net.Conn, error)) opificav1connect.AgentServiceClient {
+	protocols := new(http.Protocols)
+	protocols.SetUnencryptedHTTP2(true)
+	transport := &http.Transport{
+		Protocols: protocols,
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return dial(ctx)
+		},
+	}
+	return opificav1connect.NewAgentServiceClient(&http.Client{Transport: transport}, "http://agent")
+}
+
+// Handler returns the agent's HTTP handler.
+func Handler(s *Server) http.Handler {
+	mux := http.NewServeMux()
+	mux.Handle(opificav1connect.NewAgentServiceHandler(s))
+	return mux
+}
diff --git a/internal/agent/proc_unix.go b/internal/agent/proc_unix.go
new file mode 100644
index 0000000..514668a
--- /dev/null
+++ b/internal/agent/proc_unix.go
@@ -0,0 +1,20 @@
+//go:build unix
+
+package agent
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+// killTree runs the command in its own process group and makes cancellation
+// kill the whole group, so background processes a step started die with it.
+func killTree(cmd *exec.Cmd) {
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		if cmd.Process == nil {
+			return nil
+		}
+		return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+	}
+}
diff --git a/internal/agent/proc_windows.go b/internal/agent/proc_windows.go
new file mode 100644
index 0000000..6023f00
--- /dev/null
+++ b/internal/agent/proc_windows.go
@@ -0,0 +1,19 @@
+//go:build windows
+
+package agent
+
+import "os/exec"
+
+// killTree kills the command on cancellation.
+//
+// TODO: put the process in a job object so its children are killed too. The
+// VM is destroyed after every job, so leftovers can't outlive the job, but
+// a later step in the same job could still see them.
+func killTree(cmd *exec.Cmd) {
+	cmd.Cancel = func() error {
+		if cmd.Process == nil {
+			return nil
+		}
+		return cmd.Process.Kill()
+	}
+}
diff --git a/internal/controller/logs/logs.go b/internal/controller/logs/logs.go
new file mode 100644
index 0000000..80345bb
--- /dev/null
+++ b/internal/controller/logs/logs.go
@@ -0,0 +1,101 @@
+// Package logs stores job output on the controller's disk, one file per job.
+//
+// Workers resend chunks after a reconnect; chunks are numbered per job, and
+// anything not newer than the last stored chunk is dropped. The last number
+// is kept in memory, so after a controller restart a resent chunk may appear
+// twice. S3-compatible storage is planned (docs/ROADMAP.md, M7).
+package logs
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/google/uuid"
+)
+
+// Store appends and reads job logs.
+type Store struct {
+	Dir string
+
+	mu   sync.Mutex
+	open map[uuid.UUID]*jobLog
+}
+
+type jobLog struct {
+	mu      sync.Mutex
+	f       *os.File
+	lastSeq uint64
+}
+
+// ErrNoLog means a job has no output yet.
+var ErrNoLog = errors.New("no log for this job yet")
+
+func (s *Store) path(job uuid.UUID) string {
+	return filepath.Join(s.Dir, job.String()+".log")
+}
+
+func (s *Store) get(job uuid.UUID) (*jobLog, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.open == nil {
+		s.open = map[uuid.UUID]*jobLog{}
+	}
+	if l, ok := s.open[job]; ok {
+		return l, nil
+	}
+	if err := os.MkdirAll(s.Dir, 0o750); err != nil {
+		return nil, err
+	}
+	f, err := os.OpenFile(s.path(job), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o640)
+	if err != nil {
+		return nil, err
+	}
+	l := &jobLog{f: f}
+	s.open[job] = l
+	return l, nil
+}
+
+// Append stores chunk seq of a job's log. It reports whether the chunk was
+// new.
+func (s *Store) Append(job uuid.UUID, seq uint64, data []byte) (bool, error) {
+	l, err := s.get(job)
+	if err != nil {
+		return false, err
+	}
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if seq <= l.lastSeq {
+		return false, nil
+	}
+	if _, err := l.f.Write(data); err != nil {
+		return false, fmt.Errorf("writing log: %w", err)
+	}
+	l.lastSeq = seq
+	return true, nil
+}
+
+// Finish closes a job's log file once the job is done.
+func (s *Store) Finish(job uuid.UUID) {
+	s.mu.Lock()
+	l, ok := s.open[job]
+	delete(s.open, job)
+	s.mu.Unlock()
+	if ok {
+		l.mu.Lock()
+		l.f.Close()
+		l.mu.Unlock()
+	}
+}
+
+// Open returns a job's log.
+func (s *Store) Open(job uuid.UUID) (io.ReadCloser, error) {
+	f, err := os.Open(s.path(job))
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, ErrNoLog
+	}
+	return f, err
+}
diff --git a/internal/controller/logs/logs_test.go b/internal/controller/logs/logs_test.go
new file mode 100644
index 0000000..344dff5
--- /dev/null
+++ b/internal/controller/logs/logs_test.go
@@ -0,0 +1,38 @@
+package logs
+
+import (
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func TestAppendDropsDuplicates(t *testing.T) {
+	s := &Store{Dir: t.TempDir()}
+	job := uuid.New()
+	if _, err := s.Open(job); !errors.Is(err, ErrNoLog) {
+		t.Errorf("Open before any output: err = %v", err)
+	}
+	for _, c := range []struct {
+		seq  uint64
+		data string
+		want bool
+	}{
+		{1, "a", true}, {2, "b", true}, {2, "b", false}, {1, "a", false}, {3, "c", true},
+	} {
+		if got, err := s.Append(job, c.seq, []byte(c.data)); err != nil || got != c.want {
+			t.Errorf("Append(%d) = %v, %v; want %v", c.seq, got, err, c.want)
+		}
+	}
+	s.Finish(job)
+	r, err := s.Open(job)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	data, _ := io.ReadAll(r)
+	if string(data) != "abc" {
+		t.Errorf("log = %q, want abc", data)
+	}
+}
diff --git a/internal/controller/scheduler/scheduler.go b/internal/controller/scheduler/scheduler.go
index d31a619..45e6a5f 100644
--- a/internal/controller/scheduler/scheduler.go
+++ b/internal/controller/scheduler/scheduler.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"log/slog"
 	mrand "math/rand/v2"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -27,6 +28,7 @@ import (
 	"google.golang.org/protobuf/types/known/durationpb"
 
 	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/store/db"
 	"github.com/opifica/opifica/internal/controller/workers"
@@ -47,6 +49,8 @@ type Config struct {
 	// Interval is how often the scheduler looks for work and expired leases
 	// even when nothing woke it.
 	Interval time.Duration
+	// Logs stores job output. Nil drops it.
+	Logs *logs.Store
 }
 
 // Scheduler implements workers.MessageHandler.
@@ -56,6 +60,11 @@ type Scheduler struct {
 	log      *slog.Logger
 	cfg      Config
 	kick     chan struct{}
+
+	// owners caches which worker holds a job, so log chunks don't each
+	// need a database query to check they come from the right worker.
+	ownersMu sync.Mutex
+	owners   map[uuid.UUID]uuid.UUID
 }
 
 var _ workers.MessageHandler = (*Scheduler)(nil)
@@ -71,7 +80,7 @@ func New(st *store.Store, sessions *workers.Registry, log *slog.Logger, cfg Conf
 	if log == nil {
 		log = slog.Default()
 	}
-	return &Scheduler{store: st, sessions: sessions, log: log, cfg: cfg, kick: make(chan struct{}, 1)}
+	return &Scheduler{store: st, sessions: sessions, log: log, cfg: cfg, kick: make(chan struct{}, 1), owners: map[uuid.UUID]uuid.UUID{}}
 }
 
 // Kick makes the scheduler look for work now.
@@ -397,6 +406,7 @@ func (s *Scheduler) reapLeases(ctx context.Context) {
 			continue
 		}
 		s.log.Warn("lease expired", "job", job.ID, "worker", e.WorkerID, "attempt", job.Attempt, "now", job.State)
+		s.released(job.ID)
 		if e.WorkerID != nil {
 			// In case the worker is alive but lost track of the job.
 			s.cancelOnWorker(ctx, *e.WorkerID, job.ID, "lease expired")
@@ -454,6 +464,9 @@ func (s *Scheduler) WorkerDisconnected(worker uuid.UUID) {
 		s.log.Error("releasing unstarted jobs", "worker", worker, "err", err)
 		return
 	}
+	for _, id := range ids {
+		s.released(id)
+	}
 	if len(ids) > 0 {
 		s.log.Info("worker disconnected before starting jobs; queued them again", "worker", worker, "jobs", len(ids))
 		s.Kick()
@@ -477,11 +490,53 @@ func (s *Scheduler) HandleWorkerMessage(ctx context.Context, worker uuid.UUID, m
 			},
 		}})
 	case *opificav1.ConnectRequest_LogChunk:
-		// Log storage comes next; drop for now.
+		s.logChunk(ctx, worker, m.LogChunk)
 	}
 	return nil
 }
 
+func (s *Scheduler) logChunk(ctx context.Context, worker uuid.UUID, c *opificav1.LogChunk) {
+	if s.cfg.Logs == nil {
+		return
+	}
+	jobID, err := uuid.Parse(c.GetJobId())
+	if err != nil || !s.holds(ctx, worker, jobID) {
+		return
+	}
+	if _, err := s.cfg.Logs.Append(jobID, c.GetSequence(), c.GetData()); err != nil {
+		s.log.Error("storing log", "job", jobID, "err", err)
+	}
+}
+
+// holds reports whether worker currently holds the job.
+func (s *Scheduler) holds(ctx context.Context, worker, job uuid.UUID) bool {
+	s.ownersMu.Lock()
+	owner, ok := s.owners[job]
+	s.ownersMu.Unlock()
+	if ok {
+		return owner == worker
+	}
+	j, err := s.store.GetJob(ctx, job)
+	if err != nil || j.WorkerID == nil || *j.WorkerID != worker || (j.State != "assigned" && j.State != "running") {
+		return false
+	}
+	s.ownersMu.Lock()
+	s.owners[job] = worker
+	s.ownersMu.Unlock()
+	return true
+}
+
+// released forgets a job's owner and closes its log once the worker no
+// longer holds it.
+func (s *Scheduler) released(job uuid.UUID) {
+	s.ownersMu.Lock()
+	delete(s.owners, job)
+	s.ownersMu.Unlock()
+	if s.cfg.Logs != nil {
+		s.cfg.Logs.Finish(job)
+	}
+}
+
 func (s *Scheduler) renew(ctx context.Context, worker uuid.UUID, ids []string) {
 	if len(ids) == 0 {
 		return
@@ -526,6 +581,7 @@ func (s *Scheduler) jobUpdate(ctx context.Context, worker uuid.UUID, u *opificav
 			return
 		}
 		log.Info("job finished", "key", job.Key, "state", state)
+		s.released(jobID)
 		s.advance(ctx, job.RunID)
 
 	case opificav1.JobState_JOB_STATE_ERRORED:
@@ -535,6 +591,7 @@ func (s *Scheduler) jobUpdate(ctx context.Context, worker uuid.UUID, u *opificav
 			return
 		}
 		log.Warn("job errored on the worker", "err", u.GetError(), "attempt", job.Attempt, "now", job.State)
+		s.released(jobID)
 		s.settled(ctx, job)
 
 	default:
diff --git a/internal/integration/connect_test.go b/internal/integration/connect_test.go
index a7d31e2..fde168a 100644
--- a/internal/integration/connect_test.go
+++ b/internal/integration/connect_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
 	"github.com/opifica/opifica/internal/controller/api"
 	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/logs"
 	"github.com/opifica/opifica/internal/controller/scheduler"
 	"github.com/opifica/opifica/internal/controller/store"
 	"github.com/opifica/opifica/internal/controller/workers"
@@ -42,6 +43,7 @@ type env struct {
 	t     *testing.T
 	svc   *workers.Service
 	sched *scheduler.Scheduler
+	logs  *logs.Store
 	url   string
 }
 
@@ -65,8 +67,9 @@ func newEnv(t *testing.T) *env {
 			TouchInterval:    50 * time.Millisecond,
 		},
 	}
+	logStore := &logs.Store{Dir: t.TempDir()}
 	sched := scheduler.New(st, svc.Sessions(), quiet(), scheduler.Config{
-		Lease: 400 * time.Millisecond, Interval: 20 * time.Millisecond,
+		Lease: 400 * time.Millisecond, Interval: 20 * time.Millisecond, Logs: logStore,
 	})
 	svc.Messages = sched
 	go sched.Run(ctx)
@@ -88,7 +91,7 @@ func newEnv(t *testing.T) *env {
 		<-done
 		st.Close()
 	})
-	return &env{t: t, svc: svc, sched: sched, url: "https://" + ln.Addr().String()}
+	return &env{t: t, svc: svc, sched: sched, logs: logStore, url: "https://" + ln.Addr().String()}
 }
 
 func info() *opificav1.WorkerInfo {
@@ -107,6 +110,11 @@ func (e *env) join(pool string) *worker.Identity {
 }
 
 func (e *env) joinWith(pool string, wi *opificav1.WorkerInfo) *worker.Identity {
+	e.t.Helper()
+	return e.joinInto(pool, wi, e.t.TempDir())
+}
+
+func (e *env) joinInto(pool string, wi *opificav1.WorkerInfo, stateDir string) *worker.Identity {
 	e.t.Helper()
 	tok, err := e.svc.CreateToken(context.Background(), workers.TokenRequest{Pool: pool})
 	if err != nil {
@@ -114,7 +122,7 @@ func (e *env) joinWith(pool string, wi *opificav1.WorkerInfo) *worker.Identity {
 	}
 	id, err := worker.Join(context.Background(), worker.JoinConfig{
 		ControllerURL: e.url, Token: tok.Secret, CAFingerprint: tok.CAFingerprint,
-		StateDir: e.t.TempDir(), Info: wi,
+		StateDir: stateDir, Info: wi,
 	})
 	if err != nil {
 		e.t.Fatalf("Join: %v", err)
diff --git a/internal/integration/e2e_test.go b/internal/integration/e2e_test.go
new file mode 100644
index 0000000..fe35276
--- /dev/null
+++ b/internal/integration/e2e_test.go
@@ -0,0 +1,335 @@
+package integration
+
+import (
+	"context"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/opifica/opifica/internal/controller/scheduler"
+	"github.com/opifica/opifica/internal/pipeline"
+	"github.com/opifica/opifica/pkg/worker"
+	"github.com/opifica/opifica/pkg/worker/workertest"
+)
+
+// These tests run the whole path for real: controller, scheduler, the worker
+// runtime, a Git mirror, and a real agent per job running real commands. Only
+// the VM is faked: each "VM" is a fresh directory with its own agent.
+
+func hostOS() string {
+	if runtime.GOOS == "darwin" {
+		return "macos"
+	}
+	return runtime.GOOS
+}
+
+type realWorker struct {
+	hv     *workertest.Hypervisor
+	cancel context.CancelFunc
+	done   chan error
+}
+
+func (e *env) realWorker(capacity int) *realWorker {
+	e.t.Helper()
+	if runtime.GOOS == "windows" {
+		e.t.Skip("the test pipelines use a POSIX shell")
+	}
+	if _, err := exec.LookPath("git"); err != nil {
+		e.t.Skip("git is not installed")
+	}
+	pool := worker.PoolConfig{
+		Capacity: capacity,
+		Images:   map[string]worker.Image{"default": {Name: "default"}},
+		Warm:     map[string]int{"default": 1},
+	}
+	dir := e.t.TempDir()
+	e.joinInto("trusted", worker.Info(pool, nil), dir)
+
+	rw := &realWorker{hv: &workertest.Hypervisor{Agent: true}, done: make(chan error, 1)}
+	ctx, cancel := context.WithCancel(context.Background())
+	rw.cancel = cancel
+	go func() {
+		rw.done <- worker.Run(ctx, worker.Config{
+			ControllerURL: e.url, StateDir: dir, Hypervisor: rw.hv, Pool: pool,
+			AllowLocalRepos: true, HeartbeatInterval: 50 * time.Millisecond, Logger: quiet(),
+		})
+	}()
+	e.t.Cleanup(func() {
+		cancel()
+		<-rw.done
+		if n := rw.hv.Alive(); n != 0 {
+			e.t.Errorf("%d VMs still alive after the worker stopped", n)
+		}
+	})
+	return rw
+}
+
+// gitRepo creates a repository with the given files and returns its path
+// and commit.
+func gitRepo(t *testing.T, files map[string]string) (string, string) {
+	t.Helper()
+	dir := t.TempDir()
+	for name, content := range files {
+		p := filepath.Join(dir, name)
+		os.MkdirAll(filepath.Dir(p), 0o755)
+		if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	gitIn := func(args ...string) string {
+		cmd := exec.Command("git", args...)
+		cmd.Dir = dir
+		cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@example.com",
+			"GIT_COMMITTER_NAME=t", "GIT_COMMITTER_EMAIL=t@example.com", "GIT_CONFIG_GLOBAL="+os.DevNull)
+		out, err := cmd.CombinedOutput()
+		if err != nil {
+			t.Fatalf("git %v: %v: %s", args, err, out)
+		}
+		return strings.TrimSpace(string(out))
+	}
+	gitIn("init", "-q", "-b", "main")
+	gitIn("add", ".")
+	gitIn("commit", "-q", "-m", "test")
+	return dir, gitIn("rev-parse", "HEAD")
+}
+
+func (e *env) startRun(repo, sha, src string) uuid.UUID {
+	e.t.Helper()
+	run, err := e.sched.CreateRun(context.Background(), runRequest(repo, sha, src))
+	if err != nil {
+		e.t.Fatalf("CreateRun: %v", err)
+	}
+	return run.ID
+}
+
+func (e *env) jobLog(runID uuid.UUID, key string) string {
+	e.t.Helper()
+	jobs, err := e.svc.Store.ListRunJobs(context.Background(), runID)
+	if err != nil {
+		e.t.Fatal(err)
+	}
+	for _, j := range jobs {
+		if j.Key == key {
+			r, err := e.logs.Open(j.ID)
+			if err != nil {
+				return ""
+			}
+			defer r.Close()
+			data, _ := io.ReadAll(r)
+			return string(data)
+		}
+	}
+	e.t.Fatalf("no job %q", key)
+	return ""
+}
+
+func wantIn(t *testing.T, log string, parts ...string) {
+	t.Helper()
+	for _, p := range parts {
+		if !strings.Contains(log, p) {
+			t.Errorf("log lacks %q:\n%s", p, log)
+		}
+	}
+}
+
+func TestE2ERealCommands(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(2)
+	repo, sha := gitRepo(t, map[string]string{"README": "hello world\n", "scripts/check.sh": "echo checked in $(pwd)\n"})
+
+	runID := e.startRun(repo, sha, `
+on: push
+env:
+  GREETING: hi
+jobs:
+  build:
+    runs-on: `+hostOS()+`
+    steps:
+      - run: |
+          echo "$GREETING, the readme says: $(cat README)"
+          test "$OPIFICA_COMMIT_SHA" = "`+sha+`"
+          test "$CI" = true
+      - run: echo to-stderr >&2
+      - run: sh check.sh
+        working-directory: scripts
+  check:
+    needs: build
+    runs-on: `+hostOS()+`
+    steps:
+      - name: allowed to fail
+        run: exit 7
+        continue-on-error: true
+      - run: echo still running
+      - report: junit build/*.xml
+`)
+	run, jobs := e.waitRun(runID)
+	if run.State != "succeeded" {
+		t.Fatalf("run %s; build %s (%q), check %s (%q)\n%s", run.State, jobs["build"].State, jobs["build"].Error,
+			jobs["check"].State, jobs["check"].Error, e.jobLog(runID, "build"))
+	}
+	wantIn(t, e.jobLog(runID, "build"),
+		"==> Fetching", "==> Starting a fresh VM", "==> Step 1/3",
+		"hi, the readme says: hello world", "to-stderr", "checked in ", "/src/scripts")
+	wantIn(t, e.jobLog(runID, "check"),
+		"Step failed with exit code 7; continuing", "still running", "Test reports are not stored yet")
+}
+
+func TestE2EFailure(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, `
+on: push
+jobs:
+  test:
+    runs-on: `+hostOS()+`
+    steps:
+      - run: echo before; exit 7
+      - run: echo never
+  deploy:
+    needs: test
+    runs-on: `+hostOS()+`
+    steps: [{ run: echo deploying }]
+`)
+	run, jobs := e.waitRun(runID)
+	if run.State != "failed" || jobs["test"].State != "failed" || *jobs["test"].ExitCode != 7 || jobs["deploy"].State != "skipped" {
+		t.Fatalf("run %s; test %s exit %v; deploy %s", run.State, jobs["test"].State, jobs["test"].ExitCode, jobs["deploy"].State)
+	}
+	if jobs["test"].Attempt != 1 {
+		t.Errorf("a failing command was retried (attempt %d); only infrastructure errors should be", jobs["test"].Attempt)
+	}
+	log := e.jobLog(runID, "test")
+	wantIn(t, log, "before", "Step failed with exit code 7")
+	if strings.Contains(log, "never") {
+		t.Error("ran the step after the failing one")
+	}
+}
+
+// Every job gets a fresh VM: nothing one job writes is visible to the next.
+func TestE2EFreshVMPerJob(t *testing.T) {
+	e := newEnv(t)
+	rw := e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, `
+on: push
+jobs:
+  first:
+    runs-on: `+hostOS()+`
+    steps: [{ run: "test ! -e ../left-behind && touch ../left-behind README.modified" }]
+  second:
+    needs: first
+    runs-on: `+hostOS()+`
+    steps: [{ run: "test ! -e ../left-behind && test ! -e README.modified" }]
+`)
+	run, jobs := e.waitRun(runID)
+	if run.State != "succeeded" {
+		t.Fatalf("run %s: first %s, second %s\n%s", run.State, jobs["first"].State, jobs["second"].State, e.jobLog(runID, "second"))
+	}
+	if created, _ := rw.hv.Counts(); created < 2 {
+		t.Errorf("only %d VMs created for 2 jobs", created)
+	}
+}
+
+func TestE2ESecretDenied(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, `
+on: push
+jobs:
+  deploy:
+    runs-on: `+hostOS()+`
+    secrets: [DEPLOY_TOKEN]
+    steps:
+      - run: echo deploying
+        secrets: [DEPLOY_TOKEN]
+`)
+	run, jobs := e.waitRun(runID)
+	j := jobs["deploy"]
+	if run.State != "failed" || !strings.Contains(j.Error, "secret DEPLOY_TOKEN was denied") || j.Attempt != 1 {
+		t.Errorf("run %s; job %s, attempt %d, error %q", run.State, j.State, j.Attempt, j.Error)
+	}
+	// "deploying" alone on a line is the command's output; the step's header
+	// line ("==> Step 1/1: echo deploying") doesn't count.
+	if regexp.MustCompile(`(?m)^deploying$`).MatchString(e.jobLog(runID, "deploy")) {
+		t.Error("the step ran without its secret")
+	}
+}
+
+func TestE2ECancelStopsTheCommand(t *testing.T) {
+	e := newEnv(t)
+	rw := e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, "on: push\njobs:\n  long:\n    runs-on: "+hostOS()+"\n    steps: [{ run: echo started; sleep 300 }]\n")
+	eventually(t, "the command to start", func() bool { return strings.Contains(e.jobLog(runID, "long"), "started") })
+
+	start := time.Now()
+	if err := e.sched.CancelRun(context.Background(), runID); err != nil {
+		t.Fatal(err)
+	}
+	run, jobs := e.waitRun(runID)
+	if run.State != "cancelled" || jobs["long"].State != "cancelled" {
+		t.Errorf("run %s, job %s", run.State, jobs["long"].State)
+	}
+	if d := time.Since(start); d > 5*time.Second {
+		t.Errorf("cancelling took %v", d)
+	}
+	eventually(t, "the job's VM to be destroyed", func() bool { return rw.hv.Alive() <= 1 }) // the warm one may remain
+}
+
+func TestE2EJobTimeout(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, "on: push\njobs:\n  slow:\n    runs-on: "+hostOS()+"\n    timeout: 2s\n    steps: [{ run: sleep 60 }]\n")
+	run, jobs := e.waitRun(runID)
+	j := jobs["slow"]
+	if run.State != "failed" || j.State != "failed" || !strings.Contains(j.Error+e.jobLog(runID, "slow"), "timed out") {
+		t.Errorf("run %s; job %s, error %q\n%s", run.State, j.State, j.Error, e.jobLog(runID, "slow"))
+	}
+}
+
+func TestE2EMissingCommitFailsWithoutRetry(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(1)
+	repo, _ := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, strings.Repeat("ab", 20), "on: push\njobs:\n  a:\n    runs-on: "+hostOS()+"\n    steps: [{ run: echo hi }]\n")
+	run, jobs := e.waitRun(runID)
+	if run.State != "failed" || jobs["a"].Attempt != 1 || !strings.Contains(jobs["a"].Error, "commit not found") {
+		t.Errorf("run %s; attempt %d; error %q", run.State, jobs["a"].Attempt, jobs["a"].Error)
+	}
+}
+
+func runRequest(repo, sha, src string) scheduler.RunRequest {
+	return scheduler.RunRequest{
+		Repo: "acme/app", CloneURL: repo, Pool: "trusted", Pipeline: []byte(src),
+		Event: pipeline.Event{Type: pipeline.EventPush, Branch: "main", Ref: "refs/heads/main", SHA: sha},
+	}
+}
+
+// The connection drops while a job runs. The job keeps running, the worker
+// reconnects within the lease, and the output and result arrive: no retry.
+func TestE2EJobSurvivesReconnect(t *testing.T) {
+	e := newEnv(t)
+	e.realWorker(1)
+	repo, sha := gitRepo(t, map[string]string{"README": "x"})
+	runID := e.startRun(repo, sha, "on: push\njobs:\n  a:\n    runs-on: "+hostOS()+"\n    steps: [{ run: echo before; sleep 1; echo after }]\n")
+	eventually(t, "the command to start", func() bool { return strings.Contains(e.jobLog(runID, "a"), "before") })
+
+	for _, s := range e.svc.Sessions().List() {
+		e.svc.Sessions().Disconnect(s.WorkerID, context.Canceled)
+	}
+	run, jobs := e.waitRun(runID)
+	if run.State != "succeeded" || jobs["a"].Attempt != 1 {
+		t.Fatalf("run %s, attempt %d (%q)", run.State, jobs["a"].Attempt, jobs["a"].Error)
+	}
+	wantIn(t, e.jobLog(runID, "a"), "before", "after")
+}
diff --git a/internal/mask/mask.go b/internal/mask/mask.go
new file mode 100644
index 0000000..1b13c6c
--- /dev/null
+++ b/internal/mask/mask.go
@@ -0,0 +1,133 @@
+// Package mask replaces secret values in streamed output.
+//
+// Output arrives in arbitrary chunks, so a secret can be split across two of
+// them. Writer holds back the tail of each chunk that could be the start of
+// a secret until the next chunk (or Flush) shows whether it is.
+package mask
+
+import (
+	"bytes"
+	"io"
+	"sort"
+	"sync"
+)
+
+// Replacement is written in place of every secret value.
+const Replacement = "***"
+
+// minLen is the shortest value that is masked. Shorter values ("1", "true")
+// would mangle ordinary output while protecting nothing.
+const minLen = 4
+
+// Writer masks secrets in everything written to it before passing it on.
+type Writer struct {
+	mu      sync.Mutex
+	out     io.Writer
+	secrets [][]byte // longest first, so overlapping secrets mask fully
+	maxLen  int
+	pending []byte
+}
+
+// NewWriter returns a Writer that masks secrets in what it writes to out.
+// Values shorter than 4 bytes are not masked.
+func NewWriter(out io.Writer, secrets []string) *Writer {
+	w := &Writer{out: out}
+	seen := map[string]bool{}
+	for _, s := range secrets {
+		if len(s) < minLen || seen[s] {
+			continue
+		}
+		seen[s] = true
+		w.secrets = append(w.secrets, []byte(s))
+		w.maxLen = max(w.maxLen, len(s))
+	}
+	sort.Slice(w.secrets, func(i, j int) bool { return len(w.secrets[i]) > len(w.secrets[j]) })
+	return w
+}
+
+// Write masks and forwards p, except for a tail that might be the start of a
+// secret, which waits for the next Write or Flush.
+func (w *Writer) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if len(w.secrets) == 0 {
+		return len(p), w.write(p)
+	}
+	buf := append(w.pending, p...)
+	masked, keep := w.mask(buf, false)
+	w.pending = append(w.pending[:0:0], buf[len(buf)-keep:]...)
+	return len(p), w.write(masked)
+}
+
+// Flush writes whatever is being held back. Call it when the stream ends.
+func (w *Writer) Flush() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if len(w.pending) == 0 {
+		return nil
+	}
+	masked, _ := w.mask(w.pending, true)
+	w.pending = nil
+	return w.write(masked)
+}
+
+func (w *Writer) write(p []byte) error {
+	if len(p) == 0 {
+		return nil
+	}
+	_, err := w.out.Write(p)
+	return err
+}
+
+// mask returns buf with secrets replaced, and how many bytes at the end of
+// buf were held back (always 0 when final).
+func (w *Writer) mask(buf []byte, final bool) ([]byte, int) {
+	var out bytes.Buffer
+	i := 0
+	for i < len(buf) {
+		if n := w.matchAt(buf[i:]); n > 0 {
+			out.WriteString(Replacement)
+			i += n
+			continue
+		}
+		if !final && w.couldStartSecret(buf[i:]) {
+			return out.Bytes(), len(buf) - i
+		}
+		out.WriteByte(buf[i])
+		i++
+	}
+	return out.Bytes(), 0
+}
+
+// matchAt returns the length of the secret that b starts with, or 0.
+func (w *Writer) matchAt(b []byte) int {
+	for _, s := range w.secrets {
+		if bytes.HasPrefix(b, s) {
+			return len(s)
+		}
+	}
+	return 0
+}
+
+// couldStartSecret reports whether b (the rest of the buffer) is a proper
+// prefix of some secret, i.e. more input could complete it.
+func (w *Writer) couldStartSecret(b []byte) bool {
+	if len(b) >= w.maxLen {
+		return false
+	}
+	for _, s := range w.secrets {
+		if len(b) < len(s) && bytes.HasPrefix(s, b) {
+			return true
+		}
+	}
+	return false
+}
+
+// String masks secrets in a complete string.
+func String(s string, secrets []string) string {
+	var b bytes.Buffer
+	w := NewWriter(&b, secrets)
+	w.Write([]byte(s))
+	w.Flush()
+	return b.String()
+}
diff --git a/internal/mask/mask_test.go b/internal/mask/mask_test.go
new file mode 100644
index 0000000..16aafa2
--- /dev/null
+++ b/internal/mask/mask_test.go
@@ -0,0 +1,69 @@
+package mask
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func TestString(t *testing.T) {
+	secrets := []string{"hunter2!", "s3cr3t-token", "abc"}
+	tests := map[string]string{
+		"password is hunter2!":           "password is ***",
+		"hunter2!hunter2!":               "******",
+		"token=s3cr3t-token;":            "token=***;",
+		"abc is too short to mask":       "abc is too short to mask",
+		"nothing here":                   "nothing here",
+		"almost hunter2 but not":         "almost hunter2 but not",
+		"ends with a prefix: s3cr3t-tok": "ends with a prefix: s3cr3t-tok",
+	}
+	for in, want := range tests {
+		if got := String(in, secrets); got != want {
+			t.Errorf("String(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+// Every possible split of the input into two writes must give the same result.
+func TestSecretSplitAcrossWrites(t *testing.T) {
+	secrets := []string{"correct-horse-battery", "horse-battery-staple"}
+	in := "a correct-horse-battery-staple b horse-battery-staple c correct-hors"
+	want := String(in, secrets)
+	if strings.Contains(want, "battery") && !strings.HasSuffix(want, "correct-hors") {
+		t.Fatalf("baseline masking failed: %q", want)
+	}
+	for i := 0; i <= len(in); i++ {
+		for j := i; j <= len(in); j++ {
+			var b bytes.Buffer
+			w := NewWriter(&b, secrets)
+			w.Write([]byte(in[:i]))
+			w.Write([]byte(in[i:j]))
+			w.Write([]byte(in[j:]))
+			w.Flush()
+			if b.String() != want {
+				t.Fatalf("split at %d,%d: got %q, want %q", i, j, b.String(), want)
+			}
+		}
+	}
+}
+
+func TestByteAtATime(t *testing.T) {
+	var b bytes.Buffer
+	w := NewWriter(&b, []string{"TOPSECRET"})
+	for _, c := range []byte("x TOPSECRET y TOPSECRE") {
+		w.Write([]byte{c})
+	}
+	w.Flush()
+	if got := b.String(); got != "x *** y TOPSECRE" {
+		t.Errorf("got %q", got)
+	}
+}
+
+func TestNoSecretsPassesThrough(t *testing.T) {
+	var b bytes.Buffer
+	w := NewWriter(&b, nil)
+	w.Write([]byte("hello"))
+	if b.String() != "hello" {
+		t.Errorf("got %q without Flush; nothing should be held back", b.String())
+	}
+}
diff --git a/internal/pipeline/parse.go b/internal/pipeline/parse.go
index f8226bf..6548a0f 100644
--- a/internal/pipeline/parse.go
+++ b/internal/pipeline/parse.go
@@ -45,6 +45,7 @@ func Parse(data []byte) (*Pipeline, error) {
 	if err := yaml.Unmarshal(data, &root); err != nil {
 		errs := convertYAMLError(err)
 		addFlowHint(errs, data)
+		addColonHint(errs)
 		return nil, errs
 	}
 
@@ -120,6 +121,16 @@ func addFlowHint(errs Errors, data []byte) {
 	}
 }
 
+// addColonHint explains "mapping values are not allowed": usually a command
+// or value containing ": ", which YAML reads as a nested key.
+func addColonHint(errs Errors) {
+	for _, e := range errs {
+		if strings.Contains(e.Msg, "mapping values are not allowed") && !strings.Contains(e.Msg, "hint:") {
+			e.Msg += ` (hint: a value containing ": " must be quoted, or written as a block, e.g. run: | on its own line with the command indented below)`
+		}
+	}
+}
+
 // addColumns fills in columns yaml.v3 leaves out of decode errors, using the
 // key on the reported line.
 func addColumns(errs Errors, root *yaml.Node) {
diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go
index 1366fc5..fce6b75 100644
--- a/internal/pipeline/pipeline_test.go
+++ b/internal/pipeline/pipeline_test.go
@@ -204,6 +204,8 @@ func TestErrors(t *testing.T) {
 			[]string{"this job has no matrix"}},
 		{"unknown event field", "on: push\njobs:\n  a:\n    runs-on: linux\n    steps:\n      - run: echo ${{ event.author }}\n",
 			[]string{`unknown event field "author"`}},
+		{"colon in a command", "on: push\njobs:\n  a:\n    runs-on: linux\n    steps:\n      - run: git log --format='%h: %s'\n",
+			[]string{`hint: a value containing ": " must be quoted`}},
 		{"expression in flow mapping", "on: push\njobs:\n  a:\n    matrix: { os: [linux] }\n    runs-on: { os: ${{ matrix.os }} }\n    steps: [{run: x}]\n",
 			[]string{"hint: ${{ }} can't be used inside inline"}},
 		{"quoted expression in flow mapping is fine", "on: push\njobs:\n  a:\n    matrix: { os: [linux] }\n    runs-on: { os: \"${{ matrix.os }}\", arch: sparc }\n    steps: [{run: x}]\n",
diff --git a/pkg/worker/client.go b/pkg/worker/client.go
index bcd40fd..c492928 100644
--- a/pkg/worker/client.go
+++ b/pkg/worker/client.go
@@ -273,9 +273,11 @@ func (c *Client) Run(ctx context.Context) error {
 	}
 }
 
-// backoff is exponential from 500ms with ±25% jitter, capped at max.
+// backoff is exponential from 100ms with ±25% jitter, capped at max. The
+// first retry is quick because most drops are momentary, and a worker that
+// is back within its lease keeps its running jobs.
 func backoff(attempt int, max time.Duration) time.Duration {
-	d := time.Duration(float64(500*time.Millisecond) * math.Pow(2, float64(min(attempt, 20))))
+	d := time.Duration(float64(100*time.Millisecond) * math.Pow(2, float64(min(attempt, 20))))
 	d = min(d, max)
 	return time.Duration(float64(d) * (0.75 + 0.5*mrand.Float64()))
 }
diff --git a/pkg/worker/mirror.go b/pkg/worker/mirror.go
new file mode 100644
index 0000000..8e2c2c4
--- /dev/null
+++ b/pkg/worker/mirror.go
@@ -0,0 +1,160 @@
+package worker
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"sync"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+)
+
+// Mirrors keeps a bare mirror of each repository on the worker's disk, so
+// jobs check out from a local copy: fast, light on the forge, and the VM
+// never needs access to the forge itself.
+type Mirrors struct {
+	// Dir holds the mirrors.
+	Dir string
+	// AllowLocal permits local paths and file:// URLs as clone URLs. Only for
+	// development: a controller could otherwise make the worker read any
+	// repository on its disk.
+	AllowLocal bool
+
+	mu    sync.Mutex
+	locks map[string]*sync.Mutex
+}
+
+// ErrCommitNotFound means the repository doesn't contain the commit.
+var ErrCommitNotFound = errors.New("commit not found in the repository")
+
+var commitRe = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
+
+func (m *Mirrors) lock(key string) *sync.Mutex {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.locks == nil {
+		m.locks = map[string]*sync.Mutex{}
+	}
+	l, ok := m.locks[key]
+	if !ok {
+		l = &sync.Mutex{}
+		m.locks[key] = l
+	}
+	return l
+}
+
+func (m *Mirrors) checkURL(u string) error {
+	switch {
+	case strings.HasPrefix(u, "https://"), strings.HasPrefix(u, "ssh://"):
+		return nil
+	case regexp.MustCompile(`^[\w.-]+@[\w.-]+:`).MatchString(u): // git@host:org/repo
+		return nil
+	case strings.HasPrefix(u, "http://"):
+		return fmt.Errorf("clone URL %q: plain http is not allowed", u)
+	case m.AllowLocal:
+		return nil
+	}
+	return fmt.Errorf("clone URL %q: local repositories are not allowed on this worker", u)
+}
+
+// Ensure makes sure the mirror of cloneURL contains sha and returns the
+// mirror's path. A commit already in the mirror is used without fetching:
+// commits never change.
+func (m *Mirrors) Ensure(ctx context.Context, cloneURL string, cred *opificav1.Credential, sha string) (string, error) {
+	if !commitRe.MatchString(sha) {
+		return "", fmt.Errorf("invalid commit id %q", sha)
+	}
+	if err := m.checkURL(cloneURL); err != nil {
+		return "", err
+	}
+	env, cleanup, err := credentialEnv(cred)
+	if err != nil {
+		return "", err
+	}
+	defer cleanup()
+
+	sum := sha256.Sum256([]byte(cloneURL))
+	key := hex.EncodeToString(sum[:12])
+	path := filepath.Join(m.Dir, key+".git")
+	l := m.lock(key)
+	l.Lock()
+	defer l.Unlock()
+
+	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
+		if err := os.MkdirAll(m.Dir, 0o700); err != nil {
+			return "", err
+		}
+		tmp, err := os.MkdirTemp(m.Dir, key+".tmp-")
+		if err != nil {
+			return "", err
+		}
+		defer os.RemoveAll(tmp)
+		if _, err := runGit(ctx, env, "clone", "--quiet", "--mirror", "--", cloneURL, tmp); err != nil {
+			return "", err
+		}
+		if err := os.Rename(tmp, path); err != nil {
+			return "", err
+		}
+	}
+
+	if hasCommit(ctx, path, sha) {
+		return path, nil
+	}
+	if _, err := runGit(ctx, env, "-C", path, "fetch", "--quiet", "--prune", "origin"); err != nil {
+		return "", err
+	}
+	if hasCommit(ctx, path, sha) {
+		return path, nil
+	}
+	// Not on any branch or tag (e.g. a pull request ref): ask for it directly.
+	if _, err := runGit(ctx, env, "-C", path, "fetch", "--quiet", "origin", sha); err == nil && hasCommit(ctx, path, sha) {
+		return path, nil
+	}
+	return "", fmt.Errorf("%w: %s", ErrCommitNotFound, sha)
+}
+
+func hasCommit(ctx context.Context, repo, sha string) bool {
+	_, err := runGit(ctx, nil, "-C", repo, "cat-file", "-e", sha+"^{commit}")
+	return err == nil
+}
+
+// credentialEnv passes a token to git through the environment rather than
+// the command line, where other users of the host could see it.
+func credentialEnv(cred *opificav1.Credential) ([]string, func(), error) {
+	noop := func() {}
+	switch k := cred.GetKind().(type) {
+	case nil:
+		return nil, noop, nil
+	case *opificav1.Credential_Token:
+		basic := base64.StdEncoding.EncodeToString([]byte("x-access-token:" + k.Token))
+		return []string{
+			"GIT_CONFIG_COUNT=1",
+			"GIT_CONFIG_KEY_0=http.extraHeader",
+			"GIT_CONFIG_VALUE_0=Authorization: Basic " + basic,
+		}, noop, nil
+	case *opificav1.Credential_SshPrivateKey:
+		return nil, noop, errors.New("SSH key credentials are not supported yet")
+	}
+	return nil, noop, fmt.Errorf("unknown credential type %T", cred.GetKind())
+}
+
+func runGit(ctx context.Context, env []string, args ...string) (string, error) {
+	cmd := exec.CommandContext(ctx, "git", args...)
+	cmd.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL="+os.DevNull)
+	cmd.Env = append(cmd.Env, env...)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &stdout, &stderr
+	if err := cmd.Run(); err != nil {
+		return "", fmt.Errorf("git %s: %v: %s", args[0], err, strings.TrimSpace(stderr.String()))
+	}
+	return stdout.String(), nil
+}
diff --git a/pkg/worker/runner.go b/pkg/worker/runner.go
new file mode 100644
index 0000000..0e3dc9b
--- /dev/null
+++ b/pkg/worker/runner.go
@@ -0,0 +1,431 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"maps"
+	"math"
+	"net"
+	"slices"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"connectrpc.com/connect"
+	"github.com/google/uuid"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/internal/agent"
+)
+
+// checkoutDir is where the agent checks out the repository, relative to its
+// work directory. Step working directories are relative to it.
+const checkoutDir = "src"
+
+// RunnerConfig configures a Runner.
+type RunnerConfig struct {
+	Pool    *Pool
+	Mirrors *Mirrors
+	// Send delivers a message to the controller. It should keep trying
+	// across reconnects; see NewOutbox.
+	Send func(msg *opificav1.ConnectRequest, onSent func())
+	// GuestMirrorURL turns the mirror's path on the host into the URL the
+	// agent in the VM clones from. Hypervisor drivers expose the mirror to
+	// the guest (e.g. over vsock); the default, the host path itself, only
+	// works when the agent runs on the host (development).
+	GuestMirrorURL func(vm VM, hostPath string) (string, error)
+	// SecretTimeout bounds waiting for the controller to answer a secret
+	// request. Defaults to 30s.
+	SecretTimeout time.Duration
+	Logger        *slog.Logger
+}
+
+// Runner runs the jobs the controller assigns, each in a fresh VM.
+type Runner struct {
+	cfg RunnerConfig
+	log *slog.Logger
+
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+
+	mu       sync.Mutex
+	jobs     map[string]*runningJob
+	secrets  map[string]chan *opificav1.SecretResponse
+	draining bool
+}
+
+type runningJob struct {
+	spec   *opificav1.Job
+	cancel context.CancelCauseFunc
+	seq    atomic.Uint64
+}
+
+var errCancelledByController = errors.New("cancelled by the controller")
+
+// NewRunner creates a runner. Close it to stop every job.
+func NewRunner(cfg RunnerConfig) *Runner {
+	if cfg.GuestMirrorURL == nil {
+		cfg.GuestMirrorURL = func(_ VM, p string) (string, error) { return p, nil }
+	}
+	if cfg.SecretTimeout <= 0 {
+		cfg.SecretTimeout = 30 * time.Second
+	}
+	log := cfg.Logger
+	if log == nil {
+		log = slog.Default()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	return &Runner{
+		cfg: cfg, log: log, ctx: ctx, cancel: cancel,
+		jobs: map[string]*runningJob{}, secrets: map[string]chan *opificav1.SecretResponse{},
+	}
+}
+
+// Running returns the IDs of jobs this worker holds, for heartbeats. A job
+// stays here until its final result has been delivered, so its lease is
+// renewed until then.
+func (r *Runner) Running() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return slices.Sorted(maps.Keys(r.jobs))
+}
+
+// Close cancels every job and waits for them to finish.
+func (r *Runner) Close() {
+	r.cancel()
+	r.wg.Wait()
+}
+
+// HandleMessage handles a message from the controller.
+func (r *Runner) HandleMessage(_ context.Context, msg *opificav1.ConnectResponse) {
+	switch m := msg.GetMessage().(type) {
+	case *opificav1.ConnectResponse_AssignJob:
+		r.start(m.AssignJob.GetJob())
+	case *opificav1.ConnectResponse_CancelJob:
+		r.mu.Lock()
+		j := r.jobs[m.CancelJob.GetJobId()]
+		r.mu.Unlock()
+		if j != nil {
+			r.log.Info("job cancelled by the controller", "job", m.CancelJob.GetJobId(), "reason", m.CancelJob.GetReason())
+			j.cancel(errCancelledByController)
+		}
+	case *opificav1.ConnectResponse_SecretResponse:
+		r.mu.Lock()
+		ch := r.secrets[m.SecretResponse.GetRequestId()]
+		delete(r.secrets, m.SecretResponse.GetRequestId())
+		r.mu.Unlock()
+		if ch != nil {
+			ch <- m.SecretResponse
+		}
+	case *opificav1.ConnectResponse_Drain:
+		r.mu.Lock()
+		r.draining = true
+		r.mu.Unlock()
+		r.log.Info("draining: finishing running jobs, accepting no new ones", "reason", m.Drain.GetReason())
+	}
+}
+
+func (r *Runner) start(spec *opificav1.Job) {
+	id := spec.GetId()
+	r.mu.Lock()
+	if _, dup := r.jobs[id]; dup || r.draining || r.ctx.Err() != nil {
+		r.mu.Unlock()
+		if !dup {
+			// Not running it: let the controller give it to someone else.
+			r.cfg.Send(update(id, opificav1.JobState_JOB_STATE_ERRORED, nil, "worker is shutting down or draining"), nil)
+		}
+		return
+	}
+	ctx, cancel := context.WithCancelCause(r.ctx)
+	j := &runningJob{spec: spec, cancel: cancel}
+	r.jobs[id] = j
+	r.mu.Unlock()
+
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		defer cancel(nil)
+		state, exit, msg := r.run(ctx, j)
+		r.log.Info("job finished", "job", id, "state", state, "exit", exit, "error", msg)
+		var exitPtr *int32
+		if state == opificav1.JobState_JOB_STATE_SUCCEEDED || state == opificav1.JobState_JOB_STATE_FAILED {
+			exitPtr = &exit
+		}
+		// Forget the job only once the controller has its result.
+		r.cfg.Send(update(id, state, exitPtr, msg), func() {
+			r.mu.Lock()
+			delete(r.jobs, id)
+			r.mu.Unlock()
+		})
+	}()
+}
+
+func update(job string, state opificav1.JobState, exit *int32, msg string) *opificav1.ConnectRequest {
+	return &opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_JobUpdate{JobUpdate: &opificav1.JobUpdate{
+		JobId: job, State: state, ExitCode: exit, Error: msg,
+	}}}
+}
+
+// syslog writes an Opifica line into the job's log.
+func (r *Runner) syslog(j *runningJob, step uint32, format string, args ...any) {
+	r.logChunk(j, step, opificav1.LogStream_LOG_STREAM_SYSTEM, []byte("==> "+fmt.Sprintf(format, args...)+"\n"))
+}
+
+func (r *Runner) logChunk(j *runningJob, step uint32, stream opificav1.LogStream, data []byte) {
+	r.cfg.Send(&opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_LogChunk{LogChunk: &opificav1.LogChunk{
+		JobId: j.spec.GetId(), StepIndex: step, Sequence: j.seq.Add(1), Stream: stream, Data: data,
+	}}}, nil)
+}
+
+// jobError is a failure with the state it should be reported as.
+type jobError struct {
+	state opificav1.JobState
+	msg   string
+}
+
+func failed(format string, args ...any) *jobError {
+	return &jobError{opificav1.JobState_JOB_STATE_FAILED, fmt.Sprintf(format, args...)}
+}
+
+func errored(format string, args ...any) *jobError {
+	return &jobError{opificav1.JobState_JOB_STATE_ERRORED, fmt.Sprintf(format, args...)}
+}
+
+// run executes a job and returns how it ended. Errors in the infrastructure
+// (VM, agent, mirror) are reported as ERRORED so the controller retries the
+// job elsewhere; problems with the job itself are FAILED.
+func (r *Runner) run(ctx context.Context, j *runningJob) (opificav1.JobState, int32, string) {
+	spec := j.spec
+	if t := spec.GetTimeout().AsDuration(); t > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeoutCause(ctx, t, fmt.Errorf("job timed out after %v", t))
+		defer cancel()
+	}
+	exit, jerr := r.execute(ctx, j)
+	switch {
+	case errors.Is(context.Cause(ctx), errCancelledByController):
+		return opificav1.JobState_JOB_STATE_CANCELLED, 0, "cancelled"
+	case r.ctx.Err() != nil:
+		return opificav1.JobState_JOB_STATE_ERRORED, 0, "worker shut down during the job"
+	case ctx.Err() != nil:
+		msg := context.Cause(ctx).Error()
+		r.syslog(j, 0, "%s", msg)
+		return opificav1.JobState_JOB_STATE_FAILED, exit, msg
+	case jerr != nil:
+		r.syslog(j, 0, "%s", jerr.msg)
+		return jerr.state, exit, jerr.msg
+	case exit != 0:
+		return opificav1.JobState_JOB_STATE_FAILED, exit, ""
+	}
+	return opificav1.JobState_JOB_STATE_SUCCEEDED, 0, ""
+}
+
+func (r *Runner) execute(ctx context.Context, j *runningJob) (int32, *jobError) {
+	spec := j.spec
+	src := spec.GetSource()
+	r.cfg.Send(update(spec.GetId(), opificav1.JobState_JOB_STATE_PREPARING, nil, ""), nil)
+
+	r.syslog(j, 0, "Fetching %s at %s", src.GetCloneUrl(), short(src.GetCommitSha()))
+	mirror, err := r.cfg.Mirrors.Ensure(ctx, src.GetCloneUrl(), src.GetCredential(), src.GetCommitSha())
+	switch {
+	case errors.Is(err, ErrCommitNotFound):
+		return 0, failed("%v", err)
+	case err != nil:
+		return 0, errored("fetching the repository: %v", err)
+	}
+
+	r.syslog(j, 0, "Starting a fresh VM (image %s)", spec.GetImage())
+	lease, err := r.cfg.Pool.Acquire(ctx, spec.GetImage())
+	if err != nil {
+		return 0, errored("starting a VM: %v", err)
+	}
+	defer func() {
+		if err := lease.Release(context.WithoutCancel(ctx)); err != nil {
+			r.log.Error("destroying job VM", "job", spec.GetId(), "err", err)
+		}
+	}()
+
+	client := agent.NewClient(func(ctx context.Context) (net.Conn, error) { return lease.VM.DialAgent(ctx) })
+	if _, err := client.Ready(ctx, connect.NewRequest(&opificav1.ReadyRequest{})); err != nil {
+		return 0, errored("the VM's agent is not responding: %v", err)
+	}
+	guestURL, err := r.cfg.GuestMirrorURL(lease.VM, mirror)
+	if err != nil {
+		return 0, errored("exposing the repository to the VM: %v", err)
+	}
+	if _, err := client.Checkout(ctx, connect.NewRequest(&opificav1.CheckoutRequest{
+		MirrorUrl: guestURL, CommitSha: src.GetCommitSha(), Directory: checkoutDir,
+	})); err != nil {
+		return 0, errored("checking out %s: %v", short(src.GetCommitSha()), err)
+	}
+
+	r.cfg.Send(update(spec.GetId(), opificav1.JobState_JOB_STATE_RUNNING, nil, ""), nil)
+	steps := spec.GetSteps()
+	var failedExit int32
+	for i, step := range steps {
+		idx := uint32(i)
+		name := stepName(step)
+		r.syslog(j, idx, "Step %d/%d: %s", i+1, len(steps), name)
+		exit, jerr := r.step(ctx, j, client, idx, step)
+		if jerr != nil {
+			return exit, jerr
+		}
+		if ctx.Err() != nil {
+			return exit, nil
+		}
+		if exit != 0 {
+			if step.GetContinueOnError() {
+				r.syslog(j, idx, "Step failed with exit code %d; continuing (continue-on-error)", exit)
+				continue
+			}
+			r.syslog(j, idx, "Step failed with exit code %d", exit)
+			failedExit = exit
+			break
+		}
+	}
+	return failedExit, nil
+}
+
+func (r *Runner) step(ctx context.Context, j *runningJob, client agentClient, idx uint32, step *opificav1.Step) (int32, *jobError) {
+	switch a := step.GetAction().(type) {
+	case *opificav1.Step_Run:
+		return r.runStep(ctx, j, client, idx, step, a.Run)
+	case *opificav1.Step_Report:
+		r.syslog(j, idx, "Test reports are not stored yet; skipping")
+	case *opificav1.Step_Upload:
+		r.syslog(j, idx, "Artifact uploads are not stored yet; skipping")
+	case *opificav1.Step_Sign:
+		r.syslog(j, idx, "The signing service is not available yet; skipping")
+	default:
+		return 0, failed("step %d has no action", idx+1)
+	}
+	return 0, nil
+}
+
+type agentClient interface {
+	Exec(context.Context, *connect.Request[opificav1.ExecRequest]) (*connect.ServerStreamForClient[opificav1.ExecResponse], error)
+}
+
+func (r *Runner) runStep(ctx context.Context, j *runningJob, client agentClient, idx uint32, step *opificav1.Step, run *opificav1.RunAction) (int32, *jobError) {
+	spec := j.spec
+	env := map[string]string{
+		"CI": "true", "OPIFICA": "true",
+		"OPIFICA_JOB_ID": spec.GetId(), "OPIFICA_RUN_ID": spec.GetRunId(),
+		"OPIFICA_COMMIT_SHA": spec.GetSource().GetCommitSha(), "OPIFICA_REF": spec.GetSource().GetRef(),
+	}
+	maps.Copy(env, spec.GetEnv())
+	maps.Copy(env, step.GetEnv())
+
+	var masks []string
+	for _, name := range step.GetSecrets() {
+		value, denied, err := r.secret(ctx, spec, name)
+		if err != nil {
+			return 0, errored("requesting secret %s: %v", name, err)
+		}
+		if denied != "" {
+			return 0, failed("secret %s was denied: %s", name, denied)
+		}
+		env[name] = value
+		masks = append(masks, value)
+	}
+
+	timeout := step.GetTimeout().AsDuration()
+	if dl, ok := ctx.Deadline(); ok && (timeout == 0 || time.Until(dl) < timeout) {
+		timeout = time.Until(dl)
+	}
+	wd := checkoutDir
+	if d := run.GetWorkingDirectory(); d != "" {
+		wd = checkoutDir + "/" + strings.TrimPrefix(d, "./")
+	}
+	stream, err := client.Exec(ctx, connect.NewRequest(&opificav1.ExecRequest{
+		Command: run.GetCommand(), Shell: run.GetShell(), WorkingDirectory: wd,
+		Env: env, Mask: masks, TimeoutSeconds: uint32(math.Ceil(timeout.Seconds())),
+	}))
+	if err != nil {
+		return 0, errored("starting the step: %v", err)
+	}
+	defer stream.Close()
+	exit := int32(-1)
+	for stream.Receive() {
+		switch e := stream.Msg().GetEvent().(type) {
+		case *opificav1.ExecResponse_Stdout:
+			r.logChunk(j, idx, opificav1.LogStream_LOG_STREAM_STDOUT, e.Stdout)
+		case *opificav1.ExecResponse_Stderr:
+			r.logChunk(j, idx, opificav1.LogStream_LOG_STREAM_STDERR, e.Stderr)
+		case *opificav1.ExecResponse_ExitCode:
+			exit = e.ExitCode
+		}
+	}
+	if ctx.Err() != nil {
+		return 0, nil
+	}
+	if err := stream.Err(); err != nil {
+		var cerr *connect.Error
+		if errors.As(err, &cerr) && cerr.Code() == connect.CodeInvalidArgument {
+			return 0, failed("step %d: %s", idx+1, cerr.Message())
+		}
+		return 0, errored("running the step: %v", err)
+	}
+	if exit < 0 {
+		return 0, errored("the agent did not report an exit code")
+	}
+	return exit, nil
+}
+
+// secret asks the controller for a secret on the job's behalf.
+func (r *Runner) secret(ctx context.Context, spec *opificav1.Job, name string) (value, denied string, err error) {
+	reqID := uuid.NewString()
+	ch := make(chan *opificav1.SecretResponse, 1)
+	r.mu.Lock()
+	r.secrets[reqID] = ch
+	r.mu.Unlock()
+	defer func() {
+		r.mu.Lock()
+		delete(r.secrets, reqID)
+		r.mu.Unlock()
+	}()
+
+	r.cfg.Send(&opificav1.ConnectRequest{Message: &opificav1.ConnectRequest_SecretRequest{SecretRequest: &opificav1.SecretRequest{
+		RequestId: reqID, JobId: spec.GetId(), JobToken: spec.GetJobToken(), Name: name,
+	}}}, nil)
+	select {
+	case resp := <-ch:
+		if d, ok := resp.GetResult().(*opificav1.SecretResponse_DeniedReason); ok {
+			return "", d.DeniedReason, nil
+		}
+		return string(resp.GetValue()), "", nil
+	case <-time.After(r.cfg.SecretTimeout):
+		return "", "", errors.New("no answer from the controller")
+	case <-ctx.Done():
+		return "", "", ctx.Err()
+	}
+}
+
+func stepName(s *opificav1.Step) string {
+	if s.GetName() != "" {
+		return s.GetName()
+	}
+	switch a := s.GetAction().(type) {
+	case *opificav1.Step_Run:
+		line, _, _ := strings.Cut(strings.TrimSpace(a.Run.GetCommand()), "\n")
+		return line
+	case *opificav1.Step_Report:
+		return "report " + strings.Join(a.Report.GetPaths(), " ")
+	case *opificav1.Step_Upload:
+		return "upload " + strings.Join(a.Upload.GetPaths(), " ")
+	case *opificav1.Step_Sign:
+		return "sign " + strings.Join(a.Sign.GetPaths(), " ")
+	}
+	return "step"
+}
+
+func short(sha string) string {
+	if len(sha) > 12 {
+		return sha[:12]
+	}
+	return sha
+}
diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go
new file mode 100644
index 0000000..b94a989
--- /dev/null
+++ b/pkg/worker/worker.go
@@ -0,0 +1,206 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"maps"
+	"os"
+	"path/filepath"
+	"runtime"
+	"slices"
+	"time"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
+	"github.com/opifica/opifica/internal/version"
+)
+
+// Config is everything an OS-specific worker passes to Run.
+type Config struct {
+	ControllerURL string
+	// StateDir holds the identity from Join and the repository mirrors.
+	StateDir   string
+	Hypervisor Hypervisor
+	// Pool sizes the VM pool. Its Capacity is how many jobs run at once.
+	Pool PoolConfig
+	// Labels are reported to the controller for scheduling. "pool" is set
+	// by the join token and can't be chosen here.
+	Labels map[string]string
+	// GuestMirrorURL: see RunnerConfig.
+	GuestMirrorURL func(vm VM, hostPath string) (string, error)
+	// AllowLocalRepos: see Mirrors.AllowLocal. Development only.
+	AllowLocalRepos bool
+	// HeartbeatInterval: see ClientConfig.
+	HeartbeatInterval time.Duration
+	Logger            *slog.Logger
+}
+
+// Info describes this machine for Join and Hello.
+func Info(pool PoolConfig, labels map[string]string) *opificav1.WorkerInfo {
+	host, _ := os.Hostname()
+	info := &opificav1.WorkerInfo{
+		Hostname:          host,
+		Labels:            labels,
+		MaxConcurrentJobs: uint32(pool.Capacity),
+		Version:           version.Version,
+	}
+	switch runtime.GOOS {
+	case "linux":
+		info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_LINUX
+	case "windows":
+		info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_WINDOWS
+	case "darwin":
+		info.Os = opificav1.OperatingSystem_OPERATING_SYSTEM_MACOS
+	}
+	switch runtime.GOARCH {
+	case "amd64":
+		info.Arch = opificav1.Architecture_ARCHITECTURE_AMD64
+	case "arm64":
+		info.Arch = opificav1.Architecture_ARCHITECTURE_ARM64
+	}
+	for _, name := range slices.Sorted(maps.Keys(pool.Images)) {
+		info.Images = append(info.Images, &opificav1.ImageRef{Name: name, Digest: pool.Images[name].Digest})
+	}
+	return info
+}
+
+// Run connects to the controller and runs jobs until ctx is cancelled or the
+// controller rejects the worker for good. The worker must have joined
+// (see Join) with the same StateDir.
+func Run(ctx context.Context, cfg Config) error {
+	log := cfg.Logger
+	if log == nil {
+		log = slog.Default()
+	}
+	id, err := LoadIdentity(cfg.StateDir)
+	if err != nil {
+		return err
+	}
+	if cfg.Hypervisor == nil {
+		return errors.New("a hypervisor is required")
+	}
+	cfg.Pool.Logger = log
+	pool, err := NewPool(cfg.Hypervisor, cfg.Pool)
+	if err != nil {
+		return err
+	}
+	defer pool.Close(context.Background())
+
+	var outbox *Outbox
+	runner := NewRunner(RunnerConfig{
+		Pool:           pool,
+		Mirrors:        &Mirrors{Dir: filepath.Join(cfg.StateDir, "mirrors"), AllowLocal: cfg.AllowLocalRepos},
+		Send:           func(m *opificav1.ConnectRequest, onSent func()) { outbox.Send(m, onSent) },
+		GuestMirrorURL: cfg.GuestMirrorURL,
+		Logger:         log,
+	})
+
+	client, err := NewClient(ClientConfig{
+		ControllerURL:     cfg.ControllerURL,
+		Identity:          id,
+		Info:              Info(cfg.Pool, cfg.Labels),
+		HeartbeatInterval: cfg.HeartbeatInterval,
+		Logger:            log,
+		Status: func() (uint32, []string) {
+			warm := 0
+			for _, n := range pool.Stats().Warm {
+				warm += n
+			}
+			return uint32(warm), runner.Running()
+		},
+		OnMessage: runner.HandleMessage,
+	})
+	if err != nil {
+		return err
+	}
+	outbox = NewOutbox(client, 4096)
+
+	outboxCtx, stopOutbox := context.WithCancel(context.Background())
+	outboxDone := make(chan struct{})
+	go func() { outbox.Run(outboxCtx); close(outboxDone) }()
+
+	err = client.Run(ctx)
+
+	// Stop jobs, then give their final updates a moment to go out (they
+	// won't if the connection is gone; the controller's leases cover that).
+	runner.Close()
+	time.AfterFunc(5*time.Second, stopOutbox)
+	outbox.Close()
+	<-outboxDone
+	stopOutbox()
+	return err
+}
+
+// Outbox delivers messages to the controller in order, waiting through
+// reconnects. Job updates and logs aren't lost when the connection drops
+// briefly: they are sent once it is back.
+type Outbox struct {
+	client *Client
+	ch     chan outboxItem
+	closed chan struct{}
+}
+
+type outboxItem struct {
+	msg    *opificav1.ConnectRequest
+	onSent func()
+}
+
+// NewOutbox creates an outbox holding up to size messages. When it is full,
+// Send blocks, which slows job output down instead of using unbounded memory.
+func NewOutbox(c *Client, size int) *Outbox {
+	return &Outbox{client: c, ch: make(chan outboxItem, size), closed: make(chan struct{})}
+}
+
+// Send queues msg. onSent, if set, runs after it was handed to the stream.
+func (o *Outbox) Send(msg *opificav1.ConnectRequest, onSent func()) {
+	select {
+	case o.ch <- outboxItem{msg, onSent}:
+	case <-o.closed:
+	}
+}
+
+// Close stops accepting messages; Run delivers what is queued, then returns.
+func (o *Outbox) Close() {
+	select {
+	case <-o.closed:
+	default:
+		close(o.closed)
+	}
+}
+
+// Run sends queued messages until Close was called and the queue is empty,
+// or ctx is cancelled.
+func (o *Outbox) Run(ctx context.Context) {
+	for {
+		var it outboxItem
+		select {
+		case it = <-o.ch:
+		case <-o.closed:
+			select {
+			case it = <-o.ch:
+			default:
+				return
+			}
+		case <-ctx.Done():
+			return
+		}
+		for {
+			select {
+			case <-o.client.Connected():
+			case <-ctx.Done():
+				return
+			}
+			if err := o.client.Send(it.msg); err == nil {
+				break
+			}
+			select { // the stream just ended; wait for the next one
+			case <-time.After(50 * time.Millisecond):
+			case <-ctx.Done():
+				return
+			}
+		}
+		if it.onSent != nil {
+			it.onSent()
+		}
+	}
+}
diff --git a/pkg/worker/workertest/hypervisor.go b/pkg/worker/workertest/hypervisor.go
index 3a4c2ab..a0f71ea 100644
--- a/pkg/worker/workertest/hypervisor.go
+++ b/pkg/worker/workertest/hypervisor.go
@@ -6,11 +6,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"log/slog"
 	"net"
+	"os"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/opifica/opifica/internal/agent"
+	"github.com/opifica/opifica/internal/server"
 	"github.com/opifica/opifica/pkg/worker"
 )
 
@@ -21,6 +26,10 @@ type Hypervisor struct {
 	BootDelay time.Duration
 	// FailStart, if set, is called on every Start; a non-nil error fails it.
 	FailStart func(spec worker.VMSpec) error
+	// Agent runs a real agent for every VM, on the host, in a fresh work
+	// directory that is deleted with the VM. Jobs then really run, but with
+	// NO isolation: for tests and development only.
+	Agent bool
 
 	nextID    atomic.Int64
 	mu        sync.Mutex
@@ -82,8 +91,16 @@ type VM struct {
 	mu      sync.Mutex
 	started bool
 	gone    bool
+
+	// With Hypervisor.Agent:
+	root      string
+	agentAddr string
+	stopAgent func()
 }
 
+// Root is the VM's agent work directory (with Hypervisor.Agent).
+func (v *VM) Root() string { return v.root }
+
 var _ worker.VM = (*VM)(nil)
 
 // ID implements worker.VM.
@@ -109,10 +126,41 @@ func (v *VM) Start(ctx context.Context) error {
 	if v.gone {
 		return errors.New("vm destroyed")
 	}
+	if v.hv.Agent {
+		if err := v.startAgent(); err != nil {
+			return err
+		}
+	}
 	v.started = true
 	return nil
 }
 
+func (v *VM) startAgent() error {
+	root, err := os.MkdirTemp("", "opifica-vm-")
+	if err != nil {
+		return err
+	}
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		os.RemoveAll(root)
+		return err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
+	done := make(chan struct{})
+	go func() {
+		server.Serve(ctx, ln, agent.Handler(&agent.Server{Root: root, Log: quiet}), quiet)
+		close(done)
+	}()
+	v.root, v.agentAddr = root, ln.Addr().String()
+	v.stopAgent = func() {
+		cancel()
+		<-done
+		os.RemoveAll(root)
+	}
+	return nil
+}
+
 // DialAgent implements worker.VM with an in-memory pipe.
 func (v *VM) DialAgent(ctx context.Context) (net.Conn, error) {
 	v.mu.Lock()
@@ -120,8 +168,12 @@ func (v *VM) DialAgent(ctx context.Context) (net.Conn, error) {
 	if !v.started || v.gone {
 		return nil, errors.New("vm not running")
 	}
-	client, server := net.Pipe()
-	server.Close()
+	if v.agentAddr != "" {
+		var d net.Dialer
+		return d.DialContext(ctx, "tcp", v.agentAddr)
+	}
+	client, srv := net.Pipe()
+	srv.Close()
 	return client, nil
 }
 
@@ -130,10 +182,14 @@ func (v *VM) Destroy(context.Context) error {
 	v.mu.Lock()
 	already := v.gone
 	v.gone = true
+	stop := v.stopAgent
 	v.mu.Unlock()
 	if already {
 		return nil
 	}
+	if stop != nil {
+		stop()
+	}
 	v.hv.mu.Lock()
 	defer v.hv.mu.Unlock()
 	delete(v.hv.alive, v.id)