test(Agent验证): 覆盖预检短路与计划回读

新增硬性预检失败不调用模型、预检通过继续语义验证的参数化用例。同步验证 ValidationPlan 往返序列化、验证结果与零 token 统计，确保新字段兼容既有编排协议。

agent/tests/test_orchestration_validation_policy.py

@@ -50,11 +50,13 @@ def make_context(root: str = "root") -> ValidationContext:
         goal_id=None,
         parent_task_id=None,
         display_path="1",
-        specs=[TaskSpec(
-            version=1,
-            objective="validate",
-            acceptance_criteria=[AcceptanceCriterion("criterion", "validate result")],
-        )],
+        specs=[
+            TaskSpec(
+                version=1,
+                objective="validate",
+                acceptance_criteria=[AcceptanceCriterion("criterion", "validate result")],
+            )
+        ],
     )
     attempt = TaskAttempt(
         attempt_id="attempt-1",
@@ -140,9 +142,7 @@ async def test_default_policy_and_validation_plan_roundtrip():
         max_evidence_queries=2,
     )
     assert ValidationPlan.from_dict(deterministic.to_dict()) == deterministic
-    assert ValidationPlan.from_dict(
-        {"mode": "deterministic", "rule_ids": []}
-    ).validator_preset is None
+    assert ValidationPlan.from_dict({"mode": "deterministic", "rule_ids": []}).validator_preset is None
     with pytest.raises(ValueError, match="cannot contain"):
         ValidationPlan(rule_ids=["unexpected"])
     with pytest.raises(ValueError, match="cannot be blank"):
@@ -219,18 +219,12 @@ async def test_coordinator_freezes_default_agent_plan(tmp_path):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "verdict", [ValidationVerdict.PASSED, ValidationVerdict.FAILED]
-)
-async def test_coordinator_executes_deterministic_plan_without_agent_validator(
-    tmp_path, verdict
-):
+@pytest.mark.parametrize("verdict", [ValidationVerdict.PASSED, ValidationVerdict.FAILED])
+async def test_coordinator_executes_deterministic_plan_without_agent_validator(tmp_path, verdict):
     executor = FakeExecutor([])
     coordinator, store, _ = await make_coordinator(tmp_path, executor)
     coordinator.validation_policy = DeterministicPolicy()
-    coordinator.deterministic_validator = RuleBasedDeterministicValidator(
-        [StaticRule("c1", verdict)]
-    )
+    coordinator.deterministic_validator = RuleBasedDeterministicValidator([StaticRule("c1", verdict)])
     task_id = await create_task(coordinator, "deterministic")
 
     result = (await coordinator.dispatch_tasks("root", [task_id]))[0]
@@ -243,6 +237,33 @@ async def test_coordinator_executes_deterministic_plan_without_agent_validator(
     assert report.criterion_results[0].criterion_id == "c1"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("preflight_verdict", "validator_calls", "final_verdict"),
+    [
+        (ValidationVerdict.FAILED, 0, ValidationVerdict.FAILED),
+        (ValidationVerdict.PASSED, 1, ValidationVerdict.PASSED),
+    ],
+)
+async def test_agent_validation_skips_model_on_hard_preflight_failure(
+    tmp_path, preflight_verdict, validator_calls, final_verdict
+):
+    executor = FakeExecutor([ValidationVerdict.PASSED])
+    coordinator, store, _ = await make_coordinator(tmp_path, executor)
+    coordinator.validation_policy = FixedPolicy(ValidationPlan(preflight_rule_ids=("preflight",)))
+    coordinator.deterministic_validator = RuleBasedDeterministicValidator([StaticRule("preflight", preflight_verdict)])
+    task_id = await create_task(coordinator, "hybrid preflight")
+
+    result = (await coordinator.dispatch_tasks("root", [task_id]))[0]
+    report = (await store.load("root")).validations[result.validation_id]
+
+    assert executor.validator_calls == validator_calls
+    assert report.verdict == final_verdict
+    if preflight_verdict is ValidationVerdict.FAILED:
+        assert "Deterministic preflight rejected" in report.summary
+        assert report.execution_stats.total_tokens == 0
+
+
 def test_wire_orchestration_preserves_legacy_positional_defaults(tmp_path):
     runner = SimpleNamespace(trace_store=None, task_coordinator=None)
     config = OrchestrationConfig(max_parallel_tasks=2)
