from collections import OrderedDict
import json
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import urlparse

import pytest

from api import config as api_config
from api import models, routes, workspace


def _write_session_sidecar(monkeypatch, tmp_path, session):
    session_dir = tmp_path / "sessions"
    session_dir.mkdir(exist_ok=True)
    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
    monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
    sidecar = session_dir / f"{session.session_id}.json"
    sidecar.write_text(
        json.dumps(
            {
                "session_id": session.session_id,
                "workspace": session.workspace,
                "messages": [{"role": "user", "content": "preserve me"}],
            }
        ),
        encoding="utf-8",
    )
    return sidecar


def test_profile_default_workspace_uses_live_config_default(monkeypatch, tmp_path):
    live_default = tmp_path / "live-default"
    live_default.mkdir()

    monkeypatch.setattr(api_config, "DEFAULT_WORKSPACE", live_default)
    monkeypatch.setattr(api_config, "get_config", lambda: {})

    assert workspace._profile_default_workspace() == str(live_default.resolve())


def test_implicit_workspace_recovery_keeps_fallback_lazy(monkeypatch, tmp_path):
    valid = tmp_path / "valid"
    valid.mkdir()
    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)

    resolved, recovered = workspace.resolve_implicit_workspace_with_recovery(
        valid,
        lambda: (_ for _ in ()).throw(AssertionError("fallback should stay lazy")),
    )

    assert resolved == valid.resolve()
    assert recovered is False


def test_resolve_chat_workspace_with_recovery_repairs_missing_implicit_workspace(monkeypatch, tmp_path):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"

    session = SimpleNamespace(session_id="sess-1", workspace=str(stale))
    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))

    resolved = routes._resolve_chat_workspace_with_recovery(session, None)

    assert resolved == str(fallback.resolve())
    assert session.workspace == str(fallback.resolve())
    assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")


def test_chat_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"
    session = SimpleNamespace(
        session_id="sess-chat-save-fails",
        workspace=str(stale),
        save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
    )

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))

    with pytest.raises(routes.WorkspaceBindingPersistenceError):
        routes._resolve_chat_workspace_with_recovery(session, None)

    assert session.workspace == str(stale)


def test_resolve_chat_workspace_with_recovery_preserves_explicit_errors(monkeypatch, tmp_path):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"

    def fake_resolve(value):
        if value == str(stale):
            raise ValueError(f"Path does not exist: {stale}")
        return Path(value).resolve()

    saved = {"count": 0}

    def fake_save():
        saved["count"] += 1

    session = SimpleNamespace(session_id="sess-2", workspace=str(fallback), save=fake_save)

    monkeypatch.setattr(routes, "resolve_trusted_workspace", fake_resolve)
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))

    with pytest.raises(ValueError, match="Path does not exist"):
        routes._resolve_chat_workspace_with_recovery(session, str(stale))

    assert session.workspace == str(fallback)
    assert saved["count"] == 0


def _post_new_session_with_workspace(
    monkeypatch,
    body,
    previous_session,
    fallback,
    *,
    previous_visible=True,
):
    import api.session_lifecycle as session_lifecycle

    captured = {"new": [], "fallback": 0}

    class _NewSession:
        session_id = "new-session"
        profile = "default"
        messages = []

        def __init__(self, workspace):
            self.workspace = workspace

        def compact(self):
            return {
                "session_id": self.session_id,
                "profile": self.profile,
                "workspace": self.workspace,
            }

    def get_fallback():
        captured["fallback"] += 1
        return str(fallback)

    def create_session(**kwargs):
        captured["new"].append(kwargs)
        return _NewSession(kwargs["workspace"])

    monkeypatch.setattr(routes, "read_body", lambda _handler: body)
    monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
    monkeypatch.setattr(routes, "_csrf_exempt_path", lambda _path: False)
    monkeypatch.setattr(routes, "_handle_extension_sidecar_proxy", lambda *_a, **_k: False)
    monkeypatch.setattr(routes, "_guard_request_session_visibility", lambda *_a, **_k: True)
    monkeypatch.setattr(
        routes,
        "_session_id_visible_to_request_profile",
        lambda *_a, **_k: previous_visible,
    )
    monkeypatch.setattr(
        routes,
        "get_session",
        lambda sid, metadata_only=False: previous_session
        if sid == body.get("prev_session_id")
        else (_ for _ in ()).throw(KeyError(sid)),
    )
    monkeypatch.setattr(routes, "get_last_workspace", get_fallback)
    monkeypatch.setattr(routes, "_session_model_state_from_request", lambda *_a: (None, None))
    monkeypatch.setattr(routes, "new_session", create_session)
    monkeypatch.setattr(
        session_lifecycle,
        "_register_background_commit_thread",
        lambda _thread: False,
    )
    monkeypatch.setattr(
        routes,
        "j",
        lambda _handler, payload, status=200, **_kwargs: captured.setdefault(
            "response", (status, payload)
        ),
    )
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400, **_kwargs: captured.setdefault(
            "error", (status, message)
        ),
    )

    handler = SimpleNamespace(command="POST", headers={})
    routes.handle_post(handler, urlparse("/api/session/new"))
    return captured


