"""Tests for mobile UI structure and key script behaviours.

These tests verify the HTML/JS contract without running a browser:
- Required mobile meta tags and manifest link
- CSS dvh / safe-area-inset usage
- Key JavaScript identifiers and flow strings
- PWA static assets served correctly
- Log tail endpoint parameter pass-through
"""
import json

import pytest
from fastapi.testclient import TestClient

from dual_agent.web import INDEX_HTML


def test_claude_login_has_authorization_code_submission():
    assert 'id="auth-code"' in INDEX_HTML
    assert 'function submitLoginCode(provider)' in INDEX_HTML
    assert '/login/code' in INDEX_HTML
from dual_agent.api import create_app, MANIFEST, ICON_SVG, SERVICE_WORKER_JS, ICON_192_PNG, ICON_512_PNG
from tests._fakes import fake_orchestrator


@pytest.fixture()
def client():
    return TestClient(create_app(fake_orchestrator()))


# ── HTML structure ─────────────────────────────────────────────────────

def test_viewport_meta_present():
    assert 'name="viewport"' in INDEX_HTML
    assert "viewport-fit=cover" in INDEX_HTML
    assert "width=device-width" in INDEX_HTML


def test_dvh_used_for_height():
    assert "100dvh" in INDEX_HTML


def test_safe_area_inset_bottom_used():
    assert "env(safe-area-inset-bottom" in INDEX_HTML


def test_manifest_link_present():
    assert 'href="/manifest.webmanifest"' in INDEX_HTML


def test_apple_mobile_web_app_metas():
    assert 'apple-mobile-web-app-capable' in INDEX_HTML
    assert 'apple-mobile-web-app-status-bar-style' in INDEX_HTML


def test_pwa_service_worker_registration():
    assert "serviceWorker" in INDEX_HTML
    assert "/sw.js" in INDEX_HTML


def test_connection_banner_present():
    assert 'id="conn-banner"' in INDEX_HTML


def test_mobile_tabs_present():
    assert 'id="tabs"' in INDEX_HTML
    assert "showPanel" in INDEX_HTML


def test_confirm_overlay_present():
    assert 'id="confirm-overlay"' in INDEX_HTML
    assert "resolveConfirm" in INDEX_HTML
    assert "mobileConfirm" in INDEX_HTML


def test_polling_present():
    assert "startPolling" in INDEX_HTML
    assert "POLL_ACTIVE" in INDEX_HTML
    assert "POLL_HIDDEN" in INDEX_HTML
    assert "visibilitychange" in INDEX_HTML


def test_connection_tracking_present():
    assert "setConn" in INDEX_HTML
    assert "TypeError" in INDEX_HTML  # network-error detection


def test_request_id_sent_with_actions():
    # X-Request-Id header must be passed for every state-changing call
    assert "'X-Request-Id'" in INDEX_HTML


def test_danger_confirm_for_cancel():
    assert "act('cancel'" in INDEX_HTML


def test_danger_confirm_for_run_all():
    assert "runAll" in INDEX_HTML
    assert "连续自动执行" in INDEX_HTML  # confirm message


def test_gate_warning_for_replan():
    assert "退回规划" in INDEX_HTML


def test_log_tail_auto_load():
    assert "appendLogTail" in INDEX_HTML
    assert "showFullLogs" in INDEX_HTML


# ── mobile-logs-terminal-hidden fix verification ───────────────────────

def test_log_tail_not_gated_on_terminal_state():
    """appendLogTail must be called unconditionally so DONE/NEEDS_HUMAN tasks show logs.

    Verify that the call to appendLogTail(chat) is not wrapped inside an
    if(!isTerminal) block.  The chat.innerHTML assignment always precedes it
    in refreshTask(), and there must be no isTerminal condition between them.
    """
    chat_pos = INDEX_HTML.find("chat.innerHTML=statusHtml")
    append_pos = INDEX_HTML.find("appendLogTail(chat)")
    assert chat_pos != -1, "chat.innerHTML=statusHtml assignment not found in refreshTask"
    assert append_pos != -1, "appendLogTail(chat) call not found"
    assert append_pos > chat_pos, "appendLogTail(chat) must appear after chat.innerHTML="
    between = INDEX_HTML[chat_pos:append_pos]
    # Strip whitespace variations before searching
    between_compact = between.replace(" ", "")
    assert "if(!isTerminal)" not in between_compact, (
        "appendLogTail(chat) is gated on !isTerminal; DONE/NEEDS_HUMAN tasks won't show log tail"
    )


def test_full_logs_button_in_log_tail():
    """The 完整日志 (full logs) button must be present inside appendLogTail's output."""
    assert "完整日志" in INDEX_HTML, "Full-logs button label must exist in appendLogTail"
    assert "showFullLogs" in INDEX_HTML, "showFullLogs handler must be wired up"


