Coverage for astrocyte/cli.py: 95%

1"""Astrocyte CLI — local setup and operator tooling. 

2 

3Commands: 

4 astrocyte setup [--claude ...] Create a local memory store and wire it into coding agents. 

5 astrocyte doctor [--fix] Check the local install end to end; repair what it can. 

6 astrocyte uninstall [--claude ...] Remove Astrocyte from agent harnesses (keeps memories). 

7 astrocyte memory [search|forget|banks] See, search and remove what agents remember (this project). 

8 astrocyte mip lint <path> Validate a mip.yaml, printing errors and warnings. 

9 astrocyte mip explain <path> ... Show which rule fires for a hypothetical input. 

10 

11The CLI is intentionally minimal: argparse-based, no external dependencies, and 

12reuses the same loader / router code that runtime uses. It is the recommended 

13way to verify pipeline overrides (chunker, dedup, rerank, reflect) before 

14deploying a rule change. 

15""" 

16 

17from __future__ import annotations 

18 

19import argparse 

20import dataclasses 

21import os 

22import sys 

23import warnings 

24from pathlib import Path 

25from typing import TYPE_CHECKING, Sequence 

26 

27from astrocyte.errors import ConfigError 

28 

29# The `mip` commands import routing where they run: every agent hook enters 

30# through this module (`python -m astrocyte.cli hook …`) and must not pay for it. 

31if TYPE_CHECKING: 

32 from astrocyte.mip.schema import ForgetSpec, PipelineSpec 

33 from astrocyte.types import MetadataValue 

34 

35# --------------------------------------------------------------------------- 

36# astrocyte mip lint 

37# --------------------------------------------------------------------------- 

38 

39 

40def _cmd_mip_lint(args: argparse.Namespace) -> int: 

41 """Load and validate a mip.yaml. Returns 0 if clean, 1 on any error.""" 

42 from astrocyte.mip import load_mip_config 

43 

44 path = Path(args.path) 

45 print(f"Linting MIP config: {path}") 

46 

47 captured: list[warnings.WarningMessage] = [] 

48 try: 

49 with warnings.catch_warnings(record=True) as caught: 

50 warnings.simplefilter("always") 

51 config = load_mip_config(path) 

52 captured = list(caught) 

53 except ConfigError as exc: 

54 print(f" error: {exc}", file=sys.stderr) 

55 return 1 

56 except FileNotFoundError as exc: 

57 print(f" error: {exc}", file=sys.stderr) 

58 return 1 

59 

60 rule_count = len(config.rules or []) 

61 bank_count = len(config.banks or []) 

62 print(f" ok: {rule_count} rule(s), {bank_count} bank(s)") 

63 

64 if captured: 

65 print(f" {len(captured)} warning(s):") 

66 for w in captured: 

67 print(f" - {w.message}") 

68 return 0 

69 

70 

71# --------------------------------------------------------------------------- 

72# astrocyte mip explain 

73# --------------------------------------------------------------------------- 

74 

75 

76def _parse_kv(items: Sequence[str] | None) -> dict[str, MetadataValue]: 

77 """Parse a list of ``key=value`` strings into a metadata dict. 

78 

79 Numeric and boolean values are coerced; everything else stays as a string. 

80 """ 

81 out: dict[str, MetadataValue] = {} 

82 if not items: 

83 return out 

84 for raw in items: 

85 if "=" not in raw: 

86 raise SystemExit(f"Invalid --metadata entry (expected KEY=VALUE): {raw!r}") 

87 key, _, val = raw.partition("=") 

88 key = key.strip() 

89 val = val.strip() 

90 coerced: MetadataValue 

91 if val.lower() in ("true", "false"): 

92 coerced = val.lower() == "true" 

93 else: 

94 try: 

95 coerced = int(val) 

96 except ValueError: 

97 try: 

98 coerced = float(val) 

99 except ValueError: 

100 coerced = val 

101 out[key] = coerced 

102 return out 

103 

104 

105def _format_pipeline(pipeline: PipelineSpec | None) -> list[str]: 

106 """Render a PipelineSpec as a list of indented printable lines.""" 

107 if pipeline is None: 

108 return [] 

109 lines: list[str] = [" pipeline:"] 

110 if pipeline.version is not None: 

111 lines.append(f" version: {pipeline.version}") 

112 for field in ("chunker", "dedup", "rerank", "reflect"): 

113 spec = getattr(pipeline, field) 

114 if spec is None: 

115 continue 

116 # Show only fields that are set 

117 set_fields = { 

118 f.name: getattr(spec, f.name) for f in dataclasses.fields(spec) if getattr(spec, f.name) is not None 

119 } 

120 if set_fields: 

121 lines.append(f" {field}: {set_fields}") 

122 return lines 

123 

124 

125def _format_forget(forget: ForgetSpec | None) -> list[str]: 

126 """Render a ForgetSpec as a list of indented printable lines.""" 

127 if forget is None: 

128 return [] 

129 lines: list[str] = [" forget:"] 

130 set_fields = { 

131 f.name: getattr(forget, f.name) for f in dataclasses.fields(forget) if getattr(forget, f.name) is not None 

132 } 

133 for k, v in set_fields.items(): 

134 lines.append(f" {k}: {v}") 

135 return lines 

136 

