"""Phase 162 E2E (Playwright, mock-only): a 1-document source is BOTH visible to ```` and winnable by its accented question. Run in isolation (DB must be up: ``podman compose up -d db``): uv run pytest tests/e2e/test_search_recall_small_sources.py -v --no-cov The 2026-10-07 live miss, replayed deterministically: the owner asked ``How do I build the BESTÅ?``; the KB held ``ikea_instructions/ besta-frame-white__AA-1272068-11-100.md`` (1 document in the source, its body carrying the name ONLY as ``BESTÅ``) and the question came back empty-handed — ASCII ``besta`` documents outranked the one document that actually named the product. Three separate losses, three pins here: * **D1 (the question-side fold)** — name matching never bridged the accent: ``BESTÅ`` normalized to ``best``, ``besta-frame-…`` to ``besta``, and the digitless rule EQUALS a stem component. Test 1 pins the fix, test 2 MEASURES the miss. * **D2 (the deterministic source inventory)** — the ``lite`` outline is lossy, so a 1-doc source could be invisible to the model. Test 4 pins the mechanical ``Sources:`` tail on the stored row. * **D3 (the grep fold + search discipline)** — the stored accented body stayed grep-matchable by the ASCII name. Test 6 pins the content layer; the turn-level prose discipline is the prompt teaching (``app/rag/prompts.py``, unit-pinned in task 03). KB fixture — ``tests/fixtures/recallkb`` (tracked), THREE local-directory sources registered through the authenticated API + the REAL in-process ``POST /api/sync`` (the ``test_name_signal_read_chips.py`` pattern), six markdown documents whose text is token-controlled so the hybrid gate, the FTS list and the name-hit bonus pick the intended tiers deterministically (``synced_kb`` pins the tiers with the app's REAL retrieval — a fixture-text regression breaks at setup with a measured message): * ``ikea_instructions/`` — the small source: ONE document, the live incident's own filename, its body naming the product ONLY as the accented ``BESTÅ`` (no ASCII ``besta`` anywhere in the file). Two token controls make the mechanism visible: no ``build`` (the question's other FTS token) anywhere, and the accented name deliberately placed PAST the 24-token digest window, so the mock summary chunk leaks no ASCII ``best`` lexeme that a pre-fix FTS could have hit; * ``phase_docs/`` (4 docs) + ``players_handbook/`` (1 doc) — the ASCII distractors of the live index: every body carries ASCII ``besta`` AND ``build``, so each of the five outranks the small source on pre-fix retrieval (two fused terms — FTS + vector — against its one). Measured, never assumed (test 2 re-measures both directions; the same numbers are recorded in ``.agents/reports/162_search_recall_small_sources/``). Accented question ``How do I build the BESTÅ?``, the app's real ``retrieve()`` + ``select_suggested()`` over this KB with the mock's embeddings (gate: threshold 0.30, lexical-support floor 0.15 — the conftest mock calibration; ``best cosine`` / ``fts`` are the A8 inputs): PRE-FIX (the phase-119 ``_normalize_name``, no accent fold) name tokens ['build', 'best'] → NO name hit suggested the five ASCII distractors related ikea_instructions/besta-frame-white__… (rank 6) best cosine 0.382, fts 8 (grounded, on the WRONG documents — the live 2026-10-07 shape: an answer about the ASCII ``besta`` notes) POST-FIX (shipped) name tokens ['build', 'besta'] → name hit suggested THE SMALL SOURCE #1, then four distractors related phase_docs/156_history_source_filter.md best cosine 0.382, fts 9 (the fold adds the lexical term; the cosine is untouched — LOCKED A1) The mechanism is the term COUNT: every distractor matches the question's other FTS token ``build``, so each fuses TWO terms (vector + FTS) while the small source — no ``build``, no accent-bridgeable lexeme — fuses ONE and lands rank 6. The folded name hit is the only signal that can lift it: name hits LEAD the lexical list (``fuse`` rank 1 ⇒ 1/61) and the per-document ``BOR_NAME_HIT_BONUS`` (0.005) rides on top. The ASCII question ``How do I build a Besta?`` keeps the same document suggested #1 with the same cosine, before and after (the untouched phase-119 path, test 3). Test → contract mapping (task 04 cases 1–4): 1. ``test_accented_question_suggests_the_small_source`` — (1): the accented question grounds (``deflected: false``), the durable record's SUGGESTED slice (entries 1–5, LOCKED A3) contains ``ikea_instructions/besta-frame-white__AA-1272068-11-100.md``, and the ``done`` frame's ``related`` list — the rank-6+ remainder — does NOT (the doc moved from the rank-6 miss into the tier). A zero-read grounded turn chips nothing (phase 119 LOCKED A1). 2. ``test_under_rank_is_measured_not_assumed`` — (1, the hypothesis): with the PRE-FIX normalizer re-installed (``monkeypatch``), the same retrieval over the same KB puts the doc OUT of the top-5 (and the ``BESTÅ?`` token degrades to ``best``); with the shipped normalizer it is suggested #1. The under-rank is measured in both directions, so the pin can never silently rot. 3. ``test_ascii_question_still_name_matches`` — (2): the ASCII regression (the phase-119 path untouched) — same document in the suggested tier, grounded, same related remainder. 4. ``test_overview_tail_names_every_source_with_counts`` — (3): the stored ``kb_overview`` row ENDS with the deterministic ``Sources: …`` line, one entry per source with exact counts — the 1-doc sources visible whatever the ``lite`` digest said. 5. ``test_out_of_kb_question_still_deflects`` — (4): the gate verdict for the out-of-KB question is UNCHANGED (``deflected: true``, ``fts_hits == 0``, no chips, ``Maybe try`` chips only) — the fold must not turn a real miss into a false grounding. 6. ``test_grep_folds_the_accented_body`` — (3, content layer): ``grep_document`` with the ASCII pattern ``besta`` matches the stored accented line (D3), and the accented pattern matches too. 7. ``test_suggested_folder_lines_carry_the_small_source`` — the prompt-level witness: the mock's ``repeat your folder map`` echo ends with the suggested-folder lines, and the 1-doc source's line ``ikea_instructions/:`` is among them (its suggested doc owns the line, so the header stands alone). """ from __future__ import annotations import json import os import re import subprocess import sys import time from collections.abc import Iterator from pathlib import Path from typing import Any import httpx import pytest from playwright.sync_api import Locator, Page, expect from sqlalchemy import select, text from app.config import Settings as _Settings from app.db import SessionLocal from app.models import Document, KbOverview, QueryLog from app.rag import retriever from app.rag.agent import grep_document from app.rag.retriever import retrieve, select_related, select_suggested from e2e.auth_helpers import login from e2e.conftest import ( ADMIN_PASSWORD, SESSION_SECRET, USE_REAL_LLM, _wait_http, ) from e2e.mock_llm import TOKEN_RE, embed_text REPO = Path(__file__).resolve().parents[2] #: Own port (the phase-79 full-inventory rule: a module app binds its #: own port so a combined run never drives the session app). APP_PORT = int(os.environ.get("E2E_APP_PORT_RECALL", "8162")) APP_URL = f"http://127.0.0.1:{APP_PORT}" FIXTURES = REPO / "tests" / "fixtures" / "recallkb" MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E" # -------------------------------------------------------------------------- # The fixture KB (tracked files on disk — read back byte-for-byte) # -------------------------------------------------------------------------- #: source name → document paths. THE SMALL SOURCE is the live incident: #: one document whose body carries the product name only accented. SOURCE_DOCS: dict[str, list[str]] = { "ikea_instructions": ["besta-frame-white__AA-1272068-11-100.md"], "phase_docs": [ "140_pictorial_pdf_pages.md", "148_pdf_text_backfill.md", "152_seaweedfs_blob_ingest.md", "156_history_source_filter.md", ], "players_handbook": ["game_night_setup.md"], } IKEA_SOURCE = "ikea_instructions" IKEA_DOC = "besta-frame-white__AA-1272068-11-100.md" IKEA = f"{IKEA_SOURCE}/{IKEA_DOC}" #: Every fixture doc in ``(source, path)`` catalog order. ALL_DOCS = [f"{source}/{path}" for source in SOURCE_DOCS for path in SOURCE_DOCS[source]] #: The deterministic source inventory tail (LOCKED A2 order: name-sorted, #: singular ``doc`` at k=1) — the D2 assertion. INVENTORY = "Sources: ikea_instructions (1 doc), phase_docs (4 docs), players_handbook (1 doc)" # -------------------------------------------------------------------------- # The scripted questions # -------------------------------------------------------------------------- #: The live miss, byte-for-byte (the ``Å`` is the whole point — the file #: spells it ``BESTÅ``, the pre-phase name matcher read ``best``). ACCENTED_QUESTION = "How do I build the BESTÅ?" #: The phase-119 regression: the same product, spelled in ASCII. This #: question ALWAYS name-matched (the untouched path). ASCII_QUESTION = "How do I build a Besta?" #: The accented question + the mock's folder-map trigger — the prompt #: lens on the suggested tier (the folder lines are composed from the #: suggested documents only). FOLDER_QUESTION = ACCENTED_QUESTION + " repeat your folder map" #: The out-of-KB control (the phase-119 battery's deflection sanity). DEFLECT_QUESTION = "What is the capital of Mongolia?" def _expected_summary(content: str, source: str, path: str) -> str: """The mock lite model's byte-stable digest + the code pointer line (mirrors ``mock_llm.compose_answer``'s ``SUMMARY_MODE`` branch — the first 24 tokens of the document content — plus the summarizer's deterministic ``Source:`` line; no model output is ever trusted).""" digest = " ".join(TOKEN_RE.findall(content.lower())[:24]) return f"This document covers {digest}.\nSource: {source}/{path}" def _prefix_normalizer(value: str) -> str: """The PRE-PHASE-162 name normalizer (the measured before-state): lower + alnum scan with NO accent fold — the exact body ``_normalize_name`` had before task 01. Installed by monkeypatch in test 2 so the miss is MEASURED, not asserted from reasoning.""" return "".join(retriever._ALNUM_RE.findall(value.lower())) # -------------------------------------------------------------------------- # Fixtures # -------------------------------------------------------------------------- @pytest.fixture(scope="module") def seed_dirs() -> list[Path]: """The three local-directory sources (the registered directory's BASENAME is the source name, phase 38). The small source must be a directory of its own — that is what makes it a 1-doc source.""" dirs = [] for source, paths in SOURCE_DOCS.items(): directory = FIXTURES / source assert directory.is_dir(), f"{directory} is missing" for path in paths: assert (directory / path).is_file(), f"fixture doc {source}/{path} is missing" dirs.append(directory) return dirs @pytest.fixture(scope="module") def app_server(mock_llm: int, seed_dirs: list[Path]) -> Iterator[str]: """The real app under test — per-module app on its own port (the phase-79 inventory pattern), the mock LLM, the mock-calibrated gate, and the code defaults for everything the tiers depend on (suggested 5 / related 2 / name-hit bonus 0.005 / recency 0.0007, exactly the production ``.env``-free defaults).""" env = dict(os.environ) env.pop("DEBUGPY", None) env["BOR_ENVIRONMENT"] = "e2e" env["BOR_STATIC_DIR"] = str(REPO / "frontend") env["BOR_LLM_BASE_URL"] = ( "https://aipi.reeseapps.com/v1" if USE_REAL_LLM else f"http://127.0.0.1:{mock_llm}/v1" ) # Mock-calibrated gate (the conftest pattern): the accented and the # ASCII turns ground on the distractors' cosine (≈0.35); the # out-of-KB control stays under it. env["BOR_RELEVANCE_THRESHOLD"] = "0.30" env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15" env["BOR_LLM_RETRY_DELAY"] = "0" env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default) env.setdefault( "BOR_DATABASE_URL", "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese", ) env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD env["BOR_SESSION_SECRET"] = SESSION_SECRET # The repo's .env is the owner's — empty the git list so the registry # holds EXACTLY the three directories this suite registers, and pin # the leak-guarded corpus settings (the conftest pattern). env["BOR_GIT_SOURCES"] = "" env["BOR_DOCS_REPO"] = "" env["BOR_SUGGESTIONS"] = json.dumps( _Settings.model_fields["suggestions"].default ) env["BOR_INPUT_PLACEHOLDER"] = _Settings.model_fields["input_placeholder"].default env["BOR_FOOTER_TEXT"] = _Settings.model_fields["footer_text"].default proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"], cwd=REPO, env=env, ) try: _wait_http(f"{APP_URL}/api/health") yield APP_URL finally: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() @pytest.fixture(scope="module") def app_url(app_server: str) -> str: return app_server def _truncate_all() -> None: """Fresh registry + KB (the E2E isolation pattern): the tier margins are pinned against EXACTLY these six documents.""" with SessionLocal() as db: db.execute( text( "TRUNCATE chunks, documents, query_log, steering_notes, " "kb_overview, git_sources, folder_summaries" ) ) db.commit() def _wait_sync_done_http(client: httpx.Client, timeout_s: float = 180.0) -> dict[str, Any]: """Poll the cookie-authenticated status endpoint to a terminal state (the ``test_summary_seed_context`` pattern over plain httpx).""" deadline = time.monotonic() + timeout_s body: dict[str, Any] = {} while time.monotonic() < deadline: r = client.get("/api/sync/status") assert r.status_code == 200, r.text body = r.json() if body["state"] in ("success", "failed"): return body time.sleep(0.5) raise AssertionError(f"sync did not reach a terminal state: {body}") def _tiers(question: str) -> tuple[list[str], list[str], float, int]: """The measured tiers of one question over the synced KB, with the app's REAL hybrid retrieval and the mock's embeddings (the phase-119 measurement lens): ``(suggested, related, top cosine, fts hits)``.""" with SessionLocal() as db: chunks = retrieve(db, question, embed_text(question)) suggested = select_suggested(chunks) related = select_related( chunks, {d.id for d in suggested}, _Settings.model_fields["related_max_docs"].default, ) return ( [f"{d.source}/{d.path}" for d in suggested], [f"{d.source}/{d.path}" for d in related], max((c.cosine for c in chunks), default=0.0), sum(1 for c in chunks if c.fts_hit), ) def _assert_tiers( question: str, suggested: list[str], related: list[str] ) -> None: """Pin the engineered invariant for one scripted question, with a MEASURED message (a fixture-text regression fails at setup, not at the wire assertions).""" got_suggested, got_related, cosine, fts = _tiers(question) assert got_suggested == suggested, ( f"suggested tier drifted for {question!r}: {got_suggested} " f"(expected {suggested}; best cosine {cosine:.3f}, fts {fts})" ) assert got_related == related, ( f"related tier drifted for {question!r}: {got_related} " f"(expected {related})" ) @pytest.fixture(scope="module") def synced_kb(app_server: str, seed_dirs: list[Path]) -> None: """The story's precondition: the three-source KB synced under the deterministic mock. Registers the directories through the authenticated API, runs the REAL in-process sync, pins the stored content byte-identical to the tracked fixture files (including the accented ``BESTÅ`` surviving the round trip untouched), pins the fixture's own token controls (the mechanism the miss is reproduced with), and pins the engineered tiers for all three scripted questions.""" _truncate_all() with httpx.Client(base_url=app_server, timeout=30.0) as client: r = client.post("/api/login", json={"password": ADMIN_PASSWORD}) assert r.status_code == 204, r.text for directory in seed_dirs: r = client.post( "/api/git-sources", json={"kind": "local", "path": str(directory)} ) assert r.status_code == 201, r.text r = client.post("/api/sync") assert r.status_code == 202, r.text body = _wait_sync_done_http(client) assert body["state"] == "success", body assert body["detail"]["added"] == len(ALL_DOCS), body["detail"] assert body["detail"]["pruned"] == 0, body["detail"] with SessionLocal() as db: for source, paths in SOURCE_DOCS.items(): for path in paths: on_disk = (FIXTURES / source / path).read_text(encoding="utf-8") stored = db.scalar( select(Document).where( Document.source == source, Document.path == path ) ) assert stored is not None, f"fixture doc {source}/{path} missing" assert stored.content == on_disk, f"content drifted for {source}/{path}" assert stored.summary == _expected_summary(on_disk, source, path), ( f"summary is not the mock digest for {source}/{path}" ) schunks = [c for c in stored.chunks if c.is_summary] assert len(schunks) == 1 and schunks[0].embedding is not None, ( f"expected one embedded is_summary chunk for {source}/{path}" ) # ---- The fixture's token controls (the reproduction mechanism) ---- ikea_text = (FIXTURES / IKEA_SOURCE / IKEA_DOC).read_text(encoding="utf-8") # (a) the name exists ONLY accented — no ASCII ``besta`` in the body, # so nothing but the phase-162 fold can bridge question ↔ doc; assert "BESTÅ" in ikea_text assert "besta" not in ikea_text.lower().replace("bestå", ""), ikea_text # (b) no ``build`` either: the accented question's only OTHER FTS # token — the small source must win on the NAME signal, never on # a content-word accident; assert "build" not in ikea_text.lower() # (c) the accented name sits PAST the 24-token digest window, so the # mock's ASCII-only digest never leaks a ``best`` lexeme the # pre-phase FTS could have hit (a real ``lite`` summary of the # live PDF never wrote the name in ASCII either). digest = TOKEN_RE.findall(ikea_text.lower())[:24] assert "best" not in digest and "besta" not in digest, digest # ---- The engineered tiers (measured, stable across re-imports) ---- _assert_tiers(ACCENTED_QUESTION, SUGGESTED_ACCENTED, RELATED_ACCENTED) _assert_tiers(ASCII_QUESTION, SUGGESTED_ASCII, RELATED_ASCII) _assert_tiers(FOLDER_QUESTION, SUGGESTED_FOLDER, RELATED_FOLDER) @pytest.fixture(autouse=True) def _clean(db_ready: None) -> Iterator[None]: """Per-test query_log isolation (the KB itself is module-scoped — the scripted turns never change it).""" with SessionLocal() as db: db.execute(text("TRUNCATE query_log")) db.commit() yield with SessionLocal() as db: db.execute(text("TRUNCATE query_log")) db.commit() # -------------------------------------------------------------------------- # The pinned tiers (MEASURED with the app's real retrieval over the mock # embeddings — see the module docstring's before/after record) # -------------------------------------------------------------------------- #: POST-FIX, accented question: the small source LEADS the tier (its #: name hit leads the lexical list ⇒ ``fuse`` rank 1 = 1/61, plus the #: per-document name-hit bonus 0.005), the five ASCII distractors #: follow in measured retrieval order, one of them falls to rank 6. SUGGESTED_ACCENTED = [ IKEA, "phase_docs/140_pictorial_pdf_pages.md", "phase_docs/148_pdf_text_backfill.md", "phase_docs/152_seaweedfs_blob_ingest.md", "players_handbook/game_night_setup.md", ] RELATED_ACCENTED = ["phase_docs/156_history_source_filter.md"] #: POST-FIX, ASCII question (the untouched phase-119 path): same doc, #: same #1 slot — the only difference is the measured order of two #: distractors that trade places on the extra ``besta`` FTS term. SUGGESTED_ASCII = [ IKEA, "phase_docs/140_pictorial_pdf_pages.md", "phase_docs/148_pdf_text_backfill.md", "players_handbook/game_night_setup.md", "phase_docs/152_seaweedfs_blob_ingest.md", ] RELATED_ASCII = ["phase_docs/156_history_source_filter.md"] #: The folder-map turn (the trigger's own tokens dilute the cosine to #: 0.296 — lexical support carries the gate) keeps the same tier. SUGGESTED_FOLDER = list(SUGGESTED_ACCENTED) RELATED_FOLDER = list(RELATED_ACCENTED) #: PRE-FIX (the measured miss, test 2): the folded token never reaches #: ``besta``, so the small source has ONE fused term (vector) against #: every distractor's TWO (vector + FTS ``build``) and the five #: distractors fill the tier — the IKEA doc is rank 6, i.e. the rank-6+ #: ``related`` remainder. Exactly the live 2026-10-07 under-rank. PRE_SUGGESTED_ACCENTED = [ "phase_docs/140_pictorial_pdf_pages.md", "phase_docs/148_pdf_text_backfill.md", "phase_docs/152_seaweedfs_blob_ingest.md", "players_handbook/game_night_setup.md", "phase_docs/156_history_source_filter.md", ] PRE_RELATED_ACCENTED = [IKEA] # -------------------------------------------------------------------------- # Page helpers (the phase-119 house patterns) # -------------------------------------------------------------------------- #: Captures the raw SSE ``data:`` payloads of the /api/chat stream (a #: response clone read in the background) — wire-level assertions on the #: ``done`` frame, independent of the rendering. SSE_HOOK = """ () => { if (window.__sseInstalled) return; window.__sseInstalled = true; window.__sseFrames = []; const origFetch = window.fetch; window.fetch = async function (...args) { const res = await origFetch.apply(this, args); try { const url = typeof args[0] === 'string' ? args[0] : args[0].url; if (url.includes('/api/chat')) { res.clone().text().then((bodyText) => { for (const block of bodyText.split('\\n\\n')) { const line = block.trim(); if (line.startsWith('data: ')) { window.__sseFrames.push(line.slice(6)); } } }); } } catch (e) { /* non-clonable responses: ignored */ } return res; }; } """ def _install_page_hooks(page: Page) -> None: page.evaluate(SSE_HOOK) def _frames(page: Page) -> list[dict[str, Any]]: """The SSE frames captured since the last submit, once the hook's background read settles.""" deadline = time.monotonic() + 30.0 while True: raw = page.evaluate("() => window.__sseFrames || []") parsed = [json.loads(line) for line in raw if line] if any(f.get("type") == "done" for f in parsed): return parsed if time.monotonic() > deadline: raise AssertionError( f"SSE hook captured no `done` frame (frames so far: " f"{len(parsed)}) — hook install failed?" ) time.sleep(0.05) def _done(page: Page) -> dict[str, Any]: return next(f for f in _frames(page) if f.get("type") == "done") def _tool_frames(frames: list[dict[str, Any]]) -> list[dict[str, Any]]: return [f for f in frames if f.get("type") == "tool"] def _submit(page: Page, question: str) -> None: page.evaluate("window.__sseFrames = []") page.fill("#message-input", question) page.click("#send-btn") expect(page.locator(".msg.user .bubble").last).to_contain_text(question) def _wait_settled(page: Page) -> None: expect(page.locator(".msg.brain .bubble").last).not_to_have_text("", timeout=30_000) expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000) expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000) def _last_brain(page: Page) -> Locator: return page.locator(".msg.brain").last def _last_query_log() -> QueryLog: with SessionLocal() as db: rows = db.scalars(select(QueryLog)).all() assert len(rows) == 1, f"expected exactly one query_log row, got {len(rows)}" return rows[0] def _recorded_tiers(row: QueryLog) -> tuple[list[str], list[str]]: """The durable record's tiers (LOCKED A3: the record is the suggested tier followed by the rank-6+ remainder — the wire's only faithful witness of the suggested set, since ``done.sources`` is the READ-DOCS citation surface and ``done.suggestions`` is the deflection chips).""" recorded = [s for s in row.sources.split(", ") if s] return recorded[:5], recorded[5:] # -------------------------------------------------------------------------- # 1. (task 04 case 1) the accented question puts the 1-doc source in the # suggested top-5 — and it is no longer the rank-6 miss # -------------------------------------------------------------------------- def test_accented_question_suggests_the_small_source( page: Page, app_url: str, synced_kb: None, db_ready: None ) -> None: """``How do I build the BESTÅ?`` grounds, and the suggested tier of the turn carries ``ikea_instructions/besta-frame-white__AA-1272068- 11-100.md`` — SUGGESTED #1, measured. The suggested set is pinned on the wire through the durable record (LOCKED A3: ``sources`` = the suggested 5, then the rank-6+ remainder): the IKEA doc sits inside the first five entries, and the ``done`` frame's ``related`` list — the rank-6+ tier, the slot it occupied as the live miss — is EMPTY of it. A zero-read grounded turn chips nothing (phase 119 LOCKED A1); the answer renders grounded.""" page.set_default_timeout(30_000) login(page, app_url, next="/") _install_page_hooks(page) _submit(page, ACCENTED_QUESTION) _wait_settled(page) frames = _frames(page) assert _tool_frames(frames) == [], _tool_frames(frames) done = _done(page) assert done["deflected"] is False, done assert done["sources"] == [], done["sources"] # The miss was the IKEA doc landing in the rank-6+ remainder. It is # not there anymore — the tier it belongs to is the suggested one. related = [(s["source"], s["path"]) for s in done["related"]] assert (IKEA_SOURCE, IKEA_DOC) not in related, done["related"] # The ``suggestions`` field is the DEFLECTION contract (title chips # under a weak turn) — a grounded turn carries none, so it can never # be the tier witness (the durable record is, and is asserted below). assert done["suggestions"] == [], done["suggestions"] bubble = _last_brain(page).locator(".bubble") expect(bubble).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000) expect(page.locator(".msg.brain .source-chip")).to_have_count(0) # The de-emphasized rank-6+ row renders — and the doc it used to # hold (the miss) is a distractor now. row = _last_query_log() related_links = page.locator(".msg.brain .related-docs .related-doc") expect(related_links).to_have_count(len(RELATED_ACCENTED)) for i, path in enumerate(RELATED_ACCENTED): expect(related_links.nth(i)).to_contain_text(path) expect(related_links.filter(has_text=IKEA)).to_have_count(0) assert row.question == ACCENTED_QUESTION assert row.deflected is False suggested, related_record = _recorded_tiers(row) assert suggested == SUGGESTED_ACCENTED, row.sources assert related_record == RELATED_ACCENTED, row.sources assert suggested[0] == IKEA, row.sources # -------------------------------------------------------------------------- # 2. (task 04 case 1, the hypothesis) the under-rank is MEASURED in both # directions: pre-fix the doc loses the tier, post-fix it wins it # -------------------------------------------------------------------------- def test_under_rank_is_measured_not_assumed( monkeypatch: pytest.MonkeyPatch, synced_kb: None, db_ready: None ) -> None: """The phase-162 hypothesis, measured rather than asserted: with the PRE-phase normalizer re-installed (``lower`` + alnum scan, no accent fold) the accented question degrades to the token ``best`` — no name hit — and the SAME retrieval over the SAME KB drops ``ikea_instructions/besta-frame-white__AA-1272068-11-100.md`` out of the suggested top-5 into the rank-6+ remainder (the live miss, reproduced). With the shipped normalizer the folded ``besta`` token name-matches the filename and the doc is suggested #1. Re-running the measurement is the guard: if a future fixture or retriever change ever makes the pre-phase path win, this test fails and the recorded evidence is re-examined instead of trusted.""" # The token side of the miss, measured. assert retriever.name_hit_tokens(ACCENTED_QUESTION) == ["build", "besta"] monkeypatch.setattr(retriever, "_normalize_name", _prefix_normalizer) assert retriever.name_hit_tokens(ACCENTED_QUESTION) == ["build", "best"] pre_suggested, pre_related, _cosine, _fts = _tiers(ACCENTED_QUESTION) assert IKEA not in pre_suggested, ( f"the pre-phase path WINS the tier — the fixture no longer " f"reproduces the miss: {pre_suggested}" ) assert pre_suggested == PRE_SUGGESTED_ACCENTED, pre_suggested assert IKEA in pre_related, ( f"the miss should surface as the rank-6+ remainder: {pre_related}" ) assert pre_related == PRE_RELATED_ACCENTED, pre_related monkeypatch.undo() # And the same measurement with the shipped fold: the doc is #1. post_suggested, post_related, _cosine, _fts = _tiers(ACCENTED_QUESTION) assert post_suggested == SUGGESTED_ACCENTED assert post_related == RELATED_ACCENTED # -------------------------------------------------------------------------- # 3. (task 04 case 2) the ASCII question keeps its phase-119 name hit # -------------------------------------------------------------------------- def test_ascii_question_still_name_matches( page: Page, app_url: str, synced_kb: None, db_ready: None ) -> None: """``How do I build a Besta?`` — the question the phase-119 rule already answered — still grounds on the same document (the fold must be behavior-preserving for the ASCII path: ``besta`` normalizes to ``besta`` either way). Same tier position, same durable record, grounded, nothing chipped on a zero-read turn.""" page.set_default_timeout(30_000) login(page, app_url, next="/") _install_page_hooks(page) _submit(page, ASCII_QUESTION) _wait_settled(page) frames = _frames(page) assert _tool_frames(frames) == [], _tool_frames(frames) done = _done(page) assert done["deflected"] is False, done assert done["sources"] == [], done["sources"] expect(_last_brain(page).locator(".bubble")).to_contain_text( MOCK_ANSWER_MARKER, timeout=30_000 ) expect(page.locator(".msg.brain .source-chip")).to_have_count(0) row = _last_query_log() assert row.question == ASCII_QUESTION assert row.deflected is False suggested, related_record = _recorded_tiers(row) assert suggested == SUGGESTED_ASCII, row.sources assert suggested[0] == IKEA, row.sources assert related_record == RELATED_ASCII, row.sources # -------------------------------------------------------------------------- # 4. (task 04 case 3) the deterministic source inventory — the D2 tail # -------------------------------------------------------------------------- def test_overview_tail_names_every_source_with_counts( synced_kb: None, db_ready: None ) -> None: """The stored ``kb_overview`` row ENDS with the mechanical ``Sources: …`` line — one entry per source, exact counts, singular ``doc`` at k=1 — so the 1-document sources are visible to ```` whatever the lossy ``lite`` digest said (the 2026-10-07 live finding: a 14-source index outlined as 3 groups). The tail is ONE line, appears exactly once, and sits after the model's outline (the model input never sees it).""" with SessionLocal() as db: rows = db.scalars(select(KbOverview)).all() assert len(rows) == 1, rows content = rows[0].content lines = content.splitlines() assert lines[-1] == INVENTORY, content assert content.count("Sources: ") == 1, content # The ``lite`` outline still leads (the tail is appended, never a # replacement). assert len(lines) >= 2, content assert lines[0].startswith("Knowledge base outline:"), content for source in SOURCE_DOCS: assert source in lines[-1], content # -------------------------------------------------------------------------- # 5. (task 04 case 4) the out-of-KB control — the gate verdict unchanged # -------------------------------------------------------------------------- def test_out_of_kb_question_still_deflects( page: Page, app_url: str, synced_kb: None, db_ready: None ) -> None: """``What is the capital of Mongolia?`` matches nothing in this KB: the A8 verdict stays DEFLECTED (``deflected: true``, ``fts_hits == 0``, weak cosine under the mock-calibrated threshold), the answer carries no citation chips, and the only chips are the ``Maybe try`` suggestions. The accent fold must never turn a real miss into a false grounding (the phase-119 battery's deflection sanity, localised).""" page.set_default_timeout(30_000) login(page, app_url, next="/") _install_page_hooks(page) _submit(page, DEFLECT_QUESTION) _wait_settled(page) done = _done(page) assert done["deflected"] is True, done assert done["sources"] == [], done["sources"] assert 2 <= len(done["suggestions"]) <= 3, done["suggestions"] bubble = _last_brain(page) expect(bubble).to_have_class(re.compile(r"is-deflected")) expect(bubble.locator(".source-chip")).to_have_count(0) expect( bubble.locator(".maybe-try .suggestion-chip") ).to_have_count(len(done["suggestions"])) row = _last_query_log() assert row.question == DEFLECT_QUESTION assert row.deflected is True assert row.fts_hits == 0, row.fts_hits assert row.top_score < 0.30, row.top_score # -------------------------------------------------------------------------- # 6. (task 04 case 3 / D3) the stored accented body stays grep-matchable # -------------------------------------------------------------------------- def test_grep_folds_the_accented_body(synced_kb: None, db_ready: None) -> None: """The content layer of the same miss: the stored body spells the product ``BESTÅ`` only, and the agent's ``grep`` tool folds BOTH the pattern and the line — so the ASCII pattern ``besta`` finds the accented line (the pre-fold tool returned nothing, which is why the live turn gave up on a document it had already opened), and the accented pattern still matches itself.""" with SessionLocal() as db: stored = db.scalar( select(Document).where( Document.source == IKEA_SOURCE, Document.path == IKEA_DOC ) ) assert stored is not None assert "BESTÅ" in stored.content ascii_hits = grep_document(stored.content, "besta") assert any("BESTÅ" in line for _no, line in ascii_hits), ascii_hits accented_hits = grep_document(stored.content, "BESTÅ") assert [line for _no, line in accented_hits] == [line for _no, line in ascii_hits] # A pattern that is not the name still finds nothing (the fold is an # accent bridge, not a wildcard). assert grep_document(stored.content, "bestaa") == [] # -------------------------------------------------------------------------- # 7. the prompt-level witness — the 1-doc source's folder line rides the # HIGH prompt with the suggested tier # -------------------------------------------------------------------------- def test_suggested_folder_lines_carry_the_small_source( page: Page, app_url: str, synced_kb: None, db_ready: None ) -> None: """The accented question + the mock's ``repeat your folder map`` trigger ⇒ the answer ends with the suggested-folder lines VERBATIM. The 1-doc source contributes ``ikea_instructions/:`` — a header with no sibling entries, because its only document IS the suggested doc — the visible proof the small source reached the prompt with the tier.""" page.set_default_timeout(30_000) login(page, app_url, next="/") _install_page_hooks(page) _submit(page, FOLDER_QUESTION) _wait_settled(page) frames = _frames(page) assert _tool_frames(frames) == [], _tool_frames(frames) done = _done(page) assert done["deflected"] is False, done bubble = _last_brain(page).locator(".bubble") expect(bubble).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000) expect(bubble).to_contain_text("ikea_instructions/:") row = _last_query_log() assert row.deflected is False suggested, _related = _recorded_tiers(row) assert suggested == SUGGESTED_FOLDER, row.sources