Do

{
  "description": "do loop hooks for 5-phase workflow",
  "hooks": {
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "python3 ${CLAUDE_PLUGIN_ROOT}/hooks/stop-hook.py"
          }
        ]
      }
    ],
    "SubagentStop": [
      {
        "matcher": "code-reviewer",
        "hooks": [
          {
            "type": "command",
            "command": "python3 ${CLAUDE_PLUGIN_ROOT}/hooks/verify-loop.py"
          }
        ]
      }
    ]
  }
}

#!/usr/bin/env python3
"""
Stop hook for do skill workflow.

Checks if the do loop is complete before allowing exit.
Uses the new task directory structure under .claude/do-tasks/.
"""

import glob
import json
import os
import sys

DIR_TASKS = ".claude/do-tasks"
FILE_CURRENT_TASK = ".current-task"
FILE_TASK_JSON = "task.json"

PHASE_NAMES = {
    1: "Understand",
    2: "Clarify",
    3: "Design",
    4: "Implement",
    5: "Complete",
}


def phase_name_for(n: int) -> str:
    return PHASE_NAMES.get(n, f"Phase {n}")


def get_current_task(project_dir: str) -> str | None:
    """Read current task directory path."""
    current_task_file = os.path.join(project_dir, DIR_TASKS, FILE_CURRENT_TASK)
    if not os.path.exists(current_task_file):
        return None
    try:
        with open(current_task_file, "r", encoding="utf-8") as f:
            content = f.read().strip()
            return content if content else None
    except Exception:
        return None


def get_task_info(project_dir: str, task_dir: str) -> dict | None:
    """Read task.json data."""
    task_json_path = os.path.join(project_dir, task_dir, FILE_TASK_JSON)
    if not os.path.exists(task_json_path):
        return None
    try:
        with open(task_json_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def check_task_complete(project_dir: str, task_dir: str) -> str:
    """Check if task is complete. Returns blocking reason or empty string."""
    task_info = get_task_info(project_dir, task_dir)
    if not task_info:
        return ""

    status = task_info.get("status", "")
    if status == "completed":
        return ""

    current_phase = task_info.get("current_phase", 1)
    max_phases = task_info.get("max_phases", 5)
    phase_name = task_info.get("phase_name", phase_name_for(current_phase))
    completion_promise = task_info.get("completion_promise", "<promise>DO_COMPLETE</promise>")

    if current_phase >= max_phases:
        # Task is at final phase, allow exit
        return ""

    return (
        f"do loop incomplete: current phase {current_phase}/{max_phases} ({phase_name}). "
        f"Continue with remaining phases; use 'task.py update-phase <N>' after each phase. "
        f"Include completion_promise in final output when done: {completion_promise}. "
        f"To exit early, set status to 'completed' in task.json."
    )


def main():
    project_dir = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())

    task_dir = get_current_task(project_dir)
    if not task_dir:
        # No active task, allow exit
        sys.exit(0)

    stdin_payload = ""
    if not sys.stdin.isatty():
        try:
            stdin_payload = sys.stdin.read()
        except Exception:
            pass

    reason = check_task_complete(project_dir, task_dir)
    if not reason:
        sys.exit(0)

    print(json.dumps({"decision": "block", "reason": reason}))
    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Verify Loop Hook for do skill workflow.

SubagentStop hook that intercepts when code-reviewer agent tries to stop.
Runs verification commands to ensure code quality before allowing exit.

Mechanism:
- Intercepts SubagentStop event for code-reviewer agent
- Runs verify commands from task.json if configured
- Blocks stopping until verification passes
- Has max iterations as safety limit (MAX_ITERATIONS=5)

State file: .claude/do-tasks/.verify-state.json
"""

import json
import os
import subprocess
import sys
from datetime import datetime
from pathlib import Path

# Configuration
MAX_ITERATIONS = 5
STATE_TIMEOUT_MINUTES = 30
DIR_TASKS = ".claude/do-tasks"
FILE_CURRENT_TASK = ".current-task"
FILE_TASK_JSON = "task.json"
STATE_FILE = ".claude/do-tasks/.verify-state.json"

# Only control loop for code-reviewer agent
TARGET_AGENTS = {"code-reviewer"}


def get_project_root(cwd: str) -> str | None:
    """Find project root (directory with .claude folder)."""
    current = Path(cwd).resolve()
    while current != current.parent:
        if (current / ".claude").exists():
            return str(current)
        current = current.parent
    return None


def get_current_task(project_root: str) -> str | None:
    """Read current task directory path."""
    current_task_file = os.path.join(project_root, DIR_TASKS, FILE_CURRENT_TASK)
    if not os.path.exists(current_task_file):
        return None
    try:
        with open(current_task_file, "r", encoding="utf-8") as f:
            content = f.read().strip()
            return content if content else None
    except Exception:
        return None


def get_task_info(project_root: str, task_dir: str) -> dict | None:
    """Read task.json data."""
    task_json_path = os.path.join(project_root, task_dir, FILE_TASK_JSON)
    if not os.path.exists(task_json_path):
        return None
    try:
        with open(task_json_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def get_verify_commands(task_info: dict) -> list[str]:
    """Get verify commands from task.json."""
    return task_info.get("verify_commands", [])


def run_verify_commands(project_root: str, commands: list[str]) -> tuple[bool, str]:
    """Run verify commands and return (success, message)."""
    for cmd in commands:
        try:
            result = subprocess.run(
                cmd,
                shell=True,
                cwd=project_root,
                capture_output=True,
                timeout=120,
            )
            if result.returncode != 0:
                stderr = result.stderr.decode("utf-8", errors="replace")
                stdout = result.stdout.decode("utf-8", errors="replace")
                error_output = stderr or stdout
                if len(error_output) > 500:
                    error_output = error_output[:500] + "..."
                return False, f"Command failed: {cmd}\n{error_output}"
        except subprocess.TimeoutExpired:
            return False, f"Command timed out: {cmd}"
        except Exception as e:
            return False, f"Command error: {cmd} - {str(e)}"
    return True, "All verify commands passed"


def load_state(project_root: str) -> dict:
    """Load verify loop state."""
    state_path = os.path.join(project_root, STATE_FILE)
    if not os.path.exists(state_path):
        return {"task": None, "iteration": 0, "started_at": None}
    try:
        with open(state_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {"task": None, "iteration": 0, "started_at": None}


def save_state(project_root: str, state: dict) -> None:
    """Save verify loop state."""
    state_path = os.path.join(project_root, STATE_FILE)
    try:
        os.makedirs(os.path.dirname(state_path), exist_ok=True)
        with open(state_path, "w", encoding="utf-8") as f:
            json.dump(state, f, indent=2, ensure_ascii=False)
    except Exception:
        pass


def main():
    try:
        input_data = json.load(sys.stdin)
    except json.JSONDecodeError:
        sys.exit(0)

    hook_event = input_data.get("hook_event_name", "")
    if hook_event != "SubagentStop":
        sys.exit(0)

    subagent_type = input_data.get("subagent_type", "")
    agent_output = input_data.get("agent_output", "")
    cwd = input_data.get("cwd", os.getcwd())

    if subagent_type not in TARGET_AGENTS:
        sys.exit(0)

    project_root = get_project_root(cwd)
    if not project_root:
        sys.exit(0)

    task_dir = get_current_task(project_root)
    if not task_dir:
        sys.exit(0)

    task_info = get_task_info(project_root, task_dir)
    if not task_info:
        sys.exit(0)

    verify_commands = get_verify_commands(task_info)
    if not verify_commands:
        # No verify commands configured, allow exit
        sys.exit(0)

    # Load state
    state = load_state(project_root)

    # Reset state if task changed or too old
    should_reset = False
    if state.get("task") != task_dir:
        should_reset = True
    elif state.get("started_at"):
        try:
            started = datetime.fromisoformat(state["started_at"])
            if (datetime.now() - started).total_seconds() > STATE_TIMEOUT_MINUTES * 60:
                should_reset = True
        except (ValueError, TypeError):
            should_reset = True

    if should_reset:
        state = {
            "task": task_dir,
            "iteration": 0,
            "started_at": datetime.now().isoformat(),
        }

    # Increment iteration
    state["iteration"] = state.get("iteration", 0) + 1
    current_iteration = state["iteration"]
    save_state(project_root, state)

    # Safety check: max iterations
    if current_iteration >= MAX_ITERATIONS:
        state["iteration"] = 0
        save_state(project_root, state)
        output = {
            "decision": "allow",
            "reason": f"Max iterations ({MAX_ITERATIONS}) reached. Stopping to prevent infinite loop.",
        }
        print(json.dumps(output, ensure_ascii=False))
        sys.exit(0)

    # Run verify commands
    passed, message = run_verify_commands(project_root, verify_commands)

    if passed:
        state["iteration"] = 0
        save_state(project_root, state)
        output = {
            "decision": "allow",
            "reason": "All verify commands passed. Review phase complete.",
        }
        print(json.dumps(output, ensure_ascii=False))
        sys.exit(0)
    else:
        output = {
            "decision": "block",
            "reason": f"Iteration {current_iteration}/{MAX_ITERATIONS}. Verification failed:\n{message}\n\nPlease fix the issues and try again.",
        }
        print(json.dumps(output, ensure_ascii=False))
        sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Install/uninstall do skill to ~/.claude/skills/do"""
