fix(validation): 固化候选验收检查点并保持局部失败隔离

让确定性质量规则同时支持 Trace 和 Candidate subject，并按实际 Validation scope 筛选执行，单一 scope 的失败不再污染其他 scope。

在根候选账本持久化 scope 级 checkpoint，恢复时复用材料解析、确定性结果与 LLM 结果；验证材料和调用预算通过稳定 operation ID 幂等记账，避免发布阶段崩溃导致重复消耗。

补充已采用后代候选的祖先回溯屏障，以及发布故障、预算幂等和 scope 隔离回归测试。

cyber_agent/application/candidate.py

@@ -131,6 +131,7 @@ class CandidateLedger(ApplicationModel):
     lifecycle: tuple[CandidateLifecycleRecord, ...] = ()
     operations: tuple[CandidateOperationRecord, ...] = ()
     validations: tuple[dict[str, Any], ...] = ()
+    validation_checkpoints: tuple[dict[str, Any], ...] = ()
 
     @model_validator(mode="after")
     def validate_ledger(self) -> "CandidateLedger":
@@ -147,6 +148,18 @@ class CandidateLedger(ApplicationModel):
         operation_ids = [item.operation_id for item in self.operations]
         if len(operation_ids) != len(set(operation_ids)):
             raise ValueError("Candidate ledger contains duplicate operation IDs")
+        from cyber_agent.application.quality import CandidateValidationCheckpoint
+
+        checkpoint_keys = []
+        for raw in self.validation_checkpoints:
+            checkpoint = CandidateValidationCheckpoint.model_validate(raw)
+            checkpoint_keys.append((
+                checkpoint.candidate_ref.candidate_id,
+                checkpoint.candidate_ref.revision,
+                checkpoint.plan_hash,
+            ))
+        if len(checkpoint_keys) != len(set(checkpoint_keys)):
+            raise ValueError("Candidate ledger contains duplicate validation checkpoints")
         known = set(identities)
         parents_by_candidate: dict[
             tuple[str, int], tuple[tuple[str, int], ...]

cyber_agent/application/candidate_service.py

@@ -20,7 +20,10 @@ from cyber_agent.application.candidate import (
     CandidateVersionRequest,
 )
 from cyber_agent.application.models import canonical_json
-from cyber_agent.application.quality import CandidateValidationRecord
+from cyber_agent.application.quality import (
+    CandidateValidationCheckpoint,
+    CandidateValidationRecord,
+)
 from cyber_agent.core.agent_mode import require_mutable_trace_policy
 from cyber_agent.core.artifacts import ArtifactRef, ArtifactResolver, resolve_artifact_refs
 from cyber_agent.core.task_protocol import ensure_task_protocol, task_contract_ref
@@ -268,6 +271,89 @@ class CandidateService:
                     return record
         return None
 
+    async def begin_validation_checkpoint(
+        self,
+        trace_id: str,
+        candidate_ref: CandidateRef,
+        *,
+        plan: dict[str, Any],
+        plan_hash: str,
+        validated_at_sequence: int,
+        material_usage_recorded: bool,
+    ) -> CandidateValidationCheckpoint:
+        trace, root = await self._require_bound_trace(trace_id)
+        async with self._lock_for(root.trace_id):
+            ledger = await self._load_ledger(root.trace_id)
+            persisted = ledger.candidate(candidate_ref)
+            if persisted != candidate_ref:
+                raise CandidateStateError(
+                    "Candidate validation references an altered CandidateRef"
+                )
+            self._validate_candidate_owner(persisted, trace, root)
+            for raw in ledger.validation_checkpoints:
+                checkpoint = CandidateValidationCheckpoint.model_validate(raw)
+                if (
+                    checkpoint.candidate_ref == candidate_ref
+                    and checkpoint.plan_hash == plan_hash
+                ):
+                    return checkpoint
+            checkpoint = CandidateValidationCheckpoint(
+                candidate_ref=candidate_ref,
+                plan_hash=plan_hash,
+                validation_plan=plan,
+                validated_at_sequence=validated_at_sequence,
+                material_usage_recorded=material_usage_recorded,
+            )
+            updated = ledger.model_copy(update={
+                "validation_checkpoints": (
+                    *ledger.validation_checkpoints,
+                    checkpoint.model_dump(mode="json"),
+                ),
+            })
+            await self.store.replace_candidate_ledger(
+                root.trace_id,
+                updated.model_dump(mode="json"),
+            )
+            return checkpoint
+
+    async def update_validation_checkpoint(
+        self,
+        trace_id: str,
+        candidate_ref: CandidateRef,
+        plan_hash: str,
+        **updates: Any,
+    ) -> CandidateValidationCheckpoint:
+        trace, root = await self._require_bound_trace(trace_id)
+        async with self._lock_for(root.trace_id):
+            ledger = await self._load_ledger(root.trace_id)
+            persisted = ledger.candidate(candidate_ref)
+            if persisted != candidate_ref:
+                raise CandidateStateError(
+                    "Candidate validation references an altered CandidateRef"
+                )
+            self._validate_candidate_owner(persisted, trace, root)
+            checkpoints = []
+            result = None
+            for raw in ledger.validation_checkpoints:
+                checkpoint = CandidateValidationCheckpoint.model_validate(raw)
+                if (
+                    checkpoint.candidate_ref == candidate_ref
+                    and checkpoint.plan_hash == plan_hash
+                ):
+                    checkpoint = checkpoint.model_copy(update=updates)
+                    result = checkpoint
+                checkpoints.append(checkpoint.model_dump(mode="json"))
+            if result is None:
+                raise CandidateStateError("Candidate validation checkpoint not found")
+            updated = ledger.model_copy(update={
+                "validation_checkpoints": tuple(checkpoints),
+            })
+            await self.store.replace_candidate_ledger(
+                root.trace_id,
+                updated.model_dump(mode="json"),
+            )
+            return result
+
     async def record_validation(
         self,
         trace_id: str,
@@ -297,6 +383,22 @@ class CandidateService:
                     record.model_dump(mode="json"),
                 ),
             })
