Enhance knowledge base and API functionality

- Updated .gitignore to include local environment files for better configuration management.
- Enhanced the build_chat_request function to handle explicit empty source selections, ensuring that the knowledge base is correctly disabled when no sources are selected.
- Improved the UIMessageStreamEncoder to include additional metadata for source matches, enhancing the information available to the frontend.
- Introduced reranking logic in the rerank_results function to score matched chunks instead of entire documents, optimizing relevance and performance.
- Added comprehensive tests to validate the new functionalities, ensuring robustness in knowledge base interactions and API responses.

.gitignore

@@ -121,6 +121,8 @@ celerybeat.pid
 
 # Environments
 .env
+.env.local
+.env.*.local
 .venv
 env/
 venv/

app/api.py

@@ -33,6 +33,7 @@ from .config import (
     DEFAULT_MODEL_NAME,
     DEFAULT_SELECTED_SOURCE_KEYS,
     DEFAULT_SELECTED_SOURCES_UI,
+    SOURCE_DISPLAY_INFO,
     SOURCE_UI_TO_KEY,
 )
 
@@ -210,15 +211,25 @@ def build_chat_request(payload: ApiChatRequest) -> ChatRequest:
         )
 
     allowed_source_keys = set(AVAILABLE_SOURCES)
-    requested_source_keys = payload.sourceKeys or list(DEFAULT_SELECTED_SOURCE_KEYS)
-    source_keys = (
-        tuple(
-            dict.fromkeys(
-                key for key in requested_source_keys if key in allowed_source_keys
+    if payload.sourceKeys is not None and not payload.sourceKeys:
+        # An explicit empty selection is the user turning the knowledge base
+        # off; it must not silently coerce to the defaults. Only an *omitted*
+        # field means "use the defaults".
+        source_keys: tuple[str, ...] = ()
+    else:
+        requested_source_keys = (
+            payload.sourceKeys
+            if payload.sourceKeys is not None
+            else list(DEFAULT_SELECTED_SOURCE_KEYS)
+        )
+        source_keys = (
+            tuple(
+                dict.fromkeys(
+                    key for key in requested_source_keys if key in allowed_source_keys
+                )
             )
+            or DEFAULT_SELECTED_SOURCE_KEYS
         )
-        or DEFAULT_SELECTED_SOURCE_KEYS
-    )
     model_name = (payload.model or DEFAULT_MODEL_NAME).strip() or DEFAULT_MODEL_NAME
     if model_name not in {model["id"] for model in AVAILABLE_MODELS}:
         raise HTTPException(status_code=422, detail="Unknown model")
@@ -292,6 +303,10 @@ class UIMessageStreamEncoder:
             "sourceLabel": str(data.get("source_label", "")),
             "score": float(data.get("score", 0.0)),
             "group": str(data.get("group", "")),
+            # KB-root-relative path ("raw/docs/...") when the source is a KB
+            # file; the client uses it to resolve inline `raw/...` citations
+            # to this source's real URL.
+            "path": str(data.get("path", "")),
         }
 
     def encode(self, event: ChatEvent) -> list[dict[str, Any]]:
@@ -303,6 +318,11 @@ class UIMessageStreamEncoder:
                 {
                     "type": "data-thread",
                     "data": {"threadId": self.thread_id},
+                    # Delivered to onData only, never stored in message.parts:
+                    # nothing renders it, and a non-transient part emitted
+                    # before "start" relies on undocumented AI SDK ordering
+                    # behavior (the in-flight message is stored by reference).
+                    "transient": True,
                 }
             )
             return parts
@@ -550,8 +570,15 @@ def _source_entries() -> list[dict[str, Any]]:
     for label in AVAILABLE_SOURCES_UI:
         key = SOURCE_UI_TO_KEY[label]
         group = "courses" if key in COURSE_SOURCE_KEYS else "docs"