import argparse
import json
import os
import shutil
import sys
from pathlib import Path

SKILL_NAME = "do"
HOOK_PATH = "~/.claude/skills/do/hooks/stop-hook.py"

MODELS_JSON_TEMPLATE = {
    "agents": {
        "code-explorer": {
            "backend": "claude",
            "model": "claude-sonnet-4-5-20250929"
        },
        "code-architect": {
            "backend": "claude",
            "model": "claude-sonnet-4-5-20250929"
        },
        "code-reviewer": {
            "backend": "claude",
            "model": "claude-sonnet-4-5-20250929"
        }
    }
}

def get_settings_path() -> Path:
    return Path.home() / ".claude" / "settings.json"

def load_settings() -> dict:
    path = get_settings_path()
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

def save_settings(settings: dict):
    path = get_settings_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(settings, f, indent=2)

def add_hook(settings: dict) -> dict:
    hook_command = str(Path(HOOK_PATH).expanduser())
    hook_entry = {
        "type": "command",
        "command": f"python3 {hook_command}"
    }

    if "hooks" not in settings:
        settings["hooks"] = {}
    if "Stop" not in settings["hooks"]:
        settings["hooks"]["Stop"] = []

    stop_hooks = settings["hooks"]["Stop"]

    for item in stop_hooks:
        if "hooks" in item:
            for h in item["hooks"]:
                if "stop-hook" in h.get("command", "") and "do" in h.get("command", ""):
                    h["command"] = f"python3 {hook_command}"
                    return settings

    stop_hooks.append({"hooks": [hook_entry]})
    return settings

def remove_hook(settings: dict) -> dict:
    if "hooks" not in settings or "Stop" not in settings["hooks"]:
        return settings

    stop_hooks = settings["hooks"]["Stop"]
    new_stop_hooks = []

    for item in stop_hooks:
        if "hooks" in item:
            filtered = [h for h in item["hooks"]
                       if "stop-hook" not in h.get("command", "")
                       or "do" not in h.get("command", "")]
            if filtered:
                item["hooks"] = filtered
                new_stop_hooks.append(item)
        else:
            new_stop_hooks.append(item)

    settings["hooks"]["Stop"] = new_stop_hooks
    if not settings["hooks"]["Stop"]:
        del settings["hooks"]["Stop"]
    if not settings["hooks"]:
        del settings["hooks"]

    return settings

