✨ feat: WP-04 记忆与知识——候选版本闭环、检索管线与大周天（ADR-011/018/023）

- 迁移 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 正/缺失/过期、检索隔离与召回证据、删除传播与墓碑原文清除、大周天端到端

backend/app/memories/__init__.py

@@ -0,0 +1 @@
+"""记忆域：候选、版本、纠正、删除与检索。"""

backend/app/memories/big_cycle.py

@@ -0,0 +1,163 @@
+"""大周天执行器：回顾→对照→提议→提交的最小实现（ADR-018/023）。
+
+评估调用经模型网关（真实流）；提议产物为 persona_candidate（pending），
+经既有审核机制确认后才影响道体——候选先于写入不变。
+"""
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.gateway.adapters import AdapterRequest, GatewayError, stream_chat
+from app.models import (
+    AuditEvent,
+    BigCycle,
+    Character,
+    CharacterVersion,
+    Message,
+    OutboxEvent,
+    ProviderEndpoint,
+)
+
+EVALUATION_SYSTEM_PROMPT = (
+    "你是角色演化评估器。根据近期对话、当前道体基线与内驱力，"
+    "提出一至三条渐进式人格或关系演化建议。只输出建议正文，简洁分条。"
+)
+
+
+async def _select_endpoint(db: AsyncSession) -> ProviderEndpoint | None:
+    return (
+        await db.execute(
+            select(ProviderEndpoint)
+            .where(
+                ProviderEndpoint.enabled.is_(True),
+                ProviderEndpoint.logical_model_id == "default-chat",
+            )
+            .order_by(ProviderEndpoint.priority.asc())
+            .limit(1)
+        )
+    ).scalar_one_or_none()
+
+
+async def execute_big_cycle(
+    factory: async_sessionmaker[AsyncSession],
+    cycle_id: str,
+    transport: Any | None = None,
+) -> None:
+    """四阶段：回顾（拉取消息）→ 对照（道体/内驱力）→ 提议（模型）→ 提交（候选落库）。"""
+    async with factory() as db:
+        cycle = await db.get(BigCycle, cycle_id)
+        if cycle is None:
+            return
+        character = await db.get(Character, cycle.character_id)
+        assert character is not None
+        try:
+            cycle.status = "running"
+            await db.commit()
+
+            # 阶段一：回顾——同域近期消息（对话暂未绑定角色维度，按个人域取窗口）
+            from app.models import Conversation
+
+            messages = list(
+                (
+                    await db.execute(
+                        select(Message)
+                        .where(
+                            Message.conversation_id.in_(
+                                select(Conversation.id).where(
+                                    Conversation.personal_domain_id
+                                    == character.personal_domain_id
+                                )
+                            ),
+                            Message.role.in_(["user", "assistant"]),
+                        )
+                        .order_by(Message.created_at.desc())
+                        .limit(20)
+                    )
+                ).scalars().all()
+            )
+            def _text_of(m: Any) -> str:
+                joined = "".join(
+                    str(b.get("text", "")) for b in m.content if isinstance(b, dict)
+                )
+                return joined[:120]
+
+            recent = "\n".join(f"{m.role}: {_text_of(m)}" for m in reversed(messages))
+            cycle.phase_memory = {**cycle.phase_memory, "recall": {"messages": len(messages)}}
+            await db.commit()
+
+            # 阶段二：对照——道体与内驱力快照
+            cycle.phase_memory = {
+                **cycle.phase_memory,
+                "evaluate": {
+                    "identity_baseline": character.identity_baseline,
+                    "drives": character.drives,
+                },
+            }
+            await db.commit()
+
+            # 阶段三：提议——模型评估（真实流）
+            endpoint = await _select_endpoint(db)
+            if endpoint is None:
+                cycle.status = "failed"
+                cycle.completed_at = datetime.now(UTC)
+                cycle.phase_memory = {**cycle.phase_memory, "error": "no_endpoint_available"}
+                await db.commit()
+                return
+            prompt = [
+                {"role": "system", "content": EVALUATION_SYSTEM_PROMPT},
+                {
+                    "role": "user",
+                    "content": (
+                        f"角色：{character.name}\n道体基线：{character.identity_baseline}\n"
+                        f"内驱力：{character.drives}\n\n近期对话：\n{recent}\n\n"
+                        "请提出演化建议。"
+                    ),
+                },
+            ]
+            parts: list[str] = []
+            async for chunk in stream_chat(
+                endpoint.base_url, endpoint.model_id, endpoint.api_key_env,
+                AdapterRequest(messages=prompt, model=endpoint.model_id), transport=transport,
+            ):
+                if chunk.kind == "delta" and chunk.text:
+                    parts.append(chunk.text)
+            proposal = "".join(parts).strip()
+
+            # 阶段四：提交——persona_candidate（pending，等待人工审核）
+            version = CharacterVersion(
+                character_id=character.id,
+                kind="persona_candidate",
+                status="pending",
+                payload={"proposal": proposal, "cycle_id": cycle.id, "source": "big_cycle"},
+            )
+            db.add(version)
+            await db.flush()
+            cycle.produced_version_ids = [version.id]
+            cycle.status = "completed"
+            cycle.completed_at = datetime.now(UTC)
+            db.add(
+                AuditEvent(
+                    actor_type="system", actor_id="big_cycle", action="big_cycle.completed",
+                    aggregate_type="character", aggregate_id=character.id,
+                    payload={"cycle_id": cycle.id, "version_id": version.id},
+                )
+            )
+            db.add(
+                OutboxEvent(
+                    aggregate_type="character", aggregate_id=character.id,
+                    event_type="character.persona_candidate_submitted",
+                    payload={"cycle_id": cycle.id, "version_id": version.id},
+                )
+            )
+            await db.commit()
+        except GatewayError as exc:
+            cycle.status = "failed"
+            cycle.completed_at = datetime.now(UTC)
+            cycle.phase_memory = {**cycle.phase_memory, "error": str(exc)[:300]}
+            await db.commit()
+        except Exception:
+            await db.rollback()
+            raise

