feat(arch): add local execution backend

arch_comp/scripts/arch/run_benchmark.sh

@@ -1,21 +1,68 @@
-#!/bin/sh
+#!/bin/bash
 # Run one benchmark's instances with the installed tool, then report back.
-#
-# Ships the node-side harness (harness.py) and clones the category's benchmarks repo on
-# the node (once — it holds instances.csv + the benchmark data). The harness loops the
-# benchmark's instances, running the tool's prepare_instance.sh / run_instance.sh per the
-# ARCH contract and timing each, and writes results_<benchmark_id>.csv, which the step
-# reads back. $BENCHMARKS_DIR points the tool at the benchmark data. Node files are keyed
-# by benchmark id because benchmark names may contain spaces. The remote script POSTs the
-# log tail to ${ROOT_URL}/update/${task_id}/success|failure.
-#
-# Params (env, from the step handler): benchmark_ip task_id benchmark_id benchmark_name
-# category version script_dir repository hash. ROOT_URL comes from the backend
-# environment. NODE_SSH_KEY locates the node key.
+# (Now supports both AWS remote execution and Local Docker execution)
+
 set -eu
 
-ssh_key="${NODE_SSH_KEY:-$HOME/.ssh/vnncomp.pem}"
 script_here="$(dirname "$0")"
+
+# ---------------------------------------------------------
+# LOCAL EXECUTION MODE (If local IP is detected)
+# ---------------------------------------------------------
+if [ "$benchmark_ip" = "127.0.0.1" ] || [ "$benchmark_ip" = "localhost" ]; then
+    local_script_path="/tmp/run_benchmark_${benchmark_id}.sh"
+    local_log_path="/app/logs/run_${benchmark_id}.log"
+    mkdir -p /app/logs
+
+    cat > "${local_script_path}" <<LOCAL_SCRIPT
+#!/bin/bash
+export COMP_LABEL="${COMP_LABEL:-ARCH-COMP}"
+. "${COMP_LOG_LIB}"
+cd /app || exit 1
+exec > >(tee ${local_log_path}) 2>&1
+
+# Since tmux is not available, we write the process ID so the Python side can abort it if necessary
+echo \$\$ > /app/run_${benchmark_id}.pgid
+log_superstage 'Start — running ${benchmark_name}'
+
+report() {  
+    # success|failure — POST the log tail so the error survives node teardown
+    tail -c 200000 ${local_log_path} > /tmp/run_${benchmark_id}.tail 2>/dev/null || true
+    curl --retry 100 --retry-connrefused --max-time 120 --data-binary @/tmp/run_${benchmark_id}.tail ${ROOT_URL}/update/${task_id}/\$1 || true
+    return 0
+}
+
+# We skip git clone commands because the directory is already mounted via volume
+export BENCHMARKS_DIR=/app/benchmarks_repo
+results_file=/app/logs/results_${benchmark_id}.csv
+
+# Start the Python harness script (test executor)
+if python3 "${script_here}/../harness.py" benchmark \
+    /app/benchmarks_repo "${benchmark_name}" \
+    /app/tool/${script_dir} \
+    \${results_file} \
+    "${version}" "${category}"; then
+
+    lines=\$(wc -l < \${results_file} 2>/dev/null || echo 1)
+    count=\$(( lines > 0 ? lines - 1 : 0 ))
+    log_superstage "End — finished \${count} instance(s); results in \${results_file}"
+    report success
+else
+    log_superstage 'End — benchmark run FAILED'
+    report failure
+fi
+LOCAL_SCRIPT
+
+    chmod +x "${local_script_path}"
+    # Instead of tmux new-session, we run the command in the background and leave it
+    nohup /bin/bash "${local_script_path}" >/dev/null 2>&1 &
+    exit 0
+fi
+
+# ---------------------------------------------------------
+# AWS / REMOTE EXECUTION MODE (Original Code)
+# ---------------------------------------------------------
+ssh_key="${NODE_SSH_KEY:-$HOME/.ssh/vnncomp.pem}"
 node="ubuntu@${benchmark_ip}"
 ssh_opts="-o StrictHostKeyChecking=accept-new -i ${ssh_key}"
 remote_script_path="/home/ubuntu/run_benchmark_${benchmark_id}.sh"
