Blame for collector/core/task_record.py

"""
Task record persistence — data/tasks.json is the single source of truth.

- 进度分账（多电脑协同，后端聚合口径）：
    local_count    本机累计完成段数（increment_task_completed 加它）
    synced_count   后端已确认的本机段数（水位，只前进不后退）
    backend_count  最近一次已知后端全局数（merge/flush 用后端值无条件替换）
    completed_count = backend_count + (local_count - synced_count)
      —— 派生显示值，字段名保留使 UI 零改动，所有写路径经 _recount() 重算。
- 迁移回填：旧数据没有上述字段时，local/synced/backend 一次性回填为
  旧 completed_count（目录扫描兜底）——历史不补报、显示连续；
  此后以持久化值为准，上传后删本地文件也不影响进度。
- Task metadata (id, name, description, total_required, etc.) is persisted.
- Backend tasks are merged into the local record, keeping local-only tasks.

- 本机任务（source="local"）：
    任务页「＋ 新建任务」创建，id 前缀 local-，不进后端任务表、不参与进度上报。
    * **仅创建者可见**：created_by 记创建时的身份（"guest" 或用户名），
      filter_by_identity() 对本机任务按 created_by 过滤（缺失时对所有人可见，
      兼容手工写的数据）；assigned_user 恒为 None，仍然算「公共任务」，
      这样 merge_backend_tasks 的账号撤销清除不会删到它（另有显式豁免）。
    * **进度只在本机**：local_count 累计，synced_count/backend_count 恒 0，
      pending_sync_tasks() 跳过本机任务 —— 服务端没有对应任务，上报必被拒。
    * **删除是真删记录**：remove_local_task() 整条移除（后端不会再推送它，
      不需要 hidden 墓碑）；服务端任务仍走 mark_hidden()。
    * **名字即目录即上传项目名**：只允许字母/数字/汉字/下划线/连字符，
      保证 <RECORDING_DIR>/<name> 与 task_tag(name) 完全一致。
"""

from __future__ import annotations
import json
import os
import threading
import uuid
from datetime import datetime, timezone

from config import settings

_TASKS_FILE = os.path.join(settings.DATA_DIR, "tasks.json")

# ── 本机任务常量 ──────────────────────────────────────
LOCAL_TASK_PREFIX = "local-"
MAX_TASK_NAME_LEN = 64
MAX_TASK_DESCRIPTION_LEN = 500
MAX_TOTAL_REQUIRED = 100000

# Windows 保留设备名（不能作为目录名；大小写不敏感）
_WINDOWS_RESERVED_NAMES = frozenset(
    ["CON", "PRN", "AUX", "NUL"]
    + [f"COM{i}" for i in range(1, 10)]
    + [f"LPT{i}" for i in range(1, 10)]
)


class TaskValidationError(ValueError):
    """本机任务创建/改名被拒。code 供 UI 本地化，不把中文写进核心层。"""

    def __init__(self, code: str, detail: str = ""):
        super().__init__(code)
        self.code = code
        self.detail = detail

    def __str__(self) -> str:  # 日志/调试可读
        return f"{self.code}: {self.detail}" if self.detail else self.code


_lock = threading.RLock()  # reentrant — merge_backend_tasks calls save_tasks/load_tasks

_SEED_TASKS = [
    {
        "id": "task_001",
        "name": "cylinder_grasping",
        "description": "Grasp cylinder parts with right hand, record 16x16 tactile sensor data.",
        "status": "pending",
        "total_required": 1000,
        "assigned_at": "2026-08-07T09:00:00Z",
        "params": {"object": "cylinder", "hand": "right"},
    },
    {
        "id": "task_002",
        "name": "cube_placement",
        "description": "Place cube objects from point A to point B using both hands.",
        "status": "pending",
        "total_required": 500,
        "assigned_at": "2026-08-06T14:30:00Z",
        "params": {"object": "cube", "hand": "both"},
    },
    {
        "id": "task_003",
        "name": "valve_rotation",
        "description": "Rotate circular valve 90 degrees with single hand, record torque and tactile feedback.",
        "status": "pending",
        "total_required": 800,
        "assigned_at": "2026-08-05T11:00:00Z",
        "params": {"object": "valve", "hand": "right", "angle": 90},
    },
]


# ═══════════════════════════════════════════════════════════
#  Public API
# ═══════════════════════════════════════════════════════════

