Skip to content

tot_agent.agent

Core agentic loop, Observer pattern implementation, and Goal template builders.

Observer pattern classes

classDiagram
    class AgentObserver {
        <<abstract>>
        +on_event(event: AgentEvent) None
    }
    class ConsoleObserver {
        -_console: Console
        +on_event(event: AgentEvent) None
    }
    class LoggingObserver {
        -_log: Logger
        +on_event(event: AgentEvent) None
    }
    AgentObserver <|-- ConsoleObserver
    AgentObserver <|-- LoggingObserver
    BrowserAgent --> AgentObserver : notifies
    BrowserAgent --> AgentEvent : emits

Goal template hierarchy

classDiagram
    class GoalTemplate {
        <<abstract>>
        +build(**kwargs) str
    }
    class CreateTestsGoal {
        +count: int
        +genre: str
        +build() str
    }
    class VoteGoal {
        +username: str
        +password: str
        +vote_count: int
        +bias: str
        +build() str
    }
    class SimulateAllUsersGoal {
        +vote_count_each: int
        +build() str
    }
    class FullSeedGoal {
        +test_count: int
        +vote_rounds: int
        +build() str
    }
    GoalTemplate <|-- CreateTestsGoal
    GoalTemplate <|-- VoteGoal
    GoalTemplate <|-- SimulateAllUsersGoal
    GoalTemplate <|-- FullSeedGoal

Module reference

tot_agent.agent

agent.py — Core agentic loop and goal template builders.

Architecture — Observer pattern

The agent emits :class:AgentEvent objects at each significant step of its execution. Any number of :class:AgentObserver instances can be attached to receive these events. Two concrete observers are bundled:

  • :class:ConsoleObserver — renders events to the terminal via :mod:rich.
  • :class:LoggingObserver — writes events to the standard :mod:logging hierarchy (useful for CI/CD pipelines and file-based audit trails).

This separation means the core loop never directly calls print or console.print.

Usage::

from tot_agent.browser import BrowserManager
from tot_agent.agent import BrowserAgent, ConsoleObserver, LoggingObserver

async with BrowserManager() as bm:
    agent = BrowserAgent(bm, observers=[ConsoleObserver(), LoggingObserver()])
    summary = await agent.run("Log in as admin and take a screenshot.")

EventType

Bases: Enum

Enumeration of agent lifecycle events.

Source code in src/tot_agent/agent.py
class EventType(Enum):
    """Enumeration of agent lifecycle events."""

    GOAL_START = auto()
    """The agent has received a new goal and is about to start the loop."""
    STEP_START = auto()
    """A new iteration of the tool-call loop is beginning."""
    AGENT_TEXT = auto()
    """The model produced a text block (reasoning or narration)."""
    TOOL_CALL = auto()
    """The model requested a tool execution."""
    TOOL_RESULT = auto()
    """A tool call completed and returned a result."""
    GOAL_COMPLETE = auto()
    """The agent finished successfully (``stop_reason == "end_turn"``)."""
    STEP_LIMIT = auto()
    """The loop was terminated because :data:`~tot_agent.config.MAX_AGENT_STEPS`
    was reached."""

GOAL_START = auto() class-attribute instance-attribute

The agent has received a new goal and is about to start the loop.

STEP_START = auto() class-attribute instance-attribute

A new iteration of the tool-call loop is beginning.

AGENT_TEXT = auto() class-attribute instance-attribute

The model produced a text block (reasoning or narration).

TOOL_CALL = auto() class-attribute instance-attribute

The model requested a tool execution.

TOOL_RESULT = auto() class-attribute instance-attribute

A tool call completed and returned a result.

GOAL_COMPLETE = auto() class-attribute instance-attribute

The agent finished successfully (stop_reason == "end_turn").

STEP_LIMIT = auto() class-attribute instance-attribute

The loop was terminated because :data:~tot_agent.config.MAX_AGENT_STEPS was reached.

AgentEvent dataclass

Carries data from the agent loop to registered observers.

