From c12df4dfb42f4d7221e082e001466eb159c15347 Mon Sep 17 00:00:00 2001 From: Jason Robert Date: Tue, 11 Aug 2026 15:46:10 -0400 Subject: [PATCH 1/2] fix(cli): exclude the run `stop` is executing inside (#399) An agent smoke-testing `conductor stop` from its own workflow's `bash` tool inherited that workflow's background environment and terminated itself -- the process printed "Stopped" and was killed by what it printed. Add `conductor.cli.self_run`, which identifies the run `stop` is executing inside via three signals, tried in order: `CONDUCTOR_RUN_ID` (set on every `--web-bg` child and inherited by descendants), the legacy `CONDUCTOR_WEB_BG`/`CONDUCTOR_WEB_PORT` pair (for PID files predating the `run_id` field), and POSIX process ancestry. That run is now excluded from targeting by default: - `--all` now means "stop all *other* runs" - the no-flag auto-stop skips it - `--port ` is refused (exit 1), naming the fix - a self-only `stop`/`stop --all` prints a refusal and exits 0 - `--allow-self` restores the previous behavior exactly, printing a warning whenever it actually causes the caller's own run to be signalled Process-ancestry detection is POSIX-only; Windows relies on the env-var signals alone. --- AGENTS.md | 1 + CHANGELOG.md | 17 ++ docs/cli-reference.md | 56 +++++- src/conductor/cli/app.py | 148 +++++++++++++++- src/conductor/cli/self_run.py | 209 ++++++++++++++++++++++ tests/test_cli/test_markup_injection.py | 4 + tests/test_cli/test_self_run.py | 187 ++++++++++++++++++++ tests/test_cli/test_stop.py | 220 ++++++++++++++++++++++-- 8 files changed, 816 insertions(+), 26 deletions(-) create mode 100644 src/conductor/cli/self_run.py create mode 100644 tests/test_cli/test_self_run.py diff --git a/AGENTS.md b/AGENTS.md index c2ab67a0..b1fd8f25 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -95,6 +95,7 @@ step-by-step checklist. - `run.py` - Workflow execution command with verbose logging helpers - `bg_runner.py` - Background process forking for `--web-bg` mode. Captures the detached child's stdout/stderr to `$TMPDIR/conductor/conductor---.bg.{stderr,stdout}.log` so silent crashes (uncaught Python exceptions, `faulthandler` dumps) leave a forensic trail — DEVNULL is **not** used for stdout/stderr. Passes `CONDUCTOR_RUN_ID`, `CONDUCTOR_BG_STDERR_LOG`, and `CONDUCTOR_BG_STDOUT_LOG` to the child via env so the child's `EventLogSubscriber` shares a run id with the bg log files and surfaces both paths in `workflow_started` system metadata. Returns a `BackgroundLaunch` dataclass (`url`, `stderr_log`, `stdout_log`, `run_id`). The launcher records that same `run_id` and both capture-log paths into the PID file (issue #404), rather than the launcher-invisible `run_id=""`/`log_file=""` defaults `write_pid_file` used to fall back to. `launch_background_resume` adopts the run id from the resolved checkpoint (`_checkpoint_run_id`, mirroring `cli/run.py`'s own checkpoint-resolution precedence) instead of minting a fresh one, so the PID file, `/api/info`, the events JSONL, and the capture-log filenames all agree on one id across a resume; a checkpoint with no usable id falls back to a fresh one. - `pid.py` - PID file utilities for tracking/stopping background processes. The PID file records `run_id`, `stderr_log`, and `stdout_log` from the launch (see `bg_runner.py`); a PID file written before this field existed has `run_id` as an empty string and lacks the `stderr_log`/`stdout_log` keys entirely (they didn't exist yet) — both cases surface as JSON `null` via `conductor status --json`. + - `self_run.py` - Answers "is this PID-file entry the run I am executing inside?" for `conductor stop`'s self-exclusion (issue #399: an agent smoke-testing `stop` must not terminate its own workflow). Three signals, first match wins: (1) `CONDUCTOR_RUN_ID` matching the entry's `run_id`; (2) `CONDUCTOR_WEB_BG`/`CONDUCTOR_WEB_PORT` matching the entry's port, but *only* when the entry records no `run_id` (the pre-#411 compatibility path — an entry with a present-but-different id never falls back to this signal); (3) process ancestry (`/proc//status` `PPid:` walk + `os.getsid(0)`), POSIX-only — Windows relies on signals 1–2 alone. `partition_own_run` splits PID-file entries into `others`/`own`; `stop` targets `others` unless `--allow-self` is passed. - `update.py` - Update check and version comparison. Upgrades are delegated to the install script (`install.ps1`/`install.sh`); in-process self-upgrade was removed because on Windows the running Python interpreter sits inside the venv `uv tool install --force` is trying to recreate, which fails with "Access is denied". `conductor update` prints the OS-appropriate install-script one-liner; `conductor update --apply` spawns the installer detached (Windows: new console window; POSIX: `os.execvpe` replace) and exits the current process so file locks release. The startup hint is suppressed by `CONDUCTOR_NO_UPDATE_CHECK=1`, `--silent`, `--help`/`--version`, and the `update` subcommand itself. - **config/**: YAML loading and Pydantic schema validation diff --git a/CHANGELOG.md b/CHANGELOG.md index 298f4bdd..a3403841 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -167,6 +167,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`conductor stop` no longer kills the run it is executing inside** (#399). + An agent smoke-testing `conductor stop` from its own workflow's `bash` tool + inherited that workflow's background environment and terminated itself — + the process printed "Stopped" and was killed by what it printed. `stop` now + identifies the run it is executing inside via `CONDUCTOR_RUN_ID` (set on + every `--web-bg` child and inherited by descendants), the legacy + `CONDUCTOR_WEB_BG`/`CONDUCTOR_WEB_PORT` pair (for PID files predating + #411's `run_id` field), and POSIX process ancestry, and excludes it from + targeting by default: `--all` now means "stop all *other* runs", the + no-flag auto-stop skips it, and `--port ` is refused (exit + `1`, naming the fix). If only your own run is alive, `stop`/`stop --all` + print a refusal and exit `0` rather than erroring, since nothing named was + declined. Pass `--allow-self` to restore the previous behavior exactly; a + yellow warning is printed whenever it actually causes your own run to be + signalled. Process-ancestry detection is POSIX-only — Windows relies on + the env-var signals alone. + - **Bracketed text no longer crashes or corrupts CLI output** (#406). The same defect as #382, which #387 fixed only in `cli/run.py`. `conductor validate` died with an unhandled `MarkupError` traceback on a workflow whose `name:` diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 50c3d91c..29b5e3b6 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -305,7 +305,8 @@ conductor stop [OPTIONS] | Option | Description | |--------|-------------| | `--port PORT` | Stop the workflow running on this specific port | -| `--all` | Stop all background conductor workflows | +| `--all` | Stop all background conductor workflows (all *other* runs; see Self-Exclusion) | +| `--allow-self` | Include the run this command is executing inside (refused by default) | With no options, `conductor stop` lists running background workflows. If exactly one is found, it stops automatically. If multiple are running, it prints the list and asks you to specify `--port`. @@ -324,6 +325,54 @@ The web dashboard also exposes terminate controls that always preserve progress: shows a **"Workflow Stopped"** banner with the checkpoint path (or a clear explanation if no checkpoint could be saved). +### Self-Exclusion + +`conductor stop` never targets the run it is executing inside by default — an +agent smoke-testing this command must not terminate its own workflow (issue +#399). This matters because a background workflow's `bash` tool inherits the +parent process's environment, so an agent-invoked `conductor stop` can easily +end up pointed at the very run driving it. + +The caller's own run is identified by three signals, tried in order (first +match wins): + +1. **`CONDUCTOR_RUN_ID`** env var matching the PID file's `run_id`. Set on + every `--web-bg` child (and inherited by its descendants, including a + spawned `conductor stop`). +2. **`CONDUCTOR_WEB_BG=1` + `CONDUCTOR_WEB_PORT`** matching the entry's port — + a compatibility signal used only for PID files written before `run_id` + existed (empty `run_id`). +3. **Process ancestry** — a `/proc//status` `PPid:` walk plus a session-id + check, so any descendant of the background process (however it was + re-parented) still resolves to it. + +Effects: + +- `conductor stop` (no flags) and `conductor stop --all` never stop your own + run; `--all` means "stop all *other* runs." If only your own run is alive, + both print a refusal and exit `0` (nothing was requested by name and + nothing failed). +- `conductor stop --port ` is refused and exits `1`, naming + `--allow-self` as the remedy — here a specific target was named and + declined. +- `conductor stop --allow-self [...]` restores the pre-#399 behavior exactly, + printing a yellow warning when the run actually being stopped is your own. + +**Windows caveat**: process ancestry (signal 3) is POSIX-only. On Windows, +self-identification relies solely on the `CONDUCTOR_RUN_ID` / +`CONDUCTOR_WEB_BG`+`CONDUCTOR_WEB_PORT` env vars — an agent whose tool runner +strips `CONDUCTOR_*` env vars before spawning its shell is still exposed. + +See [`conductor status`](#conductor-status) for a non-destructive way to see +the full list of running workflows, including your own. + +### Exit Codes + +| Code | Meaning | +|------|---------| +| `0` | Stopped/listed successfully, including a self-only refusal | +| `1` | `--port` matched nothing, or matched only your own run | + ### Examples ```bash @@ -333,8 +382,11 @@ conductor stop # Stop a specific workflow by port conductor stop --port 8080 -# Stop all running background workflows +# Stop all other running background workflows conductor stop --all + +# Include the run this command is executing inside +conductor stop --allow-self --port 8080 ``` ## `conductor gate respond` diff --git a/src/conductor/cli/app.py b/src/conductor/cli/app.py index 79caed2c..0fbe032f 100644 --- a/src/conductor/cli/app.py +++ b/src/conductor/cli/app.py @@ -10,7 +10,7 @@ import os from enum import Enum from pathlib import Path -from typing import Annotated, Any +from typing import TYPE_CHECKING, Annotated, Any import typer from rich.console import Console @@ -21,6 +21,11 @@ from conductor.console import make_console, styled from conductor.exceptions import WorkflowTerminated +if TYPE_CHECKING: + # Typing-only: ``stop()`` imports ``conductor.cli.self_run`` lazily at + # runtime, matching the existing lazy import of ``conductor.cli.pid``. + from conductor.cli.self_run import OwnRunPartition + logger = logging.getLogger(__name__) @@ -1260,6 +1265,13 @@ def stop( help="Stop all background conductor workflows.", ), ] = False, + allow_self: Annotated[ + bool, + typer.Option( + "--allow-self", + help="Include the run this command is executing inside (refused by default).", + ), + ] = False, ) -> None: """Stop background workflow processes launched with --web-bg. @@ -1267,13 +1279,29 @@ def stop( is found, stops it automatically. If multiple are found, prints the list and asks you to specify --port. + \b + By default, ``stop`` never targets the run it is executing inside -- + an agent smoke-testing this command must not terminate its own + workflow (issue #399). That run is identified by ``CONDUCTOR_RUN_ID``, + the legacy ``CONDUCTOR_WEB_BG``/``CONDUCTOR_WEB_PORT`` pair, or process + ancestry, and is excluded from ``--all`` and the no-flag auto-stop; a + ``--port`` naming it is refused outright. Pass ``--allow-self`` to + include it anyway. + + \b + Exit codes: + 0 stopped/listed successfully, including a self-only refusal + 1 --port matched nothing, or matched only your own run + \b Examples: conductor stop conductor stop --port 8080 conductor stop --all + conductor stop --allow-self --port 8080 """ from conductor.cli.pid import read_pid_files, remove_pid_file + from conductor.cli.self_run import partition_own_run running = read_pid_files() @@ -1281,36 +1309,61 @@ def stop( console.print(Text.from_markup("[dim]No background workflows are currently running.[/dim]")) return + partition = partition_own_run(running) + targetable = running if allow_self else partition.others + if all_workflows: - for entry in running: + if not allow_self and not targetable: + _print_self_exclusion(partition, console, blocking=True) + return + if not allow_self and partition.own: + _print_self_exclusion(partition, console, blocking=False) + for entry in targetable: + _maybe_warn_stopping_self(entry, partition, console) _stop_process(entry, console) remove_pid_file(entry["port"]) return if port is not None: # Find the entry for the specified port - match = [e for e in running if e["port"] == port] + match = [e for e in targetable if e["port"] == port] if not match: + if not allow_self: + own_match = [e for e in partition.own if e["port"] == port] + if own_match: + _print_self_refusal_line(own_match[0], console) + _print_allow_self_hint(console) + raise typer.Exit(code=1) console.print( styled("[bold red]Error:[/bold red] No background workflow found on port {}.", port) ) - console.print(Text.from_markup("[dim]Running workflows:[/dim]")) - _print_running_list(running, console) + if not allow_self and not targetable and partition.own: + _print_self_exclusion(partition, console, blocking=False) + else: + console.print(Text.from_markup("[dim]Running workflows:[/dim]")) + _print_running_list(targetable, console) raise typer.Exit(code=1) + _maybe_warn_stopping_self(match[0], partition, console) _stop_process(match[0], console) remove_pid_file(port) return # No flags: auto-stop if exactly one, otherwise list - if len(running) == 1: - entry = running[0] + if len(targetable) == 0: + _print_self_exclusion(partition, console, blocking=True) + return + if len(targetable) == 1: + entry = targetable[0] + _maybe_warn_stopping_self(entry, partition, console) _stop_process(entry, console) remove_pid_file(entry["port"]) + if not allow_self and partition.own: + _print_self_exclusion(partition, console, blocking=False) else: console.print( styled( "[bold yellow]Multiple background workflows running ({}).[/bold yellow]", - len(running), + len(targetable), ) ) console.print( @@ -1318,7 +1371,84 @@ def stop( "[dim]Specify --port to stop a specific one, or --all to stop all.[/dim]\n" ) ) - _print_running_list(running, console) + _print_running_list(targetable, console) + if not allow_self and partition.own: + _print_self_exclusion(partition, console, blocking=False) + + +def _print_self_refusal_line(entry: dict, con: Console) -> None: + """Print the red refusal line naming the run this command is executing inside. + + Args: + entry: The PID-file dict identified as this process's own run. + con: Rich Console for output. + """ + from conductor.cli.self_run import describe_own_run + + con.print( + styled( + "[bold red]Refusing[/bold red] to stop run {} — it is the run this " + "command is executing inside.", + describe_own_run(entry), + ) + ) + + +def _print_allow_self_hint(con: Console) -> None: + """Print the dim hint pointing at the ``--allow-self`` escape hatch.""" + con.print(Text.from_markup("[dim]Use --allow-self to include it.[/dim]")) + + +def _print_self_exclusion(partition: OwnRunPartition, con: Console, *, blocking: bool) -> None: + """Print the message explaining that this run was excluded from targeting. + + Args: + partition: The result of ``partition_own_run``. ``partition.own`` + must be non-empty. + con: Rich Console for output. + blocking: True when there is nothing left to stop (prints the red + refusal line plus "No other workflows are running."); False when + other runs were still targeted (prints a yellow exclusion note). + """ + entry = partition.own[0] + if blocking: + _print_self_refusal_line(entry, con) + con.print(Text.from_markup("[dim]No other workflows are running.[/dim]")) + else: + from conductor.cli.self_run import describe_own_run + + con.print( + styled( + "[yellow]Excluded[/yellow] run {} — it is the run this command is " + "executing inside.", + describe_own_run(entry), + ) + ) + _print_allow_self_hint(con) + + +def _maybe_warn_stopping_self(entry: dict, partition: OwnRunPartition, con: Console) -> None: + """Print a yellow warning when about to signal the caller's own run. + + Only reachable via ``--allow-self`` -- without that flag, an entry + identified as this process's own run is never present in the + targetable list in the first place. + + Args: + entry: The PID-file dict about to be stopped. + partition: The result of ``partition_own_run``. + con: Rich Console for output. + """ + from conductor.cli.self_run import describe_own_run + + if any(o["port"] == entry["port"] for o in partition.own): + con.print( + styled( + "[yellow]Warning:[/yellow] stopping run {} — this is the run " + "executing this command.", + describe_own_run(entry), + ) + ) def _stop_process(entry: dict, con: Console) -> None: diff --git a/src/conductor/cli/self_run.py b/src/conductor/cli/self_run.py new file mode 100644 index 00000000..18d411e6 --- /dev/null +++ b/src/conductor/cli/self_run.py @@ -0,0 +1,209 @@ +"""Identify whether a PID-file entry is the run this process executes inside. + +Issue #399: an agent smoke-testing ``conductor stop`` sent it against the +background workflow that was, itself, executing the agent — the process +printed a "Stopped" message and was killed by the very thing it printed. This +module answers one question — *is this PID-file entry the run I am currently +running inside?* — so ``cli/app.py::stop`` can exclude that entry from +targeting by default. + +Three independent signals are tried, in order (first match wins per entry): + +1. **``CONDUCTOR_RUN_ID`` env var** matches the PID file's ``run_id``. + ``cli/bg_runner.py::_build_bg_env`` sets this on the detached background + child, so every descendant of it — including an agent's ``bash`` tool and + the ``conductor stop`` it spawns — inherits it, and + ``_finalize_background_launch`` writes the same id into the PID file's + ``run_id`` key (added in issue #411). +2. **``CONDUCTOR_WEB_BG=1`` + ``CONDUCTOR_WEB_PORT``** matching the entry's + ``port``, used *only* when the entry records no ``run_id`` (i.e. ``""``). + This is the compatibility path for PID files written before #411. Limiting + it to entries with no recorded id means an entry whose id is present and + *different* from ours is never misidentified as self. +3. **Process ancestry.** A Linux ``/proc//status`` ``PPid:`` walk + upward from ``os.getpid()``, plus a POSIX ``os.getsid(0)`` check. The + session check is precise for background runs specifically because + ``bg_runner._detachment_kwargs()`` passes ``start_new_session=True``, + making the bg child a session leader whose session id equals its own pid + — so any descendant, even one re-parented away from the direct ancestry + chain, still resolves to it. + +``os.getpid()`` itself is always in the identity set: a PID file naming the +very process running ``stop`` is definitionally self, and SIGTERM-ing +yourself is never the right behaviour. + +**Windows caveat**: signal 3 (process ancestry) is POSIX-only. On Windows, +self-identification relies solely on signals 1 and 2 (the env vars). This is +a deliberate limitation rather than an implementation gap — a +``CreateToolhelp32Snapshot``-based ancestry walk would be unexercised by +conductor's ubuntu-only CI, so it is documented here instead: an agent on +Windows whose tool runner strips ``CONDUCTOR_*`` env vars before spawning +its shell is still exposed to this issue. +""" + +from __future__ import annotations + +import contextlib +import logging +import os +from dataclasses import dataclass +from pathlib import Path + +logger = logging.getLogger(__name__) + +# Mirrors the names ``cli/bg_runner.py::_build_bg_env`` writes into the +# background child's environment. +RUN_ID_ENV = "CONDUCTOR_RUN_ID" +WEB_BG_ENV = "CONDUCTOR_WEB_BG" +WEB_PORT_ENV = "CONDUCTOR_WEB_PORT" + +# Bounds the ``/proc`` ancestry walk so a malformed or cyclic ``PPid`` chain +# cannot loop indefinitely. +_MAX_ANCESTRY_HOPS = 64 + + +def _read_ppid(pid: int) -> int | None: + """Return the parent PID of *pid* by reading ``/proc//status``. + + This is the one seam tests monkeypatch to drive :func:`own_run_pids`' + ancestry walk deterministically, independent of the real process tree. + + Args: + pid: The process ID to look up. + + Returns: + The parent PID, or ``None`` if it cannot be determined (the file + does not exist, is unreadable, or has no parseable ``PPid:`` line). + """ + try: + status = Path(f"/proc/{pid}/status").read_text() + except OSError: + return None + + for line in status.splitlines(): + if line.startswith("PPid:"): + try: + return int(line.split(":", 1)[1].strip()) + except ValueError: + return None + return None + + +def own_run_pids() -> frozenset[int]: + """Return the set of PIDs that identify "this process" for self-exclusion. + + Includes ``os.getpid()`` unconditionally, then walks ``/proc//status`` + ``PPid:`` links upward (bounded by :data:`_MAX_ANCESTRY_HOPS`, with a + ``seen`` set so a malformed cyclic chain also terminates), then adds + ``os.getsid(0)`` when available. + + Deliberately not memoized: a ``stop`` invocation calls this once, and + caching would force every test to remember to clear it. + + Returns: + A frozen set of PIDs considered to be "this run" for the purpose of + excluding a PID-file entry from ``conductor stop`` targeting. + """ + pids: set[int] = {os.getpid()} + + if Path("/proc/self/status").exists(): + seen: set[int] = set() + current = os.getpid() + for _ in range(_MAX_ANCESTRY_HOPS): + if current in seen: + break + seen.add(current) + parent = _read_ppid(current) + if parent is None or parent in pids: + break + pids.add(parent) + current = parent + + if hasattr(os, "getsid"): + with contextlib.suppress(OSError): + pids.add(os.getsid(0)) + + return frozenset(pids) + + +@dataclass(frozen=True) +class OwnRunPartition: + """The result of classifying PID-file entries against this process's identity.""" + + others: list[dict] + """Entries that are not this run — the only ones ``stop`` should target by default.""" + + own: list[dict] + """Entries identified as this run.""" + + reasons: dict[int, str] + """Why an entry (keyed by port) was classified as own: ``"run id"``, + ``"dashboard port"``, or ``"process ancestry"``. Logged at debug so a + false positive is diagnosable without being printed to the user.""" + + +def partition_own_run(entries: list[dict]) -> OwnRunPartition: + """Split PID-file entries into "others" and "own" (this process's run). + + Computes this process's identity (:func:`own_run_pids` plus the bg-launch + env vars) once, then classifies each entry against it in order: + ``run_id`` match, then the legacy dashboard-port compatibility signal + (only for entries with no recorded ``run_id``), then process ancestry. + + Args: + entries: PID-file dicts, each with at least ``pid`` and ``port`` + (and typically ``run_id``, ``workflow``). + + Returns: + An :class:`OwnRunPartition` with ``others``/``own`` preserving the + input order, and a ``reasons`` map for diagnostics. + """ + my_pids = own_run_pids() + my_run_id = os.environ.get(RUN_ID_ENV, "") + web_bg = os.environ.get(WEB_BG_ENV) == "1" + my_web_port = os.environ.get(WEB_PORT_ENV, "") + + others: list[dict] = [] + own: list[dict] = [] + reasons: dict[int, str] = {} + + for entry in entries: + port = entry.get("port") + entry_run_id = entry.get("run_id") or "" + reason: str | None = None + + if my_run_id and entry_run_id and entry_run_id.lower() == my_run_id.lower(): + reason = "run id" + elif not entry_run_id and web_bg and my_web_port and str(port) == my_web_port: + reason = "dashboard port" + elif entry.get("pid") in my_pids: + reason = "process ancestry" + + if reason is not None: + own.append(entry) + if isinstance(port, int): + reasons[port] = reason + logger.debug("PID-file entry on port %s identified as own run (%s)", port, reason) + else: + others.append(entry) + + return OwnRunPartition(others=others, own=own, reasons=reasons) + + +def describe_own_run(entry: dict) -> str: + """Return the identity fragment used to name the caller's own run in messages. + + Args: + entry: The PID-file dict identified as this process's own run. + + Returns: + The ``run_id`` when the PID file records one, otherwise + ``" (port N)"``. Always a plain ``str`` — never a + ``Text`` — since callers interpolate it through ``styled()`` (markup + guard rule F). + """ + run_id = entry.get("run_id") + if run_id: + return str(run_id) + workflow = Path(entry.get("workflow", "unknown")).stem + return f"{workflow} (port {entry.get('port')})" diff --git a/tests/test_cli/test_markup_injection.py b/tests/test_cli/test_markup_injection.py index 66c63c70..5ff22910 100644 --- a/tests/test_cli/test_markup_injection.py +++ b/tests/test_cli/test_markup_injection.py @@ -148,6 +148,10 @@ def pid_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: runs = tmp_path / "runs" runs.mkdir() monkeypatch.setattr("conductor.cli.pid.pid_dir", lambda: runs) + # Issue #399: the hardcoded ``pid: 610745`` below must never be + # misidentified as this test process's own run by a coincidental + # ancestor PID, which would make this markup test flaky. + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) return runs def _write_pid(self, pid_dir: Path, stem: str) -> None: diff --git a/tests/test_cli/test_self_run.py b/tests/test_cli/test_self_run.py new file mode 100644 index 00000000..264269df --- /dev/null +++ b/tests/test_cli/test_self_run.py @@ -0,0 +1,187 @@ +"""Tests for ``conductor.cli.self_run`` (issue #399). + +Covers: +- ``own_run_pids()`` identity signals (self, ancestry, session) +- ``_read_ppid`` ancestry-walk driving via ``own_run_pids`` +- ``partition_own_run`` classification against each of the three signals +- ``describe_own_run`` identity formatting +""" + +from __future__ import annotations + +import os +import sys + +import pytest + +from conductor.cli.self_run import ( + _MAX_ANCESTRY_HOPS, + RUN_ID_ENV, + WEB_BG_ENV, + WEB_PORT_ENV, + describe_own_run, + own_run_pids, + partition_own_run, +) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure the bg-launch identity env vars don't leak in from the real environment.""" + monkeypatch.delenv(RUN_ID_ENV, raising=False) + monkeypatch.delenv(WEB_BG_ENV, raising=False) + monkeypatch.delenv(WEB_PORT_ENV, raising=False) + + +class TestOwnRunPids: + """Tests for the real (unmocked) ``own_run_pids()``.""" + + def test_contains_own_pid_on_every_platform(self) -> None: + assert os.getpid() in own_run_pids() + + @pytest.mark.skipif( + not sys.platform.startswith("linux"), reason="ancestry walk is /proc-based (Linux only)" + ) + def test_contains_parent_pid_on_linux(self) -> None: + assert os.getppid() in own_run_pids() + + @pytest.mark.skipif(sys.platform == "win32", reason="os.getsid is POSIX-only") + def test_contains_session_id_on_posix(self) -> None: + assert os.getsid(0) in own_run_pids() + + +@pytest.mark.skipif( + not sys.platform.startswith("linux"), reason="ancestry walk is /proc-based (Linux only)" +) +class TestAncestryWalk: + """Drives ``own_run_pids()``'s ``/proc`` walk via a monkeypatched ``_read_ppid``.""" + + def test_multi_hop_chain_fully_collected(self, monkeypatch: pytest.MonkeyPatch) -> None: + me = os.getpid() + chain = {me: 100_001, 100_001: 100_002, 100_002: 100_003, 100_003: None} + monkeypatch.setattr("conductor.cli.self_run._read_ppid", lambda pid: chain.get(pid)) + + pids = own_run_pids() + + assert {me, 100_001, 100_002, 100_003}.issubset(pids) + + def test_ppid_cycle_terminates(self, monkeypatch: pytest.MonkeyPatch) -> None: + me = os.getpid() + chain = {me: 200_001, 200_001: 200_002, 200_002: 200_001} + monkeypatch.setattr("conductor.cli.self_run._read_ppid", lambda pid: chain.get(pid)) + + # Must return promptly rather than looping forever. + pids = own_run_pids() + + assert {me, 200_001, 200_002}.issubset(pids) + + def test_max_ancestry_hops_cap_holds(self, monkeypatch: pytest.MonkeyPatch) -> None: + me = os.getpid() + # Build a chain longer than the hop cap, all distinct pids. + chain_len = _MAX_ANCESTRY_HOPS + 20 + chain: dict[int, int | None] = {} + prev = me + for i in range(chain_len): + nxt = 300_000 + i + chain[prev] = nxt + prev = nxt + chain[prev] = None + monkeypatch.setattr("conductor.cli.self_run._read_ppid", lambda pid: chain.get(pid)) + + pids = own_run_pids() + + # The walk is bounded, so it cannot have collected the entire chain. + assert len(pids) <= _MAX_ANCESTRY_HOPS + 2 # +1 for self, +1 slack for getsid + + def test_unreadable_proc_stops_without_raising(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("conductor.cli.self_run._read_ppid", lambda pid: None) + + pids = own_run_pids() + + assert os.getpid() in pids + + +def _entry(pid: int, port: int, run_id: str = "", workflow: str = "/tmp/wf.yaml") -> dict: + return {"pid": pid, "port": port, "run_id": run_id, "workflow": workflow} + + +class TestPartitionOwnRun: + """Tests for ``partition_own_run``'s three-signal classification.""" + + def test_run_id_match_is_case_insensitive(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(RUN_ID_ENV, "AbC123") + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + entries = [_entry(111, 8080, run_id="abc123")] + partition = partition_own_run(entries) + + assert partition.own == entries + assert partition.others == [] + assert partition.reasons[8080] == "run id" + + def test_different_run_id_is_not_self_even_with_matching_port( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(RUN_ID_ENV, "self-run-id") + monkeypatch.setenv(WEB_BG_ENV, "1") + monkeypatch.setenv(WEB_PORT_ENV, "8080") + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + # Same port as our own web-bg port, but a *different* recorded run_id + # -- the compatibility (port) signal must not fire here. + entries = [_entry(111, 8080, run_id="someone-elses-run-id")] + partition = partition_own_run(entries) + + assert partition.own == [] + assert partition.others == entries + + def test_legacy_port_signal_fires_only_without_a_recorded_run_id( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(WEB_BG_ENV, "1") + monkeypatch.setenv(WEB_PORT_ENV, "8080") + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + entries = [_entry(111, 8080, run_id="")] + partition = partition_own_run(entries) + + assert partition.own == entries + assert partition.others == [] + assert partition.reasons[8080] == "dashboard port" + + def test_ancestry_pid_is_self(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset({111})) + + entries = [_entry(111, 8080)] + partition = partition_own_run(entries) + + assert partition.own == entries + assert partition.others == [] + assert partition.reasons[8080] == "process ancestry" + + def test_no_signal_leaves_own_empty_and_preserves_order( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + entries = [_entry(111, 8080, run_id="a"), _entry(222, 9090, run_id="b")] + partition = partition_own_run(entries) + + assert partition.own == [] + assert partition.others == entries + + +class TestDescribeOwnRun: + """Tests for ``describe_own_run``'s identity formatting.""" + + def test_prefers_run_id(self) -> None: + entry = _entry(111, 8080, run_id="abc123", workflow="/tmp/my-workflow.yaml") + assert describe_own_run(entry) == "abc123" + + def test_falls_back_to_workflow_stem_and_port(self) -> None: + entry = _entry(111, 8080, run_id="", workflow="/tmp/my-workflow.yaml") + assert describe_own_run(entry) == "my-workflow (port 8080)" + + def test_returns_plain_str_not_text(self) -> None: + entry = _entry(111, 8080, run_id="abc123") + assert type(describe_own_run(entry)) is str diff --git a/tests/test_cli/test_stop.py b/tests/test_cli/test_stop.py index 5480ff76..6d5bd697 100644 --- a/tests/test_cli/test_stop.py +++ b/tests/test_cli/test_stop.py @@ -11,7 +11,6 @@ from __future__ import annotations import json -import os from pathlib import Path from unittest.mock import patch @@ -22,6 +21,13 @@ runner = CliRunner() +# ``os.getpid()`` used to be a convenient "definitely alive" PID for these +# fixtures, but the self-exclusion rule (issue #399) treats a PID-file entry +# naming *this* test process as the caller's own run -- which would flip all +# of these tests to the refusal path. A synthetic PID that will never match +# the real test process keeps them deterministic. +_LIVE_PID = 999001 + @pytest.fixture() def pid_tmpdir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: @@ -32,13 +38,36 @@ def pid_tmpdir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: return runs_dir -def _write_pid(pid_dir: Path, pid: int, port: int, workflow: str = "/tmp/wf.yaml") -> Path: +@pytest.fixture(autouse=True) +def no_self_run(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep self-exclusion (issue #399) from perturbing the pre-existing targeting tests. + + Clears the bg-launch identity env vars and stubs ``own_run_pids`` so no + entry is ever misidentified as "this run" by a coincidental ancestor PID. + ``TestStopSelfExclusion`` overrides this per-test to exercise the actual + self-exclusion behaviour. + """ + monkeypatch.delenv("CONDUCTOR_RUN_ID", raising=False) + monkeypatch.delenv("CONDUCTOR_WEB_BG", raising=False) + monkeypatch.delenv("CONDUCTOR_WEB_PORT", raising=False) + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + +def _write_pid( + pid_dir: Path, pid: int, port: int, workflow: str = "/tmp/wf.yaml", run_id: str = "" +) -> Path: """Helper to write a PID file directly.""" name = Path(workflow).stem filepath = pid_dir / f"{name}-{port}.pid" filepath.write_text( json.dumps( - {"pid": pid, "port": port, "workflow": workflow, "started_at": "2026-03-03T00:00:00"} + { + "pid": pid, + "port": port, + "workflow": workflow, + "started_at": "2026-03-03T00:00:00", + "run_id": run_id, + } ) ) return filepath @@ -57,8 +86,7 @@ class TestStopByPort: """Test ``conductor stop --port ``.""" def test_stops_specific_port(self, pid_tmpdir: Path) -> None: - pid = os.getpid() - _write_pid(pid_tmpdir, pid, 8080) + _write_pid(pid_tmpdir, _LIVE_PID, 8080) with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -71,8 +99,7 @@ def test_stops_specific_port(self, pid_tmpdir: Path) -> None: assert "8080" in result.output def test_error_on_unknown_port(self, pid_tmpdir: Path) -> None: - pid = os.getpid() - _write_pid(pid_tmpdir, pid, 8080) + _write_pid(pid_tmpdir, _LIVE_PID, 8080) with patch("conductor.cli.pid._is_process_alive", return_value=True): result = runner.invoke(app, ["stop", "--port", "9999"]) @@ -85,9 +112,8 @@ class TestStopAll: """Test ``conductor stop --all``.""" def test_stops_all_workflows(self, pid_tmpdir: Path) -> None: - pid = os.getpid() - _write_pid(pid_tmpdir, pid, 8080, "/tmp/wf1.yaml") - _write_pid(pid_tmpdir, pid, 9090, "/tmp/wf2.yaml") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/wf1.yaml") + _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/wf2.yaml") with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -105,8 +131,7 @@ class TestStopAutoDetect: """Test ``conductor stop`` with no flags (auto-detect).""" def test_auto_stops_single_workflow(self, pid_tmpdir: Path) -> None: - pid = os.getpid() - _write_pid(pid_tmpdir, pid, 8080) + _write_pid(pid_tmpdir, _LIVE_PID, 8080) with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -118,9 +143,8 @@ def test_auto_stops_single_workflow(self, pid_tmpdir: Path) -> None: assert "Stopped" in result.output def test_lists_multiple_workflows(self, pid_tmpdir: Path) -> None: - pid = os.getpid() - _write_pid(pid_tmpdir, pid, 8080, "/tmp/wf1.yaml") - _write_pid(pid_tmpdir, pid, 9090, "/tmp/wf2.yaml") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/wf1.yaml") + _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/wf2.yaml") with patch("conductor.cli.pid._is_process_alive", return_value=True): result = runner.invoke(app, ["stop"]) @@ -188,3 +212,169 @@ def test_pid_file_is_removed_after_oserror(self, pid_tmpdir: Path) -> None: runner.invoke(app, ["stop", "--port", "8080"]) assert list(pid_tmpdir.glob("*.pid")) == [] + + +class TestStopSelfExclusion: + """Issue #399: ``conductor stop`` must never target the run it executes inside.""" + + @pytest.fixture(autouse=True) + def _clear_default_stub(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Override the module's ``no_self_run`` autouse fixture for this class. + + These tests exercise the real self-exclusion behaviour, so each test + configures its own identity signal (env var or ancestry) rather than + having ``own_run_pids`` stubbed to an empty set. + """ + monkeypatch.delenv("CONDUCTOR_RUN_ID", raising=False) + monkeypatch.delenv("CONDUCTOR_WEB_BG", raising=False) + monkeypatch.delenv("CONDUCTOR_WEB_PORT", raising=False) + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + def test_run_id_match_refuses_no_flag( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "abc123") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="abc123") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Refusing" in result.output + assert "No other workflows are running." in result.output + mock_kill.assert_not_called() + + def test_ancestry_match_refuses_no_flag( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _write_pid(pid_tmpdir, _LIVE_PID, 8080) + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset({_LIVE_PID})) + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Refusing" in result.output + mock_kill.assert_not_called() + + def test_all_stops_others_and_reports_exclusion( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/other.yaml", run_id="other-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--all"]) + + assert result.exit_code == 0 + assert "Excluded" in result.output + assert "Stopped" in result.output + assert mock_kill.call_count == 1 + + def test_all_with_only_self_sends_no_signal( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="self-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--all"]) + + assert result.exit_code == 0 + assert "No other workflows are running." in result.output + mock_kill.assert_not_called() + + def test_port_matching_own_run_exits_1( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="self-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--port", "8080"]) + + assert result.exit_code == 1 + assert "Refusing" in result.output + assert "--allow-self" in result.output + mock_kill.assert_not_called() + + def test_port_unknown_with_only_self_shows_exclusion_not_empty_table( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="self-run") + + with patch("conductor.cli.pid._is_process_alive", return_value=True): + result = runner.invoke(app, ["stop", "--port", "9999"]) + + assert result.exit_code == 1 + assert "No background workflow found on port 9999" in result.output + assert "Excluded" in result.output + assert "Running workflows:" not in result.output + + def test_allow_self_restores_stop_and_warns( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="self-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--allow-self"]) + + assert result.exit_code == 0 + assert "Stopped" in result.output + assert "Warning" in result.output + mock_kill.assert_called_once() + + def test_allow_self_with_port_stops_own_run( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, run_id="self-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--allow-self", "--port", "8080"]) + + assert result.exit_code == 0 + assert "Stopped" in result.output + assert "Warning" in result.output + mock_kill.assert_called_once() + + def test_allow_self_all_stops_both_and_warns( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/other.yaml", run_id="other-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop", "--all", "--allow-self"]) + + assert result.exit_code == 0 + assert "Warning" in result.output + assert mock_kill.call_count == 2 From 1e41bbcc2b486348303d4e81f384eec470185e5c Mon Sep 17 00:00:00 2001 From: Jason Robert Date: Tue, 11 Aug 2026 16:28:17 -0400 Subject: [PATCH 2/2] fix(cli): harden self-exclusion against malformed PID data, close test gaps Code review of PR #414 found several genuine issues in the self-exclusion logic itself and in the test suite's ability to catch a regression of it. Crash fixes in self_run.py (all verified with real reproductions): - partition_own_run crashed on a non-string run_id (e.g. int) in a PID file, taking down `stop` for every run, not just the malformed one. Now logs a warning and treats it as absent, matching pid.py::scan_pid_files' "skip, don't raise" discipline. - _read_ppid's `except OSError` didn't catch UnicodeDecodeError, which a non-UTF-8 /proc//status Name: line (settable by any unprivileged process via prctl(PR_SET_NAME)) can raise. This aborted the ancestry walk before it ever reached the os.getsid(0) fallback, defeating the safety net for any of up to 64 ancestor processes. Now decodes with errors="replace" since only the PPid: line is ever parsed. - describe_own_run crashed on `"workflow": null` via Path(None) -- precisely the message meant to reassure a user that their own run is protected. Now falls back to "unknown" the same way a missing key does. Test-suite hardening: - The one test combining a self-run and an other-run under --all asserted only generic substrings/call counts, and could not detect the own/others classification being fully inverted (verified by swapping them and rerunning -- the test still passed). Now uses distinct PIDs and asserts on which PID was actually signalled. - Added coverage for the no-flag path with a mixed self+other population (auto-stop-the-sole-other and list-remaining-others branches), previously fully unexercised -- arguably the most common real trigger for issue #399. - Added a unit test for partition_own_run with genuinely mixed input, and regression tests for each of the three crash fixes above. - OwnRunPartition now validates its own/others disjointness in __post_init__ (with slots=True), turning a future swap bug into an immediate, loud construction-time failure instead of a silent behavioral inversion. Doc/comment fixes: - Removed the dead `seen`-set cycle guard in own_run_pids (the pre-existing `parent in pids` check already terminates any cycle) and corrected the docstring that misattributed cycle termination to it. - Removed the redundant _clear_default_stub test fixture that duplicated the module-level no_self_run fixture without actually overriding it (different fixture names don't override in pytest). - Fixed a rule-F misattribution in describe_own_run's docstring, an ambiguous "naming the fix" phrase in CHANGELOG.md, a muddled causal explanation and a self-contradictory "restores ... exactly" sentence in docs/cli-reference.md, and matched stop()'s docstring backtick style to its sibling commands. - Documented the case-insensitive run_id comparison in the module docstring and docs/cli-reference.md. Verification: full test suite (5885 passed, 47 skipped, 1 pre-existing unrelated flake in test_event_log.py confirmed present on the unmodified branch), ruff, and ty all pass. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 4 +- docs/cli-reference.md | 21 ++--- src/conductor/cli/app.py | 10 +-- src/conductor/cli/self_run.py | 81 ++++++++++++++----- tests/test_cli/test_self_run.py | 136 ++++++++++++++++++++++++++++++++ tests/test_cli/test_stop.py | 87 ++++++++++++++++---- 6 files changed, 287 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3403841..3f6c0a2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -177,9 +177,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 #411's `run_id` field), and POSIX process ancestry, and excludes it from targeting by default: `--all` now means "stop all *other* runs", the no-flag auto-stop skips it, and `--port ` is refused (exit - `1`, naming the fix). If only your own run is alive, `stop`/`stop --all` + `1`, naming `--allow-self` as the remedy). If only your own run is alive, `stop`/`stop --all` print a refusal and exit `0` rather than erroring, since nothing named was - declined. Pass `--allow-self` to restore the previous behavior exactly; a + declined. Pass `--allow-self` to restore the previous targeting exactly; a yellow warning is printed whenever it actually causes your own run to be signalled. Process-ancestry detection is POSIX-only — Windows relies on the env-var signals alone. diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 29b5e3b6..dde051df 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -305,7 +305,7 @@ conductor stop [OPTIONS] | Option | Description | |--------|-------------| | `--port PORT` | Stop the workflow running on this specific port | -| `--all` | Stop all background conductor workflows (all *other* runs; see Self-Exclusion) | +| `--all` | Stop all *other* background conductor workflows (see Self-Exclusion) | | `--allow-self` | Include the run this command is executing inside (refused by default) | With no options, `conductor stop` lists running background workflows. If exactly one is found, it stops automatically. If multiple are running, it prints the list and asks you to specify `--port`. @@ -329,16 +329,18 @@ The web dashboard also exposes terminate controls that always preserve progress: `conductor stop` never targets the run it is executing inside by default — an agent smoke-testing this command must not terminate its own workflow (issue -#399). This matters because a background workflow's `bash` tool inherits the -parent process's environment, so an agent-invoked `conductor stop` can easily -end up pointed at the very run driving it. +#399). This matters because nothing about a PID-file entry says "this is the +workflow driving you" — an agent's `bash` tool, and any `conductor stop` it +spawns, sits inside that very run's process tree, so a naive `stop` treats it +as fair game just like any other run. The caller's own run is identified by three signals, tried in order (first match wins): -1. **`CONDUCTOR_RUN_ID`** env var matching the PID file's `run_id`. Set on - every `--web-bg` child (and inherited by its descendants, including a - spawned `conductor stop`). +1. **`CONDUCTOR_RUN_ID`** env var matching the PID file's `run_id` + (case-insensitively, since a manually-exported env var could differ in + case from the minted lowercase id). Set on every `--web-bg` child (and + inherited by its descendants, including a spawned `conductor stop`). 2. **`CONDUCTOR_WEB_BG=1` + `CONDUCTOR_WEB_PORT`** matching the entry's port — a compatibility signal used only for PID files written before `run_id` existed (empty `run_id`). @@ -355,8 +357,9 @@ Effects: - `conductor stop --port ` is refused and exits `1`, naming `--allow-self` as the remedy — here a specific target was named and declined. -- `conductor stop --allow-self [...]` restores the pre-#399 behavior exactly, - printing a yellow warning when the run actually being stopped is your own. +- `conductor stop --allow-self [...]` restores the pre-#399 targeting exactly + (same processes, same counts), but now prints a yellow warning when the run + being stopped is your own. **Windows caveat**: process ancestry (signal 3) is POSIX-only. On Windows, self-identification relies solely on the `CONDUCTOR_RUN_ID` / diff --git a/src/conductor/cli/app.py b/src/conductor/cli/app.py index 0fbe032f..dbc1c3aa 100644 --- a/src/conductor/cli/app.py +++ b/src/conductor/cli/app.py @@ -1280,12 +1280,12 @@ def stop( list and asks you to specify --port. \b - By default, ``stop`` never targets the run it is executing inside -- + By default, `stop` never targets the run it is executing inside -- an agent smoke-testing this command must not terminate its own - workflow (issue #399). That run is identified by ``CONDUCTOR_RUN_ID``, - the legacy ``CONDUCTOR_WEB_BG``/``CONDUCTOR_WEB_PORT`` pair, or process - ancestry, and is excluded from ``--all`` and the no-flag auto-stop; a - ``--port`` naming it is refused outright. Pass ``--allow-self`` to + workflow (issue #399). That run is identified by `CONDUCTOR_RUN_ID`, + the legacy `CONDUCTOR_WEB_BG`/`CONDUCTOR_WEB_PORT` pair, or process + ancestry, and is excluded from `--all` and the no-flag auto-stop; a + `--port` naming it is refused outright. Pass `--allow-self` to include it anyway. \b diff --git a/src/conductor/cli/self_run.py b/src/conductor/cli/self_run.py index 18d411e6..bb69c6d0 100644 --- a/src/conductor/cli/self_run.py +++ b/src/conductor/cli/self_run.py @@ -9,12 +9,13 @@ Three independent signals are tried, in order (first match wins per entry): -1. **``CONDUCTOR_RUN_ID`` env var** matches the PID file's ``run_id``. - ``cli/bg_runner.py::_build_bg_env`` sets this on the detached background - child, so every descendant of it — including an agent's ``bash`` tool and - the ``conductor stop`` it spawns — inherits it, and - ``_finalize_background_launch`` writes the same id into the PID file's - ``run_id`` key (added in issue #411). +1. **``CONDUCTOR_RUN_ID`` env var** matches the PID file's ``run_id`` + (case-insensitively, since a manually-exported env var could differ in + case from the minted lowercase id). ``cli/bg_runner.py::_build_bg_env`` + sets this on the detached background child, so every descendant of it — + including an agent's ``bash`` tool and the ``conductor stop`` it spawns — + inherits it, and ``_finalize_background_launch`` writes the same id into + the PID file's ``run_id`` key (added in issue #411). 2. **``CONDUCTOR_WEB_BG=1`` + ``CONDUCTOR_WEB_PORT``** matching the entry's ``port``, used *only* when the entry records no ``run_id`` (i.e. ``""``). This is the compatibility path for PID files written before #411. Limiting @@ -43,7 +44,6 @@ from __future__ import annotations -import contextlib import logging import os from dataclasses import dataclass @@ -76,8 +76,16 @@ def _read_ppid(pid: int) -> int | None: does not exist, is unreadable, or has no parseable ``PPid:`` line). """ try: - status = Path(f"/proc/{pid}/status").read_text() - except OSError: + # errors="replace": the kernel-sourced ``Name:`` line can be + # arbitrary, non-UTF-8 bytes (any unprivileged process can set its + # own via ``prctl(PR_SET_NAME, ...)``), but only ``PPid:`` and the + # digits after it are ever parsed below, so corruption elsewhere in + # the file is harmless — raising here would needlessly crash the + # whole ancestry walk (and the ``os.getsid(0)`` fallback after it) + # over a process with an unrelated garbled name. + status = Path(f"/proc/{pid}/status").read_text(errors="replace") + except OSError as exc: + logger.debug("Could not read /proc/%s/status: %s", pid, exc) return None for line in status.splitlines(): @@ -93,8 +101,9 @@ def own_run_pids() -> frozenset[int]: """Return the set of PIDs that identify "this process" for self-exclusion. Includes ``os.getpid()`` unconditionally, then walks ``/proc//status`` - ``PPid:`` links upward (bounded by :data:`_MAX_ANCESTRY_HOPS`, with a - ``seen`` set so a malformed cyclic chain also terminates), then adds + ``PPid:`` links upward (bounded by :data:`_MAX_ANCESTRY_HOPS`; a cyclic + chain terminates immediately, before the hop cap, because each new + parent is checked against every pid collected so far), then adds ``os.getsid(0)`` when available. Deliberately not memoized: a ``stop`` invocation calls this once, and @@ -107,12 +116,8 @@ def own_run_pids() -> frozenset[int]: pids: set[int] = {os.getpid()} if Path("/proc/self/status").exists(): - seen: set[int] = set() current = os.getpid() for _ in range(_MAX_ANCESTRY_HOPS): - if current in seen: - break - seen.add(current) parent = _read_ppid(current) if parent is None or parent in pids: break @@ -120,13 +125,15 @@ def own_run_pids() -> frozenset[int]: current = parent if hasattr(os, "getsid"): - with contextlib.suppress(OSError): + try: pids.add(os.getsid(0)) + except OSError as exc: + logger.debug("os.getsid(0) failed: %s", exc) return frozenset(pids) -@dataclass(frozen=True) +@dataclass(frozen=True, slots=True) class OwnRunPartition: """The result of classifying PID-file entries against this process's identity.""" @@ -141,6 +148,25 @@ class OwnRunPartition: ``"dashboard port"``, or ``"process ancestry"``. Logged at debug so a false positive is diagnosable without being printed to the user.""" + def __post_init__(self) -> None: + """Guard the one invariant that matters for safety: no entry is in both lists. + + ``others``/``own`` are both plain ``list[dict]`` — nothing in the type + system stops a future edit to :func:`partition_own_run` from swapping + them, which would silently invert exactly the safety property this + module exists to provide. Catching it here, at construction, turns + that mistake into an immediate ``ValueError`` instead of a quiet + misclassification. + """ + own_ports = {e.get("port") for e in self.own if isinstance(e.get("port"), int)} + other_ports = {e.get("port") for e in self.others if isinstance(e.get("port"), int)} + overlap = own_ports & other_ports + if overlap: + raise ValueError( + f"OwnRunPartition: entr{'y' if len(overlap) == 1 else 'ies'} on " + f"port(s) {sorted(overlap)} classified as both own and other" + ) + def partition_own_run(entries: list[dict]) -> OwnRunPartition: """Split PID-file entries into "others" and "own" (this process's run). @@ -170,6 +196,18 @@ def partition_own_run(entries: list[dict]) -> OwnRunPartition: for entry in entries: port = entry.get("port") entry_run_id = entry.get("run_id") or "" + if not isinstance(entry_run_id, str): + # A malformed PID file (hand-edited, partially written, or from a + # future schema) could carry a non-string run_id. Don't let one + # bad file crash `stop` for every other run: log it and treat it + # as absent, matching pid.py::scan_pid_files' own "skip, don't + # raise" discipline for malformed entries. + logger.warning( + "PID-file entry on port %r has non-string run_id (%r); ignoring for self-exclusion", + port, + entry_run_id, + ) + entry_run_id = "" reason: str | None = None if my_run_id and entry_run_id and entry_run_id.lower() == my_run_id.lower(): @@ -199,11 +237,14 @@ def describe_own_run(entry: dict) -> str: Returns: The ``run_id`` when the PID file records one, otherwise ``" (port N)"``. Always a plain ``str`` — never a - ``Text`` — since callers interpolate it through ``styled()`` (markup - guard rule F). + ``Text`` — since there's no styling to preserve here, and keeping it + a plain string forecloses a future f-string interpolation mistake + (markup guard rule F) regardless of which mechanism a caller uses to + print it. """ run_id = entry.get("run_id") if run_id: return str(run_id) - workflow = Path(entry.get("workflow", "unknown")).stem + workflow_raw = entry.get("workflow") or "unknown" + workflow = Path(str(workflow_raw)).stem return f"{workflow} (port {entry.get('port')})" diff --git a/tests/test_cli/test_self_run.py b/tests/test_cli/test_self_run.py index 264269df..f2ef4f86 100644 --- a/tests/test_cli/test_self_run.py +++ b/tests/test_cli/test_self_run.py @@ -101,6 +101,71 @@ def test_unreadable_proc_stops_without_raising(self, monkeypatch: pytest.MonkeyP assert os.getpid() in pids +@pytest.mark.skipif( + not sys.platform.startswith("linux"), reason="ancestry walk is /proc-based (Linux only)" +) +class TestReadPpid: + """Direct tests of ``_read_ppid``'s real error handling (not monkeypatched away).""" + + def test_returns_none_for_nonexistent_pid(self) -> None: + from conductor.cli.self_run import _read_ppid + + assert _read_ppid(999_999_999) is None + + def test_returns_none_for_malformed_ppid_value( + self, monkeypatch: pytest.MonkeyPatch, tmp_path + ) -> None: + from pathlib import Path as _Path + + from conductor.cli import self_run + + fake_status = tmp_path / "status" + fake_status.write_text("Name:\ttest\nPPid:\tnotanumber\n") + monkeypatch.setattr( + self_run, + "Path", + lambda p: fake_status if str(p).startswith("/proc/") else _Path(p), + ) + + assert self_run._read_ppid(12345) is None + + def test_returns_none_when_no_ppid_line_present( + self, monkeypatch: pytest.MonkeyPatch, tmp_path + ) -> None: + from pathlib import Path as _Path + + from conductor.cli import self_run + + fake_status = tmp_path / "status" + fake_status.write_text("Name:\ttest\nState:\tS (sleeping)\n") + monkeypatch.setattr( + self_run, + "Path", + lambda p: fake_status if str(p).startswith("/proc/") else _Path(p), + ) + + assert self_run._read_ppid(12345) is None + + def test_non_utf8_status_file_does_not_raise( + self, monkeypatch: pytest.MonkeyPatch, tmp_path + ) -> None: + """A process with a non-UTF-8 ``Name:`` (e.g. via ``prctl(PR_SET_NAME)``) must + not crash the ancestry walk -- only the ``PPid:`` line is ever parsed.""" + from pathlib import Path as _Path + + from conductor.cli import self_run + + fake_status = tmp_path / "status" + fake_status.write_bytes(b"Name:\tx\xff\xfey\nPPid:\t42\n") + monkeypatch.setattr( + self_run, + "Path", + lambda p: fake_status if str(p).startswith("/proc/") else _Path(p), + ) + + assert self_run._read_ppid(12345) == 42 + + def _entry(pid: int, port: int, run_id: str = "", workflow: str = "/tmp/wf.yaml") -> dict: return {"pid": pid, "port": port, "run_id": run_id, "workflow": workflow} @@ -170,6 +235,67 @@ def test_no_signal_leaves_own_empty_and_preserves_order( assert partition.own == [] assert partition.others == entries + def test_mixed_entries_partition_correctly(self, monkeypatch: pytest.MonkeyPatch) -> None: + """A genuinely mixed self+other input classifies each entry independently. + + This is the precise, cheap unit-level check for the exact scenario + issue #399 exists to fix: proven (during review) to catch an + own/others swap bug that CLI-level substring assertions missed. + """ + monkeypatch.setenv(RUN_ID_ENV, "mine") + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + entries = [_entry(111, 8080, run_id="mine"), _entry(222, 9090, run_id="other")] + partition = partition_own_run(entries) + + assert partition.own == [entries[0]] + assert partition.others == [entries[1]] + + def test_non_string_run_id_is_ignored_rather_than_crashing( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A malformed PID file with a non-string ``run_id`` must not crash classification. + + One corrupted/hand-edited PID file must not take down `stop` for + every other run (matching `pid.py::scan_pid_files`'s own "skip, + don't raise" discipline for malformed entries). + """ + monkeypatch.setenv(RUN_ID_ENV, "mine") + monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) + + entry = _entry(111, 8080) + entry["run_id"] = 12345 # malformed: an int instead of a string + partition = partition_own_run([entry]) + + assert partition.own == [] + assert partition.others == [entry] + + +class TestOwnRunPartitionInvariant: + """Tests for ``OwnRunPartition``'s constructor-enforced own/others disjointness.""" + + def test_disjoint_own_and_others_construct_cleanly(self) -> None: + from conductor.cli.self_run import OwnRunPartition + + OwnRunPartition(own=[_entry(111, 8080)], others=[_entry(222, 9090)], reasons={}) + + def test_overlapping_own_and_others_raises(self) -> None: + from conductor.cli.self_run import OwnRunPartition + + entry = _entry(111, 8080) + with pytest.raises(ValueError, match="both own and other"): + OwnRunPartition(own=[entry], others=[entry], reasons={}) + + def test_entries_missing_port_do_not_falsely_collide(self) -> None: + """Multiple entries with no ``port`` key must not be treated as an overlap.""" + from conductor.cli.self_run import OwnRunPartition + + OwnRunPartition( + own=[{"pid": 1}], + others=[{"pid": 2}], + reasons={}, + ) + class TestDescribeOwnRun: """Tests for ``describe_own_run``'s identity formatting.""" @@ -185,3 +311,13 @@ def test_falls_back_to_workflow_stem_and_port(self) -> None: def test_returns_plain_str_not_text(self) -> None: entry = _entry(111, 8080, run_id="abc123") assert type(describe_own_run(entry)) is str + + def test_null_workflow_falls_back_to_unknown_rather_than_crashing(self) -> None: + """A PID file with ``"workflow": null`` must not crash ``Path(None)``. + + ``dict.get(key, default)`` only substitutes the default when the key + is *absent*, not when it's present with value ``None`` -- so this + exercises that exact gotcha. + """ + entry = {"pid": 111, "port": 8080, "run_id": "", "workflow": None} + assert describe_own_run(entry) == "unknown (port 8080)" diff --git a/tests/test_cli/test_stop.py b/tests/test_cli/test_stop.py index 6d5bd697..963f5a79 100644 --- a/tests/test_cli/test_stop.py +++ b/tests/test_cli/test_stop.py @@ -11,6 +11,7 @@ from __future__ import annotations import json +import signal from pathlib import Path from unittest.mock import patch @@ -28,6 +29,12 @@ # the real test process keeps them deterministic. _LIVE_PID = 999001 +# A second synthetic "definitely alive but distinct from _LIVE_PID" PID, used +# in self+other mixed-population tests so assertions can pin down *which* +# entry was actually signalled rather than merely counting calls (a swapped +# own/other classification would otherwise pass the same assertions). +_OTHER_PID = 999002 + @pytest.fixture() def pid_tmpdir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: @@ -217,19 +224,6 @@ def test_pid_file_is_removed_after_oserror(self, pid_tmpdir: Path) -> None: class TestStopSelfExclusion: """Issue #399: ``conductor stop`` must never target the run it executes inside.""" - @pytest.fixture(autouse=True) - def _clear_default_stub(self, monkeypatch: pytest.MonkeyPatch) -> None: - """Override the module's ``no_self_run`` autouse fixture for this class. - - These tests exercise the real self-exclusion behaviour, so each test - configures its own identity signal (env var or ancestry) rather than - having ``own_run_pids`` stubbed to an empty set. - """ - monkeypatch.delenv("CONDUCTOR_RUN_ID", raising=False) - monkeypatch.delenv("CONDUCTOR_WEB_BG", raising=False) - monkeypatch.delenv("CONDUCTOR_WEB_PORT", raising=False) - monkeypatch.setattr("conductor.cli.self_run.own_run_pids", lambda: frozenset()) - def test_run_id_match_refuses_no_flag( self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -268,7 +262,7 @@ def test_all_stops_others_and_reports_exclusion( ) -> None: monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") - _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/other.yaml", run_id="other-run") + _write_pid(pid_tmpdir, _OTHER_PID, 9090, "/tmp/other.yaml", run_id="other-run") with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -279,7 +273,11 @@ def test_all_stops_others_and_reports_exclusion( assert result.exit_code == 0 assert "Excluded" in result.output assert "Stopped" in result.output - assert mock_kill.call_count == 1 + assert "9090" in result.output + # Pins down *which* run was signalled: a classification that swapped + # own/other would still print "Excluded"/"Stopped" and call os.kill + # exactly once, just against the wrong PID. + mock_kill.assert_called_once_with(_OTHER_PID, signal.SIGTERM) def test_all_with_only_self_sends_no_signal( self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch @@ -367,7 +365,7 @@ def test_allow_self_all_stops_both_and_warns( ) -> None: monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") - _write_pid(pid_tmpdir, _LIVE_PID, 9090, "/tmp/other.yaml", run_id="other-run") + _write_pid(pid_tmpdir, _OTHER_PID, 9090, "/tmp/other.yaml", run_id="other-run") with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -377,4 +375,61 @@ def test_allow_self_all_stops_both_and_warns( assert result.exit_code == 0 assert "Warning" in result.output + # Only the self entry should trigger the warning; a swapped + # classification would warn about the *other* entry instead, silently. + assert result.output.count("Warning") == 1 assert mock_kill.call_count == 2 + mock_kill.assert_any_call(_LIVE_PID, signal.SIGTERM) + mock_kill.assert_any_call(_OTHER_PID, signal.SIGTERM) + + def test_no_flag_mixed_auto_stops_sole_other_and_notes_exclusion( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """No flags, self-run + exactly one other run: auto-stops the other, excludes self. + + Covers `app.py`'s single-target auto-stop branch when the caller's + own run is present alongside exactly one other -- the most common + real trigger for issue #399 (an agent's own background workflow + plus one unrelated run, invoking bare `conductor stop`). + """ + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") + _write_pid(pid_tmpdir, _OTHER_PID, 9090, "/tmp/other.yaml", run_id="other-run") + + with ( + patch("conductor.cli.pid._is_process_alive", return_value=True), + patch("conductor.cli.app.os.kill") as mock_kill, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Stopped" in result.output + assert "9090" in result.output + assert "Excluded" in result.output + mock_kill.assert_called_once_with(_OTHER_PID, signal.SIGTERM) + + def test_no_flag_mixed_lists_others_only_and_notes_exclusion( + self, pid_tmpdir: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """No flags, self-run + two other runs: lists the *other* runs only. + + Covers `app.py`'s multi-target listing branch, asserting the printed + count reflects the post-exclusion `targetable` list (2), not the raw + PID-file count (3), and that the self entry's port never appears + under the running-workflows listing. + """ + monkeypatch.setenv("CONDUCTOR_RUN_ID", "self-run") + _write_pid(pid_tmpdir, _LIVE_PID, 8080, "/tmp/self.yaml", run_id="self-run") + _write_pid(pid_tmpdir, 999003, 9090, "/tmp/other1.yaml", run_id="other-run-1") + _write_pid(pid_tmpdir, 999004, 9091, "/tmp/other2.yaml", run_id="other-run-2") + + with patch("conductor.cli.pid._is_process_alive", return_value=True): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Multiple background workflows running (2)" in result.output + assert "9090" in result.output + assert "9091" in result.output + assert "Excluded" in result.output + # The self entry's port must not leak into the "running" listing. + assert "8080" not in result.output.split("Excluded")[0]