def load_tasks(identity: str | None = None) -> list[dict]:
    """Load all tasks. Filters out hidden tasks.

    completed_count 以持久化值为准（increment_task_completed 累计）；
    文件中缺该字段（旧版本数据）时一次性按目录扫描回填并写回，
    此后删会话文件不再影响进度。

    Status is auto-computed: pending(0%), in_progress(>0%), completed(100%).

    排序: 按 assigned_at 从新到旧；null/空/解析失败排最后。

    identity（新增，默认 None 兼容旧调用）:
      None   → 返回全部任务（含他人创建的本机任务，仅供内部/统计调用）
      "guest" → 仅公共任务（无 assigned_user 字段）+ 游客创建的本机任务
      用户名  → 公共任务 + assigned_user == 该用户的任务
               + 该用户创建的本机任务

    注意: 除迁移回填外不写回 tasks.json —— hidden 任务必须保留在文件中，
    否则后端再次推送时无法识别为已删除，任务会死而复生。
    """
    with _lock:
        tasks = _read_raw()
    need_backfill = False
    for t in tasks:
        if _ensure_sync_fields(t):
            need_backfill = True
        _recount(t)
        t["status"] = _compute_status(t)
    if need_backfill:
        with _lock:
            _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})
    visible = [t for t in tasks if not t.get("hidden")]
    visible = filter_by_identity(visible, identity)
    visible.sort(key=_assigned_at_key, reverse=True)  # 新→旧；null/空/非法排最后
    return visible


def filter_by_identity(tasks: list[dict], identity: str | None = None) -> list[dict]:
    """按身份过滤任务列表（纯函数，供 UI 对已有列表即时过滤）。

    identity 语义与 load_tasks 相同；返回新列表，不修改原列表。
    本机任务按 **created_by** 过滤（仅创建者可见）；assigned_user 语义只
    用于服务端任务（本机任务 assigned_user 恒为 None）。created_by 缺失的
    本机任务视为「无主」，对所有身份可见（兼容手工写的 tasks.json）。
    """
    if identity is None:
        return list(tasks)
    out: list[dict] = []
    for t in tasks:
        if is_local_task(t):
            if _local_visible(t, identity):
                out.append(t)
            continue
        if identity == "guest":
            if not _assigned_user(t):
                out.append(t)
        elif not _assigned_user(t) or _assigned_user(t) == identity:
            out.append(t)
    return out


def save_tasks(tasks: list[dict]):
    """Persist task metadata，含 completed_count（录制完成次数持久化权威值）。"""
    clean = [dict(t) for t in tasks]
    with _lock:
        _write({"tasks": clean, "updated_at": datetime.now().isoformat()})


# ═══════════════════════════════════════════════════════════
#  本机任务（采集端自建）—— 创建 / 改名 / 删除 / 校验
# ═══════════════════════════════════════════════════════════

def is_local_task(task: dict) -> bool:
    """是否采集端自建任务。source 缺失时按 id 前缀兜底（双保险）。"""
    if not isinstance(task, dict):
        return False
    if task.get("source") == "local":
        return True
    return str(_tid(task) or "").startswith(LOCAL_TASK_PREFIX)


def new_local_task_id() -> str:
    """本机任务 id：local-YYYYmmdd-HHMMSS-<4hex>，与后端 UUID 不会相撞。"""
    return "local-{}-{:04x}".format(
        datetime.now().strftime("%Y%m%d-%H%M%S"), uuid.uuid4().int & 0xFFFF)


def validate_task_name(name: str, exclude_id: str | None = None
                       ) -> tuple[bool, str]:
    """校验本机任务名，返回 (ok, 错误码)。错误码由 UI 本地化。

    规则（保证 name == task_tag(name)，即名字、录制目录、上传项目名一致）：
      * 去首尾空白后非空、≤ 64 字符；
      * 只允许字母/数字/汉字/下划线/连字符（空格、点、斜杠等一律拒绝）；
      * Windows 保留设备名（CON/PRN/AUX/NUL/COM1-9/LPT1-9）拒绝；
      * 与任何现有任务重名（大小写不敏感、含 hidden 与种子任务）拒绝。
    """
    with _lock:
        tasks = _read_raw()
    return _validate_name_locked(tasks, name, exclude_id)