backend/app/memories/routes.py

@@ -0,0 +1,199 @@
+"""记忆域路由（WP-04）。"""
+
+from typing import Annotated, Any
+
+from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request, status
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.auth.dependencies import get_current_user, require_state_change
+from app.characters.service import require_character
+from app.conversations.routes import _domain_of
+from app.db.session import DbSession
+from app.memories import service
+from app.models import Memory, MemoryCandidate, MemoryVersion, User
+
+router = APIRouter(prefix="/v1/memories", tags=["memories"])
+cycle_router = APIRouter(prefix="/v1/characters", tags=["big-cycle"])
+
+
+class CandidateIn(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    character_id: str | None = None
+    kind: str
+    sensitivity: str = "none"
+    realm: str = "reality"
+    payload: dict[str, Any]
+    source: str = "manual"
+    source_ref: dict[str, Any] = Field(default_factory=dict)
+
+
+def _candidate_out(c: MemoryCandidate, memory: Memory | None = None) -> dict[str, Any]:
+    return {
+        "id": c.id,
+        "kind": c.kind,
+        "sensitivity": c.sensitivity,
+        "realm": c.realm,
+        "payload": c.payload,
+        "source": c.source,
+        "status": c.status,
+        "created_memory_id": memory.id if memory else None,
+        "created_at": c.created_at.isoformat(),
+    }
+
+
+def _memory_out(m: Memory, v: MemoryVersion | None) -> dict[str, Any]:
+    return {
+        "id": m.id,
+        "character_id": m.character_id,
+        "kind": m.kind,
+        "sensitivity": m.sensitivity,
+        "realm": m.realm,
+        "status": m.status,
+        "payload": v.payload if v else None,
+        "version": v.version if v else None,
+        "resource_version": m.resource_version,
+        "expires_at": m.expires_at.isoformat() if m.expires_at else None,
+        "created_at": m.created_at.isoformat(),
+    }
+
+
+@router.post("/candidates", status_code=201)
+async def submit_candidate(
+    db: DbSession,
+    user: Annotated[User, Depends(require_state_change)],
+    request: Request,
+    body: CandidateIn,
+) -> dict[str, Any]:
+    domain = await _domain_of(db, user)
+    candidate, memory = await service.submit_candidate(
+        db, domain.id, body.model_dump(), getattr(request.state, "trace_id", None)
+    )
+    return _candidate_out(candidate, memory)
+
+
+@router.get("/candidates")
+async def list_candidates(
+    db: DbSession,
+    user: Annotated[User, Depends(get_current_user)],
+    status_filter: str = "pending",
+) -> dict[str, Any]:
+    from sqlalchemy import select
+
+    domain = await _domain_of(db, user)
+    stmt = (
+        select(MemoryCandidate)
+        .where(
+            MemoryCandidate.owner_personal_domain_id == domain.id,
+            MemoryCandidate.status == status_filter,
+        )
+        .order_by(MemoryCandidate.created_at.asc())
+    )
+    rows = list((await db.scalars(stmt)).all())
+    return {"data": [_candidate_out(c) for c in rows]}
+
+
+@router.post("/candidates/{candidate_id}/decisions", status_code=201)
+async def decide_candidate(
+    candidate_id: str,
+    db: DbSession,
+    user: Annotated[User, Depends(require_state_change)],
+    request: Request,
+    body: dict[str, bool],
+) -> dict[str, Any]:
+    domain = await _domain_of(db, user)
+    candidate, memory = await service.decide_candidate(
+        db, domain.id, candidate_id, bool(body.get("approve")),
+        getattr(request.state, "trace_id", None),
+    )
+    return _candidate_out(candidate, memory)
+
+
+@router.get("")
+async def search_memories(
+    db: DbSession,
+    user: Annotated[User, Depends(get_current_user)],
+    q: str | None = None,
+    character_id: str | None = None,
+    limit: int = 50,
+    cursor: str | None = None,
+) -> dict[str, Any]:
+    if not 1 <= limit <= 200:
+        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "limit 取值范围 1-200")
+    domain = await _domain_of(db, user)
+    page, next_cursor, has_more = await service.search_memories(
+        db, domain.id, character_id, q, limit, cursor
+    )
+    # 召回证据：进入结果即附（ContextSnapshot 的响应级最小形态）
+    return {
+        "data": [
+            {**_memory_out(m, v), "recall": {"matched": bool(q), "channel": "text"}}
+            for m, v in page
+        ],
+        "next_cursor": next_cursor,
+        "has_more": has_more,
+    }
+
+
+@router.patch("/{memory_id}")
+async def correct_memory(
+    memory_id: str,
+    db: DbSession,
+    user: Annotated[User, Depends(require_state_change)],
+    request: Request,
+    if_match: Annotated[str | None, Header(alias="If-Match")] = None,
+    body: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    domain = await _domain_of(db, user)
+    memory = await service.correct_memory(
+        db, domain.id, memory_id, if_match, (body or {}).get("payload", {}),
+        getattr(request.state, "trace_id", None),
+    )
+    version = await db.get(MemoryVersion, memory.current_version_id)
+    return _memory_out(memory, version)
+
+
+@router.delete("/{memory_id}", status_code=204)
+async def delete_memory(
+    memory_id: str,
+    db: DbSession,
+    user: Annotated[User, Depends(require_state_change)],
+    request: Request,
+) -> None:
+    domain = await _domain_of(db, user)
+    await service.delete_memory(
+        db, domain.id, memory_id, getattr(request.state, "trace_id", None)
+    )
+
+
+@cycle_router.post("/{character_id}/big-cycles", status_code=201)
+async def trigger_big_cycle(
+    character_id: str,
+    db: DbSession,
+    user: Annotated[User, Depends(require_state_change)],
+    request: Request,
+    background_tasks: BackgroundTasks,
+) -> dict[str, Any]:
+    """手动触发大周天：评估经模型网关，产物为待审核 persona_candidate。"""
+    from app.models import BigCycle
+
+    domain = await _domain_of(db, user)
+    character = await require_character(db, domain.id, character_id)
+    if character.status not in {"active", "resting"}:
+        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "当前状态不可触发大周天")
+    await db.commit()
+    cycle = BigCycle(
+        character_id=character.id, status="pending", trigger={"type": "manual"}
+    )
+    async with db.begin():
+        db.add(cycle)
+    await db.refresh(cycle)
+    from app.memories.big_cycle import execute_big_cycle
+
+    background_tasks.add_task(
+        execute_big_cycle,
+        request.app.state.session_factory,
+        cycle.id,
+        transport=getattr(request.app.state, "gw_transport", None),
+    )
+    return {"id": cycle.id, "character_id": character.id, "status": cycle.status}

backend/app/memories/service.py

@@ -0,0 +1,267 @@
+"""记忆域：候选、审批、不可变版本、纠正、删除与检索（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