feat: add github-actions-deploy solution

cortexapps_cli/commands/solutions.py

@@ -74,13 +74,26 @@ def getvalue(self) -> str:
 
 
 def _run_import_with_toggle(fn) -> str:
-    """Run fn() capturing stdout. On a TTY, Ctrl+o toggles live output."""
+    """Run fn() capturing stdout. On a TTY, Ctrl+o toggles live output.
+
+    If fn() raises for any reason, all buffered output is flushed to the
+    terminal before the exception propagates — no silent failures.
+    """
     real_stdout = sys.stdout
     capture = _ToggleableCapture(real_stdout)
 
     if not (_TTY_SUPPORT and sys.stdin.isatty()):
-        with contextlib.redirect_stdout(capture):
-            fn()
+        success = False
+        try:
+            with contextlib.redirect_stdout(capture):
+                fn()
+            success = True
+        finally:
+            if not success:
+                buffered = capture.getvalue()
+                if buffered:
+                    real_stdout.write(buffered)
+                    real_stdout.flush()
         return capture.getvalue()
 
     done = threading.Event()
@@ -112,16 +125,23 @@ def _listen() -> None:
 
     t = threading.Thread(target=_listen, daemon=True)
     t.start()
+    success = False
     try:
         with contextlib.redirect_stdout(capture):
             fn()
+        success = True
     finally:
         done.set()
         t.join(timeout=0.5)
         try:
             _termios.tcsetattr(fd, _termios.TCSADRAIN, old_settings)
         except Exception:
             pass
+        if not success:
+            buffered = capture.getvalue()
+            if buffered:
+                real_stdout.write(buffered)
+                real_stdout.flush()
 
     return capture.getvalue()
 
@@ -175,6 +195,53 @@ def _get_readme(tag: str, path: str | None = None) -> str | None:
         return None
 
 
+def _has_post_install(tag: str, path: str | None = None) -> bool:
+    """Return True if this solution has a post-install setup.py."""
+    try:
+        (_solutions_root(path) / tag / "setup.py").read_bytes()
+        return True
+    except Exception:
+        return False
+
+
+def _load_setup_module(solution_tag: str, solutions_dir: str | None = None):
+    """Load a solution's setup.py module. Returns the module or None if not found."""
+    import importlib.util
+
+    root = _solutions_root(solutions_dir)
+    try:
+        with as_file(root / solution_tag / "setup.py") as setup_path:
+            spec = importlib.util.spec_from_file_location(f"{solution_tag}_setup", setup_path)
+            module = importlib.util.module_from_spec(spec)
+            spec.loader.exec_module(module)
+            return module
+    except FileNotFoundError:
+        return None
+
+
+def _get_setup_description(solution_tag: str, solutions_dir: str | None = None) -> str:
+    """Return the SETUP_DESCRIPTION from a solution's setup.py, or a generic fallback."""
+    module = _load_setup_module(solution_tag, solutions_dir)
+    if module:
+        return getattr(module, "SETUP_DESCRIPTION", "This solution includes a post-install setup script.")
+    return "This solution includes a post-install setup script."
+
+
+def _run_post_install_script(solution_tag: str, solutions_dir: str | None = None, ctx=None, no_prompt: bool = False) -> None:
+    """Find and invoke the solution's setup.py main() function."""
+    module = _load_setup_module(solution_tag, solutions_dir)
+    if module is None:
+        typer.echo("No post-install setup available for this solution.")
+        return
+    kwargs = {}
+    if ctx and ctx.obj and "client" in ctx.obj:
+        client = ctx.obj["client"]
+        kwargs["cortex_api_key"] = client.api_key
+        kwargs["cortex_base_url"] = client.base_url
+    kwargs["no_prompt"] = no_prompt
+    module.main(**kwargs)
+
+
 def _extract_first_codeblock(text: str) -> str | None:
     """Return content of the first fenced code block."""
     m = re.search(r"```[^\n]*\n(.*?)```", text, re.DOTALL)
@@ -538,17 +605,6 @@ def _show_next_steps(readme: str) -> None:
             section = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'\1 (\2)', section)
         console.print()
         console.print(Markdown(section))
