feat(validation): 接入候选独立验收和父级幂等决策

将应用固化的确定性质量规则编译进现有 ValidationPlan，并由框架严格校验 Provider 返回的 rule、版本和完整结果集合。候选验收按精确 CandidateRef 与 plan_hash 独立缓存，确定性失败只阻断对应 output scope，不污染同一报告中的其他候选。

扩展 TaskReport 与 TaskReview 的候选协议：报告只能引用根账本中的本 Task 候选，父级可对已验收 revision 执行 adopt、discard 或 retry_from=output 的 revise。采用操作先持久化 intent，再以稳定 operation ID 调用业务 Repository 并固化 receipt；并发终态互斥，已提交采用形成 rewind barrier。

新增 Provider 漏项、额外结果、异常、A/B 隔离缓存、跨 Task 引用、幂等采用和采用后回溯拒绝测试。

cyber_agent/application/__init__.py

@@ -7,6 +7,7 @@ from cyber_agent.application.models import (
     AgentRole,
     ApplicationRef,
     ProviderRef,
+    QualityRuleSpec,
     RoleRunLimits,
     RunLimits,
     SkillSpec,
@@ -24,10 +25,16 @@ _LAZY_EXPORTS = {
     "CandidatePointer": ("cyber_agent.application.candidate", "CandidatePointer"),
     "CandidateRef": ("cyber_agent.application.candidate", "CandidateRef"),
     "CandidateRepository": ("cyber_agent.application.candidate", "CandidateRepository"),
+    "CandidateReviewAction": ("cyber_agent.application.candidate", "CandidateReviewAction"),
     "CandidateVersionRequest": ("cyber_agent.application.candidate", "CandidateVersionRequest"),
     "ContextMaterial": ("cyber_agent.application.ports", "ContextMaterial"),
     "ContextRequest": ("cyber_agent.application.ports", "ContextRequest"),
     "ContextResourceDescriptor": ("cyber_agent.application.ports", "ContextResourceDescriptor"),
+    "CandidateValidationRecord": ("cyber_agent.application.quality", "CandidateValidationRecord"),
+    "QualityCheckInput": ("cyber_agent.application.quality", "QualityCheckInput"),
+    "QualityCheckOutcome": ("cyber_agent.application.quality", "QualityCheckOutcome"),
+    "QualityCheckProvider": ("cyber_agent.application.quality", "QualityCheckProvider"),
+    "ValidationSubject": ("cyber_agent.application.quality", "ValidationSubject"),
     "TaskContextProvider": ("cyber_agent.application.ports", "TaskContextProvider"),
 }
 
@@ -56,15 +63,22 @@ __all__ = [
     "CandidatePointer",
     "CandidateRef",
     "CandidateRepository",
+    "CandidateReviewAction",
     "CandidateVersionRequest",
     "ContextMaterial",
     "ContextRequest",
     "ContextResourceDescriptor",
     "ProviderRef",
+    "QualityRuleSpec",
+    "QualityCheckInput",
+    "QualityCheckOutcome",
+    "QualityCheckProvider",
     "RoleRunLimits",
     "RunLimits",
     "SkillSpec",
     "TaskContextProvider",
     "ToolSet",
     "ToolSpec",
+    "ValidationSubject",
+    "CandidateValidationRecord",
 ]

cyber_agent/application/candidate.py

@@ -72,6 +72,12 @@ class CandidateAdoptionReceipt(ApplicationModel):
     committed: bool = True
 
 
+class CandidateReviewAction(ApplicationModel):
+    action: Literal["adopt", "discard", "revise"]
+    candidate_ref: CandidateRef
+    reason: str = Field(min_length=1, max_length=2_000)
+
+
 @runtime_checkable
 class CandidateRepository(Protocol):
     async def put_version(
@@ -103,13 +109,16 @@ class CandidateLifecycleRecord(ApplicationModel):
         "adoption_failed",
     ]
     effective_at_sequence: int = Field(ge=0)
+    source_trace_id: str | None = None
     reason: str | None = Field(default=None, max_length=2_000)
     receipt: CandidateAdoptionReceipt | None = None
 
 
 class CandidateOperationRecord(ApplicationModel):
     operation_id: str = Field(min_length=1, max_length=300)
-    operation: Literal["create", "fork", "merge", "adopt", "discard"]
+    operation: Literal[
+        "create", "fork", "merge", "adopt", "discard", "revise"
+    ]
     input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
     status: Literal["pending", "completed", "failed"] = "pending"
     candidate: CandidatePointer | None = None

cyber_agent/application/candidate_service.py

@@ -8,15 +8,19 @@ from typing import Any
 from weakref import WeakValueDictionary
 
 from cyber_agent.application.candidate import (
+    CandidateAdoptionReceipt,
+    CandidateAdoptionRequest,
     CandidateLedger,
     CandidateLifecycleRecord,
     CandidateOperationRecord,
     CandidatePointer,
     CandidateRef,
     CandidateRepository,
+    CandidateReviewAction,
     CandidateVersionRequest,
 )
 from cyber_agent.application.models import canonical_json
+from cyber_agent.application.quality import 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
@@ -213,7 +217,375 @@ class CandidateService:
                         "TaskReport may only reference proposed candidates"
                     )
 
