Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/en/docs/operate/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
1 change: 1 addition & 0 deletions docs/zh/docs/operate/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
12 changes: 6 additions & 6 deletions src/powercontext/builtin/runtime/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"})
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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,
)
)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
24 changes: 24 additions & 0 deletions src/powercontext/builtin/runtime/atomic_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 (
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand Down
1 change: 1 addition & 0 deletions src/powercontext/builtin/runtime/composition.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
12 changes: 12 additions & 0 deletions src/powercontext/builtin/runtime/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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:
Expand Down
41 changes: 41 additions & 0 deletions tests/builtin/runtime/test_recall_embedding_timeout_config.py
Original file line number Diff line number Diff line change
@@ -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
186 changes: 186 additions & 0 deletions tests/e2e/test_recall_embedding_timeout.py
Original file line number Diff line number Diff line change
@@ -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) == []
Loading