Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
224 changes: 224 additions & 0 deletions ifixai/cli/_branding.py
Original file line number Diff line number Diff line change
@@ -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
60 changes: 60 additions & 0 deletions ifixai/cli/_imecore_prompt.py
Original file line number Diff line number Diff line change
@@ -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()
17 changes: 10 additions & 7 deletions ifixai/cli/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
)
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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],
Expand All @@ -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,
Expand All @@ -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,
Expand Down
13 changes: 11 additions & 2 deletions ifixai/cli/reports.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import re
from pathlib import Path

import click
Expand All @@ -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"):
Expand Down
Loading
Loading