def test_log_tail_shown_for_terminal_task_via_api():
    """GET /tasks/{id}/logs must return log data for tasks in any state including DONE."""
    from pathlib import Path
    from tests._fakes import task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "DONE")

    logs_dir = orc.store.task_dir(task.task_id) / "logs"
    logs_dir.mkdir(parents=True, exist_ok=True)
    (logs_dir / "agent.log").write_text("done log line\n")

    c = TestClient(create_app(orc))
    r = c.get(f"/tasks/{task.task_id}/logs?tail=5")
    assert r.status_code == 200
    data = r.json()
    assert data["files"], "DONE task must return log files"
    assert "done log line" in data["files"][0]["tail"]


# ── mobile-device-coverage-gap: structural assertions ─────────────────
# Real-device testing (320/375/430px, keyboard, safe-area, background
# recovery) requires physical devices or a device cloud and is documented
# in tests/test_browser.py manual verification matrix.  The tests below
# assert the structural CSS/JS contracts that enable those behaviours.

def test_safe_area_inset_used_for_composer():
    """Composer / bottom input area must use safe-area-inset-bottom."""
    assert "env(safe-area-inset-bottom" in INDEX_HTML


def test_dvh_viewport_height():
    """Layout height must use 100dvh (Dynamic Viewport Height) for mobile browsers."""
    assert "100dvh" in INDEX_HTML


def test_visibility_change_triggers_refresh():
    """visibilitychange event must trigger an immediate refresh when page becomes visible."""
    assert "visibilitychange" in INDEX_HTML
    assert "document.hidden" in INDEX_HTML


def test_connection_banner_exists_for_weak_network():
    """Connection banner must exist to inform the user of network failures."""
    assert 'id="conn-banner"' in INDEX_HTML
    assert "setConn" in INDEX_HTML


def test_action_buttons_have_min_touch_target():
    """Button touch targets must be at least 44px for reliable mobile interaction."""
    assert "min-height:44px" in INDEX_HTML


def test_state_labels_defined():
    for state in ["INIT", "PLAN", "IMPLEMENT", "TEST", "REVIEW", "FIX", "DONE", "NEEDS_HUMAN"]:
        assert state in INDEX_HTML


def test_error_category_labels_defined():
    for cat in ["AUTH", "RATE_LIMIT", "TIMEOUT", "AGENT_FAILURE", "MAX_CYCLES"]:
        assert cat in INDEX_HTML


def test_task_status_card_rendered():
    assert "status-card" in INDEX_HTML
    assert "state-badge" in INDEX_HTML
    assert "err-box" in INDEX_HTML


def test_back_button_for_mobile():
    assert "bar-back" in INDEX_HTML
    assert "showPanel" in INDEX_HTML


def test_touch_min_height_set():
    # min-height:44px for touch targets
    assert "min-height:44px" in INDEX_HTML


# ── API: PWA assets ────────────────────────────────────────────────────

def test_manifest_endpoint(client):
    r = client.get("/manifest.webmanifest")
    assert r.status_code == 200
    data = r.json()
    assert data["name"] == "Dual Agent Orchestrator"
    assert data["display"] == "standalone"
    assert any(icon["src"] == "/icon.svg" for icon in data["icons"])


def test_manifest_has_start_url(client):
    r = client.get("/manifest.webmanifest")
    assert r.json()["start_url"] == "/"


def test_sw_endpoint(client):
    r = client.get("/sw.js")
    assert r.status_code == 200
    assert "fetch" in r.text
    assert "Cache-Control" in r.headers
    # SW must not cache API responses
    assert "cache" not in r.text.lower() or "Network-first" in r.text or "network" in r.text.lower()


def test_sw_no_cache_header(client):
    r = client.get("/sw.js")
    assert "no-cache" in r.headers.get("Cache-Control", "")


def test_icon_endpoint(client):
    r = client.get("/icon.svg")
    assert r.status_code == 200
    assert r.headers["content-type"].startswith("image/svg+xml")
    assert "<svg" in r.text


def test_index_returns_html(client):
    r = client.get("/")
    assert r.status_code == 200
    assert "text/html" in r.headers["content-type"]
    assert "<!doctype html>" in r.text.lower()


def test_health_endpoint_always_200(client):
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"


# ── API: Log tail parameter ───────────────────────────────────────────

def test_logs_404_when_no_logs(client):
    r = client.get("/tasks/nonexistent_task_id_x/logs")
    assert r.status_code == 404


# ── Concurrent idempotency ─────────────────────────────────────────────

