| 1 | """Serena adapter - SSE :9878. |
| 2 | |
| 3 | Calls find_symbol. Authority tier: live-infra (live LSP, always fresh). |
| 4 | |
| 5 | NOTE: Serena requires an active project. Without project activation this will |
| 6 | return empty results. The adapter degrades gracefully; use the scope filter to |
| 7 | exclude serena when querying non-code questions. |
| 8 | """ |
| 9 | from __future__ import annotations |
| 10 | import os |
| 11 | |
| 12 | from models import Hit |
| 13 | from adapters.base import BaseAdapter |
| 14 | from adapters.mcp_base import call_mcp_tool |
| 15 | |
| 16 | SERENA_URL = os.environ.get("SERENA_URL", "http://127.0.0.1:9878/sse") |
| 17 | |
| 18 | |
| 19 | def _parse_serena(blocks: list[dict]) -> list[Hit]: |
| 20 | hits = [] |
| 21 | for block in blocks: |
| 22 | results = block.get("symbols") or block.get("results") or [] |
| 23 | if isinstance(results, list): |
| 24 | for r in results: |
| 25 | if not isinstance(r, dict): |
| 26 | continue |
| 27 | hits.append(Hit( |
| 28 | engine="serena", |
| 29 | title=r.get("name") or r.get("symbol", ""), |
| 30 | path=r.get("file") or r.get("path", ""), |
| 31 | snippet=r.get("signature") or r.get("snippet", "")[:300], |
| 32 | score=float(r.get("score") or 0.6), |
| 33 | last_seen=None, |
| 34 | confidence=0.9, |
| 35 | authority_tier="live-infra", |
| 36 | )) |
| 37 | elif "text" in block: |
| 38 | hits.append(Hit( |
| 39 | engine="serena", |
| 40 | title="symbol result", |
| 41 | snippet=str(block["text"])[:300], |
| 42 | score=0.6, |
| 43 | confidence=0.9, |
| 44 | authority_tier="live-infra", |
| 45 | )) |
| 46 | return hits |
| 47 | |
| 48 | |
| 49 | class SerenaAdapter(BaseAdapter): |
| 50 | engine = "serena" |
| 51 | authority_tier = "live-infra" |
| 52 | |
| 53 | async def _fetch(self, query: str, k: int) -> list[Hit]: |
| 54 | blocks = await call_mcp_tool( |
| 55 | SERENA_URL, "find_symbol", {"query": query, "limit": k} |
| 56 | ) |
| 57 | return _parse_serena(blocks)[:k] |