Plan T14, follow-up 4 of the Node gap audit… (#13)

ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex 68b0bf9..08af9eb 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -486,6 +486,12 @@ func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {
  		}
  		fmt.Fprintln(stdout)
  	}
++	if len(s.Jobs) > 0 {
++		fmt.Fprintln(stdout, "Jobs:")
++		for _, j := range s.Jobs {
++			fmt.Fprintf(stdout, "  %s %s %s\n", j.ID, j.Function, j.State)
++		}
++	}
  	if b := s.Telemetry.Battery; b != nil {
  		fmt.Fprintf(stdout, "Battery:  %.0f%%\n", *b*100)
  	}
ddiff --git a/docs/install.md b/docs/install.md
iindex aabf7f7..73d5cfb 100644
---- a/docs/install.md
++++ b/docs/install.md
@@@ -61,7 +61,7 @@ need Python; set `"python"` in `config.json`.
  | `openvibe-node run [--dry-run]`           | run in the foreground (what the service runs)                              |
  | `openvibe-node install [--user NAME]`     | install and start the service                                              |
  | `openvibe-node uninstall`                 | stop and remove the service (config and credential are kept)               |
--| `openvibe-node status [--json]`           | link, stop latch, limits, plugins, video, battery                          |
++| `openvibe-node status [--json]`           | link, stop latch, limits, plugins, video, battery, running jobs            |
  | `openvibe-node stop`                      | **kill switch**: stop every actuator now and hold it stopped              |
  | `openvibe-node resume`                    | release the local stop (and a server e-stop)                               |
  | `openvibe-node plugins [--json]`          | plugins and their capabilities                                             |
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 7cfcbfb..1a38a68 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -216,12 +216,15 @@ A resent `job` with an id already answered gets the same answer again, never a s
  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
++`job_stdout` chunks (at most 16 KiB each as sent and `max_output_bytes` in all, valid UTF-8 with invalid bytes as
++U+FFFD, `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`:
++`job_exit` again on a `job` or `job_cancel` for that id, until `job_exit_ack`. `job_started` goes out only once the
++job's CPU limit is set; a job that fails before (no namespaces, spawn or CPU limit failed) ends `failed` without it.
++A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not started yet. `job_exit.reason`:
  
  | `reason`    | `code`  | when                                                                                                |
  |-------------|---------|-----------------------------------------------------------------------------------------------------|
ddiff --git a/docs/worker.md b/docs/worker.md
iindex cb65219..71199dc 100644
---- a/docs/worker.md
++++ b/docs/worker.md
@@@ -36,8 +36,8 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
    `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.
++  each clamped to them (the stricter value wins, as with `max_command_ms`). `max_output_bytes` caps the stdout bytes
++  one job sends (counted on the wire, after invalid UTF-8 became U+FFFD); `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,
@@@ -56,8 +56,10 @@ affected.
     `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
++   `RLIMIT_CORE` 0. `job_started` follows only once `RLIMIT_CPU` is set: if it cannot be set the process is killed
++   and the job ends `failed` with no `job_started`.
++3. **Run**: stdout streams as `job_stdout` chunks of at most 16 KiB each as sent (invalid UTF-8 becomes U+FFFD before
++   it is cut, and a UTF-8 sequence is never split); 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
@@@ -66,8 +68,12 @@ affected.
  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`).
++stop latches (`openvibe-node stop`); while the latch is set every job is refused, and `openvibe-node resume` admits
++them again. A Node shutting down kills its jobs (`stopped`); a job acked but not started yet never starts and gets its
++`job_exit` (`stopped`) all the same.
++
++`openvibe-node status` lists the jobs not ended yet under `jobs` (`--json`: `id`, `function` as `name@version`,
++`state` `admitted`, `running` or `stopping`, and `started_ms` once running); the list is empty when the worker is off.
  
  ## Isolation model and its limits
  
@@@ -89,8 +95,8 @@ What it does **not** protect against (no seccomp, no cgroups, no mount namespace
  - **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.
++  (and meet the kernel's OOM killer first); the `RLIMIT_CPU` backstop is set just after start (before `job_started`)
++  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
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex ce683e5..8bd6bc0 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -19,7 +19,6 @@ import (
  	"os"
  	"os/exec"
  	"runtime"
--	"slices"
  	"sort"
  	"strings"
  	"sync"
@@@ -44,11 +43,6 @@ type Options struct {
  	Log     *slog.Logger
  	Version string
  
--	// 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.
  	HeartbeatInterval time.Duration
  	LinkBackoffMin    time.Duration
@@@ -103,7 +97,7 @@ func New(opt Options) (*Node, error) {
  		opt.Log.Error("latch", "err", err)
  	}
  	n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),
--		jobs: newJobs(opt.RuntimeClasses), telemetry: map[string]map[string]any{}, telemMS: 500}
++		jobs: newJobs(nil), telemetry: map[string]map[string]any{}, telemMS: 500}
  	n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)
  
  	var specs []plugins.Spec
@@@ -161,10 +155,9 @@ func New(opt Options) (*Node, error) {
  		} 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)
--			}
++			// The only class advertised (status.capabilities.worker), and only with a worker to run it; with none,
++			// every job is refused "class not available".
++			n.jobs.admit, n.jobs.classes = w.Admit, []string{protocol.ClassFunction}
  			opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
  		}
  	}
@@@ -733,6 +726,8 @@ type Status struct {
  	ConfigDir string             `json:"config_dir"`
  	StateDir  string             `json:"state_dir"`
  	Telemetry protocol.Telemetry `json:"telemetry"`
++	// Jobs are the worker's jobs not ended yet; empty when the worker is off.
++	Jobs []worker.JobInfo `json:"jobs"`
  }
  
  type LinkStatus struct {
@@@ -782,6 +777,10 @@ func (n *Node) Status() Status {
  		}
  		s.Plugins = append(s.Plugins, ps)
  	}
++	s.Jobs = []worker.JobInfo{}
++	if n.worker != nil {
++		s.Jobs = n.worker.Jobs()
++	}
  	s.Video.State = video.StateOff
  	if n.pub != nil {
  		s.Video.State, s.Video.Frames, s.Video.LastError = n.pub.State(), n.pub.Frames(), n.pub.LastError()
ddiff --git a/internal/node/worker_test.go b/internal/node/worker_test.go
iindex 3f463b7..86a18e1 100644
---- a/internal/node/worker_test.go
++++ b/internal/node/worker_test.go
@@@ -18,11 +18,14 @@ import (
  	"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/localctl"
  	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/worker"
  )
  
  // 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.
++// appends a line to the marker file named in its args, prints, holds for hold_ms (1.5 s by default) 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" {
@@@ -30,14 +33,18 @@ func TestWorkerHelperProcess(t *testing.T) {
  	}
  	var args struct {
  		Marker string `json:"marker"`
++		HoldMS int64  `json:"hold_ms"`
  	}
  	_ = json.NewDecoder(os.Stdin).Decode(&args)
++	if args.HoldMS == 0 {
++		args.HoldMS = 1500
++	}
  	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)
++	time.Sleep(time.Duration(args.HoldMS) * time.Millisecond)
  	fmt.Fprint(os.NewFile(3, "result"), `{"ok":true}`)
  	os.Exit(0)
  }
@@@ -193,3 +200,89 @@ func TestJobRunsOverTheLink(t *testing.T) {
  		t.Fatalf("the function ran %q (%v), want once", b, err)
  	}
  }
