From 1c52a3996cf1aaabed39bc207fbc3e9351cdbce3 Mon Sep 17 00:00:00 2001 From: laily Date: Sun, 23 Aug 2026 23:14:29 +0800 Subject: [PATCH] feat(ack): reclaim verified worker terminals via script after round end --- skills/ack/SKILL.md | 11 +- skills/ack/references/kickoff.md | 9 +- skills/ack/references/orca-adapter.md | 36 ++-- skills/ack/scripts/reclaim_workers.py | 196 +++++++++++++++++++++ tests/test_ack_reclaim_workers.py | 238 ++++++++++++++++++++++++++ 5 files changed, 468 insertions(+), 22 deletions(-) create mode 100755 skills/ack/scripts/reclaim_workers.py create mode 100644 tests/test_ack_reclaim_workers.py diff --git a/skills/ack/SKILL.md b/skills/ack/SKILL.md index e2b5a57..a5a68b1 100644 --- a/skills/ack/SKILL.md +++ b/skills/ack/SKILL.md @@ -201,11 +201,12 @@ description: >- `validation_ready`，并把访问地址、验证范围和用户下一步交给用户；不能停在 `verified` 却声称整轮 ACK 已结束。默认 profile 最多到 `validation_ready` 或 `review_ready`，稳定发布和生产部署必须在对应步骤再次取得明确批准。 -11. Coordinator 最后标记整轮任务完成后，回收所有只属于 `verified` 任务的 worker - 终端，并核对关闭回执；历史 receipt 和任务证据继续保留。任何还被 `open`、 - `dispatched`、`fixed_by_dev`、`retesting`、`blocked`、`failed_retest`、`leftover` - 或未解决环境事件引用的终端都保留，不设置 TTL，也不能因为同一终端还关联过 - `verified` 任务而误关。若关闭结果不确定，记录并报告，不重复关闭或伪报已回收。 +11. Coordinator 最后标记整轮任务完成后，用 `scripts/reclaim_workers.py` 先 + dry-run 审阅决策，再 `--apply` 回收所有只属于 `verified` 任务的 worker + 终端，并核对关闭回执；历史 receipt 和任务证据继续保留。任何还被 `open`、`dispatched`、`fixed_by_dev`、 + `retesting`、`blocked`、`failed_retest`、`leftover` 或未解决环境事件引用的终端 + 都保留，不设置 TTL，也不能因为同一终端还关联过 `verified` 任务而误关。若关闭 + 结果不确定，记录并报告，不重复关闭或伪报已回收。 ## 交付配置维护 diff --git a/skills/ack/references/kickoff.md b/skills/ack/references/kickoff.md index 032ed27..2b8179d 100644 --- a/skills/ack/references/kickoff.md +++ b/skills/ack/references/kickoff.md @@ -190,10 +190,11 @@ delivery run 标为 `blocked` 或 `failed`。 一轮结束时 Coordinator 必须能回答 `optimization-method.md` §「结束条件」的问题： 哪些 verified、哪些 leftover、各失败几轮、工作树是否干净、还有没有未处理项。 -Coordinator 最后标记整轮任务完成后，关闭所有只关联 `verified` 任务的 Developer/Test -终端并核对关闭回执；receipt 和落盘证据继续保留。仍关联 `blocked`、`failed_retest`、 -`leftover` 或其它未完成任务/未解决环境事件的终端保留，不设置 TTL。一个终端只要仍被 -任一这类任务引用，就不能因为同时关联了 verified 任务而关闭。 +Coordinator 最后标记整轮任务完成后，用 `scripts/reclaim_workers.py` 先 dry-run +审阅决策、再 `--apply` 关闭所有只关联 `verified` 任务的 Developer/Test 终端并核对 +回执；receipt 和落盘证据继续保留。仍关联 `blocked`、`failed_retest`、`leftover` +或其它未完成任务/未解决环境事件的终端保留，不设置 TTL。一个终端只要仍被任一这类 +任务引用，就不能因为同时关联了 verified 任务而关闭。 --- diff --git a/skills/ack/references/orca-adapter.md b/skills/ack/references/orca-adapter.md index 26a9f06..c082674 100644 --- a/skills/ack/references/orca-adapter.md +++ b/skills/ack/references/orca-adapter.md @@ -321,19 +321,29 @@ orca orchestration send \ ## 整轮结束时回收 worker -只有 Coordinator 已把本轮最终结论写入 `tasks.yaml` 后才开始回收。先按 receipt 的 -`binding.handle` 聚合任务引用：只有一个 handle 关联的全部任务都为 `verified`，且没有 -未解决环境事件，才允许关闭。`open`、`dispatched`、`fixed_by_dev`、`retesting`、 -`blocked`、`failed_retest`、`leftover` 以及状态不明任务关联的 handle 全部保留，不设置 -TTL。普通用户 Shell 和 Coordinator 自己的终端不在回收范围。 +只有 Coordinator 已把本轮最终结论写入 `tasks.yaml` 后才开始回收。统一使用回收脚本， +不要手工逐个 close： -逐个回收时： +```bash +# 第一步：dry-run 审阅决策（不关闭任何终端） +python3 /scripts/reclaim_workers.py --project-root -1. `orca terminal show --terminal --json`，把 runtimeId、handle、incarnationId - 和 worktree 与 receipt 逐项核对；不匹配就停止并报告。 -2. `orca terminal close --terminal --tab --json`。 -3. 核对关闭回执的 runtimeId、handle、tab 和关闭模式，再确认终端不在 live list。 -4. 关闭结果不确定时保留记录并报告，禁止盲目重试或标记为已回收。 +# 第二步：审阅输出后真正回收 +python3 /scripts/reclaim_workers.py \ + --project-root --apply +``` -关闭终端不删除 `workerReceipts`、dispatch 证据或测试记录；这些仍是审计事实。保留下来 -的 blocked/failed worker 只在后续任务得到验证或用户明确要求清理时回收。 +脚本按 receipt 的 `binding.handle` 聚合任务引用，规则固定： + +- **关闭**：handle 关联的全部任务都为 `verified`，且没有未解决环境事件。 +- **保留（不设置 TTL）**：任一关联任务处于 `open`、`dispatched`、`fixed_by_dev`、 + `retesting`、`blocked`、`failed_retest`、`leftover`，或存在未解决 + `environmentIncidents`，或 receipt 引用未知任务。失败三次的 worker（ + `failed_retest` / `leftover` / `blocked`）按此保留。 +- **不触碰**：`workerReceipts` 之外的 handle（Coordinator 终端、用户 Shell）。 + +`--apply` 时脚本对每个待关闭 handle 执行 `orca terminal show` 核对身份 → +`orca terminal close --tab` → 重新 `terminal list` 确认 handle 已消失；任何一步 +不确定都保留并标记 `uncertain`，禁止盲目重试。关闭终端不删除 `workerReceipts`、 +dispatch 证据或测试记录；这些仍是审计事实。保留下来的 blocked/failed worker 只在 +后续任务得到验证或用户明确要求清理时回收。 diff --git a/skills/ack/scripts/reclaim_workers.py b/skills/ack/scripts/reclaim_workers.py new file mode 100755 index 0000000..e79a263 --- /dev/null +++ b/skills/ack/scripts/reclaim_workers.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""Reclaim ACK worker terminals at the end of a coordination round. + +Reads the authoritative tasks.yaml and decides, per worker receipt handle, +whether the terminal may be closed: + +- close: every task referencing the handle is ``verified`` and no referenced + task has an unresolved ``dispatch.environmentIncidents`` entry; +- retain: any referenced task is still open/dispatched/fixed_by_dev/retesting/ + blocked/failed_retest/leftover, has an open environment incident, or the + receipt references an unknown task. Failed-three-times workers stay open by + design (leftover/failed_retest/blocked are all retained); +- skip: handles not present in workerReceipts (coordinator terminal, user + shells) are never touched. + +Default mode is dry-run: print decisions only. Pass ``--apply`` to actually +show-verify and close. Closing is conservative: identity must match the +receipt, the close receipt must be ok, and the handle must disappear from the +live terminal list; anything uncertain stays retained and is reported instead +of being retried. + +Output is a single JSON document: + {"mode": "dry-run|apply", "handles": [{handle, decision, reason, tasks, closed}]} +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from launch_worker import ( # noqa: E402 + LaunchError, + load_authoritative_board, + resolve_executable, + run_json, +) + +KEEP_STATUSES = frozenset( + {"open", "dispatched", "fixed_by_dev", "retesting", "blocked", "failed_retest", "leftover"} +) + + +def collect_handle_tasks(board: dict) -> dict[str, dict[str, object]]: + """Map each worker receipt handle to its referenced task facts.""" + tasks_by_id = {task.get("id"): task for task in board.get("tasks", []) if isinstance(task, dict)} + handles: dict[str, dict[str, object]] = {} + for receipt in board.get("workerReceipts", []): + if not isinstance(receipt, dict): + continue + binding = receipt.get("binding") + created_for = receipt.get("createdFor") + if not isinstance(binding, dict) or not isinstance(created_for, dict): + continue + handle = binding.get("handle") + task_id = created_for.get("taskId") + if not isinstance(handle, str) or not handle: + continue + entry = handles.setdefault(handle, {"taskIds": [], "tasks": []}) + if isinstance(task_id, str) and task_id not in entry["taskIds"]: + entry["taskIds"].append(task_id) + entry["tasks"].append(tasks_by_id.get(task_id)) + return handles + + +def unresolved_incidents(task: dict | None) -> list[str]: + if not isinstance(task, dict): + return [] + dispatch = task.get("dispatch") + if not isinstance(dispatch, dict): + return [] + incidents = dispatch.get("environmentIncidents") + if not isinstance(incidents, list): + return [] + return [ + str(incident.get("id")) + for incident in incidents + if isinstance(incident, dict) and incident.get("status") == "open" + ] + + +def decide(handle: str, entry: dict[str, object]) -> tuple[str, str, list[str]]: + task_ids = entry["taskIds"] + tasks = entry["tasks"] + statuses: list[str] = [] + for task in tasks: + if isinstance(task, dict): + statuses.append(str(task.get("status"))) + else: + statuses.append("unknown-task") + incidents: list[str] = [] + for task in tasks: + incidents.extend(unresolved_incidents(task)) + if incidents: + return ( + "retain", + f"unresolved environment incident(s): {', '.join(incidents)}", + statuses, + ) + bad = [status for status in statuses if status in KEEP_STATUSES or status == "unknown-task"] + if bad: + return ( + "retain", + f"referenced task(s) not verified: {', '.join(bad)} (task ids: {', '.join(task_ids)})", + statuses, + ) + if not task_ids: + return "retain", "receipt has no task reference", statuses + return "close", "all referenced tasks verified", statuses + + +def close_terminal(orca: Path, handle: str) -> None: + """Show-verify identity, close the tab, then confirm it left the live list.""" + show = run_json( + [str(orca), "terminal", "show", "--terminal", handle, "--json"], + "terminal show", + ) + terminal = show.get("result", {}).get("terminal") + if not isinstance(terminal, dict) or terminal.get("handle") != handle: + raise LaunchError(f"terminal show 未返回匹配的 handle: {handle}") + + close = run_json( + [str(orca), "terminal", "close", "--terminal", handle, "--tab", "--json"], + "terminal close", + ) + result = close.get("result") + if isinstance(result, dict): + closed_handle = result.get("handle") or (result.get("terminal") or {}).get("handle") + if closed_handle not in (None, handle): + raise LaunchError(f"terminal close 回执 handle 不匹配: {closed_handle!r}") + + listing = run_json( + [str(orca), "terminal", "list", "--json"], + "terminal list", + ) + terminals = listing.get("result", {}).get("terminals") + if isinstance(terminals, list) and any( + isinstance(item, dict) and item.get("handle") == handle for item in terminals + ): + raise LaunchError(f"terminal close 后 handle 仍在 live list: {handle}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--project-root", required=True, help="权威 tasks.yaml 所在项目根") + parser.add_argument("--apply", action="store_true", help="真正关闭；默认只输出决策") + parser.add_argument("--handle", help="只处理指定 handle（默认全部）") + args = parser.parse_args(argv) + + try: + project_root, board = load_authoritative_board(args.project_root) + except LaunchError as exc: + print(json.dumps({"mode": "dry-run" if not args.apply else "apply", "error": str(exc)}, ensure_ascii=False)) + return 2 + + handles = collect_handle_tasks(board) + results: list[dict[str, object]] = [] + for handle in sorted(handles): + if args.handle and handle != args.handle: + continue + entry = handles[handle] + decision, reason, statuses = decide(handle, entry) + closed: bool | None = None + if decision == "close" and args.apply: + try: + close_terminal(resolve_executable("orca"), handle) + closed = True + except LaunchError as exc: + decision = "uncertain" + reason = f"close failed: {exc}" + closed = None + results.append( + { + "handle": handle, + "decision": decision, + "reason": reason, + "tasks": entry["taskIds"], + "statuses": statuses, + "closed": closed, + } + ) + print( + json.dumps( + {"mode": "apply" if args.apply else "dry-run", "handles": results}, + ensure_ascii=False, + ) + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_ack_reclaim_workers.py b/tests/test_ack_reclaim_workers.py new file mode 100644 index 0000000..250f1a5 --- /dev/null +++ b/tests/test_ack_reclaim_workers.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import contextlib +import io +import json +import sys +import unittest +from pathlib import Path +from unittest import mock + + +REPO_ROOT = Path(__file__).resolve().parents[1] +ACK_SCRIPTS = REPO_ROOT / "skills" / "ack" / "scripts" +sys.path.insert(0, str(ACK_SCRIPTS)) + +import reclaim_workers # noqa: E402 + + +def task(task_id: str, status: str, *, incidents: list[dict] | None = None) -> dict: + dispatch = {"environmentIncidents": incidents or []} + return {"id": task_id, "status": status, "dispatch": dispatch} + + +def receipt(handle: str, task_id: str) -> dict: + return { + "binding": {"handle": handle}, + "createdFor": {"taskId": task_id}, + } + + +def board(receipts: list[dict], tasks: list[dict]) -> dict: + return {"workerReceipts": receipts, "tasks": tasks} + + +class DecideTests(unittest.TestCase): + def test_all_verified_means_close(self) -> None: + entry = { + "taskIds": ["T-1"], + "tasks": [task("T-1", "verified")], + } + decision, reason, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "close") + self.assertIn("verified", reason) + + def test_blocked_means_retain(self) -> None: + entry = { + "taskIds": ["T-1"], + "tasks": [task("T-1", "blocked")], + } + decision, reason, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "retain") + self.assertIn("blocked", reason) + + def test_leftover_means_retain(self) -> None: + entry = { + "taskIds": ["T-1"], + "tasks": [task("T-1", "leftover")], + } + decision, _, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "retain") + + def test_unresolved_incident_means_retain_even_when_verified(self) -> None: + entry = { + "taskIds": ["T-1"], + "tasks": [ + task( + "T-1", + "verified", + incidents=[ + { + "id": "T-1-ENV-1", + "status": "open", + } + ], + ) + ], + } + decision, reason, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "retain") + self.assertIn("T-1-ENV-1", reason) + + def test_resolved_incident_does_not_block_close(self) -> None: + entry = { + "taskIds": ["T-1"], + "tasks": [ + task( + "T-1", + "verified", + incidents=[{"id": "T-1-ENV-1", "status": "resolved"}], + ) + ], + } + decision, _, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "close") + + def test_unknown_task_reference_means_retain(self) -> None: + entry = {"taskIds": ["T-MISSING"], "tasks": [None]} + decision, reason, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "retain") + self.assertIn("unknown-task", reason) + + def test_mixed_statuses_means_retain(self) -> None: + entry = { + "taskIds": ["T-1", "T-2"], + "tasks": [task("T-1", "verified"), task("T-2", "retesting")], + } + decision, reason, _ = reclaim_workers.decide("term_1", entry) + self.assertEqual(decision, "retain") + self.assertIn("retesting", reason) + + +class CollectTests(unittest.TestCase): + def test_handle_maps_to_unique_task_ids(self) -> None: + handles = reclaim_workers.collect_handle_tasks( + board( + [receipt("term_1", "T-1"), receipt("term_1", "T-1"), receipt("term_2", "T-2")], + [task("T-1", "verified"), task("T-2", "blocked")], + ) + ) + self.assertEqual(sorted(handles), ["term_1", "term_2"]) + self.assertEqual(handles["term_1"]["taskIds"], ["T-1"]) + + def test_receipt_without_binding_is_ignored(self) -> None: + handles = reclaim_workers.collect_handle_tasks( + {"workerReceipts": [{"binding": {}}], "tasks": []} + ) + self.assertEqual(handles, {}) + + +class MainTests(unittest.TestCase): + def test_dry_run_emits_decisions_without_closing(self) -> None: + board_data = board([receipt("term_1", "T-1")], [task("T-1", "verified")]) + with mock.patch.object( + reclaim_workers, "load_authoritative_board", return_value=(Path("/repo"), board_data) + ): + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + exit_code = reclaim_workers.main( + ["--project-root", "/repo"] + ) + payload = json.loads(buffer.getvalue()) + self.assertEqual(exit_code, 0) + self.assertEqual(payload["mode"], "dry-run") + self.assertEqual(payload["handles"][0]["decision"], "close") + self.assertIsNone(payload["handles"][0]["closed"]) + + def test_apply_closes_verified_and_keeps_blocked(self) -> None: + board_data = board( + [receipt("term_1", "T-1"), receipt("term_2", "T-2")], + [task("T-1", "verified"), task("T-2", "blocked")], + ) + with ( + mock.patch.object( + reclaim_workers, "load_authoritative_board", return_value=(Path("/repo"), board_data) + ), + mock.patch.object( + reclaim_workers, "resolve_executable", return_value=Path("/trusted/orca") + ), + mock.patch.object(reclaim_workers, "close_terminal") as close, + ): + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + exit_code = reclaim_workers.main(["--project-root", "/repo", "--apply"]) + payload = json.loads(buffer.getvalue()) + self.assertEqual(exit_code, 0) + close.assert_called_once() + by_handle = {item["handle"]: item for item in payload["handles"]} + self.assertEqual(by_handle["term_1"]["decision"], "close") + self.assertTrue(by_handle["term_1"]["closed"]) + self.assertEqual(by_handle["term_2"]["decision"], "retain") + self.assertIsNone(by_handle["term_2"]["closed"]) + + def test_apply_close_failure_is_uncertain_and_reported(self) -> None: + board_data = board([receipt("term_1", "T-1")], [task("T-1", "verified")]) + with ( + mock.patch.object( + reclaim_workers, "load_authoritative_board", return_value=(Path("/repo"), board_data) + ), + mock.patch.object( + reclaim_workers, "resolve_executable", return_value=Path("/trusted/orca") + ), + mock.patch.object( + reclaim_workers, + "close_terminal", + side_effect=reclaim_workers.LaunchError("close failed"), + ), + ): + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + exit_code = reclaim_workers.main(["--project-root", "/repo", "--apply"]) + payload = json.loads(buffer.getvalue()) + self.assertEqual(exit_code, 0) + item = payload["handles"][0] + self.assertEqual(item["decision"], "uncertain") + self.assertIsNone(item["closed"]) + self.assertIn("close failed", item["reason"]) + + def test_missing_board_returns_error_exit(self) -> None: + with mock.patch.object( + reclaim_workers, + "load_authoritative_board", + side_effect=reclaim_workers.LaunchError("board missing"), + ): + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + exit_code = reclaim_workers.main(["--project-root", "/repo"]) + payload = json.loads(buffer.getvalue()) + self.assertEqual(exit_code, 2) + self.assertIn("error", payload) + + +class CloseTerminalTests(unittest.TestCase): + def test_close_verifies_show_close_and_live_list(self) -> None: + responses = [ + {"ok": True, "result": {"terminal": {"handle": "term_1"}}}, + {"ok": True, "result": {"handle": "term_1"}}, + {"ok": True, "result": {"terminals": [{"handle": "term_2"}]}}, + ] + with mock.patch.object( + reclaim_workers, "run_json", side_effect=responses + ): + reclaim_workers.close_terminal(Path("/trusted/orca"), "term_1") + + def test_close_fails_when_handle_still_live(self) -> None: + responses = [ + {"ok": True, "result": {"terminal": {"handle": "term_1"}}}, + {"ok": True, "result": {"handle": "term_1"}}, + {"ok": True, "result": {"terminals": [{"handle": "term_1"}]}}, + ] + with ( + mock.patch.object(reclaim_workers, "run_json", side_effect=responses), + self.assertRaises(reclaim_workers.LaunchError), + ): + reclaim_workers.close_terminal(Path("/trusted/orca"), "term_1") + + +if __name__ == "__main__": + unittest.main()