refactor(server): tighten agent runtime + interactive SPI seams

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/mentor/MentorAgentProperties.java

@@ -13,12 +13,15 @@
 /**
  * When llmProvider + credentialMode + modelName are all set, mentor uses them directly and
  * skips the workspace-scoped AgentConfig table. Omit any one to fall back to AgentConfig.
+ *
+ * <p>The runner script name is owned by {@link MentorRunnerProfile} — operator overrides for it
+ * were never used in practice and risked drifting from the V8 flags / per-process env that the
+ * kernel pairs with the script. Bumping the runner is a code change.
  */
 @Validated
 @ConfigurationProperties(prefix = "hephaestus.mentor.agent")
 public record MentorAgentProperties(
     @DefaultValue("ghcr.io/ls1intum/hephaestus/agent-pi:latest") @NotBlank String image,
-    @DefaultValue("pi-mentor-runner.mjs") @NotBlank String runnerScript,
     @DefaultValue("100000") @Min(1) int maxPromptChars,
     @DefaultValue("") String baseUrl,
     @Nullable LlmProvider llmProvider,

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/mentor/MentorPiAdapter.java

@@ -35,7 +35,9 @@ public class MentorPiAdapter {
     public static final String ASPECT_INPUT_PREFIX = "context/target/";
 
     /** Workspace-relative directory for restored Pi SDK session JSONL files (matches the runner's {@code SESSIONS_DIR}). */
-    public static final String SESSIONS_DIR_PREFIX = ".sessions/";
+    public static final String SESSIONS_DIR_PREFIX = WorkspaceAbi.SESSIONS_DIR_PREFIX;
+
+    private static final MentorRunnerProfile PROFILE = new MentorRunnerProfile();
 
     private final PiRuntimeFactory runtimeFactory;
     private final MentorAgentProperties mentorProperties;
@@ -75,7 +77,7 @@ public InteractiveSandboxSpec buildSandboxSpec(
             null,
             true,
             llmConfig.timeoutSeconds(),
-            mentorProperties.runnerScript(),
+            PROFILE,
             extraInputs,
             "" /* no precompute step — mentor analytics arrive as aspect JSON */
         );

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/mentor/MentorRunnerProfile.java

@@ -0,0 +1,74 @@
+package de.tum.in.www1.hephaestus.agent.mentor;
+
+import de.tum.in.www1.hephaestus.agent.runtime.PiRunnerProfile;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Runner profile for the long-lived mentor chat agent.
+ *
+ * <p><b>V8 flags:</b>
+ * <ul>
+ *   <li>{@code --max-old-space-size=256} — cap V8 old-gen at 256 MB. Empirically the mentor
+ *       runtime sits at ~100 MB V8 heap; 2.5× headroom defends against a leaky session OOM-ing
+ *       the host instead of itself. Default ~1.4 GB on 64-bit lets one bad runner take the host
+ *       down.</li>
+ *   <li>{@code --no-warnings} — keeps stderr clean for ops grep against our own log prefix.</li>
+ *   <li>{@code --expose-gc} — exposes {@code global.gc()} so the runner can force a post-turn
+ *       compaction in {@code pi-mentor-runner.mjs:forwardEvent}. The flag costs nothing on its
+ *       own and is only effective when {@code global.gc()} is actually called.</li>
+ * </ul>
+ *
+ * <p>Note: {@code --disable-source-maps} was removed in Node 22 (source maps are off by default;
+ * the flag itself no longer exists and causes {@code bad option} exit 9). Do not re-add.
+ *
+ * <p>We deliberately removed {@code --max-semi-space-size=16} and {@code UV_THREADPOOL_SIZE=2}
+ * from prior revisions: the former matches the Node 22 default on 64-bit (so it was a no-op),
+ * and the latter risks serialising libuv fs/crypto bursts.
+ *
+ * <p><b>Per-process env:</b> {@code LD_PRELOAD=libjemalloc.so.2} +
+ * {@code MALLOC_CONF=background_thread:true,narenas:2,dirty_decay_ms:30000,muzzy_decay_ms:30000}.
+ * Mentor's long-lived heap benefits from jemalloc's page-decay tuning; precompute's bursty
+ * allocations don't. {@code background_thread:true} runs jemalloc's page-decay sweep on a
+ * dedicated thread — without it the mutator must re-enter the allocator to trigger decay, which
+ * a long-idle Node loop rarely does (cf. jemalloc TUNING.md). Decay window 30 s matches
+ * jemalloc upstream's "long-lived process" recommendation.
+ *
+ * <p>The path matches the {@code /usr/local/lib/libjemalloc.so.2} symlink created by the Pi
+ * Dockerfile ({@code docker/agents/pi/Dockerfile}). The symlink is per-arch by design; the env
+ * literal here is arch-independent.
+ */
+public final class MentorRunnerProfile implements PiRunnerProfile {
+
+    /** Filename of the mentor runner under {@code resources/agent/}. */
+    public static final String SCRIPT = "pi-mentor-runner.mjs";
+
+    private static final List<String> FLAGS = List.of("--max-old-space-size=256", "--no-warnings", "--expose-gc");
+
+    private static final Map<String, String> ENV;
+
+    static {
+        // LinkedHashMap preserves declaration order — the assembled command line is the same
+        // every build, which simplifies cross-build diff inspection.
+        LinkedHashMap<String, String> env = new LinkedHashMap<>();
+        env.put("LD_PRELOAD", "/usr/local/lib/libjemalloc.so.2");
+        env.put("MALLOC_CONF", "background_thread:true,narenas:2,dirty_decay_ms:30000,muzzy_decay_ms:30000");
+        ENV = Map.copyOf(env);
+    }
+
+    @Override
+    public String runnerScript() {
+        return SCRIPT;
+    }
+
+    @Override
+    public List<String> nodeFlags() {
+        return FLAGS;
+    }
+
+    @Override
+    public Map<String, String> additionalEnv() {
+        return ENV;
+    }
+}

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/mentor/chat/MentorFrameFilters.java

@@ -0,0 +1,52 @@
+package de.tum.in.www1.hephaestus.agent.mentor.chat;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.Predicate;
+
+/**
+ * JSON-RPC-aware {@link Predicate} factories for use with
+ * {@code AttachedSandbox.subscribe(Cursor, Predicate, Consumer)}.
+ *
+ * <p>The mentor sandbox is shared by {@code (userId, workspaceId)}: a second chat tab in the
+ * same workspace subscribes to the SAME frame stream. The bound thread filter drops frames
+ * whose JSON-RPC {@code params.threadId} does not match the subscriber's thread; without it,
+ * tab-A's translator would observe tab-B's text deltas and ship them down tab-A's wire.
+ *
+ * <p><b>Broadcast contract:</b> frames without a {@code params.threadId} (or with a null one)
+ * are server notifications — {@code runner_ready}, ring metadata, server status. Filters MUST
+ * pass them through unless deliberately suppressing them. {@code forThread(null)} accepts all
+ * frames; this is the test-only path used by legacy unit tests that pre-date multi-session.
+ */
+final class MentorFrameFilters {
+
+    private MentorFrameFilters() {}
+
+    /** Returns a predicate that passes broadcasts + frames matching {@code threadId}. */
+    static Predicate<JsonNode> forThread(UUID threadId) {
+        if (threadId == null) {
+            return frame -> true;
+        }
+        String expected = threadId.toString();
+        return frame -> {
+            // The mentor runner uses two top-level shapes that ride the sandbox frame stream:
+            //   • Notifications:  { method, params: { threadId?, event } }
+            //   • Responses:      { id, result | error }
+            // Responses have no threadId — they correlate to a pending call already routed by id
+            // inside MentorRunnerClient, so they MUST pass through every client's filter.
+            if (!frame.isObject()) {
+                return true;
+            }
+            JsonNode params = frame.get("params");
+            if (params == null || !params.isObject()) {
+                return true; // response / non-routed frame
+            }
+            JsonNode threadIdNode = params.get("threadId");
+            if (threadIdNode == null || threadIdNode.isNull()) {
+                return true; // broadcast notification (runner_ready, etc.)
+            }
+            return Objects.equals(threadIdNode.asText(), expected);
+        };
+    }
+}

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/mentor/chat/MentorRunnerClient.java

@@ -106,7 +106,16 @@ public synchronized void start() {
         // Cursor.FROM_NOW: skip ring-buffer replay of frames from prior turns on the same
         // reused sandbox. Without this, a second turn replays turn-1's agent_end event and
         // completes instantly with stale data.
-        this.subscription = sandbox.subscribe(Cursor.FROM_NOW, this::onFrame);
+        //
+        // Per-thread routing is pushed into the SPI as a Predicate<JsonNode>: rejected frames
+        // never enter this client's subscription queue and never consume drop-counter budget.
+        // Broadcast frames (server notifications without a thread destination) cross the
+        // filter for every client — see MentorFrameFilters.forThread.
+        this.subscription = sandbox.subscribe(
+            Cursor.FROM_NOW,
+            MentorFrameFilters.forThread(boundThreadId),
+            this::onFrame
+        );
     }
 
     public CompletableFuture<JsonNode> hello() {
@@ -241,18 +250,10 @@ private void handleEvent(JsonNode frame) {
             log.debug("Runner event frame missing params.event — ignoring");
             return;
         }
-        // Per-thread fan-out: the sandbox is shared by (userId, workspaceId), so a second
-        // chat tab in the same workspace subscribes to the same frame stream. Drop any frame
-        // whose threadId doesn't match the one this client is bound to — without the filter,
-        // tab-A's translator sees tab-B's text deltas and ships them down tab-A's wire.
-        // Notification-type frames (`runner_ready`) ship with `threadId: null` and pass
-        // through here for ALL clients; the translator drops them by event-type.
-        if (boundThreadId != null && params.has("threadId") && !params.get("threadId").isNull()) {
-            String frameThreadId = params.get("threadId").asText();
-            if (!boundThreadId.toString().equals(frameThreadId)) {
-                return;
-            }
-        }
+        // Per-thread fan-out used to filter here; the predicate is now applied server-side at
+        // subscribe() time so cross-thread frames never enter this client's dispatcher queue.
+        // Notification-type frames (`runner_ready`) ship with `threadId: null` and pass through
+        // the predicate for ALL clients; the translator drops them by event-type.
         try {
             onEvent.accept(params.get("event"));
         } catch (RuntimeException e) {

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/practice/PracticePiAdapter.java

@@ -32,6 +32,8 @@ public PracticePiAdapter(
         this.properties = properties;
     }
 
+    private static final PracticeRunnerProfile PROFILE = new PracticeRunnerProfile();
+
     public PracticeSandboxSpec buildSandboxSpec(PracticeAgentRequest request) {
         PiRuntimeFactory.PiPlan plan = runtimeFactory.build(
             new PiPlanSpec(
@@ -43,7 +45,7 @@ public PracticeSandboxSpec buildSandboxSpec(PracticeAgentRequest request) {
                 request.jobToken(),
                 request.allowInternet(),
                 request.timeoutSeconds(),
-                properties.runnerScript(),
+                PROFILE,
                 Map.of(),
                 buildPrecomputeStep()
             )

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/practice/PracticeRunnerProfile.java

@@ -0,0 +1,40 @@
+package de.tum.in.www1.hephaestus.agent.practice;
+
+import de.tum.in.www1.hephaestus.agent.runtime.PiRunnerProfile;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Runner profile for the one-shot practice-review agent.
+ *
+ * <p><b>V8 flags:</b> only {@code --no-warnings}. We do NOT cap the heap because practice
+ * routinely parses 30-file diff patches that allocate transiently; a 256 MB cap would convert
+ * worst-case-input OOMs from "rare" to "regular." We do NOT {@code --expose-gc} because the
+ * practice runner never calls {@code global.gc()} and exposing the global is a foot-gun.
+ *
+ * <p><b>Per-process env:</b> empty. Practice's bursty allocations don't benefit from jemalloc's
+ * page-decay tuning, and forcing {@code LD_PRELOAD} on a short-lived process is unmeasured
+ * overhead.
+ */
+public final class PracticeRunnerProfile implements PiRunnerProfile {
+
+    /** Filename of the practice runner under {@code resources/agent/}. */
+    public static final String SCRIPT = "pi-runner.mjs";
+
+    private static final List<String> FLAGS = List.of("--no-warnings");
+
+    @Override
+    public String runnerScript() {
+        return SCRIPT;
+    }
+
+    @Override
+    public List<String> nodeFlags() {
+        return FLAGS;
+    }
+
+    @Override
+    public Map<String, String> additionalEnv() {
+        return Map.of();
+    }
+}

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/runtime/PiAgentProperties.java

@@ -6,10 +6,15 @@
 import org.springframework.boot.context.properties.bind.DefaultValue;
 import org.springframework.validation.annotation.Validated;
 
+/**
+ * Pi-agent image/pull-policy configuration. The runner script name is owned by
+ * {@link de.tum.in.www1.hephaestus.agent.practice.PracticeRunnerProfile} — operator overrides
+ * for it were never used in practice and risked drifting from the V8 flags / per-process env
+ * that the kernel pairs with the script. Bumping the runner is a code change.
+ */
 @Validated
 @ConfigurationProperties(prefix = "hephaestus.agent.pi")
 public record PiAgentProperties(
     @DefaultValue("ghcr.io/ls1intum/hephaestus/agent-pi:latest") @NotBlank String image,
-    @DefaultValue("pi-runner.mjs") @NotBlank String runnerScript,
     @DefaultValue("IF_NOT_PRESENT") ImagePullPolicy pullPolicy
 ) {}

server/application-server/src/main/java/de/tum/in/www1/hephaestus/agent/runtime/PiPlanSpec.java

@@ -23,8 +23,14 @@
  * @param jobToken        PROXY mode job token; must be non-blank in PROXY mode
  * @param allowInternet   PROXY mode internet flag; ignored otherwise (always true)
  * @param timeoutSeconds  total sandbox timeout (must be {@code > TIMEOUT_BUFFER_SECONDS})
- * @param runnerScript    filename of the Pi runner under {@code resources/agent/}; required
- * @param extraInputs     additional workspace files keyed by relative path (e.g. {@code task.json})
+ * @param runnerProfile   per-runner-kind strategy (script filename, V8 flags, per-process env);
+ *                        the kernel reads three properties off it instead of dispatching by
+ *                        filename match
+ * @param extraInputs     additional workspace files keyed by relative path. Each key MUST be
+ *                        prefixed by {@link WorkspaceAbi#CONTEXT_TARGET_PREFIX} or appear in
+ *                        {@link WorkspaceAbi#allowedExtraInputPaths()} — the validation
+ *                        prevents adapters from writing to arbitrary workspace paths and forces
+ *                        new mount points to be declared on {@link WorkspaceAbi}
  * @param precomputeStep  shell fragment ending in {@code " && "} (or empty)
  */
 public record PiPlanSpec(
@@ -36,16 +42,16 @@ public record PiPlanSpec(
     @Nullable String jobToken,
     boolean allowInternet,
     int timeoutSeconds,
-    String runnerScript,
+    PiRunnerProfile runnerProfile,
     Map<String, byte[]> extraInputs,
     String precomputeStep
 ) {
     public PiPlanSpec {
         Objects.requireNonNull(provider, "provider");
         Objects.requireNonNull(credentialMode, "credentialMode");
-        Objects.requireNonNull(runnerScript, "runnerScript");
-        if (runnerScript.isBlank()) {
-            throw new IllegalArgumentException("runnerScript must not be blank");
+        Objects.requireNonNull(runnerProfile, "runnerProfile");
+        if (runnerProfile.runnerScript() == null || runnerProfile.runnerScript().isBlank()) {
+            throw new IllegalArgumentException("runnerProfile.runnerScript() must not be blank");
         }
         if (timeoutSeconds <= PiRuntimeFactory.TIMEOUT_BUFFER_SECONDS) {
             throw new IllegalArgumentException(
@@ -68,6 +74,32 @@ public record PiPlanSpec(
             }
         }
         extraInputs = extraInputs != null ? Map.copyOf(extraInputs) : Map.of();
+        // Fail-fast at construction: every extraInputs key must be a recognised workspace path.
+        // Adapters caught writing to arbitrary paths surface here at boot/test time instead of
+        // silently overwriting a future workspace mount-point.
+        for (String path : extraInputs.keySet()) {
+            if (path == null) {
+                throw new IllegalArgumentException("extraInputs keys must not be null");
+            }
+            boolean ok = WorkspaceAbi.allowedExtraInputPaths().contains(path);
+            if (!ok) {
+                for (String prefix : WorkspaceAbi.allowedExtraInputPrefixes()) {
+                    if (path.startsWith(prefix)) {
+                        ok = true;
+                        break;
+                    }
+                }
+            }
+            if (!ok) {
+                throw new IllegalArgumentException(
+                    "extraInputs path '" +
+                        path +
+                        "' is not a recognised workspace path: must appear in " +
+                        "WorkspaceAbi.allowedExtraInputPaths() or be prefixed by one of " +
+                        WorkspaceAbi.allowedExtraInputPrefixes()
+                );
+            }
+        }
         precomputeStep = precomputeStep != null ? precomputeStep : "";
     }
 }