Parameters:

Name Type Description Default
event_type EventType

The kind of event that occurred.

required
data dict[str, Any]

Arbitrary key-value payload; keys vary per event type.

dict()
Source code in src/tot_agent/agent.py
@dataclass
class AgentEvent:
    """Carries data from the agent loop to registered observers.

    :param EventType event_type: The kind of event that occurred.
    :param dict[str, Any] data: Arbitrary key-value payload; keys vary per event type.
    """

    event_type: EventType
    data: dict[str, Any] = field(default_factory=dict)

AgentObserver

Bases: ABC

Abstract base class for agent event observers (Observer pattern).

Subclass this and implement :meth:on_event to react to agent lifecycle events without modifying the agent core.

Source code in src/tot_agent/agent.py
class AgentObserver(ABC):
    """Abstract base class for agent event observers (Observer pattern).

    Subclass this and implement :meth:`on_event` to react to agent lifecycle
    events without modifying the agent core.
    """

    @abstractmethod
    def on_event(self, event: AgentEvent) -> None:
        """Handle an agent event.

        :param AgentEvent event: The event to handle.
        """

on_event(event) abstractmethod

Handle an agent event.

Parameters:

Name Type Description Default
event AgentEvent

The event to handle.

required
Source code in src/tot_agent/agent.py
@abstractmethod
def on_event(self, event: AgentEvent) -> None:
    """Handle an agent event.

    :param AgentEvent event: The event to handle.
    """

ConsoleObserver

Bases: AgentObserver

Renders agent events to the terminal using :class:rich.console.Console.

Parameters:

Name Type Description Default
console Console | None

Rich console instance. A new one is created if omitted.

None
Source code in src/tot_agent/agent.py
class ConsoleObserver(AgentObserver):
    """Renders agent events to the terminal using :class:`rich.console.Console`.

    :param rich.console.Console or None console: Rich console instance.  A new one is created if omitted.
    """

    def __init__(self, console: Console | None = None) -> None:
        self._console = console or Console()

    def on_event(self, event: AgentEvent) -> None:
        """Render *event* to the terminal.

        :param AgentEvent event: Agent event to render.
        """
        match event.event_type:
            case EventType.GOAL_START:
                self._console.print(
                    Panel(
                        f"[bold]Goal:[/bold] {event.data.get('goal', '')}",
                        title="Agent",
                        border_style="blue",
                    )
                )
            case EventType.STEP_START:
                step = event.data.get("step", "?")
                self._console.print(
                    f"\n[dim]-- Step {step} ------------------------------------------[/dim]"
                )
            case EventType.AGENT_TEXT:
                text = event.data.get("text", "")
                if text.strip():
                    self._console.print(f"[bold cyan]Agent:[/bold cyan] {text}")
            case EventType.TOOL_CALL:
                name = event.data.get("name", "")
                inp = event.data.get("input", {})
                preview = json.dumps(inp, ensure_ascii=False)[:120]
                self._console.print(
                    f"  [yellow]-> tool:[/yellow] [bold]{name}[/bold] [dim]{preview}[/dim]"
                )
            case EventType.TOOL_RESULT:
                is_screenshot = event.data.get("is_screenshot", False)
                if is_screenshot:
                    self._console.print("  [green]screenshot captured[/green]")
                else:
                    preview = str(event.data.get("result", ""))[:200]
                    self._console.print(f"  [green]✓[/green] {preview}")
            case EventType.GOAL_COMPLETE:
                summary = event.data.get("summary", "Goal completed.")
                self._console.print(
                    Panel(summary, title="Done", border_style="green")
                )
            case EventType.STEP_LIMIT:
                self._console.print(
                    Panel(
                        event.data.get("message", "Step limit reached."),
                        title="Step limit",
                        border_style="yellow",
                    )
                )

on_event(event)

Render event to the terminal.

Parameters:

Name Type Description Default
event AgentEvent

Agent event to render.