def test_concurrent_same_request_id_only_one_executes():
    """Two threads sending the same X-Request-Id must not both execute the action."""
    import threading
    from tests._fakes import fake_orchestrator, task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    # One app instance so both threads share the same _idempotency store.
    app = create_app(orc)
    rid = "concurrent-idem-test"

    results = []

    def do_advance():
        # Each thread owns its own TestClient to avoid httpx client threading issues.
        c = TestClient(app, raise_server_exceptions=False)
        r = c.post(
            f"/tasks/{task.task_id}/advance",
            headers={"X-Request-Id": rid},
        )
        results.append(r.status_code)

    t1 = threading.Thread(target=do_advance)
    t2 = threading.Thread(target=do_advance)
    t1.start(); t2.start()
    t1.join(); t2.join()

    # At least one must succeed (200); the duplicate gets cached 200 or 409 (in-flight).
    assert 200 in results, f"Expected at least one 200, got {results}"
    # The task must have advanced exactly once: if both got 200, they saw the same state.
    if results.count(200) == 2:
        # Both returned 200 — the second must be the cached result (same state value).
        pass  # acceptable: second hit cache before in-flight was detected
    final = orc.get_task(task.task_id)
    assert final.state.value != "INIT" or all(r == 409 for r in results)


def test_idempotency_store_claim_is_atomic():
    """claim() must return _PENDING for a concurrent call, never None twice."""
    from dual_agent.api import _IdempotencyStore, _PENDING

    store = _IdempotencyStore()
    key = "test-atomic-key"

    first = store.claim(key)
    assert first is None  # First claim succeeds

    second = store.claim(key)
    assert second is _PENDING  # Second claim sees in-flight

    # After set(), a third claim returns the cached value.
    store.set(key, {"state": "PLAN"})
    third = store.claim(key)
    assert third == {"state": "PLAN"}


def test_idempotency_discard_allows_retry():
    """discard() releases an in-flight claim so the client can retry."""
    from dual_agent.api import _IdempotencyStore, _PENDING

    store = _IdempotencyStore()
    key = "discard-key"

    store.claim(key)  # claim it
    store.discard(key)  # release on failure
    result = store.claim(key)
    assert result is None  # should be claimable again


# ── Resume / NEEDS_HUMAN recovery ─────────────────────────────────────

def test_resume_recovers_needs_human_task():
    """POST /resume on a non-operator NEEDS_HUMAN task clears the terminal state."""
    from tests._fakes import fake_orchestrator, task_in_state
    from dual_agent.domain import ErrorCategory

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    # Drive to NEEDS_HUMAN via cancel
    orc.cancel(task.task_id)
    # Manually set error_category to AGENT_FAILURE (not OPERATOR) so recovery is allowed
    nh_task = orc.get_task(task.task_id)
    orc.store.save(nh_task.model_copy(update={"error_category": ErrorCategory.AGENT_FAILURE}))

    app_client = TestClient(create_app(orc), raise_server_exceptions=False)
    r = app_client.post(f"/tasks/{task.task_id}/resume")
    assert r.status_code == 200
    assert r.json()["state"] != "NEEDS_HUMAN"


def test_resume_operator_cancel_returns_409():
    """POST /resume on an operator-cancelled task must return 409."""
    from tests._fakes import fake_orchestrator, task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    orc.cancel(task.task_id)  # puts task in NEEDS_HUMAN with OPERATOR category

    app_client = TestClient(create_app(orc), raise_server_exceptions=False)
    r = app_client.post(f"/tasks/{task.task_id}/resume")
    assert r.status_code == 409


def test_resume_done_task_returns_task():
    """POST /resume on a DONE task returns the completed task as-is."""
    from tests._fakes import fake_orchestrator, task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "DONE")
    app_client = TestClient(create_app(orc))
    r = app_client.post(f"/tasks/{task.task_id}/resume")
    assert r.status_code == 200
    assert r.json()["state"] == "DONE"


# ── AbortController / connection timeout ───────────────────────────────

def test_abort_controller_present():
    """api() must have AbortController to prevent hung fetch requests."""
    assert "AbortController" in INDEX_HTML
    assert "AbortError" in INDEX_HTML


def test_api_timeout_clears_on_success():
    """clearTimeout must be called after a successful fetch (leak prevention)."""
    assert "clearTimeout" in INDEX_HTML


def test_polling_handles_5xx_as_connection_error():
    """startPolling must treat httpStatus>=500 as a connection error, not task failure."""
    assert "httpStatus>=500" in INDEX_HTML or "httpStatus >= 500" in INDEX_HTML
    assert "服务暂时不可用" in INDEX_HTML


# ── startDiscussion idempotency ────────────────────────────────────────

