From 497d751d329aa98fa5695c01bdc46aefb4be3776 Mon Sep 17 00:00:00 2001 From: heyamei690-hash gua Date: Thu, 22 Oct 2026 13:57:59 +0800 Subject: [PATCH 1/3] feat(processing): classify artifact processing failures with safe diagnostics (#2126) Add a shared failure contract for artifact processing and apply its first slice to Memory / Atomic Memory. Failures now carry a bounded processing stage and reason plus safe diagnostics (input item count, estimated input tokens and estimator identity, effective budget, provider request id and finish reason, original exception type, and failure location). The fields are allowlisted for JSON and console output and travel from a spawned Worker to the supervisor. Provider errors are classified only when the evidence supports it: timeout, rate limit, transient unavailability, provider-confirmed context overflow, schema failures, usage limits, and embedding validation. Budget exhaustion and extraction/reconciliation rules attach business-validation diagnostics. Diagnostics themselves degrade to no fields and never change the processing outcome. Publish the per-Family coverage matrix and the Memory / Atomic Memory operation-level table. Other Families remain Partial or Missing until their own slices land; this change is part of #2126, not its completion. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- docs/en/development/meta.json | 1 + docs/en/development/observability.md | 3 + docs/en/development/processing-failures.md | 84 +++++ docs/en/docs/operate/observability.md | 1 + docs/zh/development/meta.json | 1 + docs/zh/development/observability.md | 2 + docs/zh/development/processing-failures.md | 74 ++++ docs/zh/docs/operate/observability.md | 1 + .../artifacts/atomic_memory/extraction.py | 41 ++- .../artifacts/atomic_memory/reconciliation.py | 29 +- .../builtin/artifacts/memory/extraction.py | 37 +- src/powercontext/builtin/inference/minimax.py | 47 ++- .../builtin/inference/pydantic_ai.py | 139 ++++++- .../builtin/processing_diagnostics.py | 260 +++++++++++++ .../builtin/runtime/artifact_processing.py | 10 + .../runtime/atomic_memory_processing.py | 44 ++- .../builtin/runtime/processing_contracts.py | 2 + src/powercontext/server/logging.py | 10 + .../inference/test_minimax_embedding.py | 24 ++ tests/builtin/inference/test_pydantic_ai.py | 171 ++++++++- tests/builtin/test_processing_diagnostics.py | 343 ++++++++++++++++++ tests/e2e/test_worker_logging.py | 35 ++ 22 files changed, 1316 insertions(+), 43 deletions(-) create mode 100644 docs/en/development/processing-failures.md create mode 100644 docs/zh/development/processing-failures.md create mode 100644 src/powercontext/builtin/processing_diagnostics.py create mode 100644 tests/builtin/test_processing_diagnostics.py diff --git a/docs/en/development/meta.json b/docs/en/development/meta.json index 39bb09b4c..49a8f9f72 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 869c242fc..35ab05f33 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 000000000..7c99787ac --- /dev/null +++ b/docs/en/development/processing-failures.md @@ -0,0 +1,84 @@ +# 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` | +| `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, which names the Worker or +supervisor boundary (`worker_process`, `work_admission`, and so on). 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 | Partial | +| 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 | Partial | `UsageLimitExceeded` is classified; accounting write failures are not. | + +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 fdb510051..c3832a26a 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 boundary stage plus the bounded failure stage/reason and safe accounting fields; `stage` names the Worker or supervisor boundary, `processing_stage` the 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 26d69af5e..3bedd65ef 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 8a6e40191..60319ff71 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 000000000..1233af44e --- /dev/null +++ b/docs/zh/development/processing-failures.md @@ -0,0 +1,74 @@ +# 处理失败诊断 + +本文是 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` 字段不同：后者表示 Worker 或 supervisor 边界 +（`worker_process`、`work_admission` 等）。计数与标识在写入日志前会被限界；未知或不受支持的值直接省略， +不做猜测。 + +## 隐私边界 + +诊断不得包含 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 | Partial | +| 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 | Partial | `UsageLimitExceeded` 已分类；账务写入失败未覆盖。 | + +第一个落地切片覆盖 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 05f5ff664..0ccd5f699 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，以及有界的失败 stage/reason 与安全计数字段；`stage` 表示 Worker/supervisor 边界，`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 9d0e694b8..14f9960c5 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 260038f51..2261d08ed 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 6b4608889..3261495df 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 9ecc8bf8d..e10dbf243 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" @@ -109,11 +115,46 @@ 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: + rate_limited = error.response.status_code == 429 + raise attach_diagnostics( + InferenceUnavailableError("embed"), + ProcessingDiagnostics( + stage=ProcessingStage.EMBEDDING, + reason=( + ProcessingFailureReason.RATE_LIMIT + if rate_limited + else ProcessingFailureReason.PROVIDER_UNAVAILABLE + ), + 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 5145c9d48..0b1946016 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 @@ -495,7 +513,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 +521,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=ProcessingStage.USAGE_ACCOUNTING, + 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 000000000..ee4e7f60d --- /dev/null +++ b/src/powercontext/builtin/processing_diagnostics.py @@ -0,0 +1,260 @@ +# 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" + 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 a35f2adf3..ce436b343 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 aacc62d86..24ae5534a 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 6fcbd261b..beced183c 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 08527a1d3..310de3683 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 901e623d4..5267525a3 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 diff --git a/tests/builtin/inference/test_pydantic_ai.py b/tests/builtin/inference/test_pydantic_ai.py index 9f53f8462..b0f30b24e 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,97 @@ 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_is_classified_as_usage_accounting() -> 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.USAGE_ACCOUNTING + assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT + assert diagnostics.cause_exception_type == "UsageLimitExceeded" + + +def test_embedding_validation_error_is_not_classified_as_output_parsing() -> None: + class Payload(BaseModel): + value: int + + mapped = None + try: + Payload.model_validate({"value": "not-an-integer"}) + except ValidationError as error: + mapped = _map_error(error, operation="embed", timeout_seconds=1.0) + + assert mapped is not None + diagnostics = classify_processing_error(mapped) + + assert diagnostics is not None + assert diagnostics.stage is ProcessingStage.EMBEDDING + assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + + +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()) diff --git a/tests/builtin/test_processing_diagnostics.py b/tests/builtin/test_processing_diagnostics.py new file mode 100644 index 000000000..f8aa0e909 --- /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 ffa412ea7..97b1e8fc0 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 From 2a92e52b89c35d0bb116fd7161a0b0d4c6e497d1 Mon Sep 17 00:00:00 2001 From: heyamei690-hash gua Date: Thu, 22 Oct 2026 17:55:51 +0800 Subject: [PATCH 2/3] fix(processing): align embedding, usage-limit, and MiniMax classification (#2189) Route a validation failure from an injected embedding model through the shared mapping so the public embed() path reports embedding/invalid_output and keeps the original exception type instead of classifying it as a transient outage. Report inference usage limits in their originating stage and reserve usage accounting for usage reporting or persistence. Derive MiniMax diagnostics from the HTTP status while keeping its public exception contract. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- docs/en/development/processing-failures.md | 4 +- docs/zh/development/processing-failures.md | 4 +- src/powercontext/builtin/inference/minimax.py | 22 +++++-- .../builtin/inference/pydantic_ai.py | 9 ++- .../inference/test_minimax_embedding.py | 32 ++++++++++ tests/builtin/inference/test_pydantic_ai.py | 64 ++++++++++++++----- 6 files changed, 108 insertions(+), 27 deletions(-) diff --git a/docs/en/development/processing-failures.md b/docs/en/development/processing-failures.md index 7c99787ac..429fa6568 100644 --- a/docs/en/development/processing-failures.md +++ b/docs/en/development/processing-failures.md @@ -43,7 +43,7 @@ in its processing path. | Family | Generation | Embedding | Output parsing | Business validation | Persistence | Usage accounting | |---|---|---|---|---|---|---| -| Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Partial | +| 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 | @@ -63,7 +63,7 @@ Memory / Atomic Memory is tracked at operation granularity while it is the activ | `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 | Partial | `UsageLimitExceeded` is classified; accounting write failures are not. | +| Usage accounting | Usage accounting | Missing | Usage-reporting and persistence failures are not classified; inference usage limits are reported in their originating generation stage, and the same reason is available when an embedding 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 diff --git a/docs/zh/development/processing-failures.md b/docs/zh/development/processing-failures.md index 1233af44e..44e246beb 100644 --- a/docs/zh/development/processing-failures.md +++ b/docs/zh/development/processing-failures.md @@ -37,7 +37,7 @@ telemetry 系统。 | Family | Generation | Embedding | Output parsing | Business validation | Persistence | Usage accounting | |---|---|---|---|---|---|---| -| Memory / Atomic Memory | Partial | Covered | Partial | Covered | Partial | Partial | +| 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 | @@ -57,7 +57,7 @@ Memory / Atomic Memory 作为当前切片，按 operation 粒度跟踪： | `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 | Partial | `UsageLimitExceeded` 已分类；账务写入失败未覆盖。 | +| Usage accounting | Usage accounting | Missing | 尚未分类账务上报与持久化失败；推理用量限制归入其来源的 generation 阶段，embedding adapter 上报同一限制时使用相同 reason。 | 第一个落地切片覆盖 Memory / Atomic Memory：provider 失败、预算耗尽、提取/对账校验均有分类，安全计数字段 可从子 Worker 传播到 supervisor 日志。其他 Family 在各自切片落地前保持现有行为；每次切片在同一变更中更新 diff --git a/src/powercontext/builtin/inference/minimax.py b/src/powercontext/builtin/inference/minimax.py index e10dbf243..9e541361a 100644 --- a/src/powercontext/builtin/inference/minimax.py +++ b/src/powercontext/builtin/inference/minimax.py @@ -100,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: @@ -133,16 +135,22 @@ async def _embed(self, texts: tuple[str, ...], *, embedding_type: str) -> Embedd ), ) from error except httpx.HTTPStatusError as error: - rate_limited = error.response.status_code == 429 + 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=( - ProcessingFailureReason.RATE_LIMIT - if rate_limited - else ProcessingFailureReason.PROVIDER_UNAVAILABLE - ), + reason=reason, cause_exception_type=type(error).__name__, ), ) from error diff --git a/src/powercontext/builtin/inference/pydantic_ai.py b/src/powercontext/builtin/inference/pydantic_ai.py index 0b1946016..42e5cb0dd 100644 --- a/src/powercontext/builtin/inference/pydantic_ai.py +++ b/src/powercontext/builtin/inference/pydantic_ai.py @@ -402,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: @@ -587,7 +594,7 @@ def _map_error( # noqa: C901 - bounded mapping across status codes and provider return attach_diagnostics( InvalidInferenceOutputError(operation, "provider did not return a valid result"), ProcessingDiagnostics( - stage=ProcessingStage.USAGE_ACCOUNTING, + stage=stage or ProcessingStage.GENERATION, reason=ProcessingFailureReason.USAGE_LIMIT, failure_location="inference-limits", cause_exception_type=cause_type, diff --git a/tests/builtin/inference/test_minimax_embedding.py b/tests/builtin/inference/test_minimax_embedding.py index 5267525a3..ea518c5d5 100644 --- a/tests/builtin/inference/test_minimax_embedding.py +++ b/tests/builtin/inference/test_minimax_embedding.py @@ -317,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 b0f30b24e..14dde67a3 100644 --- a/tests/builtin/inference/test_pydantic_ai.py +++ b/tests/builtin/inference/test_pydantic_ai.py @@ -1012,34 +1012,42 @@ async def scenario() -> None: asyncio.run(scenario()) -def test_usage_limit_is_classified_as_usage_accounting() -> None: +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.USAGE_ACCOUNTING + assert diagnostics.stage is ProcessingStage.GENERATION assert diagnostics.reason is ProcessingFailureReason.USAGE_LIMIT assert diagnostics.cause_exception_type == "UsageLimitExceeded" -def test_embedding_validation_error_is_not_classified_as_output_parsing() -> None: - class Payload(BaseModel): - value: int +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)) - mapped = None - try: - Payload.model_validate({"value": "not-an-integer"}) - except ValidationError as error: - mapped = _map_error(error, operation="embed", timeout_seconds=1.0) + 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), + ) - assert mapped is not None - diagnostics = classify_processing_error(mapped) + with pytest.raises(InvalidInferenceOutputError) as error: + await generator.generate(Question("bounded evidence")) - assert diagnostics is not None - assert diagnostics.stage is ProcessingStage.EMBEDDING - assert diagnostics.reason is ProcessingFailureReason.INVALID_OUTPUT + 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: @@ -1079,3 +1087,29 @@ async def scenario() -> 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()) From 70f6aa7ead82623f0a8bcb8ecb58f68713b3ec5e Mon Sep 17 00:00:00 2001 From: heyamei690-hash gua Date: Thu, 22 Oct 2026 18:21:30 +0800 Subject: [PATCH 3/3] docs(processing): complete the stage domain and usage-limit wording (#2189) Describe every stage value the log can carry, mark usage accounting as reserved, and report usage limits in their originating generation or embedding stage. Add the public embedding usage-limit regression test and rebase the branch onto the current master. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- docs/en/development/processing-failures.md | 15 +++++++++------ docs/en/docs/operate/observability.md | 2 +- docs/zh/development/processing-failures.md | 11 +++++++---- docs/zh/docs/operate/observability.md | 2 +- .../builtin/processing_diagnostics.py | 1 + tests/builtin/inference/test_pydantic_ai.py | 19 +++++++++++++++++++ 6 files changed, 38 insertions(+), 12 deletions(-) diff --git a/docs/en/development/processing-failures.md b/docs/en/development/processing-failures.md index 429fa6568..f594b26b5 100644 --- a/docs/en/development/processing-failures.md +++ b/docs/en/development/processing-failures.md @@ -11,7 +11,7 @@ Every processing failure keeps the existing identity and outcome fields (`family | Field | Meaning | Values | |---|---|---| -| `processing_stage` | The stage that owned the failure | `generation`, `embedding`, `output_parsing`, `business_validation`, `persistence`, `usage_accounting` | +| `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 | @@ -22,10 +22,13 @@ Every processing failure keeps the existing identity and outcome fields (`family | `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, which names the Worker or -supervisor boundary (`worker_process`, `work_admission`, and so on). Counts and identifiers are -bounded before they reach a log record, and unknown or unsupported values are omitted rather than -guessed. +`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 @@ -63,7 +66,7 @@ Memory / Atomic Memory is tracked at operation granularity while it is the activ | `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 generation stage, and the same reason is available when an embedding adapter reports the limit. | +| 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 diff --git a/docs/en/docs/operate/observability.md b/docs/en/docs/operate/observability.md index c3832a26a..9f226378b 100644 --- a/docs/en/docs/operate/observability.md +++ b/docs/en/docs/operate/observability.md @@ -34,7 +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 boundary stage plus the bounded failure stage/reason and safe accounting fields; `stage` names the Worker or supervisor boundary, `processing_stage` the domain stage. See [Processing failure diagnostics](../../development/processing-failures.md) | +| `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/processing-failures.md b/docs/zh/development/processing-failures.md index 44e246beb..eb419678e 100644 --- a/docs/zh/development/processing-failures.md +++ b/docs/zh/development/processing-failures.md @@ -10,7 +10,7 @@ telemetry 系统。 | 字段 | 含义 | 取值 | |---|---|---| -| `processing_stage` | 失败所属的处理阶段 | `generation`、`embedding`、`output_parsing`、`business_validation`、`persistence`、`usage_accounting` | +| `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 数 | 非负整数 | @@ -21,8 +21,11 @@ telemetry 系统。 | `failure_location` | 稳定的内部位置或操作标识 | 有界字符串 | | `cause_exception_type` | 映射前的原始 provider/库异常类型 | 有界字符串 | -`processing_stage` 与既有的 `stage` 字段不同：后者表示 Worker 或 supervisor 边界 -（`worker_process`、`work_admission` 等）。计数与标识在写入日志前会被限界；未知或不受支持的值直接省略， +`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` 表示有界的领域阶段。计数与标识在写入日志前会被限界；未知或不受支持的值直接省略， 不做猜测。 ## 隐私边界 @@ -57,7 +60,7 @@ Memory / Atomic Memory 作为当前切片，按 operation 粒度跟踪： | `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 上报同一限制时使用相同 reason。 | +| Usage accounting | Usage accounting | Missing | 尚未分类账务上报与持久化失败；推理用量限制归入其来源阶段（generation，或 embedding adapter 上报时归入 embedding）。 | 第一个落地切片覆盖 Memory / Atomic Memory：provider 失败、预算耗尽、提取/对账校验均有分类，安全计数字段 可从子 Worker 传播到 supervisor 日志。其他 Family 在各自切片落地前保持现有行为；每次切片在同一变更中更新 diff --git a/docs/zh/docs/operate/observability.md b/docs/zh/docs/operate/observability.md index 0ccd5f699..5d0f4a95c 100644 --- a/docs/zh/docs/operate/observability.md +++ b/docs/zh/docs/operate/observability.md @@ -32,7 +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，以及有界的失败 stage/reason 与安全计数字段；`stage` 表示 Worker/supervisor 边界，`processing_stage` 表示领域阶段。详见 [处理失败诊断](../../development/processing-failures.md) | +| `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/processing_diagnostics.py b/src/powercontext/builtin/processing_diagnostics.py index ee4e7f60d..056dd3895 100644 --- a/src/powercontext/builtin/processing_diagnostics.py +++ b/src/powercontext/builtin/processing_diagnostics.py @@ -31,6 +31,7 @@ class ProcessingStage(StrEnum): OUTPUT_PARSING = "output_parsing" BUSINESS_VALIDATION = "business_validation" PERSISTENCE = "persistence" + # Reserved for usage reporting or persistence failures. USAGE_ACCOUNTING = "usage_accounting" diff --git a/tests/builtin/inference/test_pydantic_ai.py b/tests/builtin/inference/test_pydantic_ai.py index 14dde67a3..af2a8760f 100644 --- a/tests/builtin/inference/test_pydantic_ai.py +++ b/tests/builtin/inference/test_pydantic_ai.py @@ -1113,3 +1113,22 @@ async def scenario() -> None: 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())