required
Source code in src/tot_agent/agent.py
def on_event(self, event: AgentEvent) -> None:
    """Render *event* to the terminal.

    :param AgentEvent event: Agent event to render.
    """
    match event.event_type:
        case EventType.GOAL_START:
            self._console.print(
                Panel(
                    f"[bold]Goal:[/bold] {event.data.get('goal', '')}",
                    title="Agent",
                    border_style="blue",
                )
            )
        case EventType.STEP_START:
            step = event.data.get("step", "?")
            self._console.print(
                f"\n[dim]-- Step {step} ------------------------------------------[/dim]"
            )
        case EventType.AGENT_TEXT:
            text = event.data.get("text", "")
            if text.strip():
                self._console.print(f"[bold cyan]Agent:[/bold cyan] {text}")
        case EventType.TOOL_CALL:
            name = event.data.get("name", "")
            inp = event.data.get("input", {})
            preview = json.dumps(inp, ensure_ascii=False)[:120]
            self._console.print(
                f"  [yellow]-> tool:[/yellow] [bold]{name}[/bold] [dim]{preview}[/dim]"
            )
        case EventType.TOOL_RESULT:
            is_screenshot = event.data.get("is_screenshot", False)
            if is_screenshot:
                self._console.print("  [green]screenshot captured[/green]")
            else:
                preview = str(event.data.get("result", ""))[:200]
                self._console.print(f"  [green]✓[/green] {preview}")
        case EventType.GOAL_COMPLETE:
            summary = event.data.get("summary", "Goal completed.")
            self._console.print(
                Panel(summary, title="Done", border_style="green")
            )
        case EventType.STEP_LIMIT:
            self._console.print(
                Panel(
                    event.data.get("message", "Step limit reached."),
                    title="Step limit",
                    border_style="yellow",
                )
            )

LoggingObserver

Bases: AgentObserver

Writes agent events to the Python :mod:logging hierarchy.

Parameters:

Name Type Description Default
log Logger | None

Logger to write to. Defaults to this module's logger.

None
Source code in src/tot_agent/agent.py
class LoggingObserver(AgentObserver):
    """Writes agent events to the Python :mod:`logging` hierarchy.

    :param logging.Logger or None log: Logger to write to.  Defaults to this module's logger.
    """

    def __init__(self, log: logging.Logger | None = None) -> None:
        self._log = log or logger

    def on_event(self, event: AgentEvent) -> None:
        """Log *event* at an appropriate level.

        :param AgentEvent event: Agent event to log.
        """
        match event.event_type:
            case EventType.GOAL_START:
                self._log.info("Goal start: %s", event.data.get("goal", ""))
            case EventType.STEP_START:
                self._log.debug("Step %s start", event.data.get("step"))
            case EventType.AGENT_TEXT:
                self._log.debug("Agent text: %s", event.data.get("text", "")[:200])
            case EventType.TOOL_CALL:
                self._log.debug(
                    "Tool call: %s %s",
                    event.data.get("name"),
                    event.data.get("input"),
                )
            case EventType.TOOL_RESULT:
                self._log.debug(
                    "Tool result (screenshot=%s): %s",
                    event.data.get("is_screenshot"),
                    str(event.data.get("result", ""))[:200],
                )
            case EventType.GOAL_COMPLETE:
                self._log.info("Goal complete: %s", event.data.get("summary", ""))
            case EventType.STEP_LIMIT:
                self._log.warning("Step limit reached: %s", event.data.get("message"))

on_event(event)

Log event at an appropriate level.

Parameters:

Name Type Description Default
event AgentEvent

Agent event to log.

