From 23a323849acf7e7d36e690ad8f0d8b6577bedabe Mon Sep 17 00:00:00 2001
From: Shreesh <shreesh.exe22@gmail.com>
Date: Mon, 2 Nov 2026 16:46:22 +0530
Subject: [PATCH 1/4] feat: richer results, stack/process/argument controls and
 a language availability check

---
 CHANGELOG.md                     |  16 +++++
 README.md                        |  16 +++++
 api/openapi.yaml                 |  49 +++++++++++++++
 cmd/stj-exec/main.go             |   1 +
 docker/go-judge/Dockerfile       |  11 +++-
 internal/gateway/handler.go      | 104 +++++++++++++++++++++++++++++++
 internal/gateway/handler_test.go |  87 ++++++++++++++++++++++++++
 internal/model/submission.go     |  16 +++--
 internal/queue/pubsub.go         |  27 ++++++++
 internal/queue/pubsub_test.go    |  36 ++++++++++-
 internal/sandbox/args.go         |  43 +++++++++++++
 internal/sandbox/args_test.go    |  48 ++++++++++++++
 internal/sandbox/client.go       |   1 +
 internal/sandbox/language.go     |  19 +++++-
 internal/sandbox/probe.go        |  67 ++++++++++++++++++++
 internal/sandbox/probe_test.go   |  67 ++++++++++++++++++++
 internal/worker/verdict_test.go  |  20 ++++++
 internal/worker/worker.go        |   9 +++
 18 files changed, 628 insertions(+), 9 deletions(-)
 create mode 100644 internal/sandbox/args.go
 create mode 100644 internal/sandbox/args_test.go
 create mode 100644 internal/sandbox/probe.go
 create mode 100644 internal/sandbox/probe_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9af9349..826db1c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,22 @@ All notable changes to the AetherCode Execution Engine are documented here.
 This project follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 and uses [semantic versioning](https://semver.org/).
 
+## [Unreleased]
+
+### Added
+
+- **Richer results** — each test in the `VERDICT` event, the webhook and the
+  stored results now carries `exit_status`, `sandbox_status` and `message`
+  beside the existing CPU, wall-clock (`run_time_ns`) and memory figures.
+- **Run controls** — `stack_limit_kb`, `max_processes` and `args` (command-line
+  arguments) on `POST /api/v1/execute`, validated at the gateway and stored with
+  the job.
+- **Language availability** — the engine checks at startup, and every five
+  minutes, that each registered language's compiler and runtime exist in the
+  sandbox. `GET /api/v1/languages` reports `available`, and `POST
+  /api/v1/execute` answers 400 for a language that is not installed instead of
+  failing the job later.
+
 ## [0.1.0] — 2026-10-04
 
 First open-source release.
diff --git a/README.md b/README.md
index 1baeeda..3af5ae5 100644
--- a/README.md
+++ b/README.md
@@ -314,6 +314,9 @@ Submit code for evaluation against test cases.
 | `priority` | string | No | Priority tier: `run` (P0 default), `submit` (P1), or `bulk` (P2) |
 | `time_limit_ms` | integer | No | Per-test CPU time ceiling in milliseconds (100 to 20000) |
 | `memory_limit_kb` | integer | No | Per-test memory ceiling in kilobytes (16384 to 2097152) |
+| `stack_limit_kb` | integer | No | Per-test stack size limit in kilobytes (1024 to 1048576) |
+| `max_processes` | integer | No | Per-test process/thread limit (1 to 512); the language default otherwise |
+| `args` | array | No | Command-line arguments passed to the program on every test, as separate argv entries, never through a shell (max 16, 4 KB in total; not for `sqlite`) |
 | `idempotency_key` | string | No | Client key; duplicate submissions return the original job identifier |
 
 **Successful Response (`201 Created`):**
@@ -394,6 +397,10 @@ data: {
         "verdict": "accepted",
         "cpu_time_ns": 23000000,
         "memory_bytes": 14221312,
+        "run_time_ns": 31000000,
+        "exit_status": 0,
+        "sandbox_status": "Accepted",
+        "message": "",
         "stdout_preview": "60\n",
         "stderr_preview": ""
       },
@@ -490,6 +497,15 @@ Cancels an active or queued submission. Marks the submission record as `failed`
 | `python` | CPython 3.12 | No | 2.0 s | 512 MB |
 | `javascript` | Node.js 24 LTS | No | 2.0 s | 512 MB |
 | `sqlite` | SQLite 3.45 (`sqlite3 -batch`) | No | 2.0 s | 512 MB |
+| `bash` | Bash | No | 2.0 s | 256 MB |
+| `rust` | rustc (`-O`) | Yes | 2.0 s | 512 MB |
+| `csharp` | Mono (`mcs`, `mono`) | Yes | 2.0 s | 512 MB (Min 128 MB) |
+| `php` | PHP CLI | No | 2.0 s | 512 MB |
+| `ruby` | Ruby | No | 2.0 s | 512 MB |
+| `kotlin` | `kotlinc` and OpenJDK 21 | Yes | 2.0 s | 512 MB (Min 256 MB) |
+| `typescript` | Bun | No | 2.0 s | 512 MB |
+
+The bundled sandbox image (`docker/go-judge`) installs every language above except `kotlin` and `typescript`, which have no Debian package. At startup, and every five minutes, the engine checks that each language's compiler and runtime exist in the sandbox: `GET /api/v1/languages` reports `available`, and `POST /api/v1/execute` answers `400` for a language that is not installed. Install `kotlinc` or `bun` in your own sandbox image to enable them.
 
 Compilation phase limits: C and C++ allow 10 s CPU and 256 MB RAM; Java allows 15 s CPU and 512 MB RAM; Go allows 30 s CPU and 1 GB RAM.
 
diff --git a/api/openapi.yaml b/api/openapi.yaml
index a3d3240..882af4d 100644
--- a/api/openapi.yaml
+++ b/api/openapi.yaml
@@ -175,6 +175,33 @@ components:
             Per-test memory limit override in kilobytes (16384–2097152 = 16 MB–2 GB).
             Omit to use the language default.
           example: 262144
+        stack_limit_kb:
+          type: integer
+          format: int64
+          minimum: 1024
+          maximum: 1048576
+          description: |
+            Per-test stack size limit in kilobytes (1 MB–1 GB). Omit to use the sandbox default.
+          example: 65536
+        max_processes:
+          type: integer
+          format: int64
+          minimum: 1
+          maximum: 512
+          description: |
+            Per-test process/thread limit. Omit to use the language default.
+          example: 4
+        args:
+          type: array
+          maxItems: 16
+          items:
+            type: string
+            maxLength: 1024
+          description: |
+            Command-line arguments passed to the program on every test, after the
+            program's own arguments, as separate argv entries (never through a shell).
+            Up to 16 arguments, 4 KB in total. Not supported for `sqlite`.
+          example: ["-n", "5"]
 
     ExecuteResponse:
       type: object
@@ -381,6 +408,13 @@ components:
           format: int64
           description: Maximum number of processes/threads the submission may spawn.
           example: 10
+        available:
+          type: boolean
+          description: |
+            False when the sandbox is missing one of the language's executables
+            (checked at startup and every five minutes); `POST /api/v1/execute`
+            then answers 400. True until the first check completes.
+          example: true
 
     # ── SSE event shapes ───────────────────────────────────────────────────────
 
@@ -554,6 +588,21 @@ components:
                   stderr:
                     type: string
                     description: Full stderr.
+                  run_time_ns:
+                    type: integer
+                    format: int64
+                    description: Wall-clock run time in nanoseconds.
+                  exit_status:
+                    type: integer
+                    description: The program's exit code, or the signal number when it was killed.
+                  sandbox_status:
+                    type: string
+                    description: The sandbox's status for the run, for example `Accepted`, `Nonzero Exit Status` or `Signalled`.
+                    example: Signalled
+                  message:
+                    type: string
+                    maxLength: 512
+                    description: The sandbox's explanation, for example the signal's name.
 
 paths:
 
diff --git a/cmd/stj-exec/main.go b/cmd/stj-exec/main.go
index c2b814b..996611f 100644
--- a/cmd/stj-exec/main.go
+++ b/cmd/stj-exec/main.go
@@ -156,6 +156,7 @@ func main() {
 	mux := http.NewServeMux()
 	h := gateway.NewTieredHandler(pool, submitStream, runStream, bulkStream, pubsub, judge, rdb, rl, unauthRL, cfg.AdmissionMaxQueueDepth, cfg.WorkerCount, trustedProxies)
 	h.Register(mux)
+	h.StartLanguageProbe(ctx, 5*time.Minute)
 
 	var handler http.Handler = mux
 
diff --git a/docker/go-judge/Dockerfile b/docker/go-judge/Dockerfile
index 98897e4..403d353 100644
--- a/docker/go-judge/Dockerfile
+++ b/docker/go-judge/Dockerfile
@@ -1,5 +1,8 @@
-# go-judge sandbox image with the 7 languages the engine supports:
-# C, C++, Java 21, Go, Python 3, JavaScript (Node), SQLite3.
+# go-judge sandbox image with the languages the engine runs:
+# C, C++, Java 21, Go, Python 3, JavaScript (Node), SQLite3, Bash, Rust, C#
+# (Mono), PHP and Ruby. Kotlin (kotlinc) and TypeScript (bun) have no Debian
+# package; until a pinned release is added here, the engine's startup probe
+# reports them as unavailable (GET /api/v1/languages) and rejects them.
 #
 # Base image pinned to v1.13.0 (the version the engine is validated against).
 # To update: docker pull criyle/go-judge:<new-tag> && docker inspect --format='{{.RepoDigests}}' criyle/go-judge:<new-tag>
@@ -13,6 +16,10 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
       openjdk-21-jdk-headless \
       golang \
       sqlite3 \
+      rustc \
+      mono-mcs mono-runtime \
+      php-cli \
+      ruby \
     && rm -rf /var/lib/apt/lists/*
 WORKDIR /opt
 COPY --from=go-judge /opt/go-judge /opt/go-judge
diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go
index 322e1c3..fed0898 100644
--- a/internal/gateway/handler.go
+++ b/internal/gateway/handler.go
@@ -16,6 +16,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
@@ -104,6 +105,10 @@ type Handler struct {
 	startTime              time.Time
 	workerCount            int
 	trustedProxies         []*net.IPNet
+	// available is the last language probe: language id to whether its
+	// executables exist in the sandbox. Nil until the first probe succeeds, in
+	// which case every language is accepted.
+	available atomic.Pointer[map[string]bool]
 }
 
 func NewHandler(pool *pgxpool.Pool, submitStream, runStream *queue.Stream, pubsub *queue.PubSub, judge *sandbox.Client, rdb *redis.Client, rateLimiter *ratelimit.Limiter, unauthLimiter *ratelimit.Limiter, admissionMaxQueueDepth int, workerCount int, trustedProxies []*net.IPNet) *Handler {
@@ -147,6 +152,11 @@ type executeRequest struct {
 	// Optional limits; omitted means the language defaults.
 	TimeLimitMs   *uint64 `json:"time_limit_ms"`
 	MemoryLimitKb *uint64 `json:"memory_limit_kb"`
+	// Optional run controls: stack size (KB), process/thread count, and
+	// command-line arguments passed to the program on every test.
+	StackLimitKb *uint64  `json:"stack_limit_kb"`
+	MaxProcesses *uint64  `json:"max_processes"`
+	Args         []string `json:"args,omitempty"`
 	// Optional output checker: exact (default), token, float, or custom.
 	Checker string `json:"checker"`
 	// Optional custom checker source code and language (e.g. "python", "cpp").
@@ -225,6 +235,10 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 		jsonError(w, "unsupported language", http.StatusBadRequest)
 		return
 	}
+	if !h.languageAvailable(req.Language) {
+		jsonError(w, "language "+req.Language+" is not installed on this engine", http.StatusBadRequest)
+		return
+	}
 	if len(req.SourceCode) == 0 {
 		jsonError(w, "source_code is required", http.StatusBadRequest)
 		return
@@ -291,6 +305,10 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 			jsonError(w, "unsupported checker_language", http.StatusBadRequest)
 			return
 		}
+		if !h.languageAvailable(req.CheckerLanguage) {
+			jsonError(w, "checker_language "+req.CheckerLanguage+" is not installed on this engine", http.StatusBadRequest)
+			return
+		}
 		req.Checker = "custom"
 	}
 	if req.InteractorCode != "" {
@@ -305,6 +323,10 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 			jsonError(w, "unsupported interactor_language", http.StatusBadRequest)
 			return
 		}
+		if !h.languageAvailable(req.InteractorLanguage) {
+			jsonError(w, "interactor_language "+req.InteractorLanguage+" is not installed on this engine", http.StatusBadRequest)
+			return
+		}
 	}
 	if len(req.Files) > 0 {
 		if err := sandbox.ValidateFiles(req.Files, 10, 128<<10); err != nil {
@@ -353,6 +375,30 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	if req.StackLimitKb != nil {
+		limits.Stack = *req.StackLimitKb * 1024
+		if limits.Stack < sandbox.MinRunStack || limits.Stack > sandbox.MaxRunStack {
+			jsonError(w, "stack_limit_kb must be between 1024 and 1048576", http.StatusBadRequest)
+			return
+		}
+	}
+	if req.MaxProcesses != nil {
+		limits.Procs = *req.MaxProcesses
+		if limits.Procs < sandbox.MinRunProcs || limits.Procs > sandbox.MaxRunProcs {
+			jsonError(w, "max_processes must be between 1 and 512", http.StatusBadRequest)
+			return
+		}
+	}
+	if err := sandbox.ValidateArgs(req.Args); err != nil {
+		jsonError(w, "args: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+	if len(req.Args) > 0 && sandbox.Languages[req.Language].StdinIsSource {
+		jsonError(w, "args are not supported for "+req.Language, http.StatusBadRequest)
+		return
+	}
+	limits.Args = req.Args
+
 	// Per-user rate limiting for authenticated callers.
 	if hasAuth {
 		if rlResult := h.rateLimiter.Check(ctx, identity.UserID); !rlResult.Allowed {
@@ -607,6 +653,9 @@ type languageInfo struct {
 	DefaultCPULimitMs    uint64 `json:"default_cpu_limit_ms"`
 	DefaultMemoryLimitKb uint64 `json:"default_memory_limit_kb"`
 	RunProcLimit         uint64 `json:"run_proc_limit"`
+	// Available is false when the sandbox is missing one of the language's
+	// executables; true while no probe has completed.
+	Available bool `json:"available"`
 }
 
 func (h *Handler) HandleLanguages(w http.ResponseWriter, r *http.Request) {
@@ -625,6 +674,7 @@ func (h *Handler) HandleLanguages(w http.ResponseWriter, r *http.Request) {
 			DefaultCPULimitMs:    l.RunCPU / 1_000_000,
 			DefaultMemoryLimitKb: l.RunMem / 1024,
 			RunProcLimit:         l.RunProc,
+			Available:            h.languageAvailable(id),
 		}
 	}
 
@@ -756,3 +806,57 @@ func (h *Handler) HandleGetSubmission(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(resp) //nolint:errcheck
 }
+
+// languageAvailable reports whether the last probe found the language's
+// executables in the sandbox. Before any probe it is true, so a slow or failing
+// probe never blocks submissions.
+func (h *Handler) languageAvailable(id string) bool {
+	probed := h.available.Load()
+	if probed == nil {
+		return true
+	}
+	ok, known := (*probed)[id]
+	return ok || !known
+}
+
+// ProbeLanguages asks the sandbox which languages are installed and remembers
+// the answer. A failed probe keeps the previous answer.
+func (h *Handler) ProbeLanguages(ctx context.Context) error {
+	probeCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	available, err := h.judge.ProbeLanguages(probeCtx)
+	if err != nil {
+		return err
+	}
+	h.available.Store(&available)
+	return nil
+}
+
+// StartLanguageProbe probes now and then every interval until ctx ends, logging
+// each language that is registered but missing from the sandbox.
+func (h *Handler) StartLanguageProbe(ctx context.Context, interval time.Duration) {
+	probe := func() {
+		if err := h.ProbeLanguages(ctx); err != nil {
+			slog.Warn("language probe failed; keeping the last answer", "err", err)
+			return
+		}
+		for id, ok := range *h.available.Load() {
+			if !ok {
+				slog.Warn("language is registered but not installed in the sandbox", "language", id)
+			}
+		}
+	}
+	probe()
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				probe()
+			}
+		}
+	}()
+}
diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go
index e72c398..db9f73c 100644
--- a/internal/gateway/handler_test.go
+++ b/internal/gateway/handler_test.go
@@ -629,3 +629,90 @@ func TestHandleExecute_MergesSubmissionEnvIntoTests(t *testing.T) {
 		}
 	}
 }
+
+func TestHandleExecute_RejectsInvalidRunControls(t *testing.T) {
+	judgeSrv := fakeJudgeServer(t, "1.13.0")
+	judge := sandbox.NewClient(judgeSrv.URL, "test-token")
+	_, rdb := newTestRedis(t)
+
+	submitStream := queue.NewStream(rdb, "ac:submissions:submit", "workers")
+	runStream := queue.NewStream(rdb, "ac:submissions:run", "workers")
+	h := NewHandler(nil, submitStream, runStream, queue.NewPubSub(rdb), judge, rdb, nil, nil, 5000, 8, nil)
+
+	cases := []struct{ name, body, msg string }{
+		{"stack too small",
+			`{"language":"c","source_code":"int main(){}","stack_limit_kb":10,"tests":[{"input":"","expected_output":""}]}`,
+			"stack_limit_kb must be between 1024 and 1048576"},
+		{"too many processes",
+			`{"language":"c","source_code":"int main(){}","max_processes":1000,"tests":[{"input":"","expected_output":""}]}`,
+			"max_processes must be between 1 and 512"},
+		{"zero processes",
+			`{"language":"c","source_code":"int main(){}","max_processes":0,"tests":[{"input":"","expected_output":""}]}`,
+			"max_processes must be between 1 and 512"},
+		{"too many args",
+			`{"language":"c","source_code":"int main(){}","args":["1","2","3","4","5","6","7","8","9","10","11","12","13","14","15","16","17"],"tests":[{"input":"","expected_output":""}]}`,
+			"args: at most 16 arguments"},
+		{"args for sqlite",
+			`{"language":"sqlite","source_code":"select 1;","args":["x"],"tests":[{"input":"","expected_output":"1"}]}`,
+			"args are not supported for sqlite"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assertExecuteError(t, postExecute(h, tc.body), http.StatusBadRequest, tc.msg)
+		})
+	}
+}
+
+func TestLanguageAvailability(t *testing.T) {
+	var runs int
+	judgeSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/run" {
+			http.NotFound(w, r)
+			return
+		}
+		runs++
+		var req sandbox.Request
+		_ = json.NewDecoder(r.Body).Decode(&req)
+		results := make([]sandbox.Result, len(req.Cmd))
+		for i, cmd := range req.Cmd {
+			results[i] = sandbox.Result{Status: sandbox.StatusAccepted}
+			if cmd.Args[2] == "/usr/bin/rustc" {
+				results[i] = sandbox.Result{Status: sandbox.StatusNonzeroExit, ExitStatus: 1}
+			}
+		}
+		_ = json.NewEncoder(w).Encode(results)
+	}))
+	t.Cleanup(judgeSrv.Close)
+	_, rdb := newTestRedis(t)
+	h := NewHandler(nil, queue.NewStream(rdb, "ac:submissions:submit", "workers"), queue.NewStream(rdb, "ac:submissions:run", "workers"), queue.NewPubSub(rdb), sandbox.NewClient(judgeSrv.URL, "t"), rdb, nil, nil, 5000, 8, nil)
+
+	if !h.languageAvailable("rust") {
+		t.Fatal("rust must be accepted before the first probe")
+	}
+	if err := h.ProbeLanguages(context.Background()); err != nil {
+		t.Fatalf("ProbeLanguages: %v", err)
+	}
+	if h.languageAvailable("rust") || !h.languageAvailable("python") {
+		t.Fatalf("rust = %v, python = %v, want rust unavailable and python available", h.languageAvailable("rust"), h.languageAvailable("python"))
+	}
+	assertExecuteError(t, postExecute(h, `{"language":"rust","source_code":"fn main(){}","tests":[{"input":"","expected_output":""}]}`),
+		http.StatusBadRequest, "language rust is not installed on this engine")
+
+	rec := httptest.NewRecorder()
+	h.HandleLanguages(rec, httptest.NewRequest(http.MethodGet, "/api/v1/languages", nil))
+	var listed []struct {
+		ID        string `json:"id"`
+		Available bool   `json:"available"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&listed); err != nil {
+		t.Fatal(err)
+	}
+	for _, l := range listed {
+		if l.ID == "rust" && l.Available {
+			t.Error("GET /languages reports rust available")
+		}
+		if l.ID == "python" && !l.Available {
+			t.Error("GET /languages reports python unavailable")
+		}
+	}
+}
diff --git a/internal/model/submission.go b/internal/model/submission.go
index a2a333d..f016d7c 100644
--- a/internal/model/submission.go
+++ b/internal/model/submission.go
@@ -52,11 +52,17 @@ type Submission struct {
 }
 
 type TestResult struct {
-	TestIndex     int
-	Verdict       Verdict
-	CPUTimeNs     int64
-	MemoryBytes   int64
-	RunTimeNs     int64
+	TestIndex   int
+	Verdict     Verdict
+	CPUTimeNs   int64
+	MemoryBytes int64
+	RunTimeNs   int64
+	// ExitStatus is the program's exit code (or the signal number when it was
+	// killed), SandboxStatus go-judge's status for the run, and Message its
+	// explanation (for example the signal's name).
+	ExitStatus    int
+	SandboxStatus string
+	Message       string
 	StdoutPreview string
 	StderrPreview string
 }
diff --git a/internal/queue/pubsub.go b/internal/queue/pubsub.go
index 6dc9f69..58b3300 100644
--- a/internal/queue/pubsub.go
+++ b/internal/queue/pubsub.go
@@ -100,6 +100,8 @@ func (p *PubSub) PrepareAndEnqueue(ctx context.Context, jobID string, set TestSe
 			"count", len(set.Tests),
 			"cpu_ns", set.Limits.CPU,
 			"mem_bytes", set.Limits.Memory,
+			"stack_bytes", set.Limits.Stack,
+			"procs", set.Limits.Procs,
 			"checker", set.Checker,
 			"checker_code", set.CheckerCode,
 			"checker_language", set.CheckerLanguage,
@@ -113,6 +115,13 @@ func (p *PubSub) PrepareAndEnqueue(ctx context.Context, jobID string, set TestSe
 				fields = append(fields, "files_json", string(filesJSON))
 			}
 		}
+		if len(set.Limits.Args) > 0 {
+			argsJSON, err := json.Marshal(set.Limits.Args)
+			if err != nil {
+				return fmt.Errorf("pubsub: marshal args: %w", err)
+			}
+			fields = append(fields, "args_json", string(argsJSON))
+		}
 		if len(set.CompilerFlags) > 0 {
 			if flagsJSON, err := json.Marshal(set.CompilerFlags); err == nil {
 				fields = append(fields, "compiler_flags_json", string(flagsJSON))
@@ -313,6 +322,24 @@ func (p *PubSub) GetTests(ctx context.Context, jobID string) (TestSet, error) {
 	if v := fields["mem_bytes"]; v != "" {
 		limits.Memory, _ = strconv.ParseUint(v, 10, 64)
 	}
+	// Absent on jobs enqueued before stack, process and argument controls existed.
+	if v := fields["stack_bytes"]; v != "" {
+		limits.Stack, _ = strconv.ParseUint(v, 10, 64)
+	}
+	if v := fields["procs"]; v != "" {
+		limits.Procs, _ = strconv.ParseUint(v, 10, 64)
+	}
+	if aj, ok := fields["args_json"]; ok {
+		var args []string
+		err := json.Unmarshal([]byte(aj), &args)
+		if err == nil {
+			err = sandbox.ValidateArgs(args)
+		}
+		if err != nil {
+			return TestSet{}, fmt.Errorf("pubsub: corrupt args in %s: %w", key, err)
+		}
+		limits.Args = args
+	}
 	var files map[string]string
 	if fj := fields["files_json"]; fj != "" {
 		_ = json.Unmarshal([]byte(fj), &files)
diff --git a/internal/queue/pubsub_test.go b/internal/queue/pubsub_test.go
index 55d81d0..61fb633 100644
--- a/internal/queue/pubsub_test.go
+++ b/internal/queue/pubsub_test.go
@@ -358,7 +358,7 @@ func TestPrepareAndGetTests_RoundTripsChecker(t *testing.T) {
 	if got.Checker != "float" {
 		t.Errorf("Checker = %q, want float", got.Checker)
 	}
-	if got.Limits != set.Limits {
+	if !reflect.DeepEqual(got.Limits, set.Limits) {
 		t.Errorf("Limits = %+v, want %+v", got.Limits, set.Limits)
 	}
 	if !reflect.DeepEqual(got.Tests, set.Tests) {
@@ -564,3 +564,37 @@ func TestCallbackSecret_EncryptedWithoutKey(t *testing.T) {
 		t.Fatal("expected error reading encrypted secret without key, got nil")
 	}
 }
+
+func TestPrepareAndGetTests_RoundTripsRunControls(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx := context.Background()
+	set := TestSet{
+		Tests:  []TestData{{Input: "1", ExpectedOutput: "1"}},
+		Limits: sandbox.RunLimits{CPU: 1_000_000_000, Memory: 64 << 20, Stack: 8 << 20, Procs: 4, Args: []string{"-n", "5"}},
+	}
+	evt := eventlog.New("job-controls", eventlog.EventQueued, nil)
+	if err := ps.PrepareAndEnqueue(ctx, "job-controls", set, evt, "ac:submissions:run"); err != nil {
+		t.Fatalf("PrepareAndEnqueue: %v", err)
+	}
+	got, err := ps.GetTests(ctx, "job-controls")
+	if err != nil {
+		t.Fatalf("GetTests: %v", err)
+	}
+	if !reflect.DeepEqual(got.Limits, set.Limits) {
+		t.Errorf("Limits = %+v, want %+v", got.Limits, set.Limits)
+	}
+}
+
+func TestGetTests_RejectsCorruptArgs(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx := context.Background()
+	for name, value := range map[string]string{"not json": "nope", "null bytes": `["a\u0000b"]`} {
+		key := "ac:tests:job-" + name
+		rdb.HSet(ctx, key, "count", 1, "0:input", "", "0:expected", "", "args_json", value)
+		if _, err := ps.GetTests(ctx, "job-"+name); err == nil {
+			t.Errorf("%s: GetTests accepted corrupt args", name)
+		}
+	}
+}
diff --git a/internal/sandbox/args.go b/internal/sandbox/args.go
new file mode 100644
index 0000000..6b8972e
--- /dev/null
+++ b/internal/sandbox/args.go
@@ -0,0 +1,43 @@
+package sandbox
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Caller-supplied command-line argument limits. The arguments are passed to
+// the program as separate argv entries, never through a shell.
+const (
+	MaxArgs         = 16
+	MaxArgBytes     = 1 << 10
+	MaxArgsTotalLen = 4 << 10
+)
+
+// ValidateArgs checks the command-line arguments of a run request.
+func ValidateArgs(args []string) error {
+	if len(args) > MaxArgs {
+		return fmt.Errorf("at most %d arguments", MaxArgs)
+	}
+	total := 0
+	for i, arg := range args {
+		if len(arg) > MaxArgBytes {
+			return fmt.Errorf("argument %d is longer than %d bytes", i, MaxArgBytes)
+		}
+		if strings.ContainsRune(arg, 0) {
+			return fmt.Errorf("argument %d contains a NUL byte", i)
+		}
+		total += len(arg)
+	}
+	if total > MaxArgsTotalLen {
+		return fmt.Errorf("arguments total more than %d bytes", MaxArgsTotalLen)
+	}
+	return nil
+}
+
+// runArgs is the program's own argv followed by the caller's arguments. It
+// always returns a fresh slice, so the shared language config is never mutated.
+func runArgs(base, extra []string) []string {
+	out := make([]string, 0, len(base)+len(extra))
+	out = append(out, base...)
+	return append(out, extra...)
+}
diff --git a/internal/sandbox/args_test.go b/internal/sandbox/args_test.go
new file mode 100644
index 0000000..4782c7a
--- /dev/null
+++ b/internal/sandbox/args_test.go
@@ -0,0 +1,48 @@
+package sandbox
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidateArgs(t *testing.T) {
+	t.Parallel()
+	tooMany := make([]string, MaxArgs+1)
+	cases := map[string]struct {
+		args []string
+		ok   bool
+	}{
+		"none":            {nil, true},
+		"plain":           {[]string{"-n", "5", "file name"}, true},
+		"too many":        {tooMany, false},
+		"one too long":    {[]string{strings.Repeat("a", MaxArgBytes+1)}, false},
+		"nul byte":        {[]string{"a\x00b"}, false},
+		"total too large": {[]string{strings.Repeat("a", MaxArgBytes), strings.Repeat("a", MaxArgBytes), strings.Repeat("a", MaxArgBytes), strings.Repeat("a", MaxArgBytes), "a"}, false},
+	}
+	for name, tc := range cases {
+		if err := ValidateArgs(tc.args); (err == nil) != tc.ok {
+			t.Errorf("%s: err = %v, want ok = %v", name, err, tc.ok)
+		}
+	}
+}
+
+func TestRunCmdCarriesArgsStackAndProcs(t *testing.T) {
+	t.Parallel()
+	lang := Languages["python"]
+	before := append([]string(nil), lang.RunArgs...)
+	cmd := BuildRunCmd(lang, "print(1)", "", "", RunLimits{Args: []string{"a", "b"}, Stack: 8 << 20, Procs: 4})
+	want := append(append([]string(nil), before...), "a", "b")
+	if strings.Join(cmd.Args, " ") != strings.Join(want, " ") {
+		t.Fatalf("args = %v, want %v", cmd.Args, want)
+	}
+	if cmd.StackLimit != 8<<20 || cmd.ProcLimit != 4 {
+		t.Fatalf("stack = %d, procs = %d, want 8 MiB and 4", cmd.StackLimit, cmd.ProcLimit)
+	}
+	if strings.Join(Languages["python"].RunArgs, " ") != strings.Join(before, " ") {
+		t.Fatal("the language's own run args were changed")
+	}
+	plain := BuildRunCmd(lang, "print(1)", "", "", RunLimits{})
+	if plain.ProcLimit != lang.RunProc || plain.StackLimit != 0 {
+		t.Fatalf("defaults changed: procs = %d, stack = %d", plain.ProcLimit, plain.StackLimit)
+	}
+}
diff --git a/internal/sandbox/client.go b/internal/sandbox/client.go
index c761a6d..bde6a66 100644
--- a/internal/sandbox/client.go
+++ b/internal/sandbox/client.go
@@ -165,6 +165,7 @@ type Cmd struct {
 	ClockLimit    uint64            `json:"clockLimit,omitempty"`
 	MemoryLimit   uint64            `json:"memoryLimit"`
 	ProcLimit     uint64            `json:"procLimit"`
+	StackLimit    uint64            `json:"stackLimit,omitempty"`
 	CopyIn        map[string]CopyIn `json:"copyIn,omitempty"`
 	CopyOut       []string          `json:"copyOut,omitempty"`
 	CopyOutCached []string          `json:"copyOutCached,omitempty"`
diff --git a/internal/sandbox/language.go b/internal/sandbox/language.go
index 93688ec..e963925 100644
--- a/internal/sandbox/language.go
+++ b/internal/sandbox/language.go
@@ -19,6 +19,11 @@ const (
 	MaxRunCPU = 20_000_000_000 // 20s
 	MinRunMem = 16 << 20       // 16MB
 	MaxRunMem = 2 << 30        // 2GB
+	// Bounds for the optional stack and process limits.
+	MinRunStack = 1 << 20 // 1MB
+	MaxRunStack = 1 << 30 // 1GB
+	MinRunProcs = 1
+	MaxRunProcs = 512
 )
 
 // RunLimits overrides a language's default CPU time (ns) and memory (bytes)
@@ -26,6 +31,14 @@ const (
 type RunLimits struct {
 	CPU    uint64
 	Memory uint64
+	// Stack is the stack size limit in bytes; zero keeps the sandbox default.
+	Stack uint64
+	// Procs is the process/thread limit; zero keeps the language default.
+	Procs uint64
+	// Args are command-line arguments passed to the program after its own
+	// arguments (validated by ValidateArgs). They travel with the limits
+	// because both are per-job run settings.
+	Args []string
 }
 
 // Effective returns the CPU and memory limits a run uses. Memory never drops
@@ -311,15 +324,19 @@ func BuildRunCmd(lang LangConfig, sourceCode, stdinData, fileID string, limits R
 func BuildRunCmdWithOptions(lang LangConfig, sourceCode, stdinData, fileID string, limits RunLimits, extraFiles map[string]string, env []string) Cmd {
 	cpu, memory := lang.Effective(limits)
 	cmd := Cmd{
-		Args:        lang.RunArgs,
+		Args:        runArgs(lang.RunArgs, limits.Args),
 		Env:         runEnv(lang.RunEnv, env),
 		Files:       []CmdFile{Stdin(stdinData), Collector("stdout", StdoutMax), Collector("stderr", StderrMax)},
 		CPULimit:    cpu,
 		ClockLimit:  ClockLimit(cpu),
 		MemoryLimit: memory,
 		ProcLimit:   lang.RunProc,
+		StackLimit:  limits.Stack,
 		CopyOut:     []string{"stdout", "stderr"},
 	}
+	if limits.Procs != 0 {
+		cmd.ProcLimit = limits.Procs
+	}
 
 	copyIn := make(map[string]CopyIn)
 	if fileID != "" {
diff --git a/internal/sandbox/probe.go b/internal/sandbox/probe.go
new file mode 100644
index 0000000..1381777
--- /dev/null
+++ b/internal/sandbox/probe.go
@@ -0,0 +1,67 @@
+package sandbox
+
+import (
+	"context"
+	"fmt"
+	"slices"
+	"sort"
+)
+
+// Binaries lists the executables a language needs inside the sandbox: its
+// compiler, if any, and the program that runs it.
+func (lang LangConfig) Binaries() []string {
+	var out []string
+	if lang.Compiled && len(lang.CompileArgs) > 0 {
+		out = append(out, lang.CompileArgs[0])
+	}
+	if len(lang.RunArgs) > 0 && lang.RunArgs[0] != lang.BinaryPath {
+		out = append(out, lang.RunArgs[0])
+	}
+	return out
+}
+
+// ProbeLanguages reports, per registered language, whether every executable it
+// needs exists in the sandbox. One request checks each distinct executable once.
+// An error means the sandbox could not be asked, not that a language is missing.
+func (c *Client) ProbeLanguages(ctx context.Context) (map[string]bool, error) {
+	seen := map[string]bool{}
+	for _, lang := range Languages {
+		for _, binary := range lang.Binaries() {
+			seen[binary] = true
+		}
+	}
+	binaries := make([]string, 0, len(seen))
+	for binary := range seen {
+		binaries = append(binaries, binary)
+	}
+	sort.Strings(binaries)
+
+	cmds := make([]Cmd, len(binaries))
+	for i, binary := range binaries {
+		cmds[i] = Cmd{
+			Args:        []string{"/usr/bin/test", "-x", binary},
+			Env:         []string{"PATH=/usr/bin:/bin"},
+			Files:       []CmdFile{Stdin(""), Collector("stdout", 0), Collector("stderr", 0)},
+			CPULimit:    1_000_000_000,
+			ClockLimit:  2_000_000_000,
+			MemoryLimit: 16 << 20,
+			ProcLimit:   1,
+		}
+	}
+	results, err := c.Run(ctx, Request{Cmd: cmds})
+	if err != nil {
+		return nil, fmt.Errorf("sandbox: probe languages: %w", err)
+	}
+	if len(results) != len(binaries) {
+		return nil, fmt.Errorf("sandbox: probe languages: got %d results for %d checks", len(results), len(binaries))
+	}
+	present := make(map[string]bool, len(binaries))
+	for i, binary := range binaries {
+		present[binary] = results[i].Status == StatusAccepted
+	}
+	available := make(map[string]bool, len(Languages))
+	for id, lang := range Languages {
+		available[id] = !slices.ContainsFunc(lang.Binaries(), func(binary string) bool { return !present[binary] })
+	}
+	return available, nil
+}
diff --git a/internal/sandbox/probe_test.go b/internal/sandbox/probe_test.go
new file mode 100644
index 0000000..f623fff
--- /dev/null
+++ b/internal/sandbox/probe_test.go
@@ -0,0 +1,67 @@
+package sandbox
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestLangBinaries(t *testing.T) {
+	t.Parallel()
+	if got := Languages["c"].Binaries(); len(got) != 1 || got[0] != "/usr/bin/gcc" {
+		t.Errorf("c binaries = %v, want only gcc: the compiled program is not a toolchain binary", got)
+	}
+	if got := Languages["java"].Binaries(); len(got) != 2 || got[0] != "/usr/bin/javac" || got[1] != "/usr/bin/java" {
+		t.Errorf("java binaries = %v, want javac and java", got)
+	}
+	if got := Languages["python"].Binaries(); len(got) != 1 || got[0] != "/usr/bin/python3" {
+		t.Errorf("python binaries = %v, want python3", got)
+	}
+}
+
+func TestProbeLanguages(t *testing.T) {
+	t.Parallel()
+	missing := map[string]bool{"/usr/bin/rustc": true, "/usr/bin/php": true}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req Request
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			t.Errorf("decode: %v", err)
+		}
+		results := make([]Result, len(req.Cmd))
+		for i, cmd := range req.Cmd {
+			results[i] = Result{Status: StatusAccepted}
+			if missing[cmd.Args[2]] {
+				results[i] = Result{Status: StatusNonzeroExit, ExitStatus: 1}
+			}
+		}
+		_ = json.NewEncoder(w).Encode(results)
+	}))
+	defer server.Close()
+
+	available, err := NewClient(server.URL, "t").ProbeLanguages(context.Background())
+	if err != nil {
+		t.Fatalf("ProbeLanguages: %v", err)
+	}
+	if available["rust"] || available["php"] {
+		t.Errorf("rust = %v, php = %v, want both unavailable", available["rust"], available["php"])
+	}
+	if !available["c"] || !available["java"] || !available["python"] {
+		t.Errorf("c, java and python = %v %v %v, want all available", available["c"], available["java"], available["python"])
+	}
+	if len(available) != len(Languages) {
+		t.Errorf("len = %d, want an entry for each of the %d languages", len(available), len(Languages))
+	}
+}
+
+func TestProbeLanguagesReportsASandboxFailure(t *testing.T) {
+	t.Parallel()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "down", http.StatusBadGateway)
+	}))
+	defer server.Close()
+	if _, err := NewClient(server.URL, "t").ProbeLanguages(context.Background()); err == nil {
+		t.Fatal("ProbeLanguages returned no error for a failing sandbox")
+	}
+}
diff --git a/internal/worker/verdict_test.go b/internal/worker/verdict_test.go
index f306b7a..bbd2458 100644
--- a/internal/worker/verdict_test.go
+++ b/internal/worker/verdict_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/model"
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/sandbox"
 )
 
 func TestVerdictResults_AlwaysIncludesFullOutput(t *testing.T) {
@@ -32,3 +33,22 @@ func TestVerdictResults_AlwaysIncludesFullOutput(t *testing.T) {
 		t.Errorf("stderr_preview = %q, want warning", s)
 	}
 }
+
+func TestJudgeResult_ReportsExitStatusAndSandboxMessage(t *testing.T) {
+	r := sandbox.Result{Status: sandbox.StatusSignalled, ExitStatus: 11, Error: "signal: segmentation fault", Time: 5_000_000, Memory: 4096, RunTime: 9_000_000, Files: map[string]string{"stdout": "", "stderr": "boom"}}
+	tr := judgeResult(r, 3, "", "exact")
+	if tr.Verdict != model.VerdictRuntimeError || tr.ExitStatus != 11 || tr.SandboxStatus != sandbox.StatusSignalled || tr.Message != "signal: segmentation fault" {
+		t.Fatalf("result = %+v, want a runtime error carrying the exit status, sandbox status and message", tr)
+	}
+	e := verdictResults([]model.TestResult{tr})[0]
+	if e["exit_status"] != 11 || e["sandbox_status"] != sandbox.StatusSignalled || e["message"] != "signal: segmentation fault" || e["run_time_ns"] != int64(9_000_000) {
+		t.Fatalf("event entry = %+v, want exit_status, sandbox_status, message and run_time_ns", e)
+	}
+}
+
+func TestVerdictResults_TruncatesTheSandboxMessage(t *testing.T) {
+	e := verdictResults([]model.TestResult{{Message: strings.Repeat("m", 2000)}})[0]
+	if s, _ := e["message"].(string); len(s) != 512 {
+		t.Fatalf("message length = %d, want 512", len(s))
+	}
+}
diff --git a/internal/worker/worker.go b/internal/worker/worker.go
index 91b0000..f9a115e 100644
--- a/internal/worker/worker.go
+++ b/internal/worker/worker.go
@@ -354,6 +354,9 @@ func verdictResults(results []model.TestResult) []map[string]any {
 			"cpu_time_ns":    r.CPUTimeNs,
 			"memory_bytes":   r.MemoryBytes,
 			"run_time_ns":    r.RunTimeNs,
+			"exit_status":    r.ExitStatus,
+			"sandbox_status": r.SandboxStatus,
+			"message":        db.Truncate(r.Message, 512),
 			"stdout_preview": db.Truncate(r.StdoutPreview, 256),
 			"stderr_preview": db.Truncate(r.StderrPreview, 256),
 			"stdout":         r.StdoutPreview,
@@ -475,6 +478,9 @@ func (w *Worker) runTestsBatched(ctx context.Context, jobID string, lang sandbox
 					CPUTimeNs:     int64(studentRes.Time),    //nolint:gosec // go-judge reports CPU time in ns
 					MemoryBytes:   int64(studentRes.Memory),  //nolint:gosec // go-judge reports memory in bytes
 					RunTimeNs:     int64(studentRes.RunTime), //nolint:gosec // go-judge reports wall-clock run time in ns
+					ExitStatus:    studentRes.ExitStatus,
+					SandboxStatus: studentRes.Status,
+					Message:       studentRes.Error,
 					StdoutPreview: studentRes.Files["stdout"],
 					StderrPreview: studentRes.Files["stderr"],
 				}
@@ -571,6 +577,9 @@ func judgeResult(r sandbox.Result, testIndex int, expectedOutput, checkerType st
 		CPUTimeNs:     int64(r.Time),    //nolint:gosec // go-judge reports CPU time in ns (non-negative)
 		MemoryBytes:   int64(r.Memory),  //nolint:gosec // go-judge reports memory usage in bytes (non-negative)
 		RunTimeNs:     int64(r.RunTime), //nolint:gosec // go-judge reports wall-clock run time in ns
+		ExitStatus:    r.ExitStatus,
+		SandboxStatus: r.Status,
+		Message:       r.Error,
 		StdoutPreview: r.Files["stdout"],
 		StderrPreview: r.Files["stderr"],
 	}

From a2302b9417295e108aafbe9860eeb87d15164102 Mon Sep 17 00:00:00 2001
From: Shreesh <shreesh.exe22@gmail.com>
Date: Mon, 2 Nov 2026 17:03:10 +0530
Subject: [PATCH 2/4] perf: stage timings, one shared event subscription,
 overlapped claim and test load, honest Retry-After, leaner JVM

---
 CHANGELOG.md                     |   8 ++
 README.md                        |  25 +++++
 go.mod                           |   1 +
 internal/gateway/handler.go      |  16 +++-
 internal/gateway/handler_test.go |  29 ++++++
 internal/metrics/drain.go        |  74 +++++++++++++++
 internal/metrics/drain_test.go   |  37 ++++++++
 internal/metrics/metrics.go      |  17 ++++
 internal/queue/hub.go            | 156 +++++++++++++++++++++++++++++++
 internal/queue/hub_test.go       | 124 ++++++++++++++++++++++++
 internal/queue/pubsub.go         |  60 ++++++------
 internal/queue/stream.go         |  21 +++++
 internal/queue/wait_test.go      |  27 ++++++
 internal/sandbox/args_test.go    |  20 ++++
 internal/sandbox/language.go     |   6 +-
 internal/worker/worker.go        |  30 +++++-
 16 files changed, 615 insertions(+), 36 deletions(-)
 create mode 100644 internal/metrics/drain.go
 create mode 100644 internal/metrics/drain_test.go
 create mode 100644 internal/queue/hub.go
 create mode 100644 internal/queue/hub_test.go
 create mode 100644 internal/queue/wait_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 826db1c..b66b98e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,14 @@ and uses [semantic versioning](https://semver.org/).
   /api/v1/execute` answers 400 for a language that is not installed instead of
   failing the job later.
 
+### Changed
+
+- **Where the time goes** — new histograms `ac_queue_wait_seconds{stream}` (time between enqueue and a worker reading the job) and `ac_job_stage_seconds{stage}` (`claim`, `load_tests`, `run`, `persist`, `publish`), beside the existing `ac_compile_duration_seconds`.
+- **Lower latency per job** — a worker now fetches a job's tests from Redis while it claims the job in PostgreSQL, instead of one after the other.
+- **One Redis subscription per gateway** — SSE clients share a single pattern subscription instead of one subscription each. A client that falls behind is closed so it reconnects and replays from the event log; no event is dropped silently.
+- **Honest `Retry-After`** — when the queue is full, `429` carries the time the queue needs to fall under the admission limit at the measured completion rate (1 to 60 s), not a fixed 10.
+- **Leaner JVM** — `java` and `kotlin` start with the serial collector and shared class data, which also lowers the thread count of each run.
+
 ## [0.1.0] — 2026-10-04
 
 First open-source release.
diff --git a/README.md b/README.md
index 3af5ae5..bb85f6d 100644
--- a/README.md
+++ b/README.md
@@ -599,6 +599,31 @@ feature/* / fix/*  ──>  development (CI Gate)  ──>  production (Protecte
 
 ---
 
+## Performance Tuning
+
+**Find the stage first.** A slow p99 is usually queueing, not execution. Per job, the engine records where the time went:
+
+| Metric | Meaning |
+|---|---|
+| `ac_queue_wait_seconds{stream}` | Enqueue to a worker reading the job |
+| `ac_job_stage_seconds{stage}` | `claim`, `load_tests`, `run`, `persist`, `publish` |
+| `ac_compile_duration_seconds{language}` | The compile step, per language |
+| `ac_job_processing_duration_seconds{language,verdict}` | Dequeue to verdict |
+
+```promql
+histogram_quantile(0.99, sum by (le) (rate(ac_queue_wait_seconds_bucket[5m])))
+histogram_quantile(0.99, sum by (le, stage) (rate(ac_job_stage_seconds_bucket[5m])))
+```
+
+If queue wait dominates, add capacity; if `run` dominates, look at the language and the limits; if `persist` or `publish` do, look at PostgreSQL and Redis.
+
+**Sizing.**
+- Keep `AC_WORKER_COUNT` equal to go-judge's `-parallelism`; more workers than sandbox slots only queue inside go-judge.
+- On a multi-socket host, run one engine plus one go-judge per NUMA node (for example `numactl --cpunodebind=0 --membind=0` for each pair, each with its own listen address and `-parallelism` set to that node's physical cores) rather than one pair on one node. The published benchmark used a single node of a 256-thread host.
+- Raise go-judge's `-pre-fork` toward its parallelism so a burst does not wait for sandbox creation.
+
+**Overload.** `AC_ADMISSION_MAX_QUEUE_DEPTH` bounds how long a job can wait. A job admitted at the limit waits about depth ÷ completion rate, so lower the limit if that exceeds your verdict SLO. Rejected callers receive `Retry-After` computed from the measured completion rate.
+
 ## Contributing
 
 Contributions are welcome. Please consult:
diff --git a/go.mod b/go.mod
index c10a745..923c864 100644
--- a/go.mod
+++ b/go.mod
@@ -39,6 +39,7 @@ require (
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.3 // indirect
 	github.com/klauspost/compress v1.20.3 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20260330125221-c963978e514e // indirect
 	github.com/magiconair/properties v1.8.10 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go
index fed0898..19bb671 100644
--- a/internal/gateway/handler.go
+++ b/internal/gateway/handler.go
@@ -439,7 +439,7 @@ func (h *Handler) HandleExecute(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		slog.Error("admission check failed, allowing", "err", err)
 	} else if !admitted {
-		w.Header().Set("Retry-After", "10")
+		w.Header().Set("Retry-After", strconv.Itoa(h.busyRetryAfter(ctx, targetStream.StreamName())))
 		jsonError(w, "server is busy, try again later", http.StatusTooManyRequests)
 		return
 	}
@@ -860,3 +860,17 @@ func (h *Handler) StartLanguageProbe(ctx context.Context, interval time.Duration
 		}
 	}()
 }
+
+// busyRetryAfter is how long a rejected caller should wait: the time for the
+// queue to fall back under the admission limit at the rate jobs are finishing,
+// between 1 and 60 seconds. Without a measured rate it is the old fixed 10.
+func (h *Handler) busyRetryAfter(ctx context.Context, stream string) int {
+	if metrics.JobsDrain.Rate() <= 0 {
+		return 10
+	}
+	depth, err := h.rdb.XLen(ctx, stream).Result()
+	if err != nil {
+		return 10
+	}
+	return metrics.JobsDrain.RetryAfter(depth-int64(h.admissionMaxQueueDepth)+1, 1, 60)
+}
diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go
index db9f73c..5d04b1b 100644
--- a/internal/gateway/handler_test.go
+++ b/internal/gateway/handler_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	goredis "github.com/redis/go-redis/v9"
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
 
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/db"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
@@ -716,3 +717,31 @@ func TestLanguageAvailability(t *testing.T) {
 		}
 	}
 }
+
+func TestBusyRetryAfterFollowsTheDrainRate(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	h := NewHandler(nil, queue.NewStream(rdb, "ac:submissions:submit", "workers"), queue.NewStream(rdb, "ac:submissions:run", "workers"), nil, nil, rdb, nil, nil, 10, 8, nil)
+	ctx := context.Background()
+	for i := 0; i < 70; i++ {
+		rdb.XAdd(ctx, &goredis.XAddArgs{Stream: "ac:submissions:run", Values: map[string]any{"submission_id": fmt.Sprint(i)}})
+	}
+
+	original := metrics.JobsDrain
+	t.Cleanup(func() { metrics.JobsDrain = original })
+
+	now := time.Unix(5000, 0)
+	metrics.JobsDrain = metrics.NewDrain(func() time.Time { return now })
+	if got := h.busyRetryAfter(ctx, "ac:submissions:run"); got != 10 {
+		t.Fatalf("Retry-After with no measured rate = %d, want the old fixed 10", got)
+	}
+
+	for i := 0; i < 201; i++ {
+		metrics.JobsDrain.Complete()
+	}
+	now = now.Add(time.Second) // the rate settles at about 60 jobs/s (alpha 0.3)
+	got := h.busyRetryAfter(ctx, "ac:submissions:run")
+	// 70 queued, limit 10: it must fall by 61 jobs, about a second at 60/s.
+	if got < 1 || got > 3 {
+		t.Fatalf("Retry-After = %d, want 1 to 3 seconds for a 61-job excess at about 60 jobs/s", got)
+	}
+}
diff --git a/internal/metrics/drain.go b/internal/metrics/drain.go
new file mode 100644
index 0000000..8947bfc
--- /dev/null
+++ b/internal/metrics/drain.go
@@ -0,0 +1,74 @@
+package metrics
+
+import (
+	"math"
+	"sync"
+	"time"
+)
+
+// Drain measures how fast jobs finish, as a smoothed rate per second. The
+// gateway turns it into an honest Retry-After when it sheds load.
+type Drain struct {
+	mu      sync.Mutex
+	rate    float64 // smoothed completions per second
+	window  time.Time
+	count   int
+	now     func() time.Time
+	started bool
+}
+
+const (
+	drainWindow = time.Second
+	// drainAlpha weights the newest second against the history.
+	drainAlpha = 0.3
+)
+
+// NewDrain returns a tracker that reads time from now.
+func NewDrain(now func() time.Time) *Drain {
+	return &Drain{now: now}
+}
+
+// Complete records one finished job.
+func (d *Drain) Complete() {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.advance()
+	d.count++
+}
+
+// Rate is the smoothed completions per second; zero until a full second passed.
+func (d *Drain) Rate() float64 {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.advance()
+	return d.rate
+}
+
+// RetryAfter is how many seconds a caller should wait for depth queued jobs to
+// drain, between lo and hi. With no measured rate it returns lo.
+func (d *Drain) RetryAfter(depth int64, lo, hi int) int {
+	rate := d.Rate()
+	if rate <= 0 || depth <= 0 {
+		return lo
+	}
+	seconds := int(math.Ceil(float64(depth) / rate))
+	return max(lo, min(hi, seconds))
+}
+
+// advance folds every full elapsed window into the smoothed rate. Idle windows
+// decay it toward zero.
+func (d *Drain) advance() {
+	now := d.now()
+	if !d.started {
+		d.started, d.window = true, now
+		return
+	}
+	for now.Sub(d.window) >= drainWindow {
+		d.rate = drainAlpha*float64(d.count) + (1-drainAlpha)*d.rate
+		d.count = 0
+		d.window = d.window.Add(drainWindow)
+	}
+}
+
+// JobsDrain is the process-wide tracker: workers feed it, the gateway reads it.
+var JobsDrain = NewDrain(time.Now)
diff --git a/internal/metrics/drain_test.go b/internal/metrics/drain_test.go
new file mode 100644
index 0000000..61f51ed
--- /dev/null
+++ b/internal/metrics/drain_test.go
@@ -0,0 +1,37 @@
+package metrics
+
+import (
+	"testing"
+	"time"
+)
+
+func TestDrainRateAndRetryAfter(t *testing.T) {
+	now := time.Unix(1000, 0)
+	d := NewDrain(func() time.Time { return now })
+
+	if got := d.RetryAfter(500, 1, 60); got != 1 {
+		t.Fatalf("RetryAfter with no measurements = %d, want the lower bound", got)
+	}
+	d.Complete() // starts the first window
+	for i := 0; i < 100; i++ {
+		d.Complete()
+	}
+	now = now.Add(time.Second)
+	rate := d.Rate()
+	if rate < 25 || rate > 35 {
+		t.Fatalf("rate after one 101-job second = %.1f, want about 30 (alpha 0.3)", rate)
+	}
+	if got := d.RetryAfter(600, 1, 60); got < 15 || got > 25 {
+		t.Fatalf("RetryAfter(600) = %d, want about 20 at 30 jobs/s", got)
+	}
+	if got := d.RetryAfter(1_000_000, 1, 60); got != 60 {
+		t.Fatalf("RetryAfter above the cap = %d, want 60", got)
+	}
+	if got := d.RetryAfter(0, 2, 60); got != 2 {
+		t.Fatalf("RetryAfter on an empty queue = %d, want the lower bound", got)
+	}
+	now = now.Add(30 * time.Second)
+	if d.Rate() > 0.1 {
+		t.Fatalf("rate after 30 idle seconds = %.3f, want it to have decayed", d.Rate())
+	}
+}
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index f8300b0..b06ab29 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -74,4 +74,21 @@ var (
 		Name:      "compile_cache_misses_total",
 		Help:      "Compile cache misses — source was not in cache, compilation required.",
 	})
+
+	QueueWaitSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ac",
+		Name:      "queue_wait_seconds",
+		Help:      "Time a job waited in its stream between being enqueued and being read by a worker.",
+		Buckets:   []float64{0.005, 0.025, 0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120},
+	}, []string{"stream"})
+
+	// JobStageSeconds splits a job's time into its stages so a slow p99 can be
+	// attributed: claim, load_tests, run, persist and publish. Compile has its
+	// own histogram (CompileDuration) because it is labelled by language.
+	JobStageSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "ac",
+		Name:      "job_stage_seconds",
+		Help:      "Time spent in each stage of processing a job.",
+		Buckets:   []float64{0.001, 0.005, 0.025, 0.1, 0.5, 1, 2.5, 5, 10, 30, 60},
+	}, []string{"stage"})
 )
diff --git a/internal/queue/hub.go b/internal/queue/hub.go
new file mode 100644
index 0000000..f40968f
--- /dev/null
+++ b/internal/queue/hub.go
@@ -0,0 +1,156 @@
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+	"sync"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
+)
+
+const (
+	eventChannelPrefix = "ac:events:"
+	// subscriberBuffer is how many events one SSE client may be behind before
+	// it is cut loose to reconnect and replay from the event log.
+	subscriberBuffer = 16
+)
+
+// hub fans one Redis pattern subscription out to every SSE client of this
+// process. Before it, each client held its own Redis subscription, so 5,000
+// open streams meant 5,000 Redis connections.
+type hub struct {
+	mu     sync.Mutex
+	subs   map[string]map[chan eventlog.Event]struct{}
+	cancel context.CancelFunc
+	done   chan struct{} // closed when the hub stops, after every client channel is closed
+}
+
+// startHub opens the shared subscription and waits for Redis to confirm it, so
+// a client that subscribes after this returns cannot miss a later event.
+func (p *PubSub) startHub() (*hub, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+	sub := p.rdb.PSubscribe(ctx, eventChannelPrefix+"*")
+	if _, err := sub.Receive(ctx); err != nil {
+		cancel()
+		_ = sub.Close()
+		return nil, err
+	}
+	h := &hub{subs: make(map[string]map[chan eventlog.Event]struct{}), cancel: cancel, done: make(chan struct{})}
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		defer h.closeAll()
+		defer sub.Close() //nolint:errcheck // best-effort: the connection is going away
+		messages := sub.Channel()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case msg, ok := <-messages:
+				if !ok {
+					return
+				}
+				h.dispatch(msg)
+			}
+		}
+	}()
+	return h, nil
+}
+
+// getHub returns the running hub, starting it on first use. A failed start is
+// retried by the next caller.
+func (p *PubSub) getHub() (*hub, error) {
+	p.hubMu.Lock()
+	defer p.hubMu.Unlock()
+	if p.hub != nil {
+		return p.hub, nil
+	}
+	h, err := p.startHub()
+	if err != nil {
+		return nil, err
+	}
+	p.hub = h
+	return h, nil
+}
+
+// stopHub ends the shared subscription. Wait calls it before waiting.
+func (p *PubSub) stopHub() {
+	p.hubMu.Lock()
+	defer p.hubMu.Unlock()
+	if p.hub != nil {
+		p.hub.cancel()
+	}
+}
+
+func (h *hub) add(jobID string, ch chan eventlog.Event) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.subs[jobID] == nil {
+		h.subs[jobID] = make(map[chan eventlog.Event]struct{})
+	}
+	h.subs[jobID][ch] = struct{}{}
+}
+
+// remove unregisters a client. It reports whether the client was still
+// registered, so the caller closes its channel only if the hub has not already.
+func (h *hub) remove(jobID string, ch chan eventlog.Event) bool {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if _, ok := h.subs[jobID][ch]; !ok {
+		return false
+	}
+	delete(h.subs[jobID], ch)
+	if len(h.subs[jobID]) == 0 {
+		delete(h.subs, jobID)
+	}
+	return true
+}
+
+// dispatch hands a message to the clients of its job. A client whose buffer is
+// full is closed instead of skipped: the SSE handler turns a closed channel
+// into a reconnect, and the reconnect replays every event from the log.
+func (h *hub) dispatch(msg *redis.Message) {
+	jobID, ok := strings.CutPrefix(msg.Channel, eventChannelPrefix)
+	if !ok || strings.Contains(jobID, ":") {
+		return
+	}
+	h.mu.Lock()
+	if len(h.subs[jobID]) == 0 {
+		h.mu.Unlock()
+		return
+	}
+	var evt eventlog.Event
+	if err := json.Unmarshal([]byte(msg.Payload), &evt); err != nil {
+		h.mu.Unlock()
+		return
+	}
+	for ch := range h.subs[jobID] {
+		select {
+		case ch <- evt:
+		default:
+			slog.Warn("sse: client fell behind, closing it to reconnect", "job_id", jobID)
+			delete(h.subs[jobID], ch)
+			close(ch)
+		}
+	}
+	if len(h.subs[jobID]) == 0 {
+		delete(h.subs, jobID)
+	}
+	h.mu.Unlock()
+}
+
+func (h *hub) closeAll() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	defer close(h.done)
+	for jobID, set := range h.subs {
+		for ch := range set {
+			close(ch)
+		}
+		delete(h.subs, jobID)
+	}
+}
diff --git a/internal/queue/hub_test.go b/internal/queue/hub_test.go
new file mode 100644
index 0000000..2e9580a
--- /dev/null
+++ b/internal/queue/hub_test.go
@@ -0,0 +1,124 @@
+package queue
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
+)
+
+func recv(t *testing.T, ch <-chan eventlog.Event) eventlog.Event {
+	t.Helper()
+	select {
+	case evt, ok := <-ch:
+		if !ok {
+			t.Fatal("channel closed, want an event")
+		}
+		return evt
+	case <-time.After(3 * time.Second):
+		t.Fatal("no event within 3s")
+	}
+	return eventlog.Event{}
+}
+
+func TestHub_OneRedisSubscriptionServesEveryClient(t *testing.T) {
+	mr, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	a1 := ps.Subscribe(ctx, "job-a")
+	a2 := ps.Subscribe(ctx, "job-a")
+	b := ps.Subscribe(ctx, "job-b")
+	// All three clients share one pattern subscription.
+	if n := mr.PubSubNumPat(); n != 1 {
+		t.Fatalf("pattern subscriptions = %d, want exactly 1", n)
+	}
+
+	if err := ps.PublishEvent(ctx, eventlog.New("job-a", eventlog.EventRunning, nil)); err != nil {
+		t.Fatal(err)
+	}
+	if got := recv(t, a1); got.JobID != "job-a" {
+		t.Errorf("a1 got %q", got.JobID)
+	}
+	if got := recv(t, a2); got.JobID != "job-a" {
+		t.Errorf("a2 got %q", got.JobID)
+	}
+	select {
+	case evt := <-b:
+		t.Fatalf("job-b received job-a's event: %+v", evt)
+	case <-time.After(150 * time.Millisecond):
+	}
+}
+
+func TestHub_CancelledClientIsRemovedAndItsChannelClosed(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx, cancel := context.WithCancel(context.Background())
+	ch := ps.Subscribe(ctx, "job-c")
+	cancel()
+	select {
+	case _, ok := <-ch:
+		if ok {
+			t.Fatal("got an event after cancel, want a closed channel")
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("channel not closed after cancel")
+	}
+	ps.hubMu.Lock()
+	defer ps.hubMu.Unlock()
+	ps.hub.mu.Lock()
+	defer ps.hub.mu.Unlock()
+	if len(ps.hub.subs["job-c"]) != 0 {
+		t.Fatal("cancelled client still registered")
+	}
+}
+
+func TestHub_AClientThatFallsBehindIsClosedSoItReplays(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	slow := ps.Subscribe(ctx, "job-d")
+	fast := ps.Subscribe(ctx, "job-e")
+
+	for i := 0; i < subscriberBuffer+5; i++ {
+		if err := ps.PublishEvent(ctx, eventlog.New("job-d", eventlog.EventRunning, nil)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// Drain the buffered events; the channel must then be closed, not stalled.
+	deadline := time.After(3 * time.Second)
+	for open := true; open; {
+		select {
+		case _, ok := <-slow:
+			open = ok
+		case <-deadline:
+			t.Fatal("slow client's channel was never closed")
+		}
+	}
+	// Other jobs are unaffected.
+	if err := ps.PublishEvent(ctx, eventlog.New("job-e", eventlog.EventRunning, nil)); err != nil {
+		t.Fatal(err)
+	}
+	recv(t, fast)
+}
+
+func TestHub_WaitStopsTheSharedSubscription(t *testing.T) {
+	_, rdb := newTestRedis(t)
+	ps := NewPubSub(rdb)
+	ctx, cancel := context.WithCancel(context.Background())
+	ch := ps.Subscribe(ctx, "job-f")
+	done := make(chan struct{})
+	go func() { ps.Wait(); close(done) }()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Wait did not return")
+	}
+	cancel()
+	if _, ok := <-ch; ok {
+		t.Fatal("channel delivered an event after the hub stopped")
+	}
+}
diff --git a/internal/queue/pubsub.go b/internal/queue/pubsub.go
index 58b3300..c73a8c1 100644
--- a/internal/queue/pubsub.go
+++ b/internal/queue/pubsub.go
@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"strconv"
 	"sync"
 	"time"
@@ -168,6 +169,9 @@ type PubSub struct {
 	rdb              *redis.Client
 	wg               sync.WaitGroup
 	webhookMasterKey []byte // nil means no encryption
+
+	hubMu sync.Mutex
+	hub   *hub // the shared event subscription, started by the first Subscribe
 }
 
 // NewPubSub creates a PubSub with no webhook secret encryption (legacy / no key).
@@ -181,9 +185,13 @@ func NewPubSubWithKey(rdb *redis.Client, webhookMasterKey []byte) *PubSub {
 	return &PubSub{rdb: rdb, webhookMasterKey: webhookMasterKey}
 }
 
-// Wait blocks until all active Subscribe goroutines have exited.
-// Call after srv.Shutdown to ensure sub.Close() completes before rdb.Close().
-func (p *PubSub) Wait() { p.wg.Wait() }
+// Wait stops the shared event subscription and blocks until every Subscribe
+// goroutine has exited. Call after srv.Shutdown so the subscription is closed
+// before rdb.Close().
+func (p *PubSub) Wait() {
+	p.stopHub()
+	p.wg.Wait()
+}
 
 func channelName(jobID string) string {
 	return "ac:events:" + jobID
@@ -230,40 +238,32 @@ func (p *PubSub) GetEventLog(ctx context.Context, jobID string) ([]eventlog.Even
 	return events, nil
 }
 
-// Subscribe returns a channel that receives events for the given job.
-// Close the context to unsubscribe.
+// Subscribe returns a channel that receives events for the given job. It is
+// registered with the process-wide hub before Subscribe returns, so events
+// published after the call are never missed. Cancel ctx to unsubscribe. The
+// channel is closed when Redis is unreachable, when the client falls behind, or
+// when the PubSub is stopped; callers treat that as "reconnect and replay".
 func (p *PubSub) Subscribe(ctx context.Context, jobID string) <-chan eventlog.Event {
-	ch := make(chan eventlog.Event, 16)
-	sub := p.rdb.Subscribe(ctx, channelName(jobID))
-
+	ch := make(chan eventlog.Event, subscriberBuffer)
+	h, err := p.getHub()
+	if err != nil {
+		slog.Error("pubsub: could not open the shared subscription", "err", err)
+		close(ch)
+		return ch
+	}
+	h.add(jobID, ch)
 	p.wg.Add(1)
 	go func() {
 		defer p.wg.Done()
-		defer close(ch)
-		defer sub.Close() //nolint:errcheck // best-effort: unsubscribe when the SSE handler's goroutine exits
-
-		msgCh := sub.Channel()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case msg, ok := <-msgCh:
-				if !ok {
-					return
-				}
-				var evt eventlog.Event
-				if err := json.Unmarshal([]byte(msg.Payload), &evt); err != nil {
-					continue
-				}
-				select {
-				case ch <- evt:
-				case <-ctx.Done():
-					return
-				}
+		select {
+		case <-ctx.Done():
+			if h.remove(jobID, ch) {
+				close(ch)
 			}
+		case <-h.done:
+			// The hub has closed every client channel.
 		}
 	}()
-
 	return ch
 }
 
diff --git a/internal/queue/stream.go b/internal/queue/stream.go
index b37c87e..05783ca 100644
--- a/internal/queue/stream.go
+++ b/internal/queue/stream.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -99,6 +101,7 @@ func (s *Stream) Consume(ctx context.Context, consumer string, handler func(ctx
 					continue
 				}
 
+				observeQueueWait(s.stream, msg.ID)
 				if err := handler(ctx, subID); err != nil {
 					slog.Error("queue: handler error", "submission_id", subID, "err", err)
 					continue
@@ -254,6 +257,7 @@ func (dc *DualConsumer) consumeOne(ctx context.Context, stream, consumer string,
 				return true
 			}
 
+			observeQueueWait(stream, msg.ID)
 			if err := handler(ctx, subID); err != nil {
 				slog.Error("queue: handler error", "stream", stream, "submission_id", subID, "err", err)
 				return true // consumed but not ACKed — reclaimer will pick it up
@@ -303,6 +307,7 @@ func (dc *DualConsumer) blockRead(ctx context.Context, consumer string, handler
 				dc.ack(ctx, streamName, msg.ID)
 				continue
 			}
+			observeQueueWait(streamName, msg.ID)
 			if err := handler(ctx, subID); err != nil {
 				slog.Error("queue: handler error", "stream", streamName, "submission_id", subID, "err", err)
 				continue
@@ -537,3 +542,19 @@ func DrainLegacyStream(ctx context.Context, wg *sync.WaitGroup, rdb *redis.Clien
 		}
 	}()
 }
+
+// observeQueueWait records how long a message sat in its stream. Redis stream
+// IDs start with the millisecond they were added, so the wait needs no extra
+// field on the message.
+func observeQueueWait(stream, id string) {
+	ms, _, ok := strings.Cut(id, "-")
+	if !ok {
+		return
+	}
+	added, err := strconv.ParseInt(ms, 10, 64)
+	if err != nil {
+		return
+	}
+	wait := time.Since(time.UnixMilli(added))
+	metrics.QueueWaitSeconds.WithLabelValues(stream).Observe(max(wait, 0).Seconds())
+}
diff --git a/internal/queue/wait_test.go b/internal/queue/wait_test.go
new file mode 100644
index 0000000..4408baa
--- /dev/null
+++ b/internal/queue/wait_test.go
@@ -0,0 +1,27 @@
+package queue
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
+)
+
+func TestObserveQueueWait(t *testing.T) {
+	before := testutil.CollectAndCount(metrics.QueueWaitSeconds)
+	added := time.Now().Add(-3 * time.Second).UnixMilli()
+	observeQueueWait("ac:test-wait", fmt.Sprintf("%d-0", added))
+	if got := testutil.CollectAndCount(metrics.QueueWaitSeconds); got != before+1 {
+		t.Fatalf("series after a valid id = %d, want %d", got, before+1)
+	}
+	// Ids that are not stream ids add nothing and do not panic.
+	for _, id := range []string{"", "nope", "abc-0", "-"} {
+		observeQueueWait("ac:test-wait", id)
+	}
+	if got := testutil.CollectAndCount(metrics.QueueWaitSeconds); got != before+1 {
+		t.Fatalf("series after invalid ids = %d, want unchanged %d", got, before+1)
+	}
+}
diff --git a/internal/sandbox/args_test.go b/internal/sandbox/args_test.go
index 4782c7a..6fe5878 100644
--- a/internal/sandbox/args_test.go
+++ b/internal/sandbox/args_test.go
@@ -46,3 +46,23 @@ func TestRunCmdCarriesArgsStackAndProcs(t *testing.T) {
 		t.Fatalf("defaults changed: procs = %d, stack = %d", plain.ProcLimit, plain.StackLimit)
 	}
 }
+
+func TestJavaStartsLeanAndKeepsItsExecutables(t *testing.T) {
+	t.Parallel()
+	for _, id := range []string{"java", "kotlin"} {
+		run := Languages[id].RunArgs
+		if run[0] != "/usr/bin/java" {
+			t.Errorf("%s: run executable = %s, want /usr/bin/java", id, run[0])
+		}
+		joined := strings.Join(run, " ")
+		for _, flag := range []string{"-XX:+UseSerialGC", "-Xshare:auto", "-XX:TieredStopAtLevel=1"} {
+			if !strings.Contains(joined, flag) {
+				t.Errorf("%s: run args %q lack %s", id, joined, flag)
+			}
+		}
+	}
+	// -J flags must come right after javac and before the options it parses.
+	if compile := Languages["java"].CompileArgs; compile[0] != "/usr/bin/javac" || !strings.HasPrefix(compile[1], "-J") {
+		t.Errorf("javac args = %v, want the -J flags straight after the executable", compile)
+	}
+}
diff --git a/internal/sandbox/language.go b/internal/sandbox/language.go
index e963925..654732f 100644
--- a/internal/sandbox/language.go
+++ b/internal/sandbox/language.go
@@ -131,14 +131,14 @@ var Languages = map[string]LangConfig{
 	},
 	"java": {
 		Compiled:    true,
-		CompileArgs: []string{"/usr/bin/javac", "-d", "/w", "/w/Main.java"},
+		CompileArgs: []string{"/usr/bin/javac", "-J-XX:TieredStopAtLevel=1", "-J-XX:+UseSerialGC", "-J-Xshare:auto", "-d", "/w", "/w/Main.java"},
 		CompileEnv:  []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"},
 		CompileCPU:  15_000_000_000,
 		CompileMem:  536_870_912,
 		CompileProc: 50,
 		SourcePath:  "/w/Main.java",
 		BinaryPath:  "/w/Main.class",
-		RunArgs:     []string{"/usr/bin/java", "-XX:+TieredCompilation", "-XX:TieredStopAtLevel=1", "-cp", "/w", "Main"},
+		RunArgs:     []string{"/usr/bin/java", "-XX:+TieredCompilation", "-XX:TieredStopAtLevel=1", "-XX:+UseSerialGC", "-Xshare:auto", "-cp", "/w", "Main"},
 		RunEnv:      []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"},
 		RunCPU:      2_000_000_000,
 		RunMem:      536_870_912,
@@ -207,7 +207,7 @@ var Languages = map[string]LangConfig{
 		CompileProc: 50,
 		SourcePath:  "/w/Main.kt",
 		BinaryPath:  "/w/Main.jar",
-		RunArgs:     []string{"/usr/bin/java", "-jar", "/w/Main.jar"},
+		RunArgs:     []string{"/usr/bin/java", "-XX:TieredStopAtLevel=1", "-XX:+UseSerialGC", "-Xshare:auto", "-jar", "/w/Main.jar"},
 		RunEnv:      []string{"PATH=/usr/bin:/bin", "JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64"},
 		RunCPU:      2_000_000_000,
 		RunMem:      536_870_912,
diff --git a/internal/worker/worker.go b/internal/worker/worker.go
index f9a115e..df7e41a 100644
--- a/internal/worker/worker.go
+++ b/internal/worker/worker.go
@@ -131,6 +131,21 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 
 	log := slog.With("submission_id", submissionID)
 
+	// The job's tests live in Redis and do not depend on the database claim, so
+	// the two round trips overlap. The result is read after the claim below.
+	type loaded struct {
+		spec jobSpec
+		err  error
+		took time.Duration
+	}
+	testsCh := make(chan loaded, 1)
+	go func() {
+		t := time.Now()
+		spec, err := w.loadTests(jobCtx, submissionID)
+		testsCh <- loaded{spec, err, time.Since(t)}
+	}()
+
+	claimStart := time.Now()
 	sub, err := db.GetSubmission(jobCtx, w.pool, id)
 	if err != nil {
 		return fmt.Errorf("worker: get submission: %w", err)
@@ -150,9 +165,12 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 		return fmt.Errorf("worker: claim submission: %w", err)
 	}
 
+	metrics.JobStageSeconds.WithLabelValues("claim").Observe(time.Since(claimStart).Seconds())
+
 	// Count only jobs we've actually claimed; deferred duration also starts here.
 	metrics.ActiveWorkers.Inc()
 	defer metrics.ActiveWorkers.Dec()
+	defer metrics.JobsDrain.Complete() // a failed job frees its slot too
 	// finalVerdict is set just before each return so the deferred histogram gets labels.
 	finalVerdict := string(model.VerdictInternalError)
 	defer func() {
@@ -176,8 +194,10 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 		}
 	}()
 
-	// Load the job's tests and limits from Redis
-	spec, err := w.loadTests(jobCtx, submissionID)
+	// The job's tests and limits, fetched from Redis while the claim ran.
+	loadedTests := <-testsCh
+	metrics.JobStageSeconds.WithLabelValues("load_tests").Observe(loadedTests.took.Seconds())
+	spec, err := loadedTests.spec, loadedTests.err
 	if err != nil {
 		return w.failSubmission(jobCtx, id, model.VerdictInternalError, fmt.Sprintf("load tests: %v", err), log)
 	}
@@ -257,6 +277,7 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 		log.Warn("failed to publish RUNNING event", "err", err)
 	}
 
+	runStart := time.Now()
 	results, totalCPU, peakMem, passed, overallVerdict, err := w.runTestsBatched(jobCtx, submissionID, lang, sub.SourceCode, spec.tests, fileID, spec.limits, spec.files, log)
 	if err != nil {
 		return w.failSubmission(jobCtx, id, model.VerdictInternalError, fmt.Sprintf("batch run: %v", err), log)
@@ -267,6 +288,8 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 	writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), 15*time.Second)
 	defer writeCancel()
 
+	metrics.JobStageSeconds.WithLabelValues("run").Observe(time.Since(runStart).Seconds())
+	persistStart := time.Now()
 	if err := db.InsertResultsAndVerdict(writeCtx, w.pool, id, results,
 		model.StatusCompleted, overallVerdict, "",
 		totalCPU, peakMem, len(spec.tests), passed); err != nil {
@@ -277,6 +300,7 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 		return fmt.Errorf("worker: write results+verdict: %w", err)
 	}
 
+	metrics.JobStageSeconds.WithLabelValues("persist").Observe(time.Since(persistStart).Seconds())
 	eventResults := verdictResults(results)
 
 	effectiveCPU, effectiveMem := lang.Effective(spec.limits)
@@ -290,9 +314,11 @@ func (w *Worker) process(ctx context.Context, submissionID string) error {
 		"applied_memory_limit_bytes": effectiveMem,
 		"results":                    eventResults,
 	}
+	publishStart := time.Now()
 	if err := w.pubsub.PublishEvent(writeCtx, eventlog.New(submissionID, eventlog.EventVerdict, verdictData)); err != nil {
 		log.Error("failed to publish VERDICT event (DB has verdict, SSE client may not)", "err", err)
 	}
+	metrics.JobStageSeconds.WithLabelValues("publish").Observe(time.Since(publishStart).Seconds())
 
 	if spec.callbackURL != "" {
 		go dispatchWebhook(context.WithoutCancel(ctx), spec.callbackURL, spec.callbackSecret, WebhookPayload{

From 716987855f855a091175000a63fa0dcf1413280b Mon Sep 17 00:00:00 2001
From: Shreesh <shreesh.exe22@gmail.com>
Date: Mon, 2 Nov 2026 17:28:56 +0530
Subject: [PATCH 3/4] style(gateway): group imports in handler test

---
 internal/gateway/handler_test.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go
index 5d04b1b..22337c6 100644
--- a/internal/gateway/handler_test.go
+++ b/internal/gateway/handler_test.go
@@ -16,10 +16,9 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	goredis "github.com/redis/go-redis/v9"
-	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
-
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/db"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
+	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/model"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/queue"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/sandbox"

From 9876a3fcfa936475d48acc07e686a228b1df8b30 Mon Sep 17 00:00:00 2001
From: Shreesh <shreesh.exe22@gmail.com>
Date: Mon, 2 Nov 2026 18:10:09 +0530
Subject: [PATCH 4/4] style(gateway): separate local imports in handler test

---
 internal/gateway/handler_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go
index 22337c6..d0d69cf 100644
--- a/internal/gateway/handler_test.go
+++ b/internal/gateway/handler_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgxpool"
 	goredis "github.com/redis/go-redis/v9"
+
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/db"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/eventlog"
 	"github.com/stjosephsplacements/AetherCode-Execution-Engine/internal/metrics"