fix(issue-21): review R1-R3 - no env as agent arguments on a reused pane, one jobs dir for a relative MAM_JOBS_DIR, stop keeps the audit-log root

Review 9a1adb46 (planner-reviewer-codex-01):

- R1 [P1] lib.sh new-session: a forwarded binding reached the agent as
  arguments. With env flags the idle-pane branch prefixed `env K=V ...` to
  the command that `herdr agent start ... --kind <agent> -- <args>` passes to
  the agent executable. An idle pane is now reused only when no environment
  has to be set; otherwise the pane is created by pane split / workspace
  create, which take --env. The dead `env` prefix code is removed (its
  send-text fallback would have typed the same line).
- R2 [P2] runner: a relative MAM_JOBS_DIR is resolved once against the
  control root, the way the wrapper (cwd=workspace_root) reads it, and that
  absolute dir is used for delegation, wait, report lookup and cleanup.
- R3 [P2] stop: the row's control_plane.delegate_job_logs_dir is restored
  (exported unless the caller set DELEGATE_JOB_LOGS_DIR), so the
  cancellation history lands in the row's audit-log root.

Tests: T3-1 and T3-3 now assert that the binding arrives as --env on the
pane-creation calls and that no env/K=V token follows `--` in agent start
(fresh and reused pane); T3-4c (audit root on stop) and T2-15 (relative
MAM_JOBS_DIR) are new. The design doc's compatibility table and a create
comment no longer describe the removed idle-pane env path.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017R3Eg7MEpSWnm1oB9ZPP6G

tests/test_issue21_worktree_prereqs.py

@@ -511,6 +511,22 @@ def _env_flags(mock_state_file, since=0) -> list:
     return [c[i + 1] for c in calls for i, t in enumerate(c[:-1]) if t == "--env"]
 
 
+def _agent_args(mock_state_file, since=0) -> list:
+    """The tokens after `--` of every `herdr agent start` call: herdr passes
+    them to the canonical agent executable (--kind), never to a shell."""
+    calls = json.loads(Path(mock_state_file).read_text()).get("calls", [])[since:]
+    starts = [c for c in calls if any(c[i:i + 2] == ["agent", "start"] for i in range(len(c) - 1))]
+    return [c[c.index("--") + 1:] if "--" in c else [] for c in starts]
+
+
+def _assert_no_env_in_agent_args(mock_state_file, since=0) -> None:
+    arg_lists = _agent_args(mock_state_file, since)
+    assert arg_lists, "no agent start was recorded"
+    for args in arg_lists:
+        assert "env" not in args, args
+        assert not [a for a in args if a.startswith(tuple(f"{k}=" for k in CP_ENV))], args
+
+
 def _ncalls(mock_state_file) -> int:
     return len(json.loads(Path(mock_state_file).read_text()).get("calls", []))
 
@@ -563,6 +579,7 @@ def test_t3_1_create_persists_and_forwards_a_foreign_binding(mam_sandbox, mock_h
         assert f"{k}={v}" in flags, flags
     assert "WORKSPACE_ROOT=" not in row["pane"]["cmd_full"]
     assert "WORKSPACE_ROOT=" not in row["start_command"]
+    _assert_no_env_in_agent_args(mock_herdr)
 
 
 def test_t3_2_default_binding_is_recorded_not_forwarded(mam_sandbox, mock_herdr, mock_agents):
@@ -602,14 +619,14 @@ def test_t3_3_4_stop_keeps_the_binding_and_resume_reforwards_it(mam_sandbox, moc
                           "--workspace", str(wt), "--agent", "claude", "--session", name],
                          capture_output=True, text=True, env=env, timeout=180)
     assert res.returncode == 0, res.stdout + res.stderr
-    # A fresh pane gets `--env K=V` flags; a reused idle pane (the stopped
-    # session's) gets the same K=V tokens as `env K=V ... <cmd>` (lib.sh D9).
-    state = json.loads(Path(mock_herdr).read_text())
-    calls = state.get("calls", [])[since:]
-    args = [a for c in calls for a in c if isinstance(a, str)] + _sent_texts(state)
-    forwarded = {tok for a in args for tok in a.split()}
+    # The stopped session left an idle pane behind. A binding that must be
+    # forwarded is never put in front of `agent start --` (herdr passes those
+    # tokens to the agent as arguments, review R1): resume takes a new pane,
+    # whose creation call carries the binding as --env.
+    flags = _env_flags(mock_herdr, since)
     for k, v in zip(CP_ENV, _binding(tmp).values()):
