chenxin
b4242f3844
- 迁移 7+8：logical_model（能力目录）、provider_endpoint（密钥只存环境变量名，永不落库）、turn、model_call（每 attempt 单独持久化：usage/TTFT/错误，Provider 未返回保持 null 不估算——ADR-003）；message.run_id 关联 - OpenAI-compatible 流式适配器：httpx SSE 解析、标准化 StreamChunk（delta/usage/done）、transport 可注入（测试与故障注入）；密钥经环境变量读取不进日志 - Run 执行器：后台执行完整生命周期——running → content.block.started → 真实 content.block.delta*（流式产生，delta_index 单调）→ content.block.completed → message.completed → 终态；assistant 消息为权威快照；ModelCall 记录 usage 与 TTFT；无可用端点时 Run 显式失败并留 error 事件（fail-closed） - post_message：响应提交后由框架 BackgroundTasks 执行 Run（显式预提交保证 Run 落库）；SSE 改为增量轮询（重放 + 等待新事件至终态，30s 上限） - 测试 54 项全绿（+2）：端到端 mock 流验证 ModelCall usage/TTFT/assistant 权威消息、无端点 Run failed + content.block.failed 事件
410 lines
14 KiB
Python
410 lines
14 KiB
Python
"""对话域服务：幂等、原子事务与越权防护。
|
||
|
||
不变量（契约 4.3/4.4 节）：Message 与可选 Run 在同一事务创建；权威状态、
|
||
AuditEvent 与 OutboxEvent 同事务提交；跨域对象按 404 fail-closed。
|
||
"""
|
||
|
||
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.conversations.schemas import RUNTIME_BLOCK_TYPES, MessageIn
|
||
from app.core import lingtai
|
||
from app.core.ids import new_id
|
||
from app.models import (
|
||
AuditEvent,
|
||
Branch,
|
||
Conversation,
|
||
IdempotencyRecord,
|
||
Message,
|
||
OutboxEvent,
|
||
Run,
|
||
)
|
||
|
||
MESSAGE_ROUTE = "/v1/conversations/{conversation_id}/messages"
|
||
|
||
|
||
def _request_hash(body: MessageIn) -> str:
|
||
canonical = json.dumps(
|
||
{
|
||
"content": [b.model_dump(exclude_none=True) for b in body.content],
|
||
"start_run": body.start_run,
|
||
},
|
||
sort_keys=True,
|
||
ensure_ascii=False,
|
||
)
|
||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
||
|
||
|
||
async def require_conversation(
|
||
db: AsyncSession, domain_id: str, conversation_id: str
|
||
) -> Conversation:
|
||
conversation = await db.get(Conversation, conversation_id)
|
||
# 跨域对象统一 404，不泄露存在性（契约第 7 节）
|
||
if conversation is None or conversation.personal_domain_id != domain_id:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "对话不存在")
|
||
return conversation
|
||
|
||
|
||
async def _get_default_branch(db: AsyncSession, conversation_id: str) -> Branch:
|
||
result = await db.execute(
|
||
select(Branch).where(Branch.conversation_id == conversation_id).limit(1)
|
||
)
|
||
branch = result.scalar_one_or_none()
|
||
assert branch is not None, "会话必须存在默认分支（创建事务保证）"
|
||
return branch
|
||
|
||
|
||
async def create_conversation(
|
||
db: AsyncSession, domain_id: str, title: str, trace_id: str | None
|
||
) -> Conversation:
|
||
conversation = Conversation(personal_domain_id=domain_id, title=title)
|
||
db.add(conversation)
|
||
await db.flush()
|
||
branch = Branch(conversation_id=conversation.id, title="主脉")
|
||
db.add(branch)
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user",
|
||
actor_id=domain_id,
|
||
action="conversation.created",
|
||
aggregate_type="conversation",
|
||
aggregate_id=conversation.id,
|
||
trace_id=trace_id,
|
||
)
|
||
)
|
||
db.add(
|
||
OutboxEvent(
|
||
aggregate_type="conversation",
|
||
aggregate_id=conversation.id,
|
||
event_type="conversation.created",
|
||
payload={"conversation_id": conversation.id, "title": title},
|
||
)
|
||
)
|
||
await db.flush()
|
||
return conversation
|
||
|
||
|
||
async def replay_or_execute_message(
|
||
db: AsyncSession,
|
||
domain_id: str,
|
||
conversation_id: str,
|
||
idempotency_key: str,
|
||
body: MessageIn,
|
||
trace_id: str | None,
|
||
) -> tuple[Message, Run | None, bool]:
|
||
"""返回（消息，Run，是否为重放）。幂等：相同键+相同请求返回原资源。"""
|
||
route = MESSAGE_ROUTE
|
||
existing = await db.execute(
|
||
select(IdempotencyRecord).where(
|
||
IdempotencyRecord.owner_domain_id == domain_id,
|
||
IdempotencyRecord.route == route,
|
||
IdempotencyRecord.key == idempotency_key,
|
||
)
|
||
)
|
||
record = existing.scalar_one_or_none()
|
||
request_hash = _request_hash(body)
|
||
if record is not None:
|
||
if record.request_hash != request_hash:
|
||
raise HTTPException(status.HTTP_409_CONFLICT, "相同幂等键对应不同请求体")
|
||
message = await db.get(Message, record.resource_ids.get("message_id"))
|
||
run = None
|
||
run_id = record.resource_ids.get("run_id")
|
||
if run_id is not None:
|
||
run = await db.get(Run, run_id)
|
||
assert message is not None
|
||
return message, run, True
|
||
|
||
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]
|
||
if forged:
|
||
raise HTTPException(
|
||
status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||
f"系统产生类 block 不允许客户端提交：{', '.join(sorted(set(forged)))}",
|
||
)
|
||
|
||
blocks: list[dict[str, Any]] = []
|
||
parser_version: str | None = None
|
||
for b in body.content:
|
||
if b.type == "markdown":
|
||
# 契约第 4 节：markdown 输入经确定性 parser 规范化为 canonical blocks
|
||
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")
|
||
message = Message(
|
||
conversation_id=conversation.id,
|
||
branch_id=branch.id,
|
||
role="user",
|
||
realm=conversation.realm,
|
||
content=blocks,
|
||
parser_version=parser_version,
|
||
)
|
||
db.add(message)
|
||
await db.flush()
|
||
|
||
run = None
|
||
if body.start_run:
|
||
# Run 创建为 queued，由网关执行器在后台推进至终态（WP-05）
|
||
run = Run(
|
||
personal_domain_id=domain_id,
|
||
conversation_id=conversation.id,
|
||
status="queued",
|
||
)
|
||
db.add(run)
|
||
await db.flush()
|
||
message.run_id = run.id
|
||
|
||
resource_ids: dict[str, Any] = {"message_id": message.id}
|
||
if run is not None:
|
||
resource_ids["run_id"] = run.id
|
||
db.add(
|
||
AuditEvent(
|
||
actor_type="user",
|
||
actor_id=domain_id,
|
||
action="message.created",
|
||
aggregate_type="message",
|
||
aggregate_id=message.id,
|
||
trace_id=trace_id,
|
||
)
|
||
)
|
||
db.add(
|
||
OutboxEvent(
|
||
aggregate_type="message",
|
||
aggregate_id=message.id,
|
||
event_type="message.created",
|
||
payload={"message_id": message.id, "conversation_id": conversation.id},
|
||
)
|
||
)
|
||
db.add(
|
||
IdempotencyRecord(
|
||
key=idempotency_key,
|
||
owner_domain_id=domain_id,
|
||
route=route,
|
||
request_hash=request_hash,
|
||
response_status=201,
|
||
resource_ids=resource_ids,
|
||
)
|
||
)
|
||
await db.flush()
|
||
return message, run, False
|
||
|
||
|
||
|
||
async def list_messages(
|
||
db: AsyncSession,
|
||
domain_id: str,
|
||
conversation_id: str,
|
||
branch_id: str | None,
|
||
limit: int,
|
||
cursor: str | None,
|
||
) -> tuple[list[Message], str | None, bool]:
|
||
"""按创建时间升序的消息列表（opaque cursor 分页，契约第 4.1 节）。"""
|
||
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)
|
||
after_created, after_id = decode_cursor(cursor)
|
||
if after_created is not None:
|
||
# (created_at, id) 复合键分页：严格大于游标位置
|
||
stmt = stmt.where(
|
||
(Message.created_at > after_created)
|
||
| ((Message.created_at == after_created) & (Message.id > after_id))
|
||
)
|
||
stmt = stmt.order_by(Message.created_at.asc(), Message.id.asc()).limit(limit + 1)
|
||
rows = list((await db.scalars(stmt)).all())
|
||
has_more = len(rows) > limit
|
||
page = rows[:limit]
|
||
next_cursor = encode_cursor(page[-1]) if has_more and page else None
|
||
return page, next_cursor, has_more
|
||
|
||
|
||
def encode_cursor(message: Message) -> str:
|
||
import base64
|
||
|
||
raw = f"{message.created_at.isoformat()}|{message.id}"
|
||
return base64.urlsafe_b64encode(raw.encode()).decode()
|
||
|
||
|
||
def decode_cursor(cursor: str | None) -> tuple[Any | None, str]:
|
||
import base64
|
||
|
||
if cursor is None:
|
||
return None, ""
|
||
try:
|
||
raw = base64.urlsafe_b64decode(cursor.encode()).decode()
|
||
created_str, _, mid = raw.partition("|")
|
||
return datetime.fromisoformat(created_str), mid
|
||
except Exception as exc:
|
||
raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "cursor 不合法") from exc
|
||
|
||
|
||
async def list_branches(
|
||
db: AsyncSession, domain_id: str, conversation_id: str
|
||
) -> list[Branch]:
|
||
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)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
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)
|
||
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
|