890 lines
38 KiB
Python
890 lines
38 KiB
Python
"""SQLAlchemy-backed roundtable background-job storage.
|
||
|
||
JSON columns (``dispatch_chain`` / ``thread_ids`` / ``chain``) are stored as
|
||
serialized strings in ``PortableLongText`` and round-tripped here with
|
||
``json.dumps`` / ``json.loads`` — same approach as ``ai_writing_sessions`` /
|
||
``roundtable_drafts``.
|
||
|
||
User isolation mirrors ``ai_writing_sessions``: methods take ``user_id`` defaulting
|
||
to ``AUTO`` (resolved from the run context) so backend graph nodes can write without
|
||
threading a user id through, while gateway routes pass it explicitly.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import delete, or_, select, update
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
|
||
from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow
|
||
from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
|
||
|
||
_UNSET = object()
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 终态：列表/清理判定用。
|
||
TERMINAL_STATUSES = ("done", "error", "cancelled")
|
||
# 非终态（活跃）：唯一活跃键去重 / 领取判定用。
|
||
ACTIVE_STATUSES = ("queued", "running", "awaiting_input")
|
||
# cancel 可命中的状态（queued/running/awaiting_input → cancel_requested）。
|
||
CANCELLABLE_STATUSES = ("queued", "running", "awaiting_input")
|
||
|
||
|
||
def compute_active_dedupe_key(
|
||
draft_id: str | None, user_id: str | None, task_id: str | None
|
||
) -> str | None:
|
||
"""活跃去重键：``sha256(scope)``。scope = draft_id + (task_id | user_id)。
|
||
|
||
- task 作业（task_id 非空）：按 (draft_id, task_id) 去重 —— 跨用户共享；
|
||
- 普通作业：按 (draft_id, user_id) 去重；
|
||
- draft_id 为空 → 无法去重，返回 None（直接创建，保持原行为）。
|
||
|
||
该键写入 ``active_dedupe_key`` 唯一索引：作业活跃时占用，进入终态时清 NULL
|
||
（MySQL/SQLite 唯一索引都允许多个 NULL）——「同一 scope 至多一个活跃作业」
|
||
由 DB 唯一键强制，INSERT 冲突即幂等命中。
|
||
"""
|
||
if not draft_id:
|
||
return None
|
||
if task_id:
|
||
scope = f"task:{draft_id}:{task_id}"
|
||
else:
|
||
scope = f"user:{draft_id}:{user_id or ''}"
|
||
return hashlib.sha256(scope.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _loads(value: Any) -> Any:
|
||
if not isinstance(value, str):
|
||
return value
|
||
if not value:
|
||
return None
|
||
try:
|
||
return json.loads(value)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _dumps(value: Any) -> str | None:
|
||
if value is None:
|
||
return None
|
||
return json.dumps(value, ensure_ascii=False)
|
||
|
||
|
||
class RoundtableJobRepository:
|
||
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
|
||
self._sf = session_factory
|
||
|
||
@staticmethod
|
||
def _row_to_dict(row: RoundtableJobRow) -> dict[str, Any]:
|
||
d = row.to_dict()
|
||
for k in ("created_at", "updated_at", "lease_until"):
|
||
if isinstance(d.get(k), datetime):
|
||
d[k] = d[k].isoformat()
|
||
for k in (
|
||
"dispatch_chain",
|
||
"thread_ids",
|
||
"chain",
|
||
"orchestration_plan",
|
||
"dialogues",
|
||
"step3",
|
||
"input_snapshot",
|
||
):
|
||
d[k] = _loads(d.get(k))
|
||
return d
|
||
|
||
# ── create ────────────────────────────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _make_row(
|
||
*,
|
||
id: str,
|
||
resolved_user_id: str | None,
|
||
draft_id: str | None,
|
||
task_id: str | None,
|
||
orchestration_mode: str,
|
||
coordinator_name: str | None,
|
||
thread_ids: dict[str, str] | None,
|
||
chain: dict[str, Any] | None,
|
||
orchestration_plan: dict[str, Any] | None,
|
||
dispatch_chain: list[dict[str, Any]] | None,
|
||
status: str,
|
||
phase: str,
|
||
request_id: str | None = None,
|
||
request_hash: str | None = None,
|
||
active_dedupe_key: str | None = None,
|
||
input_snapshot: dict[str, Any] | None = None,
|
||
) -> RoundtableJobRow:
|
||
now = datetime.now(UTC)
|
||
return RoundtableJobRow(
|
||
id=id,
|
||
draft_id=draft_id,
|
||
user_id=resolved_user_id,
|
||
task_id=(task_id or None),
|
||
request_id=(request_id or None),
|
||
request_hash=(request_hash or None),
|
||
active_dedupe_key=(active_dedupe_key or None),
|
||
input_snapshot=_dumps(input_snapshot),
|
||
status=status,
|
||
phase=phase,
|
||
cycle=0,
|
||
dispatch_chain=_dumps(dispatch_chain),
|
||
active_agent_id=None,
|
||
consensus_percentage=0,
|
||
pending_clarification=None,
|
||
thread_ids=_dumps(thread_ids),
|
||
coordinator_name=coordinator_name,
|
||
orchestration_mode=orchestration_mode,
|
||
chain=_dumps(chain),
|
||
orchestration_plan=_dumps(orchestration_plan),
|
||
error=None,
|
||
attempt=0,
|
||
version=0,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
|
||
async def create(
|
||
self,
|
||
*,
|
||
id: str,
|
||
draft_id: str | None = None,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
task_id: str | None = None,
|
||
orchestration_mode: str = "recommend",
|
||
coordinator_name: str | None = None,
|
||
thread_ids: dict[str, str] | None = None,
|
||
chain: dict[str, Any] | None = None,
|
||
orchestration_plan: dict[str, Any] | None = None,
|
||
dispatch_chain: list[dict[str, Any]] | None = None,
|
||
status: str = "queued",
|
||
phase: str = "initializing",
|
||
) -> dict[str, Any]:
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.create")
|
||
row = self._make_row(
|
||
id=id,
|
||
resolved_user_id=resolved_user_id,
|
||
draft_id=draft_id,
|
||
task_id=task_id,
|
||
orchestration_mode=orchestration_mode,
|
||
coordinator_name=coordinator_name,
|
||
thread_ids=thread_ids,
|
||
chain=chain,
|
||
orchestration_plan=orchestration_plan,
|
||
dispatch_chain=dispatch_chain,
|
||
status=status,
|
||
phase=phase,
|
||
)
|
||
async with self._sf() as session:
|
||
session.add(row)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return self._row_to_dict(row)
|
||
|
||
async def try_create_or_get_active(
|
||
self,
|
||
*,
|
||
draft_id: str | None,
|
||
user_id: str | None | _AutoSentinel,
|
||
task_id: str | None,
|
||
id: str,
|
||
orchestration_mode: str = "recommend",
|
||
coordinator_name: str | None = None,
|
||
thread_ids: dict[str, str] | None = None,
|
||
chain: dict[str, Any] | None = None,
|
||
orchestration_plan: dict[str, Any] | None = None,
|
||
dispatch_chain: list[dict[str, Any]] | None = None,
|
||
status: str = "queued",
|
||
phase: str = "initializing",
|
||
request_id: str | None = None,
|
||
request_hash: str | None = None,
|
||
input_snapshot: dict[str, Any] | None = None,
|
||
) -> tuple[dict[str, Any], bool]:
|
||
"""原子地「创建活跃作业 → 冲突则幂等返回已有活跃作业」（Phase 3 唯一活跃键版）。
|
||
|
||
一致性来自 DB **唯一索引** ``uq_roundtable_jobs_active_dedupe_key``：活跃作业占用
|
||
``active_dedupe_key = sha256(draft_id + task_id|user_id)``，进入终态时清 NULL。于是
|
||
「同一 scope 至多一个活跃作业」由唯一键强制——直接 INSERT，冲突（IntegrityError）即
|
||
说明已有活跃作业，读回它幂等返回（``created=False``）。**不再依赖 GET_LOCK 命名锁**：
|
||
唯一键在 MySQL / PostgreSQL / SQLite 上语义一致，跨进程/跨 worker 天然互斥。
|
||
|
||
极小竞态兜底：INSERT 冲突后读回时若胜者恰好已进终态（键已释放），则重试一次 INSERT。
|
||
|
||
返回 ``(job_dict, created)``。``draft_id`` 为空时无法去重，直接创建（保持原行为）。
|
||
"""
|
||
resolved_user_id = resolve_user_id(
|
||
user_id, method_name="RoundtableJobRepository.try_create_or_get_active"
|
||
)
|
||
dedupe_key = compute_active_dedupe_key(draft_id, resolved_user_id, task_id)
|
||
|
||
def _build_row() -> RoundtableJobRow:
|
||
return self._make_row(
|
||
id=id,
|
||
resolved_user_id=resolved_user_id,
|
||
draft_id=draft_id,
|
||
task_id=task_id,
|
||
orchestration_mode=orchestration_mode,
|
||
coordinator_name=coordinator_name,
|
||
thread_ids=thread_ids,
|
||
chain=chain,
|
||
orchestration_plan=orchestration_plan,
|
||
dispatch_chain=dispatch_chain,
|
||
status=status,
|
||
phase=phase,
|
||
request_id=request_id,
|
||
request_hash=request_hash,
|
||
active_dedupe_key=dedupe_key,
|
||
input_snapshot=input_snapshot,
|
||
)
|
||
|
||
# 无 draft_id → 无法去重，直接创建（保持原有行为）。
|
||
if dedupe_key is None:
|
||
row = _build_row()
|
||
async with self._sf() as session:
|
||
session.add(row)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return self._row_to_dict(row), True
|
||
|
||
for _attempt in range(2):
|
||
try:
|
||
async with self._sf() as session:
|
||
row = _build_row()
|
||
session.add(row)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return self._row_to_dict(row), True
|
||
except IntegrityError:
|
||
pass # 唯一键冲突 → 已有活跃作业（或极小竞态，见下）。
|
||
existing = await self._get_by_dedupe_key(dedupe_key)
|
||
if existing is not None:
|
||
return self._row_to_dict(existing), False
|
||
# 胜者在冲突与我们读回之间进了终态（键已释放）→ 重试 INSERT 可成功。
|
||
|
||
# 两次都冲突且读不回 —— 理论上不可达；保守读回/报错，绝不双建。
|
||
existing = await self._get_by_dedupe_key(dedupe_key)
|
||
if existing is not None:
|
||
return self._row_to_dict(existing), False
|
||
raise RuntimeError(
|
||
f"roundtable job dedupe conflict but no active row found for key {dedupe_key}"
|
||
)
|
||
|
||
async def _get_by_dedupe_key(self, dedupe_key: str) -> RoundtableJobRow | None:
|
||
"""按唯一活跃键读回占用该键的（活跃）作业行；无则 None。"""
|
||
stmt = select(RoundtableJobRow).where(
|
||
RoundtableJobRow.active_dedupe_key == dedupe_key
|
||
)
|
||
async with self._sf() as session:
|
||
return (await session.execute(stmt)).scalars().first()
|
||
|
||
# ── update progress ─────────────────────────────────────────────────────
|
||
|
||
async def update_progress(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
status=_UNSET,
|
||
phase=_UNSET,
|
||
cycle=_UNSET,
|
||
dispatch_chain=_UNSET,
|
||
active_agent_id=_UNSET,
|
||
consensus_percentage=_UNSET,
|
||
pending_clarification=_UNSET,
|
||
thread_ids=_UNSET,
|
||
coordinator_name=_UNSET,
|
||
dialogues=_UNSET,
|
||
step3=_UNSET,
|
||
error=_UNSET,
|
||
lease_owner: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""更新进度字段。
|
||
|
||
**租约门（Phase 3）**：传入 ``lease_owner`` 时走**原子条件 UPDATE**——要求 DB 里
|
||
``lease_owner`` 匹配且状态不在 ``cancel_requested/cancelled``，否则返回 None（写被
|
||
拒）。这保证：① 只有持租者能推进度；② 一旦 cancel 落库（cancel_requested），迟到的
|
||
进度写不会覆盖取消意图；③ 多 worker 下非持租者的过期写直接失效。一致性来自 DB 条件
|
||
UPDATE，而非读后判断。
|
||
|
||
**终态处理**：写入 status ∈ 终态（done/error/cancelled）时，同时清空
|
||
``active_dedupe_key``（释放唯一活跃键，允许该 scope 再建作业）与租约字段。
|
||
|
||
不传 ``lease_owner`` 时保持原「按 user 分权 + 读改写」行为（向后兼容旧调用/测试）。
|
||
"""
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.update_progress")
|
||
|
||
# ── 租约门：原子条件 UPDATE ──
|
||
if lease_owner is not None:
|
||
values: dict[str, Any] = {"updated_at": datetime.now(UTC)}
|
||
if status is not _UNSET:
|
||
values["status"] = status
|
||
if phase is not _UNSET:
|
||
values["phase"] = phase
|
||
if cycle is not _UNSET:
|
||
values["cycle"] = int(cycle)
|
||
if dispatch_chain is not _UNSET:
|
||
values["dispatch_chain"] = _dumps(dispatch_chain)
|
||
if active_agent_id is not _UNSET:
|
||
values["active_agent_id"] = active_agent_id
|
||
if consensus_percentage is not _UNSET:
|
||
values["consensus_percentage"] = int(consensus_percentage)
|
||
if pending_clarification is not _UNSET:
|
||
values["pending_clarification"] = pending_clarification
|
||
if thread_ids is not _UNSET:
|
||
values["thread_ids"] = _dumps(thread_ids)
|
||
if coordinator_name is not _UNSET:
|
||
values["coordinator_name"] = coordinator_name
|
||
if dialogues is not _UNSET:
|
||
values["dialogues"] = _dumps(dialogues)
|
||
if step3 is not _UNSET:
|
||
values["step3"] = _dumps(step3)
|
||
if error is not _UNSET:
|
||
values["error"] = error
|
||
# 终态：释放唯一活跃键 + 清租约。
|
||
if status is not _UNSET and status in TERMINAL_STATUSES:
|
||
values["active_dedupe_key"] = None
|
||
values["lease_owner"] = None
|
||
values["lease_until"] = None
|
||
# awaiting_input：任务已暂停、执行协程退出 —— 必须释放租约，否则 resume 前
|
||
# 租约未过期没人能重新领取。活跃键保留（作业仍未终态，继续占着 scope）。
|
||
elif status is not _UNSET and status == "awaiting_input":
|
||
values["lease_owner"] = None
|
||
values["lease_until"] = None
|
||
values["version"] = RoundtableJobRow.version + 1
|
||
|
||
where = [
|
||
RoundtableJobRow.id == job_id,
|
||
RoundtableJobRow.lease_owner == lease_owner,
|
||
RoundtableJobRow.status.not_in(["cancel_requested", "cancelled"]),
|
||
]
|
||
if resolved_user_id is not None:
|
||
where.append(RoundtableJobRow.user_id == resolved_user_id)
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow).where(*where).values(**values)
|
||
)
|
||
if result.rowcount == 0:
|
||
return None
|
||
await session.commit()
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
return self._row_to_dict(row) if row is not None else None
|
||
|
||
# ── 兼容路径：按 user 分权 + 读改写（不持租约的旧调用）──
|
||
async with self._sf() as session:
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
if row is None:
|
||
return None
|
||
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
||
return None
|
||
if status is not _UNSET:
|
||
row.status = status
|
||
if phase is not _UNSET:
|
||
row.phase = phase
|
||
if cycle is not _UNSET:
|
||
row.cycle = int(cycle)
|
||
if dispatch_chain is not _UNSET:
|
||
row.dispatch_chain = _dumps(dispatch_chain)
|
||
if active_agent_id is not _UNSET:
|
||
row.active_agent_id = active_agent_id
|
||
if consensus_percentage is not _UNSET:
|
||
row.consensus_percentage = int(consensus_percentage)
|
||
if pending_clarification is not _UNSET:
|
||
row.pending_clarification = pending_clarification
|
||
if thread_ids is not _UNSET:
|
||
row.thread_ids = _dumps(thread_ids)
|
||
if coordinator_name is not _UNSET:
|
||
row.coordinator_name = coordinator_name
|
||
if dialogues is not _UNSET:
|
||
row.dialogues = _dumps(dialogues)
|
||
if step3 is not _UNSET:
|
||
row.step3 = _dumps(step3)
|
||
if error is not _UNSET:
|
||
row.error = error
|
||
# 终态同样要释放唯一活跃键（兼容路径没有租约也要清键）。
|
||
if status is not _UNSET and status in TERMINAL_STATUSES:
|
||
row.active_dedupe_key = None
|
||
row.lease_owner = None
|
||
row.lease_until = None
|
||
# awaiting_input 同样释放租约（与租约门路径一致）。
|
||
elif status is not _UNSET and status == "awaiting_input":
|
||
row.lease_owner = None
|
||
row.lease_until = None
|
||
row.updated_at = datetime.now(UTC)
|
||
await session.commit()
|
||
await session.refresh(row)
|
||
return self._row_to_dict(row)
|
||
|
||
async def mark_draft_persist_pending(self, job_id: str) -> bool:
|
||
"""把草稿回写 pending 标记置 1（``_write_run_to_draft`` 重试耗尽时调用）。
|
||
|
||
一致性来自 DB：绝不退化为 last-write-wins 盲写，而是留下可被 outbox 补偿器
|
||
（Phase 3）发现的标记。返回是否命中了作业行。
|
||
"""
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow)
|
||
.where(RoundtableJobRow.id == job_id)
|
||
.values(draft_persist_pending=True, updated_at=datetime.now(UTC))
|
||
)
|
||
if result.rowcount == 0:
|
||
return False
|
||
await session.commit()
|
||
return True
|
||
|
||
# ── Phase 3：租约领取 / 续租 / 释放（dispatcher） ────────────────────────
|
||
|
||
async def list_claimable(self, *, now: datetime, limit: int = 20) -> list[str]:
|
||
"""列出可领取的作业 id：``queued``（新作业/续跑入队）或 ``running`` 但租约已过期
|
||
（持租 worker 崩溃/卡死）。按 updated_at 升序（先来先领），最多 limit 个。
|
||
|
||
这只是「候选列表」；真正的领取靠 {@link claim_job} 的条件 UPDATE —— 多 worker 并发
|
||
扫到同一候选时，只有一个 UPDATE 能命中（一致性来自 DB）。
|
||
"""
|
||
stmt = (
|
||
select(RoundtableJobRow.id)
|
||
.where(
|
||
or_(
|
||
RoundtableJobRow.status == "queued",
|
||
(RoundtableJobRow.status == "running")
|
||
& (RoundtableJobRow.lease_until.is_not(None))
|
||
& (RoundtableJobRow.lease_until < now),
|
||
)
|
||
)
|
||
.order_by(RoundtableJobRow.updated_at.asc())
|
||
.limit(limit)
|
||
)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
return [r[0] for r in result.all()]
|
||
|
||
async def list_cancel_reapable(self, *, now: datetime, limit: int = 20) -> list[str]:
|
||
"""列出可以由 dispatcher 收口的取消请求。
|
||
|
||
``cancel_requested`` 是取消的权威意图，但持租 worker 可能在看门狗收口前崩溃。
|
||
仅当作业没有租约或租约已过期时才允许 dispatcher 无租约收口；仍持有有效租约的
|
||
作业继续由原 worker 的看门狗处理，避免抢占仍在正常退出的执行器。
|
||
"""
|
||
stmt = (
|
||
select(RoundtableJobRow.id)
|
||
.where(
|
||
RoundtableJobRow.status == "cancel_requested",
|
||
or_(
|
||
RoundtableJobRow.lease_owner.is_(None),
|
||
RoundtableJobRow.lease_until.is_(None),
|
||
RoundtableJobRow.lease_until < now,
|
||
),
|
||
)
|
||
.order_by(RoundtableJobRow.updated_at.asc())
|
||
.limit(limit)
|
||
)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
return [r[0] for r in result.all()]
|
||
|
||
async def claim_job(
|
||
self, job_id: str, *, lease_owner: str, lease_until: datetime
|
||
) -> dict[str, Any] | None:
|
||
"""原子条件 UPDATE 领取一个作业：仅当它仍是可领取状态时才成功。
|
||
|
||
成功条件（DB 保证，跨 worker 互斥）：``status='queued'``，或 ``status='running'``
|
||
且租约已过期。成功则置 ``status='running'``、写 ``lease_owner/lease_until``、
|
||
``attempt+1``、``version+1``，返回行；否则返回 None（别的 worker 先领了 / 状态已变）。
|
||
"""
|
||
values = {
|
||
"status": "running",
|
||
"lease_owner": lease_owner,
|
||
"lease_until": lease_until,
|
||
"attempt": RoundtableJobRow.attempt + 1,
|
||
"version": RoundtableJobRow.version + 1,
|
||
"updated_at": datetime.now(UTC),
|
||
}
|
||
where = [
|
||
RoundtableJobRow.id == job_id,
|
||
or_(
|
||
RoundtableJobRow.status == "queued",
|
||
(RoundtableJobRow.status == "running")
|
||
& (RoundtableJobRow.lease_until.is_not(None))
|
||
& (RoundtableJobRow.lease_until < datetime.now(UTC)),
|
||
),
|
||
]
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow).where(*where).values(**values)
|
||
)
|
||
if result.rowcount == 0:
|
||
return None
|
||
await session.commit()
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
if row is None:
|
||
return None
|
||
# Phase 7 observability: attempt >= 2 means this job was previously
|
||
# claimed by another worker whose lease expired (crash / OOM / GC
|
||
# stall) and is now being reclaimed. This is the key signal for
|
||
# diagnosing worker instability under load.
|
||
if row.attempt >= 2:
|
||
logger.warning(
|
||
"roundtable_jobs lease takeover: job=%s reclaimed by %s "
|
||
"(attempt=%d — previous worker's lease expired)",
|
||
job_id, lease_owner, row.attempt,
|
||
)
|
||
return self._row_to_dict(row)
|
||
|
||
async def renew_lease(
|
||
self, job_id: str, *, lease_owner: str, lease_until: datetime
|
||
) -> bool:
|
||
"""心跳续租：仅持租者可延长 ``lease_until``。返回是否续租成功。"""
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow)
|
||
.where(
|
||
RoundtableJobRow.id == job_id,
|
||
RoundtableJobRow.lease_owner == lease_owner,
|
||
RoundtableJobRow.status == "running",
|
||
)
|
||
.values(lease_until=lease_until, updated_at=datetime.now(UTC))
|
||
)
|
||
if result.rowcount == 0:
|
||
return False
|
||
await session.commit()
|
||
return True
|
||
|
||
async def release_lease(self, job_id: str, *, lease_owner: str | None = None) -> bool:
|
||
"""释放租约（进入 awaiting_input 等「暂停但非终态」时调用）。
|
||
|
||
置 ``lease_owner/lease_until`` 为 NULL，使该作业可被任意 worker 重新领取（续跑）。
|
||
传入 ``lease_owner`` 时仅当匹配才释放（防止误放他人的租约）。
|
||
"""
|
||
where = [RoundtableJobRow.id == job_id]
|
||
if lease_owner is not None:
|
||
where.append(RoundtableJobRow.lease_owner == lease_owner)
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow)
|
||
.where(*where)
|
||
.values(lease_owner=None, lease_until=None, updated_at=datetime.now(UTC))
|
||
)
|
||
if result.rowcount == 0:
|
||
return False
|
||
await session.commit()
|
||
return True
|
||
|
||
# ── Phase 3：resume / cancel 条件状态机 ──────────────────────────────────
|
||
|
||
async def queue_for_resume(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
input_snapshot: dict[str, Any] | None = None,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
task_id: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""原子条件 UPDATE：``awaiting_input → queued``（续跑入队），仅第一个成功方有效。
|
||
|
||
多 worker / 重复点击下可能有多个 resume 请求并发；条件 UPDATE（WHERE status=
|
||
'awaiting_input'）保证只有一个能把它翻成 queued，其余返回 None（路由据此拒绝）。
|
||
成功后清租约（交给 dispatcher 重新领取），并把含续跑语义的新 snapshot 写回。
|
||
"""
|
||
resolved_user_id = resolve_user_id(
|
||
user_id, method_name="RoundtableJobRepository.queue_for_resume"
|
||
)
|
||
values: dict[str, Any] = {
|
||
"status": "queued",
|
||
"phase": "initializing",
|
||
"pending_clarification": None,
|
||
"lease_owner": None,
|
||
"lease_until": None,
|
||
"version": RoundtableJobRow.version + 1,
|
||
"updated_at": datetime.now(UTC),
|
||
}
|
||
if input_snapshot is not None:
|
||
values["input_snapshot"] = _dumps(input_snapshot)
|
||
where = [RoundtableJobRow.id == job_id, RoundtableJobRow.status == "awaiting_input"]
|
||
if task_id is not None:
|
||
where.append(RoundtableJobRow.task_id == task_id)
|
||
elif resolved_user_id is not None:
|
||
where.append(RoundtableJobRow.user_id == resolved_user_id)
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow).where(*where).values(**values)
|
||
)
|
||
if result.rowcount == 0:
|
||
return None
|
||
await session.commit()
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
return self._row_to_dict(row) if row is not None else None
|
||
|
||
async def request_cancel(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
task_id: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""原子条件 UPDATE：可取消态（queued/running/awaiting_input）→ ``cancel_requested``。
|
||
|
||
cancel 的**第一步**（权威意图落库）。成功后路由再 best-effort 取消本进程任务；
|
||
实际终态 ``cancelled`` 由持租者看门狗（检测到 cancel_requested）或 ``finalize_cancel``
|
||
写入 —— 保证「已取消」终态由持租者收口，避免竞态。重复 cancel 幂等（已 cancel_requested/
|
||
cancelled 时返回 None，路由据此回 409 或返回现状）。
|
||
"""
|
||
resolved_user_id = resolve_user_id(
|
||
user_id, method_name="RoundtableJobRepository.request_cancel"
|
||
)
|
||
values: dict[str, Any] = {
|
||
"status": "cancel_requested",
|
||
"version": RoundtableJobRow.version + 1,
|
||
"updated_at": datetime.now(UTC),
|
||
}
|
||
where = [
|
||
RoundtableJobRow.id == job_id,
|
||
RoundtableJobRow.status.in_(list(CANCELLABLE_STATUSES)),
|
||
]
|
||
if task_id is not None:
|
||
where.append(RoundtableJobRow.task_id == task_id)
|
||
elif resolved_user_id is not None:
|
||
where.append(RoundtableJobRow.user_id == resolved_user_id)
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow).where(*where).values(**values)
|
||
)
|
||
if result.rowcount == 0:
|
||
return None
|
||
await session.commit()
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
return self._row_to_dict(row) if row is not None else None
|
||
|
||
async def finalize_cancel(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
lease_owner: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""把 ``cancel_requested``（或孤儿 queued/running）收口为终态 ``cancelled``。
|
||
|
||
由持租者看门狗（检测 cancel_requested 后）或路由（作业尚在 queued、无持租者时）调用。
|
||
条件 UPDATE：仅当 ``status='cancel_requested'`` 时翻转；同时清唯一活跃键 + 租约。
|
||
传入 ``lease_owner`` 时要求租约匹配（持租者收口），否则不加租约条件（queued 无租约）。
|
||
"""
|
||
values: dict[str, Any] = {
|
||
"status": "cancelled",
|
||
"active_dedupe_key": None,
|
||
"lease_owner": None,
|
||
"lease_until": None,
|
||
"version": RoundtableJobRow.version + 1,
|
||
"updated_at": datetime.now(UTC),
|
||
}
|
||
where = [RoundtableJobRow.id == job_id, RoundtableJobRow.status == "cancel_requested"]
|
||
if lease_owner is not None:
|
||
where.append(RoundtableJobRow.lease_owner == lease_owner)
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow).where(*where).values(**values)
|
||
)
|
||
if result.rowcount == 0:
|
||
return None
|
||
await session.commit()
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
return self._row_to_dict(row) if row is not None else None
|
||
|
||
# ── Phase 3：outbox 补偿器（draft_persist_pending 投影） ─────────────────
|
||
|
||
async def list_draft_persist_pending(self, *, limit: int = 50) -> list[dict[str, Any]]:
|
||
"""列出 ``draft_persist_pending=1`` 的作业行（补偿器待投影清单）。"""
|
||
stmt = (
|
||
select(RoundtableJobRow)
|
||
.where(RoundtableJobRow.draft_persist_pending.is_(True))
|
||
.order_by(RoundtableJobRow.updated_at.asc())
|
||
.limit(limit)
|
||
)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
return [self._row_to_dict(r) for r in result.scalars()]
|
||
|
||
async def clear_draft_persist_pending(self, job_id: str) -> bool:
|
||
"""投影成功后清 ``draft_persist_pending`` 标记。返回是否命中。"""
|
||
async with self._sf() as session:
|
||
result = await session.execute(
|
||
update(RoundtableJobRow)
|
||
.where(RoundtableJobRow.id == job_id)
|
||
.values(draft_persist_pending=False, updated_at=datetime.now(UTC))
|
||
)
|
||
if result.rowcount == 0:
|
||
return False
|
||
await session.commit()
|
||
return True
|
||
|
||
# ── read ────────────────────────────────────────────────────────────────
|
||
|
||
async def get(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
task_id: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
# task 作业按 task_id 共享、**不按 user 分权**：给定 task_id 时只校验作业属于该
|
||
# task（忽略 user_id）。否则走原「按 user 分权」逻辑。
|
||
if task_id is not None:
|
||
async with self._sf() as session:
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
if row is None or row.task_id != task_id:
|
||
return None
|
||
return self._row_to_dict(row)
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.get")
|
||
async with self._sf() as session:
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
if row is None:
|
||
return None
|
||
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
||
return None
|
||
return self._row_to_dict(row)
|
||
|
||
async def get_by_draft(
|
||
self,
|
||
draft_id: str,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
task_id: str | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""Return the most-recently-updated job for a draft (or None).
|
||
|
||
``task_id`` set → task 作业查询：按 ``draft_id`` + ``task_id`` 过滤，**不按 user
|
||
分权**（同一 task 的作业跨用户共享）。否则按 user 分权。
|
||
"""
|
||
stmt = select(RoundtableJobRow).where(RoundtableJobRow.draft_id == draft_id)
|
||
if task_id is not None:
|
||
stmt = stmt.where(RoundtableJobRow.task_id == task_id)
|
||
else:
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.get_by_draft")
|
||
if resolved_user_id is not None:
|
||
stmt = stmt.where(RoundtableJobRow.user_id == resolved_user_id)
|
||
stmt = stmt.order_by(RoundtableJobRow.updated_at.desc()).limit(1)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
row = result.scalars().first()
|
||
return self._row_to_dict(row) if row is not None else None
|
||
|
||
async def list_by_user(
|
||
self,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
limit: int = 50,
|
||
) -> list[dict[str, Any]]:
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.list_by_user")
|
||
stmt = select(RoundtableJobRow)
|
||
if resolved_user_id is not None:
|
||
stmt = stmt.where(RoundtableJobRow.user_id == resolved_user_id)
|
||
stmt = stmt.order_by(RoundtableJobRow.updated_at.desc()).limit(limit)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
return [self._row_to_dict(r) for r in result.scalars()]
|
||
|
||
async def status_map_by_user(
|
||
self,
|
||
user_id: str,
|
||
*,
|
||
limit: int = 200,
|
||
) -> dict[str, dict[str, str]]:
|
||
"""``{draft_id: {"jobId": ..., "status": ...}}`` for the latest job per draft.
|
||
|
||
Drives the history dropdown's per-draft status badge (Phase 6). Walks jobs
|
||
newest-first and keeps the first (latest) one seen for each draft_id.
|
||
"""
|
||
stmt = (
|
||
select(RoundtableJobRow.id, RoundtableJobRow.draft_id, RoundtableJobRow.status)
|
||
.where(RoundtableJobRow.user_id == user_id)
|
||
.where(RoundtableJobRow.draft_id.is_not(None))
|
||
.order_by(RoundtableJobRow.updated_at.desc())
|
||
.limit(limit)
|
||
)
|
||
out: dict[str, dict[str, str]] = {}
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
for job_id, draft_id, status in result.all():
|
||
if draft_id is None or draft_id in out:
|
||
continue
|
||
out[draft_id] = {"jobId": job_id, "status": status}
|
||
return out
|
||
|
||
async def status_map_by_drafts(
|
||
self,
|
||
draft_ids: list[str],
|
||
*,
|
||
limit: int = 400,
|
||
) -> dict[str, dict[str, str]]:
|
||
"""``{draft_id: {"jobId": ..., "status": ...}}`` for the latest job per draft.
|
||
|
||
Like :meth:`status_map_by_user` but filtered by an explicit draft-id list
|
||
instead of by owner — used by the taskId-mode history dropdown, where a
|
||
task's drafts may span multiple users (task-bound drafts are shared).
|
||
"""
|
||
ids = [d for d in draft_ids if d]
|
||
if not ids:
|
||
return {}
|
||
stmt = (
|
||
select(RoundtableJobRow.id, RoundtableJobRow.draft_id, RoundtableJobRow.status)
|
||
.where(RoundtableJobRow.draft_id.in_(ids))
|
||
.order_by(RoundtableJobRow.updated_at.desc())
|
||
.limit(limit)
|
||
)
|
||
out: dict[str, dict[str, str]] = {}
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
for job_id, draft_id, status in result.all():
|
||
if draft_id is None or draft_id in out:
|
||
continue
|
||
out[draft_id] = {"jobId": job_id, "status": status}
|
||
return out
|
||
|
||
# ── delete / cleanup ─────────────────────────────────────────────────────
|
||
|
||
async def delete(
|
||
self,
|
||
job_id: str,
|
||
*,
|
||
user_id: str | None | _AutoSentinel = AUTO,
|
||
) -> bool:
|
||
resolved_user_id = resolve_user_id(user_id, method_name="RoundtableJobRepository.delete")
|
||
async with self._sf() as session:
|
||
row = await session.get(RoundtableJobRow, job_id)
|
||
if row is None:
|
||
return False
|
||
if resolved_user_id is not None and row.user_id != resolved_user_id:
|
||
return False
|
||
await session.delete(row)
|
||
await session.commit()
|
||
return True
|
||
|
||
async def list_older_than(
|
||
self,
|
||
cutoff: datetime,
|
||
*,
|
||
only_finished: bool = False,
|
||
limit: int = 500,
|
||
) -> list[str]:
|
||
"""列出 updated_at 早于 cutoff 的作业 id（最多 limit 个），供清理任务用。
|
||
|
||
only_finished=True 时只列终态（done/error/cancelled），避免清掉还在跑的作业。
|
||
"""
|
||
stmt = select(RoundtableJobRow.id).where(RoundtableJobRow.updated_at < cutoff)
|
||
if only_finished:
|
||
stmt = stmt.where(RoundtableJobRow.status.in_(TERMINAL_STATUSES))
|
||
stmt = stmt.order_by(RoundtableJobRow.updated_at.asc()).limit(limit)
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
return [row[0] for row in result.all()]
|
||
|
||
async def delete_by_ids(self, job_ids: list[str]) -> int:
|
||
if not job_ids:
|
||
return 0
|
||
stmt = delete(RoundtableJobRow).where(RoundtableJobRow.id.in_(job_ids))
|
||
async with self._sf() as session:
|
||
result = await session.execute(stmt)
|
||
await session.commit()
|
||
return int(result.rowcount or 0)
|