Agent/tests/test_recursive_validation_core.py

import json
import tempfile
import unittest

from pydantic import ValidationError

from cyber_agent.core.validation import (
    LLMValidator,
    ScopeValidationResult,
    ValidationCheck,
    ValidationPolicy,
    ValidatorSettings,
    aggregate_validation_results,
    build_validation_packet,
    parse_scope_validation_result,
    scope_validation_error,
    persist_validation_policy,
    require_validation_policy,
)
from cyber_agent.core.artifacts import MaterialIssue
from cyber_agent.core.validator_web import (
    ValidatorToolLimits,
    ValidatorToolSession,
)
from cyber_agent.trace.models import Message, Trace
from cyber_agent.trace.store import FileSystemTraceStore


BRIEF = {
    "objective": "verify one claim",
    "reason": "parent needs a checked fact",
    "completion_criteria": ["the claim is checked"],
    "expected_outputs": ["one conclusion"],
    "constraints": [],
    "validation_scopes": [],
}
ANCHOR = {
    "objective": "publish a reliable answer",
    "completion_criteria": ["all facts are supported"],
    "constraints": ["do not invent sources"],
}


def plan_for(*, scopes=None, root=False, head=2):
    brief = dict(BRIEF, validation_scopes=scopes or [])
    policy = ValidationPolicy()
    return policy.compile_plan(
        task_brief=brief,
        task_brief_version=1,
        root_task_anchor=ANCHOR,
        task_report={"summary": "done"},
        candidate_output="candidate" if root else None,
        evaluated_head_sequence=head,
        materials=[],
        material_issues=[],
        model_by_scope={scope: "fake" for scope in (
            "evidence", "hypothesis", "output", "task", "root"
        )},
        root=root,
    )


