diff --git a/.env.example b/.env.example index e7fcda91e2..85ff0e5538 100644 --- a/.env.example +++ b/.env.example @@ -105,6 +105,14 @@ POWERCONTEXT_SERVER_RUNTIME_DREAM_MAX_PENDING_PER_SCOPE=32 # is absent and makes no model call. # POWERCONTEXT_SERVER_RUNTIME_DECISION_ASSISTANCE_ENABLED=true +# 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 +# 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. +# POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_THRESHOLD=0.5 + # Recall-sufficiency gate. Disabled by default; enabling expands thin recall up to two rounds. # POWERCONTEXT_SERVER_RUNTIME_RECALL_GATE_ENABLED=true # POWERCONTEXT_SERVER_RUNTIME_RECALL_GATE_MAX_ROUNDS=2 diff --git a/openapi/powercontext.yaml b/openapi/powercontext.yaml index 2388150194..3801c74951 100644 --- a/openapi/powercontext.yaml +++ b/openapi/powercontext.yaml @@ -8194,6 +8194,17 @@ components: memory: $ref: "#/components/schemas/ArtifactReference" nullable: true + held_count: + type: integer + minimum: 0 + default: 0 + description: Number of source windows held by the Memory write gate. + hold_codes: + type: array + items: + type: string + default: [] + description: Structured Memory write gate refusal codes for held windows. FlushTopicMemoryRequest: type: object additionalProperties: false diff --git a/src/powercontext/builtin/artifacts/memory/__init__.py b/src/powercontext/builtin/artifacts/memory/__init__.py index a3c28e2981..7b4142f825 100644 --- a/src/powercontext/builtin/artifacts/memory/__init__.py +++ b/src/powercontext/builtin/artifacts/memory/__init__.py @@ -25,6 +25,7 @@ MemoryEntryInactiveError, MemoryEntryNotFoundError, MemoryLayerError, + MemoryWriteRejectedError, ) from powercontext.builtin.artifacts.memory.extraction import ( DefaultMemoryEvidenceProjector, @@ -84,7 +85,12 @@ MemorySearchChannels, MemorySearchRequest, MemoryUnitOfWork, + MemoryWriteAssessment, + MemoryWriteGate, + MemoryWriteGateRequest, MemoryWritePlan, + MemoryWriteRejectionCode, + MemoryWriteVerdict, ) from powercontext.builtin.artifacts.memory.reranking import ( MEMORY_RERANK_INSTRUCTIONS, @@ -170,7 +176,13 @@ "MemoryService", "MemoryUnitOfWork", "MemoryUsedSearchMode", + "MemoryWriteAssessment", + "MemoryWriteGate", + "MemoryWriteGateRequest", "MemoryWritePlan", + "MemoryWriteRejectedError", + "MemoryWriteRejectionCode", + "MemoryWriteVerdict", "memory_extraction_instructions", "memory_extraction_instructions_version", ] diff --git a/src/powercontext/builtin/artifacts/memory/errors.py b/src/powercontext/builtin/artifacts/memory/errors.py index de507eb781..fefdcf4fd8 100644 --- a/src/powercontext/builtin/artifacts/memory/errors.py +++ b/src/powercontext/builtin/artifacts/memory/errors.py @@ -110,5 +110,19 @@ def __init__(self, code: str) -> None: super().__init__(messages.get(code, f"invalid memory citation: {code}")) +class MemoryWriteRejectedError(MemoryLayerError, RuntimeError): + """A structured, caller-visible refusal to apply one Memory write. + + ``code`` and ``reason`` carry the gate's decision to the host, so a refused write is + observable rather than silently dropped. + """ + + def __init__(self, code: str, reason: str | None = None) -> None: + self.code = code + self.reason = reason + detail = "" if reason is None else f": {reason}" + super().__init__(f"memory write was rejected ({code}){detail}") + + class MemoryBackendConfigurationError(MemoryLayerError, RuntimeError): """Raised when a repository cannot satisfy its declared configuration.""" diff --git a/src/powercontext/builtin/artifacts/memory/protocols.py b/src/powercontext/builtin/artifacts/memory/protocols.py index 7efda1bcc5..933425d961 100644 --- a/src/powercontext/builtin/artifacts/memory/protocols.py +++ b/src/powercontext/builtin/artifacts/memory/protocols.py @@ -17,9 +17,11 @@ from __future__ import annotations from contextlib import AbstractAsyncContextManager +from dataclasses import dataclass +from enum import StrEnum from typing import Protocol -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict from powercontext.artifacts import Artifact, ArtifactRef from powercontext.builtin.artifacts.memory.models import ( @@ -65,11 +67,72 @@ class MemoryCommit(BaseModel): projections: tuple[MemoryProjection, ...] +class MemoryWriteRejectionCode(StrEnum): + """Structured, caller-visible vocabulary for a held Memory write. + + Names mirror the evidence-selection vocabulary so a host can branch on one stable set of + codes instead of parsing prose. + """ + + NEEDS_EVIDENCE = "needs_evidence" + EVIDENCE_LIMIT_EXCEEDED = "evidence_limit_exceeded" + INSUFFICIENT_COVERAGE = "insufficient_coverage" + + +class MemoryWriteVerdict(StrEnum): + """The complete verdict vocabulary a Memory write gate may produce.""" + + ACCEPT = "accept" + FLAG = "flag" + HOLD = "hold" + + +class MemoryWriteAssessment(BaseModel): + """One gate verdict with the structured refusal a caller can observe. + + ``HOLD`` always carries both a ``code`` and a ``reason``: a refused write is visible to + its caller, never silently dropped. ``ACCEPT``/``FLAG`` leave ``code`` unset. + """ + + model_config = ConfigDict(frozen=True) + + verdict: MemoryWriteVerdict + policy_id: str + code: MemoryWriteRejectionCode | None = None + reason: str | None = None + used_fallback: bool = False + + +@dataclass(frozen=True, slots=True) +class MemoryWriteGateRequest: + """A bounded projection of one pending Memory write for sufficiency judgement.""" + + candidates: tuple[str, ...] + evidence: tuple[str, ...] + expected_revision: int | None = None + + +class MemoryWriteGate(Protocol): + """Judge whether a pending Memory write is supported by its cited evidence. + + A gate only classifies: it never writes, approves, rejects, or deletes anything. A missing + or failing gate must be treated by callers as a pass-through, never as a hold. + """ + + policy_id: str + + async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessment: + """Return one verdict for a candidate set and its bounded evidence projection.""" + + ... + + class MemoryWritePlan(BaseModel): """A side-effect-free result that can be committed in an outer transaction.""" result: Memory | None commit: MemoryCommit | None + decision: MemoryWriteAssessment | None = None class MemorySearchRequest(BaseModel): diff --git a/src/powercontext/builtin/artifacts/memory/service.py b/src/powercontext/builtin/artifacts/memory/service.py index c168f98c7e..e0727ceda2 100644 --- a/src/powercontext/builtin/artifacts/memory/service.py +++ b/src/powercontext/builtin/artifacts/memory/service.py @@ -16,6 +16,7 @@ from __future__ import annotations +import logging from collections.abc import Callable, Sequence from contextlib import nullcontext from dataclasses import dataclass @@ -24,6 +25,9 @@ from typing import Literal, Protocol, TypeAlias, TypeVar, overload from uuid import uuid4 +from pydantic import BaseModel + +from powercontext._logging import log_safely from powercontext.artifacts import Artifact, ArtifactLineage, ArtifactRef from powercontext.builtin.artifacts.memory.canonical import ( canonical_embedding, @@ -48,6 +52,7 @@ MemoryCapacityExceededError, MemoryEntryInactiveError, MemoryEntryNotFoundError, + MemoryWriteRejectedError, ) from powercontext.builtin.artifacts.memory.fusion import ( admit_fts_candidates, @@ -85,7 +90,12 @@ MemoryCommit, MemoryProjection, MemorySearchRequest, + MemoryWriteAssessment, + MemoryWriteGate, + MemoryWriteGateRequest, MemoryWritePlan, + MemoryWriteRejectionCode, + MemoryWriteVerdict, ) from powercontext.builtin.artifacts.memory.reranking import MemoryReranker from powercontext.builtin.artifacts.prompt.service import ScopedPrompts, current_prompt, prompt_operation @@ -99,22 +109,29 @@ ) from powercontext.builtin.tags import TagFilter from powercontext.errors import RevisionConflictError -from powercontext.sources import Source, SourceRef +from powercontext.sources import TEXT_EVIDENCE_PROJECTION_KEY, Source, SourceObservation, SourceRef, TextEvidence MemoryRememberMode: TypeAlias = Literal["append", "extract", "auto"] IdFactory: TypeAlias = Callable[[str], str] ValueT = TypeVar("ValueT") +_GATE_EVIDENCE_ITEM_LIMIT = 32 +_GATE_EVIDENCE_TEXT_LIMIT = 2000 +logger = logging.getLogger(__name__) class _SourceResolver(Protocol): async def get(self, source: Source, /) -> Source: ... + async def get_ref(self, ref: SourceRef, /) -> Source: ... + def as_ref(self, source: Source, /) -> SourceRef: ... class _ArtifactResolver(Protocol): async def get(self, artifact: Artifact[object], /) -> Artifact[object]: ... + async def get_ref(self, ref: ArtifactRef, /) -> Artifact[object]: ... + @dataclass(frozen=True, slots=True) class _OperationEvidence: @@ -132,6 +149,51 @@ class _EntryMaterial: content_hash: str +@dataclass(frozen=True, slots=True) +class _GateEvidenceEntry: + text: str + complete: bool + + +@dataclass(frozen=True, slots=True) +class _GateCandidateEvidence: + index: int + material: _EntryMaterial + + +@dataclass(frozen=True, slots=True) +class _GateEvidenceProjection: + entries: tuple[str, ...] + rejection: MemoryWriteAssessment | None = None + + +@dataclass(slots=True) +class _GateEvidenceBuilder: + policy_id: str + entries: list[str] + + def append(self, entry: _GateEvidenceEntry) -> MemoryWriteAssessment | None: + if not entry.complete: + return self._budget_rejection() + if any(current == entry.text for current in self.entries): + return None + if len(self.entries) >= _GATE_EVIDENCE_ITEM_LIMIT: + return self._budget_rejection() + self.entries.append(entry.text) + return None + + def projection(self) -> _GateEvidenceProjection: + return _GateEvidenceProjection(tuple(self.entries)) + + def _budget_rejection(self) -> MemoryWriteAssessment: + return MemoryWriteAssessment( + verdict=MemoryWriteVerdict.HOLD, + policy_id=self.policy_id, + code=MemoryWriteRejectionCode.EVIDENCE_LIMIT_EXCEEDED, + reason="the cited evidence exceeds the gate evidence budget", + ) + + class _InvalidMemoryOperationError(ValueError): def __init__(self, code: str) -> None: messages = { @@ -164,6 +226,15 @@ def _require_tag_filter(capabilities: MemoryCapabilities, tag_filter: TagFilter raise CapabilityNotSupportedError("tag-filter") +def _annotate_reason(reason: str | None, flagged_reason: str | None) -> str | None: + """Fill a missing audit reason from a flagged gate verdict without overwriting a caller's.""" + + normalized = normalize_reason(reason) + if normalized is not None or flagged_reason is None: + return normalized + return normalize_reason(flagged_reason) + + class MemoryService: """Validate and orchestrate Memory operations without exposing storage details.""" @@ -179,6 +250,7 @@ def __init__( artifact_resolver: _ArtifactResolver | None = None, id_factory: IdFactory | None = None, prompt_context: ScopedPrompts | None = None, + write_gate: MemoryWriteGate | None = None, capacity_budget: MemoryCapacityBudget | None = None, compaction: MemoryCompactionPolicy | None = None, max_history_revisions: int = 100, @@ -186,6 +258,7 @@ def __init__( self._backend = backend self._prompt_context = prompt_context self._candidate_pipeline = candidate_pipeline + self._write_gate = write_gate self._embedding_model = embedding_model if rerank_candidate_limit < 1: raise _InvalidMemoryOperationError("search-limit") @@ -400,15 +473,15 @@ async def remember( ) -> Memory | None: """Append or extract validated entry changes against one exact head.""" - return await self.apply( - await self.plan_remember( - memory=memory, - sources=sources, - artifacts=artifacts, - entries=entries, - mode=mode, - ) + plan = await self.plan_remember( + memory=memory, + sources=sources, + artifacts=artifacts, + entries=entries, + mode=mode, ) + _raise_if_write_held(plan) + return await self.apply(plan) async def plan_remember( self, @@ -455,15 +528,25 @@ async def plan_remember( if not candidates: return MemoryWritePlan(result=base, commit=None) + assessment = await self._assess_write(base, candidates, evidence, current_entries) + if assessment is not None and assessment.verdict is MemoryWriteVerdict.HOLD: + # A refused write stays visible: the caller reads the structured code and reason + # from the plan. The plan carries no commit, so nothing is written. + return MemoryWritePlan(result=base, commit=None, decision=assessment) + + flagged_reason = ( + assessment.reason if assessment is not None and assessment.verdict is MemoryWriteVerdict.FLAG else None + ) commit = await self._prepare_commit( base=base, candidates=candidates, evidence=evidence, current_entries=current_entries, + flagged_reason=flagged_reason, ) if commit is None: - return MemoryWritePlan(result=base, commit=None) - return MemoryWritePlan(result=commit.memory, commit=commit) + return MemoryWritePlan(result=base, commit=None, decision=assessment) + return MemoryWritePlan(result=commit.memory, commit=commit, decision=assessment) async def apply(self, plan: MemoryWritePlan, /) -> Memory | None: """Apply one prepared write through this service's transaction boundary.""" @@ -1271,6 +1354,122 @@ async def _candidates( ) ) + async def _assess_write( + self, + base: Memory | None, + candidates: tuple[MemoryEntryInput, ...], + evidence: _OperationEvidence, + current_entries: tuple[MemoryEntryVersion, ...] | None, + ) -> MemoryWriteAssessment | None: + """Ask the configured gate about one candidate set; ``None`` means no gate is active.""" + + if self._write_gate is None: + return None + projection = await self._gate_evidence(base, candidates, evidence, current_entries) + if projection.rejection is not None: + _log_gate_assessment(projection.rejection) + return projection.rejection + try: + return await self._write_gate.assess( + MemoryWriteGateRequest( + candidates=tuple(candidate.text for candidate in candidates), + evidence=projection.entries, + expected_revision=None if base is None else base.revision, + ) + ) + except Exception: + return MemoryWriteAssessment( + verdict=MemoryWriteVerdict.ACCEPT, + policy_id=self._write_gate.policy_id, + used_fallback=True, + ) + + async def _gate_evidence( + self, + base: Memory | None, + candidates: tuple[MemoryEntryInput, ...], + evidence: _OperationEvidence, + current_entries: tuple[MemoryEntryVersion, ...] | None, + ) -> _GateEvidenceProjection: + builder = _GateEvidenceBuilder(self._write_gate_policy_id(), []) + for candidate in await self._gate_candidate_evidence(base, candidates, evidence, current_entries): + for entry in await self._candidate_gate_evidence(candidate): + if rejection := builder.append(entry): + return _GateEvidenceProjection(tuple(builder.entries), rejection) + return builder.projection() + + async def _gate_candidate_evidence( + self, + base: Memory | None, + candidates: tuple[MemoryEntryInput, ...], + evidence: _OperationEvidence, + current_entries: tuple[MemoryEntryVersion, ...] | None, + ) -> tuple[_GateCandidateEvidence, ...]: + current_by_entry = {} if current_entries is None else {entry.entry_id: entry for entry in current_entries} + targeted: set[str] = set() + resolved: list[_GateCandidateEvidence] = [] + for index, candidate in enumerate(candidates, start=1): + previous = None + if candidate.entry is not None: + _, previous = await self._claim_revision_target(candidate, base, current_by_entry, targeted) + material = await self._material_from_candidate( + candidate, + evidence.sources, + evidence.artifacts, + previous=previous, + ) + resolved.append(_GateCandidateEvidence(index=index, material=material)) + return tuple(resolved) + + async def _candidate_gate_evidence(self, candidate: _GateCandidateEvidence) -> tuple[_GateEvidenceEntry, ...]: + entries = [ + await self._source_ref_gate_evidence(candidate.index, source) for source in candidate.material.sources + ] + entries.extend([ + await self._artifact_ref_gate_evidence(candidate.index, artifact) + for artifact in candidate.material.artifacts + ]) + return tuple(entries) + + def _source_gate_evidence(self, candidate_index: int, source: Source) -> _GateEvidenceEntry: + ref = self._source_refs((source,))[0] + identity = _candidate_gate_identity(candidate_index, f"source:{ref.source_type}:{ref.source_id}") + content = _source_gate_content(source) + if isinstance(content, str) and content.strip(): + return _bounded_gate_evidence(identity, content) + return _incomplete_gate_evidence(identity) + + @staticmethod + def _artifact_gate_evidence(candidate_index: int, artifact: Artifact[object]) -> _GateEvidenceEntry: + ref = artifact.as_ref() + return _bounded_gate_evidence( + _candidate_gate_identity(candidate_index, f"artifact:{ref.family}:{ref.artifact_id}@{ref.revision}"), + _artifact_gate_content(artifact), + ) + + async def _source_ref_gate_evidence(self, candidate_index: int, ref: SourceRef) -> _GateEvidenceEntry: + identity = _candidate_gate_identity(candidate_index, f"source:{ref.source_type}:{ref.source_id}") + if self._source_resolver is None: + return _incomplete_gate_evidence(identity) + try: + return self._source_gate_evidence(candidate_index, await self._source_resolver.get_ref(ref)) + except Exception: + return _incomplete_gate_evidence(identity) + + async def _artifact_ref_gate_evidence(self, candidate_index: int, ref: ArtifactRef) -> _GateEvidenceEntry: + identity = _candidate_gate_identity(candidate_index, f"artifact:{ref.family}:{ref.artifact_id}@{ref.revision}") + if self._artifact_resolver is None: + return _incomplete_gate_evidence(identity) + try: + return self._artifact_gate_evidence(candidate_index, await self._artifact_resolver.get_ref(ref)) + except Exception: + return _incomplete_gate_evidence(identity) + + def _write_gate_policy_id(self) -> str: + if self._write_gate is not None: + return self._write_gate.policy_id + return "memory.write-gate" + async def _prepare_commit( self, *, @@ -1278,6 +1477,7 @@ async def _prepare_commit( candidates: tuple[MemoryEntryInput, ...], evidence: _OperationEvidence, current_entries: tuple[MemoryEntryVersion, ...] | None, + flagged_reason: str | None = None, ) -> MemoryCommit | None: memory_id = base.artifact_id if base is not None else self._new_id("memory") next_revision = 1 if base is None else base.revision + 1 @@ -1316,7 +1516,7 @@ async def _prepare_commit( entry_id=entry_id, from_entry_version_id=None, to_entry_version_id=version.entry_version_id, - reason=normalize_reason(candidate.reason), + reason=_annotate_reason(candidate.reason, flagged_reason), ) ) continue @@ -1355,7 +1555,7 @@ async def _prepare_commit( entry_id=entry_id, from_entry_version_id=previous.entry_version_id, to_entry_version_id=version.entry_version_id, - reason=normalize_reason(candidate.reason), + reason=_annotate_reason(candidate.reason, flagged_reason), ) ) @@ -1481,7 +1681,9 @@ async def _canonical_candidate_artifacts( result: list[ArtifactRef] = [] allowed_refs = tuple(artifact.as_ref() for artifact in allowed) for value in values: - canonical = value if self._artifact_resolver is None else await self._artifact_resolver.get(value) + canonical = _matching_allowed_artifact(value, allowed) + if canonical is None: + canonical = value if self._artifact_resolver is None else await self._artifact_resolver.get(value) reference = canonical.as_ref() if reference not in (*allowed_refs, *previous): raise InvalidMemoryEvidenceError("artifact-outside") @@ -1604,6 +1806,91 @@ def _manifest_entry(version: MemoryEntryVersion, *, state: Literal["active", "in ) +def _raise_if_write_held(plan: MemoryWritePlan) -> None: + decision = plan.decision + if decision is None or decision.verdict is not MemoryWriteVerdict.HOLD: + return + code = "unspecified" if decision.code is None else decision.code.value + raise MemoryWriteRejectedError(code, decision.reason) + + +def _bounded_gate_evidence(identity: str, content: str) -> _GateEvidenceEntry: + normalized = normalize_text(content) + return _GateEvidenceEntry( + text=f"{identity}\n{normalized[:_GATE_EVIDENCE_TEXT_LIMIT]}", + complete=len(normalized) <= _GATE_EVIDENCE_TEXT_LIMIT, + ) + + +def _incomplete_gate_evidence(identity: str) -> _GateEvidenceEntry: + return _GateEvidenceEntry(text=identity, complete=False) + + +def _candidate_gate_identity(candidate_index: int, identity: str) -> str: + return f"candidate:{candidate_index} {identity}" + + +def _source_gate_content(source: Source) -> str | None: + content = getattr(source, "content", None) + if isinstance(content, str): + return content + if isinstance(source, SourceObservation): + try: + evidence = TextEvidence.model_validate(source.projection(TEXT_EVIDENCE_PROJECTION_KEY)) + except Exception: + return None + return evidence.content + return None + + +def _log_gate_assessment(assessment: MemoryWriteAssessment) -> None: + event = { + MemoryWriteVerdict.HOLD: "memory.write-gate.hold", + MemoryWriteVerdict.FLAG: "memory.write-gate.flag", + }.get(assessment.verdict, "memory.write-gate.assess") + log_safely( + logger, + logging.INFO, + "Memory write gate assessed a pending write", + extra={ + "event": event, + "decision_kind": "memory.write-gate", + "policy_id": assessment.policy_id, + "verdict": assessment.verdict.value, + "code": None if assessment.code is None else assessment.code.value, + "used_fallback": assessment.used_fallback, + }, + ) + + +def _artifact_gate_content(artifact: Artifact[object]) -> str: + content = artifact.content + if isinstance(content, BaseModel): + return content.model_dump_json() + return str(content) + + +def _matching_allowed_artifact( + value: Artifact[object], + allowed: Sequence[Artifact[object]], +) -> Artifact[object] | None: + reference = value.as_ref() + for artifact in allowed: + if artifact.as_ref() == reference: + return artifact + # Pydantic validates MemoryEntryInput artifacts through the generic Artifact[object] + # annotation, which strips the concrete subclass family. Fall back to the operation's + # canonical evidence set when the revision identity and body match exactly. + matches = [ + artifact + for artifact in allowed + if artifact.artifact_id == value.artifact_id + and artifact.revision == value.revision + and artifact.content == value.content + ] + return matches[0] if len(matches) == 1 else None + + def _canonical_source_refs(values: Sequence[SourceRef]) -> tuple[SourceRef, ...]: keyed = {canonical_json(value.model_dump(mode="json")): value for value in values} return tuple(keyed[key] for key in sorted(keyed)) diff --git a/src/powercontext/builtin/runtime/__init__.py b/src/powercontext/builtin/runtime/__init__.py index 5953e26801..927352e734 100644 --- a/src/powercontext/builtin/runtime/__init__.py +++ b/src/powercontext/builtin/runtime/__init__.py @@ -99,6 +99,7 @@ RuntimeConfig, ) from powercontext.builtin.runtime.decision_model import ( + DecisionKind, DecisionModel, DecisionModelOption, DecisionModelRequest, @@ -109,6 +110,15 @@ StructuredDecisionModel, ) from powercontext.builtin.runtime.errors import InvalidRuntimeRequestError, TopicMemoryProcessingUnavailableError +from powercontext.builtin.runtime.memory_write_gate import ( + DecisionMemoryWriteGate, + MemoryWriteAssessment, + MemoryWriteGate, + MemoryWriteGateRequest, + MemoryWriteRejectionCode, + MemoryWriteVerdict, + build_memory_write_gate, +) from powercontext.builtin.runtime.models import ( ApproveArtifactCandidateRequest, CaptureSource, @@ -223,6 +233,8 @@ "ContextAssemblySection", "CreateDreamRunRequest", "DatabaseConfig", + "DecisionKind", + "DecisionMemoryWriteGate", "DecisionModel", "DecisionModelOption", "DecisionModelRequest", @@ -291,6 +303,11 @@ "MemoryMutationResult", "MemoryRevisionChanges", "MemorySearchPage", + "MemoryWriteAssessment", + "MemoryWriteGate", + "MemoryWriteGateRequest", + "MemoryWriteRejectionCode", + "MemoryWriteVerdict", "ModelUsageDay", "ModelUsageOperation", "ModelUsagePurpose", @@ -361,6 +378,7 @@ "TopicMemoryProcessingUnavailableError", "UsageStatistics", "WorkApplication", + "build_memory_write_gate", "dependency_readiness_probe", "open_builtin_contexts", "open_builtin_runtime", diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index 02fc9c7262..db60812d3a 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -64,11 +64,14 @@ MemoryHit, MemoryQueryEmbedding, MemoryService, + MemoryWritePlan, + MemoryWriteVerdict, ) from powercontext.builtin.artifacts.memory.errors import ( CapabilityNotSupportedError, InvalidMemoryCitationError, MemoryEntryNotFoundError, + MemoryWriteRejectedError, ) from powercontext.builtin.artifacts.profile.service import RelationalProfileService from powercontext.builtin.artifacts.prompt import ( @@ -2394,7 +2397,9 @@ async def remember(self, request: RememberMemoryRequest, /) -> MemoryMutationRes service = context.artifacts.memory current = await _head_or_none(service, context.artifacts.memory_artifact_id) _validate_expected_revision(current, request.expected_revision) - updated = await service.remember(memory=current, entries=request.entries, mode="append") + plan = await service.plan_remember(memory=current, entries=request.entries, mode="append") + _raise_if_held(plan) + updated = await service.apply(plan) if updated is None: raise _RuntimeStateError("empty-write") return MemoryMutationResult( @@ -2526,7 +2531,7 @@ async def revise(self, request: ReviseMemoryEntryRequest, /) -> MemoryMutationRe context.artifacts.memory_artifact_id, request.citation, ) - updated = await service.remember( + plan = await service.plan_remember( memory=current, entries=( MemoryEntryInput( @@ -2538,6 +2543,8 @@ async def revise(self, request: ReviseMemoryEntryRequest, /) -> MemoryMutationRe ), mode="append", ) + _raise_if_held(plan) + updated = await service.apply(plan) if updated is None: raise _RuntimeStateError("empty-write") revised = next(item for item in await service.entries(updated) if item.entry_id == entry.entry_id) @@ -2854,16 +2861,22 @@ async def run(self) -> None: if span is not None: span.set_outcome("failure") else: - outcome = "success" if result.processed else "noop" + outcome = "hold" if result.held_count else "success" if result.processed else "noop" _log_scheduled_processing( outcome, operation="process_source_window", started_at=started_at, source_count=result.source_count, + held_count=result.held_count, + hold_codes=result.hold_codes, ) if span is not None: span.set_outcome(outcome) - span.set_attributes({"powercontext.background.source_count": result.source_count}) + span.set_attributes({ + "powercontext.background.source_count": result.source_count, + "powercontext.background.memory_held_count": result.held_count, + "powercontext.background.memory_hold_codes": ",".join(result.hold_codes), + }) class ScheduledExperienceProcessor: @@ -2933,6 +2946,8 @@ def _log_scheduled_processing( error: Exception | None = None, source_count: int | None = None, candidate_count: int | None = None, + held_count: int | None = None, + hold_codes: tuple[str, ...] = (), ) -> None: extra = { "event": "background.operation.completed", @@ -2945,6 +2960,10 @@ def _log_scheduled_processing( extra["source_count"] = source_count if candidate_count is not None: extra["candidate_count"] = candidate_count + if held_count is not None: + extra["held_count"] = held_count + if hold_codes: + extra["hold_codes"] = hold_codes level = logging.ERROR if error is not None else logging.INFO log_safely( logger, @@ -3508,6 +3527,16 @@ def _is_stale_memory_search(error: CapabilityNotSupportedError | InvalidMemoryCi ) +def _raise_if_held(plan: MemoryWritePlan) -> None: + """Surface a gate refusal as a structured error so the caller can read code and reason.""" + + decision = plan.decision + if decision is None or decision.verdict is not MemoryWriteVerdict.HOLD: + return + code = "unspecified" if decision.code is None else decision.code.value + raise MemoryWriteRejectedError(code, decision.reason) + + def _validate_expected_revision(memory: Memory | None, expected_revision: int | None) -> None: if expected_revision is None: return diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 835b825470..831f5f69ae 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -44,6 +44,7 @@ MemoryHit, MemoryRerankDecision, MemoryReranker, + MemoryWriteGate, ) from powercontext.builtin.artifacts.profile.generation import PROFILE_INSTRUCTIONS, LLMProfileGenerator from powercontext.builtin.artifacts.profile.service import ( @@ -137,6 +138,7 @@ LLMDecisionModel, ) from powercontext.builtin.runtime.family_processing import FAMILY_BINDINGS, FamilyWorkerSpec, run_family_worker +from powercontext.builtin.runtime.memory_write_gate import build_memory_write_gate from powercontext.builtin.runtime.models import MemorySearchMode, RuntimeCapabilities from powercontext.builtin.runtime.processing_discovery import SourceProcessingPendingProvider, enabled_profile_scopes from powercontext.builtin.runtime.processing_registry import ( @@ -319,6 +321,38 @@ def _require_decision_backend(runtime: RuntimeConfig, configured: DecisionModel raise BuiltinConfigurationError("decision-model") +def _configured_memory_write_gate( + injected: MemoryWriteGate | None, + decision_model: DecisionModel | None, + runtime: RuntimeConfig, +) -> MemoryWriteGate | None: + """Resolve the Memory write gate: an explicit injection wins, then configuration builds one. + + The gate is auxiliary and fail-open by contract, which is the opposite of the decision role: + an enabled gate whose decision backend is unavailable logs a warning and passes writes through + instead of failing startup, so a misconfigured gate can never block Memory writes. + """ + + if injected is not None: + return injected + if not runtime.memory_write_gate_enabled: + return None + gate = build_memory_write_gate( + decision_model, + enabled=True, + hold_on=runtime.memory_write_gate_hold_on, + threshold=runtime.memory_write_gate_threshold, + ) + if gate is None: + log_safely( + logger, + logging.WARNING, + "Memory write gate is enabled but no decision backend is available; writes pass through", + extra={"event": "memory.write-gate.unavailable", "decision_kind": "memory.write-gate"}, + ) + return gate + + @asynccontextmanager async def open_builtin_runtime( config: BuiltinConfig, @@ -339,6 +373,7 @@ async def open_builtin_runtime( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, decision_model: DecisionModel | None = None, + memory_write_gate: MemoryWriteGate | None = None, instrumentation: InstrumentationSettings | None = None, scope_cache_observer: ScopeCacheObserver | None = None, topic_memory_search_observer: Callable[[str, bool], None] | None = None, @@ -388,7 +423,10 @@ async def open_builtin_runtime( or skill_generator is None or handoff_pipeline is None or (config.runtime.memory_rerank_enabled and memory_reranker is None) - or (config.runtime.decision_assistance_enabled and decision_model is None) + or ( + (config.runtime.decision_assistance_enabled or config.runtime.memory_write_gate_enabled) + and decision_model is None + ) ) else (None, None, None, None, None, None, None, None, None, None, None) ) @@ -423,6 +461,7 @@ async def open_builtin_runtime( tracing, timeout_seconds=config.inference.decision_timeout_seconds or config.inference.generation_timeout_seconds, ) + configured_gate = _configured_memory_write_gate(memory_write_gate, configured_decision, config.runtime) if embedding_model is None: configured_embedding_source, readiness_embedding = await _embedding_models( config.inference, @@ -457,6 +496,7 @@ async def open_builtin_runtime( token_estimator=token_estimator, memory_reranker=configured_reranker, decision_model=configured_decision, + memory_write_gate=configured_gate, source_registry=configured_source_registry, cursor_secret=cursor_secret, tracing=tracing, @@ -818,6 +858,7 @@ async def open_builtin_contexts( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, decision_model: DecisionModel | None = None, + memory_write_gate: MemoryWriteGate | None = None, source_registry: SourceDefinitionRegistry | None = None, cursor_secret: bytes | None = None, tracing: RuntimeTracing | None = None, @@ -877,6 +918,7 @@ async def open_builtin_contexts( token_estimator=configured_token_estimator, memory_reranker=memory_reranker, decision_model=decision_model, + memory_write_gate=memory_write_gate, memory_rerank_candidate_limit=config.runtime.memory_rerank_candidate_limit, memory_capacity_budget=MemoryCapacityBudget( max_active_entries=config.runtime.memory_max_active_entries, @@ -945,6 +987,7 @@ async def open_builtin_contexts( token_estimator=configured_token_estimator, memory_reranker=memory_reranker, decision_model=decision_model, + memory_write_gate=memory_write_gate, memory_rerank_candidate_limit=config.runtime.memory_rerank_candidate_limit, memory_capacity_budget=MemoryCapacityBudget( max_active_entries=config.runtime.memory_max_active_entries, @@ -1069,7 +1112,10 @@ async def _generation_pipelines( if ( settings.generation_model is None and (not runtime.memory_rerank_enabled or settings.rerank_model is None) - and not (runtime.decision_assistance_enabled and settings.decision_model is not None) + and not ( + (runtime.decision_assistance_enabled or runtime.memory_write_gate_enabled) + and settings.decision_model is not None + ) ): return (None, None, None, None, None, None, None, None, None, None, None) @@ -1355,7 +1401,7 @@ async def _generation_decision( ) -> tuple[DecisionModel | None, ReadinessProbe | None]: """Build the opt-in decision backend, reusing the generation model when not overridden.""" - if not runtime.decision_assistance_enabled: + if not runtime.decision_assistance_enabled and not runtime.memory_write_gate_enabled: return None, None from pydantic_ai.settings import ModelSettings, merge_model_settings diff --git a/src/powercontext/builtin/runtime/config.py b/src/powercontext/builtin/runtime/config.py index 4a8dccd7aa..983f15681e 100644 --- a/src/powercontext/builtin/runtime/config.py +++ b/src/powercontext/builtin/runtime/config.py @@ -142,6 +142,12 @@ def reject_boolean_worker_quota(cls, value: Any) -> Any: memory_rerank_enabled: bool = False memory_rerank_candidate_limit: int = Field(default=30, ge=1, le=100) decision_assistance_enabled: bool = False + memory_write_gate_enabled: bool = False + # 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) 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) diff --git a/src/powercontext/builtin/runtime/decision_model.py b/src/powercontext/builtin/runtime/decision_model.py index a6a2485489..c30bb6a0ac 100644 --- a/src/powercontext/builtin/runtime/decision_model.py +++ b/src/powercontext/builtin/runtime/decision_model.py @@ -66,6 +66,16 @@ class DecisionOutcome(StrEnum): ABSTAIN = "abstain" +class DecisionKind(StrEnum): + """Stable, low-cardinality consumer labels for one decision request. + + A single naming source keeps call sites, telemetry, and tracing from drifting into + ad-hoc strings; the direction of a positive answer is never encoded here. + """ + + MEMORY_WRITE_GATE = "memory.write-gate" + + @dataclass(frozen=True, slots=True) class DecisionRequest: """One bounded question and the material a decision backend may judge. @@ -319,6 +329,7 @@ async def evaluate(self, request: DecisionModelRequest, /) -> DecisionModelResul "DECISION_INSTRUCTIONS", "DECISION_INSTRUCTIONS_VERSION", "DecisionInput", + "DecisionKind", "DecisionModel", "DecisionModelOption", "DecisionModelRequest", diff --git a/src/powercontext/builtin/runtime/family_processing.py b/src/powercontext/builtin/runtime/family_processing.py index ef7a0ee616..78bca0f721 100644 --- a/src/powercontext/builtin/runtime/family_processing.py +++ b/src/powercontext/builtin/runtime/family_processing.py @@ -77,8 +77,10 @@ async def _run_family_worker( spec: FamilyWorkerSpec, assignment: ArtifactProcessingWorkAssignment ) -> ArtifactProcessingWorkerCompletion: from powercontext.builtin.runtime.composition import ( + _configured_memory_write_gate, _dream_generator, _embedding_models, + _fail_open_decision_model, _generation_pipelines, _prompt_registry, _usage_reporting_embedding_model, @@ -90,6 +92,13 @@ async def _run_family_worker( pipelines = await _generation_pipelines( config.inference, config.runtime, resources, None, BUILTIN_SOURCE_REGISTRY ) + decision_model = _fail_open_decision_model( + None, + pipelines[7], + None, + timeout_seconds=config.inference.decision_timeout_seconds or config.inference.generation_timeout_seconds, + ) + memory_write_gate = _configured_memory_write_gate(None, decision_model, config.runtime) embedding, _ = await _embedding_models(config.inference, resources, None) contexts = await resources.enter_async_context( open_builtin_contexts( @@ -97,6 +106,8 @@ async def _run_family_worker( candidate_pipeline=pipelines[1], experience_pipeline=pipelines[2], embedding_model=_usage_reporting_embedding_model(embedding), + decision_model=decision_model, + memory_write_gate=memory_write_gate, prompt_registry=_prompt_registry( config.runtime, ( diff --git a/src/powercontext/builtin/runtime/memory_write_gate.py b/src/powercontext/builtin/runtime/memory_write_gate.py new file mode 100644 index 0000000000..cb0940172b --- /dev/null +++ b/src/powercontext/builtin/runtime/memory_write_gate.py @@ -0,0 +1,196 @@ +# 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. + +"""Gate Memory writes on a decision-model evidence-sufficiency verdict. + +The gate answers one narrow question: are the citations behind a pending Memory write strong +enough to record it now? It never decides content, only whether to proceed. A ``HOLD`` is a +visible refusal — it carries a structured code and a bounded reason back to its caller — and +never a silent drop or an automatic approval. Any backend failure degrades to ``ACCEPT`` so an +unavailable judge can never block a write. +""" + +from __future__ import annotations + +import logging +from typing import Literal + +from powercontext._logging import log_safely +from powercontext.builtin.artifacts.memory.protocols import ( + MemoryWriteAssessment, + MemoryWriteGate, + MemoryWriteGateRequest, + MemoryWriteRejectionCode, + MemoryWriteVerdict, +) +from powercontext.builtin.runtime.decision_model import ( + DecisionKind, + DecisionModel, + DecisionOutcome, + DecisionRequest, + DecisionResult, +) + +logger = logging.getLogger(__name__) + +_GATE_QUESTION = ( + "Do the cited evidence items fail to support the proposed memory candidates, so that " + "writing them now would record unsupported content?" +) +# A held write cites the same bounded vocabulary as evidence selection; the ceiling mirrors +# that selector so an over-wide citation set is reported instead of silently accepted. +_MAX_EVIDENCE_ITEMS = 32 +_MAX_SUBJECT_LENGTH = 4000 +_MAX_REASON_LENGTH = 512 +_DEFAULT_REASON = "the cited evidence does not clearly support this memory write" + + +def build_memory_write_gate( + decision_model: DecisionModel | None, + *, + enabled: bool, + hold_on: Literal["yes", "no"] = "yes", + threshold: float | None = None, +) -> MemoryWriteGate | None: + """Build the opt-in gate, or return ``None`` while it stays disabled. + + The direction is supplied by configuration and must be calibrated before the gate is + enabled; a disabled gate and a missing backend both resolve to ``None``. + """ + + if not enabled or decision_model is None: + return None + return DecisionMemoryWriteGate(decision_model, hold_on=DecisionOutcome(hold_on), threshold=threshold) + + +class DecisionMemoryWriteGate: + """Map one decision-model verdict onto a Memory write verdict. + + ``hold_on`` is the calibrated outcome that means "the cited evidence is insufficient"; a + clear verdict in that direction becomes ``HOLD``, the opposite becomes ``ACCEPT``, and an + abstention or fallback passes the write through untouched. When a threshold is configured, + a hold-direction verdict whose confidence falls below it is downgraded to ``FLAG`` — written, + but annotated as uncertain. + """ + + def __init__( + self, + decision_model: DecisionModel, + /, + *, + hold_on: DecisionOutcome = DecisionOutcome.YES, + threshold: float | None = None, + ) -> None: + if hold_on is DecisionOutcome.ABSTAIN: + raise ValueError("the hold direction cannot be abstain") # noqa: TRY003 + self._decision_model = decision_model + self._hold_on = hold_on + self._threshold = threshold + self.policy_id = decision_model.policy_id + + 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( + verdict=MemoryWriteVerdict.HOLD, + policy_id=self.policy_id, + code=_rejection_code(request), + reason="the candidate batch exceeds the gate assessment budget", + ) + self._log(assessment) + return assessment + decision = await self._decision_model.evaluate( + DecisionRequest( + decision_kind=DecisionKind.MEMORY_WRITE_GATE.value, + question=_GATE_QUESTION, + subject=_bounded_subject(request.candidates), + evidence=request.evidence, + ) + ) + assessment = self._map(request, decision) + self._log(assessment) + return assessment + + def _map(self, request: MemoryWriteGateRequest, decision: DecisionResult) -> MemoryWriteAssessment: + if decision.used_fallback or decision.outcome is DecisionOutcome.ABSTAIN: + return MemoryWriteAssessment( + verdict=MemoryWriteVerdict.ACCEPT, + policy_id=decision.policy_id, + used_fallback=decision.used_fallback, + ) + if decision.outcome is not self._hold_on: + 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, + ) + + def _log(self, assessment: MemoryWriteAssessment) -> None: + event = { + MemoryWriteVerdict.HOLD: "memory.write-gate.hold", + MemoryWriteVerdict.FLAG: "memory.write-gate.flag", + }.get(assessment.verdict, "memory.write-gate.assess") + log_safely( + logger, + logging.INFO, + "Memory write gate assessed a pending write", + extra={ + "event": event, + "decision_kind": DecisionKind.MEMORY_WRITE_GATE.value, + "policy_id": assessment.policy_id, + "verdict": assessment.verdict.value, + "code": None if assessment.code is None else assessment.code.value, + "used_fallback": assessment.used_fallback, + }, + ) + + +def _rejection_code(request: MemoryWriteGateRequest) -> MemoryWriteRejectionCode: + if not request.evidence: + return MemoryWriteRejectionCode.NEEDS_EVIDENCE + if len(request.evidence) > _MAX_EVIDENCE_ITEMS: + return MemoryWriteRejectionCode.EVIDENCE_LIMIT_EXCEEDED + return MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE + + +def _bounded_subject(candidates: tuple[str, ...]) -> str: + return "\n".join(candidates)[:_MAX_SUBJECT_LENGTH] + + +def _subject_exceeds_limit(candidates: tuple[str, ...]) -> bool: + return len("\n".join(candidates)) > _MAX_SUBJECT_LENGTH + + +def _bounded_reason(value: str | None) -> str: + if value is None: + return _DEFAULT_REASON + normalized = value.strip() + return normalized[:_MAX_REASON_LENGTH] if normalized else _DEFAULT_REASON + + +__all__ = [ + "DecisionMemoryWriteGate", + "MemoryWriteAssessment", + "MemoryWriteGate", + "MemoryWriteGateRequest", + "MemoryWriteRejectionCode", + "MemoryWriteVerdict", + "build_memory_write_gate", +] diff --git a/src/powercontext/builtin/runtime/models.py b/src/powercontext/builtin/runtime/models.py index d7f37cf654..7d95e54b65 100644 --- a/src/powercontext/builtin/runtime/models.py +++ b/src/powercontext/builtin/runtime/models.py @@ -117,13 +117,19 @@ class RuntimeCapabilities(BaseModel): class MemoryFlushResult(BaseModel): - """Result of processing one scoped Source window.""" + """Result of processing one scoped Source window. + + ``held_count`` and ``hold_codes`` expose a gate refusal to the caller: a held window + advances its cursor but writes no Memory, and the structured code says why. + """ previous_cursor: int high_watermark: int current_cursor: int source_count: int memory_ref: ArtifactRef | None + held_count: int = 0 + hold_codes: tuple[str, ...] = () @property def processed(self) -> bool: diff --git a/src/powercontext/builtin/runtime/relational.py b/src/powercontext/builtin/runtime/relational.py index 1b96afc8d4..2089d2b09b 100644 --- a/src/powercontext/builtin/runtime/relational.py +++ b/src/powercontext/builtin/runtime/relational.py @@ -67,7 +67,9 @@ MemoryQueryEmbedding, MemoryReranker, MemoryService, + MemoryWriteGate, MemoryWritePlan, + MemoryWriteVerdict, ) from powercontext.builtin.artifacts.profile import Profile from powercontext.builtin.artifacts.profile.management import ProfileManagementWriter @@ -301,6 +303,7 @@ class _ScopedServices: memory_reranker: MemoryReranker | None memory_rerank_candidate_limit: int decision_model: DecisionModel | None + memory_write_gate: MemoryWriteGate | None memory_capacity_budget: MemoryCapacityBudget memory_compaction: MemoryCompactionPolicy memory_max_history_revisions: int @@ -366,6 +369,7 @@ def memory( connection=connection, ), id_factory=self.id_factory, + write_gate=self.memory_write_gate, ) def evidence(self, authorize: EvidenceAuthorizer | None = None) -> EvidenceResolver: @@ -521,6 +525,7 @@ def __init__( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, decision_model: DecisionModel | None = None, + memory_write_gate: MemoryWriteGate | None = None, memory_rerank_candidate_limit: int = 30, memory_capacity_budget: MemoryCapacityBudget | None = None, memory_compaction: MemoryCompactionPolicy | None = None, @@ -675,6 +680,7 @@ def __init__( self._token_estimator = token_estimator self._memory_reranker = memory_reranker self._decision_model = decision_model + self._memory_write_gate = memory_write_gate self._memory_rerank_candidate_limit = memory_rerank_candidate_limit self._memory_capacity_budget = ( MemoryCapacityBudget() if memory_capacity_budget is None else memory_capacity_budget @@ -1418,6 +1424,7 @@ def _services_for(self, scope_id: str) -> _ScopedServices: memory_reranker=self._memory_reranker, memory_rerank_candidate_limit=self._memory_rerank_candidate_limit, decision_model=self._decision_model, + memory_write_gate=self._memory_write_gate, memory_capacity_budget=self._memory_capacity_budget, memory_compaction=self._memory_compaction, memory_max_history_revisions=self._memory_max_history_revisions, @@ -1453,6 +1460,14 @@ def __init__( def as_ref(self, source: Source, /) -> SourceRef: return self._catalog.as_ref(source) + async def get_ref(self, ref: SourceRef, /) -> Source: + try: + async with self._database.connection(self._connection) as connection: + (stored,) = await self._access.require_for_generation(connection, self._scope_id, (ref,)) + except RepositoryNotFoundError: + raise SourceNotFoundError(ref) from None + return stored.value + async def get(self, source: Source, /) -> Source: try: async with self._database.connection(self._connection) as connection: @@ -1557,14 +1572,17 @@ def __init__( self._bound_connection = connection async def get(self, artifact: Artifact[object], /) -> Artifact[object]: + try: + return await self.get_ref(artifact.as_ref()) + except ArtifactNotFoundError: + raise ArtifactNotFoundError(artifact) from None + + async def get_ref(self, ref: ArtifactRef, /) -> Artifact[object]: try: async with self._database.connection(self._bound_connection) as connection: - return cast( - Artifact[object], - await self._repository.get(connection, self._scope_id, artifact.as_ref()), - ) + return cast(Artifact[object], await self._repository.get(connection, self._scope_id, ref)) except RepositoryNotFoundError: - raise ArtifactNotFoundError(artifact) from None + raise ArtifactNotFoundError(ref) from None class _RelationalTriggers: @@ -1686,6 +1704,7 @@ async def flush( prepared = ( None if not sources else await self._prepare_memory(sources, authorize_snapshot=authorize_snapshot) ) + held = _is_held_write(prepared) commit = None if prepared is None else prepared.commit with self._stage( _MEMORY_COMMIT_STAGE, @@ -1719,6 +1738,8 @@ async def flush( current_cursor=action.through, source_count=len(sources), memory_ref=None if updated is None else updated.as_ref(), + held_count=1 if held else 0, + hold_codes=_hold_codes(prepared), ) async def _sources( @@ -1983,6 +2004,26 @@ def _validate_schema_value(name: str, schema: Mapping[str, Any], value: object) raise InvalidSourceObservationError("schema", f"value does not match {name!r}") from error +def _is_held_write(plan: MemoryWritePlan | None) -> bool: + """Report whether the gate refused this prepared write.""" + + if plan is None: + return False + decision = plan.decision + return decision is not None and decision.verdict is MemoryWriteVerdict.HOLD + + +def _hold_codes(plan: MemoryWritePlan | None) -> tuple[str, ...]: + """Expose the structured refusal code of a held write to the window caller.""" + + if not _is_held_write(plan) or plan is None: + return () + decision = plan.decision + if decision is None or decision.code is None: + return () + return (decision.code.value,) + + def _scoped_id_factory(memory_artifact_id: str, delegate: IdFactory | None) -> IdFactory: def new_id(kind: str) -> str: if kind == "memory": diff --git a/src/powercontext/http/_generated/models.py b/src/powercontext/http/_generated/models.py index 7d41c3f5af..4ff5c26339 100644 --- a/src/powercontext/http/_generated/models.py +++ b/src/powercontext/http/_generated/models.py @@ -3235,6 +3235,12 @@ class FlushMemoryResponse(BaseModel): high_watermark: Annotated[StrictInt, Field(ge=0)] processed_source_count: Annotated[StrictInt, Field(ge=0)] memory: ArtifactReference | None = None + held_count: Annotated[ + StrictInt, Field(description="Number of source windows held by the Memory write gate.", ge=0) + ] = 0 + hold_codes: Annotated[ + list[StrictStr], Field(description="Structured Memory write gate refusal codes for held windows.") + ] = [] class FlushTopicMemoryResponse(BaseModel): diff --git a/src/powercontext/http/_generated/schema.py b/src/powercontext/http/_generated/schema.py index 508871f72a..4f89eb9cf5 100644 --- a/src/powercontext/http/_generated/schema.py +++ b/src/powercontext/http/_generated/schema.py @@ -7397,6 +7397,18 @@ "high_watermark": {"type": "integer", "minimum": 0.0}, "processed_source_count": {"type": "integer", "minimum": 0.0}, "memory": {"$ref": "#/components/schemas/ArtifactReference", "nullable": True}, + "held_count": { + "type": "integer", + "minimum": 0.0, + "description": "Number of source windows held by the Memory write gate.", + "default": 0, + }, + "hold_codes": { + "items": {"type": "string"}, + "type": "array", + "description": "Structured Memory write gate refusal codes for held windows.", + "default": [], + }, }, "additionalProperties": False, "type": "object", diff --git a/src/powercontext/server/app.py b/src/powercontext/server/app.py index 58e5d7649a..fcc85e1a37 100644 --- a/src/powercontext/server/app.py +++ b/src/powercontext/server/app.py @@ -66,6 +66,7 @@ MemoryCapacityExceededError, MemoryEntryInactiveError, MemoryEntryNotFoundError, + MemoryWriteRejectedError, ) from powercontext.builtin.artifacts.memory.models import MemoryCapacity as RuntimeMemoryCapacity from powercontext.builtin.artifacts.prompt import GeneratePromptDemonstrations, PromptError @@ -5481,6 +5482,31 @@ def _map_domain_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | source_ingestion = _map_source_ingestion_error(error) if source_ingestion is not None: return source_ingestion + memory_error = _map_memory_error(error) + if memory_error is not None: + return memory_error + if isinstance( + error, + ( + HandoffScopeMismatchError, + InvalidHandoffReferenceError, + InvalidRuntimeRequestError, + ), + ): + return ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + "invalid_request", + "The request is invalid.", + _invalid_request_details(error), + ) + if isinstance(error, InferenceTimeoutError): + return status.HTTP_503_SERVICE_UNAVAILABLE, "inference_timeout", "Model inference timed out.", None + if isinstance(error, InferenceUnavailableError): + return status.HTTP_503_SERVICE_UNAVAILABLE, "inference_unavailable", "Model inference is unavailable.", None + return status.HTTP_500_INTERNAL_SERVER_ERROR, "internal_error", "The Server failed.", None + + +def _map_memory_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | None] | None: if isinstance(error, ArtifactNotFoundError): return status.HTTP_404_NOT_FOUND, "artifact_not_found", "The requested Artifact was not found.", None if isinstance(error, MemoryEntryNotFoundError): @@ -5488,6 +5514,13 @@ def _map_domain_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | memory_conflict = _map_memory_conflict_error(error) if memory_conflict is not None: return memory_conflict + if isinstance(error, MemoryWriteRejectedError): + return ( + status.HTTP_422_UNPROCESSABLE_CONTENT, + "memory_write_rejected", + "The Memory write was rejected by the configured gate.", + {"code": error.code, "reason": error.reason}, + ) if isinstance(error, CapabilityNotSupportedError): return ( status.HTTP_422_UNPROCESSABLE_CONTENT, @@ -5501,9 +5534,6 @@ def _map_domain_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | InvalidMemoryCandidateError, InvalidMemoryCitationError, InvalidMemoryEvidenceError, - HandoffScopeMismatchError, - InvalidHandoffReferenceError, - InvalidRuntimeRequestError, ), ): return ( @@ -5512,11 +5542,7 @@ def _map_domain_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | "The request is invalid.", _invalid_request_details(error), ) - if isinstance(error, InferenceTimeoutError): - return status.HTTP_503_SERVICE_UNAVAILABLE, "inference_timeout", "Model inference timed out.", None - if isinstance(error, InferenceUnavailableError): - return status.HTTP_503_SERVICE_UNAVAILABLE, "inference_unavailable", "Model inference is unavailable.", None - return status.HTTP_500_INTERNAL_SERVER_ERROR, "internal_error", "The Server failed.", None + return None def _map_memory_conflict_error(error: Exception) -> tuple[int, str, str, dict[str, Any] | None] | None: diff --git a/src/powercontext/server/mapping.py b/src/powercontext/server/mapping.py index 913bd45d90..bd13026be5 100644 --- a/src/powercontext/server/mapping.py +++ b/src/powercontext/server/mapping.py @@ -561,6 +561,8 @@ def flush_response(value: MemoryFlushResult) -> FlushMemoryResponse: high_watermark=value.high_watermark, processed_source_count=value.source_count, memory=None if value.memory_ref is None else artifact_reference(value.memory_ref), + held_count=value.held_count, + hold_codes=list(value.hold_codes), ) diff --git a/tests/builtin/runtime/test_batch2_config.py b/tests/builtin/runtime/test_batch2_config.py new file mode 100644 index 0000000000..f5f0b1fb17 --- /dev/null +++ b/tests/builtin/runtime/test_batch2_config.py @@ -0,0 +1,54 @@ +# 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.runtime import RuntimeConfig +from powercontext.server.settings import ServerSettings + + +def test_memory_write_gate_is_disabled_by_default() -> None: + config = RuntimeConfig() + + assert config.memory_write_gate_enabled is False + assert config.memory_write_gate_hold_on == "yes" + assert config.memory_write_gate_threshold is None + + +def test_the_hold_direction_is_configurable() -> None: + assert RuntimeConfig(memory_write_gate_hold_on="no").memory_write_gate_hold_on == "no" + + +def test_an_unknown_hold_direction_is_rejected() -> None: + with pytest.raises(ValidationError): + RuntimeConfig.model_validate({"memory_write_gate_hold_on": "maybe"}) + + +@pytest.mark.parametrize("threshold", [-0.1, 1.1]) +def test_the_strength_threshold_is_bounded(threshold: float) -> None: + with pytest.raises(ValidationError): + RuntimeConfig.model_validate({"memory_write_gate_threshold": threshold}) + + +def test_the_gate_settings_load_from_the_server_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_ENABLED", "true") + monkeypatch.setenv("POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_HOLD_ON", "no") + + runtime = ServerSettings().runtime + + assert runtime.memory_write_gate_enabled is True + assert runtime.memory_write_gate_hold_on == "no" diff --git a/tests/builtin/runtime/test_family_processing.py b/tests/builtin/runtime/test_family_processing.py index d0f6a30ab1..3f843d7cf2 100644 --- a/tests/builtin/runtime/test_family_processing.py +++ b/tests/builtin/runtime/test_family_processing.py @@ -17,12 +17,15 @@ from __future__ import annotations import asyncio +from contextlib import asynccontextmanager from dataclasses import replace from functools import partial 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.experience import ExperienceCandidateInput, ExperienceContent from powercontext.builtin.artifacts.memory import MemoryEntryInput from powercontext.builtin.inference.models import GenerationResult, InferenceUsage @@ -31,7 +34,7 @@ from powercontext.builtin.persistence.processing_intents import ArtifactProcessingIntentRepository from powercontext.builtin.persistence.processing_migration import bootstrap_processing_schema from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile -from powercontext.builtin.persistence.supervision import ArtifactProcessingLeaseRepository +from powercontext.builtin.persistence.supervision import ArtifactProcessingFence, ArtifactProcessingLeaseRepository from powercontext.builtin.persistence.tables import ( ARTIFACT_CANDIDATE_HEADS_TABLE, ARTIFACT_HEADS_TABLE, @@ -49,6 +52,7 @@ ) from powercontext.builtin.runtime.processing_contracts import ( ArtifactProcessingWorkAssignment, + ArtifactProcessingWorkerCompletion, ArtifactProcessingWorkerOutcome, ) from powercontext.builtin.runtime.processing_registry import canonical_processing_manifest @@ -91,6 +95,22 @@ async def generate(self, value): return "# Preferences\n\nVerify every change." +class _FakeDecisionModel: + policy_id = "test.worker.decision.v1" + + async def evaluate(self, request, /): + raise AssertionError + + +class _WorkerProfiles: + generator = None + max_sources = 0 + + +class _WorkerContexts: + profiles = _WorkerProfiles() + + async def prepare(profile, family): contexts = RelationalContexts( database=profile.database, candidate_pipeline=MemoryPipeline(), experience_pipeline=ExperiencePipeline() @@ -328,6 +348,69 @@ async def scenario(): asyncio.run(scenario()) +def test_spawned_memory_worker_reconstructs_configured_write_gate(monkeypatch, tmp_path): + captured: dict[str, object] = {} + + async def fake_generation_pipelines(*_args, **_kwargs): + return (None, MemoryPipeline(), None, None, None, None, None, _FakeDecisionModel(), None, None, None) + + async def fake_embedding_models(*_args, **_kwargs): + return object(), None + + def fake_usage_reporting_embedding_model(value): + return value + + def fake_prompt_registry(*_args, **_kwargs): + return object() + + @asynccontextmanager + async def fake_open_builtin_contexts(*_args, **kwargs): + captured.update(kwargs) + yield _WorkerContexts() + + async def fake_process_family_invocation(contexts, assignment, *, config, security=None, dream_generator=None): + assert contexts is not None + assert assignment.artifact_family == "memory" + assert config.runtime.memory_write_gate_enabled is True + assert security is None + assert dream_generator is None + return ArtifactProcessingWorkerCompletion() + + monkeypatch.setattr(composition, "_generation_pipelines", fake_generation_pipelines) + monkeypatch.setattr(composition, "_embedding_models", fake_embedding_models) + monkeypatch.setattr(composition, "_usage_reporting_embedding_model", fake_usage_reporting_embedding_model) + monkeypatch.setattr(composition, "_prompt_registry", fake_prompt_registry) + monkeypatch.setattr(composition, "open_builtin_contexts", fake_open_builtin_contexts) + monkeypatch.setattr(family_processing, "process_family_invocation", fake_process_family_invocation) + + async def scenario(): + config = BuiltinConfig( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'worker-gate.db'}"), + 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", + ) + + result = await family_processing._run_family_worker(FamilyWorkerSpec(config=config), assignment) + + assert result.outcome == ArtifactProcessingWorkerOutcome.SUCCEEDED + assert captured["decision_model"] is not None + assert captured["memory_write_gate"] is not None + + asyncio.run(scenario()) + + def test_dedicated_fence_cannot_commit_another_family(tmp_path): async def scenario(): config = BuiltinConfig(database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'wrong-family.db'}")) diff --git a/tests/builtin/runtime/test_memory_write_gate_contract.py b/tests/builtin/runtime/test_memory_write_gate_contract.py new file mode 100644 index 0000000000..5812649e24 --- /dev/null +++ b/tests/builtin/runtime/test_memory_write_gate_contract.py @@ -0,0 +1,322 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import asyncio + +import pytest + +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import ( + DecisionKind, + DecisionOutcome, + DecisionRequest, + DecisionResult, + FailOpenDecisionModel, +) +from powercontext.builtin.runtime.memory_write_gate import ( + DecisionMemoryWriteGate, + MemoryWriteGateRequest, + MemoryWriteRejectionCode, + MemoryWriteVerdict, + build_memory_write_gate, +) + +# A known-answer pair used to orient the hold direction. A durable preference should pass an +# evidence check; filler should not. The direction must be established from this pair before the +# gate is enabled, so a backend with the opposite polarity is caught instead of silently holding +# the wrong writes. +_PROBE_PREFERENCE = "Always run the full test suite before committing." +_PROBE_FILLER = "ok sounds good sure" + + +class _StaticDecisionModel: + """A backend that always returns one prepared verdict.""" + + policy_id = "powercontext.decision.static.v1" + + def __init__(self, result: DecisionResult) -> None: + self._result = result + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + return self._result + + +class _RecordingDecisionModel(_StaticDecisionModel): + def __init__(self, result: DecisionResult) -> None: + super().__init__(result) + self.requests: list[DecisionRequest] = [] + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + self.requests.append(request) + return await super().evaluate(request) + + +class _FailingDecisionModel: + """A backend whose every evaluation raises.""" + + policy_id = "powercontext.decision.failing.v1" + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + raise ValueError("backend unavailable") # noqa: TRY003 + + +class _PolarityBackend: + """A backend that calls one known answer insufficient and the other sufficient.""" + + policy_id = "powercontext.decision.polarity.v1" + + def __init__(self, *, insufficient_for: frozenset[str]) -> None: + self._insufficient_for = insufficient_for + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + outcome = DecisionOutcome.YES if request.subject in self._insufficient_for else DecisionOutcome.NO + return DecisionResult(outcome, self.policy_id, InferenceUsage(requests=1)) + + +def _verdict( + outcome: DecisionOutcome, + *, + confidence: float | None = None, + rationale: str | None = None, + used_fallback: bool = False, +) -> DecisionResult: + return DecisionResult( + outcome, + "powercontext.decision.static.v1", + InferenceUsage(requests=1), + rationale=rationale, + confidence=confidence, + used_fallback=used_fallback, + ) + + +def test_decision_kind_values_are_stable() -> None: + assert DecisionKind.MEMORY_WRITE_GATE.value == "memory.write-gate" + + +def test_gate_vocabulary_is_complete() -> None: + assert {verdict.value for verdict in MemoryWriteVerdict} == {"accept", "flag", "hold"} + assert {code.value for code in MemoryWriteRejectionCode} == { + "needs_evidence", + "evidence_limit_exceeded", + "insufficient_coverage", + } + + +def test_a_supporting_verdict_accepts_the_write() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.NO)), hold_on=DecisionOutcome.YES) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.code is None + assert assessment.reason is None + + asyncio.run(scenario()) + + +def test_the_hold_direction_is_read_from_configuration() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.NO)), hold_on=DecisionOutcome.NO) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + + asyncio.run(scenario()) + + +def test_a_hold_carries_a_structured_code_and_reason() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES, rationale="the citation is thin")), + hold_on=DecisionOutcome.YES, + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + assert assessment.code is MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE + assert assessment.reason == "the citation is thin" + + asyncio.run(scenario()) + + +def test_a_hold_without_citations_reports_needs_evidence() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), hold_on=DecisionOutcome.YES) + + assessment = await gate.assess(_request(evidence=())) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + assert assessment.code is MemoryWriteRejectionCode.NEEDS_EVIDENCE + assert assessment.reason is not None + + asyncio.run(scenario()) + + +def test_a_hold_beyond_the_evidence_ceiling_reports_limit_exceeded() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), hold_on=DecisionOutcome.YES) + + assessment = await gate.assess(_request(evidence=tuple(f"source:task:{index}" for index in range(40)))) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + assert assessment.code is MemoryWriteRejectionCode.EVIDENCE_LIMIT_EXCEEDED + + asyncio.run(scenario()) + + +def test_an_oversized_candidate_batch_is_held_before_backend_assessment() -> None: + async def scenario() -> None: + backend = _RecordingDecisionModel(_verdict(DecisionOutcome.NO)) + gate = DecisionMemoryWriteGate(backend, hold_on=DecisionOutcome.YES) + + assessment = await gate.assess( + _request( + candidates=("x" * 4000, "UNASSESSED_TAIL"), + evidence=("source:task:1\nsupporting text",), + ) + ) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + assert assessment.code is MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE + assert assessment.reason == "the candidate batch exceeds the gate assessment budget" + assert backend.requests == [] + + asyncio.run(scenario()) + + +def test_a_low_confidence_hold_is_written_but_flagged() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES, confidence=0.2, rationale="uncertain")), + hold_on=DecisionOutcome.YES, + threshold=0.5, + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.FLAG + assert assessment.code is None + assert assessment.reason == "uncertain" + + asyncio.run(scenario()) + + +def test_a_confident_hold_still_holds_with_a_threshold() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.YES, confidence=0.9)), + hold_on=DecisionOutcome.YES, + threshold=0.5, + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.HOLD + + asyncio.run(scenario()) + + +def test_a_deliberate_abstention_passes_the_write_through() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate( + _StaticDecisionModel(_verdict(DecisionOutcome.ABSTAIN)), hold_on=DecisionOutcome.YES + ) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.used_fallback is False + + asyncio.run(scenario()) + + +def test_a_failing_backend_is_fail_open() -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(FailOpenDecisionModel(_FailingDecisionModel()), hold_on=DecisionOutcome.YES) + + assessment = await gate.assess(_request(evidence=("source:task:1",))) + + assert assessment.verdict is MemoryWriteVerdict.ACCEPT + assert assessment.code is None + assert assessment.reason is None + assert assessment.used_fallback is True + + asyncio.run(scenario()) + + +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) + + +def test_noul_polarity_probe_pairs_a_known_answer_with_the_direction() -> None: + async def scenario() -> None: + backend = _PolarityBackend(insufficient_for=frozenset({_PROBE_FILLER})) + gate = DecisionMemoryWriteGate(backend, hold_on=DecisionOutcome.YES) + + preference = await gate.assess(_request(candidates=(_PROBE_PREFERENCE,))) + filler = await gate.assess(_request(candidates=(_PROBE_FILLER,))) + + assert preference.verdict is MemoryWriteVerdict.ACCEPT + assert filler.verdict is MemoryWriteVerdict.HOLD + + asyncio.run(scenario()) + + +def test_noul_polarity_probe_exposes_a_contradictory_backend() -> None: + async def scenario() -> None: + # The same pair with the opposite polarity: the probe must fail rather than let a + # reversed backend enable the gate against real preferences. + backend = _PolarityBackend(insufficient_for=frozenset({_PROBE_PREFERENCE})) + gate = DecisionMemoryWriteGate(backend, hold_on=DecisionOutcome.YES) + + preference = await gate.assess(_request(candidates=(_PROBE_PREFERENCE,))) + + assert preference.verdict is MemoryWriteVerdict.HOLD + + asyncio.run(scenario()) + + +def test_the_gate_stays_disabled_without_an_enabled_flag() -> None: + assert build_memory_write_gate(_StaticDecisionModel(_verdict(DecisionOutcome.YES)), enabled=False) is None + + +def test_the_gate_stays_disabled_without_a_backend() -> None: + assert build_memory_write_gate(None, enabled=True) is None + + +def test_the_gate_is_built_from_configuration() -> None: + gate = build_memory_write_gate( + _StaticDecisionModel(_verdict(DecisionOutcome.NO)), + enabled=True, + hold_on="no", + threshold=0.4, + ) + + assert isinstance(gate, DecisionMemoryWriteGate) + assert gate.policy_id == "powercontext.decision.static.v1" + + +def _request( + *, + candidates: tuple[str, ...] = ("Remember the contract change.",), + evidence: tuple[str, ...] = (), +) -> MemoryWriteGateRequest: + return MemoryWriteGateRequest(candidates=candidates, evidence=evidence, expected_revision=1) diff --git a/tests/builtin/runtime/test_memory_write_gate_paths.py b/tests/builtin/runtime/test_memory_write_gate_paths.py new file mode 100644 index 0000000000..34ed5fb519 --- /dev/null +++ b/tests/builtin/runtime/test_memory_write_gate_paths.py @@ -0,0 +1,612 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path + +import pytest +from pydantic import BaseModel + +from powercontext.builtin.artifacts.experience import Experience, ExperienceContent, ExperienceDraft +from powercontext.builtin.artifacts.memory import ( + MemoryCandidateRequest, + MemoryEntryInput, + MemoryWriteAssessment, + MemoryWriteGateRequest, + MemoryWriteRejectionCode, + MemoryWriteVerdict, +) +from powercontext.builtin.artifacts.memory.errors import MemoryWriteRejectedError +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.runtime import ( + BuiltinConfig, + BuiltinRuntime, + CaptureSource, + MemoryFlushResult, + RememberMemoryRequest, + RuntimeConfig, + SubmitSourceObservation, + open_builtin_contexts, + open_builtin_runtime, +) +from powercontext.builtin.runtime.decision_model import ( + DecisionOutcome, + DecisionRequest, + DecisionResult, + FailOpenDecisionModel, +) +from powercontext.builtin.runtime.memory_write_gate import DecisionMemoryWriteGate +from powercontext.builtin.scope import ScopeDraft +from powercontext.builtin.sources import ContentCapture, ContentSource +from powercontext.server import mapping +from powercontext.server.app import _map_error +from powercontext.sources import ( + TEXT_EVIDENCE_PROJECTION_KEY, + AdapterSourceDefinition, + Source, + SourceDefinitionRegistry, + SourceMaterialization, + TextEvidence, + manifest_for_definition, + project_source_for_transport, +) + +_SCRIPTED_POLICY_ID = "test.memory.write-gate.v1" + + +class _RemoteNoteCapture(BaseModel): + source_id: str + content: str + + +class _RemoteNoteSource(Source): + note: str + + +class _RemoteNoteAdapter: + name = "remote-note" + input_class = _RemoteNoteCapture + source_class = _RemoteNoteSource + + async def resolve(self, value: _RemoteNoteCapture, /) -> _RemoteNoteSource: + return _RemoteNoteSource( + name=value.source_id, materialization=SourceMaterialization.CAPTURED, note=value.content + ) + + async def read(self, source: _RemoteNoteSource, /) -> _RemoteNoteCapture: + return _RemoteNoteCapture(source_id=source.name, content=source.note) + + +class _RemoteNoteTextEvidenceProjection: + name = TEXT_EVIDENCE_PROJECTION_KEY.name + version = TEXT_EVIDENCE_PROJECTION_KEY.version + source_class = _RemoteNoteSource + output_class: type[BaseModel] = TextEvidence + + def project(self, source: _RemoteNoteSource, /) -> TextEvidence: + return TextEvidence(source_type="remote-note", source_id=source.name, content=source.note) + + +_REMOTE_NOTE_DEFINITION = AdapterSourceDefinition( + _RemoteNoteAdapter(), + projections=(_RemoteNoteTextEvidenceProjection(),), +) + + +class _ScriptedGate: + """A gate that returns one prepared assessment and records its requests.""" + + policy_id = _SCRIPTED_POLICY_ID + + def __init__(self, assessment: MemoryWriteAssessment) -> None: + self._assessment = assessment + self.requests: list[MemoryWriteGateRequest] = [] + + async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessment: + self.requests.append(request) + return self._assessment + + +class _FailingGate: + policy_id = "test.memory.write-gate.failing.v1" + + async def assess(self, request: MemoryWriteGateRequest, /) -> MemoryWriteAssessment: + raise ValueError("gate unavailable") # noqa: TRY003 + + +class _FailingDecisionModel: + policy_id = "test.decision.failing.v1" + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + raise ValueError("backend unavailable") # noqa: TRY003 + + +class _InsufficientDecisionModel: + """A backend that always answers "evidence is insufficient" (the hold direction).""" + + policy_id = "test.decision.insufficient.v1" + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + return DecisionResult(DecisionOutcome.YES, self.policy_id, InferenceUsage(requests=1)) + + +class _ContentCandidatePipeline: + async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntryInput, ...]: + return tuple( + MemoryEntryInput(kind="fact", text=source.content, sources=(source,)) + for source in request.sources + if isinstance(source, ContentSource) + ) + + +def _assessment( + verdict: MemoryWriteVerdict, + *, + code: MemoryWriteRejectionCode | None = None, + reason: str | None = None, +) -> MemoryWriteAssessment: + return MemoryWriteAssessment(verdict=verdict, policy_id=_SCRIPTED_POLICY_ID, code=code, reason=reason) + + +def _config(tmp_path: Path, runtime: RuntimeConfig | None = None, database: str = "gate.db") -> BuiltinConfig: + return BuiltinConfig( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / database}"), + runtime=RuntimeConfig() if runtime is None else runtime, + ) + + +async def _create_scope(runtime: BuiltinRuntime, idempotency_key: str) -> str: + assert runtime.scopes is not None + scope = await runtime.scopes.create( + ScopeDraft(title="Gate Test", summary="Memory write gate path test", idempotency_key=idempotency_key) + ) + return scope.scope_id + + +def test_an_accepted_write_behaves_like_the_baseline(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + service = (await contexts.get("project")).artifacts.memory + + plan = await service.plan_remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Accepted."),), + mode="append", + ) + + assert plan.commit is not None + assert plan.decision is not None + assert plan.decision.verdict is MemoryWriteVerdict.ACCEPT + assert [change.reason for change in plan.commit.memory.content.changes] == [None] + assert gate.requests + + asyncio.run(scenario()) + + +def test_a_flagged_write_is_annotated_and_still_committed(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.FLAG, reason="evidence is thin")) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + service = (await contexts.get("project")).artifacts.memory + + stored = await service.remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Flagged."),), + mode="append", + ) + + assert stored is not None + assert [change.reason for change in stored.content.changes] == ["evidence is thin"] + + asyncio.run(scenario()) + + +def test_a_flagged_write_preserves_an_existing_candidate_reason(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.FLAG, reason="evidence is thin")) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + service = (await contexts.get("project")).artifacts.memory + + stored = await service.remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Annotated.", reason="an explicit reason"),), + mode="append", + ) + + assert stored is not None + assert [change.reason for change in stored.content.changes] == ["an explicit reason"] + + asyncio.run(scenario()) + + +def test_config_enables_the_gate_over_the_decision_backend(tmp_path: Path) -> None: + async def scenario() -> None: + config = _config(tmp_path, RuntimeConfig(memory_write_gate_enabled=True), database="enabled.db") + async with open_builtin_runtime(config, decision_model=_InsufficientDecisionModel()) as runtime: + scope_id = await _create_scope(runtime, "gate-config-enabled") + with pytest.raises(MemoryWriteRejectedError) as error: + await runtime.memory.for_scope(scope_id).remember( + RememberMemoryRequest(entries=(MemoryEntryInput(kind="note", text="Held by config."),)) + ) + + # The config-built gate is active, and the explicit write cites no evidence. + assert error.value.code == "needs_evidence" + + asyncio.run(scenario()) + + +def test_enabling_the_gate_without_a_backend_warns_and_passes_writes_through( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + async def scenario() -> None: + config = _config(tmp_path, RuntimeConfig(memory_write_gate_enabled=True), database="unavailable.db") + async with open_builtin_runtime(config) as runtime: + scope_id = await _create_scope(runtime, "gate-config-unavailable") + written = await runtime.memory.for_scope(scope_id).remember( + RememberMemoryRequest(entries=(MemoryEntryInput(kind="note", text="Written anyway."),)) + ) + + assert written.memory_ref is not None + + with caplog.at_level(logging.WARNING, logger="powercontext.builtin.runtime.composition"): + asyncio.run(scenario()) + + assert any("no decision backend is available" in message for message in caplog.messages) + assert "memory.write-gate.unavailable" in {getattr(record, "event", None) for record in caplog.records} + + +def test_a_held_write_is_not_committed_and_stays_visible(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate( + _assessment( + MemoryWriteVerdict.HOLD, + code=MemoryWriteRejectionCode.NEEDS_EVIDENCE, + reason="the candidate cites no evidence", + ) + ) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + service = (await contexts.get("project")).artifacts.memory + + plan = await service.plan_remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Held."),), + mode="append", + ) + + assert plan.commit is None + assert plan.decision is not None + assert plan.decision.verdict is MemoryWriteVerdict.HOLD + assert plan.decision.code is MemoryWriteRejectionCode.NEEDS_EVIDENCE + assert plan.decision.reason == "the candidate cites no evidence" + # No head is written, and the refusal is not silently dropped. + with pytest.raises(MemoryWriteRejectedError) as error: + await service.remember(memory=None, entries=(MemoryEntryInput(kind="note", text="Held."),)) + assert error.value.code == "needs_evidence" + + asyncio.run(scenario()) + + +def test_a_failing_backend_leaves_the_write_unchanged(tmp_path: Path) -> None: + async def scenario() -> None: + gate = DecisionMemoryWriteGate(FailOpenDecisionModel(_FailingDecisionModel()), hold_on=DecisionOutcome.YES) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + service = (await contexts.get("project")).artifacts.memory + + stored = await service.remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Passed through."),), + mode="append", + ) + + assert stored is not None + + asyncio.run(scenario()) + + +def test_a_failing_injected_gate_leaves_the_write_unchanged(tmp_path: Path) -> None: + async def scenario() -> None: + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=_FailingGate()) as contexts: + service = (await contexts.get("project")).artifacts.memory + + stored = await service.remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Passed through."),), + mode="append", + ) + + assert stored is not None + + asyncio.run(scenario()) + + +def test_revisions_pass_inherited_source_content_to_the_gate(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + context = await contexts.get("project") + source, _ = await context.sources.capture( + ContentCapture(source_id="db-requirements", content="The database requirement is MySQL 8.") + ) + initial = await context.artifacts.memory.remember( + memory=None, + sources=(source,), + entries=(MemoryEntryInput(kind="fact", text="Use MySQL 8.", sources=(source,)),), + mode="append", + ) + assert initial is not None + entry = (await context.artifacts.memory.entries(initial))[0] + + revised = await context.artifacts.memory.remember( + memory=initial, + entries=(MemoryEntryInput(kind="fact", text="Use PostgreSQL.", entry=entry),), + mode="append", + ) + + assert revised is not None + assert "The database requirement is MySQL 8." in "\n".join(gate.requests[-1].evidence) + + asyncio.run(scenario()) + + +def test_artifact_evidence_passes_content_to_the_gate(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate( + _assessment( + MemoryWriteVerdict.HOLD, + code=MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE, + reason="artifact content was inspected", + ) + ) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + context = await contexts.get("project") + draft = ExperienceDraft( + content=ExperienceContent( + situation="The write path used SQLite.", + action="Checked the gate request.", + outcome="The controlled outcome was HOLD.", + lesson="Opposite outcome text must be visible to the judge.", + ), + ) + async with contexts.database.transaction() as connection: + stored = await contexts.repositories.artifacts.create( + connection, "project", "experience-db-outcome", draft + ) + artifact = Experience.model_validate(stored.model_dump(mode="json")) + + plan = await context.artifacts.memory.plan_remember( + memory=None, + artifacts=(artifact,), + entries=(MemoryEntryInput(kind="fact", text="Gate outcome was ACCEPT.", artifacts=(artifact,)),), + mode="append", + ) + + assert plan.commit is None + evidence = "\n".join(gate.requests[-1].evidence) + assert "Opposite outcome text must be visible to the judge." in evidence + + asyncio.run(scenario()) + + +def test_gate_reads_remote_source_text_evidence_projection(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + registry = SourceDefinitionRegistry((_REMOTE_NOTE_DEFINITION,)) + source = await registry.resolve( + _RemoteNoteCapture(source_id="remote-1", content="Remote projection says use MySQL.") + ) + observed = project_source_for_transport(registry, source) + async with open_builtin_contexts( + _config(tmp_path), + memory_write_gate=gate, + ) as contexts: + scope = await contexts.scopes.create( + ScopeDraft(title="Remote", summary="Remote source test", idempotency_key="remote-source-gate") + ) + await contexts.register_source_definition(manifest_for_definition(_REMOTE_NOTE_DEFINITION)) + await contexts.submit_source_observation( + SubmitSourceObservation(scope_id=scope.scope_id, observation=observed) + ) + context = await contexts.get(scope.scope_id) + + plan = await context.artifacts.memory.plan_remember( + memory=None, + sources=(observed,), + entries=(MemoryEntryInput(kind="fact", text="Use MySQL.", sources=(observed,)),), + mode="append", + ) + + assert plan.commit is not None + assert any("Remote projection says use MySQL." in item for item in gate.requests[-1].evidence) + + asyncio.run(scenario()) + + +def test_gate_budgets_only_effective_candidate_citations(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + context = await contexts.get("project") + short_source, _ = await context.sources.capture(ContentCapture(source_id="short", content="Use MySQL.")) + unrelated_long_source, _ = await context.sources.capture( + ContentCapture(source_id="build-log", content="unrelated " * 300) + ) + + plan = await context.artifacts.memory.plan_remember( + memory=None, + sources=(short_source, unrelated_long_source), + entries=(MemoryEntryInput(kind="fact", text="Use MySQL.", sources=(short_source,)),), + mode="append", + ) + + assert plan.commit is not None + evidence = "\n".join(gate.requests[-1].evidence) + assert "Use MySQL." in evidence + assert "build-log" not in evidence + + asyncio.run(scenario()) + + +def test_gate_preserves_each_candidate_citation_mapping(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + context = await contexts.get("project") + alpha, _ = await context.sources.capture(ContentCapture(source_id="alpha", content="Alpha uses MySQL.")) + beta, _ = await context.sources.capture(ContentCapture(source_id="beta", content="Beta uses PostgreSQL.")) + + await context.artifacts.memory.plan_remember( + memory=None, + sources=(alpha, beta), + entries=( + MemoryEntryInput(kind="fact", text="Alpha uses MySQL.", sources=(alpha,)), + MemoryEntryInput(kind="fact", text="Beta uses PostgreSQL.", sources=(beta,)), + ), + mode="append", + ) + + evidence = "\n".join(gate.requests[-1].evidence) + assert "candidate:1 source:content:alpha" in evidence + assert "Alpha uses MySQL." in evidence + assert "candidate:2 source:content:beta" in evidence + assert "Beta uses PostgreSQL." in evidence + + asyncio.run(scenario()) + + +def test_incomplete_gate_evidence_is_held_before_backend_assessment(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate(_assessment(MemoryWriteVerdict.ACCEPT)) + async with open_builtin_contexts(_config(tmp_path), memory_write_gate=gate) as contexts: + context = await contexts.get("project") + long_source, _ = await context.sources.capture( + ContentCapture(source_id="corrected-requirements", content=f"{'PostgreSQL first. ' * 150}Use MySQL.") + ) + + plan = await context.artifacts.memory.plan_remember( + memory=None, + sources=(long_source,), + entries=(MemoryEntryInput(kind="fact", text="Use PostgreSQL.", sources=(long_source,)),), + mode="append", + ) + + assert plan.commit is None + assert plan.decision is not None + assert plan.decision.verdict is MemoryWriteVerdict.HOLD + assert plan.decision.code is MemoryWriteRejectionCode.EVIDENCE_LIMIT_EXCEEDED + assert gate.requests == [] + + asyncio.run(scenario()) + + +def test_without_a_gate_the_plan_carries_no_decision(tmp_path: Path) -> None: + async def scenario() -> None: + async with open_builtin_contexts(_config(tmp_path)) as contexts: + service = (await contexts.get("project")).artifacts.memory + + plan = await service.plan_remember( + memory=None, + entries=(MemoryEntryInput(kind="note", text="Plain."),), + mode="append", + ) + + assert plan.decision is None + assert plan.commit is not None + + asyncio.run(scenario()) + + +def test_the_explicit_write_surfaces_a_hold_as_a_structured_error(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate( + _assessment( + MemoryWriteVerdict.HOLD, + code=MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE, + reason="the citation is thin", + ) + ) + async with open_builtin_runtime(_config(tmp_path), memory_write_gate=gate) as runtime: + scope_id = await _create_scope(runtime, "gate-explicit-hold") + with pytest.raises(MemoryWriteRejectedError) as error: + await runtime.memory.for_scope(scope_id).remember( + RememberMemoryRequest(entries=(MemoryEntryInput(kind="note", text="Rejected."),)) + ) + + assert error.value.code == "insufficient_coverage" + assert error.value.reason == "the citation is thin" + + asyncio.run(scenario()) + + +def test_the_ingestion_window_reports_a_hold_and_still_advances(tmp_path: Path) -> None: + async def scenario() -> None: + gate = _ScriptedGate( + _assessment( + MemoryWriteVerdict.HOLD, + code=MemoryWriteRejectionCode.INSUFFICIENT_COVERAGE, + reason="the window evidence is thin", + ) + ) + async with open_builtin_runtime( + _config(tmp_path), + candidate_pipeline=_ContentCandidatePipeline(), + memory_write_gate=gate, + ) as runtime: + scope_id = await _create_scope(runtime, "gate-ingestion-hold") + await runtime.sources.for_scope(scope_id).capture( + CaptureSource(source_id="task-1", content="A durable note.", metadata={}) + ) + + result = await runtime.memory.for_scope(scope_id).flush() + + assert result.held_count == 1 + assert result.hold_codes == ("insufficient_coverage",) + assert result.processed is True + assert result.memory_ref is None + assert gate.requests + assert any("A durable note." in item for item in gate.requests[0].evidence) + + asyncio.run(scenario()) + + +def test_flush_response_preserves_gate_hold_details() -> None: + response = mapping.flush_response( + MemoryFlushResult( + previous_cursor=0, + high_watermark=2, + current_cursor=2, + source_count=1, + memory_ref=None, + held_count=1, + hold_codes=("insufficient_coverage",), + ) + ) + + assert response.held_count == 1 + assert response.hold_codes == ["insufficient_coverage"] + + +def test_memory_write_rejection_maps_to_a_structured_transport_error() -> None: + status_code, code, message, details = _map_error( + MemoryWriteRejectedError("insufficient_coverage", "the citation is thin") + ) + + assert status_code == 422 + assert code == "memory_write_rejected" + assert "rejected" in message + assert details == {"code": "insufficient_coverage", "reason": "the citation is thin"} diff --git a/tests/e2e/test_topic_memory_product_chain.py b/tests/e2e/test_topic_memory_product_chain.py index 77b864bd89..8b4f648b9c 100644 --- a/tests/e2e/test_topic_memory_product_chain.py +++ b/tests/e2e/test_topic_memory_product_chain.py @@ -16,6 +16,7 @@ import argparse import json +import logging import os import subprocess from collections.abc import Mapping @@ -25,6 +26,7 @@ import pytest +from tests.e2e.topic_memory_product import common as common_module from tests.e2e.topic_memory_product import harness from tests.e2e.topic_memory_product.common import ( ArtifactIdentity, @@ -165,6 +167,37 @@ def test_worker_failure_capture_cannot_be_reported_as_pass() -> None: require_no_worker_failures("E1", failures) +def test_worker_failure_capture_records_current_supervisor_event() -> None: + record = logging.LogRecord( + name="powercontext.builtin.runtime.artifact_processing", + level=logging.ERROR, + pathname=__file__, + lineno=0, + msg="Artifact processing failed", + args=(), + exc_info=None, + ) + record.event = "artifact_processing.failed" # type: ignore[attr-defined] + record.stage = "worker" # type: ignore[attr-defined] + record.error_code = "TimeoutError" # type: ignore[attr-defined] + record.exception_type = "TimeoutError" # type: ignore[attr-defined] + record.retry_count = 1 # type: ignore[attr-defined] + + common_capture = common_module._WorkerFailureCapture() + common_capture.handle(record) + harness_capture = harness._WorkerFailureCapture() + harness_capture.handle(record) + + expected = { + "stage": "worker", + "error_code": "TimeoutError", + "exception_type": "TimeoutError", + "failure_count": 1, + } + assert common_capture.failures == [expected] + assert harness_capture.failures == [expected] + + def test_e1_codex_generation_and_plugin_subprocesses_exclude_layer_secrets( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/e2e/topic_memory_product/common.py b/tests/e2e/topic_memory_product/common.py index 99c66d4a1b..8cff034eb9 100644 --- a/tests/e2e/topic_memory_product/common.py +++ b/tests/e2e/topic_memory_product/common.py @@ -81,13 +81,16 @@ def __init__(self) -> None: self.failures: list[dict[str, object]] = [] def emit(self, record: logging.LogRecord) -> None: - if getattr(record, "event", None) != "artifact_processing.worker.failed": + if getattr(record, "event", None) not in { + "artifact_processing.failed", + "artifact_processing.worker.failed", + }: return self.failures.append({ "stage": str(getattr(record, "stage", "unknown")), "error_code": str(getattr(record, "error_code", "unknown")), "exception_type": str(getattr(record, "exception_type", "unknown")), - "failure_count": int(getattr(record, "failure_count", 0)), + "failure_count": int(getattr(record, "failure_count", getattr(record, "retry_count", 0))), }) @@ -641,7 +644,7 @@ def run_e0(directory: Path) -> dict[str, object]: database=SQLiteConfig(url=f"sqlite+aiosqlite:///{runtime_directory / 'runtime.db'}"), runtime=RuntimeConfig( topic_memory_source_window_limit=1, - artifact_processing_worker_timeout_seconds=20, + artifact_processing_worker_timeout_seconds=60, ), inference=InferenceConfig( generation_model="openai-chat:r8-fake-generation", diff --git a/tests/e2e/topic_memory_product/harness.py b/tests/e2e/topic_memory_product/harness.py index e63857cb7c..fb84c8da47 100644 --- a/tests/e2e/topic_memory_product/harness.py +++ b/tests/e2e/topic_memory_product/harness.py @@ -200,13 +200,16 @@ def __init__(self) -> None: self.failures: list[dict[str, object]] = [] def emit(self, record: logging.LogRecord) -> None: - if getattr(record, "event", None) != "artifact_processing.worker.failed": + if getattr(record, "event", None) not in { + "artifact_processing.failed", + "artifact_processing.worker.failed", + }: return self.failures.append({ "stage": str(getattr(record, "stage", "unknown")), "error_code": str(getattr(record, "error_code", "unknown")), "exception_type": str(getattr(record, "exception_type", "unknown")), - "failure_count": int(getattr(record, "failure_count", 0)), + "failure_count": int(getattr(record, "failure_count", getattr(record, "retry_count", 0))), })