69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""
|
|
AutoDev - Resume Manager
|
|
Handles state persistence and resuming from worklog.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from .logger import Logger
|
|
from . import config
|
|
|
|
|
|
class ResumeManager:
|
|
def __init__(self, logger: Logger, workdir: str):
|
|
self.logger = logger
|
|
self.workdir = workdir
|
|
self.state_path = os.path.join(workdir, ".autodev_state.json")
|
|
|
|
def save_state(self, current_step: int, plan: dict, status: str = "in_progress",
|
|
desc_hash: str = ""):
|
|
state = {
|
|
"current_step": current_step,
|
|
"total_steps": len(plan.get("steps", [])),
|
|
"status": status,
|
|
"plan_hash": hash(json.dumps(plan, sort_keys=True)),
|
|
"desc_hash": desc_hash,
|
|
}
|
|
with open(self.state_path, "w") as f:
|
|
json.dump(state, f, indent=2)
|
|
|
|
def load_state(self) -> dict | None:
|
|
if not os.path.exists(self.state_path):
|
|
return None
|
|
try:
|
|
with open(self.state_path, "r") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
return None
|
|
|
|
def get_resume_step(self) -> int:
|
|
"""Determine which step to resume from based on worklog."""
|
|
state = self.load_state()
|
|
if state and state.get("status") == "in_progress":
|
|
step = state.get("current_step", 0)
|
|
self.logger.log("resume", f"Resuming from step {step + 1}")
|
|
return step
|
|
return 0
|
|
|
|
def mark_complete(self, plan: dict, desc_hash: str = ""):
|
|
self.save_state(len(plan.get("steps", [])), plan, status="complete",
|
|
desc_hash=desc_hash)
|
|
self.logger.log("phase_complete", "all")
|
|
|
|
def mark_failed(self, step: int, plan: dict, reason: str, desc_hash: str = ""):
|
|
self.save_state(step, plan, status="failed", desc_hash=desc_hash)
|
|
self.logger.log("failed", f"Step {step + 1}: {reason}", "error")
|
|
|
|
def description_changed(self, desc_hash: str) -> bool:
|
|
"""Check if description.txt has changed since last run."""
|
|
state = self.load_state()
|
|
if not state:
|
|
return False
|
|
old_hash = state.get("desc_hash", "")
|
|
if not old_hash:
|
|
return False # No hash stored — can't tell, assume unchanged
|
|
return old_hash != desc_hash
|
|
|
|
def is_fresh_start(self) -> bool:
|
|
return not os.path.exists(self.state_path)
|