Merge pull request #139 from openqj/feat-longtask-p0-three

长任务 P0 三件：断点进库、错误分类+熔断、Run/Step 时间线

cloud/db.py

@@ -805,6 +805,32 @@ POSTGRES_SCHEMA = (
     "action TEXT, "
     "detail TEXT, "
     "created_at DOUBLE PRECISION)",
+    # P0 三件：断点与步骤事实源（PG 新库一次建齐；老库走
+    # server/worker 的 ensure 幂等补表，不丢数据）。
+    # task_checkpoints：每轮断点快照（worker 镜像线程 + 终局
+    # 导入写），(task_id, round_no) 唯一；断点丢失时可据此
+    # 重建 host-state 续跑，不再只信容器里的 host-state.json。
+    "CREATE TABLE IF NOT EXISTS task_checkpoints("
+    "task_id TEXT, "
+    "round_no INTEGER, "
+    "thread_id TEXT, "
+    "goal_status TEXT, "
+    "state_sha TEXT, "
+    "verify_out TEXT, "
+    "created_at DOUBLE PRECISION, "
+    "PRIMARY KEY(task_id, round_no))",
+    # task_steps：Run/Step 时间线（worker 侧 claim/resume/final
+    # + host 每轮 round + 镜像导入），timeline 接口直读此表。
+    "CREATE TABLE IF NOT EXISTS task_steps("
+    "id TEXT PRIMARY KEY, "
+    "task_id TEXT, "
+    "step_no INTEGER, "
+    "kind TEXT, "
+    "summary TEXT, "
+    "state_sha TEXT, "
+    "created_at DOUBLE PRECISION)",
+    "CREATE INDEX IF NOT EXISTS idx_task_steps_task"
+    " ON task_steps(task_id, step_no)",
     "CREATE TABLE IF NOT EXISTS projects("
     "id TEXT PRIMARY KEY, "
     "name TEXT, "

cloud/migrations.py

@@ -75,6 +75,11 @@ MIGRATION_TABLES = (
     "memory_settings",
     "task_memory_flags",
     "memory_extraction_log",
+    # P0 三件：断点与步骤表（TEXT 主键 / 复合主键，无身份列，
+    # 不需要序列拨动；sqlite 侧由 server/worker ensure 建出，
+    # 老库没有时迁移记 0，不中断其余表）。
+    "task_checkpoints",
+    "task_steps",
 )
 
 _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"})

cloud/sandbox.py

@@ -117,6 +117,57 @@ def _parse_goal_output(stdout):
     return None, None
 
 
+# P0 三件其二：goal 错误签名。permanent = 同代码、同断点、同配置
+# 下重试必然再失败（断点丢失、配置缺失、旧语义误用），必须判
+# failed 终态进死信，绝不能判 interrupted 进自动恢复；transient =
+# 网络/网关抖动，重试有意义，走原有 interrupted 路径。未知错误
+# 不分类（None），走原有逻辑，兜底靠 worker 的无进展熔断。
+_GOAL_PERMANENT_MARKERS = (
+    "goal-resume 拒绝",
+    "KeyError",
+    "没有会话线程",
+    "没有 goal 记录",
+    "provider not found",
+    "Model provider",
+    "Missing environment variable",
+    "任务已存在",
+)
+_GOAL_TRANSIENT_MARKERS = (
+    "timed out",
+    "Timeout",
+    "429",
+    "502",
+    "503",
+    "RemoteDisconnected",
+    "IncompleteRead",
+    "Connection reset",
+    "connection refused",
+)
+
+
+def _classify_goal_error(stdout, stderr):
+    """返回 (kind, snippet)：kind 为 permanent/transient/None。
+
+    snippet 取命中标记所在行（截 500 字），供终态 error_text 用；
+    命中多个时 permanent 优先于 transient（永久错误里常混着
+    网络字样，反过来会把死信误判成可重试）。
+    """
+    combined = (stdout or "") + "\n" + (stderr or "")
+    for marker in _GOAL_PERMANENT_MARKERS:
+        if marker in combined:
+            for line in combined.splitlines():
+                if marker in line:
+                    return "permanent", line.strip()[:500]
+            return "permanent", marker
+    for marker in _GOAL_TRANSIENT_MARKERS:
+        if marker in combined:
+            for line in combined.splitlines():
+                if marker in line:
+                    return "transient", line.strip()[:500]
+            return "transient", marker
+    return None, None
+
+
 def _goal_resume_refusal(stdout):
     """host.py goal-resume 的拒绝行（断点丢失时），无则返回 None。
 
@@ -584,6 +635,8 @@ class LocalExecutor(Executor):
         interrupted_reason = _goal_interrupted_reason(stdout_full)
         refusal = _goal_resume_refusal(stdout_full)
         combined = stdout_full + stderr
+        perm_kind, perm_snippet = _classify_goal_error(
+            stdout_full, stderr)
         if refusal is not None and met is None:
             # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。
             met, goal_error = None, refusal
@@ -601,6 +654,14 @@ class LocalExecutor(Executor):
             status = "interrupted"
             goal_error = (interrupted_reason
                           or "host goal 异常中断，断点可续跑")
+        elif (met is None and returncode != 0
+                and perm_kind == "permanent"):
+            # P0 三件其二：永久错误（断点 KeyError、配置缺失等）
+            # 的无结论异常退出——同代码同断点重试必然再失败，
+            # 判 failed 终态进死信，绝不判 interrupted 空转。
+            # host 已明言 interrupted 的在上一分支，不受影响。
+            status = "failed"
+            goal_error = perm_snippet or "goal 永久错误"
         elif returncode != 0:
             if met is None:
                 # 无结论的异常退出（进程被杀、连接断后恢复时抛错
@@ -1301,6 +1362,8 @@ class DockerExecutor(Executor):
         met, goal_error = _parse_goal_output(stdout_full)
         interrupted_reason = _goal_interrupted_reason(stdout_full)
         combined = stdout_full + stderr
+        perm_kind, perm_snippet = _classify_goal_error(
+            stdout_full, stderr)
         if "已存在" in combined:
             met, goal_error = None, (
                 "host 任务已存在：goal 失败请用继续而非重试——"
@@ -1314,6 +1377,13 @@ class DockerExecutor(Executor):
             status = "interrupted"
             goal_error = (interrupted_reason
                           or "host goal 异常中断，断点可续跑")
+        elif (met is None and returncode != 0
+                and perm_kind == "permanent"):
+            # P0 三件其二：永久错误无结论异常退出 → failed 死信
+            # （同 local 侧注释；host 已明言 interrupted 的在上
+            # 一分支，不受影响）。
+            status = "failed"
+            goal_error = perm_snippet or "goal 永久错误"
         elif returncode != 0:
             if met is None:
                 # 无结论的异常退出（进程被杀、连接断后恢复时抛错
@@ -1407,6 +1477,8 @@ class DockerExecutor(Executor):
         met, goal_error = _parse_goal_output(stdout_full)
         interrupted_reason = _goal_interrupted_reason(stdout_full)
         refusal = _goal_resume_refusal(stdout_full)
+        perm_kind, perm_snippet = _classify_goal_error(
+            stdout_full, stderr)
         if refusal is not None and met is None:
             # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。
             status = "failed"
@@ -1417,6 +1489,13 @@ class DockerExecutor(Executor):
             status = "interrupted"
             goal_error = (interrupted_reason
                           or "host goal-resume 异常中断，断点可续跑")
+        elif (met is None and returncode != 0
+                and perm_kind == "permanent"):
+            # P0 三件其二：续跑侧永久错误（典型如断点 KeyError）
+            # → failed 死信；任务 2 的 12 秒空转就是此分支缺失
+            # 造成的（当时无结论异常退出被判 interrupted）。
+            status = "failed"
+            goal_error = perm_snippet or "goal 永久错误"
         elif returncode != 0:
             if met is None:
                 # 无结论的异常退出：断点在盘上，记 interrupted
@@ -1508,6 +1587,8 @@ class DockerExecutor(Executor):
         met, goal_error = _parse_goal_output(stdout_full)
         interrupted_reason = _goal_interrupted_reason(stdout_full)
         refusal = _goal_resume_refusal(stdout_full)
+        perm_kind, perm_snippet = _classify_goal_error(
+            stdout_full, stderr)
         if refusal is not None and met is None:
             # 断点丢失拒绝：终态 failed，不进自动恢复（见 helper）。
             status = "failed"
@@ -1518,6 +1599,11 @@ class DockerExecutor(Executor):
             status = "interrupted"
             goal_error = (interrupted_reason
                           or "host goal-resume 异常中断，断点可续跑")
+        elif (met is None and returncode != 0
+                and perm_kind == "permanent"):
+            # P0 三件其二：续跑侧永久错误 → failed 死信（同上）。
+            status = "failed"
+            goal_error = perm_snippet or "goal 永久错误"
         elif returncode != 0:
             if met is None:
                 # 无结论的异常退出：断点在盘上，记 interrupted

cloud/server.py

@@ -1126,6 +1126,19 @@ def migrate_tasks_table(conn):
         "interrupted_reason": "TEXT",
         "auto_resume_count": "INTEGER DEFAULT 0",
         "no_time_limit": "INTEGER DEFAULT 0",
+        # P0 三件：自动恢复退避与无进展熔断（见 worker
+        # auto_resume_interrupted）：resume_after 是下次允许
+        # 重排队的时间戳（指数退避）；auto_resume_stall_count
+        # 记连续无进展重排队次数（到 3 即停、标
+        # auto_resume_stalled 等人工）；checkpoint_round_at_last_
+        # resume 记上次重排队时库中断点轮次（无进展判定的
+        # 基线，NULL = 从未记过）；last_error_signature 记
+        # 最近一次终局的错误签名（permanent/transient/None，
+        # 供排障，不作恢复判定——判定在 sandbox 终态里）。
+        "resume_after": "DOUBLE PRECISION DEFAULT 0",
+        "auto_resume_stall_count": "INTEGER DEFAULT 0",
+        "checkpoint_round_at_last_resume": "INTEGER",
+        "last_error_signature": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -1204,6 +1217,40 @@ def _attachments_map(conn, task_ids):
     return out
 
 
+def _ensure_task_checkpoints_table(conn):
+    """幂等建断点表（P0 三件其一）：server init_db（PG 与 SQLite
+    两支）与 worker 迁移兜底共用。DOUBLE PRECISION 在两后端都可用。
+    """
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_checkpoints("
+        "task_id TEXT, "
+        "round_no INTEGER, "
+        "thread_id TEXT, "
+        "goal_status TEXT, "
+        "state_sha TEXT, "
+        "verify_out TEXT, "
+        "created_at DOUBLE PRECISION, "
+        "PRIMARY KEY(task_id, round_no))")
+
+
+def _ensure_task_steps_table(conn):
+    """幂等建步骤表（P0 三件其三）：timeline 接口直读此表。
+    id 由写入方给 uuid hex（两后端都不依赖自增语义）。
+    """
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_steps("
+        "id TEXT PRIMARY KEY, "
+        "task_id TEXT, "
+        "step_no INTEGER, "
+        "kind TEXT, "
+        "summary TEXT, "
+        "state_sha TEXT, "
+        "created_at DOUBLE PRECISION)")
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_task_steps_task"
+        " ON task_steps(task_id, step_no)")
+
+
 def _ensure_followups_table(conn):
     """幂等建后续消息表：委托给 peer_messages（两边同一份 DDL，
     server init_db、各 followup 接口与 worker 投递路径共用）。
@@ -1530,6 +1577,8 @@ def init_db(db_path):
             _backfill_default_quotas(conn)
             _ensure_gateway_keys(conn)
             _ensure_uploads_tables(conn)
+            _ensure_task_checkpoints_table(conn)
+            _ensure_task_steps_table(conn)
             conn.commit()
             return
         conn.execute(
@@ -1564,6 +1613,8 @@ def init_db(db_path):
         _ensure_quotas_table(conn)
         _backfill_default_quotas(conn)
         _ensure_uploads_tables(conn)
+        _ensure_task_checkpoints_table(conn)
+        _ensure_task_steps_table(conn)
         # worker 每轮 poll 都按 status 取任务（见 load_tasks 的
         # WHERE status IN (...)），无索引时是全表扫描；任务表随
         # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。
@@ -2099,6 +2150,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._create_followup(user_id, rest[: -len("/followups")])
             elif rest.endswith("/followups") and method == "GET":
                 self._list_followups(user_id, rest[: -len("/followups")])
+            elif rest.endswith("/timeline") and method == "GET":
+                self._task_timeline(user_id, rest[: -len("/timeline")])
             elif rest.endswith("/chat-history") and method == "GET":
                 self._task_chat_history(
                     user_id, rest[: -len("/chat-history")])
@@ -4602,6 +4655,49 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         self._send_json(200, {
             "followups": [_followup_to_dict(r) for r in rows]})
 
+    def _task_timeline(self, user_id, task_id):
+        """GET /tasks/{id}/timeline（P0 三件其三）：一个任务一张
+        时间线。steps 直读 task_steps（claim/resume/round/final），
+        checkpoints 直读 task_checkpoints（每轮断点快照）；旧
+        任务（部署前跑完、无步骤记录）返回空数组，不编造。
+        """
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        conn = self._db()
+        try:
+            _ensure_task_checkpoints_table(conn)
+            _ensure_task_steps_table(conn)
+            step_rows = conn.execute(
+                "SELECT step_no, kind, summary, state_sha, created_at"
+                " FROM task_steps WHERE task_id = ?"
+                " ORDER BY step_no, created_at, id",
+                (task_id,)).fetchall()
+            cp_rows = conn.execute(
+                "SELECT round_no, thread_id, goal_status, state_sha,"
+                " verify_out, created_at FROM task_checkpoints"
+                " WHERE task_id = ? ORDER BY round_no",
+                (task_id,)).fetchall()
+        finally:
+            conn.close()
+        self._send_json(200, {
+            "task_id": task_id,
+            "steps": [{"step_no": r[0], "kind": r[1], "summary": r[2],
+                       "state_sha": r[3], "created_at": r[4]}
+                      for r in step_rows],
+            "checkpoints": [{"round_no": r[0], "thread_id": r[1],
+                             "goal_status": r[2], "state_sha": r[3],
+                             "verify_out": r[4], "created_at": r[5]}
+                            for r in cp_rows],
+        })
+
     def _task_chat_history(self, user_id, task_id):
         """GET /tasks/{id}/chat-history：真正的交替历史，一次性
         按时间交错拼好返回，前端不再猜顺序。
@@ -6541,6 +6637,21 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 " updated_at = ? WHERE id = ?",
                 (now, task_id),
             )
+            # P0 三件其三：时间线 pause 步骤（best-effort）。
+            try:
+                _ensure_task_steps_table(conn)
+                _max = conn.execute(
+                    "SELECT MAX(step_no) FROM task_steps"
+                    " WHERE task_id = ?", (task_id,)).fetchone()
+                conn.execute(
+                    "INSERT INTO task_steps(id, task_id, step_no, kind,"
+                    " summary, state_sha, created_at)"
+                    " VALUES(?, ?, ?, 'pause', '用户暂停', NULL, ?)",
+                    (uuid.uuid4().hex, task_id,
+                     ((_max[0] or 0) + 1 if _max and _max[0] else 1),
+                     now))
+            except Exception:
+                pass
             conn.commit()
             task = _task_to_dict(
                 conn.execute(

cloud/tests/test_p0_three.py

@@ -0,0 +1,328 @@
+"""P0 三件单测：断点进库、错误分类+熔断、Run/Step 时间线。
+
+只测本路改动，不跑真实模型、不联网。
+- 分类：KeyError/配置缺失等永久错误判 failed 死信；网络抖动
+  等瞬时错误保持 interrupted；未知错误不分类（走原有逻辑）。
+- 熔断：退避期内不重排队；库断点轮次连续 3 次无进展即停，
+  标 auto_resume_stalled；有进展时 stall 清零。
+- 断点：worker 镜像把磁盘 host-state 断点写进 task_checkpoints；
+  条目丢失后可按库断点重建最小条目（不造假：库里无断点时
+  返回 False）。
+- 时间线：timeline 接口按任务返回 steps 与 checkpoints。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.sandbox import (LocalExecutor, _classify_goal_error)
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.sandbox import (LocalExecutor, _classify_goal_error)
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+
+
+def _insert_task(db_path, task_id, status="interrupted", **cols):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
+            (task_id, "p1", "do work", status, "tester", now, now),
+        )
+        for key, value in cols.items():
+            conn.execute(
+                "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                (value, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _task_row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        return conn.execute(
+            "SELECT status, goal_resume_pending, auto_resume_count,"
+            " auto_resume_stall_count, checkpoint_round_at_last_resume,"
+            " resume_after, interrupted_reason FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+
+
+class ClassifyTestCase(unittest.TestCase):
+    def test_permanent_keyerror(self):
+        kind, snippet = _classify_goal_error(
+            "Traceback (most recent call last):\n"
+            "KeyError: '3829f5f3efb049dfafaecbd15232064c'\n", "")
+        self.assertEqual(kind, "permanent")
+        self.assertIn("KeyError", snippet)
+
+    def test_permanent_missing_env(self):
+        kind, _ = _classify_goal_error(
+            "", "Missing environment variable: SUB2API_API_KEY")
+        self.assertEqual(kind, "permanent")
+
+    def test_transient_gateway(self):
+        kind, _ = _classify_goal_error("", "HTTP 503 Service Unavailable")
+        self.assertEqual(kind, "transient")
+
+    def test_unknown_not_classified(self):
+        kind, snippet = _classify_goal_error("some odd output", "")
+        self.assertIsNone(kind)
+        self.assertIsNone(snippet)
+
+    def test_keyerror_no_conclusion_is_failed(self):
+        fake = SimpleNamespace(
+            returncode=1,
+            stdout="Traceback (most recent call last):\n"
+                   "KeyError: 't1'\n",
+            stderr="")
+        result = LocalExecutor("/fake/host.py")._goal_result_from_runner(
+            "t1", fake)
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("KeyError", result["goal_error"])
+
+    def test_transient_no_conclusion_stays_interrupted(self):
+        fake = SimpleNamespace(
+            returncode=1,
+            stdout="",
+            stderr="RemoteDisconnected: connection closed")
+        result = LocalExecutor("/fake/host.py")._goal_result_from_runner(
+            "t1", fake)
+        self.assertEqual(result["status"], "interrupted")
+
+
+class StallGuardTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+        self.worker = Worker(
+            self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"},
+            task_timeout=1800)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _checkpoint(self, task_id, rounds):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO task_checkpoints(task_id, round_no,"
+                " thread_id, goal_status, state_sha, verify_out,"
+                " created_at) VALUES(?, ?, 'th', 'goal-running',"
+                " 'sha', '', ?)", (task_id, rounds, time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def test_backoff_blocks_early_resume(self):
+        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
+                     interrupted_reason="sweep", paused_by_user=0,
+                     resume_after=time.time() + 600)
+        self.assertEqual(self.worker.auto_resume_interrupted(), [])
+        self.assertEqual(_task_row(self.db_path, "T1")[0], "interrupted")
+
+    def test_three_no_progress_stalls(self):
+        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
+                     interrupted_reason="sweep", paused_by_user=0,
+                     checkpoint_round_at_last_resume=5,
+                     auto_resume_stall_count=2)
+        self._checkpoint("T1", 5)
+        self.assertEqual(self.worker.auto_resume_interrupted(), [])
+        row = _task_row(self.db_path, "T1")
+        self.assertEqual(row[0], "interrupted")
+        self.assertEqual(row[6], "auto_resume_stalled")
+
+    def test_progress_resets_stall(self):
+        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
+                     interrupted_reason="sweep", paused_by_user=0,
+                     checkpoint_round_at_last_resume=5,
+                     auto_resume_stall_count=2)
+        self._checkpoint("T1", 6)
+        self.assertEqual(self.worker.auto_resume_interrupted(), ["T1"])
+        row = _task_row(self.db_path, "T1")
+        self.assertEqual(row[0], "queued")
+        self.assertEqual(row[3], 0)
+        self.assertEqual(row[4], 6)
+        self.assertGreater(row[5], time.time())
+
+
+class CheckpointMirrorTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "test.db")
+        init_db(self.db_path)
+        self.state_dir = os.path.join(self.tmp.name, "work")
+        os.makedirs(self.state_dir)
+        cfg_path = os.path.join(self.tmp.name, "host-config.json")
+        with open(cfg_path, "w", encoding="utf-8") as fh:
+            json.dump({"state": "work/host-state.json"}, fh)
+        self.state_path = os.path.join(self.state_dir, "host-state.json")
+        with open(self.state_path, "w", encoding="utf-8") as fh:
+            json.dump({"version": 1, "tasks": {"T1": {
+                "goal_rounds_done": 3, "thread_id": "th1",
+                "goal_status": "goal-running",
+                "goal_last_verify": "ok"}}}, fh)
+        self.worker = Worker(
+            self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"},
+            task_timeout=1800)
+        self.worker.executor = LocalExecutor(
+            "/fake/host.py", config_path=cfg_path)
+        _insert_task(self.db_path, "T1", status="running",
+                     goal_mode=1, goal="g", verify_command=None,
+                     goal_max_rounds=0)
+        self.task = {"id": "T1", "project_id": "p1"}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_mirror_then_rebuild(self):
+        self.assertEqual(self.worker._mirror_checkpoint_once(self.task), 3)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT round_no, thread_id FROM task_checkpoints"
+                " WHERE task_id = 'T1'").fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row, (3, "th1"))
+        # 模拟任务 2 的断点丢失：磁盘条目被删，库里有断点。
+        with open(self.state_path, "w", encoding="utf-8") as fh:
+            json.dump({"version": 1, "tasks": {}}, fh)
+        self.assertTrue(
+            self.worker._rebuild_host_state_from_checkpoint(self.task))
+        with open(self.state_path, encoding="utf-8") as fh:
+            entry = json.load(fh)["tasks"]["T1"]
+        self.assertEqual(entry["goal_rounds_done"], 3)
+        self.assertEqual(entry["thread_id"], "th1")
+        self.assertTrue(entry["rebuilt_from_checkpoint"])
+
+    def test_mirror_imports_host_steps(self):
+        steps_dir = os.path.join(self.state_dir, "host-steps")
+        os.makedirs(steps_dir)
+        with open(os.path.join(steps_dir, "T1.jsonl"),
+                  "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(
+                {"round": 1, "turn_status": "completed",
+                 "verify_out": "", "ts": time.time()}) + "\n")
+            fh.write("not json\n")
+            fh.write(json.dumps(
+                {"round": 2, "turn_status": "completed",
+                 "verify_out": "", "ts": time.time()}) + "\n")
+        self.worker._mirror_checkpoint_once(self.task)
+        self.worker._mirror_checkpoint_once(self.task)  # 去重：不重复插
+        conn = sqlite3.connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT step_no, kind FROM task_steps"
+                " WHERE task_id = 'T1' ORDER BY step_no").fetchall()
+        finally:
+            conn.close()
+        self.assertEqual(rows, [(1, "round"), (2, "round")])
+
+    def test_rebuild_without_checkpoint_is_false(self):
+        with open(self.state_path, "w", encoding="utf-8") as fh:
+            json.dump({"version": 1, "tasks": {}}, fh)
+        self.assertFalse(
+            self.worker._rebuild_host_state_from_checkpoint(self.task))
+
+
+class TimelineApiTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        _insert_task(self.db_path, "T9", status="completed",
+                     goal_mode=1, goal="g")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET project_id = ? WHERE id = 'T9'",
+                (self.project["id"],))
+            conn.execute(
+                "INSERT INTO task_steps(id, task_id, step_no, kind,"
+                " summary, state_sha, created_at)"
+                " VALUES('s1', 'T9', 1, 'claim', 'worker 认领', NULL, ?)",
+                (time.time(),))
+            conn.execute(
+                "INSERT INTO task_checkpoints(task_id, round_no,"
+                " thread_id, goal_status, state_sha, verify_out,"
+                " created_at) VALUES('T9', 2, 'th', 'goal-running',"
+                " 'sha', '', ?)", (time.time(),))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        headers = {"Content-Type": "application/json"}
+        if key:
+            headers["X-API-Key"] = key
+        conn.request(method, path,
+                     json.dumps(body) if body is not None else None,
+                     headers)
+        resp = conn.getresponse()
+        data = resp.read()
+        conn.close()
+        try:
+            return resp.status, json.loads(data)
+        except Exception:
+            return resp.status, data
+
+    def test_timeline_returns_steps_and_checkpoints(self):
+        status, body = self._req(
+            "GET", "/tasks/T9/timeline", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(body["steps"]), 1)
+        self.assertEqual(body["steps"][0]["kind"], "claim")
+        self.assertEqual(len(body["checkpoints"]), 1)
+        self.assertEqual(body["checkpoints"][0]["round_no"], 2)
+
+    def test_timeline_missing_task_404(self):
+        status, _ = self._req(
+            "GET", "/tasks/nope/timeline", key=self.owner_key)
+        self.assertEqual(status, 404)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -119,6 +119,7 @@ Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下，
 """
 
 import argparse
+import hashlib
 import json
 import os
 from datetime import datetime, timezone
@@ -205,6 +206,23 @@ GOAL_MAX_ROUNDS_CAP = 100
 _ALREADY_EXISTS_MARKER = "已存在"
 
 
+def _uuid_hex():
+    """uuid hex（task_steps id 用，两后端都不依赖自增语义）。"""
+    return uuid.uuid4().hex
+
+
+def _worker_host_name(executor, task):
+    """host 任务名：优先 executor._host_name（与 sandbox 同款
+    派生），没有 executor 时回落任务 id（uuid hex 本就是安全名）。"""
+    fn = getattr(executor, "_host_name", None)
+    if callable(fn):
+        try:
+            return fn(task)
+        except Exception:
+            pass
+    return str(task.get("id"))
+
+
 def _default_quota_usd_micros():
     """无配额行时的默认配额（微美元），与 server 侧同值。
 