++
++// statusJobs is the jobs field of `openvibe-node status` (over the local control socket).
++func (e *env) statusJobs() []worker.JobInfo {
++	e.t.Helper()
++	r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStatus}, 2*time.Second)
++	if err != nil || !r.OK {
++		e.t.Fatalf("status: %+v %v", r, err)
++	}
++	var s struct {
++		Jobs []worker.JobInfo `json:"jobs"`
++	}
++	if err := json.Unmarshal(r.Data, &s); err != nil || s.Jobs == nil {
++		e.t.Fatalf("status %s: %v", r.Data, err)
++	}
++	return s.Jobs
++}
++
++// TestLocalStopEndsJobs: `openvibe-node status` lists a running job; `openvibe-node stop` ends it stopped and new jobs
++// are refused with local_stop; after `resume` jobs run again, and status lists none once they ended.
++func TestLocalStopEndsJobs(t *testing.T) {
++	e := startWorkerNode(t)
++	marker := filepath.Join(e.paths.StateDir, "starts")
++	hello := func(n, holdMS int) protocol.Job {
++		j := testJob(jobID(n))
++		j.Artifact = &protocol.Artifact{Name: "hello", Version: "1.0.0"}
++		j.Args = json.RawMessage(fmt.Sprintf(`{"marker":%q,"hold_ms":%d}`, marker, holdMS))
++		j.Limits.MemBytes = 4 << 30
++		return j
++	}
++	exitOf := func(id string) protocol.JobExit {
++		t.Helper()
++		f, err := e.conn.Expect(protocol.TypeJobExit, 15*time.Second, func(m protocol.Message) bool {
++			return m.(protocol.JobExit).ID == id
++		})
++		if err != nil {
++			t.Fatal(err)
++		}
++		return f.Msg.(protocol.JobExit)
++	}
++	started := func(id string) {
++		t.Helper()
++		if _, err := e.conn.Expect(protocol.TypeJobStarted, 10*time.Second, func(m protocol.Message) bool {
++			return m.(protocol.JobStarted).ID == id
++		}); err != nil {
++			t.Fatal(err)
++		}
++	}
++
++	if jobs := e.statusJobs(); len(jobs) != 0 {
++		t.Fatalf("jobs before any: %+v", jobs)
++	}
++	a := hello(1, 60000)
++	if r := e.jobReply(a); r != (protocol.Ack{ID: a.ID}) {
++		t.Fatalf("%+v", r)
++	}
++	started(a.ID)
++	jobs := e.statusJobs()
++	if len(jobs) != 1 || jobs[0].ID != a.ID || jobs[0].Function != "[email protected]" || jobs[0].State != "running" || jobs[0].StartedMS == 0 {
++		t.Fatalf("status jobs while running: %+v", jobs)
++	}
++
++	e.ctl(localctl.CmdStop)
++	if ex := exitOf(a.ID); ex.Reason != protocol.ExitStopped || ex.Code != nil {
++		t.Fatalf("%+v", ex)
++	}
++	if nk := e.jobNack(hello(2, 0), protocol.JobStopped); nk.FaultCode != protocol.FaultLocalStop {
++		t.Fatalf("while stopped: %+v", nk)
++	}
++	if jobs := e.statusJobs(); len(jobs) != 0 {
++		t.Fatalf("status jobs after the stop: %+v", jobs)
++	}
++
++	e.ctl(localctl.CmdResume)
++	e.expectEstopState(false, false)
++	b := hello(3, 0)
++	if r := e.jobReply(b); r != (protocol.Ack{ID: b.ID}) {
++		t.Fatalf("after resume: %+v", r)
++	}
++	started(b.ID)
++	if ex := exitOf(b.ID); ex.Reason != protocol.ExitExited {
++		t.Fatalf("%+v", ex)
++	}
++	if jobs := e.statusJobs(); len(jobs) != 0 {
++		t.Fatalf("status jobs after the end: %+v", jobs)
++	}
++}
ddiff --git a/internal/worker/run_linux_test.go b/internal/worker/run_linux_test.go
iindex 80e5864..8890bf4 100644
---- a/internal/worker/run_linux_test.go
++++ b/internal/worker/run_linux_test.go
@@@ -16,6 +16,7 @@ import (
  	"strings"
  	"testing"
  	"time"
++	"unicode/utf8"
  
  	"github.com/OpenVibers/OpenVibe.Node/internal/config"
  	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
@@@ -35,7 +36,7 @@ func TestHelperProcess(t *testing.T) {
  		fmt.Println("line 1")
  		time.Sleep(300 * time.Millisecond)
  		fmt.Println("line 2 é")
--		time.Sleep(1900 * time.Millisecond)
++		time.Sleep(2900 * time.Millisecond) // well past the second usage tick, so both are sent even under -race
  		fmt.Fprintf(result, `{"got":%s}`, bytes.TrimSpace(in))
  	case "sleep":
  		fmt.Println("sleeping")
@@@ -45,6 +46,11 @@ func TestHelperProcess(t *testing.T) {
  		for {
  			os.Stdout.WriteString(line)
  		}
++	case "badflood": // invalid UTF-8: every lone 0xff is sent as U+FFFD, 3 bytes
++		line := strings.Repeat("\xffx", 511) + "\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.
@@@ -249,29 +255,32 @@ func TestCancelKill(t *testing.T) {
  	}
  }
  
--// 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.
++// TestOutputCap: stdout past max_output_bytes is cut at the cap and the job is killed (limit); chunks are numbered
++// from 1, valid UTF-8 and never larger than chunkBytes. Both bounds count the bytes sent, after invalid UTF-8 became
++// U+FFFD: ASCII is cut at exactly the cap, a U+FFFD is never split, so it may stop up to 2 bytes short.
  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))