def passed_scope_response(plan, scope, *, evidence_refs=None):
    return json.dumps({
        "scope": scope,
        "outcome": "passed",
        "checks": [
            {
                "check_id": check.check_id,
                "status": "passed",
                "evidence_refs": list(evidence_refs or []),
                "issue": None,
            }
            for check in plan.checks_for_scope(scope)
        ],
        "reason": "every planned check passed",
        "retry_from": 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)
        self.assertTrue(any(item.check_id == "task.criterion.1" for item in plan.checks))
        same = plan_for(scopes=["evidence", "output"])
        self.assertEqual(plan.plan_hash, same.plan_hash)
        changed = plan_for(scopes=["evidence", "output"], head=3)
        self.assertNotEqual(plan.plan_hash, changed.plan_hash)

    def test_root_is_framework_owned(self):
        self.assertEqual(["root"], plan_for(root=True).effective_scopes)
        with self.assertRaises(ValueError):
            ValidationPolicy().compile_plan(
                task_brief=dict(BRIEF, validation_scopes=["root"]),
                task_brief_version=1,
                root_task_anchor=ANCHOR,
                task_report={},
                candidate_output=None,
                evaluated_head_sequence=1,
                materials=[],
                material_issues=[],
                model_by_scope={},
                root=False,
            )

    def test_parser_binds_plan_checks_and_framework_ids(self):
        plan = plan_for()
        result = parse_scope_validation_result(
            passed_scope_response(plan, "task"),
            plan=plan,
            expected_scope="task",
            validator_trace_id="validator-1",
        )
        self.assertEqual("validator-1", result.validator_trace_id)
        forged = json.loads(passed_scope_response(plan, "task"))
        forged["checks"].append({
            "check_id": "task.forged",
            "status": "passed",
            "evidence_refs": [],
            "issue": None,
        })
        with self.assertRaisesRegex(ValueError, "do not match plan"):
            parse_scope_validation_result(
                json.dumps(forged),
                plan=plan,
                expected_scope="task",
                validator_trace_id="validator-1",
            )

    def test_evidence_pass_requires_opened_source_ids(self):
        plan = plan_for(scopes=["evidence"])
        content = passed_scope_response(plan, "evidence", evidence_refs=["src-1"])
        with self.assertRaisesRegex(ValueError, "opened"):
            parse_scope_validation_result(
                content,
                plan=plan,
                expected_scope="evidence",
                validator_trace_id="validator",
                opened_source_ids=set(),
            )
        result = parse_scope_validation_result(
            content,
            plan=plan,
            expected_scope="evidence",
            validator_trace_id="validator",
            opened_source_ids={"src-1"},
        )
        self.assertEqual("passed", result.outcome)

    def test_aggregate_priority_and_retry_are_framework_owned(self):
        plan = plan_for(scopes=["evidence", "output"])
        results = []
        for scope, outcome in zip(plan.effective_scopes, ["failed", "unknown", "passed"]):
            results.append(ScopeValidationResult(
                validator_trace_id=f"validator-{scope}",
                scope=scope,
                outcome=outcome,
                checks=[ValidationCheck(
                    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"
                    else "evidence" if scope == "evidence"
                    else "output" if scope == "output"
                    else "task_definition"
                ),
                plan_hash=plan.plan_hash,
            ))
        aggregate = aggregate_validation_results(
            evaluated_trace_id="child",
            plan=plan,
            scope_results=results,
        )
        self.assertEqual("failed", aggregate.outcome)
        self.assertEqual("evidence", aggregate.retry_from)

        errored = list(results)
        errored[-1] = scope_validation_error(
            validator_trace_id="validator-task",
            scope="task",
            plan=plan,
            reason="invalid JSON",
        )
        aggregate = aggregate_validation_results(
            evaluated_trace_id="child",
            plan=plan,
            scope_results=errored,
        )
        self.assertEqual("error", aggregate.outcome)
        self.assertIsNone(aggregate.retry_from)

    def test_packet_keeps_contract_and_newest_main_path(self):
        trajectory = [
            {
                "sequence": index,
                "role": "tool",
                "name": "read_file",
                "content": f"marker-{index}-" + ("x" * 300),
            }
            for index in range(1, 8)
        ]
        trajectory.append({
            "sequence": 99,
            "role": "assistant",
            "content": "side-secret",
            "branch_type": "compression",
        })
        packet = build_validation_packet(
            validation_scope="task",
            validation_plan=plan_for(),
            task_brief=BRIEF,
            task_report={"summary": "report-kept"},
            trajectory=trajectory,
            max_chars=2_500,
        )
        self.assertLessEqual(len(packet), 2_500)
        self.assertIn("report-kept", packet)
        self.assertIn("marker-7", packet)
        self.assertNotIn("marker-1", packet)
        self.assertNotIn("side-secret", packet)

    def test_each_scope_has_fixed_rubric_and_task_has_dynamic_checks(self):
        plan = plan_for(scopes=["evidence", "hypothesis", "output"])
        for scope in ("evidence", "hypothesis", "output", "task"):
            self.assertTrue(any(
                item.scope == scope and item.check_id.startswith(f"{scope}.policy.")
                for item in plan.checks
            ))
        self.assertTrue(any(
            item.check_id == "task.criterion.1"
            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"])
        policy = ValidationPolicy()
        common = {
            "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": 2,
            "materials": [],
            "material_issues": [],
            "model_by_scope": {scope: "fake" for scope in (
                "evidence", "hypothesis", "output", "task", "root"
            )},
            "root": False,
        }
        variants = [
            {"task_brief_version": 2},
            {"task_report": {"summary": "changed"}},
            {"evaluated_head_sequence": 3},
            {"model_by_scope": {**common["model_by_scope"], "task": "stronger"}},
            {"material_issues": [MaterialIssue(
                artifact_id="script",
                outcome="unknown",
                reason="temporarily unavailable",
            )]},
        ]
        for override in variants:
            with self.subTest(override=override):
                changed = policy.compile_plan(**{**common, **override})
                self.assertNotEqual(base.plan_hash, changed.plan_hash)

    def test_policy_snapshot_detects_tampering(self):
        context = {}
        persist_validation_policy(
            context,
            ValidationPolicy(),
            ValidatorSettings(search_provider="disabled"),
        )
        restored, settings = require_validation_policy(context)
        self.assertEqual("disabled", settings.search_provider)
        self.assertEqual(ValidationPolicy().policy_hash, restored.policy_hash)
        context["validation_policy"]["global_rules"] = "approve everything"
        with self.assertRaisesRegex(ValueError, "hash"):
            require_validation_policy(context)

    def test_parser_rejects_missing_duplicate_wrong_scope_and_outcome(self):
        plan = plan_for()
        valid = json.loads(passed_scope_response(plan, "task"))
        variants = []
        missing = json.loads(json.dumps(valid))
        missing["checks"].pop()
        variants.append(missing)
        duplicate = json.loads(json.dumps(valid))
        duplicate["checks"].append(duplicate["checks"][0])
        variants.append(duplicate)
        wrong_scope = json.loads(json.dumps(valid))
        wrong_scope["scope"] = "output"
        variants.append(wrong_scope)
        wrong_outcome = json.loads(json.dumps(valid))
        wrong_outcome["outcome"] = "failed"
        wrong_outcome["retry_from"] = "task_definition"
        variants.append(wrong_outcome)
        for payload in variants:
            with self.subTest(payload=payload), self.assertRaises(
                (ValueError, ValidationError)
            ):
                parse_scope_validation_result(
                    json.dumps(payload),
                    plan=plan,
                    expected_scope="task",
                    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(
            validator_trace_id="validator-task",
            scope="task",
            outcome="unknown",
            checks=[ValidationCheck(
                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,
        )
        aggregate = aggregate_validation_results(
            evaluated_trace_id="child",
            plan=plan,
            scope_results=[scope],
        )
        self.assertEqual("unknown", aggregate.outcome)
        self.assertEqual("task_definition", aggregate.retry_from)


class FakeLLM:
    def __init__(self, responses):
        self.responses = list(responses)
        self.calls = []

    async def __call__(self, **kwargs):
        self.calls.append(kwargs)
        response = self.responses.pop(0)
        if isinstance(response, Exception):
            raise response
        return response


def response(content="", tool_calls=None):
    return {
        "content": content,
        "tool_calls": tool_calls,
        "prompt_tokens": 10,
        "completion_tokens": 5,
        "cost": 0.01,
        "finish_reason": "tool_calls" if tool_calls else "stop",
    }


class LLMValidatorTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.store = FileSystemTraceStore(self.temp.name)
        self.evaluated = Trace(
            trace_id="root@delegate-child",
            mode="agent",
            task="child task",
            uid="user-1",
            model="fake",
            current_goal_id="1",
            context={
                "agent_mode": "recursive",
                "agent_mode_revision": 2,
                "root_trace_id": "root",
                "agent_depth": 1,
            },
        )
        await self.store.create_trace(self.evaluated)
        self.trajectory = [Message.create(
            trace_id=self.evaluated.trace_id,
            role="tool",
            sequence=1,
            content={"tool_name": "read_file", "result": "actual output"},
        )]

    async def asyncTearDown(self):
        self.temp.cleanup()

    async def test_task_scope_is_one_tool_free_call(self):
        plan = plan_for()
        llm = FakeLLM([response(passed_scope_response(plan, "task"))])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        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("passed", run.result.outcome, run.result.model_dump())
        self.assertEqual(1, len(llm.calls))
        self.assertEqual([], llm.calls[0]["tools"])
        trace = await self.store.get_trace(run.trace_id)
        self.assertEqual("completed", trace.status)
        self.assertEqual("task", trace.context["validation_scope"])

    async def test_evidence_scope_runs_private_search_and_open_loop(self):
        plan = plan_for(scopes=["evidence"])

        class Provider:
            async def search(self, query, max_results):
                return [{
                    "title": "Official",
                    "link": "https://example.com/fact",
                    "snippet": "official page",
                }]

        async def page_fetcher(url, resolver=None):
            del resolver
            return {
                "source_id": "src-official",
                "url": url,
                "final_url": url,
                "title": "Official",
                "content_type": "text/plain",
                "retrieved_at": "2026-01-01T00:00:00+00:00",
                "content_sha256": "b" * 64,
                "text": "The official figure is 12%.",
                "truncated": False,
                "untrusted_material": True,
            }

        def session_factory(scope, allowed_urls, trace_id):
            del scope, trace_id
            return ValidatorToolSession(
                provider=Provider(),
                allowed_urls=allowed_urls,
                limits=ValidatorToolLimits(5, 10, 15),
                page_fetcher=page_fetcher,
            )

        llm = FakeLLM([
            response(tool_calls=[{
                "id": "search-1",
                "type": "function",
                "function": {
                    "name": "validator_web_search",
                    "arguments": json.dumps({"query": "official figure"}),
                },
            }]),
            response(tool_calls=[{
                "id": "open-1",
                "type": "function",
                "function": {
                    "name": "validator_open_url",
                    "arguments": json.dumps({"url": "https://example.com/fact"}),
                },
            }]),
            response(passed_scope_response(
                plan,
                "evidence",
                evidence_refs=["src-official"],
            )),
            response(passed_scope_response(plan, "task")),
        ])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
            tool_session_factory=session_factory,
        )
        run = await validator.validate_plan(
            evaluated_trace=self.evaluated,
            trajectory=self.trajectory,
            plan=plan,
            root_task_anchor=ANCHOR,
            task_brief=dict(BRIEF, validation_scopes=["evidence"]),
            task_report={"summary": "done"},
            candidate_output=None,
            materials=[],
            material_issues=[],
            model_by_scope={"evidence": "fake", "task": "fake"},
        )
        self.assertEqual("passed", run.result.outcome, run.result.model_dump())
        self.assertEqual(2, len(run.trace_ids))
        evidence_messages = await self.store.get_trace_messages(run.trace_ids[0])
        self.assertEqual(
            ["system", "user", "assistant", "tool", "assistant", "tool", "assistant"],
            [item.role for item in evidence_messages],
        )
        self.assertEqual(
            {"validator_web_search", "validator_open_url"},
            {
                item["function"]["name"]
                for item in (await self.store.get_trace(run.trace_ids[0])).tools
            },
        )

    async def test_invalid_output_fails_without_format_correction(self):
        plan = plan_for()
        llm = FakeLLM([response("not-json")])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        run = 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"},
        )
        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(
            task_brief=dict(BRIEF, validation_scopes=["output"]),
            task_brief_version=1,
            root_task_anchor=ANCHOR,
            task_report={"summary": "claimed output"},
            candidate_output=None,
            evaluated_head_sequence=1,
            materials=[],
            material_issues=[MaterialIssue(
                artifact_id="script:missing",
                outcome="failed",
                reason="artifact does not exist",
            )],
            model_by_scope={"output": "fake", "task": "fake"},
            root=False,
        )
        llm = FakeLLM([])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        run = await validator.validate_plan(
            evaluated_trace=self.evaluated,
            trajectory=[],
            plan=plan,
            root_task_anchor=ANCHOR,
            task_brief=dict(BRIEF, validation_scopes=["output"]),
            task_report={"summary": "claimed output"},
            candidate_output=None,
            materials=[],
            material_issues=[MaterialIssue(
                artifact_id="script:missing",
                outcome="failed",
                reason="artifact does not exist",
            )],
            model_by_scope={"output": "fake", "task": "fake"},
        )
        self.assertEqual("failed", run.result.outcome)
        self.assertEqual(2, len(run.trace_ids))
        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"])
        issue = MaterialIssue(
            artifact_id="script:missing",
            outcome="failed",
            reason="script does not exist",
            scopes=["output", "task", "root"],
        )
        plan = ValidationPolicy().compile_plan(
            task_brief=brief,
            task_brief_version=1,
            root_task_anchor=ANCHOR,
            task_report={"summary": "claimed output"},
            candidate_output=None,
            evaluated_head_sequence=1,
            materials=[],
            material_issues=[issue],
            model_by_scope={
                "hypothesis": "fake", "output": "fake", "task": "fake",
            },
            root=False,
        )
        llm = FakeLLM([response(passed_scope_response(plan, "hypothesis"))])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        run = await validator.validate_plan(
            evaluated_trace=self.evaluated,
            trajectory=[],
            plan=plan,
            root_task_anchor=ANCHOR,
            task_brief=brief,
            task_report={"summary": "claimed output"},
            candidate_output=None,
            materials=[],
            material_issues=[issue],
            model_by_scope={
                "hypothesis": "fake", "output": "fake", "task": "fake",
            },
        )

        self.assertEqual(1, len(llm.calls))
        self.assertEqual(
            ["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"])
        output_result = ScopeValidationResult(
            validator_trace_id="existing-validator-output",
            scope="output",
            outcome="passed",
            checks=[ValidationCheck(
                check_id=item.check_id,
                status="passed",
            ) for item in plan.checks_for_scope("output")],
            reason="already checked",
            retry_from=None,
            plan_hash=plan.plan_hash,
        )
        llm = FakeLLM([response(passed_scope_response(plan, "task"))])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        run = await validator.validate_plan(
            evaluated_trace=self.evaluated,
            trajectory=[],
            plan=plan,
            root_task_anchor=ANCHOR,
            task_brief=dict(BRIEF, validation_scopes=["output"]),
            task_report={"summary": "done"},
            candidate_output=None,
            materials=[],
            material_issues=[],
            model_by_scope={"output": "fake", "task": "fake"},
            resume_scope_results=[output_result],
        )
        self.assertEqual("passed", run.result.outcome)
        self.assertEqual(1, len(llm.calls))
        self.assertEqual(
            ["existing-validator-output", run.trace_ids[1]],
            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=[{
            "id": "forged",
            "type": "function",
            "function": {
                "name": "validator_web_search",
                "arguments": json.dumps({"query": "should not run"}),
            },
        }])])
        validator = LLMValidator(
            llm_call=llm,
            trace_store=self.store,
            policy=ValidationPolicy(),
        )
        run = 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"},
        )
        self.assertEqual("error", run.result.outcome)
        self.assertIn("unavailable tool", run.result.issues[0])


if __name__ == "__main__":
    unittest.main()