✨ feat: WP-03 三界归真作用域与消息版本谱系（ADR-020 落地）

- 迁移 5：conversation.realm（默认真界，检查约束）；message.realm 快照、canonized_at、parent_message_id 自引用与 variant_kind（origin/revision/candidate）
- 界域端点：GET conversation（realm/当前分支）与 POST realm（入境/出境显式切换，留审计——ADR-020 切换必须可见）
- 归真端点 POST /v1/messages/{id}/canonizations：scope 必须与消息界域一致（不一致 422）；幂等（重复归真 200 原状态）；境界归真产出 world.canonized Outbox 事件（世界知识候选入口，WP-04 消费）；幻境归真仅标记试演记录
- 编辑版本 POST /v1/messages/{id}/revisions：新消息 variant=revision 引用原消息（markdown 修订同样经灵台引擎规范化），原消息永不修改
- 测试 44 项全绿（+8）：界域快照、归真正/反/幂等、幻境试演、revision 原消息不变、跨域 401/403/404 fail-closed

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 ###

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}

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

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]:

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"
+        ),
     )
 
 

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)