Skip to content
Open
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/reference/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ Server settings use the `POWERCONTEXT_SERVER_` prefix.
| `POWERCONTEXT_SERVER_METRICS_ENABLED` | `true` | Expose Prometheus metrics at `/metrics` |
| `POWERCONTEXT_SERVER_TRACING_ENABLED` | `false` | Enable span recording and OTLP export |
| `POWERCONTEXT_SERVER_DATABASE_URL` | user data SQLite file | SQLAlchemy async database URL |
| `POWERCONTEXT_SERVER_RUNTIME_SCOPE_CACHE_SIZE` | `128` | Inactive scope compositions retained by the Runtime; in-flight scopes are never evicted |
| `POWERCONTEXT_SERVER_RUNTIME_SOURCE_WINDOW_LIMIT` | `100` | Maximum Sources processed in one activation |
| `POWERCONTEXT_SERVER_RUNTIME_MEMORY_EXTRACTION_PROFILE` | `coding` | Memory selection policy: `coding` or `conversation` |
| `POWERCONTEXT_SERVER_RUNTIME_MEMORY_RERANK_ENABLED` | `false` | Apply listwise reranking after coarse Memory retrieval |
Expand Down
1 change: 1 addition & 0 deletions docs/zh/docs/reference/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ Server 配置使用 `POWERCONTEXT_SERVER_` 前缀。
| `POWERCONTEXT_SERVER_METRICS_ENABLED` | `true` | 在 `/metrics` 暴露 Prometheus metrics |
| `POWERCONTEXT_SERVER_TRACING_ENABLED` | `false` | 启用 span recording 和 OTLP export |
| `POWERCONTEXT_SERVER_DATABASE_URL` | 用户数据目录下的 SQLite 文件 | SQLAlchemy 异步数据库 URL |
| `POWERCONTEXT_SERVER_RUNTIME_SCOPE_CACHE_SIZE` | `128` | Runtime 保留的非活动 scope composition 数量;进行中的 scope 不会被驱逐 |
| `POWERCONTEXT_SERVER_RUNTIME_SOURCE_WINDOW_LIMIT` | `100` | 单次 activation 最多处理的 Source 数量 |
| `POWERCONTEXT_SERVER_RUNTIME_MEMORY_EXTRACTION_PROFILE` | `coding` | Memory 选择策略:`coding` 或 `conversation` |
| `POWERCONTEXT_SERVER_RUNTIME_MEMORY_RERANK_ENABLED` | `false` | 在 Memory 粗召回后应用 listwise rerank |
Expand Down
149 changes: 149 additions & 0 deletions src/powercontext/builtin/runtime/_scope_cache.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
# 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.

"""Bounded lifecycle for scope-local Runtime resources."""
Comment thread
thunguo marked this conversation as resolved.

from __future__ import annotations

import asyncio
from collections import OrderedDict
from collections.abc import Callable, Iterator
from contextlib import contextmanager, suppress
from dataclasses import dataclass, field

DEFAULT_SCOPE_CACHE_SIZE = 128

ScopeEvictor = Callable[[str], None]
ScopeCacheObserver = Callable[[int, int], None]


@dataclass(frozen=True, slots=True)
class ScopeCacheCounts:
"""Low-cardinality snapshot of inactive cached and currently active scopes."""

cached: int
active: int


@dataclass(slots=True)
class _ScopeEntry:
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
leases: int = 0


class ScopeCache:
"""Retain a bounded LRU of inactive scopes without evicting in-flight work."""

def __init__(
self,
capacity: int,
*,
evictor: ScopeEvictor | None = None,
observer: ScopeCacheObserver | None = None,
) -> None:
if capacity < 1:
raise ValueError("scope cache capacity must be positive") # noqa: TRY003
self.capacity = capacity
self._evictor = evictor
self._observer = observer
self._entries: OrderedDict[str, _ScopeEntry] = OrderedDict()
self._observe()

@contextmanager
def lease(self, scope_id: str, /) -> Iterator[None]:
"""Keep one scope and its serialization lock alive for an operation."""

