From 585242fc8f4a41e084999d826307d99bbd59f804 Mon Sep 17 00:00:00 2001 From: host Date: Thu, 29 Oct 2026 15:13:01 +0800 Subject: [PATCH] =?UTF-8?q?=E9=95=BF=E4=BB=BB=E5=8A=A1=20P0=20=E4=B8=89?= =?UTF-8?q?=E4=BB=B6=EF=BC=9A=E6=96=AD=E7=82=B9=E8=BF=9B=E5=BA=93=E3=80=81?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E5=88=86=E7=B1=BB+=E7=86=94=E6=96=AD?= =?UTF-8?q?=E3=80=81Run/Step=20=E6=97=B6=E9=97=B4=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 断点进库：task_checkpoints 表（PG 新库 + 老库幂等补表 + 迁移清单 对齐）；worker 在 goal 任务执行期间每 30 秒把磁盘 host-state 断点 镜像进库，终局再补一次；断点续跑认领前，若 host-state 条目丢失 但库里有断点，先按库断点重建最小条目再跑，库里也没有断点时不 造假、走原有拒绝路径。 错误分类+熔断：sandbox 四处解析加永久错误分支（KeyError、配置 缺失等无结论异常退出判 failed 死信，不再判 interrupted；host 已 明言 interrupted 的不受影响，未知错误不分类）；worker 自动恢复 加退避（10 秒起指数退避、封顶 300 秒）与无进展熔断（库断点轮次 连续 3 次不变即标 auto_resume_stalled 等人工）。 时间线：task_steps 表 + host 每轮写 host-steps 步骤文件 + worker 导入；GET /tasks/{id}/timeline 一次返回 steps 与 checkpoints； 认领、暂停、自动恢复、终局都记步骤。旧任务返回空数组，不编造。 单测：cloud/tests/test_p0_three.py 14 项（分类、熔断、镜像重建、 步骤导入去重、timeline 接口）。 --- cloud/db.py | 26 +++ cloud/migrations.py | 5 + cloud/sandbox.py | 86 ++++++++ cloud/server.py | 111 ++++++++++ cloud/tests/test_p0_three.py | 328 ++++++++++++++++++++++++++++ cloud/worker.py | 410 ++++++++++++++++++++++++++++++++++- host.py | 19 ++ 7 files changed, 982 insertions(+), 3 deletions(-) create mode 100644 cloud/tests/test_p0_three.py diff --git a/cloud/db.py b/cloud/db.py index 51b08a0..277f460 100644 --- a/cloud/db.py +++ b/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, " diff --git a/cloud/migrations.py b/cloud/migrations.py index c7a0cd1..ea1889b 100644 --- a/cloud/migrations.py +++ b/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"}) diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 5a1b0db..ff801ff 100644 --- a/cloud/sandbox.py +++ b/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 diff --git a/cloud/server.py b/cloud/server.py index 1e0f08e..d5a3997 100644 --- a/cloud/server.py +++ b/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( diff --git a/cloud/tests/test_p0_three.py b/cloud/tests/test_p0_three.py new file mode 100644 index 0000000..4dcd061 --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 5be6fb9..d704ac9 100644 --- a/cloud/worker.py +++ b/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//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 偏好下两者又都是 diff --git a/host.py b/host.py index 177050c..0606b12 100644 --- a/host.py +++ b/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"