Sandbox worker: per-session egress budget, rate limits, and the isolation facts Live registers (#302)

ddiff --git a/deploy/sandbox/sandbox-worker.env.example b/deploy/sandbox/sandbox-worker.env.example
iindex 0511f5a..3806d8a 100644
---- a/deploy/sandbox/sandbox-worker.env.example
++++ b/deploy/sandbox/sandbox-worker.env.example
@@@ -14,6 +14,13 @@ CODES_SANDBOX_MAX_SESSIONS=2
  # The longest time box a session may have (the owner sets 5 to 30 minutes).
  CODES_SANDBOX_TIME_BOX_MAX_MIN=30
  
++# Each session's egress budget in MiB, down and up (default 500 and 100). When it is spent the session's transfer is cut
++# off and every new request is refused (429, code egress.budget) until the session is wiped; the session report says so.
++# The request rates (10/s per destination host, burst 30; 50/s per session) and 8 connections per host are the proxy's
++# own limits.
++CODES_SANDBOX_EGRESS_MB_DOWN=500
++CODES_SANDBOX_EGRESS_MB_UP=100
++
  # Node for the supervisor inside the sandbox (bound read-only), and the headless Chromium install.sh put in place.
  CODES_SANDBOX_NODE=/usr/bin/node
  #CODES_SANDBOX_BROWSER=/opt/openvibe-codes-sandbox/browser/chromium_headless_shell-NNNN/chrome-headless-shell-linux64/chrome-headless-shell
ddiff --git a/sandbox/egress-proxy.js b/sandbox/egress-proxy.js
iindex ac02470..33043e5 100644
---- a/sandbox/egress-proxy.js
++++ b/sandbox/egress-proxy.js
@@@ -19,13 +19,27 @@
   * port, address, allowed or the refusal's code — never a path, a header or a body) goes to onDecision for the
   * owner's log.
   *
-- *   const proxy = createEgressProxy({ lookup, dial, hostAddresses, ports, maxConnections, onDecision });
++ * A session has a budget and limits of its own (design §1; the owner sets the bytes through CODES_SANDBOX_EGRESS_MB_*):
++ *
++ *   - bytes: 500 MB down and 100 MB up by default. Every response body and tunnel counts both ways; an open tunnel
++ *     is closed as soon as the budget is spent, and a download is cut off at the byte that spends it. New requests
++ *     after that are 429 with `code=egress.budget`, and the session's report (stats()) says it is spent;
++ *   - requests: a token bucket per destination host (10/s, burst 30) and one for the session (50/s, burst 50). A
++ *     request over either is 429 `egress.rate` without a socket;
++ *   - connections: at most 8 open at once to one destination host (a CONNECT tunnel or an HTTP request in flight).
++ *     One more is 429 `egress.connections`.
++ *
++ * The counters are per proxy, and a proxy is one sandbox session, so one session can never spend another's budget.
++ *
++ *   const proxy = createEgressProxy({ lookup, dial, hostAddresses, ports, maxConnections,
++ *       budgetDownBytes, budgetUpBytes, hostRequests, sessionRequests, maxConnectionsPerHost, now, onDecision });
   *   await proxy.listen('/run/…/net/egress.sock') or proxy.listen({ port: 0, host: '127.0.0.1' })   (tests)
   *   await proxy.check(host, port) → { ok, host, port, address, family } | { ok: false, code, reason }
++ *   proxy.stats() → what this session has spent, its limits and the refusals it made
   *   proxy.address(), await proxy.close()
   *
   * lookup (dns.lookup-shaped) and dial ({ host: address, port } → net.Socket) are injectable for the tests; the worker
-- * uses the system resolver and net.connect.
++ * uses the system resolver and net.connect. now() is the clock for the token buckets (tests).
   */
  const dns = require('dns');
  const fs = require('fs');
@@@ -47,17 +61,81 @@ function defaultHostAddresses() {
  
  const defaultDial = ({ host, port }) => net.connect({ host, port });
  
++const DEFAULT_BUDGET_DOWN = 500 * 1024 * 1024;
++const DEFAULT_BUDGET_UP = 100 * 1024 * 1024;
++const DEFAULT_HOST_REQUESTS = Object.freeze({ rate: 10, burst: 30 });
++const DEFAULT_SESSION_REQUESTS = Object.freeze({ rate: 50, burst: 50 });
++const DEFAULT_MAX_PER_HOST = 8;
++
  function createEgressProxy({
      lookup = dns.lookup, dial = defaultDial, hostAddresses = defaultHostAddresses, ports = PORTS,
      maxConnections = 64, connectTimeoutMs = 10_000, idleMs = 120_000, lookupTimeoutMs = 5_000,
++    budgetDownBytes = DEFAULT_BUDGET_DOWN, budgetUpBytes = DEFAULT_BUDGET_UP,
++    hostRequests = DEFAULT_HOST_REQUESTS, sessionRequests = DEFAULT_SESSION_REQUESTS,
++    maxConnectionsPerHost = DEFAULT_MAX_PER_HOST, now = Date.now,
      onDecision = () => {}, log = console,
  } = {}) {
      const allowedPorts = new Set(ports.map(Number));
      const sockets = new Set();
      let server = null;
++    const clock = typeof now === 'function' ? now : Date.now;
++    // A non-positive budget or rate means "no limit" for that knob, never zero (a 0-byte budget would brick the
++    // session, and an env var left empty must not do that).
++    const limitOf = (v, def) => (Number(v) > 0 ? Number(v) : Number(def) > 0 ? Number(def) : Infinity);
++    const limits = { down: limitOf(budgetDownBytes, DEFAULT_BUDGET_DOWN), up: limitOf(budgetUpBytes, DEFAULT_BUDGET_UP) };
++    const hostRate = { rate: Number(hostRequests.rate) > 0 ? Number(hostRequests.rate) : 0, burst: Math.max(1, Number(hostRequests.burst) || DEFAULT_HOST_REQUESTS.burst) };
++    const sessionRate = { rate: Number(sessionRequests.rate) > 0 ? Number(sessionRequests.rate) : 0, burst: Math.max(1, Number(sessionRequests.burst) || DEFAULT_SESSION_REQUESTS.burst) };
++    const perHost = Number(maxConnectionsPerHost) > 0 ? Number(maxConnectionsPerHost) : Infinity;
++
++    const bytes = { down: 0, up: 0 };
++    let spent = false;
++    let spentAt = null;
++    let spentDirection = null;
++    const refusals = { budget: 0, rate: 0, connections: 0 };
++    const hosts = new Map();   // host → { tokens, last, active }
++    const sessionBucket = { tokens: sessionRate.burst, last: clock() };
  
      const refuse = (host, port, code, reason) => ({ ok: false, host, port, code, reason });
  
++    /** A token bucket: burst tokens, refilled at rate per second. No rate = unlimited. */
++    function take(bucket, cfg) {
++        if (!(cfg.rate > 0)) return true;
++        const t = Number(clock()) || Date.now();
++        bucket.tokens = Math.min(cfg.burst, bucket.tokens + Math.max(0, t - bucket.last) / 1000 * cfg.rate);
++        bucket.last = t;
++        if (bucket.tokens < 1) return false;
++        bucket.tokens -= 1;
++        return true;
++    }
++
++    function hostState(host) {
++        let h = hosts.get(host);
++        if (!h) { h = { tokens: hostRate.burst, last: clock(), active: 0 }; hosts.set(host, h); }
++        return h;
++    }
++
++    /** Count bytes both ways; at the byte that spends the budget the transfer is marked and the sockets it is on close. */
++    function account(direction, n, host, port, method) {
++        if (spent || !(n > 0)) return spent ? 0 : n;
++        const left = limits[direction] - bytes[direction];
++        const taken = Math.min(n, Number.isFinite(left) ? Math.max(0, left) : n);
++        bytes[direction] += taken;
++        if (taken < n) {
++            spent = true;
++            spentAt = new Date().toISOString();
++            spentDirection = direction;
++            report(method || 'TRANSFER', { host, port, code: 'egress.budget', reason: `the session's ${direction} budget is spent` });
++        }
++        return taken;
++    }
++
++    function spentRefusal(host, port, method) {
++        refusals.budget++;
++        const d = refuse(host, port, 'egress.budget', "the session's egress budget is spent");
++        report(method, d);
++        return d;
++    }
++
      function resolveAll(host) {
          return new Promise((resolve) => {
              let settled = false;
@@@ -112,8 +190,17 @@ function createEgressProxy({
          res.end(`${text}\n`);
      }
  
++    /** The session's own budget is checked first: once it is spent every new request says so, whatever it asks for. */
++    function budgetDenied(host, port, method) {
++        return spentRefusal(host, port, method);
++    }
++
      async function onRequest(req, res) {
          if (req.socket.ovRefused) return;
++        if (spent) {
++            const shared = budgetDenied(null, null, req.method);
++            return deny(res, 429, shared.code, "The session's egress budget is spent");
++        }
          let url;
          try { url = new URL(req.url); } catch { return deny(res, 400, 'egress.bad_request', 'This is a forward proxy: send an absolute http:// URL, or CONNECT for https'); }
          if (url.protocol === 'https:') return deny(res, 400, 'egress.use_connect', 'Use CONNECT for https');
@@@ -122,6 +209,24 @@ function createEgressProxy({
          const d = await check(url.hostname, url.port ? Number(url.port) : 80);
          report(req.method, d);
          if (!d.ok) return deny(res, 403, d.code, `Refused: ${d.reason}`);
++        if (spent) {
++            const shared = budgetDenied(d.host, d.port, req.method);
++            return deny(res, 429, shared.code, "The session's egress budget is spent");
++        }
++        const h = hostState(d.host);
++        if (!take(sessionBucket, sessionRate) || !take(h, hostRate)) {
++            refusals.rate++;
++            const limited = refuse(d.host, d.port, 'egress.rate', 'this session is sending requests too fast');
++            report(req.method, limited);
++            return deny(res, 429, limited.code, 'Too many requests: slow down');
++        }
++        if (h.active >= perHost) {
++            refusals.connections++;
++            const busy = refuse(d.host, d.port, 'egress.connections', `this session has ${h.active} connections open to ${d.host}`);
++            report(req.method, busy);
++            return deny(res, 429, busy.code, `Too many connections to ${d.host}: one at a time, please`);
++        }
++        h.active++;
  
          const headers = {};
          for (let i = 0; i < req.rawHeaders.length; i += 2) {
@@@ -137,37 +242,90 @@ function createEgressProxy({
              host: d.address, port: d.port, method: req.method, path: url.pathname + url.search, headers, timeout: connectTimeoutMs,
              createConnection: () => dial({ host: d.address, port: d.port }),
          });
++        let released = false;
++        const release = () => { if (!released) { released = true; h.active = Math.max(0, h.active - 1); } };
++        upstream.on('close', release);
          upstream.on('response', (up) => {
              const out = [];
              for (let i = 0; i < up.rawHeaders.length; i += 2) if (!HOP.has(up.rawHeaders[i].toLowerCase())) out.push(up.rawHeaders[i], up.rawHeaders[i + 1]);
              res.writeHead(up.statusCode, up.statusMessage, out);
--            up.pipe(res);
++            // The body is counted as it arrives: a download that spends the budget is cut off at that byte, and the
++            // client sees the response end early (the next request gets the 429).
++            up.on('data', (chunk) => {
++                if (res.destroyed) return;
++                const n = account('down', chunk.length, d.host, d.port, req.method);
++                if (n > 0) {
++                    const body = n === chunk.length ? chunk : chunk.subarray(0, n);
++                    if (!res.write(body)) { up.pause(); res.once('drain', () => up.resume()); }
++                }
++                if (n < chunk.length) { up.destroy(); res.end(); }
++            });
++            up.on('end', () => { if (!res.destroyed && !res.writableEnded) res.end(); });
++            up.on('error', () => { if (!res.headersSent) deny(res, 502, 'egress.upstream', 'The site did not answer'); else res.destroy(); });
++        });
++        req.on('data', (chunk) => {
++            if (account('up', chunk.length, d.host, d.port, req.method) < chunk.length) {
++                req.destroy();
++                upstream.destroy();
++                if (!res.headersSent) deny(res, 429, 'egress.budget', "The session's egress budget is spent");
++            }
          });
          upstream.on('timeout', () => upstream.destroy(Object.assign(new Error('timeout'), { code: 'ETIMEDOUT' })));
--        upstream.on('error', () => deny(res, 502, 'egress.upstream', 'The site did not answer'));
++        upstream.on('error', () => { release(); deny(res, 502, 'egress.upstream', 'The site did not answer'); });
          req.pipe(upstream);
      }
  
      async function onConnect(req, socket, head) {
          socket.on('error', () => {});
          if (socket.ovRefused) return;
++        if (spent) {
++            const shared = budgetDenied(null, null, 'CONNECT');
++            return socket.end(refusalHead(429, shared.code, "The session's egress budget is spent"));
++        }
          const m = /^\[([0-9a-fA-F:.]+)\]:(\d{1,5})$/.exec(req.url) || /^([^\s:[\]/@]+):(\d{1,5})$/.exec(req.url);
          if (!m) return socket.end(refusalHead(400, 'egress.bad_request', 'CONNECT needs host:port'));
          const d = await check(m[1], Number(m[2]));
          report('CONNECT', d);
          if (!d.ok) return socket.end(refusalHead(403, d.code, `Refused: ${d.reason}`));
++        if (spent) {
++            const shared = budgetDenied(d.host, d.port, 'CONNECT');
++            return socket.end(refusalHead(429, shared.code, "The session's egress budget is spent"));
++        }
++        const h = hostState(d.host);
++        if (!take(sessionBucket, sessionRate) || !take(h, hostRate)) {
++            refusals.rate++;
++            const limited = refuse(d.host, d.port, 'egress.rate', 'this session is sending requests too fast');
++            report('CONNECT', limited);
++            return socket.end(refusalHead(429, limited.code, 'Too many requests: slow down'));
++        }
++        if (h.active >= perHost) {
++            refusals.connections++;
++            const busy = refuse(d.host, d.port, 'egress.connections', `this session has ${h.active} connections open to ${d.host}`);
++            report('CONNECT', busy);
++            return socket.end(refusalHead(429, busy.code, `Too many connections to ${d.host}: one at a time, please`));
++        }
++        h.active++;
          const up = dial({ host: d.address, port: d.port });
++        let released = false;
++        const release = () => { if (!released) { released = true; h.active = Math.max(0, h.active - 1); } };
          const timer = setTimeout(() => up.destroy(), connectTimeoutMs);
++        const cut = (direction) => (chunk) => {
++            const n = account(direction, chunk.length, d.host, d.port, 'CONNECT');
++            if (n < chunk.length) { up.destroy(); socket.destroy(); }
++        };
          up.once('connect', () => {
              clearTimeout(timer);
              socket.write('HTTP/1.1 200 Connection Established\r\n\r\n');
              if (head && head.length) up.write(head);
++            // Both directions count: the tunnel closes the moment either budget is spent.
++            up.on('data', cut('down'));
++            socket.on('data', cut('up'));
              up.pipe(socket);
              socket.pipe(up);
          });
          up.on('error', () => { clearTimeout(timer); if (!socket.destroyed) socket.end(refusalHead(502, 'egress.upstream', 'The site did not answer')); });
--        socket.on('close', () => up.destroy());
--        up.on('close', () => socket.destroy());
++        up.on('close', () => { clearTimeout(timer); release(); socket.destroy(); });
++        socket.on('close', () => { release(); up.destroy(); });
      }
  
      function onConnection(sock) {
@@@ -207,6 +365,16 @@ function createEgressProxy({
          },
          address: () => server && server.address(),
          connections: () => sockets.size,
++        /** What this session has spent and the limits it ran under (the worker's session report carries this). */
++        stats: () => ({
++            down_bytes: bytes.down, up_bytes: bytes.up,
++            down_limit_bytes: Number.isFinite(limits.down) ? limits.down : null,
++            up_limit_bytes: Number.isFinite(limits.up) ? limits.up : null,
++            spent, spent_at: spentAt, spent_direction: spentDirection,
++            budget_refusals: refusals.budget, rate_refusals: refusals.rate, connection_refusals: refusals.connections,
++            host_requests_per_second: hostRate.rate || null, session_requests_per_second: sessionRate.rate || null,
++            max_connections_per_host: Number.isFinite(perHost) ? perHost : null,
++        }),
          async close() {
              if (!server) return;
              for (const s of sockets) s.destroy();
@@@ -216,4 +384,7 @@ function createEgressProxy({
      };
  }
  
--module.exports = { createEgressProxy, PORTS, defaultHostAddresses };
++module.exports = {
++    createEgressProxy, PORTS, defaultHostAddresses,
++    DEFAULT_BUDGET_DOWN, DEFAULT_BUDGET_UP, DEFAULT_HOST_REQUESTS, DEFAULT_SESSION_REQUESTS, DEFAULT_MAX_PER_HOST,
++};
ddiff --git a/sandbox/session.js b/sandbox/session.js
iindex 6f7e3f8..e7d2721 100644
---- a/sandbox/session.js
++++ b/sandbox/session.js
@@@ -12,8 +12,9 @@
   *       sbx.ended (Promise of { reason }), sbx.state ('running' | 'wiping' | 'wiped'), sbx.id, sbx.pid, sbx.dir, sbx.cgroup
   *
   *   opts: runtimeDir (short: a socket path must fit 108 bytes), limits { memoryMb, diskMb, pids, cpuPercent }, timeBoxMs,
-- *         limiter (sandbox/limits.js; null only with allowUnlimited, for tests), egress { lookup, dial, hostAddresses }
-- *         (sandbox/egress-proxy.js), onEgress(decision), nodePath, browserPath, rootfs, bwrap, seccomp, log
++ *         limiter (sandbox/limits.js; null only with allowUnlimited, for tests), egress { lookup, dial, hostAddresses,
++ *         budgetDownBytes, budgetUpBytes, hostRequests, sessionRequests, maxConnectionsPerHost } (sandbox/egress-proxy.js),
++ *         onEgress(decision), nodePath, browserPath, rootfs, bwrap, seccomp, log
   *
   * How it is made (sandbox/bwrap.js has the whole list): bubblewrap with its own user, pid, network, ipc, uts and cgroup
   * namespaces, a read-only root of /usr and a few /etc entries, size-limited tmpfs for /workspace, /home/agent and /tmp,
@@@ -336,7 +337,11 @@ class Sandbox {
      }
  
      info() {
--        return { id: this.id, state: this.state, started_at: new Date(this.startedAt).toISOString(), ends_at: new Date(this.endsAt).toISOString(), limits: this.limits };
++        return {
++            id: this.id, state: this.state, started_at: new Date(this.startedAt).toISOString(), ends_at: new Date(this.endsAt).toISOString(), limits: this.limits,
++            // What the session has spent on egress and the limits it ran under; spent is true once the budget is gone.
++            egress: this._proxy && this._proxy.stats ? this._proxy.stats() : null,
++        };
      }
  }
  
ddiff --git a/sandbox/worker.js b/sandbox/worker.js
iindex a8b2d1f..289161f 100644
---- a/sandbox/worker.js
++++ b/sandbox/worker.js
@@@ -67,6 +67,15 @@ function loadConfig(env = process.env) {
              cpuPercent: int(env.CODES_SANDBOX_CPU_PERCENT, 100, 10, 6400),
          },
          maxSessions: int(env.CODES_SANDBOX_MAX_SESSIONS, 2, 1, 64),
++        // Each session's egress budget (design §1): 500 MB down and 100 MB up unless the owner sets it. The rate and
++        // connection limits are the proxy's own defaults (10/s burst 30 per host, 50/s per session, 8 per host).
++        egress: {
++            budgetDownBytes: int(env.CODES_SANDBOX_EGRESS_MB_DOWN, 500, 1, 1024 * 1024) * MB,
++            budgetUpBytes: int(env.CODES_SANDBOX_EGRESS_MB_UP, 100, 1, 1024 * 1024) * MB,
++            hostRequests: { rate: 10, burst: 30 },
++            sessionRequests: { rate: 50, burst: 50 },
++            maxConnectionsPerHost: 8,
++        },
          // The owner's time box is 5 to 30 minutes (design §3); the worker's own ceiling is that, never more.
          timeBoxMinMinutes: 5,
          timeBoxMaxMinutes: int(env.CODES_SANDBOX_TIME_BOX_MAX_MIN, 30, 5, 30),
@@@ -118,6 +127,15 @@ function createWorker(opts = {}) {
      /** What the worker can enforce, as it found it (never a secret, a path or an address). */
      function isolation() {
          const c = (lastSelfTest && lastSelfTest.checks) || {};
++        // Live reads these five facts (server/agent-streams/sandbox.js REQUIRED_ISOLATION) and refuses a report that
++        // does not say each one exactly: false, false, false, true, 'ephemeral'. An unproven fact is left out, so a
++        // worker that cannot show it never registers instead of claiming it.
++        const live = {};
++        if (c.network_namespace === true) live.host_network = false;
++        if (c.clean_env === true && c.no_host_files === true && c.no_capabilities === true) live.credentials = false;
++        if (c.no_host_files === true) live.profile = false;
++        if (c.egress_refuses_private === true && c.no_direct_egress === true) live.egress_proxy = true;
++        if (c.wipe === true) live.workspace = 'ephemeral';
          return {
              sandbox: 'bubblewrap', bubblewrap: support && support.version || null,
              user_namespace: !!(support && support.ok), network_namespace: c.network_namespace === true,
@@@ -126,7 +144,8 @@ function createWorker(opts = {}) {
              host_files: c.no_host_files === true ? 'none' : 'unknown',
              capabilities: c.no_capabilities === true ? 'none' : 'unknown',
              seccomp: c.seccomp === true,
--            workspace: 'tmpfs, wiped with the sandbox',
++            ...live,
++            workspace_kind: 'tmpfs, wiped with the sandbox',
              limits: {
                  kind: limiter && limiter.ok ? limiter.kind : 'none', enforced: c.limits === true,
                  memory_mb: config.limits.memoryMb, disk_mb: config.limits.diskMb, pids: config.limits.pids, cpu_percent: config.limits.cpuPercent,
@@@ -204,7 +223,7 @@ function createWorker(opts = {}) {
          const egressLog = [];
          const sbx = await createSandbox({
              id, runtimeDir: config.runtimeDir, limiter, limits: config.limits, timeBoxMs: box * 60_000, nodePath: config.nodePath,
--            browserPath: config.browserPath, log,
++            browserPath: config.browserPath, egress: config.egress, log,
              onEgress: (d) => { egressLog.push(d); if (egressLog.length > 500) egressLog.shift(); },
          });
          const entry = { id, sbx, stream: stream ? String(stream) : null, egressLog, created: now() };
@@@ -213,7 +232,18 @@ function createWorker(opts = {}) {
          return entry;
      }
  
--    const view = (e) => ({ ...e.sbx.info(), stream_id: e.stream, egress: { allowed: e.egressLog.filter((d) => d.allowed).length, refused: e.egressLog.filter((d) => !d.allowed).length } });
++    // The session's report: what it spent on egress (spent and why), beside the allow/refuse counts of its log.
++    const view = (e) => {
++        const info = e.sbx.info();
++        return {
++            ...info, stream_id: e.stream,
++            egress: {
++                ...(info.egress || {}),
++                allowed: e.egressLog.filter((d) => d.allowed).length,
++                refused: e.egressLog.filter((d) => !d.allowed).length,
++            },
++        };
++    };
  
      // ── The local API (a Unix socket; the runner on this host is its only client) ──
  
ddiff --git a/test/helpers/sandbox-world.js b/test/helpers/sandbox-world.js
iindex 86c43c9..e9d5360 100644
---- a/test/helpers/sandbox-world.js
++++ b/test/helpers/sandbox-world.js
@@@ -27,6 +27,8 @@ const SELF_IP = '93.184.215.99';        // public, and one of the host's own add
  
  const NAMES = {
      'origin.example': [ORIGIN_IP],
++    // The same public origin under a second name: per-destination limits can be told apart from the session's.
++    'alt.example': [ORIGIN_IP],
      'smtp.example': [SMTP_IP],
      'self.example': [SELF_IP],
      'private.example': ['10.0.0.5'],
@@@ -82,6 +84,11 @@ async function startWorld() {
          const u = new URL(req.url, 'http://origin.example');
          if (u.pathname === '/redirect') { res.writeHead(302, { location: u.searchParams.get('to') || '/' }); return res.end(); }
          if (u.pathname === '/page') { res.writeHead(200, { 'content-type': 'text/html' }); return res.end('<!doctype html><title>Origin page</title><h1 id="h">origin-ok</h1>'); }
++        if (u.pathname === '/big') {
++            const n = Math.min(8 * 1024 * 1024, Math.max(1, parseInt(u.searchParams.get('n'), 10) || 1024));
++            res.writeHead(200, { 'content-type': 'application/octet-stream' });
++            return res.end(Buffer.alloc(n, 0x61));
++        }
          res.writeHead(200, { 'content-type': 'text/plain' });
          res.end('origin-ok');
      });
ddiff --git a/test/sandbox-egress-proxy.test.js b/test/sandbox-egress-proxy.test.js
iindex f824b2f..b12c844 100644
---- a/test/sandbox-egress-proxy.test.js
++++ b/test/sandbox-egress-proxy.test.js
@@@ -182,6 +182,129 @@ const connect = (port, authority) => raw(port, `CONNECT ${authority} HTTP/1.1\r\
          await small.close();
      });
  
++    await check('a download that spends the session\'s budget is cut off, and new requests get 429 egress.budget', async () => {
++        const budget = createEgressProxy({ ...world.egress, budgetDownBytes: 256, budgetUpBytes: 4096 });
++        await budget.listen({ port: 0, host: '127.0.0.1' });
++        const p = budget.address().port;
++        const first = await get(p, 'http://origin.example/big?n=65536');
++        assert.strictEqual(first.status, 200);
++        assert.ok(first.body.length <= 256, `the body stops at the budget (${first.body.length} bytes)`);
++        assert.ok(first.body.length > 0, 'some of it was delivered before the cut-off');
++        const stats = budget.stats();
++        assert.strictEqual(stats.spent, true, 'the session report says the budget is spent');
++        assert.strictEqual(stats.spent_direction, 'down');
++        assert.strictEqual(stats.down_bytes, 256, 'it counted exactly the budget');
++        const second = await get(p, 'http://origin.example/');
++        assert.strictEqual(second.status, 429);
++        assert.match(String(second.headers['x-ov-egress']), /^refused; code=egress\.budget$/);
++        assert.ok(budget.stats().budget_refusals >= 1);
++        // The tunnel counts too, and closes when the budget is spent.
++        const tunnel = createEgressProxy({ ...world.egress, budgetDownBytes: 2, budgetUpBytes: 4096 });
++        await tunnel.listen({ port: 0, host: '127.0.0.1' });
++        const t = await new Promise((resolve) => {
++            const s = net.connect(tunnel.address().port, '127.0.0.1', () => s.write('CONNECT origin.example:443 HTTP/1.1\r\nHost: origin.example:443\r\n\r\n'));
++            let buf = ''; let open = false;
++            s.on('data', (d) => { buf += d; if (!open && buf.includes('\r\n\r\n')) { open = true; buf = ''; s.write('ping'); } });
++            const done = () => resolve({ open, reply: buf });
++            s.on('end', done);
++            s.on('error', done);
++            setTimeout(() => { s.destroy(); done(); }, 3000);
++        });
++        assert.strictEqual(t.open, true, 'the tunnel opened');
++        await new Promise((r) => setTimeout(r, 100));
++        assert.strictEqual(tunnel.stats().spent, true, 'the tunnel spent the down budget and closed');
++        assert.strictEqual(tunnel.stats().spent_direction, 'down');
++        const after = await get(tunnel.address().port, 'http://origin.example/');
++        assert.strictEqual(after.status, 429);
++        assert.match(String(after.headers['x-ov-egress']), /code=egress\.budget/);
++
++        // The other direction counts too: a tunnel whose upload spends the up budget closes as well.
++        const upload = createEgressProxy({ ...world.egress, budgetDownBytes: 4096, budgetUpBytes: 2 });
++        await upload.listen({ port: 0, host: '127.0.0.1' });
++        const u = await new Promise((resolve) => {
++            const s = net.connect(upload.address().port, '127.0.0.1', () => s.write('CONNECT origin.example:443 HTTP/1.1\r\nHost: origin.example:443\r\n\r\n'));
++            let buf = ''; let open = false;
++            s.on('data', (d) => { buf += d; if (!open && buf.includes('\r\n\r\n')) { open = true; s.write('ping'); } });
++            const done = () => resolve({ open });
++            s.on('end', done);
++            s.on('error', done);
++            setTimeout(() => { s.destroy(); done(); }, 3000);
++        });
++        assert.strictEqual(u.open, true, 'the tunnel opened');
++        await new Promise((r) => setTimeout(r, 100));
++        assert.strictEqual(upload.stats().spent, true);
++        assert.strictEqual(upload.stats().spent_direction, 'up', 'the up budget is the one that went');
++
++        // Another session has its own budget: the same download there is not affected.
++        const other = createEgressProxy({ ...world.egress, budgetDownBytes: 256, budgetUpBytes: 4096 });
++        await other.listen({ port: 0, host: '127.0.0.1' });
++        const ok = await get(other.address().port, 'http://origin.example/');
++        assert.strictEqual(ok.status, 200);
++        assert.strictEqual(ok.body, 'origin-ok');
++        assert.strictEqual(other.stats().spent, false, 'a fresh session has its own budget');
++        for (const s of [budget, tunnel, upload, other]) await s.close();
++    });
++
++    await check('a session holds at most the per-host connection cap', async () => {
++        const capped = createEgressProxy({ ...world.egress, maxConnectionsPerHost: 2 });
++        await capped.listen({ port: 0, host: '127.0.0.1' });
++        const p = capped.address().port;
++        const held = [];
++        for (let i = 0; i < 2; i++) {
++            await new Promise((resolve, reject) => {
++                const s = net.connect(p, '127.0.0.1', () => s.write('CONNECT origin.example:443 HTTP/1.1\r\nHost: origin.example:443\r\n\r\n'));
++                let buf = '';
++                s.on('data', (d) => { buf += d; if (buf.includes('\r\n\r\n')) resolve(s); });
++                s.on('error', reject);
++                held.push(s);
++                setTimeout(() => reject(new Error('the tunnel did not open')), 3000);
++            });
++        }
++        const third = await get(p, 'http://origin.example/');
++        assert.strictEqual(third.status, 429);
++        assert.match(String(third.headers['x-ov-egress']), /code=egress\.connections/);
++        assert.ok(capped.stats().connection_refusals >= 1);
++        for (const s of held) s.destroy();
++        await capped.close();
++    });
++
++    await check('a flood to one host is rate-limited, another host is not, and another session is unaffected', async () => {
++        const limited = createEgressProxy({
++            ...world.egress, hostRequests: { rate: 2, burst: 3 }, sessionRequests: { rate: 50, burst: 50 },
++        });
++        await limited.listen({ port: 0, host: '127.0.0.1' });
++        const p = limited.address().port;
++        const statuses = [];
++        for (let i = 0; i < 6; i++) statuses.push((await get(p, 'http://origin.example/')).status);
++        assert.deepStrictEqual(statuses.slice(0, 3), [200, 200, 200], 'the burst gets through');
++        assert.ok(statuses.slice(3).every((s) => s === 429), `the flood is refused (${statuses.join(',')})`);
++        const denied = await get(p, 'http://origin.example/');
++        assert.match(String(denied.headers['x-ov-egress']), /code=egress\.rate/);
++        // The per-host bucket is the host's: a different destination still has its own.
++        assert.strictEqual((await get(p, 'http://alt.example/')).status, 200);
++        assert.ok(limited.stats().rate_refusals >= 1);
++        // A second session starts with its own buckets.
++        const fresh = createEgressProxy({ ...world.egress, hostRequests: { rate: 2, burst: 3 } });
++        await fresh.listen({ port: 0, host: '127.0.0.1' });
++        assert.strictEqual((await get(fresh.address().port, 'http://origin.example/')).status, 200);
++        await limited.close();
++        await fresh.close();
++    });
++
++    await check('the per-session request rate is shared by the session\'s destinations', async () => {
++        const limited = createEgressProxy({
++            ...world.egress, hostRequests: { rate: 100, burst: 100 }, sessionRequests: { rate: 1, burst: 2 },
++        });
++        await limited.listen({ port: 0, host: '127.0.0.1' });
++        const p = limited.address().port;
++        assert.strictEqual((await get(p, 'http://origin.example/')).status, 200);
++        assert.strictEqual((await get(p, 'http://alt.example/')).status, 200);
++        const third = await get(p, 'http://origin.example/');
++        assert.strictEqual(third.status, 429);
++        assert.match(String(third.headers['x-ov-egress']), /code=egress\.rate/);
++        await limited.close();
++    });
++
      await proxy.close();
      await world.close();
      console.log(failures ? `\n${failures} failed` : '\nall passed');
ddiff --git a/test/sandbox-worker.test.js b/test/sandbox-worker.test.js
iindex 1363cb4..c296a4c 100644
---- a/test/sandbox-worker.test.js
++++ b/test/sandbox-worker.test.js
@@@ -237,6 +237,14 @@ async function check(name, fn) {
          assert.strictEqual(body.isolation.network_namespace, true);
          assert.strictEqual(body.isolation.host_credentials, 'none');
          assert.strictEqual(body.isolation.limits.enforced, true);
++        // The five facts Live's registration requires (server/agent-streams/sandbox.js REQUIRED_ISOLATION), exactly.
++        assert.deepStrictEqual(
++            {
++                host_network: body.isolation.host_network, credentials: body.isolation.credentials, profile: body.isolation.profile,
++                egress_proxy: body.isolation.egress_proxy, workspace: body.isolation.workspace,
++            },
++            { host_network: false, credentials: false, profile: false, egress_proxy: true, workspace: 'ephemeral' },
++            'the report says what Live needs to register a sandbox worker');
          for (const c of calls) {
              assert.ok(!String(c.init.body).includes(CODES_SECRET), 'the Codes secret never travels');
              assert.ok(!String(c.init.body).includes(tmp), 'no path of the host');