Running Tests

"""
Run Voxli tests against your chatbot or AI agent.

1. Create a test run
2. Get all tests for the scenario
3. Simulate each test
"""

import os
import time
import requests


def poll_next_message(endpoint: str, headers: dict, timeout: int = 30) -> dict | None:
    """Poll next-message until the tester is ready or the chat ends.

    Returns None when the chat is over, otherwise a dict with either a
    `message` (free text) or an `action` (an ActionInvocation) field.
    """
    start_time = time.time()
    while True:
        response = requests.post(endpoint, headers=headers)
        response.raise_for_status()
        data = response.json()
        if data["ready"]:
            if data.get("end_chat"):
                return None
            return {"message": data.get("message"), "action": data.get("action")}
        if time.time() - start_time > timeout:
            raise TimeoutError("Timed out waiting for message")
        time.sleep(1)


# --- Configuration ---
api_key = os.getenv("VOXLI_API_KEY")
base_url = os.getenv("VOXLI_API_URL", "https://api.voxli.io")
scenario_id = os.getenv("VOXLI_SCENARIO_ID")
agent_id = os.getenv("VOXLI_AGENT_ID")

headers = {"Authorization": f"Bearer {api_key}"}

# 1. Create a test run
run = requests.post(f"{base_url}/runs/", headers=headers, json={
    "scenario": scenario_id,
    "agent": agent_id,
    "status": "running"
}).json()
run_id = run["id"]

# 2. Get all tests for this scenario
tests = requests.get(f"{base_url}/scenarios/{scenario_id}/tests", headers=headers).json()["data"]

# 3. Simulate each test
for test in tests:
    # 3a. Create a test result entry
    result = requests.post(f"{base_url}/test-results/", headers=headers, json={
        "test": test["id"],
        "run": run_id,
        "agent": agent_id
    }).json()
    result_id = result["id"]
    generate_endpoint = f"{base_url}/test-results/{result_id}/next-message"
    conversation_endpoint = f"{base_url}/test-results/{result_id}/conversation"

    # 3b. Get first turn from Voxli
    turn = poll_next_message(generate_endpoint, headers)

    # 3c. Conversation loop
    while turn is not None:
        # TODO: Replace with your agent's response
        start = time.monotonic()
        if turn.get("action"):
            # Tester invoked a registered action instead of typing. Apply
            # it in your system, then record the chatbot's follow-up.
            agent_response = your_agent.apply_action(
                turn["action"]["name"],
                turn["action"].get("arguments", {}),
            )
        else:
            agent_response = your_agent.process(turn["message"])
        response_time_ms = round((time.monotonic() - start) * 1000)

        # 3d. Record agent response (include metadata for performance tracking)
        requests.post(
            conversation_endpoint,
            headers=headers, json={
                "type": "message",
                "content": agent_response,
                "metadata": {
                    "responseTime": response_time_ms,
                    "inputTokens": input_tokens,
                    "outputTokens": output_tokens,
                    "cost": cost,
                }
            }
        )

        # 3e. Get next turn from Voxli
        turn = poll_next_message(generate_endpoint, headers)

print(f"Test run {run_id} completed.")

import time

start = time.monotonic()
agent_response = get_agent_response(tester_message)  # your agent logic
response_time_ms = round((time.monotonic() - start) * 1000)

requests.post(conversation_endpoint, headers=headers, json={
    "type": "message",
    "content": agent_response,
    "metadata": {
        "responseTime": response_time_ms,
        "inputTokens": input_tokens,
        "outputTokens": output_tokens,
        "cost": cost,
    }
})

"""
Attach data after a chat ends, then finish the result yourself.

1. Create a test result with completionMode "manual"
2. Run the conversation loop until end_chat
3. Attach post-chat data to the conversation
4. Finish the result to start scoring
"""

import os
import time
import requests

api_key = os.getenv("VOXLI_API_KEY")
base_url = os.getenv("VOXLI_API_URL", "https://api.voxli.io")
test_id = os.getenv("VOXLI_TEST_ID")
agent_id = os.getenv("VOXLI_AGENT_ID")

headers = {"Authorization": f"Bearer {api_key}"}

# 1. Create a test result that waits for an explicit finish call.
result = requests.post(f"{base_url}/test-results/", headers=headers, json={
    "test": test_id,
    "agent": agent_id,
    "completionMode": "manual",
}).json()
result_id = result["id"]
conversation_endpoint = f"{base_url}/test-results/{result_id}/conversation"

# 2. Run the conversation loop as usual (see the complete example).
# Every recorded entry gets a server-generated id. Keep the ids you
# care about; they are also returned by GET /test-results/{id}.
reply = requests.post(conversation_endpoint, headers=headers, json={
    "type": "message",
    "content": agent_response,
}).json()
reply_id = reply["conversation"][-1]["id"]

# ... keep relaying turns until next-message returns end_chat: true ...

# 3a. Merge late data into an earlier message, for example metrics or
# tracing details that your observability stack delivers after the fact.
requests.patch(f"{conversation_endpoint}/{reply_id}", headers=headers, json={
    "metadata": {
        "responseTime": 840,
        "traceId": "trace-8c14",
    }
})

# 3b. Insert a tool call at the position where it actually happened.
requests.post(conversation_endpoint, headers=headers, json={
    "type": "tool",
    "name": "lookup_order",
    "metadata": {"order_id": "A-1042"},
    "after": reply_id,
})

# 3c. Internal events recorded now are visible to scoring but were never
# part of the chat, so they cannot influence the conversation itself.
requests.post(conversation_endpoint, headers=headers, json={
    "type": "internal-event",
    "name": "session_summary",
    "metadata": {"handoff": False, "sentiment": "positive"},
})

# 4. Finish the result. Scoring runs in the background, so poll the
# result until its status turns "completed".
requests.post(f"{base_url}/test-results/{result_id}/finish", headers=headers)

while True:
    status = requests.get(
        f"{base_url}/test-results/{result_id}", headers=headers
    ).json()["status"]
    if status in ("completed", "failed", "canceled"):
        break
    time.sleep(2)

print(f"Test result {result_id} finished with status {status}.")

new_result = requests.post(
    f"{base_url}/test-results/{result_id}/retry",
    headers=headers,
).json()
# new_result['id'] is the new PENDING attempt, attached to the same run as
# the original. new_result['metadata']['previous_test_result_id'] points
# back to the original, which is now canceled.