"""Unit: the admin "Sync sources" button contract (phase 32, task 02).

The browser behavior is E2E-covered (tests/e2e/test_sync_button.py,
task 03); here we pin the source-level wiring — the anonymous-safe
ship-hidden button markup, the Sources-page admin reveal on the SAME
cached whoami (no extra fetch), the sync state machine that now lives
in sources.js (owner rework 2026-08-28: the button is Sources-page
only, so the page owns the machine it drives — 2 s poll, 202 start
/ 409 adoption / 403 hide, terminal labels, the "bor:sync-status"
event with the status object as detail, the single-poll-loop guard, no
client-side hard timeout), the page's #sync-result line +
#sync-error-banner, the lazily created error modal (phase 41), and the
§7.4 never-stale CSS (spin + reduced-motion opt-out,
disabled state, 44px floor, contrast pair) — so a silent regression is
caught without a browser.

Phase 64 (task 02): unit coverage of the ``app/api/sync.py`` status
contract — the idle response dict is pinned in full (every pre-existing
key unchanged, plus ``current_file``/``files_done``/``files_total`` as
null/0/0); mid-run the status reports the file the (mock) import is
processing, through the runner's own hook closure (no file yet during
the clone/pull phase — A4); the terminal states clear
``current_file`` while keeping the run's final counts.

Phase 98 (task 01): the status's phase machine joins the pins — the
four new keys (``phase`` / ``current_summary`` / ``summaries_done`` /
``summaries_total``) ride along as null/0/0/0 idle, the prelude keeps
``phase: null``, the import reports ``phase: "import"``, and the
terminal states clear ``phase`` + ``current_summary`` while keeping
the run's final summary counts.

Phase 98 (task 02): the running-state title rule is adjusted — the
FULL untruncated LABEL (the file label, "Writing KB overview…", or
"Summarizing folders… <folder> (n/m)") rides the button title for
EVERY running label; the phase-64 "title only when there is a current
file" (remove-when-null) rule is gone.

Phase 128 (task 01): the label becomes PARTS — #sync-label ships a
main span (the idle text pre-JS) + a hidden count span (the pinned
"(n/m)"), the builders return { main, count }, and the single writer
setSyncLabel(main, count, chunk — the chunk part is phase 129) lands
the parts + the untruncated full label (main + " " + count +
" " + chunk) on the title + #sync-result; the running-state /
settle pins move onto that shape (their labels stay byte-identical
as full strings). Phase 129 (task 04): #sync-label ships a THIRD
hidden span (the pinned "· chunk n/m" — the in-flight file's
embed-batch position), rendered only while a file's embed batch is in
flight (the import phase after the first batch — hidden in every
other state); the markup pin below pins the exact three-part ship
shape.

Phase 149 (task 01): the running button becomes the SPINNER — the
writer gains the `running` flag (the running entry passes `true`,
the settle the default `false`), which reveals the detail box
(#sync-progress — the full untruncated label, presentational
aria-hidden; #sync-result stays the sole announcer) and owns the
button's accessible name (the full live label while running, the
ship value "Sync sources" restored on every settle); the
`is-running` class travels with the spin (added in the running
entry, removed in the settle). The pins below move onto that shape
(the aria-label write leaves the running entry for the writer).

Phase 129 (task 03): the status contract gains the in-flight file's
EMBED-chunk position — ``current_chunk`` (the 1-based first chunk of
the in-flight token-budgeted batch) + ``chunks_total`` (the file's
total content chunks): the idle dict pins them 0/0, the running
dict's pins gain them, and the new mid-run pair test reads the
positions between ticks (1/38 → 13/38) through the runner's own
``_chunk_hook`` closure; the convention split is pinned — the chunk
position clears at every point the import phase ends (the
overview/summaries transitions + both terminals) while the FILE
counts keep their final values.

Phase 134 (task 01): the FIFTH state — ``cancelled`` — joins the
contract. The admin-only ``POST /api/sync/cancel`` (the 403 is pinned
in the integration suite — the router's ``require_admin`` dependency,
no DB/HTTP layer here) is 202 ``cancel requested`` while a run is in
flight (it sets the module ``_cancel_event`` AND ``_task.cancel()``)
and 409 ``no sync is running`` idle / after a terminal run. A
user-requested cancel converts the ``CancelledError`` into the
``cancelled`` terminal (``error`` None, ``detail`` empty, the final
file counts kept, the task ends CANCELLED); an UNREQUESTED cancel
(bare ``task.cancel()`` — the app-shutdown path) re-raises with the
state untouched (never swallowed — the completion criterion's pin);
``start_sync`` clears a stale event so the next run completes to
``success``. The cancel runs here are driven through the REAL
``start_sync`` on a worker loop — ``sync_api._task`` is the endpoint's
target — and the cancel is issued FROM that loop thread (a same-thread
``Task.cancel()`` — a cross-thread cancel would schedule through the
loop's selector wake, which a parked loop never receives).

Phase 134 (task 02): the FRONTEND cancel contract is pinned at the
source level (the browser behavior is E2E-covered in
tests/e2e/test_sync_cancel.py, task 03): the ship-hidden
``#sync-cancel-btn`` markup right after ``#sync-btn`` (the visible
"Cancel" label is the accessible name — WCAG 2.4), the
``.sync-cancel-btn`` secondary-pill CSS (the 44px floor, the AA
ink-soft pair, the guarded hover, the :disabled + ``[hidden]`` rules,
the full-width mobile rule next to the main button's), the show/hide
helpers (show = unhide + enable + aria-label, hide = hide + disable),
the two choke points (``enterSyncRunningState`` shows for kind
"sync" / hides for kind "upload" — the upload job is NOT cancellable
in this phase — and ``settleSyncButton`` hides on EVERY settle), the
``requestCancel`` never-stale lifecycle (disable FIRST, then
POST /api/sync/cancel; the 403 backstop, the silent 409, the 202
wait, the phase-55 retry copy + re-enable on any other failure), the
poll tree's new 5th branch (after failed, before the upload branches
— the 7-branch tree is now 8), the neutral ``applySyncCancelled``
(no modal, no banner, no error styling), and the
``initSyncButton`` re-attach of a ``cancelled`` run.
"""
from __future__ import annotations

import asyncio
import logging
import re
import threading
import time
from collections.abc import Callable, Iterator, Sequence
from datetime import datetime
from pathlib import Path
from typing import Any

import pytest
from fastapi import HTTPException

from app.api import sync as sync_api
from app.config import Settings
from app.models import GitSource
from app.rag.importer import ImportSummary
from app.rag.llm import EmbeddingError

FRONTEND = Path(__file__).resolve().parents[2] / "frontend"
ASSETS = FRONTEND / "assets"

HEADER_JS = ASSETS / "header.js"
SOURCES_JS = ASSETS / "sources.js"
STYLES_CSS = ASSETS / "styles.css"
# Phase 76 (task 02): sources.html is folded into the ONE-document shell
# — the sync markup now lives in the RAG view section of index.html.
SHELL_HTML = FRONTEND / "index.html"


def _text(path: Path) -> str:
    assert path.is_file(), f"missing frontend file: {path}"
    return path.read_text(encoding="utf-8")


def _rag_view(html: str) -> str:
    """The RAG view section of the shell (view-scoped page-sub scope —
    the shell carries one .page-sub per view, so a whole-file match
    would hit the earlier views first)."""
    i = html.find('<section class="view" id="view-rag"')
    assert i != -1, "the RAG view section must be in the shell"
    j = html.find('<section class="view" id="view-git-sources"', i)
    assert j != -1, "the Sources view section must follow the RAG view"
    return html[i:j]


def _body(js: str, fn_name: str) -> str:
    """The source of the first `function <fn_name>` in js (brace
    balanced — since phase 76 task 02 the functions live inside
    mount(root), so the closing brace is indented, not column 0)."""
    fn = js.find(f"function {fn_name}")
    assert fn != -1, f"{fn_name} must be defined"
    open_idx = js.find("{", fn)
    depth = 0
    for i in range(open_idx, len(js)):
        c = js[i]
        if c == "{":
            depth += 1
        elif c == "}":
            depth -= 1
            if depth == 0:
                return js[fn : i + 1]
    raise AssertionError(f"unbalanced braces in {fn_name}")


# ---------- the shell's RAG view: anonymous-safe ship-hidden markup ----------


def test_sync_button_ships_hidden_and_labeled() -> None:
    """#sync-btn SHIPS with the hidden attribute (anonymous-safe —
    sources.js reveals it for the admin at view boot), is a real
    <button type="button">, and carries aria-label="Sync sources" so
    the accessible name stays stable across the label states."""
    tag = re.search(r"<button[^>]*id=\"sync-btn\"[^>]*>", _text(SHELL_HTML))
    assert tag, "the shell must carry the #sync-btn button (RAG view)"
    attrs = tag.group(0)
    assert 'class="sync-btn"' in attrs
    assert 'type="button"' in attrs
    assert re.search(r"\bhidden\b", attrs), "#sync-btn must ship hidden"
    assert 'aria-label="Sync sources"' in attrs


def test_sync_button_has_icon_and_label_parts() -> None:
    """The button body is a refresh-cycle svg (aria-hidden — decorative,
    the spin is the visible running state) + the .sync-label span with
    the idle text, so the label can be swapped by sources.js.

    Phase 128 (task 01): #sync-label is PARTS — the MAIN span
    (.sync-label-main) carries the idle text (the pre-JS label renders
    from the static markup; the part CSS may ellipsize it), the
    COUNT span (.sync-label-count — the pinned "(n/m)" position,
    never clipped) ships hidden + empty (there is no count at idle),
    and — phase 129 (task 04) — the CHUNK span (.sync-label-chunk —
    the pinned "· chunk n/m" embed-batch position) ships hidden +
    empty too (no chunk position at idle — the part renders only
    while a file's embed batch is in flight). The class/id stay (the
    module query + the E2E label assertions)."""
    text = _text(SHELL_HTML)
    btn = text[text.find('id="sync-btn"') : text.find("</button>", text.find('id="sync-btn"'))]
    assert re.search(r'<svg[^>]*class="sync-icon"[^>]*aria-hidden="true"', btn)
    # Phase 128 (task 01) + Phase 129 (task 04): the parts markup —
    # the exact three-part shape (the main span holds the idle text,
    # the count + chunk spans are hidden + empty).
    label = re.search(
        r'<span\b[^>]*id="sync-label"[^>]*>'
        r'<span class="sync-label-main">Sync sources</span>'
        r'<span class="sync-label-count" hidden></span>'
        r'<span class="sync-label-chunk" hidden></span>'
        r"</span>",
        btn,
    )
    assert label, (
        "the #sync-label parts markup (main + hidden count + hidden chunk) is missing"
    )
    assert 'class="sync-label"' in label.group(0)


