brain-of-reese/tests/integration/test_git_sources_upload.py

"""Integration: the admin archive-upload API (phase 49, task 02;
backgrounded in phase 64, task 03; scan deferred in phase 90, task 01).

Real Postgres (``podman compose up -d db``); the upload dir is pointed
at a fresh tmp dir per test by monkeypatching the router's
``get_settings`` (the ``test_git_sources_api.py`` pattern — the dev
``.env`` never leaks in).

Phase 90 (task 01) contract under test: ``POST`` answers **202** the
moment the archive is safely on disk, and the rest — unpack → swap →
row upsert, and **nothing else** (no model check, no import, no
overview refresh — the scan is the RAG page's "Sync sources" button's
job, phase 90 A1) — runs in a **background task** behind
``GET /api/git-sources/upload/status`` (the phase-32 ``SyncStatus``
shape). The success ``detail`` is the no-count ``{"message":
"uploaded"}`` payload and ``current_file`` / ``files_done`` /
``files_total`` stay null/0/0 for the whole run (phase 90 A2 — the key
set is unchanged; the live file label belongs to the sync). Every
post-receive scenario observes the **status endpoint, polled until
terminal** — never the HTTP response; post-202 failures are status
states (``failed`` + sanitized error, phase 64 A5), never HTTP errors.
The upload touches NO KB state: zero documents, zero chunks, no
overview row — asserted on the happy paths (the sync suites already
prove a fresh upload row is imported on the next sync, and the phase-90
E2E proves the full loop with an edited ignore list).

The inline gates are unchanged:

* anonymous → 403 ``{"detail": "admin only"}`` (the router's
  ``require_admin`` covers the new status route too);
* name/format gate → 422: a non-archive extension names the accepted
  formats; a ``..`` / separator / empty-stem name (including a bare
  ``tar.gz``) is rejected with the task-01 message — and the upload
  dir is never created for a rejected name (control characters never
  reach the app: the multipart transport percent-encodes them — the
  task-01 branch for them is covered in ``test_archive_upload.py``);
* one upload at a time → 409 ``an upload is already in progress`` —
  the gate is the module-level **flag** (checked and set with no await
  in between, BEFORE the receive — the handler now awaits before the
  background task exists), held through the whole background run;
* streaming cap → 413 naming the ``upload_max_mb`` cap; the temp
  ``.upload`` file is removed and the flag is released (no stray
  ``.`` files in the upload dir; the next upload is not 409).

Background branches (phase 64 A5): unpack failure (zip-slip, tar
symlink escape, corrupt archive), zero-entry archive, swap failure,
the concurrent-insert ``IntegrityError`` backstop, and a cancelled run
(the ``finally`` cleans both temps + releases the flag) — each lands
in the status as ``failed`` (with the sanitized error) or ``success``
(with the no-count payload in ``detail``); a FAILED upload **never**
touches the previous folder, row, or KB of an earlier good upload (the
no-partial-state locked decision, asserted explicitly).

The mid-run observations (409 window, in-flight status, TRUNCATE
regression) park the run in a SYNCHRONOUS gate (unpack or swap — the
run has no await: phase 90 removed the scan). A synchronous park also
holds the 202's response delivery on the first client (the TestClient
portal wakes the main thread with a loop callback that cannot run
while the loop is blocked), so those tests DRIVE ``_run_upload``
directly on a worker loop and observe from a SECOND client (its own
loop), the module-level status, or the test's own thread. The
flip side of the fast, synchronous-only run is pinned at the handler
level instead: by the time the 202 is observed, the background run
(unpack → swap → upsert, queued on the loop BEFORE the response is
delivered) has already landed its terminal state — the status is the
source of truth, and the 202 body never carries the outcome.

``git_sources`` / ``documents`` / ``chunks`` / ``kb_overview`` are
global state: truncated around every test.
"""
from __future__ import annotations

import asyncio
import contextlib
import io
import logging
import os
import re
import tarfile
import threading
import time
import zipfile
from collections.abc import Iterator
from pathlib import Path
from typing import Any, Literal

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

from app.api import git_sources as git_sources_api
from app.config import Settings
from app.db import SessionLocal
from app.main import app as fastapi_app
from app.models import GitSource
from app.rag.archive_upload import ArchiveUploadError, swap_in, unpack_archive
from tests.conftest import ADMIN_PASSWORD


@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()