def test_start_discussion_sends_request_id():
    """startDiscussion must send X-Request-Id to POST /discussions for idempotency."""
    assert "'X-Request-Id':reqId" in INDEX_HTML or "'X-Request-Id': reqId" in INDEX_HTML


def test_start_discussion_uses_guard():
    """startDiscussion must use guard() to prevent duplicate submissions."""
    assert "正在开启讨论" in INDEX_HTML


# ── Logs pagination ────────────────────────────────────────────────────

def test_logs_total_lines_used_in_js():
    """Frontend must read total_lines from log response for pagination."""
    assert "total_lines" in INDEX_HTML


def test_logs_pagination_controls_present():
    """Log viewer must show pagination controls for large logs."""
    assert "加载更早内容" in INDEX_HTML
    assert "已全部加载" in INDEX_HTML
    assert "dataset.loaded" in INDEX_HTML


def test_logs_offset_api():
    """GET /tasks/{id}/logs with offset parameter returns paginated content with total_lines."""
    from pathlib import Path

    orc = fake_orchestrator()
    from tests._fakes import task_in_state
    task = task_in_state(orc, "INIT")

    logs_dir = orc.store.task_dir(task.task_id) / "logs"
    logs_dir.mkdir(parents=True, exist_ok=True)
    log_file = logs_dir / "agent.log"
    log_file.write_text("".join(f"line {i}\n" for i in range(50)))

    c = TestClient(create_app(orc))

    # Default tail=10: last 10 lines (lines 40-49)
    r = c.get(f"/tasks/{task.task_id}/logs?tail=10")
    assert r.status_code == 200
    data = r.json()
    assert data["files"][0]["total_lines"] == 50
    assert "line 49" in data["files"][0]["tail"]
    assert "line 39" not in data["files"][0]["tail"]

    # offset=10: skip last 10, take 10 before that (lines 30-39)
    r2 = c.get(f"/tasks/{task.task_id}/logs?tail=10&offset=10")
    assert r2.status_code == 200
    data2 = r2.json()
    assert data2["files"][0]["total_lines"] == 50
    assert "line 39" in data2["files"][0]["tail"]
    assert "line 40" not in data2["files"][0]["tail"]

    # offset=50: beyond all lines → empty tail
    r3 = c.get(f"/tasks/{task.task_id}/logs?tail=10&offset=50")
    assert r3.status_code == 200
    assert r3.json()["files"][0]["tail"] == ""


# ── Discussion idempotency ─────────────────────────────────────────────

def test_start_discussion_idempotency():
    """POST /discussions with same X-Request-Id must not create duplicate discussions."""
    from unittest.mock import MagicMock
    from dual_agent.discussion import DiscussionRecord
    from pathlib import Path

    orc = fake_orchestrator()
    mock_record = DiscussionRecord(discussion_id="disc-idem-test", repo_path=Path("/tmp"))
    mock_advisor = MagicMock()
    mock_advisor.start.return_value = mock_record

    c = TestClient(create_app(orc, advisor=mock_advisor), raise_server_exceptions=False)
    rid = "disc-create-idem-01"
    r1 = c.post("/discussions",
                json={"repo_path": "/tmp", "opening": "test question"},
                headers={"X-Request-Id": rid})
    r2 = c.post("/discussions",
                json={"repo_path": "/tmp", "opening": "test question"},
                headers={"X-Request-Id": rid})

    assert r1.status_code == 201
    assert r2.status_code == 201
    # advisor.start() must only be called once; second response is cached
    assert mock_advisor.start.call_count == 1


def test_discussion_to_task_idempotency():
    """POST /discussions/{id}/task with same X-Request-Id must not create duplicate tasks."""
    from unittest.mock import MagicMock
    from dual_agent.domain import TaskSpec
    from pathlib import Path
    import tempfile

    orc = fake_orchestrator()
    repo = Path(tempfile.mkdtemp())
    import subprocess
    subprocess.run(["git", "-C", str(repo), "init", "-q"], check=True)
    subprocess.run(
        ["git", "-C", str(repo), "-c", "user.name=T", "-c", "user.email=t@t.invalid",
         "commit", "--allow-empty", "-m", "init", "-q"],
        check=True,
    )
    real_task = orc.create_task(TaskSpec(repo_path=repo, goal="test"))

    mock_advisor = MagicMock()
    mock_advisor.to_task.return_value = real_task

    c = TestClient(create_app(orc, advisor=mock_advisor), raise_server_exceptions=False)
    rid = "to-task-idem-01"
    r1 = c.post("/discussions/some-disc-id/task", headers={"X-Request-Id": rid})
    r2 = c.post("/discussions/some-disc-id/task", headers={"X-Request-Id": rid})

    assert r1.status_code == 201
    assert r2.status_code == 201
    assert mock_advisor.to_task.call_count == 1


