fix: enforce per-job process isolation

docs/architecture.md

@@ -71,9 +71,13 @@ That permits parallelism across unrelated PRs/issues while serializing a single
 Dispatch has two external side effects in live mode:
 
 1. apply GitHub 👀 reaction when possible;
-2. send one OpenClaw agent task with prompt rules and repository role context.
+2. run one local OpenClaw agent task with prompt rules and repository role
+   context.
 
 If dispatch fails, the job is marked `blocked`, `last_error` is stored, and the lock is released.
+Each attempt runs in its own transient systemd scope. Agent and tool descendants
+stay inside that scope, so the worker limit is a global concurrency boundary and
+job-level remediation cannot stop the executor or another worker.
 
 ## Prompt resources
 

docs/operations.md

@@ -277,29 +277,35 @@ the latest semantic heartbeat, visible OpenClaw output, and persisted
 CPU/I/O/PID-tree activity to decide whether an old running job looks stalled.
 Use `--progress-warn-seconds` to tune how long a running job can go without a
 semantic or visible progress update before the monitor considers it quiet.
-The alert wrapper uses the same composite stalled-job alert before automatic
-unlock or child termination. It does not unlock every old running job; it passes
-only the job ids that the monitor flagged as stalled.
+Stalled-job detection is alert-only: age or lack of visible progress is not
+proof that a process is orphaned, so it never terminates or unlocks a job.
 
 Each live dispatch also records its executor generation, worker id, root PID,
 parent PID, process group/session ids, and Linux process start time in the job
 metadata. The start time prevents PID reuse from making a dead job look alive.
-When the monitor finds a running job whose registered root process is dead,
-reparented, reused, or owned by another executor generation, it restarts the
-complete executor systemd cgroup and leaves the affected work blocked. This
-restart is deliberately broader than `killpg`: tools may create new process
-groups or sessions, but they remain in the service cgroup and are therefore
-terminated together. `KillMode=control-group` and `TimeoutStopSec=30s` in the
-executor unit make that cleanup explicit.
+Every attempt runs in a deterministic transient systemd scope named
+`github-agent-bridge-job-<job-id>-attempt-<attempt>.scope`. The monitor validates
+the exact unit name, cgroup path, PID, and process start time. If that ownership
+does not match, it stops only the validated job scope and marks only that job
+`blocked`. It never restarts the shared executor as job-level remediation. If
+scope metadata is absent or does not match the job and attempt, the monitor
+alerts and leaves the process and job untouched for manual inspection.
+
+Bridge jobs use OpenClaw's local embedded mode. Agent and tool processes
+therefore run in that job's scope instead of being owned and recoverable by the
+gateway. `KillMode=control-group` terminates the complete tree for one attempt,
+including tools that create their own process groups or sessions, without
+touching the executor or another worker. The configured worker count is also
+the global concurrency limit for bridge work. Normal local dispatches share a
+versioned per-thread OpenClaw session key that cannot collide with legacy
+gateway-dispatched runs. Explicit recovery retries receive an attempt-scoped
+rescue key so stale local session state cannot race or resume the new attempt.
 
 Set `GITHUB_AGENT_BRIDGE_KILL_STALE_CHILDREN=1` in the private systemd env file
-to let `github-agent-bridge-monitor-alert` terminate stale executor child
-process groups before retrying stalled jobs. The wrapper samples every direct
-executor child and its descendants, then only terminates children when the whole
-sample has been idle for `GITHUB_AGENT_BRIDGE_PROC_IDLE_SECONDS` seconds. It
-sends `SIGTERM`, waits `GITHUB_AGENT_BRIDGE_TERMINATE_GRACE_SECONDS`, and then
-uses `SIGKILL` if the child process group is still present. Keep this disabled
-unless the bridge host is allowed to clean up stuck OpenClaw runs automatically.
+only if legacy process-activity sampling is still required. This option no
+longer authorizes termination based on age or idleness; automatic remediation
+is restricted to an explicit ownership mismatch and the exact validated
+per-job scope.
 
 ## Dashboard API service
 

src/github_agent_bridge/dispatch.py

@@ -6,6 +6,7 @@
 import signal
 import subprocess
 import threading
+import time
 from dataclasses import dataclass
 from importlib import resources
 from enum import StrEnum
@@ -14,7 +15,8 @@
 from . import feedback
 from .models import GitHubContext, Job
 from .policy import DEFAULT_REPO_ROLE, Policy, Route, complexity_from_metadata
-from .process_inspection import process_stat
+from .job_isolation import is_expected_job_scope, is_job_scope_name, job_scope_unit
+from .process_inspection import cgroup_pids, process_stat
 from .session_correlation import (
     normalize_session_id,
     session_id_for_job,
@@ -480,6 +482,9 @@ def __init__(
         work_timeout_seconds: int = 3600,
         cli_grace_seconds: int = 60,
         feedback_db_path: str | None = None,
+        systemd_run_bin: str | None = "systemd-run",
+        systemctl_bin: str = "systemctl",
+        executor_unit: str = "github-agent-bridge.service",
     ):
         self.openclaw_bin = openclaw_bin
         self.node_bin = node_bin
@@ -490,10 +495,14 @@ def __init__(
         self.work_timeout_seconds = work_timeout_seconds
         self.cli_grace_seconds = cli_grace_seconds
         self.feedback_db_path = feedback_db_path
+        self.systemd_run_bin = systemd_run_bin
+        self.systemctl_bin = systemctl_bin
+        self.executor_unit = executor_unit
         self.mode = mode
         self._shutdown_event = threading.Event()
         self._process_lock = threading.Lock()
         self._active_processes: set[subprocess.Popen] = set()
+        self._active_units: set[str] = set()
 
     def shutdown(self, kill_grace_seconds: float = 5.0) -> None:
         """Cancel active CLI process groups and prevent new dispatches."""
@@ -502,6 +511,9 @@ def shutdown(self, kill_grace_seconds: float = 5.0) -> None:
         self._shutdown_event.set()
         with self._process_lock:
             processes = list(self._active_processes)
+            units = list(self._active_units)
+        for unit in units:
+            self.stop_job_scope(unit)
         for proc in processes:
             self._signal_process_group(proc, signal.SIGTERM)
         if processes and kill_grace_seconds >= 0:
@@ -529,6 +541,103 @@ def _kill_processes_after_grace(self, processes: list[subprocess.Popen], grace_s
         for proc in processes:
             self._signal_process_group(proc, signal.SIGKILL)
 
+    def stop_job_scope(self, unit: str) -> bool:
+        if not self.systemd_run_bin or not is_job_scope_name(unit):
+            return False
+        try:
+            result = subprocess.run(
+                [self.systemctl_bin, "--user", "stop", unit],
+                check=False,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                timeout=max(5, self.cli_grace_seconds),
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            return False
+        return result.returncode == 0
+
+    def stop_job(self, job: Job) -> bool:
+        runtime = job.metadata.get("runtime_process")
+        if not isinstance(runtime, dict):
+            return False
+        unit = str(runtime.get("unit") or "")
+        if not is_expected_job_scope(unit, job.id, job.attempts):
+            return False
+        return self.stop_job_scope(unit)
+
+    def _job_command(self, job: Job, agent_command: list[str], env: dict[str, str]) -> tuple[list[str], str | None]:
+        if not self.systemd_run_bin:
+            return agent_command, None
+        unit = job_scope_unit(job.id, job.attempts)
+        command = [
+            self.systemd_run_bin,
+            "--user",
+            "--scope",
+            "--quiet",
+            "--collect",
+            f"--unit={unit}",
+            "--property=KillMode=control-group",
+            f"--property=BindsTo={self.executor_unit}",
+            f"--property=PartOf={self.executor_unit}",
+            f"--property=After={self.executor_unit}",
+        ]
+        for key in (
+            "PATH",
+            "GITHUB_AGENT_BRIDGE_ACTION_MODE",
+            "GITHUB_AGENT_BRIDGE_WORK_INTENT",
+            "GITHUB_AGENT_BRIDGE_ALLOW_REPOSITORY_WRITE",
+            "GITHUB_AGENT_BRIDGE_ALLOW_PUSH",
+        ):
+            if key in env:
+                command.append(f"--setenv={key}={env[key]}")
+        command.extend(["--", *agent_command])
+        return command, unit
+
+    def _scope_process_identity(
+        self,
+        launcher_pid: int,
+        unit: str,
+        timeout_seconds: float = 10.0,
+    ) -> dict[str, int | str] | None:
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            try:
+                result = subprocess.run(
+                    [
+                        self.systemctl_bin,
+                        "--user",
+                        "show",
+                        unit,
+                        "--property=ControlGroup",
+                        "--value",
+                    ],
+                    check=False,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.DEVNULL,
+                    text=True,
+                    timeout=2,
+                )
+            except (OSError, subprocess.TimeoutExpired):
+                return None
+            control_group = (result.stdout or "").strip()
+            if result.returncode == 0 and control_group:
+                for pid in cgroup_pids(control_group):
+                    stat = process_stat(pid)
+                    if stat and stat.get("state") != "Z":
+                        return {
+                            "pid": pid,
+                            "ppid": int(stat["ppid"]),
+                            "pgid": int(stat["pgid"]),
+                            "sid": int(stat["sid"]),
+                            "start_time_ticks": int(stat["start_time_ticks"]),
+                            "launcher_pid": launcher_pid,
+                            "unit": unit,
+                            "control_group": control_group,
+                        }
+            threading.Event().wait(0.05)
+        return None
+
     def timeout_for(self, job: Job) -> int:
         if job.work_intent == "review_only":
             return self.review_timeout_seconds
@@ -600,10 +709,14 @@ def dispatch(
         policy: Policy,
         reaction_ok: bool | None = None,
         activity_callback: Callable[[str, str, str | None], None] | None = None,
-        process_callback: Callable[[dict[str, int]], bool | None] | None = None,
+        process_callback: Callable[[dict[str, int | str]], bool | None] | None = None,
     ) -> DispatchResult:
         agent, channel, to = self.route_for(job, policy)
-        cmd = [self.openclaw_bin, "agent"]
+        # Local embedded execution is a hard concurrency boundary: the OpenClaw
+        # process and every tool it starts remain descendants of this worker.
+        # Gateway-dispatched sessions can be recovered independently and would
+        # otherwise escape both the worker limit and this service's cgroup.
+        cmd = [self.openclaw_bin, "agent", "--local"]
         if agent:
             cmd += ["--agent", agent]
         model_route = policy.model_route_for(
@@ -618,11 +731,12 @@ def dispatch(
             cmd += ["--thinking", model_route.thinking]
         agent_timeout = self.timeout_for(job)
         fresh_session = bool(job.metadata.get("fresh_session_on_retry")) and job.attempts > 1
-        session_key = (
-            session_key_for_rescue(job.work_key, job.id, job.attempts)
-            if fresh_session
-            else session_key_for_work(job.work_key)
-        )
+        # Keep normal local dispatches on a stable per-thread key. Its versioned
+        # local namespace cannot collide with legacy gateway-dispatched runs;
+        # only an explicit recovery retry needs an attempt-scoped rescue key.
+        session_key = session_key_for_work(job.work_key)
+        if fresh_session:
+            session_key = session_key_for_rescue(job.work_key, job.id, job.attempts)
         if fresh_session or job.work_intent == "work_allowed":
             default_session_id = session_id_for_job_attempt(job.id, job.attempts)
             session_id = normalize_session_id(default_session_id)
@@ -653,31 +767,45 @@ def dispatch(
         env["GITHUB_AGENT_BRIDGE_ALLOW_PUSH"] = "1" if job.work_intent == "work_allowed" else "0"
         if self.node_bin:
             env["PATH"] = os.path.dirname(self.node_bin) + os.pathsep + env.get("PATH", "")
+        cmd, unit = self._job_command(job, cmd, env)
         if self.mode != RunMode.LIVE:
             return DispatchResult(True, 0, "side effects skipped", "", False, reaction_ok, cmd)
         with self._process_lock:
             if self._shutdown_event.is_set():
                 return DispatchResult(False, 130, "", "executor shutdown requested before dispatch", False, reaction_ok, cmd, True)
             proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, start_new_session=True)
             self._active_processes.add(proc)
+            if unit:
+                self._active_units.add(unit)
         if process_callback:
-            stat = process_stat(proc.pid)
-            identity = {
-                "pid": proc.pid,
-                "ppid": int(stat["ppid"]) if stat else os.getpid(),
-                "pgid": int(stat["pgid"]) if stat else proc.pid,
-                "sid": int(stat["sid"]) if stat else proc.pid,
-                "start_time_ticks": int(stat["start_time_ticks"]) if stat else 0,
-            }
+            if unit:
+                identity = self._scope_process_identity(proc.pid, unit)
+            else:
+                stat = process_stat(proc.pid)
+                identity = (
+                    {
+                        "pid": proc.pid,
+                        "ppid": int(stat["ppid"]),
+                        "pgid": int(stat["pgid"]),
+                        "sid": int(stat["sid"]),
+                        "start_time_ticks": int(stat["start_time_ticks"]),
+                    }
+                    if stat
+                    else None
+                )
             try:
-                registered = bool(stat) and process_callback(identity) is not False
+                registered = identity is not None and process_callback(identity) is not False
             except Exception:
                 registered = False
             if not registered:
+                if unit:
+                    self.stop_job_scope(unit)
                 self._signal_process_group(proc, signal.SIGKILL)
                 proc.wait()
                 with self._process_lock:
                     self._active_processes.discard(proc)
+                    if unit:
+                        self._active_units.discard(unit)
                 return DispatchResult(
                     False,
                     125,
@@ -715,9 +843,12 @@ def read_stream(stream, chunks: list[str], event_type: str) -> None:
                 stdout_thread.join(timeout=1)
                 stderr_thread.join(timeout=1)
                 out, err = "".join(stdout_chunks), "".join(stderr_chunks)
-                cancelled = self._shutdown_event.is_set() and proc.returncode != 0
-                return DispatchResult(proc.returncode == 0, proc.returncode, (out or "")[:2000], (err or "")[:4000], False, reaction_ok, cmd, cancelled)
+                cancelled = self._shutdown_event.is_set()
+                returncode = 130 if cancelled and proc.returncode == 0 else proc.returncode
+                return DispatchResult(not cancelled and returncode == 0, returncode, (out or "")[:2000], (err or "")[:4000], False, reaction_ok, cmd, cancelled)
             except subprocess.TimeoutExpired:
+                if unit:
+                    self.stop_job_scope(unit)
                 self._signal_process_group(proc, signal.SIGKILL)
                 proc.wait()
                 stdout_thread.join(timeout=1)
@@ -727,3 +858,5 @@ def read_stream(stream, chunks: list[str], event_type: str) -> None:
         finally:
             with self._process_lock:
                 self._active_processes.discard(proc)
+                if unit:
+                    self._active_units.discard(unit)