weichen
28ed7005dd
Extend agentctl SQLite with catalog tables, CJK FTS5 recall, hard/soft overlap gates, fail-open embeddings, backfill, and MCP/skills hooks so cross-conversation designs cannot silently conflict.
1487 lines
71 KiB
Python
1487 lines
71 KiB
Python
#!/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")
|
||
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<!-- touch -->\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"<stderr unavailable: {exc}>"
|
||
|
||
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()
|