diff --git a/README.md b/README.md index 08ce2b8..811840c 100644 --- a/README.md +++ b/README.md @@ -143,7 +143,8 @@ turn 可以真正并发(不是轮流排队)。 | 沙箱边界 | read-only thread 的写文件请求被拦下,文件确实没建出来 | 已实测 | | 审批控制 | untrusted + deny 时 agent 明确收到审批拒绝,没拿到命令输出 | 已实测 | | 模型内子代理 | 3 子代理并行检查并汇总,视角是真的不同,不是重复一句话 | 已实测 | -| 中断 / 续跑控制 | `turn/interrupt` 后状态为 interrupted;`host.py interrupt --task` 已接协议;`compact` 需先 resume(host.py 的 compact 命令已包好这一步);超时回合也记 interrupted 且保留部分文本 | 已实测(interrupt 真 server 回 no active turn 时不改状态) | +| 中断 / 续跑控制 | `turn/interrupt` 后状态为 interrupted;`host.py interrupt --task` 已接协议,直接中断不生效时写 `/interrupt-requests/.json`,属主在等 turn 时每秒轮询到后用自己的 live 连接自中断(跨进程可靠路径);`compact` 需先 resume(host.py 的 compact 命令已包好这一步);超时回合也记 interrupted 且保留部分文本 | 单测已过(等待循环、请求路径、透传);真 server 跨进程中断尚未实弹,属主自中断逻辑与直接中断共享同一协议调用 | +| 长等待进度 | 等 turn 时每满 `--progress-interval`(默认 30 秒,0 关闭)往 stderr 打一行 `turn progress: elapsed=…s` 并往链路文件记一条 `kind=progress`(console 按 fallback 当 JSON 行展示);进度绝不走 stdout,不破坏 dispatch 的 JSON 与 goal 的结论行/用量行解析 | 单测已过(等待循环进度回调);真模型长等待下的进度尚未实弹 | | 合并冲突处理 | 先合的进主仓,后合同一行的冲突 abort,主仓干净,冲突 worktree 留着人工解 | 已实测 | | 自动化幂等 | 未到期的自动化 run-due 时不重复派发 | 已实测 | | 宿主提供 MCP 工具 | 隔离 CODEX_HOME 注册 stdio MCP,agent 能发现并调用;注意 MCP 审批是 elicitation 协议,不是通用 decision 字段,host.py 已分流处理 | 已实测(修过一轮才通) | diff --git a/cloud/README.md b/cloud/README.md index e3461bb..e94a5d6 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -683,7 +683,9 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 `GET /metrics` 的 `by_status` 恒含七个状态键(queued / running / completed / failed / interrupted / merged / merge-failed),缺失 状态为 0 而不是缺键。 -- 任务级超时:worker 的 `--task-timeout`(默认 1800 秒),与另外 +- 任务级超时:worker 的 `--task-timeout`(默认 1800 秒,0 表示 + 不限时、与 host `--turn-timeout 0` 同约定,worker 级不设截止, + 但 Docker 路径仍受 `--docker-timeout` 约束),与另外 两个参数不要混淆:`--docker-timeout` 是执行器级的单次 docker run 超时,`--max-seconds` 是 `run_loop` 的整体墙钟上限;三者互不覆盖, Docker 路径下 worker 取 `min(task_timeout, docker_timeout)` diff --git a/cloud/tests/test_prod_worker_timeout.py b/cloud/tests/test_prod_worker_timeout.py index f0fe161..1f8cf1d 100644 --- a/cloud/tests/test_prod_worker_timeout.py +++ b/cloud/tests/test_prod_worker_timeout.py @@ -123,6 +123,27 @@ class WorkerTimeoutTestCase(unittest.TestCase): self.assertIn("worker_timeout", _audit_actions(db_path)) tmp.cleanup() + def test_task_timeout_zero_means_unlimited(self): + # 对齐 host turn_timeout:0 归一化为 None(不限时), + # 而不是在 popen 轮询第一轮就立即超时。sleep 0.5 的 + # 任务在 task_timeout=0 下应正常完成,库里是 completed。 + tmp = tempfile.TemporaryDirectory() + + def factory(task): + return subprocess.Popen( + ["sleep", "0.5"], stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, start_new_session=True) + + worker, db_path, task = _make_worker( + tmp.name, process_factory=factory, task_timeout=0, + max_retries=0, retry_backoff_seconds=0) + self.assertIsNone(worker.task_timeout) + result = worker.run_one(task) + self.assertEqual(result["status"], "completed") + status, _ = _task_row(db_path) + self.assertEqual(status, "completed") + tmp.cleanup() + def test_docker_timeout_converts_to_failed_and_audit(self): tmp = tempfile.TemporaryDirectory() diff --git a/cloud/worker.py b/cloud/worker.py index 4b017ba..c1d9503 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -68,6 +68,12 @@ interrupted,不用执行结果覆盖它。 runner / 无超时的 subprocess.run)无法在中途抢占一个已阻塞 的调用,超时只能靠执行器自身的 timeout,这是诚实边界, 不要假装同步路径也能中途杀掉。 +- 对齐 host.py 的 turn_timeout 约定:task_timeout 传 0 时 + 归一化为 None(不限时,worker 级不设截止),与 host 的 + “0 = 不限时”同约定;默认 1800 不变,不改默认就不会悄悄 + 挂起自动化。worker 级不限时时,Docker 路径仍受执行器 + docker_timeout 约束(timeout=None 时 _call_runner 用 + executor 自身的 timeout),不是全链路无限。 - max_retries(默认 1 次重试,CLI --max-retries 可调,0 表示 不重试)是对模型任务失败的有限重试:取消不重试;可 fallback 的失败优先走真实 fallback(见下),不再做同模型重试(fallback @@ -311,6 +317,14 @@ class Worker: self._loop_stop = False self.logger = get_logger("cloud.worker") # 任务级超时 / 失败重试(见模块 docstring 的生产级说明)。 + # 对齐 host turn_timeout:0 归一化为 None(不限时), + # 负数同理;默认 1800 不变。归一化后 _run_attempt 里 + # task_timeout is None 的分支本来就走“无 worker 级截止” + #(Docker 路径仍受 executor docker_timeout 约束), + # 不用再改各执行路径;不归一化的话 0 会在 popen 轮询 + # 里第一轮就触发“已到截止时间”而立即超时,是反直觉的 bug。 + if task_timeout is not None and task_timeout <= 0: + task_timeout = None self.task_timeout = task_timeout self.max_retries = max_retries self.retry_backoff_seconds = retry_backoff_seconds @@ -2102,10 +2116,12 @@ def main(argv=None): " worker 必须开,否则它会抢走本该进" " Docker 容器的 NULL 任务") parser.add_argument("--task-timeout", type=float, default=1800, - help="每任务截止时间(秒,默认 1800);超时后真杀" - "容器/进程组、任务置 failed;与 --docker-timeout" - "(单次 docker run 超时)取较小值,与" - " --max-seconds(run_loop 整体墙钟上限)不同") + help="每任务截止时间(秒,默认 1800);0 表示不限时" + "(与 host --turn-timeout 0 同约定,worker 级" + "不设截止,但 Docker 路径仍受 --docker-timeout" + "约束);超时后真杀容器/进程组、任务置 failed;" + "与 --docker-timeout(单次 docker run 超时)取" + "较小值,与 --max-seconds(run_loop 整体墙钟上限)不同") parser.add_argument("--max-retries", type=int, default=1, help="模型任务失败重试次数(默认 1,0 表示不重试;" "取消与已记 fallback 的失败不重试)") diff --git a/host.py b/host.py index c2f524b..fbedb99 100644 --- a/host.py +++ b/host.py @@ -111,6 +111,75 @@ def _partial_text_from_events(events): return "\n".join(parts) +def _interrupt_request_path(state_path, task_name): + """跨进程中断请求文件路径:/interrupt-requests/.json。 + + 属主进程在等 turn 时每秒轮询这个文件,另一个进程的 + `host.py interrupt --task` 直接协议中断不生效时就写这个文件, + 属主用自己的 live 连接发 turn/interrupt——只有属主的连接 + 一定拥有这个 turn,这是跨进程中断可靠的原因。文件本身不 + 由 run_task_turn 创建,只由 interrupt 命令写、属主读到后删。 + """ + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name)) + return os.path.join( + os.path.dirname(os.path.abspath(state_path)), + "interrupt-requests", safe + ".json") + + +def _wait_turn(q, timeout, progress_interval=30, on_progress=None, + interrupt_check=None): + """等 turn/completed 的轮询等待(对齐 Codex App 的长等待语义)。 + + 原来是一次性 q.get(timeout) / q.get(),长等待全程静默、另一 + 进程无法插手;现在按 1 秒切片轮询,每片做两件事: + 1) interrupt_check() 为真时立刻返回 ("interrupted-requested", None), + 由调用方用自己的 live 连接发 turn/interrupt 再记 interrupted; + 2) progress_interval > 0 且距上次进度已满一个间隔时调 + on_progress(elapsed_seconds),让调用方打流式进度。 + 返回:("completed", done) / ("timeout", None) / + ("interrupted-requested", None)。timeout 为 None 或 <=0 时 + 不限时,只等 done 或中断请求;正数时到时返回 timeout。 + """ + start = time.monotonic() + next_progress = progress_interval if progress_interval and \ + progress_interval > 0 else None + while True: + elapsed = time.monotonic() - start + if timeout and timeout > 0: + remaining = timeout - elapsed + if remaining <= 0: + return "timeout", None + slice_s = min(1.0, remaining) + else: + slice_s = 1.0 + # 切片还要受下一个进度点约束,否则一个长切片会把 + # 中间的进度检查全吞掉(timeout=0.35、间隔=0.1 时只会 + # 打 1 次进度,而不是 3 次;单测抓住了这个 bug)。 + if next_progress is not None: + slice_s = min(slice_s, max(0.01, next_progress - elapsed)) + try: + done = q.get(timeout=slice_s) + return "completed", done + except queue.Empty: + pass + elapsed = time.monotonic() - start + if interrupt_check is not None: + try: + if interrupt_check(): + return "interrupted-requested", None + except Exception: + pass + if next_progress is not None and elapsed >= next_progress: + if on_progress is not None: + try: + on_progress(elapsed) + except Exception: + pass + next_progress += progress_interval + if timeout and timeout > 0 and elapsed >= timeout: + return "timeout", None + + def _add_usage(a, b): """两份用量相加(任一为空时返回另一份;都为空返回 None)。""" if not a: @@ -202,10 +271,14 @@ def _summarize_trace_item(method, params): class Server: """单个 app-server 进程,多 thread 并发。""" - def __init__(self, codex_bin=None, codex_home=None, approval="deny"): + def __init__(self, codex_bin=None, codex_home=None, approval="deny", + progress_interval=30): self.codex_bin = codex_bin or default_codex_bin() self.codex_home = codex_home self.approval = approval + # 流式进度间隔(秒):run_turn 等待时每满一个间隔打一次 + # 进度;0 表示关闭进度输出,但中断轮询仍每秒进行。 + self.progress_interval = progress_interval self.send_lock = threading.Lock() self.next_id = 1 self.pending = {} @@ -432,7 +505,7 @@ class Server: def run_turn(self, thread_id, prompt, model=None, effort=None, parent_turn_id=None, root_turn_id=None, timeout=600, - sandbox=None): + sandbox=None, progress_interval=None, interrupt_file=None): # 开 turn 前先看 server 是否还活着,死了就重拉并 resume # 这个 thread(thread 持久化在 CODEX_HOME 里,重拉不丢), # 否则 turn/start 会直接 BrokenPipe 失败。 @@ -477,10 +550,82 @@ class Server: # 或模型自己结束);正数才按秒计时。默认仍是 600, # 由调用方(CLI --turn-timeout / 配置 turn_timeout) # 显式改,不改默认就不会悄悄挂起自动化。 - if timeout and timeout > 0: - done = q.get(timeout=timeout) - else: - done = q.get() + # 等待不再是一次性 q.get():按 1 秒切片轮询,每满 + # progress_interval 打一次流式进度(stderr,绝不 + # 走 stdout——dispatch/run 的 stdout 是 JSON、goal + # 的 stdout 要被 worker 解析结论行与用量行,进度 + # 进 stdout 会把这两种解析都打坏;链路文件里也 + # 同步记一条 kind=progress,console 按 fallback + # 把它当 JSON 行展示),并每秒检查跨进程中断 + # 请求文件(见 _interrupt_request_path)。 + interval = self.progress_interval if \ + progress_interval is None else progress_interval + + def _on_progress(elapsed): + sys.stderr.write( + "turn progress: elapsed=%ds\n" + % int(elapsed)) + sys.stderr.flush() + if self.trace_path is not None and \ + self._trace_lines < TRACE_MAX_LINES: + try: + with self.events_lock: + n_events = len(self.events) + with open(self.trace_path, "a", + encoding="utf-8") as f: + f.write(json.dumps( + {"kind": "progress", + "elapsed_s": int(elapsed), + "events": n_events}, + ensure_ascii=False) + "\n") + self._trace_lines += 1 + except Exception: + pass + + def _interrupt_check(): + return bool(interrupt_file) and \ + os.path.exists(interrupt_file) + + wait_status, done = _wait_turn( + q, timeout, progress_interval=interval, + on_progress=_on_progress, + interrupt_check=_interrupt_check) + if wait_status == "interrupted-requested": + # 属主用自己的 live 连接发中断:只有这个连接 + # 一定拥有该 turn,另一进程的直接中断未必生效。 + # 中断请求本身已明确,属主不再继续等:即使 + # 服务端回 error 也记 interrupted(任务可续跑, + # 不是失败),并把请求文件删掉避免重复触发; + # 服务端侧的停止是 best-effort,stderr 留一行。 + try: + resp = self.interrupt_turn(thread_id, turn_id) + if resp and "error" in resp: + sys.stderr.write( + "turn interrupt request returned error: " + "%s\n" % (resp,)) + sys.stderr.flush() + except Exception as exc: + sys.stderr.write( + "turn interrupt request failed: %s\n" % (exc,)) + sys.stderr.flush() + try: + if interrupt_file: + os.remove(interrupt_file) + except Exception: + pass + with self.events_lock: + evs = list(self.events[start_events:]) + return {"status": "interrupted", + "text": _partial_text_from_events(evs), + "turn_id": turn_id, + "usage": _sum_turn_usage(evs, thread_id, turn_id)} + if wait_status == "timeout": + with self.events_lock: + evs = list(self.events[start_events:]) + return {"status": "timeout", + "text": _partial_text_from_events(evs), + "turn_id": turn_id, + "usage": _sum_turn_usage(evs, thread_id, turn_id)} except queue.Empty: with self.events_lock: evs = list(self.events[start_events:]) @@ -534,9 +679,11 @@ class Server: 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)。""" + 注意:直接中断只对本 server 进程正在跑的 turn 可靠,另一进程的 + turn 未必能被本进程的中断请求停下;跨进程的可靠路径是 + 中断请求文件(见 _interrupt_request_path):interrupt 命令 + 直接中断不生效时写文件,属主进程在等 turn 时每秒轮询到后 + 用自己的 live 连接发中断(run_turn 的 interrupted-requested 分支)。""" if not thread_id or not turn_id: return None return self.request("turn/interrupt", @@ -1442,21 +1589,36 @@ def dispatch(server, state, name, prompt, repo, model, effort, sandbox, def run_task_turn(server, state, task, prompt, model, effort, - sandbox=None, timeout=None): + sandbox=None, timeout=None, interrupt_file=None): # task 自带 sandbox 时优先用它(只读评审任务不能被全局默认覆盖), # 否则用调用方显式传的 sandbox;两者都没有时不传 sandboxPolicy, # 保留 thread 原有设置(旧任务没有 sandbox 记录时尤其重要)。 # timeout 为 None 时不传给 server(沿用 server 默认 600), # 这样旧 FakeServer(不接受 timeout 参数)不受影响;显式 - # 传了才透传,0 由 server 侧解释为不限时。 + # 传了才透传,0 由 server 侧解释为不限时。interrupt_file 同理: + # 不传时不进 server 参数,旧 FakeServer 不受影响;传了时 + # 属主在等 turn 时每秒轮询它(见 _interrupt_request_path)。 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, - **turn_kwargs) + if interrupt_file is not None: + turn_kwargs["interrupt_file"] = interrupt_file + try: + result = server.run_turn(task["thread_id"], prompt, model=model, + effort=effort, sandbox=use_sandbox, + **turn_kwargs) + except TypeError as exc: + # 兼容旧 server:不接受 interrupt_file 参数时退回不传它 + # 再跑一次(goal 单测的 FakeServer 就是这款);其它 + # TypeError 一律不吞,原样抛出。 + if "interrupt_file" not in str(exc): + raise + turn_kwargs.pop("interrupt_file", None) + result = server.run_turn(task["thread_id"], prompt, model=model, + 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, @@ -1475,7 +1637,8 @@ def run_task_turn(server, state, task, prompt, model, effort, def run_task_turn_with_retry(server, state, task, prompt, model, effort, - sandbox=None, cfg=None, timeout=None): + sandbox=None, cfg=None, timeout=None, + interrupt_file=None): """失败自动重试:默认 max_retries=0 时行为不变(只跑一次); 配了重试时,失败后同 prompt 再跑,最多 max_retries 次, 每次重试前先查预算,超预算直接停,不烧钱。""" @@ -1497,7 +1660,8 @@ 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, timeout=timeout) + model, effort, sandbox, timeout=timeout, + interrupt_file=interrupt_file) # 各次尝试的真实用量都要算(失败尝试花的钱也是钱), # 最后一次的结果里带汇总,不只带最后一次。 usage = _add_usage(usage, result.get("usage")) @@ -1566,7 +1730,10 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, f"验收命令:{verify}\n" f"上一轮验收输出:\n{last_verify}") result = run_task_turn(server, state, task, prompt, - model, effort, sandbox, timeout=timeout) + model, effort, sandbox, timeout=timeout, + interrupt_file=_interrupt_request_path( + state_path, task["name"]) + if task.get("name") else None) goal_usage = _add_usage(goal_usage, result.get("usage")) last_turn_status = result["status"] last_turn_text = result.get("text", "") or "" @@ -1664,6 +1831,11 @@ def main(): help="单轮等待上限(秒);0 表示不限时、对齐 Codex App " "等 turn/completed 的语义;不传时用配置 " "turn_timeout,未配则默认 600") + ap.add_argument("--progress-interval", type=float, default=None, + help="长等待流式进度间隔(秒);每满一个间隔往 stderr " + "打一行 turn progress 并往链路文件记一条 " + "kind=progress;0 表示关闭进度输出;不传时用配置 " + "progress_interval,未配则默认 30") sub = ap.add_subparsers(dest="cmd", required=True) p = sub.add_parser("init") @@ -1782,6 +1954,12 @@ def main(): args.turn_timeout = cfg.get("turn_timeout") if args.turn_timeout is None: args.turn_timeout = 600 + # progress-interval 解析:同款 CLI > 配置 > 默认 30;0 保持为 0 + #(关闭进度输出),不能用 or 串联。 + if args.progress_interval is None: + args.progress_interval = cfg.get("progress_interval") + if args.progress_interval is None: + args.progress_interval = 30 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 @@ -2010,7 +2188,8 @@ def main(): print("host daemon stopped", flush=True) return srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home, - approval=args.approval) + approval=args.approval, + progress_interval=args.progress_interval) # 实时链路:只给跑 turn 的命令设链路文件(dispatch/run/goal/ # delegate);只读命令不设。名字映射与 worker 派发一致: # dispatch/goal/delegate 用 --name,run 用 --task。 @@ -2042,7 +2221,8 @@ def main(): result = run_task_turn_with_retry(srv, state, task, args.task, args.model, args.effort, args.sandbox, cfg=cfg, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, args.name)) if result["status"] != "completed": notify_failure(args.state, cfg, args.name, result) save_state(args.state, state) @@ -2104,7 +2284,8 @@ def main(): srv.resume_thread(t["thread_id"]) result = run_task_turn(srv, state, t, build_fix_prompt(t), args.model, args.effort, args.sandbox, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, args.task)) if result["status"] == "completed": for c in oc: c["status"] = "resolved" @@ -2189,7 +2370,8 @@ def main(): result = run_task_turn_with_retry(srv, state, t, args.prompt, args.model, args.effort, args.sandbox, cfg=cfg, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, args.task)) if result["status"] != "completed": notify_failure(args.state, cfg, args.task, result) save_state(args.state, state) @@ -2206,12 +2388,57 @@ def main(): else: srv.resume_thread(t["thread_id"]) resp = srv.interrupt_turn(t["thread_id"], turn_id) - # 只有服务端接受中断才改状态;它说“没有活动回合” - # 时说明该轮已结束,硬标 interrupted 会掩盖真实终态。 + # 直接中断生效(快路径)时沿用老行为:只有服务端 + # 接受中断才改状态;它说“没有活动回合”时说明该轮 + # 已结束,硬标 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}") + req = _interrupt_request_path(args.state, args.task) + try: + if os.path.exists(req): + os.remove(req) + except Exception: + pass + print(f"interrupt {args.task}: turn={turn_id} resp={resp}") + else: + # 直接中断不生效(典型是另一进程在等这个 turn, + # 本进程的 server 并不拥有它):写中断请求文件, + # 属主进程在等 turn 时每秒轮询到后用自己的 + # live 连接发 turn/interrupt 并记 interrupted。 + # 写完后最多等 10 秒看属主是否已响应,超时也 + # 不删文件——属主下次轮询时仍会处理,调用方 + # 也可 Ctrl-C 属主进程 + recover 兜底。 + req = _interrupt_request_path(args.state, args.task) + try: + os.makedirs(os.path.dirname(req), exist_ok=True) + with open(req, "w", encoding="utf-8") as f: + json.dump({"task": args.task, + "thread_id": t["thread_id"], + "turn_id": turn_id, + "requested_at": time.time()}, + f, ensure_ascii=False) + except Exception as exc: + print(f"interrupt {args.task}: 写中断请求失败:{exc}") + return + print(f"interrupt {args.task}: 直接中断未生效 " + f"(resp={resp}),已写中断请求 {req}," + f"等待属主进程处理") + for _ in range(10): + time.sleep(1) + try: + cur = json.load(open(args.state, + encoding="utf-8")) + if (cur.get("tasks", {}).get(args.task, {}) + .get("status") == "interrupted"): + print(f"interrupt {args.task}: 属主已响应," + f"任务已记 interrupted") + break + except Exception: + pass + else: + print(f"interrupt {args.task}: 属主 10 秒内未响应," + f"请求文件仍在 {req},属主下次轮询时会处理") elif args.cmd == "delegate": parent = state["tasks"][args.parent_task] # 子 thread:cwd 与父任务相同 worktree 时只读更安全, @@ -2228,7 +2455,8 @@ def main(): save_state(args.state, state) result = run_task_turn(srv, state, sub, args.task, args.model, args.effort, args.sandbox, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, args.name)) save_state(args.state, state) print(json.dumps({"sub_task": args.name, "status": sub["status"], "text": result.get("text", "")}, @@ -2239,7 +2467,8 @@ def main(): f"子任务 {args.name} 的结论如下,请汇总到主任务结论里," f"不要重复做子任务的工作:\n{result.get('text', '')}", args.model, args.effort, args.sandbox, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, args.parent_task)) save_state(args.state, state) print(f"reported to parent: {rep.get('text', '')[:1000]}") elif args.cmd == "goal": @@ -2361,7 +2590,8 @@ def main(): return job["name"], run_task_turn_with_retry( srv, state, t, job["task"], args.model, args.effort, args.sandbox, cfg=cfg, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, job["name"])) with ThreadPoolExecutor(max_workers=len(ready)) as ex: for name, result in ex.map( _run, [tasks[j["name"]] for j in ready]): @@ -2397,7 +2627,8 @@ def main(): save_state(args.state, state) result = run_task_turn(srv, state, task, auto["task"], args.model, args.effort, args.sandbox, - timeout=args.turn_timeout) + timeout=args.turn_timeout, + interrupt_file=_interrupt_request_path(args.state, name)) 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 8f28956..5a80df7 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -890,6 +890,101 @@ class GoalLoopLongTaskTest(unittest.TestCase): self.assertIn("verify timeout", task["goal_last_verify"]) + +class ProgressInterruptTest(unittest.TestCase): + """长等待流式进度 + 跨进程中断请求:只测可单测的等待循环与 + 路径/透传,不跑模型、不拉 app-server。""" + + def test_wait_turn_completed(self): + import queue + import threading + + q = queue.Queue() + + def _put(): + import time as _time + _time.sleep(0.05) + q.put({"turn": "done"}) + + threading.Thread(target=_put, daemon=True).start() + status, done = host._wait_turn(q, timeout=5) + self.assertEqual(status, "completed") + self.assertEqual(done, {"turn": "done"}) + + def test_wait_turn_timeout(self): + import queue + import time as _time + + q = queue.Queue() + start = _time.monotonic() + status, done = host._wait_turn(q, timeout=0.2, + progress_interval=10) + elapsed = _time.monotonic() - start + self.assertEqual(status, "timeout") + self.assertIsNone(done) + self.assertGreaterEqual(elapsed, 0.2) + self.assertLess(elapsed, 1.0) + + def test_wait_turn_interrupt_request(self): + import queue + + q = queue.Queue() + calls = [] + + def _check(): + calls.append(1) + return len(calls) >= 2 + + status, done = host._wait_turn(q, timeout=5, + interrupt_check=_check) + self.assertEqual(status, "interrupted-requested") + self.assertIsNone(done) + + def test_wait_turn_progress_calls(self): + import queue + + q = queue.Queue() + seen = [] + status, _ = host._wait_turn(q, timeout=0.35, + progress_interval=0.1, + on_progress=seen.append) + self.assertEqual(status, "timeout") + self.assertGreaterEqual(len(seen), 2) + self.assertEqual(seen, sorted(seen)) + + def test_interrupt_request_path(self): + path = host._interrupt_request_path("/tmp/abc/host-state.json", + "my task/a") + self.assertEqual(path, "/tmp/abc/interrupt-requests/my-task-a.json") + + def test_run_task_turn_passes_interrupt_file(self): + class RecordingServer: + def __init__(self): + self.seen = None + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None, timeout=None, interrupt_file=None): + self.seen = interrupt_file + return {"status": "interrupted", "text": "part", + "turn_id": "t1"} + + srv = RecordingServer() + task = {"name": "g", "thread_id": "x", "sandbox": None, + "turns": []} + state = {"tasks": {"g": task}, "automations": []} + result = host.run_task_turn(srv, state, task, "do it", "m", + "low", interrupt_file="/tmp/req.json") + self.assertEqual(result["status"], "interrupted") + self.assertEqual(task["status"], "interrupted") + self.assertEqual(srv.seen, "/tmp/req.json") + + def test_progress_line_is_parse_safe(self): + line = "turn progress: elapsed=30s" + self.assertNotIn("goal met", line) + self.assertNotIn("goal unmet", line) + self.assertNotIn("usage input=", line) + + if __name__ == "__main__": unittest.main()