GodopuandClaude Opus 5
12cba25c18
Two failures the full suite caught, both mine. finalize.py started in lib_py/loop, which made publish_event import a sibling package. That is the rule phase 1 wrote down and phase 2a already made me move the engine for: under a frozen loop, delegation is bound to one tree while the parent may be another, so reaching sideways mixes them. The manifest is written into the job's own report directory and the CLI that declares it is a delegation entry point, so finalize belongs there. Second time on the same rule, and the guard caught it both times. The publish_event --help golden no longer matched, because phase 2b intentionally added the finalization options. The diff is exactly those three options and nothing else, so the capture is refreshed rather than the assertion weakened, and the test now states when a refresh is legitimate: an intended extension, never drift nobody meant. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
1215 lines
50 KiB
Python
1215 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""Issue #5 Phase 1 — delegation extraction characterization & contract tests.
|
|
|
|
Written BEFORE the `lib_py/delegation/` extraction (P1-00) so the refactor has a
|
|
real regression oracle: the three shell helpers in `lib.sh` had zero direct test
|
|
coverage. Every test here must pass identically against the legacy inline bodies
|
|
and against the Python-backed ones, which is why the shell tests are
|
|
parametrized over `MAM_DELEGATION_PY` in {"0", "1"}.
|
|
|
|
Sections
|
|
1. C-01..C-09 shell helper golden behaviour (non-fatal contract, stdout shape,
|
|
daemon PID/dedup/logging).
|
|
2. CLI contract snapshots — argparse option names/choices/defaults, asserted
|
|
against the parser rather than --help text so cosmetic wording is free.
|
|
3. Auto-completion invariant guards — the three surfaces that decide job status
|
|
must not shift by a single line during Phase 1.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
SKILLS = REPO_ROOT / ".agents" / "skills"
|
|
FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "phase1"
|
|
|
|
DELEGATION_MODES = ("0", "1") # MAM_DELEGATION_PY: legacy inline body / python body
|
|
|
|
|
|
# ==============================================================================
|
|
# Helpers
|
|
# ==============================================================================
|
|
|
|
def _sandbox_skills(mam_sandbox) -> Path:
|
|
return Path(mam_sandbox) / "skills"
|
|
|
|
|
|
def _run_helper(mam_sandbox, snippet: str, env=None, timeout=30):
|
|
"""Source the sandbox lib.sh and run a shell snippet against it."""
|
|
lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
|
|
run_env = dict(os.environ)
|
|
run_env.pop("HERDR_SESSION_NAME", None)
|
|
run_env.pop("HERDR_SERVER_NAME", None)
|
|
if env:
|
|
run_env.update({k: str(v) for k, v in env.items()})
|
|
return subprocess.run(
|
|
["bash", "-c", 'source "{}" >/dev/null 2>&1; {}'.format(lib_sh, snippet)],
|
|
capture_output=True, text=True, env=run_env, timeout=timeout,
|
|
)
|
|
|
|
|
|
def _stub_env(tmp_path, *, exit_code=0, stdout=""):
|
|
"""Point _delegate_py_bin at the recording stub via AGENT_PYTHON_BIN."""
|
|
log = tmp_path / "stub_argv.log"
|
|
return {
|
|
"AGENT_PYTHON_BIN": str(FIXTURES / "fake_python.sh"),
|
|
"STUB_ARGV_LOG": str(log),
|
|
"STUB_EXIT": str(exit_code),
|
|
"STUB_STDOUT": stdout,
|
|
}, log
|
|
|
|
|
|
def _install_fake_monitor(mam_sandbox) -> Path:
|
|
"""Replace the sandbox reconcile.sh with a sleeping stub and return its path."""
|
|
target = (Path(mam_sandbox) / ".agents" / "skills"
|
|
/ "multi-agent-mux-monitor" / "scripts" / "reconcile.sh")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text((FIXTURES / "fake_reconcile.sh").read_text(encoding="utf-8"),
|
|
encoding="utf-8")
|
|
target.chmod(0o755)
|
|
return target
|
|
|
|
|
|
def _kill(pids):
|
|
for pid in pids:
|
|
try:
|
|
os.kill(int(pid), signal.SIGKILL)
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
|
|
def _alive(pid) -> bool:
|
|
try:
|
|
os.kill(int(pid), 0)
|
|
return True
|
|
except (OSError, ValueError):
|
|
return False
|
|
|
|
|
|
# ==============================================================================
|
|
# 1. Shell helper golden behaviour (C-01 .. C-09)
|
|
# ==============================================================================
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c01_publish_event_empty_job_is_noop(mam_sandbox, mode):
|
|
"""C-01: an empty job id returns 0 without ever starting an interpreter."""
|
|
env, log = _stub_env(mam_sandbox, exit_code=7)
|
|
env["MAM_DELEGATION_PY"] = mode
|
|
res = _run_helper(mam_sandbox, 'delegate_publish_event "" started', env=env)
|
|
assert res.returncode == 0, res.stderr
|
|
assert not log.exists(), "python must not be invoked for an empty job id"
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c02_publish_event_missing_script_is_non_fatal(mam_sandbox, mode):
|
|
"""C-02: a missing publish_event.py must not abort the caller."""
|
|
script = (_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job"
|
|
/ "scripts" / "publish_event.py")
|
|
script.unlink()
|
|
env, _ = _stub_env(mam_sandbox)
|
|
env["MAM_DELEGATION_PY"] = mode
|
|
res = _run_helper(mam_sandbox, 'delegate_publish_event j1 started', env=env)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c03_publish_event_broker_failure_is_non_fatal(mam_sandbox, mode):
|
|
"""C-03: a non-zero interpreter exit (broker down) still returns 0."""
|
|
env, log = _stub_env(mam_sandbox, exit_code=2)
|
|
env["MAM_DELEGATION_PY"] = mode
|
|
res = _run_helper(
|
|
mam_sandbox,
|
|
'delegate_publish_event j1 completed "done"; echo "rc=$?"',
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
assert "rc=0" in res.stdout, res.stdout
|
|
assert log.exists(), "the interpreter should have been invoked"
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c04_publish_event_forwards_job_event_detail(mam_sandbox, mode):
|
|
"""C-04: the job id, event and detail reach the interpreter argv intact."""
|
|
env, log = _stub_env(mam_sandbox)
|
|
env["MAM_DELEGATION_PY"] = mode
|
|
res = _run_helper(
|
|
mam_sandbox,
|
|
'delegate_publish_event abc123 progress "half way"',
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
argv = log.read_text(encoding="utf-8")
|
|
assert "--job abc123" in argv, argv
|
|
assert "--event progress" in argv, argv
|
|
assert "--detail half way" in argv, argv
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c05_submit_job_stdout_is_exactly_the_job_id(mam_sandbox, mode):
|
|
"""C-05: create_session.sh captures this with $( ) — stdout must be the id alone."""
|
|
env, _ = _stub_env(mam_sandbox, stdout="deadbeef")
|
|
env["MAM_DELEGATION_PY"] = mode
|
|
res = _run_helper(
|
|
mam_sandbox,
|
|
'delegate_submit_job "a prompt" claude-code herdr:claude',
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
assert res.stdout == "deadbeef\n", repr(res.stdout)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c06_watchdog_missing_monitor_script_errors(mam_sandbox, mode):
|
|
"""C-06: exact stderr wording and rc=1 when reconcile.sh is absent."""
|
|
workdir = Path(mam_sandbox) / "empty_ws"
|
|
workdir.mkdir()
|
|
expected = (workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
|
|
/ "scripts" / "reconcile.sh")
|
|
res = _run_helper(
|
|
mam_sandbox,
|
|
'start_watchdog job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode},
|
|
)
|
|
assert res.returncode == 1, res.stdout
|
|
assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c07_watchdog_spawns_daemon_with_workdir_and_log(mam_sandbox, mode):
|
|
"""C-07: prints a live numeric PID, anchors cwd at workdir, appends to the log."""
|
|
_install_fake_monitor(mam_sandbox)
|
|
workdir = Path(mam_sandbox)
|
|
(workdir / ".mam").mkdir(exist_ok=True)
|
|
spawned = []
|
|
try:
|
|
res = _run_helper(
|
|
mam_sandbox,
|
|
'start_watchdog job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode},
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
pid = res.stdout.strip()
|
|
assert re.fullmatch(r"\d+", pid), repr(res.stdout)
|
|
spawned.append(pid)
|
|
deadline = time.time() + 5
|
|
log_file = workdir / ".mam" / "multi-agent-mux-monitor.log"
|
|
while time.time() < deadline and not log_file.exists():
|
|
time.sleep(0.1)
|
|
assert log_file.exists(), "monitor log was not created"
|
|
assert _alive(pid), "spawned daemon is not running"
|
|
finally:
|
|
_kill(spawned)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c08_watchdog_second_call_reuses_running_daemon(mam_sandbox, mode):
|
|
"""C-08: the pgrep pre-check must return the existing PID, not spawn a second."""
|
|
monitor = _install_fake_monitor(mam_sandbox)
|
|
workdir = Path(mam_sandbox)
|
|
(workdir / ".mam").mkdir(exist_ok=True)
|
|
spawned = []
|
|
try:
|
|
first = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode})
|
|
pid1 = first.stdout.strip()
|
|
spawned.append(pid1)
|
|
assert _alive(pid1)
|
|
second = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode})
|
|
assert second.returncode == 0, second.stderr
|
|
assert second.stdout.strip() == pid1, (first.stdout, second.stdout)
|
|
matches = subprocess.run(
|
|
["pgrep", "-f", "bash {} --subscribe".format(monitor)],
|
|
capture_output=True, text=True,
|
|
).stdout.split()
|
|
spawned.extend(matches)
|
|
assert len(matches) == 1, "expected exactly one daemon, got {}".format(matches)
|
|
finally:
|
|
_kill(spawned)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", DELEGATION_MODES)
|
|
def test_c09_start_watchdog_and_ensure_subscriber_daemon_agree(mam_sandbox, mode):
|
|
"""C-09: `start_watchdog` survives as an alias once the python body lands.
|
|
|
|
Passes on the legacy tree (only start_watchdog defined) and on the refactored
|
|
tree (both defined and behaviourally identical)."""
|
|
_install_fake_monitor(mam_sandbox)
|
|
workdir = Path(mam_sandbox)
|
|
(workdir / ".mam").mkdir(exist_ok=True)
|
|
probe = _run_helper(
|
|
mam_sandbox,
|
|
'declare -F start_watchdog >/dev/null && echo HAS_SW; '
|
|
'declare -F ensure_subscriber_daemon >/dev/null && echo HAS_ESD',
|
|
env={"MAM_DELEGATION_PY": mode},
|
|
)
|
|
assert "HAS_SW" in probe.stdout, "start_watchdog must always remain defined"
|
|
if "HAS_ESD" not in probe.stdout:
|
|
return # legacy tree: alias not introduced yet (P1-02)
|
|
|
|
spawned = []
|
|
try:
|
|
a = _run_helper(mam_sandbox, 'start_watchdog job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode})
|
|
b = _run_helper(mam_sandbox, 'ensure_subscriber_daemon job1 "{}"'.format(workdir),
|
|
env={"MAM_DELEGATION_PY": mode})
|
|
spawned += [a.stdout.strip(), b.stdout.strip()]
|
|
assert a.returncode == b.returncode == 0, (a.stderr, b.stderr)
|
|
assert a.stdout.strip() == b.stdout.strip(), (a.stdout, b.stdout)
|
|
finally:
|
|
_kill(spawned)
|
|
|
|
|
|
# ==============================================================================
|
|
# 2. CLI contract snapshots
|
|
# ==============================================================================
|
|
|
|
def _parser_snapshot(parser):
|
|
"""{option string: (nargs-ish kind, choices, default)} for one parser level."""
|
|
snap = {}
|
|
for action in parser._actions:
|
|
if not action.option_strings:
|
|
snap["<positional:{}>".format(action.dest)] = (
|
|
action.type.__name__ if action.type else None,
|
|
tuple(action.choices) if action.choices else None,
|
|
action.default,
|
|
)
|
|
continue
|
|
key = ",".join(sorted(action.option_strings))
|
|
snap[key] = (
|
|
action.type.__name__ if action.type else None,
|
|
tuple(action.choices) if action.choices else None,
|
|
action.default,
|
|
)
|
|
return snap
|
|
|
|
|
|
def _import_delegation(mam_sandbox, name):
|
|
"""Import a delegation module the way production callers do (top-level name)."""
|
|
scripts = str(_sandbox_skills(mam_sandbox) / "multi-agent-mux-delegate-job" / "scripts")
|
|
if scripts not in sys.path:
|
|
sys.path.insert(0, scripts)
|
|
return __import__(name)
|
|
|
|
|
|
def test_publish_event_cli_contract_snapshot(mam_sandbox):
|
|
publish_event = _import_delegation(mam_sandbox, "publish_event")
|
|
assert publish_event.VALID_EVENTS == (
|
|
"started", "permission_required", "progress", "completed", "error", "cancelled",
|
|
)
|
|
assert publish_event.TERMINAL_EVENTS == ("completed", "error", "cancelled")
|
|
# main() builds its parser inline, so --help is the only way to exercise it.
|
|
with pytest.raises(SystemExit) as exc:
|
|
publish_event.main(["--help"])
|
|
assert exc.value.code == 0
|
|
# Defaults that shell and wrapper callers depend on.
|
|
src = Path(publish_event.__file__).read_text(encoding="utf-8")
|
|
assert '"--attempts", type=int, default=3' in src
|
|
assert '"--registry-dir", default=DEFAULT_REGISTRY_DIR' in src
|
|
assert '"--event", default="progress", choices=VALID_EVENTS' in src
|
|
|
|
|
|
def test_registry_cli_contract_snapshot(mam_sandbox):
|
|
registry = _import_delegation(mam_sandbox, "registry")
|
|
parser = registry._build_parser()
|
|
top = _parser_snapshot(parser)
|
|
assert "--registry-dir" in top
|
|
subparsers = [a for a in parser._actions if hasattr(a, "choices") and isinstance(a.choices, dict)]
|
|
assert subparsers, "registry must expose subcommands"
|
|
commands = set(subparsers[0].choices)
|
|
assert commands == {
|
|
"register", "list", "get", "status", "update", "get-feedback", "pick", "logs",
|
|
}, commands
|
|
reg = _parser_snapshot(subparsers[0].choices["register"])
|
|
assert reg["--prompt"][2] is None
|
|
assert reg["--agent"][2] == "claude-code"
|
|
assert reg["--agent-session"][2] == "herdr:claude"
|
|
assert reg["--role"][2] == "Worker"
|
|
assert reg["--timeout"][2] == 3600
|
|
assert reg["--idle-timeout"][2] == 120
|
|
assert reg["--bits"][2] == 32
|
|
assert reg["--job-type"][1] == ("direct", "loop", "discuss")
|
|
assert reg["--max-iterations"][2] == 5
|
|
assert registry.VALID_STATUSES == (
|
|
"pending", "running", "completed", "error", "cancelled",
|
|
)
|
|
|
|
|
|
def test_job_subscriber_cli_contract_snapshot(mam_sandbox):
|
|
job_subscriber = _import_delegation(mam_sandbox, "job_subscriber")
|
|
assert job_subscriber.TERMINAL_EVENTS == ("completed", "error")
|
|
src = Path(job_subscriber.__file__).read_text(encoding="utf-8")
|
|
for doc in (" 0 all watched jobs reached", " 1 any watched job reached",
|
|
" 2 timed out"):
|
|
assert doc in src, doc
|
|
assert 'return 3' in src, "fatal-error exit code 3 must survive"
|
|
|
|
|
|
# ==============================================================================
|
|
# 3. Auto-completion invariant guards (Phase 1 must not move these)
|
|
# ==============================================================================
|
|
|
|
def test_event_to_status_mapping_is_frozen(mam_sandbox):
|
|
publish_event = _import_delegation(mam_sandbox, "publish_event")
|
|
assert publish_event.EVENT_TO_STATUS == {
|
|
"started": "running",
|
|
"completed": "completed",
|
|
"error": "error",
|
|
"cancelled": "cancelled",
|
|
}
|
|
|
|
|
|
def test_wrapper_exit_code_mapping_literals_are_frozen():
|
|
wrapper = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
|
text = wrapper.read_text(encoding="utf-8")
|
|
for literal in (
|
|
'if [[ $sub_rc -eq 0 ]]; then',
|
|
'elif [[ $sub_rc -eq 1 ]]; then',
|
|
'elif [[ $sub_rc -eq 3 ]]; then',
|
|
'job_status="completed"',
|
|
'job_status="error"',
|
|
):
|
|
assert literal in text, literal
|
|
|
|
|
|
def test_reconcile_terminal_mutation_literals_are_frozen():
|
|
reconcile = SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
|
|
text = reconcile.read_text(encoding="utf-8")
|
|
for literal in (
|
|
"s['delegate_job_id'] = None",
|
|
"s['status'] = 'terminated'",
|
|
"kill-session",
|
|
"if _event in ('completed', 'cancelled')",
|
|
):
|
|
assert literal in text, literal
|
|
|
|
|
|
# ==============================================================================
|
|
# 4. P1-01 — package binding, provenance and context matrix (PV-1, PV-4, PV-5)
|
|
# ==============================================================================
|
|
|
|
DELEGATION_MODULES = ("mqtt_common", "registry", "publish_event", "job_subscriber")
|
|
PKG_DIR = SKILLS / "lib_py" / "delegation"
|
|
SHIM_DIR = SKILLS / "multi-agent-mux-delegate-job" / "scripts"
|
|
|
|
|
|
def test_pv5_all_four_shims_share_one_bootstrap_block():
|
|
"""PV-5: the four shims differ only by the module name they bind."""
|
|
normalised = {}
|
|
for mod in DELEGATION_MODULES:
|
|
text = (SHIM_DIR / (mod + ".py")).read_text(encoding="utf-8")
|
|
normalised[mod] = text.replace(mod, "{mod}")
|
|
base = normalised["mqtt_common"]
|
|
for mod, text in normalised.items():
|
|
assert text == base, "shim {} diverged from the shared bootstrap".format(mod)
|
|
assert "sys.path.insert" not in base, "the shim must never take sys.path precedence"
|
|
assert "sys.path.append(_SKILLS)" in base, "standalone fallback must remain"
|
|
|
|
|
|
def test_pv1_identity_and_provenance_of_every_alias(mam_sandbox, tmp_path_factory):
|
|
"""PV-1: each alias IS the package module, and all four share one tree."""
|
|
import lib_py.delegation
|
|
|
|
paths = set()
|
|
for mod in DELEGATION_MODULES:
|
|
alias = _import_delegation(mam_sandbox, mod)
|
|
pkg_mod = getattr(lib_py.delegation, mod, None)
|
|
if pkg_mod is None:
|
|
import importlib
|
|
pkg_mod = importlib.import_module("lib_py.delegation." + mod)
|
|
assert alias is pkg_mod, "{}: alias is not the package module".format(mod)
|
|
paths.add(str(Path(alias.__file__).resolve().parent))
|
|
|
|
assert len(paths) == 1, "delegation modules span multiple trees: {}".format(paths)
|
|
bound = Path(paths.pop())
|
|
real = PKG_DIR.resolve()
|
|
# Provenance, not just identity: the bound tree is either the real package or a
|
|
# copy under pytest's own basetemp. Checked against the actual basetemp rather
|
|
# than sniffing for "pytest-" in the path, so an unrelated directory that merely
|
|
# looks temporary cannot satisfy it.
|
|
basetemp = Path(tmp_path_factory.getbasetemp()).resolve()
|
|
try:
|
|
bound.relative_to(basetemp)
|
|
under_basetemp = True
|
|
except ValueError:
|
|
under_basetemp = False
|
|
assert bound == real or under_basetemp, (
|
|
"delegation bound to an unexpected tree: {} (real={}, basetemp={})".format(
|
|
bound, real, basetemp)
|
|
)
|
|
# the parent package must never be dragged along with it
|
|
assert Path(lib_py.__file__).resolve().parent == (SKILLS / "lib_py").resolve()
|
|
|
|
|
|
def test_delegation_package_imports_no_sibling_lib_py_modules():
|
|
"""D1(b): a frozen delegation package must not reach into the parent tree."""
|
|
import ast
|
|
|
|
offenders = []
|
|
for path in sorted(PKG_DIR.glob("*.py")):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
names = []
|
|
if isinstance(node, ast.Import):
|
|
names = [a.name for a in node.names]
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
names = [node.module]
|
|
for name in names:
|
|
if name in DELEGATION_MODULES:
|
|
offenders.append("{}:{} top-level `{}`".format(path.name, node.lineno, name))
|
|
if name == "lib_py" or (
|
|
name.startswith("lib_py.") and not name.startswith("lib_py.delegation")
|
|
):
|
|
offenders.append("{}:{} sibling `{}`".format(path.name, node.lineno, name))
|
|
assert not offenders, "forbidden imports inside lib_py.delegation:\n " + "\n ".join(offenders)
|
|
|
|
|
|
def test_importing_delegation_does_not_mutate_environment(mam_sandbox):
|
|
"""D6: importing the library must leave os.environ alone; env loads at entry points."""
|
|
import importlib
|
|
mqtt_common = _import_delegation(mam_sandbox, "mqtt_common")
|
|
before = set(os.environ)
|
|
importlib.reload(mqtt_common)
|
|
assert set(os.environ) == before, "library import injected environment variables"
|
|
# idempotent: the second call is a no-op
|
|
mqtt_common.ensure_env_loaded()
|
|
keys = set(os.environ)
|
|
mqtt_common.ensure_env_loaded()
|
|
assert set(os.environ) == keys
|
|
|
|
|
|
def test_shim_help_output_matches_pre_move_capture():
|
|
"""The shim must be indistinguishable from the implementation at the CLI.
|
|
|
|
The captures were taken before the phase 1 move to prove the move changed
|
|
nothing. They are refreshed only when the CLI is *intentionally* extended
|
|
(phase 2b added the finalization options to publish_event); the guard's job
|
|
is to catch drift nobody meant, which it still does.
|
|
"""
|
|
for mod in ("publish_event", "registry", "job_subscriber"):
|
|
golden = (FIXTURES / "help-{}.txt".format(mod)).read_text(encoding="utf-8")
|
|
res = subprocess.run(
|
|
[sys.executable, str(SHIM_DIR / (mod + ".py")), "--help"],
|
|
capture_output=True, text=True, timeout=60,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
assert res.stdout + res.stderr == golden, "--help drifted for {}".format(mod)
|
|
|
|
|
|
# --- PV-4: the context matrix, each scenario in a fresh interpreter ------------
|
|
|
|
@pytest.fixture(scope="module")
|
|
def shim_trees(tmp_path_factory):
|
|
"""Four copies of the live skills tree: real / sandbox / sandbox2 / freeze."""
|
|
import shutil
|
|
|
|
root = tmp_path_factory.mktemp("shim_trees")
|
|
for name in ("real", "sandbox", "sandbox2", "freeze"):
|
|
dest = root / name
|
|
(dest / "multi-agent-mux-delegate-job").mkdir(parents=True)
|
|
shutil.copytree(SKILLS / "lib_py", dest / "lib_py")
|
|
shutil.copytree(SHIM_DIR, dest / "multi-agent-mux-delegate-job" / "scripts")
|
|
return root
|
|
|
|
|
|
def _run_scenario(shim_trees, scenario, pythonpath=None):
|
|
env = dict(os.environ)
|
|
env.pop("PYTHONPATH", None)
|
|
if pythonpath:
|
|
env["PYTHONPATH"] = str(shim_trees / pythonpath)
|
|
res = subprocess.run(
|
|
[sys.executable, str(FIXTURES / "shim_contexts.py"), scenario]
|
|
+ [str(shim_trees / t) for t in ("real", "sandbox", "sandbox2", "freeze")],
|
|
capture_output=True, text=True, env=env, timeout=120,
|
|
)
|
|
assert res.returncode == 0, "scenario {} failed:\n{}".format(scenario, res.stderr)
|
|
return json.loads(res.stdout)
|
|
|
|
|
|
def _delegation_trees(report):
|
|
keys = ("alias_mqtt_common", "alias_publish_event", "pkg_delegation",
|
|
"pkg_mqtt_common", "pkg_publish_event")
|
|
return {report[k] for k in keys if report[k] is not None}
|
|
|
|
|
|
@pytest.mark.parametrize("scenario,expected_delegation", [
|
|
("A", "sandbox"), # sandbox shim binds first
|
|
("A_pe", "sandbox"), # non-leaf module alone: dependencies must follow it
|
|
("B", "sandbox"), # real lib_py.agents already imported: delegation still binds to shim tree
|
|
("B2", "real"), # real delegation already bound: bind-once keeps it
|
|
("D", "sandbox"), # a second sandbox must not rebind
|
|
("E", "sandbox"), # direct package import after a shim resolves to the bound tree
|
|
])
|
|
def test_pv4_pytest_family_binds_one_tree_and_spares_the_parent(
|
|
shim_trees, scenario, expected_delegation
|
|
):
|
|
rep = _run_scenario(shim_trees, scenario)
|
|
assert _delegation_trees(rep) == {expected_delegation}, rep
|
|
assert rep["lib_py"] == "real", "parent lib_py must stay on the real tree: {}".format(rep)
|
|
if rep["lib_py_agents"] is not None:
|
|
assert rep["lib_py_agents"] == "real", rep
|
|
for name, same in rep["identity"].items():
|
|
if same is not None:
|
|
assert same, "identity broken for {} in {}".format(name, scenario)
|
|
assert "sandbox" not in rep["lib_py_path_entries"], rep
|
|
assert "sandbox2" not in rep["lib_py_path_entries"], rep
|
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["C1", "C1_pe"])
|
|
def test_pv4_standalone_bootstraps_from_its_own_tree(shim_trees, scenario):
|
|
"""K4: no ambient lib_py anywhere, so the shim may append its own tree."""
|
|
rep = _run_scenario(shim_trees, scenario)
|
|
assert _delegation_trees(rep) == {"freeze"}, rep
|
|
assert rep["lib_py"] == "freeze", rep
|
|
|
|
|
|
@pytest.mark.parametrize("scenario", ["C2", "C2_pe"])
|
|
def test_pv4_frozen_copy_wins_over_a_live_pythonpath(shim_trees, scenario):
|
|
"""K2: the shape every delegated job runs -- frozen script, live PYTHONPATH.
|
|
|
|
B-13 requires the frozen code to execute; the parent lib_py may stay live
|
|
because D1(b) forbids the delegation package from importing siblings."""
|
|
rep = _run_scenario(shim_trees, scenario, pythonpath="real")
|
|
assert _delegation_trees(rep) == {"freeze"}, rep
|
|
assert rep["lib_py"] == "real", rep
|
|
for name, same in rep["identity"].items():
|
|
if same is not None:
|
|
assert same, name
|
|
|
|
|
|
def test_pv4_k3_live_helpers_resolve_to_the_live_tree(shim_trees):
|
|
rep = _run_scenario(shim_trees, "K3")
|
|
assert _delegation_trees(rep) == {"real"}, rep
|
|
assert rep["lib_py"] == "real", rep
|
|
|
|
|
|
def test_k1_module_execution_follows_pythonpath(shim_trees):
|
|
"""K1: `python -m lib_py.delegation.<mod>` is resolved by PYTHONPATH, not by a shim."""
|
|
for tree in ("real", "freeze"):
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = str(shim_trees / tree)
|
|
res = subprocess.run(
|
|
[sys.executable, "-c",
|
|
"import lib_py.delegation as d; print(list(d.__path__)[0])"],
|
|
capture_output=True, text=True, env=env, timeout=60,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
assert res.stdout.strip() == str(shim_trees / tree / "lib_py" / "delegation"), res.stdout
|
|
|
|
|
|
def test_k6_subprocess_executes_the_copy_it_was_given(shim_trees):
|
|
"""K6: e2e-style subprocess execution runs the tree the path points at."""
|
|
env = dict(os.environ)
|
|
env.pop("PYTHONPATH", None)
|
|
shim = shim_trees / "sandbox" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
|
|
res = subprocess.run([sys.executable, str(shim), "--help"],
|
|
capture_output=True, text=True, env=env, timeout=60)
|
|
assert res.returncode == 0, res.stderr
|
|
probe = subprocess.run(
|
|
[sys.executable, "-c",
|
|
"import sys; sys.path.insert(0, {!r}); import publish_event;"
|
|
" print(publish_event.__file__)".format(str(shim.parent))],
|
|
capture_output=True, text=True, env=env, timeout=60,
|
|
)
|
|
assert probe.returncode == 0, probe.stderr
|
|
assert str(shim_trees / "sandbox") in probe.stdout, probe.stdout
|
|
|
|
|
|
# ==============================================================================
|
|
# 5. P1-02 — daemon module, cli router, shell-to-python equivalence
|
|
# ==============================================================================
|
|
|
|
def _delegation_env(mam_sandbox):
|
|
"""Env for running the package out of the sandbox copy."""
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = str(_sandbox_skills(mam_sandbox))
|
|
env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
|
|
return env
|
|
|
|
|
|
def test_daemon_missing_monitor_raises_with_the_legacy_path(mam_sandbox):
|
|
from lib_py.delegation import daemon
|
|
|
|
workdir = Path(mam_sandbox) / "no_monitor"
|
|
workdir.mkdir()
|
|
with pytest.raises(FileNotFoundError) as exc:
|
|
daemon.ensure_subscriber_daemon("job1", str(workdir))
|
|
assert exc.value.args[0] == str(daemon.monitor_script_path(str(workdir)))
|
|
|
|
|
|
def test_daemon_spawn_dedup_and_survives_parent_exit(mam_sandbox):
|
|
"""The daemon must outlive the shell that started it and never double-spawn."""
|
|
from lib_py.delegation import daemon
|
|
|
|
monitor = _install_fake_monitor(mam_sandbox)
|
|
workdir = str(mam_sandbox)
|
|
spawned = []
|
|
try:
|
|
first = daemon.ensure_subscriber_daemon("job1", workdir)
|
|
spawned += first
|
|
assert len(first) == 1 and _alive(first[0])
|
|
assert (Path(workdir) / ".mam" / "multi-agent-mux-monitor.log").exists()
|
|
second = daemon.ensure_subscriber_daemon("job1", workdir)
|
|
assert second == first, "a running daemon must be reused, not replaced"
|
|
# detached into its own session, so a parent's process-group signal misses it
|
|
pgid = os.getpgid(int(first[0]))
|
|
assert pgid != os.getpgid(os.getpid())
|
|
assert daemon.find_running(str(monitor)) == first
|
|
finally:
|
|
_kill(spawned)
|
|
|
|
|
|
def test_cli_ensure_daemon_reproduces_the_legacy_failure_line(mam_sandbox):
|
|
workdir = Path(mam_sandbox) / "no_monitor_cli"
|
|
workdir.mkdir()
|
|
res = subprocess.run(
|
|
[sys.executable, "-m", "lib_py.delegation.cli",
|
|
"ensure-daemon", "--job", "j1", "--workdir", str(workdir)],
|
|
capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
|
|
)
|
|
assert res.returncode == 1
|
|
expected = str(workdir / ".agents" / "skills" / "multi-agent-mux-monitor"
|
|
/ "scripts" / "reconcile.sh")
|
|
assert res.stderr.strip() == "ERROR: monitor script not found: {}".format(expected)
|
|
assert res.stdout == ""
|
|
|
|
|
|
def test_cli_submit_prints_only_the_job_id(mam_sandbox):
|
|
reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
res = subprocess.run(
|
|
[sys.executable, "-m", "lib_py.delegation.cli", "submit",
|
|
"--prompt", "hello", "--agent", "claude-code",
|
|
"--agent-session", "herdr:claude", "--registry-dir", str(reg_dir)],
|
|
capture_output=True, text=True, env=_delegation_env(mam_sandbox), timeout=60,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
job_id = res.stdout.strip()
|
|
assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
|
|
assert res.stdout == job_id + "\n", "stdout must carry the id and nothing else"
|
|
assert (reg_dir / (job_id + ".json")).exists()
|
|
|
|
|
|
@pytest.mark.parametrize("entry", ["script", "cli"])
|
|
def test_publish_failure_semantics_match_across_entry_points(mam_sandbox, entry):
|
|
"""G-1/G-2/G-4 semantics must be identical through the shim and through the cli.
|
|
|
|
Broker points at a non-routable port, so nothing touches the network."""
|
|
reg_dir = Path(mam_sandbox) / ".mam" / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
env = _delegation_env(mam_sandbox)
|
|
job_id = subprocess.run(
|
|
[sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "p",
|
|
"--registry-dir", str(reg_dir)],
|
|
capture_output=True, text=True, env=env, timeout=60,
|
|
).stdout.strip()
|
|
|
|
record = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
|
|
record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False}
|
|
(reg_dir / (job_id + ".json")).write_text(json.dumps(record), encoding="utf-8")
|
|
|
|
if entry == "cli":
|
|
cmd = [sys.executable, "-m", "lib_py.delegation.cli", "publish"]
|
|
else:
|
|
cmd = [sys.executable, str(SHIM_DIR / "publish_event.py")]
|
|
res = subprocess.run(
|
|
cmd + ["--job", job_id, "--event", "completed", "--detail", "done",
|
|
"--attempts", "1", "--registry-dir", str(reg_dir)],
|
|
capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120,
|
|
)
|
|
assert res.returncode == 2, (res.stdout, res.stderr)
|
|
assert res.stdout == "", "publish must stay silent on stdout"
|
|
after = json.loads((reg_dir / (job_id + ".json")).read_text(encoding="utf-8"))
|
|
assert after["status"] == "completed", "status syncs even when the publish fails"
|
|
assert after.get("last_seq") == 1, "seq is consumed on failure by design"
|
|
audit = (Path(mam_sandbox) / ".mam" / "delegate_job_logs" / job_id / "events.ndjson")
|
|
events = [json.loads(l) for l in audit.read_text(encoding="utf-8").splitlines() if l.strip()]
|
|
published = [e for e in events if e.get("event") == "published"]
|
|
assert published and published[-1]["published"] is False
|
|
assert published[-1]["publish_error"]
|
|
|
|
|
|
# ==============================================================================
|
|
# 6. P1-03 — transports, parity and the benchmark harness (no network at all)
|
|
# ==============================================================================
|
|
|
|
def _prepared(topic="python/mqtt/jobs/j/events", host="broker.example", port=1883):
|
|
from lib_py.delegation import mqtt_common, publish_event
|
|
|
|
config = mqtt_common.broker_config_from_env({"host": host, "port": port})
|
|
return publish_event.PreparedEvent(
|
|
job_id="j", event="progress", detail="d", seq=1, topic=topic, retain=False,
|
|
payload={"seq": 1}, body=b'{"seq": 1}', config=config,
|
|
)
|
|
|
|
|
|
def _fake_factory(counters, publishes_ok=True, connect_rc=0):
|
|
from lib_py.delegation.bench import FakeClient, FakeInfo
|
|
|
|
def factory(_role, _config):
|
|
client = FakeClient(connect_latency=0.0, counters=counters)
|
|
if not publishes_ok:
|
|
def unacked(*_a, **_k):
|
|
counters["publish"] = counters.get("publish", 0) + 1
|
|
return FakeInfo(published=False)
|
|
client.publish = unacked
|
|
if connect_rc:
|
|
original = client.connect
|
|
|
|
def connect(host, port, keepalive=60):
|
|
original(host, port, keepalive)
|
|
client.on_connect(client, None, None, connect_rc, None)
|
|
client.connect = connect
|
|
return client
|
|
return factory
|
|
|
|
|
|
def test_persistent_publisher_reuses_one_connection():
|
|
"""The whole point of the prototype: N publishes, 1 connection."""
|
|
from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher
|
|
|
|
prepared = _prepared()
|
|
one_counters, persist_counters = {}, {}
|
|
one = OneShotPublisher(client_factory=_fake_factory(one_counters))
|
|
persist = PersistentPublisher(client_factory=_fake_factory(persist_counters))
|
|
try:
|
|
for _ in range(5):
|
|
assert one.publish_with_retry(prepared, attempts=1) == (True, None)
|
|
assert persist.publish_with_retry(prepared, attempts=1) == (True, None)
|
|
finally:
|
|
persist.close()
|
|
assert one_counters["connect"] == 5
|
|
assert persist_counters["connect"] == 1
|
|
assert one_counters["publish"] == persist_counters["publish"] == 5
|
|
|
|
|
|
def test_transport_parity_of_registry_and_audit_records(mam_sandbox, monkeypatch):
|
|
"""Same input, two transports: registry and audit records must match field for field."""
|
|
from lib_py.delegation import mqtt_common, publish_event, registry
|
|
from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher
|
|
|
|
monkeypatch.chdir(mam_sandbox)
|
|
reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
|
|
os.makedirs(reg_dir, exist_ok=True)
|
|
|
|
results = {}
|
|
for label, publisher in (
|
|
("one_shot", OneShotPublisher(client_factory=_fake_factory({}))),
|
|
("persistent", PersistentPublisher(client_factory=_fake_factory({}))),
|
|
):
|
|
job_id = registry.register_job("parity " + label, registry_dir=reg_dir)
|
|
prepared = publish_event.prepare_event(
|
|
job_id=job_id, event="completed", detail="done", registry_dir=reg_dir,
|
|
)
|
|
ok, err = publisher.publish_with_retry(prepared, attempts=1)
|
|
publish_event.record_outcome(prepared, reg_dir, ok, err)
|
|
if hasattr(publisher, "close"):
|
|
publisher.close()
|
|
record = mqtt_common.load_job(job_id, reg_dir)
|
|
events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
|
|
audit = [e for e in events if e.get("event") == "published"][-1]
|
|
results[label] = {
|
|
"ok": ok,
|
|
"status": record["status"],
|
|
"last_seq": record.get("last_seq"),
|
|
"retain": audit["retain"],
|
|
"published": audit["published"],
|
|
"seq": audit["seq"],
|
|
"has_sig": "hmac_sig" in audit["payload"].get("data", {}),
|
|
"source_event": audit["source_event"],
|
|
}
|
|
|
|
assert results["one_shot"] == results["persistent"], results
|
|
assert results["one_shot"]["retain"] is True, "terminal events must retain"
|
|
assert results["one_shot"]["status"] == "completed"
|
|
|
|
|
|
def test_persistent_publisher_unacked_publish_retries_then_records_failure(mam_sandbox, monkeypatch):
|
|
"""No PUBACK -> exception -> with_retry -> record_outcome(published=False)."""
|
|
from lib_py.delegation import mqtt_common, publish_event, registry
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
monkeypatch.chdir(mam_sandbox)
|
|
reg_dir = str(Path(mam_sandbox) / ".mam" / "jobs")
|
|
os.makedirs(reg_dir, exist_ok=True)
|
|
job_id = registry.register_job("unacked", registry_dir=reg_dir)
|
|
prepared = publish_event.prepare_event(
|
|
job_id=job_id, event="completed", detail="d", registry_dir=reg_dir,
|
|
)
|
|
counters = {}
|
|
publisher = PersistentPublisher(client_factory=_fake_factory(counters, publishes_ok=False))
|
|
try:
|
|
ok, err = publisher.publish_with_retry(prepared, attempts=3)
|
|
finally:
|
|
publisher.close()
|
|
assert ok is False and "not acknowledged" in err
|
|
assert counters["publish"] == 3, "every attempt must reach the transport"
|
|
assert counters["connect"] == 1, "retries reuse the connection"
|
|
|
|
publish_event.record_outcome(prepared, reg_dir, ok, err)
|
|
record = mqtt_common.load_job(job_id, reg_dir)
|
|
assert record["status"] == "completed", "status syncs even when the publish failed"
|
|
assert record.get("last_seq") == prepared.seq
|
|
audit = [e for e in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())
|
|
if e.get("event") == "published"][-1]
|
|
assert audit["published"] is False and audit["publish_error"]
|
|
|
|
|
|
def test_persistent_publisher_refuses_a_second_broker():
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
publisher = PersistentPublisher(client_factory=_fake_factory({}))
|
|
try:
|
|
assert publisher.publish_with_retry(_prepared(host="a.example"), attempts=1)[0] is True
|
|
ok, err = publisher.publish_with_retry(_prepared(host="b.example"), attempts=1)
|
|
assert ok is False
|
|
assert "bound to a.example" in err, err
|
|
finally:
|
|
publisher.close()
|
|
|
|
|
|
def test_bench_fake_transport_schema_and_counts():
|
|
from lib_py.delegation import bench
|
|
|
|
result = bench.run_benchmark(n=7, transport="fake", connect_latency_ms=1.0)
|
|
assert result["transport"] == "fake"
|
|
for arm in ("one_shot", "persistent"):
|
|
assert result[arm]["publishes"] == 7
|
|
assert result[arm]["failures"] == 0
|
|
assert set(result[arm]["latency_ms"]) == {"p50", "p95", "max"}
|
|
assert result["one_shot"]["client_connects"] == 7
|
|
assert result["persistent"]["client_connects"] == 1
|
|
assert result["persistent"]["latency_ms"]["p50"] < result["one_shot"]["latency_ms"]["p50"]
|
|
|
|
|
|
def test_bench_broker_transport_requires_a_job():
|
|
from lib_py.delegation import bench
|
|
|
|
with pytest.raises(ValueError):
|
|
bench.run_benchmark(n=1, transport="broker")
|
|
|
|
|
|
# ==============================================================================
|
|
# 7. Self-review regressions — the connect handshake is per-connection
|
|
#
|
|
# Found reviewing phase 1: PersistentPublisher held the CONNACK result on the
|
|
# instance, so a refused attempt left its reason code behind for the next one.
|
|
# The retry then either rejected a healthy connection or, worse, accepted one
|
|
# whose CONNACK had never arrived. Both need an asynchronous callback to
|
|
# reproduce, which is how real paho delivers it.
|
|
# ==============================================================================
|
|
|
|
class _AsyncAckClient(object):
|
|
"""paho-shaped stub: connect() returns at once, CONNACK lands on a timer."""
|
|
|
|
def __init__(self, rc_sequence, calls, ack_delay=0.05):
|
|
self._rc_sequence = rc_sequence
|
|
self._calls = calls
|
|
self._ack_delay = ack_delay
|
|
self.on_connect = None
|
|
|
|
def connect(self, host, port, keepalive=60):
|
|
import threading
|
|
idx = self._calls["n"]
|
|
self._calls["n"] += 1
|
|
rc = self._rc_sequence[min(idx, len(self._rc_sequence) - 1)]
|
|
if rc is None:
|
|
return # models a broker that never acknowledges
|
|
timer = threading.Timer(self._ack_delay,
|
|
lambda: self.on_connect(self, None, None, rc, None))
|
|
timer.daemon = True
|
|
timer.start()
|
|
|
|
def loop_start(self):
|
|
return None
|
|
|
|
def loop_stop(self):
|
|
self._calls["loop_stop"] = self._calls.get("loop_stop", 0) + 1
|
|
|
|
def reconnect_delay_set(self, **_kw):
|
|
return None
|
|
|
|
def publish(self, *_a, **_k):
|
|
from lib_py.delegation.bench import FakeInfo
|
|
self._calls["publish"] = self._calls.get("publish", 0) + 1
|
|
return FakeInfo()
|
|
|
|
def disconnect(self):
|
|
self._calls["disconnect"] = self._calls.get("disconnect", 0) + 1
|
|
|
|
|
|
def _async_factory(rc_sequence, calls):
|
|
def factory(_role, _config):
|
|
return _AsyncAckClient(rc_sequence, calls)
|
|
return factory
|
|
|
|
|
|
def test_persistent_publisher_recovers_on_the_attempt_after_a_refusal():
|
|
"""A refused connection must not decide the next attempt's outcome."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
calls = {"n": 0}
|
|
publisher = PersistentPublisher(client_factory=_async_factory([5, 0], calls))
|
|
try:
|
|
ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
|
|
finally:
|
|
publisher.close()
|
|
assert ok is True, err
|
|
assert calls["n"] == 2, (
|
|
"attempt 2 had a healthy connection and should have been the last one; "
|
|
"got {} connect() calls".format(calls["n"])
|
|
)
|
|
assert calls.get("disconnect", 0) >= 1, "the refused client must be torn down, not leaked"
|
|
|
|
|
|
def test_persistent_publisher_never_accepts_a_connection_that_did_not_ack():
|
|
"""A silent broker must time out even when an earlier attempt acked."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
import lib_py.delegation.publisher as publisher_mod
|
|
|
|
calls = {"n": 0}
|
|
# first connection is refused (leaves a reason code behind), second never acks
|
|
publisher = PersistentPublisher(client_factory=_async_factory([5, None], calls))
|
|
original = publisher_mod.CONNECT_ACK_TIMEOUT
|
|
publisher_mod.CONNECT_ACK_TIMEOUT = 0.3 # keep the test quick
|
|
try:
|
|
ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
|
|
finally:
|
|
publisher_mod.CONNECT_ACK_TIMEOUT = original
|
|
publisher.close()
|
|
assert ok is False
|
|
assert "no CONNACK" in err, (
|
|
"the silent second connection must time out on its own evidence, not "
|
|
"inherit the first attempt's reason code; got: {}".format(err)
|
|
)
|
|
|
|
|
|
def test_persistent_publisher_reuses_a_healthy_connection_across_events():
|
|
"""The successful path still connects exactly once."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
calls = {"n": 0}
|
|
publisher = PersistentPublisher(client_factory=_async_factory([0], calls))
|
|
try:
|
|
for _ in range(4):
|
|
assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
|
|
finally:
|
|
publisher.close()
|
|
assert calls["n"] == 1, "a healthy connection must be reused"
|
|
assert calls["publish"] == 4
|
|
|
|
|
|
# ==============================================================================
|
|
# 8. Cross-review regressions (job 60c980ac)
|
|
#
|
|
# R1 (codex): `python -m` puts the working directory ahead of PYTHONPATH, so a
|
|
# frozen loop whose cwd contained a lib_py package silently ran live
|
|
# code. The shell helpers now use an explicit bootstrap instead.
|
|
# R2 (codex): the persistent connection was reused on host/port alone, ignoring
|
|
# credentials and TLS.
|
|
# F-1 (claude): paho queues a publish issued while disconnected and then raises
|
|
# RuntimeError, which sat outside the retry tuple -- so the one state
|
|
# a persistent connection exists to survive failed on first attempt.
|
|
# ==============================================================================
|
|
|
|
def _decoy_tree(root: Path) -> Path:
|
|
"""A directory that looks like a skills tree and screams if it is imported."""
|
|
pkg = root / "lib_py" / "delegation"
|
|
pkg.mkdir(parents=True)
|
|
(root / "lib_py" / "__init__.py").write_text("", encoding="utf-8")
|
|
(pkg / "__init__.py").write_text("", encoding="utf-8")
|
|
(pkg / "cli.py").write_text(
|
|
"def main(argv=None):\n"
|
|
" print('DECOY-TREE-EXECUTED')\n"
|
|
" return 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
return root
|
|
|
|
|
|
def test_r1_shell_helper_ignores_a_lib_py_in_the_working_directory(mam_sandbox, tmp_path):
|
|
"""R1: the helper must run its own tree even when cwd provides lib_py.
|
|
|
|
This is the frozen-loop shape: a decoy package sitting in the working
|
|
directory must not win over the tree the sourced lib.sh belongs to.
|
|
"""
|
|
decoy = _decoy_tree(tmp_path / "decoy_ws")
|
|
(decoy / ".mam" / "jobs").mkdir(parents=True)
|
|
|
|
lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh"
|
|
env = dict(os.environ)
|
|
env.pop("HERDR_SESSION_NAME", None)
|
|
env.pop("PYTHONPATH", None)
|
|
env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent")
|
|
|
|
res = subprocess.run(
|
|
["bash", "-c",
|
|
'cd "{}" && source "{}" >/dev/null 2>&1 && '
|
|
'delegate_submit_job "r1 probe" claude-code herdr:claude'.format(decoy, lib_sh)],
|
|
capture_output=True, text=True, env=env, timeout=120,
|
|
)
|
|
assert "DECOY-TREE-EXECUTED" not in res.stdout, (
|
|
"the working directory's lib_py shadowed the helper's own tree:\n" + res.stdout
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
job_id = res.stdout.strip()
|
|
assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout)
|
|
assert (decoy / ".mam" / "jobs" / (job_id + ".json")).exists(), (
|
|
"the real cli should have registered the job in the cwd registry"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("field,first,second", [
|
|
("username", "account-a", "account-b"),
|
|
("password", "secret-a", "secret-b"),
|
|
("tls", False, True),
|
|
("ca_certs", None, "/tmp/ca.pem"),
|
|
])
|
|
def test_r2_persistent_publisher_refuses_a_different_connection_identity(field, first, second):
|
|
"""R2: same endpoint is not the same connection when auth or TLS differ."""
|
|
from lib_py.delegation import mqtt_common, publish_event
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
built = []
|
|
|
|
def factory(_role, config):
|
|
built.append(getattr(config, field))
|
|
from lib_py.delegation.bench import FakeClient
|
|
client = FakeClient(connect_latency=0.0, counters={})
|
|
return client
|
|
|
|
def prepared(value):
|
|
config = mqtt_common.broker_config_from_env(
|
|
{"host": "same.example", "port": 1883, field: value})
|
|
return publish_event.PreparedEvent(
|
|
"j", "progress", "d", 1, "t", False, {}, b"{}", config)
|
|
|
|
publisher = PersistentPublisher(client_factory=factory)
|
|
try:
|
|
ok, _err = publisher.publish_with_retry(prepared(first), attempts=1)
|
|
assert ok is True
|
|
ok2, err2 = publisher.publish_with_retry(prepared(second), attempts=1)
|
|
finally:
|
|
publisher.close()
|
|
|
|
assert ok2 is False, "an event with a different {} must not reuse the connection".format(field)
|
|
assert field in err2, err2
|
|
assert str(second) not in err2, "the error must name the field, never its value"
|
|
assert built == [first], "the second event must not have been published on the first client"
|
|
|
|
|
|
def test_r2_identical_configuration_still_reuses_the_connection():
|
|
"""The guard must not become so strict that reuse never happens."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
counters = {"n": 0}
|
|
|
|
def factory(_role, _config):
|
|
counters["n"] += 1
|
|
from lib_py.delegation.bench import FakeClient
|
|
return FakeClient(connect_latency=0.0, counters={})
|
|
|
|
publisher = PersistentPublisher(client_factory=factory)
|
|
try:
|
|
for _ in range(3):
|
|
assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None)
|
|
finally:
|
|
publisher.close()
|
|
assert counters["n"] == 1
|
|
|
|
|
|
class _FlakyConnectionClient(object):
|
|
"""paho-shaped: the first N publishes are issued while disconnected."""
|
|
|
|
def __init__(self, no_conn_times, calls):
|
|
import paho.mqtt.client as mqtt
|
|
self._mqtt = mqtt
|
|
self._no_conn_times = no_conn_times
|
|
self._calls = calls
|
|
self.on_connect = None
|
|
|
|
def connect(self, host, port, keepalive=60):
|
|
self._calls["connect"] = self._calls.get("connect", 0) + 1
|
|
self.on_connect(self, None, None, 0, None)
|
|
|
|
def loop_start(self):
|
|
return None
|
|
|
|
def loop_stop(self):
|
|
return None
|
|
|
|
def reconnect_delay_set(self, **_kw):
|
|
return None
|
|
|
|
def publish(self, *_a, **_k):
|
|
from lib_py.delegation.bench import FakeInfo
|
|
n = self._calls.get("publish", 0)
|
|
self._calls["publish"] = n + 1
|
|
if n < self._no_conn_times:
|
|
return FakeInfo(published=False, rc=int(self._mqtt.MQTT_ERR_NO_CONN))
|
|
return FakeInfo(published=True, rc=0)
|
|
|
|
def disconnect(self):
|
|
return None
|
|
|
|
|
|
def test_f1_publish_issued_while_disconnected_is_retried_until_reconnect():
|
|
"""F-1: the gap between a drop and paho's reconnect must be survivable.
|
|
|
|
paho queues such a publish and raises RuntimeError from wait_for_publish,
|
|
which is not in the retry tuple; it used to end the attempt immediately."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
calls = {}
|
|
publisher = PersistentPublisher(
|
|
client_factory=lambda _r, _c: _FlakyConnectionClient(2, calls))
|
|
try:
|
|
ok, err = publisher.publish_with_retry(_prepared(), attempts=3)
|
|
finally:
|
|
publisher.close()
|
|
assert ok is True, "should have recovered on the third attempt, got: {}".format(err)
|
|
assert calls["publish"] == 3, "every attempt must reach the transport"
|
|
assert calls["connect"] == 1, "recovery must reuse the connection, not rebuild it"
|
|
|
|
|
|
def test_f1_disconnected_publish_still_fails_when_it_never_recovers():
|
|
"""Retrying must not mask a genuinely dead connection."""
|
|
from lib_py.delegation.publisher import PersistentPublisher
|
|
|
|
calls = {}
|
|
publisher = PersistentPublisher(
|
|
client_factory=lambda _r, _c: _FlakyConnectionClient(99, calls))
|
|
try:
|
|
ok, err = publisher.publish_with_retry(_prepared(), attempts=2)
|
|
finally:
|
|
publisher.close()
|
|
assert ok is False
|
|
assert "not connected" in err, err
|
|
assert calls["publish"] == 2
|
|
|
|
|
|
def test_one_shot_publish_has_a_single_implementation():
|
|
"""O-1: the CLI path and the service path must not drift apart again."""
|
|
from lib_py.delegation import publish_event
|
|
from lib_py.delegation.publisher import OneShotPublisher
|
|
|
|
source = Path(publish_event.__file__).read_text(encoding="utf-8")
|
|
assert "OneShotPublisher().publish_raw(" in source, (
|
|
"_publish_once must delegate to the transport module"
|
|
)
|
|
for leaked in ("client.loop_start()", "no CONNACK from broker", "wait_for_publish("):
|
|
assert leaked not in source, (
|
|
"publish_event still carries its own copy of the transport logic: {}".format(leaked)
|
|
)
|
|
assert publish_event.CONNECT_ACK_TIMEOUT is not None
|
|
assert hasattr(OneShotPublisher, "publish_raw")
|