"""Tests for token auth middleware and X-Request-Id idempotency.

Auth tests:
- No token configured → all requests pass through
- Token configured → 401 without token
- Token via Authorization: Bearer <token> → passes
- Token via cookie da_token → passes
- Token via ?token= → NOT supported (returns login bridge, not a redirect)
- /health never requires auth

Idempotency tests:
- Same X-Request-Id on advance/approve/cancel/resume → same result, no double execution
- Different X-Request-Id → executes again (state permitting)
- No X-Request-Id → normal execution
- Repeated cancel with same id → idempotent (task stays NEEDS_HUMAN)
"""
import pytest
from fastapi.testclient import TestClient

from dual_agent.api import create_app
from tests._fakes import fake_orchestrator, task_in_state


# ── Auth tests ─────────────────────────────────────────────────────────

def test_no_token_configured_passes_all():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc))
    assert client.get("/health").status_code == 200
    assert client.get("/tasks").status_code == 200


def test_token_configured_blocks_without_auth():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    r = client.get("/tasks")
    assert r.status_code == 401
    assert "WWW-Authenticate" in r.headers


def test_token_via_bearer_header():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    r = client.get("/tasks", headers={"Authorization": "Bearer secret123"})
    assert r.status_code == 200


def test_wrong_bearer_token_rejected():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    r = client.get("/tasks", headers={"Authorization": "Bearer wrong"})
    assert r.status_code == 401


def test_token_via_cookie():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    client.cookies.set("da_token", "secret123")
    r = client.get("/tasks")
    assert r.status_code == 200


def test_wrong_cookie_rejected():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    client.cookies.set("da_token", "bad")
    r = client.get("/tasks")
    assert r.status_code == 401


def test_token_via_query_param_no_longer_redirects():
    # ?token= in the URL leaks the token to server access logs.
    # It has been removed; users must use the #token= hash fragment (never
    # sent to server) or the login form (POST /cookie with Authorization header).
    orc = fake_orchestrator()
    client = TestClient(
        create_app(orc, token="secret123"),
        raise_server_exceptions=False,
        follow_redirects=False,
    )
    r = client.get("/?token=secret123")
    # Unauthenticated GET / → login bridge page (not a redirect, not access granted)
    assert r.status_code == 200
    assert r.status_code != 302
    assert "da_token" not in r.cookies


def test_health_bypasses_auth():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    r = client.get("/health")
    assert r.status_code == 200


def test_pwa_assets_bypass_auth():
    """PWA static assets must be accessible without auth so the service worker
    can cache them during install (before the session cookie is set)."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret123"), raise_server_exceptions=False)
    assert client.get("/sw.js").status_code == 200
    assert client.get("/manifest.webmanifest").status_code == 200
    assert client.get("/icon.svg").status_code == 200


def test_token_not_in_401_body():
    """Token must never be echoed in error responses."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="supersecret"), raise_server_exceptions=False)
    r = client.get("/tasks")
    assert "supersecret" not in r.text


def test_claude_authorization_code_can_be_submitted():
    class Auth:
        def __init__(self):
            self.calls = []

        def submit_code(self, provider, code):
            self.calls.append((provider, code))
            return True

    auth = Auth()
    client = TestClient(create_app(fake_orchestrator(), auth=auth))
    response = client.post("/auth/claude/login/code", json={"code": "abc#123"})

    assert response.status_code == 200
    assert response.json() == {"submitted": True}
    assert auth.calls == [("claude", "abc#123")]


# ── Idempotency tests ──────────────────────────────────────────────────

def _headers(req_id: str) -> dict:
    return {"X-Request-Id": req_id}


def test_no_request_id_works_normally():
    """Endpoints work without X-Request-Id (backward compatible)."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    r = client.post(f"/tasks/{task.task_id}/advance")
    assert r.status_code == 200


def test_same_request_id_returns_cached_advance():
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "test-idem-01"
    r1 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers(rid))
    assert r1.status_code == 200
    state_after_first = r1.json()["state"]

    # Second call with same id must not advance again
    r2 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers(rid))
    assert r2.status_code == 200
    assert r2.json()["state"] == state_after_first


def test_different_request_id_executes_again():
    """A fresh X-Request-Id advances the state machine again."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    r1 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers("idem-a"))
    assert r1.status_code == 200
    state1 = r1.json()["state"]

    r2 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers("idem-b"))
    assert r2.status_code == 200
    # State may differ (or not, depending on fake adapter), but request was executed
    assert r2.json()["task_id"] == task.task_id


