feat: QueryTranslator with domain-aware system prompt and category hint injection

app/llm/query_translator.py

@@ -11,7 +11,10 @@ from __future__ import annotations
 import json
 import logging
 from dataclasses import dataclass
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from app.platforms.ebay.categories import EbayCategoryCache
 
 log = logging.getLogger(__name__)
 
@@ -85,6 +88,80 @@ def _parse_response(raw: str) -> SearchParamsResponse:
     )
 
 
+# ── System prompt template ────────────────────────────────────────────────────
+
+_SYSTEM_PROMPT_TEMPLATE = """\
+You are a search assistant for Snipe, an eBay listing intelligence tool.
+Your job is to translate a natural-language description of what someone is looking for
+into a structured eBay search configuration.
+
+Return ONLY a JSON object with these exact fields — no preamble, no markdown, no extra keys:
+  base_query        (string)  Primary search term, short — e.g. "RTX 3080", "vintage Leica"
+  must_include_mode (string)  One of: "all" (AND), "any" (OR), "groups" (CNF: pipe=OR within group, comma=AND between groups)
+  must_include      (string)  Filter string per mode — leave blank if nothing to filter
+  must_exclude      (string)  Comma-separated terms to exclude — e.g. "mining,for parts,broken"
+  max_price         (number|null)  Maximum price in USD, or null
+  min_price         (number|null)  Minimum price in USD, or null
+  condition         (array)   Any of: "new", "used", "for_parts" — empty array means any condition
+  category_id       (string|null)  eBay category ID from the list below, or null if no match
+  explanation       (string)  One plain sentence summarizing what you built
+
+eBay "groups" mode syntax example: to find a GPU that is BOTH (nvidia OR amd) AND (16gb OR 8gb):
+  must_include_mode: "groups"
+  must_include: "nvidia|amd, 16gb|8gb"
+
+Phrase "like new", "open box", "refurbished" -> condition: ["used"]
+Phrase "broken", "for parts", "not working" -> condition: ["for_parts"]
+If unsure about condition, use an empty array.
+
+Available eBay categories (use category_id verbatim if one fits — otherwise omit):
+{category_hints}
+
+If none match, omit category_id (set to null). Respond with valid JSON only. No commentary outside the JSON object.
+"""
+
+
+# ── QueryTranslator ───────────────────────────────────────────────────────────
+
 class QueryTranslator:
-    """Stub — implemented in Task 6."""
+    """Translates natural-language search descriptions into SearchParamsResponse.
-    pass
+
+    Args:
+        category_cache: An EbayCategoryCache instance (may have empty cache).
+        llm_router: An LLMRouter instance from circuitforge_core.
+    """
+
+    def __init__(self, category_cache: "EbayCategoryCache", llm_router: object) -> None:
+        self._cache = category_cache
+        self._llm_router = llm_router
+
+    def translate(self, natural_language: str) -> SearchParamsResponse:
+        """Translate a natural-language query into a SearchParamsResponse.
+
+        Raises QueryTranslatorError if the LLM fails or returns bad JSON.
+        """
+        # Extract up to 10 keywords for category hint lookup
+        keywords = [w for w in natural_language.split()[:10] if len(w) > 2]
+        hints = self._cache.get_relevant(keywords, limit=30)
+        if not hints:
+            hints = self._cache.get_all_for_prompt(limit=40)
+
+        if hints:
+            category_hints = "\n".join(f"{cid}: {path}" for cid, path in hints)
+        else:
+            category_hints = "(no categories cached — omit category_id)"
+
+        system_prompt = _SYSTEM_PROMPT_TEMPLATE.format(category_hints=category_hints)
+
+        try:
+            raw = self._llm_router.complete(
+                natural_language,
+                system=system_prompt,
+                max_tokens=512,
+            )
+        except Exception as exc:
+            raise QueryTranslatorError(
+                f"LLM backend error: {exc}", raw=""
+            ) from exc
+
+        return _parse_response(raw)

tests/test_query_translator.py

@@ -2,11 +2,12 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from app.llm.query_translator import QueryTranslatorError, SearchParamsResponse, _parse_response
+from app.llm.query_translator import QueryTranslator, QueryTranslatorError, SearchParamsResponse, _parse_response
 
 
 # ── _parse_response ───────────────────────────────────────────────────────────
