feat(memory): align semantic and KB recall

coding-agent/helm/templates/configmap-agent-config.yaml

@@ -87,6 +87,7 @@ data:
     {{- if ne .Values.kb.maxDistance nil }}
     max_distance = {{ .Values.kb.maxDistance }}
     {{- end }}
+    defer_when_semantic_memory_hits = {{ .Values.kb.deferWhenSemanticMemoryHits }}
     corpus = {{ .Values.kb.corpus | quote }}
     search_corpora = {{ .Values.kb.searchCorpora | toJson }}
     {{- end }}

coding-agent/helm/values.yaml

@@ -130,6 +130,7 @@ kb:
   embeddingBaseUrl: https://api.siliconflow.cn/v1
   embeddingDim: 1024
   maxDistance: null
+  deferWhenSemanticMemoryHits: false
   corpus: notes
   searchCorpora:
     - sre

coding-agent/src/coding_agent/core/app.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-import os
 import inspect
+import os
 import uuid
 from collections.abc import Mapping
 from contextlib import contextmanager
@@ -18,27 +18,27 @@
 from agentkit.runtime.hookspecs import HOOK_SPECS
 from agentkit.runtime.pipeline import Pipeline, PipelineContext
 from agentkit.tape.tape import Tape
-
-from coding_agent.core.agent_identity import legacy_agent_id_str
 from coding_agent.approval import ApprovalPolicy
+from coding_agent.core.agent_identity import legacy_agent_id_str
 from coding_agent.environment import LocalEnvironment
 from coding_agent.environment.additional_roots import with_additional_workspace_roots
+from coding_agent.observability import build_observation_sink
 from coding_agent.plugins.approval import ApprovalPlugin
 from coding_agent.plugins.core_tools import CoreToolsPlugin
 from coding_agent.plugins.doom_detector import DoomDetectorPlugin
+from coding_agent.plugins.kb import KBPlugin
 from coding_agent.plugins.llm_provider import LLMProviderPlugin
 from coding_agent.plugins.mcp import MCPPlugin
 from coding_agent.plugins.memory import MemoryPlugin
-from coding_agent.plugins.kb import KBPlugin
 from coding_agent.plugins.metrics import SessionMetricsPlugin
-from coding_agent.observability import build_observation_sink
 from coding_agent.plugins.parallel_executor import ParallelExecutorPlugin
 from coding_agent.plugins.semantic_memory import SemanticMemoryPlugin
 from coding_agent.plugins.shell_session import ShellSessionPlugin
 from coding_agent.plugins.skills import SkillsPlugin
 from coding_agent.plugins.storage import StoragePlugin
 from coding_agent.plugins.summarizer import SummarizerPlugin
 from coding_agent.subagents.coordinator import ChildWorkerCoordinator
+from coding_agent.tools.web_search import create_web_search_backend
 from coding_agent.topics.memory import MemoryReviewStore
 from coding_agent.topics.range_index import TopicRangeIndex
 from coding_agent.topics.semantic_backends import (
@@ -53,7 +53,6 @@
     SemanticMemoryReviewSyncService,
     SemanticMemorySyncer,
 )
-from coding_agent.tools.web_search import create_web_search_backend
 
 ToolFilter = Any
 
@@ -529,6 +528,11 @@ def create_child_pipeline(
     kb_db_path = kb_cfg.get("db_path", "kb")
     if not isinstance(kb_db_path, str):
         raise ValueError("[kb].db_path must be a string")
+    kb_defer_when_semantic_memory_hits = kb_cfg.get(
+        "defer_when_semantic_memory_hits", False
+    )
+    if not isinstance(kb_defer_when_semantic_memory_hits, bool):
+        raise TypeError("[kb].defer_when_semantic_memory_hits must be a boolean")
     memory_review_store = MemoryReviewStore(
         data_dir / kb_db_path / "reviewed_memory.jsonl",
         candidate_writes_enabled=memory_cfg.effective_write_enabled,
@@ -674,11 +678,16 @@ async def _execute_tool_async(
                 ),
                 corpus=kb_corpus,
                 search_corpora=kb_search_corpora,
+                defer_when_semantic_memory_hits=kb_defer_when_semantic_memory_hits,
             ),
         }
     )
 
     enabled_plugins = cfg.plugins or list(plugin_factories.keys())
+    _validate_kb_semantic_plugin_order(
+        enabled_plugins,
+        kb_defer_when_semantic_memory_hits=kb_defer_when_semantic_memory_hits,
+    )
     for plugin_name in enabled_plugins:
         factory = plugin_factories.get(plugin_name)
         if factory is None:
@@ -798,6 +807,22 @@ async def _memory_handler(directive: Any) -> None:
     return pipeline, ctx
 
 