required
Source code in src/tot_agent/agent.py
def on_event(self, event: AgentEvent) -> None:
    """Log *event* at an appropriate level.

    :param AgentEvent event: Agent event to log.
    """
    match event.event_type:
        case EventType.GOAL_START:
            self._log.info("Goal start: %s", event.data.get("goal", ""))
        case EventType.STEP_START:
            self._log.debug("Step %s start", event.data.get("step"))
        case EventType.AGENT_TEXT:
            self._log.debug("Agent text: %s", event.data.get("text", "")[:200])
        case EventType.TOOL_CALL:
            self._log.debug(
                "Tool call: %s %s",
                event.data.get("name"),
                event.data.get("input"),
            )
        case EventType.TOOL_RESULT:
            self._log.debug(
                "Tool result (screenshot=%s): %s",
                event.data.get("is_screenshot"),
                str(event.data.get("result", ""))[:200],
            )
        case EventType.GOAL_COMPLETE:
            self._log.info("Goal complete: %s", event.data.get("summary", ""))
        case EventType.STEP_LIMIT:
            self._log.warning("Step limit reached: %s", event.data.get("message"))

BrowserAgent

Vision-capable agentic loop that drives a Playwright browser.

The agent calls Claude with :data:~tot_agent.tools.TOOL_DEFINITIONS and iterates until the model reaches end_turn or the step cap is hit.

Parameters:

Name Type Description Default
bm BrowserManager

An initialised :class:~tot_agent.browser.BrowserManager.

required
observers list[AgentObserver] | None

Observers to notify at each lifecycle event. Defaults to [ConsoleObserver(), LoggingObserver()].

None
model str

Claude model identifier. Defaults to :data:~tot_agent.config.AGENT_MODEL.

AGENT_MODEL
max_steps int

Maximum tool-call iterations per goal. Defaults to :data:~tot_agent.config.MAX_AGENT_STEPS.

MAX_AGENT_STEPS
api_key str | None

Anthropic API key. Defaults to :data:~tot_agent.config.ANTHROPIC_API_KEY.

ANTHROPIC_API_KEY

Raises:

Type Description
RuntimeError

At construction time if api_key is None.