137 

138def _cmd_mip_explain(args: argparse.Namespace) -> int: 

139 """Show which rule(s) match a hypothetical input and the resulting decision.""" 

140 from astrocyte.mip import MipRouter, load_mip_config 

141 from astrocyte.mip.rule_engine import RuleEngineInput, evaluate_rules 

142 

143 path = Path(args.path) 

144 try: 

145 config = load_mip_config(path) 

146 except (ConfigError, FileNotFoundError) as exc: 

147 print(f"error: {exc}", file=sys.stderr) 

148 return 1 

149 

150 metadata = _parse_kv(args.metadata) 

151 rule_input = RuleEngineInput( 

152 content=args.content or "", 

153 content_type=args.content_type, 

154 metadata=metadata or None, 

155 tags=list(args.tag) if args.tag else None, 

156 pii_detected=args.pii_detected, 

157 source=args.source, 

158 ) 

159 

160 print("Input:") 

161 print(f" content_type: {rule_input.content_type!r}") 

162 print(f" metadata: {rule_input.metadata}") 

163 print(f" tags: {rule_input.tags}") 

164 print(f" pii_detected: {rule_input.pii_detected}") 

165 print(f" source: {rule_input.source!r}") 

166 

167 # 1. Show every matching rule (mechanical eval, before escalation policy) 

168 sorted_rules = sorted(config.rules or [], key=lambda r: r.priority) 

169 matches = evaluate_rules(sorted_rules, rule_input) 

170 print(f"\nMatched rules ({len(matches)}):") 

171 if not matches: 

172 print(" (none)") 

173 for m in matches: 

174 print(f" - {m.rule.name} (priority={m.rule.priority}, override={m.rule.override}, confidence={m.confidence})") 

175 

176 # 2. Show what the synchronous router would actually return 

177 router = MipRouter(config) 

178 decision = router.route_sync(rule_input) 

179 print("\nDecision (sync routing):") 

180 if decision is None: 

181 print(" → escalation required (would call intent layer at runtime)") 

182 return 0 

183 

184 print(f" resolved_by: {decision.resolved_by}") 

185 print(f" rule_name: {decision.rule_name}") 

186 print(f" bank_id: {decision.bank_id}") 

187 print(f" tags: {decision.tags}") 

188 print(f" retain_policy: {decision.retain_policy}") 

189 print(f" confidence: {decision.confidence}") 

190 for line in _format_pipeline(decision.pipeline): 

191 print(line) 

192 for line in _format_forget(decision.forget): 

193 print(line) 

194 return 0 

195 

196 

197# --------------------------------------------------------------------------- 

198# argparse plumbing 

199# --------------------------------------------------------------------------- 

200 

201 

202def _build_parser() -> argparse.ArgumentParser: 

203 parser = argparse.ArgumentParser(prog="astrocyte", description="Astrocyte operator CLI") 

204 sub = parser.add_subparsers(dest="command", required=True) 

205 

206 mip = sub.add_parser("mip", help="MIP (Memory Intent Protocol) tools") 

207 mip_sub = mip.add_subparsers(dest="mip_command", required=True) 

208 

209 lint = mip_sub.add_parser("lint", help="Validate a mip.yaml file") 

210 lint.add_argument("path", help="Path to mip.yaml") 

211 lint.set_defaults(func=_cmd_mip_lint) 

212 

213 explain = mip_sub.add_parser( 

214 "explain", 

215 help="Show which rule fires for a hypothetical retain input", 

216 ) 

217 explain.add_argument("path", help="Path to mip.yaml") 

218 explain.add_argument("--content", default="", help="Inbound content (text body)") 

219 explain.add_argument("--content-type", default=None, help="Content type, e.g. text|conversation|document") 

220 explain.add_argument("--metadata", action="append", default=[], help="Metadata KEY=VALUE (repeatable)") 

221 explain.add_argument("--tag", action="append", default=[], help="Tag (repeatable)") 

222 explain.add_argument("--source", default=None, help="Source identifier") 

223 explain.add_argument("--pii-detected", action="store_true", help="Mark PII as already detected") 

224 explain.set_defaults(func=_cmd_mip_explain) 

225 

226 from astrocyte.harness.commands import register as register_harness_commands 

227 

228 register_harness_commands(sub) 

229 return parser 

230 

231 

232def _utf8_stdio(windows: bool = os.name == "nt") -> None: 

233 """Speak UTF-8 on Windows. Output piped there (an agent running the CLI, 

234 CI, a redirect) is encoded with the ANSI code page, cp1252, which has no 

235 ✓ or ─: every status line would raise UnicodeEncodeError. And hook 

236 payloads arrive on stdin as UTF-8 JSON.""" 

237 if not windows: 

238 return 

239 for stream in (sys.stdin, sys.stdout, sys.stderr): 

240 reconfigure = getattr(stream, "reconfigure", None) 

241 if reconfigure is not None: 

242 reconfigure(encoding="utf-8", errors="replace") 

243 

244 

245def main(argv: Sequence[str] | None = None) -> int: 

246 _utf8_stdio() 

247 parser = _build_parser() 

248 args = parser.parse_args(argv) 

249 return int(args.func(args) or 0) 

250 

251 

252if __name__ == "__main__": 

253 raise SystemExit(main())