Agent/cyber_agent/core/memory.py

"""
Memory 系统（Phase 2+）

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

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

from __future__ import annotations

import glob as _glob
import hashlib
import json
import logging
from dataclasses import dataclass
from fnmatch import fnmatchcase
from pathlib import Path, PurePosixPath
from typing import Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)


MEMORY_IDENTITY_CONTEXT_KEY = "memory_identity"


class MemoryPathError(ValueError):
    """记忆配置或 Dream 更新中的路径不安全。"""


@dataclass
class MemoryConfig:
    """持久化记忆配置（见 cyber_agent/docs/memory.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 _normalize_path_parts(path: str, *, label: str) -> Tuple[str, ...]:
    """校验并拆分相对 POSIX 路径。

    MemoryConfig.files 是跨平台持久化配置，因此不接受平台特有的
    反斜杠分隔符，也不容忍 ``.``/``..`` 或空路径段的隐式归一化。
    """
    if not isinstance(path, str) or not path:
        raise MemoryPathError(f"{label} must be a non-empty string")
    if "\x00" in path or "\\" in path:
        raise MemoryPathError(f"{label} contains an invalid separator")
    pure = PurePosixPath(path)
    if pure.is_absolute():
        raise MemoryPathError(f"{label} must be relative")
    parts = tuple(path.split("/"))
    if any(part in ("", ".", "..") for part in parts):
        raise MemoryPathError(f"{label} is not a normalized relative path")
    return parts


def normalize_memory_relative_path(path: str) -> str:
    """返回经严格校验的 Dream 相对文件路径。"""
    return "/".join(_normalize_path_parts(path, label="memory path"))


def _normalize_memory_pattern(pattern: str) -> str:
    """返回经严格校验的 MemoryConfig.files 模式。"""
    return "/".join(_normalize_path_parts(pattern, label="memory file pattern"))


def _match_pattern_parts(
    path_parts: Tuple[str, ...],
    pattern_parts: Tuple[str, ...],
    path_index: int = 0,
    pattern_index: int = 0,
) -> bool:
    """按 glob 语义匹配：``*`` 不跨目录，独立 ``**`` 可跨任意层级。"""
    if pattern_index == len(pattern_parts):
        return path_index == len(path_parts)
    pattern = pattern_parts[pattern_index]
    if pattern == "**":
        return _match_pattern_parts(
            path_parts, pattern_parts, path_index, pattern_index + 1
        ) or (
            path_index < len(path_parts)
            and _match_pattern_parts(
                path_parts, pattern_parts, path_index + 1, pattern_index
            )
        )
    return (
        path_index < len(path_parts)
        and fnmatchcase(path_parts[path_index], pattern)
        and _match_pattern_parts(
            path_parts, pattern_parts, path_index + 1, pattern_index + 1
        )
    )


def is_declared_memory_path(config: MemoryConfig, relative_path: str) -> bool:
    """判断路径是否被 ``MemoryConfig.files`` 精确授权。"""
    if not config.files:
        return False
    path_parts = _normalize_path_parts(relative_path, label="memory path")
    for raw_pattern in config.files:
        pattern = _normalize_memory_pattern(raw_pattern)
        if _match_pattern_parts(path_parts, tuple(pattern.split("/"))):
            return True
    return False


def resolve_memory_path(config: MemoryConfig, relative_path: str) -> Path:
    """解析已声明的记忆路径，并拒绝符号链接和前缀碰撞逃逸。"""
    if not config.base_path:
        raise MemoryPathError("memory base_path is not configured")
    normalized = normalize_memory_relative_path(relative_path)
    if not is_declared_memory_path(config, normalized):
        raise MemoryPathError(
            f"memory path is not declared by MemoryConfig.files: {normalized}"
        )
    base = Path(config.base_path).expanduser().resolve(strict=False)
    target = (base / normalized).resolve(strict=False)
    try:
        target.relative_to(base)
    except ValueError as exc:
        raise MemoryPathError(
            f"memory path escapes base_path: {normalized}"
        ) from exc
    return target


def compute_memory_identity(config: MemoryConfig) -> str:
    """由真实 base 目录和允许模式生成稳定的记忆身份。

    purpose 和 prompt 只影响展示/反思方式，不改变哪组持久化文件属于该身份。
    """
    if not config.base_path:
        raise ValueError("memory_config.base_path is required")
    patterns = sorted(
        _normalize_memory_pattern(pattern) for pattern in (config.files or {})
    )
    canonical = {
        "schema_version": 1,
        "base_path": str(Path(config.base_path).expanduser().resolve(strict=False)),
        "file_patterns": patterns,
    }
    payload = json.dumps(
        canonical, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")
    return f"memory-v1:{hashlib.sha256(payload).hexdigest()}"


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).expanduser()
    if not base.exists():
        logger.debug(f"[Memory] base_path 不存在: {base}")
        return []
    base_resolved = base.resolve(strict=False)

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

    for key, purpose in config.files.items():
        try:
            safe_pattern = _normalize_memory_pattern(key)
        except MemoryPathError as exc:
            logger.warning(f"[Memory] 忽略不安全的文件模式 {key!r}: {exc}")
            continue
        # 展开 glob；直接路径也走 glob（无通配符时返回单条或空）
        pattern = str(base / safe_pattern)
        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:
            candidate = Path(fs_path)
            if not candidate.is_file():
                continue
            resolved = candidate.resolve(strict=False)
            try:
                rel = resolved.relative_to(base_resolved).as_posix()
            except ValueError:
                logger.warning(f"[Memory] 拒绝读取 base_path 之外的文件: {fs_path}")
                continue
            try:
                if not is_declared_memory_path(config, rel):
                    logger.warning(f"[Memory] 拒绝读取未声明的记忆文件: {rel}")
                    continue
            except MemoryPathError as exc:
                logger.warning(f"[Memory] 拒绝读取非规范路径 {rel!r}: {exc}")
                continue
            if rel in seen:
                continue
            seen.add(rel)
            try:
                content = resolved.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"