"""A confirmed worker close preserves partial validation but revokes readiness."""
from __future__ import annotations

import pytest

from apps.server.tests.test_discovery import item, search_candidate
from apps.server.tests.test_learning import client, initialize
from apps.server.tests.test_postgres import pg_client
from apps.server.tests.test_reading import NOTE_ID, TITLE, detail_output, learned_detail


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


def test_budget_close_keeps_partial_evidence_revokes_ready_and_never_replays(budget_client):
    client = budget_client
    initialize(client)
    _, search = search_candidate(client, [item(TITLE)])
    run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "准备装修"}).json()
    task, candidate, _ = learned_detail(client, seek=True)
    prefix = "/api/learning/tasks/" + task["id"]
    endpoint = "/api/discovery/runs/" + run["id"] + "/readings"
    payload = {"detail_candidate_id": candidate["id"], "note_id": NOTE_ID}
    first = client.post(endpoint, json=payload).json()
    assert first["status"] == "completed"
    store = client.app_reference.state.store
    old_candidate = store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],))
    old_run = store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],))
    old_reading = store.one("SELECT data_json FROM reading_runs WHERE id=$1", (first["id"],))
    old_validation = store.one("SELECT data_json FROM validations WHERE id=$1", (first["validation_id"],))
    worker = client.worker_reference
    original = worker.call
    flag = "true"

    async def timeout_with_partial(method, params):
        value = await original(method, params)
        if method == "verify" and params.get("note_binding"):
            value.update(passed=False, outputs={"note_detail": value["outputs"]["note_detail"]}, errors=["验证预算已到，保留已完成输出"], error_codes=["VERIFY_BUDGET"], session_closed=flag)
            if flag is True:
                worker.sessions.pop(params["task_id"], None)  # Simulate actual context close.
        return value

    worker.call = timeout_with_partial
    # Only an actual boolean close acknowledgment can invalidate the session.
    malformed = client.post(endpoint, json=payload).json()
    assert malformed["status"] == "partial"
    before_close = client.get(prefix).json()
    assert before_close["browser_readiness"]["ready"] is True
    flag = True
    response = client.post(endpoint, json=payload)
    assert response.status_code == 201, response.text
    partial = response.json()
    assert partial["status"] == "partial" and partial["detail"] == detail_output() and partial["comments"] is None
    assert partial["validation_id"] and partial["error_codes"] == ["VERIFY_BUDGET"]
    assert partial["seek_results"][0]["target_reached"] is True
    assert partial["evidence"] and all(client.get(item["screenshot_url"]).status_code == 200 for item in partial["evidence"])
    task_after = client.get(prefix).json()
    assert task_after["status"] == "paused" and task_after["browser_status"] == "closed"
    assert task_after["browser_readiness"]["ready"] is False and task_after["browser_readiness"]["next_action"] == "open_browser"
    validation = next(item for item in task_after["validations"] if item["id"] == partial["validation_id"])
    assert validation["status"] == "failed" and validation["error_codes"] == ["VERIFY_BUDGET"]
    assert validation["environment"]["browser_session_closed"] is True
    assert validation["actual_outputs"]["note_detail"] == detail_output() and validation["evidence"] == partial["evidence"]
    calls = len(worker.calls)
    next_reading = client.post(endpoint, json=payload).json()
    assert next_reading["status"] == "failed" and next_reading["validation_id"] is None
    assert next_reading["error_codes"] == ["browser_session_required"] and next_reading["detail"] is None
    assert worker.calls[calls:] == []  # No status/open/ready/verify or automatic retry.
    assert client.get("/api/discovery/readings/" + partial["id"]).json() == partial
    assert client.get("/api/discovery/readings/" + first["id"]).json() == first
    assert store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)) == old_candidate
    assert store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],)) == old_run
    assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (first["id"],)) == old_reading
    assert store.one("SELECT data_json FROM validations WHERE id=$1", (first["validation_id"],)) == old_validation