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.
156 lines
7.1 KiB
Python
156 lines
7.1 KiB
Python
"""CodeGraph governance + read-only audit commands."""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class CodegraphGovernanceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / 'repo'
|
|
self.root.mkdir()
|
|
(self.root / 'scripts').mkdir()
|
|
for name in ('agent-worktree.sh', 'agent_worktree.py', 'agentctl.py',
|
|
'agentctl', 'agentctl_w4.py', 'agentctl_catalog.py'):
|
|
shutil.copy2(SOURCE / name, self.root / 'scripts' / name)
|
|
(self.root / 'configs').mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(SOURCE.parent / 'configs' / 'agent-governance.json',
|
|
self.root / 'configs' / 'agent-governance.json')
|
|
(self.root / '.gitignore').write_text('.worktrees/\n.locks/\n.codegraph/\n')
|
|
(self.root / 'source.txt').write_text('base\n')
|
|
doc = self.root / '.doc/status'
|
|
doc.mkdir(parents=True)
|
|
ops = self.root / '.doc/ops'
|
|
ops.mkdir(parents=True)
|
|
(ops / 'CURRENT.md').write_text(
|
|
'# 当前焦点\n\n## 焦点\n\n0. x\n\n## 并行认领\n\n'
|
|
'| owner | task | branch | worktree | mutex | status |\n'
|
|
'|---|---|---|---|---|---|\n'
|
|
'| Cursor | seed | feature/seed | .worktrees/seed | m | done |\n\n'
|
|
'## 清理接管与遗留现场\n\n| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n'
|
|
'|---|---|---|---|\n| — | none | — | — |\n\n'
|
|
'## 集成槽位\n| holder | phase | since | scope | queue |\n'
|
|
'|--------|-------|-------|-------|-------|\n'
|
|
'| - | idle | 2026-09-21 | - | - |\n\n'
|
|
'## 受控约束\n\n- none\n\n## 活跃入口\n\n- x\n'
|
|
)
|
|
(doc / 'FEATURE-MATRIX.md').write_text(
|
|
'# 矩阵\n\n| record_id | 模块 | 声明范围 | 状态 | 发布关系 | 证据 | 剩余工作 / 关闭条件 | mutex |\n'
|
|
'|------|----------|------|----------|------|---------------------|-------|-------|\n'
|
|
'| alpha-mod | Alpha | scope | done | non_blocker | e | closed | a |\n'
|
|
'| beta-mod | Beta | scope | partial | non_blocker | e | open | b |\n'
|
|
)
|
|
self.env = {**os.environ, 'PYTHONDONTWRITEBYTECODE': '1'}
|
|
for name in ('AGENTCTL_RUN_ID', 'AGENTCTL_TASK_ID'):
|
|
self.env.pop(name, None)
|
|
self.git('init', '-b', 'dev')
|
|
self.git('config', 'user.email', 'test@example.invalid')
|
|
self.git('config', 'user.name', 'Test')
|
|
self.git('add', '.')
|
|
self.git('commit', '-qm', 'fixture')
|
|
|
|
def run_cmd(self, args, cwd=None, env=None, timeout=30):
|
|
return subprocess.run(args, cwd=cwd or self.root, env=env or self.env,
|
|
text=True, capture_output=True, timeout=timeout)
|
|
|
|
def git(self, *args, cwd=None):
|
|
result = self.run_cmd(['git', *args], cwd=cwd)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
return result
|
|
|
|
def ctl(self, *args, cwd=None, check=True):
|
|
result = self.run_cmd(
|
|
[sys.executable, str(self.root / 'scripts/agentctl.py'), *args], cwd=cwd)
|
|
if check:
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
return result
|
|
|
|
def test_codegraph_sync_fail_open_without_binary(self):
|
|
self.ctl('init')
|
|
env = {**self.env, 'PATH': '/usr/bin:/bin'}
|
|
result = self.run_cmd(
|
|
[sys.executable, str(self.root / 'scripts/agentctl.py'),
|
|
'hooks', 'codegraph-sync'],
|
|
env=env)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload['status'], 'ok')
|
|
self.assertEqual(payload['hook'], 'codegraph-sync')
|
|
self.assertEqual(payload['codegraph']['status'], 'not_installed')
|
|
self.assertEqual(payload['codegraph'].get('exit_code', 0), 0)
|
|
|
|
def test_hooks_install_includes_codegraph_sync_hooks(self):
|
|
self.ctl('init')
|
|
self.ctl('hooks', 'install')
|
|
for name in ('post-merge', 'post-checkout'):
|
|
path = self.root / '.git/hooks' / name
|
|
self.assertTrue(path.exists(), name)
|
|
text = path.read_text()
|
|
self.assertIn('agentctl-managed-v1', text)
|
|
self.assertIn('hooks codegraph-sync', text)
|
|
|
|
def test_docs_drift_detects_done_regression(self):
|
|
self.ctl('init')
|
|
first = self.ctl('docs', 'drift')
|
|
payload = json.loads(first.stdout)
|
|
self.assertEqual(payload['status'], 'ok')
|
|
self.assertEqual(payload['baseline'], 'first_snapshot')
|
|
self.assertEqual(payload['regressions'], [])
|
|
# Regress alpha-mod done → planned on disk + commit so tip reflects it.
|
|
matrix = self.root / '.doc/status/FEATURE-MATRIX.md'
|
|
text = matrix.read_text().replace(
|
|
'| alpha-mod | Alpha | scope | done |',
|
|
'| alpha-mod | Alpha | scope | planned |')
|
|
matrix.write_text(text)
|
|
self.git('add', '.doc/status/FEATURE-MATRIX.md')
|
|
self.git('commit', '-qm', 'regress alpha')
|
|
second = self.ctl('docs', 'drift')
|
|
payload = json.loads(second.stdout)
|
|
self.assertTrue(any(
|
|
r['record_id'] == 'alpha-mod' and r['severity'] == 'hard'
|
|
for r in payload['regressions']), payload)
|
|
open_ids = {row['record_id'] for row in payload['open_capabilities']}
|
|
self.assertIn('alpha-mod', open_ids)
|
|
self.assertIn('beta-mod', open_ids)
|
|
|
|
def test_cleanup_report_never_deletes(self):
|
|
self.ctl('init')
|
|
helper = self.root / 'scripts/agent-worktree.sh'
|
|
created = self.run_cmd(['bash', str(helper), 'create', 'feature/task-a'])
|
|
self.assertEqual(created.returncode, 0, created.stderr)
|
|
wt = Path(json.loads(created.stdout)['worktree'])
|
|
self.assertTrue(wt.is_dir())
|
|
report = self.ctl('cleanup', 'report')
|
|
payload = json.loads(report.stdout)
|
|
self.assertEqual(payload['status'], 'ok')
|
|
self.assertGreaterEqual(payload['summary']['total'], 1)
|
|
self.assertTrue(wt.is_dir(), 'cleanup report must not remove worktrees')
|
|
|
|
def test_create_skips_codegraph_when_missing(self):
|
|
self.ctl('init')
|
|
helper = self.root / 'scripts/agent-worktree.sh'
|
|
env = {**self.env, 'PATH': '/nonexistent'}
|
|
# Keep bash/git on a minimal path: prepend system essentials.
|
|
env['PATH'] = '/usr/bin:/bin'
|
|
# Hide only codegraph by putting a stub dir without it first... PATH without
|
|
# codegraph is enough if which looks at PATH. Ensure no codegraph on PATH.
|
|
result = self.run_cmd(
|
|
['bash', str(helper), 'create', 'feature/task-b'], env=env, timeout=60)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
out = json.loads(result.stdout)
|
|
self.assertEqual(out.get('codegraph'), 'not_installed')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|