Coverage for astrocyte/harness/hosts.py: 96%

1"""Agent harnesses Astrocyte can register its MCP server with. 

2 

3Each host knows where its harness keeps MCP registrations, how to read the 

4current ``astrocyte`` entry, and how to write one. Writes go through the 

5harness's own CLI where it has one (``claude``, ``codex``, ``gemini``) — the 

6CLI owns its file format and locking, and ``~/.claude.json`` in particular is 

7rewritten constantly by a running Claude Code. JSON-configured harnesses 

8(Cursor, Windsurf, Copilot CLI) are edited directly, atomically, refusing to 

9touch a file that is not valid JSON. 

10 

11Every write is verified by re-reading the harness's config: CLI exit codes 

12are not trustworthy here (``claude mcp add`` exits 0 and changes nothing when 

13the name already exists), and an unverified "installed" is the failure mode 

14this whole module exists to prevent. 

15""" 

16 

17from __future__ import annotations 

18 

19import copy 

20import json 

21import os 

22import re 

23import shutil 

24import subprocess 

25import tomllib 

26from abc import ABC, abstractmethod 

27from dataclasses import dataclass 

28from pathlib import Path 

29from typing import Any, Literal 

30 

31SERVER_NAME = "astrocyte" 

32 

33Status = Literal["installed", "updated", "unchanged", "removed", "absent", "planned", "skipped", "failed"] 

34 

35 

36_VERB = {"installed": "install", "updated": "update"} 

37 

38 

39class HostConfigError(Exception): 

40 """A harness config file exists but cannot be read safely.""" 

41 

42 

43@dataclass(frozen=True) 

44class ServerSpec: 

45 """The MCP server entry setup registers: an absolute command and args.""" 

46 

47 command: str 

48 args: tuple[str, ...] 

49 

50 

51@dataclass(frozen=True) 

52class Registration: 

53 """What a harness currently has registered under ``astrocyte``.""" 

54 

55 command: str 

56 args: tuple[str, ...] 

57 

58 def matches(self, spec: ServerSpec) -> bool: 

59 return self.command == spec.command and self.args == spec.args 

60 

61 

62@dataclass(frozen=True) 

63class Outcome: 

64 host: str 

65 status: Status 

66 detail: str = "" 

67 

68 

69def _home() -> Path: 

70 # $HOME on POSIX; USERPROFILE on Windows (which ignores HOME, as do the 

71 # agents there: their configs live under the profile even when Git Bash 

72 # sets HOME elsewhere). 

73 return Path.home() 

74 

75 

76def _read_json(path: Path) -> dict[str, Any]: 

77 if not path.exists(): 

78 return {} 

79 try: 

80 data = json.loads(path.read_text(encoding="utf-8") or "{}") 

81 except json.JSONDecodeError as e: 

82 raise HostConfigError(f"{path} is not valid JSON ({e}); fix it by hand, then re-run") from e 

83 if not isinstance(data, dict): 

84 raise HostConfigError(f"{path} is not a JSON object; fix it by hand, then re-run") 

85 return data 

86 

87 

88def edit_json(path: Path, change: Any) -> None: 

89 """Apply ``change(data)`` to a JSON file: refuses invalid JSON before 

90 writing anything, keeps a one-time backup of the user's original, writes 

91 atomically, and preserves the file's permissions.""" 

92 data = _read_json(path) 

93 change(data) 

94 path.parent.mkdir(parents=True, exist_ok=True) 

95 if path.exists(): 

96 backup = path.with_name(path.name + ".astrocyte-bak") 

97 if not backup.exists(): 

98 shutil.copy2(path, backup) 

99 tmp = path.with_name(path.name + ".astrocyte-tmp") 

100 tmp.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") 

101 if path.exists(): 

102 shutil.copymode(path, tmp) 

103 os.replace(tmp, path) # atomic: a reader never sees a half-written file 

104 

105 

106def _registration_from(entry: Any) -> Registration | None: 

107 if not isinstance(entry, dict) or not entry.get("command"): 

108 return None 

109 return Registration(command=str(entry["command"]), args=tuple(str(a) for a in entry.get("args") or ())) 

110 

111 

112class Host(ABC): 

113 key: str 

114 label: str 

115 # A consent gate the harness applies to new MCP servers, which setup 

116 # cannot (and should not) bypass. Shown after a fresh install so "it 

