phase: 147_sync_changed_only
All defaults match. Verification complete — everything is green with no defects found. **Phase 147 final verification pass — report** - All 3 task files found in `complete/147_sync_changed_only/` → verification pass only; no code changes needed. - Verified: migration 0025 (`documents.file_hash` nullable String(64), no index) + model field; fast path at top of `_index_file` (streamed 1 MiB sha256 → row lookup → `_unchanged_doc`); stamping on all three branches' create/update + shared helper; pre-phase NULL-hash fallthrough; retriever/`content_hash` untouched. - **Tests:** `uv run pytest --cov=app --cov-report=term-missing` → 3438 passed, `app/` coverage **99%** (>90% ✓); `uv run pytest tests/e2e/test_sync_changed_only.py -v --no-cov` → 3 passed in isolation ✓; `uv run ruff check . && uv run pyright` → all checks passed, 0 errors ✓; prior-phase `tests/e2e/test_sync_button.py` → 3 passed (no regression) ✓. - **Completion criteria:** all 7 pass — (1) column + round-trip ✓ (2) unchanged text/image/garbage-PDF/pictorial-PDF cost zero LLM/extraction/vision, `summary.unchanged` counted, rows byte-still ✓ (3) changed file re-indexes + re-stamps; bytes-changed/content-same stamps only ✓ (4) backfill/D1/D4 + log wordings pinned verbatim ✓ (5) `content_hash`/retriever byte-identical ✓ (6) all gates green ✓ (7) no prior-phase behavior change ✓. - **Documentation:** full review of README.md (1,031 lines) and AGENTS.md against the code: every `.env.example`/code `BOR_*` var documented, all config defaults match `app/config.py`, quick-start/QA commands and referenced test files exist, `file://` URL + troubleshooting claims accurate; the phase's README "Updating the documents" edit is accurate. **Verified consistent — no changes needed.** - **Notable:** no live-infrastructure changes (dev Postgres container used for the standard test gates only). Working tree left uncommitted per harness protocol. - **Next pending phase:** `148_pdf_pages_in_sources`.
This commit is contained in:
@@ -0,0 +1,408 @@
|
||||
"""Phase 147 E2E (Playwright): only what changed is synced.
|
||||
|
||||
Source: ``TODO.md:4`` — "Local sources, like PDFs, are synced every
|
||||
time even if they haven't changed. This results in really long sync
|
||||
times for PDFs that are mostly pictoral. Only sync documents if they've
|
||||
changed since the last sync (hash the document contents to check)."
|
||||
|
||||
Run in isolation (DB must be up: ``podman compose up -d db``; git on
|
||||
PATH — a documented environment prerequisite, phase 28):
|
||||
|
||||
uv run pytest tests/e2e/test_sync_changed_only.py -v --no-cov
|
||||
|
||||
End-to-end proof through the real UI + the real sync path (a real
|
||||
``git clone`` of a local ``file://`` fixture repo, a real import
|
||||
against the mock LLM): the SECOND sync of an unchanged source reports
|
||||
everything unchanged and touches NO document row (``indexed_at`` /
|
||||
``content_hash`` / ``file_hash`` all stay still — a re-index would
|
||||
bump ``indexed_at``); a source change syncs ONLY the changed part (new
|
||||
file → ``1 added``; modified file → ``1 updated`` with only its own
|
||||
row moving). NO PDF here — the unit matrix (task 02) carries the
|
||||
vision-skip proof; this suite pins the WIRE (the owner's visible
|
||||
``#sync-result`` line) + the UI settle + the row-level proof, fast and
|
||||
deterministic.
|
||||
|
||||
Per-module app env (the E2E conftest pattern, module-scoped): this
|
||||
suite's app boots with ``BOR_GIT_SOURCES=file://<fixture repo>`` and
|
||||
its own ``BOR_SOURCES_DIR`` on its own port (env-overridable, the
|
||||
house port-constant pattern). The session app (no git sources) is
|
||||
never started in this isolated run, so no port clash.
|
||||
|
||||
The three tests SHARE the settled state on purpose (one module-scoped
|
||||
truncate, not per-test — the suite's point is the run 2 → run 3 →
|
||||
run 4 sequence):
|
||||
|
||||
1. ``test_second_sync_unchanged`` — run 1 (clone + import) indexes
|
||||
``N_DOCS`` documents and stamps every row's ``file_hash`` non-NULL;
|
||||
run 2 (unchanged bytes) settles ``0 added · N unchanged`` and moves
|
||||
NO row (the load-bearing assertion).
|
||||
2. ``test_change_syncs_only_the_change`` — commit a NEW doc to the
|
||||
fixture repo; run 3 settles ``1 added · N unchanged``; the new row
|
||||
appears stamped, the pre-existing rows do not move.
|
||||
3. ``test_changed_file_reindexes`` — modify the sentinel doc; run 4
|
||||
settles ``0 added · 1 updated · N unchanged``; exactly the
|
||||
sentinel row moves (``indexed_at`` bumped, ``content_hash`` and
|
||||
``file_hash`` re-stamped).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from playwright.sync_api import Page, expect
|
||||
from sqlalchemy import select, text
|
||||
|
||||
from app.db import SessionLocal
|
||||
from app.models import Document
|
||||
from e2e.auth_helpers import login
|
||||
from e2e.conftest import (
|
||||
ADMIN_PASSWORD,
|
||||
SESSION_SECRET,
|
||||
USE_REAL_LLM,
|
||||
_wait_http,
|
||||
)
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
# The house port-constant pattern (module app, env-overridable) — the
|
||||
# session app (8123) is never started in this isolated run.
|
||||
APP_PORT = int(os.environ.get("E2E_APP_PORT_SYNCCHANGED", "8157"))
|
||||
APP_URL = f"http://127.0.0.1:{APP_PORT}"
|
||||
|
||||
# The fixture repo's directory name — the source name the import
|
||||
# records for it (the repo_name basename rule, phase 28).
|
||||
GIT_SOURCE_NAME = "changed-notes"
|
||||
|
||||
# The three importable fixture docs (task 03: a few importable docs,
|
||||
# NO PDF — the unit matrix covers the vision-skip proof).
|
||||
SENTINEL_DOC = "notes/sync-fixture.md"
|
||||
SECOND_DOC = "notes/second-note.md"
|
||||
THIRD_DOC = "notes/third-note.md"
|
||||
#: The doc test 2 commits AFTER run 2 (``1 added``).
|
||||
NEW_DOC = "notes/new-note.md"
|
||||
N_DOCS = 3
|
||||
|
||||
#: "Synced HH:MM" — the local-time last-result label (sources.js's
|
||||
#: fmtSyncTime), any hour/minute.
|
||||
SYNCED_LABEL = re.compile(r"Synced \d{1,2}:\d{2}")
|
||||
|
||||
#: Real git clone/pull + import against the mock LLM (+ the KB
|
||||
#: overview regeneration after a KB-changing run) — generous budget,
|
||||
#: no client timeout (the test_sync_button.py pattern).
|
||||
SYNC_TIMEOUT_MS = 90_000
|
||||
|
||||
#: One document row's row-level proof snapshot (phase 147: the
|
||||
#: ``file_hash`` stamp joins the pre-phase pair).
|
||||
DocRow = tuple[datetime | None, str | None, str | None]
|
||||
|
||||
|
||||
def _git(cwd: Path, *args: str) -> None:
|
||||
"""Run git in *cwd*; a non-zero exit fails the fixture/test loudly."""
|
||||
proc = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True)
|
||||
if proc.returncode != 0:
|
||||
raise AssertionError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")
|
||||
|
||||
|
||||
def _commit(cwd: Path, message: str) -> None:
|
||||
"""The house fixture commit — identity inline, never signed."""
|
||||
_git(
|
||||
cwd,
|
||||
"-c", "user.email=e@x", "-c", "user.name=t",
|
||||
"-c", "commit.gpgsign=false", # the fixture commits never sign
|
||||
"commit", "-qm", message,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def sync_git_repo(tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
"""A real one-commit git repo with ``N_DOCS`` importable docs that
|
||||
the sync must clone (the test_sync_button.py recipe, module-safe
|
||||
``tmp_path_factory`` + real ``git`` subprocess calls)."""
|
||||
root = tmp_path_factory.mktemp("sync_changed_git")
|
||||
repo = root / GIT_SOURCE_NAME
|
||||
(repo / "notes").mkdir(parents=True)
|
||||
(repo / "notes" / "sync-fixture.md").write_text(
|
||||
"# Sync fixture note\n"
|
||||
"\n"
|
||||
"One small note that exists only to prove the changed-only sync\n"
|
||||
"end to end: a real git clone, a real import, an idempotent\n"
|
||||
"second run — and the sentinel test 3 modifies.\n"
|
||||
"\n"
|
||||
"Marker: RESE-SYNC-CHANGED-SENTINEL-7f41\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(repo / "notes" / "second-note.md").write_text(
|
||||
"# Second note\n"
|
||||
"\n"
|
||||
"A second unchanged doc — the sync's unchanged count is over\n"
|
||||
"the whole source, so it must stay still through runs 2–4.\n"
|
||||
"\n"
|
||||
"Marker: RESE-SECOND-NOTE-5c0b\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(repo / "notes" / "third-note.md").write_text(
|
||||
"# Third note\n"
|
||||
"\n"
|
||||
"A third unchanged doc (rounds out the source — the settle\n"
|
||||
"line's unchanged count is the KB size, not a singleton).\n"
|
||||
"\n"
|
||||
"Marker: RESE-THIRD-NOTE-91de\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
_git(repo, "init", "-q")
|
||||
_git(repo, "add", "-A")
|
||||
_commit(repo, "one")
|
||||
assert (repo / ".git").is_dir()
|
||||
return repo
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def app_server(mock_llm: int, sync_git_repo: Path) -> Iterator[str]:
|
||||
"""The real app under test — per-module env: the sync's subject is
|
||||
a real ``file://`` git source with its own checkout dir (the
|
||||
conftest session app boots without ``BOR_GIT_SOURCES`` and is never
|
||||
started in this isolated run)."""
|
||||
env = dict(os.environ)
|
||||
env.pop("DEBUGPY", None)
|
||||
env["BOR_ENVIRONMENT"] = "e2e"
|
||||
env["BOR_STATIC_DIR"] = str(REPO / "frontend")
|
||||
env["BOR_LLM_BASE_URL"] = (
|
||||
"https://aipi.reeseapps.com/v1"
|
||||
if USE_REAL_LLM
|
||||
else f"http://127.0.0.1:{mock_llm}/v1"
|
||||
)
|
||||
# Mock-calibrated threshold (conftest pattern) — keeps every story
|
||||
# suite's deterministic gate behavior; production default stays 0.62.
|
||||
env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
|
||||
env.setdefault(
|
||||
"BOR_DATABASE_URL",
|
||||
"postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
|
||||
)
|
||||
# Phase 16: admin auth must be set or create_app() refuses to boot.
|
||||
env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
|
||||
env["BOR_SESSION_SECRET"] = SESSION_SECRET
|
||||
# Phase 147: the sync's subject — one real local repo.
|
||||
env["BOR_GIT_SOURCES"] = f"file://{sync_git_repo}"
|
||||
env["BOR_SOURCES_DIR"] = str(sync_git_repo.parent / "checkouts")
|
||||
proc = subprocess.Popen(
|
||||
[sys.executable, "-m", "uvicorn", "app.main:app",
|
||||
"--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"],
|
||||
cwd=REPO,
|
||||
env=env,
|
||||
)
|
||||
try:
|
||||
_wait_http(f"{APP_URL}/api/health")
|
||||
yield APP_URL
|
||||
finally:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def app_url(app_server: str) -> str:
|
||||
return app_server
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def _clean_kb_once(app_url: str) -> Iterator[None]:
|
||||
"""Fresh KB ONCE per module — deliberately NOT per test: the suite's
|
||||
point is the run 2 → run 3 → run 4 sequence, and tests 2–3 build on
|
||||
the settled state test 1 leaves behind. git_sources too: the shared
|
||||
Postgres may carry leftover registry rows from other suites, and
|
||||
effective_sources prefers the DB list over this app's
|
||||
BOR_GIT_SOURCES env fallback."""
|
||||
body = httpx.get(f"{app_url}/api/health", timeout=5).json()
|
||||
if body["db"] != "up":
|
||||
pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
|
||||
with SessionLocal() as db:
|
||||
db.execute(
|
||||
text("TRUNCATE chunks, documents, query_log, kb_overview, git_sources")
|
||||
)
|
||||
db.commit()
|
||||
yield
|
||||
|
||||
|
||||
def _doc_rows() -> dict[str, DocRow]:
|
||||
"""Every document row of the fixture source, keyed by path — the
|
||||
row-level proof (the test_sync_button.py DB-assertion idiom,
|
||||
extended with the phase-147 ``file_hash`` stamp)."""
|
||||
with SessionLocal() as db:
|
||||
docs = db.execute(
|
||||
select(Document).where(Document.source == GIT_SOURCE_NAME)
|
||||
).scalars().all()
|
||||
return {d.path: (d.indexed_at, d.content_hash, d.file_hash) for d in docs}
|
||||
|
||||
|
||||
def _click_sync_and_wait_settle(page: Page) -> None:
|
||||
"""Click the admin sync button and wait for the settle — the
|
||||
"Syncing…" → "Synced HH:MM" transition with the button re-enabled
|
||||
at the terminal state (§7.4 never-stale, the test_sync_button.py
|
||||
idiom)."""
|
||||
btn = page.locator("#sync-btn")
|
||||
btn.click()
|
||||
expect(btn).to_be_disabled()
|
||||
expect(page.locator("#sync-label")).to_have_text("Syncing…")
|
||||
expect(page.locator("#sync-label")).to_have_text(
|
||||
SYNCED_LABEL, timeout=SYNC_TIMEOUT_MS
|
||||
)
|
||||
expect(btn).to_be_enabled()
|
||||
|
||||
|
||||
def _expect_settle_line(page: Page, expected: str) -> None:
|
||||
"""The owner's surface: the VISIBLE ``#sync-result`` text in its
|
||||
house count shape (sources.js's fmtSyncResult — the zero
|
||||
"updated" part is omitted: ``0 added · N unchanged``,
|
||||
``1 added · N unchanged``, ``0 added · 1 updated · N unchanged``)."""
|
||||
expect(page.locator("#sync-result")).to_have_text(
|
||||
expected, timeout=SYNC_TIMEOUT_MS
|
||||
)
|
||||
|
||||
|
||||
def _expect_rows_moved(
|
||||
before: dict[str, DocRow], after: dict[str, DocRow], *, moved: set[str]
|
||||
) -> None:
|
||||
"""Exactly the rows in *moved* may differ — every other row's
|
||||
``indexed_at``/``content_hash``/``file_hash`` is byte-still (a
|
||||
re-index would bump ``indexed_at``; the changed-only sync must not
|
||||
leak into neighbors). New rows may appear only within *moved*; the
|
||||
suite never expects a prune."""
|
||||
gained = set(after) - set(before)
|
||||
assert gained <= moved, f"unexpected rows added: {sorted(gained)}"
|
||||
lost = set(before) - set(after)
|
||||
assert not lost, f"unexpected rows pruned: {sorted(lost)}"
|
||||
for path, snapshot in before.items():
|
||||
if path in moved:
|
||||
continue
|
||||
assert after[path] == snapshot, (
|
||||
f"{path} moved though only {sorted(moved)} may move: "
|
||||
f"{snapshot} -> {after[path]}"
|
||||
)
|
||||
|
||||
|
||||
# --- 1. Run 1 stamps, run 2 re-indexes nothing -----------------------------
|
||||
|
||||
|
||||
def test_second_sync_unchanged(page: Page, app_url: str, db_ready: None) -> None:
|
||||
"""Run 1 (the real clone + import) indexes ``N_DOCS`` docs and
|
||||
stamps EVERY row's ``file_hash`` non-NULL (the first sync's job);
|
||||
run 2 (unchanged bytes) settles ``0 added · N unchanged`` and
|
||||
touches NO document row — ``indexed_at``/``content_hash``/
|
||||
``file_hash`` all still (the load-bearing assertion: a re-index
|
||||
would bump ``indexed_at`` to now and the owner's long syncs would
|
||||
be back)."""
|
||||
login(page, app_url) # lands on /sources.html (the button's home)
|
||||
|
||||
# --- run 1: clone + import (everything is new) -----------------------
|
||||
_click_sync_and_wait_settle(page)
|
||||
_expect_settle_line(page, f"{N_DOCS} added")
|
||||
after_run1 = _doc_rows()
|
||||
assert set(after_run1) == {SENTINEL_DOC, SECOND_DOC, THIRD_DOC}, after_run1
|
||||
for path, (indexed_at, content_hash, file_hash) in after_run1.items():
|
||||
assert indexed_at is not None and content_hash, path
|
||||
# The first sync stamped the hash on every row (the fast
|
||||
# path's precondition — NULL would mean run 2 re-indexed).
|
||||
assert file_hash is not None and len(file_hash) == 64, (
|
||||
f"{path} must carry a non-NULL file_hash after run 1"
|
||||
)
|
||||
|
||||
# --- run 2: unchanged bytes → zero re-index --------------------------
|
||||
_click_sync_and_wait_settle(page)
|
||||
_expect_settle_line(page, f"0 added · {N_DOCS} unchanged")
|
||||
after_run2 = _doc_rows()
|
||||
_expect_rows_moved(after_run1, after_run2, moved=set())
|
||||
|
||||
|
||||
# --- 2. A new file syncs as `1 added`, nothing else moves ------------------
|
||||
|
||||
|
||||
def test_change_syncs_only_the_change(
|
||||
page: Page, app_url: str, db_ready: None, sync_git_repo: Path
|
||||
) -> None:
|
||||
"""From the settled state (test 1), commit a NEW doc to the fixture
|
||||
repo's working tree: run 3 syncs ONLY the change — the settle line
|
||||
reads ``1 added · N unchanged`` (the unchanged count is still run
|
||||
1's doc count), the new row exists stamped, and the pre-existing
|
||||
rows' ``indexed_at``/``content_hash`` are untouched."""
|
||||
login(page, app_url)
|
||||
before = _doc_rows()
|
||||
assert len(before) == N_DOCS, "the settled state from test 1 is required"
|
||||
|
||||
(sync_git_repo / "notes" / "new-note.md").write_text(
|
||||
"# New note\n"
|
||||
"\n"
|
||||
"Committed after run 2 — only this file may move in run 3.\n"
|
||||
"\n"
|
||||
"Marker: RESE-NEW-DOC-3d8a\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
_git(sync_git_repo, "add", "-A")
|
||||
_commit(sync_git_repo, "add new-note")
|
||||
|
||||
_click_sync_and_wait_settle(page)
|
||||
_expect_settle_line(page, f"1 added · {N_DOCS} unchanged")
|
||||
|
||||
after = _doc_rows()
|
||||
assert set(after) - set(before) == {NEW_DOC}, after.keys()
|
||||
_, new_content_hash, new_file_hash = after[NEW_DOC]
|
||||
assert new_content_hash, "the added row must carry its content hash"
|
||||
assert new_file_hash is not None and len(new_file_hash) == 64, (
|
||||
"the added row must carry its file_hash stamp"
|
||||
)
|
||||
# The pre-existing rows are byte-still (the added file must not
|
||||
# drag them into a re-index).
|
||||
_expect_rows_moved(before, after, moved={NEW_DOC})
|
||||
|
||||
|
||||
# --- 3. A modified file syncs as `1 updated`, only its own row -------------
|
||||
|
||||
|
||||
def test_changed_file_reindexes(
|
||||
page: Page, app_url: str, db_ready: None, sync_git_repo: Path
|
||||
) -> None:
|
||||
"""Modify the fixture sentinel doc (append a line) + commit: run 4
|
||||
syncs ONLY the change — the settle line reads
|
||||
``0 added · 1 updated · N unchanged`` (N = the run-1 count — the
|
||||
test-2 doc is unchanged too), and EXACTLY the sentinel row moves:
|
||||
its ``indexed_at`` is bumped, its ``content_hash`` and
|
||||
``file_hash`` re-stamped; every other row is byte-still."""
|
||||
login(page, app_url)
|
||||
before = _doc_rows()
|
||||
assert len(before) == N_DOCS + 1, "the settled state from test 2 is required"
|
||||
|
||||
sentinel = sync_git_repo / "notes" / "sync-fixture.md"
|
||||
sentinel.write_text(
|
||||
sentinel.read_text(encoding="utf-8")
|
||||
+ "\nAmended: the sentinel was touched — exactly this row may move.\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
_git(sync_git_repo, "add", "-A")
|
||||
_commit(sync_git_repo, "modify sentinel")
|
||||
|
||||
_click_sync_and_wait_settle(page)
|
||||
_expect_settle_line(page, f"0 added · 1 updated · {N_DOCS} unchanged")
|
||||
|
||||
after = _doc_rows()
|
||||
_expect_rows_moved(before, after, moved={SENTINEL_DOC})
|
||||
old_indexed_at, old_content_hash, old_file_hash = before[SENTINEL_DOC]
|
||||
new_indexed_at, new_content_hash, new_file_hash = after[SENTINEL_DOC]
|
||||
assert new_content_hash != old_content_hash, (
|
||||
"the content changed — content_hash must be re-stamped"
|
||||
)
|
||||
assert new_file_hash != old_file_hash, (
|
||||
"the bytes changed — file_hash must be re-stamped"
|
||||
)
|
||||
assert (
|
||||
old_indexed_at is not None
|
||||
and new_indexed_at is not None
|
||||
and new_indexed_at > old_indexed_at
|
||||
), "the re-index must bump indexed_at (run 1 stamped it; run 4 re-indexed)"
|
||||
@@ -1636,10 +1636,11 @@ def test_ikea_tar_sparse_pages_are_described_and_rerun_unchanged(
|
||||
(the owner's rationale — more information is better for the
|
||||
embeddings), the title is the metadata title, the digest is over
|
||||
the FINAL content, and a re-sync with the same fake lands
|
||||
``unchanged`` (digest stable, byte-stable row, NO re-embed; the
|
||||
vision calls are re-issued before the unchanged check can skip
|
||||
them — the documented no-cache cost, the phase-133 convention,
|
||||
A21 endorses it)."""
|
||||
``unchanged`` through the phase-147 FAST PATH (the raw bytes are
|
||||
the change signal: byte-stable row, NO re-embed, and the vision
|
||||
calls are NOT re-issued — the pass never runs on an unchanged
|
||||
file; the pre-phase "vision re-issued before the unchanged check"
|
||||
no-cache cost is gone)."""
|
||||
pdf_path = ikea_pdf
|
||||
llm = _image_llm(
|
||||
tmp_path,
|
||||
@@ -1724,20 +1725,19 @@ def test_ikea_tar_sparse_pages_are_described_and_rerun_unchanged(
|
||||
f"(min_chars={min_chars})"
|
||||
]
|
||||
|
||||
# The re-sync with the SAME fake: ``unchanged`` — the final
|
||||
# content is a pure function of the fixture + the canned
|
||||
# replies, so the digest matches and the row is skipped (no
|
||||
# re-chunk, no re-embed, no re-summary, no KB change). The
|
||||
# vision calls ARE re-issued (the documented no-cache cost —
|
||||
# the phase-133 convention; a non-deterministic vision model
|
||||
# re-describes the row each sync: a cost, not a bug).
|
||||
# The re-sync with the SAME fake: ``unchanged`` through the
|
||||
# phase-147 fast path — the raw bytes are the change signal,
|
||||
# so the pass never re-runs: no re-render, the vision calls
|
||||
# are NOT re-issued (the pre-phase "vision re-issued before
|
||||
# the unchanged check can skip them" no-cache cost is gone),
|
||||
# no re-chunk, no re-embed, no re-summary, no KB change.
|
||||
embeds1 = len(llm.calls)
|
||||
pictorial1 = len(llm.pictorial_texts)
|
||||
content1, digest1 = doc.content, doc.content_hash
|
||||
s2 = asyncio.run(import_sources([pdf_path.parent], llm, session=db))
|
||||
assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
|
||||
assert s2.pdfs_pictorial == 1 # the pass re-fired
|
||||
assert len(llm.pictorial_texts) == 2 * pictorial1 # re-issued (cost)
|
||||
assert s2.pdfs_pictorial == 0 # the pass did not re-fire
|
||||
assert len(llm.pictorial_texts) == pictorial1 # NO re-issued vision
|
||||
assert len(llm.calls) == embeds1, "unchanged: no re-embed"
|
||||
assert s2.changed_paths == [], "unchanged: no KB change recorded"
|
||||
doc2 = db.scalar(
|
||||
|
||||
@@ -0,0 +1,366 @@
|
||||
"""Integration: migration 0025 (documents.file_hash) schema contract
|
||||
(phase 147, task 01).
|
||||
|
||||
Drives the **real Alembic engine** against the live dev database
|
||||
(``podman compose up -d db``), mirroring the house pattern of
|
||||
``test_migration_0024.py`` (information_schema assertions on the state
|
||||
the migration must leave). The tests target the 0024 → 0025 step
|
||||
explicitly so later migrations cannot break the pins:
|
||||
|
||||
* upgrade 0024 → 0025 → ``file_hash`` exists with the full contract —
|
||||
VARCHAR(64), NULLABLE, NO server default (NULL = "unknown" for
|
||||
pre-phase rows; a migration cannot backfill — it cannot see the
|
||||
source files on disk) — and is UNINDEXED (the fast path compares it
|
||||
in Python after the unique ``(source, path)`` lookup — internal
|
||||
bookkeeping, never a query surface) while the 0024 ``documents``
|
||||
schema survives — in particular ``content_hash`` stays NOT NULL and
|
||||
INDEXED (the retriever's dedupe/caching identity, untouched);
|
||||
* rows inserted while the DB is at 0024 backfill ``file_hash`` to NULL
|
||||
— the pre-phase shape, no backfill, no default;
|
||||
* the ORM contract agrees: a freshly inserted ``Document`` with no
|
||||
``file_hash`` reads ``file_hash is None`` through a fresh session
|
||||
(the pre-phase shape), and an explicit digest round-trips;
|
||||
* downgrade to 0024 → the column is GONE (A13) while the row + its
|
||||
content survive; upgrade back to 0025 → the column is back
|
||||
(round-trip).
|
||||
|
||||
The ``alembic`` fixture guarantees the DB ends at head even if a test
|
||||
fails or the process is interrupted.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from alembic.config import Config
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from alembic import command
|
||||
from app.db import SessionLocal, db_available
|
||||
from app.models import Document
|
||||
|
||||
SOURCE = "mig0025"
|
||||
CONTENT = "content here"
|
||||
CONTENT_HASH = hashlib.sha256(CONTENT.encode()).hexdigest()
|
||||
FILE_HASH = hashlib.sha256(b"raw source bytes").hexdigest()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def alembic(db: Session) -> Iterator[Config]:
|
||||
"""Real Alembic config bound to the dev DB (URL from app settings).
|
||||
|
||||
Starts at head (repairs an interrupted earlier run); teardown
|
||||
upgrades to head no matter what happened, so the dev DB is never
|
||||
left below head.
|
||||
"""
|
||||
if not db_available():
|
||||
pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
|
||||
cfg = Config() # no alembic.ini file — env.py gets the URL from app config
|
||||
cfg.set_main_option("script_location", "alembic")
|
||||
command.upgrade(cfg, "head")
|
||||
try:
|
||||
yield cfg
|
||||
finally:
|
||||
# Release the test session's open transaction BEFORE the repair
|
||||
# DDL: an idle-in-transaction SELECT holds an ACCESS SHARE lock
|
||||
# on ``documents``, which would deadlock the repair's
|
||||
# ``ALTER TABLE`` (0025) forever.
|
||||
db.rollback()
|
||||
command.upgrade(cfg, "head")
|
||||
|
||||
|
||||
def _version(db: Session) -> str | None:
|
||||
return db.execute(text("SELECT version_num FROM alembic_version")).scalar()
|
||||
|
||||
|
||||
def _column(db: Session, column: str) -> tuple[Any, ...] | None:
|
||||
"""(data_type, is_nullable, column_default, char_max_length) for one
|
||||
documents column."""
|
||||
row = db.execute(
|
||||
text(
|
||||
"SELECT data_type, is_nullable, column_default, character_maximum_length"
|
||||
" FROM information_schema.columns"
|
||||
" WHERE table_name = 'documents' AND column_name = :c"
|
||||
),
|
||||
{"c": column},
|
||||
).fetchone()
|
||||
return tuple(row) if row is not None else None
|
||||
|
||||
|
||||
def _index_covering(db: Session, column: str) -> str | None:
|
||||
"""The name of any documents index whose definition mentions the
|
||||
column (None = UNINDEXED)."""
|
||||
return db.execute(
|
||||
text(
|
||||
"SELECT indexname FROM pg_indexes"
|
||||
" WHERE tablename = 'documents' AND indexdef ILIKE :like"
|
||||
),
|
||||
{"like": f"%{column}%"},
|
||||
).scalar()
|
||||
|
||||
|
||||
def _insert_sql(db: Session, path: str) -> uuid.UUID:
|
||||
"""Insert one documents row with the PRE-0025 column set (the 0024
|
||||
shape — the file_hash column, when present, is omitted so a NULL
|
||||
backfill is what the row reads)."""
|
||||
doc_id = uuid.uuid4()
|
||||
db.execute(
|
||||
text(
|
||||
"INSERT INTO documents (id, source, path, full_path, title, content,"
|
||||
" content_hash) VALUES (:id, :s, :p, :fp, :t, :c, :h)"
|
||||
),
|
||||
{
|
||||
"id": doc_id,
|
||||
"s": SOURCE,
|
||||
"p": path,
|
||||
"fp": f"/tmp/{path}",
|
||||
"t": f"Doc {path}",
|
||||
"c": CONTENT,
|
||||
"h": CONTENT_HASH,
|
||||
},
|
||||
)
|
||||
db.commit()
|
||||
return doc_id
|
||||
|
||||
|
||||
def _delete(db: Session, doc_id: uuid.UUID) -> None:
|
||||
db.execute(text("DELETE FROM documents WHERE id = :id"), {"id": doc_id})
|
||||
db.commit()
|
||||
|
||||
|
||||
def _assert_0024_schema_survives(db: Session) -> None:
|
||||
"""The 0024 documents contract must survive the additive upgrade —
|
||||
in particular ``content_hash`` stays NOT NULL and indexed (the
|
||||
retriever's identity — untouched by phase 147)."""
|
||||
content = _column(db, "content_hash")
|
||||
assert content is not None and content[0] == "character varying" and content[1] == "NO", (
|
||||
"documents.content_hash (0001) must keep its 0024 contract"
|
||||
)
|
||||
assert content[3] == 64, "documents.content_hash must keep String(64)"
|
||||
assert (
|
||||
_index_covering(db, "content_hash") is not None
|
||||
), "the ix_documents_content_hash index must survive the upgrade"
|
||||
|
||||
indexed = _column(db, "indexed_at")
|
||||
assert indexed is not None and indexed[0] == "timestamp with time zone", (
|
||||
"documents.indexed_at (0001) must survive the upgrade"
|
||||
)
|
||||
created = _column(db, "created_at")
|
||||
assert created is not None and created[1] == "NO", (
|
||||
"documents.created_at (0020) must survive the upgrade"
|
||||
)
|
||||
manual = _column(db, "created_at_manual")
|
||||
assert manual is not None and manual[0] == "boolean" and manual[1] == "NO", (
|
||||
"documents.created_at_manual (0020) must survive the upgrade"
|
||||
)
|
||||
is_image = _column(db, "is_image")
|
||||
assert is_image is not None and is_image[0] == "boolean" and is_image[1] == "NO", (
|
||||
"documents.is_image (0022) must survive the upgrade"
|
||||
)
|
||||
image_path = _column(db, "image_path")
|
||||
assert image_path is not None and image_path[1] == "YES", (
|
||||
"documents.image_path (0022) must survive the upgrade"
|
||||
)
|
||||
image_ctx = _column(db, "image_context")
|
||||
assert image_ctx is not None and image_ctx[1] == "YES", (
|
||||
"documents.image_context (0023) must survive the upgrade"
|
||||
)
|
||||
# Match by definition, not name: the dev DB's ``documents`` indexes
|
||||
# carry a one-off repair prefix (``documents_heal_new_…_*`` — a
|
||||
# manual dev-DB heal, never a migration), so the name itself is not
|
||||
# a stable pin.
|
||||
uq = db.execute(
|
||||
text(
|
||||
"SELECT indexname FROM pg_indexes"
|
||||
" WHERE tablename = 'documents'"
|
||||
" AND indexdef LIKE 'CREATE UNIQUE INDEX %USING btree (source, path)'",
|
||||
)
|
||||
).scalar()
|
||||
assert uq is not None, (
|
||||
"the unique (source, path) index must survive the upgrade"
|
||||
)
|
||||
|
||||
|
||||
def test_upgrade_to_0025_adds_file_hash(db: Session, alembic: Config) -> None:
|
||||
"""Upgrade 0024 → 0025: ``file_hash`` exists with the full contract
|
||||
(VARCHAR(64), NULLABLE, NO server default — NULL = "unknown" for
|
||||
pre-phase rows — and UNINDEXED: the fast path compares it in Python
|
||||
after the unique ``(source, path)`` lookup), is ABSENT at 0024, and
|
||||
every pre-0025 row backfills ``file_hash`` to NULL (no backfill
|
||||
possible — a migration cannot see the source files on disk) — while
|
||||
the 0024 table contract (incl. the INDEXED ``content_hash``)
|
||||
survives."""
|
||||
command.downgrade(alembic, "0024") # start from the pre-0025 state
|
||||
assert _version(db) == "0024"
|
||||
assert _column(db, "file_hash") is None, "file_hash must be absent at 0024"
|
||||
|
||||
pre_id = _insert_sql(db, "pre.md") # the 0024 column set
|
||||
try:
|
||||
command.upgrade(alembic, "0025")
|
||||
assert _version(db) == "0025", "alembic_version must be at 0025"
|
||||
|
||||
col = _column(db, "file_hash")
|
||||
assert col is not None, "documents.file_hash is missing"
|
||||
assert col[0] == "character varying", "file_hash must be VARCHAR"
|
||||
assert col[1] == "YES", "file_hash must be NULLABLE (NULL = unknown)"
|
||||
assert col[2] is None, (
|
||||
"file_hash must carry NO server default — NULL is the meaningful"
|
||||
" pre-phase state"
|
||||
)
|
||||
assert col[3] == 64, "file_hash must be VARCHAR(64) — a sha256 hex digest"
|
||||
assert _index_covering(db, "file_hash") is None, (
|
||||
"file_hash must be UNINDEXED — it is compared in Python after the"
|
||||
" unique (source, path) lookup, never a query surface"
|
||||
)
|
||||
|
||||
# The pre-0025 row backfilled to NULL (the pre-phase shape — one
|
||||
# full pipeline pass stamps it on the next sync).
|
||||
assert db.execute(
|
||||
text("SELECT content_hash, file_hash FROM documents WHERE id = :id"),
|
||||
{"id": pre_id},
|
||||
).fetchone() == (CONTENT_HASH, None), (
|
||||
"every pre-0025 row must backfill file_hash to NULL — no"
|
||||
" backfill, no default — while content_hash survives"
|
||||
)
|
||||
|
||||
_assert_0024_schema_survives(db)
|
||||
finally:
|
||||
_delete(db, pre_id)
|
||||
|
||||
|
||||
def test_orm_file_hash_round_trips(db: Session, alembic: Config) -> None:
|
||||
"""The ORM contract agrees with the column contract: a freshly
|
||||
inserted ``Document`` with NO ``file_hash`` reads ``file_hash is
|
||||
None`` through a FRESH session (the pre-phase shape — NULL until the
|
||||
first post-upgrade sync stamps it), and an explicit digest
|
||||
round-trips the value."""
|
||||
command.upgrade(alembic, "head")
|
||||
row_stamped = Document(
|
||||
source=SOURCE,
|
||||
path="stamped.md",
|
||||
full_path="/tmp/stamped.md",
|
||||
title="stamped",
|
||||
content=CONTENT,
|
||||
content_hash=CONTENT_HASH,
|
||||
file_hash=FILE_HASH,
|
||||
)
|
||||
row_unstamped = Document(
|
||||
source=SOURCE,
|
||||
path="unstamped.md",
|
||||
full_path="/tmp/unstamped.md",
|
||||
title="unstamped",
|
||||
content=CONTENT,
|
||||
content_hash=CONTENT_HASH,
|
||||
)
|
||||
db.add(row_stamped)
|
||||
db.add(row_unstamped)
|
||||
db.commit()
|
||||
try:
|
||||
with SessionLocal() as fresh:
|
||||
reloaded_stamped = fresh.get(Document, row_stamped.id)
|
||||
assert reloaded_stamped is not None, "the stamped row must be readable"
|
||||
assert reloaded_stamped.file_hash == FILE_HASH, (
|
||||
"the explicit file_hash must round-trip through the DB"
|
||||
)
|
||||
reloaded_unstamped = fresh.get(Document, row_unstamped.id)
|
||||
assert reloaded_unstamped is not None, "the unstamped row must be readable"
|
||||
assert reloaded_unstamped.file_hash is None, (
|
||||
"a row created without file_hash must read NULL (the pre-phase"
|
||||
" shape)"
|
||||
)
|
||||
assert reloaded_unstamped.content_hash == CONTENT_HASH, (
|
||||
"content_hash must be untouched by the file_hash addition"
|
||||
)
|
||||
finally:
|
||||
_delete(db, row_stamped.id)
|
||||
_delete(db, row_unstamped.id)
|
||||
|
||||
|
||||
def test_downgrade_to_0024_drops_the_column(db: Session, alembic: Config) -> None:
|
||||
"""Downgrade 0025 → 0024: the file_hash column is gone (A13 — fully
|
||||
reversible, the hash is re-computable from the source files) while
|
||||
the row + its content (incl. the stamped file_hash value's sibling
|
||||
``content_hash``) survive, and the rest of the 0024 table contract
|
||||
is intact."""
|
||||
command.upgrade(alembic, "head")
|
||||
row = Document(
|
||||
source=SOURCE,
|
||||
path="down.md",
|
||||
full_path="/tmp/down.md",
|
||||
title="down",
|
||||
content=CONTENT,
|
||||
content_hash=CONTENT_HASH,
|
||||
file_hash=FILE_HASH,
|
||||
)
|
||||
db.add(row)
|
||||
db.commit()
|
||||
try:
|
||||
command.downgrade(alembic, "0024")
|
||||
assert _version(db) == "0024"
|
||||
assert _column(db, "file_hash") is None, "file_hash must be dropped"
|
||||
|
||||
surviving = db.execute(
|
||||
text(
|
||||
"SELECT source, path, content, content_hash, is_image, image_path,"
|
||||
" image_context FROM documents WHERE id = :id"
|
||||
),
|
||||
{"id": row.id},
|
||||
).fetchone()
|
||||
assert surviving is not None and surviving[0] == SOURCE, (
|
||||
"the row must survive the column drop"
|
||||
)
|
||||
assert surviving[1] == "down.md" and surviving[2] == CONTENT, (
|
||||
"path + content must survive the drop"
|
||||
)
|
||||
assert surviving[3] == CONTENT_HASH, (
|
||||
"content_hash (the retriever's identity) must survive the drop"
|
||||
)
|
||||
assert surviving[4] is False and surviving[5] is None and surviving[6] is None, (
|
||||
"is_image + image_path + image_context must survive the drop"
|
||||
)
|
||||
assert _index_covering(db, "content_hash") is not None, (
|
||||
"the content_hash index must survive the downgrade"
|
||||
)
|
||||
finally:
|
||||
_delete(db, row.id)
|
||||
# Repair: the fixture teardown re-upgrades to head.
|
||||
|
||||
|
||||
def test_upgrade_round_trip_restores_the_column(db: Session, alembic: Config) -> None:
|
||||
"""Downgrade to 0024, then upgrade back to 0025: ``file_hash`` is
|
||||
back with the full contract (VARCHAR(64), NULLABLE, NO server
|
||||
default, UNINDEXED), a row seeded at 0024 reads ``file_hash is
|
||||
None`` (the pre-phase shape on the round-trip), and the 0024 schema
|
||||
(incl. the indexed ``content_hash``) is intact."""
|
||||
command.downgrade(alembic, "0024")
|
||||
pre_id = _insert_sql(db, "round.md")
|
||||
try:
|
||||
command.upgrade(alembic, "0025")
|
||||
assert _version(db) == "0025", "round-trip upgrade must land at 0025"
|
||||
|
||||
col = _column(db, "file_hash")
|
||||
assert col is not None, "documents.file_hash must be back"
|
||||
assert col[0] == "character varying", "file_hash must be VARCHAR after the round-trip"
|
||||
assert col[1] == "YES", "file_hash must be NULLABLE after the round-trip"
|
||||
assert col[2] is None, (
|
||||
"file_hash must still carry NO server default after the round-trip"
|
||||
)
|
||||
assert col[3] == 64, "file_hash must be VARCHAR(64) after the round-trip"
|
||||
assert _index_covering(db, "file_hash") is None, (
|
||||
"file_hash must stay UNINDEXED after the round-trip"
|
||||
)
|
||||
|
||||
assert db.execute(
|
||||
text("SELECT file_hash FROM documents WHERE id = :id"),
|
||||
{"id": pre_id},
|
||||
).scalar() is None, (
|
||||
"a row seeded at 0024 must read file_hash NULL after the round-trip"
|
||||
)
|
||||
|
||||
_assert_0024_schema_survives(db)
|
||||
finally:
|
||||
_delete(db, pre_id)
|
||||
+17
-17
@@ -1990,18 +1990,19 @@ def test_fallback_unrenderable_page_gets_the_placeholder(
|
||||
_cleanup_source(db, "halfblind")
|
||||
|
||||
|
||||
def test_fallback_rerun_lands_unchanged_no_reembed(
|
||||
def test_fallback_rerun_unchanged_fast_path_no_reburn(
|
||||
db, tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Digest stability (the task's re-burn guard): the digest runs
|
||||
over the constructed fallback content — a re-sync of an unchanged
|
||||
garbage PDF (same pages + the deterministic fake's same replies)
|
||||
re-imports as UNCHANGED: the fallback re-runs (the gate + the
|
||||
render are cheap; the vision calls are re-issued — the documented
|
||||
no-cache cost), but the digest MATCHES, so the row is skipped —
|
||||
no re-chunk, no re-embed, no re-summary, no KB change (the row is
|
||||
not ``updated`` and records no changed path). The doc's content
|
||||
and digest are byte-stable across the re-sync."""
|
||||
"""Phase 147 (task 02) — the owner's core complaint, fixed: a
|
||||
re-sync of an unchanged garbage PDF (same raw bytes) takes the
|
||||
FAST PATH at the top of ``_index_file`` (the file hash matches)
|
||||
— the fallback does NOT re-run: the gate never fires
|
||||
(``pdfs_screenshotted`` stays 0 on run 2), no re-extract, no
|
||||
re-render, and the vision calls are NOT re-issued (the previous
|
||||
"vision calls re-issued before the unchanged check can skip them"
|
||||
no-cache cost is gone). The row re-imports as UNCHANGED — no
|
||||
re-chunk, no re-embed, no KB change (not ``updated``, no changed
|
||||
path) — and the doc's content/digest are byte-stable."""
|
||||
root = tmp_path / "stable"
|
||||
root.mkdir()
|
||||
_build_garbage_pdf(root / "stable.pdf")
|
||||
@@ -2010,19 +2011,18 @@ def test_fallback_rerun_lands_unchanged_no_reembed(
|
||||
try:
|
||||
s1 = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (s1.added, s1.pdfs_screenshotted) == (1, 1)
|
||||
assert llm.vision_calls == 2
|
||||
assert llm.vision_calls == 2 # one per page — the first index
|
||||
embeds1 = len(llm.calls)
|
||||
doc1 = _pdf_doc(db, "stable", "stable.pdf")
|
||||
content1, digest1 = doc1.content, doc1.content_hash
|
||||
|
||||
s2 = asyncio.run(import_sources([root], llm, session=db))
|
||||
# The fallback re-ran (gate re-fired, vision re-issued) — but
|
||||
# the row itself re-imported as unchanged (no re-embed, no
|
||||
# re-chunk, no KB change: the stable replies give the same
|
||||
# digest, so the upsert is skipped entirely).
|
||||
# The fast path fired (file hash match) — the fallback never
|
||||
# ran on run 2: no gate, no render, no vision re-issued, no
|
||||
# re-embed, no KB change.
|
||||
assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
|
||||
assert s2.pdfs_screenshotted == 1 # the gate re-fired
|
||||
assert llm.vision_calls == 4 # the vision calls were re-issued
|
||||
assert s2.pdfs_screenshotted == 0 # the gate never re-fired
|
||||
assert llm.vision_calls == 2 # NO re-issued vision call
|
||||
assert len(llm.calls) == embeds1, "unchanged: no re-embed"
|
||||
assert s2.changed_paths == [], "unchanged: no KB change recorded"
|
||||
|
||||
|
||||
@@ -0,0 +1,626 @@
|
||||
"""Phase 147, task 02 — the unchanged fast path (unit matrix).
|
||||
|
||||
The owner's TODO L4: "Local sources, like PDFs, are synced every time
|
||||
even if they haven't changed. This results in really long sync times
|
||||
for PDFs that are mostly pictoral." The fix: the sha256 of the RAW
|
||||
source file's bytes (:func:`app.rag.importer._file_hash` — streamed in
|
||||
1 MiB chunks) is computed ONCE at the top of ``_index_file``, BEFORE
|
||||
the image/text/PDF dispatch, and compared against
|
||||
``documents.file_hash`` (migration 0025). A match returns through the
|
||||
shared unchanged path (``_unchanged_doc``) with ZERO llm calls (embed
|
||||
+ chat, both counted through the fakes), ZERO PDF extraction, and
|
||||
ZERO vision work — while the preserved unchanged semantics (the
|
||||
phase-118 summary-NULL backfill, the phase-106 D1 manual-date lock,
|
||||
the D4 date refresh, the counts, the log wordings) survive verbatim.
|
||||
|
||||
Pinned here (the load-bearing pins are the ZERO-llm-call assertions on
|
||||
the second sync):
|
||||
|
||||
* the fast path fires for all three document classes (text, image,
|
||||
garbage-layer PDF, pictorial PDF) and returns before any branch code;
|
||||
* a changed file (any byte) re-indexes fully and re-stamps the hash;
|
||||
* a bytes-changed/content-same file (a PDF metadata edit) re-runs the
|
||||
branch's pipeline but lands unchanged (no content change) and
|
||||
stamps the new hash — the next sync is fast;
|
||||
* a pre-phase row (``file_hash`` NULL) runs the branch's content-digest
|
||||
path (today's behavior) and gets STAMPED — the next sync is fast;
|
||||
* the D1 manual-date lock and the D4 date refresh (may go older,
|
||||
counts ``dates_updated``) hold on the fast path;
|
||||
* the phase-118 backfill on the fast path runs from the STORED
|
||||
``doc.content`` (one ``is_summary`` chunk, no content re-embed);
|
||||
* the streamed hash is transparent (a >1 MiB file hashes identically
|
||||
to ``hashlib.sha256(bytes).hexdigest()``).
|
||||
|
||||
The E2E surface (the idempotent second sync through the real UI +
|
||||
mock LLM) is pinned in ``tests/e2e/test_sync_changed_only.py``
|
||||
(task 03).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
import pymupdf as fitz
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.config import Settings
|
||||
from app.models import Document
|
||||
from app.rag.importer import _file_hash, import_sources
|
||||
from tests.fakes import FakeEmbedder
|
||||
from tests.unit.test_importer import (
|
||||
_build_garbage_pdf,
|
||||
_PageVisionLLM,
|
||||
_pdf_settings,
|
||||
_png_bytes,
|
||||
_VisionLLM,
|
||||
)
|
||||
from tests.unit.test_pictorial_import import (
|
||||
_build_ikea_like_pdf,
|
||||
_PictorialLLM,
|
||||
_settings,
|
||||
)
|
||||
|
||||
#: mtime granularity tolerance (the ``test_importer_dates.py``
|
||||
#: convention).
|
||||
_TOL = timedelta(milliseconds=50)
|
||||
|
||||
OLD_2021 = datetime(2021, 6, 1, 12, 0, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
def _utime(path: Path, when: datetime) -> None:
|
||||
ts = when.timestamp()
|
||||
os.utime(path, (ts, ts))
|
||||
|
||||
|
||||
def _doc(db, source: str, rel: str) -> Document:
|
||||
doc = db.scalar(select(Document).where(Document.source == source, Document.path == rel))
|
||||
assert doc is not None, f"no documents row for ({source!r}, {rel!r})"
|
||||
return doc
|
||||
|
||||
|
||||
def _cleanup_source(db, source: str) -> None:
|
||||
for doc in db.scalars(select(Document).where(Document.source == source)).all():
|
||||
db.delete(doc)
|
||||
db.commit()
|
||||
|
||||
|
||||
# ---------- the streamed hash -------------------------------------------------
|
||||
|
||||
|
||||
def test_file_hash_streamed_matches_direct_hash(tmp_path: Path) -> None:
|
||||
"""The chunking is transparent: a >1 MiB file (1.5 MiB here —
|
||||
spans multiple 1 MiB reads) hashes identically to
|
||||
``hashlib.sha256(bytes).hexdigest()``, an empty file hashes the
|
||||
empty byte string, and the result is a 64-char lowercase hex
|
||||
digest (the ``documents.file_hash`` column width)."""
|
||||
block = hashlib.sha256(b"seed-147").digest() # 32 B
|
||||
data = block * 49152 # 1.5 MiB — well past one 1 MiB chunk
|
||||
big = tmp_path / "big.bin"
|
||||
big.write_bytes(data)
|
||||
expected = hashlib.sha256(data).hexdigest()
|
||||
assert len(expected) == 64 and expected == expected.lower()
|
||||
assert _file_hash(big) == expected
|
||||
|
||||
empty = tmp_path / "empty.bin"
|
||||
empty.write_bytes(b"")
|
||||
assert _file_hash(empty) == hashlib.sha256(b"").hexdigest()
|
||||
|
||||
|
||||
# ---------- the fast path: zero llm on the second sync ------------------------
|
||||
|
||||
|
||||
def test_text_unchanged_second_sync_costs_zero_llm(db, tmp_path: Path) -> None:
|
||||
"""Text class: index a file, capture the llm call count (embed +
|
||||
chat both counted); re-run → ``unchanged == 1``, ZERO new llm
|
||||
calls, ``files == 1``, the doc's ``indexed_at``/``content_hash``
|
||||
untouched, and ``file_hash`` stamped after run 1 (equal to
|
||||
``_file_hash`` of the raw bytes)."""
|
||||
root = tmp_path / "fasttext"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\nmarkdown body\n")
|
||||
llm = FakeEmbedder()
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (first.added, first.unchanged, first.files) == (1, 0, 1)
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
# Stamped after run 1 — the raw-bytes digest.
|
||||
assert doc.file_hash == _file_hash(path)
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
indexed1, digest1 = doc.indexed_at, doc.content_hash
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged, second.files) == (0, 0, 1, 1)
|
||||
# THE pin: zero new llm calls of any kind.
|
||||
assert len(llm.calls) == embeds1, "fast path: no re-embed"
|
||||
assert len(llm.chat_calls) == chats1, "fast path: no lite/vision call"
|
||||
assert second.changed_paths == [] # no KB change recorded
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.indexed_at == indexed1 # the re-index did not happen
|
||||
assert doc2.content_hash == digest1
|
||||
assert doc2.file_hash == _file_hash(path)
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_garbage_pdf_unchanged_second_sync_costs_zero_vision(db, tmp_path: Path) -> None:
|
||||
"""Garbage-layer PDF class (the owner's core complaint): a no-text
|
||||
layer's PDF indexed through the page-screenshot fallback (two
|
||||
vision calls); re-run → ``unchanged == 1``, ZERO new llm calls —
|
||||
the fallback vision did NOT re-fire (the pre-phase "vision
|
||||
re-issued before the unchanged check can skip them" cost is
|
||||
gone), the gate never re-fires, and the content row is
|
||||
byte-identical."""
|
||||
root = tmp_path / "fastgarbage"
|
||||
root.mkdir()
|
||||
path = root / "garbage.pdf"
|
||||
_build_garbage_pdf(path)
|
||||
llm = _PageVisionLLM()
|
||||
llm.settings = _pdf_settings(tmp_path)
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (first.added, first.pdfs_screenshotted) == (1, 1)
|
||||
assert llm.vision_calls == 2 # one per page — the first index
|
||||
doc = _doc(db, root.name, "garbage.pdf")
|
||||
assert doc.file_hash == _file_hash(path)
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
content1, digest1, indexed1 = doc.content, doc.content_hash, doc.indexed_at
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.pdfs_screenshotted == 0 # the gate never re-fired
|
||||
# THE pin: the fallback vision did NOT re-fire.
|
||||
assert llm.vision_calls == 2, "fast path: no re-issued vision call"
|
||||
assert len(llm.chat_calls) == chats1
|
||||
assert len(llm.calls) == embeds1, "fast path: no re-embed"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "garbage.pdf")
|
||||
assert doc2.content == content1 # content row byte-identical
|
||||
assert doc2.content_hash == digest1
|
||||
assert doc2.indexed_at == indexed1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_pictorial_pdf_unchanged_second_sync_costs_zero_vision(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""Pictorial PDF class (the IKEA assembly-instruction class,
|
||||
TODO L5): a sparse-page PDF indexed with the pictorial pass ON
|
||||
(six vision calls); re-run → ``unchanged == 1``, ZERO new llm
|
||||
calls — the pictorial pass did NOT re-fire (no re-render, no
|
||||
vision), and the content is byte-stable."""
|
||||
root = tmp_path / "fastikea"
|
||||
root.mkdir()
|
||||
path = root / "ikea.pdf"
|
||||
_build_ikea_like_pdf(path)
|
||||
llm = _PictorialLLM()
|
||||
llm.settings = _settings(tmp_path) # pdf_pictorial ON (code default)
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (first.added, first.pdfs_pictorial, first.pdf_pictorial_pages) == (1, 1, 6)
|
||||
assert llm.pictorial_calls == 6 # one per candidate — the first index
|
||||
doc = _doc(db, root.name, "ikea.pdf")
|
||||
assert doc.file_hash == _file_hash(path)
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
content1, digest1 = doc.content, doc.content_hash
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.pdfs_pictorial == 0 # the pass did not re-fire
|
||||
# THE pin: the pictorial vision did NOT re-fire.
|
||||
assert llm.pictorial_calls == 6, "fast path: no re-issued vision call"
|
||||
assert len(llm.chat_calls) == chats1
|
||||
assert len(llm.calls) == embeds1, "fast path: no re-embed"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "ikea.pdf")
|
||||
assert doc2.content == content1 # byte-stable content
|
||||
assert doc2.content_hash == digest1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_image_unchanged_second_sync_costs_zero_llm(db, tmp_path: Path) -> None:
|
||||
"""Image class: a standalone image indexed through the vision
|
||||
describe (one chat call); re-run → ``unchanged == 1``, ZERO new
|
||||
llm calls — the describe did NOT re-fire, no re-embed, and the
|
||||
row (content + ``indexed_at``) is untouched."""
|
||||
root = tmp_path / "fastimg"
|
||||
root.mkdir()
|
||||
path = root / "pic.png"
|
||||
path.write_bytes(_png_bytes(4, 4, (0.2, 0.4, 0.6)))
|
||||
llm = _VisionLLM()
|
||||
llm.settings = Settings(
|
||||
_env_file=None, # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
|
||||
images=True,
|
||||
image_dir=str(tmp_path / "images"),
|
||||
)
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (first.added, first.images_failed) == (1, 0)
|
||||
doc = _doc(db, root.name, "pic.png")
|
||||
assert doc.file_hash == _file_hash(path)
|
||||
assert doc.content == _VisionLLM.DESCRIPTION
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
indexed1 = doc.indexed_at
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
# THE pin: no re-describe, no re-embed.
|
||||
assert len(llm.chat_calls) == chats1, "fast path: no re-describe"
|
||||
assert len(llm.calls) == embeds1, "fast path: no re-embed"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "pic.png")
|
||||
assert doc2.content == _VisionLLM.DESCRIPTION
|
||||
assert doc2.indexed_at == indexed1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
# ---------- changed files: full re-index + re-stamp ---------------------------
|
||||
|
||||
|
||||
def test_changed_file_reindexes_and_re_stamps(db, tmp_path: Path) -> None:
|
||||
"""A changed file (one byte flipped in a text file) re-indexes
|
||||
FULLY — the llm calls fire again (embed + summary) — and
|
||||
``file_hash`` is RE-STAMPED with the new raw-bytes digest;
|
||||
``content_hash`` moves with the content."""
|
||||
root = tmp_path / "fastchange"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\nv1\n")
|
||||
llm = FakeEmbedder()
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert first.added == 1
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
hash1, content_digest1 = doc.file_hash, doc.content_hash
|
||||
assert hash1 == _file_hash(path)
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
|
||||
path.write_text("# Note\n\nv2\n") # the bytes changed
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 1, 0)
|
||||
# The full pipeline re-fired.
|
||||
assert len(llm.calls) > embeds1, "a changed file re-embeds"
|
||||
assert len(llm.chat_calls) > chats1, "a changed file re-summarizes"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.file_hash == _file_hash(path) # re-stamped
|
||||
assert doc2.file_hash != hash1
|
||||
assert doc2.content_hash != content_digest1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_pdf_bytes_changed_content_same_stamps_without_content_change(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""The bytes-changed/content-same edge (a PDF metadata edit with
|
||||
identical extracted text): the fast path does NOT fire (the raw
|
||||
bytes moved), the branch re-runs its pipeline (the fallback's
|
||||
vision re-issues — the honest cost of a byte change), but the
|
||||
CONTENT digest matches — the row is UNCHANGED (no content change,
|
||||
no re-embed), and the shared helper STAMPS the new hash, so the
|
||||
NEXT sync takes the fast path (zero further vision)."""
|
||||
root = tmp_path / "fastmeta"
|
||||
root.mkdir()
|
||||
path = root / "meta.pdf"
|
||||
_build_garbage_pdf(path)
|
||||
llm = _PageVisionLLM()
|
||||
llm.settings = _pdf_settings(tmp_path)
|
||||
try:
|
||||
first = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (first.added, first.pdfs_screenshotted) == (1, 1)
|
||||
doc = _doc(db, root.name, "meta.pdf")
|
||||
content1, digest1 = doc.content, doc.content_hash
|
||||
hash1 = doc.file_hash
|
||||
assert llm.vision_calls == 2
|
||||
embeds1 = len(llm.calls)
|
||||
|
||||
# A metadata-only edit: the raw bytes change, the extracted
|
||||
# text (and the rendered pages) do not.
|
||||
with fitz.open(str(path)) as d:
|
||||
d.set_metadata({"title": "metadata-only edit"})
|
||||
d.saveIncr()
|
||||
new_hash = _file_hash(path)
|
||||
assert new_hash != hash1 # the bytes really moved
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
# The branch re-ran (gate re-fired, vision re-issued — the byte
|
||||
# change's cost) but the content digest matched: the row is
|
||||
# unchanged, no content change, no re-embed.
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.pdfs_screenshotted == 1 # the branch re-ran its gate
|
||||
assert llm.vision_calls == 4 # the fallback re-issued (bytes changed)
|
||||
assert len(llm.calls) == embeds1, "content-same: no re-embed"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "meta.pdf")
|
||||
assert doc2.content == content1 # content byte-identical
|
||||
assert doc2.content_hash == digest1
|
||||
assert doc2.file_hash == new_hash # STAMPED by the shared helper
|
||||
|
||||
# The NEXT sync is fast (the stamp landed): zero further vision.
|
||||
third = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (third.added, third.updated, third.unchanged) == (0, 0, 1)
|
||||
assert third.pdfs_screenshotted == 0
|
||||
assert llm.vision_calls == 4, "the third sync is the fast path"
|
||||
assert len(llm.calls) == embeds1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
# ---------- pre-phase rows (NULL file_hash) -----------------------------------
|
||||
|
||||
|
||||
def test_pre_phase_null_file_hash_runs_branch_path_and_stamps(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""A row that predates the column (``file_hash`` NULL) + an
|
||||
unchanged file: the fast path cannot fire, so the branch's
|
||||
content-digest path runs (today's behavior — the digest matches,
|
||||
no re-embed), and the shared helper STAMPS the hash — the THIRD
|
||||
sync then takes the fast path (proof the stamp landed)."""
|
||||
root = tmp_path / "fastlegacy"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\nmarkdown body\n")
|
||||
llm = FakeEmbedder()
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
doc.file_hash = None # simulate the pre-phase row
|
||||
db.commit()
|
||||
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
# The content digest matched — no re-embed, no re-summary.
|
||||
assert len(llm.calls) == embeds1
|
||||
assert len(llm.chat_calls) == chats1
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.file_hash == _file_hash(path) # stamped by the helper
|
||||
|
||||
# The third sync takes the FAST PATH (the stamp is committed):
|
||||
third = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert third.unchanged == 1
|
||||
assert len(llm.calls) == embeds1
|
||||
assert len(llm.chat_calls) == chats1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_pre_phase_null_file_hash_image_runs_branch_and_stamps(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""The image flavour of the branch path: a pre-phase row
|
||||
(``file_hash`` NULL) + an unchanged image — the fast path cannot
|
||||
fire, so the image branch's content-digest check runs (the raw
|
||||
BYTES digest — the image's content identity — matches BEFORE the
|
||||
expensive describe, so no re-describe, no re-embed), the row is
|
||||
unchanged, and the shared helper STAMPS the hash — the next sync
|
||||
is fast.
|
||||
"""
|
||||
root = tmp_path / "fastimglegacy"
|
||||
root.mkdir()
|
||||
path = root / "pic.png"
|
||||
path.write_bytes(_png_bytes(4, 4, (0.2, 0.4, 0.6)))
|
||||
llm = _VisionLLM()
|
||||
llm.settings = Settings(
|
||||
_env_file=None, # pyright: ignore[reportCallIssue] (kwarg exists at runtime)
|
||||
images=True,
|
||||
image_dir=str(tmp_path / "images"),
|
||||
)
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
doc = _doc(db, root.name, "pic.png")
|
||||
doc.file_hash = None # simulate the pre-phase row
|
||||
db.commit()
|
||||
|
||||
chats1 = len(llm.chat_calls)
|
||||
embeds1 = len(llm.calls)
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
# The branch's byte-digest check matched before the describe —
|
||||
# zero llm work of any kind.
|
||||
assert len(llm.chat_calls) == chats1
|
||||
assert len(llm.calls) == embeds1, "content-same: no re-embed"
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "pic.png")
|
||||
assert doc2.file_hash == _file_hash(path) # stamped by the helper
|
||||
assert doc2.content == _VisionLLM.DESCRIPTION
|
||||
|
||||
# The next sync takes the fast path (zero further calls):
|
||||
third = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert third.unchanged == 1
|
||||
assert len(llm.chat_calls) == chats1
|
||||
assert len(llm.calls) == embeds1
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
# ---------- the preserved semantics on the fast path --------------------------
|
||||
|
||||
|
||||
def test_manual_date_row_fast_path_no_date_write(db, tmp_path: Path) -> None:
|
||||
"""D1 on the fast path: an unchanged file carrying the owner's
|
||||
correction (``created_at_manual``) gets NO date write — the moved
|
||||
mtime does not refresh it, ``dates_updated`` stays 0, the
|
||||
correction + the flag survive, and the row's stamp is intact
|
||||
(zero llm calls)."""
|
||||
root = tmp_path / "fastmanual"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\ncontent for the KB\n")
|
||||
llm = FakeEmbedder()
|
||||
correction = datetime(2023, 5, 5, 9, 30, 0, tzinfo=UTC)
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
doc.created_at = correction
|
||||
doc.created_at_manual = True
|
||||
db.commit()
|
||||
_utime(path, OLD_2021) # the source moved to a DIFFERENT date
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.dates_updated == 0 # the manual lock withheld it
|
||||
assert len(llm.calls) == embeds1 and len(llm.chat_calls) == chats1
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.created_at == correction # the correction survived
|
||||
assert doc2.created_at_manual is True # the flag intact
|
||||
assert doc2.file_hash == _file_hash(path) # the stamp intact
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_manual_date_null_hash_row_stamps_without_date_write(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""D1 on the BRANCH path (a pre-phase NULL-hash row that also
|
||||
carries the manual correction): the shared helper's stamp is the
|
||||
only write — it COMMITS (so the next sync is fast even for a
|
||||
manual-date row) — while the DATE stays the manual value
|
||||
(``dates_updated`` 0, the flag intact)."""
|
||||
root = tmp_path / "fastmanualnull"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\ncontent for the KB\n")
|
||||
llm = FakeEmbedder()
|
||||
correction = datetime(2023, 5, 5, 9, 30, 0, tzinfo=UTC)
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
doc.created_at = correction
|
||||
doc.created_at_manual = True
|
||||
doc.file_hash = None # pre-phase row
|
||||
db.commit()
|
||||
_utime(path, OLD_2021) # a refresh is DUE — the lock must withhold it
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.dates_updated == 0
|
||||
# No summary work of any kind on this run (run 1's summary
|
||||
# count is the baseline).
|
||||
assert len(llm.chat_calls) == chats1
|
||||
assert len(llm.calls) == embeds1
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.created_at == correction # the correction survived
|
||||
assert doc2.created_at_manual is True
|
||||
assert doc2.file_hash == _file_hash(path) # the stamp committed
|
||||
|
||||
# The next sync is fast (the stamp landed despite the D1 lock):
|
||||
third = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert third.unchanged == 1 and third.dates_updated == 0
|
||||
db.expire_all()
|
||||
assert _doc(db, root.name, "note.md").created_at == correction
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_unchanged_fast_path_date_refresh_may_go_older(
|
||||
db, tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""D4 on the fast path: an unchanged file whose mtime-derived date
|
||||
moved (OLDER — no monotonic guard) refreshes the stored date on
|
||||
the fast path: ``dates_updated == 1``, the date lands, and zero
|
||||
llm calls (the refresh is bookkeeping, not a re-index). The
|
||||
PRE-CHANGE log wordings are pinned verbatim (``import: unchanged``
|
||||
+ ``import: date-refreshed``)."""
|
||||
root = tmp_path / "fastdate"
|
||||
root.mkdir()
|
||||
path = root / "note.md"
|
||||
path.write_text("# Note\n\ncontent for the KB\n")
|
||||
llm = FakeEmbedder()
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
_utime(path, OLD_2021) # the source date moved; content identical
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="app.importer"):
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
|
||||
assert second.dates_updated == 1 # the refresh ran (fast path)
|
||||
assert len(llm.calls) == embeds1 and len(llm.chat_calls) == chats1
|
||||
# The preserved wordings (the shared helper's log lines, verbatim
|
||||
# from the per-branch blocks).
|
||||
messages = [r.getMessage() for r in caplog.records]
|
||||
assert f"import: unchanged source={root.name} path=note.md" in messages
|
||||
assert any(
|
||||
m.startswith(f"import: date-refreshed source={root.name} path=note.md date=")
|
||||
for m in messages
|
||||
)
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert abs(doc2.created_at - OLD_2021) <= _TOL
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
|
||||
|
||||
def test_null_summary_backfill_on_fast_path(db, tmp_path: Path) -> None:
|
||||
"""Phase 118 (A2) via the fast path: an unchanged doc whose
|
||||
summary is still NULL (the stored content is intact) gets the
|
||||
summary-only backfill from the STORED ``doc.content`` — one
|
||||
``is_summary`` chunk, no content re-embed, counted
|
||||
``summary_backfilled`` (never ``summaries``), never
|
||||
added/updated/pruned. The ``lite`` call fires (the backfill's own
|
||||
work — the fast path's "zero llm calls" is "unless the backfill
|
||||
fires"); nothing else does."""
|
||||
root = tmp_path / "fastbfill"
|
||||
root.mkdir()
|
||||
(root / "note.md").write_text("# Note\n\nmarkdown body\n")
|
||||
llm = FakeEmbedder()
|
||||
try:
|
||||
asyncio.run(import_sources([root], llm, session=db))
|
||||
doc = _doc(db, root.name, "note.md")
|
||||
# Simulate the NULL-summary target row: the content stays, only
|
||||
# the summary + its chunk go away (the phase-118 convention).
|
||||
doc.summary = None
|
||||
for c in [c for c in doc.chunks if c.is_summary]:
|
||||
doc.chunks.remove(c)
|
||||
db.commit()
|
||||
assert doc.file_hash is not None # the fast path can fire
|
||||
|
||||
embeds1, chats1 = len(llm.calls), len(llm.chat_calls)
|
||||
second = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (second.added, second.updated, second.pruned) == (0, 0, 0)
|
||||
assert (second.unchanged, second.summary_backfilled, second.summaries) == (1, 1, 0)
|
||||
assert second.summary_errors == 0
|
||||
# Exactly the backfill's own work: one lite call, one embed
|
||||
# batch (the summary chunk only — no content re-embed).
|
||||
assert len(llm.chat_calls) == chats1 + 1
|
||||
assert len(llm.calls) == embeds1 + 1
|
||||
assert second.changed_paths == [] # no KB change recorded
|
||||
|
||||
db.expire_all()
|
||||
doc2 = _doc(db, root.name, "note.md")
|
||||
assert doc2.summary is not None
|
||||
schunks = [c for c in doc2.chunks if c.is_summary]
|
||||
assert len(schunks) == 1 and schunks[0].position == -1
|
||||
assert schunks[0].content == doc2.summary
|
||||
content = [c for c in doc2.chunks if not c.is_summary]
|
||||
assert len(content) == 1 and content[0].embedding is not None
|
||||
finally:
|
||||
_cleanup_source(db, root.name)
|
||||
@@ -128,6 +128,81 @@ def test_documents_unique_source_path() -> None:
|
||||
assert uq, "documents must be unique on (source, path) — the upsert key"
|
||||
|
||||
|
||||
def test_documents_file_hash_column_contract() -> None:
|
||||
"""Phase 147 (task 01): ``documents.file_hash`` — the sha256 of the
|
||||
RAW SOURCE FILE's bytes (the sync's change signal): VARCHAR(64),
|
||||
NULLABLE (NULL = "unknown" — pre-phase rows until the first
|
||||
post-upgrade sync stamps it), NO server default, and NO index
|
||||
(the fast path compares it in Python after the unique
|
||||
``(source, path)`` lookup — internal bookkeeping, never a query
|
||||
surface). ``content_hash`` (0001) stays untouched: NOT NULL and
|
||||
still indexed — the retriever's dedupe/caching identity."""
|
||||
from sqlalchemy import String as SAString
|
||||
|
||||
documents = Base.metadata.tables["documents"]
|
||||
col = documents.c["file_hash"]
|
||||
assert col.nullable is True, ("file_hash must be NULLABLE — NULL is the pre-phase state")
|
||||
assert isinstance(col.type, SAString), "file_hash must be String(64)"
|
||||
assert col.type.length == 64, "file_hash must be String(64) — a sha256 hex digest"
|
||||
assert col.server_default is None, (
|
||||
"file_hash must carry NO server default — NULL is the meaningful"
|
||||
" pre-phase state"
|
||||
)
|
||||
# No index: neither the inline flag (None = unspecified) nor any
|
||||
# table-level index may cover file_hash (unlike content_hash — a
|
||||
# query/exposure surface).
|
||||
assert not col.index, ("file_hash must be UNINDEXED — compared in Python")
|
||||
covering = [
|
||||
idx
|
||||
for idx in documents.indexes
|
||||
if any(c.name == "file_hash" for c in idx.columns)
|
||||
]
|
||||
assert not covering, "no table-level index may cover file_hash"
|
||||
uq_covering = [
|
||||
c
|
||||
for c in documents.constraints
|
||||
if isinstance(c, UniqueConstraint) and any(col.name == "file_hash" for col in c.columns)
|
||||
]
|
||||
assert not uq_covering, "no unique constraint may cover file_hash"
|
||||
|
||||
# content_hash is untouched: still NOT NULL and indexed.
|
||||
content = documents.c["content_hash"]
|
||||
assert content.nullable is False, ("content_hash must stay NOT NULL (0001)")
|
||||
assert content.index is True, "content_hash must stay INDEXED (0001)"
|
||||
content_covering = [
|
||||
idx for idx in documents.indexes if any(c.name == "content_hash" for c in idx.columns)
|
||||
]
|
||||
assert content_covering, "the content_hash index must survive phase 147"
|
||||
|
||||
|
||||
def test_document_file_hash_orm_defaults_to_none() -> None:
|
||||
"""A freshly constructed ``Document`` (no DB) exposes ``file_hash``:
|
||||
the omitted kwarg and explicit ``None`` read NULL (the pre-phase
|
||||
shape — NULL is backfilled with a real digest only by the first
|
||||
post-upgrade sync), and a set digest round-trips as-is (the real
|
||||
DB round-trip is pinned in ``test_migration_0025.py``)."""
|
||||
from app.models import Document
|
||||
|
||||
base = dict(
|
||||
source="s",
|
||||
path="p.md",
|
||||
full_path="/tmp/s/p.md",
|
||||
title="p",
|
||||
content="c",
|
||||
content_hash="1" * 64,
|
||||
)
|
||||
stamped = Document(**base, file_hash="a" * 64)
|
||||
assert stamped.file_hash == "a" * 64
|
||||
|
||||
explicit_null = Document(**base, file_hash=None)
|
||||
assert explicit_null.file_hash is None
|
||||
|
||||
omitted = Document(**base)
|
||||
assert omitted.file_hash is None, (
|
||||
"an omitted file_hash must read NULL (the pre-phase shape)"
|
||||
)
|
||||
|
||||
|
||||
def test_doc_drafts_token_is_unique_not_null() -> None:
|
||||
"""Phase 59: the draft's URL credential — an unguessable uuid4,
|
||||
UNIQUE + NOT NULL (no "un-drafted" state, unlike the NULLable
|
||||
|
||||
@@ -28,11 +28,12 @@ fixtures; the real tar is pinned in
|
||||
others are described), the unreadable-page placeholder skip (the
|
||||
zero-byte ``render_pdf_pages`` slot burns no vision call, the page
|
||||
keeps its text only).
|
||||
* the digest-stability rerun — a stable file + a stable vision reply
|
||||
re-syncs as ``unchanged`` (no re-embed, byte-stable content +
|
||||
digest); the vision calls are re-issued before the unchanged check
|
||||
can skip them — the documented no-cache cost (the phase-133
|
||||
convention, A21 endorses it).
|
||||
* the phase-147 rerun — an unchanged file re-syncs through the FAST
|
||||
PATH (the raw bytes are the change signal): ``unchanged``, no
|
||||
re-embed, byte-stable content + digest — and the vision calls are
|
||||
NOT re-issued (the pass never runs on an unchanged file; the
|
||||
pre-phase "vision re-issued before the unchanged check" no-cache
|
||||
cost is gone).
|
||||
|
||||
The E2E surface (the real UI + the mock LLM's ``PICTORIAL_PAGE_MODE``
|
||||
branch) is pinned in ``tests/e2e/test_pictorial_pdf.py`` (task 04).
|
||||
@@ -565,18 +566,22 @@ def test_unreadable_page_placeholder_skips_the_vision_call(
|
||||
_cleanup_source(db, "halfblind")
|
||||
|
||||
|
||||
def test_rerun_lands_unchanged_with_byte_stable_content(
|
||||
def test_rerun_lands_unchanged_fast_path_no_revision(
|
||||
db, tmp_path: Path
|
||||
) -> None:
|
||||
"""Digest stability (A4 — the phase-126/133 convention): the
|
||||
digest runs over the FINAL content, so a re-sync of an unchanged
|
||||
IKEA-class PDF (same pages + the stable fake's same replies)
|
||||
re-imports as UNCHANGED: no re-chunk, no re-embed, no re-summary,
|
||||
no KB change (not ``updated``, no changed path), and the content +
|
||||
digest are byte-stable. The vision calls ARE re-issued before the
|
||||
unchanged check can skip them — the documented no-cache cost (the
|
||||
phase-133 convention; a non-deterministic vision model re-describes
|
||||
the row each sync — a cost, not a bug; A21 endorses it)."""
|
||||
"""Phase 147 (task 02) — the owner's "really long sync times for
|
||||
PDFs that are mostly pictoral" complaint, fixed: a re-sync of an
|
||||
unchanged IKEA-class PDF (same raw bytes) takes the FAST PATH at
|
||||
the top of ``_index_file`` (the file hash matches) — the
|
||||
pictorial pass does NOT re-run: no re-render, the vision calls are
|
||||
NOT re-issued (``pictorial_calls`` stays at run 1's 6 — the
|
||||
pre-phase "vision re-issued before the unchanged check can skip
|
||||
them" no-cache cost is gone), ``pdfs_pictorial`` stays 0 on run 2.
|
||||
The row re-imports as UNCHANGED: no re-chunk, no re-embed, no
|
||||
re-summary, no KB change (not ``updated``, no changed path), and
|
||||
the content + digest are byte-stable. (A non-deterministic vision
|
||||
model no longer flips the row's digest either — the raw bytes
|
||||
decide.)"""
|
||||
root = tmp_path / "stable"
|
||||
root.mkdir()
|
||||
pdf = root / "ikea.pdf"
|
||||
@@ -586,17 +591,18 @@ def test_rerun_lands_unchanged_with_byte_stable_content(
|
||||
try:
|
||||
s1 = asyncio.run(import_sources([root], llm, session=db))
|
||||
assert (s1.added, s1.pdfs_pictorial, s1.pdf_pictorial_pages) == (1, 1, 6)
|
||||
assert llm.pictorial_calls == 6
|
||||
assert llm.pictorial_calls == 6 # one per candidate — first index
|
||||
embeds1 = len(llm.calls)
|
||||
doc1 = _pdf_doc(db, "stable", "ikea.pdf")
|
||||
content1, digest1 = doc1.content, doc1.content_hash
|
||||
|
||||
s2 = asyncio.run(import_sources([root], llm, session=db))
|
||||
# The pass re-ran (gate re-fired, vision re-issued — the
|
||||
# documented cost) — but the row itself re-imported unchanged.
|
||||
# The fast path fired (file hash match) — the pass never ran
|
||||
# on run 2: no render, no vision re-issued, no re-embed, no KB
|
||||
# change.
|
||||
assert (s2.added, s2.updated, s2.unchanged) == (0, 0, 1)
|
||||
assert s2.pdfs_pictorial == 1 # the pass re-fired
|
||||
assert llm.pictorial_calls == 12 # the vision calls were re-issued
|
||||
assert s2.pdfs_pictorial == 0 # the pass did not re-fire
|
||||
assert llm.pictorial_calls == 6 # NO re-issued vision call
|
||||
assert len(llm.calls) == embeds1, "unchanged: no re-embed"
|
||||
assert s2.changed_paths == [], "unchanged: no KB change recorded"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user