tegwick
93bf49479b
Some checks failed
ci / test (push) Has been cancelled
Add --engagement, --agents, and --bootstrap-cadence flags to scaffold hourly/daily/weekly engagement schedules. Hourly bootstrap keeps cadence: daily with hourly cron overrides per coulomb-loop ADR-003. Document activity-core requirements in activity-core-handoff-engagement.md. Closes KAIZEN-WP-0008 T02 and T04.
244 lines
8.2 KiB
Python
244 lines
8.2 KiB
Python
"""CLI + module tests for scheduled agent execution (ADR-005, WP-0006)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from click.testing import CliRunner
|
|
|
|
from kaizen_agentic.cli import cli
|
|
from kaizen_agentic.schedule import (
|
|
ScheduleError,
|
|
engagement_schedule_yaml,
|
|
parse_schedule,
|
|
schedule_path,
|
|
validate_schedule,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def runner() -> CliRunner:
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def project_dir(tmp_path: Path) -> Path:
|
|
root = tmp_path / "demo-project"
|
|
root.mkdir()
|
|
return root
|
|
|
|
|
|
class TestScheduleModule:
|
|
def test_engagement_schedule_yaml_hourly_preset(self):
|
|
text = engagement_schedule_yaml(
|
|
"coulomb-loop",
|
|
agents=["coach", "optimization"],
|
|
bootstrap_cadence="hourly",
|
|
)
|
|
assert "Engagement: coulomb-loop bootstrap" in text
|
|
body = "\n".join(line for line in text.splitlines() if not line.startswith("#"))
|
|
schedule = parse_schedule(yaml.safe_load(body))
|
|
coach = schedule.entry_for("coach")
|
|
assert coach is not None
|
|
assert coach.cadence == "daily"
|
|
assert coach.cron == "15 * * * *"
|
|
|
|
def test_parse_requires_version(self):
|
|
with pytest.raises(ScheduleError):
|
|
parse_schedule({"agents": {}})
|
|
|
|
def test_parse_rejects_non_mapping(self):
|
|
with pytest.raises(ScheduleError):
|
|
parse_schedule(["not", "a", "mapping"])
|
|
|
|
def test_validate_flags_unknown_agent_and_bad_cadence(self):
|
|
schedule = parse_schedule(
|
|
{
|
|
"version": "1",
|
|
"agents": {
|
|
"coach": {"cadence": "weekly", "enabled": True},
|
|
"made-up": {"cadence": "hourly"},
|
|
},
|
|
}
|
|
)
|
|
errors = validate_schedule(schedule, known_agents=["coach", "optimization"])
|
|
assert any("hourly" in e for e in errors)
|
|
assert any("made-up" in e for e in errors)
|
|
|
|
def test_validate_clean_schedule(self):
|
|
schedule = parse_schedule(
|
|
{"version": "1", "agents": {"coach": {"cadence": "weekly"}}}
|
|
)
|
|
assert validate_schedule(schedule, known_agents=["coach"]) == []
|
|
|
|
|
|
class TestScheduleCli:
|
|
def test_init_creates_default_schedule(self, runner: CliRunner, project_dir: Path):
|
|
result = runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
assert result.exit_code == 0
|
|
path = schedule_path(project_dir)
|
|
assert path.exists()
|
|
assert "coach" in path.read_text()
|
|
|
|
def test_engagement_init_hourly_bootstrap(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"schedule",
|
|
"init",
|
|
"--target",
|
|
str(project_dir),
|
|
"--engagement",
|
|
"coulomb-loop",
|
|
"--agents",
|
|
"coach,optimization",
|
|
"--bootstrap-cadence",
|
|
"hourly",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
text = schedule_path(project_dir).read_text()
|
|
assert "Engagement: coulomb-loop bootstrap" in text
|
|
assert "cron: 15 * * * *" in text
|
|
assert "cadence: daily" in text
|
|
assert "Engagement: coulomb-loop" in result.output
|
|
|
|
def test_engagement_init_validates_unknown_agent(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"schedule",
|
|
"init",
|
|
"--target",
|
|
str(project_dir),
|
|
"--engagement",
|
|
"demo",
|
|
"--agents",
|
|
"not-a-real-agent",
|
|
],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "unknown agent" in result.output
|
|
|
|
def test_engagement_flags_require_engagement_slug(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"schedule",
|
|
"init",
|
|
"--target",
|
|
str(project_dir),
|
|
"--agents",
|
|
"coach",
|
|
],
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "--engagement" in result.output
|
|
|
|
def test_init_no_overwrite_without_force(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
result = runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
assert result.exit_code == 0
|
|
assert "already exists" in result.output
|
|
|
|
def test_validate_passes_on_default(self, runner: CliRunner, project_dir: Path):
|
|
runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
result = runner.invoke(
|
|
cli, ["schedule", "validate", "--target", str(project_dir)]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "valid" in result.output
|
|
|
|
def test_validate_missing_file_errors(self, runner: CliRunner, project_dir: Path):
|
|
result = runner.invoke(
|
|
cli, ["schedule", "validate", "--target", str(project_dir)]
|
|
)
|
|
assert result.exit_code == 1
|
|
|
|
def test_validate_rejects_bad_schema(self, runner: CliRunner, project_dir: Path):
|
|
path = schedule_path(project_dir)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("version: '1'\nagents:\n not-an-agent:\n cadence: weekly\n")
|
|
result = runner.invoke(
|
|
cli, ["schedule", "validate", "--target", str(project_dir)]
|
|
)
|
|
assert result.exit_code == 1
|
|
assert "not-an-agent" in result.output
|
|
|
|
def test_list_shows_enabled(self, runner: CliRunner, project_dir: Path):
|
|
runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
result = runner.invoke(cli, ["schedule", "list", "--target", str(project_dir)])
|
|
assert result.exit_code == 0
|
|
assert "coach" in result.output
|
|
# tdd-workflow is disabled by default; hidden without --all
|
|
assert "tdd-workflow" not in result.output
|
|
|
|
def test_list_all_shows_disabled(self, runner: CliRunner, project_dir: Path):
|
|
runner.invoke(cli, ["schedule", "init", "--target", str(project_dir)])
|
|
result = runner.invoke(
|
|
cli, ["schedule", "list", "--all", "--target", str(project_dir)]
|
|
)
|
|
assert "tdd-workflow" in result.output
|
|
|
|
def test_prepare_markdown_includes_agent_prompt(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
result = runner.invoke(
|
|
cli, ["schedule", "prepare", "coach", "--target", str(project_dir)]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "Scheduled Run Orientation: coach" in result.output
|
|
assert "## Agent Prompt" in result.output
|
|
assert "Coach Agent" in result.output
|
|
assert "## Session Close" in result.output
|
|
|
|
def test_prepare_json_format(self, runner: CliRunner, project_dir: Path):
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"schedule",
|
|
"prepare",
|
|
"coach",
|
|
"--target",
|
|
str(project_dir),
|
|
"--format",
|
|
"json",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.output)
|
|
assert payload["agent"] == "coach"
|
|
assert payload["agent_prompt_found"] is True
|
|
assert payload["session_close"]
|
|
|
|
def test_prepare_unknown_agent_notes_missing(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
result = runner.invoke(
|
|
cli,
|
|
["schedule", "prepare", "no-such-agent", "--target", str(project_dir)],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "not found in registry" in result.output
|
|
|
|
def test_prepare_includes_memory_when_present(
|
|
self, runner: CliRunner, project_dir: Path
|
|
):
|
|
runner.invoke(cli, ["memory", "init", "coach", "--target", str(project_dir)])
|
|
result = runner.invoke(
|
|
cli, ["schedule", "prepare", "coach", "--target", str(project_dir)]
|
|
)
|
|
assert "## Project Memory" in result.output
|
|
assert "Project Context" in result.output
|