diff --git a/docs/en/development/processing-failures.md b/docs/en/development/processing-failures.md index bf3b4ff2e5..55026dc11a 100644 --- a/docs/en/development/processing-failures.md +++ b/docs/en/development/processing-failures.md @@ -55,10 +55,10 @@ in its processing path. |---|---|---|---|---|---|---| | Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Missing | | Topic Memory | Partial | Covered | Partial | Covered | Partial | Missing | -| Experience | Partial | N/A | Partial | Missing | Missing | Missing | -| Profile | Partial | N/A | Partial | Missing | Missing | Missing | -| Skill | Partial | N/A | Partial | Missing | Missing | Missing | -| Handoff | Partial | N/A | Partial | Missing | Missing | Missing | +| Experience | Partial | N/A | Partial | Partial | Missing | Missing | +| Profile | Partial | N/A | Partial | Partial | Missing | Missing | +| Skill | Partial | N/A | Partial | Partial | Missing | Missing | +| Handoff | Partial | N/A | Partial | Partial | Missing | Missing | | Prompt | Partial | N/A | Partial | Missing | Missing | Missing | Memory / Atomic Memory and Topic Memory are tracked at operation granularity while they are the active slices: @@ -86,10 +86,31 @@ Topic Memory operation-level coverage: | Publish / persistence | Persistence | Partial | Projection content and vector-completeness rules are business validation; storage-invariant failures map to `persistence` / `storage_failure`; unsupported capabilities report `configuration`; publisher conflicts keep their domain outcomes, and other storage failures keep unknown. | | Usage accounting | Usage accounting | Missing | The window provider budget is reported as `generation` / `usage_limit`; accounting failures are not classified. | -The first two rollout slices cover Memory / Atomic Memory and Topic Memory: provider failures, -budget exhaustion, and extraction/reconciliation/domain validation are classified, and safe -accounting fields travel from the child Worker to the supervisor log. Other Families keep their -current behavior until their own slice lands; the matrix is updated in the same change as each slice. +Experience, Profile, Skill, and Handoff operation-level coverage: + +| Operation | Stage | Status | Notes | +|---|---|---|---| +| `experience.incubate` | Business validation | Partial | Bounded input and out-of-window evidence carry explicit diagnostics; structural limits do not establish context overflow. | +| `profile.generate` | Business validation | Partial | Invalid Markdown carries the input count and original validation exception type without exposing the content. | +| Experience / Skill typed generation | Output parsing | Partial | Invalid typed results use the shared schema mapping; unclassified provider details remain unknown. | +| Dream (`refine_experience`, `revise_profile`, `derive_skill`, `revise_skill`) | Generation / business validation | Partial | Caught attempt failures emit `dream.attempt.failed` using the existing worker/request identity. Evidence limits, unavailable evidence, and rejected request targets carry `business_validation` and their failure location. | +| `handoff.generate` | Business validation | Partial | Objective, citation, rollover, and byte-limit violations are invalid output, not model resource exhaustion. | +| Publication / persistence | Persistence | Missing | These slices do not yet classify all storage entry points. | +| Usage accounting | Usage accounting | Missing | Accounting failures remain separate from processing outcomes. | + +The matrices describe classification coverage, not completion of the whole issue. Dream failure +records contain bounded diagnostic fields only, never exception text or evidence content. Its +persisted run outcome and retry policy remain authoritative; the failure record is an observation, +not a new durable failure history. Durable-block classification currently covers Topic Memory only; +Dream uses its existing Run outcomes and attempt logs, not a durable scheduler block. + +A timeout from Dream's own execution deadline produces the terminal Run code `budget_exceeded`. +Its attempt log reports `generation` / `usage_limit` with `failure_location=budget_exceeded` and +`exception_type=TimeoutError`. The stage identifies the owner of the execution budget. A bare +`TimeoutError` raised before that deadline expires follows the terminal `generation_failed` path +and logs `timeout` without inferring a stage. +Provider `InferenceTimeoutError` keeps its operation-specific stage and `timeout` reason and remains +retryable within the existing Run budget. ## Rollout diff --git a/docs/zh/development/processing-failures.md b/docs/zh/development/processing-failures.md index ec9bccbdb3..19e7567c55 100644 --- a/docs/zh/development/processing-failures.md +++ b/docs/zh/development/processing-failures.md @@ -49,10 +49,10 @@ Worker/supervisor 值（`worker_process`、`work_admission`、`supervisor`、`le |---|---|---|---|---|---|---| | Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Missing | | Topic Memory | Partial | Covered | Partial | Covered | Partial | Missing | -| Experience | Partial | N/A | Partial | Missing | Missing | Missing | -| Profile | Partial | N/A | Partial | Missing | Missing | Missing | -| Skill | Partial | N/A | Partial | Missing | Missing | Missing | -| Handoff | Partial | N/A | Partial | Missing | Missing | Missing | +| Experience | Partial | N/A | Partial | Partial | Missing | Missing | +| Profile | Partial | N/A | Partial | Partial | Missing | Missing | +| Skill | Partial | N/A | Partial | Partial | Missing | Missing | +| Handoff | Partial | N/A | Partial | Partial | Missing | Missing | | Prompt | Partial | N/A | Partial | Missing | Missing | Missing | Memory / Atomic Memory 与 Topic Memory 作为当前切片，按 operation 粒度跟踪： @@ -80,9 +80,27 @@ Topic Memory 的 operation 级覆盖： | Publish / persistence | Persistence | Partial | 投影内容及向量完整性规则归业务校验；存储不变量失败归 `persistence` / `storage_failure`；不支持的能力归 `configuration`；publisher 冲突保留既有领域结果，其他存储失败保持 unknown。 | | Usage accounting | Usage accounting | Missing | window provider budget 归为 `generation` / `usage_limit`；账务失败未分类。 | -前两个落地切片覆盖 Memory / Atomic Memory 与 Topic Memory：provider 失败、预算耗尽、提取/对账/领域校验 -均有分类，安全计数字段可从子 Worker 传播到 supervisor 日志。其他 Family 在各自切片落地前保持现有行为； -每次切片在同一变更中更新本矩阵。 +Experience、Profile、Skill 与 Handoff 的 operation 级覆盖： + +| Operation | Stage | 状态 | 说明 | +|---|---|---|---| +| `experience.incubate` | Business validation | Partial | 有界输入和窗口外证据携带明确诊断；结构上限不证明模型上下文溢出。 | +| `profile.generate` | Business validation | Partial | 无效 Markdown 携带输入数量与原始校验异常类型，不暴露内容。 | +| Experience / Skill typed generation | Output parsing | Partial | 无效 typed 结果使用共享 schema 映射；无法分类的 provider 细节保持 unknown。 | +| Dream（`refine_experience`、`revise_profile`、`derive_skill`、`revise_skill`） | Generation / business validation | Partial | 捕获的 attempt 失败通过 `dream.attempt.failed` 记录，复用 worker/request 身份。证据超过上限、证据不可用或请求目标不受支持时，记录 `business_validation` 和失败位置。 | +| `handoff.generate` | Business validation | Partial | objective、引用、rollover、字节上限违规均为无效输出，不是模型资源耗尽。 | +| Publication / persistence | Persistence | Missing | 这些切片尚未分类全部存储入口。 | +| Usage accounting | Usage accounting | Missing | 账务失败仍与处理结果分离。 | + +矩阵描述分类覆盖情况，不表示整个 issue 已完成。Dream 失败记录只包含有界诊断字段，不含异常文本或证据正文。 +持久化 run 结果和重试规则仍是权威状态；日志属于观测，不新增持久化失败历史。当前 durable block 分类只覆盖 +Topic Memory；Dream 使用既有 Run 结果和 attempt 日志，不产生持久化调度阻塞。 + +Dream 自身的执行期限耗尽时，Run 以 `budget_exceeded` 终止。对应 attempt 日志记录 +`generation` / `usage_limit` 和 `failure_location=budget_exceeded`，与执行预算结果一致，同时保留 +`exception_type=TimeoutError`。这里的 stage 表示执行预算的归属。期限尚未耗尽时抛出的裸 +`TimeoutError` 走 `generation_failed` 终止路径，日志记录 `timeout`，省略未知阶段。 +Provider 的 `InferenceTimeoutError` 保留操作对应的 stage 和 `timeout` 原因，仍可在既有 Run 预算内重试。 ## 落地顺序 diff --git a/src/powercontext/builtin/artifacts/experience/incubation.py b/src/powercontext/builtin/artifacts/experience/incubation.py index ee7bf4ebdb..48c5abf322 100644 --- a/src/powercontext/builtin/artifacts/experience/incubation.py +++ b/src/powercontext/builtin/artifacts/experience/incubation.py @@ -22,6 +22,12 @@ from powercontext.builtin.artifacts.experience.models import ExperienceContent from powercontext.builtin.inference import GenerationResult, InvalidInferenceOutputError, StructuredGenerator +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.builtin.sources import CONTENT_SOURCE_NAME, ContentSource from powercontext.sources import Source, SourceRef @@ -136,9 +142,17 @@ def _incubation_input( try: return ExperienceIncubationInput(evidence=tuple(projected)), evidence except ValidationError as error: - raise InvalidInferenceOutputError( - "experience-incubate", - "eligible Task Outcome evidence exceeded the bounded input contract", + raise attach_diagnostics( + InvalidInferenceOutputError( + "experience-incubate", + "eligible Task Outcome evidence exceeded the bounded input contract", + ), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + input_item_count=len(projected), + failure_location="experience-incubate-input", + cause_exception_type=type(error).__name__, + ), ) from error @@ -161,9 +175,16 @@ def _selected_sources(values: tuple[str, ...], evidence: dict[str, SourceRef]) - try: source = evidence[evidence_id] except KeyError: - raise InvalidInferenceOutputError( - "experience-incubate", - "candidate cited evidence outside the current Task Outcome window", + raise attach_diagnostics( + InvalidInferenceOutputError( + "experience-incubate", + "candidate cited evidence outside the current Task Outcome window", + ), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="experience-incubate-evidence", + ), ) from None key = (source.source_type, source.source_id) if key not in seen: diff --git a/src/powercontext/builtin/artifacts/handoff/generation.py b/src/powercontext/builtin/artifacts/handoff/generation.py index 8d941151d5..bdb76c593f 100644 --- a/src/powercontext/builtin/artifacts/handoff/generation.py +++ b/src/powercontext/builtin/artifacts/handoff/generation.py @@ -36,6 +36,12 @@ HandoffStatement, ) from powercontext.builtin.inference import GenerationResult, InvalidInferenceOutputError, StructuredGenerator +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.sources import Source HandoffGenerationEvidenceType: TypeAlias = Literal["source", "artifact"] @@ -146,9 +152,17 @@ async def generate(self, request: HandoffGenerationRequest, /) -> HandoffDraft: omissions=tuple(_omission(value, citations) for value in output.omissions), ) except ValidationError as error: - raise InvalidInferenceOutputError( - "handoff-generate", - "generated content violates the Handoff Draft contract", + raise attach_diagnostics( + InvalidInferenceOutputError( + "handoff-generate", + "generated content violates the Handoff Draft contract", + ), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="handoff-generate-contract", + cause_exception_type=type(error).__name__, + ), ) from error @@ -230,7 +244,7 @@ def _statement( citations: Mapping[str, HandoffCitation], ) -> HandoffStatement: if not value.evidence_ids: - raise InvalidInferenceOutputError("handoff-generate", "statement does not cite evidence") + raise _invalid_handoff_output("statement does not cite evidence", "handoff-generation-citation") # noqa: TRY003 return HandoffStatement( text=value.text, citations=tuple(_citation(evidence_id, citations) for evidence_id in dict.fromkeys(value.evidence_ids)), @@ -251,12 +265,22 @@ def _citation(evidence_id: str, citations: Mapping[str, HandoffCitation]) -> Han try: return citations[evidence_id] except KeyError: - raise InvalidInferenceOutputError( - "handoff-generate", - "generated content cites evidence outside the request", + raise _invalid_handoff_output( # noqa: TRY003 + "generated content cites evidence outside the request", "handoff-generation-citation" ) from None +def _invalid_handoff_output(detail: str, location: str) -> InvalidInferenceOutputError: + return attach_diagnostics( + InvalidInferenceOutputError("handoff-generate", detail), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=location, + ), + ) + + def _validated_json(value: object) -> JsonValue: try: return _JSON_VALUE.validate_python(value) diff --git a/src/powercontext/builtin/artifacts/profile/service.py b/src/powercontext/builtin/artifacts/profile/service.py index 2aac6a888c..7ac0270b5f 100644 --- a/src/powercontext/builtin/artifacts/profile/service.py +++ b/src/powercontext/builtin/artifacts/profile/service.py @@ -53,6 +53,12 @@ from powercontext.builtin.persistence.profile import ProfilePolicyRepository from powercontext.builtin.persistence.sources import SourceRepository from powercontext.builtin.persistence.tables import PROFILE_POLICIES_TABLE +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.builtin.records import BaseValueConflictError, BaseValueNotFoundError from powercontext.builtin.scope.errors import ScopeNotFoundError from powercontext.builtin.scope.repository import ScopeRepository @@ -266,7 +272,16 @@ async def _flush( # noqa: C901 except ValueError as error: from powercontext.builtin.inference import InvalidInferenceOutputError - raise InvalidInferenceOutputError("profile-generate", "invalid Markdown") from error + raise attach_diagnostics( + InvalidInferenceOutputError("profile-generate", "invalid Markdown"), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + input_item_count=len(evidence), + failure_location="profile-generate-markdown", + cause_exception_type=type(error).__name__, + ), + ) from error if current is not None and current.content.content == markdown: markdown = None try: diff --git a/src/powercontext/builtin/dream/service.py b/src/powercontext/builtin/dream/service.py index bb9924d6ef..08ead62591 100644 --- a/src/powercontext/builtin/dream/service.py +++ b/src/powercontext/builtin/dream/service.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import logging from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext from datetime import datetime @@ -26,6 +27,7 @@ from sqlalchemy import update from sqlalchemy.ext.asyncio import AsyncConnection +from powercontext._logging import log_safely from powercontext.artifacts import ArtifactRef from powercontext.builtin.artifacts.atomic_memory.models import AtomicMemoryContent from powercontext.builtin.artifacts.experience import ExperienceContent @@ -84,10 +86,12 @@ from powercontext.builtin.persistence.processing_intents import ArtifactProcessingIntentRepository from powercontext.builtin.persistence.profile import ProfilePolicyRepository from powercontext.builtin.persistence.tables import CANDIDATE_HEADS_TABLE, SCOPES_TABLE +from powercontext.builtin.processing_diagnostics import safe_classify_processing_error, safe_log_fields from powercontext.builtin.review.errors import ArtifactTargetConflictError, InvalidCandidateError from powercontext.builtin.review.generation import SkillGenerationOrigin, validate_skill_lineage from powercontext.builtin.review.models import Candidate, CandidateStatus from powercontext.builtin.review.service import ReviewService +from powercontext.builtin.runtime.processing_contracts import processing_error_code from powercontext.builtin.runtime.processing_execution import InvocationAlreadyHandled, ScopeInvocation from powercontext.builtin.tags import TagPreconditionError @@ -97,6 +101,7 @@ ReviewFactory = Callable[[str, AsyncConnection | None], ReviewService] CandidateAttester = Callable[[AsyncConnection, DreamRecord, str, str], Awaitable[None]] CatalogFactory = Callable[[str, AsyncConnection | None], CatalogChangeService] +logger = logging.getLogger(__name__) class DreamService: @@ -293,21 +298,56 @@ async def execute(self) -> bool: # noqa: C901 - fenced admission, fair dispatch if record.run.terminal: await self._complete_invocation(connection) return True + deadline: asyncio.Timeout | None = None try: async with self._transaction() as connection: remaining = _remaining_seconds(record, await database_now(connection)) - async with asyncio.timeout(max(remaining, 0)): + deadline = asyncio.timeout(max(remaining, 0)) + async with deadline: await self._execute(record) except (ArtifactProcessingLeadershipLostError, InvocationAlreadyHandled): raise except (InferenceTimeoutError, InferenceUnavailableError) as error: + self._log_attempt_failure(record, error) await self._fail_or_retry(record, _error_code(error), retry=True) - except TimeoutError: - await self._fail_or_retry(record, "budget_exceeded", retry=False) except Exception as error: - await self._fail_or_retry(record, _error_code(error), retry=False) + if isinstance(error, TimeoutError) and deadline is not None and deadline.expired(): + self._log_attempt_failure(record, error, classification_error=DreamError("budget_exceeded")) + await self._fail_or_retry(record, "budget_exceeded", retry=False) + else: + self._log_attempt_failure(record, error) + await self._fail_or_retry(record, _error_code(error), retry=False) return True + def _log_attempt_failure( + self, + record: DreamRecord, + error: Exception, + *, + classification_error: Exception | None = None, + ) -> None: + work = self.processing + diagnostic_error = error if classification_error is None else classification_error + log_safely( + logger, + logging.WARNING, + "Dream processing attempt failed", + extra={ + "event": "dream.attempt.failed", + "operation": record.run.operation, + "unit": "stage", + "outcome": "failure", + "family": operation_spec(record.run.operation).processing_family, + "scope": record.run.scope_id, + "binding": None if work is None else work.assignment.binding_name, + "worker_id": None if work is None else work.assignment.worker_id, + "request_generation": None if work is None else work.assignment.claimed_request_generation, + "exception_type": type(error).__name__, + "error_code": processing_error_code(type(error)), + **safe_log_fields(safe_classify_processing_error(diagnostic_error)), + }, + ) + async def _complete_invocation(self, connection: AsyncConnection) -> None: if self.processing is None: raise RuntimeError("Dream execution requires a Supervisor invocation") # noqa: TRY003 diff --git a/src/powercontext/builtin/processing_diagnostics.py b/src/powercontext/builtin/processing_diagnostics.py index 1655adeba3..53667b0cf9 100644 --- a/src/powercontext/builtin/processing_diagnostics.py +++ b/src/powercontext/builtin/processing_diagnostics.py @@ -220,6 +220,93 @@ def _topic_memory_diagnostics(code: str) -> ProcessingDiagnostics | None: return None +def _dream_error_diagnostics(code: str) -> ProcessingDiagnostics | None: + """Classify the bounded Dream and Evidence rejection vocabulary.""" + + value = code.lower() + if value == "capability_unavailable": + return ProcessingDiagnostics( + stage=ProcessingStage.GENERATION, + reason=ProcessingFailureReason.CONFIGURATION, + failure_location=value, + ) + if value == "budget_exceeded": + return ProcessingDiagnostics( + stage=ProcessingStage.GENERATION, + reason=ProcessingFailureReason.USAGE_LIMIT, + failure_location=value, + ) + if value == "capacity_exceeded": + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.USAGE_LIMIT, + failure_location=value, + ) + if value == "invalid_generation_output": + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=value, + ) + if value in { + "unsupported_target", + "invalid_target", + "invalid_dream_operation", + "invalid_artifact_family", + "invalid_idempotency_key", + "evidence_unavailable", + "reference_not_found", + "invalid_memory_citation", + "policy_changed", + "evidence_limit_exceeded", + "memory_entry_inactive", + "source_not_eligible", + "scope_not_found", + }: + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + failure_location=value, + ) + if value in {"artifact_conflict", "attempt_conflict", "tag_precondition_failed"}: + return ProcessingDiagnostics(failure_location=value) + if value in {"access_revoked", "access_unavailable", "dream_request_unavailable", "generation_failed"}: + return ProcessingDiagnostics(failure_location=value) + if value in {"idempotency_conflict", "needs_evidence"}: + return ProcessingDiagnostics(failure_location=value) + return None + + +def _handoff_error_diagnostics(code: str) -> ProcessingDiagnostics | None: + """Classify the bounded Handoff generation rejection vocabulary.""" + + value = code.lower() + if value in {"output", "evidence"}: + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=value, + ) + if value == "budget": + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=value, + ) + if value == "objective": + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=value, + ) + if value in {"rollover_next_action", "rollover_quality"}: + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location=value, + ) + return None + + def safe_classify_processing_error(error: BaseException) -> ProcessingDiagnostics | None: """Never let classification itself change a processing outcome.""" @@ -235,7 +322,10 @@ def classify_block_code(stage: str | None, code: str | None) -> ProcessingDiagno if not isinstance(code, str) or not code: return None try: - return _topic_memory_diagnostics(code) if stage == "topic_memory" else None + if stage == "topic_memory": + return _topic_memory_diagnostics(code) + else: + return None except Exception: return None @@ -257,7 +347,7 @@ def stage_for_operation(operation: object) -> ProcessingStage | None: value = operation.lower() if "embed" in value or "index" in value: return ProcessingStage.EMBEDDING - if "extract" in value or "generate" in value or "reconcile" in value or "incubate" in value: + if value == "generation" or any(term in value for term in ("extract", "generate", "reconcile", "incubate")): return ProcessingStage.GENERATION return None @@ -393,6 +483,18 @@ def classify_processing_error(error: BaseException) -> ProcessingDiagnostics | N reason=ProcessingFailureReason.CONFIGURATION, failure_location=capability or None, ) + if "DreamError" in names: + return _dream_error_diagnostics(str(getattr(error, "code", "") or "")) + if "InvalidHandoffGenerationError" in names: + return _handoff_error_diagnostics(str(getattr(error, "code", "") or "")) + if "HandoffGenerationUnavailableError" in names: + return ProcessingDiagnostics( + stage=ProcessingStage.GENERATION, + reason=ProcessingFailureReason.CONFIGURATION, + ) + if "EvidenceResolutionError" in names: + code = str(getattr(error, "code", "") or "").lower() + return _dream_error_diagnostics(code) return None diff --git a/src/powercontext/builtin/runtime/relational.py b/src/powercontext/builtin/runtime/relational.py index 52fb49ddbd..ffb0932214 100644 --- a/src/powercontext/builtin/runtime/relational.py +++ b/src/powercontext/builtin/runtime/relational.py @@ -174,6 +174,12 @@ from powercontext.builtin.persistence.topic_memory_index import NoTopicMemoryIndex, TopicMemoryIndex from powercontext.builtin.persistence.topic_memory_management import TopicMemoryManagementWriter from powercontext.builtin.portability import PortableBundleService +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.builtin.publication import ArtifactPublicationApplication from powercontext.builtin.review.generation import ( GeneratedCandidateResult, @@ -1967,9 +1973,17 @@ def _validate_experience_plans( ) -> None: available = {(row.ref.source_type, row.ref.source_id) for row in rows} if any((source.source_type, source.source_id) not in available for plan in plans for source in plan.sources): - raise InvalidInferenceOutputError( - "experience-incubate", - "pipeline cited a Source outside the current incubation window", + raise attach_diagnostics( + InvalidInferenceOutputError( + "experience-incubate", + "pipeline cited a Source outside the current incubation window", + ), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + input_item_count=len(rows), + failure_location="experience-incubate-evidence", + ), ) diff --git a/tests/builtin/artifacts/handoff/test_generation.py b/tests/builtin/artifacts/handoff/test_generation.py index 6c11460176..1ebfb0e03d 100644 --- a/tests/builtin/artifacts/handoff/test_generation.py +++ b/tests/builtin/artifacts/handoff/test_generation.py @@ -28,6 +28,11 @@ LLMHandoffGenerationPipeline, ) from powercontext.builtin.inference import GenerationResult, InvalidInferenceOutputError +from powercontext.builtin.processing_diagnostics import ( + ProcessingFailureReason, + ProcessingStage, + classify_processing_error, +) from powercontext.builtin.sources import ContentSource from powercontext.sources import SourceRef @@ -115,7 +120,7 @@ async def scenario() -> None: ) ) - with pytest.raises(InvalidInferenceOutputError, match="outside the request"): + with pytest.raises(InvalidInferenceOutputError, match="outside the request") as raised: await pipeline.generate( HandoffGenerationRequest( objective="Complete parser error handling.", @@ -123,6 +128,11 @@ async def scenario() -> None: max_bytes=4096, ) ) + diagnostics = classify_processing_error(raised.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + assert diagnostics.failure_location == "handoff-generation-citation" asyncio.run(scenario()) @@ -143,7 +153,7 @@ async def scenario() -> None: ) ) - with pytest.raises(InvalidInferenceOutputError, match="does not cite evidence"): + with pytest.raises(InvalidInferenceOutputError, match="does not cite evidence") as raised: await pipeline.generate( HandoffGenerationRequest( objective="Complete parser error handling.", @@ -151,5 +161,10 @@ async def scenario() -> None: max_bytes=4096, ) ) + diagnostics = classify_processing_error(raised.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + assert diagnostics.failure_location == "handoff-generation-citation" asyncio.run(scenario()) diff --git a/tests/builtin/artifacts/handoff/test_service.py b/tests/builtin/artifacts/handoff/test_service.py index 14c42542f7..c916ad37df 100644 --- a/tests/builtin/artifacts/handoff/test_service.py +++ b/tests/builtin/artifacts/handoff/test_service.py @@ -39,6 +39,11 @@ PrepareHandoff, PrepareHandoffHint, ) +from powercontext.builtin.processing_diagnostics import ( + ProcessingFailureReason, + ProcessingStage, + classify_processing_error, +) from powercontext.errors import RevisionConflictError from powercontext.sources import Source, SourceMaterialization, SourceRef @@ -380,6 +385,12 @@ async def scenario() -> None: ) assert caught.value.code == code + diagnostics = classify_processing_error(caught.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + assert diagnostics.failure_location == code + assert await service.latest() is None asyncio.run(scenario()) diff --git a/tests/builtin/test_processing_diagnostics.py b/tests/builtin/test_processing_diagnostics.py index 542aedf349..90fe08397a 100644 --- a/tests/builtin/test_processing_diagnostics.py +++ b/tests/builtin/test_processing_diagnostics.py @@ -32,6 +32,16 @@ AtomicMemoryExtractionOutput, AtomicMemoryGenerationPipeline, ) +from powercontext.builtin.artifacts.experience.incubation import ( + ExperienceIncubationCandidate, + ExperienceIncubationOutput, + LLMExperienceCandidatePipeline, +) +from powercontext.builtin.artifacts.experience.models import ExperienceContent +from powercontext.builtin.artifacts.handoff.errors import ( + HandoffGenerationUnavailableError, + InvalidHandoffGenerationError, +) from powercontext.builtin.artifacts.memory.extraction import MemoryExtractionCandidate, _selected_evidence from powercontext.builtin.artifacts.topic_memory.errors import ( TopicMemoryCapabilityError, @@ -39,6 +49,8 @@ TopicMemoryStorageInvariantError, ) from powercontext.builtin.artifacts.topic_memory.generation import TopicMemoryGenerationError +from powercontext.builtin.dream.models import DreamError +from powercontext.builtin.evidence.models import EvidenceResolutionError from powercontext.builtin.inference.errors import ( InferenceTimeoutError, InferenceUnavailableError, @@ -54,6 +66,8 @@ AtomicMemorySearchRequest, RelationalAtomicMemoryIndex, ) +from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile +from powercontext.builtin.persistence.tables import BUILTIN_TABLES from powercontext.builtin.processing_diagnostics import ( ProcessingDiagnostics, ProcessingFailureReason, @@ -65,8 +79,11 @@ ) from powercontext.builtin.runtime.atomic_memory_processing import AtomicMemorySourceWindowProcessor from powercontext.builtin.runtime.processing_contracts import ArtifactProcessingWorkerFailure +from powercontext.builtin.runtime.relational import RelationalContexts +from powercontext.builtin.scope import ScopeDraft +from powercontext.builtin.sources import ContentSource from powercontext.server.logging import JsonFormatter -from powercontext.sources import SourceRef +from powercontext.sources import SourceMaterialization, SourceRef def test_diagnostics_bound_and_sanitize_every_field() -> None: @@ -533,4 +550,193 @@ def test_durable_blocks_classify_only_for_topic_memory() -> None: assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT assert diagnostics.failure_location == "window_attempt_limit" assert classify_block_code("memory", "window_attempt_limit") is None + assert classify_block_code("dream", "budget_exceeded") is None assert classify_block_code("topic_memory", "") is None + + +# -- Dream, Handoff, and Evidence error classification -- + + +@pytest.mark.parametrize( + ("code", "stage", "reason"), + [ + ("capability_unavailable", ProcessingStage.GENERATION, ProcessingFailureReason.CONFIGURATION), + ("unsupported_target", ProcessingStage.BUSINESS_VALIDATION, None), + ("budget_exceeded", ProcessingStage.GENERATION, ProcessingFailureReason.USAGE_LIMIT), + ("capacity_exceeded", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.USAGE_LIMIT), + ("invalid_generation_output", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("invalid_target", ProcessingStage.BUSINESS_VALIDATION, None), + ("invalid_dream_operation", ProcessingStage.BUSINESS_VALIDATION, None), + ("invalid_artifact_family", ProcessingStage.BUSINESS_VALIDATION, None), + ("invalid_idempotency_key", ProcessingStage.BUSINESS_VALIDATION, None), + ("evidence_unavailable", ProcessingStage.BUSINESS_VALIDATION, None), + ("reference_not_found", ProcessingStage.BUSINESS_VALIDATION, None), + ("invalid_memory_citation", ProcessingStage.BUSINESS_VALIDATION, None), + ("policy_changed", ProcessingStage.BUSINESS_VALIDATION, None), + ("artifact_conflict", None, None), + ("attempt_conflict", None, None), + ("tag_precondition_failed", None, None), + ("scope_not_found", ProcessingStage.BUSINESS_VALIDATION, None), + ("evidence_limit_exceeded", ProcessingStage.BUSINESS_VALIDATION, None), + ("memory_entry_inactive", ProcessingStage.BUSINESS_VALIDATION, None), + ("source_not_eligible", ProcessingStage.BUSINESS_VALIDATION, None), + ("access_revoked", None, None), + ("access_unavailable", None, None), + ("dream_request_unavailable", None, None), + ("generation_failed", None, None), + ("idempotency_conflict", None, None), + ("needs_evidence", None, None), + ], +) +def test_dream_errors_classify_by_code(code, stage, reason) -> None: + diagnostics = classify_processing_error(DreamError(code)) + + assert diagnostics is not None + assert diagnostics.stage is stage + assert diagnostics.reason is reason + assert diagnostics.failure_location == code + + +def test_unknown_dream_code_keeps_unknown_classification() -> None: + assert classify_processing_error(DreamError("future_dream_code")) is None + + +@pytest.mark.parametrize( + ("code", "stage", "reason"), + [ + ("output", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("evidence", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("budget", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("objective", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("rollover_next_action", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("rollover_quality", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ], +) +def test_handoff_generation_errors_classify_by_code(code, stage, reason) -> None: + diagnostics = classify_processing_error(InvalidHandoffGenerationError(code)) + + assert diagnostics is not None + assert diagnostics.stage is stage + assert diagnostics.reason is reason + assert diagnostics.failure_location == code + + +def test_unknown_handoff_code_keeps_unknown_classification() -> None: + assert classify_processing_error(InvalidHandoffGenerationError("future_handoff_code")) is None + + +def test_handoff_generation_unavailable_classifies_as_configuration() -> None: + diagnostics = classify_processing_error(HandoffGenerationUnavailableError()) + + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.CONFIGURATION + + +def test_evidence_resolution_error_classifies_by_code() -> None: + diagnostics = classify_processing_error(EvidenceResolutionError("reference_not_found")) + + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is None + assert diagnostics.failure_location == "reference_not_found" + + +@pytest.mark.parametrize( + "code", + [ + "invalid_generation_output", + "needs_evidence", + "evidence_limit_exceeded", + "reference_not_found", + "evidence_unavailable", + "memory_entry_inactive", + "source_not_eligible", + ], +) +def test_evidence_resolution_error_codes_classify(code) -> None: + diagnostics = classify_processing_error(EvidenceResolutionError(code)) + + assert diagnostics is not None + assert diagnostics.failure_location == code + + +def test_dream_block_stage_is_not_supported() -> None: + assert classify_block_code("dream", "budget_exceeded") is None + assert classify_block_code("dream", "unknown_future_code") is None + + +def test_unknown_evidence_resolution_error_code_is_unclassified() -> None: + assert classify_processing_error(EvidenceResolutionError("completely_unknown")) is None + + +@pytest.mark.parametrize("outside_window", [False, True]) +def test_experience_incubation_validates_typed_output_before_citations(outside_window) -> None: + class Generator: + async def generate(self, value): + if not outside_window: + return GenerationResult(output={"candidates": "private-invalid-output"}) + return GenerationResult( + output=ExperienceIncubationOutput( + candidates=( + ExperienceIncubationCandidate( + proposal=ExperienceContent( + situation="A write timed out.", + action="Replayed the same request.", + outcome="One row remained.", + lesson="Reuse the idempotency key.", + ), + evidence_ids=("private-outside-window",), + ), + ) + ) + ) + + pipeline = LLMExperienceCandidatePipeline(cast(Any, Generator())) + source = ContentSource( + name="task-1", + materialization=SourceMaterialization.CAPTURED, + content="private-source", + metadata={"kind": "task-outcome"}, + ) + with pytest.raises(InvalidInferenceOutputError) as raised: + asyncio.run(pipeline.incubate((source,))) + diagnostics = classify_processing_error(raised.value) + assert diagnostics is not None + assert diagnostics.stage is ( + ProcessingStage.BUSINESS_VALIDATION if outside_window else ProcessingStage.OUTPUT_PARSING + ) + assert diagnostics.reason is ( + ProcessingFailureReason.INVALID_OUTPUT if outside_window else ProcessingFailureReason.SCHEMA_INVALID + ) + assert "private" not in json.dumps(diagnostics.log_fields()) + + +def test_profile_markdown_rejection_preserves_source_cursor_and_publication() -> None: + class Generator: + async def generate(self, value): + return " " + + async def scenario(): + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as profile: + contexts = RelationalContexts(database=profile.database) + scope = await contexts.scopes.create( + ScopeDraft(title="Profile validation", summary="Safe output", idempotency_key="profile-validation") + ) + await contexts.records.create_source(scope.scope_id, "content", "private-source") + await contexts.profiles.put_policy(scope.scope_id, generation_enabled=True, expected_version=0) + contexts.profiles.generator = Generator() + with pytest.raises(InvalidInferenceOutputError) as raised: + await contexts.profiles.flush(scope.scope_id) + diagnostics = classify_processing_error(raised.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + assert diagnostics.input_item_count == 1 + assert diagnostics.cause_exception_type == "ValueError" + assert "private" not in json.dumps(diagnostics.log_fields()) + async with profile.database.transaction() as connection: + assert await contexts.profiles.cursors.load(connection, scope.scope_id, "profile-source-window") is None + assert await contexts.profiles.latest(connection, scope.scope_id) is None + + asyncio.run(scenario()) diff --git a/tests/e2e/test_artifact_dreaming.py b/tests/e2e/test_artifact_dreaming.py index 483ae3d768..c9bef9abc1 100644 --- a/tests/e2e/test_artifact_dreaming.py +++ b/tests/e2e/test_artifact_dreaming.py @@ -17,6 +17,8 @@ from __future__ import annotations import asyncio +import json +import logging import os from collections.abc import Iterator from contextlib import nullcontext, suppress @@ -61,6 +63,7 @@ from powercontext.builtin.inference.models import GenerationResult, InferenceUsage from powercontext.builtin.persistence.oceanbase import OceanBaseConfig, OceanBaseProfile from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.processing_diagnostics import ProcessingStage, classify_processing_error from powercontext.builtin.records import ArtifactWrite, BaseValueConflictError from powercontext.builtin.runtime import ( ApproveCandidateRequest, @@ -1611,14 +1614,20 @@ async def scenario() -> None: asyncio.run(scenario()) -def test_transient_failure_retries_same_projection_with_bounded_calls(database: DatabaseConfig) -> None: - from powercontext.builtin.inference.errors import InferenceUnavailableError +@pytest.mark.parametrize("diagnostic_failure", [None, "classification", "logging"]) +@pytest.mark.parametrize("provider_failure", ["unavailable", "timeout"]) +def test_transient_failure_retries_same_projection_with_bounded_calls( + database: DatabaseConfig, caplog, monkeypatch, diagnostic_failure, provider_failure +) -> None: + from powercontext.builtin.inference.errors import InferenceTimeoutError, InferenceUnavailableError class TransientGenerator(Generator): async def generate(self, value): result = await super().generate(value) if len(self.inputs) == 1: - raise InferenceUnavailableError("generation") + if provider_failure == "timeout": + raise InferenceTimeoutError("generation", 1) + raise InferenceUnavailableError("generation", "private-provider-response") return result async def scenario() -> None: @@ -1630,6 +1639,16 @@ async def scenario() -> None: accepted = await runtime.dream.for_scope(scope).create( CreateDreamRunRequest(operation="refine_experience", artifacts=(citation,), idempotency_key="retry") ) + + def reject_observation(*args, **kwargs): + raise RuntimeError("private-diagnostic-failure") + + if diagnostic_failure == "classification": + monkeypatch.setattr( + "powercontext.builtin.processing_diagnostics.classify_processing_error", reject_observation + ) + elif diagnostic_failure == "logging": + monkeypatch.setattr("powercontext.builtin.dream.service.logger.log", reject_observation) await process_pending(runtime) first = await runtime.dream.for_scope(scope).get(GetDreamRunRequest(run_id=accepted.run_id)) assert first.status == "queued" @@ -1641,8 +1660,34 @@ async def scenario() -> None: assert complete.input_manifest == first.input_manifest assert generator.inputs[0] == generator.inputs[1] assert complete.usage.input_tokens is None # The failed call's token usage is unknown. + records = [record for record in caplog.records if getattr(record, "event", None) == "dream.attempt.failed"] + if diagnostic_failure == "logging": + assert records == [] + return + assert len(records) == 1 + record = records[0] + assert record.operation == "refine_experience" + assert record.family == "experience" + assert record.worker_id + assert record.request_generation >= 1 + from powercontext.server.logging import JsonFormatter, _HumanContextFilter + + payload = JsonFormatter().format(record) + if diagnostic_failure == "classification": + assert "processing_stage" not in json.loads(payload) + assert "private-diagnostic-failure" not in payload + return + assert record.processing_stage == "generation" + expected_reason = "timeout" if provider_failure == "timeout" else "provider_unavailable" + assert record.failure_reason == expected_reason + assert json.loads(payload)["failure_reason"] == expected_reason + _HumanContextFilter().filter(record) + assert f'failure_reason="{expected_reason}"' in record.processing_context + assert "private-provider-response" not in payload + assert "private-provider-response" not in caplog.text - asyncio.run(scenario()) + with caplog.at_level(logging.WARNING, logger="powercontext.builtin.dream.service"): + asyncio.run(scenario()) def test_explicit_dream_preserves_incubation_cursor_and_pending_source_work(database: DatabaseConfig) -> None: @@ -1691,6 +1736,111 @@ async def scenario() -> None: asyncio.run(scenario()) +@pytest.mark.parametrize("diagnostic_failure", [None, "classification", "logging"]) +def test_execution_deadline_matches_terminal_budget_diagnostics( + database: DatabaseConfig, caplog, monkeypatch, diagnostic_failure +) -> None: + from types import SimpleNamespace + + async def scenario() -> None: + deadline: asyncio.Timeout | None = None + + def execution_timeout(seconds): + nonlocal deadline + deadline = asyncio.timeout(seconds) + return deadline + + class DeadlineGenerator(Generator): + async def generate(self, value): + assert deadline is not None + deadline.reschedule(asyncio.get_running_loop().time()) + return await super().generate(value) + + monkeypatch.setattr("powercontext.builtin.dream.service.asyncio", SimpleNamespace(timeout=execution_timeout)) + generator = DeadlineGenerator(blocked=True) + async with open_builtin_runtime( + config(database), candidate_pipeline=atomic_memory_pipeline(MemoryPipeline()), dream_generator=generator + ) as runtime: + scope, _, citation = await seed(runtime) + request = CreateDreamRunRequest( + operation="refine_experience", artifacts=(citation,), idempotency_key="deadline" + ) + accepted = await runtime.dream.for_scope(scope).create(request) + + def reject_observation(*args, **kwargs): + raise RuntimeError("private-diagnostic-failure") + + if diagnostic_failure == "classification": + monkeypatch.setattr( + "powercontext.builtin.processing_diagnostics.classify_processing_error", reject_observation + ) + elif diagnostic_failure == "logging": + monkeypatch.setattr("powercontext.builtin.dream.service.logger.log", reject_observation) + await process_pending(runtime) + run = await runtime.dream.for_scope(scope).get(GetDreamRunRequest(run_id=accepted.run_id)) + assert (run.status, run.error, run.candidate, run.attempt_count) == ("failed", "budget_exceeded", None, 1) + assert len(generator.inputs) == 1 + assert (await runtime.review.for_scope(scope).list(ListCandidatesRequest())).candidates == () + assert (await runtime.dream.for_scope(scope).create(request)).run_id == run.run_id + await process_pending(runtime) + assert len(generator.inputs) == 1 + records = [record for record in caplog.records if getattr(record, "event", None) == "dream.attempt.failed"] + if diagnostic_failure == "logging": + assert records == [] + return + assert len(records) == 1 + record = records[0] + assert record.exception_type == "TimeoutError" + assert record.operation == "refine_experience" and record.family == "experience" + assert record.worker_id and record.request_generation >= 1 + from powercontext.server.logging import JsonFormatter, _HumanContextFilter + + payload = JsonFormatter().format(record) + assert "private" not in payload + if diagnostic_failure == "classification": + assert "processing_stage" not in json.loads(payload) + return + assert record.processing_stage == "generation" + assert record.failure_reason == "usage_limit" + assert record.failure_location == run.error + assert json.loads(payload)["failure_location"] == "budget_exceeded" + _HumanContextFilter().filter(record) + assert 'failure_reason="usage_limit"' in record.processing_context + + with caplog.at_level(logging.WARNING, logger="powercontext.builtin.dream.service"): + asyncio.run(scenario()) + + +def test_dependency_timeout_preserves_timeout_diagnostics(database: DatabaseConfig, caplog) -> None: + class TimedOutGenerator(Generator): + async def generate(self, value): + await super().generate(value) + raise TimeoutError("private-dependency-timeout") + + async def scenario() -> None: + generator = TimedOutGenerator() + async with open_builtin_runtime( + config(database), candidate_pipeline=atomic_memory_pipeline(MemoryPipeline()), dream_generator=generator + ) as runtime: + scope, _, citation = await seed(runtime) + accepted = await runtime.dream.for_scope(scope).create( + CreateDreamRunRequest( + operation="refine_experience", artifacts=(citation,), idempotency_key="dependency-timeout" + ) + ) + await process_pending(runtime) + run = await runtime.dream.for_scope(scope).get(GetDreamRunRequest(run_id=accepted.run_id)) + assert (run.status, run.error, run.candidate, run.attempt_count) == ("failed", "generation_failed", None, 1) + (record,) = [entry for entry in caplog.records if getattr(entry, "event", None) == "dream.attempt.failed"] + assert record.exception_type == "TimeoutError" + assert record.failure_reason == "timeout" + assert not hasattr(record, "processing_stage") + assert "private-dependency-timeout" not in caplog.text + + with caplog.at_level(logging.WARNING, logger="powercontext.builtin.dream.service"): + asyncio.run(scenario()) + + def test_provider_exceeding_output_budget_leaves_no_candidate(database: DatabaseConfig) -> None: class OversizedGenerator(Generator): async def generate(self, value): @@ -2809,10 +2959,15 @@ async def scenario() -> None: stored = await runtime.experience.for_scope(scope).get(GetExperienceRequest(artifact=target)) assert stored.lineage.sources == (source.source_ref,) assert stored.lineage.artifacts == (*refs, *(citations if revision == 1 else ())) - with pytest.raises(EvidenceResolutionError, match="evidence_limit_exceeded"): + with pytest.raises(EvidenceResolutionError, match="evidence_limit_exceeded") as limited: await runtime.dream.for_scope(scope).create( CreateDreamRunRequest(operation="derive_skill", artifacts=(target,), idempotency_key="bounded") ) + diagnostics = classify_processing_error(limited.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is None + assert diagnostics.failure_location == "evidence_limit_exceeded" if citations: candidate = await runtime.skill.for_scope(scope).propose( ProposeSkillRequest( @@ -2826,10 +2981,15 @@ async def scenario() -> None: ) ) await forget_memory(runtime, scope, citations[0]) - with pytest.raises(EvidenceResolutionError, match="memory_entry_inactive"): + with pytest.raises(EvidenceResolutionError, match="memory_entry_inactive") as inactive: await runtime.review.for_scope(scope).approve( ApproveCandidateRequest(candidate_id=candidate.candidate_id, expected_version=candidate.version) ) + diagnostics = classify_processing_error(inactive.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is None + assert diagnostics.failure_location == "memory_entry_inactive" pending = await runtime.review.for_scope(scope).get( GetCandidateRequest(candidate_id=candidate.candidate_id) )