Optimize hook injection fallback and filtering

hooks/brainlayer-prompt-search.py

@@ -40,6 +40,11 @@
 ENTITY_TOKEN_RE = re.compile(r"[A-Za-z0-9]+(?:-[A-Za-z0-9]+)*|[\u0590-\u05FF]+")
 _ENTITY_CACHE = None
 _ENTITY_CACHE_DB_PATH = None
+LOW_CONFIDENCE_FALLBACK_THRESHOLD = 0.30
+LOW_CONFIDENCE_FALLBACK_MESSAGE = (
+    "No high-confidence memories found. Use brain_search() for deeper retrieval."
+)
+_KG_ENTITY_CHUNKS_RELATION_TYPE_CACHE = {}
 
 
 def get_session_context(conn, session_id: str, limit: int = 3) -> list[str]:
@@ -400,11 +405,9 @@ def truncate(text, max_chars=80):
     if len(text) <= max_chars:
         return text
     candidate = text[:max_chars]
-    search_start = max(0, max_chars - 40)
-    for sep in (". ", "! ", "? ", "| "):
-        idx = candidate.rfind(sep, search_start)
-        if idx > 0:
-            return candidate[: idx + len(sep) - 1] + "..."
+    sentence_matches = list(re.finditer(r"[.!?](?=(?:\s|$))", candidate))
+    if sentence_matches:
+        return candidate[: sentence_matches[-1].end()] + "..."
     return candidate.rsplit(" ", 1)[0] + "..."
 
 
@@ -428,6 +431,22 @@ def _get_connection_cache_key(conn):
     return f"conn:{id(conn)}"
 
 
+def _kg_entity_chunks_has_relation_type(conn):
+    cache_key = _get_connection_cache_key(conn)
+    cached = _KG_ENTITY_CHUNKS_RELATION_TYPE_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+
+    try:
+        columns = {row[1] for row in conn.execute("PRAGMA table_info(kg_entity_chunks)").fetchall()}
+    except sqlite3.Error:
+        columns = set()
+
+    has_column = "relation_type" in columns
+    _KG_ENTITY_CHUNKS_RELATION_TYPE_CACHE[cache_key] = has_column
+    return has_column
+
+
 def _load_entity_cache(conn=None):
     global _ENTITY_CACHE, _ENTITY_CACHE_DB_PATH
 
@@ -621,12 +640,17 @@ def detect_entities_in_prompt(prompt, conn=None):
 def get_entity_chunks(entity_id, conn, limit=3):
     """Get top linked chunk summaries for an entity."""
     try:
+        relation_filter = ""
+        if _kg_entity_chunks_has_relation_type(conn):
+            relation_filter = "AND COALESCE(ec.relation_type, '') != 'co_occurs_with'"
+
         rows = conn.execute(
-            """
+            f"""
             SELECT c.content, c.created_at, c.project
             FROM kg_entity_chunks ec
             JOIN chunks c ON c.id = ec.chunk_id
             WHERE ec.entity_id = ?
+              {relation_filter}
               AND COALESCE(c.project, '') != 'eval-sandbox'
               AND COALESCE(c.tags, '') NOT LIKE '%"eval-test"%'
             ORDER BY ec.relevance DESC
@@ -710,6 +734,24 @@ def select_adaptive_injection_rows(rows, entity_count=0, light_mode=False):
     return strategic_reorder(selected[:MAX_ADAPTIVE_INJECTION])
 
 
+def build_low_confidence_fallback(rows):
+    if not rows:
+        return LOW_CONFIDENCE_FALLBACK_MESSAGE
+
+    top_row = rows[0]
+    if isinstance(top_row, dict):
+        relevance = top_row.get("relevance")
+        if relevance is None:
+            relevance = top_row.get("rrf_score")
+    else:
+        relevance = None
+
+    if relevance is not None and relevance < LOW_CONFIDENCE_FALLBACK_THRESHOLD:
+        return LOW_CONFIDENCE_FALLBACK_MESSAGE
+
+    return None
+
+
 def _ensure_src_on_syspath():
     src_path = Path(__file__).resolve().parents[1] / "src"
     if src_path.exists():
@@ -1014,6 +1056,7 @@ def main():
     new_chunk_ids = []
     new_briefs = []
     entities_detected = 0
+    fallback_rows = []
 
     def finalize_and_exit(*, mode=None):
         final_mode = mode or telemetry_mode
@@ -1172,6 +1215,18 @@ def finalize_and_exit(*, mode=None):
                     if len(filtered_rows) >= base_limit:
                         break
 
+            fallback_rows = [
+                {
+                    "id": chunk_id,
+                    "content": content,
+                    "importance": importance,
+                    "project": project,
+                    "tags": tags,
+                    "created_at": created_at,
+                    "rrf_score": 0.0,
+                }
+                for chunk_id, content, importance, project, tags, created_at in filtered_rows
+            ]
             new_chunk_ids, new_briefs = inject_search_results(lines, filtered_rows, deep, label=label)
     except sqlite3.Error:
         pass
@@ -1186,6 +1241,11 @@ def finalize_and_exit(*, mode=None):
             "Run brain_search() to verify before stating any personal details (hardware, history, names)."
         )
 
