"""Phase 183 E2E (Playwright, mock-only): a mass prune leaves nothing behind.

Source: owner incident 2026-09-15 → 2026-10-15 — the brain-of-code sync
pruned **3,008 litellm docs** and the app hit 4 GB of RAM, locking up a
3,904 MB VM (no swap, sharing RAM with a 3.1 GB Postgres) hard enough that
boot −1 has no shutdown record at all. The cause was NOT the delete itself:
``Document.chunks`` carried ``cascade="all, delete-orphan"`` with **no
``passive_deletes``**, so every ``session.delete(doc)`` first SELECTed the
document's whole chunk collection — hydrating every 768-dim
``chunks.embedding`` (~24.8 KiB of Python each vs 3 KiB on disk) purely to
be thrown away, and nothing was released before the single commit
(``app/db.py`` builds sessions with ``expire_on_commit=False``).

The fix under test: ``Document.chunks`` is now ``passive_deletes=True``, so
the rows go **in the database** through ``chunks_document_id_fkey``'s
``ON DELETE CASCADE`` (alembic 0001 L54) and Python never sees them. That
is a *display-of-work* change with exactly one owner-visible contract: the
prune must still remove **every chunk row of every pruned document** — zero
orphans — while changing nothing else (the counts line, the KB tree, the
surviving RAG). This module pins that contract end to end, through the
pipeline the owner actually drives (the Sources page Sync button) and
through the admin's total source removal.

Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_prune_cascade_cleanup.py -v --no-cov

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is skipped — the gate is the deterministic
prune count + the exact orphan-row query, and test 2's grounding needs the
mock-calibrated relevance thresholds (the conftest convention).

Per-module app env (the ``test_local_directory_sources.py`` /
``test_force_push_resync.py`` module-app pattern): this module's app boots
on its OWN port (the conftest session app owns 8123 in a combined run; a
same-port second uvicorn dies on bind and would drive the wrong server),
against the mock LLM, with a dedicated ``BOR_SOURCES_DIR`` /
``BOR_PDF_PAGE_DIR`` under ``tmp_path_factory`` and ``BOR_GIT_SOURCES``
forced empty (the owner's local gitignored ``.env`` must never leak a real
source into the run — the registry rows this suite creates through the
authenticated API are the only sources). The fixture KB is a plain
(non-git) host directory per test under ``tmp_path`` — the app server runs
on the same host, so the path is visible to it — with the stable basename
:data:`SOURCE_NAME` (``kind='local'`` rows index under
``Path(path).name``, ``app/rag/source_removal.py``) so the SQL assertions
name one source whatever the pytest temp root is.

Isolation: the shared E2E Postgres is truncated BEFORE **and** AFTER every
test (``chunks, documents, query_log, kb_overview, git_sources,
sources_meta, folder_summaries``) — a leftover registry row would flip a
sync onto another suite's source, and a leftover document under this
suite's deterministic source name would survive the prune and falsify the
count assertions.

Test → observable mapping (Playwright Mapping Rule):
1. ``test_deleted_files_prune_without_orphans`` — sync 9 files (the counts
   line reports "9 added"), delete 6 of them upstream, re-sync: the counts
   line reports "0 added · 3 unchanged · 6 pruned", the 6 documents are
   gone from the KB (the tree row is absent, ``GET /api/documents/content``
   404s the pair, ``GET /api/docs`` holds exactly the 3 survivors), the
   pruned documents' **chunk rows are all gone** (the test's own SQL — the
   no-orphan half of the phase-183 D4 contract) and the surviving chunk
   total is exactly the sum of the 3 survivors' chunks.
2. ``test_surviving_kb_still_answers_after_a_partial_prune`` — the same
   pruned state answers a chat question about a SURVIVING document: the
   turn settles with a real answer (no error state, no empty bubble, the
   composer recovered) and the durable ``query_log`` row is NOT deflected
   and records the surviving document — a partial prune never damages the
   RAG.
3. ``test_source_removal_prunes_every_document_of_the_source`` — the admin
   path (``DELETE /api/git-sources/{id}``, ``app/api/git_sources.py``): the
   second mass-delete surface ``passive_deletes=True`` covers. Every
   document AND every chunk row of the removed source is gone (the same
   orphan query), the count matches what was indexed there, the row is
   gone, and the OTHER registered source's documents and chunks are
   untouched.
"""
from __future__ import annotations

import os
import re
import subprocess
import sys
import time
import uuid
from collections.abc import Iterator
from datetime import datetime
from pathlib import Path
from typing import Any

