"""SQLAlchemy-backed task-analysis store.

Mirrors the memory implementation's normalisation: JSON payloads
(chain_snapshot / stages / seat_results / blocks / direction_notes) are stored
as LONGTEXT JSON strings; datetimes are ISO strings in dict form.
"""

from __future__ import annotations

import json
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import delete, func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from deerflow.persistence.task_analysis.base import TaskAnalysisStore
from deerflow.persistence.task_analysis.model import (
    TaskAnalysisBindingRow,
    TaskAnalysisContentVersionRow,
    TaskAnalysisRunMessageRow,
    TaskAnalysisRunRow,
    TaskAnalysisSettingRow,
    TaskAnalysisTaskRow,
    TaskDataBlockDefRow,
    TaskDataBlockRow,
    TaskDataPeriodRow,
    TaskDataSubmissionRow,
)


def _iso(value: datetime | str | None) -> str | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.isoformat()
    return str(value)


def _utcnow() -> datetime:
    return datetime.now(UTC)


def _parse_dt(value: Any) -> datetime | None:
    if value is None or isinstance(value, datetime):
        return value
    try:
        return datetime.fromisoformat(str(value))
    except ValueError:
        return None


def _dump(value: Any) -> str | None:
    return None if value is None else json.dumps(value, ensure_ascii=False)


def _load(value: str | None, default: Any) -> Any:
    if value is None or value == "":
        return default
    try:
        return json.loads(value)
    except (TypeError, ValueError):
        return default


def _task_dict(row: TaskAnalysisTaskRow) -> dict[str, Any]:
    data = row.to_dict()
    data["created_at"] = _iso(row.created_at)
    data["updated_at"] = _iso(row.updated_at)
    return data


def _binding_dict(row: TaskAnalysisBindingRow) -> dict[str, Any]:
    data = row.to_dict()
    data["chain_snapshot"] = _load(row.chain_snapshot, {})
    data["direction_notes"] = _load(row.direction_notes, [])
    for key in ("next_run_at", "last_run_at", "created_at", "updated_at"):
        data[key] = _iso(getattr(row, key))
    return data


def _run_dict(row: TaskAnalysisRunRow) -> dict[str, Any]:
    data = row.to_dict()
    data["scheduled_for"] = _iso(row.scheduled_for)
    data["started_at"] = _iso(row.started_at)
    data["finished_at"] = _iso(row.finished_at)
    data["created_at"] = _iso(row.created_at)
    data["updated_at"] = _iso(row.updated_at)
    for key, default in (
        ("stages_snapshot", []),
        ("stage_goals_snapshot", []),
        ("seat_names_snapshot", {}),
        ("seat_results", {}),
    ):
        data[key] = _load(getattr(row, key), default)
    return data


def _message_dict(row: TaskAnalysisRunMessageRow) -> dict[str, Any]:
    data = row.to_dict()
    data["created_at"] = _iso(row.created_at)
    return data


def _version_dict(row: TaskAnalysisContentVersionRow) -> dict[str, Any]:
    data = row.to_dict()
    data["blocks"] = _load(row.blocks, {})
    data["created_at"] = _iso(row.created_at)
    return data


def _block_def_dict(row: TaskDataBlockDefRow) -> dict[str, Any]:
    data = row.to_dict()
    data["created_at"] = _iso(row.created_at)
    data["updated_at"] = _iso(row.updated_at)
    return data


def _period_dict(row: TaskDataPeriodRow) -> dict[str, Any]:
    data = row.to_dict()
    data["opened_at"] = _iso(row.opened_at)
    data["created_at"] = _iso(row.created_at)
    return data


def _task_block_dict(row: TaskDataBlockRow) -> dict[str, Any]:
    data = row.to_dict()
    data["payload"] = _load(row.payload, {})
    data["updated_at"] = _iso(row.updated_at)
    data["created_at"] = _iso(row.created_at)
    return data


def _submission_dict(row: TaskDataSubmissionRow) -> dict[str, Any]:
    data = row.to_dict()
    data["payload"] = _load(row.payload, {})
    data["created_at"] = _iso(row.created_at)
    return data