def test_cancel_is_idempotent_with_same_id():
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "cancel-idem-01"
    r1 = client.post(f"/tasks/{task.task_id}/cancel", headers=_headers(rid))
    assert r1.status_code == 200
    assert r1.json()["state"] == "NEEDS_HUMAN"

    r2 = client.post(f"/tasks/{task.task_id}/cancel", headers=_headers(rid))
    assert r2.status_code == 200
    # Same cached result
    assert r2.json()["state"] == "NEEDS_HUMAN"


def test_approve_idempotency_on_wrong_state():
    """Approving a task that is not in a gate state returns current state, not 409."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "approve-idem-01"
    # First approve: task not in gated state → no-op, returns current state
    r1 = client.post(f"/tasks/{task.task_id}/approve", headers=_headers(rid))
    assert r1.status_code == 200
    # Second approve with same id → cached, no error
    r2 = client.post(f"/tasks/{task.task_id}/approve", headers=_headers(rid))
    assert r2.status_code == 200
    assert r2.json()["state"] == r1.json()["state"]


def test_invalid_transition_new_request_id_is_409():
    """InvalidTransition must never be cached as 200 for any request-id."""
    from unittest.mock import patch
    from dual_agent.state_machine import InvalidTransition as IT

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc), raise_server_exceptions=False)
    # Patch approve to simulate an InvalidTransition (e.g. from a race condition)
    with patch.object(orc, "approve", side_effect=IT("state conflict")):
        r1 = client.post(f"/tasks/{task.task_id}/approve", headers=_headers("new-id-1"))
        r2 = client.post(f"/tasks/{task.task_id}/approve", headers=_headers("new-id-2"))
    assert r1.status_code == 409
    assert r2.status_code == 409


def test_invalid_task_id_returns_404():
    orc = fake_orchestrator()
    client = TestClient(create_app(orc), raise_server_exceptions=False)
    for endpoint in ["advance", "approve", "cancel", "resume", "run"]:
        r = client.post(f"/tasks/does_not_exist_abc/{endpoint}")
        assert r.status_code == 404, f"{endpoint} should 404"


def test_tampered_task_id_get_endpoints_return_404():
    """GET sub-endpoints must return 404 (not 500) for invalid/tampered task IDs."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc), raise_server_exceptions=False)
    bad_ids = ["../etc/passwd", "not valid!", "a" * 100]
    get_endpoints = ["conversation", "detail", "logs"]
    for bad_id in bad_ids:
        for endpoint in get_endpoints:
            r = client.get(f"/tasks/{bad_id}/{endpoint}")
            assert r.status_code in (404, 422), (
                f"GET /tasks/{bad_id!r}/{endpoint} returned {r.status_code}, expected 404 or 422"
            )


