董宏凯andCursor
75963c9ad9
岗位会商写操作引入 command_id 幂等与版本锁，前端支持同键重试；圆桌后台作业改为 DB 租约 dispatcher、去重与草稿 outbox 补偿，并持久化 intent turns。 Co-authored-by: Cursor <cursoragent@cursor.com>
61 lines
3.2 KiB
Python
61 lines
3.2 KiB
Python
"""ORM model for intent-determination turn single-flight (Phase 4).
|
|
|
|
一条 ``intent_turns`` 行 = 第一步「确定意图」对话的**一个回合**对 ``(thread_id,
|
|
client_turn_id)`` 的占用记录。前端为每个回合生成稳定的 ``client_turn_id``，重试 / 双击 /
|
|
网络层重发都复用它；后端用 ``UNIQUE(thread_id, client_turn_id)`` 做跨 worker 单飞：
|
|
|
|
* 同一 ``(thread_id, client_turn_id)`` 至多一条行 —— INSERT 冲突即幂等命中：
|
|
- 已 ``done`` → 原样回放缓存的终态帧，**不再启 run**（杜绝双发意图 / 重复澄清卡）；
|
|
- 在跑且租约未过期 → 该请求是重复的 in-flight 请求，丢弃（原请求仍在流式吐结果）；
|
|
- 在跑但租约过期（worker 崩溃）→ 抢占续跑；
|
|
- ``message_hash`` 不同（同一 client_turn_id 却是不同内容）→ 409 客户端 bug。
|
|
|
|
一致性只来自 DB 唯一键 + 条件 UPDATE（租约 CAS），不依赖进程内 ``RunManager``
|
|
（后者是每进程内存态，多 worker 下看不到彼此的 run，无法单飞）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import String, Text, UniqueConstraint
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.types import BeijingDateTime, PortableLongText
|
|
|
|
|
|
class IntentTurnRow(Base):
|
|
__tablename__ = "intent_turns"
|
|
__table_args__ = (
|
|
# 跨 worker 单飞的根：同一 (thread_id, client_turn_id) 至多一条。INSERT 冲突即幂等。
|
|
UniqueConstraint("thread_id", "client_turn_id", name="uq_intent_turns_thread_client"),
|
|
)
|
|
|
|
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
|
thread_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
|
|
# 前端每回合生成的稳定幂等键；重试 / 双击复用同一个。
|
|
client_turn_id: Mapped[str] = mapped_column(String(64), nullable=False)
|
|
# 本回合 message 的 sha256：检测「同 client_turn_id 但内容变了」的客户端 bug（→ 409）。
|
|
message_hash: Mapped[str] = mapped_column(String(64), nullable=False)
|
|
# running（在跑）/ done（已出终态帧，可回放）/ error。
|
|
status: Mapped[str] = mapped_column(
|
|
String(16), nullable=False, default="running", server_default="running"
|
|
)
|
|
# 终态帧的完整 JSON（done/error 时写入）：重复请求命中时原样回放，不再启 run。
|
|
result_json: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
|
|
error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
# 租约：领走的 worker 标记；过期（worker 崩溃）后可被条件 UPDATE 抢占续跑。
|
|
lease_owner: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
|
lease_until: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
|
|
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC)
|
|
)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
BeijingDateTime(),
|
|
nullable=False,
|
|
default=lambda: datetime.now(UTC),
|
|
onupdate=lambda: datetime.now(UTC),
|
|
)
|