# ── LAN startup banner ─────────────────────────────────────────────────

def test_stage_progress_bar_present():
    """UI must contain progress bar elements and STAGE_ORDER for task progress."""
    assert "STAGE_ORDER" in INDEX_HTML
    assert "stageProgressHtml" in INDEX_HTML
    assert "progress-fill" in INDEX_HTML


def test_offline_503_detection_present():
    """api() must handle service-worker synthetic 503 as a connection error."""
    assert "detail==='offline'" in INDEX_HTML or "detail==\\'offline\\'" in INDEX_HTML or "'offline'" in INDEX_HTML
    assert "offline" in INDEX_HTML


# ── PWA shell caching (offline-pwa-shell-not-cached) ──────────────────

def test_sw_shell_includes_root():
    """Service worker SHELL list must include '/' so the app shell is cached for offline."""
    assert "'/'," in SERVICE_WORKER_JS or "'/'" in SERVICE_WORKER_JS


def test_sw_offline_navigate_can_match_root():
    """SW offline fallback uses caches.match('/') — '/' must be in the cache list."""
    assert "caches.match('/')" in SERVICE_WORKER_JS
    # '/' must be in SHELL so it gets cached and match() can return something useful
    assert "'/'" in SERVICE_WORKER_JS


# ── Retry reqId reuse (retry-generates-new-operation-id) ──────────────

def test_guard_op_key_reuse_mechanism_present():
    """guard() must retain reqId across network failures via _opReqIds map."""
    assert "_opReqIds" in INDEX_HTML


def test_guard_accepts_op_key_parameter():
    """All state-changing callers must pass an opKey so retries reuse the same reqId."""
    # Check that guard calls include opKey arguments (template literals with task/disc prefix)
    assert "task:advance:" in INDEX_HTML
    # approve/cancel/resume/run go through act() which uses `task:${name}:${current}`
    assert "task:${name}:" in INDEX_HTML
    assert "disc:say:" in INDEX_HTML
    assert "task:say:" in INDEX_HTML


def test_guard_clears_op_key_on_success():
    """guard() must delete the opKey entry after a successful request (not on network error)."""
    assert "networkFailed" in INDEX_HTML
    assert "delete _opReqIds" in INDEX_HTML


# ── /cookie login bridge (token-query-leaks-to-logs) ──────────────────

def test_login_bridge_html_exported():
    """LOGIN_BRIDGE_HTML constant must exist in api module."""
    from dual_agent.api import LOGIN_BRIDGE_HTML
    assert "token" in LOGIN_BRIDGE_HTML.lower()
    assert "/cookie" in LOGIN_BRIDGE_HTML


def test_cookie_endpoint_in_sw_js_exempt():
    """Service worker must not interfere with POST /cookie (it's a navigate, not shell)."""
    # SW uses network-first, so /cookie is always fetched from network. Just verify
    # the navigate fallback doesn't pretend /cookie is the root.
    assert "navigate" in SERVICE_WORKER_JS


def test_say_idempotency_key_includes_text():
    """task:say opKey must include message text so different messages get different IDs."""
    assert "task:say:${current}:${text}" in INDEX_HTML


# ── Auto-reply stable request ID (mobile-review-001) ──────────────────

def test_auto_reply_uses_stable_request_id():
    """startDiscussion() auto-reply must use record.discussion_id+':auto-reply'
    as the X-Request-Id, not genId().  A stable ID ensures that if the reply
    request times out and the user retries, the server-side idempotency cache
    blocks the duplicate reply execution."""
    assert "record.discussion_id+':auto-reply'" in INDEX_HTML


# ── POST /discussions body-hash in cache key (mobile-review-003) ───────

def test_start_discussion_different_body_same_request_id_not_cached():
    """POST /discussions with the same X-Request-Id but a different body must
    not return the cached result of the first request (the body hash is part
    of the cache key so each distinct payload gets an independent slot)."""
    from unittest.mock import MagicMock
    from dual_agent.discussion import DiscussionRecord
    from pathlib import Path

    orc = fake_orchestrator()
    records = [
        DiscussionRecord(discussion_id="disc-body-01", repo_path=Path("/tmp")),
        DiscussionRecord(discussion_id="disc-body-02", repo_path=Path("/tmp")),
    ]
    call_count = [0]

    def _side_effect(spec):
        r = records[min(call_count[0], 1)]
        call_count[0] += 1
        return r

    mock_advisor = MagicMock()
    mock_advisor.start.side_effect = _side_effect

    c = TestClient(create_app(orc, advisor=mock_advisor), raise_server_exceptions=False)
    rid = "same-rid-diff-body"
    r1 = c.post("/discussions",
                json={"repo_path": "/tmp", "opening": "question one"},
                headers={"X-Request-Id": rid})
    r2 = c.post("/discussions",
                json={"repo_path": "/tmp", "opening": "question two"},
                headers={"X-Request-Id": rid})

    assert r1.status_code == 201
    assert r2.status_code == 201
    assert mock_advisor.start.call_count == 2, (
        "Different body should produce a different cache key and re-execute advisor.start"
    )
    assert r1.json()["discussion_id"] == "disc-body-01"
    assert r2.json()["discussion_id"] == "disc-body-02"


