Load the orchestration state, or None if it does not exist.
()
| 160 | |
| 161 | |
| 162 | def load_state() -> dict | None: |
| 163 | """Load the orchestration state, or None if it does not exist.""" |
| 164 | if not STATE_PATH.exists(): |
| 165 | return None |
| 166 | try: |
| 167 | with open(STATE_PATH, "r", encoding="utf-8") as f: |
| 168 | data = json.load(f) |
| 169 | # Minimal validation |
| 170 | if not isinstance(data, dict) or "kernels" not in data: |
| 171 | raise ValueError("State file missing 'kernels' key") |
| 172 | return data |
| 173 | except (json.JSONDecodeError, ValueError, OSError) as exc: |
| 174 | print(f"WARNING: Orchestration state corrupted ({exc}). Re-initializing.") |
| 175 | return None |
| 176 | |
| 177 | |
| 178 | def save_state(state: dict) -> None: |
no outgoing calls
no test coverage detected