+            checkpoints = []
+            for raw in updated.validation_checkpoints:
+                checkpoint = CandidateValidationCheckpoint.model_validate(raw)
+                if (
+                    checkpoint.candidate_ref == record.candidate_ref
+                    and checkpoint.plan_hash == record.plan_hash
+                ):
+                    result = record.validation_result
+                    checkpoint = checkpoint.model_copy(update={
+                        "scope_results": tuple(result.get("scope_results", [])),
+                        "aggregate_result": result,
+                    })
+                checkpoints.append(checkpoint.model_dump(mode="json"))
+            updated = updated.model_copy(update={
+                "validation_checkpoints": tuple(checkpoints),
+            })
             await self.store.replace_candidate_ledger(
                 root.trace_id,
                 updated.model_dump(mode="json"),
@@ -643,6 +745,18 @@ class CandidateService:
                 if lifecycle.state != "adopted":
                     continue
                 candidate = ledger.candidate(lifecycle.candidate)
+                ancestor_ids: set[str] = set()
+                for start_trace_id in {
+                    lifecycle.source_trace_id,
+                    candidate.owner_trace_id,
+                }:
+                    current_trace_id = start_trace_id
+                    while current_trace_id and current_trace_id not in ancestor_ids:
+                        ancestor_ids.add(current_trace_id)
+                        current = await self.store.get_trace(current_trace_id)
+                        current_trace_id = (
+                            current.parent_trace_id if current is not None else None
+                        )
                 crosses_review = (
                     lifecycle.source_trace_id == trace_id
                     and lifecycle.effective_at_sequence > after_sequence
@@ -651,7 +765,14 @@ class CandidateService:
                     candidate.owner_trace_id == trace_id
                     and candidate.created_at_sequence > after_sequence
                 )
-                if crosses_review or crosses_version:
+                incomparable_ancestor = (
+                    trace_id in ancestor_ids
+                    and trace_id not in {
+                        lifecycle.source_trace_id,
+                        candidate.owner_trace_id,
+                    }
+                )
+                if crosses_review or crosses_version or incomparable_ancestor:
                     raise CandidateStateError(
                         "Cannot rewind across a committed candidate adoption"
                     )
@@ -765,6 +886,7 @@ class CandidateService:
             lifecycle=lifecycle,
             operations=operations,
             validations=ledger.validations,
+            validation_checkpoints=ledger.validation_checkpoints,
         )
 
     @staticmethod

