diff --git a/docs/en/development/meta.json b/docs/en/development/meta.json index 39bb09b4cd..49a8f9f72a 100644 --- a/docs/en/development/meta.json +++ b/docs/en/development/meta.json @@ -9,6 +9,7 @@ "plugin-contract", "pydantic-ai-inference", "observability", + "processing-failures", "remote-access-implementation", "dashboard", "desktop-delivery-plan" diff --git a/docs/en/development/observability.md b/docs/en/development/observability.md index 869c242fcb..35ab05f339 100644 --- a/docs/en/development/observability.md +++ b/docs/en/development/observability.md @@ -74,6 +74,9 @@ JSON formatting deliberately ignores unknown extras. When a new field is require 4. Update the operational field contract in both locales and add a focused observable-output regression test. Review compatibility before renaming or removing published fields or event names. +See [Processing failure diagnostics](processing-failures.md) for the shared failure vocabulary and the +per-Family coverage matrix. + Service identity comes from `src/powercontext/server/telemetry.py`; logs and traces use the same resource identity. Error classification comes from `powercontext._logging.error_category`, with domain-owned categories such as `blocked` set at their owning boundary. Do not create independent vocabularies in each adapter. diff --git a/docs/en/development/processing-failures.md b/docs/en/development/processing-failures.md new file mode 100644 index 0000000000..f594b26b54 --- /dev/null +++ b/docs/en/development/processing-failures.md @@ -0,0 +1,87 @@ +# Processing failure diagnostics + +This page is the shared contract for classifying artifact-processing failures and for the safe +diagnostics attached to them. It extends the existing operational logging fields; it does not +introduce a second telemetry system. + +## Shared fields + +Every processing failure keeps the existing identity and outcome fields (`family`, `operation`, +`worker_id`, `request_generation`, `outcome`, `error_code`, `exception_type`) and may add: + +| Field | Meaning | Values | +|---|---|---| +| `processing_stage` | The stage that owned the failure | `generation`, `embedding`, `output_parsing`, `business_validation`, `persistence`, `usage_accounting` (reserved for usage reporting or persistence) | +| `failure_reason` | A bounded reason, only when evidence supports it | `context_overflow`, `output_truncated`, `rate_limit`, `timeout`, `schema_invalid`, `provider_unavailable`, `configuration`, `invalid_output`, `storage_failure`, `usage_limit` | +| `input_item_count` | Number of input items in the failing call | Non-negative integer | +| `estimated_input_tokens` | Estimated input tokens | Non-negative integer | +| `token_estimator_id` | Identity of the estimator that produced the estimate | Bounded string | +| `effective_budget` | Budget applied to the call | Non-negative integer | +| `provider_request_id` | Provider request identity, when reported | Bounded string | +| `finish_reason` | Provider finish reason, when reported | Bounded string | +| `failure_location` | Stable internal location or operation identifier | Bounded string | +| `cause_exception_type` | Original provider or library exception type before mapping | Bounded string | + +`processing_stage` is distinct from the existing `stage` field. `stage` names the emitting boundary or +domain location: Worker and supervisor values (`worker_process`, `work_admission`, `supervisor`, +`lease_renewal`, `scope_discovery`, `pending_observation`, `term_revocation`, `worker_termination`), +domain block names such as `topic_memory`, and the child Worker's `worker` fallback when an exception +carries no stage. A child failure without a domain block may omit `stage` entirely. `processing_stage` +is the bounded domain stage. Counts and identifiers are bounded before they reach a log record, and +unknown or unsupported values are omitted rather than guessed. + +## Privacy boundary + +Diagnostics never carry Source or Artifact bodies, prompts, model responses, credentials, +authorization headers, SQL parameters, or arbitrary exception text. Provider identifiers and reasons +are length-limited printable text. A field being allowlisted is not a redaction engine: values must +still be safe at the call site. + +## Coverage matrix + +`Covered` means the stage has a stable classification and a test. `Partial` means some failures are +classified through the inferred operation or exception type but not every raise site is tagged. +`Missing` means the stage is not classified yet. `N/A` means the Family does not perform that stage +in its processing path. + +| Family | Generation | Embedding | Output parsing | Business validation | Persistence | Usage accounting | +|---|---|---|---|---|---|---| +| Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Missing | +| Topic Memory | Partial | Partial | Partial | Missing | Missing | 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 | +| Prompt | Partial | N/A | Partial | Missing | Missing | Missing | + +Memory / Atomic Memory is tracked at operation granularity while it is the active slice: + +| Operation | Stage | Status | Notes | +|---|---|---|---| +| `atomic-memory.extract` | Generation | Partial | Provider timeout, unavailability, rate limit, and configuration are classified; a provider finish reason is only visible where the adapter exposes it. | +| `atomic-memory.extract` | Output parsing | Partial | Schema and Pydantic output failures are classified; some provider truncation paths surface as schema failures. | +| `atomic-memory.extract` | Business validation | Covered | Candidate content and cited-evidence rules carry `business_validation` / `invalid_output`. | +| `atomic-memory.reconcile` | Generation | Partial | Same provider mapping as extraction. | +| `atomic-memory.reconcile` | Output parsing | Partial | Same schema mapping as extraction. | +| `atomic-memory.reconcile` | Business validation | Covered | Reconciliation output and content rules carry `business_validation` / `invalid_output`. | +| Embedding (extract, reconcile, query) | Embedding | Covered | Provider status, wrong vector count/order/type, and embedding profile/input failures are classified. | +| Publish / index | Persistence | Partial | `AtomicMemoryIndexError` codes are mapped; other storage failures keep unknown. | +| Usage accounting | Usage accounting | Missing | Usage-reporting and persistence failures are not classified; inference usage limits are reported in their originating stage (generation, or embedding when an adapter reports the limit). | + +The first rollout slice covers Memory / Atomic Memory: provider failures, budget exhaustion, and +extraction/reconciliation 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. + +## Rollout + +1. Shared diagnostics contract, allowlisted fields, Worker propagation, and the Memory / Atomic + Memory slice (this change). +2. Topic Memory: tag its staged generation, index/embedding, and persistence failures. +3. Experience, Profile, and Skill: tag generation output validation and persistence entry points. +4. Handoff and Prompt: tag generation and publication failures, then add usage-accounting + classification and complete the matrix. + +For new instrumentation, cover the observable behavior: successful classification, propagation +across a real spawned Worker, sensitive-value exclusion, and that a diagnostics failure does not +change the business outcome. diff --git a/docs/en/docs/operate/observability.md b/docs/en/docs/operate/observability.md index fdb5100512..9f226378b6 100644 --- a/docs/en/docs/operate/observability.md +++ b/docs/en/docs/operate/observability.md @@ -34,6 +34,7 @@ including code stages. Code-specific diagnostic fields are included only for `co | `request_id`, `trace_id`, `span_id` | Bound ingress request and current trace context, when available | | `worker_id`, `family`, `binding`, `request_generation`, `supervisor_group`, `supervisor_generation` | One worker attempt and its accepted work/fence; correlate child output with supervisor diagnostics | | `duration_ms`, `error_code`, `error_category` | Duration in milliseconds, diagnostic code, and bounded operational error classification | +| `stage`, `processing_stage`, `failure_reason`, `input_item_count`, `estimated_input_tokens`, `token_estimator_id`, `effective_budget`, `provider_request_id`, `finish_reason`, `failure_location`, `cause_exception_type` | Processing failure stage/reason and safe accounting fields; `stage` names the emitting boundary or domain location (Worker/supervisor stage, domain block name, or `worker` fallback), `processing_stage` the bounded domain stage. See [Processing failure diagnostics](../../development/processing-failures.md) | Server roles are `all` (API and supervisor), `api`, and `background`; child processes use `worker`. Explicit logging configuration without a deployment role uses `server`. JSON identity fields correspond to trace resource attributes diff --git a/docs/zh/development/meta.json b/docs/zh/development/meta.json index 26d69af5ee..3bedd65ef2 100644 --- a/docs/zh/development/meta.json +++ b/docs/zh/development/meta.json @@ -9,6 +9,7 @@ "plugin-contract", "pydantic-ai-inference", "observability", + "processing-failures", "remote-access-implementation", "dashboard", "desktop-delivery-plan" diff --git a/docs/zh/development/observability.md b/docs/zh/development/observability.md index 8a6e401917..60319ff712 100644 --- a/docs/zh/development/observability.md +++ b/docs/zh/development/observability.md @@ -75,6 +75,8 @@ JSON formatter 会忽略未登记的 extras。确需新增字段时： 4. 更新中英文 operational field contract，并补充针对可观察输出的 regression test。 重命名或删除已发布字段、event name 前，必须检查 compatibility。 +共享失败词汇与各 Family 覆盖矩阵见 [处理失败诊断](processing-failures.md)。 + Service identity 来自 `src/powercontext/server/telemetry.py`，logs 和 traces 共用同一 resource identity。 Error classification 来自 `powercontext._logging.error_category`；`blocked` 等 domain-owned category 由所属 boundary 明确设置。各 adapter 不应自行定义不同词汇。 diff --git a/docs/zh/development/processing-failures.md b/docs/zh/development/processing-failures.md new file mode 100644 index 0000000000..eb419678e6 --- /dev/null +++ b/docs/zh/development/processing-failures.md @@ -0,0 +1,77 @@ +# 处理失败诊断 + +本文是 Artifact 处理失败分类与安全诊断的共享契约。它扩展现有的可观测性日志字段，不引入第二套 +telemetry 系统。 + +## 共享字段 + +每次处理失败保留现有的身份与结果字段（`family`、`operation`、`worker_id`、`request_generation`、 +`outcome`、`error_code`、`exception_type`），并可按需增加： + +| 字段 | 含义 | 取值 | +|---|---|---| +| `processing_stage` | 失败所属的处理阶段 | `generation`、`embedding`、`output_parsing`、`business_validation`、`persistence`、`usage_accounting`（预留给账务上报或持久化） | +| `failure_reason` | 有证据支持时才给出的有界原因 | `context_overflow`、`output_truncated`、`rate_limit`、`timeout`、`schema_invalid`、`provider_unavailable`、`configuration`、`invalid_output`、`storage_failure`、`usage_limit` | +| `input_item_count` | 失败调用涉及的输入条目数 | 非负整数 | +| `estimated_input_tokens` | 估算的输入 token 数 | 非负整数 | +| `token_estimator_id` | 产生估算值的估算器标识 | 有界字符串 | +| `effective_budget` | 本次调用生效的预算 | 非负整数 | +| `provider_request_id` | provider 返回的请求标识 | 有界字符串 | +| `finish_reason` | provider 返回的结束原因 | 有界字符串 | +| `failure_location` | 稳定的内部位置或操作标识 | 有界字符串 | +| `cause_exception_type` | 映射前的原始 provider/库异常类型 | 有界字符串 | + +`processing_stage` 与既有的 `stage` 字段不同：`stage` 表示发出失败的位置或领域位置，包括 +Worker/supervisor 值（`worker_process`、`work_admission`、`supervisor`、`lease_renewal`、`scope_discovery`、 +`pending_observation`、`term_revocation`、`worker_termination`）、领域 block 名称（如 `topic_memory`）， +以及异常未携带 stage 时 child Worker 的 `worker` 兜底；没有领域 block 的 child 失败可能完全省略 `stage`。 +`processing_stage` 表示有界的领域阶段。计数与标识在写入日志前会被限界；未知或不受支持的值直接省略， +不做猜测。 + +## 隐私边界 + +诊断不得包含 Source/Artifact 正文、prompt、模型响应、凭据、授权头、SQL 参数或任意异常文本。provider +标识与原因只保留有长度上限的可打印文本。字段进入白名单不等于自动脱敏，调用点仍必须保证值本身安全。 + +## 覆盖矩阵 + +`Covered` 表示该阶段已有稳定分类与测试；`Partial` 表示部分失败可经 operation 或异常类型推断，但尚未覆盖 +所有抛出点；`Missing` 表示尚未分类；`N/A` 表示该 Family 的处理路径不执行该阶段。 + +| Family | Generation | Embedding | Output parsing | Business validation | Persistence | Usage accounting | +|---|---|---|---|---|---|---| +| Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Missing | +| Topic Memory | Partial | Partial | Partial | Missing | Missing | 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 | +| Prompt | Partial | N/A | Partial | Missing | Missing | Missing | + +Memory / Atomic Memory 作为当前切片，按 operation 粒度跟踪： + +| Operation | Stage | 状态 | 说明 | +|---|---|---|---| +| `atomic-memory.extract` | Generation | Partial | provider timeout、不可用、限流、配置错误已分类；仅 adapter 暴露 finish reason 时可见。 | +| `atomic-memory.extract` | Output parsing | Partial | schema 与 Pydantic 输出失败已分类；部分 provider 截断仍表现为 schema 失败。 | +| `atomic-memory.extract` | Business validation | Covered | 候选内容与引用证据规则输出 `business_validation` / `invalid_output`。 | +| `atomic-memory.reconcile` | Generation | Partial | 与提取相同的 provider 映射。 | +| `atomic-memory.reconcile` | Output parsing | Partial | 与提取相同的 schema 映射。 | +| `atomic-memory.reconcile` | Business validation | Covered | 对账输出与内容规则输出 `business_validation` / `invalid_output`。 | +| Embedding（提取、对账、查询） | Embedding | Covered | provider 状态、向量数量/顺序/类型错误、profile/input 失败均已分类。 | +| Publish / index | Persistence | Partial | `AtomicMemoryIndexError` 的 code 已映射；其他存储失败保持 unknown。 | +| Usage accounting | Usage accounting | Missing | 尚未分类账务上报与持久化失败；推理用量限制归入其来源阶段（generation，或 embedding adapter 上报时归入 embedding）。 | + +第一个落地切片覆盖 Memory / Atomic Memory：provider 失败、预算耗尽、提取/对账校验均有分类，安全计数字段 +可从子 Worker 传播到 supervisor 日志。其他 Family 在各自切片落地前保持现有行为；每次切片在同一变更中更新 +本矩阵。 + +## 落地顺序 + +1. 共享诊断契约、日志白名单字段、Worker 传播，以及 Memory / Atomic Memory 切片（本次变更）。 +2. Topic Memory：标注分阶段生成、索引/嵌入与持久化失败。 +3. Experience、Profile、Skill：标注生成输出校验与持久化入口。 +4. Handoff、Prompt：标注生成与发布失败，再补 usage accounting 分类并补全矩阵。 + +新增 instrumentation 时，需要覆盖可观察行为：分类正确、跨真实 spawn Worker 传播、敏感值不出现，以及诊断 +失败不改变业务结果。 diff --git a/docs/zh/docs/operate/observability.md b/docs/zh/docs/operate/observability.md index 05f5ff664a..5d0f4a95cd 100644 --- a/docs/zh/docs/operate/observability.md +++ b/docs/zh/docs/operate/observability.md @@ -32,6 +32,7 @@ Code 专用诊断字段仅在 `code.*` events 中输出。 | `request_id`、`trace_id`、`span_id` | 已绑定的 ingress request 和当前 trace context；没有上下文时省略 | | `worker_id`、`family`、`binding`、`request_generation`、`supervisor_group`、`supervisor_generation` | 一次 worker attempt 及其已接受的工作和 fence，用于关联 child 与 supervisor logs | | `duration_ms`、`error_code`、`error_category` | 毫秒耗时、诊断 code 和有限取值的 operational error category | +| `stage`、`processing_stage`、`failure_reason`、`input_item_count`、`estimated_input_tokens`、`token_estimator_id`、`effective_budget`、`provider_request_id`、`finish_reason`、`failure_location`、`cause_exception_type` | 处理失败 stage/reason 与安全计数字段；`stage` 表示发出失败的位置或领域位置（Worker/supervisor stage、领域 block 名称或 `worker` 兜底），`processing_stage` 表示有界的领域阶段。详见 [处理失败诊断](../../development/processing-failures.md) | Server role 为 `all`（API 与 supervisor）、`api` 或 `background`，child process 使用 `worker`。 显式配置 logging 但没有指定 deployment role 时使用 `server`。JSON identity 字段对应 trace resource attributes： diff --git a/src/powercontext/builtin/artifacts/atomic_memory/extraction.py b/src/powercontext/builtin/artifacts/atomic_memory/extraction.py index 9d0e694b8f..14f9960c54 100644 --- a/src/powercontext/builtin/artifacts/atomic_memory/extraction.py +++ b/src/powercontext/builtin/artifacts/atomic_memory/extraction.py @@ -19,7 +19,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal -from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter +from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError from powercontext.artifacts import ArtifactRef from powercontext.builtin.artifacts.atomic_memory.models import AtomicMemoryContent @@ -27,6 +27,12 @@ from powercontext.builtin.inference import InvalidInferenceOutputError, StructuredGenerator, TokenEstimator from powercontext.builtin.persistence.records import _canonical_source_text from powercontext.builtin.persistence.sources import SourceRepository, StoredSource +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.sources import TEXT_EVIDENCE_PROJECTION_KEY, Source, SourceObservation, SourceRef, TextEvidence _JSON = TypeAdapter(JsonValue) @@ -120,13 +126,38 @@ class AtomicMemoryGenerationPipeline: async def extract(self, value: AtomicMemoryExtractionInput) -> tuple[AtomicMemoryCandidate, ...]: result = await self.extractor.generate(value) - output = AtomicMemoryExtractionOutput.model_validate(result.output) + try: + output = AtomicMemoryExtractionOutput.model_validate(result.output) + except ValidationError as error: + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-extract", "output does not match the extraction schema"), + ProcessingDiagnostics( + stage=ProcessingStage.OUTPUT_PARSING, + reason=ProcessingFailureReason.SCHEMA_INVALID, + failure_location="atomic-memory-extract", + ), + ) from error supplied = {item.evidence_id for item in value.evidence} for candidate in output.candidates: - AtomicMemoryContent(kind=candidate.kind, text=candidate.text) + try: + AtomicMemoryContent(kind=candidate.kind, text=candidate.text) + except ValidationError as error: + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-extract", "candidate content is invalid"), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="atomic-memory-extract", + ), + ) from error if not set(candidate.evidence_ids) <= supplied: - raise InvalidInferenceOutputError( - "atomic-memory-extract", "candidate cites unavailable Source evidence" + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-extract", "candidate cites unavailable Source evidence"), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="atomic-memory-extract", + ), ) return output.candidates diff --git a/src/powercontext/builtin/artifacts/atomic_memory/reconciliation.py b/src/powercontext/builtin/artifacts/atomic_memory/reconciliation.py index 260038f513..2261d08ede 100644 --- a/src/powercontext/builtin/artifacts/atomic_memory/reconciliation.py +++ b/src/powercontext/builtin/artifacts/atomic_memory/reconciliation.py @@ -20,13 +20,19 @@ from dataclasses import dataclass, replace from typing import Any, Literal, NoReturn -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator from powercontext.artifacts import ArtifactRef from powercontext.builtin.artifacts.atomic_memory.errors import AtomicMemoryConflictError from powercontext.builtin.artifacts.atomic_memory.extraction import LANGUAGE_RULES, TIME_RULES, AtomicMemoryEvidence from powercontext.builtin.artifacts.atomic_memory.models import AtomicMemoryContent, AtomicMemoryRead from powercontext.builtin.inference import InvalidInferenceOutputError +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.sources import SourceRef ATOMIC_MEMORY_RECONCILIATION_INSTRUCTIONS = f""" @@ -222,7 +228,17 @@ def apply( if output.content is None: _invalid("write content is missing") - content = AtomicMemoryContent(kind=output.content.kind, text=output.content.text) + try: + content = AtomicMemoryContent(kind=output.content.kind, text=output.content.text) + except ValidationError as error: + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-reconcile", "write content is invalid"), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="atomic-memory-reconcile", + ), + ) from error inputs = (proposal, *targets) origins = tuple({read.ref.artifact_id: read for item in inputs for read in item.origins}.values()) evidence_by_id = {item.evidence_id: item for item in request.evidence} @@ -330,4 +346,11 @@ def unique_evidence( def _invalid(reason: str) -> NoReturn: - raise InvalidInferenceOutputError("atomic-memory-reconcile", reason) + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-reconcile", reason), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="atomic-memory-reconcile", + ), + ) diff --git a/src/powercontext/builtin/artifacts/memory/extraction.py b/src/powercontext/builtin/artifacts/memory/extraction.py index 6b46088895..3261495df0 100644 --- a/src/powercontext/builtin/artifacts/memory/extraction.py +++ b/src/powercontext/builtin/artifacts/memory/extraction.py @@ -17,7 +17,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Literal, Protocol, TypeAlias +from typing import Literal, NoReturn, Protocol, TypeAlias from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError @@ -26,6 +26,12 @@ from powercontext.builtin.artifacts.memory.models import MemoryEntryInput, MemoryEntryVersion from powercontext.builtin.artifacts.memory.protocols import MemoryCandidateRequest from powercontext.builtin.inference import GenerationResult, InvalidInferenceOutputError, StructuredGenerator +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.sources import Source MemoryExtractionIntent: TypeAlias = Literal["add", "revise"] @@ -194,16 +200,14 @@ def _selected_evidence( evidence: Mapping[str, Source | Artifact[object]], ) -> tuple[tuple[Source, ...], tuple[Artifact[object], ...]]: if not proposal.evidence_ids: - raise InvalidInferenceOutputError("memory-extract", "candidate does not cite evidence") + _business_invalid("candidate does not cite evidence") selected_sources: list[Source] = [] selected_artifacts: list[Artifact[object]] = [] for evidence_id in dict.fromkeys(proposal.evidence_ids): try: value = evidence[evidence_id] except KeyError: - raise InvalidInferenceOutputError( - "memory-extract", "candidate cites evidence outside the request" - ) from None + _business_invalid("candidate cites evidence outside the request") if isinstance(value, Source): selected_sources.append(value) else: @@ -218,24 +222,33 @@ def _revision_target( ) -> MemoryEntryVersion | None: if proposal.intent == "add": if proposal.entry_id is not None: - raise InvalidInferenceOutputError("memory-extract", "add candidate must not identify an existing entry") + _business_invalid("add candidate must not identify an existing entry") return None if proposal.intent != "revise": - raise InvalidInferenceOutputError("memory-extract", "candidate has an unsupported intent") + _business_invalid("candidate has an unsupported intent") if proposal.entry_id is None: - raise InvalidInferenceOutputError("memory-extract", "revise candidate must identify an active entry") + _business_invalid("revise candidate must identify an active entry") try: entry = current_entries[proposal.entry_id] except KeyError: - raise InvalidInferenceOutputError( - "memory-extract", "revise candidate does not identify an active entry" - ) from None + _business_invalid("revise candidate does not identify an active entry") if proposal.entry_id in revised_entries: - raise InvalidInferenceOutputError("memory-extract", "an active entry can only be revised once per extraction") + _business_invalid("an active entry can only be revised once per extraction") revised_entries.add(proposal.entry_id) return entry +def _business_invalid(reason: str) -> NoReturn: + raise attach_diagnostics( + InvalidInferenceOutputError("memory-extract", reason), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + failure_location="memory-extract", + ), + ) from None + + def _validated_json(value: object) -> JsonValue: try: return _JSON_VALUE.validate_python(value) diff --git a/src/powercontext/builtin/inference/minimax.py b/src/powercontext/builtin/inference/minimax.py index 9ecc8bf8d0..9e541361a5 100644 --- a/src/powercontext/builtin/inference/minimax.py +++ b/src/powercontext/builtin/inference/minimax.py @@ -41,6 +41,12 @@ InvalidInferenceOutputError, ) from powercontext.builtin.inference.models import EmbeddingResult, InferenceUsage +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) _MINIMAX_DOCUMENT_TYPE = "db" _MINIMAX_QUERY_TYPE = "query" @@ -94,7 +100,9 @@ async def embed_query(self, texts: tuple[str, ...], /) -> EmbeddingResult: return await self._embed(texts, embedding_type=_MINIMAX_QUERY_TYPE) - async def _embed(self, texts: tuple[str, ...], *, embedding_type: str) -> EmbeddingResult: + async def _embed( # noqa: C901 - bounded transport and HTTP status mapping + self, texts: tuple[str, ...], *, embedding_type: str + ) -> EmbeddingResult: """Embed one MiniMax document or query batch.""" if not texts: @@ -109,11 +117,52 @@ async def _embed(self, texts: tuple[str, ...], *, embedding_type: str) -> Embedd except (InvalidInferenceOutputError, InferenceConfigurationError): raise except TimeoutError as error: - raise InferenceTimeoutError("embed", self._timeout) from error + raise attach_diagnostics( + InferenceTimeoutError("embed", self._timeout), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=ProcessingFailureReason.TIMEOUT, + cause_exception_type=type(error).__name__, + ), + ) from error except httpx.TimeoutException as error: - raise InferenceTimeoutError("embed", self._timeout) from error + raise attach_diagnostics( + InferenceTimeoutError("embed", self._timeout), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=ProcessingFailureReason.TIMEOUT, + cause_exception_type=type(error).__name__, + ), + ) from error + except httpx.HTTPStatusError as error: + status = error.response.status_code + if status == 429: + reason = ProcessingFailureReason.RATE_LIMIT + elif status in {408, 504}: + reason = ProcessingFailureReason.TIMEOUT + elif status in {401, 403}: + reason = ProcessingFailureReason.CONFIGURATION + elif status in {409, 425} or status >= 500: + reason = ProcessingFailureReason.PROVIDER_UNAVAILABLE + else: + reason = None + raise attach_diagnostics( + InferenceUnavailableError("embed"), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=reason, + cause_exception_type=type(error).__name__, + ), + ) from error except httpx.HTTPError as error: - raise InferenceUnavailableError("embed") from error + raise attach_diagnostics( + InferenceUnavailableError("embed"), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=ProcessingFailureReason.PROVIDER_UNAVAILABLE, + cause_exception_type=type(error).__name__, + ), + ) from error return result async def _embed_batches(self, texts: tuple[str, ...], *, embedding_type: str) -> EmbeddingResult: diff --git a/src/powercontext/builtin/inference/pydantic_ai.py b/src/powercontext/builtin/inference/pydantic_ai.py index 5145c9d48b..42e5cb0dd7 100644 --- a/src/powercontext/builtin/inference/pydantic_ai.py +++ b/src/powercontext/builtin/inference/pydantic_ai.py @@ -42,6 +42,13 @@ GenerationResult, InferenceUsage, ) +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, + stage_for_operation, +) class PydanticAIConfigurationError(InferenceConfigurationError): @@ -128,7 +135,18 @@ async def request( ) -> ModelResponse: response = await self.wrapped.request(messages, model_settings, model_request_parameters) if response.state != "complete": - raise InvalidInferenceOutputError("generate", "provider continuation is not allowed") + finish_reason = getattr(response, "finish_reason", None) + truncated = str(finish_reason or "").lower() in {"length", "max_tokens"} + raise attach_diagnostics( + InvalidInferenceOutputError("generate", "provider continuation is not allowed"), + ProcessingDiagnostics( + stage=ProcessingStage.GENERATION, + reason=ProcessingFailureReason.OUTPUT_TRUNCATED if truncated else None, + provider_request_id=getattr(response, "provider_response_id", None), + finish_reason=finish_reason, + failure_location="provider-response", + ), + ) from None return response @@ -384,6 +402,13 @@ async def _embed(self, texts: tuple[str, ...], *, input_type: Literal["document" raise except InvalidInferenceOutputError: raise + except ValidationError as error: + # Malformed provider output is an invalid-output failure, not a + # transient availability failure; match the structured mapping. + mapped = _map_error(error, operation="embed", timeout_seconds=self._limits.timeout_seconds) + if mapped is None: + raise + raise mapped from error except ValueError as error: raise InferenceUnavailableError("embed") from error except Exception as error: @@ -495,7 +520,7 @@ def _checked_rule(error: BaseException) -> InvalidInferenceOutputError | None: return None -def _map_error( +def _map_error( # noqa: C901 - bounded mapping across status codes and provider exception classes error: Exception, *, operation: str, @@ -503,23 +528,122 @@ def _map_error( ) -> InferenceError | None: if isinstance(error, InferenceError): return error + stage = stage_for_operation(operation) + cause_type = type(error).__name__ if isinstance(error, TimeoutError): - return InferenceTimeoutError(operation, timeout_seconds) + return attach_diagnostics( + InferenceTimeoutError(operation, timeout_seconds), + ProcessingDiagnostics(stage=stage, reason=ProcessingFailureReason.TIMEOUT, cause_exception_type=cause_type), + ) if isinstance(error, ModelHTTPError): if error.status_code in {408, 504}: - return InferenceTimeoutError(operation, timeout_seconds) - if error.status_code in {409, 425, 429} or error.status_code >= 500: - return InferenceUnavailableError(operation) - return PydanticAIConfigurationError("provider-rejected", detail=f"HTTP {error.status_code}") + return attach_diagnostics( + InferenceTimeoutError(operation, timeout_seconds), + ProcessingDiagnostics( + stage=stage, + reason=ProcessingFailureReason.TIMEOUT, + cause_exception_type=cause_type, + ), + ) + if error.status_code in {409, 425, 429}: + return attach_diagnostics( + InferenceUnavailableError(operation), + ProcessingDiagnostics( + stage=stage, + reason=( + ProcessingFailureReason.RATE_LIMIT + if error.status_code == 429 + else ProcessingFailureReason.PROVIDER_UNAVAILABLE + ), + cause_exception_type=cause_type, + ), + ) + if error.status_code >= 500: + return attach_diagnostics( + InferenceUnavailableError(operation), + ProcessingDiagnostics( + stage=stage, + reason=ProcessingFailureReason.PROVIDER_UNAVAILABLE, + cause_exception_type=cause_type, + ), + ) + reason = ( + ProcessingFailureReason.CONTEXT_OVERFLOW + if _looks_like_context_overflow(error) + else ProcessingFailureReason.CONFIGURATION + ) + return attach_diagnostics( + PydanticAIConfigurationError("provider-rejected", detail=f"HTTP {error.status_code}"), + ProcessingDiagnostics( + stage=stage, + reason=reason, + failure_location="provider-response", + cause_exception_type=cause_type, + ), + ) if isinstance(error, (ModelAPIError, ConcurrencyLimitExceeded, OSError)): - return InferenceUnavailableError(operation) - if isinstance(error, (UnexpectedModelBehavior, UsageLimitExceeded, ValidationError)): - return InvalidInferenceOutputError(operation, "provider did not return a valid result") + return attach_diagnostics( + InferenceUnavailableError(operation), + ProcessingDiagnostics( + stage=stage, + reason=ProcessingFailureReason.PROVIDER_UNAVAILABLE, + cause_exception_type=cause_type, + ), + ) + if isinstance(error, UsageLimitExceeded): + return attach_diagnostics( + InvalidInferenceOutputError(operation, "provider did not return a valid result"), + ProcessingDiagnostics( + stage=stage or ProcessingStage.GENERATION, + reason=ProcessingFailureReason.USAGE_LIMIT, + failure_location="inference-limits", + cause_exception_type=cause_type, + ), + ) + if isinstance(error, (UnexpectedModelBehavior, ValidationError)): + embedding = stage is ProcessingStage.EMBEDDING + return attach_diagnostics( + InvalidInferenceOutputError(operation, "provider did not return a valid result"), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING if embedding else ProcessingStage.OUTPUT_PARSING, + reason=( + ProcessingFailureReason.INVALID_OUTPUT if embedding else ProcessingFailureReason.SCHEMA_INVALID + ), + failure_location="provider-output", + cause_exception_type=cause_type, + ), + ) if isinstance(error, (PydanticSchemaGenerationError, PydanticUserError, UserError)): - return PydanticAIConfigurationError("pydantic-rejected") + return attach_diagnostics( + PydanticAIConfigurationError("pydantic-rejected"), + ProcessingDiagnostics( + stage=stage, + reason=ProcessingFailureReason.CONFIGURATION, + cause_exception_type=cause_type, + ), + ) return None +def _looks_like_context_overflow(error: ModelHTTPError) -> bool: + """Classify only when the provider's own response names a context limit.""" + + body = getattr(error, "body", None) + text = str(body if isinstance(body, (str, bytes)) else body or "").lower() + return any( + term in text + for term in ( + "context length", + "context window", + "maximum context", + "too many tokens", + "prompt is too long", + "input is too long", + "token limit", + ) + ) + + __all__ = [ "InferenceLimits", "PydanticAIConfigurationError", diff --git a/src/powercontext/builtin/processing_diagnostics.py b/src/powercontext/builtin/processing_diagnostics.py new file mode 100644 index 0000000000..056dd38957 --- /dev/null +++ b/src/powercontext/builtin/processing_diagnostics.py @@ -0,0 +1,261 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Bounded, content-free failure diagnostics shared by every processing Family.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, TypeVar + +_ErrorT = TypeVar("_ErrorT", bound=BaseException) + + +class ProcessingStage(StrEnum): + """The processing stage that owned a failure.""" + + GENERATION = "generation" + EMBEDDING = "embedding" + OUTPUT_PARSING = "output_parsing" + BUSINESS_VALIDATION = "business_validation" + PERSISTENCE = "persistence" + # Reserved for usage reporting or persistence failures. + USAGE_ACCOUNTING = "usage_accounting" + + +class ProcessingFailureReason(StrEnum): + """A bounded reason, only when the observed evidence supports it.""" + + CONTEXT_OVERFLOW = "context_overflow" + OUTPUT_TRUNCATED = "output_truncated" + RATE_LIMIT = "rate_limit" + TIMEOUT = "timeout" + SCHEMA_INVALID = "schema_invalid" + PROVIDER_UNAVAILABLE = "provider_unavailable" + CONFIGURATION = "configuration" + INVALID_OUTPUT = "invalid_output" + STORAGE_FAILURE = "storage_failure" + USAGE_LIMIT = "usage_limit" + + +_MAX_COUNT = 2**31 - 1 +_TEXT_LIMITS = { + "token_estimator_id": 128, + "provider_request_id": 128, + "finish_reason": 64, + "failure_location": 128, + "cause_exception_type": 128, +} + + +def _enum(enum_type: type[StrEnum], value: object) -> Any: + if value is None: + return None + try: + return enum_type(value) + except ValueError: + return None + + +def _count(value: object) -> int | None: + if not isinstance(value, int) or isinstance(value, bool) or value < 0 or value > _MAX_COUNT: + return None + return value + + +def _text(value: object, limit: int) -> str | None: + if not isinstance(value, str): + return None + cleaned = "".join(character for character in value if character.isprintable()).strip() + if not cleaned: + return None + return cleaned[:limit] + + +@dataclass(frozen=True, slots=True) +class ProcessingDiagnostics: + """Safe causal context for one processing failure. + + Every field is bounded: counts are non-negative, provider identifiers and + reasons are printable and length-limited, and no field carries prompts, + Source or Artifact bodies, model responses, credentials, or SQL parameters. + """ + + stage: ProcessingStage | None = None + reason: ProcessingFailureReason | None = None + input_item_count: int | None = None + estimated_input_tokens: int | None = None + token_estimator_id: str | None = None + effective_budget: int | None = None + provider_request_id: str | None = None + finish_reason: str | None = None + failure_location: str | None = None + cause_exception_type: str | None = None + + def __post_init__(self) -> None: + object.__setattr__(self, "stage", _enum(ProcessingStage, self.stage)) + object.__setattr__(self, "reason", _enum(ProcessingFailureReason, self.reason)) + object.__setattr__(self, "input_item_count", _count(self.input_item_count)) + object.__setattr__(self, "estimated_input_tokens", _count(self.estimated_input_tokens)) + object.__setattr__(self, "effective_budget", _count(self.effective_budget)) + for name, limit in _TEXT_LIMITS.items(): + object.__setattr__(self, name, _text(getattr(self, name), limit)) + + def log_fields(self) -> dict[str, str | int]: + """Return only the allowlisted operational fields that are set.""" + + fields: dict[str, str | int] = {} + if self.stage is not None: + fields["processing_stage"] = self.stage.value + if self.reason is not None: + fields["failure_reason"] = self.reason.value + for name in ( + "input_item_count", + "estimated_input_tokens", + "token_estimator_id", + "effective_budget", + "provider_request_id", + "finish_reason", + "failure_location", + "cause_exception_type", + ): + value = getattr(self, name) + if value is not None: + fields[name] = value + return fields + + +def attach_diagnostics(error: _ErrorT, diagnostics: ProcessingDiagnostics | None) -> _ErrorT: + """Attach safe diagnostics to an exception without changing its type or message.""" + + if diagnostics is not None: + setattr(error, "processing_diagnostics", diagnostics) # noqa: B010 - dynamic exception attribute + return error + + +def diagnostics_of(error: BaseException) -> ProcessingDiagnostics | None: + value = getattr(error, "processing_diagnostics", None) + return value if isinstance(value, ProcessingDiagnostics) else None + + +def safe_classify_processing_error(error: BaseException) -> ProcessingDiagnostics | None: + """Never let classification itself change a processing outcome.""" + + try: + return classify_processing_error(error) + except Exception: + return None + + +def safe_log_fields(diagnostics: ProcessingDiagnostics | None) -> dict[str, str | int]: + """Never let a diagnostics field failure suppress the failure record.""" + + if diagnostics is None: + return {} + try: + return diagnostics.log_fields() + except Exception: + return {} + + +def stage_for_operation(operation: object) -> ProcessingStage | None: + if not isinstance(operation, str) or not operation: + return 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: + return ProcessingStage.GENERATION + return None + + +def classify_processing_error(error: BaseException) -> ProcessingDiagnostics | None: # noqa: C901 - bounded vocabulary + """Derive a bounded classification when no raise site attached one.""" + + attached = diagnostics_of(error) + if attached is not None: + return attached + + names = {base.__name__ for base in type(error).__mro__} + operation = getattr(error, "operation", None) + stage = stage_for_operation(operation) + if "InferenceTimeoutError" in names or isinstance(error, TimeoutError): + return ProcessingDiagnostics(stage=stage, reason=ProcessingFailureReason.TIMEOUT) + if "InferenceUnavailableError" in names: + status_code = getattr(error, "status_code", None) + reason = ( + ProcessingFailureReason.RATE_LIMIT if status_code == 429 else ProcessingFailureReason.PROVIDER_UNAVAILABLE + ) + return ProcessingDiagnostics(stage=stage, reason=reason) + if "InvalidInferenceOutputError" in names: + detail = str(getattr(error, "detail", "") or "").lower() + operation_value = operation.lower() if isinstance(operation, str) else "" + if "token budget" in detail or "budget" in operation_value: + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.CONTEXT_OVERFLOW, + ) + if any(term in detail for term in ("schema", "typed", "wrong output", "invalid output tree", "valid result")): + return ProcessingDiagnostics( + stage=ProcessingStage.OUTPUT_PARSING, + reason=ProcessingFailureReason.SCHEMA_INVALID, + ) + if "vector" in detail: + return ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=ProcessingFailureReason.INVALID_OUTPUT, + ) + return ProcessingDiagnostics( + stage=stage or ProcessingStage.OUTPUT_PARSING, + reason=ProcessingFailureReason.INVALID_OUTPUT, + ) + if "InferenceConfigurationError" in names: + return ProcessingDiagnostics(stage=stage, reason=ProcessingFailureReason.CONFIGURATION) + if "AtomicMemoryIndexError" in names: + code = str(getattr(error, "code", "") or "").lower() + if code in {"embedding-profile", "embedding-input", "embedding-result"}: + return ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=( + ProcessingFailureReason.CONFIGURATION + if code == "embedding-profile" + else ProcessingFailureReason.INVALID_OUTPUT + ), + ) + if code in {"content", "prepared-content"}: + return ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + ) + if code == "family": + return ProcessingDiagnostics(reason=ProcessingFailureReason.CONFIGURATION) + return ProcessingDiagnostics( + stage=ProcessingStage.PERSISTENCE, + reason=ProcessingFailureReason.STORAGE_FAILURE, + ) + return None + + +__all__ = [ + "ProcessingDiagnostics", + "ProcessingFailureReason", + "ProcessingStage", + "attach_diagnostics", + "classify_processing_error", + "diagnostics_of", + "safe_classify_processing_error", + "safe_log_fields", + "stage_for_operation", +] diff --git a/src/powercontext/builtin/runtime/artifact_processing.py b/src/powercontext/builtin/runtime/artifact_processing.py index a35f2adf36..ce436b343b 100644 --- a/src/powercontext/builtin/runtime/artifact_processing.py +++ b/src/powercontext/builtin/runtime/artifact_processing.py @@ -58,6 +58,7 @@ database_utc_now, ) from powercontext.builtin.persistence.tables import ARTIFACT_PROCESSING_LEASES_TABLE +from powercontext.builtin.processing_diagnostics import safe_classify_processing_error, safe_log_fields from powercontext.builtin.runtime.observations import DurationObservations, DurationObservationSnapshot from powercontext.builtin.runtime.processing_contracts import ( ArtifactProcessingBlock, @@ -1558,12 +1559,18 @@ def _log_failure( cancelled = isinstance(error, asyncio.CancelledError) or ( isinstance(error, _WorkerExecutionError) and error.failure.cancelled ) + diagnostics = ( + error.failure.diagnostics + if isinstance(error, _WorkerExecutionError) and error.failure.diagnostics is not None + else safe_classify_processing_error(error) + ) log_safely( logger, logging.INFO if cancelled else logging.ERROR, "Artifact processing cancelled" if cancelled else "Artifact processing failed", extra={ "event": "artifact_processing.failed", + **safe_log_fields(diagnostics), "stage": error.failure.stage if isinstance(error, _WorkerExecutionError) else _safe_error_attribute(error, "stage", stage), @@ -1715,6 +1722,7 @@ def _run_spawned_worker( ) except BaseException as error: error_code = processing_error_code(type(error), _safe_error_attribute(error, "error_code", "")) + diagnostics = safe_classify_processing_error(error) block = error.block if isinstance(error, ArtifactProcessingBlockedError) else None level = logging.ERROR outcome = "failure" @@ -1740,6 +1748,7 @@ def _run_spawned_worker( "stage": None if block is None else block.stage, "exception_type": type(error).__name__, "duration_ms": max(time.perf_counter() - started_at, 0) * 1000, + **safe_log_fields(diagnostics), }, ) sender.send( @@ -1755,6 +1764,7 @@ def _run_spawned_worker( block=error.block if isinstance(error, ArtifactProcessingBlockedError) else None, cancelled=isinstance(error, asyncio.CancelledError), model_usage=None if collector is None else collector.snapshot(), + diagnostics=diagnostics, ) ) finally: diff --git a/src/powercontext/builtin/runtime/atomic_memory_processing.py b/src/powercontext/builtin/runtime/atomic_memory_processing.py index aacc62d86f..24ae5534ac 100644 --- a/src/powercontext/builtin/runtime/atomic_memory_processing.py +++ b/src/powercontext/builtin/runtime/atomic_memory_processing.py @@ -78,6 +78,12 @@ SOURCE_JOURNAL_HEADS_TABLE, SOURCES_TABLE, ) +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.builtin.runtime.atomic_memory import AtomicMemoryApplication, deferred_decision_audit from powercontext.builtin.runtime.decision_policy import emit_decision_observation from powercontext.builtin.runtime.models import MemoryFlushResult @@ -809,7 +815,7 @@ async def _read_transaction(self, context, *, write_policy_first=False, consiste def _stage(self, name, attributes): return nullcontext(None) if self.tracing is None else self.tracing.stage(name, attributes=attributes) - def _fits(self, pipeline, value, key): + def _estimate(self, pipeline, value, key): prompt = current_prompt(key) instructions = ( ( @@ -820,15 +826,30 @@ def _fits(self, pipeline, value, key): if prompt is None else prompt.compiled_instructions ) - return ( - pipeline.estimator.estimate(f"{instructions}\n{value.model_dump_json()}") <= self.config.input_tokens_limit - ) + return pipeline.estimator.estimate(f"{instructions}\n{value.model_dump_json()}") + + def _fits(self, pipeline, value, key): + return self._estimate(pipeline, value, key) <= self.config.input_tokens_limit def _require_budget(self, pipeline, value, key): - if not self._fits(pipeline, value, key): - raise InvalidInferenceOutputError( - "atomic-memory-input-budget", "A complete comparison input exceeds its token budget" - ) + estimated = self._estimate(pipeline, value, key) + if estimated <= self.config.input_tokens_limit: + return + error = InvalidInferenceOutputError( + "atomic-memory-input-budget", "A complete comparison input exceeds its token budget" + ) + raise attach_diagnostics( + error, + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.CONTEXT_OVERFLOW, + input_item_count=_input_item_count(value), + estimated_input_tokens=estimated, + token_estimator_id=pipeline.estimator.profile.estimator_id, + effective_budget=self.config.input_tokens_limit, + failure_location=key, + ), + ) @staticmethod def _require_window(window, after, through): @@ -955,3 +976,10 @@ def _window_operation_id(prepared) -> str: def _conflict(detail: str) -> NoReturn: raise AtomicMemoryConflictError(detail) + + +def _input_item_count(value: object) -> int | None: + evidence = getattr(value, "evidence", None) + if isinstance(evidence, (tuple, list)): + return len(evidence) + return None diff --git a/src/powercontext/builtin/runtime/processing_contracts.py b/src/powercontext/builtin/runtime/processing_contracts.py index 6fcbd261bd..beced183c6 100644 --- a/src/powercontext/builtin/runtime/processing_contracts.py +++ b/src/powercontext/builtin/runtime/processing_contracts.py @@ -22,6 +22,7 @@ from typing import Protocol from powercontext.builtin.persistence.supervision import ArtifactProcessingFence +from powercontext.builtin.processing_diagnostics import ProcessingDiagnostics class ArtifactProcessingWorkerOutcome(StrEnum): @@ -100,6 +101,7 @@ class ArtifactProcessingWorkerFailure: block: ArtifactProcessingBlock | None = None cancelled: bool = False model_usage: ArtifactProcessingModelUsageDelta | None = None + diagnostics: ProcessingDiagnostics | None = None @dataclass(frozen=True, slots=True) diff --git a/src/powercontext/server/logging.py b/src/powercontext/server/logging.py index 08527a1d3a..310de3683d 100644 --- a/src/powercontext/server/logging.py +++ b/src/powercontext/server/logging.py @@ -57,6 +57,16 @@ "attempts", "requests", "tokens", + "processing_stage", + "failure_reason", + "input_item_count", + "estimated_input_tokens", + "token_estimator_id", + "effective_budget", + "provider_request_id", + "finish_reason", + "failure_location", + "cause_exception_type", ) _OPERATIONAL_FIELDS = ( "event", diff --git a/tests/builtin/inference/test_minimax_embedding.py b/tests/builtin/inference/test_minimax_embedding.py index 901e623d4a..ea518c5d57 100644 --- a/tests/builtin/inference/test_minimax_embedding.py +++ b/tests/builtin/inference/test_minimax_embedding.py @@ -35,6 +35,11 @@ _embedding_model_name, is_minimax_embedding, ) +from powercontext.builtin.processing_diagnostics import ( + ProcessingFailureReason, + ProcessingStage, + classify_processing_error, +) TEST_PROFILE = EmbeddingProfile( profile_id="mm-v1", @@ -159,6 +164,25 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_embed_classifies_rate_limited_http_status() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(429, json={"base_resp": {"status_code": 1002}}) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InferenceUnavailableError) as error: + await model.embed(("alpha",)) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.RATE_LIMIT + assert diagnostics.cause_exception_type == "HTTPStatusError" + + asyncio.run(scenario()) + + def test_embed_maps_http_timeout_to_inference_timeout() -> None: def handler(request: httpx.Request) -> httpx.Response: raise httpx.ReadTimeout("provider did not respond", request=request) # noqa: TRY003 @@ -293,3 +317,35 @@ def test_embedding_model_name_strips_provider_prefix() -> None: assert _embedding_model_name("minimax:embo-01") == "embo-01" assert _embedding_model_name("embo-01") == "embo-01" assert _embedding_model_name(None) == "" + + +@pytest.mark.parametrize( + ("status_code", "reason"), + [ + (401, ProcessingFailureReason.CONFIGURATION), + (403, ProcessingFailureReason.CONFIGURATION), + (408, ProcessingFailureReason.TIMEOUT), + (504, ProcessingFailureReason.TIMEOUT), + (409, ProcessingFailureReason.PROVIDER_UNAVAILABLE), + (425, ProcessingFailureReason.PROVIDER_UNAVAILABLE), + (500, ProcessingFailureReason.PROVIDER_UNAVAILABLE), + (400, None), + ], +) +def test_embed_classifies_http_status_reason(status_code, reason) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(status_code, json={"base_resp": {"status_code": 1000}}) + + async def scenario() -> None: + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = _model(http_client=client) + with pytest.raises(InferenceUnavailableError) as error: + await model.embed(("alpha",)) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is reason + assert diagnostics.cause_exception_type == "HTTPStatusError" + + asyncio.run(scenario()) diff --git a/tests/builtin/inference/test_pydantic_ai.py b/tests/builtin/inference/test_pydantic_ai.py index 9f53f8462a..af2a8760f6 100644 --- a/tests/builtin/inference/test_pydantic_ai.py +++ b/tests/builtin/inference/test_pydantic_ai.py @@ -25,6 +25,7 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from pydantic import BaseModel, ValidationError from pydantic_ai import Embedder from pydantic_ai.embeddings import ( EmbeddingModel as PydanticAIEmbeddingModelBase, @@ -37,7 +38,7 @@ TestEmbeddingModel, ) from pydantic_ai.embeddings.result import EmbedInputType -from pydantic_ai.exceptions import ModelHTTPError +from pydantic_ai.exceptions import ModelHTTPError, UsageLimitExceeded from pydantic_ai.messages import ( FinishReason, ModelMessage, @@ -67,8 +68,14 @@ PydanticAIEmbeddingModel, PydanticAIStructuredGenerator, StreamingAggregateModel, + _map_error, probe_pydantic_ai_model, ) +from powercontext.builtin.processing_diagnostics import ( + ProcessingFailureReason, + ProcessingStage, + classify_processing_error, +) @dataclass(frozen=True, slots=True) @@ -315,6 +322,11 @@ async def scenario() -> None: await generator.generate(Question("bounded evidence")) assert "secret" not in str(error.value) assert isinstance(error.value.__cause__, ModelHTTPError) + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.RATE_LIMIT + assert diagnostics.cause_exception_type == "ModelHTTPError" asyncio.run(scenario()) @@ -847,6 +859,10 @@ async def scenario() -> None: await adapter.embed(("bounded text",)) assert error.value.__cause__ is provider_error assert "secret" not in str(error.value) + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.PROVIDER_UNAVAILABLE asyncio.run(scenario()) @@ -865,6 +881,65 @@ async def scenario() -> None: assert error.value.detail == "HTTP 400" assert "HTTP 400" in str(error.value) assert "secret provider body" not in str(error.value) + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.reason is ProcessingFailureReason.CONFIGURATION + + asyncio.run(scenario()) + + +def test_embedding_adapter_classifies_a_provider_context_limit_without_leaking_the_body() -> None: + async def scenario() -> None: + provider_error = ModelHTTPError( + 400, + "result-model", + {"error": {"message": "maximum context length is 8192 tokens, body secret-provider-body"}}, + ) + adapter = PydanticAIEmbeddingModel( + embedder=Embedder(ResultEmbeddingModel((), error=provider_error)), + profile=TEST_PROFILE, + ) + + with pytest.raises(PydanticAIConfigurationError) as error: + await adapter.embed(("bounded text",)) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.CONTEXT_OVERFLOW + assert "secret-provider-body" not in str(error.value) + + asyncio.run(scenario()) + + +def test_structured_generator_classifies_a_length_finish_reason_as_truncated_output() -> None: + async def truncated(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + del messages, info + return ModelResponse( + parts=[TextPart("partial")], + provider_response_id="req-truncated", + finish_reason="length", + state="incomplete", + ) + + async def scenario() -> None: + generator = PydanticAIStructuredGenerator( + model=FunctionModel(truncated), + instructions="Return a value.", + input_type=Question, + output_type=Answer, + limits=InferenceLimits(allow_continuations=False), + ) + + with pytest.raises(InvalidInferenceOutputError) as error: + await generator.generate(Question("bounded evidence")) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.OUTPUT_TRUNCATED + assert diagnostics.finish_reason == "length" + assert diagnostics.provider_request_id == "req-truncated" asyncio.run(scenario()) @@ -910,3 +985,150 @@ async def scenario() -> None: await adapter.embed(("bounded text",)) asyncio.run(scenario()) + + +@pytest.mark.parametrize("status_code", [409, 425]) +def test_structured_generator_keeps_transient_provider_conflicts_unavailable(status_code) -> None: + async def conflict(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + del messages, info + raise ModelHTTPError(status_code, "test-model", {"error": "try again"}) + + async def scenario() -> None: + generator = PydanticAIStructuredGenerator( + model=FunctionModel(conflict), + instructions="Return a value.", + input_type=Question, + output_type=Answer, + ) + + with pytest.raises(InferenceUnavailableError) as error: + await generator.generate(Question("bounded evidence")) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.reason is ProcessingFailureReason.PROVIDER_UNAVAILABLE + assert diagnostics.cause_exception_type == "ModelHTTPError" + + asyncio.run(scenario()) + + +def test_usage_limit_maps_to_the_originating_generation_stage() -> None: + mapped = _map_error(UsageLimitExceeded("request limit"), operation="generate", timeout_seconds=1.0) + + assert mapped is not None + diagnostics = classify_processing_error(mapped) + + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT + assert diagnostics.cause_exception_type == "UsageLimitExceeded" + + +def test_usage_limit_public_generate_path_reports_generation_stage() -> None: + async def respond(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + del messages, info + return ModelResponse(parts=[TextPart('{"value":"stable"}')], usage=RequestUsage(output_tokens=11)) + + async def scenario() -> None: + generator = PydanticAIStructuredGenerator( + model=FunctionModel(respond), + instructions="Return a value.", + input_type=Question, + output_type=Answer, + limits=InferenceLimits(output_tokens_limit=10), + ) + + with pytest.raises(InvalidInferenceOutputError) as error: + await generator.generate(Question("bounded evidence")) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT + assert diagnostics.cause_exception_type == "UsageLimitExceeded" + + asyncio.run(scenario()) + + +def test_a_413_without_context_evidence_is_not_context_overflow() -> None: + mapped = _map_error( + ModelHTTPError(413, "test-model", {"error": "upload too large"}), + operation="generate", + timeout_seconds=1.0, + ) + + assert mapped is not None + diagnostics = classify_processing_error(mapped) + + assert diagnostics is not None + assert diagnostics.reason is ProcessingFailureReason.CONFIGURATION + + +def test_non_length_incomplete_response_does_not_claim_truncation() -> None: + async def filtered(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: + del messages, info + return ModelResponse(parts=[TextPart("partial")], finish_reason="content_filter", state="incomplete") + + async def scenario() -> None: + generator = PydanticAIStructuredGenerator( + model=FunctionModel(filtered), + instructions="Return a value.", + input_type=Question, + output_type=Answer, + limits=InferenceLimits(allow_continuations=False), + ) + + with pytest.raises(InvalidInferenceOutputError) as error: + await generator.generate(Question("bounded evidence")) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.reason is None + assert diagnostics.finish_reason == "content_filter" + + asyncio.run(scenario()) + + +def test_embedding_public_path_maps_validation_error_to_invalid_output() -> None: + class Payload(BaseModel): + value: int + + with pytest.raises(ValidationError) as captured: + Payload.model_validate({"value": "not-an-integer"}) + validation_error = captured.value + + async def scenario() -> None: + adapter = PydanticAIEmbeddingModel( + embedder=Embedder(ResultEmbeddingModel((), error=validation_error)), + profile=TEST_PROFILE, + ) + + with pytest.raises(InvalidInferenceOutputError) as error: + await adapter.embed(("bounded text",)) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + assert diagnostics.cause_exception_type == "ValidationError" + + asyncio.run(scenario()) + + +def test_embedding_public_path_reports_usage_limit_in_embedding_stage() -> None: + async def scenario() -> None: + adapter = PydanticAIEmbeddingModel( + embedder=Embedder(ResultEmbeddingModel((), error=UsageLimitExceeded("limit"))), + profile=TEST_PROFILE, + ) + + with pytest.raises(InvalidInferenceOutputError) as error: + await adapter.embed(("bounded text",)) + + diagnostics = classify_processing_error(error.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT + assert diagnostics.cause_exception_type == "UsageLimitExceeded" + + asyncio.run(scenario()) diff --git a/tests/builtin/test_processing_diagnostics.py b/tests/builtin/test_processing_diagnostics.py new file mode 100644 index 0000000000..f8aa0e909e --- /dev/null +++ b/tests/builtin/test_processing_diagnostics.py @@ -0,0 +1,343 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Shared processing failure diagnostics stay bounded, safe, and classifiable.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import pickle +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from powercontext.builtin.artifacts.atomic_memory.extraction import ( + AtomicMemoryCandidate, + AtomicMemoryEvidence, + AtomicMemoryExtractionInput, + AtomicMemoryExtractionOutput, + AtomicMemoryGenerationPipeline, +) +from powercontext.builtin.artifacts.memory.extraction import MemoryExtractionCandidate, _selected_evidence +from powercontext.builtin.inference.errors import ( + InferenceTimeoutError, + InferenceUnavailableError, + InvalidInferenceOutputError, +) +from powercontext.builtin.inference.models import GenerationResult +from powercontext.builtin.inference.pydantic_ai import PydanticAIConfigurationError +from powercontext.builtin.inference.tokens import character_token_estimator +from powercontext.builtin.persistence.atomic_memory_index import AtomicMemoryIndexError +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, + classify_processing_error, + safe_classify_processing_error, +) +from powercontext.builtin.runtime.atomic_memory_processing import AtomicMemorySourceWindowProcessor +from powercontext.builtin.runtime.processing_contracts import ArtifactProcessingWorkerFailure +from powercontext.server.logging import JsonFormatter +from powercontext.sources import SourceRef + + +def test_diagnostics_bound_and_sanitize_every_field() -> None: + diagnostics = ProcessingDiagnostics( + stage=cast(Any, "generation"), + reason=cast(Any, "context_overflow"), + input_item_count=3, + estimated_input_tokens=120, + token_estimator_id="character\x00weighted", # noqa: S106 - estimator identity, not a secret + effective_budget=100, + provider_request_id=" req\nid ", + finish_reason="length\x00", + failure_location="location" * 40, + ) + + assert diagnostics.stage is ProcessingStage.GENERATION + assert diagnostics.reason is ProcessingFailureReason.CONTEXT_OVERFLOW + assert diagnostics.token_estimator_id == "characterweighted" # noqa: S105 - estimator identity + assert diagnostics.provider_request_id == "reqid" + assert diagnostics.finish_reason == "length" + assert diagnostics.failure_location is not None + assert len(diagnostics.failure_location) == 128 + + +def test_diagnostics_reject_invalid_enums_and_counts() -> None: + diagnostics = ProcessingDiagnostics( + stage=cast(Any, "unknown-stage"), + reason=cast(Any, "unknown-reason"), + input_item_count=-1, + estimated_input_tokens=True, + effective_budget=2**40, + ) + + assert diagnostics.stage is None + assert diagnostics.reason is None + assert diagnostics.input_item_count is None + assert diagnostics.estimated_input_tokens is None + assert diagnostics.effective_budget is None + + +def test_log_fields_only_include_set_values() -> None: + fields = ProcessingDiagnostics( + stage=ProcessingStage.OUTPUT_PARSING, + reason=ProcessingFailureReason.SCHEMA_INVALID, + provider_request_id="req-1", + ).log_fields() + + assert fields == { + "processing_stage": "output_parsing", + "failure_reason": "schema_invalid", + "provider_request_id": "req-1", + } + + +def test_attached_diagnostics_take_precedence_over_derived_classification() -> None: + error = InvalidInferenceOutputError("atomic-memory-extract", "provider did not return a valid result") + attached = ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.INVALID_OUTPUT, + ) + attach_diagnostics(error, attached) + + assert classify_processing_error(error) is attached + + +@pytest.mark.parametrize( + ("error", "stage", "reason"), + [ + (InferenceTimeoutError("generate", 5.0), ProcessingStage.GENERATION, ProcessingFailureReason.TIMEOUT), + (InferenceTimeoutError("embed", 5.0), ProcessingStage.EMBEDDING, ProcessingFailureReason.TIMEOUT), + ( + InferenceUnavailableError("embed"), + ProcessingStage.EMBEDDING, + ProcessingFailureReason.PROVIDER_UNAVAILABLE, + ), + ( + InvalidInferenceOutputError( + "atomic-memory-input-budget", "A complete comparison input exceeds its token budget" + ), + ProcessingStage.BUSINESS_VALIDATION, + ProcessingFailureReason.CONTEXT_OVERFLOW, + ), + ( + InvalidInferenceOutputError("generate", "provider did not return a valid result"), + ProcessingStage.OUTPUT_PARSING, + ProcessingFailureReason.SCHEMA_INVALID, + ), + ( + InvalidInferenceOutputError("embed", "provider returned the wrong vector count"), + ProcessingStage.EMBEDDING, + ProcessingFailureReason.INVALID_OUTPUT, + ), + ( + PydanticAIConfigurationError("provider-rejected"), + None, + ProcessingFailureReason.CONFIGURATION, + ), + ], +) +def test_classifies_known_failures(error, stage, reason) -> None: + diagnostics = classify_processing_error(error) + + assert diagnostics is not None + assert diagnostics.stage is stage + assert diagnostics.reason is reason + + +def test_rate_limited_unavailable_error_is_distinguished() -> None: + error = InferenceUnavailableError("generate") + setattr(error, "status_code", 429) # noqa: B010 - provider status attached by the adapter + + diagnostics = classify_processing_error(error) + + assert diagnostics is not None + assert diagnostics.reason is ProcessingFailureReason.RATE_LIMIT + + +def test_worker_failure_carries_diagnostics_across_a_process_boundary() -> None: + diagnostics = ProcessingDiagnostics( + stage=ProcessingStage.GENERATION, + reason=ProcessingFailureReason.OUTPUT_TRUNCATED, + provider_request_id="req-1", + finish_reason="length", + ) + failure = ArtifactProcessingWorkerFailure( + stage="worker", + error_code="invalid_model_output", + exception_type="InvalidInferenceOutputError", + traceback="", + diagnostics=diagnostics, + ) + + restored = pickle.loads(pickle.dumps(failure)) # noqa: S301 - round-trip of a trusted local object + + assert restored.diagnostics == diagnostics + + +def test_budget_failure_reports_safe_accounting_fields() -> None: + processor = object.__new__(AtomicMemorySourceWindowProcessor) + processor.config = SimpleNamespace(input_tokens_limit=10) + pipeline = SimpleNamespace( + estimator=character_token_estimator(), + extraction_instructions="extract", + ) + value = SimpleNamespace( + evidence=(object(), object(), object()), + model_dump_json=lambda: "a long body that exceeds the configured budget", + ) + + with pytest.raises(InvalidInferenceOutputError) as raised: + processor._require_budget(pipeline, value, "atomic_memory.extract") + + diagnostics = classify_processing_error(raised.value) + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.BUSINESS_VALIDATION + assert diagnostics.reason is ProcessingFailureReason.CONTEXT_OVERFLOW + assert diagnostics.input_item_count == 3 + assert diagnostics.estimated_input_tokens is not None + assert diagnostics.estimated_input_tokens > 10 + assert diagnostics.token_estimator_id == "character:weighted" # noqa: S105 - estimator identity + assert diagnostics.effective_budget == 10 + assert diagnostics.failure_location == "atomic_memory.extract" + + +def test_json_formatter_serializes_the_new_processing_fields() -> None: + record = logging.makeLogRecord({ + "msg": "Artifact processing failed", + "event": "artifact_processing.failed", + "outcome": "failure", + "processing_stage": "generation", + "failure_reason": "context_overflow", + "input_item_count": 3, + "estimated_input_tokens": 120, + "token_estimator_id": "character:weighted", + "effective_budget": 100, + "provider_request_id": "req-1", + "finish_reason": "length", + "failure_location": "atomic_memory.extract", + "unexpected_extra": "must not be serialized", + }) + + payload = json.loads(JsonFormatter().format(record)) + + assert payload["processing_stage"] == "generation" + assert payload["failure_reason"] == "context_overflow" + assert payload["input_item_count"] == 3 + assert payload["provider_request_id"] == "req-1" + assert "unexpected_extra" not in payload + + +@pytest.mark.parametrize( + ("code", "stage", "reason"), + [ + ("embedding-result", ProcessingStage.EMBEDDING, ProcessingFailureReason.INVALID_OUTPUT), + ("embedding-profile", ProcessingStage.EMBEDDING, ProcessingFailureReason.CONFIGURATION), + ("content", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("prepared-content", ProcessingStage.BUSINESS_VALIDATION, ProcessingFailureReason.INVALID_OUTPUT), + ("family", None, ProcessingFailureReason.CONFIGURATION), + ("stale-recall", ProcessingStage.PERSISTENCE, ProcessingFailureReason.STORAGE_FAILURE), + ], +) +def test_atomic_memory_index_errors_classify_by_code(code, stage, reason) -> None: + diagnostics = classify_processing_error(AtomicMemoryIndexError(code, "bounded detail")) + + assert diagnostics is not None + assert diagnostics.stage is stage + assert diagnostics.reason is reason + + +def test_safe_classifier_never_escapes() -> None: + class ExplodingError(RuntimeError): + @property + def operation(self) -> str: + raise RuntimeError("exploded") + + assert safe_classify_processing_error(ExplodingError()) is None + + +def test_memory_extraction_rules_attach_business_validation_diagnostics() -> None: + candidate = MemoryExtractionCandidate(intent="add", kind="fact", text="A fact", evidence_ids=()) + + with pytest.raises(InvalidInferenceOutputError) as raised: + _selected_evidence(candidate, {}) + + 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 == "memory-extract" + + +def test_atomic_memory_content_validation_attaches_business_validation_diagnostics() -> None: + class EmptyCandidateGenerator: + async def generate(self, value): + del value + return GenerationResult( + output=AtomicMemoryExtractionOutput( + candidates=(AtomicMemoryCandidate(kind="fact", text="", evidence_ids=("evidence-1",)),) + ) + ) + + pipeline = AtomicMemoryGenerationPipeline( + extractor=EmptyCandidateGenerator(), + reconciler=cast(Any, None), + estimator=character_token_estimator(), + ) + request = AtomicMemoryExtractionInput( + evidence=( + AtomicMemoryEvidence( + evidence_id="evidence-1", + source_ref=SourceRef(source_type="content", source_id="source-1"), + journal_position=1, + content={"text": "bounded"}, + source_metadata={}, + ), + ) + ) + + with pytest.raises(InvalidInferenceOutputError) as raised: + asyncio.run(pipeline.extract(request)) + + 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 == "atomic-memory-extract" + + +def test_console_filter_renders_the_new_processing_fields() -> None: + from powercontext.server.logging import _HumanContextFilter + + record = logging.makeLogRecord({ + "msg": "Artifact processing failed", + "event": "artifact_processing.failed", + "outcome": "failure", + "processing_stage": "generation", + "failure_reason": "rate_limit", + "provider_request_id": "req-1", + "cause_exception_type": "ModelHTTPError", + }) + + _HumanContextFilter().filter(record) + + processing_context = getattr(record, "processing_context", "") + assert 'processing_stage="generation"' in processing_context + assert 'failure_reason="rate_limit"' in processing_context + assert 'cause_exception_type="ModelHTTPError"' in processing_context diff --git a/tests/e2e/test_worker_logging.py b/tests/e2e/test_worker_logging.py index ffa412ea73..97b1e8fc0c 100644 --- a/tests/e2e/test_worker_logging.py +++ b/tests/e2e/test_worker_logging.py @@ -43,6 +43,12 @@ from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile from powercontext.builtin.persistence.supervision import ArtifactProcessingFence from powercontext.builtin.persistence.tables import SHARED_TABLES +from powercontext.builtin.processing_diagnostics import ( + ProcessingDiagnostics, + ProcessingFailureReason, + ProcessingStage, + attach_diagnostics, +) from powercontext.builtin.runtime.artifact_processing import ( ArtifactProcessingBinding, ArtifactProcessingSupervisor, @@ -85,6 +91,23 @@ def _raise_inference_failure(scope_id: str) -> None: raise InferenceUnavailableError("generation", _PRIVATE) if scope_id == "invalid-output": raise InvalidInferenceOutputError("memory-extract", _PRIVATE) + if scope_id == "diagnosed-failure": + raise attach_diagnostics( + InvalidInferenceOutputError( + "atomic-memory-input-budget", "A complete comparison input exceeds its token budget" + ), + ProcessingDiagnostics( + stage=ProcessingStage.BUSINESS_VALIDATION, + reason=ProcessingFailureReason.CONTEXT_OVERFLOW, + input_item_count=3, + estimated_input_tokens=120, + token_estimator_id="character:weighted", # noqa: S106 - estimator identity, not a secret + effective_budget=100, + provider_request_id="req-1", + finish_reason="length", + failure_location="atomic_memory.extract", + ), + ) def _logging_worker(assignment: ArtifactProcessingWorkAssignment) -> ArtifactProcessingWorkerCompletion: @@ -188,6 +211,7 @@ async def run() -> None: ("cancellation", "cancelled", "cancelled"), ("cancellation-subclass", "cancelled", "cancelled"), ("cancellation-name-collision", "internal", "failure"), + ("diagnosed-failure", "dependency", "failure"), ], ) def test_supervisor_logs_isolate_request_identity_and_preserve_child_cancellation( @@ -247,6 +271,17 @@ async def run() -> None: if scope_id == "cancellation-subclass": assert failure["exception_type"] == child["exception_type"] == "_WorkerCancelled" assert "request_id" not in failure + if scope_id == "diagnosed-failure": + for record in (failure, child): + assert record["processing_stage"] == "business_validation" + assert record["failure_reason"] == "context_overflow" + assert record["input_item_count"] == 3 + assert record["estimated_input_tokens"] == 120 + assert record["token_estimator_id"] == "character:weighted" # noqa: S105 - estimator identity + assert record["effective_budget"] == 100 + assert record["provider_request_id"] == "req-1" + assert record["finish_reason"] == "length" + assert record["failure_location"] == "atomic_memory.extract" caller = next(record for record in records if record.get("event") == "caller.observed") assert caller["request_id"] == "unrelated-parent-request" assert _PRIVATE not in output.getvalue() + child_output.out + child_output.err