From ede7c6afeedd31cec1478ee2b35df7f98c9f836d Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 02:45:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=95=BF=E4=BB=BB=E5=8A=A1=E5=AF=B9?= =?UTF-8?q?=E9=BD=90=20Codex=20App=20+=20/goal=20=E6=A3=80=E6=9F=A5?= =?UTF-8?q?=E7=82=B9=E7=BB=AD=E8=B7=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - --turn-timeout / turn_timeout 可配置，0 不限时（对齐桌面 App 等待语义）；超时回合保留部分文本，任务记 interrupted 而非 failed - Server.interrupt_turn + host.py interrupt --task（协议 turn/interrupt，无活动回合时不改任务状态） - goal 循环抽成 run_goal_loop：每轮落盘检查点（含首轮前初始保存），goal-resume 从下一轮续跑；终局 goal-met/goal-unmet 后拒绝续跑 - 回合超时/中断时下一轮提示写明是回合中断而非验收失败；预算耗尽时记 interrupted 不冒充 unmet；--verify-timeout（默认 600，0 不限时，超时记 rc 124） - 云端解析契约未动：goal met/unmet after rounds、round 行、usage 行格式保持 - 真模型实弹：端到端、断点续跑、正常多轮反馈修复（低档、gpt-5.6-terra） --- README.md | 4 +- host.py | 316 +++++++++++++++++++++++++++++++++++++-------- tests/test_host.py | 239 ++++++++++++++++++++++++++++++++++ 3 files changed, 502 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 1e3efd7..08ce2b8 100644 --- a/README.md +++ b/README.md @@ -138,12 +138,12 @@ turn 可以真正并发（不是轮流排队）。 | 线程续跑 / fork / 归档 | `run`（resume + 新回合）、`fork`、`archive`、`list` | 已实测 | | 宿主级子代理委派 | `delegate`：子 thread（默认 read-only）跑完，结果可 `--report-to-parent` 汇总回主 thread | 已实测 | | 模型内子代理 | prompt 明确要求派生，见 `examples/subagent-prompt.txt`；与宿主委派是两层，别混 | 已实测（模板 + 协议） | -| 长任务 / 目标循环 | `goal`：同一 thread 多回合，直到 `--verify` 验收命令通过或到 `--max-rounds` | 已实测（1 轮通过） | +| 长任务 / 目标循环 | `goal`：同一 thread 多回合，直到 `--verify` 验收命令通过或到 `--max-rounds`；每轮落盘检查点，中断后 `goal-resume` 从下一轮续跑（goal-met / goal-unmet 终局后不许续跑，须新建）；`--turn-timeout 0` 不限时，`--verify-timeout` 限验测；回合超时下一轮提示写明是回合中断不是验收失败 | 已实测：端到端、断点续跑、正常多轮反馈修复 | | 自动化 / 评审队列 | `automations --add` + `run-due`：到期的任务自动派发，结果进 host-state.json 评审队列 | 已实测 | | 沙箱边界 | read-only thread 的写文件请求被拦下，文件确实没建出来 | 已实测 | | 审批控制 | untrusted + deny 时 agent 明确收到审批拒绝，没拿到命令输出 | 已实测 | | 模型内子代理 | 3 子代理并行检查并汇总，视角是真的不同，不是重复一句话 | 已实测 | -| 中断 / 续跑控制 | `turn/interrupt` 后状态为 interrupted；`compact` 需先 resume（host.py 的 compact 命令已包好这一步） | 已实测 | +| 中断 / 续跑控制 | `turn/interrupt` 后状态为 interrupted；`host.py interrupt --task` 已接协议；`compact` 需先 resume（host.py 的 compact 命令已包好这一步）；超时回合也记 interrupted 且保留部分文本 | 已实测（interrupt 真 server 回 no active turn 时不改状态） | | 合并冲突处理 | 先合的进主仓，后合同一行的冲突 abort，主仓干净，冲突 worktree 留着人工解 | 已实测 | | 自动化幂等 | 未到期的自动化 run-due 时不重复派发 | 已实测 | | 宿主提供 MCP 工具 | 隔离 CODEX_HOME 注册 stdio MCP，agent 能发现并调用；注意 MCP 审批是 elicitation 协议，不是通用 decision 字段，host.py 已分流处理 | 已实测（修过一轮才通） | diff --git a/host.py b/host.py index 6e3115a..c2f524b 100644 --- a/host.py +++ b/host.py @@ -96,6 +96,21 @@ def _sum_turn_usage(events, thread_id, turn_id): return total +def _partial_text_from_events(events): + """超时回合的 best-effort 部分文本：从事件窗口里已完成的 + agentMessage 收集文本（不含流式 delta，避免重复拼接）。 + 找不到时返回空串，调用方不得把部分文本当成最终结论。""" + parts = [] + for ev in events or []: + p = ev.get("params") or {} + item = p.get("item") or {} + if item.get("type") == "agentMessage" and item.get("text"): + text = item["text"] + if not parts or parts[-1] != text: + parts.append(text) + return "\n".join(parts) + + def _add_usage(a, b): """两份用量相加（任一为空时返回另一份；都为空返回 None）。""" if not a: @@ -457,11 +472,21 @@ class Server: t = resp["result"].get("turn") or {} turn_id = t.get("id") try: - done = q.get(timeout=timeout) + # 对齐 Codex App 的长任务语义：timeout 为 None 或 0 + # 表示不限时、一直等到 turn/completed（由用户中断 + # 或模型自己结束）；正数才按秒计时。默认仍是 600， + # 由调用方（CLI --turn-timeout / 配置 turn_timeout） + # 显式改，不改默认就不会悄悄挂起自动化。 + if timeout and timeout > 0: + done = q.get(timeout=timeout) + else: + done = q.get() except queue.Empty: with self.events_lock: evs = list(self.events[start_events:]) - return {"status": "timeout", "text": "", "turn_id": turn_id, + return {"status": "timeout", + "text": _partial_text_from_events(evs), + "turn_id": turn_id, "usage": _sum_turn_usage(evs, thread_id, turn_id)} turn = (done.get("params") or {}).get("turn") or {} text = "" @@ -506,6 +531,18 @@ class Server: finally: self.turn_queues.pop(thread_id, None) + def interrupt_turn(self, thread_id, turn_id): + """对齐 Codex App 的停止按钮：向 app-server 发 turn/interrupt。 + 协议要求 threadId + turnId；turn_id 缺失时无法中断，返回 None。 + 注意：中断只对本 server 进程正在跑的 turn 可靠，另一进程的 + turn 未必能被本进程的中断请求停下，此时应直接 Ctrl-C + 等待中的那个进程（其 finally 会清队列，recover 会标 interrupted）。""" + if not thread_id or not turn_id: + return None + return self.request("turn/interrupt", + {"threadId": thread_id, "turnId": turn_id}, + timeout=120) + def _turn_error_from_history(self, thread_id, turn_id): """从 codex thread_history 库里取失败回合的 error message。 @@ -1405,27 +1442,40 @@ def dispatch(server, state, name, prompt, repo, model, effort, sandbox, def run_task_turn(server, state, task, prompt, model, effort, - sandbox=None): + sandbox=None, timeout=None): # task 自带 sandbox 时优先用它（只读评审任务不能被全局默认覆盖）， # 否则用调用方显式传的 sandbox；两者都没有时不传 sandboxPolicy， # 保留 thread 原有设置（旧任务没有 sandbox 记录时尤其重要）。 + # timeout 为 None 时不传给 server（沿用 server 默认 600）， + # 这样旧 FakeServer（不接受 timeout 参数）不受影响；显式 + # 传了才透传，0 由 server 侧解释为不限时。 use_sandbox = task.get("sandbox") or sandbox started = time.time() + turn_kwargs = {} + if timeout is not None: + turn_kwargs["timeout"] = timeout result = server.run_turn(task["thread_id"], prompt, model=model, - effort=effort, sandbox=use_sandbox) + effort=effort, sandbox=use_sandbox, + **turn_kwargs) finished = time.time() record_turn_usage(state, failed=(result["status"] != "completed")) task["turns"].append({"prompt": prompt, "result": result, "started_at": started, "finished_at": finished}) task["last_turn_at"] = finished - task["status"] = "completed" if result["status"] == "completed" \ - else "failed" + # 对齐 Codex App：超时 / 被中断不是失败，是可续跑的中断态； + # 只有真正的 failed 才记 failed，recover 才能区分两者。 + if result["status"] == "completed": + task["status"] = "completed" + elif result["status"] in ("timeout", "interrupted"): + task["status"] = "interrupted" + else: + task["status"] = "failed" task["last_text"] = result.get("text", "") return result def run_task_turn_with_retry(server, state, task, prompt, model, effort, - sandbox=None, cfg=None): + sandbox=None, cfg=None, timeout=None): """失败自动重试：默认 max_retries=0 时行为不变（只跑一次）； 配了重试时，失败后同 prompt 再跑，最多 max_retries 次， 每次重试前先查预算，超预算直接停，不烧钱。""" @@ -1447,7 +1497,7 @@ def run_task_turn_with_retry(server, state, task, prompt, model, effort, task["last_text"] = "" break result = run_task_turn(server, state, task, prompt_try, - model, effort, sandbox) + model, effort, sandbox, timeout=timeout) # 各次尝试的真实用量都要算（失败尝试花的钱也是钱）， # 最后一次的结果里带汇总，不只带最后一次。 usage = _add_usage(usage, result.get("usage")) @@ -1458,6 +1508,129 @@ def run_task_turn_with_retry(server, state, task, prompt, model, effort, return result +def run_goal_loop(server, state, state_path, task, goal_text, verify, + max_rounds, model, effort, sandbox, cfg=None, + timeout=None, verify_timeout=600, start_round=1, + last_verify="", last_turn_status=None, last_turn_text=""): + """目标循环（长任务版，对齐 Codex App 的可续跑语义）。 + + 每轮：turn → 验测 → 把真实结果喂回下一轮；每轮结束立即落盘 + （检查点），进程被杀 / VM 回收后可用 goal-resume 从 + goal_rounds_done 的下一轮续跑，而不是从头重来。 + 与云端 worker 的解析契约（cloud/sandbox.py）： + 结论行必须是 "goal met after rounds" / "goal unmet after rounds"， + 每轮行必须是 "round N: turn=... verify_rc=... verify_out=..."， + 用量行必须是 "usage input=... cached=... ..." 格式，改动时 + 不能动这三行；两种结局都 exit 0（main 不抛异常即 0）。 + 预算 stop 是第三种结局：打印 "goal stopped after rounds: ..." + 且任务记 interrupted（明天预算重置后可续跑），不冒充 unmet。 + goal-unmet 穷尽轮次后不允许 goal-resume（按既有规则须新建 + 任务），goal-resume 只接受 goal-running / interrupted / running。 + """ + task["goal"] = goal_text + task["verify_command"] = verify + task["goal_max_rounds"] = int(max_rounds) + task["goal_rounds_done"] = int(start_round) - 1 + task["goal_last_verify"] = last_verify + # 循环前先落盘一次：否则第一轮 turn 进行中被杀时，磁盘上的 + # 任务还没有 goal/verify_command，goal-resume 会因“没有记录” + # 而拒绝——断点续跑必须在第一轮开始前就已可续跑。 + task["status"] = "goal-running" + save_state(state_path, state) + ok = False + budget_stopped = False + goal_usage = None + for round_no in range(int(start_round), int(max_rounds) + 1): + if cfg is not None and budget_exceeded(state, cfg): + budget_stopped = True + print(f"goal stopped: 预算已用完（max_daily_turns），" + f"第 {round_no} 轮未开始，任务可续跑") + break + if round_no == 1 and not last_verify and not last_turn_status: + prompt = goal_text + elif last_turn_status in ("timeout", "interrupted"): + # 上轮回合根本没跑完：说清是回合中断，不是验收失败， + # 否则 agent 会对着验收输出修一个不存在的问题。 + prompt = (f"上一轮回合未完成（状态：{last_turn_status}），" + f"请继续完成目标，不要重复已完成的部分。\n" + f"目标：{goal_text}\n" + f"验收命令：{verify}\n" + f"上一轮验收输出：\n{last_verify}\n" + f"上一轮部分文本：\n{last_turn_text[:1000]}") + else: + # 生产教训：只说“验收未通过” agent 不知道改什么， + # 2026-10-25 实测 prod-retry 3 轮都原地不动。 + # 必须把验收命令和失败输出喂回去，才是可用的目标循环。 + prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n" + f"目标：{goal_text}\n" + f"验收命令：{verify}\n" + f"上一轮验收输出：\n{last_verify}") + result = run_task_turn(server, state, task, prompt, + model, effort, sandbox, timeout=timeout) + goal_usage = _add_usage(goal_usage, result.get("usage")) + last_turn_status = result["status"] + last_turn_text = result.get("text", "") or "" + try: + v_rc, v_out = None, None + if verify_timeout and verify_timeout > 0: + v = subprocess.run(verify, shell=True, cwd=task["cwd"], + capture_output=True, text=True, + timeout=verify_timeout) + else: + v = subprocess.run(verify, shell=True, cwd=task["cwd"], + capture_output=True, text=True) + v_rc = v.returncode + v_out = (v.stdout + v.stderr).strip()[:1000] or \ + f"exit code {v.returncode}" + except subprocess.TimeoutExpired as exc: + # 验测本身也可能很长（测试套件）：超时是验测失败， + # 不是进程挂死，喂回下一轮让 agent 缩小验测范围。 + v_rc = 124 + out = (exc.stdout or "") if isinstance(exc.stdout, str) \ + else "" + err = (exc.stderr or "") if isinstance(exc.stderr, str) \ + else "" + v_out = (f"verify timeout after {verify_timeout}s: " + f"{(out + err).strip()[:900]}") + last_verify = v_out + task["goal_rounds_done"] = round_no + task["goal_last_verify"] = last_verify + task["goal_last_turn_status"] = last_turn_status + # 检查点：非终局轮次先记 goal-running，终局轮次由下面 + # 统一改写；这样被杀后 state 里能看出循环没跑完。 + task["status"] = "goal-running" + save_state(state_path, state) + print(f"round {round_no}: turn={result['status']} " + f"verify_rc={v_rc} verify_out={last_verify[:300]}") + if v_rc == 0: + ok = True + break + if budget_stopped: + task["status"] = "interrupted" + task["goal_stopped_reason"] = "budget exceeded: max_daily_turns" + else: + task["status"] = "goal-met" if ok else "goal-unmet" + if task["status"] == "goal-unmet": + notify_failure(state_path, cfg, task["name"], + {"status": "goal-unmet", "text": last_verify[:200]}) + save_state(state_path, state) + if budget_stopped: + print("goal stopped after rounds: budget exceeded") + else: + print(f"goal {'met' if ok else 'unmet'} after rounds") + if goal_usage: + print(f"usage input={goal_usage['input_tokens']} " + f"cached={goal_usage['cached_input_tokens']} " + f"cache_write={goal_usage['cache_write_input_tokens']} " + f"output={goal_usage['output_tokens']} " + f"reasoning={goal_usage['reasoning_output_tokens']} " + f"total={goal_usage['total_tokens']} " + f"completions={goal_usage['completions']}") + return {"ok": ok, "budget_stopped": budget_stopped, + "rounds_done": task["goal_rounds_done"], + "last_verify": last_verify, "usage": goal_usage} + + def _load_config_file(path): if path and os.path.exists(path): try: @@ -1487,6 +1660,10 @@ def main(): ap.add_argument("--sandbox", default=None) ap.add_argument("--approval-policy", default=None) ap.add_argument("--approval", default=None, choices=["auto", "deny"]) + ap.add_argument("--turn-timeout", type=int, default=None, + help="单轮等待上限（秒）；0 表示不限时、对齐 Codex App " + "等 turn/completed 的语义；不传时用配置 " + "turn_timeout，未配则默认 600") sub = ap.add_subparsers(dest="cmd", required=True) p = sub.add_parser("init") @@ -1533,6 +1710,7 @@ def main(): p = sub.add_parser("fork"); p.add_argument("--task", required=True); p.add_argument("--name", required=True); p.add_argument("--sandbox", default=None) p = sub.add_parser("compact"); p.add_argument("--task", required=True) p = sub.add_parser("run"); p.add_argument("--task", required=True); p.add_argument("--prompt", required=True) + p = sub.add_parser("interrupt"); p.add_argument("--task", required=True) p = sub.add_parser("delegate") p.add_argument("--parent-task", required=True); p.add_argument("--task", required=True) p.add_argument("--name", required=True); p.add_argument("--report-to-parent", action="store_true") @@ -1540,6 +1718,14 @@ def main(): p.add_argument("--name", required=True); p.add_argument("--task", required=True) p.add_argument("--repo", default=None); p.add_argument("--verify", required=True) p.add_argument("--max-rounds", type=int, default=3) + p.add_argument("--verify-timeout", type=int, default=600, + help="验测命令上限（秒）；0 表示不限时，默认 600") + p = sub.add_parser("goal-resume") + p.add_argument("--task", required=True) + p.add_argument("--max-rounds", type=int, default=None, + help="不传时沿用原任务的 goal_max_rounds；传了则覆盖总轮次") + p.add_argument("--verify-timeout", type=int, default=600, + help="验测命令上限（秒）；0 表示不限时，默认 600") p = sub.add_parser("parallel"); p.add_argument("--jobs", required=True) p = sub.add_parser("automations"); p.add_argument("--add", action="store_true") p.add_argument("--name"); p.add_argument("--task"); p.add_argument("--repo") @@ -1590,6 +1776,12 @@ def main(): args.sandbox = args.sandbox or cfg.get("sandbox") args.approval_policy = args.approval_policy or cfg.get("approval_policy") or "never" args.approval = args.approval or cfg.get("approval") or "deny" + # turn-timeout 解析：CLI > 配置 > 默认 600；0 保持为 0（不限时）， + # 不能用 or 串联，否则 0 会被当成缺失而吞掉。 + if args.turn_timeout is None: + args.turn_timeout = cfg.get("turn_timeout") + if args.turn_timeout is None: + args.turn_timeout = 600 args.codex_home = args.codex_home or _resolve_path(cfg_dir, cfg.get("codex_home")) args.codex_bin = args.codex_bin or _resolve_path(cfg_dir, cfg.get("codex_bin")) args.state = args.state or _resolve_path(cfg_dir, cfg.get("state")) or DEFAULT_STATE @@ -1849,7 +2041,8 @@ def main(): save_state(args.state, state) result = run_task_turn_with_retry(srv, state, task, args.task, args.model, args.effort, - args.sandbox, cfg=cfg) + args.sandbox, cfg=cfg, + timeout=args.turn_timeout) if result["status"] != "completed": notify_failure(args.state, cfg, args.name, result) save_state(args.state, state) @@ -1910,7 +2103,8 @@ def main(): return srv.resume_thread(t["thread_id"]) result = run_task_turn(srv, state, t, build_fix_prompt(t), - args.model, args.effort, args.sandbox) + args.model, args.effort, args.sandbox, + timeout=args.turn_timeout) if result["status"] == "completed": for c in oc: c["status"] = "resolved" @@ -1994,11 +2188,30 @@ def main(): srv.resume_thread(t["thread_id"]) result = run_task_turn_with_retry(srv, state, t, args.prompt, args.model, args.effort, - args.sandbox, cfg=cfg) + args.sandbox, cfg=cfg, + timeout=args.turn_timeout) if result["status"] != "completed": notify_failure(args.state, cfg, args.task, result) save_state(args.state, state) print(json.dumps(result, indent=2, ensure_ascii=False)) + elif args.cmd == "interrupt": + t = state["tasks"][args.task] + turn_id = None + for turn in reversed(t.get("turns", [])): + turn_id = (turn.get("result") or {}).get("turn_id") + if turn_id: + break + if not turn_id: + print(f"interrupt {args.task}: 没有可中断的 turn_id") + else: + srv.resume_thread(t["thread_id"]) + resp = srv.interrupt_turn(t["thread_id"], turn_id) + # 只有服务端接受中断才改状态；它说“没有活动回合” + # 时说明该轮已结束，硬标 interrupted 会掩盖真实终态。 + if resp and "error" not in resp: + t["status"] = "interrupted" + save_state(args.state, state) + print(f"interrupt {args.task}: turn={turn_id} resp={resp}") elif args.cmd == "delegate": parent = state["tasks"][args.parent_task] # 子 thread：cwd 与父任务相同 worktree 时只读更安全， @@ -2014,7 +2227,8 @@ def main(): state["tasks"][args.name] = sub save_state(args.state, state) result = run_task_turn(srv, state, sub, args.task, - args.model, args.effort, args.sandbox) + args.model, args.effort, args.sandbox, + timeout=args.turn_timeout) save_state(args.state, state) print(json.dumps({"sub_task": args.name, "status": sub["status"], "text": result.get("text", "")}, @@ -2024,7 +2238,8 @@ def main(): rep = run_task_turn(srv, state, parent, f"子任务 {args.name} 的结论如下，请汇总到主任务结论里，" f"不要重复做子任务的工作：\n{result.get('text', '')}", - args.model, args.effort, args.sandbox) + args.model, args.effort, args.sandbox, + timeout=args.turn_timeout) save_state(args.state, state) print(f"reported to parent: {rep.get('text', '')[:1000]}") elif args.cmd == "goal": @@ -2032,46 +2247,35 @@ def main(): args.model, args.effort, args.sandbox, args.approval_policy) save_state(args.state, state) - ok = False - last_verify = "" - goal_usage = None - for round_no in range(1, args.max_rounds + 1): - if round_no == 1: - prompt = args.task - else: - # 生产教训：只说“验收未通过” agent 不知道改什么， - # 2026-10-25 实测 prod-retry 3 轮都原地不动。 - # 必须把验收命令和失败输出喂回去，才是可用的目标循环。 - prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n" - f"目标：{args.task}\n" - f"验收命令：{args.verify}\n" - f"上一轮验收输出：\n{last_verify}") - result = run_task_turn(srv, state, task, prompt, - args.model, args.effort, args.sandbox) - goal_usage = _add_usage(goal_usage, result.get("usage")) - v = subprocess.run(args.verify, shell=True, cwd=task["cwd"], - capture_output=True, text=True) - last_verify = (v.stdout + v.stderr).strip()[:1000] or \ - f"exit code {v.returncode}" - print(f"round {round_no}: turn={result['status']} " - f"verify_rc={v.returncode} verify_out={last_verify[:300]}") - if v.returncode == 0: - ok = True - break - task["status"] = "goal-met" if ok else "goal-unmet" - if not ok: - notify_failure(args.state, cfg, args.name, - {"status": "goal-unmet", "text": last_verify[:200]}) - save_state(args.state, state) - print(f"goal {'met' if ok else 'unmet'} after rounds") - if goal_usage: - print(f"usage input={goal_usage['input_tokens']} " - f"cached={goal_usage['cached_input_tokens']} " - f"cache_write={goal_usage['cache_write_input_tokens']} " - f"output={goal_usage['output_tokens']} " - f"reasoning={goal_usage['reasoning_output_tokens']} " - f"total={goal_usage['total_tokens']} " - f"completions={goal_usage['completions']}") + run_goal_loop(srv, state, args.state, task, args.task, + args.verify, args.max_rounds, args.model, + args.effort, args.sandbox, cfg=cfg, + timeout=args.turn_timeout, + verify_timeout=args.verify_timeout) + elif args.cmd == "goal-resume": + t = state["tasks"][args.task] + if t.get("status") not in ("goal-running", "interrupted", + "running"): + print(f"goal-resume 拒绝：{args.task} 当前状态是 " + f"{t.get('status')}，已终局或已达标；" + f"goal-unmet / goal-met 后请新建任务，不要同任务续跑") + return + if not t.get("goal") or not t.get("verify_command"): + print(f"goal-resume 拒绝：{args.task} 没有 goal/verify_command 记录") + return + max_rounds = args.max_rounds or t.get("goal_max_rounds") or 3 + srv.resume_thread(t["thread_id"]) + last_turn = (t.get("turns") or [{}])[-1] + last_result = last_turn.get("result") or {} + run_goal_loop(srv, state, args.state, t, t["goal"], + t["verify_command"], max_rounds, + args.model, args.effort, args.sandbox, cfg=cfg, + timeout=args.turn_timeout, + verify_timeout=args.verify_timeout, + start_round=int(t.get("goal_rounds_done") or 0) + 1, + last_verify=t.get("goal_last_verify") or "", + last_turn_status=t.get("goal_last_turn_status"), + last_turn_text=last_result.get("text", "") or "") elif args.cmd == "parallel": jobs = json.load(open(args.jobs)) # 依赖 + 优先级：先拓扑排序，缺依赖/成环直接报错不跑； @@ -2156,7 +2360,8 @@ def main(): job, t = job_task return job["name"], run_task_turn_with_retry( srv, state, t, job["task"], args.model, - args.effort, args.sandbox, cfg=cfg) + args.effort, args.sandbox, cfg=cfg, + timeout=args.turn_timeout) with ThreadPoolExecutor(max_workers=len(ready)) as ex: for name, result in ex.map( _run, [tasks[j["name"]] for j in ready]): @@ -2191,7 +2396,8 @@ def main(): args.approval_policy) save_state(args.state, state) result = run_task_turn(srv, state, task, auto["task"], - args.model, args.effort, args.sandbox) + args.model, args.effort, args.sandbox, + timeout=args.turn_timeout) auto["last_run"] = now if result["status"] != "completed": notify_failure(args.state, cfg, name, result) diff --git a/tests/test_host.py b/tests/test_host.py index be8b548..8f28956 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -651,6 +651,245 @@ class TurnUsageSumTest(unittest.TestCase): self.assertEqual(result["usage"]["completions"], 2) +class LongTurnAlignmentTest(unittest.TestCase): + def test_partial_text_collects_agent_messages_only(self): + events = [ + {"method": "item/completed", + "params": {"item": {"type": "agentMessage", + "text": "first part"}}}, + {"method": "item/agentMessage/delta", + "params": {"delta": "should be ignored"}}, + {"method": "item/completed", + "params": {"item": {"type": "toolCall", + "text": "not a message"}}}, + {"method": "item/completed", + "params": {"item": {"type": "agentMessage", + "text": "second part"}}}, + ] + self.assertEqual(host._partial_text_from_events(events), + "first part\nsecond part") + self.assertEqual(host._partial_text_from_events([]), "") + + def test_timeout_maps_to_interrupted_and_threads_timeout(self): + class FakeServer: + def __init__(self): + self.seen_timeout = None + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + self.seen_timeout = timeout + return {"status": "timeout", "text": "partial", + "turn_id": "u1"} + + task = {"name": "t", "thread_id": "x", "sandbox": None, + "max_retries": 0, "retry_count": 0, "turns": []} + state = {"tasks": {"t": task}, "automations": []} + srv = FakeServer() + result = host.run_task_turn(srv, state, task, "do it", "m", "low", + timeout=0) + self.assertEqual(result["status"], "timeout") + self.assertEqual(task["status"], "interrupted") + self.assertEqual(task["last_text"], "partial") + self.assertEqual(srv.seen_timeout, 0) + + def test_timeout_none_keeps_old_fake_server_working(self): + class OldFakeServer: + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None): + return {"status": "completed", "text": "ok"} + + task = {"name": "t", "thread_id": "x", "sandbox": None, + "max_retries": 0, "retry_count": 0, "turns": []} + state = {"tasks": {"t": task}, "automations": []} + result = host.run_task_turn(OldFakeServer(), state, task, "do it", + "m", "low") + self.assertEqual(result["status"], "completed") + self.assertEqual(task["status"], "completed") + + def test_interrupt_turn_sends_thread_and_turn_ids(self): + srv = object.__new__(host.Server) + seen = {} + + def fake_request(method, params=None, timeout=120): + seen["method"] = method + seen["params"] = params + seen["timeout"] = timeout + return {"result": {}} + + srv.request = fake_request + self.assertEqual(srv.interrupt_turn("th1", "tu1"), + {"result": {}}) + self.assertEqual(seen["method"], "turn/interrupt") + self.assertEqual(seen["params"], + {"threadId": "th1", "turnId": "tu1"}) + self.assertIsNone(srv.interrupt_turn("th1", None)) + self.assertIsNone(srv.interrupt_turn(None, "tu1")) + + +class GoalLoopLongTaskTest(unittest.TestCase): + def _task(self, tmp): + return {"name": "g", "thread_id": "x", "cwd": tmp, "sandbox": None, + "max_retries": 0, "retry_count": 0, "turns": []} + + def test_goal_met_on_first_round_checkpoints(self): + class FakeServer: + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + return {"status": "completed", "text": "done", + "turn_id": "u1"} + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + result = host.run_goal_loop(FakeServer(), state, state_path, + task, "make it", "true", 3, "m", + "low", None, cfg={}, timeout=600) + self.assertTrue(result["ok"]) + self.assertEqual(task["status"], "goal-met") + self.assertEqual(task["goal_rounds_done"], 1) + self.assertEqual(task["goal"], "make it") + self.assertEqual(task["verify_command"], "true") + saved = host.load_state(state_path) + self.assertEqual(saved["tasks"]["g"]["status"], "goal-met") + + def test_goal_unmet_after_max_rounds_and_timeout_feedback(self): + class FakeServer: + def __init__(self): + self.prompts = [] + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + self.prompts.append(prompt) + if len(self.prompts) == 1: + return {"status": "timeout", "text": "partial work", + "turn_id": "u1"} + return {"status": "completed", "text": "done", + "turn_id": "u2"} + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + srv = FakeServer() + result = host.run_goal_loop(srv, state, state_path, + task, "make it", "false", 2, "m", + "low", None, cfg={}, timeout=600) + self.assertFalse(result["ok"]) + self.assertEqual(task["status"], "goal-unmet") + self.assertEqual(task["goal_rounds_done"], 2) + # 第二轮的提示必须说清是回合中断，不是验收失败。 + self.assertIn("timeout", srv.prompts[1]) + self.assertIn("未完成", srv.prompts[1]) + saved = host.load_state(state_path) + self.assertEqual(saved["tasks"]["g"]["goal_rounds_done"], 2) + + def test_goal_resume_continues_from_checkpoint(self): + class FakeServer: + def __init__(self): + self.prompts = [] + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + self.prompts.append(prompt) + return {"status": "completed", "text": "done", + "turn_id": "u9"} + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + srv = FakeServer() + # 模拟被杀在检查点后的任务：已跑 1 轮、验测失败、状态 goal-running。 + task["status"] = "goal-running" + task["goal"] = "make it" + task["verify_command"] = "true" + task["goal_max_rounds"] = 3 + task["goal_rounds_done"] = 1 + task["goal_last_verify"] = "missing file" + task["goal_last_turn_status"] = "completed" + task["turns"] = [{"prompt": "make it", + "result": {"status": "completed", + "text": "partial"}}] + result = host.run_goal_loop(srv, state, state_path, task, + "make it", "true", 3, "m", "low", + None, cfg={}, timeout=600, + start_round=2, + last_verify="missing file", + last_turn_status="completed") + self.assertTrue(result["ok"]) + self.assertEqual(task["status"], "goal-met") + self.assertEqual(task["goal_rounds_done"], 2) + self.assertIn("missing file", srv.prompts[-1]) + + def test_goal_fields_saved_before_first_turn(self): + class FakeServer: + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + raise RuntimeError("killed during turn") + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + with self.assertRaises(RuntimeError): + host.run_goal_loop(FakeServer(), state, state_path, + task, "make it", "true", 3, "m", "low", + None, cfg={}, timeout=600) + saved = host.load_state(state_path) + saved_task = saved["tasks"]["g"] + self.assertEqual(saved_task["goal"], "make it") + self.assertEqual(saved_task["verify_command"], "true") + self.assertEqual(saved_task["goal_rounds_done"], 0) + self.assertEqual(saved_task["status"], "goal-running") + + def test_goal_budget_stop_is_interrupted_not_unmet(self): + class FakeServer: + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + raise AssertionError("budget stop must not run a turn") + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + today = __import__("time").strftime("%Y-%m-%d") + state["usage"] = {"date": today, "turns": 5, "failed_turns": 0} + result = host.run_goal_loop(FakeServer(), state, state_path, + task, "make it", "true", 3, "m", + "low", None, + cfg={"max_daily_turns": 5}, + timeout=600) + self.assertTrue(result["budget_stopped"]) + self.assertFalse(result["ok"]) + self.assertEqual(task["status"], "interrupted") + + def test_goal_verify_timeout_is_a_verify_failure(self): + class FakeServer: + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None): + return {"status": "completed", "text": "done", + "turn_id": "u1"} + + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + state = {"tasks": {}, "automations": []} + task = self._task(tmp) + state["tasks"]["g"] = task + result = host.run_goal_loop(FakeServer(), state, state_path, + task, "make it", "sleep 5", 1, "m", + "low", None, cfg={}, timeout=600, + verify_timeout=1) + self.assertFalse(result["ok"]) + self.assertEqual(task["status"], "goal-unmet") + self.assertIn("verify timeout", task["goal_last_verify"]) + + if __name__ == "__main__": unittest.main()