chenxin/hunjing
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
40221170aefc56854e3ddc61eaefe4e4ba81b1d4
hunjing/backend/app/models.py
T
chenxin 40221170ae ✨ feat: WP-02 角色域——完整生命周期、铸魂、内驱力与随缘 
- 迁移 6：character（生命周期六态检查约束、起源、现实人物治理标记、道体基线、内驱力）+ character_version（baseline/first_statement/drive_revision/reforging/persona_candidate 五类不可变版本与审核态）
- 创建：手动与随缘（确定性词库组合，同种子同候选、一次一个——不是抽卡）；现实人物治理（已故拒绝、未声明授权拒绝）
- 启镜（draft→active + 首个 baseline 版本）与初照自陈（候选→确认归真并入道体）
- 生命周期端点：静修/归藏/复照（归档可恢复、复照不抹历史）+ 死亡两步确认（7 天冷静期可撤销、未满确认 422）；非法状态迁移 422
- 铸魂（显式 confirm 强确认，直接更新道体）与人格候选（pending→审核批准才并入道体，不静默覆盖）
- 内驱力编辑：写当前值并落 drive_revision 版本（可追溯，ADR-018）
- 测试 52 项全绿（+8）：状态机全路径、随缘确定性、现实人物三例、铸魂确认、候选审核与重复审核 409、跨域 fail-closed
2026-09-10 18:11:15 +08:00

