改进心跳机制和每日总结检查

TevinClaw

yesterday c1421933de3545033a1a56036376b1487155321c

改进心跳机制和每日总结检查

- HEARTBEAT.md: 重构任务清单结构，每日总结作为每日任务的第一项
- SKILL.md: 添加完整的session扫描检查流程和判断标准
- daily_check.py: 新增扫描所有session文件功能（包括.reset.和.deleted.归档）

改进点：
1. 扫描所有今日修改的session文件，不只是当前活跃session
2. 提取飞书渠道用户消息并识别重要事件
3. 明确判断标准：怎样才算'没有遗漏'
4. 预留任务扩展空间，方便添加更多心跳任务

 workspace/HEARTBEAT.md

@@ -6,23 +6,80 @@

## 任务清单

### 1. 三层记忆每日总结（由 memory-management 技能处理）
### 每日任务（晚上10点后执行）

#### 1. 三层记忆每日总结（由 memory-management 技能处理）

**触发条件**: 时间 ≥ 22:00 且当日无 L2 记录  
**执行技能**: [memory-management](../skills/memory-management/SKILL.md)  
**执行脚本**: `skills/memory-management/scripts/daily_check.py`

**逻辑**:
**执行逻辑**:
```
收到心跳请求
    │
    ▼
读取 HEARTBEAT.md 获取任务清单
    │
    ▼
时间 ≥ 22:00 ?
    └── 是 → 今日 L2 已存在 ?
              └── 否 → 执行每日总结
    ├── 否 → 回复 HEARTBEAT_OK（时间未到）
    │
    └── 是 → 检查今日 L2 是否存在
              │
              ├── 是 → 回复 HEARTBEAT_OK（已记录）
              │
              └── 否 → 执行每日检查脚本
                        │
                        ▼
                  扫描所有session文件
                  （当前活跃 + .reset.归档 + .deleted.删除）
                        │
                        ▼
                  分析内容识别重要事件
                        │
                        ▼
                  生成每日总结建议
```

**完整检查流程（确保无遗漏）**:

1. **扫描所有Session文件**
   - 当前活跃: `*.jsonl`
   - 重置归档: `*.jsonl.reset.*`
   - 删除归档: `*.jsonl.deleted.*`
   - 检查今日修改时间戳

2. **提取飞书渠道对话**
   - 解析每个session文件
   - 识别 `channel: feishu` 的消息
   - 提取用户发送的文本内容

3. **识别重要事件类型**
   - 技能安装/更新（关键词: skill, 安装, 创建）
   - 配置变更（关键词: config, 配置, API key）
   - 定时任务（关键词: cron, 定时）
   - 重要对话/决策

4. **判断标准（怎样才算"没有遗漏"）**
   ```
   ✅ 检查完成标准：
   ├── 已扫描今日所有修改过的session文件（≥1个）
   ├── 已检查.reset.和.deleted.归档文件
   ├── 已提取飞书渠道对话记录
   ├── 已识别所有重要事件类型
   └── 已生成L2记录或确认无需记录
   
   ❌ 遗漏警告：
   ├── 发现今日session文件 > 0
   ├── 但今日L2记录不存在
   └── → 必须人工检查补充
```

**动作**:
- 扫描当日活动、决策、事件
- 创建 L2 记录 (`memory/journal/YYYY-MM-DD.md`)
- 更新 L0 索引
- 执行 `daily_check.py` 扫描所有session
- 如发现活动但未记录 → 提示需要补充L2
- 更新 MEMORY.md 的"最近活动"摘要
- 检查 L0 大小

---

 workspace/skills/memory-management/SKILL.md

@@ -112,10 +112,23 @@

