Session watcher: queue and history for the live author session, launch ledger, --why

CHANGELOG.md

@@ -8,6 +8,16 @@ Versions follow [SemVer](https://semver.org).
 
 ### Added
 
+- The session watcher (docs/design/session-watcher.md): a thread beside the
+  live author session answers two new tool verbs within seconds. `queue`
+  shows the kernel's jobs in the cluster queue, every agent's, each launch
+  with its author's `--why` (a new optional one-line field on `launch`), the
+  pending reason, elapsed and limit, and the GPU lane's node states from
+  sinfo; `history` lists this run's launches sleep by sleep and how each job
+  ended. Both ride the `sync` marker protocol; a deployment without the
+  watcher times out with a plain message. The kernel keeps a per-run
+  append-only launch ledger (`launches.jsonl`) that both views read, and the
+  wake text echoes each launch's why.
 - `climb/status.json` carries the fleet's queue: the kernel's own Slurm jobs
   (tick chain, sessions, wakes, evals, launches), each attributed to an agent,
   with state, elapsed time, partition and submit time. Jobs on the account

src/outerloop/attempt.py

@@ -18,6 +18,7 @@
 import os
 import re
 import shutil
+import time
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass
 from dataclasses import replace as dc_replace
@@ -45,6 +46,7 @@
     ensure_regular_git_dir,
 )
 from outerloop.harness import Harness, SessionResult, default_binary, redact
+from outerloop.launchlog import append_ended, append_submitted
 from outerloop.markers import has_marker
 from outerloop.measure import DispatchedMeasurer, DispatchSettings
 from outerloop.orchestrator import (
@@ -84,6 +86,9 @@
     save_record,
     stamp_outage,
 )
+from outerloop.runstate import (
+    run_dir as run_dir_of,
+)
 from outerloop.syscall import CHANNEL_DIR_NAMES, MAX_ARTIFACT_BYTES, SyscallRequest, channel_dir
 from outerloop.syscall import ensure_excluded as syscall_excluded
 from outerloop.syscall import install_tool as syscall_install_tool
@@ -452,10 +457,34 @@ def _park_run(
                 "minutes": launch.minutes,
                 "artifacts": list(launch.artifacts),
                 **({"array": launch.array} if launch.array > 1 else {}),
+                **({"why": redact(launch.why, secrets)} if launch.why else {}),
             }
             for launch in parked.syscall.launches
         ]
         stage["syscall_note"] = redact(parked.syscall.note, secrets)
+        if parked.syscall.launches:
+            # the run's launch ledger (`history`, and the queue view's labels):
+            # ids align with launch_jobs order, as _stage_launch_job_ids reads them
+            if parked.launch_afterany:
+                launch_ids = afterany_ids(parked.launch_afterany)
+            elif parked.phase == "author-sleep":
+                launch_ids = list(job_ids)
+            else:
+                launch_ids = []
+            ledger_launches = tuple(
+                dc_replace(launch, why=redact(launch.why, secrets))
+                for launch in parked.syscall.launches
+            )
+            _best_effort(
+                "launch ledger",
+                lambda: append_submitted(
+                    run_dir_of(run_root, record.run_id),
+                    sleep=parked.sleeps_used,
+                    launches=ledger_launches,
+                    job_ids=launch_ids,
+                    at=now,
+                ),
+            )
         # (the session id the wake resumes is the record's own
         # resume_session_id, set below for every park — no stage duplicate)
         stage["launches_used"] = parked.launches_used
@@ -654,6 +683,26 @@ def launcher(sha: str, request: SyscallRequest) -> str:
     return launcher
 
 
+def _make_watcher(
+    dispatch: DispatchSettings, run_root: Path, run_id: str, workspace: Path, config: RunConfig
+) -> Callable[[], Any]:
+    """The session watcher for one run: a thread beside the harness that
+    answers `queue` and `history` from the channel (docs/design/session-watcher.md).
+    Shared by the first pass and every wake leg."""
+    from outerloop.watcher import SessionWatcher, WatcherContext
+
+    ctx = WatcherContext(
+        workspace=workspace,
+        run_root=run_root,
+        run_id=run_id,
+        target=config.target,
+        agent_id=config.agent_id,
+        compute=dispatch.compute,
+        gpu_partition=dispatch.gpu_partition,
+    )
+    return lambda: SessionWatcher(ctx)
+
+
 def _make_admission(dispatch: DispatchSettings, gpus: int = 0) -> Callable[[SyscallRequest], str]:
     """The admission check for this benchmark's launches: only GPU launches on a
     cluster queue are ever refused (admission.queue_saturated)."""