117 # doesn't work in X" is never the first experience. 

118 next_step: str = "" 

119 

120 @abstractmethod 

121 def detected(self) -> bool: 

122 """Is this harness installed for the current user?""" 

123 

124 @abstractmethod 

125 def config_file(self) -> Path: 

126 """The file holding this harness's MCP registrations.""" 

127 

128 @abstractmethod 

129 def registration(self) -> Registration | None: 

130 """The current ``astrocyte`` entry; raises HostConfigError if unreadable.""" 

131 

132 @abstractmethod 

133 def _write(self, spec: ServerSpec) -> None: ... 

134 

135 @abstractmethod 

136 def _remove(self) -> None: ... 

137 

138 def install(self, spec: ServerSpec, *, dry_run: bool = False) -> Outcome: 

139 try: 

140 current = self.registration() 

141 except HostConfigError as e: 

142 return Outcome(self.label, "failed", str(e)) 

143 if current is not None and current.matches(spec): 

144 return Outcome(self.label, "unchanged", str(self.config_file())) 

145 verb: Status = "updated" if current is not None else "installed" 

146 if dry_run: 

147 return Outcome(self.label, "planned", f"would {_VERB[verb]} {self.config_file()}") 

148 try: 

149 self._write(spec) 

150 after = self.registration() 

151 except (HostConfigError, OSError, subprocess.SubprocessError) as e: 

152 return Outcome(self.label, "failed", str(e)) 

153 if after is None or not after.matches(spec): 

154 return Outcome(self.label, "failed", f"wrote {self.config_file()} but the entry did not take effect") 

155 return Outcome(self.label, verb, str(self.config_file())) 

156 

157 def uninstall(self, *, dry_run: bool = False) -> Outcome: 

158 try: 

159 if self.registration() is None: 

160 return Outcome(self.label, "absent", str(self.config_file())) 

161 if dry_run: 

162 return Outcome(self.label, "planned", f"would remove from {self.config_file()}") 

163 self._remove() 

164 if self.registration() is not None: 

165 return Outcome(self.label, "failed", f"entry still present in {self.config_file()}") 

166 except (HostConfigError, OSError, subprocess.SubprocessError) as e: 

167 return Outcome(self.label, "failed", str(e)) 

168 return Outcome(self.label, "removed", str(self.config_file())) 

169 

170 

171# ── CLI-managed harnesses ──────────────────────────────────────────────── 

172 

173 

174class _CliHost(Host): 

175 """Writes through the harness's own CLI; reads the file it maintains.""" 

176 

177 binary: str 

178 

179 def _cli(self) -> str: 

180 found = shutil.which(self.binary) 

181 if not found: 

182 raise HostConfigError( 

183 f"`{self.binary}` is not on PATH, so its MCP config can't be edited safely. " 

184 f"Add it to PATH and re-run, or register manually: {self.manual_command()}" 

185 ) 

186 return found 

187 

188 def _run(self, *args: str, check: bool = True) -> None: 

189 proc = subprocess.run( 

190 [self._cli(), *args], 

191 capture_output=True, 

192 text=True, 

193 timeout=60, 

194 # Gemini's `mcp add` writes project scope relative to the cwd; 

195 # pinning cwd to HOME keeps stray project files out of repos. 

196 cwd=str(_home()), 

197 ) 

198 if check and proc.returncode != 0: 

199 raise subprocess.SubprocessError( 

200 f"`{self.binary} {' '.join(args)}` exited {proc.returncode}: " 

201 f"{(proc.stderr or proc.stdout).strip()[:300]}" 

202 ) 

203 

204 @abstractmethod 

205 def manual_command(self, spec: ServerSpec | None = None) -> str: ... 

206 

207 

208# ── lifecycle hooks (automatic memory) ─────────────────────────────────── 

209 

210 

211class HookHost: 

212 """Installs ``astrocyte hook`` commands into a harness's hooks file. 

213 

214 Claude Code and Codex share one layout — ``{"hooks": {Event: [{"hooks": 

215 [entry]}]}}`` — and one input/output contract, so one installer serves 

216 both. Entries are recognised by command, never by position: the user's 

217 own hooks in the same file are left exactly as they were. 

218 """ 

219 

220 label: str 

221 #: Event → (``astrocyte hook`` subcommand, extra entry fields). 

222 HOOK_EVENTS: dict[str, tuple[str, dict[str, Any]]] 

