Local

npm install -g @voxli/cli

voxli auth

"""
Run Voxli test conversations.

Environment variables:
- VOXLI_API_TOKEN: Your API key (required)
- VOXLI_TEST_RESULT_IDS: JSON array of test result IDs (required)
"""

import json
import os
import sys
import time
import requests


def get_agent_response(message: str) -> str:
    """
    Replace this with your actual agent integration.

    Receives a message from the Voxli tester and returns
    your chatbot's response.
    """
    return "The meaning of life is 42."


# In interactive sessions the tester is a person (or agent) typing, so `ready`
# can stay false for minutes between turns — far longer than an automated run.
# Voxli ends abandoned sessions on its side, but don't rely on that alone: this
# runner is a billed CI job, so it holds its own deadline and turn cap and exits
# on its own if the server never says end_chat.
SESSION_DEADLINE_SECONDS = 90 * 60
MAX_TURNS = 200
POLL_INTERVAL_SECONDS = 2


def poll_next_message(endpoint: str, headers: dict, deadline: float) -> str | None:
    """Wait for the next tester message, or None once the chat is over.

    `deadline` is an absolute time covering the whole session, not one turn,
    so a long chat cannot extend the job indefinitely one turn at a time.
    """
    while True:
        response = requests.post(endpoint, headers=headers)
        response.raise_for_status()
        data = response.json()
        if data["ready"]:
            return None if data.get("end_chat") else data["message"]
        if time.time() > deadline:
            raise TimeoutError("Session deadline reached waiting for a message")
        time.sleep(POLL_INTERVAL_SECONDS)


def run_test(test_result_id: str, api_key: str, base_url: str):
    headers = {"Authorization": f"Bearer {api_key}"}
    conversation_endpoint = f"{base_url}/test-results/{test_result_id}/conversation"
    generate_endpoint = f"{base_url}/test-results/{test_result_id}/next-message"
    deadline = time.time() + SESSION_DEADLINE_SECONDS

    # Get first message from Voxli
    tester_message = poll_next_message(generate_endpoint, headers, deadline)

    # Converse until Voxli ends the chat (end_chat=true), bounded by our own
    # deadline and turn cap so the job always terminates.
    for _ in range(MAX_TURNS):
        if tester_message is None:
            print(f"Test {test_result_id} completed")
            return

        agent_response = get_agent_response(tester_message)

        # Record agent response
        response = requests.post(
            conversation_endpoint, headers=headers, json={"type": "message", "content": agent_response}
        )
        response.raise_for_status()

        # Get next message from Voxli
        tester_message = poll_next_message(generate_endpoint, headers, deadline)

    print(f"Test {test_result_id} stopped after {MAX_TURNS} turns")


def main():
    api_key = os.getenv("VOXLI_API_TOKEN")
    test_result_ids_raw = os.getenv("VOXLI_TEST_RESULT_IDS")
    base_url = os.getenv("VOXLI_API_URL", "https://api.voxli.io")

    if not api_key or not test_result_ids_raw:
        print("Error: VOXLI_API_TOKEN and VOXLI_TEST_RESULT_IDS are required")
        sys.exit(1)

    test_result_ids = json.loads(test_result_ids_raw)

    for test_result_id in test_result_ids:
        run_test(test_result_id, api_key, base_url)


if __name__ == "__main__":
    main()

voxli listen --command "python run_test.py"