cyber_agent/application/models.py

@@ -97,6 +97,7 @@ class QualityRuleSpec(ApplicationModel):
     rule_id: str = Field(pattern=r"^[a-z][a-z0-9_.-]{2,100}$")
     version: str = Field(min_length=1, max_length=100)
     scope: ValidationScopeName = "output"
+    subject_types: tuple[Literal["trace", "candidate"], ...] = ("candidate",)
     criterion: str = Field(min_length=1, max_length=2_000)
     config: dict[str, Any] = Field(default_factory=dict)
     timeout_seconds: float = Field(default=10.0, gt=0, le=120)
@@ -104,6 +105,10 @@ class QualityRuleSpec(ApplicationModel):
     @model_validator(mode="after")
     def validate_config(self) -> "QualityRuleSpec":
         canonical_json(self.config)
+        if not self.subject_types or len(self.subject_types) != len(
+            set(self.subject_types)
+        ):
+            raise ValueError("subject_types must be non-empty and unique")
         return self
 
 
@@ -219,10 +224,6 @@ class AgentApplication(ApplicationModel):
             raise ValueError("quality_rules must contain unique rule_id values")
         if self.quality_rules and not self.quality_provider_ref:
             raise ValueError("quality_rules require quality_provider_ref")
-        if any(item.scope != "output" for item in self.quality_rules):
-            raise ValueError(
-                "M3B quality_rules currently validate candidate output only"
-            )
         return self
 
     def role(self, role_id: str) -> AgentRole:

cyber_agent/application/quality.py

@@ -26,21 +26,31 @@ class ValidationSubject(ApplicationModel):
 class QualityCheckInput(ApplicationModel):
     subject: ValidationSubject
     rules: tuple[QualityRuleSpec, ...] = Field(min_length=1)
-    material: ValidationMaterial
+    materials: tuple[ValidationMaterial, ...] = Field(min_length=1)
 
     @model_validator(mode="after")
     def validate_material(self) -> "QualityCheckInput":
-        if self.subject.subject_type != "candidate" or self.subject.candidate_ref is None:
-            raise ValueError("quality checks currently require a candidate subject")
-        ref = self.subject.candidate_ref.artifact_ref
-        if (
-            self.material.artifact_id != ref.artifact_id
-            or self.material.version != ref.version
-            or self.material.content_hash != ref.content_hash
-        ):
-            raise ValueError("quality material does not match the CandidateRef")
+        if any(self.subject.subject_type not in rule.subject_types for rule in self.rules):
+            raise ValueError("quality rule does not support this subject type")
+        if self.subject.subject_type == "candidate":
+            assert self.subject.candidate_ref is not None
+            ref = self.subject.candidate_ref.artifact_ref
+            if len(self.materials) != 1:
+                raise ValueError("candidate quality checks require one exact material")
+            material = self.materials[0]
+            if (
+                material.artifact_id != ref.artifact_id
+                or material.version != ref.version
+                or material.content_hash != ref.content_hash
+            ):
+                raise ValueError("quality material does not match the CandidateRef")
         return self
 
+    @property
+    def material(self) -> ValidationMaterial:
+        """Compatibility accessor for providers that consume one primary material."""
+        return self.materials[0]
+
 
 class QualityCheckOutcome(ApplicationModel):
     rule_id: str = Field(min_length=1, max_length=100)
@@ -82,3 +92,19 @@ class CandidateValidationRecord(ApplicationModel):
         if result.plan_hash != self.plan_hash:
             raise ValueError("candidate validation result plan hash mismatch")
         return self
+
+
+class CandidateValidationCheckpoint(ApplicationModel):
+    """Durable per-scope candidate validation state before final aggregation."""
+
+    schema_version: Literal[1] = 1
+    candidate_ref: CandidateRef
+    plan_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    validation_plan: dict[str, Any]
+    scope_results: tuple[dict[str, Any], ...] = ()
+    fixed_checks: tuple[dict[str, Any], ...] = ()
+    fixed_scope_errors: dict[str, str] = Field(default_factory=dict)
+    quality_completed: bool = False
+    material_usage_recorded: bool = False
+    aggregate_result: dict[str, Any] | None = None
+    validated_at_sequence: int = Field(ge=0)

