"""Integration: the admin sources CRUD API (phase 35, task 02; local
kind, phase 38, task 02).

Real Postgres (``podman compose up -d db``); the ``BOR_GIT_SOURCES``
fallback is exercised deterministically by monkeypatching the router's
``get_settings`` with a fresh ``Settings(_env_file=None, git_sources=…)``
(same pattern as ``test_sync_api.py`` — the dev ``.env`` never leaks in).

Contract under test:

* anonymous → 403 ``{"detail": "admin only"}`` on GET, POST (git and
  local), and DELETE (phase 16 pattern, same as ``/api/sync``);
* GET — rows carry ``kind`` + ``path`` (phase 38); empty table + env
  set → the git-only env rows (``kind: "git"``, ``path: null``) with
  ``from_env: true`` and null ``id``/``added_at``; empty table + empty
  env → ``sources: []`` with ``from_env: true``; any DB rows →
  ``from_env: false`` and the env var is ignored (the phase's locked
  decision); DB rows ordered by ``(added_at, id)``;
* POST ``kind=git`` (default) — 201 stored trimmed; duplicate (even with
  different surrounding whitespace) → 409 with a generic detail that
  never echoes the URL (credential safety), including when only the DB
  unique index catches it; bad shape / blank / >500 chars → 422, also
  input-free (the phase-35 contract, unchanged);
* POST ``kind=local`` (phase 38) — existing directory → 201, stored row
  carries ``kind=local`` + the path expanded (``~`` resolved) and
  trimmed; relative / missing / not-a-directory path → 422 naming the
  path (not a secret); duplicate path → 409 naming the path (unique
  index as backstop); wrong field combinations (git without url, local
  without path, both kinds' fields) → 422;
* DELETE — 204 and gone; an emptied table falls back to the env list
  again; unknown id → 404;
* ignore paths (phase 89) — POST accepts the RAW box lines for both
  kinds (optional, absent → ``[]``), stored normalized (A1) with the A4
  fixed-detail 422s (shared gate with PATCH); GET reports each row's
  stored list (env rows ``[]``);
* hidden-folders flag (phase 105) — POST accepts ``include_hidden`` for
  both kinds (optional, absent → stored ``False``, A4); GET reports the
  stored flag (env rows ``False`` — no DB row to store a flag on);
* ``PATCH /{source_id}`` (admin-only) — phase 89 A5 + phase 105: each
  field is optional, present-wins — ``ignore_paths`` when present
  REPLACES the list wholesale (A5 — an empty list clears all),
  ``include_hidden`` when present sets the flag, both absent → 200
  no-op, 404 unknown id, 422 fixed details for the A4 limits, a
  rejected list never half-applies the flag, the row otherwise
  unchanged.

``git_sources`` is global state: truncated around every test.

Phase 121 (task 02 — clone-time credential + output sanitization):
POST normalizes an old-style embedded ``user:pass@`` URL into the bare
URL + ``token`` column (explicit ``token`` wins — LOCKED A6), the
duplicate check runs on the bare URL, the PATCH token is tri-state
(absent/None = no change, non-empty = replace, "" = clear) and
re-normalizes a legacy row's URL, and every response (list DB rows,
list env rows, POST 201, PATCH 200) is token-free — asserted on the
RAW response text — while a token row's SYNC clone receives the
injected ``https://x-access-token:<token>@…`` URL with a
credential-free checkout path (mock ``clone_or_pull``) and a legacy
row's clone still receives its ORIGINAL stored URL (the credential
keeps working).

Phase 135 (task 01 — branch storage + reporting): the output shapes
gain ``branch`` (NULL = the repo's default branch) — list DB rows
report the stored value, env-fallback rows report ``null`` (no DB row
to store one on), and POST 201 / PATCH 200 report the stored value.

Phase 135 (task 02 — branch validation + editing): POST stores a
branch on GIT rows only (absent/empty → NULL = the default branch;
non-empty → :func:`app.rag.git_sources.validate_branch_name`, a
``ValueError`` → the FIXED-detail 422 ``not a valid branch name`` —
the input is never echoed, and the row is not created); a ``branch``
on a LOCAL row is 422 ``branch is only valid for git sources`` (POST
and PATCH alike — local rows are walked, not cloned); PATCH edits
the branch with the token's EXACT tri-state — absent/None = no
change, non-empty = set (re-validated), empty string = clear (stored
NULL).
"""
from __future__ import annotations

import time
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, text
from sqlalchemy.orm import Session

from app.api import git_sources as git_sources_api
from app.api import sync as sync_api
from app.config import Settings
from app.main import app as fastapi_app
from app.models import GitSource
from app.rag import git_sources as git_sources_resolver
from app.rag.importer import ImportSummary


@pytest.fixture(autouse=True)
def clean_git_sources(db: Session) -> Iterator[None]:
    """The stored list is global state: reset around every test."""
    db.execute(text("TRUNCATE git_sources"))
    db.commit()
    yield
    db.execute(text("TRUNCATE git_sources"))
    db.commit()


def _settings(git_sources: str = "", **overrides: object) -> Settings:
    """Fresh settings with the ``.env`` file ignored; the explicit
    kwargs beat any process env leaks (test_sync_api pattern).

    ``**overrides`` carries the per-test dir pins (``sources_dir`` /
    ``upload_dir`` — the phase-69 total-removal DELETE must never aim a
    rmtree at the operator's real ``~/bor-sources``).
    """
    return Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        git_sources=git_sources,
        **overrides,  # pyright: ignore[reportCallIssue]
    )


# --- anonymous -------------------------------------------------------------


def test_anonymous_gets_403_on_all_routes(client: TestClient, db: Session) -> None:
    r = client.get("/api/git-sources")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    r = client.post("/api/git-sources", json={"url": "https://anon.example.com/x.git"})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    # The phase-38 local kind is gated the same way.
    r = client.post("/api/git-sources", json={"kind": "local", "path": "/tmp"})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    r = client.delete(f"/api/git-sources/{uuid.uuid4()}")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    # The phase-89 ignore-list PATCH is gated the same way — and so is
    # the phase-105 bool-only payload (the toggle's exact request).
    r = client.patch(f"/api/git-sources/{uuid.uuid4()}", json={"ignore_paths": ["a"]})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    r = client.patch(f"/api/git-sources/{uuid.uuid4()}", json={"include_hidden": True})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    # Nothing landed in the table.
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


# --- GET: env fallback -----------------------------------------------------


def test_get_empty_table_with_env_returns_env_rows(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: _settings("https://a.example.com/one.git, git@b.example.com:two.git ,"),
    )

    r = admin_client.get("/api/git-sources")
    assert r.status_code == 200
    body = r.json()
    assert body["from_env"] is True
    # Whitespace-trimmed, empty entries dropped, order preserved; null
    # ids; the env fallback is git-only (phase 38: kind + path fields).
    assert body["sources"] == [
        {
            "id": None,
            "kind": "git",
            "url": "https://a.example.com/one.git",
            "path": None,
            "added_at": None,
            # Env rows have no DB row to store a list on (phase 89), a
            # flag on (phase 105) or a branch on (phase 135).
            "ignore_paths": [],
            "include_hidden": False,
            "branch": None,
        },
        {
            "id": None,
            "kind": "git",
            "url": "git@b.example.com:two.git",
            "path": None,
            "added_at": None,
            "ignore_paths": [],
            "include_hidden": False,
            "branch": None,
        },
    ]


def test_get_empty_table_with_empty_env_returns_empty_list(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())

    r = admin_client.get("/api/git-sources")
    assert r.status_code == 200
    assert r.json() == {"sources": [], "from_env": True}


