weichen
f9f055bee1
Implement loopback HTTP console with claim new, integrate prepare, and pi packet sharing the same helpers as CLI; ship SPA and MCP brief/claim_archive wrappers.
281 lines
12 KiB
Python
281 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Catalog governance regressions (schema, FTS5 CJK, overlap gates, vector degrade)."""
|
|
|
|
import json
|
|
import pathlib
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
|
|
CATALOG_SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl_catalog.py"
|
|
POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json"
|
|
|
|
|
|
def install_governance_policy(root: pathlib.Path) -> None:
|
|
dest = root / "configs"
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json")
|
|
|
|
|
|
class CatalogTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="agentctl-catalog-")
|
|
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", "Catalog Test")
|
|
self.git("config", "user.email", "catalog-test@example.invalid")
|
|
(self.root / "scripts").mkdir()
|
|
shutil.copy2(SOURCE, self.root / "scripts" / "agentctl.py")
|
|
shutil.copy2(SOURCE.parent / "agentctl_w4.py", self.root / "scripts" / "agentctl_w4.py")
|
|
shutil.copy2(CATALOG_SOURCE, self.root / "scripts" / "agentctl_catalog.py")
|
|
vis = SOURCE.parent / "agentctl_visibility.py"
|
|
if vis.exists():
|
|
shutil.copy2(vis, self.root / "scripts" / "agentctl_visibility.py")
|
|
cons = SOURCE.parent / "agentctl_console.py"
|
|
if cons.exists():
|
|
shutil.copy2(cons, self.root / "scripts" / "agentctl_console.py")
|
|
install_governance_policy(self.root)
|
|
(self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
|
|
for directory in ("src/a", "src/b", ".doc/ops", ".doc/status"):
|
|
(self.root / directory).mkdir(parents=True, exist_ok=True)
|
|
(self.root / ".doc/ops/CURRENT.md").write_text(
|
|
"# Coordination\n\n| owner | task | branch | worktree | mutex | status |\n"
|
|
"| --- | --- | --- | --- | --- | --- |\n"
|
|
"| Alice | alpha | feature/alpha | .worktrees/alpha | src | in_progress |\n",
|
|
encoding="utf-8",
|
|
)
|
|
(self.root / ".doc/status/FEATURE-MATRIX.md").write_text(
|
|
"# Features\n\n| record_id | 模块 | 状态 |\n| --- | --- | --- |\n"
|
|
"| core | core | planned |\n| intent-plan | 多意图 | done |\n",
|
|
encoding="utf-8",
|
|
)
|
|
self.git("add", ".")
|
|
self.git("commit", "-m", "fixture")
|
|
self.ctl("init")
|
|
|
|
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 ctl(self, *args, check=True, cwd=None):
|
|
result = subprocess.run(
|
|
["python3", "-B", str(self.root / "scripts" / "agentctl.py"), *args],
|
|
cwd=cwd or self.root, text=True, capture_output=True,
|
|
)
|
|
if check and result.returncode != 0:
|
|
raise AssertionError(result.stderr or result.stdout or f"exit {result.returncode}")
|
|
out = (result.stdout or "").strip()
|
|
err = (result.stderr or "").strip()
|
|
payload = None
|
|
for stream in (out, err):
|
|
if not stream:
|
|
continue
|
|
try:
|
|
payload = json.loads(stream.splitlines()[-1])
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return result.returncode, payload, out, err
|
|
|
|
def make_worktree(self, task_id: str):
|
|
wt = self.root / ".worktrees" / task_id
|
|
self.git("worktree", "add", "-b", f"feature/{task_id}", str(wt))
|
|
return wt
|
|
|
|
def test_schema_and_crud(self):
|
|
code, payload, _, _ = self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-demo-overlap",
|
|
"--title", "多意图持久计划",
|
|
"--statement", "多意图持久计划 shadow 灰度放量",
|
|
"--owner", "Cursor",
|
|
"--feature", "intent-plan",
|
|
)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(payload["status"], "added")
|
|
_, listed, _, _ = self.ctl("catalog", "list")
|
|
self.assertEqual(listed["status"], "ok")
|
|
self.assertEqual(len(listed["requirements"]), 1)
|
|
_, shown, _, _ = self.ctl("catalog", "show", "--id", "R-demo-overlap")
|
|
self.assertEqual(shown["requirement"]["id"], "R-demo-overlap")
|
|
_, status, _, _ = self.ctl("catalog", "status")
|
|
self.assertGreaterEqual(status["requirements"], 1)
|
|
|
|
def test_cjk_fts_digram_recall(self):
|
|
"""CJK digrams must hit under unicode61 char-split (trigram would miss 灰度)."""
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-gray",
|
|
"--title", "灰度放量",
|
|
"--statement", "多意图持久计划 shadow 灰度放量",
|
|
"--owner", "Cursor",
|
|
)
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-other",
|
|
"--title", "语义摘要",
|
|
"--statement", "语义摘要 REPLACE fail-closed 灰度",
|
|
"--owner", "Cursor",
|
|
)
|
|
# Direct FTS verification including trigram contrast.
|
|
import sys
|
|
sys.path.insert(0, str(self.root / "scripts"))
|
|
import agentctl_catalog as cat
|
|
db = sqlite3.connect(self.root / ".locks" / "state.sqlite")
|
|
db.row_factory = sqlite3.Row
|
|
# Create trigram index for contrast.
|
|
db.execute("CREATE VIRTUAL TABLE IF NOT EXISTS trigram_probe USING fts5(body, tokenize='trigram')")
|
|
db.execute("INSERT INTO trigram_probe VALUES (?)", ("多意图持久计划 shadow 灰度放量",))
|
|
db.execute("INSERT INTO trigram_probe VALUES (?)", ("语义摘要 REPLACE fail-closed 灰度",))
|
|
tri = db.execute("SELECT count(*) FROM trigram_probe WHERE trigram_probe MATCH ?", ('"灰度"',)).fetchone()[0]
|
|
self.assertEqual(tri, 0, "trigram must miss digram 灰度")
|
|
hits = cat.search(db, "灰度", top=10)
|
|
self.assertGreaterEqual(len(hits), 2, f"CJK split must recall digram; got {hits}")
|
|
db.close()
|
|
_, payload, _, _ = self.ctl("catalog", "search", "--query", "灰度")
|
|
self.assertGreaterEqual(len(payload["hits"]), 2)
|
|
|
|
def test_hard_same_req_id_conflict(self):
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-owned",
|
|
"--title", "owned",
|
|
"--statement", "already owned requirement",
|
|
"--owner", "Alice",
|
|
"--source-task", "alpha-owner",
|
|
)
|
|
wt = self.make_worktree("beta-conflict")
|
|
code, payload, _, err = self.ctl(
|
|
"task", "register",
|
|
"--id", "beta-conflict",
|
|
"--owner", "Bob",
|
|
"--scope", "src/b",
|
|
"--requirement", "R-owned",
|
|
"--intent", "already owned requirement",
|
|
cwd=wt,
|
|
check=False,
|
|
)
|
|
self.assertNotEqual(code, 0)
|
|
denied = payload or json.loads(err.splitlines()[-1])
|
|
self.assertEqual(denied["code"], "REQUIREMENT_CONFLICT")
|
|
self.assertEqual(denied["kind"], "same_req_id")
|
|
|
|
def test_soft_overlap_requires_ack(self):
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-existing",
|
|
"--title", "治理目录库",
|
|
"--statement", "跨对话需求重叠检测与治理目录库",
|
|
"--owner", "Alice",
|
|
"--source-task", "alpha-owner",
|
|
)
|
|
wt = self.make_worktree("soft-probe")
|
|
# Register without catalog args first.
|
|
self.ctl(
|
|
"task", "register",
|
|
"--id", "soft-probe",
|
|
"--owner", "Bob",
|
|
"--scope", "src/b",
|
|
cwd=wt,
|
|
)
|
|
code, payload, _, _ = self.ctl(
|
|
"catalog", "overlap",
|
|
"--task", "soft-probe",
|
|
"--statement", "跨对话需求重叠检测治理目录",
|
|
"--requirement", "R-new-soft",
|
|
)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(payload["status"], "ack_required")
|
|
self.assertTrue(payload["pending_soft"])
|
|
oid = payload["pending_soft"][0]["id"]
|
|
# Exclusive with blocked until ack.
|
|
code, denied, _, err = self.ctl(
|
|
"with", "--task", "soft-probe", "--resource", "path:src/b", "--", "true",
|
|
cwd=wt, check=False,
|
|
)
|
|
self.assertNotEqual(code, 0)
|
|
body = denied or json.loads(err.splitlines()[-1])
|
|
self.assertEqual(body["code"], "OVERLAP_ACK_REQUIRED")
|
|
self.ctl(
|
|
"catalog", "ack",
|
|
"--task", "soft-probe",
|
|
"--overlap", oid,
|
|
"--decision", "proceed",
|
|
"--reason", "incremental extension of same theme",
|
|
"--decided-by", "Bob",
|
|
)
|
|
code, _, _, _ = self.ctl(
|
|
"with", "--task", "soft-probe", "--resource", "path:src/b", "--", "true",
|
|
cwd=wt,
|
|
)
|
|
self.assertEqual(code, 0)
|
|
|
|
def test_vector_layer_degrades_without_key(self):
|
|
# Enable vector in policy for this repo.
|
|
policy_path = self.root / "configs" / "agent-governance.json"
|
|
data = json.loads(policy_path.read_text(encoding="utf-8"))
|
|
data["catalog_vector"] = {
|
|
"enabled": True, "timeout_seconds": 2, "fail_open": True, "cosine_threshold": 0.82,
|
|
}
|
|
policy_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-vec",
|
|
"--title", "向量降级",
|
|
"--statement", "向量库可降级不阻断门禁",
|
|
"--owner", "Cursor",
|
|
)
|
|
# No EMBEDDING_API_KEY → degraded, still returns FTS soft or none without crash.
|
|
env = {"EMBEDDING_ENDPOINT_URL": "http://127.0.0.1:9/embeddings", "EMBEDDING_API_KEY": ""}
|
|
result = subprocess.run(
|
|
["python3", "-B", str(self.root / "scripts" / "agentctl.py"),
|
|
"catalog", "overlap", "--task", "no-task-yet", "--statement", "向量库可降级不阻断"],
|
|
cwd=self.root, text=True, capture_output=True,
|
|
env={**dict(**{k: v for k, v in __import__('os').environ.items()}), **env},
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout.strip().splitlines()[-1])
|
|
self.assertIn(payload.get("vector_layer"), ("degraded", "disabled", "ok"))
|
|
|
|
def test_hard_proceed_ack_rejected(self):
|
|
self.ctl(
|
|
"catalog", "add",
|
|
"--id", "R-hard",
|
|
"--title", "mutex feature",
|
|
"--statement", "same feature ownership",
|
|
"--owner", "Alice",
|
|
"--feature", "core",
|
|
"--source-task", "holder-a",
|
|
)
|
|
wt = self.make_worktree("hard-ack")
|
|
self.ctl("task", "register", "--id", "hard-ack", "--owner", "Bob", "--scope", "src/b", cwd=wt)
|
|
_, payload, _, _ = self.ctl(
|
|
"catalog", "overlap",
|
|
"--task", "hard-ack",
|
|
"--statement", "same feature ownership",
|
|
"--feature", "core",
|
|
"--requirement", "R-hard-b",
|
|
)
|
|
self.assertTrue(payload["hard"])
|
|
oid = payload["hard"][0]["id"]
|
|
code, denied, _, err = self.ctl(
|
|
"catalog", "ack",
|
|
"--task", "hard-ack",
|
|
"--overlap", oid,
|
|
"--decision", "proceed",
|
|
"--reason", "should fail",
|
|
check=False,
|
|
)
|
|
self.assertNotEqual(code, 0)
|
|
body = denied or json.loads(err.splitlines()[-1])
|
|
self.assertEqual(body["code"], "REQUIREMENT_CONFLICT")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|