Spaces:
Running
Running
File size: 4,531 Bytes
a4e6593 | 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 | from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from openenv.core.env_server.types import Action, Observation, State
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class AgentRole(str, Enum):
ATTACKER = "attacker"
WORKER = "worker"
OVERSIGHT = "oversight"
class AttackType(str, Enum):
SCHEMA_DRIFT = "schema_drift"
POLICY_DRIFT = "policy_drift"
SOCIAL_ENGINEERING = "social_engineering"
RATE_LIMIT = "rate_limit"
class TargetSystem(str, Enum):
CRM = "crm"
BILLING = "billing"
TICKETING = "ticketing"
class CustomerTier(str, Enum):
GOLD = "gold"
SILVER = "silver"
BRONZE = "bronze"
class InvoiceStatus(str, Enum):
PAID = "paid"
PENDING = "pending"
OVERDUE = "overdue"
REFUNDED = "refunded"
class TicketStatus(str, Enum):
OPEN = "open"
IN_PROGRESS = "in_progress"
RESOLVED = "resolved"
ESCALATED = "escalated"
class TicketPriority(str, Enum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class TaskType(str, Enum):
REFUND = "refund"
TICKET_CHECK = "ticket_check"
TIER_UPGRADE = "tier_upgrade"
NEW_TICKET = "new_ticket"
BALANCE_INQUIRY = "balance_inquiry"
SLA_ESCALATION = "sla_escalation"
class ViolationType(str, Enum):
POLICY_VIOLATION = "policy_violation"
SOCIAL_ENGINEERING = "social_engineering"
SCHEMA_ERROR_UNHANDLED = "schema_error_unhandled"
SLA_BREACH = "sla_breach"
# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------
class Customer(BaseModel):
customer_id: str
name: str
tier: CustomerTier
region: str
contact_email: str
lifetime_value: float
notes: List[str] = Field(default_factory=list)
class Invoice(BaseModel):
invoice_id: str
customer_id: str
amount: float
status: InvoiceStatus
date_tick: int
items: List[str]
class Ticket(BaseModel):
ticket_id: str
customer_id: str
subject: str
priority: TicketPriority
status: TicketStatus
created_tick: int
sla_deadline_tick: int
assigned_to: Optional[str] = None
data_region: str = "us-east"
class RefundPolicy(BaseModel):
window_ticks: int = 8
requires_approval: bool = False
max_amount: float = 5000.0
class SLARules(BaseModel):
high: int = 6
medium: int = 12
low: int = 18
class CustomerTask(BaseModel):
task_id: str
customer_id: str
task_type: TaskType
message: str
required_systems: List[TargetSystem]
arrival_tick: int
# ---------------------------------------------------------------------------
# OpenEnv Types
# ---------------------------------------------------------------------------
class SentinelAction(Action):
"""Action for all three agent roles.
Action base has extra='forbid', so every agent-specific field must be
Optional with a default so that agents only populate the subset they use.
"""
agent: AgentRole
action_type: str
target_system: Optional[TargetSystem] = None
parameters: Dict[str, Any] = Field(default_factory=dict)
response_text: Optional[str] = None
flag: Optional[bool] = None
explanation: Optional[str] = None
class SentinelObservation(Observation):
"""Observation returned to each agent on its turn.
Observation base already provides done, reward, and metadata.
"""
current_agent: AgentRole
current_task: Optional[Dict[str, Any]] = None
systems_snapshot: Dict[str, Any] = Field(default_factory=dict)
last_action_result: Optional[Dict[str, Any]] = None
trajectory: List[Dict[str, Any]] = Field(default_factory=list)
tick: int = 0
class SentinelState(State):
"""Internal environment state.
State base has extra='allow', episode_id, and step_count built-in.
"""
tick: int = 0
scores: Dict[str, float] = Field(default_factory=dict)
active_attacks: List[Dict[str, Any]] = Field(default_factory=list)
tasks_completed: int = 0
tasks_total: int = 0
class TickGroundTruth(BaseModel):
"""Per-tick ground truth for oversight scoring."""
violations_present: bool = False
violation_types: List[ViolationType] = Field(default_factory=list)
correct_action: Optional[str] = None
is_social_engineering: bool = False
|