From f9f055bee1e517591b2c2bab462f2cc91818dce8 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 24 Sep 2026 14:13:27 +0000 Subject: [PATCH 1/3] feat(agentctl): add governance console P1 serve and action APIs 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. --- .../agentctl-integrate-closeout/SKILL.md | 17 +- .../skills/agentctl-worktree-setup/SKILL.md | 19 +- .cursor/skills/pi-subagent-dispatch/SKILL.md | 14 +- .doc/ops/CURRENT.md | 1 + .../2026-09-24-governance-console-p1-impl.md | 35 ++ ...09-24-governance-console-p1-test-report.md | 68 ++ .../unittest-agentctl-suite.txt | 31 + .../unittest-serve.txt | 19 + scripts/agentctl.py | 53 +- scripts/agentctl_console.py | 591 ++++++++++++++++++ scripts/governance-console/index.html | 199 ++++++ scripts/mcp/agentctl-mcp.mjs | 41 ++ scripts/tests/test_agent_worktree.py | 2 +- scripts/tests/test_agentctl.py | 2 + scripts/tests/test_agentctl_catalog.py | 3 + scripts/tests/test_agentctl_serve.py | 366 +++++++++++ scripts/tests/test_codegraph_governance.py | 2 +- scripts/tests/test_ops_guards.py | 2 +- scripts/tests/test_pi_subagents.py | 3 + 19 files changed, 1440 insertions(+), 28 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-24-governance-console-p1-impl.md create mode 100644 .doc/test-reports/2026-09-24-governance-console-p1-test-report.md create mode 100644 .doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-agentctl-suite.txt create mode 100644 .doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-serve.txt create mode 100644 scripts/agentctl_console.py create mode 100644 scripts/governance-console/index.html create mode 100644 scripts/tests/test_agentctl_serve.py diff --git a/.cursor/skills/agentctl-integrate-closeout/SKILL.md b/.cursor/skills/agentctl-integrate-closeout/SKILL.md index 6b59080c..80608b53 100644 --- a/.cursor/skills/agentctl-integrate-closeout/SKILL.md +++ b/.cursor/skills/agentctl-integrate-closeout/SKILL.md @@ -9,19 +9,18 @@ description: Validate a merge candidate, enqueue for daemon ff-only integrate, a ## 候选 +优先一层封装（创建 integrate worktree、`task register`、`--no-ff` merge；**不** validate/enqueue）： + ```bash D=$(git -C /path/to/Agent-System rev-parse HEAD) # 冻结 expected-dev -ID=integrate--YYYY-MM-DD -git worktree add -b feature/$ID .worktrees/$ID "$D" -cd .worktrees/$ID -python3 ../../scripts/agentctl.py task register --id "$ID" --owner Cursor --scope . --capability git:dev -python3 ../../scripts/agentctl.py with --task "$ID" \ - --resource path:. --resource git:index --resource git:dev -- bash -lc ' - git merge --no-ff -m "merge: candidate into dev" feature/ -' +python3 scripts/agentctl.py integrate prepare \ + --task --expected-dev "$D" \ + [--integrate-id integrate--YYYY-MM-DD] +# 进入返回的 worktree 继续 validate → enqueue +cd .worktrees/integrate--YYYY-MM-DD # 或 emit 中的 worktree ``` -第一父必须等于 `$D`。 +第一父必须等于 `$D`。手工 `worktree add` + `with` + `merge --no-ff` 仍可用。 ## validate → enqueue（daemon ff） diff --git a/.cursor/skills/agentctl-worktree-setup/SKILL.md b/.cursor/skills/agentctl-worktree-setup/SKILL.md index 622a9c0f..cfb2a5bc 100644 --- a/.cursor/skills/agentctl-worktree-setup/SKILL.md +++ b/.cursor/skills/agentctl-worktree-setup/SKILL.md @@ -42,21 +42,18 @@ python3 ../../scripts/agentctl.py catalog ack \ --reason '<理由>' ``` -## CURRENT 认领（docs apply） +## CURRENT 认领 + +优先用一层封装（内部仍走 `docs apply` CAS）： ```bash -python3 ../../scripts/agentctl.py docs snapshot \ - --file .doc/ops/CURRENT.md --record "$ID" > /tmp/claim-snap.json -# 新建行：expected_sha256=null -cat > /tmp/claim.json < | $ID | feature/$ID | $WT | | in_progress |" -}]} -EOF -python3 ../../scripts/agentctl.py docs apply --task "$ID" --request /tmp/claim.json +python3 ../../scripts/agentctl.py claim new \ + --task "$ID" --owner --mutex \ + [--branch feature/$ID] [--worktree "$WT"] [--replace] ``` +手工 `docs snapshot` + `docs apply` 仍可用；行已存在且内容不同时需 `--replace`。 + ## short-term 模板字段 `status`、`write_scope`、`mutex`、`base_dev_sha`、`review_state`/`integration_state`/`deployment`/`cleanup`/`remote_sync`。命名 `.doc/ops/short-term/YYYY-MM-DD-.md`。 diff --git a/.cursor/skills/pi-subagent-dispatch/SKILL.md b/.cursor/skills/pi-subagent-dispatch/SKILL.md index 2088f1e2..0df59a6a 100644 --- a/.cursor/skills/pi-subagent-dispatch/SKILL.md +++ b/.cursor/skills/pi-subagent-dispatch/SKILL.md @@ -13,8 +13,18 @@ description: Dispatch gmktec PI analysis/review/test subagents via project MCP p 1. `pi_subagent_roles` — 只读预览角色/模型（不调 PI）。 2. 选 `role_id`（research → planning → reviews → diagnosis → test_run / acceptance_check → report）。 -3. `pi_subagent_run` — 在 agentctl 租约下异步启动。 -4. `pi_subagent_result` — 取回收据；轮询至终态。 +3. （推荐）`agentctl pi packet` 生成派发包；**不**触发 run： + ```bash + python3 -B scripts/agentctl.py pi packet \ + --task "$ID" --role research \ + --goal '…' --authorization-ref 'user:…' \ + --source-sha "$(git rev-parse HEAD)" \ + [--input-file …] [--approved-command '…']… \ + --request-id "$RID" --write + # → .doc/test-reports/assets/pi-subagents///packet.json + ``` +4. `pi_subagent_run` — 按 packet 字段在 agentctl 租约下异步启动。 +5. `pi_subagent_result` — 取回收据；轮询至终态。 CLI 等价：`python3 -B scripts/pi_subagents.py roles|run|result`；`run` 必须包在 `agentctl with --resource path:.doc/test-reports/assets/pi-subagents//` 内。 diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index eeae149e..2de341f1 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -18,6 +18,7 @@ | owner | task | branch | worktree | mutex | status | |---|---|---|---|---|---| +| Cursor | governance-console-p1-impl-2026-09-24 | feature/governance-console-p1-impl-2026-09-24 | .worktrees/governance-console-p1-impl-2026-09-24 | agentctl-console | in_progress | | Cursor | governance-console-p1-design-2026-09-24 | feature/governance-console-p1-design-2026-09-24 | .worktrees/governance-console-p1-design-2026-09-24 | agentctl-console-design | done | | Cursor | governance-visibility-p0-2026-09-24 | feature/governance-visibility-p0-2026-09-24 | .worktrees/governance-visibility-p0-2026-09-24 | agentctl-visibility | done | | Cursor | worktree-cleanup-closeout-2026-09-24 | feature/worktree-cleanup-closeout-2026-09-24 | .worktrees/worktree-cleanup-closeout-2026-09-24 | worktree-cleanup | in_progress | diff --git a/.doc/ops/short-term/2026-09-24-governance-console-p1-impl.md b/.doc/ops/short-term/2026-09-24-governance-console-p1-impl.md new file mode 100644 index 00000000..6764552f --- /dev/null +++ b/.doc/ops/short-term/2026-09-24-governance-console-p1-impl.md @@ -0,0 +1,35 @@ +# governance-console P1 实施 + +- task_id: `governance-console-p1-impl-2026-09-24` +- status: `in_progress` +- mutex: `agentctl-console` +- requirement: `R-governance-console-p1-impl` +- design: `.doc/ops/short-term/2026-09-24-governance-console-plan.md` (`governance-console-p1-r1`) +- authorization: 用户 `/goal 按照方案进行实现` +- write_scope: `scripts/` `.cursor/skills/` `.doc/` +- base_dev_sha: `6eb2d11eeb1840286b189c07e0da524541dcbbc8` +- review_state: pending +- integration_state: pending +- deployment: not_required +- cleanup: pending +- remote_sync: not_requested + +## 交付 + +| 阶段 | 产物 | +|---|---| +| P1-a | `agentctl serve` + `scripts/governance-console/index.html`；GET 八端点 | +| P1-b | `agentctl claim new` + `POST /api/claim/new` | +| P1-c | `agentctl integrate prepare` + `POST /api/integrate/prepare` | +| P1-d | `agentctl pi packet` + `POST /api/pi/packet`；skill 引用 packet | +| P1-e | SPA 四区+认领/Prepare；MCP `brief`/`claim_archive` | + +核心模块：`scripts/agentctl_console.py`（CLI/HTTP 同构）。 + +## 验证 + +见 [.doc/test-reports/2026-09-24-governance-console-p1-test-report.md](../../test-reports/2026-09-24-governance-console-p1-test-report.md)。 + +## 禁止 + +不 push；不碰产品 `web/`；不自动 `queue enqueue` 以外的 ff；不实现 `/api/shell`。 diff --git a/.doc/test-reports/2026-09-24-governance-console-p1-test-report.md b/.doc/test-reports/2026-09-24-governance-console-p1-test-report.md new file mode 100644 index 00000000..8cfe0d61 --- /dev/null +++ b/.doc/test-reports/2026-09-24-governance-console-p1-test-report.md @@ -0,0 +1,68 @@ +# governance-console P1 测试报告 + +## 一、 测试背景 (Background) + +P0 已交付 `collect_*` / `brief` / `claim archive` / table 渲染。本任务按定稿规格 `governance-console-p1-r1`（`.doc/ops/short-term/2026-09-24-governance-console-plan.md`）实施 P1-a…e：loopback HTTP console、高频动作白名单与 SPA。 + +## 二、 测试目标 (Test Objectives) + +- `agentctl serve` 只读八端点 + SPA；非 loopback 拒绝；未知 `/api/shell` → 404 +- `claim new` CLI/HTTP：新建、幂等、缺能力 403、缺 task 400 +- `integrate prepare`：`first_parent == expected-dev`；脏 worktree 拒绝；不自动 enqueue +- `pi packet`：拒绝符号 `source_sha`；`--write` 落盘且不触发 run +- 工程工具层回归：`test_agentctl*.py` 全绿 + +## 三、 测试方案与环境 (Test Plan & Environment) + +- OS：Linux；Python 3.14；隔离临时 git 仓 fixture +- 验证层级：验证契约「工程工具验收」（非产品 PI） +- 套件：`scripts/tests/test_agentctl_serve.py` + `unittest discover -p 'test_agentctl*.py'` +- 真机冒烟：worktree 内 `agentctl serve` + curl `/api/status` `/api/brief` `/` + +## 四、 预期结果 (Expected Results) + +见方案 §9.1–9.2 矩阵。 + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 `test_agentctl_serve.py`（8） + +| 用例 | 判定 | +|---|---| +| non_loopback_host_denied | PASS | +| get_eight_endpoints_and_spa | PASS | +| unknown_post_shell_404 | PASS | +| claim_new_missing_task_400 | PASS | +| claim_new_without_docs_publish_403 | PASS | +| claim_new_cli_create_replace_and_replay | PASS | +| integrate_prepare_first_parent_and_dirty_refuse | PASS | +| pi_packet_rejects_symbolic_and_writes | PASS | + +证据：`.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-serve.txt` + +### 5.2 `test_agentctl*.py` 套件 + +- **输入**：`python3 -B -m unittest discover -s scripts/tests -p 'test_agentctl*.py' -v` +- **实际输出**：`Ran 110 tests … OK`（约 83s） +- **判定**：PASS +- 证据：`.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-agentctl-suite.txt` + +### 5.3 真机 serve 冒烟 + +- `/api/status` → 含 `tasks`/`main` +- `/api/brief` → markdown 以 `# agentctl brief` 开头 +- `/` → 200 HTML（worktree 静态回退） +- `POST /api/shell` → `NOT_FOUND` + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 套件 | 结果 | +|---|---| +| test_agentctl_serve | 8/8 PASS | +| test_agentctl* discover | 110/110 PASS | + +产品 PI / `go test`：**NOT REQUIRED**（工程工具变更，依据验证契约与 PROJECT 验证入口）。 + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +P1-a…e 验收达标，可合入 local_dev。后续可选：SSE 推送、Enqueue 二次确认写入口（当前故意仅 CLI 指引）、浏览器手动四区清单勾选。 diff --git a/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-agentctl-suite.txt b/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-agentctl-suite.txt new file mode 100644 index 00000000..61e80290 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-agentctl-suite.txt @@ -0,0 +1,31 @@ +..................................................../home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/asyncio/events.py:706: ResourceWarning: unclosed database in + class _Local(threading.local): +ResourceWarning: Enable tracemalloc to get the object allocation traceback +/home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/asyncio/events.py:706: ResourceWarning: unclosed database in + class _Local(threading.local): +ResourceWarning: Enable tracemalloc to get the object allocation traceback +/home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/asyncio/events.py:706: ResourceWarning: unclosed database in + class _Local(threading.local): +ResourceWarning: Enable tracemalloc to get the object allocation traceback +/home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/asyncio/events.py:706: ResourceWarning: unclosed database in + class _Local(threading.local): +ResourceWarning: Enable tracemalloc to get the object allocation traceback +/home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/asyncio/events.py:706: ResourceWarning: unclosed database in + class _Local(threading.local): +ResourceWarning: Enable tracemalloc to get the object allocation traceback +{"demoted": [{"from": "running", "run_id": "hb-stale", "to": "suspect"}, {"from": "suspect", "run_id": "sus-stale", "to": "quarantine"}], "heartbeat_ttl_seconds": 300, "status": "ok", "suspect_ttl_seconds": 600} +.{"demoted": [{"from": "running", "run_id": "run-unknown", "to": "suspect"}], "heartbeat_ttl_seconds": 300, "status": "ok", "suspect_ttl_seconds": 600} +...usage: python3.14 -m unittest [-h] [--repo REPO] + {init,status,doctor,brief,serve,task,with,guard,worktree-check,docs,queue,cleanup,recover,operation,freeze,unfreeze,hooks,validate,integrate,deploy,claim,quarantine,daemon,pi,catalog} ... +python3.14 -m unittest: error: unrecognized arguments: --head-sha deadbeef +.........................../home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +./home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +...../home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +..................... +---------------------------------------------------------------------- +Ran 110 tests in 82.892s + +OK diff --git a/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-serve.txt b/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-serve.txt new file mode 100644 index 00000000..5ed8219e --- /dev/null +++ b/.doc/test-reports/assets/2026-09-24-governance-console-p1/unittest-serve.txt @@ -0,0 +1,19 @@ +test_claim_new_cli_create_replace_and_replay (test_agentctl_serve.ConsoleServeTests.test_claim_new_cli_create_replace_and_replay) ... ok +test_claim_new_missing_task_400 (test_agentctl_serve.ConsoleServeTests.test_claim_new_missing_task_400) ... /home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +ok +test_claim_new_without_docs_publish_403 (test_agentctl_serve.ConsoleServeTests.test_claim_new_without_docs_publish_403) ... /home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +ok +test_get_eight_endpoints_and_spa (test_agentctl_serve.ConsoleServeTests.test_get_eight_endpoints_and_spa) ... ok +test_integrate_prepare_first_parent_and_dirty_refuse (test_agentctl_serve.ConsoleServeTests.test_integrate_prepare_first_parent_and_dirty_refuse) ... ok +test_non_loopback_host_denied (test_agentctl_serve.ConsoleServeTests.test_non_loopback_host_denied) ... ok +test_pi_packet_rejects_symbolic_and_writes (test_agentctl_serve.ConsoleServeTests.test_pi_packet_rejects_symbolic_and_writes) ... ok +test_unknown_post_shell_404 (test_agentctl_serve.ConsoleServeTests.test_unknown_post_shell_404) ... /home/linuxbrew/.linuxbrew/Cellar/python@3.14/3.14.7/lib/python3.14/tempfile.py:484: ResourceWarning: Implicitly cleaning up + _warnings.warn(self.warn_message, ResourceWarning) +ok + +---------------------------------------------------------------------- +Ran 8 tests in 5.459s + +OK diff --git a/scripts/agentctl.py b/scripts/agentctl.py index 5653c7e4..6ddc333f 100755 --- a/scripts/agentctl.py +++ b/scripts/agentctl.py @@ -28,6 +28,7 @@ os.environ.setdefault('PYTHONDONTWRITEBYTECODE', '1') import agentctl_w4 as w4 # noqa: E402 import agentctl_catalog as catalog # noqa: E402 import agentctl_visibility as visibility # noqa: E402 +import agentctl_console as console # noqa: E402 VERSION = 1 POLICY_REL = 'configs/agent-governance.json' @@ -2972,6 +2973,10 @@ def parser(): sp.add_argument('--format',choices=['json','table'],default='json') brief=sub.add_parser('brief', help='One-page governance briefing (Markdown)') brief.add_argument('--format',choices=['markdown','json'],default='markdown') + srv=sub.add_parser('serve', help='Loopback governance HTTP console') + srv.add_argument('--host', default='127.0.0.1') + srv.add_argument('--port', type=int, default=7432) + srv.add_argument('--static', default='scripts/governance-console') task=sub.add_parser('task').add_subparsers(dest='task_action',required=True) reg=task.add_parser('register');reg.add_argument('--id',required=True);reg.add_argument('--owner',required=True);reg.add_argument('--scope',action='append',required=True);reg.add_argument('--capability',action='append',default=[]) reg.add_argument('--requirement',help='Catalog R-ID to bind at register') @@ -3013,7 +3018,16 @@ def parser(): hooks.add_parser('install');hc=hooks.add_parser('check');hc.add_argument('--merge',action='store_true');hr=hooks.add_parser('ref');hr.add_argument('state') hooks.add_parser('codegraph-sync', help='Fail-open codegraph sync for post-merge/checkout') v=sub.add_parser('validate');v.add_argument('--task',required=True);v.add_argument('command',nargs=argparse.REMAINDER) - i=sub.add_parser('integrate');i.add_argument('--task',required=True);i.add_argument('--expected-dev',required=True);i.add_argument('--request-id') + i=sub.add_parser('integrate') + i.add_argument('--task') + i.add_argument('--expected-dev') + i.add_argument('--request-id') + i_sub=i.add_subparsers(dest='integrate_action') + ip=i_sub.add_parser('prepare', help='Build integrate candidate worktree + no-ff merge') + ip.add_argument('--task',required=True) + ip.add_argument('--expected-dev',required=True) + ip.add_argument('--integrate-id') + ip.add_argument('--request-id') dep=sub.add_parser('deploy').add_subparsers(dest='deploy_action',required=True) dr=dep.add_parser('run-loop') dr.add_argument('--task',required=True);dr.add_argument('--id',required=True) @@ -3035,6 +3049,15 @@ def parser(): carc.add_argument('--retain-days',type=int,default=7) carc.add_argument('--include-undated',action='store_true') carc.add_argument('--apply',action='store_true') + cnew=cl.add_parser('new', help='Create CURRENT claim row via docs apply') + cnew.add_argument('--task',required=True) + cnew.add_argument('--owner') + cnew.add_argument('--mutex',default='-') + cnew.add_argument('--branch') + cnew.add_argument('--worktree') + cnew.add_argument('--status',default='in_progress') + cnew.add_argument('--replace',action='store_true') + cnew.add_argument('--request-id') cc=cl.add_parser('cancel');cc.add_argument('--task',required=True);cc.add_argument('--id',required=True);cc.add_argument('--reason',required=True) qtn=sub.add_parser('quarantine').add_subparsers(dest='quarantine_action',required=True) qtn.add_parser('sweep') @@ -3049,6 +3072,18 @@ def parser(): drun.add_argument('--audit-every',type=int,default=30,dest='audit_every', help='Every N cycles run cleanup report + docs drift (0=off)') dm.add_parser('status') + pi_root=sub.add_parser('pi').add_subparsers(dest='pi_action', required=True) + pp=pi_root.add_parser('packet', help='Generate PI dispatch packet JSON') + pp.add_argument('--task',required=True) + pp.add_argument('--role',required=True) + pp.add_argument('--goal',required=True) + pp.add_argument('--authorization-ref',required=True) + pp.add_argument('--source-sha') + pp.add_argument('--timeout-seconds',type=int,default=600) + pp.add_argument('--input-file',action='append',default=[]) + pp.add_argument('--approved-command',action='append',default=[]) + pp.add_argument('--request-id') + pp.add_argument('--write',action='store_true') cat = sub.add_parser('catalog').add_subparsers(dest='catalog_action', required=True) cadd = cat.add_parser('add') cadd.add_argument('--id', required=True) @@ -3087,7 +3122,7 @@ def main(): write_actions = { 'init', 'task', 'with', 'docs', 'validate', 'integrate', 'queue', 'deploy', 'recover', 'freeze', 'unfreeze', 'hooks', 'guard', 'cleanup', 'claim', - 'quarantine', 'daemon', 'catalog', + 'quarantine', 'daemon', 'catalog', 'serve', 'pi', } # status/doctor remain readable without a policy file and never create one. if args.action in write_actions: @@ -3123,6 +3158,11 @@ def main(): emit_payload(payload, getattr(args, 'format', 'json'), visibility.render_status_table) elif args.action=='brief': return cmd_brief(repo, args) + elif args.action=='serve': + return console.serve(repo, args) + elif args.action=='pi': + if args.pi_action=='packet': + emit(console.pi_packet(repo, args)); return 0 elif args.action=='task': if args.task_action=='register': if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}',args.id):deny('INVALID_TASK','Invalid task id') @@ -3221,7 +3261,12 @@ def main(): elif args.hook_action=='codegraph-sync':return hooks_codegraph_sync(repo) else:hook_ref(repo,args.state) elif args.action=='validate':return validate(repo,args) - elif args.action=='integrate':return integrate(repo,args) + elif args.action=='integrate': + if getattr(args, 'integrate_action', None)=='prepare': + emit(console.integrate_prepare(repo, args)); return 0 + if not getattr(args, 'task', None) or not getattr(args, 'expected_dev', None): + deny('INVALID_REQUEST', 'integrate requires --task and --expected-dev (or: integrate prepare)') + return integrate(repo, args) elif args.action=='deploy':return deploy_cmd(repo,args) elif args.action=='claim': if args.claim_action=='acquire':return claim_acquire(repo,args) @@ -3229,6 +3274,8 @@ def main(): if args.claim_action=='status':return claim_status(repo,args) if args.claim_action=='list':return claim_list(repo,args) if args.claim_action=='archive':return claim_archive(repo,args) + if args.claim_action=='new': + emit(console.claim_new(repo,args)); return 0 if args.claim_action=='cancel':return claim_cancel(repo,args) elif args.action=='quarantine': if args.quarantine_action=='sweep':return quarantine_sweep(repo) diff --git a/scripts/agentctl_console.py b/scripts/agentctl_console.py new file mode 100644 index 00000000..a2461a92 --- /dev/null +++ b/scripts/agentctl_console.py @@ -0,0 +1,591 @@ +#!/usr/bin/env python3 +"""Governance console helpers: claim_new, integrate_prepare, pi_packet, serve. + +Shared by CLI and HTTP (same-process). Raises agentctl.Denied on refusal. +Returns payload dicts; CLI wrappers call emit(). +""" +from __future__ import annotations + +import argparse +import contextlib +import json +import mimetypes +import re +import tempfile +import threading +import time +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Optional +from urllib.parse import parse_qs, urlparse + +import agentctl as ac +import agentctl_visibility as visibility + +try: + import pi_subagents as pi +except Exception: # noqa: BLE001 — optional for environments without full PI deps + pi = None + +CURRENT_REL = '.doc/ops/CURRENT.md' +DEFAULT_HOST = '127.0.0.1' +DEFAULT_PORT = 7432 +DEFAULT_STATIC = 'scripts/governance-console' +FORBIDDEN_BODY_KEYS = frozenset({'command', 'shell', 'argv'}) + +HTTP_STATUS = { + 'INVALID_REQUEST': 400, + 'CAPABILITY_REQUIRED': 403, + 'PATH_OUT_OF_SCOPE': 403, + 'BASE_CHANGED': 409, + 'WORKTREE_MISMATCH': 409, + 'LOCK_BUSY': 409, + 'UNMANAGED_DOCUMENT_CHANGE': 409, + 'DIRTY_TREE': 409, + 'FROZEN': 409, + 'EVIDENCE_REQUIRED': 422, + 'REQUEST_ID_REUSED': 409, + 'RECOVERY_REQUIRED': 500, +} + + +def _caps(task) -> list: + raw = task['capabilities'] + return json.loads(raw) if isinstance(raw, str) else list(raw or []) + + +def _rel_worktree(path: Path, main: Path) -> str: + try: + return str(path.resolve().relative_to(main.resolve())) + except ValueError: + text = str(path) + if '/.worktrees/' in text.replace('\\', '/'): + return '.worktrees/' + text.replace('\\', '/').split('/.worktrees/', 1)[1] + return text + + +def claim_new(repo: ac.Repo, args) -> dict: + """Create or replace a CURRENT claim row via single-row docs apply.""" + task = ac.task_row(repo, args.task) + if repo.current != Path(task['worktree']) or repo.current == repo.main: + ac.deny('WORKTREE_MISMATCH', 'claim new must run inside the task worktree') + if 'docs:publish' not in _caps(task) and 'docs:admin' not in _caps(task): + ac.deny('CAPABILITY_REQUIRED', 'claim new requires docs:publish') + owner = getattr(args, 'owner', None) or task.get('owner') or 'Cursor' + branch = getattr(args, 'branch', None) or task.get('branch') or f"feature/{args.task}" + worktree = getattr(args, 'worktree', None) or _rel_worktree(Path(task['worktree']), repo.main) + mutex = getattr(args, 'mutex', None) or '-' + status = getattr(args, 'status', None) or 'in_progress' + if status not in ac.PROJECTION_STATUS_VOCAB: + ac.deny('INVALID_REQUEST', 'status must be a projection vocab value') + content = ac.safe_path(repo.current, CURRENT_REL).read_text(encoding='utf-8') + existing = ac.doc_rows(content, args.task) + expected = ac.digest(existing[0][1]) if existing else None + replacement = ( + f"| {owner} | {args.task} | {branch} | {worktree} | {mutex} | {status} |" + ) + if existing and existing[0][1].strip() == replacement.strip(): + return { + 'status': 'idempotent', + 'record_id': args.task, + 'request_id': getattr(args, 'request_id', None), + 'row': replacement, + } + if existing and not getattr(args, 'replace', False): + ac.deny('INVALID_REQUEST', 'row exists; pass --replace') + request_id = getattr(args, 'request_id', None) or f"{args.task}-claim-new-{uuid.uuid4().hex[:8]}" + request = { + 'request_id': request_id, + 'changes': [{ + 'file': CURRENT_REL, + 'record_id': args.task, + 'expected_sha256': expected, + 'replacement': replacement, + }], + } + req_path = Path(tempfile.mkstemp(prefix='agentctl-claim-new-', suffix='.json')[1]) + try: + req_path.write_text(json.dumps(request, ensure_ascii=False), encoding='utf-8') + captured = [] + with _capture_emit(captured): + ac.docs_apply(repo, argparse.Namespace(task=args.task, request=str(req_path))) + applied = next((c for c in captured if isinstance(c, dict)), {}) + status = applied.get('status') + if status == 'replayed': + out_status = 'replayed' + else: + out_status = 'replaced' if existing else 'created' + return { + 'status': out_status, + 'record_id': args.task, + 'request_id': request_id, + 'run_id': applied.get('run_id'), + 'row': replacement, + } + finally: + with contextlib.suppress(OSError): + req_path.unlink() + + +def integrate_prepare(repo: ac.Repo, args) -> dict: + """Build integrate candidate worktree and --no-ff merge task branch.""" + task = ac.task_row(repo, args.task) + expected = getattr(args, 'expected_dev', None) + if not expected or not re.fullmatch(r'[0-9a-f]{40}|[0-9a-f]{64}', expected): + ac.deny('INVALID_REQUEST', 'expected-dev must be a full commit hash') + tip = ac.git(repo.main, 'rev-parse', 'refs/heads/dev') + if tip != expected: + ac.deny('BASE_CHANGED', 'refs/heads/dev tip differs from expected-dev', + tip=tip, expected_dev=expected) + integrate_id = getattr(args, 'integrate_id', None) or f"integrate-{args.task}" + if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}', integrate_id): + ac.deny('INVALID_REQUEST', 'Invalid integrate-id') + branch = f'feature/{integrate_id}' + integ_path = repo.main / '.worktrees' / integrate_id + task_branch = task['branch'] + if not task_branch: + ac.deny('INVALID_REQUEST', 'Source task has no branch') + + if integ_path.exists(): + integ_repo = ac.Repo(str(integ_path)) + head = ac.git(integ_path, 'rev-parse', 'HEAD') + first = ac.git(integ_path, 'rev-parse', 'HEAD^1', check=False) + if first.returncode == 0 and first.stdout.strip() == expected: + # Already prepared on this base. + return { + 'status': 'prepared', + 'integrate_task': integrate_id, + 'worktree': str(integ_path), + 'head': head, + 'expected_dev': expected, + 'first_parent': expected, + 'idempotent': True, + } + porcelain = ac.git(integ_path, 'status', '--porcelain', '--untracked-files=all') + if porcelain: + ac.deny('INVALID_REQUEST', 'Existing integrate worktree is dirty', + path=str(integ_path)) + # Different base: refuse rather than reset. + ac.deny('BASE_CHANGED', 'Integrate worktree exists for a different base; remove it first', + path=str(integ_path), head=head) + else: + integ_path.parent.mkdir(exist_ok=True) + ac.git(repo.main, 'worktree', 'add', '-b', branch, str(integ_path), expected) + + # Register integrate task (must run as checkout identity). + _register_integrate_task(integ_path, integrate_id, owner=task.get('owner') or 'Cursor') + integ_repo = ac.Repo(str(integ_path)) + integ_task = ac.task_row(integ_repo, integrate_id) + request_id = getattr(args, 'request_id', None) or f"{integrate_id}-prepare" + msg = f"merge: candidate {args.task} into dev" + with ac.Lease(integ_repo, integ_task, ['path:.', 'git:index', 'git:dev'], + request_id, ac.digest(msg + expected + task_branch), kind='integrate_prepare') as lease: + if lease.replay: + head = ac.git(integ_path, 'rev-parse', 'HEAD') + return { + 'status': 'replayed', + 'run_id': lease.request_id, + 'integrate_task': integrate_id, + 'worktree': str(integ_path), + 'head': head, + 'expected_dev': expected, + } + result = ac.git(integ_path, 'merge', '--no-ff', '-m', msg, task_branch, check=False) + if result.returncode != 0: + with contextlib.suppress(Exception): + ac.git(integ_path, 'merge', '--abort', check=False) + lease.finish(1, True) + ac.deny('GIT_ERROR', 'merge --no-ff failed', stderr=(result.stderr or '')[:500]) + head = ac.git(integ_path, 'rev-parse', 'HEAD') + first_parent = ac.git(integ_path, 'rev-parse', 'HEAD^1') + if first_parent != expected: + lease.finish(1, True) + ac.deny('BASE_CHANGED', 'first parent is not expected-dev', + first_parent=first_parent, expected_dev=expected) + lease.finish() + return { + 'status': 'prepared', + 'integrate_task': integrate_id, + 'worktree': str(integ_path), + 'head': head, + 'expected_dev': expected, + 'first_parent': first_parent, + 'run_id': request_id, + 'idempotent': False, + } + + +def _register_integrate_task(integ_path: Path, integrate_id: str, owner: str): + """Register integrate task from that checkout (subprocess avoids cwd gymnastics).""" + import subprocess + import sys + agentctl = Path(__file__).resolve() + cmd = [ + sys.executable, '-B', str(agentctl.parent / 'agentctl.py'), + 'task', 'register', + '--id', integrate_id, '--owner', owner, + '--scope', '.', + '--capability', 'git:dev', + '--capability', 'docs:publish', + ] + proc = subprocess.run(cmd, cwd=str(integ_path), capture_output=True, text=True) + if proc.returncode != 0: + # Already registered with same identity is ok if output says registered/exists matching. + err = (proc.stderr or proc.stdout or '').strip() + try: + payload = json.loads(err.splitlines()[-1]) + except Exception: + payload = {} + if payload.get('code') == 'TASK_EXISTS': + return + if payload.get('status') == 'registered': + return + ac.deny('INVALID_REQUEST', 'Failed to register integrate task', detail=err[:500]) + + +def pi_packet(repo: ac.Repo, args) -> dict: + """Build PI dispatch packet JSON; optional write under report assets.""" + task = ac.task_row(repo, args.task) + wt = Path(task['worktree']) + role = getattr(args, 'role', None) or getattr(args, 'role_id', None) + known = getattr(pi, 'KNOWN_ROLES', None) if pi else None + if not role: + ac.deny('INVALID_REQUEST', 'role is required') + if known is not None and role not in known: + ac.deny('UNKNOWN_ROLE', 'Unknown role_id', role_id=role) + goal = getattr(args, 'goal', None) + auth = getattr(args, 'authorization_ref', None) + if not goal or not auth: + ac.deny('INVALID_REQUEST', 'goal and authorization-ref are required') + raw_sha = getattr(args, 'source_sha', None) + if not raw_sha: + raw_sha = ac.git(wt, 'rev-parse', 'HEAD') + if pi is not None: + try: + source_sha = pi.freeze_source_sha(wt, raw_sha) + except Exception as exc: # ExecutorError + code = getattr(exc, 'code', 'INVALID_SOURCE_SHA') + ac.deny(code, getattr(exc, 'message', str(exc)), **getattr(exc, 'details', {})) + else: + if not re.fullmatch(r'[0-9a-f]{40}|[0-9a-f]{64}', raw_sha): + ac.deny('INVALID_SOURCE_SHA', 'source_sha must be a full commit hash') + source_sha = raw_sha + request_id = getattr(args, 'request_id', None) or str(uuid.uuid4()) + rid_re = getattr(pi, 'REQUEST_ID_RE', re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$')) + if not rid_re.fullmatch(request_id): + ac.deny('INVALID_REQUEST', 'Invalid request_id') + input_files = list(getattr(args, 'input_file', None) or getattr(args, 'input_files', None) or []) + approved = [] + for item in (getattr(args, 'approved_command', None) or []): + if isinstance(item, list): + approved.append(item) + elif isinstance(item, str): + approved.append(item.split()) + else: + ac.deny('INVALID_REQUEST', 'approved-command must be string or list') + timeout = int(getattr(args, 'timeout_seconds', None) or 600) + packet = { + 'task_id': args.task, + 'request_id': request_id, + 'role_id': role, + 'goal': goal, + 'authorization_ref': auth, + 'source_sha': source_sha, + 'input_files': input_files, + 'timeout_seconds': timeout, + 'approved_commands': approved, + } + wrote = None + if getattr(args, 'write', False): + rel_root = getattr(pi, 'REPORT_ROOT_REL', '.doc/test-reports/assets/pi-subagents') + rel = f"{rel_root}/{args.task}/{request_id}/packet.json" + # Write under task worktree with lease. + if repo.current != wt: + ac.deny('WORKTREE_MISMATCH', 'pi packet --write must run in the task worktree') + dest = ac.safe_path(repo.current, rel) + dest.parent.mkdir(parents=True, exist_ok=True) + body = json.dumps(packet, ensure_ascii=False, indent=2) + '\n' + with ac.Lease(repo, task, [f'path:{rel}'], f"{args.task}-pi-packet-{request_id}", + ac.digest(body), kind='pi_packet') as lease: + if lease.replay: + return {'status': 'replayed', 'run_id': lease.request_id, 'packet': packet, 'path': rel} + ac.atomic_write(dest, body) + lease.finish() + wrote = rel + return {'status': 'ok', 'packet': packet, 'path': wrote} + + +@contextlib.contextmanager +def _capture_emit(bucket: list): + original = ac.emit + + def _cap(value, stream=None): + bucket.append(value) + + ac.emit = _cap + try: + yield bucket + finally: + ac.emit = original + + +def denied_http_status(code: str) -> int: + if code in HTTP_STATUS: + return HTTP_STATUS[code] + if code.startswith('UNKNOWN_'): + return 404 + return 500 + + +def build_brief_text(repo: ac.Repo) -> str: + status = ac.collect_status_payload(repo) + try: + docs = ac.collect_docs_drift_payload(repo) + except Exception as exc: # noqa: BLE001 + docs = {'status': 'error', 'error': str(exc)} + try: + cleanup = ac.collect_cleanup_payload(repo) + except Exception as exc: # noqa: BLE001 + cleanup = {'status': 'error', 'error': str(exc), 'worktrees': []} + try: + cat = ac.collect_catalog_drift_payload(repo) + except Exception as exc: # noqa: BLE001 + cat = {'status': 'error', 'error': str(exc)} + try: + daemon = ac.collect_daemon_payload(repo) + except Exception as exc: # noqa: BLE001 + daemon = {'status': 'error', 'error': str(exc)} + return visibility.build_brief( + status=status, docs_drift=docs, cleanup=cleanup, + catalog_drift=cat, daemon=daemon, + ) + + +def _queue_payload(repo: ac.Repo, all_items: bool = False) -> dict: + states = None if all_items else ('pending', 'claimed') + with repo.db(True) as db: + if states: + rows = db.execute( + "SELECT * FROM queue_items WHERE state IN (?,?) ORDER BY priority DESC, created ASC", + states).fetchall() + else: + rows = db.execute('SELECT * FROM queue_items ORDER BY created ASC').fetchall() + return {'status': 'ok', 'items': [ac.queue_row(r) for r in rows]} + + +def _claims_payload(repo: ac.Repo, all_items: bool = False) -> dict: + with repo.db() as db: + try: + if all_items: + rows = [dict(r) for r in db.execute( + 'SELECT * FROM claims ORDER BY updated DESC').fetchall()] + else: + rows = [dict(r) for r in db.execute( + "SELECT * FROM claims WHERE state='held' ORDER BY updated DESC").fetchall()] + except Exception: + rows = [] + return {'status': 'ok', 'claims': rows} + + +def serve(repo: ac.Repo, args) -> int: + host = getattr(args, 'host', None) or DEFAULT_HOST + port = int(getattr(args, 'port', None) or DEFAULT_PORT) + if host not in ('127.0.0.1', 'localhost', '::1'): + ac.deny('INVALID_REQUEST', 'serve host must be loopback (127.0.0.1)') + static_rel = getattr(args, 'static', None) or DEFAULT_STATIC + static_root = (repo.main / static_rel).resolve() + if not str(static_root).startswith(str(repo.main.resolve())): + ac.deny('PATH_OUT_OF_SCOPE', 'static root must stay under main worktree') + # Prefer main tip; if not yet integrated, allow the same relative path from + # the current checkout when it remains under the main worktree tree. + if not (static_root / 'index.html').is_file(): + alt = (repo.current / static_rel).resolve() + if str(alt).startswith(str(repo.main.resolve())) and (alt / 'index.html').is_file(): + static_root = alt + + class Handler(BaseHTTPRequestHandler): + def log_message(self, fmt, *a): + pass + + def _send(self, code: int, body: Any, content_type='application/json; charset=utf-8'): + if isinstance(body, (dict, list)): + data = json.dumps(body, ensure_ascii=False, sort_keys=True).encode('utf-8') + elif isinstance(body, str): + data = body.encode('utf-8') + else: + data = body + self.send_response(code) + self.send_header('Content-Type', content_type) + self.send_header('Content-Length', str(len(data))) + self.send_header('Cache-Control', 'no-store') + self.end_headers() + self.wfile.write(data) + + def _read_json(self) -> dict: + length = int(self.headers.get('Content-Length') or 0) + if length > 1_000_000: + ac.deny('INVALID_REQUEST', 'request body too large') + raw = self.rfile.read(length) if length else b'{}' + try: + data = json.loads(raw.decode('utf-8') or '{}') + except Exception: + ac.deny('INVALID_REQUEST', 'JSON body required') + if not isinstance(data, dict): + ac.deny('INVALID_REQUEST', 'JSON object required') + if FORBIDDEN_BODY_KEYS & set(data): + ac.deny('INVALID_REQUEST', 'command/shell/argv fields are forbidden') + return data + + def do_GET(self): # noqa: N802 + try: + parsed = urlparse(self.path) + path = parsed.path + qs = parse_qs(parsed.query) + fmt = (qs.get('format') or ['json'])[0] + all_items = (qs.get('all') or ['0'])[0] in ('1', 'true', 'yes') + if path == '/api/status': + payload = ac.collect_status_payload(repo) + if fmt == 'table': + return self._send(200, visibility.render_status_table(payload), 'text/plain; charset=utf-8') + return self._send(200, payload) + if path == '/api/brief': + brief_fmt = (qs.get('format') or ['markdown'])[0] + text = build_brief_text(repo) + if brief_fmt == 'json': + return self._send(200, {'status': 'ok', 'brief': text}) + return self._send(200, text, 'text/markdown; charset=utf-8') + if path == '/api/queue': + payload = _queue_payload(repo, all_items) + if fmt == 'table': + return self._send(200, visibility.render_queue_table(payload), 'text/plain; charset=utf-8') + return self._send(200, payload) + if path == '/api/claims': + payload = _claims_payload(repo, all_items) + if fmt == 'table': + return self._send(200, visibility.render_claim_table(payload), 'text/plain; charset=utf-8') + return self._send(200, payload) + if path == '/api/cleanup': + payload = ac.collect_cleanup_payload(repo) + if fmt == 'table': + return self._send(200, visibility.render_cleanup_table(payload), 'text/plain; charset=utf-8') + return self._send(200, payload) + if path == '/api/docs/drift': + return self._send(200, ac.collect_docs_drift_payload(repo)) + if path == '/api/catalog/drift': + return self._send(200, ac.collect_catalog_drift_payload(repo)) + if path == '/api/daemon': + return self._send(200, ac.collect_daemon_payload(repo)) + if path in ('/', '/index.html'): + return self._static('index.html') + if path.startswith('/static/'): + return self._static(path[len('/static/'):]) + return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': path}) + except ac.Denied as exc: + return self._send(denied_http_status(exc.code), { + 'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details, + }) + except Exception as exc: # noqa: BLE001 + return self._send(500, { + 'status': 'denied', 'code': 'RECOVERY_REQUIRED', 'message': str(exc)[:300], + }) + + def do_POST(self): # noqa: N802 + try: + path = urlparse(self.path).path + body = self._read_json() + if path == '/api/claim/new': + ns = argparse.Namespace( + task=body.get('task'), + owner=body.get('owner'), + mutex=body.get('mutex', '-'), + branch=body.get('branch'), + worktree=body.get('worktree'), + status=body.get('status', 'in_progress'), + replace=bool(body.get('replace')), + request_id=body.get('request_id'), + ) + if not ns.task: + ac.deny('INVALID_REQUEST', 'task required') + # claim_new must run in task worktree + task = ac.task_row(repo, ns.task) + task_repo = ac.Repo(task['worktree']) + return self._send(200, claim_new(task_repo, ns)) + if path == '/api/claim/archive': + ns = argparse.Namespace( + task=body.get('task'), + retain_days=int(body.get('retain_days', 7)), + include_undated=bool(body.get('include_undated')), + apply=True, + ) + if not ns.task: + ac.deny('INVALID_REQUEST', 'task required') + task = ac.task_row(repo, ns.task) + task_repo = ac.Repo(task['worktree']) + captured = [] + with _capture_emit(captured): + ac.claim_archive(task_repo, ns) + return self._send(200, captured[-1] if captured else {'status': 'ok'}) + if path == '/api/integrate/prepare': + ns = argparse.Namespace( + task=body.get('task'), + expected_dev=body.get('expected_dev'), + integrate_id=body.get('integrate_id'), + request_id=body.get('request_id'), + ) + if not ns.task or not ns.expected_dev: + ac.deny('INVALID_REQUEST', 'task and expected_dev required') + return self._send(200, integrate_prepare(repo, ns)) + if path == '/api/pi/packet': + ns = argparse.Namespace( + task=body.get('task'), + role=body.get('role') or body.get('role_id'), + goal=body.get('goal'), + authorization_ref=body.get('authorization_ref'), + source_sha=body.get('source_sha'), + timeout_seconds=body.get('timeout_seconds', 600), + input_file=body.get('input_files') or [], + approved_command=body.get('approved_commands') or [], + request_id=body.get('request_id'), + write=bool(body.get('write')), + ) + if not ns.task: + ac.deny('INVALID_REQUEST', 'task required') + task = ac.task_row(repo, ns.task) + task_repo = ac.Repo(task['worktree']) if ns.write else repo + return self._send(200, pi_packet(task_repo, ns)) + return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': path}) + except ac.Denied as exc: + return self._send(denied_http_status(exc.code), { + 'status': 'denied', 'code': exc.code, 'message': exc.message, **exc.details, + }) + except Exception as exc: # noqa: BLE001 + return self._send(500, { + 'status': 'denied', 'code': 'RECOVERY_REQUIRED', 'message': str(exc)[:300], + }) + + def _static(self, name: str): + name = name.lstrip('/') + if '..' in name or name.startswith('/'): + return self._send(400, {'status': 'denied', 'code': 'INVALID_REQUEST', 'message': 'bad path'}) + target = (static_root / name).resolve() + if not str(target).startswith(str(static_root)) or not target.is_file(): + return self._send(404, {'status': 'denied', 'code': 'NOT_FOUND', 'message': name}) + ctype = mimetypes.guess_type(str(target))[0] or 'application/octet-stream' + return self._send(200, target.read_bytes(), ctype) + + server = ThreadingHTTPServer((host, port), Handler) + ac.emit({ + 'status': 'serving', + 'host': host, + 'port': port, + 'static': str(static_root), + 'url': f'http://{host}:{port}/', + }) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + return 0 diff --git a/scripts/governance-console/index.html b/scripts/governance-console/index.html new file mode 100644 index 00000000..37e64fe8 --- /dev/null +++ b/scripts/governance-console/index.html @@ -0,0 +1,199 @@ + + + + + +agentctl console + + + +
+

agentctl console

+
loading…
+ + + + + +
+
+
+

1. 任务 / 租约

+
+
+
+

2. 队列与集成

+
+

Enqueue 请用 CLI：agentctl queue enqueue --task … --expected-dev … --head …

+
+
+

3. 文档漂移

+
+
+
+

4. 清理积压

+
+
+
+

brief

+ 

+  

+

+

+
+
+  

+    
claim new

+    
+    
+    
+    
+    
须在已注册任务 worktree 上下文由 serve 代其 Repo 执行。

+    
+      
+      
+    
+  

+
+
+
+  

+    
integrate prepare

+    
+    
+    
+    
+      
+      
+    
+  

+
+
+
+
+
diff --git a/scripts/mcp/agentctl-mcp.mjs b/scripts/mcp/agentctl-mcp.mjs
index d05a218e..6090a488 100755
--- a/scripts/mcp/agentctl-mcp.mjs
+++ b/scripts/mcp/agentctl-mcp.mjs
@@ -262,6 +262,33 @@ const TOOLS = [
     description: 'Refresh catalog features from MATRIX; report done→non-done regressions',
     inputSchema: { type: 'object', properties: {}, additionalProperties: false },
   },
+  {
+    name: 'brief',
+    description: 'One-page governance briefing (markdown or json)',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        format: { type: 'string', enum: ['markdown', 'json'] },
+      },
+      additionalProperties: false,
+    },
+  },
+  {
+    name: 'claim_archive',
+    description: 'Archive done CURRENT claim rows (default dry-run; set apply=true to write)',
+    inputSchema: {
+      type: 'object',
+      properties: {
+        task: { type: 'string' },
+        retain_days: { type: 'number' },
+        include_undated: { type: 'boolean' },
+        apply: { type: 'boolean' },
+        cwd: { type: 'string', description: 'task worktree absolute path' },
+      },
+      required: ['task'],
+      additionalProperties: false,
+    },
+  },
 ];
 
 function callTool(name, args = {}) {
@@ -396,6 +423,20 @@ function callTool(name, args = {}) {
       const r = runAgentctl(['catalog', 'drift']);
       return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
     }
+    case 'brief': {
+      const argv = ['brief'];
+      if (args.format) argv.push('--format', args.format);
+      const r = runAgentctl(argv);
+      return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
+    }
+    case 'claim_archive': {
+      const argv = ['claim', 'archive', '--task', args.task];
+      if (args.retain_days != null) argv.push('--retain-days', String(args.retain_days));
+      if (args.include_undated) argv.push('--include-undated');
+      if (args.apply) argv.push('--apply');
+      const r = runAgentctl(argv, { cwd: args.cwd || REPO });
+      return r.ok ? textResult(r.stdout.trim()) : errResult(r.stderr || r.stdout || r.error);
+    }
     default:
       return errResult(`unknown tool: ${name}`);
   }
