server/response_specs.py

"""
Response templates for the response validation task.

Each template defines what a correct API response looks like for a given
request type. The environment generates a broken response by injecting
issues (wrong status code, missing fields, wrong types, extra fields).

Response issue types:
- wrong_status_code: Response has incorrect HTTP status code
- missing_response_field: Required field missing from response body
- wrong_response_type: Field present but wrong data type
- extra_response_field: Unexpected field in response (data leak risk)
- inconsistent_error_format: Error response doesn't follow spec format
"""

import copy
import random
from typing import Any, Dict, List, Tuple

# Response issue types the agent must identify
RESPONSE_ISSUE_TYPES = [
    "wrong_status_code",
    "missing_response_field",
    "wrong_response_type",
    "extra_response_field",
    "inconsistent_error_format",
]


# Maps API operation type to expected success response
RESPONSE_TEMPLATES = [
    {
        "name": "Create Resource",
        "success_status": 201,
        "success_body": {
            "id": "res_abc123",
            "status": "created",
            "created_at": "2025-01-15T10:30:00Z",
        },
        "required_response_fields": ["id", "status", "created_at"],
        "field_types": {"id": "string", "status": "string", "created_at": "string"},
        "error_status": 400,
        "error_body": {
            "error": {"code": "VALIDATION_ERROR", "message": "Invalid input", "details": []},
        },
    },
    {
        "name": "List Resources",
        "success_status": 200,
        "success_body": {
            "data": [{"id": "res_1", "name": "Item 1"}, {"id": "res_2", "name": "Item 2"}],
            "total": 2,
            "page": 1,
            "per_page": 20,
        },
        "required_response_fields": ["data", "total", "page", "per_page"],
        "field_types": {"data": "array", "total": "integer", "page": "integer", "per_page": "integer"},
        "error_status": 401,
        "error_body": {
            "error": {"code": "UNAUTHORIZED", "message": "Invalid API key", "details": []},
        },
    },
    {
        "name": "Update Resource",
        "success_status": 200,
        "success_body": {
            "id": "res_abc123",
            "status": "updated",
            "updated_at": "2025-01-15T12:00:00Z",
        },
        "required_response_fields": ["id", "status", "updated_at"],
        "field_types": {"id": "string", "status": "string", "updated_at": "string"},
        "error_status": 404,
        "error_body": {
            "error": {"code": "NOT_FOUND", "message": "Resource not found", "details": []},
        },
    },
    {
        "name": "Delete Resource",
        "success_status": 204,
        "success_body": {},
        "required_response_fields": [],
        "field_types": {},
        "error_status": 403,
        "error_body": {
            "error": {"code": "FORBIDDEN", "message": "Insufficient permissions", "details": []},
        },
    },
    {
        "name": "Batch Operation",
        "success_status": 200,
        "success_body": {
            "processed": 5,
            "failed": 0,
            "results": [
                {"id": "item_1", "status": "success"},
                {"id": "item_2", "status": "success"},
            ],
        },
        "required_response_fields": ["processed", "failed", "results"],
        "field_types": {"processed": "integer", "failed": "integer", "results": "array"},
        "error_status": 422,
        "error_body": {
            "error": {"code": "UNPROCESSABLE", "message": "Batch validation failed", "details": []},
        },
    },
    {
        "name": "Authentication",
        "success_status": 200,
        "success_body": {
            "access_token": "eyJhbGciOiJIUzI1NiJ9.token",
            "token_type": "Bearer",
            "expires_in": 3600,
        },
        "required_response_fields": ["access_token", "token_type", "expires_in"],
        "field_types": {"access_token": "string", "token_type": "string", "expires_in": "integer"},
        "error_status": 401,
        "error_body": {
            "error": {"code": "INVALID_CREDENTIALS", "message": "Bad credentials", "details": []},
        },
    },
    {
        "name": "File Upload",
        "success_status": 201,
        "success_body": {
            "file_id": "file_xyz789",
            "filename": "report.pdf",
            "size_bytes": 1048576,
            "url": "https://cdn.example.com/files/file_xyz789",
        },
        "required_response_fields": ["file_id", "filename", "size_bytes", "url"],
        "field_types": {"file_id": "string", "filename": "string", "size_bytes": "integer", "url": "string"},
        "error_status": 413,
        "error_body": {
            "error": {"code": "PAYLOAD_TOO_LARGE", "message": "File exceeds 10MB limit", "details": []},
        },
    },
    {
        "name": "Search Query",
        "success_status": 200,
        "success_body": {
            "query": "test search",
            "results": [{"id": "doc_1", "score": 0.95, "title": "Test Document"}],
            "total_results": 1,
            "search_time_ms": 42,
        },
        "required_response_fields": ["query", "results", "total_results", "search_time_ms"],
        "field_types": {"query": "string", "results": "array", "total_results": "integer", "search_time_ms": "integer"},
        "error_status": 400,
        "error_body": {
            "error": {"code": "INVALID_QUERY", "message": "Query syntax error", "details": []},
        },
    },
]


