Class _RecordingHooks

tests/flows/test_runner.py:19–45

Test hook that records every lifecycle call on a shared list.

class _RecordingHooks(RunHooks[Any]):
    """Test hook that records every lifecycle call on a shared list."""

    def __init__(self, label: str, log: list[tuple[str, str]]) -> None:
        self.label = label
        self.log = log

    async def on_llm_start(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_llm_start"))

    async def on_llm_end(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_llm_end"))

    async def on_agent_start(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_agent_start"))

    async def on_agent_end(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_agent_end"))

    async def on_handoff(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_handoff"))

    async def on_tool_start(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_tool_start"))

    async def on_tool_end(self, *_: Any, **__: Any) -> None:
        self.log.append((self.label, "on_tool_end"))


class ComposeHooksTests(unittest.TestCase):