Source code in src/tot_agent/agent.py
class BrowserAgent:
    """Vision-capable agentic loop that drives a Playwright browser.

    The agent calls Claude with :data:`~tot_agent.tools.TOOL_DEFINITIONS` and
    iterates until the model reaches ``end_turn`` or the step cap is hit.

    :param BrowserManager bm: An initialised :class:`~tot_agent.browser.BrowserManager`.
    :param list[AgentObserver] or None observers: Observers to notify at each lifecycle event.  Defaults
        to ``[ConsoleObserver(), LoggingObserver()]``.
    :param str model: Claude model identifier.  Defaults to
        :data:`~tot_agent.config.AGENT_MODEL`.
    :param int max_steps: Maximum tool-call iterations per goal.  Defaults to
        :data:`~tot_agent.config.MAX_AGENT_STEPS`.
    :param str or None api_key: Anthropic API key.  Defaults to
        :data:`~tot_agent.config.ANTHROPIC_API_KEY`.
    :raises RuntimeError: At construction time if *api_key* is ``None``.
    """

    def __init__(
        self,
        bm: BrowserManager,
        observers: list[AgentObserver] | None = None,
        model: str = AGENT_MODEL,
        max_steps: int = MAX_AGENT_STEPS,
        api_key: str | None = ANTHROPIC_API_KEY,
        client: Any | None = None,
    ) -> None:
        if client is None and api_key is None:
            raise RuntimeError(
                "ANTHROPIC_API_KEY is not set. "
                "Export it as an environment variable or add it to .env."
            )
        self.bm = bm
        self.model = model
        self.max_steps = max_steps
        self._observers: list[AgentObserver] = observers or [
            ConsoleObserver(),
            LoggingObserver(),
        ]
        self._client = client or _create_default_client(api_key)
        self._system = _build_system_prompt(SIM_USERS)

    # ------------------------------------------------------------------
    # Observer management
    # ------------------------------------------------------------------

    def add_observer(self, observer: AgentObserver) -> None:
        """Register an additional observer.

        :param AgentObserver observer: Observer to add.
        """
        self._observers.append(observer)

    def remove_observer(self, observer: AgentObserver) -> None:
        """Deregister an observer.

        :param AgentObserver observer: Observer to remove.
        """
        self._observers.remove(observer)

    def _emit(self, event: AgentEvent) -> None:
        """Notify all registered observers of *event*.

        :param AgentEvent event: The event to broadcast.
        """
        for obs in self._observers:
            obs.on_event(event)

    # ------------------------------------------------------------------
    # Core loop
    # ------------------------------------------------------------------

    async def _create_message_response(self, messages: list[dict[str, Any]]) -> Any:
        """Call the model client without blocking the event loop."""
        create = self._client.messages.create
        kwargs = {
            "model": self.model,
            "max_tokens": 4096,
            "system": self._system,
            "tools": TOOL_DEFINITIONS,
            "messages": messages,
        }

        if inspect.iscoroutinefunction(create):
            return await create(**kwargs)
        return await asyncio.to_thread(create, **kwargs)

    async def run(self, goal: str) -> str:
        """Execute the agent loop until the goal is achieved or the step cap is hit.

        :param str goal: Plain-English objective for the agent.
        :returns: A human-readable summary of what was accomplished.
        :rtype: str
        """
        self._emit(AgentEvent(EventType.GOAL_START, {"goal": goal}))
        logger.info("Agent run started: %s", goal[:80])

        messages: list[dict] = [{"role": "user", "content": goal}]

        for step in range(1, self.max_steps + 1):
            self._emit(AgentEvent(EventType.STEP_START, {"step": step}))

            response = await self._create_message_response(messages)

            text_parts: list[str] = []
            tool_calls: list[Any] = []

            for block in response.content:
                if block.type == "text":
                    text_parts.append(block.text)
                    self._emit(
                        AgentEvent(EventType.AGENT_TEXT, {"text": block.text})
                    )
                elif block.type == "tool_use":
                    tool_calls.append(block)

            # Append the full assistant response to history.
            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn" or not tool_calls:
                summary = "\n".join(text_parts) or "Goal completed."
                self._emit(
                    AgentEvent(EventType.GOAL_COMPLETE, {"summary": summary})
                )
                logger.info("Agent completed in %d step(s)", step)
                return summary

            # Execute tool calls.
            tool_results: list[dict] = []
            for tc in tool_calls:
                self._emit(
                    AgentEvent(
                        EventType.TOOL_CALL,
                        {"name": tc.name, "input": tc.input},
                    )
                )
                result = await dispatch(tc.name, tc.input, self.bm)
                is_screenshot = isinstance(result, dict) and result.get("type") == "screenshot"
                self._emit(
                    AgentEvent(
                        EventType.TOOL_RESULT,
                        {"result": result, "is_screenshot": is_screenshot},
                    )
                )
                tool_results.append(format_tool_result(tc.id, result))

            messages.append({"role": "user", "content": tool_results})

        msg = f"Reached step limit ({self.max_steps}). Last goal: {goal}"
        self._emit(AgentEvent(EventType.STEP_LIMIT, {"message": msg}))
        logger.warning("Step limit reached after %d steps", self.max_steps)
        return msg

add_observer(observer)

Register an additional observer.

Parameters:

Name Type Description Default
observer AgentObserver

Observer to add.

required
Source code in src/tot_agent/agent.py
def add_observer(self, observer: AgentObserver) -> None:
    """Register an additional observer.

    :param AgentObserver observer: Observer to add.
    """
    self._observers.append(observer)

remove_observer(observer)

Deregister an observer.

Parameters:

Name Type Description Default
observer AgentObserver

Observer to remove.

required
Source code in src/tot_agent/agent.py
def remove_observer(self, observer: AgentObserver) -> None:
    """Deregister an observer.

    :param AgentObserver observer: Observer to remove.
    """
    self._observers.remove(observer)

run(goal) async

Execute the agent loop until the goal is achieved or the step cap is hit.

Parameters:

Name Type Description Default
goal str

Plain-English objective for the agent.

required

Returns:

Type Description
str

A human-readable summary of what was accomplished.

