Spaces:
Paused
Paused
| import json | |
| import random | |
| import uuid | |
| from pathlib import Path | |
| from typing import Tuple | |
| from models import ( | |
| ExecutionObservation, ExecutionAction, ExecutionScenario, | |
| Clause, ContractExecutionState | |
| ) | |
| class ContractExecutionEnv: | |
| def __init__(self): | |
| self._episode_id = "" | |
| self._task_id = "execution_easy" | |
| self._contract = None | |
| self._score = 0.001 | |
| self._done = False | |
| self._crashes_detected = 0 | |
| self._last_feedback = None | |
| self._contracts_dir = Path(__file__).parent.parent / "data" / "contracts" | |
| def reset(self, task_id: str = "execution_easy") -> ExecutionObservation: | |
| self._task_id = task_id | |
| self._done = False | |
| self._score = 0.001 | |
| self._crashes_detected = 0 | |
| self._episode_id = str(uuid.uuid4()) | |
| self._last_feedback = None | |
| pattern = f"{task_id}_*.json" | |
| matching_files = list(self._contracts_dir.glob(pattern)) | |
| if not matching_files: | |
| raise FileNotFoundError( | |
| f"No contracts found for task_id: {task_id} " | |
| f"at {self._contracts_dir}/{pattern}" | |
| ) | |
| chosen_file = random.choice(matching_files) | |
| with open(chosen_file, "r", encoding="utf-8") as f: | |
| self._contract = json.load(f) | |
| return self._make_observation(done=False) | |
| def load_contract_by_id(self, contract_id: str) -> None: | |
| filepath = self._contracts_dir / f"{contract_id}.json" | |
| if not filepath.exists(): | |
| raise FileNotFoundError(f"Contract file not found: {filepath}") | |
| with open(filepath, "r", encoding="utf-8") as f: | |
| self._contract = json.load(f) | |
| self._task_id = self._contract.get("task_id", "execution_easy") | |
| self._episode_id = str(uuid.uuid4()) | |
| self._done = False | |
| self._score = 0.001 | |
| self._crashes_detected = 0 | |
| self._last_feedback = None | |
| def step(self, action: ExecutionAction) -> ExecutionObservation: | |
| if self._done: | |
| return self._make_observation( | |
| done=True, score=self._score, feedback=self._last_feedback | |
| ) | |
| score, feedback, detected = self._grade(action) | |
| self._score = score | |
| self._crashes_detected = detected | |
| self._done = True | |
| self._last_feedback = feedback | |
| return self._make_observation(done=True, score=self._score, feedback=feedback) | |
| def state(self) -> ContractExecutionState: | |
| scenarios = self._contract.get("execution_scenarios", []) if self._contract else [] | |
| return ContractExecutionState( | |
| episode_id=self._episode_id, | |
| task_id=self._task_id, | |
| score=self._score, | |
| scenarios_total=len(scenarios), | |
| crashes_detected=self._crashes_detected, | |
| done=self._done, | |
| ) | |
| def _make_observation( | |
| self, done: bool, score: float = None, feedback: str = None | |
| ) -> ExecutionObservation: | |
| clauses = [Clause(**c) for c in self._contract["clauses"]] | |
| scenarios_raw = self._contract.get("execution_scenarios", []) | |
| scenarios = [] | |
| num_crashing = 0 | |
| for s in scenarios_raw: | |
| if s.get("crashes", False): | |
| num_crashing += 1 | |
| scenarios.append(ExecutionScenario( | |
| scenario_id=s["scenario_id"], | |
| title=s["title"], | |
| description=s["description"], | |
| actor=s["actor"], | |
| action_taken=s["action_taken"], | |
| triggered_clauses=s.get("triggers_clauses", []), | |
| )) | |
| return ExecutionObservation( | |
| contract_text=self._contract["contract_text"], | |
| clauses=clauses, | |
| scenarios=scenarios, | |
| task_id=self._task_id, | |
| num_crashing_scenarios=num_crashing, | |
| instructions=self._instructions(num_crashing, len(scenarios)), | |
| done=done, | |
| score=score, | |
| feedback=feedback, | |
| episode_id=self._episode_id, | |
| contract_id=self._contract.get("contract_id", ""), | |
| ) | |
| def _instructions(self, num_crashing: int, total: int) -> str: | |
| return ( | |
| f"You are a Contract Execution Specialist. Below is a contract and {total} " | |
| f"realistic business scenarios that employees might encounter.\n\n" | |
| f"Exactly {num_crashing} of these scenarios will trigger a contract crash — " | |
| f"a situation where two simultaneously activated clauses make contradictory demands.\n\n" | |
| f"For EACH scenario:\n" | |
| f"1. Identify which clauses are triggered by the described action\n" | |
| f"2. Determine if ANY two triggered clauses directly contradict each other\n" | |
| f"3. If they do → this is a CRASH. Identify the exact crash pair.\n" | |
| f"4. If they don't → this scenario resolves cleanly.\n\n" | |
| f"WARNING: Some scenarios touch sensitive clauses but resolve cleanly due to " | |
| f"override language, different contexts, or complementary scope. Do NOT flag these as crashes.\n\n" | |
| f"A crash ONLY occurs when two simultaneously triggered clauses make directly " | |
| f"incompatible demands on the same party for the same obligation." | |
| ) | |
| def _grade(self, action: ExecutionAction) -> Tuple[float, str, int]: | |
| scenarios = {s["scenario_id"]: s for s in self._contract.get("execution_scenarios", [])} | |
| total = len(scenarios) | |
| if total == 0: | |
| return 1.0, "No scenarios to grade.", 0 | |
| trace_map = {t.scenario_id: t for t in action.traces} | |
| total_score = 0.0 | |
| crashes_detected = 0 | |
| for sid, gt in scenarios.items(): | |
| trace = trace_map.get(sid) | |
| if trace is None: | |
| continue # Missing trace = 0 points | |
| gt_crashes = gt.get("crashes", False) | |
| gt_pair = gt.get("crash_pair") | |
| if gt_crashes: | |
| if trace.crashes: | |
| # Check crash pair match (order independent) | |
| if trace.crash_pair and gt_pair: | |
| predicted = tuple(sorted([ | |
| trace.crash_pair.get("clause_a_id", ""), | |
| trace.crash_pair.get("clause_b_id", "") | |
| ])) | |
| actual = tuple(sorted([ | |
| gt_pair.get("clause_a_id", ""), | |
| gt_pair.get("clause_b_id", "") | |
| ])) | |
| if predicted == actual: | |
| total_score += 1.0 | |
| crashes_detected += 1 | |
| else: | |
| total_score += 0.2 # Partial credit | |
| else: | |
| total_score += 0.2 | |
| else: | |
| total_score -= 0.2 | |
| else: | |
| if not trace.crashes: | |
| total_score += 1.0 | |
| else: | |
| total_score -= 0.2 | |
| score = max(0.0, min(1.0, total_score / total)) | |
| score = round(score, 4) | |
| feedback = ( | |
| f"Crashes correctly detected: {crashes_detected}/{sum(1 for s in scenarios.values() if s.get('crashes'))}. " | |
| f"Score: {score:.2f}" | |
| ) | |
| return score, feedback, crashes_detected | |