+    async def resolve_for_validation(
+        self,
+        trace_id: str,
+        candidate_ref: CandidateRef,
+    ):
+        """Resolve one exact report-owned candidate through the trusted resolver."""
+        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(
+                    "CandidateRef does not match the root ledger"
+                )
+            self._validate_candidate_owner(persisted, trace, root)
+        materials, issues = await resolve_artifact_refs(
+            [candidate_ref.artifact_ref],
+            resolver=self.artifact_resolver,
+            root_trace_id=root.trace_id,
+            uid=trace.uid,
+        )
+        if issues or len(materials) != 1:
+            reason = issues[0].reason if issues else "Artifact was not resolved"
+            raise CandidateStateError(
+                f"Candidate ArtifactRef cannot be validated: {reason}"
+            )
+        return materials[0]
+
+    async def cached_validation(
+        self,
+        trace_id: str,
+        candidate_ref: CandidateRef,
+        plan_hash: str,
+    ) -> CandidateValidationRecord | None:
+        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)
+            self._validate_candidate_owner(persisted, trace, root)
+            for raw in ledger.validations:
+                record = CandidateValidationRecord.model_validate(raw)
+                if (
+                    record.candidate_ref == candidate_ref
+                    and record.plan_hash == plan_hash
+                ):
+                    return record
+        return None
+
+    async def record_validation(
+        self,
+        trace_id: str,
+        record: CandidateValidationRecord,
+    ) -> None:
+        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(record.candidate_ref)
+            if persisted != record.candidate_ref:
+                raise CandidateStateError(
+                    "Candidate validation references an altered CandidateRef"
+                )
+            self._validate_candidate_owner(persisted, trace, root)
+            retained = tuple(
+                raw for raw in ledger.validations
+                if not (
+                    CandidateValidationRecord.model_validate(raw).candidate_ref
+                    == record.candidate_ref
+                    and CandidateValidationRecord.model_validate(raw).plan_hash
+                    == record.plan_hash
+                )
+            )
+            updated = ledger.model_copy(update={
+                "validations": (
+                    *retained,
+                    record.model_dump(mode="json"),
+                ),
+            })
+            await self.store.replace_candidate_ledger(
+                root.trace_id,
+                updated.model_dump(mode="json"),
+            )
+
+    async def apply_review_actions(
+        self,
+        parent_trace_id: str,
+        child_trace_id: str,
+        *,
+        report_refs: list[CandidateRef],
+        actions: list[CandidateReviewAction],
+        effective_at_sequence: int,
+    ) -> list[CandidateLifecycleRecord]:
+        """Apply parent-owned candidate decisions behind one root lock."""
+        if not actions:
+            return []
+        parent = await self.store.get_trace(parent_trace_id)
+        child = await self.store.get_trace(child_trace_id)
+        if parent is None or child is None:
+            raise CandidateStateError("Parent or child Trace not found")
+        if child.parent_trace_id != parent_trace_id or child.uid != parent.uid:
+            raise CandidateStateError("Candidate review requires a direct child")
+        _parent, root = await self._require_bound_trace(parent_trace_id)
+        child_root = child.context.get("root_trace_id") or child.trace_id
+        if (
+            child_root != root.trace_id
+            or child.context.get("application_ref")
+            != self.binding.application_ref.model_dump(mode="json")
+        ):
+            raise CandidateStateError("Candidate review application/root mismatch")
+        report_by_key = {
+            (item.candidate_id, item.revision): item for item in report_refs
+        }
+        action_keys = [
+            (item.candidate_ref.candidate_id, item.candidate_ref.revision)
+            for item in actions
+        ]
+        if len(action_keys) != len(set(action_keys)):
+            raise CandidateStateError("Candidate review actions must be unique")
+        async with self._lock_for(root.trace_id):
+            ledger = await self._load_ledger(root.trace_id)
+            validation_by_key: dict[tuple[str, int], CandidateValidationRecord] = {}
+            for raw in ledger.validations:
+                record = CandidateValidationRecord.model_validate(raw)
+                key = (
+                    record.candidate_ref.candidate_id,
+                    record.candidate_ref.revision,
+                )
+                validation_by_key[key] = record
+            for action, key in zip(actions, action_keys):
+                if key not in report_by_key or report_by_key[key] != action.candidate_ref:
+                    raise CandidateStateError(
+                        "Candidate action must reference the exact child TaskReport"
+                    )
+                persisted = ledger.candidate(action.candidate_ref)
+                if persisted != action.candidate_ref or persisted.owner_trace_id != child_trace_id:
+                    raise CandidateStateError(
+                        "Candidate action references another Task or altered revision"
+                    )
+                record = validation_by_key.get(key)
+                if record is None:
+                    raise CandidateStateError(
+                        "Candidate action requires framework validation"
+                    )
+                from cyber_agent.core.validation import ValidationResult
+
+                outcome = ValidationResult.model_validate(
+                    record.validation_result
+                ).outcome
+                if action.action == "adopt" and outcome != "passed":
+                    raise CandidateStateError(
+                        "Only a passed candidate revision can be adopted"
+                    )
+                if action.action == "revise" and outcome == "passed":
+                    raise CandidateStateError(
+                        "Candidate revise requires a non-passed validation"
+                    )
+                state = ledger.current_state(action.candidate_ref)
+                expected_operation_id = (
+                    f"candidate-review:{parent_trace_id}:{effective_at_sequence}:"
+                    f"{key[0]}:{key[1]}:{action.action}"
+                )
+                same_completed_operation = any(
+                    item.operation_id == expected_operation_id
+                    and item.status == "completed"
+                    for item in ledger.operations
+                )
+                if (
+                    state not in {"proposed", "adoption_pending"}
+                    and not same_completed_operation
+                ):
+                    raise CandidateStateError(
+                        f"Candidate already has terminal state: {state}"
+                    )
+
+            applied: list[CandidateLifecycleRecord] = []
+            for action in actions:
+                ledger, record = await self._apply_review_action(
+                    ledger,
+                    action,
+                    parent_trace_id=parent_trace_id,
+                    root_trace_id=root.trace_id,
+                    effective_at_sequence=effective_at_sequence,
+                )
+                applied.append(record)
+            return applied
+
+    async def _apply_review_action(
+        self,
+        ledger: CandidateLedger,
+        action: CandidateReviewAction,
+        *,
+        parent_trace_id: str,
+        root_trace_id: str,
+        effective_at_sequence: int,
+    ) -> tuple[CandidateLedger, CandidateLifecycleRecord]:
+        pointer = CandidatePointer(
+            candidate_id=action.candidate_ref.candidate_id,
+            revision=action.candidate_ref.revision,
+        )
+        operation_id = (
+            f"candidate-review:{parent_trace_id}:{effective_at_sequence}:"
+            f"{pointer.candidate_id}:{pointer.revision}:{action.action}"
+        )
+        input_hash = sha256(
+            canonical_json(action).encode("utf-8")
+        ).hexdigest()
+        existing = next(
+            (item for item in ledger.operations if item.operation_id == operation_id),
+            None,
+        )
+        if existing is not None:
+            if existing.input_hash != input_hash:
+                raise CandidateStateError("Candidate review payload changed during recovery")
+            if existing.status == "completed":
+                record = next(
+                    item for item in reversed(ledger.lifecycle)
+                    if item.operation_id == operation_id
+                )
+                return ledger, record
+            if existing.status == "failed":
+                raise CandidateStateError(existing.error or "Candidate review failed")
+        else:
+            operation = CandidateOperationRecord(
+                operation_id=operation_id,
+                operation=action.action,
+                input_hash=input_hash,
+                candidate=pointer,
+            )
+            ledger = ledger.model_copy(update={
+                "operations": (*ledger.operations, operation),
+            })
+        if action.action != "adopt":
+            record = CandidateLifecycleRecord(
+                operation_id=operation_id,
+                candidate=pointer,
+                state="discarded",
+                effective_at_sequence=effective_at_sequence,
+                source_trace_id=parent_trace_id,
+                reason=(
+                    f"retry_from=output: {action.reason}"
+                    if action.action == "revise"
+                    else action.reason
+                ),
+            )
+            ledger = self._finish_review_operation(ledger, operation_id, record)
+            await self.store.replace_candidate_ledger(
+                root_trace_id,
+                ledger.model_dump(mode="json"),
+            )
+            return ledger, record
+
+        pending = CandidateLifecycleRecord(
+            operation_id=operation_id,
+            candidate=pointer,
+            state="adoption_pending",
+            effective_at_sequence=effective_at_sequence,
+            source_trace_id=parent_trace_id,
+            reason=action.reason,
+        )
+        if ledger.current_state(pointer) != "adoption_pending":
+            ledger = ledger.model_copy(update={
+                "lifecycle": (*ledger.lifecycle, pending),
+            })
+            await self.store.replace_candidate_ledger(
+                root_trace_id,
+                ledger.model_dump(mode="json"),
+            )
+        try:
+            receipt = CandidateAdoptionReceipt.model_validate(
+                await self.repository.commit_adoption(CandidateAdoptionRequest(
+                    operation_id=operation_id,
+                    candidate_ref=action.candidate_ref,
+                ))
+            )
+            if receipt.operation_id != operation_id or not receipt.committed:
+                raise CandidateStateError("Candidate adoption receipt is invalid")
+        except Exception as exc:
+            failed = CandidateLifecycleRecord(
+                operation_id=operation_id,
+                candidate=pointer,
+                state="adoption_failed",
+                effective_at_sequence=effective_at_sequence,
+                source_trace_id=parent_trace_id,
+                reason=str(exc)[:2_000],
+            )
+            ledger = self._finish_review_operation(
+                ledger,
+                operation_id,
+                failed,
+                error=str(exc),
+            )
+            await self.store.replace_candidate_ledger(
+                root_trace_id,
+                ledger.model_dump(mode="json"),
+            )
+            raise
+        adopted = CandidateLifecycleRecord(
+            operation_id=operation_id,
+            candidate=pointer,
+            state="adopted",
+            effective_at_sequence=effective_at_sequence,
+            source_trace_id=parent_trace_id,
+            reason=action.reason,
+            receipt=receipt,
+        )
+        ledger = self._finish_review_operation(ledger, operation_id, adopted)
+        # A failure here intentionally leaves the durable pending intent. The
+        # Repository sees the same operation_id when the review is replayed.
+        await self.store.replace_candidate_ledger(
+            root_trace_id,
+            ledger.model_dump(mode="json"),
+        )
+        return ledger, adopted
+
+    @staticmethod
+    def _finish_review_operation(
+        ledger: CandidateLedger,
+        operation_id: str,
+        lifecycle: CandidateLifecycleRecord,
+        *,
+        error: str | None = None,
+    ) -> CandidateLedger:
+        operations = tuple(
+            item.model_copy(update={
+                "status": "failed" if error else "completed",
+                "error": error[:2_000] if error else None,
+            })
+            if item.operation_id == operation_id else item
+            for item in ledger.operations
+        )
+        return ledger.model_copy(update={
+            "operations": operations,
+            "lifecycle": (*ledger.lifecycle, lifecycle),
+        })
+
+    async def assert_rewind_allowed(
+        self,
+        trace_id: str,
+        after_sequence: int,
+    ) -> None:
+        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)
+            for lifecycle in ledger.lifecycle:
+                if lifecycle.state != "adopted":
+                    continue
+                candidate = ledger.candidate(lifecycle.candidate)
+                crosses_review = (
+                    lifecycle.source_trace_id == trace_id
+                    and lifecycle.effective_at_sequence > after_sequence
+                )
+                crosses_version = (
+                    candidate.owner_trace_id == trace_id
+                    and candidate.created_at_sequence > after_sequence
+                )
+                if crosses_review or crosses_version:
+                    raise CandidateStateError(
+                        "Cannot rewind across a committed candidate adoption"
+                    )
+
     async def _require_owner_trace(self, trace_id: str):