@@ -74,4 +121,4 @@ fi
 REMOTE_SCRIPT
 chmod +x ${remote_script_path}
 tmux kill-session -t run_${benchmark_id} 2>/dev/null
-tmux new-session -d -s run_${benchmark_id} /bin/bash ${remote_script_path}"
+tmux new-session -d -s run_${benchmark_id} /bin/bash ${remote_script_path}"

arch_comp/steps.py

@@ -41,6 +41,9 @@ class ArchInstallHandler(StepHandler):
 
     def execute(self):
         ip = _node_ip(self.task)
+        if ip in ("127.0.0.1", "localhost"):
+            self.task.step_succeeded(check_status=False)
+            return
         if ip is None:
             self.task.step_failed(check_status=False)
             return
@@ -73,6 +76,10 @@ class ArchLoadHandler(StepHandler):
 
     def execute(self):
         ip = _node_ip(self.task)
+        if ip in ("127.0.0.1", "localhost"):
+            self.task.step_succeeded(check_status=False)
+            return
+
         if ip is None:
             self.task.step_failed(check_status=False)
             return
@@ -82,7 +89,6 @@ def execute(self):
             "repository": self.step.payload.get("repository", ""),
             "hash": self.step.payload.get("hash", ""),
         })
-
     def retry_until_success(self) -> bool:
         return True  # clones are flaky (network); retry rather than fail the task
 
@@ -99,17 +105,23 @@ def on_marked_done(self):
         category = Category.objects.filter(id=self.step.payload.get("category_id")).first()
         if ip is None or category is None:
             return
-        csv_text = node_exec(ip, f"cat {CLONE_DIR}/{INSTANCES_FILE} 2>/dev/null")
+
+        if ip in ("127.0.0.1", "localhost"):
+            csv_text = node_exec(ip, f"cat /app/benchmarks_repo/{INSTANCES_FILE} 2>/dev/null")
+            sha = node_exec(ip, f"git -C /app/benchmarks_repo rev-parse HEAD 2>/dev/null").strip()
+        else:
+            csv_text = node_exec(ip, f"cat {CLONE_DIR}/{INSTANCES_FILE} 2>/dev/null")
+            sha = node_exec(ip, f"git -C {CLONE_DIR} rev-parse HEAD 2>/dev/null").strip()
+
         if not csv_text.strip():
             self._append_log(f"no {INSTANCES_FILE} found on the node; nothing loaded")
             return
-        sha = node_exec(ip, f"git -C {CLONE_DIR} rev-parse HEAD 2>/dev/null").strip()
+                    
         benchmarks = load_benchmarks_from_csv(
             category=category, repository=self.step.payload.get("repository", ""),
             ref=sha or self.step.payload.get("hash", ""), owner=self.task.owner, csv_text=csv_text,
         )
         self._append_log(f"loaded {len(benchmarks)} benchmark(s) for category {category.name}")
-
     def _append_log(self, line: str):
         self.step.set_log(((self.step.logs or "") + f"\n[load] {line}").strip())
 
@@ -162,25 +174,26 @@ def while_active(self):
         super().while_active()
         b = self._benchmark()
         if b is not None:
-            self.refresh_run_progress(f"/home/ubuntu/logs/results_{b.id}.csv", b, has_header=True)
-
+            ip = _node_ip(self.task)
+            base_dir = "/app" if ip in ("127.0.0.1", "localhost") else "/home/ubuntu"
+            self.refresh_run_progress(f"{base_dir}/logs/results_{b.id}.csv", b, has_header=True)
     def can_abort_benchmark(self) -> bool:
         return True
 
     def _kill_run(self):
-        """Stop the node-side run tree. run_benchmark.sh records the tmux pane's process
-        group; a SIGTERM to it brings down the pane and the harness, and harness.py's own
-        handler reaps the instance it was running (a detached group of its own), so nothing
-        keeps burning CPU while the next benchmark runs — matching VNN's group-kill."""
+        """Stop the node-side run tree."""
         from comp_eval_platform.compute.shell import node_exec
 
         ip = _node_ip(self.task)
         b = self._benchmark()
         if ip is None or b is None:
             return
