Merge pull request #19 from openqj/host-error-surfacing

Surface failed-turn reason from thread_history in dispatch output

cloud/worker.py

@@ -905,6 +905,14 @@ class Worker:
             parsed = None
         if isinstance(parsed, dict) and parsed.get("text"):
             return str(parsed["text"])
+        # dispatch 输出现在带 error 字段（host.py 从 thread_history
+        # 库里捞回的失败原因，如 404）；text 为空时优先用它，
+        # 否则两次失败的拼接里只剩 host_status=404 这种代号。
+        if isinstance(parsed, dict) and parsed.get("error"):
+            err = parsed["error"]
+            if isinstance(err, dict):
+                err = err.get("message") or str(err)
+            return str(err)[:1000]
         # host.py 崩溃时 stdout 常为空、真正的报错在 stderr（例如
         # “缺少 --repo”、Read-only file system）；只用 stdout 时
         # 最终错误会被写成通用的 host execution failed，排障无从

host.py

@@ -316,13 +316,59 @@ class Server:
                             item.get("phase") == "final_answer" and \
                             item.get("text"):
                         text = item["text"]
-            return {"status": turn.get("status", "completed"), "text": text,
+            status = turn.get("status", "completed")
+            error = turn.get("error")
+            if status != "completed" and not error:
+                # app-server 的 turn/completed 事件常常只给
+                # status=failed，不给失败原因（采样失败的 404
+                # 只写在它本地的 thread_history 库里，事件里没有）。
+                # 此时 best-effort 从 codex_home 的 thread_history
+                # 库里按 thread_id + turn_id 查 error_json，把
+                # message 捞回来给调用方（worker 靠它判断是否可
+                # fallback）；查不到、库不存在、表结构变了都只当
+                # 没有原因，绝不因此把回合本身搞失败。
+                error = self._turn_error_from_history(
+                    thread_id, turn.get("id") or turn_id)
+            return {"status": status, "text": text,
                     "turn_id": turn.get("id") or turn_id,
                     "root_turn_id": turn.get("rootTurnId"),
-                    "error": turn.get("error")}
+                    "error": error}
         finally:
             self.turn_queues.pop(thread_id, None)
 
+    def _turn_error_from_history(self, thread_id, turn_id):
+        """从 codex thread_history 库里取失败回合的 error message。
+
+        只读、只在回合失败且事件本身没给原因时调用；任何异常
+        都返回 None。库路径：codex_home/thread_history_1.sqlite，
+        codex_home 未指定时用 ~/.codex。
+        """
+        if not thread_id or not turn_id:
+            return None
+        try:
+            import sqlite3
+            home = self.codex_home or os.path.expanduser("~/.codex")
+            db_path = os.path.join(home, "thread_history_1.sqlite")
+            if not os.path.exists(db_path):
+                return None
+            conn = sqlite3.connect("file:%s?mode=ro" % db_path, uri=True)
+            try:
+                row = conn.execute(
+                    "SELECT error_json FROM thread_turns"
+                    " WHERE thread_id = ? AND turn_id = ?",
+                    (thread_id, turn_id)).fetchone()
+            finally:
+                conn.close()
+            if not row or not row[0]:
+                return None
+            parsed = json.loads(row[0])
+            msg = parsed.get("message") if isinstance(parsed, dict) else None
+            if not msg:
+                return None
+            return str(msg)[:1000]
+        except Exception:
+            return None
+
     def close(self):
         try:
             self.proc.stdin.close()
@@ -1623,7 +1669,8 @@ def main():
             save_state(args.state, state)
             print(json.dumps({"task": args.name, "thread_id": task["thread_id"],
                               "cwd": task["cwd"], "status": task["status"],
-                              "text": result.get("text", "")},
+                              "text": result.get("text", ""),
+                              "error": result.get("error")},
                              indent=2, ensure_ascii=False))
         elif args.cmd == "list":
             for name, t in state["tasks"].items():

tests/test_host.py

@@ -248,6 +248,53 @@ class RetryAndBudgetTest(unittest.TestCase):
             self.assertIn("t1 failed", Path(notify_log).read_text())
 
 
+class TurnErrorFromHistoryTest(unittest.TestCase):
+    def _server_with_history(self, tmp, error_json):
+        import sqlite3
+        db = os.path.join(tmp, "thread_history_1.sqlite")
+        conn = sqlite3.connect(db)
+        conn.execute(
+            "CREATE TABLE thread_turns(thread_id TEXT, turn_id TEXT,"
+            " error_json TEXT)")
+        if error_json is not None:
+            conn.execute(
+                "INSERT INTO thread_turns VALUES('th1', 'tu1', ?)",
+                (error_json,))
+        conn.commit()
+        conn.close()
+        srv = object.__new__(host.Server)
+        srv.codex_home = tmp
+        return srv
+
+    def test_failed_turn_recovers_404_from_history(self):
+        # 事件里 status=failed 但 error 为空时，从 thread_history
+        # 库里按 thread_id + turn_id 捞回 404 原因（主机实测回归：
+        # 坏模型回合的 404 只在库里，事件里没有，导致 worker
+        # 永远看不到可 fallback 信号）。
+        import json as _json
+        with tempfile.TemporaryDirectory() as tmp:
+            err = _json.dumps({
+                "message": "unexpected status 404 Not Found: Model "
+                           "\"gpt-9.9-nonexistent\" is not available",
+                "codexErrorInfo": {"httpConnectionFailed":
+                                   {"httpStatusCode": 404}}})
+            srv = self._server_with_history(tmp, err)
+            got = srv._turn_error_from_history("th1", "tu1")
+            self.assertIn("404", got)
+            self.assertIn("gpt-9.9-nonexistent", got)
+
+    def test_missing_history_returns_none_without_crashing(self):
+        # 库不存在、行不存在、JSON 损坏都只返回 None，不抛错。
+        with tempfile.TemporaryDirectory() as tmp:
+            srv = self._server_with_history(tmp, None)
+            self.assertIsNone(srv._turn_error_from_history("th1", "tu1"))
+            self.assertIsNone(srv._turn_error_from_history("th1", None))
+            self.assertIsNone(srv._turn_error_from_history(None, "tu1"))
+        with tempfile.TemporaryDirectory() as tmp:
+            srv = self._server_with_history(tmp, "not-json{")
+            self.assertIsNone(srv._turn_error_from_history("th1", "tu1"))
+
+
 class StateVersionTest(unittest.TestCase):
     def test_version_increments_and_conflict_logged(self):
         with tempfile.TemporaryDirectory() as tmp: