1from __future__ import annotations
2
3import importlib.util
4import json
5import pkgutil
6import shutil
7from pathlib import Path
8
9import click
10
11
12def _get_agent_dirs() -> list[Path]:
13 """Get list of agents/.claude/ directories from installed plain.* and plainx.* packages."""
14 agent_dirs: list[Path] = []
15
16 # Search plain.* packages
17 try:
18 import plain
19
20 # Check core plain package (namespace package)
21 plain_spec = importlib.util.find_spec("plain")
22 if plain_spec and plain_spec.submodule_search_locations:
23 for location in plain_spec.submodule_search_locations:
24 agent_dir = Path(location) / "agents" / ".claude"
25 if agent_dir.exists() and agent_dir.is_dir():
26 agent_dirs.append(agent_dir)
27 break
28
29 # Check other plain.* subpackages
30 if hasattr(plain, "__path__"):
31 for importer, modname, ispkg in pkgutil.iter_modules(
32 plain.__path__, "plain."
33 ):
34 if ispkg:
35 try:
36 spec = importlib.util.find_spec(modname)
37 if spec and spec.origin:
38 agent_dir = Path(spec.origin).parent / "agents" / ".claude"
39 if agent_dir.exists() and agent_dir.is_dir():
40 agent_dirs.append(agent_dir)
41 except Exception:
42 continue
43 except Exception:
44 pass
45
46 # Search plainx.* packages
47 try:
48 import plainx # ty: ignore[unresolved-import]
49
50 # Check plainx.* subpackages
51 if hasattr(plainx, "__path__"):
52 for importer, modname, ispkg in pkgutil.iter_modules(
53 plainx.__path__, "plainx."
54 ):
55 if ispkg:
56 try:
57 spec = importlib.util.find_spec(modname)
58 if spec and spec.origin:
59 agent_dir = Path(spec.origin).parent / "agents" / ".claude"
60 if agent_dir.exists() and agent_dir.is_dir():
61 agent_dirs.append(agent_dir)
62 except Exception:
63 continue
64 except Exception:
65 pass
66
67 return agent_dirs
68
69
70def _install_agent_dir(source_dir: Path, dest_dir: Path) -> tuple[int, int]:
71 """Copy contents of a source agents/.claude/ dir to the project's .claude/ dir.
72
73 Handles skills/ subdirectories and rules/ files.
74 Returns (installed_count, removed_count) for reporting.
75 """
76 installed_count = 0
77
78 # Copy skills (directories containing SKILL.md)
79 source_skills = source_dir / "skills"
80 if source_skills.exists():
81 dest_skills = dest_dir / "skills"
82 dest_skills.mkdir(parents=True, exist_ok=True)
83 for skill_dir in source_skills.iterdir():
84 if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists():
85 dest_skill = dest_skills / skill_dir.name
86 # Check mtime to skip unchanged
87 if dest_skill.exists():
88 source_mtime = (skill_dir / "SKILL.md").stat().st_mtime
89 dest_mtime = (
90 (dest_skill / "SKILL.md").stat().st_mtime
91 if (dest_skill / "SKILL.md").exists()
92 else 0
93 )
94 if source_mtime <= dest_mtime:
95 continue
96 shutil.rmtree(dest_skill)
97 shutil.copytree(skill_dir, dest_skill)
98 installed_count += 1
99
100 # Copy rules (individual .md files)
101 source_rules = source_dir / "rules"
102 if source_rules.exists():
103 dest_rules = dest_dir / "rules"
104 dest_rules.mkdir(parents=True, exist_ok=True)
105 for rule_file in source_rules.iterdir():
106 if rule_file.is_file() and rule_file.suffix == ".md":
107 dest_rule = dest_rules / rule_file.name
108 # Check mtime to skip unchanged
109 if dest_rule.exists():
110 if rule_file.stat().st_mtime <= dest_rule.stat().st_mtime:
111 continue
112 shutil.copy2(rule_file, dest_rule)
113 installed_count += 1
114
115 return installed_count, 0
116
117
118def _is_plain_item(name: str) -> bool:
119 """Check if a name is from plain or plainx namespace."""
120 return name.startswith("plain") # Matches both 'plain' and 'plainx'
121
122
123def _cleanup_orphans(dest_dir: Path, agent_dirs: list[Path]) -> int:
124 """Remove plain* and plainx* items from .claude/ that no longer exist in any source package."""
125 removed_count = 0
126
127 # Collect all source skill and rule names
128 source_skills: set[str] = set()
129 source_rules: set[str] = set()
130 for agent_dir in agent_dirs:
131 skills_dir = agent_dir / "skills"
132 if skills_dir.exists():
133 for d in skills_dir.iterdir():
134 if d.is_dir() and (d / "SKILL.md").exists():
135 source_skills.add(d.name)
136 rules_dir = agent_dir / "rules"
137 if rules_dir.exists():
138 for f in rules_dir.iterdir():
139 if f.is_file() and f.suffix == ".md":
140 source_rules.add(f.name)
141
142 # Remove orphaned skills
143 dest_skills = dest_dir / "skills"
144 if dest_skills.exists():
145 for dest in dest_skills.iterdir():
146 if (
147 dest.is_dir()
148 and _is_plain_item(dest.name)
149 and dest.name not in source_skills
150 ):
151 shutil.rmtree(dest)
152 removed_count += 1
153
154 # Remove orphaned rules
155 dest_rules = dest_dir / "rules"
156 if dest_rules.exists():
157 for dest in dest_rules.iterdir():
158 if (
159 dest.is_file()
160 and _is_plain_item(dest.name)
161 and dest.suffix == ".md"
162 and dest.name not in source_rules
163 ):
164 dest.unlink()
165 removed_count += 1
166
167 return removed_count
168
169
170def _cleanup_session_hook(dest_dir: Path) -> None:
171 """Remove the old plain agent context SessionStart hook from settings.json."""
172 settings_file = dest_dir / "settings.json"
173
174 if not settings_file.exists():
175 return
176
177 settings = json.loads(settings_file.read_text())
178
179 hooks = settings.get("hooks", {})
180 session_hooks = hooks.get("SessionStart", [])
181
182 # Remove any plain agent or plain-context.md hooks
183 session_hooks = [h for h in session_hooks if "plain agent" not in str(h)]
184 session_hooks = [h for h in session_hooks if "plain-context.md" not in str(h)]
185
186 if session_hooks:
187 hooks["SessionStart"] = session_hooks
188 else:
189 hooks.pop("SessionStart", None)
190
191 if hooks:
192 settings["hooks"] = hooks
193 else:
194 settings.pop("hooks", None)
195
196 if settings:
197 settings_file.write_text(json.dumps(settings, indent=2) + "\n")
198 else:
199 settings_file.unlink()
200
201
202@click.group()
203def agent() -> None:
204 """AI agent integration for Plain projects"""
205 pass
206
207
208@agent.command()
209def install() -> None:
210 """Install skills and rules to agent directories"""
211 cwd = Path.cwd()
212 claude_dir = cwd / ".claude"
213
214 if not claude_dir.exists():
215 click.secho("No .claude/ directory found.", fg="yellow")
216 return
217
218 agent_dirs = _get_agent_dirs()
219
220 # Clean up orphaned plain-* items
221 removed_count = _cleanup_orphans(claude_dir, agent_dirs)
222
223 # Install from each package
224 total_installed = 0
225 for source_dir in agent_dirs:
226 installed, _ = _install_agent_dir(source_dir, claude_dir)
227 total_installed += installed
228
229 # Clean up old session hook
230 _cleanup_session_hook(claude_dir)
231
232 parts = []
233 if total_installed > 0:
234 parts.append(f"installed {total_installed}")
235 if removed_count > 0:
236 parts.append(f"removed {removed_count}")
237 click.echo(f"Agent: {', '.join(parts)} in .claude/") if parts else click.echo(
238 "Agent: up to date"
239 )
240
241
242@agent.command()
243def skills() -> None:
244 """List available skills from installed packages"""
245 agent_dirs = _get_agent_dirs()
246
247 skill_names = []
248 for agent_dir in agent_dirs:
249 skills_dir = agent_dir / "skills"
250 if skills_dir.exists():
251 for d in skills_dir.iterdir():
252 if d.is_dir() and (d / "SKILL.md").exists():
253 skill_names.append(d.name)
254
255 if not skill_names:
256 click.echo("No skills found in installed packages.")
257 return
258
259 click.echo("Available skills:")
260 for name in sorted(skill_names):
261 click.echo(f" - {name}")