File size: 5,654 Bytes
5cf6185 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | """
Typed Pydantic models for the API Contract Debugger environment.
The environment gives an agent a broken OpenAPI-style spec and asks it to
fix contract violations by proposing targeted field-level corrections.
"""
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Optional
from openenv.core.env_server.types import Action, Observation, State
from pydantic import Field
# ---------------------------------------------------------------------------
# Domain types
# ---------------------------------------------------------------------------
class FieldType(str, Enum):
"""Supported JSON Schema primitive types."""
STRING = "string"
INTEGER = "integer"
NUMBER = "number"
BOOLEAN = "boolean"
ARRAY = "array"
OBJECT = "object"
NULL = "null"
class HttpMethod(str, Enum):
GET = "GET"
POST = "POST"
PUT = "PUT"
PATCH = "PATCH"
DELETE = "DELETE"
class ActionKind(str, Enum):
"""What kind of fix the agent is proposing."""
ADD_FIELD = "add_field" # Add a missing required field
REMOVE_FIELD = "remove_field" # Remove a forbidden/extra field
CHANGE_TYPE = "change_type" # Fix a field's type
CHANGE_STATUS = "change_status" # Fix an HTTP status code
NO_OP = "no_op" # Agent explicitly passes this step
# ---------------------------------------------------------------------------
# API Spec domain models (not OpenEnv base classes)
# ---------------------------------------------------------------------------
class FieldSpec(dict):
"""A JSON Schema-like field definition. Stored as plain dict for flexibility."""
pass
class EndpointSpec(dict):
"""A single endpoint definition: method, path, request_body, response."""
pass
# ---------------------------------------------------------------------------
# OpenEnv Action
# ---------------------------------------------------------------------------
class DebugAction(Action):
"""
A single fix proposed by the agent.
The agent targets one endpoint + one field and proposes exactly one change.
"""
kind: ActionKind = Field(
...,
description="The type of fix being applied",
)
endpoint_index: int = Field(
...,
ge=0,
description="0-based index into the endpoint list",
)
location: str = Field(
...,
description=(
"Where in the endpoint to apply the fix. "
"One of: 'request_body', 'response_body', 'status_code'"
),
)
field_name: Optional[str] = Field(
default=None,
description="Field name to add/remove/change (null for status_code fixes)",
)
new_value: Optional[Any] = Field(
default=None,
description=(
"The corrected value. "
"For CHANGE_TYPE: a FieldType string. "
"For ADD_FIELD: a dict with 'type' (and optional 'description'). "
"For CHANGE_STATUS: an integer HTTP status code. "
"For REMOVE_FIELD / NO_OP: null."
),
)
# ---------------------------------------------------------------------------
# OpenEnv Observation
# ---------------------------------------------------------------------------
class Violation(dict):
"""
Describes a single detected contract violation.
Keys: endpoint_index, location, field_name, violation_type, description
"""
pass
class DebugObservation(Observation):
"""
What the agent sees after each reset() / step().
"""
task_name: str = Field(
...,
description="Which task is currently active (easy / medium / hard)",
)
task_description: str = Field(
...,
description="Human-readable description of the task objective",
)
endpoints: List[Dict[str, Any]] = Field(
...,
description="Current (potentially partially-fixed) endpoint specs",
)
violations: List[Dict[str, Any]] = Field(
default_factory=list,
description="List of detected violations still present in the spec",
)
violations_fixed_this_step: int = Field(
default=0,
description="How many violations the last action resolved",
)
violations_introduced_this_step: int = Field(
default=0,
description="How many new violations the last action introduced",
)
total_violations_at_start: int = Field(
...,
description="Number of violations at episode start (for progress tracking)",
)
step_count: int = Field(
default=0,
description="Steps taken so far in this episode",
)
max_steps: int = Field(
default=10,
description="Maximum steps allowed per episode",
)
last_action_error: Optional[str] = Field(
default=None,
description="Error message if the last action was malformed / out-of-range",
)
# ---------------------------------------------------------------------------
# OpenEnv State
# ---------------------------------------------------------------------------
class DebugState(State):
"""
Full internal state of the environment (not exposed to the agent by default).
"""
task_name: str = Field(default="")
original_endpoints: List[Dict[str, Any]] = Field(default_factory=list)
current_endpoints: List[Dict[str, Any]] = Field(default_factory=list)
golden_endpoints: List[Dict[str, Any]] = Field(default_factory=list)
violations: List[Dict[str, Any]] = Field(default_factory=list)
total_violations_at_start: int = Field(default=0)
max_steps: int = Field(default=10)
|