import pytest
from playwright.sync_api import Page, expect
from sqlalchemy import func, select, text
from sqlalchemy.orm import Session

from app.db import SessionLocal
from app.models import Chunk, Document, QueryLog
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)
from e2e.sync_helpers import counts_line

REPO = Path(__file__).resolve().parents[2]

# Phase 79 (task 04, full inventory): the conftest session app owns its
# port in a combined run — this module app binds its own port instead.
# Env-overridable.
APP_PORT = int(os.environ.get("E2E_APP_PORT_PRUNECASCADE", "8199"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

SOURCES_URL = "/sources.html"

#: The fixture directory's basename — the ``source`` the importer records
#: for a ``kind='local'`` row (``Path(path).name``), so it is the same
#: string in the DB whatever ``tmp_path`` this test got.
SOURCE_NAME = "prune-cascade"
KEEPER_NAME = "prune-keeper"

#: The fixture layout: 9 importable files across one folder, 6 of which
#: the test deletes upstream (the incident's shape — a bulk deletion in a
#: synced source — scaled to what a test sync walks in seconds).
KEEP_PATHS = [f"notes/keep-{n}.md" for n in (1, 2, 3)]
GONE_PATHS = [f"notes/gone-{n}.md" for n in range(1, 7)]
ALL_PATHS = KEEP_PATHS + GONE_PATHS

#: The sentinel inside the FIRST survivor — test 2's question is answerable
#: ONLY from that document, so a grounded, non-deflected turn proves the
#: surviving chunk rows are still retrievable after the prune.
SENTINEL_DOC = KEEP_PATHS[0]
SENTINEL = "warden-lighthouse-RESE-PRUNECASCADE-7f31"
QUESTION = "Which keeper tends the warden lighthouse beacon log?"

#: The mock LLM's byte-stable grounded-answer marker (``mock_llm.
#: compose_answer``'s fallback branch) — its presence proves a real,
#: grounded answer streamed (the DEFLECT_MODE branch answers with
#: different, deterministic text).
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"

#: "Synced HH:MM" — the local-time last-result label (sources.js's
#: fmtSyncTime), any hour/minute.
SYNCED_LABEL = re.compile(r"Synced \d{1,2}:\d{2}")


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(scope="module")
def app_server(mock_llm: int, tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
    """The real app under test — per-module env: the mock LLM, the
    mock-calibrated gate (so test 2's grounding is deterministic), an
    empty ``BOR_GIT_SOURCES`` fallback, and scratch
    ``BOR_SOURCES_DIR`` / ``BOR_PDF_PAGE_DIR`` under
    ``tmp_path_factory`` (nothing in this suite clones a repo or renders a
    PDF — the dirs are set so the run never writes into the operator's
    own). The session app is never started in this isolated run, so no
    port clash."""
    if USE_REAL_LLM:
        pytest.skip(
            "mock-only suite — the grounding pin is the mock's byte-stable "
            "answer on the mock-calibrated gate (E2E_REAL_LLM unsupported)"
        )
    scratch = tmp_path_factory.mktemp("bor_prune_cascade")
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = (
        "https://aipi.reeseapps.com/v1"
        if USE_REAL_LLM
        else f"http://127.0.0.1:{mock_llm}/v1"
    )
    # Mock-calibrated gate (the conftest pattern, phase 112/113): the
    # token-overlap mock embeddings score far below the production 0.62,
    # and BOTH floors must stay <= the threshold or the app refuses to
    # boot (app/config.py's typo guards).
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    # Phase 16: admin auth must be set or create_app() refuses to boot.
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    # No env fallback rows: the registry this suite writes through the
    # authenticated API is the only source list, so every sync's counts
    # are exactly this test's doing.
    env["BOR_GIT_SOURCES"] = ""
    env["BOR_SOURCES_DIR"] = str(scratch / "checkouts")
    env["BOR_PDF_PAGE_DIR"] = str(scratch / "pdf-pages")
    proc = subprocess.Popen(
        [
            sys.executable, "-m", "uvicorn", "app.main:app",
            "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning",
        ],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{APP_URL}/api/health")
        yield APP_URL
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="module")
def app_url(app_server: str) -> str:
    return app_server


def _truncate_all() -> None:
    """Fresh registry + KB per test (the E2E isolation pattern,
    ``test_source_removal_cleanup.py`` / ``test_force_push_resync.py``).
    ``sources_meta`` (the KB generation counter) and
    ``folder_summaries`` reset with the KB — the sync's overview/bump is an
    upsert, and a leftover folder summary would ride the very
    ``changed_paths`` feed the prune writes."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, kb_overview, "
                "git_sources, sources_meta, folder_summaries"
            )
        )
        db.commit()


@pytest.fixture(autouse=True)
def _clean(db_ready: None) -> Iterator[None]:
    _truncate_all()
    yield
    _truncate_all()


# ---------------------------------------------------------------------------
# The fixture source
# ---------------------------------------------------------------------------


def _doc_body(path: str) -> str:
    """One small markdown file per fixture path. The FIRST survivor is the
    ONLY file carrying the sentinel's words, so test 2's question resolves
    to exactly one document (and the mock's token-overlap cosine —
    ``sqrt(shared/total)`` — clears the mock-calibrated 0.30 gate only for
    that one)."""
    if path == SENTINEL_DOC:
        return (
            "# Keeper log\n"
            "\n"
            "The keeper tends the warden lighthouse beacon log.\n"
            f"Marker: {SENTINEL}\n"
        )
    return (
        f"# {path}\n"
        "\n"
        f"One unrelated note from the {SOURCE_NAME} fixture source.\n"
    )


def _make_source_dir(root: Path, name: str, paths: list[str]) -> Path:
    """A plain (non-git) directory of importable files — the shape the
    owner's local sources have. Returns the registered root."""
    base = root / name
    for rel in paths:
        target = base / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(_doc_body(rel), encoding="utf-8")
    assert not (base / ".git").exists()
    return base


# ---------------------------------------------------------------------------
# Helpers: the DB side of the contract (the phase-183 D4 assertions)
# ---------------------------------------------------------------------------


def _doc_ids(db: Session, source: str, paths: list[str]) -> list[str]:
    """The indexed document ids of ``(source, path)``, sorted."""
    return sorted(
        str(doc_id)
        for (doc_id,) in db.execute(
            select(Document.id).where(
                Document.source == source, Document.path.in_(paths)
            )
        ).all()
    )


def _chunk_count(db: Session, document_ids: list[str]) -> int:
    """The chunk rows still belonging to ``document_ids`` — THE orphan
    probe. Non-zero means the documents are gone and their embeddings (and
    their FTS rows) stayed behind."""
    if not document_ids:
        return 0
    total = db.scalar(
        select(func.count())
        .select_from(Chunk)
        .where(Chunk.document_id.in_([uuid.UUID(x) for x in document_ids]))
    )
    return int(total or 0)


def _total_chunks(db: Session) -> int:
    return int(db.scalar(select(func.count()).select_from(Chunk)) or 0)


def _chunk_counts_by_doc(
    db: Session, source: str, paths: list[str] | None = None
) -> dict[str, int]:
    """``path`` → chunk count, for the surviving-total assertion."""
    query = (
        select(Document.path, func.count(Chunk.id))
        .outerjoin(Chunk, Chunk.document_id == Document.id)
        .where(Document.source == source)
        .group_by(Document.path)
    )
    if paths is not None:
        query = query.where(Document.path.in_(paths))
    rows = db.execute(query).all()
    return {str(path): int(count) for path, count in rows}


def _sources_in_db(db: Session) -> list[str]:
    return sorted(
        {
            str(name)
            for (name,) in db.execute(
                select(Document.source).distinct()
            ).all()
        }
    )


# ---------------------------------------------------------------------------
# Helpers: the UI / API side
# ---------------------------------------------------------------------------


def _started_at(body: dict[str, Any]) -> float:
    """The status body's ``started_at`` as a wall-clock epoch (the server
    stamps it in UTC, this process runs on the same host — the
    ``test_git_source_branch`` idiom)."""
    started = body.get("started_at")
    if not started:
        return 0.0
    try:
        return datetime.fromisoformat(str(started)).timestamp()
    except ValueError:
        return 0.0


def _sync_from_sources_page(page: Page, app_url: str) -> dict[str, Any]:
    """Click the Sources page **Sync sources** button and wait for the
    run's TERMINAL state (the ``test_force_push_resync._wait_sync_state``
    pattern: the module app is shared across a test's two runs, so a
    terminal is accepted only after a ``running`` tick was seen or its
    ``started_at`` post-dates the click). Settles the UI on success
    (``#sync-label`` → "Synced HH:MM" — the never-stale contract, §7.4)."""
    login(page, app_url, next=SOURCES_URL)
    expect(page.locator("#sign-out-btn")).to_be_visible(timeout=15_000)
    btn = page.locator("#sync-btn")
    expect(btn).to_be_visible()
    clicked_at = time.time()
    btn.click()
    expect(btn).to_be_disabled()
    expect(page.locator("#sync-label")).to_have_text("Syncing…")

    deadline = time.monotonic() + 90.0
    saw_running = False
    body: dict[str, Any] = {}
    while time.monotonic() < deadline:
        r = page.request.get(f"{app_url}/api/sync/status")
        assert r.status == 200, r.text
        body = r.json()
        state = body["state"]
        if state == "running":
            saw_running = True
        elif state in ("success", "failed") and (
            saw_running or _started_at(body) >= clicked_at - 1.0
        ):
            break
        time.sleep(0.2)
    else:
        raise AssertionError(f"sync never reached a terminal state: {body}")

    assert body["state"] == "success", f"sync failed: {body}"
    expect(btn).to_be_enabled(timeout=15_000)
    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=15_000)
    # The counts line is the owner-visible prune report (phase 150 — the
    # single sync text surface, sources.js fmtSyncResult).
    return body


def _expect_counts_line(page: Page, expected: str) -> None:
    """The Sources page's counts line reads exactly ``expected`` —
    "9 added" / "0 added · 3 unchanged · 6 pruned"."""
    expect(page.locator("#sync-progress-text")).to_have_text(counts_line(expected), timeout=15_000)


def _add_local_source(page: Page, app_url: str, path: Path) -> str:
    """Register a local directory through the authenticated API (the
    phase-49 shape — the page's own form for this is gone; the admin
    cookie rides ``page.request``). Returns the row id."""
    r = page.request.post(
        f"{app_url}/api/git-sources",
        data={"kind": "local", "path": str(path)},
    )
    assert r.status == 201, f"expected 201 for {path}: {r.status} {r.text}"
    row_id = str(r.json()["id"])
    assert row_id
    return row_id


def _api_docs(page: Page, app_url: str) -> list[dict[str, Any]]:
    """``GET /api/docs`` as the signed-in page (admin cookie rides along)."""
    r = page.request.get(f"{app_url}/api/docs")
    assert r.status == 200, r.text
    return list(r.json()["documents"])


def _api_doc_pairs(page: Page, app_url: str) -> list[tuple[str, str]]:
    return sorted((d["source"], d["path"]) for d in _api_docs(page, app_url))


def _api_sources(page: Page, app_url: str) -> list[str]:
    """The distinct ``source`` labels the KB holds, read through the
    signed-in API (the same view the owner gets on the Sources page)."""
    return sorted({d["source"] for d in _api_docs(page, app_url)})


def _drill_to_fixture_folder(page: Page) -> None:
    """Open the KB catalogue down to the fixture's ``notes`` folder (the
    drill-down tree is the phase-97 shape, the
    ``test_force_push_resync._expect_fixture_row`` idiom)."""
    crumb = page.locator("#kb-crumb")
    if not crumb.is_hidden():
        page.click("#kb-crumb a.kb-crumb-link")
    page.click(f'#folders-tbody a.folder-link:text-is("{SOURCE_NAME}")')
    page.click('#folders-tbody a.folder-link:text-is("notes")')


# ---------------------------------------------------------------------------
# 1. Upstream deletion + re-sync: the prune is exact and leaves no orphans
# ---------------------------------------------------------------------------


def test_deleted_files_prune_without_orphans(
    page: Page, app_url: str, db_ready: None, tmp_path: Path
) -> None:
    """The incident's own shape, end to end: 9 files sync, 6 are deleted
    upstream, the re-sync prunes exactly those 6 — through the Sources page
    Sync button, with the counts line as the owner-visible report — and the
    pruned documents' chunk rows are gone FROM THE DATABASE (the phase-183
    D4 no-orphan contract, asserted by this test's own SQL)."""
    page.set_default_timeout(30_000)
    source = _make_source_dir(tmp_path, SOURCE_NAME, ALL_PATHS)

    login(page, app_url, next=SOURCES_URL)
    _add_local_source(page, app_url, source)

    # --- sync 1: everything is new ------------------------------------
    body = _sync_from_sources_page(page, app_url)
    assert body["detail"]["added"] == len(ALL_PATHS), body
    assert body["detail"]["pruned"] == 0, body
    _expect_counts_line(page, f"{len(ALL_PATHS)} added")
    assert _api_doc_pairs(page, app_url) == [
        (SOURCE_NAME, path) for path in sorted(ALL_PATHS)
    ]

    # The subject of the prune, captured BEFORE it happens: the 6 doomed
    # document ids + every survivor's chunk count. The orphan query below
    # needs ids that no longer exist by the time it runs.
    with SessionLocal() as db:
        gone_ids = _doc_ids(db, SOURCE_NAME, GONE_PATHS)
        keep_counts = _chunk_counts_by_doc(db, SOURCE_NAME, KEEP_PATHS)
        assert _chunk_count(db, gone_ids) > 0, "the doomed docs must be embedded"
    assert len(gone_ids) == len(GONE_PATHS), "the doomed documents must exist first"
    assert all(keep_counts[p] > 0 for p in KEEP_PATHS), keep_counts
    assert sum(keep_counts.values()) > 0

    # --- the upstream deletion (the test owns the source dir) ----------
    for rel in GONE_PATHS:
        (source / rel).unlink()
    assert not any((source / rel).exists() for rel in GONE_PATHS)

    # --- sync 2: the prune ---------------------------------------------
    body = _sync_from_sources_page(page, app_url)
    assert body["detail"]["pruned"] == len(GONE_PATHS), body
    assert body["detail"]["added"] == 0, body
    assert body["detail"]["updated"] == 0, body
    assert body["detail"]["unchanged"] == len(KEEP_PATHS), body
    _expect_counts_line(
        page,
        f"0 added · {len(KEEP_PATHS)} unchanged · {len(GONE_PATHS)} pruned",
    )

    # The KB no longer serves the deleted files: the catalogue row is gone,
    # the content route 404s the pair, and the total is exactly 3.
    _drill_to_fixture_folder(page)
    expect(page.locator("#docs-tbody tr", has_text=KEEP_PATHS[0])).to_have_count(1)
    expect(page.locator("#docs-tbody tr", has_text=GONE_PATHS[0])).to_have_count(0)
    r = page.request.get(
        f"{app_url}/api/documents/content"
        f"?source={SOURCE_NAME}&path={GONE_PATHS[0]}"
    )
    assert r.status == 404, f"a pruned document must 404, got {r.status}: {r.text}"
    docs = _api_docs(page, app_url)
    assert len(docs) == len(KEEP_PATHS), docs
    assert {d["path"] for d in docs} == set(KEEP_PATHS), docs

    # THE phase-183 assertion: the pruned documents left NO chunk rows
    # behind (the FK cascade, not the ORM, removed them) and the surviving
    # chunk total is exactly the survivors' own chunks — nothing orphaned,
    # nothing collateral-damaged.
    with SessionLocal() as db:
        assert _chunk_count(db, gone_ids) == 0, (
            "pruned documents left orphan chunk rows — the FK cascade did "
            "not run (Document.chunks' passive_deletes without ON DELETE "
            "CASCADE)"
        )
        assert _doc_ids(db, SOURCE_NAME, GONE_PATHS) == []
        after = _chunk_counts_by_doc(db, SOURCE_NAME, KEEP_PATHS)
        assert after == keep_counts, (
            f"the surviving documents' chunks changed across the prune: {after}"
        )
        assert _total_chunks(db) == sum(keep_counts.values())
        assert _sources_in_db(db) == [SOURCE_NAME]


# ---------------------------------------------------------------------------
# 2. The surviving KB still answers after a partial prune
# ---------------------------------------------------------------------------


def test_surviving_kb_still_answers_after_a_partial_prune(
    page: Page, app_url: str, db_ready: None, tmp_path: Path
) -> None:
    """The prune's other half: what stays must still work. With the 6 files
    pruned, the chat answers a question that only the surviving
    :data:`SENTINEL_DOC` can answer — a real streamed answer (no empty
    bubble, no error state, the composer recovered) whose durable
    ``query_log`` row is NOT deflected and names the surviving document."""
    page.set_default_timeout(30_000)
    source = _make_source_dir(tmp_path, SOURCE_NAME, ALL_PATHS)

    login(page, app_url, next=SOURCES_URL)
    _add_local_source(page, app_url, source)
    _sync_from_sources_page(page, app_url)
    for rel in GONE_PATHS:
        (source / rel).unlink()
    body = _sync_from_sources_page(page, app_url)
    assert body["detail"]["pruned"] == len(GONE_PATHS), body

    # The chat turn, through the real shell.
    page.goto(f"{app_url}/")
    expect(page.locator("#message-input")).to_be_visible(timeout=30_000)
    # A healthy KB: the offline banner stays hidden (the prune did not
    # empty or break the index the page reports on).
    expect(page.locator("#kb-banner")).to_be_hidden()
    page.fill("#message-input", QUESTION)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(QUESTION)
    bubble = page.locator(".msg.brain .bubble").last
    # Never an empty answer bubble, and the grounded (not deflected) mock
    # answer streamed.
    expect(bubble).not_to_have_text("", timeout=30_000)
    expect(bubble).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000)
    # Never stale: the composer recovers (§7.4), and no error state landed.
    expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)

    # The durable record: grounded on the SURVIVING document.
    with SessionLocal() as db:
        rows = db.scalars(select(QueryLog)).all()
        assert len(rows) == 1, [r.question for r in rows]
        row = rows[0]
        assert row.question == QUESTION
        assert row.deflected is False, (
            f"a partial prune broke retrieval — the turn deflected "
            f"(top_score={row.top_score}, fts_hits={row.fts_hits})"
        )
        assert f"{SOURCE_NAME}/{SENTINEL_DOC}" in row.sources, row.sources
        assert all(GONE_PATHS[i] not in row.sources for i in range(len(GONE_PATHS))), (
            f"a pruned document still grounds a turn: {row.sources}"
        )


