Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion act_operator/act_operator/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
__version__ = "0.5.4"
__version__ = "0.6.0"
151 changes: 150 additions & 1 deletion act_operator/act_operator/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@

from __future__ import annotations

import json
import shutil
import tempfile
from pathlib import Path

import typer
Expand All @@ -11,6 +13,7 @@

from .utils import (
CASTS_DIR,
ENCODING_UTF8,
LANGGRAPH_FILE,
PYPROJECT_FILE,
Language,
Expand All @@ -28,7 +31,6 @@
BASE_NODE_FILE = "base_node.py"
BASE_GRAPH_FILE = "base_graph.py"
DEFAULT_LANGUAGE_CHOICE = 1
ENCODING_UTF8 = "utf-8"

console = Console()
app = typer.Typer(help="Act Operator", invoke_without_command=True)
Expand Down Expand Up @@ -733,6 +735,153 @@ def cast_command(
_generate_cast_project(act_path=act_path, cast_name=cast_raw, language=lang)


def _detect_project_language(act_path: Path) -> str:
"""Detect project language by checking README.md for Korean characters.

Args:
act_path: Path to the Act project.

Returns:
Language code string ("en" or "kr").
"""
readme_path = act_path / "README.md"
if not readme_path.exists():
return Language.ENGLISH.value

try:
content = readme_path.read_text(encoding=ENCODING_UTF8)
if any("\uac00" <= ch <= "\ud7af" for ch in content):
return Language.KOREAN.value
except OSError:
pass

return Language.ENGLISH.value


def _detect_first_cast(act_path: Path) -> NameVariants:
"""Detect the first Cast from langgraph.json graphs registry.

Args:
act_path: Path to the Act project.

Returns:
NameVariants for the first registered Cast.
"""
fallback = build_name_variants("default cast")

langgraph_path = act_path / LANGGRAPH_FILE
if not langgraph_path.exists():
return fallback

try:
content = langgraph_path.read_text(encoding=ENCODING_UTF8)
payload = json.loads(content)
graphs = payload.get("graphs", {})
if not graphs:
return fallback

first_slug = next(iter(graphs))
# Derive cast_snake from graph path: "./casts/cast_snake/graph.py:cast_snake_graph"
graph_ref = graphs[first_slug]
# Extract the snake name from the path segment between casts/ and /graph.py
parts = graph_ref.split("/")
for i, part in enumerate(parts):
if part == "casts" and i + 1 < len(parts):
cast_snake = parts[i + 1]
raw_name = cast_snake.replace("_", " ")
return build_name_variants(raw_name)

return fallback
except (OSError, json.JSONDecodeError, ValueError, StopIteration):
return fallback


def _upgrade_skills(
scaffold_root: Path, act_path: Path, context: dict[str, str]
) -> int:
"""Render latest skills from scaffold and replace project skills.

Args:
scaffold_root: Path to the scaffold template directory.
act_path: Path to the Act project.
context: Cookiecutter context variables.

Returns:
Number of skill directories upgraded.

Raises:
typer.Exit: If rendering or file operations fail.
"""
with tempfile.TemporaryDirectory(prefix="act_upgrade_") as tmp_dir:
tmp_path = Path(tmp_dir) / "rendered"
tmp_path.mkdir()

try:
render_cookiecutter_template(scaffold_root, tmp_path, context)
except Exception as error:
console.print(f"[red]Failed to render scaffold: {error}[/red]")
raise typer.Exit(code=EXIT_CODE_ERROR) from error

rendered_skills = tmp_path / ".claude" / "skills"
if not rendered_skills.exists():
console.print("[red]No skills found in scaffold template.[/red]")
raise typer.Exit(code=EXIT_CODE_ERROR)

target_claude = act_path / ".claude"
target_skills = target_claude / "skills"
backup_skills = target_claude / "skills.bak"

# Clean up previous backup
if backup_skills.exists():
shutil.rmtree(backup_skills)

# Backup existing skills
if target_skills.exists():
shutil.move(str(target_skills), str(backup_skills))
console.print("[dim]Backed up existing skills to .claude/skills.bak/[/dim]")

# Copy rendered skills
target_claude.mkdir(parents=True, exist_ok=True)
shutil.copytree(str(rendered_skills), str(target_skills))

skill_dirs = [d for d in rendered_skills.iterdir() if d.is_dir()]
return len(skill_dirs)


@app.command("upgrade")
def upgrade_command(
act_path: Path = CAST_ACT_PATH_OPTION,
) -> None:
"""Upgrade .claude/skills/ in an existing Act project to the latest version.

Args:
act_path: Path to the existing Act project.
"""
act_path = act_path.resolve()
_ensure_act_project(act_path)

# Detect project settings
language = _detect_project_language(act_path)
act = build_name_variants(act_path.name)
cast = _detect_first_cast(act_path)

console.print("[bold]Upgrading .claude/skills/ ...[/bold]")

scaffold_root = _get_scaffold_root()
context = _build_template_context(act, cast, language)

skill_count = _upgrade_skills(scaffold_root, act_path, context)

lang_display = Language.from_string(language).display_name
table = Table(show_header=False)
table.add_row("Act", act.title)
table.add_row("Language", lang_display)
table.add_row("Skills", str(skill_count))
table.add_row("Location", str(act_path / ".claude" / "skills"))
console.print(table)
console.print("[bold green]Skills upgraded successfully![/bold green]")


def main() -> None:
"""Entry point for the Act Operator CLI."""
app()
59 changes: 59 additions & 0 deletions act_operator/act_operator/tests/integration/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,65 @@ def test_init_derives_act_name_from_path(tmp_path: Path) -> None:
assert "custom-act" in _read(project_pyproject)


def _create_sample_project(tmp_path: Path, lang: str = "en") -> Path:
"""Helper to scaffold a sample Act project for testing."""
target_dir = tmp_path / "sample-act"
result = runner.invoke(
app,
[
"--path",
str(target_dir),
"--act-name",
"Sample Act",
"--cast-name",
"Primary Cast",
"--lang",
lang,
],
)
assert result.exit_code == 0, result.stdout
return target_dir


def test_upgrade_replaces_skills(tmp_path: Path) -> None:
target_dir = _create_sample_project(tmp_path)
skill_file = target_dir / ".claude" / "skills" / "testing-cast" / "SKILL.md"
assert skill_file.exists()

# Corrupt a skill file
original = _read(skill_file)
skill_file.write_text("corrupted content", encoding="utf-8")
assert _read(skill_file) == "corrupted content"

# Upgrade should restore it
result = runner.invoke(app, ["upgrade", "--path", str(target_dir)])
combined = (result.stdout or "") + (result.stderr or "")
assert result.exit_code == 0, combined
assert "upgraded successfully" in combined
assert _read(skill_file) == original


def test_upgrade_creates_backup(tmp_path: Path) -> None:
target_dir = _create_sample_project(tmp_path)
skills_dir = target_dir / ".claude" / "skills"
assert skills_dir.exists()

result = runner.invoke(app, ["upgrade", "--path", str(target_dir)])
assert result.exit_code == 0, result.stdout

backup_dir = target_dir / ".claude" / "skills.bak"
assert backup_dir.exists()
assert (backup_dir / "testing-cast").exists()


def test_upgrade_fails_on_invalid_project(tmp_path: Path) -> None:
invalid_dir = tmp_path / "not-a-project"
invalid_dir.mkdir()

result = runner.invoke(app, ["upgrade", "--path", str(invalid_dir)])
assert result.exit_code != 0


def test_init_aborts_on_non_empty_dir(tmp_path: Path) -> None:
target_dir = tmp_path / "existing"
target_dir.mkdir()
Expand Down