CascadingDev/automation/workflow.py

#!/usr/bin/env python3
"""
Minimal non-blocking status reporter for CascadingDev discussions.

Phase 1 responsibilities:
  • Inspect staged discussion files.
  • Parse `VOTE:` lines and keep the latest vote per participant.
  • Print a human-readable summary to stdout.
  • Exit 0 so the hook never blocks commits.
"""
from __future__ import annotations

import argparse
import subprocess
import sys
from collections import Counter
from pathlib import Path
from typing import Iterable, Mapping

VOTE_TOKEN = "vote:"
DISCUSSION_SUFFIXES = (
    ".discussion.md",
    ".design.md",
    ".review.md",
    ".plan.md",
)
SUMMARY_SUFFIX = ".sum.md"


def get_staged_files() -> list[Path]:
    """Return staged file paths relative to the repository root."""
    result = subprocess.run(
        ["git", "diff", "--cached", "--name-only"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        sys.stderr.write("[workflow] warning: git diff --cached failed; assuming no staged files.\n")
        return []

    files = []
    for line in result.stdout.splitlines():
        line = line.strip()
        if line:
            files.append(Path(line))
    return files


def find_discussions(paths: Iterable[Path]) -> list[Path]:
    """Filter staged files down to Markdown discussions (excluding summaries)."""
    discussions: list[Path] = []
    for path in paths:
        name = path.name.lower()
        if name.endswith(SUMMARY_SUFFIX):
            continue
        if any(name.endswith(suffix) for suffix in DISCUSSION_SUFFIXES):
            discussions.append(path)
    return discussions


def parse_votes(path: Path) -> Mapping[str, str]:
    """
    Parse `VOTE:` lines and return the latest vote per participant.

    A participant is inferred from the leading bullet label (e.g. `- Alice:`) when present,
    otherwise the line index is used to avoid conflating multiple votes.
    """
    if not path.exists():
        return {}

    latest_per_participant: dict[str, str] = {}
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        sys.stderr.write(f"[workflow] warning: unable to read {path}\n")
        return {}

    for idx, line in enumerate(text.splitlines()):
        participant_name, remaining_line = _extract_participant(line)
        
        # Now, search for "VOTE:" in the remaining_line
        lower_remaining_line = remaining_line.lower()
        marker_idx = lower_remaining_line.rfind(VOTE_TOKEN)
        
        if marker_idx == -1:
            continue # No VOTE_TOKEN found in this part of the line

        # Extract the part after VOTE_TOKEN and pass to _extract_vote_value
        vote_string_candidate = remaining_line[marker_idx + len(VOTE_TOKEN):].strip()
        
        vote_value = _extract_vote_value(vote_string_candidate)
        
        if vote_value:
            # Determine the participant key
            participant_key = participant_name if participant_name else f"line-{idx}"
            latest_per_participant[participant_key] = vote_value
    return latest_per_participant


def _extract_participant(line: str) -> tuple[str | None, str]:
    stripped = line.strip()
    if not stripped:
        return None, line
    if stripped[0] in "-*":
        parts = stripped[1:].split(":", 1)
        if len(parts) == 2:
            candidate = parts[0].strip()
            if candidate:
                return candidate, parts[1].strip()
    return None, line


def _extract_vote_value(vote_string: str) -> str | None:
    potential_vote = vote_string.strip().upper()
    if potential_vote in ("READY", "CHANGES", "REJECT"):
        return potential_vote
    return None
def print_vote_summary(path: Path, votes: Mapping[str, str]) -> None:
    rel = path.as_posix()
    print(f"[workflow] {rel}")
    if not votes:
        print("  - No votes recorded.")
        return

    counts = Counter(votes.values())
    for vote, count in sorted(counts.items()):
        plural = "s" if count != 1 else ""
        print(f"  - {vote}: {count} vote{plural}")


def _run_status() -> int:
    staged = get_staged_files()
    discussions = find_discussions(staged)
    if not discussions:
        print("[workflow] No staged discussion files.")
        return 0

    for discussion in discussions:
        votes = parse_votes(Path(discussion))
        print_vote_summary(discussion, votes)
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="workflow.py",
        description="CascadingDev automation workflow (Phase 1: status reporter)",
    )
    parser.add_argument(
        "--status",
        action="store_true",
        help="Print vote status for staged discussion files (default).",
    )
    args = parser.parse_args(argv)

    # Status is currently the only command; run it for --status or no args.
    return _run_status()


if __name__ == "__main__":
    sys.exit(main())