def create_local_task(name: str, total_required: int, description: str = "",
                      created_by: str = "guest") -> dict:
    """新建本机任务并落盘，返回任务 dict。

    失败抛 TaskValidationError（code 见 validate_task_name / total_invalid /
    description_too_long）。任务仅创建者可见，进度只在本机累计。
    """
    total = _validate_total(total_required)
    desc = (description or "").strip()
    if len(desc) > MAX_TASK_DESCRIPTION_LEN:
        raise TaskValidationError("description_too_long")

    with _lock:
        tasks = _read_raw()
        ok, code = _validate_name_locked(tasks, name)
        if not ok:
            raise TaskValidationError(code)
        existing_ids = {_tid(t) for t in tasks}
        task_id = new_local_task_id()
        while task_id in existing_ids:
            task_id = new_local_task_id()
        task = {
            "id": task_id,
            "name": (name or "").strip(),
            "description": desc,
            "source": "local",
            "created_by": created_by or "guest",
            "status": "pending",
            "total_required": total,
            "assigned_at": datetime.now(timezone.utc).isoformat(),
            "assigned_user": None,
            # 进度分账三字段显式写 0：避开旧数据迁移回填（否则会去扫目录）
            "local_count": 0,
            "synced_count": 0,
            "backend_count": 0,
            "completed_count": 0,
        }
        tasks.append(task)
        _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})
    task["status"] = _compute_status(task)
    return dict(task)


def rename_local_task(task_id: str, new_name: str,
                      allow_with_data: bool = False) -> dict:
    """本机任务改名。只允许改本机任务，且默认拒绝「已有录制目录」的任务。

    改名不迁移录制目录：本地已有数据时改名字会让旧目录失去归属、且可能
    与上传队列里的路径脱节，因此默认拒绝（code="has_data"）；新名字的
    目录已存在（孤儿目录）也拒绝（code="name_dir_exists"）。
    """
    with _lock:
        tasks = _read_raw()
        target = _find(tasks, task_id)
        if target is None:
            raise TaskValidationError("task_not_found")
        if not is_local_task(target):
            raise TaskValidationError("not_local")
        old_name = (target.get("name") or "").strip()
        clean = (new_name or "").strip()
        if clean == old_name:
            target["status"] = _compute_status(target)
            return dict(target)
        ok, code = _validate_name_locked(tasks, clean, exclude_id=task_id)
        if not ok:
            raise TaskValidationError(code)
        if not allow_with_data:
            old_dir = _local_task_dir(old_name)
            if old_dir and os.path.isdir(old_dir):
                raise TaskValidationError("has_data")
        new_dir = _local_task_dir(clean)
        if new_dir and os.path.isdir(new_dir):
            raise TaskValidationError("name_dir_exists")
        target["name"] = clean
        target["status"] = _compute_status(target)
        _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})
        return dict(target)


def remove_local_task(task_id: str) -> bool:
    """硬移除本机任务（不写 hidden 墓碑）。

    后端永远不会推送本机任务，不需要防复活；整条移除后同名任务可以重建。
    只对 source=local 的条目生效，服务端任务返回 False（仍走 mark_hidden）。
    """
    with _lock:
        tasks = _read_raw()
        target = _find(tasks, task_id)
        if target is None or not is_local_task(target):
            return False
        kept = [t for t in tasks if _tid(t) != task_id]
        _write({"tasks": kept, "updated_at": datetime.now().isoformat()})
        return True


def _local_visible(task: dict, identity: str | None) -> bool:
    """本机任务可见性：只有创建者身份可见；无 created_by 视为无主（所有人可见）。"""
    creator = task.get("created_by")
    if not creator:
        return True
    return creator == identity


def _local_task_dir(name: str) -> str:
    """本机任务录制目录（懒 import helpers，避免 task_record 顶层拉重依赖）。"""
    if not name:
        return ""
    from core.helpers import task_dir_of
    return task_dir_of(settings.RECORDING_DIR, name)


def _validate_name_locked(tasks: list[dict], name: str,
                          exclude_id: str | None = None) -> tuple[bool, str]:
    """名字校验（调用方持 _lock 并传入当前任务表）。"""
    clean = (name or "").strip()
    if not clean:
        return False, "name_empty"
    if len(clean) > MAX_TASK_NAME_LEN:
        return False, "name_too_long"
    if any(not (c.isalnum() or c in "_-") for c in clean):
        return False, "name_illegal"
    if clean.upper() in _WINDOWS_RESERVED_NAMES:
        return False, "name_reserved"
    key = clean.casefold()
    for t in tasks:
        if exclude_id and _tid(t) == exclude_id:
            continue
        if str(t.get("name") or "").casefold() == key:
            return False, "name_duplicate"
    return True, ""