def install_models_json():
    """Install ~/.codeagent/models.json if not exists"""
    path = Path.home() / ".codeagent" / "models.json"
    if path.exists():
        print(f"⚠ {path} already exists, skipping")
        return

    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(MODELS_JSON_TEMPLATE, f, indent=2)
    print(f"✓ Created {path}")

def install():
    src = Path(__file__).parent.resolve()
    dest = Path.home() / ".claude" / "skills" / SKILL_NAME

    dest.mkdir(parents=True, exist_ok=True)

    exclude = {".git", "__pycache__", ".DS_Store", "install.py"}

    for item in src.iterdir():
        if item.name in exclude:
            continue
        target = dest / item.name
        if target.exists():
            if target.is_dir():
                shutil.rmtree(target)
            else:
                target.unlink()
        if item.is_dir():
            shutil.copytree(item, target)
        else:
            shutil.copy2(item, target)

    settings = load_settings()
    settings = add_hook(settings)
    save_settings(settings)

    install_models_json()

    print(f"✓ Installed to {dest}")
    print(f"✓ Hook added to settings.json")

def uninstall():
    dest = Path.home() / ".claude" / "skills" / SKILL_NAME

    settings = load_settings()
    settings = remove_hook(settings)
    save_settings(settings)
    print(f"✓ Hook removed from settings.json")

    if dest.exists():
        shutil.rmtree(dest)
        print(f"✓ Removed {dest}")
    else:
        print(f"⚠ {dest} not found")

def main():
    parser = argparse.ArgumentParser(description="Install/uninstall do skill")
    parser.add_argument("--uninstall", "-u", action="store_true", help="Uninstall the skill")
    args = parser.parse_args()

    if args.uninstall:
        uninstall()
    else:
        install()

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Get context for current task.