@@ -868,6 +886,15 @@ class Worker:
         从 goal_rounds_done 下一轮起），同时清旧结果、清
         interrupted_reason、auto_resume_count+1；上限
         self.auto_resume_max 到顶的任务停住并记审计，等人工。
+        P0 三件其二的熔断（与次数上限并存，防确定性失败空转）：
+        - 退避：重排队时写 resume_after = now + min(300,
+          10 * 2**stall)，未到时间不重排队，不再 12 秒一轮烧次数。
+        - 无进展熔断：库中断点轮次（task_checkpoints 最大
+          round_no）与上次重排队时记的基线相同 → stall_count+1；
+          stall 到 3 时停住，interrupted_reason 标
+          auto_resume_stalled 并记审计，等人工（任务 2 的
+          KeyError 空转：分类已把永久错误判 failed；万一分类
+          漏网，无进展熔断是第二道闸）。
         返回自动续跑的任务 id 列表；任何异常只记日志，不炸
         run_loop（恢复是辅助路径，不能带走常驻 worker）。
         """
@@ -878,7 +905,9 @@ class Worker:
                 self._ensure_result_columns(conn)
                 rows = conn.execute(
                     "SELECT id, project_id, created_by, parent_task_id,"
-                    " model, interrupted_reason FROM tasks"
+                    " model, interrupted_reason, resume_after,"
+                    " auto_resume_stall_count,"
+                    " checkpoint_round_at_last_resume FROM tasks"
                     " WHERE status = 'interrupted'"
                     " AND (paused_by_user IS NULL OR paused_by_user = 0)"
                     " AND (interrupted_reason IS NULL OR"
@@ -895,6 +924,8 @@ class Worker:
                 for row in rows:
                     task_id = row[0]
                     project_id = row[1]
+                    if row[6] and now < row[6]:
+                        continue
                     if row[5] == "quota":
                         # 配额中断：现查配额，仍超限或查不到
                         # 就留在 interrupted，下轮再查，不重排队。
@@ -908,6 +939,31 @@ class Worker:
                             _qerr = "quota_check_failed"
                         if _qerr is not None:
                             continue
+                    try:
+                        cp = conn.execute(
+                            "SELECT MAX(round_no) FROM task_checkpoints"
+                            " WHERE task_id = ?", (task_id,)).fetchone()
+                        cp_round = cp[0] if cp and cp[0] is not None else -1
+                    except Exception:
+                        cp_round = -1
+                    prev = row[8]
+                    stall = 0
+                    if prev is not None:
+                        stall = (row[7] or 0) + 1 if cp_round == prev else 0
+                    if stall >= 3:
+                        cur = conn.execute(
+                            "UPDATE tasks SET interrupted_reason ="
+                            " 'auto_resume_stalled',"
+                            " auto_resume_stall_count = ?,"
+                            " updated_at = ?"
+                            " WHERE id = ? AND status = 'interrupted'",
+                            (stall, now, task_id))
+                        if cur.rowcount == 1:
+                            self._audit(
+                                project_id, "worker_auto_resume_stalled",
+                                task_id)
+                        continue
+                    backoff = min(300, 10 * (2 ** stall))
                     cur = conn.execute(
                         "UPDATE tasks SET status = 'queued',"
                         " goal_resume_pending = 1,"
@@ -916,16 +972,23 @@ class Worker:
                         " interrupted_reason = NULL,"
                         " auto_resume_count ="
                         " COALESCE(auto_resume_count, 0) + 1,"
+                        " auto_resume_stall_count = ?,"
+                        " checkpoint_round_at_last_resume = ?,"
+                        " resume_after = ?,"
                         " updated_at = ?"
                         " WHERE id = ? AND status = 'interrupted'"
                         " AND (paused_by_user IS NULL"
                         " OR paused_by_user = 0)",
-                        (now, task_id),
+                        (stall, cp_round, now + backoff, now, task_id),
                     )
                     if cur.rowcount == 1:
                         resumed.append(task_id)
                         self._audit(project_id, "worker_auto_resumed",
                                     task_id)
+                        self._record_step(
+                            conn, task_id, "auto_resume",
+                            "自动续跑（无进展计数 %d，下次退避 %d 秒）"
+                            % (stall, backoff))
                 conn.commit()
             finally:
                 conn.close()
@@ -934,6 +997,249 @@ class Worker:
                       message="auto resume failed: %s" % exc)
         return resumed
 
+    def _record_step(self, conn, task_id, kind, summary, step_no=None,
+                     state_sha=None):
+        """向 task_steps 写一行（P0 三件其三，时间线事实源）。
+
+        best-effort：写失败只记日志，绝不影响调用方（认领/
+        恢复/终局）的正常流程。step_no 未给时按该任务已有
+        最大 step_no + 1 自增（kind=round 的调用方显式传
+        round_no，保证 host 轮次与时间线 step_no 一致）。
+        """
+        try:
+            if step_no is None:
+                row = conn.execute(
+                    "SELECT MAX(step_no) FROM task_steps"
+                    " WHERE task_id = ?", (task_id,)).fetchone()
+                step_no = (row[0] or 0) + 1 if row and row[0] else 1
+            conn.execute(
+                "INSERT INTO task_steps(id, task_id, step_no, kind,"
+                " summary, state_sha, created_at) VALUES(?, ?, ?, ?,"
+                " ?, ?, ?)",
+                (_uuid_hex(), task_id, step_no, kind, summary[:500]
+                 if summary else None, state_sha, time.time()))
+        except Exception as exc:
+            log_event(self.logger, "task_step_record_failed",
+                      message="step record failed: %s" % exc)
+
+    def _checkpoint_round(self, conn, task_id):
+        """库中断点最大轮次（无断点时 -1）；表缺失时也返回 -1。"""
+        try:
+            row = conn.execute(
+                "SELECT MAX(round_no) FROM task_checkpoints"
+                " WHERE task_id = ?", (task_id,)).fetchone()
+            return row[0] if row and row[0] is not None else -1
+        except Exception:
+            return -1
+
+    def _checkpoint_mirror_loop(self, task, stop):
+        """断点镜像线程（P0 三件其一）：goal 任务执行期间每 30 秒
+        把磁盘 host-state 断点镜像进库一次；执行结束由
+        _run_attempt 的 finally 再补一次终局镜像。线程本身绝不
+        抛（_mirror_checkpoint_once 已全量兜底），stop 后即退。
+        """
+        try:
+            self._mirror_checkpoint_once(task)
+        except Exception:
+            pass
+        while not stop.wait(30):
+            try:
+                self._mirror_checkpoint_once(task)
+            except Exception:
+                pass
+
+    def _host_state_path_for_task(self, task):
+        """断点镜像的 host-state 文件路径（P0 三件其一）。
+
+        租户 Docker：tenants/<safe>/work/host-state.json；
+        非租户 Docker：executor extra_mounts 里 dest=/app/work
+        的宿主路径（若无则回落 /opt/codex-app-server/work）；
+        Local：读 executor config 的 state 项（相对配置目录）。
+        找不到时返回 None，镜像静默跳过（不影响任务执行）。
+        """
+        try:
+            project_id = task.get("project_id")
+            if getattr(self, "tenant_isolation", False) and project_id:
+                safe = self._tenant_safe_name(project_id)
+                base = os.path.join(self.tenants_root, safe, "work")
+                return os.path.join(base, "host-state.json")
+            mounts = getattr(self.executor, "extra_mounts", None) or []
+            for item in mounts:
+                src, dst = (item if isinstance(item, tuple)
+                            else (None, None))
+                if dst == "/app/work" and src:
+                    return os.path.join(src, "host-state.json")
+            cfg = getattr(self.executor, "config_path", None)
+            if cfg and os.path.exists(cfg):
+                with open(cfg, encoding="utf-8") as fh:
+                    data = json.load(fh)
+                state = data.get("state") or "host-state.json"
+                if not os.path.isabs(state):
+                    state = os.path.join(
+                        os.path.dirname(os.path.abspath(cfg)), state)
+                return state
+            return os.path.join("/opt/codex-app-server/work",
+                                "host-state.json")
+        except Exception:
+            return None
+
+    def _mirror_checkpoint_once(self, task):
+        """把磁盘 host-state 里的本任务断点镜像进库（单次）。
+
+        读 state → 取 tasks[host 名] 条目 → (task_id, round_no)
+        upsert 进 task_checkpoints（state_sha 是条目 JSON 的
+        sha256，轮次相同但线程/状态变了也能看出来）→ 同时
+        导入同目录 host-steps/<任务>.jsonl 里的 round 步骤。
+        任何异常静默（镜像是辅助路径，执行本身不受影响）。
+        返回镜像到的轮次（-1 = 没镜像到）。
+        """
+        try:
+            state_path = self._host_state_path_for_task(task)
+            if not state_path or not os.path.exists(state_path):
+                return -1
+            with open(state_path, encoding="utf-8") as fh:
+                state = json.load(fh)
+            host_name = _worker_host_name(self.executor, task)
+            entry = (state.get("tasks") or {}).get(host_name)
+            if not isinstance(entry, dict):
+                return -1
+            rounds = int(entry.get("goal_rounds_done") or 0)
+            state_sha = hashlib.sha256(
+                json.dumps(entry, sort_keys=True,
+                           default=str).encode("utf-8")).hexdigest()
+            conn = _db_connect(self.db_path)
+            try:
+                self._ensure_result_columns(conn)
+                existing = conn.execute(
+                    "SELECT 1 FROM task_checkpoints"
+                    " WHERE task_id = ? AND round_no = ?",
+                    (task["id"], rounds)).fetchone()
+                if existing:
+                    conn.execute(
+                        "UPDATE task_checkpoints SET thread_id = ?,"
+                        " goal_status = ?, state_sha = ?,"
+                        " verify_out = ?, created_at = ?"
+                        " WHERE task_id = ? AND round_no = ?",
+                        (entry.get("thread_id"),
+                         entry.get("goal_status"),
+                         state_sha,
+                         (entry.get("goal_last_verify") or "")[:500],
+                         time.time(), task["id"], rounds))
+                else:
+                    conn.execute(
+                        "INSERT INTO task_checkpoints(task_id, round_no,"
+                        " thread_id, goal_status, state_sha, verify_out,"
+                        " created_at) VALUES(?, ?, ?, ?, ?, ?, ?)",
+                        (task["id"], rounds, entry.get("thread_id"),
+                         entry.get("goal_status"), state_sha,
+                         (entry.get("goal_last_verify") or "")[:500],
+                         time.time()))
+                self._import_host_steps(conn, task, state_path)
+                conn.commit()
+            finally:
+                conn.close()
+            return rounds
+        except Exception:
+            return -1
+
+    def _import_host_steps(self, conn, task, state_path):
+        """把 host-steps/<任务>.jsonl 的 round 行导入 task_steps。
+
+        按 (task_id, kind='round', step_no) 去重：已入库的
+        step_no 跳过；文件不存在或格式坏行都跳过不炸。
+        """
+        try:
+            steps_path = os.path.join(
+                os.path.dirname(os.path.abspath(state_path)),
+                "host-steps", "%s.jsonl" % _worker_host_name(
+                    self.executor, task))
+            if not os.path.exists(steps_path):
+                return
+            existing = {row[0] for row in conn.execute(
+                "SELECT step_no FROM task_steps WHERE task_id = ?"
+                " AND kind = 'round'", (task["id"],)).fetchall()}
+            with open(steps_path, encoding="utf-8") as fh:
+                for line in fh:
+                    try:
+                        item = json.loads(line)
+                    except Exception:
+                        continue
+                    step_no = item.get("round")
+                    if not isinstance(step_no, int) or step_no in existing:
+                        continue
+                    conn.execute(
+                        "INSERT INTO task_steps(id, task_id, step_no,"
+                        " kind, summary, state_sha, created_at)"
+                        " VALUES(?, ?, ?, 'round', ?, NULL, ?)",
+                        (_uuid_hex(), task["id"], step_no,
+                         ("第 %d 轮 turn=%s verify=%s" % (
+                             step_no, item.get("turn_status"),
+                             (item.get("verify_out") or "")[:200])),
+                         item.get("ts") or time.time()))
+                    existing.add(step_no)
+        except Exception:
+            pass
+
+    def _rebuild_host_state_from_checkpoint(self, task):
+        """断点丢失时的重建（P0 三件其一，任务 2 类故障的解法）。
+
+        host-state 里没有本任务条目、但库里有断点时：从库里
+        最新断点 + tasks 行的 goal/verify_command/max_rounds
+        拼一个最小 host-state 条目写回（goal_status=interrupted，
+        轮次与 thread_id 以库为准），再由正常 goal-resume
+        路径续跑。拼不出来（库里也无断点）返回 False，
+        调用方走原有拒绝/失败路径，不造假断点。
+        """
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                self._ensure_result_columns(conn)
+                row = conn.execute(
+                    "SELECT round_no, thread_id, goal_status"
+                    " FROM task_checkpoints WHERE task_id = ?"
+                    " ORDER BY round_no DESC LIMIT 1",
+                    (task["id"],)).fetchone()
+                if not row:
+                    return False
+                trow = conn.execute(
+                    "SELECT goal, verify_command, goal_max_rounds"
+                    " FROM tasks WHERE id = ?", (task["id"],)).fetchone()
+                if not trow:
+                    return False
+            finally:
+                conn.close()
+            state_path = self._host_state_path_for_task(task)
+            if not state_path:
+                return False
+            host_name = _worker_host_name(self.executor, task)
+            if os.path.exists(state_path):
+                with open(state_path, encoding="utf-8") as fh:
+                    state = json.load(fh)
+            else:
+                state = {"version": 1, "tasks": {}}
+            tasks_map = state.setdefault("tasks", {})
+            if host_name in tasks_map:
+                return True
+            tasks_map[host_name] = {
+                "goal": trow[0],
+                "verify_command": trow[1],
+                "goal_max_rounds": trow[2] or 0,
+                "goal_rounds_done": row[0],
+                "thread_id": row[1],
+                "goal_status": "interrupted",
+                "goal_last_verify": "",
+                "rebuilt_from_checkpoint": True,
+            }
+            os.makedirs(os.path.dirname(
+                os.path.abspath(state_path)), exist_ok=True)
+            tmp_path = state_path + ".tmp"
+            with open(tmp_path, "w", encoding="utf-8") as fh:
+                json.dump(state, fh)
+            os.replace(tmp_path, state_path)
+            return True
+        except Exception:
+            return False
+
     def _ensure_result_columns(self, conn):
         """worker 侧的迁移兜底：与 server.migrate_tasks_table 同列。
 