+        display = SOURCE_DISPLAY_INFO.get(key, {})
         entry: dict[str, Any] = {
             "label": label,
+            # Display metadata is registry-owned (single source of truth);
+            # the frontend renders these verbatim instead of keeping its own
+            # per-source maps or reshaping the label.
+            "shortLabel": display.get("ui_label") or label,
+            "description": display.get("description"),
+            "infoUrl": display.get("url"),
             "key": key,
             "group": group,
             "selectedByDefault": label in defaults,

app/chat_service.py

@@ -818,15 +818,17 @@ def build_agent(
     enabled_tools: tuple[str, ...] = (),
     include_thoughts: bool = False,
     kb_agents_instructions: str | None = None,
+    include_local_tools: bool = True,
 ):
     # 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,
-        run_kb_command,
-    ]
+    # An explicit empty source selection turns the knowledge base off: no
+    # retrieval, no KB browsing, and a system prompt that says so.
+    tools: list[Any] = (
+        [retrieve_tutor_context, run_kb_command] if include_local_tools else []
+    )
     middleware: list[AgentMiddleware] = [
         ContextEditingMiddleware(
             edits=[
@@ -876,7 +878,10 @@ def build_agent(
         model=model,
         tools=tools,
         system_prompt=build_system_prompt(
-            model_name, enabled_tools, kb_agents_instructions
+            model_name,
+            enabled_tools,
+            kb_agents_instructions,
+            include_local_tools=include_local_tools,
         ),
         context_schema=AppContext,
         checkpointer=CHECKPOINTER,
@@ -893,8 +898,9 @@ def model_provider_and_name(model_name: str) -> tuple[str, str]:
 def effective_tool_names(
     model_name: str,
     enabled_tools: tuple[str, ...],
+    include_local_tools: bool = True,
 ) -> tuple[str, ...]:
-    names = ["retrieve_tutor_context", "run_kb_command"]
+    names = ["retrieve_tutor_context", "run_kb_command"] if include_local_tools else []
     enabled = set(enabled_tools)
     if is_google_genai_model(model_name):
         if "web_search" in enabled:
@@ -950,7 +956,11 @@ def agent_run_config(
     message_id: str,
 ) -> dict[str, Any]:
     provider, actual_model = model_provider_and_name(request.model_name)
-    tools = effective_tool_names(request.model_name, request.enabled_tools)
+    tools = effective_tool_names(
+        request.model_name,
+        request.enabled_tools,
+        include_local_tools=bool(request.source_keys),
+    )
     source_labels = [
         SOURCE_KEY_TO_LABEL.get(source_key, source_key)
         for source_key in request.source_keys
@@ -1003,7 +1013,12 @@ 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)
+    include_local_tools = bool(request.source_keys)
+    effective_tools = effective_tool_names(
+        request.model_name,
+        request.enabled_tools,
+        include_local_tools=include_local_tools,
+    )
     if "url_context" in effective_tools:
         _record_evidence(web_evidence, url_context_evidence(request.query))
 
@@ -1017,7 +1032,10 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
             request.model_name,
             enabled_tools=tuple(request.enabled_tools),
             include_thoughts=include_reasoning,
-            kb_agents_instructions=ensure_kb_agents_instructions(),
+            kb_agents_instructions=(
+                ensure_kb_agents_instructions() if include_local_tools else None
+            ),
+            include_local_tools=include_local_tools,
         )
 
     agent = await asyncio.to_thread(_build_agent_for_request)

app/chat_types.py

@@ -19,6 +19,9 @@ class SourceMatch:
     source_label: str
     score: float
     group: str = ""
+    # KB-root-relative file path ("raw/docs/...") for manifest-backed matches;
+    # lets the client map inline `raw/...` citations to this source's real URL.
+    path: str = ""
 
 
 @dataclass(frozen=True, slots=True)

app/chroma_rag.py

@@ -964,6 +964,20 @@ def embed_texts(
     return vectors
 
 
+def _rerank_document(result: SearchResult) -> str:
+    """Text the reranker scores for a result.
+
+    For ``retrieve_doc`` results ``content`` is the *entire* document, which
+    both dilutes the relevance signal toward the document's average (instead of
+    the chunk that actually matched) and can overflow Cohere's per-document
+    token limit when several full docs are reranked together. Score the matched
+    chunk instead; the full document is still what gets returned for the answer.
+    """
+    if result.retrieve_doc and result.chunk_content:
+        return result.chunk_content
+    return result.content
+
+
 def rerank_results(
     client: cohere.ClientV2,
     query: str,
@@ -977,7 +991,7 @@ def rerank_results(
     response = client.rerank(
         model=model,
         query=query,
-        documents=[result.content for result in results],
+        documents=[_rerank_document(result) for result in results],
         top_n=min(top_n, len(results)),
     )
 

app/config.py

@@ -11,6 +11,7 @@ from data.scraping_scripts.source_registry import (
     COURSE_SOURCE_KEYS,
     DEFAULT_SELECTED_SOURCE_KEYS,
     DEFAULT_SELECTED_SOURCES_UI,
+    SOURCE_DISPLAY_INFO,
     SOURCE_KEY_TO_LABEL,
     SOURCE_UI_TO_KEY,
 )
@@ -151,6 +152,7 @@ __all__ = [
     "KB_DIR",
     "KB_INDEX_PATH",
     "KB_MANIFEST_PATH",
+    "SOURCE_DISPLAY_INFO",
     "SOURCE_KEY_TO_LABEL",
     "SOURCE_UI_TO_KEY",
     "VECTOR_COLLECTION_NAME",

app/kb_manifest.py

@@ -1,15 +1,17 @@
 from __future__ import annotations
 
 import json
+import logging
 import re
 from dataclasses import dataclass
-from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
 from .chat_types import SourceMatch
 from .config import COURSE_SOURCE_KEYS, KB_DIR, SOURCE_KEY_TO_LABEL
 
+logger = logging.getLogger(__name__)
+
 KB_DOC_SCHEME_RE = re.compile(r"^kb://doc/(?P<doc_id>[^)\]\s]+)$")
 RAW_PATH_RE = re.compile(r"(?:data/kb/)?raw/[^\s)\]>,:]+?\.(?:mdx|md)")
 
@@ -35,18 +37,57 @@ def _normalize_path(value: str) -> str:
     return value
 
 
-@lru_cache(maxsize=4)
+def kb_root_path(value: str) -> str:
+    """KB-root-relative form of a manifest path ("raw/docs/...") — the shape
+    the model is instructed to cite and the client matches against."""
+    value = value.strip()
+    if value.startswith("./"):
+        value = value[2:]
+    if value.startswith(f"{KB_DIR}/"):
+        value = value[len(KB_DIR) + 1 :]
+    return value
+
+
+# Cache parsed manifests per kb_dir, but only once the file actually exists, so
+# a lookup during the first-start download window does not pin an empty result.
+_MANIFEST_CACHE: dict[str, tuple[KbManifestEntry, ...]] = {}
+
+
 def load_manifest_entries(kb_dir: str = KB_DIR) -> tuple[KbManifestEntry, ...]:
+    cached = _MANIFEST_CACHE.get(kb_dir)
+    if cached is not None:
+        return cached
+
     manifest_path = Path(kb_dir) / "generated" / "corpus_manifest.jsonl"
     if not manifest_path.exists():
+        # Do not cache the missing-file case: the KB bundle may still be
+        # downloading on first start, so retry on the next call instead of
+        # pinning an empty manifest (and thus never resolving citations) for
+        # the process lifetime.
         return ()
 
     entries: list[KbManifestEntry] = []
     with manifest_path.open("r", encoding="utf-8") as handle:
-        for line in handle:
+        for line_number, line in enumerate(handle, start=1):
             if not line.strip():
                 continue
-            row = json.loads(line)
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as exc:
+                logger.warning(
+                    "Skipping malformed manifest line %d in %s: %s",
+                    line_number,
+                    manifest_path,
+                    exc,
+                )
+                continue
+            if not isinstance(row, dict):
+                logger.warning(
+                    "Skipping non-object manifest line %d in %s",
+                    line_number,
+                    manifest_path,
+                )
+                continue
             entries.append(
                 KbManifestEntry(
                     doc_id=str(row.get("doc_id") or ""),
@@ -57,7 +98,9 @@ def load_manifest_entries(kb_dir: str = KB_DIR) -> tuple[KbManifestEntry, ...]:
                     path=str(row.get("path") or ""),
                 )
             )
-    return tuple(entries)
+    result = tuple(entries)
+    _MANIFEST_CACHE[kb_dir] = result
+    return result
 
 
 def manifest_indexes(
@@ -72,6 +115,11 @@ def manifest_indexes(
     by_url: dict[str, KbManifestEntry] = {}
     by_path: dict[str, KbManifestEntry] = {}
     by_title: dict[str, KbManifestEntry] = {}
+    # Titles like "Introduction"/"Quickstart" recur across docs; a plain dict
+    # would resolve a bare-label citation to whichever doc was ingested last and
+    # surface a misleading source card. Track collisions and refuse to resolve
+    # an ambiguous title to any single doc.
+    ambiguous_titles: set[str] = set()
     for entry in load_manifest_entries(kb_dir):
         if entry.doc_id:
             by_doc_id[entry.doc_id] = entry
@@ -82,7 +130,15 @@ def manifest_indexes(
             if entry.path.startswith(f"{KB_DIR}/"):
                 by_path[entry.path[len(KB_DIR) + 1 :]] = entry
         if entry.title:
-            by_title[entry.title.strip().lower()] = entry
+            title_key = entry.title.strip().lower()
+            if title_key in ambiguous_titles:
+                continue
+            existing = by_title.get(title_key)
+            if existing is not None and existing.doc_id != entry.doc_id:
+                ambiguous_titles.add(title_key)
+                del by_title[title_key]
+            else:
+                by_title[title_key] = entry
     return by_doc_id, by_url, by_path, by_title
 
 
@@ -99,6 +155,7 @@ def source_match_from_manifest(
         group="courses"
         if entry.source in COURSE_SOURCE_KEYS
         else entry.source_group or "docs",
+        path=kb_root_path(entry.path),
     )
 
 
@@ -188,6 +245,7 @@ def source_match_payload(
         "source_label": match.source_label,
         "score": match.score,
         "group": match.group,
+        "path": match.path,
     }
     if call_id:
         payload["call_id"] = call_id

app/kb_shell.py

@@ -5,8 +5,10 @@ import re
 import shlex
 import shutil
 import subprocess
+import threading
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
 
 DEFAULT_KB_DIR = Path(os.getenv("AI_TUTOR_KB_DIR", "data/kb"))
 DEFAULT_TIMEOUT_SECONDS = 8
@@ -44,6 +46,7 @@ RG_FLAG_OPTIONS = frozenset(
 GREP_FLAG_OPTIONS = frozenset(
     {"-i", "--ignore-case", "-w", "--word-regexp", "-F", "-E"}
 )
+GREP_VALUE_OPTIONS = frozenset({"-m", "--max-count"})
 LS_FLAG_OPTIONS = frozenset({"-1", "-a", "-l", "-la", "-al"})
 WC_FLAG_OPTIONS = frozenset({"-l", "-w", "-c", "-m"})
 
@@ -170,12 +173,16 @@ def _build_rg(tokens: list[str], root: Path) -> list[str]:
     pattern = positionals[0]
     paths = [_relative_path_arg(value, root) for value in positionals[1:]] or ["."]
     _reject_unbounded_raw_search("rg", paths, has_max_count)
+    # `--` ends option parsing so a pattern beginning with `-`/`--` (e.g.
+    # `--pre=/bin/sh`, `--hostname-bin`) can never be re-interpreted by rg as a
+    # flag that spawns an external process or escapes the read-only jail.
     return [
         "rg",
         "--color=never",
         "--line-number",
         "--no-heading",
         *options,
+        "--",
         pattern,
         *paths,
     ]
@@ -184,6 +191,7 @@ def _build_rg(tokens: list[str], root: Path) -> list[str]:
 def _build_grep(tokens: list[str], root: Path) -> list[str]:
     options: list[str] = []
     positionals: list[str] = []
+    has_max_count = False
     idx = 1
     parsing_options = True
     while idx < len(tokens):
@@ -196,6 +204,13 @@ def _build_grep(tokens: list[str], root: Path) -> list[str]:
             options.append(token)
             idx += 1
             continue
+        if parsing_options and token in GREP_VALUE_OPTIONS:
+            if idx + 1 >= len(tokens):
+                raise KbCommandError(f"{token} requires a value")
+            options.extend([token, _numeric_value(tokens[idx + 1], token)])
+            has_max_count = True
+            idx += 2
+            continue
         positionals.append(token)
         parsing_options = False
         idx += 1
@@ -203,7 +218,14 @@ def _build_grep(tokens: list[str], root: Path) -> list[str]:
         raise KbCommandError("`grep` requires a search pattern")
     pattern = positionals[0]
     paths = [_relative_path_arg(value, root) for value in positionals[1:]] or ["."]
-    return ["grep", "-R", "-n", "--color=never", *options, pattern, *paths]
+    # Bound grep the same way as rg: a broad `raw/` recursion needs an explicit
+    # -m/--max-count (or a narrower path) so it can't emit unbounded output.
+    _reject_unbounded_raw_search("grep", paths, has_max_count)
+    # Lowercase `-r` follows only command-line symlinks (not symlinks discovered
+    # during the walk), unlike `-R`; per-file paths found during recursion are
+    # not re-validated by the jail, so the more permissive `-R` is avoided.
+    # `--` ends option parsing so a `-`/`--` pattern can't be re-read as a flag.
+    return ["grep", "-r", "-n", "--color=never", *options, "--", pattern, *paths]
 
 
 def _build_find(tokens: list[str], root: Path) -> list[str]:
@@ -329,11 +351,46 @@ def build_kb_command_argv(
     return builders[executable](tokens, resolved_root), resolved_root
 
 
-def _truncate(text: str, max_chars: int) -> tuple[str, bool]:
-    if len(text) <= max_chars:
-        return text, False
+def _drain_stream(stream: Any, cap: int) -> tuple[str, bool]:
+    """Read a child pipe to EOF but retain at most ``cap`` characters.
+
+    Reading continues past the cap (discarding the overflow) so the child never
+    blocks on a full pipe, but peak memory is bounded by ``cap`` instead of the
+    command's full output size.
+    """
+    chunks: list[str] = []
+    retained = 0
+    truncated = False
+    try:
+        while True:
+            data = stream.read(8192)
+            if not data:
+                break
+            if retained >= cap:
+                # Already full; this read is pure overflow we discard.
+                truncated = True
+                continue
+            room = cap - retained
+            if len(data) <= room:
+                chunks.append(data)
+                retained += len(data)
+            else:
+                chunks.append(data[:room])
+                retained = cap
+                truncated = True
+    finally:
+        try:
+            stream.close()
+        except Exception:
+            pass
+    return "".join(chunks), truncated
+
+
+def _apply_truncation_marker(text: str, truncated: bool, max_chars: int) -> str:
+    if not truncated:
+        return text
     marker = f"\n... output truncated to {max_chars} characters ..."
-    return text[: max(0, max_chars - len(marker))] + marker, True
+    return text[: max(0, max_chars - len(marker))] + marker
 
 
 def run_kb_command(
@@ -346,6 +403,7 @@ def run_kb_command(
     argv, resolved_root = build_kb_command_argv(command, root=root)
     timeout = max(1, min(int(timeout_seconds), 30))
     output_limit = max(1_000, min(int(max_output_chars), 80_000))
+    stderr_limit = min(output_limit, 8_000)
     env = {
         "HOME": os.getenv("HOME", ""),
         "LANG": os.getenv("LANG", "C.UTF-8"),
@@ -353,30 +411,55 @@ def run_kb_command(
         "PATH": os.getenv("PATH", ""),
     }
     timed_out = False
+    stdout_box: dict[str, Any] = {"text": "", "truncated": False}
+    stderr_box: dict[str, Any] = {"text": "", "truncated": False}
+
+    # Stream both pipes through capped reader threads so a command that emits a
+    # lot (e.g. `cat` of a large file) cannot spike memory to its full output
+    # size before truncation: capture_output buffers everything, this does not.
+    # argv is fully validated/allowlisted by build_kb_command_argv above.
+    process = subprocess.Popen(
+        argv,
+        cwd=resolved_root,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        env=env,
+    )
+
+    def _read(stream: Any, cap: int, box: dict[str, Any]) -> None:
+        box["text"], box["truncated"] = _drain_stream(stream, cap)
+
+    readers = [
+        threading.Thread(target=_read, args=(process.stdout, output_limit, stdout_box)),
+        threading.Thread(target=_read, args=(process.stderr, stderr_limit, stderr_box)),
+    ]
+    for reader in readers:
+        reader.start()
+
     try:
-        completed = subprocess.run(
-            argv,
-            cwd=resolved_root,
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-            env=env,
-            check=False,
-        )
-        exit_code = int(completed.returncode)
-        stdout = completed.stdout or ""
-        stderr = completed.stderr or ""
-    except subprocess.TimeoutExpired as exc:
+        process.wait(timeout=timeout)
+        exit_code = int(process.returncode)
+    except subprocess.TimeoutExpired:
         timed_out = True
         exit_code = 124
-        stdout = exc.stdout if isinstance(exc.stdout, str) else ""
-        stderr = exc.stderr if isinstance(exc.stderr, str) else ""
+        process.kill()
+        process.wait()
+    for reader in readers:
+        reader.join()
+
+    stdout = _apply_truncation_marker(
+        stdout_box["text"], bool(stdout_box["truncated"]), output_limit
+    )
+    stdout_truncated = bool(stdout_box["truncated"])
+    stderr = _apply_truncation_marker(
+        stderr_box["text"], bool(stderr_box["truncated"]), stderr_limit
+    )
+    stderr_truncated = bool(stderr_box["truncated"])
+    if timed_out:
         stderr = (
             stderr + "\n" if stderr else ""
         ) + f"Command timed out after {timeout}s."
-
-    stdout, stdout_truncated = _truncate(stdout, output_limit)
-    stderr, stderr_truncated = _truncate(stderr, min(output_limit, 8_000))
     return KbCommandResult(
         command=command,
         argv=argv,

app/prompts.py

@@ -158,6 +158,24 @@ ANSWERING_RULES = """## Answering rules
 
 The retrieval tool returns JSON with matched passages and source metadata."""
 
+NO_KB_NOTE = """## Knowledge base disabled
+
+The user deselected every knowledge-base source for this conversation, so the
+corpus retrieval and KB browsing tools are unavailable. Answer from general
+knowledge (and the web tools above, when available). Do not claim to have
+searched the course corpus; mention briefly when an answer would be stronger
+with course sources enabled."""
+
+NO_KB_ANSWERING_RULES = """## Answering rules
+
+- When you use web tools, cite sources with inline Markdown links: place a
+  `[short source title](url)` citation in the same sentence, immediately
+  after the claim it supports.
+- Synthesize a clear teaching explanation. Prefer a few solid paragraphs over
+  shallow bullet lists.
+- Include complete, runnable code blocks when code is relevant.
+- End with a short invitation for a follow-up question."""
+
 
 def _provider_key(model_name: str) -> str:
     normalized = (model_name or "").strip()
@@ -215,10 +233,11 @@ def build_system_prompt(
     model_name: str,
     enabled_tools: tuple[str, ...],
     kb_agents_instructions: str | None = None,
+    include_local_tools: bool = True,
 ) -> str:
     provider = _provider_key(model_name)
     enabled = set(enabled_tools)
-    tool_lines = [RETRIEVAL_TOOL_LINE, *KB_TOOL_LINES]
+    tool_lines = [RETRIEVAL_TOOL_LINE, *KB_TOOL_LINES] if include_local_tools else []
     usage_sections: list[str] = []
     provider_web_tools = WEB_TOOL_LINES.get(provider, {})
     provider_web_usage = WEB_USAGE_SECTIONS.get(provider, {})
@@ -227,26 +246,33 @@ def build_system_prompt(
             tool_lines.append(provider_web_tools[key])
             usage_sections.append(provider_web_usage[key])
 
-    if len(tool_lines) == 1:
-        intro = "You have one tool available:"
-    else:
-        intro = f"You have {len(tool_lines)} tools available:"
-
-    parts = [
-        BASE_PROMPT_HEADER,
-        f"{intro}\n\n" + "\n".join(tool_lines),
-        RETRIEVAL_USAGE_SECTION,
-        KB_USAGE_SECTION,
-    ]
+    parts = [BASE_PROMPT_HEADER]
+    if tool_lines:
+        if len(tool_lines) == 1:
+            intro = "You have one tool available:"
+        else:
+            intro = f"You have {len(tool_lines)} tools available:"
+        parts.append(f"{intro}\n\n" + "\n".join(tool_lines))
+
+    if include_local_tools:
+        parts.append(RETRIEVAL_USAGE_SECTION)
+        parts.append(KB_USAGE_SECTION)
     if usage_sections:
         parts.append(
             "## When to use web search / URL reading\n\n" + "\n\n".join(usage_sections)
         )
-        parts.append(
-            "Prefer `retrieve_tutor_context` first when the question is clearly about\n"
-            "course material. Combine tools when it helps (e.g. retrieve corpus\n"
-            "context, then search the web for the latest update)."
-        )
+        if include_local_tools:
+            parts.append(
+                "Prefer `retrieve_tutor_context` first when the question is clearly about\n"
+                "course material. Combine tools when it helps (e.g. retrieve corpus\n"
+                "context, then search the web for the latest update)."
+            )
+    if not include_local_tools:
+        # The user deselected every source: the prompt must not describe or
+        # instruct tools the agent does not have this turn.
+        parts.append(NO_KB_NOTE)
+        parts.append(NO_KB_ANSWERING_RULES)
+        return "\n\n".join(parts) + "\n"
     if kb_agents_instructions is None:
         kb_agents_instructions = load_kb_agents_instructions()
     if kb_agents_instructions:

data/scraping_scripts/source_registry.py

@@ -268,6 +268,120 @@ SOURCE_KEY_TO_LABEL = {
     "agentic_ai_engineering": "Agentic AI Engineering",
 }
 
+# Display metadata served to the UI via /api/tools. The frontend renders these
+# verbatim (single source of truth: adding a source here means no separate UI
+# edit). `ui_label` is the short sidebar name; `label` in SOURCE_KEY_TO_LABEL
+# stays the full name used on source cards and in prompts. Keep descriptions
+# free of em-dashes (frontend user-facing text convention).
+SOURCE_DISPLAY_INFO: dict[str, dict[str, str]] = {
+    "transformers": {
+        "ui_label": "Transformers",
+        "description": (
+            "Hugging Face library for state-of-the-art NLP and multimodal "
+            "models. Load, run, and train pretrained transformers."
+        ),
+        "url": "https://huggingface.co/docs/transformers",
+    },
+    "peft": {
+        "ui_label": "PEFT",
+        "description": (
+            "Parameter-Efficient Fine-Tuning: LoRA, prefix tuning, and other "
+            "methods for adapting large models with minimal compute."
+        ),
+        "url": "https://huggingface.co/docs/peft",
+    },
+    "trl": {
+        "ui_label": "TRL",
+        "description": (
+            "Train language models with reinforcement learning. Covers SFT, "
+            "DPO, PPO, and other alignment techniques."
+        ),
+        "url": "https://huggingface.co/docs/trl",
+    },
+    "llama_index": {
+        "ui_label": "LlamaIndex",
+        "description": (
+            "Framework for building RAG apps: ingestion, indexing, "
+            "retrievers, and query engines over your own data."
+        ),
+        "url": "https://docs.llamaindex.ai",
+    },
+    "langchain": {
+        "ui_label": "LangChain",
+        "description": (
+            "Framework for building LLM apps: chains, agents, tool-calling, "
+            "and production observability."
+        ),
+        "url": "https://docs.langchain.com/oss/python/langchain/overview",
+    },
+    "langgraph": {
+        "ui_label": "LangGraph",
+        "description": (
+            "Graph-based runtime for reliable, stateful AI agents with "
+            "persistence, streaming, human review, and deployment patterns."
+        ),
+        "url": "https://docs.langchain.com/oss/python/langgraph/overview",
+    },
+    "deep_agents": {
+        "ui_label": "Deep Agents",
+        "description": (
+            "LangChain's deep agent harness for planning, delegation, "
+            "filesystem context, and longer-running agent workflows."
+        ),
+        "url": "https://docs.langchain.com/oss/python/deepagents/overview",
+    },
+    "openai_docs": {
+        "ui_label": "OpenAI",
+        "description": (
+            "Official OpenAI API, Agents SDK, and Codex documentation from "
+            "the developer docs Markdown index."
+        ),
+        "url": "https://developers.openai.com",
+    },
+    "claude_code_docs": {
+        "ui_label": "Claude Code",
+        "description": (
+            "Official Claude Code and Claude Agent SDK documentation from "
+            "Anthropic's Markdown index."
+        ),
+        "url": "https://code.claude.com/docs/en/overview",
+    },
+    "full_stack_ai_engineering": {
+        "ui_label": "Full Stack AI Engineering",
+        "description": (
+            "Full-stack LLM engineering, covering RAG, fine-tuning, "
+            "evaluation, and deploying production systems end-to-end. The "
+            "deepest technical course."
+        ),
+        "url": "https://academy.towardsai.net/courses/beginner-to-advanced-llm-dev",
+    },
+    "beginner_python_for_ai_engineering": {
+        "ui_label": "Beginner Python for AI Engineering",
+        "description": (
+            "Python for the LLM era: API integration, using open-source "
+            "models, and core training/testing workflows. Assumes no prior "
+            "Python."
+        ),
+        "url": "https://academy.towardsai.net/courses/python-for-genai",
+    },
+    "master_ai_for_work": {
+        "ui_label": "Master AI For Work",
+        "description": (
+            "Non-engineer course on using AI tools (ChatGPT, Claude, etc.) "
+            "for workplace productivity and rolling them out across a team."
+        ),
+        "url": "https://academy.towardsai.net/courses/ai-business-professionals",
+    },
+    "agentic_ai_engineering": {
+        "ui_label": "Agentic AI Engineering",
+        "description": (
+            "Designing, building, evaluating, and deploying production-grade "
+            "AI agents end-to-end."
+        ),
+        "url": "https://academy.towardsai.net/courses/agent-engineering",
+    },
+}
+
 UI_SOURCE_KEYS = (
     "openai_docs",
     "claude_code_docs",

tests/test_api.py

@@ -64,6 +64,17 @@ class ApiTestCase(unittest.TestCase):
                 for source in retrieval["sources"]
             )
         )
+        # Display metadata is registry-owned and served per source so the
+        # frontend renders it verbatim (no client-side maps or label edits).
+        for source in retrieval["sources"]:
+            self.assertTrue(source["shortLabel"])
+            self.assertTrue(source["description"])
+            self.assertTrue(str(source["infoUrl"]).startswith("https://"))
+        transformers = next(
+            source for source in retrieval["sources"] if source["key"] == "transformers"
+        )
+        self.assertEqual(transformers["label"], "Transformers Docs")
+        self.assertEqual(transformers["shortLabel"], "Transformers")
         # Gemini is the default model, so web search + url reading are present.
         tool_keys = {tool["key"] for tool in tools}
         self.assertIn("web_search", tool_keys)
@@ -184,6 +195,10 @@ class ApiTestCase(unittest.TestCase):
         part_types = [part["type"] for part in parts]
 
         self.assertIn("data-thread", part_types)
+        data_thread = next(part for part in parts if part["type"] == "data-thread")
+        # Transient: consumed via onData only; must not land in message.parts
+        # (it would also rely on undocumented ordering when emitted pre-start).
+        self.assertTrue(data_thread.get("transient"))
         self.assertIn("start", part_types)
         self.assertIn("start-step", part_types)
         self.assertIn("reasoning-start", part_types)
@@ -499,6 +514,7 @@ class ApiTestCase(unittest.TestCase):
                             "source_label": "PEFT Docs",
                             "score": 0.9,
                             "group": "docs",
+                            "path": "raw/docs/peft/lora.md",
                         }
                     ],
                 },
@@ -511,6 +527,7 @@ class ApiTestCase(unittest.TestCase):
         self.assertEqual(matches[0]["docId"], "peft:lora")
         self.assertEqual(matches[0]["sourceKey"], "peft")
         self.assertEqual(matches[0]["score"], 0.9)
+        self.assertEqual(matches[0]["path"], "raw/docs/peft/lora.md")
 
     def test_chat_rejects_oversized_query(self) -> None:
         from app.api import MAX_QUERY_CHARS
@@ -678,6 +695,21 @@ class ApiTestCase(unittest.TestCase):
 
         self.assertEqual(response.status_code, 422)
 
+    def test_explicit_empty_source_keys_disable_retrieval(self) -> None:
+        from app.api import ApiChatRequest, build_chat_request
+        from app.config import DEFAULT_SELECTED_SOURCE_KEYS
+
+        # Explicit [] is the user turning the knowledge base off; it must not
+        # silently coerce to the defaults (the UI shows it as "off").
+        explicit_empty = build_chat_request(
+            ApiChatRequest(query="What is RAG?", sourceKeys=[])
+        )
+        self.assertEqual(explicit_empty.source_keys, ())
+
+        # An omitted field still means "use the defaults".
+        omitted = build_chat_request(ApiChatRequest(query="What is RAG?"))
+        self.assertEqual(omitted.source_keys, tuple(DEFAULT_SELECTED_SOURCE_KEYS))
+
 
 LIVE_API_E2E = pytest.mark.skipif(
     os.getenv("RUN_LIVE_API_E2E") != "1",

tests/test_chat_service.py

@@ -444,6 +444,42 @@ class ChatServiceTestCase(unittest.TestCase):
         self.assertEqual(source_matches[0].data["source_key"], "peft")
         self.assertNotIn("call_id", source_matches[0].data)
 
+    def test_stream_chat_disables_local_tools_when_no_sources_selected(self) -> None:
+        # An explicit empty source selection (UI "Knowledge base: off") must
+        # build the agent without retrieval/KB tools instead of silently
+        # retrieving from the defaults.
+        agent = FakeStreamingAgent([])
+        build_agent_mock = MagicMock(return_value=agent)
+        self.addCleanup(_drop_thread_record, "thread_no_sources")
+        request = ChatRequest(
+            query="What is RAG?",
+            source_keys=(),
+            model_name="google-genai:gemini-3.5-flash",
+            include_reasoning=False,
+            enabled_tools=(),
+        )
+
+        async def collect_events():
+            return [event async for event in stream_chat(request)]
+
+        with (
+            patch("app.chat_service.build_agent", build_agent_mock),
+            patch("app.chat_service.new_thread_id", return_value="thread_no_sources"),
+        ):
+            asyncio.run(collect_events())
+
+        build_agent_mock.assert_called_once()
+        self.assertFalse(build_agent_mock.call_args.kwargs["include_local_tools"])
+
+        self.assertEqual(
+            effective_tool_names(
+                "google-genai:gemini-3.5-flash",
+                ("web_search",),
+                include_local_tools=False,
+            ),
+            ("google_search",),
+        )
+
     def test_checkpoint_history_collapses_tool_using_turn(self) -> None:
         # A tool-using turn checkpoints as [Human, AI(tool_calls, empty text),
         # ToolMessage, AI(answer)]; the visible transcript has one assistant

tests/test_chroma_rag.py

@@ -17,6 +17,7 @@ from app.chroma_rag import (
     heading_aware_markdown_chunks,
     reciprocal_rank_fusion,
     load_bm25_index,
+    rerank_results,
     SearchResult,
 )
 
@@ -170,6 +171,74 @@ After the example.
         self.assertIn("Heading path: Guide > Setup", setup_record.text)
         self.assertIn("Context: Situated", setup_record.text)
 
+    def test_rerank_scores_matched_chunk_for_retrieve_doc_results(self) -> None:
+        # retrieve_doc results carry the whole document in `content`; the
+        # reranker must score the matched chunk (`chunk_content`) instead, so
+        # relevance is not diluted toward the doc average and the payload stays
+        # within Cohere's per-document token limit.
+        full_doc = "Intro paragraph.\n" * 500
+        results = [
+            SearchResult(
+                chunk_id="doc-chunk",
+                doc_id="doc-1",
+                title="Doc",
+                url="",
+                source="test",
+                retrieve_doc=True,
+                tokens=4000,
+                score=0.5,
+                content=full_doc,
+                chunk_content="the matched chunk about AutoModel",
+                heading_path="section",
+                retrieval_method="dense",
+            ),
+            SearchResult(
+                chunk_id="plain-chunk",
+                doc_id="doc-2",
+                title="Plain",
+                url="",
+                source="test",
+                retrieve_doc=False,
+                tokens=100,
+                score=0.4,
+                content="formatted chunk body",
+                chunk_content="raw chunk body",
+                heading_path="section",
+                retrieval_method="dense",
+            ),
+        ]
+
+        captured: dict[str, list[str]] = {}
+
+        class _FakeItem:
+            def __init__(self, index: int, score: float) -> None:
+                self.index = index
+                self.relevance_score = score
+
+        class _FakeResponse:
+            def __init__(self, items: list["_FakeItem"]) -> None:
+                self.results = items
+
+        class _FakeCohere:
+            def rerank(self, *, model, query, documents, top_n):  # type: ignore[no-untyped-def]
+                captured["documents"] = list(documents)
+                return _FakeResponse(
+                    [
+                        _FakeItem(i, 1.0 - i * 0.1)
+                        for i in range(min(top_n, len(documents)))
+                    ]
+                )
+
+        reranked = rerank_results(_FakeCohere(), "AutoModel", results)
+
+        # The full document never reaches the reranker; the matched chunk does.
+        self.assertEqual(
+            captured["documents"],
+            ["the matched chunk about AutoModel", "formatted chunk body"],
+        )
+        # The returned result still carries the full document for the answer.
+        self.assertEqual(reranked[0].content, full_doc)
+
     def test_rrf_prefers_overlap_across_ranked_lists(self) -> None:
         dense_only = self._result("dense-only", 0.9, "dense")
         overlap_dense = self._result("overlap", 0.7, "dense")

tests/test_kb_manifest.py

@@ -0,0 +1,217 @@
+import json
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from app.chat_types import SourceMatch
+from app.kb_manifest import (
+    kb_root_path,
+    load_manifest_entries,
+    resolve_manifest_reference,
+    source_match_payload,
+)
+
+
+def _write_manifest(kb_dir: Path, rows: list[dict]) -> None:
+    generated = kb_dir / "generated"
+    generated.mkdir(parents=True, exist_ok=True)
+    lines = [json.dumps(row) for row in rows]
+    (generated / "corpus_manifest.jsonl").write_text("\n".join(lines))
+
+
+class KbRootPathTestCase(unittest.TestCase):
+    def test_strips_kb_dir_prefix(self) -> None:
+        self.assertEqual(
+            kb_root_path("data/kb/raw/docs/peft/lora.md"),
+            "raw/docs/peft/lora.md",
+        )
+
+    def test_strips_leading_dot_slash(self) -> None:
+        self.assertEqual(
+            kb_root_path("./raw/docs/peft/lora.md"),
+            "raw/docs/peft/lora.md",
+        )
+
+    def test_keeps_root_relative_path(self) -> None:
+        self.assertEqual(
+            kb_root_path("raw/docs/peft/lora.md"),
+            "raw/docs/peft/lora.md",
+        )
+
+
+class ManifestPathResolutionTestCase(unittest.TestCase):
+    def test_manifest_match_carries_kb_root_path(self) -> None:
+        # The client maps inline `raw/...` citations to the resolved URL via
+        # this path, so it must be KB-root-relative regardless of how the
+        # manifest spells it.
+        with TemporaryDirectory() as tmp:
+            kb_dir = Path(tmp)
+            _write_manifest(
+                kb_dir,
+                [
+                    {
+                        "doc_id": "peft:lora",
+                        "title": "LoRA",
+                        "url": "https://example.com/lora",
+                        "source": "peft",
+                        "source_group": "docs",
+                        "path": "data/kb/raw/docs/peft/lora.md",
+                    }
+                ],
+            )
+
+            match = resolve_manifest_reference(
+                "raw/docs/peft/lora.md", kb_dir=str(kb_dir)
+            )
+
+            self.assertIsNotNone(match)
+            assert match is not None
+            self.assertEqual(match.url, "https://example.com/lora")
+            self.assertEqual(match.path, "raw/docs/peft/lora.md")
+
+            by_doc_scheme = resolve_manifest_reference(
+                "kb://doc/peft:lora", kb_dir=str(kb_dir)
+            )
+            self.assertIsNotNone(by_doc_scheme)
+            assert by_doc_scheme is not None
+            self.assertEqual(by_doc_scheme.path, "raw/docs/peft/lora.md")
+
+
+class AmbiguousTitleResolutionTestCase(unittest.TestCase):
+    def test_duplicate_title_does_not_resolve_to_arbitrary_doc(self) -> None:
+        # Two distinct docs share the title "Introduction"; a bare-label
+        # citation must not silently resolve to whichever was ingested last.
+        with TemporaryDirectory() as tmp:
+            kb_dir = Path(tmp)
+            _write_manifest(
+                kb_dir,
+                [
+                    {
+                        "doc_id": "peft:intro",
+                        "title": "Introduction",
+                        "url": "https://example.com/peft/intro",
+                        "source": "peft",
+                        "source_group": "docs",
+                        "path": "data/kb/raw/docs/peft/intro.md",
+                    },
+                    {
+                        "doc_id": "trl:intro",
+                        "title": "Introduction",
+                        "url": "https://example.com/trl/intro",
+                        "source": "trl",
+                        "source_group": "docs",
+                        "path": "data/kb/raw/docs/trl/intro.md",
+                    },
+                    {
+                        "doc_id": "peft:lora",
+                        "title": "LoRA",
+                        "url": "https://example.com/peft/lora",
+                        "source": "peft",
+                        "source_group": "docs",
+                        "path": "data/kb/raw/docs/peft/lora.md",
+                    },
+                ],
+            )
+
+            self.assertIsNone(
+                resolve_manifest_reference(
+                    "Introduction", label="Introduction", kb_dir=str(kb_dir)
+                )
+            )
+
+            # A unique title still resolves by label.
+            lora = resolve_manifest_reference("LoRA", label="LoRA", kb_dir=str(kb_dir))
+            self.assertIsNotNone(lora)
+            assert lora is not None
+            self.assertEqual(lora.doc_id, "peft:lora")
+
+            # Both same-title docs remain resolvable by their unambiguous keys.
+            by_url = resolve_manifest_reference(
+                "https://example.com/trl/intro", kb_dir=str(kb_dir)
+            )
+            self.assertIsNotNone(by_url)
+            assert by_url is not None
+            self.assertEqual(by_url.doc_id, "trl:intro")
+
+
+class ManifestLoaderTestCase(unittest.TestCase):
+    def test_missing_manifest_is_not_cached_empty(self) -> None:
+        # A lookup before the first-start bundle download must not pin an empty
+        # manifest for the process lifetime; once the file appears it loads.
+        with TemporaryDirectory() as tmp:
+            kb_dir = Path(tmp)
+            self.assertEqual(load_manifest_entries(str(kb_dir)), ())
+
+            _write_manifest(
+                kb_dir,
+                [
+                    {
+                        "doc_id": "peft:lora",
+                        "title": "LoRA",
+                        "url": "https://example.com/lora",
+                        "source": "peft",
+                        "source_group": "docs",
+                        "path": "data/kb/raw/docs/peft/lora.md",
+                    }
+                ],
+            )
+
+            entries = load_manifest_entries(str(kb_dir))
+            self.assertEqual([entry.doc_id for entry in entries], ["peft:lora"])
+
+    def test_malformed_line_is_skipped_not_fatal(self) -> None:
+        with TemporaryDirectory() as tmp:
+            kb_dir = Path(tmp)
+            generated = kb_dir / "generated"
+            generated.mkdir(parents=True, exist_ok=True)
+            good_a = json.dumps(
+                {
+                    "doc_id": "peft:lora",
+                    "title": "LoRA",
+                    "url": "https://example.com/lora",
+                    "source": "peft",
+                    "source_group": "docs",
+                    "path": "data/kb/raw/docs/peft/lora.md",
+                }
+            )
+            good_b = json.dumps(
+                {
+                    "doc_id": "trl:intro",
+                    "title": "Intro",
+                    "url": "https://example.com/intro",
+                    "source": "trl",
+                    "source_group": "docs",
+                    "path": "data/kb/raw/docs/trl/intro.md",
+                }
+            )
+            (generated / "corpus_manifest.jsonl").write_text(
+                f"{good_a}\n{{not valid json\n[1, 2, 3]\n{good_b}\n"
+            )
+
+            entries = load_manifest_entries(str(kb_dir))
+            self.assertEqual(
+                [entry.doc_id for entry in entries], ["peft:lora", "trl:intro"]
+            )
+
+
+class SourceMatchPayloadTestCase(unittest.TestCase):
+    def test_payload_includes_path(self) -> None:
+        match = SourceMatch(
+            doc_id="peft:lora",
+            title="LoRA",
+            url="https://example.com/lora",
+            source_key="peft",
+            source_label="PEFT Docs",
+            score=1.0,
+            group="docs",
+            path="raw/docs/peft/lora.md",
+        )
+
+        payload = source_match_payload(match, message_id="m1")
+
+        self.assertEqual(payload["path"], "raw/docs/peft/lora.md")
+        self.assertNotIn("call_id", payload)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_kb_shell.py

@@ -8,7 +8,12 @@ import pytest
 
 from data.scraping_scripts.build_kb_artifacts import build_kb_artifacts
 from data.scraping_scripts.update_kb_wiki import update_kb_wiki
-from app.kb_shell import KbCommandError, format_command_payload, run_kb_command
+from app.kb_shell import (
+    KbCommandError,
+    build_kb_command_argv,
+    format_command_payload,
+    run_kb_command,
+)
 
 
 def write_jsonl(path: Path, rows: list[dict]) -> None:
@@ -82,6 +87,20 @@ def test_run_kb_command_can_read_all_raw_sources(kb_dir: Path) -> None:
     assert "agentic_ai_engineering" in result.stdout
 
 
+def test_run_kb_command_caps_large_output_without_buffering_all(kb_dir: Path) -> None:
+    # A command that emits far more than the cap must be truncated to the cap;
+    # the streaming reader retains at most `output_limit` chars regardless of how
+    # much the child writes, so peak memory is bounded by the cap, not the file.
+    big = kb_dir / "raw" / "big.txt"
+    big.write_text("A" * 200_000)
+
+    result = run_kb_command("cat raw/big.txt", root=kb_dir, max_output_chars=1000)
+
+    assert result.truncated
+    assert len(result.stdout) <= 1000
+    assert "output truncated" in result.stdout
+
+
 def test_run_kb_command_rejects_shell_chaining(kb_dir: Path) -> None:
     with pytest.raises(KbCommandError):
         run_kb_command("rg LoraConfig | head", root=kb_dir)
@@ -92,6 +111,26 @@ def test_run_kb_command_rejects_path_traversal(kb_dir: Path) -> None:
         run_kb_command("cat ../outside.md", root=kb_dir)
 
 
+def test_rg_pattern_is_separated_by_double_dash(kb_dir: Path) -> None:
+    # A pattern beginning with `-`/`--` (e.g. ripgrep's `--pre`, which spawns an
+    # external preprocessor binary) must be forced into the positional pattern
+    # slot by a `--` separator so it can never be re-parsed by rg as a flag.
+    argv, _ = build_kb_command_argv("rg --pre=/bin/sh raw/docs/peft", root=kb_dir)
+    assert "--" in argv
+    sep = argv.index("--")
+    assert argv[sep + 1 :] == ["--pre=/bin/sh", "raw/docs/peft"]
+    # The dangerous token never precedes the separator (i.e. is never an option).
+    assert "--pre=/bin/sh" not in argv[:sep]
+
+
+def test_grep_pattern_is_separated_by_double_dash(kb_dir: Path) -> None:
+    argv, _ = build_kb_command_argv("grep --label=foo raw/docs/peft", root=kb_dir)
+    assert "--" in argv
+    sep = argv.index("--")
+    assert argv[sep + 1 :] == ["--label=foo", "raw/docs/peft"]
+    assert "--label=foo" not in argv[:sep]
+
+
 def test_run_kb_command_rejects_unbounded_broad_raw_search(kb_dir: Path) -> None:
     with pytest.raises(KbCommandError, match="requires -m"):
         run_kb_command("rg LoraConfig raw", root=kb_dir)
@@ -100,3 +139,19 @@ def test_run_kb_command_rejects_unbounded_broad_raw_search(kb_dir: Path) -> None
 
     assert result.exit_code == 0
     assert "LoraConfig" in result.stdout
+
+
+def test_grep_is_bounded_like_rg_and_does_not_follow_symlinks(kb_dir: Path) -> None:
+    # grep must reject an unbounded broad raw recursion, mirroring rg.
+    with pytest.raises(KbCommandError, match="requires -m"):
+        run_kb_command("grep LoraConfig raw", root=kb_dir)
+
+    # An explicit -m/--max-count satisfies the bound and works end-to-end.
+    argv, _ = build_kb_command_argv("grep -m 20 LoraConfig raw", root=kb_dir)
+    assert argv[:4] == ["grep", "-r", "-n", "--color=never"]
+    assert "-R" not in argv  # never the symlink-following recursion mode
+    assert "-m" in argv
+
+    result = run_kb_command("grep -m 20 LoraConfig raw/docs/peft", root=kb_dir)
+    assert result.exit_code == 0
+    assert "LoraConfig" in result.stdout

tests/test_prompts.py

@@ -48,6 +48,37 @@ def test_load_kb_agents_instructions_returns_empty_for_missing_path(
     assert load_kb_agents_instructions(tmp_path / "missing.md") == ""
 
 
+def test_build_system_prompt_without_local_tools_describes_none_of_them(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    # Explicit empty source selection: the prompt must not describe or
+    # instruct retrieval/KB tools the agent does not have this turn.
+    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))
+
+    prompt = build_system_prompt(
+        "google-genai:gemini-3.5-flash",
+        ("web_search",),
+        include_local_tools=False,
+    )
+
+    assert "retrieve_tutor_context" not in prompt
+    assert "run_kb_command" not in prompt
+    assert "## Local KB Instructions" not in prompt
+    assert "## Knowledge base disabled" in prompt
+    assert "google_search" in prompt
+
+    no_tools = build_system_prompt(
+        "google-genai:gemini-3.5-flash",
+        (),
+        include_local_tools=False,
+    )
+    assert "tools available" not in no_tools
+    assert "one tool available" not in no_tools
+
+
 def test_build_system_prompt_uses_explicit_instructions_without_file_read(
     tmp_path: Path,
     monkeypatch,