diff --git a/docs/architecture.md b/docs/architecture.md index 45982d3..40ce41e 100644 --- a/docs/architecture.md +++ b/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 diff --git a/docs/operations.md b/docs/operations.md index a3ca749..94c471b 100644 --- a/docs/operations.md +++ b/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--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 diff --git a/src/github_agent_bridge/dispatch.py b/src/github_agent_bridge/dispatch.py index a3166c3..824e8ec 100644 --- a/src/github_agent_bridge/dispatch.py +++ b/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,6 +767,7 @@ 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: @@ -660,24 +775,37 @@ def dispatch( 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) diff --git a/src/github_agent_bridge/executor.py b/src/github_agent_bridge/executor.py index 8949c38..3454a18 100644 --- a/src/github_agent_bridge/executor.py +++ b/src/github_agent_bridge/executor.py @@ -81,14 +81,14 @@ def work_one(self, worker_id: str | None = None) -> bool: ack_ok = self.github.react_ack_no_comment(job.context) summary = "non-actionable review; skipped dispatch" detail = f"eyes={reaction_ok} ack={ack_ok}" - self.queue.finish(job.id, "done", summary, detail) + self.queue.finish(job.id, "done", summary, detail, expected_locked_by=worker_id) return True if job.action == "reply_comment" and job.context.comment_id and not assigned_to_bot and not self.github.issue_comment_addresses_current_user(job.context): reaction_ok = self.react_eyes_for_job_contexts(job) ack_ok = self.github.react_ack_no_comment(job.context) summary = "comment not addressed to bot and bot not assigned; skipped dispatch" detail = f"eyes={reaction_ok} ack={ack_ok}" - self.queue.finish(job.id, "done", summary, detail) + self.queue.finish(job.id, "done", summary, detail, expected_locked_by=worker_id) return True if job.action == "reply_comment" and job.work_intent == "review_only" and (assigned_to_bot or authored_by_bot): reason = "PR/issue assigned to authenticated bot" if assigned_to_bot else "PR authored by authenticated bot" @@ -142,11 +142,11 @@ def work_one(self, worker_id: str | None = None) -> bool: if job.attempts <= self.config.missing_followup_retries: self.queue.requeue_running(job.id, "agent finished without visible GitHub follow-up; auto-requeued", detail) return True - self._finish(job, "blocked", summary, detail, notify_completion=True) + self._finish(job, worker_id, "blocked", summary, detail, notify_completion=True) return True summary = "👀 reaction ok + agent dispatch queued" if reaction_ok else "agent dispatch queued; reaction failed or unavailable" detail = f"followup_url={followup_url}; {result.detail}" if followup_url else result.detail - self._finish(job, "done", summary, detail, notify_completion=True, followup_url=followup_url) + self._finish(job, worker_id, "done", summary, detail, notify_completion=True, followup_url=followup_url) else: reason = ( "executor shutdown interrupted dispatch" @@ -159,7 +159,7 @@ def work_one(self, worker_id: str | None = None) -> bool: if followup_url: summary = "dispatch failed after producing visible GitHub follow-up" detail = f"followup_url={followup_url}; {reason}; {result.detail}" - self._finish(job, "blocked", summary, detail, notify_completion=True, followup_url=followup_url) + self._finish(job, worker_id, "blocked", summary, detail, notify_completion=True, followup_url=followup_url) return True if self._dispatch_failure_is_retryable(result) and job.attempts <= self.config.transient_dispatch_retries: self.queue.requeue_running( @@ -169,14 +169,15 @@ def work_one(self, worker_id: str | None = None) -> bool: fresh_session=self._dispatch_failure_needs_fresh_session(result), ) return True - self._finish(job, "blocked", reason, result.detail, notify_completion=True) + self._finish(job, worker_id, "blocked", reason, result.detail, notify_completion=True) except Exception as exc: - self._finish(job, "blocked", f"executor exception: {type(exc).__name__}", str(exc), notify_completion=dispatched) + self._finish(job, worker_id, "blocked", f"executor exception: {type(exc).__name__}", str(exc), notify_completion=dispatched) return True def _finish( self, job, + worker_id: str, status: str, summary: str, detail: str | None = None, @@ -184,7 +185,15 @@ def _finish( notify_completion: bool = False, followup_url: str | None = None, ) -> None: - self.queue.finish(job.id, status, summary, detail) + finished = self.queue.finish( + job.id, + status, + summary, + detail, + expected_locked_by=worker_id, + ) + if not finished: + return if not notify_completion: return actors = [actor for actor in [job.trigger_actor, *self.queue.coalesced_trigger_actors(job.id)] if actor] @@ -287,6 +296,10 @@ def run(self) -> None: worker_ids = [f"{self.executor_id}/worker-{i}" for i in range(worker_count)] threads: list[threading.Thread] = [] try: + stop_job = getattr(self.dispatcher, "stop_job", None) + if callable(stop_job): + for orphaned_job in self.queue.list_jobs(status="running", limit=1_000_000): + stop_job(orphaned_job) self.queue.block_running( "orphaned running job recovered at executor startup", "No prior executor process owns this running job. It was blocked, not auto-requeued, to avoid duplicate external actions.", diff --git a/src/github_agent_bridge/job_isolation.py b/src/github_agent_bridge/job_isolation.py new file mode 100644 index 0000000..f019bae --- /dev/null +++ b/src/github_agent_bridge/job_isolation.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import re + + +JOB_SCOPE_PATTERN = re.compile( + r"^github-agent-bridge-job-(?P\d+)-attempt-(?P\d+)\.scope$" +) + + +def job_scope_unit(job_id: int, attempt: int) -> str: + safe_attempt = max(1, int(attempt or 1)) + return f"github-agent-bridge-job-{int(job_id)}-attempt-{safe_attempt}.scope" + + +def is_expected_job_scope(unit: str, job_id: int, attempt: int) -> bool: + return unit == job_scope_unit(job_id, attempt) and JOB_SCOPE_PATTERN.fullmatch(unit) is not None + + +def is_job_scope_name(unit: str) -> bool: + return JOB_SCOPE_PATTERN.fullmatch(unit) is not None diff --git a/src/github_agent_bridge/monitor.py b/src/github_agent_bridge/monitor.py index 1dedf5a..fef8a85 100644 --- a/src/github_agent_bridge/monitor.py +++ b/src/github_agent_bridge/monitor.py @@ -12,8 +12,9 @@ from typing import Any from .dashboard_data import inspect_db_read_only +from .job_isolation import is_expected_job_scope from .observability import DEFAULT_PROCESS_SAMPLE_RETENTION_SECONDS, recent_process_samples, record_monitor_observation -from .process_inspection import direct_children, process_identity_matches +from .process_inspection import direct_children, process_cgroup, process_identity_matches ALERT_RELEASE_AVAILABLE = "monitor.release_available" @@ -89,7 +90,8 @@ def text(self) -> str: "- runtime detail: " f"job={job.get('id')} state={runtime.get('state', '-')} " f"pid={runtime.get('pid', '-')} ppid={runtime.get('ppid', '-')} " - f"pgid={runtime.get('pgid', '-')} sid={runtime.get('sid', '-')}" + f"pgid={runtime.get('pgid', '-')} sid={runtime.get('sid', '-')} " + f"unit={runtime.get('unit', '-')}" ) children = metrics.get("executor_children") or [] if children: @@ -290,11 +292,12 @@ def monitor( _add_alert(metrics, alerts, ALERT_EXECUTOR_SERVICE, f"executor service is {executor_state}") if metrics.get("running_jobs") and executor_state == "active" and not children: tracking_id = metrics.get("executor_process_tracking_id") - expects_live_process = not tracking_id or any( + expects_executor_child = not tracking_id or any( (job.get("runtime_process") or {}).get("state") == "running" + and not (job.get("runtime_process") or {}).get("unit") for job in metrics.get("running_jobs", []) ) - if expects_live_process: + if expects_executor_child: _add_alert( metrics, alerts, @@ -387,6 +390,26 @@ def _runtime_process_problem( start_time_ticks = int(runtime["start_time_ticks"]) except (KeyError, TypeError, ValueError): return "runtime process identity is incomplete" + unit = str(runtime.get("unit") or "") + if unit: + job_id = int(job.get("id") or 0) + attempt = int(job.get("attempts") or 0) + if not is_expected_job_scope(unit, job_id, attempt): + return f"runtime unit {unit} does not match job {job_id} attempt {attempt}" + if _is_active(unit) != "active": + return f"job scope {unit} is not active" + control_group = str(runtime.get("control_group") or "") + if not control_group: + return "runtime cgroup identity is incomplete" + if not process_identity_matches(pid, start_time_ticks): + return f"PID {pid} is dead, zombie, or reused" + actual_control_group = process_cgroup(pid) + if actual_control_group != control_group: + return ( + f"PID {pid} cgroup {actual_control_group or '-'} " + f"does not match {control_group}" + ) + return None if ppid != executor_pid: return f"runtime parent {ppid} does not match executor PID {executor_pid}" if not process_identity_matches(pid, start_time_ticks, expected_ppid=executor_pid): diff --git a/src/github_agent_bridge/monitor_alert.py b/src/github_agent_bridge/monitor_alert.py index c020076..af2de40 100644 --- a/src/github_agent_bridge/monitor_alert.py +++ b/src/github_agent_bridge/monitor_alert.py @@ -5,12 +5,14 @@ import os import re import signal +import sqlite3 import subprocess import time from dataclasses import dataclass from pathlib import Path from .monitor import ALERT_RUNNING_NO_EXECUTOR_CHILD, ALERT_RUNNING_PROCESS_MISMATCH +from .job_isolation import is_expected_job_scope from .observability import configure_sentry @@ -294,14 +296,7 @@ def terminate_process_group(pid: int, grace_seconds: int) -> str: def running_job_ids(output: str) -> list[str]: - ids = re.findall(r"running job (\d+)\b", output) - no_child_alert = ( - f"[{ALERT_RUNNING_NO_EXECUTOR_CHILD}]" in output - or "running jobs exist but executor has no child process" in output - ) - if no_child_alert: - ids.extend(re.findall(r"running detail: job=(\d+)\b", output)) - return list(dict.fromkeys(ids)) + return list(dict.fromkeys(re.findall(r"running job (\d+)\b", output))) def process_mismatch_job_ids(output: str) -> list[str]: @@ -310,7 +305,6 @@ def process_mismatch_job_ids(output: str) -> list[str]: return list( dict.fromkeys( re.findall(r"running job (\d+) process ownership mismatch", output) - + re.findall(r"running detail: job=(\d+)\b", output) ) ) @@ -341,23 +335,63 @@ def block_orphaned_jobs( return proc.stdout +def runtime_scope_for_job(config: AlertConfig, job_id: str) -> str | None: + try: + numeric_job_id = int(job_id) + con = sqlite3.connect(_expand(config.db)) + con.row_factory = sqlite3.Row + row = con.execute( + "SELECT attempts, metadata_json FROM jobs WHERE id=? AND status='running'", + (numeric_job_id,), + ).fetchone() + con.close() + except (OSError, sqlite3.Error, TypeError, ValueError): + return None + if row is None: + return None + try: + metadata = json.loads(row["metadata_json"] or "{}") + runtime = metadata.get("runtime_process") or {} + unit = str(runtime.get("unit") or "") + attempt = int(row["attempts"]) + except (AttributeError, TypeError, ValueError, json.JSONDecodeError): + return None + return unit if is_expected_job_scope(unit, numeric_job_id, attempt) else None + + def reconcile_process_mismatch(config: AlertConfig, job_ids: list[str]) -> str: - """Restart the whole executor cgroup so detached descendants cannot survive.""" - main_pid = get_main_pid(config.executor_unit) - restart_output = "" - if main_pid and main_pid != "0": - proc = _run(["systemctl", "--user", "restart", config.executor_unit]) - returncode = int(getattr(proc, "returncode", 0) or 0) - restart_output = f"executor cgroup restart rc={returncode}\n" - if proc.stdout: - restart_output += proc.stdout - blocked_output = block_orphaned_jobs( - config, - job_ids, - "The registered root process no longer matched this running job. The complete executor cgroup was restarted so detached descendants were stopped; the job was blocked and not auto-requeued.", - older_than_seconds=0, - ) - return restart_output + blocked_output + """Stop only mismatched job scopes; never restart the shared executor.""" + output: list[str] = [] + isolated_ids: list[str] = [] + for job_id in dict.fromkeys(job_ids): + unit = runtime_scope_for_job(config, job_id) + if not unit: + output.append( + f"job {job_id}: no validated per-job scope; left running for manual inspection" + ) + continue + before = (_run(["systemctl", "--user", "is-active", unit]).stdout or "").strip() + if before in {"active", "activating", "deactivating"}: + proc = _run(["systemctl", "--user", "stop", unit]) + if int(getattr(proc, "returncode", 0) or 0) != 0: + output.append(f"job {job_id}: failed to stop {unit}") + continue + after = (_run(["systemctl", "--user", "is-active", unit]).stdout or "").strip() + if after in {"active", "activating", "deactivating"}: + output.append(f"job {job_id}: {unit} remained {after}; job left running") + continue + isolated_ids.append(job_id) + output.append(f"job {job_id}: stopped isolated scope {unit}") + if isolated_ids: + blocked = block_orphaned_jobs( + config, + isolated_ids, + "The registered process no longer matched this running job. Its isolated job scope was stopped; the executor and other workers were left running. The job was blocked and not auto-requeued.", + older_than_seconds=0, + ) + if blocked: + output.append(blocked.rstrip()) + return "\n".join(output) + ("\n" if output else "") def maybe_unlock_stale(config: AlertConfig, output: str) -> str: @@ -374,23 +408,12 @@ def maybe_unlock_stale(config: AlertConfig, output: str) -> str: job_ids, "The executor service is inactive and no process owns this stale running job. It was not auto-requeued.", ) - child_output = "" - if has_child_processes(main_pid): - if not config.kill_stale_children: - return "" - sample_output = sample_executor_activity(config, main_pid=main_pid) - sample = load_proc_state(config.proc_state_file) - idle_seconds = int(sample.get("idle_seconds") or 0) - if sample.get("active_since_last_sample", True) or idle_seconds < config.proc_idle_seconds: - return sample_output - results = [terminate_process_group(pid, config.terminate_grace_seconds) for pid in child_pids(main_pid)] - child_output = sample_output + "terminated stale child processes:\n" + "\n".join(results) + "\n" - blocked_output = block_orphaned_jobs( - config, - job_ids, - "No live executor child owns this stale running job. It was blocked after process reconciliation and not auto-requeued.", - ) - return child_output + blocked_output + # Age or lack of visible progress is not proof that a process is orphaned. + # Keep these alerts observational while the executor is alive; automatic + # termination is reserved for an explicit runtime ownership mismatch above. + if config.kill_stale_children: + return sample_executor_activity(config, main_pid=main_pid) + return "" def load_state(path: Path) -> tuple[str, int]: diff --git a/src/github_agent_bridge/process_inspection.py b/src/github_agent_bridge/process_inspection.py index effd549..3dec8c7 100644 --- a/src/github_agent_bridge/process_inspection.py +++ b/src/github_agent_bridge/process_inspection.py @@ -5,6 +5,7 @@ PROC_ROOT = Path("/proc") +CGROUP_ROOT = Path("/sys/fs/cgroup") def process_exists(pid: int, proc_root: Path = PROC_ROOT) -> bool: @@ -57,6 +58,39 @@ def process_io(pid: int, proc_root: Path = PROC_ROOT) -> dict[str, int] | None: return values or None +def process_cgroup(pid: int, proc_root: Path = PROC_ROOT) -> str | None: + """Return the unified cgroup v2 path for a process.""" + try: + lines = (proc_root / str(pid) / "cgroup").read_text(encoding="utf-8").splitlines() + except OSError: + return None + for line in lines: + hierarchy, controllers, path = line.split(":", 2) + if hierarchy == "0" and controllers == "" and path.startswith("/"): + return path + return None + + +def cgroup_pids(control_group: str, cgroup_root: Path = CGROUP_ROOT) -> list[int]: + """List processes directly attached to an exact cgroup v2 path.""" + normalized = control_group.strip() + if not normalized.startswith("/") or ".." in normalized.split("/"): + return [] + try: + lines = (cgroup_root / normalized.lstrip("/") / "cgroup.procs").read_text( + encoding="utf-8" + ).splitlines() + except OSError: + return [] + pids: list[int] = [] + for line in lines: + try: + pids.append(int(line.strip())) + except ValueError: + continue + return sorted(set(pids)) + + def direct_child_pids(pid: int, proc_root: Path = PROC_ROOT) -> list[int]: children: list[int] = [] try: diff --git a/src/github_agent_bridge/queue.py b/src/github_agent_bridge/queue.py index d31d775..e056877 100644 --- a/src/github_agent_bridge/queue.py +++ b/src/github_agent_bridge/queue.py @@ -190,7 +190,7 @@ def register_runtime_process( job_id: int, worker_id: str, executor_id: str, - identity: dict[str, int], + identity: dict[str, int | str], ) -> bool: """Persist the exact process that owns a running job.""" now = utc_now() @@ -215,6 +215,10 @@ def register_runtime_process( "start_time_ticks": int(identity["start_time_ticks"]), "registered_at": now, } + for key in ("launcher_pid", "unit", "control_group"): + value = identity.get(key) + if value is not None: + runtime_process[key] = int(value) if key == "launcher_pid" else str(value) metadata["runtime_process"] = runtime_process con.execute( "UPDATE jobs SET metadata_json=?, updated_at=? WHERE id=? AND status='running' AND locked_by=?", @@ -259,29 +263,49 @@ def mark_runtime_process_exited(self, job_id: int, worker_id: str) -> bool: con.commit() return bool(cur.rowcount) - def finish(self, job_id: int, status: str, summary: str, detail: str | None = None) -> None: + def finish( + self, + job_id: int, + status: str, + summary: str, + detail: str | None = None, + *, + expected_locked_by: str | None = None, + ) -> bool: now = utc_now() with self.connect() as con: - row = con.execute("SELECT work_key FROM jobs WHERE id=?", (job_id,)).fetchone() + con.execute("BEGIN IMMEDIATE") + where = "id=?" + args: list[object] = [job_id] + if expected_locked_by is not None: + where += " AND status='running' AND locked_by=?" + args.append(expected_locked_by) + row = con.execute(f"SELECT work_key FROM jobs WHERE {where}", args).fetchone() + if row is None: + con.commit() + return False metadata = self._job_metadata(con, job_id) cancellation = metadata.get("cancellation") if isinstance(cancellation, dict) and cancellation.get("state") in {"requested", "cancelled"}: status = "blocked" summary = str(cancellation.get("summary") or summary) detail = str(cancellation.get("detail") or detail or "") - con.execute( - "UPDATE jobs SET status=?, last_error=?, locked_by=NULL, finished_at=COALESCE(finished_at, ?), updated_at=? WHERE id=?", - (status, detail if status == "blocked" else None, now, now, job_id), - ) + finished_at = "COALESCE(finished_at, ?)" else: - con.execute( - "UPDATE jobs SET status=?, last_error=?, locked_by=NULL, finished_at=?, updated_at=? WHERE id=?", - (status, detail if status == "blocked" else None, now, now, job_id), - ) + finished_at = "?" + cur = con.execute( + f"UPDATE jobs SET status=?, last_error=?, locked_by=NULL, finished_at={finished_at}, updated_at=? WHERE {where}", + (status, detail if status == "blocked" else None, now, now, *args), + ) + if not cur.rowcount: + con.commit() + return False self._log(con, job_id, row["work_key"] if row else None, status, summary, detail) session_id = metadata.get("openclaw_session_id") or session_id_for_job(job_id) self._session_event(con, job_id, row["work_key"] if row else None, str(session_id), status, summary, detail) self._progress(con, job_id, row["work_key"] if row else None, "semantic", status, summary, detail) + con.commit() + return True def request_cancel_running( self, diff --git a/src/github_agent_bridge/session_correlation.py b/src/github_agent_bridge/session_correlation.py index bd229f7..f293bcf 100644 --- a/src/github_agent_bridge/session_correlation.py +++ b/src/github_agent_bridge/session_correlation.py @@ -6,6 +6,7 @@ SESSION_ID_PREFIX = "github-agent-bridge-job" SESSION_KEY_PREFIX = "github-agent-bridge" +LOCAL_SESSION_KEY_NAMESPACE = "local-v2" SESSION_ID_PATTERN = re.compile(r"[^A-Za-z0-9_.:-]+") @@ -19,7 +20,7 @@ def session_id_for_job_attempt(job_id: int, attempt: int) -> str: def session_key_for_work(work_key: str) -> str: - return f"{SESSION_KEY_PREFIX}:{normalize_session_id(work_key)}" + return f"{SESSION_KEY_PREFIX}:{LOCAL_SESSION_KEY_NAMESPACE}:{normalize_session_id(work_key)}" def session_key_for_rescue(work_key: str, job_id: int, attempt: int) -> str: diff --git a/tests/test_job_isolation.py b/tests/test_job_isolation.py new file mode 100644 index 0000000..01d11f7 --- /dev/null +++ b/tests/test_job_isolation.py @@ -0,0 +1,21 @@ +from github_agent_bridge.job_isolation import ( + is_expected_job_scope, + is_job_scope_name, + job_scope_unit, +) + + +def test_job_scope_unit_is_deterministic_per_job_attempt(): + assert job_scope_unit(3025, 2) == ( + "github-agent-bridge-job-3025-attempt-2.scope" + ) + + +def test_job_scope_validation_rejects_other_jobs_and_arbitrary_units(): + unit = job_scope_unit(3025, 2) + + assert is_expected_job_scope(unit, 3025, 2) is True + assert is_expected_job_scope(unit, 3026, 2) is False + assert is_expected_job_scope(unit, 3025, 3) is False + assert is_job_scope_name("github-agent-bridge.service") is False + assert is_job_scope_name("github-agent-bridge-job-3025-attempt-2.scope;rm") is False diff --git a/tests/test_modes_cli.py b/tests/test_modes_cli.py index 7c8886b..bf833c5 100644 --- a/tests/test_modes_cli.py +++ b/tests/test_modes_cli.py @@ -77,19 +77,25 @@ def test_shadow_dispatch_returns_command_without_running(): result = OpenClawDispatcher(openclaw_bin="definitely-not-present", mode=RunMode.SHADOW).dispatch(make_job(), Policy(trusted_orgs={"gisce"}), reaction_ok=True) assert result.ok is True assert result.command + assert result.command[0] == "systemd-run" + assert "--scope" in result.command + assert "--unit=github-agent-bridge-job-1-attempt-1.scope" in result.command assert "agent" in result.command + assert "--local" in result.command assert "--model" not in result.command assert "--thinking" not in result.command assert "--session-id" in result.command assert result.command[result.command.index("--session-id") + 1] == "github-agent-bridge-job-1-attempt-1" assert "--session-key" in result.command - assert result.command[result.command.index("--session-key") + 1] == "github-agent-bridge:gisce-erp-1" + assert result.command[result.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) assert result.command[result.command.index("--verbose") + 1] == "on" assert "--timeout" in result.command assert "3600" in result.command -def test_work_allowed_dispatch_uses_fresh_session_id_per_job_attempt_with_stable_session_key(): +def test_work_allowed_dispatch_uses_fresh_session_id_and_stable_thread_key(): dispatcher = OpenClawDispatcher(openclaw_bin="definitely-not-present", mode=RunMode.SHADOW) policy = Policy(trusted_orgs={"gisce"}) first = dispatcher.dispatch(make_job(), policy, reaction_ok=True) @@ -102,9 +108,15 @@ def test_work_allowed_dispatch_uses_fresh_session_id_per_job_attempt_with_stable assert first.command[first.command.index("--session-id") + 1] == "github-agent-bridge-job-1-attempt-1" assert second.command[second.command.index("--session-id") + 1] == "github-agent-bridge-job-2-attempt-1" assert retry.command[retry.command.index("--session-id") + 1] == "github-agent-bridge-job-2-attempt-2" - assert first.command[first.command.index("--session-key") + 1] == "github-agent-bridge:gisce-erp-1" - assert first.command[first.command.index("--session-key") + 1] == second.command[second.command.index("--session-key") + 1] - assert second.command[second.command.index("--session-key") + 1] == retry.command[retry.command.index("--session-key") + 1] + assert first.command[first.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) + assert second.command[second.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) + assert retry.command[retry.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) def test_compaction_retry_uses_fresh_session_key_and_id_for_review_only_work(): @@ -118,7 +130,7 @@ def test_compaction_retry_uses_fresh_session_key_and_id_for_review_only_work(): assert result.command assert result.command[result.command.index("--session-id") + 1] == "github-agent-bridge-job-2-attempt-2" assert result.command[result.command.index("--session-key") + 1] == ( - "github-agent-bridge:gisce-erp-1:fresh:2:attempt:2" + "github-agent-bridge:local-v2:gisce-erp-1:fresh:2:attempt:2" ) @@ -134,7 +146,7 @@ def test_work_allowed_dispatch_ignores_legacy_session_id_metadata(): assert result.command[result.command.index("--session-id") + 1] == "github-agent-bridge-job-2-attempt-2" -def test_review_only_dispatch_session_key_remains_stable_for_same_github_thread(): +def test_review_only_dispatch_uses_stable_thread_key(): dispatcher = OpenClawDispatcher(openclaw_bin="definitely-not-present", mode=RunMode.SHADOW) policy = Policy(trusted_orgs={"gisce"}) first = dispatcher.dispatch(make_job("review_only"), policy, reaction_ok=True) @@ -144,8 +156,12 @@ def test_review_only_dispatch_session_key_remains_stable_for_same_github_thread( assert second.command assert first.command[first.command.index("--session-id") + 1] == "github-agent-bridge-job-1" assert second.command[second.command.index("--session-id") + 1] == "github-agent-bridge-job-2" - assert first.command[first.command.index("--session-key") + 1] == "github-agent-bridge:gisce-erp-1" - assert first.command[first.command.index("--session-key") + 1] == second.command[second.command.index("--session-key") + 1] + assert first.command[first.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) + assert second.command[second.command.index("--session-key") + 1] == ( + "github-agent-bridge:local-v2:gisce-erp-1" + ) def test_review_only_dispatch_uses_shorter_timeout(): @@ -211,7 +227,12 @@ def test_live_dispatch_streams_openclaw_output_to_activity_callback(tmp_path): events = [] processes = [] - result = OpenClawDispatcher(openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=1).dispatch( + result = OpenClawDispatcher( + openclaw_bin=str(openclaw), + mode=RunMode.LIVE, + cli_grace_seconds=1, + systemd_run_bin=None, + ).dispatch( make_job(), Policy(trusted_orgs={"gisce"}), reaction_ok=True, @@ -246,7 +267,12 @@ def test_live_dispatch_streams_partial_openclaw_output_before_process_exits(tmp_ monkeypatch.setenv("DONE_FILE", str(done)) callback_observed_done = [] - result = OpenClawDispatcher(openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=1).dispatch( + result = OpenClawDispatcher( + openclaw_bin=str(openclaw), + mode=RunMode.LIVE, + cli_grace_seconds=1, + systemd_run_bin=None, + ).dispatch( make_job(), Policy(trusted_orgs={"gisce"}), reaction_ok=True, @@ -272,7 +298,12 @@ def test_dispatcher_shutdown_terminates_active_process_group(tmp_path, monkeypat ) openclaw.chmod(0o755) monkeypatch.setenv("STARTED_FILE", str(started)) - dispatcher = OpenClawDispatcher(openclaw_bin=str(openclaw), mode=RunMode.LIVE, cli_grace_seconds=1) + dispatcher = OpenClawDispatcher( + openclaw_bin=str(openclaw), + mode=RunMode.LIVE, + cli_grace_seconds=1, + systemd_run_bin=None, + ) results = [] thread = threading.Thread( target=lambda: results.append(dispatcher.dispatch(make_job(), Policy(trusted_orgs={"gisce"}))), diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 8661218..4192a0c 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -240,6 +240,49 @@ def test_monitor_accepts_registered_job_process_identity(tmp_path, monkeypatch): assert "monitor.running_process_mismatch" not in report.metrics.get("alert_codes", []) +def test_monitor_accepts_process_in_matching_job_scope(tmp_path, monkeypatch): + db = tmp_path / "bridge.sqlite3" + q = JobQueue(db) + q.enqueue(notif(), Policy(trusted_orgs={"gisce"})) + executor_id = "executor-123-deadbeef" + worker_id = f"{executor_id}/worker-0" + job = q.claim_next(worker_id) + assert job is not None + unit = f"github-agent-bridge-job-{job.id}-attempt-{job.attempts}.scope" + control_group = f"/user.slice/{unit}" + q.register_runtime_process( + job.id, + worker_id, + executor_id, + { + "pid": 456, + "ppid": 789, + "pgid": 456, + "sid": 456, + "start_time_ticks": 999, + "launcher_pid": 789, + "unit": unit, + "control_group": control_group, + }, + ) + q.set_state("executor_process_tracking_id", executor_id) + monkeypatch.setattr(monitor_module, "_is_active", lambda unit: "active") + monkeypatch.setattr(monitor_module, "_main_pid", lambda unit: 123) + monkeypatch.setattr( + monitor_module, + "_direct_children", + lambda pid: [{"pid": 789, "cmd": "systemd-run --scope"}], + ) + monkeypatch.setattr(monitor_module, "_last_service_result", lambda unit: ("success", "0", 42)) + monkeypatch.setattr(monitor_module, "process_identity_matches", lambda *args, **kwargs: True) + monkeypatch.setattr(monitor_module, "process_cgroup", lambda pid: control_group) + + report = monitor(db) + + assert "monitor.running_process_mismatch" not in report.metrics.get("alert_codes", []) + assert f"unit={unit}" in report.text() + + def test_monitor_persists_process_samples_and_alert_state(tmp_path, monkeypatch): db = tmp_path / "bridge.sqlite3" q = JobQueue(db) diff --git a/tests/test_monitor_alert.py b/tests/test_monitor_alert.py index 81f3e72..e96a64e 100644 --- a/tests/test_monitor_alert.py +++ b/tests/test_monitor_alert.py @@ -1,8 +1,12 @@ from __future__ import annotations +from dataclasses import replace from pathlib import Path from github_agent_bridge import monitor_alert +from github_agent_bridge.models import Notification +from github_agent_bridge.policy import Policy +from github_agent_bridge.queue import JobQueue def make_config(tmp_path: Path) -> monitor_alert.AlertConfig: @@ -41,36 +45,27 @@ def test_load_state_accepts_legacy_shell_format(tmp_path): assert monitor_alert.load_state(state_file) == ("abc", 123) -def test_maybe_unlock_stale_blocks_when_executor_has_no_children(tmp_path, monkeypatch): +def test_maybe_unlock_stale_is_alert_only_without_per_job_kill_enabled(tmp_path, monkeypatch): config = make_config(tmp_path) calls = [] monkeypatch.setattr(monitor_alert, "get_main_pid", lambda unit="github-agent-bridge.service": "123") monkeypatch.setattr(monitor_alert, "has_child_processes", lambda pid: False) - def fake_run(args, check=False): - calls.append(args) - return type("Proc", (), {"stdout": '{"blocked":[7],"count":1}\n'})() - - monkeypatch.setattr(monitor_alert, "_run", fake_run) + monkeypatch.setattr(monitor_alert, "_run", lambda args, check=False: calls.append(args)) output = monitor_alert.maybe_unlock_stale(config, "running job 7 owner/repo#1 age 1200s > 900s") - assert output == '{"blocked":[7],"count":1}\n' - assert "block-running" in calls[0] - assert calls[0][-2:] == ["--job-id", "7"] + assert output == "" + assert calls == [] -def test_maybe_unlock_stale_blocks_no_child_running_detail_ids(tmp_path, monkeypatch): +def test_no_child_alert_does_not_treat_all_running_detail_rows_as_orphans(tmp_path, monkeypatch): config = make_config(tmp_path) calls = [] monkeypatch.setattr(monitor_alert, "get_main_pid", lambda unit="github-agent-bridge.service": "123") monkeypatch.setattr(monitor_alert, "has_child_processes", lambda pid: False) - def fake_run(args, check=False): - calls.append(args) - return type("Proc", (), {"stdout": '{"blocked":[568,570],"count":2}' + "\n"})() - - monkeypatch.setattr(monitor_alert, "_run", fake_run) + monkeypatch.setattr(monitor_alert, "_run", lambda args, check=False: calls.append(args)) output = monitor_alert.maybe_unlock_stale( config, @@ -83,12 +78,11 @@ def fake_run(args, check=False): ), ) - assert output == '{"blocked":[568,570],"count":2}\n' - assert "block-running" in calls[0] - assert calls[0][-4:] == ["--job-id", "568", "--job-id", "570"] + assert output == "" + assert calls == [] -def test_maybe_unlock_stale_uses_no_child_alert_code_for_detail_ids(tmp_path, monkeypatch): +def test_no_child_alert_code_does_not_expand_to_unrelated_running_details(tmp_path, monkeypatch): config = make_config(tmp_path) calls = [] monkeypatch.setattr(monitor_alert, "get_main_pid", lambda unit="github-agent-bridge.service": "123") @@ -110,12 +104,11 @@ def fake_run(args, check=False): ), ) - assert output == '{"blocked":[568],"count":1}\n' - assert "block-running" in calls[0] - assert calls[0][-2:] == ["--job-id", "568"] + assert output == "" + assert calls == [] -def test_maybe_unlock_stale_kills_children_and_blocks_jobs_when_enabled(tmp_path, monkeypatch): +def test_maybe_unlock_stale_never_stops_job_based_on_age_alone(tmp_path, monkeypatch): base = make_config(tmp_path) config = monitor_alert.AlertConfig( bridge_bin=base.bridge_bin, @@ -136,22 +129,18 @@ def test_maybe_unlock_stale_kills_children_and_blocks_jobs_when_enabled(tmp_path proc_idle_seconds=base.proc_idle_seconds, ) monkeypatch.setattr(monitor_alert, "get_main_pid", lambda unit="github-agent-bridge.service": "123") - monkeypatch.setattr(monitor_alert, "has_child_processes", lambda pid: True) - monkeypatch.setattr(monitor_alert, "child_pids", lambda pid: [456]) - monkeypatch.setattr(monitor_alert, "sample_executor_activity", lambda config, main_pid=None, now=None: "proc sample\n") - monkeypatch.setattr(monitor_alert, "load_proc_state", lambda path: {"active_since_last_sample": False, "idle_seconds": 300}) - monkeypatch.setattr(monitor_alert, "terminate_process_group", lambda pid, grace: f"pid {pid}: killed") - - def fake_run(args, check=False): - return type("Proc", (), {"stdout": "{\"blocked\":[7],\"count\":1}\n"})() - - monkeypatch.setattr(monitor_alert, "_run", fake_run) + calls = [] + monkeypatch.setattr(monitor_alert, "_run", lambda args, check=False: calls.append(args)) + monkeypatch.setattr( + monitor_alert, + "sample_executor_activity", + lambda config, main_pid=None, now=None: "proc sample\n", + ) output = monitor_alert.maybe_unlock_stale(config, "running job 7 owner/repo#1 age 1200s > 900s") - assert "pid 456: killed" in output - assert '{"blocked":[7],"count":1}' in output - assert "requeued" not in output + assert output == "proc sample\n" + assert calls == [] def test_maybe_unlock_stale_blocks_jobs_when_executor_is_inactive(tmp_path, monkeypatch): @@ -172,14 +161,18 @@ def fake_run(args, check=False): assert "unlock-stale" not in calls[0] -def test_process_mismatch_restarts_complete_executor_cgroup_and_blocks_without_age_gate(tmp_path, monkeypatch): +def test_process_mismatch_stops_only_job_scope_and_blocks_without_age_gate(tmp_path, monkeypatch): config = make_config(tmp_path) calls = [] - monkeypatch.setattr(monitor_alert, "get_main_pid", lambda unit="github-agent-bridge.service": "123") + unit = "github-agent-bridge-job-7-attempt-2.scope" + monkeypatch.setattr(monitor_alert, "runtime_scope_for_job", lambda config, job_id: unit) + active_states = iter(["active", "inactive"]) def fake_run(args, check=False): calls.append(args) - if "restart" in args: + if "is-active" in args: + return type("Proc", (), {"stdout": next(active_states) + "\n", "returncode": 0})() + if "stop" in args: return type("Proc", (), {"stdout": "", "returncode": 0})() return type("Proc", (), {"stdout": '{"blocked":[7],"count":1}\n', "returncode": 0})() @@ -194,13 +187,95 @@ def fake_run(args, check=False): ), ) - assert calls[0] == ["systemctl", "--user", "restart", "github-agent-bridge.service"] - assert "block-running" in calls[1] - assert calls[1][calls[1].index("--older-than") + 1] == "0" - assert "executor cgroup restart rc=0" in output + assert ["systemctl", "--user", "restart", "github-agent-bridge.service"] not in calls + assert calls[0] == ["systemctl", "--user", "is-active", unit] + assert calls[1] == ["systemctl", "--user", "stop", unit] + assert calls[2] == ["systemctl", "--user", "is-active", unit] + assert "block-running" in calls[3] + assert calls[3][calls[3].index("--older-than") + 1] == "0" + assert f"stopped isolated scope {unit}" in output assert '"blocked":[7]' in output +def test_process_mismatch_selects_only_explicitly_mismatched_job(): + output = "\n".join( + [ + "- [monitor.running_process_mismatch] running job 7 process ownership mismatch: PID 456 is dead", + "- running detail: job=7 key=owner/repo#1", + "- running detail: job=8 key=owner/repo#2", + ] + ) + + assert monitor_alert.process_mismatch_job_ids(output) == ["7"] + + +def test_process_mismatch_without_validated_scope_does_not_kill_or_block(tmp_path, monkeypatch): + config = make_config(tmp_path) + calls = [] + monkeypatch.setattr(monitor_alert, "runtime_scope_for_job", lambda config, job_id: None) + monkeypatch.setattr( + monitor_alert, + "_run", + lambda args, check=False: calls.append(args), + ) + + output = monitor_alert.reconcile_process_mismatch(config, ["7"]) + + assert calls == [] + assert "left running for manual inspection" in output + + +def test_runtime_scope_for_job_requires_exact_job_attempt_unit(tmp_path): + db = tmp_path / "bridge.sqlite3" + queue = JobQueue(db) + notification = Notification( + uid=1, + message_id="", + subject="Re: [owner/repo] Scope test", + from_addr="notifications@github.com", + body="@pilipilisbot https://github.com/owner/repo/issues/1#issuecomment-1", + auth={"spf": True, "dkim": True, "dmarc": True}, + ) + queued, _ = queue.enqueue( + notification, + Policy(trusted_orgs={"owner"}, bot_logins={"pilipilisbot"}), + ) + assert queued is not None + worker_id = "executor-123-deadbeef/worker-0" + job = queue.claim_next(worker_id) + assert job is not None + unit = f"github-agent-bridge-job-{job.id}-attempt-{job.attempts}.scope" + queue.register_runtime_process( + job.id, + worker_id, + "executor-123-deadbeef", + { + "pid": 456, + "ppid": 789, + "pgid": 456, + "sid": 456, + "start_time_ticks": 999, + "launcher_pid": 789, + "unit": unit, + "control_group": f"/user.slice/{unit}", + }, + ) + config = replace(make_config(tmp_path), db=str(db)) + + assert monitor_alert.runtime_scope_for_job(config, str(job.id)) == unit + + with queue.connect() as con: + row = con.execute("SELECT metadata_json FROM jobs WHERE id=?", (job.id,)).fetchone() + metadata = monitor_alert.json.loads(row["metadata_json"]) + metadata["runtime_process"]["unit"] = "github-agent-bridge-job-999-attempt-1.scope" + con.execute( + "UPDATE jobs SET metadata_json=? WHERE id=?", + (monitor_alert.json.dumps(metadata), job.id), + ) + + assert monitor_alert.runtime_scope_for_job(config, str(job.id)) is None + + def test_sample_executor_activity_tracks_all_executor_children(tmp_path, monkeypatch): config = make_config(tmp_path) monkeypatch.setattr(monitor_alert, "has_child_processes", lambda pid: True) diff --git a/tests/test_process_inspection.py b/tests/test_process_inspection.py index 9e787fa..a6ce877 100644 --- a/tests/test_process_inspection.py +++ b/tests/test_process_inspection.py @@ -2,7 +2,13 @@ from pathlib import Path -from github_agent_bridge.process_inspection import direct_children, inspect_process, process_identity_matches +from github_agent_bridge.process_inspection import ( + cgroup_pids, + direct_children, + inspect_process, + process_cgroup, + process_identity_matches, +) def write_proc(root: Path, pid: int, *, ppid: int, cmd: str, cpu_user: int = 1, cpu_system: int = 2, read_bytes: int = 3, write_bytes: int = 4) -> None: @@ -44,3 +50,26 @@ def test_process_identity_matches_pid_birth_and_parent(tmp_path): assert process_identity_matches(30, 3000, expected_ppid=10, proc_root=tmp_path) is True assert process_identity_matches(30, 2999, expected_ppid=10, proc_root=tmp_path) is False assert process_identity_matches(30, 3000, expected_ppid=11, proc_root=tmp_path) is False + + +def test_process_cgroup_reads_unified_v2_path(tmp_path): + proc = tmp_path / "40" + proc.mkdir() + (proc / "cgroup").write_text( + "1:name=systemd:/legacy\n" + "0::/user.slice/user-1000.slice/app.slice/job.scope\n", + encoding="utf-8", + ) + + assert process_cgroup(40, proc_root=tmp_path) == ( + "/user.slice/user-1000.slice/app.slice/job.scope" + ) + + +def test_cgroup_pids_reads_only_exact_valid_path(tmp_path): + group = tmp_path / "user.slice" / "job.scope" + group.mkdir(parents=True) + (group / "cgroup.procs").write_text("42\n41\n42\ninvalid\n", encoding="utf-8") + + assert cgroup_pids("/user.slice/job.scope", cgroup_root=tmp_path) == [41, 42] + assert cgroup_pids("../job.scope", cgroup_root=tmp_path) == [] diff --git a/tests/test_queue.py b/tests/test_queue.py index 1c268ae..315543d 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -634,3 +634,26 @@ def test_runtime_process_is_bound_to_running_job_worker(tmp_path): runtime = q.get(queued.id).metadata["runtime_process"] assert runtime["state"] == "exited" assert runtime["exited_at"].endswith("Z") + + +def test_worker_finish_cannot_overwrite_job_blocked_by_monitor(tmp_path): + q = JobQueue(tmp_path / "q.sqlite3") + queued, _ = q.enqueue(notif(1, "<1@github.com>", BODY1), policy()) + worker_id = "executor-123-deadbeef/worker-0" + claimed = q.claim_next(worker_id) + assert claimed is not None + + assert q.block_running("scope stopped", "isolated process mismatch", job_ids=[claimed.id]) == [ + claimed.id + ] + assert q.finish( + claimed.id, + "done", + "late worker result", + expected_locked_by=worker_id, + ) is False + + stored = q.get(queued.id) + assert stored is not None + assert stored.status == "blocked" + assert stored.last_error == "isolated process mismatch"