"""数据层：PostgreSQL 17（docker-compose 提供），通过 psycopg 3 访问。

多校隔离：本实例只看 settings.school_id 这所学校的数据（knowledge / tasks / system_accounts / learned_entries）。
settings 表是实例级配置，不分学校。对外函数签名与旧 MySQL 版本一致。
"""

import json
import re
import time
import uuid
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from typing import Any

import psycopg
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb

from backend.config import KNOWLEDGE_DIR, PROFILES_DIR, settings
from shared import knowledge_match

SCHEMA_PATH = Path(__file__).parent / "schema.sql"

_TASK_COLUMNS = {
    "student_id",
    "knowledge_id",
    "status",
    "current_step",
    "inputs",
    "artifacts",
    "skill_version",
    "browser_profile_path",
    "pause_reason",
    "next_check_at",
}
_JSON_FIELDS = {"inputs", "artifacts"}
LEARNED_STATUSES = {"draft", "verified", "rejected"}
LEARNED_CATEGORIES = knowledge_match.LEARNED_CATEGORIES
AUTO_VERIFY_CATEGORIES = knowledge_match.AUTO_VERIFY_CATEGORIES


def _sid() -> str:
    return settings.school_id or "default"


# ---------- 连接 ----------


def _connect() -> psycopg.Connection:
    return psycopg.connect(
        host=settings.pg_host,
        port=settings.pg_port,
        user=settings.pg_user,
        password=settings.pg_password,
        dbname=settings.pg_database,
        connect_timeout=5,
        row_factory=dict_row,
        options="-c timezone=Asia/Shanghai",
    )


@contextmanager
def get_conn():
    """每次调用一个短连接：成功提交，异常回滚。MVP 阶段够用，后续可换连接池。"""
    conn = _connect()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def _fetchone(conn, sql: str, params: tuple | list = ()) -> dict | None:
    with conn.cursor() as cur:
        cur.execute(sql, params)
        return cur.fetchone()


def _fetchall(conn, sql: str, params: tuple | list = ()) -> list[dict]:
    with conn.cursor() as cur:
        cur.execute(sql, params)
        return list(cur.fetchall())


def _execute(conn, sql: str, params: tuple | list = ()) -> int:
    with conn.cursor() as cur:
        cur.execute(sql, params)
        return cur.rowcount


def _insert_id(conn, sql: str, params: tuple | list = ()) -> int:
    """INSERT ... RETURNING id"""
    with conn.cursor() as cur:
        cur.execute(sql, params)
        row = cur.fetchone()
        return int(row["id"]) if row else 0


# ---------- 初始化 ----------


def init_db(retries: int = 40, delay: float = 1.5) -> None:
    """建表并同步 knowledge/*.yaml。容器首次启动要几秒才能接受连接，所以带重试。"""
    last_err: Exception | None = None
    for attempt in range(1, retries + 1):
        try:
            with get_conn() as conn:
                _apply_schema(conn)
                _apply_prospecting_knowledge_schema(conn)
                _seed_knowledge(conn)
            return
        except psycopg.Error as e:
            last_err = e
            if attempt == 1:
                print(f"[db] 等待 PostgreSQL {settings.pg_host}:{settings.pg_port} 就绪 ... ({e})", flush=True)
            time.sleep(delay)
    raise RuntimeError(
        f"无法连接 PostgreSQL {settings.pg_host}:{settings.pg_port}/{settings.pg_database}。"
        f"请先在项目根目录执行 docker compose up -d。原始错误：{last_err}"
    )


def _strip_comment(line: str) -> str:
    """去掉行尾 -- 注释。语句要拼成一行执行，留着注释会把后面的列定义一起注释掉。"""
    in_quote = False
    for i, ch in enumerate(line):
        if ch == "'":
            in_quote = not in_quote
        elif ch == "-" and not in_quote and line[i : i + 2] == "--":
            return line[:i]
    return line


def _apply_schema(conn) -> None:
    sql = SCHEMA_PATH.read_text(encoding="utf-8")
    for stmt in sql.split(";"):
        body = " ".join(_strip_comment(line) for line in stmt.splitlines()).strip()
        if body:
            _execute(conn, body)


