diff --git a/scripts/e2e-job-flow.sh b/scripts/e2e-job-flow.sh
new file mode 100755
index 0000000..3cb076e
--- /dev/null
+++ b/scripts/e2e-job-flow.sh
@@ -0,0 +1,242 @@
+#!/usr/bin/env bash
+# End-to-end test of the print loop, with every machine faked: upload a file to
+# the REAL object store through the presigned URL the API hands out, let the
+# REAL scheduler dispatch it over the REAL /ws/edge/v1 to a stand-in gateway,
+# which downloads it from the REAL presigned GET and hashes it, then reports
+# the print running and finished. The job must be completed, and the bytes the
+# gateway received must be the bytes uploaded.
+#
+# SAFETY: dispatch is ON here, deliberately - so this is only safe because the
+# only gateway this backend instance can have is the stand-in (a real proxy
+# would have to be pointed at this instance's random port). No printer is
+# touched.
+#
+# Needs the real object-store settings: set -a; . ../dev-secrets.sh; set +a
+#
+# This is the test that the unit tests cannot be. tests/authz.test.ts proves
+# tokens are verified and that the role ladder orders correctly; only this
+# proves that each route actually asks, that scoping is applied in the query
+# rather than after it, and that credentials stay out of responses - the three
+# things that were wrong when every route was unauthenticated.
+#
+# Four users, two organisations, no printers or gateways needed:
+#   owner_a, operator_a, viewer_a  in org A, which owns farm A and printer A
+#   outsider_b                     in org B, which owns nothing
+#
+# Usage:
+#   scripts/dev-env.sh up && scripts/dev-env.sh schema
+#   scripts/e2e-authz.sh
+#
+#   KEEP_DATA=1   leave the seeded organisations behind for inspection
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$REPO_ROOT"
+
+PORT="${PORT:-3037}"
+AUTH_PORT="${AUTH_PORT:-3038}"
+BASE="http://localhost:$PORT"
+DATABASE_URL="${DATABASE_URL:-postgres://continuum:continuum@localhost:5432/continuum}"
+VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
+EDGE_JWT_SECRET="${EDGE_JWT_SECRET:-e2e-edge-secret}"
+# Presigning a PUT is local HMAC with no network call, so POST /jobs works
+# offline - but the SDK refuses to sign with empty credentials.
+R2_ACCESS_KEY_ID="${R2_ACCESS_KEY_ID:-}"
+R2_SECRET_ACCESS_KEY="${R2_SECRET_ACCESS_KEY:-}"
+R2_ENDPOINT="${R2_ENDPOINT:-}"
+export DATABASE_URL VALKEY_URL EDGE_JWT_SECRET PORT
+export R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY R2_ENDPOINT
+
+WORK="$(mktemp -d)"
+BACKEND_PID=""
+AUTH_PID=""
+FAILURES=0
+
+# `uuidgen` is not installed everywhere; the kernel's generator always is.
+new_uuid() {
+  if [ -r /proc/sys/kernel/random/uuid ]; then
+    cat /proc/sys/kernel/random/uuid
+  else
+    python3 -c 'import uuid; print(uuid.uuid4())'
+  fi
+}
+
+# Uses DATABASE_URL rather than a hardcoded localhost. The backend under test
+# reads that variable, so a psql pointed anywhere else seeds fixtures into a
+# different database than the one being exercised - and the failure message
+# below names DATABASE_URL, which would then be naming a URL nothing tried.
+psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; }
+
+cleanup() {
+  local rc=$?
+  # Never `kill $PID` with an unset PID: `kill 0` signals the whole process
+  # group, which includes this script and whatever launched it.
+  local pids=()
+  [ -n "$BACKEND_PID" ] && pids+=("$BACKEND_PID")
+  [ -n "$AUTH_PID" ] && pids+=("$AUTH_PID")
+  if [ ${#pids[@]} -gt 0 ]; then
+    kill "${pids[@]}" 2>/dev/null || true
+    sleep 0.3
+    kill -9 "${pids[@]}" 2>/dev/null || true
+  fi
+  if [ -z "${KEEP_DATA:-}" ]; then
+    for org in "${ORG_A:-}" "${ORG_B:-}"; do
+      [ -n "$org" ] && psql_q "DELETE FROM organisations WHERE id = '$org'" >/dev/null 2>&1 || true
+    done
+  fi
+  if [ $rc -ne 0 ]; then
+    echo
+    echo "--- backend log (tail) ---"; tail -30 "$WORK/backend.log" 2>/dev/null || true
+    echo "--- fake gateway log ---"; cat "$WORK/gw.log" 2>/dev/null || true
+  fi
+  rm -rf "$WORK"
+  exit $rc
+}
+trap cleanup EXIT
+
+fail() { echo "FAIL: $*" >&2; exit 1; }
+ok()   { echo "  ok: $*"; }
+bad()  { echo "  FAILED: $*" >&2; FAILURES=$((FAILURES + 1)); }
+
+# `req <method> <path> <token|-> [json body]` -> prints the status code,
+# leaves the response body in $WORK/last.json.
+req() {
+  local method="$1" path="$2" token="$3" body="${4:-}"
+  local args=(-s -o "$WORK/last.json" -w '%{http_code}' -X "$method" "$BASE$path")
+  [ "$token" != "-" ] && args+=(-H "authorization: Bearer $token")
+  [ -n "$body" ] && args+=(-H 'content-type: application/json' -d "$body")
+  curl "${args[@]}"
+}
+
+# `expect <code> <method> <path> <token> [body] -- <description>`
+expect() {
+  local want="$1"; shift
+  local method="$1" path="$2" token="$3"; shift 3
+  local body=""
+  if [ "${1:-}" != "--" ]; then body="$1"; shift; fi
+  shift # the --
+  local what="$*"
+  local got; got="$(req "$method" "$path" "$token" "$body")"
+  if [ "$got" = "$want" ]; then
+    ok "$what ($got)"
+  else
+    bad "$what: expected $want, got $got - $(head -c 300 "$WORK/last.json")"
+  fi
+}
+
+body_has()    { grep -q "$1" "$WORK/last.json" || bad "expected '$1' in: $(head -c 300 "$WORK/last.json")"; }
+body_lacks()  { grep -q "$1" "$WORK/last.json" && bad "'$1' leaked into: $(head -c 300 "$WORK/last.json")" || true; }
+json()        { python3 -c "import json,sys; d=json.load(sys.stdin); print($1)" < "$WORK/last.json"; }
+
+# --- 0. preconditions ------------------------------------------------------
+
+echo "==> checking the database is up and migrated"
+psql_q "SELECT 1" >/dev/null 2>&1 || fail "no database at $DATABASE_URL - run scripts/dev-env.sh up"
+psql_q "SELECT org_id FROM farms LIMIT 0" >/dev/null 2>&1 \
+  || fail "farms.org_id is missing - run scripts/dev-env.sh schema"
+psql_q "SELECT role FROM org_members LIMIT 0" >/dev/null 2>&1 \
+  || fail "org_members is missing - run scripts/dev-env.sh schema"
+ok "schema has the org layer"
+
+# --- 1. the issuer and the backend ----------------------------------------
+
+echo "==> starting the dev-auth JWKS issuer on :$AUTH_PORT"
+bun scripts/dev-auth.ts --port "$AUTH_PORT" > "$WORK/auth.log" 2>&1 &
+AUTH_PID=$!
+AUTH_URL="http://localhost:$AUTH_PORT"
+for _ in $(seq 1 40); do
+  curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 && break
+  sleep 0.25
+done
+curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 \
+  || fail "the dev-auth issuer did not come up: $(cat "$WORK/auth.log")"
+export SUPABASE_URL="$AUTH_URL"
+ok "issuer up, SUPABASE_URL=$SUPABASE_URL"
+
+mint() { curl -sf "$AUTH_URL/token?sub=$1" | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])'; }
+
+
+echo "==> starting continuum-backend on :$PORT"
+bun run src/index.ts > "$WORK/backend.log" 2>&1 &
+BACKEND_PID=$!
+for _ in $(seq 1 60); do
+  curl -sf "$BASE/health" >/dev/null 2>&1 && break
+  sleep 0.5
+done
+curl -sf "$BASE/health" >/dev/null 2>&1 || fail "backend did not become healthy: $(cat "$WORK/backend.log")"
+ok "backend healthy"
+
+# --- 2. a farm with one idle printer and a stand-in gateway ----------------
+
+OWNER_ID="$(new_uuid)"; OWNER="$(mint "$OWNER_ID")"
+expect 201 POST /orgs "$OWNER" '{"name":"e2e job-flow org"}' -- "create an org"
+ORG_A="$(json 'd["id"]')"
+expect 201 POST /farms "$OWNER" "{\"name\":\"e2e job-flow farm\",\"orgId\":\"$ORG_A\"}" -- "create a farm"
+FARM_A="$(json 'd["id"]')"
+expect 201 POST /printers "$OWNER" \
+  "{\"farmId\":\"$FARM_A\",\"name\":\"Fake P1\",\"vendor\":\"bambu\",\"gatewayPrinterId\":\"fakep\"}" -- "claim a printer"
+PRINTER_A="$(json 'd["id"]')"
+expect 200 POST "/farms/$FARM_A/edge-proxy-token" "$OWNER" -- "mint the gateway token"
+PROXY_TOKEN="$(json 'd["edgeProxyToken"]')"
+expect 200 POST /auth/edge-token - "{\"farmId\":\"$FARM_A\",\"edgeProxyToken\":\"$PROXY_TOKEN\"}" -- "trade it for an edge JWT"
+EDGE_JWT="$(json 'd["token"]')"
+FRAMES="$WORK/frames.jsonl"; : > "$FRAMES"
+FARM_ID="$FARM_A" REPORT_IDLE=fakep bun scripts/fake-gateway-serve.ts "ws://localhost:$PORT/ws/edge/v1" "$EDGE_JWT" 90 "$FRAMES" > "$WORK/gw.log" 2>&1 &
+GW_PID=$!
+for _ in $(seq 1 40); do grep -q ready "$WORK/gw.log" 2>/dev/null && break; sleep 0.25; done
+grep -q ready "$WORK/gw.log" || fail "the stand-in gateway did not connect: $(cat "$WORK/gw.log")"
+for _ in $(seq 1 40); do
+  [ "$(psql_q "SELECT status FROM printers WHERE id='$PRINTER_A'")" = "idle" ] && break
+  sleep 0.25
+done
+[ "$(psql_q "SELECT status FROM printers WHERE id='$PRINTER_A'")" = "idle" ] && ok "the printer reports idle" || fail "the printer never reported idle"
+
+# --- 3. upload, dispatch, print --------------------------------------------
+
+echo "==> upload a file through the presigned URL"
+head -c 200000 /dev/urandom > "$WORK/cube.3mf"
+WANT="$(sha256sum "$WORK/cube.3mf" | cut -d' ' -f1)"
+expect 201 POST /jobs "$OWNER" "{\"farmId\":\"$FARM_A\",\"fileName\":\"cube.3mf\",\"contentType\":\"application/octet-stream\"}" -- "queue a job"
+JOB_A="$(json 'd["job"]["id"]')"
+OBJECT_KEY="$(json 'd["job"]["r2ObjectKey"]')"
+UPLOAD_URL="$(json 'd["uploadUrl"]')"
+cleanup_object() { [ -n "${OBJECT_KEY:-}" ] && bun -e "
+import { s3 } from './src/s3';
+import { DeleteObjectCommand } from '@aws-sdk/client-s3';
+await s3.send(new DeleteObjectCommand({ Bucket: process.env.R2_BUCKET ?? 'continuum', Key: process.argv[1] }));
+" "$OBJECT_KEY" >/dev/null 2>&1 || true; }
+trap 'cleanup_object; cleanup' EXIT
+CODE="$(curl -s -o /dev/null -w '%{http_code}' -X PUT -H 'Content-Type: application/octet-stream' --data-binary @"$WORK/cube.3mf" "$UPLOAD_URL")"
+[ "$CODE" = "200" ] && ok "the file reached the object store (HTTP 200)" || fail "the upload returned HTTP $CODE"
+
+echo "==> the scheduler dispatches it and the print runs to completion"
+SEEN=""
+for _ in $(seq 1 80); do
+  STATUS="$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_A'")"
+  case " $SEEN " in *" $STATUS "*) ;; *) SEEN="$SEEN $STATUS" ;; esac
+  [ "$STATUS" = "completed" ] && break
+  [ "$STATUS" = "failed" ] && break
+  sleep 1
+done
+[ "$STATUS" = "completed" ] && ok "the job completed (saw:$SEEN)" || bad "the job ended as '$STATUS' (saw:$SEEN)"
+case "$SEEN" in *printing*) ok "it was seen printing on the way" ;; *) bad "the job was never printing: $SEEN" ;; esac
+
+python3 - "$FRAMES" "$WANT" <<'PYCHK' && ok "the gateway got start_print for this job, and downloaded exactly the bytes that were uploaded" || bad "the gateway's view does not match: $(cat "$FRAMES")"
+import json,sys
+frames=[json.loads(l) for l in open(sys.argv[1])]
+start=[f for f in frames if f["type"]=="start_print"]
+dl=[f for f in frames if f["type"]=="downloaded"]
+assert len(start)==1, f"expected one start_print, got {len(start)}"
+assert start[0]["file_name"]=="cube.3mf" and start[0]["printer_id"]=="fakep", start[0]
+assert len(dl)==1 and dl[0]["sha256"]==sys.argv[2] and dl[0]["bytes"]==200000, dl
+PYCHK
+
+kill "$GW_PID" 2>/dev/null || true
+
+echo
+if [ "$FAILURES" -eq 0 ]; then
+  echo "PASS - a file went from upload to a finished print, byte for byte, through every real component but the printer."
+else
+  echo "FAIL - $FAILURES assertion(s) failed." >&2
+  exit 1
+fi
diff --git a/scripts/fake-gateway-serve.ts b/scripts/fake-gateway-serve.ts
index 9252454..2bb30f4 100644
--- a/scripts/fake-gateway-serve.ts
+++ b/scripts/fake-gateway-serve.ts
@@ -6,6 +6,12 @@
  * (running, then failed a few seconds later), so the console has something to
  * alert on.
  *
+ * With REPORT_IDLE=<gateway-printer-id> it reports that printer idle on
+ * connect, so the scheduler treats it as free. A `start_print` command is
+ * accepted, the file is downloaded from its presigned URL and hashed (the
+ * hash is appended to the frames file, never the URL), and the printer then
+ * reports running and finished, like a real print that goes well.
+ *
  * Needs FARM_ID in the environment (the hello must name the token's farm).
  *
  * Usage: bun scripts/fake-gateway-serve.ts <ws-url> <edge-jwt> <seconds> <frames-file> [refuse-access-code]
@@ -31,10 +37,30 @@ ws.onerror = () => {
 ws.onopen = () => {
   // farm_id is required by the hello schema; the backend checks it against the token's farm.
   ws.send(JSON.stringify({ type: "hello", gateway_id: "fake-gw", farm_id: process.env.FARM_ID }));
+  if (process.env.REPORT_IDLE) {
+    ws.send(JSON.stringify({ type: "telemetry", printers: [{ printer_id: process.env.REPORT_IDLE, name: "Fake", state: "IDLE", percent: 0, print_error: 0 }] }));
+  }
   console.log("ready");
 };
+
+async function printFile(message: { command_id: string; printer_id: string; command: { job_id: string; file_name: string; url: string } }) {
+  const report = (state: string, percent: number) =>
+    ws.send(JSON.stringify({ type: "telemetry", printers: [{ printer_id: message.printer_id, name: "Fake", state, percent, print_error: 0 }] }));
+  const body = new Uint8Array(await (await fetch(message.command.url)).arrayBuffer());
+  const digest = new Bun.CryptoHasher("sha256").update(body).digest("hex");
+  appendFileSync(framesFile, JSON.stringify({ type: "downloaded", file_name: message.command.file_name, bytes: body.length, sha256: digest }) + "\n");
+  report("RUNNING", 20);
+  setTimeout(() => report("RUNNING", 60), 1500);
+  setTimeout(() => report("FINISH", 100), 3000);
+}
 ws.onmessage = (event) => {
   const message = JSON.parse(String(event.data));
+  if (message.type === "command" && message.command?.action === "start_print") {
+    appendFileSync(framesFile, JSON.stringify({ type: "start_print", printer_id: message.printer_id, file_name: message.command.file_name, job_id: message.command.job_id }) + "\n");
+    ws.send(JSON.stringify({ type: "command_result", command_id: message.command_id, ok: true }));
+    void printFile(message);
+    return;
+  }
   if (message.type !== "provision_printer" && message.type !== "remove_printer") return;
   appendFileSync(framesFile, JSON.stringify(message) + "\n");
   if (message.type === "provision_printer" && process.env.TELEMETRY_AFTER_PROVISION === "1") {