+        trace, root = await self._require_bound_trace(trace_id)
+        state = ensure_task_protocol(trace.context)
+        if state.get("task_report") is not None:
+            raise CandidateStateError("Candidates cannot change after TaskReport submission")
+        if state.get("pending_reviews") or state.get("next_actions"):
+            raise CandidateStateError("Resolve protocol lifecycle work before candidates")
+        return trace, root
+
+    async def _require_bound_trace(self, trace_id: str):
         trace = await self.store.get_trace(trace_id)
         if trace is None:
             raise CandidateStateError(f"Trace not found: {trace_id}")
@@ -233,11 +605,6 @@ class CandidateService:
             raise CandidateStateError("Candidate root or owner mismatch")
         if root.context.get("application_ref") != application_ref:
             raise CandidateStateError("Candidate root application mismatch")
-        state = ensure_task_protocol(trace.context)
-        if state.get("task_report") is not None:
-            raise CandidateStateError("Candidates cannot change after TaskReport submission")
-        if state.get("pending_reviews") or state.get("next_actions"):
-            raise CandidateStateError("Resolve protocol lifecycle work before candidates")
         return trace, root
 
     async def _load_ledger(self, root_trace_id: str) -> CandidateLedger:

cyber_agent/application/models.py

@@ -91,6 +91,22 @@ class ToolSet(ApplicationModel):
         return self
 
 
+class QualityRuleSpec(ApplicationModel):
+    """A deterministic rule frozen into an application version."""
+
+    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"
+    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)
+
+    @model_validator(mode="after")
+    def validate_config(self) -> "QualityRuleSpec":
+        canonical_json(self.config)
+        return self
+
+
 class RoleRunLimits(ApplicationModel):
     max_iterations: int | None = Field(default=None, gt=0)
     max_depth: int | None = Field(default=None, ge=0)
@@ -159,6 +175,7 @@ class AgentApplication(ApplicationModel):
     artifact_resolver_ref: ProviderRef | None = None
     candidate_repository_ref: ProviderRef | None = None
     quality_provider_ref: ProviderRef | None = None
+    quality_rules: tuple[QualityRuleSpec, ...] = ()
     event_projector_ref: ProviderRef | None = None
     validation_policy: dict[str, Any] = Field(default_factory=dict)
     run_limits: RunLimits = Field(default_factory=RunLimits)
@@ -197,6 +214,15 @@ class AgentApplication(ApplicationModel):
             raise ValueError(
                 "candidate_repository_ref requires artifact_resolver_ref"
             )
+        rule_ids = [item.rule_id for item in self.quality_rules]
+        if len(rule_ids) != len(set(rule_ids)):
+            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

@@ -0,0 +1,84 @@
+"""Deterministic candidate-quality contracts and strict provider boundary."""
+
+from __future__ import annotations
+
+from typing import Any, Literal, Protocol, Sequence
+
+from pydantic import Field, model_validator
+
+from cyber_agent.application.candidate import CandidateRef
+from cyber_agent.application.models import ApplicationModel, QualityRuleSpec
+from cyber_agent.core.artifacts import ValidationMaterial
+
+
+class ValidationSubject(ApplicationModel):
+    subject_type: Literal["trace", "candidate"]
+    trace_id: str = Field(min_length=1)
+    candidate_ref: CandidateRef | None = None
+
+    @model_validator(mode="after")
+    def validate_reference(self) -> "ValidationSubject":
+        if (self.subject_type == "candidate") != (self.candidate_ref is not None):
+            raise ValueError("candidate subjects require exactly one CandidateRef")
+        return self
+
+
+class QualityCheckInput(ApplicationModel):
+    subject: ValidationSubject
+    rules: tuple[QualityRuleSpec, ...] = Field(min_length=1)
+    material: ValidationMaterial
+
+    @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")
+        return self
+
+
+class QualityCheckOutcome(ApplicationModel):
+    rule_id: str = Field(min_length=1, max_length=100)
+    rule_version: str = Field(min_length=1, max_length=100)
+    status: Literal["passed", "failed", "unknown", "error"]
+    evidence_refs: tuple[str, ...] = ()
+    issue: str | None = Field(default=None, max_length=2_000)
+
+    @model_validator(mode="after")
+    def validate_status(self) -> "QualityCheckOutcome":
+        if self.status == "passed" and self.issue is not None:
+            raise ValueError("passed quality outcome requires issue=null")
+        if self.status != "passed" and not self.issue:
+            raise ValueError(f"{self.status} quality outcome requires issue")
+        return self
+
+
+class QualityCheckProvider(Protocol):
+    async def check(
+        self,
+        request: QualityCheckInput,
+    ) -> Sequence[QualityCheckOutcome]: ...
+
+
+class CandidateValidationRecord(ApplicationModel):
+    schema_version: Literal[1] = 1
+    candidate_ref: CandidateRef
+    plan_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    validation_result: dict[str, Any]
+    validated_at_sequence: int = Field(ge=0)
+
+    @model_validator(mode="after")
+    def validate_result(self) -> "CandidateValidationRecord":
+        # Lazy import keeps application declarations independent from the
+        # Validator implementation while still rejecting malformed ledgers.
+        from cyber_agent.core.validation import ValidationResult
+
+        result = ValidationResult.model_validate(self.validation_result)
+        if result.plan_hash != self.plan_hash:
+            raise ValueError("candidate validation result plan hash mismatch")
+        return self

cyber_agent/core/runner.py

@@ -115,6 +115,8 @@ from cyber_agent.core.artifacts import (
     resolve_artifact_refs,
 )
 from cyber_agent.core.validation import (
+    ValidationCheck,
+    ValidationCheckSpec,
     LLMValidator,
     ScopeValidationResult,
     ValidationPolicy,
@@ -528,6 +530,7 @@ class AgentRunner:
         candidate_output: Optional[str] = None,
         deterministic_failure: Optional[Dict[str, Any]] = None,
         root_validator: bool = False,
+        candidate_ref: Any = None,
     ) -> ValidationRun:
         """编译Plan、解析材料、顺序运行Scope并持久化聚合缓存。"""
         if not self.trace_store or not self.llm_call:
@@ -549,6 +552,14 @@ class AgentRunner:
         )
         policy, settings = require_validation_policy(root.context)
         state = ensure_task_protocol(evaluated.context)
+        if candidate_ref is not None:
+            from cyber_agent.application.candidate import CandidateRef
+
+            candidate_ref = CandidateRef.model_validate(candidate_ref)
+            if self.candidate_service is None:
+                raise ValueError("Candidate validation requires CandidateService")
+            if root_validator:
+                raise ValueError("Candidate validation cannot be root validation")
         runtime_policy = policy_from_context(evaluated.context)
         authoritative_brief = state.get("task_brief")
         if authoritative_brief is not None:
@@ -623,6 +634,34 @@ class AgentRunner:
         )
         materials.extend(resolved_materials)
         material_issues.extend(resolved_issues)
