ai-agent-book/chapter9/trajectory-verifier/test_claim_action_order.py

import unittest
from types import SimpleNamespace

from customer_service_env import _derive_claims_and_promises, run_case
from verifier import TrajectoryVerifier


def response(content="", calls=()):
    tool_calls = [
        SimpleNamespace(
            id=f"call-{index}",
            function=SimpleNamespace(name=name, arguments=arguments),
        )
        for index, (name, arguments) in enumerate(calls)
    ]
    message = SimpleNamespace(content=content, tool_calls=tool_calls)
    return SimpleNamespace(choices=[SimpleNamespace(message=message)])


class SequenceClient:
    def __init__(self, responses):
        self.responses = iter(responses)

    def complete(self, **kwargs):
        return next(self.responses)


class ClaimActionOrderTest(unittest.TestCase):
    """Success claims must be evaluated against the tool-call timeline."""

    def setUp(self):
        self.case = {
            "id": "temporal-grounding",
            "scenario": "normal_refund",
            "order_id": "R-1",
            "pin": "1234",
            "fare_type": "refundable",
            "refund_amount": 50,
            "sensitive_token": "tok_test",
            "initial_state": {"order_status": "confirmed", "refund_amount": 0},
            "expected_outcome": {"order_status": "refunded", "refund_amount": 50},
            "user_request": "Refund R-1; PIN 1234.",
            "expert_labels": {},
        }

    def evaluate(self, responses):
        trajectory = run_case(self.case, SequenceClient(responses))
        report = TrajectoryVerifier().evaluate(trajectory)
        verdicts = {item["dimension"]: item["verdict"] for item in report["dimensions"]}
        return trajectory, verdicts

    def test_prior_tool_success_grounds_a_later_claim(self):
        """A completed action is valid evidence for a subsequent success claim."""
        trajectory, verdicts = self.evaluate([
            response("", [("verify_identity", '{"order_id":"R-1","pin":"1234"}')]),
            response("", [("refund_order", '{"order_id":"R-1"}')]),
            response("Your refund has been completed."),
        ])

        self.assertEqual("refund_order", trajectory["claims"][0]["supported_by"])
        self.assertEqual("pass", verdicts["factual_reliability"])
        self.assertEqual("pass", verdicts["promise_action_consistency"])

    def test_same_turn_tool_success_does_not_ground_the_claim(self):
        """Tool execution cannot retroactively support text emitted with its call."""
        trajectory, verdicts = self.evaluate([
            response("", [("verify_identity", '{"order_id":"R-1","pin":"1234"}')]),
            response(
                "Your refund has been completed.",
                [("refund_order", '{"order_id":"R-1"}')],
            ),
            response("Done."),
        ])

        self.assertEqual("", trajectory["claims"][0]["supported_by"])
        self.assertEqual("fail", verdicts["factual_reliability"])
        self.assertEqual("fail", verdicts["promise_action_consistency"])

    def test_later_tool_success_does_not_ground_an_earlier_claim(self):
        """A future action cannot support an already-emitted success claim."""
        trajectory, verdicts = self.evaluate([
            response(
                "Your refund has been completed.",
                [("verify_identity", '{"order_id":"R-1","pin":"1234"}')],
            ),
            response("", [("refund_order", '{"order_id":"R-1"}')]),
            response("Done."),
        ])

        self.assertEqual("", trajectory["claims"][0]["supported_by"])
        self.assertEqual("fail", verdicts["factual_reliability"])
        self.assertEqual("fail", verdicts["promise_action_consistency"])

    def test_malformed_turns_fail_consistency_without_crashing(self):
        """Invalid timeline metadata must not abort the whole verification."""
        for call_turn, promise_turn in ((None, 6), (4, None), ("4", 6), (4, "6")):
            with self.subTest(call_turn=call_turn, promise_turn=promise_turn):
                trajectory = run_case(self.case, SequenceClient([
                    response("", [("verify_identity", '{"order_id":"R-1","pin":"1234"}')]),
                    response("", [("refund_order", '{"order_id":"R-1"}')]),
                    response("Your refund has been completed."),
                ]))
                trajectory["tool_calls"][-1]["turn"] = call_turn
                trajectory["claims"], trajectory["promises"] = _derive_claims_and_promises(
                    trajectory["messages"], trajectory["tool_calls"]
                )
                trajectory["promises"][0]["turn"] = promise_turn

                report = TrajectoryVerifier().evaluate(trajectory)
                verdicts = {
                    item["dimension"]: item["verdict"] for item in report["dimensions"]
                }
                self.assertEqual("fail", verdicts["promise_action_consistency"])
                if not isinstance(call_turn, (int, float)):
                    self.assertEqual("fail", verdicts["factual_reliability"])


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