def test_sync_result_is_the_aria_live_announcer() -> None:
    """#sync-result sits right after the button and is a polite live
    region (role="status" + aria-live="polite") — the last-result /
    counts announcement for screen readers."""
    text = _text(SHELL_HTML)
    tag = re.search(r'<span[^>]*id="sync-result"[^>]*>', text)
    assert tag, "the shell must carry the #sync-result announcer (RAG view)"
    attrs = tag.group(0)
    assert 'role="status"' in attrs
    assert 'aria-live="polite"' in attrs
    assert text.find('id="sync-result"') > text.find("</button>", text.find('id="sync-btn"'))


def test_sync_error_banner_is_a_hidden_alert() -> None:
    """The failure banner uses the chat error-banner markup style
    (kb-banner + is-error) and role="alert", shipping hidden —
    sources.js un-hides it with the error text on a failed run."""
    text = _text(SHELL_HTML)
    tag = re.search(r'<div[^>]*id="sync-error-banner"[^>]*>', text)
    assert tag, "the shell must carry the #sync-error-banner (RAG view)"
    attrs = tag.group(0)
    assert "kb-banner" in attrs and "is-error" in attrs
    assert 'role="alert"' in attrs
    assert re.search(r"\bhidden\b", attrs)
    assert 'id="sync-error-text"' in text
    # The banner lives in the page content, not the 64px header bar.
    assert text.find('id="sync-error-banner"') > text.find('<main id="main"')


def test_page_sub_copy_mentions_the_button() -> None:
    """The page-sub copy names the button as the one-click way to pull
    the latest and re-import (the import CLI docs live elsewhere).
    Phase 61: the copy describes the current source model (git repos +
    local directories + uploaded archives), not the old ~/Homelab +
    ~/Deployments clone. Phase 76 (task 02): scoped to the RAG view —
    the shell carries one .page-sub per view."""
    sub = re.search(r'<p class="page-sub">(.*?)</p>', _rag_view(_text(SHELL_HTML)), re.DOTALL)
    assert sub, "the RAG view must keep the .page-sub copy"
    copy = re.sub(r"\s+", " ", sub.group(1))  # the markup wraps lines
    assert "Press <strong>Sync sources</strong>" in copy
    assert "pull the latest and re-import" in copy
    assert "git repositories," in copy
    assert "local directories, and uploaded archives" in copy


def test_sources_page_stays_cdn_free() -> None:
    """No-CDN rule (PLAN §7.3, A11): the new button markup adds no
    external references — same-origin assets only (the integration
    test_index_html_served_locally re-checks this on the served page)."""
    text = _text(SHELL_HTML)
    assert 'src="https://' not in text
    assert 'href="https://' not in text


# ---------- sources.js: the admin reveal (view boot) ----------


def test_sources_js_reveals_sync_btn_on_the_admin_branch() -> None:
    """The view boot (mount's tail, phase 76 task 02) reveals #sync-btn
    for the admin on the SAME cached whoami fetchIsAdmin() reads (no
    extra fetch — header.js keeps the single /api/whoami call site; the
    header itself is booted exactly once, by the chat module at shell
    boot); anonymous users never leave the ship-hidden default."""
    js = _text(SOURCES_JS)
    boot = js[js.find("view boot (phase 76 task 02)"):]
    assert "const admin = await fetchIsAdmin()" in boot
    assert "syncBtn.hidden = !admin" in boot, "#sync-btn must join the admin reveal"
    # The reveal must not introduce a second whoami call site.
    header = _text(HEADER_JS)
    assert header.count('fetch("/api/whoami")') == 1
    assert 'fetch("/api/whoami")' not in js


# ---------- sources.js: the sync state machine (moved here from
# ---------- header.js in the owner rework 2026-08-28) ----------


def test_sources_js_owns_the_sync_button_elements() -> None:
    """The button refs are module-level and null-safe: #sync-btn, the
    .sync-label inside the button, the .sync-icon inside the button — a
    page without the markup is a complete no-op."""
    js = _text(SOURCES_JS)
    assert 'querySelector("#sync-btn")' in js
    assert 'syncBtn.querySelector(".sync-label")' in js
    assert 'syncBtn.querySelector(".sync-icon")' in js


def test_sources_js_calls_the_sync_api() -> None:
    """The click posts to POST /api/sync and the poll loop GETs
    /api/sync/status — both through the same-origin API (A10)."""
    js = _text(SOURCES_JS)
    assert 'fetch("/api/sync", { method: "POST" })' in js
    assert 'fetch("/api/sync/status")' in js


def test_sources_js_polls_every_2000ms() -> None:
    """The feedback loop is a 2000 ms poll of the status endpoint,
    re-scheduled one tick at a time (setTimeout, not setInterval — an
    in-flight fetch can never overlap the next tick)."""
    js = _text(SOURCES_JS)
    assert "SYNC_POLL_MS = 2000" in js
    assert "setTimeout(tick, SYNC_POLL_MS)" in js
    assert "setInterval" not in js


def test_sources_js_adopts_409_and_starts_on_202() -> None:
    """202 (started) and 409 (a run started elsewhere — e.g. a second
    tab) both enter the running state and start polling: the UI never
    starts a second run, it adopts the in-flight one.

    Phase 64 (task 04): the entry is the sync-kind live label with no
    file yet (the run is just starting — bare "Syncing…", A4)."""
    js = _text(SOURCES_JS)
    assert "r.status === 202 || r.status === 409" in js
    idx = js.find("r.status === 202 || r.status === 409")
    branch = js[idx : idx + 600]
    assert "enterSyncRunningState(\"sync\", null, 0, 0)" in branch
    assert "startSyncPolling()" in branch


def test_sources_js_hides_the_button_on_403() -> None:
    """A 403 anywhere (POST, the status poll, the load re-attach) is
    treated as not-admin: the button hides — defense in depth behind
    the whoami reveal (the primary gate)."""
    js = _text(SOURCES_JS)
    assert len(re.findall(r"r\.status === 403", js)) >= 3, (
        "POST, the status poll, and the load re-attach must all handle 403"
    )
    assert js.count("syncBtn.hidden = true") >= 3, (
        "every 403 branch must hide the button"
    )


def test_sources_js_running_state_is_never_stale() -> None:
    """Entering the running state disables the button, sets aria-busy,
    spins the icon, and swaps the label to the live file label
    (fmtSyncLabel — the §7.4 feedback while the poll waits) — and a
    fresh run starts clean: the previous failure's title / aria-label /
    .is-error come off NOW, not when the run settles.

    Phase 64 (task 04): #sync-result (the aria-live announcer) carries
    the FULL untruncated label. Phase 98 (task 02): the title rule is
    adjusted — the FULL untruncated LABEL (the file label, the overview
    label, or the summaries label — whatever the running form is) rides
    the button title for EVERY running label; the "title only when
    there is a current file" (remove-when-null) rule is gone.
    Phase 128 (task 01): the entry hands the label PARTS (the
    phase-aware split — sync + status → fmtSyncPhaseLabel, the
    fall-through → the byte-identical fmtSyncLabel) to the SINGLE
    writer, setSyncLabel(main, count, chunk — phase 129: the chunk
    part destructures with a "" default), which lands the main part,
    the pinned count part (hidden when absent), the pinned chunk part
    (phase 129 — hidden when absent), and the FULL untruncated label
    on the button title + #sync-result (A4) — the direct span /
    title / announcer writes are GONE from the entry (they live in
    setSyncLabel). Phase 149 (task 01): the writer call passes
    `running = true` (the detail box reveals with the full label),
    and the button's accessible name rides in the writer too (the
    full live label — the icon-only running button keeps a name);
    the entry no longer sets aria-label at all. The `is-running`
    class is added here, with the spin."""
    js = _text(SOURCES_JS)
    body = _body(js, "enterSyncRunningState")
    assert "syncBtn.disabled = true" in body
    assert 'syncBtn.setAttribute("aria-busy", "true")' in body
    # the parts split (phase 98) is intact…
    assert "fmtSyncPhaseLabel(status)" in body
    assert "fmtSyncLabel(kind, currentFile, done, total)" in body
    # …and the single writer lands the parts (A4 in it — the title is
    # set for EVERY running label: the file-only rule is adjusted
    # away) with the phase-149 `running` flag (the box reveals).
    assert "setSyncLabel(main, count, chunk, true)" in body
    assert body.count("setSyncLabel(") == 1, "the entry writes ONLY through the parts writer"
    assert "syncBtn.title =" not in body, ("the title write lives in setSyncLabel (A4)")
    assert "syncLabel.textContent" not in body, ("no direct span write — the parts")
    assert "syncResult.textContent" not in body, ("the announcer write lives in setSyncLabel (A4)")
    assert 'syncBtn.setAttribute("aria-label"' not in body, (
        "the aria-label write lives in setSyncLabel (phase 149: the full"
        " live label while running)"
    )
    assert "syncBtn.classList.remove(\"is-error\")" in body
    assert "syncIcon.classList.add(\"is-spinning\")" in body
    assert "syncBtn.classList.add(\"is-running\")" in body, (
        "the icon-only running button class travels with the spin"
    )


def test_sources_js_terminal_states() -> None:
    """Terminal rendering: success → enabled + 'Synced HH:MM' (local
    time of finished_at); failed → enabled + retry-ready 'Sync sources'
    label + the sanitized error in the button's title + aria-label +
    the .is-error class. The counts formatting (fmtSyncResult) lives
    here ('added' always announced, zero terms omitted — a no-op
    re-sync reads '0 added · 1 unchanged')."""
    js = _text(SOURCES_JS)

    success = _body(js, "applySyncSuccess")
    assert '"Synced"' in success and "fmtSyncTime(status.finished_at)" in success

    failure = _body(js, "applySyncFailure")
    assert 'settleSyncButton("Sync sources")' in failure  # retry-ready
    assert "syncBtn.title = error" in failure
    assert 'syncBtn.setAttribute("aria-label", error)' in failure
    assert "syncBtn.classList.add(\"is-error\")" in failure
    assert "sanitizeSyncError(status.error)" in failure

    result = _body(js, "fmtSyncResult")
    assert "added" in result and "unchanged" in result
    assert " · " in result
    assert "> 0" in result, "zero terms must be omitted"

    time = _body(js, "fmtSyncTime")
    assert "getHours()" in time and "getMinutes()" in time, "local HH:MM of finished_at"


def test_sources_js_failed_error_is_sanitized() -> None:
    """The button's title/aria-label error is sanitized for the
    attributes: the server already masks credentials (sync.py
    _sanitize_error); the page collapses whitespace to a single line
    and caps the length, and a missing error still names a failure."""
    js = _text(SOURCES_JS)
    body = _body(js, "sanitizeSyncError")
    assert "replace(/\\s+/g, \" \")" in body, "single line for the attributes"
    assert "200" in body, "long errors (chatty git stderr) are capped"
    assert '"The sync failed."' in body