Reads the current task's jsonl files and returns context for specified agent.
Used by inject-context hook to build agent prompts.
"""

import json
import os
import sys
from pathlib import Path

DIR_TASKS = ".claude/do-tasks"
FILE_CURRENT_TASK = ".current-task"
FILE_TASK_JSON = "task.json"


def get_project_root() -> str:
    return os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())


def get_current_task(project_root: str) -> str | None:
    current_task_file = os.path.join(project_root, DIR_TASKS, FILE_CURRENT_TASK)
    if not os.path.exists(current_task_file):
        return None
    try:
        with open(current_task_file, "r", encoding="utf-8") as f:
            content = f.read().strip()
            return content if content else None
    except Exception:
        return None


def read_file_content(base_path: str, file_path: str) -> str | None:
    full_path = os.path.join(base_path, file_path)
    if os.path.exists(full_path) and os.path.isfile(full_path):
        try:
            with open(full_path, "r", encoding="utf-8") as f:
                return f.read()
        except Exception:
            return None
    return None


def read_jsonl_entries(base_path: str, jsonl_path: str) -> list[tuple[str, str]]:
    full_path = os.path.join(base_path, jsonl_path)
    if not os.path.exists(full_path):
        return []

    results = []
    try:
        with open(full_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    item = json.loads(line)
                    file_path = item.get("file") or item.get("path")
                    if not file_path:
                        continue
                    content = read_file_content(base_path, file_path)
                    if content:
                        results.append((file_path, content))
                except json.JSONDecodeError:
                    continue
    except Exception:
        pass
    return results


def get_agent_context(project_root: str, task_dir: str, agent_type: str) -> str:
    """Get complete context for specified agent."""
    context_parts = []

    # Read agent-specific jsonl
    agent_jsonl = os.path.join(task_dir, f"{agent_type}.jsonl")
    agent_entries = read_jsonl_entries(project_root, agent_jsonl)

    for file_path, content in agent_entries:
        context_parts.append(f"=== {file_path} ===\n{content}")

    # Read prd.md
    prd_content = read_file_content(project_root, os.path.join(task_dir, "prd.md"))
    if prd_content:
        context_parts.append(f"=== {task_dir}/prd.md (Requirements) ===\n{prd_content}")

    return "\n\n".join(context_parts)


def get_task_info(project_root: str, task_dir: str) -> dict | None:
    """Get task.json data."""
    task_json_path = os.path.join(project_root, task_dir, FILE_TASK_JSON)
    if not os.path.exists(task_json_path):
        return None
    try:
        with open(task_json_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def main():
    import argparse

    parser = argparse.ArgumentParser(description="Get context for current task")
    parser.add_argument("agent", nargs="?", choices=["implement", "check", "debug"],
                        help="Agent type (optional, returns task info if not specified)")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    project_root = get_project_root()
    task_dir = get_current_task(project_root)

    if not task_dir:
        if args.json:
            print(json.dumps({"error": "No active task"}))
        else:
            print("No active task.", file=sys.stderr)
        sys.exit(1)

    task_info = get_task_info(project_root, task_dir)

    if not args.agent:
        if args.json:
            print(json.dumps({"task_dir": task_dir, "task_info": task_info}))
        else:
            print(f"Task: {task_dir}")
            if task_info:
                print(f"Title: {task_info.get('title', 'N/A')}")
                print(f"Phase: {task_info.get('current_phase', '?')}/{task_info.get('max_phases', 5)}")
        sys.exit(0)

    context = get_agent_context(project_root, task_dir, args.agent)

    if args.json:
        print(json.dumps({
            "task_dir": task_dir,
            "agent": args.agent,
            "context": context,
            "task_info": task_info,
        }))
    else:
        print(context)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Initialize do skill workflow - wrapper around task.py.

Creates a task directory under .claude/do-tasks/ with:
- task.md: Task metadata (YAML frontmatter) + requirements (Markdown body)

If --worktree is specified, also creates a git worktree for isolated development.
"""

import argparse
import sys

from task import create_task, PHASE_NAMES


def die(msg: str):
    print(f"Error: {msg}", file=sys.stderr)
    sys.exit(1)


