fix(validator): 保证校验计划与所有结果完全一致

将 Validator policy 升级到 recursive-validator-2.2，新计划仅为真实材料解析硬错误生成 deterministic，其余任务、输出与根约束统一交由 llm 检查。

LLM 返回、材料失败、输入超限、取消和执行异常全部使用计划内的完整 check_id 集合；阻断后的未执行项显式标记 unknown，不再制造 input_limit 或 framework.error 等计划外 ID。

ScopeValidationResult 在构造、checkpoint 恢复和聚合时重复校验 outcome 与逐项状态的一致性，阻断伪造 passed 但 checks 实际失败的缓存结果。

补充正常输出、框架失败、checkpoint 恢复、生命周期和 replan 回归，确保 plan_hash 变化后旧断点失效。

cyber_agent/core/validation.py

@@ -231,6 +231,7 @@ class ScopeValidationResult(_StrictModel):
 
     @model_validator(mode="after")
     def validate_outcome(self) -> "ScopeValidationResult":
+        _require_scope_outcome_matches_checks(self.outcome, self.checks)
         if self.outcome == "passed" and self.retry_from is not None:
             raise ValueError("passed requires retry_from=null")
         if self.outcome in {"failed", "unknown"} and self.retry_from is None:
@@ -240,6 +241,34 @@ class ScopeValidationResult(_StrictModel):
         return self
 
 
+def _require_scope_outcome_matches_checks(
+    outcome: ValidationOutcome,
+    checks: Sequence[ValidationCheck],
+) -> None:
+    """防止 Scope 总结论与逐项状态相互矛盾。"""
+    statuses = {check.status for check in checks}
+    if not statuses:
+        raise ValueError("scope result must contain at least one check")
+    if outcome == "passed":
+        if statuses != {"passed"}:
+            raise ValueError("passed scope requires every check to pass")
+        return
+    if outcome == "failed":
+        if "failed" not in statuses:
+            raise ValueError("failed scope requires at least one failed check")
+        return
+    if outcome == "unknown":
+        if "failed" in statuses or "unknown" not in statuses:
+            raise ValueError(
+                "unknown scope requires an unknown check and no failed checks"
+            )
+        return
+    if "failed" in statuses or "unknown" not in statuses:
+        raise ValueError(
+            "error scope requires an unknown check and no failed checks"
+        )
+
+
 class ValidationResult(_StrictModel):
     """所有有效 Scope 的聚合结果；直接父级只审核这一份。"""
 
@@ -277,7 +306,7 @@ class ValidationResult(_StrictModel):
 class ValidationPolicy(_StrictModel):
     """由可信应用注入并在根 Trace 固化的最低验收规则。"""
 
-    policy_version: str = "recursive-validator-2.1"
+    policy_version: str = "recursive-validator-2.2"
     global_rules: str = _GLOBAL_RULES
     rubrics: dict[ValidationScope, list[str]] = Field(
         default_factory=lambda: {
@@ -360,14 +389,14 @@ class ValidationPolicy(_StrictModel):
                         check_id=f"task.criterion.{index}",
                         scope=scope,
                         criterion=str(criterion),
-                        method="deterministic_and_llm",
+                        method="llm",
                     ))
                 for index, expected in enumerate(brief.get("expected_outputs", []), start=1):
                     checks.append(ValidationCheckSpec(
                         check_id=f"task.expected_output.{index}",
                         scope=scope,
                         criterion=f"Expected output is actually delivered: {expected}",
-                        method="deterministic_and_llm",
+                        method="llm",
                     ))
             if scope == "root":
                 for index, criterion in enumerate(anchor.get("completion_criteria", []), start=1):
@@ -375,14 +404,14 @@ class ValidationPolicy(_StrictModel):
                         check_id=f"root.criterion.{index}",
                         scope=scope,
                         criterion=str(criterion),