def test_sources_js_settles_the_button_on_terminal() -> None:
    """settleSyncButton re-enables the control, drops aria-busy, the
    title, and the failed affordances, and un-spins the icon — the
    button can never sit disabled after a run reaches a terminal state
    (failed included: retry-ready). Phase 128 (task 01): the settle /
    idle label ("Sync sources", "Synced HH:MM") goes through the parts
    writer — setSyncLabel(label), count hidden (absent), the full
    string riding the title + #sync-result (A4, every label state).
    Phase 149 (task 01): the `is-running` class is removed WITH the
    spin (every terminal + both idles pass through the settle), and
    the writer call's default `running = false` hides the detail box
    + restores the idle accessible name."""
    js = _text(SOURCES_JS)
    body = _body(js, "settleSyncButton")
    assert "syncBtn.disabled = false" in body
    assert 'syncBtn.removeAttribute("aria-busy")' in body
    assert 'syncBtn.removeAttribute("title")' in body
    assert 'syncBtn.setAttribute("aria-label", "Sync sources")' in body
    assert "syncIcon.classList.remove(\"is-spinning\")" in body
    assert "syncBtn.classList.remove(\"is-running\")" in body, (
        "the running class travels with the spin — the settle removes both"
    )
    assert "setSyncLabel(label)" in body, ("the settle label goes through the parts writer")
    assert "setSyncLabel(label, true)" not in body, (
        "the settle is NEVER a running state — the box hides (phase 149)"
    )
    assert "syncLabel.textContent" not in body, ("no direct span write — the parts")


def test_sources_js_never_starts_a_second_poll_loop() -> None:
    """startSyncPolling is guarded by the module-level timer: a 409
    adoption, a reload re-attach, or a stray call can never run two
    poll loops at once (phase completion criterion)."""
    js = _text(SOURCES_JS)
    fn = js.find("function startSyncPolling")
    head = js[fn : js.find("const tick", fn)]
    assert re.search(r"if\s*\(\s*syncPollTimer\s*!==\s*null\s*\)\s*return", head), (
        "the single-loop guard must be the first statement"
    )
    assert "clearTimeout(syncPollTimer)" in _body(js, "stopSyncPolling")


def test_sources_js_has_no_client_side_hard_timeout() -> None:
    """Phase locked decision: a sync can legitimately run for minutes
    (and outlive the page), so there is NO client-side hard timeout —
    the 2 s poll is the feedback loop and the server state is
    authoritative (the 120 s LLM-turn guard must not leak into the
    sync path)."""
    js = _text(SOURCES_JS)
    assert "TURN_TIMEOUT" not in js
    sync_start = js.find("Sync sources button (Sources page only)")
    assert sync_start != -1, "the sync section marker comment"
    assert "120" not in js[sync_start:]


def test_sources_js_reattaches_on_load_admin_only() -> None:
    """initSyncButton (run at page boot, after the click binding)
    awaits the SAME cached whoami — ADMIN ONLY (non-admins never poll,
    the status endpoint is admin-only): a running run re-enters the
    running state (reload mid-sync), a terminal run renders its last
    result, idle settles retry-ready; the click binding wires
    startSync to the button."""
    js = _text(SOURCES_JS)
    body = _body(js, "initSyncButton")
    assert "await fetchIsAdmin()" in body, "admin-only boot (no extra fetch)"
    assert 'fetch("/api/sync/status")' in body
    assert 'status.state === "running"' in body
    assert 'status.state === "success"' in body
    assert 'status.state === "failed"' in body
    assert (
        'syncBtn.addEventListener("click", startSync);\n    initSyncButton();'
        in js
    ), "the click binding and the boot re-attach ship together, guarded on syncBtn"


def test_sources_js_emits_bor_sync_status_on_state_changes() -> None:
    """Every state change dispatches window 'bor:sync-status' with the
    status object as detail — the channel other page scripts (or
    future ones) can subscribe to. The click path emits the synthetic
    running frame IMMEDIATELY (no 2 s poll lag)."""
    js = _text(SOURCES_JS)
    assert (
        'window.dispatchEvent(new CustomEvent("bor:sync-status", { detail: status }))'
        in js
    )
    for fn in ("applySyncSuccess", "applySyncFailure", "applySyncIdle"):
        assert "emitSyncStatus" in _body(js, fn), f"{fn} must emit its frame"
    # the running frame: synthetic on click, the real object on boot
    assert 'emitSyncStatus({ state: "running" })' in js
    assert "emitSyncStatus(status)" in _body(js, "initSyncButton")


# ---------- sources.js: the sync failure modal (phase 41, TODO.md L4) ----------


def test_sources_js_owns_a_lazily_created_sync_modal() -> None:
    """The modal is page-owned and created lazily ONCE (module-level
    `let syncModal = null`): a .sync-modal-backdrop holding a
    .sync-modal panel with role="alertdialog" + aria-modal +
    labelled/described ids + the close button, appended to
    document.body — no page-markup changes needed (the section marker
    records that the former header.js modal was recreated here)."""
    js = _text(SOURCES_JS)
    assert "let syncModal = null" in js, "module-level once-only modal ref"
    assert 'role="alertdialog"' in js
    assert 'aria-modal="true"' in js
    assert 'aria-labelledby="sync-modal-title"' in js
    assert 'aria-describedby="sync-modal-error"' in js
    assert "sync failure modal" in js, "the modal section marker"
    create = _body(js, "createSyncModal")
    assert "sync-modal-backdrop" in create
    assert 'document.body.appendChild(backdrop)' in create
    assert "return backdrop" in create, "the module ref must hold the created element"
    assert "Sync failed" in create, "the dialog title"
    assert 'class="sync-modal-close"' in create
    assert 'aria-label="Close error dialog"' in create


def test_sync_modal_error_is_rendered_via_text_content() -> None:
    """The error text is ALWAYS set via textContent (XSS-safe — no
    innerHTML with user data in the open path), and it is set BEFORE
    the already-open check, so a second failure while open updates the
    text IN PLACE (no stacking, no focus jump)."""
    js = _text(SOURCES_JS)
    body = _body(js, "showSyncModal")
    assert 'querySelector("#sync-modal-error").textContent' in body
    assert "innerHTML" not in body, "the open path never touches innerHTML"
    text = body.find("textContent")
    open_check = body.find('contains("is-open")')
    assert text != -1 and open_check != -1 and text < open_check, (
        "the in-place update happens while the modal is already open"
    )


def test_sync_modal_focus_goes_in_and_out_to_sync_btn() -> None:
    """On open: document.activeElement is remembered and focus moves
    to the close button — with a fallback to #sync-btn when the active
    element is <body> (the run's disabled button dropped focus there;
    the close must still land on the control that started the run);
    on close: focus returns to the remembered element (guarded by
    document.contains — a detached target is a no-op)."""
    js = _text(SOURCES_JS)
    open_body = _body(js, "showSyncModal")
    assert "const active = document.activeElement" in open_body
    assert "active !== document.body" in open_body, (
        "the disabled-button window leaves focus on <body> — the fallback"
    )
    assert "? active : syncBtn" in open_body, ("the fallback remembers #sync-btn")
    assert 'querySelector(".sync-modal-close").focus()' in open_body
    close_body = _body(js, "closeSyncModal")
    assert "document.contains(target)" in close_body
    assert "target.focus()" in close_body
    assert 'contains("is-open")' in close_body, "closing a closed modal is a no-op"


def test_sync_modal_closes_via_button_esc_and_backdrop() -> None:
    """All three dismissal paths call the SAME close function: the
    close button, Esc (ONE document keydown binding, acting only while
    the modal is open), and a click on the backdrop element itself —
    the event.target check keeps clicks bubbling from the panel from
    closing it."""
    js = _text(SOURCES_JS)
    create = _body(js, "createSyncModal")
    assert 'addEventListener("click", closeSyncModal)' in create
    assert 'e.key === "Escape"' in create
    assert 'addEventListener("keydown"' in create
    assert 'backdrop.classList.contains("is-open")' in create
    assert "e.target === backdrop" in create


def test_sources_js_apply_sync_failure_opens_the_modal_after_the_event() -> None:
    """applySyncFailure opens the modal with the SANITIZED error, and
    the call comes AFTER the existing button title/aria/.is-error
    lines + emitSyncStatus (the bor:sync-status event — those lines
    stay byte-identical; the modal is additive)."""
    js = _text(SOURCES_JS)
    body = _body(js, "applySyncFailure")
    call = body.find("showSyncModal(")
    emit = body.find("emitSyncStatus(status)")
    assert call != -1, "the modal must open from the failure path"
    assert emit != -1 and call > emit, "the event still emits first (byte-identical)"
    assert "showSyncModal(error)" in body, "the sanitized error goes to the modal"
    # The pre-existing affordances stay (the Sources banner contract).
    assert "syncBtn.title = error" in body
    assert 'syncBtn.setAttribute("aria-label", error)' in body
    assert 'syncBtn.classList.add("is-error")' in body
    assert "emitSyncStatus(status)" in body


# ---------- sources.js: the page-specific result line + banner ----------


def test_sources_js_renders_result_line_and_banner_directly() -> None:
    """sources.js owns the page-specific rendering on the Sources page:
    #sync-result (aria-live) + #sync-error-banner (role=alert), driven
    directly by the state appliers: success → the counts (fmtSyncResult)
    + a live catalog refresh; failed → the banner with the sanitized
    error; every state change still emits 'bor:sync-status' (the status
    object as detail) for other subscribers."""
    js = _text(SOURCES_JS)
    success = _body(js, "applySyncSuccess")
    assert "syncResult.textContent = fmtSyncResult(status.detail)" in success, (
        "the counts live in status.detail — a bare status renders all zeros"
    )
    assert "loadTree()" in success, (
        "the catalog re-fetches live on a successful sync (phase 97: the tree)"
    )
    failure = _body(js, "applySyncFailure")
    assert "showSyncError(error)" in failure
    assert "emitSyncStatus" in _body(js, "applySyncIdle")
    assert (
        'window.dispatchEvent(new CustomEvent("bor:sync-status", { detail: status }))'
        in _body(js, "emitSyncStatus")
    )


def test_header_js_no_longer_owns_the_sync_machine() -> None:
    """The state machine is GONE from header.js (sources.js owns it —
    the button is Sources-page only, so the page drives it): no button
    refs, no POST, no status poll, no button-state helpers, no modal.
    sources.js keeps every one of them."""
    header = _text(HEADER_JS)
    for gone in (
        "SYNC_POLL_MS",
        "syncPollTimer",
        "startSyncPolling",
        "stopSyncPolling",
        "enterSyncRunningState",
        "settleSyncButton",
        "applySyncSuccess",
        "applySyncFailure",
        "applySyncIdle",
        "initSyncButton",
        "syncModal",
        "syncBtn",
        'fetch("/api/sync", { method: "POST" })',
        'fetch("/api/sync/status")',
        'querySelector("#sync-btn")',
    ):
        assert gone not in header, f"{gone!r} must be gone from header.js (sources.js owns it)"
    js = _text(SOURCES_JS)
    for kept in (
        "SYNC_POLL_MS = 2000",
        "syncPollTimer",
        "startSyncPolling",
        "enterSyncRunningState",
        "initSyncButton",
        'fetch("/api/sync", { method: "POST" })',
        'fetch("/api/sync/status")',
        'querySelector("#sync-btn")',
    ):
        assert kept in js, f"{kept!r} must be in sources.js (the owner)"