entry = self._entries.get(scope_id)
if entry is None:
self._make_room()
entry = _ScopeEntry()
self._entries[scope_id] = entry
entry.leases += 1
self._entries.move_to_end(scope_id)
self._observe()
try:
yield
finally:
current = self._entries.get(scope_id)
if current is not entry or entry.leases < 1:
raise RuntimeError("scope cache lease invariant violated") # noqa: TRY003
entry.leases -= 1
self._entries.move_to_end(scope_id)
self._trim()
self._observe()

def lock(self, scope_id: str, /) -> asyncio.Lock:
"""Return the serialization lock protected by the caller's current lease."""

entry = self._entries.get(scope_id)
if entry is None or entry.leases < 1:
raise RuntimeError("scope cache entry is not leased") # noqa: TRY003
return entry.lock

@property
def counts(self) -> ScopeCacheCounts:
"""Return inactive cached and active scope counts without scope identifiers."""

inactive = sum(entry.leases == 0 for entry in self._entries.values())
active = sum(entry.leases > 0 for entry in self._entries.values())
return ScopeCacheCounts(cached=inactive, active=active)

def clear(self) -> None:
"""Evict all entries after the Runtime has drained its operations."""

if any(entry.leases > 0 for entry in self._entries.values()):
raise RuntimeError("cannot clear active scope cache entries") # noqa: TRY003
for scope_id in tuple(self._entries):
self._evict(scope_id)
self._observe()

def _make_room(self) -> None:
while len(self._entries) >= self.capacity:
scope_id = self._oldest_inactive_scope()
if scope_id is None:
return
self._evict(scope_id)

def _trim(self) -> None:
while len(self._entries) > self.capacity:
scope_id = self._oldest_inactive_scope()
if scope_id is None:
return
self._evict(scope_id)

def _oldest_inactive_scope(self) -> str | None:
return next((scope_id for scope_id, entry in self._entries.items() if entry.leases == 0), None)

def _evict(self, scope_id: str) -> None:
entry = self._entries.pop(scope_id)
if entry.leases > 0:
raise RuntimeError("cannot evict an active scope cache entry") # noqa: TRY003
if self._evictor is not None:
self._evictor(scope_id)

def _observe(self) -> None:
if self._observer is None:
return
counts = self.counts
with suppress(Exception):
self._observer(counts.cached, counts.active)


__all__ = [
"DEFAULT_SCOPE_CACHE_SIZE",
"ScopeCache",
"ScopeCacheCounts",
"ScopeCacheObserver",
"ScopeEvictor",
]
86 changes: 65 additions & 21 deletions src/powercontext/builtin/runtime/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
from datetime import UTC, datetime
from pathlib import Path
from time import perf_counter
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, ValidationError