223 #: ``--host`` passed to ``astrocyte hook``; None for Claude Code, whose 

224 #: hooks predate the flag (keeps existing installs "unchanged"). 

225 hook_dialect: str | None = None 

226 #: A consent gate the harness applies to new hooks (cf. Host.next_step). 

227 hooks_next_step: str = "" 

228 #: Do this harness's hooks save finished turns, or only recall? 

229 captures_turns: bool = True 

230 #: Opt-in recall on file reads and edits: (event, tool matcher, extra entry 

231 #: fields), or None where the harness has no such hook (yet). 

232 FILE_HOOK: tuple[str, str, dict[str, Any]] | None = None 

233 #: Hooks on tool calls, installed with the HOOK_EVENTS ones: event → 

234 #: (``astrocyte hook`` subcommand, tool matcher, extra entry fields). 

235 TOOL_HOOKS: dict[str, tuple[str, str, dict[str, Any]]] = {} 

236 # Matches the current ``<python> -I -m astrocyte.cli hook …`` form and the 

237 # earlier ``astrocyte hook …`` console-script form, with or without --host. 

238 _OURS = re.compile( 

239 r"astrocyte(?:\.cli)?['\"]?\s+hook\s+(session-start|prompt|stop|file|edit)(?:\s+--host\s+[a-z-]+)?\s*$" 

240 ) 

241 

242 @abstractmethod 

243 def hooks_file(self) -> Path: ... 

244 

245 def _events(self) -> dict[str, tuple[str, dict[str, Any]]]: 

246 """Every always-installed hook: HOOK_EVENTS plus TOOL_HOOKS.""" 

247 return {**self.HOOK_EVENTS, **{e: (sub, extra) for e, (sub, _, extra) in self.TOOL_HOOKS.items()}} 

248 

249 def _matchers(self) -> dict[str, str]: 

250 return {event: matcher for event, (_, matcher, _) in self.TOOL_HOOKS.items()} 

251 

252 def hooks_hint(self) -> str: 

253 """What to do after installing new or changed hooks ("" for nothing).""" 

254 return self.hooks_next_step 

255 

256 @classmethod 

257 def _is_ours(cls, hook: Any) -> bool: 

258 return isinstance(hook, dict) and bool(cls._OURS.search(str(hook.get("command", "")))) 

259 

260 def hook_commands(self) -> dict[str, str | None]: 

261 """Our installed hook command per event (None where absent).""" 

262 hooks = _read_json(self.hooks_file()).get("hooks", {}) 

263 found: dict[str, str | None] = dict.fromkeys(self._events()) 

264 if not isinstance(hooks, dict): 

265 return found 

266 for event in self._events(): 

267 for group in hooks.get(event) or []: 

268 for hook in (group or {}).get("hooks") or []: 

269 if self._is_ours(hook): 

270 found[event] = hook["command"] 

271 return found 

272 

273 def _wanted(self, prefix: str, file_recall: bool = False) -> dict[str, dict[str, Any]]: 

274 """``prefix`` is a ready-quoted command, e.g. ``'/py' -I -m astrocyte.cli``.""" 

275 flag = f" --host {self.hook_dialect}" if self.hook_dialect else "" 

276 wanted = { 

277 event: {"type": "command", "command": f"{prefix} hook {sub}{flag}", **extra} 

278 for event, (sub, extra) in self._events().items() 

279 } 

280 if file_recall and self.FILE_HOOK is not None: 

281 event, _, extra = self.FILE_HOOK 

282 wanted[event] = {"type": "command", "command": f"{prefix} hook file{flag}", **extra} 

283 return wanted 

284 

285 def _all_ours(self) -> dict[str, str]: 

286 """Every event holding one of our hooks, base or file recall.""" 

287 hooks = _read_json(self.hooks_file()).get("hooks", {}) 

288 found: dict[str, str] = {} 

289 for event, groups in (hooks.items() if isinstance(hooks, dict) else ()): 

290 for group in groups if isinstance(groups, list) else (): 

291 for hook in (group or {}).get("hooks") or []: 

292 if self._is_ours(hook): 

293 found[event] = hook["command"] 

294 return found 

295 

296 def file_recall_command(self) -> str | None: 

297 """Our file-recall hook command, if installed.""" 

298 return self._all_ours().get(self.FILE_HOOK[0]) if self.FILE_HOOK else None 

299 

