Coverage for src / osiris_cli / recipes.py: 0%
87 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-27 17:41 +0200
1from __future__ import annotations
3import asyncio
4import json
5from dataclasses import dataclass
6from datetime import datetime
7from pathlib import Path
8from typing import Dict, List, Optional, Any
10import os
12from .config import CONFIG_DIR
13from .mcp import mcp_client
14from .tools import tools
15from .utils import log_activity
18RECIPE_LIBRARY = {
19 "diagnose": {
20 "description": "Gather repo status and recent history for identifying flaky tests or failing builds.",
21 "steps": [
22 {"type": "shell", "name": "Git status", "command": "git status -sb"},
23 {"type": "shell", "name": "Recent commits", "command": "git log -5 --oneline"},
24 {"type": "shell", "name": "Open ports (light check)", "command": "ss -tlnp | head"},
25 ],
26 },
27 "deploy": {
28 "description": "Prepare a deployment readiness summary, run quick sanity checks, and capture diffs.",
29 "steps": [
30 {"type": "shell", "name": "Workspace diff", "command": "git status -sb"},
31 {"type": "shell", "name": "Diff stats", "command": "git diff --stat"},
32 {"type": "shell", "name": "List docker services", "command": "docker ps --format '{{.Names}} ({{.Status}})'"},
33 ],
34 },
35 "refactor": {
36 "description": "Plan and surface files that need interface updates.",
37 "steps": [
38 {"type": "shell", "name": "Code tree", "command": "find . -maxdepth 2 -type f | head"},
39 {"type": "shell", "name": "Large files", "command": "du -h . | sort -hr | head"},
40 {"type": "shell", "name": "Prepare diff", "command": "git diff HEAD~1 --stat || true"},
41 ],
42 },
43 "triage": {
44 "description": "GitHub triage: collect open issue counts for osiris + gather recent issue titles.",
45 "steps": [
46 {
47 "type": "mcp",
48 "name": "Query issue counts",
49 "server": "github",
50 "payload": {
51 "query": """
52 query IssueCounts {
53 repository(owner: "TheOsirisLabs", name: "osiris") {
54 issues(states: OPEN) { totalCount }
55 pullRequests(states: OPEN) { totalCount }
56 }
57 }
58 """
59 },
60 },
61 {
62 "type": "mcp",
63 "name": "Recent open issues list",
64 "server": "github",
65 "payload": {
66 "query": """
67 query OpenIssues {
68 repository(owner: "TheOsirisLabs", name: "osiris") {
69 issues(states: OPEN, last: 5, orderBy: {field: CREATED_AT, direction: DESC}) {
70 nodes { number title url }
71 }
72 }
73 }
74 """
75 },
76 },
77 ],
78 },
79 "db_health": {
80 "description": "Query Postgres via MCP for version and uptime.",
81 "steps": [
82 {
83 "type": "mcp",
84 "name": "Postgres version",
85 "server": "postgres",
86 "payload": {"sql": "SELECT version();"},
87 },
88 {
89 "type": "mcp",
90 "name": "Recent 5 tables",
91 "server": "postgres",
92 "payload": {
93 "sql": "SELECT table_name FROM information_schema.tables WHERE table_schema='public' LIMIT 5;"
94 },
95 },
96 ],
97 },
98 "ui_check": {
99 "description": "Browser MCP check that ensures a page is reachable and matches title.",
100 "steps": [
101 {
102 "type": "mcp",
103 "name": "Check UI status",
104 "server": "browser",
105 "payload": {
106 "url": os.getenv("BROWSER_MCP_URL", "https://example.com")
107 },
108 },
109 ],
110 },
111 "ui_journey": {
112 "description": "Playwright MCP journey: visit page, wait, capture screenshot metadata.",
113 "steps": [
114 {
115 "type": "mcp",
116 "name": "Playwright visit",
117 "server": "playwright",
118 "payload": {"steps": [{"action": "wait", "seconds": 2}, {"action": "screenshot"}]},
119 },
120 ],
121 },
122 "deploy_ready": {
123 "description": "Composite deployment readiness check (GitHub + Postgres + UI + Slack note).",
124 "steps": [
125 {
126 "type": "mcp",
127 "name": "Issue/pull count quick check",
128 "server": "github",
129 "payload": {
130 "query": """
131 query DeployCounts {
132 repository(owner: "TheOsirisLabs", name: "osiris") {
133 issues(states: OPEN) { totalCount }
134 pullRequests(states: OPEN) { totalCount }
135 }
136 }
137 """
138 },
139 },
140 {
141 "type": "mcp",
142 "name": "DB version note",
143 "server": "postgres",
144 "payload": {"sql": "SELECT version();"},
145 },
146 {
147 "type": "mcp",
148 "name": "UI smoke test",
149 "server": "browser",
150 "payload": {"url": os.getenv("BROWSER_MCP_URL", "https://example.com")},
151 },
152 {
153 "type": "mcp",
154 "name": "Slack deployment notice",
155 "server": "slack",
156 "payload": {
157 "url": os.getenv("SLACK_WEBHOOK_URL", ""),
158 "text": "Osiris deployment readiness check completed. Review /activity or ~/.osiris/recipes/deploy_ready_* for details.",
159 },
160 },
161 ],
162 },
163 "fs_inspect": {
164 "description": "Filesystem MCP inspection of the configured root.",
165 "steps": [
166 {
167 "type": "mcp",
168 "name": "List root contents",
169 "server": "filesystem",
170 "payload": {"action": "list", "path": "."},
171 },
172 {
173 "type": "mcp",
174 "name": "Read README (if present)",
175 "server": "filesystem",
176 "payload": {"action": "read", "path": "README.md"},
177 },
178 ],
179 },
180 "monitor_report": {
181 "description": "Send a monitoring heartbeat summarizing last recipes and status.",
182 "steps": [
183 {
184 "type": "mcp",
185 "name": "Send monitoring payload",
186 "server": "monitoring",
187 "payload": {
188 "status": "ok",
189 "details": {
190 "last_recipe": os.getenv("LAST_RECIPE", "unknown"),
191 "provider": os.getenv("OSIRIS_PROVIDER", "unknown"),
192 },
193 },
194 },
195 ],
196 },
197}
200@dataclass
201class RecipeResult:
202 name: str
203 command: str
204 output: str
207class RecipeRunner:
208 def __init__(self):
209 self.registry = tools.tools
210 self.recipe_dir = CONFIG_DIR / "recipes"
212 def list_recipes(self) -> List[str]:
213 return list(RECIPE_LIBRARY.keys())
215 def describe(self, recipe_id: str) -> Optional[str]:
216 recipe = RECIPE_LIBRARY.get(recipe_id)
217 if not recipe:
218 return None
219 return recipe.get("description")
221 def run(self, recipe_id: str, dry_run: bool = False) -> List[RecipeResult]:
222 recipe = RECIPE_LIBRARY.get(recipe_id)
223 if not recipe:
224 raise ValueError(f"No recipe named {recipe_id}")
225 results: List[RecipeResult] = []
226 for step in recipe["steps"]:
227 name = step["name"]
228 step_type = step.get("type", "shell")
229 if step_type == "mcp":
230 command_desc = f"mcp:{step.get('server', 'unknown')}"
231 output = "[DRY RUN] MCP call skipped." if dry_run else self._run_mcp(step)
232 else:
233 command_desc = step.get("command", "[unknown]")
234 output = "[DRY RUN] Command not executed." if dry_run else self._run_shell(command_desc)
235 results.append(RecipeResult(name=name, command=command_desc, output=output))
236 self._persist(recipe_id, results)
237 return results
239 def history(self, recipe_id: Optional[str] = None) -> List[Path]:
240 if not self.recipe_dir.exists():
241 return []
242 files = sorted(self.recipe_dir.glob("*.md"), reverse=True)
243 if recipe_id:
244 files = [f for f in files if f.stem.startswith(recipe_id)]
245 return files
247 def _run_shell(self, command: str) -> str:
248 impl = self.registry.get("run_shell_command", {}).get("implementation")
249 if not impl:
250 return "run_shell_command not registered"
251 try:
252 result = impl(command=command)
253 log_activity("recipe_step", {"command": command, "result_preview": str(result)[:200]})
254 return result
255 except Exception as exc: # pragma: no cover
256 log_activity("recipe_error", {"command": command, "error": str(exc)})
257 return f"ERROR: {exc}"
259 def _run_mcp(self, step: Dict[str, Any]) -> str:
260 server = step.get("server")
261 payload = step.get("payload", {})
262 if not server:
263 log_activity("recipe_error", {"step": step, "error": "missing server"})
264 return "ERROR: missing MCP server name"
265 try:
266 result = asyncio.run(mcp_client.call(server, payload))
267 log_activity("recipe_mcp", {"server": server, "payload": payload, "result": result})
268 return json.dumps(result, ensure_ascii=False)
269 except Exception as exc:
270 log_activity("recipe_mcp_error", {"server": server, "payload": payload, "error": str(exc)})
271 return f"ERROR: {exc}"
273 def _persist(self, recipe_id: str, results: List[RecipeResult]) -> None:
274 self.recipe_dir.mkdir(parents=True, exist_ok=True)
275 timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
276 path = self.recipe_dir / f"{recipe_id}_{timestamp}.md"
277 lines = [f"# Recipe {recipe_id}", f"run: {timestamp} UTC", ""]
278 for result in results:
279 lines.append(f"## {result.name}")
280 lines.append(f"- **Command:** `{result.command}`")
281 lines.append("```")
282 lines.append(result.output.strip())
283 lines.append("```")
284 lines.append("")
285 path.write_text("\n".join(lines))
286 log_activity("recipe_complete", {"recipe": recipe_id, "path": str(path)})