Agent/agent/core/memory.py

"""
Memory 系统（Phase 2+）

详见 agent/docs/memory-plan.md。核心概念：
- Memory：Agent 身份私有的主观记忆，Markdown 文件，人类可读写
- Dream：记忆反思操作（回顾多个 trace 的执行历史，更新记忆文件）

本模块只提供 MemoryConfig 数据类和记忆文件加载逻辑。
Dream 操作在 agent/core/dream.py（Phase 3）。
"""

from __future__ import annotations

import glob as _glob
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)


@dataclass
class MemoryConfig:
    """持久化记忆配置（见 agent/docs/memory-plan.md 第五节）"""

    base_path: str = ""
    # 记忆文件根目录。所有文件路径相对此目录解析。

    files: Optional[Dict[str, str]] = None
    # {路径模式: 用途说明}
    # key 支持两种形式：
    #   - 直接路径："core/identity.md"
    #   - glob 模式："relationships/*.md"、"journals/2026/**.md"
    # value 是人类可读的用途说明（注入时作为文件分隔标题的一部分）。
    # 框架只负责按 key 解析文件内容；组织结构由配置者决定。

    dream_prompt: str = ""
    # Dream 跨 trace 整合 prompt；空则使用默认（Phase 3 定义）

    reflect_prompt: str = ""
    # Per-trace 记忆反思 prompt；空则使用默认（Phase 3 定义）


def load_memory_files(config: MemoryConfig) -> List[Tuple[str, str, str]]:
    """按 MemoryConfig.files 的 key 解析磁盘上的记忆文件。

    Returns:
        List[(relative_path, purpose, content)]，按 files 声明顺序扁平化，
        文件不存在则跳过（记 debug 日志），内容为空也保留（方便人类看到占位）。
    """
    if not config.base_path or not config.files:
        return []

    base = Path(config.base_path)
    if not base.exists():
        logger.debug(f"[Memory] base_path 不存在: {base}")
        return []

    results: List[Tuple[str, str, str]] = []
    seen: set[str] = set()  # 去重（多个 glob 可能命中同一个文件）

    for key, purpose in config.files.items():
        # 展开 glob；直接路径也走 glob（无通配符时返回单条或空）
        pattern = str(base / key)
        matched_paths = sorted(_glob.glob(pattern, recursive=True))

        if not matched_paths:
            # 直接路径没命中时给个 debug（可能还没写第一版）
            logger.debug(f"[Memory] {key} 没有匹配文件（尚未创建）")
            continue

        for fs_path in matched_paths:
            rel = str(Path(fs_path).relative_to(base))
            if rel in seen:
                continue
            seen.add(rel)
            try:
                content = Path(fs_path).read_text(encoding="utf-8")
            except Exception as e:
                logger.warning(f"[Memory] 读取失败 {fs_path}: {e}")
                continue
            results.append((rel, purpose, content))

    return results


def format_memory_injection(files: List[Tuple[str, str, str]]) -> str:
    """把加载结果格式化为可注入到上下文的 markdown 段。"""
    if not files:
        return ""
    parts = ["## 你的长期记忆\n\n以下是你作为此 Agent 身份积累的记忆（人类可直接编辑）：\n"]
    for rel, purpose, content in files:
        header = f"### `{rel}`"
        if purpose:
            header += f" — {purpose}"
        parts.append(header)
        parts.append(content.rstrip() or "_（空文件，尚未积累内容）_")
        parts.append("")  # 空行分隔
    return "\n".join(parts).rstrip() + "\n"