def test_task_not_found_not_cached():
    """A 404 must not be cached as a successful idempotency result."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "idem-404"
    # Try with wrong id first → 404
    r1 = client.post("/tasks/nonexistent/advance", headers=_headers(rid))
    assert r1.status_code == 404
    # Same rid on valid task must still execute
    r2 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers(rid))
    assert r2.status_code == 200


# ── /tasks/{task_id}/say idempotency ──────────────────────────────────

_SAY_RESULT = {"task_id": "t1", "state": "INIT", "goal": "test", "cycle": 0}


def test_say_idempotency_same_request_id():
    """POST /say with the same X-Request-Id and text must not call orchestrator.say twice."""
    from unittest.mock import patch

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "say-idem-01"
    payload = {"text": "hello"}

    with patch.object(orc, "say", return_value=_SAY_RESULT) as mock_say:
        r1 = client.post(f"/tasks/{task.task_id}/say", json=payload, headers=_headers(rid))
        r2 = client.post(f"/tasks/{task.task_id}/say", json=payload, headers=_headers(rid))

    assert r1.status_code == 200
    assert r2.status_code == 200
    assert mock_say.call_count == 1, f"orchestrator.say was called {mock_say.call_count} times, expected 1"


def test_say_different_request_id_executes_again():
    """POST /say with a different X-Request-Id should call orchestrator.say again."""
    from unittest.mock import patch

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    payload = {"text": "hello again"}

    with patch.object(orc, "say", return_value=_SAY_RESULT) as mock_say:
        r1 = client.post(f"/tasks/{task.task_id}/say", json=payload, headers=_headers("say-a"))
        r2 = client.post(f"/tasks/{task.task_id}/say", json=payload, headers=_headers("say-b"))

    assert r1.status_code == 200
    assert r2.status_code == 200
    assert mock_say.call_count == 2


def test_say_no_request_id_always_executes():
    """POST /say without X-Request-Id bypasses idempotency (backward compat)."""
    from unittest.mock import patch

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    payload = {"text": "no-id message"}

    with patch.object(orc, "say", return_value=_SAY_RESULT) as mock_say:
        client.post(f"/tasks/{task.task_id}/say", json=payload)
        client.post(f"/tasks/{task.task_id}/say", json=payload)

    assert mock_say.call_count == 2


# ── /cookie token-exchange endpoint ───────────────────────────────────

def test_cookie_endpoint_sets_cookie_with_correct_token():
    """POST /cookie with correct bearer token must set da_token cookie and return 204."""
    orc = fake_orchestrator()
    client = TestClient(
        create_app(orc, token="mytoken"),
        follow_redirects=False,
        raise_server_exceptions=False,
    )
    r = client.post("/cookie", headers={"Authorization": "Bearer mytoken"})
    assert r.status_code == 204
    assert "da_token" in r.cookies or "Set-Cookie" in r.headers


def test_cookie_endpoint_rejects_wrong_token():
    """POST /cookie with wrong token must return 401."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="mytoken"), raise_server_exceptions=False)
    r = client.post("/cookie", headers={"Authorization": "Bearer wrongtoken"})
    assert r.status_code == 401


def test_cookie_endpoint_accessible_without_auth():
    """POST /cookie must not be blocked by the auth middleware (bootstrap endpoint)."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="mytoken"), raise_server_exceptions=False)
    # POST /cookie without any auth credential should reach the handler (returns 401 from handler)
    r = client.post("/cookie")
    assert r.status_code != 404  # endpoint exists
    assert r.status_code in (204, 401)  # not blocked by middleware (which would give WWW-Authenticate)


def test_unauthenticated_get_root_returns_login_page_not_401():
    """GET / without auth must return the login bridge page (200), not 401."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="secret"), raise_server_exceptions=False)
    r = client.get("/")
    assert r.status_code == 200
    assert "text/html" in r.headers["content-type"]
    assert "令牌" in r.text  # login page mentions token


def test_login_page_does_not_contain_token():
    """The login bridge page must never echo the configured token."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="topsecret99"), raise_server_exceptions=False)
    r = client.get("/")
    assert "topsecret99" not in r.text


# ── In-process idempotency limitation (mobile-review-002) ─────────────

def test_network_timeout_retry_same_reqid_returns_cached():
    """After a network timeout the client retries with the same X-Request-Id.
    The server must return the cached result without re-executing the action.
    This is the primary in-process idempotency guarantee."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    client = TestClient(create_app(orc))
    rid = "timeout-retry-002"

    r1 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers(rid))
    assert r1.status_code == 200
    state_cached = r1.json()["state"]

    r2 = client.post(f"/tasks/{task.task_id}/advance", headers=_headers(rid))
    assert r2.status_code == 200
    # Must return cached state, not re-execute the advance
    assert r2.json()["state"] == state_cached


