ai-companion/app/agent/memory_policy.py

"""记忆写入/删除的策略层。

**为什么需要独立的一层**：交接文档明确要求
"不要仅依赖'请不要误记'的 system prompt"。

模型可以提议动作，**这一层拥有最终决定权**。理由：

  - system prompt 是软的，模型会漂移；策略是硬的，可测试、可回归
  - "我今天吃了香菜"不能被记成"用户不吃香菜"这种稳定偏好——
    这不是措辞问题，是**语义等级**问题，必须在代码里判定
  - 身份未确认时不得碰任何私人记忆，这也不能交给模型自觉

输入：用户原话、工具请求参数、身份确认状态、已有同键记忆。
输出：allow_confirmed / allow_candidate / needs_confirmation / reject。

**本模块不依赖 LLM、不依赖网络、不依赖真实存储**，可以纯单元测试。
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from enum import Enum

from ..memory.base import ConfirmStatus, Memory, MemoryKind


class MemoryDecision(str, Enum):
    """策略层的裁决。"""
    ALLOW_CONFIRMED = "allow_confirmed"      # 用户明确要求 → 写 confirmed
    ALLOW_CANDIDATE = "allow_candidate"      # 只是观察/推断 → 写 candidate
    NEEDS_CONFIRMATION = "needs_confirmation"  # 目标不明确 → 转入确认节点
    REJECT = "reject"                        # 不允许（越权、身份未确认等）


@dataclass
class PolicyResult:
    decision: MemoryDecision
    reason: str
    #: 规范化后的写入内容。策略层可以改写模型给的 kind/key/value。
    kind: MemoryKind = MemoryKind.FACT
    key: str = ""
    value: str = ""
    #: NEEDS_CONFIRMATION 时，候选目标（供确认节点展示）
    candidates: list[Memory] = field(default_factory=list)
    #: REJECT/NEEDS_CONFIRMATION 时给模型看的话（不冒充已执行）
    message: str = ""

    @property
    def confirm_status(self) -> ConfirmStatus:
        return (ConfirmStatus.CONFIRMED
                if self.decision is MemoryDecision.ALLOW_CONFIRMED
                else ConfirmStatus.CANDIDATE)


# ---------------------------------------------------------------- 显式意图

#: 用户明确要求"记住"的说法。命中才允许写 confirmed。
#: 刻意保守：宁可多判成 candidate（不参与召回），也不要误记成稳定事实。
EXPLICIT_REMEMBER_PATTERNS = (
    "记住", "记一下", "记下", "帮我记", "帮我记住",
    "别忘了", "别忘记", "不要忘",
    "以后都", "以后要", "以后别", "今后都",
    "我的偏好是", "我习惯", "我通常", "我总是",
)

#: 明确表示"这是这一次的事"的说法。命中则**降级**为 candidate，
#: 即使句中同时出现"记住"之类的词。
#: 例："记住我今天吃了香菜"——语义是"记下今天这件事"，不是稳定偏好。
#: 注意不要收太宽的词。比如"先"看着像临时，但"我习惯先喝汤"是稳定偏好，
#: 收进来会造成漏判（把偏好降级成候选）。宁可少收。
EPISODIC_MARKERS = (
    "今天", "昨天", "刚才", "刚刚", "这次", "这一回", "暂时", "目前",
    "这会儿", "眼下",
)

#: 主语是"今天/这次"这类时间词时，更可能是事件而非偏好
_TIME_LEAD_RE = re.compile(r"^(今天|昨天|刚才|刚刚|这次|这会|这回|现在|目前)")


def has_explicit_remember_intent(user_text: str) -> bool:
    """用户是否明确要求记住（而不是模型自己觉得值得记）。

    注意：本函数调用下面定义的 `looks_episodic`（Python 晚绑定，没问题）。
    """
    t = (user_text or "").strip()
    if not t:
        return False
    if not any(p in t for p in EXPLICIT_REMEMBER_PATTERNS):
        return False
    # 明确是"这一次"的事 → 不算稳定偏好意图。
    # **必须复用 looks_episodic**：早先这里只检查句首时间词，
    # 而"记住我今天吃了香菜"的时间词在中间，导致漏判（测试逮到的）。
    if looks_episodic(t):
        return False
    return True


def looks_episodic(user_text: str) -> bool:
    """是否是"这一次发生的事"而非稳定事实。"""
    t = (user_text or "").strip()
    if not t:
        return False
    if _TIME_LEAD_RE.search(t):
        return True
    return any(m in t for m in EPISODIC_MARKERS)


# ---------------------------------------------------------------- 越权检查

#: 模型**不得**传入的字段——身份与授权范围必须由服务端注入。
FORBIDDEN_ARG_KEYS = (
    "owner_user_id", "user_id", "owner", "uid",
    "scope", "confirm_status", "trust_level",
    "session_id", "turn_id",
)


def find_forbidden_args(args: dict) -> list[str]:
    """返回模型试图传入的越权字段名。"""
    if not isinstance(args, dict):
        return []
    lower = {str(k).lower() for k in args.keys()}
    return [k for k in FORBIDDEN_ARG_KEYS if k.lower() in lower]


# ---------------------------------------------------------------- 写入裁决

def decide_remember(
    *,
    user_text: str,
    args: dict,
    identity_confirmed: bool,
    model_claims_explicit: bool = False,
) -> PolicyResult:
    """裁决一次 memory_remember 请求。

    参数：
        user_text:         本轮用户原话（策略的主要依据）
        args:              模型给的工具参数（含 kind/key/value）
        identity_confirmed: 当前身份是否已确认
        model_claims_explicit: 模型自称"用户明确要求记住"。
            仅作参考，**不采信**——必须由 user_text 判定。
    """
    # 1. 身份未确认：一律拒绝，不碰任何私人记忆
    if not identity_confirmed:
        return PolicyResult(
            decision=MemoryDecision.REJECT,
            reason="身份未确认",
            message="现在还不能记录私人信息。请先确认身份。",
        )

    # 2. 越权字段：拒绝（不静默忽略，避免模型以为生效了）
    forbidden = find_forbidden_args(args)
    if forbidden:
        return PolicyResult(
            decision=MemoryDecision.REJECT,
            reason=f"模型传入越权字段 {forbidden}",
            message=f"参数不合法：不接受 {', '.join(forbidden)}。",
        )

    value = str(args.get("value") or "").strip()
    if not value:
        return PolicyResult(
            decision=MemoryDecision.REJECT,
            reason="value 为空",
            message="没有可记录的内容。",
        )

    kind = _coerce_kind(args.get("kind"))
    key = str(args.get("key") or "").strip()

    # 3. 判定语义等级。
    #    注意：以 user_text 为准，不采信 model_claims_explicit。
    explicit = has_explicit_remember_intent(user_text)
    episodic = looks_episodic(user_text)

    if explicit and not episodic:
        return PolicyResult(
            decision=MemoryDecision.ALLOW_CONFIRMED,
            reason="用户明确要求记住",
            kind=kind, key=key, value=value,
        )

    # 4. 其余一律 candidate：模型推断、总结、或"这一次"的事件
    reason = "属于一次性事件" if episodic else "非用户明确要求，仅作观察"
    return PolicyResult(
        decision=MemoryDecision.ALLOW_CANDIDATE,
        reason=reason,
        kind=kind, key=key, value=value,
    )


def _coerce_kind(raw) -> MemoryKind:
    """把模型给的 kind 收敛到枚举；不认识的一律当 FACT。"""
    try:
        return MemoryKind(str(raw))
    except (ValueError, TypeError):
        return MemoryKind.FACT


# ---------------------------------------------------------------- 删除裁决

#: 模糊删除的说法——命中则必须先确认，不得猜目标
AMBIGUOUS_FORGET_MARKERS = (
    "所有", "全部", "都删", "清空", "关于我的", "我的资料", "我的信息",
    "这些", "那些", "乱七八糟",
)


def decide_forget(
    *,
    user_text: str,
    matches: list[Memory],
    identity_confirmed: bool,
) -> PolicyResult:
    """裁决一次 memory_forget 请求。

    规则（交接文档 §八）：
      - 精确匹配一条 → 返回待删除摘要，可执行
      - 多条或模糊 → 进入确认节点，**不得猜测删除目标**
      - 删除必须同时匹配 owner（由存储层保证，这里只判数量与措辞）
    """
    if not identity_confirmed:
        return PolicyResult(
            decision=MemoryDecision.REJECT,
            reason="身份未确认",
            message="现在不能操作私人记忆。",
        )

    if not matches:
        return PolicyResult(
            decision=MemoryDecision.REJECT,
            reason="没有匹配的记忆",
            message="没有找到相关的记忆。",
        )

    t = (user_text or "").strip()
    ambiguous_wording = any(m in t for m in AMBIGUOUS_FORGET_MARKERS)

    if len(matches) == 1 and not ambiguous_wording:
        m = matches[0]
        return PolicyResult(
            decision=MemoryDecision.ALLOW_CONFIRMED,
            reason="精确匹配一条",
            kind=m.kind, key=m.key, value=m.value,
            candidates=[m],
            message=f"将删除「{m.text}」。",
        )

    # 多条 / 模糊 → 必须确认
    return PolicyResult(
        decision=MemoryDecision.NEEDS_CONFIRMATION,
        reason=f"匹配到 {len(matches)} 条" + ("，且措辞模糊" if ambiguous_wording else ""),
        candidates=list(matches),
        message=(
            f"找到 {len(matches)} 条相关记忆，需要你确认删哪一条：\n"
            + "\n".join(f"  {i+1}. {m.text}" for i, m in enumerate(matches[:5]))
        ),
    )


__all__ = [
    "MemoryDecision", "PolicyResult",
    "decide_remember", "decide_forget",
    "has_explicit_remember_intent", "looks_episodic",
    "find_forbidden_args",
    "EXPLICIT_REMEMBER_PATTERNS", "EPISODIC_MARKERS", "FORBIDDEN_ARG_KEYS",
]