Spaces:
Sleeping
Sleeping
File size: 3,477 Bytes
ee933ab | 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 | """V3 Episode State — enhanced for unified agent architecture.
Changes from V2:
- Removed hypothesis_stack (no hypothesis gating)
- Removed judge/teacher fields
- Added deep flakiness signals cache
- Added patch history tracking
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
try:
from models import RunRecord, PatchRecord, RewardBreakdown
except ImportError:
from ..models import RunRecord, PatchRecord, RewardBreakdown
class EpisodeState(BaseModel):
"""V3 server-side episode state."""
# Identity
episode_id: str
test_identifier: str
repo_path: str
# Step tracking
step_count: int = 0
max_steps: int = 8
done: bool = False
# Code snapshots
original_test_source: str = ""
original_source_under_test: str = ""
current_test_source: str = ""
current_source_under_test: str = ""
# Run data
run_history: List[RunRecord] = Field(default_factory=list)
baseline_pass_rate: float = 0.0
current_pass_rate: float = 0.0
baseline_entropy: float = 0.0
env_type: str = "unknown"
should_train: bool = True
preflight_result: Dict[str, Any] = Field(default_factory=dict)
# Patch tracking
patches_applied: List[PatchRecord] = Field(default_factory=list)
total_diff_lines: int = 0
# Deep flakiness signals (cached from initial detection)
module_cache_violations: List[str] = Field(default_factory=list)
fixture_scope_risks: List[str] = Field(default_factory=list)
mock_residue_sites: List[str] = Field(default_factory=list)
import_side_effect_files: List[str] = Field(default_factory=list)
async_contamination_alive: bool = False
# Causal frontier
failure_frontier: str = ""
call_chain_to_frontier: List[str] = Field(default_factory=list)
boundary_crossings: List[str] = Field(default_factory=list)
# iDFlakies signals
order_dependency_detected: bool = False
infrastructure_sensitive: bool = False
# Stack trace
failing_stack_trace: str = ""
last_error_type: Optional[str] = None
# Causal graph
causal_graph: Optional[Dict[str, Any]] = None
causal_hints: List[str] = Field(default_factory=list)
# Last action tracking (for multi-step episodes)
last_think_text: str = ""
last_patch_text: str = ""
last_reward: float = 0.0
last_reward_breakdown: Dict[str, float] = Field(default_factory=dict)
last_patch_result: Dict[str, Any] = Field(default_factory=dict)
last_done_reason: str = ""
# Per-step think summaries — powers diversity penalty + hypothesis trail prompt.
# Each entry: {step, categories, entities, reason_signatures, oracle_score,
# pass_rate_after, reward}
step_think_history: List[Dict[str, Any]] = Field(default_factory=list)
# File tree
file_tree: List[str] = Field(default_factory=list)
# Regression tracking
regression_detected: bool = False
@property
def steps_remaining(self) -> int:
return max(0, self.max_steps - self.step_count)
@property
def is_terminal(self) -> bool:
return self.done or self.step_count >= self.max_steps
@property
def pass_rate_delta(self) -> float:
return round(self.current_pass_rate - self.baseline_pass_rate, 4)
|