@@ -976,7 +1282,11 @@ class Worker:
                 "paused_by_user": "INTEGER DEFAULT 0",
                 "interrupted_reason": "TEXT",
                 "auto_resume_count": "INTEGER DEFAULT 0",
-                "no_time_limit": "INTEGER DEFAULT 0"}.items():
+                "no_time_limit": "INTEGER DEFAULT 0",
+                "resume_after": "DOUBLE PRECISION DEFAULT 0",
+                "auto_resume_stall_count": "INTEGER DEFAULT 0",
+                "checkpoint_round_at_last_resume": "INTEGER",
+                "last_error_signature": "TEXT"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -984,6 +1294,32 @@ class Worker:
                         % (name, coltype))
                 except Exception:
                     pass
+        # P0 三件的表（worker 直连老库时也补，与 server ensure 同 DDL）。
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_checkpoints("
+                "task_id TEXT, "
+                "round_no INTEGER, "
+                "thread_id TEXT, "
+                "goal_status TEXT, "
+                "state_sha TEXT, "
+                "verify_out TEXT, "
+                "created_at DOUBLE PRECISION, "
+                "PRIMARY KEY(task_id, round_no))")
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_steps("
+                "id TEXT PRIMARY KEY, "
+                "task_id TEXT, "
+                "step_no INTEGER, "
+                "kind TEXT, "
+                "summary TEXT, "
+                "state_sha TEXT, "
+                "created_at DOUBLE PRECISION)")
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_task_steps_task"
+                " ON task_steps(task_id, step_no)")
+        except Exception:
+            pass
 
     def _set_status_if_running(self, task_id, status):
         """仅当任务仍为 running 时改状态，返回是否改成功。
@@ -3084,6 +3420,20 @@ class Worker:
         _watchdog_stop = None
         _watchdog_thread = None
         _quota_watchdog_thread = None
+        # P0 三件其一：goal 任务起断点镜像线程（任何执行器）；
+        # 非 goal 任务没有断点语义，不起。终局镜像在 finally 补。
+        _mirror_stop = None
+        _mirror_thread = None
+        if goal_mode:
+            try:
+                _mirror_stop = threading.Event()
+                _mirror_thread = threading.Thread(
+                    target=self._checkpoint_mirror_loop,
+                    args=(task, _mirror_stop), daemon=True)
+                _mirror_thread.start()
+            except Exception:
+                _mirror_stop = None
+                _mirror_thread = None
         if is_docker and hasattr(self.executor, "kill_container"):
             try:
                 _container_name = self.executor.container_name(task)
@@ -3306,6 +3656,16 @@ class Worker:
                     _watchdog_thread.join(timeout=2)
                 if _quota_watchdog_thread is not None:
                     _quota_watchdog_thread.join(timeout=2)
+            if _mirror_stop is not None:
+                _mirror_stop.set()
+                if _mirror_thread is not None:
+                    _mirror_thread.join(timeout=2)
+                # 终局镜像：执行结束时的断点（最后一轮）必须落库，
+                # 否则中断后库断点比实际少一轮；失败不影响结果写回。
+                try:
+                    self._mirror_checkpoint_once(task)
+                except Exception:
+                    pass
             # 同步路径执行结束：清掉取消用的容器标记（取消路径里
             # 若已清过，这里再清一次无害）。进程模式的 proc 由
             # _execute_via_process 自己的 finally 清，这里只清
@@ -3770,6 +4130,18 @@ class Worker:
                                            error=error)
         else:
             self._audit(project_id, "worker_interrupted", task_id)
+        # P0 三件其三：时间线终局步骤（best-effort）。
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                self._ensure_result_columns(conn)
+                self._record_step(conn, task_id, "final",
+                                  "终态 %s" % final)
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception:
+            pass
         host_name = self.executor._host_name(task) \
             if hasattr(self.executor, "_host_name") else task_id
         return {"dispatched": True, "task_id": task_id,
@@ -3834,6 +4206,18 @@ class Worker:
         if not self._claim(task_id, executor=self._executor_name()):
             return {"dispatched": False, "reason": "already_claimed",
                     "task_id": task_id}
+        # P0 三件其三：时间线 claim 步骤（best-effort，写失败
+        # 不影响执行；step_no 由 _record_step 自增）。
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                self._ensure_result_columns(conn)
+                self._record_step(conn, task_id, "claim", "worker 认领")
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception:
+            pass
         task = self.get_task(task_id) or task
         # G10：goal 断点续跑标记——认领后立即清掉库里标记（本轮
         # 已认领为 running，不会重复触发；本轮若再被杀回
@@ -3845,6 +4229,14 @@ class Worker:
         if task.get("goal_resume_pending"):
             task = dict(task)
             task["_goal_resume_breakpoint"] = True
+            # P0 三件其一：断点丢失重建——host-state 里条目已丢
+            # 但库里有断点时，先按库断点拼最小条目写回，再由
+            # 正常 goal-resume 续跑；库里也没有断点时返回 False，
+            # 正常路径继续（host 拒绝 → failed，不造假断点）。
+            try:
+                self._rebuild_host_state_from_checkpoint(task)
+            except Exception:
+                pass
             try:
                 conn = _db_connect(self.db_path)
                 try:
@@ -4460,6 +4852,18 @@ class Worker:
                        else "worker_failed"),
                       task_id=task_id, project_id=task["project_id"],
                       status=final)
+            # P0 三件其三：时间线终局步骤（best-effort）。
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    self._ensure_result_columns(conn)
+                    self._record_step(conn, task_id, "final",
+                                      "终态 %s" % final)
+                    conn.commit()
+                finally:
+                    conn.close()
+            except Exception:
+                pass
             if result.get("goal_met") is False:
                 # goal 未达只发 goal-unmet，不再重复发 failed：
                 # 同一结局两条通知会刷屏，digest 偏好下两者又都是

host.py

@@ -2060,6 +2060,25 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify,
         print(f"round {round_no}: turn={result['status']} "
               f"verify_rc={v_rc if v_rc is not None else '-'} "
               f"verify_out={last_verify[:300]}")
+        # P0 三件其三：每轮步骤落盘（与 host-state 同目录的
+        # host-steps/<任务>.jsonl，worker 镜像/终局时导入
+        # task_steps，供 timeline 接口读；写失败只记不炸循环）。
+        try:
+            _steps_dir = os.path.join(
+                os.path.dirname(os.path.abspath(state_path)),
+                "host-steps")
+            os.makedirs(_steps_dir, exist_ok=True)
+            with open(os.path.join(
+                    _steps_dir, "%s.jsonl" % task["name"]),
+                    "a", encoding="utf-8") as _fh:
+                _fh.write(json.dumps(
+                    {"round": round_no,
+                     "turn_status": result["status"],
+                     "verify_out": last_verify[:200],
+                     "ts": time.time()},
+                    ensure_ascii=False) + "\n")
+        except Exception:
+            pass
         if verify is None:
             # 模型自决：只有完整跑完的回合里的标记行才算数。
             if (result["status"] == "completed"