def main():
    parser = argparse.ArgumentParser(
        description="Initialize do skill workflow with task directory"
    )
    parser.add_argument("--max-phases", type=int, default=5, help="Default: 5")
    parser.add_argument(
        "--completion-promise",
        default="<promise>DO_COMPLETE</promise>",
        help="Default: <promise>DO_COMPLETE</promise>",
    )
    parser.add_argument("--worktree", action="store_true", help="Enable worktree mode")
    parser.add_argument("prompt", nargs="+", help="Task description")
    args = parser.parse_args()

    if args.max_phases < 1:
        die("--max-phases must be a positive integer")

    prompt = " ".join(args.prompt)
    result = create_task(title=prompt, use_worktree=args.worktree)

    task_data = result["task_data"]
    worktree_dir = result.get("worktree_dir", "")

    print(f"Initialized: {result['relative_path']}")
    print(f"task_id: {task_data['id']}")
    print(f"phase: 1/{task_data['max_phases']} ({PHASE_NAMES[1]})")
    print(f"completion_promise: {task_data['completion_promise']}")
    print(f"use_worktree: {task_data['use_worktree']}")
    print(f"export DO_TASK_DIR={result['relative_path']}")

    if worktree_dir:
        print(f"worktree_dir: {worktree_dir}")
        print(f"export DO_WORKTREE_DIR={worktree_dir}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Task Directory Management CLI for do skill workflow.

Commands:
  create <title>              - Create a new task directory with task.md
  start <task-dir>            - Set current task pointer
  finish                      - Clear current task pointer
  list                        - List active tasks
  status                      - Show current task status
  update-phase <N>            - Update current phase
"""

import argparse
import os
import random
import re
import string
import subprocess
import sys
from datetime import datetime
from pathlib import Path

# Directory constants
DIR_TASKS = ".claude/do-tasks"
FILE_CURRENT_TASK = ".current-task"
FILE_TASK_MD = "task.md"

PHASE_NAMES = {
    1: "Understand",
    2: "Clarify",
    3: "Design",
    4: "Implement",
    5: "Complete",
}


def get_project_root() -> str:
    """Get project root from env or cwd."""
    return os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())


def get_tasks_dir(project_root: str) -> str:
    """Get tasks directory path."""
    return os.path.join(project_root, DIR_TASKS)


def get_current_task_file(project_root: str) -> str:
    """Get current task pointer file path."""
    return os.path.join(project_root, DIR_TASKS, FILE_CURRENT_TASK)


def generate_task_id() -> str:
    """Generate short task ID: MMDD-XXXX format."""
    date_part = datetime.now().strftime("%m%d")
    random_part = ''.join(random.choices(string.ascii_lowercase + string.digits, k=4))
    return f"{date_part}-{random_part}"


def read_task_md(task_md_path: str) -> dict | None:
    """Read task.md and parse YAML frontmatter + body."""
    if not os.path.exists(task_md_path):
        return None

    try:
        with open(task_md_path, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception:
        return None

    # Parse YAML frontmatter
    match = re.match(r'^---\n(.*?)\n---\n(.*)$', content, re.DOTALL)
    if not match:
        return None

    frontmatter_str = match.group(1)
    body = match.group(2)

    # Simple YAML parsing (no external deps)
    frontmatter = {}
    for line in frontmatter_str.split('\n'):
        if ':' in line:
            key, value = line.split(':', 1)
            key = key.strip()
            value = value.strip()
            # Handle quoted strings
            if value.startswith('"') and value.endswith('"'):
                value = value[1:-1]
            elif value == 'true':
                value = True
            elif value == 'false':
                value = False
            elif value.isdigit():
                value = int(value)
            frontmatter[key] = value

    return {"frontmatter": frontmatter, "body": body}


def write_task_md(task_md_path: str, frontmatter: dict, body: str) -> bool:
    """Write task.md with YAML frontmatter + body."""
    try:
        lines = ["---"]
        for key, value in frontmatter.items():
            if isinstance(value, bool):
                lines.append(f"{key}: {str(value).lower()}")
            elif isinstance(value, int):
                lines.append(f"{key}: {value}")
            elif isinstance(value, str) and ('<' in value or '>' in value or ':' in value):
                lines.append(f'{key}: "{value}"')
            else:
                lines.append(f'{key}: "{value}"' if isinstance(value, str) else f"{key}: {value}")
        lines.append("---")
        lines.append("")
        lines.append(body)

        with open(task_md_path, "w", encoding="utf-8") as f:
            f.write('\n'.join(lines))
        return True
    except Exception:
        return False


def create_worktree(project_root: str, task_id: str) -> str:
    """Create a git worktree for the task. Returns the worktree directory path."""
    # Get git root
    result = subprocess.run(
        ["git", "-C", project_root, "rev-parse", "--show-toplevel"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Not a git repository: {project_root}")
    git_root = result.stdout.strip()

    # Calculate paths
    worktree_dir = os.path.join(git_root, ".worktrees", f"do-{task_id}")
    branch_name = f"do/{task_id}"

    # Create worktree with new branch
    result = subprocess.run(
        ["git", "-C", git_root, "worktree", "add", "-b", branch_name, worktree_dir],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Failed to create worktree: {result.stderr}")

    return worktree_dir


def create_task(title: str, use_worktree: bool = False) -> dict:
    """Create a new task directory with task.md."""
    project_root = get_project_root()
    tasks_dir = get_tasks_dir(project_root)
    os.makedirs(tasks_dir, exist_ok=True)

    task_id = generate_task_id()
    task_dir = os.path.join(tasks_dir, task_id)

    os.makedirs(task_dir, exist_ok=True)

    # Create worktree if requested
    worktree_dir = ""
    if use_worktree:
        try:
            worktree_dir = create_worktree(project_root, task_id)
        except RuntimeError as e:
            print(f"Warning: {e}", file=sys.stderr)
            use_worktree = False

    frontmatter = {
        "id": task_id,
        "title": title,
        "status": "in_progress",
        "current_phase": 1,
        "phase_name": PHASE_NAMES[1],
        "max_phases": 5,
        "use_worktree": use_worktree,
        "worktree_dir": worktree_dir,
        "created_at": datetime.now().isoformat(),
        "completion_promise": "<promise>DO_COMPLETE</promise>",
    }

    body = f"""# Requirements

{title}

## Context

## Progress
"""

    task_md_path = os.path.join(task_dir, FILE_TASK_MD)
    write_task_md(task_md_path, frontmatter, body)

    current_task_file = get_current_task_file(project_root)
    relative_task_dir = os.path.relpath(task_dir, project_root)
    with open(current_task_file, "w", encoding="utf-8") as f:
        f.write(relative_task_dir)

    return {
        "task_dir": task_dir,
        "relative_path": relative_task_dir,
        "task_data": frontmatter,
        "worktree_dir": worktree_dir,
    }


def get_current_task(project_root: str) -> str | None:
    """Read current task directory path."""
    current_task_file = get_current_task_file(project_root)
    if not os.path.exists(current_task_file):
        return None

    try:
        with open(current_task_file, "r", encoding="utf-8") as f:
            content = f.read().strip()
            return content if content else None
    except Exception:
        return None


def start_task(task_dir: str) -> bool:
    """Set current task pointer."""
    project_root = get_project_root()
    tasks_dir = get_tasks_dir(project_root)

    if os.path.isabs(task_dir):
        full_path = task_dir
        relative_path = os.path.relpath(task_dir, project_root)
    else:
        if not task_dir.startswith(DIR_TASKS):
            full_path = os.path.join(tasks_dir, task_dir)
            relative_path = os.path.join(DIR_TASKS, task_dir)
        else:
            full_path = os.path.join(project_root, task_dir)
            relative_path = task_dir

    if not os.path.exists(full_path):
        print(f"Error: Task directory not found: {full_path}", file=sys.stderr)
        return False

    current_task_file = get_current_task_file(project_root)
    os.makedirs(os.path.dirname(current_task_file), exist_ok=True)

    with open(current_task_file, "w", encoding="utf-8") as f:
        f.write(relative_path)

    return True


def finish_task() -> bool:
    """Clear current task pointer."""
    project_root = get_project_root()
    current_task_file = get_current_task_file(project_root)

    if os.path.exists(current_task_file):
        os.remove(current_task_file)

    return True


def list_tasks() -> list[dict]:
    """List all task directories."""
    project_root = get_project_root()
    tasks_dir = get_tasks_dir(project_root)

    if not os.path.exists(tasks_dir):
        return []

    tasks = []
    current_task = get_current_task(project_root)

    for entry in sorted(os.listdir(tasks_dir), reverse=True):
        entry_path = os.path.join(tasks_dir, entry)
        if not os.path.isdir(entry_path):
            continue

        task_md_path = os.path.join(entry_path, FILE_TASK_MD)
        if not os.path.exists(task_md_path):
            continue

        parsed = read_task_md(task_md_path)
        if parsed:
            task_data = parsed["frontmatter"]
        else:
            task_data = {"id": entry, "title": entry, "status": "unknown"}

        relative_path = os.path.join(DIR_TASKS, entry)
        task_data["path"] = relative_path
        task_data["is_current"] = current_task == relative_path
        tasks.append(task_data)

    return tasks


def get_status() -> dict | None:
    """Get current task status."""
    project_root = get_project_root()
    current_task = get_current_task(project_root)

    if not current_task:
        return None

    task_dir = os.path.join(project_root, current_task)
    task_md_path = os.path.join(task_dir, FILE_TASK_MD)

    parsed = read_task_md(task_md_path)
    if not parsed:
        return None

    task_data = parsed["frontmatter"]
    task_data["path"] = current_task
    return task_data


def update_phase(phase: int) -> bool:
    """Update current task phase."""
    project_root = get_project_root()
    current_task = get_current_task(project_root)

    if not current_task:
        print("Error: No active task.", file=sys.stderr)
        return False

    task_dir = os.path.join(project_root, current_task)
    task_md_path = os.path.join(task_dir, FILE_TASK_MD)

    parsed = read_task_md(task_md_path)
    if not parsed:
        print("Error: task.md not found or invalid.", file=sys.stderr)
        return False

    frontmatter = parsed["frontmatter"]
    frontmatter["current_phase"] = phase
    frontmatter["phase_name"] = PHASE_NAMES.get(phase, f"Phase {phase}")

    if not write_task_md(task_md_path, frontmatter, parsed["body"]):
        print("Error: Failed to write task.md.", file=sys.stderr)
        return False

    return True


def main():
    parser = argparse.ArgumentParser(
        description="Task directory management for do skill workflow"
    )
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    # create command
    create_parser = subparsers.add_parser("create", help="Create a new task")
    create_parser.add_argument("title", nargs="+", help="Task title")
    create_parser.add_argument("--worktree", action="store_true", help="Enable worktree mode")

    # start command
    start_parser = subparsers.add_parser("start", help="Set current task")
    start_parser.add_argument("task_dir", help="Task directory path")

    # finish command
    subparsers.add_parser("finish", help="Clear current task")

    # list command
    subparsers.add_parser("list", help="List all tasks")

    # status command
    subparsers.add_parser("status", help="Show current task status")

    # update-phase command
    phase_parser = subparsers.add_parser("update-phase", help="Update current phase")
    phase_parser.add_argument("phase", type=int, help="Phase number (1-5)")

    args = parser.parse_args()

    if args.command == "create":
        title = " ".join(args.title)
        result = create_task(title, args.worktree)
        print(f"Created task: {result['relative_path']}")
        print(f"Task ID: {result['task_data']['id']}")
        print(f"Phase: 1/{result['task_data']['max_phases']} (Understand)")
        print(f"Worktree: {result['task_data']['use_worktree']}")

    elif args.command == "start":
        if start_task(args.task_dir):
            print(f"Started task: {args.task_dir}")
        else:
            sys.exit(1)

    elif args.command == "finish":
        if finish_task():
            print("Task finished, current task cleared.")
        else:
            sys.exit(1)

    elif args.command == "list":
        tasks = list_tasks()
        if not tasks:
            print("No tasks found.")
        else:
            for task in tasks:
                marker = "* " if task.get("is_current") else "  "
                phase = task.get("current_phase", "?")
                max_phase = task.get("max_phases", 5)
                status = task.get("status", "unknown")
                print(f"{marker}{task['id']} [{status}] phase {phase}/{max_phase}")
                print(f"    {task.get('title', 'No title')}")

    elif args.command == "status":
        status = get_status()
        if not status:
            print("No active task.")
        else:
            print(f"Task: {status['id']}")
            print(f"Title: {status.get('title', 'No title')}")
            print(f"Status: {status.get('status', 'unknown')}")
            print(f"Phase: {status.get('current_phase', '?')}/{status.get('max_phases', 5)}")
            print(f"Worktree: {status.get('use_worktree', False)}")
            print(f"Path: {status['path']}")

    elif args.command == "update-phase":
        if update_phase(args.phase):
            phase_name = PHASE_NAMES.get(args.phase, f"Phase {args.phase}")
            print(f"Updated to phase {args.phase} ({phase_name})")
        else:
            sys.exit(1)

    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()