fix(memory): 隔离 Dream 身份并加固记忆写入事务

以真实 base_path 和声明文件模式生成 memory_identity，并要求 DreamScope 同时精确匹配 uid、agent_type 与 memory_identity，禁止无过滤器扫描全部 Trace。

移除 Runner 共享 MemoryConfig 的依赖，改由每次工具调用的隐藏上下文传递配置和身份，避免并发 Runner 串用记忆目录。

统一校验规范化相对路径、声明 glob、目录边界和符号链接逃逸；所有更新先持久化 staging，提交中途失败时逆序恢复已发布文件。

仅在全部记忆文件更新成功后消费 reflection；新增 13 项安全测试，覆盖并发隔离、路径穿越、前缀碰撞、未声明文件、符号链接及多文件回滚。

cyber_agent/core/dream.py

@@ -7,20 +7,31 @@ Dream：记忆反思操作（Phase 3）
                              用 dream_prompt 指导 LLM 更新记忆文件
 
 对外入口：
-    run_dream(store, llm_call, memory_config, trace_filter=None, model=...)
+    run_dream(store, llm_call, memory_config, dream_scope, model=...)
 """
 
 from __future__ import annotations
 
 import json
 import logging
+import os
 import re
+import tempfile
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
 
-from cyber_agent.core.memory import MemoryConfig, load_memory_files, format_memory_injection
+from cyber_agent.core.memory import (
+    MEMORY_IDENTITY_CONTEXT_KEY,
+    MemoryConfig,
+    MemoryPathError,
+    compute_memory_identity,
+    format_memory_injection,
+    load_memory_files,
+    normalize_memory_relative_path,
+    resolve_memory_path,
+)
 from cyber_agent.trace.models import Trace
 from cyber_agent.trace.store import FileSystemTraceStore
 
@@ -65,6 +76,42 @@ DEFAULT_DREAM_PROMPT = """你正在整理自己的长期记忆。下面是你最
 
 # ===== 数据结构 =====
 
+@dataclass(frozen=True)
+class DreamScope:
+    """Dream 可见 Trace 的完整身份边界。
+
+    uid 允许为 None，但仍按 None 做精确匹配；不会将其视为
+    “不过滤”。
+    """
+
+    uid: Optional[str]
+    agent_type: str
+    memory_identity: str
+
+    def __post_init__(self) -> None:
+        if not self.agent_type:
+            raise ValueError("DreamScope.agent_type is required")
+        if not self.memory_identity:
+            raise ValueError("DreamScope.memory_identity is required")
+
+    def matches(self, trace: Optional[Trace]) -> bool:
+        if trace is None:
+            return False
+        context = trace.context if isinstance(trace.context, dict) else {}
+        return (
+            trace.uid == self.uid
+            and trace.agent_type == self.agent_type
+            and context.get(MEMORY_IDENTITY_CONTEXT_KEY) == self.memory_identity
+        )
+
+
+class DreamScopeError(ValueError):
+    """DreamScope 与 MemoryConfig 或 Trace 身份不一致。"""
+
+
+class DreamPlanError(ValueError):
+    """LLM 产生的记忆更新计划不安全或不完整。"""
+
 @dataclass
 class DreamReport:
     per_trace_summaries: Dict[str, str] = field(default_factory=dict)  # {trace_id: summary}
@@ -84,6 +131,7 @@ async def per_trace_reflect(
     llm_call: LLMCall,
     trace_id: str,
     memory_config: MemoryConfig,
+    dream_scope: DreamScope,
     model: str = "gpt-4o-mini",
 ) -> Optional[str]:
     """为单个 trace 生成反思摘要，写入 cognition_log，更新 reflected_at_sequence。
@@ -92,8 +140,8 @@ async def per_trace_reflect(
         反思摘要字符串；若 trace 没有新消息或 LLM 返回空，返回 None。
     """
     trace = await store.get_trace(trace_id)
-    if not trace:
-        logger.debug(f"[Dream] trace 不存在: {trace_id}")
+    if not dream_scope.matches(trace):
+        logger.warning(f"[Dream] trace 超出 DreamScope，拒绝反思: {trace_id}")
         return None
 
     start_seq = (trace.reflected_at_sequence or 0) + 1
@@ -198,21 +246,17 @@ async def cross_trace_integrate(
     store: FileSystemTraceStore,
     llm_call: LLMCall,
     memory_config: MemoryConfig,
-    trace_filter: Optional[Callable[[Trace], bool]] = None,
+    dream_scope: DreamScope,
     model: str = "gpt-4o",
 ) -> Tuple[int, List[str], str]:
     """汇总各 trace 未消化的 reflection 事件，用 LLM 更新记忆文件。
 
-    Args:
-        trace_filter: 可选的 trace 过滤函数（例如按 agent_type / owner）；
-                      None 表示扫描 TraceStore 下所有 trace。
-
     Returns:
         (consumed_reflection_count, updated_file_paths, reasoning)
     """