def test_session_new_recovers_deleted_workspace_inherited_from_visible_previous_session(
    monkeypatch, tmp_path
):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"
    previous = SimpleNamespace(
        session_id="previous-session",
        profile="default",
        workspace=str(stale),
    )
    body = {
        "workspace": str(stale),
        "workspace_inherited_from_prev_session": True,
        "prev_session_id": previous.session_id,
        "profile": "default",
        "worktree": False,
    }

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    captured = _post_new_session_with_workspace(
        monkeypatch,
        body,
        previous,
        fallback,
    )

    assert "error" not in captured
    assert captured["response"][0] == 200
    assert captured["response"][1]["session"]["workspace"] == str(fallback.resolve())
    assert captured["new"][0]["workspace"] == str(fallback.resolve())
    assert captured["fallback"] == 1


@pytest.mark.parametrize(
    ("mark_inherited", "previous_visible", "stored_workspace"),
    [
        pytest.param(False, True, "requested", id="explicit-request"),
        pytest.param(True, False, "requested", id="foreign-previous-session"),
        pytest.param(True, True, "different", id="stored-workspace-mismatch"),
    ],
)
def test_session_new_keeps_deleted_unverified_workspace_strict(
    monkeypatch,
    tmp_path,
    mark_inherited,
    previous_visible,
    stored_workspace,
):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"
    previous = SimpleNamespace(
        session_id="previous-session",
        profile="default",
        workspace=(str(stale) if stored_workspace == "requested" else str(tmp_path / "other")),
    )
    body = {
        "workspace": str(stale),
        "prev_session_id": previous.session_id,
        "profile": "default",
        "worktree": False,
    }
    if mark_inherited:
        body["workspace_inherited_from_prev_session"] = True

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    captured = _post_new_session_with_workspace(
        monkeypatch,
        body,
        previous,
        fallback,
        previous_visible=previous_visible,
    )

    assert captured["error"][0] == 400
    assert "does not exist" in captured["error"][1].lower()
    assert captured["new"] == []
    assert captured["fallback"] == 0


def test_chat_recovery_preserves_existing_implicit_trust_error(monkeypatch, tmp_path):
    home = tmp_path / "home"
    fallback = home / "fallback"
    fallback.mkdir(parents=True)
    outside = tmp_path / "outside"
    outside.mkdir()
    saved = {"count": 0}
    session = SimpleNamespace(
        session_id="sess-untrusted",
        workspace=str(outside),
        save=lambda: saved.__setitem__("count", saved["count"] + 1),
    )

    monkeypatch.setattr(workspace, "_home_path", lambda: home)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(workspace, "_BOOT_DEFAULT_WORKSPACE", fallback)
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))

    with pytest.raises(ValueError, match="outside the user home directory"):
        routes._resolve_chat_workspace_with_recovery(session, None)

    assert session.workspace == str(outside)
    assert saved["count"] == 0


@pytest.mark.parametrize(
    "terminal_cfg",
    [
        pytest.param(
            {"backend": "ssh", "cwd": "/Users/joeyshiue"},
            id="cwd-absolute",
        ),
        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
    ],
)
def test_chat_recovery_preserves_remote_workspace_rejection(
    monkeypatch, tmp_path, terminal_cfg
):
    candidate = "/Users/other/projects/demo"
    fallback_path = tmp_path / "fallback"
    fallback_path.mkdir()
    fallback_calls = {"count": 0}
    saved = {"count": 0}
    session = SimpleNamespace(
        session_id="sess-remote-untrusted",
        workspace=candidate,
        save=lambda: saved.__setitem__("count", saved["count"] + 1),
    )

    monkeypatch.setattr(
        api_config,
        "get_config",
        lambda: {"terminal": terminal_cfg},
    )
    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)

    def fallback():
        fallback_calls["count"] += 1
        return fallback_path

    monkeypatch.setattr(routes, "get_last_workspace", fallback)

    with pytest.raises(ValueError, match="Path does not exist"):
        routes._resolve_chat_workspace_with_recovery(session, None)

    assert fallback_calls["count"] == 0
    assert session.workspace == candidate
    assert saved["count"] == 0


def test_list_dir_recovers_missing_implicit_session_workspace(monkeypatch, tmp_path):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"
    session = SimpleNamespace(
        session_id="sess-list",
        workspace=str(stale),
    )
    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
    captured = {}

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_session", lambda _sid: session)
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))

    def fake_list_dir(workspace_path, rel_path):
        captured["workspace"] = workspace_path
        captured["rel_path"] = rel_path
        return []

    monkeypatch.setattr(routes, "list_dir", fake_list_dir)
    monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)

    payload = routes._handle_list_dir(
        object(), urlparse("/api/list?session_id=sess-list&path=.")
    )

    assert captured == {"workspace": fallback.resolve(), "rel_path": "."}
    assert session.workspace == str(fallback.resolve())
    assert str(fallback.resolve()) in sidecar.read_text(encoding="utf-8")
    assert payload == {
        "entries": [],
        "signature": "sig",
        "path": ".",
        "workspace": str(fallback.resolve()),
        "workspace_recovered": True,
    }


def test_list_recovery_stays_bound_when_global_fallback_changes(
    monkeypatch, tmp_path
):
    """A later mutation must use the root that the Files pane displayed."""
    fallback_a = tmp_path / "fallback-a"
    fallback_b = tmp_path / "fallback-b"
    fallback_a.mkdir()
    fallback_b.mkdir()
    stale = tmp_path / "deleted-workspace"
    selected = {"workspace": str(fallback_a)}
    session = SimpleNamespace(
        session_id="sess-authority",
        workspace=str(stale),
        profile=None,
    )
    sidecar = _write_session_sidecar(monkeypatch, tmp_path, session)
    captured = {}

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
    monkeypatch.setattr(
        routes, "get_last_workspace", lambda: selected["workspace"]
    )
    def capture_list(workspace_path, _rel):
        captured["listed"] = workspace_path
        return []

    monkeypatch.setattr(routes, "list_dir", capture_list)
    monkeypatch.setattr(routes, "dir_signature", lambda *_args: "sig")
    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)

    payload = routes._handle_list_dir(
        object(), urlparse("/api/list?session_id=sess-authority&path=.")
    )
    selected["workspace"] = str(fallback_b)

    assert payload["workspace"] == str(fallback_a.resolve())
    assert session.workspace == str(fallback_a.resolve())
    assert captured["listed"] == fallback_a.resolve()
    assert str(fallback_a.resolve()) in sidecar.read_text(encoding="utf-8")


def test_persisted_list_recovery_anchors_later_create_dir_to_fallback_a(
    monkeypatch, tmp_path
):
    """Reproduce the Maintainer's A→B sequence across fresh sidecar loads."""
    session_dir = tmp_path / "sessions"
    session_dir.mkdir()
    fallback_a = tmp_path / "fallback-a"
    fallback_b = tmp_path / "fallback-b"
    fallback_a.mkdir()
    fallback_b.mkdir()
    stale = tmp_path / "deleted-workspace"
    selected = {"workspace": str(fallback_a)}
    sid = "sess-http-authority"

    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
    monkeypatch.setattr(models, "SESSIONS", OrderedDict())
    monkeypatch.setattr(models, "_write_session_index", lambda **_kwargs: None)
    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_last_workspace", lambda: selected["workspace"])
    monkeypatch.setattr(models, "get_last_workspace", lambda: selected["workspace"])
    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400: {"error": message, "status": status},
    )

    session = models.Session(
        session_id=sid,
        workspace=str(stale),
        messages=[{"role": "user", "content": "preserve transcript"}],
    )
    session.save(skip_index=True)
    models.SESSIONS.clear()

    listed = routes._handle_list_dir(
        object(), urlparse(f"/api/list?session_id={sid}&path=.")
    )
    persisted = models.Session.load(sid)
    assert listed["workspace"] == str(fallback_a.resolve())
    assert listed["workspace_recovered"] is True
    assert persisted is not None
    assert persisted.workspace == str(fallback_a.resolve())
    assert persisted.messages == [{"role": "user", "content": "preserve transcript"}]

    models.SESSIONS.clear()
    selected["workspace"] = str(fallback_b)
    created = routes._handle_create_dir(
        object(), {"session_id": sid, "path": "gate-wrong-root"}
    )

    assert created == {"ok": True, "path": "gate-wrong-root"}
    assert (fallback_a / "gate-wrong-root").is_dir()
    assert not (fallback_b / "gate-wrong-root").exists()


