From e333dc5ea5e25f85fe688c14f2ffbb8cdd25feea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Fri, 11 Sep 2026 10:54:20 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20WP-04=20=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=E4=B8=8E=E7=9F=A5=E8=AF=86=E2=80=94=E2=80=94=E5=80=99=E9=80=89?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E9=97=AD=E7=8E=AF=E3=80=81=E6=A3=80=E7=B4=A2?= =?UTF-8?q?=E7=AE=A1=E7=BA=BF=E4=B8=8E=E5=A4=A7=E5=91=A8=E5=A4=A9=EF=BC=88?= =?UTF-8?q?ADR-011/018/023=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 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 正/缺失/过期、检索隔离与召回证据、删除传播与墓碑原文清除、大周天端到端 --- ...994e5b4337b_记忆与知识_候选_版本_大周天.py | 120 ++++++++ backend/app/main.py | 4 + backend/app/memories/__init__.py | 1 + backend/app/memories/big_cycle.py | 163 +++++++++++ backend/app/memories/routes.py | 199 +++++++++++++ backend/app/memories/service.py | 267 ++++++++++++++++++ backend/app/models.py | 142 ++++++++++ backend/tests/test_conversations.py | 173 ++++++++++++ 8 files changed, 1069 insertions(+) create mode 100644 backend/alembic/versions/a994e5b4337b_记忆与知识_候选_版本_大周天.py create mode 100644 backend/app/memories/__init__.py create mode 100644 backend/app/memories/big_cycle.py create mode 100644 backend/app/memories/routes.py create mode 100644 backend/app/memories/service.py diff --git a/backend/alembic/versions/a994e5b4337b_记忆与知识_候选_版本_大周天.py b/backend/alembic/versions/a994e5b4337b_记忆与知识_候选_版本_大周天.py new file mode 100644 index 0000000..8c9d674 --- /dev/null +++ b/backend/alembic/versions/a994e5b4337b_记忆与知识_候选_版本_大周天.py @@ -0,0 +1,120 @@ +"""记忆与知识：候选、版本、大周天 + +Revision ID: a994e5b4337b +Revises: 1c1f7edb7cbe +Create Date: 2026-09-11 10:48:52.696975 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = 'a994e5b4337b' +down_revision: str | Sequence[str] | None = '1c1f7edb7cbe' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('big_cycle', + sa.Column('id', sa.String(), nullable=False), + sa.Column('character_id', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('trigger', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('phase_memory', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('produced_version_ids', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('pending', 'running', 'completed', 'failed')", name=op.f('ck_big_cycle_status_values')), + sa.ForeignKeyConstraint(['character_id'], ['character.id'], name=op.f('fk_big_cycle_character'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_big_cycle')) + ) + op.create_index(op.f('ix_big_cycle_character_id'), 'big_cycle', ['character_id'], unique=False) + op.create_table('memory', + sa.Column('id', sa.String(), nullable=False), + sa.Column('owner_personal_domain_id', sa.String(), nullable=False), + sa.Column('character_id', sa.String(), nullable=True), + sa.Column('kind', sa.String(), nullable=False), + sa.Column('sensitivity', sa.String(), nullable=False), + sa.Column('realm', sa.String(), nullable=False), + sa.Column('status', sa.String(), nullable=False), + sa.Column('valid_from', sa.DateTime(timezone=True), nullable=True), + sa.Column('expires_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('pinned', sa.Boolean(), nullable=False), + sa.Column('current_version_id', sa.String(), nullable=True), + sa.Column('resource_version', sa.Integer(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("kind IN ('fact', 'preference', 'event', 'relationship')", name=op.f('ck_memory_candidate_kind_values')), + sa.CheckConstraint("realm IN ('reality', 'world', 'simulation')", name=op.f('ck_memory_candidate_realm_values')), + sa.CheckConstraint("sensitivity IN ('none', 'low', 'medium', 'high')", name=op.f('ck_memory_candidate_sensitivity_values')), + sa.CheckConstraint("status IN ('active', 'superseded', 'archived', 'deleted')", name=op.f('ck_memory_status_values')), + sa.ForeignKeyConstraint(['character_id'], ['character.id'], name=op.f('fk_memory_character'), ondelete='cascade'), + sa.ForeignKeyConstraint(['current_version_id'], ['memory_version.id'], name=op.f('fk_memory_memory_version'), ondelete='set null', use_alter=True), + sa.ForeignKeyConstraint(['owner_personal_domain_id'], ['personal_domain.id'], name=op.f('fk_memory_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_memory')) + ) + op.create_index(op.f('ix_memory_character_id'), 'memory', ['character_id'], unique=False) + op.create_index('ix_memory_domain_status_kind', 'memory', ['owner_personal_domain_id', 'status', 'kind'], unique=False) + op.create_index(op.f('ix_memory_owner_personal_domain_id'), 'memory', ['owner_personal_domain_id'], unique=False) + op.create_table('memory_candidate', + sa.Column('id', sa.String(), nullable=False), + sa.Column('owner_personal_domain_id', sa.String(), nullable=False), + sa.Column('character_id', sa.String(), nullable=True), + sa.Column('kind', sa.String(), nullable=False), + sa.Column('sensitivity', sa.String(), nullable=False), + sa.Column('realm', sa.String(), nullable=False), + sa.Column('payload', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('source', sa.String(), nullable=False), + sa.Column('source_ref', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('content_hash', sa.String(), nullable=True), + sa.Column('status', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('decided_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("kind IN ('fact', 'preference', 'event', 'relationship')", name=op.f('ck_memory_candidate_kind_values')), + sa.CheckConstraint("realm IN ('reality', 'world', 'simulation')", name=op.f('ck_memory_candidate_realm_values')), + sa.CheckConstraint("sensitivity IN ('none', 'low', 'medium', 'high')", name=op.f('ck_memory_candidate_sensitivity_values')), + sa.CheckConstraint("source IN ('manual', 'auto', 'big_cycle', 'canonization')", name=op.f('ck_memory_candidate_source_values')), + sa.CheckConstraint("status IN ('pending', 'approved', 'rejected', 'expired')", name=op.f('ck_memory_candidate_status_values')), + sa.ForeignKeyConstraint(['character_id'], ['character.id'], name=op.f('fk_memory_candidate_character'), ondelete='cascade'), + sa.ForeignKeyConstraint(['owner_personal_domain_id'], ['personal_domain.id'], name=op.f('fk_memory_candidate_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_memory_candidate')) + ) + op.create_index(op.f('ix_memory_candidate_owner_personal_domain_id'), 'memory_candidate', ['owner_personal_domain_id'], unique=False) + op.create_table('memory_version', + sa.Column('id', sa.String(), nullable=False), + sa.Column('memory_id', sa.String(), nullable=False), + sa.Column('version', sa.Integer(), nullable=False), + sa.Column('payload', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('supersedes_version_id', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['memory_id'], ['memory.id'], name=op.f('fk_memory_version_memory'), ondelete='cascade'), + sa.ForeignKeyConstraint(['supersedes_version_id'], ['memory_version.id'], name=op.f('fk_memory_version_memory_version'), ondelete='set null'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_memory_version')) + ) + op.create_index(op.f('ix_memory_version_memory_id'), 'memory_version', ['memory_id'], unique=False) + op.create_index('uq_memory_version_number', 'memory_version', ['memory_id', 'version'], unique=True) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index('uq_memory_version_number', table_name='memory_version') + op.drop_index(op.f('ix_memory_version_memory_id'), table_name='memory_version') + op.drop_table('memory_version') + op.drop_index(op.f('ix_memory_candidate_owner_personal_domain_id'), table_name='memory_candidate') + op.drop_table('memory_candidate') + op.drop_index(op.f('ix_memory_owner_personal_domain_id'), table_name='memory') + op.drop_index('ix_memory_domain_status_kind', table_name='memory') + op.drop_index(op.f('ix_memory_character_id'), table_name='memory') + op.drop_table('memory') + op.drop_index(op.f('ix_big_cycle_character_id'), table_name='big_cycle') + op.drop_table('big_cycle') + # ### end Alembic commands ### diff --git a/backend/app/main.py b/backend/app/main.py index c1af63b..1142a0a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -10,6 +10,8 @@ from app.core.config import get_settings from app.core.errors import install_error_handlers from app.core.middleware import RequestContextMiddleware, configure_logging from app.db.session import init_engine +from app.memories.routes import cycle_router +from app.memories.routes import router as memories_router def create_app() -> FastAPI: @@ -29,7 +31,9 @@ def create_app() -> FastAPI: app.include_router(conversations_router) app.include_router(run_router) app.include_router(message_router) + app.include_router(cycle_router) # 先于 characters：避免通配生命周期路由拦截 app.include_router(characters_router) + app.include_router(memories_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/memories/__init__.py b/backend/app/memories/__init__.py new file mode 100644 index 0000000..6a66066 --- /dev/null +++ b/backend/app/memories/__init__.py @@ -0,0 +1 @@ +"""记忆域：候选、版本、纠正、删除与检索。""" diff --git a/backend/app/memories/big_cycle.py b/backend/app/memories/big_cycle.py new file mode 100644 index 0000000..9aaf6bb --- /dev/null +++ b/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 diff --git a/backend/app/memories/routes.py b/backend/app/memories/routes.py new file mode 100644 index 0000000..61b7776 --- /dev/null +++ b/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} diff --git a/backend/app/memories/service.py b/backend/app/memories/service.py new file mode 100644 index 0000000..4174c93 --- /dev/null +++ b/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 diff --git a/backend/app/models.py b/backend/app/models.py index f1da2cd..e6d4127 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -472,3 +472,145 @@ class ModelCall(Base): ), Index("uq_model_call_turn_attempt", "turn_id", "attempt", unique=True), ) + + +MEMORY_KIND_CK = CheckConstraint( + "kind IN ('fact', 'preference', 'event', 'relationship')", name="kind_values" +) +SENSITIVITY_CK = CheckConstraint( + "sensitivity IN ('none', 'low', 'medium', 'high')", name="sensitivity_values" +) +MEMORY_REALM_CK = CheckConstraint( + "realm IN ('reality', 'world', 'simulation')", name="realm_values" +) + + +class MemoryCandidate(Base): + """记忆候选：待判定提取结果，不参与召回（ADR-011）。""" + + __tablename__ = "memory_candidate" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("memory_candidate_") + ) + owner_personal_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + character_id: Mapped[str | None] = mapped_column( + ForeignKey("character.id", ondelete="cascade"), nullable=True + ) + kind: Mapped[str] = mapped_column(String, nullable=False) + sensitivity: Mapped[str] = mapped_column(String, nullable=False, default="none") + realm: Mapped[str] = mapped_column(String, nullable=False, default="reality") + payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + source: Mapped[str] = mapped_column(String, nullable=False, default="manual") + source_ref: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + content_hash: Mapped[str | None] = mapped_column(String, nullable=True) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + MEMORY_KIND_CK, + SENSITIVITY_CK, + MEMORY_REALM_CK, + CheckConstraint( + "status IN ('pending', 'approved', 'rejected', 'expired')", name="status_values" + ), + CheckConstraint( + "source IN ('manual', 'auto', 'big_cycle', 'canonization')", name="source_values" + ), + ) + + +class MemoryVersion(Base): + """不可变记忆版本：纠正/取代/恢复都创建新版本（ADR-011）。""" + + __tablename__ = "memory_version" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("memory_version_") + ) + memory_id: Mapped[str] = mapped_column( + ForeignKey("memory.id", ondelete="cascade"), nullable=False, index=True + ) + version: Mapped[int] = mapped_column(nullable=False) + payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + supersedes_version_id: Mapped[str | None] = mapped_column( + ForeignKey("memory_version.id", ondelete="set null"), nullable=True + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + Index("uq_memory_version_number", "memory_id", "version", unique=True), + ) + + +class Memory(Base): + """权威长期记忆：候选批准后创建；当前版本由指针引用。""" + + __tablename__ = "memory" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("memory_")) + owner_personal_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + character_id: Mapped[str | None] = mapped_column( + ForeignKey("character.id", ondelete="cascade"), nullable=True, index=True + ) + kind: Mapped[str] = mapped_column(String, nullable=False) + sensitivity: Mapped[str] = mapped_column(String, nullable=False, default="none") + realm: Mapped[str] = mapped_column(String, nullable=False, default="reality") + status: Mapped[str] = mapped_column(String, nullable=False, default="active") + valid_from: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + pinned: Mapped[bool] = mapped_column(nullable=False, default=False) + current_version_id: Mapped[str | None] = mapped_column( + ForeignKey("memory_version.id", ondelete="set null", use_alter=True), nullable=True + ) + resource_version: Mapped[int] = mapped_column(nullable=False, default=1) # ETag 源 + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + __table_args__ = ( + MEMORY_KIND_CK, + SENSITIVITY_CK, + MEMORY_REALM_CK, + CheckConstraint( + "status IN ('active', 'superseded', 'archived', 'deleted')", name="status_values" + ), + Index("ix_memory_domain_status_kind", "owner_personal_domain_id", "status", "kind"), + ) + + +class BigCycle(Base): + """大周天循环：cycle_ 聚合根，四阶段评估（ADR-018/023）。""" + + __tablename__ = "big_cycle" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("cycle_")) + character_id: Mapped[str] = mapped_column( + ForeignKey("character.id", ondelete="cascade"), nullable=False, index=True + ) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending") + trigger: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + phase_memory: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + produced_version_ids: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint( + "status IN ('pending', 'running', 'completed', 'failed')", name="status_values" + ), + ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index a8aaa13..0852eaa 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -1,5 +1,6 @@ """垂直切片测试：幂等原子事务、IDOR fail-closed、CSRF、SSE 与回滚。""" +import contextlib import json import pytest @@ -708,3 +709,175 @@ class TestGatewayExecution: assert "content.block.failed" in types failed = next(e for e in evs if e.type == "content.block.failed") assert failed.data["error"] == "no_endpoint_available" + + +@contextlib.asynccontextmanager +async def _test_db(): + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from tests.conftest import TEST_DATABASE_URL + + engine = create_async_engine(TEST_DATABASE_URL) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + yield factory + finally: + await engine.dispose() + + +@pytest.mark.usefixtures("client") +class TestMemories: + async def _submit(self, client, headers, body) -> dict: + resp = await client.post("/v1/memories/candidates", json=body, headers=headers) + assert resp.status_code == 201, resp.text + return resp.json() + + async def test_manual_candidate_review_flow(self, client: AsyncClient) -> None: + headers = await login(client, "mem1@example.com") + c = await self._submit(client, headers, { + "kind": "relationship", "sensitivity": "high", + "payload": {"text": "用户偏好深夜交谈", "note": "高影响关系结论"}, + }) + assert c["status"] == "pending" # 敏感默认人工 + approve = await client.post( + f"/v1/memories/candidates/{c['id']}/decisions", + json={"approve": True}, headers=headers, + ) + assert approve.status_code == 201 + assert approve.json()["status"] == "approved" + assert approve.json()["created_memory_id"] is not None + + async def test_whitelist_auto_approve_only_low_sensitivity_fact(self, client) -> None: + headers = await login(client, "mem2@example.com") + auto = await self._submit(client, headers, { + "kind": "fact", "sensitivity": "low", "source": "auto", + "payload": {"text": "用户住在杭州"}, + }) + assert auto["status"] == "approved" # 白名单自动写入 + assert auto["created_memory_id"] is not None + # 高敏感自动候选不自动批准 + sensitive = await self._submit(client, headers, { + "kind": "fact", "sensitivity": "high", "source": "auto", + "payload": {"text": "敏感事实"}, + }) + assert sensitive["status"] == "pending" + # 非 fact 也不自动 + pref = await self._submit(client, headers, { + "kind": "preference", "sensitivity": "low", "source": "auto", + "payload": {"text": "喜欢猫"}, + }) + assert pref["status"] == "pending" + + async def test_correction_with_etag_and_conflict(self, client) -> None: + headers = await login(client, "mem3@example.com") + c = await self._submit(client, headers, { + "kind": "fact", "sensitivity": "low", "source": "auto", + "payload": {"text": "原记录"}, + }) + mid = c["created_memory_id"] + # 缺 If-Match：412 + no_header = await client.patch( + f"/v1/memories/{mid}", json={"payload": {"text": "纠正"}}, headers=headers + ) + assert no_header.status_code == 412 + # 错误版本：412 + stale = await client.patch( + f"/v1/memories/{mid}", json={"payload": {"text": "纠正"}}, + headers={**headers, "If-Match": '"99"'}, + ) + assert stale.status_code == 412 + # 正确 ETag → v2 + ok = await client.patch( + f"/v1/memories/{mid}", json={"payload": {"text": "纠正后"}}, + headers={**headers, "If-Match": '"1"'}, + ) + assert ok.status_code == 200 + body = ok.json() + assert body["version"] == 2 and body["payload"]["text"] == "纠正后" + assert body["resource_version"] == 2 + + async def test_search_filters_and_isolation(self, client) -> None: + headers = await login(client, "mem4@example.com") + for text in ("杭州西湖", "喜欢跑步", "工作在软件园"): + await self._submit(client, headers, { + "kind": "fact", "sensitivity": "low", "source": "auto", + "payload": {"text": text}, + }) + hits = await client.get("/v1/memories?q=西湖", headers=headers) + data = hits.json()["data"] + assert len(data) == 1 and "西湖" in data[0]["payload"]["text"] + assert data[0]["recall"]["channel"] == "text" # 召回证据 + # 跨域隔离 + other = await login(client, "mem4-evil@example.com") + leak = await client.get("/v1/memories?q=西湖", headers=other) + assert leak.json()["data"] == [] + + async def test_delete_purges_content_and_blocks_recall(self, client) -> None: + from sqlalchemy import select as sa_select + + from app.models import MemoryVersion as MV + + headers = await login(client, "mem5@example.com") + c = await self._submit(client, headers, { + "kind": "fact", "sensitivity": "low", "source": "auto", + "payload": {"text": "将被删除的秘密"}, + }) + mid = c["created_memory_id"] + resp = await client.delete(f"/v1/memories/{mid}", headers=headers) + assert resp.status_code == 204 + # 不再召回 + after = await client.get("/v1/memories?q=秘密", headers=headers) + assert after.json()["data"] == [] + # 墓碑：版本原文已清除（契约：墓碑不存被删原文） + async with _test_db() as factory: + async with factory() as session: + versions = list( + ( + await session.execute(sa_select(MV).where(MV.memory_id == mid)) + ).scalars().all() + ) + assert versions and all(v.payload == {} for v in versions) + + +@pytest.mark.usefixtures("client") +class TestBigCycle: + async def test_manual_trigger_produces_pending_candidate(self, client, db) -> None: + + from app.models import BigCycle, CharacterVersion + + headers = await login(client, "bc1@example.com") + c = await client.post("/v1/characters", json={"name": "砚舟"}, headers=headers) + cid = c.json()["id"] + await client.post(f"/v1/characters/{cid}/activations", headers=headers) + # 先有一条对话消息供回顾 + conv = await _create_conversation(client, headers) + await client.post( + f"/v1/conversations/{conv}/messages", + json={"content": [{"type": "text", "text": "最近在想什么呢"}], "start_run": False}, + headers={**headers, "Idempotency-Key": "idem-bc1"}, + ) + trigger = await client.post(f"/v1/characters/{cid}/big-cycles", headers=headers) + assert trigger.status_code == 201, trigger.text + cycle_id = trigger.json()["id"] + # 轮询等待 cycle 终态 + import asyncio + + deadline = asyncio.get_event_loop().time() + 10 + cycle = None + while asyncio.get_event_loop().time() < deadline: + async with db() as session: + cycle = await session.get(BigCycle, cycle_id) + if cycle and cycle.status in {"completed", "failed"}: + break + await asyncio.sleep(0.05) + assert cycle is not None and cycle.status == "completed", ( + cycle.status if cycle else "missing" + ) + assert cycle.produced_version_ids + async with db() as session: + version = await session.get(CharacterVersion, cycle.produced_version_ids[0]) + assert version is not None + assert version.kind == "persona_candidate" + assert version.status == "pending" # 候选先于写入 + assert version.payload["source"] == "big_cycle" + assert version.payload["proposal"] # 模型真实输出