def _validate_total(total_required) -> int:
    """目标段数：1 ~ MAX_TOTAL_REQUIRED 的整数（bool 不算）。"""
    if isinstance(total_required, bool) or not isinstance(total_required, int):
        raise TaskValidationError("total_invalid")
    if total_required < 1 or total_required > MAX_TOTAL_REQUIRED:
        raise TaskValidationError("total_invalid")
    return total_required


def merge_backend_tasks(backend_tasks: list[dict], view_scope: str | None = None) -> list[dict]:
    """Merge backend tasks into local record. Returns merged list.

    - Backend tasks update existing entries by id, new ones are added.
    - Local-only tasks are preserved.
    - Invalid/incomplete tasks are filtered out.
    - 同名去重: 同一 (name, assigned_user) 组出现多个可见任务时，只保留
      assigned_at 最新的一个（后端推送的优先胜出——"根据后台指令更新任务细节"），
      其余标 hidden。公共任务（均无 assigned_user）分组结果与旧版等价。
    - view_scope（新增）: 当前会话身份。"guest"/None 不触发撤销清除；
      用户名时，本地 assigned_user == 该用户名、且本轮后端未返回的任务
      （已被后端撤销）从本地清除。游客轮询绝不触发撤销清除——防止把
      他人指派任务误删。本机任务（source=local）无论身份一律豁免：它们
      不在后端任务表里，按「后端没返回」清除会直接删掉自建任务。
    """
    with _lock:
        local = _read_raw()

    backend_ids = {_tid(t) for t in backend_tasks}

    # ── 阶段 0: 撤销清除（仅账号轮询；游客/无身份完全跳过）──
    if view_scope not in (None, "guest"):
        local = [
            lt for lt in local
            if is_local_task(lt)
            or not (_assigned_user(lt) == view_scope and _tid(lt) not in backend_ids)
        ]

    candidates: list[dict] = []

    # ── 阶段 1: 后端任务（按 id 合并，逻辑与现状一致）──
    for bt in backend_tasks:
        if not _is_valid(bt):
            continue
        bid = _tid(bt)
        local_match = _find(local, bid)
        backend_global = max(0, int(bt.get("completed_count") or 0))
        if local_match:
            # 保留本地覆盖字段
            for key in ("total_required", "params", "assigned_user"):
                if key in local_match and key not in bt:
                    bt[key] = local_match[key]
            # 进度分账：后端全局数权威；本机完成数/已同步水位沿用本地
            _ensure_sync_fields(local_match)
            bt["local_count"] = local_match["local_count"]
            bt["synced_count"] = local_match["synced_count"]
            bt["backend_count"] = backend_global
            # 保留删除标记：用户删过的任务不能被后端推送复活
            if local_match.get("hidden"):
                bt["hidden"] = True
        else:
            # 新任务：本机 0 段全同步；backend_count 取后端全局数
            # （保留 current_count 作新任务初值的旧语义）
            bt["local_count"] = 0
            bt["synced_count"] = 0
            bt["backend_count"] = backend_global
        candidates.append(bt)

    # ── 阶段 2: 本地独有任务 ──
    for lt in local:
        if not _is_valid(lt):
            continue
        if _tid(lt) not in backend_ids:
            candidates.append(dict(lt))

    # ── 阶段 3: 同名去重（(name, assigned_user) 墓碑）──
    groups: dict[tuple, list[dict]] = {}
    for c in candidates:
        groups.setdefault((c["name"], _assigned_user(c)), []).append(c)
    for group in groups.values():
        visible = [g for g in group if not g.get("hidden")]
        if len(visible) <= 1:
            continue  # 无碰撞（含整组已 hidden）→ 不动
        backend_visible = [v for v in visible if _tid(v) in backend_ids]
        pool = backend_visible if backend_visible else visible
        winner = max(pool, key=_assigned_at_key)
        for g in visible:
            if g is not winner:
                g["hidden"] = True  # 持久化墓碑，防止复活

    for t in candidates:
        _ensure_sync_fields(t)
        _recount(t)
        t["status"] = _compute_status(t)
    save_tasks(candidates)
    return load_tasks()