# ── Poll generation counter (mobile-review-003) ────────────────────────

def test_poll_generation_counter_declared():
    """_pollGen must be declared in script state so timer accumulation is prevented."""
    assert "_pollGen" in INDEX_HTML


def test_poll_generation_checked_in_tick():
    """tick() must compare gen against _pollGen before executing and before rescheduling."""
    assert "gen!==_pollGen" in INDEX_HTML or "gen !== _pollGen" in INDEX_HTML
    assert "gen===_pollGen" in INDEX_HTML or "gen === _pollGen" in INDEX_HTML


def test_stop_polling_invalidates_generation():
    """stopPolling() must increment _pollGen to invalidate in-flight ticks."""
    assert "_pollGen++" in INDEX_HTML


def test_start_polling_captures_generation():
    """startPolling() must capture gen=_pollGen immediately after stopPolling so
    any tick spawned by the previous startPolling() is automatically invalidated."""
    assert "const gen=_pollGen" in INDEX_HTML or "const gen = _pollGen" in INDEX_HTML


# ── Cookie security attributes (mobile-review-002) ────────────────────

def test_cookie_secure_when_behind_https_proxy():
    """POST /cookie with X-Forwarded-Proto: https must set the Secure cookie flag."""
    from tests._fakes import fake_orchestrator as _fo
    orc = _fo()
    c = TestClient(create_app(orc, token="tok"), raise_server_exceptions=False)
    r = c.post(
        "/cookie",
        headers={"Authorization": "Bearer tok", "X-Forwarded-Proto": "https"},
    )
    assert r.status_code == 204
    assert "Secure" in r.headers.get("Set-Cookie", "")


def test_cookie_not_secure_over_plain_http():
    """POST /cookie without X-Forwarded-Proto must NOT set Secure (cookie works over HTTP)."""
    from tests._fakes import fake_orchestrator as _fo
    orc = _fo()
    c = TestClient(create_app(orc, token="tok"), raise_server_exceptions=False)
    r = c.post("/cookie", headers={"Authorization": "Bearer tok"})
    assert r.status_code == 204
    assert "Secure" not in r.headers.get("Set-Cookie", "")


def test_cookie_has_httponly_and_samesite():
    """Cookie must always have HttpOnly and SameSite=strict regardless of HTTPS mode."""
    from tests._fakes import fake_orchestrator as _fo
    orc = _fo()
    c = TestClient(create_app(orc, token="tok"), raise_server_exceptions=False)
    r = c.post("/cookie", headers={"Authorization": "Bearer tok"})
    assert r.status_code == 204
    cookie_header = r.headers.get("Set-Cookie", "").lower()
    assert "httponly" in cookie_header
    assert "samesite=strict" in cookie_header


# ── LAN HTTP plaintext warning (mobile-review-002) ────────────────────

def test_lan_http_plaintext_warning_in_cli():
    """serve command must emit a plaintext-over-HTTP warning when LAN mode is active."""
    from dual_agent.cli import _lan_ips  # noqa: PLC2701  – internal but stable
    # Check that the warning text exists in cli.py source
    import inspect
    from dual_agent import cli
    src = inspect.getsource(cli)
    assert "PLAINTEXT" in src or "plaintext" in src or "UNENCRYPTED" in src or "unencrypted" in src, (
        "serve command must warn that LAN HTTP transmits the token in cleartext"
    )


# ── Page-refresh retry safety (mobile-review-001) ─────────────────────

def test_page_refresh_retry_with_new_reqid_hits_state_machine():
    """After a page refresh the client generates a new reqId.  The server-side
    idempotency cache is empty for the new key so the action re-executes.
    The state machine (InvalidTransition / TaskBusy) must prevent invalid transitions.
    """
    from tests._fakes import fake_orchestrator, task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    c = TestClient(create_app(orc), raise_server_exceptions=False)

    # First advance: INIT → some state
    r1 = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": "pre-refresh-001"})
    assert r1.status_code == 200
    state_after_first = r1.json()["state"]

    # Simulate page refresh: new reqId is generated (old one is lost in JS memory).
    # The server re-executes; state machine may advance again or return 409.
    r2 = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": "post-refresh-001"})
    assert r2.status_code in (200, 409), f"Expected 200 or 409, got {r2.status_code}"

    # Task must remain in a valid, non-INIT state regardless of outcome.
    final = orc.get_task(task.task_id)
    assert final.state is not None
    assert final.state.value != "INIT"