diff --git a/scripts/tests/test_agent_worktree.py b/scripts/tests/test_agent_worktree.py
index 03f6a20b..4090f5b3 100644
--- a/scripts/tests/test_agent_worktree.py
+++ b/scripts/tests/test_agent_worktree.py
@@ -24,7 +24,7 @@ class WorktreeTest(unittest.TestCase):
         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", "agentctl_visibility.py"):
+        for name in ("agent-worktree.sh", "agent_worktree.py", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py", "agentctl_console.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",
diff --git a/scripts/tests/test_agentctl.py b/scripts/tests/test_agentctl.py
index 2e07f9ed..f1f64561 100644
--- a/scripts/tests/test_agentctl.py
+++ b/scripts/tests/test_agentctl.py
@@ -39,6 +39,7 @@ class AgentctlTests(unittest.TestCase):
         shutil.copy2(SOURCE.parent / "agentctl_w4.py", self.root / "scripts" / "agentctl_w4.py")
         shutil.copy2(SOURCE.parent / "agentctl_catalog.py", self.root / "scripts" / "agentctl_catalog.py")
         shutil.copy2(SOURCE.parent / "agentctl_visibility.py", self.root / "scripts" / "agentctl_visibility.py")
+        shutil.copy2(SOURCE.parent / "agentctl_console.py", 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"):
@@ -1555,6 +1556,7 @@ class ProcessPlatformTests(unittest.TestCase):
         shutil.copy2(SOURCE.parent / "agentctl_w4.py", root / "scripts" / "agentctl_w4.py")
         shutil.copy2(SOURCE.parent / "agentctl_catalog.py", root / "scripts" / "agentctl_catalog.py")
         shutil.copy2(SOURCE.parent / "agentctl_visibility.py", root / "scripts" / "agentctl_visibility.py")
+        shutil.copy2(SOURCE.parent / "agentctl_console.py", root / "scripts" / "agentctl_console.py")
         install_governance_policy(root)
         (root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
         (root / ".doc/ops").mkdir(parents=True)
diff --git a/scripts/tests/test_agentctl_catalog.py b/scripts/tests/test_agentctl_catalog.py
index 713cfc6f..0daf5655 100644
--- a/scripts/tests/test_agentctl_catalog.py
+++ b/scripts/tests/test_agentctl_catalog.py
@@ -36,6 +36,9 @@ class CatalogTests(unittest.TestCase):
         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"):
diff --git a/scripts/tests/test_agentctl_serve.py b/scripts/tests/test_agentctl_serve.py
new file mode 100644
index 00000000..2657d7d4
--- /dev/null
+++ b/scripts/tests/test_agentctl_serve.py
@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+"""Black-box tests for agentctl serve / claim new / integrate prepare / pi packet."""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+import urllib.error
+import urllib.request
+
+SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
+POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json"
+CONSOLE_STATIC = SOURCE.parent / "governance-console"
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class ConsoleServeTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(prefix="agentctl-serve-")
+        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", "Agentctl Test")
+        self.git("config", "user.email", "agentctl-test@example.invalid")
+        (self.root / "scripts").mkdir()
+        for name in (
+            "agentctl.py", "agentctl_w4.py", "agentctl_catalog.py",
+            "agentctl_visibility.py", "agentctl_console.py", "pi_subagents.py",
+            "agent_worktree.py",
+        ):
+            src = SOURCE.parent / name
+            if src.exists():
+                shutil.copy2(src, self.root / "scripts" / name)
+        static_dest = self.root / "scripts" / "governance-console"
+        static_dest.mkdir(parents=True, exist_ok=True)
+        if (CONSOLE_STATIC / "index.html").exists():
+            shutil.copy2(CONSOLE_STATIC / "index.html", static_dest / "index.html")
+        else:
+            (static_dest / "index.html").write_text("ok", encoding="utf-8")
+        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/a", ".doc/ops", ".doc/status", ".doc/test-reports"):
+            (self.root / directory).mkdir(parents=True, exist_ok=True)
+        (self.root / "src/a" / ".keep").write_text("keep\n", encoding="utf-8")
+        self._write_current([])
+        (self.root / ".doc/status/FEATURE-MATRIX.md").write_text(
+            "# Features\n\n| record_id | 模块 | 状态 |\n| --- | --- | --- |\n| core | core | planned |\n",
+            encoding="utf-8")
+        self.git("add", ".")
+        self.git("commit", "-m", "fixture")
+        self.cli("init")
+        self._server = None
+        self._thread = None
+
+    def tearDown(self):
+        if self._server is not None:
+            self._server.shutdown()
+            self._server.server_close()
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+
+    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 cli(self, *args, cwd=None, check=True):
+        result = subprocess.run(
+            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"), *args],
+            cwd=cwd or self.root, text=True, capture_output=True, timeout=60,
+        )
+        if check:
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        return result
+
+    def _last_json(self, text: str):
+        lines = [ln for ln in text.splitlines() if ln.strip()]
+        return json.loads(lines[-1])
+
+    def _write_current(self, claim_rows):
+        rows = [
+            "| owner | task | branch | worktree | mutex | status |",
+            "| --- | --- | --- | --- | --- | --- |",
+            *claim_rows,
+        ]
+        body = (
+            "# Coordination\n\n## 焦点\n\nfocus\n\n## 并行认领\n\n"
+            + "\n".join(rows) + "\n\n"
+            "## 清理接管与遗留现场\n\n"
+            "| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n"
+            "| --- | --- | --- | --- |\n"
+            "| none | none | - | - |\n\n"
+            "## 集成槽位\n\n"
+            "| holder | phase | since | scope | queue |\n"
+            "|--------|-------|-------|-------|-------|\n"
+            "| - | idle | 2026-09-24 | test | - |\n\n"
+            "## 受控约束\n\n- c\n\n## 活跃入口\n\n- e\n"
+        )
+        (self.root / ".doc/ops/CURRENT.md").write_text(body, encoding="utf-8")
+
+    def initialized_task(self, task="alpha", capabilities=("docs:publish",)):
+        path = self.root / ".worktrees" / task
+        if not path.exists():
+            self.git("worktree", "add", "-b", "feature/" + task, str(path), "dev")
+        args = ["task", "register", "--id", task, "--owner", "Test Agent",
+                "--scope", "src/a", "--scope", ".doc"]
+        for cap in capabilities:
+            args.extend(("--capability", cap))
+        self.cli(*args, cwd=path)
+        return path
+
+    def start_server(self, host="127.0.0.1", port=None):
+        port = port or _free_port()
+        sys.path.insert(0, str(self.root / "scripts"))
+        import agentctl as ac  # noqa: E402
+        import agentctl_console as console  # noqa: E402
+
+        # Ensure policy loaded for this fixture repo.
+        repo = ac.Repo(str(self.root))
+        ac.require_policy(repo)
+        args = type("A", (), {"host": host, "port": port, "static": "scripts/governance-console"})()
+        # Build server without serve_forever so tests control lifecycle.
+        if host not in ("127.0.0.1", "localhost", "::1"):
+            with self.assertRaises(ac.Denied) as ctx:
+                console.serve(repo, args)
+            return None, ctx.exception
+
+        from http.server import ThreadingHTTPServer
+        # Re-enter serve setup by calling internal pattern: use CLI subprocess for smoke,
+        # and for unit endpoints use Handler via temporary monkey of serve_forever.
+        holder = {}
+
+        def run():
+            class Stoppable(ThreadingHTTPServer):
+                allow_reuse_address = True
+
+            # Reconstruct Handler by invoking console.serve until emit, then hijack.
+            # Simpler: spawn `agentctl serve` subprocess.
+            pass
+
+        proc = subprocess.Popen(
+            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"),
+             "serve", "--host", host, "--port", str(port)],
+            cwd=str(self.root), text=True,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        )
+
+        def _stop_proc(p=proc):
+            if p.poll() is None:
+                p.terminate()
+                try:
+                    p.wait(timeout=3)
+                except subprocess.TimeoutExpired:
+                    p.kill()
+            if p.stdout:
+                p.stdout.close()
+            if p.stderr:
+                p.stderr.close()
+        self.addCleanup(_stop_proc)
+        # Wait until port accepts connections.
+        deadline = time.time() + 8
+        while time.time() < deadline:
+            if proc.poll() is not None:
+                out = (proc.stdout.read() if proc.stdout else "") + (proc.stderr.read() if proc.stderr else "")
+                raise AssertionError(f"serve exited early: {out}")
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                    break
+            except OSError:
+                time.sleep(0.05)
+        else:
+            proc.kill()
+            raise AssertionError("serve did not bind in time")
+        return port, proc
+
+    def http_json(self, port, path, method="GET", body=None, expect_status=200):
+        data = None if body is None else json.dumps(body).encode("utf-8")
+        req = urllib.request.Request(
+            f"http://127.0.0.1:{port}{path}",
+            data=data, method=method,
+            headers={"Content-Type": "application/json"} if data is not None else {},
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                raw = resp.read().decode("utf-8")
+                status = resp.status
+                ctype = resp.headers.get("Content-Type", "")
+        except urllib.error.HTTPError as exc:
+            raw = exc.read().decode("utf-8")
+            status = exc.code
+            ctype = exc.headers.get("Content-Type", "")
+        self.assertEqual(status, expect_status, raw)
+        if "json" in ctype or raw[:1] in "{[":
+            return json.loads(raw)
+        return raw
+
+    def test_non_loopback_host_denied(self):
+        sys.path.insert(0, str(self.root / "scripts"))
+        import agentctl as ac  # noqa: E402
+        import agentctl_console as console  # noqa: E402
+        repo = ac.Repo(str(self.root))
+        ac.require_policy(repo)
+        args = type("A", (), {"host": "0.0.0.0", "port": 7432, "static": "scripts/governance-console"})()
+        with self.assertRaises(ac.Denied) as ctx:
+            console.serve(repo, args)
+        self.assertEqual(ctx.exception.code, "INVALID_REQUEST")
+
+    def test_get_eight_endpoints_and_spa(self):
+        port, _ = self.start_server()
+        for path in (
+            "/api/status", "/api/brief", "/api/queue", "/api/claims",
+            "/api/cleanup", "/api/docs/drift", "/api/catalog/drift", "/api/daemon",
+        ):
+            payload = self.http_json(port, path)
+            self.assertIsInstance(payload, (dict, str))
+            if path == "/api/brief":
+                self.assertTrue(isinstance(payload, str) and payload.startswith("#"))
+            else:
+                self.assertIsInstance(payload, dict)
+        brief_json = self.http_json(port, "/api/brief?format=json")
+        self.assertIn("brief", brief_json)
+        html = self.http_json(port, "/", expect_status=200)
+        self.assertIn("agentctl console", html if isinstance(html, str) else str(html))
+
+    def test_unknown_post_shell_404(self):
+        port, _ = self.start_server()
+        payload = self.http_json(port, "/api/shell", method="POST", body={}, expect_status=404)
+        self.assertEqual(payload["code"], "NOT_FOUND")
+
+    def test_claim_new_missing_task_400(self):
+        port, _ = self.start_server()
+        payload = self.http_json(port, "/api/claim/new", method="POST", body={}, expect_status=400)
+        self.assertEqual(payload["code"], "INVALID_REQUEST")
+
+    def test_claim_new_without_docs_publish_403(self):
+        self.initialized_task(capabilities=())
+        port, _ = self.start_server()
+        payload = self.http_json(
+            port, "/api/claim/new", method="POST",
+            body={"task": "alpha", "owner": "Test Agent", "mutex": "m1"},
+            expect_status=403,
+        )
+        self.assertEqual(payload["code"], "CAPABILITY_REQUIRED")
+
+    def test_claim_new_cli_create_replace_and_replay(self):
+        task = self.initialized_task(capabilities=("docs:publish",))
+        rid = "alpha-claim-new-test1"
+        created = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m1", "--request-id", rid, cwd=task,
+        ).stdout)
+        self.assertEqual(created["status"], "created")
+        text = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertIn("| Test Agent | alpha |", text)
+        self.assertIn("| m1 | in_progress |", text)
+        # Identical row → idempotent (no second apply)
+        again = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m1", "--request-id", rid, cwd=task,
+        ).stdout)
+        self.assertEqual(again["status"], "idempotent")
+        # Without replace while row differs → refuse
+        denied = self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m2", "--request-id", "alpha-claim-new-test2", cwd=task, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)
+        replaced = self._last_json(self.cli(
+            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
+            "--mutex", "m2", "--replace", "--request-id", "alpha-claim-new-test3", cwd=task,
+        ).stdout)
+        self.assertEqual(replaced["status"], "replaced")
+        text2 = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
+        self.assertIn("| m2 | in_progress |", text2)
+
+    def test_integrate_prepare_first_parent_and_dirty_refuse(self):
+        feat = self.initialized_task(task="feat1", capabilities=("docs:publish",))
+        (feat / "src/a").mkdir(parents=True, exist_ok=True)
+        (feat / "src/a/note.txt").write_text("hello\n", encoding="utf-8")
+        self.git("add", "src/a/note.txt", cwd=feat)
+        self.git("commit", "-m", "feat change", cwd=feat)
+        expected = self.git("rev-parse", "refs/heads/dev")
+        # Prepare from main cwd is allowed by design.
+        out = self._last_json(self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", "integrate-feat1",
+        ).stdout)
+        self.assertEqual(out["status"], "prepared")
+        self.assertEqual(out["first_parent"], expected)
+        integ = self.root / ".worktrees" / "integrate-feat1"
+        self.assertTrue(integ.is_dir())
+        # Idempotent on same base
+        again = self._last_json(self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", "integrate-feat1",
+        ).stdout)
+        self.assertTrue(again.get("idempotent"))
+        # Dirty refuse on different attempt id that reuses? Use existing dirty tree.
+        (integ / "dirt.txt").write_text("x", encoding="utf-8")
+        # Advance expected to force non-idempotent path that hits dirty check —
+        # tip still same so idempotent returns early. Force by corrupting first-parent check:
+        # Create second integrate with dirty pre-created empty? Create dirty empty dir as worktree path.
+        dirty_id = "integrate-dirty"
+        dirty_path = self.root / ".worktrees" / dirty_id
+        # Manually add a fake dirty worktree at expected tip then make dirty.
+        self.git("worktree", "add", "-b", f"feature/{dirty_id}", str(dirty_path), expected)
+        (dirty_path / "dirt2.txt").write_text("y", encoding="utf-8")
+        denied = self.cli(
+            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
+            "--integrate-id", dirty_id, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)
+
+    def test_pi_packet_rejects_symbolic_and_writes(self):
+        task = self.initialized_task(capabilities=("docs:publish",))
+        # Ensure HEAD tracked file for freeze if needed
+        denied = self.cli(
+            "pi", "packet", "--task", "alpha", "--role", "research",
+            "--goal", "g", "--authorization-ref", "user:test",
+            "--source-sha", "HEAD", cwd=task, check=False,
+        )
+        self.assertNotEqual(denied.returncode, 0)
+        blob = denied.stderr + denied.stdout
+        self.assertTrue(
+            "INVALID_SOURCE_SHA" in blob or "INVALID_REQUEST" in blob or "denied" in blob.lower(),
+            blob,
+        )
+        head = self.git("rev-parse", "HEAD", cwd=task)
+        ok = self._last_json(self.cli(
+            "pi", "packet", "--task", "alpha", "--role", "research",
+            "--goal", "research goal", "--authorization-ref", "user:test",
+            "--source-sha", head, "--request-id", "pkt-1",
+            "--approved-command", "python3 -c print(1)",
+            "--write", cwd=task,
+        ).stdout)
+        self.assertEqual(ok["status"], "ok")
+        self.assertEqual(ok["packet"]["source_sha"], head)
+        self.assertEqual(ok["packet"]["role_id"], "research")
+        rel = ok["path"]
+        self.assertTrue(rel.endswith("packet.json"))
+        written = json.loads((task / rel).read_text(encoding="utf-8"))
+        self.assertEqual(written["request_id"], "pkt-1")
+        self.assertEqual(written["approved_commands"], [["python3", "-c", "print(1)"]])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/scripts/tests/test_codegraph_governance.py b/scripts/tests/test_codegraph_governance.py
index 036b68d4..2e3ba0c8 100644
--- a/scripts/tests/test_codegraph_governance.py
+++ b/scripts/tests/test_codegraph_governance.py
@@ -22,7 +22,7 @@ class CodegraphGovernanceTest(unittest.TestCase):
         (self.root / 'scripts').mkdir()
         for name in ('agent-worktree.sh', 'agent_worktree.py', 'agentctl.py',
                      'agentctl', 'agentctl_w4.py', 'agentctl_catalog.py',
-                     'agentctl_visibility.py'):
+                     'agentctl_visibility.py', 'agentctl_console.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',
diff --git a/scripts/tests/test_ops_guards.py b/scripts/tests/test_ops_guards.py
index 1b20f59c..af3c1c68 100644
--- a/scripts/tests/test_ops_guards.py
+++ b/scripts/tests/test_ops_guards.py
@@ -19,7 +19,7 @@ class OpsGuardsTest(unittest.TestCase):
         self.root = Path(self.tmp.name) / "repo"
         self.root.mkdir()
         (self.root / "scripts").mkdir()
