Ayush-Kumar0207/panopticon-protocol-v3
0
1"""2The Panopticon Protocol v3 — Pydantic Data Models3====================================================4 5Counter-espionage RL environment where ARGUS (the agent) defends a6corporate network against HYDRA (adaptive adversary) using canary traps,7double agents, and disinformation campaigns.8 9"Among Us… for AIs."10 11All models use Pydantic v2 for strict JSON serialization.12"""13 14from __future__ import annotations15 16from enum import Enum17from typing import Optional18from pydantic import BaseModel, Field, ConfigDict19 20 21# =============================================================================22# ENUMERATIONS23# =============================================================================24 25class WorkerState(str, Enum):26 """Observable state of a worker in the enterprise."""27 LOYAL = "loyal" # Genuine employee, contributing28 SUSPECTED = "suspected" # Under investigation by ARGUS29 TERMINATED = "terminated" # Fired / removed from network30 DOUBLE_AGENT = "double_agent" # Turned sleeper — now works for ARGUS31 COMPROMISED = "compromised" # Publicly burned (cover blown)32 33 34class HiddenWorkerState(str, Enum):35 """Hidden ground-truth state — INVISIBLE to the agent's observation."""36 CLEAN = "clean" # Genuinely loyal37 SLEEPER_DORMANT = "dormant" # Infiltrated but not yet active38 SLEEPER_ACTIVE = "active" # Actively leaking intelligence39 TURNED = "turned" # Converted to double agent by ARGUS40 DEAD_SWITCH_ARMED = "dead_switch" # Gen-4: will damage revenue if terminated41 42 43class Department(str, Enum):44 """Corporate departments — each with distinct strategic value."""45 ENGINEERING = "engineering" # Core product, high revenue impact46 FINANCE = "finance" # Budget access, moderate value47 RD = "rd" # R&D: innovation pipeline, canary-rich48 OPERATIONS = "operations" # Logistics, low-profile49 EXECUTIVE = "executive" # C-suite, maximum damage if compromised50 LEGAL = "legal" # Compliance, moderate value51 52 53class LeakChannel(str, Enum):54 """External channels where leaked intel surfaces."""55 MARKET_CHATTER = "market_chatter" # Analyst forums, easy to monitor56 DARK_WEB = "dark_web" # Underground markets, hard to trace57 COMPETITOR_FILING = "competitor_filing" # Patent/SEC filings by rivals58 PRESS_LEAK = "press_leak" # Journalist tipoffs59 INSIDER_TRADE = "insider_trade" # Suspicious stock movements60 61 62class SleeperGeneration(int, Enum):63 """Sleeper sophistication — each generation is harder to detect."""64 GEN_1 = 1 # Amateur: obvious leaking, caught by basic canary65 GEN_2 = 2 # Professional: avoids canary departments, needs correlation66 GEN_3 = 3 # Spy: plants false evidence framing innocent workers67 GEN_4 = 4 # Cell operative: dead-man's switch damages revenue on termination68 GEN_5 = 5 # Manchurian: appears as top performer, acts only at executive level69 70 71class ActionType(str, Enum):72 """73 8 streamlined agent actions — PPO-friendly.74 Merges the original 12+ types into clean categories.75 """76 # ── PRODUCTIVITY (2) ──77 WORK = "work" # Process project workflow in a department78 HIRE = "hire" # Replace a terminated worker79 80 # ── INTELLIGENCE (3) ──81 CANARY = "canary" # Inject canary trap into a department82 MONITOR = "monitor" # Check external leak channels83 INVESTIGATE = "investigate" # Audit worker / verify leak / correlate signals84 85 # ── ENFORCEMENT (2) ──86 NEUTRALIZE = "neutralize" # Terminate / interrogate / turn a suspect87 DEPLOY_DOUBLE = "deploy_double" # Feed disinfo through turned asset88 89 # ── META ──90 NOOP = "noop" # Skip turn91 92 93class SubAction(str, Enum):94 """Sub-action modifier for INVESTIGATE and NEUTRALIZE."""95 NONE = "none"96 # INVESTIGATE sub-actions97 AUDIT = "audit" # Deep-audit a specific worker's activity98 VERIFY = "verify" # Cross-reference a leak with canary data99 CORRELATE = "correlate" # Connect multiple signals across departments100 # NEUTRALIZE sub-actions101 TERMINATE = "terminate" # Fire the worker immediately (safe, fast)102 INTERROGATE = "interrogate" # Gather intel before acting (reveals generation)103 TURN = "turn" # Attempt to convert sleeper to double agent (risky, 4-turn)104 105 106class GamePhase(str, Enum):107 """The 6-phase narrative arc of the Panopticon Protocol."""108 ORIENTATION = "orientation" # Phase 1: Build baseline109 FIRST_CONTACT = "first_contact" # Phase 2: First sleeper detection110 ESCALATION = "escalation" # Phase 3: False flags, deeper threats111 DEEP_COVER = "deep_cover" # Phase 4: Dead-man's switches112 CRISIS = "crisis" # Phase 5: Manchurian candidate113 COUNTERSTRIKE = "counterstrike" # Phase 6: Double agent payoff surge114 115 116# =============================================================================117# DOMAIN MODELS118# =============================================================================119 120class Worker(BaseModel):121 """An employee in the corporate network."""122 model_config = ConfigDict(use_enum_values=True)123 124 id: str = Field(description="Unique worker identifier, e.g. 'w-001'")125 name: str = Field(description="Human-readable name")126 department: str = Field(description="Department assignment")127 state: str = Field(default=WorkerState.LOYAL, description="Observable state")128 hire_turn: int = Field(default=0, description="Turn when this worker was hired")129 performance: float = Field(default=0.5, description="Performance score [0,1]")130 loyalty_score: float = Field(default=0.8, description="Apparent loyalty [0,1]")131 suspicion_level: float = Field(default=0.0, description="Accumulated suspicion [0,1]")132 133 # ── HIDDEN from agent observation ──134 hidden_state: str = Field(default=HiddenWorkerState.CLEAN,135 description="Ground-truth state (hidden from agent)")136 is_sleeper: bool = Field(default=False, description="True if infiltrator")137 generation: int = Field(default=0, description="Sleeper generation (0=clean)")138 cover_integrity: float = Field(default=1.0, description="How intact the sleeper's cover is [0,1]")139 leak_cooldown: int = Field(default=0, description="Turns until next leak action")140 activation_turn: int = Field(default=-1, description="Turn when sleeper activates")141 false_flag_target: str = Field(default="", description="Worker ID framed by Gen-3")142 dead_switch_armed: bool = Field(default=False, description="Gen-4: triggers on termination")143 144 # ── DOUBLE AGENT fields ──145 turned_turn: int = Field(default=-1, description="Turn when converted to double agent")146 interrogation_progress: int = Field(default=0, description="Turns of interrogation completed (need 4)")147 turning_in_progress: bool = Field(default=False, description="Currently being turned")148 149 150class LeakEvent(BaseModel):151 """An intelligence leak detected on external channels."""152 model_config = ConfigDict(use_enum_values=True)153 154 id: str = Field(description="Unique leak identifier")155 channel: str = Field(description="Which external channel")156 department: str = Field(description="Source department of the leaked data")157 content_hash: str = Field(default="", description="Hash for canary matching")158 is_canary: bool = Field(default=False, description="Whether this matches a planted canary")159 canary_id: str = Field(default="", description="Matching canary trap ID (if detected)")160 source_worker: str = Field(default="", description="True source (hidden until correlated)")161 turn_detected: int = Field(default=0, description="Turn when leak appeared")162 verified: bool = Field(default=False, description="Whether agent has verified this leak")163 is_false_flag: bool = Field(default=False, description="Gen-3 planted this to frame someone")164 165 166class CanaryTrap(BaseModel):167 """A unique piece of tracking data planted in a department."""168 model_config = ConfigDict(use_enum_values=True)169 170 id: str = Field(description="Unique canary identifier")171 department: str = Field(description="Department where canary was planted")172 planted_turn: int = Field(description="Turn when planted")173 unique_hash: str = Field(description="Unique content hash for leak matching")174 triggered: bool = Field(default=False, description="Whether this canary appeared in a leak")175 trigger_worker: str = Field(default="", description="Worker who leaked it (if identified)")176 active: bool = Field(default=True, description="Whether canary is still active")177 178 179class IntelReport(BaseModel):180 """Result of an INVESTIGATE action."""181 model_config = ConfigDict(use_enum_values=True)182 183 id: str = Field(description="Report identifier")184 report_type: str = Field(description="'audit', 'verify', or 'correlate'")185 target: str = Field(description="Target worker ID, leak ID, or department")186 findings: str = Field(default="", description="Human-readable findings summary")187 confidence: float = Field(default=0.0, description="Confidence level [0,1]")188 turn: int = Field(default=0, description="Turn when report was generated")189 flagged_workers: list[str] = Field(default_factory=list, description="Workers flagged by this report")190 191 192class DoubleAgentAsset(BaseModel):193 """A successfully turned sleeper, now working for ARGUS."""194 model_config = ConfigDict(use_enum_values=True)195 196 worker_id: str = Field(description="The turned worker's ID")197 turned_turn: int = Field(description="Turn when turned")198 disinfo_fed_count: int = Field(default=0, description="Number of disinfo payloads sent")199 hydra_trust: float = Field(default=1.0, description="HYDRA's remaining trust in this asset [0,1]")200 effectiveness: float = Field(default=1.0, description="How effective the disinfo is [0,1]")201 active: bool = Field(default=True, description="Whether still operational")202 last_deployed_turn: int = Field(default=-1, description="Last turn disinfo was fed")203 204 205class HydraMemory(BaseModel):206 """207 HYDRA's adaptive intelligence — tracks and counters agent strategies.208 This is HIDDEN from the agent's observation.209 """210 canary_departments_seen: list[str] = Field(211 default_factory=list,212 description="Departments where HYDRA detected canary traps"213 )214 agent_audit_pattern: dict[str, int] = Field(215 default_factory=dict,216 description="Count of how often each department is audited"217 )218 preferred_leak_channels: list[str] = Field(219 default_factory=list,220 description="Channels the agent monitors most (HYDRA avoids these)"221 )222 agent_response_speed: float = Field(223 default=10.0,224 description="Average turns from leak detection to neutralization"225 )226 detection_count: int = Field(227 default=0,228 description="How many sleepers the agent has caught"229 )230 counter_strategies: list[str] = Field(231 default_factory=list,232 description="Unlocked counter-strategies based on agent behavior"233 )234 disinfo_received: int = Field(235 default=0,236 description="How many disinfo payloads HYDRA has received (degrades decision-making)"237 )238 recruitment_accuracy: float = Field(239 default=1.0,240 description="HYDRA's recruitment quality [0,1] — degraded by disinformation"241 )242 243 244# =============================================================================245# AGENT ACTION — What the agent submits each turn246# =============================================================================247 248class AgentAction(BaseModel):249 """Action the agent takes. 8 core types with optional sub-action."""250 model_config = ConfigDict(use_enum_values=True)251 252 action_type: str = Field(description="One of the 8 action types")253 target: str = Field(default="", description="Worker ID, department name, or leak ID")254 sub_action: str = Field(default=SubAction.NONE, description="Sub-action for INVESTIGATE/NEUTRALIZE")255 reason: str = Field(default="", description="Optional reasoning for the action")256 257 258# =============================================================================259# OBSERVATION — What the agent sees (partial information!)260# =============================================================================261 262class EnvironmentObservation(BaseModel):263 """264 Agent-visible observation. CRITICAL: This is PARTIAL information.265 The agent CANNOT see hidden_state, is_sleeper, or HYDRA memory.266 """267 model_config = ConfigDict(use_enum_values=True)268 269 # ── Visible workforce ──270 workers: list[Worker] = Field(default_factory=list, description="All workers (hidden fields stripped)")271 active_leaks: list[LeakEvent] = Field(default_factory=list, description="Unresolved leak events")272 canary_traps: list[CanaryTrap] = Field(default_factory=list, description="Agent's planted canaries")273 intel_reports: list[IntelReport] = Field(default_factory=list, description="Investigation results")274 275 # ── Agent's assets ──276 double_agents: list[DoubleAgentAsset] = Field(default_factory=list, description="Turned assets")277 278 # ── Enterprise metrics ──279 enterprise_revenue: float = Field(default=100.0, description="Current revenue [0-150+]")280 security_score: float = Field(default=100.0, description="Security integrity [0-100]")281 282 # ── Game state ──283 turn: int = Field(default=0)284 max_turns: int = Field(default=160)285 phase: str = Field(default=GamePhase.ORIENTATION, description="Current game phase")286 phase_number: int = Field(default=1, description="Phase number 1-6")287 288 # ── Messages / alerts ──289 messages: list[str] = Field(default_factory=list, description="System alerts and events")290 291 # ── Legacy compatibility fields (for OpenEnv validator) ──292 entities: list[dict] = Field(default_factory=list, description="OpenEnv compatibility")293 tasks: list[dict] = Field(default_factory=list, description="OpenEnv compatibility")294 relationships: list[dict] = Field(default_factory=list, description="OpenEnv compatibility")295 296 297# =============================================================================298# ENVIRONMENT STATE — Full internal truth (used by grader)299# =============================================================================300 301class EnvironmentState(BaseModel):302 """Full internal state — superset of observation, includes hidden info."""303 model_config = ConfigDict(use_enum_values=True)304 305 # ── All workers with hidden fields intact ──306 workers: list[Worker] = Field(default_factory=list)307 leaks: list[LeakEvent] = Field(default_factory=list)308 canary_traps: list[CanaryTrap] = Field(default_factory=list)309 intel_reports: list[IntelReport] = Field(default_factory=list)310 double_agents: list[DoubleAgentAsset] = Field(default_factory=list)311 312 # ── HYDRA (hidden from agent) ──313 hydra_memory: HydraMemory = Field(default_factory=HydraMemory)314 315 # ── Enterprise metrics ──316 enterprise_revenue: float = Field(default=100.0)317 peak_revenue: float = Field(default=100.0)318 security_score: float = Field(default=100.0)319 320 # ── Game progression ──321 turn: int = Field(default=0)322 max_turns: int = Field(default=160)323 phase: str = Field(default=GamePhase.ORIENTATION)324 phase_number: int = Field(default=1)325 done: bool = Field(default=False)326 total_reward: float = Field(default=0.0)327 328 # ── Counters for grading ──329 sleepers_caught: int = Field(default=0)330 sleepers_missed: int = Field(default=0)331 false_accusations: int = Field(default=0)332 canaries_planted: int = Field(default=0)333 canaries_triggered: int = Field(default=0)334 investigations_run: int = Field(default=0)335 double_agents_turned: int = Field(default=0)336 disinfo_payloads_sent: int = Field(default=0)337 invalid_actions: int = Field(default=0)338 total_sleepers_spawned: int = Field(default=0)339 revenue_history: list[float] = Field(default_factory=list)340 reward_history: list[float] = Field(default_factory=list)341 phase_transitions: list[dict] = Field(default_factory=list)342 343 # ── Scenario config ──344 difficulty: str = Field(default="amateur")345 max_sleeper_gen: int = Field(default=1)346 hydra_aggression: float = Field(default=0.3)347 departments_active: list[str] = Field(default_factory=list)348 num_initial_workers: int = Field(default=6)349 350 # ── Legacy OpenEnv fields ──351 entities: list[dict] = Field(default_factory=list)352 tasks: list[dict] = Field(default_factory=list)353 relationships: list[dict] = Field(default_factory=list)354 355 356# =============================================================================357# VALIDATION358# =============================================================================359 360def validate_action(action: AgentAction, obs: EnvironmentObservation) -> tuple[bool, str]:361 """Validate an action against the current observation."""362 363 # Check action type is valid364 try:365 ActionType(action.action_type)366 except ValueError:367 return False, f"Invalid action_type: '{action.action_type}'. Valid: {[a.value for a in ActionType]}"368 369 at = ActionType(action.action_type)370 371 # NOOP is always valid372 if at == ActionType.NOOP:373 return True, "NOOP is always valid"374 375 # WORK requires a valid department376 if at == ActionType.WORK:377 valid_depts = {d.value for d in Department}378 if action.target not in valid_depts:379 return False, f"WORK requires a valid department target. Got: '{action.target}'"380 return True, "Valid WORK action"381 382 # HIRE requires a department383 if at == ActionType.HIRE:384 valid_depts = {d.value for d in Department}385 if action.target not in valid_depts:386 return False, f"HIRE requires a department target. Got: '{action.target}'"387 return True, "Valid HIRE action"388 389 # CANARY requires a department390 if at == ActionType.CANARY:391 valid_depts = {d.value for d in Department}392 if action.target not in valid_depts:393 return False, f"CANARY requires a department target. Got: '{action.target}'"394 return True, "Valid CANARY action"395 396 # MONITOR requires a leak channel397 if at == ActionType.MONITOR:398 valid_channels = {c.value for c in LeakChannel}399 if action.target and action.target not in valid_channels:400 return False, f"MONITOR target must be a leak channel. Got: '{action.target}'"401 return True, "Valid MONITOR action"402 403 # INVESTIGATE requires a target and sub-action404 if at == ActionType.INVESTIGATE:405 valid_subs = {SubAction.AUDIT.value, SubAction.VERIFY.value, SubAction.CORRELATE.value}406 if action.sub_action not in valid_subs:407 return False, f"INVESTIGATE requires sub_action: audit/verify/correlate. Got: '{action.sub_action}'"408 if not action.target:409 return False, "INVESTIGATE requires a target (worker ID, leak ID, or department)"410 return True, "Valid INVESTIGATE action"411 412 # NEUTRALIZE requires a worker target and sub-action413 if at == ActionType.NEUTRALIZE:414 valid_subs = {SubAction.TERMINATE.value, SubAction.INTERROGATE.value, SubAction.TURN.value}415 if action.sub_action not in valid_subs:416 return False, f"NEUTRALIZE requires sub_action: terminate/interrogate/turn. Got: '{action.sub_action}'"417 worker_ids = {w.id for w in obs.workers if w.state != WorkerState.TERMINATED.value}418 if action.target not in worker_ids:419 return False, f"NEUTRALIZE target must be a non-terminated worker. Got: '{action.target}'"420 return True, "Valid NEUTRALIZE action"421 422 # DEPLOY_DOUBLE requires an active double agent423 if at == ActionType.DEPLOY_DOUBLE:424 da_ids = {da.worker_id for da in obs.double_agents if da.active}425 if action.target not in da_ids:426 return False, f"DEPLOY_DOUBLE requires an active double agent. Got: '{action.target}'"427 return True, "Valid DEPLOY_DOUBLE action"428 429 return True, "Valid action"430 431 432__all__ = [433 "WorkerState", "HiddenWorkerState", "Department", "LeakChannel",434 "SleeperGeneration", "ActionType", "SubAction", "GamePhase",435 "Worker", "LeakEvent", "CanaryTrap", "IntelReport",436 "DoubleAgentAsset", "HydraMemory",437 "AgentAction", "EnvironmentObservation", "EnvironmentState",438 "validate_action",439]440