Source code in src/tot_agent/agent.py
async def run(self, goal: str) -> str:
    """Execute the agent loop until the goal is achieved or the step cap is hit.

    :param str goal: Plain-English objective for the agent.
    :returns: A human-readable summary of what was accomplished.
    :rtype: str
    """
    self._emit(AgentEvent(EventType.GOAL_START, {"goal": goal}))
    logger.info("Agent run started: %s", goal[:80])

    messages: list[dict] = [{"role": "user", "content": goal}]

    for step in range(1, self.max_steps + 1):
        self._emit(AgentEvent(EventType.STEP_START, {"step": step}))

        response = await self._create_message_response(messages)

        text_parts: list[str] = []
        tool_calls: list[Any] = []

        for block in response.content:
            if block.type == "text":
                text_parts.append(block.text)
                self._emit(
                    AgentEvent(EventType.AGENT_TEXT, {"text": block.text})
                )
            elif block.type == "tool_use":
                tool_calls.append(block)

        # Append the full assistant response to history.
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn" or not tool_calls:
            summary = "\n".join(text_parts) or "Goal completed."
            self._emit(
                AgentEvent(EventType.GOAL_COMPLETE, {"summary": summary})
            )
            logger.info("Agent completed in %d step(s)", step)
            return summary

        # Execute tool calls.
        tool_results: list[dict] = []
        for tc in tool_calls:
            self._emit(
                AgentEvent(
                    EventType.TOOL_CALL,
                    {"name": tc.name, "input": tc.input},
                )
            )
            result = await dispatch(tc.name, tc.input, self.bm)
            is_screenshot = isinstance(result, dict) and result.get("type") == "screenshot"
            self._emit(
                AgentEvent(
                    EventType.TOOL_RESULT,
                    {"result": result, "is_screenshot": is_screenshot},
                )
            )
            tool_results.append(format_tool_result(tc.id, result))

        messages.append({"role": "user", "content": tool_results})

    msg = f"Reached step limit ({self.max_steps}). Last goal: {goal}"
    self._emit(AgentEvent(EventType.STEP_LIMIT, {"message": msg}))
    logger.warning("Step limit reached after %d steps", self.max_steps)
    return msg

GoalTemplate

Base class for structured goal strings (Template Method pattern).

Subclass and override :meth:build to create reusable, parameterised goal strings for common test scenarios.

Source code in src/tot_agent/agent.py
class GoalTemplate:
    """Base class for structured goal strings (Template Method pattern).

    Subclass and override :meth:`build` to create reusable, parameterised
    goal strings for common test scenarios.
    """

    def build(self, **kwargs: Any) -> str:
        """Render the goal string from the provided keyword arguments.

        :returns: A plain-English goal string ready to pass to
            :meth:`BrowserAgent.run`.
        :rtype: str
        :raises NotImplementedError: In the base class.
        """
        raise NotImplementedError

build(**kwargs)

Render the goal string from the provided keyword arguments.

Returns:

Type Description
str

A plain-English goal string ready to pass to :meth:BrowserAgent.run.

Raises:

Type Description
NotImplementedError

In the base class.

Source code in src/tot_agent/agent.py
def build(self, **kwargs: Any) -> str:
    """Render the goal string from the provided keyword arguments.

    :returns: A plain-English goal string ready to pass to
        :meth:`BrowserAgent.run`.
    :rtype: str
    :raises NotImplementedError: In the base class.
    """
    raise NotImplementedError

CreateTestsGoal

Bases: GoalTemplate

Goal: create A/B tests with real book covers.

Parameters:

Name Type Description Default
count int

Number of tests to create. Defaults to 5.

5
genre str

Book genre, or "mixed" for variety. Defaults to "mixed".