def test_list_recovery_persistence_failure_fails_closed(monkeypatch, tmp_path):
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    stale = tmp_path / "deleted-workspace"
    calls = {"list_dir": 0}
    session = SimpleNamespace(
        session_id="sess-save-fails",
        workspace=str(stale),
        save=lambda **_kwargs: (_ for _ in ()).throw(OSError("disk full")),
    )

    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(workspace, "load_workspaces", lambda: [])
    monkeypatch.setattr(routes, "get_session", lambda _sid, **_kwargs: session)
    monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
    monkeypatch.setattr(
        routes,
        "list_dir",
        lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
    )
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400: {"error": message, "status": status},
    )

    payload = routes._handle_list_dir(
        object(), urlparse("/api/list?session_id=sess-save-fails&path=.")
    )

    assert payload["status"] >= 400
    assert "persist" in payload["error"].lower()
    assert calls["list_dir"] == 0
    assert session.workspace == str(stale)


def test_list_dir_does_not_recover_unpersistable_cli_workspace(
    monkeypatch, tmp_path
):
    stale = tmp_path / "deleted-cli-workspace"
    fallback = tmp_path / "fallback"
    fallback.mkdir()
    calls = {"fallback": 0, "list_dir": 0}

    monkeypatch.setattr(routes, "get_session", lambda _sid: (_ for _ in ()).throw(KeyError(_sid)))
    monkeypatch.setattr(
        routes,
        "get_cli_sessions",
        lambda: [{"session_id": "cli-stale", "workspace": str(stale)}],
    )

    def get_fallback():
        calls["fallback"] += 1
        return str(fallback)

    monkeypatch.setattr(routes, "get_last_workspace", get_fallback)
    monkeypatch.setattr(
        routes,
        "list_dir",
        lambda *_args: calls.__setitem__("list_dir", calls["list_dir"] + 1),
    )
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400: {"error": message, "status": status},
    )

    payload = routes._handle_list_dir(
        object(), urlparse("/api/list?session_id=cli-stale&path=.")
    )

    assert payload["status"] == 404
    assert calls == {"fallback": 0, "list_dir": 0}


@pytest.mark.parametrize(
    "terminal_cfg",
    [
        pytest.param(
            {"backend": "ssh", "cwd": "/Users/joeyshiue"},
            id="cwd-absolute",
        ),
        pytest.param({"backend": "ssh"}, id="cwd-omitted"),
        pytest.param({"backend": "ssh", "cwd": ""}, id="cwd-empty"),
        pytest.param({"backend": "ssh", "cwd": "."}, id="cwd-dot"),
    ],
)
def test_list_dir_preserves_remote_workspace_rejection(
    monkeypatch, tmp_path, terminal_cfg
):
    candidate = "/Users/other/projects/demo"
    fallback_path = tmp_path / "fallback"
    fallback_path.mkdir()
    session = SimpleNamespace(session_id="sess-list-remote", workspace=candidate)
    calls = {"fallback": 0, "list_dir": 0}

    monkeypatch.setattr(
        api_config,
        "get_config",
        lambda: {"terminal": terminal_cfg},
    )
    monkeypatch.setattr(workspace, "_home_path", lambda: tmp_path)
    monkeypatch.setattr(routes, "get_session", lambda _sid: session)

    def fallback():
        calls["fallback"] += 1
        return fallback_path

    def fake_list_dir(*_args):
        calls["list_dir"] += 1
        return []

    monkeypatch.setattr(routes, "get_last_workspace", fallback)
    monkeypatch.setattr(routes, "list_dir", fake_list_dir)
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400: {"error": message, "status": status},
    )

    payload = routes._handle_list_dir(
        object(), urlparse("/api/list?session_id=sess-list-remote&path=.")
    )

    assert isinstance(payload, dict)
    assert payload["status"] == 404
    assert "Path does not exist" in payload["error"]
    assert calls == {"fallback": 0, "list_dir": 0}