++	w, s := newRunWorker(t, config.WorkerCaps{MaxOutputBytes: capBytes}, "flood", "badflood")
++	for i, mode := range []string{"flood", "badflood"} {
++		j := testJob(i+1, mode)
++		run(t, w, j)
++		ex := s.exit(t, j.ID)
++		if ex.Reason != protocol.ExitLimit || ex.Code != nil {
++			t.Fatalf("%s: %+v", mode, ex)
++		}
++		total, seq := 0, uint64(0)
++		for _, m := range s.all() {
++			if f, ok := m.(protocol.JobStdout); ok && f.ID == j.ID {
++				if f.ChunkSeq != seq+1 || len(f.Chunk) > chunkBytes || !utf8.ValidString(f.Chunk) {
++					t.Fatalf("%s: chunk %d after %d, %d bytes, valid %v", mode, f.ChunkSeq, seq, len(f.Chunk), utf8.ValidString(f.Chunk))
++				}
++				seq = f.ChunkSeq
++				total += len(f.Chunk)
  			}
--			seq = f.ChunkSeq
--			total += len(f.Chunk)
  		}
--	}
--	if total != capBytes {
--		t.Fatalf("%d stdout bytes sent, cap %d", total, capBytes)
++		if total > capBytes || total <= capBytes-utf8.UTFMax || (mode == "flood" && total != capBytes) {
++			t.Fatalf("%s: %d stdout bytes sent, cap %d", mode, total, capBytes)
++		}
  	}
  }
  
ddiff --git a/internal/worker/worker.go b/internal/worker/worker.go
iindex 412bcd3..8d05c4d 100644
---- a/internal/worker/worker.go
++++ b/internal/worker/worker.go
@@@ -9,6 +9,7 @@ package worker
  import (
  	"bytes"
  	"encoding/json"
++	"fmt"
  	"io"
  	"log/slog"
  	"os"
@@@ -57,6 +58,7 @@ type job struct {
  
  	reason    string      // why it was killed (the first reason wins); "" while nothing killed it
  	launched  bool        // Launch was called
++	started   bool        // job_started was sent (its CPU limit is set)
  	proc      *os.Process // set while the process runs
  	startedMS int64
  	exit      *protocol.JobExit // set once it ended
@@@ -197,13 +199,43 @@ func (w *Worker) SetStopped(fault string) {
  	}
  }
  
++// JobInfo is a job that has not ended, as `openvibe-node status` lists it.
++type JobInfo struct {
++	ID        string `json:"id"`
++	Function  string `json:"function"` // name@version
++	State     string `json:"state"`    // admitted (not started yet), running, or stopping (killed, not ended yet)
++	StartedMS int64  `json:"started_ms,omitempty"`
++}
++
++// Jobs lists the jobs admitted or running, by id; ended jobs are not listed. Never nil.
++func (w *Worker) Jobs() []JobInfo {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	out := []JobInfo{}
++	for _, jb := range w.jobs {
++		if jb.exit != nil {
++			continue
++		}
++		i := JobInfo{ID: jb.req.ID, Function: jb.fn.Name + "@" + jb.fn.Version, State: "admitted"}
++		if jb.started {
++			i.State, i.StartedMS = "running", jb.startedMS
++		}
++		if jb.reason != "" {
++			i.State = "stopping"
++		}
++		out = append(out, i)
++	}
++	sort.Slice(out, func(a, b int) bool { return out[a].ID < out[b].ID })
++	return out
++}
++
  // 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 {
++		if jb.exit == nil && jb.started && jb.proc != nil {
  			msgs = append(msgs, protocol.JobStarted{ID: jb.req.ID, StartedMS: jb.startedMS})
  		}
  	}
