From 5c3c2eb861a73552fc084aad1e1a3ca4f600d75a Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 30 Oct 2026 00:15:31 +0900 Subject: [PATCH] fix(reset): refuse id-less claude/grok targets, gate submit during a reset, validate the spec (#78) Review round 1 of PR #82: - A running claude/grok row without a recorded id is refused by the precheck (stop would fill it with a folder-wide guess the purge then deletes); record-id never records a pre-assigned guess, and a target is skipped (id_unverified) when the row's id after the normal stop is not the one recorded before it. - reset takes the update admission lock exclusively while it publishes .mam/loop-guard-active and .mam/reset-in-progress; delegate submit refuses on a live reset marker (mam_reset_admit). Before each target's stop its open jobs are re-read (skipped as open_job, never cancelled), and a row whose agent/server no longer matches the spec is skipped. - read_spec validates every stored field (herdr_server required, never defaulted) and rejects duplicate names. - Signal tests wait 60 s for the kill-session gate (this Mac needs ~18 s). Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib.sh | 17 + .agents/skills/lib_py/reset.py | 117 ++++-- .../multi-agent-mux-delegate-job | 5 +- .agents/skills/multi-agent-mux-reset/SKILL.md | 16 +- .../multi-agent-mux-reset/scripts/reset.sh | 62 ++- tests/test_issue78_reset.py | 374 +++++++++++++++++- 6 files changed, 529 insertions(+), 62 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index e1bb19f..9d22298 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -3348,6 +3348,23 @@ mam_update_admit_release() { eval "exec ${1:-7}>&-" } +# Issue #78: a reset (multi-agent-mux-reset) publishes .mam/reset-in-progress +# under the exclusive admission lock before its precheck. Delegate submit checks +# it while it holds that workspace's admission, so a job is either registered +# before the reset's precheck (which refuses on it) or not at all. +# mam_reset_admit [workspace]: rc 0 no live reset, 1 refused (stderr). +mam_reset_admit() { + local root="${1:-$WORKSPACE_ROOT}" marker st + marker="$root/.mam/reset-in-progress" + [ -f "$marker" ] || return 0 + # shellcheck disable=SC1091 + declare -F _mam_lock_holder_state >/dev/null || source "$SKILL_DIR/multi-agent-mux-loop/scripts/loop_lock.sh" + st="$(_mam_lock_holder_state "$marker")" + [ "$st" != "stale" ] || return 0 + echo "ERROR: a reset of $root is in progress (marker $marker, holder $st); its agents are being purged and recreated; try again when it has finished" >&2 + return 1 +} + # mam_update_admit_at : what create and resume take for an agent they # start in - mam_update_admit for the control root (fd 7) and for every # other installed workspace at or above (fds 20 and up), whose update diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py index 6aaf6e9..7beaf54 100644 --- a/.agents/skills/lib_py/reset.py +++ b/.agents/skills/lib_py/reset.py @@ -20,7 +20,7 @@ import sys import tempfile from typing import Dict, List, Mapping, Optional, Tuple -from lib_py.agents.base import AGENT_SESSION_ENV +from lib_py.agents.base import AGENT_SESSION_ENV, POST_DISCOVERED from lib_py.agents.registry import agent_of_row, get_adapter, own_key from lib_py.update import open_jobs, row_server @@ -40,10 +40,21 @@ def spec_path(workspace: str) -> str: return os.path.join(_real(workspace), SPEC_REL) +def _rows_by_name(state: dict) -> Dict[str, dict]: + return { + str(s.get("name")): s + for s in (state.get("herdr_sessions") or []) + if isinstance(s, dict) and s.get("name") + } + + def _state(workspace: str = "") -> dict: raw = os.environ.get("MAM_STATE_JSON") if raw is not None: return json.loads(raw or "{}") or {} + # Only --dry-run gets here (reset.sh passes MAM_STATE_JSON everywhere + # else): it does not source lib.sh, and reads a copy of the DB because a + # read on macOS rewrites -shm, so the dry run changes nothing under .mam. yaml_path = os.environ.get("AGENT_SESSIONS_YAML") or ( os.path.join(_real(workspace), ".mam", "agent-sessions.yaml") if workspace else "" ) @@ -136,6 +147,15 @@ def select_targets(workspace: str, state: dict) -> dict: key = own_key(agent) raw_id = row.get(key) old_id = str(raw_id).strip() if isinstance(raw_id, str) and raw_id.strip() else None + if old_id is None and get_adapter(agent).id_lifecycle != POST_DISCOVERED: + # Its id is given at spawn, so none on the row means none was ever + # pinned (the monitor defers an ambiguous one); stop would fill it + # with a folder-wide guess that the purge would then delete. + errors.append( + "running session %s (%s) has no recorded conversation id; record it (monitor) " + "or stop/purge it by hand first" % (name, agent) + ) + continue targets.append({ "name": name, "agent": agent, @@ -219,14 +239,28 @@ def read_spec(workspace: str) -> dict: agents = doc.get("agents") if not isinstance(agents, list): raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path) + names = set() for item in agents: if not isinstance(item, dict): raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path) name = item.get("name") agent = item.get("agent") role = item.get("role") - if not (isinstance(name, str) and name and isinstance(agent, str) and agent and isinstance(role, str) and role): + server = item.get("herdr_server") + old_id = item.get("old_id") + if not ( + all(isinstance(v, str) and v for v in (name, agent, role, server)) + and isinstance(item.get("herdr_workspace"), str) + and "old_id" in item + and (old_id is None or (isinstance(old_id, str) and old_id)) + and isinstance(item.get("old_created_at"), str) + ): raise SpecError("spec file %s has an incomplete agent entry; check it, then remove it by hand" % path) + if name in names: + raise SpecError( + "spec file %s lists %s twice; check it, then remove it by hand" % (path, name) + ) + names.add(name) adapter = get_adapter(agent) if adapter is None or adapter.name != agent or not own_key(agent): raise SpecError( @@ -245,12 +279,7 @@ def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[st """ ws_real = _real(workspace) spec = read_spec(ws_real) - rows_by_name = { - str(s.get("name")): s - for s in (state.get("herdr_sessions") or []) - if isinstance(s, dict) and s.get("name") - } - row = rows_by_name.get(session_name) + row = _rows_by_name(state).get(session_name) if row is None: return None changed = False @@ -262,6 +291,10 @@ def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[st old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None if old_id: return old_id + # Only a post-discovered id is captured by this pane's pid (#40); a + # pre-assigned agent's is a folder-wide guess, never recorded. + if get_adapter(entry["agent"]).id_lifecycle != POST_DISCOVERED: + return None cur_created_at = str(row.get("herdr_session_created_at") or "") old_created_at = str(entry.get("old_created_at") or "") if cur_created_at != old_created_at: @@ -433,15 +466,22 @@ def precheck( ) ) - rows_by_name = { - str(s.get("name")): s - for s in (state.get("herdr_sessions") or []) - if isinstance(s, dict) and s.get("name") - } - blockers.extend(_caller_blockers(targets, rows_by_name, env, ancestors=ancestors)) + blockers.extend(_caller_blockers(targets, _rows_by_name(state), env, ancestors=ancestors)) return blockers +def target_open_jobs(workspace: str, state: dict, env: Mapping[str, str], session_name: str) -> List[str]: + """Open (or unreadable) job records addressed to one target, re-read just + before its stop: a job registered after the precheck is never cancelled.""" + out: List[str] = [] + for j in open_jobs(_real(workspace), state, env): + if j.get("unreadable"): + out.append("job record %s cannot be read" % j["path"]) + elif j["agent_session"] == "herdr:%s" % session_name: + out.append("job %s (%s) is open" % (j["job_id"], j["status"])) + return out + + # --------------------------------------------------------------------------- D1: entries shell lines def format_entry_line(entry: dict, row: Optional[dict]) -> str: @@ -454,16 +494,27 @@ def format_entry_line(entry: dict, row: Optional[dict]) -> str: ("R_NAME", str(entry.get("name") or "")), ("R_AGENT", str(entry.get("agent") or "")), ("R_ROLE", str(entry.get("role") or "")), - ("R_SERVER", str(entry.get("herdr_server") or "default")), + ("R_SERVER", str(entry.get("herdr_server") or "")), ("R_LABEL", str(entry.get("herdr_workspace") or "")), ("R_OLD_ID", str(entry.get("old_id") or "")), ("R_ROW_ID", row_id), + ("R_ROW_AGENT", (agent_of_row(row) or "") if row else ""), + ("R_ROW_SERVER", str(row_server(row)) if row else ""), ] return " ".join("%s=%s" % (k, shlex.quote(v)) for k, v in pairs) # --------------------------------------------------------------------------- D8: result table +SKIP_RESULTS = { + "no_id": "skipped: no conversation id recorded", + "purge_refused": "skipped: purge refused", + "open_job": "skipped: job open", + "id_unverified": "skipped: id not verified", + "changed": "skipped: row changed", +} + + def _parse_skipped(items: Optional[List[str]]) -> Dict[str, str]: out: Dict[str, str] = {} for raw in items or []: @@ -497,11 +548,7 @@ def build_table( failed_spec: Optional[str] = None, ) -> Tuple[str, int]: """Format the D8 result table and compute the D9 exit code (0, 1, 4, or 5).""" - rows_by_name = { - str(s.get("name")): s - for s in (state.get("herdr_sessions") or []) - if isinstance(s, dict) and s.get("name") - } + rows_by_name = _rows_by_name(state) skipped_map = _parse_skipped(skipped_items) failed = _parse_failed(failed_spec) failed_name = failed[0] if failed else None @@ -539,11 +586,7 @@ def build_table( result_col = "not reached" has_skip_or_incomplete = True elif name in skipped_map: - kind = skipped_map[name] - if kind == "no_id": - result_col = "skipped: no conversation id recorded" - else: - result_col = "skipped: purge refused" + result_col = SKIP_RESULTS.get(skipped_map[name], SKIP_RESULTS["purge_refused"]) has_skip_or_incomplete = True elif r_state == "done": result_col = "done" @@ -598,11 +641,7 @@ def format_dry_run( refusals: List[str] = list(extra_refusals or []) refusals.extend(precheck(ws_real, state, env)) - rows_by_name = { - str(s.get("name")): s - for s in (state.get("herdr_sessions") or []) - if isinstance(s, dict) and s.get("name") - } + rows_by_name = _rows_by_name(state) sel = select_targets(ws_real, state) spath = spec_path(ws_real) action_map = { @@ -628,7 +667,7 @@ def format_dry_run( e["name"], e["agent"], e["role"], - e.get("herdr_server") or "default", + e["herdr_server"], e.get("herdr_workspace") or "", e.get("old_id") or "-", e.get("old_created_at") or "-", @@ -694,6 +733,10 @@ def main(argv: Optional[List[str]] = None) -> int: s.add_argument("--session", default="") s.add_argument("--warn-new", action="store_true") + s = sub.add_parser("open-jobs") + s.add_argument("--workspace", required=True) + s.add_argument("--session", required=True) + s = sub.add_parser("record-id") s.add_argument("--workspace", required=True) s.add_argument("--session", required=True) @@ -766,17 +809,19 @@ def main(argv: Optional[List[str]] = None) -> int: % (spec_path(ws), ", ".join(extra)), file=sys.stderr, ) - rows_by_name = { - str(s.get("name")): s - for s in (state.get("herdr_sessions") or []) - if isinstance(s, dict) and s.get("name") - } + rows_by_name = _rows_by_name(state) for e in spec["agents"]: if a.session and e.get("name") != a.session: continue print(format_entry_line(e, rows_by_name.get(e["name"]))) return 0 + if a.cmd == "open-jobs": + lines = target_open_jobs(ws, state, os.environ, a.session) + for line in lines: + print(line) + return 1 if lines else 0 + if a.cmd == "record-id": try: record_old_id(ws, state, a.session) diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index b3e97f6..259b0ca 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -306,7 +306,8 @@ cmd_submit() { # its row's, or with no row the live agent's on the server delivery uses) # runs; each is held, fd 7 then 20.., until the record is written (lib.sh # mam_update_admit). A dry-run record is never dispatched and is not - # counted by the update. + # counted by the update. Issue #78: likewise while a reset of one runs + # (mam_reset_admit, checked under the same admission). local admit_fds=() target_ws tws fd=20 if [[ "$DRY_RUN" != "1" ]]; then target_ws="$(MAM_STATE_JSON="$(load_state_json)" PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$PY" -m lib_py.update \ @@ -315,9 +316,11 @@ cmd_submit() { --herdr "$(_resolve_real_herdr_path || true)")" \ || { echo "ERROR: cannot resolve the workspace of $AGENT_SESSION; nothing was registered" >&2; exit 1; } mam_update_admit "$WORKSPACE_ROOT" || exit 1 + mam_reset_admit "$WORKSPACE_ROOT" || exit 1 while IFS= read -r tws; do [[ -n "$tws" ]] || continue mam_update_admit "$tws" "$fd" || exit 1 + mam_reset_admit "$tws" || exit 1 admit_fds+=("$fd"); fd=$((fd + 1)) done <<< "$target_ws" fi diff --git a/.agents/skills/multi-agent-mux-reset/SKILL.md b/.agents/skills/multi-agent-mux-reset/SKILL.md index 7b74635..da8ddd5 100644 --- a/.agents/skills/multi-agent-mux-reset/SKILL.md +++ b/.agents/skills/multi-agent-mux-reset/SKILL.md @@ -20,13 +20,13 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] | Item | Behaviour | |---|---| -| Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Worktree rows, `stopped`/`terminated` rows, and `orchestrator_uuids` are skipped and listed with their reason. | +| Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Rows with any other status, worktree rows, and `orchestrator_uuids` are skipped and listed with their reason. | | Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required (exits `3` otherwise). `--dry-run` inspects without `--yes` and changes nothing. | -| Prechecks | Refuses before any change when an open job exists (prints the `registry.py ... close --request cancelled` command), when `.mam/loop-guard-active` or `.mam/update-in-progress` is held, when a target row lacks a valid agent or role, or when the caller itself is one of the targets. | -| Lock & concurrency | Takes update admission (`mam_update_admit`), acquires `.mam/loop-guard-active` (`mam_acquire_loop_lock`), then releases update admission so `run_loop.sh`, `update.sh`, and concurrent resets refuse while the reset runs. | +| Prechecks | Refuses before any change when an open job exists (prints the `registry.py ... close --request cancelled` command), when `.mam/loop-guard-active` or `.mam/update-in-progress` is held, when a target row lacks a valid agent or role, when a running `claude`/`grok` target has no recorded conversation id (the monitor had not pinned one; record it, or stop/purge it by hand first), when `.mam/reset-spec.json` is invalid, or when the caller itself is one of the targets. | +| Lock & concurrency | Takes the update admission lock exclusively, acquires `.mam/loop-guard-active` and publishes `.mam/reset-in-progress`, then releases the admission lock and runs the prechecks. `run_loop.sh`, `update.sh` and another reset refuse on the loop marker; `multi-agent-mux-delegate-job submit` refuses on the reset marker. | | Spec file & rerun | Writes `.mam/reset-spec.json` atomically before the first stop/purge and removes it only when all targets reach `done`. Rerunning resumes from the spec file without re-purging already recreated agents (`created` or `done`). | -| Sequence per target | Sequential (never parallel): ① normal `stop_session.sh --reason reset` (captures verified ID onto the row while herdr is live and sets `status: stopped`), ② checks that the row has a recorded conversation ID, ③ `stop_session.sh --reason reset --purge-conversation --yes` (using the row's recorded ID, never a CLI UUID), then `create_session.sh` with the original name, agent, role, `--herdr-session`, and `--herdr-workspace` (onboarding enabled by default). | -| When skipped (exit 4) | If step ② finds no recorded ID or step ③ exits `4` (in-use gate), the target is skipped and the reset continues to the next agent: the skipped agent is stopped but resumable (its row keeps its id) and its conversation files are untouched; rerun once its holders end, or resume it. | +| Sequence per target | Sequential (never parallel): ⓪ re-checks that its row still has the spec's agent and herdr server and that no job addressed to it is open, ① normal `stop_session.sh --reason reset` (a post-discovered agent's id is captured onto the row while herdr is live; `status: stopped`), ② checks that the row's id is the one recorded before the stop, ③ `stop_session.sh --reason reset --purge-conversation --yes` (using the row's recorded ID, never a CLI UUID), then `create_session.sh` with the original name, agent, role, `--herdr-session`, and `--herdr-workspace` (onboarding enabled by default). | +| When skipped (exit 4) | The target is skipped and the reset continues to the next agent when step ⓪ finds a changed row (`row changed`) or an open job (`job open`) - nothing was changed for it -, or when step ② finds no id (`no conversation id recorded`) or an id other than the recorded one (`id not verified`), or step ③ exits `4` (`purge refused`, in-use gate) - it is stopped but resumable (its row keeps its id) and its conversation files are untouched. Rerun once the cause is gone, or resume it. | | Exit | Meaning | |---|---| @@ -34,9 +34,9 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] | 1 | Refused by precheck (nothing changed), corrupt/mismatched `.mam/reset-spec.json`, or a `stop`/`create` step failed (`.mam/reset-spec.json` is kept so rerunning resumes) | | 2 | Invalid arguments | | 3 | Not a terminal and `-y`/`--yes` was not passed; nothing was changed | -| 4 | Completed the pass, but one or more agents were skipped (purge refused with stop exit 4, or no conversation ID recorded); `.mam/reset-spec.json` is kept for rerun | +| 4 | Completed the pass, but one or more agents were skipped (see *When skipped*); `.mam/reset-spec.json` is kept for rerun | | 5 | No agent was skipped or failed, but one or more newly created post-discovered agents do not have their new conversation ID recorded yet; `.mam/reset-spec.json` is kept so rerunning re-checks without re-purging | -| other | Under `set -e`, an unexpected sub-step failure passes its nonzero exit code through | +| other | 130 / 143 / 129 when interrupted by SIGINT / SIGTERM / SIGHUP (the in-flight stop/create ends first, nothing further runs, `.mam/reset-spec.json` is kept); otherwise, under `set -e`, an unexpected internal step failure passes its nonzero exit code through | ## Safety Notes @@ -44,5 +44,5 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] - **No external process termination**: processes holding conversation files open (reported via `lsof` in `in_use_hints`) are never signalled by MAM; end them yourself when safe, then rerun. - **Shared stores preserved**: Grok's shared `memory-v2.db` is left intact during purge; only session rows and per-session files are removed. - **Orchestrator hook during reset**: while `.mam/loop-guard-active` is held by the reset, the AGY orchestrator's direct file-edit hook (`loop_delegation_guard.sh`) blocks direct edits until the reset finishes. -- **Post-precheck job submissions**: `multi-agent-mux-delegate-job submit` does not check `.mam/loop-guard-active`; do not submit new jobs to resident agents while a reset is running. +- **Job submissions during a reset**: `multi-agent-mux-delegate-job submit` refuses with `a reset of is in progress` until the reset ends; submit again afterwards. A job registered for a target by other means after the precheck makes that target `job open` (skipped, not cancelled). - **Same-type post-discovered ordering**: when resetting multiple post-discovered agents of the same type, the second `create_session.sh` waits up to `MAM_ID_ORDER_WAIT_S` for the first agent's ID to be recorded; if it times out, rerun the reset once the ID is recorded. diff --git a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh index 5963263..7978b7a 100755 --- a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh +++ b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh @@ -42,6 +42,7 @@ fi MAM="$WS/.mam" MARKER="${MAM_LOOP_MARKER:-$MAM/loop-guard-active}" +RESET_MARKER="$MAM/reset-in-progress" SPEC_FILE="$MAM/reset-spec.json" # shellcheck disable=SC1091 @@ -82,26 +83,38 @@ fi source "$WS/.agents/skills/lib.sh" PY="$(_delegate_py_bin)" -mam_update_admit "$WS" || exit 1 +_refuse_locked() { + exec 7>&- + echo "ERROR: reset refused; nothing was changed:" >&2 + echo " - $1" >&2 + exit 1 +} +# The update admission lock is taken exclusively, as update.sh does, while the +# reset publishes its markers: a delegate submit admitted earlier has written +# its job record (the precheck below sees it), and one admitted later sees +# .mam/reset-in-progress and refuses. +mkdir -p "$MAM" +exec 7>>"$MAM/update-admission.lock" +case "$("$PY" -m lib_py.pane_lock acquire --fd 7 --wait-env MAM_UPDATE_ADMISSION_WAIT_S)" in + locked) ;; + timeout) _refuse_locked "a create, resume, loop, delegate submit, stop or orc-onboard kept the admission lock past MAM_UPDATE_ADMISSION_WAIT_S; try again" ;; + *) _refuse_locked "the admission lock $MAM/update-admission.lock could not be decided (lib_py.pane_lock)" ;; +esac +if [ -f "$MAM/update-in-progress" ]; then + upd_st="$(_mam_lock_holder_state "$MAM/update-in-progress")" + if [ "$upd_st" != "stale" ]; then + _refuse_locked "an update of $WS is in progress (pid $(_mam_lock_field "$MAM/update-in-progress" pid), holder $upd_st); wait for it to finish" + fi +fi _acquire_rc=0 mam_acquire_loop_lock "$MARKER" || _acquire_rc=$? case "$_acquire_rc" in 0) ;; - 3) - mam_update_admit_release - _holder_pid="$(_mam_lock_field "$MARKER" "pid" 2>/dev/null || echo "unknown")" - echo "ERROR: reset refused; nothing was changed:" >&2 - echo " - a review loop or another reset is running (pid $_holder_pid); wait for it to end" >&2 - exit 1 - ;; - *) - mam_update_admit_release - echo "ERROR: reset refused; nothing was changed:" >&2 - echo " - cannot determine whether a review loop or another reset is running (ps unavailable?)" >&2 - exit 1 - ;; + 3) _refuse_locked "a review loop or another reset is running (pid $(_mam_lock_field "$MARKER" "pid" 2>/dev/null || echo "unknown")); wait for it to end" ;; + *) _refuse_locked "cannot determine whether a review loop or another reset is running (ps unavailable?)" ;; esac _release_reset_lock() { + mam_release_loop_lock "$RESET_MARKER" || true mam_release_loop_lock "$MARKER" || true } _on_reset_signal() { @@ -113,7 +126,10 @@ trap _release_reset_lock EXIT trap '_on_reset_signal 130' INT trap '_on_reset_signal 143' TERM trap '_on_reset_signal 129' HUP -mam_update_admit_release +if ! mam_acquire_loop_lock "$RESET_MARKER"; then + _refuse_locked "cannot publish $RESET_MARKER" +fi +exec 7>&- STATE_JSON="" if ! STATE_JSON="$(load_state_json strict)"; then @@ -189,6 +205,17 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do continue ;; pending) + if [ "$R_ROW_AGENT" != "$R_AGENT" ] || [ "$R_ROW_SERVER" != "$R_SERVER" ]; then + echo "skipped $R_NAME: its row ($R_ROW_AGENT on herdr server $R_ROW_SERVER) no longer matches $SPEC_FILE ($R_AGENT on $R_SERVER); nothing was changed for it; check the row, then rerun the reset" + SKIPPED_ARGS+=(--skipped "$R_NAME:changed") + continue + fi + jobs_out="" + jobs_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset open-jobs --workspace "$WS" --session "$R_NAME")" || { + echo "skipped $R_NAME: a job was registered for it after the precheck (${jobs_out//$'\n'/; }); nothing was changed for it; rerun the reset once the job has ended" + SKIPPED_ARGS+=(--skipped "$R_NAME:open_job") + continue + } STOP_CMD=(bash "$SKILL_DIR/multi-agent-mux-stop/scripts/stop_session.sh" --session "$R_NAME" --agent "$R_AGENT") if [ -n "$R_SERVER" ]; then STOP_CMD+=(--herdr-session "$R_SERVER") @@ -223,6 +250,11 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do SKIPPED_ARGS+=(--skipped "$R_NAME:no_id") continue fi + if [ "$R_ROW_ID" != "$R_OLD_ID" ]; then + echo "skipped $R_NAME: the stop left id $R_ROW_ID on its row, not the one recorded before the reset (${R_OLD_ID:-none}); it may be another conversation's, so nothing is purged; check it, purge it by hand if it is this agent's, and rerun the reset" + SKIPPED_ARGS+=(--skipped "$R_NAME:id_unverified") + continue + fi echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes" purge_rc=0 diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py index 8befe59..dc9b2d2 100644 --- a/tests/test_issue78_reset.py +++ b/tests/test_issue78_reset.py @@ -458,6 +458,27 @@ def test_select_targets_and_plan_cli(tmp_path: Path): assert "bad-role-claude" in res_pre.stdout assert "mystery-row" in res_pre.stdout + # Review B2: claude/grok get their id at spawn, so a running row without one + # is refused (stop would fill it with a folder-wide guess); a post-discovered + # agent without one stays a target (its id is captured by its own pane). + no_id_state = { + "herdr_sessions": [ + _row("reviewer-claude-noid", "claude", ws, own_id=None), + _row("reviewer-grok-noid", "grok", ws, own_id=None), + _row("reviewer-agy-noid", "agy", ws, own_id=None), + ] + } + res_noid = _run_reset_py(["plan", "--workspace", str(ws)], state=no_id_state) + assert res_noid.returncode == 1 + noid_plan = json.loads(res_noid.stdout) + assert [t["name"] for t in noid_plan["targets"]] == ["reviewer-agy-noid"] + assert len(noid_plan["errors"]) == 2 + for name, agent in (("reviewer-claude-noid", "claude"), ("reviewer-grok-noid", "grok")): + assert any( + "running session %s (%s) has no recorded conversation id" % (name, agent) in e + for e in noid_plan["errors"] + ) + def test_spec_file_write_read_and_validation(tmp_path: Path): """D5: write-spec atomically writes .mam/reset-spec.json; corrupt, wrong-version, @@ -526,6 +547,34 @@ def test_spec_file_write_read_and_validation(tmp_path: Path): with pytest.raises(SpecError, match="unknown agent"): read_spec(str(ws)) + # Review B2 (codex): every stored field is validated; a lost herdr_server is + # never replaced by "default", and a name listed twice is refused + good = dict(targets[0]) + for mutate in ( + lambda e: e.pop("herdr_server"), + lambda e: e.update(herdr_server=""), + lambda e: e.update(herdr_server=None), + lambda e: e.pop("herdr_workspace"), + lambda e: e.update(herdr_workspace=None), + lambda e: e.pop("old_id"), + lambda e: e.update(old_id=""), + lambda e: e.update(old_id=7), + lambda e: e.pop("old_created_at"), + lambda e: e.update(old_created_at=None), + ): + bad = dict(good) + mutate(bad) + spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [bad]}), encoding="utf-8") + with pytest.raises(SpecError, match="incomplete agent entry"): + read_spec(str(ws)) + spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [good, dict(good)]}), encoding="utf-8") + with pytest.raises(SpecError, match="lists reviewer-claude-01 twice"): + read_spec(str(ws)) + # Control: legal empty values (label, created_at) and an explicit null old_id pass + ok = dict(good, herdr_workspace="", old_created_at="", old_id=None) + spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [ok]}), encoding="utf-8") + assert read_spec(str(ws))["agents"] == [ok] + # Corrupt JSON spath.write_text("{corrupt", encoding="utf-8") with pytest.raises(SpecError, match="check it, then remove it by hand"): @@ -1335,7 +1384,7 @@ with open(spy_log, "a", encoding="utf-8") as lf: if verb == "kill-session" and not os.path.exists(entered): with open(entered, "w", encoding="utf-8") as fh: fh.write("1") - deadline = time.time() + 15.0 + deadline = time.time() + 60.0 while not os.path.exists(release) and time.time() < deadline: time.sleep(0.05) @@ -1366,7 +1415,7 @@ os.execv(real, [real, *raw_args]) text=True, ) try: - deadline = time.time() + 15.0 + deadline = time.time() + 60.0 while not entered_gate.exists() and time.time() < deadline: assert proc.poll() is None, "reset.sh exited before reaching kill-session" time.sleep(0.05) @@ -1414,3 +1463,324 @@ os.execv(real, [real, *raw_args]) assert not any( l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines() ) + + +# --------------------------------------------------------------------------- Review round 1 (PR #82) + +def _write_spec_file(sandbox: Path, agents: List[dict]) -> Path: + spec = sandbox / ".mam" / "reset-spec.json" + spec.write_text( + json.dumps({"version": 1, "workspace": os.path.realpath(str(sandbox)), "written_at": "x", "agents": agents}), + encoding="utf-8", + ) + return spec + + +def _spec_entry(name: str, agent: str, old_id: Optional[str], server: str = "default") -> dict: + return { + "name": name, + "agent": agent, + "role": "reviewer", + "herdr_server": server, + "herdr_workspace": "ws", + "old_id": old_id, + "old_created_at": "2026-10-01T00:00:00Z", + } + + +def _result_of(stdout: str, name: str) -> str: + line = [l for l in stdout.splitlines() if l.startswith(name + " ")][-1] + return line.split(None, 6)[6] + + +def _live_marker(path: Path, holder: subprocess.Popen) -> None: + ls = " ".join( + subprocess.run(["ps", "-p", str(holder.pid), "-o", "lstart="], capture_output=True, text=True).stdout.split() + ) + path.write_text("pid=%d\nlstart=%s\nstarted=x\n" % (holder.pid, ls), encoding="utf-8") + + +def test_record_id_never_records_a_pre_assigned_guess(tmp_path: Path): + """Review B2 (claude): a claude/grok entry with old_id=None never takes the id + a normal stop guessed onto its row; a post-discovered one still does.""" + from lib_py.reset import read_spec, write_spec + + ws = tmp_path / "ws" + ws.mkdir() + write_spec(str(ws), [_spec_entry("c-noid", "claude", None), _spec_entry("a-noid", "agy", None)]) + state = { + "herdr_sessions": [ + _row("c-noid", "claude", ws, status="stopped", own_id=UA), + _row("a-noid", "agy", ws, status="stopped", own_id=UB), + ] + } + for name in ("c-noid", "a-noid"): + res = _run_reset_py(["record-id", "--workspace", str(ws), "--session", name], state=state) + assert res.returncode == 0, res.stderr + old = {e["name"]: e["old_id"] for e in read_spec(str(ws))["agents"]} + assert old == {"c-noid": None, "a-noid": UB} + + +def test_claude_without_recorded_id_is_refused_and_unregistered_conversation_kept( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """Review B2 (claude) (c): a running claude row with no id and an unregistered + claude conversation in the same folder: the reset refuses before any change, + so the conversation that is not provably this pane's is never purged.""" + name = "reviewer-claude-01" + tf = _claude_transcript(mam_sandbox, mam_sandbox, UC) + _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude") + _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=None)]) + spy_log = _install_herdr_spy(mam_sandbox) + yaml_before = (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() + + res = _run_reset(mam_sandbox, "--yes") + assert res.returncode == 1, res.stdout + res.stderr + assert "running session %s (claude) has no recorded conversation id" % name in res.stderr + assert "nothing was changed" in res.stderr + assert tf.is_file() + assert (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() == yaml_before + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + assert not any(c["verb"] in ("kill-session", "pane close", "agent start") for c in _read_spy_log(spy_log)) + + res_dry = _run_reset(mam_sandbox, "--dry-run") + assert res_dry.returncode == 1 + assert "has no recorded conversation id" in res_dry.stdout + + +def test_rerun_skips_a_target_whose_row_id_is_not_the_recorded_one( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """Review B2 (claude) (b): after the normal stop the row's id must equal the + spec's old_id. A: the row's id changed after the spec was written (UC, spec + UA). B: an older spec holds a claude entry with old_id null, and the stop + guesses an unregistered conversation onto its row. Neither is purged.""" + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + files = [_claude_transcript(mam_sandbox, mam_sandbox, u) for u in (UA, UC, UD)] + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [_row(name_a, "claude", mam_sandbox, own_id=UC), _row(name_b, "claude", mam_sandbox, own_id=None)], + ) + _write_spec_file(mam_sandbox, [_spec_entry(name_a, "claude", UA), _spec_entry(name_b, "claude", None)]) + + res = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res.returncode == 4, res.stdout + res.stderr + assert _result_of(res.stdout, name_a) == "skipped: id not verified" + assert _result_of(res.stdout, name_b) == "skipped: id not verified" + assert not any(l.startswith("reset [") and "--purge-conversation" in l for l in res.stdout.splitlines()) + for f in files: + assert f.is_file(), f + rows = _read_rows(mam_sandbox) + assert rows[name_a]["claude_session_id_own"] == UC + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + + +def test_spec_server_is_validated_and_a_changed_row_is_not_stopped( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """Review B2 (codex): a spec entry that lost herdr_server is refused before any + stop (never `--herdr-session default`); a valid spec whose server no longer + matches the row skips that target untouched.""" + name = "reviewer-claude-01" + tf = _claude_transcript(mam_sandbox, mam_sandbox, UA) + _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude") + _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)]) + spy_log = _install_herdr_spy(mam_sandbox) + + entry = _spec_entry(name, "claude", UA) + entry.pop("herdr_server") + spec = _write_spec_file(mam_sandbox, [entry]) + spec_text = spec.read_text(encoding="utf-8") + res = _run_reset(mam_sandbox, "--yes") + assert res.returncode == 1, res.stdout + res.stderr + assert "incomplete agent entry" in res.stderr + assert "--herdr-session" not in res.stdout + assert spec.read_text(encoding="utf-8") == spec_text + assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log)) + + _write_spec_file(mam_sandbox, [_spec_entry(name, "claude", UA, server="isolated-server")]) + res_changed = _run_reset(mam_sandbox, "--yes") + assert res_changed.returncode == 4, res_changed.stdout + res_changed.stderr + assert _result_of(res_changed.stdout, name) == "skipped: row changed" + assert "reset [%s]:" % name not in res_changed.stdout + assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log)) + assert _read_rows(mam_sandbox)[name]["status"] == "running" + assert tf.is_file() + + +def test_job_after_precheck_skips_its_target_and_submit_is_refused_during_reset( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """Review B1 (codex), "reset first": while the reset runs (A's stop), a + delegate submit to B is refused by .mam/reset-in-progress and registers + nothing; a record for B written by other means is seen by B's re-check, so B + is skipped with its job left open and its conversation kept.""" + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA) + tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [_row(name_a, "claude", mam_sandbox, own_id=UA), _row(name_b, "claude", mam_sandbox, own_id=UB)], + ) + + bin_dir = mam_sandbox / "bin" + herdr_bin = bin_dir / "herdr" + herdr_real = bin_dir / "herdr.real" + herdr_bin.rename(herdr_real) + probe = mam_sandbox / "submit_probe.json" + jobs_dir = mam_sandbox / ".mam" / "jobs" + job_file = jobs_dir / "job78late.json" + submit = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job" + herdr_bin.write_text( + f"""#!{sys.executable} +import json, os, subprocess, sys + +real = {repr(str(herdr_real))} +probe = {repr(str(probe))} +args = list(sys.argv[1:]) +while args and args[0] in ("-L", "--server", "-s", "--session"): + args = args[2:] +if args[:2] == ["pane", "close"] or args[:1] == ["kill-session"]: + if not os.path.exists(probe): + marker = {repr(str(mam_sandbox / ".mam" / "reset-in-progress"))} + res = subprocess.run( + ["bash", {repr(str(submit))}, "submit", "--agent", "claude-code", + "--agent-session", "herdr:{name_b}", "--prompt", "hi"], + cwd={repr(str(mam_sandbox))}, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=120, + ) + jobs = sorted(os.listdir({repr(str(jobs_dir))})) if os.path.isdir({repr(str(jobs_dir))}) else [] + with open(probe, "w", encoding="utf-8") as fh: + json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr, + "marker": os.path.isfile(marker), "jobs": jobs}}, fh) + os.makedirs({repr(str(jobs_dir))}, exist_ok=True) + with open({repr(str(job_file))}, "w", encoding="utf-8") as fh: + json.dump({{"job_id": "job78late", "agent_session": "herdr:{name_b}", "status": "pending"}}, fh) +os.execv(real, [real, *sys.argv[1:]]) +""", + encoding="utf-8", + ) + herdr_bin.chmod(0o755) + + res = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res.returncode == 4, res.stdout + res.stderr + + p = json.loads(probe.read_text(encoding="utf-8")) + assert p["marker"] is True + assert p["rc"] != 0, p + assert "a reset of %s is in progress" % mam_sandbox in p["stderr"], p + assert "registered job" not in p["stdout"] + assert p["jobs"] == [] + + assert _result_of(res.stdout, name_a) == "done" + assert _result_of(res.stdout, name_b) == "skipped: job open" + assert "job78late" in res.stdout + assert "reset [%s]:" % name_b not in res.stdout + assert not tf_a.exists() + assert tf_b.is_file() + assert json.loads(job_file.read_text(encoding="utf-8"))["status"] == "pending" + rows = _read_rows(mam_sandbox) + assert rows[name_b]["status"] == "running" and rows[name_b]["claude_session_id_own"] == UB + assert not (mam_sandbox / ".mam" / "reset-in-progress").exists() + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + + +def _descendant_args(root_pid: int) -> List[str]: + res = subprocess.run(["ps", "-eo", "pid=,ppid=,args="], capture_output=True, text=True, timeout=5) + kids: Dict[int, List[int]] = {} + args: Dict[int, str] = {} + for line in res.stdout.splitlines(): + parts = line.split(None, 2) + if len(parts) == 3 and parts[0].isdigit() and parts[1].isdigit(): + kids.setdefault(int(parts[1]), []).append(int(parts[0])) + args[int(parts[0])] = parts[2] + out, todo = [], [root_pid] + while todo: + for c in kids.get(todo.pop(), []): + out.append(args.get(c, "")) + todo.append(c) + return out + + +def test_reset_waits_for_an_admitted_submit_and_sees_its_job(mam_sandbox: Path, mock_herdr: Path, mock_agents): + """Review B1 (codex), "submit first": a submit that holds the admission lock + shared until its record is written makes the reset wait before it publishes + its markers; the precheck then sees the job and refuses, nothing changed.""" + import fcntl + + name = "reviewer-claude-01" + tf = _claude_transcript(mam_sandbox, mam_sandbox, UA) + _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude") + _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)]) + spy_log = _install_herdr_spy(mam_sandbox) + + lock = open(mam_sandbox / ".mam" / "update-admission.lock", "a") + fcntl.flock(lock, fcntl.LOCK_SH) + script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh" + proc = subprocess.Popen( + ["bash", str(script), "--yes"], + cwd=str(mam_sandbox), + env=dict(os.environ, MAM_UPDATE_ADMISSION_WAIT_S="60", PYTHONDONTWRITEBYTECODE="1"), + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.time() + 60 + while time.time() < deadline: + assert proc.poll() is None, proc.communicate() + if any("lib_py.pane_lock acquire --fd 7" in a for a in _descendant_args(proc.pid)): + break + time.sleep(0.1) + else: + pytest.fail("reset.sh never waited for the admission lock") + assert not (mam_sandbox / ".mam" / "reset-in-progress").exists() + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + jobs_dir = mam_sandbox / ".mam" / "jobs" + jobs_dir.mkdir(parents=True, exist_ok=True) + (jobs_dir / "job78first.json").write_text( + json.dumps({"job_id": "job78first", "agent_session": "herdr:%s" % name, "status": "pending"}), + encoding="utf-8", + ) + finally: + fcntl.flock(lock, fcntl.LOCK_UN) + lock.close() + stdout, stderr = proc.communicate(timeout=120) + assert proc.returncode == 1, stdout + stderr + assert "job78first" in stderr and "nothing was changed" in stderr + assert tf.is_file() + assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log)) + assert not (mam_sandbox / ".mam" / "reset-in-progress").exists() + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + + +def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path): + lib = mam_sandbox / ".agents" / "skills" / "lib.sh" + marker = mam_sandbox / ".mam" / "reset-in-progress" + + def admit(): + return subprocess.run( + ["bash", "-c", 'source "$1" && mam_reset_admit "$2"', "_", str(lib), str(mam_sandbox)], + cwd=str(mam_sandbox), capture_output=True, text=True, timeout=60, + ) + + assert admit().returncode == 0 + holder = subprocess.Popen(["sleep", "120"]) + try: + _live_marker(marker, holder) + res = admit() + assert res.returncode == 1 + assert "a reset of %s is in progress" % mam_sandbox in res.stderr + finally: + holder.kill() + holder.wait() + res_stale = admit() + assert res_stale.returncode == 0, res_stale.stderr