|
|
|
@@ -38,6 +38,7 @@ from lib_py import control_plane # noqa: E402
|
|
|
|
|
from lib_py.delegation import engine, job_subscriber, mqtt_common, registry # noqa: E402
|
|
|
|
|
from test_issue21_phase2_loop_a import _mutate, _prepare, _rows, _skill, _wt # noqa: E402
|
|
|
|
|
from test_issue21_phase2_loop_a import ctl # noqa: E402,F401
|
|
|
|
|
from test_issue21_worktree_prereqs import _env_flags, _sent_texts # noqa: E402
|
|
|
|
|
|
|
|
|
|
UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
|
|
|
|
|
DROP = ("MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")
|
|
|
|
@@ -153,9 +154,9 @@ def test_registry_cli_writes_and_reads_next_to_the_registry(moved):
|
|
|
|
|
# L3 — the wrapper's last line names the directory it wrote
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
def _wrapper(ws: Path, *args, **extra) -> subprocess.CompletedProcess:
|
|
|
|
|
def _wrapper(ws: Path, *args, cwd: Path = None, **extra) -> subprocess.CompletedProcess:
|
|
|
|
|
return subprocess.run(["bash", str(WRAPPER), *args], capture_output=True, text=True,
|
|
|
|
|
env=_env(ws, **extra), cwd=str(ws), timeout=120)
|
|
|
|
|
env=_env(ws, **extra), cwd=str(cwd or ws), timeout=120)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _only_job_id(reg: Path) -> str:
|
|
|
|
@@ -280,6 +281,10 @@ def test_worktree_prepare_keeps_the_logs_next_to_the_control_root_registry(ctl,
|
|
|
|
|
jid = row["delegate_job_id"]
|
|
|
|
|
assert (main / "elsewhere" / "delegate_job_logs" / jid / "status.json").is_file()
|
|
|
|
|
assert not (main / ".mam" / "delegate_job_logs").exists()
|
|
|
|
|
# the pane carries the registry, not the audit-log root derived from it
|
|
|
|
|
flags = _env_flags(mock_herdr)
|
|
|
|
|
assert any(f == f"MAM_JOBS_DIR={os.path.realpath(reg)}" for f in flags), flags
|
|
|
|
|
assert not [f for f in flags if f.startswith("DELEGATE_JOB_LOGS_DIR=")], flags
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@@ -380,3 +385,90 @@ def test_old_audit_terminal_does_not_complete_the_redispatched_attempt(moved, mo
|
|
|
|
|
pending, terminal = {jid}, {}
|
|
|
|
|
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
|
|
|
|
|
assert pending == set() and terminal == {jid: "completed"}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# plan §2-B (user decision): the derivable audit-log root is recorded, not forwarded
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
def _cp_cli(ws: Path, *args, **extra) -> subprocess.CompletedProcess:
|
|
|
|
|
return subprocess.run([sys.executable, "-m", "lib_py.control_plane", *args], capture_output=True, text=True,
|
|
|
|
|
env=_env(ws, **extra), cwd=str(ws), timeout=60)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_prefix_forwards_only_what_cannot_be_derived(moved, monkeypatch, tmp_path):
|
|
|
|
|
ws, reg, logs = moved
|
|
|
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
|
|
|
|
|
d = control_plane.describe(str(ws / "wt"))
|
|
|
|
|
assert d["forward"] is True
|
|
|
|
|
assert d["delegate_job_logs_dir"] == os.path.realpath(logs), "the row still records the sibling"
|
|
|
|
|
p = control_plane.prefix(d)
|
|
|
|
|
assert "WORKSPACE_ROOT=" in p and "AGENT_SESSIONS_YAML=" in p and f"MAM_JOBS_DIR={os.path.realpath(reg)} " in p, p
|
|
|
|
|
assert "DELEGATE_JOB_LOGS_DIR=" not in p, p
|
|
|
|
|
# an audit-log root that cannot be derived is still forwarded
|
|
|
|
|
custom = tmp_path / "custom-logs"
|
|
|
|
|
monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(custom))
|
|
|
|
|
d2 = control_plane.describe(str(ws / "wt"))
|
|
|
|
|
assert d2["delegate_job_logs_dir"] == os.path.realpath(custom)
|
|
|
|
|
assert f"DELEGATE_JOB_LOGS_DIR={os.path.realpath(custom)} " in control_plane.prefix(d2)
|
|
|
|
|
# the CLI lib.sh calls agrees on both
|
|
|
|
|
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR")
|
|
|
|
|
res = _cp_cli(ws, "json", str(ws / "wt"), MAM_JOBS_DIR=str(reg))
|
|
|
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
|
cli = _cp_cli(ws, "prefix", res.stdout.strip(), MAM_JOBS_DIR=str(reg))
|
|
|
|
|
assert cli.returncode == 0 and "MAM_JOBS_DIR=" in cli.stdout and "DELEGATE_JOB_LOGS_DIR=" not in cli.stdout, cli.stdout
|
|
|
|
|
res = _cp_cli(ws, "json", str(ws / "wt"), MAM_JOBS_DIR=str(reg), DELEGATE_JOB_LOGS_DIR=str(custom))
|
|
|
|
|
cli = _cp_cli(ws, "prefix", res.stdout.strip(), MAM_JOBS_DIR=str(reg), DELEGATE_JOB_LOGS_DIR=str(custom))
|
|
|
|
|
assert f"DELEGATE_JOB_LOGS_DIR={os.path.realpath(custom)} " in cli.stdout, cli.stdout
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_create_binds_the_sibling_and_does_not_forward_it(mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
|
|
|
|
tmp = mam_sandbox
|
|
|
|
|
reg, sibling = tmp / "elsewhere" / "jobs", tmp / "elsewhere" / "delegate_job_logs"
|
|
|
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
|
|
|
|
|
monkeypatch.delenv("DELEGATE_JOB_LOGS_DIR", raising=False)
|
|
|
|
|
wt = tmp / "wt"
|
|
|
|
|
wt.mkdir()
|
|
|
|
|
res = subprocess.run(["bash", _skill(tmp, "multi-agent-mux-create", "scripts", "create_session.sh"),
|
|
|
|
|
"--workspace", str(wt), "--agent", "claude", "--role", "creator", "--onboard"],
|
|
|
|
|
capture_output=True, text=True, cwd=str(tmp), timeout=180)
|
|
|
|
|
assert res.returncode == 0, res.stdout + res.stderr
|
|
|
|
|
(row,) = _rows(tmp)
|
|
|
|
|
assert row["control_plane"]["mam_jobs_dir"] == os.path.realpath(reg)
|
|
|
|
|
assert row["control_plane"]["delegate_job_logs_dir"] == os.path.realpath(sibling)
|
|
|
|
|
flags = _env_flags(mock_herdr)
|
|
|
|
|
assert any(f == f"MAM_JOBS_DIR={os.path.realpath(reg)}" for f in flags), flags
|
|
|
|
|
assert not [f for f in flags if f.startswith("DELEGATE_JOB_LOGS_DIR=")], flags
|
|
|
|
|
# the injected publish commands still pin the root per command
|
|
|
|
|
texts = [t for t in _sent_texts(json.loads(Path(mock_herdr).read_text())) if "publish_event.py" in t]
|
|
|
|
|
assert texts and f"DELEGATE_JOB_LOGS_DIR={os.path.realpath(sibling)} " in "".join(texts), texts
|
|
|
|
|
jid = row["delegate_job_id"]
|
|
|
|
|
assert (sibling / jid / "status.json").is_file()
|
|
|
|
|
assert not (tmp / ".mam" / "delegate_job_logs").exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_a_forwarded_pane_moves_its_logs_with_the_registry(tmp_path, monkeypatch):
|
|
|
|
|
"""The issue's scene: a worktree creator pane after §2-B carries the control
|
|
|
|
|
root's WORKSPACE_ROOT and MAM_JOBS_DIR but no DELEGATE_JOB_LOGS_DIR."""
|
|
|
|
|
for k in DROP:
|
|
|
|
|
monkeypatch.delenv(k, raising=False)
|
|
|
|
|
main, wt, exp = tmp_path / "main", tmp_path / "wt", tmp_path / "exp" / "jobs"
|
|
|
|
|
real_logs = main / ".mam" / "delegate_job_logs"
|
|
|
|
|
real_logs.mkdir(parents=True)
|
|
|
|
|
wt.mkdir()
|
|
|
|
|
pane = dict(WORKSPACE_ROOT=str(main), MAM_JOBS_DIR=str(main / ".mam" / "jobs"))
|
|
|
|
|
res = _wrapper(main, "submit", "--dry-run", "--agent", "human", "--registry-dir", str(exp),
|
|
|
|
|
"--workdir", str(wt), "--prompt", "t", cwd=wt, **pane)
|
|
|
|
|
assert res.returncode == 0, res.stdout[-800:] + res.stderr[-800:]
|
|
|
|
|
jid = _only_job_id(exp)
|
|
|
|
|
assert res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(tmp_path / "exp")) / "delegate_job_logs" / jid)
|
|
|
|
|
assert (tmp_path / "exp" / "delegate_job_logs" / jid / "meta.json").is_file()
|
|
|
|
|
assert os.listdir(real_logs) == [], "the experiment leaked into the control root"
|
|
|
|
|
# the usual run in that pane is unchanged: the default registry's sibling
|
|
|
|
|
res = _wrapper(main, "submit", "--dry-run", "--agent", "human", "--workdir", str(wt), "--prompt", "t",
|
|
|
|
|
cwd=wt, **pane)
|
|
|
|
|
assert res.returncode == 0, res.stdout[-800:] + res.stderr[-800:]
|
|
|
|
|
jid2 = _only_job_id(main / ".mam" / "jobs")
|
|
|
|
|
assert res.stdout.strip().splitlines()[-1] == str(Path(os.path.realpath(main)) / ".mam" / "delegate_job_logs" / jid2)
|
|
|
|
|
assert os.listdir(real_logs) == [jid2]
|
|
|
|
|
assert not (wt / ".mam" / "delegate_job_logs").exists()
|