-                        method="deterministic_and_llm",
+                        method="llm",
                     ))
                 for index, constraint in enumerate(anchor.get("constraints", []), start=1):
                     checks.append(ValidationCheckSpec(
                         check_id=f"root.constraint.{index}",
                         scope=scope,
                         criterion=f"Global constraint is satisfied: {constraint}",
-                        method="deterministic_and_llm",
+                        method="llm",
                     ))
             scoped_issues = [
                 issue for issue in material_issues if scope in issue.scopes
@@ -682,7 +711,7 @@ def parse_scope_validation_result(
         raise ValueError(
             f"{expected_scope} non-pass must use retry_from={expected_retry}"
         )
-    return ScopeValidationResult(
+    result = ScopeValidationResult(
         validator_trace_id=validator_trace_id,
         scope=expected_scope,
         outcome=decision.outcome,
@@ -691,24 +720,116 @@ def parse_scope_validation_result(
         retry_from=decision.retry_from,
         plan_hash=plan.plan_hash,
     )
+    return _require_scope_result_integrity(
+        result,
+        plan=plan,
+        expected_scope=expected_scope,
+    )
 
 
-def scope_validation_error(
+def _require_scope_result_integrity(
+    result: ScopeValidationResult,
+    *,
+    plan: ValidationPlan,
+    expected_scope: ValidationScope,
+) -> ScopeValidationResult:
+    """对 LLM、框架生成和断点恢复结果执行同一份 Plan 约束。"""
+    if result.scope != expected_scope:
+        raise ValueError(
+            f"scope result has scope {result.scope!r}, expected {expected_scope!r}"
+        )
+    if result.plan_hash != plan.plan_hash:
+        raise ValueError("scope result plan_hash does not match validation plan")
+    _require_scope_outcome_matches_checks(result.outcome, result.checks)
+
+    planned_ids = [item.check_id for item in plan.checks_for_scope(expected_scope)]
+    returned_ids = [item.check_id for item in result.checks]
+    if len(returned_ids) != len(set(returned_ids)):
+        raise ValueError("scope result contains duplicate check_id values")
+    extras = set(returned_ids) - set(planned_ids)
+    missing = set(planned_ids) - set(returned_ids)
+    if extras or missing:
+        raise ValueError(
+            "scope result checks do not match plan: "
+            f"extras={sorted(extras)}, missing={sorted(missing)}"
+        )
+    by_id = {item.check_id: item for item in result.checks}
+    required_statuses = {
+        by_id[item.check_id].status
+        for item in plan.checks_for_scope(expected_scope)
+        if item.required
+    }
+    if result.outcome == "passed" and required_statuses != {"passed"}:
+        raise ValueError("passed scope requires every required plan check to pass")
+    return result
+
+
+def _framework_scope_result(
     *,
     validator_trace_id: str,
     scope: ValidationScope,
-    plan_hash: str,
+    plan: ValidationPlan,
+    outcome: Literal["failed", "unknown", "error"],
     reason: str,
+    primary_check_id: str | None = None,
 ) -> ScopeValidationResult:
+    """用 Plan 内的全部 check 表达框架失败，不制造计划外 ID。"""
     detail = reason.strip() or "Validator failed without an error description"
-    return ScopeValidationResult(
+    planned = plan.checks_for_scope(scope)
+    planned_ids = {item.check_id for item in planned}
+    if not planned:
+        raise ValueError(f"validation plan has no checks for scope {scope!r}")
+    if primary_check_id is not None and primary_check_id not in planned_ids:
+        raise ValueError(
+            f"primary check {primary_check_id!r} is not planned for scope {scope!r}"
+        )
+    primary = primary_check_id or planned[0].check_id
+    checks: list[ValidationCheck] = []
+    for spec in planned:
+        is_primary = spec.check_id == primary
+        status: CheckStatus = (
+            "failed" if outcome == "failed" and is_primary else "unknown"
+        )
+        if is_primary:
+            issue = detail
+        elif outcome == "error":
+            issue = f"Not evaluated because Validator execution failed: {detail}"
+        else:
+            issue = f"Not evaluated because {primary} blocked this scope: {detail}"
+        checks.append(ValidationCheck(
+            check_id=spec.check_id,
+            status=status,
+            issue=issue,
+        ))
+    result = ScopeValidationResult(
         validator_trace_id=validator_trace_id,
         scope=scope,
-        outcome="error",
-        checks=[],
+        outcome=outcome,
+        checks=checks,
         reason=detail,
-        retry_from=None,
-        plan_hash=plan_hash,
+        retry_from=None if outcome == "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,
+    scope: ValidationScope,
+    plan: ValidationPlan,
+    reason: str,
+) -> ScopeValidationResult:
+    return _framework_scope_result(
+        validator_trace_id=validator_trace_id,
+        scope=scope,
+        plan=plan,
+        outcome="error",
+        reason=reason,
     )
 
 
@@ -721,6 +842,12 @@ def aggregate_validation_results(
     """按固定优先级聚合Scope结果，并选择最早可靠回退层。"""
     if [item.scope for item in scope_results] != plan.effective_scopes:
         raise ValueError("scope results must match plan order exactly")
+    for scope, result in zip(plan.effective_scopes, scope_results):
+        _require_scope_result_integrity(
+            result,
+            plan=plan,
+            expected_scope=scope,
+        )
     outcome = max(
         (item.outcome for item in scope_results),
         key=lambda item: _OUTCOME_PRIORITY[item],
@@ -729,11 +856,7 @@ def aggregate_validation_results(
         check.issue or result.reason
         for result in scope_results
         if result.outcome != "passed"
-        for check in (result.checks or [ValidationCheck(
-            check_id="framework.error",
-            status="failed" if result.outcome == "failed" else "unknown",
-            issue=result.reason,
-        )])
+        for check in result.checks
         if check.status != "passed"
     ]
     issues = list(dict.fromkeys(item for item in issues if item))
@@ -760,14 +883,13 @@ def validation_error(
     evaluated_trace_id: str,
     scope: ValidationScope,
     reason: str,
-    plan_hash: str | None = None,
+    plan: ValidationPlan,
 ) -> ValidationResult:
     """兼容入口：把一个Scope错误包装为一份聚合失败关闭结果。"""
-    resolved_hash = plan_hash or ("0" * 64)
     scope_result = scope_validation_error(
         validator_trace_id=validator_trace_id,
         scope=scope,
-        plan_hash=resolved_hash,
+        plan=plan,
         reason=reason,
     )
     return ValidationResult(
@@ -776,7 +898,7 @@ def validation_error(
         scope_results=[scope_result],
         issues=[scope_result.reason],
         retry_from=None,
-        plan_hash=resolved_hash,
+        plan_hash=plan.plan_hash,
     )
 
 
@@ -826,11 +948,15 @@ class LLMValidator:
         """顺序执行所有Scope；每完成一项即可回调持久化断点。"""
         started = time.monotonic()
         runs: list[_ScopeRun] = []
-        resumed = {
-            item.scope: item
-            for item in resume_scope_results
-            if item.plan_hash == plan.plan_hash
-        }
+        resumed: dict[ValidationScope, ScopeValidationResult] = {}
+        for item in resume_scope_results:
+            if item.plan_hash != plan.plan_hash:
+                continue
+            resumed[item.scope] = _require_scope_result_integrity(
+                item,
+                plan=plan,
+                expected_scope=item.scope,
+            )
         for scope in plan.effective_scopes:
             if scope in resumed:
                 result = resumed[scope]
@@ -1072,7 +1198,7 @@ class LLMValidator:
             result = scope_validation_error(
                 validator_trace_id=validator_trace_id,
                 scope=scope,
-                plan_hash=plan.plan_hash,
+                plan=plan,
                 reason=f"Validator failed: {exc}",
             )
             await self._finish_trace(
@@ -1114,7 +1240,7 @@ class LLMValidator:
             result = scope_validation_error(
                 validator_trace_id=validator_trace_id,
                 scope=scope,
-                plan_hash=plan.plan_hash,
+                plan=plan,
                 reason=reason,
             )
         else:
@@ -1125,22 +1251,15 @@ class LLMValidator:
                 ),
                 None,
             )
-            check = ValidationCheck(
-                check_id=(
-                    deterministic_check.check_id
-                    if deterministic_check else f"{scope}.deterministic"
-                ),
-                status=outcome,
-                issue=reason,
-            )
-            result = ScopeValidationResult(
+            result = _framework_scope_result(
                 validator_trace_id=validator_trace_id,
                 scope=scope,
+                plan=plan,
                 outcome=outcome,
-                checks=[check],
                 reason=reason,
-                retry_from=_SCOPE_RETRY[scope],
-                plan_hash=plan.plan_hash,
+                primary_check_id=(
+                    deterministic_check.check_id if deterministic_check else None
+                ),
             )
         await self._store_terminal_message(trace, result)
         return _ScopeRun(result=result, trace_id=validator_trace_id)
@@ -1152,18 +1271,13 @@ class LLMValidator:
         scope: ValidationScope,
         reason: str,
     ) -> _ScopeRun:
-        result = ScopeValidationResult(
+        detail = f"Required validation material is unavailable: {reason}"
+        result = _framework_scope_result(
             validator_trace_id=trace.trace_id,
             scope=scope,
+            plan=plan,
             outcome="unknown",
-            checks=[ValidationCheck(
-                check_id=f"{scope}.input_limit",
-                status="unknown",
-                issue=f"Required validation material is unavailable: {reason}",
-            )],
-            reason=f"Required validation material is unavailable: {reason}",
-            retry_from=_SCOPE_RETRY[scope],
-            plan_hash=plan.plan_hash,
+            reason=detail,
         )
         await self._store_terminal_message(trace, result)
         return _ScopeRun(result=result, trace_id=trace.trace_id)

tests/test_recursive_replan_context.py

@@ -83,11 +83,11 @@ def validation_result(
         validator_trace_id=f"{child_trace_id}@validator",
         scope="task",
         outcome=outcome,
-        checks=([] if outcome == "error" else [ValidationCheck(
+        checks=[ValidationCheck(
             check_id="task.fixture",
-            status=outcome,
+            status="unknown" if outcome == "error" else outcome,
             issue=None if outcome == "passed" else reason,
-        )]),
+        )],
         reason=reason,
         retry_from=retry_from,
         plan_hash=plan_hash,

tests/test_recursive_validation_core.py

@@ -79,6 +79,9 @@ def passed_scope_response(plan, scope, *, evidence_refs=None):
 
 
 class ValidationPolicyTest(unittest.TestCase):
+    def test_policy_version_invalidates_old_checkpoints(self):
+        self.assertEqual("recursive-validator-2.2", ValidationPolicy().policy_version)
+
     def test_scope_order_and_mandatory_task_are_stable(self):
         plan = plan_for(scopes=["output", "evidence", "output"])
         self.assertEqual(["evidence", "output", "task"], plan.effective_scopes)
@@ -157,10 +160,10 @@ class ValidationPolicyTest(unittest.TestCase):
                 scope=scope,
                 outcome=outcome,
                 checks=[ValidationCheck(
-                    check_id=f"{scope}.policy.1",
+                    check_id=check.check_id,
                     status=outcome,
                     issue=None if outcome == "passed" else f"{scope} gap",
-                )],
+                ) for check in plan.checks_for_scope(scope)],
                 reason=f"{scope} result",
                 retry_from=(
                     None if outcome == "passed"
@@ -182,7 +185,7 @@ class ValidationPolicyTest(unittest.TestCase):
         errored[-1] = scope_validation_error(
             validator_trace_id="validator-task",
             scope="task",
-            plan_hash=plan.plan_hash,
+            plan=plan,
             reason="invalid JSON",
         )
         aggregate = aggregate_validation_results(
@@ -232,13 +235,48 @@ class ValidationPolicyTest(unittest.TestCase):
             ))
         self.assertTrue(any(
             item.check_id == "task.criterion.1"
-            and item.method == "deterministic_and_llm"
+            and item.method == "llm"
             for item in plan.checks
         ))
         self.assertTrue(any(
             item.check_id == "task.expected_output.1"
+            and item.method == "llm"
             for item in plan.checks
         ))
+        self.assertNotIn(
+            "deterministic_and_llm",
+            {item.method for item in plan.checks},
+        )
+        root_plan = plan_for(root=True)
+        self.assertTrue(all(
+            item.method == "llm"
+            for item in root_plan.checks
+            if item.check_id.startswith(("root.criterion.", "root.constraint."))
+        ))
+
+    def test_only_material_resolution_failures_compile_as_deterministic(self):
+        issue = MaterialIssue(
+            artifact_id="missing-output",
+            outcome="failed",
+            reason="artifact does not exist",
+            scopes=["output"],
+        )
+        plan = ValidationPolicy().compile_plan(
+            task_brief=dict(BRIEF, validation_scopes=["output"]),
+            task_brief_version=1,
+            root_task_anchor=ANCHOR,
+            task_report={"summary": "done"},
+            candidate_output=None,
+            evaluated_head_sequence=1,
+            materials=[],
+            material_issues=[issue],
+            model_by_scope={"output": "fake", "task": "fake"},
+            root=False,
+        )
+        deterministic = [
+            item for item in plan.checks if item.method == "deterministic"
+        ]
+        self.assertEqual(["output.material.1"], [item.check_id for item in deterministic])
 
     def test_plan_hash_changes_for_every_authoritative_input(self):
         base = plan_for(scopes=["output"])
@@ -315,6 +353,56 @@ class ValidationPolicyTest(unittest.TestCase):
                     validator_trace_id="validator",
                 )
 
+    def test_scope_result_rejects_passed_with_failed_check(self):
+        plan = plan_for()
+        checks = [
+            ValidationCheck(
+                check_id=item.check_id,
+                status="failed" if index == 0 else "passed",
+                issue="forged failure" if index == 0 else None,
+            )
+            for index, item in enumerate(plan.checks_for_scope("task"))
+        ]
+        with self.assertRaisesRegex(ValidationError, "every check to pass"):
+            ScopeValidationResult(
+                validator_trace_id="validator-task",
+                scope="task",
+                outcome="passed",
+                checks=checks,
+                reason="forged pass",
+                retry_from=None,
+                plan_hash=plan.plan_hash,
+            )
+
+    def test_aggregate_revalidates_mutated_scope_outcome(self):
+        plan = plan_for()
+        checks = [
+            ValidationCheck(
+                check_id=item.check_id,
+                status="failed" if index == 0 else "passed",
+                issue="real failure" if index == 0 else None,
+            )
+            for index, item in enumerate(plan.checks_for_scope("task"))
+        ]
+        forged = ScopeValidationResult(
+            validator_trace_id="validator-task",
+            scope="task",
+            outcome="failed",
+            checks=checks,
+            reason="real failure",
+            retry_from="task_definition",
+            plan_hash=plan.plan_hash,
+        )
+        # 模拟持久化边界之外绕过 Pydantic 的内存篡改；聚合仍必须失败关闭。
+        object.__setattr__(forged, "outcome", "passed")
+        object.__setattr__(forged, "retry_from", None)
+        with self.assertRaisesRegex(ValueError, "every check to pass"):
+            aggregate_validation_results(
+                evaluated_trace_id="child",
+                plan=plan,
+                scope_results=[forged],
+            )
+
     def test_unknown_aggregate_preserves_recoverable_retry(self):
         plan = plan_for()
         scope = ScopeValidationResult(
@@ -322,10 +410,10 @@ class ValidationPolicyTest(unittest.TestCase):
             scope="task",
             outcome="unknown",
             checks=[ValidationCheck(
-                check_id="task.policy.1",
+                check_id=check.check_id,
                 status="unknown",
                 issue="required material is unavailable",
-            )],
+            ) for check in plan.checks_for_scope("task")],
             reason="insufficient material",
             retry_from="task_definition",
             plan_hash=plan.plan_hash,
@@ -533,6 +621,46 @@ class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
         )
         self.assertEqual("error", run.result.outcome)
         self.assertEqual(1, len(llm.calls))
+        scope_result = run.result.scope_results[0]
+        self.assertEqual(
+            {item.check_id for item in plan.checks_for_scope("task")},
+            {item.check_id for item in scope_result.checks},
+        )
+        self.assertTrue(all(item.status == "unknown" for item in scope_result.checks))
+
+    async def test_input_packet_failure_uses_only_all_planned_check_ids(self):
+        plan = plan_for()
+        llm = FakeLLM([])
+        validator = LLMValidator(
+            llm_call=llm,
+            trace_store=self.store,
+            policy=ValidationPolicy(),
+            max_input_chars=8,
+        )
+        run = await validator.validate_plan(
+            evaluated_trace=self.evaluated,
+            trajectory=self.trajectory,
+            plan=plan,
+            root_task_anchor=ANCHOR,
+            task_brief=BRIEF,
+            task_report={"summary": "done"},
+            candidate_output=None,
+            materials=[],
+            material_issues=[],
+            model_by_scope={"task": "fake"},
+        )
+
+        self.assertEqual([], llm.calls)
+        self.assertEqual("unknown", run.result.outcome)
+        scope_result = run.result.scope_results[0]
+        self.assertEqual(
+            [item.check_id for item in plan.checks_for_scope("task")],
+            [item.check_id for item in scope_result.checks],
+        )
+        self.assertTrue(all(item.status == "unknown" for item in scope_result.checks))
+        self.assertFalse(any(
+            item.check_id.endswith("input_limit") for item in scope_result.checks
+        ))
 
     async def test_deterministic_material_failure_skips_all_llm_calls(self):
         plan = ValidationPolicy().compile_plan(
@@ -578,6 +706,12 @@ class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual([], llm.calls)
         for trace_id in run.trace_ids:
             self.assertEqual("failed", (await self.store.get_trace(trace_id)).status)
+        for result in run.result.scope_results:
+            self.assertEqual(
+                {item.check_id for item in plan.checks_for_scope(result.scope)},
+                {item.check_id for item in result.checks},
+            )
+            self.assertEqual(1, sum(item.status == "failed" for item in result.checks))
 
     async def test_material_failure_only_skips_affected_scopes(self):
         brief = dict(BRIEF, validation_scopes=["hypothesis", "output"])
@@ -627,6 +761,11 @@ class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
             ["passed", "failed", "failed"],
             [item.outcome for item in run.result.scope_results],
         )
+        for result in run.result.scope_results:
+            self.assertEqual(
+                {item.check_id for item in plan.checks_for_scope(result.scope)},
+                {item.check_id for item in result.checks},
+            )
 
     async def test_matching_scope_checkpoint_is_not_run_again(self):
         plan = plan_for(scopes=["output"])
@@ -668,6 +807,50 @@ class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
             run.trace_ids,
         )
 
+    async def test_checkpoint_revalidates_mutated_scope_outcome(self):
+        plan = plan_for()
+        checks = [
+            ValidationCheck(
+                check_id=item.check_id,
+                status="failed" if index == 0 else "passed",
+                issue="real failure" if index == 0 else None,
+            )
+            for index, item in enumerate(plan.checks_for_scope("task"))
+        ]
+        forged = ScopeValidationResult(
+            validator_trace_id="existing-validator-task",
+            scope="task",
+            outcome="failed",
+            checks=checks,
+            reason="real failure",
+            retry_from="task_definition",
+            plan_hash=plan.plan_hash,
+        )
+        object.__setattr__(forged, "outcome", "passed")
+        object.__setattr__(forged, "retry_from", None)
+        llm = FakeLLM([])
+        validator = LLMValidator(
+            llm_call=llm,
+            trace_store=self.store,
+            policy=ValidationPolicy(),
+        )
+
+        with self.assertRaisesRegex(ValueError, "every check to pass"):
+            await validator.validate_plan(
+                evaluated_trace=self.evaluated,
+                trajectory=[],
+                plan=plan,
+                root_task_anchor=ANCHOR,
+                task_brief=BRIEF,
+                task_report={"summary": "done"},
+                candidate_output=None,
+                materials=[],
+                material_issues=[],
+                model_by_scope={"task": "fake"},
+                resume_scope_results=[forged],
+            )
+        self.assertEqual([], llm.calls)
+
     async def test_task_scope_forged_private_tool_is_an_error(self):
         plan = plan_for()
         llm = FakeLLM([response(tool_calls=[{

tests/test_recursive_validator_lifecycle_complex.py

@@ -15,6 +15,10 @@ from cyber_agent.core.resource_budget import (
     ResourceBudget,
     ResourceBudgetController,
 )
+from cyber_agent.core.run_snapshot import (
+    RunConfigSnapshotV1,
+    persist_run_config_snapshot,
+)
 from cyber_agent.core.runner import AgentRunner, RunConfig
 from cyber_agent.core.task_protocol import (
     TaskReport,
@@ -532,20 +536,29 @@ class FiveLevelValidatorLifecycleTest(unittest.IsolatedAsyncioTestCase):
             )
             self.assertEqual("passed", run.result.outcome)
 
-        result = await runner.run_result(
-            [],
-            RunConfig(
-                trace_id=self.root_id,
-                tools=["publish_script"],
-                tool_groups=[],
-                enable_research_flow=False,
-                knowledge=KnowledgeConfig(
-                    enable_extraction=False,
-                    enable_completion_extraction=False,
-                    enable_injection=False,
-                ),
+        resume_config = RunConfig(
+            trace_id=self.root_id,
+            model="fake",
+            uid="user-1",
+            tools=["publish_script"],
+            tool_groups=[],
+            enable_research_flow=False,
+            knowledge=KnowledgeConfig(
+                enable_extraction=False,
+                enable_completion_extraction=False,
+                enable_injection=False,
+            ),
+        )
+        root = await self.store.get_trace(self.root_id)
+        persist_run_config_snapshot(
+            root.context,
+            RunConfigSnapshotV1.from_run_config(
+                resume_config,
+                memory_identity=None,
             ),
         )
+        await self.store.update_trace(self.root_id, context=root.context)
+        result = await runner.run_result([], resume_config)
         self.assertEqual("completed", result["status"])
         root = await self.store.get_trace(self.root_id)
         state = ensure_task_protocol(root.context)