+        candidate_material = None
+        if candidate_ref is not None:
+            try:
+                candidate_material = await self.candidate_service.resolve_for_validation(
+                    evaluated_trace_id,
+                    candidate_ref,
+                )
+                materials = [
+                    item for item in materials
+                    if not (
+                        item.artifact_id == candidate_material.artifact_id
+                        and item.version == candidate_material.version
+                    )
+                ]
+                materials.append(candidate_material)
+                candidate_output = json.dumps(
+                    candidate_material.content,
+                    ensure_ascii=False,
+                    sort_keys=True,
+                    separators=(",", ":"),
+                )
+            except Exception as exc:
+                material_issues.append(MaterialIssue(
+                    artifact_id=candidate_ref.artifact_ref.artifact_id,
+                    outcome="error",
+                    reason=f"Candidate material cannot be resolved: {exc}",
+                    scopes=["output"],
+                ))
         total_material_chars = sum(material_chars(item) for item in materials)
         if deterministic_failure:
             material_issues.append(MaterialIssue(
@@ -645,6 +684,35 @@ class AgentRunner:
                 "expected_outputs": expected_outputs or [],
                 "validation_scopes": [] if scope == "task" else [scope],
             }
+        validation_subject = None
+        quality_specs: list[ValidationCheckSpec] = []
+        quality_manifest: list[dict[str, Any]] = []
+        fixed_checks: dict[str, ValidationCheck] = {}
+        fixed_scope_errors: dict[ValidationScope, str] = {}
+        quality_rules: tuple[Any, ...] = ()
+        if candidate_ref is not None:
+            from cyber_agent.application.quality import ValidationSubject
+
+            validation_subject = ValidationSubject(
+                subject_type="candidate",
+                trace_id=evaluated_trace_id,
+                candidate_ref=candidate_ref,
+            )
+            quality_rules = tuple(
+                self.application_binding.application.quality_rules
+            )
+            quality_specs = [
+                ValidationCheckSpec(
+                    check_id=f"quality.{rule.rule_id}",
+                    scope=rule.scope,
+                    criterion=rule.criterion,
+                    method="deterministic",
+                )
+                for rule in quality_rules
+            ]
+            quality_manifest = [
+                item.model_dump(mode="json") for item in quality_rules
+            ]
         plan = policy.compile_plan(
             task_brief=task_brief,
             task_brief_version=task_brief_version,
@@ -657,9 +725,37 @@ class AgentRunner:
             model_by_scope=model_by_scope,
             root=root_validator,
             task_progress=task_progress,
+            subject=validation_subject,
+            quality_checks=quality_specs,
+            quality_manifest=quality_manifest,
         )
 
         cached = state.get("task_report_validation")
+        if candidate_ref is not None:
+            candidate_cached = await self.candidate_service.cached_validation(
+                evaluated_trace_id,
+                candidate_ref,
+                plan.plan_hash,
+            )
+            if candidate_cached is not None:
+                aggregate = ValidationResult.model_validate(
+                    candidate_cached.validation_result
+                )
+                return ValidationRun(
+                    result=aggregate,
+                    trace_ids=[
+                        item.validator_trace_id
+                        for item in aggregate.scope_results
+                    ],
+                    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:
@@ -695,11 +791,12 @@ class AgentRunner:
                 "validated_at_sequence": plan.evaluated_head_sequence,
                 "material_usage_recorded": total_material_chars == 0,
             }
-            state["task_report_validation"] = validation_cache
-            await self.trace_store.update_trace(
-                evaluated_trace_id,
-                context=evaluated.context,
-            )
+            if candidate_ref is None:
+                state["task_report_validation"] = validation_cache
+                await self.trace_store.update_trace(
+                    evaluated_trace_id,
+                    context=evaluated.context,
+                )
 
         if (
             total_material_chars
@@ -709,21 +806,24 @@ class AgentRunner:
                 evaluated_trace_id,
                 material_chars_count=total_material_chars,
             )
-            fresh = await self.trace_store.get_trace(evaluated_trace_id)
-            if not fresh:
-                raise ValueError(f"Trace not found: {evaluated_trace_id}")
-            fresh_state = ensure_task_protocol(fresh.context)
-            fresh_cache = fresh_state.get("task_report_validation")
-            if (
-                not isinstance(fresh_cache, dict)
-                or fresh_cache.get("plan_hash") != plan.plan_hash
-            ):
-                raise ValueError("Validation cache changed while recording materials")
-            fresh_cache["material_usage_recorded"] = True
-            await self.trace_store.update_trace(
-                evaluated_trace_id,
-                context=fresh.context,
-            )
+            if candidate_ref is not None:
+                validation_cache["material_usage_recorded"] = True
+            else:
+                fresh = await self.trace_store.get_trace(evaluated_trace_id)
+                if not fresh:
+                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
+                fresh_state = ensure_task_protocol(fresh.context)
+                fresh_cache = fresh_state.get("task_report_validation")
+                if (
+                    not isinstance(fresh_cache, dict)
+                    or fresh_cache.get("plan_hash") != plan.plan_hash
+                ):
+                    raise ValueError("Validation cache changed while recording materials")
+                fresh_cache["material_usage_recorded"] = True
+                await self.trace_store.update_trace(
+                    evaluated_trace_id,
+                    context=fresh.context,
+                )
 
         lineage_event = None
         if (
@@ -802,6 +902,19 @@ class AgentRunner:
         )
         try:
             async def persist_scope(result: ScopeValidationResult) -> None:
+                if candidate_ref is not None:
+                    by_scope = {
+                        item.get("scope"): item
+                        for item in validation_cache.get("scope_results", [])
+                        if isinstance(item, dict)
+                    }
+                    by_scope[result.scope] = result.model_dump(mode="json")
+                    validation_cache["scope_results"] = [
+                        by_scope[item]
+                        for item in plan.effective_scopes
+                        if item in by_scope
+                    ]
+                    return
                 fresh = await self.trace_store.get_trace(evaluated_trace_id)
                 if not fresh:
                     raise ValueError(f"Trace not found: {evaluated_trace_id}")
@@ -840,7 +953,22 @@ class AgentRunner:
                 source_urls=source_urls,
                 resume_scope_results=resume_scope_results,
                 on_scope_result=persist_scope,
+                fixed_checks=fixed_checks,
+                fixed_scope_errors=fixed_scope_errors,
             )
+            if candidate_ref is not None:
+                from cyber_agent.application.quality import CandidateValidationRecord
+
+                await self.candidate_service.record_validation(
+                    evaluated_trace_id,
+                    CandidateValidationRecord(
+                        candidate_ref=candidate_ref,
+                        plan_hash=plan.plan_hash,
+                        validation_result=run.result.model_dump(mode="json"),
+                        validated_at_sequence=plan.evaluated_head_sequence,
+                    ),
+                )
+                return run
             fresh = await self.trace_store.get_trace(evaluated_trace_id)
             if not fresh:
                 raise ValueError(f"Trace not found: {evaluated_trace_id}")
@@ -861,6 +989,78 @@ class AgentRunner:
             if lineage_event is not None:
                 self.release_recursive_trace(evaluated_trace_id, lineage_event)
 