**任务清单：**
- [ ] 检查今日是否有重要决策需要记录到L2
- [ ] **检查飞书渠道历史** — 如用户询问"检查昨天的每日总结"，需读取所有session并提取飞书渠道的完整聊天记录补充到L2
- [ ] **完整检查飞书渠道历史** — 执行以下步骤确保无遗漏：
  1. 列出所有今日活跃的session（包括.reset.归档文件）
  2. 扫描 `~/.openclaw/agents/main/sessions/` 目录下今日修改的所有 `.jsonl*` 文件
  3. 读取每个session文件，提取飞书渠道的对话内容
  4. 检查是否有未记录到L2的重要活动
- [ ] 更新 MEMORY.md 的"最近活动"摘要
- [ ] 确保 L0 层不超过 4KB

**判断标准：怎样才算"没有遗漏"？**
```
检查完成标准：
├── 已扫描所有session文件（当前活跃 + .reset.归档 + .deleted.删除）
├── 已提取飞书渠道所有对话记录
├── 已识别所有重要事件（技能安装、配置变更、定时任务等）
└── 已在L2中记录或确认无需记录
```

**重要提醒：**
> ⚠️ **飞书历史检查**：当用户说"检查昨天的每日总结"或类似表述时，必须：
> 1. 使用 `sessions_list` 查找过去48小时的活跃session

 workspace/skills/memory-management/scripts/daily_check.py

@@ -1,18 +1,26 @@
#!/usr/bin/env python3
"""
每日记忆检查脚本
在晚上10点后触发，检查今日是否已写入L2
在晚上10点后触发，检查今日是否已写入L2，并扫描session确保无遗漏
"""

import os
import sys
import json
import re
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Tuple


def get_workspace_path() -> Path:
    """获取workspace路径。"""
    return Path.home() / ".openclaw" / "workspace"


def get_sessions_path() -> Path:
    """获取sessions路径。"""
    return Path.home() / ".openclaw" / "agents" / "main" / "sessions"


def check_today_journal() -> bool:
@@ -38,6 +46,152 @@
    return f"{kb:.1f}KB"


def get_today_session_files() -> List[Path]:
    """
    获取今日所有session文件（包括.reset.和.deleted.归档）
    这是确保"没有遗漏"的关键步骤
    """
    sessions_dir = get_sessions_path()
    if not sessions_dir.exists():
        return []
    
    today = datetime.now()
    today_files = []
    
    # 扫描所有.jsonl文件（包括.reset.和.deleted.）
    for file in sessions_dir.glob("*.jsonl*"):
        try:
            # 检查文件修改时间
            mtime = datetime.fromtimestamp(file.stat().st_mtime)
            if mtime.date() == today.date():
                today_files.append(file)
        except (OSError, ValueError):
            continue
    
    # 按修改时间排序
    today_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
    return today_files


def extract_feishu_messages(file_path: Path, max_messages: int = 50) -> List[Dict]:
    """
    从session文件中提取飞书渠道的消息
    返回用户发送的消息列表
    """
    messages = []
    
    try:
        with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
            for line_num, line in enumerate(f):
                if line_num >= max_messages * 3:  # 限制读取行数
                    break
                
                line = line.strip()
                if not line:
                    continue
                
                try:
                    record = json.loads(line)
                    
                    # 只处理消息类型
                    if record.get("type") != "message":
                        continue
                    
                    msg = record.get("message", {})
                    if not msg:
                        continue
                    
                    # 检查是否是用户消息（role为user）
                    if msg.get("role") != "user":
                        continue
                    
                    # 提取内容
                    content_list = msg.get("content", [])
                    if not content_list:
                        continue
                    
                    # 查找文本内容
                    text_content = ""
                    for item in content_list:
                        if isinstance(item, dict) and item.get("type") == "text":
                            text = item.get("text", "")
                            # 过滤掉系统消息
                            if text and not text.startswith("[") and len(text) > 10:
                                text_content = text
                                break
                    
                    if text_content:
                        messages.append({
                            "timestamp": record.get("timestamp", ""),
                            "content": text_content[:200]  # 限制长度
                        })
                        
                        if len(messages) >= max_messages:
                            break
                            
                except json.JSONDecodeError:
                    continue
                    
    except (IOError, OSError) as e:
        print(f"  警告：无法读取文件 {file_path.name}: {e}")
    
    return messages


