9.2The ConstrainedTestHarness Class

from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class HarnessConfig:
    max_steps: int = 30
    timeout_seconds: int = 300
    max_tokens: int = 50_000
    allowed_domains: list[str] = None
    allowed_actions: list[str] = None
    require_assertion: bool = True
    screenshot_on_failure: bool = True

class ConstrainedTestHarness:
    def __init__(self, agent, config: HarnessConfig):
        self.agent = agent
        self.config = config
        self.step_count = 0
        self.token_count = 0
        self.start_time = None
        self.violations = []

    def execute(self, test_objective: str) -> TestResult:
        self.start_time = time.time()

        while True:
            # Check all guardrails before each step
            violation = self.check_guardrails()
            if violation:
                return TestResult(
                    status="ABORTED",
                    reason=f"Guardrail violation: {violation}",
                    steps_taken=self.step_count,
                    violations=self.violations
                )

            # Let the agent take one step
            action = self.agent.next_action()

            # Validate the action before execution
            if not self.is_action_allowed(action):
                self.violations.append(f"Blocked action: {action}")
                return TestResult(
                    status="BLOCKED",
                    reason=f"Action not allowed: {action}",
                    steps_taken=self.step_count
                )

            # Execute and record
            result = self.agent.execute_action(action)
            self.step_count += 1
            self.token_count += result.tokens_used

            if result.is_terminal:
                # Validate the test result
                if self.config.require_assertion and not result.has_assertion:
                    return TestResult(
                        status="INVALID",
                        reason="Test completed without any assertion",
                        steps_taken=self.step_count
                    )
                return result

    def check_guardrails(self) -> Optional[str]:
        if self.step_count >= self.config.max_steps:
            return f"Max steps ({self.config.max_steps}) exceeded"
        elapsed = time.time() - self.start_time
        if elapsed > self.config.timeout_seconds:
            return f"Timeout ({self.config.timeout_seconds}s) exceeded"
        if self.token_count >= self.config.max_tokens:
            return f"Token budget ({self.config.max_tokens}) exhausted"
        return None

    def is_action_allowed(self, action) -> bool:
        if self.config.allowed_actions:
            if action.type not in self.config.allowed_actions:
                return False
        if self.config.allowed_domains and action.type == "NAVIGATE":
            from urllib.parse import urlparse
            domain = urlparse(action.url).netloc
            if domain not in self.config.allowed_domains:
                return False
        return True