#!/usr/bin/env python3 """Black-box regressions for the local PI subagent executor (no paid model calls).""" from __future__ import annotations import fcntl import hashlib import json import os import pathlib import selectors import shutil import socket import subprocess import sys import tempfile import threading import time import unittest ROOT = pathlib.Path(__file__).resolve().parents[2] SOURCE_AGENTCTL = ROOT / "scripts" / "agentctl.py" SOURCE_W4 = ROOT / "scripts" / "agentctl_w4.py" SOURCE_EXEC = ROOT / "scripts" / "pi_subagents.py" SOURCE_MCP = ROOT / "scripts" / "mcp" / "pi-subagents-mcp.mjs" SOURCE_EXT = ROOT / "scripts" / "pi-subagent-readonly.ts" POLICY_SOURCE = ROOT / "configs" / "agent-governance.json" MODEL_CONFIG_SOURCE = ROOT / ".doc" / "ops" / "2026-09-24-subagent-model-config.md" KNOWN_ROLES = ( "research", "planning", "requirements_review", "plan_review", "code_review", "diagnosis", "test_run", "acceptance_check", "stage_notes", "report", "auxiliary", ) KNOWN_ALIASES = ("qwen_max", "qwen_flash", "glm_full", "glm_flash", "minimax_aux") def sha256_text(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest() FAKE_PI = r'''#!/usr/bin/env python3 import json, os, sys, time, pathlib argv_log = os.environ.get("FAKE_PI_ARGV_LOG") if argv_log: pathlib.Path(argv_log).write_text(json.dumps(sys.argv[1:]), encoding="utf-8") snap_probe = os.environ.get("FAKE_PI_SNAP_PROBE") if snap_probe: cwd = pathlib.Path(os.getcwd()).resolve() snap_env = os.environ.get("PI_SUBAGENT_SNAPSHOT_ROOT") or str(cwd) snap_root = pathlib.Path(snap_env).resolve() inventory = [] for root, dirs, files in os.walk(snap_root): dirs[:] = sorted(d for d in dirs if d != ".pi-subagent-snapshot") rel = pathlib.Path(root).relative_to(snap_root) for name in sorted(files): if name == ".pi-subagent-snapshot": continue path = (rel / name).as_posix() if str(rel) != "." else name inventory.append(path) ext_arg = None args = sys.argv[1:] for i, a in enumerate(args): if a in ("-e", "--extension") and i + 1 < len(args): ext_arg = args[i + 1] break ext_exists = bool(ext_arg and pathlib.Path(ext_arg).is_file()) inside = False if ext_arg: try: pathlib.Path(ext_arg).resolve().relative_to(snap_root) inside = True except ValueError: inside = False pathlib.Path(snap_probe).write_text(json.dumps({ "cwd": str(cwd), "snapshot_root": str(snap_root), "inventory": inventory, "extension": ext_arg, "extension_exists": ext_exists, "extension_inside_snapshot": inside, }), encoding="utf-8") mode = os.environ.get("FAKE_PI_MODE", "ok") provider = os.environ.get("FAKE_PI_PROVIDER", "qwen-token-plan-cn") model = os.environ.get("FAKE_PI_MODEL", "qwen3.8-flash") text = os.environ.get("FAKE_PI_TEXT", "review complete: looks fine") sleep_s = float(os.environ.get("FAKE_PI_SLEEP", "0")) if sleep_s: time.sleep(sleep_s) if mode == "hang": time.sleep(3600) if mode == "crash": sys.exit(42) events = [{"type": "session", "version": 3, "id": "fake", "cwd": os.getcwd()}] if mode != "no_agent_start": events.append({"type": "agent_start"}) events.append({"type": "turn_start"}) msg = { "role": "assistant", "content": [{"type": "text", "text": text}], "stopReason": ( "error" if mode == "stop_error" else ("aborted" if mode == "aborted" else ("toolUse" if mode == "tool_use_stop" else "stop")) ), "provider": provider, "model": model, } if mode == "empty_text": msg["content"] = [{"type": "text", "text": ""}] if mode == "thinking_only": msg["content"] = [{"type": "thinking", "thinking": "secret thoughts"}] if mode != "no_message_end": events.append({"type": "message_end", "message": msg}) events.append({"type": "turn_end", "message": msg, "toolResults": []}) if mode not in ("no_agent_end", "crash_mid"): will_retry = mode == "will_retry" events.append({"type": "agent_end", "messages": [msg], "willRetry": will_retry}) if mode not in ("no_settled",) and not will_retry: events.append({"type": "agent_settled"}) if mode == "activity_after_settled": events.append({"type": "agent_start"}) for ev in events: print(json.dumps(ev), flush=True) if mode == "nonzero": sys.exit(7) sys.exit(0) ''' BASE_CONFIG = MODEL_CONFIG_SOURCE.read_text(encoding="utf-8") if MODEL_CONFIG_SOURCE.exists() else "" class PiSubagentsTests(unittest.TestCase): def setUp(self): self.assertTrue(SOURCE_AGENTCTL.exists(), "agentctl.py must exist") self.temp = tempfile.TemporaryDirectory(prefix="pi-subagent-test-") self.addCleanup(self.temp.cleanup) self.root = pathlib.Path(self.temp.name) / "repo" self.root.mkdir() self.git("init", "-b", "dev") self.git("config", "user.name", "PI Subagent Test") self.git("config", "user.email", "pi-subagent-test@example.invalid") (self.root / "scripts" / "mcp").mkdir(parents=True) (self.root / ".cursor").mkdir(parents=True) shutil.copy2(SOURCE_AGENTCTL, self.root / "scripts" / "agentctl.py") if SOURCE_W4.exists(): shutil.copy2(SOURCE_W4, self.root / "scripts" / "agentctl_w4.py") catalog_src = ROOT / "scripts" / "agentctl_catalog.py" if catalog_src.exists(): shutil.copy2(catalog_src, self.root / "scripts" / "agentctl_catalog.py") vis_src = ROOT / "scripts" / "agentctl_visibility.py" if vis_src.exists(): shutil.copy2(vis_src, self.root / "scripts" / "agentctl_visibility.py") cons_src = ROOT / "scripts" / "agentctl_console.py" if cons_src.exists(): shutil.copy2(cons_src, self.root / "scripts" / "agentctl_console.py") if SOURCE_EXEC.exists(): shutil.copy2(SOURCE_EXEC, self.root / "scripts" / "pi_subagents.py") if SOURCE_MCP.exists(): shutil.copy2(SOURCE_MCP, self.root / "scripts" / "mcp" / "pi-subagents-mcp.mjs") if SOURCE_EXT.exists(): shutil.copy2(SOURCE_EXT, self.root / "scripts" / "pi-subagent-readonly.ts") dest = self.root / "configs" dest.mkdir(parents=True, exist_ok=True) shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json") (self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8") for directory in ("src", ".doc/ops", ".doc/status", ".doc/test-reports/assets"): (self.root / directory).mkdir(parents=True, exist_ok=True) (self.root / "src" / "sample.py").write_text("print('hello')\n", encoding="utf-8") (self.root / ".doc" / "ops" / "CURRENT.md").write_text( "# Coordination\n\n| Task | Owner | Branch | State |\n| --- | --- | --- | --- |\n" "| alpha | Alice | feature/alpha | active |\n", encoding="utf-8") (self.root / ".doc" / "status" / "FEATURE-MATRIX.md").write_text( "# Features\n\n| record_id | 模块 | 状态 |\n| --- | --- | --- |\n| core | core | planned |\n", encoding="utf-8") self.config_path = self.root / ".doc" / "ops" / "2026-09-24-subagent-model-config.md" self.config_path.write_text(BASE_CONFIG or self._minimal_config(), encoding="utf-8") self.fake_pi = pathlib.Path(self.temp.name) / "fake-pi" self.fake_pi.write_text(FAKE_PI, encoding="utf-8") self.fake_pi.chmod(0o755) self.argv_log = pathlib.Path(self.temp.name) / "pi-argv.json" self.git("add", ".") self.git("commit", "-m", "fixture") self.source_sha = self.git("rev-parse", "HEAD") def _minimal_config(self): return """# Subagent 模型分配配置 config_revision：`subagent-models-2026-09-24-r1` ## 2. 允许的模型池 | alias | provider | model（精确 ID） | 使用限制 | |---|---|---|---| | qwen_max | qwen-token-plan-cn | qwen3.8-max | ok | | qwen_flash | qwen-token-plan-cn | qwen3.8-flash | ok | | glm_full | zai-coding-cn | glm-5.3 | ok | | glm_flash | zai-coding-cn | glm-5.3-flash | ok | | minimax_aux | minimax-cn | MiniMax-M3 | aux only | ## 3. 角色路由表 | role_id | 工作 | enabled | 默认模型 | 升级角色 | 执行边界 | |---|---|---|---|---|---| | research | r | true | qwen_flash | planning | ro | | planning | p | true | qwen_max | none | ro | | requirements_review | rr | true | glm_full | none | ro | | plan_review | pr | true | glm_full | none | ro | | code_review | cr | true | glm_full | none | ro | | diagnosis | d | true | glm_full | none | ro | | test_run | t | true | glm_flash | diagnosis | test | | acceptance_check | a | true | glm_flash | diagnosis | check | | stage_notes | s | true | qwen_flash | report | notes | | report | rp | true | qwen_max | none | report | | auxiliary | aux | false | minimax_aux | none | aux | ## 4. 执行与升级参数 | 参数 | 默认值 | 语义 | |---|---|---| | max_parallel_pi_tasks | 2 | limit | | implementation_executor | cursor_auto | cursor only | | unavailable_action | pause_role | pause | | automatic_fallback | none | no fallback | | reload_boundary | next_dispatch | next | | recursive_delegation | false | no recurse | """ def git(self, *args, cwd=None): return subprocess.run( ["git", *args], cwd=cwd or self.root, text=True, capture_output=True, check=True ).stdout.strip() def env(self, extra=None): base = { **os.environ, "PI_SUBAGENT_PI_BIN": str(self.fake_pi), "FAKE_PI_ARGV_LOG": str(self.argv_log), "FAKE_PI_MODE": "ok", "FAKE_PI_PROVIDER": "qwen-token-plan-cn", "FAKE_PI_MODEL": "qwen3.8-flash", "PYTHONDONTWRITEBYTECODE": "1", } # Drop inherited auth/runtime pointers; tests inject only what they need. for key in list(base): if key.startswith("AGENTCTL_") or key.startswith("PI_CODING_") or key == "XDG_RUNTIME_DIR": base.pop(key, None) if extra: base.update(extra) for key in list(base): if key.startswith("AGENTCTL_"): base.pop(key, None) return base def cli(self, *args, cwd=None, check=True, env=None, stdin=None): script = self.root / "scripts" / "pi_subagents.py" result = subprocess.run( [sys.executable, "-B", str(script), *args], cwd=cwd or self.root, text=True, capture_output=True, env=env or self.env(), input=stdin, timeout=60, ) if check: self.assertEqual(result.returncode, 0, result.stdout + "\n" + result.stderr) return result def agentctl(self, *args, cwd=None, check=True, env=None): result = subprocess.run( [sys.executable, "-B", str(self.root / "scripts" / "agentctl.py"), *args], cwd=cwd or self.root, text=True, capture_output=True, env=env, timeout=30, ) if check: self.assertEqual(result.returncode, 0, result.stdout + result.stderr) return result def initialized_task(self, task="alpha", scopes=None): scopes = scopes or ( "src", ".doc", ".doc/test-reports/assets/pi-subagents", ) if not (self.root / ".locks").exists(): self.agentctl("init") path = self.root / ".worktrees" / task self.git("worktree", "add", "-b", "feature/" + task, str(path), "dev") args = ["task", "register", "--id", task, "--owner", "Test Agent"] for scope in scopes: args.extend(("--scope", scope)) self.agentctl(*args, cwd=path) return path def last_json(self, text): lines = [ln for ln in text.splitlines() if ln.strip()] self.assertTrue(lines, text) return json.loads(lines[-1]) def result_json(self, result): """Prefer executor JSON (code/run_status) over agentctl status lines on stderr.""" candidates = [] for blob in (result.stdout, result.stderr): for ln in (blob or "").splitlines(): ln = ln.strip() if not ln: continue try: obj = json.loads(ln) except Exception: continue if isinstance(obj, dict): candidates.append(obj) for obj in reversed(candidates): if "code" in obj or "run_status" in obj or "roles" in obj or "models" in obj: return obj self.assertTrue(candidates, result.stdout + "\n" + result.stderr) return candidates[-1] def report_dir(self, task, request_id, worktree=None): base = worktree or self.root return base / ".doc" / "test-reports" / "assets" / "pi-subagents" / task / request_id def base_request(self, **overrides): req = { "task_id": "alpha", "role_id": "research", "request_id": "req-001", "goal": "Locate sample.py and summarize", "authorization_ref": "user:test-auth", "source_sha": self.source_sha, "input_files": ["src/sample.py"], "timeout_seconds": 30, } req.update(overrides) return req def run_under_lease(self, task_path, request, resources=None, env_extra=None, check=True): resources = resources or [ f"path:.doc/test-reports/assets/pi-subagents/{request['task_id']}/{request['request_id']}" ] req_path = pathlib.Path(self.temp.name) / f"{request['request_id']}.json" req_path.write_text(json.dumps(request), encoding="utf-8") env = self.env(env_extra) wt_config = task_path / ".doc/ops/2026-09-24-subagent-model-config.md" argv = ["with", "--task", request["task_id"], "--wait", "0"] for res in resources: argv.extend(("--resource", res)) argv.extend(( "--", sys.executable, "-B", str(task_path / "scripts" / "pi_subagents.py"), "run", "--request", str(req_path), "--config", str(wt_config), )) result = self.agentctl(*argv, cwd=task_path, check=False, env=env) if check: self.assertEqual(result.returncode, 0, result.stdout + "\n" + result.stderr) return result # --- presence / import --- def test_00_executor_module_present(self): self.assertTrue(SOURCE_EXEC.exists(), "scripts/pi_subagents.py must be implemented") self.assertTrue(SOURCE_MCP.exists(), "scripts/mcp/pi-subagents-mcp.mjs must be implemented") self.assertTrue(SOURCE_EXT.exists(), "scripts/pi-subagent-readonly.ts must be implemented") # --- config parsing --- def test_config_catalog_parses_known_pool_and_roles(self): task = self.initialized_task() result = self.cli("roles", "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task) payload = self.last_json(result.stdout) self.assertEqual(payload["config_revision"], "subagent-models-2026-09-24-r1") self.assertEqual(set(payload["models"].keys()), set(KNOWN_ALIASES)) self.assertEqual(set(payload["roles"].keys()), set(KNOWN_ROLES)) self.assertFalse(payload["roles"]["auxiliary"]["enabled"]) self.assertEqual(payload["params"]["max_parallel_pi_tasks"], 2) self.assertEqual(payload["params"]["implementation_executor"], "cursor_auto") self.assertEqual(payload["params"]["automatic_fallback"], "none") self.assertIn("config_sha256", payload) self.assertEqual(payload["config_sha256"], sha256_text( (task / ".doc/ops/2026-09-24-subagent-model-config.md").read_text(encoding="utf-8"))) def test_roles_does_not_invoke_pi(self): task = self.initialized_task() if self.argv_log.exists(): self.argv_log.unlink() self.cli("roles", "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task) self.assertFalse(self.argv_log.exists(), "roles must not spawn PI") def test_config_rejects_unknown_alias_and_cycle(self): task = self.initialized_task() bad = (task / ".doc/ops/2026-09-24-subagent-model-config.md").read_text(encoding="utf-8") bad = bad.replace("| research | 检索、代码定位、资料摘录、需求初稿 | true | qwen_flash | planning |", "| research | 检索、代码定位、资料摘录、需求初稿 | true | not_a_model | planning |") if "not_a_model" not in bad: bad = bad.replace("| research | r | true | qwen_flash | planning | ro |", "| research | r | true | not_a_model | planning | ro |") (task / ".doc/ops/2026-09-24-subagent-model-config.md").write_text(bad, encoding="utf-8") result = self.cli("config", "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task, check=False) self.assertNotEqual(result.returncode, 0) err = self.result_json(result) self.assertEqual(err["code"], "UNKNOWN_ALIAS") cyclic = BASE_CONFIG or self._minimal_config() cyclic = cyclic.replace("| planning | 需求分析、方案制定、复杂问题综合分析 | true | qwen_max | none |", "| planning | 需求分析、方案制定、复杂问题综合分析 | true | qwen_max | research |") if "| planning | p | true | qwen_max | none | ro |" in cyclic: cyclic = cyclic.replace("| planning | p | true | qwen_max | none | ro |", "| planning | p | true | qwen_max | research | ro |") (task / ".doc/ops/2026-09-24-subagent-model-config.md").write_text(cyclic, encoding="utf-8") result = self.cli("config", "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task, check=False) self.assertNotEqual(result.returncode, 0) err = self.result_json(result) self.assertEqual(err["code"], "UPGRADE_CYCLE") def test_disabled_auxiliary_and_m3_critical_remap_denied(self): task = self.initialized_task() # auxiliary disabled by default req = self.base_request(role_id="auxiliary", request_id="aux-1") result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertEqual(err["code"], "ROLE_DISABLED") # remap code_review default model to minimax_aux text = (task / ".doc/ops/2026-09-24-subagent-model-config.md").read_text(encoding="utf-8") text = text.replace("| code_review | 代码审查与回归风险检查 | true | glm_full | none |", "| code_review | 代码审查与回归风险检查 | true | minimax_aux | none |") if "minimax_aux | none |" not in text.split("code_review")[1][:200]: text = text.replace("| code_review | cr | true | glm_full | none | ro |", "| code_review | cr | true | minimax_aux | none | ro |") (task / ".doc/ops/2026-09-24-subagent-model-config.md").write_text(text, encoding="utf-8") req = self.base_request(role_id="code_review", request_id="m3-deny", goal="review", ) result = self.run_under_lease(task, req, check=False, env_extra={"FAKE_PI_PROVIDER": "minimax-cn", "FAKE_PI_MODEL": "MiniMax-M3"}) err = self.result_json(result) self.assertIn(err["code"], ("ROLE_MODEL_FORBIDDEN", "MINIMAX_ROLE_FORBIDDEN")) # --- lease / naked refusal --- def test_naked_run_without_lease_refused(self): task = self.initialized_task() req = self.base_request() req_path = pathlib.Path(self.temp.name) / "naked.json" req_path.write_text(json.dumps(req), encoding="utf-8") result = self.cli( "run", "--request", str(req_path), "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task, check=False) self.assertNotEqual(result.returncode, 0) err = self.result_json(result) self.assertEqual(err["code"], "LEASE_REQUIRED") def test_forged_lease_env_refused(self): task = self.initialized_task() req = self.base_request(request_id="forge-1") req_path = pathlib.Path(self.temp.name) / "forge.json" req_path.write_text(json.dumps(req), encoding="utf-8") forged = self.env({ "AGENTCTL_RUN_ID": "not-a-real-run", "AGENTCTL_TASK_ID": "alpha", "AGENTCTL_GENERATION": "1", "AGENTCTL_EPOCH": "fake-epoch", "AGENTCTL_FENCE": "0", }) result = self.cli( "run", "--request", str(req_path), "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task, check=False, env=forged) self.assertNotEqual(result.returncode, 0) err = self.result_json(result) self.assertIn(err["code"], ("LEASE_REQUIRED", "STALE_GENERATION")) def test_scope_refusal_outside_report_path(self): task = self.initialized_task(scopes=("src",)) # no report scope req = self.base_request(request_id="noscope") result = self.run_under_lease( task, req, resources=["path:.doc/test-reports/assets/pi-subagents/alpha/noscope"], check=False) self.assertNotEqual(result.returncode, 0) # --- input validation --- def test_input_path_traversal_symlink_secret_refusal(self): task = self.initialized_task() for bad, code in ( (["../etc/passwd"], "PATH_OUT_OF_SCOPE"), ([".env"], "PATH_OUT_OF_SCOPE"), (["src/../.git/config"], "PATH_OUT_OF_SCOPE"), ): req = self.base_request(request_id="bad-" + code.lower() + str(hash(tuple(bad)) % 1000), input_files=bad) result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertEqual(err["code"], code, err) # symlink refusal link = task / "src" / "link.py" try: link.symlink_to(task / "src" / "sample.py") except OSError: self.skipTest("symlink unsupported") self.git("add", "src/link.py", cwd=task) # uncommitted symlink should be refused; commit it then refuse as symlink try: self.git("commit", "-m", "symlink", cwd=task) except subprocess.CalledProcessError: pass sha = self.git("rev-parse", "HEAD", cwd=task) req = self.base_request(request_id="symlink-1", input_files=["src/link.py"], source_sha=sha) result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertIn(err["code"], ("SYMLINK_REFUSED", "PATH_OUT_OF_SCOPE", "INPUT_NOT_REGULAR")) # --- successful run / argv flags / receipts --- def test_run_emits_exact_pi_flags_and_receipt(self): task = self.initialized_task() req = self.base_request(request_id="ok-1") result = self.run_under_lease(task, req) payload = self.last_json(result.stdout) self.assertEqual(payload["run_status"], "succeeded") # Non-test roles: model completion is pending review, not auto pass. self.assertEqual(payload["verdict"], "not_applicable") self.assertEqual(payload.get("review_state"), "pending") self.assertNotEqual(payload["verdict"], payload["run_status"]) argv = json.loads(self.argv_log.read_text(encoding="utf-8")) joined = " ".join(argv) for flag in ( "--no-session", "--mode", "json", "--print", "--offline", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-context-files", "--no-approve", "--no-extensions", "--provider", "qwen-token-plan-cn", "--model", "qwen3.8-flash", "--tools", ): self.assertIn(flag, argv if flag.startswith("--") and flag != "--tools" else joined) self.assertIn("read,grep,find,ls", joined.replace(" ", "")) self.assertTrue(any(a == "-e" or a == "--extension" for a in argv)) rdir = self.report_dir("alpha", "ok-1", task) self.assertTrue((rdir / "receipt.json").is_file()) self.assertTrue((rdir / "report.md").is_file()) receipt = json.loads((rdir / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(receipt["configured_provider"], "qwen-token-plan-cn") self.assertEqual(receipt["configured_model"], "qwen3.8-flash") self.assertEqual(receipt["source_sha"], self.source_sha) self.assertIn("config_sha256", receipt) self.assertIn("input_files", receipt) mode = oct((rdir / "receipt.json").stat().st_mode)[-3:] self.assertIn(mode, ("600", "640", "644")) # prefer 600; accept umask variance in CI def test_extension_outside_model_snapshot_inventory(self): """Model snapshot must contain only declared inputs; -e guard lives outside it.""" task = self.initialized_task() probe = pathlib.Path(self.temp.name) / "snap-probe.json" req = self.base_request(request_id="snap-clean-1") self.run_under_lease( task, req, env_extra={"FAKE_PI_SNAP_PROBE": str(probe)}) self.assertTrue(probe.is_file(), "fake PI must record snapshot probe") data = json.loads(probe.read_text(encoding="utf-8")) declared = list(req["input_files"]) self.assertEqual(sorted(data["inventory"]), sorted(declared), data) self.assertFalse( any(p == "pi-subagent-readonly.ts" or p.endswith("/pi-subagent-readonly.ts") or p.startswith(".pi-extensions/") for p in data["inventory"]), data) self.assertTrue(data.get("extension"), data) self.assertTrue(data.get("extension_exists"), "extension must exist on disk during PI run") self.assertFalse(data.get("extension_inside_snapshot"), data) snap = pathlib.Path(data["snapshot_root"]).resolve() self.assertEqual(pathlib.Path(data["cwd"]).resolve(), snap, data) ext = pathlib.Path(data["extension"]).resolve() try: ext.relative_to(snap) self.fail(f"extension unexpectedly under snapshot: {ext}") except ValueError: pass def test_model_mismatch_fails(self): task = self.initialized_task() req = self.base_request(request_id="mm-1") result = self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_PROVIDER": "other", "FAKE_PI_MODEL": "wrong"}) # may succeed process but receipt run_status failed rdir = self.report_dir("alpha", "mm-1", task) if (rdir / "receipt.json").exists(): receipt = json.loads((rdir / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(receipt["run_status"], "failed") self.assertNotEqual(receipt.get("verdict"), "pass") else: err = self.result_json(result) self.assertEqual(err["code"], "MODEL_MISMATCH") def test_malformed_and_empty_outputs_fail(self): task = self.initialized_task() for mode, rid in (("empty_text", "empty-1"), ("no_agent_end", "noend-1"), ("stop_error", "sterr-1"), ("thinking_only", "think-1"), ("nonzero", "nz-1"), ("will_retry", "retry-1"), ("tool_use_stop", "toolstop-1"), ("no_settled", "nosettle-1"), ("activity_after_settled", "after-1")): req = self.base_request(request_id=rid) self.run_under_lease(task, req, check=False, env_extra={"FAKE_PI_MODE": mode}) receipt = json.loads( (self.report_dir("alpha", rid, task) / "receipt.json").read_text(encoding="utf-8")) self.assertIn(receipt["run_status"], ("failed", "unknown")) self.assertNotEqual(receipt.get("verdict"), "pass") # --- idempotency --- def test_idempotent_replay_and_digest_mismatch(self): task = self.initialized_task() req = self.base_request(request_id="idem-1") self.run_under_lease(task, req) first = json.loads( (self.report_dir("alpha", "idem-1", task) / "receipt.json").read_text(encoding="utf-8")) if self.argv_log.exists(): self.argv_log.unlink() result = self.run_under_lease(task, req) payload = self.last_json(result.stdout) self.assertTrue(payload.get("replay") or payload.get("run_status") == first["run_status"]) self.assertFalse(self.argv_log.exists(), "replay must not re-invoke PI") req2 = dict(req) req2["goal"] = "different goal same request_id" result = self.run_under_lease(task, req2, check=False) err = self.result_json(result) self.assertEqual(err["code"], "REQUEST_DIGEST_MISMATCH") def test_config_reload_snapshot_on_dispatch(self): task = self.initialized_task() req = self.base_request(request_id="snap-1") self.run_under_lease(task, req) receipt = json.loads( (self.report_dir("alpha", "snap-1", task) / "receipt.json").read_text(encoding="utf-8")) old_sha = receipt["config_sha256"] cfg = task / ".doc/ops/2026-09-24-subagent-model-config.md" text = cfg.read_text(encoding="utf-8") text = text.replace("config_revision：`subagent-models-2026-09-24-r1`", "config_revision：`subagent-models-2026-09-24-r2`") cfg.write_text(text + "\n\n", encoding="utf-8") req2 = self.base_request(request_id="snap-2") self.run_under_lease(task, req2) receipt2 = json.loads( (self.report_dir("alpha", "snap-2", task) / "receipt.json").read_text(encoding="utf-8")) self.assertNotEqual(old_sha, receipt2["config_sha256"]) self.assertEqual(receipt2["config_revision"], "subagent-models-2026-09-24-r2") # --- test_run governance --- def test_test_run_requires_approved_commands_and_records_failure(self): task = self.initialized_task() # other roles reject approved_commands req = self.base_request(request_id="badcmd", role_id="research", approved_commands=[["true"]]) result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertEqual(err["code"], "APPROVED_COMMANDS_NOT_ALLOWED") # test_run without plan/commands fails req = self.base_request(request_id="tr-miss", role_id="test_run", goal="run tests", approved_commands=None) result = self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_PROVIDER": "zai-coding-cn", "FAKE_PI_MODEL": "glm-5.3-flash"}) err = self.result_json(result) self.assertIn(err["code"], ("APPROVED_COMMANDS_REQUIRED", "PLAN_DIGEST_REQUIRED")) # failing approved command cannot be promoted by model success text plan = {"commands": [["false"]], "purpose": "unit"} plan_digest = sha256_text(json.dumps(plan, sort_keys=True)) req = self.base_request( request_id="tr-fail", role_id="test_run", goal="execute approved plan", approved_commands=[["false"]], plan_digest=plan_digest, authorization_ref="user:test-plan", ) self.run_under_lease( task, req, check=False, env_extra={ "FAKE_PI_PROVIDER": "zai-coding-cn", "FAKE_PI_MODEL": "glm-5.3-flash", "FAKE_PI_TEXT": "All tests passed!", }) receipt = json.loads( (self.report_dir("alpha", "tr-fail", task) / "receipt.json").read_text(encoding="utf-8")) self.assertNotEqual(receipt["verdict"], "pass") self.assertTrue(receipt.get("test_outcomes")) self.assertNotEqual(receipt["test_outcomes"][0].get("exit_code"), 0) # shared env ops unsupported req = self.base_request( request_id="tr-env", role_id="test_run", goal="deploy", approved_commands=[["true"]], plan_digest=sha256_text(json.dumps({"commands": [["true"]]}, sort_keys=True)), shared_env_ops=[{"op": "restart", "target": "dev"}], authorization_ref="user:test-plan", ) result = self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_PROVIDER": "zai-coding-cn", "FAKE_PI_MODEL": "glm-5.3-flash"}) err = self.result_json(result) self.assertEqual(err["code"], "SHARED_ENV_UNSUPPORTED") # --- timeout / process cleanup --- def test_timeout_kills_process_group(self): task = self.initialized_task() req = self.base_request(request_id="to-1", timeout_seconds=1) marker = pathlib.Path(self.temp.name) / "child-alive" # wrap fake pi to spawn a child that touches marker while hanging wrapper = pathlib.Path(self.temp.name) / "hang-pi" wrapper.write_text( "#!/usr/bin/env python3\n" "import os,sys,time,subprocess,json\n" f"log={str(self.argv_log)!r}\n" "open(log,'w').write(json.dumps(sys.argv[1:]))\n" f"marker={str(marker)!r}\n" "child=subprocess.Popen([sys.executable,'-c'," "'import time,pathlib,os; pathlib.Path(os.environ[\"M\"]).write_text(\"1\"); time.sleep(30)']," "env={**os.environ,'M':marker}, start_new_session=False)\n" "time.sleep(30)\n", encoding="utf-8") wrapper.chmod(0o755) started = time.monotonic() result = self.run_under_lease( task, req, check=False, env_extra={"PI_SUBAGENT_PI_BIN": str(wrapper)}) elapsed = time.monotonic() - started self.assertLess(elapsed, 15) receipt = json.loads( (self.report_dir("alpha", "to-1", task) / "receipt.json").read_text(encoding="utf-8")) self.assertIn(receipt["run_status"], ("failed", "cancelled", "unknown")) # give a moment; child should be reaped (marker may exist but process dead) time.sleep(0.5) # best-effort: no orphan sleep 30 from wrapper still running for this test uid/cwd self.assertIsNotNone(result) # --- concurrency --- def test_global_concurrency_limit(self): task = self.initialized_task() blockers = [] def hold(rid): req = self.base_request(request_id=rid, timeout_seconds=8) result = self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_MODE": "hang"}) blockers.append(result) threads = [] for rid in ("c1", "c2"): t = threading.Thread(target=hold, args=(rid,)) t.start() threads.append(t) # Wait until both slots are held (receipts show running) deadline = time.monotonic() + 10 while time.monotonic() < deadline: r1 = self.report_dir("alpha", "c1", task) / "receipt.json" r2 = self.report_dir("alpha", "c2", task) / "receipt.json" if r1.is_file() and r2.is_file(): break time.sleep(0.1) req = self.base_request(request_id="c3", timeout_seconds=3) result = self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_MODE": "ok"}) err_text = result.stderr + result.stdout self.assertNotEqual(result.returncode, 0, err_text) err = self.result_json(result) self.assertEqual(err["code"], "CONCURRENCY_LIMIT") # Do not pkill: broad patterns can match the harness itself. Hang runs # are bounded by timeout_seconds and reaped by the executor. for t in threads: t.join(timeout=20) # --- additional regressions --- def test_source_sha_rejects_symbolic_refs(self): task = self.initialized_task() req = self.base_request(request_id="sha-head", source_sha="HEAD") result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertEqual(err["code"], "INVALID_SOURCE_SHA") def test_corrupt_receipt_fail_closed(self): task = self.initialized_task() req = self.base_request(request_id="corrupt-1") self.run_under_lease(task, req) rdir = self.report_dir("alpha", "corrupt-1", task) (rdir / "receipt.json").write_text("{not-json", encoding="utf-8") result = self.run_under_lease(task, req, check=False) err = self.result_json(result) self.assertEqual(err["code"], "RECEIPT_CORRUPT") def test_replay_keeps_old_config_after_disable(self): task = self.initialized_task() req = self.base_request(request_id="replay-cfg") self.run_under_lease(task, req) first = json.loads( (self.report_dir("alpha", "replay-cfg", task) / "receipt.json").read_text(encoding="utf-8")) cfg = task / ".doc/ops/2026-09-24-subagent-model-config.md" text = cfg.read_text(encoding="utf-8") if "| research | r | true |" in text: text = text.replace("| research | r | true |", "| research | r | false |") else: text = text.replace( "| research | 检索、代码定位、资料摘录、需求初稿 | true |", "| research | 检索、代码定位、资料摘录、需求初稿 | false |") cfg.write_text(text, encoding="utf-8") if self.argv_log.exists(): self.argv_log.unlink() result = self.run_under_lease(task, req) payload = self.last_json(result.stdout) self.assertTrue(payload.get("replay")) self.assertEqual(payload["config_sha256"], first["config_sha256"]) self.assertFalse(self.argv_log.exists()) def test_readonly_guard_rejects_at_prefix(self): """Exercise extension path policy via Node strip-types if available.""" if not SOURCE_EXT.exists(): self.fail("extension missing") outside = pathlib.Path(self.temp.name) / "outside.txt" outside.write_text("SECRET", encoding="utf-8") snap = pathlib.Path(self.temp.name) / "snap" snap.mkdir() (snap / "ok.txt").write_text("hi", encoding="utf-8") (snap / "stray.txt").write_text("nope", encoding="utf-8") sub = snap / "sub" sub.mkdir() (sub / "nested.txt").write_text("x", encoding="utf-8") try: (snap / "evil.link").symlink_to(snap / "ok.txt") except OSError: pass probe = pathlib.Path(self.temp.name) / "guard-probe.mjs" probe.write_text( "import { pathToFileURL } from 'node:url';\n" "import fs from 'node:fs';\n" f"const extPath = {str(SOURCE_EXT.resolve())!r};\n" f"const root = {str(snap.resolve())!r};\n" f"const outside = {str(outside.resolve())!r};\n" "process.env.PI_SUBAGENT_SNAPSHOT_ROOT = root;\n" "process.env.PI_SUBAGENT_ALLOWED_FILES = JSON.stringify(['ok.txt']);\n" "const mod = await import(pathToFileURL(extPath).href);\n" "const handlers = [];\n" "const pi = { on(ev, fn) { handlers.push(fn); } };\n" "(mod.default || mod)(pi);\n" "const handler = handlers[0];\n" "const cases = [\n" " { toolName: 'read', input: { path: '@' + outside } },\n" " { toolName: 'read', input: { path: '~/outside.txt' } },\n" " { toolName: 'read', input: { path: outside } },\n" " { toolName: 'read', input: { path: 'ok.txt' } },\n" " { toolName: 'ls', input: {} },\n" # default '.' " { toolName: 'grep', input: { path: '.' } },\n" " { toolName: 'read', input: { path: 'stray.txt' } },\n" " { toolName: 'read', input: { path: 'sub' } },\n" # dir with non-allowlisted nested " { toolName: 'read', input: { file: 'ok.txt' } },\n" # unsupported carrier " { toolName: 'read', input: { path: 'evil.link' } },\n" "];\n" "const out = [];\n" "for (const c of cases) {\n" " const r = await handler(c, {});\n" " out.push({ tool: c.toolName, input: c.input, blocked: !!(r && r.block), reason: r && r.reason,\n" " pathCanon: c.input && c.input.path });\n" "}\n" "process.stdout.write(JSON.stringify(out));\n", encoding="utf-8") result = subprocess.run( ["node", "--experimental-strip-types", str(probe)], capture_output=True, text=True, timeout=30) if result.returncode != 0 and "strip-types" in (result.stderr or ""): self.skipTest("node --experimental-strip-types unavailable") self.assertEqual(result.returncode, 0, result.stderr) rows = json.loads(result.stdout) self.assertTrue(rows[0]["blocked"], rows) # @ self.assertTrue(rows[1]["blocked"], rows) # ~ self.assertTrue(rows[2]["blocked"], rows) # absolute outside self.assertFalse(rows[3]["blocked"], rows) # ok.txt self.assertTrue(rows[3]["pathCanon"].startswith(str(snap.resolve())), rows) self.assertTrue(rows[4]["blocked"], rows) # ls default . has stray.txt self.assertTrue(rows[5]["blocked"], rows) # grep . self.assertTrue(rows[6]["blocked"], rows) # stray file self.assertTrue(rows[7]["blocked"], rows) # dir with nested non-allow self.assertTrue(rows[8]["blocked"], rows) # file carrier self.assertTrue(rows[9]["blocked"], rows) # symlink def test_readonly_guard_allows_exact_root_listing(self): """Exact-root ls/grep must pass when snapshot inventory is allowlisted-only.""" if not SOURCE_EXT.exists(): self.fail("extension missing") snap = pathlib.Path(self.temp.name) / "snap-exact-root" snap.mkdir() (snap / "sample.txt").write_text("ok", encoding="utf-8") nested = snap / "nested" nested.mkdir() (nested / "inner.txt").write_text("in", encoding="utf-8") dirty = pathlib.Path(self.temp.name) / "snap-dirty" dirty.mkdir() (dirty / "sample.txt").write_text("ok", encoding="utf-8") (dirty / "stray.txt").write_text("nope", encoding="utf-8") try: (dirty / "evil.link").symlink_to(dirty / "sample.txt") except OSError: pass empty = pathlib.Path(self.temp.name) / "snap-empty" empty.mkdir() probe = pathlib.Path(self.temp.name) / "guard-exact-root.mjs" probe.write_text( "import { pathToFileURL } from 'node:url';\n" f"const extPath = {str(SOURCE_EXT.resolve())!r};\n" f"const clean = {str(snap.resolve())!r};\n" f"const dirty = {str(dirty.resolve())!r};\n" f"const empty = {str(empty.resolve())!r};\n" "const mod = await import(pathToFileURL(extPath).href);\n" "async function run(root, allow, cases) {\n" " process.env.PI_SUBAGENT_SNAPSHOT_ROOT = root;\n" " process.env.PI_SUBAGENT_ALLOWED_FILES = JSON.stringify(allow);\n" " const handlers = [];\n" " const pi = { on(ev, fn) { handlers.push(fn); } };\n" " (mod.default || mod)(pi);\n" " const handler = handlers[0];\n" " const out = [];\n" " for (const c of cases) {\n" " const input = c.input ? { ...c.input } : {};\n" " const r = await handler({ toolName: c.toolName, input }, {});\n" " out.push({ tag: c.tag, blocked: !!(r && r.block), reason: r && r.reason });\n" " }\n" " return out;\n" "}\n" "const cleanAllow = ['sample.txt', 'nested/inner.txt'];\n" "const rows = [];\n" "rows.push(...await run(clean, cleanAllow, [\n" " { tag: 'ls_dot', toolName: 'ls', input: { path: '.' } },\n" " { tag: 'ls_omit', toolName: 'ls', input: {} },\n" " { tag: 'grep_default', toolName: 'grep', input: {} },\n" " { tag: 'ls_nested', toolName: 'ls', input: { path: 'nested' } },\n" "]));\n" "rows.push(...await run(dirty, ['sample.txt'], [\n" " { tag: 'stray_file', toolName: 'read', input: { path: 'stray.txt' } },\n" " { tag: 'symlink', toolName: 'read', input: { path: 'evil.link' } },\n" " { tag: 'ls_dirty_root', toolName: 'ls', input: { path: '.' } },\n" "]));\n" "rows.push(...await run(empty, ['sample.txt'], [\n" " { tag: 'empty_root_ls', toolName: 'ls', input: { path: '.' } },\n" "]));\n" "process.stdout.write(JSON.stringify(rows));\n", encoding="utf-8") result = subprocess.run( ["node", "--experimental-strip-types", str(probe)], capture_output=True, text=True, timeout=30) if result.returncode != 0 and "strip-types" in (result.stderr or ""): self.skipTest("node --experimental-strip-types unavailable") self.assertEqual(result.returncode, 0, result.stderr) rows = {r["tag"]: r for r in json.loads(result.stdout)} self.assertFalse(rows["ls_dot"]["blocked"], rows) self.assertFalse(rows["ls_omit"]["blocked"], rows) self.assertFalse(rows["grep_default"]["blocked"], rows) self.assertFalse(rows["ls_nested"]["blocked"], rows) self.assertTrue(rows["stray_file"]["blocked"], rows) self.assertTrue(rows["symlink"]["blocked"], rows) self.assertTrue(rows["ls_dirty_root"]["blocked"], rows) self.assertFalse(rows["empty_root_ls"]["blocked"], rows) def test_output_cap_on_stdout_flood(self): task = self.initialized_task() flood = pathlib.Path(self.temp.name) / "flood-pi" flood.write_text( "#!/usr/bin/env python3\n" "import sys\n" "sys.stdout.write('x' * (5 * 1024 * 1024))\n" "sys.stdout.flush()\n", encoding="utf-8") flood.chmod(0o755) req = self.base_request(request_id="flood-1", timeout_seconds=10) self.run_under_lease( task, req, check=False, env_extra={"PI_SUBAGENT_PI_BIN": str(flood)}) rdir = self.report_dir("alpha", "flood-1", task) receipt = json.loads((rdir / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(receipt["run_status"], "failed") codes = {e.get("code") for e in receipt.get("errors") or []} self.assertTrue( "OUTPUT_CAP" in codes or "PI_PARSE" in codes or "PI_NONZERO" in codes or "MALFORMED_JSON" in codes, receipt) def test_leader_exit_reaps_orphan_child(self): task = self.initialized_task() marker = pathlib.Path(self.temp.name) / "orphan-child-alive" leader = pathlib.Path(self.temp.name) / "leader-exit-pi" leader.write_text( "#!/usr/bin/env python3\n" "import os, sys, subprocess, time, pathlib\n" f"marker={str(marker)!r}\n" "child=subprocess.Popen([sys.executable,'-c'," "'import time,pathlib,os; pathlib.Path(os.environ[\"M\"]).write_text(str(os.getpid())); time.sleep(30)']," "env={**os.environ,'M':marker}, start_new_session=False)\n" "print(child.pid, flush=True)\n" "time.sleep(0.2)\n" "sys.exit(0)\n", encoding="utf-8") leader.chmod(0o755) req = self.base_request(request_id="orphan-1", timeout_seconds=8) self.run_under_lease( task, req, check=False, env_extra={"PI_SUBAGENT_PI_BIN": str(leader)}) receipt = json.loads( (self.report_dir("alpha", "orphan-1", task) / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(receipt["run_status"], "failed", receipt) codes = {e.get("code") for e in receipt.get("errors") or []} self.assertIn("ORPHAN_CHILD", codes, receipt) time.sleep(0.3) if marker.is_file(): try: cpid = int(marker.read_text().strip()) except ValueError: cpid = None if cpid: # child must be gone; never kill arbitrary recycled PID without identity alive = pathlib.Path(f"/proc/{cpid}").exists() self.assertFalse(alive, f"orphan child {cpid} still alive") def test_orphan_result_is_readonly(self): task = self.initialized_task() req = self.base_request(request_id="orphan-ro") self.run_under_lease(task, req) rdir = self.report_dir("alpha", "orphan-ro", task) receipt_path = rdir / "receipt.json" data = json.loads(receipt_path.read_text(encoding="utf-8")) data["run_status"] = "running" data["supervisor_pid"] = 1 data["supervisor_start"] = 0 data["supervisor_boot_id"] = "not-the-real-boot" receipt_path.write_text(json.dumps(data, indent=2), encoding="utf-8") before = hashlib.sha256(receipt_path.read_bytes()).hexdigest() result = self.cli( "result", "--task", "alpha", "--request-id", "orphan-ro", cwd=task, check=False) payload = self.result_json(result) self.assertEqual(payload.get("run_status"), "unknown") after = hashlib.sha256(receipt_path.read_bytes()).hexdigest() self.assertEqual(before, after, "result must not mutate receipt on disk") def test_approved_commands_do_not_mutate_model_snapshot(self): task = self.initialized_task() plan = {"commands": [["python3", "-c", "open('src/sample.py','w').write('MUTATED'); " "import os; os.symlink('src/sample.py','src/evil.link')"]], "purpose": "mutate"} plan_digest = sha256_text(json.dumps(plan, sort_keys=True)) req = self.base_request( request_id="snap-iso", role_id="test_run", goal="mutate snapshot", approved_commands=plan["commands"], plan_digest=plan_digest, authorization_ref="user:test-plan", ) self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_PROVIDER": "zai-coding-cn", "FAKE_PI_MODEL": "glm-5.3-flash"}) receipt = json.loads( (self.report_dir("alpha", "snap-iso", task) / "receipt.json").read_text(encoding="utf-8")) # Model input hash must remain original sample.py content self.assertEqual(receipt["input_files"][0]["sha256"], sha256_text("print('hello')\n")) self.assertIn("test_outcomes", receipt) # Evidence files persisted under report dir ev = receipt["test_outcomes"][0] self.assertTrue(ev.get("stdout_path") or ev.get("stderr_path")) if ev.get("stdout_path"): self.assertTrue((self.report_dir("alpha", "snap-iso", task) / ev["stdout_path"]).is_file()) def test_failed_replay_stays_failure(self): task = self.initialized_task() req = self.base_request(request_id="fail-replay") self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_MODE": "nonzero"}) first = json.loads( (self.report_dir("alpha", "fail-replay", task) / "receipt.json").read_text(encoding="utf-8")) self.assertEqual(first["run_status"], "failed") self.assertFalse(first.get("ok")) result = self.run_under_lease(task, req, check=False) self.assertNotEqual(result.returncode, 0) payload = self.result_json(result) self.assertTrue(payload.get("replay")) self.assertEqual(payload.get("run_status"), "failed") self.assertFalse(payload.get("ok")) def test_scrub_env_drops_pi_dir_for_tests(self): # Import executor scrub directly from source import importlib.util spec = importlib.util.spec_from_file_location( "pi_subagents_under_test", SOURCE_EXEC) mod = importlib.util.module_from_spec(spec) # agentctl must be importable sys.path.insert(0, str(SOURCE_EXEC.parent)) spec.loader.exec_module(mod) leaked = mod.scrub_env({ "PATH": "/usr/bin", "PI_CODING_AGENT_DIR": "/tmp/real-auth", "PI_SUBAGENT_SNAPSHOT_ROOT": "/tmp/snap", "XDG_RUNTIME_DIR": "/run/user/1000", "CURSOR_API_KEY": "secret", }, for_pi=False) self.assertNotIn("PI_CODING_AGENT_DIR", leaked) self.assertNotIn("PI_SUBAGENT_SNAPSHOT_ROOT", leaked) self.assertNotIn("XDG_RUNTIME_DIR", leaked) self.assertNotIn("CURSOR_API_KEY", leaked) # --- crash leaves unknown --- def test_crash_before_receipt_terminal_is_unknown_on_status(self): task = self.initialized_task() # Simulate: write running receipt then crash without finalize by using mode crash after # The executor should persist running first; crash mid leaves unknown when queried. req = self.base_request(request_id="crash-1") self.run_under_lease( task, req, check=False, env_extra={"FAKE_PI_MODE": "crash"}) result = self.cli( "status", "--task", "alpha", "--request-id", "crash-1", "--config", str(task / ".doc/ops/2026-09-24-subagent-model-config.md"), cwd=task, check=False) payload = self.result_json(result) self.assertIn(payload.get("run_status"), ("failed", "unknown")) # --- MCP --- def _mcp_stderr_preview(self, proc, limit=4000): """Non-blocking stderr peek — never block-read a live process pipe.""" if not proc.stderr: return "" try: fd = proc.stderr.fileno() flags = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) chunk = proc.stderr.read(limit) return chunk or "" except Exception as exc: return f"" def _mcp_recv_line(self, proc, timeout=15.0): """Bounded NDJSON line receive from MCP stdout.""" fd = proc.stdout.fileno() flags = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) sel = selectors.DefaultSelector() sel.register(proc.stdout, selectors.EVENT_READ) buf = "" deadline = time.monotonic() + timeout try: while time.monotonic() < deadline: remaining = max(0.01, deadline - time.monotonic()) events = sel.select(min(0.5, remaining)) if events: try: chunk = proc.stdout.read(8192) except BlockingIOError: chunk = "" if chunk: buf += chunk if "\n" in buf: line, _rest = buf.split("\n", 1) return line elif proc.poll() is not None: break elif proc.poll() is not None: try: chunk = proc.stdout.read() or "" buf += chunk except Exception: pass if "\n" in buf: return buf.split("\n", 1)[0] break finally: sel.close() err = self._mcp_stderr_preview(proc) self.fail(f"MCP RPC timeout/no line; stderr_preview={err!r} buf={buf[:200]!r}") def test_mcp_initialize_list_run_result(self): if not (self.root / "scripts" / "mcp" / "pi-subagents-mcp.mjs").exists(): self.fail("MCP server missing") task = self.initialized_task() mcp = self.root / "scripts" / "mcp" / "pi-subagents-mcp.mjs" env = self.env({"AGENTCTL_REPO": str(self.root)}) env["PI_SUBAGENT_PI_BIN"] = str(self.fake_pi) proc = subprocess.Popen( ["node", str(mcp)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(self.root), env=env) def rpc(msg, timeout=20.0): if "id" not in msg and str(msg.get("method", "")).startswith("notifications/"): proc.stdin.write(json.dumps(msg) + "\n") proc.stdin.flush() return None proc.stdin.write(json.dumps(msg) + "\n") proc.stdin.flush() line = self._mcp_recv_line(proc, timeout=timeout) self.assertTrue(line.strip(), f"empty MCP line; stderr={self._mcp_stderr_preview(proc)!r}") return json.loads(line) try: init = rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "0"}}}) self.assertEqual(init["result"]["serverInfo"]["name"], "pi-subagents") rpc({"jsonrpc": "2.0", "method": "notifications/initialized"}) tools = rpc({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) names = {t["name"] for t in tools["result"]["tools"]} self.assertEqual(names, {"pi_subagent_roles", "pi_subagent_run", "pi_subagent_result"}) roles = rpc({"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "pi_subagent_roles", "arguments": {"worktree": str(task), "task_id": "alpha"}}}) self.assertFalse(roles["result"].get("isError")) body = json.loads(roles["result"]["content"][0]["text"]) self.assertIn("roles", body) run = rpc({"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "pi_subagent_run", "arguments": { "worktree": str(task), "task_id": "alpha", "request": self.base_request(request_id="mcp-1"), }}}) self.assertIn("result", run) self.assertFalse(run["result"].get("isError"), run["result"]) deadline = time.monotonic() + 25 result_body = None while time.monotonic() < deadline: got = rpc({"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "pi_subagent_result", "arguments": { "worktree": str(task), "task_id": "alpha", "request_id": "mcp-1", }}}) text = got["result"]["content"][0]["text"] result_body = json.loads(text) if result_body.get("run_status") == "succeeded": break if result_body.get("run_status") in ("failed", "cancelled", "unknown"): self.fail(f"MCP run did not succeed: {result_body}") if got["result"].get("isError") and result_body.get("code") not in ( "NOT_FOUND", "UNKNOWN_REQUEST"): if result_body.get("code"): self.fail(f"MCP supervisor rejected: {result_body}") time.sleep(0.3) self.assertIsNotNone(result_body) self.assertEqual(result_body.get("run_status"), "succeeded", result_body) rdir = self.report_dir("alpha", "mcp-1", task) self.assertTrue((rdir / "receipt.json").is_file()) self.assertTrue((rdir / "report.md").is_file()) # nested request schema present for auto-dispatch req_schema = next(t for t in tools["result"]["tools"] if t["name"] == "pi_subagent_run") props = req_schema["inputSchema"]["properties"]["request"]["properties"] for key in ("role_id", "request_id", "goal", "authorization_ref", "source_sha", "input_files", "timeout_seconds"): self.assertIn(key, props) self.assertNotIn("config_path", props) finally: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=3) def _start_mcp(self, mcp_path, cwd): env = self.env({"AGENTCTL_REPO": str(self.root)}) env["PI_SUBAGENT_PI_BIN"] = str(self.fake_pi) proc = subprocess.Popen( ["node", str(mcp_path)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(cwd), env=env) return proc def _mcp_rpc(self, proc, msg, timeout=20.0): if "id" not in msg and str(msg.get("method", "")).startswith("notifications/"): proc.stdin.write(json.dumps(msg) + "\n") proc.stdin.flush() return None proc.stdin.write(json.dumps(msg) + "\n") proc.stdin.flush() line = self._mcp_recv_line(proc, timeout=timeout) self.assertTrue(line.strip(), f"empty MCP line; stderr={self._mcp_stderr_preview(proc)!r}") return json.loads(line) def test_mcp_from_worktree_uses_main_agentctl(self): """Server INSTALL_ROOT may be a linked worktree; agentctl must resolve from main.""" task = self.initialized_task() mcp_wt = task / "scripts" / "mcp" / "pi-subagents-mcp.mjs" self.assertTrue(mcp_wt.is_file(), "worktree must contain MCP script from fixture commit") # Distinct paths: worktree install vs main tip self.assertNotEqual(mcp_wt.resolve(), (self.root / "scripts/mcp/pi-subagents-mcp.mjs").resolve()) proc = self._start_mcp(mcp_wt, task) try: self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "0"}}}) self._mcp_rpc(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"}) roles = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "pi_subagent_roles", "arguments": {"worktree": str(task), "task_id": "alpha"}}}) self.assertFalse(roles["result"].get("isError"), roles) run = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "pi_subagent_run", "arguments": { "worktree": str(task), "task_id": "alpha", "request": self.base_request(request_id="mcp-main-1"), }}}) self.assertFalse(run["result"].get("isError"), run) deadline = time.monotonic() + 25 while time.monotonic() < deadline: got = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "pi_subagent_result", "arguments": { "worktree": str(task), "task_id": "alpha", "request_id": "mcp-main-1", }}}) body = json.loads(got["result"]["content"][0]["text"]) if body.get("run_status") == "succeeded": break if body.get("run_status") in ("failed", "cancelled", "unknown") or got["result"].get("isError"): self.fail(body) time.sleep(0.2) else: self.fail("timeout waiting for MCP run via worktree-installed server") self.assertTrue((self.report_dir("alpha", "mcp-main-1", task) / "receipt.json").is_file()) finally: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() def test_mcp_surfaces_role_disabled_and_scope_refusal(self): task = self.initialized_task() mcp = self.root / "scripts" / "mcp" / "pi-subagents-mcp.mjs" proc = self._start_mcp(mcp, self.root) try: self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}}) self._mcp_rpc(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"}) # Disabled auxiliary — typed failure, no receipt forever run = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "pi_subagent_run", "arguments": { "worktree": str(task), "task_id": "alpha", "request": self.base_request( request_id="mcp-aux", role_id="auxiliary"), }}}) self.assertFalse(run["result"].get("isError"), "accept is async") deadline = time.monotonic() + 15 saw = None while time.monotonic() < deadline: got = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "pi_subagent_result", "arguments": { "worktree": str(task), "task_id": "alpha", "request_id": "mcp-aux", }}}) saw = json.loads(got["result"]["content"][0]["text"]) if got["result"].get("isError") and saw.get("code") == "ROLE_DISABLED": break if saw.get("run_status") == "running": self.assertFalse(got["result"].get("isError"), "running must not be isError") time.sleep(0.2) else: self.fail(f"expected ROLE_DISABLED via MCP result, got {saw}") self.assertEqual(saw.get("code"), "ROLE_DISABLED") # No unleased source writes under worktree for rejected run mcp_tmp = list((task / ".doc/test-reports/assets/pi-subagents").glob(".mcp-*")) self.assertEqual(mcp_tmp, []) finally: proc.terminate() try: proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() # Scope-refused registered task (no report scope) task2 = self.initialized_task(task="beta", scopes=("src",)) proc2 = self._start_mcp(mcp, self.root) try: self._mcp_rpc(proc2, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}}) self._mcp_rpc(proc2, {"jsonrpc": "2.0", "method": "notifications/initialized"}) req = self.base_request(request_id="mcp-scope", task_id="beta") run = self._mcp_rpc(proc2, {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "pi_subagent_run", "arguments": { "worktree": str(task2), "task_id": "beta", "request": req, }}}) self.assertFalse(run["result"].get("isError")) deadline = time.monotonic() + 15 saw = None while time.monotonic() < deadline: got = self._mcp_rpc(proc2, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "pi_subagent_result", "arguments": { "worktree": str(task2), "task_id": "beta", "request_id": "mcp-scope", }}}) saw = json.loads(got["result"]["content"][0]["text"]) if got["result"].get("isError") and saw.get("code"): break time.sleep(0.2) else: self.fail(f"expected scope/lease error via MCP, got {saw}") self.assertTrue(got["result"].get("isError")) self.assertEqual(saw.get("code"), "PATH_OUT_OF_SCOPE", saw) finally: proc2.terminate() try: proc2.wait(timeout=5) except subprocess.TimeoutExpired: proc2.kill() def test_mcp_running_poll_is_not_error(self): task = self.initialized_task() mcp = self.root / "scripts" / "mcp" / "pi-subagents-mcp.mjs" env = self.env({"AGENTCTL_REPO": str(self.root), "FAKE_PI_MODE": "hang", "PI_SUBAGENT_PI_BIN": str(self.fake_pi)}) proc = subprocess.Popen( ["node", str(mcp)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=str(self.root), env=env) try: self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}}) self._mcp_rpc(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"}) self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "pi_subagent_run", "arguments": { "worktree": str(task), "task_id": "alpha", "request": self.base_request( request_id="mcp-hang2", timeout_seconds=12), }}}) deadline = time.monotonic() + 10 saw_running = False while time.monotonic() < deadline: got = self._mcp_rpc(proc, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "pi_subagent_result", "arguments": { "worktree": str(task), "task_id": "alpha", "request_id": "mcp-hang2", }}}) body = json.loads(got["result"]["content"][0]["text"]) if body.get("run_status") == "running": self.assertFalse(got["result"].get("isError"), body) saw_running = True break if got["result"].get("isError") and body.get("code") in ("NOT_FOUND", "UNKNOWN_REQUEST"): time.sleep(0.15) continue time.sleep(0.15) self.assertTrue(saw_running, "expected a non-error running poll") finally: proc.terminate() try: proc.wait(timeout=8) except subprocess.TimeoutExpired: proc.kill() if __name__ == "__main__": unittest.main()