def get_random_response_template(rng: random.Random) -> Dict[str, Any]:
    """Pick a random response template."""
    return copy.deepcopy(rng.choice(RESPONSE_TEMPLATES))


def inject_response_issues(
    template: Dict[str, Any],
    rng: random.Random,
    issue_count: int = 1,
) -> Tuple[Dict[str, Any], int, List[Dict[str, str]]]:
    """Inject issues into a response and return (broken_body, broken_status, ground_truths).

    Each ground truth has: issue_type, description, affected_field (if applicable).
    """
    # Decide if we break a success response or an error response
    use_success = rng.random() < 0.6
    if use_success:
        body = copy.deepcopy(template["success_body"])
        status = template["success_status"]
    else:
        body = copy.deepcopy(template["error_body"])
        status = template["error_status"]

    ground_truths: List[Dict[str, str]] = []
    available_issues = list(RESPONSE_ISSUE_TYPES)
    rng.shuffle(available_issues)

    injected = 0
    for issue_type in available_issues:
        if injected >= issue_count:
            break

        if issue_type == "wrong_status_code":
            wrong_codes = [200, 201, 204, 301, 400, 401, 403, 404, 422, 429, 500, 502, 503]
            wrong_codes = [c for c in wrong_codes if c != status]
            old_status = status
            status = rng.choice(wrong_codes)
            ground_truths.append({
                "issue_type": "wrong_status_code",
                "description": f"Expected status {old_status}, got {status}",
                "affected_field": "status_code",
                "correct_value": str(old_status),
            })
            injected += 1

        elif issue_type == "missing_response_field" and template["required_response_fields"]:
            fields = list(template["required_response_fields"])
            rng.shuffle(fields)
            field = fields[0]
            if field in body:
                del body[field]
                ground_truths.append({
                    "issue_type": "missing_response_field",
                    "description": f"Required field '{field}' missing from response",
                    "affected_field": field,
                })
                injected += 1

        elif issue_type == "wrong_response_type" and template["field_types"]:
            typed_fields = [f for f in template["field_types"] if f in body]
            if typed_fields:
                field = rng.choice(typed_fields)
                original_type = template["field_types"][field]
                # Replace with wrong type
                if original_type == "string":
                    body[field] = 12345
                elif original_type == "integer":
                    body[field] = "not_a_number"
                elif original_type == "array":
                    body[field] = "should_be_array"
                else:
                    body[field] = [1, 2, 3]
                ground_truths.append({
                    "issue_type": "wrong_response_type",
                    "description": f"Field '{field}' should be {original_type}",
                    "affected_field": field,
                })
                injected += 1

        elif issue_type == "extra_response_field":
            leak_fields = {
                "internal_id": "int_9f8a2b",
                "debug_trace": "stack trace at line 42",
                "db_query": "SELECT * FROM users WHERE id=123",
                "server_ip": "10.0.0.42",
                "session_token": "sess_leaked_abc",
            }
            field, value = rng.choice(list(leak_fields.items()))
            body[field] = value
            ground_truths.append({
                "issue_type": "extra_response_field",
                "description": f"Unexpected field '{field}' in response (potential data leak)",
                "affected_field": field,
            })
            injected += 1

        elif issue_type == "inconsistent_error_format" and not use_success:
            # Break the error format -- flatten it or use wrong keys
            variants = [
                {"msg": body.get("error", {}).get("message", "error"), "err_code": "UNKNOWN"},
                {"error_message": "Something went wrong", "status": "error"},
                {"errors": [{"msg": "bad request"}]},
            ]
            body = rng.choice(variants)
            ground_truths.append({
                "issue_type": "inconsistent_error_format",
                "description": "Error response doesn't follow standard format: {error: {code, message, details}}",
                "affected_field": "error",
            })
            injected += 1

    return body, status, ground_truths