def test_idempotency_cache_cleared_after_restart():
    """After a server restart (new app instance) the in-memory cache is empty.
    A retry with the same X-Request-Id re-executes the action.  The task state
    machine (InvalidTransition / TaskBusy) is the final safety net.
    This test documents the known single-process limitation of _IdempotencyStore.
    """
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")

    # First app instance: advance INIT → next state
    c1 = TestClient(create_app(orc))
    r1 = c1.post(f"/tasks/{task.task_id}/advance", headers=_headers("restart-idem-001"))
    assert r1.status_code == 200
    assert r1.json()["state"] != "INIT"

    # Simulate restart: new app instance has an empty idempotency cache.
    # The orchestrator/store is reused (task state is persisted on disk).
    c2 = TestClient(create_app(orc), raise_server_exceptions=False)
    r2 = c2.post(f"/tasks/{task.task_id}/advance", headers=_headers("restart-idem-001"))
    # Cache is gone → action re-executes against the current persisted task state.
    # Valid outcomes: another 200 (advances again) or 409 (state machine blocks invalid transition).
    assert r2.status_code in (200, 409)
    # Task must remain in a valid, consistent state — never corrupted.
    final = orc.get_task(task.task_id)
    assert final.state is not None
    assert final.state.value != "INIT"  # first advance always took effect


def test_multi_worker_independent_caches():
    """Two separate app instances (simulating multiple uvicorn workers) each have
    independent idempotency caches. A request handled by worker-A is unknown to
    worker-B; re-executing against the persisted task state must be safe.
    Documents the known single-process limitation of _IdempotencyStore.
    """
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")

    # Worker A handles the first request
    ca = TestClient(create_app(orc))
    ra = ca.post(f"/tasks/{task.task_id}/advance", headers=_headers("worker-idem-001"))
    assert ra.status_code == 200
    state_a = ra.json()["state"]
    assert state_a != "INIT"

    # Worker B has no knowledge of worker A's cache — re-executes the action.
    cb = TestClient(create_app(orc), raise_server_exceptions=False)
    rb = cb.post(f"/tasks/{task.task_id}/advance", headers=_headers("worker-idem-001"))
    # State machine (InvalidTransition/TaskBusy) is the safety net for cross-worker retries.
    assert rb.status_code in (200, 409), f"Expected 200 or 409, got {rb.status_code}"
    final = orc.get_task(task.task_id)
    assert final.state is not None  # task always remains in a valid state


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

def test_cookie_secure_when_behind_https_proxy():
    """POST /cookie with X-Forwarded-Proto: https must include the Secure flag."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="lan-tok"), raise_server_exceptions=False)
    r = client.post(
        "/cookie",
        headers={"Authorization": "Bearer lan-tok", "X-Forwarded-Proto": "https"},
    )
    assert r.status_code == 204
    assert "Secure" in r.headers.get("Set-Cookie", ""), (
        "Secure cookie flag must be set when X-Forwarded-Proto: https is present"
    )


def test_cookie_not_secure_over_plain_http():
    """POST /cookie without X-Forwarded-Proto must NOT include Secure (HTTP must work)."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="lan-tok"), raise_server_exceptions=False)
    r = client.post("/cookie", headers={"Authorization": "Bearer lan-tok"})
    assert r.status_code == 204
    assert "Secure" not in r.headers.get("Set-Cookie", ""), (
        "Secure flag must be absent over plain HTTP so the cookie is actually stored"
    )


def test_cookie_always_httponly_and_samesite_strict():
    """HttpOnly and SameSite=strict must be present regardless of transport."""
    orc = fake_orchestrator()
    client = TestClient(create_app(orc, token="lan-tok"), raise_server_exceptions=False)
    r = client.post("/cookie", headers={"Authorization": "Bearer lan-tok"})
    assert r.status_code == 204
    hdr = r.headers.get("Set-Cookie", "").lower()
    assert "httponly" in hdr, "Cookie must be HttpOnly"
    assert "samesite=strict" in hdr, "Cookie must be SameSite=strict"


# ── Disk-backed idempotency (mobile-idempotency-not-durable) ──────────

def test_task_action_idem_record_written_to_disk():
    """After a task action, an idempotency record must be persisted to {task_dir}/idem/."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    c = TestClient(create_app(orc), raise_server_exceptions=False)
    rid = "disk-idem-write-001"
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 200
    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    assert idem_dir.is_dir(), "idem/ directory must be created after a successful action"
    files = list(idem_dir.glob("advance_*.json"))
    assert files, "at least one advance idem record must exist on disk"


def test_disk_idem_survives_in_process_restart():
    """A new app instance (empty in-memory cache) must find the disk record and not re-execute."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "disk-restart-001"

    c1 = TestClient(create_app(orc), raise_server_exceptions=False)
    r1 = c1.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r1.status_code == 200
    state_after = r1.json()["state"]

    # New app instance = empty in-memory store; disk still holds the record.
    c2 = TestClient(create_app(orc), raise_server_exceptions=False)
    r2 = c2.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r2.status_code == 200, "Must return cached result even after in-memory cache is gone"
    assert r2.json()["state"] == state_after, "Disk hit must return the original state, not advance again"


