graph LR
    subgraph "METR时长边界指数曲线"
        A["GPT-4<br/>~4min<br/>2023Q1"] --> B["Claude 3.5<br/>~15min<br/>2024Q3"]
        B --> C["GPT-5.2<br/>~2h<br/>2025Q3"]
        C --> D["Opus 4.6<br/>~14.5h<br/>2026Q1"]
        D --> E["2027预测<br/>~100-200h"]
    end

    subgraph "任务层级映射"
        T1["分钟级: bug修复"] --> T2["小时级: 功能开发"]
        T2 --> T3["日级: 模块重构"]
        T3 --> T4["周级: 系统迁移"]
    end

    D -.->|50%覆盖| T3
    E -.->|50%覆盖| T4

graph TD
    subgraph "DeepPlanning规划深度衰减曲线"
        S1["1-5步: >80%<br/>✅ 简单推理"] --> S2["6-10步: ~65%<br/>⚠️ 分化开始"]
        S2 --> S3["11-20步: ~40%<br/>🔴 认知负荷临界"]
        S3 --> S4["20-50步: ~15%<br/>💀 规划链断裂"]
        S4 --> S5["50+步: <5%<br/>❌ 全面崩塌"]
    end

    subgraph "崩塌根因"
        R1["上下文窗口溢出"] --> RC["遗忘早期约束"]
        R2["错误累积效应"] --> RC
        R3["反馈循环缺失"] --> RC
        RC --> FAIL["任务链崩溃"]
    end

    S4 --> R1
    S4 --> R2
    S4 --> R3

# DeepPlanning评估核心逻辑（简化）
from deepplanning import DeepPlanningBenchmark, RubricScorer

bench = DeepPlanningBenchmark(
    difficulty_levels=["easy", "medium", "hard", "extreme"],
    constraint_types=["time_budget", "resource_limit", "feasibility"]
)

# 对模型进行规划深度梯度测试
results = {}
for depth in [5, 10, 20, 50, 100]:
    task = bench.sample_tasks(
        min_steps=depth, 
        max_steps=depth * 1.5,
        n_tasks=50
    )

    scores = []
    for t in task:
        plan = model.plan(task=t.description, constraints=t.constraints)
        rubric = RubricScorer.evaluate(
            plan=plan,
            ground_truth=t.optimal_plan,
            dimensions=[
                "completeness",    # 规划完整性
                "ordering",        # 步骤顺序正确性
                "constraint_adherence",  # 约束遵守度
                "information_gathering"  # 主动信息收集策略
            ]
        )
        scores.append(rubric.weighted_score())

    results[depth] = {
        "mean_score": np.mean(scores),
        "cliff_detected": check_score_cliff(scores, threshold=0.3)
    }

# 输出规划深度断层分析
plot_planning_cliff(results)

graph TB
    subgraph "长程Agent Harness三层架构"
        L1["Layer 1: Planner Agent<br/>任务分解·约束检查·里程碑设定"]
        L2["Layer 2: Executor Agent<br/>代码生成·工具调用·状态推进"]
        L3["Layer 3: Evaluator Agent<br/>质量把关·回滚决策·上下文压缩"]

        L1 -->|"子任务列表"| L2
        L2 -->|"执行结果"| L3
        L3 -->|"通过/回滚"| L2
        L3 -->|"里程碑报告"| L1
        L1 -->|"重新规划"| L2
    end

    subgraph "状态持久化层"
        CK["Checkpoint Store"]
        HB["Heartbeat Monitor"]
        SQ["State Queue"]
    end

    L2 <--> CK
    L2 <--> HB
    L3 <--> SQ

# 生产级长程Agent的Checkpoint-Resume核心逻辑
import json, hashlib
from datetime import datetime
from pathlib import Path

class AgentCheckpoint:
    """长程Agent的检查点管理器"""

    def __init__(self, task_id: str, store_path: str = "./checkpoints"):
        self.task_id = task_id
        self.store = Path(store_path) / task_id
        self.store.mkdir(parents=True, exist_ok=True)

    def save(self, step: int, state: dict, plan: list):
        """保存检查点：包含执行状态和剩余计划"""
        checkpoint = {
            "task_id": self.task_id,
            "step": step,
            "timestamp": datetime.now().isoformat(),
            "state_hash": hashlib.sha256(
                json.dumps(state, sort_keys=True).encode()
            ).hexdigest()[:16],
            "state": state,          # 当前执行上下文
            "remaining_plan": plan,  # 剩余规划步骤
            "token_count": state.get("total_tokens", 0)
        }
        path = self.store / f"step_{step:04d}.json"
        path.write_text(json.dumps(checkpoint, indent=2, ensure_ascii=False))
        return path

    def resume(self) -> tuple[int, dict, list]:
        """从最新检查点恢复"""
        checkpoints = sorted(self.store.glob("step_*.json"))
        if not checkpoints:
            return 0, {}, []
        latest = json.loads(checkpoints[-1].read_text())
        return latest["step"], latest["state"], latest["remaining_plan"]

    def should_checkpoint(self, tokens_used: int, 
                          steps_completed: int) -> bool:
        """决策何时保存检查点"""
        # 每消耗50K tokens或每完成5步保存一次
        TOKEN_INTERVAL = 50_000
        STEP_INTERVAL = 5
        last_ckpt_steps = len(list(self.store.glob("step_*.json")))
        return (
            tokens_used % TOKEN_INTERVAL < 5000 or
            steps_completed % STEP_INTERVAL == 0
        )


# 使用示例：长程Agent主循环
def run_long_task(task_description: str, model, max_steps: int = 100):
    ckpt = AgentCheckpoint(task_id=hashlib.md5(
        task_description.encode()).hexdigest()[:12])

    step, state, plan = ckpt.resume()
    if step > 0:
        print(f"从步骤 {step} 恢复执行，剩余 {len(plan)} 个子任务")

    for i, subtask in enumerate(plan[step:], start=step):
        result = model.execute(subtask, context=state)
        state.update(result.new_state)

        # 评估子任务结果
        eval_result = model.evaluate(result, subtask.acceptance_criteria)
        if not eval_result.passed:
            # 回滚到上一个检查点
            step, state, plan = ckpt.resume()
            plan = replan(plan, step, eval_result.failure_reason)
            continue

        if ckpt.should_checkpoint(state["total_tokens"], i):
            ckpt.save(step=i, state=state, plan=plan)

    return state

graph TD
    START["任务时长预估"] --> Q1{"<30分钟?"}
    Q1 -->|是| A["单Agent<br/>Gemini Flash / Sonnet<br/>成本优先"]
    Q1 -->|否| Q2{"<4小时?"}
    Q2 -->|是| B["双Agent<br/>Opus 4.6 规划+执行<br/>质量优先"]
    Q2 -->|否| Q3{"<14小时?"}
    Q3 -->|是| C["三Agent Harness<br/>Opus规划 + Codex执行<br/>+ Evaluator把关"]
    Q3 -->|否| Q4{"是否可拆分?"}
    Q4 -->|是| D["拆分子任务<br/>每个<4h，并行执行<br/>用B模式逐个处理"]
    Q4 -->|否| E["人机协作<br/>AI处理可自动化部分<br/>人工决策关键节点"]