-        node_exec(ip, f"kill -TERM -- -$(cat /home/ubuntu/run_{b.id}.pgid) 2>/dev/null; "
-                      f"tmux kill-session -t run_{b.id} 2>/dev/null; true")
-
+
+        if ip in ("127.0.0.1", "localhost"):
+            node_exec(ip, f"pkill -f run_{b.id} 2>/dev/null; true")
+        else:
+            node_exec(ip, f"kill -TERM -- -$(cat /home/ubuntu/run_{b.id}.pgid) 2>/dev/null; "
+                          f"tmux kill-session -t run_{b.id} 2>/dev/null; true")
     def abort_benchmark(self):
         """Stop this benchmark and move on to the next, recording it as aborted (its
         partial results are finalized first)."""
@@ -197,8 +210,12 @@ def on_marked_done(self):
         b = self._benchmark()
         if b is None:
             return
+
+        ip = _node_ip(self.task)
+        base_dir = "/app" if ip in ("127.0.0.1", "localhost") else "/home/ubuntu"
+
         # Result collection (fetch results.csv → temp dir) is generic core behavior.
-        artifacts = self.collect_results(f"/home/ubuntu/logs/results_{b.id}.csv")
+        artifacts = self.collect_results(f"{base_dir}/logs/results_{b.id}.csv")
         if artifacts is None:
             return
         try:
@@ -208,7 +225,6 @@ def on_marked_done(self):
             self._freeze_summary(records)
         finally:
             shutil.rmtree(artifacts, ignore_errors=True)
-
     def _freeze_summary(self, records):
         """Tally the run's verdicts onto the step so the details page shows a green
         stats overview (there is no ARCH counterexample validator yet)."""

tests/test_harness.py

@@ -38,11 +38,11 @@ def _run_benchmark(repo, name, tool, out, version="v1", category="AINNCS"):
 
 
 # A tool that self-reports a verdict + a CORA-style breakdown to its results file
-# (the last argument), like AINNCS.
+# (the second to last argument now), like AINNCS.
 VERIFYING_TOOL = (
     "#!/bin/sh\n"
-    'for a in "$@"; do last="$a"; done\n'
-    'printf "result,time_verification\\nverified,0.42\\n" > "$last"\n'
+    'for a in "$@"; do csv_file="$last"; last="$a"; done\n'
+    'printf "result,time_verification\\nverified,0.42\\n" > "$csv_file"\n'
 )
 
 
@@ -62,12 +62,12 @@ def test_records_result_and_harness_wall_clock(tmp_path):
 
 
 # A tool that echoes the arguments it received back into the results file, to check the
-# harness passes them as <version> <category> <benchmark> <instance> ... <results_file>.
+# harness passes them as <version> <category> <benchmark> <instance> ... <results_file> <figures_dir>.
 ECHO_TOOL = (
     "#!/bin/sh\n"
-    'for a in "$@"; do last="$a"; done\n'
-    'printf "result,seen_version,seen_category,seen_benchmark,seen_instance\\n" > "$last"\n'
-    'printf "unknown,%s,%s,%s,%s\\n" "$1" "$2" "$3" "$4" >> "$last"\n'
+    'for a in "$@"; do csv_file="$last"; last="$a"; done\n'
+    'printf "result,seen_version,seen_category,seen_benchmark,seen_instance\\n" > "$csv_file"\n'
+    'printf "unknown,%s,%s,%s,%s\\n" "$1" "$2" "$3" "$4" >> "$csv_file"\n'
 )
 
 
@@ -85,9 +85,9 @@ def test_forwards_version_category_then_columns(tmp_path):
 def test_optional_timeout_column_caps_the_run(tmp_path):
     slow_tool = (
         "#!/bin/sh\n"
-        'for a in "$@"; do last="$a"; done\n'
+        'for a in "$@"; do csv_file="$last"; last="$a"; done\n'
         'sleep 2\n'
-        'printf "result\\nverified\\n" > "$last"\n'
+        'printf "result\\nverified\\n" > "$csv_file"\n'
     )
     repo = _repo(tmp_path / "repo", "benchmark,instance,timeout\nACC,slow,0.5\n")
     tool = _tool(tmp_path / "tool", slow_tool)