Read current phase state. Returns: Phase state dict, or None if file doesn't exist or is stale
(self)
| 90 | temp_file.replace(self.state_file) |
| 91 | |
| 92 | def get_phase(self) -> Optional[dict[str, Any]]: |
| 93 | """ |
| 94 | Read current phase state. |
| 95 | |
| 96 | Returns: |
| 97 | Phase state dict, or None if file doesn't exist or is stale |
| 98 | """ |
| 99 | if not self.state_file.exists(): |
| 100 | return None |
| 101 | |
| 102 | try: |
| 103 | with open(self.state_file, "r", encoding="utf-8") as f: |
| 104 | state = json.load(f) |
| 105 | |
| 106 | # Check if stale (process no longer exists) |
| 107 | if self._is_stale(state): |
| 108 | return None |
| 109 | |
| 110 | return state |
| 111 | except (json.JSONDecodeError, KeyError, OSError): |
| 112 | # Corrupted or unreadable state file |
| 113 | return None |
| 114 | |
| 115 | def clear(self) -> None: |
| 116 | """Remove phase state file (call on successful completion).""" |
no test coverage detected