300 def _strip_ours(self, data: dict) -> None: 

301 hooks = data.get("hooks") 

302 if not isinstance(hooks, dict): 

303 return 

304 for event in list(hooks): 

305 if not isinstance(hooks[event], list): 

306 continue # not an event (e.g. Codex's per-hook trust state) 

307 groups = [] 

308 for group in hooks.get(event) or []: 

309 kept = [h for h in (group or {}).get("hooks") or [] if not self._is_ours(h)] 

310 if kept: 

311 groups.append({**group, "hooks": kept}) 

312 if groups: 

313 hooks[event] = groups 

314 else: 

315 hooks.pop(event) 

316 if not hooks: 

317 data.pop("hooks") 

318 

319 def install_hooks(self, prefix: str, *, dry_run: bool = False, file_recall: bool = False) -> Outcome: 

320 label = f"{self.label} hooks" 

321 wanted = self._wanted(prefix, file_recall) 

322 try: 

323 current = self.hook_commands() 

324 # Every hook of ours, so switching file recall off removes it. 

325 unchanged = self._all_ours() == {e: entry["command"] for e, entry in wanted.items()} 

326 except HostConfigError as e: 

327 return Outcome(label, "failed", str(e)) 

328 if unchanged: 

329 return Outcome(label, "unchanged", str(self.hooks_file())) 

330 verb: Status = "updated" if any(current.values()) else "installed" 

331 if dry_run: 

332 return Outcome(label, "planned", f"would {_VERB[verb]} {self.hooks_file()}") 

333 matcher = self._matchers() 

334 if file_recall and self.FILE_HOOK: 

335 matcher[self.FILE_HOOK[0]] = self.FILE_HOOK[1] 

336 

337 def change(data: dict) -> None: 

338 self._strip_ours(data) # replace stale entries; never touch the user's own hooks 

339 hooks = data.setdefault("hooks", {}) 

340 for event, entry in wanted.items(): 

341 group = {"matcher": matcher[event], "hooks": [entry]} if event in matcher else {"hooks": [entry]} 

342 hooks.setdefault(event, []).append(group) 

343 

344 try: 

345 edit_json(self.hooks_file(), change) 

346 after = self._all_ours() 

347 except (HostConfigError, OSError) as e: 

348 return Outcome(label, "failed", str(e)) 

349 if any(after.get(e) != wanted[e]["command"] for e in wanted): 

350 return Outcome(label, "failed", f"wrote {self.hooks_file()} but the hooks did not take effect") 

351 return Outcome(label, verb, str(self.hooks_file())) 

352 

353 def uninstall_hooks(self, *, dry_run: bool = False) -> Outcome: 

354 label = f"{self.label} hooks" 

355 try: 

356 if not self._all_ours(): 

357 return Outcome(label, "absent", str(self.hooks_file())) 

358 if dry_run: 

359 return Outcome(label, "planned", f"would remove from {self.hooks_file()}") 

360 edit_json(self.hooks_file(), self._strip_ours) 

361 except (HostConfigError, OSError) as e: 

362 return Outcome(label, "failed", str(e)) 

363 return Outcome(label, "removed", str(self.hooks_file())) 

364 

365 

366class ClaudeCodeHost(HookHost, _CliHost): 

367 key, label, binary = "claude", "Claude Code", "claude" 

368 

369 def _dir(self) -> Path: 

370 return Path(os.environ["CLAUDE_CONFIG_DIR"]) if os.environ.get("CLAUDE_CONFIG_DIR") else _home() / ".claude" 

371 

372 def detected(self) -> bool: 

373 return self._dir().is_dir() or shutil.which(self.binary) is not None 

374 

375 def config_file(self) -> Path: 

376 # User-scope servers live in .claude.json beside (or, with 

377 # CLAUDE_CONFIG_DIR, inside) the config directory. 

378 if os.environ.get("CLAUDE_CONFIG_DIR"): 

379 return Path(os.environ["CLAUDE_CONFIG_DIR"]) / ".claude.json" 

380 return _home() / ".claude.json" 

381 

382 def registration(self) -> Registration | None: 

383 return _registration_from(_read_json(self.config_file()).get("mcpServers", {}).get(SERVER_NAME)) 

384 

385 def _write(self, spec: ServerSpec) -> None: 

386 # `claude mcp add` will not replace an existing name (it prints 

387 # "already exists" and exits 0), so remove first. 

