Merge pull request #63 from openqj/goal-long-task

CI verify pass 2m10s，合并后走新链路推生产。

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 已分流处理 | 已实测（修过一轮才通） |

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)

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()