BigPizzaV3/CodexPlusPlus · error · RuntimeError
error
Error message
error
What it means
In _collect_turn, a notification with method 'error' arrives from the app-server (turn aborted, invalid turn id, internal model error). The handler raises it so the user's prompt fails fast instead of waiting for a turn/completed that will never come.
Solutions
- Inspect the error payload in the notification for the underlying cause (quota, model unavailable, malformed turn).
- Retry the turn if the cause is transient; otherwise surface the message back to the WeChat user.
- Verify the turn was started with a valid session/turn id from the previous turn/start response.
Defensive patterns
Strategy: try-catch
When it happens
Trigger: Thrown at tools/codex-wechat/codex_wechat.py:538 when the library encounters an invalid state.
Common situations: See trigger scenarios.
AI-assisted analysis of BigPizzaV3/CodexPlusPlus@b1ed92e5e4 (2026-09-19).
Data as JSON: /api/errors/c850071b48cf5ee5.
Report an issue: GitHub.
Appendix: source
Thrown at tools/codex-wechat/codex_wechat.py:538
elif method == "item/agentMessage/delta":
if final_item_id and params.get("turnId") == turn_id and params.get("itemId") == final_item_id:
final_chunks.append(str(params.get("delta") or ""))
elif method == "item/completed":
item = params.get("item") or {}
if item.get("type") == "agentMessage" and item.get("phase") == "final_answer":
text = str(item.get("text") or "")
if text:
final_chunks = [text]
elif method == "turn/completed" and params.get("turn", {}).get("id") == turn_id:
turn = params.get("turn") or {}
if turn.get("status") == "failed":
error = str((turn.get("error") or {}).get("message") or "turn failed")
completed = True
if not completed:
raise TimeoutError("app-server turn 超时")
if error:
raise RuntimeError(error)
reply = "".join(final_chunks).strip()
return reply or "Codex 没有返回最终答复。"
class CodexRunner:
def __init__(self, cfg: BridgeConfig):
self.cfg = cfg
self.app_server: AppServerCodexClient | None = None
if cfg.codex.backend == "app-server":
self.app_server = AppServerCodexClient(cfg)
atexit.register(self.app_server.close)
def run(self, user_id: str, prompt: str) -> str:
if self.app_server:
try:
log("调用 codex app-server")
return self.app_server.run(user_id, prompt)View on GitHub (pinned to b1ed92e5e4)