Surface failed-turn reason: host.py recovers it from thread_history, dispatch prints it

Real bad-model turn: relay returns 404, codex retries sampling
5 times, turn fails, but app-server's turn/completed event
carries status=failed with NO error and NO text, so worker
never saw any fallbackable signal. The 404 actually lives in
codex's thread_history_1.sqlite (thread_turns.error_json).
host.py Server.run_turn now best-effort recovers that message
(read-only query by thread_id+turn_id, any failure -> None,
never breaks the turn itself), and dispatch CLI prints the
error field (run CLI already printed the full result). Worker
_attempt_error prefers the error field when text is empty, so
a double failure keeps the real reasons, not just status codes.
Tests: host TurnErrorFromHistoryTest (recovery + missing/corrupt
DB -> None), full host + cloud suites pass.

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