def test_get_orders_db_rows_by_added_at_then_id(admin_client: TestClient, db: Session) -> None:
    base = datetime.now(UTC) - timedelta(hours=3)
    db.add_all(
        [
            GitSource(url="https://example.com/oldest.git", added_at=base),
            GitSource(url="https://example.com/second.git", added_at=base + timedelta(hours=1)),
            GitSource(url="https://example.com/newest.git", added_at=base + timedelta(hours=2)),
            # Same transaction → identical server-stamped added_at (≈ now,
            # after the explicit rows): the id tie-break decides their order.
            GitSource(url="https://example.com/tie-a.git"),
            GitSource(url="https://example.com/tie-b.git"),
        ]
    )
    db.commit()
    tie_rows = db.scalars(
        select(GitSource).where(GitSource.url.like("https://example.com/tie-%"))
    ).all()
    tie_ids = sorted(row.id for row in tie_rows)

    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is False
    assert [s["url"] for s in body["sources"][:3]] == [
        "https://example.com/oldest.git",
        "https://example.com/second.git",
        "https://example.com/newest.git",
    ]
    assert [s["id"] for s in body["sources"][3:]] == [str(i) for i in tie_ids]
    # DB rows carry real ids + timestamps (the env shape has neither).
    for s in body["sources"]:
        assert s["id"] is not None
        assert s["added_at"] is not None


# --- POST: create ----------------------------------------------------------


def test_post_creates_trimmed_and_list_stops_using_env(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: _settings("https://env.example.com/env.git"),
    )

    r = admin_client.post("/api/git-sources", json={"url": "  https://new.example.com/repo.git  "})
    assert r.status_code == 201
    body = r.json()
    assert body["url"] == "https://new.example.com/repo.git"  # trimmed
    uuid.UUID(body["id"])
    assert body["added_at"] is not None
    # Phase 89: the response gains ``ignore_paths`` — absent at create
    # time → ``[]``; phase 105 adds ``include_hidden`` — absent →
    # ``False`` (A4); phase 135 adds ``branch`` — absent → ``null``.
    assert set(body) == {"id", "url", "added_at", "ignore_paths", "include_hidden", "branch"}
    assert body["ignore_paths"] == []
    assert body["include_hidden"] is False
    assert body["branch"] is None

    # The DB row now wins: from_env False, the env URL is gone from the list.
    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is False
    assert [s["url"] for s in body["sources"]] == ["https://new.example.com/repo.git"]


def test_post_accepts_accepted_url_shapes(admin_client: TestClient) -> None:
    for url in (
        "https://github.com/owner/repo.git",
        "http://git.local/repo.git",
        "ssh://git@example.com/repo.git",
        "git@github.com:owner/repo.git",
        # Phase 135 (task 05): the file:// local transport — the E2E
        # fixture repos the UI adds end to end (the phase-28/32/107
        # transport-true idiom, promoted from the env fallback).
        "file:///srv/repos/homelab-notes",
    ):
        r = admin_client.post("/api/git-sources", json={"url": url})
        assert r.status_code == 201, f"{url} must be accepted: {r.text}"
        assert r.json()["url"] == url


def test_post_duplicate_url_returns_409_without_echoing_url(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The 409 detail is a fixed generic string — URLs may embed
    ``user:pass@`` credentials (phase 32's masking discipline)."""
    monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())
    url = "https://user:secret@example.com/creds.git"

    assert admin_client.post("/api/git-sources", json={"url": url}).status_code == 201
    # Same URL with different surrounding whitespace: the trim makes it a
    # duplicate too.
    r = admin_client.post("/api/git-sources", json={"url": f"  {url}\t"})
    assert r.status_code == 409
    detail = r.json()["detail"]
    assert detail == "a git source with this URL already exists"
    assert url not in detail
    assert "user:secret" not in detail
    # Exactly one row stored.
    assert len(admin_client.get("/api/git-sources").json()["sources"]) == 1


def test_post_concurrent_insert_backstop_still_409(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """If the duplicate pre-check misses (a concurrent insert lands
    between the check and the commit), the DB unique index still yields
    the generic 409 — never a 500."""
    monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())
    url = "https://example.com/backstop.git"
    assert admin_client.post("/api/git-sources", json={"url": url}).status_code == 201

    real_select = git_sources_api.select

    def blind_select(*args: Any, **kwargs: Any) -> Any:
        if args and args[0] is GitSource:  # the duplicate pre-check
            # …now never matches — only the unique index can catch it.
            return real_select(GitSource).where(GitSource.url == "zz-never-matches")
        return real_select(*args, **kwargs)

    monkeypatch.setattr(git_sources_api, "select", blind_select)
    r = admin_client.post("/api/git-sources", json={"url": url})
    assert r.status_code == 409
    detail = r.json()["detail"]
    assert detail == "a git source with this URL already exists"
    assert url not in detail


def test_post_rejects_invalid_shapes_without_echoing_input(
    admin_client: TestClient, db: Session
) -> None:
    """Bad shapes are 422 with a detail that never repeats the submitted
    value."""
    for bad in ("not a url", "host:repo", "ftp://example.com/x.git"):
        r = admin_client.post("/api/git-sources", json={"url": bad})
        assert r.status_code == 422, f"{bad!r} must be rejected"
        assert bad not in r.text, f"422 detail must not echo the input ({bad!r})"
    # Nothing stored.
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_post_rejects_blank_and_oversized_urls(admin_client: TestClient, db: Session) -> None:
    """Whitespace-only (trim → empty) and >500-char URLs are 422; the
    500-char boundary passes."""
    assert admin_client.post("/api/git-sources", json={"url": " \t\n "}).status_code == 422
    assert admin_client.post("/api/git-sources", json={"url": "x" * 501}).status_code == 422
    boundary = "https://" + "x" * 492  # exactly 500 chars
    assert len(boundary) == 500
    assert admin_client.post("/api/git-sources", json={"url": boundary}).status_code == 201
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 1


# --- POST: local kind (phase 38, task 02) ----------------------------------


def test_post_local_creates_stored_row_with_expanded_path(
    admin_client: TestClient,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``kind=local`` + an existing directory → 201; the stored row
    carries ``kind=local`` and the path expanded (``~`` resolved via the
    server's ``HOME``, whitespace trimmed)."""
    monkeypatch.setenv("HOME", str(tmp_path / "home"))
    real_dir = tmp_path / "home" / "notes"
    real_dir.mkdir(parents=True)

    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": "  ~/notes\t"})
    assert r.status_code == 201, r.text
    body = r.json()
    # The phase-35 response shape is unchanged — the local row reports
    # its (expanded) path in ``url``; ``kind`` + ``path`` via GET;
    # phase 89 adds ``ignore_paths`` (absent → ``[]``); phase 105 adds
    # ``include_hidden`` (absent → ``False``, A4); phase 135 adds
    # ``branch`` (absent → ``null``).
    assert set(body) == {"id", "url", "added_at", "ignore_paths", "include_hidden", "branch"}
    assert body["ignore_paths"] == []
    assert body["include_hidden"] is False
    assert body["branch"] is None
    uuid.UUID(body["id"])
    assert body["url"] == str(real_dir)
    assert body["added_at"] is not None

    row = admin_client.get("/api/git-sources").json()["sources"][0]
    assert row["kind"] == "local"
    assert row["path"] == str(real_dir)
    assert row["url"] == str(real_dir)
    assert row["id"] is not None
    assert row["added_at"] is not None


def test_post_local_stores_trimmed_normalized_path(
    admin_client: TestClient, tmp_path: Path
) -> None:
    """Absolute path with surrounding whitespace + trailing slash →
    stored clean (trimmed, normalized)."""
    real_dir = tmp_path / "plain"
    real_dir.mkdir()

    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": f"  {real_dir}/  "})
    assert r.status_code == 201, r.text
    row = admin_client.get("/api/git-sources").json()["sources"][0]
    assert row["path"] == str(real_dir)


def test_post_local_relative_path_returns_422_naming_path(
    admin_client: TestClient, db: Session
) -> None:
    """A relative path fails loud at add-time — 422 naming the path
    (relative or not, it is never stored)."""
    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": "relative/dir"})
    assert r.status_code == 422
    assert r.json()["detail"] == "local source path is not a directory: relative/dir"
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_post_local_missing_path_returns_422_naming_path(
    admin_client: TestClient, db: Session
) -> None:
    """A missing (absolute) path is a user error → 422 naming the path so
    the owner sees exactly which directory failed."""
    missing = f"/nonexistent/bor-test-{uuid.uuid4()}"
    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": missing})
    assert r.status_code == 422
    assert r.json()["detail"] == f"local source path is not a directory: {missing}"
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_post_local_file_not_dir_returns_422(
    admin_client: TestClient, tmp_path: Path
) -> None:
    """An existing *file* is not a directory → 422 naming the path."""
    a_file = tmp_path / "a-file.md"
    a_file.write_text("not a directory")
    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": str(a_file)})
    assert r.status_code == 422
    assert r.json()["detail"] == f"local source path is not a directory: {a_file}"


def test_post_local_duplicate_path_returns_409_naming_path(
    admin_client: TestClient, tmp_path: Path
) -> None:
    """Duplicate path (even with different surrounding whitespace) → 409
    naming the path (a path is not a secret, unlike a git URL)."""
    real_dir = tmp_path / "dups"
    real_dir.mkdir()

    assert admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": f"  {real_dir}\t"})
    assert r.status_code == 409
    detail = r.json()["detail"]
    assert detail == f"a local source with this path already exists: {real_dir}"
    # Exactly one row stored.
    assert len(admin_client.get("/api/git-sources").json()["sources"]) == 1


def test_post_local_concurrent_insert_backstop_still_409(
    admin_client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """If the duplicate pre-check misses (a concurrent insert lands
    between the check and the commit), the DB unique index on ``path``
    still yields the 409 naming the path — never a 500."""
    real_dir = tmp_path / "backstop"
    real_dir.mkdir()
    assert (
        admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
        .status_code
        == 201
    )

    real_select = git_sources_api.select

    def blind_select(*args: Any, **kwargs: Any) -> Any:
        if args and args[0] is GitSource:  # the duplicate pre-check
            # …now never matches — only the unique index can catch it.
            return real_select(GitSource).where(GitSource.url == "zz-never-matches")
        return real_select(*args, **kwargs)

    monkeypatch.setattr(git_sources_api, "select", blind_select)
    r = admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
    assert r.status_code == 409
    assert r.json()["detail"] == f"a local source with this path already exists: {real_dir}"


def test_post_wrong_field_combinations_return_422(
    admin_client: TestClient, db: Session, tmp_path: Path
) -> None:
    """git without url, local without path, and both kinds' fields are
    422 with fixed details — nothing is stored."""
    real_dir = tmp_path / "combo"
    real_dir.mkdir()

    assert admin_client.post("/api/git-sources", json={"kind": "git"}).status_code == 422
    r = admin_client.post(
        "/api/git-sources",
        json={"kind": "git", "url": "https://example.com/both.git", "path": str(real_dir)},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "a git source takes a url, not a path"
    assert admin_client.post("/api/git-sources", json={"kind": "local"}).status_code == 422
    # Whitespace-only path trims to empty → 422 as well (the schema's
    # min-length guard).
    assert (
        admin_client.post("/api/git-sources", json={"kind": "local", "path": "  "}).status_code
        == 422
    )
    r = admin_client.post(
        "/api/git-sources",
        json={"kind": "local", "url": "https://example.com/both.git", "path": str(real_dir)},
    )
    assert r.status_code == 422
    assert r.json()["detail"] == "a local source takes a path, not a url"
    # Unknown kind and an oversized path are 422 too.
    assert (
        admin_client.post(
            "/api/git-sources", json={"kind": "svn", "url": "https://example.com/x.git"}
        ).status_code
        == 422
    )
    assert (
        admin_client.post(
            "/api/git-sources", json={"kind": "local", "path": "/" + "x" * 2000}
        ).status_code
        == 422
    )
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_get_mixed_kinds_in_added_order(
    admin_client: TestClient, db: Session, tmp_path: Path
) -> None:
    """GET mixes git + local rows in ``(added_at, id)`` order; git rows
    report ``path: null``, local rows their stored path."""
    git_url = "https://example.com/mixed.git"
    db.add(GitSource(url=git_url, kind="git", added_at=datetime.now(UTC) - timedelta(hours=1)))
    db.commit()
    real_dir = tmp_path / "mixed"
    real_dir.mkdir()
    assert admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})

    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is False
    assert [s["url"] for s in body["sources"]] == [git_url, str(real_dir)]
    git_row, local_row = body["sources"]
    assert git_row["kind"] == "git"
    assert git_row["path"] is None
    assert git_row["id"] is not None
    assert local_row["kind"] == "local"
    assert local_row["path"] == str(real_dir)
    assert local_row["id"] is not None


# --- DB rows win over env ---------------------------------------------------


def test_db_rows_win_over_env(admin_client: TestClient, db: Session, monkeypatch) -> None:
    """Seed a row AND set the env: the GET returns only the DB rows and
    ``from_env: false`` — the env var is ignored once the table has rows."""
    monkeypatch.setattr(
        git_sources_api, "get_settings", lambda: _settings("https://env.example.com/env.git")
    )
    db.add(GitSource(url="https://db.example.com/db.git"))
    db.commit()

    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is False
    assert [s["url"] for s in body["sources"]] == ["https://db.example.com/db.git"]
    for s in body["sources"]:
        assert s["id"] is not None
        assert s["added_at"] is not None


# --- DELETE -----------------------------------------------------------------


def test_delete_removes_row_and_falls_back_to_env(
    admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    # Phase 69: the DELETE is a total removal — the (absent) checkout
    # dir it would rmtree is pinned at fresh tmp dirs, so the test never
    # depends on the operator's real ``~/bor-sources`` being clean; the
    # document prune it performs owns the KB tables the same way the
    # other suites do (no row's source name may shadow real docs).
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: _settings(
            "https://env.example.com/env.git",
            sources_dir=str(tmp_path / "sources"),
            upload_dir=str(tmp_path / "uploads"),
        ),
    )
    created = admin_client.post("/api/git-sources", json={"url": "https://new.example.com/x.git"})
    assert created.status_code == 201

    assert admin_client.delete(f"/api/git-sources/{created.json()['id']}").status_code == 204

    # The table is empty again → the env fallback is live once more
    # (git-only rows, phase 38).
    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is True
    assert body["sources"] == [
        {
            "id": None,
            "kind": "git",
            "url": "https://env.example.com/env.git",
            "path": None,
            "added_at": None,
            "ignore_paths": [],  # env rows: no DB row to store a list on
            "include_hidden": False,  # … a flag on (phase 105)
            "branch": None,  # … or a branch on (phase 135)
        }
    ]
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


def test_delete_unknown_id_returns_404(admin_client: TestClient) -> None:
    r = admin_client.delete(f"/api/git-sources/{uuid.uuid4()}")
    assert r.status_code == 404
    assert r.json() == {"detail": "git source not found"}


def test_delete_invalid_id_returns_422(admin_client: TestClient) -> None:
    assert admin_client.delete("/api/git-sources/not-a-uuid").status_code == 422


# --- ignore paths (phase 89) ----------------------------------------------


def test_post_stores_normalized_ignore_paths_both_kinds(
    admin_client: TestClient, tmp_path: Path
) -> None:
    """POST accepts the RAW box lines for both kinds (A1 normalization
    in the API layer) — the normalized list is stored and round-trips
    through GET."""
    r = admin_client.post(
        "/api/git-sources",
        json={
            "url": "https://example.com/ig.git",
            "ignore_paths": ["/my/files/", "  my/files2  ", "x"],
        },
    )
    assert r.status_code == 201, r.text
    assert r.json()["ignore_paths"] == ["my/files", "my/files2", "x"]

    real_dir = tmp_path / "ig"
    real_dir.mkdir()
    r = admin_client.post(
        "/api/git-sources",
        json={"kind": "local", "path": str(real_dir), "ignore_paths": ["//skip/", "keep"]},
    )
    assert r.status_code == 201, r.text
    assert r.json()["ignore_paths"] == ["skip", "keep"]

    # Round-trip through GET (keyed by url — (added_at, id) order of two
    # same-millisecond inserts is not the point under test).
    by_url = {
        s["url"]: s["ignore_paths"] for s in admin_client.get("/api/git-sources").json()["sources"]
    }
    assert by_url["https://example.com/ig.git"] == ["my/files", "my/files2", "x"]
    assert by_url[str(real_dir)] == ["skip", "keep"]


def test_post_without_ignore_paths_reports_empty_list(
    admin_client: TestClient, db: Session
) -> None:
    """Absent ``ignore_paths`` at create time → ``[]`` — both in the 201
    response and in the GET round-trip (the migration's server default)."""
    r = admin_client.post("/api/git-sources", json={"url": "https://example.com/none.git"})
    assert r.status_code == 201
    assert r.json()["ignore_paths"] == []
    body = admin_client.get("/api/git-sources").json()
    assert body["sources"][0]["ignore_paths"] == []
    # The stored value is the JSONB server default, not Python-only.
    row = db.scalars(select(GitSource)).one()
    assert row.ignore_paths == []


def test_post_rejects_invalid_ignore_paths_like_patch(
    admin_client: TestClient, db: Session
) -> None:
    """POST shares ``_validate_ignore_paths`` with PATCH — the same A4
    fixed 422 details; nothing is stored."""
    for payload, detail in (
        (["  "], "ignore paths must be non-empty"),
        ([f"e{i}" for i in range(201)], "a source has at most 200 ignore paths"),
        (["a" * 501], "an ignore path exceeds 500 characters"),
    ):
        r = admin_client.post(
            "/api/git-sources",
            json={"url": "https://example.com/bad-ignore.git", "ignore_paths": payload},
        )
        assert r.status_code == 422, f"{detail!r}: {r.text}"
        assert r.json()["detail"] == detail
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_get_reports_stored_ignore_paths_and_env_rows_empty(
    admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
    """GET reports each DB row's stored list; env-fallback rows (no DB
    row to store a list on) report ``[]`` with ``from_env: true``."""
    db.add(GitSource(url="https://example.com/stored.git", ignore_paths=["docs/drafts"]))
    db.commit()
    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is False
    assert body["sources"][0]["ignore_paths"] == ["docs/drafts"]

    db.execute(text("DELETE FROM git_sources"))
    db.commit()
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: _settings("https://a.example.com/env.git"),
    )
    body = admin_client.get("/api/git-sources").json()
    assert body["from_env"] is True
    assert body["sources"] == [
        {
            "id": None,
            "kind": "git",
            "url": "https://a.example.com/env.git",
            "path": None,
            "added_at": None,
            "ignore_paths": [],
            "include_hidden": False,
            "branch": None,  # env rows: no DB row to store a branch on (135)
        }
    ]


def test_patch_replaces_ignore_paths_including_clear(admin_client: TestClient) -> None:
    """PATCH 200 — REPLACE semantics (A5): the body list, normalized,
    becomes the row's whole list; an empty list clears all; ``id``/
    ``url``/``added_at`` are unchanged; the response is the
    ``GitSourceOut`` shape incl. the new list."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/patch.git", "ignore_paths": ["old/"]},
    )
    assert created.status_code == 201
    before = created.json()

    r = admin_client.patch(f"/api/git-sources/{before['id']}", json={"ignore_paths": ["a/", "b"]})
    assert r.status_code == 200, r.text
    body = r.json()
    assert set(body) == {"id", "url", "added_at", "ignore_paths", "include_hidden", "branch"}
    assert body["ignore_paths"] == ["a", "b"]  # normalized
    assert body["include_hidden"] is False  # untouched by a list-only PATCH
    assert body["branch"] is None  # untouched by a list-only PATCH
    for key in ("id", "url", "added_at"):
        assert body[key] == before[key]
    # Round-trip through GET.
    assert (
        admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == ["a", "b"]
    )

    # An empty list clears all.
    r = admin_client.patch(f"/api/git-sources/{before['id']}", json={"ignore_paths": []})
    assert r.status_code == 200, r.text
    assert r.json()["ignore_paths"] == []
    assert admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == []


def test_patch_empty_body_is_a_noop_200(admin_client: TestClient) -> None:
    """Phase 105: BOTH fields absent (or explicit None) → 200 no-op —
    the row is untouched. (The pre-phase-105 pin of an absent
    ``ignore_paths`` as a 422 is retired — the list is optional now; a
    PRESENT list keeps the phase-89 A5 replace semantics.)"""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/noop.git", "ignore_paths": ["a/b"]},
    )
    assert created.status_code == 201
    before = created.json()

    r = admin_client.patch(f"/api/git-sources/{before['id']}", json={})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["ignore_paths"] == ["a/b"]
    assert body["include_hidden"] is False
    for key in ("id", "url", "added_at"):
        assert body[key] == before[key]
    # Explicit None values are "absent" too — still a no-op.
    r = admin_client.patch(
        f"/api/git-sources/{before['id']}", json={"ignore_paths": None, "include_hidden": None}
    )
    assert r.status_code == 200, r.text
    assert r.json()["ignore_paths"] == ["a/b"]
    assert r.json()["include_hidden"] is False
    # The row never changed — round-trip through GET.
    assert admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == ["a/b"]


def test_patch_unknown_id_returns_404(admin_client: TestClient) -> None:
    r = admin_client.patch(f"/api/git-sources/{uuid.uuid4()}", json={"ignore_paths": ["a"]})
    assert r.status_code == 404
    assert r.json() == {"detail": "git source not found"}


def test_patch_invalid_id_returns_422(admin_client: TestClient) -> None:
    r = admin_client.patch("/api/git-sources/not-a-uuid", json={"ignore_paths": []})
    assert r.status_code == 422


def test_patch_422s_are_fixed_details(admin_client: TestClient) -> None:
    """The A4 422s are exact fixed strings (never echoing the input),
    and a rejected PATCH leaves the row's list unchanged."""
    created = admin_client.post(
        "/api/git-sources", json={"url": "https://example.com/v.git", "ignore_paths": ["keep"]}
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    for payload, detail in (
        # Whitespace-only → empty after normalization: 422, not a silent
        # drop (the UI drops blank lines client-side; the API is
        # defensive).
        (["   "], "ignore paths must be non-empty"),
        (list(f"e{i}" for i in range(201)), "a source has at most 200 ignore paths"),
        (["a" * 501], "an ignore path exceeds 500 characters"),
    ):
        r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": payload})
        assert r.status_code == 422, f"{detail!r}: {r.text}"
        assert r.json()["detail"] == detail
    # Nothing changed by the failed PATCHes.
    assert admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == ["keep"]


def test_patch_accepts_a4_boundaries(admin_client: TestClient) -> None:
    """Exactly 200 entries and a 500-char entry (post-normalization)
    are the accepted edge of A4."""
    created = admin_client.post("/api/git-sources", json={"url": "https://example.com/bnd.git"})
    assert created.status_code == 201
    sid = created.json()["id"]

    r = admin_client.patch(
        f"/api/git-sources/{sid}", json={"ignore_paths": [f"e/{i}" for i in range(200)]}
    )
    assert r.status_code == 200, r.text
    assert len(r.json()["ignore_paths"]) == 200

    long_entry = "a" * 500
    assert len(long_entry) == 500
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": [long_entry]})
    assert r.status_code == 200, r.text
    assert r.json()["ignore_paths"] == [long_entry]


# --- hidden-folders flag (phase 105) ----------------------------------------


def test_get_reports_include_hidden_default_false(admin_client: TestClient, db: Session) -> None:
    """A fresh stored row (no flag passed) reports ``include_hidden:
    false`` — in the GET round-trip AND on the model (the column's
    server default, A4)."""
    db.add(GitSource(url="https://example.com/fresh.git"))
    db.commit()
    body = admin_client.get("/api/git-sources").json()
    assert body["sources"][0]["include_hidden"] is False
    row = db.scalars(select(GitSource)).one()
    assert row.include_hidden is False


def test_post_stores_include_hidden_both_kinds(
    admin_client: TestClient, tmp_path: Path
) -> None:
    """POST accepts ``include_hidden`` for both kinds (present → stored
    as sent); the 201 body and the GET round-trip report it."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/hidden-git.git", "include_hidden": True},
    )
    assert r.status_code == 201, r.text
    assert r.json()["include_hidden"] is True

    real_dir = tmp_path / "hidden-local"
    real_dir.mkdir()
    r = admin_client.post(
        "/api/git-sources",
        json={"kind": "local", "path": str(real_dir), "include_hidden": True},
    )
    assert r.status_code == 201, r.text
    assert r.json()["include_hidden"] is True

    by_url = {
        s["url"]: s["include_hidden"]
        for s in admin_client.get("/api/git-sources").json()["sources"]
    }
    assert by_url["https://example.com/hidden-git.git"] is True
    assert by_url[str(real_dir)] is True


def test_patch_bool_only_sets_flag_leaves_list(admin_client: TestClient) -> None:
    """The toggle's exact payload — ``{"include_hidden": …}`` alone:
    the flag is set, the ignore list is UNCHANGED (present-wins, phase
    105)."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/toggle.git", "ignore_paths": ["a/b"]},
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    r = admin_client.patch(f"/api/git-sources/{sid}", json={"include_hidden": True})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["include_hidden"] is True
    assert body["ignore_paths"] == ["a/b"]  # untouched
    assert admin_client.get("/api/git-sources").json()["sources"][0]["include_hidden"] is True

    # And back off again.
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"include_hidden": False})
    assert r.status_code == 200, r.text
    assert r.json()["include_hidden"] is False
    assert r.json()["ignore_paths"] == ["a/b"]


def test_patch_list_only_replaces_list_leaves_flag(admin_client: TestClient) -> None:
    """The dialog's exact payload — ``{"ignore_paths": …}`` alone:
    the list is REPLACED (normalized, phase-89 A5), the flag is
    UNCHANGED — byte-identical to the pre-phase-105 dialog PATCH."""
    created = admin_client.post(
        "/api/git-sources",
        json={
            "url": "https://example.com/dialog.git",
            "ignore_paths": ["a/b"],
            "include_hidden": True,
        },
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": ["c/d", " e "]})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["ignore_paths"] == ["c/d", "e"]  # normalized
    assert body["include_hidden"] is True  # untouched
    assert admin_client.get("/api/git-sources").json()["sources"][0]["include_hidden"] is True


def test_patch_both_fields_apply_independently(admin_client: TestClient) -> None:
    """``{"ignore_paths": [], "include_hidden": true}`` — the list is
    cleared AND the flag is set in one request."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/both.git", "ignore_paths": ["old/one"]},
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    r = admin_client.patch(
        f"/api/git-sources/{sid}", json={"ignore_paths": [], "include_hidden": True}
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["ignore_paths"] == []  # cleared
    assert body["include_hidden"] is True  # set


def test_patch_422_does_not_half_apply_flag(admin_client: TestClient) -> None:
    """A PRESENT bad list 422s with the fixed A4 details — and the
    OTHER field in the same body never half-applies: after each 422
    the flag is still the stored value (validation raises before any
    assignment, request transaction untouched)."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/half.git", "ignore_paths": ["keep"]},
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    for payload, detail in (
        (["   "], "ignore paths must be non-empty"),
        ([f"e{i}" for i in range(201)], "a source has at most 200 ignore paths"),
        (["a" * 501], "an ignore path exceeds 500 characters"),
    ):
        r = admin_client.patch(
            f"/api/git-sources/{sid}", json={"ignore_paths": payload, "include_hidden": True}
        )
        assert r.status_code == 422, f"{detail!r}: {r.text}"
        assert r.json()["detail"] == detail
    # Nothing half-applied: list AND flag are both still the originals.
    row = admin_client.get("/api/git-sources").json()["sources"][0]
    assert row["ignore_paths"] == ["keep"]
    assert row["include_hidden"] is False


# --- token: write-path normalization (phase 121, task 02) ------------------

TOKEN = "ghp_phase121secrettoken"


def test_post_with_token_stores_column_and_never_echoes(
    admin_client: TestClient, db: Session
) -> None:
    """Bare URL + masked token → the token is stored in the dedicated
    column and appears in NO API response — asserted on the RAW text of
    both the 201 and the GET list (the response shapes have no token
    field by contract; this pins that nothing else smuggles it out)."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://github.com/owner/private.git", "token": TOKEN},
    )
    assert r.status_code == 201, r.text
    body = r.json()
    assert body["url"] == "https://github.com/owner/private.git"
    assert set(body) == {"id", "url", "added_at", "ignore_paths", "include_hidden", "branch"}
    assert body["branch"] is None  # no branch sent → the default-branch NULL
    assert TOKEN not in r.text

    row = db.scalars(select(GitSource)).one()
    assert row.url == "https://github.com/owner/private.git"  # bare
    assert row.token == TOKEN  # the column, not the URL

    r = admin_client.get("/api/git-sources")
    assert r.status_code == 200
    assert TOKEN not in r.text
    assert r.json()["sources"][0]["url"] == "https://github.com/owner/private.git"


def test_post_embedded_token_url_normalizes_to_bare_plus_column(
    admin_client: TestClient, db: Session
) -> None:
    """The TODO L5 paste — old-style ``user:token@`` URL with no token
    field: stored BARE, the embedded PASSWORD part moved to the token
    column, and every response is token-free (raw text)."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": f"https://myuser:{TOKEN}@github.com/owner/private-repo.git"},
    )
    assert r.status_code == 201, r.text
    assert r.json()["url"] == "https://github.com/owner/private-repo.git"
    assert TOKEN not in r.text
    assert "myuser" not in r.text

    row = db.scalars(select(GitSource)).one()
    assert row.url == "https://github.com/owner/private-repo.git"
    assert row.token == TOKEN  # the password part, not the whole userinfo

    r = admin_client.get("/api/git-sources")
    assert TOKEN not in r.text
    assert r.json()["sources"][0]["url"] == "https://github.com/owner/private-repo.git"


def test_post_explicit_token_wins_over_embedded(
    admin_client: TestClient, db: Session
) -> None:
    """LOCKED A6 — the masked field and the pasted URL disagree: the
    explicit field is the intent and beats the embedded credential."""
    r = admin_client.post(
        "/api/git-sources",
        json={
            "url": f"https://myuser:{TOKEN}@github.com/owner/private.git",
            "token": "ghp_explicitwins",
        },
    )
    assert r.status_code == 201, r.text
    assert r.json()["url"] == "https://github.com/owner/private.git"

    row = db.scalars(select(GitSource)).one()
    assert row.url == "https://github.com/owner/private.git"
    assert row.token == "ghp_explicitwins"


def test_post_username_as_token_form_moves_whole_run(
    admin_client: TestClient, db: Session
) -> None:
    """The documented GitHub shape (no colon) — the whole userinfo run
    is the credential."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": f"https://{TOKEN}@github.com/owner/private.git"},
    )
    assert r.status_code == 201, r.text
    row = db.scalars(select(GitSource)).one()
    assert row.url == "https://github.com/owner/private.git"
    assert row.token == TOKEN


def test_post_same_repo_different_token_is_409_on_bare_url(
    admin_client: TestClient, db: Session
) -> None:
    """The duplicate check runs on the NORMALIZED bare URL: the same
    repo pasted with a different credential is the same source — 409,
    not a second row (both via embedded and via explicit token)."""
    assert (
        admin_client.post(
            "/api/git-sources",
            json={"url": f"https://user1:{TOKEN}@github.com/owner/dup.git"},
        ).status_code
        == 201
    )
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://user2:other-cred@github.com/owner/dup.git"},
    )
    assert r.status_code == 409
    assert r.json()["detail"] == "a git source with this URL already exists"
    assert TOKEN not in r.text
    # And the bare form of the same repo 409s too.
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://github.com/owner/dup.git", "token": "another"},
    )
    assert r.status_code == 409
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 1


def test_post_local_kind_token_is_inert_and_never_echoed(
    admin_client: TestClient, tmp_path: Path, db: Session
) -> None:
    """Local-kind sources have no URL credential (the design does not
    touch them): a token on a local row is stored inert (never used —
    local rows are walked, not cloned) and, like on git rows, never
    echoed."""
    real_dir = tmp_path / "local-tok"
    real_dir.mkdir()
    r = admin_client.post(
        "/api/git-sources",
        json={"kind": "local", "path": str(real_dir), "token": TOKEN},
    )
    assert r.status_code == 201, r.text
    assert TOKEN not in r.text
    assert r.json()["url"] == str(real_dir)

    row = db.scalars(select(GitSource)).one()
    assert row.token == TOKEN  # inert — clone_url_for never sees it
    assert TOKEN not in admin_client.get("/api/git-sources").text


# --- token: PATCH tri-state (phase 121, task 02) ---------------------------


def test_patch_token_replace_clear_and_noop(
    admin_client: TestClient, db: Session
) -> None:
    """The tri-state on a clean-URL row: absent/None = no change,
    non-empty = replace, "" = clear (stored NULL) — the token is never
    in any response (raw text)."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://github.com/owner/patch.git", "token": TOKEN},
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    # Absent (and explicit None) = no change.
    # The endpoint commits in its own session — expire this one's
    # identity map before reading the row back (the house pattern for
    # cross-session reads).
    def stored() -> GitSource:
        db.expire_all()
        row = db.get(GitSource, uuid.UUID(sid))
        assert row is not None
        return row

    r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": ["a"]})
    assert r.status_code == 200, r.text
    assert TOKEN not in r.text
    assert r.json()["url"] == "https://github.com/owner/patch.git"
    assert stored().token == TOKEN

    r = admin_client.patch(
        f"/api/git-sources/{sid}", json={"ignore_paths": None, "token": None}
    )
    assert r.status_code == 200
    assert stored().token == TOKEN

    # Non-empty = replace (the URL is untouched — a clean URL comes
    # back byte-identical).
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"token": "ghp_replaced"})
    assert r.status_code == 200, r.text
    assert TOKEN not in r.text
    assert "ghp_replaced" not in r.text
    assert r.json()["url"] == "https://github.com/owner/patch.git"
    row = stored()
    assert row.url == "https://github.com/owner/patch.git"
    assert row.token == "ghp_replaced"

    # "" = clear — stored NULL.
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"token": ""})
    assert r.status_code == 200, r.text
    row = stored()
    assert row.token is None
    assert r.json()["url"] == "https://github.com/owner/patch.git"


def test_patch_token_on_legacy_row_renormalizes_url(
    admin_client: TestClient, db: Session
) -> None:
    """A pre-phase row (credential embedded in the stored URL, token
    NULL) gets its userinfo stripped (moved to the column) the first
    time an explicit credential is written — and a "" clear also moves
    it (a cleared credential is gone from both the column and the URL;
    the owner explicitly asked for no credential)."""
    legacy = f"https://user:{TOKEN}@github.com/owner/legacy.git"
    db.add(GitSource(url=legacy))
    db.commit()
    row = db.scalars(select(GitSource)).one()
    sid = row.id

    r = admin_client.patch(f"/api/git-sources/{sid}", json={"token": "ghp_new"})
    assert r.status_code == 200, r.text
    assert TOKEN not in r.text
    assert "ghp_new" not in r.text
    assert r.json()["url"] == "https://github.com/owner/legacy.git"
    db.expire_all()  # the endpoint committed in its own session
    row = db.get(GitSource, sid)
    assert row is not None
    assert row.url == "https://github.com/owner/legacy.git"  # normalized bare
    assert row.token == "ghp_new"

    # "" clears the (now column) credential — the URL stays bare.
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"token": ""})
    assert r.status_code == 200, r.text
    db.expire_all()
    row = db.get(GitSource, sid)
    assert row is not None
    assert row.url == "https://github.com/owner/legacy.git"
    assert row.token is None


def test_patch_token_409_backstop_on_renormalized_collision(
    admin_client: TestClient, db: Session
) -> None:
    """A legacy embedded row + a bare row for the same repo can only
    coexist pre-phase; re-normalizing the legacy row's URL on a token
    PATCH makes the stored URLs collide — the unique index yields the
    generic 409 (never a 500) and the failed PATCH leaves the row
    untouched."""
    legacy = f"https://user:{TOKEN}@github.com/owner/collide.git"
    bare = "https://github.com/owner/collide.git"
    db.add_all([GitSource(url=legacy), GitSource(url=bare)])
    db.commit()
    legacy_row = db.scalars(select(GitSource).where(GitSource.url == legacy)).one()

    r = admin_client.patch(f"/api/git-sources/{legacy_row.id}", json={"token": "ghp_x"})
    assert r.status_code == 409
    assert r.json()["detail"] == "a git source with this URL already exists"
    assert TOKEN not in r.text
    # The rollback left the legacy row exactly as it was.
    db.expire_all()
    row = db.get(GitSource, legacy_row.id)
    assert row is not None
    assert row.url == legacy
    assert row.token is None


# --- token: output sanitization (phase 121, task 02) ------------------------


def test_get_masks_legacy_embedded_token_row(admin_client: TestClient, db: Session) -> None:
    """Completion criterion: a legacy row (token embedded in the
    stored URL, token column NULL) clones with its original URL but
    its API output is token-free — the stored DB value is untouched,
    the response is bare."""
    legacy = f"https://user:{TOKEN}@github.com/owner/legacy.git"
    db.add(GitSource(url=legacy))
    db.commit()

    r = admin_client.get("/api/git-sources")
    assert r.status_code == 200
    assert TOKEN not in r.text
    assert "user:" not in r.text
    assert r.json()["sources"][0]["url"] == "https://github.com/owner/legacy.git"
    # The stored value is untouched (the clone still authenticates).
    assert db.scalars(select(GitSource)).one().url == legacy


def test_get_masks_env_fallback_rows_with_embedded_token(
    admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """An env URL can embed a token too — the ENV VALUE itself is
    untouched (the config is the operator's), only the response is
    masked."""
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: _settings(f"https://user:{TOKEN}@env.example.com/env.git"),
    )
    r = admin_client.get("/api/git-sources")
    assert r.status_code == 200
    assert TOKEN not in r.text
    assert r.json()["from_env"] is True
    assert r.json()["sources"][0]["url"] == "https://env.example.com/env.git"


def test_get_clean_urls_are_byte_identical(admin_client: TestClient, db: Session) -> None:
    """The phase-50/35 contract through the mask: credential-free
    stored URLs (including ``git@`` and local paths) surface verbatim."""
    urls = [
        "https://github.com/owner/clean.git",
        "git@github.com:owner/scp.git",
        "ssh://git@example.com/repo.git",
    ]
    base = datetime.now(UTC) - timedelta(hours=1)
    # Distinct added_at: same-timestamp rows order by random uuid4 id.
    db.add_all(
        GitSource(url=u, added_at=base + timedelta(minutes=i)) for i, u in enumerate(urls)
    )
    db.commit()
    body = admin_client.get("/api/git-sources").json()
    assert [s["url"] for s in body["sources"]] == urls


# --- token: the sync clone URL (phase 121, task 02) -------------------------


@pytest.fixture()
def sync_admin_client() -> Iterator[TestClient]:
    """A context-managed, logged-in client — one app event loop across
    requests (the sync background task must survive between the POST
    and the polls; the test_sync_api.py pattern)."""
    with TestClient(fastapi_app) as client:
        r = client.post("/api/login", json={"password": "test-admin-password"})
        assert r.status_code == 204
        yield client


def _stub_sync_stack(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> list[tuple[str, Path]]:
    """The sync pipeline minus the real git/import/LLM layers (the
    test_sync_api.py pattern): a fresh settings dir, an empty env list
    (the DB row wins), a no-op model probe, a zero-count import
    (change-gated steps skipped), and a recording clone.
    Returns the clone call list."""
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            sources_dir=str(tmp_path / "sources"),
        ),
    )
    monkeypatch.setattr(
        git_sources_resolver,
        "get_settings",
        lambda: Settings(_env_file=None, git_sources=""),  # pyright: ignore[reportCallIssue]
    )

    async def fake_check_models(llm: object) -> None:
        return None

    monkeypatch.setattr(sync_api, "check_models", fake_check_models)

    clone_calls: list[tuple[str, Path]] = []

    def fake_clone_or_pull(url: str, dest: Path | str, branch: str | None = None) -> Path:
        # phase 135: the row's branch pin (None for this file's rows)
        dest = Path(dest)
        dest.mkdir(parents=True, exist_ok=True)
        clone_calls.append((url, dest))
        return dest

    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone_or_pull)

    async def fake_import_sources(*args: object, **kwargs: object) -> ImportSummary:
        return ImportSummary()  # all-zero: the change-gated steps skip

    monkeypatch.setattr(sync_api, "import_sources", fake_import_sources)
    # The unchanged-walk gap probe: no candidates → no generator call
    # (hermetic — the real probe reads the global KB state, and the
    # generator would burn real lite calls).
    monkeypatch.setattr(sync_api, "missing_folder_summaries", lambda db: [])
    return clone_calls


def _poll_sync(client: TestClient, want: str, timeout: float = 5.0) -> dict:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        body = client.get("/api/sync/status").json()
        if body["state"] == want:
            return body
        assert body["state"] == "running", body
        time.sleep(0.02)
    raise AssertionError(f"sync did not reach {want!r} in {timeout}s")


