diff --git a/altk_evolve/cli/cli.py b/altk_evolve/cli/cli.py index e0c53428..695339d5 100644 --- a/altk_evolve/cli/cli.py +++ b/altk_evolve/cli/cli.py @@ -407,6 +407,41 @@ def consolidate_entities( raise typer.Exit(1) +@entities_app.command("select") +def select_entities( + namespace: Annotated[str, typer.Argument(help="Namespace to select guidelines from")], + task: Annotated[str, typer.Option("--task", "-q", help="Task instruction to retrieve guidelines for")], + top_k: Annotated[Optional[int], typer.Option("--top-k", "-k", help="Max task-specific guidelines beyond the core")] = None, + core_support: Annotated[Optional[int], typer.Option("--core-support", "-c", help="Support threshold for the always-on core")] = None, +): + """Select an always-on core plus the top-k task-relevant guidelines (dosage-aware retrieval).""" + client = get_client() + + try: + selection = client.select_guidelines(namespace, task, top_k=top_k, core_support=core_support) + except NamespaceNotFoundException: + console.print(f"[red]Namespace '{namespace}' not found.[/red]") + raise typer.Exit(1) + except ValueError as e: + console.print(f"[red]Retrieval unavailable:[/red] {e}") + console.print("[yellow]Configure the embedding model/backend before selecting guidelines.[/yellow]") + raise typer.Exit(1) + + if not selection.all: + console.print("[yellow]No guidelines found for this task.[/yellow]") + return + + console.print(f"[bold]Core ({len(selection.core)}):[/bold]") + for entity in selection.core: + console.print(f" • {entity.content}") + console.print(f"\n[bold]Retrieved ({len(selection.retrieved)}):[/bold]") + for entity in selection.retrieved: + console.print(f" • {entity.content}") + console.print( + f"\n[dim]Total: {len(selection.all)} guidelines ({len(selection.core)} core + {len(selection.retrieved)} retrieved)[/dim]" + ) + + # ============================================================================= # Sync Commands # ============================================================================= diff --git a/altk_evolve/config/evolve.py b/altk_evolve/config/evolve.py index 03a9b494..6b028c58 100644 --- a/altk_evolve/config/evolve.py +++ b/altk_evolve/config/evolve.py @@ -1,3 +1,4 @@ +from pydantic import Field, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from typing import Literal @@ -17,6 +18,30 @@ class EvolveConfig(BaseSettings): # time, not by deleting entities here. consolidation_mode: Literal["none", "lossless", "lossy"] = "lossless" lossy_target_num_guidelines: int = 12 + # Dosage-aware retrieval knobs (see docs: capability-dependent dosage). + # static - inject the whole playbook (best for strong models; current default behavior) + # retrieval - inject core (support >= core_support) + top-k task-relevant guidelines + # Default is "static" so existing get_guidelines callers are unaffected; set "retrieval" + # to opt get_guidelines into the dosage-aware path. + injection_mode: Literal["static", "retrieval"] = "static" + retrieval_top_k: int = Field(default=10, ge=0) + core_support: int = Field(default=3, ge=1) + # Non-destructive sup2/sup3 floor on the candidate pool. Constrained to <= core_support + # (see validator below) so it can never drop a guideline that qualifies for the core. + min_support: int = Field(default=1, ge=1) + retrieval_similarity_key: Literal["source_task", "guideline_text"] = "source_task" + retrieval_near_core_thresh: float = Field(default=0.75, ge=0.0, le=1.0) + retrieval_dedup_thresh: float = Field(default=0.90, ge=0.0, le=1.0) + evidence_filter: Literal["all", "success", "failure"] = "all" + + @model_validator(mode="after") + def _check_support_thresholds(self) -> "EvolveConfig": + if self.min_support > self.core_support: + raise ValueError( + f"min_support ({self.min_support}) must be <= core_support ({self.core_support}); " + "a floor above the core threshold would drop guidelines that qualify for the core." + ) + return self # to reload settings call evolve_config.__init__() diff --git a/altk_evolve/frontend/client/evolve_client.py b/altk_evolve/frontend/client/evolve_client.py index 3117edcb..f565b676 100644 --- a/altk_evolve/frontend/client/evolve_client.py +++ b/altk_evolve/frontend/client/evolve_client.py @@ -1,4 +1,5 @@ import logging +from typing import TYPE_CHECKING, cast from altk_evolve.backend.base import BaseEntityBackend from altk_evolve.config.evolve import EvolveConfig @@ -7,9 +8,23 @@ from altk_evolve.schema.exceptions import NamespaceAlreadyExistsException, NamespaceNotFoundException from altk_evolve.schema.guidelines import ConsolidationResult +if TYPE_CHECKING: + from altk_evolve.llm.guidelines.retrieval import GuidelineSelection, SimilarityKey + logger = logging.getLogger(__name__) +def _filter_by_evidence(entities: list[RecordedEntity], evidence_filter: str) -> list[RecordedEntity]: + """Keep guidelines matching an evidence polarity. Unknown evidence (None) is always kept.""" + if evidence_filter == "success": + allowed = {"success", "both", None} + elif evidence_filter == "failure": + allowed = {"failure", "both", None} + else: # "all" + return entities + return [e for e in entities if (e.metadata or {}).get("evidence") in allowed] + + class EvolveClient: """Wrapper client around evolve entity backends.""" @@ -246,6 +261,61 @@ def consolidate_guidelines(self, namespace_id: str, threshold: float | None = No support_after=support_after, ) + def select_guidelines( + self, + namespace_id: str, + task_query: str, + top_k: int | None = None, + core_support: int | None = None, + min_support: int | None = None, + evidence_filter: str | None = None, + limit: int = 10000, + ) -> "GuidelineSelection": + """Select the always-on core plus the top-k task-relevant guidelines for a task. + + This is the dosage-aware alternative to injecting the whole playbook: the core + (guidelines with support >= ``core_support``) is always returned, plus up to + ``top_k`` further guidelines whose source task is most similar to ``task_query``. + Unset arguments fall back to the corresponding ``config`` values. + + Args: + namespace_id: Namespace to select guidelines from. + task_query: The current task instruction to retrieve for. + top_k: Max retrieved (non-core) guidelines. Defaults to ``config.retrieval_top_k``. + core_support: Support threshold for the always-on core. Defaults to config. + min_support: Non-destructive support floor on the candidate pool. Defaults to config. + evidence_filter: ``"all"`` / ``"success"`` / ``"failure"``. Defaults to config. + limit: Max guideline entities to fetch. + + Returns: + A ``GuidelineSelection`` (core + retrieved). + """ + from altk_evolve.llm.guidelines.retrieval import GuidelineSelection, select_guidelines + + top_k = top_k if top_k is not None else getattr(self.config, "retrieval_top_k", 10) + core_support = core_support if core_support is not None else getattr(self.config, "core_support", 3) + min_support = min_support if min_support is not None else getattr(self.config, "min_support", 1) + evidence_filter = evidence_filter or getattr(self.config, "evidence_filter", "all") + similarity_key = getattr(self.config, "retrieval_similarity_key", "source_task") + near_core_thresh = getattr(self.config, "retrieval_near_core_thresh", 0.75) + dedup_thresh = getattr(self.config, "retrieval_dedup_thresh", 0.90) + + entities = self.get_all_entities(namespace_id, filters={"type": "guideline"}, limit=limit) + entities = _filter_by_evidence(entities, str(evidence_filter)) + if not entities: + return GuidelineSelection(core=[], retrieved=[]) + + return select_guidelines( + entities, + task_query, + top_k=int(top_k), + core_support=int(core_support), + min_support=int(min_support), + similarity_key=cast("SimilarityKey", similarity_key), + near_core_thresh=float(near_core_thresh), + dedup_thresh=float(dedup_thresh), + ) + # Convenience methods for common patterns def namespace_exists(self, namespace_id: str) -> bool: """Check if a namespace exists.""" diff --git a/altk_evolve/frontend/mcp/mcp_server.py b/altk_evolve/frontend/mcp/mcp_server.py index a012b47d..e00080c3 100644 --- a/altk_evolve/frontend/mcp/mcp_server.py +++ b/altk_evolve/frontend/mcp/mcp_server.py @@ -305,15 +305,74 @@ def get_guidelines( Provide a task description and receive applicable best practices and guidelines. This tool is maintained for backward compatibility. Use 'get_entities' for more generic queries. + Honors ``EvolveConfig.injection_mode``: 'static' (default) returns the whole guideline set, + while 'retrieval' routes through the dosage-aware core + top-k path (see get_relevant_guidelines). + Args: task: A description of the task you want guidelines for user_id: Optional caller user ID. Logged for attribution; does not filter results. namespace_id: Optional namespace override. Falls back to the configured default. session_id: Optional session/thread ID. Logged for attribution; does not filter results. """ + from altk_evolve.config.evolve import evolve_config + + if evolve_config.injection_mode == "retrieval": + return get_relevant_guidelines(task, user_id=user_id, namespace_id=namespace_id, session_id=session_id) return get_entities_logic(task, "guideline", user_id=user_id, namespace_id=namespace_id, session_id=session_id) +@mcp.tool() +def get_relevant_guidelines( + task: str, + top_k: int | None = None, + core_support: int | None = None, + namespace_id: str | None = None, + user_id: str | None = None, + session_id: str | None = None, +) -> str: + """ + Get a dosage-aware set of guidelines for a task: an always-on core plus the top-k + guidelines most relevant to this task (retrieved by similarity to the tasks they were + learned from). + + Prefer this over 'get_guidelines'/'get_entities' for weaker models, where injecting the + whole playbook can hurt and a small, targeted set helps. + + Args: + task: A description of the task you want guidelines for. + top_k: Max task-specific guidelines to add beyond the core. Defaults to config. + core_support: Support threshold for the always-on core. Defaults to config. + namespace_id: Optional namespace override. Falls back to the configured default. + user_id: Optional caller user ID. Logged for attribution; does not filter results. + session_id: Optional session/thread ID. Logged for attribution; does not filter results. + """ + from altk_evolve.llm.guidelines.retrieval import format_selection + + resolved_ns = _resolve_namespace(namespace_id) + # Log only non-sensitive metadata at INFO; task is arbitrary user text (see save_trajectory). + logger.info( + "get_relevant_guidelines (namespace=%s, top_k=%s, core_support=%s, task_len=%s, user_present=%s, session_present=%s)", + resolved_ns, + top_k, + core_support, + len(task), + user_id is not None, + session_id is not None, + ) + logger.debug("get_relevant_guidelines task=%s", task) + client = get_client() + try: + selection = client.select_guidelines(resolved_ns, task, top_k=top_k, core_support=core_support) + except NamespaceNotFoundException: + _evict_namespace(resolved_ns) + resolved_ns = _resolve_namespace(namespace_id) + selection = client.select_guidelines(resolved_ns, task, top_k=top_k, core_support=core_support) + except ValueError as e: + logger.error("Retrieval unavailable for get_relevant_guidelines: %s", e) + return f"Guideline retrieval unavailable: {e}" + return format_selection(selection) + + def _empty_store_user_facts_response(user_id: str) -> str: return json.dumps({"user_id": user_id, "stored_count": 0, "updates": []}) diff --git a/altk_evolve/llm/guidelines/retrieval.py b/altk_evolve/llm/guidelines/retrieval.py new file mode 100644 index 00000000..1cba1f25 --- /dev/null +++ b/altk_evolve/llm/guidelines/retrieval.py @@ -0,0 +1,152 @@ +"""Dosage-aware guideline selection: always-on core + per-task retrieved guidelines. + +Instead of injecting the whole playbook every task (best only for strong models), this +selects a small, task-relevant dose (best for weaker models — the capability-dependent +dosage finding): + + core = guidelines whose ``support`` >= ``core_support`` (they recurred across many + tasks, so they generalise) — always included. + retrieved = up to ``top_k`` further guidelines whose SOURCE task (``task_description``) + is most similar to the current task ("a lesson from a task like this one"), + after dropping ones already covered by the core or duplicating each other. + +``min_support`` applies a non-destructive support threshold (the sup2/sup3 filter) to the +candidate pool without deleting anything from the store. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Literal + +import numpy as np + +from altk_evolve.llm.guidelines.clustering import _get_sentence_transformer +from altk_evolve.schema.core import RecordedEntity + +logger = logging.getLogger(__name__) + +SimilarityKey = Literal["source_task", "guideline_text"] + + +@dataclass(frozen=True) +class GuidelineSelection: + """Result of :func:`select_guidelines` — core (always-on) plus retrieved guidelines.""" + + core: list[RecordedEntity] = field(default_factory=list) + retrieved: list[RecordedEntity] = field(default_factory=list) + + @property + def all(self) -> list[RecordedEntity]: + return [*self.core, *self.retrieved] + + +def _support(entity: RecordedEntity) -> int: + try: + return max(1, int((entity.metadata or {}).get("support", 1) or 1)) + except (TypeError, ValueError): + return 1 + + +def _key_text(entity: RecordedEntity, similarity_key: SimilarityKey) -> str: + if similarity_key == "source_task": + return str((entity.metadata or {}).get("task_description", "") or entity.content) + return str(entity.content) + + +def _embed(texts: list[str], embedding_model: str) -> np.ndarray: + model = _get_sentence_transformer(embedding_model) + return np.asarray(model.encode(texts, normalize_embeddings=True)) + + +def select_guidelines( + entities: list[RecordedEntity], + task_query: str, + *, + top_k: int = 10, + core_support: int = 3, + min_support: int = 1, + similarity_key: SimilarityKey = "source_task", + near_core_thresh: float = 0.75, + dedup_thresh: float = 0.90, + embedding_model: str | None = None, +) -> GuidelineSelection: + """Select the always-on core plus the top-``top_k`` task-relevant guidelines. + + Args: + entities: Candidate guideline entities (carrying ``support`` and, for + ``similarity_key="source_task"``, ``task_description`` in metadata). + task_query: The current task instruction to retrieve for. + top_k: Maximum number of retrieved (non-core) guidelines. + core_support: Guidelines with support >= this are always included. + min_support: Non-destructive sup2/sup3 floor applied to the whole pool *before* the + core/candidate split. Callers should keep ``min_support <= core_support`` (enforced + by ``EvolveConfig``) so the floor never drops a guideline that qualifies for the core. + similarity_key: Retrieve by the source ``task_description`` (default) or by the + guideline text itself. For ``"source_task"``, a candidate lacking a + ``task_description`` falls back to being ranked by its own content. + near_core_thresh: Drop a candidate whose content cosine to any core guideline is + >= this (already covered by the core). + dedup_thresh: Drop a candidate whose content cosine to an already-kept candidate is + >= this. + embedding_model: SentenceTransformer model name. Defaults to the configured model. + + Returns: + A :class:`GuidelineSelection` (core first, then retrieved by descending relevance). + """ + if embedding_model is None: + from altk_evolve.config.milvus import milvus_other_settings + + embedding_model = milvus_other_settings.embedding_model + + pool = [e for e in entities if _support(e) >= min_support] + core = [e for e in pool if _support(e) >= core_support] + candidates = [e for e in pool if _support(e) < core_support] + + if top_k <= 0 or not candidates: + return GuidelineSelection(core=core, retrieved=[]) + + core_content_emb = _embed([str(e.content) for e in core], embedding_model) if core else None + cand_content_emb = _embed([str(e.content) for e in candidates], embedding_model) + cand_key_emb = ( + cand_content_emb + if similarity_key == "guideline_text" + else _embed([_key_text(e, similarity_key) for e in candidates], embedding_model) + ) + query_emb = _embed([task_query], embedding_model)[0] + + # Drop candidates already covered by the core, then dedup the remainder, then rank by + # similarity of the source task to the current task. + kept: list[tuple[int, float]] = [] # (candidate index, relevance score) + kept_content: list[np.ndarray] = [] + order = sorted(range(len(candidates)), key=lambda i: float(cand_key_emb[i] @ query_emb), reverse=True) + for i in order: + content_vec = cand_content_emb[i] + if core_content_emb is not None and float(np.max(core_content_emb @ content_vec)) >= near_core_thresh: + continue + if kept_content and float(np.max(np.stack(kept_content) @ content_vec)) >= dedup_thresh: + continue + kept.append((i, float(cand_key_emb[i] @ query_emb))) + kept_content.append(content_vec) + if len(kept) >= top_k: + break + + retrieved = [candidates[i] for i, _ in kept] + return GuidelineSelection(core=core, retrieved=retrieved) + + +def format_selection(selection: GuidelineSelection) -> str: + """Render a selection as an injectable guidelines block (mirrors the retrieved-tips agent).""" + lines = [ + "Guidelines learned from past task attempts (follow these carefully; they address the most common mistakes):", + "", + ] + for entity in selection.core: + lines.append(f"- {entity.content}") + if selection.retrieved: + lines.append("") + lines.append("Additional task-specific guidelines retrieved from similar past tasks:") + for entity in selection.retrieved: + lines.append(f"- {entity.content}") + return "\n".join(lines) diff --git a/tests/unit/test_retrieval.py b/tests/unit/test_retrieval.py new file mode 100644 index 00000000..ada6167f --- /dev/null +++ b/tests/unit/test_retrieval.py @@ -0,0 +1,236 @@ +"""Unit tests for dosage-aware guideline selection (core + top-k retrieval).""" + +from datetime import datetime +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest + +from altk_evolve.llm.guidelines.retrieval import GuidelineSelection, format_selection, select_guidelines +from altk_evolve.schema.core import RecordedEntity + + +def _make_entity( + entity_id: str, + content: str, + task_description: str = "", + support: int = 1, + evidence: str | None = None, +) -> RecordedEntity: + metadata: dict = {"support": support} + if task_description: + metadata["task_description"] = task_description + if evidence is not None: + metadata["evidence"] = evidence + return RecordedEntity( + id=entity_id, + content=content, + type="guideline", + metadata=metadata, + created_at=datetime(2025, 1, 1), + ) + + +def _vec(text: str) -> list[float]: + """Deterministic orthogonal-ish embedding keyed on a marker word in the text.""" + t = text.lower() + if "spotify" in t: + return [1.0, 0.0, 0.0] + if "venmo" in t: + return [0.0, 1.0, 0.0] + if "email" in t: + return [0.0, 0.0, 1.0] + return [0.5, 0.5, 0.0] + + +def _mock_encode(texts, normalize_embeddings=True): + return np.array([_vec(t) for t in texts]) + + +@pytest.mark.unit +@patch("altk_evolve.llm.guidelines.retrieval._get_sentence_transformer") +class TestSelectGuidelines: + def _model(self, mock_st): + model = MagicMock() + model.encode = _mock_encode + mock_st.return_value = model + return model + + def test_core_always_included_regardless_of_similarity(self, mock_st): + self._model(mock_st) + # Core (support 3) is about email; task is about spotify -> dissimilar, still included. + core = _make_entity("c", "Handle email pagination", task_description="email task", support=3) + cand = _make_entity("s", "Spotify search tip", task_description="spotify task", support=1) + + sel = select_guidelines([core, cand], "do something on spotify", top_k=5, core_support=3, embedding_model="test") + + assert [e.id for e in sel.core] == ["c"] + assert [e.id for e in sel.retrieved] == ["s"] + + def test_topk_ranked_by_source_task_similarity(self, mock_st): + self._model(mock_st) + cands = [ + _make_entity("spotify", "Spotify rule", task_description="a spotify task", support=1), + _make_entity("venmo", "Venmo rule", task_description="a venmo task", support=1), + _make_entity("email", "Email rule", task_description="an email task", support=1), + ] + sel = select_guidelines(cands, "pay someone on venmo", top_k=1, core_support=3, embedding_model="test") + + assert sel.core == [] + assert [e.id for e in sel.retrieved] == ["venmo"] + + def test_min_support_filters_candidate_pool(self, mock_st): + self._model(mock_st) + cands = [ + _make_entity("weak", "Venmo weak", task_description="a venmo task", support=1), + _make_entity("strong", "Venmo strong", task_description="a venmo task", support=2), + ] + sel = select_guidelines(cands, "pay on venmo", top_k=5, core_support=3, min_support=2, embedding_model="test") + + assert [e.id for e in sel.retrieved] == ["strong"] + + def test_near_core_candidate_dropped(self, mock_st): + self._model(mock_st) + # Candidate content duplicates the core content -> dropped as already covered. + core = _make_entity("c", "Venmo rule", task_description="venmo core", support=3) + dup = _make_entity("d", "Venmo rule", task_description="a venmo task", support=1) + fresh = _make_entity("f", "Email rule", task_description="an email task", support=1) + + sel = select_guidelines([core, dup, fresh], "venmo payment", top_k=5, core_support=3, near_core_thresh=0.9, embedding_model="test") + + assert [e.id for e in sel.core] == ["c"] + assert "d" not in {e.id for e in sel.retrieved} + assert "f" in {e.id for e in sel.retrieved} + + def test_dedup_drops_near_identical_candidate(self, mock_st): + self._model(mock_st) + # v1 and v2 have identical content (same embedding) -> dedup keeps one; email is distinct. + v1 = _make_entity("v1", "Venmo rule", task_description="a venmo task", support=1) + v2 = _make_entity("v2", "Venmo rule", task_description="another venmo task", support=1) + email = _make_entity("email", "Email rule", task_description="an email task", support=1) + + sel = select_guidelines([v1, v2, email], "venmo payment", top_k=5, core_support=3, dedup_thresh=0.9, embedding_model="test") + + retrieved_ids = {e.id for e in sel.retrieved} + assert len(retrieved_ids & {"v1", "v2"}) == 1 # exactly one of the duplicates survives + assert "email" in retrieved_ids + assert len(sel.retrieved) == 2 + + def test_topk_zero_returns_core_only(self, mock_st): + self._model(mock_st) + core = _make_entity("c", "Email rule", task_description="email", support=3) + cand = _make_entity("s", "Spotify rule", task_description="spotify", support=1) + sel = select_guidelines([core, cand], "spotify task", top_k=0, core_support=3, embedding_model="test") + assert [e.id for e in sel.core] == ["c"] + assert sel.retrieved == [] + + def test_no_candidates_returns_core_only(self, mock_st): + self._model(mock_st) + core = _make_entity("c", "Email rule", task_description="email", support=3) + sel = select_guidelines([core], "email task", top_k=5, core_support=3, embedding_model="test") + assert [e.id for e in sel.core] == ["c"] + assert sel.retrieved == [] + + +@pytest.mark.unit +def test_format_selection_renders_core_and_retrieved(): + core = [_make_entity("c", "Always paginate", support=3)] + retrieved = [_make_entity("r", "Use exact match search", support=1)] + text = format_selection(GuidelineSelection(core=core, retrieved=retrieved)) + assert "Always paginate" in text + assert "Additional task-specific guidelines" in text + assert "Use exact match search" in text + + +@pytest.mark.unit +class TestClientSelectGuidelines: + def test_select_guidelines_filters_evidence_and_delegates(self): + from altk_evolve.frontend.client.evolve_client import EvolveClient + + entities = [ + _make_entity("ok", "keep me", support=1, evidence="failure"), + _make_entity("drop", "drop me", support=1, evidence="success"), + _make_entity("unknown", "keep unknown", support=1), + ] + mock_backend = MagicMock() + client = EvolveClient.__new__(EvolveClient) + client.backend = mock_backend + client.config = MagicMock() + client.config.retrieval_top_k = 5 + client.config.core_support = 3 + client.config.min_support = 1 + client.config.evidence_filter = "failure" + client.config.retrieval_similarity_key = "source_task" + client.config.retrieval_near_core_thresh = 0.75 + client.config.retrieval_dedup_thresh = 0.90 + + captured = {} + + def fake_select(ents, task_query, **kwargs): + captured["ids"] = [e.id for e in ents] + return GuidelineSelection(core=list(ents), retrieved=[]) + + with ( + patch.object(client, "get_all_entities", return_value=entities), + patch("altk_evolve.llm.guidelines.retrieval.select_guidelines", side_effect=fake_select), + ): + client.select_guidelines("ns", "some task") + + # success-evidence guideline dropped; failure + unknown kept. + assert captured["ids"] == ["ok", "unknown"] + + def test_select_guidelines_fetches_full_unlimited_pool(self): + """Guards against a silent cap: the candidate pool must not be limited to the + get_all_entities default (100), or retrieval would ignore most guidelines.""" + from altk_evolve.frontend.client.evolve_client import EvolveClient + + client = EvolveClient.__new__(EvolveClient) + client.backend = MagicMock() + client.config = MagicMock() + client.config.retrieval_top_k = 5 + client.config.core_support = 3 + client.config.min_support = 1 + client.config.evidence_filter = "all" + client.config.retrieval_similarity_key = "source_task" + client.config.retrieval_near_core_thresh = 0.75 + client.config.retrieval_dedup_thresh = 0.90 + + with patch.object(client, "get_all_entities", return_value=[]) as mock_get: + client.select_guidelines("ns", "some task") + + mock_get.assert_called_once_with("ns", filters={"type": "guideline"}, limit=10000) + + +@pytest.mark.unit +class TestInjectionModeRouting: + """get_guidelines must honor EvolveConfig.injection_mode (default 'static').""" + + def test_static_mode_uses_full_playbook_path(self): + import altk_evolve.config.evolve as cfg + import altk_evolve.frontend.mcp.mcp_server as mcp + + with ( + patch.object(cfg.evolve_config, "injection_mode", "static"), + patch.object(mcp, "get_entities_logic", return_value="STATIC") as static_path, + patch.object(mcp, "get_relevant_guidelines", return_value="RETRIEVAL") as retrieval_path, + ): + out = mcp.get_guidelines("do a task") + + assert out == "STATIC" + static_path.assert_called_once() + retrieval_path.assert_not_called() + + def test_retrieval_mode_routes_through_dosage_path(self): + import altk_evolve.config.evolve as cfg + import altk_evolve.frontend.mcp.mcp_server as mcp + + with ( + patch.object(cfg.evolve_config, "injection_mode", "retrieval"), + patch.object(mcp, "get_entities_logic", return_value="STATIC") as static_path, + patch.object(mcp, "get_relevant_guidelines", return_value="RETRIEVAL") as retrieval_path, + ): + out = mcp.get_guidelines("do a task") + + assert out == "RETRIEVAL" + retrieval_path.assert_called_once() + static_path.assert_not_called()