kaizen-agentic/src/kaizen_agentic/schedule.py

"""Repo-local scheduled agent execution manifest (.kaizen/schedule.yml).

ADR-005 defines the schedule contract: which agents run on what cadence in an
opted-in repo. kaizen-agentic owns parsing, validation, and preparing an
orientation bundle for a scheduled run. It does **not** run cron schedules or
invoke Claude — activity-core fires the cron and a coding-agent session executes
the prepared bundle.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

import yaml

SCHEDULE_RELATIVE_PATH = Path(".kaizen") / "schedule.yml"
SCHEDULE_VERSION = "1"
VALID_CADENCES = ("daily", "weekly", "monthly")

# Sensible defaults for `schedule init` — coach + optimization weekly, the
# heavier tdd-workflow review monthly and disabled until an operator opts in.
DEFAULT_AGENTS: Dict[str, Dict[str, Any]] = {
    "coach": {"cadence": "weekly", "cron": "0 9 * * 1", "enabled": True},
    "optimization": {"cadence": "weekly", "cron": "0 10 * * 1", "enabled": True},
    "tdd-workflow": {"cadence": "monthly", "enabled": False},
}
DEFAULT_TIMEZONE = "Europe/Berlin"


class ScheduleError(Exception):
    """Raised when a schedule manifest cannot be parsed."""


@dataclass
class ScheduleEntry:
    """One scheduled agent run declaration."""

    agent: str
    cadence: str
    enabled: bool = True
    cron: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        data: Dict[str, Any] = {"cadence": self.cadence, "enabled": self.enabled}
        if self.cron:
            data["cron"] = self.cron
        return data


@dataclass
class Schedule:
    """Parsed `.kaizen/schedule.yml` manifest."""

    version: str
    timezone: Optional[str] = None
    entries: List[ScheduleEntry] = field(default_factory=list)
    source_path: Optional[Path] = None

    def entry_for(self, agent: str) -> Optional[ScheduleEntry]:
        for entry in self.entries:
            if entry.agent == agent:
                return entry
        return None

    def enabled_entries(self) -> List[ScheduleEntry]:
        return [e for e in self.entries if e.enabled]


def schedule_path(project_root: Path) -> Path:
    """Return the canonical schedule.yml path for a project root."""
    return Path(project_root) / SCHEDULE_RELATIVE_PATH


def parse_schedule(data: Any, source_path: Optional[Path] = None) -> Schedule:
    """Parse a raw mapping into a Schedule (structural errors raise).

    Semantic validation (known agents, cadence values) is handled by
    :func:`validate_schedule` so callers can collect actionable error lists.
    """
    if not isinstance(data, dict):
        raise ScheduleError("schedule.yml must be a YAML mapping at the top level")

    version = data.get("version")
    if version is None:
        raise ScheduleError("schedule.yml is missing required key: version")
    version = str(version)

    timezone = data.get("timezone")
    if timezone is not None and not isinstance(timezone, str):
        raise ScheduleError("timezone must be a string")

    agents = data.get("agents", {})
    if not isinstance(agents, dict):
        raise ScheduleError("agents must be a mapping of agent-name -> settings")

    entries: List[ScheduleEntry] = []
    for name, settings in agents.items():
        if settings is None:
            settings = {}
        if not isinstance(settings, dict):
            raise ScheduleError(f"agent '{name}' settings must be a mapping")
        cron = settings.get("cron")
        if cron is not None and not isinstance(cron, str):
            raise ScheduleError(f"agent '{name}' cron must be a string")
        entries.append(
            ScheduleEntry(
                agent=str(name),
                cadence=str(settings.get("cadence", "")),
                enabled=bool(settings.get("enabled", True)),
                cron=cron,
            )
        )

    return Schedule(
        version=version,
        timezone=timezone,
        entries=entries,
        source_path=source_path,
    )


def load_schedule(path: Path) -> Schedule:
    """Load and parse a schedule.yml file (raises ScheduleError)."""
    path = Path(path)
    if not path.exists():
        raise ScheduleError(f"schedule file not found: {path}")
    try:
        raw = yaml.safe_load(path.read_text(encoding="utf-8"))
    except yaml.YAMLError as exc:  # pragma: no cover - passthrough message
        raise ScheduleError(f"invalid YAML in {path}: {exc}") from exc
    return parse_schedule(raw, source_path=path)


def validate_schedule(
    schedule: Schedule, known_agents: Optional[List[str]] = None
) -> List[str]:
    """Return a list of human-readable validation errors (empty == valid)."""
    errors: List[str] = []

    if schedule.version != SCHEDULE_VERSION:
        errors.append(
            f"unsupported version '{schedule.version}' "
            f"(expected '{SCHEDULE_VERSION}')"
        )

    if not schedule.entries:
        errors.append("no agents declared under 'agents:'")

    seen: set = set()
    known = set(known_agents) if known_agents is not None else None
    for entry in schedule.entries:
        if entry.agent in seen:
            errors.append(f"duplicate agent entry: {entry.agent}")
        seen.add(entry.agent)

        if entry.cadence not in VALID_CADENCES:
            errors.append(
                f"agent '{entry.agent}': invalid cadence '{entry.cadence}' "
                f"(expected one of {', '.join(VALID_CADENCES)})"
            )

        if known is not None and entry.agent not in known:
            errors.append(
                f"agent '{entry.agent}' is not an installed or packaged agent"
            )

    return errors


def default_schedule_yaml(timezone: str = DEFAULT_TIMEZONE) -> str:
    """Render the default schedule.yml scaffold for `schedule init`."""
    document = {
        "version": SCHEDULE_VERSION,
        "timezone": timezone,
        "agents": dict(DEFAULT_AGENTS),
    }
    header = (
        "# Kaizen scheduled agent execution manifest (ADR-005)\n"
        "# Declares which agents run on what cadence in this repo.\n"
        "# Validate with: kaizen-agentic schedule validate\n"
    )
    body = yaml.safe_dump(document, sort_keys=False, default_flow_style=False)
    return header + body