-        assert f"{k}={v}" in forwarded, (res.stderr[-1500:], json.dumps(calls)[:1500])
+        assert f"{k}={v}" in flags, (res.stderr[-1500:], flags)
+    _assert_no_env_in_agent_args(mock_herdr, since)
     (row,) = _rows(tmp)
     assert row["control_plane"] == _binding(tmp)
 
@@ -690,3 +707,70 @@ d['herdr_sessions'] = [{{
     assert res.returncode == 0, res.stdout + res.stderr
     (row,) = _rows(tmp)
     assert row["control_plane"] == _binding(tmp)
+
+
+def test_t3_4c_stop_keeps_the_audit_trail_in_the_rows_log_root(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
+    """Review R3: with no MAM_JOBS_DIR / DELEGATE_JOB_LOGS_DIR in the caller,
+    stop cancels the job in the row's registry and writes its history to the
+    row's audit-log root - not to $WORKSPACE_ROOT/.mam/delegate_job_logs."""
+    from lib_py.delegation import registry
+    tmp = mam_sandbox
+    jobs, logs = tmp / "custom-jobs", tmp / "custom-logs"
+    # the job was registered under that binding (its `registered` audit entry lives there)
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(logs))
+    jid = registry.register_job("t3-4c", registry_dir=str(jobs), agent_session="herdr:logs-creator-claude")
+    binding = {**_binding(tmp), "mam_jobs_dir": str(jobs), "delegate_job_logs_dir": str(logs)}
+    _mutate(tmp, f"""
+d['herdr_sessions'] = [{{
+    'name': 'logs-creator-claude', 'status': 'running', 'role': 'creator',
+    'delegate_job_id': {jid!r}, 'pane': {{'cwd': {str(tmp)!r}, 'cmd': 'claude'}},
+    'control_plane': {binding!r},
+}}]
+""")
+    env = {k: v for k, v in os.environ.items() if k not in ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")}
+    res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-stop", "scripts", "stop_session.sh"),
+                          "--session", "logs-creator-claude", "--agent", "claude"],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert json.loads((jobs / f"{jid}.json").read_text())["status"] == "cancelled"
+    assert not (tmp / ".mam" / "delegate_job_logs" / jid).exists(), "history misrouted to the default log root"
+    audit = [json.loads(l) for l in (logs / jid / "events.ndjson").read_text().splitlines() if l.strip()]
+    assert {"progress", "cancelled"} <= {e.get("source_event") for e in audit}, audit
+    assert any(e.get("event") == "status_changed" for e in audit), audit
+
+
+def test_t2_15_relative_mam_jobs_dir_means_the_same_dir_to_runner_and_wrapper(tmp_path, monkeypatch):
+    """Review R2: the wrapper runs with cwd=workspace_root and absolutises a
+    relative registry there; the runner must look in that same directory."""
+    from lib_py.loop import runner as lr
+    ctl, caller = tmp_path / "control", tmp_path / "caller"
+    ctl.mkdir()
+    caller.mkdir()
+    monkeypatch.chdir(caller)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(ctl))
+    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
+    monkeypatch.setenv("MAM_JOBS_DIR", ".mam/jobs")
+    r = _runner()
+    jobs = Path(os.path.realpath(ctl)) / ".mam" / "jobs"
+    assert r.jobs_dir == jobs and r.jobs_dir.is_absolute()
+    report = jobs / "j21" / "reviewer-reports" / "report-final.md"
+    report.parent.mkdir(parents=True)
+    report.write_text("[VERDICT: PASS]\n")
+    (jobs / "j21.json").write_text(json.dumps({"status": "completed"}))
+    assert r.find_report("j21") == report
+    assert r.wait_for_job("j21", max_wait=2) is True
+    seen = []
+
+    def fake_run(cmd, **kw):
+        seen.append(cmd)
+
+        class P:
+            returncode = 0
+            stdout = "registered job: j22\n"
+            stderr = ""
+        return P()
+
+    monkeypatch.setattr(lr.subprocess, "run", fake_run)
+    r.delegate_job("creator-claude-01", "claude", "Worker", "go")
+    assert seen[0][seen[0].index("--registry-dir") + 1] == str(jobs), "the wrapper is told the same absolute dir"
+    assert not (caller / ".mam").exists()