diff --git a/src/powercontext/builtin/runtime/__init__.py b/src/powercontext/builtin/runtime/__init__.py index 9dfbf5a71..a353a3564 100644 --- a/src/powercontext/builtin/runtime/__init__.py +++ b/src/powercontext/builtin/runtime/__init__.py @@ -98,6 +98,13 @@ InferenceConfig, RuntimeConfig, ) +from powercontext.builtin.runtime.decision_model import ( + DecisionModel, + DecisionModelOption, + DecisionModelRequest, + DecisionModelResult, + StructuredDecisionModel, +) from powercontext.builtin.runtime.errors import InvalidRuntimeRequestError, TopicMemoryProcessingUnavailableError from powercontext.builtin.runtime.models import ( ApproveArtifactCandidateRequest, @@ -213,6 +220,10 @@ "ContextAssemblySection", "CreateDreamRunRequest", "DatabaseConfig", + "DecisionModel", + "DecisionModelOption", + "DecisionModelRequest", + "DecisionModelResult", "DreamApplication", "DreamRun", "DreamRunPage", @@ -337,6 +348,7 @@ "Statistics", "StatisticsApplication", "StatisticsPeriod", + "StructuredDecisionModel", "SubmitSourceObservation", "TopicMemoryApplication", "TopicMemoryFlushResult", diff --git a/src/powercontext/builtin/runtime/config.py b/src/powercontext/builtin/runtime/config.py index 3a8f97638..cdf84e35c 100644 --- a/src/powercontext/builtin/runtime/config.py +++ b/src/powercontext/builtin/runtime/config.py @@ -139,6 +139,7 @@ def reject_boolean_worker_quota(cls, value: Any) -> Any: source_window_limit: int = Field(default=100, ge=1) context_assembly_max_entries: int = Field(default=8, ge=1) memory_extraction_profile: MemoryExtractionProfile = MemoryExtractionProfile.CODING + decision_assistance_enabled: bool = False memory_rerank_enabled: bool = False memory_rerank_candidate_limit: int = Field(default=30, ge=1, le=100) recall_gate_enabled: bool = False @@ -248,6 +249,12 @@ class InferenceConfig(BaseModel): embedding_normalization: Literal["none", "unit"] = "unit" embedding_timeout_seconds: float = Field(default=30.0, gt=0) embedding_batch_size: int = Field(default=10, ge=1) + decision_model: str | None = None + decision_base_url: AnyHttpUrl | None = None + decision_headers: dict[str, SecretStr] = Field(default_factory=dict, repr=False) + decision_model_settings: dict[str, JsonValue] = Field(default_factory=dict) + decision_timeout_seconds: float | None = Field(default=None, gt=0) + decision_max_requests: int | None = Field(default=None, ge=1) rerank_model: str | None = None rerank_base_url: AnyHttpUrl | None = None rerank_headers: dict[str, SecretStr] = Field(default_factory=dict, repr=False) @@ -255,7 +262,7 @@ class InferenceConfig(BaseModel): rerank_timeout_seconds: float | None = Field(default=None, gt=0) rerank_max_requests: int | None = Field(default=None, ge=1) - @field_validator("generation_model", "embedding_model", "embedding_profile_id", "rerank_model") + @field_validator("generation_model", "embedding_model", "embedding_profile_id", "decision_model", "rerank_model") @classmethod def validate_optional_identifier(cls, value: str | None) -> str | None: if value is None: @@ -275,7 +282,7 @@ def validate_normalization(cls, value: object) -> object: raise ValueError("embedding normalization must be 'none' or 'unit'") # noqa: TRY003 return normalized - @field_validator("generation_headers", "embedding_headers", "rerank_headers") + @field_validator("generation_headers", "embedding_headers", "decision_headers", "rerank_headers") @classmethod def validate_headers(cls, value: dict[str, SecretStr]) -> dict[str, SecretStr]: normalized_names: set[str] = set() @@ -290,7 +297,12 @@ def validate_headers(cls, value: dict[str, SecretStr]) -> dict[str, SecretStr]: normalized_names.add(normalized_name) return value - @field_validator("generation_model_settings", "embedding_model_settings", "rerank_model_settings") + @field_validator( + "generation_model_settings", + "embedding_model_settings", + "decision_model_settings", + "rerank_model_settings", + ) @classmethod def reserve_headers_field(cls, value: dict[str, JsonValue]) -> dict[str, JsonValue]: if "extra_headers" in value: @@ -310,14 +322,32 @@ def validate_embedding_profile(self) -> Self: @model_validator(mode="after") def validate_workload_overrides(self) -> Self: + self._validate_generation_overrides() + self._validate_embedding_overrides() + self._validate_decision_overrides() + self._validate_rerank_overrides() + return self + + def _validate_generation_overrides(self) -> None: if self.generation_model is None and self.generation_model_settings: raise ValueError("generation_model_settings requires generation_model") # noqa: TRY003 if self.generation_model is None and (self.generation_base_url is not None or self.generation_headers): raise ValueError("generation overrides require generation_model") # noqa: TRY003 + self._validate_generation_budget() + + def _validate_embedding_overrides(self) -> None: if self.embedding_model is None and ( self.embedding_base_url is not None or self.embedding_headers or self.embedding_model_settings ): raise ValueError("embedding overrides require a complete embedding profile") # noqa: TRY003 + + def _validate_decision_overrides(self) -> None: + if self.decision_base_url is not None and self.decision_model is None: + raise ValueError("decision_base_url requires decision_model") # noqa: TRY003 + if self.decision_model is None and (self.decision_headers or self.decision_model_settings): + raise ValueError("decision overrides require decision_model") # noqa: TRY003 + + def _validate_rerank_overrides(self) -> None: if self.rerank_base_url is not None and self.rerank_model is None: raise ValueError("rerank_base_url requires rerank_model") # noqa: TRY003 if ( @@ -326,6 +356,8 @@ def validate_workload_overrides(self) -> Self: and (self.rerank_headers or self.rerank_model_settings) ): raise ValueError("rerank overrides require rerank_model or generation_model") # noqa: TRY003 + + def _validate_generation_budget(self) -> None: max_tokens = self.generation_model_settings.get("max_tokens") if max_tokens is not None and ( not isinstance(max_tokens, int) or isinstance(max_tokens, bool) or max_tokens < 1 @@ -343,7 +375,6 @@ def validate_workload_overrides(self) -> Self: raise ValueError( # noqa: TRY003 f"Topic Memory generation budget is invalid: {error.error_code}" ) from error - return self class ExternalSkillsConfig(BaseModel): diff --git a/src/powercontext/builtin/runtime/decision_model.py b/src/powercontext/builtin/runtime/decision_model.py new file mode 100644 index 000000000..80d6fcb44 --- /dev/null +++ b/src/powercontext/builtin/runtime/decision_model.py @@ -0,0 +1,161 @@ +# 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. + +"""Provider-neutral decision model contracts for runtime gates.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Annotated, Protocol + +from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator, model_validator + +from powercontext.builtin.inference import InferenceError, InferenceUsage, StructuredGenerator + + +class DecisionModelOption(BaseModel): + """One selectable answer exposed to a deterministic runtime decision.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + option_id: Annotated[str, Field(min_length=1, max_length=128)] + label: Annotated[str, Field(min_length=1, max_length=512)] + description: Annotated[str | None, Field(min_length=1, max_length=4096)] = None + + @field_validator("option_id", "label", "description") + @classmethod + def validate_trimmed_text(cls, value: str | None) -> str | None: + if value is None: + return None + if value != value.strip(): + raise ValueError("DecisionModel text fields must be trimmed") # noqa: TRY003 + return value + + +class DecisionModelRequest(BaseModel): + """A bounded, explicit-choice decision made outside the main agent loop.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + operation: Annotated[str, Field(min_length=1, max_length=128)] + question: Annotated[str, Field(min_length=1, max_length=8192)] + options: Annotated[tuple[DecisionModelOption, ...], Field(min_length=2, max_length=128)] + context: Annotated[str | None, Field(min_length=1, max_length=32768)] = None + metadata: Mapping[str, JsonValue] = Field(default_factory=dict) + + @field_validator("operation", "question", "context") + @classmethod + def validate_trimmed_text(cls, value: str | None) -> str | None: + if value is None: + return None + if value != value.strip(): + raise ValueError("DecisionModel text fields must be trimmed") # noqa: TRY003 + return value + + @model_validator(mode="after") + def validate_option_ids(self) -> DecisionModelRequest: + option_ids = [option.option_id for option in self.options] + if len(option_ids) != len(set(option_ids)): + raise ValueError("DecisionModel option IDs must be unique") # noqa: TRY003 + return self + + +class DecisionModelResult(BaseModel): + """A selected option plus portable usage and fallback metadata.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + selected_option_id: Annotated[str | None, Field(min_length=1, max_length=128)] = None + confidence: Annotated[float | None, Field(ge=0.0, le=1.0)] = None + scores: Mapping[str, Annotated[float, Field(ge=0.0, le=1.0)]] = Field(default_factory=dict) + rationale: Annotated[str | None, Field(min_length=1, max_length=4096)] = None + used_fallback: bool = False + fallback_reason: Annotated[str | None, Field(min_length=1, max_length=256)] = None + usage: InferenceUsage = Field(default_factory=lambda: InferenceUsage(requests=0)) + + @field_validator("selected_option_id", "rationale", "fallback_reason") + @classmethod + def validate_trimmed_text(cls, value: str | None) -> str | None: + if value is None: + return None + if value != value.strip(): + raise ValueError("DecisionModel text fields must be trimmed") # noqa: TRY003 + return value + + @model_validator(mode="after") + def validate_fallback(self) -> DecisionModelResult: + if self.used_fallback: + if self.selected_option_id is not None: + raise ValueError("DecisionModel fallback results must not select an option") # noqa: TRY003 + if self.fallback_reason is None: + raise ValueError("DecisionModel fallback results require fallback_reason") # noqa: TRY003 + elif self.fallback_reason is not None: + raise ValueError("DecisionModel fallback_reason requires used_fallback") # noqa: TRY003 + return self + + +class DecisionModel(Protocol): + """Evaluate explicit-choice runtime decisions without exposing provider SDKs.""" + + policy_id: str + + async def evaluate(self, request: DecisionModelRequest, /) -> DecisionModelResult: + """Return one provider-neutral decision result or an explicit fallback.""" + + ... + + +class StructuredDecisionModel: + """Adapt a structured generator into a fail-open DecisionModel.""" + + def __init__( + self, + generator: StructuredGenerator[DecisionModelRequest, DecisionModelResult], + *, + policy_id: str, + ) -> None: + self._generator = generator + self.policy_id = policy_id + + async def evaluate(self, request: DecisionModelRequest, /) -> DecisionModelResult: + try: + result = await self._generator.generate(request) + except InferenceError as error: + return DecisionModelResult(used_fallback=True, fallback_reason=type(error).__name__) + output = result.output.model_copy(update={"usage": result.usage}) + option_ids = {option.option_id for option in request.options} + if output.used_fallback: + return output + if output.selected_option_id not in option_ids: + return DecisionModelResult( + used_fallback=True, + fallback_reason="invalid_output", + usage=result.usage, + ) + if any(option_id not in option_ids for option_id in output.scores): + return DecisionModelResult( + used_fallback=True, + fallback_reason="invalid_output", + usage=result.usage, + ) + return output + + +__all__ = [ + "DecisionModel", + "DecisionModelOption", + "DecisionModelRequest", + "DecisionModelResult", + "StructuredDecisionModel", +] diff --git a/tests/builtin/runtime/test_decision_model.py b/tests/builtin/runtime/test_decision_model.py new file mode 100644 index 000000000..f12c0803f --- /dev/null +++ b/tests/builtin/runtime/test_decision_model.py @@ -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. + +from __future__ import annotations + +import asyncio + +import pytest +from pydantic import ValidationError + +from powercontext.builtin.inference import GenerationResult, InferenceUnavailableError, InferenceUsage +from powercontext.builtin.runtime import ( + DecisionModelOption, + DecisionModelRequest, + DecisionModelResult, + StructuredDecisionModel, +) + + +def test_decision_request_requires_unique_explicit_options() -> None: + with pytest.raises(ValidationError, match="option IDs must be unique"): + DecisionModelRequest( + operation="memory.write_gate", + question="Should this candidate be written?", + options=( + DecisionModelOption(option_id="write", label="Write"), + DecisionModelOption(option_id="write", label="Also write"), + ), + ) + + +def test_decision_result_fallback_must_be_explicit_and_non_selecting() -> None: + with pytest.raises(ValidationError, match="fallback results must not select an option"): + DecisionModelResult( + selected_option_id="write", + used_fallback=True, + fallback_reason="provider unavailable", + ) + + with pytest.raises(ValidationError, match="fallback results require fallback_reason"): + DecisionModelResult(used_fallback=True) + + +def test_decision_result_can_report_fail_open_without_usage() -> None: + result = DecisionModelResult(used_fallback=True, fallback_reason="provider unavailable") + + assert result.selected_option_id is None + assert result.usage.requests == 0 + + +def test_structured_decision_model_reports_usage_from_generator() -> None: + class Generator: + async def generate(self, value: DecisionModelRequest, /) -> GenerationResult[DecisionModelResult]: + assert value.operation == "handoff.consult" + return GenerationResult( + output=DecisionModelResult(selected_option_id="continue", confidence=0.8), + usage=InferenceUsage(requests=1, input_tokens=7, output_tokens=3), + ) + + async def scenario() -> None: + model = StructuredDecisionModel(Generator(), policy_id="test") + result = await model.evaluate( + DecisionModelRequest( + operation="handoff.consult", + question="Can the next agent continue?", + options=( + DecisionModelOption(option_id="continue", label="Continue"), + DecisionModelOption(option_id="escalate", label="Escalate"), + ), + ) + ) + + assert result.selected_option_id == "continue" + assert result.usage.input_tokens == 7 + + asyncio.run(scenario()) + + +@pytest.mark.parametrize( + "output", + [ + DecisionModelResult(selected_option_id="missing"), + DecisionModelResult(), + DecisionModelResult(selected_option_id="write", scores={"missing": 0.9}), + ], +) +def test_structured_decision_model_rejects_outputs_outside_requested_options(output) -> None: + class Generator: + async def generate(self, value: DecisionModelRequest, /) -> GenerationResult[DecisionModelResult]: + return GenerationResult( + output=output, + usage=InferenceUsage(requests=1, input_tokens=7, output_tokens=3), + ) + + async def scenario() -> None: + model = StructuredDecisionModel(Generator(), policy_id="test") + result = await model.evaluate( + DecisionModelRequest( + operation="memory.write_gate", + question="Should this candidate be written?", + options=( + DecisionModelOption(option_id="write", label="Write"), + DecisionModelOption(option_id="defer", label="Defer"), + ), + ) + ) + + assert result.selected_option_id is None + assert result.used_fallback is True + assert result.fallback_reason == "invalid_output" + assert result.usage.input_tokens == 7 + + asyncio.run(scenario()) + + +def test_structured_decision_model_fails_open_on_provider_unavailable() -> None: + class Generator: + async def generate(self, value: DecisionModelRequest, /) -> GenerationResult[DecisionModelResult]: + raise InferenceUnavailableError("decision") + + async def scenario() -> None: + model = StructuredDecisionModel(Generator(), policy_id="test") + result = await model.evaluate( + DecisionModelRequest( + operation="memory.write_gate", + question="Should this candidate be written?", + options=( + DecisionModelOption(option_id="write", label="Write"), + DecisionModelOption(option_id="defer", label="Defer"), + ), + ) + ) + + assert result.selected_option_id is None + assert result.used_fallback is True + assert result.fallback_reason == "InferenceUnavailableError" + + asyncio.run(scenario()) diff --git a/tests/test_server_generation.py b/tests/test_server_generation.py index a37588a59..4df2decd5 100644 --- a/tests/test_server_generation.py +++ b/tests/test_server_generation.py @@ -35,3 +35,17 @@ def test_generation_model_settings_reject_static_headers() -> None: generation_model="provider:model", generation_model_settings={"extra_headers": {"Authorization": "Bearer secret"}}, ) + + +def test_decision_overrides_require_decision_model() -> None: + with pytest.raises(ValidationError, match="decision overrides require decision_model"): + InferenceConfig( + decision_model_settings={ + "temperature": 0, + } + ) + + +def test_decision_base_url_requires_decision_model() -> None: + with pytest.raises(ValidationError, match="decision_base_url requires decision_model"): + InferenceConfig.model_validate({"decision_base_url": "https://decision.example/v1"})