+    async def _run_quality_checks(
+        self,
+        subject: Any,
+        material: ValidationMaterial,
+        rules: tuple[Any, ...],
+    ) -> tuple[dict[str, ValidationCheck], dict[ValidationScope, str]]:
+        """Execute one frozen quality batch and reject provider-shaped plans."""
+        from cyber_agent.application.quality import (
+            QualityCheckInput,
+            QualityCheckOutcome,
+        )
+
+        provider = self.application_binding.services.quality_provider
+        check_ids = {rule.rule_id: f"quality.{rule.rule_id}" for rule in rules}
+
+        def provider_error(reason: str):
+            return (
+                {
+                    check_id: ValidationCheck(
+                        check_id=check_id,
+                        status="unknown",
+                        issue=reason,
+                    )
+                    for check_id in check_ids.values()
+                },
+                {"output": reason},
+            )
+
+        if provider is None:
+            return provider_error("QualityCheckProvider is unavailable")
+        request = QualityCheckInput(
+            subject=subject,
+            rules=rules,
+            material=material,
+        )
+        try:
+            raw = await asyncio.wait_for(
+                provider.check(request),
+                timeout=max(rule.timeout_seconds for rule in rules),
+            )
+            outcomes = [QualityCheckOutcome.model_validate(item) for item in raw]
+        except Exception as exc:
+            return provider_error(f"QualityCheckProvider failed: {exc}")
+        returned = [(item.rule_id, item.rule_version) for item in outcomes]
+        expected = [(item.rule_id, item.version) for item in rules]
+        if len(returned) != len(set(returned)) or set(returned) != set(expected):
+            extras = sorted(set(returned) - set(expected))
+            missing = sorted(set(expected) - set(returned))
+            return provider_error(
+                "QualityCheckProvider results do not match the frozen rules: "
+                f"extras={extras}, missing={missing}"
+            )
+        by_rule = {item.rule_id: item for item in outcomes}
+        checks: dict[str, ValidationCheck] = {}
+        errors: dict[ValidationScope, str] = {}
+        for rule in rules:
+            outcome = by_rule[rule.rule_id]
+            check_id = check_ids[rule.rule_id]
+            checks[check_id] = ValidationCheck(
+                check_id=check_id,
+                status=("unknown" if outcome.status == "error" else outcome.status),
+                evidence_refs=list(outcome.evidence_refs),
+                issue=outcome.issue,
+            )
+            if outcome.status == "error":
+                errors[rule.scope] = outcome.issue or "Quality check failed"
+        await self.record_recursive_validation_usage(
+            subject.trace_id,
+            tool_calls=len(rules),
+        )
+        return checks, errors
+
     async def dream(
         self,
         memory_config: MemoryConfig,
@@ -4025,6 +4225,8 @@ class AgentRunner:
 
         # 2. 找到安全截断点（确保不截断在 tool_call 和 tool response 之间）
         cutoff = self._find_safe_cutoff(all_messages, after_sequence)
+        if self.candidate_service is not None:
+            await self.candidate_service.assert_rewind_allowed(trace_id, cutoff)
 
         # 3. 快照并重建 GoalTree
         if goal_tree:

cyber_agent/core/task_protocol.py

@@ -17,7 +17,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
 from cyber_agent.core.artifacts import ArtifactRef
 from cyber_agent.core.task_contract import TaskContractRef
 from cyber_agent.core.validation import RetryFrom
-from cyber_agent.application.candidate import CandidateRef
+from cyber_agent.application.candidate import CandidateRef, CandidateReviewAction
 
 
 TaskOutcome = Literal["satisfied", "partial", "failed", "protocol_error"]
@@ -236,9 +236,24 @@ class TaskReview(StrictProtocolModel):
     reason: str = Field(min_length=1)
     approved_next_task: TaskBrief | None = None
     retry_from: RetryFrom | None = None
+    candidate_actions: list[CandidateReviewAction] = Field(
+        default_factory=list,
+        max_length=20,
+    )
 
     @model_validator(mode="after")
     def validate_approved_next_task(self) -> "TaskReview":
+        action_types = {item.action for item in self.candidate_actions}
+        if "revise" in action_types and self.decision != "REVISE_CHILD":
+            raise ValueError("candidate revise requires REVISE_CHILD")
+        if self.decision == "REVISE_CHILD" and action_types - {"revise", "discard"}:
+            raise ValueError("REVISE_CHILD cannot adopt a candidate")
+        identities = [
+            (item.candidate_ref.candidate_id, item.candidate_ref.revision)
+            for item in self.candidate_actions
+        ]
+        if len(identities) != len(set(identities)):
+            raise ValueError("candidate_actions must reference unique candidates")
         if self.decision in {"ACCEPT_REFINE", "DESCEND_AGAIN"}:
             if not self.approved_next_task:
                 raise ValueError(f"{self.decision} requires approved_next_task")

cyber_agent/core/validation.py

@@ -17,6 +17,7 @@ from typing import Any, Literal, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
+from cyber_agent.application.quality import ValidationSubject
 from cyber_agent.core.artifacts import MaterialIssue, ValidationMaterial
 from cyber_agent.core.validator_web import ValidatorToolSession
 from cyber_agent.trace.models import Message, Trace
@@ -162,6 +163,8 @@ class ValidationPlan(_StrictModel):
     effective_scopes: list[ValidationScope] = Field(min_length=1)
     checks: list[ValidationCheckSpec] = Field(min_length=1)
     material_manifest: list[dict[str, Any]] = Field(default_factory=list)
+    subject: ValidationSubject | None = None
+    quality_manifest: list[dict[str, Any]] = Field(default_factory=list)
     plan_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
 
     @model_validator(mode="after")
@@ -369,12 +372,19 @@ class ValidationPolicy(_StrictModel):
         model_by_scope: Mapping[str, str],
         root: bool,
         task_progress: Mapping[str, Any] | BaseModel | None = None,
+        subject: ValidationSubject | None = None,
+        quality_checks: Sequence[ValidationCheckSpec] = (),
+        quality_manifest: Sequence[Mapping[str, Any]] = (),
     ) -> ValidationPlan:
         """从任务合同和权威材料确定性编译完整检查计划。"""
         brief = _jsonable(task_brief) or {}
         anchor = _jsonable(root_task_anchor) or {}
         requested = brief.get("validation_scopes", []) if isinstance(brief, dict) else []
-        scopes = self.effective_scopes(requested, root=root)
+        scopes = (
+            ["output"]
+            if subject is not None and subject.subject_type == "candidate"
+            else self.effective_scopes(requested, root=root)
+        )
         if isinstance(brief, dict):
             brief = dict(brief)
             brief["validation_scopes"] = [
@@ -431,6 +441,9 @@ class ValidationPolicy(_StrictModel):
                     ),
                     method="deterministic",
                 ))
