Fetch digest-pinned Media inputs and upload large job results

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex aa797f3..cc155ee 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -237,13 +237,13 @@ Otherwise it advertises no class and refuses every well-formed job with `class n
  
  | direction        | type           | fields                                                                                    |
  |------------------|----------------|-------------------------------------------------------------------------------------------|
--| server → device  | `job`          | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, `none`, `public`, `openvibe-only`; absent means `deny`, which the Node never widens to its own `worker.egress`; `public` and `openvibe-only` run under the stricter of what is asked and what `worker.egress` allows, and are refused when it is `none`), `inputs` (files pinned by Media id and sha256: **refused**, `job inputs not supported`, until the Node can stage and verify them)} |
++| server → device  | `job`          | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, `none`, `public`, `openvibe-only`; absent means `deny`, which the Node never widens to its own `worker.egress`; `public` and `openvibe-only` run under the stricter of what is asked and what `worker.egress` allows, and are refused when it is `none`), `inputs` (at most 32 files, each `name`, `media_id`, `sha256`, optional `size_bytes`: fetched and checked against the digest before the job starts when `worker.media` is on, the job ending `failed` with nothing run otherwise; **refused**, `job inputs not supported`, when it is off)} |
  | server → device  | `job_cancel`   | `id`                                                                                      |
  | server → device  | `job_exit_ack` | `id`                                                                                      |
  | device → server  | `job_started`  | `id`, `started_ms`                                                                        |
  | device → server  | `job_stdout`   | `id`, `chunk_seq` (the job's own counter from 1; the envelope owns `seq`), `chunk`        |
  | device → server  | `job_usage`    | `id`, `started_ms`, `second`, `cpu_ms`                                                    |
--| device → server  | `job_exit`     | `id`, `reason` (`exited`, `cancelled`, `ttl`, `limit`, `stopped`, `failed`), `code`, `result`, `usage` {`started_ms`, `wall_ms`, `cpu_ms`, `mem_peak_bytes`} |
++| device → server  | `job_exit`     | `id`, `reason` (`exited`, `cancelled`, `ttl`, `limit`, `stopped`, `failed`), `code`, `result` (a result over 256 KiB uploaded to Media is `{"media_id": …}`), `usage` {`started_ms`, `wall_ms`, `cpu_ms`, `mem_peak_bytes`} |
  
  A `job` is answered `ack` (accepted) or `nack` keyed by the job id. A nack's `message` names the first problem found,
  checked in this order:
ddiff --git a/docs/worker.md b/docs/worker.md
iindex daa2c3d..0b68eb9 100644
---- a/docs/worker.md
++++ b/docs/worker.md
@@@ -20,6 +20,13 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
    "run_as": {"uid": 2001, "gid": 2001},
    "allow_same_user": false,
    "egress": "none",
++  "media": {
++    "enabled": false,
++    "endpoint": "",
++    "app": "",
++    "token_env": "OPENVIBE_MEDIA_TOKEN",
++    "max_bytes": 67108864
++  },
    "caps": {
      "max_ttl_ms": 600000,
      "max_wall_ms": 300000,
@@@ -104,6 +111,14 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
    them. The refused ranges above still win over an entry that overlaps them.
  - `egress_deny` (with `public` or `openvibe-only`): more IPv4 CIDRs to refuse, on top of those always refused: the
    Node's own cell or WireGuard peers when they hold public addresses.
++- `media` (off by default): the OpenVibe.Media objects API (`/api/v2/<app>/objects`) a job's `inputs` are fetched
++  from and a result over 256 KiB is uploaded to. `enabled` needs `endpoint` (https only, no credentials, query or
++  fragment) and `app`. Which app and token a Node uses is deployment configuration, chosen by the operator who enables
++  it: OpenVibe.Run's own app (a token holding `media.object.read` and `media.object.upload`), never Node's. The token
++  is read from the environment variable `token_env` names (default `OPENVIBE_MEDIA_TOKEN`) when a job needs it: it is
++  never written in `config.json` (a `token` key fails the load), and it appears in no job's environment, no log line
++  and no error; it is sent only to `endpoint`'s own host. `max_bytes` caps each input and each uploaded result.
++  Off, a job naming `inputs` is refused and a result over 256 KiB ends the job `limit`, as before.
  
  The classes `function` and `code` — advertised as the reserved Fabric capabilities `worker:function` and `worker:code` in
  `status.capabilities.worker` (the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — are each offered only when `enabled` is true, the OS is Linux
@@@ -132,10 +147,20 @@ AppArmor; a Node running as root with `run_as` is not affected.
  
  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. A job carrying `inputs` is refused (`job inputs not supported`) until
--   the Node can place them in its working directory and check their size and sha256: it never runs one with the
--   requested files and the digest check silently dropped. So is a job naming a `net` this host cannot enforce.
--2. **Start**: the Node makes the job's cgroup and starts itself again (`/proc/self/exe`) as the **sandbox init**,
++   it ended) and never starts a second process. A job carrying `inputs` is refused (`job inputs not supported`) unless
++   `media` is enabled: it never runs one with the requested files and the digest check silently dropped. So is a job
++   naming a `net` this host cannot enforce.
++2. **Inputs** (with `media` on, before anything of the job runs): each input is fetched by the Node from
++   `GET <endpoint>/api/v2/<app>/objects/<media_id>/download?redirect=1` into the job's directory under its `name`
++   (the contract's file-name pattern, unique, checked before the `ack`, so it cannot leave that directory). Every
++   redirect hop must stay https (at most 3; the token is dropped on a hop to another host), the body is cut off at
++   `max_bytes` (and at what is left of `max_disk_bytes` for all inputs together) while it streams, and the bytes that
++   arrive, from whatever host served them, must match the pinned `sha256` (and `size_bytes` when given). The requests
++   are bounded by the job's `ttl_ms` from receipt and cancelled when it is killed. Any failure (Media unreachable, a
++   status other than 2xx, a redirect off https, too many bytes, another digest, `token_env` unset) ends the job
++   `failed`, with no `job_started`: no sandbox is made and nothing runs. Under `egress` `openvibe-only` an unreachable
++   Media therefore fails closed. The Node fetches, not the job: inputs give a job no network of its own.
++3. **Start**: the Node makes the job's cgroup and starts itself again (`/proc/self/exe`) as the **sandbox init**,
     with its name as its only argument and its spec (the function's command and environment included) in a memfd, so
     that no host user reads the environment in its `/proc/<pid>/cmdline` (the Node checks that it shows the name
     alone), straight into that cgroup (`clone3` `CLONE_INTO_CGROUP`), in its own process group and in new user, mount,
@@@ -144,18 +169,23 @@ AppArmor; a Node running as root with `run_as` is not affected.
     `RLIMIT_CORE` 0, `RLIMIT_NOFILE` 1024, `RLIMIT_NPROC`, `RLIMIT_FSIZE` and `RLIMIT_AS`; drops every capability; sets
     `no_new_privs` and the seccomp filter; reports that it is ready and waits for the Node's go; and only then
     executes the function, with only fds 0-3 open. The function
--   works in a fresh directory `/tmp/openvibe-job-*` of its own `/tmp`; its environment is built from scratch
++   works in a fresh directory `/tmp/openvibe-job-*` of its own `/tmp`, where the init copied its inputs (regular files
++   only, opened without following a symlink) before the private root covered the Node's copies: they count towards
++   `max_disk_bytes` and its memory, and are removed with the job's directory when it ends; its environment is built from scratch
     (`PATH=/usr/local/bin:/usr/bin:/bin`, `HOME` and `TMPDIR` that directory, `LANG=C.UTF-8`, `OPENVIBE_JOB_ID`, and
     the function's `env`; nothing inherited from the Node); `args` come as one line of JSON on stdin; stderr is
     discarded. While the init waits (so it is sure to be alive), the Node checks `NoNewPrivs`, the seccomp filter and
     that nothing but `/tmp`, `/proc` and the device nodes is mounted writable, and sets `RLIMIT_CPU` again from
     outside; only then does it let the init execute the function. `job_started` follows once that `execve`
     succeeded: otherwise 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
++4. **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
++5. **End**: the function's result is what it writes to **fd 3** (JSON, at most 256 KiB), reported only when it exits
++   0. With `media` on a result may be up to `max_bytes`: one over 256 KiB is uploaded as a private Media object
++   (`POST …/objects`, `PUT …/objects/<id>/content`, `POST …/objects/<id>/complete`, each to `endpoint`, never
++   redirected, the stored sha256 checked) and `job_exit.result` is `{"media_id": "med_…"}`; the bytes never ride the
++   control link. An upload that fails ends the job `failed` (its `code` kept). `job_exit` carries the reason, the exit code (null when it was killed) and the usage, and is resent until
     `job_exit_ack`. The job's cgroup is then killed (`cgroup.kill`) and removed.
  
  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
@@@ -263,7 +293,7 @@ Refused with `nack` (keyed by the job id; the first matching row wins, after the
  | `message`                             | `fault_code`               | when                                                    |
  |---------------------------------------|----------------------------|---------------------------------------------------------|
  | `class not available`                 | `unsupported`              | the worker is off (disabled, not Linux, probe failed), or no declared entry has the job's class (`function` or `code`) |
--| `job inputs not supported`            | `unsupported`              | the job carries `inputs`: this Node cannot stage them and check their digests yet, and never runs a job with them silently dropped |
++| `job inputs not supported`            | `unsupported`              | the job carries `inputs` and `media` is off: this Node never runs a job with them silently dropped |
  | `net policy not supported`            | `unsupported`              | the job names a `net` this host cannot enforce (`public` or `openvibe-only` while `egress` is `none`) |
  | `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                                    |
