长任务与失败恢复

import hashlib, json
from dataclasses import dataclass, field
@dataclass
class CompactionPolicy:
    window: int = 128_000
    trigger_ratio: float = 0.60      # 到 60% 就开始压，留出余量
    keep_recent_turns: int = 4       # 最近几轮原样保留
    max_single_obs_ratio: float = 0.10   # 单条工具结果不得超过窗口的 10%
    summarize: object = None         # 注入的摘要函数（调用模型）
@dataclass
class Step:
    turn: int
    tool: str
    status: str
    brief: str                       # 该步的一句话说明
    result_fingerprint: str          # 结果指纹，用于判重与去噪
    full_output: str | None = None   # 完整输出：压缩后丢弃，只留摘要
def estimate_tokens(messages: list[dict]) -> int:
    """粗略估算。生产环境请用真实 tokenizer，这里按字符数近似"""
    return sum(len(str(m.get("content", ""))) for m in messages) // 2
def trim_single_observation(obs: str, limit_chars: int) -> str:
    """
    单条超长结果的头尾保留裁剪。
    关键：必须显式标注省略了多少，否则模型会以为这就是全部内容。
    """
    if len(obs) <= limit_chars:
        return obs
    head = obs[: int(limit_chars * 0.6)]
    tail = obs[-int(limit_chars * 0.3):]
    omitted = len(obs) - len(head) - len(tail)
    return f"{head}\n……[此处省略 {omitted} 字符]……\n{tail}"
def compact(messages: list[dict], steps: list[Step], policy: CompactionPolicy) -> tuple[list[dict], str]:
    """
    返回 (压缩后的消息列表, 压缩事件说明)。
    三件事按顺序做：单条裁剪 → 旧步骤摘要 → 结论固化到头部。
    """
    events = []
    # ---- 第一步：单条超长结果裁剪（成本最低，先做） ----
    limit = int(policy.window * policy.max_single_obs_ratio * 2)   # 字符数近似
    for m in messages:
        if m.get("role") == "tool" and len(str(m.get("content", ""))) > limit:
            before = len(str(m["content"]))
            m["content"] = trim_single_observation(str(m["content"]), limit)
            events.append(f"裁剪单条工具结果 {before}→{limit} 字符")
    # ---- 第二步：用量仍超阈值，则摘要旧步骤 ----
    if estimate_tokens(messages) > policy.window * policy.trigger_ratio:
        old = steps[:-policy.keep_recent_turns] if len(steps) > policy.keep_recent_turns else []
        if old:
            summary = structured_summary(old, policy.summarize)
            # 用摘要替换掉旧的具体消息，但保留最近几轮原文
            kept = messages[:3] + [{"role": "user", "content": f"【已完成步骤摘要】\n{summary}"}]
            kept += messages[-(policy.keep_recent_turns * 2):]
            messages = kept
            events.append(f"摘要压缩 {len(old)} 个步骤，上下文降至 {estimate_tokens(messages)} token")
    # ---- 第三步：把不可丢失的结论固化到头部（防止被后续压缩丢掉） ----
    pinned = [s for s in steps if s.status == "ok" and s.brief]
    if pinned:
        digest = "；".join(f"{s.tool}:{s.brief}" for s in pinned[-10:])
        # 头部插入「钉住的事实」，这部分不再参与后续压缩
        messages.insert(2, {"role": "user", "content": f"【已确认的关键结论】{digest}"})
    return messages, "　|　".join(events) if events else "无需压缩"
def structured_summary(steps: list[Step], summarize_fn) -> str:
    """
    结构化摘要：固定四段。自由文本摘要会越摘越糊，
    而固定字段能保证「目标 / 已完成 / 结论 / 待办」都不丢。
    """
    done = [s for s in steps if s.status == "ok"]
    failed = [s for s in steps if s.status != "ok"]
    return (
        f"目标：见任务描述（未变更）\n"
        f"已完成（{len(done)}）：" + "；".join(f"{s.tool}→{s.brief}" for s in done) + "\n"
        f"失败/已放弃（{len(failed)}）：" + "；".join(f"{s.tool}→{s.status}" for s in failed) + "\n"
        f"待办：目标中尚未被任何步骤覆盖的部分"
    )
def fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()[:12]

import json, os, time
from dataclasses import dataclass, asdict, field
STATE_DIR = os.environ.get("HT_STATE_DIR", "./.ht_state")
@dataclass
class Checkpoint:
    task_id: str
    goal: str                                  # 原始目标，永不压缩
    constraints: list[str]                     # 用户明确给出的约束，永不压缩
    pinned_conclusions: list[str]              # 已确认结论，只追加
    pending: list[str]                         # 待办
    side_effects: list[dict] = field(default_factory=list)  # 已发生的副作用
    version: int = 1
    updated_at: float = field(default_factory=time.time)
def save(cp: Checkpoint) -> None:
    """原子写：先写临时文件再重命名，避免中断导致快照损坏"""
    os.makedirs(STATE_DIR, exist_ok=True)
    path = os.path.join(STATE_DIR, f"{cp.task_id}.json")
    tmp = path + ".tmp"
    cp.updated_at = time.time()
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(asdict(cp), f, ensure_ascii=False, indent=1)
    os.replace(tmp, path)          # 原子替换
def load(task_id: str) -> Checkpoint | None:
    path = os.path.join(STATE_DIR, f"{task_id}.json")
    if not os.path.exists(path):
        return None
    with open(path, encoding="utf-8") as f:
        return Checkpoint(**json.load(f))
def idempotent_side_effect(cp: Checkpoint, step_no: int, action: str, run) -> dict:
    """
    所有副作用必须走这里。
    恢复后重复调用会命中 side_effects 记录，直接返回上次结果。
    """
    key = f"{cp.task_id}:{step_no}:{action}"
    for rec in cp.side_effects:
        if rec["key"] == key:
            return {"status": "already_done", "result": rec["result"]}
    result = run()
    cp.side_effects.append({"key": key, "action": action,
                            "result": result, "at": time.time()})
    save(cp)                    # 立即落盘：副作用发生后必须马上可查
    return {"status": "executed", "result": result}
def resume(task_id: str, current_tools: set[str]) -> tuple[Checkpoint, list[str]]:
    """
    恢复：只重放未完成步骤，并检查工具集是否变更。
    如果某个待办步骤依赖的工具已经不存在，必须显式报告而不是静默跳过。
    """
    cp = load(task_id)
    if cp is None:
        raise FileNotFoundError(f"没有找到任务 {task_id} 的检查点")
    problems = []
    for step in list(cp.pending):
        tool = step.split("::")[0] if "::" in step else step
        if tool not in current_tools:
            problems.append(f"待办步骤依赖的工具 {tool} 已不可用，需重新规划：{step}")
    return cp, problems