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 2263335a..0db19e36 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 `--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 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. + - **A pricing hook that silently prices nothing is now reported** (#386). #265 warns when the provider pricing hook *raises*; the companion case — a hook that never raises and returns `None` for everything — looked identical to diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 417bc5c2..08b02394 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 *other* background conductor workflows (see Self-Exclusion) | +| `--allow-self` | Include the run this command is executing inside (refused by default) | | `--force` | Proceed when the run's identity cannot be confirmed (see [Identity and `--force`](#identity-and---force)) | | `--json` | Emit a machine-readable result per workflow on stdout instead of prose | @@ -315,8 +316,8 @@ With no options, `conductor stop` lists running background workflows. If exactly | Code | Meaning | |------|---------| -| `0` | Every targeted workflow is confirmed stopped, or was already gone | -| `1` | `--port` matched no running workflow, or the target was ambiguous | +| `0` | Every targeted workflow is confirmed stopped, or was already gone — including a self-only refusal (see Self-Exclusion) | +| `1` | `--port` matched no running workflow, the target was ambiguous, or `--port` matched only your own run | | `2` | At least one workflow survived, or could not be confirmed stopped | Exit `2` is deliberately not a synonym for failure to signal — it means Conductor could not *prove* the process is gone. A run that ignored every rung and a run whose liveness could not be probed both land here, because both leave you with something you should look at by hand. @@ -358,6 +359,50 @@ 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 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` + (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`). +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 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` / +`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. + ### Examples ```bash @@ -367,8 +412,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 731fb1a3..68db64fe 100644 --- a/src/conductor/cli/app.py +++ b/src/conductor/cli/app.py @@ -11,7 +11,7 @@ from datetime import UTC, datetime 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 @@ -22,6 +22,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__) @@ -1272,6 +1277,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, force: Annotated[ bool, typer.Option( @@ -1310,10 +1322,21 @@ def stop( dashboard, because a recorded PID may since have been recycled onto an unrelated process. Use --force to override that check. + \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 every targeted workflow is confirmed stopped (or was already gone) - 1 --port matched no running workflow, or the target was ambiguous + 0 every targeted workflow is confirmed stopped (or was already + gone), including a self-only refusal + 1 --port matched no running workflow, the target was ambiguous, + or --port matched only your own run 2 at least one workflow survived or could not be confirmed stopped \b @@ -1322,10 +1345,12 @@ def stop( conductor stop --port 8080 conductor stop --all conductor stop --all --json + conductor stop --allow-self --port 8080 """ import json from conductor.cli.pid import read_pid_files, remove_pid_file_at + from conductor.cli.self_run import partition_own_run running = read_pid_files() @@ -1338,11 +1363,44 @@ def stop( ) return + partition = partition_own_run(running) + targetable = running if allow_self else partition.others + auto_detected_single = False + if all_workflows: - targets = running + if not allow_self and not targetable: + if json_output: + output_console.print_json( + json.dumps({"stopped": [], "failed": []}), ensure_ascii=True + ) + else: + _print_self_exclusion(partition, console, blocking=True) + return + targets = targetable + if not allow_self and partition.own and not json_output: + _print_self_exclusion(partition, console, blocking=False) elif port is not None: - targets = [e for e in running if e["port"] == port] + targets = [e for e in targetable if e["port"] == port] if not targets: + if not allow_self: + own_match = [e for e in partition.own if e["port"] == port] + if own_match: + if json_output: + output_console.print_json( + json.dumps( + { + "error": ( + f"port {port} is the run this command is executing " + "inside; pass --allow-self to include it" + ) + } + ), + ensure_ascii=True, + ) + else: + _print_self_refusal_line(own_match[0], console) + _print_allow_self_hint(console) + raise typer.Exit(code=1) if json_output: output_console.print_json( json.dumps({"error": f"no background workflow on port {port}"}), @@ -1355,11 +1413,21 @@ def stop( 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) - elif len(running) == 1: - targets = running + elif len(targetable) == 0: + if json_output: + output_console.print_json(json.dumps({"stopped": [], "failed": []}), ensure_ascii=True) + else: + _print_self_exclusion(partition, console, blocking=True) + return + elif len(targetable) == 1: + targets = targetable + auto_detected_single = True else: # Ambiguous: list rather than guess which run the user meant. This is # a failure to act, so it must not report success to automation. @@ -1372,7 +1440,7 @@ def stop( console.print( styled( "[bold yellow]Multiple background workflows running ({}).[/bold yellow]", - len(running), + len(targetable), ) ) console.print( @@ -1380,13 +1448,19 @@ 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) raise typer.Exit(code=1) # Prose goes to ``console`` (stderr); JSON goes to ``output_console`` # (stdout). They cannot corrupt each other, so diagnostics stay visible # even in --json mode. - results = [_stop_process(entry, console, force=force) for entry in targets] + results = [] + for entry in targets: + if allow_self: + _maybe_warn_stopping_self(entry, partition, console) + results.append(_stop_process(entry, console, force=force)) for entry, result in zip(targets, results, strict=True): if result["outcome"] in ("stopped", "already-exited"): @@ -1425,11 +1499,91 @@ def stop( "failed": [r for r in results if r["outcome"] not in ("stopped", "already-exited")], } output_console.print_json(json.dumps(payload), ensure_ascii=True) + elif auto_detected_single and not allow_self and partition.own: + # Single-target auto-stop: the exclusion note comes after the stop + # so the user sees "Stopped " before being told their own run + # was left out of consideration, matching the --all branch's note. + _print_self_exclusion(partition, console, blocking=False) if any(r["outcome"] not in ("stopped", "already-exited") for r in results): raise typer.Exit(code=2) +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), + ) + ) + + class Identity(str, Enum): """Result of checking that a PID file describes the process on its port. diff --git a/src/conductor/cli/self_run.py b/src/conductor/cli/self_run.py new file mode 100644 index 00000000..bb69c6d0 --- /dev/null +++ b/src/conductor/cli/self_run.py @@ -0,0 +1,250 @@ +"""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`` + (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 + 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 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: + # 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(): + 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`; 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 + 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(): + current = os.getpid() + for _ in range(_MAX_ANCESTRY_HOPS): + parent = _read_ppid(current) + if parent is None or parent in pids: + break + pids.add(parent) + current = parent + + if hasattr(os, "getsid"): + try: + pids.add(os.getsid(0)) + except OSError as exc: + logger.debug("os.getsid(0) failed: %s", exc) + + return frozenset(pids) + + +@dataclass(frozen=True, slots=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 __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). + + 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 "" + 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(): + 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 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_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_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..f2ef4f86 --- /dev/null +++ b/tests/test_cli/test_self_run.py @@ -0,0 +1,323 @@ +"""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 + + +@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} + + +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 + + 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.""" + + 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 + + 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 4a2f56c0..f4d78142 100644 --- a/tests/test_cli/test_stop.py +++ b/tests/test_cli/test_stop.py @@ -6,11 +6,14 @@ - Auto-stop when exactly one workflow is running - Listing when multiple workflows are running - Error cases (no running workflows, invalid port) +- Self-exclusion (issue #399): ``stop`` must never target the run it + executes inside """ from __future__ import annotations import contextlib +import importlib import json import os import re @@ -24,8 +27,29 @@ from conductor.cli.app import Identity, app from conductor.cli.pid import Liveness +# ``conductor.cli.__init__`` does ``from conductor.cli.app import app``, which +# rebinds the *package's* ``app`` attribute to the Typer instance -- shadowing +# the submodule of the same name. ``import conductor.cli.app as x`` resolves +# through that shadowed attribute (via IMPORT_FROM) and would silently hand +# back the Typer app instead of the module, so ``importlib.import_module`` is +# used here instead to get the real module object to patch/wrap attributes on. +app_module = importlib.import_module("conductor.cli.app") + 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 + +# 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: @@ -36,12 +60,23 @@ def pid_tmpdir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: return runs_dir +@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 = "a1b2c3d4", + 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 @@ -65,9 +100,9 @@ def _stops_cleanly() -> Iterator[None]: """Patch the ladder so the target is confirmed dead on the first rung. These tests cover *routing* — which PID files get targeted by ``--port`` / - ``--all`` / auto-detect — not the escalation ladder itself, which has its - own module (``test_stop_ladder.py``). Patching the outcome also keeps them - free of the ladder's real bounded waits. + ``--all`` / auto-detect / self-exclusion — not the escalation ladder + itself, which has its own module (``test_stop_ladder.py``). Patching the + outcome also keeps them free of the ladder's real bounded waits. """ with ( patch("conductor.cli.pid._is_process_alive", return_value=True), @@ -79,6 +114,20 @@ def _stops_cleanly() -> Iterator[None]: yield +@contextlib.contextmanager +def _spy_stop_process() -> Iterator[object]: + """Wrap the real ``_stop_process`` so calls are recorded without changing behaviour. + + Used by the self-exclusion tests to pin down *which* PID-file entries + were actually targeted, the same way ``TestStopAll`` above pins down + call args by patching ``_stop_process`` directly -- except here the real + ladder still runs (under ``_stops_cleanly()``), so the printed "Stopped" + / "Excluded" / "Warning" text is genuine rather than asserted on faith. + """ + with patch.object(app_module, "_stop_process", wraps=app_module._stop_process) as spy: + yield spy + + class TestStopNoRunning: """Test behavior when no background workflows are running.""" @@ -92,8 +141,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 _stops_cleanly(): result = runner.invoke(app, ["stop", "--port", "8080"]) @@ -103,8 +151,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"]) @@ -117,13 +164,12 @@ 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 _stops_cleanly(), patch("conductor.cli.app._stop_process") as stop_one: stop_one.return_value = { - "pid": pid, + "pid": _LIVE_PID, "port": 0, "workflow": "wf", "run_id": "", @@ -142,8 +188,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 _stops_cleanly(): result = runner.invoke(app, ["stop"]) @@ -152,9 +197,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"]) @@ -262,3 +306,205 @@ def test_pid_file_is_removed_when_process_confirmed_gone(self, pid_tmpdir: Path) assert result.exit_code == 0 assert "already exited" in result.output assert list(pid_tmpdir.glob("*.pid")) == [] + + +class TestStopSelfExclusion: + """Issue #399: ``conductor stop`` must never target the run it executes inside.""" + + 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.object(app_module, "_stop_process") as stop_spy, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Refusing" in result.output + assert "No other workflows are running." in result.output + stop_spy.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.object(app_module, "_stop_process") as stop_spy, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 0 + assert "Refusing" in result.output + stop_spy.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, _OTHER_PID, 9090, "/tmp/other.yaml", run_id="other-run") + + with _stops_cleanly(), _spy_stop_process() as spy: + result = runner.invoke(app, ["stop", "--all"]) + + assert result.exit_code == 0 + assert "Excluded" in result.output + assert "Stopped" in result.output + assert "9090" in result.output + # Pins down *which* run was targeted: a classification that swapped + # own/other would still print "Excluded"/"Stopped", just against the + # wrong entry. + assert [c.args[0]["port"] for c in spy.call_args_list] == [9090] + + 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.object(app_module, "_stop_process") as stop_spy, + ): + result = runner.invoke(app, ["stop", "--all"]) + + assert result.exit_code == 0 + assert "No other workflows are running." in result.output + stop_spy.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.object(app_module, "_stop_process") as stop_spy, + ): + result = runner.invoke(app, ["stop", "--port", "8080"]) + + assert result.exit_code == 1 + assert "Refusing" in result.output + assert "--allow-self" in result.output + stop_spy.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 _stops_cleanly(), _spy_stop_process() as spy: + result = runner.invoke(app, ["stop", "--allow-self"]) + + assert result.exit_code == 0 + assert "Stopped" in result.output + assert "Warning" in result.output + assert spy.call_count == 1 + + 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 _stops_cleanly(), _spy_stop_process() as spy: + result = runner.invoke(app, ["stop", "--allow-self", "--port", "8080"]) + + assert result.exit_code == 0 + assert "Stopped" in result.output + assert "Warning" in result.output + assert spy.call_count == 1 + + 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, _OTHER_PID, 9090, "/tmp/other.yaml", run_id="other-run") + + with _stops_cleanly(), _spy_stop_process() as spy: + result = runner.invoke(app, ["stop", "--all", "--allow-self"]) + + 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 spy.call_count == 2 + assert sorted(c.args[0]["port"] for c in spy.call_args_list) == [8080, 9090] + + 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 _stops_cleanly(), _spy_stop_process() as spy: + 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 + assert [c.args[0]["port"] for c in spy.call_args_list] == [9090] + + 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), + patch.object(app_module, "_stop_process") as stop_spy, + ): + result = runner.invoke(app, ["stop"]) + + assert result.exit_code == 1 + 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] + stop_spy.assert_not_called() diff --git a/tests/test_cli/test_stop_ladder.py b/tests/test_cli/test_stop_ladder.py index b291fbfe..2c072487 100644 --- a/tests/test_cli/test_stop_ladder.py +++ b/tests/test_cli/test_stop_ladder.py @@ -67,6 +67,27 @@ def pid_tmpdir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: return runs_dir +@pytest.fixture(autouse=True) +def no_self_run(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep self-exclusion (issue #399) from perturbing these ladder tests. + + These tests use small, arbitrary PIDs (e.g. ``1``, ``2``, ``4242``) as + stand-ins for "some other process" -- they predate issue #399 and are not + about self-exclusion at all. But ``conductor.cli.self_run.own_run_pids()`` + walks real process ancestry, and in a shallow-PID-namespace environment + (e.g. a container) that walk can genuinely include low PIDs like ``1`` or + ``2``, which would misclassify these entries as the caller's own run and + silently exclude them from ``stop --all`` targeting. Clearing the env + vars and stubbing ``own_run_pids`` keeps these pre-existing tests exactly + as deterministic as they were before #399; self-exclusion itself is + covered by ``test_stop.py::TestStopSelfExclusion``. + """ + 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 _entry(pid: int = 4242, port: int = 8080, run_id: str = _RUN_ID) -> dict: """Build a PID-file dict shaped like ``read_pid_files`` output.""" return {