@@@ -216,14 +248,24 @@ func (w *Worker) Resend() {
  	}
  }
  
--// Close kills every job (job_exit says stopped), refuses new ones and waits up to 5 s for them to end.
++// Close kills every job (job_exit says stopped), refuses new ones and waits up to 5 s for them to end. A job admitted
++// but not launched yet never will be: its job_exit (stopped) is sent here.
  func (w *Worker) Close() {
  	w.mu.Lock()
  	w.closed = true
++	var never []protocol.JobExit
  	for _, jb := range w.jobs {
  		w.killLocked(jb, protocol.ExitStopped)
++		if jb.exit == nil && !jb.launched {
++			ex := notRun(jb.req.ID, protocol.ExitStopped)
++			w.endLocked(jb, ex)
++			never = append(never, ex)
++		}
  	}
  	w.mu.Unlock()
++	for _, ex := range never {
++		w.send(ex)
++	}
  	done := make(chan struct{})
  	go func() { w.wg.Wait(); close(done) }()
  	select {
@@@ -256,15 +298,25 @@ func (w *Worker) run(jb *job) {
  	defer w.wg.Done()
  	ex := w.execute(jb)
  	w.mu.Lock()
++	w.endLocked(jb, ex)
++	w.mu.Unlock()
++	w.log.Info("job ended", "id", ex.ID, "reason", ex.Reason, "wall_ms", ex.Usage.WallMS)
++	w.send(ex)
++}
++
++// endLocked records jb's job_exit, kept until job_exit_ack; past endedMax the oldest ended jobs are forgotten.
++func (w *Worker) endLocked(jb *job, ex protocol.JobExit) {
  	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)
++}
++
++// notRun is the job_exit of a job whose process never ran: no code, no result, no usage.
++func notRun(id, reason string) protocol.JobExit {
++	return protocol.JobExit{ID: id, Reason: reason, Result: json.RawMessage("null")}
  }
  
  // execute runs jb's process to its end and returns its job_exit.
@@@ -274,7 +326,7 @@ func (w *Worker) execute(jb *job) 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")}
++		return notRun(id, reason)
  	}
  	w.mu.Lock()
  	reason := jb.reason
@@@ -334,11 +386,19 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
  		return never(protocol.ExitFailed, err)
  	}
  	pid := cmd.Process.Pid
--	w.send(protocol.JobStarted{ID: id, StartedMS: jb.startedMS})
++	// job_started only once RLIMIT_CPU is set; a job whose limit cannot be set is killed and ends failed without it.
  	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)
++		w.mu.Lock()
++		killGroup(cmd.Process)
++		jb.proc = nil
++		w.mu.Unlock()
++		_ = cmd.Wait()
++		return never(protocol.ExitFailed, fmt.Errorf("its CPU limit could not be set: %w", err))
  	}
++	w.mu.Lock()
++	jb.started = true
++	w.mu.Unlock()
++	w.send(protocol.JobStarted{ID: id, StartedMS: jb.startedMS})
  
  	stop, watched := make(chan struct{}), make(chan accounting, 1)
  	go func() { watched <- w.watch(jb, pid, start, stop) }()
@@@ -431,41 +491,47 @@ func (w *Worker) watch(jb *job, pid int, start time.Time, stop <-chan struct{})
  	}
  }
  