def mark_hidden(task_id: str):
    """Mark a task as hidden so it no longer appears in the UI."""
    with _lock:
        tasks = _read_raw()
        for t in tasks:
            if _tid(t) == task_id:
                t["hidden"] = True
                break
        _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})


def refresh_progress() -> list[dict]:
    """Reload task list with persisted completion counts."""
    return load_tasks()


def increment_task_completed(task_name: str, task_id: str | None = None,
                             assigned_user: str | None = None) -> dict | None:
    """一次录制完成：把对应任务的录制完成次数持久化 +1，返回更新后的任务。

    匹配优先级：
      1) task_id 提供时按 id 精确匹配（须 name 一致）
      2) (name, assigned_user) 精确匹配（assigned_user 为 None 时匹配公共任务）
      3) 同名条目唯一时回退唯一匹配（兼容旧数据/手动临时任务名）
      4) 多名同名且身份不匹配 → 返回 None 不计数（跨身份串扰防护）
    task_id/assigned_user 由调用方从所选任务（_current_task）提供，
    而非会话身份——避免"登录用户录制公共任务却按用户名匹配"的错配。
    任务不在 tasks.json（手动临时任务名）时不计数，返回 None。
    会话目录随后被删除也不影响该计数——进度以持久化值为准。
    """
    if not task_name:
        return None
    with _lock:
        tasks = _read_raw()
        target = None
        if task_id:
            for t in tasks:
                if _tid(t) == task_id and t.get("name") == task_name:
                    target = t
                    break
        if target is None:
            for t in tasks:
                if (t.get("name") == task_name
                        and _assigned_user(t) == (assigned_user or None)):
                    target = t
                    break
        if target is None:
            matches = [t for t in tasks if t.get("name") == task_name]
            if len(matches) == 1:
                target = matches[0]
        if target is None:
            return None
        _ensure_sync_fields(target)
        target["local_count"] = int(target.get("local_count", 0) or 0) + 1
        _recount(target)
        target["status"] = _compute_status(target)
        _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})
        return dict(target)


def pending_sync_tasks() -> list[dict]:
    """所有 local_count > synced_count 的任务快照（含 hidden，不按身份过滤——
    身份不符由服务端 401/403 处理，跳过本轮即可）。

    纯读路径：直接 _read_raw()，跳过未迁移记录（无 local_count），
    绝不触发迁移写盘（flush 只上报，不改变本地计数语义）。
    本机任务（source=local）显式跳过：后端没有对应任务，上报必被拒 400。
    """
    with _lock:
        tasks = _read_raw()
    out = []
    for t in tasks:
        if is_local_task(t):
            continue
        local = t.get("local_count")
        if not isinstance(local, int):
            continue
        synced = int(t.get("synced_count", 0) or 0)
        tid = _tid(t)
        if local > synced and tid:
            out.append({"id": tid, "name": t.get("name", ""),
                        "local_count": local, "synced_count": synced})
    return out


def mark_synced(task_id: str, synced_count: int, backend_count: int) -> dict | None:
    """flush 成功后回写同步状态。

    synced_count 必须是「快照水位 + 本次增量」（不是当前 local_count——
    快照与响应之间 UI 线程可能又录了一段）；水位只前进不后退；
    backend_count 用响应值无条件替换（服务端权威，迁移基线可能高于真实
    全局数，不能 max-前向）。返回更新后的任务。
    """
    with _lock:
        tasks = _read_raw()
        target = _find(tasks, task_id)
        if target is None or not isinstance(target.get("local_count"), int):
            return None
        if synced_count > (int(target.get("synced_count", 0) or 0)):
            target["synced_count"] = synced_count
        target["backend_count"] = max(0, int(backend_count or 0))
        _recount(target)
        target["status"] = _compute_status(target)
        _write({"tasks": tasks, "updated_at": datetime.now().isoformat()})
        return dict(target)


def task_by_id(task_id: str) -> dict | None:
    tasks = load_tasks()
    return _find(tasks, task_id)


# ═══════════════════════════════════════════════════════════
#  Internal
# ═══════════════════════════════════════════════════════════