# ---------- styles.css: the §7.4 states ----------


def test_sync_button_css_ghost_pill_and_disabled_state() -> None:
    """.sync-btn is a pill in the shared button family (contrast pair
    ink-soft on surface ≥ 4.5:1), with a ≥44px touch floor and a
    :disabled state (never stale — the busy look is visible)."""
    css = _text(STYLES_CSS)
    block = re.search(r"\.sync-btn\s*\{([^}]*)\}", css)
    assert block, "styles.css must define .sync-btn"
    body = block.group(1)
    assert "min-height: 44px" in body
    assert "color: var(--ink-soft)" in body
    assert "border-radius: 999px" in body
    disabled = re.search(r"\.sync-btn:disabled\s*\{([^}]*)\}", css)
    assert disabled, ".sync-btn:disabled must be styled"
    assert "cursor: wait" in disabled.group(1)


def test_sync_icon_spins_and_respects_reduced_motion() -> None:
    """The running state spins the refresh icon on the shared spin
    keyframes (1s linear infinite), and prefers-reduced-motion stills
    it — the existing opt-out pattern."""
    css = _text(STYLES_CSS)
    spin = re.search(r"\.sync-btn \.sync-icon\.is-spinning\s*\{([^}]*)\}", css)
    assert spin, "the .is-spinning state must be styled"
    assert "animation: spin 1s linear infinite" in spin.group(1)
    spinner = r"\.sync-btn \.sync-icon\.is-spinning\s*\{([^}]*)\}"
    reduced = re.search(r"@media \(prefers-reduced-motion: reduce\)\s*\{\s*" + spinner, css)
    assert reduced, "the spin must opt out under prefers-reduced-motion"
    assert "animation: none" in reduced.group(1)
    assert "@keyframes spin" in css, "the spin keyframes are shared (pre-existing)"


def test_sync_result_is_styled() -> None:
    """#sync-result (the aria-live last-result line) is styled in the
    theme tokens — soft ink, small mono, no wrap in the header bar."""
    css = _text(STYLES_CSS)
    block = re.search(r"\.sync-result\s*\{([^}]*)\}", css)
    assert block, "styles.css must define .sync-result"
    assert "var(--ink-soft)" in block.group(1)


# ---------- styles.css: the sync failure modal (phase 41, TODO.md L4) ----------


def test_sync_modal_css_error_palette_and_stacking() -> None:
    """.sync-modal-backdrop: fixed, full-viewport, the --bg-82% dim
    (phase 92: color-mix of the identity variable), z-index above the
    sticky header; .sync-modal: the centered ≈28rem panel on
    the phase-08 error palette (panel on --err-bg, 1px --err-line
    border, --err-ink error text, --ink title — all computed ≥4.5:1);
    open/close via .is-open (visibility/opacity)."""
    css = _text(STYLES_CSS)
    backdrop = re.search(r"\.sync-modal-backdrop\s*\{([^}]*)\}", css)
    assert backdrop, "styles.css must define .sync-modal-backdrop"
    b = backdrop.group(1)
    assert "position: fixed" in b
    assert "inset: 0" in b
    assert "z-index: 1000" in b, "above the sticky header (20) + skip-link (100)"
    assert "color-mix(in srgb, var(--bg) 82%, transparent)" in b, (
        "the dim over the page (phase 92: --bg at 82%)"
    )
    open_state = re.search(r"\.sync-modal-backdrop\.is-open\s*\{([^}]*)\}", css)
    assert open_state, ".is-open must be the open state"
    assert "visibility: visible" in open_state.group(1)
    assert "opacity: 1" in open_state.group(1)
    panel = re.search(r"\.sync-modal\s*\{([^}]*)\}", css)
    assert panel, "styles.css must define .sync-modal"
    p = panel.group(1)
    assert "max-width: 28rem" in p
    assert "var(--err-bg)" in p
    assert "var(--err-line)" in p
    title = re.search(r"#sync-modal-title\s*\{([^}]*)\}", css)
    assert title, "the modal title must be styled"
    assert "var(--ink)" in title.group(1)
    error = re.search(r"#sync-modal-error\s*\{([^}]*)\}", css)
    assert error, "the modal error line must be styled"
    assert "var(--err-ink)" in error.group(1)


def test_sync_modal_close_button_touch_floor() -> None:
    """The close button keeps the 44px touch floor at every width (the
    global 3px :focus-visible outline applies — no per-button
    override)."""
    css = _text(STYLES_CSS)
    block = re.search(r"\.sync-modal-close\s*\{([^}]*)\}", css)
    assert block, "styles.css must define .sync-modal-close"
    assert "min-width: 44px" in block.group(1)
    assert "min-height: 44px" in block.group(1)


def test_sync_modal_respects_reduced_motion() -> None:
    """The open/close fade is stilled under prefers-reduced-motion —
    the phase-25 / doc-modal opt-out pattern."""
    css = _text(STYLES_CSS)
    reduced = re.search(
        r"@media \(prefers-reduced-motion: reduce\)\s*\{\s*\.sync-modal-backdrop\s*\{([^}]*)\}",
        css,
    )
    assert reduced, "the backdrop fade must opt out under prefers-reduced-motion"
    assert "transition: none" in reduced.group(1)


# ---------- phase 64 (task 02): per-file progress on the sync status ----------
#
# The GET /api/sync/status contract in app/api/sync.py (task 04 renders
# the live file label on the button — its pins live in
# tests/unit/test_frontend_sync_upload.py). The runner's seams are
# monkeypatched on ``app.api.sync`` (the house mock-import pattern from
# tests/integration/test_sync_api.py); the background task runs on a
# worker thread's own loop so the test can read the status mid-run.
# No DB, no HTTP: every seam is faked.


@pytest.fixture()
def fresh_sync_status() -> Iterator[None]:
    """The module-level status object + task are process-global: reset
    them before AND after every progress test (the integration suite's
    ``_fresh_sync_state`` pattern).

    Phase 134 (task 01): the module ``_cancel_event`` joins the reset —
    a stale (set) event from a previous test's cancelled run would
    otherwise bleed into the next run's boundary checks."""
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()
    yield
    sync_api._status = sync_api.SyncStatus()
    sync_api._task = None
    sync_api._cancel_event = asyncio.Event()


class _GatedClone:
    """A ``clone_or_pull`` that parks between start and finish on a
    threading gate, so the test can read the status during the
    clone/pull phase (A4: no file yet). Phase 135: accepts the row's
    branch pin (the third argument — ``None`` for this file's rows)."""

    def __init__(self, started: threading.Event, release: threading.Event) -> None:
        self.started = started
        self.release = release
        self.calls: list[tuple[str, Path]] = []

    def __call__(self, url: str, dest: Path | str, branch: str | None = None) -> Path:
        dest = Path(dest)
        self.calls.append((url, dest))
        dest.mkdir(parents=True, exist_ok=True)
        (dest / "notes.md").write_text("# repo\ncontent for the KB\n", encoding="utf-8")
        self.started.set()
        self.release.wait()  # blocking is fine: the clone is a sync call
        return dest


class _GatedImport:
    """The mock import: fires the runner's OWN progress hook once (the
    progress-shaped call goes through the real hook closure — the
    closure under test), parks on a threading gate so the test can read
    the status mid-run, then returns the canned summary (or raises
    ``fail``).

    Phase 129 (task 03): optionally ALSO fires the in-flight file's
    chunk-position pairs (``chunk_pairs``) through the runner's OWN
    ``_chunk_hook`` closure: pair one lands before the main park
    (readable mid-run), and every later pair lands after the main
    park + its own park (``chunk_release``), so each position is
    readable on the status between ticks."""

    def __init__(
        self,
        summary: ImportSummary,
        started: threading.Event,
        release: threading.Event,
        fail: BaseException | None = None,
        chunk_pairs: Sequence[tuple[int, int]] = (),
        chunk_release: threading.Event | None = None,
    ) -> None:
        self.summary = summary
        self.started = started
        self.release = release
        self.fail = fail
        self.hook_calls: list[tuple[str, str, int, int]] = []
        self.prune_flags: list[bool] = []
        self.chunk_pairs = list(chunk_pairs)
        self.chunk_release = chunk_release
        self.chunk_hook_calls: list[tuple[int, int]] = []

    async def __call__(
        self,
        sources: list[Path],
        llm: object,
        *,
        prune: bool = False,
        limit: int | None = None,
        session: object = None,
        progress: Callable[[str, str, int, int], None] | None = None,
        on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
        ignore_by_root: dict[str, list[str]] | None = None,  # phase 89
        include_hidden_by_root: dict[str, bool] | None = None,  # phase 105
        doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,  # phase 106
    ) -> ImportSummary:
        self.prune_flags.append(prune)
        if progress is not None:
            progress("repo", "notes/deep.md", 1, 3)
            self.hook_calls.append(("repo", "notes/deep.md", 1, 3))
        if on_chunk_progress is not None and self.chunk_pairs:
            # Phase 129 (task 03): the file's chunk-position pairs,
            # fired through the runner's OWN hook closure (the closure
            # under test) — pair one before the main park, each later
            # pair after the main park + its own park (each park lets
            # the test read the status mid-run). The runner passes the
            # hook only when the wiring is intact — a regression to
            # None skips this branch and the pair tests time out.
            on_chunk_progress(*self.chunk_pairs[0])
            self.chunk_hook_calls.append(self.chunk_pairs[0])
            self.started.set()
            await asyncio.to_thread(self.release.wait)
            for pair in self.chunk_pairs[1:]:
                on_chunk_progress(*pair)
                self.chunk_hook_calls.append(pair)
                if self.chunk_release is not None:
                    await asyncio.to_thread(self.chunk_release.wait)
        else:
            self.started.set()
            await asyncio.to_thread(self.release.wait)  # park without freezing the worker loop
        if self.fail is not None:
            raise self.fail
        return self.summary


