"""Fixed execution contracts; fake browser only, temporary SQLite / dedicated PG."""
from __future__ import annotations

import copy
import hashlib
import json
import os
import sqlite3
import subprocess
import sys
from pathlib import Path
from uuid import uuid4

import pytest

from apps.server.db import PostgresStore, SCHEMA, Store, digest, encoded
from apps.server.learning_contracts import EXECUTION_CONTRACT, EXECUTION_PROFILE, execution_metadata, execution_metadata_matches
from apps.server.tests.test_discovery import item, search_candidate
from apps.server.tests.test_learning import client, initialize, working_candidate
from apps.server.tests.test_postgres import pg_client
from apps.server.tests.test_reading import NOTE_ID, learned_detail
from apps.server.worker import safe_failure_capture, safe_request_observations

META = execution_metadata(EXECUTION_PROFILE)


@pytest.fixture(params=["client", "pg_client"])
def profile_client(request):
    return request.getfixturevalue(request.param)


def install_profile_worker(client):
    """Add actual worker echoes, with bounded corruption hooks for each RPC."""
    worker = client.worker_reference
    previous_call, previous_capture, previous_post = worker.call, worker.capture, client.post
    profiles, state = {}, {"method": None, "place": "root", "mutation": None}

    def capture(task_id):
        result = previous_capture(task_id)
        result.update(execution_metadata(profiles.get(task_id)))
        return result

    async def call(method, params):
        if method == "open":
            profiles[params["task_id"]] = params.get("execution_profile")
        result = await previous_call(method, params)
        if method != "close" and not (method == "status" and result.get("status") == "closed"):
            result.update(execution_metadata(profiles.get(params["task_id"])))
        if state["method"] == method:
            target = result if state["place"] == "root" else result[state["place"]]
            state["mutation"](target)
        return result

    def post(url, **kwargs):
        if str(url) == "/api/learning/tasks" and "json" in kwargs:
            kwargs["json"] = {"execution_profile": EXECUTION_PROFILE, **kwargs["json"]}
        return previous_post(url, **kwargs)

    worker.capture, worker.call, client.post = capture, call, post
    return state, profiles


def raw_facts(store):
    return {table: store.rows(f"SELECT * FROM {table} ORDER BY id") for table in ("observations", "learning_steps", "candidates", "validations")}


def test_fixed_descriptor_and_native_metadata_are_exact_not_caller_combinations():
    assert META == {"execution_profile": EXECUTION_PROFILE, "execution_contract": EXECUTION_CONTRACT}
    assert execution_metadata(None) == {}
    assert execution_metadata_matches({}, None)
    for value in ("legacy", "", {}, True):
        with pytest.raises(ValueError):
            execution_metadata(value)
    for value in ({}, {"execution_profile": EXECUTION_PROFILE}, META | {"execution_contract": EXECUTION_CONTRACT | {"extra": True}}, META | {"execution_contract": EXECUTION_CONTRACT | {"engine": "browser_use"}}):
        assert not execution_metadata_matches(value, EXECUTION_PROFILE)
    request = {"method": "POST", "origin": "https://www.xiaohongshu.com", "pathname": "/api/telemetry", "resource_type": "xhr", "automating": True, "blocked": False,
               "policy_version": "xhs-page-native-v1", "policy_rule": "native_page_traffic"}
    assert safe_request_observations([request])[0]["policy_rule"] == "native_page_traffic"
    for mutation in ({"policy_version": "xhs-readonly-ui-experimental-v1"}, {"policy_rule": "collect"}, {"blocked": True}, {"automating": False}):
        assert "policy_rule" not in safe_request_observations([request | mutation])[0]
    capture = {"url": "http://127.0.0.1/fixture", "elements": [], "screenshot_path": "synthetic.png", **META}
    assert execution_metadata_matches(safe_failure_capture(capture), EXECUTION_PROFILE)
    assert safe_failure_capture(capture | {"execution_contract": {}}) is None
    legacy = {key: value for key, value in capture.items() if key not in META}
    assert execution_metadata_matches(safe_failure_capture(legacy), None)


