chenxin
e333dc5ea5
- 迁移 9：memory_candidate（来源/敏感度/作用域/状态检查约束）、memory（resource_version 作 ETag 源、有效期、当前版本指针）、memory_version（(memory_id, version) 唯一、取代链）、big_cycle（cycle_ 聚合根、phase_memory、产物版本引用） - 候选闭环：手动提交；白名单自动批准仅限 auto+fact+低敏感（撤销窗 72h 记录于审计）；敏感/高影响默认人工审核；批准同事务写 Memory+Version+Audit+Outbox（契约 4.4） - 纠正：If-Match 校验 resource_version（缺失/过期 412），创建新版本不原地覆盖（ADR-011） - 删除：立即阻断召回 + 清除全部版本原文（墓碑不存被删原文，契约 4.4） - 检索管线：数据库层域/角色预过滤 → 状态/有效期过滤 → 文本匹配 → 响应级召回证据（语义检索待 embedding 端点） - 大周天最小实现：手动触发 → 后台四阶段（回顾/对照/提议/提交）→ 模型真实评估产出 pending persona_candidate（候选先于写入）；无端点 fail-closed - 测试 60 项全绿（+6）：审批流、白名单三例、ETag 正/缺失/过期、检索隔离与召回证据、删除传播与墓碑原文清除、大周天端到端
268 lines
9.3 KiB
Python
268 lines
9.3 KiB
Python
"""记忆域：候选、审批、不可变版本、纠正、删除与检索（WP-04 / ADR-011）。"""
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import (
|
|
AuditEvent,
|
|
Memory,
|
|
MemoryCandidate,
|
|
MemoryVersion,
|
|
OutboxEvent,
|
|
)
|
|
|
|
# 白名单自动批准：仅自动提取、fact、低敏感（契约 4.2：撤销窗随批准记录）
|
|
AUTO_APPROVE_KINDS = {"fact"}
|
|
AUTO_APPROVE_SENSITIVITY = {"none", "low"}
|
|
|
|
|
|
def _content_hash(payload: dict[str, Any]) -> str:
|
|
canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False)
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
async def require_memory(db: AsyncSession, domain_id: str, memory_id: str) -> Memory:
|
|
memory = await db.get(Memory, memory_id)
|
|
if memory is None or memory.owner_personal_domain_id != domain_id:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "记忆不存在")
|
|
return memory
|
|
|
|
|
|
async def submit_candidate(
|
|
db: AsyncSession,
|
|
domain_id: str,
|
|
body: dict[str, Any],
|
|
trace_id: str | None,
|
|
) -> tuple[MemoryCandidate, Memory | None]:
|
|
"""提交候选；白名单（auto+fact+低敏感）自动批准并直接创建 Memory（撤销窗记录于候选）。"""
|
|
kind = body.get("kind", "")
|
|
sensitivity = body.get("sensitivity", "none")
|
|
payload = body.get("payload", {})
|
|
source = body.get("source", "manual")
|
|
if not isinstance(payload, dict) or not payload:
|
|
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "payload 不能为空")
|
|
from datetime import timedelta
|
|
|
|
window_until = None
|
|
auto_approved = (
|
|
source == "auto" and kind in AUTO_APPROVE_KINDS and sensitivity in AUTO_APPROVE_SENSITIVITY
|
|
)
|
|
if auto_approved:
|
|
# 撤销窗：白名单自动写入后 72 小时内可一键撤销（产品策略可调）
|
|
window_until = datetime.now(UTC) + timedelta(hours=72)
|
|
candidate = MemoryCandidate(
|
|
owner_personal_domain_id=domain_id,
|
|
character_id=body.get("character_id"),
|
|
kind=kind,
|
|
sensitivity=sensitivity,
|
|
realm=body.get("realm", "reality"),
|
|
payload=payload,
|
|
source=source,
|
|
source_ref=body.get("source_ref", {}),
|
|
content_hash=_content_hash(payload),
|
|
status="pending",
|
|
)
|
|
db.add(candidate)
|
|
await db.flush()
|
|
memory = None
|
|
if auto_approved:
|
|
memory = await _approve_into_memory(db, domain_id, candidate, trace_id, window_until)
|
|
else:
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="user", actor_id=domain_id, action="memory.candidate_submitted",
|
|
aggregate_type="memory_candidate", aggregate_id=candidate.id,
|
|
payload={"kind": kind, "sensitivity": sensitivity}, trace_id=trace_id,
|
|
)
|
|
)
|
|
await db.flush()
|
|
return candidate, memory
|
|
|
|
|
|
async def _approve_into_memory(
|
|
db: AsyncSession,
|
|
domain_id: str,
|
|
candidate: MemoryCandidate,
|
|
trace_id: str | None,
|
|
window_until: datetime | None = None,
|
|
) -> Memory:
|
|
"""批准候选：同一事务写 Memory + MemoryVersion(v1) + Audit + Outbox（契约 4.4）。"""
|
|
candidate.status = "approved"
|
|
candidate.decided_at = datetime.now(UTC)
|
|
memory = Memory(
|
|
owner_personal_domain_id=candidate.owner_personal_domain_id,
|
|
character_id=candidate.character_id,
|
|
kind=candidate.kind,
|
|
sensitivity=candidate.sensitivity,
|
|
realm=candidate.realm,
|
|
status="active",
|
|
)
|
|
db.add(memory)
|
|
await db.flush()
|
|
version = MemoryVersion(memory_id=memory.id, version=1, payload=candidate.payload)
|
|
db.add(version)
|
|
await db.flush()
|
|
memory.current_version_id = version.id
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="user", actor_id=domain_id, action="memory.created",
|
|
aggregate_type="memory", aggregate_id=memory.id,
|
|
payload={
|
|
"candidate_id": candidate.id, "auto": window_until is not None,
|
|
"window_until": window_until.isoformat() if window_until else None,
|
|
},
|
|
trace_id=trace_id,
|
|
)
|
|
)
|
|
db.add(
|
|
OutboxEvent(
|
|
aggregate_type="memory", aggregate_id=memory.id,
|
|
event_type="memory.created", payload={"memory_id": memory.id},
|
|
)
|
|
)
|
|
await db.flush()
|
|
return memory
|
|
|
|
|
|
async def decide_candidate(
|
|
db: AsyncSession,
|
|
domain_id: str,
|
|
candidate_id: str,
|
|
approve: bool,
|
|
trace_id: str | None,
|
|
) -> tuple[MemoryCandidate, Memory | None]:
|
|
"""人工审批：敏感与高影响默认走此路径（契约：候选先于长期写入）。"""
|
|
candidate = await db.get(MemoryCandidate, candidate_id)
|
|
if candidate is None or candidate.owner_personal_domain_id != domain_id:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "候选不存在")
|
|
if candidate.status != "pending":
|
|
raise HTTPException(status.HTTP_409_CONFLICT, "候选不在待审状态")
|
|
if not approve:
|
|
candidate.status = "rejected"
|
|
candidate.decided_at = datetime.now(UTC)
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="user", actor_id=domain_id, action="memory.candidate_rejected",
|
|
aggregate_type="memory_candidate", aggregate_id=candidate.id, trace_id=trace_id,
|
|
)
|
|
)
|
|
await db.flush()
|
|
return candidate, None
|
|
memory = await _approve_into_memory(db, domain_id, candidate, trace_id)
|
|
return candidate, memory
|
|
|
|
|
|
async def correct_memory(
|
|
db: AsyncSession,
|
|
domain_id: str,
|
|
memory_id: str,
|
|
if_match: str | None,
|
|
payload: dict[str, Any],
|
|
trace_id: str | None,
|
|
) -> Memory:
|
|
"""纠正：If-Match 校验 resource_version，创建新版本（不原地覆盖）。"""
|
|
memory = await require_memory(db, domain_id, memory_id)
|
|
expected = str(memory.resource_version)
|
|
if if_match is None or if_match.strip('"') != expected:
|
|
raise HTTPException(
|
|
status.HTTP_412_PRECONDITION_FAILED,
|
|
"版本不匹配：请以最新 resource_version 重试",
|
|
)
|
|
last = (
|
|
await db.execute(
|
|
select(MemoryVersion.version)
|
|
.where(MemoryVersion.memory_id == memory.id)
|
|
.order_by(MemoryVersion.version.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one()
|
|
version = MemoryVersion(
|
|
memory_id=memory.id,
|
|
version=last + 1,
|
|
payload=payload,
|
|
supersedes_version_id=memory.current_version_id,
|
|
)
|
|
db.add(version)
|
|
await db.flush()
|
|
memory.current_version_id = version.id
|
|
memory.resource_version += 1
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="user", actor_id=domain_id, action="memory.corrected",
|
|
aggregate_type="memory", aggregate_id=memory.id,
|
|
payload={"new_version": version.version}, trace_id=trace_id,
|
|
)
|
|
)
|
|
await db.flush()
|
|
return memory
|
|
|
|
|
|
async def delete_memory(
|
|
db: AsyncSession, domain_id: str, memory_id: str, trace_id: str | None
|
|
) -> Memory:
|
|
"""删除：立即阻止召回；清除全部版本原文（墓碑不存被删原文，契约 4.4）。"""
|
|
memory = await require_memory(db, domain_id, memory_id)
|
|
memory.status = "deleted"
|
|
versions = (
|
|
await db.execute(select(MemoryVersion).where(MemoryVersion.memory_id == memory.id))
|
|
).scalars().all()
|
|
for v in versions:
|
|
v.payload = {} # 原文清除：墓碑只留元数据
|
|
db.add(
|
|
AuditEvent(
|
|
actor_type="user", actor_id=domain_id, action="memory.deleted",
|
|
aggregate_type="memory", aggregate_id=memory.id, trace_id=trace_id,
|
|
)
|
|
)
|
|
db.add(
|
|
OutboxEvent(
|
|
aggregate_type="memory", aggregate_id=memory.id,
|
|
event_type="memory.deleted", payload={"memory_id": memory.id},
|
|
)
|
|
)
|
|
await db.flush()
|
|
return memory
|
|
|
|
|
|
async def search_memories(
|
|
db: AsyncSession,
|
|
domain_id: str,
|
|
character_id: str | None,
|
|
query: str | None,
|
|
limit: int,
|
|
cursor: str | None,
|
|
) -> tuple[list[tuple[Memory, MemoryVersion | None]], str | None, bool]:
|
|
"""检索管线（最小）：数据库层授权预过滤 → 状态/有效期过滤 → 文本匹配 → 召回证据。
|
|
|
|
语义检索（pgvector）随 embedding 端点接入；当前文本匹配以个人域规模为前提
|
|
（域过滤前置 + ix_memory_domain_status_kind 索引）。
|
|
"""
|
|
stmt = select(Memory, MemoryVersion).outerjoin(
|
|
MemoryVersion, Memory.current_version_id == MemoryVersion.id
|
|
).where(
|
|
Memory.owner_personal_domain_id == domain_id,
|
|
Memory.status == "active",
|
|
)
|
|
if character_id is not None:
|
|
stmt = stmt.where(Memory.character_id == character_id)
|
|
now = datetime.now(UTC)
|
|
stmt = stmt.where(
|
|
(Memory.expires_at.is_(None)) | (Memory.expires_at > now),
|
|
)
|
|
if query:
|
|
stmt = stmt.where(MemoryVersion.payload["text"].astext.ilike(f"%{query}%"))
|
|
if cursor:
|
|
stmt = stmt.where(Memory.id > cursor)
|
|
stmt = stmt.order_by(Memory.id.asc()).limit(limit + 1)
|
|
rows = list((await db.execute(stmt)).all())
|
|
has_more = len(rows) > limit
|
|
page: list[tuple[Memory, MemoryVersion | None]] = [(m, v) for m, v in rows[:limit]]
|
|
next_cursor = page[-1][0].id if has_more and page else None
|
|
return page, next_cursor, has_more
|