diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index 7c431f7..67b01e4 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -253,6 +253,16 @@ def parse_args() -> argparse.Namespace: context = sub.add_parser("context", help="Build an agent-ready KB context bundle for a question.") context.add_argument("query", nargs="+") + context.add_argument( + "--retrieval-query", + help="Optional read-only semantic query; operational contracts remain bound to the exact operator query.", + ) + context.add_argument( + "--retrieval-anchor", + action="append", + default=[], + help="Bounded topic anchor used only to rank read-only claim retrieval; may be repeated.", + ) context.add_argument("--limit", type=int, default=4) context.add_argument("--context-limit", type=int, default=6) add_output_flag(context) @@ -1951,13 +1961,24 @@ def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str: return f"[`{text}`]({claim_url(claim_id)})" -def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[str, Any]]: +def find_claims( + args: argparse.Namespace, + query: str, + limit: int, + retrieval_anchors: list[str] | None = None, +) -> list[dict[str, Any]]: patterns = [f"%{term}%" for term in query_terms(query)] + anchor_terms = query_terms(" ".join(retrieval_anchors or [])) if retrieval_anchors else [] + anchor_patterns = [f"%{term}%" for term in anchor_terms] min_score = 2 if len(patterns) >= 3 else 1 + anchor_limit = max(1, min(2, limit)) sql = f""" with params(patterns) as ( values ({sql_array(patterns)}) ), + anchors(patterns) as ( + values ({sql_array(anchor_patterns)}) + ), ranked as ( select c.id, c.type, @@ -1974,6 +1995,15 @@ def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[s where lower(tag) like pattern ) ) as score, + ( + select count(*) + from anchors a, unnest(a.patterns) pattern + where lower(c.text) like pattern + or exists ( + select 1 from unnest(coalesce(c.tags, '{{}}'::text[])) tag + where lower(tag) like pattern + ) + ) as anchor_score, (select count(*) from claim_evidence ce where ce.claim_id = c.id) as evidence_count, (select count(*) from claim_edges e where e.from_claim = c.id or e.to_claim = c.id) as edge_count from claims c @@ -1983,6 +2013,29 @@ def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[s select ranked.*, max(score) over () as max_score from ranked + ), + raw_selected as ( + select eligible.* + from eligible + where score >= case when max_score >= {min_score} then {min_score} else 1 end + order by score desc, evidence_count desc, edge_count desc, + coalesce(confidence, 0) desc, length(text), text, id + limit {limit} + ), + anchor_selected as ( + select eligible.* + from eligible + where anchor_score >= 1 + order by anchor_score desc, score desc, evidence_count desc, edge_count desc, + coalesce(confidence, 0) desc, length(text), text, id + limit {anchor_limit} + ), + lanes as ( + select raw_selected.*, 'raw'::text as selection_lane + from raw_selected + union all + select anchor_selected.*, 'anchor'::text as selection_lane + from anchor_selected ) select jsonb_build_object( 'id', id::text, @@ -1991,16 +2044,31 @@ def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[s 'confidence', confidence, 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), 'score', score, + 'anchor_score', anchor_score, 'evidence_count', evidence_count, - 'edge_count', edge_count + 'edge_count', edge_count, + '_selection_lane', selection_lane )::text - from eligible - where score >= 1 - and score >= case when max_score >= {min_score} then {min_score} else 1 end - order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text), text, id - limit {limit}; + from lanes + order by case selection_lane when 'raw' then 0 else 1 end, + score desc, anchor_score desc, evidence_count desc, edge_count desc, + coalesce(confidence, 0) desc, length(text), text, id; """ - return psql_json(args, sql) + rows = psql_json(args, sql) + raw_rows: list[dict[str, Any]] = [] + anchor_rows: list[dict[str, Any]] = [] + for row in rows: + lane = row.pop("_selection_lane", None) + if lane == "raw": + raw_rows.append(row) + elif lane == "anchor": + anchor_rows.append(row) + if not anchor_patterns or any(int(row.get("anchor_score") or 0) > 0 for row in raw_rows): + return raw_rows[:limit] + + raw_ids = {str(row.get("id")) for row in raw_rows} + backfill = [row for row in anchor_rows if str(row.get("id")) not in raw_ids][:anchor_limit] + return backfill + raw_rows[: max(0, limit - len(backfill))] def find_context_rows(args: argparse.Namespace, query: str, limit: int) -> list[dict[str, Any]]: @@ -3339,6 +3407,8 @@ def build_retrieval_receipt( return { "schema": RETRIEVAL_RECEIPT_SCHEMA, "query_sha256": hashlib.sha256(str(data.get("query") or "").encode("utf-8")).hexdigest(), + "retrieval_query_sha256": data.get("retrieval_query_sha256"), + "retrieval_anchors_sha256": data.get("retrieval_anchors_sha256"), "semantic_context_sha256": canonical_json_sha256(semantic_context), "artifact_state_sha256": canonical_json_sha256(artifact_states), "claim_ids": [str(claim.get("id")) for claim in data.get("claims", []) if claim.get("id")], @@ -3362,16 +3432,26 @@ def build_retrieval_receipt( } -def _bundle_once(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]: +def _bundle_once( + args: argparse.Namespace, + query: str, + claim_limit: int, + context_limit: int, + retrieval_query: str | None = None, + retrieval_anchors: list[str] | None = None, +) -> dict[str, Any]: + semantic_query = retrieval_query or query contracts = context_operational_contracts(args, query) - claims = find_claims(args, query, claim_limit) + claims = find_claims(args, semantic_query, claim_limit, retrieval_anchors) claim_ids = [claim["id"] for claim in claims] evidence = load_evidence(args, claim_ids, 4) edges = load_edges(args, claim_ids, 6) bind_claim_evidence_challenge(contracts, claims, evidence) return { "query": query, - "terms": query_terms(query), + "retrieval_query_sha256": hashlib.sha256(semantic_query.encode("utf-8")).hexdigest(), + "retrieval_anchors_sha256": canonical_json_sha256(retrieval_anchors or []), + "terms": query_terms(semantic_query), "operational_contracts": contracts, "compiled_response": compile_operational_response(contracts), "claims": [ @@ -3383,11 +3463,18 @@ def _bundle_once(args: argparse.Namespace, query: str, claim_limit: int, context } for claim in claims ], - "context_rows": find_context_rows(args, query, context_limit), + "context_rows": find_context_rows(args, semantic_query, context_limit), } -def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]: +def bundle( + args: argparse.Namespace, + query: str, + claim_limit: int, + context_limit: int, + retrieval_query: str | None = None, + retrieval_anchors: list[str] | None = None, +) -> dict[str, Any]: try: max_attempts = max(1, min(int(os.environ.get("TELEO_KB_READ_ATTEMPTS", "3")), 5)) except ValueError as exc: @@ -3396,7 +3483,7 @@ def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit last_markers: tuple[dict[str, Any], dict[str, Any]] | None = None for attempt in range(1, max_attempts + 1): before = database_read_marker(args) - data = _bundle_once(args, query, claim_limit, context_limit) + data = _bundle_once(args, query, claim_limit, context_limit, retrieval_query, retrieval_anchors) after = database_read_marker(args) last_markers = (before, after) consistency_status = "stable_wal_marker" if _read_marker_stable(before, after) else "wal_changed_during_read" @@ -3607,7 +3694,14 @@ def main() -> int: if args.command == "status": data = status(args) elif args.command in {"search", "context"}: - data = bundle(args, args.query, args.limit, args.context_limit) + data = bundle( + args, + args.query, + args.limit, + args.context_limit, + getattr(args, "retrieval_query", None), + getattr(args, "retrieval_anchor", None), + ) elif args.command == "show": claim = get_claim(args, args.claim_id) if not claim: diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py index 95132b3..eae9191 100644 --- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py +++ b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py @@ -9,7 +9,7 @@ import re import subprocess import sys import threading -from collections import OrderedDict +from collections import Counter, OrderedDict from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path @@ -23,6 +23,10 @@ CONTEXT_ROW_LIMIT = 4 MAX_CLAIM_TEXT_CHARS = 1_000 MAX_CONTEXT_BODY_CHARS = 800 MAX_EVIDENCE_EXCERPT_CHARS = 600 +MAX_RETRIEVAL_ANCHORS = 8 +MAX_RETRIEVAL_SUBJECT_ANCHORS = 2 +MAX_RETRIEVAL_FALLBACK_ANCHORS = 3 +RETRIEVAL_HISTORY_USER_TURNS = 6 DATABASE_REASONING_PROTOCOL_VERSION = "leo-db-reasoning-v2" WORD_RE = re.compile(r"\b\w+(?:[-']\w+)*\b") LIST_PREFIX_RE = re.compile(r"^(\s*(?:[-*]|\d+[.)])\s+)(.*)$") @@ -31,6 +35,120 @@ REQUESTED_SUBJECT_RE = re.compile( r"name the subject exactly once as\s+`(?P[^`\r\n]{1,120})`\s+in your answer", re.I, ) +RETRIEVAL_TOKEN_RE = re.compile(r"[A-Za-z][A-Za-z0-9.+#'-]{2,}") +RETRIEVAL_ANCHOR_STOPWORDS = frozenset( + { + "about", + "across", + "after", + "all", + "also", + "and", + "answer", + "actually", + "any", + "anything", + "are", + "argue", + "around", + "before", + "being", + "been", + "better", + "both", + "but", + "can", + "could", + "did", + "does", + "each", + "for", + "from", + "had", + "has", + "have", + "how", + "into", + "less", + "more", + "not", + "now", + "only", + "our", + "over", + "same", + "say", + "should", + "still", + "than", + "that", + "the", + "their", + "them", + "then", + "they", + "this", + "through", + "use", + "was", + "were", + "what", + "when", + "where", + "which", + "who", + "why", + "will", + "with", + "without", + "would", + "you", + "your", + "asserting", + "based", + "body", + "candidate", + "canonical", + "challenge", + "claim", + "concrete", + "confidence", + "context", + "database", + "direct", + "distinguish", + "discussion", + "evidence", + "exact", + "explain", + "explicit", + "existing", + "falsifier", + "give", + "identify", + "initial", + "inference", + "minimum", + "missing", + "most", + "nothing", + "opposite", + "outcome", + "requirements", + "return", + "revised", + "reviewable", + "source", + "stage", + "state", + "strongest", + "support", + "test", + "tool", + "while", + "write", + } +) COMPILED_CONTRACT_IDS = frozenset( { "mixed_packet_composition", @@ -96,6 +214,8 @@ def _retrieval_receipt_trace( safe = { "schema": value.get("schema"), "query_sha256": value.get("query_sha256"), + "retrieval_query_sha256": value.get("retrieval_query_sha256"), + "retrieval_anchors_sha256": value.get("retrieval_anchors_sha256"), "semantic_context_sha256": value.get("semantic_context_sha256"), "artifact_state_sha256": value.get("artifact_state_sha256"), "claim_ids": sorted(str(item) for item in value.get("claim_ids") or []), @@ -134,6 +254,100 @@ def _failure_context(reason: str) -> str: ) +def _history_message_text(message: Any) -> str: + if not isinstance(message, dict) or message.get("role") != "user": + return "" + content = message.get("content") + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + parts: list[str] = [] + for item in content: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict) and isinstance(item.get("text"), str): + parts.append(str(item["text"])) + return " ".join(parts) + + +def _is_subject_token(token: str) -> bool: + return token.isupper() or any(char.isupper() for char in token[1:]) + + +def _contextual_retrieval_query(user_message: str, conversation_history: Any) -> tuple[str, list[str]]: + """Return the exact query plus bounded user-only topic anchors for separate ranking.""" + + query = str(user_message or "")[:MAX_QUERY_CHARS] + if not isinstance(conversation_history, list): + return query, [] + + prior_user_turns: list[str] = [] + skipped_current = False + for message in reversed(conversation_history): + text = _history_message_text(message).strip() + if not text: + continue + if not skipped_current and text == query.strip(): + skipped_current = True + continue + prior_user_turns.append(text) + if len(prior_user_turns) >= RETRIEVAL_HISTORY_USER_TURNS: + break + + counts: Counter[str] = Counter() + recency: dict[str, int] = {} + surfaces: dict[str, str] = {} + subject_tokens: set[str] = set() + for turn_index, text in enumerate(prior_user_turns): + seen_in_turn: set[str] = set() + for token in RETRIEVAL_TOKEN_RE.findall(text): + surface = token.strip("-'.+") + if surface.casefold().endswith("'s"): + surface = surface[:-2] + normalized = surface.casefold() + if len(normalized) < 3 or normalized in RETRIEVAL_ANCHOR_STOPWORDS or normalized in seen_in_turn: + continue + seen_in_turn.add(normalized) + counts[normalized] += 1 + recency.setdefault(normalized, RETRIEVAL_HISTORY_USER_TURNS - turn_index) + surfaces.setdefault(normalized, surface) + if _is_subject_token(surface): + subject_tokens.add(normalized) + + current_terms: set[str] = set() + for token in RETRIEVAL_TOKEN_RE.findall(query): + surface = token.strip("-'.+") + if surface.casefold().endswith("'s"): + surface = surface[:-2] + normalized = surface.casefold() + if len(normalized) < 3 or normalized in RETRIEVAL_ANCHOR_STOPWORDS or normalized in current_terms: + continue + current_terms.add(normalized) + counts[normalized] += 1 + recency[normalized] = RETRIEVAL_HISTORY_USER_TURNS + 1 + surfaces[normalized] = surface + if _is_subject_token(surface): + subject_tokens.add(normalized) + + subject_anchors = sorted( + subject_tokens, + key=lambda term: ( + -counts[term], + -recency[term], + term, + ), + ) + if subject_anchors: + selected = subject_anchors[:MAX_RETRIEVAL_SUBJECT_ANCHORS] + else: + repeated = [term for term, count in counts.items() if count >= 2] + repeated.sort(key=lambda term: (-counts[term], -recency[term], term)) + selected = repeated[:MAX_RETRIEVAL_FALLBACK_ANCHORS] + anchors = [surfaces[term] for term in selected[:MAX_RETRIEVAL_ANCHORS]] + return query, anchors + + def _requires_database_truth(query: str) -> bool: lowered = query.lower() named_packet_question = "helmer" in lowered and any( @@ -345,12 +559,14 @@ def _take_snapshot(session_id: str, query_sha256: str) -> dict[str, Any] | None: def _load_database_snapshot( user_message: str, *, + conversation_history: Any = None, hermes_home: Path | None = None, runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, ) -> dict[str, Any]: """Load one immutable DB-contract snapshot for generation and validation.""" query = str(user_message or "")[:MAX_QUERY_CHARS] + retrieval_query, retrieval_anchors = _contextual_retrieval_query(query, conversation_history) home = hermes_home or Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes"))) kb_tool = home / "bin" / "kb_tool.py" query_sha256 = hashlib.sha256(query.encode("utf-8")).hexdigest() @@ -358,6 +574,11 @@ def _load_database_snapshot( "generated_at_utc": datetime.now(timezone.utc).isoformat(), "event": "pre_llm_call", "query_sha256": query_sha256, + "retrieval_query_sha256": hashlib.sha256(retrieval_query.encode("utf-8")).hexdigest(), + "retrieval_anchor_count": len(retrieval_anchors), + "retrieval_anchors_sha256": hashlib.sha256( + json.dumps(retrieval_anchors, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest(), "source": "kb_tool.py --local context", } @@ -373,13 +594,21 @@ def _load_database_snapshot( "--local", "context", query, - "--limit", - str(CONTEXT_CLAIM_LIMIT), - "--context-limit", - str(CONTEXT_ROW_LIMIT), - "--format", - "json", + "--retrieval-query", + retrieval_query, ] + for anchor in retrieval_anchors: + command.extend(("--retrieval-anchor", anchor)) + command.extend( + ( + "--limit", + str(CONTEXT_CLAIM_LIMIT), + "--context-limit", + str(CONTEXT_ROW_LIMIT), + "--format", + "json", + ) + ) try: completed = runner(command, capture_output=True, text=True, timeout=timeout, check=False) except subprocess.TimeoutExpired: @@ -454,12 +683,20 @@ def _load_database_snapshot( def build_database_context( user_message: str, *, + conversation_history: Any = None, hermes_home: Path | None = None, runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, ) -> str: """Return current operational contracts for one exact operator question.""" - return str(_load_database_snapshot(user_message, hermes_home=hermes_home, runner=runner)["context"]) + return str( + _load_database_snapshot( + user_message, + conversation_history=conversation_history, + hermes_home=hermes_home, + runner=runner, + )["context"] + ) def _word_count(value: str) -> int: @@ -1082,7 +1319,7 @@ def _requires_compiled_fallback(issues: list[str]) -> bool: def _pre_llm_call(**kwargs: Any) -> dict[str, str]: user_message = str(kwargs.get("user_message") or "") - snapshot = _load_database_snapshot(user_message) + snapshot = _load_database_snapshot(user_message, conversation_history=kwargs.get("conversation_history")) _store_snapshot(str(kwargs.get("session_id") or ""), snapshot) return {"context": str(snapshot["context"])} diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index c0fda17..1c2adae 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -336,6 +336,8 @@ def test_plugin_injects_bounded_retrieval_rows_and_writes_body_redacted_trace(tm "retrieval_receipt": { "schema": "livingip.teleoKbRetrievalReceipt.v1", "query_sha256": "1" * 64, + "retrieval_query_sha256": "4" * 64, + "retrieval_anchors_sha256": "5" * 64, "semantic_context_sha256": "2" * 64, "artifact_state_sha256": "3" * 64, "claim_ids": ["claim-1"], @@ -402,12 +404,15 @@ def test_plugin_injects_bounded_retrieval_rows_and_writes_body_redacted_trace(tm assert record["contract_ids"] == ["reply_budget", "source_intake"] assert record["database_reasoning_protocol_version"] == "leo-db-reasoning-v2" assert record["retrieval_receipt"]["semantic_context_sha256"] == "2" * 64 + assert record["retrieval_receipt"]["retrieval_query_sha256"] == "4" * 64 + assert record["retrieval_receipt"]["retrieval_anchors_sha256"] == "5" * 64 assert record["retrieval_receipt"]["read_consistency"]["system_identifier"] == "system-1" assert record["retrieval_receipt"]["counts"] == {"claims": 1, "context_rows": 1, "evidence_rows": 1} assert record["retrieval_receipt"]["injected_rows_sha256"] == injected_rows_sha256 assert len(record["retrieval_receipt"]["receipt_sha256"]) == 64 command, kwargs = calls[0] assert command[2:5] == ["--local", "context", "operator secret-shaped but nonsecret message"] + assert command[command.index("--retrieval-query") + 1] == "operator secret-shaped but nonsecret message" claim_limit = int(command[command.index("--limit") + 1]) context_limit = int(command[command.index("--context-limit") + 1]) assert 0 < claim_limit <= 8 @@ -417,6 +422,85 @@ def test_plugin_injects_bounded_retrieval_rows_and_writes_body_redacted_trace(tm assert kwargs["timeout"] == module.DEFAULT_TIMEOUT_SECONDS +def test_plugin_anchors_followup_retrieval_to_recent_user_topics_without_assistant_text() -> None: + module = load_plugin() + query = "Narrow that candidate and return the revised falsifier." + history = [ + { + "role": "user", + "content": "Does LivingIP token ownership create governance over narrative and licensing?", + }, + {"role": "assistant", "content": "UNRELATED_ASSISTANT_SECRET about GLP-1."}, + { + "role": "user", + "content": "Compare LivingIP ownership with governance rights and treasury control.", + }, + {"role": "user", "content": query}, + ] + + retrieval_query, anchors = module._contextual_retrieval_query(query, history) + + assert retrieval_query == query + assert anchors == ["LivingIP"] + assert "UNRELATED_ASSISTANT_SECRET" not in retrieval_query + assert len(anchors) <= module.MAX_RETRIEVAL_ANCHORS + + +def test_plugin_keeps_operational_query_separate_from_contextual_anchors(tmp_path: Path) -> None: + module = load_plugin() + home = tmp_path / "profile" + kb_tool = home / "bin" / "kb_tool.py" + kb_tool.parent.mkdir(parents=True) + kb_tool.write_text("# fixture\n", encoding="utf-8") + calls = [] + payload = { + "claims": [], + "context_rows": [], + "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}], + "compiled_response": None, + } + + def fake_runner(command, **kwargs): + calls.append(command) + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="") + + query = "Revise that candidate." + module.build_database_context( + query, + conversation_history=[ + {"role": "user", "content": "LivingIP ownership needs governance and treasury rights."}, + {"role": "user", "content": "LivingIP token ownership is not narrative control."}, + {"role": "user", "content": query}, + ], + hermes_home=home, + runner=fake_runner, + ) + + command = calls[0] + assert command[4] == query + retrieval_query = command[command.index("--retrieval-query") + 1] + assert retrieval_query == query + anchors = [command[index + 1] for index, item in enumerate(command) if item == "--retrieval-anchor"] + assert anchors == ["LivingIP"] + assert all("treasury rights" not in item for item in anchors) + + +def test_plugin_uses_repeated_user_topics_only_when_no_explicit_subject_anchor() -> None: + module = load_plugin() + query = "Narrow that position." + history = [ + {"role": "user", "content": "Compare ownership governance with licensing constraints."}, + {"role": "assistant", "content": "UNRELATED_ASSISTANT_SECRET governance ownership."}, + {"role": "user", "content": "Ownership governance may not imply narrative control."}, + {"role": "user", "content": query}, + ] + + retrieval_query, anchors = module._contextual_retrieval_query(query, history) + + assert retrieval_query == query + assert {item.casefold() for item in anchors} == {"ownership", "governance"} + + def test_plugin_replaces_contract_violating_draft_with_compiled_db_response(tmp_path: Path, monkeypatch) -> None: module = load_plugin() home = tmp_path / "profile" diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 76bfa56..bafa184 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -1411,7 +1411,7 @@ def test_vps_bridge_context_receipt_is_deterministic_and_retries_moving_wal(monk monkeypatch.setattr(module, "database_read_marker", lambda _args: next(markers)) monkeypatch.setattr(module, "context_operational_contracts", contracts) monkeypatch.setattr(module, "compile_operational_response", lambda _contracts: None) - monkeypatch.setattr(module, "find_claims", lambda _args, _query, _limit: []) + monkeypatch.setattr(module, "find_claims", lambda _args, _query, _limit, _anchors=None: []) monkeypatch.setattr(module, "load_evidence", lambda _args, _ids, _limit: {}) monkeypatch.setattr(module, "load_edges", lambda _args, _ids, _limit: {}) monkeypatch.setattr(module, "find_context_rows", lambda _args, _query, _limit: []) @@ -2211,7 +2211,7 @@ def test_vps_bridge_context_corpus_does_not_expose_persona_source_ref(monkeypatc assert "p.voice, p.role" in captured["sql"] -def test_vps_bridge_ranked_retrieval_keeps_one_match_recall_floor(monkeypatch) -> None: +def test_vps_bridge_ranked_retrieval_boosts_bounded_anchors_and_keeps_one_match_recall_floor(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") sql_statements = [] @@ -2220,14 +2220,75 @@ def test_vps_bridge_ranked_retrieval_keeps_one_match_recall_floor(monkeypatch) - return [] monkeypatch.setattr(module, "psql_json", capture_sql) - module.find_claims(SimpleNamespace(), "singular ceramic topic with several wrapper terms", 4) + module.find_claims( + SimpleNamespace(), + "singular ceramic topic with several wrapper terms", + 4, + ["LivingIP", "ownership"], + ) module.find_context_rows(SimpleNamespace(), "singular ceramic topic with several wrapper terms", 4) assert len(sql_statements) == 2 - for sql in sql_statements: - assert "max(score) over () as max_score" in sql - assert "score >= 1" in sql - assert "case when max_score >= 2 then 2 else 1 end" in sql + claim_sql, context_sql = sql_statements + assert "anchors(patterns) as" in claim_sql + assert "%livingip%" in claim_sql + assert "%ownership%" in claim_sql + assert "anchor_score" in claim_sql + assert "raw_selected as" in claim_sql + assert "anchor_selected as" in claim_sql + assert "'_selection_lane', selection_lane" in claim_sql + assert "where anchor_score >= 1" in claim_sql + assert "max(score) over () as max_score" in claim_sql + assert "max(score) over () as max_score" in context_sql + assert "case when max_score >= 2 then 2 else 1 end" in context_sql + + +def test_vps_bridge_ranked_retrieval_has_no_match_all_anchor_when_anchors_absent(monkeypatch) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + captured = {} + + def capture_sql(_args, sql): + captured["sql"] = sql + return [] + + monkeypatch.setattr(module, "psql_json", capture_sql) + module.find_claims(SimpleNamespace(), "specific topic", 4) + + assert "array[]::text[]" in captured["sql"] + assert "'%%'" not in captured["sql"] + + +def test_vps_bridge_ranked_retrieval_preserves_raw_results_when_subject_already_present(monkeypatch) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + rows = [ + {"id": "raw-subject", "score": 4, "anchor_score": 1, "_selection_lane": "raw"}, + {"id": "raw-cross-domain", "score": 3, "anchor_score": 0, "_selection_lane": "raw"}, + {"id": "anchor-duplicate", "score": 4, "anchor_score": 1, "_selection_lane": "anchor"}, + ] + monkeypatch.setattr(module, "psql_json", lambda _args, _sql: rows) + + result = module.find_claims(SimpleNamespace(), "LivingIP ownership", 4, ["LivingIP"]) + + assert [row["id"] for row in result] == ["raw-subject", "raw-cross-domain"] + assert all("_selection_lane" not in row for row in result) + + +def test_vps_bridge_ranked_retrieval_backfills_subject_without_discarding_top_raw_rows(monkeypatch) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + rows = [ + *( + {"id": f"raw-{index}", "score": 5 - index, "anchor_score": 0, "_selection_lane": "raw"} + for index in range(4) + ), + {"id": "subject-1", "score": 1, "anchor_score": 1, "_selection_lane": "anchor"}, + {"id": "subject-2", "score": 0, "anchor_score": 1, "_selection_lane": "anchor"}, + ] + monkeypatch.setattr(module, "psql_json", lambda _args, _sql: rows) + + result = module.find_claims(SimpleNamespace(), "revise that conclusion", 4, ["LivingIP"]) + + assert [row["id"] for row in result] == ["subject-1", "subject-2", "raw-0", "raw-1"] + assert all("_selection_lane" not in row for row in result) def test_vps_bridge_combines_identity_and_restart_contracts_for_independent_paraphrase() -> None: