diff --git a/roar/cli/commands/register.py b/roar/cli/commands/register.py index be392f98..2007a83c 100644 --- a/roar/cli/commands/register.py +++ b/roar/cli/commands/register.py @@ -367,7 +367,10 @@ def register( and not yes and not dry_run and not confirm_defaulted_active_session_publish( - session_hash=target, command_name="roar register", start_dir=str(ctx.cwd) + session_hash=target, + command_name="roar register", + start_dir=str(ctx.cwd), + roar_dir=ctx.roar_dir, ) ): click.echo("Registration aborted.") diff --git a/roar/cli/commands/status.py b/roar/cli/commands/status.py index e7025cf5..5bd3e09a 100644 --- a/roar/cli/commands/status.py +++ b/roar/cli/commands/status.py @@ -40,6 +40,16 @@ def status(ctx: RoarContext, untracked_dirs: bool) -> None: except StatusQueryError as exc: raise click.ClickException(str(exc)) from exc + # A concurrent `roar run`/`roar build` has no row in the DB until it + # completes, so the session summary above can't reflect it — name that + # gap here instead of showing a job count that's about to change. Warn, + # never block; unconditional (not gated on hints) since this is a real + # data-freshness risk, not a UX tip. + from ...execution.runtime.active_runs import in_flight_run_warnings + + for warning in in_flight_run_warnings(ctx.roar_dir): + click.echo(warning, err=True) + # `reproduce` is the on-mission verb but is undiscoverable at the # surfaces a user actually visits. Status is one of those surfaces: # the user is staring at the active session and may want to know diff --git a/roar/cli/publish_intent.py b/roar/cli/publish_intent.py index f6aed364..3e54b7ac 100644 --- a/roar/cli/publish_intent.py +++ b/roar/cli/publish_intent.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path import click @@ -127,7 +128,11 @@ def _confirm_or_explain_noninteractive(prompt: str, *, default: bool, command_na def confirm_defaulted_active_session_publish( - *, session_hash: str, command_name: str, start_dir: str | None = None + *, + session_hash: str, + command_name: str, + start_dir: str | None = None, + roar_dir: Path | None = None, ) -> bool: """Prompt before a target-less invocation publishes the whole active session. @@ -137,13 +142,23 @@ def confirm_defaulted_active_session_publish( by-accident publish. This gate fires independent of ``publish_intent.anonymous``: an attributed, private publish of the wrong (too-broad) scope is still a real mistake, so it can't rely on the anonymous/public confirmation to catch it. + + Job rows are only written when a job completes, so a concurrent `roar run`/ + `roar build` in another terminal has nothing in the DB yet for this prompt + to see — it can silently publish a session that's still being added to. + `roar_dir`, when given, is checked for other live in-flight run markers so + that risk can be named here instead of passing silently. """ + from ..execution.runtime.active_runs import in_flight_run_warnings + click.echo("") click.echo(f"Will publish to: {_publish_url_preview(start_dir, session_hash)}") click.echo( f"No target given — the whole active session ({_preview_hash(session_hash)}) will be " "published, including every job and artifact recorded in it so far." ) + for warning in in_flight_run_warnings(roar_dir): + click.echo(warning) click.echo(f"Use `{command_name} -y` to skip this confirmation in scripts.") click.echo("") return _confirm_or_explain_noninteractive( diff --git a/roar/execution/runtime/active_runs.py b/roar/execution/runtime/active_runs.py new file mode 100644 index 00000000..18c4fb99 --- /dev/null +++ b/roar/execution/runtime/active_runs.py @@ -0,0 +1,144 @@ +""" +Active-run markers for concurrent-run detection. + +`RunCoordinator.execute()` writes one of these markers before starting the +tracer and removes it when the run finishes (see `active_run_marker`), so a +separate `roar register` invocation against the same `.roar` directory can +warn when a `roar run`/`roar build` still appears to be in flight. Mirrors +the PID-liveness pattern already used for the S3 proxy daemon +(`execution/cluster/proxy.py`'s `proxy.json` + `os.kill(pid, 0)`), scoped to +this host rather than a shared/remote signal. +""" + +from __future__ import annotations + +import contextlib +import json +import os +import time +from collections.abc import Iterator +from pathlib import Path +from typing import Any + + +def _markers_dir(roar_dir: Path) -> Path: + return Path(roar_dir) / "active_runs" + + +def write_marker(roar_dir: Path, *, pid: int, command: list[str], job_type: str | None) -> None: + """Record that `pid` has a run/build in flight against `roar_dir`. + + Best-effort: a missing/unwritable `.roar` dir must never fail the run. + """ + try: + markers_dir = _markers_dir(roar_dir) + markers_dir.mkdir(parents=True, exist_ok=True) + info = { + "pid": pid, + "started_at": time.time(), + "command": command, + "job_type": job_type, + } + (markers_dir / f"{pid}.json").write_text(json.dumps(info)) + except OSError: + pass + + +def remove_marker(roar_dir: Path, *, pid: int) -> None: + """Remove a marker written by `write_marker`. Best-effort; never raises.""" + with contextlib.suppress(OSError): + (_markers_dir(roar_dir) / f"{pid}.json").unlink(missing_ok=True) + + +@contextlib.contextmanager +def active_run_marker( + roar_dir: Path, *, pid: int, command: list[str], job_type: str | None +) -> Iterator[None]: + """Write a marker for the duration of the enclosed block, cleaned up on exit.""" + write_marker(roar_dir, pid=pid, command=command, job_type=job_type) + try: + yield + finally: + remove_marker(roar_dir, pid=pid) + + +def _is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except OSError: + return False + return True + + +def list_active_runs(roar_dir: Path) -> list[dict[str, Any]]: + """Return markers for runs that still appear to be alive on this host. + + Self-healing: markers for PIDs that are no longer alive (crash, `kill -9`, + a run that finished without going through `active_run_marker`'s cleanup) + are deleted as they're found rather than reported. + """ + markers_dir = _markers_dir(roar_dir) + if not markers_dir.is_dir(): + return [] + + active: list[dict[str, Any]] = [] + for marker_path in markers_dir.glob("*.json"): + try: + info = json.loads(marker_path.read_text()) + pid = int(info["pid"]) + except (OSError, ValueError, KeyError, json.JSONDecodeError): + marker_path.unlink(missing_ok=True) + continue + + if _is_alive(pid): + active.append(info) + else: + marker_path.unlink(missing_ok=True) + + return active + + +def format_elapsed(seconds: float) -> str: + """Render a small elapsed-time hint, e.g. ``2m14s`` or ``43s``.""" + total = max(0, int(seconds)) + minutes, secs = divmod(total, 60) + return f"{minutes}m{secs}s" if minutes else f"{secs}s" + + +def in_flight_run_warnings(roar_dir: Path | None) -> list[str]: + """Warning lines for `roar run`/`roar build` processes still active on this host. + + Shared by any surface that wants to name this risk — `roar register`'s + defaulted-active-session prompt and `roar status`'s summary both call this + rather than duplicating the marker lookup + formatting. + + Best-effort: a missing/unreadable marker dir means "nothing detected", not + an error — this must never block or fail the caller. + """ + if roar_dir is None: + return [] + try: + markers = list_active_runs(roar_dir) + except Exception: + return [] + + now = time.time() + own_pid = os.getpid() + lines: list[str] = [] + for marker in markers: + pid = marker.get("pid") + if pid == own_pid: + continue + job_type = marker.get("job_type") or "run" + started_at = marker.get("started_at") + elapsed = format_elapsed(now - started_at) if isinstance(started_at, (int, float)) else "?" + command = marker.get("command") + command_preview = " ".join(command) if isinstance(command, list) and command else None + detail = f"pid {pid}, started {elapsed} ago" + if command_preview: + detail += f": `{command_preview}`" + lines.append( + f"Warning: a roar {job_type} ({detail}) appears to still be in progress — " + "the active session may be incomplete." + ) + return lines diff --git a/roar/execution/runtime/coordinator.py b/roar/execution/runtime/coordinator.py index 29796933..be6b5ccc 100644 --- a/roar/execution/runtime/coordinator.py +++ b/roar/execution/runtime/coordinator.py @@ -96,6 +96,15 @@ def execute(self, ctx: RunContext) -> RunResult: Returns: RunResult with execution details """ + from .active_runs import active_run_marker + + with active_run_marker( + ctx.roar_dir, pid=os.getpid(), command=list(ctx.command), job_type=ctx.job_type + ): + return self._execute_traced(ctx) + + def _execute_traced(self, ctx: RunContext) -> RunResult: + """Body of `execute()`, run inside the active-run marker's lifetime.""" from .signal_handler import ProcessSignalHandler self.logger.debug( diff --git a/tests/execution/runtime/test_active_runs.py b/tests/execution/runtime/test_active_runs.py new file mode 100644 index 00000000..85f02586 --- /dev/null +++ b/tests/execution/runtime/test_active_runs.py @@ -0,0 +1,139 @@ +"""Unit tests for active-run marker write/list/cleanup behavior.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from roar.execution.runtime.active_runs import ( + active_run_marker, + format_elapsed, + in_flight_run_warnings, + list_active_runs, + remove_marker, + write_marker, +) + + +def _markers_dir(roar_dir: Path) -> Path: + return roar_dir / "active_runs" + + +def test_write_marker_creates_file_with_expected_fields(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + write_marker(roar_dir, pid=1234, command=["python", "train.py"], job_type="run") + + marker_path = _markers_dir(roar_dir) / "1234.json" + assert marker_path.exists() + info = json.loads(marker_path.read_text()) + assert info["pid"] == 1234 + assert info["command"] == ["python", "train.py"] + assert info["job_type"] == "run" + assert isinstance(info["started_at"], float) + + +def test_write_marker_is_best_effort_on_unwritable_dir(tmp_path: Path) -> None: + """A path component blocked by a file (not a dir) must not raise.""" + blocker = tmp_path / "blocked" + blocker.write_text("not a directory") + write_marker(blocker / ".roar", pid=1, command=["x"], job_type="run") # must not raise + + +def test_remove_marker_is_best_effort_when_missing(tmp_path: Path) -> None: + remove_marker(tmp_path / ".roar", pid=999) # must not raise, no file exists + + +def test_active_run_marker_cleans_up_on_normal_exit(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + with active_run_marker(roar_dir, pid=42, command=["x"], job_type="run"): + assert (_markers_dir(roar_dir) / "42.json").exists() + assert not (_markers_dir(roar_dir) / "42.json").exists() + + +def test_active_run_marker_cleans_up_on_exception(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + try: + with active_run_marker(roar_dir, pid=43, command=["x"], job_type="build"): + assert (_markers_dir(roar_dir) / "43.json").exists() + raise RuntimeError("boom") + except RuntimeError: + pass + assert not (_markers_dir(roar_dir) / "43.json").exists() + + +def test_list_active_runs_returns_empty_when_no_markers_dir(tmp_path: Path) -> None: + assert list_active_runs(tmp_path / ".roar") == [] + + +def test_list_active_runs_reports_live_pid(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + # Our own PID is guaranteed alive for the duration of the test. + write_marker(roar_dir, pid=os.getpid(), command=["python", "train.py"], job_type="run") + + active = list_active_runs(roar_dir) + + assert len(active) == 1 + assert active[0]["pid"] == os.getpid() + # Still on disk — a live PID's marker is not touched. + assert (_markers_dir(roar_dir) / f"{os.getpid()}.json").exists() + + +def test_list_active_runs_self_heals_dead_pid(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + # PID 2**30 is astronomically unlikely to be a live process. + dead_pid = 2**30 + write_marker(roar_dir, pid=dead_pid, command=["python", "train.py"], job_type="run") + + active = list_active_runs(roar_dir) + + assert active == [] + assert not (_markers_dir(roar_dir) / f"{dead_pid}.json").exists() + + +def test_list_active_runs_self_heals_corrupt_marker(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + markers_dir = _markers_dir(roar_dir) + markers_dir.mkdir(parents=True) + (markers_dir / "garbage.json").write_text("{not json") + + active = list_active_runs(roar_dir) + + assert active == [] + assert not (markers_dir / "garbage.json").exists() + + +def test_format_elapsed_under_a_minute() -> None: + assert format_elapsed(43) == "43s" + + +def test_format_elapsed_over_a_minute() -> None: + assert format_elapsed(134) == "2m14s" + + +def test_in_flight_run_warnings_empty_when_roar_dir_is_none() -> None: + assert in_flight_run_warnings(None) == [] + + +def test_in_flight_run_warnings_empty_with_no_markers(tmp_path: Path) -> None: + assert in_flight_run_warnings(tmp_path / ".roar") == [] + + +def test_in_flight_run_warnings_excludes_own_pid(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + write_marker(roar_dir, pid=os.getpid(), command=["python", "train.py"], job_type="run") + assert in_flight_run_warnings(roar_dir) == [] + + +def test_in_flight_run_warnings_reports_other_live_pid(tmp_path: Path) -> None: + roar_dir = tmp_path / ".roar" + other_pid = os.getppid() # guaranteed alive for the duration of the test + write_marker(roar_dir, pid=other_pid, command=["python", "train.py"], job_type="build") + + warnings = in_flight_run_warnings(roar_dir) + + assert len(warnings) == 1 + assert f"pid {other_pid}" in warnings[0] + assert "train.py" in warnings[0] + assert "roar build" in warnings[0] + assert "still be in progress" in warnings[0] diff --git a/tests/unit/test_cli_query_errors.py b/tests/unit/test_cli_query_errors.py index 8bd22b70..9463b140 100644 --- a/tests/unit/test_cli_query_errors.py +++ b/tests/unit/test_cli_query_errors.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os from unittest.mock import MagicMock, patch from click.testing import CliRunner @@ -33,6 +34,35 @@ def test_status_cli_exits_non_zero_without_active_session(tmp_path) -> None: assert "No active session." in result.output +def test_status_cli_has_no_in_flight_warning_by_default(tmp_path) -> None: + ctx = _ctx(tmp_path) + with patch("roar.cli.commands.status.render_status", return_value="STATUS BODY"): + result = CliRunner().invoke(status, obj=ctx) + + assert result.exit_code == 0, result.output + assert "still be in progress" not in result.output + + +def test_status_cli_warns_on_in_flight_run(tmp_path) -> None: + """A live `roar run`/`roar build` marker must surface a warning in `roar + status` too — the same gap register's defaulted-active-session prompt + warns about: job rows (and this summary) don't reflect a run until it + completes, so a concurrent run in another terminal is otherwise invisible + here.""" + from roar.execution.runtime.active_runs import write_marker + + ctx = _ctx(tmp_path) + other_pid = os.getppid() # guaranteed alive for the duration of the test + write_marker(ctx.roar_dir, pid=other_pid, command=["python", "train.py"], job_type="run") + + with patch("roar.cli.commands.status.render_status", return_value="STATUS BODY"): + result = CliRunner().invoke(status, obj=ctx) + + assert result.exit_code == 0, result.output + assert f"pid {other_pid}" in result.output + assert "still be in progress" in result.output + + def test_show_cli_exits_non_zero_for_missing_path_lookup(tmp_path) -> None: result = CliRunner().invoke(show, ["artifact.bin"], obj=_ctx(tmp_path)) diff --git a/tests/unit/test_proxy_coordinator.py b/tests/unit/test_proxy_coordinator.py index f244c8e0..5af08d4a 100644 --- a/tests/unit/test_proxy_coordinator.py +++ b/tests/unit/test_proxy_coordinator.py @@ -1,8 +1,11 @@ """Tests for RunCoordinator runtime-resource integration.""" +import os from pathlib import Path from unittest.mock import MagicMock, patch +import pytest + from roar.core.exceptions import TracerPreflightError from roar.execution.runtime.coordinator import RunCoordinator from roar.execution.runtime.resources import RuntimeObservationBundle, RuntimeResourceStart @@ -160,3 +163,66 @@ def test_runtime_resource_is_stopped_when_tracer_log_missing(self): assert result.exit_code == 1 assert resource.stop_calls == [1] + + +class TestActiveRunMarkerLifecycle: + """`execute()` must mark itself in-flight for `roar register` to discover, + and clean up on every exit path (success, tracer error, exception).""" + + def _markers_dir(self, roar_dir: Path) -> Path: + return roar_dir / "active_runs" + + def test_marker_exists_during_tracer_execution_and_is_removed_after(self, tmp_path): + from roar.execution.runtime.active_runs import list_active_runs + + roar_dir = tmp_path / ".roar" + ctx = _make_ctx() + ctx.roar_dir = roar_dir + + mock_tracer = _make_mock_tracer() + observed_active = {} + + def _execute_side_effect(*args, **kwargs): + # Called mid-`execute()`, before the marker is cleaned up. + observed_active["runs"] = list_active_runs(roar_dir) + return _make_tracer_result() + + mock_tracer.execute.side_effect = _execute_side_effect + coord = RunCoordinator(tracer_service=mock_tracer) + + mock_prov = MagicMock() + mock_prov.collect.return_value = {"data": {"read_files": [], "written_files": []}} + with ( + patch("os.path.exists", return_value=True), + patch("roar.integrations.config.load_config", return_value={}), + patch("roar.execution.provenance.ProvenanceService", return_value=mock_prov), + patch.object(coord, "_record_job", return_value=(1, "abc123", [], [], [], [], {})), + patch.object(coord, "_backup_previous_outputs"), + patch.object(coord, "_cleanup_logs"), + ): + coord.execute(ctx) + + assert len(observed_active["runs"]) == 1 + assert observed_active["runs"][0]["pid"] == os.getpid() + assert observed_active["runs"][0]["command"] == ctx.command + assert list_active_runs(roar_dir) == [] + assert list(self._markers_dir(roar_dir).glob("*.json")) == [] + + def test_marker_is_removed_when_tracer_raises(self, tmp_path): + from roar.execution.runtime.active_runs import list_active_runs + + roar_dir = tmp_path / ".roar" + ctx = _make_ctx() + ctx.roar_dir = roar_dir + + mock_tracer = _make_mock_tracer() + mock_tracer.execute.side_effect = RuntimeError("tracer blew up") + coord = RunCoordinator(tracer_service=mock_tracer) + + with ( + patch.object(coord, "_backup_previous_outputs"), + pytest.raises(RuntimeError, match="tracer blew up"), + ): + coord.execute(ctx) + + assert list_active_runs(roar_dir) == [] diff --git a/tests/unit/test_register_cli.py b/tests/unit/test_register_cli.py index 22d5e331..15fe37d1 100644 --- a/tests/unit/test_register_cli.py +++ b/tests/unit/test_register_cli.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from pathlib import Path from unittest.mock import MagicMock, patch @@ -480,6 +481,63 @@ def test_register_cli_accepts_defaulted_active_session_publish_prompt(tmp_path: assert request.target == session_hash +def test_register_cli_defaulted_active_session_prompt_has_no_in_flight_warning_by_default( + tmp_path: Path, +) -> None: + """No active-run marker present -> no in-flight warning line.""" + runner = CliRunner() + session_hash = "3" * 64 + with ( + patch("roar.cli.publish_intent._is_logged_in", return_value=True), + patch( + "roar.application.query.status.compute_active_session_hash", + return_value=session_hash, + ), + patch("roar.cli.commands.register.register_lineage_target") as mock_register, + ): + mock_register.return_value = _fake_result() + result = runner.invoke(register, [], input="y\n", obj=_mock_context(tmp_path)) + + assert result.exit_code == 0, result.output + assert "still be in progress" not in result.output + + +def test_register_cli_defaulted_active_session_prompt_warns_on_in_flight_run( + tmp_path: Path, +) -> None: + """A live `roar run`/`roar build` marker in this repo's `.roar` dir must + surface a warning in the defaulted-active-session prompt — this is the gap + a bare `roar register` in one terminal previously had no way to see a + still-running job in another terminal of the same session. + """ + from roar.execution.runtime.active_runs import write_marker + + runner = CliRunner() + session_hash = "4" * 64 + ctx = _mock_context(tmp_path) + # Our own PID would be excluded as "self" — use the parent (pytest's own + # process), which is guaranteed alive for the duration of the test, to + # simulate a genuinely different in-flight process. + other_pid = os.getppid() + write_marker(ctx.roar_dir, pid=other_pid, command=["python", "train.py"], job_type="run") + + with ( + patch("roar.cli.publish_intent._is_logged_in", return_value=True), + patch( + "roar.application.query.status.compute_active_session_hash", + return_value=session_hash, + ), + patch("roar.cli.commands.register.register_lineage_target") as mock_register, + ): + mock_register.return_value = _fake_result() + result = runner.invoke(register, [], input="n\n", obj=ctx) + + assert f"pid {other_pid}" in result.output + assert "train.py" in result.output + assert "still be in progress" in result.output + mock_register.assert_not_called() + + def test_register_cli_yes_skips_defaulted_active_session_prompt(tmp_path: Path) -> None: """--yes bypasses the new prompt (as well as the pre-existing ones).""" runner = CliRunner()