@@ -780,6 +829,7 @@ def _end(result: AttemptResult, drop_refs: list[str]) -> AttemptOutcome:
             minutes=int(item.get("minutes") or 1),
             artifacts=tuple(str(a) for a in item.get("artifacts", [])),
             array=int(item.get("array") or 1),
+            why=str(item.get("why") or ""),
         )
         for item in _stage_launches(record)
     )
@@ -795,6 +845,10 @@ def _end(result: AttemptResult, drop_refs: list[str]) -> AttemptOutcome:
         results = annotate_launch_states(results, _stage_launch_job_ids(record), status_of)
     launches_used = int(record.stage.get("launches_used", 0))  # type: ignore[call-overload]
     sleeps_used = int(record.stage.get("sleeps_used", 0))  # type: ignore[call-overload]
+    _best_effort(
+        "launch ledger",
+        lambda: append_ended(run_dir, sleep=sleeps_used, results=results, at=time.time()),
+    )
     gpu_hours_used = _reconcile_launch_hours(record, dispatch, bench.gpus, launches)
     wake_text = render_wake(
         results,
@@ -878,6 +932,7 @@ def changed_paths() -> list[str]:
             improve_prompt=wake_text,
             launcher=_make_launcher(dispatch, run_dir, workspace, run_id, gpus=bench.gpus),
             admission=_make_admission(dispatch, gpus=bench.gpus),
+            watcher=_make_watcher(dispatch, run_root, run_id, workspace, config),
             tree_of=lambda sha: ws.git("rev-parse", f"{sha}^{{tree}}").strip(),
             judged=judged or _stage_judged(record),
             launches_used=launches_used,
@@ -953,6 +1008,7 @@ def _stage_syscall_launches(record: RunRecord) -> tuple:
             minutes=int(item.get("minutes") or 1),
             artifacts=tuple(str(a) for a in item.get("artifacts", [])),
             array=int(item.get("array") or 1),
+            why=str(item.get("why") or ""),
         )
         for item in _stage_launches(record)
     )
@@ -1648,6 +1704,7 @@ def resume_run(
                             minutes=int(item.get("minutes") or 1),
                             artifacts=tuple(str(a) for a in item.get("artifacts", [])),
                             array=int(item.get("array") or 1),
+                            why=str(item.get("why") or ""),
                         )
                         for item in _stage_launches(record)
                     ),
@@ -2730,6 +2787,11 @@ def snapshot() -> str:
                 line_divergence=line_divergence,
                 launcher=launcher,
                 admission=admission,
+                watcher=(
+                    _make_watcher(dispatch, run_root, run_id, workspace, config)
+                    if dispatch is not None
+                    else None
+                ),
                 tree_of=lambda sha: ws.git("rev-parse", f"{sha}^{{tree}}").strip(),
             )
         except RunParked as p:
@@ -3123,7 +3185,6 @@ def _on_sigterm(signum: int, frame: object) -> None:
 def main() -> int:
     import argparse
     import os
-    import time
     from datetime import UTC, datetime
 
     arm_sigterm_containment()

src/outerloop/brief.py

@@ -392,9 +392,12 @@ def render(brief: SessionBrief) -> str:
             "inside your own session time — it costs no budget:",
             "",
             f"    python {_CHANNEL}/syscall launch --name <handle> "
-            "--minutes <N> [--array <K>] --artifact <repo-relative file> -- <command>",
+            '--minutes <N> [--array <K>] [--why "one line"] '
+            "--artifact <repo-relative file> -- <command>",
             f"    python {_CHANNEL}/syscall submit [--minutes <N>]",
             f"    python {_CHANNEL}/syscall siblings",
+            f"    python {_CHANNEL}/syscall queue",
+            f"    python {_CHANNEL}/syscall history",
             f"    python {_CHANNEL}/syscall sync",
             f"    python {_CHANNEL}/syscall sleep",
             "",
@@ -418,7 +421,10 @@ def render(brief: SessionBrief) -> str:
                 if brief.gpu_hour_budget > 0
                 else []
             ),
