#!/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") 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()