-        for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py"):
+        for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py", "agentctl_catalog.py", "agentctl_visibility.py", "agentctl_console.py"):
             shutil.copy2(SOURCE / name, self.root / "scripts" / name)
         (self.root / ".gitignore").write_text(".worktrees/\n.locks/\nconfigs/.env.*\n!configs/.env.*.example\ndata/\n")
         (self.root / ".dockerignore").write_text(".doc\n.locks\n.worktrees\n")
diff --git a/scripts/tests/test_pi_subagents.py b/scripts/tests/test_pi_subagents.py
index 989704d1..6d012194 100644
--- a/scripts/tests/test_pi_subagents.py
+++ b/scripts/tests/test_pi_subagents.py
@@ -152,6 +152,9 @@ class PiSubagentsTests(unittest.TestCase):
         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():

From f5cdae1a41e4027338bbf1c1c06100f53f901364 Mon Sep 17 00:00:00 2001
From: weichen 
Date: Thu, 24 Sep 2026 14:15:50 +0000
Subject: [PATCH 2/3] fix(agentctl): fence integrate prepare merge via
 supervise

Pre-commit requires AGENTCTL_FENCE; run merge --no-ff through supervise like integrate ff-only.
---
 scripts/agentctl_console.py | 25 ++++++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/scripts/agentctl_console.py b/scripts/agentctl_console.py