@pytest.fixture(autouse=True)
def clean_documents(db: Session) -> Iterator[None]:
    """Phase 90: the upload touches no KB state — these tables must be
    empty BEFORE every test so the ``_kb_untouched`` assertions (zero
    documents / chunks / overview rows) are meaningful, and they stay
    empty after (global state, truncated around every test)."""
    db.execute(text("TRUNCATE chunks, documents, kb_overview"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents, kb_overview"))
    db.commit()


@pytest.fixture(autouse=True)
def fresh_upload_status() -> Iterator[None]:
    """The module-level upload status + flag are process-global: reset
    them before AND after every test (the ``fresh_sync_status`` pattern
    from ``tests/unit/test_sync_button.py``) — a previous test's
    terminal state must never leak into the next one's idle-shape
    assertions or its 409 expectations."""
    git_sources_api._upload_status = git_sources_api.UploadStatus()
    git_sources_api._upload_in_progress = False
    yield
    git_sources_api._upload_status = git_sources_api.UploadStatus()
    git_sources_api._upload_in_progress = False


@pytest.fixture()
def upload_client() -> Iterator[TestClient]:
    """Admin-signed client (the context-manager form keeps one event
    loop for the whole test — the background task the 202 creates runs
    on it, so the client must stay open until the run is observed in a
    terminal state)."""
    with TestClient(fastapi_app) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
        yield client


def _point_at(monkeypatch: pytest.MonkeyPatch, upload_dir: Path, upload_max_mb: int = 512) -> None:
    """Fresh settings on the router's module: the tmp upload dir and
    (optionally) a shrunk cap — the dev ``.env`` never leaks in."""
    monkeypatch.setattr(
        git_sources_api,
        "get_settings",
        lambda: Settings(
            _env_file=None,  # pyright: ignore[reportCallIssue]
            upload_dir=str(upload_dir),
            upload_max_mb=upload_max_mb,
        ),
    )


#: tarfile write modes used by the tests (uncompressed + gzip).
_TAR_WRITE_MODES = Literal["w", "w:gz"]


def _tarball(path: Path, files: dict[str, str], compress: _TAR_WRITE_MODES = "w:gz") -> Path:
    with tarfile.open(path, compress) as tf:
        for rel, content in files.items():
            data = content.encode("utf-8")
            info = tarfile.TarInfo(rel)
            info.size = len(data)
            info.mtime = 0
            tf.addfile(info, io.BytesIO(data))
    return path


def _zip(path: Path, files: dict[str, str]) -> Path:
    with zipfile.ZipFile(path, "w") as zf:
        for rel, content in files.items():
            zf.writestr(rel, content)
    return path


def _targz_bytes(files: dict[str, str]) -> bytes:
    """A tarball in memory (no temp file on disk)."""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        for rel, content in files.items():
            data = content.encode("utf-8")
            info = tarfile.TarInfo(rel)
            info.size = len(data)
            info.mtime = 0
            tf.addfile(info, io.BytesIO(data))
    return buf.getvalue()


def _post(client: TestClient, filename: str, payload: bytes):
    """One upload request (multipart, the page's exact shape)."""
    return client.post(
        "/api/git-sources/upload",
        files={"file": (filename, payload, "application/octet-stream")},
    )


def _wait_status(client: TestClient, timeout: float = 30.0) -> dict[str, Any]:
    """Poll ``GET /api/git-sources/upload/status`` until the run is
    terminal (``success`` / ``failed``) and return the terminal body.

    Phase 64: the 202 answers BEFORE the background run completes —
    every post-receive assertion observes the status endpoint, not the
    HTTP response. Phase 90: the run is short (unpack → swap → row
    upsert, no scan), so a terminal state is usually already there by
    the first poll — the poll loop is the shape contract, not a
    timing assumption."""
    deadline = time.monotonic() + timeout
    body: dict[str, Any] = {}
    while True:
        r = client.get("/api/git-sources/upload/status")
        assert r.status_code == 200, r.text
        body = r.json()
        if body["state"] in ("success", "failed"):
            return body
        if time.monotonic() >= deadline:
            raise AssertionError(f"upload never reached a terminal state: {body}")
        time.sleep(0.05)


def _row(db: Session, path: str) -> GitSource | None:
    db.expire_all()
    return db.scalar(select(GitSource).where(GitSource.path == path))


def _docs(client: TestClient) -> list[tuple[str, str]]:
    body = client.get("/api/docs").json()
    return [(d["source"], d["path"]) for d in body["documents"]]


def _count_rows(db: Session) -> int:
    return db.execute(text("SELECT count(*) FROM git_sources")).scalar_one()


def _kb_untouched(db: Session, client: TestClient) -> None:
    """Phase 90: the upload never touches the KB — zero documents,
    zero chunks, no overview row (the scan that populated them moved
    to the RAG page's "Sync sources" button)."""
    assert _docs(client) == []
    assert db.execute(text("SELECT count(*) FROM chunks")).scalar_one() == 0
    assert db.execute(text("SELECT count(*) FROM kb_overview")).scalar_one() == 0


def _good_upload(
    client: TestClient, name: str = "safe", files: dict[str, str] | None = None
) -> None:
    """A successful upload of a two-sentinel tarball (202 → status
    ``success`` — no KB state, phase 90) — the baseline state the
    no-partial-state tests protect."""
    payload_files = files or {
        "alpha.md": "# Alpha\noriginal sentinel one\n",
        "bravo.md": "# Bravo\noriginal sentinel two\n",
    }
    r = _post(client, f"{name}.tar.gz", _targz_bytes(payload_files))
    assert r.status_code == 202, r.text
    assert _wait_status(client)["state"] == "success"


def _assert_previous_intact(
    client: TestClient, db: Session, upload_dir: Path, name: str, files: dict[str, str]
) -> None:
    """The no-partial-state locked decision, asserted explicitly: after a
    FAILED upload the previous folder's content, the row, and the KB
    are all exactly as the good upload left them (phase 90: an upload
    touches no KB state — zero documents) — and no temp file
    survived."""
    folder = upload_dir / name
    assert {p.name: p.read_text(encoding="utf-8") for p in folder.iterdir()} == files
    assert _row(db, str(folder)) is not None  # the row is still there
    assert _count_rows(db) == 1  # …and no second row appeared
    assert _docs(client) == []  # the KB was never touched (phase 90)
    assert [p.name for p in upload_dir.iterdir()] == [name]  # no stray temp


class _GatedUnpack:
    """Parks inside ``unpack_archive`` — a SYNCHRONOUS seam (the app
    loop blocks while parked — the observer must come from a second
    client or the test's own thread, never the first client's loop) —
    then runs the REAL unpack so the run completes for real."""

    def __init__(self, started: threading.Event, release: threading.Event) -> None:
        self.started = started
        self.release = release

    def __call__(self, archive: Path, target: Path, max_bytes: int) -> None:
        self.started.set()
        self.release.wait(30.0)
        unpack_archive(archive, target, max_bytes)


class _GatedSwap:
    """Parks inside ``swap_in`` (the same synchronous-seam contract as
    :class:`_GatedUnpack`) — then runs the REAL swap so the run
    completes for real."""

    def __init__(self, started: threading.Event, release: threading.Event) -> None:
        self.started = started
        self.release = release

    def __call__(self, new_dir: Path, final_dir: Path) -> None:
        self.started.set()
        self.release.wait(30.0)
        swap_in(new_dir, final_dir)


@contextlib.contextmanager
def _second_client() -> Iterator[TestClient]:
    """A fresh admin-signed client on its OWN event loop — the mid-run
    observer for tests whose first client's loop is blocked in a
    synchronous park (phase 90: the run has no await, so its gates are
    synchronous)."""
    with TestClient(fastapi_app) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
        yield client


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


def test_anonymous_upload_and_status_get_403(client: TestClient, db: Session) -> None:
    r = _post(client, "homelab.tar.gz", b"not an archive at all")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    # The status route sits behind the same router-level dependency.
    r = client.get("/api/git-sources/upload/status")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    assert _count_rows(db) == 0


def test_idle_upload_status_pins_full_shape(upload_client: TestClient) -> None:
    """Idle: the full response dict is pinned — identical key set to
    ``GET /api/sync/status`` (phase 32), progress keys null/0/0."""
    assert upload_client.get("/api/git-sources/upload/status").json() == {
        "state": "idle",
        "started_at": None,
        "finished_at": None,
        "detail": {},
        "error": None,
        "current_file": None,
        "files_done": 0,
        "files_total": 0,
    }


# --- name / format gate -----------------------------------------------------


def test_non_archive_extension_gets_422_naming_formats(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)

    r = _post(upload_client, "notes.txt", b"hello world")
    assert r.status_code == 422
    assert r.json()["detail"] == "only .tar, .tar.gz, .tgz or .zip archives are accepted"
    assert _count_rows(db) == 0
    # The name gate runs before the dir is created — nothing on disk.
    assert not uploads.exists()


@pytest.mark.parametrize(
    "filename",
    [
        "../evil.zip",  # ``..`` + separator
        "sub/dir.tar.gz",  # separator
        "tar.gz",  # bare suffix → empty stem
        ".tar.gz",  # empty stem
        # (control characters in the filename are percent-encoded by the
        # multipart transport before the app ever sees them — the
        # ``archive_source_name`` branch for them is covered in
        # tests/unit/test_archive_upload.py)
    ],
)
def test_unsafe_name_gets_422(
    upload_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    filename: str,
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)

    r = _post(upload_client, filename, b"content")
    assert r.status_code == 422, f"{filename!r} must be rejected: {r.text}"
    assert r.json()["detail"] != "only .tar, .tar.gz, .tgz or .zip archives are accepted"
    assert _count_rows(db) == 0
    assert not uploads.exists()


# --- one at a time (409) ----------------------------------------------------


def test_second_upload_while_one_is_in_flight_returns_409(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """While a background run is in flight, the second upload is 409 —
    the module-level flag, held from the first receive, is the gate —
    and after the first run lands in a terminal state, uploads are
    accepted again (a full real run this time).

    The park is a SYNCHRONOUS seam (phase 90: the run is unpack → swap
    → row upsert, no await), so the in-flight run is driven directly
    on a worker loop (the house background-task pattern from
    ``test_sync_button.py``); the second upload comes from a real HTTP
    client (its own loop) to observe the 409 while the first run is
    held."""
    uploads = tmp_path / "uploads"
    uploads.mkdir()
    _point_at(monkeypatch, uploads)
    archive = _tarball(tmp_path / "homelab.tar.gz", {"alpha.md": "# Alpha\nx\n"})
    temp_upload = uploads / ".homelab.0.upload"
    temp_unpack = uploads / ".homelab.0.unpack"
    temp_upload.write_bytes(archive.read_bytes())
    started = threading.Event()
    release = threading.Event()
    monkeypatch.setattr(git_sources_api, "unpack_archive", _GatedUnpack(started, release))

    # The handler holds the flag when it creates the task — simulate.
    git_sources_api._upload_in_progress = True
    errors: list[BaseException] = []

    def _run() -> None:
        try:
            asyncio.run(
                git_sources_api._run_upload(
                    "homelab", "homelab.tar.gz", len(archive.read_bytes()),
                    uploads, temp_upload, temp_unpack,
                )
            )
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    try:
        assert started.wait(15.0), "background run did not reach its unpack"
        # The run is in flight: the flag (not the task's done-ness) is
        # held — the second upload is 409.
        with _second_client() as second:
            r = _post(second, "homelab.tar.gz", archive.read_bytes())
        assert r.status_code == 409
        assert r.json() == {"detail": "an upload is already in progress"}
        release.set()
        thread.join(20)
    finally:
        release.set()
    assert not thread.is_alive()
    assert errors == [], f"the run raised: {errors!r}"
    terminal = git_sources_api._upload_status
    assert terminal.state == "success", terminal
    assert terminal.detail == {"message": "uploaded"}
    # The folder is in place, no temp survived, the flag is released.
    assert [p.name for p in uploads.iterdir()] == ["homelab"]
    assert git_sources_api._upload_in_progress is False

    # The flag was released: the next upload goes through for real
    # (a full HTTP run — 202 + the terminal status).
    with TestClient(fastapi_app) as third:
        assert third.post("/api/login", json={"password": ADMIN_PASSWORD}).status_code == 204
        r = _post(third, "homelab.tar.gz", archive.read_bytes())
        assert r.status_code == 202, r.text
        assert _wait_status(third)["state"] == "success"


def test_upload_refused_while_flag_held(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """The smallest test seam: the module-level flag itself. Held →
    409 and nothing happens (no dir, no row) — even though the status
    is still ``idle`` (no run exists at all): the flag, not a
    task/state check, is the gate. That is what the receive window
    needs — the handler awaits the (1 MiB-chunk) streaming receive
    BEFORE the background task exists, so only the flag can stop a
    concurrent POST from starting a second run."""
    _point_at(monkeypatch, tmp_path / "uploads")
    monkeypatch.setattr(git_sources_api, "_upload_in_progress", True)

    r = _post(upload_client, "a.tar.gz", b"x")
    assert r.status_code == 409
    assert r.json() == {"detail": "an upload is already in progress"}
    # The flag — not the status state — is the gate:
    assert upload_client.get("/api/git-sources/upload/status").json()["state"] == "idle"
    assert _count_rows(db) == 0
    assert not (tmp_path / "uploads").exists()


# --- the 202 lands with the archive on disk (phase 64 A2) --------------------


def test_202_is_followed_immediately_by_the_terminal_status(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """The 202 is the "archive safely on disk" moment (phase 64 A2) and
    never carries the outcome (the 202 body is the ``UploadAccepted``
    name only); the outcome lives in the status, which the UI polls.

    Phase 90: the run is a fast, synchronous-only pipeline (unpack →
    swap → upsert, no await) queued on the event loop BEFORE the 202's
    response is delivered (the TestClient portal wakes the main thread
    with a loop callback that runs AFTER the run's queued step), so
    the flip side of the deferred scan is deterministic at the handler
    level: by the time the 202 is observed, the run has ALREADY landed
    its terminal state — no mid-flight ``running`` window to race.
    (The in-flight ``running`` shape itself is pinned at the runner
    level in ``test_background_run_starts_with_the_archive_on_disk``
    and ``test_mid_run_status_carries_no_file_progress``.)"""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    r = _post(upload_client, "homelab.tar.gz", _targz_bytes({"alpha.md": "# Alpha\nx\n"}))
    assert r.status_code == 202, r.text
    # The 202 body names the safe source — and nothing else (no counts,
    # no state): the outcome is the status's, not the response's.
    assert r.json() == {"detail": "upload received", "name": "homelab"}
    # The very next status read is already terminal (the run finished
    # before the 202 was even delivered).
    s = upload_client.get("/api/git-sources/upload/status").json()
    assert s["state"] == "success", s
    assert s["detail"] == {"message": "uploaded"}
    assert s["error"] is None
    assert s["current_file"] is None
    assert s["files_done"] == 0 and s["files_total"] == 0
    assert s["started_at"] is not None and s["finished_at"] is not None
    # The folder is in place, one row, no dotfile temp, KB untouched.
    assert (uploads / "homelab").is_dir()
    assert [p.name for p in uploads.iterdir()] == ["homelab"]
    assert _count_rows(db) == 1
    _kb_untouched(db, upload_client)


def test_background_run_starts_with_the_archive_on_disk(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """A2 pinned at the runner level: the background run receives the
    archive SAFELY ON DISK — parked inside unpack (before the real
    unpack and the post-unpack unlink), the dotfile temp upload file
    still exists, and the module-level status (what
    ``GET /upload/status`` reads) is ``running`` with no progress yet
    (phase 90 A2: the upload has no file-level progress at all). After
    release the run completes for real (real unpack + swap) and no
    dotfile temp survives.

    ``_run_upload`` is driven directly on a worker loop (the house
    background-task pattern from ``test_sync_button.py``) — the park is
    a synchronous block, which would also stall the 202's response
    delivery through HTTP, so the runner-level observation is the
    deterministic one."""
    uploads = tmp_path / "uploads"
    uploads.mkdir()
    _point_at(monkeypatch, uploads)

    class _DummySession:
        def scalar(self, statement: object) -> None:
            return None

        def add(self, row: object) -> None:
            pass

        def commit(self) -> None:
            pass

        def close(self) -> None:
            pass

    monkeypatch.setattr(git_sources_api, "SessionLocal", _DummySession)

    # A real, valid archive in the temp upload file — exactly what the
    # handler leaves behind before the 202.
    real_archive = _tarball(tmp_path / "homelab.tar.gz", {"alpha.md": "# Alpha\nx\n"})
    temp_upload = uploads / ".homelab.0.upload"
    temp_unpack = uploads / ".homelab.0.unpack"
    temp_upload.write_bytes(real_archive.read_bytes())

    # The handler holds the flag when it creates the task — simulate.
    git_sources_api._upload_in_progress = True
    started = threading.Event()
    release = threading.Event()
    monkeypatch.setattr(git_sources_api, "unpack_archive", _GatedUnpack(started, release))

    errors: list[BaseException] = []

    def _run() -> None:
        try:
            asyncio.run(
                git_sources_api._run_upload(
                    "homelab",
                    "homelab.tar.gz",
                    len(real_archive.read_bytes()),
                    uploads,
                    temp_upload,
                    temp_unpack,
                )
            )
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    try:
        assert started.wait(15.0), "the run never reached its unpack"
        # The archive is on disk while the run is in flight (A2): the
        # temp upload file exists — the run is parked before the real
        # unpack and the post-unpack unlink.
        assert temp_upload.is_file()
        # The module-level status (what GET /upload/status reads) is
        # running, with no progress (phase 90 A2).
        status = git_sources_api._upload_status
        assert status.state == "running"
        assert status.current_file is None
        assert status.files_done == 0 and status.files_total == 0
        assert status.detail == {} and status.error is None
        assert status.started_at is not None and status.finished_at is None
        release.set()
        thread.join(20)
    finally:
        release.set()
    assert not thread.is_alive()
    assert errors == [], f"the run raised: {errors!r}"
    terminal = git_sources_api._upload_status
    assert terminal.state == "success"
    assert terminal.detail == {"message": "uploaded"}
    assert terminal.current_file is None
    assert terminal.files_done == 0 and terminal.files_total == 0
    # The folder is in place and no dotfile temp survived the run.
    assert [p.name for p in uploads.iterdir()] == ["homelab"]
    assert git_sources_api._upload_in_progress is False


# --- mid-run status: no file progress (phase 90 A2) ---------------------------


@pytest.mark.parametrize(
    "swap_error",
    [
        pytest.param(None, id="success-terminal"),
        pytest.param(
            OSError(
                "rename of https://u:p@aipi.example.com/x "
                "failed: connection refused"
            ),
            id="failed-terminal",
        ),
    ],
)
def test_mid_run_status_carries_no_file_progress(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    db: Session,
    swap_error: BaseException | None,
) -> None:
    """Mid-run: the status carries NO file progress — ``current_file``
    null, ``files_done`` / ``files_total`` 0/0 — for the WHOLE upload
    run (phase 90 A2: the live file label belongs to the sync); the
    endpoint (observed from a second client on its own loop while the
    run is parked in its synchronous unpack gate) reports ``running``
    with empty ``detail`` / null ``error``. Terminal states keep 0/0:
    ``success`` carries the no-count ``{"message": "uploaded"}``
    payload; ``failed`` carries the sanitized error (credentials
    masked).

    ``_run_upload`` is driven directly on a worker loop (the house
    background-task pattern from ``test_sync_button.py``) — the park is
    a synchronous block, so the endpoint observation must come from a
    client with its own loop.
    """
    uploads = tmp_path / "uploads"
    uploads.mkdir()
    _point_at(monkeypatch, uploads)
    archive = _tarball(tmp_path / "homelab.tar.gz", {"alpha.md": "# Alpha\nx\n"})
    temp_upload = uploads / ".homelab.0.upload"
    temp_unpack = uploads / ".homelab.0.unpack"
    temp_upload.write_bytes(archive.read_bytes())
    started = threading.Event()
    release = threading.Event()
    monkeypatch.setattr(git_sources_api, "unpack_archive", _GatedUnpack(started, release))
    if swap_error is not None:
        def failing_swap(new_dir: Path, final_dir: Path) -> None:
            raise swap_error

        monkeypatch.setattr(git_sources_api, "swap_in", failing_swap)

    # The handler holds the flag when it creates the task — simulate.
    git_sources_api._upload_in_progress = True
    errors: list[BaseException] = []

    def _run() -> None:
        try:
            asyncio.run(
                git_sources_api._run_upload(
                    "homelab", "homelab.tar.gz", len(archive.read_bytes()),
                    uploads, temp_upload, temp_unpack,
                )
            )
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    try:
        assert started.wait(15.0), "the run never reached its unpack"
        # While the run is parked: no file progress (phase 90 A2) — the
        # keys exist in the key set but are null/0/0 — and the endpoint
        # (a second client: the run's loop is blocked in the park)
        # reports running with no detail/error yet.
        with _second_client() as second:
            s = second.get("/api/git-sources/upload/status").json()
        assert s["state"] == "running"
        assert s["current_file"] is None
        assert s["files_done"] == 0 and s["files_total"] == 0
        assert s["error"] is None and s["detail"] == {}
        assert s["started_at"] is not None and s["finished_at"] is None
        release.set()
        thread.join(20)
    finally:
        release.set()
    assert not thread.is_alive()
    assert errors == [], f"the run raised: {errors!r}"

    # Terminal (read from the same object the endpoint serializes):
    # still no file progress; the state carries the payload.
    terminal = git_sources_api._upload_status
    assert terminal.current_file is None
    assert terminal.files_done == 0 and terminal.files_total == 0
    assert terminal.finished_at is not None
    if swap_error is None:
        assert terminal.state == "success", terminal
        assert terminal.error is None
        assert terminal.detail == {"message": "uploaded"}
        # The endpoint agrees (same object, same key set).
        with _second_client() as second:
            s = second.get("/api/git-sources/upload/status").json()
        assert s["state"] == "success" and s["detail"] == {"message": "uploaded"}
        _kb_untouched(db, _admin_client())
    else:
        assert terminal.state == "failed", terminal
        assert terminal.detail == {}
        error = terminal.error or ""
        assert "*****@aipi.example.com" in error  # credentials masked
        assert "u:p" not in error
        assert "connection refused" in error  # the reason survives
        # A5: a pre-swap failure leaves no folder, no row, no KB state.
        assert not (uploads / "homelab").exists()
        assert _count_rows(db) == 0
        _kb_untouched(db, _admin_client())


# --- cancellation: the finally cleans temps + releases the flag --------------


def test_cancelled_run_cleans_temps_and_releases_the_flag(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """App shutdown cancels the background task: ``CancelledError`` is
    deliberately NOT caught (the ``_run_sync`` rule — swallowing it
    would mask a real stop), and the ``finally`` still runs: both
    temps are gone and the flag is released, so nothing lingers after
    the stop.

    Phase 90: the run is a synchronous pipeline (unpack → swap → row
    upsert — no await), so a cancel that lands mid-run is delivered
    when the run completes (CPython marks a task cancelled when a
    cancel was requested before completion) — ``await task`` still
    raises ``CancelledError``, exactly as before. ``_run_upload`` is
    driven directly on a worker loop (the house background-task
    pattern from ``test_sync_button.py``); the park is a synchronous
    gate in ``swap_in``, and the cancel comes from the test's own
    thread (``Task.cancel`` only sets a flag — safe across threads
    while the task is blocked in synchronous code)."""
    uploads = tmp_path / "uploads"
    uploads.mkdir()
    _point_at(monkeypatch, uploads)

    monkeypatch.setattr(git_sources_api, "unpack_archive", lambda a, t, cap: None)
    started = threading.Event()
    release = threading.Event()

    def gated_swap(new_dir: Path, final_dir: Path) -> None:
        # The no-op swap: the park itself is the seam (the real swap
        # would finish the run before the cancel could land).
        started.set()
        release.wait(30.0)

    monkeypatch.setattr(git_sources_api, "swap_in", gated_swap)

    class _DummySession:
        def scalar(self, statement: object) -> None:
            return None

        def add(self, row: object) -> None:
            pass

        def commit(self) -> None:
            pass

        def close(self) -> None:
            pass

    monkeypatch.setattr(git_sources_api, "SessionLocal", _DummySession)

    # The handler holds the flag when it creates the task — simulate.
    git_sources_api._upload_in_progress = True
    temp_upload = uploads / ".homelab.0.upload"
    temp_unpack = uploads / ".homelab.0.unpack"
    temp_upload.write_bytes(b"compressed bytes")
    temp_unpack.mkdir()
    (temp_unpack / "alpha.md").write_text("# A\n", encoding="utf-8")

    task_holder: list[asyncio.Task[None]] = []
    errors: list[BaseException] = []

    def _run() -> None:
        async def _main() -> None:
            task = asyncio.create_task(
                git_sources_api._run_upload(
                    "homelab", "homelab.tar.gz", 4, uploads, temp_upload, temp_unpack
                )
            )
            task_holder.append(task)
            await task

        try:
            asyncio.run(_main())
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    try:
        for _ in range(200):
            if task_holder:
                break
            time.sleep(0.01)
        assert task_holder, "the run task was never created"
        assert started.wait(15.0), "the run never reached its swap"
        # The app-shutdown shape: cancel the in-flight task. The run is
        # blocked in synchronous code, so the cancel is delivered when
        # it completes — still delivered (not swallowed).
        task_holder[0].cancel()
        # Free the gated swap AFTER the cancel: the run finishes, the
        # finally cleans up, the cancel lands on the completed task.
        release.set()
        thread.join(20)
    finally:
        release.set()  # backstop — the run may already be past the gate
    assert not thread.is_alive()
    # The CancelledError propagated (not swallowed) — and the finally
    # did its cleanup: no temp survives, the flag is released.
    assert len(errors) == 1
    assert isinstance(errors[0], asyncio.CancelledError)
    assert not temp_upload.exists()
    assert not temp_unpack.exists()
    assert git_sources_api._upload_in_progress is False


# --- in-flight run must not block a concurrent TRUNCATE (regression, --------
# --- phase 49 task 03; now structural in phases 64/90) -----------------------


def _admin_client() -> TestClient:
    """A one-shot admin client (no portal lifetime to manage) for
    terminal-state observations after a direct-drive run."""
    client = TestClient(fastapi_app)
    r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
    return client


def test_in_flight_upload_does_not_block_a_concurrent_truncate(
    monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """While the background run is in flight, a concurrent TRUNCATE of
    the KB + registry tables (the E2E isolation-fixture pattern) must
    complete.

    Regression for the phase-49 task-03 deadlock, now structural
    (phases 64/90): the handler no longer touches the DB at all, the
    run's row upsert commits in its own short-lived session (closed
    before the run ends), and the run holds NO session across a scan —
    the scan itself moved to the sync button (phase 90). This pins it:
    with the run held in flight (driven directly on a worker loop,
    parked inside its swap — the synchronous-seam contract), the
    TRUNCATE completes well inside its 5 s ``lock_timeout`` (without
    the fix it times out with a lock-not-available error)."""
    uploads = tmp_path / "uploads"
    uploads.mkdir()
    _point_at(monkeypatch, uploads)
    archive = _tarball(tmp_path / "homelab.tar.gz", {"alpha.md": "# Alpha\nx\n"})
    temp_upload = uploads / ".homelab.0.upload"
    temp_unpack = uploads / ".homelab.0.unpack"
    temp_upload.write_bytes(archive.read_bytes())
    started = threading.Event()
    release = threading.Event()
    monkeypatch.setattr(git_sources_api, "swap_in", _GatedSwap(started, release))

    # The handler holds the flag when it creates the task — simulate.
    git_sources_api._upload_in_progress = True
    errors: list[BaseException] = []

    def _run() -> None:
        try:
            asyncio.run(
                git_sources_api._run_upload(
                    "homelab", "homelab.tar.gz", len(archive.read_bytes()),
                    uploads, temp_upload, temp_unpack,
                )
            )
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    try:
        assert started.wait(15.0), "background run did not reach its swap"
        # The E2E isolation TRUNCATE, exactly as the story fixtures run
        # it — must complete while the run is held in flight.
        with SessionLocal() as tr, tr.begin():
            tr.execute(text("SET LOCAL lock_timeout = '5s'"))
            tr.execute(
                text("TRUNCATE chunks, documents, query_log, kb_overview, git_sources")
            )
        release.set()
        thread.join(20)
    finally:
        release.set()
    assert not thread.is_alive()
    assert errors == [], f"the run raised: {errors!r}"
    terminal = git_sources_api._upload_status
    assert terminal.state == "success", terminal
    assert terminal.detail == {"message": "uploaded"}
    # The upsert's short-lived session is closed by now: one row,
    # no stray temp.
    assert _count_rows(db) == 1
    assert [p.name for p in uploads.iterdir()] == ["homelab"]


# --- streaming cap (413) ----------------------------------------------------


def test_oversized_upload_gets_413_and_leaves_no_temp(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads, upload_max_mb=1)

    # 2 MiB of incompressible bytes with a 1 MiB cap → 413 naming the
    # cap (inline — the run never starts).
    big = os.urandom(2 * 1024 * 1024)
    r = _post(upload_client, "big.zip", big)
    assert r.status_code == 413
    assert "1 MiB" in r.json()["detail"]
    assert _count_rows(db) == 0
    assert uploads.exists()  # the dir was created before streaming
    assert list(uploads.iterdir()) == []  # the temp .upload file is gone

    # The inline exception path released the flag: the next upload is
    # NOT 409. Boundary: EXACTLY the cap is not 413 (the check is
    # strictly >) — the stream completes, 202 lands, and the garbage
    # bytes fail at the BACKGROUND unpack instead (a status state,
    # phase 64 A5 — never an HTTP error post-202).
    r = _post(upload_client, "exact.zip", os.urandom(1024 * 1024))
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert "could not unpack the archive" in (status["error"] or "")
    assert list(uploads.iterdir()) == []


# --- unpack safety: failed runs leave the previous state intact ---------------


def test_zip_slip_archive_fails_the_run_and_previous_state_is_intact(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "alpha.md": "# Alpha\noriginal sentinel one\n",
        "bravo.md": "# Bravo\noriginal sentinel two\n",
    }
    _good_upload(upload_client, "safe", files)

    evil = _zip(tmp_path / "safe.zip", {"../evil.txt": "pwned"})
    r = _post(upload_client, "safe.zip", evil.read_bytes())
    assert r.status_code == 202, r.text  # on disk → the failure is a status state
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert "traversal" in (status["error"] or "")
    assert not (tmp_path / "evil.txt").exists()  # the escape never landed

    _assert_previous_intact(upload_client, db, uploads, "safe", files)


def test_tar_symlink_escape_fails_the_run_and_previous_state_is_intact(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "alpha.md": "# Alpha\noriginal sentinel one\n",
        "bravo.md": "# Bravo\noriginal sentinel two\n",
    }
    _good_upload(upload_client, "safe", files)

    evil = tmp_path / "safe.tar"
    with tarfile.open(evil, "w") as tf:
        info = tarfile.TarInfo("link")
        info.type = tarfile.SYMTYPE
        info.linkname = "/etc/passwd"
        tf.addfile(info)
    r = _post(upload_client, "safe.tar", evil.read_bytes())
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert "escape" in (status["error"] or "")

    _assert_previous_intact(upload_client, db, uploads, "safe", files)


def test_corrupt_archive_fails_the_run_and_previous_state_is_intact(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "alpha.md": "# Alpha\noriginal sentinel one\n",
        "bravo.md": "# Bravo\noriginal sentinel two\n",
    }
    _good_upload(upload_client, "safe", files)

    # A truncated zip (the EOCD is cut off) is not a zip and not a tar.
    good_zip = _zip(tmp_path / "good.zip", {"alpha.md": "# Alpha\nx\n"})
    good_bytes = good_zip.read_bytes()
    truncated = good_bytes[: len(good_bytes) // 2]
    r = _post(upload_client, "safe.zip", truncated)
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert "could not unpack the archive" in (status["error"] or "")

    # A zero-byte "archive" fails the same way.
    r = _post(upload_client, "safe.tar", b"")
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert "could not unpack the archive" in (status["error"] or "")

    _assert_previous_intact(upload_client, db, uploads, "safe", files)


def test_swap_failure_fails_the_run_and_previous_state_is_intact(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """A rename failure in ``swap_in`` (OS error) → status ``failed``
    with its message; the previous folder/row are untouched and no
    temp survives (the run's finally cleans the temp sibling). A
    failure here leaves the previous folder/row untouched (phase 64
    A5); the KB was never touched by either upload (phase 90)."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "alpha.md": "# Alpha\noriginal sentinel one\n",
        "bravo.md": "# Bravo\noriginal sentinel two\n",
    }
    _good_upload(upload_client, "safe", files)

    def failing_swap(new_dir: Path, final_dir: Path) -> None:
        raise ArchiveUploadError("could not replace the previous folder")

    monkeypatch.setattr(git_sources_api, "swap_in", failing_swap)
    r = _post(upload_client, "safe.tar.gz", _targz_bytes({"alpha.md": "# A\nx\n"}))
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert status["error"] == "could not replace the previous folder"

    _assert_previous_intact(upload_client, db, uploads, "safe", files)


def test_zero_entry_archive_fails_the_run(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """A completely empty archive (zero entries) → status ``failed``
    ``the archive contains no files`` — both containers — with no
    folder, row, or temp file left behind (a pre-swap failure leaves
    the KB/folders/rows untouched, phase 64 A5)."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)

    empty_targz = tmp_path / "empty.tar.gz"
    with tarfile.open(empty_targz, "w:gz"):
        pass
    r = _post(upload_client, "empty.tar.gz", empty_targz.read_bytes())
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert status["error"] == "the archive contains no files"

    empty_zip = tmp_path / "empty.zip"
    with zipfile.ZipFile(empty_zip, "w"):
        pass
    r = _post(upload_client, "empty.zip", empty_zip.read_bytes())
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert status["error"] == "the archive contains no files"

    assert _count_rows(db) == 0
    assert not (uploads / "empty").exists()
    assert list(uploads.iterdir()) == []


def test_concurrent_row_insert_fails_the_run_with_the_path_named(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """The upsert's unique-index backstop: the pre-check sees no row,
    but the commit hits the unique constraint (a concurrent insert the
    pre-check missed) → status ``failed`` naming the path. The folder
    stays — the row exists (the concurrent one) and the next
    sync/re-upload sees it (phase 64 A5); no temp survives and the
    short-lived session was closed."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)

    class _ConcurrentInsertSession:
        """The short-lived upsert session, faked: the pre-check misses
        (no row), the commit hits the unique index."""

        def __init__(self) -> None:
            self.closed = False
            self.added: list[object] = []

        def scalar(self, statement: object) -> None:
            return None

        def add(self, row: object) -> None:
            self.added.append(row)

        def commit(self) -> None:
            raise IntegrityError("INSERT INTO git_sources", {}, Exception("duplicate key"))

        def rollback(self) -> None:
            pass

        def close(self) -> None:
            self.closed = True

    sessions: list[_ConcurrentInsertSession] = []

    def _fake_session_factory() -> _ConcurrentInsertSession:
        s = _ConcurrentInsertSession()
        sessions.append(s)
        return s

    monkeypatch.setattr(git_sources_api, "SessionLocal", _fake_session_factory)
    r = _post(
        upload_client,
        "homelab.tar.gz",
        _tarball(tmp_path / "h.tar.gz", {"alpha.md": "# Alpha\nx\n"}).read_bytes(),
    )
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "failed"
    assert status["error"] == (
        f"a local source with this path already exists: {uploads / 'homelab'}"
    )
    # The folder stays (the row exists — the concurrent insert — and
    # the next sync sees it); no temp survived; the session was closed.
    assert (uploads / "homelab").is_dir()
    assert [p.name for p in uploads.iterdir()] == ["homelab"]
    assert len(sessions) == 1
    assert sessions[0].added, "the run attempted the insert (the pre-check missed)"
    assert sessions[0].closed


# --- happy path -------------------------------------------------------------


def test_happy_path_tar_gz(
    upload_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {"alpha.md": "# Alpha\nfirst sentinel\n", "bravo.md": "# Bravo\nsecond sentinel\n"}
    payload = _tarball(tmp_path / "homelab.tar.gz", files).read_bytes()

    with caplog.at_level(logging.INFO, logger="app.api.git_sources"):
        r = _post(upload_client, "homelab.tar.gz", payload)
        assert r.status_code == 202, r.text
        # 202 = the archive is on disk (phase 64 A2) — the safe name
        # rides back in the ``UploadAccepted`` body (the toast moment).
        assert r.json() == {"detail": "upload received", "name": "homelab"}
        status = _wait_status(upload_client)

    assert status["state"] == "success", status
    assert status["error"] is None
    # Phase 90 A2: the success detail is the no-count "uploaded"
    # payload — no import counts ride the upload anymore (the scan
    # lives in the sync's status).
    assert status["detail"] == {"message": "uploaded"}
    # No file progress, ever: current_file null, counts 0/0.
    assert status["current_file"] is None
    assert status["files_done"] == 0 and status["files_total"] == 0
    assert status["started_at"] is not None and status["finished_at"] is not None

    # Unpacked under the tmp upload dir, dotfile temps cleaned up.
    folder = uploads / "homelab"
    assert folder.is_dir()
    assert {p.name: p.read_text(encoding="utf-8") for p in folder.iterdir()} == files
    assert [p.name for p in uploads.iterdir()] == ["homelab"]

    # One kind=local row; the NOT-NULL url column carries the path.
    row = _row(db, str(folder))
    assert row is not None
    assert row.kind == "local"
    assert row.url == str(folder)
    assert row.path == str(folder)
    assert _count_rows(db) == 1

    # Phase 90: the upload touches NO KB state — zero documents, zero
    # chunks, no overview row — and the scan machinery is gone from
    # the module entirely (the sync button owns it).
    _kb_untouched(db, upload_client)
    for gone in ("import_sources", "check_models"):
        assert not hasattr(git_sources_api, gone), f"{gone} must not be imported"

    # The per-upload log line (PLAN §9 / AGENTS.md rule 10) — unpack +
    # register only: no file counts, the state rides the line.
    lines = [
        rec.getMessage()
        for rec in caplog.records
        if rec.getMessage().startswith("upload: finished")
    ]
    assert len(lines) == 1, lines
    match = re.match(
        r"^upload: finished name=homelab file=homelab\.tar\.gz bytes=(\d+) "
        r"total_ms=\d+ state=success$",
        lines[0],
    )
    assert match, lines[0]
    assert int(match.group(1)) == len(payload)  # the compressed bytes in


@pytest.mark.parametrize(
    ("filename", "builder"),
    [
        ("notes.zip", "zip"),
        ("plain.tar", "tar"),
        ("tgz.tgz", "targz"),
    ],
)
def test_happy_path_other_accepted_formats(
    upload_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    db: Session,
    tmp_path: Path,
    filename: str,
    builder: str,
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {"alpha.md": "# Alpha\nsentinel\n"}
    make = {"zip": _zip, "tar": lambda p, f: _tarball(p, f, "w"), "targz": _tarball}[builder]
    payload = make(tmp_path / filename, files).read_bytes()

    r = _post(upload_client, filename, payload)
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}
    expected_name = {"notes.zip": "notes", "plain.tar": "plain", "tgz.tgz": "tgz"}[filename]
    assert r.json()["name"] == expected_name  # the safe name in the 202 body
    # The folder is unpacked in full, one row, and the KB is untouched
    # (phase 90 — the sync decides what gets indexed).
    folder = uploads / expected_name
    assert {p.name: p.read_text(encoding="utf-8") for p in folder.iterdir()} == files
    assert _count_rows(db) == 1
    _kb_untouched(db, upload_client)


# --- re-upload, same name: in-place replace ---------------------------------


def test_reupload_same_name_replaces_in_place(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    v1 = {
        "alpha.md": "# Alpha\nv1 content\n",
        "bravo.md": "# Bravo\nv1 content\n",
        "charlie.md": "# Charlie\nv1 content\n",
    }
    r = _post(upload_client, "homelab.tar.gz", _tarball(tmp_path / "v1.tar.gz", v1).read_bytes())
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}
    row_before = _row(db, str(uploads / "homelab"))
    assert row_before is not None
    added_at_before = row_before.added_at

    # v2: the in-place-replace scenario, now observed via the SECOND
    # run's status and the folder on disk (the upload no longer scans —
    # phase 90).
    v2 = {"alpha.md": "# Alpha\nv2 CHANGED content\n", "delta.md": "# Delta\nbrand new\n"}
    r = _post(upload_client, "homelab.tar.gz", _tarball(tmp_path / "v2.tar.gz", v2).read_bytes())
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}

    # Exactly ONE row for the path, and its added_at survived (the
    # second run's upsert left the existing row alone).
    assert _count_rows(db) == 1
    row_after = _row(db, str(uploads / "homelab"))
    assert row_after is not None
    assert row_after.id == row_before.id
    assert row_after.added_at == added_at_before

    # The folder holds ONLY the new archive's files (full replacement —
    # no stale files from v1), and the KB stays untouched (phase 90:
    # what the index does with them is the sync's call).
    folder = uploads / "homelab"
    assert {p.name: p.read_text(encoding="utf-8") for p in folder.iterdir()} == v2
    assert [p.name for p in uploads.iterdir()] == ["homelab"]
    _kb_untouched(db, upload_client)


def test_non_a9_archive_is_a_valid_replacement(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """An archive with only non-importable files is a VALID
    replacement (phase 90, A1): the swap happens, the row registers —
    and the KB stays untouched (what the index does with the files is
    the sync's call, not the upload's)."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)

    r = _post(
        upload_client,
        "notes.tar.gz",
        _tarball(tmp_path / "a.tar.gz", {"readme.md": "# Readme\nv1\n"}).read_bytes(),
    )
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}

    r = _post(
        upload_client,
        "notes.tar.gz",
        _tarball(tmp_path / "b.tar.gz", {"binary.bin": "not an importable format"}).read_bytes(),
    )
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status  # still a valid replacement
    assert status["detail"] == {"message": "uploaded"}

    assert {p.name for p in (uploads / "notes").iterdir()} == {"binary.bin"}
    assert _count_rows(db) == 1
    _kb_untouched(db, upload_client)


# --- phase 89: re-uploads preserve the row's saved ignore list ---------------


def test_reupload_preserves_saved_ignore_list(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase 89/90: a re-upload of an EXISTING source leaves the row —
    and its saved ignore list — exactly in place (the upsert is
    leave-as-is for existing rows). Phase 90: the upload no longer
    scans, so the list's effect on the KB is the SYNC's (its suites
    prove the honored-ignore import; the phase-90 E2E proves the full
    upload → edit list → sync loop) — the upload's contract is
    preservation + every file kept on disk: the ignore is about the
    index, not the folder."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "keep.md": "# Keep\nin scope\n",
        "ignore/secret.md": "# Secret\nignored\n",
    }
    r = _post(upload_client, "docs.tar.gz", _targz_bytes(files))
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}
    # The fresh row has no list yet — and nothing is indexed (phase 90).
    folder = uploads / "docs"
    row = _row(db, str(folder))
    assert row is not None
    assert (row.ignore_paths or []) == []
    _kb_untouched(db, upload_client)

    # Save the ignore list on the row via the API (phase 89, task 03).
    r = upload_client.patch(
        f"/api/git-sources/{row.id}", json={"ignore_paths": ["ignore/"]}
    )
    assert r.status_code == 200, r.text
    assert r.json()["ignore_paths"] == ["ignore"]  # stored normalized

    # Re-upload the SAME archive (same name → in-place replace).
    r = _post(upload_client, "docs.tar.gz", _targz_bytes(files))
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}

    # The re-upload left the row's list in place (the existing row is
    # untouched by the upsert), and the folder keeps every file.
    row_after = _row(db, str(folder))
    assert row_after is not None
    assert row_after.ignore_paths == ["ignore"]
    assert {p.name for p in folder.iterdir()} == {"keep.md", "ignore"}
    assert _count_rows(db) == 1
    _kb_untouched(db, upload_client)


def test_upload_new_source_registers_row_with_empty_ignore_list(
    upload_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
) -> None:
    """Phase 89/90 regression: an upload of a NEW source name (no row
    yet, hence no ignore list) registers the row with the
    server-default empty list and keeps every file on disk — including
    nested files — exactly as before; nothing is indexed until the
    owner presses "Sync sources" (phase 90)."""
    uploads = tmp_path / "uploads"
    _point_at(monkeypatch, uploads)
    files = {
        "alpha.md": "# Alpha\nroot file\n",
        "sub/deep.md": "# Deep\nnested file\n",
    }
    r = _post(upload_client, "fresh.tar.gz", _targz_bytes(files))
    assert r.status_code == 202, r.text
    status = _wait_status(upload_client)
    assert status["state"] == "success", status
    assert status["detail"] == {"message": "uploaded"}
    # The fresh row carries the server-default empty list: nothing was
    # ignored.
    folder = uploads / "fresh"
    row = _row(db, str(folder))
    assert row is not None
    assert (row.ignore_paths or []) == []
    # Every file (incl. nested) is on disk — the KB is untouched.
    assert (
        {p.relative_to(folder).as_posix() for p in folder.rglob("*") if p.is_file()} == set(files)
    )
    _kb_untouched(db, upload_client)