cyber_agent/core/resource_budget.py

@@ -8,7 +8,7 @@ Runner 的模型调用、Sub-Agent 创建和 Validator 沿用同一快照。动
 from __future__ import annotations
 
 import asyncio
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 import math
 import os
@@ -222,6 +222,7 @@ class ResourceUsage:
     started_at: str
     last_denial: dict[str, Any] | None = None
     exhausted_reason: str | None = None
+    validation_operation_ids: list[str] = field(default_factory=list)
 
     @classmethod
     def new(cls, *, total_agents: int = 1, started_at: datetime | None = None) -> "ResourceUsage":
@@ -244,6 +245,8 @@ class ResourceUsage:
         if not isinstance(value, Mapping):
             raise ResourceBudgetStateError("resource usage must be an object")
         expected = {field.name for field in cls.__dataclass_fields__.values()}
+        value = dict(value)
+        value.setdefault("validation_operation_ids", [])
         unknown = set(value) - expected
         missing = expected - set(value)
         if unknown or missing:
@@ -252,7 +255,7 @@ class ResourceUsage:
                 f"missing={sorted(missing)}, unknown={sorted(unknown)}"
             )
         try:
-            usage = cls(**dict(value))
+            usage = cls(**value)
             usage.validate()
             return usage
         except (TypeError, ValueError) as exc:
@@ -311,6 +314,16 @@ class ResourceUsage:
         }
         if self.exhausted_reason is not None and self.exhausted_reason not in valid_reasons:
             raise ValueError("exhausted_reason is invalid")
+        if (
+            not isinstance(self.validation_operation_ids, list)
+            or len(self.validation_operation_ids)
+            != len(set(self.validation_operation_ids))
+            or any(
+                not isinstance(item, str) or not item or len(item) > 500
+                for item in self.validation_operation_ids
+            )
+        ):
+            raise ValueError("validation_operation_ids is invalid")
 
     def to_dict(self) -> dict[str, Any]:
         self.validate()
@@ -567,6 +580,7 @@ class ResourceBudgetController:
         tool_calls: int = 0,
         material_chars: int = 0,
         provider_cost_usd: float = 0.0,
