From 36e1c36650fc96cf7bea0667d573093b436a1e2d Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 30 Jun 2026 10:48:11 +0800 Subject: [PATCH 1/2] feat(memory): align semantic and kb recall Use the current runtime prompt for both semantic memory and KB retrieval, add an optional KB defer gate when semantic memory has grounded the same query, and lazily derive durable topic range indexes for semantic recall. Verification: uv run pytest tests/coding_agent/plugins/test_kb_plugin.py tests/coding_agent/plugins/test_semantic_memory.py tests/coding_agent/test_context_system_smoke.py tests/coding_agent/test_bootstrap.py tests/deploy/test_helm_chart.py -v Verification: uv run ruff check/format --check on touched files --- .../templates/configmap-agent-config.yaml | 1 + coding-agent/helm/values.yaml | 1 + coding-agent/src/coding_agent/core/app.py | 37 +++- coding-agent/src/coding_agent/plugins/kb.py | 74 ++++++- .../coding_agent/plugins/semantic_memory.py | 61 +++++- .../topics/range_index_builder.py | 85 ++++++++ .../coding_agent/plugins/test_kb_plugin.py | 188 +++++++++++++++++- .../plugins/test_semantic_memory.py | 137 ++++++++++++- .../tests/coding_agent/test_bootstrap.py | 36 +++- .../coding_agent/test_context_system_smoke.py | 139 +++++++++++++ coding-agent/tests/deploy/test_helm_chart.py | 25 ++- 11 files changed, 758 insertions(+), 26 deletions(-) create mode 100644 coding-agent/src/coding_agent/topics/range_index_builder.py diff --git a/coding-agent/helm/templates/configmap-agent-config.yaml b/coding-agent/helm/templates/configmap-agent-config.yaml index d5ee57e66..539ee7156 100644 --- a/coding-agent/helm/templates/configmap-agent-config.yaml +++ b/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 }} diff --git a/coding-agent/helm/values.yaml b/coding-agent/helm/values.yaml index 54028cdc7..43a79d79d 100644 --- a/coding-agent/helm/values.yaml +++ b/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 diff --git a/coding-agent/src/coding_agent/core/app.py b/coding-agent/src/coding_agent/core/app.py index d3f1c2f87..cdcb24026 100644 --- a/coding-agent/src/coding_agent/core/app.py +++ b/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,20 +18,19 @@ 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 @@ -39,6 +38,7 @@ 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, diff --git a/coding-agent/src/coding_agent/plugins/kb.py b/coding-agent/src/coding_agent/plugins/kb.py index 327a62699..6adaeaec5 100644 --- a/coding-agent/src/coding_agent/plugins/kb.py +++ b/coding-agent/src/coding_agent/plugins/kb.py @@ -1,14 +1,20 @@ 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, @@ -16,12 +22,17 @@ 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 [] diff --git a/coding-agent/src/coding_agent/plugins/semantic_memory.py b/coding-agent/src/coding_agent/plugins/semantic_memory.py index 8391c47c5..9ec555306 100644 --- a/coding-agent/src/coding_agent/plugins/semantic_memory.py +++ b/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, *, diff --git a/coding-agent/src/coding_agent/topics/range_index_builder.py b/coding-agent/src/coding_agent/topics/range_index_builder.py new file mode 100644 index 000000000..94b2691b8 --- /dev/null +++ b/coding-agent/src/coding_agent/topics/range_index_builder.py @@ -0,0 +1,85 @@ +"""Build deterministic topic range indexes from durable topic stores.""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from inspect import isawaitable +from typing import Protocol, cast + +from coding_agent.topics.range_index import TopicRangeIndex +from coding_agent.topics.store import TopicRecord, TopicStatus + +logger = logging.getLogger(__name__) + + +class TopicRangeListingStore(Protocol): + async def list_topics( + self, + *, + session_id: str | None = None, + tape_id: str | None = None, + status: TopicStatus | None = None, + after_created_at: datetime | None = None, + after_topic_id: str | None = None, + limit: int = 100, + offset: int = 0, + ) -> Sequence[TopicRecord]: ... + + +@dataclass(frozen=True) +class TopicRangeIndexBuildResult: + index: TopicRangeIndex + topic_count: int + skipped_count: int = 0 + + +async def build_topic_range_index_from_store( + topic_store: object | None, + *, + max_topics: int = 10_000, +) -> TopicRangeIndexBuildResult | None: + """Derive a deterministic recall index from finalized durable topics.""" + + if topic_store is None: + return None + if ( + isinstance(max_topics, bool) + or not isinstance(max_topics, int) + or max_topics <= 0 + ): + raise ValueError("max_topics must be positive") + + list_topics = getattr(topic_store, "list_topics", None) + if not callable(list_topics): + return None + + result = list_topics(status="finalized", limit=max_topics + 1) + if not isawaitable(result): + raise TypeError("topic_store.list_topics must be async") + topics = tuple(cast(Sequence[TopicRecord], await result)) + if len(topics) > max_topics: + raise RuntimeError("too many finalized topics to build semantic topic index") + + index = TopicRangeIndex() + skipped_count = 0 + for topic in topics: + try: + index.index_topic(topic) + except ValueError as exc: + skipped_count += 1 + logger.warning( + "Skipping recall-unsafe finalized topic while building semantic topic index", + extra={ + "topic_id": topic.topic_id, + "session_id": topic.session_id, + "reason": str(exc), + }, + ) + return TopicRangeIndexBuildResult( + index=index, + topic_count=len(topics), + skipped_count=skipped_count, + ) diff --git a/coding-agent/tests/coding_agent/plugins/test_kb_plugin.py b/coding-agent/tests/coding_agent/plugins/test_kb_plugin.py index 41e96a4d6..de30647ce 100644 --- a/coding-agent/tests/coding_agent/plugins/test_kb_plugin.py +++ b/coding-agent/tests/coding_agent/plugins/test_kb_plugin.py @@ -6,10 +6,20 @@ import pytest from agentkit.observability import SpanRecord +from agentkit.runtime.messages import ( + RuntimeMessage, + RuntimeMessageKind, + SequencedRuntimeMessage, +) +from agentkit.runtime.pipeline import PipelineContext from agentkit.tape.models import Entry from agentkit.tape.tape import Tape -from coding_agent.kb import DocumentChunk, KB, KBSearchResult +from coding_agent.kb import KB, DocumentChunk, KBSearchResult from coding_agent.plugins.kb import KBPlugin +from coding_agent.plugins.semantic_memory import ( + SEMANTIC_MEMORY_GROUNDING_MARKER_KEY, + semantic_grounding_query_digest, +) def _fake_embed(texts: list[str]) -> list[list[float]]: @@ -230,6 +240,182 @@ def tracking_search(query: str, k: int = 5, *, corpora=None): assert indexed_plugin._snapshot is not None assert indexed_plugin._snapshot.last_user_msg == "What API endpoints exist?" + def test_runtime_prompt_takes_precedence_over_stale_tape_user( + self, indexed_plugin: KBPlugin + ): + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "stale README prompt"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.runtime_messages.append( + SequencedRuntimeMessage( + sequence=1, + message=RuntimeMessage( + message_id="runtime-query", + kind=RuntimeMessageKind.USER_STEER, + payload={"text": "fresh runtime prompt"}, + ), + ) + ) + assert indexed_plugin._kb is not None + captured: list[str] = [] + + def tracking_search(query: str, k: int = 5, *, corpora=None): + del k, corpora + captured.append(query) + return [] + + indexed_plugin._kb.search_sync = tracking_search + + indexed_plugin.build_context(tape=tape, ctx=ctx) + + assert captured == ["fresh runtime prompt"] + + def test_runtime_prompt_ignores_later_system_notice(self, indexed_plugin: KBPlugin): + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "stale README prompt"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.runtime_messages.extend( + [ + SequencedRuntimeMessage( + sequence=1, + message=RuntimeMessage( + message_id="runtime-query", + kind=RuntimeMessageKind.USER_STEER, + payload={"text": "fresh runtime prompt"}, + ), + ), + SequencedRuntimeMessage( + sequence=2, + message=RuntimeMessage( + message_id="system-notice", + kind=RuntimeMessageKind.SYSTEM_NOTICE, + payload={"text": "do not use as retrieval query"}, + ), + ), + ] + ) + assert indexed_plugin._kb is not None + captured: list[str] = [] + + def tracking_search(query: str, k: int = 5, *, corpora=None): + del k, corpora + captured.append(query) + return [] + + indexed_plugin._kb.search_sync = tracking_search + + indexed_plugin.build_context(tape=tape, ctx=ctx) + + assert captured == ["fresh runtime prompt"] + + def test_defers_when_semantic_memory_already_injected_context( + self, indexed_plugin: KBPlugin + ): + indexed_plugin._defer_when_semantic_memory_hits = True + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "How does auth work?"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = { + "query_digest": semantic_grounding_query_digest("How does auth work?"), + "tape_entry_count": len(tape), + "hit_count": 1, + } + assert indexed_plugin._kb is not None + + def fail_search(*args, **kwargs): + del args, kwargs + raise AssertionError("KB search should not run after semantic grounding") + + indexed_plugin._kb.search_sync = fail_search + + assert indexed_plugin.build_context(tape=tape, ctx=ctx) == [] + assert SEMANTIC_MEMORY_GROUNDING_MARKER_KEY not in ctx.config + + def test_defer_flag_does_not_skip_when_semantic_memory_has_no_hits( + self, indexed_plugin: KBPlugin + ): + indexed_plugin._defer_when_semantic_memory_hits = True + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "How does auth work?"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = { + "query_digest": semantic_grounding_query_digest("How does auth work?"), + "tape_entry_count": len(tape), + "hit_count": 0, + } + + result = indexed_plugin.build_context(tape=tape, ctx=ctx) + + assert len(result) == 1 + assert "## Repo references" in result[0]["content"] + + def test_defer_flag_does_not_skip_on_stale_semantic_marker( + self, indexed_plugin: KBPlugin + ): + indexed_plugin._defer_when_semantic_memory_hits = True + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "How does auth work?"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = { + "query_digest": semantic_grounding_query_digest("previous prompt"), + "tape_entry_count": len(tape), + "hit_count": 1, + } + + result = indexed_plugin.build_context(tape=tape, ctx=ctx) + + assert len(result) == 1 + assert "## Repo references" in result[0]["content"] + assert SEMANTIC_MEMORY_GROUNDING_MARKER_KEY not in ctx.config + + def test_defer_flag_does_not_skip_on_invalid_semantic_hit_count( + self, indexed_plugin: KBPlugin + ): + indexed_plugin._defer_when_semantic_memory_hits = True + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "How does auth work?"}, + ) + ) + ctx = PipelineContext(tape=tape) + ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = { + "query_digest": semantic_grounding_query_digest("How does auth work?"), + "tape_entry_count": len(tape), + "hit_count": True, + } + + result = indexed_plugin.build_context(tape=tape, ctx=ctx) + + assert len(result) == 1 + assert "## Repo references" in result[0]["content"] + def test_search_corpora_passed_to_kb_search(self, tmp_path: Path): db_path = tmp_path / "kb_db" kb = KB( diff --git a/coding-agent/tests/coding_agent/plugins/test_semantic_memory.py b/coding-agent/tests/coding_agent/plugins/test_semantic_memory.py index 659045906..ceef352e1 100644 --- a/coding-agent/tests/coding_agent/plugins/test_semantic_memory.py +++ b/coding-agent/tests/coding_agent/plugins/test_semantic_memory.py @@ -5,17 +5,20 @@ import pytest from agentkit.plugin.registry import PluginRegistry +from agentkit.runtime.hook_runtime import HookRuntime from agentkit.runtime.messages import ( RuntimeMessage, RuntimeMessageKind, SequencedRuntimeMessage, ) -from agentkit.runtime.pipeline import PipelineContext -from agentkit.runtime.hook_runtime import HookRuntime -from agentkit.runtime.pipeline import Pipeline +from agentkit.runtime.pipeline import Pipeline, PipelineContext from agentkit.tape.models import Entry from agentkit.tape.tape import Tape -from coding_agent.plugins.semantic_memory import SemanticMemoryPlugin +from coding_agent.plugins.semantic_memory import ( + SEMANTIC_MEMORY_GROUNDING_MARKER_KEY, + SemanticMemoryPlugin, + semantic_grounding_query_digest, +) from coding_agent.topics.memory import ( MemoryReviewStore, TopicDerivedMemoryCandidate, @@ -35,11 +38,20 @@ class FakeTopicStore: def __init__(self, topics: tuple[TopicRecord, ...]) -> None: self.topics = {topic.topic_id: topic for topic in topics} self.loaded: list[str] = [] + self.listed: list[dict[str, object]] = [] async def load_topic(self, topic_id: str) -> TopicRecord | None: self.loaded.append(topic_id) return self.topics.get(topic_id) + async def list_topics(self, **kwargs: object) -> tuple[TopicRecord, ...]: + self.listed.append(dict(kwargs)) + return tuple( + topic + for topic in self.topics.values() + if topic.status == kwargs.get("status") + ) + @pytest.mark.asyncio async def test_build_context_rehydrates_accepted_memory_without_rendering_hit_text() -> ( @@ -134,6 +146,123 @@ async def test_build_context_rehydrates_topic_hits_from_authoritative_store() -> assert "jwt middleware backend sentinel must never render" not in rendered +@pytest.mark.asyncio +async def test_build_context_derives_topic_range_index_from_topic_store() -> None: + topic = _topic( + "topic-auth", + title="Auth gateway", + summary="Authoritative topic summary says JWT middleware lives in auth gateway.", + ) + topic_store = FakeTopicStore((topic,)) + index = SafeSemanticMemoryIndex(FakeSemanticMemoryBackend()) + plugin = SemanticMemoryPlugin( + semantic_index=index, + memory_review_store=MemoryReviewStore(), + read_enabled=True, + topic_store=topic_store, + ) + + tape = _tape("jwt middleware") + result = await plugin.build_context(tape=tape) + + assert topic_store.listed == [{"status": "finalized", "limit": 10001}] + assert result + rendered = result[0]["content"] + assert "Cross-topic recall references" in rendered + assert "Authoritative topic summary says JWT middleware lives in auth gateway." in ( + rendered + ) + + second_result = await plugin.build_context(tape=tape) + + assert second_result == result + assert topic_store.listed == [{"status": "finalized", "limit": 10001}] + + +@pytest.mark.asyncio +async def test_build_context_skips_recall_unsafe_topics_from_derived_index() -> None: + safe_topic = _topic( + "topic-auth", + title="Auth gateway", + summary="JWT middleware lives in auth gateway.", + ) + unsafe_topic = _topic( + "topic-unsafe", + title="Unsafe topic", + summary="stdout: raw command output must not enter recall context", + ) + topic_store = FakeTopicStore((unsafe_topic, safe_topic)) + index = SafeSemanticMemoryIndex(FakeSemanticMemoryBackend()) + plugin = SemanticMemoryPlugin( + semantic_index=index, + memory_review_store=MemoryReviewStore(), + read_enabled=True, + topic_store=topic_store, + ) + + result = await plugin.build_context(tape=_tape("jwt middleware")) + + assert topic_store.listed == [{"status": "finalized", "limit": 10001}] + assert result + rendered = result[0]["content"] + assert "JWT middleware lives in auth gateway." in rendered + assert "stdout: raw command output" not in rendered + + +@pytest.mark.asyncio +async def test_build_context_records_grounding_hit_count_in_pipeline_context() -> None: + topic = _topic( + "topic-auth", + title="Auth gateway", + summary="JWT middleware lives in auth gateway.", + ) + index = SafeSemanticMemoryIndex(FakeSemanticMemoryBackend()) + await index.upsert( + SemanticMemoryDocument( + memory_id=SemanticDocId.for_topic(topic), + text="jwt middleware", + metadata={"kind": "topic_summary"}, + source_refs=(SemanticSourceRef.for_topic(topic),), + ) + ) + plugin = SemanticMemoryPlugin( + semantic_index=index, + memory_review_store=MemoryReviewStore(), + read_enabled=True, + topic_store=FakeTopicStore((topic,)), + ) + tape = _tape("jwt middleware") + ctx = PipelineContext(tape=tape, session_id="session-runtime") + + await plugin.build_context(tape=tape, ctx=ctx) + + assert ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] == { + "query_digest": semantic_grounding_query_digest("jwt middleware"), + "tape_entry_count": len(tape), + "hit_count": 1, + } + + +@pytest.mark.asyncio +async def test_build_context_clears_stale_grounding_hit_count() -> None: + plugin = SemanticMemoryPlugin( + semantic_index=SafeSemanticMemoryIndex(FakeSemanticMemoryBackend()), + memory_review_store=MemoryReviewStore(), + read_enabled=True, + ) + tape = Tape(tape_id="tape-semantic") + ctx = PipelineContext(tape=tape, session_id="session-runtime") + ctx.config[SEMANTIC_MEMORY_GROUNDING_MARKER_KEY] = { + "query_digest": "stale", + "tape_entry_count": 1, + "hit_count": 3, + } + + await plugin.build_context(tape=tape, ctx=ctx) + + assert SEMANTIC_MEMORY_GROUNDING_MARKER_KEY not in ctx.config + + @pytest.mark.asyncio async def test_build_context_uses_runtime_prompt_when_tape_has_no_user_message() -> ( None diff --git a/coding-agent/tests/coding_agent/test_bootstrap.py b/coding-agent/tests/coding_agent/test_bootstrap.py index d530237b3..b71452638 100644 --- a/coding-agent/tests/coding_agent/test_bootstrap.py +++ b/coding-agent/tests/coding_agent/test_bootstrap.py @@ -2,12 +2,13 @@ from typing import Any, cast import pytest + from agentkit.environment import WorkspaceSummary from agentkit.observability import NoopObservationSink from agentkit.runtime import AgentRunContext, ContextBudget from agentkit.runtime.pipeline import PipelineContext -from agentkit.tools import tool from agentkit.tape.tape import Tape +from agentkit.tools import tool from coding_agent.__main__ import create_agent, create_child_pipeline from coding_agent.environment import ( CloudCommandResult, @@ -245,6 +246,7 @@ def test_kb_plugin_requires_explicit_enablement(self, tmp_path): index_extensions = [".md"] corpus = "sre" search_corpora = ["sre", "notes"] +defer_when_semantic_memory_hits = true """.strip() ) @@ -260,6 +262,7 @@ def test_kb_plugin_requires_explicit_enablement(self, tmp_path): assert kb_plugin._max_distance == 0.8 assert kb_plugin._corpus == "sre" assert kb_plugin._search_corpora == ("sre", "notes") + assert kb_plugin._defer_when_semantic_memory_hits is True def test_create_agent_installs_persistent_memory_review_store(self, tmp_path): config_path = tmp_path / "agent.toml" @@ -446,6 +449,37 @@ def test_create_agent_respects_enabled_plugins_order(self, tmp_path): assert pipeline._registry.plugin_ids() == ["storage", "core_tools"] + def test_create_agent_rejects_kb_defer_before_semantic_memory(self, tmp_path): + config_path = tmp_path / "agent.toml" + config_path.write_text( + """ +[agent] +name = "test-agent" +model = "claude-sonnet-4-20250514" +provider = "anthropic" + +[agent.plugins] +enabled = ["kb", "semantic_memory"] + +[memory.semantic] +enabled = true +backend = "fake" + +[kb] +defer_when_semantic_memory_hits = true +""".strip() + ) + + with pytest.raises( + ValueError, + match="requires semantic_memory to run before kb", + ): + create_agent( + config_path=config_path, + data_dir=tmp_path / "data", + api_key="sk-test", + ) + def test_create_agent_uses_mcp_servers_override(self, tmp_path): config_path = tmp_path / "agent.toml" config_path.write_text( diff --git a/coding-agent/tests/coding_agent/test_context_system_smoke.py b/coding-agent/tests/coding_agent/test_context_system_smoke.py index 6a679dfe6..f68d0393b 100644 --- a/coding-agent/tests/coding_agent/test_context_system_smoke.py +++ b/coding-agent/tests/coding_agent/test_context_system_smoke.py @@ -2,11 +2,17 @@ from __future__ import annotations +from datetime import UTC, datetime from pathlib import Path import pytest from agentkit.plugin.registry import PluginRegistry +from agentkit.runtime.messages import ( + RuntimeMessage, + RuntimeMessageKind, + SequencedRuntimeMessage, +) from agentkit.runtime.hook_runtime import HookRuntime from agentkit.runtime.hookspecs import HOOK_SPECS from agentkit.runtime.pipeline import Pipeline, PipelineContext @@ -15,6 +21,24 @@ from coding_agent.kb import KB from coding_agent.plugins.kb import KBPlugin from coding_agent.plugins.memory import MemoryPlugin +from coding_agent.plugins.semantic_memory import SemanticMemoryPlugin +from coding_agent.topics.memory import MemoryReviewStore +from coding_agent.topics.semantic_backends import FakeSemanticMemoryBackend +from coding_agent.topics.semantic_index import ( + SafeSemanticMemoryIndex, + SemanticDocId, + SemanticMemoryDocument, + SemanticSourceRef, +) +from coding_agent.topics.store import TopicRecord + + +class FakeTopicStore: + def __init__(self, topics: tuple[TopicRecord, ...]) -> None: + self._topics = {topic.topic_id: topic for topic in topics} + + async def load_topic(self, topic_id: str) -> TopicRecord | None: + return self._topics.get(topic_id) @pytest.mark.asyncio @@ -122,6 +146,92 @@ async def test_context_system_smoke_combines_retrieval_failure_and_memory( assert "[Memory]" not in rendered +@pytest.mark.asyncio +async def test_semantic_memory_and_kb_use_same_runtime_prompt_to_avoid_stale_kb( + tmp_path: Path, +) -> None: + db_path = tmp_path / "kb-db" + kb = KB( + db_path=db_path, + embedding_dim=4, + embedding_fn=_fresh_old_embed, + chunk_size=1000, + chunk_overlap=0, + ) + await kb.index_file( + Path("README.md"), + "Old README snapshot says the deployed image tag is legacy-o6n.", + ) + + topic = _topic( + "topic-o6n-image", + title="o6n deployed image", + summary="Fresh topic summary says the deployed image tag is new-o6n.", + ) + semantic_index = SafeSemanticMemoryIndex(FakeSemanticMemoryBackend()) + await semantic_index.upsert( + SemanticMemoryDocument( + memory_id=SemanticDocId.for_topic(topic), + text="fresh runtime deployed image tag new-o6n", + metadata={"kind": "topic_summary"}, + source_refs=(SemanticSourceRef.for_topic(topic),), + ) + ) + semantic_memory = SemanticMemoryPlugin( + semantic_index=semantic_index, + memory_review_store=MemoryReviewStore(), + read_enabled=True, + topic_store=FakeTopicStore((topic,)), + ) + kb_plugin = KBPlugin( + db_path=db_path, + embedding_dim=4, + top_k=3, + max_distance=0.01, + embedding_fn=_fresh_old_embed, + ) + registry = PluginRegistry(specs=HOOK_SPECS) + registry.register(semantic_memory) + registry.register(kb_plugin) + runtime = HookRuntime(registry, specs=HOOK_SPECS) + pipeline = Pipeline(runtime=runtime, registry=registry) + tape = Tape() + tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "old README deployed image"}, + ) + ) + ctx = PipelineContext( + tape=tape, + session_id="session-smoke", + config={"system_prompt": "You are a test agent."}, + ) + ctx.runtime_messages.append( + SequencedRuntimeMessage( + sequence=1, + message=RuntimeMessage( + message_id="runtime-query", + kind=RuntimeMessageKind.USER_STEER, + payload={"text": "fresh runtime deployed image tag"}, + ), + ) + ) + + await pipeline.mount(ctx) + await pipeline._stage_build_context(ctx) + + rendered = "\n".join( + message["content"] + for message in ctx.messages + if message.get("role") == "system" + ) + assert "Fresh topic summary says the deployed image tag is new-o6n." in rendered + assert "Old README snapshot says the deployed image tag is legacy-o6n." not in ( + rendered + ) + + def _context_system_embed(texts: list[str]) -> list[list[float]]: vectors: list[list[float]] = [] for text in texts: @@ -135,6 +245,19 @@ def _context_system_embed(texts: list[str]) -> list[list[float]]: return vectors +def _fresh_old_embed(texts: list[str]) -> list[list[float]]: + vectors: list[list[float]] = [] + for text in texts: + lower = text.lower() + if "old" in lower or "legacy" in lower or "readme" in lower: + vectors.append([1.0, 0.0, 0.0, 0.0]) + elif "fresh" in lower or "new-o6n" in lower: + vectors.append([0.0, 1.0, 0.0, 0.0]) + else: + vectors.append([0.0, 0.0, 1.0, 0.0]) + return vectors + + def _repo_file_evidence(repo_path: str) -> dict[str, str]: return { "kind": "repo_file", @@ -142,3 +265,19 @@ def _repo_file_evidence(repo_path: str) -> dict[str, str]: "label": repo_path, "repo_path": repo_path, } + + +def _topic(topic_id: str, *, title: str, summary: str) -> TopicRecord: + return TopicRecord( + topic_id=topic_id, + tape_id="tape-topic", + session_id="session-topic", + kind="coding", + status="finalized", + title=title, + summary=summary, + owner=None, + topic_initial_seq=2, + topic_finalized_seq=9, + created_at=datetime(2026, 6, 29, tzinfo=UTC), + ) diff --git a/coding-agent/tests/deploy/test_helm_chart.py b/coding-agent/tests/deploy/test_helm_chart.py index 3c47c8ab5..2e1656d91 100644 --- a/coding-agent/tests/deploy/test_helm_chart.py +++ b/coding-agent/tests/deploy/test_helm_chart.py @@ -13,7 +13,6 @@ from coding_agent.__main__ import create_agent - ROOT = Path(__file__).resolve().parents[2] CHART = ROOT / "helm" @@ -247,6 +246,18 @@ def test_helm_semantic_memory_enabled_renders_config() -> None: } +def test_helm_semantic_memory_runs_before_kb_when_both_enabled() -> None: + docs = _render( + "--set", + "memory.semantic.enabled=true", + "--set", + "kb.enabled=true", + ) + plugins = _agent_toml(docs)["agent"]["plugins"]["enabled"] + + assert plugins.index("semantic_memory") < plugins.index("kb") + + def test_helm_semantic_memory_enabled_bootstraps_runtime(tmp_path: Path) -> None: data_mount = tmp_path / "data" docs = _render( @@ -894,6 +905,7 @@ def test_helm_kb_enabled_renders_config_and_plugin() -> None: "embedding_model": "BAAI/bge-m3", "embedding_base_url": "https://api.siliconflow.cn/v1", "embedding_dim": 1024, + "defer_when_semantic_memory_hits": False, "corpus": "notes", "search_corpora": ["sre", "notes"], } @@ -905,6 +917,17 @@ def test_helm_kb_enabled_can_render_optional_max_distance() -> None: assert _agent_toml(docs)["kb"]["max_distance"] == 0.42 +def test_helm_kb_enabled_can_defer_when_semantic_memory_hits() -> None: + docs = _render( + "--set", + "kb.enabled=true", + "--set", + "kb.deferWhenSemanticMemoryHits=true", + ) + + assert _agent_toml(docs)["kb"]["defer_when_semantic_memory_hits"] is True + + def test_helm_kb_index_disabled_does_not_render_cronjob() -> None: docs = _render("--set", "kb.enabled=true", "--set", "kb.index.enabled=false") From 4a3f54ff6d9364182ddb1f6622f531cf2312d268 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 30 Jun 2026 11:07:40 +0800 Subject: [PATCH 2/2] test(memory): align semantic smoke topic store Add list_topics to the smoke FakeTopicStore so the test double matches the lazy semantic topic-index contract. Verification: uv run pytest tests/coding_agent/test_context_system_smoke.py tests/coding_agent/plugins/test_semantic_memory.py tests/coding_agent/plugins/test_kb_plugin.py -v Verification: uv run ruff check tests/coding_agent/test_context_system_smoke.py && uv run ruff format --check tests/coding_agent/test_context_system_smoke.py --- .../tests/coding_agent/test_context_system_smoke.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/coding-agent/tests/coding_agent/test_context_system_smoke.py b/coding-agent/tests/coding_agent/test_context_system_smoke.py index f68d0393b..104135d57 100644 --- a/coding-agent/tests/coding_agent/test_context_system_smoke.py +++ b/coding-agent/tests/coding_agent/test_context_system_smoke.py @@ -40,6 +40,13 @@ def __init__(self, topics: tuple[TopicRecord, ...]) -> None: async def load_topic(self, topic_id: str) -> TopicRecord | None: return self._topics.get(topic_id) + async def list_topics(self, **kwargs: object) -> tuple[TopicRecord, ...]: + return tuple( + topic + for topic in self._topics.values() + if topic.status == kwargs.get("status") + ) + @pytest.mark.asyncio async def test_context_system_smoke_combines_retrieval_failure_and_memory(