def _apply_prospecting_knowledge_schema(conn) -> None:
    """Additive v0.3 ledger. Historical learned rows and counters stay untouched."""
    statements = [
        """CREATE TABLE IF NOT EXISTS prospecting_methods (
            workspace_id VARCHAR(64) NOT NULL, id VARCHAR(80) NOT NULL,
            method_key VARCHAR(200) NOT NULL, version VARCHAR(40) NOT NULL,
            spec JSONB NOT NULL, created_from_task_id VARCHAR(128) NOT NULL DEFAULT '',
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            PRIMARY KEY (workspace_id, id), UNIQUE (workspace_id, method_key, version))""",
        """CREATE TABLE IF NOT EXISTS prospecting_validation_events (
            workspace_id VARCHAR(64) NOT NULL, id VARCHAR(80) NOT NULL,
            method_id VARCHAR(80) NOT NULL, task_id VARCHAR(128) NOT NULL,
            object_digest VARCHAR(80) NOT NULL, event_key VARCHAR(200) NOT NULL,
            payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            PRIMARY KEY (workspace_id, id), UNIQUE (workspace_id, method_id, event_key),
            FOREIGN KEY (workspace_id, method_id) REFERENCES prospecting_methods(workspace_id, id))""",
        """CREATE INDEX IF NOT EXISTS idx_prospecting_validation_method
            ON prospecting_validation_events(workspace_id, method_id, created_at)""",
        """CREATE INDEX IF NOT EXISTS idx_prospecting_validation_task
            ON prospecting_validation_events(workspace_id, task_id)""",
        """CREATE TABLE IF NOT EXISTS prospecting_business_feedback (
            workspace_id VARCHAR(64) NOT NULL, id VARCHAR(80) NOT NULL,
            task_id VARCHAR(128) NOT NULL, result_version VARCHAR(128) NOT NULL,
            payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            PRIMARY KEY (workspace_id, id))""",
        """CREATE INDEX IF NOT EXISTS idx_prospecting_feedback_task
            ON prospecting_business_feedback(workspace_id, task_id, created_at)""",
    ]
    for statement in statements:
        _execute(conn, statement)


def _seed_knowledge(conn) -> None:
    """knowledge/*.yaml 是出厂配置：写入本实例所属学校（YAML 里可用 school_id 指定别的学校）。"""
    import yaml

    for yaml_file in sorted(KNOWLEDGE_DIR.glob("*.yaml")):
        data = yaml.safe_load(yaml_file.read_text(encoding="utf-8"))
        # 当前产品只发布拓客知识；学校包作为历史资料保留，不进入新工作区。
        if data.get("id") != "xiaohongshu_prospecting":
            continue
        school = str(data.get("school_id") or _sid())
        _execute(
            conn,
            """INSERT INTO knowledge (school_id, id, name, content) VALUES (%s, %s, %s, %s)
               ON CONFLICT (school_id, id) DO UPDATE
               SET name = EXCLUDED.name, content = EXCLUDED.content, updated_at = NOW()""",
            (school, data["id"], data["name"], Jsonb(data)),
        )


# ---------- settings（实例级） ----------


def get_setting(key: str, default: str = "") -> str:
    with get_conn() as conn:
        row = _fetchone(conn, 'SELECT "value" FROM settings WHERE "key" = %s', (key,))
    return row["value"] if row else default


def set_setting(key: str, value: str) -> None:
    with get_conn() as conn:
        _execute(
            conn,
            """INSERT INTO settings ("key", "value") VALUES (%s, %s)
               ON CONFLICT ("key") DO UPDATE SET "value" = EXCLUDED."value" """,
            (key, value),
        )


# ---------- tasks ----------


def create_task(knowledge_id: str, student_id: str = "default", inputs: dict | None = None) -> dict:
    task_id = f"task_{uuid.uuid4().hex[:12]}"
    profile_path = str(PROFILES_DIR / student_id)
    Path(profile_path).mkdir(parents=True, exist_ok=True)
    with get_conn() as conn:
        _execute(
            conn,
            """INSERT INTO tasks (id, school_id, student_id, knowledge_id, status, inputs, artifacts, browser_profile_path)
               VALUES (%s, %s, %s, %s, 'collecting', %s, %s, %s)""",
            (task_id, _sid(), student_id, knowledge_id, Jsonb(inputs or {}), Jsonb({}), profile_path),
        )
    return get_task(task_id)


def get_task(task_id: str) -> dict | None:
    with get_conn() as conn:
        row = _fetchone(conn, "SELECT * FROM tasks WHERE id = %s AND school_id = %s", (task_id, _sid()))
    return _row_to_task(row) if row else None


def update_task(task_id: str, **fields: Any) -> dict | None:
    if not fields:
        return get_task(task_id)
    unknown = set(fields) - _TASK_COLUMNS
    if unknown:
        raise ValueError(f"tasks 不存在字段: {sorted(unknown)}")
    sets: list[str] = []
    values: list[Any] = []
    for k, v in fields.items():
        sets.append(f"{k} = %s")
        values.append(Jsonb(v) if k in _JSON_FIELDS and v is not None else v)
    sets.append("updated_at = NOW()")
    values += [task_id, _sid()]
    with get_conn() as conn:
        _execute(conn, f"UPDATE tasks SET {', '.join(sets)} WHERE id = %s AND school_id = %s", values)
    return get_task(task_id)


def list_tasks(limit: int = 20) -> list[dict]:
    with get_conn() as conn:
        rows = _fetchall(
            conn,
            "SELECT * FROM tasks WHERE school_id = %s ORDER BY updated_at DESC LIMIT %s",
            (_sid(), int(limit)),
        )
    return [_row_to_task(r) for r in rows]


def due_tracking_tasks(limit: int = 5) -> list[dict]:
    """到了回查时间的已提交任务（本校）。"""
    with get_conn() as conn:
        rows = _fetchall(
            conn,
            """SELECT * FROM tasks
               WHERE school_id = %s AND status = 'submitted' AND next_check_at IS NOT NULL AND next_check_at <= NOW()
               ORDER BY next_check_at LIMIT %s""",
            (_sid(), int(limit)),
        )
    return [_row_to_task(r) for r in rows]


# ---------- knowledge ----------


def get_knowledge(knowledge_id: str) -> dict | None:
    with get_conn() as conn:
        row = _fetchone(conn, "SELECT * FROM knowledge WHERE school_id = %s AND id = %s", (_sid(), knowledge_id))
    if not row:
        return None
    content = _load_json(row["content"], {})
    return {"id": row["id"], "name": row["name"], "school_id": row["school_id"], **content}


def list_knowledge() -> list[dict]:
    with get_conn() as conn:
        rows = _fetchall(conn, "SELECT id, name FROM knowledge WHERE school_id = %s ORDER BY name", (_sid(),))
    return [{"id": r["id"], "name": r["name"]} for r in rows]


# ---------- run_logs ----------


def add_run_log(task_id: str, step_id: str, verdict: str, agent_history: Any = None) -> int:
    history = Jsonb(json.loads(json.dumps(agent_history, ensure_ascii=False, default=str))) if agent_history else None
    with get_conn() as conn:
        return _insert_id(
            conn,
            """INSERT INTO run_logs (task_id, step_id, verdict, agent_history, screenshots, finished_at)
               VALUES (%s, %s, %s, %s, %s, NOW()) RETURNING id""",
            (task_id, step_id, verdict, history, Jsonb([])),
        )


# ---------- system_accounts（密码本） ----------


def upsert_account(student_id: str, system_id: str, username: str, password_enc: str, role: str = "") -> dict:
    """密码本：一个人在一个系统只有一条账号，role 是该账号在该系统里的身份（学生 / 辅导员 …）。"""
    with get_conn() as conn:
        _execute(
            conn,
            """INSERT INTO system_accounts (school_id, student_id, system_id, role, username, password_enc)
               VALUES (%s, %s, %s, %s, %s, %s)
               ON CONFLICT (school_id, student_id, system_id) DO UPDATE
               SET role = EXCLUDED.role, username = EXCLUDED.username, password_enc = EXCLUDED.password_enc,
                   updated_at = NOW()""",
            (_sid(), student_id, system_id, (role or "").strip(), username, password_enc),
        )
    return get_account(student_id, system_id)


def get_account(student_id: str, system_id: str) -> dict | None:
    """含 password_enc（密文），仅供编排层解密使用，不要直接返回给前端。"""
    with get_conn() as conn:
        row = _fetchone(
            conn,
            "SELECT * FROM system_accounts WHERE school_id = %s AND student_id = %s AND system_id = %s",
            (_sid(), student_id, system_id),
        )
    return dict(row) if row else None


def list_accounts(student_id: str | None = None) -> list[dict]:
    sql = "SELECT id, student_id, system_id, role, username, created_at, updated_at FROM system_accounts WHERE school_id = %s"
    params: list[Any] = [_sid()]
    if student_id:
        sql += " AND student_id = %s"
        params.append(student_id)
    with get_conn() as conn:
        rows = _fetchall(conn, sql + " ORDER BY student_id, system_id", params)
    return [{**r, "created_at": _fmt_dt(r["created_at"]), "updated_at": _fmt_dt(r["updated_at"])} for r in rows]


def delete_account(account_id: int) -> None:
    with get_conn() as conn:
        _execute(conn, "DELETE FROM system_accounts WHERE id = %s AND school_id = %s", (int(account_id), _sid()))


# ---------- learned_entries（自动沉淀） ----------


def _row_to_learned(row: dict) -> dict:
    return {
        **row,
        "created_at": _fmt_dt(row.get("created_at")),
        "updated_at": _fmt_dt(row.get("updated_at")),
        "verified_at": _fmt_dt(row.get("verified_at")),
    }


# 判重规则与中心服务端共用一份（shared/knowledge_match.py），保证两边合并结果一致
_similarity = knowledge_match.similarity
_same_knowledge = knowledge_match.same_knowledge
_pick_category = knowledge_match.pick_category


def _operation_contract_identity(key: str) -> tuple[str, str, int] | None:
    """Versioned source capabilities have a stronger identity than their prose."""
    match = re.fullmatch(r"xhs_operation_([a-z][a-z0-9_]*)_(note|comment|reply)_v([1-9][0-9]*)", key)
    if match:
        return match.group(1), match.group(2), int(match.group(3))
    # Keep the old namespace isolated too, without repairing any historical
    # note-key/comment-value row or merging it into a new operation contract.
    match = re.fullmatch(r"xhs_visible_(note|comment|reply)_v([1-9][0-9]*)", key)
    return ("legacy_visible", match.group(1), int(match.group(2))) if match else None


def _same_scoped_knowledge(knowledge_id: str, business_id: str, a_key: str, a_value: str, b_key: str, b_value: str) -> bool:
    if knowledge_id == "xiaohongshu_prospecting" and business_id == "prospecting":
        a_identity, b_identity = _operation_contract_identity(a_key), _operation_contract_identity(b_key)
        if a_identity is not None or b_identity is not None:
            return a_identity is not None and a_identity == b_identity
    return _same_knowledge(a_key, a_value, b_key, b_value)


def _load_scope_rows(conn, knowledge_id: str, business_id: str) -> list[dict]:
    return _fetchall(
        conn,
        """SELECT id, "key", "value", status, category, observations, source_task_id FROM learned_entries
           WHERE school_id = %s AND knowledge_id = %s AND business_id = %s AND status IN ('draft', 'verified')
           ORDER BY id""",
        (_sid(), knowledge_id, business_id),
    )


def _promote(conn, entry_id: int, by: str) -> None:
    """把一条记录置为 verified：同键只保留这一条 verified。"""
    row = _fetchone(conn, "SELECT * FROM learned_entries WHERE id = %s", (int(entry_id),))
    if not row:
        return
    _execute(
        conn,
        """DELETE FROM learned_entries
           WHERE school_id = %s AND knowledge_id = %s AND business_id = %s AND "key" = %s
             AND status = 'verified' AND id <> %s""",
        (row["school_id"], row["knowledge_id"], row["business_id"], row["key"], int(entry_id)),
    )
    _execute(
        conn,
        """UPDATE learned_entries SET status = 'verified', verified_at = NOW(), verified_by = %s, updated_at = NOW()
           WHERE id = %s""",
        (by, int(entry_id)),
    )


def add_learned_drafts(knowledge_id: str, business_id: str, entries: list[dict], source_task_id: str | None, *, auto_promote: bool = True) -> dict:
    """写入 / 合并草稿，并按规则自动晋升。普通知识按内容判重。

    小红书显式操作合同按能力、来源类型、版本隔离；新命名空间不继承旧错配行。

    - 与某条已 verified 的知识相同：跳过
    - 与某条 draft 相同：合并（保留原 key、取更长的 value、类别取非 note 的），来自不同任务则观察次数 +1；
      达到门槛且类别属于控件技巧 / 入口路径：自动 verified（verified_by=auto）
    - 否则新建 draft
    返回 {"written", "merged", "promoted", "skipped"}。
    """
    stats = {"written": 0, "merged": 0, "promoted": 0, "skipped": 0}
    if not entries:
        return stats
    min_obs = max(1, int(settings.learned_auto_verify_min_obs))
    with get_conn() as conn:
        rows = _load_scope_rows(conn, knowledge_id, business_id)
        for e in entries:
            key = str(e.get("key") or "").strip()[:128]
            value = str(e.get("value") or "").strip()
            category = str(e.get("category") or "note").strip()
            if category not in LEARNED_CATEGORIES:
                category = "note"
            if not key or not value:
                continue

            match = next((r for r in rows if _same_scoped_knowledge(knowledge_id, business_id, key, value, r["key"], r["value"])), None)
            if match and match["status"] == "verified":
                stats["skipped"] += 1
                continue
            if match is None:
                clash = next((r for r in rows if r["key"] == key and r["status"] == "draft"), None)
                if clash:
                    _execute(conn, "DELETE FROM learned_entries WHERE id = %s", (clash["id"],))
                    rows = [r for r in rows if r["id"] != clash["id"]]
                new_id = _insert_id(
                    conn,
                    """INSERT INTO learned_entries
                       (school_id, knowledge_id, business_id, "key", "value", status, category, observations, source_task_id)
                       VALUES (%s, %s, %s, %s, %s, 'draft', %s, 1, %s) RETURNING id""",
                    (_sid(), knowledge_id, business_id, key, value, category, source_task_id),
                )
                rows.append({"id": new_id, "key": key, "value": value, "status": "draft", "category": category,
                             "observations": 1, "source_task_id": source_task_id})
                stats["written"] += 1
                continue

            same_task = match["source_task_id"] == source_task_id
            observations = int(match["observations"] or 1) + (0 if same_task else 1)
            richer = value if len(value) > len(match["value"] or "") else match["value"]
            cat = _pick_category(category, match["category"] or "note")
            _execute(
                conn,
                """UPDATE learned_entries SET "value" = %s, observations = %s, category = %s, source_task_id = %s,
                       updated_at = NOW() WHERE id = %s""",
                (richer, observations, cat, source_task_id, match["id"]),
            )
            match.update({"value": richer, "observations": observations, "category": cat, "source_task_id": source_task_id})
            stats["merged"] += 1
            if auto_promote and observations >= min_obs and cat in AUTO_VERIFY_CATEGORIES:
                _promote(conn, match["id"], "auto")
                match["status"] = "verified"
                stats["promoted"] += 1
    return stats


def consolidate_learned(knowledge_id: str, business_id: str) -> dict:
    """把已有草稿按内容归并：同一条知识只留一行，观察次数累加，类别取非 note，达到门槛的自动晋升。
    与已 verified 相同的草稿直接删除。返回 {"clusters", "removed", "promoted"}。"""
    stats = {"clusters": 0, "removed": 0, "promoted": 0}
    min_obs = max(1, int(settings.learned_auto_verify_min_obs))
    with get_conn() as conn:
        rows = _load_scope_rows(conn, knowledge_id, business_id)
        verified = [r for r in rows if r["status"] == "verified"]
        drafts = [r for r in rows if r["status"] == "draft"]
        keep: list[dict] = []
        for d in drafts:
            if any(_same_scoped_knowledge(knowledge_id, business_id, d["key"], d["value"], v["key"], v["value"]) for v in verified):
                _execute(conn, "DELETE FROM learned_entries WHERE id = %s", (d["id"],))
                stats["removed"] += 1
            else:
                keep.append(d)
        used: set[int] = set()
        for i, a in enumerate(keep):
            if a["id"] in used:
                continue
            cluster = [a]
            for b in keep[i + 1:]:
                if b["id"] in used:
                    continue
                if any(_same_scoped_knowledge(knowledge_id, business_id, b["key"], b["value"], c["key"], c["value"]) for c in cluster):
                    cluster.append(b)
                    used.add(b["id"])
            used.add(a["id"])
            if len(cluster) == 1:
                continue
            stats["clusters"] += 1
            head = max(cluster, key=lambda r: len(r["value"] or ""))
            observations = sum(int(r["observations"] or 1) for r in cluster)
            tasks = {r["source_task_id"] for r in cluster if r["source_task_id"]}
            observations = max(observations, len(tasks))
            cat = _pick_category(*[r["category"] or "note" for r in cluster])
            for r in cluster:
                if r["id"] != head["id"]:
                    _execute(conn, "DELETE FROM learned_entries WHERE id = %s", (r["id"],))
                    stats["removed"] += 1
            _execute(
                conn,
                "UPDATE learned_entries SET observations = %s, category = %s, updated_at = NOW() WHERE id = %s",
                (observations, cat, head["id"]),
            )
            if observations >= min_obs and cat in AUTO_VERIFY_CATEGORIES:
                _promote(conn, head["id"], "auto")
                stats["promoted"] += 1
    return stats


def list_learned(
    knowledge_id: str | None = None,
    business_id: str | None = None,
    status: str | None = None,
    source_task_id: str | None = None,
) -> list[dict]:
    sql = "SELECT * FROM learned_entries WHERE school_id = %s"
    params: list[Any] = [_sid()]
    if knowledge_id:
        sql += " AND knowledge_id = %s"
        params.append(knowledge_id)
    if business_id is not None:
        sql += " AND business_id = %s"
        params.append(business_id)
    if status:
        sql += " AND status = %s"
        params.append(status)
    if source_task_id:
        sql += " AND source_task_id = %s"
        params.append(source_task_id)
    with get_conn() as conn:
        rows = _fetchall(conn, sql + " ORDER BY business_id, status, id", params)
    return [_row_to_learned(r) for r in rows]


def get_learned(entry_id: int) -> dict | None:
    with get_conn() as conn:
        row = _fetchone(conn, "SELECT * FROM learned_entries WHERE id = %s AND school_id = %s", (int(entry_id), _sid()))
    return _row_to_learned(row) if row else None


def set_learned_status(entry_id: int, status: str, by: str = "human") -> dict | None:
    if status not in LEARNED_STATUSES:
        raise ValueError(f"非法状态: {status}")
    with get_conn() as conn:
        row = _fetchone(conn, "SELECT * FROM learned_entries WHERE id = %s AND school_id = %s", (int(entry_id), _sid()))
        if not row:
            return None
        if status == "verified":
            _promote(conn, int(entry_id), by)
        else:
            _execute(
                conn,
                """DELETE FROM learned_entries
                   WHERE school_id = %s AND knowledge_id = %s AND business_id = %s AND "key" = %s
                     AND status = %s AND id <> %s""",
                (_sid(), row["knowledge_id"], row["business_id"], row["key"], status, int(entry_id)),
            )
            _execute(conn, "UPDATE learned_entries SET status = %s, updated_at = NOW() WHERE id = %s", (status, int(entry_id)))
    return get_learned(entry_id)


def apply_hub_entries(entries: list[dict]) -> dict:
    """把中心服务端审核后的条目落到本地。

    按内容匹配本地已有条目：匹配上就改状态和内容（中心可能改写过），匹配不上且是采纳的就新建。
    中心驳回的条目在本地也置为 rejected，防止本地自动晋升过的错知识继续注入提示词。
    """
    stats = {"verified": 0, "rejected": 0, "created": 0, "skipped": 0}
    if not entries:
        return stats
    with get_conn() as conn:
        for e in entries:
            knowledge_id = str(e.get("knowledge_id") or "")
            business_id = str(e.get("business_id") or "")
            key = str(e.get("key") or "").strip()[:128]
            value = str(e.get("value") or "").strip()
            status = str(e.get("status") or "")
            category = knowledge_match.normalise_category(e.get("category"))
            if not key or not value or status not in ("verified", "rejected"):
                stats["skipped"] += 1
                continue

            rows = _load_scope_rows(conn, knowledge_id, business_id)
            match = next((r for r in rows if knowledge_match.same_knowledge(key, value, r["key"], r["value"])), None)
            if match is None:
                if status != "verified":
                    stats["skipped"] += 1
                    continue
                _execute(
                    conn,
                    """INSERT INTO learned_entries
                       (school_id, knowledge_id, business_id, "key", "value", status, category, observations,
                        verified_at, verified_by)
                       VALUES (%s, %s, %s, %s, %s, 'verified', %s, 1, NOW(), 'hub')
                       ON CONFLICT (school_id, knowledge_id, business_id, "key", status) DO UPDATE
                       SET "value" = EXCLUDED."value", category = EXCLUDED.category, updated_at = NOW()""",
                    (_sid(), knowledge_id, business_id, key, value, category),
                )
                stats["created"] += 1
                continue

            # 同 key 同状态有唯一约束，先清掉会冲突的另一条
            _execute(
                conn,
                """DELETE FROM learned_entries
                   WHERE school_id = %s AND knowledge_id = %s AND business_id = %s AND "key" = %s
                     AND status = %s AND id <> %s""",
                (_sid(), knowledge_id, business_id, match["key"], status, match["id"]),
            )
            _execute(
                conn,
                """UPDATE learned_entries
                   SET "value" = %s, status = %s, category = %s, verified_at = NOW(), verified_by = 'hub',
                       updated_at = NOW()
                   WHERE id = %s""",
                (value, status, category, match["id"]),
            )
            stats[status] += 1
    return stats


def update_learned_value(entry_id: int, value: str) -> dict | None:
    with get_conn() as conn:
        _execute(
            conn,
            'UPDATE learned_entries SET "value" = %s, updated_at = NOW() WHERE id = %s AND school_id = %s',
            (value.strip(), int(entry_id), _sid()),
        )
    return get_learned(entry_id)


def trial_learned(knowledge_id: str, business_id: str | None, limit: int) -> list[dict]:
    """挑一批「待验证线索」注入提示词。

    草稿不再压着等人审——真假只有跑一遍才知道。按可信度排序取前 limit 条：
    已被证实过的优先，然后是观察次数多的，最后是新的。被证伪过的排最后（还没到降级门槛但先别太信）。
    """
    if limit <= 0:
        return []
    with get_conn() as conn:
        rows = _fetchall(
            conn,
            """SELECT * FROM learned_entries
               WHERE school_id = %s AND knowledge_id = %s AND status = 'draft'
                 AND (business_id = '' OR business_id = %s)
               ORDER BY contradicted ASC, confirmed DESC, observations DESC, id DESC
               LIMIT %s""",
            (_sid(), knowledge_id, business_id or "", int(limit)),
        )
    return [_row_to_learned(r) for r in rows]


def record_trials(task_id: str, entry_ids: list[int]) -> int:
    """记下这次任务注入了哪些待验证线索，结果待回填。"""
    if not entry_ids:
        return 0
    with get_conn() as conn:
        for entry_id in entry_ids:
            _execute(
                conn,
                """INSERT INTO learned_trials (school_id, entry_id, task_id) VALUES (%s, %s, %s)
                   ON CONFLICT (entry_id, task_id) DO NOTHING""",
                (_sid(), int(entry_id), task_id),
            )
        _execute(
            conn,
            f"""UPDATE learned_entries SET trials = trials + 1, last_trial_at = NOW()
                WHERE school_id = %s AND id IN ({",".join(["%s"] * len(entry_ids))})""",
            [_sid(), *[int(i) for i in entry_ids]],
        )
    return len(entry_ids)


def apply_trial_outcomes(task_id: str, outcomes: list[dict], min_confirm: int, max_contradict: int) -> dict:
    """回填 Agent 对每条线索的判断，并据此升级 / 降级。

    outcome 取 confirmed（与页面相符）/ contradicted（与页面不符）/ unused（这次没用上）。
    被证实够次数就升为 verified（来源记 trial）；被证伪到上限就降为 rejected，不再注入。
    """
    stats = {"confirmed": 0, "contradicted": 0, "unused": 0, "promoted": 0, "demoted": 0}
    if not outcomes:
        return stats
    with get_conn() as conn:
        for item in outcomes:
            try:
                entry_id = int(item.get("id") or 0)
            except (TypeError, ValueError):
                continue
            outcome = str(item.get("outcome") or "unused")
            if entry_id <= 0 or outcome not in ("confirmed", "contradicted", "unused"):
                continue
            note = str(item.get("note") or "")[:500]
            row = _fetchone(conn, "SELECT * FROM learned_entries WHERE id = %s AND school_id = %s", (entry_id, _sid()))
            if not row:
                continue
            stats[outcome] += 1
            _execute(
                conn,
                """INSERT INTO learned_trials (school_id, entry_id, task_id, outcome, note)
                   VALUES (%s, %s, %s, %s, %s)
                   ON CONFLICT (entry_id, task_id) DO UPDATE SET outcome = EXCLUDED.outcome, note = EXCLUDED.note""",
                (_sid(), entry_id, task_id, outcome, note),
            )
            if outcome == "unused":
                continue
            column = "confirmed" if outcome == "confirmed" else "contradicted"
            _execute(
                conn,
                f'UPDATE learned_entries SET {column} = {column} + 1, last_outcome = %s, updated_at = NOW() WHERE id = %s',
                (f"{outcome}：{note}"[:500], entry_id),
            )
            confirmed = int(row["confirmed"] or 0) + (1 if outcome == "confirmed" else 0)
            contradicted = int(row["contradicted"] or 0) + (1 if outcome == "contradicted" else 0)

            if row["status"] != "draft":
                continue
            if contradicted >= max_contradict:
                _execute(
                    conn,
                    """DELETE FROM learned_entries
                       WHERE school_id = %s AND knowledge_id = %s AND business_id = %s AND "key" = %s
                         AND status = 'rejected' AND id <> %s""",
                    (_sid(), row["knowledge_id"], row["business_id"], row["key"], entry_id),
                )
                _execute(conn, "UPDATE learned_entries SET status = 'rejected', updated_at = NOW() WHERE id = %s", (entry_id,))
                stats["demoted"] += 1
            elif confirmed >= min_confirm and contradicted == 0:
                _promote(conn, entry_id, "trial")
                stats["promoted"] += 1
    return stats


def trial_history(entry_id: int, limit: int = 20) -> list[dict]:
    with get_conn() as conn:
        rows = _fetchall(
            conn,
            "SELECT * FROM learned_trials WHERE entry_id = %s ORDER BY at DESC LIMIT %s",
            (int(entry_id), int(limit)),
        )
    return [{**r, "at": _fmt_dt(r.get("at"))} for r in rows]


def verified_learned(knowledge_id: str, business_id: str | None) -> dict[str, str]:
    """运行时注入用：全局条目（business_id 为空）+ 该业务条目，业务条目覆盖同键全局条目。"""
    with get_conn() as conn:
        rows = _fetchall(
            conn,
            """SELECT business_id, "key", "value" FROM learned_entries
               WHERE school_id = %s AND knowledge_id = %s AND status = 'verified'
                 AND (business_id = '' OR business_id = %s)
               ORDER BY business_id, id""",
            (_sid(), knowledge_id, business_id or ""),
        )
    merged: dict[str, str] = {}
    for r in rows:
        merged[r["key"]] = r["value"]
    return merged


# ---------- helpers ----------


def _load_json(value: Any, default: Any) -> Any:
    """JSONB 列 psycopg 直接返回 dict / list；兼容字符串形式。"""
    if value is None:
        return default
    if isinstance(value, (dict, list)):
        return value
    try:
        return json.loads(value)
    except (TypeError, ValueError):
        return default


def _fmt_dt(value: Any) -> Any:
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d %H:%M:%S")
    return value


def _row_to_task(row: dict) -> dict:
    return {
        "id": row["id"],
        "school_id": row.get("school_id"),
        "student_id": row["student_id"],
        "knowledge_id": row["knowledge_id"],
        "status": row["status"],
        "current_step": row["current_step"],
        "inputs": _load_json(row["inputs"], {}),
        "artifacts": _load_json(row["artifacts"], {}),
        "skill_version": row["skill_version"],
        "browser_profile_path": row["browser_profile_path"],
        "pause_reason": row["pause_reason"],
        "next_check_at": _fmt_dt(row["next_check_at"]),
        "created_at": _fmt_dt(row["created_at"]),
        "updated_at": _fmt_dt(row["updated_at"]),
    }