class TaskAnalysisRepository(TaskAnalysisStore):
    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self._sf = session_factory

    # ── tasks ───────────────────────────────────────────────────────────────

    async def create_task(self, data: dict[str, Any]) -> dict[str, Any]:
        source = str(data["source_task_id"])
        async with self._sf() as session:
            existing = await session.execute(
                select(TaskAnalysisTaskRow).where(TaskAnalysisTaskRow.source_task_id == source)
            )
            if existing.scalar_one_or_none() is not None:
                raise ValueError(f"source_task_id already exists: {source}")
            row = TaskAnalysisTaskRow(
                id=str(data.get("id") or f"ta-{source}"),
                source_task_id=source,
                name=str(data.get("name") or ""),
                description=data.get("description"),
            )
            session.add(row)
            # 建任务即种 6 条数据块定义（块 key 定死，标题后续可改）。
            from deerflow.persistence.task_analysis.data_blocks import default_block_defs

            for definition in default_block_defs():
                session.add(
                    TaskDataBlockDefRow(
                        id=f"tdbd-{row.id}-{definition['block_key']}",
                        task_id=row.id,
                        block_key=definition["block_key"],
                        title=definition["title"],
                        sort_order=definition["sort_order"],
                    )
                )
            await session.commit()
            await session.refresh(row)
            return _task_dict(row)

    async def get_task(self, task_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisTaskRow, task_id)
            return _task_dict(row) if row else None

    async def get_task_by_source(self, source_task_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisTaskRow).where(TaskAnalysisTaskRow.source_task_id == source_task_id)
            )
            row = result.scalar_one_or_none()
            return _task_dict(row) if row else None

    async def list_tasks(self) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisTaskRow).order_by(TaskAnalysisTaskRow.created_at.desc())
            )
            tasks = [_task_dict(row) for row in result.scalars()]
            bindings = await session.execute(select(TaskAnalysisBindingRow))
            by_task: dict[str, dict[str, Any]] = {}
            for row in bindings.scalars():
                by_task[row.external_task_id] = _binding_dict(row)
            for task in tasks:
                task["binding"] = by_task.get(task["id"])
            return tasks

    async def delete_task_cascade(self, task_id: str) -> bool:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisTaskRow, task_id)
            if row is None:
                return False
            run_ids = (
                (
                    await session.execute(
                        select(TaskAnalysisRunRow.id).where(TaskAnalysisRunRow.external_task_id == task_id)
                    )
                )
                .scalars()
                .all()
            )
            await session.execute(
                delete(TaskAnalysisBindingRow).where(TaskAnalysisBindingRow.external_task_id == task_id)
            )
            await session.execute(
                delete(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
            )
            await session.execute(
                delete(TaskDataPeriodRow).where(TaskDataPeriodRow.task_id == task_id)
            )
            await session.execute(
                delete(TaskDataBlockRow).where(TaskDataBlockRow.task_id == task_id)
            )
            await session.execute(
                delete(TaskDataSubmissionRow).where(TaskDataSubmissionRow.task_id == task_id)
            )
            if run_ids:
                await session.execute(
                    delete(TaskAnalysisRunMessageRow).where(TaskAnalysisRunMessageRow.run_id.in_(run_ids))
                )
                await session.execute(
                    delete(TaskAnalysisContentVersionRow).where(
                        TaskAnalysisContentVersionRow.run_id.in_(run_ids)
                    )
                )
            await session.execute(
                delete(TaskAnalysisRunRow).where(TaskAnalysisRunRow.external_task_id == task_id)
            )
            await session.delete(row)
            await session.commit()
            return True

    # ── bindings ────────────────────────────────────────────────────────────

    async def get_binding(self, external_task_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisBindingRow).where(
                    TaskAnalysisBindingRow.external_task_id == external_task_id
                )
            )
            row = result.scalar_one_or_none()
            return _binding_dict(row) if row else None

    async def get_binding_by_id(self, binding_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisBindingRow, binding_id)
            return _binding_dict(row) if row else None

    async def upsert_binding(self, external_task_id: str, data: dict[str, Any]) -> dict[str, Any]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisBindingRow).where(
                    TaskAnalysisBindingRow.external_task_id == external_task_id
                )
            )
            row = result.scalar_one_or_none()
            if row is None:
                row = TaskAnalysisBindingRow(
                    id=str(data.get("id") or f"tab-{external_task_id}"),
                    external_task_id=external_task_id,
                    chain_id=str(data["chain_id"]),
                    created_by=data.get("created_by"),
                )
                session.add(row)
            row.chain_id = str(data.get("chain_id") or row.chain_id)
            row.chain_snapshot = _dump(data.get("chain_snapshot") or {})
            row.cron_expr = str(data.get("cron_expr") or row.cron_expr or "")
            row.timezone = str(data.get("timezone") or row.timezone or "Asia/Shanghai")
            if "schedule_type" in data:
                row.schedule_type = str(data.get("schedule_type") or "cron")
            row.model = data.get("model")
            if "enabled" in data:
                row.enabled = bool(data["enabled"])
            if "next_run_at" in data:
                row.next_run_at = _parse_dt(data.get("next_run_at"))
            row.coordinator_prompt = data.get("coordinator_prompt")
            if "direction_notes" in data:
                row.direction_notes = _dump(list(data["direction_notes"] or []))
            await session.commit()
            await session.refresh(row)
            return _binding_dict(row)

    async def delete_binding(self, external_task_id: str) -> bool:
        async with self._sf() as session:
            result = await session.execute(
                delete(TaskAnalysisBindingRow).where(
                    TaskAnalysisBindingRow.external_task_id == external_task_id
                )
            )
            await session.commit()
            return bool(result.rowcount)

    async def due_bindings(self, now: datetime, *, limit: int = 20) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisBindingRow)
                .where(
                    TaskAnalysisBindingRow.enabled.is_(True),
                    TaskAnalysisBindingRow.next_run_at.is_not(None),
                    TaskAnalysisBindingRow.next_run_at <= now,
                )
                .order_by(TaskAnalysisBindingRow.next_run_at.asc())
                .limit(limit)
            )
            return [_binding_dict(row) for row in result.scalars()]

    async def update_binding(self, binding_id: str, patch: dict[str, Any]) -> dict[str, Any] | None:
        allowed = {
            "chain_id", "cron_expr", "timezone", "schedule_type", "enabled", "next_run_at",
            "last_run_at", "last_status", "coordinator_prompt", "model",
        }
        values: dict[str, Any] = {k: v for k, v in patch.items() if k in allowed}
        if "next_run_at" in values:
            values["next_run_at"] = _parse_dt(values["next_run_at"])
        if "last_run_at" in values:
            values["last_run_at"] = _parse_dt(values["last_run_at"])
        if "direction_notes" in patch:
            values["direction_notes"] = _dump(list(patch["direction_notes"] or []))
        if not values:
            row = await self._get_binding_row(binding_id)
            return _binding_dict(row) if row else None
        async with self._sf() as session:
            await session.execute(
                update(TaskAnalysisBindingRow).where(TaskAnalysisBindingRow.id == binding_id).values(**values)
            )
            await session.commit()
            row = await session.get(TaskAnalysisBindingRow, binding_id)
            return _binding_dict(row) if row else None

    async def _get_binding_row(self, binding_id: str) -> TaskAnalysisBindingRow | None:
        async with self._sf() as session:
            return await session.get(TaskAnalysisBindingRow, binding_id)

    async def last_bound_chain_id(self) -> str | None:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisBindingRow.chain_id)
                .order_by(TaskAnalysisBindingRow.updated_at.desc())
                .limit(1)
            )
            return result.scalar_one_or_none()

    # ── runs ────────────────────────────────────────────────────────────────

    async def create_run_once(self, data: dict[str, Any]) -> dict[str, Any]:
        binding_id = str(data["binding_id"])
        scheduled_for = _parse_dt(data["scheduled_for"]) or datetime.now(UTC)
        import uuid as _uuid

        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisRunRow).where(
                    TaskAnalysisRunRow.binding_id == binding_id,
                    TaskAnalysisRunRow.scheduled_for == scheduled_for,
                )
            )
            existing = result.scalar_one_or_none()
            if existing is not None:
                return _run_dict(existing)
            row = TaskAnalysisRunRow(
                id=str(data.get("id") or f"tar-{_uuid.uuid4()}"),
                binding_id=binding_id,
                external_task_id=str(data["external_task_id"]),
                trigger=str(data.get("trigger") or "schedule"),
                scheduled_for=scheduled_for,
                status=str(data.get("status") or "pending"),
                stages_snapshot=_dump(data.get("stages_snapshot") or []),
                stage_goals_snapshot=_dump(data.get("stage_goals_snapshot") or []),
                seat_names_snapshot=_dump(data.get("seat_names_snapshot") or {}),
                seat_results=_dump(data.get("seat_results") or {}),
            )
            session.add(row)
            await session.commit()
            await session.refresh(row)
            return _run_dict(row)

    async def get_run(self, run_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisRunRow, run_id)
            return _run_dict(row) if row else None

    async def list_runs(self, external_task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisRunRow)
                .where(TaskAnalysisRunRow.external_task_id == external_task_id)
                .order_by(TaskAnalysisRunRow.started_at.desc())
                .limit(limit)
            )
            return [_run_dict(row) for row in result.scalars()]

    async def update_run(self, run_id: str, patch: dict[str, Any]) -> dict[str, Any] | None:
        allowed = {
            "trigger", "status", "finished_at", "error", "retry_count",
            "stages_snapshot", "stage_goals_snapshot", "seat_names_snapshot", "seat_results",
        }
        values: dict[str, Any] = {}
        for key, value in patch.items():
            if key not in allowed:
                continue
            if key == "finished_at":
                values[key] = _parse_dt(value)
            elif key in {"stages_snapshot", "stage_goals_snapshot", "seat_names_snapshot", "seat_results"}:
                values[key] = _dump(value)
            else:
                values[key] = value
        if not values:
            row = await self._get_run_row(run_id)
            return _run_dict(row) if row else None
        async with self._sf() as session:
            await session.execute(
                update(TaskAnalysisRunRow).where(TaskAnalysisRunRow.id == run_id).values(**values)
            )
            await session.commit()
            row = await session.get(TaskAnalysisRunRow, run_id)
            return _run_dict(row) if row else None

    async def _get_run_row(self, run_id: str) -> TaskAnalysisRunRow | None:
        async with self._sf() as session:
            return await session.get(TaskAnalysisRunRow, run_id)

    async def stuck_runs(self, older_than: datetime, *, limit: int = 20) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisRunRow)
                .where(
                    TaskAnalysisRunRow.status.in_(("pending", "running")),
                    TaskAnalysisRunRow.started_at < older_than,
                )
                .order_by(TaskAnalysisRunRow.started_at.asc())
                .limit(limit)
            )
            return [_run_dict(row) for row in result.scalars()]

    # ── messages ────────────────────────────────────────────────────────────

    async def append_run_messages(self, run_id: str, messages: list[dict[str, Any]]) -> int:
        import uuid as _uuid

        async with self._sf() as session:
            existing = (
                (
                    await session.execute(
                        select(TaskAnalysisRunMessageRow.id).where(
                            TaskAnalysisRunMessageRow.run_id == run_id
                        )
                    )
                )
                .scalars()
                .all()
            )
            base_seq = len(existing)
            rows = [
                TaskAnalysisRunMessageRow(
                    id=str(_uuid.uuid4()),
                    run_id=run_id,
                    seq=base_seq + index,
                    seat_agent_id=str(message.get("seat_agent_id") or ""),
                    stage_index=int(message.get("stage_index") or 0),
                    event_type=str(message.get("event_type") or "llm.ai"),
                    content=message.get("content"),
                )
                for index, message in enumerate(messages)
            ]
            session.add_all(rows)
            await session.commit()
            return len(rows)

    async def list_run_messages(
        self, run_id: str, *, seat: str | None = None, limit: int = 500
    ) -> list[dict[str, Any]]:
        async with self._sf() as session:
            stmt = (
                select(TaskAnalysisRunMessageRow)
                .where(TaskAnalysisRunMessageRow.run_id == run_id)
                .order_by(TaskAnalysisRunMessageRow.seq.asc())
                .limit(limit)
            )
            if seat is not None:
                stmt = stmt.where(TaskAnalysisRunMessageRow.seat_agent_id == seat)
            result = await session.execute(stmt)
            return [_message_dict(row) for row in result.scalars()]

    # ── versions ────────────────────────────────────────────────────────────

    async def create_version(self, data: dict[str, Any]) -> dict[str, Any]:
        import uuid as _uuid

        run_id = str(data["run_id"])
        async with self._sf() as session:
            await session.execute(
                delete(TaskAnalysisContentVersionRow).where(
                    TaskAnalysisContentVersionRow.run_id == run_id
                )
            )
            row = TaskAnalysisContentVersionRow(
                id=str(_uuid.uuid4()),
                external_task_id=str(data["external_task_id"]),
                run_id=run_id,
                blocks=_dump(data.get("blocks") or {}),
            )
            session.add(row)
            await session.commit()
            await session.refresh(row)
            return _version_dict(row)

    async def list_versions(
        self, external_task_id: str, *, limit: int = 50, offset: int = 0
    ) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskAnalysisContentVersionRow)
                .where(TaskAnalysisContentVersionRow.external_task_id == external_task_id)
                .order_by(TaskAnalysisContentVersionRow.created_at.desc())
                .offset(offset)
                .limit(limit)
            )
            return [_version_dict(row) for row in result.scalars()]

    # ── data delivery: block defs / periods / submissions ────────────────────

    async def ensure_block_defs(self, task_id: str) -> list[dict[str, Any]]:
        from deerflow.persistence.task_analysis.data_blocks import default_block_defs

        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
            )
            existing_keys = {row.block_key for row in result.scalars()}
            missing = [
                definition
                for definition in default_block_defs()
                if definition["block_key"] not in existing_keys
            ]
            if not missing:
                return await self._ordered_defs(session, task_id)
            for definition in missing:
                session.add(
                    TaskDataBlockDefRow(
                        id=f"tdbd-{task_id}-{definition['block_key']}",
                        task_id=task_id,
                        block_key=definition["block_key"],
                        title=definition["title"],
                        sort_order=definition["sort_order"],
                    )
                )
            try:
                await session.commit()
            except IntegrityError:
                # 并发补种：另一个请求先种好了，沿用其结果。
                await session.rollback()
            return await self._ordered_defs(session, task_id)

    @staticmethod
    async def _ordered_defs(session: AsyncSession, task_id: str) -> list[dict[str, Any]]:
        result = await session.execute(
            select(TaskDataBlockDefRow)
            .where(TaskDataBlockDefRow.task_id == task_id)
            .order_by(TaskDataBlockDefRow.sort_order.asc(), TaskDataBlockDefRow.block_key.asc())
        )
        return [_block_def_dict(row) for row in result.scalars()]

    async def list_block_defs(self, task_id: str) -> list[dict[str, Any]]:
        async with self._sf() as session:
            return await self._ordered_defs(session, task_id)

    async def update_block_defs(self, task_id: str, updates: list[dict[str, Any]]) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataBlockDefRow).where(TaskDataBlockDefRow.task_id == task_id)
            )
            rows = {row.block_key: row for row in result.scalars()}
            touched = False
            for item in updates:
                block_key = str(item.get("block_key") or "")
                row = rows.get(block_key)
                if row is None:
                    raise ValueError(f"未知数据块：{block_key}")
                patch: dict[str, Any] = {}
                if item.get("title") is not None:
                    patch["title"] = str(item["title"])
                if item.get("sort_order") is not None:
                    patch["sort_order"] = int(item["sort_order"])
                if patch:
                    await session.execute(
                        update(TaskDataBlockDefRow)
                        .where(TaskDataBlockDefRow.id == row.id)
                        .values(**patch)
                    )
                    touched = True
            if touched:
                await session.commit()
            return await self._ordered_defs(session, task_id)

    async def open_period(
        self,
        task_id: str,
        *,
        source: str,
        run_id: str | None = None,
        opened_at: datetime | None = None,
    ) -> dict[str, Any]:
        import uuid as _uuid

        for _attempt in range(3):
            async with self._sf() as session:
                if run_id:
                    existing = await session.execute(
                        select(TaskDataPeriodRow).where(TaskDataPeriodRow.run_id == run_id)
                    )
                    row = existing.scalar_one_or_none()
                    if row is not None:
                        return _period_dict(row)
                max_index = (
                    await session.execute(
                        select(func.max(TaskDataPeriodRow.period_index)).where(
                            TaskDataPeriodRow.task_id == task_id
                        )
                    )
                ).scalar()
                row = TaskDataPeriodRow(
                    id=f"tdp-{_uuid.uuid4()}",
                    task_id=task_id,
                    period_index=(max_index or 0) + 1,
                    opened_at=opened_at or _utcnow(),
                    trigger=str(source or "cron"),
                    run_id=run_id,
                )
                session.add(row)
                try:
                    await session.commit()
                except IntegrityError:
                    # 并发开期撞了期号唯一键（或 run_id 幂等锚点）——重查/重试。
                    await session.rollback()
                    continue
                await session.refresh(row)
                return _period_dict(row)
        raise RuntimeError(f"open_period 连续冲突，task={task_id} source={source}")

    async def get_period(self, task_id: str, period_index: int) -> dict[str, Any] | None:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataPeriodRow).where(
                    TaskDataPeriodRow.task_id == task_id,
                    TaskDataPeriodRow.period_index == period_index,
                )
            )
            row = result.scalar_one_or_none()
            return _period_dict(row) if row else None

    async def current_period(self, task_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataPeriodRow)
                .where(TaskDataPeriodRow.task_id == task_id)
                .order_by(TaskDataPeriodRow.period_index.desc())
                .limit(1)
            )
            row = result.scalar_one_or_none()
            return _period_dict(row) if row else None

    async def list_periods(self, task_id: str, *, limit: int = 500) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataPeriodRow)
                .where(TaskDataPeriodRow.task_id == task_id)
                .order_by(TaskDataPeriodRow.period_index.desc())
                .limit(limit)
            )
            rows = [_period_dict(row) for row in result.scalars()]
            rows.reverse()  # 时间轴语义：按期号升序返回
            return rows

    async def append_submission(
        self,
        task_id: str,
        period_index: int,
        block_key: str,
        payload: dict[str, Any],
        *,
        submitted_by: str | None = None,
    ) -> dict[str, Any]:
        import uuid as _uuid

        async with self._sf() as session:
            base_seq = (
                (
                    await session.execute(
                        select(func.count(TaskDataSubmissionRow.id)).where(
                            TaskDataSubmissionRow.task_id == task_id,
                            TaskDataSubmissionRow.period_index == period_index,
                            TaskDataSubmissionRow.block_key == block_key,
                        )
                    )
                )
                .scalar_one()
            )
            row = TaskDataSubmissionRow(
                id=f"tds-{_uuid.uuid4()}",
                task_id=task_id,
                period_index=period_index,
                block_key=block_key,
                payload=_dump(payload),
                submitted_by=submitted_by,
                seq=int(base_seq),
            )
            session.add(row)
            await session.commit()
            await session.refresh(row)
            return _submission_dict(row)

    async def list_submissions(
        self, task_id: str, period_index: int, block_key: str, *, limit: int = 500
    ) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataSubmissionRow)
                .where(
                    TaskDataSubmissionRow.task_id == task_id,
                    TaskDataSubmissionRow.period_index == period_index,
                    TaskDataSubmissionRow.block_key == block_key,
                )
                .order_by(TaskDataSubmissionRow.seq.asc())
                .limit(limit)
            )
            return [_submission_dict(row) for row in result.scalars()]

    async def upsert_period_block(
        self, task_id: str, period_index: int, block_key: str, payload: dict[str, Any]
    ) -> dict[str, Any]:
        import uuid as _uuid

        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataBlockRow).where(
                    TaskDataBlockRow.task_id == task_id,
                    TaskDataBlockRow.period_index == period_index,
                    TaskDataBlockRow.block_key == block_key,
                )
            )
            row = result.scalar_one_or_none()
            if row is None:
                row = TaskDataBlockRow(
                    id=f"tdb-{_uuid.uuid4()}",
                    task_id=task_id,
                    period_index=period_index,
                    block_key=block_key,
                    payload=_dump(payload),
                )
                session.add(row)
            else:
                row.payload = _dump(payload)
                row.updated_at = _utcnow()
            await session.commit()
            await session.refresh(row)
            return _task_block_dict(row)

    async def list_task_blocks(self, task_id: str, *, limit: int = 2000) -> list[dict[str, Any]]:
        async with self._sf() as session:
            result = await session.execute(
                select(TaskDataBlockRow)
                .where(TaskDataBlockRow.task_id == task_id)
                .order_by(TaskDataBlockRow.period_index.asc(), TaskDataBlockRow.block_key.asc())
                .limit(limit)
            )
            return [_task_block_dict(row) for row in result.scalars()]

    # ── settings ────────────────────────────────────────────────────────────

    async def get_setting(self, key: str) -> str | None:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisSettingRow, key)
            return row.value if row else None

    async def set_setting(self, key: str, value: str) -> None:
        async with self._sf() as session:
            row = await session.get(TaskAnalysisSettingRow, key)
            if row is None:
                session.add(TaskAnalysisSettingRow(key=key, value=value))
            else:
                row.value = value
            await session.commit()