def _read_raw() -> list[dict]:
    if not os.path.isfile(_TASKS_FILE):
        _init_seed()
    try:
        with open(_TASKS_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data.get("tasks", [])
    except (json.JSONDecodeError, OSError):
        _init_seed()
        return [dict(t) for t in _SEED_TASKS]


def _write(data: dict):
    os.makedirs(os.path.dirname(_TASKS_FILE), exist_ok=True)
    tmp = _TASKS_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    os.replace(tmp, _TASKS_FILE)


def _init_seed():
    _write({"tasks": [dict(t) for t in _SEED_TASKS], "updated_at": datetime.now().isoformat()})


def _tid(task: dict) -> str:
    return task.get("id", task.get("task_id", ""))


def _assigned_user(task: dict):
    """任务的指派用户；null/空串/缺失统一归一化为 None（= 公共任务）。"""
    return task.get("assigned_user") or None


def _assigned_at_key(task: dict) -> tuple:
    """排序键: 带有效时间的排前，null/空/解析失败排后。
    (0, raw) 与 (1, ts) 首元素不同，跨类不会比较第二元素。"""
    raw = task.get("assigned_at") or ""
    if not raw:
        return (0, "")
    try:
        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
        return (1, dt.timestamp())
    except ValueError:
        return (0, raw)  # 解析失败: 排最后，按原始字符串保序


def _find(tasks: list[dict], task_id: str) -> dict | None:
    for t in tasks:
        if _tid(t) == task_id:
            return t
    return None


def _is_valid(task: dict) -> bool:
    name = task.get("name", task.get("task_name", ""))
    tid = task.get("id", task.get("task_id", ""))
    return bool(name and name.strip()) and bool(tid and tid.strip()) and task.get("total_required", 0) > 0


def _compute_status(task: dict) -> str:
    total = task.get("total_required", 0)
    completed = task.get("completed_count", 0)
    if total <= 0:
        return "pending"
    if completed >= total:
        return "completed"
    if completed > 0:
        return "in_progress"
    return "pending"


def _count_with_fallback(task: dict) -> int:
    """任务的录制完成次数：优先持久化值，缺字段（旧版本数据）回退目录扫描。"""
    cur = task.get("completed_count")
    if isinstance(cur, int) and cur >= 0:
        return cur
    return _count_sessions(task.get("name", ""))


def _count_sessions(task_name: str) -> int:
    """扫描录制目录统计 episode 数 —— 仅用于旧数据迁移回填初值，非进度权威来源。

    v1.1.0 池化布局：list_task_episodes（videos/data 文件组扫描，与
    episodes 每段文件一致——删除时两者同步移除）；旧格式任务目录
    回退按会话子目录计数。
    """
    if not task_name:
        return 0
    task_dir = os.path.join(settings.RECORDING_DIR, task_name)
    if not os.path.isdir(task_dir):
        return 0
    try:
        from core.helpers import detect_session_format, list_task_episodes
        if detect_session_format(task_dir) == "pooled":
            return len(list_task_episodes(task_dir))
    except ImportError:
        pass
    count = 0
    try:
        for entry in os.listdir(task_dir):
            ses_dir = os.path.join(task_dir, entry)
            if not os.path.isdir(ses_dir):
                continue
            if os.path.isfile(os.path.join(ses_dir, "meta", "info.json")):
                count += 1
    except OSError:
        pass
    return count


def _ensure_sync_fields(task: dict) -> bool:
    """补全 local/synced/backend 三字段（迁移）；返回是否发生迁移（需写回）。

    迁移（幂等，字段存在且为合法 int 即跳过）：
      local_count   = 旧 completed_count（目录扫描兜底）
      synced_count  = local_count（历史不补报，避免重复计数）
      backend_count = local_count（历史基线保显示连续；首次 merge/flush
                      用真实全局数无条件替换）
    """
    changed = False
    local = task.get("local_count")
    if not isinstance(local, int) or local < 0:
        task["local_count"] = _count_with_fallback(task)
        changed = True
    synced = task.get("synced_count")
    if not isinstance(synced, int) or synced < 0:
        task["synced_count"] = task["local_count"]
        changed = True
    backend = task.get("backend_count")
    if not isinstance(backend, int) or backend < 0:
        task["backend_count"] = task["local_count"]
        changed = True
    return changed


def _recount(task: dict) -> int:
    """显示值 = backend_count + (local_count - synced_count)，写回 completed_count。

    防御：synced 不超 local（异常水位按已同步处理，宁可少显示不可多显示）。
    """
    local = int(task.get("local_count", 0) or 0)
    synced = min(int(task.get("synced_count", 0) or 0), local)
    task["completed_count"] = int(task.get("backend_count", 0) or 0) + (local - synced)
    return task["completed_count"]