def test_nullable_legacy_and_explicit_profile_roundtrip_do_not_rewrite_history(profile_client):
    client = profile_client
    initialize(client)
    old_task, old_candidate = working_candidate(client)
    store = client.app_reference.state.store
    facts = raw_facts(store)
    assert store.one("SELECT execution_profile FROM learning_tasks WHERE id=$1", (old_task["id"],))["execution_profile"] is None
    assert "execution_profile" not in client.get("/api/learning/tasks/" + old_task["id"]).json()
    base = {key: old_task[key] for key in ("name", "platform", "entry_url", "goal", "account_key", "parameters")}
    null = client.post("/api/learning/tasks", json=base | {"execution_profile": None})
    assert null.status_code == 201 and "execution_profile" not in null.json()
    for bad in ("legacy", "browser_use", "", {}, True):
        assert client.post("/api/learning/tasks", json=base | {"execution_profile": bad}).status_code == 422
    install_profile_worker(client)
    task, candidate = working_candidate(client)
    assert task["execution_profile"] == EXECUTION_PROFILE and "execution_contract" not in task
    assert all(candidate["manifest"][key] == value for key, value in META.items())
    assert digest(candidate["manifest"]) == candidate["digest"]
    endpoint = "/api/learning/candidates/" + candidate["id"]
    checked = client.post(endpoint + "/check").json()
    assert checked["status"] == "passed" and "execution_profile" not in checked["environment"]
    verify = client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}}).json()
    assert verify["status"] == "passed" and all(verify["environment"][key] == value for key, value in META.items())
    wire = [params for method, params in client.worker_reference.calls if method == "verify"][-1]
    assert all(wire[key] == value for key, value in META.items())
    detail = client.get("/api/learning/tasks/" + task["id"]).json()
    assert all(detail["browser_readiness"][key] == value for key, value in META.items())
    assert all(execution_metadata_matches(value, EXECUTION_PROFILE) for value in detail["observations"])
    # Neither action nor candidate creation can override a task's contract.
    for path, data in (("/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results", "execution_profile": EXECUTION_PROFILE}),
                       ("/candidates", {"execution_profile": EXECUTION_PROFILE})):
        assert client.post("/api/learning/tasks/" + task["id"] + path, json=data).status_code == 422
    old = client.get("/api/learning/candidates/" + old_candidate["id"]).json()
    assert old["digest"] == old_candidate["digest"] and old["manifest"] == old_candidate["manifest"]
    current = raw_facts(store)
    for table, rows in facts.items():
        assert [row for row in current[table] if row["id"] in {item["id"] for item in rows}] == rows


def test_profile_echo_missing_open_status_capture_and_action_never_become_success(profile_client):
    client = profile_client
    initialize(client)
    state, _ = install_profile_worker(client)
    task, _ = working_candidate(client)
    prefix = "/api/learning/tasks/" + task["id"]
    store = client.app_reference.state.store
    old_steps = store.rows("SELECT * FROM learning_steps WHERE task_id=$1", (task["id"],))
    old_obs = store.one("SELECT count(*) AS n FROM observations WHERE task_id=$1", (task["id"],))["n"]
    def strip(value):
        value.pop("execution_profile", None)
        value.pop("execution_contract", None)
    for method, place, path, data in (("capture", "root", "/browser/capture", {}),
                                    ("action", "root", "/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results"}),
                                    ("action", "before", "/browser/action", {"kind": "read", "selector": '[data-testid="result"]', "output_name": "results"}),
                                    ("status", "root", "/browser/capture", {}), ("open", "root", "/browser/open", {})):
        state.update(method=None)
        assert client.post(prefix + "/browser/open").status_code == 200
        assert client.post(prefix + "/browser/ready", json={"identity_label": "测试"}).status_code == 200
        state.update(method=method, place=place, mutation=strip)
        response = client.post(prefix + path, json=data)
        assert response.status_code == 409, (method, place, response.text)
        assert not client.app_reference.state.learning.browser_readiness(task["id"])["ready"]
    assert store.rows("SELECT * FROM learning_steps WHERE task_id=$1", (task["id"],)) == old_steps
    assert store.one("SELECT count(*) AS n FROM observations WHERE task_id=$1", (task["id"],))["n"] == old_obs


def test_new_profile_cannot_relabel_old_evidence_or_accept_mutated_manifest(profile_client):
    client = profile_client
    initialize(client)
    old_task, candidate = working_candidate(client)
    service = client.app_reference.state.learning
    original = raw_facts(service.store)
    service.store.execute("UPDATE learning_tasks SET execution_profile=$1 WHERE id=$2", (EXECUTION_PROFILE, old_task["id"]))
    result = client.post("/api/learning/tasks/" + old_task["id"] + "/candidates", json={})
    assert result.status_code == 409 and result.json()["detail"]["code"] == "EXECUTION_PROFILE_MISMATCH"
    assert service.static_errors(candidate)
    for change in ({}, {"execution_profile": "unknown", "execution_contract": EXECUTION_CONTRACT}, META):
        changed = copy.deepcopy(candidate)
        changed["manifest"].update(change)
        changed["digest"] = digest(changed["manifest"])
        assert service.static_errors(changed)
    assert raw_facts(service.store) == original


def test_verify_bad_actual_profile_or_capture_records_failure_without_usable_outputs(profile_client):
    client = profile_client
    initialize(client)
    state, _ = install_profile_worker(client)
    task, candidate = working_candidate(client)
    endpoint = "/api/learning/candidates/" + candidate["id"]
    assert client.post(endpoint + "/check").json()["status"] == "passed"
    for mutate in (lambda result: result.pop("execution_contract"),
                   lambda result: result.update(execution_profile="unknown"),
                   lambda result: result["evidence"][0].pop("execution_profile")):
        state.update(method=None)
        prefix = "/api/learning/tasks/" + task["id"]
        client.post(prefix + "/browser/open")
        client.post(prefix + "/browser/ready", json={"identity_label": "测试"})
        state.update(method="verify", place="root", mutation=mutate)
        response = client.post(endpoint + "/verify", json={"expected_outputs": {"results": "建材"}})
        assert response.status_code == 200, response.text
        result = response.json()
        assert result["status"] == "failed" and "EXECUTION_PROFILE_MISMATCH" in result["error_codes"]
        assert result["actual_outputs"] == {} and result["evidence"] == []
        assert "execution_profile" not in result["environment"]


def test_new_discovery_and_reading_keep_actual_validation_profile_without_upgrading_old_search(profile_client):
    client = profile_client
    initialize(client)
    _, old_search = search_candidate(client, [item("准备装修求推荐靠谱公司")])
    old_run = client.post("/api/discovery/runs", json={"search_candidate_id": old_search["id"], "keyword": "装修", "region": "", "max_selected": 1}).json()
    assert old_run["status"] == "completed" and "execution_profile" not in old_run
    install_profile_worker(client)
    _, candidate, _ = learned_detail(client)
    result = client.post("/api/discovery/runs/" + old_run["id"] + "/readings", json={"detail_candidate_id": candidate["id"], "note_id": NOTE_ID})
    assert result.status_code == 201, result.text
    reading = result.json()
    assert reading["status"] == "completed" and all(reading[key] == value for key, value in META.items())
    assert client.get("/api/discovery/runs/" + old_run["id"]).json() == old_run
    _, new_search = search_candidate(client, [item("准备装修求推荐靠谱公司")])
    new_run = client.post("/api/discovery/runs", json={"search_candidate_id": new_search["id"], "keyword": "装修", "region": "", "max_selected": 1}).json()
    assert new_run["status"] == "completed" and all(new_run[key] == value for key, value in META.items())


def old_sqlite(home):
    home.mkdir(parents=True, exist_ok=True)
    target = home / "learning.sqlite"
    with sqlite3.connect(target) as conn:
        conn.executescript(SCHEMA.replace(", execution_profile TEXT", ""))
        conn.execute("INSERT INTO operators VALUES(?,?,?,?)", ("owner", "old", "not-used", "2026-10-25"))
        conn.execute("INSERT INTO learning_tasks VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", ("old-task", "owner", "old", "demo", "http://127.0.0.1/demo", "", "old", "{}", "draft", None, "2026-10-25", "2026-10-25"))
        manifest = {"schema_version": 2, "steps": [], "parameters": {}}
        conn.execute("INSERT INTO candidates VALUES(?,?,?,?,?,?,?)", ("old-candidate", "old-task", "old", 1, digest(manifest), encoded(manifest), "2026-10-25"))
    return target


def test_sqlite_existing_schema_upgrade_is_idempotent_and_preserves_old_manifest(tmp_path):
    target = old_sqlite(tmp_path / "old")
    with sqlite3.connect(target) as conn:
        original = conn.execute("SELECT digest,manifest_json FROM candidates").fetchall()
    store = Store(target.parent)
    assert store.one("SELECT execution_profile FROM learning_tasks")["execution_profile"] is None
    assert store.one("SELECT version FROM schema_versions WHERE version=10")
    assert [(row["digest"], row["manifest_json"]) for row in store.rows("SELECT digest,manifest_json FROM candidates")] == original
    assert Store(target.parent).one("SELECT count(*) AS n FROM schema_versions WHERE version=10")["n"] == 1


def test_postgres_existing_column_upgrade_and_old_sqlite_backup_migration(pg_client, tmp_path):
    store = pg_client.app_reference.state.store
    store.execute("ALTER TABLE learning_tasks DROP COLUMN execution_profile")
    upgraded = PostgresStore(store.home, pg_client.test_database_url, schema=pg_client.test_schema)
    assert upgraded.one("SELECT count(*) AS n FROM information_schema.columns WHERE table_schema=$1 AND table_name='learning_tasks' AND column_name='execution_profile'", (pg_client.test_schema,))["n"] == 1
    assert upgraded.one("SELECT version FROM schema_versions WHERE version=10")
    home = tmp_path / "legacy-import"
    source = old_sqlite(home)
    original_hash = hashlib.sha256(source.read_bytes()).hexdigest()
    schema = "assistant_learning_test_" + uuid4().hex
    assert pg_client.test_database_url and "/assistant_learning_test" in pg_client.test_database_url
    env = {**os.environ, "ASSISTANT_DATABASE_URL": pg_client.test_database_url, "ASSISTANT_DATABASE_SCHEMA": schema, "ASSISTANT_LEARNING_HOME": str(home), "PYTHONUTF8": "1"}
    try:
        completed = subprocess.run([sys.executable, "-m", "scripts.migrate_learning"], cwd=Path(__file__).resolve().parents[3], env=env, capture_output=True, timeout=60)
        assert completed.returncode == 0, "isolated legacy migration failed"
        summary = json.loads(completed.stdout)
        assert summary["result"] == "copied"
        imported = PostgresStore(home, pg_client.test_database_url, schema=schema)
        assert imported.one("SELECT execution_profile FROM learning_tasks")["execution_profile"] is None
        assert imported.one("SELECT manifest_json FROM candidates")["manifest_json"] == encoded({"schema_version": 2, "steps": [], "parameters": {}})
        assert hashlib.sha256(source.read_bytes()).hexdigest() == original_hash
        backups = list((home / "backups").glob("*.sqlite"))
        assert len(backups) == 1
        with sqlite3.connect(backups[0]) as conn:
            assert "execution_profile" not in {row[1] for row in conn.execute("PRAGMA table_info(learning_tasks)")}
    finally:
        import psycopg
        from psycopg import sql
        with psycopg.connect(pg_client.test_database_url, connect_timeout=5) as conn:
            conn.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema)))