Plan T14, follow-up 3 of the Node gap audit… (#12)

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex 2116e08..1afae82 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -24,7 +24,12 @@ jobs:
        - run: pip install av numpy
        - run: test -z "$(gofmt -l .)" || (gofmt -l . && exit 1)
        - run: go vet ./...
++      # The job worker isolates jobs in unprivileged user and network namespaces, which Ubuntu's AppArmor restricts by
++      # default; its tests skip where they cannot run, and OPENVIBE_WORKER_TESTS=require makes them fail here instead.
++      - run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 || true
        - run: go test -race -count=1 ./...
++        env:
++          OPENVIBE_WORKER_TESTS: require
  
    build:
      name: cross-compile
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 0c0ed38..7cfcbfb 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -180,9 +180,10 @@ Commands are never queued while offline and never replayed after a reconnect.
  
  ### Jobs
  
--The job frames of OpenVibe.Contracts `platform.job-frame@1` (plan T14 Run) ride the same link and envelope.
--**Running jobs is not implemented yet:** the Node advertises no runtime class, so it refuses every well-formed job
--with `class not available`, and nothing executes.
++The job frames of OpenVibe.Contracts `platform.job-frame@1` (plan T14 Run) ride the same link and envelope. The Node
++runs class `function` only, and only when the owner turned the worker on in the local config, the OS is Linux and a
++boot-time probe could start a process in its own user, network and PID namespaces ([worker.md](worker.md)).
++Otherwise it advertises no class and refuses every well-formed job with `class not available`, and nothing executes.
  
  | direction        | type           | fields                                                                                    |
  |------------------|----------------|-------------------------------------------------------------------------------------------|
@@@ -205,11 +206,34 @@ checked in this order:
  | `function job needs an artifact with an exact version` | `bad_value` | a `function` job without `artifact`, or a name or version outside the contract's pattern (a range) |
  | `net policy not supported`                   | `unsupported` | `net` is present and not `deny`                             |
  | `missing or invalid args, ttl_ms or limits`  | `bad_value`   | `args` is not an object, or `ttl_ms` or a limit is missing or below 1 |
--| `class not available`                        | `unsupported` | the class is valid but this Node does not run it (today: every class) |
--
--A resent `job` with an id already answered gets the same answer again, never a second acceptance; the Node
--remembers the last 1024 well-formed ids. `job_cancel` for an id that is not running (unknown, refused or ended) and
--`job_exit_ack` for an unknown id are ignored, with no answer: an `ack` keyed by a job id means the job was accepted.
++| `class not available`                        | `unsupported` | the class is valid but this Node does not run it (every class but `function`; every class while the worker is off) |
++| `the e-stop or local stop is latched`        | `estopped` or `local_stop` | the stop latch is set: no job starts until it clears |
++| `class not available`                        | `shutting_down` | the Node is stopping                                      |
++| `unknown artifact`                           | `unsupported` | `artifact` is not a function the owner declared, at that exact version |
++| `worker busy`                                | `not_ready`   | `worker.caps.max_jobs` jobs are already running              |
++
++A resent `job` with an id already answered gets the same answer again, never a second acceptance and never a second
++process: `ack` while it runs, its `job_exit` once it ended; the Node remembers the last 1024 well-formed ids.
++
++An accepted job runs once the `ack` is out: `job_started` (`started_ms` anchors every usage second), its stdout as
++`job_stdout` chunks (at most 16 KiB each, UTF-8, `chunk_seq` from 1; best effort, not resent), one `job_usage` for
++every wall-clock second once it has fully elapsed (`second` from 0, `cpu_ms` that second's CPU time), then
++`job_exit`. Its `usage` is authoritative for metering (Bot writes `usage-sample` readings `service:"run"`,
++`operation:"function.invoke"`, unit `s`, idempotency key `run:<job id>:<n>`; none of those fields changes on a
++resend). The Node resends `job_started` for running jobs and `job_exit` for ended ones after every reconnect, and
++`job_exit` again on a `job` or `job_cancel` for that id, until `job_exit_ack`. `job_exit.reason`:
++
++| `reason`    | `code`  | when                                                                                                |
++|-------------|---------|-----------------------------------------------------------------------------------------------------|
++| `exited`    | status  | the process ended by itself (`code` is null if a signal it did not get from the worker ended it); `result` is what it wrote to fd 3 when it exited 0 |
++| `cancelled` | null    | `job_cancel`: the process group is killed; a job cancelled before it started never starts            |
++| `ttl`       | null    | `ttl_ms` (from receipt, clamped to `worker.caps.max_ttl_ms`) ran out, before or while it ran         |
++| `limit`     | null    | `wall_ms`, `cpu_ms` or `mem_bytes` (each clamped to the local cap), more stdout than `max_output_bytes`, or more than 256 KiB of result |
++| `stopped`   | null    | the e-stop or the local stop latched, or the Node shut down                                          |
++| `failed`    | null    | the worker could not run it: the namespaces could not be created, the spawn failed, or its CPU limit could not be set |
++
++`job_cancel` for an id that is not running (unknown, refused or ended) and `job_exit_ack` for an unknown id are
++ignored, with no answer: an `ack` keyed by a job id means the job was accepted.
  A Node that runs jobs lists its classes in `status.capabilities.worker` as `{"runtime_classes": ["function"]}`; the
  key is absent while it runs none, and `worker` is reserved as a plugin name.
  
ddiff --git a/docs/worker.md b/docs/worker.md
nnew file mode 100644
iindex 0000000..cb65219
---- /dev/null
++++ b/docs/worker.md
@@@ -0,0 +1,123 @@
++# The job worker (`function` jobs)
++
++The Node can run **function jobs** (OpenVibe.Contracts `platform.job@1`, class `function`) sent over the control link
++([protocol.md](protocol.md#jobs)). A job is untrusted input: it names a function, and the Node runs it only if the
++owner declared that function, at that exact version, in the local config. Nothing is downloaded and no path comes from
++the server. The worker is **off by default**; off, the Node refuses every job `class not available`, as before.
++
++## Config
++
++`config.json`, section `worker` (defaults shown; a zero cap means the default):
++
++```json
++"worker": {
++  "enabled": false,
++  "functions": [
++    {"name": "thumbnail", "version": "1.2.0", "command": ["/opt/fn/thumbnail/bin/run"], "env": {"MODE": "fast"}}
++  ],
++  "run_as": {"uid": 2001, "gid": 2001},
++  "allow_same_user": false,
++  "caps": {
++    "max_ttl_ms": 600000,
++    "max_wall_ms": 300000,
++    "max_cpu_ms": 300000,
++    "max_mem_bytes": 536870912,
++    "max_output_bytes": 1048576,
++    "max_jobs": 1
++  }
++}
++```
++
++- `functions`: `name` and `version` match a job's `artifact` exactly (the contract's patterns; listed once each);
++  `command[0]` is an absolute path. `env` is added to the job's environment.
++- `run_as`: the unprivileged uid and gid jobs run as (not 0; no supplementary groups). It needs the Node to run as
++  root, which the system install does. Create a dedicated user for it with no login and no groups.
++- `allow_same_user`: run jobs as the Node's own (non-root) user instead. A job can then read every file the Node can,
++  `credential.json` included, and use its groups (`gpio`, `dialout`, `video`): **for development only**. With neither
++  `run_as` nor `allow_same_user` the worker stays off.
++- `caps`: the owner's local caps. A job's `ttl_ms`, `limits.wall_ms`, `limits.cpu_ms` and `limits.mem_bytes` are
++  each clamped to them (the stricter value wins, as with `max_command_ms`). `max_output_bytes` caps one job's stdout;
++  `max_jobs` how many run at once.
++
++The class `function` is advertised in `status.capabilities.worker` only when `enabled` is true, the OS is Linux and a
++**boot-time probe** passes: it starts `/bin/sh` the way a job is started and checks that the process got user,
++network and PID namespaces of its own, that its CPU limit can be set and that its CPU and memory can be read. If the
++probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu 24.04
++and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is not
++affected.
++
++## How a job runs
++
++1. **Admit** (before the `ack`): refused unless the stop latch is clear, the artifact is declared and fewer than
++   `max_jobs` jobs run. An id the worker already holds is answered from it (`ack` while it runs, its `job_exit` once
++   it ended) and never starts a second process.
++2. **Start**: a fresh working directory (`$TMPDIR/openvibe-job-*`, owned by `run_as`, removed afterwards); an
++   environment built from scratch (`PATH=/usr/local/bin:/usr/bin:/bin`, `HOME` and `TMPDIR` the working directory,
++   `LANG=C.UTF-8`, `OPENVIBE_JOB_ID`, and the function's `env`; nothing inherited from the Node); `args` as one line of
++   JSON on stdin; stderr discarded. The process starts in its own process group and in new user, network and PID
++   namespaces, with `PR_SET_PDEATHSIG` SIGKILL, then gets `RLIMIT_CPU` (at `cpu_ms`, rounded up to a second) and
++   `RLIMIT_CORE` 0. `job_started` follows.
++3. **Run**: stdout streams as `job_stdout`; a watchdog samples CPU time and resident memory of every process in the
++   job's PID namespace every 200 ms and sends `job_usage` for every second once it fully elapsed.
++4. **End**: the function's result is what it writes to **fd 3** (JSON, at most 256 KiB), reported only when it exits
++   0. `job_exit` carries the reason, the exit code (null when it was killed) and the usage, and is resent until
++   `job_exit_ack`.
++
++Every kill is SIGKILL to the whole process group and to the leader, which is PID 1 of the job's PID namespace, so its
++death also kills every process the job left behind. The watchdog kills at `ttl_ms` (from receipt), `wall_ms` (from
++start), `cpu_ms`, `mem_bytes`, at `max_output_bytes` of stdout, on `job_cancel`, and the moment the e-stop or the local
++stop latches (`openvibe-node stop`); while the latch is set every job is refused. A Node shutting down kills its jobs
++(`stopped`).
++
++## Isolation model and its limits
++
++In five lines:
++
++1. Only owner-declared executables run, matched by exact name and version; the server chooses which and with what JSON.
++2. Each job is a separate unprivileged process (`run_as`) in its own process group and user, network and PID namespaces.
++3. The network namespace has only a loopback that is down: no IP network, not even to the Node or `localhost`.
++4. Nothing of the Node's environment reaches it; its working directory is fresh and removed after the job.
++5. ttl, wall, CPU, memory and output are enforced by a per-job watchdog that SIGKILLs the whole group; `RLIMIT_CPU` backs the CPU cap in the kernel.
++
++What it does **not** protect against (no seccomp, no cgroups, no mount namespace yet):
++
++- **Files**: there is no filesystem isolation. A job reads and writes whatever its uid may: with `run_as`, every
++  world-readable file and world-writable directory; with `allow_same_user`, everything the Node can, including
++  `credential.json` and the local control socket (so it could clear the local stop).
++- **Unix sockets by path** (Docker's, D-Bus's, the Node's control socket) are reachable: a network namespace does not
++  cover filesystem sockets; only their file permissions do.
++- **The kernel**: no seccomp filter, so the whole syscall surface is open, including what user namespaces expose to
++  unprivileged code (a common source of local privilege escalations). Keep the kernel patched.
++- **Memory and CPU are sampled**, not enforced by cgroups: a burst between two 200 ms samples can exceed `mem_bytes`
++  (and meet the kernel's OOM killer first); the `RLIMIT_CPU` backstop is set just after start and covers each process
++  on its own, not their sum.
++- **No limit on processes, threads, disk or I/O**: a fork bomb is bounded only by the uid's `RLIMIT_NPROC`, disk use
++  in the working directory or `/tmp` only by free space; `/tmp` and `/dev/shm` are shared with the machine.
++- **`/proc` is the host's**: a job can list the machine's processes and their command lines (not signal them: its PID
++  namespace has no host PIDs).
++- **The `args` and the result are trusted to the function**: the worker passes the server's JSON unchanged; validating
++  it is the function's job.
++
++## Refusals and exit reasons
++
++Refused with `nack` (keyed by the job id; the first matching row wins, after the checks every Node does, in
++[protocol.md](protocol.md#jobs)):
++
++| `message`                             | `fault_code`               | when                                                    |
++|---------------------------------------|----------------------------|---------------------------------------------------------|
++| `class not available`                 | `unsupported`              | the worker is off (disabled, not Linux, probe failed), or the class is not `function` |
++| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set                                   |
++| `class not available`                 | `shutting_down`            | the Node is stopping                                    |
++| `unknown artifact`                    | `unsupported`              | no declared function has that name and exact version    |
++| `worker busy`                         | `not_ready`                | `max_jobs` jobs are running                             |
++
++Accepted jobs end with `job_exit.reason`:
++
++| `reason`    | when                                                                                                      |
++|-------------|-----------------------------------------------------------------------------------------------------------|
++| `exited`    | the process ended by itself; `code` is its status (null if a signal the worker did not send ended it)      |
++| `cancelled` | `job_cancel`; a job cancelled before it started never starts                                               |
++| `ttl`       | `ttl_ms` (clamped) ran out, before or while it ran                                                         |
++| `limit`     | `wall_ms`, `cpu_ms` (sampled, or `RLIMIT_CPU`'s SIGXCPU) or `mem_bytes` exceeded, stdout past `max_output_bytes`, or a result over 256 KiB |
++| `stopped`   | the e-stop or local stop latched, or the Node shut down                                                    |
++| `failed`    | the job could not run isolated: namespaces not created, spawn error, working directory not made, CPU limit not set; never run with less isolation |
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex d7d7df7..bd5ce13 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -17,6 +17,7 @@ import (
  	"io/fs"
  	"os"
  	"path/filepath"
++	"regexp"
  	"runtime"
  	"strings"
  )
@@@ -106,6 +107,66 @@ type Config struct {
  	// Python is the interpreter for plugins without an explicit command (default: <state>/venv/bin/python, then python3).
  	Python   string `json:"python,omitempty"`
  	LogLevel string `json:"log_level,omitempty"`
++	// Worker runs `function` jobs from the control link (docs/worker.md). Off unless enabled.
++	Worker WorkerConfig `json:"worker,omitzero"`
++}
++
++// WorkerConfig is the job worker. Disabled (the default), every job is refused "class not available". Only the
++// functions listed here ever run: a job names one by name and exact version, and nothing is downloaded.
++type WorkerConfig struct {
++	Enabled   bool             `json:"enabled,omitempty"`
++	Functions []FunctionConfig `json:"functions,omitempty"`
++	// RunAs is the dedicated unprivileged uid/gid jobs run as; it needs the Node to run as root. Without it the Node
++	// refuses to run jobs unless AllowSameUser is set.
++	RunAs *RunAs `json:"run_as,omitempty"`
++	// AllowSameUser runs jobs as the Node's own user (no run_as). A job can then read every file the Node can,
++	// including credential.json, and use the Node user's groups (gpio, dialout, video): for development only.
++	AllowSameUser bool       `json:"allow_same_user,omitempty"`
++	Caps          WorkerCaps `json:"caps,omitzero"`
++}
++
++// FunctionConfig is one function a job may run: Command is started with the job's args as JSON on stdin.
++type FunctionConfig struct {
++	Name    string            `json:"name"`
++	Version string            `json:"version"`
++	Command []string          `json:"command"` // Command[0] is an absolute path
++	Env     map[string]string `json:"env,omitempty"`
++}
++
++// RunAs is a uid and gid on this machine.
++type RunAs struct {
++	UID uint32 `json:"uid"`
++	GID uint32 `json:"gid"`
++}
++
++// WorkerCaps are the owner's local caps on every job; a job's own ttl_ms and limits are clamped to them (the
++// stricter value wins). Zero means the default (DefaultWorkerCaps).
++type WorkerCaps struct {
++	MaxTTLMS       int64 `json:"max_ttl_ms,omitempty"`
++	MaxWallMS      int64 `json:"max_wall_ms,omitempty"`
++	MaxCPUMS       int64 `json:"max_cpu_ms,omitempty"`
++	MaxMemBytes    int64 `json:"max_mem_bytes,omitempty"`
++	MaxOutputBytes int64 `json:"max_output_bytes,omitempty"` // stdout of one job, in bytes
++	MaxJobs        int   `json:"max_jobs,omitempty"`         // jobs running at once
++}
++
++// DefaultWorkerCaps are the caps for every field the config leaves at zero.
++var DefaultWorkerCaps = WorkerCaps{MaxTTLMS: 600000, MaxWallMS: 300000, MaxCPUMS: 300000, MaxMemBytes: 512 << 20,
++	MaxOutputBytes: 1 << 20, MaxJobs: 1}
++
++// WithDefaults fills the zero fields of c from DefaultWorkerCaps.
++func (c WorkerCaps) WithDefaults() WorkerCaps {
++	d := DefaultWorkerCaps
++	for _, f := range []struct{ v, def *int64 }{{&c.MaxTTLMS, &d.MaxTTLMS}, {&c.MaxWallMS, &d.MaxWallMS},
++		{&c.MaxCPUMS, &d.MaxCPUMS}, {&c.MaxMemBytes, &d.MaxMemBytes}, {&c.MaxOutputBytes, &d.MaxOutputBytes}} {
++		if *f.v <= 0 {
++			*f.v = *f.def
++		}
++	}
++	if c.MaxJobs <= 0 {
++		c.MaxJobs = d.MaxJobs
++	}
++	return c
  }
  
  type LocalLimits struct {
@@@ -239,6 +300,35 @@ func (c *Config) Validate() error {
  			return errors.New("config: limits must be between 0 and 1")
  		}
  	}
++	return c.Worker.validate()
++}
++
++var (
++	functionNameRe    = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)
++	functionVersionRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)
++)
++
++func (w WorkerConfig) validate() error {
++	seen := map[string]bool{}
++	for _, f := range w.Functions {
++		if !functionNameRe.MatchString(f.Name) || !functionVersionRe.MatchString(f.Version) {
++			return fmt.Errorf("config: worker function %q version %q: name or version is not a valid artifact name or exact version", f.Name, f.Version)
++		}
++		if seen[f.Name+"@"+f.Version] {
++			return fmt.Errorf("config: worker function %s@%s listed twice", f.Name, f.Version)
++		}
++		seen[f.Name+"@"+f.Version] = true
++		if len(f.Command) == 0 || !filepath.IsAbs(f.Command[0]) {
++			return fmt.Errorf("config: worker function %s@%s needs a command starting with an absolute path", f.Name, f.Version)
++		}
++	}
++	if w.RunAs != nil && (w.RunAs.UID == 0 || w.RunAs.GID == 0) {
++		return errors.New("config: worker.run_as must not be root (uid or gid 0)")
++	}
++	c := w.Caps
++	if c.MaxTTLMS < 0 || c.MaxWallMS < 0 || c.MaxCPUMS < 0 || c.MaxMemBytes < 0 || c.MaxOutputBytes < 0 || c.MaxJobs < 0 {
++		return errors.New("config: worker.caps must not be negative")
++	}
  	return nil
  }
  
ddiff --git a/internal/node/jobs.go b/internal/node/jobs.go
iindex a37f1d5..cc61074 100644
---- a/internal/node/jobs.go
++++ b/internal/node/jobs.go
@@@ -4,18 +4,21 @@ import (
  	"sync"
  
  	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/safety"
  )
  
  // jobSeenMax bounds how many job ids the Node remembers answers for.
  const jobSeenMax = 1024
  
--// jobs answers job frames. Nothing runs yet: a job is refused unless its class is advertised, and an accepted job is
--// only remembered (the worker that runs it fills classes and executes accepted jobs).
++// jobs answers job frames: a job is refused unless its class is advertised and admit (the worker) takes it. Running
++// jobs live in the worker's own table, not here.
  type jobs struct {
  	mu      sync.Mutex
  	classes []string                    // runtime classes this Node runs and advertises; empty: none
--	answers map[string]protocol.Message // job id → the ack or nack it got, resent unchanged for a repeated job
++	answers map[string]protocol.Message // job id → the ack or nack it got, or its job_exit once it ended
  	order   []string                    // ids in answers, oldest first
++	// admit reserves an accepted job in the worker or returns the fault and reason to nack it with; nil accepts.
++	admit func(protocol.Job) (fault, reason string)
  }
  
  func newJobs(classes []string) *jobs {
@@@ -38,6 +41,9 @@ func (j *jobs) request(r protocol.JobRequest) (answer protocol.Message, accepted
  		return a, false
  	}
  	fault, reason := r.Check(j.classes)
++	if fault == "" && j.admit != nil {
++		fault, reason = j.admit(r.Job)
++	}
  	if fault != "" {
  		answer = protocol.Nack{ID: id, FaultCode: fault, Message: reason}
  	} else {
@@@ -55,6 +61,15 @@ func (j *jobs) request(r protocol.JobRequest) (answer protocol.Message, accepted
  	return answer, accepted
  }
  
++// ended makes a job's job_exit the answer to a resent job frame for it.
++func (j *jobs) ended(e protocol.JobExit) {
++	j.mu.Lock()
++	defer j.mu.Unlock()
++	if _, ok := j.answers[e.ID]; ok {
++		j.answers[e.ID] = e
++	}
++}
++
  // accepted reports whether id was accepted (and so would be running).
  func (j *jobs) accepted(id string) bool {
  	j.mu.Lock()
@@@ -63,8 +78,15 @@ func (j *jobs) accepted(id string) bool {
  	return ok
  }
  
--// job answers a job frame with ack or nack, keyed by the job id.
++// job answers a job frame with ack or nack, keyed by the job id, and runs an accepted job. A job the worker holds is
++// answered from it (an ack while it runs, its job_exit once it ended) and never started twice.
  func (n *Node) job(r protocol.JobRequest) {
++	if n.worker != nil {
++		if m, ok := n.worker.Known(r.Job.ID); ok {
++			n.send(m)
++			return
++		}
++	}
  	answer, accepted := n.jobs.request(r)
  	if nk, ok := answer.(protocol.Nack); ok {
  		n.log.Info("job refused", "id", r.Job.ID, "class", r.Job.Class, "reason", nk.Message, "err", r.Err)
@@@ -72,14 +94,42 @@ func (n *Node) job(r protocol.JobRequest) {
  		n.log.Info("job accepted", "id", r.Job.ID, "class", r.Job.Class)
  	}
  	n.send(answer)
++	if accepted && n.worker != nil {
++		n.worker.Launch(r.Job.ID)
++	}
++}
++
++// sendJobFrame sends a worker frame; a job_exit also becomes the answer to a resent job frame.
++func (n *Node) sendJobFrame(m protocol.Message) {
++	if e, ok := m.(protocol.JobExit); ok {
++		n.jobs.ended(e)
++	}
++	n.send(m)
  }
  
  // jobCancel stops an accepted job. An unknown, refused or ended id is ignored, as platform.job-frame@1 says, and no
--// frame answers it: an ack keyed by a job id means the job was accepted.
++// frame answers it: an ack keyed by a job id means the job was accepted. An ended job's unacked job_exit is resent.
  func (n *Node) jobCancel(c protocol.JobCancel) {
++	if n.worker != nil {
++		if !n.worker.Cancel(c.ID) {
++			n.log.Debug("job_cancel for a job that is not running ignored", "id", c.ID)
++		}
++		return
++	}
  	if !n.jobs.accepted(c.ID) {
  		n.log.Debug("job_cancel for a job that is not running ignored", "id", c.ID)
  		return
  	}
--	n.log.Warn("job_cancel: job execution is not implemented, nothing to stop", "id", c.ID)
++	n.log.Warn("job_cancel: no worker runs jobs on this Node, nothing to stop", "id", c.ID)
++}
++
++// stopFault is the fault a job is refused with while the stop latch is set; "" when it is clear.
++func stopFault(st safety.LatchState) string {
++	switch {
++	case st.Local:
++		return protocol.FaultLocalStop
++	case st.Remote:
++		return protocol.FaultEstopped
++	}
++	return ""
  }
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 7579acc..ce683e5 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -19,6 +19,7 @@ import (
  	"os"
  	"os/exec"
  	"runtime"
++	"slices"
  	"sort"
  	"strings"
  	"sync"
@@@ -32,6 +33,7 @@ import (
  	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
  	"github.com/OpenVibers/OpenVibe.Node/internal/safety"
  	"github.com/OpenVibers/OpenVibe.Node/internal/video"
++	"github.com/OpenVibers/OpenVibe.Node/internal/worker"
  )
  
  // Options configure a Node.
@@@ -42,8 +44,9 @@ type Options struct {
  	Log     *slog.Logger
  	Version string
  
--	// RuntimeClasses are the job classes this Node runs and advertises in status.capabilities.worker. None yet:
--	// every job is refused "class not available" until the worker that executes jobs fills this in.
++	// RuntimeClasses are the job classes this Node runs and advertises in status.capabilities.worker. New adds
++	// `function` when config.worker is enabled and its isolation probe passes; with none, every job is refused
++	// "class not available".
  	RuntimeClasses []string
  
  	// For tests.
@@@ -60,9 +63,11 @@ type Node struct {
  	latch *safety.Latch
  	dedup *safety.Dedup
  	jobs  *jobs
--	mgr   *plugins.Manager
--	link  *link.Link
--	pub   *video.Publisher
++	// worker runs function jobs; nil unless config.worker is enabled on Linux and its boot-time probe passed.
++	worker *worker.Worker
++	mgr    *plugins.Manager
++	link   *link.Link
++	pub    *video.Publisher
  
  	ctx context.Context
  
@@@ -149,6 +154,20 @@ func New(opt Options) (*Node, error) {
  				ICEServers: opt.Creds.ICEServers, Log: opt.Log, IncludeLoopback: opt.VideoLoopback})
  		}
  	}
++	if opt.Config.Worker.Enabled {
++		w := worker.New(opt.Config.Worker, n.sendJobFrame, opt.Log)
++		if err := w.Probe(); err != nil {
++			opt.Log.Error("worker off: jobs cannot run isolated here, every job is refused", "err", err)
++		} else {
++			n.worker = w
++			w.SetStopped(stopFault(n.latch.State()))
++			n.jobs.admit = w.Admit
++			if !slices.Contains(n.jobs.classes, protocol.ClassFunction) {
++				n.jobs.classes = append(n.jobs.classes, protocol.ClassFunction)
++			}
++			opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
++		}
++	}
  	n.latch.OnChange(n.onLatch)
  	return n, nil
  }
@@@ -188,6 +207,9 @@ func (n *Node) Run(ctx context.Context) error {
  	<-ctx.Done()
  	n.log.Info("shutting down: stopping every actuator")
  	n.mgr.StopAll()
++	if n.worker != nil {
++		n.worker.Close()
++	}
  	cancelPlugins()
  	n.mgr.Wait()
  	wg.Wait()
@@@ -236,8 +258,11 @@ func (n *Node) Frame(f protocol.Frame) {
  	case protocol.JobCancel:
  		n.jobCancel(m)
  	case protocol.JobExitAck:
--		// No job runs yet, so no job_exit is ever waiting for this.
--		n.log.Debug("job_exit_ack ignored", "id", m.ID)
++		if n.worker != nil {
++			n.worker.ExitAck(m.ID)
++		} else {
++			n.log.Debug("job_exit_ack ignored: no worker", "id", m.ID)
++		}
  	}
  }
  
@@@ -278,6 +303,9 @@ func (n *Node) applyConfig(c protocol.Config) {
  	n.mu.Unlock()
  	if first {
  		n.sendStatus()
++		if n.worker != nil {
++			n.worker.Resend()
++		}
  	}
  	n.sendEstopState()
  	if n.link != nil {
@@@ -428,6 +456,9 @@ func (n *Node) send(m protocol.Message) {
  // ---- latch ----
  
  func (n *Node) onLatch(st safety.LatchState) {
++	if n.worker != nil {
++		n.worker.SetStopped(stopFault(st)) // kills every job at once while stopped
++	}
  	if st.Stopped() {
  		n.log.Warn("stop latched", "remote", st.Remote, "local", st.Local)
  		n.mgr.EstopAll()
ddiff --git a/internal/node/worker_test.go b/internal/node/worker_test.go
nnew file mode 100644
iindex 0000000..3f463b7
---- /dev/null
++++ b/internal/node/worker_test.go
@@@ -0,0 +1,195 @@
++//go:build linux
++
++package node
++
++import (
++	"context"
++	"encoding/json"
++	"fmt"
++	"io"
++	"log/slog"
++	"os"
++	"path/filepath"
++	"slices"
++	"strings"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++	"github.com/OpenVibers/OpenVibe.Node/internal/link"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// TestWorkerHelperProcess is the job's process when this test binary runs as the `hello` function (`-- job`): it
++// appends a line to the marker file named in its args, prints, holds for 1.5 s and writes its result to fd 3.
++func TestWorkerHelperProcess(t *testing.T) {
++	i := slices.Index(os.Args, "--")
++	if i < 0 || i+1 >= len(os.Args) || os.Args[i+1] != "job" {
++		return
++	}
++	var args struct {
++		Marker string `json:"marker"`
++	}
++	_ = json.NewDecoder(os.Stdin).Decode(&args)
++	if f, err := os.OpenFile(args.Marker, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600); err == nil {
++		fmt.Fprintln(f, "started")
++		f.Close()
++	}
++	fmt.Println("hello from the job")
++	time.Sleep(1500 * time.Millisecond)
++	fmt.Fprint(os.NewFile(3, "result"), `{"ok":true}`)
++	os.Exit(0)
++}
++
++// startWorkerNode pairs a Node with no plugin and no video whose worker runs the `hello` function, and waits for the
++// status that advertises it. It is skipped where the kernel does not let this user create the job namespaces.
++func startWorkerNode(t *testing.T) *env {
++	t.Helper()
++	exe, err := os.Executable()
++	if err != nil {
++		t.Fatal(err)
++	}
++	home, err := os.MkdirTemp("", "ovn")
++	if err != nil {
++		t.Fatal(err)
++	}
++	t.Cleanup(func() { os.RemoveAll(home) })
++	paths := config.HomePaths(home)
++	srv := fakebot.New()
++	t.Cleanup(srv.Close)
++	srv.Config = protocol.Config{HeartbeatMS: 100, Limits: protocol.Limits{MaxCommandMS: 1000}}
++	srv.AddCode("TEST2345")
++	creds, err := link.Pair(context.Background(), nil, srv.URL(), protocol.PairRequest{Code: "TEST2345", AgentVersion: "test",
++		DeviceKind: "onboard"})
++	if err != nil {
++		t.Fatal(err)
++	}
++	cfg := config.Default()
++	cfg.Plugins = nil
++	cfg.Video = config.VideoConfig{Source: "off"}
++	cfg.Worker = config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: config.WorkerCaps{MaxMemBytes: 8 << 30},
++		Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{exe, "-test.run=^TestWorkerHelperProcess$", "--", "job"}}}}
++	if err := cfg.Validate(); err != nil {
++		t.Fatal(err)
++	}
++	n, err := New(Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
++		LinkBackoffMin: 50 * time.Millisecond, LatchPoll: 50 * time.Millisecond, Log: slog.New(slog.NewTextHandler(io.Discard, nil))})
++	if err != nil {
++		t.Fatal(err)
++	}
++	if !slices.Contains(n.jobs.runtimeClasses(), protocol.ClassFunction) {
++		if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {
++			t.Fatal("job isolation unavailable: the worker probe failed")
++		}
++		t.Skip("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead)")
++	}
++	ctx, cancel := context.WithCancel(context.Background())
++	done := make(chan struct{})
++	go func() { n.Run(ctx); close(done) }()
++	e := &env{t: t, srv: srv, node: n, paths: paths, cancel: cancel, done: done}
++	t.Cleanup(e.stop)
++	e.conn = e.nextConn()
++	_, err = e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {
++		_, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]
++		return ok
++	})
++	if err != nil {
++		t.Fatal(err)
++	}
++	return e
++}
++
++// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
++// and job_exit arrive in that order; a job frame resent while it runs is acked again and starts no second process;
++// after the end a resent job is answered with the same job_exit, before and after job_exit_ack.
++func TestJobRunsOverTheLink(t *testing.T) {
++	e := startWorkerNode(t)
++	marker := filepath.Join(e.paths.StateDir, "starts")
++	j := testJob(jobID(1))
++	j.Artifact = &protocol.Artifact{Name: "hello", Version: "1.0.0"}
++	j.Args = json.RawMessage(fmt.Sprintf(`{"marker":%q}`, marker))
++	j.Limits.MemBytes = 4 << 30
++	isJob := func(m protocol.Message) bool {
++		switch f := m.(type) {
++		case protocol.JobStarted:
++			return f.ID == j.ID
++		case protocol.JobExit:
++			return f.ID == j.ID
++		}
++		return false
++	}
++
++	if r := e.jobReply(j); r != (protocol.Ack{ID: j.ID}) {
++		t.Fatalf("%+v", r)
++	}
++	if _, err := e.conn.Expect(protocol.TypeJobStarted, 10*time.Second, isJob); err != nil {
++		t.Fatal(err)
++	}
++	if r := e.jobReply(j); r != (protocol.Ack{ID: j.ID}) {
++		t.Fatalf("resent while running: %+v", r)
++	}
++	f, err := e.conn.Expect(protocol.TypeJobExit, 15*time.Second, isJob)
++	if err != nil {
++		t.Fatal(err)
++	}
++	ex := f.Msg.(protocol.JobExit)
++	if ex.Reason != protocol.ExitExited || ex.Code == nil || *ex.Code != 0 || string(ex.Result) != `{"ok":true}` || ex.Usage.StartedMS == nil {
++		t.Fatalf("%+v %s", ex, ex.Result)
++	}
++
++	// Order on the wire: ack, job_started, job_stdout, job_usage, job_exit; one job_started only.
++	var kinds []string
++	var out strings.Builder
++	started := 0
++	for _, f := range e.conn.Frames() {
++		switch m := f.Msg.(type) {
++		case protocol.JobStarted:
++			if m.ID == j.ID {
++				started++
++				kinds = append(kinds, f.Type)
++			}
++		case protocol.JobStdout:
++			if m.ID == j.ID {
++				out.WriteString(m.Chunk)
++				kinds = append(kinds, f.Type)
++			}
++		case protocol.JobUsage:
++			if m.ID == j.ID && m.Second == 0 && m.StartedMS == *ex.Usage.StartedMS {
++				kinds = append(kinds, f.Type)
++			}
++		case protocol.JobExit:
++			if m.ID == j.ID {
++				kinds = append(kinds, f.Type)
++			}
++		}
++	}
++	kinds = slices.Compact(kinds)
++	want := []string{protocol.TypeJobStarted, protocol.TypeJobStdout, protocol.TypeJobUsage, protocol.TypeJobExit}
++	if started != 1 || !slices.Equal(kinds, want) || out.String() != "hello from the job\n" {
++		t.Fatalf("%d job_started, frames %v, stdout %q", started, kinds, out.String())
++	}
++
++	// Ended: a resent job gets the same job_exit, before and after job_exit_ack, and never runs again.
++	for _, ack := range []bool{false, true} {
++		if ack {
++			if err := e.conn.JobExitAck(j.ID); err != nil {
++				t.Fatal(err)
++			}
++		}
++		if err := e.conn.Job(j); err != nil {
++			t.Fatal(err)
++		}
++		f, err := e.conn.Expect(protocol.TypeJobExit, 5*time.Second, isJob)
++		if err != nil {
++			t.Fatal(err)
++		}
++		if again := f.Msg.(protocol.JobExit); again.Reason != ex.Reason || *again.Usage.StartedMS != *ex.Usage.StartedMS || again.Usage.WallMS != ex.Usage.WallMS {
++			t.Fatalf("resent job answered %+v, first job_exit %+v", again, ex)
++		}
++	}
++	b, err := os.ReadFile(marker)
++	if err != nil || string(b) != "started\n" {
++		t.Fatalf("the function ran %q (%v), want once", b, err)
++	}
++}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex b3df2cf..81ceee5 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -97,6 +97,11 @@ const (
  	JobNetUnsupported = "net policy not supported"
  	JobBadLimits      = "missing or invalid args, ttl_ms or limits"
  	JobNotAvailable   = "class not available"
++
++	// Refused by a running worker (platform.job@1: a missing artifact, the local kill switch, limits it cannot honor).
++	JobUnknownArtifact = "unknown artifact"                    // not a function declared in the Node's local config
++	JobStopped         = "the e-stop or local stop is latched" // fault estopped or local_stop
++	JobBusy            = "worker busy"                         // worker.caps.max_jobs are running; fault not_ready
  )
  
  // job_exit reasons.
ddiff --git a/internal/worker/run_linux_test.go b/internal/worker/run_linux_test.go
nnew file mode 100644
iindex 0000000..80e5864
---- /dev/null
++++ b/internal/worker/run_linux_test.go
@@@ -0,0 +1,304 @@
++//go:build linux
++
++package worker
++
++import (
++	"bytes"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"io"
++	"io/fs"
++	"net"
++	"os"
++	"path/filepath"
++	"slices"
++	"strings"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// TestHelperProcess is the job's process when this test binary runs as a function (`-- <mode>`); otherwise it does
++// nothing.
++func TestHelperProcess(t *testing.T) {
++	i := slices.Index(os.Args, "--")
++	if i < 0 || i+1 >= len(os.Args) {
++		return
++	}
++	result := os.NewFile(3, "result")
++	switch os.Args[i+1] {
++	case "stream":
++		in, _ := io.ReadAll(os.Stdin)
++		fmt.Println("line 1")
++		time.Sleep(300 * time.Millisecond)
++		fmt.Println("line 2 é")
++		time.Sleep(1900 * time.Millisecond)
++		fmt.Fprintf(result, `{"got":%s}`, bytes.TrimSpace(in))
++	case "sleep":
++		fmt.Println("sleeping")
++		time.Sleep(time.Hour)
++	case "flood":
++		line := strings.Repeat("x", 1023) + "\n"
++		for {
++			os.Stdout.WriteString(line)
++		}
++	case "env":
++		cwd, _ := os.Getwd()
++		// A down loopback still lets a socket bind 127.0.0.1; connecting to it is what fails.
++		lerr := errors.New("no listener")
++		if l, err := net.Listen("tcp", "127.0.0.1:0"); err == nil {
++			_, lerr = net.DialTimeout("tcp", l.Addr().String(), time.Second)
++		}
++		_, derr := net.DialTimeout("tcp", "192.0.2.1:9", time.Second)
++		_ = json.NewEncoder(result).Encode(map[string]any{"env": os.Environ(), "cwd": cwd, "pid": os.Getpid(),
++			"uid": os.Getuid(), "loopback": fmt.Sprint(lerr), "dial": fmt.Sprint(derr)})
++	}
++	os.Exit(0)
++}
++
++// newRunWorker is a worker running the helper modes, skipped where the kernel does not let this user create the
++// namespaces (CI sets OPENVIBE_WORKER_TESTS=require to fail instead).
++func newRunWorker(t *testing.T, caps config.WorkerCaps, modes ...string) (*Worker, *sink) {
++	t.Helper()
++	if caps.MaxMemBytes == 0 {
++		caps.MaxMemBytes = 8 << 30
++	}
++	s := &sink{}
++	w := New(helperConfig(t, caps, modes...), s.send, quiet())
++	if err := w.Probe(); err != nil {
++		if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {
++			t.Fatalf("job isolation unavailable: %v", err)
++		}
++		t.Skipf("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead): %v", err)
++	}
++	t.Cleanup(w.Close)
++	return w, s
++}
++
++func run(t *testing.T, w *Worker, j protocol.Job) {
++	t.Helper()
++	if f, r := w.Admit(j); f != "" {
++		t.Fatalf("%s refused: %s %s", j.ID, f, r)
++	}
++	w.Launch(j.ID)
++}
++
++// TestRunStreamsAndMeters: a function gets its args on stdin, its stdout arrives as job_stdout in order, one
++// job_usage per elapsed wall-clock second anchored on job_started, and job_exit carries code 0, the result it wrote to
++// fd 3 and usage covering every second sent. The job_exit is resent until job_exit_ack, then the job is forgotten.
++func TestRunStreamsAndMeters(t *testing.T) {
++	w, s := newRunWorker(t, config.WorkerCaps{}, "stream")
++	j := testJob(1, "stream")
++	run(t, w, j)
++	ex := s.exit(t, j.ID)
++	if ex.Reason != protocol.ExitExited || ex.Code == nil || *ex.Code != 0 || string(ex.Result) != `{"got":{"n":7}}` {
++		t.Fatalf("%+v result %s", ex, ex.Result)
++	}
++	if ex.Usage.StartedMS == nil || ex.Usage.WallMS < 2000 || ex.Usage.CPUMS == nil || ex.Usage.MemPeakBytes == nil || *ex.Usage.MemPeakBytes == 0 {
++		t.Fatalf("usage %+v", ex.Usage)
++	}
++	frames := s.all()
++	started, ok := frames[0].(protocol.JobStarted)
++	if !ok || started.StartedMS != *ex.Usage.StartedMS {
++		t.Fatalf("first frame %+v, want job_started at %d", frames[0], *ex.Usage.StartedMS)
++	}
++	if _, ok := frames[len(frames)-1].(protocol.JobExit); !ok {
++		t.Fatalf("last frame %+v, want job_exit", frames[len(frames)-1])
++	}
++	var out strings.Builder
++	var chunk uint64
++	var second int64
++	for _, m := range frames {
++		switch f := m.(type) {
++		case protocol.JobStdout:
++			if f.ChunkSeq != chunk+1 {
++				t.Fatalf("chunk_seq %d after %d", f.ChunkSeq, chunk)
++			}
++			chunk = f.ChunkSeq
++			out.WriteString(f.Chunk)
++		case protocol.JobUsage:
++			if f.StartedMS != started.StartedMS || f.Second != second || f.CPUMS == nil {
++				t.Fatalf("usage %+v, want second %d at %d", f, second, started.StartedMS)
++			}
++			second++
++		}
++	}
++	if out.String() != "line 1\nline 2 é\n" {
++		t.Fatalf("stdout %q", out.String())
++	}
++	if second < 2 || second*1000 > ex.Usage.WallMS {
++		t.Fatalf("%d usage seconds for wall_ms %d", second, ex.Usage.WallMS)
++	}
++
++	if m, ok := w.Known(j.ID); !ok || m.(protocol.JobExit).Reason != protocol.ExitExited {
++		t.Fatalf("a resent job is answered %+v, want its job_exit", m)
++	}
++	w.Resend()
++	if again := s.wait(t, protocol.TypeJobExit, j.ID, 2, 5*time.Second).(protocol.JobExit); *again.Usage.StartedMS != *ex.Usage.StartedMS || again.Usage.WallMS != ex.Usage.WallMS {
++		t.Fatalf("resent job_exit %+v differs from %+v", again.Usage, ex.Usage)
++	}
++	w.ExitAck(j.ID)
++	if _, ok := w.Known(j.ID); ok {
++		t.Fatal("job still held after job_exit_ack")
++	}
++}
++
++// TestIsolation: the job sees none of the Node's environment, has no network (not even loopback), is PID 1 of its own
++// PID namespace, runs as the configured user, and its working directory is fresh and removed afterwards.
++func TestIsolation(t *testing.T) {
++	t.Setenv("OPENVIBE_NODE_SECRET", "leak")
++	w, s := newRunWorker(t, config.WorkerCaps{}, "env")
++	j := testJob(1, "env")
++	run(t, w, j)
++	ex := s.exit(t, j.ID)
++	var r struct {
++		Env            []string
++		Cwd            string
++		PID, UID       int
++		Loopback, Dial string
++	}
++	if err := json.Unmarshal(ex.Result, &r); err != nil || ex.Reason != protocol.ExitExited {
++		t.Fatalf("%+v %s: %v", ex, ex.Result, err)
++	}
++	if r.PID != 1 || r.UID != os.Getuid() {
++		t.Fatalf("pid %d uid %d, want 1 and %d", r.PID, r.UID, os.Getuid())
++	}
++	if r.Loopback == "<nil>" || r.Dial == "<nil>" {
++		t.Fatalf("network reachable: loopback %s, dial %s", r.Loopback, r.Dial)
++	}
++	for _, kv := range r.Env {
++		if strings.HasPrefix(kv, "OPENVIBE_NODE_SECRET=") {
++			t.Fatalf("the Node's environment leaked: %v", r.Env)
++		}
++	}
++	if !slices.Contains(r.Env, "OPENVIBE_JOB_ID="+j.ID) || !slices.Contains(r.Env, "HOME="+r.Cwd) {
++		t.Fatalf("env %v", r.Env)
++	}
++	if !strings.HasPrefix(filepath.Base(r.Cwd), "openvibe-job-") {
++		t.Fatalf("cwd %s", r.Cwd)
++	}
++	if _, err := os.Stat(r.Cwd); !errors.Is(err, fs.ErrNotExist) {
++		t.Fatalf("working directory %s not removed: %v", r.Cwd, err)
++	}
++}
++
++// TestTTLAndLimitKill: ttl_ms (from receipt) and wall_ms (from start) kill the process group, each clamped by the
++// local cap when that is stricter; job_exit says ttl or limit with no exit code.
++func TestTTLAndLimitKill(t *testing.T) {
++	for i, c := range []struct {
++		name      string
++		caps      config.WorkerCaps
++		ttl, wall int64
++		reason    string
++	}{
++		{"ttl", config.WorkerCaps{}, 800, 60000, protocol.ExitTTL},
++		{"ttl clamped by the local cap", config.WorkerCaps{MaxTTLMS: 800}, 60000, 60000, protocol.ExitTTL},
++		{"wall_ms", config.WorkerCaps{}, 60000, 800, protocol.ExitLimit},
++		{"wall_ms clamped by the local cap", config.WorkerCaps{MaxWallMS: 800}, 60000, 60000, protocol.ExitLimit},
++	} {
++		t.Run(c.name, func(t *testing.T) {
++			w, s := newRunWorker(t, c.caps, "sleep")
++			j := testJob(i+1, "sleep")
++			j.TTLMS, j.Limits.WallMS = c.ttl, c.wall
++			begin := time.Now()
++			run(t, w, j)
++			ex := s.exit(t, j.ID)
++			if ex.Reason != c.reason || ex.Code != nil || ex.Usage.StartedMS == nil || s.count(protocol.TypeJobStarted, j.ID) != 1 {
++				t.Fatalf("%+v", ex)
++			}
++			if d := time.Since(begin); d > 5*time.Second {
++				t.Fatalf("killed after %s", d)
++			}
++		})
++	}
++}
++
++// TestCancelKill: job_cancel kills a running job (cancelled, no code); a cancel after the end resends the unacked
++// job_exit and is ignored once acked; a job cancelled before it started never starts.
++func TestCancelKill(t *testing.T) {
++	w, s := newRunWorker(t, config.WorkerCaps{MaxJobs: 2}, "sleep")
++	j := testJob(1, "sleep")
++	run(t, w, j)
++	s.wait(t, protocol.TypeJobStarted, j.ID, 1, 10*time.Second)
++	if !w.Cancel(j.ID) {
++		t.Fatal("running job unknown")
++	}
++	ex := s.exit(t, j.ID)
++	if ex.Reason != protocol.ExitCancelled || ex.Code != nil || ex.Usage.StartedMS == nil {
++		t.Fatalf("%+v", ex)
++	}
++	w.Cancel(j.ID)
++	s.wait(t, protocol.TypeJobExit, j.ID, 2, 5*time.Second)
++	w.ExitAck(j.ID)
++	if w.Cancel(j.ID) {
++		t.Fatal("cancel of an acked job was not ignored")
++	}
++
++	p := testJob(2, "sleep")
++	if f, r := w.Admit(p); f != "" {
++		t.Fatalf("%s %s", f, r)
++	}
++	w.Cancel(p.ID)
++	w.Launch(p.ID)
++	ex = s.exit(t, p.ID)
++	if ex.Reason != protocol.ExitCancelled || ex.Usage.StartedMS != nil || ex.Usage.WallMS != 0 || s.count(protocol.TypeJobStarted, p.ID) != 0 {
++		t.Fatalf("%+v", ex)
++	}
++}
++
++// TestOutputCap: stdout past max_output_bytes is cut at exactly the cap and the job is killed (limit); chunks are
++// numbered from 1 and never larger than chunkBytes.
++func TestOutputCap(t *testing.T) {
++	const capBytes = 64 << 10
++	w, s := newRunWorker(t, config.WorkerCaps{MaxOutputBytes: capBytes}, "flood")
++	j := testJob(1, "flood")
++	run(t, w, j)
++	ex := s.exit(t, j.ID)
++	if ex.Reason != protocol.ExitLimit || ex.Code != nil {
++		t.Fatalf("%+v", ex)
++	}
++	total, seq := 0, uint64(0)
++	for _, m := range s.all() {
++		if f, ok := m.(protocol.JobStdout); ok {
++			if f.ChunkSeq != seq+1 || len(f.Chunk) > chunkBytes {
++				t.Fatalf("chunk %d after %d, %d bytes", f.ChunkSeq, seq, len(f.Chunk))
++			}
++			seq = f.ChunkSeq
++			total += len(f.Chunk)
++		}
++	}
++	if total != capBytes {
++		t.Fatalf("%d stdout bytes sent, cap %d", total, capBytes)
++	}
++}
++
++// TestLatchKillsAndRefuses: setting the stop latch kills every running job at once (stopped) and refuses new jobs with
++// the latch's fault; once cleared, jobs run again.
++func TestLatchKillsAndRefuses(t *testing.T) {
++	w, s := newRunWorker(t, config.WorkerCaps{MaxJobs: 2}, "sleep")
++	a, b := testJob(1, "sleep"), testJob(2, "sleep")
++	run(t, w, a)
++	run(t, w, b)
++	s.wait(t, protocol.TypeJobStarted, a.ID, 1, 10*time.Second)
++	s.wait(t, protocol.TypeJobStarted, b.ID, 1, 10*time.Second)
++	w.SetStopped(protocol.FaultEstopped)
++	for _, j := range []protocol.Job{a, b} {
++		if ex := s.exit(t, j.ID); ex.Reason != protocol.ExitStopped || ex.Code != nil {
++			t.Fatalf("%+v", ex)
++		}
++	}
++	if f, r := w.Admit(testJob(3, "sleep")); f != protocol.FaultEstopped || r != protocol.JobStopped {
++		t.Fatalf("while latched: %q %q", f, r)
++	}
++	w.SetStopped("")
++	c := testJob(4, "sleep")
++	run(t, w, c)
++	s.wait(t, protocol.TypeJobStarted, c.ID, 1, 10*time.Second)
++	w.Cancel(c.ID)
++	if ex := s.exit(t, c.ID); ex.Reason != protocol.ExitCancelled {
++		t.Fatalf("%+v", ex)
++	}
++}
ddiff --git a/internal/worker/worker.go b/internal/worker/worker.go
nnew file mode 100644
iindex 0000000..412bcd3
---- /dev/null
++++ b/internal/worker/worker.go
@@@ -0,0 +1,517 @@
++// Package worker runs `function` jobs (OpenVibe.Contracts platform.job@1) from the control link. It runs only the
++// functions the owner declared in the local config, each job as one child process in its own process group and in new
++// user, network and PID namespaces (no network at all), with a scrubbed environment and a fresh working directory, and
++// kills it at its ttl, at its limits, on job_cancel and while the stop latch is set. It never runs a job with less
++// isolation: a job whose namespaces cannot be created fails without running, and a Node whose boot-time Probe fails
++// does not advertise the class. docs/worker.md has the model and its limits.
++package worker
++
++import (
++	"bytes"
++	"encoding/json"
++	"io"
++	"log/slog"
++	"os"
++	"os/exec"
++	"sort"
++	"strings"
++	"sync"
++	"time"
++	"unicode/utf8"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++const (
++	chunkBytes  = 16 << 10  // the most stdout bytes one job_stdout carries
++	resultBytes = 256 << 10 // the most bytes of result a function may write to fd 3
++	sampleEvery = 200 * time.Millisecond
++	endedMax    = 1024 // ended jobs kept for their job_exit until job_exit_ack
++)
++
++// Worker runs function jobs and sends their frames with send.
++type Worker struct {
++	cfg  config.WorkerConfig
++	caps config.WorkerCaps
++	send func(protocol.Message)
++	log  *slog.Logger
++
++	// isolate puts a job's command in its namespaces; tests replace it to stand for a kernel that cannot.
++	isolate func(*exec.Cmd, config.WorkerConfig) error
++
++	mu      sync.Mutex
++	stopped string          // the fault code while the stop latch is set; "" when clear
++	closed  bool            // shutting down
++	jobs    map[string]*job // admitted, running, or ended with its job_exit not yet acked
++	ended   []string        // ids of ended jobs in jobs, oldest first
++	wg      sync.WaitGroup
++}
++
++type job struct {
++	req      protocol.Job
++	fn       config.FunctionConfig
++	received time.Time
++	ttl      time.Duration
++	limits   protocol.JobLimits // the job's, clamped to the local caps
++
++	reason    string      // why it was killed (the first reason wins); "" while nothing killed it
++	launched  bool        // Launch was called
++	proc      *os.Process // set while the process runs
++	startedMS int64
++	exit      *protocol.JobExit // set once it ended
++}
++
++// New returns a Worker for cfg. It runs nothing until Probe has passed and a job is admitted and launched.
++func New(cfg config.WorkerConfig, send func(protocol.Message), log *slog.Logger) *Worker {
++	if log == nil {
++		log = slog.Default()
++	}
++	return &Worker{cfg: cfg, caps: cfg.Caps.WithDefaults(), send: send, log: log, isolate: isolate, jobs: map[string]*job{}}
++}
++
++// Probe starts a process the way a job is started and checks that it got user, network and PID namespaces of its own
++// and that its CPU limit can be set and its usage read. The class must not be advertised when Probe fails.
++func (w *Worker) Probe() error { return probe(w.cfg, w.isolate) }
++
++// Admit reserves a job, or returns the fault code and reason to nack it with. It runs nothing (Launch does, once the
++// ack is out). An id the worker already holds is admitted again without a second reservation.
++func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	if _, ok := w.jobs[j.ID]; ok {
++		return "", ""
++	}
++	if w.stopped != "" {
++		return w.stopped, protocol.JobStopped
++	}
++	if w.closed {
++		return protocol.FaultShuttingDown, protocol.JobNotAvailable
++	}
++	if j.Class != protocol.ClassFunction {
++		return protocol.FaultUnsupported, protocol.JobNotAvailable
++	}
++	fn, ok := w.function(j.Artifact)
++	if !ok {
++		return protocol.FaultUnsupported, protocol.JobUnknownArtifact
++	}
++	running := 0
++	for _, jb := range w.jobs {
++		if jb.exit == nil {
++			running++
++		}
++	}
++	if running >= w.caps.MaxJobs {
++		return protocol.FaultNotReady, protocol.JobBusy
++	}
++	w.jobs[j.ID] = &job{req: j, fn: fn, received: time.Now(), ttl: ms(min(j.TTLMS, w.caps.MaxTTLMS)),
++		limits: protocol.JobLimits{WallMS: min(j.Limits.WallMS, w.caps.MaxWallMS), CPUMS: min(j.Limits.CPUMS, w.caps.MaxCPUMS),
++			MemBytes: min(j.Limits.MemBytes, w.caps.MaxMemBytes)}}
++	return "", ""
++}
++
++func (w *Worker) function(a *protocol.Artifact) (config.FunctionConfig, bool) {
++	if a == nil {
++		return config.FunctionConfig{}, false
++	}
++	for _, f := range w.cfg.Functions {
++		if f.Name == a.Name && f.Version == a.Version {
++			return f, true
++		}
++	}
++	return config.FunctionConfig{}, false
++}
++
++// Launch runs an admitted job on its own goroutine. A job already launched, or unknown, is left alone.
++func (w *Worker) Launch(id string) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	jb, ok := w.jobs[id]
++	if !ok || jb.launched || w.closed {
++		return
++	}
++	jb.launched = true
++	w.wg.Add(1)
++	go w.run(jb)
++}
++
++// Known returns what a resent job frame for id is answered with: an ack while the job waits or runs, its job_exit
++// once it ended (until job_exit_ack).
++func (w *Worker) Known(id string) (protocol.Message, bool) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	jb, ok := w.jobs[id]
++	switch {
++	case !ok:
++		return nil, false
++	case jb.exit != nil:
++		return *jb.exit, true
++	}
++	return protocol.Ack{ID: id}, true
++}
++
++// Cancel kills a job (its job_exit says cancelled; one not started yet never starts). For an ended job whose job_exit
++// is not acked yet it resends that job_exit. An unknown id is ignored; Cancel reports whether id was known.
++func (w *Worker) Cancel(id string) bool {
++	w.mu.Lock()
++	jb, ok := w.jobs[id]
++	var resend *protocol.JobExit
++	if ok {
++		resend = jb.exit
++		w.killLocked(jb, protocol.ExitCancelled)
++	}
++	w.mu.Unlock()
++	if resend != nil {
++		w.send(*resend)
++	}
++	return ok
++}
++
++// ExitAck forgets an ended job: its usage is recorded, so its job_exit is not resent any more.
++func (w *Worker) ExitAck(id string) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	if jb, ok := w.jobs[id]; !ok || jb.exit == nil {
++		return
++	}
++	delete(w.jobs, id)
++	for i, e := range w.ended {
++		if e == id {
++			w.ended = append(w.ended[:i], w.ended[i+1:]...)
++			break
++		}
++	}
++}
++
++// SetStopped follows the stop latch (the e-stop or the local stop). A fault code kills every job at once (job_exit
++// says stopped) and refuses new ones with that fault until SetStopped("").
++func (w *Worker) SetStopped(fault string) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	w.stopped = fault
++	if fault == "" {
++		return
++	}
++	for _, jb := range w.jobs {
++		w.killLocked(jb, protocol.ExitStopped)
++	}
++}
++
++// Resend sends again, after a reconnect, job_started for every running job and job_exit for every ended job whose
++// job_exit is not acked yet.
++func (w *Worker) Resend() {
++	w.mu.Lock()
++	var msgs []protocol.Message
++	for _, jb := range w.jobs {
++		if jb.exit == nil && jb.proc != nil {
++			msgs = append(msgs, protocol.JobStarted{ID: jb.req.ID, StartedMS: jb.startedMS})
++		}
++	}
++	for _, id := range w.ended {
++		msgs = append(msgs, *w.jobs[id].exit)
++	}
++	w.mu.Unlock()
++	for _, m := range msgs {
++		w.send(m)
++	}
++}
++
++// Close kills every job (job_exit says stopped), refuses new ones and waits up to 5 s for them to end.
++func (w *Worker) Close() {
++	w.mu.Lock()
++	w.closed = true
++	for _, jb := range w.jobs {
++		w.killLocked(jb, protocol.ExitStopped)
++	}
++	w.mu.Unlock()
++	done := make(chan struct{})
++	go func() { w.wg.Wait(); close(done) }()
++	select {
++	case <-done:
++	case <-time.After(5 * time.Second):
++		w.log.Error("jobs did not end within 5 s of being killed")
++	}
++}
++
++func (w *Worker) kill(jb *job, reason string) {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	w.killLocked(jb, reason)
++}
++
++// killLocked records why jb is killed and SIGKILLs its process group; a job not started yet then never starts.
++func (w *Worker) killLocked(jb *job, reason string) {
++	if jb.exit != nil {
++		return
++	}
++	if jb.reason == "" {
++		jb.reason = reason
++	}
++	if jb.proc != nil {
++		killGroup(jb.proc)
++	}
++}
++
++func (w *Worker) run(jb *job) {
++	defer w.wg.Done()
++	ex := w.execute(jb)
++	w.mu.Lock()
++	jb.exit = &ex
++	w.ended = append(w.ended, ex.ID)
++	for len(w.ended) > endedMax {
++		delete(w.jobs, w.ended[0])
++		w.ended = w.ended[1:]
++	}
++	w.mu.Unlock()
++	w.log.Info("job ended", "id", ex.ID, "reason", ex.Reason, "wall_ms", ex.Usage.WallMS)
++	w.send(ex)
++}
++
++// execute runs jb's process to its end and returns its job_exit.
++func (w *Worker) execute(jb *job) protocol.JobExit {
++	id := jb.req.ID
++	never := func(reason string, err error) protocol.JobExit {
++		if err != nil {
++			w.log.Error("job not run", "id", id, "err", err)
++		}
++		return protocol.JobExit{ID: id, Reason: reason, Result: json.RawMessage("null")}
++	}
++	w.mu.Lock()
++	reason := jb.reason
++	w.mu.Unlock()
++	if reason != "" { // cancelled or stopped before it started
++		return never(reason, nil)
++	}
++	dir, err := os.MkdirTemp("", "openvibe-job-")
++	if err != nil {
++		return never(protocol.ExitFailed, err)
++	}
++	defer os.RemoveAll(dir)
++	if r := w.cfg.RunAs; r != nil {
++		if err := os.Chown(dir, int(r.UID), int(r.GID)); err != nil {
++			return never(protocol.ExitFailed, err)
++		}
++	}
++	cmd := exec.Command(jb.fn.Command[0], jb.fn.Command[1:]...)
++	cmd.Dir, cmd.Env = dir, jobEnv(jb.fn, dir, id)
++	cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))
++	if err := w.isolate(cmd, w.cfg); err != nil {
++		return never(protocol.ExitFailed, err)
++	}
++	outR, outW, err := os.Pipe()
++	if err != nil {
++		return never(protocol.ExitFailed, err)
++	}
++	defer outR.Close()
++	resR, resW, err := os.Pipe()
++	if err != nil {
++		outW.Close()
++		return never(protocol.ExitFailed, err)
++	}
++	defer resR.Close()
++	cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}
++
++	w.mu.Lock()
++	if jb.reason == "" && time.Since(jb.received) >= jb.ttl {
++		jb.reason = protocol.ExitTTL
++	}
++	reason = jb.reason
++	var start time.Time
++	if reason == "" {
++		err = cmd.Start()
++		start = time.Now()
++		if err == nil {
++			jb.proc, jb.startedMS = cmd.Process, start.UnixMilli()
++		}
++	}
++	w.mu.Unlock()
++	outW.Close()
++	resW.Close()
++	if reason != "" {
++		return never(reason, nil)
++	}
++	if err != nil {
++		return never(protocol.ExitFailed, err)
++	}
++	pid := cmd.Process.Pid
++	w.send(protocol.JobStarted{ID: id, StartedMS: jb.startedMS})
++	if err := limitCPU(pid, jb.limits.CPUMS); err != nil {
++		w.log.Error("job killed: its CPU limit could not be set", "id", id, "err", err)
++		w.kill(jb, protocol.ExitFailed)
++	}
++
++	stop, watched := make(chan struct{}), make(chan accounting, 1)
++	go func() { watched <- w.watch(jb, pid, start, stop) }()
++	var result []byte
++	var readers sync.WaitGroup
++	readers.Add(2)
++	go func() { defer readers.Done(); w.pump(jb, outR) }()
++	go func() { defer readers.Done(); result = w.readResult(jb, resR) }()
++	readers.Wait()
++	_ = cmd.Wait()
++	wall := time.Since(start)
++	close(stop)
++	acc := <-watched
++
++	w.mu.Lock()
++	reason = jb.reason
++	jb.proc = nil
++	w.mu.Unlock()
++	code, cpuLimit := exitStatus(cmd.ProcessState)
++	if reason == "" {
++		reason = protocol.ExitExited
++		if cpuLimit {
++			reason = protocol.ExitLimit
++		}
++	}
++	ex := protocol.JobExit{ID: id, Reason: reason, Code: code, Result: json.RawMessage("null")}
++	if r := bytes.TrimSpace(result); reason == protocol.ExitExited && code != nil && *code == 0 && len(r) > 0 && json.Valid(r) {
++		ex.Result = json.RawMessage(r)
++	}
++	// wall_ms is at least (n+1) s for every usage second n already sent, as platform.job-frame@1 requires.
++	wallMS := max(wall.Milliseconds(), acc.seconds*1000)
++	cpuMS := max(acc.cpuMS, (cmd.ProcessState.UserTime() + cmd.ProcessState.SystemTime()).Milliseconds())
++	mem := max(acc.memPeak, maxRSS(cmd.ProcessState))
++	started := jb.startedMS
++	ex.Usage = protocol.JobExitUsage{StartedMS: &started, WallMS: wallMS, CPUMS: &cpuMS, MemPeakBytes: &mem}
++	return ex
++}
++
++// accounting is what watch measured: usage seconds sent, and the CPU time and peak resident memory sampled.
++type accounting struct{ seconds, cpuMS, memPeak int64 }
++
++// watch is a job's watchdog until stop closes: it kills the job at its ttl (from receipt), at wall_ms (from start)
++// and when sampled CPU time or resident memory exceed the limits, and sends job_usage for every wall-clock second
++// once it has fully elapsed.
++func (w *Worker) watch(jb *job, pid int, start time.Time, stop <-chan struct{}) accounting {
++	var acc accounting
++	ttl := time.NewTimer(time.Until(jb.received.Add(jb.ttl)))
++	defer ttl.Stop()
++	wall := time.NewTimer(time.Until(start.Add(ms(jb.limits.WallMS))))
++	defer wall.Stop()
++	usage := time.NewTimer(time.Until(start.Add(time.Second)))
++	defer usage.Stop()
++	tick := time.NewTicker(sampleEvery)
++	defer tick.Stop()
++	var ns string
++	var lastCPU int64
++	sample := func() {
++		if ns == "" {
++			ns, _ = pidNS(pid)
++		}
++		cpu, rss, err := sampleProcs(pid, ns)
++		if err != nil {
++			return
++		}
++		acc.cpuMS, acc.memPeak = max(acc.cpuMS, cpu), max(acc.memPeak, rss)
++		if acc.cpuMS > jb.limits.CPUMS || rss > jb.limits.MemBytes {
++			w.kill(jb, protocol.ExitLimit)
++		}
++	}
++	sample()
++	for {
++		select {
++		case <-stop:
++			return acc
++		case <-ttl.C:
++			w.kill(jb, protocol.ExitTTL)
++		case <-wall.C:
++			w.kill(jb, protocol.ExitLimit)
++		case <-tick.C:
++			sample()
++		case <-usage.C:
++			for time.Since(start) >= time.Duration(acc.seconds+1)*time.Second {
++				cpu := acc.cpuMS - lastCPU
++				lastCPU = acc.cpuMS
++				w.send(protocol.JobUsage{ID: jb.req.ID, StartedMS: jb.startedMS, Second: acc.seconds, CPUMS: &cpu})
++				acc.seconds++
++			}
++			usage.Reset(time.Until(start.Add(time.Duration(acc.seconds+1) * time.Second)))
++		}
++	}
++}
++
++// pump sends the job's stdout as job_stdout chunks of at most chunkBytes, never splitting a UTF-8 sequence, and
++// kills the job (limit) once it wrote more than max_output_bytes; the excess is read and dropped.
++func (w *Worker) pump(jb *job, r io.Reader) {
++	buf := make([]byte, chunkBytes-utf8.UTFMax) // room for a rune carried over from the last read
++	var carry []byte
++	var total int64
++	var seq uint64
++	over := false
++	emit := func(b []byte) {
++		seq++
++		w.send(protocol.JobStdout{ID: jb.req.ID, ChunkSeq: seq, Chunk: strings.ToValidUTF8(string(b), "�")})
++	}
++	for {
++		n, err := r.Read(buf)
++		if n > 0 && !over {
++			b := buf[:n]
++			if left := w.caps.MaxOutputBytes - total; int64(n) > left {
++				b, over = b[:left], true
++			}
++			total += int64(len(b))
++			b = append(carry, b...)
++			cut := completeRunes(b)
++			if over {
++				cut = len(b)
++			}
++			if cut > 0 {
++				emit(b[:cut])
++			}
++			carry = append([]byte(nil), b[cut:]...)
++			if over {
++				w.kill(jb, protocol.ExitLimit)
++			}
++		}
++		if err != nil {
++			if len(carry) > 0 {
++				emit(carry)
++			}
++			return
++		}
++	}
++}
++
++// completeRunes is how many leading bytes of b end on a rune boundary: a rune cut by the end of a read waits for
++// the next read. Invalid bytes count as complete (they become U+FFFD).
++func completeRunes(b []byte) int {
++	for i := len(b) - 1; i >= 0 && i >= len(b)-utf8.UTFMax; i-- {
++		if utf8.RuneStart(b[i]) {
++			if utf8.FullRune(b[i:]) {
++				return len(b)
++			}
++			return i
++		}
++	}
++	return len(b)
++}
++
++// readResult reads what the function writes to fd 3, its result; more than resultBytes kills the job (limit).
++func (w *Worker) readResult(jb *job, r io.Reader) []byte {
++	b, _ := io.ReadAll(io.LimitReader(r, resultBytes+1))
++	if len(b) > resultBytes {
++		w.kill(jb, protocol.ExitLimit)
++		_, _ = io.Copy(io.Discard, r)
++		return nil
++	}
++	return b
++}
++
++// jobEnv is the whole environment of a job: nothing is inherited from the Node (no credential, no secrets), only a
++// fixed PATH, the working directory as HOME and TMPDIR, the job id, and the function's own env from the config.
++func jobEnv(fn config.FunctionConfig, dir, id string) []string {
++	env := map[string]string{"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": dir, "TMPDIR": dir, "LANG": "C.UTF-8",
++		"OPENVIBE_JOB_ID": id}
++	for k, v := range fn.Env {
++		env[k] = v
++	}
++	out := make([]string, 0, len(env))
++	for k, v := range env {
++		out = append(out, k+"="+v)
++	}
++	sort.Strings(out)
++	return out
++}
++
++func ms(v int64) time.Duration { return time.Duration(v) * time.Millisecond }
ddiff --git a/internal/worker/worker_linux.go b/internal/worker/worker_linux.go
nnew file mode 100644
iindex 0000000..7cdbee9
---- /dev/null
++++ b/internal/worker/worker_linux.go
@@@ -0,0 +1,189 @@
++//go:build linux
++
++package worker
++
++import (
++	"errors"
++	"fmt"
++	"os"
++	"os/exec"
++	"strconv"
++	"strings"
++	"syscall"
++	"unsafe"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++)
++
++// clkTck is USER_HZ, the unit of /proc/<pid>/stat times: 100 on every architecture Go runs Linux on.
++const clkTck = 100
++
++var pageSize = int64(os.Getpagesize())
++
++// isolate starts cmd in its own process group and in new user, network and PID namespaces: no network but a loopback
++// that is down, and when the process exits every process it left behind is killed with it. It runs as worker.run_as
++// (with no supplementary groups; the Node must be root) or, with allow_same_user, as the Node's own non-root user.
++func isolate(cmd *exec.Cmd, cfg config.WorkerConfig) error {
++	uid, gid := os.Geteuid(), os.Getegid()
++	attr := &syscall.SysProcAttr{Setpgid: true, Pdeathsig: syscall.SIGKILL,
++		Cloneflags: syscall.CLONE_NEWUSER | syscall.CLONE_NEWNET | syscall.CLONE_NEWPID}
++	switch {
++	case cfg.RunAs != nil:
++		if uid != 0 {
++			return errors.New("worker.run_as needs the Node to run as root")
++		}
++		r := cfg.RunAs
++		attr.UidMappings = []syscall.SysProcIDMap{{ContainerID: int(r.UID), HostID: int(r.UID), Size: 1}}
++		attr.GidMappings = []syscall.SysProcIDMap{{ContainerID: int(r.GID), HostID: int(r.GID), Size: 1}}
++		attr.GidMappingsEnableSetgroups = true
++		attr.Credential = &syscall.Credential{Uid: r.UID, Gid: r.GID} // no Groups: setgroups(0) drops root's groups
++	case cfg.AllowSameUser:
++		if uid == 0 {
++			return errors.New("the Node runs as root: set worker.run_as instead of allow_same_user")
++		}
++		attr.UidMappings = []syscall.SysProcIDMap{{ContainerID: uid, HostID: uid, Size: 1}}
++		attr.GidMappings = []syscall.SysProcIDMap{{ContainerID: gid, HostID: gid, Size: 1}}
++	default:
++		return errors.New("set worker.run_as (or allow_same_user, for development) to run jobs")
++	}
++	cmd.SysProcAttr = attr
++	return nil
++}
++
++// probe starts /bin/sh the way a job is started and checks that it got user, network and PID namespaces of its own,
++// that its CPU limit can be set and that its usage can be read.
++func probe(cfg config.WorkerConfig, isolate func(*exec.Cmd, config.WorkerConfig) error) error {
++	cmd := exec.Command("/bin/sh", "-c", "read x")
++	cmd.Env, cmd.Dir = []string{"PATH=/usr/bin:/bin"}, "/"
++	if err := isolate(cmd, cfg); err != nil {
++		return err
++	}
++	in, err := cmd.StdinPipe()
++	if err != nil {
++		return err
++	}
++	if err := cmd.Start(); err != nil {
++		in.Close()
++		return fmt.Errorf("starting a process in new user, network and PID namespaces: %w", err)
++	}
++	defer func() { killGroup(cmd.Process); in.Close(); _ = cmd.Wait() }()
++	pid := cmd.Process.Pid
++	for _, ns := range []string{"user", "net", "pid"} {
++		mine, err1 := os.Readlink("/proc/self/ns/" + ns)
++		its, err2 := os.Readlink("/proc/" + strconv.Itoa(pid) + "/ns/" + ns)
++		if err1 != nil || err2 != nil || mine == its {
++			return fmt.Errorf("the probe process did not get a %s namespace of its own (%v, %v)", ns, err1, err2)
++		}
++	}
++	if err := limitCPU(pid, 1000); err != nil {
++		return err
++	}
++	if _, _, err := statOf(pid); err != nil {
++		return fmt.Errorf("reading the probe's CPU and memory: %w", err)
++	}
++	return nil
++}
++
++// killGroup SIGKILLs the job's process group while its leader lives, then the leader; the leader is PID 1 of the
++// job's PID namespace, so its death also kills anything that left the group.
++func killGroup(p *os.Process) {
++	if p.Signal(syscall.Signal(0)) == nil {
++		_ = syscall.Kill(-p.Pid, syscall.SIGKILL)
++	}
++	_ = p.Kill()
++}
++
++// limitCPU sets the process's RLIMIT_CPU (SIGXCPU at cpuMS rounded up to a second, SIGKILL a second later; every
++// process it starts inherits it) and RLIMIT_CORE 0: the kernel's backstop behind the sampled cpu_ms check.
++func limitCPU(pid int, cpuMS int64) error {
++	sec := uint64((cpuMS + 999) / 1000)
++	limits := []struct {
++		res int
++		lim syscall.Rlimit
++	}{{syscall.RLIMIT_CPU, syscall.Rlimit{Cur: sec, Max: sec + 1}}, {syscall.RLIMIT_CORE, syscall.Rlimit{}}}
++	for _, l := range limits {
++		lim := l.lim
++		if _, _, e := syscall.RawSyscall6(syscall.SYS_PRLIMIT64, uintptr(pid), uintptr(l.res), uintptr(unsafe.Pointer(&lim)), 0, 0, 0); e != 0 {
++			return fmt.Errorf("prlimit: %w", e)
++		}
++	}
++	return nil
++}
++
++func pidNS(pid int) (string, error) { return os.Readlink("/proc/" + strconv.Itoa(pid) + "/ns/pid") }
++
++// sampleProcs sums the CPU time and resident memory of every process in the job's PID namespace ns (the leader alone
++// while ns is unknown), so a process that left the process group is still counted.
++func sampleProcs(pid int, ns string) (cpuMS, rssBytes int64, err error) {
++	if ns == "" {
++		return statOf(pid)
++	}
++	ents, err := os.ReadDir("/proc")
++	if err != nil {
++		return 0, 0, err
++	}
++	for _, e := range ents {
++		p, err := strconv.Atoi(e.Name())
++		if err != nil {
++			continue
++		}
++		if l, err := os.Readlink("/proc/" + e.Name() + "/ns/pid"); err != nil || l != ns {
++			continue
++		}
++		if c, r, err := statOf(p); err == nil {
++			cpuMS, rssBytes = cpuMS+c, rssBytes+r
++		}
++	}
++	return cpuMS, rssBytes, nil
++}
++
++// statOf reads one process's CPU time (its own and its reaped children's) and resident memory from /proc.
++func statOf(pid int) (cpuMS, rssBytes int64, err error) {
++	b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat")
++	if err != nil {
++		return 0, 0, err
++	}
++	s := string(b)
++	i := strings.LastIndexByte(s, ')')
++	if i < 0 {
++		return 0, 0, errors.New("malformed /proc stat")
++	}
++	f := strings.Fields(s[i+1:]) // f[0] is field 3, state
++	if len(f) < 22 {
++		return 0, 0, errors.New("short /proc stat")
++	}
++	var ticks int64
++	for _, k := range []int{11, 12, 13, 14} { // utime, stime, cutime, cstime
++		v, err := strconv.ParseInt(f[k], 10, 64)
++		if err != nil {
++			return 0, 0, err
++		}
++		ticks += v
++	}
++	rss, err := strconv.ParseInt(f[21], 10, 64)
++	if err != nil {
++		return 0, 0, err
++	}
++	return ticks * 1000 / clkTck, rss * pageSize, nil
++}
++
++// exitStatus is the process's exit code (nil when a signal ended it) and whether RLIMIT_CPU ended it.
++func exitStatus(ps *os.ProcessState) (code *int, cpuLimit bool) {
++	ws, ok := ps.Sys().(syscall.WaitStatus)
++	if !ok {
++		return nil, false
++	}
++	if ws.Exited() {
++		c := ws.ExitStatus()
++		return &c, false
++	}
++	return nil, ws.Signaled() && ws.Signal() == syscall.SIGXCPU
++}
++
++// maxRSS is the peak resident memory the kernel reports for the process and the children it waited for.
++func maxRSS(ps *os.ProcessState) int64 {
++	if r, ok := ps.SysUsage().(*syscall.Rusage); ok {
++		return int64(r.Maxrss) * 1024
++	}
++	return 0
++}
ddiff --git a/internal/worker/worker_other.go b/internal/worker/worker_other.go
nnew file mode 100644
iindex 0000000..30ec593
---- /dev/null
++++ b/internal/worker/worker_other.go
@@@ -0,0 +1,35 @@
++//go:build !linux
++
++package worker
++
++import (
++	"errors"
++	"os"
++	"os/exec"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++)
++
++// errNotLinux: the namespaces the worker isolates jobs with exist on Linux only, so elsewhere no job ever runs.
++var errNotLinux = errors.New("the job worker runs on Linux only")
++
++func isolate(*exec.Cmd, config.WorkerConfig) error { return errNotLinux }
++
++func probe(config.WorkerConfig, func(*exec.Cmd, config.WorkerConfig) error) error { return errNotLinux }
++
++func killGroup(p *os.Process) { _ = p.Kill() }
++
++func limitCPU(int, int64) error { return errNotLinux }
++
++func pidNS(int) (string, error) { return "", errNotLinux }
++
++func sampleProcs(int, string) (int64, int64, error) { return 0, 0, errNotLinux }
++
++func exitStatus(ps *os.ProcessState) (*int, bool) {
++	if c := ps.ExitCode(); c >= 0 {
++		return &c, false
++	}
++	return nil, false
++}
++
++func maxRSS(*os.ProcessState) int64 { return 0 }
ddiff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go
nnew file mode 100644
iindex 0000000..c93ed0a
---- /dev/null
++++ b/internal/worker/worker_test.go
@@@ -0,0 +1,187 @@
++package worker
++
++import (
++	"encoding/json"
++	"errors"
++	"fmt"
++	"io"
++	"log/slog"
++	"os"
++	"os/exec"
++	"sync"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// sink records every frame the worker sends.
++type sink struct {
++	mu     sync.Mutex
++	frames []protocol.Message
++}
++
++func (s *sink) send(m protocol.Message) {
++	s.mu.Lock()
++	defer s.mu.Unlock()
++	s.frames = append(s.frames, m)
++}
++
++func (s *sink) all() []protocol.Message {
++	s.mu.Lock()
++	defer s.mu.Unlock()
++	return append([]protocol.Message(nil), s.frames...)
++}
++
++func frameID(m protocol.Message) string {
++	switch f := m.(type) {
++	case protocol.JobStarted:
++		return f.ID
++	case protocol.JobStdout:
++		return f.ID
++	case protocol.JobUsage:
++		return f.ID
++	case protocol.JobExit:
++		return f.ID
++	}
++	return ""
++}
++
++// count is how many frames of type typ for job id were sent.
++func (s *sink) count(typ, id string) int {
++	n := 0
++	for _, m := range s.all() {
++		if m.MessageType() == typ && frameID(m) == id {
++			n++
++		}
++	}
++	return n
++}
++
++// wait returns the n-th (from 1) frame of type typ for job id, waiting up to timeout for it.
++func (s *sink) wait(t *testing.T, typ, id string, n int, timeout time.Duration) protocol.Message {
++	t.Helper()
++	end := time.Now().Add(timeout)
++	for {
++		seen := 0
++		for _, m := range s.all() {
++			if m.MessageType() == typ && frameID(m) == id {
++				if seen++; seen == n {
++					return m
++				}
++			}
++		}
++		if time.Now().After(end) {
++			t.Fatalf("no %s #%d for %s within %s", typ, n, id, timeout)
++		}
++		time.Sleep(10 * time.Millisecond)
++	}
++}
++
++func (s *sink) exit(t *testing.T, id string) protocol.JobExit {
++	t.Helper()
++	return s.wait(t, protocol.TypeJobExit, id, 1, 20*time.Second).(protocol.JobExit)
++}
++
++func testJob(n int, name string) protocol.Job {
++	return protocol.Job{ID: fmt.Sprintf("job_01JAB2C3D4E5F6G7H8J9K%05d", n), Class: protocol.ClassFunction,
++		Artifact: &protocol.Artifact{Name: name, Version: "1.0.0"}, Args: json.RawMessage(`{"n":7}`), TTLMS: 60000,
++		Limits: protocol.JobLimits{WallMS: 60000, CPUMS: 60000, MemBytes: 4 << 30}}
++}
++
++// helperConfig declares one function per helper mode, each running this test binary's TestHelperProcess.
++func helperConfig(t *testing.T, caps config.WorkerCaps, modes ...string) config.WorkerConfig {
++	t.Helper()
++	exe, err := os.Executable()
++	if err != nil {
++		t.Fatal(err)
++	}
++	cfg := config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: caps}
++	for _, m := range modes {
++		cfg.Functions = append(cfg.Functions, config.FunctionConfig{Name: m, Version: "1.0.0",
++			Command: []string{exe, "-test.run=^TestHelperProcess$", "--", m}})
++	}
++	return cfg
++}
++
++func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
++
++// TestAdmitRefusals: only a declared function at its exact version is admitted, never another class; max_jobs bounds
++// the jobs held at once; while the stop latch is set every job is refused with its fault; nothing ever starts.
++func TestAdmitRefusals(t *testing.T) {
++	s := &sink{}
++	w := New(helperConfig(t, config.WorkerCaps{MaxJobs: 1}, "sleep"), s.send, quiet())
++	w.isolate = func(*exec.Cmd, config.WorkerConfig) error { return errors.New("not in this test") }
++
++	wrongVersion := testJob(2, "sleep")
++	wrongVersion.Artifact.Version = "1.0.1"
++	code := testJob(3, "sleep")
++	code.Class = protocol.ClassCode
++	for _, c := range []struct {
++		j             protocol.Job
++		fault, reason string
++	}{
++		{testJob(1, "nope"), protocol.FaultUnsupported, protocol.JobUnknownArtifact},
++		{wrongVersion, protocol.FaultUnsupported, protocol.JobUnknownArtifact},
++		{code, protocol.FaultUnsupported, protocol.JobNotAvailable},
++	} {
++		if f, r := w.Admit(c.j); f != c.fault || r != c.reason {
++			t.Errorf("%s %+v: %q %q, want %q %q", c.j.Class, c.j.Artifact, f, r, c.fault, c.reason)
++		}
++	}
++
++	held := testJob(4, "sleep")
++	if f, r := w.Admit(held); f != "" {
++		t.Fatalf("%s %s", f, r)
++	}
++	if f, r := w.Admit(held); f != "" {
++		t.Fatalf("a held id re-admitted: %s %s", f, r)
++	}
++	if f, r := w.Admit(testJob(5, "sleep")); f != protocol.FaultNotReady || r != protocol.JobBusy {
++		t.Fatalf("over max_jobs: %q %q", f, r)
++	}
++	w.Cancel(held.ID)
++	w.Launch(held.ID)
++	if ex := s.exit(t, held.ID); ex.Reason != protocol.ExitCancelled {
++		t.Fatalf("%+v", ex)
++	}
++
++	w.SetStopped(protocol.FaultLocalStop)
++	if f, r := w.Admit(testJob(6, "sleep")); f != protocol.FaultLocalStop || r != protocol.JobStopped {
++		t.Fatalf("latched: %q %q", f, r)
++	}
++	w.SetStopped("")
++	if f, r := w.Admit(testJob(6, "sleep")); f != "" {
++		t.Fatalf("after the latch cleared: %q %q", f, r)
++	}
++	for _, m := range s.all() {
++		if _, ok := m.(protocol.JobStarted); ok {
++			t.Fatalf("a refused job started: %+v", m)
++		}
++	}
++}
++
++// TestIsolationUnavailableRefused: when the namespaces cannot be created the probe fails (the class is not
++// advertised), and a job whose isolation fails ends `failed` without its process ever starting.
++func TestIsolationUnavailableRefused(t *testing.T) {
++	s := &sink{}
++	w := New(helperConfig(t, config.WorkerCaps{}, "sleep"), s.send, quiet())
++	w.isolate = func(*exec.Cmd, config.WorkerConfig) error { return errors.New("no user namespaces") }
++	t.Cleanup(w.Close)
++	if err := w.Probe(); err == nil {
++		t.Fatal("the probe passed without namespaces")
++	}
++	j := testJob(1, "sleep")
++	if f, r := w.Admit(j); f != "" {
++		t.Fatalf("%s %s", f, r)
++	}
++	w.Launch(j.ID)
++	ex := s.exit(t, j.ID)
++	if ex.Reason != protocol.ExitFailed || ex.Code != nil || ex.Usage.WallMS != 0 || ex.Usage.StartedMS != nil || string(ex.Result) != "null" {
++		t.Fatalf("%+v", ex)
++	}
++	if n := s.count(protocol.TypeJobStarted, j.ID); n != 0 {
++		t.Fatalf("job_started sent %d times for a job that could not be isolated", n)
++	}
++}