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..104135d57 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,31 @@ 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) + + 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 @@ -122,6 +153,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 +252,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 +272,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")