feat: 长任务补齐——流式进度 + 跨进程中断 + worker 超时约定

- 长等待流式进度：run_turn 改 1 秒切片轮询（_wait_turn），每满 --progress-interval（默认 30 秒，0 关闭，CLI > 配置 > 默认）往 stderr 打 turn progress 并往链路文件记 kind=progress；进度绝不走 stdout，不破坏 dispatch JSON 与 goal 结论行/用量行解析（经 cloud 真正则验过）
- 跨进程中断：interrupt 直接不生效时写 <state_dir>/interrupt-requests/<task>.json，属主等 turn 时每秒轮询到后用自己的 live 连接发 turn/interrupt 并记 interrupted；10 秒内未响应也不删文件，属主下次轮询仍处理；接线覆盖 dispatch/run/goal/goal-resume/delegate/parallel/fix-review/run-due；旧 server 不接受 interrupt_file 时退回不传再跑
- cloud worker 超时约定：--task-timeout 0 归一化为 None（不限时，与 host turn_timeout 同约定，Docker 路径仍受 docker_timeout 约束），默认 1800 不变；超时仍记 failed（现有单测合同不动）

测试：host 56 全过（新增 7）、cloud 526 全过（4 跳过，新增 1 个 0=不限时）。

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