feat(register): warn when a run/build appears in flight during defaul…

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.")

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

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(

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

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(

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]