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 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-30 00:15:31 +09:00
1 parent f90a705316
commit 5c3c2eb861
6 files changed
+529 -62

No files matched your search

+17
View File
@@ -3348,6 +3348,23 @@ mam_update_admit_release() {
eval "exec ${1:-7}>&-" 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 <cwd>: what create and resume take for an agent they # mam_update_admit_at <cwd>: what create and resume take for an agent they
# start in <cwd> - mam_update_admit for the control root (fd 7) and for every # start in <cwd> - mam_update_admit for the control root (fd 7) and for every
# other installed workspace at or above <cwd> (fds 20 and up), whose update # other installed workspace at or above <cwd> (fds 20 and up), whose update
+81 -36
View File
@@ -20,7 +20,7 @@ import sys
import tempfile import tempfile
from typing import Dict, List, Mapping, Optional, Tuple 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.agents.registry import agent_of_row, get_adapter, own_key
from lib_py.update import open_jobs, row_server 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) 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: def _state(workspace: str = "") -> dict:
raw = os.environ.get("MAM_STATE_JSON") raw = os.environ.get("MAM_STATE_JSON")
if raw is not None: if raw is not None:
return json.loads(raw or "{}") or {} 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 ( yaml_path = os.environ.get("AGENT_SESSIONS_YAML") or (
os.path.join(_real(workspace), ".mam", "agent-sessions.yaml") if workspace else "" 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) key = own_key(agent)
raw_id = row.get(key) raw_id = row.get(key)
old_id = str(raw_id).strip() if isinstance(raw_id, str) and raw_id.strip() else None 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({ targets.append({
"name": name, "name": name,
"agent": agent, "agent": agent,
@@ -219,14 +239,28 @@ def read_spec(workspace: str) -> dict:
agents = doc.get("agents") agents = doc.get("agents")
if not isinstance(agents, list): if not isinstance(agents, list):
raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path) raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path)
names = set()
for item in agents: for item in agents:
if not isinstance(item, dict): if not isinstance(item, dict):
raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path) raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path)
name = item.get("name") name = item.get("name")
agent = item.get("agent") agent = item.get("agent")
role = item.get("role") 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) 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) adapter = get_adapter(agent)
if adapter is None or adapter.name != agent or not own_key(agent): if adapter is None or adapter.name != agent or not own_key(agent):
raise SpecError( raise SpecError(
@@ -245,12 +279,7 @@ def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[st
""" """
ws_real = _real(workspace) ws_real = _real(workspace)
spec = read_spec(ws_real) spec = read_spec(ws_real)
rows_by_name = { row = _rows_by_name(state).get(session_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)
if row is None: if row is None:
return None return None
changed = False 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 old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
if old_id: if old_id:
return 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 "") cur_created_at = str(row.get("herdr_session_created_at") or "")
old_created_at = str(entry.get("old_created_at") or "") old_created_at = str(entry.get("old_created_at") or "")
if cur_created_at != old_created_at: if cur_created_at != old_created_at:
@@ -433,15 +466,22 @@ def precheck(
) )
) )
rows_by_name = { blockers.extend(_caller_blockers(targets, _rows_by_name(state), env, ancestors=ancestors))
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))
return blockers 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 # --------------------------------------------------------------------------- D1: entries shell lines
def format_entry_line(entry: dict, row: Optional[dict]) -> str: 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_NAME", str(entry.get("name") or "")),
("R_AGENT", str(entry.get("agent") or "")), ("R_AGENT", str(entry.get("agent") or "")),
("R_ROLE", str(entry.get("role") 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_LABEL", str(entry.get("herdr_workspace") or "")),
("R_OLD_ID", str(entry.get("old_id") or "")), ("R_OLD_ID", str(entry.get("old_id") or "")),
("R_ROW_ID", row_id), ("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) return " ".join("%s=%s" % (k, shlex.quote(v)) for k, v in pairs)
# --------------------------------------------------------------------------- D8: result table # --------------------------------------------------------------------------- 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]: def _parse_skipped(items: Optional[List[str]]) -> Dict[str, str]:
out: Dict[str, str] = {} out: Dict[str, str] = {}
for raw in items or []: for raw in items or []:
@@ -497,11 +548,7 @@ def build_table(
failed_spec: Optional[str] = None, failed_spec: Optional[str] = None,
) -> Tuple[str, int]: ) -> Tuple[str, int]:
"""Format the D8 result table and compute the D9 exit code (0, 1, 4, or 5).""" """Format the D8 result table and compute the D9 exit code (0, 1, 4, or 5)."""
rows_by_name = { rows_by_name = _rows_by_name(state)
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
skipped_map = _parse_skipped(skipped_items) skipped_map = _parse_skipped(skipped_items)
failed = _parse_failed(failed_spec) failed = _parse_failed(failed_spec)
failed_name = failed[0] if failed else None failed_name = failed[0] if failed else None
@@ -539,11 +586,7 @@ def build_table(
result_col = "not reached" result_col = "not reached"
has_skip_or_incomplete = True has_skip_or_incomplete = True
elif name in skipped_map: elif name in skipped_map:
kind = skipped_map[name] result_col = SKIP_RESULTS.get(skipped_map[name], SKIP_RESULTS["purge_refused"])
if kind == "no_id":
result_col = "skipped: no conversation id recorded"
else:
result_col = "skipped: purge refused"
has_skip_or_incomplete = True has_skip_or_incomplete = True
elif r_state == "done": elif r_state == "done":
result_col = "done" result_col = "done"
@@ -598,11 +641,7 @@ def format_dry_run(
refusals: List[str] = list(extra_refusals or []) refusals: List[str] = list(extra_refusals or [])
refusals.extend(precheck(ws_real, state, env)) refusals.extend(precheck(ws_real, state, env))
rows_by_name = { rows_by_name = _rows_by_name(state)
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
sel = select_targets(ws_real, state) sel = select_targets(ws_real, state)
spath = spec_path(ws_real) spath = spec_path(ws_real)
action_map = { action_map = {
@@ -628,7 +667,7 @@ def format_dry_run(
e["name"], e["name"],
e["agent"], e["agent"],
e["role"], e["role"],
e.get("herdr_server") or "default", e["herdr_server"],
e.get("herdr_workspace") or "", e.get("herdr_workspace") or "",
e.get("old_id") or "-", e.get("old_id") or "-",
e.get("old_created_at") 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("--session", default="")
s.add_argument("--warn-new", action="store_true") 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 = sub.add_parser("record-id")
s.add_argument("--workspace", required=True) s.add_argument("--workspace", required=True)
s.add_argument("--session", 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)), % (spec_path(ws), ", ".join(extra)),
file=sys.stderr, file=sys.stderr,
) )
rows_by_name = { rows_by_name = _rows_by_name(state)
str(s.get("name")): s
for s in (state.get("herdr_sessions") or [])
if isinstance(s, dict) and s.get("name")
}
for e in spec["agents"]: for e in spec["agents"]:
if a.session and e.get("name") != a.session: if a.session and e.get("name") != a.session:
continue continue
print(format_entry_line(e, rows_by_name.get(e["name"]))) print(format_entry_line(e, rows_by_name.get(e["name"])))
return 0 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": if a.cmd == "record-id":
try: try:
record_old_id(ws, state, a.session) record_old_id(ws, state, a.session)
@@ -306,7 +306,8 @@ cmd_submit() {
# its row's, or with no row the live agent's on the server delivery uses) # 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 # 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 # 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 local admit_fds=() target_ws tws fd=20
if [[ "$DRY_RUN" != "1" ]]; then if [[ "$DRY_RUN" != "1" ]]; then
target_ws="$(MAM_STATE_JSON="$(load_state_json)" PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$PY" -m lib_py.update \ 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)")" \ --herdr "$(_resolve_real_herdr_path || true)")" \
|| { echo "ERROR: cannot resolve the workspace of $AGENT_SESSION; nothing was registered" >&2; exit 1; } || { echo "ERROR: cannot resolve the workspace of $AGENT_SESSION; nothing was registered" >&2; exit 1; }
mam_update_admit "$WORKSPACE_ROOT" || exit 1 mam_update_admit "$WORKSPACE_ROOT" || exit 1
mam_reset_admit "$WORKSPACE_ROOT" || exit 1
while IFS= read -r tws; do while IFS= read -r tws; do
[[ -n "$tws" ]] || continue [[ -n "$tws" ]] || continue
mam_update_admit "$tws" "$fd" || exit 1 mam_update_admit "$tws" "$fd" || exit 1
mam_reset_admit "$tws" || exit 1
admit_fds+=("$fd"); fd=$((fd + 1)) admit_fds+=("$fd"); fd=$((fd + 1))
done <<< "$target_ws" done <<< "$target_ws"
fi fi
@@ -20,13 +20,13 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run]
| Item | Behaviour | | 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. | | 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. | | 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 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. | | 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`). | | 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). | | 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) | 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. | | 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 | | 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) | | 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 | | 2 | Invalid arguments |
| 3 | Not a terminal and `-y`/`--yes` was not passed; nothing was changed | | 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 | | 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 ## 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. - **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. - **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. - **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 <WS> 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. - **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.
@@ -42,6 +42,7 @@ fi
MAM="$WS/.mam" MAM="$WS/.mam"
MARKER="${MAM_LOOP_MARKER:-$MAM/loop-guard-active}" MARKER="${MAM_LOOP_MARKER:-$MAM/loop-guard-active}"
RESET_MARKER="$MAM/reset-in-progress"
SPEC_FILE="$MAM/reset-spec.json" SPEC_FILE="$MAM/reset-spec.json"
# shellcheck disable=SC1091 # shellcheck disable=SC1091
@@ -82,26 +83,38 @@ fi
source "$WS/.agents/skills/lib.sh" source "$WS/.agents/skills/lib.sh"
PY="$(_delegate_py_bin)" 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 _acquire_rc=0
mam_acquire_loop_lock "$MARKER" || _acquire_rc=$? mam_acquire_loop_lock "$MARKER" || _acquire_rc=$?
case "$_acquire_rc" in case "$_acquire_rc" in
0) ;; 0) ;;
3) 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" ;;
mam_update_admit_release *) _refuse_locked "cannot determine whether a review loop or another reset is running (ps unavailable?)" ;;
_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
;;
esac esac
_release_reset_lock() { _release_reset_lock() {
mam_release_loop_lock "$RESET_MARKER" || true
mam_release_loop_lock "$MARKER" || true mam_release_loop_lock "$MARKER" || true
} }
_on_reset_signal() { _on_reset_signal() {
@@ -113,7 +126,10 @@ trap _release_reset_lock EXIT
trap '_on_reset_signal 130' INT trap '_on_reset_signal 130' INT
trap '_on_reset_signal 143' TERM trap '_on_reset_signal 143' TERM
trap '_on_reset_signal 129' HUP 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="" STATE_JSON=""
if ! STATE_JSON="$(load_state_json strict)"; then if ! STATE_JSON="$(load_state_json strict)"; then
@@ -189,6 +205,17 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do
continue continue
;; ;;
pending) 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") STOP_CMD=(bash "$SKILL_DIR/multi-agent-mux-stop/scripts/stop_session.sh" --session "$R_NAME" --agent "$R_AGENT")
if [ -n "$R_SERVER" ]; then if [ -n "$R_SERVER" ]; then
STOP_CMD+=(--herdr-session "$R_SERVER") 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") SKIPPED_ARGS+=(--skipped "$R_NAME:no_id")
continue continue
fi 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" echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes"
purge_rc=0 purge_rc=0
+372 -2
View File
@@ -458,6 +458,27 @@ def test_select_targets_and_plan_cli(tmp_path: Path):
assert "bad-role-claude" in res_pre.stdout assert "bad-role-claude" in res_pre.stdout
assert "mystery-row" 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): def test_spec_file_write_read_and_validation(tmp_path: Path):
"""D5: write-spec atomically writes .mam/reset-spec.json; corrupt, wrong-version, """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"): with pytest.raises(SpecError, match="unknown agent"):
read_spec(str(ws)) 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 # Corrupt JSON
spath.write_text("{corrupt", encoding="utf-8") spath.write_text("{corrupt", encoding="utf-8")
with pytest.raises(SpecError, match="check it, then remove it by hand"): 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): if verb == "kill-session" and not os.path.exists(entered):
with open(entered, "w", encoding="utf-8") as fh: with open(entered, "w", encoding="utf-8") as fh:
fh.write("1") fh.write("1")
deadline = time.time() + 15.0 deadline = time.time() + 60.0
while not os.path.exists(release) and time.time() < deadline: while not os.path.exists(release) and time.time() < deadline:
time.sleep(0.05) time.sleep(0.05)
@@ -1366,7 +1415,7 @@ os.execv(real, [real, *raw_args])
text=True, text=True,
) )
try: try:
deadline = time.time() + 15.0 deadline = time.time() + 60.0
while not entered_gate.exists() and time.time() < deadline: while not entered_gate.exists() and time.time() < deadline:
assert proc.poll() is None, "reset.sh exited before reaching kill-session" assert proc.poll() is None, "reset.sh exited before reaching kill-session"
time.sleep(0.05) time.sleep(0.05)
@@ -1414,3 +1463,324 @@ os.execv(real, [real, *raw_args])
assert not any( assert not any(
l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines() 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