@@ -70,3 +71,100 @@ def test_parse_response_missing_required_field():
     })
     with pytest.raises(QueryTranslatorError):
         _parse_response(raw)
+
+
+# ── QueryTranslator (integration with mocked LLMRouter) ──────────────────────
+
+from app.platforms.ebay.categories import EbayCategoryCache
+from circuitforge_core.db import get_connection, run_migrations
+
+
+@pytest.fixture
+def db_with_categories(tmp_path):
+    conn = get_connection(tmp_path / "test.db")
+    run_migrations(conn, Path("app/db/migrations"))
+    cache = EbayCategoryCache(conn)
+    cache._seed_bootstrap()
+    return conn
+
+
+def _make_translator(db_conn, llm_response: str) -> QueryTranslator:
+    from app.platforms.ebay.categories import EbayCategoryCache
+    cache = EbayCategoryCache(db_conn)
+    mock_router = MagicMock()
+    mock_router.complete.return_value = llm_response
+    return QueryTranslator(category_cache=cache, llm_router=mock_router)
+
+
+def test_translate_returns_search_params(db_with_categories):
+    llm_out = json.dumps({
+        "base_query": "RTX 3080",
+        "must_include_mode": "groups",
+        "must_include": "rtx|geforce, 3080",
+        "must_exclude": "mining,for parts",
+        "max_price": 300.0,
+        "min_price": None,
+        "condition": ["used"],
+        "category_id": "27386",
+        "explanation": "Searching for used RTX 3080 GPUs under $300.",
+    })
+    t = _make_translator(db_with_categories, llm_out)
+    result = t.translate("used RTX 3080 under $300 no mining")
+    assert result.base_query == "RTX 3080"
+    assert result.max_price == 300.0
+
+
+def test_translate_injects_category_hints(db_with_categories):
+    """The system prompt sent to the LLM must contain category_id hints."""
+    llm_out = json.dumps({
+        "base_query": "GPU",
+        "must_include_mode": "all",
+        "must_include": "",
+        "must_exclude": "",
+        "max_price": None,
+        "min_price": None,
+        "condition": [],
+        "category_id": None,
+        "explanation": "Searching for GPUs.",
+    })
+    t = _make_translator(db_with_categories, llm_out)
+    t.translate("GPU")
+    call_args = t._llm_router.complete.call_args
+    system_prompt = call_args.kwargs.get("system") or call_args.args[1]
+    # Bootstrap seeds "27386" for Graphics Cards — should appear in the prompt
+    assert "27386" in system_prompt
+
+
+def test_translate_empty_category_cache_still_works(tmp_path):
+    """No crash when category cache is empty — prompt uses fallback text."""
+    from circuitforge_core.db import get_connection, run_migrations
+    conn = get_connection(tmp_path / "empty.db")
+    run_migrations(conn, Path("app/db/migrations"))
+    # Do NOT seed bootstrap — empty cache
+    llm_out = json.dumps({
+        "base_query": "vinyl",
+        "must_include_mode": "all",
+        "must_include": "",
+        "must_exclude": "",
+        "max_price": None,
+        "min_price": None,
+        "condition": [],
+        "category_id": None,
+        "explanation": "Searching for vinyl records.",
+    })
+    t = _make_translator(conn, llm_out)
+    result = t.translate("vinyl records")
+    assert result.base_query == "vinyl"
+    call_args = t._llm_router.complete.call_args
+    system_prompt = call_args.kwargs.get("system") or call_args.args[1]
+    assert "If none match" in system_prompt
+
+
+def test_translate_llm_error_raises_query_translator_error(db_with_categories):
+    from app.platforms.ebay.categories import EbayCategoryCache
+    cache = EbayCategoryCache(db_with_categories)
+    mock_router = MagicMock()
+    mock_router.complete.side_effect = RuntimeError("all backends exhausted")
+    t = QueryTranslator(category_cache=cache, llm_router=mock_router)
+    with pytest.raises(QueryTranslatorError, match="LLM backend"):
+        t.translate("used GPU")