-    console.print()
-    console.print(
-        "[magenta]Planned for Q4 2027:[/magenta] CQL metadata traversal will enable scorecard rules "
-        "across relationship chains — for example, a Vulnerability Scorecard checking that no "
-        "deployed service-version has open Snyk issues:"
-    )
-    console.print(
-        "  [dim]entity.destinations(relationshipType = \"environments\", depth = 3)\n"
-        "    .filter((d) => d.type == \"service-version\")\n"
-        "    .all((sv) => sv.snyk.issues == 0)[/dim]"
-    )
 
 
 def _post_install_menu(
@@ -604,6 +660,12 @@ def install(
     ctx: typer.Context,
     solution: str = typer.Option(..., "--solution", "-s", help="Solution tag"),
     no_prompt: bool = typer.Option(False, "--no-prompt", help="Skip the post-install interactive menu"),
+    skip_post_install_setup: bool = typer.Option(
+        False,
+        "--skip-post-install-setup",
+        help="Skip the post-install setup script prompt",
+        is_flag=True,
+    ),
 ):
     """Install a solution."""
     solutions_dir = ctx.obj.get("solutions_dir") if ctx.obj else None
@@ -643,6 +705,20 @@ def _do_import() -> None:
     else:
         typer.echo(output)
 
+    # Post-install setup hook — runs before the informational menu
+    if not no_prompt and not skip_post_install_setup and _has_post_install(solution, solutions_dir):
+        state_file = Path.home() / ".cortex" / "solutions" / f"{solution}.json"
+        if state_file.exists():
+            typer.echo(f"\nRetrieving previous responses from: {state_file}")
+        desc = _get_setup_description(solution, solutions_dir)
+        typer.echo(f"\n{desc}")
+        if typer.confirm("Run setup now?", default=True):
+            _run_post_install_script(solution, solutions_dir=solutions_dir, ctx=ctx)
+        else:
+            typer.echo(f"\nRun setup later with: cortex solutions post-install -s {solution}")
+    elif skip_post_install_setup and _has_post_install(solution, solutions_dir):
+        typer.echo(f"\nRun setup later with: cortex solutions post-install -s {solution}")
+
     if not no_prompt:
         readme = _get_readme(solution, solutions_dir)
         if readme:
@@ -660,6 +736,27 @@ def _do_import() -> None:
             _post_install_menu(readme, import_report=output, entity_tags=entity_tags, ui_url=ui_url)
 
 
+@app.command(name="post-install")
+def post_install(
+    ctx: typer.Context,
+    solution: str = typer.Option(..., "--solution", "-s", help="Solution tag"),
+    no_prompt: bool = typer.Option(
+        False,
+        "--no-prompt",
+        "-N",
+        help="Use saved answers from ~/.cortex/solutions/<tag>.json without prompting.",
+    ),
+):
+    """Run post-install setup for a solution."""
+    solutions_dir = ctx.obj.get("solutions_dir") if ctx.obj else None
+    if solution not in _list_solution_tags(solutions_dir):
+        avail = ", ".join(_list_solution_tags(solutions_dir))
+        typer.echo(f"Error: Solution '{solution}' not found. Available: {avail}")
+        raise typer.Exit(1)
+    ctx.obj["client"] = _build_client(ctx)
+    _run_post_install_script(solution, solutions_dir=solutions_dir, ctx=ctx, no_prompt=no_prompt)
+
+
 @app.command()
 def uninstall(
     ctx: typer.Context,

cortexapps_cli/solutions/_lib/setup_base.py

@@ -0,0 +1,169 @@
+import getpass
+import json
+import os
+import sys
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Optional
+
+
+class SolutionSetup(ABC):
+    """
+    Base class for solution post-install setup scripts.
+    Subclasses define solution_tag, collect_prompts(), and steps().
+    """
+
+    solution_tag: str  # must be set by subclass
+
+    def __init__(self, state_dir: Optional[Path] = None, no_prompt: bool = False):
+        self._no_prompt = no_prompt
+        self._secret_keys: set = set()
+        self._answers: dict = {}
+
+        solutions_dir = state_dir or Path.home() / ".cortex" / "solutions"
+        solutions_dir.mkdir(parents=True, exist_ok=True)
+        self._state_file = solutions_dir / f"{self.solution_tag}.json"
+
+        data = self._load_file()
+        self._answers = data.get("answers", {})
+        self._state: dict = data.get("state", {})
+
+        self._migrate_old_state()
+
+    def _load_file(self) -> dict:
+        if self._state_file.exists():
+            try:
+                return json.loads(self._state_file.read_text())
+            except (json.JSONDecodeError, OSError):
+                return {}
+        return {}
+
+    def _save_file(self) -> None:
+        data = {
+            "answers": {k: v for k, v in self._answers.items() if k not in self._secret_keys},
+            "state": self._state,
+        }
+        self._state_file.write_text(json.dumps(data, indent=2))
+
+    def _save_state(self) -> None:
+        self._save_file()
+
+    def _migrate_old_state(self) -> None:
+        """Move state from the old flat ~/.cortex/setup-<tag>.json into the new file."""
+        old_file = Path.home() / ".cortex" / f"setup-{self.solution_tag}.json"
+        if not old_file.exists():
+            return
+        try:
+            old_data = json.loads(old_file.read_text())
+            if old_data and not self._state:
+                self._state.update(old_data)
+                self._save_file()
+            old_file.unlink()
+        except Exception:
+            pass
+
+    def prompt(
+        self,
+        key: str,
+        message: str,
+        env_var: Optional[str] = None,
+        default: Optional[str] = None,
+        secret: bool = False,
+    ) -> str:
+        """Prompt for a value. Uses saved answer or env var when available."""
+        if secret:
+            self._secret_keys.add(key)
+
+        # Non-secret: use saved answer when --no-prompt
+        if self._no_prompt and not secret and key in self._answers:
+            return self._answers[key]
+
+        # Non-secret: saved answer takes precedence over any derived default
+        if not secret and key in self._answers:
+            default = self._answers[key]
+
+        if env_var:
+            env_val = os.environ.get(env_var)
+            if env_val:
+                masked = "********" if secret else env_val
+                if self._no_prompt or self.confirm(f"{message} [{masked} from {env_var}]", default=True):
+                    self._answers[key] = env_val
+                    return env_val
+
+        # Secrets in --no-prompt mode still need a prompt if no env var provided
+        if self._no_prompt and secret and key not in self._answers:
+            print(f"  (secret required — no env var set for {key})", file=sys.stderr)
+
+        prompt_str = message
+        if default:
+            prompt_str += f" [{default}]"
+        prompt_str += ": "
+
+        if secret:
+            value = getpass.getpass(prompt_str).strip()
+        else:
+            value = input(prompt_str).strip()
+        if not value:
+            value = default or ""
+        self._answers[key] = value
+        return value
+
+    def confirm(self, message: str, default: bool = True) -> bool:
+        """Y|N confirmation prompt. Auto-accepts default when --no-prompt."""
+        if self._no_prompt:
+            return default
+        hint = "[Y/n]" if default else "[y/N]"
+        response = input(f"{message} {hint}: ").strip().lower()
+        if not response:
+            return default
+        return response in ("y", "yes")
+
+    def already_done(self, key: str) -> bool:
+        """Return True if this step was previously completed."""
+        return self._state.get(key, False)
+
+    def mark_done(self, key: str) -> None:
+        """Mark a step as completed in the persistent state file."""
+        self._state[key] = True
+        self._save_file()
+
+    @abstractmethod
+    def collect_prompts(self) -> None:
+        """Collect all user inputs upfront before executing steps."""
+
+    @abstractmethod
+    def steps(self) -> list[tuple[str, callable]]:
+        """Return ordered list of (label, callable) tuples."""
+
+    def post_steps(self) -> None:
+        """Optional hook called after all steps complete. Override in subclass."""
+
+    def run(self) -> None:
+        """Collect prompts then execute steps with progress display."""
+        if self._no_prompt and self._answers:
+            saved = {k: v for k, v in self._answers.items() if k not in self._secret_keys}
+            if saved:
+                print("Using saved configuration:")
+                for k, v in saved.items():
+                    print(f"  {k}: {v}")
+                print()
+
+        self.collect_prompts()
+        self._save_file()
+
+        print()
+        step_list = self.steps()
+        total = len(step_list)
+        for i, (label, fn) in enumerate(step_list, 1):
+            try:
+                detail = fn()
+                print(f"[{i}/{total}] {label}... \u2713")
+                if detail:
+                    lines = [detail] if isinstance(detail, str) else detail
+                    for line in lines:
+                        print(f"  {line}")
+                print()
+            except Exception as e:
+                print(f"[{i}/{total}] {label}... \u2717  {e}", file=sys.stderr)
+                raise SystemExit(1)
+        self.post_steps()