diff --git a/ifixai/cli/_branding.py b/ifixai/cli/_branding.py new file mode 100644 index 00000000..c34a936f --- /dev/null +++ b/ifixai/cli/_branding.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import os +import sys +import threading +import time +from dataclasses import dataclass, field +from typing import Iterable + +import click + +from ifixai.types import InspectionCategory + + +_SPINNER_FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" + + +class _Spinner: + def __init__(self, message: str) -> None: + self._message = message + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + self._frame = 0 + self._start_time = 0.0 + + def start(self) -> None: + if self._thread is not None: + return + self._start_time = time.monotonic() + sys.stdout.write(self._render() + "\n") + sys.stdout.flush() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def stop(self) -> None: + if self._thread is None: + return + self._stop_event.set() + self._thread.join(timeout=0.5) + self._thread = None + + def _render(self) -> str: + glyph = _SPINNER_FRAMES[self._frame % len(_SPINNER_FRAMES)] + elapsed = int(time.monotonic() - self._start_time) if self._start_time else 0 + suffix = f" ({elapsed}s)" if elapsed >= 3 else "" + return _truecolor(f" {glyph} {self._message}{suffix}", _DIM_RGB) + + def _run(self) -> None: + while not self._stop_event.wait(0.12): + self._frame += 1 + sys.stdout.write("\033[1F\033[2K" + self._render() + "\n") + sys.stdout.flush() + + +_LOGO_LINES: tuple[str, ...] = ( + "██ ███████ ██ ██ ██ █████ ██", + "██ ██ ██ ██ ██ ██ ██ ██", + "██ █████ ██ ███ ███████ ██", + "██ ██ ██ ██ ██ ██ ██ ██", + "██ ██ ██ ██ ██ ██ ██ ██", +) + +_ACCENT_RGB = (232, 99, 42) +_DIM_RGB = (110, 110, 117) + +_CATEGORY_COLORS: dict[InspectionCategory, tuple[int, int, int]] = { + InspectionCategory.FABRICATION: (255, 139, 92), + InspectionCategory.MANIPULATION: (255, 99, 99), + InspectionCategory.DECEPTION: (167, 139, 250), + InspectionCategory.UNPREDICTABILITY: (251, 191, 36), + InspectionCategory.OPACITY: (96, 165, 250), +} + +_CATEGORY_ORDER: tuple[InspectionCategory, ...] = ( + InspectionCategory.FABRICATION, + InspectionCategory.MANIPULATION, + InspectionCategory.DECEPTION, + InspectionCategory.UNPREDICTABILITY, + InspectionCategory.OPACITY, +) + +_BAR_WIDTH = 26 +_BLOCK_FULL = "█" +_BLOCK_EMPTY = "·" + + +def supports_color(stream=None) -> bool: + s = stream or sys.stdout + if os.environ.get("NO_COLOR"): + return False + if not hasattr(s, "isatty"): + return False + return bool(s.isatty()) + + +def _truecolor(text: str, rgb: tuple[int, int, int], bold: bool = False) -> str: + if not supports_color(): + return text + r, g, b = rgb + prefix = f"\033[38;2;{r};{g};{b}m" + if bold: + prefix = "\033[1m" + prefix + return f"{prefix}{text}\033[0m" + + +def print_startup_banner(version: str, *, quiet: bool = False) -> None: + if quiet or not supports_color(): + return + click.echo() + for line in _LOGO_LINES: + click.echo(" " + _truecolor(line, _ACCENT_RGB, bold=True)) + click.echo() + click.echo(_truecolor(f" ™ · v{version} · powered by iMe", _DIM_RGB)) + click.echo() + + +@dataclass +class _CategoryRow: + category: InspectionCategory + total: int = 0 + done: int = 0 + failed: int = 0 + + +@dataclass +class CategoryProgress: + rows: dict[InspectionCategory, _CategoryRow] = field(default_factory=dict) + _printed_lines: int = 0 + _started: bool = False + _interactive: bool = False + _spinner: _Spinner | None = None + + @classmethod + def from_totals(cls, totals: dict[InspectionCategory, int]) -> "CategoryProgress": + rows = {cat: _CategoryRow(category=cat, total=totals.get(cat, 0)) for cat in _CATEGORY_ORDER} + return cls(rows=rows) + + def start(self) -> None: + if self._started: + return + self._started = True + self._interactive = supports_color() + if not self._interactive: + return + total = sum(r.total for r in self.rows.values()) + if total == 0: + return + self._spinner = _Spinner(f"Running {total} tests") + self._spinner.start() + self._printed_lines = 1 + + def record(self, category: InspectionCategory, passing: bool) -> None: + row = self.rows.get(category) + if row is None: + return + if row.total == 0: + row.total = max(1, row.total) + row.done += 1 + if not passing: + row.failed += 1 + self._redraw() + + def finalize(self) -> None: + if self._spinner is not None: + self._spinner.stop() + self._spinner = None + self._redraw(final=True) + if self._interactive and self._printed_lines: + click.echo() + + def _redraw(self, *, final: bool = False) -> None: + if not self._started: + return + rendered = self._render() + if not rendered: + return + if self._spinner is not None: + self._spinner.stop() + self._spinner = None + if self._interactive and self._printed_lines: + sys.stdout.write(f"\033[{self._printed_lines}F") + for line in rendered: + sys.stdout.write("\033[2K") + sys.stdout.write(line + "\n") + sys.stdout.flush() + self._printed_lines = len(rendered) + elif self._interactive and not self._printed_lines: + for line in rendered: + sys.stdout.write(line + "\n") + sys.stdout.flush() + self._printed_lines = len(rendered) + elif not self._interactive and final: + for line in rendered: + click.echo(line) + + def _render(self) -> list[str]: + out: list[str] = [] + for cat in _CATEGORY_ORDER: + row = self.rows[cat] + if row.total == 0: + continue + label = cat.value.upper().ljust(16) + ratio = row.done / row.total if row.total else 0.0 + filled = int(round(ratio * _BAR_WIDTH)) + bar = _BLOCK_FULL * filled + _BLOCK_EMPTY * (_BAR_WIDTH - filled) + colored_bar = _truecolor(bar, _CATEGORY_COLORS[cat], bold=True) + count = f"{row.done}/{row.total}".rjust(7) + tail = "" + if row.done >= row.total: + if row.failed == 0: + tail = " " + _truecolor("✓", (74, 222, 128)) + else: + tail = " " + _truecolor(f"✗ {row.failed} failed", (239, 68, 68)) + out.append(f" {label} {colored_bar} {count}{tail}") + return out + + +def category_totals_from_specs(specs: Iterable[object]) -> dict[InspectionCategory, int]: + counts: dict[InspectionCategory, int] = {cat: 0 for cat in _CATEGORY_ORDER} + for spec in specs: + cat = getattr(spec, "category", None) + if isinstance(cat, InspectionCategory): + counts[cat] = counts.get(cat, 0) + 1 + return counts diff --git a/ifixai/cli/_imecore_prompt.py b/ifixai/cli/_imecore_prompt.py new file mode 100644 index 00000000..3a4b2510 --- /dev/null +++ b/ifixai/cli/_imecore_prompt.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import os +import sys + +import click + +from ifixai.cli._branding import _ACCENT_RGB, _DIM_RGB, _truecolor + +ENV_NO_PROMPT = "IFIXAI_NO_PROMPT" + + +def print_imecore_conclusion(*, quiet: bool) -> None: + if quiet: + return + if os.environ.get(ENV_NO_PROMPT): + return + if not sys.stdout.isatty(): + _print_plain_conclusion() + return + + click.echo() + click.echo(click.style("Conclusion", bold=True)) + click.echo( + " The report above isn't a bug list. It's the absence of an alignment layer." + ) + click.echo() + click.echo(" " + _truecolor("iFixAi measures it. iMe ends it.", _ACCENT_RGB, bold=True)) + click.echo() + click.echo( + " iMe is the deterministic alignment runtime: non-LLM, six constitutional" + ) + click.echo(" rules, five-stage pipeline.") + click.echo() + click.echo( + " " + _truecolor("Probabilistic guardrails fail. Deterministic rules don't.", _ACCENT_RGB, bold=True) + ) + click.echo() + click.echo(" " + _truecolor("Limited release. Selected deployments.", _DIM_RGB)) + click.echo( + " Request access → " + _truecolor("https://ifixai.ai/ime", _ACCENT_RGB, bold=True) + ) + click.echo() + + +def _print_plain_conclusion() -> None: + click.echo() + click.echo("Conclusion") + click.echo(" The report above isn't a bug list. It's the absence of an alignment layer.") + click.echo() + click.echo(" iFixAi measures it. iMe ends it.") + click.echo() + click.echo(" iMe is the deterministic alignment runtime: non-LLM, six constitutional") + click.echo(" rules, five-stage pipeline.") + click.echo() + click.echo(" Probabilistic guardrails fail. Deterministic rules don't.") + click.echo() + click.echo(" Limited release. Selected deployments.") + click.echo(" Request access → https://ifixai.ai/ime") + click.echo() diff --git a/ifixai/cli/orchestrator.py b/ifixai/cli/orchestrator.py index f67f6d78..ea7ba9ef 100644 --- a/ifixai/cli/orchestrator.py +++ b/ifixai/cli/orchestrator.py @@ -185,13 +185,13 @@ def _print_insufficient_evidence_summary(result: TestRunResult) -> None: insufficient = [br for br in result.test_results if br.insufficient_evidence] total = len(result.test_results) if not insufficient: - click.echo(click.style(f"Insufficient evidence: 0/{total} inspections.", fg="green")) + click.echo(click.style(f"0 out of {total} tests have failed.", fg="green")) return inspection_ids = ", ".join(sorted(br.test_id for br in insufficient)) click.echo( click.style( - f"Insufficient evidence: {len(insufficient)}/{total} inspections excluded " - f"from aggregate ({inspection_ids}). Wrap your provider in a governance layer " + f"{len(insufficient)} out of {total} tests have failed " + f"({inspection_ids}). Wrap your provider in a governance layer " f"or run with ≥2 provider credentials. See docs/methodology.md.", fg="yellow", ) @@ -233,14 +233,17 @@ async def execute_tests( sut_temperature: float = 0.0, sut_seed: int | None = None, self_judged: bool = False, + progress_callback=None, ) -> TestRunResult | None: try: - load_fixture(fixture) + loaded_fixture = load_fixture(fixture) except FileNotFoundError as exc: click.echo(click.style(f"Fixture error: {exc}", fg="red")) return None + effective_callback = progress_callback or _progress_callback + try: if test_id: single_result = await run_single( @@ -270,7 +273,7 @@ async def execute_tests( system_name=system_name, system_version=system_version, provider=provider, - fixture_name=fixture, + fixture_name=loaded_fixture.metadata.name, overall_score=single_result.score, strategic_score=single_result.score, test_results=[single_result], @@ -288,7 +291,7 @@ async def execute_tests( model=model, system_prompt=system_prompt, timeout=timeout, - progress_callback=_progress_callback, + progress_callback=effective_callback, pipeline_config=pipeline_config, judge_config=judge_config, governor=governor, @@ -308,7 +311,7 @@ async def execute_tests( model=model, system_prompt=system_prompt, timeout=timeout, - progress_callback=_progress_callback, + progress_callback=effective_callback, pipeline_config=pipeline_config, judge_config=judge_config, governor=governor, diff --git a/ifixai/cli/reports.py b/ifixai/cli/reports.py index 21e3bcbf..65306066 100644 --- a/ifixai/cli/reports.py +++ b/ifixai/cli/reports.py @@ -1,3 +1,4 @@ +import re from pathlib import Path import click @@ -9,14 +10,22 @@ from ifixai.types import TestRunResult +def _slugify(value: str) -> str: + if not value: + return "unknown" + stem = Path(value).stem if ("/" in value or "\\" in value) else value + cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", stem).strip("-").lower() + return cleaned or "unknown" + + def save_reports( result: TestRunResult, output_dir: str, report_format: str ) -> None: out_path = Path(output_dir) out_path.mkdir(parents=True, exist_ok=True) - system_slug = result.system_name.lower().replace(" ", "-") - fixture_slug = result.fixture_name.lower().replace(" ", "-") + system_slug = _slugify(result.system_name) + fixture_slug = _slugify(result.fixture_name) base_name = f"ifixai-{system_slug}-{fixture_slug}" if report_format in ("json", "both"): diff --git a/ifixai/cli/run.py b/ifixai/cli/run.py index fab230a6..a734f419 100644 --- a/ifixai/cli/run.py +++ b/ifixai/cli/run.py @@ -12,11 +12,17 @@ import click +from ifixai import __version__ as IFIXAI_VERSION +from ifixai.cli._branding import ( + CategoryProgress, + print_startup_banner, + supports_color, +) +from ifixai.cli._imecore_prompt import print_imecore_conclusion from ifixai.cli.orchestrator import ( _build_judge_config, _eval_mode_declaration, _lookup_env_api_key, - _print_inconclusive_summary, _print_insufficient_evidence_summary, _resolve_judge_label, _resolve_standard_eval_mode, @@ -388,6 +394,14 @@ def _print_concurrency_banner(resolved: int) -> None: "accept a seed will record seed_supported_by_provider=false in the " "manifest; the seed is still recorded.", ) +@click.option( + "--quiet", + "-q", + is_flag=True, + default=False, + help="Suppress the startup banner, in-place progress bars, and the post-run " + "iMe Core conclusion. Stdout still contains scores so CI gates keep working.", +) def run( provider: str | None, api_key: str | None, @@ -420,8 +434,10 @@ def run( sut_temperature: float, sut_seed: int | None, grounding: str, + quiet: bool, ) -> None: """Run ifixai against a target AI assistant.""" + print_startup_banner(IFIXAI_VERSION, quiet=quiet) resolved_concurrency = _resolve_concurrency(concurrency, no_parallel) _print_concurrency_banner(resolved_concurrency) concurrency_governor = ConcurrencyGovernor(resolved_concurrency) @@ -837,6 +853,26 @@ def run( if sys.platform == "win32": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + + cat_progress: CategoryProgress | None = None + progress_cb = None + if not quiet and not test and supports_color(): + from ifixai.tests.registry import ALL_SPECS, STRATEGIC_TESTS + + targeted = ( + [s for s in ALL_SPECS if s.test_id in STRATEGIC_TESTS] + if strategic + else list(ALL_SPECS) + ) + totals: dict = {} + for s in targeted: + cat = getattr(s, "category", None) + if cat is not None: + totals[cat] = totals.get(cat, 0) + 1 + cat_progress = CategoryProgress.from_totals(totals) + cat_progress.start() + progress_cb = _make_category_progress_callback(cat_progress) + result = asyncio.run( execute_tests( provider=provider, @@ -856,9 +892,13 @@ def run( sut_temperature=sut_temperature, sut_seed=sut_seed, self_judged=(eval_mode == "self"), + progress_callback=progress_cb, ) ) + if cat_progress is not None: + cat_progress.finalize() + if result is None: sys.exit(1) @@ -871,7 +911,17 @@ def run( click.echo(f" Strategic Score: {redacted}") click.echo(f" Passed: {redacted}") else: - click.echo(f" Overall Score: {result.overall_score:.1%}") + score_line = f" Overall Score: {result.overall_score:.1%}" + if ( + result.overall_score is not None + and min_score is not None + and result.overall_score < min_score + ): + score_line += click.style( + f" (Score {result.overall_score:.1%} is below minimum {min_score:.1%})", + fg="red", + ) + click.echo(score_line) click.echo(f" Grade: {result.grade.value}") click.echo(f" Strategic Score: {result.strategic_score:.1%}") verdict = ( @@ -882,10 +932,9 @@ def run( click.echo(f" Passed: {verdict}") click.echo() - _print_inconclusive_summary(result) _print_insufficient_evidence_summary(result) - click.echo() + click.echo(click.style("Access your Full Report here:", bold=True)) save_reports(result, output, report_format) run_mode = RunMode.FULL if profile.lower() == "full" else RunMode.STANDARD @@ -929,12 +978,9 @@ def run( b14_seed=b14_seed if b14_seed is not None else 20260422, b30_seed=b30_seed if b30_seed is not None else 20260422, ) - manifest_path = write_manifest(manifest, Path(reliability_out)) - click.echo() - click.echo(click.style("Run manifest", bold=True)) - click.echo(f" Mode: {manifest.mode.value}") - click.echo(f" Run ID: {manifest.run_id}") - click.echo(f" Manifest: {manifest_path}") + write_manifest(manifest, Path(reliability_out)) + + print_imecore_conclusion(quiet=quiet) if result.overall_score is None: click.echo( @@ -945,14 +991,18 @@ def run( ) sys.exit(2) if result.overall_score < min_score: - click.echo( - click.style( - f"Score {result.overall_score:.1%} is below minimum {min_score:.1%}", - fg="red", - ) - ) sys.exit(2) + +def _make_category_progress_callback(progress: "CategoryProgress"): + def _cb(bid: str, index: int, total: int, bench_result: object) -> None: + category = getattr(bench_result, "category", None) + passing = bool(getattr(bench_result, "passing", False)) + if category is not None: + progress.record(category, passing) + return _cb + + def gather_interactive_config() -> tuple[str, str, str | None, str | None]: """Run the interactive guided mode to collect provider configuration.