388 self._remove() 

389 self._run("mcp", "add", "--scope", "user", SERVER_NAME, "--", spec.command, *spec.args) 

390 

391 def _remove(self) -> None: 

392 self._run("mcp", "remove", "--scope", "user", SERVER_NAME, check=False) # exits 1 when absent 

393 

394 def manual_command(self, spec: ServerSpec | None = None) -> str: 

395 tail = f"{spec.command} {' '.join(spec.args)}" if spec else "<astrocyte-mcp> --config <path>" 

396 return f"claude mcp add --scope user {SERVER_NAME} -- {tail}" 

397 

398 #: Stop is deliberately *not* async: `claude -p` exits as soon as it prints 

399 #: its answer and takes background hooks with it, so an async capture never 

400 #: ran in headless use (observed). The hook only spools the turn — ~0.35 s, 

401 #: after the response is already on screen. 

402 HOOK_EVENTS = { 

403 "SessionStart": ("session-start", {"timeout": 15}), 

404 "UserPromptSubmit": ("prompt", {"timeout": 5}), 

405 "Stop": ("stop", {"timeout": 10}), 

406 } 

407 #: After a file tool runs, its memories are added next to the result 

408 #: (``hookSpecificOutput.additionalContext``; code.claude.com/docs/en/hooks). 

409 FILE_HOOK = ("PostToolUse", "Read|Edit|Write|NotebookEdit", {"timeout": 5}) 

410 

411 def hooks_file(self) -> Path: 

412 return self._dir() / "settings.json" 

413 

414 

415 

416 

417 

418class CodexHost(HookHost, _CliHost): 

419 key, label, binary = "codex", "Codex CLI", "codex" 

420 

421 def _dir(self) -> Path: 

422 return Path(os.environ["CODEX_HOME"]) if os.environ.get("CODEX_HOME") else _home() / ".codex" 

423 

424 def detected(self) -> bool: 

425 return self._dir().is_dir() or shutil.which(self.binary) is not None 

426 

427 def config_file(self) -> Path: 

428 return self._dir() / "config.toml" 

429 

430 def registration(self) -> Registration | None: 

431 path = self.config_file() 

432 if not path.exists(): 

433 return None 

434 try: 

435 data = tomllib.loads(path.read_text(encoding="utf-8")) 

436 except tomllib.TOMLDecodeError as e: 

437 raise HostConfigError(f"{path} is not valid TOML ({e}); fix it by hand, then re-run") from e 

438 return _registration_from(data.get("mcp_servers", {}).get(SERVER_NAME)) 

439 

440 def _write(self, spec: ServerSpec) -> None: 

441 self._remove() 

442 self._run("mcp", "add", SERVER_NAME, "--", spec.command, *spec.args) 

443 

444 def _remove(self) -> None: 

445 self._run("mcp", "remove", SERVER_NAME, check=False) 

446 

447 def manual_command(self, spec: ServerSpec | None = None) -> str: 

448 tail = f"{spec.command} {' '.join(spec.args)}" if spec else "<astrocyte-mcp> --config <path>" 

449 return f"codex mcp add {SERVER_NAME} -- {tail}" 

450 

451 # Codex speaks Claude Code's hook contract (same events, same 

452 # ``hookSpecificOutput.additionalContext``; verified on codex-cli 0.160), 

453 # but its Stop payload carries the reply as ``last_assistant_message`` 

454 # rather than a transcript we can parse — hence its own dialect — and 

455 # names no files, so PostToolUse on apply_patch (its one file-writing 

456 # tool; reads go through the shell) notes them. Timeouts are seconds. 

457 HOOK_EVENTS = { 

458 "SessionStart": ("session-start", {"timeout": 15}), 

459 "UserPromptSubmit": ("prompt", {"timeout": 5}), 

460 "Stop": ("stop", {"timeout": 10}), 

461 } 

462 TOOL_HOOKS = {"PostToolUse": ("edit", "apply_patch", {"timeout": 5})} 

463 hook_dialect = "codex" 

464 # The released value (public API); hooks_hint() is what setup prints. 

465 hooks_next_step = ( 

466 "Codex runs new or changed hooks only once you trust them: in Codex, run /hooks and trust " 

467 "the three astrocyte hooks." 

468 ) 

469 

470 def hooks_hint(self) -> str: 