+    if not lines:
+        fallback = build_low_confidence_fallback(fallback_rows)
+        if fallback:
+            lines.append(fallback)
+
     if lines:
         print("\n".join(lines))
 

tests/test_adaptive_injection.py

@@ -154,3 +154,22 @@ def test_fallback_to_fts_only(self, prompt_search, monkeypatch):
 
         assert used_hybrid is False
         assert [row["id"] for row in rows] == ["fts-best"]
+
+    def test_no_results_falls_back_to_low_confidence_message(self, prompt_search):
+        message = prompt_search.build_low_confidence_fallback([])
+
+        assert message == "No high-confidence memories found. Use brain_search() for deeper retrieval."
+
+    def test_low_relevance_rows_emit_fallback_message(self, prompt_search):
+        rows = [_row("low", 0.29)]
+
+        message = prompt_search.build_low_confidence_fallback(rows)
+
+        assert message == "No high-confidence memories found. Use brain_search() for deeper retrieval."
+
+    def test_high_relevance_rows_do_not_emit_fallback_message(self, prompt_search):
+        rows = [_row("high", 0.30)]
+
+        message = prompt_search.build_low_confidence_fallback(rows)
+
+        assert message is None

tests/test_hook_slim.py

@@ -55,6 +55,16 @@ def test_truncate_prefers_sentence_boundary(self, hook):
         assert result.endswith("...")
         assert len(result) <= 83
 
+    def test_truncate_cuts_at_last_sentence_end_before_limit(self, hook):
+        text = (
+            "Short sentence. Another complete sentence. "
+            "This final sentence extends well beyond the truncation limit and should be omitted."
+        )
+
+        result = hook.truncate(text, max_chars=55)
+
+        assert result == "Short sentence. Another complete sentence...."
+
 
 class TestResultCountCap:
     """Result count must be capped at MAX_ADAPTIVE_INJECTION (3)."""

tests/test_prompt_classification.py

@@ -35,7 +35,16 @@ def make_hook_db(db_path: Path) -> None:
     )
     conn.execute("CREATE VIRTUAL TABLE chunks_fts USING fts5(chunk_id UNINDEXED, content)")
     conn.execute("CREATE TABLE kg_entities (id TEXT PRIMARY KEY, name TEXT, entity_type TEXT)")
-    conn.execute("CREATE TABLE kg_entity_chunks (entity_id TEXT, chunk_id TEXT, relevance REAL)")
+    conn.execute(
+        """
+        CREATE TABLE kg_entity_chunks (
+            entity_id TEXT,
+            chunk_id TEXT,
+            relevance REAL,
+            relation_type TEXT
+        )
+        """
+    )
 
     conn.execute(
         """
@@ -79,8 +88,26 @@ def make_hook_db(db_path: Path) -> None:
         ("person-theo", "Theo Browne", "person"),
     )
     conn.execute(
-        "INSERT INTO kg_entity_chunks (entity_id, chunk_id, relevance) VALUES (?, ?, ?)",
-        ("person-theo", "chunk-theo", 0.9),
+        "INSERT INTO kg_entity_chunks (entity_id, chunk_id, relevance, relation_type) VALUES (?, ?, ?, ?)",
+        ("person-theo", "chunk-theo", 0.9, "mentioned_in"),
+    )
+    conn.execute(
+        """
+        INSERT INTO chunks (id, content, importance, project, tags, created_at)
+        VALUES (?, ?, ?, ?, ?, ?)
+        """,
+        (
+            "chunk-theo-noise",
+            "Theo Browne appears in low-signal co-occurrence noise that should stay hidden.",
+            5,
+            "brainlayer",
+            '["people"]',
+            "2026-04-03T10:00:00Z",
+        ),
+    )
+    conn.execute(
+        "INSERT INTO kg_entity_chunks (entity_id, chunk_id, relevance, relation_type) VALUES (?, ?, ?, ?)",
+        ("person-theo", "chunk-theo-noise", 0.99, "co_occurs_with"),
     )
     conn.commit()
     conn.close()
@@ -137,6 +164,7 @@ def test_classify_long_prompt_not_casual():
 
 def test_command_skips_retrieval(monkeypatch, capsys):
     module = load_prompt_search_module()
+    monkeypatch.setattr(module, "get_db_path", lambda: None)
 
     def fail_connect(*args, **kwargs):
         raise AssertionError("sqlite3.connect should not be called for command prompts")
@@ -148,6 +176,7 @@ def fail_connect(*args, **kwargs):
 
 def test_casual_skips_retrieval(monkeypatch, capsys):
     module = load_prompt_search_module()
+    monkeypatch.setattr(module, "get_db_path", lambda: None)
 
     def fail_connect(*args, **kwargs):
         raise AssertionError("sqlite3.connect should not be called for casual prompts")
@@ -167,6 +196,7 @@ def test_entity_route_injects_card(tmp_path, monkeypatch, capsys):
 
     assert "[Entity: Theo Browne" in output
     assert "Theo Browne is linked to BrainLayer collaboration notes." in output
+    assert "co-occurrence noise" not in output
 
 
 def test_knowledge_route_injects_chunks(tmp_path, monkeypatch, capsys):