def test_disk_idem_different_reqid_executes_again():
    """A different X-Request-Id must bypass the disk cache and execute again."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")

    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r1 = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": "rid-aaa"})
    assert r1.status_code == 200

    r2 = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": "rid-bbb"})
    # New reqId → executes again; state machine may advance or reject, both are valid.
    assert r2.status_code in (200, 409)


# ── Cross-worker pending marker (cross-worker-idempotency-race) ───────

def test_pending_marker_written_before_execution():
    """A .pending file must be created atomically before the action executes,
    providing cross-process protection when the completed .json is not yet present."""
    import threading

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "pending-marker-001"

    # Hold the action long enough to check the .pending file mid-flight
    pending_existed = []
    orig_advance = orc.advance

    def slow_advance(task_id):
        idem_dir = orc.store.task_dir(task_id) / "idem"
        from dual_agent.api import _safe_idem_key
        key = _safe_idem_key(rid)
        pending = idem_dir / f"advance_{key}.pending"
        pending_existed.append(pending.is_file())
        return orig_advance(task_id)

    orc.advance = slow_advance

    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 200
    assert pending_existed == [True], ".pending must exist while action is executing"


def test_pending_marker_removed_after_success():
    """.pending marker must be cleaned up once the .json record is written."""
    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "pending-cleanup-001"

    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 200

    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    from dual_agent.api import _safe_idem_key
    key = _safe_idem_key(rid)
    assert (idem_dir / f"advance_{key}.json").is_file(), ".json record must exist"
    assert not (idem_dir / f"advance_{key}.pending").is_file(), ".pending must be removed after success"


def test_pending_marker_blocks_concurrent_cross_process():
    """A second request finding a live .pending file must get 409, not execute."""
    from dual_agent.api import _load_task_idem, _PENDING, _safe_idem_key

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "cross-proc-pending-001"

    # Manually write a .pending file to simulate another process mid-execution
    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    idem_dir.mkdir(parents=True, exist_ok=True)
    key = _safe_idem_key(rid)
    (idem_dir / f"advance_{key}.pending").write_bytes(b"")

    # _load_task_idem must return _PENDING for a fresh .pending file
    result = _load_task_idem(orc.store, task.task_id, rid, "advance")
    assert result is _PENDING, "_load_task_idem must return _PENDING when .pending exists"

    # The HTTP layer must surface this as a 409
    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 409


def test_stale_pending_marker_treated_as_cache_miss():
    """A .pending file older than _PENDING_STALE_SECS must be ignored (process died)."""
    import os as _os
    import time as _time
    from dual_agent.api import _load_task_idem, _PENDING, _safe_idem_key, _PENDING_STALE_SECS

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "stale-pending-001"

    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    idem_dir.mkdir(parents=True, exist_ok=True)
    key = _safe_idem_key(rid)
    pending_path = idem_dir / f"advance_{key}.pending"
    pending_path.write_bytes(b"")
    # Back-date the file by 2x the stale threshold
    old_mtime = _time.time() - (_PENDING_STALE_SECS * 2)
    _os.utime(str(pending_path), (old_mtime, old_mtime))

    result = _load_task_idem(orc.store, task.task_id, rid, "advance")
    assert result is None, "Stale .pending must be ignored (treated as cache miss)"


def test_stale_pending_marker_deleted_by_load():
    """_load_task_idem must delete a stale .pending file so subsequent claim can succeed."""
    import os as _os
    import time as _time
    from dual_agent.api import _load_task_idem, _safe_idem_key, _PENDING_STALE_SECS

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "stale-delete-001"

    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    idem_dir.mkdir(parents=True, exist_ok=True)
    key = _safe_idem_key(rid)
    pending_path = idem_dir / f"advance_{key}.pending"
    pending_path.write_bytes(b"")
    old_mtime = _time.time() - (_PENDING_STALE_SECS * 2)
    _os.utime(str(pending_path), (old_mtime, old_mtime))

    _load_task_idem(orc.store, task.task_id, rid, "advance")
    assert not pending_path.is_file(), (
        "_load_task_idem must delete stale .pending so _claim_task_idem_pending can create a fresh one"
    )


def test_stale_pending_lock_does_not_permanently_block_request():
    """After a process crash leaves a stale .pending file, the next HTTP request with the same
    X-Request-Id must succeed (not return a permanent 409) — the idem-stale-pending-lock fix."""
    import os as _os
    import time as _time
    from dual_agent.api import _safe_idem_key, _PENDING_STALE_SECS

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "stale-recovery-001"

    # Simulate a crashed process that left a stale .pending file.
    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    idem_dir.mkdir(parents=True, exist_ok=True)
    key = _safe_idem_key(rid)
    pending_path = idem_dir / f"advance_{key}.pending"
    pending_path.write_bytes(b"")
    old_mtime = _time.time() - (_PENDING_STALE_SECS * 2)
    _os.utime(str(pending_path), (old_mtime, old_mtime))

    # The request must succeed (not return 409) despite the stale file.
    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 200, (
        f"Stale .pending must not permanently block the request; got {r.status_code}. "
        "Expected 200 — the stale file should be cleaned up and the action re-executed."
    )
    assert r.json()["state"] != "INIT", "Task must have advanced after stale pending was cleared"


def test_pending_marker_removed_on_exception():
    """.pending marker must be cleaned up when the action raises an exception."""
    from unittest.mock import patch
    from dual_agent.state_machine import InvalidTransition
    from dual_agent.api import _safe_idem_key

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "pending-exception-001"

    with patch.object(orc, "advance", side_effect=InvalidTransition("bad state")):
        c = TestClient(create_app(orc), raise_server_exceptions=False)
        r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 409

    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    key = _safe_idem_key(rid)
    assert not (idem_dir / f"advance_{key}.pending").is_file(), ".pending must be removed on exception"


# ── Regression: resume-final-verify-can-mark-done ────────────────────

def test_recover_from_needs_human_after_final_verify_stays_at_final_verify():
    """recover() on a NEEDS_HUMAN task with last_successful_stage=FINAL_VERIFY must
    resume at FINAL_VERIFY, not skip to DONE.

    Root cause: last_successful_stage is written BEFORE the verdict is applied, so a
    failed FINAL_VERIFY (CHANGES_REQUIRED → exhausted fix cycles) ends up with
    last_successful_stage=FINAL_VERIFY while state=NEEDS_HUMAN.  Mapping this to DONE
    would mark incomplete work as complete.
    """
    from dual_agent.domain import TaskState, ErrorCategory

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

    # Directly write the state that a real failure scenario produces:
    # task is NEEDS_HUMAN because FINAL_VERIFY requested changes and fix cycles exhausted.
    task = orc.store.load(task.task_id)
    updated = task.model_copy(update={
        "state": TaskState.NEEDS_HUMAN,
        "last_successful_stage": TaskState.FINAL_VERIFY.value,
        "error": "FINAL_VERIFY requested changes; max fix cycles reached",
    })
    orc.store.save(updated)

    resumed = orc.recover(task.task_id)
    assert resumed.state == TaskState.FINAL_VERIFY, (
        f"recover() mapped last_successful_stage=FINAL_VERIFY to {resumed.state!r}; "
        "expected FINAL_VERIFY so the verification re-runs rather than skipping to DONE"
    )
    assert resumed.state != TaskState.DONE, "recover() must not mark task as DONE"


# ── Regression: lan-without-auth-exposes-control-api ─────────────────

def test_serve_lan_without_token_exits_with_error():
    """--lan without --token must exit(1) with a clear error message, not start the server."""
    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(app, ["serve", "--lan"])
    assert result.exit_code == 1, (
        f"Expected exit code 1 for --lan without --token, got {result.exit_code}. "
        f"Output: {result.output!r}"
    )
    assert "token" in result.output.lower(), (
        "Error message must mention token so the operator knows how to fix it"
    )


def test_serve_lan_with_token_does_not_exit_early(monkeypatch, tmp_path):
    """--lan with --token must not exit early with code 1 (LAN auth guard does not fire)."""
    import uvicorn as _uvicorn

    # Prevent uvicorn from actually starting a server — we only need the guard logic.
    monkeypatch.setattr(_uvicorn, "run", lambda *a, **kw: None)

    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["serve", "--lan", "--token", "test-tok", "--root", str(tmp_path), "--fake"],
    )
    assert result.exit_code == 0, (
        f"--lan with --token should not exit with error, got {result.exit_code}. "
        f"Output: {result.output!r}"
    )


# ── Regression: TLS serve options (lan-bearer-token-plaintext) ───────

def test_serve_tls_cert_without_key_exits_error(tmp_path):
    """--tls-cert without --tls-key must exit with error."""
    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["serve", "--lan", "--token", "tok", "--tls-cert", str(tmp_path / "cert.pem"), "--fake",
         "--root", str(tmp_path)],
    )
    assert result.exit_code == 1
    assert "tls-key" in result.output.lower() or "key" in result.output.lower()


def test_serve_tls_key_without_cert_exits_error(tmp_path):
    """--tls-key without --tls-cert must exit with error."""
    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["serve", "--lan", "--token", "tok", "--tls-key", str(tmp_path / "key.pem"), "--fake",
         "--root", str(tmp_path)],
    )
    assert result.exit_code == 1
    assert "tls-cert" in result.output.lower() or "cert" in result.output.lower()


def test_serve_tls_banner_shows_https(monkeypatch, tmp_path):
    """When --tls-cert + --tls-key are provided, the startup banner must show https://."""
    import uvicorn as _uvicorn

    captured_kwargs = {}

    def fake_run(app, **kwargs):
        captured_kwargs.update(kwargs)

    monkeypatch.setattr(_uvicorn, "run", fake_run)
    # Create dummy cert/key files so Path resolution succeeds
    cert_path = tmp_path / "cert.pem"
    key_path = tmp_path / "key.pem"
    cert_path.write_text("CERT")
    key_path.write_text("KEY")

    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(
        app,
        [
            "serve", "--lan", "--token", "tok",
            "--tls-cert", str(cert_path),
            "--tls-key", str(key_path),
            "--root", str(tmp_path), "--fake",
        ],
    )
    assert result.exit_code == 0, f"Exit code: {result.exit_code}\nOutput: {result.output}"
    assert "https://" in result.output, (
        "Startup banner must show https:// when TLS cert+key are configured; "
        f"got: {result.output!r}"
    )
    assert "http://" not in result.output.replace("https://", ""), (
        "Startup banner must not show plain http:// when TLS is active"
    )


