diff --git a/docs/en/docs/operate/configuration.md b/docs/en/docs/operate/configuration.md index 0bb5f8ec8f..39555fc615 100644 --- a/docs/en/docs/operate/configuration.md +++ b/docs/en/docs/operate/configuration.md @@ -140,6 +140,7 @@ Server settings use the `POWERCONTEXT_SERVER_` prefix. | `POWERCONTEXT_SERVER_RUNTIME_RECALL_GATE_ALLOW_WITH_RERANK` | `false` | Whether expansion may still run when `MEMORY_RERANK_ENABLED` is set; by default reranking ends the expansion | | `POWERCONTEXT_SERVER_RUNTIME_RECALL_FTS_WEIGHT` | `1.0` | Relative FTS contribution in hybrid Atomic Memory and Topic Memory RRF; `0` excludes FTS-only matches from hybrid results | | `POWERCONTEXT_SERVER_RUNTIME_RECALL_VECTOR_WEIGHT` | `1.0` | Relative vector contribution in hybrid Atomic Memory and Topic Memory RRF; `0` excludes vector-only matches from hybrid results | +| `POWERCONTEXT_SERVER_RUNTIME_RECALL_EMBEDDING_TIMEOUT_SECONDS` | `0.25` | Seconds each Atomic Memory or Topic Memory search inside a context preparation waits for the query embedding, at most `60`. A preparation searches the Scope and each of its context references in turn, so the waits add up. When a budget expires, that search continues with lexical matching only and logs `memory.search.embedding_fallback` or `topic_memory.search.embedding_fallback`. A remote embedding provider needs more than the default; keep the summed waits inside the host's Hook request timeout, 3 seconds on most hosts | | `POWERCONTEXT_SERVER_RUNTIME_MEMORY_SCHEDULE_SECONDS` | unset | Memory automatic admission interval; `SCHEDULE_SECONDS` remains a compatibility alias | | `POWERCONTEXT_SERVER_RUNTIME_TOPIC_MEMORY_SCHEDULE_SECONDS` | unset | Topic Memory automatic admission interval; unset disables new automatic admission | | `POWERCONTEXT_SERVER_RUNTIME_TOPIC_MEMORY_SOURCE_WINDOW_LIMIT` | `10` | Maximum Sources per Topic Memory Window, capped at 100; one Scope invocation can finish several Windows | diff --git a/docs/zh/docs/operate/configuration.md b/docs/zh/docs/operate/configuration.md index 0f7080fc8a..ccf9fe48c4 100644 --- a/docs/zh/docs/operate/configuration.md +++ b/docs/zh/docs/operate/configuration.md @@ -131,6 +131,7 @@ Server 配置使用 `POWERCONTEXT_SERVER_` 前缀。 | `POWERCONTEXT_SERVER_RUNTIME_RECALL_GATE_ALLOW_WITH_RERANK` | `false` | 已启用 `MEMORY_RERANK_ENABLED` 时是否仍允许追加搜索；默认在 rerank 之后不再追加 | | `POWERCONTEXT_SERVER_RUNTIME_RECALL_FTS_WEIGHT` | `1.0` | hybrid Atomic Memory 和 Topic Memory RRF 中 FTS 的相对贡献；`0` 会从 hybrid 结果中排除仅命中 FTS 的候选 | | `POWERCONTEXT_SERVER_RUNTIME_RECALL_VECTOR_WEIGHT` | `1.0` | hybrid Atomic Memory 和 Topic Memory RRF 中 vector 的相对贡献；`0` 会从 hybrid 结果中排除仅命中 vector 的候选 | +| `POWERCONTEXT_SERVER_RUNTIME_RECALL_EMBEDDING_TIMEOUT_SECONDS` | `0.25` | 一次上下文准备里每次 Atomic Memory 或 Topic Memory 检索等待查询 embedding 的秒数，上限 `60`。准备会依次检索该 Scope 及其每个 context reference，等待时间会累加。某次等待超时后，该次检索只用词法匹配继续，并记录 `memory.search.embedding_fallback` 或 `topic_memory.search.embedding_fallback`。远程 embedding 提供商需要比默认值更长的时间；累计等待应控制在宿主 Hook 请求超时之内，多数宿主为 3 秒 | | `POWERCONTEXT_SERVER_RUNTIME_MEMORY_SCHEDULE_SECONDS` | 未设置 | Memory 自动准入间隔；`SCHEDULE_SECONDS` 保留为兼容别名 | | `POWERCONTEXT_SERVER_RUNTIME_TOPIC_MEMORY_SCHEDULE_SECONDS` | 未设置 | Topic Memory 自动准入间隔；未设置时不接纳新的自动调用 | | `POWERCONTEXT_SERVER_RUNTIME_TOPIC_MEMORY_SOURCE_WINDOW_LIMIT` | `10` | 每个 Topic Memory Window 的 Source 数量上限，硬上限为 100；一次 Scope 调用可完成多个 Window | diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index 80c36bc4c3..5260afa7d0 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -364,10 +364,6 @@ def __call__( logger = logging.getLogger(__name__) -# Leave room for database reads and assembly within the default one-second Hook request. -_CONTEXT_TOPIC_EMBEDDING_TIMEOUT_SECONDS = 0.25 -_CONTEXT_MEMORY_EMBEDDING_TIMEOUT_SECONDS = 0.25 - _MEMORY_CAPTURE_STAGE = "memory.capture" _MEMORY_CAPTURE_SOURCE_COUNT = "powercontext.memory.capture.source_count" _WORKING_NOTE_RESERVED_METADATA = frozenset({"kind", "session_id", "note_kind", "expires_at"}) @@ -1561,7 +1557,7 @@ async def _recall_scope( mode="auto", admission=admission, query_embedding=reuse, - embedding_timeout_seconds=_CONTEXT_MEMORY_EMBEDDING_TIMEOUT_SECONDS, + embedding_timeout_seconds=self._runtime.recall_embedding_timeout_seconds, allow_embedding=allow_embedding, recovery_admission=None if policy is None else policy.round2_admission, context=_PREPARE_ATOMIC_CONTEXT.get(), @@ -1650,7 +1646,7 @@ async def _topic_memory_hits( SearchTopicMemoryRequest(query=bounded_query, limit=limit), admission=admission, query_embedding=reuse, - embedding_timeout_seconds=_CONTEXT_TOPIC_EMBEDDING_TIMEOUT_SECONDS, + embedding_timeout_seconds=self._runtime.recall_embedding_timeout_seconds, allow_embedding=allow_embedding, ) ) @@ -3630,6 +3626,7 @@ def __init__( source_window_limit: int = 100, context_assembly_max_entries: int = 8, recall_sufficiency_policy: RecallSufficiencyPolicy | None = None, + recall_embedding_timeout_seconds: float = 0.25, scope_cache_size: int = DEFAULT_SCOPE_CACHE_SIZE, scope_evictor: ScopeEvictor | None = None, scope_cache_observer: ScopeCacheObserver | None = None, @@ -3687,6 +3684,8 @@ def __init__( raise _RuntimeConfigurationError("source_window_limit") if context_assembly_max_entries < 1: raise _RuntimeConfigurationError("context_assembly_max_entries") + if not recall_embedding_timeout_seconds > 0: + raise _RuntimeConfigurationError("recall_embedding_timeout_seconds") if scope_cache_size < 1: raise _RuntimeConfigurationError("scope_cache_size") self._provider = provider @@ -3768,6 +3767,7 @@ def __init__( self.source_window_limit = source_window_limit self.context_assembly_max_entries = context_assembly_max_entries self.recall_sufficiency_policy = recall_sufficiency_policy + self.recall_embedding_timeout_seconds = recall_embedding_timeout_seconds self._scope_cache = ScopeCache( scope_cache_size, evictor=scope_evictor, diff --git a/src/powercontext/builtin/runtime/atomic_memory.py b/src/powercontext/builtin/runtime/atomic_memory.py index 0aad55967b..177b52af4e 100644 --- a/src/powercontext/builtin/runtime/atomic_memory.py +++ b/src/powercontext/builtin/runtime/atomic_memory.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import logging from contextlib import asynccontextmanager, nullcontext from dataclasses import dataclass, replace from hashlib import sha256 @@ -29,6 +30,7 @@ from sqlalchemy.ext.asyncio import AsyncConnection from typing_extensions import override +from powercontext._logging import log_safely from powercontext.artifacts import ArtifactRef, ArtifactSearchContractError, ArtifactSearchMatch from powercontext.artifacts.search import ArtifactSearchExecutionContext from powercontext.builtin.artifacts.atomic_memory.errors import ( @@ -89,6 +91,27 @@ from powercontext.builtin.runtime.artifact_merge import ArtifactMergeAccess, merge_artifact_tags from powercontext.builtin.tags import TagFilter +logger = logging.getLogger(__name__) + + +def _log_embedding_fallback(error: Exception) -> None: + """Record that a bounded query embedding gave way to lexical retrieval; the query itself is never logged.""" + + log_safely( + logger, + logging.WARNING, + "Memory search fell back to lexical retrieval", + extra={ + "event": "memory.search.embedding_fallback", + "outcome": "fallback", + "mode": "text", + "error_code": "inference_timeout" + if isinstance(error, (InferenceTimeoutError, TimeoutError)) + else "inference_unavailable", + "unit": "memory", + }, + ) + class AtomicMemoryAccess(ArtifactMergeAccess): def __init__(self) -> None: @@ -665,6 +688,7 @@ async def _resolve_query_embedding( raise InferenceUnavailableError("embedding", "Atomic query embedding timed out") from exc if requested_mode != "auto" or not capabilities.fts: raise + _log_embedding_fallback(exc) return "text", None, None, 1 if len(result.vectors) != 1: raise AtomicMemoryIndexError("embedding-result", "Expected one query vector") diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index f3e1b03f99..827c122959 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -788,6 +788,7 @@ async def run_profile(scope_id, high): ), context_assembly_max_entries=config.runtime.context_assembly_max_entries, recall_sufficiency_policy=RecallSufficiencyPolicy.from_runtime_config(config.runtime), + recall_embedding_timeout_seconds=config.runtime.recall_embedding_timeout_seconds, scope_cache_size=config.runtime.scope_cache_size, scope_evictor=contexts.evict, scope_cache_observer=scope_cache_observer, diff --git a/src/powercontext/builtin/runtime/config.py b/src/powercontext/builtin/runtime/config.py index a58ad106e2..32347ba2ce 100644 --- a/src/powercontext/builtin/runtime/config.py +++ b/src/powercontext/builtin/runtime/config.py @@ -180,6 +180,11 @@ def validate_memory_capacity_order(self) -> RuntimeConfig: recall_gate_allow_with_rerank: bool = False recall_fts_weight: float = Field(default=1.0, ge=0.0, allow_inf_nan=False) recall_vector_weight: float = Field(default=1.0, ge=0.0, allow_inf_nan=False) + # Budget for embedding the query in each Atomic Memory or Topic Memory search of a context preparation. A + # preparation searches the Scope and each context reference in turn, so the waits add up inside the Hook + # request; when a budget expires, that search continues with lexical matching only. A remote embedding + # provider needs more than the default. + recall_embedding_timeout_seconds: float = Field(default=0.25, gt=0.0, le=60.0, allow_inf_nan=False) profile_schedule_enabled: bool = False profile_cron: str = "0 2 * * *" profile_timezone: str = "Asia/Shanghai" @@ -209,6 +214,13 @@ def validate_recall_gate_threshold_order(self): raise ValueError(_RECALL_GATE_ROUND2_ORDER_ERROR) return self + @field_validator("recall_embedding_timeout_seconds", mode="before") + @classmethod + def reject_boolean_recall_embedding_timeout(cls, value: Any) -> Any: + if isinstance(value, bool): + raise ValueError("recall embedding timeout must be a number") # noqa: TRY003, TRY004 - Pydantic validation + return value + @field_validator("recall_fts_weight", "recall_vector_weight", mode="before") @classmethod def reject_boolean_recall_weight(cls, value: Any) -> Any: diff --git a/tests/builtin/runtime/test_recall_embedding_timeout_config.py b/tests/builtin/runtime/test_recall_embedding_timeout_config.py new file mode 100644 index 0000000000..482e3720d7 --- /dev/null +++ b/tests/builtin/runtime/test_recall_embedding_timeout_config.py @@ -0,0 +1,41 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The recall embedding budget is a bounded runtime setting with an environment mapping.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pydantic import ValidationError + +from powercontext.builtin.runtime.config import RuntimeConfig +from powercontext.server.settings import ServerSettings + + +def test_recall_embedding_timeout_defaults_to_the_hook_sized_budget() -> None: + assert RuntimeConfig().recall_embedding_timeout_seconds == 0.25 + + +@pytest.mark.parametrize("value", [0, -1, 61, float("nan"), True]) +def test_recall_embedding_timeout_rejects_values_outside_the_budget_range(value: Any) -> None: + with pytest.raises(ValidationError): + RuntimeConfig(recall_embedding_timeout_seconds=value) + + +def test_recall_embedding_timeout_is_read_from_the_server_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("POWERCONTEXT_SERVER_RUNTIME_RECALL_EMBEDDING_TIMEOUT_SECONDS", "1.5") + + assert ServerSettings().runtime.recall_embedding_timeout_seconds == 1.5 diff --git a/tests/e2e/test_recall_embedding_timeout.py b/tests/e2e/test_recall_embedding_timeout.py new file mode 100644 index 0000000000..22de698c25 --- /dev/null +++ b/tests/e2e/test_recall_embedding_timeout.py @@ -0,0 +1,186 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The configured recall embedding budget decides whether a context preparation waits for a slow embedding. + +The query embedding of ``prepare_context`` is bounded so a Hook request stays bounded. With a remote provider the +round trip can exceed the default, and retrieval then continues with lexical matching only. The budget is a +runtime setting; these scenarios show it reaching the Atomic Memory search and being reported when it expires. +""" + +from __future__ import annotations + +import asyncio +import json +import math +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any + +import pytest + +from powercontext.builtin.artifacts.atomic_memory import AtomicMemoryContent +from powercontext.builtin.artifacts.memory import EmbeddingProfile +from powercontext.builtin.inference import EmbeddingResult, InferenceTimeoutError, InferenceUnavailableError +from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.runtime import BuiltinConfig, PrepareContextRequest, RuntimeConfig, open_builtin_runtime +from powercontext.builtin.runtime.application import BuiltinRuntime +from powercontext.builtin.scope import ScopeDraft +from powercontext.server.logging import JsonFormatter + +_QUERY = "alpha beta gamma" +_MEMORY = "alpha beta gamma memory" +_EVENT = "memory.search.embedding_fallback" +_PROFILE = EmbeddingProfile( + profile_id="recall-budget", model="deterministic", dimension=2, distance="l2", normalization="unit" +) + + +class _FailingQueryEmbedding: + """Embed entries at once but fail the query embedding with ``error``.""" + + profile = _PROFILE + + def __init__(self, error: Exception) -> None: + self.error = error + + async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: + if any(text == _QUERY for text in texts): + raise self.error + return EmbeddingResult(vectors=tuple((0.3, math.sqrt(1.0 - 0.3**2)) for _ in texts)) + + +class _SlowQueryEmbedding: + """Embed entries at once but hold the query embedding for ``delay`` seconds.""" + + profile = _PROFILE + + def __init__(self, delay: float) -> None: + self.delay = delay + self.query_completed = False + + async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: + if any(text == _QUERY for text in texts): + await asyncio.sleep(self.delay) + self.query_completed = True + similarities = tuple(1.0 if text.endswith(_QUERY) or text == _QUERY else 0.3 for text in texts) + return EmbeddingResult(vectors=tuple((value, math.sqrt(1.0 - value**2)) for value in similarities)) + + +@asynccontextmanager +async def _runtime(database: Path, budget: float, embedding: Any) -> AsyncIterator[BuiltinRuntime]: + config = BuiltinConfig( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{database}"), + runtime=RuntimeConfig(recall_embedding_timeout_seconds=budget), + ) + async with open_builtin_runtime( + config, scheduler_path=database.with_suffix(".scheduler.db"), embedding_model=embedding + ) as runtime: + yield runtime + + +async def _scope_with_memory(runtime: BuiltinRuntime, key: str) -> str: + assert runtime.scopes is not None + assert runtime.atomic_memory is not None + created = await runtime.scopes.create( + ScopeDraft(title="Budget", summary="Recall embedding budget", idempotency_key=key) + ) + await runtime.atomic_memory.for_scope(created.scope_id).create((AtomicMemoryContent(kind="fact", text=_MEMORY),)) + return created.scope_id + + +def _request() -> PrepareContextRequest: + return PrepareContextRequest.model_validate({ + "query": _QUERY, + "max_bytes": 8000, + "assembly": {"sections": [{"family": "memory", "limit": 8}]}, + }) + + +def _fallbacks(caplog: pytest.LogCaptureFixture) -> list[Any]: + return [record for record in caplog.records if getattr(record, "event", None) == _EVENT] + + +def _assert_fallback_record(record: Any, error_code: str) -> None: + """The event classifies the failure and carries no query, exception or provider text.""" + + serialized = JsonFormatter().format(record) + payload = json.loads(serialized) + assert record.levelname == "WARNING" + assert record.exc_info is None + assert payload["event"] == _EVENT + assert payload["outcome"] == "fallback" + assert payload["mode"] == "text" + assert payload["unit"] == "memory" + assert payload["error_code"] == error_code + assert "exception" not in payload + assert _QUERY not in serialized + + +def test_an_expired_budget_keeps_lexical_recall_and_reports_the_fallback( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + embedding = _SlowQueryEmbedding(delay=0.5) + + async def scenario() -> Any: + async with _runtime(tmp_path / "expired.db", 0.05, embedding) as runtime: + scope_id = await _scope_with_memory(runtime, "expired") + return await runtime.context.for_scope(scope_id).prepare(_request()) + + prepared = asyncio.run(scenario()) + + assert prepared.status == "ready" + assert _MEMORY in (prepared.content or "") + assert embedding.query_completed is False + (record,) = _fallbacks(caplog) + _assert_fallback_record(record, "inference_timeout") + + +@pytest.mark.parametrize( + ("error", "error_code"), + [ + (InferenceUnavailableError("embed"), "inference_unavailable"), + (InferenceTimeoutError("embed", 1.0), "inference_timeout"), + ], +) +def test_a_failing_provider_keeps_lexical_recall_and_classifies_the_failure( + tmp_path: Path, caplog: pytest.LogCaptureFixture, error: Exception, error_code: str +) -> None: + async def scenario() -> Any: + async with _runtime(tmp_path / "failing.db", 5.0, _FailingQueryEmbedding(error)) as runtime: + scope_id = await _scope_with_memory(runtime, "failing") + return await runtime.context.for_scope(scope_id).prepare(_request()) + + prepared = asyncio.run(scenario()) + + assert prepared.status == "ready" + assert _MEMORY in (prepared.content or "") + (record,) = _fallbacks(caplog) + _assert_fallback_record(record, error_code) + + +def test_a_sufficient_budget_waits_for_the_query_embedding(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + embedding = _SlowQueryEmbedding(delay=0.3) + + async def scenario() -> Any: + async with _runtime(tmp_path / "waited.db", 5.0, embedding) as runtime: + scope_id = await _scope_with_memory(runtime, "waited") + return await runtime.context.for_scope(scope_id).prepare(_request()) + + prepared = asyncio.run(scenario()) + + assert prepared.status == "ready" + assert embedding.query_completed is True + assert _fallbacks(caplog) == []