+def _validate_kb_semantic_plugin_order(
+    enabled_plugins: list[str],
+    *,
+    kb_defer_when_semantic_memory_hits: bool,
+) -> None:
+    if not kb_defer_when_semantic_memory_hits:
+        return
+    if "kb" not in enabled_plugins or "semantic_memory" not in enabled_plugins:
+        return
+    if enabled_plugins.index("semantic_memory") > enabled_plugins.index("kb"):
+        raise ValueError(
+            "[kb].defer_when_semantic_memory_hits requires semantic_memory "
+            "to run before kb in [agent.plugins].enabled"
+        )
+
+
 def create_agent(
     config_path: Path | None = None,
     data_dir: Path | None = None,

coding-agent/src/coding_agent/plugins/kb.py

@@ -1,27 +1,38 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any
 
 from agentkit.observability import ObservationSink, record_span
+from agentkit.runtime.messages import RuntimeMessageKind
+from agentkit.runtime.pipeline import PipelineContext
 from agentkit.tape.tape import Tape
-
+from coding_agent.kb import KB, KBSearchResult
+from coding_agent.plugins.semantic_memory import (
+    SEMANTIC_MEMORY_GROUNDING_MARKER_KEY,
+    semantic_grounding_query_digest,
+)
 from coding_agent.topics.context_pack import (
     ContextPack,
     ContextPackItem,
     ContextPackRenderer,
     ContextPackSection,
     EvidenceRef,
 )
-from coding_agent.kb import KB, KBSearchResult
 
 logger = logging.getLogger(__name__)
 
 _CHUNK_TRUNCATE = 500
 _CONTEXT_PACK_RENDERER = ContextPackRenderer(max_item_chars=_CHUNK_TRUNCATE)
+_RUNTIME_QUERY_KINDS = frozenset(
+    {
+        RuntimeMessageKind.USER_STEER,
+        RuntimeMessageKind.SUBAGENT_MESSAGE,
+    }
+)
 
 
 @dataclass
@@ -50,6 +61,7 @@ def __init__(
         text_extensions: list[str] | set[str] | None = None,
         corpus: str = "default",
         search_corpora: list[str] | tuple[str, ...] | None = None,
+        defer_when_semantic_memory_hits: bool = False,
         embedding_fn: Callable[[list[str]], list[list[float]]] | None = None,
     ) -> None:
         self._db_path = db_path
@@ -72,6 +84,9 @@ def __init__(
         self._search_corpora = (
             tuple(search_corpora) if search_corpora is not None else None
         )
+        if not isinstance(defer_when_semantic_memory_hits, bool):
+            raise TypeError("defer_when_semantic_memory_hits must be a boolean")
+        self._defer_when_semantic_memory_hits = defer_when_semantic_memory_hits
         normalized_extensions = index_extensions
         if normalized_extensions is None and text_extensions is not None:
             normalized_extensions = list(text_extensions)
@@ -120,13 +135,20 @@ def do_mount(self, **kwargs: Any) -> dict[str, Any]:
     def build_context(
         self, tape: Tape | None = None, **kwargs: Any
     ) -> list[dict[str, Any]]:
-        del kwargs
         if tape is None or not self._has_table or self._kb is None:
             return []
 
-        user_message = _latest_user_message(tape)
+        user_message = _latest_runtime_prompt_message(kwargs.get("ctx"))
+        if user_message is None:
+            user_message = _latest_user_message(tape)
         if user_message is None:
             return []
+        if self._defer_when_semantic_memory_hits and _take_semantic_memory_hit_count(
+            kwargs.get("ctx"),
+            tape=tape,
+            query=user_message,
+        ):
+            return []
 
         if self._snapshot is not None and self._snapshot.last_user_msg == user_message:
             _record_retrieval_cache_hit(
@@ -180,6 +202,24 @@ def build_context(
         return grounding
 
 
+def _take_semantic_memory_hit_count(ctx: Any, *, tape: Tape, query: str) -> int:
+    if not isinstance(ctx, PipelineContext):
+        return 0
+    marker = ctx.config.pop(SEMANTIC_MEMORY_GROUNDING_MARKER_KEY, None)
+    if not isinstance(marker, Mapping):
+        return 0
+    if marker.get("query_digest") != semantic_grounding_query_digest(query):
+        return 0
+    if marker.get("tape_entry_count") != len(tape):
+        return 0
+    hit_count = marker.get("hit_count", 0)
+    if isinstance(hit_count, bool):
+        return 0
+    if isinstance(hit_count, int):
+        return max(hit_count, 0)
+    return 0
+
+
 def _observation_sink_from_context(ctx: Any) -> ObservationSink | None:
     if ctx is None:
         return None
@@ -208,6 +248,28 @@ def _latest_user_message(tape: Tape) -> str | None:
     return None
 
 
+def _latest_runtime_prompt_message(ctx: object) -> str | None:
+    if not isinstance(ctx, PipelineContext):
+        return None
+
+    for item in reversed(ctx.runtime_messages):
+        message = item.message
+        if message.kind not in _RUNTIME_QUERY_KINDS:
+            continue
+        text = _runtime_payload_text(message.payload)
+        if text is not None:
+            return text
+    return None
+
+
+def _runtime_payload_text(payload: Mapping[str, object]) -> str | None:
+    for key in ("text", "message", "content"):
+        value = payload.get(key)
+        if isinstance(value, str) and value.strip():
+            return value
+    return None
+
+
 def _format_grounding_messages(results: list[KBSearchResult]) -> list[dict[str, Any]]:
     if not results:
         return []

coding-agent/src/coding_agent/plugins/semantic_memory.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping
-from datetime import UTC, datetime
+import hashlib
 import inspect
-from typing import Any, Callable
+from collections.abc import Callable, Mapping
+from datetime import UTC, datetime
+from typing import Any
 
 from agentkit.runtime.messages import RuntimeMessageKind
 from agentkit.runtime.pipeline import PipelineContext
@@ -16,6 +17,7 @@
     memory_candidate_session_id,
 )
 from coding_agent.topics.range_index import TopicRangeIndex, require_recall_safe_text
+from coding_agent.topics.range_index_builder import build_topic_range_index_from_store
 from coding_agent.topics.recall_context import (
     TopicRecallPlanner,
     TopicRecallPlannerInput,
@@ -28,13 +30,13 @@
 )
 from coding_agent.topics.store import TopicRecord
 
-
 _RUNTIME_QUERY_KINDS = frozenset(
     {
         RuntimeMessageKind.USER_STEER,
         RuntimeMessageKind.SUBAGENT_MESSAGE,
     }
 )
+SEMANTIC_MEMORY_GROUNDING_MARKER_KEY = "semantic_memory.grounding_marker"
 
 
 class _NoopTopicStore:
@@ -66,13 +68,25 @@ def __init__(
         self._limit = limit
         self._topic_store = _validate_topic_store(topic_store)
         self._topic_index = _validate_topic_index(topic_index)
+        self._derived_topic_index: TopicRangeIndex | None = None
 
     def hooks(self) -> dict[str, Callable[..., Any]]:
         return {"build_context": self.build_context}
 
+    async def _topic_index_for_context(self) -> TopicRangeIndex:
+        if self._topic_index is not None:
+            return self._topic_index
+        if self._derived_topic_index is None:
+            derived = await build_topic_range_index_from_store(self._topic_store)
+            self._derived_topic_index = (
+                TopicRangeIndex() if derived is None else derived.index
+            )
+        return self._derived_topic_index
+
     async def build_context(
         self, tape: Tape | None = None, **kwargs: Any
     ) -> list[dict[str, object]]:
+        _clear_semantic_memory_grounding_marker(kwargs.get("ctx"))
         if not self._read_enabled:
             return []
         if tape is None:
@@ -87,9 +101,10 @@ async def build_context(
             return []
 
         session_id = _session_id_from_context(kwargs.get("ctx"))
+        topic_index = await self._topic_index_for_context()
         planner = SemanticRecallPlanner(
             topic_planner=TopicRecallPlanner(
-                topic_index=self._topic_index,
+                topic_index=topic_index,
                 accepted_memories=_accepted_memories_for_context(
                     self._memory_review_store,
                     session_id=session_id,
@@ -110,6 +125,12 @@ async def build_context(
                 enabled=self._read_enabled,
             )
         )
+        _set_semantic_memory_grounding_marker(
+            kwargs.get("ctx"),
+            tape=tape,
+            query=user_message,
+            hit_count=len(plan.topic_results) + len(plan.accepted_memories),
+        )
         return recall_context_messages(plan, enabled=self._read_enabled)
 
 
@@ -168,9 +189,11 @@ def _validate_topic_store(
     return topic_store
 
 
-def _validate_topic_index(topic_index: TopicRangeIndex | None) -> TopicRangeIndex:
+def _validate_topic_index(
+    topic_index: TopicRangeIndex | None,
+) -> TopicRangeIndex | None:
     if topic_index is None:
-        return TopicRangeIndex()
+        return None
     if not isinstance(topic_index, TopicRangeIndex):
         raise TypeError("topic_index must be TopicRangeIndex")
     return topic_index
@@ -182,6 +205,30 @@ def _session_id_from_context(ctx: object) -> str | None:
     return None
 
 
+def _clear_semantic_memory_grounding_marker(ctx: object) -> None:
+    if isinstance(ctx, PipelineContext):
+        ctx.config.pop(SEMANTIC_MEMORY_GROUNDING_MARKER_KEY, None)
+
+
+def _set_semantic_memory_grounding_marker(
+    ctx: object,
+    *,
+    tape: Tape,
+    query: str,
+    hit_count: int,
+) -> None:
+    if isinstance(ctx, PipelineContext):
+        ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = {
+            "query_digest": semantic_grounding_query_digest(query),
+            "tape_entry_count": len(tape),
+            "hit_count": hit_count,
+        }
+
+
+def semantic_grounding_query_digest(query: str) -> str:
+    return hashlib.sha256(query.encode("utf-8")).hexdigest()
+
+
 def _accepted_memories_for_context(
     review_store: MemoryReviewStore,
     *,