339 lines
13 KiB
Python
Executable File
339 lines
13 KiB
Python
Executable File
#!/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. A completed role can be reclaimed before the whole task is
|
|
verified with ``--completed-task-id`` and ``--completed-role``; this is only
|
|
valid after that role has reported success.
|
|
|
|
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"}
|
|
)
|
|
COMPLETED_ROLE_STATUSES = {
|
|
"developer": frozenset({"fixed_by_dev", "retesting", "verified"}),
|
|
"test": frozenset({"retesting", "verified"}),
|
|
}
|
|
|
|
|
|
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."""
|
|
try:
|
|
show = run_json(
|
|
[str(orca), "terminal", "show", "--terminal", handle, "--json"],
|
|
"terminal show",
|
|
)
|
|
except LaunchError:
|
|
# A role may already have been reclaimed before the final verified-task
|
|
# sweep. Treat a handle absent from the live list as idempotent success;
|
|
# retain other show failures for manual reconciliation.
|
|
listing = run_json(
|
|
[str(orca), "terminal", "list", "--json"],
|
|
"terminal list",
|
|
)
|
|
terminals = listing.get("result", {}).get("terminals")
|
|
if isinstance(terminals, list) and not any(
|
|
isinstance(item, dict) and item.get("handle") == handle for item in terminals
|
|
):
|
|
return
|
|
raise
|
|
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 completed_role_handle(board: dict, task_id: str, role: str) -> str:
|
|
"""Resolve the receipt-bound terminal for a role that just succeeded."""
|
|
|
|
if role not in COMPLETED_ROLE_STATUSES:
|
|
raise LaunchError(f"不支持的完成角色: {role}")
|
|
tasks = board.get("tasks")
|
|
tasks_by_id = (
|
|
{task.get("id"): task for task in tasks if isinstance(task, dict)}
|
|
if isinstance(tasks, list)
|
|
else {}
|
|
)
|
|
task = tasks_by_id.get(task_id)
|
|
if not isinstance(task, dict):
|
|
raise LaunchError(f"找不到 ACK task: {task_id}")
|
|
status = task.get("status")
|
|
if status not in COMPLETED_ROLE_STATUSES[role]:
|
|
raise LaunchError(
|
|
f"{role} 完成回收要求任务处于 {sorted(COMPLETED_ROLE_STATUSES[role])}，当前是 {status!r}"
|
|
)
|
|
if unresolved_incidents(task):
|
|
raise LaunchError(f"task {task_id} 存在未解决环境事件，不能回收 {role} terminal")
|
|
|
|
dispatch = task.get("dispatch")
|
|
role_dispatch = dispatch.get(role) if isinstance(dispatch, dict) else None
|
|
receipt_id = role_dispatch.get("receiptId") if isinstance(role_dispatch, dict) else None
|
|
if not isinstance(receipt_id, str) or not receipt_id:
|
|
raise LaunchError(f"task {task_id} 缺少 {role} receiptId")
|
|
|
|
receipts = board.get("workerReceipts")
|
|
receipt = next(
|
|
(
|
|
item
|
|
for item in receipts
|
|
if isinstance(item, dict) and item.get("id") == receipt_id
|
|
),
|
|
None,
|
|
) if isinstance(receipts, list) else None
|
|
if not isinstance(receipt, dict):
|
|
raise LaunchError(f"找不到 {role} receipt: {receipt_id}")
|
|
created_for = receipt.get("createdFor")
|
|
binding = receipt.get("binding")
|
|
if (
|
|
not isinstance(created_for, dict)
|
|
or created_for.get("taskId") != task_id
|
|
or created_for.get("role") != role
|
|
or not isinstance(binding, dict)
|
|
or not isinstance(binding.get("handle"), str)
|
|
or not binding.get("handle")
|
|
):
|
|
raise LaunchError(f"{role} receipt 与 task/role/handle 不匹配: {receipt_id}")
|
|
return binding["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（默认全部）")
|
|
parser.add_argument("--completed-task-id", help="刚完成的 ACK task ID；与 --completed-role 一起使用")
|
|
parser.add_argument(
|
|
"--completed-role",
|
|
choices=("developer", "test"),
|
|
help="刚成功完成的角色；与 --completed-task-id 一起使用",
|
|
)
|
|
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
|
|
|
|
if bool(args.completed_task_id) != bool(args.completed_role):
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"error": "--completed-task-id 与 --completed-role 必须同时提供",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return 2
|
|
if args.completed_task_id:
|
|
if args.handle:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"error": "完成角色回收不能同时使用 --handle",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return 2
|
|
try:
|
|
handle = completed_role_handle(board, args.completed_task_id, args.completed_role)
|
|
closed: bool | None = None
|
|
decision = "close-completed-role"
|
|
reason = f"{args.completed_role} reported success for task {args.completed_task_id}"
|
|
if args.apply:
|
|
close_terminal(resolve_executable("orca"), handle)
|
|
closed = True
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"handles": [
|
|
{
|
|
"handle": handle,
|
|
"decision": decision,
|
|
"reason": reason,
|
|
"tasks": [args.completed_task_id],
|
|
"role": args.completed_role,
|
|
"closed": closed,
|
|
}
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return 0
|
|
except LaunchError as exc:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"mode": "apply" if args.apply else "dry-run",
|
|
"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())
|