Enhance chat service and configuration management

- Introduced URL extraction functionality to deduplicate and trim URLs from user queries, improving evidence handling for web citations.
- Updated the build_agent function to incorporate knowledge base agent instructions into the system prompt, ensuring proper cache management.
- Enhanced the ensure_kb_agents_md function for atomic file writing, preventing race conditions during concurrent access.
- Improved local vector database verification to ensure completeness before proceeding with operations, reducing potential errors.
- Added comprehensive tests for new functionalities, including URL extraction and knowledge base instructions handling, ensuring robustness and reliability.

app/chat_service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+import re
 import shlex
 import time
 from dataclasses import dataclass
@@ -44,7 +45,7 @@ from .kb_manifest import (
     source_match_key,
     source_match_payload,
 )
-from .prompts import build_system_prompt
+from .prompts import build_system_prompt, ensure_kb_agents_instructions
 from .provider_events import (
     GoogleSearchActivity,
     extract_anthropic_source_matches,
@@ -813,7 +814,11 @@ def build_agent(
     model_name: str,
     enabled_tools: tuple[str, ...] = (),
     include_thoughts: bool = False,
+    kb_agents_instructions: str | None = None,
 ):
+    # kb_agents_instructions is part of the cache key on purpose: an agent
+    # built before data/kb/AGENTS.md existed must not pin its degraded
+    # system prompt for the process lifetime.
     model = build_chat_model(model_name, include_thoughts=include_thoughts)
     tools: list[Any] = [
         retrieve_tutor_context,
@@ -867,7 +872,9 @@ def build_agent(
     return create_agent(
         model=model,
         tools=tools,
-        system_prompt=build_system_prompt(model_name, enabled_tools),
+        system_prompt=build_system_prompt(
+            model_name, enabled_tools, kb_agents_instructions
+        ),
         context_schema=AppContext,
         checkpointer=CHECKPOINTER,
         middleware=middleware,
@@ -899,6 +906,41 @@ def effective_tool_names(
     return tuple(names)
 
 
+_QUERY_URL_PATTERN = re.compile(r"https?://[^\s<>()\[\]\"']+")
+
+
+def extract_query_urls(text: str) -> list[str]:
+    """Pull http(s) URLs out of a user query, in order, deduped."""
+    urls: list[str] = []
+    for raw_url in _QUERY_URL_PATTERN.findall(text):
+        url = raw_url.rstrip(".,;:!?")
+        if url and url not in urls:
+            urls.append(url)
+    return urls
+
+
+def url_context_evidence(query: str) -> list[SourceMatch]:
+    """Evidence for URLs the user pasted when `url_context` is enabled.
+
+    langchain-google-genai (4.2.4) never surfaces Gemini's
+    `url_context_metadata`, so fetches are invisible to the stream. Treating
+    pasted URLs as web evidence at least lets the model's citation of a
+    fetched page resolve to a source card instead of being dropped.
+    """
+    return [
+        SourceMatch(
+            doc_id=f"url_context::{url}",
+            title=url,
+            url=url,
+            source_key="url_context",
+            source_label="Web page",
+            score=1.0,
+            group="web",
+        )
+        for url in extract_query_urls(query)
+    ]
+
+
 def agent_run_config(
     request: ChatRequest,
     active_thread_id: str,
@@ -958,12 +1000,16 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
     # decide whether consecutive deltas need a paragraph break between them.
     reasoning_deltas_are_blocks = is_google_genai_model(request.model_name)
     google_search = GoogleSearchActivity(message_id, web_evidence)
+    effective_tools = effective_tool_names(request.model_name, request.enabled_tools)
+    if "url_context" in effective_tools:
+        _record_evidence(web_evidence, url_context_evidence(request.query))
 
     logger.info("Running query: %s", request.query)
     agent = build_agent(
         request.model_name,
         enabled_tools=tuple(request.enabled_tools),
         include_thoughts=include_reasoning,
+        kb_agents_instructions=ensure_kb_agents_instructions(),
     )
     _evict_idle_threads()
     active_thread_id, fork_checkpoint_id = sync_thread_with_history(
@@ -1200,6 +1246,11 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
             retrieval_evidence=retrieval_evidence,
             shell_evidence=shell_evidence,
             web_evidence=web_evidence,
+            # url_context is the only tool whose fetches are invisible to the
+            # stream (the other web tools report results into web_evidence),
+            # so only then is a cited-but-unmatched URL plausibly a fetched
+            # page rather than a memory link: keep it as a low-trust Web chip.
+            keep_unresolved_sources="url_context" in effective_tools,
         )
     )
 

app/config.py

@@ -1,6 +1,7 @@
 import logging
 import os
 from pathlib import Path
+from threading import Lock
 
 from dotenv import load_dotenv
 
@@ -44,6 +45,7 @@ VECTOR_DB_DIR = "data/chroma-db-all_sources"
 VECTOR_COLLECTION_NAME = "chroma-db-all_sources"
 DOCUMENT_DICT_PATH = f"{VECTOR_DB_DIR}/document_dict_all_sources.pkl"
 BM25_INDEX_PATH = f"{VECTOR_DB_DIR}/bm25_index_all_sources.pkl"
+CHROMA_SQLITE_PATH = f"{VECTOR_DB_DIR}/chroma.sqlite3"
 KB_DIR = "data/kb"
 KB_MANIFEST_PATH = f"{KB_DIR}/generated/corpus_manifest.jsonl"
 KB_INDEX_PATH = f"{KB_DIR}/wiki/index.md"
@@ -58,42 +60,80 @@ AVAILABLE_MODELS: tuple[dict[str, str], ...] = (
 )
 
 
+_BUNDLE_LOCK = Lock()
+_BUNDLE_READY = False
+_KB_AGENTS_WRITE_LOCK = Lock()
+
+
+def _bundle_complete() -> bool:
+    required = (
+        VECTOR_DB_DIR,
+        CHROMA_SQLITE_PATH,
+        DOCUMENT_DICT_PATH,
+        BM25_INDEX_PATH,
+        KB_MANIFEST_PATH,
+        KB_INDEX_PATH,
+    )
+    return all(os.path.exists(path) for path in required)
+
+
 def ensure_kb_agents_md() -> None:
-    """Overwrite data/kb/AGENTS.md from the in-git template on every startup."""
+    """Write data/kb/AGENTS.md from the in-git template.
+
+    Atomic (temp file + os.replace, so a concurrent prompt build never reads
+    a truncated file) and skipped when the content already matches.
+    """
     template_path = Path(KB_AGENTS_TEMPLATE_PATH)
     if not template_path.exists():
         return
+    content = template_path.read_text(encoding="utf-8")
     target = Path(KB_AGENTS_PATH)
-    target.parent.mkdir(parents=True, exist_ok=True)
-    target.write_text(template_path.read_text(encoding="utf-8"), encoding="utf-8")
+    with _KB_AGENTS_WRITE_LOCK:
+        try:
+            if target.read_text(encoding="utf-8") == content:
+                return
+        except OSError:
+            pass
+        target.parent.mkdir(parents=True, exist_ok=True)
+        tmp_path = target.with_name(target.name + ".tmp")
+        tmp_path.write_text(content, encoding="utf-8")
+        os.replace(tmp_path, target)
 
 
 def ensure_local_vector_db() -> None:
-    needs_download = not (
-        os.path.exists(VECTOR_DB_DIR)
-        and os.path.exists(DOCUMENT_DICT_PATH)
-        and os.path.exists(KB_MANIFEST_PATH)
-        and os.path.exists(KB_INDEX_PATH)
-    )
-    if needs_download:
-        logger.warning(
-            "Vector database does not exist locally, downloading from Hugging Face"
-        )
-        from huggingface_hub import snapshot_download
-
-        # Mute httpx's per-file flood during the cold-start download only.
-        httpx_logger = logging.getLogger("httpx")
-        previous_level = httpx_logger.level
-        httpx_logger.setLevel(logging.WARNING)
-        try:
-            snapshot_download(
-                repo_id="towardsai-tutors/ai-tutor-vector-db",
-                local_dir="data",
-                repo_type="dataset",
+    """Make sure the vector-db/KB bundle and AGENTS.md exist locally.
+
+    A flag check once the bundle has been verified, so per-tool-call use is
+    effectively free; the verification and download run single-flight under
+    a lock so parallel first-turn tool calls cannot race two downloads.
+    """
+    global _BUNDLE_READY
+    if _BUNDLE_READY:
+        return
+    with _BUNDLE_LOCK:
+        if _BUNDLE_READY:
+            return
+        if not _bundle_complete():
+            logger.warning(
+                "Vector database missing or incomplete locally, downloading "
+                "from Hugging Face"
             )
-        finally:
-            httpx_logger.setLevel(previous_level)
-    ensure_kb_agents_md()
+            from huggingface_hub import snapshot_download
+
+            # Mute httpx's per-file flood during the cold-start download only.
+            httpx_logger = logging.getLogger("httpx")
+            previous_level = httpx_logger.level
+            httpx_logger.setLevel(logging.WARNING)
+            try:
+                snapshot_download(
+                    repo_id="towardsai-tutors/ai-tutor-vector-db",
+                    local_dir="data",
+                    repo_type="dataset",
+                )
+            finally:
+                httpx_logger.setLevel(previous_level)
+        ensure_kb_agents_md()
+        _BUNDLE_READY = _bundle_complete() and os.path.exists(KB_AGENTS_PATH)
 
 
 __all__ = [

app/prompts.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import logging
 import os
 from pathlib import Path
 
+logger = logging.getLogger(__name__)
+
 
 BASE_PROMPT_HEADER = """You are an AI teacher for applied AI, LLM, RAG, and Python topics.
 
@@ -183,7 +186,36 @@ def load_kb_agents_instructions(path: Path | None = None) -> str:
         return ""
 
 
-def build_system_prompt(model_name: str, enabled_tools: tuple[str, ...]) -> str:
+def ensure_kb_agents_instructions() -> str:
+    """Load the Local KB Instructions, materializing the file if missing.
+
+    `data/kb/AGENTS.md` is generated from an in-git template, so it can be
+    written without the KB bundle download. Returns "" (with a loud log)
+    only when even the template is unavailable.
+    """
+    instructions = load_kb_agents_instructions()
+    if instructions:
+        return instructions
+    try:
+        from .config import ensure_kb_agents_md
+
+        ensure_kb_agents_md()
+    except OSError as exc:
+        logger.warning("Could not materialize KB agent instructions: %s", exc)
+    instructions = load_kb_agents_instructions()
+    if not instructions:
+        logger.warning(
+            "KB agent instructions unavailable; the system prompt will omit "
+            "the Local KB section for this request."
+        )
+    return instructions
+
+
+def build_system_prompt(
+    model_name: str,
+    enabled_tools: tuple[str, ...],
+    kb_agents_instructions: str | None = None,
+) -> str:
     provider = _provider_key(model_name)
     enabled = set(enabled_tools)
     tool_lines = [RETRIEVAL_TOOL_LINE, *KB_TOOL_LINES]
@@ -215,7 +247,8 @@ def build_system_prompt(model_name: str, enabled_tools: tuple[str, ...]) -> str:
             "course material. Combine tools when it helps (e.g. retrieve corpus\n"
             "context, then search the web for the latest update)."
         )
-    kb_agents_instructions = load_kb_agents_instructions()
+    if kb_agents_instructions is None:
+        kb_agents_instructions = load_kb_agents_instructions()
     if kb_agents_instructions:
         parts.append(
             "## Local KB Instructions\n\n"

tests/test_api.py

@@ -325,11 +325,15 @@ def live_api_server() -> Iterator[int]:
         thread.join(timeout=10)
 
 
-def live_chat_payload(messages: list[dict], thread_id: str = "") -> dict:
+def live_chat_payload(
+    messages: list[dict],
+    thread_id: str = "",
+    enabled_tools: list[str] | None = None,
+) -> dict:
     return {
         "messages": messages,
         "sourceKeys": ["peft", "transformers"],
-        "enabledTools": [],
+        "enabledTools": enabled_tools or [],
         "model": os.getenv("LIVE_API_E2E_MODEL", "google-genai:gemini-3.5-flash"),
         "includeReasoning": False,
         "threadId": thread_id,
@@ -535,6 +539,40 @@ def test_live_api_edit_forks_thread_from_checkpoint() -> None:
         )
 
 
+@LIVE_API_E2E
+def test_live_api_pasted_url_gets_source_card_with_url_context() -> None:
+    """The paste-a-URL flow must not lose its source card.
+
+    Gemini's url_context fetches never reach the stream (the library drops
+    url_context_metadata), so the pasted URL is synthesized as web evidence;
+    whether the model cites it inline or cites nothing, a source card for
+    the page must surface.
+    """
+    require_live_api_e2e_prereqs()
+
+    page_url = "https://huggingface.co/docs/peft/index"
+    prompt = (
+        f"Read {page_url} and answer in two sentences: what is PEFT? "
+        "Cite that page inline in your answer."
+    )
+
+    with live_api_server() as port:
+        parts = post_live_api_chat(
+            port,
+            live_chat_payload(
+                [live_user_message(prompt)],
+                enabled_tools=["url_context"],
+            ),
+        )
+
+    assert "finish" in live_part_types(parts)
+    assert live_answer_text(parts)
+    sources = live_sources(parts)
+    assert any(source["url"] == page_url for source in sources), (
+        f"pasted URL missing from source cards: {sources}"
+    )
+
+
 @LIVE_API_E2E
 def test_live_api_stream_obeys_retrieval_only_instruction() -> None:
     require_live_api_e2e_prereqs()

tests/test_chat_service.py

@@ -22,6 +22,7 @@ from app.chat_service import (
     build_agent,
     checkpoint_messages_to_history,
     effective_tool_names,
+    extract_query_urls,
     resolve_answer_citations,
     retrieve_tutor_context,
     sync_thread_with_history,
@@ -106,6 +107,20 @@ class FakeStreamingAgent(FakeAgent):
         }
 
 
+class FakeAnswerAgent(FakeAgent):
+    """Agent that streams nothing and answers with one final AI message."""
+
+    def __init__(self, answer: str):
+        super().__init__([])
+        self.answer = answer
+
+    async def astream(self, *args, **kwargs):
+        yield {
+            "type": "updates",
+            "data": {"model": {"messages": [AIMessage(content=self.answer)]}},
+        }
+
+
 class ChatServiceTestCase(unittest.TestCase):
     def test_effective_tool_names_follow_provider(self) -> None:
         self.assertEqual(
@@ -710,6 +725,175 @@ class ChatServiceTestCase(unittest.TestCase):
             ),
         )
 
+    def test_build_agent_cache_busts_when_kb_instructions_appear(self) -> None:
+        build_agent.cache_clear()
+        created_agents = []
+
+        def fake_create_agent(**kwargs):
+            agent = types.SimpleNamespace(kwargs=kwargs)
+            created_agents.append(agent)
+            return agent
+
+        try:
+            with (
+                patch(
+                    "app.chat_service.build_chat_model",
+                    return_value=types.SimpleNamespace(_llm_type="fake-chat-model"),
+                ),
+                patch("app.chat_service.create_agent", side_effect=fake_create_agent),
+            ):
+                degraded = build_agent(
+                    "google-genai:gemini-3.5-flash",
+                    enabled_tools=(),
+                    include_thoughts=False,
+                    kb_agents_instructions="",
+                )
+                healthy = build_agent(
+                    "google-genai:gemini-3.5-flash",
+                    enabled_tools=(),
+                    include_thoughts=False,
+                    kb_agents_instructions="# KB rules marker",
+                )
+                healthy_again = build_agent(
+                    "google-genai:gemini-3.5-flash",
+                    enabled_tools=(),
+                    include_thoughts=False,
+                    kb_agents_instructions="# KB rules marker",
+                )
+        finally:
+            build_agent.cache_clear()
+
+        # A degraded (no KB instructions) build must not pin the cache entry.
+        self.assertIsNot(degraded, healthy)
+        self.assertIs(healthy, healthy_again)
+        self.assertNotIn(
+            "## Local KB Instructions", created_agents[0].kwargs["system_prompt"]
+        )
+        self.assertIn("# KB rules marker", created_agents[1].kwargs["system_prompt"])
+
+    def test_stream_chat_passes_ensured_kb_instructions_to_build_agent(self) -> None:
+        agent = FakeAnswerAgent("Answer.")
+        request = ChatRequest(
+            query="hello",
+            source_keys=("peft",),
+            model_name="google-genai:gemini-3.5-flash",
+            include_reasoning=False,
+            enabled_tools=(),
+        )
+        self.addCleanup(_drop_thread_record, "thread_kbwire")
+
+        async def collect_events():
+            return [event async for event in stream_chat(request)]
+
+        with (
+            patch(
+                "app.chat_service.build_agent", return_value=agent
+            ) as build_agent_mock,
+            patch(
+                "app.chat_service.ensure_kb_agents_instructions",
+                return_value="# Ensured rules",
+            ),
+            patch("app.chat_service.new_thread_id", return_value="thread_kbwire"),
+        ):
+            asyncio.run(collect_events())
+
+        self.assertEqual(
+            build_agent_mock.call_args.kwargs["kb_agents_instructions"],
+            "# Ensured rules",
+        )
+
+    def test_extract_query_urls_dedupes_and_trims_punctuation(self) -> None:
+        self.assertEqual(
+            extract_query_urls(
+                "Read https://example.com/post, then https://example.com/post "
+                "and (https://other.example/page)."
+            ),
+            ["https://example.com/post", "https://other.example/page"],
+        )
+        self.assertEqual(extract_query_urls("no links here"), [])
+
+    def run_answer_stream(
+        self,
+        answer: str,
+        *,
+        query: str,
+        enabled_tools: tuple[str, ...],
+        thread_id: str,
+    ) -> list:
+        agent = FakeAnswerAgent(answer)
+        request = ChatRequest(
+            query=query,
+            source_keys=("peft",),
+            model_name="google-genai:gemini-3.5-flash",
+            include_reasoning=False,
+            enabled_tools=enabled_tools,
+        )
+        self.addCleanup(_drop_thread_record, thread_id)
+
+        async def collect_events():
+            return [event async for event in stream_chat(request)]
+
+        with (
+            patch("app.chat_service.build_agent", return_value=agent),
+            patch("app.chat_service.new_thread_id", return_value=thread_id),
+            patch("app.chat_service.resolve_manifest_reference", return_value=None),
+        ):
+            events = asyncio.run(collect_events())
+        return [event for event in events if event.type == "source_match"]
+
+    def test_cited_pasted_url_resolves_via_url_context_evidence(self) -> None:
+        sources = self.run_answer_stream(
+            "Summary. See [the post](https://example.com/post).",
+            query="Read https://example.com/post and summarize it",
+            enabled_tools=("url_context",),
+            thread_id="thread_urlctx",
+        )
+
+        self.assertEqual(len(sources), 1)
+        self.assertEqual(sources[0].data["source_key"], "url_context")
+        self.assertEqual(sources[0].data["url"], "https://example.com/post")
+        self.assertEqual(sources[0].data["group"], "web")
+
+    def test_unmatched_web_citation_kept_only_for_url_context(self) -> None:
+        # Only url_context fetches invisibly; the other web tools report
+        # their results into evidence, so an unmatched citation there still
+        # means a memory link and stays gated.
+        answer = "From [some site](https://unsourced.example/page)."
+
+        kept = self.run_answer_stream(
+            answer,
+            query="What does that site say?",
+            enabled_tools=("url_context",),
+            thread_id="thread_webkeep",
+        )
+        self.assertEqual(len(kept), 1)
+        self.assertEqual(kept[0].data["source_key"], "web")
+        self.assertEqual(kept[0].data["group"], "web")
+
+        for enabled_tools, thread_id in (
+            (("web_search",), "thread_webdrop_search"),
+            ((), "thread_webdrop_none"),
+        ):
+            dropped = self.run_answer_stream(
+                answer,
+                query="What does that site say?",
+                enabled_tools=enabled_tools,
+                thread_id=thread_id,
+            )
+            self.assertEqual(dropped, [])
+
+    def test_pasted_url_surfaces_in_fallback_when_nothing_cited(self) -> None:
+        sources = self.run_answer_stream(
+            "Here is a summary with no inline citations.",
+            query="Summarize https://example.com/post please",
+            enabled_tools=("url_context",),
+            thread_id="thread_urlfallback",
+        )
+
+        self.assertEqual(len(sources), 1)
+        self.assertEqual(sources[0].data["source_key"], "url_context")
+        self.assertEqual(sources[0].data["url"], "https://example.com/post")
+
     def test_idle_threads_evicted_with_checkpoints_and_records(self) -> None:
         self.addCleanup(_drop_thread_record, "thread_stale")
         self.addCleanup(_drop_thread_record, "thread_fresh")

tests/test_config.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import os
+from contextlib import ExitStack
+from pathlib import Path
+from unittest.mock import patch
+
+from app import config
+
+
+def _patched_bundle(tmp_path: Path) -> ExitStack:
+    db_dir = tmp_path / "chroma"
+    kb_dir = tmp_path / "kb"
+    stack = ExitStack()
+    for target, value in (
+        ("VECTOR_DB_DIR", db_dir),
+        ("CHROMA_SQLITE_PATH", db_dir / "chroma.sqlite3"),
+        ("DOCUMENT_DICT_PATH", db_dir / "document_dict.pkl"),
+        ("BM25_INDEX_PATH", db_dir / "bm25.pkl"),
+        ("KB_MANIFEST_PATH", kb_dir / "generated" / "corpus_manifest.jsonl"),
+        ("KB_INDEX_PATH", kb_dir / "wiki" / "index.md"),
+        ("KB_AGENTS_PATH", kb_dir / "AGENTS.md"),
+        ("KB_AGENTS_TEMPLATE_PATH", tmp_path / "template.md"),
+    ):
+        stack.enter_context(patch(f"app.config.{target}", str(value)))
+    stack.enter_context(patch("app.config._BUNDLE_READY", False))
+    return stack
+
+
+def _write_bundle_files(tmp_path: Path, *, include_bm25: bool = True) -> None:
+    db_dir = tmp_path / "chroma"
+    db_dir.mkdir(parents=True, exist_ok=True)
+    (db_dir / "chroma.sqlite3").write_bytes(b"")
+    (db_dir / "document_dict.pkl").write_bytes(b"")
+    if include_bm25:
+        (db_dir / "bm25.pkl").write_bytes(b"")
+    (tmp_path / "kb" / "generated").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "kb" / "generated" / "corpus_manifest.jsonl").write_text(
+        "", encoding="utf-8"
+    )
+    (tmp_path / "kb" / "wiki").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "kb" / "wiki" / "index.md").write_text("# index", encoding="utf-8")
+    (tmp_path / "template.md").write_text("# KB rules", encoding="utf-8")
+
+
+def test_missing_bm25_triggers_download(tmp_path: Path) -> None:
+    # An interrupted download that left BM25 behind must be repaired, not
+    # silently degrade hybrid retrieval to dense-only forever.
+    _write_bundle_files(tmp_path, include_bm25=False)
+
+    with _patched_bundle(tmp_path):
+        with patch("huggingface_hub.snapshot_download") as snapshot_download:
+            config.ensure_local_vector_db()
+
+        snapshot_download.assert_called_once()
+        # The mock download produced nothing, so readiness must stay false
+        # (the next call retries instead of trusting a broken bundle).
+        assert config._BUNDLE_READY is False
+
+
+def test_complete_bundle_skips_download_and_caches_readiness(tmp_path: Path) -> None:
+    _write_bundle_files(tmp_path)
+
+    with _patched_bundle(tmp_path):
+        with patch("huggingface_hub.snapshot_download") as snapshot_download:
+            config.ensure_local_vector_db()
+            assert snapshot_download.call_count == 0
+            assert config._BUNDLE_READY is True
+            assert (tmp_path / "kb" / "AGENTS.md").read_text(
+                encoding="utf-8"
+            ) == "# KB rules"
+
+            # Once verified, later calls are flag-checks: no re-stat, no
+            # download attempt even if files vanish mid-process.
+            (tmp_path / "chroma" / "bm25.pkl").unlink()
+            config.ensure_local_vector_db()
+            assert snapshot_download.call_count == 0
+
+
+def test_ensure_kb_agents_md_writes_once_and_atomically(tmp_path: Path) -> None:
+    _write_bundle_files(tmp_path)
+
+    with _patched_bundle(tmp_path):
+        with patch("app.config.os.replace", wraps=os.replace) as replace:
+            config.ensure_kb_agents_md()
+            assert replace.call_count == 1
+            # Unchanged content: no rewrite at all.
+            config.ensure_kb_agents_md()
+            assert replace.call_count == 1
+
+            # Changed template: rewritten through the atomic path.
+            (tmp_path / "template.md").write_text("# KB rules v2", encoding="utf-8")
+            config.ensure_kb_agents_md()
+            assert replace.call_count == 2
+
+        agents_path = tmp_path / "kb" / "AGENTS.md"
+        assert agents_path.read_text(encoding="utf-8") == "# KB rules v2"
+        assert not agents_path.with_name("AGENTS.md.tmp").exists()

tests/test_prompts.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
 from pathlib import Path
+from unittest.mock import patch
 
-from app.prompts import build_system_prompt, load_kb_agents_instructions
+from app.prompts import (
+    build_system_prompt,
+    ensure_kb_agents_instructions,
+    load_kb_agents_instructions,
+)
 
 
 def test_build_system_prompt_includes_local_kb_agents_instructions(
@@ -41,3 +46,61 @@ def test_load_kb_agents_instructions_returns_empty_for_missing_path(
     tmp_path: Path,
 ) -> None:
     assert load_kb_agents_instructions(tmp_path / "missing.md") == ""
+
+
+def test_build_system_prompt_uses_explicit_instructions_without_file_read(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    # The file on disk must be ignored when instructions are passed in.
+    agents_path = tmp_path / "AGENTS.md"
+    agents_path.write_text("# From disk\n", encoding="utf-8")
+    monkeypatch.setenv("AI_TUTOR_KB_AGENTS_PATH", str(agents_path))
+
+    explicit = build_system_prompt(
+        "google-genai:gemini-3.5-flash",
+        (),
+        kb_agents_instructions="# Explicit KB rules",
+    )
+    assert "# Explicit KB rules" in explicit
+    assert "# From disk" not in explicit
+
+    empty = build_system_prompt(
+        "google-genai:gemini-3.5-flash",
+        (),
+        kb_agents_instructions="",
+    )
+    assert "## Local KB Instructions" not in empty
+
+
+def test_ensure_kb_agents_instructions_materializes_from_template(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    template_path = tmp_path / "kb_agents_template.md"
+    template_path.write_text("# Generated KB rules\n", encoding="utf-8")
+    agents_path = tmp_path / "kb" / "AGENTS.md"
+    monkeypatch.setenv("AI_TUTOR_KB_AGENTS_PATH", str(agents_path))
+
+    with (
+        patch("app.config.KB_AGENTS_PATH", str(agents_path)),
+        patch("app.config.KB_AGENTS_TEMPLATE_PATH", str(template_path)),
+    ):
+        instructions = ensure_kb_agents_instructions()
+
+    assert instructions == "# Generated KB rules"
+    assert agents_path.is_file()
+
+
+def test_ensure_kb_agents_instructions_empty_when_template_missing(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    agents_path = tmp_path / "kb" / "AGENTS.md"
+    monkeypatch.setenv("AI_TUTOR_KB_AGENTS_PATH", str(agents_path))
+
+    with (
+        patch("app.config.KB_AGENTS_PATH", str(agents_path)),
+        patch("app.config.KB_AGENTS_TEMPLATE_PATH", str(tmp_path / "missing.md")),
+    ):
+        assert ensure_kb_agents_instructions() == ""