def test_sync_token_row_clones_with_injected_url_and_bare_checkout(
    sync_admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """A token row's sync: ``clone_or_pull`` receives the injected
    ``https://x-access-token:<token>@…`` URL, the checkout directory is
    derived from the BARE URL (credential-free), and no token leaks
    into the status surface (raw text)."""
    db.add(GitSource(url="https://github.com/owner/priv.git", token=TOKEN))
    db.commit()
    clone_calls = _stub_sync_stack(monkeypatch, tmp_path)

    assert sync_admin_client.post("/api/sync").status_code == 202
    body = _poll_sync(sync_admin_client, "success")
    assert body["error"] is None
    assert TOKEN not in str(body)

    assert len(clone_calls) == 1
    clone_url, dest = clone_calls[0]
    assert clone_url == f"https://x-access-token:{TOKEN}@github.com/owner/priv.git"
    # The checkout name is the bare repo name — no userinfo, no token.
    assert dest.name == "priv"
    assert dest == tmp_path / "sources" / "priv"
    assert TOKEN not in str(dest)


def test_sync_legacy_row_clones_with_original_stored_url(
    sync_admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Completion criterion: a pre-phase row (credential embedded in
    the stored URL, token NULL) produces the ORIGINAL stored URL at
    clone time — the credential keeps working — and the checkout name
    is still the credential-free repo name (repo_name on the bare-URL
    semantics: the basename after the last / of the stored URL)."""
    stored = f"https://user:{TOKEN}@github.com/owner/priv.git"
    db.add(GitSource(url=stored))
    db.commit()
    clone_calls = _stub_sync_stack(monkeypatch, tmp_path)

    assert sync_admin_client.post("/api/sync").status_code == 202
    body = _poll_sync(sync_admin_client, "success")
    assert body["error"] is None

    assert len(clone_calls) == 1
    clone_url, dest = clone_calls[0]
    assert clone_url == stored  # the original stored URL, verbatim
    assert dest.name == "priv"


def test_sync_public_row_clones_with_verbatim_url(
    sync_admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Public repos behave byte-identically to pre-phase: a NULL-token
    row's clone URL is the stored URL itself (no injection)."""
    url = "https://github.com/owner/public.git"
    db.add(GitSource(url=url))
    db.commit()
    clone_calls = _stub_sync_stack(monkeypatch, tmp_path)

    assert sync_admin_client.post("/api/sync").status_code == 202
    assert _poll_sync(sync_admin_client, "success")["error"] is None

    assert clone_calls == [(url, tmp_path / "sources" / "public")]


# --- branch: create + validation (phase 135, task 02) ----------------------


def test_post_git_source_with_branch_stores_and_reports(
    admin_client: TestClient, db: Session
) -> None:
    """POST a git source with ``branch: "release"`` → 201 reports the
    branch; the stored model row carries it; the GET round-trip
    reports it too (201, list DB row)."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/branched.git", "branch": "release"},
    )
    assert r.status_code == 201, r.text
    body = r.json()
    assert body["branch"] == "release"
    assert set(body) == {"id", "url", "added_at", "ignore_paths", "include_hidden", "branch"}

    row = db.scalars(select(GitSource)).one()
    assert row.branch == "release"  # the column, not response-only

    listed = admin_client.get("/api/git-sources").json()["sources"][0]
    assert listed["branch"] == "release"


def test_post_git_source_branch_is_trimmed(admin_client: TestClient) -> None:
    """Surrounding whitespace on the branch is trimmed (the schema's
    ``_trim_branch`` + the API's defensive strip) before storage."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/trim.git", "branch": "  release/1.2\t"},
    )
    assert r.status_code == 201, r.text
    assert r.json()["branch"] == "release/1.2"


def test_post_git_source_empty_branch_stores_null(admin_client: TestClient) -> None:
    """Absent/None AND trimmed-EMPTY branch at create → stored NULL
    (the repo's default branch — the pre-phase-135 behavior): the
    create-time split from PATCH's clear leg (GitSourceIn docstring)."""
    for i, branch_value in enumerate((None, "", "   ")):
        payload: dict[str, object] = {
            "url": f"https://example.com/def-{i}.git"
        }  # distinct URLs: the loop must not 409 on itself
        if branch_value is not None:
            payload["branch"] = branch_value
        r = admin_client.post("/api/git-sources", json=payload)
        assert r.status_code == 201, r.text
        assert r.json()["branch"] is None


def test_post_git_source_invalid_branch_422_fixed_detail_no_row(
    admin_client: TestClient, db: Session
) -> None:
    """Invalid branch names are 422 with the EXACT fixed detail
    ``not a valid branch name`` — the input is never echoed (the
    credential-safety discipline, applied for consistency) — and no
    row is created."""
    for bad in ("a..b", "a b", "a@b", "-x", ".x", "/x", "x/", "x.", "HEAD", "head"):
        r = admin_client.post(
            "/api/git-sources",
            json={"url": "https://example.com/bad-branch.git", "branch": bad},
        )
        assert r.status_code == 422, f"{bad!r} must be rejected: {r.text}"
        assert r.json()["detail"] == "not a valid branch name"  # EXACT
        assert bad not in r.text, f"422 must not echo the input ({bad!r})"
    # Nothing stored.
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_post_git_source_oversized_branch_422(admin_client: TestClient, db: Session) -> None:
    """>200 chars: the schema's ``max_length=200`` rejects it before
    the handler (FastAPI's standard schema 422 — the same house
    precedent as the oversized-URL 422, which this router also
    answers with the framework's standard shape); no row is created.
    The API-layer branch 422s (charset/sequence/leading/trailing/
    HEAD) are the FIXED-detail ones — see the adjacent tests."""
    r = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/long-branch.git", "branch": "a" * 201},
    )
    assert r.status_code == 422
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


def test_post_local_source_with_branch_422(
    admin_client: TestClient, tmp_path: Path, db: Session
) -> None:
    """A ``branch`` on a LOCAL row is 422 with the fixed detail (local
    rows are walked, not cloned) — for BOTH valid and invalid branch
    names (the kind check precedes validation); nothing is stored."""
    real_dir = tmp_path / "local-branch"
    real_dir.mkdir()
    for bad in ("main", "a..b"):
        r = admin_client.post(
            "/api/git-sources",
            json={"kind": "local", "path": str(real_dir), "branch": bad},
        )
        assert r.status_code == 422, f"{bad!r} must be rejected: {r.text}"
        assert r.json()["detail"] == "branch is only valid for git sources"
    assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0


# --- branch: PATCH tri-state (phase 135, task 02) ---------------------------


def test_patch_branch_set_change_clear_and_noop(
    admin_client: TestClient, db: Session
) -> None:
    """The branch tri-state on a git row — the token's exact
    semantics field-for-field:

    * set: ``"hotfix/x"`` → 200 reports it (stored + GET round-trip);
    * change: ``"hotfix/y"`` → 200 reports the new value;
    * clear: ``""`` → 200 reports ``null`` (stored NULL = the
      default branch); whitespace-only trims to empty → the same
      clear leg;
    * no-op: branch ABSENT (or explicit None) → 200, the stored
      branch survives (a flag-only PATCH must not touch it).
    """
    created = admin_client.post(
        "/api/git-sources", json={"url": "https://example.com/patch-branch.git"}
    )
    assert created.status_code == 201
    sid = created.json()["id"]
    assert created.json()["branch"] is None  # started on the default branch

    def stored_branch() -> str | None:
        # The endpoint commits in its own session — expire this one's
        # identity map before reading back (the house pattern).
        db.expire_all()
        row = db.get(GitSource, uuid.UUID(sid))
        assert row is not None
        return row.branch

    # SET — "hotfix/x".
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": "hotfix/x"})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] == "hotfix/x"
    assert stored_branch() == "hotfix/x"
    assert (
        admin_client.get("/api/git-sources").json()["sources"][0]["branch"] == "hotfix/x"
    )

    # CHANGE — "hotfix/y".
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": "hotfix/y"})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] == "hotfix/y"
    assert stored_branch() == "hotfix/y"

    # NO-OP — an unrelated field only: the branch is UNTOUCHED.
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"include_hidden": True})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] == "hotfix/y"
    assert r.json()["include_hidden"] is True
    assert stored_branch() == "hotfix/y"

    # NO-OP — explicit None is absent too.
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": None})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] == "hotfix/y"
    assert stored_branch() == "hotfix/y"

    # CLEAR — "" (the tri-state's clear leg; a deliberate empty save).
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": ""})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] is None
    assert stored_branch() is None

    # Whitespace-only trims to empty → the same clear leg.
    admin_client.patch(f"/api/git-sources/{sid}", json={"branch": "hotfix/x"})
    r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": "   "})
    assert r.status_code == 200, r.text
    assert r.json()["branch"] is None
    assert stored_branch() is None


def test_patch_branch_invalid_422_leaves_row_unchanged(
    admin_client: TestClient, db: Session
) -> None:
    """A PRESENT invalid branch 422s with the EXACT fixed detail —
    and the stored branch is untouched (validation raises before the
    assignment; the row's other fields in the same body never
    half-apply)."""
    created = admin_client.post(
        "/api/git-sources",
        json={"url": "https://example.com/patch-bad.git", "branch": "stable"},
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    for bad in ("a..b", "HEAD", "x/"):
        r = admin_client.patch(
            f"/api/git-sources/{sid}", json={"branch": bad, "include_hidden": True}
        )
        assert r.status_code == 422, f"{bad!r}: {r.text}"
        assert r.json()["detail"] == "not a valid branch name"
        assert bad not in r.text
    # Nothing half-applied: the branch AND the flag are still the
    # originals.
    row = admin_client.get("/api/git-sources").json()["sources"][0]
    assert row["branch"] == "stable"
    assert row["include_hidden"] is False


def test_patch_branch_on_local_row_422(admin_client: TestClient, tmp_path: Path) -> None:
    """A PRESENT branch on a LOCAL row is the same fixed 422 as POST
    (local rows are walked, not cloned) — valid name or not; the row
    is untouched."""
    real_dir = tmp_path / "local-patch"
    real_dir.mkdir()
    created = admin_client.post(
        "/api/git-sources", json={"kind": "local", "path": str(real_dir)}
    )
    assert created.status_code == 201
    sid = created.json()["id"]

    for branch_value in ("main", "a..b"):
        r = admin_client.patch(f"/api/git-sources/{sid}", json={"branch": branch_value})
        assert r.status_code == 422, f"{branch_value!r}: {r.text}"
        assert r.json()["detail"] == "branch is only valid for git sources"
    row = admin_client.get("/api/git-sources").json()["sources"][0]
    assert row["branch"] is None
    assert row["kind"] == "local"