def _patch_sync_seams(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    fake_import: object,  # any import-shaped fake (the phase-134 fakes too)
    fake_clone: object,  # any clone-shaped fake (the phase-134 fakes too)
) -> None:
    """The runner's seams, monkeypatched on ``app.api.sync`` (the house
    mock-import pattern): fresh settings (no ``.env`` leak), a no-op
    model probe, a sentinel LLM client, one git row, the gated clone +
    import, a no-op overview, a no-op folder-summary step (phase 94 —
    the sentinel LLM client has no ``chat``), and the DB-free
    sources-version step (dummy session + pinned counters)."""
    monkeypatch.setattr(
        sync_api,
        "get_settings",
        lambda: Settings(_env_file=None, sources_dir=str(tmp_path / "bor")),  # pyright: ignore[reportCallIssue]
    )

    async def fake_probe(llm: object) -> None:
        pass

    monkeypatch.setattr(sync_api, "check_models", fake_probe)
    monkeypatch.setattr(sync_api, "LLMClient", lambda: object())
    monkeypatch.setattr(
        sync_api,
        "effective_sources",
        lambda session: (
            # kind explicit: the Python-side default applies at INSERT
            # flush, not on an in-memory instance
            [GitSource(url="https://git.example.com/repo.git", kind="git")],
            "db",
        ),
    )
    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
    monkeypatch.setattr(sync_api, "import_sources", fake_import)

    async def fake_overview(llm: object, session: object = None) -> bool:
        return False

    monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)

    async def fake_folder_summaries(
        db: object,
        llm: object,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        # Phase 98 (task 01): the runner wires the progress hook into
        # the generation branches — the fake accepts it (the state-
        # machine tests' canned run never fires it: zero summary
        # counters, like a no-gap skip). Phase 127 (task 03): the
        # changed-KB branch passes the change set as changed_docs= —
        # the fake accepts it (these canned runs record nothing).
        return {"generated": 0, "failed": 0, "pruned": 0}

    monkeypatch.setattr(sync_api, "generate_folder_summaries", fake_folder_summaries)
    # Phase 96 (task 03): the unchanged-walk gap probe is DB-free in
    # these state-machine tests — no gap, so the folder step stays
    # skipped exactly as before the gate change.
    monkeypatch.setattr(sync_api, "missing_folder_summaries", lambda session: [])

    class _DummySession:
        def close(self) -> None:
            pass

        def commit(self) -> None:
            pass

    monkeypatch.setattr(sync_api, "SessionLocal", _DummySession)
    monkeypatch.setattr(sync_api, "bump_sources_version", lambda session: 1)
    monkeypatch.setattr(sync_api, "current_sources_version", lambda session: 1)


def _start_run() -> tuple[threading.Thread, list[BaseException]]:
    """Run the module-level runner on a worker thread's own event loop
    (the house background-task pattern), capturing any unexpected
    exception — the runner is supposed to die in state, never raise."""
    errors: list[BaseException] = []

    def _run() -> None:
        try:
            asyncio.run(sync_api._run_sync())
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    return thread, errors


def test_idle_status_pins_full_shape_including_progress_keys(
    fresh_sync_status: None,
) -> None:
    """Idle: the three phase-64 progress keys ride along as null/0/0,
    the two phase-129 chunk keys as 0/0, the four phase-98 phase keys
    as null/0/0/0, and EVERY pre-existing key is unchanged — the full
    response dict is pinned, so the current UI and every existing
    consumer keep working."""
    assert sync_api.sync_status() == {
        "state": "idle",
        "started_at": None,
        "finished_at": None,
        "detail": {},
        "error": None,
        "current_file": None,
        "files_done": 0,
        "files_total": 0,
        # Phase 129 (task 03): the chunk-position keys — 0/0 idle
        # (the dataclass defaults, no special-casing).
        "current_chunk": 0,
        "chunks_total": 0,
        # Phase 98 (task 01): the phase-machine keys — null/0/0/0 idle.
        "phase": None,
        "current_summary": None,
        "summaries_done": 0,
        "summaries_total": 0,
    }


def test_mid_run_status_reports_current_file(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Mid-run: the status carries the file the (mock) import is
    processing — assigned through the runner's own hook closure; during
    the clone/pull phase (A4) no file is reported yet. The success
    terminal clears ``current_file`` but keeps the final counts."""
    clone = _GatedClone(threading.Event(), threading.Event())
    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)

    thread, errors = _start_run()
    try:
        assert clone.started.wait(5.0), "the run never reached the clone phase"
        # Clone/pull prelude: running — but no file yet (A4: bare
        # "Syncing…") and no phase yet either (D1: the prelude stays
        # null — the bare label's pin), and no chunk position yet
        # (phase 129: 0/0 before the first embed batch).
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] is None
        assert s["current_file"] is None
        assert s["current_summary"] is None
        assert s["files_done"] == 0 and s["files_total"] == 0
        assert s["summaries_done"] == 0 and s["summaries_total"] == 0
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        clone.release.set()
        assert fake_import.started.wait(5.0), "the run never reached the import"
        # Import phase: the hook's file is live on the status, and the
        # phase machine names the phase (the summary span has not
        # started — its counters are still 0/0); no embed batch has
        # fired the chunk hook (this canned import passes no pairs),
        # so the chunk position stays 0/0.
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] == "import"
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["files_done"] == 1
        assert s["files_total"] == 3
        assert s["current_summary"] is None
        assert s["summaries_done"] == 0 and s["summaries_total"] == 0
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        fake_import.release.set()
    finally:
        clone.release.set()
        fake_import.release.set()
        thread.join(10.0)
    assert not thread.is_alive()
    assert errors == [], f"the runner raised: {errors!r}"
    # Wiring: prune=True is preserved, and the hook fired through the
    # runner's own closure (the recorded call is what the closure
    # assigned to the status above).
    assert fake_import.prune_flags == [True]
    assert fake_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]
    # Success terminal: current_file null, final file counts retained;
    # the phase machine clears phase + current_summary — the fake
    # folder step never fired the summary hook, so the summary counts
    # stay 0/0 (the keep-final-counts convention keeps what fired);
    # the chunk position clears with the file (phase 129 — it is a
    # sub-position of a done file, never a keep-final counter).
    s = sync_api.sync_status()
    assert s["state"] == "success"
    assert s["phase"] is None
    assert s["current_summary"] is None
    assert s["current_file"] is None
    assert s["files_done"] == 1 and s["files_total"] == 3
    assert s["summaries_done"] == 0 and s["summaries_total"] == 0
    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
    assert s["error"] is None
    assert s["started_at"] is not None and s["finished_at"] is not None


def test_failed_terminal_clears_current_file_keeps_counts(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Terminal (failed): a run that dies inside the import clears
    ``current_file`` but keeps the hook's final counts — the last
    position is useful context next to the (sanitized) error.

    Phase 129 (task 03): the convention split extends to the chunk
    position — the (about-to-fail) import reports a live pair
    (1/38), the failed terminal clears it to 0/0 (a sub-position of a
    file that's done/abandoned), while the FILE counts keep their
    final values next to the error."""
    clone = _GatedClone(threading.Event(), threading.Event())
    fake_import = _GatedImport(
        ImportSummary(),
        threading.Event(),
        threading.Event(),
        fail=EmbeddingError(
            "embeddings request to https://u:p@aipi.example.com/v1 "
            "failed: connection refused"
        ),
        chunk_pairs=[(1, 38)],  # phase 129: the live pair before the raise
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)

    thread, errors = _start_run()
    try:
        assert clone.started.wait(5.0), "the run never reached the clone phase"
        clone.release.set()
        assert fake_import.started.wait(5.0), "the chunk pair never landed"
        # While the (about-to-fail) import is parked: the file is live,
        # and so is its chunk position (the pair the fake fired through
        # the runner's own closure).
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["files_done"] == 1 and s["files_total"] == 3
        assert s["current_chunk"] == 1 and s["chunks_total"] == 38
        fake_import.release.set()
    finally:
        clone.release.set()
        fake_import.release.set()
        thread.join(10.0)
    assert not thread.is_alive()
    assert errors == []
    # Wiring: the pair fired through the runner's own closure.
    assert fake_import.chunk_hook_calls == [(1, 38)]
    s = sync_api.sync_status()
    assert s["state"] == "failed"
    assert s["current_file"] is None  # cleared in the terminal state
    assert s["phase"] is None  # phase 98: cleared in the terminal state
    assert s["current_summary"] is None
    assert s["files_done"] == 1 and s["files_total"] == 3  # final counts kept
    assert s["summaries_done"] == 0 and s["summaries_total"] == 0
    # Phase 129 (task 03): the chunk position clears in the FAILED
    # terminal too — the convention split, pinned (the FILE counts
    # above keep their final values, the sub-position does not).
    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
    assert s["detail"] == {}
    error = s["error"] or ""
    assert "*****@aipi.example.com" in error  # credentials masked
    assert "u:p" not in error
    assert "connection refused" in error  # the reason survives


# ---------- phase 129 (task 03): the in-flight file's embed-chunk ----------
# ---------- position (current_chunk / chunks_total) ----------------------
#
# Same seams as the phase-64 block above: the fake import fires the
# runner's OWN ``_chunk_hook`` closure (the task under test) with the
# file's chunk-position pairs, parking between pairs so the test reads
# the status BETWEEN ticks (the A6 batch-granularity contract: 14 =
# the first chunk of the in-flight token-budgeted batch).


def _poll_until(
    predicate: Callable[[], bool],
    timeout: float = 5.0,
    what: str = "condition",
) -> None:
    """Poll (main thread) until *predicate* — the worker-thread run's
    state is only observable between its parks."""
    deadline = time.monotonic() + timeout
    while not predicate():
        if time.monotonic() >= deadline:
            raise AssertionError(f"timed out waiting for {what}")
        time.sleep(0.01)


def test_mid_run_status_reports_chunk_position_between_ticks(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Phase 129 (task 03): the status carries the in-flight file's
    embed-chunk position — the hook's (first, total) pair climbs
    through the file's batches (1/38 → 13/38, read between ticks),
    paired with the file the ``progress`` hook named; the prelude
    reports 0/0 (no embed batch yet), and the success terminal clears
    the position to 0/0 while the FILE counts keep their final values
    (the convention split — the chunk position is a sub-position of a
    done file, never a keep-final counter)."""
    clone = _GatedClone(threading.Event(), threading.Event())
    import_release = threading.Event()
    chunk_release = threading.Event()
    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        import_release,
        chunk_pairs=[(1, 38), (13, 38)],  # two batches of a 38-chunk file
        chunk_release=chunk_release,
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)

    thread, errors = _start_run()
    try:
        assert clone.started.wait(5.0), "the run never reached the clone phase"
        # Prelude: the chunk position is 0/0 (no file, no batch yet —
        # alongside the bare-label pins of the phase-64 test above).
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["current_file"] is None
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        clone.release.set()
        assert fake_import.started.wait(5.0), "the first chunk pair never landed"
        # Import phase, between ticks: pair one is live on the status,
        # paired with the file the progress hook named (the two hooks
        # are paired — the position belongs to that file).
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] == "import"
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["files_done"] == 1 and s["files_total"] == 3
        assert s["current_chunk"] == 1 and s["chunks_total"] == 38
        # Release pair one: the import fires pair two (13/38 — the
        # second batch of the 38-chunk file) and parks again.
        import_release.set()
        _poll_until(
            lambda: sync_api.sync_status()["current_chunk"] == 13,
            what="the second chunk pair on the status",
        )
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] == "import"
        # The position CLIMBED (batch granularity) — total unchanged
        # (the file's chunk count, not the batch size).
        assert s["current_chunk"] == 13 and s["chunks_total"] == 38
        assert s["current_file"] == "repo/notes/deep.md"
        assert s["files_done"] == 1 and s["files_total"] == 3
        chunk_release.set()
    finally:
        clone.release.set()
        import_release.set()
        chunk_release.set()
        thread.join(10.0)
    assert not thread.is_alive()
    assert errors == [], f"the runner raised: {errors!r}"
    # Wiring: the pairs fired through the runner's OWN closure (the
    # status reads above came through it; a wiring regression to
    # None would have skipped the firing branch and timed out).
    assert fake_import.chunk_hook_calls == [(1, 38), (13, 38)]
    # Success terminal: the chunk position clears (0/0) ...
    s = sync_api.sync_status()
    assert s["state"] == "success"
    assert s["phase"] is None
    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
    # ... while the FILE counts keep their final values (the
    # convention split, pinned alongside the phase-64 terminal pin).
    assert s["current_file"] is None
    assert s["files_done"] == 1 and s["files_total"] == 3


def test_chunk_position_clears_at_phase_transitions_and_terminal(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Phase 129 (task 03): the chunk position is CLEARED (0/0) at
    every point the import phase ends. This run's summary changes the
    KB, so BOTH post-import transitions run and are pinned: the
    import→``"overview"`` transition and the import→``"summaries"``
    transition (the changed-KB branch) each clear it to 0/0, and the
    success terminal keeps it 0/0 — through the whole span the FILE
    counts keep their final values (the convention split: the file
    counts survive for the result line's context, the chunk position
    is a sub-position of a file that's done)."""
    clone = _GatedClone(threading.Event(), threading.Event())
    import_release = threading.Event()
    overview_started = threading.Event()
    overview_release = threading.Event()
    folders_started = threading.Event()
    folders_release = threading.Event()
    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        import_release,
        chunk_pairs=[(7, 38)],  # a live pair to clear at each transition
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)

    # Park the post-import steps (re-monkeypatched over the seams
    # helper's no-ops): the test reads the status DURING the
    # "overview" phase and DURING the "summaries" phase — the two
    # import-phase-ending transitions the task pins.
    async def gated_overview(llm: object, session: object = None) -> bool:
        overview_started.set()
        await asyncio.to_thread(overview_release.wait)
        return True

    async def gated_folders(
        db: object,
        llm: object,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        folders_started.set()
        await asyncio.to_thread(folders_release.wait)
        return {"generated": 0, "failed": 0, "pruned": 0}

    monkeypatch.setattr(sync_api, "regenerate_overview", gated_overview)
    monkeypatch.setattr(sync_api, "generate_folder_summaries", gated_folders)

    thread, errors = _start_run()
    try:
        assert clone.started.wait(5.0), "the run never reached the clone phase"
        clone.release.set()
        assert fake_import.started.wait(5.0), "the chunk pair never landed"
        # Mid-import: the live pair (7/38) is on the status.
        s = sync_api.sync_status()
        assert s["phase"] == "import"
        assert s["current_chunk"] == 7 and s["chunks_total"] == 38
        # Release the import: it returns (added+updated > 0) → the
        # phase machine moves to "overview" and the position clears.
        import_release.set()
        assert overview_started.wait(5.0), "the run never reached the overview"
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] == "overview"
        # CLEARED at the import→overview transition (0/0) ... the file
        # counts KEEP their final values (the convention split).
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        assert s["files_done"] == 1 and s["files_total"] == 3
        overview_release.set()
        # ... and the changed-KB branch moves to "summaries" (the
        # second import-phase-ending transition — cleared there too,
        # idempotently) before parking.
        assert folders_started.wait(5.0), "the run never reached the summaries"
        s = sync_api.sync_status()
        assert s["state"] == "running"
        assert s["phase"] == "summaries"
        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
        assert s["files_done"] == 1 and s["files_total"] == 3
        folders_release.set()
    finally:
        clone.release.set()
        import_release.set()
        overview_release.set()
        folders_release.set()
        thread.join(10.0)
    assert not thread.is_alive()
    assert errors == [], f"the runner raised: {errors!r}"
    assert fake_import.chunk_hook_calls == [(7, 38)]
    # Success terminal: still 0/0 (the terminal clears too — same
    # spots as current_file), file counts kept.
    s = sync_api.sync_status()
    assert s["state"] == "success"
    assert s["phase"] is None
    assert s["current_file"] is None
    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
    assert s["files_done"] == 1 and s["files_total"] == 3


# ---------- phase 134 (task 01): the cooperative cancel (5th state) ----------
#
# The cancel runs are driven through the REAL ``start_sync`` on a
# worker thread's own event loop — ``sync_api._task`` is the running
# task (the cancel endpoint's target), unlike the direct-``_run_sync``
# tests above. Two driver shapes:
#
# * ``_run_via_start_sync`` — the ACTION runs ON the loop thread once
#   the run is in flight (a same-thread ``Task.cancel()`` — a
#   cross-thread cancel would schedule through the loop's selector
#   wake, which a parked loop never receives); the loop is drained
#   afterwards so the runner's terminal step (the
#   ``except CancelledError`` conversion) lands before the test reads
#   the status.
# * ``_start_run_via_api`` — the OLD house pattern (blocking threading
#   gates; the loop thread may park inside a sync span — the
#   clone/``file_commit_dates``) with the main thread orchestrating
#   the event set / releases.


def _fast_clone(url: str, dest: Path | str, branch: str | None = None) -> Path:
    """A non-gated ``clone_or_pull`` — the loop thread must STAY FREE
    for the loop-thread action, so nothing in the prelude may block it
    (the import's gate is an ``asyncio.to_thread`` park — loop-safe).
    Phase 135: accepts the row's branch pin (the third argument —
    ``None`` for this file's rows)."""
    dest = Path(dest)
    dest.mkdir(parents=True, exist_ok=True)
    (dest / "notes.md").write_text("# repo\ncontent for the KB\n", encoding="utf-8")
    return dest


def _run_via_start_sync(
    action: Callable[[asyncio.AbstractEventLoop], Any],
) -> tuple[threading.Thread, list[BaseException], dict[str, Any]]:
    """Start a run through the real ``start_sync`` on a worker thread's
    own event loop; once the run is in flight, run *action* ON that
    loop thread, then drain the loop so the runner's terminal step
    (the ``except CancelledError`` conversion) lands before the test
    reads the status. Returns (thread, the exceptions the run
    surfaced — the CancelledError of a converted cancel, the action's
    result under ``"result"``)."""
    errors: list[BaseException] = []
    out: dict[str, Any] = {}
    started = threading.Event()

    def _run() -> None:
        loop = asyncio.new_event_loop()
        try:
            asyncio.set_event_loop(loop)
            loop.run_until_complete(sync_api.start_sync())
            started.set()
            out["result"] = action(loop)
            for _ in range(20):
                if sync_api._task is not None and sync_api._task.done():
                    break
                loop.run_until_complete(asyncio.sleep(0))
            assert sync_api._task is not None and sync_api._task.done(), (
                "the runner task never finished"
            )
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)
        finally:
            asyncio.set_event_loop(None)
            loop.close()

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    assert started.wait(5.0), "start_sync never ran"
    return thread, errors, out


def _start_run_via_api() -> tuple[threading.Thread, list[BaseException]]:
    """The house background-task pattern over the real ``start_sync`` —
    the worker loop thread may park inside a SYNC span (the blocking
    clone / ``file_commit_dates`` park); the main thread orchestrates
    the event set / gate releases. The runner's surfaced exception
    (the CancelledError of a converted cancel) is captured."""
    errors: list[BaseException] = []

    def _run() -> None:
        loop = asyncio.new_event_loop()
        try:
            asyncio.set_event_loop(loop)
            loop.run_until_complete(sync_api.start_sync())
            if sync_api._task is not None:
                loop.run_until_complete(sync_api._task)
        except BaseException as e:  # noqa: BLE001 — surfaced to the test
            errors.append(e)
        finally:
            asyncio.set_event_loop(None)
            loop.close()

    thread = threading.Thread(target=_run, daemon=True)
    thread.start()
    return thread, errors


def test_cancel_is_409_idle_and_after_a_terminal_run(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The 409 contract (the exact detail ``no sync is running``):
    idle (never started — ``_task`` None) AND after a terminal run
    (``_task`` done) — the harmless race the UI's poll settles (no
    error copy, no state change)."""
    fake_import = _GatedImport(
        ImportSummary(files=1, added=1), threading.Event(), threading.Event()
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, _fast_clone)

    # Idle (never started): 409, the state untouched.
    with pytest.raises(HTTPException) as idle:
        asyncio.run(sync_api.cancel_sync())
    assert idle.value.status_code == 409
    assert idle.value.detail == "no sync is running"
    assert sync_api.sync_status()["state"] == "idle"

    # A full (uncancelled) run to the success terminal ...
    def act(loop: asyncio.AbstractEventLoop) -> None:
        assert fake_import.started.wait(5.0), "the run never reached the import"
        fake_import.release.set()
        task = sync_api._task
        assert task is not None
        loop.run_until_complete(task)

    thread, errors, out = _run_via_start_sync(act)
    thread.join(10.0)
    assert not thread.is_alive()
    assert errors == [], f"the runner raised: {errors!r}"
    assert sync_api.sync_status()["state"] == "success"

    # ... and after a terminal run the cancel is 409 again.
    with pytest.raises(HTTPException) as after:
        asyncio.run(sync_api.cancel_sync())
    assert after.value.status_code == 409
    assert after.value.detail == "no sync is running"
    # The state is untouched by the rejected cancel.
    assert sync_api.sync_status()["state"] == "success"


def test_cancel_endpoint_stops_the_in_flight_run(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The 202 path, end to end at the function level: a started run
    (parked inside the import) + ``cancel_sync`` → 202 ``cancel
    requested``; the event is set AND the task is cancelled, the run
    converts the ``CancelledError`` into the ``cancelled`` terminal —
    ``error`` None, ``detail`` empty, ``finished_at`` set, the FINAL
    file counts kept (the hook fired 1/3 before the park), the file /
    phase / summary / chunk positions cleared — the PLAN §9 line is
    logged, and the task ends CANCELLED (the re-raise; the status
    object stays authoritative)."""
    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, _fast_clone)

    def act(loop: asyncio.AbstractEventLoop) -> dict[str, str]:
        assert fake_import.started.wait(5.0), "the run never reached the import"
        return loop.run_until_complete(sync_api.cancel_sync())

    try:
        with caplog.at_level(logging.INFO, logger="app.api.sync"):
            thread, errors, out = _run_via_start_sync(act)
            thread.join(10.0)
    finally:
        fake_import.release.set()  # free the to_thread park thread (non-daemon)
    assert not thread.is_alive()
    assert out.get("result") == {"detail": "cancel requested"}
    # The endpoint did BOTH: the event (the cooperative marker) ...
    assert sync_api._cancel_event.is_set()
    # ... and the task cancel (the task ended CANCELLED — the re-raise
    # happened; a swallowed cancel would have ended the task plain).
    # The CancelledError itself is stored in the task (nothing awaits
    # it directly — start_sync's create_task), so no thread-level
    # exception surfaces — and nothing ELSE may have either.
    assert sync_api._task is not None
    assert sync_api._task.cancelled()
    assert errors == [], f"the runner raised: {errors!r}"
    s = sync_api.sync_status()
    assert s["state"] == "cancelled"
    assert s["error"] is None  # intent, not failure
    assert s["detail"] == {}
    assert s["started_at"] is not None and s["finished_at"] is not None
    assert s["current_file"] is None
    assert s["phase"] is None
    assert s["current_summary"] is None
    assert s["files_done"] == 1 and s["files_total"] == 3  # final counts kept
    assert s["summaries_done"] == 0 and s["summaries_total"] == 0
    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
    # PLAN §9: the ample-logging line for the converted cancel.
    assert any(
        r.getMessage()
        == "sync: cancelled by user state=cancelled files_done=1 files_total=3"
        for r in caplog.records
    )