@@@ -277,6 +307,6 @@ Accepted jobs end with `job_exit.reason`:
  | `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` (sampled, or the cgroup's OOM kill) exceeded, stdout past `max_output_bytes`, or a result over 256 KiB |
++| `limit`     | `wall_ms`, `cpu_ms` (sampled, or `RLIMIT_CPU`'s SIGXCPU) or `mem_bytes` (sampled, or the cgroup's OOM kill) exceeded, stdout past `max_output_bytes`, or a result over 256 KiB (over `media.max_bytes` with `media` on) |
  | `stopped`   | the e-stop or local stop latched, or the Node shut down                                                    |
--| `failed`    | the job could not run isolated: sandbox, cgroup or namespaces not set up, spawn error, working directory not made, CPU limit not set; never run with less isolation |
++| `failed`    | the job could not run isolated: sandbox, cgroup or namespaces not set up, spawn error, working directory not made, CPU limit not set; never run with less isolation. Also: an input not fetched or not matching its digest (nothing ran), or a large result not uploaded |
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex cae6c55..f3904a6 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -16,6 +16,7 @@ import (
  	"fmt"
  	"io/fs"
  	"net/netip"
++	"net/url"
  	"os"
  	"path/filepath"
  	"regexp"
@@@ -138,11 +139,65 @@ type WorkerConfig struct {
  	// EgressDeny is more IPv4 CIDRs a public or openvibe-only job is refused (the Node's cell, WireGuard peers with
  	// public addresses), on top of the ranges always refused.
  	EgressDeny []string `json:"egress_deny,omitempty"`
++	// Media is where a job's inputs come from and a result over 256 KiB is uploaded to. Off (the default), a job
++	// naming inputs is refused and a result over 256 KiB ends the job limit.
++	Media MediaConfig `json:"media,omitzero"`
  	// NodeDirs are the Node's own directories (config, credential, state, control socket), which no job may see; the
  	// Node sets them, the config file does not.
  	NodeDirs []string `json:"-"`
  }
  
++// MediaConfig is the OpenVibe.Media objects client the worker fetches inputs and uploads large results with. The
++// token is read from the environment variable TokenEnv names when a job needs it: it never lies in the config file
++// and appears in no job's environment, log line or error. Which app and token a Node uses is the operator's choice
++// (Run's own app, never Node's), made when they enable it.
++type MediaConfig struct {
++	Enabled  bool   `json:"enabled,omitempty"`
++	Endpoint string `json:"endpoint,omitempty"`  // https only, e.g. https://media.openvibe.network
++	App      string `json:"app,omitempty"`       // the :app of /api/v2/:app/objects
++	TokenEnv string `json:"token_env,omitempty"` // default DefaultMediaTokenEnv
++	MaxBytes int64  `json:"max_bytes,omitempty"` // the most bytes one input or uploaded result may take; default 64 MiB
++}
++
++// DefaultMediaTokenEnv is the variable worker.media's token is read from when token_env is not set.
++const DefaultMediaTokenEnv = "OPENVIBE_MEDIA_TOKEN"
++
++// WithDefaults fills token_env and max_bytes.
++func (m MediaConfig) WithDefaults() MediaConfig {
++	if m.TokenEnv == "" {
++		m.TokenEnv = DefaultMediaTokenEnv
++	}
++	if m.MaxBytes == 0 {
++		m.MaxBytes = 64 << 20
++	}
++	return m
++}
++
++var (
++	mediaAppRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$`)
++	envNameRe  = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]{0,127}$`)
++)
++
++func (m MediaConfig) validate() error {
++	if m.Endpoint != "" {
++		u, err := url.Parse(m.Endpoint)
++		if err != nil || u.Scheme != "https" || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" {
++			return errors.New("config: worker.media.endpoint must be an https:// URL with no credentials, query or fragment")
++		}
++	}
++	switch {
++	case m.App != "" && !mediaAppRe.MatchString(m.App):
++		return fmt.Errorf("config: worker.media.app %q is not an app id", m.App)
++	case m.TokenEnv != "" && !envNameRe.MatchString(m.TokenEnv):
++		return fmt.Errorf("config: worker.media.token_env %q is not an environment variable name", m.TokenEnv)
++	case m.MaxBytes < 0:
++		return errors.New("config: worker.media.max_bytes must not be negative")
++	case m.Enabled && (m.Endpoint == "" || m.App == ""):
++		return errors.New("config: worker.media.enabled needs worker.media.endpoint and worker.media.app")
++	}
++	return nil
++}
++
  // The worker.egress policies.
  const (
  	EgressNone         = "none"
@@@ -425,7 +480,7 @@ func (w WorkerConfig) validate() error {
  		c.MaxCPUMillis < 0 || c.MaxPids < 0 || c.MaxDiskBytes < 0 || c.MaxIOBps < 0 || c.MaxIOPS < 0 || c.MaxVMBytes < 0 {
  		return errors.New("config: worker.caps must not be negative")
  	}
--	return nil
++	return w.Media.validate()
  }
  
  // PluginCommand resolves a plugin's argv.
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
iindex 838333c..6ed96a5 100644
---- a/internal/config/config_test.go
++++ b/internal/config/config_test.go
@@@ -148,3 +148,47 @@ func TestWorkerEgress(t *testing.T) {
  		}
  	}
  }
