""" core/agent/recovery.py ====================== 🌟 pi 会话层恢复逻辑的 Python 1:1 移植 对照 pi-main 源码： packages/coding-agent/src/core/agent-session.ts _runAgentPrompt (行 1074): prompt → while _handlePostAgentRun(): continue() _handlePostAgentRun (行 1088): 基于「最后一条 assistant 消息」的三路决策 _prepareRetry (行 2811): 计数 → 移除坏消息 → 退避睡眠 _isRetryableError (行 2770): 溢出交给压缩；其余走重试判定 _checkCompaction (行 ~2034): 溢出压缩恢复（只试一次） packages/ai/src/utils/retry.ts isRetryableAssistantError (行 223) + 可重试/不可重试正则模式表 退避: baseDelayMs × 2^(attempt-1)（默认 2s, 4s, 8s；maxRetries 默认 3） packages/ai/src/utils/overflow.ts isContextOverflow (行 134)：三种溢出识别 isRecoverableLength：length 且 output < 原始输出上限 🌟 决策表（与 pi 完全一致，基于最后一条 assistant 消息）： (1) stop_reason=="error"： a. 是上下文溢出 → 删除坏消息 + 压缩 + continue（只试一次） b. 命中可重试模式（429/5xx/超时/断连…且非配额耗尽） 且未超 maxRetries → 删除坏消息 + 退避(2s×2^(n-1)) + continue c. 其他 → 结束 (2) isRecoverableLength（length 且实际输出 < 原始 maxTokens， 说明是被上下文窗口挤断而非输出上限截断）且未尝试过 → 删除该助手消息 + 压缩 + continue（只试一次） (3) 队列里还有 followUp/steering → continue（续跑） (4) 都不满足 → settle """ from __future__ import annotations import os import re import time from typing import Callable, Optional from .compaction import CompactionSettings, _content_text, compact_context from .context import estimate_context_tokens, should_compact from .types import (AgentMessage, AgentConfig, RetryConfig, RunResult) # ========== 压缩诊断日志（与 main_window.diag_log 同一规范： # [HH:MM:SS.mmm] 时间戳 + 项目根目录追加 + UTF-8 + 失败静默） ========== _COMPACT_LOG_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "compaction_diag.log") def compact_diag_log(line: str): """压缩日志只记录：同一行内容 print（控制台）+ 追加写 compaction_diag.log""" t = time.time() stamp = (f"[{time.strftime('%H:%M:%S', time.localtime(t))}" f".{int(t * 1000) % 1000:03d}]") print(f"{stamp} {line}", flush=True) try: with open(_COMPACT_LOG_PATH, "a", encoding="utf-8") as f: f.write(f"{stamp} {line}\n") except Exception: pass def _settings_of(cfg: AgentConfig) -> CompactionSettings: """AgentConfig 压缩参数 → CompactionSettings（1:1 映射 pi settings 字段）""" return CompactionSettings(reserve_tokens=cfg.compaction_reserve, keep_recent_tokens=cfg.compaction_keep_recent) def _cut_ids_of(msgs: list, new_msgs: list) -> tuple: """🆕 压缩持久化切点：new_msgs = [摘要] + 保留尾巴（旧 msgs 的连续后缀）。 按 identity 定位尾巴起点 → (cut_before_id, first_retained_id)（DB 行 id）。 ⚠️ 行粒度：同一 DB 行（assistant timeline）会被 build_api_context 回放成 多条 API 消息（asst/tool 条目共享同一 _db_msg_id）。若尾巴起点落在某行 回放序列中间，必须回退到该行第一条，否则 cut_before 与 first_retained 会是同一行 → DB 自环（链死循环卡死）。 回退后下一轮会从该行完整重放（保守方向，宁多勿漏）。 任一为空 = 无法在 DB 链上切（如尾巴全是轮中内存消息）→ UI 不插标记。""" if not new_msgs or getattr(new_msgs[0], "kind", None) != "compaction_summary": return "", "" tail = new_msgs[1:] if not tail: return "", "" first = tail[0] for i, m in enumerate(msgs): if m is first: # 回退到同一 DB 行回放序列的头部 fid = getattr(first, "db_msg_id", "") or "" j = i while j > 0 and fid and getattr(msgs[j - 1], "db_msg_id", "") == fid: j -= 1 if j == 0: return "", "" # 没有“之前”的行 → 无切点 return (getattr(msgs[j - 1], "db_msg_id", "") or "", getattr(first, "db_msg_id", "") or "") return "", "" # 尾巴首条不在旧列表（不应发生）→ 不插标记 # ====================================================================== # 对照 retry.ts 的模式表（原样移植） # ====================================================================== _NON_RETRYABLE_LIMIT_RE = re.compile( r"GoUsageLimitError|FreeUsageLimitError|" r"Monthly usage limit reached|available balance|" r"insufficient_quota|out of budget|quota exceeded|billing", re.IGNORECASE, ) _RETRYABLE_RE = re.compile( r"overloaded|rate.?limit|too many requests|429|500|502|503|504|524|" r"service.?unavailable|server.?error|internal.?error|" r"provider.?returned.?error|exceeded request buffer limit while retrying upstream|" r"network.?error|connection.?error|connection.?refused|connection.?lost|" r"other side closed|fetch failed|getaddrinfo|ENOTFOUND|EAI_AGAIN|" r"upstream.?connect|reset before headers|socket hang up|timeout|timed?\s?out", re.IGNORECASE, ) # 对照 overflow.ts OVERFLOW_PATTERNS（各服务商话术） _OVERFLOW_RE = re.compile( r"context\s*window|prompt is too long|maximum context|too many tokens|" r"context length|exceeds the (model|maximum) context|" r"maximum context length is \d+ tokens|exceeds the maximum allowed input length|" r"longer than the model's context length|exceeds the available context size|" r"greater than the context length|exceeded model token limit|" r"Range of input length should be|configured context size|" r"长于|超出.*上下文|上下文.*(超出|超过|不足)", re.IGNORECASE, ) # 限速类不是溢出（对照 overflow.ts NON_OVERFLOW_PATTERNS 的判定意图） _NON_OVERFLOW_RE = re.compile( r"rate.?limit|too many requests|429|overloaded", re.IGNORECASE) # ====================================================================== # 判定函数（对照 pi overflow.ts / retry.ts，基于 assistant 消息） # ====================================================================== def is_context_overflow(message: AgentMessage, context_window: int = 0) -> bool: """ 对照 isContextOverflow 的三种情形： 1. error + 消息文本命中溢出模式（且不是限速类） 2. 静默溢出：stop 但 usage.input > contextWindow（z.ai 风格） 3. 服务端截断式溢出：length 且 output==0 且 input ≥ 0.99×窗口 """ if message.role != "assistant": return False if message.stop_reason == "error" and message.error_message: if not _NON_OVERFLOW_RE.search(message.error_message) \ and _OVERFLOW_RE.search(message.error_message): return True usage = message.usage or {} input_tokens = int(usage.get("input", 0) or 0) + int(usage.get("cacheRead", 0) or 0) if context_window and message.stop_reason == "stop": if input_tokens > context_window: return True if context_window and message.stop_reason == "length" \ and int(usage.get("output", 0) or 0) == 0: if input_tokens >= context_window * 0.99: return True return False def is_retryable_assistant_error(message: AgentMessage) -> bool: """对照 isRetryableAssistantError：基于 errorMessage 文本模式""" if message.role != "assistant" or message.stop_reason != "error": return False if not message.error_message: return False if _NON_RETRYABLE_LIMIT_RE.search(message.error_message): return False return bool(_RETRYABLE_RE.search(message.error_message)) def is_recoverable_length(message: AgentMessage, desired_max_output: int) -> bool: """ 对照 pi isRecoverableLength： length 停止 且 实际输出 < 「原始输出上限」（钳制前的 maxTokens）。 """ if message.role != "assistant" or message.stop_reason != "length": return False if desired_max_output <= 0: return False out = int((message.usage or {}).get("output", 0) or 0) return out < desired_max_output def compute_retry_delay_ms(attempt: int, cfg: RetryConfig) -> float: """对照 pi _prepareRetry: baseDelayMs × 2^(attempt-1)（无抖动、默认无封顶）""" delay = cfg.base_delay_ms * (cfg.factor ** (attempt - 1)) if cfg.max_delay_ms and delay > cfg.max_delay_ms: delay = cfg.max_delay_ms return float(delay) def find_last_assistant(messages) -> Optional[AgentMessage]: """对照 agent-session _findLastAssistantMessage""" for m in reversed(messages): if m.role == "assistant": return m return None # ====================================================================== # 编排器 —— 对照 agent-session 的 run 循环 # ====================================================================== class AgentRunner: """ 对照 agent-session.ts 的 run 编排（agent 之外的「会话层」）。 haocode 里由 AgentWorker(QThread) 持有一个 AgentRunner。 summarize_fn(transcript) -> str：压缩用的非流式 LLM 调用（缺省 → 禁压缩）。 """ def __init__(self, agent: Agent, summarize_fn: Optional[Callable[[str], str]] = None, on_retry_scheduled: Optional[Callable] = None, on_retry_finished: Optional[Callable] = None, on_compaction_started: Optional[Callable] = None, on_compaction_finished: Optional[Callable] = None): self.agent = agent self.summarize_fn = summarize_fn # 🆕 M3: 重试回调（对照 pi retryAssistantCall 的 # onRetryScheduled / onRetryFinished）——UI 可提示「正在重试」 self.on_retry_scheduled = on_retry_scheduled self.on_retry_finished = on_retry_finished # 🆕 压缩开始回调（摘要 LLM 调用阻塞前触发）——UI 显示「执行中」动态气泡 self.on_compaction_started = on_compaction_started # 🆕 压缩完成回调（压缩结束瞬间触发，不等整轮 run 结束） # ——前端气泡即时定格「已完成上下文压缩」（与思考块定格同款时机） self.on_compaction_finished = on_compaction_finished self._overflow_recovered = False # 对照 _overflowRecoveryAttempted self._retry_attempt = 0 # 对照 _retryAttempt self.last_action = "none" # 诊断: none/retry/overflow_compact/length_compact/queued self.compactions_performed = 0 # 成功压缩次数（UI 提示用） # 🆕 压缩事件队列（UI 气泡用：摘要全文/前后 token/耗时/路径） self.compaction_events: list = [] # 🆕 轮中主动压缩护栏（同一 run 内；AgentRunner 每轮新建 → 自动复位） self._mid_turn_fail_streak = 0 self._mid_turn_exhausted = False # ------------------------------------------------------------------ # 发送前压缩（🌟 1:1 对照 pi shouldCompact：tokens > window - reserveTokens） # ------------------------------------------------------------------ def pre_prompt_compaction(self) -> bool: if self.summarize_fn is None: return False cfg = self.agent.config msgs = self.agent.state.messages m = cfg.model # 🆕 G2: 透传 system_prompt/tools —— 无锚点时度量「下一请求真实载荷」 should, tokens = should_compact(msgs, m, cfg.compaction_reserve, system_prompt=cfg.system_prompt, tools=cfg.tools) # 🆕 每次请求前的校验：print + 写 compaction_diag.log（触发与否都记） compact_diag_log( f"[PRE_CHECK] model={m.name} window={m.context_window} " f"reserve={cfg.compaction_reserve} " f"threshold={m.context_window - cfg.compaction_reserve} " f"est={tokens} triggered={str(should).lower()}") if not should: return False self._notify_compaction_started("pre_prompt", tokens) t0 = time.time() new_msgs = compact_context(msgs, cfg.model, self.summarize_fn, settings=_settings_of(cfg)) if new_msgs is not None: cut_before, first_retained = _cut_ids_of(msgs, new_msgs) self.agent.state.messages = new_msgs self.compactions_performed += 1 self._record_compaction("pre_prompt", tokens, new_msgs, t0, cut_before_id=cut_before, first_retained_id=first_retained) return True self._record_compaction_failed("pre_prompt", tokens, t0) return False # ------------------------------------------------------------------ # 🆕 轮中主动压缩（haocode 增强，偏离 pi 1:1） # ------------------------------------------------------------------ def compact_if_needed(self, msgs: list) -> Optional[list]: """接线到 AgentConfig.compact_fn：内层循环每次 LLM 请求前调用。 与轮首 pre_prompt_compaction 同公式（tokens > window − reserve）。 轮中单条工具输出可把上下文顶出窗口（轮中内存工具结果没有 4000 字回放上限），发请求前主动压缩，避免「一次失败往返 + 响应式兜底」。返回新消息列表（发生了压缩）或 None（不需要/ 不可压缩 → 原样发请求，响应式 overflow 安全网仍在）。 防刷屏护栏：同一 run 内连续失败 2 次后不再尝试（形状无有效 切点，如单条巨型条目，硬试无意义）；成功则复位计数。 """ if self.summarize_fn is None or self._mid_turn_exhausted: return None cfg = self.agent.config m = cfg.model # 🆕 G2: 透传 system_prompt/tools（与轮首同口径） should, tokens = should_compact(msgs, m, cfg.compaction_reserve, system_prompt=cfg.system_prompt, tools=cfg.tools) compact_diag_log( f"[PRE_CHECK] path=mid_turn est={tokens} " f"threshold={m.context_window - cfg.compaction_reserve} " f"triggered={str(should).lower()}") if not should: return None if self._do_compaction("mid_turn", msgs=msgs): self._mid_turn_fail_streak = 0 return list(self.agent.state.messages) self._mid_turn_fail_streak += 1 if self._mid_turn_fail_streak >= 2: self._mid_turn_exhausted = True compact_diag_log( "[PRE_CHECK] path=mid_turn exhausted " "（连续 2 次失败，本轮不再尝试）") return None # ------------------------------------------------------------------ # 主入口 —— 对照 _runAgentPrompt # ------------------------------------------------------------------ def run(self, message) -> RunResult: self.pre_prompt_compaction() self._overflow_recovered = False self._retry_attempt = 0 result = self.agent.prompt(message) return self._post_loop(result) def _post_loop(self, result: RunResult) -> RunResult: # 对照: while (await this._handlePostAgentRun()) await this.agent.continue() while True: action = self._handle_post_agent_run(result) if action == "stop": return result if action == "retry": result = self.agent.continue_() # 🆕 M3: 重试结果回调（成功 = 下一轮未以 error/aborted 收尾） if self.on_retry_finished is not None: la = find_last_assistant(self.agent.state.messages) ok = (la is not None and la.stop_reason not in ("error", "aborted")) try: self.on_retry_finished(bool(ok)) except Exception: pass continue if action == "compact_retry": if not (self._remove_last_bad_assistant() and self._do_compaction(self.last_action)): return result # 压缩不可用 → 放弃，交 UI 报错 result = self.agent.continue_() continue if action == "queued": result = self.agent.continue_() continue return result # ------------------------------------------------------------------ # 三路决策 —— 对照 _handlePostAgentRun # ------------------------------------------------------------------ def _handle_post_agent_run(self, result: RunResult) -> str: msg = find_last_assistant(self.agent.state.messages) if msg is None: return "stop" cfg = self.agent.config # (1) 错误路径 if msg.stop_reason == "error": # a. 溢出 → 压缩恢复（只试一次） if is_context_overflow(msg, cfg.model.context_window) \ and not self._overflow_recovered \ and self.summarize_fn is not None: self._overflow_recovered = True self.last_action = "overflow_compact" return "compact_retry" # b. 可重试错误 → 移除坏消息 + 退避 + continue if is_retryable_assistant_error(msg) \ and self._prepare_retry(cfg.retry, reason=msg.error_message or ""): self.last_action = "retry" return "retry" return "stop" # (2) 可恢复的 length 截断 if is_recoverable_length(msg, cfg.model.max_tokens) \ and not self._overflow_recovered \ and self.summarize_fn is not None: self._overflow_recovered = True self.last_action = "length_compact" return "compact_retry" # (3) 队列里还有消息 → 续跑 if self.agent.has_queued(): self.last_action = "queued" return "queued" # (4) settle return "stop" # ------------------------------------------------------------------ # 对照 _prepareRetry：计数 → 移除坏消息 → 退避睡眠 # ------------------------------------------------------------------ def _prepare_retry(self, cfg: RetryConfig, reason: str = "") -> bool: if self._retry_attempt >= cfg.max_attempts: return False self._retry_attempt += 1 delay_ms = compute_retry_delay_ms(self._retry_attempt, cfg) # 移除错误助手消息（对照: messages.slice(0,-1)，会话历史保留由 UI 层负责） messages = self.agent.state.messages if messages and messages[-1].role == "assistant": self.agent.state.messages = messages[:-1] # 🆕 M3: 通知重试已调度（参数: attempt, max_attempts, delay_ms, reason） if self.on_retry_scheduled is not None: try: self.on_retry_scheduled(self._retry_attempt, cfg.max_attempts, delay_ms, reason) except Exception: pass time.sleep(delay_ms / 1000.0) return True # ------------------------------------------------------------------ # 压缩恢复辅助 # ------------------------------------------------------------------ def _remove_last_bad_assistant(self) -> bool: msgs = self.agent.state.messages if msgs and msgs[-1].role == "assistant" \ and msgs[-1].stop_reason in ("length", "error"): self.agent.state.messages = msgs[:-1] return True return False def _do_compaction(self, path: str = "reactive", msgs: Optional[list] = None) -> bool: if self.summarize_fn is None: return False cfg = self.agent.config if msgs is None: msgs = self.agent.state.messages tokens_before = estimate_context_tokens(msgs).tokens self._notify_compaction_started(path, tokens_before) t0 = time.time() new_msgs = compact_context(msgs, cfg.model, self.summarize_fn, settings=_settings_of(cfg)) if new_msgs is not None: cut_before, first_retained = _cut_ids_of(msgs, new_msgs) self.agent.state.messages = new_msgs self.compactions_performed += 1 self._record_compaction(path, tokens_before, new_msgs, t0, cut_before_id=cut_before, first_retained_id=first_retained) return True self._record_compaction_failed(path, tokens_before, t0) return False # ------------------------------------------------------------------ # 🆕 压缩可视化辅助：开始/完成 两阶段（print + 日志 + 事件供 UI 气泡） # ------------------------------------------------------------------ def _notify_compaction_started(self, path: str, tokens_before: int) -> None: compact_diag_log(f"[COMPACT_START] path={path} before={tokens_before}") if self.on_compaction_started is not None: try: self.on_compaction_started(path) except Exception: pass def _record_compaction_failed(self, path: str, tokens_before: int, t0: float) -> None: """压缩不可行（如历史无法切分）→ 气泡显示失败原因，不能永远卡在「执行中」""" duration_ms = int((time.time() - t0) * 1000) compact_diag_log( f"[COMPACT_FAIL] path={path} before={tokens_before} " f"duration_ms={duration_ms}") ev = { "summary": "本次压缩未能执行（历史无法切分），继续以原有历史对话。", "before": tokens_before, "after": tokens_before, "duration_ms": duration_ms, "path": path, "failed": True, } self.compaction_events.append(ev) if self.on_compaction_finished is not None: try: self.on_compaction_finished(dict(ev)) except Exception: pass def _record_compaction(self, path: str, tokens_before: int, new_msgs: list, t0: float, cut_before_id: str = "", first_retained_id: str = "") -> None: summary = "" if new_msgs and getattr(new_msgs[0], "kind", None) == "compaction_summary": summary = _content_text(new_msgs[0].content) or "" tokens_after = estimate_context_tokens(new_msgs).tokens duration_ms = int((time.time() - t0) * 1000) compact_diag_log( f"[COMPACT_DONE] path={path} before={tokens_before} " f"after={tokens_after} summary_chars={len(summary)} " f"duration_ms={duration_ms} " f"cut_before={cut_before_id or '-'} first_retained={first_retained_id or '-'}") ev = { "summary": summary, "before": tokens_before, "after": tokens_after, "duration_ms": duration_ms, "path": path, # 🆕 压缩持久化切点：UI 收到后在 DB 链上插标记行（两者齐备才插） "cut_before_id": cut_before_id, "first_retained_id": first_retained_id, } self.compaction_events.append(ev) if self.on_compaction_finished is not None: try: self.on_compaction_finished(dict(ev)) except Exception: pass