# ── guard() opKey retention across refresh failures (refresh-failure-drops-operation-retry-key) ──

def test_guard_catches_refresh_errors_not_unhandled():
    """guard() must catch refresh() errors rather than letting them propagate unhandled."""
    assert "try{await refresh();}catch" in INDEX_HTML, (
        "guard() must wrap refresh() in try/catch so refresh failures don't become "
        "unhandled Promise rejections that silently drop the opKey"
    )


def test_guard_retains_opkey_when_fn_succeeds_but_refresh_fails():
    """When fn() succeeds but refresh() fails, opKey must be retained for the next retry."""
    assert "fnSucceeded" in INDEX_HTML, "guard() must track whether fn() itself succeeded"
    assert "refreshNetFailed" in INDEX_HTML, "guard() must track whether refresh() failed"


def test_guard_opkey_deletion_is_after_refresh():
    """opKey must only be deleted AFTER the refresh() try/catch, not before it."""
    refresh_pos = INDEX_HTML.find("try{await refresh();}catch")
    delete_pos = INDEX_HTML.find("delete _opReqIds[opKey]")
    assert refresh_pos != -1, "try{await refresh();}catch not found in guard()"
    assert delete_pos != -1, "delete _opReqIds[opKey] not found"
    assert delete_pos > refresh_pos, (
        f"delete _opReqIds (offset {delete_pos}) must come AFTER the refresh() try/catch "
        f"(offset {refresh_pos}), otherwise opKey is lost before refresh confirms success"
    )


# ── Cancel-recovery contract (cancel-recovery-contract-mismatch) ──────

def test_operator_cancel_hides_resume_button():
    """Resume button must be hidden for OPERATOR-cancelled tasks (server rejects recovery)."""
    assert "error_category!=='OPERATOR'" in INDEX_HTML, (
        "resumeBtn condition must exclude OPERATOR-cancelled tasks via error_category!=='OPERATOR'"
    )


def test_cancel_confirmation_reflects_no_recovery():
    """Cancel confirmation must not promise recovery; must direct user to create a new task."""
    assert "不可通过恢复" in INDEX_HTML or "须创建新任务" in INDEX_HTML, (
        "Cancel confirm text must say the task cannot be recovered via resume "
        "and that a new task is needed to continue"
    )


def test_operator_cancel_status_card_message():
    """NEEDS_HUMAN + OPERATOR category must display a 'create new task' message in status card."""
    assert "error_category==='OPERATOR'" in INDEX_HTML, (
        "UI must check error_category==='OPERATOR' to show the create-new-task guidance "
        "when a task is in NEEDS_HUMAN state due to operator cancellation"
    )


# ── Concurrent state mutation safety (unsafe-concurrent-state-mutations) ──

def test_approve_uses_task_lock():
    """approve() must acquire the in-process and file-based task locks like advance() does."""
    from dual_agent.services import Orchestrator
    import inspect
    src = inspect.getsource(Orchestrator.approve)
    assert "_lock_for" in src, "approve() must acquire the in-process task lock via _lock_for()"
    assert "task_lock" in src, "approve() must acquire the file-based task lock via task_lock()"


def test_say_uses_task_lock():
    """say() must acquire the in-process and file-based task locks like advance() does."""
    from dual_agent.services import Orchestrator
    import inspect
    src = inspect.getsource(Orchestrator.say)
    assert "_lock_for" in src, "say() must acquire the in-process task lock via _lock_for()"
    assert "task_lock" in src, "say() must acquire the file-based task lock via task_lock()"


def test_approve_and_advance_concurrent_leave_valid_state():
    """approve() and advance() running concurrently must leave the task in a valid state."""
    from tests._fakes import task_in_state
    import threading
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    app = create_app(orc)
    results = []

    def do_approve():
        c = TestClient(app, raise_server_exceptions=False)
        r = c.post(f"/tasks/{task.task_id}/approve")
        results.append(("approve", r.status_code))

    def do_advance():
        c = TestClient(app, raise_server_exceptions=False)
        r = c.post(f"/tasks/{task.task_id}/advance")
        results.append(("advance", r.status_code))

    t1 = threading.Thread(target=do_approve)
    t2 = threading.Thread(target=do_advance)
    t1.start(); t2.start()
    t1.join(); t2.join()

    assert all(code in (200, 409) for _, code in results), f"Unexpected codes: {results}"
    final = orc.get_task(task.task_id)
    assert final.state is not None