471 return ("Codex runs new or changed hooks only once you trust them: in Codex, run /hooks and trust " 

472 "the four astrocyte hooks.") 

473 

474 # Codex reads hooks from hooks.json and from config.toml, and warns on 

475 # every run when both define some ("prefer a single representation"). So 

476 # ours go where the user's hooks already are: hooks.json only if it holds 

477 # hooks of theirs, otherwise config.toml — between marker comments, the 

478 # convention other tools writing that file use, so it can be removed 

479 # exactly. Every TOML edit is verified structurally: the file must parse 

480 # to the old contents plus (or minus) our entries, and nothing else. 

481 _BEGIN = "# >>> astrocyte hooks >>>" 

482 _END = "# <<< astrocyte hooks <<<" 

483 

484 def _json_file(self) -> Path: 

485 return self._dir() / "hooks.json" 

486 

487 def _toml(self) -> tuple[str, dict[str, Any]]: 

488 path = self.config_file() 

489 text = path.read_text(encoding="utf-8") if path.exists() else "" 

490 try: 

491 return text, tomllib.loads(text) 

492 except tomllib.TOMLDecodeError as e: 

493 raise HostConfigError(f"{path} is not valid TOML ({e}); fix it by hand, then re-run") from e 

494 

495 @staticmethod 

496 def _entries(data: dict[str, Any]) -> list[Any]: 

497 hooks = data.get("hooks") 

498 if not isinstance(hooks, dict): 

499 return [] 

500 return [h for groups in hooks.values() if isinstance(groups, list) 

501 for group in groups for h in (group or {}).get("hooks") or []] 

502 

503 def _has_foreign_hooks(self, data: dict[str, Any]) -> bool: 

504 return any(not self._is_ours(h) for h in self._entries(data)) 

505 

506 def _holds_ours(self, data: dict[str, Any]) -> bool: 

507 return any(self._is_ours(h) for h in self._entries(data)) 

508 

509 def hooks_file(self) -> Path: 

510 """Where our hooks are, or would be installed.""" 

511 return self._json_file() if self._has_foreign_hooks(_read_json(self._json_file())) else self.config_file() 

512 

513 def hook_commands(self) -> dict[str, str | None]: 

514 found: dict[str, str | None] = dict.fromkeys(self._events()) 

515 for data in (_read_json(self._json_file()), self._toml()[1]): 

516 hooks = data.get("hooks") 

517 if not isinstance(hooks, dict): 

518 continue 

519 for event in self._events(): 

520 for group in hooks.get(event) or []: 

521 for hook in (group or {}).get("hooks") or []: 

522 if self._is_ours(hook): 

523 found[event] = hook["command"] 

524 return found 

525 

526 def _unmarked(self, text: str) -> str: 

527 """``text`` without our marked block (and the blank line before it).""" 

528 if self._BEGIN not in text: 

529 return text 

530 head, _, rest = text.partition(self._BEGIN) 

531 _, sep, tail = rest.partition(self._END) 

532 if not sep: 

533 raise HostConfigError(f"{self.config_file()}: '{self._BEGIN}' has no closing marker; fix it by hand") 

534 return head.rstrip("\n") + ("\n" if head.strip() else "") + tail.lstrip("\n") 

535 

536 def _block(self, wanted: dict[str, dict[str, Any]]) -> str: 

537 lines = [self._BEGIN, "# Automatic memory, managed by `astrocyte setup` (remove: astrocyte setup --no-hooks)."] 

538 for event, entry in wanted.items(): 

539 lines += ["", f"[[hooks.{event}]]"] 

540 if event in self._matchers(): 

541 lines.append(f"matcher = {json.dumps(self._matchers()[event])}") 

542 lines += ["", f"[[hooks.{event}.hooks]]"] 

543 # JSON string escapes are valid TOML basic-string escapes. 

544 lines += [f"{k} = {json.dumps(v, ensure_ascii=False)}" for k, v in entry.items()] 

545 return "\n".join([*lines, self._END]) + "\n" 

546 

547 def _write_toml(self, text: str, expected: dict[str, Any]) -> None: 

548 path = self.config_file() 

549 try: 

550 after = tomllib.loads(text) 

551 except tomllib.TOMLDecodeError as e: 

552 after = {"<unparseable>": str(e)} 

553 if after != expected: 

