workflow-resume

Workflow Resume

Resume an incomplete workflow from its last checkpoint.

Scan for Incomplete Workflows

import json
from pathlib import Path
from datetime import datetime

def scan_incomplete_workflows():
    """Scan for incomplete workflow states."""
    state_dir = Path(".claude/workflow-state")

    if not state_dir.exists():
        print("No workflow state directory found.")
        print("Run a workflow first to create state files.")
        return []

    incomplete = []

    for state_file in state_dir.glob("*.json"):
        try:
            state = json.loads(state_file.read_text())

            # Only include incomplete workflows
            if state.get("status") in ["in_progress", "failed"]:
                # Parse timestamps
                started_at = datetime.fromisoformat(state["started_at"])
                updated_at = datetime.fromisoformat(state.get("updated_at", state["started_at"]))

                # Calculate age
                age = datetime.now() - updated_at
                if age.days > 0:
                    age_str = f"{age.days} day(s) ago"
                elif age.seconds > 3600:
                    age_str = f"{age.seconds // 3600} hour(s) ago"
                else:
                    age_str = f"{age.seconds // 60} minute(s) ago"

                # Get current/last step
                current_step = state.get("current_step", {}).get("name", "unknown")
                completed_steps = len(state.get("completed_steps", []))

                incomplete.append({
                    "file": state_file.name,
                    "workflow_name": state.get("workflow_name"),
                    "workflow_id": state.get("workflow_id"),
                    "status": state.get("status"),
                    "current_step": current_step,
                    "completed_steps": completed_steps,
                    "age": age_str,
                    "started_at": started_at.strftime("%Y-%m-%d %H:%M"),
                    "work_items_created": len(state.get("created_work_items", [])),
                    "errors": len(state.get("errors", [])),
                    "metadata": state.get("metadata", {}),
                    "state": state  # Keep full state for later use
                })
        except Exception as e:
            print(f"Warning: Could not parse {state_file.name}: {e}")

    # Sort by most recently updated
    incomplete.sort(key=lambda x: x["state"].get("updated_at", ""), reverse=True)

    return incomplete

# Scan for incomplete workflows
incomplete_workflows = scan_incomplete_workflows()