def test_serve_tls_passes_ssl_kwargs_to_uvicorn(monkeypatch, tmp_path):
    """uvicorn.run must receive ssl_certfile and ssl_keyfile when --tls-cert/--tls-key set."""
    import uvicorn as _uvicorn

    captured_kwargs = {}

    def fake_run(app, **kwargs):
        captured_kwargs.update(kwargs)

    monkeypatch.setattr(_uvicorn, "run", fake_run)
    cert_path = tmp_path / "cert.pem"
    key_path = tmp_path / "key.pem"
    cert_path.write_text("CERT")
    key_path.write_text("KEY")

    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    runner.invoke(
        app,
        [
            "serve", "--lan", "--token", "tok",
            "--tls-cert", str(cert_path),
            "--tls-key", str(key_path),
            "--root", str(tmp_path), "--fake",
        ],
    )
    assert "ssl_certfile" in captured_kwargs, "uvicorn.run must receive ssl_certfile"
    assert "ssl_keyfile" in captured_kwargs, "uvicorn.run must receive ssl_keyfile"
    assert captured_kwargs["ssl_certfile"] == str(cert_path)
    assert captured_kwargs["ssl_keyfile"] == str(key_path)


def test_serve_no_tls_banner_warns_plaintext(monkeypatch, tmp_path):
    """Without TLS, the LAN banner must warn about plaintext transport."""
    import uvicorn as _uvicorn

    monkeypatch.setattr(_uvicorn, "run", lambda *a, **kw: None)

    from typer.testing import CliRunner
    from dual_agent.cli import app

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["serve", "--lan", "--token", "tok", "--root", str(tmp_path), "--fake"],
    )
    assert result.exit_code == 0
    output = result.output.upper()
    assert "UNENCRYPTED" in output or "PLAINTEXT" in output or "TRANSPORT-UNPROTECTED" in output, (
        "LAN mode without TLS must warn about plaintext transport; "
        f"got: {result.output!r}"
    )