Expand Down Expand Up @@ -73,6 +73,12 @@
from powercontext.builtin.inference.usage import bind_usage_reporter
from powercontext.builtin.review.generation import GeneratedCandidateResult, ReviewedGenerationService
from powercontext.builtin.review.service import ReviewService
from powercontext.builtin.runtime._scope_cache import (
DEFAULT_SCOPE_CACHE_SIZE,
ScopeCache,
ScopeCacheObserver,
ScopeEvictor,
)
from powercontext.builtin.runtime.errors import InvalidRuntimeRequestError
from powercontext.builtin.runtime.models import (
ApproveArtifactCandidateRequest,
Expand Down Expand Up @@ -252,7 +258,7 @@ def __init__(self, runtime: BuiltinRuntime, scope_id: str) -> None:
self.scope_id = validate_scope_id(scope_id)

async def overview(self, *, period: StatisticsPeriod = StatisticsPeriod.THIRTY_DAYS) -> Statistics:
async with self._runtime._operation():
async with self._runtime._scope_operation(self.scope_id):
return await self._runtime._statistics(self.scope_id).overview(period, self._runtime._clock())

async def record_model_usage(
Expand All @@ -263,12 +269,13 @@ async def record_model_usage(
/,
) -> None:
try:
await self._runtime._statistics(self.scope_id).record(
purpose,
operation,
usage,
self._runtime._clock().astimezone(UTC).date(),
)
async with self._runtime._scope_operation(self.scope_id):
await self._runtime._statistics(self.scope_id).record(
purpose,
operation,
usage,
self._runtime._clock().astimezone(UTC).date(),
)
except Exception as error:
log_safely(
logger,
Expand All @@ -286,10 +293,11 @@ async def record_model_usage(

async def record_recall(self, measurement: RecallTokenMeasurement, /) -> None:
try:
await self._runtime._statistics(self.scope_id).record_recall(
measurement,
self._runtime._clock().astimezone(UTC).date(),
)
async with self._runtime._scope_operation(self.scope_id):
await self._runtime._statistics(self.scope_id).record_recall(
measurement,
self._runtime._clock().astimezone(UTC).date(),
)
except Exception as error:
log_safely(
logger,
Expand Down Expand Up @@ -323,6 +331,10 @@ def __init__(self, runtime: BuiltinRuntime, scope_id: str) -> None:
self.scope_id = validate_scope_id(scope_id)

async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext:
async with self._runtime._scope_operation(self.scope_id):
return await self._prepare(request)

async def _prepare(self, request: PrepareContextRequest, /) -> PreparedContext:
builder = PreparedContextBuilder()
async with (
self._runtime._context(self.scope_id, embedding_purpose=ModelUsagePurpose.MEMORY_RECALL) as context,
Expand Down Expand Up @@ -1167,6 +1179,9 @@ def __init__(
provider: PowerContextProvider[BuiltinSources, BuiltinArtifacts, BuiltinTriggers],
capabilities: RuntimeCapabilities,
source_window_limit: int = 100,
scope_cache_size: int = DEFAULT_SCOPE_CACHE_SIZE,
scope_evictor: ScopeEvictor | None = None,
scope_cache_observer: ScopeCacheObserver | None = None,
scope_ids: ScopeIds | None = None,
review_service: ReviewServiceFactory | None = None,
generation_service: GenerationServiceFactory | None = None,
Expand All @@ -1182,6 +1197,8 @@ def __init__(
) -> None:
if source_window_limit < 1:
raise _RuntimeConfigurationError("source_window_limit")
if scope_cache_size < 1:
raise _RuntimeConfigurationError("scope_cache_size")
self._provider = provider
self._capabilities = capabilities
self._review_service = review_service
Expand All @@ -1196,11 +1213,16 @@ def __init__(
self._clock = _utc_now if clock is None else clock
self._tracing = tracing
self.source_window_limit = source_window_limit
self._locks: dict[str, asyncio.Lock] = {}
self._scope_cache = ScopeCache(
scope_cache_size,
evictor=scope_evictor,
observer=scope_cache_observer,
)
self._processor_lock = asyncio.Lock()
self._close_lock = asyncio.Lock()
self._lifecycle = asyncio.Condition()
self._active_operations = 0
self._operation_depths: dict[asyncio.Task[Any], int] = {}
self._closing = False
self._closed = False
self._scheduler: AsyncIOScheduler | None = None
Expand Down Expand Up @@ -1330,21 +1352,40 @@ async def close(self) -> None:
unregister_processor(self._scheduler_runtime_key)
self._scheduler_runtime_key = None
self._scheduler = None
self._scope_cache.clear()
self._closed = True

@asynccontextmanager
async def _operation(self) -> AsyncIterator[None]:
task = asyncio.current_task()
if task is None:
raise RuntimeError("Runtime operations require an asyncio Task") # noqa: TRY003
async with self._lifecycle:
if self._closing or self._closed:
raise _RuntimeStateError("closed")
self._active_operations += 1
depth = self._operation_depths.get(task, 0)
if depth == 0:
if self._closing or self._closed:
raise _RuntimeStateError("closed")
self._active_operations += 1
self._operation_depths[task] = depth + 1
try:
yield
finally:
async with self._lifecycle:
self._active_operations -= 1
if self._active_operations == 0:
self._lifecycle.notify_all()
depth = self._operation_depths[task] - 1
if depth > 0:
self._operation_depths[task] = depth
else:
del self._operation_depths[task]
self._active_operations -= 1
if self._active_operations == 0:
self._lifecycle.notify_all()

@asynccontextmanager
async def _scope_operation(self, scope_id: str) -> AsyncIterator[None]:
scope = validate_scope_id(scope_id)
async with self._operation():
with self._scope_cache.lease(scope):
yield

@asynccontextmanager
async def _scoped_operation(
Expand All @@ -1355,7 +1396,7 @@ async def _scoped_operation(
embedding_purpose: ModelUsagePurpose | None = None,
) -> AsyncIterator[None]:
scope = validate_scope_id(scope_id)
async with self._operation():
async with self._scope_operation(scope):
with bind_usage_reporter(
self.statistics.for_scope(scope).record_model_usage,
generation_purpose=generation_purpose,
Expand All @@ -1381,11 +1422,14 @@ async def _context(
context = await self._provider.get(validate_scope_id(scope_id))
yield context

def _lock(self, scope_id: str) -> asyncio.Lock:
return self._scope_cache.lock(validate_scope_id(scope_id))
Comment thread
thunguo marked this conversation as resolved.

@asynccontextmanager
async def _locked(self, scope_id: str) -> AsyncIterator[None]:
"""Serialize writes for one scope and trace only the wait, not the critical section."""

lock = self._locks.setdefault(validate_scope_id(scope_id), asyncio.Lock())
lock = self._lock(scope_id)
acquired = False
try:
# Injected tracing must never leak the lock, so the release is armed before the stage is closed.
Expand Down
5 changes: 5 additions & 0 deletions src/powercontext/builtin/runtime/composition.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
from powercontext.builtin.persistence.sqlite.memory_index import SQLiteMemoryFTSIndex, SQLiteMemoryVectorIndex
from powercontext.builtin.persistence.sqlite.profile import SQLiteConfig, SQLiteProfile
from powercontext.builtin.persistence.tables import BUILTIN_TABLES
from powercontext.builtin.runtime._scope_cache import ScopeCacheObserver
from powercontext.builtin.runtime.application import BuiltinRuntime
from powercontext.builtin.runtime.config import BuiltinConfig, ExternalSkillsConfig, InferenceConfig, RuntimeConfig
from powercontext.builtin.runtime.models import MemorySearchMode, RuntimeCapabilities
Expand Down Expand Up @@ -167,6 +168,7 @@ async def open_builtin_runtime(
token_estimator: TokenEstimator | None = None,
memory_reranker: MemoryReranker | None = None,
instrumentation: InstrumentationSettings | None = None,
scope_cache_observer: ScopeCacheObserver | None = None,
tracing: RuntimeTracing | None = None,
) -> AsyncIterator[BuiltinRuntime]:
"""Open the selected database, inference adapters, and built-in runtime."""
Expand Down Expand Up @@ -265,6 +267,9 @@ async def open_builtin_runtime(
handoff_generation=contexts.handoff_generation,
),
source_window_limit=config.runtime.source_window_limit,
scope_cache_size=config.runtime.scope_cache_size,
scope_evictor=contexts.evict,
scope_cache_observer=scope_cache_observer,
scope_ids=contexts.scope_ids,
review_service=contexts.review,
generation_service=contexts.generation,
Expand Down
2 changes: 2 additions & 0 deletions src/powercontext/builtin/runtime/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,13 @@
from powercontext.builtin.persistence.oceanbase import OceanBaseConfig
from powercontext.builtin.persistence.seekdb import SeekDBConfig
from powercontext.builtin.persistence.sqlite import SQLiteConfig
from powercontext.builtin.runtime._scope_cache import DEFAULT_SCOPE_CACHE_SIZE


class RuntimeConfig(BaseModel):
"""Built-in runtime policy and scheduler configuration."""

scope_cache_size: int = Field(default=DEFAULT_SCOPE_CACHE_SIZE, ge=1)
source_window_limit: int = Field(default=100, ge=1)
memory_extraction_profile: MemoryExtractionProfile = MemoryExtractionProfile.CODING
memory_rerank_enabled: bool = False
Expand Down
Loading
Loading