| 1 | """cbm-homelab adapter - SSE :9877. |
| 2 | |
| 3 | Calls search_graph across indexed projects. Authority tier: recent-file. |
| 4 | |
| 5 | CBM's search_graph requires a project parameter. We discover the project list |
| 6 | by calling list_projects first (cached after first success), then fan out |
| 7 | search_graph calls to the most relevant projects in parallel. |
| 8 | |
| 9 | If list_projects is unavailable we fall back to the project list embedded |
| 10 | in the "project not found" error that search_graph returns without a project arg. |
| 11 | """ |
| 12 | from __future__ import annotations |
| 13 | import asyncio |
| 14 | import json |
| 15 | import os |
| 16 | |
| 17 | from models import Hit |
| 18 | from adapters.base import BaseAdapter |
| 19 | from adapters.mcp_base import call_mcp_tool |
| 20 | |
| 21 | CBM_URL = os.environ.get("CBM_URL", "http://REDACTED-IP:9877/sse") |
| 22 | |
| 23 | _PROJECT_CACHE: list[str] | None = None |
| 24 | |
| 25 | MAX_PROJECTS = 6 |
| 26 | |
| 27 | |
| 28 | async def _get_projects() -> list[str]: |
| 29 | """Return list of available CBM project names, cached.""" |
| 30 | global _PROJECT_CACHE |
| 31 | if _PROJECT_CACHE is not None: |
| 32 | return _PROJECT_CACHE |
| 33 | |
| 34 | blocks = await call_mcp_tool(CBM_URL, "list_projects", {}, timeout=3.5) |
| 35 | for block in blocks: |
| 36 | projects = block.get("projects") if isinstance(block, dict) else None |
| 37 | if isinstance(projects, list): |
| 38 | _PROJECT_CACHE = [str(p) for p in projects] |
| 39 | return _PROJECT_CACHE |
| 40 | |
| 41 | err_blocks = await call_mcp_tool(CBM_URL, "search_graph", {"query": "status", "limit": 1}, timeout=3.5) |
| 42 | for block in err_blocks: |
| 43 | raw = block.get("text", "") if isinstance(block, dict) else "" |
| 44 | try: |
| 45 | parsed = json.loads(raw) |
| 46 | avail = parsed.get("available_projects") or [] |
| 47 | if avail: |
| 48 | _PROJECT_CACHE = [str(p) for p in avail] |
| 49 | return _PROJECT_CACHE |
| 50 | except Exception: |
| 51 | pass |
| 52 | |
| 53 | _PROJECT_CACHE = [] |
| 54 | return _PROJECT_CACHE |
| 55 | |
| 56 | |
| 57 | def _rank_projects(projects: list[str], query: str) -> list[str]: |
| 58 | """Pick the most query-relevant projects to search (heuristic).""" |
| 59 | query_lower = query.lower() |
| 60 | query_words = set(query_lower.split()) |
| 61 | |
| 62 | def relevance(proj: str) -> int: |
| 63 | p = proj.lower() |
| 64 | score = 0 |
| 65 | for word in query_words: |
| 66 | if word in p: |
| 67 | score += 3 |
| 68 | if any(k in p for k in ("hermes", "treetrace", "jobforge", "oversight", |
| 69 | "darwin", "personal", "homelab", "portfolio")): |
| 70 | score += 2 |
| 71 | if any(k in p for k in ("archive", "bounty", "staging", "triage", |
| 72 | "mirror", "cache", "plugins")): |
| 73 | score -= 1 |
| 74 | return score |
| 75 | |
| 76 | ranked = sorted(projects, key=relevance, reverse=True) |
| 77 | return ranked[:MAX_PROJECTS] |
| 78 | |
| 79 | |
| 80 | def _parse_cbm_block(block: dict) -> list[Hit]: |
| 81 | """Parse one content block from search_graph into Hit objects.""" |
| 82 | results = block.get("results") or block.get("nodes") or [] |
| 83 | hits = [] |
| 84 | if isinstance(results, list): |
| 85 | for r in results: |
| 86 | if not isinstance(r, dict): |
| 87 | continue |
| 88 | name = r.get("name") or r.get("title") or r.get("qualified_name", "") |
| 89 | if not name: |
| 90 | continue |
| 91 | file_path = r.get("file") or r.get("path") or r.get("file_path", "") |
| 92 | snippet = (r.get("summary") or r.get("snippet") or |
| 93 | r.get("signature") or "")[:300] |
| 94 | raw = r.get("score") or r.get("relevance") or r.get("rank") or 0 |
| 95 | try: |
| 96 | score = min(1.0, max(0.0, -float(raw) / 10.0)) if float(raw) < 0 else float(raw) |
| 97 | except Exception: |
| 98 | score = 0.5 |
| 99 | hits.append(Hit( |
| 100 | engine="cbm", |
| 101 | title=str(name), |
| 102 | path=str(file_path), |
| 103 | snippet=str(snippet), |
| 104 | score=score, |
| 105 | last_seen=None, |
| 106 | confidence=0.8, |
| 107 | authority_tier="recent-file", |
| 108 | )) |
| 109 | elif isinstance(block, dict) and "text" in block: |
| 110 | text = str(block["text"]) |
| 111 | if '"error"' not in text: |
| 112 | hits.append(Hit( |
| 113 | engine="cbm", |
| 114 | title="code graph result", |
| 115 | snippet=text[:300], |
| 116 | score=0.5, |
| 117 | confidence=0.7, |
| 118 | authority_tier="recent-file", |
| 119 | )) |
| 120 | return hits |
| 121 | |
| 122 | |
| 123 | class CbmAdapter(BaseAdapter): |
| 124 | engine = "cbm" |
| 125 | authority_tier = "recent-file" |
| 126 | |
| 127 | async def _fetch(self, query: str, k: int) -> list[Hit]: |
| 128 | projects = await _get_projects() |
| 129 | if not projects: |
| 130 | return [] |
| 131 | |
| 132 | selected = _rank_projects(projects, query) |
| 133 | if not selected: |
| 134 | return [] |
| 135 | |
| 136 | tasks = [ |
| 137 | asyncio.create_task( |
| 138 | call_mcp_tool(CBM_URL, "search_graph", |
| 139 | {"project": proj, "query": query, "limit": k}, |
| 140 | timeout=3.0) |
| 141 | ) |
| 142 | for proj in selected |
| 143 | ] |
| 144 | all_blocks_list = await asyncio.gather(*tasks, return_exceptions=True) |
| 145 | |
| 146 | all_hits: list[Hit] = [] |
| 147 | seen: set[str] = set() |
| 148 | for blocks in all_blocks_list: |
| 149 | if isinstance(blocks, Exception) or not blocks: |
| 150 | continue |
| 151 | for block in blocks: |
| 152 | for hit in _parse_cbm_block(block): |
| 153 | key = hit.path or hit.title |
| 154 | if key not in seen: |
| 155 | seen.add(key) |
| 156 | all_hits.append(hit) |
| 157 | |
| 158 | all_hits.sort(key=lambda h: h.score, reverse=True) |
| 159 | return all_hits[:k] |