diff --git a/changelog.d/964-long-run-handoff.md b/changelog.d/964-long-run-handoff.md new file mode 100644 index 00000000..7c667d31 --- /dev/null +++ b/changelog.d/964-long-run-handoff.md @@ -0,0 +1,6 @@ +Added +`scripts/dsh-superagent-run.sh` hands a long evolution run off a +dsh-superagent worker: it starts the run in a detached `tmux` session +(`setsid` fallback), writes `waiting.json` with the run id, round and run +directory, and returns at once, so a worker turn no longer holds its slot for +the length of the run (#964). diff --git a/docs/dsh-super-agent.md b/docs/dsh-super-agent.md index d1ba8066..2e498df0 100644 --- a/docs/dsh-super-agent.md +++ b/docs/dsh-super-agent.md @@ -118,6 +118,80 @@ Do not start a nested run from inside a run's turn. A worker that evolves a harness which itself starts runs multiplies cost with no ceiling — keep one run per Ticket, and make the second one a follow-up Ticket. +### Handing a run off instead of holding the slot + +A worker's background shell job is killed at 30 minutes; an evolution run is +often longer. Start it outside the worker's process group and give the slot +back: + +```bash +bash scripts/dsh-superagent-run.sh --store "$RUN_STORE" -- \ + agentdescent --store "$RUN_STORE" evolve spec.json --detach \ + --budget 5 --usd-per-call 0.01 +# waiting 3f9c1a… round 2 /runs/3f9c1a… +# report /runs/3f9c1a…/waiting.json +``` + +`--budget` needs `--usd-per-call`: the engine counts calls, and a per-call price +is the only way a dollar figure becomes a stop rule. With one and not the other +the CLI exits 2 before creating anything (`budget_usd needs usd_per_call`), so +the pair is what makes the run stop itself and write why. + +Two stores, one variable: the wrapper's `--store` is where the report goes, and +the run's store is the CLI's own `--store`, so pass the same value on both. The +wrapper cannot see the child's store — give it only to the wrapper and `evolve` +writes to the default store, the wrapper waits out its deadline and reports +`found: false`, and the read-back below cannot find the run. + +It starts the command in a detached `tmux` session (`setsid` when tmux is +absent), waits for the run's own `status.json`, writes `waiting.json` beside it +with `run_id`, `round`, `state`, `run_dir`, `driver`, `session` and `log`, and +exits 0 within seconds. Post the run id, the round and the run directory on the +Ticket and end the turn; nothing holds the slot. + +A store is reused for every Ticket, so it already holds earlier runs. The +wrapper only reports a `status.json` that appears or changes after it launches, +so a handoff into a used store reports `found: false` until its own run writes +one, instead of naming a finished run from last week. The tmux session carries a +per-launch suffix for the same reason: two handoffs into stores both called +`runs` get two sessions, not one shared, refused name. One live handoff per +store: two launched into one store inside the same wait window both read that +store's `waiting.json` and the second can report the first's run, so give each +handoff its own store or wait for the first to finish. + +Read it back from a follow-up Trigger, a person, or a scheduled check — the +wrapper adds no daemon and no scheduler. `--store` is a **global** flag, so it +goes before the verb: after the verb the CLI exits 2 with `unrecognized +arguments`, and without it the run is looked up in the default store +(`~/.agentdescent/runs`, or `$AGENTDESCENT_HOME/runs`), not in the handoff's. + +```bash +agentdescent --store "$RUN_STORE" status "$RUN_ID" --brief # one line; safe to poll +agentdescent --store "$RUN_STORE" show "$RUN_ID" # the artifact, once there is one +``` + +Measured against a handoff of a fake run, a second in — `status` keeps it to +one line, and `show` reports the gap instead of inventing a result: + +```console +$ agentdescent --store "$RUN_STORE" status fake-run-1 --brief +fake-run-1 running round 2 calls=0 skill_dir:./skill +$ agentdescent --store "$RUN_STORE" show fake-run-1 +run_id: fake-run-1 +state: running +note: no result yet +``` + +What the wrapper changes, and what it cannot: + +| failure | what happens | +|---|---| +| the worker's turn ends, or its background job is killed | with the `--detach` above, the run is in its own session — not the worker's process group, and not the wrapper's tmux session either, which holds only the launcher — so it keeps going when either ends | +| no tmux on the host (`setsid` fallback) | survives the group kill, not a reboot; the run directory is still the handoff | +| the host restarts | the run directory reads `failed`; `agentdescent resume ` continues it | +| an orphan outlives the Ticket | `agentdescent --store "$RUN_STORE" cancel ` sends SIGTERM to the run's process group; `tmux kill-session -t ` only stops the wrapper's own launcher, because a `--detach` run is not in that session | +| the budget runs out | the run stops itself and writes why into its status; the follow-up reads it, nobody is waiting | + ## 7. The check ```bash diff --git a/scripts/dsh-superagent-run.sh b/scripts/dsh-superagent-run.sh new file mode 100644 index 00000000..85e4312b --- /dev/null +++ b/scripts/dsh-superagent-run.sh @@ -0,0 +1,118 @@ +#!/usr/bin/env bash +# Start a long evolution run outside the worker's process group, then hand it +# back (#964). The worker reports `waiting` and ends its turn; a follow-up +# Trigger, a person or a scheduled check reads the run with `status`/`show`. +# +# dsh-superagent-run.sh --store DIR -- agentdescent --store DIR evolve spec.json --detach +# +# The wrapper's --store is where the report goes; the run's store is the CLI's +# own --store. Give both the same value, or the run lands in the default store +# and the report says found: false. +# +# tmux is preferred because dsh-superagent already depends on it and a session +# is not in the worker's process group, so the kill that ends the worker's +# background job does not reach the run. `setsid` is the fallback: it survives +# that kill, not a reboot -- the run directory, not the process, is what a +# worker hands on. +# +# Exit 0 once the run is detached and DIR/waiting.json is written. It never +# waits for the run: a handoff that blocks is the failure this replaces. +set -euo pipefail + +WAIT="${DSH_RUN_HANDOFF_WAIT:-10}" # seconds to wait for the run's status.json + +die() { echo "error: $*" >&2; exit 2; } +usage() { awk 'NR > 1 && /^#/ {sub(/^# ?/, ""); print; next} NR > 1 {exit}' "$0"; } + +STORE=""; CMD=() +while [ $# -gt 0 ]; do + case "$1" in + --store) [ $# -ge 2 ] || die "--store needs a directory"; STORE="$2"; shift 2 ;; + --) shift; CMD=("$@"); break ;; + -h|--help) usage; exit 0 ;; + *) die "unknown argument: $1 (try --help)" ;; + esac +done +[ -n "$STORE" ] || die "--store is required" +[ ${#CMD[@]} -gt 0 ] || die "nothing to start: give -- " +mkdir -p "$STORE" +STORE="$(cd "$STORE" && pwd)" +command -v python3 >/dev/null || die "python3 is required to write the handoff report" + +LOG="$STORE/worker.log" +# The store's basename is not unique -- a worker that hands off two runs keeps +# both in a directory called `runs` -- so the session carries a per-launch +# suffix. Two handoffs then never ask tmux for the same name (#1084 review). +SESSION="ad-$(printf '%s' "$(basename "$STORE")" | tr -c 'A-Za-z0-9_.-' '-')-$$-$RANDOM" +QUOTED="$(printf '%q ' "${CMD[@]}")" + +# What the store already holds, taken before the launch: a reused store always +# has earlier runs in it, and reporting the newest of those as this handoff is +# worse than reporting nothing (#1084 review). Only a status.json that this +# launch creates or rewrites may be reported. +SNAP="$STORE/.handoff-$$-$RANDOM.json" +trap 'rm -f "$SNAP"' EXIT +python3 - "$STORE" "$SNAP" <<'PY' +import glob, json, os, sys +store, out = sys.argv[1:3] +snap = {} +paths = [os.path.join(store, "status.json")] + sorted( + glob.glob(os.path.join(store, "*", "status.json"))) +for path in paths: + if os.path.isfile(path): + st = os.stat(path) + snap[path] = [st.st_mtime, st.st_size] +with open(out, "w", encoding="utf-8") as fh: + json.dump(snap, fh) +PY +if command -v tmux >/dev/null 2>&1; then + DRIVER=tmux + tmux new-session -d -s "$SESSION" -c "$PWD" "$QUOTED >> '$LOG' 2>&1" \ + || die "tmux refused the session $SESSION" +elif command -v setsid >/dev/null 2>&1; then + DRIVER=setsid + setsid sh -c "$QUOTED >> '$LOG' 2>&1" >/dev/null 2>&1 & +else + die "neither tmux nor setsid is available; this host cannot hand off a run" +fi + +# The run owns its status.json (`//status.json`, or the store +# itself when the caller owns the run directory). Absent is reported as such: +# an invented run id is worse than a gap a Trigger can see. +python3 - "$STORE" "$WAIT" "$DRIVER" "$SESSION" "$LOG" "$QUOTED" "$SNAP" <<'PY' +import glob, json, os, sys, time +store, wait, driver, session, log, command, snap_path = sys.argv[1:8] +before = json.load(open(snap_path, encoding="utf-8")) +deadline, found = time.time() + float(wait), None +while time.time() < deadline: + paths = [p for p in [os.path.join(store, "status.json")] + + sorted(glob.glob(os.path.join(store, "*", "status.json"))) + if os.path.isfile(p)] + fresh = [] + for path in paths: + try: + st = os.stat(path) + except OSError: + continue + seen = before.get(path) + if seen is None or [st.st_mtime, st.st_size] != seen: + fresh.append(path) + if fresh: + found = max(fresh, key=os.path.getmtime) + break + time.sleep(0.2) +run_dir = os.path.dirname(found) if found else store +status = json.load(open(found)) if found else {} +report = {"status": "waiting", "found": bool(found), + "run_id": status.get("run_id", "") if found else "", + "round": status.get("round", 0), "state": status.get("state", "unknown"), + "run_dir": run_dir, "driver": driver, + "session": session if driver == "tmux" else "", "log": log, + "command": command.strip()} +path = os.path.join(store, "waiting.json") +with open(path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") +print(f"waiting {report['run_id'] or '(no run yet)'} round {report['round']} {run_dir}") +print(f"report {path}") +PY diff --git a/tests/test_dsh_superagent_run.py b/tests/test_dsh_superagent_run.py new file mode 100644 index 00000000..33aebbb7 --- /dev/null +++ b/tests/test_dsh_superagent_run.py @@ -0,0 +1,253 @@ +"""The long-run handoff (#964): the page's command, and the run outliving a turn. + +Two checks, and neither needs a model: + +* the documented command shape is real -- the wrapper exists and takes the + flags the page passes it, and `agentdescent evolve` still answers `--detach`, + which is the flag the page tells a worker to use; +* gated behind ``AGENTDESCENT_LONG_RUN_SMOKE=1``, a fake run longer than the + caller's own limit survives the caller's turn: the wrapper returns while the + run is still going, the report names the run id and round, and no tmux + session (hence no orphan) is left once it finishes. +""" +import json +import os +import shlex +import shutil +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from agentdescent import cli, runstore + +REPO = Path(__file__).resolve().parent.parent +WRAPPER = REPO / "scripts" / "dsh-superagent-run.sh" +PAGE = REPO / "docs" / "dsh-super-agent.md" + +FAKE = ''' +import json, os, sys, time +store = sys.argv[1] +run_dir = os.path.join(store, "fake-run-1") +os.makedirs(run_dir, exist_ok=True) +path = os.path.join(run_dir, "status.json") + +def write(**kw): + with open(path, "w", encoding="utf-8") as fh: + json.dump(kw, fh) + +write(run_id="fake-run-1", round=2, state="running", pid=os.getpid()) +time.sleep(3) +write(run_id="fake-run-1", round=3, state="finished", pid=os.getpid()) +''' + + +def _page_handoff_example() -> str: + """The page's fenced example that starts a run through the wrapper.""" + page = PAGE.read_text(encoding="utf-8") + for block in page.split("```"): + if "scripts/dsh-superagent-run.sh" in block: + return block + raise AssertionError("the page has no example that starts the wrapper") + + +def _handoff_argv(): + """The page's example as the shell would hand it over: (wrapper, child) argv. + + Parsed, not matched as a substring. The block is a command a worker copies, + so its flags have to be the ones the CLI accepts *as a pair* -- a + substring check passes a command that exits 2, which is how `--budget` went + out without its `--usd-per-call` (#1084 review). + """ + block = _page_handoff_example() + # The shell's line continuations, then the `#` lines showing its output -- + # both are page furniture, not arguments. + code = block.replace("\\\n", " ") + code = " ".join(line for line in code.splitlines() + if not line.strip().startswith("#")) + argv = shlex.split(code) + assert argv[0] == "bash", f"the example does not invoke the wrapper with bash: {argv}" + assert "--" in argv, f"the example does not separate the wrapper's flags: {argv}" + cut = argv.index("--") + return argv[1:cut], argv[cut + 1:] + + +def _store_flag(argv) -> str: + """The value after `--store`, which is what must agree on both sides.""" + assert "--store" in argv, f"no --store in {argv}" + return argv[argv.index("--store") + 1] + + +def test_the_page_hands_the_run_the_store_the_wrapper_reports_into(): + # Two stores, one variable. The wrapper's `--store` is where the report + # goes; the run's store is the CLI's own global flag. Naming only the + # wrapper's store made `evolve` write to the default store, so the wrapper + # waited out its deadline and reported found: false, and the page's + # read-back could not find the run (#1084 review). + wrapper_argv, child_argv = _handoff_argv() + assert Path(wrapper_argv[1]).name == "dsh-superagent-run.sh", wrapper_argv + + # Drop the child's program name ("agentdescent", or `python -m + # agentdescent.cli`); the CLI's own parser then says whether the rest is a + # command it has. + start = next((i for i, token in enumerate(child_argv) + if token in ("agentdescent", "agentdescent.cli")), None) + assert start is not None, f"the example does not run agentdescent: {child_argv}" + cli_argv = child_argv[start + 1:] + try: + args = cli.build_parser().parse_args(cli_argv) + except SystemExit as e: # argparse exits 2 on a flag the CLI does not have + raise AssertionError( + f"the page's handoff command is not one the CLI accepts: {cli_argv} " + f"(exit {e.code})") from e + + assert _store_flag(wrapper_argv) == _store_flag(cli_argv), ( + "the report's store and the run's store must be the same value") + assert getattr(args, "detach", False), "the handoff must start the run detached" + # The section promises a run that stops itself and writes why, which is what + # --budget buys; the CLI refuses that flag without a per-call price. + assert args.budget is not None, ( + "the section documents a capped run, so the example carries --budget") + assert args.usd_per_call, ( + f"the page passes --budget {args.budget} without --usd-per-call, so the " + "documented command exits 2 and starts nothing") + runstore.require_budget_pair(args.budget, args.usd_per_call) + + +def test_the_page_documents_a_wrapper_that_exists_and_takes_its_flags(): + assert WRAPPER.is_file(), f"no {WRAPPER}" + # Invoked as `bash scripts/dsh-superagent-run.sh`, the way the page and + # `scripts/verify-dsh-superagent.sh` (committed 644) are, so the mode is not + # part of the contract; the command shape below is. + page = PAGE.read_text(encoding="utf-8") + assert "scripts/dsh-superagent-run.sh" in page, "the page does not name the wrapper" + assert "--store" in page and "--detach" in page + helped = subprocess.run(["bash", str(WRAPPER), "--help"], cwd=REPO, + capture_output=True, text=True, timeout=60) + assert helped.returncode == 0, helped.stderr + assert "--store" in helped.stdout, "the wrapper's usage omits --store" + evolve = subprocess.run( + [sys.executable, "-m", "agentdescent.cli", "evolve", "--help"], + cwd=REPO, capture_output=True, text=True, timeout=120) + assert evolve.returncode == 0, evolve.stderr + assert "--detach" in evolve.stdout, "the page tells a worker to pass a flag the CLI dropped" + + +@pytest.mark.skipif(os.environ.get("AGENTDESCENT_LONG_RUN_SMOKE") != "1", + reason="set AGENTDESCENT_LONG_RUN_SMOKE=1 to run the detached smoke") +def test_a_run_outlives_the_turn_that_started_it_and_leaves_no_orphan(tmp_path): + store = tmp_path / "runs" + fake = tmp_path / "fake_run.py" + fake.write_text(FAKE, encoding="utf-8") + + started = time.monotonic() + handed = subprocess.run( + ["bash", str(WRAPPER), "--store", str(store), "--", + sys.executable, str(fake), str(store)], + cwd=REPO, capture_output=True, text=True, timeout=60) + elapsed = time.monotonic() - started + + assert handed.returncode == 0, handed.stderr + assert elapsed < 2.0, f"the handoff blocked for {elapsed:.1f}s, so it held the slot" + report = json.loads((store / "waiting.json").read_text(encoding="utf-8")) + assert report["status"] == "waiting" + assert report["found"] is True + assert report["run_id"] == "fake-run-1" + assert report["round"] == 2 + assert Path(report["run_dir"]).name == "fake-run-1" + + # The caller's turn is over and the run is still going: that is the handoff. + pid = json.loads((Path(report["run_dir"]) / "status.json") + .read_text(encoding="utf-8"))["pid"] + os.kill(pid, 0) + + deadline = time.monotonic() + 30 + status = {} + while time.monotonic() < deadline: + status = json.loads((Path(report["run_dir"]) / "status.json") + .read_text(encoding="utf-8")) + if status["state"] == "finished": + break + time.sleep(0.2) + assert status["state"] == "finished", status + assert status["round"] == 3 + + if report["driver"] == "tmux" and shutil.which("tmux"): + # The run writes `finished` before its own process exits, and tmux + # destroys the session after the pane's command ends, so the session is + # still listed for a moment -- measured ~1.7s on a loaded host. Poll for + # it to go; only a session that stays is an orphan. `=` asks for that + # exact name: a bare target also prefix-matches a longer one, so + # `-t ad-runs` can answer about an unrelated ad-runs-2. + poll_started = time.monotonic() + deadline = poll_started + 15 + left = 0 + while time.monotonic() < deadline: + left = subprocess.run( + ["tmux", "has-session", "-t", "=" + report["session"]], + capture_output=True, text=True).returncode + if left != 0: + break + time.sleep(0.1) + waited = time.monotonic() - poll_started + assert left != 0, ( + f"orphan tmux session {report['session']!r} still listed after " + f"{waited:.1f}s (last `tmux has-session` rc={left})") + + +@pytest.mark.skipif(os.environ.get("AGENTDESCENT_LONG_RUN_SMOKE") != "1", + reason="set AGENTDESCENT_LONG_RUN_SMOKE=1 to run the detached smoke") +def test_a_reused_store_does_not_report_an_earlier_run(tmp_path): + # A store is reused for every Ticket, so it always holds earlier runs. The + # first poll used to take the newest `*/status.json` it could see and report + # that run -- a finished one -- as the handoff (#1084 review). Nothing here + # writes a status.json, so the report must say it found nothing. + store = tmp_path / "runs" + old = store / "old-run" + old.mkdir(parents=True) + (old / "status.json").write_text( + json.dumps({"run_id": "old-run", "round": 7, "state": "finished"}), + encoding="utf-8") + handed = subprocess.run( + ["bash", str(WRAPPER), "--store", str(store), "--", "sleep", "1"], + cwd=REPO, capture_output=True, text=True, timeout=60, + env=dict(os.environ, DSH_RUN_HANDOFF_WAIT="1")) + assert handed.returncode == 0, handed.stderr + report = json.loads((store / "waiting.json").read_text(encoding="utf-8")) + assert report["found"] is False, report + assert report["run_id"] == "", report + assert report["round"] == 0, report + assert os.path.realpath(report["run_dir"]) == os.path.realpath(str(store)), report + + +@pytest.mark.skipif(os.environ.get("AGENTDESCENT_LONG_RUN_SMOKE") != "1", + reason="set AGENTDESCENT_LONG_RUN_SMOKE=1 to run the detached smoke") +def test_two_handoffs_into_stores_called_runs_do_not_take_one_session(tmp_path): + # The session name came from the store's basename, so two handoffs into + # stores both called `runs` asked tmux for the same session and the second + # died with "tmux refused the session" (#1084 review). + if not shutil.which("tmux"): + pytest.skip("no tmux on this host: the setsid fallback has no session name") + fake = tmp_path / "fake_run.py" + fake.write_text(FAKE, encoding="utf-8") + reports = [] + for name in ("a", "b"): + store = tmp_path / name / "runs" + handed = subprocess.run( + ["bash", str(WRAPPER), "--store", str(store), "--", + sys.executable, str(fake), str(store)], + cwd=REPO, capture_output=True, text=True, timeout=60) + assert handed.returncode == 0, handed.stderr + reports.append(json.loads((store / "waiting.json").read_text(encoding="utf-8"))) + assert reports[0]["driver"] == "tmux", reports + assert reports[0]["session"] and reports[1]["session"], reports + assert reports[0]["session"] != reports[1]["session"], reports + + deadline = time.monotonic() + 15 + while time.monotonic() < deadline and any( + subprocess.run(["tmux", "has-session", "-t", "=" + report["session"]], + capture_output=True, text=True).returncode == 0 + for report in reports): + time.sleep(0.1)