554 raise HostConfigError( 

555 f"refusing to edit {path}: the change would alter more than Astrocyte's hooks " 

556 "(hand-edited marker block, or hook events written as inline arrays?). " 

557 "Fix it by hand, or move your hooks into ~/.codex/hooks.json, then re-run" 

558 ) 

559 path.parent.mkdir(parents=True, exist_ok=True) 

560 if path.exists(): 

561 backup = path.with_name(path.name + ".astrocyte-bak") 

562 if not backup.exists(): 

563 shutil.copy2(path, backup) 

564 tmp = path.with_name(path.name + ".astrocyte-tmp") 

565 tmp.write_text(text, encoding="utf-8") 

566 if path.exists(): 

567 shutil.copymode(path, tmp) 

568 os.replace(tmp, path) 

569 

570 def _strip_json(self) -> None: 

571 path = self._json_file() 

572 data = _read_json(path) 

573 if not self._holds_ours(data): 

574 return 

575 self._strip_ours(data) 

576 if data == {}: 

577 path.unlink() # held nothing but ours; even an empty file is a second hooks source 

578 else: 

579 edit_json(path, self._strip_ours) 

580 

581 def _strip_toml(self) -> None: 

582 text, data = self._toml() 

583 if not self._holds_ours(data): 

584 return 

585 expected = copy.deepcopy(data) 

586 self._strip_ours(expected) 

587 self._write_toml(self._unmarked(text), expected) 

588 

589 def install_hooks(self, prefix: str, *, dry_run: bool = False) -> Outcome: 

590 label = f"{self.label} hooks" 

591 wanted = self._wanted(prefix) 

592 try: 

593 current = self.hook_commands() 

594 target = self.hooks_file() 

595 text, data = self._toml() 

596 # Where ours are now; "unchanged" also requires the right place. 

597 held = {self._json_file()} if self._holds_ours(_read_json(self._json_file())) else set() 

598 if self._holds_ours(data): 

599 held.add(self.config_file()) 

600 settled = held == {target} and (target == self._json_file() or self._BEGIN in text) 

601 except HostConfigError as e: 

602 return Outcome(label, "failed", str(e)) 

603 if settled and all(current[e] == wanted[e]["command"] for e in wanted): 

604 return Outcome(label, "unchanged", str(target)) 

605 verb: Status = "updated" if any(current.values()) else "installed" 

606 if dry_run: 

607 return Outcome(label, "planned", f"would {_VERB[verb]} {target}") 

608 try: 

609 if target == self._json_file(): 

610 self._strip_toml() 

611 return super().install_hooks(prefix) # hooks.json: the shared JSON installer 

612 self._strip_json() 

613 expected = copy.deepcopy(data) 

614 self._strip_ours(expected) 

615 hooks = expected.setdefault("hooks", {}) 

616 for event, entry in wanted.items(): 

617 group = {"matcher": self._matchers()[event]} if event in self._matchers() else {} 

618 hooks.setdefault(event, []).append({**group, "hooks": [entry]}) 

619 body = self._unmarked(text) 

620 body = body.rstrip("\n") + "\n\n" if body.strip() else "" 

621 self._write_toml(body + self._block(wanted), expected) 

622 after = self.hook_commands() 

623 except (HostConfigError, OSError) as e: 

624 return Outcome(label, "failed", str(e)) 

625 if any(after[e] != wanted[e]["command"] for e in wanted): 

626 return Outcome(label, "failed", f"wrote {target} but the hooks did not take effect") 

627 return Outcome(label, verb, str(target)) 

628 

629 def uninstall_hooks(self, *, dry_run: bool = False) -> Outcome: 

630 label = f"{self.label} hooks" 

631 try: 

632 if not any(self.hook_commands().values()): 

633 return Outcome(label, "absent", str(self.hooks_file())) 

634 if dry_run: 

635 return Outcome(label, "planned", f"would remove from {self.hooks_file()}") 

636 self._strip_json() 

637 self._strip_toml() 

638 except (HostConfigError, OSError) as e: 

639 return Outcome(label, "failed", str(e)) 

640 return Outcome(label, "removed", str(self.hooks_file())) 

641 

642 

643class GeminiHost(_CliHost): 

644 key, label, binary = "gemini", "Gemini CLI", "gemini" 

645 next_step = "Gemini CLI only loads MCP servers in trusted folders: trust your project folder when prompted." 

646 

647 def detected(self) -> bool: 

648 return (_home() / ".gemini").is_dir() or shutil.which(self.binary) is not None