# ---------------------------------------------------------------------------
# 3. Total source removal: every document AND chunk row of the source
# ---------------------------------------------------------------------------


def test_source_removal_prunes_every_document_of_the_source(
    page: Page, app_url: str, db_ready: None, tmp_path: Path
) -> None:
    """The second mass-delete surface the keyword covers
    (``delete_git_source``, ``app/api/git_sources.py`` — the endpoint that
    took 47 s for 2,303 immich docs on the incident day): removing a
    registered source removes every one of its documents and chunks with
    zero orphans, and the OTHER registered source stays byte-identical.

    The endpoint's contract is ``204`` with **no body** (phase 69), so the
    prune count is pinned against what the KB held immediately before the
    DELETE rather than a response field."""
    page.set_default_timeout(30_000)
    doomed = _make_source_dir(tmp_path, SOURCE_NAME, ALL_PATHS)
    keeper = _make_source_dir(
        tmp_path, KEEPER_NAME, ["notes/keeper-1.md", "notes/keeper-2.md"]
    )

    login(page, app_url, next=SOURCES_URL)
    doomed_id = _add_local_source(page, app_url, doomed)
    _add_local_source(page, app_url, keeper)
    _sync_from_sources_page(page, app_url)
    assert _api_sources(page, app_url) == [SOURCE_NAME, KEEPER_NAME]

    with SessionLocal() as db:
        doomed_doc_ids = _doc_ids(db, SOURCE_NAME, ALL_PATHS)
        doomed_chunks = _chunk_count(db, doomed_doc_ids)
        keeper_before = _chunk_counts_by_doc(db, KEEPER_NAME)
    assert len(doomed_doc_ids) == len(ALL_PATHS)
    assert doomed_chunks > 0, "the fixture must embed chunks to prune"

    r = page.request.delete(f"{app_url}/api/git-sources/{doomed_id}")
    assert r.status == 204, f"expected 204, got {r.status}: {r.text}"

    # Every document of the source is gone and left NO chunk rows; the
    # remaining source is untouched, chunks included.
    with SessionLocal() as db:
        assert _chunk_count(db, doomed_doc_ids) == 0, (
            "source removal left orphan chunk rows — the FK cascade did "
            "not run (Document.chunks' passive_deletes without ON DELETE "
            "CASCADE)"
        )
        assert (
            db.scalar(
                select(func.count())
                .select_from(Document)
                .where(Document.source == SOURCE_NAME)
            )
            or 0
        ) == 0
        assert _sources_in_db(db) == [KEEPER_NAME]
        assert _chunk_counts_by_doc(db, KEEPER_NAME) == keeper_before
        assert _total_chunks(db) == sum(keeper_before.values())

    # The registry row is gone too, and the API agrees with the DB.
    r = page.request.get(f"{app_url}/api/git-sources")
    assert r.status == 200, r.text
    rows = r.json()["sources"]
    assert doomed_id not in [row["id"] for row in rows], rows
    assert len(rows) == 1 and rows[0]["path"] == str(keeper), rows
    assert _api_doc_pairs(page, app_url) == [
        (KEEPER_NAME, f"notes/keeper-{n}.md") for n in (1, 2)
    ]
    # The doomed files stay on disk — a foreign local directory is never
    # touched (the phase-69 locked decision, unaffected by phase 183).
    assert doomed.is_dir() and all((doomed / p).is_file() for p in KEEP_PATHS)