**Phase 194 — final verification pass: all green.**
**Verified (all 5 tasks already complete; nothing re-implemented needed)**
- `app/rag/importer.py`: `_prescan_pending` + `_pending_hash_stmt` (scalar `documents.file_hash`, own session, ONE `asyncio.to_thread` span, REPLACES the stats pre-walk), `on_scan_progress`, `progress` pair = pending position, `0` branch = byte-identical phase-193 walk.
- `app/config.py`/`settings_catalog.py`/`.env.example`: `sync_pending_count=1` + validator; `app/api/sync.py`: `"scan"` phase, `scan_done`/`scan_total` (additive after `files_total`, kept at every terminal), forced/throttled RSS split; `frontend/assets/sources.js`: `fmtSyncPhaseLabel` scan branch first-arm, `fmtSyncLabel` untouched.
**Tests / lint / coverage (exact commands)**
- `uv run pytest --cov=app --cov-report=term-missing` → **4835 passed**, `app/` 99% (7240 stmts / 16 miss), exit 0
- `uv run pytest tests/e2e/test_sync_pending_count.py -v --no-cov` → **8 passed in 150.81s** (phase suite, isolated)
- 14 sync-lineage E2E suites, one pytest process each (`test_sync_remaining_count`, `…_chunk_progress`, `…_progress_box`, `…_peak_memory_visible`, `…_label_mobile`, `…_upload_progress`, `test_archive_upload_sources`, `…_image_progress`, `…_summary_visibility`, `…_status_freshness`, `test_smoke`, `test_fast_path_deferred_content`, `test_sync_session_reset`, `test_sync_git_freeze`) → **all exit 0**
- `uv run ruff check .` → All checks passed; `uv run pyright` → **0 errors**
**Completion criteria** — all met: `N remaining` = predicted change set (E2E arms 1/2/5, first import counts the walk, no-change → no count part); `scan` phase visible + flips to `import`, scan pair reset/kept at terminals (`test_sync_scan_phase.py`); `BOR_SYNC_PENDING_COUNT=0` walk semantics ("7 remaining" arm) + **negative refuses startup naming the field (verified live by importing `app.main`)**; pre-scan off-loop + cancel lands `cancelled` (integration pins); CLI/upload byte-identical (unedited pins green); payload additive (`_PRE_187_STATUS_KEYS | _MEMORY_KEYS | _SCAN_KEYS`); suite green, coverage >90%, lint/types clean.
**Documentation** — 3 fixes: README sync "States" bullet had an **unclosed parenthesis** (pre-existing, 193/194) and the scan-label "total is 0" clause was ambiguous about a `=0` run (it still counts the walk once the import starts); **AGENTS.md rule 17 had a stray `)`** (pre-existing from phase 191). Also verified consistency both directions for all 87 settings ↔ README ↔ `.env.example` ↔ settings catalog, the pages table ↔ `app/main.py` routes, no stale `BOR_` names, and README/AGENTS claims through phase 194.
**Notes/deviations** — no live-infra changes (prod sync knob ships default-on, no deployment touched); no git index/history commands run (all changes left in the working tree for the harness); commit criterion left to the harness as instructed.
**Next pending phase:** none — `.agents/phases/todo/` holds only `194_sync_pending_count`, which is now ready to move to `complete/`.
928 lines
39 KiB
Python
928 lines
39 KiB
Python
"""Integration: phase 194 (task 02) — the ``"scan"`` phase on the status surface.
|
|
|
|
The owner report that opened the phase (2026-10-19): *"it said '1216
|
|
remaining' when it only had like 10 files to sync"*. Task 01 moved the
|
|
count onto the pre-scan's PREDICTED change set; this file pins the
|
|
status half (design D3, owner-confirmed A2 — the pre-scan is a VISIBLE
|
|
stage, not an invisible pause):
|
|
|
|
* the phase machine is ``null`` → ``"scan"`` (set immediately before
|
|
``import_sources``, where the scan actually runs) → ``"import"``
|
|
(flipped by the FIRST per-file ``progress`` call) → ``"overview"`` →
|
|
``"summaries"`` → the terminal clears ``phase``;
|
|
* ``scan_done`` / ``scan_total`` carry the scan's own position through
|
|
that stage, reset with the run, and are KEPT at every terminal beside
|
|
the final file counts (the phase-64 keep-final-values convention);
|
|
* the payload is ADDITIVE — the pre-phase key set plus exactly those two
|
|
keys, byte-identical in name and type, stable from idle through the
|
|
terminal;
|
|
* the phase-187 sampling discipline holds on the new stage: the ``"scan"``
|
|
set and the ``scan``→``import`` flip are TRANSITIONS and force the
|
|
``/proc`` read, while the scan's own per-file hook is THROTTLED — one
|
|
read per interval, never one per file (LOCKED A4);
|
|
* the two degenerate/terminal edges: an empty knowledge base sits in
|
|
``"scan"`` and never reaches ``"import"`` (no ``progress`` call exists
|
|
to flip it) and a cancel DURING the scan lands the ``cancelled``
|
|
terminal with the scan counts kept and nothing indexed (the phase-134
|
|
contract, the app-side twin of the importer-side pin in
|
|
``tests/integration/test_sync_pending_count.py``).
|
|
|
|
**Why one arm is choreographed and one is real.** An in-process import
|
|
never awaits real I/O (the embedder is a fake), so the loop cannot answer a
|
|
status poll from inside it — a real run's ``"import"`` phase is therefore
|
|
NOT reliably observable, and the real arm pins what the terminal keeps
|
|
(both pairs, divergent on the incremental run). The flip itself gets a
|
|
deterministic pin from the choreographed double (the ``test_sync_api``
|
|
gated-import idiom), which drives the runner's OWN closures and parks at
|
|
every frame the test reads. The scan, by contrast, runs in a
|
|
``to_thread`` worker, so the loop is FREE through it — that is exactly why
|
|
its frames are observable, and why the blocking-hash fixture slows ONLY the
|
|
scan window (a blanket stub would park the loop through the import too).
|
|
|
|
House patterns (``test_sync_api.py`` / ``test_sync_git_freeze.py``): the
|
|
``sync_client`` fixture is a CONTEXT-MANAGED ``TestClient`` — one app event
|
|
loop across requests, so the background run survives between the POST and
|
|
the polls; the module ``_status`` / ``_task`` / ``_cancel_event`` /
|
|
``_rss_sampled_at`` and the ``git_sources`` / ``sources_meta`` /
|
|
``folder_summaries`` / ``documents`` global state are reset around every
|
|
test; the probe is stubbed so nothing reaches the network, and the import
|
|
runs for real over a temp ``kind=local`` source with the deterministic
|
|
``FakeEmbedder``. The scan is made observable with a PLAIN blocking
|
|
``_file_hash`` stub scoped to the scan window (the phase-179 timing-fixture
|
|
idiom): the point under test is WHICH phase the run is in WHEN the page
|
|
polls, and a scan of instant hashes is over before the first poll.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable, Iterator
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
import app.rag.importer as importer
|
|
from app.api import sync as sync_api
|
|
from app.config import Settings
|
|
from app.db import SessionLocal, db_available
|
|
from app.main import app as fastapi_app
|
|
from app.models import Document, GitSource
|
|
from app.rag import git_sources as git_sources_resolver
|
|
from app.rag.importer import ImportSummary
|
|
from app.rag.llm import LLMClient
|
|
from tests.conftest import ADMIN_PASSWORD
|
|
from tests.fakes import FakeEmbedder
|
|
|
|
#: The probe's own source-name prefix — nothing else in the KB uses it,
|
|
#: so the module's DELETE-by-source cleanup can never touch another
|
|
#: suite's rows.
|
|
PREFIX = "phase194-t02-"
|
|
|
|
#: One blocking ``_file_hash`` per file. Three files is a ~360 ms scan:
|
|
#: long enough that a 10 ms poll catches several distinct positions inside
|
|
#: the phase (the frames are what the UI's 2 s poll would render), short
|
|
#: enough that the suite stays fast.
|
|
HASH_BLOCK_S = 0.12
|
|
|
|
|
|
# --- fixtures (the house sync hygiene) --------------------------------------
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_sync_state() -> Iterator[None]:
|
|
"""The module-level status object + task + cancel event + sampling
|
|
guard are process-global — reset before AND after every test so no
|
|
terminal state, stale cancel or stale throttle window leaks into the
|
|
next one (``test_sync_api`` / ``test_sync_git_freeze``)."""
|
|
sync_api._status = sync_api.SyncStatus()
|
|
sync_api._task = None
|
|
sync_api._cancel_event = asyncio.Event()
|
|
sync_api._rss_sampled_at = float("-inf")
|
|
yield
|
|
sync_api._status = sync_api.SyncStatus()
|
|
sync_api._task = None
|
|
sync_api._cancel_event = asyncio.Event()
|
|
sync_api._rss_sampled_at = float("-inf")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_git_sources(db: Session) -> Iterator[None]:
|
|
"""The runner resolves through the real ``git_sources`` table —
|
|
truncated around every test (the rows this file seeds are
|
|
``kind=local``)."""
|
|
db.execute(text("TRUNCATE git_sources"))
|
|
db.commit()
|
|
yield
|
|
db.execute(text("TRUNCATE git_sources"))
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_sources_version() -> Iterator[None]:
|
|
"""The phase-53 bump is global mutable state — pin it to the
|
|
migration-0010 seed (0) around every test (own session: the runner
|
|
bumps through its own short-lived ``SessionLocal``)."""
|
|
if not db_available():
|
|
pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
|
|
session = SessionLocal()
|
|
try:
|
|
session.execute(text("UPDATE sources_meta SET version = 0 WHERE id = 1"))
|
|
session.commit()
|
|
yield
|
|
finally:
|
|
session.execute(text("UPDATE sources_meta SET version = 0 WHERE id = 1"))
|
|
session.commit()
|
|
session.close()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_folder_summaries(db: Session) -> Iterator[None]:
|
|
"""``folder_summaries`` is global state the real generator writes on a
|
|
changed KB — truncated around every test."""
|
|
db.execute(text("TRUNCATE folder_summaries"))
|
|
db.commit()
|
|
yield
|
|
db.execute(text("TRUNCATE folder_summaries"))
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_probe_documents(db: Session) -> Iterator[None]:
|
|
"""The real imports of this module write ``documents``/``chunks`` —
|
|
removed before and after every test, narrowed to this module's own
|
|
source names (the ``test_sync_pending_count`` hygiene)."""
|
|
_drop(db)
|
|
yield
|
|
_drop(db)
|
|
|
|
|
|
def _drop(db: Session) -> None:
|
|
db.execute(
|
|
text("DELETE FROM chunks WHERE document_id IN "
|
|
"(SELECT id FROM documents WHERE source LIKE :p)"),
|
|
{"p": f"{PREFIX}%"},
|
|
)
|
|
db.execute(text("DELETE FROM documents WHERE source LIKE :p"), {"p": f"{PREFIX}%"})
|
|
db.commit()
|
|
db.expunge_all()
|
|
|
|
|
|
@pytest.fixture()
|
|
def sync_client() -> Iterator[TestClient]:
|
|
"""Context-managed ``TestClient`` — one app event loop across requests
|
|
(the background run must survive between the POST and the polls)."""
|
|
with TestClient(fastapi_app) as client:
|
|
yield client
|
|
|
|
|
|
# --- helpers ----------------------------------------------------------------
|
|
|
|
|
|
def _settings(sources_dir: str) -> Settings:
|
|
return Settings(_env_file=None, sources_dir=sources_dir) # pyright: ignore[reportCallIssue]
|
|
|
|
|
|
def _stub_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Empty env fallback (the dev ``.env`` never leaks in) — this file's
|
|
runs resolve their one local row from the table."""
|
|
monkeypatch.setattr(
|
|
git_sources_resolver,
|
|
"get_settings",
|
|
lambda: Settings(_env_file=None, git_sources=""), # pyright: ignore[reportCallIssue]
|
|
)
|
|
|
|
|
|
def _seed_local(db: Session, path: Path) -> None:
|
|
"""A ``kind=local`` row exactly as the phase-38 API stores it."""
|
|
db.add(GitSource(url=str(path), kind="local", path=str(path)))
|
|
db.commit()
|
|
|
|
|
|
def _local_source(tmp_path: Path, name: str, files: dict[str, str]) -> Path:
|
|
"""One source root (always created — the empty-KB arm needs a real
|
|
empty directory, not a missing one), with *files* under it."""
|
|
assert name.startswith(PREFIX), f"the probe source must be {PREFIX}-prefixed"
|
|
root = tmp_path / name
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
for rel, body in files.items():
|
|
target = root / rel
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(body, encoding="utf-8")
|
|
return root
|
|
|
|
|
|
def _login(client: TestClient) -> None:
|
|
r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
|
|
assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
|
|
|
|
|
|
def _poll(client: TestClient, want: str, timeout: float = 20.0) -> dict:
|
|
"""Poll until ``state == want``, failing loudly on any other state."""
|
|
deadline = time.monotonic() + timeout
|
|
body: dict = {}
|
|
while time.monotonic() < deadline:
|
|
body = client.get("/api/sync/status").json()
|
|
if body["state"] == want:
|
|
return body
|
|
assert body["state"] == "running", (
|
|
f"unexpected state {body['state']!r} while waiting for {want!r}: {body}"
|
|
)
|
|
time.sleep(0.02)
|
|
raise AssertionError(f"sync did not reach {want!r} within {timeout}s: {body}")
|
|
|
|
|
|
def _watch_scan(client: TestClient, timeout: float = 20.0) -> tuple[list[dict], dict | None]:
|
|
"""Poll through the run's SCAN window; stop at the first frame that is
|
|
no longer a scan.
|
|
|
|
Returns ``(frames, terminal)`` — ``terminal`` is non-``None`` when the
|
|
run finished inside the window. Both stops are honest: the scan runs
|
|
OFF the loop (so its frames are served, and this list is exactly what
|
|
the label was built from), while the import that follows is one
|
|
straight run of the importer that can finish before the next poll.
|
|
The flip frame therefore gets pinned deterministically by the
|
|
choreographed test below; what the real run pins here is the scan's
|
|
own frames plus the values the terminal keeps."""
|
|
frames: list[dict] = []
|
|
terminal: dict | None = None
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
body = client.get("/api/sync/status").json()
|
|
if body["state"] != "running":
|
|
terminal = body
|
|
break
|
|
frames.append(body)
|
|
if body["phase"] != "scan":
|
|
break
|
|
time.sleep(0.01)
|
|
else:
|
|
raise AssertionError(f"the run never left the scan phase: {frames[-1]}")
|
|
return frames, terminal
|
|
|
|
|
|
class _FakeScanImport:
|
|
"""A parked ``import_sources`` double for the app-side edges.
|
|
|
|
It records the two position closures the runner passes (the wiring
|
|
pin) and fires NOTHING — which is exactly what a run with no
|
|
importable file does (no scan report, so no position, and no
|
|
``progress`` call to flip the phase; the importer-side proof lives in
|
|
the empty-KB arm below and in ``test_sync_pending_count.py``). The
|
|
park lets the test read the status inside the ``"scan"`` phase without
|
|
racing the pipeline; the caller always releases it in ``finally``.
|
|
"""
|
|
|
|
def __init__(self, summary: ImportSummary | None = None) -> None:
|
|
self.summary = summary if summary is not None else ImportSummary()
|
|
self.started = threading.Event()
|
|
self.release = threading.Event()
|
|
self.progress_hook: Callable[[str, str, int, int], None] | None = None
|
|
self.scan_hook: Callable[[int, int], None] | None = None
|
|
|
|
async def __call__(
|
|
self,
|
|
sources: list[Path],
|
|
llm: LLMClient,
|
|
*,
|
|
prune: bool = False,
|
|
limit: int | None = None,
|
|
session: Session | None = None,
|
|
progress: Callable[[str, str, int, int], None] | None = None,
|
|
on_chunk_progress: Callable[[int, int], None] | None = None,
|
|
on_image_progress: Callable[[int, int], None] | None = None,
|
|
on_scan_progress: Callable[[int, int], None] | None = None,
|
|
ignore_by_root: dict[str, list[str]] | None = None,
|
|
include_hidden_by_root: dict[str, bool] | None = None,
|
|
doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,
|
|
) -> ImportSummary:
|
|
self.progress_hook = progress
|
|
self.scan_hook = on_scan_progress
|
|
self.started.set()
|
|
await asyncio.to_thread(self.release.wait)
|
|
return self.summary
|
|
|
|
|
|
def _install_counting_memory(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
|
|
"""Replace the two memory helpers on ``app.api.sync`` with recorders.
|
|
|
|
``_sample_rss`` calls them ONLY when the throttle lets it, so one
|
|
entry per entry below is one real read — and each entry carries the
|
|
phase (and the scan/file position) at that instant, which is the
|
|
difference between "a transition forced a read" and "a per-file hook
|
|
read once per file" (LOCKED A4)."""
|
|
reads: list[dict[str, Any]] = []
|
|
|
|
def _record() -> None:
|
|
reads.append(
|
|
{
|
|
"phase": sync_api._status.phase,
|
|
"scan_done": sync_api._status.scan_done,
|
|
"files_done": sync_api._status.files_done,
|
|
"current_file": sync_api._status.current_file,
|
|
}
|
|
)
|
|
|
|
def fake_rss_mb() -> int:
|
|
_record()
|
|
return 100
|
|
|
|
def fake_rss_peak_mb() -> int:
|
|
return 120
|
|
|
|
monkeypatch.setattr(sync_api, "rss_mb", fake_rss_mb)
|
|
monkeypatch.setattr(sync_api, "rss_peak_mb", fake_rss_peak_mb)
|
|
return reads
|
|
|
|
|
|
def _slow_hash(
|
|
monkeypatch: pytest.MonkeyPatch, block: float = HASH_BLOCK_S
|
|
) -> dict[str, bool]:
|
|
"""Make ONLY the pre-scan window slow, and return its flag.
|
|
|
|
``_file_hash`` is called by the scan AND by the processing loop, so a
|
|
blanket blocking stub would also park the event loop through the whole
|
|
import (an in-process import never awaits real I/O, so the loop cannot
|
|
answer a status poll inside it) — the frames this file needs would
|
|
then be unservable rather than absent. The flag is raised around
|
|
:func:`app.rag.importer._prescan_pending` only, so the scan's window is
|
|
the fixture's window: the loop stays free exactly while the run is in
|
|
the phase under test (the phase-179 off-loop claim, observed from the
|
|
HTTP side), and the import after it runs at full speed.
|
|
|
|
The flag is a plain ``dict`` entry written/read from two threads: the
|
|
house hook contract (a GIL-atomic store) applies here for the same
|
|
reason it applies to the runner's closures.
|
|
"""
|
|
real_hash = importer._file_hash
|
|
real_prescan = importer._prescan_pending
|
|
state = {"in_scan": False}
|
|
|
|
def slow(path: Path) -> str:
|
|
if state["in_scan"]:
|
|
time.sleep(block)
|
|
return real_hash(path)
|
|
|
|
def spy_prescan(*args: Any, **kwargs: Any) -> Any:
|
|
state["in_scan"] = True
|
|
try:
|
|
return real_prescan(*args, **kwargs)
|
|
finally:
|
|
state["in_scan"] = False
|
|
|
|
monkeypatch.setattr(importer, "_file_hash", slow)
|
|
monkeypatch.setattr(importer, "_prescan_pending", spy_prescan)
|
|
return state
|
|
|
|
|
|
class _ChoreographedImport:
|
|
"""An ``import_sources`` double that replays the importer's hook script
|
|
through the runner's OWN closures, parking at every point the test
|
|
reads the status.
|
|
|
|
The choreography (all four stops are ``threading.Event`` pairs, so no
|
|
test here races the pipeline):
|
|
|
|
1. fire ``scan_pairs[:1]`` through ``on_scan_progress`` → ``first``
|
|
(the test reads the scan's FIRST position);
|
|
2. on ``go_on``: fire the rest → ``scanned`` (the test reads the last
|
|
position, still inside the phase);
|
|
3. on ``go_import``: fire ``progress_call`` through ``progress`` —
|
|
THE FLIP → ``flipped`` (the test reads the ``"import"`` frame);
|
|
4. on ``release``: return the canned summary.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
summary: ImportSummary,
|
|
scan_pairs: list[tuple[int, int]],
|
|
progress_call: tuple[str, str, int, int],
|
|
) -> None:
|
|
self.summary = summary
|
|
self.scan_pairs = list(scan_pairs)
|
|
self.progress_call = progress_call
|
|
self.first = threading.Event()
|
|
self.go_on = threading.Event()
|
|
self.scanned = threading.Event()
|
|
self.go_import = threading.Event()
|
|
self.flipped = threading.Event()
|
|
self.release = threading.Event()
|
|
self.scan_hook: Callable[[int, int], None] | None = None
|
|
self.progress_hook: Callable[[str, str, int, int], None] | None = None
|
|
|
|
async def __call__(
|
|
self,
|
|
sources: list[Path],
|
|
llm: LLMClient,
|
|
*,
|
|
prune: bool = False,
|
|
limit: int | None = None,
|
|
session: Session | None = None,
|
|
progress: Callable[[str, str, int, int], None] | None = None,
|
|
on_chunk_progress: Callable[[int, int], None] | None = None,
|
|
on_image_progress: Callable[[int, int], None] | None = None,
|
|
on_scan_progress: Callable[[int, int], None] | None = None,
|
|
ignore_by_root: dict[str, list[str]] | None = None,
|
|
include_hidden_by_root: dict[str, bool] | None = None,
|
|
doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,
|
|
) -> ImportSummary:
|
|
# The wiring pin: both closures arrive live (None here would mean
|
|
# the runner stopped passing them).
|
|
self.scan_hook = on_scan_progress
|
|
self.progress_hook = progress
|
|
assert on_scan_progress is not None and progress is not None
|
|
for pair in self.scan_pairs[:1]:
|
|
on_scan_progress(*pair)
|
|
self.first.set()
|
|
await asyncio.to_thread(self.go_on.wait)
|
|
for pair in self.scan_pairs[1:]:
|
|
on_scan_progress(*pair)
|
|
self.scanned.set()
|
|
await asyncio.to_thread(self.go_import.wait)
|
|
progress(*self.progress_call)
|
|
self.flipped.set()
|
|
await asyncio.to_thread(self.release.wait)
|
|
return self.summary
|
|
|
|
|
|
def _stub_probe(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
async def fake_check_models(llm: LLMClient) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(sync_api, "check_models", fake_check_models)
|
|
|
|
|
|
def _install_local_run(
|
|
monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path, source: Path
|
|
) -> None:
|
|
"""Point one app run at *source*: local row, tmp sources root, stubbed
|
|
probe, the deterministic in-process embedder (the real import, no
|
|
network)."""
|
|
_seed_local(db, source)
|
|
_stub_env(monkeypatch)
|
|
monkeypatch.setattr(
|
|
sync_api, "get_settings", lambda: _settings(sources_dir=str(tmp_path / "bor"))
|
|
)
|
|
_stub_probe(monkeypatch)
|
|
monkeypatch.setattr(sync_api, "LLMClient", lambda: FakeEmbedder())
|
|
|
|
|
|
_THREE_FILES = {
|
|
"a.md": "# A\n\nalpha content\n",
|
|
"b.md": "# B\n\nbravo content\n",
|
|
"notes/c.md": "# C\n\ncharlie content\n",
|
|
}
|
|
|
|
|
|
def _log_lines(caplog: pytest.LogCaptureFixture, prefix: str) -> list[str]:
|
|
return [r.getMessage() for r in caplog.records if r.getMessage().startswith(prefix)]
|
|
|
|
|
|
# --- the phase-machine sequence --------------------------------------------
|
|
|
|
|
|
def test_real_run_reports_the_scan_frames_and_keeps_both_pairs(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""D3 through the REAL pipeline, twice (one changed file included).
|
|
|
|
Run 1 (a first import): the prelude reports ``phase: null`` with
|
|
``scan_*`` 0/0, then the run reports ``"scan"`` with ``scan_total`` at
|
|
the importable count and ``scan_done`` climbing through AT LEAST two
|
|
distinct positions (the frames the ``Scanning… (n/total)`` label
|
|
needs) with ``current_file`` null throughout — the scan names no file.
|
|
The terminal clears ``phase`` and KEEPS the final scan pair beside the
|
|
file pair, and on a first import both pairs are 3/3 (A1's degenerate
|
|
parity: walk == work, the only shape the phase-193 suite ever saw).
|
|
|
|
Run 2 (one file edited — the owner's prod shape at 1-of-3): the scan
|
|
denominator is STILL the importable 3, while the kept file pair is
|
|
``1/1`` and the terminal reports exactly one updated document. That
|
|
divergence — the scan counted the walk, the count counted the work —
|
|
IS the phase, and it survives on the terminal where no poll can miss
|
|
it."""
|
|
source = _local_source(tmp_path, f"{PREFIX}machine", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
_slow_hash(monkeypatch)
|
|
|
|
_login(sync_client)
|
|
idle = sync_client.get("/api/sync/status").json()
|
|
assert idle["phase"] is None
|
|
assert idle["scan_done"] == 0 and idle["scan_total"] == 0
|
|
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
frames, _late = _watch_scan(sync_client)
|
|
scan_frames = [f for f in frames if f["phase"] == "scan"]
|
|
assert scan_frames, f"the run never reported the scan phase: {frames}"
|
|
assert all(f["current_file"] is None for f in scan_frames), (
|
|
"the scan named a file — current_file stays null until the first "
|
|
"progress call flips the phase to import"
|
|
)
|
|
# ``0`` is the transition frame — the phase is set before the scan
|
|
# reports its first file, and the UI's total===0 guard renders it bare.
|
|
totals = {f["scan_total"] for f in scan_frames}
|
|
assert totals <= {0, 3} and max(totals) == 3, totals
|
|
positions = sorted({f["scan_done"] for f in scan_frames if f["scan_total"] == 3})
|
|
assert len(positions) >= 2, (
|
|
f"only scan position(s) {positions} were observable — the label needs "
|
|
"the scan's own n/total to move (the fixture scan blocks on purpose)"
|
|
)
|
|
assert positions[0] >= 1 and positions[-1] <= 3
|
|
|
|
body = _poll(sync_client, "success")
|
|
assert body["detail"]["added"] == 3
|
|
assert body["phase"] is None and body["current_file"] is None
|
|
# The keep-final-values convention, beside the file counts.
|
|
assert body["scan_total"] == 3 and body["scan_done"] == 3
|
|
assert body["files_total"] == 3 and body["files_done"] == 3
|
|
|
|
# Run 2 — one changed file of three (the owner's prod case, scaled).
|
|
(source / "b.md").write_text("# B\n\nbravo content EDITED\n", encoding="utf-8")
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
frames2, _late2 = _watch_scan(sync_client)
|
|
scan2 = [f for f in frames2 if f["phase"] == "scan"]
|
|
totals2 = {f["scan_total"] for f in scan2 if f["scan_total"]}
|
|
assert scan2 and totals2 == {3}, (
|
|
"the scan denominator is the importable count, not the prediction"
|
|
)
|
|
assert sorted({f["scan_done"] for f in scan2 if f["scan_total"] == 3})[-1] >= 2
|
|
|
|
body2 = _poll(sync_client, "success")
|
|
assert (body2["detail"]["added"], body2["detail"]["updated"]) == (0, 1)
|
|
assert body2["phase"] is None
|
|
# The divergence the phase exists for: the kept scan pair is the WALK
|
|
# it did (3 files hashed), the kept file pair is the WORK (1 file).
|
|
assert (body2["scan_done"], body2["scan_total"]) == (3, 3)
|
|
assert (body2["files_done"], body2["files_total"]) == (1, 1)
|
|
|
|
|
|
def test_scan_flips_to_import_on_the_first_progress_call(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The sequence, frame by frame, with the flip pinned.
|
|
|
|
The in-process import runs straight through without awaiting real I/O,
|
|
so a poll cannot reliably land inside the ``"import"`` phase (that is
|
|
why the real arm above pins the terminal's kept pairs). This arm drives
|
|
the runner's OWN closures through a choreographed double — the
|
|
``test_sync_api`` gated-import idiom — and reads the status at each
|
|
stop: prelude ``null``/0/0 → ``"scan"`` at position 1 of 3 → still
|
|
``"scan"`` at position 3 of 3 → ``"import"`` with the PREDICTED
|
|
``files_total`` (1, the count that reaches ``N remaining``) and the
|
|
scan pair still showing 3/3 → the terminal, which clears ``phase`` and
|
|
keeps both pairs."""
|
|
source = _local_source(tmp_path, f"{PREFIX}flip", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
gated = _ChoreographedImport(
|
|
ImportSummary(files=3, updated=1, unchanged=2),
|
|
scan_pairs=[(1, 3), (2, 3), (3, 3)],
|
|
progress_call=(source.name, "b.md", 1, 1),
|
|
)
|
|
monkeypatch.setattr(sync_api, "import_sources", gated)
|
|
|
|
_login(sync_client)
|
|
idle = sync_client.get("/api/sync/status").json()
|
|
assert idle["state"] == "idle"
|
|
assert idle["phase"] is None
|
|
assert idle["scan_done"] == 0 and idle["scan_total"] == 0
|
|
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
try:
|
|
assert gated.first.wait(10.0), "the run never reached the scan"
|
|
s1 = sync_client.get("/api/sync/status").json()
|
|
assert s1["state"] == "running"
|
|
assert s1["phase"] == "scan"
|
|
assert (s1["scan_done"], s1["scan_total"]) == (1, 3)
|
|
assert s1["current_file"] is None
|
|
# Still the pre-import phase: no file count yet.
|
|
assert (s1["files_done"], s1["files_total"]) == (0, 0)
|
|
|
|
gated.go_on.set()
|
|
assert gated.scanned.wait(10.0), "the scan never reported its last file"
|
|
s2 = sync_client.get("/api/sync/status").json()
|
|
assert s2["phase"] == "scan"
|
|
assert (s2["scan_done"], s2["scan_total"]) == (3, 3)
|
|
|
|
gated.go_import.set()
|
|
assert gated.flipped.wait(10.0), "the progress hook never fired"
|
|
s3 = sync_client.get("/api/sync/status").json()
|
|
assert s3["phase"] == "import", (
|
|
"the first progress call must flip scan -> import: "
|
|
f"{ {k: s3[k] for k in ('phase', 'current_file', 'files_done', 'files_total')} }"
|
|
)
|
|
assert s3["current_file"] == f"{source.name}/b.md"
|
|
# The flip already carries the PREDICTED denominator — the number
|
|
# the label turns into "N remaining".
|
|
assert (s3["files_done"], s3["files_total"]) == (1, 1)
|
|
# … and the scan pair stays where the scan left it.
|
|
assert (s3["scan_done"], s3["scan_total"]) == (3, 3)
|
|
finally:
|
|
gated.go_on.set()
|
|
gated.go_import.set()
|
|
gated.release.set()
|
|
|
|
body = _poll(sync_client, "success")
|
|
assert body["phase"] is None and body["current_file"] is None
|
|
assert (body["scan_done"], body["scan_total"]) == (3, 3)
|
|
assert (body["files_done"], body["files_total"]) == (1, 1)
|
|
|
|
|
|
# --- the additive payload ----------------------------------------------------
|
|
|
|
|
|
def test_scan_keys_are_additive_and_stable_from_idle_to_terminal(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The additive-only contract on the wire (the phase-187 discipline).
|
|
|
|
The payload's key set is EXACTLY the pre-phase-194 set plus
|
|
``scan_done`` / ``scan_total``; it does not change between idle,
|
|
mid-run and the terminal (no key appears or disappears with the
|
|
phase); and both new keys are plain ``int`` — 0 when there is nothing
|
|
to report, never ``null`` and never a plausible position."""
|
|
expected_pre = {
|
|
"state",
|
|
"started_at",
|
|
"finished_at",
|
|
"detail",
|
|
"error",
|
|
"current_file",
|
|
"files_done",
|
|
"files_total",
|
|
"current_chunk",
|
|
"chunks_total",
|
|
"images_done",
|
|
"images_total",
|
|
"phase",
|
|
"current_summary",
|
|
"summaries_done",
|
|
"summaries_total",
|
|
"rss_mb",
|
|
"rss_peak_mb",
|
|
}
|
|
source = _local_source(tmp_path, f"{PREFIX}keys", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
_stub_probe(monkeypatch)
|
|
|
|
_login(sync_client)
|
|
idle = sync_client.get("/api/sync/status").json()
|
|
assert set(idle) == expected_pre | {"scan_done", "scan_total"}, set(idle)
|
|
assert idle["scan_done"] == 0 and idle["scan_total"] == 0
|
|
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
mid = sync_client.get("/api/sync/status").json()
|
|
assert set(mid) == expected_pre | {"scan_done", "scan_total"}
|
|
assert mid["state"] == "running"
|
|
|
|
body = _poll(sync_client, "success")
|
|
assert set(body) == expected_pre | {"scan_done", "scan_total"}
|
|
for key in ("scan_done", "scan_total"):
|
|
assert isinstance(body[key], int) and not isinstance(body[key], bool), body
|
|
# Every pre-existing key keeps its type too (the additive pin's other
|
|
# half — the file pair is still ints, the phase still a string|null).
|
|
assert isinstance(body["files_done"], int) and isinstance(body["files_total"], int)
|
|
assert body["phase"] is None
|
|
|
|
|
|
# --- the sampling discipline ------------------------------------------------
|
|
|
|
|
|
def test_scan_and_import_transitions_force_the_sample_and_the_scan_hook_does_not(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Phase 187's rule, extended to the new stage (LOCKED A4).
|
|
|
|
Every entry below is one REAL ``/proc`` read (the two helpers are only
|
|
called when the throttle lets ``_sample_rss`` through), tagged with
|
|
the phase it happened in. Collapsing consecutive same-phase reads, the
|
|
sequence is one read per transition: run start (``phase`` still null)
|
|
→ ``"scan"`` → ``"import"`` (the flip) → ``"overview"`` →
|
|
``"summaries"`` → the terminal (null again). The scan transition's
|
|
read carries ``scan_done == 0`` — it is the transition, not a file —
|
|
and there is EXACTLY one read inside the whole scan phase, so a
|
|
per-file scan read (one ``/proc`` read per file, the regression this
|
|
pins) fails here even though the fixture scanned three files."""
|
|
source = _local_source(tmp_path, f"{PREFIX}sampling", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
_slow_hash(monkeypatch)
|
|
reads = _install_counting_memory(monkeypatch)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
body = _poll(sync_client, "success")
|
|
|
|
assert reads, "the run sampled no memory at all"
|
|
phases = [r["phase"] for r in reads]
|
|
transitions = [p for i, p in enumerate(phases) if i == 0 or p != phases[i - 1]]
|
|
assert transitions == [None, "scan", "import", "overview", "summaries", None], reads
|
|
|
|
scan_reads = [r for r in reads if r["phase"] == "scan"]
|
|
assert len(scan_reads) == 1, (
|
|
f"the scan phase cost {len(scan_reads)} /proc reads for 3 files — the "
|
|
"per-file scan hook must sample UNFORCED (one read per interval)"
|
|
)
|
|
assert scan_reads[0]["scan_done"] == 0, (
|
|
"the scan-phase read is not the transition — the forced sample must "
|
|
"land when the phase is SET, before the first file is scanned"
|
|
)
|
|
flip_reads = [r for r in reads if r["phase"] == "import"]
|
|
assert flip_reads and flip_reads[0]["current_file"] is not None, (
|
|
"no read at the scan->import flip — the transition is a forced "
|
|
f"sampling point: {reads}"
|
|
)
|
|
# The payload reports what the sampler wrote, not a stale pair.
|
|
assert (body["rss_mb"], body["rss_peak_mb"]) == (100, 120), body
|
|
|
|
|
|
# --- the degenerate empty knowledge base ------------------------------------
|
|
|
|
|
|
def test_empty_kb_stays_in_scan_and_never_reaches_import(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The D1 degenerate clause, pinned on BOTH sides.
|
|
|
|
Importer side (real): an empty source dir makes the pre-scan report
|
|
NOTHING — no ``on_scan_progress`` call, no ``progress`` call, a zero
|
|
denominator. App side (the double that fires nothing, i.e. exactly
|
|
what that run reports): the status sits in ``"scan"`` — the flip
|
|
lives in the per-file hook and no hook ever fires — with ``scan_*``
|
|
0/0 and ``current_file`` null, and the terminal clears ``phase``
|
|
while keeping the 0/0 pair (the UI's ``total === 0`` guard then
|
|
renders the bare label with no count part)."""
|
|
# Importer side: the real scan of an empty root reports nothing.
|
|
empty = _local_source(tmp_path, f"{PREFIX}empty", {})
|
|
llm = FakeEmbedder()
|
|
scan_calls: list[tuple[int, int]] = []
|
|
progress_calls: list[tuple[str, str, int, int]] = []
|
|
summary = asyncio.run(
|
|
importer.import_sources(
|
|
[empty],
|
|
llm,
|
|
progress=lambda s, r, d, t: progress_calls.append((s, r, d, t)),
|
|
on_scan_progress=lambda p, t: scan_calls.append((p, t)),
|
|
)
|
|
)
|
|
assert (scan_calls, progress_calls) == ([], [])
|
|
assert summary.files == 0
|
|
|
|
# App side: a run that reports nothing is a run that stays "scan".
|
|
_install_local_run(monkeypatch, db, tmp_path, empty)
|
|
fake_import = _FakeScanImport()
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
try:
|
|
assert fake_import.started.wait(10.0), "the run never reached the import"
|
|
# Wiring: the runner passed BOTH live closures (so the empty
|
|
# phase above is the scan reporting nothing, not a missing hook).
|
|
assert fake_import.scan_hook is not None
|
|
assert fake_import.progress_hook is not None
|
|
body = sync_client.get("/api/sync/status").json()
|
|
assert body["state"] == "running"
|
|
assert body["phase"] == "scan"
|
|
assert body["current_file"] is None
|
|
assert body["scan_done"] == 0 and body["scan_total"] == 0
|
|
assert body["files_done"] == 0 and body["files_total"] == 0
|
|
finally:
|
|
fake_import.release.set()
|
|
|
|
terminal = _poll(sync_client, "success")
|
|
assert terminal["phase"] is None
|
|
assert terminal["scan_done"] == 0 and terminal["scan_total"] == 0
|
|
assert terminal["files_done"] == 0 and terminal["files_total"] == 0
|
|
|
|
|
|
# --- the cancel during the pre-scan -----------------------------------------
|
|
|
|
|
|
def test_cancel_during_the_scan_lands_cancelled_with_the_scan_counts_kept(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""The phase-134 contract at the pipeline's NEW await (the app-side
|
|
twin of the importer-side cancel pin).
|
|
|
|
The scan is blocking on purpose, so the cancel provably lands inside
|
|
it: ``POST /api/sync/cancel`` answers, the ``cancelled`` terminal
|
|
arrives well before the abandoned scan's remaining work could finish,
|
|
``phase`` clears, the position the scan had reached is KEPT beside the
|
|
(still zero) file pair, ``error`` stays null and ``detail`` empty (a
|
|
cancel is intent, not failure), and NOTHING was indexed — the walk
|
|
never started. The lifecycle line keeps its exact pre-phase prefix
|
|
(LOCKED A2's append-only rule)."""
|
|
source = _local_source(tmp_path, f"{PREFIX}cancel", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
_slow_hash(monkeypatch, block=0.4)
|
|
|
|
_login(sync_client)
|
|
with caplog.at_level(logging.INFO, logger="app.api.sync"):
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
# Wait for proof the run is INSIDE the scan: the phase, and a
|
|
# position that moved.
|
|
deadline = time.monotonic() + 20.0
|
|
live: dict = {}
|
|
while time.monotonic() < deadline:
|
|
live = sync_client.get("/api/sync/status").json()
|
|
if live["phase"] == "scan" and live["scan_done"] >= 1:
|
|
break
|
|
assert live["state"] == "running", live
|
|
time.sleep(0.01)
|
|
assert live.get("phase") == "scan" and live["scan_done"] >= 1, (
|
|
f"the cancel never landed inside the scan: {live}"
|
|
)
|
|
position_at_cancel = live["scan_done"]
|
|
|
|
t0 = time.monotonic()
|
|
r = sync_client.post("/api/sync/cancel")
|
|
assert r.status_code == 202, f"cancel not accepted: {r.status_code} {r.text}"
|
|
assert time.monotonic() - t0 < 2.0, "the cancel request itself stalled"
|
|
|
|
body = _poll(sync_client, "cancelled", timeout=3.0)
|
|
|
|
assert body["error"] is None
|
|
assert body["detail"] == {}
|
|
assert body["phase"] is None
|
|
assert body["current_file"] is None
|
|
# The scan's final position is kept (the keep-final-values
|
|
# convention), and it is the position the cancel found — the run did
|
|
# not finish the scan it was cancelled in.
|
|
assert body["scan_total"] == 3
|
|
assert body["scan_done"] >= position_at_cancel
|
|
assert body["files_done"] == 0 and body["files_total"] == 0
|
|
|
|
# Nothing was indexed: the walk never started.
|
|
assert db.scalar(
|
|
select(select(Document).where(Document.source == source.name).exists())
|
|
) is False
|
|
|
|
# The lifecycle line keeps its pre-phase-194 prefix exactly.
|
|
cancelled = _log_lines(caplog, "sync: cancelled by user state=cancelled files_done=")
|
|
assert len(cancelled) == 1, cancelled
|
|
|
|
|
|
# --- the kill switch's status shape ----------------------------------------
|
|
|
|
|
|
def test_kill_switch_off_reports_no_scan_position_but_the_same_machine(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""``BOR_SYNC_PENDING_COUNT=0`` on the STATUS side.
|
|
|
|
The importer takes the phase-193 stats pre-walk, which reports NOTHING,
|
|
so the scan pair stays 0/0 for the whole run (the page renders no
|
|
count from it) while ``files_done`` / ``files_total`` carry the WALK
|
|
position again — 3 walked, not 1 predicted. The phase machine itself is
|
|
UNCHANGED by the switch (the pre-``import_sources`` set is still
|
|
``"scan"`` and the first file still flips it: that flip is pinned in
|
|
the choreographed test above, and the switch's importer-side halves are
|
|
pinned in ``tests/unit/test_sync_pending_count.py`` and
|
|
``tests/integration/test_sync_pending_count.py``). The pins below read
|
|
the TERMINAL, which is where both pairs survive deterministically."""
|
|
source = _local_source(tmp_path, f"{PREFIX}off", _THREE_FILES)
|
|
_install_local_run(monkeypatch, db, tmp_path, source)
|
|
# The switch lives on the settings object the import reads (llm).
|
|
real_embedder = FakeEmbedder()
|
|
real_embedder.settings = Settings(_env_file=None, sync_pending_count=0) # pyright: ignore[reportCallIssue]
|
|
monkeypatch.setattr(sync_api, "LLMClient", lambda: real_embedder)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
body = _poll(sync_client, "success")
|
|
|
|
assert body["detail"]["added"] == 3
|
|
# No scan, so no position — kept at 0/0, the value the UI renders as
|
|
# "no count part" rather than a stale or invented one.
|
|
assert body["scan_done"] == 0 and body["scan_total"] == 0
|
|
# The WALK count is back (phase-193 semantics, byte-for-byte).
|
|
assert (body["files_done"], body["files_total"]) == (3, 3)
|
|
assert body["phase"] is None
|