index a2461a92..ba2bf0bc 100644
--- a/scripts/agentctl_console.py
+++ b/scripts/agentctl_console.py
@@ -10,6 +10,7 @@ import argparse
 import contextlib
 import json
 import mimetypes
+import os
 import re
 import tempfile
 import threading
@@ -177,8 +178,9 @@ def integrate_prepare(repo: ac.Repo, args) -> dict:
     _register_integrate_task(integ_path, integrate_id, owner=task.get('owner') or 'Cursor')
     integ_repo = ac.Repo(str(integ_path))
     integ_task = ac.task_row(integ_repo, integrate_id)
-    request_id = getattr(args, 'request_id', None) or f"{integrate_id}-prepare"
+    request_id = getattr(args, 'request_id', None) or f"{integrate_id}-prepare-{uuid.uuid4().hex[:8]}"
     msg = f"merge: candidate {args.task} into dev"
+    # New request_id when recovering after a failed fence-less attempt.
     with ac.Lease(integ_repo, integ_task, ['path:.', 'git:index', 'git:dev'],
                   request_id, ac.digest(msg + expected + task_branch), kind='integrate_prepare') as lease:
         if lease.replay:
@@ -191,12 +193,25 @@ def integrate_prepare(repo: ac.Repo, args) -> dict:
                 'head': head,
                 'expected_dev': expected,
             }
