"""线程生命周期补充：worktree 快照、任务外部引用、自动化心跳绑定。

本模块只补控制面已有能力的缺口，不发明执行能力：

- worktree 快照（task_worktree_snapshots）：归档线程时真正要
  保住的是“未提交修改 + 未推送 commit 的位置 + 非忽略文件
  清单”。快照记 branch / head_sha / git status（截 4000 字）/
  git diff（截 20000 字）/ untracked 路径清单。注意诚实边界：
  untracked 文件只记路径不拷内容，diff 只覆盖已跟踪文件的
  修改；快照是事后可查的证据，不是可一键还原的备份。恢复
  （restore）只在 worktree 目录还在时解归档；目录已删时
  回 409 并附快照（重建需要仓库路径，服务端没有，不编造）。
- 任务外部引用（task_refs）：任务附件表只记文件包路径，
  交付物里的“生成的文档 / 构建结果 / 外部服务结果引用”
  没有地方放。refs 只记 kind + ref + note 的引用行，不存
  文件内容、不校验外部链接可达——引用就是引用。
- 自动化心跳绑定（automation_bindings）：自动化定义本身
  只有 interval + prompt，没有“绑到哪个线程”。绑定记
  automation -> task + mode（heartbeat | cron）。heartbeat
  是“到期唤醒已有线程”（走 thread/message 的投递路）；
  cron 是“到期独立跑新任务”（走自动化自身的产出路）。
  本模块只记绑定与取绑定；到期触发仍由外部触发器
  （host run-due / 未来的 run_due_automation）调用
  /heartbeat 端点，服务端没有也不会偷偷起定时器。

三张表都走 UNMIGRATED_TABLES（见 cloud/migrations.py）：
运行时 ensure 建表，与 task_handoffs 同款，不进
POSTGRES_SCHEMA 快照。"""

import json
import time
import uuid

# 行取值统一走 peer_messages.row_get（A7）：本模块旧私有版在
# 真实行类型（None / dict / sqlite3.Row）下与之结果完全一致
# （None 特判、dict .get、缺列回 None），调用处仍用 _row_get
# 别名，行为不变；旧版宽 Exception 兜底的病态行不在契约内，
# 见 peer_messages.row_get docstring。
from cloud.peer_messages import row_get as _row_get

SNAPSHOT_STATUS_LIMIT = 4000
SNAPSHOT_DIFF_LIMIT = 20000
REF_LIMIT = 2000
NOTE_LIMIT = 1000
REF_KINDS = ("doc", "build", "external", "pr", "file")
BINDING_MODES = ("heartbeat", "cron")


# -- worktree 快照 ------------------------------------------------------

def ensure_snapshots_table(conn):
    conn.execute(
        "CREATE TABLE IF NOT EXISTS task_worktree_snapshots("
        "id TEXT PRIMARY KEY, "
        "task_id TEXT, "
        "project_id TEXT, "
        "worktree_path TEXT, "
        "branch TEXT, "
        "head_sha TEXT, "
        "status_text TEXT, "
        "diff_text TEXT, "
        "untracked_json TEXT, "
        "created_at DOUBLE PRECISION, "
        "restored_at DOUBLE PRECISION, "
        "restore_note TEXT)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_snapshots_task"
        " ON task_worktree_snapshots(task_id, created_at)")


def snapshot_to_dict(row):
    try:
        untracked = json.loads(_row_get(row, "untracked_json") or "[]")
    except ValueError:
        untracked = []
    if not isinstance(untracked, list):
        untracked = []
    return {
        "id": _row_get(row, "id"),
        "task_id": _row_get(row, "task_id"),
        "project_id": _row_get(row, "project_id"),
        "worktree_path": _row_get(row, "worktree_path"),
        "branch": _row_get(row, "branch"),
        "head_sha": _row_get(row, "head_sha"),
        "status_text": _row_get(row, "status_text"),
        "diff_text": _row_get(row, "diff_text"),
        "untracked": untracked,
        "created_at": _row_get(row, "created_at"),
        "restored_at": _row_get(row, "restored_at"),
        "restore_note": _row_get(row, "restore_note"),
    }


def create_snapshot(conn, task_id, project_id, worktree_path, branch,
                    head_sha, status_text, diff_text, untracked, now=None):
    """插一条快照并提交。status/diff 在调用方截断，这里再截
    一次防直接调用方越界；untracked 只接受 list[str]。"""
    ensure_snapshots_table(conn)
    snap_id = uuid.uuid4().hex
    now = time.time() if now is None else now
    untracked = [str(p) for p in (untracked or []) if isinstance(p, str)]
    conn.execute(
        "INSERT INTO task_worktree_snapshots(id, task_id, project_id,"
        " worktree_path, branch, head_sha, status_text, diff_text,"
        " untracked_json, created_at, restored_at, restore_note)"
        " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL)",
        (snap_id, task_id, project_id, worktree_path, branch,
         head_sha, (status_text or "")[:SNAPSHOT_STATUS_LIMIT],
         (diff_text or "")[:SNAPSHOT_DIFF_LIMIT],
         json.dumps(untracked), now))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM task_worktree_snapshots WHERE id = ?",
        (snap_id,)).fetchone()
    return snapshot_to_dict(row)


def list_snapshots(conn, task_id):
    ensure_snapshots_table(conn)
    rows = conn.execute(
        "SELECT * FROM task_worktree_snapshots WHERE task_id = ?"
        " ORDER BY created_at DESC, id",
        (task_id,)).fetchall()
    return [snapshot_to_dict(r) for r in rows]


def latest_snapshot(conn, task_id):
    rows = list_snapshots(conn, task_id)
    return rows[0] if rows else None


def mark_restored(conn, snapshot_id, note, now=None):
    ensure_snapshots_table(conn)
    now = time.time() if now is None else now
    cur = conn.execute(
        "UPDATE task_worktree_snapshots SET restored_at = ?,"
        " restore_note = ? WHERE id = ?",
        (now, note, snapshot_id))
    conn.commit()
    if cur.rowcount != 1:
        return None
    row = conn.execute(
        "SELECT * FROM task_worktree_snapshots WHERE id = ?",
        (snapshot_id,)).fetchone()
    return snapshot_to_dict(row)


# -- 任务外部引用 ---------------------------------------------------------

def ensure_refs_table(conn):
    conn.execute(
        "CREATE TABLE IF NOT EXISTS task_refs("
        "id TEXT PRIMARY KEY, "
        "task_id TEXT, "
        "project_id TEXT, "
        "kind TEXT, "
        "ref TEXT, "
        "note TEXT, "
        "created_by TEXT, "
        "created_at DOUBLE PRECISION)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_refs_task"
        " ON task_refs(task_id, created_at)")


def ref_to_dict(row):
    return {
        "id": _row_get(row, "id"),
        "task_id": _row_get(row, "task_id"),
        "project_id": _row_get(row, "project_id"),
        "kind": _row_get(row, "kind"),
        "ref": _row_get(row, "ref"),
        "note": _row_get(row, "note"),
        "created_by": _row_get(row, "created_by"),
        "created_at": _row_get(row, "created_at"),
    }


def add_ref(conn, task_id, project_id, kind, ref, note, created_by,
            now=None):
    """校验后插引用行。kind 不在白名单、ref 为空或超长、
    note 超长时抛 ValueError（调用方转 400）。"""
    if kind not in REF_KINDS:
        raise ValueError("kind must be one of %s" % ", ".join(REF_KINDS))
    if not ref or not isinstance(ref, str) or not ref.strip():
        raise ValueError("ref required")
    ref = ref.strip()
    if len(ref) > REF_LIMIT:
        raise ValueError("ref too long")
    if note is not None and len(str(note)) > NOTE_LIMIT:
        raise ValueError("note too long")
    ensure_refs_table(conn)
    ref_id = uuid.uuid4().hex
    now = time.time() if now is None else now
    conn.execute(
        "INSERT INTO task_refs(id, task_id, project_id, kind, ref,"
        " note, created_by, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?)",
        (ref_id, task_id, project_id, kind, ref, note, created_by,
         now))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM task_refs WHERE id = ?", (ref_id,)).fetchone()
    return ref_to_dict(row)


def list_refs(conn, task_id):
    ensure_refs_table(conn)
    rows = conn.execute(
        "SELECT * FROM task_refs WHERE task_id = ?"
        " ORDER BY created_at, id",
        (task_id,)).fetchall()
    return [ref_to_dict(r) for r in rows]


def delete_ref(conn, task_id, ref_id):
    """只删属于该任务的引用行；行不存在或不属于该任务时
    返回 False（调用方一律 404，不泄露跨任务存在性）。"""
    ensure_refs_table(conn)
    cur = conn.execute(
        "DELETE FROM task_refs WHERE id = ? AND task_id = ?",
        (ref_id, task_id))
    conn.commit()
    return cur.rowcount == 1


# -- 自动化心跳绑定 --------------------------------------------------------

def ensure_bindings_table(conn):
    conn.execute(
        "CREATE TABLE IF NOT EXISTS automation_bindings("
        "automation_id TEXT PRIMARY KEY, "
        "project_id TEXT, "
        "task_id TEXT, "
        "mode TEXT, "
        "created_at DOUBLE PRECISION)")


def binding_to_dict(row):
    if row is None:
        return None
    return {
        "automation_id": _row_get(row, "automation_id"),
        "project_id": _row_get(row, "project_id"),
        "task_id": _row_get(row, "task_id"),
        "mode": _row_get(row, "mode"),
        "created_at": _row_get(row, "created_at"),
    }


def bind_automation(conn, automation_id, project_id, task_id, mode,
                    now=None):
    if mode not in BINDING_MODES:
        raise ValueError("mode must be heartbeat or cron")
    ensure_bindings_table(conn)
    now = time.time() if now is None else now
    conn.execute(
        "INSERT INTO automation_bindings(automation_id, project_id,"
        " task_id, mode, created_at) VALUES(?, ?, ?, ?, ?)"
        " ON CONFLICT(automation_id) DO UPDATE SET"
        " task_id = excluded.task_id, mode = excluded.mode,"
        " created_at = excluded.created_at",
        (automation_id, project_id, task_id, mode, now))
    conn.commit()
    row = conn.execute(
        "SELECT * FROM automation_bindings WHERE automation_id = ?",
        (automation_id,)).fetchone()
    return binding_to_dict(row)


def get_binding(conn, automation_id):
    ensure_bindings_table(conn)
    row = conn.execute(
        "SELECT * FROM automation_bindings WHERE automation_id = ?",
        (automation_id,)).fetchone()
    return binding_to_dict(row)


def unbind_automation(conn, automation_id):
    ensure_bindings_table(conn)
    cur = conn.execute(
        "DELETE FROM automation_bindings WHERE automation_id = ?",
        (automation_id,))
    conn.commit()
    return cur.rowcount == 1