Coverage for astrocyte/harness/commands.py: 97%
245 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""``astrocyte setup`` / ``uninstall`` / ``doctor`` command implementations."""
3from __future__ import annotations
5import json
6import sys
7from argparse import Namespace
8from pathlib import Path
10from .choices import load_choices, save_choices
11from .doctor import Check, apply_fixes, run_checks
12from .hosts import ALL_HOSTS, SUPPORTED_HOSTS, HookHost, Host, Outcome, host_by_key, hosts
13from .localconfig import SetupError, choose_providers, render_config, write_config
14from .paths import config_path, database_path
15from .server import HookPathError, handshake, hook_prefix, locate_mcp_server, server_spec
17HOST_KEYS = tuple(cls.key for cls in ALL_HOSTS) # v0.16.0's value: part of the public API
18SUPPORTED_HOST_KEYS = tuple(cls.key for cls in SUPPORTED_HOSTS)
20_MARK = {"ok": "✓", "warn": "!", "fail": "✗", "info": "·"}
21_OUTCOME_MARK = {
22 "installed": "✓", "updated": "✓", "unchanged": "✓", "removed": "✓", "absent": "·",
23 "planned": "→", "skipped": "·", "failed": "✗",
24}
27def _selected_hosts(args: Namespace) -> tuple[list[Host], bool]:
28 """Hosts named by flags, else every detected one. Returns (hosts, explicit)."""
29 named = [k for k in SUPPORTED_HOST_KEYS if getattr(args, k, False)]
30 if named:
31 return [host_by_key(k) for k in named], True
32 return [h for h in hosts() if h.detected()], False
35def _print_outcome(o: Outcome) -> None:
36 print(f" {_OUTCOME_MARK[o.status]} {o.host:<12} {o.status:<9} {o.detail}")
39def cmd_setup(args: Namespace) -> int:
40 cfg_path = Path(args.config).expanduser() if args.config else config_path()
41 dry = args.dry_run
43 print("Astrocyte setup\n")
44 # 1. Config — the user's file once written; never overwritten.
45 if cfg_path.is_file():
46 print(f" ✓ config keeping existing {cfg_path}")
47 else:
48 try:
49 choice = choose_providers()
50 except SetupError as e:
51 print(f" ✗ config {e}", file=sys.stderr)
52 return 1
53 if dry:
54 print(f" → config would write {cfg_path}")
55 else:
56 write_config(cfg_path, render_config(choice, database_path()))
57 print(f" ✓ config wrote {cfg_path}")
58 print(f" completions: {choice.llm_why}")
59 if choice.embedding_why:
60 print(f" embeddings: {choice.embedding_why}")
61 print(f" memories: {database_path()}")
63 # 2. The server binary from *this* installation.
64 found = locate_mcp_server()
65 if found is None:
66 print(" ✗ server astrocyte-mcp not found. Install with: uv tool install 'astrocyte[local]'",
67 file=sys.stderr)
68 return 1
69 if found.ephemeral:
70 print(f" ! server {found.command} is in a cache that may be pruned;\n"
71 " for a durable install run: uv tool install 'astrocyte[local]'")
72 spec = server_spec(found.command, cfg_path)
74 # 3. Prove the server starts before wiring anything to it.
75 if not dry and not args.no_verify:
76 result = handshake(spec)
77 if not result.ok:
78 print(f" ✗ server failed to start: {result.detail}\n"
79 " nothing was wired. Fix the above, then re-run.", file=sys.stderr)
80 return 1
81 print(f" ✓ server starts and answers ({result.detail}, {result.seconds:.1f}s)")
82 warmed = _warm_embeddings(cfg_path)
83 if warmed:
84 print(f" ✓ model {warmed}")
86 # Memories hold conversations: a store other accounts can read (created
87 # before stores were made private) is tightened here.
88 if not dry:
89 from .doctor import _make_store_private
91 for line in _make_store_private(cfg_path):
92 print(f" ✓ store {line}")
94 # 4. Harnesses. Naming one records the choice made for it; a plain run
95 # leaves alone what the user switched off before.
96 targets, explicit = _selected_hosts(args)
97 if not targets:
98 print("\n No agent harnesses detected. Supported: " + ", ".join(c.label for c in SUPPORTED_HOSTS)
99 + ".\n Install one, or name it explicitly, e.g. astrocyte setup --claude")
100 return 1
101 choices = load_choices(cfg_path)
102 if explicit:
103 for h in targets:
104 choices.off.discard(h.key)
105 (choices.hooks_off.add if args.no_hooks else choices.hooks_off.discard)(h.key)
106 for h in targets: # opt-in file recall, where the harness has a file hook
107 if getattr(h, "FILE_HOOK", None) and getattr(args, "file_recall", False):
108 choices.file_recall.add(h.key)
109 elif getattr(args, "no_file_recall", False):
110 choices.file_recall.discard(h.key)
111 else:
112 print("\n Detected: " + ", ".join(h.label for h in targets))
113 if args.no_hooks:
114 choices.hooks_off |= {h.key for h in targets if isinstance(h, HookHost)}
115 switched_off = [h for h in targets if h.key in choices.off]
116 targets = [h for h in targets if h.key not in choices.off]
117 print()
118 outcomes = [Outcome(h.label, "skipped", f"switched off; astrocyte setup --{h.key} turns it back on")
119 for h in switched_off]
120 outcomes += [h.install(spec, dry_run=dry) for h in targets]
122 # 5. Automatic memory (lifecycle hooks, where the harness has them).
123 hook_hosts = [h for h in targets if isinstance(h, HookHost)]
124 auto_memory = [h for h in hook_hosts if h.key not in choices.hooks_off]
125 try:
126 prefix, prefix_error = hook_prefix(found.command), ""
127 except HookPathError as e: # Windows: a path no shell-neutral command can name
128 prefix, prefix_error = None, str(e)
129 for h in hook_hosts:
130 if h in auto_memory:
131 extra = {"file_recall": h.key in choices.file_recall} if h.FILE_HOOK else {}
132 outcomes.append(h.install_hooks(prefix, dry_run=dry, **extra) if prefix
133 else Outcome(f"{h.label} hooks", "failed", prefix_error))
134 continue
135 # Off by choice (--no-hooks, now or before), including a previous install.
136 outcome = h.uninstall_hooks(dry_run=dry)
137 if outcome.status != "absent":
138 outcomes.append(outcome)
139 elif not args.no_hooks:
140 outcomes.append(Outcome(f"{h.label} hooks", "skipped",
141 f"automatic memory off; astrocyte setup --{h.key} turns it on"))
142 for o in outcomes:
143 _print_outcome(o)
144 if not dry:
145 save_choices(cfg_path, choices)
146 if not targets:
147 print("\nEvery detected agent is switched off; nothing was wired. "
148 f"Name one to turn it back on, e.g. astrocyte setup --{switched_off[0].key}")
149 return 0
151 failed = [o for o in outcomes if o.status == "failed"]
152 if dry:
153 print("\nDry run — nothing was changed.")
154 elif failed:
155 print(f"\n{len(failed)} harness(es) could not be wired; see above. `astrocyte doctor` re-checks.")
156 else:
157 fresh = {o.host for o in outcomes if o.status in ("installed", "updated")}
158 notes = [h.next_step for h in targets if h.label in fresh and h.next_step]
159 notes += [h.hooks_hint() for h in hook_hosts if f"{h.label} hooks" in fresh and h.hooks_hint()]
160 if notes:
161 print("\nOne more step in some agents:")
162 for note in notes:
163 print(f" • {note}")
164 if auto_memory:
165 capture = [h.label for h in auto_memory if h.captures_turns]
166 recall_only = [h.label for h in auto_memory if not h.captures_turns]
167 print("\nAutomatic memory is on in interactive sessions, including ones already open.")
168 if capture:
169 print(f" {_join(capture)}: each finished turn is saved to that project's local memory,\n"
170 " and relevant memories are added to new prompts.")
171 if recall_only:
172 print(f" {_join(recall_only)}: relevant memories (saved by your other agents) are added to\n"
173 " new prompts; its own turns are not saved yet.")
174 file_recall = [h.label for h in auto_memory if h.FILE_HOOK and h.key in choices.file_recall]
175 if file_recall:
176 print(f" {_join(file_recall)}: after it reads or edits a file, memories of earlier turns that\n"
177 " touched that file are added too (file recall; off with --no-file-recall).")
178 print("Headless runs (`claude -p`, `codex exec`, `agy -p`, `copilot -p`) are left alone.\n"
179 "Pause with ASTROCYTE_HOOKS=off, or remove with: astrocyte setup --no-hooks")
180 seed = _seed_files(Path.cwd())
181 if seed:
182 print("\nGive this project's memory a head start from its agent instructions:\n"
183 f" astrocyte memory import {' '.join(seed)}")
184 print("\nDone. Start a new session in your agent to load the Astrocyte memory tools.\n"
185 "Verify any time with: astrocyte doctor")
186 return 1 if failed else 0
189# Instruction files coding agents already read: importing them means memory
190# starts with the project's conventions instead of empty.
191SEED_FILES = ("CLAUDE.md", "AGENTS.md", "GEMINI.md", ".github/copilot-instructions.md", ".cursorrules")
194def _seed_files(cwd: Path) -> list[str]:
195 from .project import project_root
197 root = project_root(str(cwd))
198 if not (root / ".git").exists():
199 return [] # not standing in a project
200 return [name for name in SEED_FILES if (root / name).is_file()]
203def _warm_embeddings(cfg_path: Path) -> str:
204 """Fetch and load a local embedding model now (bge-small, ~130 MB on first
205 run), so the first session doesn't wait for the download. Best effort:
206 the server already started; a failure here only means a slower first use."""
207 import asyncio
208 import time
210 from astrocyte.config import load_config
211 from astrocyte.wiring import resolve_llm_provider
213 try:
214 config = load_config(str(cfg_path))
215 if config.embedding_provider != "local_embeddings":
216 return ""
217 started = time.perf_counter()
218 print(" … model loading the local embedding model (downloaded once, ~130 MB)", flush=True)
219 asyncio.run(resolve_llm_provider(config).embed(["warm"]))
220 return f"local embedding model ready ({time.perf_counter() - started:.1f}s)"
221 except Exception as e: # noqa: BLE001 — never fail setup over a warm-up
222 return f"could not load the embedding model yet ({type(e).__name__}); it will load on first use"
225def cmd_uninstall(args: Namespace) -> int:
226 targets, explicit = _selected_hosts(args)
227 print("Removing Astrocyte from agent harnesses\n")
228 outcomes = [h.uninstall(dry_run=args.dry_run) for h in targets]
229 outcomes += [h.uninstall_hooks(dry_run=args.dry_run) for h in targets if isinstance(h, HookHost)]
230 for o in outcomes:
231 _print_outcome(o)
232 cfg_path = config_path()
233 if explicit:
234 # Removing one agent is a standing choice; removing them all (no
235 # flags) is a teardown, which a later `astrocyte setup` reverses.
236 if not args.dry_run:
237 choices = load_choices(cfg_path)
238 choices.off |= {h.key for h in targets}
239 save_choices(cfg_path, choices)
240 print(f"\nastrocyte setup will leave {_join([h.label for h in targets])} switched off;\n"
241 f"turn it back on with: astrocyte setup {' '.join('--' + h.key for h in targets)}")
242 print(f"\nYour config ({cfg_path}) and memories ({database_path()}) were left in place;\n"
243 "delete them yourself if you no longer want them.")
244 return 1 if any(o.status == "failed" for o in outcomes) else 0
247def _render(checks: list[Check]) -> None:
248 width = max(len(c.area) for c in checks)
249 for c in checks:
250 print(f" {_MARK[c.level]} {c.area:<{width}} {c.summary}")
251 if c.fix and c.level in ("fail", "warn"):
252 print(f" {' ' * width} fix: {c.fix}")
255def cmd_doctor(args: Namespace) -> int:
256 cfg_path = Path(args.config).expanduser() if args.config else config_path()
257 checks = run_checks(cfg_path, model_probes=not args.skip_models)
258 repaired: list[str] = []
259 if args.fix and any(c.fixable for c in checks):
260 repaired = apply_fixes(cfg_path, checks)
261 checks = run_checks(cfg_path, model_probes=not args.skip_models)
263 failures = [c for c in checks if c.level == "fail"]
264 if args.json:
265 print(json.dumps({"ok": not failures, "checks": [c.as_dict() for c in checks], "repaired": repaired},
266 indent=2))
267 return 1 if failures else 0
269 print("Astrocyte doctor\n")
270 if repaired:
271 print("Repaired:")
272 for line in repaired:
273 print(f" ✓ {line}")
274 print()
275 _render(checks)
276 if failures:
277 fixable = any(c.fixable for c in failures)
278 hint = " Run `astrocyte doctor --fix` to repair what it can." if fixable and not args.fix else ""
279 print(f"\n{len(failures)} problem(s).{hint}")
280 return 1
281 print("\nAll good.")
282 return 0
285def register(sub) -> None:
286 """Add setup / uninstall / doctor to the ``astrocyte`` CLI."""
288 def host_flags(p) -> None:
289 for cls in SUPPORTED_HOSTS:
290 p.add_argument(f"--{cls.key}", action="store_true", help=f"only {cls.label}")
291 p.add_argument("--dry-run", action="store_true", help="show what would change, change nothing")
293 setup = sub.add_parser(
294 "setup",
295 help="Create a local memory store and wire it into your coding agents",
296 description="Writes ~/.config/astrocyte/astrocyte.yaml (if absent), verifies the MCP server starts, "
297 "then registers it with every detected agent harness (or only those named).",
298 )
299 host_flags(setup)
300 setup.add_argument("--config", help="config path (default: ~/.config/astrocyte/astrocyte.yaml)")
301 setup.add_argument("--no-verify", action="store_true", help="skip the server start-up check")
302 setup.add_argument("--file-recall", action="store_true",
303 help="also recall memories of earlier turns when the agent reads or edits a file "
304 "(Claude Code; opt-in, remembered)")
305 setup.add_argument("--no-file-recall", action="store_true", help="turn file recall off again")
306 setup.add_argument("--no-hooks", action="store_true",
307 help="don't enable automatic memory (Claude Code / Codex capture + recall hooks); "
308 "removes them if present")
309 setup.set_defaults(func=cmd_setup)
311 hook = sub.add_parser("hook", help="(called by agent hooks) automatic memory for one lifecycle event")
312 hook.add_argument("event", choices=["session-start", "prompt", "stop", "file", "edit"])
313 hook.add_argument("--host", choices=["claude", "codex", "antigravity", "copilot"], default="claude",
314 help="the agent firing the hook")
315 hook.set_defaults(func=lambda a: _run_hook(a.event, a.host))
317 agentd = sub.add_parser("agentd", help="(started on demand) keep memory warm for agent hooks")
318 agentd.add_argument("--config", help="config path (default: ~/.config/astrocyte/astrocyte.yaml)")
319 agentd.set_defaults(func=lambda a: _run_agentd(a.config))
321 uninstall = sub.add_parser("uninstall", help="Remove Astrocyte from agent harnesses (keeps your memories)")
322 host_flags(uninstall)
323 uninstall.set_defaults(func=cmd_uninstall)
325 doctor = sub.add_parser("doctor", help="Check the local install end to end; --fix repairs it")
326 doctor.add_argument("--fix", action="store_true", help="repair a missing config and broken harness entries")
327 doctor.add_argument("--json", action="store_true", help="machine-readable output")
328 doctor.add_argument("--skip-models", action="store_true", help="skip the embedding/completion probes")
329 doctor.add_argument("--config", help="config path (default: ~/.config/astrocyte/astrocyte.yaml)")
330 doctor.set_defaults(func=cmd_doctor)
332 from .memories import register as register_memory
334 register_memory(sub)
336 from .team import register as register_team
338 register_team(sub)
341def _run_hook(event: str, host: str) -> int:
342 from .hooks import main
344 return main(event, host)
347def _run_agentd(config: str | None) -> int:
348 from .agentd import run
350 return run(Path(config).expanduser() if config else config_path())
353def _join(names: list[str]) -> str:
354 return names[0] if len(names) == 1 else ", ".join(names[:-1]) + " and " + names[-1]