diff --git a/.env.example b/.env.example index f2766a4848..645355b31a 100644 --- a/.env.example +++ b/.env.example @@ -109,6 +109,14 @@ POWERCONTEXT_SERVER_RUNTIME_DREAM_MAX_PENDING_PER_SCOPE=32 # Memory write gate: opt-in evidence-sufficiency check before a write commits. Disabled by # default; when unset no gate runs and no extra model call is made. # POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_ENABLED=true +# `disabled` does not construct a gate. `shadow` records an internal observation but always lets +# the Memory write proceed. `advisory` turns an insufficient-evidence result into FLAG, while +# `enforcing` preserves the visible ACCEPT/FLAG/HOLD outcomes. +# POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_MODE=shadow +# The default makes no decision-model call. Set `local_only` only for an in-process or loopback +# decision backend controlled by this deployment. `hosted_redacted` is rejected until a shared +# PowerContext content sanitizer is separately reviewed. +# POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_PRIVACY_BOUNDARY=no_external_call # Direction only (which verdict means the cited evidence is insufficient): "yes" or "no". # POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_HOLD_ON=yes # Optional confidence floor below which a hold becomes a written-but-annotated change. diff --git a/docs/en/docs/operate/troubleshoot.md b/docs/en/docs/operate/troubleshoot.md index 16b97da35a..fb5457c358 100644 --- a/docs/en/docs/operate/troubleshoot.md +++ b/docs/en/docs/operate/troubleshoot.md @@ -259,15 +259,15 @@ so the previous database remains available for recovery: Because `pc_scopes.parent_scope_id` is self-referential, keep ancestor Scope rows before their descendants in the exported `pc_scopes` data. If the source predates the three Skill lifecycle tables (`pc_skill_packages`, `pc_agent_skill_targets`, and - `pc_skill_publications`), the Profile tables, `pc_topic_memory_work_budgets`, or - `pc_receipt_migration_review`, remove the absent tables from their respective layers. + `pc_skill_publications`), the Profile tables, `pc_topic_memory_work_budgets`, `pc_recall_effort_daily`, + `pc_receipt_migration_review`, or `pc_decision_observations`, remove the absent tables from their respective layers. When a work-budget table exists, restore it together with Cursors so failure allowances survive the migration. Layer 1 contains parents and tables without foreign keys: ```bash obloader -D --csv \ - --table 'pc_scopes,pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_memory_source_windows,pc_artifact_processing_leases,pc_artifact_processing_binding_states,pc_artifact_processing_pending,pc_artifact_processing_auto_wave_targets,pc_artifact_processing_sequences,pc_artifact_processing_intents,pc_topic_memory_processing_targets,pc_artifact_processing_schema,pc_artifact_processing_migration_receipts,pc_topic_memory_work_budgets,pc_topic_memory_retrieval_shape,pc_connector_checkpoints,pc_source_definition_manifests,pc_external_skill_registrations,pc_skill_packages,pc_agent_skill_targets,pc_skill_publications,pc_model_usage_daily,pc_recall_token_daily,pc_recall_effort_daily,pc_receipt_migration_review' \ + --table 'pc_scopes,pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_memory_source_windows,pc_artifact_processing_leases,pc_artifact_processing_binding_states,pc_artifact_processing_pending,pc_artifact_processing_auto_wave_targets,pc_artifact_processing_sequences,pc_artifact_processing_intents,pc_topic_memory_processing_targets,pc_artifact_processing_schema,pc_artifact_processing_migration_receipts,pc_topic_memory_work_budgets,pc_topic_memory_retrieval_shape,pc_connector_checkpoints,pc_source_definition_manifests,pc_external_skill_registrations,pc_skill_packages,pc_agent_skill_targets,pc_skill_publications,pc_model_usage_daily,pc_recall_token_daily,pc_recall_effort_daily,pc_receipt_migration_review,pc_decision_observations' \ -f ``` diff --git a/docs/en/rfcs/1770-decision-policy-governance.md b/docs/en/rfcs/1770-decision-policy-governance.md index 9910cd96ab..a86ca75844 100644 --- a/docs/en/rfcs/1770-decision-policy-governance.md +++ b/docs/en/rfcs/1770-decision-policy-governance.md @@ -58,7 +58,7 @@ behaviour. RFC 1652 reserves Memory consolidation, supersession, and lifecycle e shared policy and observation contract, each consumer will reinvent its own threshold, fallback, privacy, and audit rules. -The expected outcome is a small shared layer that lets implementation PRs proceed in a consistent order: +For a policy that seeks promotion beyond `shadow` or `advisory`, the shared layer supports this evidence sequence: 1. define a policy in shadow mode; 2. collect observations without changing domain behaviour; @@ -66,7 +66,9 @@ The expected outcome is a small shared layer that lets implementation PRs procee 4. promote only the narrow action a policy has earned; 5. keep domain authority, review, persistence, and access control in the owning service. -This is an enablement RFC, not an adoption recommendation for a specific backend. +This is an enablement RFC, not an adoption recommendation for a specific backend or a commitment to implement every +listed consumer. A consumer may stop at `disabled`, `shadow`, or `advisory`; it does not need a replay runner, dashboard, +or cross-consumer evaluation bundle unless it seeks a stronger mode. # Guide-level explanation @@ -350,8 +352,8 @@ Shadow mode requirements: 2. Every gate attempt emits a `DecisionObservation`. 3. Observations distinguish no-call, local-rule, successful backend, backend fallback, privacy fallback, and parse failure. -4. The policy can be replayed or rescored without re-running the whole workload when enough exact references remain - resolvable. +4. A policy considered for `enforcing` can be replayed or rescored without re-running the whole workload when enough + exact references remain resolvable. Replay has two forms: @@ -361,6 +363,10 @@ Replay has two forms: Both forms must report their input coverage. A replay result that cannot resolve enough original material is not a negative result for the policy; it is an incomplete replay. +Replay/rescore is required promotion evidence, not a required runtime feature for every `shadow` or `advisory` consumer. +A consumer without that evidence must not be promoted to `enforcing`. +Memory write-gate promotion evidence is tracked separately in [#1930](https://github.com/oceanbase/powercontext/issues/1930). + ## Promotion criteria A policy must declare promotion criteria before moving beyond shadow. @@ -484,9 +490,6 @@ are candidate backends or reference points, not policy owners. # Unresolved questions -- What is the first concrete storage target for `DecisionObservation`: structured logs, evaluation artifacts, or a - private internal ledger? -- Which subset of observations should be retained by default, and for how long? - Should policy definitions live as Python constants, JSON/YAML resources, or both? - Which numeric promotion criteria should #1742 use if it is revised to conform to this RFC? - Do hosted-provider privacy boundaries need a user-facing configuration document before the first hosted backend is @@ -499,6 +502,5 @@ are candidate backends or reference points, not policy owners. - A policy registry page in the dashboard showing active policies, modes, fallback rates, and last replay results. - A CLI command that replays one policy against recent observations and prints candidate thresholds. - A standard evaluation bundle for decision policies, shared by Jev, Laya, local rules, and future backends. -- A private decision-observation ledger with retention controls and export for offline evaluation. - Policy-aware Review Inbox grouping, where a review item links back to the policy observation that suggested it. - Local-only decision backends for air-gapped deployments, using the same policy and observation contract. diff --git a/docs/zh/docs/operate/troubleshoot.md b/docs/zh/docs/operate/troubleshoot.md index a1d6291af3..a664e93bc6 100644 --- a/docs/zh/docs/operate/troubleshoot.md +++ b/docs/zh/docs/operate/troubleshoot.md @@ -250,16 +250,15 @@ collation，但不会包含数据库 URL 或凭据。 删除它。由于 `pc_scopes.parent_scope_id` 自引用 `pc_scopes`，导出的 `pc_scopes` 数据必须让祖先 Scope 记录排在 后代记录之前。 源数据库早于三张 Skill 生命周期表（`pc_skill_packages`、`pc_agent_skill_targets` 和 - `pc_skill_publications`）、Profile 表或 `pc_receipt_migration_review` 时，应从对应层删除缺失的表。 - - 旧版本没有 `pc_topic_memory_work_budgets` 时可从清单移除该表；若该表存在，必须与 Cursor 一起恢复， - 保留已消耗的失败额度，避免迁移后重置累计成本。 + `pc_skill_publications`）、Profile 表、`pc_topic_memory_work_budgets`、`pc_recall_effort_daily`、 + `pc_receipt_migration_review` 或 `pc_decision_observations` 时，应从对应层删除缺失的表。 + 如果存在 work-budget 表，必须与 Cursor 一起恢复，保留已消耗的失败额度，避免迁移后重置累计成本。 第 1 层包含父表和无外键的表： ```bash obloader -D --csv \ - --table 'pc_scopes,pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_memory_source_windows,pc_artifact_processing_leases,pc_artifact_processing_binding_states,pc_artifact_processing_pending,pc_artifact_processing_auto_wave_targets,pc_artifact_processing_sequences,pc_artifact_processing_intents,pc_topic_memory_processing_targets,pc_artifact_processing_schema,pc_artifact_processing_migration_receipts,pc_topic_memory_work_budgets,pc_topic_memory_retrieval_shape,pc_connector_checkpoints,pc_source_definition_manifests,pc_external_skill_registrations,pc_skill_packages,pc_agent_skill_targets,pc_skill_publications,pc_model_usage_daily,pc_recall_token_daily,pc_recall_effort_daily,pc_receipt_migration_review' \ + --table 'pc_scopes,pc_source_journal_heads,pc_sources,pc_artifacts,pc_source_cursors,pc_memory_source_windows,pc_artifact_processing_leases,pc_artifact_processing_binding_states,pc_artifact_processing_pending,pc_artifact_processing_auto_wave_targets,pc_artifact_processing_sequences,pc_artifact_processing_intents,pc_topic_memory_processing_targets,pc_artifact_processing_schema,pc_artifact_processing_migration_receipts,pc_topic_memory_work_budgets,pc_topic_memory_retrieval_shape,pc_connector_checkpoints,pc_source_definition_manifests,pc_external_skill_registrations,pc_skill_packages,pc_agent_skill_targets,pc_skill_publications,pc_model_usage_daily,pc_recall_token_daily,pc_recall_effort_daily,pc_receipt_migration_review,pc_decision_observations' \ -f ``` diff --git a/docs/zh/rfcs/1770-decision-policy-governance.md b/docs/zh/rfcs/1770-decision-policy-governance.md index 804f471020..4906b8d7e8 100644 --- a/docs/zh/rfcs/1770-decision-policy-governance.md +++ b/docs/zh/rfcs/1770-decision-policy-governance.md @@ -44,7 +44,7 @@ 项目已经在这个边界上有具体压力。#1742 提出了 Memory 写入时的 evidence gate。#1745 提出了 decision-model reranker，并明确把 rerank 的 fail-closed 行为与 advisory gate 的 fail-open 行为分开。RFC 1652 将 Memory consolidation、supersession 和 lifecycle effects 留给 review-gated 流程。如果没有共享 policy 与 observation 契约，每个消费者都会重新发明自己的阈值、fallback、隐私和审计规则。 -预期结果是一层很小的共享机制，使实现 PR 能按一致顺序推进： +对于寻求从 `shadow` 或 `advisory` 晋升的 policy，这层共享机制支持以下证据链： 1. 先用 shadow mode 定义 policy； 2. 在不改变领域行为的前提下收集 observation； @@ -52,7 +52,9 @@ 4. 只 promote 该 policy 已经证明可以承担的窄域动作； 5. 领域权威、review、持久化和访问控制仍由所属服务负责。 -这是一个 enablement RFC，不是对某个具体后端的采用建议。 +这是一个 enablement RFC，不是对某个具体后端的采用建议，也不承诺实现列出的每个 consumer。consumer 可以停在 +`disabled`、`shadow` 或 `advisory`，除非要进入更强模式，否则不需要实现 replay runner、dashboard 或跨 consumer +evaluation bundle。 # Guide-level explanation @@ -285,7 +287,7 @@ Shadow mode 要求： 1. 最终领域输出与 no-gate 路径 byte-for-byte 或语义等价。 2. 每次 gate attempt 都产生 `DecisionObservation`。 3. observation 区分 no-call、local-rule、successful backend、backend fallback、privacy fallback 和 parse failure。 -4. 只要仍能解析足够精确引用，policy 就可以在不重跑整个 workload 的情况下 replay 或 rescore。 +4. 被考虑进入 `enforcing` 的 policy，在仍能解析足够精确引用时，可以不重跑整个 workload 而 replay 或 rescore。 Replay 有两种形式： @@ -294,6 +296,10 @@ Replay 有两种形式： 两者都必须报告输入覆盖率。无法解析足够原始材料的 replay 结果不是 policy 的负面结果；它是不完整 replay。 +replay/rescore 是 promotion 所需的证据，不是每个 `shadow` 或 `advisory` consumer 都必须交付的运行时功能。缺少该 +证据的 consumer 不得晋升到 `enforcing`。 +Memory write gate 的 promotion evidence 另由 [#1930](https://github.com/oceanbase/powercontext/issues/1930) 跟踪。 + ## Promotion criteria policy 在超越 shadow 前必须先声明 promotion criteria。 @@ -394,8 +400,6 @@ PowerContext 已经有相关基础： # Unresolved questions -- `DecisionObservation` 的第一个具体存储目标是什么：结构化日志、evaluation artifact，还是私有内部 ledger？ -- 默认应保留哪些 observation，保留多久？ - policy definition 应以 Python constant、JSON/YAML resource，还是两者并存？ - 如果 #1742 调整为符合本 RFC，它应使用哪些数字 promotion criteria？ - 第一个托管后端启用前，是否需要先有面向用户的 hosted-provider privacy configuration 文档？ @@ -406,6 +410,5 @@ PowerContext 已经有相关基础： - 在 dashboard 中增加 policy registry 页面，展示 active policy、mode、fallback rate 和最近 replay 结果。 - 增加 CLI 命令，对最近 observation replay 一个 policy，并输出候选阈值。 - 为 decision policy 提供标准 evaluation bundle，供 Jev、Laya、本地规则和未来后端共用。 -- 增加带 retention controls 的私有 decision-observation ledger，并支持导出做离线 evaluation。 - Review Inbox 支持按 policy 分组，让 review item 反链到触发它的 policy observation。 - 面向气隙部署的 local-only decision backend，复用相同 policy 与 observation 契约。 diff --git a/src/powercontext/builtin/artifacts/memory/protocols.py b/src/powercontext/builtin/artifacts/memory/protocols.py index 616c030ab1..5e002b6494 100644 --- a/src/powercontext/builtin/artifacts/memory/protocols.py +++ b/src/powercontext/builtin/artifacts/memory/protocols.py @@ -18,12 +18,13 @@ from dataclasses import dataclass from enum import StrEnum -from typing import Protocol +from typing import Protocol, runtime_checkable from pydantic import BaseModel, ConfigDict from powercontext.artifacts import Artifact from powercontext.builtin.artifacts.memory.models import MemoryEntryInput, MemoryEntryVersion +from powercontext.builtin.decision_observations import DecisionObservation from powercontext.sources import Source @@ -78,6 +79,21 @@ class MemoryWriteGateRequest: candidates: tuple[str, ...] evidence: tuple[str, ...] expected_revision: int | None = None + scope_id: str = "unscoped" + operation_id: str | None = None + subject_refs: tuple[str, ...] = () + evidence_refs: tuple[str, ...] = () + observation_sink: MemoryWriteObservationSink | None = None + defer_observation_emission: bool = False + + +class MemoryWriteObservationSink(Protocol): + """Receive a bounded gate observation after its owning domain action is final.""" + + async def record(self, observation: DecisionObservation, /) -> None: + """Persist or collect one content-free observation.""" + + ... class MemoryWriteGate(Protocol): @@ -95,6 +111,18 @@ async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessm ... +@runtime_checkable +class MemoryWriteGatePreflight(Protocol): + """Optionally map a deterministic service-side rejection through the gate policy.""" + + async def assess_preflight( + self, request: MemoryWriteGateRequest, rejection: MemoryWriteAssessment, / + ) -> MemoryWriteAssessment: + """Return the mode-aware assessment for one bounded preflight rejection.""" + + ... + + class CandidatePipeline(Protocol): """Produce untrusted Memory candidates from canonical bounded evidence.""" diff --git a/src/powercontext/builtin/decision_observations.py b/src/powercontext/builtin/decision_observations.py new file mode 100644 index 0000000000..dcdb0c824e --- /dev/null +++ b/src/powercontext/builtin/decision_observations.py @@ -0,0 +1,159 @@ +# 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. + +"""Runtime-independent contracts for bounded decision-policy sidecars.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC, datetime +from enum import StrEnum +from typing import Annotated +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, JsonValue, model_validator + +from powercontext.builtin.inference import InferenceUsage + + +class DecisionPolicyMode(StrEnum): + """Runtime mode for one versioned decision policy.""" + + DISABLED = "disabled" + SHADOW = "shadow" + ADVISORY = "advisory" + ENFORCING = "enforcing" + + +class DecisionFailurePolicy(StrEnum): + """How a consumer treats backend failure when it owns a domain action.""" + + FAIL_OPEN = "fail_open" + FAIL_CLOSED = "fail_closed" + + +class DecisionPrivacyBoundary(StrEnum): + """Content boundary declared by a policy before backend calls.""" + + LOCAL_ONLY = "local_only" + HOSTED_REDACTED = "hosted_redacted" + REFERENCES_ONLY = "references_only" + NO_EXTERNAL_CALL = "no_external_call" + + +class DecisionCoverage(StrEnum): + """Whether a policy evaluation actually judged the supplied content.""" + + ADJUDICATED = "adjudicated" + UNADJUDICATED = "unadjudicated" + + +class DecisionVerdict(StrEnum): + """Domain-neutral verdict before an owning service maps it to an action.""" + + ALLOW = "allow" + DENY = "deny" + REVIEW = "review" + UNKNOWN = "unknown" + + +class DecisionAssessmentSource(StrEnum): + """The source that produced the assessment's substantive judgement.""" + + LOCAL_RULE = "local_rule" + DECISION_MODEL = "decision_model" + NONE = "none" + + +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class DecisionPolicy(_StrictModel): + """Versioned, reviewable policy manifest for one bounded runtime question.""" + + policy_id: Annotated[str, Field(min_length=1, max_length=256)] + decision_kind: Annotated[str, Field(min_length=1, max_length=128)] + version: Annotated[str, Field(min_length=1, max_length=64)] + consumer: Annotated[str, Field(min_length=1, max_length=128)] + mode: DecisionPolicyMode + failure_policy: DecisionFailurePolicy + privacy_boundary: DecisionPrivacyBoundary + local_rules: tuple[Annotated[str, Field(min_length=1, max_length=128)], ...] = () + question: Annotated[str, Field(min_length=1, max_length=8192)] + subject_selector: Annotated[str, Field(min_length=1, max_length=512)] + evidence_selector: Annotated[str, Field(min_length=1, max_length=512)] + outcome_mapping: Mapping[str, Annotated[str, Field(min_length=1, max_length=128)]] = Field(default_factory=dict) + promotion_criteria: tuple[Annotated[str, Field(min_length=1, max_length=512)], ...] = () + + +class DecisionAssessment(_StrictModel): + """Policy-level assessment before domain-specific action mapping.""" + + policy_id: Annotated[str, Field(min_length=1, max_length=256)] + policy_version: Annotated[str, Field(min_length=1, max_length=64)] + mode: DecisionPolicyMode + coverage: DecisionCoverage + verdict: DecisionVerdict + source: DecisionAssessmentSource + reason: Annotated[str | None, Field(min_length=1, max_length=4096)] = None + confidence: Annotated[float | None, Field(ge=0.0, le=1.0)] = None + used_fallback: bool = False + usage: InferenceUsage = Field(default_factory=lambda: InferenceUsage(requests=0)) + latency_ms: Annotated[float | None, Field(ge=0.0, allow_inf_nan=False)] = None + + @model_validator(mode="after") + def validate_fallback_coverage(self) -> DecisionAssessment: + if self.used_fallback and self.coverage is not DecisionCoverage.UNADJUDICATED: + raise ValueError("fallback assessments must be unadjudicated") # noqa: TRY003 + return self + + +class DecisionObservation(_StrictModel): + """Audit/replay sidecar for one policy evaluation attempt.""" + + observation_id: Annotated[str, Field(min_length=1, max_length=128)] = Field( + default_factory=lambda: f"decision-observation-{uuid4().hex}" + ) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + operation_id: Annotated[str, Field(min_length=1, max_length=256)] + scope_id: Annotated[str, Field(min_length=1, max_length=256)] + consumer: Annotated[str, Field(min_length=1, max_length=128)] + policy_id: Annotated[str, Field(min_length=1, max_length=256)] + policy_version: Annotated[str, Field(min_length=1, max_length=64)] + mode: DecisionPolicyMode + subject_refs: tuple[Annotated[str, Field(min_length=1, max_length=2200)], ...] = () + evidence_refs: tuple[Annotated[str, Field(min_length=1, max_length=2200)], ...] = () + privacy_boundary: DecisionPrivacyBoundary + privacy_outcome: Annotated[str | None, Field(min_length=1, max_length=128)] = None + provider_id: Annotated[str | None, Field(min_length=1, max_length=128)] = None + backend_model_id: Annotated[str | None, Field(min_length=1, max_length=256)] = None + model_policy_id: Annotated[str | None, Field(min_length=1, max_length=256)] = None + assessment: DecisionAssessment + final_action: Annotated[str, Field(min_length=1, max_length=128)] + fallback_reason: Annotated[str | None, Field(min_length=1, max_length=256)] = None + metadata: Mapping[str, JsonValue] = Field(default_factory=dict) + + +__all__ = [ + "DecisionAssessment", + "DecisionAssessmentSource", + "DecisionCoverage", + "DecisionFailurePolicy", + "DecisionObservation", + "DecisionPolicy", + "DecisionPolicyMode", + "DecisionPrivacyBoundary", + "DecisionVerdict", +] diff --git a/src/powercontext/builtin/persistence/__init__.py b/src/powercontext/builtin/persistence/__init__.py index 0e36bfe664..cfefdfd7b3 100644 --- a/src/powercontext/builtin/persistence/__init__.py +++ b/src/powercontext/builtin/persistence/__init__.py @@ -22,6 +22,7 @@ from powercontext.builtin.persistence.candidates import CandidateRepository from powercontext.builtin.persistence.connectors import ConnectorCheckpointRepository from powercontext.builtin.persistence.database import AsyncDatabase +from powercontext.builtin.persistence.decision_observations import DecisionObservationRepository from powercontext.builtin.persistence.errors import ( ArtifactProcessingLeadershipLostError, ArtifactProcessingWaveIncompleteError, @@ -78,6 +79,7 @@ "CompositeTopicMemoryIndex", "ConnectorCheckpointRepository", "DatabaseClosedError", + "DecisionObservationRepository", "ExternalSkillRepository", "GenerationConflictError", "IdentityMismatchError", diff --git a/src/powercontext/builtin/persistence/decision_observations.py b/src/powercontext/builtin/persistence/decision_observations.py new file mode 100644 index 0000000000..1543e34613 --- /dev/null +++ b/src/powercontext/builtin/persistence/decision_observations.py @@ -0,0 +1,170 @@ +# 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. + +"""Scope-private persistence for bounded RFC 1770 decision observations.""" + +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any + +from sqlalchemy import delete, insert, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncConnection + +from powercontext.builtin.decision_observations import DecisionObservation +from powercontext.builtin.persistence.codec import dump_model, load_model, stored_bytes +from powercontext.builtin.persistence.errors import StoredPayloadConflictError +from powercontext.builtin.persistence.tables import DECISION_OBSERVATIONS_TABLE +from powercontext.builtin.sources import validate_scope_id + +_KIND = "decision-observation" +_SAFE_REASON = re.compile(r"[a-z][a-z0-9_]{0,127}\Z") +_SAFE_REFERENCE = re.compile( + r"(?:candidate|evidence):[0-9]+|(?:candidate:[0-9]+ )?(?:entry|source|artifact):[A-Za-z0-9_.:@-]{1,2200}\Z" +) +_SAFE_METADATA_KEYS = frozenset({"candidate_count", "evidence_count", "incomplete_subject_count"}) + + +class DecisionObservationRepository: + """Append and read bounded decision sidecars without retaining input bodies.""" + + async def append(self, connection: AsyncConnection, observation: DecisionObservation, /) -> bool: + """Append an observation, accepting only an exact idempotent replay.""" + + _validate_observation(observation) + scope = validate_scope_id(observation.scope_id) + values = _values(observation, scope) + try: + await _ensure_sqlite_outer_transaction(connection) + async with connection.begin_nested(): + await connection.execute(insert(DECISION_OBSERVATIONS_TABLE).values(**values)) + except IntegrityError: + existing = ( + await connection.execute( + select(DECISION_OBSERVATIONS_TABLE.c.payload).where( + DECISION_OBSERVATIONS_TABLE.c.scope_id == scope, + DECISION_OBSERVATIONS_TABLE.c.observation_id == observation.observation_id, + ) + ) + ).scalar_one_or_none() + if existing is None: + raise + if stored_bytes(existing, column="payload") != values["payload"]: + raise StoredPayloadConflictError(_KIND, (scope, observation.observation_id)) from None + return False + return True + + async def observations(self, connection: AsyncConnection, scope_id: str, /) -> tuple[DecisionObservation, ...]: + """Read one Scope's sidecars in immutable creation order.""" + + scope = validate_scope_id(scope_id) + rows = ( + await connection.execute( + select(DECISION_OBSERVATIONS_TABLE) + .where(DECISION_OBSERVATIONS_TABLE.c.scope_id == scope) + .order_by( + DECISION_OBSERVATIONS_TABLE.c.created_at, + DECISION_OBSERVATIONS_TABLE.c.observation_id, + ) + ) + ).mappings() + return tuple(_decode(row) for row in rows) + + async def prune_before(self, connection: AsyncConnection, scope_id: str, before: datetime, /) -> int: + """Remove only this Scope's sidecars older than its explicit retention boundary.""" + + scope = validate_scope_id(scope_id) + result = await connection.execute( + delete(DECISION_OBSERVATIONS_TABLE).where( + DECISION_OBSERVATIONS_TABLE.c.scope_id == scope, + DECISION_OBSERVATIONS_TABLE.c.created_at < before, + ) + ) + return int(result.rowcount or 0) + + +def _values(observation: DecisionObservation, scope_id: str) -> dict[str, object]: + assessment = observation.assessment + usage = assessment.usage + return { + "scope_id": scope_id, + "observation_id": observation.observation_id, + "operation_id": observation.operation_id, + "created_at": observation.created_at, + "consumer": observation.consumer, + "policy_id": observation.policy_id, + "policy_version": observation.policy_version, + "mode": observation.mode.value, + "privacy_boundary": observation.privacy_boundary.value, + "coverage": assessment.coverage.value, + "verdict": assessment.verdict.value, + "assessment_source": assessment.source.value, + "final_action": observation.final_action, + "used_fallback": assessment.used_fallback, + "fallback_reason": observation.fallback_reason, + "provider_id": observation.provider_id, + "backend_model_id": observation.backend_model_id, + "model_policy_id": observation.model_policy_id, + "requests": usage.requests, + "input_tokens": usage.input_tokens, + "output_tokens": usage.output_tokens, + "latency_ms": assessment.latency_ms, + "payload": dump_model(observation, kind=_KIND, name=observation.observation_id), + } + + +def _decode(row: Any) -> DecisionObservation: + identifier = str(row["observation_id"]) + return load_model( + DecisionObservation, + stored_bytes(row["payload"], column="payload"), + kind=_KIND, + name=identifier, + ) + + +def _validate_observation(observation: DecisionObservation, /) -> None: + assessment = observation.assessment + if ( + assessment.policy_id != observation.policy_id + or assessment.policy_version != observation.policy_version + or assessment.mode is not observation.mode + ): + raise ValueError("decision observation assessment policy must match the observation") # noqa: TRY003 + if not all(_SAFE_REFERENCE.fullmatch(value) for value in (*observation.subject_refs, *observation.evidence_refs)): + raise ValueError("decision observation references must be structured identifiers") # noqa: TRY003 + for reason in (observation.assessment.reason, observation.fallback_reason): + if reason is not None and _SAFE_REASON.fullmatch(reason) is None: + raise ValueError("decision observation reasons must be bounded codes") # noqa: TRY003 + if set(observation.metadata) - _SAFE_METADATA_KEYS or any( + isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in observation.metadata.values() + ): + raise ValueError("decision observation metadata must contain only nonnegative safe counters") # noqa: TRY003 + + +async def _ensure_sqlite_outer_transaction(connection: AsyncConnection, /) -> None: + if connection.dialect.name != "sqlite" or not connection.in_transaction() or connection.in_nested_transaction(): + return + raw = await connection.get_raw_connection() + driver = getattr(raw, "driver_connection", None) + in_transaction = getattr(driver, "in_transaction", None) + if in_transaction is None: + in_transaction = getattr(getattr(driver, "_conn", None), "in_transaction", None) + if in_transaction is False: + await connection.exec_driver_sql("BEGIN") + + +__all__ = ["DecisionObservationRepository"] diff --git a/src/powercontext/builtin/persistence/tables.py b/src/powercontext/builtin/persistence/tables.py index 6725a5a34e..69c5191fa4 100644 --- a/src/powercontext/builtin/persistence/tables.py +++ b/src/powercontext/builtin/persistence/tables.py @@ -21,6 +21,7 @@ Column, Date, DateTime, + Float, ForeignKeyConstraint, Index, Integer, @@ -1096,6 +1097,46 @@ def _entry_text_type(): STATISTICS_TABLES = (MODEL_USAGE_DAILY_TABLE, RECALL_TOKEN_DAILY_TABLE, RECALL_EFFORT_DAILY_TABLE) +DECISION_OBSERVATIONS_TABLE = Table( + "pc_decision_observations", + SHARED_METADATA, + Column("scope_id", identity_string(MAX_SCOPE_ID_LENGTH), primary_key=True), + Column("observation_id", identity_string(128), primary_key=True), + Column("operation_id", identity_string(256), nullable=False), + Column("created_at", DateTime(timezone=True), nullable=False), + Column("consumer", identity_string(128), nullable=False), + Column("policy_id", identity_string(256), nullable=False), + Column("policy_version", identity_string(64), nullable=False), + Column("mode", identity_string(16), nullable=False), + Column("privacy_boundary", identity_string(32), nullable=False), + Column("coverage", identity_string(16), nullable=False), + Column("verdict", identity_string(16), nullable=False), + Column("assessment_source", identity_string(32), nullable=False), + Column("final_action", identity_string(128), nullable=False), + Column("used_fallback", Boolean, nullable=False), + Column("fallback_reason", identity_string(128)), + Column("provider_id", identity_string(128)), + Column("backend_model_id", identity_string(256)), + Column("model_policy_id", identity_string(256)), + Column("requests", BigInteger, nullable=False), + Column("input_tokens", BigInteger), + Column("output_tokens", BigInteger), + Column("latency_ms", Float), + Column("payload", _canonical_payload_type(), nullable=False), + Index("ix_pc_decision_observations_scope_created", "scope_id", "created_at", "observation_id"), + Index("ix_pc_decision_observations_operation", "scope_id", "operation_id", "created_at"), + CheckConstraint("requests >= 0", name="ck_pc_decision_observations_requests_nonnegative"), + CheckConstraint( + "input_tokens IS NULL OR input_tokens >= 0", name="ck_pc_decision_observations_input_tokens_nonnegative" + ), + CheckConstraint( + "output_tokens IS NULL OR output_tokens >= 0", name="ck_pc_decision_observations_output_tokens_nonnegative" + ), + CheckConstraint("latency_ms IS NULL OR latency_ms >= 0", name="ck_pc_decision_observations_latency_nonnegative"), +) + +DECISION_OBSERVATION_TABLES = (DECISION_OBSERVATIONS_TABLE,) + DREAM_RUNS_TABLE = Table( "pc_dream_runs", SHARED_METADATA, @@ -1221,6 +1262,7 @@ def _entry_text_type(): + TOPIC_MEMORY_TABLES + MEMORY_TABLES + STATISTICS_TABLES + + DECISION_OBSERVATION_TABLES + RECURRENCE_TABLES + (ARTIFACT_TAGS_TABLE, DREAM_RUNS_TABLE, RECEIPT_MIGRATION_REVIEW_TABLE) ) diff --git a/src/powercontext/builtin/runtime/__init__.py b/src/powercontext/builtin/runtime/__init__.py index 120c9a16e6..3cf2bc71d2 100644 --- a/src/powercontext/builtin/runtime/__init__.py +++ b/src/powercontext/builtin/runtime/__init__.py @@ -113,6 +113,18 @@ DecisionResult, StructuredDecisionModel, ) +from powercontext.builtin.runtime.decision_policy import ( + DecisionAssessment, + DecisionAssessmentSource, + DecisionCoverage, + DecisionFailurePolicy, + DecisionObservation, + DecisionPolicy, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, + assess_decision_result, +) from powercontext.builtin.runtime.errors import InvalidRuntimeRequestError, TopicMemoryProcessingUnavailableError from powercontext.builtin.runtime.memory_write_gate import ( DecisionMemoryWriteGate, @@ -237,15 +249,24 @@ "ContextAssemblySection", "CreateDreamRunRequest", "DatabaseConfig", + "DecisionAssessment", + "DecisionAssessmentSource", + "DecisionCoverage", + "DecisionFailurePolicy", "DecisionKind", "DecisionMemoryWriteGate", "DecisionModel", "DecisionModelOption", "DecisionModelRequest", "DecisionModelResult", + "DecisionObservation", "DecisionOutcome", + "DecisionPolicy", + "DecisionPolicyMode", + "DecisionPrivacyBoundary", "DecisionRequest", "DecisionResult", + "DecisionVerdict", "DreamApplication", "DreamRun", "DreamRunPage", @@ -377,6 +398,7 @@ "TopicMemoryProcessingUnavailableError", "UsageStatistics", "WorkApplication", + "assess_decision_result", "build_memory_write_gate", "dependency_readiness_probe", "open_builtin_contexts", diff --git a/src/powercontext/builtin/runtime/atomic_memory_processing.py b/src/powercontext/builtin/runtime/atomic_memory_processing.py index 0570aad02a..e3eb4d9721 100644 --- a/src/powercontext/builtin/runtime/atomic_memory_processing.py +++ b/src/powercontext/builtin/runtime/atomic_memory_processing.py @@ -16,13 +16,18 @@ from __future__ import annotations +import logging import math +from base64 import urlsafe_b64encode from contextlib import asynccontextmanager, nullcontext from dataclasses import dataclass, replace +from hashlib import sha256 from typing import TYPE_CHECKING, Literal, NoReturn from sqlalchemy import update +from sqlalchemy.exc import OperationalError +from powercontext._logging import log_safely from powercontext.artifacts import ArtifactLineage, ArtifactRef from powercontext.artifacts.search import ArtifactSearchExecutionContext from powercontext.builtin.artifacts.atomic_memory.errors import AtomicMemoryConflictError @@ -41,8 +46,14 @@ AtomicMemoryWorkingItem, ) from powercontext.builtin.artifacts.memory.canonical import canonical_embedding +from powercontext.builtin.artifacts.memory.protocols import ( + MemoryWriteGate, + MemoryWriteGateRequest, + MemoryWriteVerdict, +) from powercontext.builtin.artifacts.prompt.errors import PromptError from powercontext.builtin.artifacts.prompt.service import PromptService, current_prompt +from powercontext.builtin.decision_observations import DecisionObservation from powercontext.builtin.inference import ( InferenceTimeoutError, InferenceUnavailableError, @@ -51,11 +62,13 @@ ) from powercontext.builtin.persistence.atomic_memory_index import AtomicMemoryIndexError, AtomicMemoryRelatedRequest from powercontext.builtin.persistence.cursors import SourceCursorRepository +from powercontext.builtin.persistence.decision_observations import DecisionObservationRepository from powercontext.builtin.persistence.errors import GenerationConflictError from powercontext.builtin.persistence.memory_windows import MemorySourceWindowRepository from powercontext.builtin.persistence.sources import SourceRepository from powercontext.builtin.persistence.tables import SOURCE_JOURNAL_HEADS_TABLE from powercontext.builtin.runtime.atomic_memory import AtomicMemoryApplication, deferred_decision_audit +from powercontext.builtin.runtime.decision_policy import emit_decision_observation from powercontext.builtin.runtime.models import MemoryFlushResult from powercontext.builtin.runtime.processing_execution import ScopeInvocation from powercontext.builtin.runtime.protocols import RuntimeTracing @@ -67,6 +80,9 @@ from powercontext.builtin.runtime.config import RuntimeConfig +logger = logging.getLogger(__name__) + + @dataclass(frozen=True) class AtomicMemoryProcessingConfig: related_mode: Literal["auto", "fts", "vector", "hybrid"] = "auto" @@ -108,6 +124,9 @@ def __init__( pipeline: object | None, prompts: PromptService | None = None, config: AtomicMemoryProcessingConfig | None = None, + memory_write_gate: MemoryWriteGate | None = None, + decision_observations: DecisionObservationRepository | None = None, + decision_observation_retention_days: int | None = None, tracing: RuntimeTracing | None = None, ) -> None: self.application = application @@ -116,6 +135,9 @@ def __init__( self.pipeline = None if pipeline is None else require_atomic_memory_pipeline(pipeline) self.prompts = prompts self.config = config or AtomicMemoryProcessingConfig() + self.memory_write_gate = memory_write_gate + self.decision_observations = decision_observations + self.decision_observation_retention_days = decision_observation_retention_days self.windows = MemorySourceWindowRepository() self.tracing = tracing @@ -191,7 +213,7 @@ async def _flush_once(self, scope_id, limit, processing, context, span) -> Memor pipeline = self.pipeline if eligible else None try: workset, consulted_artifacts = await self._prepare(scope_id, eligible, pipeline, context) - prepared = await self._inspect_decisions(scope_id, workset, context) + prepared, decision, observation = await self._inspect_decisions(scope_id, workset, context) except (InferenceTimeoutError, InvalidInferenceOutputError) as error: reducible = (isinstance(error, InferenceTimeoutError) and error.operation == "generate") or ( isinstance(error, InvalidInferenceOutputError) and error.operation == "atomic-memory-input-budget" @@ -199,21 +221,36 @@ async def _flush_once(self, scope_id, limit, processing, context, span) -> Memor if reducible and action.through - action.after > 1: await self._reduce_window(scope_id, context, processing, cursor, state, action, high_watermark) raise - await self._publish( - scope_id, - context, - processing, - cursor, - transition, - action, - window, - eligible, - consulted_artifacts, - workset, - prepared, + try: + finalized = await self._publish( + scope_id, + context, + processing, + cursor, + transition, + action, + window, + eligible, + consulted_artifacts, + workset, + prepared, + decision, + observation, + high_watermark, + ) + except Exception: + self._emit_failed_observation(observation) + raise + if finalized is not None: + emit_decision_observation(finalized) + return self._result( + action.after, + action.through, high_watermark, + len(eligible), + held=decision is not None and decision.verdict is MemoryWriteVerdict.HOLD, + hold_codes=() if decision is None or decision.code is None else (decision.code.value,), ) - return self._result(action.after, action.through, high_watermark, len(eligible)) async def _reduce_window(self, scope_id, context, processing, cursor, state, action, high_watermark): async with self.application.database.transaction() as connection: @@ -263,7 +300,8 @@ async def _inspect_decisions(self, scope_id, workset, context): ) plans.append(plan) prepared = tuple([await application.service.prepare_change(plan) for plan in plans]) - return prepared + decision, observation = await self._assess_write_gate(scope_id, workset, prepared) + return prepared, decision, observation async def _publish( self, @@ -278,10 +316,13 @@ async def _publish( consulted_artifacts, workset, prepared, + decision, + observation, high_watermark, ): application = self.application writes = tuple(write for item in prepared for write in item.plan.writes) + finalized = None with self._stage( "memory.commit", { @@ -319,8 +360,20 @@ async def _publish( ) for ref in consulted_artifacts: await self._authorize_artifact(connection, scope_id, context, ref) - await application.service.commit_window( - connection, scope_id, prepared, context, read_set=tuple(workset.reads.values()) + results = () + if decision is None or decision.verdict is not MemoryWriteVerdict.HOLD: + results = await application.service.commit_window( + connection, scope_id, prepared, context, read_set=tuple(workset.reads.values()) + ) + finalized = await self._finalize_observation( + connection, + observation, + final_action=( + "memory_write_hold" + if decision is not None and decision.verdict is MemoryWriteVerdict.HOLD + else "memory_write_applied" + ), + subject_refs=tuple(_artifact_reference(result.primary.ref) for result in results), ) await self.cursors.save( connection, @@ -332,6 +385,7 @@ async def _publish( await self.windows.clear_consumed(connection, scope_id, action.through) if processing is not None: await processing.complete(connection, remaining_work=action.through < high_watermark) + return finalized async def _prepare(self, scope_id, eligible, pipeline, context): workset = AtomicMemoryWindowWorkset() @@ -574,8 +628,76 @@ def _require_window(window, after, through): if tuple(item.journal_position for item in window) != tuple(range(after + 1, through + 1)): _conflict("Source journal window is not an exact contiguous interval") + async def _assess_write_gate(self, scope_id, workset, prepared): + gate = self.memory_write_gate + if gate is None or not prepared: + return None, None + items = workset.changes() + collector = _PendingObservationSink() + request = MemoryWriteGateRequest( + candidates=tuple(item.content.text for item in items), + evidence=tuple(_gate_evidence_text(item) for item in items for item in item.evidence), + scope_id=scope_id, + operation_id=_window_operation_id(prepared), + subject_refs=tuple(f"candidate:{index}" for index in range(1, len(items) + 1)), + evidence_refs=tuple(dict.fromkeys(_evidence_reference(value) for item in items for value in item.evidence)), + observation_sink=collector, + defer_observation_emission=True, + ) + try: + decision = await gate.assess(request) + except Exception: + return None, None + return decision, collector.observation + + async def _finalize_observation(self, connection, observation, *, final_action, subject_refs): + if observation is None or self.decision_observations is None: + return None + finalized = observation.model_copy( + update={ + "final_action": final_action, + "subject_refs": observation.subject_refs if final_action == "memory_write_hold" else subject_refs, + } + ) + try: + async with connection.begin_nested(): + await self.decision_observations.append(connection, finalized) + if self.decision_observation_retention_days is not None: + from datetime import UTC, datetime, timedelta + + await self.decision_observations.prune_before( + connection, + finalized.scope_id, + datetime.now(UTC) - timedelta(days=self.decision_observation_retention_days), + ) + except Exception as error: + # SQLite RAISE(ROLLBACK, ...) invalidates the owner transaction and + # destroys this savepoint. Continuing would consume the source window + # in a new transaction after the Memory write was lost. + if ( + connection.dialect.name == "sqlite" + and isinstance(error, OperationalError) + and "no such savepoint" in str(error) + ): + raise + log_safely( + logger, + logging.WARNING, + "Decision observation persistence failed", + extra={"event": "decision.observation_sink_failed", "exception_type": type(error).__name__}, + ) + return None + return finalized + + @staticmethod + def _emit_failed_observation(observation): + if observation is not None: + emit_decision_observation( + observation.model_copy(update={"final_action": "memory_write_failed", "subject_refs": ()}) + ) + @staticmethod - def _result(after, through, high_watermark, count): + def _result(after, through, high_watermark, count, *, held=False, hold_codes=()): return MemoryFlushResult( previous_cursor=after, current_cursor=through, @@ -583,8 +705,46 @@ def _result(after, through, high_watermark, count): source_count=count, memory_ref=None, remaining_work=through < high_watermark, + held_count=1 if held else 0, + hold_codes=hold_codes, ) +@dataclass(slots=True) +class _PendingObservationSink: + observation: DecisionObservation | None = None + + async def record(self, observation: DecisionObservation, /) -> None: + self.observation = observation + + +def _gate_evidence_text(value) -> str: + if isinstance(value, AtomicMemoryEvidence): + import json + + return json.dumps(value.content, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return value.content.text + + +def _reference_component(value: str) -> str: + return urlsafe_b64encode(value.encode("utf-8")).decode("ascii").rstrip("=") + + +def _artifact_reference(reference: ArtifactRef) -> str: + return f"artifact:{_reference_component(reference.family)}:{_reference_component(reference.artifact_id)}@{reference.revision}" + + +def _evidence_reference(value) -> str: + if isinstance(value, AtomicMemoryEvidence): + reference = value.source_ref + return f"source:{_reference_component(reference.source_type)}:{_reference_component(reference.source_id)}" + return _artifact_reference(value.artifact_ref) + + +def _window_operation_id(prepared) -> str: + identities = ":".join(item.plan.primary_artifact_id for item in prepared) + return f"atomic-memory-window:{sha256(identities.encode('utf-8')).hexdigest()}" + + def _conflict(detail: str) -> NoReturn: raise AtomicMemoryConflictError(detail) diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index a5da80e56d..988560dcd9 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -23,8 +23,10 @@ from contextlib import AsyncExitStack, asynccontextmanager, nullcontext from datetime import timedelta from functools import partial +from ipaddress import ip_address from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast +from urllib.parse import urlsplit from pydantic import AnyHttpUrl, JsonValue, SecretStr from typing_extensions import override @@ -218,10 +220,6 @@ def __init__(self, issue: str) -> None: "custom inference endpoints require an OpenAI- or Anthropic-compatible model identifier" ), "inference-profile": "validated inference profile is incomplete", - "legacy-memory-write-gate": ( - "Atomic Memory does not support the legacy Memory write gate; explicitly disable " - "memory_write_gate_enabled and remove any injected memory_write_gate before upgrading or starting" - ), "memory-reranker": "Memory reranking requires a configured generation or rerank model, or injected reranker", "scheduled-experience-pipeline": "scheduled Experience incubation requires a candidate pipeline", "scheduled-pipeline": "scheduled Source processing requires a candidate pipeline", @@ -250,11 +248,6 @@ def __init__(self, issue: str) -> None: super().__init__(messages[issue]) -def _validate_legacy_memory_write_gate(runtime: RuntimeConfig, injected: MemoryWriteGate | None = None) -> None: - if runtime.memory_write_gate_enabled or injected is not None: - raise BuiltinConfigurationError("legacy-memory-write-gate") - - class _DefinitionEvidenceProjector(DefaultMemoryEvidenceProjector): def __init__(self, definitions: SourceDefinitionRegistry, projection: SourceProjectionKey) -> None: self._definitions = definitions @@ -320,6 +313,9 @@ def __init__(self, delegate: DecisionModel, tracing: RuntimeTracing) -> None: self._delegate = delegate self._tracing = tracing self.policy_id = delegate.policy_id + self.is_local_only = bool(getattr(delegate, "is_local_only", False)) + self.provider_id = getattr(delegate, "provider_id", None) + self.backend_model_id = getattr(delegate, "backend_model_id", None) async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: with self._tracing.stage( @@ -380,6 +376,8 @@ def _configured_memory_write_gate( enabled=True, hold_on=runtime.memory_write_gate_hold_on, threshold=runtime.memory_write_gate_threshold, + mode=runtime.memory_write_gate_mode, + privacy_boundary=runtime.memory_write_gate_privacy_boundary, ) if gate is None: log_safely( @@ -463,8 +461,6 @@ async def open_builtin_runtime( a second observation. A disabled gate invokes neither recorder nor sink. """ - _validate_legacy_memory_write_gate(config.runtime, memory_write_gate) - async with AsyncExitStack() as resources: configured_source_registry = source_registry or BUILTIN_SOURCE_REGISTRY _validate_processing_source_registry(config, artifact_processing_bindings, configured_source_registry) @@ -992,8 +988,6 @@ async def open_builtin_contexts( ) -> AsyncIterator[RelationalContexts]: """Open the selected database and expose scope-bound PowerContext providers.""" - _validate_legacy_memory_write_gate(config.runtime, memory_write_gate) - database = config.database configured_token_estimator = character_token_estimator() if token_estimator is None else token_estimator if isinstance(database, SQLiteConfig): @@ -1050,6 +1044,8 @@ async def open_builtin_contexts( token_estimator=configured_token_estimator, memory_reranker=memory_reranker, decision_model=decision_model, + memory_write_gate=memory_write_gate, + decision_observation_retention_days=config.runtime.decision_observation_retention_days, memory_rerank_candidate_limit=config.runtime.memory_rerank_candidate_limit, prompt_registry=prompt_registry, prompt_demonstrators=prompt_demonstrators, @@ -1242,8 +1238,6 @@ async def _generation_pipelines( ReadinessProbe | None, ReadinessProbe | None, ]: - _validate_legacy_memory_write_gate(runtime) - if ( settings.generation_model is None and (not runtime.memory_rerank_enabled or settings.rerank_model is None) @@ -1558,6 +1552,7 @@ async def _generation_decision( decision_provider_model = generation_provider_model decision_model = generation_model + decision_model_name = settings.decision_model or settings.generation_model inherits_generation = settings.decision_model is None decision_headers = ( _merge_headers(settings.generation_headers, settings.decision_headers) @@ -1566,7 +1561,6 @@ async def _generation_decision( ) separate_decision_model = settings.decision_model is not None or bool(settings.decision_headers) if separate_decision_model: - decision_model_name = settings.decision_model or settings.generation_model if decision_model_name is None: raise BuiltinConfigurationError("decision-model") decision_provider_model, decision_model = await _open_pydantic_ai_model( @@ -1602,7 +1596,16 @@ async def _generation_decision( model_settings=decision_request_settings, name="decision_evaluate", ) - generated_decision = LLMDecisionModel(UsageReportingStructuredGenerator(decision_generator)) + generated_decision = LLMDecisionModel( + UsageReportingStructuredGenerator(decision_generator), + is_local_only=_is_loopback_endpoint( + settings.decision_base_url if settings.decision_model is not None else settings.generation_base_url + ), + provider_id=None if decision_model_name is None else decision_model_name.partition(":")[0], + backend_model_id=None + if decision_model_name is None + else decision_model_name.partition(":")[2] or decision_model_name, + ) decision_readiness: ReadinessProbe | None = None if separate_decision_model or settings.decision_model_settings: @@ -1627,8 +1630,6 @@ async def probe_decision() -> None: async def preflight_builtin_runtime(config: BuiltinConfig) -> None: """Validate Runtime composition without opening persistence or making requests.""" - _validate_legacy_memory_write_gate(config.runtime) - async with AsyncExitStack() as resources: await _generation_pipelines( config.inference, @@ -1732,10 +1733,22 @@ def create_provider(provider_name: str) -> Provider[Any]: if headers: from openai import AsyncOpenAI + http_client = _endpoint_http_client(follow_redirects=False) if base_url is not None else None client = AsyncOpenAI( base_url=None if base_url is None else str(base_url), api_key=os.getenv("OPENAI_API_KEY") or "api-key-not-set", default_headers=_resolve_headers(headers), + http_client=http_client, + ) + resources.push_async_callback(client.close) + return OpenAIProvider(openai_client=client) + if base_url is not None and _is_loopback_endpoint(base_url): + from openai import AsyncOpenAI + + client = AsyncOpenAI( + base_url=str(base_url), + api_key=os.getenv("OPENAI_API_KEY") or "api-key-not-set", + http_client=_endpoint_http_client(follow_redirects=False), ) resources.push_async_callback(client.close) return OpenAIProvider(openai_client=client) @@ -1751,6 +1764,7 @@ def create_provider(provider_name: str) -> Provider[Any]: base_url=None if base_url is None else str(base_url), api_key=api_key or os.getenv("ANTHROPIC_API_KEY") or "api-key-not-set", default_headers=default_headers, + http_client=_endpoint_http_client(follow_redirects=False) if base_url is not None else None, ) resources.push_async_callback(client.close) return AnthropicProvider(anthropic_client=client) @@ -1763,6 +1777,34 @@ def create_provider(provider_name: str) -> Provider[Any]: return create_provider +def _endpoint_http_client(*, follow_redirects: bool): + """Create the SDK transport with an explicit redirect policy for custom endpoints.""" + + import httpx + + return httpx.AsyncClient(follow_redirects=follow_redirects) + + +def _is_loopback_endpoint(endpoint: AnyHttpUrl | str | None) -> bool: + """Allow only explicit local names and addresses for raw decision content.""" + + if endpoint is None: + return False + try: + host = urlsplit(str(endpoint)).hostname + except ValueError: + return False + if host is None: + return False + host = host.lower() + if host == "localhost": + return True + try: + return ip_address(host).is_loopback + except ValueError: + return False + + def _resolve_headers(headers: Mapping[str, SecretStr]) -> dict[str, str]: return {name: value.get_secret_value() for name, value in headers.items()} diff --git a/src/powercontext/builtin/runtime/config.py b/src/powercontext/builtin/runtime/config.py index 474621f81c..03b2795fdd 100644 --- a/src/powercontext/builtin/runtime/config.py +++ b/src/powercontext/builtin/runtime/config.py @@ -47,6 +47,7 @@ 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 +from powercontext.builtin.runtime.decision_policy import DecisionPolicyMode, DecisionPrivacyBoundary _HTTP_FIELD_NAME_PATTERN = re.compile(r"[!#$%&'*+\-.^_`|~0-9A-Za-z]+") _RECALL_GATE_BASE_MIN_SEMANTIC_SIMILARITY = 0.3 @@ -143,11 +144,14 @@ def reject_boolean_worker_quota(cls, value: Any) -> Any: memory_rerank_candidate_limit: int = Field(default=30, ge=1, le=100) decision_assistance_enabled: bool = False memory_write_gate_enabled: bool = False + memory_write_gate_mode: DecisionPolicyMode = DecisionPolicyMode.SHADOW + memory_write_gate_privacy_boundary: DecisionPrivacyBoundary = DecisionPrivacyBoundary.NO_EXTERNAL_CALL # Direction only: which verdict means "evidence is insufficient". The strength threshold # stays unset until a calibration probe establishes it, so a hold never depends on a made-up # number. memory_write_gate_hold_on: Literal["yes", "no"] = "yes" memory_write_gate_threshold: float | None = Field(default=None, ge=0.0, le=1.0) + decision_observation_retention_days: int | None = Field(default=None, ge=1, le=3650) memory_max_active_entries: int = Field(default=5_000, ge=1, le=100_000) memory_max_manifest_entries: int = Field(default=10_000, ge=1, le=200_000) memory_max_manifest_bytes: int = Field(default=4_194_304, ge=1_024, le=67_108_864) @@ -159,6 +163,8 @@ def reject_boolean_worker_quota(cls, value: Any) -> Any: def validate_memory_capacity_order(self) -> RuntimeConfig: if self.memory_max_active_entries > self.memory_max_manifest_entries: raise ValueError("memory_max_active_entries cannot exceed memory_max_manifest_entries") # noqa: TRY003 + if self.memory_write_gate_privacy_boundary is DecisionPrivacyBoundary.HOSTED_REDACTED: + raise ValueError("memory_write_gate_privacy_boundary does not support hosted_redacted without a sanitizer") # noqa: TRY003 return self recall_gate_enabled: bool = False diff --git a/src/powercontext/builtin/runtime/decision_model.py b/src/powercontext/builtin/runtime/decision_model.py index c30bb6a0ac..0128f9c325 100644 --- a/src/powercontext/builtin/runtime/decision_model.py +++ b/src/powercontext/builtin/runtime/decision_model.py @@ -144,8 +144,19 @@ class LLMDecisionModel: policy_id = DECISION_INSTRUCTIONS_VERSION - def __init__(self, generator: StructuredGenerator[DecisionInput, DecisionOutput], /) -> None: + def __init__( + self, + generator: StructuredGenerator[DecisionInput, DecisionOutput], + /, + *, + is_local_only: bool = False, + provider_id: str | None = None, + backend_model_id: str | None = None, + ) -> None: self._generator = generator + self.is_local_only = is_local_only + self.provider_id = provider_id + self.backend_model_id = backend_model_id async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: """Map the model's validated answer onto one portable result.""" @@ -177,6 +188,9 @@ class FailOpenDecisionModel: def __init__(self, delegate: DecisionModel, /, *, timeout_seconds: float | None = None) -> None: self._delegate = delegate self.policy_id = delegate.policy_id + self.is_local_only = bool(getattr(delegate, "is_local_only", False)) + self.provider_id = getattr(delegate, "provider_id", None) + self.backend_model_id = getattr(delegate, "backend_model_id", None) self._timeout_seconds = timeout_seconds async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: diff --git a/src/powercontext/builtin/runtime/decision_observation_ledger.py b/src/powercontext/builtin/runtime/decision_observation_ledger.py new file mode 100644 index 0000000000..8348868579 --- /dev/null +++ b/src/powercontext/builtin/runtime/decision_observation_ledger.py @@ -0,0 +1,58 @@ +# 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. + +"""Relational sidecar sink for bounded decision-policy observations.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from sqlalchemy.ext.asyncio import AsyncConnection + +from powercontext.builtin.decision_observations import DecisionObservation +from powercontext.builtin.persistence.database import AsyncDatabase +from powercontext.builtin.persistence.decision_observations import DecisionObservationRepository + + +class RelationalDecisionObservationSink: + """Persist a sidecar in the caller's transaction when one owns the domain write.""" + + def __init__( + self, + database: AsyncDatabase, + repository: DecisionObservationRepository, + /, + *, + retention_days: int | None = None, + connection: AsyncConnection | None = None, + ) -> None: + self._database = database + self._repository = repository + self._retention_days = retention_days + self._connection = connection + + async def record(self, observation: DecisionObservation, /) -> None: + """Append one sidecar, opening a short transaction only when no connection is bound.""" + + async with self._database.connection(self._connection) as connection: + await self._repository.append(connection, observation) + if self._retention_days is not None: + await self._repository.prune_before( + connection, + observation.scope_id, + datetime.now(UTC) - timedelta(days=self._retention_days), + ) + + +__all__ = ["RelationalDecisionObservationSink"] diff --git a/src/powercontext/builtin/runtime/decision_policy.py b/src/powercontext/builtin/runtime/decision_policy.py new file mode 100644 index 0000000000..7330167263 --- /dev/null +++ b/src/powercontext/builtin/runtime/decision_policy.py @@ -0,0 +1,202 @@ +# 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. + +"""Shared governance vocabulary for narrow runtime decision policies.""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping + +from powercontext._logging import log_safely +from powercontext.builtin.decision_observations import ( + DecisionAssessment, + DecisionAssessmentSource, + DecisionCoverage, + DecisionFailurePolicy, + DecisionObservation, + DecisionPolicy, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, +) +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionResult + +logger = logging.getLogger(__name__) + + +_DEFAULT_OUTCOME_MAPPING: Mapping[DecisionOutcome, DecisionVerdict] = { + DecisionOutcome.YES: DecisionVerdict.ALLOW, + DecisionOutcome.NO: DecisionVerdict.DENY, + DecisionOutcome.ABSTAIN: DecisionVerdict.UNKNOWN, +} + + +def assess_decision_result( + policy: DecisionPolicy, + result: DecisionResult, + /, + *, + outcome_mapping: Mapping[DecisionOutcome | str, DecisionVerdict | str] | None = None, + latency_ms: float | None = None, +) -> DecisionAssessment: + """Map a low-level decision result into the RFC 1770 assessment vocabulary.""" + + mapping = _normalized_outcome_mapping(policy, outcome_mapping) + if result.used_fallback: + return DecisionAssessment( + policy_id=policy.policy_id, + policy_version=policy.version, + mode=policy.mode, + coverage=DecisionCoverage.UNADJUDICATED, + verdict=DecisionVerdict.UNKNOWN, + source=DecisionAssessmentSource.NONE, + reason="decision_model_fallback", + confidence=result.confidence, + used_fallback=True, + usage=result.usage, + latency_ms=latency_ms, + ) + if result.outcome is DecisionOutcome.ABSTAIN: + return DecisionAssessment( + policy_id=policy.policy_id, + policy_version=policy.version, + mode=policy.mode, + coverage=DecisionCoverage.UNADJUDICATED, + verdict=DecisionVerdict.UNKNOWN, + source=DecisionAssessmentSource.DECISION_MODEL, + reason=_bounded_reason(result.rationale, default="abstain"), + confidence=result.confidence, + usage=result.usage, + latency_ms=latency_ms, + ) + return DecisionAssessment( + policy_id=policy.policy_id, + policy_version=policy.version, + mode=policy.mode, + coverage=DecisionCoverage.ADJUDICATED, + verdict=mapping[result.outcome], + source=DecisionAssessmentSource.DECISION_MODEL, + reason=_bounded_reason(result.rationale), + confidence=result.confidence, + usage=result.usage, + latency_ms=latency_ms, + ) + + +def unadjudicated_decision_assessment( + policy: DecisionPolicy, + /, + *, + reason: str, + used_fallback: bool = True, +) -> DecisionAssessment: + """Record a no-call policy outcome without claiming the content was judged.""" + + return DecisionAssessment( + policy_id=policy.policy_id, + policy_version=policy.version, + mode=policy.mode, + coverage=DecisionCoverage.UNADJUDICATED, + verdict=DecisionVerdict.UNKNOWN, + source=DecisionAssessmentSource.NONE, + reason=reason, + used_fallback=used_fallback, + usage=InferenceUsage(requests=0), + ) + + +def assess_local_rule( + policy: DecisionPolicy, + /, + *, + verdict: DecisionVerdict, + reason: str, +) -> DecisionAssessment: + """Record a deterministic policy rule that substantively judged the content.""" + + return DecisionAssessment( + policy_id=policy.policy_id, + policy_version=policy.version, + mode=policy.mode, + coverage=DecisionCoverage.ADJUDICATED, + verdict=verdict, + source=DecisionAssessmentSource.LOCAL_RULE, + reason=reason, + usage=InferenceUsage(requests=0), + ) + + +def emit_decision_observation(observation: DecisionObservation, /) -> None: + """Log bounded policy outcome fields without turning logs into a content ledger.""" + + log_safely( + logger, + logging.INFO, + "Decision policy evaluated", + extra={ + "event": "decision.observation", + "consumer": observation.consumer, + "policy_id": observation.policy_id, + "policy_version": observation.policy_version, + "mode": observation.mode.value, + "privacy_boundary": observation.privacy_boundary.value, + "privacy_outcome": observation.privacy_outcome, + "coverage": observation.assessment.coverage.value, + "verdict": observation.assessment.verdict.value, + "final_action": observation.final_action, + "fallback_reason": observation.fallback_reason, + "candidate_count": observation.metadata.get("candidate_count", 0), + "evidence_count": observation.metadata.get("evidence_count", 0), + }, + ) + + +def _normalized_outcome_mapping( + policy: DecisionPolicy, + override: Mapping[DecisionOutcome | str, DecisionVerdict | str] | None, +) -> dict[DecisionOutcome, DecisionVerdict]: + raw = override if override is not None else policy.outcome_mapping + normalized = dict(_DEFAULT_OUTCOME_MAPPING) + for key, value in raw.items(): + outcome = key if isinstance(key, DecisionOutcome) else DecisionOutcome(str(key)) + verdict = value if isinstance(value, DecisionVerdict) else DecisionVerdict(str(value)) + normalized[outcome] = verdict + normalized[DecisionOutcome.ABSTAIN] = DecisionVerdict.UNKNOWN + return normalized + + +def _bounded_reason(value: str | None, *, default: str | None = None) -> str | None: + if value is None: + return default + normalized = value.strip() + return normalized[:4096] if normalized else default + + +__all__ = [ + "DecisionAssessment", + "DecisionAssessmentSource", + "DecisionCoverage", + "DecisionFailurePolicy", + "DecisionObservation", + "DecisionPolicy", + "DecisionPolicyMode", + "DecisionPrivacyBoundary", + "DecisionVerdict", + "assess_decision_result", + "assess_local_rule", + "emit_decision_observation", + "unadjudicated_decision_assessment", +] diff --git a/src/powercontext/builtin/runtime/memory_write_gate.py b/src/powercontext/builtin/runtime/memory_write_gate.py index c3381ebafc..4c9bbf195c 100644 --- a/src/powercontext/builtin/runtime/memory_write_gate.py +++ b/src/powercontext/builtin/runtime/memory_write_gate.py @@ -25,6 +25,7 @@ import json import logging +from time import perf_counter from typing import Literal from powercontext._logging import log_safely @@ -42,6 +43,20 @@ DecisionRequest, DecisionResult, ) +from powercontext.builtin.runtime.decision_policy import ( + DecisionAssessment, + DecisionCoverage, + DecisionFailurePolicy, + DecisionObservation, + DecisionPolicy, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, + assess_decision_result, + assess_local_rule, + emit_decision_observation, + unadjudicated_decision_assessment, +) logger = logging.getLogger(__name__) @@ -55,6 +70,55 @@ _MAX_SUBJECT_LENGTH = 4000 _MAX_REASON_LENGTH = 512 _DEFAULT_REASON = "the cited evidence does not clearly support this memory write" +_MEMORY_WRITE_POLICY = DecisionPolicy( + policy_id="memory.write.evidence_sufficiency.v1", + decision_kind=DecisionKind.MEMORY_WRITE_GATE.value, + version="1", + consumer="memory_write_gate", + mode=DecisionPolicyMode.ENFORCING, + failure_policy=DecisionFailurePolicy.FAIL_OPEN, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, + local_rules=("candidate_batch_budget", "evidence_projection_budget"), + question=_GATE_QUESTION, + subject_selector="candidate_text", + evidence_selector="candidate_citations", + outcome_mapping={ + DecisionOutcome.YES.value: DecisionVerdict.DENY.value, + DecisionOutcome.NO.value: DecisionVerdict.ALLOW.value, + DecisionOutcome.ABSTAIN.value: DecisionVerdict.UNKNOWN.value, + }, + promotion_criteria=( + "labeled calibration and held-out evaluation", + "measured fallback, latency, cost, and friction", + "rollback path", + ), +) + + +def memory_write_policy( + *, + hold_on: DecisionOutcome = DecisionOutcome.YES, + threshold: float | None = None, + mode: DecisionPolicyMode = DecisionPolicyMode.SHADOW, + privacy_boundary: DecisionPrivacyBoundary = DecisionPrivacyBoundary.LOCAL_ONLY, +) -> DecisionPolicy: + """Return the immutable Memory write policy with its configured runtime boundaries.""" + + if hold_on is DecisionOutcome.ABSTAIN: + raise ValueError("the hold direction cannot be abstain") # noqa: TRY003 + pass_on = DecisionOutcome.NO if hold_on is DecisionOutcome.YES else DecisionOutcome.YES + return _MEMORY_WRITE_POLICY.model_copy( + update={ + "version": _policy_version(hold_on, threshold), + "mode": mode, + "privacy_boundary": privacy_boundary, + "outcome_mapping": { + hold_on.value: DecisionVerdict.DENY.value, + pass_on.value: DecisionVerdict.ALLOW.value, + DecisionOutcome.ABSTAIN.value: DecisionVerdict.UNKNOWN.value, + }, + } + ) def build_memory_write_gate( @@ -63,6 +127,8 @@ def build_memory_write_gate( enabled: bool, hold_on: Literal["yes", "no"] = "yes", threshold: float | None = None, + mode: DecisionPolicyMode = DecisionPolicyMode.SHADOW, + privacy_boundary: DecisionPrivacyBoundary = DecisionPrivacyBoundary.LOCAL_ONLY, ) -> MemoryWriteGate | None: """Build the opt-in gate, or return ``None`` while it stays disabled. @@ -70,9 +136,22 @@ def build_memory_write_gate( enabled; a disabled gate and a missing backend both resolve to ``None``. """ - if not enabled or decision_model is None: + if not enabled or mode is DecisionPolicyMode.DISABLED: return None - return DecisionMemoryWriteGate(decision_model, hold_on=DecisionOutcome(hold_on), threshold=threshold) + policy = memory_write_policy( + hold_on=DecisionOutcome(hold_on), + threshold=threshold, + mode=mode, + privacy_boundary=privacy_boundary, + ) + if decision_model is None and privacy_boundary is not DecisionPrivacyBoundary.NO_EXTERNAL_CALL: + return None + return DecisionMemoryWriteGate( + decision_model, + hold_on=DecisionOutcome(hold_on), + threshold=threshold, + policy=policy, + ) class DecisionMemoryWriteGate: @@ -87,31 +166,89 @@ class DecisionMemoryWriteGate: def __init__( self, - decision_model: DecisionModel, + decision_model: DecisionModel | None, /, *, - hold_on: DecisionOutcome = DecisionOutcome.YES, + hold_on: DecisionOutcome | None = None, threshold: float | None = None, + policy: DecisionPolicy | None = None, + allow_model_evaluation: bool | None = None, ) -> None: if hold_on is DecisionOutcome.ABSTAIN: raise ValueError("the hold direction cannot be abstain") # noqa: TRY003 + if policy is None: + resolved_hold_on = DecisionOutcome.YES if hold_on is None else hold_on + policy = memory_write_policy( + hold_on=resolved_hold_on, + threshold=threshold, + mode=DecisionPolicyMode.ENFORCING, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, + ) + else: + if policy.mode is DecisionPolicyMode.DISABLED: + raise ValueError("a disabled policy cannot construct a memory write gate") # noqa: TRY003 + resolved_hold_on = _policy_hold_on(policy) + if hold_on is not None and hold_on is not resolved_hold_on: + raise ValueError("the hold direction must match the supplied policy") # noqa: TRY003 + _validate_memory_write_policy(policy, hold_on=resolved_hold_on, threshold=threshold) self._decision_model = decision_model - self._hold_on = hold_on + self._hold_on = resolved_hold_on self._threshold = threshold - self.policy_id = decision_model.policy_id + self._policy = policy + self._allow_model_evaluation = ( + decision_model is not None + and self._policy.privacy_boundary is DecisionPrivacyBoundary.LOCAL_ONLY + and bool(getattr(decision_model, "is_local_only", False)) + and allow_model_evaluation is not False + ) + self.policy_id = "memory.write-gate.no-backend" if decision_model is None else decision_model.policy_id + + @property + def mode(self) -> DecisionPolicyMode: + """Expose the configured mode for composition and focused diagnostics.""" + + return self._policy.mode async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessment: """Judge one pending write and return a caller-visible verdict.""" if _subject_exceeds_limit(request.candidates): - assessment = MemoryWriteAssessment( + policy_assessment = assess_local_rule( + self._policy, + verdict=DecisionVerdict.DENY, + reason="candidate_batch_budget", + ) + held = MemoryWriteAssessment( verdict=MemoryWriteVerdict.HOLD, policy_id=self.policy_id, code=_rejection_code(request), reason="the candidate batch exceeds the gate assessment budget", ) + assessment = self._mode_assessment(held) + await self._observe(request, policy_assessment, assessment) + self._log(assessment) + return assessment + if not self._allow_model_evaluation: + policy_assessment = unadjudicated_decision_assessment(self._policy, reason="privacy_boundary") + assessment = MemoryWriteAssessment( + verdict=MemoryWriteVerdict.ACCEPT, + policy_id=self.policy_id, + used_fallback=True, + ) + await self._observe(request, policy_assessment, assessment) self._log(assessment) return assessment + if self._decision_model is None: + policy_assessment = unadjudicated_decision_assessment(self._policy, reason="decision_model_unavailable") + assessment = MemoryWriteAssessment( + verdict=MemoryWriteVerdict.ACCEPT, + policy_id=self.policy_id, + used_fallback=True, + ) + await self._observe(request, policy_assessment, assessment) + self._log(assessment) + return assessment + started = perf_counter() decision = await self._decision_model.evaluate( DecisionRequest( decision_kind=DecisionKind.MEMORY_WRITE_GATE.value, @@ -120,29 +257,131 @@ async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessm evidence=request.evidence, ) ) + latency_ms = (perf_counter() - started) * 1000 assessment = self._map(request, decision) + policy_assessment = _assess_memory_write_decision( + decision, hold_on=self._hold_on, policy=self._policy + ).model_copy(update={"latency_ms": latency_ms}) + await self._observe(request, policy_assessment, assessment) + self._log(assessment) + return assessment + + async def assess_preflight( + self, request: MemoryWriteGateRequest, rejection: MemoryWriteAssessment, / + ) -> MemoryWriteAssessment: + """Apply this policy's mode and observation handling to a service-side evidence rejection.""" + + policy_assessment = assess_local_rule( + self._policy, + verdict=DecisionVerdict.DENY, + reason="evidence_projection_budget", + ) + assessment = self._mode_assessment(rejection) + await self._observe(request, policy_assessment, assessment) self._log(assessment) return assessment def _map(self, request: MemoryWriteGateRequest, decision: DecisionResult) -> MemoryWriteAssessment: - if decision.used_fallback or decision.outcome is DecisionOutcome.ABSTAIN: + assessment = _assess_memory_write_decision(decision, hold_on=self._hold_on, policy=self._policy) + if assessment.coverage is DecisionCoverage.UNADJUDICATED: return MemoryWriteAssessment( verdict=MemoryWriteVerdict.ACCEPT, policy_id=decision.policy_id, - used_fallback=decision.used_fallback, + used_fallback=assessment.used_fallback, ) - if decision.outcome is not self._hold_on: + if self._policy.mode is DecisionPolicyMode.SHADOW: + return MemoryWriteAssessment( + verdict=MemoryWriteVerdict.ACCEPT, + policy_id=decision.policy_id, + used_fallback=assessment.used_fallback, + ) + if assessment.verdict is DecisionVerdict.ALLOW: return MemoryWriteAssessment(verdict=MemoryWriteVerdict.ACCEPT, policy_id=decision.policy_id) reason = _bounded_reason(decision.rationale) if self._threshold is not None and decision.confidence is not None and decision.confidence < self._threshold: return MemoryWriteAssessment(verdict=MemoryWriteVerdict.FLAG, policy_id=decision.policy_id, reason=reason) - return MemoryWriteAssessment( - verdict=MemoryWriteVerdict.HOLD, - policy_id=decision.policy_id, - code=_rejection_code(request), - reason=reason, + return self._mode_assessment( + MemoryWriteAssessment( + verdict=MemoryWriteVerdict.HOLD, + policy_id=decision.policy_id, + code=_rejection_code(request), + reason=reason, + ) ) + def _mode_assessment(self, assessment: MemoryWriteAssessment) -> MemoryWriteAssessment: + if self._policy.mode is DecisionPolicyMode.SHADOW: + return MemoryWriteAssessment(verdict=MemoryWriteVerdict.ACCEPT, policy_id=assessment.policy_id) + if self._policy.mode is DecisionPolicyMode.ADVISORY and assessment.verdict is MemoryWriteVerdict.HOLD: + return MemoryWriteAssessment( + verdict=MemoryWriteVerdict.FLAG, + policy_id=assessment.policy_id, + reason=assessment.reason, + ) + return assessment + + async def _observe( + self, + request: MemoryWriteGateRequest, + policy_assessment: DecisionAssessment, + assessment: MemoryWriteAssessment, + ) -> None: + observation = DecisionObservation( + operation_id=request.operation_id or f"memory-write:{request.expected_revision or 'new'}", + scope_id=request.scope_id, + consumer=self._policy.consumer, + policy_id=self._policy.policy_id, + policy_version=self._policy.version, + mode=self._policy.mode, + subject_refs=request.subject_refs + or tuple(f"candidate:{index}" for index in range(1, len(request.candidates) + 1)), + evidence_refs=request.evidence_refs + or tuple(f"evidence:{index}" for index in range(1, len(request.evidence) + 1)), + privacy_boundary=self._policy.privacy_boundary, + privacy_outcome=self._privacy_outcome(policy_assessment), + provider_id=_model_identity(self._decision_model, "provider_id"), + backend_model_id=_model_identity(self._decision_model, "backend_model_id"), + model_policy_id=None if policy_assessment.source.value == "none" else self.policy_id, + assessment=_safe_observation_assessment(policy_assessment), + final_action=f"memory_write_{assessment.verdict.value}", + fallback_reason=_safe_observation_reason(policy_assessment.reason) + if policy_assessment.used_fallback + else None, + metadata={"candidate_count": len(request.candidates), "evidence_count": len(request.evidence)}, + ) + if not request.defer_observation_emission: + emit_decision_observation(observation) + if request.observation_sink is None: + return + try: + await request.observation_sink.record(observation) + except Exception as error: + log_safely( + logger, + logging.WARNING, + "Decision observation sink failed", + extra={ + "event": "decision.observation_sink_failed", + "consumer": observation.consumer, + "policy_id": observation.policy_id, + "operation_id": observation.operation_id, + "error_type": type(error).__name__, + }, + ) + + def _privacy_outcome(self, assessment: DecisionAssessment) -> str: + if assessment.source.value == "local_rule": + return "local_rule" + if assessment.source.value == "decision_model": + return "local_model_called" + if self._policy.privacy_boundary is DecisionPrivacyBoundary.NO_EXTERNAL_CALL: + return "no_external_call" + if not self._allow_model_evaluation: + return "local_only_refused" + if assessment.used_fallback: + return "local_model_fallback" + return "no_call" + def _log(self, assessment: MemoryWriteAssessment) -> None: event = { MemoryWriteVerdict.HOLD: "memory.write-gate.hold", @@ -171,6 +410,83 @@ def _rejection_code(request: MemoryWriteGateRequest) -> MemoryWriteRejectionCode return MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE +def _assess_memory_write_decision( + decision: DecisionResult, + /, + *, + hold_on: DecisionOutcome, + policy: DecisionPolicy = _MEMORY_WRITE_POLICY, +) -> DecisionAssessment: + if hold_on is DecisionOutcome.ABSTAIN: + raise ValueError("the hold direction cannot be abstain") # noqa: TRY003 + pass_on = DecisionOutcome.NO if hold_on is DecisionOutcome.YES else DecisionOutcome.YES + return assess_decision_result( + policy, + decision, + outcome_mapping={ + hold_on: DecisionVerdict.DENY, + pass_on: DecisionVerdict.ALLOW, + DecisionOutcome.ABSTAIN: DecisionVerdict.UNKNOWN, + }, + ) + + +def _policy_version(hold_on: DecisionOutcome, threshold: float | None) -> str: + """Bind every action-changing calibration input into the reviewable policy version.""" + + threshold_value = "none" if threshold is None else format(threshold, ".17g") + return f"1;hold_on={hold_on.value};threshold={threshold_value}" + + +def _policy_hold_on(policy: DecisionPolicy, /) -> DecisionOutcome: + """Validate and recover the one decision outcome that the policy maps to a hold.""" + + holding = [ + DecisionOutcome(outcome) + for outcome, verdict in policy.outcome_mapping.items() + if verdict == DecisionVerdict.DENY.value + ] + if len(holding) != 1 or holding[0] is DecisionOutcome.ABSTAIN: + raise ValueError("the supplied policy must map exactly one decision outcome to deny") # noqa: TRY003 + return holding[0] + + +def _validate_memory_write_policy( + policy: DecisionPolicy, + /, + *, + hold_on: DecisionOutcome, + threshold: float | None, +) -> None: + """Reject a supplied policy whenever the specialized gate would not execute it literally.""" + + if policy.decision_kind != DecisionKind.MEMORY_WRITE_GATE.value: + raise ValueError("the supplied policy must describe the memory write gate") # noqa: TRY003 + if policy.consumer != "memory_write_gate": + raise ValueError("the supplied policy must use the memory write gate consumer") # noqa: TRY003 + if policy.failure_policy is not DecisionFailurePolicy.FAIL_OPEN: + raise ValueError("the supplied policy must fail open") # noqa: TRY003 + if policy.question != _GATE_QUESTION: + raise ValueError("the supplied policy question must match the memory write gate") # noqa: TRY003 + if policy.subject_selector != "candidate_text" or policy.evidence_selector != "candidate_citations": + raise ValueError("the supplied policy selectors must match the memory write gate") # noqa: TRY003 + pass_on = DecisionOutcome.NO if hold_on is DecisionOutcome.YES else DecisionOutcome.YES + expected_mapping = { + hold_on.value: DecisionVerdict.DENY.value, + pass_on.value: DecisionVerdict.ALLOW.value, + DecisionOutcome.ABSTAIN.value: DecisionVerdict.UNKNOWN.value, + } + if dict(policy.outcome_mapping) != expected_mapping: + raise ValueError("the supplied policy outcome mapping must match the memory write gate") # noqa: TRY003 + if policy.version != _policy_version(hold_on, threshold): + raise ValueError("the operational parameters must match the supplied policy version") # noqa: TRY003 + + +def _model_identity(model: DecisionModel | None, attribute: str, /) -> str | None: + value = None if model is None else getattr(model, attribute, None) + return value if isinstance(value, str) and value.strip() else None + + def _bounded_subject(candidates: tuple[str, ...]) -> str: return _candidate_subject(candidates)[:_MAX_SUBJECT_LENGTH] @@ -194,6 +510,24 @@ def _bounded_reason(value: str | None) -> str: return normalized[:_MAX_REASON_LENGTH] if normalized else _DEFAULT_REASON +def _safe_observation_assessment(assessment: DecisionAssessment, /) -> DecisionAssessment: + """Keep durable sidecars free of model-provided rationale text.""" + + return assessment.model_copy(update={"reason": _safe_observation_reason(assessment.reason)}) + + +def _safe_observation_reason(reason: str | None, /) -> str | None: + """Map gate-internal explanations to bounded machine-readable audit codes.""" + + return { + "candidate_batch_budget": "candidate_batch_budget", + "evidence_projection_budget": "evidence_projection_budget", + "privacy_boundary": "privacy_boundary", + "decision_model_fallback": "decision_model_fallback", + "abstain": "abstain", + }.get(reason, "decision_model_verdict" if reason is not None else None) + + __all__ = [ "DecisionMemoryWriteGate", "MemoryWriteAssessment", @@ -201,5 +535,7 @@ def _bounded_reason(value: str | None) -> str: "MemoryWriteGateRequest", "MemoryWriteRejectionCode", "MemoryWriteVerdict", + "_assess_memory_write_decision", "build_memory_write_gate", + "memory_write_policy", ] diff --git a/src/powercontext/builtin/runtime/relational.py b/src/powercontext/builtin/runtime/relational.py index f5651897c1..6e0ab08457 100644 --- a/src/powercontext/builtin/runtime/relational.py +++ b/src/powercontext/builtin/runtime/relational.py @@ -66,6 +66,7 @@ MemoryQueryEmbedding, MemoryReranker, ) +from powercontext.builtin.artifacts.memory.protocols import MemoryWriteGate from powercontext.builtin.artifacts.profile import Profile from powercontext.builtin.artifacts.profile.management import ProfileManagementWriter from powercontext.builtin.artifacts.profile.models import ProfileCandidateProposal @@ -130,6 +131,7 @@ from powercontext.builtin.persistence.connectors import ConnectorCheckpointRepository from powercontext.builtin.persistence.cursors import SourceCursorRepository from powercontext.builtin.persistence.database import AsyncDatabase +from powercontext.builtin.persistence.decision_observations import DecisionObservationRepository from powercontext.builtin.persistence.errors import RepositoryNotFoundError, StoredPayloadConflictError from powercontext.builtin.persistence.experience_index import ExperienceIndex, NoExperienceIndex from powercontext.builtin.persistence.external_skills import ExternalSkillRepository @@ -282,6 +284,7 @@ class _Repositories: processing_leases: ArtifactProcessingLeaseRepository processing_binding_states: ArtifactProcessingBindingStateRepository topic_memories: TopicMemoryRepository + decision_observations: DecisionObservationRepository @dataclass(frozen=True, slots=True) @@ -474,6 +477,8 @@ def __init__( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, decision_model: DecisionModel | None = None, + memory_write_gate: MemoryWriteGate | None = None, + decision_observation_retention_days: int | None = None, memory_rerank_candidate_limit: int = 30, id_factory: IdFactory | None = None, handoff_artifact_id: str = "handoff", @@ -560,6 +565,7 @@ def __init__( processing_leases=ArtifactProcessingLeaseRepository(), processing_binding_states=ArtifactProcessingBindingStateRepository(), topic_memories=topic_memory_repository, + decision_observations=DecisionObservationRepository(), ) self._model_usage = _ModelUsageRecorder( database, @@ -594,6 +600,9 @@ def __init__( pipeline=candidate_pipeline, prompts=self.prompts, config=atomic_memory_processing_config, + memory_write_gate=memory_write_gate, + decision_observations=self.repositories.decision_observations, + decision_observation_retention_days=decision_observation_retention_days, tracing=tracing, ) self._generation_receipts = HandoffGenerationReceipts( diff --git a/src/powercontext/server/logging.py b/src/powercontext/server/logging.py index c33141597a..1c5f13d67d 100644 --- a/src/powercontext/server/logging.py +++ b/src/powercontext/server/logging.py @@ -61,6 +61,17 @@ "source_count", "held_count", "hold_codes", + "consumer", + "policy_id", + "policy_version", + "privacy_boundary", + "privacy_outcome", + "coverage", + "verdict", + "final_action", + "fallback_reason", + "candidate_count", + "evidence_count", "trace_id", "span_id", *_PROCESSING_FIELDS, diff --git a/tests/builtin/persistence/test_decision_observation_repository.py b/tests/builtin/persistence/test_decision_observation_repository.py new file mode 100644 index 0000000000..59dc7dab02 --- /dev/null +++ b/tests/builtin/persistence/test_decision_observation_repository.py @@ -0,0 +1,228 @@ +# 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 +from datetime import UTC, datetime, timedelta + +import pytest + +from powercontext.builtin.persistence import DecisionObservationRepository +from powercontext.builtin.persistence.decision_observations import ( + DecisionObservationRepository as SameDecisionObservationRepository, +) +from powercontext.builtin.persistence.errors import StoredPayloadConflictError +from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile +from powercontext.builtin.persistence.tables import ( + DECISION_OBSERVATION_TABLES, + SHARED_TABLES, +) +from powercontext.builtin.runtime.decision_observation_ledger import RelationalDecisionObservationSink +from powercontext.builtin.runtime.decision_policy import ( + DecisionAssessment, + DecisionAssessmentSource, + DecisionCoverage, + DecisionObservation, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, +) + +SCOPE = "scope-a" +CREATED_AT = datetime(2026, 10, 10, 8, 30, tzinfo=UTC) + + +def _observation( + *, observation_id: str = "decision-observation-1", created_at: datetime = CREATED_AT +) -> DecisionObservation: + return DecisionObservation( + observation_id=observation_id, + created_at=created_at, + operation_id="memory-write:memory-a@2", + scope_id=SCOPE, + consumer="memory_write_gate", + policy_id="memory.write.evidence_sufficiency.v1", + policy_version="1", + mode=DecisionPolicyMode.ADVISORY, + subject_refs=("entry:entry-a@version-a",), + evidence_refs=("source:task:1",), + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, + provider_id="local-provider", + backend_model_id="decision-model-a", + model_policy_id="powercontext.decision.static.v1", + assessment=DecisionAssessment( + policy_id="memory.write.evidence_sufficiency.v1", + policy_version="1", + mode=DecisionPolicyMode.ADVISORY, + coverage=DecisionCoverage.ADJUDICATED, + verdict=DecisionVerdict.DENY, + source=DecisionAssessmentSource.DECISION_MODEL, + reason="decision_model_verdict", + ), + final_action="memory_write_flag", + metadata={"candidate_count": 1, "evidence_count": 1}, + ) + + +def test_repository_is_exported_from_the_persistence_package() -> None: + assert DecisionObservationRepository is SameDecisionObservationRepository + + +def test_decision_observation_tables_are_registered_in_builtin_tables() -> None: + from powercontext.builtin.persistence.tables import BUILTIN_TABLES + + assert {table.name for table in DECISION_OBSERVATION_TABLES} <= {table.name for table in BUILTIN_TABLES} + + +def test_repository_appends_and_reads_back_one_bounded_observation() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + async with SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile: + async with profile.database.transaction() as connection: + inserted = await repository.append(connection, _observation()) + async with profile.database.transaction() as connection: + stored = await repository.observations(connection, SCOPE) + + assert inserted is True + assert stored == (_observation(),) + + asyncio.run(scenario()) + + +def test_repository_accepts_gate_generated_ordinal_references() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + observation = _observation().model_copy( + update={"subject_refs": ("candidate:1",), "evidence_refs": ("evidence:1",)} + ) + async with SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile: + async with profile.database.transaction() as connection: + inserted = await repository.append(connection, observation) + async with profile.database.transaction() as connection: + stored = await repository.observations(connection, SCOPE) + + assert inserted is True + assert stored == (observation,) + + asyncio.run(scenario()) + + +def test_repository_treats_an_exact_observation_replay_as_idempotent() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + observation = _observation() + async with SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile: + async with profile.database.transaction() as connection: + first = await repository.append(connection, observation) + replayed = await repository.append(connection, observation) + async with profile.database.transaction() as connection: + stored = await repository.observations(connection, SCOPE) + + assert first is True + assert replayed is False + assert stored == (observation,) + + asyncio.run(scenario()) + + +def test_repository_rejects_an_observation_with_a_mismatched_assessment_policy() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + inconsistent = _observation().model_copy(update={"policy_id": "memory.write.other_policy.v1"}) + async with ( + SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile, + profile.database.transaction() as connection, + ): + with pytest.raises(ValueError, match="assessment policy"): + await repository.append(connection, inconsistent) + + asyncio.run(scenario()) + + +def test_repository_rejects_an_observation_id_reused_for_different_content() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + observation = _observation() + conflicting = observation.model_copy(update={"final_action": "memory_write_accept"}) + async with ( + SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile, + profile.database.transaction() as connection, + ): + await repository.append(connection, observation) + with pytest.raises(StoredPayloadConflictError): + await repository.append(connection, conflicting) + + asyncio.run(scenario()) + + +def test_repository_rejects_unbounded_metadata_before_persistence() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + forbidden = _observation().model_copy(update={"metadata": {"prompt": "secret candidate text"}}) + async with ( + SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile, + profile.database.transaction() as connection, + ): + with pytest.raises(ValueError, match="metadata"): + await repository.append(connection, forbidden) + + asyncio.run(scenario()) + + +def test_repository_prunes_only_observations_older_than_the_retention_boundary() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + old = _observation(observation_id="decision-observation-old", created_at=CREATED_AT - timedelta(days=1)) + current = _observation(observation_id="decision-observation-current") + async with ( + SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile, + profile.database.transaction() as connection, + ): + await repository.append(connection, old) + await repository.append(connection, current) + removed = await repository.prune_before(connection, SCOPE, CREATED_AT) + stored = await repository.observations(connection, SCOPE) + + assert removed == 1 + assert stored == (current,) + + asyncio.run(scenario()) + + +def test_relational_sink_prunes_only_the_current_scope_when_retention_is_configured() -> None: + async def scenario() -> None: + repository = DecisionObservationRepository() + current = _observation(observation_id="decision-observation-current", created_at=datetime.now(UTC)) + old_current_scope = _observation( + observation_id="decision-observation-old", created_at=current.created_at - timedelta(days=3) + ) + other_scope = _observation( + observation_id="decision-observation-other-scope", + created_at=current.created_at - timedelta(days=3), + ).model_copy(update={"scope_id": "scope-b"}) + async with SQLiteProfile.open(SQLiteConfig(), tables=SHARED_TABLES + DECISION_OBSERVATION_TABLES) as profile: + async with profile.database.transaction() as connection: + await repository.append(connection, old_current_scope) + await repository.append(connection, other_scope) + sink = RelationalDecisionObservationSink(profile.database, repository, retention_days=2) + await sink.record(current) + async with profile.database.transaction() as connection: + current_scope = await repository.observations(connection, SCOPE) + other_scope_observations = await repository.observations(connection, "scope-b") + + assert current_scope == (current,) + assert other_scope_observations == (other_scope,) + + asyncio.run(scenario()) diff --git a/tests/builtin/persistence/test_mysql_schema.py b/tests/builtin/persistence/test_mysql_schema.py index a95975bf1a..ae0b90a5b4 100644 --- a/tests/builtin/persistence/test_mysql_schema.py +++ b/tests/builtin/persistence/test_mysql_schema.py @@ -15,7 +15,7 @@ import re from pathlib import Path -from sqlalchemy import BigInteger, Date, Integer, LargeBinary, String, Table +from sqlalchemy import BigInteger, Date, DateTime, Integer, LargeBinary, String, Table from sqlalchemy.dialects import mysql from sqlalchemy.schema import CreateTable, ForeignKeyConstraint, PrimaryKeyConstraint, UniqueConstraint @@ -51,6 +51,8 @@ def _column_budget(column) -> int: return 4 if isinstance(column.type, Date): return 3 + if isinstance(column.type, DateTime): + return 8 raise _UnbudgetedColumnTypeError(column.type) diff --git a/tests/builtin/runtime/test_decision_composition.py b/tests/builtin/runtime/test_decision_composition.py index 10f616ad58..4bea56d89c 100644 --- a/tests/builtin/runtime/test_decision_composition.py +++ b/tests/builtin/runtime/test_decision_composition.py @@ -20,12 +20,14 @@ from typing import Any import pytest +from pydantic import AnyHttpUrl, SecretStr from powercontext.builtin.inference import InferenceUnavailableError, InferenceUsage from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime import ( BuiltinConfig, RuntimeConfig, + composition, open_builtin_runtime, preflight_builtin_runtime, ) @@ -186,3 +188,63 @@ def test_decision_role_is_not_registered_as_an_mcp_tool() -> None: from powercontext.server import mcp assert all("decision" not in operation_id for operation_id in mcp._MCP_OPERATION_IDS) + + +@pytest.mark.parametrize( + ("endpoint", "expected"), + [ + ("http://localhost:11434/v1", True), + ("http://127.0.0.1:11434/v1", True), + ("http://[::1]:11434/v1", True), + ("http://0.0.0.0:11434/v1", False), + ("http://192.168.1.2:11434/v1", False), + ("https://api.example.test/v1", False), + ], +) +def test_only_explicit_loopback_decision_endpoints_are_local(endpoint: str, expected: bool) -> None: + assert composition._is_loopback_endpoint(AnyHttpUrl(endpoint)) is expected + + +@pytest.mark.parametrize( + ("endpoint", "expected"), + [ + ("http://127.0.0.1:11434/v1", True), + ("https://api.example.test/v1", False), + ], +) +def test_loopback_classifier_accepts_runtime_string_overrides(endpoint: str, expected: bool) -> None: + assert composition._is_loopback_endpoint(endpoint) is expected + + +def test_loopback_openai_provider_disables_redirects() -> None: + async def scenario() -> None: + async with AsyncExitStack() as resources: + factory = composition._unbounded_provider_factory( + AnyHttpUrl("http://127.0.0.1:11434/v1"), + {}, + workload="generation", + resources=resources, + ) + + provider = factory("openai-chat") + + assert provider.client._client.follow_redirects is False # pyright: ignore[reportPrivateUsage] + + asyncio.run(scenario()) + + +def test_loopback_anthropic_headers_provider_disables_redirects() -> None: + async def scenario() -> None: + async with AsyncExitStack() as resources: + factory = composition._unbounded_provider_factory( + AnyHttpUrl("http://127.0.0.1:11434"), + {"x-api-key": SecretStr("test-key")}, + workload="generation", + resources=resources, + ) + + provider = factory("anthropic") + + assert provider.client._client.follow_redirects is False # pyright: ignore[reportPrivateUsage] + + asyncio.run(scenario()) diff --git a/tests/builtin/runtime/test_decision_config.py b/tests/builtin/runtime/test_decision_config.py index fd15467073..26320cc140 100644 --- a/tests/builtin/runtime/test_decision_config.py +++ b/tests/builtin/runtime/test_decision_config.py @@ -14,19 +14,44 @@ from __future__ import annotations -from typing import Any +from typing import Any, cast import pytest from pydantic import AnyHttpUrl, SecretStr, ValidationError from powercontext.builtin.runtime import RuntimeConfig from powercontext.builtin.runtime.config import InferenceConfig +from powercontext.builtin.runtime.decision_policy import DecisionPolicyMode, DecisionPrivacyBoundary def test_decision_assistance_is_disabled_by_default() -> None: assert RuntimeConfig().decision_assistance_enabled is False +def test_memory_write_gate_defaults_to_no_external_call() -> None: + assert RuntimeConfig().memory_write_gate_privacy_boundary is DecisionPrivacyBoundary.NO_EXTERNAL_CALL + + +def test_memory_write_gate_defaults_to_shadow_mode() -> None: + assert RuntimeConfig().memory_write_gate_mode is DecisionPolicyMode.SHADOW + + +def test_decision_observation_retention_is_disabled_until_explicitly_configured() -> None: + assert RuntimeConfig().decision_observation_retention_days is None + assert RuntimeConfig(decision_observation_retention_days=30).decision_observation_retention_days == 30 + + +@pytest.mark.parametrize("retention_days", [0, -1]) +def test_decision_observation_retention_rejects_nonpositive_values(retention_days: int) -> None: + with pytest.raises(ValidationError): + RuntimeConfig(decision_observation_retention_days=retention_days) + + +def test_memory_write_gate_rejects_hosted_redaction_without_a_sanitizer() -> None: + with pytest.raises(ValidationError, match="hosted_redacted"): + RuntimeConfig(memory_write_gate_privacy_boundary=cast(DecisionPrivacyBoundary, "hosted_redacted")) + + def test_decision_inference_defaults_are_unset() -> None: config = InferenceConfig() diff --git a/tests/builtin/runtime/test_decision_policy.py b/tests/builtin/runtime/test_decision_policy.py new file mode 100644 index 0000000000..837bdcead0 --- /dev/null +++ b/tests/builtin/runtime/test_decision_policy.py @@ -0,0 +1,152 @@ +# 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 pytest +from pydantic import ValidationError + +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionResult +from powercontext.builtin.runtime.decision_policy import ( + DecisionAssessment, + DecisionAssessmentSource, + DecisionCoverage, + DecisionFailurePolicy, + DecisionObservation, + DecisionPolicy, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, + assess_decision_result, +) + + +def _policy() -> DecisionPolicy: + return DecisionPolicy( + policy_id="memory.write.evidence_sufficiency.v1", + decision_kind="memory.write-gate", + version="1", + consumer="memory_write_gate", + mode=DecisionPolicyMode.SHADOW, + failure_policy=DecisionFailurePolicy.FAIL_OPEN, + privacy_boundary=DecisionPrivacyBoundary.REFERENCES_ONLY, + question="Does the cited evidence fail to support the candidate?", + subject_selector="candidate_text", + evidence_selector="candidate_citations", + outcome_mapping={ + "yes": "deny", + "no": "allow", + "abstain": "unknown", + }, + promotion_criteria=("replay evidence exists",), + ) + + +def test_fallback_assessment_is_unadjudicated_even_when_domain_will_pass_through() -> None: + assessment = assess_decision_result( + _policy(), + DecisionResult( + DecisionOutcome.ABSTAIN, + "backend.policy.v1", + InferenceUsage(requests=0), + used_fallback=True, + ), + ) + + assert assessment.coverage is DecisionCoverage.UNADJUDICATED + assert assessment.verdict is DecisionVerdict.UNKNOWN + assert assessment.source is DecisionAssessmentSource.NONE + assert assessment.used_fallback is True + assert assessment.reason == "decision_model_fallback" + + +def test_abstention_is_unknown_without_claiming_content_was_judged() -> None: + assessment = assess_decision_result( + _policy(), + DecisionResult( + DecisionOutcome.ABSTAIN, + "backend.policy.v1", + InferenceUsage(requests=1), + rationale="not enough evidence", + ), + ) + + assert assessment.coverage is DecisionCoverage.UNADJUDICATED + assert assessment.verdict is DecisionVerdict.UNKNOWN + assert assessment.source is DecisionAssessmentSource.DECISION_MODEL + assert assessment.used_fallback is False + assert assessment.reason == "not enough evidence" + + +def test_substantive_decision_result_uses_policy_outcome_mapping() -> None: + yes = assess_decision_result( + _policy(), + DecisionResult( + DecisionOutcome.YES, + "backend.policy.v1", + InferenceUsage(requests=1, input_tokens=3, output_tokens=1), + confidence=0.8, + ), + ) + no = assess_decision_result( + _policy(), + DecisionResult(DecisionOutcome.NO, "backend.policy.v1", InferenceUsage(requests=1)), + ) + + assert yes.coverage is DecisionCoverage.ADJUDICATED + assert yes.verdict is DecisionVerdict.DENY + assert yes.confidence == 0.8 + assert yes.usage.input_tokens == 3 + assert no.coverage is DecisionCoverage.ADJUDICATED + assert no.verdict is DecisionVerdict.ALLOW + + +def test_assessment_rejects_fallback_that_claims_adjudicated_coverage() -> None: + with pytest.raises(ValidationError, match="fallback assessments must be unadjudicated"): + DecisionAssessment( + policy_id="memory.write.evidence_sufficiency.v1", + policy_version="1", + mode=DecisionPolicyMode.SHADOW, + coverage=DecisionCoverage.ADJUDICATED, + verdict=DecisionVerdict.ALLOW, + source=DecisionAssessmentSource.NONE, + used_fallback=True, + usage=InferenceUsage(requests=0), + ) + + +def test_observation_keeps_policy_assessment_distinct_from_domain_action() -> None: + assessment = assess_decision_result( + _policy(), + DecisionResult(DecisionOutcome.YES, "backend.policy.v1", InferenceUsage(requests=1)), + ) + + observation = DecisionObservation( + operation_id="remember:scope-a:42", + scope_id="scope-a", + consumer="memory_write_gate", + policy_id=assessment.policy_id, + policy_version=assessment.policy_version, + mode=assessment.mode, + subject_refs=("candidate:1",), + evidence_refs=("source:task:1",), + privacy_boundary=DecisionPrivacyBoundary.REFERENCES_ONLY, + assessment=assessment, + final_action="memory_write_hold", + ) + + assert observation.assessment.verdict is DecisionVerdict.DENY + assert observation.final_action == "memory_write_hold" + assert observation.evidence_refs == ("source:task:1",) diff --git a/tests/builtin/runtime/test_family_processing.py b/tests/builtin/runtime/test_family_processing.py index f4f5c3658b..13700d48b0 100644 --- a/tests/builtin/runtime/test_family_processing.py +++ b/tests/builtin/runtime/test_family_processing.py @@ -24,8 +24,6 @@ import pytest from sqlalchemy import func, select -import powercontext.builtin.runtime.composition as composition -import powercontext.builtin.runtime.family_processing as family_processing from powercontext.builtin.artifacts.atomic_memory.extraction import AtomicMemoryCandidate, AtomicMemoryExtractionOutput from powercontext.builtin.artifacts.experience import ExperienceCandidateInput, ExperienceContent from powercontext.builtin.inference import InferenceTimeoutError @@ -461,43 +459,6 @@ async def scenario(): asyncio.run(scenario()) -def test_spawned_memory_worker_rejects_legacy_gate_before_model_creation(monkeypatch, tmp_path): - def unexpected_initialization(*_args, **_kwargs): - pytest.fail("Legacy gate configuration must fail before model and worker initialization") - - monkeypatch.setattr(composition, "_embedding_models", unexpected_initialization) - monkeypatch.setattr(composition, "_prompt_registry", unexpected_initialization) - monkeypatch.setattr(composition, "open_builtin_contexts", unexpected_initialization) - monkeypatch.setattr(family_processing, "process_family_invocation", unexpected_initialization) - - async def scenario(): - database_path = tmp_path / "worker-gate.db" - config = BuiltinConfig( - database=SQLiteConfig(url=f"sqlite+aiosqlite:///{database_path}"), - inference=InferenceConfig(generation_model="test"), - runtime=RuntimeConfig(memory_write_gate_enabled=True), - ) - assignment = ArtifactProcessingWorkAssignment( - binding_name=FAMILY_BINDINGS["memory"], - scope_id="scope-a", - artifact_family="memory", - claimed_request_generation=1, - fence=ArtifactProcessingFence( - supervisor_group="global", - holder_id="worker-test", - supervisor_generation=1, - lease_mode="single-process", - ), - worker_id="worker-1", - ) - - with pytest.raises(composition.BuiltinConfigurationError, match="legacy Memory write gate"): - await family_processing._run_family_worker(FamilyWorkerSpec(config=config), assignment) - assert not database_path.exists() - - asyncio.run(scenario()) - - def test_memory_worker_completion_preserves_hold_details(tmp_path): async def scenario(): assignment = ArtifactProcessingWorkAssignment( diff --git a/tests/builtin/runtime/test_memory_window_recovery.py b/tests/builtin/runtime/test_memory_window_recovery.py index a5d79d99ce..22fafb2c62 100644 --- a/tests/builtin/runtime/test_memory_window_recovery.py +++ b/tests/builtin/runtime/test_memory_window_recovery.py @@ -18,14 +18,20 @@ from uuid import uuid4 import pytest +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError, OperationalError from powercontext.builtin.artifacts.atomic_memory.extraction import AtomicMemoryCandidate, AtomicMemoryExtractionOutput +from powercontext.builtin.decision_observations import DecisionPrivacyBoundary from powercontext.builtin.inference import GenerationResult, InferenceTimeoutError, InferenceUnavailableError +from powercontext.builtin.persistence.decision_observations import DecisionObservationRepository from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile from powercontext.builtin.persistence.tables import BUILTIN_TABLES from powercontext.builtin.runtime.composition import _initialize_atomic_memory_authority +from powercontext.builtin.runtime.memory_write_gate import build_memory_write_gate from powercontext.builtin.runtime.relational import RelationalContexts from powercontext.builtin.scope import ScopeDraft +from powercontext.server.logging import JsonFormatter from tests.e2e.dream_support import atomic_memory_pipeline, memory_source_text @@ -124,6 +130,205 @@ async def scenario(): asyncio.run(scenario()) +def test_atomic_memory_gate_finalizes_one_applied_observation_in_the_commit_transaction( + tmp_path, caplog: pytest.LogCaptureFixture +): + """A no-call gate still has one durable, final audit outcome after Atomic Memory publishes.""" + + async def scenario(): + gate = build_memory_write_gate( + None, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + assert gate is not None + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile: + contexts = await _ready_contexts( + profile, + candidate_pipeline=atomic_memory_pipeline(Pipeline()), + memory_write_gate=gate, + ) + scope = await create_scope(contexts, 1) + context = await contexts.get(scope) + + assert (await context.triggers.flush(limit=1)).current_cursor == 1 + + async with profile.database.transaction() as connection: + observations = await DecisionObservationRepository().observations(connection, scope) + assert len(observations) == 1 + observation = observations[0] + assert observation.final_action == "memory_write_applied" + assert observation.privacy_outcome == "no_external_call" + assert observation.subject_refs + + caplog.set_level("INFO", logger="powercontext.builtin.runtime.decision_policy") + asyncio.run(scenario()) + events = [record for record in caplog.records if getattr(record, "event", None) == "decision.observation"] + assert [record.__dict__["final_action"] for record in events] == ["memory_write_applied"] + + +def test_atomic_memory_gate_emits_failed_outcome_after_late_transaction_rollback( + tmp_path, caplog: pytest.LogCaptureFixture +): + async def scenario(): + gate = build_memory_write_gate( + None, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + assert gate is not None + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-rollback.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile: + contexts = await _ready_contexts( + profile, + candidate_pipeline=atomic_memory_pipeline(Pipeline()), + memory_write_gate=gate, + ) + scope = await create_scope(contexts, 1) + context = await contexts.get(scope) + async with profile.database.transaction() as connection: + await connection.execute( + text( + "CREATE TRIGGER reject_memory_cursor " + "BEFORE INSERT ON pc_source_cursors " + "BEGIN SELECT RAISE(ABORT, 'cursor rejected'); END" + ) + ) + + with pytest.raises(IntegrityError, match="cursor rejected"): + await context.triggers.flush(limit=1) + + assert (await context.triggers.cursor()).sequence == 0 + assert not (await contexts.atomic_memory.for_scope(scope).list()).items + async with profile.database.transaction() as connection: + assert not await DecisionObservationRepository().observations(connection, scope) + + caplog.set_level("INFO", logger="powercontext.builtin.runtime.decision_policy") + asyncio.run(scenario()) + events = [record for record in caplog.records if getattr(record, "event", None) == "decision.observation"] + assert [record.__dict__["final_action"] for record in events] == ["memory_write_failed"] + + +def test_atomic_memory_audit_failure_does_not_block_shadow_write(tmp_path, caplog: pytest.LogCaptureFixture): + async def scenario(): + gate = build_memory_write_gate( + None, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + assert gate is not None + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-audit.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile: + contexts = await _ready_contexts( + profile, + candidate_pipeline=atomic_memory_pipeline(Pipeline()), + memory_write_gate=gate, + ) + scope = await create_scope(contexts, 1) + context = await contexts.get(scope) + async with profile.database.transaction() as connection: + await connection.execute( + text( + "CREATE TRIGGER reject_decision_observation " + "BEFORE INSERT ON pc_decision_observations " + "BEGIN SELECT RAISE(ABORT, 'audit rejected'); END" + ) + ) + + assert (await context.triggers.flush(limit=1)).current_cursor == 1 + assert len((await contexts.atomic_memory.for_scope(scope).list()).items) == 1 + async with profile.database.transaction() as connection: + assert not await DecisionObservationRepository().observations(connection, scope) + + caplog.set_level("WARNING", logger="powercontext.builtin.runtime.atomic_memory_processing") + asyncio.run(scenario()) + diagnostics = [ + record for record in caplog.records if getattr(record, "event", None) == "decision.observation_sink_failed" + ] + assert len(diagnostics) == 1 + + +def test_atomic_memory_audit_rollback_does_not_consume_source_window(tmp_path): + """An audit failure that rolls back the owner transaction must retain the source window.""" + + async def scenario(): + gate = build_memory_write_gate( + None, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + assert gate is not None + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-audit-rollback.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile: + contexts = await _ready_contexts( + profile, + candidate_pipeline=atomic_memory_pipeline(Pipeline()), + memory_write_gate=gate, + ) + scope = await create_scope(contexts, 1) + context = await contexts.get(scope) + async with profile.database.transaction() as connection: + await connection.execute( + text( + "CREATE TRIGGER rollback_decision_observation " + "BEFORE INSERT ON pc_decision_observations " + "BEGIN SELECT RAISE(ROLLBACK, 'audit rolled back'); END" + ) + ) + + with pytest.raises(OperationalError, match="savepoint"): + await context.triggers.flush(limit=1) + assert (await context.triggers.cursor()).sequence == 0 + assert not (await contexts.atomic_memory.for_scope(scope).list()).items + + asyncio.run(scenario()) + + +def test_atomic_memory_audit_failure_log_omits_database_exception(tmp_path, caplog: pytest.LogCaptureFixture): + """Audit diagnostics expose only a bounded exception class, never a driver payload or traceback.""" + + marker = "x" * 8192 + + async def scenario(): + gate = build_memory_write_gate( + None, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + assert gate is not None + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-audit-log.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile: + contexts = await _ready_contexts( + profile, + candidate_pipeline=atomic_memory_pipeline(Pipeline()), + memory_write_gate=gate, + ) + scope = await create_scope(contexts, 1) + context = await contexts.get(scope) + async with profile.database.transaction() as connection: + await connection.execute( + text( + "CREATE TRIGGER reject_decision_observation_with_marker " + "BEFORE INSERT ON pc_decision_observations " + f"BEGIN SELECT RAISE(ABORT, '{marker}'); END" + ) + ) + + assert (await context.triggers.flush(limit=1)).current_cursor == 1 + + caplog.set_level("WARNING", logger="powercontext.builtin.runtime.atomic_memory_processing") + asyncio.run(scenario()) + diagnostic = next( + record for record in caplog.records if getattr(record, "event", None) == "decision.observation_sink_failed" + ) + rendered = JsonFormatter().format(diagnostic) + assert diagnostic.exc_info is None + assert diagnostic.__dict__["exception_type"] == "IntegrityError" + assert marker not in rendered + assert '"exception_type":"IntegrityError"' in rendered + + @pytest.mark.parametrize("error", [InferenceTimeoutError("embed", 60), InferenceUnavailableError("generate")]) def test_non_extraction_failures_do_not_reduce_source_window(error): async def scenario(): diff --git a/tests/builtin/runtime/test_memory_write_gate_contract.py b/tests/builtin/runtime/test_memory_write_gate_contract.py index 04087ab462..a6843cebab 100644 --- a/tests/builtin/runtime/test_memory_write_gate_contract.py +++ b/tests/builtin/runtime/test_memory_write_gate_contract.py @@ -15,6 +15,7 @@ from __future__ import annotations import asyncio +import logging import pytest @@ -26,13 +27,22 @@ DecisionResult, FailOpenDecisionModel, ) +from powercontext.builtin.runtime.decision_policy import ( + DecisionCoverage, + DecisionObservation, + DecisionPolicyMode, + DecisionPrivacyBoundary, + DecisionVerdict, +) from powercontext.builtin.runtime.memory_write_gate import ( DecisionMemoryWriteGate, MemoryWriteGateRequest, MemoryWriteRejectionCode, MemoryWriteVerdict, + _assess_memory_write_decision, _candidate_subject, build_memory_write_gate, + memory_write_policy, ) # A known-answer pair used to orient the hold direction. A durable preference should pass an @@ -47,6 +57,7 @@ class _StaticDecisionModel: """A backend that always returns one prepared verdict.""" policy_id = "powercontext.decision.static.v1" + is_local_only = True def __init__(self, result: DecisionResult) -> None: self._result = result @@ -65,19 +76,36 @@ async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: return await super().evaluate(request) +class _IdentifiedDecisionModel(_StaticDecisionModel): + provider_id = "test-provider" + backend_model_id = "test-model" + + class _FailingDecisionModel: """A backend whose every evaluation raises.""" policy_id = "powercontext.decision.failing.v1" + is_local_only = True async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: raise ValueError("backend unavailable") # noqa: TRY003 +class _RecordingObservationSink: + """Collect the safe sidecar a gate offers to durable runtime storage.""" + + def __init__(self) -> None: + self.observations: list[DecisionObservation] = [] + + async def record(self, observation: DecisionObservation, /) -> None: + self.observations.append(observation) + + class _PolarityBackend: """A backend that calls one known answer insufficient and the other sufficient.""" policy_id = "powercontext.decision.polarity.v1" + is_local_only = True def __init__(self, *, insufficient_for: frozenset[str]) -> None: self._insufficient_for = frozenset(_candidate_subject((value,)) for value in insufficient_for) @@ -141,6 +169,71 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_direct_gate_binds_its_effective_hold_direction_into_the_observation_version() -> None: + async def scenario() -> None: + sink = _RecordingObservationSink() + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), hold_on=DecisionOutcome.NO) + + await gate.assess(_request(evidence=("source:task:1",), observation_sink=sink)) + + assert sink.observations[0].policy_version == "1;hold_on=no;threshold=none" + + asyncio.run(scenario()) + + +def test_direct_gate_rejects_a_disabled_or_conflicting_policy() -> None: + disabled = memory_write_policy(mode=DecisionPolicyMode.DISABLED) + with pytest.raises(ValueError, match="disabled policy"): + DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), policy=disabled) + + policy = memory_write_policy(hold_on=DecisionOutcome.NO) + with pytest.raises(ValueError, match="hold direction"): + DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES)), + hold_on=DecisionOutcome.YES, + policy=policy, + ) + + threshold_policy = memory_write_policy(threshold=0.4) + with pytest.raises(ValueError, match="operational parameters"): + DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), policy=threshold_policy) + + +def test_direct_gate_rejects_policy_fields_it_does_not_execute() -> None: + model = _StaticDecisionModel(_verdict(DecisionOutcome.YES)) + question_policy = memory_write_policy().model_copy(update={"question": "Does this unrelated question hold?"}) + with pytest.raises(ValueError, match="question"): + DecisionMemoryWriteGate(model, policy=question_policy) + + mapping_policy = memory_write_policy().model_copy( + update={ + "outcome_mapping": { + DecisionOutcome.YES.value: DecisionVerdict.DENY.value, + DecisionOutcome.NO.value: "review", + DecisionOutcome.ABSTAIN.value: DecisionVerdict.UNKNOWN.value, + } + } + ) + with pytest.raises(ValueError, match="outcome mapping"): + DecisionMemoryWriteGate(model, policy=mapping_policy) + + +def test_observation_records_backend_identity_and_evaluation_latency() -> None: + async def scenario() -> None: + sink = _RecordingObservationSink() + gate = DecisionMemoryWriteGate(_IdentifiedDecisionModel(_verdict(DecisionOutcome.NO))) + + await gate.assess(_request(evidence=("source:task:1",), observation_sink=sink)) + + observation = sink.observations[0] + assert observation.provider_id == "test-provider" + assert observation.backend_model_id == "test-model" + assert observation.assessment.latency_ms is not None + assert observation.assessment.latency_ms >= 0 + + asyncio.run(scenario()) + + def test_a_hold_carries_a_structured_code_and_reason() -> None: async def scenario() -> None: gate = DecisionMemoryWriteGate( @@ -202,6 +295,41 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_shadow_mode_passes_an_oversized_candidate_batch_without_calling_the_backend() -> None: + async def scenario() -> None: + backend = _RecordingDecisionModel(_verdict(DecisionOutcome.YES)) + gate = DecisionMemoryWriteGate( + backend, + policy=memory_write_policy(mode=DecisionPolicyMode.SHADOW), + ) + + assessment = await gate.assess( + _request(candidates=("x" * 4000, "UNASSESSED_TAIL"), evidence=("source:task:1",)) + ) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.used_fallback is False + assert backend.requests == [] + + asyncio.run(scenario()) + + +def test_advisory_mode_flags_an_insufficient_write_without_holding_it() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES, rationale="evidence is insufficient")), + policy=memory_write_policy(mode=DecisionPolicyMode.ADVISORY), + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.FLAG + assert assessment.reason == "evidence is insufficient" + assert assessment.code is None + + asyncio.run(scenario()) + + def test_a_low_confidence_hold_is_written_but_flagged() -> None: async def scenario() -> None: gate = DecisionMemoryWriteGate( @@ -262,6 +390,117 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_memory_write_decision_policy_keeps_fallback_unadjudicated() -> None: + assessment = _assess_memory_write_decision( + _verdict(DecisionOutcome.ABSTAIN, used_fallback=True), + hold_on=DecisionOutcome.YES, + ) + + assert assessment.coverage is DecisionCoverage.UNADJUDICATED + assert assessment.verdict is DecisionVerdict.UNKNOWN + assert assessment.used_fallback is True + + +def test_memory_write_decision_policy_uses_configured_hold_direction() -> None: + yes_holds = _assess_memory_write_decision(_verdict(DecisionOutcome.YES), hold_on=DecisionOutcome.YES) + no_holds = _assess_memory_write_decision(_verdict(DecisionOutcome.YES), hold_on=DecisionOutcome.NO) + + assert yes_holds.coverage is DecisionCoverage.ADJUDICATED + assert yes_holds.verdict is DecisionVerdict.DENY + assert no_holds.coverage is DecisionCoverage.ADJUDICATED + assert no_holds.verdict is DecisionVerdict.ALLOW + + +def test_gate_emits_policy_observation_without_raw_subject_or_evidence(caplog: pytest.LogCaptureFixture) -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES)), + policy=memory_write_policy(mode=DecisionPolicyMode.SHADOW), + ) + + assessment = await gate.assess(_request(candidates=("secret claim",), evidence=("secret evidence",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + + caplog.set_level(logging.INFO, logger="powercontext.builtin.runtime.decision_policy") + asyncio.run(scenario()) + + record = next(record for record in caplog.records if getattr(record, "event", None) == "decision.observation") + assert record.__dict__["policy_id"] == "memory.write.evidence_sufficiency.v1" + assert record.__dict__["mode"] == "shadow" + assert record.__dict__["privacy_boundary"] == "local_only" + assert record.__dict__["privacy_outcome"] == "local_model_called" + assert record.__dict__["coverage"] == "adjudicated" + assert "secret claim" not in str(record.__dict__) + assert "secret evidence" not in str(record.__dict__) + + +def test_gate_offers_a_bounded_observation_to_the_request_sink() -> None: + async def scenario() -> None: + sink = _RecordingObservationSink() + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES))) + + assessment = await gate.assess( + MemoryWriteGateRequest( + candidates=("secret claim",), + evidence=("secret evidence",), + expected_revision=1, + scope_id="scope-a", + operation_id="memory-write:memory-a@2", + subject_refs=("entry:entry-a@version-a",), + evidence_refs=("source:task:1",), + observation_sink=sink, + ) + ) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + assert len(sink.observations) == 1 + observation = sink.observations[0] + assert observation.scope_id == "scope-a" + assert observation.operation_id == "memory-write:memory-a@2" + assert observation.subject_refs == ("entry:entry-a@version-a",) + assert observation.evidence_refs == ("source:task:1",) + assert observation.final_action == "memory_write_hold" + assert observation.privacy_outcome == "local_model_called" + assert "secret claim" not in observation.model_dump_json() + assert "secret evidence" not in observation.model_dump_json() + + asyncio.run(scenario()) + + +def test_a_failing_observation_sink_keeps_the_gate_verdict_and_logs_only_its_type( + caplog: pytest.LogCaptureFixture, +) -> None: + class _FailingObservationSink: + async def record(self, observation: DecisionObservation, /) -> None: + raise RuntimeError("secret sink failure detail") # noqa: TRY003 + + async def scenario() -> None: + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES))) + + assessment = await gate.assess( + MemoryWriteGateRequest( + candidates=("secret claim",), + evidence=("secret evidence",), + expected_revision=1, + scope_id="scope-a", + operation_id="memory-write:memory-a@2", + observation_sink=_FailingObservationSink(), + ) + ) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + + caplog.set_level(logging.WARNING, logger="powercontext.builtin.runtime.memory_write_gate") + asyncio.run(scenario()) + + record = next( + record for record in caplog.records if getattr(record, "event", None) == "decision.observation_sink_failed" + ) + assert record.__dict__["error_type"] == "RuntimeError" + assert "secret sink failure detail" not in str(record.__dict__) + + def test_the_hold_direction_cannot_be_abstain() -> None: with pytest.raises(ValueError, match="hold direction"): DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.NO)), hold_on=DecisionOutcome.ABSTAIN) @@ -318,15 +557,113 @@ def test_the_gate_is_built_from_configuration() -> None: enabled=True, hold_on="no", threshold=0.4, + mode=DecisionPolicyMode.SHADOW, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, ) assert isinstance(gate, DecisionMemoryWriteGate) assert gate.policy_id == "powercontext.decision.static.v1" + assert gate.mode is DecisionPolicyMode.SHADOW + + +def test_new_memory_write_policies_default_to_shadow_mode() -> None: + assert memory_write_policy().mode is DecisionPolicyMode.SHADOW + + +def test_memory_write_policy_version_binds_the_hold_direction_and_threshold() -> None: + default = memory_write_policy(hold_on=DecisionOutcome.YES, threshold=None) + different_direction = memory_write_policy(hold_on=DecisionOutcome.NO, threshold=None) + different_threshold = memory_write_policy(hold_on=DecisionOutcome.YES, threshold=0.4) + + assert len({default.version, different_direction.version, different_threshold.version}) == 3 + + +def test_disabled_mode_does_not_construct_a_memory_write_gate() -> None: + gate = build_memory_write_gate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES)), + enabled=True, + mode=DecisionPolicyMode.DISABLED, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, + ) + + assert gate is None + + +def test_no_external_call_boundary_skips_the_backend_and_emits_an_unadjudicated_observation( + caplog: pytest.LogCaptureFixture, +) -> None: + async def scenario() -> None: + backend = _RecordingDecisionModel(_verdict(DecisionOutcome.YES)) + gate = build_memory_write_gate( + backend, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ) + + assert isinstance(gate, DecisionMemoryWriteGate) + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.used_fallback is True + assert backend.requests == [] + + caplog.set_level(logging.INFO, logger="powercontext.builtin.runtime.decision_policy") + asyncio.run(scenario()) + + record = next(record for record in caplog.records if getattr(record, "event", None) == "decision.observation") + assert record.__dict__["coverage"] == "unadjudicated" + assert record.__dict__["verdict"] == "unknown" + assert record.__dict__["fallback_reason"] == "privacy_boundary" + + +def test_direct_gate_construction_honors_the_no_external_call_boundary() -> None: + async def scenario() -> None: + backend = _RecordingDecisionModel(_verdict(DecisionOutcome.YES)) + gate = DecisionMemoryWriteGate( + backend, + policy=memory_write_policy(privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL), + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.used_fallback is True + assert backend.requests == [] + + asyncio.run(scenario()) + + +def test_local_only_boundary_refuses_a_backend_without_a_locality_attestation() -> None: + class _RemoteDecisionModel(_RecordingDecisionModel): + is_local_only = False + + async def scenario() -> None: + backend = _RemoteDecisionModel(_verdict(DecisionOutcome.YES)) + gate = build_memory_write_gate( + backend, + enabled=True, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY, + ) + + assert isinstance(gate, DecisionMemoryWriteGate) + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.used_fallback is True + assert backend.requests == [] + + asyncio.run(scenario()) def _request( *, candidates: tuple[str, ...] = ("Remember the contract change.",), evidence: tuple[str, ...] = (), + observation_sink: _RecordingObservationSink | None = None, ) -> MemoryWriteGateRequest: - return MemoryWriteGateRequest(candidates=candidates, evidence=evidence, expected_revision=1) + return MemoryWriteGateRequest( + candidates=candidates, + evidence=evidence, + expected_revision=1, + observation_sink=observation_sink, + ) diff --git a/tests/builtin/runtime/test_memory_write_gate_paths.py b/tests/builtin/runtime/test_memory_write_gate_paths.py index c487ce0bdd..e0efda31e6 100644 --- a/tests/builtin/runtime/test_memory_write_gate_paths.py +++ b/tests/builtin/runtime/test_memory_write_gate_paths.py @@ -12,15 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Atomic Memory refuses the legacy Memory write gate before any write path starts.""" +"""Atomic Memory receives the configured Memory write gate before any write path starts.""" from __future__ import annotations import asyncio from pathlib import Path -import pytest - from powercontext.builtin.artifacts.memory import ( MemoryWriteAssessment, MemoryWriteGateRequest, @@ -30,7 +28,6 @@ from powercontext.builtin.inference import InferenceUsage from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime import BuiltinConfig, RuntimeConfig, open_builtin_contexts, open_builtin_runtime -from powercontext.builtin.runtime.composition import BuiltinConfigurationError from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionRequest, DecisionResult _SCRIPTED_POLICY_ID = "test.memory.write-gate.v1" @@ -75,27 +72,25 @@ def _config(tmp_path: Path, runtime: RuntimeConfig | None = None, database: str ) -def test_config_rejects_the_legacy_gate_before_the_decision_backend(tmp_path: Path) -> None: +def test_runtime_allows_the_gate_before_the_decision_backend(tmp_path: Path) -> None: async def scenario() -> None: config = _config(tmp_path, RuntimeConfig(memory_write_gate_enabled=True), database="enabled.db") - with pytest.raises(BuiltinConfigurationError, match=r"Atomic Memory.*legacy Memory write gate"): - async with open_builtin_runtime(config, decision_model=_InsufficientDecisionModel()): - pytest.fail("Legacy gate configuration must fail before Runtime startup") + async with open_builtin_runtime(config, decision_model=_InsufficientDecisionModel()): + pass asyncio.run(scenario()) -def test_config_rejects_the_legacy_gate_without_a_backend(tmp_path: Path) -> None: +def test_runtime_allows_the_no_call_gate_without_a_backend(tmp_path: Path) -> None: async def scenario() -> None: config = _config(tmp_path, RuntimeConfig(memory_write_gate_enabled=True), database="unavailable.db") - with pytest.raises(BuiltinConfigurationError, match=r"Atomic Memory.*legacy Memory write gate"): - async with open_builtin_contexts(config): - pytest.fail("Legacy gate configuration must fail before Contexts startup") + async with open_builtin_runtime(config): + pass asyncio.run(scenario()) -def test_runtime_rejects_an_injected_legacy_gate_before_an_explicit_write(tmp_path: Path) -> None: +def test_runtime_accepts_an_injected_gate_before_an_explicit_write(tmp_path: Path) -> None: async def scenario() -> None: gate = _ScriptedGate( _assessment( @@ -104,15 +99,14 @@ async def scenario() -> None: reason="the citation is thin", ) ) - with pytest.raises(BuiltinConfigurationError, match=r"Atomic Memory.*legacy Memory write gate"): - async with open_builtin_runtime(_config(tmp_path), memory_write_gate=gate): - pytest.fail("Injected legacy gates must fail before Runtime startup") + async with open_builtin_runtime(_config(tmp_path), memory_write_gate=gate): + pass assert gate.requests == [] asyncio.run(scenario()) -def test_contexts_reject_an_injected_legacy_gate_before_ingestion(tmp_path: Path) -> None: +def test_contexts_accept_an_injected_gate_before_ingestion(tmp_path: Path) -> None: async def scenario() -> None: gate = _ScriptedGate( _assessment( @@ -121,9 +115,8 @@ async def scenario() -> None: reason="the window evidence is thin", ) ) - with pytest.raises(BuiltinConfigurationError, match=r"Atomic Memory.*legacy Memory write gate"): - async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate): - pytest.fail("Injected legacy gates must fail before Contexts startup") + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate): + pass assert gate.requests == [] asyncio.run(scenario()) diff --git a/tests/builtin/runtime/test_processing_composition.py b/tests/builtin/runtime/test_processing_composition.py index 3dc90a4a6e..6feeb5ed0a 100644 --- a/tests/builtin/runtime/test_processing_composition.py +++ b/tests/builtin/runtime/test_processing_composition.py @@ -246,18 +246,10 @@ async def scenario() -> None: asyncio.run(scenario()) -def test_injected_legacy_memory_write_gate_is_rejected_before_worker_composition(tmp_path: Path) -> None: - async def scenario() -> None: - config = BuiltinConfig( - database=_sqlite(tmp_path / "memory-gate-injection.db"), - inference=InferenceConfig(generation_model="test"), - runtime=RuntimeConfig(artifact_processing_families=("memory",)), - ) - with pytest.raises(BuiltinConfigurationError, match="legacy Memory write gate"): - async with open_builtin_runtime(config, memory_write_gate=_InjectedGate()): - pytest.fail("Runtime must reject the legacy Memory write gate before worker composition") +def test_injected_memory_write_gate_is_preserved_for_atomic_memory_composition() -> None: + gate = _InjectedGate() - asyncio.run(scenario()) + assert composition._configured_memory_write_gate(gate, None, RuntimeConfig()) is gate @pytest.mark.parametrize("role", ["all", "api"]) diff --git a/tests/test_server_logging.py b/tests/test_server_logging.py index 339ce36dc8..80c7523140 100644 --- a/tests/test_server_logging.py +++ b/tests/test_server_logging.py @@ -121,6 +121,17 @@ def test_json_formatter_emits_stable_operational_fields() -> None: "code": "insufficient_coverage", "held_count": 1, "hold_codes": ("insufficient_coverage",), + "consumer": "memory_write_gate", + "policy_id": "memory.write.evidence_sufficiency.v1", + "policy_version": "1", + "privacy_boundary": "local_only", + "privacy_outcome": "local_model_called", + "coverage": "adjudicated", + "verdict": "deny", + "final_action": "memory_write_hold", + "fallback_reason": "decision_model_fallback", + "candidate_count": 1, + "evidence_count": 2, "ignored": "not serialized", }) OperationalContextFilter().filter(record) @@ -136,6 +147,16 @@ def test_json_formatter_emits_stable_operational_fields() -> None: assert payload["code"] == "insufficient_coverage" assert payload["held_count"] == 1 assert payload["hold_codes"] == ["insufficient_coverage"] + assert payload["consumer"] == "memory_write_gate" + assert payload["policy_id"] == "memory.write.evidence_sufficiency.v1" + assert payload["policy_version"] == "1" + assert payload["privacy_boundary"] == "local_only" + assert payload["privacy_outcome"] == "local_model_called" + assert payload["coverage"] == "adjudicated" + assert payload["verdict"] == "deny" + assert payload["final_action"] == "memory_write_hold" + assert payload["fallback_reason"] == "decision_model_fallback" + assert (payload["candidate_count"], payload["evidence_count"]) == (1, 2) assert "ignored" not in payload