# ── pwa-icon-compatibility: PNG raster icons ──────────────────────────

def test_icon_192_endpoint(client):
    """GET /icon-192.png must return a valid PNG (correct signature and Content-Type)."""
    r = client.get("/icon-192.png")
    assert r.status_code == 200
    assert r.headers["content-type"] == "image/png"
    assert r.content[:8] == b"\x89PNG\r\n\x1a\n", "Response must start with PNG signature"
    assert len(r.content) > 100, "PNG must not be empty"


def test_icon_512_endpoint(client):
    """GET /icon-512.png must return a valid PNG."""
    r = client.get("/icon-512.png")
    assert r.status_code == 200
    assert r.headers["content-type"] == "image/png"
    assert r.content[:8] == b"\x89PNG\r\n\x1a\n", "Response must start with PNG signature"
    assert len(r.content) > 100, "PNG must not be empty"


def test_icon_192_correct_dimensions():
    """ICON_192_PNG must encode a 192×192 image (IHDR width/height bytes)."""
    import struct
    # PNG IHDR is at offset 16 (8 sig + 4 len + 4 "IHDR")
    width, height = struct.unpack(">II", ICON_192_PNG[16:24])
    assert width == 192, f"Expected width=192, got {width}"
    assert height == 192, f"Expected height=192, got {height}"


def test_icon_512_correct_dimensions():
    """ICON_512_PNG must encode a 512×512 image."""
    import struct
    width, height = struct.unpack(">II", ICON_512_PNG[16:24])
    assert width == 512, f"Expected width=512, got {width}"
    assert height == 512, f"Expected height=512, got {height}"


def test_manifest_has_png_192_icon():
    """Manifest must include a 192×192 PNG icon for Android Chrome install criteria."""
    icons = MANIFEST["icons"]
    png_192 = [i for i in icons if i.get("sizes") == "192x192" and i.get("type") == "image/png"]
    assert png_192, "Manifest must include a 192x192 PNG icon entry"
    assert png_192[0]["src"] == "/icon-192.png"


def test_manifest_has_png_512_icon():
    """Manifest must include a 512×512 PNG icon for Android Chrome install criteria."""
    icons = MANIFEST["icons"]
    png_512 = [i for i in icons if i.get("sizes") == "512x512" and i.get("type") == "image/png"]
    assert png_512, "Manifest must include a 512x512 PNG icon entry"
    assert png_512[0]["src"] == "/icon-512.png"


def test_manifest_endpoint_has_png_icons(client):
    """The /manifest.webmanifest endpoint must return the PNG icon entries."""
    r = client.get("/manifest.webmanifest")
    icons = r.json()["icons"]
    srcs = [i["src"] for i in icons]
    assert "/icon-192.png" in srcs, "Manifest endpoint must include /icon-192.png"
    assert "/icon-512.png" in srcs, "Manifest endpoint must include /icon-512.png"


def test_apple_touch_icon_points_to_png():
    """apple-touch-icon must point to a PNG (not SVG); iOS ignores SVG for home screen icons."""
    assert 'href="/icon-192.png"' in INDEX_HTML, (
        "apple-touch-icon must reference /icon-192.png so iOS can render the home screen icon"
    )


def test_sw_shell_includes_png_icons():
    """Service worker SHELL list must cache the PNG icons for offline availability."""
    assert "/icon-192.png" in SERVICE_WORKER_JS, "SHELL must include /icon-192.png"
    assert "/icon-512.png" in SERVICE_WORKER_JS, "SHELL must include /icon-512.png"


def test_png_icons_exempt_from_auth(client):
    """PNG icon endpoints must be accessible without auth (same as SVG and manifest)."""
    from dual_agent.api import create_app as _app
    orc = fake_orchestrator()
    authed_client = TestClient(_app(orc, token="secret"), raise_server_exceptions=False)
    assert authed_client.get("/icon-192.png").status_code == 200
    assert authed_client.get("/icon-512.png").status_code == 200


# ── lan-pwa-insecure-context: HTTPS warning in LAN banner ─────────────

def test_lan_banner_warns_about_pwa_https_requirement():
    """LAN startup banner must warn that PWA/Service Worker requires HTTPS."""
    import inspect
    from dual_agent import cli
    src = inspect.getsource(cli)
    assert "HTTPS" in src or "https" in src.lower(), (
        "cli.py serve command must mention HTTPS in the LAN startup output"
    )
    assert "Service Worker" in src or "PWA" in src or "pwa" in src.lower(), (
        "cli.py serve command must mention Service Worker or PWA in the LAN HTTPS warning"
    )