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