520 lines
24 KiB
Python
520 lines
24 KiB
Python
"""
|
||
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
|