'mixed'
Source code in src/tot_agent/agent.py
class CreateTestsGoal(GoalTemplate):
    """Goal: create A/B tests with real book covers.

    :param int count: Number of tests to create.  Defaults to ``5``.
    :param str genre: Book genre, or ``"mixed"`` for variety.  Defaults to
        ``"mixed"``.
    """

    def __init__(self, count: int = 5, genre: str = "mixed") -> None:
        self.count = count
        self.genre = genre

    def build(self, **kwargs: Any) -> str:
        """Render the create-tests goal string.

        :returns: Goal string.
        :rtype: str
        """
        genre_hint = (
            f"Use {self.genre} genre covers."
            if self.genre != "mixed"
            else "Use a variety of genres."
        )
        return (
            f"Create {self.count} A/B tests on the book cover testing platform. "
            f"{genre_hint} "
            "For each test: fetch two different book covers, log in as admin if not already, "
            "navigate to the test creation page, fill in the test name and paste the two "
            "cover URLs, and submit. Take a screenshot after each submission to confirm it "
            "worked. Report how many tests were successfully created."
        )

build(**kwargs)

Render the create-tests goal string.

Returns:

Type Description
str

Goal string.

Source code in src/tot_agent/agent.py
def build(self, **kwargs: Any) -> str:
    """Render the create-tests goal string.

    :returns: Goal string.
    :rtype: str
    """
    genre_hint = (
        f"Use {self.genre} genre covers."
        if self.genre != "mixed"
        else "Use a variety of genres."
    )
    return (
        f"Create {self.count} A/B tests on the book cover testing platform. "
        f"{genre_hint} "
        "For each test: fetch two different book covers, log in as admin if not already, "
        "navigate to the test creation page, fill in the test name and paste the two "
        "cover URLs, and submit. Take a screenshot after each submission to confirm it "
        "worked. Report how many tests were successfully created."
    )

VoteGoal

Bases: GoalTemplate

Goal: have a single user vote on existing tests.

Parameters:

Name Type Description Default
username str

Voter's username.

required
password str

Voter's password.

required
vote_count int

Number of tests to vote on. Defaults to 3.

3
bias str

Voting bias hint. Defaults to "random".

'random'
Source code in src/tot_agent/agent.py
class VoteGoal(GoalTemplate):
    """Goal: have a single user vote on existing tests.

    :param str username: Voter's username.
    :param str password: Voter's password.
    :param int vote_count: Number of tests to vote on.  Defaults to ``3``.
    :param str bias: Voting bias hint.  Defaults to ``"random"``.
    """

    _BIAS_HINTS: ClassVar[dict[str, str]] = {
        "prefers_dark": "Prefer covers with darker colour palettes.",
        "prefers_bright": "Prefer covers with brighter, more colourful designs.",
        "prefers_illustrated": "Prefer covers with illustrated or artistic designs.",
        "random": "Vote randomly -- pick either cover.",
    }

    def __init__(
        self,
        username: str,
        password: str,
        vote_count: int = 3,
        bias: str = "random",
    ) -> None:
        self.username = username
        self.password = password
        self.vote_count = vote_count
        self.bias = bias

    def build(self, **kwargs: Any) -> str:
        """Render the vote goal string.

        :returns: Goal string.
        :rtype: str
        """
        bias_hint = self._BIAS_HINTS.get(self.bias, "Vote randomly.")
        return (
            f"Log in as user '{self.username}' (password: '{self.password}'). "
            f"Navigate to the A/B tests page and cast votes on up to {self.vote_count} tests. "
            f"{bias_hint} "
            "Take a screenshot before and after each vote to verify it registered. "
            "Report which tests you voted on and which cover you chose each time."
        )

build(**kwargs)

Render the vote goal string.

Returns:

Type Description
str

Goal string.

Source code in src/tot_agent/agent.py
def build(self, **kwargs: Any) -> str:
    """Render the vote goal string.

    :returns: Goal string.
    :rtype: str
    """
    bias_hint = self._BIAS_HINTS.get(self.bias, "Vote randomly.")
    return (
        f"Log in as user '{self.username}' (password: '{self.password}'). "
        f"Navigate to the A/B tests page and cast votes on up to {self.vote_count} tests. "
        f"{bias_hint} "
        "Take a screenshot before and after each vote to verify it registered. "
        "Report which tests you voted on and which cover you chose each time."
    )