+        operation_id: str | None = None,
     ) -> ResourceUsage:
         """原子预留 Validator 工具调用或登记可信材料字符。
 
@@ -588,8 +602,16 @@ class ResourceBudgetController:
             raise ValueError("provider_cost_usd must be a non-negative number")
         if not tool_calls and not material_chars and not provider_cost_usd:
             return await self.get_usage(root_trace_id)
+        if operation_id is not None and (
+            not isinstance(operation_id, str)
+            or not operation_id
+            or len(operation_id) > 500
+        ):
+            raise ValueError("operation_id must be a non-empty string up to 500 chars")
         async with self._lock(root_trace_id):
             usage = await self.get_usage(root_trace_id)
+            if operation_id is not None and operation_id in usage.validation_operation_ids:
+                return usage
             self._raise_if_terminal(usage, budget)
             await self._check_time_locked(root_trace_id, usage, budget)
             proposed_calls = usage.validation_tool_calls + tool_calls
@@ -604,6 +626,8 @@ class ResourceBudgetController:
             usage.validation_tool_calls = proposed_calls
             usage.validation_material_chars += material_chars
             usage.total_cost_usd += float(provider_cost_usd)
+            if operation_id is not None:
+                usage.validation_operation_ids.append(operation_id)
             exceeded: BudgetDimension | None = None
             if (
                 budget.enabled

cyber_agent/core/runner.py

@@ -113,6 +113,7 @@ from cyber_agent.core.artifacts import (
     ValidationMaterial,
     extract_artifact_refs,
     material_chars,
+    material_content_hash,
     resolve_artifact_refs,
 )
 from cyber_agent.core.validation import (
@@ -505,6 +506,7 @@ class AgentRunner:
         tool_calls: int = 0,
         material_chars_count: int = 0,
         provider_cost_usd: float = 0.0,
+        operation_id: str | None = None,
     ) -> None:
         """把 Validator网页工具和真实材料字符计入同一棵树的预算。"""
         resolved = await self._resource_budget_for_trace(trace_id)
@@ -519,6 +521,7 @@ class AgentRunner:
             tool_calls=tool_calls,
             material_chars=material_chars_count,
             provider_cost_usd=provider_cost_usd,
+            operation_id=operation_id,
         )
 
     async def validate_recursive_trace(
@@ -693,17 +696,51 @@ class AgentRunner:
         fixed_checks: dict[str, ValidationCheck] = {}
         fixed_scope_errors: dict[ValidationScope, str] = {}
         quality_rules: tuple[Any, ...] = ()
-        if candidate_ref is not None:
+        quality_materials: tuple[ValidationMaterial, ...] = ()
+        if self.application_binding is not None:
             from cyber_agent.application.quality import ValidationSubject
 
+            brief_payload = (
+                task_brief.model_dump(mode="json")
+                if hasattr(task_brief, "model_dump")
+                else (task_brief or {})
+            )
+            requested_quality_scopes = (
+                {"output"}
+                if candidate_ref is not None
+                else set(policy.effective_scopes(
+                    brief_payload.get("validation_scopes", []),
+                    root=root_validator,
+                ))
+            )
             validation_subject = ValidationSubject(
-                subject_type="candidate",
+                subject_type=("candidate" if candidate_ref is not None else "trace"),
                 trace_id=evaluated_trace_id,
                 candidate_ref=candidate_ref,
             )
             quality_rules = tuple(
-                self.application_binding.application.quality_rules
+                rule
+                for rule in self.application_binding.application.quality_rules
+                if validation_subject.subject_type in rule.subject_types
+                and rule.scope in requested_quality_scopes
             )
+            if candidate_ref is not None and candidate_material is not None:
+                quality_materials = (candidate_material,)
+            elif candidate_ref is None:
+                quality_materials = tuple(materials)
+                if isinstance(task_report, dict):
+                    report_material = ValidationMaterial(
+                        artifact_id=f"trace:{evaluated_trace_id}:task_report",
+                        version=str(progress_revision or 0),
+                        content_hash=material_content_hash(task_report),
+                        kind="framework.task_report",
+                        mime_type="application/json",
+                        root_trace_id=root_trace_id,
+                        uid=evaluated.uid,
+                        content=task_report,
+                    )
+                    quality_materials = (*quality_materials, report_material)
+                    total_material_chars += material_chars(report_material)
             quality_specs = [
                 ValidationCheckSpec(
                     check_id=f"quality.{rule.rule_id}",
@@ -753,12 +790,6 @@ class AgentRunner:
                     cached=True,
                 )
             cached = None
-            if quality_rules and candidate_material is not None:
-                fixed_checks, fixed_scope_errors = await self._run_quality_checks(
-                    validation_subject,
-                    candidate_material,
-                    quality_rules,
-                )
         validation_cache: dict[str, Any]
         resume_scope_results: list[ScopeValidationResult] = []
         if isinstance(cached, dict) and cached.get("plan_hash") == plan.plan_hash:
@@ -801,6 +832,77 @@ class AgentRunner:
                     context=evaluated.context,
                 )
 
+        candidate_checkpoint = None
+        if candidate_ref is not None:
+            candidate_checkpoint = await self.candidate_service.begin_validation_checkpoint(
+                evaluated_trace_id,
+                candidate_ref,
+                plan=plan.model_dump(mode="json"),
+                plan_hash=plan.plan_hash,
+                validated_at_sequence=plan.evaluated_head_sequence,
+                material_usage_recorded=total_material_chars == 0,
+            )
+            validation_cache = {
+                "validation_plan": candidate_checkpoint.validation_plan,
+                "plan_hash": candidate_checkpoint.plan_hash,
+                "scope_results": list(candidate_checkpoint.scope_results),
+                "aggregate_result": candidate_checkpoint.aggregate_result,
+                "validated_at_sequence": candidate_checkpoint.validated_at_sequence,
+                "material_usage_recorded": (
+                    candidate_checkpoint.material_usage_recorded
+                ),
+            }
+            resume_scope_results = [
+                ScopeValidationResult.model_validate(item)
+                for item in candidate_checkpoint.scope_results
+            ]
+            if quality_rules:
+                if candidate_checkpoint.quality_completed:
+                    fixed_checks = {
+                        item["check_id"]: ValidationCheck.model_validate(item)
+                        for item in candidate_checkpoint.fixed_checks
+                    }
+                    fixed_scope_errors = {
+                        key: value
+                        for key, value in candidate_checkpoint.fixed_scope_errors.items()
+                    }
+                else:
+                    fixed_checks, fixed_scope_errors = await self._run_quality_checks(
+                        validation_subject,
+                        quality_materials,
+                        quality_rules,
+                        usage_operation_prefix=(
+                            f"candidate-validation:{candidate_ref.candidate_id}:"
+                            f"{candidate_ref.revision}:{plan.plan_hash}"
+                        ),
+                    )
+                    candidate_checkpoint = (
+                        await self.candidate_service.update_validation_checkpoint(
+                            evaluated_trace_id,
+                            candidate_ref,
+                            plan.plan_hash,
+                            fixed_checks=tuple(
+                                item.model_dump(mode="json")
+                                for item in fixed_checks.values()
+                            ),
+                            fixed_scope_errors={
+                                key: value
+                                for key, value in fixed_scope_errors.items()
+                            },
+                            quality_completed=True,
+                        )
+                    )
+
+        if candidate_ref is None and quality_rules:
+            fixed_checks, fixed_scope_errors = await self._run_quality_checks(
+                validation_subject,
+                quality_materials,
+                quality_rules,
+                usage_operation_prefix=(
+                    f"trace-validation:{evaluated_trace_id}:{plan.plan_hash}"
+                ),
+            )
+
         if (
             total_material_chars
             and not validation_cache.get("material_usage_recorded", False)
@@ -808,9 +910,20 @@ class AgentRunner:
             await self.record_recursive_validation_usage(
                 evaluated_trace_id,
                 material_chars_count=total_material_chars,
+                operation_id=(
+                    f"validation-materials:{evaluated_trace_id}:{plan.plan_hash}"
+                ),
             )
             if candidate_ref is not None:
                 validation_cache["material_usage_recorded"] = True
+                candidate_checkpoint = (
+                    await self.candidate_service.update_validation_checkpoint(
+                        evaluated_trace_id,
+                        candidate_ref,
+                        plan.plan_hash,
+                        material_usage_recorded=True,
+                    )
+                )
             else:
                 fresh = await self.trace_store.get_trace(evaluated_trace_id)
                 if not fresh:
@@ -917,6 +1030,12 @@ class AgentRunner:
                         for item in plan.effective_scopes
                         if item in by_scope
                     ]
+                    await self.candidate_service.update_validation_checkpoint(
+                        evaluated_trace_id,
+                        candidate_ref,
+                        plan.plan_hash,
+                        scope_results=tuple(validation_cache["scope_results"]),
+                    )
                     return
                 fresh = await self.trace_store.get_trace(evaluated_trace_id)
                 if not fresh:
@@ -1009,8 +1128,30 @@ class AgentRunner:
     async def _run_quality_checks(
         self,
         subject: Any,
-        material: ValidationMaterial,
+        materials: tuple[ValidationMaterial, ...],
         rules: tuple[Any, ...],
+        usage_operation_prefix: str,
+    ) -> tuple[dict[str, ValidationCheck], dict[ValidationScope, str]]:
+        checks: dict[str, ValidationCheck] = {}
+        errors: dict[ValidationScope, str] = {}
+        for scope in dict.fromkeys(rule.scope for rule in rules):
+            scoped_rules = tuple(rule for rule in rules if rule.scope == scope)
+            scoped_checks, scoped_errors = await self._run_quality_check_batch(
+                subject,
+                materials,
+                scoped_rules,
+                usage_operation_id=f"{usage_operation_prefix}:quality:{scope}",
+            )
+            checks.update(scoped_checks)
+            errors.update(scoped_errors)
+        return checks, errors
+
+    async def _run_quality_check_batch(
+        self,
+        subject: Any,
+        materials: tuple[ValidationMaterial, ...],
+        rules: tuple[Any, ...],
+        usage_operation_id: str,
     ) -> tuple[dict[str, ValidationCheck], dict[ValidationScope, str]]:
         """Execute one frozen quality batch and reject provider-shaped plans."""
         from cyber_agent.application.quality import (
@@ -1031,15 +1172,17 @@ class AgentRunner:
                     )
                     for check_id in check_ids.values()
                 },
-                {"output": reason},
+                {rule.scope: reason for rule in rules},
             )
 
         if provider is None:
             return provider_error("QualityCheckProvider is unavailable")
+        if not materials:
+            return provider_error("QualityCheckProvider has no authorized material")
         request = QualityCheckInput(
             subject=subject,
             rules=rules,
-            material=material,
+            materials=materials,
         )
         try:
             raw = await asyncio.wait_for(
@@ -1076,6 +1219,7 @@ class AgentRunner:
             await self.record_recursive_validation_usage(
                 subject.trace_id,
                 tool_calls=len(rules),
+                operation_id=usage_operation_id,
             )
         except Exception as exc:
             return provider_error(f"Quality validation budget failed: {exc}")

tests/test_candidate_quality.py

@@ -19,6 +19,7 @@ from cyber_agent.core.artifacts import (
     ValidationMaterial,
     material_content_hash,
 )
+from cyber_agent.core.task_protocol import ensure_task_protocol
 from cyber_agent.tools.registry import ToolRegistry
 from cyber_agent.trace.store import FileSystemTraceStore
 
@@ -73,7 +74,10 @@ class QualityProvider:
         self.calls += 1
         outcomes = []
         for rule in request.rules:
-            failed = "{{placeholder}}" in request.material.content["text"]
+            failed = "{{placeholder}}" in json.dumps(
+                request.material.content,
+                ensure_ascii=False,
+            )
             outcomes.append(QualityCheckOutcome(
                 rule_id=rule.rule_id,
                 rule_version=rule.version,
@@ -124,6 +128,7 @@ class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
                 rule_id="no_placeholder",
                 version="1",
                 criterion="Candidate contains no placeholder text",
+                subject_types=("trace", "candidate"),
             ),),
         )
         registry = ApplicationRegistry()
@@ -154,6 +159,7 @@ class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
                             "issue": None,
                         }
                         for item in packet["validation_plan"]["checks"]
+                        if item["scope"] == scope
                     ],
                     "reason": "candidate output is complete",
                     "retry_from": None,
@@ -194,6 +200,14 @@ class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
             effective_at_sequence=sequence,
         )
 
+    async def _bind_current_progress_to_report(self):
+        trace = await self.store.get_trace(self.trace.trace_id)
+        state = ensure_task_protocol(trace.context)
+        state["task_report_progress_revision"] = state[
+            "task_progress_head_revision"
+        ]
+        await self.store.update_trace(trace.trace_id, context=trace.context)
+
     async def test_deterministic_and_llm_checks_are_separate_and_cached(self):
         candidate = await self._candidate("finished", 2)
         first = await self.runner.validate_recursive_trace(
@@ -266,3 +280,116 @@ class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
                     item.status == "unknown" for item in result.checks
                 ))
         self.assertEqual(0, self.llm_calls)
+
+    async def test_trace_quality_failure_only_short_circuits_its_output_scope(self):
+        await self._bind_current_progress_to_report()
+        run = await self.runner.validate_recursive_trace(
+            self.trace.trace_id,
+            task_brief={
+                "objective": "produce checked copy",
+                "reason": "the root needs a complete output",
+                "completion_criteria": ["copy is complete"],
+                "expected_outputs": ["copy"],
+                "validation_scopes": ["output"],
+            },
+            task_report={
+                "summary": "draft",
+                "outputs": [{"text": "{{placeholder}}"}],
+                "evidence": [],
+            },
+        )
+        by_scope = {item.scope: item for item in run.result.scope_results}
+        self.assertEqual("failed", by_scope["output"].outcome)
+        self.assertEqual("passed", by_scope["task"].outcome)
+        self.assertEqual(1, self.quality.calls)
+        self.assertEqual(1, self.llm_calls)
+
+    async def test_trace_rule_is_skipped_when_scope_is_not_effective(self):
+        await self._bind_current_progress_to_report()
+        run = await self.runner.validate_recursive_trace(
+            self.trace.trace_id,
+            scope="task",
+            task_report={"summary": "{{placeholder}}"},
+        )
+        self.assertEqual("passed", run.result.outcome)
+        self.assertEqual(0, self.quality.calls)
+        self.assertEqual(1, self.llm_calls)
+
+    async def test_candidate_scope_checkpoint_survives_final_publish_crash(self):
+        candidate = await self._candidate("finished", 2)
+        original = self.runner.candidate_service.record_validation
+        attempts = 0
+
+        async def fail_final_publish_once(*args, **kwargs):
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                raise OSError("final candidate validation publish failed")
+            return await original(*args, **kwargs)
+
+        with patch.object(
+            self.runner.candidate_service,
+            "record_validation",
+            side_effect=fail_final_publish_once,
+        ):
+            with self.assertRaisesRegex(OSError, "final candidate validation"):
+                await self.runner.validate_recursive_trace(
+                    self.trace.trace_id,
+                    candidate_ref=candidate,
+                )
+            recovered = await self.runner.validate_recursive_trace(
+                self.trace.trace_id,
+                candidate_ref=candidate,
+            )
+
+        self.assertEqual("passed", recovered.result.outcome)
+        self.assertEqual(1, self.quality.calls)
+        self.assertEqual(1, self.llm_calls)
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger(self.trace.trace_id)
+        )
+        self.assertEqual(1, len(ledger.validation_checkpoints))
+        self.assertEqual(1, len(ledger.validations))
+
+    async def test_candidate_material_usage_is_idempotent_across_checkpoint_crash(self):
+        candidate = await self._candidate("finished", 2)
+        root_id = self.trace.trace_id
+        before = await self.runner.resource_budget.get_usage(root_id)
+        original = self.runner.candidate_service.update_validation_checkpoint
+        failed_once = False
+
+        async def fail_material_checkpoint_once(*args, **kwargs):
+            nonlocal failed_once
+            if kwargs.get("material_usage_recorded") and not failed_once:
+                failed_once = True
+                raise OSError("material checkpoint publish failed")
+            return await original(*args, **kwargs)
+
+        with patch.object(
+            self.runner.candidate_service,
+            "update_validation_checkpoint",
+            side_effect=fail_material_checkpoint_once,
+        ):
+            with self.assertRaisesRegex(OSError, "material checkpoint"):
+                await self.runner.validate_recursive_trace(
+                    self.trace.trace_id,
+                    candidate_ref=candidate,
+                )
+            after_failure = await self.runner.resource_budget.get_usage(root_id)
+            recovered = await self.runner.validate_recursive_trace(
+                self.trace.trace_id,
+                candidate_ref=candidate,
+            )
+        after_retry = await self.runner.resource_budget.get_usage(root_id)
+
+        self.assertEqual("passed", recovered.result.outcome)
+        self.assertGreater(
+            after_failure.validation_material_chars,
+            before.validation_material_chars,
+        )
+        self.assertEqual(
+            after_failure.validation_material_chars,
+            after_retry.validation_material_chars,
+        )
+        self.assertEqual(1, self.quality.calls)
+        self.assertEqual(1, self.llm_calls)

tests/test_candidate_service.py

@@ -478,6 +478,47 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
                 effective_at_sequence=10,
             )
 
+    async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
+        await self._create_child("child-a")
+        await self.store.create_trace(Trace(
+            trace_id="grandchild-a",
+            mode="agent",
+            agent_type="writer",
+            uid="user-1",
+            parent_trace_id="child-a",
+            context=self._context(
+                "root-a",
+                task_brief=TaskBrief(
+                    objective="write descendant candidate",
+                    reason="the child needs an option",
+                    completion_criteria=["candidate is complete"],
+                    expected_outputs=["one candidate"],
+                ),
+            ),
+        ))
+        candidate = await self.service.manage(
+            "grandchild-a",
+            operation="create",
+            content={"text": "nested candidate"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("grandchild-a", candidate)
+        await self.service.apply_review_actions(
+            "child-a",
+            "grandchild-a",
+            report_refs=[candidate],
+            actions=[CandidateReviewAction(
+                action="adopt",
+                candidate_ref=candidate,
+                reason="commit descendant output",
+            )],
+            effective_at_sequence=9,
+        )
+
+        with self.assertRaisesRegex(CandidateStateError, "committed"):
+            await self.service.assert_rewind_allowed("root-a", 100)
+
     async def test_failed_candidate_can_only_be_revised_or_discarded(self):
         await self._create_child("child-a")
         candidate = await self.service.manage(