++
++// TestWorkerMedia: worker.media is off unless enabled; enabled it needs an https endpoint (no credentials in it) and an
++// app, and its token is named, never written: a key the block does not know fails the load.
++func TestWorkerMedia(t *testing.T) {
++	ok := []MediaConfig{{}, {Endpoint: "https://media.openvibe.network"},
++		{Enabled: true, Endpoint: "https://media.openvibe.network", App: "prj_01JAB2C3D4E5F6G7H8J9K0MNPQ", TokenEnv: "RUN_MEDIA_TOKEN", MaxBytes: 1 << 20}}
++	for _, m := range ok {
++		c := Default()
++		c.Worker.Media = m
++		if err := c.Validate(); err != nil {
++			t.Fatalf("%+v: %v", m, err)
++		}
++	}
++	bad := []MediaConfig{{Enabled: true}, {Enabled: true, Endpoint: "https://media.openvibe.network"},
++		{Enabled: true, Endpoint: "http://media.openvibe.network", App: "run"},
++		{Enabled: true, Endpoint: "https://user:[email protected]", App: "run"},
++		{Enabled: true, Endpoint: "https://media.openvibe.network?token=x", App: "run"},
++		{Enabled: true, Endpoint: "https://media.openvibe.network", App: "../admin"},
++		{Enabled: true, Endpoint: "https://media.openvibe.network", App: "run", TokenEnv: "A=B"},
++		{Enabled: true, Endpoint: "https://media.openvibe.network", App: "run", MaxBytes: -1}}
++	for _, m := range bad {
++		c := Default()
++		c.Worker.Media = m
++		if err := c.Validate(); err == nil {
++			t.Fatalf("%+v validated", m)
++		}
++	}
++	if d := (MediaConfig{}).WithDefaults(); d.TokenEnv != "OPENVIBE_MEDIA_TOKEN" || d.MaxBytes != 64<<20 {
++		t.Fatalf("defaults %+v", d)
++	}
++	path := filepath.Join(t.TempDir(), "config.json")
++	if err := os.WriteFile(path, []byte(`{"worker":{"media":{"enabled":true,"endpoint":"https://media.openvibe.network","app":"run"}}}`), 0o600); err != nil {
++		t.Fatal(err)
++	}
++	if c, err := Load(path); err != nil || !c.Worker.Media.Enabled || c.Worker.Media.App != "run" {
++		t.Fatalf("%+v %v", c, err)
++	}
++	if err := os.WriteFile(path, []byte(`{"worker":{"media":{"enabled":true,"endpoint":"https://media.openvibe.network","app":"run","token":"x"}}}`), 0o600); err != nil {
++		t.Fatal(err)
++	}
++	if _, err := Load(path); err == nil {
++		t.Fatal("a token written in the config file loaded")
++	}
++}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 6ea7506..73b60b2 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -351,6 +351,9 @@ var (
  // ValidJobID reports whether id is a job_<ULID>.
  func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }
  
++// ValidMediaID reports whether id is a Media object id, med_<ULID>.
++func ValidMediaID(id string) bool { return mediaIDRe.MatchString(id) }
++
  // ValidJobInputs reports whether the job's inputs are well formed: at most maxJobInputs, each a Media object named by
  // its contract's patterns, with names unique and carrying no path separator. An empty list is valid; nil is, too.
  func ValidJobInputs(in []JobInput) bool {
ddiff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go
iindex 7f6cb59..b9ef117 100644
---- a/internal/protocol/protocol_test.go
++++ b/internal/protocol/protocol_test.go
@@@ -7,6 +7,7 @@ import (
  	"os"
  	"path/filepath"
  	"reflect"
++	"strings"
  	"testing"
  )
  
@@@ -313,6 +314,12 @@ func TestJobCheck(t *testing.T) {
  		{"inputs taken", func(j *Job) { j.Inputs = goodInputs() }, []string{ClassFunction}, ""},
  		{"input without a digest", func(j *Job) { j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID}} }, []string{ClassFunction}, JobBadInputs},
  		{"input with a path", func(j *Job) { j.Inputs = []JobInput{{Name: "a/b.bin", MediaID: mediaID, SHA256: digest}} }, []string{ClassFunction}, JobBadInputs},
++		{"input escaping the job dir", func(j *Job) { j.Inputs = []JobInput{{Name: "..", MediaID: mediaID, SHA256: digest}} }, []string{ClassFunction}, JobBadInputs},
++		{"input hidden name", func(j *Job) { j.Inputs = []JobInput{{Name: ".bashrc", MediaID: mediaID, SHA256: digest}} }, []string{ClassFunction}, JobBadInputs},
++		{"input with a 63-char digest", func(j *Job) { j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest[:63]}} }, []string{ClassFunction}, JobBadInputs},
++		{"input with an upper-case digest", func(j *Job) {
++			j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: strings.ToUpper(digest)}}
++		}, []string{ClassFunction}, JobBadInputs},
  		{"input twice under one name", func(j *Job) {
  			j.Inputs = []JobInput{{Name: "a.bin", MediaID: mediaID, SHA256: digest}, {Name: "a.bin", MediaID: mediaID, SHA256: digest}}
  		}, []string{ClassFunction}, JobBadInputs},
ddiff --git a/internal/protocol/testdata/contracts/README.md b/internal/protocol/testdata/contracts/README.md
iindex d48156b..8b9ec34 100644
---- a/internal/protocol/testdata/contracts/README.md
++++ b/internal/protocol/testdata/contracts/README.md
@@@ -1,6 +1,6 @@
  # OpenVibe.Contracts job-frame fixtures
  
--Pinned to **OpenVibe.Contracts `4029434`** (PR #7, plan T14 Run): each `.json` file is a verbatim copy of
++Pinned to **OpenVibe.Contracts `93e9bb2`** (v0.92.0, unchanged since `4029434`, PR #7, plan T14 Run): each `.json` file is a verbatim copy of
  `fixtures/platform.job-frame/valid/<name>.json`, one frame of `contracts/platform/job-frame.v1.json`
  (`platform.job-frame@1`, whose `job` field is `platform.job@1`).
  
ddiff --git a/internal/worker/run_linux_test.go b/internal/worker/run_linux_test.go
iindex b1a333b..93a0fcd 100644
---- a/internal/worker/run_linux_test.go
++++ b/internal/worker/run_linux_test.go
@@@ -144,6 +144,16 @@ func TestHelperProcess(t *testing.T) {
  		_ = json.NewEncoder(result).Encode(out)
  	case "quick":
  		fmt.Fprint(result, `{"ok":true}`)
++	case "inputs": // every file of its working directory, and its environment
++		out := map[string]string{"env": strings.Join(os.Environ(), "\n")}
++		ents, _ := os.ReadDir(".")
++		for _, e := range ents {
++			b, err := os.ReadFile(e.Name())
++			out[e.Name()] = string(b) + errString(err)
++		}
++		_ = json.NewEncoder(result).Encode(out)
++	case "bigresult": // a result over resultBytes
++		fmt.Fprintf(result, `{"x":"%s"}`, strings.Repeat("y", 300<<10))
  	case "snapshot": // looks for what TestArtifactSnapshot adds to its artifact directory on the host once it started
  		dir := filepath.Dir(os.Args[0])
  		for end := time.Now().Add(2 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
ddiff --git a/internal/worker/sandbox_linux.go b/internal/worker/sandbox_linux.go
iindex ef01853..518a23e 100644
---- a/internal/worker/sandbox_linux.go
++++ b/internal/worker/sandbox_linux.go
@@@ -39,10 +39,12 @@ const (
  
  // sandboxSpec is everything the sandbox init needs, from the Node.
  type sandboxSpec struct {
--	Root  string     `json:"root"` // the host directory the private root is mounted on
--	Work  string     `json:"work"` // the working directory, inside the private root
--	Disk  int64      `json:"disk"` // the size of the job's /tmp, and the most its artifact's copy may take
--	Binds []bindSpec `json:"binds"`
++	Root string `json:"root"` // the host directory the private root is mounted on
++	Work string `json:"work"` // the working directory, inside the private root
++	Disk int64  `json:"disk"` // the size of the job's /tmp, and the most its artifact's copy may take
++	// Inputs are the files the Node fetched into Root, copied into Work before the private root covers them.
++	Inputs []string   `json:"inputs,omitempty"`
++	Binds  []bindSpec `json:"binds"`
  	// Artifact is where the copy of the artifact directory open on artifactFD goes (its host path); "" for none.
  	Artifact string     `json:"artifact,omitempty"`
  	Links    []linkSpec `json:"links"`
@@@ -120,6 +122,14 @@ func enterSandbox() error {
  		return nil
  	}
  	r := s.Root
++	// Root itself, before the private root is mounted on it: the job's inputs lie there.
++	var inputs *os.File
++	if len(s.Inputs) > 0 {
++		if inputs, err = os.Open(r); err != nil {
++			return step("opening the job's inputs", err)
++		}
++		defer inputs.Close()
++	}
  	if err := step("making mounts private", syscall.Mount("", "/", "", syscall.MS_REC|syscall.MS_PRIVATE, "")); err != nil {
  		return err
  	}
@@@ -137,6 +147,9 @@ func enterSandbox() error {
  	if err := os.MkdirAll(r+s.Work, 0o700); err != nil {
  		return err
  	}
++	if err := step("copying the job's inputs", copyInputs(inputs, r+s.Work, s.Inputs)); err != nil {
++		return err
++	}
  	for _, b := range s.Binds {
  		if err := step("bind-mounting "+b.Path, bind(r, b)); err != nil {
  			return err
@@@ -279,6 +292,40 @@ func bind(r string, b bindSpec) error {
  	return readOnlyTree(dst)
  }
  
++// copyInputs copies the regular files names in the directory from (opened before the private root covered it) into
++// work, the job's working directory in its /tmp: they count towards its size (max_disk_bytes) and its memory.
++func copyInputs(from *os.File, work string, names []string) error {
++	for _, name := range names {
++		if filepath.Base(name) != name || name == "." || name == ".." {
++			return fmt.Errorf("input name %q", name)
++		}
++		fd, err := syscall.Openat(int(from.Fd()), name, syscall.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0)
++		if err != nil {
++			return fmt.Errorf("%s: %w", name, err)
++		}
++		src := os.NewFile(uintptr(fd), name)
++		err = func() error {
++			defer src.Close()
++			if st, err := src.Stat(); err != nil || !st.Mode().IsRegular() {
++				return fmt.Errorf("%s is not a regular file (%v)", name, err)
++			}
++			dst, err := os.OpenFile(filepath.Join(work, name), os.O_WRONLY|os.O_CREATE|os.O_EXCL|syscall.O_NOFOLLOW, 0o600)
++			if err != nil {
++				return err
++			}
++			_, err = io.Copy(dst, src)
++			if cerr := dst.Close(); err == nil {
++				err = cerr
++			}
++			return err
++		}()
++		if err != nil {
++			return err
++		}
++	}
++	return nil
++}
++
  // stageArtifact copies the artifact directory open on artifactFD, the host's src, into a tmpfs of at most size bytes
  // mounted where src leads in the private root r, and makes it read-only: the job sees the tree as it was when it
  // started, never what the host adds to it later (a socket or a FIFO a host process could be reached through, say).
ddiff --git a/internal/worker/worker.go b/internal/worker/worker.go
iindex f4f5927..0256fe6 100644
---- a/internal/worker/worker.go
++++ b/internal/worker/worker.go
@@@ -10,10 +10,13 @@ package worker
  
  import (
  	"bytes"
++	"cmp"
++	"context"
  	"encoding/json"
  	"fmt"
  	"io"
  	"log/slog"
++	"net/http"
  	"os"
  	"os/exec"
  	"path/filepath"
@@@ -43,6 +46,8 @@ type Worker struct {
  
  	// isolate puts a job's command in its sandbox; tests replace it to stand for a kernel that cannot.
  	isolate isolateFunc
++	// mediaRT carries the worker.media requests; nil is http.DefaultTransport (tests trust their own server with it).
++	mediaRT http.RoundTripper
  
  	mu      sync.Mutex
  	stopped string          // the fault code while the stop latch is set; "" when clear
@@@ -61,6 +66,9 @@ type plan struct {
  	fn     config.FunctionConfig // zero for the probe
  	limits protocol.JobLimits
  	egress string // the job's network policy (egressFor); "" is none
++	// inputs are the names of the files fetched into root, which the sandbox copies into the job's working directory
++	// before covering root with the private root.
++	inputs []string
  }
  
  // egressFor is the network policy a job runs under: the stricter of the net the job names and the host's worker.egress,
@@@ -97,6 +105,7 @@ type job struct {
  	launched  bool        // Launch was called
  	started   bool        // job_started was sent (its CPU limit is set)
  	proc      *os.Process // set while the process runs
++	abort     func()      // cancels its worker.media requests; set while it fetches its inputs or uploads its result
  	startedMS int64
  	exit      *protocol.JobExit // set once it ended
  }
@@@ -130,11 +139,14 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
  	if j.Class != protocol.ClassFunction && j.Class != protocol.ClassCode {
  		return protocol.FaultUnsupported, protocol.JobNotAvailable
  	}
--	// inputs are refused until the Node can stage them and check their digests (platform.job@1): a job never runs
--	// with the files and the digest check silently dropped.
--	if len(j.Inputs) > 0 {
++	// inputs are fetched from worker.media and checked against their digests before the job starts (platform.job@1):
++	// without it a job naming inputs is refused, never run with the files and the digest check silently dropped.
++	if len(j.Inputs) > 0 && !w.cfg.Media.Enabled {
  		return protocol.FaultUnsupported, protocol.JobInputsRefused
  	}
++	if !protocol.ValidJobInputs(j.Inputs) {
++		return protocol.FaultBadValue, protocol.JobBadInputs
++	}
  	if _, ok := egressFor(w.cfg.Egress, j.Net); !ok {
  		return protocol.FaultUnsupported, protocol.JobNetUnsupported
  	}
@@@ -353,6 +365,9 @@ func (w *Worker) killLocked(jb *job, reason string) {
  	if jb.proc != nil {
  		killGroup(jb.proc)
  	}
++	if jb.abort != nil {
++		jb.abort()
++	}
  }
  
  func (w *Worker) run(jb *job) {
@@@ -400,9 +415,22 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
  		return never(protocol.ExitFailed, err)
  	}
  	defer os.RemoveAll(dir)
++	// Every input is fetched and checked before anything of the job runs: one that fails ends it failed, never started.
++	inputs, err := w.fetchInputs(jb, dir)
++	if err != nil {
++		w.mu.Lock()
++		reason := jb.reason
++		w.mu.Unlock()
++		if reason != "" {
++			return never(reason, nil)
++		}
++		return never(protocol.ExitFailed, err)
++	}
  	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)
++		for _, p := range append([]string{"."}, inputs...) { // the job's user reads its inputs to copy them
++			if err := os.Chown(filepath.Join(dir, p), int(r.UID), int(r.GID)); err != nil {
++				return never(protocol.ExitFailed, err)
++			}
  		}
  	}
  	outR, outW, err := os.Pipe()
@@@ -424,7 +452,7 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
  	cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))
  	cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}
  	egress, _ := egressFor(w.cfg.Egress, jb.req.Net) // Admit refused what the host cannot enforce
--	sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egress})
++	sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egress, inputs: inputs})
  	if err != nil {
  		outW.Close()
  		resW.Close()
@@@ -509,7 +537,12 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
  	}
  	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)
++		if ex.Result, err = w.storeResult(jb, r); err != nil {
++			w.log.Error("job result not stored", "id", id, "err", err)
++			w.mu.Lock()
++			ex.Reason, ex.Result = cmp.Or(jb.reason, protocol.ExitFailed), json.RawMessage("null") // killed meanwhile, or failed
++			w.mu.Unlock()
++		}
  	}
  	// 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)
@@@ -636,10 +669,15 @@ func completeRunes(b []byte) int {
  	return len(b)
  }
  
--// readResult reads what the function writes to fd 3, its result; more than resultBytes kills the job (limit).
++// readResult reads what the function writes to fd 3, its result; more than resultBytes kills the job (limit), or with
++// worker.media on, more than its max_bytes (storeResult uploads a result over resultBytes).
  func (w *Worker) readResult(jb *job, r io.Reader) []byte {
--	b, _ := io.ReadAll(io.LimitReader(r, resultBytes+1))
--	if len(b) > resultBytes {
++	limit := int64(resultBytes)
++	if w.cfg.Media.Enabled {
++		limit = max(limit, w.cfg.Media.WithDefaults().MaxBytes)
++	}
++	b, _ := io.ReadAll(io.LimitReader(r, limit+1))
++	if int64(len(b)) > limit {
  		w.kill(jb, protocol.ExitLimit)
  		_, _ = io.Copy(io.Discard, r)
  		return nil
@@@ -647,6 +685,72 @@ func (w *Worker) readResult(jb *job, r io.Reader) []byte {
  	return b
  }
  
++// mediaClient returns jb's worker.media client, bounded by its ttl from receipt, and the context its requests run
++// in, which killing jb cancels; done releases it.
++func (w *Worker) mediaClient(jb *job) (c *mediaClient, ctx context.Context, done func(), err error) {
++	deadline := jb.received.Add(jb.ttl)
++	if c, err = newMediaClient(w.cfg.Media, w.mediaRT, deadline); err != nil {
++		return nil, nil, nil, err
++	}
++	ctx, cancel := context.WithDeadline(context.Background(), deadline)
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	if jb.reason != "" {
++		cancel()
++		return nil, nil, nil, fmt.Errorf("killed (%s)", jb.reason)
++	}
++	jb.abort = cancel
++	return c, ctx, func() {
++		w.mu.Lock()
++		jb.abort = nil
++		w.mu.Unlock()
++		cancel()
++	}, nil
++}
++
++// fetchInputs fetches jb's inputs into dir and returns their names: each at most worker.media.max_bytes, all of them
++// together at most worker.caps.max_disk_bytes (the size of the job's /tmp they are copied to), each matching its
++// pinned sha256.
++func (w *Worker) fetchInputs(jb *job, dir string) ([]string, error) {
++	if len(jb.req.Inputs) == 0 {
++		return nil, nil
++	}
++	c, ctx, done, err := w.mediaClient(jb)
++	if err != nil {
++		return nil, err
++	}
++	defer done()
++	left := w.caps.MaxDiskBytes
++	names := make([]string, 0, len(jb.req.Inputs))
++	for _, in := range jb.req.Inputs {
++		n, err := c.fetchInput(ctx, in, dir, min(c.cfg.MaxBytes, left))
++		if err != nil {
++			return nil, err
++		}
++		left -= n
++		names = append(names, in.Name)
++	}
++	return names, nil
++}
++
++// storeResult is the job_exit result for r, a function's valid JSON result: r itself up to resultBytes, beyond that
++// {"media_id": …} of the Media object it is uploaded as (only read past resultBytes with worker.media on).
++func (w *Worker) storeResult(jb *job, r []byte) (json.RawMessage, error) {
++	if len(r) <= resultBytes {
++		return json.RawMessage(r), nil
++	}
++	c, ctx, done, err := w.mediaClient(jb)
++	if err != nil {
++		return nil, err
++	}
++	defer done()
++	id, err := c.uploadResult(ctx, jb.req.ID, r)
++	if err != nil {
++		return nil, err
++	}
++	return json.Marshal(map[string]string{"media_id": id})
++}
++
  // 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 {
ddiff --git a/internal/worker/worker_linux.go b/internal/worker/worker_linux.go
iindex 5d64b8a..62a2932 100644
---- a/internal/worker/worker_linux.go
++++ b/internal/worker/worker_linux.go
@@@ -72,7 +72,7 @@ func isolate(cmd *exec.Cmd, cfg config.WorkerConfig, p plan) (*sandbox, error) {
  	}
  	sb.goW, sb.child = gw, append(sb.child, gr)
  	files = append(files, sw, gr) // statusFD, goFD
--	spec := sandboxSpec{Root: p.root, Work: cmd.Dir, Disk: caps.MaxDiskBytes, Links: links,
++	spec := sandboxSpec{Root: p.root, Work: cmd.Dir, Disk: caps.MaxDiskBytes, Links: links, Inputs: p.inputs,
  		Argv: append([]string{cmd.Path}, cmd.Args[1:]...), Env: cmd.Env}
  	if art != nil {
  		spec.Artifact = p.fn.Artifact()
ddiff --git a/internal/worker/worker_media.go b/internal/worker/worker_media.go
nnew file mode 100644
iindex 0000000..68cac8d
---- /dev/null
++++ b/internal/worker/worker_media.go
@@@ -0,0 +1,204 @@
++package worker
++
++import (
++	"bytes"
++	"context"
++	"crypto/sha256"
++	"encoding/hex"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"io"
++	"net/http"
++	"net/url"
++	"os"
++	"path/filepath"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// maxRedirects is how many redirects a download follows (Media answers one: to the object's signed URL).
++const maxRedirects = 3
++
++// mediaClient is the OpenVibe.Media objects client of one job (worker.media; docs/worker.md). Its token is read from
++// the environment variable worker.media.token_env names and goes only in the Authorization header of a request to the
++// endpoint's own host: never in a job's environment, a log line or an error.
++type mediaClient struct {
++	cfg   config.MediaConfig
++	base  *url.URL // the endpoint
++	token string
++	http  *http.Client
++}
++
++// newMediaClient returns the client of a job that must end by deadline (its ttl from receipt): every request is
++// bounded by it. It fails when worker.media is off or the token variable is unset, so a job's inputs are never
++// skipped and its result never dropped quietly.
++func newMediaClient(cfg config.MediaConfig, rt http.RoundTripper, deadline time.Time) (*mediaClient, error) {
++	if !cfg.Enabled {
++		return nil, errors.New("worker.media is off")
++	}
++	cfg = cfg.WithDefaults()
++	token := os.Getenv(cfg.TokenEnv)
++	if token == "" {
++		return nil, fmt.Errorf("worker.media: %s is not set", cfg.TokenEnv)
++	}
++	base, err := url.Parse(cfg.Endpoint)
++	if err != nil || base.Scheme != "https" || base.Host == "" {
++		return nil, errors.New("worker.media.endpoint is not an https URL")
++	}
++	left := time.Until(deadline)
++	if left <= 0 {
++		return nil, errors.New("worker.media: the job's ttl has passed")
++	}
++	if rt == nil {
++		rt = http.DefaultTransport
++	}
++	c := &mediaClient{cfg: cfg, base: base, token: token}
++	c.http = &http.Client{Transport: rt, Timeout: left, CheckRedirect: c.redirect}
++	return c, nil
++}
++
++// redirect re-checks every hop of a download: https only, at most maxRedirects, and the token is sent to the
++// endpoint's own host only (a signed URL on a storage host carries its own authority). The bytes at the end are
++// checked against the pinned digest whatever host served them.
++func (c *mediaClient) redirect(req *http.Request, via []*http.Request) error {
++	if via[0].Method != http.MethodGet { // a redirected POST arrives here as a GET
++		return errors.New("worker.media: an upload was redirected")
++	}
++	if len(via) > maxRedirects {
++		return errors.New("worker.media: too many redirects")
++	}
++	if req.URL.Scheme != "https" {
++		return errors.New("worker.media: a redirect left https")
++	}
++	if req.URL.Host != c.base.Host {
++		req.Header.Del("Authorization")
++	} else {
++		req.Header.Set("Authorization", "Bearer "+c.token)
++	}
++	return nil
++}
++
++// objectsURL is {endpoint}/api/v2/{app}/objects followed by the path elements given, each escaped.
++func (c *mediaClient) objectsURL(elem ...string) string {
++	p := []string{"api", "v2", url.PathEscape(c.cfg.App), "objects"}
++	for _, e := range elem {
++		p = append(p, url.PathEscape(e))
++	}
++	return c.base.JoinPath(p...).String()
++}
++
++// do sends req with the token and returns the response of a 2xx answer. An error names the status or the transport's
++// cause, never a URL (a signed URL is a credential too) and never the token.
++func (c *mediaClient) do(req *http.Request) (*http.Response, error) {
++	req.Header.Set("Authorization", "Bearer "+c.token)
++	resp, err := c.http.Do(req)
++	if err != nil {
++		var ue *url.Error
++		if errors.As(err, &ue) {
++			err = ue.Err
++		}
++		return nil, err
++	}
++	if resp.StatusCode/100 != 2 {
++		resp.Body.Close()
++		return nil, fmt.Errorf("status %d", resp.StatusCode)
++	}
++	return resp, nil
++}
++
++// fetchInput downloads one digest-pinned input into dir as in.Name (protocol.ValidJobInputs made it a plain file
++// name), reading at most limit bytes while streaming: a body over limit, a size other than the one declared, or a
++// sha256 other than the pinned one returns an error, and the caller ends the job failed before its process starts.
++func (c *mediaClient) fetchInput(ctx context.Context, in protocol.JobInput, dir string, limit int64) (int64, error) {
++	fail := func(err error) (int64, error) { return 0, fmt.Errorf("input %s (%s): %w", in.Name, in.MediaID, err) }
++	if !protocol.ValidMediaID(in.MediaID) || filepath.Base(in.Name) != in.Name || in.Name == "." || in.Name == ".." {
++		return fail(errors.New("malformed"))
++	}
++	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.objectsURL(in.MediaID, "download")+"?redirect=1", nil)
++	if err != nil {
++		return fail(err)
++	}
++	resp, err := c.do(req)
++	if err != nil {
++		return fail(err)
++	}
++	defer resp.Body.Close()
++	if resp.Request.URL.Scheme != "https" { // CheckRedirect refuses it; never trust a single check on this path
++		return fail(errors.New("not served over https"))
++	}
++	if resp.ContentLength > limit {
++		return fail(fmt.Errorf("%d bytes, over the %d allowed", resp.ContentLength, limit))
++	}
++	f, err := os.OpenFile(filepath.Join(dir, in.Name), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
++	if err != nil {
++		return fail(err)
++	}
++	h := sha256.New()
++	n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(resp.Body, limit+1))
++	if cerr := f.Close(); err == nil {
++		err = cerr
++	}
++	switch {
++	case err != nil:
++		return fail(err)
++	case n > limit:
++		return fail(fmt.Errorf("over the %d bytes allowed", limit))
++	case in.SizeBytes > 0 && n != in.SizeBytes:
++		return fail(fmt.Errorf("%d bytes, not the %d declared", n, in.SizeBytes))
++	case hex.EncodeToString(h.Sum(nil)) != in.SHA256:
++		return fail(errors.New("sha256 does not match"))
++	}
++	return n, nil
++}
++
++// uploadResult stores data, a job's JSON result, as a private Media object (init, PUT content, complete, each to the
++// endpoint, the upload never redirected) and returns its id.
++func (c *mediaClient) uploadResult(ctx context.Context, jobID string, data []byte) (string, error) {
++	sum := sha256.Sum256(data)
++	digest := hex.EncodeToString(sum[:])
++	send := func(method, u string, body []byte, ctype string, out any) error {
++		req, err := http.NewRequestWithContext(ctx, method, u, bytes.NewReader(body))
++		if err != nil {
++			return err
++		}
++		req.Header.Set("Content-Type", ctype)
++		resp, err := c.do(req)
++		if err != nil {
++			return err
++		}
++		defer resp.Body.Close()
++		if out == nil {
++			return nil
++		}
++		return json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(out)
++	}
++	created, _ := json.Marshal(map[string]any{"kind": "file", "visibility": "private", "size_bytes": len(data),
++		"content_hash": digest, "mime_type": "application/json", "filename": jobID + ".json",
++		"metadata": map[string]string{"job_id": jobID}})
++	var made struct {
++		ID string `json:"id"`
++	}
++	if err := send(http.MethodPost, c.objectsURL(), created, "application/json", &made); err != nil {
++		return "", fmt.Errorf("result upload: init: %w", err)
++	}
++	if !protocol.ValidMediaID(made.ID) {
++		return "", errors.New("result upload: init answered no object id")
++	}
++	var put struct {
++		ContentHash string `json:"content_hash"`
++	}
++	if err := send(http.MethodPut, c.objectsURL(made.ID, "content"), data, "application/json", &put); err != nil {
++		return "", fmt.Errorf("result upload: content: %w", err)
++	}
++	if put.ContentHash != digest {
++		return "", errors.New("result upload: Media stored other bytes")
++	}
++	done, _ := json.Marshal(map[string]string{"content_hash": digest})
++	if err := send(http.MethodPost, c.objectsURL(made.ID, "complete"), done, "application/json", nil); err != nil {
++		return "", fmt.Errorf("result upload: complete: %w", err)
++	}
++	return made.ID, nil
++}
ddiff --git a/internal/worker/worker_media_test.go b/internal/worker/worker_media_test.go
nnew file mode 100644
iindex 0000000..856a6fe
---- /dev/null
++++ b/internal/worker/worker_media_test.go
@@@ -0,0 +1,315 @@
++//go:build linux
++
++package worker
++
++import (
++	"bytes"
++	"crypto/sha256"
++	"encoding/hex"
++	"encoding/json"
++	"errors"
++	"io"
++	"log/slog"
++	"net/http"
++	"net/http/httptest"
++	"os"
++	"os/exec"
++	"path/filepath"
++	"strings"
++	"sync"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++const (
++	testMediaToken = "ov-media-test-token-3f1c9a"
++	testMediaEnv   = "OPENVIBE_TEST_MEDIA_TOKEN"
++	inputID        = "med_01JAB2C3D4E5F6G7H8J9K0MNPQ"
++	otherID        = "med_01JAB2C3D4E5F6G7H8J9K0MNPR"
++	resultID       = "med_01JAB2C3D4E5F6G7H8J9K0MNPS"
++)
++
++func digestOf(b []byte) string {
++	s := sha256.Sum256(b)
++	return hex.EncodeToString(s[:])
++}
++
++// mediaFake is OpenVibe.Media's objects API on TLS: download answers a 302 to where (by default the object's bytes
++// on a second host, as a signed storage URL is), and uploads are kept.
++type mediaFake struct {
++	srv, store *httptest.Server
++	mu         sync.Mutex
++	objects    map[string][]byte
++	where      func(id string) string // the download redirect's target
++	uploaded   []byte
++	bounce     bool     // every API request is redirected to the storage host
++	storeAuth  []string // Authorization headers the storage host saw
++	apiAuth    []string // Authorization headers the API saw
++}
++
++func newMediaFake(t *testing.T) *mediaFake {
++	m := &mediaFake{objects: map[string][]byte{}}
++	m.store = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++		m.mu.Lock()
++		m.storeAuth = append(m.storeAuth, r.Header.Get("Authorization"))
++		b, ok := m.objects[strings.TrimPrefix(r.URL.Path, "/o/")]
++		m.mu.Unlock()
++		if !ok {
++			http.NotFound(w, r)
++			return
++		}
++		_, _ = w.Write(b)
++	}))
++	t.Cleanup(m.store.Close)
++	m.where = func(id string) string { return m.store.URL + "/o/" + id + "?exp=1&sig=s3cr3t" }
++	m.srv = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++		m.mu.Lock()
++		defer m.mu.Unlock()
++		m.apiAuth = append(m.apiAuth, r.Header.Get("Authorization"))
++		if m.bounce {
++			http.Redirect(w, r, m.store.URL+"/o/x", http.StatusFound)
++			return
++		}
++		if r.Header.Get("Authorization") != "Bearer "+testMediaToken {
++			http.Error(w, "unauthorized", http.StatusUnauthorized)
++			return
++		}
++		p := strings.TrimPrefix(r.URL.Path, "/api/v2/run/objects")
++		switch {
++		case r.Method == http.MethodGet && strings.HasSuffix(p, "/download") && r.URL.Query().Get("redirect") == "1":
++			http.Redirect(w, r, m.where(strings.TrimSuffix(strings.TrimPrefix(p, "/"), "/download")), http.StatusFound)
++		case r.Method == http.MethodPost && p == "":
++			w.WriteHeader(http.StatusCreated)
++			_, _ = w.Write([]byte(`{"id":"` + resultID + `","upload":{"method":"PUT"}}`))
++		case r.Method == http.MethodPut && p == "/"+resultID+"/content":
++			m.uploaded, _ = io.ReadAll(r.Body)
++			_ = json.NewEncoder(w).Encode(map[string]any{"id": resultID, "content_hash": digestOf(m.uploaded)})
++		case r.Method == http.MethodPost && p == "/"+resultID+"/complete":
++			_, _ = w.Write([]byte(`{"id":"` + resultID + `","lifecycle_status":"ready"}`))
++		default:
++			http.NotFound(w, r)
++		}
++	}))
++	t.Cleanup(m.srv.Close)
++	return m
++}
++
++func (m *mediaFake) config(cfg *config.WorkerConfig) {
++	cfg.Media = config.MediaConfig{Enabled: true, Endpoint: m.srv.URL, App: "run", TokenEnv: testMediaEnv, MaxBytes: 1 << 20}
++}
++
++// mediaWorker is a worker whose jobs fetch from m, with its log in log and isolate standing for the sandbox: it records
++// the plan and the job's environment, and runs nothing.
++func mediaWorker(t *testing.T, m *mediaFake) (w *Worker, s *sink, log *logBuf, isolated func() (plan, []string, bool)) {
++	t.Helper()
++	t.Setenv(testMediaEnv, testMediaToken)
++	cfg := helperConfig(t, config.WorkerCaps{}, "sleep")
++	m.config(&cfg)
++	s, log = &sink{}, &logBuf{}
++	w = New(cfg, s.send, slog.New(slog.NewTextHandler(log, nil)))
++	w.mediaRT = m.srv.Client().Transport
++	var mu sync.Mutex
++	var got plan
++	var env []string
++	called := false
++	w.isolate = func(cmd *exec.Cmd, _ config.WorkerConfig, p plan) (*sandbox, error) {
++		mu.Lock()
++		defer mu.Unlock()
++		got, env, called = p, cmd.Env, true
++		// what lies in the job's directory when its sandbox would be made, read now: execute removes it after
++		for _, n := range p.inputs {
++			b, err := os.ReadFile(filepath.Join(p.root, n))
++			got.inputs = append(got.inputs, n+"="+string(b)+"|"+errString(err))
++		}
++		return nil, errors.New("not in this test")
++	}
++	t.Cleanup(w.Close)
++	return w, s, log, func() (plan, []string, bool) {
++		mu.Lock()
++		defer mu.Unlock()
++		return got, env, called
++	}
++}
++
++func errString(err error) string {
++	if err != nil {
++		return err.Error()
++	}
++	return ""
++}
++
++func inputJob(n int, ins ...protocol.JobInput) protocol.Job {
++	j := testJob(n, "sleep")
++	j.Inputs = ins
++	return j
++}
++
++// TestInputFetched: an input whose bytes match its pinned sha256 lies in the job's directory, under its name, when the
++// sandbox is made; the token went to the API only (not to the storage host it redirected to) and is in neither the
++// job's environment nor the log.
++func TestInputFetched(t *testing.T) {
++	m := newMediaFake(t)
++	data := []byte("frame bytes\n")
++	m.objects[inputID] = data
++	w, s, log, isolated := mediaWorker(t, m)
++	j := inputJob(1, protocol.JobInput{Name: "clip.mp4", MediaID: inputID, SHA256: digestOf(data), SizeBytes: int64(len(data))})
++	run(t, w, j)
++	s.exit(t, j.ID)
++	p, env, called := isolated()
++	if !called || len(p.inputs) != 2 || p.inputs[0] != "clip.mp4" || p.inputs[1] != "clip.mp4=frame bytes\n|" {
++		t.Fatalf("called %v, inputs %q", called, p.inputs)
++	}
++	for _, e := range env {
++		if strings.Contains(e, testMediaToken) || strings.HasPrefix(e, testMediaEnv+"=") {
++			t.Fatalf("the token reached the job's environment: %q", e)
++		}
++	}
++	if strings.Contains(log.String(), testMediaToken) || strings.Contains(log.String(), "s3cr3t") {
++		t.Fatalf("the token or a signed URL was logged: %s", log)
++	}
++	m.mu.Lock()
++	defer m.mu.Unlock()
++	if len(m.storeAuth) != 1 || m.storeAuth[0] != "" {
++		t.Fatalf("the storage host saw Authorization %q", m.storeAuth)
++	}
++}
++
++// TestInputRefused: a flipped byte, a body over max_bytes, a size other than declared, no token, a redirect to http
++// or to another object's bytes each end the job failed with nothing run: no sandbox, no job_started, no job_stdout,
++// and no token in the log.
++func TestInputRefused(t *testing.T) {
++	data := []byte("frame bytes\n")
++	flipped := append([]byte(nil), data...)
++	flipped[0] ^= 1
++	good := protocol.JobInput{Name: "clip.mp4", MediaID: inputID, SHA256: digestOf(data)}
++	for _, c := range []struct {
++		name, why string // why: in the logged error
++		set       func(t *testing.T, m *mediaFake, in *protocol.JobInput)
++	}{
++		{"flipped byte", "sha256 does not match", func(t *testing.T, m *mediaFake, in *protocol.JobInput) { m.objects[inputID] = flipped }},
++		{"over max_bytes", "over the 1048576 bytes allowed", func(t *testing.T, m *mediaFake, in *protocol.JobInput) {
++			big := bytes.Repeat([]byte("x"), 1<<20+1)
++			m.objects[inputID], in.SHA256 = big, digestOf(big)
++		}},
++		{"size not as declared", "not the 13 declared", func(t *testing.T, m *mediaFake, in *protocol.JobInput) { in.SizeBytes = int64(len(data)) + 1 }},
++		{"token_env unset", "OPENVIBE_TEST_MEDIA_TOKEN is not set", func(t *testing.T, m *mediaFake, in *protocol.JobInput) { t.Setenv(testMediaEnv, "") }},
++		{"redirect to http", "a redirect left https", func(t *testing.T, m *mediaFake, in *protocol.JobInput) {
++			m.where = func(id string) string { return strings.Replace(m.store.URL, "https:", "http:", 1) + "/o/" + id }
++		}},
++		{"redirect to another digest", "sha256 does not match", func(t *testing.T, m *mediaFake, in *protocol.JobInput) {
++			m.objects[otherID] = []byte("other bytes\n")
++			m.where = func(string) string { return m.store.URL + "/o/" + otherID }
++		}},
++		{"media unreachable", "connect", func(t *testing.T, m *mediaFake, in *protocol.JobInput) { m.srv.Close() }},
++	} {
++		t.Run(c.name, func(t *testing.T) {
++			m := newMediaFake(t)
++			m.objects[inputID] = data
++			w, s, log, isolated := mediaWorker(t, m)
++			in := good
++			c.set(t, m, &in)
++			j := inputJob(1, in)
++			run(t, w, j)
++			ex := s.exit(t, j.ID)
++			if ex.Reason != protocol.ExitFailed || ex.Code != nil || ex.Usage.StartedMS != nil || string(ex.Result) != "null" {
++				t.Fatalf("%+v", ex)
++			}
++			if _, _, called := isolated(); called {
++				t.Fatal("the job's sandbox was made")
++			}
++			if n, o := s.count(protocol.TypeJobStarted, j.ID), s.count(protocol.TypeJobStdout, j.ID); n != 0 || o != 0 {
++				t.Fatalf("%d job_started, %d job_stdout", n, o)
++			}
++			if strings.Contains(log.String(), testMediaToken) || strings.Contains(log.String(), "s3cr3t") {
++				t.Fatalf("the token or a signed URL was logged: %s", log)
++			}
++			if !strings.Contains(log.String(), "job not run") || !strings.Contains(log.String(), c.why) {
++				t.Fatalf("the failure was not logged: %s", log)
++			}
++		})
++	}
++}
++
++// TestAdmitInputs: with worker.media on a job naming inputs is admitted; malformed inputs are refused even when Check
++// was skipped.
++func TestAdmitInputs(t *testing.T) {
++	m := newMediaFake(t)
++	w, _, _, _ := mediaWorker(t, m)
++	if f, r := w.Admit(inputJob(1, protocol.JobInput{Name: "a.bin", MediaID: inputID, SHA256: strings.Repeat("a", 64)})); f != "" {
++		t.Fatalf("%s %s", f, r)
++	}
++	if _, r := w.Admit(inputJob(2, protocol.JobInput{Name: "../a.bin", MediaID: inputID, SHA256: strings.Repeat("a", 64)})); r != protocol.JobBadInputs {
++		t.Fatalf("a path-escaping name: %q", r)
++	}
++}
++
++// TestResultUploaded: a result over 256 KiB is uploaded as a Media object and job_exit carries {"media_id": …}; one
++// within 256 KiB rides the link as it is; an upload Media redirects is refused.
++func TestResultUploaded(t *testing.T) {
++	m := newMediaFake(t)
++	w, _, _, _ := mediaWorker(t, m)
++	jb := &job{req: testJob(1, "sleep"), received: time.Now(), ttl: time.Minute}
++	big := []byte(`{"frames":"` + strings.Repeat("x", resultBytes) + `"}`)
++	got, err := w.storeResult(jb, big)
++	if err != nil || string(got) != `{"media_id":"`+resultID+`"}` {
++		t.Fatalf("%s %v", got, err)
++	}
++	if !bytes.Equal(m.uploaded, big) {
++		t.Fatalf("uploaded %d bytes, want the %d of the result", len(m.uploaded), len(big))
++	}
++	if got, err := w.storeResult(jb, []byte(`{"ok":true}`)); err != nil || string(got) != `{"ok":true}` {
++		t.Fatalf("%s %v", got, err)
++	}
++	if jb.abort != nil {
++		t.Fatal("the upload's context outlived it")
++	}
++	m.mu.Lock()
++	m.bounce = true
++	m.mu.Unlock()
++	if _, err := w.storeResult(jb, big); err == nil {
++		t.Fatal("a redirected upload succeeded")
++	}
++	t.Setenv(testMediaEnv, "")
++	if _, err := w.storeResult(jb, big); err == nil {
++		t.Fatal("uploaded without a token")
++	}
++	// ended past its ttl: no request is made at all
++	jb.received = time.Now().Add(-2 * time.Minute)
++	t.Setenv(testMediaEnv, testMediaToken)
++	if _, err := w.storeResult(jb, big); err == nil {
++		t.Fatal("uploaded past the job's ttl")
++	}
++}
++
++// TestRunInputsAndBigResult: in a real sandbox, a fetched input is in the job's working directory and a result over
++// 256 KiB arrives as a media_id.
++func TestRunInputsAndBigResult(t *testing.T) {
++	m := newMediaFake(t)
++	data := []byte("frame bytes\n")
++	m.objects[inputID] = data
++	t.Setenv(testMediaEnv, testMediaToken)
++	cfg := helperConfig(t, config.WorkerCaps{MaxMemBytes: 8 << 30}, "inputs", "bigresult")
++	m.config(&cfg)
++	w, s := newRunWorkerConfig(t, cfg)
++	w.mediaRT = m.srv.Client().Transport
++	j := testJob(1, "inputs")
++	j.Inputs = []protocol.JobInput{{Name: "clip.mp4", MediaID: inputID, SHA256: digestOf(data)}}
++	run(t, w, j)
++	ex := s.exit(t, j.ID)
++	var got map[string]string
++	if err := json.Unmarshal(ex.Result, &got); err != nil || ex.Reason != protocol.ExitExited || got["clip.mp4"] != string(data) {
++		t.Fatalf("%+v %s: %v", ex, ex.Result, err)
++	}
++	if strings.Contains(got["env"], testMediaToken) {
++		t.Fatal("the token reached the job's environment")
++	}
++	b := testJob(2, "bigresult")
++	run(t, w, b)
++	ex = s.exit(t, b.ID)
++	if ex.Reason != protocol.ExitExited || string(ex.Result) != `{"media_id":"`+resultID+`"}` || len(m.uploaded) <= resultBytes {
++		t.Fatalf("%+v %s (%d bytes uploaded)", ex, ex.Result, len(m.uploaded))
++	}
++}