@@ -268,9 +289,7 @@ def test_wire_orchestration_preserves_legacy_positional_defaults(tmp_path):
 async def test_wire_orchestration_applies_custom_validation_dependencies(tmp_path):
     runner = SimpleNamespace(trace_store=None, task_coordinator=None)
     policy = DeterministicPolicy()
-    validator = RuleBasedDeterministicValidator(
-        [StaticRule("c1", ValidationVerdict.PASSED)]
-    )
+    validator = RuleBasedDeterministicValidator([StaticRule("c1", ValidationVerdict.PASSED)])
     provider = EvidenceProvider()
     coordinator = wire_orchestration(
         runner,
@@ -287,9 +306,7 @@ async def test_wire_orchestration_applies_custom_validation_dependencies(tmp_pat
     task_id = await create_task(coordinator, "wired deterministic validation")
 
     result = (await coordinator.dispatch_tasks("root", [task_id]))[0]
-    report = (await coordinator.task_store.load("root")).validations[
-        result.validation_id
-    ]
+    report = (await coordinator.task_store.load("root")).validations[result.validation_id]
 
     assert runner.task_coordinator is coordinator
     assert coordinator.validation_policy is policy
@@ -302,9 +319,7 @@ async def test_wire_orchestration_applies_custom_validation_dependencies(tmp_pat
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("rule_ids", [[], ["unknown"]])
-async def test_deterministic_plan_must_cover_hard_criteria_without_unknowns(
-    tmp_path, rule_ids
-):
+async def test_deterministic_plan_must_cover_hard_criteria_without_unknowns(tmp_path, rule_ids):
     executor = FakeExecutor([])
     coordinator, _, _ = await make_coordinator(tmp_path, executor)
     coordinator.validation_policy = FixedPolicy(
@@ -364,11 +379,7 @@ class EvidenceProvider:
         if self.error:
             raise self.error
         return EvidenceProviderResult(
-            items=(
-                self.items
-                if self.items is not None
-                else [{"index": index} for index in range(self.item_count)]
-            ),
+            items=(self.items if self.items is not None else [{"index": index} for index in range(self.item_count)]),
             evidence_refs=[ArtifactRef(uri="memory://evidence", version="1")],
         )
 
@@ -470,9 +481,7 @@ async def test_wire_orchestration_applies_custom_evidence_provider(tmp_path):
             limit=100,
         )
         response = await coordinator.query_evidence(actor, request)
-        report = (await coordinator.task_store.load("root")).validations[
-            request.validation_id
-        ]
+        report = (await coordinator.task_store.load("root")).validations[request.validation_id]
 
         assert coordinator.validation_policy is policy
         assert coordinator.evidence_provider is provider
@@ -487,9 +496,7 @@ async def test_wire_orchestration_applies_custom_evidence_provider(tmp_path):
 @pytest.mark.asyncio
 async def test_coordinator_evidence_ownership_and_persistent_budget(tmp_path):
     provider = EvidenceProvider()
-    coordinator, store, operation, actor, request = await running_evidence_validation(
-        tmp_path, provider
-    )
+    coordinator, store, operation, actor, request = await running_evidence_validation(tmp_path, provider)
     response = await coordinator.query_evidence(actor, request)
     assert response.queries_used == 1
     assert response.queries_remaining == 0
@@ -529,9 +536,7 @@ async def test_coordinator_evidence_ownership_and_persistent_budget(tmp_path):
         (EvidenceProvider(items=[{"bad": object()}]), "non-JSON"),
     ],
 )
-async def test_evidence_provider_failures_are_validation_errors(
-    tmp_path, provider, message
-):
+async def test_evidence_provider_failures_are_validation_errors(tmp_path, provider, message):
     timeout = 0.001 if provider.delay else 0.1
     coordinator, store, operation, actor, request = await running_evidence_validation(
         tmp_path,