diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index c74849e..10adeed 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -1501,7 +1501,9 @@ _herdr_scope_is_pane_inherited() { # in the shim heredoc, which stays byte-identical (test_issue23_24 S11). # # _mam_pane_claim_target : prints the server a new-session call -# targets ("" for the default one); rc 1 for any other call. +# names ("" for the ambient target, which lib_py.pane_lock resolves to the +# caller pane's server - the one herdr answers on - or the default server); +# rc 1 for any other call. _mam_pane_claim_target() { local sess="${HERDR_SESSION_NAME:-}" while [ "${1:-}" = "-L" ]; do sess="${2:-}"; shift 2 || return 1; done diff --git a/.agents/skills/lib_py/delegation/wrapper.py b/.agents/skills/lib_py/delegation/wrapper.py index 16efecb..a4ea77d 100644 --- a/.agents/skills/lib_py/delegation/wrapper.py +++ b/.agents/skills/lib_py/delegation/wrapper.py @@ -268,9 +268,14 @@ def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] = within DISPATCH_SIGNAL_WINDOW, or - never signalled - created within its own ack_timeout_sec: a producer that registers and injects without a waiter (create --onboard) waits that long for `started` - (review 55219be1 R2). A live pending record whose admission names - its current dispatch token is reported as `admitted`: it holds the - session as a running job does, whatever its place in the FIFO. + (review 55219be1 R2). Such a never-signalled record is reported as + `injected` and a live pending record whose admission names its + current dispatch token as `admitted`: both hold the session as a + running job does, whatever their place in the FIFO - the prompt is + already in the pane (review 5a09a13f F-A: in the FIFO an `injected` + record of the same second lost to a lower job id). A waiter's own + record is unsignalled only until its first scan (wait_open_jobs + signals first). stale: [(job_id, reason)] - non-terminal records nothing has touched for longer than that: a wrapper killed before or while dispatching, a zombie's unclosed record, an onboarding never acknowledged. They @@ -307,17 +312,21 @@ def open_jobs(registry_dir: str, session: str, job_id: str, now: Optional[int] = seen, window = _parse_ts(rec.get("created_at")), _int_field(rec, "ack_timeout_sec", DEFAULT_ACK_TIMEOUT_SEC) why = f"no dispatch signal and no started within {window}s" if seen is not None and now - seen <= window: - admitted = bool(rec.get(ADMITTED_TOKEN_FIELD)) and \ - rec.get(ADMITTED_TOKEN_FIELD) == rec.get("dispatch_token") - live.append((str(rec.get("created_at") or ""), other, "admitted" if admitted else status)) + if not rec.get(DISPATCH_SIGNAL_FIELD): + kind = "injected" + elif rec.get(ADMITTED_TOKEN_FIELD) and rec.get(ADMITTED_TOKEN_FIELD) == rec.get("dispatch_token"): + kind = "admitted" + else: + kind = status + live.append((str(rec.get("created_at") or ""), other, kind)) else: stale.append((other, f"pending, {why}")) return live, stale def _first_in_line(live, mine) -> bool: - """No live running or admitted job, and `mine` earliest among the live - pending ones (FIFO by (created_at, job_id)).""" + """No live running, admitted or injected job, and `mine` earliest among + the live pending ones (FIFO by (created_at, job_id)).""" if any(st != "pending" for _c, _j, st in live): return False return min([(c, j) for c, j, _st in live] + [mine]) == mine @@ -449,9 +458,12 @@ def wait_open_jobs(a: argparse.Namespace) -> str: running (_started_own). The bound is min(--wait-s, this job's remaining wall clock). The identity is pinned at the start: --dispatch-token (the token the wrapper read for this - attempt), else the record's token read once. Every DISPATCH_SIGNAL_EVERY s - the waiter signals on its own record so later waiters see it alive; a - waiter whose wrapper is gone stops (OwnerGone) and its signal goes stale. + attempt), else the record's token read once. The waiter signals on its own + record before its first scan - an unsignalled live record is one a + producer injected without a waiter (open_jobs: `injected`), and this one + is not - and every DISPATCH_SIGNAL_EVERY s after, so other waiters see it + alive; a waiter whose wrapper is gone stops (OwnerGone) and its signal + goes stale. With --hold the admission is kept alive after the answer (_start_hold).""" me = _load_record(a.registry_dir, a.job) or {} token = getattr(a, "dispatch_token", None) or me.get("dispatch_token") @@ -465,6 +477,9 @@ def wait_open_jobs(a: argparse.Namespace) -> str: rec = _load_record(a.registry_dir, a.job) if not _own(rec, token, a.agent_session): return "clear" if _started_own(rec, token, a.agent_session) else "superseded" + if last_signal is None or now - last_signal >= DISPATCH_SIGNAL_EVERY: + _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session) + last_signal = now live, stale = open_jobs(a.registry_dir, a.agent_session, a.job, now) for other, reason in stale: if other not in reported: @@ -476,9 +491,6 @@ def wait_open_jobs(a: argparse.Namespace) -> str: return "clear" if now >= deadline: return "busy " + " ".join(j for _c, j, _st in live) - if last_signal is None or now - last_signal >= DISPATCH_SIGNAL_EVERY: - _signal_dispatch_wait(a.registry_dir, a.job, token, a.agent_session) - last_signal = now _sleep(a, 1) diff --git a/.agents/skills/lib_py/pane_lock.py b/.agents/skills/lib_py/pane_lock.py index 42815e3..ef18d04 100644 --- a/.agents/skills/lib_py/pane_lock.py +++ b/.agents/skills/lib_py/pane_lock.py @@ -8,7 +8,13 @@ and the later `agent start` failed with agent_pane_busy. The shim now holds one lock per server from the choice to the claim. pane_lock.py path --session S the lock file of herdr session S, under the - HOME whose socket the session uses + HOME whose socket the session uses. S empty + (or `default`) is the shim's ambient target: + herdr then answers on the server of the pane + this runs in (HERDR_SESSION / HERDR_SOCKET_PATH, + as lib.sh _herdr_caller_server reads them), or + the default server outside any pane - the + lock is that server's (review e8e471ff R1) pane_lock.py acquire --fd N flock(LOCK_EX) on the inherited fd N; prints `locked`, or `timeout` once MAM_PANE_CLAIM_WAIT_S (default 120, minimum 1) has passed @@ -32,9 +38,26 @@ from lib_py import ints DEFAULT_WAIT_S = 120 +def caller_server(env: Mapping[str, str]) -> str: + """The server whose pane this process runs in, by the pane's exports: + HERDR_SESSION when present, else HERDR_SOCKET_PATH's layout + ($HOME/.config/herdr/herdr.sock -> default, .../sessions//herdr.sock + -> name); "" outside a pane or for an unrecognised layout.""" + if env.get("HERDR_SESSION"): + return env["HERDR_SESSION"] + sock = env.get("HERDR_SOCKET_PATH") or "" + if sock.endswith("/.config/herdr/herdr.sock"): + return "default" + if sock.endswith("/herdr.sock") and "/.config/herdr/sessions/" in sock: + return sock[:-len("/herdr.sock")].rsplit("/", 1)[-1] + return "" + + def lock_path(session: Optional[str], env: Mapping[str, str]) -> str: home = env.get("HOME") or env.get("HOME_DIR") or os.path.expanduser("~") - return os.path.join(home, ".local", "state", "mam", "pane-lock", (session or "default") + ".lock") + if session in (None, "", "default"): + session = caller_server(env) or "default" + return os.path.join(home, ".local", "state", "mam", "pane-lock", session + ".lock") def wait_s(env: Mapping[str, str]) -> int: diff --git a/tests/test_issue26_ints.py b/tests/test_issue26_ints.py index 99dad06..703bf3d 100644 --- a/tests/test_issue26_ints.py +++ b/tests/test_issue26_ints.py @@ -88,7 +88,8 @@ INPUTS = ["30", "0700", "0900", " 30 ", "0", "-5", "abc", "", "1e3", "9999999999 # DELEGATE_JOB_/RECONCILE_…:-…}` that is neither here nor normalised # through INT_KNOBS, so a new knob has to be classified to land. # Issue #17 item 4: MAM_DELEGATE_FAIL_CLOSED and MAM_STALL_WATCH are read by -# lib_py/delegation/wrapper.py now, not by shell code. +# lib_py/delegation/wrapper.py now, not by shell code; MAM_PANE_CLAIM_WAIT_S +# (minimum 1, default 120) by lib_py/pane_lock.py (test_issue50_pane_lock P5). NOT_ARITHMETIC = { "MAM_ALLOW_UNSIGNED_DISPATCH": "string compare", "MAM_DELEGATE_JOB_NO_MAIN": "string compare", "MAM_NO_AUTOSTART": "string compare", "MAM_SHIM_DEBUG": "string compare", diff --git a/tests/test_issue50_dispatch.py b/tests/test_issue50_dispatch.py index 8940d0c..298d677 100644 --- a/tests/test_issue50_dispatch.py +++ b/tests/test_issue50_dispatch.py @@ -115,6 +115,8 @@ def test_u1_open_jobs_table(tmp_path, status, age, signal, ack, want): got = {j for _c, j, _s in live}, {j for j, _r in stale} if want == "live": assert got == ({other}, set()) + if status == "pending": # never signalled: injected by its producer, a holder; signalled: a waiter in the FIFO + assert live[0][2] == ("injected" if signal is None else "pending") elif want == "stale": assert got == (set(), {other}), got assert stale[0][1].startswith(status + ", no ") @@ -169,8 +171,8 @@ def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending # A ends: clear _edit(tmp_path, a, status="completed") assert decide.wait_open_jobs(_ns(tmp_path, me)) == "clear" - # an earlier live pending waiter B: this job waits behind it (busy at the bound) - b = _register(tmp_path, created_at=_iso(time.time() - 30)) + # an earlier live pending waiter B (signalled: a waiter is on it): this job waits behind it (busy at the bound) + b = _register(tmp_path, created_at=_iso(time.time() - 30), dispatch_wait_at=_iso(time.time())) assert decide.wait_open_jobs(_ns(tmp_path, me)) == f"busy {b}" # a later one does not hold this job up _edit(tmp_path, b, created_at=_iso(time.time() + 30)) @@ -180,11 +182,32 @@ def test_u2_clear_only_when_no_live_running_and_this_job_is_the_earliest_pending def test_u2b_a_tie_on_created_at_is_broken_by_job_id(tmp_path): same = _iso(time.time()) _register_as(tmp_path, "aaaa0001", created_at=same) - _register_as(tmp_path, "aaaa0002", created_at=same) + _register_as(tmp_path, "aaaa0002", created_at=same, dispatch_wait_at=same) # a live waiter assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0001")) == "clear" assert decide.wait_open_jobs(_ns(tmp_path, "aaaa0002")) == "busy aaaa0001" +# review 5a09a13f F-A: a live pending record nobody signals on is one its +# producer injected without a waiter (create --onboard, a job sent by hand); +# the prompt is in the pane, so it holds the session - never a FIFO peer that +# a lower job id in the same second, or an earlier second, could pass + +@pytest.mark.parametrize("onboarding,waiter", [("ffffffff", "00000001"), ("00000001", "ffffffff")]) +def test_u2h_an_injected_record_holds_the_session_whatever_its_place_in_the_fifo(tmp_path, onboarding, waiter): + same = _iso(time.time()) + _register_as(tmp_path, onboarding, created_at=same) # unsignalled, ack 180: live + _register_as(tmp_path, waiter, created_at=same) + assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}" + assert "dispatch_admitted_token" not in _rec(tmp_path, waiter) and "dispatch_wait_at" in _rec(tmp_path, waiter) + assert "dispatch_wait_at" not in _rec(tmp_path, onboarding) + # registered after the waiter (which was still in its preflight): the same + _edit(tmp_path, onboarding, created_at=_iso(_at(same) + 1)) + assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == f"busy {onboarding}" + # control: the same record with a waiter's signal on it is a FIFO peer + _edit(tmp_path, onboarding, created_at=same, dispatch_wait_at=same) + assert decide.wait_open_jobs(_ns(tmp_path, waiter)) == ("clear" if waiter < onboarding else f"busy {onboarding}") + + def test_u2c_a_dead_waiter_is_ignored_and_reported(tmp_path, capsys): b = _register(tmp_path, ack=5, created_at=_iso(time.time() - 60), dispatch_wait_at=_iso(time.time() - WINDOW - 1)) me = _register(tmp_path) @@ -473,11 +496,13 @@ def test_u4_open_jobs_prints_live_and_stale(tmp_path): a = _register(tmp_path, status="running") _event(tmp_path, a, time.time()) b = _register(tmp_path, created_at=_iso(time.time() - 1000)) + c = _register(tmp_path) # within its ack budget, no waiter on it me = _register(tmp_path) res = _cli(tmp_path, "open-jobs", "--job", me, "--agent-session", "herdr:s", "--registry-dir", str(tmp_path)) assert res.returncode == 0, res.stderr assert f"{a} running" in res.stdout assert f"{b} stale pending, no dispatch signal and no started within 180s" in res.stdout + assert f"{c} injected" in res.stdout def _hold_pids(jid): diff --git a/tests/test_issue50_dispatch_herdr.py b/tests/test_issue50_dispatch_herdr.py index c56de70..1e55d87 100644 --- a/tests/test_issue50_dispatch_herdr.py +++ b/tests/test_issue50_dispatch_herdr.py @@ -19,6 +19,7 @@ so the herdr idle wait stays and the registry wait runs before it. from __future__ import annotations import calendar +import fcntl import json import os import re @@ -522,15 +523,29 @@ def _onboarding(box: Box, ack: int) -> str: def test_c10_a_pending_onboarding_within_its_ack_budget_blocks_the_session_past_the_signal_window(box): + # review 5a09a13f F-A: the onboarding and the first job are registered in + # the same second (create registers, then injects; the job's wrapper + # registers a moment later), where the FIFO's (created_at, job_id) once + # let a lower id pass the injected onboarding; the second job comes a + # different second + while time.time() % 1 > 0.05: + time.sleep(0.005) t0 = time.time() - a = _onboarding(box, 180) + a = box.register(ack_timeout_sec=180) known = box.job_ids() for label in ("before", "after"): if label == "after": time.sleep(max(0.0, t0 + 31.5 - time.time())) # A is now older than the signal window proc = box.submit(prompt="work " + label, MAM_DISPATCH_WAIT_S="3") + if label == "before": + prompt = box.herdr("agent", "prompt", SESSION, "Onboarding " + a) + assert prompt.returncode == 0, prompt.stdout + prompt.stderr b = box.wait_new_job(known) known.add(b) + print(label, "created", a, box.record(a)["created_at"], b, box.record(b)["created_at"], + "registered at +%.2fs" % (time.time() - t0)) + if label == "before": + assert box.record(b)["created_at"] == box.record(a)["created_at"], "the same-second tie this exercises" rc, out, err = box.finish(proc) assert rc == 5, (label, out + err) rec = box.record(b) @@ -727,6 +742,49 @@ def test_a3_three_concurrent_creates_in_one_folder_take_distinct_panes(local_mqt _teardown(b) +def test_s2_an_ambient_and_a_named_call_share_the_servers_pane_claim(): + """review e8e471ff R1: a call with the ambient target - no -L, HERDR_SESSION_NAME + empty or `default` - from a pane of server S is answered by S (the pane's + HERDR_SESSION / HERDR_SOCKET_PATH), so it claims under S's lock, the one a + call naming S takes: held, both wait; racing, they take distinct panes.""" + b = _make_box(f"t50-{uuid.uuid4().hex[:8]}") + ambient = dict(HERDR_SESSION_NAME="default", HERDR_SESSION=b.sess, + HERDR_SOCKET_PATH=str(b.home / ".config/herdr/sessions" / b.sess / "herdr.sock")) + new_session = 'herdr new-session -d -s {name} -c "$WORKSPACE_ROOT" "claude --dangerously-skip-permissions"' + try: + assert b.sh("herdr agent list").returncode == 0 # the server is up before the race + lock = b.home / ".local/state/mam/pane-lock" / f"{b.sess}.lock" + lock.parent.mkdir(parents=True, exist_ok=True) + with lock.open("a") as held: + fcntl.flock(held, fcntl.LOCK_EX) + for env in ({}, ambient): + res = b.sh(new_session.format(name="t50-held-creator-claude"), MAM_PANE_CLAIM_WAIT_S="1", **env) + assert res.returncode == 1 and "another pane claim" in res.stderr, (env, res.stdout, res.stderr) + assert json.loads(b.herdr("agent", "list").stdout)["result"]["agents"] == [] + _hold_pane_creation(b, 8) + procs = [] + for name, env in (("t50-named-creator-claude", {}), ("t50-ambient-creator-claude", ambient)): + if procs: + time.sleep(2) + out = tempfile.TemporaryFile("w+", dir=str(b.home)) + err = tempfile.TemporaryFile("w+", dir=str(b.home)) + procs.append((name, subprocess.Popen(["bash", "-c", f'source "{SKILLS}/lib.sh"\n' + new_session.format(name=name)], + stdout=out, stderr=err, text=True, cwd=str(b.ctl), env=b.env(**env)), out, err)) + outcome = {} + for name, proc, out, err in procs: + rc = proc.wait(timeout=120) + out.seek(0); err.seek(0) + outcome[name] = (rc, out.read(), err.read()) + assert all(o[0] == 0 for o in outcome.values()), {n: (o[0], o[2][-800:]) for n, o in outcome.items()} + assert "agent_pane_busy" not in "".join(o[1] + o[2] for o in outcome.values()) + panes = _agent_panes(b) + assert sorted(panes) == sorted(n for n, *_ in procs), panes + assert len({p for p, _c in panes.values()}) == 2, ("two agents on one pane", panes) + assert all(Path(c).resolve() == b.ctl.resolve() for _p, c in panes.values()), panes + finally: + _teardown(b) + + def test_s_three_concurrent_new_sessions_in_one_folder_take_distinct_panes(): """The shim alone, no create around it: three `new-session` into one cwd on one running server at once. Before the pane claim was serialised, two of diff --git a/tests/test_issue50_pane_lock.py b/tests/test_issue50_pane_lock.py index 4237b47..809bb66 100644 --- a/tests/test_issue50_pane_lock.py +++ b/tests/test_issue50_pane_lock.py @@ -55,6 +55,31 @@ def test_p1_the_lock_file_is_per_server_under_the_sessions_home(tmp_path): assert res.stdout.strip() == pane_lock.lock_path("t50-x", env), res.stderr +# review e8e471ff R1: the ambient target (no -L, HERDR_SESSION_NAME empty or +# `default`) is answered by the server of the pane the caller runs in - the +# pane's HERDR_SESSION / HERDR_SOCKET_PATH say which - so that is the lock; +# only outside a pane is it the default server's + +def test_p1b_the_ambient_target_locks_the_callers_server(tmp_path): + home = {"HOME": str(tmp_path)} + sock = {"HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/sessions/t50-x/herdr.sock")} + named = pane_lock.lock_path("t50-x", home) + for ambient in ("", "default", None): + assert pane_lock.lock_path(ambient, {**home, **sock}) == named, ambient + assert pane_lock.lock_path(ambient, {**home, "HERDR_SESSION": "t50-x"}) == named, ambient + assert pane_lock.lock_path(ambient, {**home, "HERDR_SESSION": "t50-x", # HERDR_SESSION first + "HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/sessions/t50-y/herdr.sock")}) == named + assert pane_lock.lock_path(ambient, {**home, "HERDR_SOCKET_PATH": str(tmp_path / ".config/herdr/herdr.sock")}) \ + == pane_lock.lock_path("default", home) == str(tmp_path / ".local/state/mam/pane-lock/default.lock") + assert pane_lock.lock_path(ambient, {**home, "HERDR_SOCKET_PATH": "/elsewhere/herdr.sock"}).endswith("/default.lock") + assert pane_lock.lock_path(ambient, home).endswith("/default.lock") + assert pane_lock.lock_path("t50-y", {**home, **sock}) != named, "a named target is that server's, wherever it is called from" + outside = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")} # this test may run in a pane + res = subprocess.run([sys.executable, "-m", "lib_py.pane_lock", "path", "--session", ""], capture_output=True, text=True, + env=dict(outside, PYTHONPATH=str(SKILLS), HOME=str(tmp_path), **sock), timeout=60) + assert res.stdout.strip() == named, res.stderr + + def test_p2_a_free_lock_is_taken_at_once_and_a_held_one_times_out(tmp_path): lock = tmp_path / "s.lock" t0 = time.time()