-            "`status` shows staged launches and remaining budget; `note ...` "
+            "`status` shows staged launches and remaining budget; `queue` shows "
+            "the kernel's jobs in the cluster queue right now — every agent's, "
+            "each launch with its `--why` — and `history` this run's launches and "
+            "how each ended, both within seconds while you work; `note ...` "
             "leaves a reminder echoed back to you on wake. `--artifact` must "
             "name a file your command actually writes, anywhere under the repo "
             f"tree — the `{_CHANNEL}/` channel does not exist in the job, so "

src/outerloop/climbboard.py

@@ -946,6 +946,9 @@ def queue_rows(root: Path, target: str, snapshot: list[dict[str, str]]) -> list[
                 "elapsed": str(job.get("elapsed", "")),
                 "partition": str(job.get("partition", "")),
                 "submitted": str(job.get("submitted", "")),
+                "reason": str(job.get("reason", "")),
+                "gres": str(job.get("gres", "")),
+                "limit": str(job.get("limit", "")),
                 "agent": agent,
                 "run_id": run_id,
             }

src/outerloop/compute.py

@@ -136,7 +136,17 @@ def to_argv(self) -> list[str]:
 
 # `reason` and `gres` feed launch admission (admission.queue_saturated: why a job
 # waits, whether it holds GPUs); the board reads the first six by key
-QUEUE_FIELDS = ("id", "name", "state", "elapsed", "partition", "submitted", "reason", "gres")
+QUEUE_FIELDS = (
+    "id",
+    "name",
+    "state",
+    "elapsed",
+    "partition",
+    "submitted",
+    "reason",
+    "gres",
+    "limit",
+)
 
 
 class Compute(Protocol):
@@ -149,6 +159,7 @@ def pending_reason(self, job_id: str) -> str: ...
     def job_partition(self, job_id: str) -> str: ...
     def active_job_names(self) -> list[str]: ...
     def queue_snapshot(self) -> list[dict[str, str]]: ...
+    def lane_load(self, partition: str) -> dict[str, int]: ...
     def job_id_for_name(self, name: str) -> str: ...
     def cancel(self, job_id: str) -> None: ...
 
@@ -270,7 +281,7 @@ def queue_snapshot(self) -> list[dict[str, str]]:
         on failure, like active_job_names."""
         try:
             result = self.runner(
-                ["squeue", "--me", "--noheader", "-o", "%i|%j|%T|%M|%P|%V|%r|%b"],
+                ["squeue", "--me", "--noheader", "-o", "%i|%j|%T|%M|%P|%V|%r|%b|%l"],
                 self.command_timeout_s,
             )
         except (OSError, subprocess.TimeoutExpired) as exc:
@@ -284,6 +295,30 @@ def queue_snapshot(self) -> list[dict[str, str]]:
                 rows.append(dict(zip(QUEUE_FIELDS, parts, strict=True)))
         return rows
 
+    def lane_load(self, partition: str) -> dict[str, int]:
+        """Node counts by state on a lane (a partition or a comma-separated
+        list), from sinfo: {"idle": 3, "mixed": 20, "allocated": 11}. Context
+        for the queue view, nothing a caller acts on. Empty when no lane is
+        named; raises SlurmQueryError on a failed query."""
+        if not partition:
+            return {}
+        try:
+            result = self.runner(
+                ["sinfo", "--noheader", "-p", partition, "-o", "%T %D"], self.command_timeout_s
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise SlurmQueryError(f"sinfo did not run: {exc}") from exc
+        if result.returncode != 0:
+            raise SlurmQueryError(f"sinfo failed ({result.returncode}): {result.stderr.strip()}")
+        load: dict[str, int] = {}
+        for line in result.stdout.splitlines():
+            parts = line.split()
+            if len(parts) != 2 or not parts[1].isdigit():
+                continue
+            state = parts[0].rstrip("*~#!%$@^-")  # sinfo's state flags (draining, no-respond...)
+            load[state] = load.get(state, 0) + int(parts[1])
+        return load
+
     def job_id_for_name(self, name: str) -> str:
         """The id of this user's PENDING/RUNNING job with exactly `name`, or
         "" if none. Authoritative for "is this still live" independent of any
@@ -478,6 +513,9 @@ def active_job_names(self) -> list[str]:
     def queue_snapshot(self) -> list[dict[str, str]]:
         return []
 
+    def lane_load(self, partition: str) -> dict[str, int]:
+        return {}  # no lanes in the monolith
+
     def job_id_for_name(self, name: str) -> str:
         return ""