diff --git a/backend/alembic/versions/8f48659a83f6_三界作用域与消息版本谱系.py b/backend/alembic/versions/8f48659a83f6_三界作用域与消息版本谱系.py new file mode 100644 index 0000000..f7206ea --- /dev/null +++ b/backend/alembic/versions/8f48659a83f6_三界作用域与消息版本谱系.py @@ -0,0 +1,42 @@ +"""三界作用域与消息版本谱系 + +Revision ID: 8f48659a83f6 +Revises: 5cf6d328c10a +Create Date: 2026-09-10 13:22:56.404702 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '8f48659a83f6' +down_revision: str | Sequence[str] | None = '5cf6d328c10a' +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.add_column('conversation', sa.Column('realm', sa.String(), nullable=False)) + op.add_column('message', sa.Column('realm', sa.String(), nullable=False)) + op.add_column('message', sa.Column('canonized_at', sa.DateTime(timezone=True), nullable=True)) + op.add_column('message', sa.Column('parent_message_id', sa.String(), nullable=True)) + op.add_column('message', sa.Column('variant_kind', sa.String(), nullable=False)) + op.create_foreign_key(op.f('fk_message_message'), 'message', 'message', ['parent_message_id'], ['id'], ondelete='set null') + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_constraint(op.f('fk_message_message'), 'message', type_='foreignkey') + op.drop_column('message', 'variant_kind') + op.drop_column('message', 'parent_message_id') + op.drop_column('message', 'canonized_at') + op.drop_column('message', 'realm') + op.drop_column('conversation', 'realm') + # ### end Alembic commands ### diff --git a/backend/app/conversations/routes.py b/backend/app/conversations/routes.py index 202a216..0a1172f 100644 --- a/backend/app/conversations/routes.py +++ b/backend/app/conversations/routes.py @@ -206,3 +206,96 @@ async def set_current_branch( db, domain.id, conversation_id, body.get("branch_id", "") ) return {"id": conversation.id, "current_branch_id": conversation.current_branch_id or ""} + + +@router.get("/{conversation_id}") +async def get_conversation( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(get_current_user)], +) -> dict[str, object]: + domain = await _domain_of(db, user) + conversation = await service.require_conversation(db, domain.id, conversation_id) + from app.conversations.schemas import ConversationOut + + return ConversationOut( + id=conversation.id, + title=conversation.title, + current_branch_id=conversation.current_branch_id, + created_at=conversation.created_at.isoformat(), + ).model_dump() | {"realm": conversation.realm} + + +@router.post("/{conversation_id}/realm") +async def set_realm( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, str], +) -> dict[str, str]: + """入境/出境：切换对话界域（显式、留审计）。""" + domain = await _domain_of(db, user) + conversation = await service.set_realm( + db, + domain.id, + conversation_id, + body.get("realm", ""), + getattr(request.state, "trace_id", None), + ) + return {"id": conversation.id, "realm": conversation.realm} + + +message_router = APIRouter(prefix="/v1/messages", tags=["messages"]) + + +@message_router.post("/{message_id}/canonizations", status_code=201, response_model=None) +async def canonize_message( + message_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, str], +) -> JSONResponse | dict[str, str | None]: + """归真：携带 canonical_scope（ADR-020）；重复归真幂等返回原状态（200）。""" + from fastapi.responses import JSONResponse + + domain = await _domain_of(db, user) + message, replayed = await service.canonize_message( + db, domain.id, message_id, body.get("scope", ""), getattr(request.state, "trace_id", None) + ) + payload = { + "message_id": message.id, + "canonical_scope": message.realm, + "canonized_at": message.canonized_at.isoformat() if message.canonized_at else None, + } + if replayed: + return JSONResponse(status_code=200, content=payload) + return payload + + +@message_router.post("/{message_id}/revisions", status_code=201) +async def revise_message( + message_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: MessageIn, +) -> dict[str, object]: + """编辑版本：新消息引用原消息；原消息永不修改。""" + domain = await _domain_of(db, user) + revision = await service.revise_message( + db, domain.id, message_id, body, getattr(request.state, "trace_id", None) + ) + from app.conversations.schemas import MessageOut + + return MessageOut( + id=revision.id, + conversation_id=revision.conversation_id, + branch_id=revision.branch_id, + role=revision.role, + content=revision.content, + parser_version=revision.parser_version, + created_at=revision.created_at.isoformat(), + ).model_dump() | {"parent_message_id": revision.parent_message_id, + "variant_kind": revision.variant_kind} diff --git a/backend/app/conversations/service.py b/backend/app/conversations/service.py index ff9b2a8..da8775b 100644 --- a/backend/app/conversations/service.py +++ b/backend/app/conversations/service.py @@ -42,7 +42,7 @@ def _request_hash(body: MessageIn) -> str: return hashlib.sha256(canonical.encode("utf-8")).hexdigest() -async def _require_conversation( +async def require_conversation( db: AsyncSession, domain_id: str, conversation_id: str ) -> Conversation: conversation = await db.get(Conversation, conversation_id) @@ -121,7 +121,7 @@ async def replay_or_execute_message( assert message is not None return message, run, True - conversation = await _require_conversation(db, domain_id, conversation_id) + conversation = await require_conversation(db, domain_id, conversation_id) branch = await _get_default_branch(db, conversation.id) forged = [b.type for b in body.content if b.type in RUNTIME_BLOCK_TYPES] @@ -148,6 +148,7 @@ async def replay_or_execute_message( conversation_id=conversation.id, branch_id=branch.id, role="user", + realm=conversation.realm, content=blocks, parser_version=parser_version, ) @@ -242,7 +243,7 @@ async def list_messages( cursor: str | None, ) -> tuple[list[Message], str | None, bool]: """按创建时间升序的消息列表（opaque cursor 分页，契约第 4.1 节）。""" - conversation = await _require_conversation(db, domain_id, conversation_id) + conversation = await require_conversation(db, domain_id, conversation_id) stmt = select(Message).where(Message.conversation_id == conversation.id) if branch_id is not None: stmt = stmt.where(Message.branch_id == branch_id) @@ -284,7 +285,7 @@ def decode_cursor(cursor: str | None) -> tuple[Any | None, str]: async def list_branches( db: AsyncSession, domain_id: str, conversation_id: str ) -> list[Branch]: - conversation = await _require_conversation(db, domain_id, conversation_id) + conversation = await require_conversation(db, domain_id, conversation_id) result = await db.execute( select(Branch).where(Branch.conversation_id == conversation.id).order_by(Branch.created_at) ) @@ -295,10 +296,146 @@ async def set_current_branch( db: AsyncSession, domain_id: str, conversation_id: str, branch_id: str ) -> Conversation: """切换当前路径：分支必须属于该会话（跨域/跨会话统一 404/422 fail-closed）。""" - conversation = await _require_conversation(db, domain_id, conversation_id) + conversation = await require_conversation(db, domain_id, conversation_id) branch = await db.get(Branch, branch_id) if branch is None or branch.conversation_id != conversation.id: raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "分支不属于该会话") conversation.current_branch_id = branch_id await db.flush() return conversation + + +async def set_realm( + db: AsyncSession, + domain_id: str, + conversation_id: str, + realm: str, + trace_id: str | None, +) -> Conversation: + """入境/出境：显式切换对话界域并留审计（ADR-020：切换必须可见）。""" + if realm not in {"reality", "world", "simulation"}: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "realm 取值不合法") + conversation = await require_conversation(db, domain_id, conversation_id) + previous = conversation.realm + conversation.realm = realm + db.add( + AuditEvent( + actor_type="user", + actor_id=domain_id, + action="conversation.realm_changed", + aggregate_type="conversation", + aggregate_id=conversation.id, + payload={"from": previous, "to": realm}, + trace_id=trace_id, + ) + ) + await db.flush() + return conversation + + +async def _require_message( + db: AsyncSession, domain_id: str, message_id: str +) -> Message: + message = await db.get(Message, message_id) + if message is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "消息不存在") + conversation = await db.get(Conversation, message.conversation_id) + assert conversation is not None + if conversation.personal_domain_id != domain_id: + raise HTTPException(status.HTTP_404_NOT_FOUND, "消息不存在") + return message + + +async def canonize_message( + db: AsyncSession, + domain_id: str, + message_id: str, + scope: str, + trace_id: str | None, +) -> tuple[Message, bool]: + """归真：scope 必须与消息所在界域一致（ADR-020）；重复归真幂等返回原状态。 + + 境界归真产出 world.canonized Outbox 事件（世界知识候选入口，WP-04 消费）； + 幻境归真仅标记试演记录，不产生长期记忆通道。 + """ + if scope not in {"reality", "world", "simulation"}: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "scope 取值不合法") + message = await _require_message(db, domain_id, message_id) + if message.realm != scope: + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + f"归真作用域与消息界域不一致：消息在 {message.realm}，请求 {scope}", + ) + if message.canonized_at is not None: + return message, True # 幂等：重复归真返回原状态 + message.canonized_at = datetime.now(UTC) + event_type = "message.canonized" if scope != "world" else "world.canonized" + db.add( + AuditEvent( + actor_type="user", + actor_id=domain_id, + action="message.canonized", + aggregate_type="message", + aggregate_id=message.id, + payload={"canonical_scope": scope}, + trace_id=trace_id, + ) + ) + db.add( + OutboxEvent( + aggregate_type="message", + aggregate_id=message.id, + event_type=event_type, + payload={"message_id": message.id, "canonical_scope": scope}, + ) + ) + await db.flush() + return message, False + + +async def revise_message( + db: AsyncSession, + domain_id: str, + message_id: str, + body: MessageIn, + trace_id: str | None, +) -> Message: + """编辑版本：创建引用原消息的新 Message（variant=revision），原消息不变。""" + original = await _require_message(db, domain_id, message_id) + blocks: list[dict[str, Any]] = [] + parser_version: str | None = None + for b in body.content: + if b.type == "markdown": + normalized = lingtai.transform_markdown(b.text) + parser_version = str(normalized["parser_version"]) + blocks.extend(normalized["blocks"]) + else: + blocks.append({**b.model_dump(exclude_none=True), "status": "completed"}) + for block in blocks: + block["id"] = new_id("block_") + block.setdefault("status", "completed") + revision = Message( + conversation_id=original.conversation_id, + branch_id=original.branch_id, + role=original.role, + realm=original.realm, + parent_message_id=original.id, + variant_kind="revision", + content=blocks, + parser_version=parser_version, + ) + db.add(revision) + await db.flush() # 生成 revision.id 后再写审计 + db.add( + AuditEvent( + actor_type="user", + actor_id=domain_id, + action="message.revised", + aggregate_type="message", + aggregate_id=revision.id, + payload={"revises": original.id}, + trace_id=trace_id, + ) + ) + await db.flush() + return revision diff --git a/backend/app/main.py b/backend/app/main.py index 2e7e09a..5ca729f 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -3,8 +3,8 @@ from fastapi import FastAPI from app.auth.routes import router as auth_router +from app.conversations.routes import message_router, run_router from app.conversations.routes import router as conversations_router -from app.conversations.routes import run_router from app.core.config import get_settings from app.core.errors import install_error_handlers from app.core.middleware import RequestContextMiddleware, configure_logging @@ -27,6 +27,7 @@ def create_app() -> FastAPI: app.include_router(auth_router) app.include_router(conversations_router) app.include_router(run_router) + app.include_router(message_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/models.py b/backend/app/models.py index dae371a..cd422ae 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -142,6 +142,11 @@ class OutboxEvent(Base): ) +CONVERSATION_REALM_CK = CheckConstraint( + "realm IN ('reality', 'world', 'simulation')", name="realm_values" +) + + class Conversation(Base): """对话容器：归属个人域（垂直切片 WP-03 最小集）。""" @@ -154,6 +159,7 @@ class Conversation(Base): ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True ) title: Mapped[str] = mapped_column(String, nullable=False, default="未命名对话") + realm: Mapped[str] = mapped_column(String, nullable=False, default="reality") current_branch_id: Mapped[str | None] = mapped_column( ForeignKey("branch.id", ondelete="set null", use_alter=True), nullable=True ) @@ -164,6 +170,8 @@ class Conversation(Base): DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) + __table_args__ = (CONVERSATION_REALM_CK,) + class Branch(Base): """对话分支：创建会话时生成默认主分支。""" @@ -193,6 +201,12 @@ class Message(Base): ForeignKey("branch.id", ondelete="cascade"), nullable=False, index=True ) role: Mapped[str] = mapped_column(String, nullable=False) + realm: Mapped[str] = mapped_column(String, nullable=False, default="reality") + canonized_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + parent_message_id: Mapped[str | None] = mapped_column( + ForeignKey("message.id", ondelete="set null"), nullable=True + ) + variant_kind: Mapped[str] = mapped_column(String, nullable=False, default="origin") content: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False) parser_version: Mapped[str | None] = mapped_column(String, nullable=True) created_at: Mapped[datetime] = mapped_column( @@ -201,6 +215,12 @@ class Message(Base): __table_args__ = ( CheckConstraint("role IN ('user', 'assistant', 'system', 'tool')", name="role_values"), + CheckConstraint( + "realm IN ('reality', 'world', 'simulation')", name="realm_values" + ), + CheckConstraint( + "variant_kind IN ('origin', 'revision', 'candidate')", name="variant_kind_values" + ), ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index eefc189..bbf4d5d 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -326,3 +326,136 @@ class TestPaginationAndBranches: f"/v1/conversations/{cid}/current-branch", json={"branch_id": bid} ) assert resp.status_code in (401, 403) # 未带新会话 Cookie：401；带错 Cookie：拒绝 + + +@pytest.mark.usefixtures("client") +class TestRealmsAndCanonization: + async def _make_message(self, client, headers, cid, key, text="hi") -> dict: + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "text", "text": text}], "start_run": False}, + headers={**headers, "Idempotency-Key": key}, + ) + assert resp.status_code == 201, resp.text + return resp.json()["message"] + + async def test_realm_switch_snapshots_into_new_messages(self, client: AsyncClient) -> None: + headers = await login(client, "realm1@example.com") + cid = await _create_conversation(client, headers) + detail = await client.get(f"/v1/conversations/{cid}", headers=headers) + assert detail.json()["realm"] == "reality" # 默认真界 + sw = await client.post( + f"/v1/conversations/{cid}/realm", json={"realm": "simulation"}, headers=headers + ) + assert sw.status_code == 200 and sw.json()["realm"] == "simulation" + await self._make_message(client, headers, cid, "idem-sim1") + await client.get(f"/v1/conversations/{cid}/messages", headers=headers) + # （realm 字段在消息响应中暂未透出，以归真行为间接验证） + # 回到真界 + await client.post( + f"/v1/conversations/{cid}/realm", json={"realm": "reality"}, headers=headers + ) + + async def test_canonize_reality_message(self, client: AsyncClient) -> None: + headers = await login(client, "can1@example.com") + cid = await _create_conversation(client, headers) + msg = await self._make_message(client, headers, cid, "idem-can1") + resp = await client.post( + f"/v1/messages/{msg['id']}/canonizations", + json={"scope": "reality"}, + headers=headers, + ) + assert resp.status_code == 201 + assert resp.json()["canonical_scope"] == "reality" + assert resp.json()["canonized_at"] is not None + + async def test_canonize_idempotent_replay(self, client: AsyncClient) -> None: + headers = await login(client, "can2@example.com") + cid = await _create_conversation(client, headers) + msg = await self._make_message(client, headers, cid, "idem-can2") + first = await client.post( + f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"}, headers=headers + ) + second = await client.post( + f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"}, headers=headers + ) + assert first.status_code == 201 and second.status_code == 200 + assert second.json()["canonized_at"] == first.json()["canonized_at"] + + async def test_canonize_scope_mismatch_rejected(self, client: AsyncClient) -> None: + """真界消息不能归真到境界/幻境作用域（ADR-020 作用域一致）。""" + headers = await login(client, "can3@example.com") + cid = await _create_conversation(client, headers) + msg = await self._make_message(client, headers, cid, "idem-can3") + resp = await client.post( + f"/v1/messages/{msg['id']}/canonizations", json={"scope": "world"}, headers=headers + ) + assert resp.status_code == 422 + + async def test_simulation_message_canonizes_as_rehearsal(self, client: AsyncClient) -> None: + """幻境消息归真仅标记试演记录（ADR-020）。""" + headers = await login(client, "can4@example.com") + cid = await _create_conversation(client, headers) + await client.post( + f"/v1/conversations/{cid}/realm", json={"realm": "simulation"}, headers=headers + ) + msg = await self._make_message(client, headers, cid, "idem-can4") + resp = await client.post( + f"/v1/messages/{msg['id']}/canonizations", json={"scope": "simulation"}, headers=headers + ) + assert resp.status_code == 201 + + async def test_canonize_cross_domain_404(self, client: AsyncClient) -> None: + alice = await login(client, "can5@example.com") + cid = await _create_conversation(client, alice) + msg = await self._make_message(client, alice, cid, "idem-can5") + await login(client, "can5-evil@example.com") + resp = await client.post( + f"/v1/messages/{msg['id']}/canonizations", json={"scope": "reality"} + ) + assert resp.status_code in (401, 403, 404) # 未携带新会话 Cookie/CSRF：拒绝 + + +@pytest.mark.usefixtures("client") +class TestRevisions: + async def test_revision_keeps_original_intact(self, client: AsyncClient) -> None: + + headers = await login(client, "rev1@example.com") + cid = await _create_conversation(client, headers) + original = None + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "text", "text": "原始内容"}], "start_run": False}, + headers={**headers, "Idempotency-Key": "idem-rev1"}, + ) + original = resp.json()["message"] + rev = await client.post( + f"/v1/messages/{original['id']}/revisions", + json={"content": [{"type": "markdown", "text": "修订后\n```python\nprint(2)\n```"}]}, + headers=headers, + ) + assert rev.status_code == 201, rev.text + body = rev.json() + assert body["parent_message_id"] == original["id"] + assert body["variant_kind"] == "revision" + # markdown 修订经引擎规范化为两块 + assert [b["type"] for b in body["content"]] == ["markdown", "code"] + # 原消息内容不变：列表中两条，原消息文本仍是"原始内容" + listing = await client.get(f"/v1/conversations/{cid}/messages", headers=headers) + by_id = {m["id"]: m for m in listing.json()["data"]} + assert by_id[original["id"]]["content"][0]["text"] == "原始内容" + + async def test_revision_cross_domain_404(self, client: AsyncClient) -> None: + alice = await login(client, "rev2@example.com") + cid = await _create_conversation(client, alice) + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "text", "text": "x"}], "start_run": False}, + headers={**alice, "Idempotency-Key": "idem-rev2"}, + ) + mid = resp.json()["message"]["id"] + await login(client, "rev2-evil@example.com") + rev = await client.post( + f"/v1/messages/{mid}/revisions", json={"content": [{"type": "text", "text": "篡改"}]} + ) + assert rev.status_code in (401, 403, 404)