377 lines
16 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""首批领域模型：身份、会话、服务账号、审计与 Outbox（WP-01）。
- 主键为带前缀的 UUIDv7 文本（ADR-015）；Session ID 为 CSPRNG opaque（秘密不用 UUID）。
- 命名约定见 app/db/base.py（pk_/uq_/ix_/fk_，AGENTS.md 8.6）。
- AuditEvent 与 OutboxEvent 不可变：无 updated_at，修正写新记录（契约第 4.4 节）。
"""
from datetime import datetime
from typing import Any
from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from app.core.ids import new_id
from app.db.base import Base
class User(Base):
__tablename__ = "user"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("user_"))
email: Mapped[str] = mapped_column(String, nullable=False, unique=True)
display_name: Mapped[str] = mapped_column(String, nullable=False)
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()
)
class PersonalDomain(Base):
__tablename__ = "personal_domain"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: new_id("domain_")
)
user_id: Mapped[str] = mapped_column(
ForeignKey("user.id", ondelete="restrict"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String, nullable=False)
# 区域驻留（ADR-024 之前的数据驻留决策）：默认区域，迁移走正式协议
data_region: Mapped[str] = mapped_column(String, nullable=False, default="default")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
class Session(Base):
"""Web 会话：ID 为 CSPRNG opaque value，数据库只存校验值（ADR-015）。"""
__tablename__ = "session"
id: Mapped[str] = mapped_column(String, primary_key=True)
user_id: Mapped[str] = mapped_column(
ForeignKey("user.id", ondelete="cascade"), nullable=False, index=True
)
token_hash: Mapped[str] = mapped_column(String, nullable=False, unique=True)
csrf_token_hash: Mapped[str] = mapped_column(String, nullable=False)
expires_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, index=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
class ServiceAccount(Base):
"""自动化账号：API key 明文只显示一次，只保存哈希、前缀、scope 与撤销状态。"""
__tablename__ = "service_account"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: new_id("service_account_")
)
domain_id: Mapped[str] = mapped_column(
ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False
)
name: Mapped[str] = mapped_column(String, nullable=False)
key_prefix: Mapped[str] = mapped_column(String, nullable=False, index=True)
key_hash: Mapped[str] = mapped_column(String, nullable=False)
scopes: 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()
)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
__table_args__ = (Index("uq_service_account_domain_name", "domain_id", "name", unique=True),)
class AuditEvent(Base):
"""不可变审计链：状态与 Outbox 在同一事务提交（契约第 4.4 节）。"""
__tablename__ = "audit_event"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("audit_"))
occurred_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
actor_type: Mapped[str] = mapped_column(String, nullable=False)
actor_id: Mapped[str] = mapped_column(String, nullable=False)
action: Mapped[str] = mapped_column(String, nullable=False)
aggregate_type: Mapped[str] = mapped_column(String, nullable=False)
aggregate_id: Mapped[str] = mapped_column(String, nullable=False)
payload: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
trace_id: Mapped[str | None] = mapped_column(String, nullable=True)
__table_args__ = (
# 聚合时间线是审计的主要查询路径（最左前缀覆盖聚合定位）
Index("ix_audit_event_aggregate", "aggregate_type", "aggregate_id", "occurred_at"),
Index("ix_audit_event_occurred_at", "occurred_at"),
)
class OutboxEvent(Base):
"""Transactional Outbox：已提交领域事实的可靠投递队列。"""
__tablename__ = "outbox_event"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("event_"))
occurred_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
aggregate_type: Mapped[str] = mapped_column(String, nullable=False)
aggregate_id: Mapped[str] = mapped_column(String, nullable=False)
event_type: Mapped[str] = mapped_column(String, nullable=False)
payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
attempts: Mapped[int] = mapped_column(nullable=False, default=0)
delivered_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
__table_args__ = (
CheckConstraint(
"status IN ('pending', 'delivered', 'failed')",
name="status_values",
),
# 投递循环按状态+时间扫描待处理事件
Index("ix_outbox_event_status_occurred", "status", "occurred_at"),
Index("ix_outbox_event_aggregate", "aggregate_type", "aggregate_id"),
)
CONVERSATION_REALM_CK = CheckConstraint(
"realm IN ('reality', 'world', 'simulation')", name="realm_values"
)
class Conversation(Base):
"""对话容器：归属个人域（垂直切片 WP-03 最小集）。"""
__tablename__ = "conversation"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: new_id("conversation_")
)
personal_domain_id: Mapped[str] = mapped_column(
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
)
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__ = (CONVERSATION_REALM_CK,)
class Branch(Base):
"""对话分支：创建会话时生成默认主分支。"""
__tablename__ = "branch"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("branch_"))
conversation_id: Mapped[str] = mapped_column(
ForeignKey("conversation.id", ondelete="cascade"), nullable=False, index=True
)
title: Mapped[str] = mapped_column(String, nullable=False, default="主脉")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
class Message(Base):
"""消息：content 为 canonical ContentBlock JSON（本切片为 Text/Markdown 子集）。"""
__tablename__ = "message"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("message_"))
conversation_id: Mapped[str] = mapped_column(
ForeignKey("conversation.id", ondelete="cascade"), nullable=False, index=True
)
branch_id: Mapped[str] = mapped_column(
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(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
__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"
),
)
class Run(Base):
"""Agent 执行：状态机见 Agent 运行时第 3 节；本切片空转即完成。"""
__tablename__ = "run"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("run_"))
personal_domain_id: Mapped[str] = mapped_column(
ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True
)
conversation_id: Mapped[str | None] = mapped_column(
ForeignKey("conversation.id", ondelete="cascade"), nullable=True, index=True
)
status: Mapped[str] = mapped_column(String, nullable=False, default="queued")
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 ('queued', 'running', 'waiting_approval', "
"'completed', 'failed', 'cancelled')",
name="status_values",
),
Index("ix_run_domain_status", "personal_domain_id", "status"),
)
class IdempotencyRecord(Base):
"""幂等键记录：按 认证主体+路由+键 隔离，保存请求哈希与结果引用（契约第 6 节）。"""
__tablename__ = "idempotency_record"
key: Mapped[str] = mapped_column(String, primary_key=True)
owner_domain_id: Mapped[str] = mapped_column(
ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False
)
route: Mapped[str] = mapped_column(String, nullable=False)
request_hash: Mapped[str] = mapped_column(String, nullable=False)
response_status: Mapped[int] = mapped_column(nullable=False)
resource_ids: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
__table_args__ = (
Index("uq_idempotency_owner_route_key", "owner_domain_id", "route", "key", unique=True),
)
class RunEvent(Base):
"""Run 事件流：持久化后可重放；sequence 在 Run 内单调（契约第 5 节）。"""
__tablename__ = "run_event"
id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("event_"))
run_id: Mapped[str] = mapped_column(
ForeignKey("run.id", ondelete="cascade"), nullable=False
)
sequence: Mapped[int] = mapped_column(nullable=False)
type: Mapped[str] = mapped_column(String, nullable=False)
data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
__table_args__ = (
CheckConstraint(
"type IN ('content.block.started', 'content.block.delta', 'content.block.progress', "
"'content.block.completed', 'content.block.failed', 'content.block.cancelled', "
"'message.completed', 'run.status.changed')",
name="type_values",
),
# Run 内 sequence 单调且唯一（重放去重的服务端保证）
Index("uq_run_event_run_sequence", "run_id", "sequence", unique=True),
)
class CharacterVersion(Base):
"""角色版本链：baseline/初照自陈/内驱力修订/铸魂/人格候选（不可变版本）。"""
__tablename__ = "character_version"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: new_id("character_version_")
)
character_id: Mapped[str] = mapped_column(
ForeignKey("character.id", ondelete="cascade"), nullable=False, index=True
)
kind: Mapped[str] = mapped_column(String, nullable=False)
status: Mapped[str] = mapped_column(String, nullable=False, default="approved")
payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()
)
__table_args__ = (
CheckConstraint(
"kind IN ('baseline', 'first_statement', 'drive_revision', 'reforging', "
"'persona_candidate')",
name="kind_values",
),
CheckConstraint("status IN ('pending', 'approved', 'rejected')", name="status_values"),
)
class Character(Base):
"""角色：稳定身份、生命周期状态、道体基线与内驱力（WP-02）。"""
__tablename__ = "character"
id: Mapped[str] = mapped_column(
String, primary_key=True, default=lambda: new_id("character_")
)
personal_domain_id: Mapped[str] = mapped_column(
ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True
)
name: Mapped[str] = mapped_column(String, nullable=False)
# 生命周期：灵胚 → 启镜 → 静修/归藏（可复照）；死亡走两步确认，deceased 不可逆
status: Mapped[str] = mapped_column(String, nullable=False, default="draft")
origin: Mapped[str] = mapped_column(String, nullable=False, default="original")
based_on_real_person: Mapped[bool] = mapped_column(nullable=False, default=False)
real_person_attested: Mapped[bool] = mapped_column(nullable=False, default=False)
# 道体基线（identity_baseline）：最小稳定身份锚点，普通对话/记忆不可静默覆盖
identity_baseline: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)
# 内驱力（character_drives）：声明式动机维度，用户可审阅编辑（ADR-018）
drives: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False, default=list)
current_version_id: Mapped[str | None] = mapped_column(
ForeignKey("character_version.id", ondelete="set null", use_alter=True), nullable=True
)
death_requested_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
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__ = (
CheckConstraint(
"status IN ('draft', 'active', 'resting', 'archived', 'death_pending', 'deceased')",
name="status_values",
),
CheckConstraint(
"origin IN ('original', 'serendipity', 'template', 'import', 'fork')",
name="origin_values",
),
)
Reference in new issue View Git Blame Copy Permalink