# ── Regression: PID-based pending staleness (long-running-idempotency-race) ──

def test_claim_idem_pending_writes_pid(tmp_path):
    """_claim_task_idem_pending must write the current process PID into the .pending file."""
    import os
    from unittest.mock import MagicMock
    from dual_agent.api import _claim_task_idem_pending, _safe_idem_key

    store = MagicMock()
    store.task_dir.return_value = tmp_path / "task-1"

    rid = "pid-write-test"
    key = _safe_idem_key(rid)
    result = _claim_task_idem_pending(store, "task-1", rid, "advance")
    assert result is True

    pending_path = tmp_path / "task-1" / "idem" / f"advance_{key}.pending"
    assert pending_path.is_file(), ".pending file must be created"
    content = pending_path.read_text(encoding="utf-8").strip()
    assert content == str(os.getpid()), (
        f"Pending file must contain the current PID ({os.getpid()}), got {content!r}"
    )


def test_is_pending_stale_live_pid_returns_false(tmp_path):
    """_is_pending_stale must return False when .pending contains the live process PID."""
    import os
    from pathlib import Path
    from dual_agent.api import _is_pending_stale

    pending_path = tmp_path / "test.pending"
    pending_path.write_text(str(os.getpid()), encoding="utf-8")
    assert _is_pending_stale(pending_path) is False, (
        "Pending file with live PID must NOT be considered stale"
    )