def test_unrequested_cancel_is_never_swallowed_and_never_converts(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The shutdown contract pin (the completion criterion): a bare
    ``task.cancel()`` WITHOUT the event set (the app-shutdown path —
    no one pressed Cancel) re-raises the ``CancelledError`` with the
    state UNTOUCHED — the task dies mid-run in ``running`` (no
    terminal state, exactly as today's shutdown path does), the event
    stays unset, and nothing is swallowed."""
    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, _fast_clone)

    def act(loop: asyncio.AbstractEventLoop) -> None:
        assert fake_import.started.wait(5.0), "the run never reached the import"
        assert sync_api._cancel_event.is_set() is False
        task = sync_api._task
        assert task is not None
        task.cancel()  # NO endpoint, NO event — the shutdown path
        loop.run_until_complete(task)  # raises the CancelledError

    try:
        thread, errors, out = _run_via_start_sync(act)
        thread.join(10.0)
    finally:
        fake_import.release.set()  # free the to_thread park thread (non-daemon)
    assert not thread.is_alive()
    assert "result" not in out  # the action raised
    assert len(errors) == 1
    assert isinstance(errors[0], asyncio.CancelledError)
    s = sync_api.sync_status()
    assert s["state"] == "running"  # the task died mid-run — no terminal
    assert s["finished_at"] is None
    assert s["error"] is None
    assert s["detail"] == {}
    assert sync_api._cancel_event.is_set() is False


def test_check_cancelled_raises_only_when_the_event_is_set(
    fresh_sync_status: None,
) -> None:
    """The boundary helper is pure polling: a no-op while the event is
    unset, ``asyncio.CancelledError`` when it is set (the run's
    ``except CancelledError`` clause converts THAT into the
    ``cancelled`` terminal)."""
    assert sync_api._cancel_event.is_set() is False
    sync_api._check_cancelled()  # unset: no raise
    sync_api._cancel_event.set()
    with pytest.raises(asyncio.CancelledError):
        sync_api._check_cancelled()


def test_fresh_run_clears_a_stale_cancel_event(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The stale-event pin: run ONE is cancelled (the terminal keeps
    the event SET — only start_sync clears it), then a FRESH run
    starts → ``start_sync`` cleared the event before the task was
    created → the new run completes to ``success`` with no cancel
    request (a stale event cannot bleed into the next run)."""
    first_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    _patch_sync_seams(monkeypatch, tmp_path, first_import, _fast_clone)

    def act1(loop: asyncio.AbstractEventLoop) -> dict[str, str]:
        assert first_import.started.wait(5.0), "run 1 never reached the import"
        return loop.run_until_complete(sync_api.cancel_sync())

    try:
        thread, errors, out = _run_via_start_sync(act1)
        thread.join(10.0)
    finally:
        first_import.release.set()  # free the to_thread park thread
    assert not thread.is_alive()
    assert errors == [], f"run 1 raised: {errors!r}"
    task1 = sync_api._task
    assert task1 is not None
    assert task1.cancelled()  # the re-raise ended it cancelled
    assert out.get("result") == {"detail": "cancel requested"}
    assert sync_api.sync_status()["state"] == "cancelled"
    assert sync_api._cancel_event.is_set()  # the terminal keeps the event set

    # The FRESH run — a new gated import (the first one's gates are
    # spent), the same fast clone.
    second_import = _GatedImport(
        ImportSummary(files=1, added=1), threading.Event(), threading.Event()
    )
    monkeypatch.setattr(sync_api, "import_sources", second_import)

    def act2(loop: asyncio.AbstractEventLoop) -> None:
        assert second_import.started.wait(5.0), "run 2 never reached the import"
        # start_sync cleared the stale event before the task was
        # created — by the import park it is definitely clear.
        assert sync_api._cancel_event.is_set() is False
        second_import.release.set()
        task = sync_api._task
        assert task is not None
        loop.run_until_complete(task)

    try:
        thread2, errors2, out2 = _run_via_start_sync(act2)
        thread2.join(10.0)
    finally:
        second_import.release.set()  # free the to_thread park thread
    assert not thread2.is_alive()
    assert errors2 == [], f"the fresh run raised: {errors2!r}"
    s = sync_api.sync_status()
    assert s["state"] == "success"  # no cancel request — clean completion


def test_event_alone_cancels_at_the_import_boundary(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The cooperative path WITHOUT ``task.cancel()``: the event is set
    while the run is parked in the clone prelude (simulating a
    registered cancel whose ``_task.cancel()`` hit a blocking
    subprocess and cannot interrupt it) — the span runs to completion
    and the NEXT boundary (immediately before ``import_sources``)
    converts it: the import never runs (the hook never fired), the
    state is ``cancelled``, the counts stay 0/0."""
    clone_parked = threading.Event()
    clone_release = threading.Event()
    clone_calls: list[tuple[str, Path]] = []

    def parking_clone(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)
        (dest / "notes.md").write_text("# repo\ncontent for the KB\n", encoding="utf-8")
        clone_calls.append((url, dest))
        clone_parked.set()
        clone_release.wait()  # blocking is fine: the clone is a sync call
        return dest

    fake_import = _GatedImport(
        ImportSummary(files=3, added=1, updated=1, unchanged=1),
        threading.Event(),
        threading.Event(),
    )
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, parking_clone)

    thread, errors = _start_run_via_api()
    try:
        assert clone_parked.wait(5.0), "the run never reached the clone"
        # The event ALONE (no task.cancel — the subprocess cannot be
        # interrupted): the cooperative boundary must catch it.
        sync_api._cancel_event.set()
    finally:
        clone_release.set()
    thread.join(10.0)
    assert not thread.is_alive()
    assert clone_calls == [("https://git.example.com/repo.git", tmp_path / "bor" / "repo")]
    assert len(errors) == 1
    assert isinstance(errors[0], asyncio.CancelledError)
    assert fake_import.hook_calls == []  # the import never ran
    s = sync_api.sync_status()
    assert s["state"] == "cancelled"
    assert s["error"] is None
    assert s["detail"] == {}
    assert s["files_done"] == 0 and s["files_total"] == 0
    assert s["current_file"] is None
    assert s["phase"] is None


def test_event_set_during_import_cancels_before_the_lite_spans(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The overview + summaries boundary: the cancel's event lands
    WHILE the import span runs (the fake import sets it mid-call —
    the endpoint would have set it too) — the run converts at the
    boundary immediately before ``if summary.added + summary.updated
    > 0:``: the overview AND the folder summaries never start, the
    state is ``cancelled``, and the import's final file counts are
    kept."""
    overview_calls: list[Any] = []
    folder_calls: list[Any] = []

    class _CancelDuringImport:
        def __init__(self, summary: ImportSummary) -> None:
            self.summary = summary
            self.calls = 0

        async def __call__(
            self,
            sources: list[Path],
            llm: object,
            *,
            prune: bool = False,
            limit: int | None = None,
            session: object = None,
            progress: Callable[[str, str, int, int], None] | None = None,
            on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
            ignore_by_root: dict[str, list[str]] | None = None,  # phase 89
            include_hidden_by_root: dict[str, bool] | None = None,  # phase 105
            doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,  # phase 106
        ) -> ImportSummary:
            self.calls += 1
            if progress is not None:
                progress("repo", "notes/deep.md", 1, 2)
            # The user's cancel lands mid-import (the endpoint's
            # _task.cancel() would interrupt here at the next per-file
            # await; the event ALONE isolates the boundary check — the
            # next boundary is the overview gate).
            sync_api._cancel_event.set()
            return self.summary

    fake_import = _CancelDuringImport(ImportSummary(files=2, added=1, updated=0))
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, _fast_clone)

    async def spy_overview(llm: object, session: object = None) -> bool:
        overview_calls.append(llm)
        return False

    async def spy_folders(
        db: object,
        llm: object,
        *,
        skip: bool = False,
        only_missing: bool = False,
        changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
        on_progress: Callable[[int, int, str, str], None] | None = None,
    ) -> dict[str, int]:
        folder_calls.append(llm)
        return {"generated": 0, "failed": 0, "pruned": 0}

    monkeypatch.setattr(sync_api, "regenerate_overview", spy_overview)
    monkeypatch.setattr(sync_api, "generate_folder_summaries", spy_folders)

    thread, errors = _start_run_via_api()
    thread.join(10.0)
    assert not thread.is_alive()
    assert fake_import.calls == 1  # the import itself ran ...
    assert len(errors) == 1
    assert isinstance(errors[0], asyncio.CancelledError)
    assert overview_calls == []  # ... but the overview never started
    assert folder_calls == []  # ... and neither did the folder summaries
    s = sync_api.sync_status()
    assert s["state"] == "cancelled"
    assert s["error"] is None
    assert s["detail"] == {}
    assert s["files_done"] == 1 and s["files_total"] == 2  # final counts kept
    assert s["current_file"] is None
    assert s["phase"] is None
    assert s["current_summary"] is None


def test_event_cancels_at_the_top_of_the_source_loop(
    fresh_sync_status: None,
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """The per-source boundary (two git rows): the event is set while
    the FIRST source's post-clone date walk runs (a blocking span —
    the cooperative consequence: it runs to completion) — the cancel
    converts at the top of iteration TWO, before the second source's
    clone/pull: exactly one clone ran, the import never started, the
    state is ``cancelled``."""
    parked = threading.Event()
    release = threading.Event()
    clone_calls: list[tuple[str, Path]] = []

    def parking_dates(root: Path) -> dict[str, datetime]:
        parked.set()
        release.wait()  # blocking is fine: the date walk is a sync call
        return {}

    def fast_clone(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)
        (dest / "notes.md").write_text("# repo\ncontent for the KB\n", encoding="utf-8")
        clone_calls.append((url, dest))
        return dest

    class _NeverReachedImport:
        def __init__(self) -> None:
            self.calls = 0

        async def __call__(self, *args: Any, **kwargs: Any) -> ImportSummary:
            self.calls += 1
            return ImportSummary()

    fake_import = _NeverReachedImport()
    _patch_sync_seams(monkeypatch, tmp_path, fake_import, fast_clone)
    monkeypatch.setattr(sync_api, "file_commit_dates", parking_dates)
    # Two git rows: the cancel must fire at the top of iteration TWO
    # (after the first source's git ops ran to completion).
    monkeypatch.setattr(
        sync_api,
        "effective_sources",
        lambda session: (
            [
                GitSource(url="https://git.example.com/one.git", kind="git"),
                GitSource(url="https://git.example.com/two.git", kind="git"),
            ],
            "db",
        ),
    )

    thread, errors = _start_run_via_api()
    try:
        assert parked.wait(5.0), "the run never reached the first date walk"
        sync_api._cancel_event.set()
    finally:
        release.set()
    thread.join(10.0)
    assert not thread.is_alive()
    assert len(errors) == 1
    assert isinstance(errors[0], asyncio.CancelledError)
    # Exactly the first source's git ops ran — the second clone never
    # started (the boundary fired before it).
    assert clone_calls == [("https://git.example.com/one.git", tmp_path / "bor" / "one")]
    assert fake_import.calls == 0
    s = sync_api.sync_status()
    assert s["state"] == "cancelled"
    assert s["error"] is None
    assert s["detail"] == {}
    assert s["files_done"] == 0 and s["files_total"] == 0
    assert s["current_file"] is None


# ---------- phase 134 (task 02): the frontend cancel contract ----------
#
# The browser behavior is E2E-covered (tests/e2e/test_sync_cancel.py,
# task 03); here the source-level contract is pinned the same way as
# the main button above — so a silent regression (a hidden leak, a
# branch-order swap, a stale re-enable) is caught without a browser.


def test_sync_cancel_button_ships_hidden_and_labeled() -> None:
    """#sync-cancel-btn SHIPS hidden (anonymous-safe + the idle state
    — sources.js reveals it the same way it reveals #sync-btn) in the
    sync-control row right after the main button, with the visible
    "Cancel" label (the accessible name — WCAG 2.4, never
    icon-only)."""
    text = _text(SHELL_HTML)
    m = re.search(r"<button\b([^>]*)id=\"sync-cancel-btn\"([^>]*)>([^<]*)</button>", text)
    assert m, "the shell must carry the #sync-cancel-btn (RAG view)"
    attrs = m.group(1) + m.group(2)
    assert 'class="sync-cancel-btn"' in attrs
    assert 'type="button"' in attrs
    assert re.search(r"\bhidden\b", attrs), "#sync-cancel-btn must ship hidden"
    assert m.group(3).strip() == "Cancel", "the visible label is the accessible name"
    # Same row: right after the main button closes, before the announcer.
    main_end = text.find("</button>", text.find('id="sync-btn"'))
    cancel_start = text.find('id="sync-cancel-btn"')
    assert -1 < main_end < cancel_start < text.find('id="sync-result"')


def test_sync_cancel_btn_css_secondary_pill_and_states() -> None:
    """.sync-cancel-btn is the house SECONDARY pill (the
    .kb-summary-cancel language): >=44px touch floor, --line border,
    the AA ink-soft text pair on the surface, the global 3px
    :focus-visible ring (no per-button override), the :disabled look
    (the request in flight — one request at a time), and the [hidden]
    rule that keeps the attribute authoritative over the display. The
    hover is guarded on :not(:disabled) — a disabled button under the
    cursor never reads as actionable."""
    css = _text(STYLES_CSS)
    block = re.search(r"\.sync-cancel-btn\s*\{([^}]*)\}", css)
    assert block, "styles.css must define .sync-cancel-btn"
    body = block.group(1)
    for prop in (
        "display: inline-flex",
        "min-height: 44px",
        "border-radius: 999px",
        "border: 1px solid var(--line)",
        "color: var(--ink-soft)",
    ):
        assert prop in body, f".sync-cancel-btn must carry {prop!r}"
    hover = re.search(r"\.sync-cancel-btn:hover:not\(:disabled\)\s*\{([^}]*)\}", css)
    assert hover, "the hover must be guarded on :not(:disabled)"
    disabled = re.search(r"\.sync-cancel-btn:disabled\s*\{([^}]*)\}", css)
    assert disabled, ".sync-cancel-btn:disabled must be styled"
    assert "opacity: 0.6" in disabled.group(1)
    hidden_rule = re.search(r"\.sync-cancel-btn\[hidden\]\s*\{([^}]*)\}", css)
    assert hidden_rule, "the [hidden] rule must beat the display"
    assert "display: none" in hidden_rule.group(1)


def test_sync_cancel_btn_css_full_width_on_mobile() -> None:
    """≤640px: the sync-control row already wraps (the 360px case) —
    the Cancel pill stacks full-width under the full-width Sync pill
    (the same mobile rule as the main button, next to it)."""
    css = _text(STYLES_CSS)
    main = css.find(".sync-btn { width: 100%; }")
    assert main != -1, "the main button's mobile full-width rule"
    near = css[main : main + 300]
    assert ".sync-cancel-btn { width: 100%; }" in near, (
        "the Cancel pill must join the main button's mobile full-width rule"
    )


def test_sources_js_owns_the_cancel_button_and_helpers() -> None:
    """The control ref is null-safe like the main button's, and the
    show/hide helpers carry the exact contract: show = unhide +
    enable + aria-label "Cancel sync" (the stable accessible name);
    hide = hide + disable (a hidden control is never focusable or
    activatable in any state — the never-stale guarantee)."""
    js = _text(SOURCES_JS)
    assert 'querySelector("#sync-cancel-btn")' in js
    show = _body(js, "showSyncCancelBtn")
    assert "syncCancelBtn.hidden = false" in show
    assert "syncCancelBtn.disabled = false" in show
    assert 'syncCancelBtn.setAttribute("aria-label", "Cancel sync")' in show
    hide = _body(js, "hideSyncCancelBtn")
    assert "syncCancelBtn.hidden = true" in hide
    assert "syncCancelBtn.disabled = true" in hide


def test_sources_js_cancel_visible_only_while_a_sync_runs() -> None:
    """The show/hide contract's two choke points: enterSyncRunningState
    shows the control for kind "sync" (the live click, the re-attach,
    every running tick) and hides it for kind "upload" (the upload job
    is NOT cancellable in this phase — its run keeps the control
    hidden); settleSyncButton hides it on EVERY settle (every
    terminal + both idles pass through it)."""
    js = _text(SOURCES_JS)
    enter = _body(js, "enterSyncRunningState")
    assert 'if (kind === "sync") showSyncCancelBtn()' in enter
    assert "else hideSyncCancelBtn();" in enter
    settle = _body(js, "settleSyncButton")
    assert "hideSyncCancelBtn()" in settle, (
        "every settle (terminal + idle) must hide the Cancel control"
    )


def test_sources_js_cancel_request_is_never_stale() -> None:
    """requestCancel: the control disables IMMEDIATELY (before the
    fetch — one request at a time, no double fire) and #sync-result
    clears with it; then the branch set — 403 the whoami backstop
    (stop polling, hide the main button, settle idle — the same depth
    as every other 403 in this module), 409 the silent no-op (the
    poll settles), 202 (r.ok) nothing (the poll's cancelled branch
    settles everything), any other non-2xx / network failure the
    phase-55 neutral retry copy + re-enable (the run is still going —
    the owner can retry)."""
    js = _text(SOURCES_JS)
    body = _body(js, "requestCancel")
    disable = body.find("syncCancelBtn.disabled = true")
    fetch = body.find('fetch("/api/sync/cancel", { method: "POST" })')
    assert disable != -1 and fetch != -1 and disable < fetch, (
        "the disable must land BEFORE the fetch (never stale)"
    )
    assert 'syncResult.textContent = ""' in body, "the result line clears with the click"
    assert "r.status === 403" in body
    assert "stopSyncPolling()" in body
    assert "syncBtn.hidden = true" in body
    assert "applySyncIdle()" in body
    assert "r.status === 409" in body
    assert "r.ok" in body
    # The phase-55 neutral retry copy + the re-enable — in the
    # network-failure catch AND the non-2xx fall-through.
    assert body.count("Couldn't cancel the sync — try again.") == 2
    assert body.count("syncCancelBtn.disabled = false") == 2


def test_sources_js_wires_the_cancel_click_next_to_the_main() -> None:
    """The click listener is armed right after the main button's
    (guarded on the control's presence — a page without the markup is
    a complete no-op, the main button's convention)."""
    js = _text(SOURCES_JS)
    main = js.find('syncBtn.addEventListener("click", startSync)')
    cancel = js.find('syncCancelBtn.addEventListener("click", requestCancel)')
    assert main != -1 and cancel != -1
    assert cancel > main, "the cancel wiring sits next to the main button's"


def test_sources_js_poll_tree_cancelled_branch_in_order() -> None:
    """The poll tick's decision tree (now 8 branches) gains the
    cancelled branch AFTER the failed branch and BEFORE the
    upload-success branch: stop polling + applySyncCancelled — the
    neutral terminal, not the failure path."""
    js = _text(SOURCES_JS)
    start = js.find("function startSyncPolling")
    end = js.find("async function startSync", start)
    tick = js[start:end]
    failed = tick.find('syncStatus.state === "failed"')
    cancelled = tick.find('syncStatus.state === "cancelled"')
    upload_success = tick.find('uploadStatus.state === "success"')
    assert -1 < failed < cancelled < upload_success, (
        "the cancelled branch must sit between failed and upload success"
    )
    branch = tick[cancelled : cancelled + 300]
    assert "stopSyncPolling()" in branch
    assert "applySyncCancelled(syncStatus)" in branch


def test_sources_js_apply_sync_cancelled_is_neutral() -> None:
    """applySyncCancelled renders the NEUTRAL result (A1 — cancel is
    intent, not failure): settle the main button retry-ready (which
    also hides the Cancel control), the "Sync cancelled" line on the
    aria-live result (superseding the settle's label write), the
    error banner hidden (a failed run's banner must not survive), the
    status frame emitted — NO modal, NO banner, NO error styling."""
    js = _text(SOURCES_JS)
    body = _body(js, "applySyncCancelled")
    assert 'settleSyncButton("Sync sources")' in body
    assert 'syncResult.textContent = "Sync cancelled"' in body
    assert "hideSyncError()" in body
    assert "emitSyncStatus(status)" in body
    assert "showSyncModal" not in body, "cancel is intent — no modal"
    assert "showSyncError" not in body, "cancel is intent — no banner"
    assert "is-error" not in body, "cancel is intent — no error styling"


def test_sources_js_reattaches_a_cancelled_run() -> None:
    """initSyncButton: a reload during/right after a cancel renders
    the neutral cancelled result (the branch sits after the failed
    branch, before the upload check) — never a stuck running state."""
    js = _text(SOURCES_JS)
    body = _body(js, "initSyncButton")
    assert 'status.state === "cancelled"' in body
    assert "applySyncCancelled(status)" in body
    failed = body.find('status.state === "failed"')
    cancelled = body.find('status.state === "cancelled"')
    assert -1 < failed < cancelled


def test_sources_js_section_header_documents_the_cancel_contract() -> None:
    """The house pattern: the sync section's header docstring carries
    the decision tree (now 8 branches — the cancelled one is the
    5th) and the Cancel control's show/hide contract."""
    js = _text(SOURCES_JS)
    start = js.find("Sync sources button (Sources page only)")
    assert start != -1, "the sync section marker comment"
    header = js[start : js.find("const syncBtn = root.querySelector", start)]
    assert "5. sync cancelled" in header, "the tree's cancelled branch (the 5th)"
    assert "8. both idle" in header, "the tree is now 8 branches"
    assert "#sync-cancel-btn" in header, "the control's contract is documented"