--// 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.
++// pump sends the job's stdout as job_stdout chunks, 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. Invalid bytes become U+FFFD first, and both
++// bounds count the bytes sent: each chunk is at most chunkBytes and all of them together at most max_output_bytes.
  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
++	buf := make([]byte, chunkBytes)
  	var carry []byte
--	var total int64
++	left := w.caps.MaxOutputBytes
  	var seq uint64
  	over := false
++	// emit sends b's complete runes, as valid UTF-8, in chunks within both bounds; it sets over at the output cap.
  	emit := func(b []byte) {
--		seq++
--		w.send(protocol.JobStdout{ID: jb.req.ID, ChunkSeq: seq, Chunk: strings.ToValidUTF8(string(b), "�")})
++		s := strings.ToValidUTF8(string(b), "�")
++		for len(s) > 0 && !over {
++			n := min(len(s), chunkBytes)
++			if int64(n) > left {
++				n, over = int(left), true
++			}
++			for n < len(s) && n > 0 && !utf8.RuneStart(s[n]) {
++				n--
++			}
++			if n > 0 {
++				seq++
++				w.send(protocol.JobStdout{ID: jb.req.ID, ChunkSeq: seq, Chunk: s[:n]})
++				left -= int64(n)
++			}
++			s = s[n:]
++		}
++		if over {
++			w.kill(jb, protocol.ExitLimit)
++		}
  	}
  	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...)
++			b := append(carry, buf[:n]...)
  			cut := completeRunes(b)
--			if over {
--				cut = len(b)
--			}
--			if cut > 0 {
--				emit(b[:cut])
--			}
++			emit(b[:cut])
  			carry = append([]byte(nil), b[cut:]...)
--			if over {
--				w.kill(jb, protocol.ExitLimit)
--			}
  		}
  		if err != nil {
--			if len(carry) > 0 {
++			if len(carry) > 0 && !over {
  				emit(carry)
  			}
  			return
ddiff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go
iindex c93ed0a..31c8112 100644
---- a/internal/worker/worker_test.go
++++ b/internal/worker/worker_test.go
@@@ -185,3 +185,37 @@ func TestIsolationUnavailableRefused(t *testing.T) {
  		t.Fatalf("job_started sent %d times for a job that could not be isolated", n)
  	}
  }
++
++// TestCloseEndsUnlaunched: a job admitted but not launched when the worker closes gets its job_exit (stopped, never
++// run), is answered with it when resent, and a later Launch starts nothing.
++func TestCloseEndsUnlaunched(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("not in this test") }
++	j := testJob(1, "sleep")
++	if f, r := w.Admit(j); f != "" {
++		t.Fatalf("%s %s", f, r)
++	}
++	if jobs := w.Jobs(); len(jobs) != 1 || jobs[0] != (JobInfo{ID: j.ID, Function: "[email protected]", State: "admitted"}) {
++		t.Fatalf("jobs %+v", jobs)
++	}
++	w.Close()
++	ex := s.exit(t, j.ID)
++	if ex.Reason != protocol.ExitStopped || ex.Code != nil || ex.Usage.StartedMS != nil || string(ex.Result) != "null" {
++		t.Fatalf("%+v", ex)
++	}
++	if m, ok := w.Known(j.ID); !ok || m.(protocol.JobExit).Reason != protocol.ExitStopped {
++		t.Fatalf("a resent job is answered %+v, want its job_exit", m)
++	}
++	if jobs := w.Jobs(); len(jobs) != 0 {
++		t.Fatalf("an ended job listed: %+v", jobs)
++	}
++	w.Launch(j.ID)
++	time.Sleep(100 * time.Millisecond)
++	if n, e := s.count(protocol.TypeJobStarted, j.ID), s.count(protocol.TypeJobExit, j.ID); n != 0 || e != 1 {
++		t.Fatalf("%d job_started and %d job_exit after Close", n, e)
++	}
++	if f, _ := w.Admit(testJob(2, "sleep")); f != protocol.FaultShuttingDown {
++		t.Fatalf("admitted after Close: %q", f)
++	}
++}