+    _validate_dream_scope(memory_config, dream_scope)
     all_traces = await store.list_traces(limit=1000)
-    if trace_filter:
-        all_traces = [t for t in all_traces if trace_filter(t)]
+    all_traces = [t for t in all_traces if dream_scope.matches(t)]
 
     # 收集所有未消化的 reflection 事件
     reflections: List[Tuple[str, Dict[str, Any]]] = []  # [(trace_id, event)]
@@ -229,8 +273,6 @@ async def cross_trace_integrate(
 
     # 读当前记忆文件
     existing_files = load_memory_files(memory_config)
-    existing_by_path = {rel: (purpose, content) for rel, purpose, content in existing_files}
-
     user_content = _build_dream_input(reflections, existing_files, memory_config)
     prompt = memory_config.dream_prompt or DEFAULT_DREAM_PROMPT
 
@@ -254,42 +296,209 @@ async def cross_trace_integrate(
         logger.error(f"[Dream] LLM 输出无法解析为 JSON 计划，原文: {raw[:500]}")
         return 0, [], ""
 
-    updated_paths: List[str] = []
-    base = Path(memory_config.base_path)
+    try:
+        prepared_updates = _prepare_memory_updates(memory_config, plan)
+    except (DreamPlanError, MemoryPathError) as exc:
+        logger.error(f"[Dream] 拒绝不安全的记忆更新计划: {exc}")
+        return 0, [], ""
 
-    for update in plan.get("updates", []):
-        rel_path = update.get("path", "")
-        new_content = update.get("new_content", "")
-        if not rel_path:
-            continue
-        # 安全检查：禁止路径穿越
-        target = (base / rel_path).resolve()
-        if not str(target).startswith(str(base.resolve())):
-            logger.warning(f"[Dream] 拒绝写入 base_path 之外的路径: {rel_path}")
-            continue
-        target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_text(new_content, encoding="utf-8")
-        updated_paths.append(rel_path)
-        logger.info(f"[Dream] 已更新记忆文件: {rel_path} ({len(new_content)} chars)")
+    # LLM 返回后再次检查 Trace 身份，防止在长耗时调用期间边界变化。
+    for trace_id in {trace_id for trace_id, _ in reflections}:
+        if not dream_scope.matches(await store.get_trace(trace_id)):
+            raise DreamScopeError(
+                f"trace left DreamScope before memory update: {trace_id}"
+            )
 
-    # 标记所有参与的 reflection 为已消化
+    updated_paths = _transactional_write_memory_updates(
+        memory_config,
+        prepared_updates,
+    )
+
+    # 只有全部记忆文件原子替换成功后，才标记 reflection 已消化。
     consumed_at = datetime.now().isoformat()
+    reflections_by_trace: Dict[str, List[Dict[str, Any]]] = {}
     for trace_id, event in reflections:
-        log = await store.get_cognition_log(trace_id)
-        events = log.get("events", log.get("entries", []))
-        target_ts = event.get("timestamp")
-        for e in events:
-            if (
-                e.get("type") == "reflection"
-                and not e.get("consumed_at")
-                and e.get("timestamp") == target_ts
-            ):
-                e["consumed_at"] = consumed_at
-        log_file = store._get_cognition_log_file(trace_id)
-        log_file.write_text(json.dumps(log, indent=2, ensure_ascii=False), encoding="utf-8")
+        reflections_by_trace.setdefault(trace_id, []).append(event)
+    consumed_count = 0
+    for trace_id, trace_reflections in reflections_by_trace.items():
+        consumed_count += await store.mark_reflections_consumed(
+            trace_id,
+            trace_reflections,
+            consumed_at,
+        )
 
     reasoning = plan.get("reasoning", "")
-    return len(reflections), updated_paths, reasoning
+    return consumed_count, updated_paths, reasoning
+
+
+def _validate_dream_scope(
+    memory_config: MemoryConfig,
+    dream_scope: DreamScope,
+) -> None:
+    expected_identity = compute_memory_identity(memory_config)
+    if dream_scope.memory_identity != expected_identity:
+        raise DreamScopeError(
+            "DreamScope.memory_identity does not match the active MemoryConfig"
+        )
+
+
+def _prepare_memory_updates(
+    memory_config: MemoryConfig,
+    plan: Dict[str, Any],
+) -> List[Tuple[str, Path, str]]:
+    updates = plan.get("updates")
+    if not isinstance(updates, list):
+        raise DreamPlanError("updates must be a list")
+    prepared: List[Tuple[str, Path, str]] = []
+    seen: set[str] = set()
+    for index, update in enumerate(updates):
+        if not isinstance(update, dict):
+            raise DreamPlanError(f"updates[{index}] must be an object")
+        rel_path = update.get("path")
+        new_content = update.get("new_content")
+        if not isinstance(rel_path, str) or not isinstance(new_content, str):
+            raise DreamPlanError(
+                f"updates[{index}] requires string path and new_content"
+            )
+        normalized = normalize_memory_relative_path(rel_path)
+        if normalized != rel_path:
+            raise DreamPlanError(f"path is not normalized: {rel_path!r}")
+        if normalized in seen:
+            raise DreamPlanError(f"duplicate update path: {normalized}")
+        seen.add(normalized)
+        prepared.append(
+            (normalized, resolve_memory_path(memory_config, normalized), new_content)
+        )
+    return prepared
+
+
+def _atomic_write_text(target: Path, content: str) -> None:
+    """在同目录写临时文件、fsync，然后原子替换目标。"""
+    tmp_path: Optional[Path] = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=str(target.parent),
+            prefix=f".{target.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as tmp:
+            tmp.write(content)
+            tmp.flush()
+            os.fsync(tmp.fileno())
+            tmp_path = Path(tmp.name)
+        os.replace(tmp_path, target)
+        tmp_path = None
+        try:
+            directory_fd = os.open(str(target.parent), os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        except OSError:
+            # 某些文件系统不支持对目录 fsync；文件本身仍已 fsync + replace。
+            logger.debug(f"[Dream] 目录 fsync 不可用: {target.parent}")
+    finally:
+        if tmp_path is not None:
+            try:
+                tmp_path.unlink(missing_ok=True)
+            except OSError:
+                logger.warning(f"[Dream] 清理临时文件失败: {tmp_path}")
+
+
+def _stage_text(target: Path, content: str) -> Path:
+    """Durably stage one replacement next to its target without publishing it."""
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        mode="w",
+        encoding="utf-8",
+        dir=str(target.parent),
+        prefix=f".{target.name}.",
+        suffix=".dream-stage",
+        delete=False,
+    ) as staged:
+        staged.write(content)
+        staged.flush()
+        os.fsync(staged.fileno())
+        return Path(staged.name)
+
+
+def _transactional_write_memory_updates(
+    memory_config: MemoryConfig,
+    prepared_updates: List[Tuple[str, Path, str]],
+) -> List[str]:
+    """Publish a multi-file Dream plan as one recoverable in-process unit.
+
+    Filesystems do not provide a native multi-file rename transaction.  We
+    therefore stage every new value first and keep every old value until all
+    replacements succeed.  A later replacement failure rolls already-published
+    files back before the exception escapes, so reflections remain retryable
+    without exposing a partially committed plan.
+    """
+    staged_updates: List[Tuple[str, Path, str, Path]] = []
+    old_values: Dict[Path, Optional[str]] = {}
+    committed: List[Path] = []
+    try:
+        for rel_path, target, new_content in prepared_updates:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            checked_target = resolve_memory_path(memory_config, rel_path)
+            if checked_target != target:
+                raise MemoryPathError(
+                    f"memory target changed before staging: {rel_path}"
+                )
+            old_values[target] = (
+                target.read_text(encoding="utf-8") if target.exists() else None
+            )
+            staged_updates.append(
+                (rel_path, target, new_content, _stage_text(target, new_content))
+            )
+
+        for rel_path, target, new_content, staged_path in staged_updates:
+            checked_target = resolve_memory_path(memory_config, rel_path)
+            if checked_target != target:
+                raise MemoryPathError(
+                    f"memory target changed before commit: {rel_path}"
+                )
+            os.replace(staged_path, target)
+            committed.append(target)
+            try:
+                directory_fd = os.open(str(target.parent), os.O_RDONLY)
+                try:
+                    os.fsync(directory_fd)
+                finally:
+                    os.close(directory_fd)
+            except OSError:
+                logger.debug("[Dream] 目录 fsync 不可用: %s", target.parent)
+            logger.info(
+                "[Dream] 已更新记忆文件: %s (%d chars)",
+                rel_path,
+                len(new_content),
+            )
+        return [rel_path for rel_path, *_ in staged_updates]
+    except Exception as commit_error:
+        rollback_errors: List[str] = []
+        for target in reversed(committed):
+            try:
+                old_content = old_values[target]
+                if old_content is None:
+                    target.unlink(missing_ok=True)
+                else:
+                    _atomic_write_text(target, old_content)
+            except Exception as rollback_error:  # pragma: no cover - fatal FS failure
+                rollback_errors.append(f"{target}: {rollback_error}")
+        if rollback_errors:
+            raise RuntimeError(
+                "Dream update failed and rollback was incomplete: "
+                + "; ".join(rollback_errors)
+            ) from commit_error
+        raise
+    finally:
+        for _rel_path, _target, _content, staged_path in staged_updates:
+            try:
+                staged_path.unlink(missing_ok=True)
+            except OSError:
+                logger.warning("[Dream] 清理 staged 文件失败: %s", staged_path)
 
 
 def _build_dream_input(
@@ -342,15 +551,14 @@ async def run_dream(
     store: FileSystemTraceStore,
     llm_call: LLMCall,
     memory_config: MemoryConfig,
-    trace_filter: Optional[Callable[[Trace], bool]] = None,
+    dream_scope: DreamScope,
     reflect_model: str = "gpt-4o-mini",
     dream_model: str = "gpt-4o",
 ) -> DreamReport:
     """执行完整的 dream 流程：per_trace_reflect → cross_trace_integrate。
 
     Args:
-        trace_filter: 筛选需要反思的 trace（例如按 agent_type 或 owner）；
-                      None 表示扫描所有 trace
+        dream_scope: 必填的 uid + agent_type + memory_identity 精确边界
         reflect_model: per-trace 反思用的模型（轻量模型即可）
         dream_model:   跨 trace 整合用的模型（需要更强推理能力）
     """
@@ -359,18 +567,23 @@ async def run_dream(
     if not memory_config.base_path:
         logger.warning("[Dream] memory_config.base_path 未配置，跳过")
         return report
+    _validate_dream_scope(memory_config, dream_scope)
 
     # Phase 1: per-trace reflect
     all_traces = await store.list_traces(limit=1000)
-    if trace_filter:
-        all_traces = [t for t in all_traces if trace_filter(t)]
+    all_traces = [t for t in all_traces if dream_scope.matches(t)]
 
     for t in all_traces:
         if (t.reflected_at_sequence or 0) >= t.last_sequence:
             continue
         try:
             summary = await per_trace_reflect(
-                store, llm_call, t.trace_id, memory_config, model=reflect_model,
+                store,
+                llm_call,
+                t.trace_id,
+                memory_config,
+                dream_scope,
+                model=reflect_model,
             )
             if summary:
                 report.per_trace_summaries[t.trace_id] = summary
@@ -382,7 +595,7 @@ async def run_dream(
     try:
         consumed, updated, reasoning = await cross_trace_integrate(
             store, llm_call, memory_config,
-            trace_filter=trace_filter, model=dream_model,
+            dream_scope=dream_scope, model=dream_model,
         )
         report.consumed_reflection_count = consumed
         report.updated_files = updated

cyber_agent/core/memory.py

@@ -12,14 +12,24 @@ 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, field
-from pathlib import Path
+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 第五节）"""
@@ -42,6 +52,116 @@ class MemoryConfig:
     # 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 解析磁盘上的记忆文件。
 
@@ -52,17 +172,23 @@ def load_memory_files(config: MemoryConfig) -> List[Tuple[str, str, str]]:
     if not config.base_path or not config.files:
         return []
 
-    base = Path(config.base_path)
+    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 / key)
+        pattern = str(base / safe_pattern)
         matched_paths = sorted(_glob.glob(pattern, recursive=True))
 
         if not matched_paths:
@@ -71,12 +197,27 @@ def load_memory_files(config: MemoryConfig) -> List[Tuple[str, str, str]]:
             continue
 
         for fs_path in matched_paths:
-            rel = str(Path(fs_path).relative_to(base))
+            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 = Path(fs_path).read_text(encoding="utf-8")
+                content = resolved.read_text(encoding="utf-8")
             except Exception as e:
                 logger.warning(f"[Memory] 读取失败 {fs_path}: {e}")
                 continue

cyber_agent/tools/builtin/memory.py

@@ -15,6 +15,8 @@ from __future__ import annotations
 import logging
 from typing import Optional
 
+from cyber_agent.core.dream import DreamScope, run_dream
+from cyber_agent.core.memory import MemoryConfig
 from cyber_agent.tools import tool, ToolResult, ToolContext
 
 logger = logging.getLogger(__name__)
@@ -46,12 +48,20 @@ async def dream(
             error="runner not in context",
         )
 
-    memory_config = getattr(runner, "_current_memory_config", None)
-    if memory_config is None:
+    memory_config = context.get("memory_config") if context else None
+    if not isinstance(memory_config, MemoryConfig):
         return ToolResult(
             title="❌ dream 不可用",
-            output="当前 Agent 未配置 MemoryConfig，不是 memory-bearing Agent",
-            error="memory not configured",
+            output="当前工具调用未注入 MemoryConfig，不是 memory-bearing Agent",
+            error="memory_config not in tool context",
+        )
+
+    dream_scope = context.get("dream_scope") if context else None
+    if not isinstance(dream_scope, DreamScope):
+        return ToolResult(
+            title="❌ dream 不可用",
+            output="当前工具调用未注入 DreamScope，为避免跨身份扫描已拒绝执行",
+            error="dream_scope not in tool context",
         )
 
     if not runner.trace_store or not runner.llm_call:
@@ -61,11 +71,11 @@ async def dream(
             error="runner dependencies missing",
         )
 
-    from cyber_agent.core.dream import run_dream
     report = await run_dream(
         store=runner.trace_store,
         llm_call=runner.llm_call,
         memory_config=memory_config,
+        dream_scope=dream_scope,
         reflect_model=reflect_model or "gpt-4o-mini",
         dream_model=dream_model or "gpt-4o",
     )

tests/test_dream_security.py

@@ -0,0 +1,552 @@
+import asyncio
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from cyber_agent.core.dream import (
+    DreamScope,
+    DreamScopeError,
+    cross_trace_integrate,
+)
+from cyber_agent.core.memory import (
+    MEMORY_IDENTITY_CONTEXT_KEY,
+    MemoryConfig,
+    MemoryPathError,
+    compute_memory_identity,
+    is_declared_memory_path,
+    load_memory_files,
+    resolve_memory_path,
+)
+from cyber_agent.tools.builtin.memory import dream as dream_tool
+from cyber_agent.trace.models import Trace
+
+
+class FakeDreamStore:
+    def __init__(self, base_path: Path, traces):
+        self.base_path = base_path
+        self.traces = {trace.trace_id: trace for trace in traces}
+        self.base_path.mkdir(parents=True, exist_ok=True)
+
+    async def list_traces(self, limit=1000):
+        return list(self.traces.values())[:limit]
+
+    async def get_trace(self, trace_id):
+        return self.traces.get(trace_id)
+
+    async def get_cognition_log(self, trace_id):
+        path = self._get_cognition_log_file(trace_id)
+        if not path.exists():
+            return {"trace_id": trace_id, "events": []}
+        return json.loads(path.read_text(encoding="utf-8"))
+
+    async def mark_reflections_consumed(
+        self, trace_id, reflections, consumed_at
+    ):
+        log = await self.get_cognition_log(trace_id)
+        wanted = {
+            (tuple(item.get("sequence_range") or []), item.get("summary", ""))
+            for item in reflections
+        }
+        changed = 0
+        for event in log.get("events", []):
+            key = (
+                tuple(event.get("sequence_range") or []),
+                event.get("summary", ""),
+            )
+            if (
+                event.get("type") == "reflection"
+                and not event.get("consumed_at")
+                and key in wanted
+            ):
+                event["consumed_at"] = consumed_at
+                changed += 1
+        self._get_cognition_log_file(trace_id).write_text(
+            json.dumps(log),
+            encoding="utf-8",
+        )
+        return changed
+
+    def _get_cognition_log_file(self, trace_id):
+        path = self.base_path / trace_id / "cognition_log.json"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        return path
+
+    def set_log(self, trace_id, events):
+        self._get_cognition_log_file(trace_id).write_text(
+            json.dumps({"trace_id": trace_id, "events": events}),
+            encoding="utf-8",
+        )
+
+
+def make_trace(trace_id, scope, *, uid=None, agent_type=None, identity=None):
+    return Trace(
+        trace_id=trace_id,
+        mode="agent",
+        uid=scope.uid if uid is None else uid,
+        agent_type=scope.agent_type if agent_type is None else agent_type,
+        context={
+            MEMORY_IDENTITY_CONTEXT_KEY: (
+                scope.memory_identity if identity is None else identity
+            )
+        },
+    )
+
+
+def reflection(summary):
+    return {
+        "type": "reflection",
+        "timestamp": f"ts-{summary}",
+        "sequence_range": [1, 2],
+        "summary": summary,
+    }
+
+
+class MemorySecurityTests(unittest.TestCase):
+    def test_identity_uses_real_base_and_file_patterns(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp) / "memory"
+            base.mkdir()
+            alias = Path(tmp) / "memory-alias"
+            alias.symlink_to(base, target_is_directory=True)
+
+            first = MemoryConfig(
+                base_path=str(base),
+                files={"journals/**/*.md": "journals", "identity.md": "self"},
+                dream_prompt="first prompt",
+            )
+            second = MemoryConfig(
+                base_path=str(alias),
+                files={"identity.md": "different purpose", "journals/**/*.md": ""},
+                dream_prompt="different prompt",
+            )
+            self.assertEqual(
+                compute_memory_identity(first), compute_memory_identity(second)
+            )
+
+            third = MemoryConfig(
+                base_path=str(base),
+                files={"identity.md": "self"},
+            )
+            self.assertNotEqual(
+                compute_memory_identity(first), compute_memory_identity(third)
+            )
+
+    def test_declared_glob_does_not_let_star_cross_directories(self):
+        config = MemoryConfig(
+            base_path="/tmp/not-used",
+            files={"relationships/*.md": "people", "journals/**/*.md": "logs"},
+        )
+        self.assertTrue(is_declared_memory_path(config, "relationships/a.md"))
+        self.assertFalse(
+            is_declared_memory_path(config, "relationships/private/a.md")
+        )
+        self.assertTrue(is_declared_memory_path(config, "journals/2026/07/a.md"))
+
+    def test_resolution_rejects_non_normal_paths_and_symlink_escape(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            base = root / "memory"
+            outside = root / "outside"
+            base.mkdir()
+            outside.mkdir()
+            (base / "link").symlink_to(outside, target_is_directory=True)
+            config = MemoryConfig(base_path=str(base), files={"**/*.md": "all"})
+
+            for unsafe in (
+                "../outside/escape.md",
+                "./identity.md",
+                "/tmp/escape.md",
+                "nested//identity.md",
+                "nested\\identity.md",
+                "link/escape.md",
+            ):
+                with self.subTest(path=unsafe):
+                    with self.assertRaises(MemoryPathError):
+                        resolve_memory_path(config, unsafe)
+
+    def test_loader_does_not_follow_symlink_outside_base(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            base = root / "memory"
+            outside = root / "outside"
+            base.mkdir()
+            outside.mkdir()
+            (outside / "secret.md").write_text("secret", encoding="utf-8")
+            (base / "safe.md").write_text("safe", encoding="utf-8")
+            (base / "linked.md").symlink_to(outside / "secret.md")
+            config = MemoryConfig(base_path=str(base), files={"*.md": "root"})
+
+            loaded = load_memory_files(config)
+            self.assertEqual(loaded, [("safe.md", "root", "safe")])
+
+
+class DreamIsolationTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+        self.memory_root = self.root / "memory"
+        self.memory_root.mkdir()
+        self.config = MemoryConfig(
+            base_path=str(self.memory_root),
+            files={"identity.md": "self", "journals/**/*.md": "logs"},
+        )
+        self.scope = DreamScope(
+            uid="user-1",
+            agent_type="writer",
+            memory_identity=compute_memory_identity(self.config),
+        )
+
+    async def asyncTearDown(self):
+        self.tmp.cleanup()
+
+    async def test_filters_uid_agent_type_and_memory_identity_together(self):
+        matching = make_trace("matching", self.scope)
+        wrong_uid = make_trace("wrong-uid", self.scope, uid="user-2")
+        wrong_agent = make_trace("wrong-agent", self.scope, agent_type="researcher")
+        wrong_memory = make_trace("wrong-memory", self.scope, identity="memory-v1:other")
+        store = FakeDreamStore(
+            self.root / "traces",
+            [matching, wrong_uid, wrong_agent, wrong_memory],
+        )
+        for trace in (matching, wrong_uid, wrong_agent, wrong_memory):
+            store.set_log(trace.trace_id, [reflection(trace.trace_id)])
+
+        calls = []
+
+        async def llm_call(**kwargs):
+            calls.append(kwargs)
+            return {"content": '{"updates": [], "reasoning": "merged"}'}
+
+        consumed, updated, reasoning = await cross_trace_integrate(
+            store, llm_call, self.config, self.scope
+        )
+
+        self.assertEqual((consumed, updated, reasoning), (1, [], "merged"))
+        prompt = calls[0]["messages"][1]["content"]
+        self.assertIn("matching", prompt)
+        self.assertNotIn("wrong-uid", prompt)
+        self.assertNotIn("wrong-agent", prompt)
+        self.assertNotIn("wrong-memory", prompt)
+        self.assertTrue(
+            (await store.get_cognition_log("matching"))["events"][0].get(
+                "consumed_at"
+            )
+        )
+        for trace_id in ("wrong-uid", "wrong-agent", "wrong-memory"):
+            self.assertNotIn(
+                "consumed_at",
+                (await store.get_cognition_log(trace_id))["events"][0],
+            )
+
+    async def test_concurrent_dreams_keep_memory_configs_isolated(self):
+        other_root = self.root / "other-memory"
+        other_root.mkdir()
+        other_config = MemoryConfig(
+            base_path=str(other_root),
+            files={"identity.md": "self"},
+        )
+        other_scope = DreamScope(
+            uid=self.scope.uid,
+            agent_type=self.scope.agent_type,
+            memory_identity=compute_memory_identity(other_config),
+        )
+        first_trace = make_trace("first", self.scope)
+        second_trace = make_trace("second", other_scope)
+        store = FakeDreamStore(
+            self.root / "traces", [first_trace, second_trace]
+        )
+        store.set_log(first_trace.trace_id, [reflection("first-summary")])
+        store.set_log(second_trace.trace_id, [reflection("second-summary")])
+
+        async def first_llm(**kwargs):
+            await asyncio.sleep(0)
+            self.assertIn("first-summary", kwargs["messages"][1]["content"])
+            self.assertNotIn("second-summary", kwargs["messages"][1]["content"])
+            return {
+                "content": json.dumps(
+                    {
+                        "updates": [
+                            {"path": "identity.md", "new_content": "first"}
+                        ],
+                        "reasoning": "first",
+                    }
+                )
+            }
+
+        async def second_llm(**kwargs):
+            await asyncio.sleep(0)
+            self.assertIn("second-summary", kwargs["messages"][1]["content"])
+            self.assertNotIn("first-summary", kwargs["messages"][1]["content"])
+            return {
+                "content": json.dumps(
+                    {
+                        "updates": [
+                            {"path": "identity.md", "new_content": "second"}
+                        ],
+                        "reasoning": "second",
+                    }
+                )
+            }
+
+        first_result, second_result = await asyncio.gather(
+            cross_trace_integrate(
+                store, first_llm, self.config, self.scope
+            ),
+            cross_trace_integrate(
+                store, second_llm, other_config, other_scope
+            ),
+        )
+
+        self.assertEqual(first_result[0], 1)
+        self.assertEqual(second_result[0], 1)
+        self.assertEqual(
+            (self.memory_root / "identity.md").read_text(encoding="utf-8"),
+            "first",
+        )
+        self.assertEqual(
+            (other_root / "identity.md").read_text(encoding="utf-8"),
+            "second",
+        )
+
+    async def test_scope_identity_must_match_memory_config(self):
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        store.set_log(trace.trace_id, [reflection("summary")])
+        wrong_scope = DreamScope(
+            uid=self.scope.uid,
+            agent_type=self.scope.agent_type,
+            memory_identity="memory-v1:wrong",
+        )
+
+        async def llm_call(**kwargs):
+            self.fail("LLM must not run for a mismatched scope")
+
+        with self.assertRaises(DreamScopeError):
+            await cross_trace_integrate(
+                store, llm_call, self.config, wrong_scope
+            )
+
+    async def test_allowed_updates_are_written_then_reflection_is_consumed(self):
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        store.set_log(trace.trace_id, [reflection("summary")])
+
+        async def llm_call(**kwargs):
+            return {
+                "content": json.dumps(
+                    {
+                        "updates": [
+                            {"path": "identity.md", "new_content": "new identity"},
+                            {
+                                "path": "journals/2026/entry.md",
+                                "new_content": "new journal",
+                            },
+                        ],
+                        "reasoning": "useful",
+                    }
+                )
+            }
+
+        result = await cross_trace_integrate(
+            store, llm_call, self.config, self.scope
+        )
+
+        self.assertEqual(
+            result, (1, ["identity.md", "journals/2026/entry.md"], "useful")
+        )
+        self.assertEqual(
+            (self.memory_root / "identity.md").read_text(encoding="utf-8"),
+            "new identity",
+        )
+        self.assertEqual(
+            (self.memory_root / "journals/2026/entry.md").read_text(
+                encoding="utf-8"
+            ),
+            "new journal",
+        )
+        self.assertTrue(
+            (await store.get_cognition_log(trace.trace_id))["events"][0].get(
+                "consumed_at"
+            )
+        )
+        self.assertEqual(list(self.memory_root.rglob("*.tmp")), [])
+
+    async def test_invalid_update_rejects_entire_plan_without_consuming(self):
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        unsafe_paths = [
+            "../memory-evil/escape.md",
+            "/tmp/escape.md",
+            "not-declared.txt",
+            "journals/../escape.md",
+            "journals\\escape.md",
+        ]
+
+        for index, unsafe_path in enumerate(unsafe_paths):
+            with self.subTest(path=unsafe_path):
+                event = reflection(f"summary-{index}")
+                store.set_log(trace.trace_id, [event])
+
+                async def llm_call(**kwargs):
+                    return {
+                        "content": json.dumps(
+                            {
+                                "updates": [
+                                    {
+                                        "path": unsafe_path,
+                                        "new_content": "must not write",
+                                    }
+                                ],
+                                "reasoning": "bad",
+                            }
+                        )
+                    }
+
+                result = await cross_trace_integrate(
+                    store, llm_call, self.config, self.scope
+                )
+                self.assertEqual(result, (0, [], ""))
+                self.assertNotIn(
+                    "consumed_at",
+                    (await store.get_cognition_log(trace.trace_id))["events"][0],
+                )
+
+        self.assertFalse((self.root / "memory-evil" / "escape.md").exists())
+
+    async def test_symlink_escape_is_rejected_without_consuming(self):
+        outside = self.root / "outside"
+        outside.mkdir()
+        (self.memory_root / "journals").symlink_to(
+            outside, target_is_directory=True
+        )
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        store.set_log(trace.trace_id, [reflection("summary")])
+
+        async def llm_call(**kwargs):
+            return {
+                "content": json.dumps(
+                    {
+                        "updates": [
+                            {
+                                "path": "journals/escape.md",
+                                "new_content": "must not escape",
+                            }
+                        ],
+                        "reasoning": "bad",
+                    }
+                )
+            }
+
+        result = await cross_trace_integrate(
+            store, llm_call, self.config, self.scope
+        )
+        self.assertEqual(result, (0, [], ""))
+        self.assertFalse((outside / "escape.md").exists())
+        self.assertNotIn(
+            "consumed_at",
+            (await store.get_cognition_log(trace.trace_id))["events"][0],
+        )
+
+    async def test_failed_atomic_replace_preserves_old_file_and_does_not_consume(self):
+        target = self.memory_root / "identity.md"
+        target.write_text("old identity", encoding="utf-8")
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        store.set_log(trace.trace_id, [reflection("summary")])
+
+        async def llm_call(**kwargs):
+            return {
+                "content": json.dumps(
+                    {
+                        "updates": [
+                            {"path": "identity.md", "new_content": "new identity"}
+                        ],
+                        "reasoning": "useful",
+                    }
+                )
+            }
+
+        with patch("cyber_agent.core.dream.os.replace", side_effect=OSError("boom")):
+            with self.assertRaises(OSError):
+                await cross_trace_integrate(
+                    store, llm_call, self.config, self.scope
+                )
+
+        self.assertEqual(target.read_text(encoding="utf-8"), "old identity")
+        self.assertNotIn(
+            "consumed_at",
+            (await store.get_cognition_log(trace.trace_id))["events"][0],
+        )
+        self.assertEqual(list(self.memory_root.glob(".identity.md.*.tmp")), [])
+
+    async def test_multi_file_failure_rolls_back_every_published_update(self):
+        identity = self.memory_root / "identity.md"
+        journal = self.memory_root / "journals" / "today.md"
+        journal.parent.mkdir()
+        identity.write_text("old identity", encoding="utf-8")
+        journal.write_text("old journal", encoding="utf-8")
+        trace = make_trace("trace", self.scope)
+        store = FakeDreamStore(self.root / "traces", [trace])
+        store.set_log(trace.trace_id, [reflection("summary")])
+
+        async def llm_call(**kwargs):
+            return {
+                "content": json.dumps({
+                    "updates": [
+                        {"path": "identity.md", "new_content": "new identity"},
+                        {"path": "journals/today.md", "new_content": "new journal"},
+                    ],
+                    "reasoning": "useful",
+                })
+            }
+
+        real_replace = os.replace
+        replace_calls = 0
+
+        def fail_second_target(source, destination):
+            nonlocal replace_calls
+            replace_calls += 1
+            if replace_calls == 2:
+                raise OSError("second replacement failed")
+            return real_replace(source, destination)
+
+        with patch(
+            "cyber_agent.core.dream.os.replace",
+            side_effect=fail_second_target,
+        ):
+            with self.assertRaisesRegex(OSError, "second replacement failed"):
+                await cross_trace_integrate(
+                    store, llm_call, self.config, self.scope
+                )
+
+        self.assertEqual("old identity", identity.read_text(encoding="utf-8"))
+        self.assertEqual("old journal", journal.read_text(encoding="utf-8"))
+        self.assertNotIn(
+            "consumed_at",
+            (await store.get_cognition_log(trace.trace_id))["events"][0],
+        )
+        self.assertEqual(
+            [],
+            list(self.memory_root.rglob("*.dream-stage")),
+        )
+
+    async def test_builtin_dream_requires_memory_config_and_scope_in_context(self):
+        class Runner:
+            trace_store = object()
+            llm_call = object()
+
+        missing = await dream_tool(context={"runner": Runner()})
+        self.assertEqual(missing.error, "memory_config not in tool context")
+
+        missing_scope = await dream_tool(
+            context={"runner": Runner(), "memory_config": self.config}
+        )
+        self.assertEqual(missing_scope.error, "dream_scope not in tool context")
+
+
+if __name__ == "__main__":
+    unittest.main()