def test_is_pending_stale_dead_pid_returns_true(tmp_path):
    """_is_pending_stale must return True when .pending contains a PID that no longer exists."""
    from dual_agent.api import _is_pending_stale
    import subprocess

    # Spawn a process and wait for it to exit so we have a known-dead PID
    proc = subprocess.Popen(["true"])
    proc.wait()
    dead_pid = proc.pid

    pending_path = tmp_path / "test.pending"
    pending_path.write_text(str(dead_pid), encoding="utf-8")
    assert _is_pending_stale(pending_path) is True, (
        f"Pending file with dead PID {dead_pid} must be considered stale"
    )


def test_load_task_idem_live_pid_pending_returns_pending_sentinel(tmp_path):
    """_load_task_idem must return _PENDING when .pending contains a live process PID."""
    import os
    from unittest.mock import MagicMock
    from dual_agent.api import _load_task_idem, _PENDING, _safe_idem_key

    store = MagicMock()
    task_dir = tmp_path / "task-live"
    task_dir.mkdir()
    store.task_dir.return_value = task_dir

    rid = "live-pid-test"
    key = _safe_idem_key(rid)
    idem_dir = task_dir / "idem"
    idem_dir.mkdir()
    pending_path = idem_dir / f"advance_{key}.pending"
    pending_path.write_text(str(os.getpid()), encoding="utf-8")

    result = _load_task_idem(store, "task-live", rid, "advance")
    assert result is _PENDING, (
        "A .pending file with a live PID must cause _load_task_idem to return _PENDING "
        "(blocking the concurrent request), not None (which would allow re-execution)"
    )


def test_http_409_when_live_pid_holds_pending(tmp_path):
    """POST /advance with live-PID .pending must return 409 (concurrent request in flight)."""
    import os
    from dual_agent.api import _safe_idem_key
    from tests._fakes import fake_orchestrator, task_in_state

    orc = fake_orchestrator()
    task = task_in_state(orc, "INIT")
    rid = "live-pid-http-test"

    # Manually place a .pending file with the current (live) PID
    idem_dir = orc.store.task_dir(task.task_id) / "idem"
    idem_dir.mkdir(parents=True, exist_ok=True)
    key = _safe_idem_key(rid)
    (idem_dir / f"advance_{key}.pending").write_text(str(os.getpid()), encoding="utf-8")

    c = TestClient(create_app(orc), raise_server_exceptions=False)
    r = c.post(f"/tasks/{task.task_id}/advance", headers={"X-Request-Id": rid})
    assert r.status_code == 409, (
        f"Live-PID pending marker must block the request with 409; got {r.status_code}"
    )