-        result = ac.git(integ_path, 'merge', '--no-ff', '-m', msg, task_branch, check=False)
-        if result.returncode != 0:
+        # Merge commit must run with AGENTCTL_* fence (pre-commit hook), same as integrate().
+        env = os.environ.copy()
+        env.update(
+            AGENTCTL_RUN_ID=lease.request_id,
+            AGENTCTL_TASK_ID=integrate_id,
+            AGENTCTL_GENERATION=str(lease.row['generation']),
+            AGENTCTL_EPOCH=integ_repo.identity()['epoch'],
+            AGENTCTL_FENCE=str(lease.row.get('fence') or 0),
+        )
+        code, orphan = ac.supervise(
+            integ_repo, lease,
+            ['git', 'merge', '--no-ff', '-m', msg, task_branch],
+            integ_path, env,
+        )
+        if code or orphan:
             with contextlib.suppress(Exception):
                 ac.git(integ_path, 'merge', '--abort', check=False)
-            lease.finish(1, True)
-            ac.deny('GIT_ERROR', 'merge --no-ff failed', stderr=(result.stderr or '')[:500])
+            lease.finish(code or 1, True)
+            ac.deny('GIT_ERROR', 'merge --no-ff failed', exit_code=code, orphan=bool(orphan))
         head = ac.git(integ_path, 'rev-parse', 'HEAD')
         first_parent = ac.git(integ_path, 'rev-parse', 'HEAD^1')
         if first_parent != expected:

From 77c602c8f4fcb74e5e64bc3490840c1491f82adf Mon Sep 17 00:00:00 2001
From: weichen 
Date: Thu, 24 Sep 2026 14:16:48 +0000
Subject: [PATCH 3/3] fix(agentctl): resume integrate prepare from clean
 expected tip

After abort, allow a clean worktree parked on expected-dev to continue merge instead of BASE_CHANGED.
---
 scripts/agentctl_console.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/scripts/agentctl_console.py b/scripts/agentctl_console.py
index ba2bf0bc..c119f265 100644
--- a/scripts/agentctl_console.py
+++ b/scripts/agentctl_console.py
@@ -167,9 +167,11 @@ def integrate_prepare(repo: ac.Repo, args) -> dict:
         if porcelain:
             ac.deny('INVALID_REQUEST', 'Existing integrate worktree is dirty',
                     path=str(integ_path))
-        # Different base: refuse rather than reset.
-        ac.deny('BASE_CHANGED', 'Integrate worktree exists for a different base; remove it first',
-                path=str(integ_path), head=head)
+        if head != expected:
+            # Different base: refuse rather than reset.
+            ac.deny('BASE_CHANGED', 'Integrate worktree exists for a different base; remove it first',
+                    path=str(integ_path), head=head)
+        # Clean checkout parked on expected-dev (e.g. after abort): continue to merge.
     else:
         integ_path.parent.mkdir(exist_ok=True)
         ac.git(repo.main, 'worktree', 'add', '-b', branch, str(integ_path), expected)