refactor(application): 收紧投影时机和验证失败边界

协议服务在持有 Task/Candidate 锁时只追加规范事件，不同步调用业务 Projector；投影统一由 Run 完成、恢复和 API 对账驱动，避免外部事务延长内核临界区。

确定性质量规则的预算登记失败现在也转换为计划完整的 error 结果，不在候选缓存创建前抛出半完成状态；应用事件全量对账按根隔离错误，单个损坏根不阻断其他应用。

cyber_agent/application/candidate_service.py

@@ -282,15 +282,15 @@ class CandidateService:
                     "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
-                )
-            )
+            retained = []
+            for raw in ledger.validations:
+                item = CandidateValidationRecord.model_validate(raw)
+                if (
+                    item.candidate_ref == record.candidate_ref
+                    and item.plan_hash == record.plan_hash
+                ):
+                    continue
+                retained.append(raw)
             updated = ledger.model_copy(update={
                 "validations": (
                     *retained,

cyber_agent/application/events.py

@@ -162,7 +162,7 @@ class RunEventService:
     async def emit_after_commit(self, **kwargs: Any) -> None:
         """Never roll back protocol state because journaling/projection failed."""
         try:
-            await self.emit(**kwargs)
+            await self.emit(project=False, **kwargs)
         except Exception:
             logger.exception("Failed to emit/project canonical run event")
 

cyber_agent/application/runtime.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import logging
 from types import MappingProxyType
 from typing import Any, Callable, Mapping
 
@@ -21,6 +22,9 @@ from cyber_agent.core.validation import ValidationPolicy
 from cyber_agent.tools.registry import ToolRegistry
 
 
+logger = logging.getLogger(__name__)
+
+
 @dataclass(frozen=True)
 class ApplicationServices:
     """Non-serializable providers and private tool implementations."""
@@ -290,7 +294,13 @@ class ApplicationRuntime:
         }
         total = 0
         for root_trace_id in sorted(roots):
-            total += await self.reconcile_events(root_trace_id)
+            try:
+                total += await self.reconcile_events(root_trace_id)
+            except Exception:
+                logger.exception(
+                    "Application event reconciliation failed for root %s",
+                    root_trace_id,
+                )
         return total
 
     def new_run(

cyber_agent/core/runner.py

@@ -1071,10 +1071,13 @@ class AgentRunner:
             )
             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),
-        )
+        try:
+            await self.record_recursive_validation_usage(
+                subject.trace_id,
+                tool_calls=len(rules),
+            )
+        except Exception as exc:
+            return provider_error(f"Quality validation budget failed: {exc}")
         return checks, errors
 
     async def dream(

tests/test_application_reference.py

@@ -286,6 +286,7 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
             self.assertEqual(1, components.candidates.adoption_calls)
             self.assertEqual(1, len(components.candidates.adoptions))
 
+            await restored_runner.event_service.try_pump(root.trace_id)
             before_replay = dict(components.projector.rows)
             await runtime.reconcile_events(root.trace_id)
             await runtime.reconcile_events(root.trace_id)