SimulateAllUsersGoal

Bases: GoalTemplate

Goal: simulate all configured users voting.

Parameters:

Name Type Description Default
vote_count_each int

Number of votes per user. Defaults to 2.

2
Source code in src/tot_agent/agent.py
class SimulateAllUsersGoal(GoalTemplate):
    """Goal: simulate all configured users voting.

    :param int vote_count_each: Number of votes per user.  Defaults to ``2``.
    """

    def __init__(self, vote_count_each: int = 2) -> None:
        self.vote_count_each = vote_count_each

    def build(self, **kwargs: Any) -> str:
        """Render the simulate-all-users goal string.

        :returns: Goal string.
        :rtype: str
        """
        lines = [
            f"Switch to user '{u.username}', login with password '{u.password}', "
            f"then vote on {self.vote_count_each} tests (bias: {u.voting_bias})."
            for u in SIM_USERS
        ]
        steps = " Then ".join(lines)
        return (
            "Simulate a full round of voting across all users. "
            + steps
            + " After all users have voted, navigate to the dashboard or results page "
            "as admin and take a final screenshot showing the vote tallies."
        )

build(**kwargs)

Render the simulate-all-users goal string.

Returns:

Type Description
str

Goal string.

Source code in src/tot_agent/agent.py
def build(self, **kwargs: Any) -> str:
    """Render the simulate-all-users goal string.

    :returns: Goal string.
    :rtype: str
    """
    lines = [
        f"Switch to user '{u.username}', login with password '{u.password}', "
        f"then vote on {self.vote_count_each} tests (bias: {u.voting_bias})."
        for u in SIM_USERS
    ]
    steps = " Then ".join(lines)
    return (
        "Simulate a full round of voting across all users. "
        + steps
        + " After all users have voted, navigate to the dashboard or results page "
        "as admin and take a final screenshot showing the vote tallies."
    )

FullSeedGoal

Bases: GoalTemplate

Goal: create tests, run voting simulation, view results.

Parameters:

Name Type Description Default
test_count int

Number of A/B tests to create. Defaults to 5.

5
vote_rounds int

Voting rounds per user. Defaults to 1.

1
Source code in src/tot_agent/agent.py
class FullSeedGoal(GoalTemplate):
    """Goal: create tests, run voting simulation, view results.

    :param int test_count: Number of A/B tests to create.  Defaults to ``5``.
    :param int vote_rounds: Voting rounds per user.  Defaults to ``1``.
    """

    def __init__(self, test_count: int = 5, vote_rounds: int = 1) -> None:
        self.test_count = test_count
        self.vote_rounds = vote_rounds

    def build(self, **kwargs: Any) -> str:
        """Render the full-seed goal string.

        :returns: Goal string.
        :rtype: str
        """
        return (
            f"Seed the A/B testing platform with {self.test_count} tests and then simulate voting. "
            "Step 1: As admin, create the tests using real book cover images from Open Library. "
            f"Step 2: Have each simulated user vote on all available tests ({self.vote_rounds} round). "
            "Step 3: As admin, view the results dashboard and take a final screenshot. "
            "Report a summary of everything that was done."
        )

build(**kwargs)

Render the full-seed goal string.

Returns:

Type Description
str

Goal string.

Source code in src/tot_agent/agent.py
def build(self, **kwargs: Any) -> str:
    """Render the full-seed goal string.

    :returns: Goal string.
    :rtype: str
    """
    return (
        f"Seed the A/B testing platform with {self.test_count} tests and then simulate voting. "
        "Step 1: As admin, create the tests using real book cover images from Open Library. "
        f"Step 2: Have each simulated user vote on all available tests ({self.vote_rounds} round). "
        "Step 3: As admin, view the results dashboard and take a final screenshot. "
        "Report a summary of everything that was done."
    )