def analyze_sessions_for_events() -> Tuple[bool, List[str]]:
    """
    分析今日session，检查是否有重要事件需要记录
    返回：(是否需要补充记录, 事件列表)
    """
    print("\n🔍 扫描今日session文件（检查是否遗漏）:")
    
    session_files = get_today_session_files()
    
    if not session_files:
        print("  ⚠️  未找到今日session文件")
        return False, []
    
    print(f"  找到 {len(session_files)} 个session文件:")
    for f in session_files:
        mtime = datetime.fromtimestamp(f.stat().st_mtime)
        print(f"    - {f.name} ({mtime.strftime('%H:%M')})")
    
    # 关键词列表，用于识别重要事件
    important_keywords = [
        "安装", "创建", "配置", "定时任务", "cron", "技能", "skill",
        "早报", "更新", "修改", "决策", "设定"
    ]
    
    found_events = []
    total_user_messages = 0
    
    for file_path in session_files:
        messages = extract_feishu_messages(file_path, max_messages=20)
        total_user_messages += len(messages)
        
        for msg in messages:
            content = msg["content"]
            # 检查是否包含重要事件关键词
            for keyword in important_keywords:
                if keyword in content and len(content) > 20:
                    event_summary = content[:100] + "..." if len(content) > 100 else content
                    if event_summary not in found_events:
                        found_events.append(event_summary)
                    break
    
    print(f"\n  提取到 {total_user_messages} 条用户消息")
    
    if found_events:
        print(f"  识别到 {len(found_events)} 个可能的重要事件:")
        for i, event in enumerate(found_events[:5], 1):  # 只显示前5个
            print(f"    {i}. {event}")
    
    # 判断是否需要补充记录
    needs_update = len(found_events) >= 2 and not check_today_journal()
    
    return needs_update, found_events


def main():
    """主函数。"""
    today_str = datetime.now().strftime("%Y-%m-%d")
@@ -51,7 +205,18 @@
        print("  ✅ 今日已有journal记录")
    else:
        print("  ⚠️  今日尚未创建journal记录")
        print("  💡 建议：如有重要决策或事件，写入L2详情层")
    
    # 关键步骤：扫描session文件确保无遗漏
    needs_update, events = analyze_sessions_for_events()
    
    if needs_update:
        print(f"\n🚨 发现遗漏：今日有session活动但未写入L2")
        print(f"   识别到 {len(events)} 个事件需要记录")
        print("   建议：执行 '检查今天的session并生成总结'")
    elif has_today_journal:
        print("\n  ✅ 已记录L2，session扫描完成")
    else:
        print("\n  ⚠️  今日无重要活动或已记录完毕")
    
    # 检查L0大小
    l0_size = get_l0_size()
@@ -67,16 +232,28 @@
    
    print("\n" + "=" * 50)
    print("📋 每日维护清单:")
    if not has_today_journal:
        print("  [ ] 如有重要事件，写入今日L2")
    else:
    if has_today_journal:
        print("  [x] L2记录已存在")
    else:
        print("  [ ] 如有重要事件，写入今日L2")
    
    if session_files := get_today_session_files():
        print(f"  [x] 已扫描 {len(session_files)} 个session文件")
    else:
        print("  [-] 今日无session活动")
    
    print("  [ ] 检查MEMORY.md最近活动摘要")
    if l0_size > 3500:
        print("  [ ] L0接近限制，考虑归档到L1")
    print("  [ ] 确认L0层引用链接有效")
    
    return 0 if has_today_journal else 1
    # 返回状态码
    if needs_update:
        return 2  # 需要补充记录
    elif not has_today_journal:
        return 1  # 无L2记录
    else:
        return 0  # 一切正常


if __name__ == "__main__":