+        checks.extend(quality_checks)
+        if any(check.scope not in scopes for check in quality_checks):
+            raise ValueError("quality checks must belong to an effective scope")
 
         manifest = [
             {
@@ -475,6 +488,8 @@ class ValidationPolicy(_StrictModel):
             "candidate_output": candidate_output,
             "model_by_scope": dict(model_by_scope),
             "tool_policy_version": self.tool_policy_version,
+            "subject": _jsonable(subject),
+            "quality_manifest": _jsonable(quality_manifest),
         }
         return ValidationPlan(
             policy_version=self.policy_version,
@@ -486,6 +501,8 @@ class ValidationPolicy(_StrictModel):
             effective_scopes=scopes,
             checks=checks,
             material_manifest=manifest,
+            subject=subject,
+            quality_manifest=[dict(item) for item in quality_manifest],
             plan_hash=_sha(base),
         )
 
@@ -620,6 +637,7 @@ def build_validation_packet(
     candidate_output: str | None = None,
     validation_plan: ValidationPlan | None = None,
     materials: Sequence[ValidationMaterial] = (),
+    fixed_checks: Mapping[str, ValidationCheck] | None = None,
     max_chars: int = MAX_VALIDATION_INPUT_CHARS,
 ) -> str:
     """固定保留任务合同、Plan和真实材料，再按最新优先裁剪主路径。"""
@@ -628,10 +646,20 @@ def build_validation_packet(
         completion_criteria = brief.get("completion_criteria")
     if expected_outputs is None and isinstance(brief, dict):
         expected_outputs = brief.get("expected_outputs")
+    serialized_plan = _jsonable(validation_plan)
+    if isinstance(serialized_plan, dict) and fixed_checks:
+        serialized_plan = dict(serialized_plan)
+        serialized_plan["checks"] = [
+            item for item in serialized_plan.get("checks", [])
+            if item.get("check_id") not in fixed_checks
+        ]
     packet: dict[str, Any] = {
         VALIDATION_PROTOCOL_MARKER: True,
         "validation_scope": validation_scope,
-        "validation_plan": _jsonable(validation_plan),
+        "validation_plan": serialized_plan,
+        "framework_checks": [
+            item.model_dump(mode="json") for item in (fixed_checks or {}).values()
+        ],
         "root_task_anchor": _jsonable(root_task_anchor),
         "task_brief": brief,
         "completion_criteria": _jsonable(completion_criteria or []),
@@ -696,6 +724,7 @@ def parse_scope_validation_result(
     expected_scope: ValidationScope,
     validator_trace_id: str,
     opened_source_ids: set[str] | None = None,
+    fixed_checks: Mapping[str, ValidationCheck] | None = None,
 ) -> ScopeValidationResult:
     """严格绑定Plan check_id，并由框架注入Validator Trace ID。"""
     raw = json.loads(content)
@@ -706,7 +735,19 @@ def parse_scope_validation_result(
         raise ValueError(
             f"validator returned scope {decision.scope!r}, expected {expected_scope!r}"
         )
-    planned = {item.check_id: item for item in plan.checks_for_scope(expected_scope)}
+    fixed_checks = fixed_checks or {}
+    all_planned = {
+        item.check_id: item for item in plan.checks_for_scope(expected_scope)
+    }
+    unknown_fixed = set(fixed_checks) - set(all_planned)
+    if unknown_fixed:
+        raise ValueError(
+            f"framework checks are not in plan: {sorted(unknown_fixed)}"
+        )
+    planned = {
+        check_id: spec for check_id, spec in all_planned.items()
+        if check_id not in fixed_checks
+    }
     returned_ids = [item.check_id for item in decision.checks]
     if len(returned_ids) != len(set(returned_ids)):
         raise ValueError("validator returned duplicate check_id values")
@@ -731,11 +772,18 @@ def parse_scope_validation_result(
         raise ValueError(
             f"{expected_scope} non-pass must use retry_from={expected_retry}"
         )
+    combined_by_id = {**fixed_checks, **{
+        item.check_id: item for item in decision.checks
+    }}
+    combined = [
+        combined_by_id[item.check_id]
+        for item in plan.checks_for_scope(expected_scope)
+    ]
     result = ScopeValidationResult(
         validator_trace_id=validator_trace_id,
         scope=expected_scope,
         outcome=decision.outcome,
-        checks=decision.checks,
+        checks=combined,
         reason=decision.reason,
         retry_from=decision.retry_from,
         plan_hash=plan.plan_hash,
@@ -837,6 +885,72 @@ def _framework_scope_result(
     )
 
 
+def _scope_result_with_fixed_checks(
+    *,
+    validator_trace_id: str,
+    scope: ValidationScope,
+    plan: ValidationPlan,
+    fixed_checks: Mapping[str, ValidationCheck],
+    error_reason: str | None = None,
+) -> ScopeValidationResult:
+    """Complete a blocked deterministic scope without asking the model."""
+    planned = plan.checks_for_scope(scope)
+    fixed_ids = set(fixed_checks)
+    unknown = fixed_ids - {item.check_id for item in planned}
+    if unknown:
+        raise ValueError(f"framework checks are not in plan: {sorted(unknown)}")
+    failing = next(
+        (item for item in fixed_checks.values() if item.status == "failed"),
+        None,
+    )
+    unresolved = next(
+        (item for item in fixed_checks.values() if item.status == "unknown"),
+        None,
+    )
+    primary = failing or unresolved
+    if error_reason is not None:
+        outcome: ValidationOutcome = "error"
+        reason = error_reason
+    elif failing is not None:
+        outcome = "failed"
+        reason = failing.issue or "Deterministic quality check failed"
+    elif unresolved is not None:
+        outcome = "unknown"
+        reason = unresolved.issue or "Deterministic quality check is unknown"
+    else:
+        outcome = "passed"
+        reason = "All deterministic checks passed"
+    checks: list[ValidationCheck] = []
+    for spec in planned:
+        if spec.check_id in fixed_checks:
+            checks.append(fixed_checks[spec.check_id])
+        elif outcome == "passed":
+            raise ValueError("scope still has model checks to execute")
+        else:
+            blocker = primary.check_id if primary else "quality provider"
+            checks.append(ValidationCheck(
+                check_id=spec.check_id,
+                status="unknown",
+                issue=f"Not evaluated because {blocker} blocked this scope: {reason}",
+            ))
+    result = ScopeValidationResult(
+        validator_trace_id=validator_trace_id,
+        scope=scope,
+        outcome=outcome,
+        checks=checks,
+        reason=reason,
+        retry_from=(
+            None if outcome in {"passed", "error"} else _SCOPE_RETRY[scope]
+        ),
+        plan_hash=plan.plan_hash,
+    )
+    return _require_scope_result_integrity(
+        result,
+        plan=plan,
+        expected_scope=scope,
+    )
+
+
 def scope_validation_error(
     *,
     validator_trace_id: str,
@@ -965,9 +1079,13 @@ class LLMValidator:
         source_urls: Sequence[str] = (),
         resume_scope_results: Sequence[ScopeValidationResult] = (),
         on_scope_result: ScopeResultCallback | None = None,
+        fixed_checks: Mapping[str, ValidationCheck] | None = None,
+        fixed_scope_errors: Mapping[ValidationScope, str] | None = None,
     ) -> ValidationRun:
         """顺序执行所有Scope；每完成一项即可回调持久化断点。"""
         started = time.monotonic()
+        fixed_checks = fixed_checks or {}
+        fixed_scope_errors = fixed_scope_errors or {}
         runs: list[_ScopeRun] = []
         resumed: dict[ValidationScope, ScopeValidationResult] = {}
         for item in resume_scope_results:
@@ -993,7 +1111,31 @@ class LLMValidator:
             )
             try:
                 blocking_issue = _blocking_material_issue(material_issues, scope)
-                if blocking_issue is not None:
+                scope_fixed = {
+                    check_id: check
+                    for check_id, check in fixed_checks.items()
+                    if check_id in {
+                        item.check_id for item in plan.checks_for_scope(scope)
+                    }
+                }
+                fixed_blocked = (
+                    scope in fixed_scope_errors
+                    or any(item.status != "passed" for item in scope_fixed.values())
+                )
+                remaining_checks = [
+                    item for item in plan.checks_for_scope(scope)
+                    if item.check_id not in scope_fixed
+                ]
+                if fixed_blocked or (scope_fixed and not remaining_checks):
+                    run = await self.record_scope_fixed(
+                        evaluated_trace=evaluated_trace,
+                        plan=plan,
+                        scope=scope,
+                        fixed_checks=scope_fixed,
+                        error_reason=fixed_scope_errors.get(scope),
+                        validator_trace_id=trace_id,
+                    )
+                elif blocking_issue is not None:
                     run = await self.record_scope_non_success(
                         evaluated_trace=evaluated_trace,
                         plan=plan,
@@ -1017,6 +1159,7 @@ class LLMValidator:
                         model=model_by_scope.get(scope, ""),
                         source_urls=source_urls,
                         validator_trace_id=trace_id,
+                        fixed_checks=scope_fixed,
                     )
             finally:
                 if self.trace_release:
@@ -1079,6 +1222,7 @@ class LLMValidator:
         source_urls: Sequence[str],
         validator_trace_id: str,
         task_progress: Mapping[str, Any] | BaseModel | None = None,
+        fixed_checks: Mapping[str, ValidationCheck] | None = None,
     ) -> _ScopeRun:
         """创建一个Scope Trace并运行严格白名单的模型/工具循环。"""
         session = (
@@ -1108,6 +1252,7 @@ class LLMValidator:
                 candidate_output=candidate_output,
                 validation_plan=plan,
                 materials=materials,
+                fixed_checks=fixed_checks,
                 max_chars=self.max_input_chars,
             )
         except Exception as exc:
@@ -1164,6 +1309,7 @@ class LLMValidator:
                         expected_scope=scope,
                         validator_trace_id=validator_trace_id,
                         opened_source_ids=(session.opened_source_ids if session else set()),
+                        fixed_checks=fixed_checks,
                     )
                     duration_ms = int((time.monotonic() - started) * 1000)
                     await self._finish_trace(
@@ -1241,6 +1387,35 @@ class LLMValidator:
                 duration_ms=int((time.monotonic() - started) * 1000),
             )
 
+    async def record_scope_fixed(
+        self,
+        *,
+        evaluated_trace: Trace,
+        plan: ValidationPlan,
+        scope: ValidationScope,
+        fixed_checks: Mapping[str, ValidationCheck],
+        error_reason: str | None,
+        validator_trace_id: str,
+    ) -> _ScopeRun:
+        """Persist an all-deterministic or deterministically blocked scope."""
+        trace = self._new_trace(
+            trace_id=validator_trace_id,
+            evaluated_trace=evaluated_trace,
+            model=None,
+            scope=scope,
+            plan_hash=plan.plan_hash,
+        )
+        await self.trace_store.create_trace(trace)
+        result = _scope_result_with_fixed_checks(
+            validator_trace_id=validator_trace_id,
+            scope=scope,
+            plan=plan,
+            fixed_checks=fixed_checks,
+            error_reason=error_reason,
+        )
+        await self._store_terminal_message(trace, result)
+        return _ScopeRun(result=result, trace_id=validator_trace_id)
+
     async def record_scope_non_success(
         self,
         *,

cyber_agent/tools/builtin/subagent.py

@@ -356,12 +356,26 @@ async def _record_pending_task_reports(
                 },
             )
         validation_result = validation_run.result
-        state["pending_reviews"][child_trace_id] = pending_review_entry(
+        candidate_validations = []
+        for candidate_ref in report.candidate_refs:
+            candidate_run = await runner.validate_recursive_trace(
+                child_trace_id,
+                task_brief=task_brief,
+                task_report=report_payload,
+                candidate_ref=candidate_ref,
+            )
+            candidate_validations.append({
+                "candidate_ref": candidate_ref.model_dump(mode="json"),
+                "validation_result": candidate_run.result.model_dump(mode="json"),
+            })
+        pending_entry = pending_review_entry(
             goal_id=goal_id,
             report=report,
             validation_result=validation_result.model_dump(),
             received_at_sequence=received_at_sequence,
         )
+        pending_entry["candidate_validations"] = candidate_validations
+        state["pending_reviews"][child_trace_id] = pending_entry
         reports.append(report)
     await store.update_trace(parent_trace.trace_id, context=parent_trace.context)
     if goal_id:

cyber_agent/tools/builtin/task_protocol.py

@@ -35,6 +35,7 @@ from cyber_agent.core.task_protocol import (
 )
 from cyber_agent.core.task_protocol_service import TaskProtocolStateError
 from cyber_agent.core.validation import ValidationResult
+from cyber_agent.application.candidate import CandidateReviewAction
 from cyber_agent.tools import tool
 
 
@@ -153,6 +154,19 @@ async def submit_task_report(
         )
     except ValidationError as exc:
         return _error(f"Invalid TaskReport: {exc}")
+    if parsed_submission.candidate_refs:
+        candidate_service = context.get("candidate_service")
+        if candidate_service is None:
+            return _error(
+                "TaskReport CandidateRefs require the bound CandidateService"
+            )
+        try:
+            await candidate_service.validate_report_refs(
+                trace_id,
+                parsed_submission.candidate_refs,
+            )
+        except ValueError as exc:
+            return _error(f"Invalid TaskReport CandidateRefs: {exc}")
     if policy.requires_task_progress and parsed_submission.outcome == "satisfied":
         readiness_error = task_progress_readiness_error(state)
         if readiness_error:
@@ -206,6 +220,7 @@ async def review_task_result(
     decision: ReviewDecision,
     reason: str,
     approved_next_task: TaskBrief | None = None,
+    candidate_actions: list[CandidateReviewAction] | None = None,
     context: dict | None = None,
 ) -> dict[str, Any]:
     """审核一份直属孩子的待处理 ``TaskReport``。
@@ -305,6 +320,7 @@ async def review_task_result(
                 if decision == "REPLAN_CURRENT"
                 else None
             ),
+            "candidate_actions": candidate_actions or [],
         })
         report = TaskReport.model_validate(entry["task_report"])
         if report.child_trace_id != child_trace_id:
@@ -372,6 +388,20 @@ async def review_task_result(
             return _error(f"Approved TaskBrief context is invalid: {exc}")
 
     reviewed_at_sequence = int(context.get("sequence", 0) or 0)
+    if review.candidate_actions:
+        candidate_service = context.get("candidate_service")
+        if candidate_service is None:
+            return _error("Candidate actions require the bound CandidateService")
+        try:
+            await candidate_service.apply_review_actions(
+                parent_trace_id,
+                child_trace_id,
+                report_refs=report.candidate_refs,
+                actions=review.candidate_actions,
+                effective_at_sequence=reviewed_at_sequence,
+            )
+        except ValueError as exc:
+            return _error(f"Invalid candidate action: {exc}")
     context_ref = None
     context_ref_warning = None
     try:

tests/test_candidate_quality.py

@@ -0,0 +1,268 @@
+import json
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cyber_agent.application import (
+    AgentApplication,
+    AgentRole,
+    ApplicationRegistry,
+    ApplicationRuntime,
+    ApplicationServices,
+    ProviderRef,
+    QualityRuleSpec,
+)
+from cyber_agent.application.candidate import CandidateLedger
+from cyber_agent.application.quality import QualityCheckOutcome
+from cyber_agent.core.artifacts import (
+    ArtifactRef,
+    ValidationMaterial,
+    material_content_hash,
+)
+from cyber_agent.tools.registry import ToolRegistry
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+class CandidateRepository:
+    def __init__(self):
+        self.requests = {}
+
+    async def put_version(self, request):
+        self.requests.setdefault(request.operation_id, request)
+        saved = self.requests[request.operation_id]
+        return ArtifactRef(
+            artifact_id=f"candidate:{saved.candidate_id}:{saved.revision}",
+            version=str(saved.revision),
+            content_hash=material_content_hash(saved.content),
+            kind="candidate.output",
+            mime_type="application/json",
+        )
+
+    async def merge(self, request):
+        return await self.put_version(request)
+
+    async def load_version(self, candidate_ref):
+        return next(
+            item.content for item in self.requests.values()
+            if item.candidate_id == candidate_ref.candidate_id
+            and item.revision == candidate_ref.revision
+        )
+
+    async def resolve(self, ref, root_trace_id, uid):
+        request = next(
+            item for item in self.requests.values()
+            if ref.artifact_id == f"candidate:{item.candidate_id}:{item.revision}"
+        )
+        return ValidationMaterial(
+            **ref.model_dump(mode="json"),
+            root_trace_id=root_trace_id,
+            uid=uid,
+            content=request.content,
+        )
+
+    async def commit_adoption(self, request):
+        raise NotImplementedError
+
+
+class QualityProvider:
+    def __init__(self):
+        self.mode = "normal"
+        self.calls = 0
+
+    async def check(self, request):
+        self.calls += 1
+        outcomes = []
+        for rule in request.rules:
+            failed = "{{placeholder}}" in request.material.content["text"]
+            outcomes.append(QualityCheckOutcome(
+                rule_id=rule.rule_id,
+                rule_version=rule.version,
+                status="failed" if failed else "passed",
+                issue="placeholder remains" if failed else None,
+            ))
+        if self.mode == "missing":
+            return []
+        if self.mode == "extra":
+            outcomes.append(QualityCheckOutcome(
+                rule_id="unexpected",
+                rule_version="1",
+                status="passed",
+            ))
+        if self.mode == "error":
+            raise RuntimeError("provider unavailable")
+        return outcomes
+
+
+class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp.name)
+        self.repository = CandidateRepository()
+        self.quality = QualityProvider()
+        application = AgentApplication(
+            application_id="quality.reference",
+            application_version="1",
+            root_role="writer",
+            roles=(AgentRole(
+                role_id="writer",
+                model="validator-model",
+                system_prompt="write",
+            ),),
+            artifact_resolver_ref=ProviderRef(
+                provider_id="artifacts",
+                provider_version="1",
+            ),
+            candidate_repository_ref=ProviderRef(
+                provider_id="candidates",
+                provider_version="1",
+            ),
+            quality_provider_ref=ProviderRef(
+                provider_id="quality",
+                provider_version="1",
+            ),
+            quality_rules=(QualityRuleSpec(
+                rule_id="no_placeholder",
+                version="1",
+                criterion="Candidate contains no placeholder text",
+            ),),
+        )
+        registry = ApplicationRegistry()
+        self.binding = registry.register(
+            application,
+            ApplicationServices(
+                tool_registry=ToolRegistry(),
+                artifact_resolver=self.repository,
+                candidate_repository=self.repository,
+                quality_provider=self.quality,
+            ),
+        )
+        self.llm_calls = 0
+
+        async def llm_call(**kwargs):
+            self.llm_calls += 1
+            packet = json.loads(kwargs["messages"][-1]["content"])
+            scope = packet["validation_scope"]
+            return {
+                "content": json.dumps({
+                    "scope": scope,
+                    "outcome": "passed",
+                    "checks": [
+                        {
+                            "check_id": item["check_id"],
+                            "status": "passed",
+                            "evidence_refs": [],
+                            "issue": None,
+                        }
+                        for item in packet["validation_plan"]["checks"]
+                    ],
+                    "reason": "candidate output is complete",
+                    "retry_from": None,
+                }),
+                "tool_calls": [],
+            }
+
+        runtime = ApplicationRuntime(
+            registry=registry,
+            trace_store=self.store,
+            llm_call=llm_call,
+        )
+        self.runner, config = runtime.new_run(
+            "quality.reference",
+            "1",
+            uid="user-1",
+            root_task_anchor={
+                "objective": "write",
+                "completion_criteria": ["complete"],
+                "constraints": [],
+            },
+        )
+        with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
+            self.trace, _goal, _sequence = await self.runner._prepare_new_trace(
+                [{"role": "user", "content": "start"}],
+                config,
+            )
+
+    async def asyncTearDown(self):
+        self.temp.cleanup()
+
+    async def _candidate(self, text, sequence):
+        return await self.runner.candidate_service.manage(
+            self.trace.trace_id,
+            operation="create",
+            content={"text": text},
+            parent_refs=[],
+            effective_at_sequence=sequence,
+        )
+
+    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(
+            self.trace.trace_id,
+            candidate_ref=candidate,
+        )
+        checks = first.result.scope_results[0].checks
+        self.assertEqual("passed", first.result.outcome)
+        self.assertIn("quality.no_placeholder", [item.check_id for item in checks])
+        self.assertTrue(any(item.check_id.startswith("output.policy") for item in checks))
+        self.assertEqual(1, self.quality.calls)
+        self.assertEqual(1, self.llm_calls)
+
+        second = await self.runner.validate_recursive_trace(
+            self.trace.trace_id,
+            candidate_ref=candidate,
+        )
+        self.assertTrue(second.cached)
+        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.validations))
+
+    async def test_failed_candidate_does_not_call_llm_or_pollute_peer(self):
+        good = await self._candidate("finished", 2)
+        bad = await self._candidate("{{placeholder}}", 3)
+        good_run = await self.runner.validate_recursive_trace(
+            self.trace.trace_id,
+            candidate_ref=good,
+        )
+        bad_run = await self.runner.validate_recursive_trace(
+            self.trace.trace_id,
+            candidate_ref=bad,
+        )
+        self.assertEqual("passed", good_run.result.outcome)
+        self.assertEqual("failed", bad_run.result.outcome)
+        self.assertEqual(1, self.llm_calls)
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger(self.trace.trace_id)
+        )
+        self.assertEqual(2, len(ledger.validations))
+
+    async def test_provider_shape_and_exception_fail_closed_with_complete_plan(self):
+        for index, mode in enumerate(("missing", "extra", "error"), start=2):
+            with self.subTest(mode=mode):
+                self.quality.mode = mode
+                candidate = await self._candidate(f"finished-{mode}", index)
+                run = await self.runner.validate_recursive_trace(
+                    self.trace.trace_id,
+                    candidate_ref=candidate,
+                )
+                result = run.result.scope_results[0]
+                self.assertEqual("error", result.outcome)
+                self.assertEqual(
+                    {
+                        item.check_id
+                        for item in result.checks
+                    },
+                    {
+                        "quality.no_placeholder",
+                        *{
+                            item.check_id for item in result.checks
+                            if item.check_id.startswith("output.policy")
+                        },
+                    },
+                )
+                self.assertTrue(all(
+                    item.status == "unknown" for item in result.checks
+                ))
+        self.assertEqual(0, self.llm_calls)

tests/test_candidate_service.py

@@ -12,9 +12,12 @@ from cyber_agent.application import (
     ProviderRef,
 )
 from cyber_agent.application.candidate import (
+    CandidateAdoptionReceipt,
     CandidateLedger,
     CandidatePointer,
+    CandidateReviewAction,
 )
+from cyber_agent.application.quality import CandidateValidationRecord
 from cyber_agent.application.candidate_service import (
     CandidateService,
     CandidateStateError,
@@ -25,6 +28,11 @@ from cyber_agent.core.artifacts import (
     material_content_hash,
 )
 from cyber_agent.core.task_protocol import TaskBrief, new_task_protocol
+from cyber_agent.core.validation import (
+    ScopeValidationResult,
+    ValidationCheck,
+    ValidationResult,
+)
 from cyber_agent.tools.registry import ToolRegistry
 from cyber_agent.trace.models import Trace
 from cyber_agent.trace.store import (
@@ -39,6 +47,8 @@ class MemoryCandidateRepository:
         self.put_calls = 0
         self.merge_calls = 0
         self.bad_artifact = False
+        self.adoptions = {}
+        self.adoption_calls = 0
 
     async def put_version(self, request):
         self.put_calls += 1
@@ -90,7 +100,12 @@ class MemoryCandidateRepository:
         )
 
     async def commit_adoption(self, request):
-        raise NotImplementedError
+        self.adoption_calls += 1
+        self.adoptions.setdefault(request.operation_id, request.candidate_ref)
+        return CandidateAdoptionReceipt(
+            operation_id=request.operation_id,
+            external_id=f"published:{request.candidate_ref.candidate_id}",
+        )
 
 
 def application():
@@ -177,6 +192,43 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             ),
         ))
 
+    async def _record_validation(self, trace_id, candidate, outcome="passed"):
+        plan_hash = material_content_hash({
+            "candidate": candidate.model_dump(mode="json"),
+            "outcome": outcome,
+        })
+        status = "passed" if outcome == "passed" else "failed"
+        scope = ScopeValidationResult(
+            validator_trace_id=f"{trace_id}@validator",
+            scope="output",
+            outcome=outcome,
+            checks=[ValidationCheck(
+                check_id="quality.rule",
+                status=status,
+                issue=None if status == "passed" else "needs revision",
+            )],
+            reason="checked",
+            retry_from=None if outcome == "passed" else "output",
+            plan_hash=plan_hash,
+        )
+        result = ValidationResult(
+            evaluated_trace_id=trace_id,
+            outcome=outcome,
+            scope_results=[scope],
+            issues=[] if outcome == "passed" else ["needs revision"],
+            retry_from=None if outcome == "passed" else "output",
+            plan_hash=plan_hash,
+        )
+        await self.service.record_validation(
+            trace_id,
+            CandidateValidationRecord(
+                candidate_ref=candidate,
+                plan_hash=plan_hash,
+                validation_result=result.model_dump(mode="json"),
+                validated_at_sequence=candidate.created_at_sequence,
+            ),
+        )
+
     async def test_create_fork_merge_and_idempotent_recovery(self):
         first, duplicate = await asyncio.gather(
             self.service.manage(
@@ -341,6 +393,90 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         with self.assertRaisesRegex(TraceStoreCorruptionError, "candidate ledger"):
             await self.store.get_candidate_ledger("root-a")
 
+    async def test_review_adoption_is_idempotent_and_creates_rewind_barrier(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("child-a", candidate)
+        action = CandidateReviewAction(
+            action="adopt",
+            candidate_ref=candidate,
+            reason="selected exact revision",
+        )
+        first = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=9,
+        )
+        second = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=9,
+        )
+        self.assertEqual(first, second)
+        self.assertEqual(1, self.repository.adoption_calls)
+        self.assertEqual(1, len(self.repository.adoptions))
+        with self.assertRaisesRegex(CandidateStateError, "committed"):
+            await self.service.assert_rewind_allowed("root-a", 8)
+        await self.service.assert_rewind_allowed("root-a", 9)
+        with self.assertRaisesRegex(CandidateStateError, "terminal"):
+            await self.service.apply_review_actions(
+                "root-a",
+                "child-a",
+                report_refs=[candidate],
+                actions=[CandidateReviewAction(
+                    action="discard",
+                    candidate_ref=candidate,
+                    reason="too late",
+                )],
+                effective_at_sequence=10,
+            )
+
+    async def test_failed_candidate_can_only_be_revised_or_discarded(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "placeholder"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("child-a", candidate, outcome="failed")
+        with self.assertRaisesRegex(CandidateStateError, "passed"):
+            await self.service.apply_review_actions(
+                "root-a",
+                "child-a",
+                report_refs=[candidate],
+                actions=[CandidateReviewAction(
+                    action="adopt",
+                    candidate_ref=candidate,
+                    reason="invalid",
+                )],
+                effective_at_sequence=9,
+            )
+        records = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[CandidateReviewAction(
+                action="revise",
+                candidate_ref=candidate,
+                reason="remove placeholder",
+            )],
+            effective_at_sequence=10,
+        )
+        self.assertEqual("discarded", records[0].state)
+        self.assertIn("retry_from=output", records[0].reason)
+
 
 if __name__ == "__main__":
     unittest.main()