GodopuandClaude Opus 5.5
a7bf07bb1c
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
247 lines
10 KiB
Python
247 lines
10 KiB
Python
"""Issue #68: worktree_prepare.sh onboards by default (--no-onboard skips), and
|
|
create_session.sh --no-onboard warns for post-discovered-id agents (adapter
|
|
id_lifecycle == POST_DISCOVERED).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
if str(SKILLS) not in sys.path:
|
|
sys.path.insert(0, str(SKILLS))
|
|
|
|
from lib_py.agents.base import POST_DISCOVERED # noqa: E402
|
|
from lib_py.agents.registry import get_adapter # noqa: E402
|
|
from test_issue21_phase2_loop_a import _prepare, _rows, _wt, ctl # noqa: E402,F401
|
|
from test_issue63_same_cwd_ids import _mock_herdr_agent, _read_rows, _row, _write_rows # noqa: E402
|
|
|
|
ALL_AGENTS = ("claude", "agy", "hermes", "grok", "opencode", "codex")
|
|
EXPECTED_POST_DISCOVERED = {
|
|
"agy": 1,
|
|
"codex": 1,
|
|
"hermes": 1,
|
|
"opencode": 1,
|
|
"claude": 0,
|
|
"grok": 0,
|
|
}
|
|
UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I)
|
|
|
|
|
|
def _norm_uuid(text: str) -> str:
|
|
return UUID_RE.sub("<UUID>", text)
|
|
|
|
|
|
def _warn_lines(stderr: str) -> list[str]:
|
|
return [l for l in stderr.splitlines() if l.startswith("WARNING: --no-onboard:")]
|
|
|
|
|
|
@pytest.mark.parametrize("agent", ALL_AGENTS)
|
|
def test_facts_expose_post_discovered(agent):
|
|
env = dict(os.environ, PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
|
|
res = subprocess.run(
|
|
[sys.executable, "-m", "lib_py.agents", "facts", agent],
|
|
capture_output=True, text=True, check=True, env=env,
|
|
)
|
|
eval_res = subprocess.run(
|
|
["bash", "-c", f"set -euo pipefail; eval {shlex.quote(res.stdout)}; echo \"$MAM_ID_POST_DISCOVERED\""],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
val = int(eval_res.stdout.strip())
|
|
adapter = get_adapter(agent)
|
|
assert val == (1 if adapter.id_lifecycle == POST_DISCOVERED else 0)
|
|
assert val == EXPECTED_POST_DISCOVERED[agent]
|
|
|
|
|
|
@pytest.mark.parametrize("agent", ALL_AGENTS)
|
|
def test_no_onboard_warning_dry_run(mam_sandbox, mock_herdr, mock_agents, monkeypatch, agent):
|
|
monkeypatch.setenv("XAI_API_KEY", "test-key")
|
|
ws = mam_sandbox / f"ws-dry-{agent}"
|
|
ws.mkdir(parents=True, exist_ok=True)
|
|
create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
|
|
res = subprocess.run(
|
|
["bash", str(create_sh), "--dry-run", "--no-onboard", "--workspace", str(ws),
|
|
"--agent", agent, "--role", "creator", "--session", f"t68-{agent}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=60,
|
|
)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
warns = _warn_lines(res.stderr)
|
|
assert len(warns) == EXPECTED_POST_DISCOVERED[agent], res.stderr
|
|
if warns:
|
|
assert "Issue #63" in warns[0] and agent in warns[0]
|
|
|
|
# Control: --dry-run with default onboarding emits 0 warning lines for all agents
|
|
ctl_res = subprocess.run(
|
|
["bash", str(create_sh), "--dry-run", "--workspace", str(ws),
|
|
"--agent", agent, "--role", "creator", "--session", f"t68-ctl-{agent}"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=60,
|
|
)
|
|
assert ctl_res.returncode == 0, ctl_res.stdout + ctl_res.stderr
|
|
assert _warn_lines(ctl_res.stderr) == [], ctl_res.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("agent", ALL_AGENTS)
|
|
def test_no_onboard_warning_keeps_exit_and_row(mam_sandbox, mock_herdr, mock_agents, monkeypatch, agent):
|
|
monkeypatch.setenv("XAI_API_KEY", "test-key")
|
|
ws = mam_sandbox / f"ws-real-{agent}"
|
|
ws.mkdir(parents=True, exist_ok=True)
|
|
session_name = f"t68-{agent}"
|
|
create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
lib_sh = mam_sandbox / ".agents" / "skills" / "lib.sh"
|
|
ws_slug = subprocess.run(
|
|
["bash", "-c", f'source "{lib_sh}" && derive_workspace_slug "{ws}"'],
|
|
capture_output=True, text=True, check=True, cwd=str(mam_sandbox),
|
|
).stdout.strip()
|
|
|
|
res = subprocess.run(
|
|
["bash", str(create_sh), "--no-onboard", "--workspace", str(ws),
|
|
"--agent", agent, "--role", "creator", "--session", session_name],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), timeout=120,
|
|
)
|
|
# 1. Exit code
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
# 2. Stable row fields (checked BEFORE warning assertion so base b9c828f fails only at step 3)
|
|
rows = _read_rows(mam_sandbox)
|
|
assert session_name in rows
|
|
row = rows[session_name]
|
|
pre_assigned = agent in ("claude", "grok")
|
|
expected_cmd_full = _norm_uuid(
|
|
get_adapter(agent).spawn_spec(
|
|
str(mock_agents / agent),
|
|
"00000000-0000-0000-0000-000000000000" if pre_assigned else "",
|
|
)
|
|
)
|
|
expected_source = "assigned" if pre_assigned else ("pending-discovery" if agent in ("opencode", "codex") else None)
|
|
expected_verified = False if agent in ("claude", "grok", "opencode", "codex") else None
|
|
expected_vis = "assigned (awaiting first message)" if pre_assigned else "unverified"
|
|
expected_cp = {
|
|
"workspace_root": os.path.realpath(mam_sandbox),
|
|
"agent_sessions_yaml": os.path.realpath(mam_sandbox / ".mam" / "agent-sessions.yaml"),
|
|
"mam_jobs_dir": os.path.realpath(mam_sandbox / ".mam" / "jobs"),
|
|
"delegate_job_logs_dir": os.path.realpath(mam_sandbox / ".mam" / "delegate_job_logs"),
|
|
}
|
|
|
|
assert row["name"] == session_name
|
|
assert row["status"] == "running"
|
|
assert row["role"] == "creator"
|
|
assert row["delegate_job_id"] is None
|
|
assert row["herdr_server"] == ws_slug
|
|
assert row["herdr_workspace"] == ws_slug.removeprefix("mam-")
|
|
assert row["pane"]["cmd"] == agent
|
|
assert _norm_uuid(row["pane"]["cmd_full"]) == expected_cmd_full
|
|
assert os.path.realpath(row["pane"]["cwd"]) == os.path.realpath(ws)
|
|
assert row.get("session_id_source") == expected_source
|
|
assert row.get("session_id_verified") == expected_verified
|
|
assert row["last_visible_status"] == expected_vis
|
|
assert row["control_plane"] == expected_cp
|
|
|
|
# 3. Warning line count
|
|
warns = _warn_lines(res.stderr)
|
|
assert len(warns) == EXPECTED_POST_DISCOVERED[agent], res.stderr
|
|
if warns:
|
|
assert "Issue #63" in warns[0] and agent in warns[0]
|
|
|
|
|
|
def test_no_onboard_warning_precedes_a_later_refusal(mam_sandbox, mock_herdr, mock_agents):
|
|
ws = mam_sandbox / "c68-ws"
|
|
ws.mkdir(parents=True, exist_ok=True)
|
|
epoch_a = int(time.time()) - 100
|
|
name_a, name_b = "c68-ws-creator-agy", "c68-ws-reviewer-agy"
|
|
create_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
|
|
holder = subprocess.Popen(["sleep", "120"])
|
|
try:
|
|
_write_rows(mam_sandbox, [_row(name_a, "agy", ws, epoch_a, pid=holder.pid)])
|
|
_mock_herdr_agent(mock_herdr, name_a, ws, holder.pid, epoch_a, pane_id="w1:p_a")
|
|
|
|
env0 = dict(os.environ, MAM_ID_ORDER_WAIT_S="0")
|
|
res = subprocess.run(
|
|
["bash", str(create_sh), "--workspace", str(ws), "--agent", "agy",
|
|
"--role", "reviewer", "--session", name_b, "--no-onboard"],
|
|
capture_output=True, text=True, cwd=str(mam_sandbox), env=env0, timeout=120,
|
|
)
|
|
assert res.returncode == 1, res.stdout + res.stderr
|
|
warns = _warn_lines(res.stderr)
|
|
assert len(warns) == 1, res.stderr
|
|
assert "Issue #63" in warns[0]
|
|
assert name_a in res.stderr
|
|
assert res.stderr.index("WARNING: --no-onboard:") < res.stderr.index(name_a), res.stderr
|
|
finally:
|
|
holder.kill()
|
|
holder.wait()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"opts,expect_onboard",
|
|
[
|
|
([], True),
|
|
(["--onboard"], True),
|
|
(["--no-onboard"], False),
|
|
(["--no-onboard", "--onboard"], True),
|
|
],
|
|
)
|
|
def test_prepare_forwards_onboarding_choice(ctl, opts, expect_onboard):
|
|
main = ctl
|
|
args_file = main / "captured_create_args.txt"
|
|
stub = main / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
|
stub.write_text(f'#!/usr/bin/env bash\nprintf "%s\\n" "$*" > "{args_file}"\nexit 0\n')
|
|
wt = _wt(main, "fwd")
|
|
try:
|
|
res = _prepare(main, wt, "--creator-agent", "claude", *opts)
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
tokens = args_file.read_text().split()
|
|
if expect_onboard:
|
|
assert "--onboard" in tokens, tokens
|
|
assert "--no-onboard" not in tokens, tokens
|
|
else:
|
|
assert "--no-onboard" in tokens, tokens
|
|
assert "--onboard" not in tokens, tokens
|
|
finally:
|
|
subprocess.run(["git", "worktree", "remove", "--force", str(wt)], cwd=str(main), capture_output=True)
|
|
|
|
|
|
def test_prepare_onboards_by_default_end_to_end(ctl, mock_herdr, mock_agents):
|
|
main = ctl
|
|
wt = _wt(main, "onboard-def")
|
|
try:
|
|
res = _prepare(main, wt, "--creator-agent", "claude")
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
(row,) = _rows(main)
|
|
jid = row["delegate_job_id"]
|
|
assert jid, row
|
|
job_path = main / ".mam" / "jobs" / f"{jid}.json"
|
|
assert job_path.is_file(), f"missing job record {job_path}"
|
|
rec = json.loads(job_path.read_text())
|
|
assert rec["status"] == "completed", rec
|
|
assert f"onboarding completed: job {jid}" in res.stdout, res.stdout
|
|
finally:
|
|
subprocess.run(["git", "worktree", "remove", "--force", str(wt)], cwd=str(main), capture_output=True)
|
|
|
|
|
|
def test_prepare_no_onboard_sends_none(ctl, mock_herdr, mock_agents):
|
|
main = ctl
|
|
wt = _wt(main, "no-onboard")
|
|
try:
|
|
res = _prepare(main, wt, "--creator-agent", "claude", "--no-onboard")
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
(row,) = _rows(main)
|
|
assert row["delegate_job_id"] is None, row
|
|
jobs_dir = main / ".mam" / "jobs"
|
|
if jobs_dir.is_dir():
|
|
for jf in jobs_dir.glob("*.json"):
|
|
rec = json.loads(jf.read_text())
|
|
assert rec.get("agent_session") != f"herdr:{row['name']}", rec
|
|
finally:
|
|
subprocess.run(["git", "worktree", "remove", "--force", str(wt)], cwd=str(main), capture_output=True)
|