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