Merge pull request #64 from openqj/goal-align-2

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

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` 已接协议，直接中断不生效时写 `<state_dir>/interrupt-requests/<task>.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 已分流处理 | 已实测（修过一轮才通） |

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

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

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 的失败不重试）")

host.py

@@ -111,6 +111,75 @@ def _partial_text_from_events(events):
     return "\n".join(parts)
 
 
+def _interrupt_request_path(state_path, task_name):
+    """跨进程中断请求文件路径：<state_dir>/interrupt-requests/<task>.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)

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