feat(cli): phase-aware scan progress and clearer verbosity flags

argus/cli.py

@@ -81,16 +81,118 @@ def _clear_line(self) -> None:
         self._stream.write("\r" + (" " * 80) + "\r")
         self._stream.flush()
 
+    @property
+    def enabled(self) -> bool:
+        """True when the spinner thread is actively drawing."""
+        return self._enabled
+
+    def update_message(self, message: str) -> None:
+        """Replace the spinner's status text in place.
+
+        Called from progress callbacks at phase transitions so the
+        user sees ``[2/4] scanner-opengrep — trivy (12s)`` updating
+        instead of a static "Running container scan". No-op when the
+        spinner isn't drawing — callers can pass through unconditionally.
+        """
+        # Pad with trailing spaces to overwrite any longer previous
+        # message without leaving stale characters on the line.
+        prev = self._message
+        self._message = message
+        if self._enabled and len(prev) > len(message):
+            self._stream.write("\r" + (" " * len(prev)) + "\r")
+            self._stream.flush()
+
+
+def _configure_logger(args: argparse.Namespace, output_dir: str | None = None):
+    """Set up the ``argus`` logger at the level the user's flags imply.
+
+    Mapping:
+      * ``--verbose`` / ``--debug``: DEBUG (full firehose)
+      * ``--quiet``: WARNING (suppress INFO lines from the engine)
+      * default:    INFO (per-phase status, no DEBUG noise)
+
+    Returns the configured logger so callers can also use it directly.
+    """
+    import logging
+    from argus.audit import get_logger
+
+    log = get_logger(
+        "argus",
+        output_dir=output_dir,
+        verbose=_verbose_enabled(args),
+    )
+    if _quiet_enabled(args):
+        for handler in log.handlers:
+            if isinstance(handler, logging.StreamHandler) and not isinstance(
+                handler, logging.FileHandler,
+            ):
+                handler.setLevel(logging.WARNING)
+    return log
+
+
+def _verbose_enabled(args: argparse.Namespace) -> bool:
+    """``--verbose`` / ``--debug`` — full firehose mode.
+
+    The two are aliases: ``--verbose`` is preserved for backward
+    compatibility, ``--debug`` is the clearer name new users learn.
+    """
+    return getattr(args, "verbose", False) or getattr(args, "debug", False)
+
+
+def _quiet_enabled(args: argparse.Namespace) -> bool:
+    """``--quiet`` / ``-q`` — suppress per-phase INFO lines.
+
+    Spinner stays drawing (just doesn't update its message) unless
+    ``--no-spinner`` is also passed. Composes with ``--no-spinner``:
+    ``--quiet --no-spinner`` means fully silent (CI exit-code-only).
+    """
+    return getattr(args, "quiet", False)
+
 
 def _spinner_enabled(args: argparse.Namespace) -> bool:
     """Enable spinner for interactive terminals unless explicitly disabled."""
     if getattr(args, "no_spinner", False):
         return False
-    if getattr(args, "verbose", False):
+    if _verbose_enabled(args):
         return False
     return sys.stderr.isatty()
 
 
+def _make_progress_emitter(
+    args: argparse.Namespace,
+    spinner: "_TerminalSpinner | None",
+):
+    """Return a progress callback that routes phase events to the right surface.
+
+    Output target depends on the flag combination:
+      * ``--quiet`` or ``--verbose``: no-op (logger handles verbose; quiet
+        suppresses progress on purpose).
+      * Spinner is drawing: update its message in place.
+      * No spinner (``--no-spinner`` or non-TTY): print a persistent
+        INFO line to stderr per phase event so CI logs and step-away
+        terminals get scrollback.
+    """
+    if _quiet_enabled(args) or _verbose_enabled(args):
+        return lambda *a, **kw: None
+
+    if spinner is not None and spinner.enabled:
+        def update_spinner(idx: int, total: int, name: str, phase: str, elapsed_ms: int):
+            elapsed_s = elapsed_ms // 1000
+            spinner.update_message(
+                f"[{idx}/{total}] {name} — {phase} ({elapsed_s}s)"
+            )
+        return update_spinner
+
+    def print_line(idx: int, total: int, name: str, phase: str, elapsed_ms: int):
+        elapsed_s = elapsed_ms // 1000
+        print(
+            f"[{idx}/{total}] {name} — {phase} ({elapsed_s}s)",
+            file=sys.stderr,
+            flush=True,
+        )
+    return print_line
+
+
 def _make_run_dir(base_dir: str) -> str:
     """Create a timestamped subdirectory for this scan run.
 
@@ -486,9 +588,25 @@ def _build_scan_parser(subparsers: argparse._SubParsersAction) -> None:
         help="List available scanners and exit",
     )
     scan_parser.add_argument(
-        "--verbose", "-v",
+        "--verbose",
         action="store_true",
-        help="Enable verbose output",
+        help="Alias for --debug. Full firehose: subprocess output, "
+             "vulnerability-DB updates, every engine log line.",
+    )
+    scan_parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="Full firehose: subprocess output, vulnerability-DB updates, "
+             "every engine log line. Use when troubleshooting; the default "
+             "phase-aware progress is enough for normal scans.",
+    )
+    scan_parser.add_argument(
+        "--quiet", "-q",
+        action="store_true",
+        help="Suppress per-phase progress lines. The spinner still draws "
+             "(use --no-spinner to suppress that too). Final summary "
+             "still prints. Compose with --no-spinner for fully silent "
+             "CI exit-code-only mode.",
     )
     scan_parser.add_argument(
         "--no-spinner",
@@ -1754,15 +1872,14 @@ def _cmd_container_scan(
     that path is kept for backward compatibility with any caller that
     still bypasses ``cmd_scan``.
     """
-    from argus.audit import get_logger
     from argus.container import ContainerEngine
     from argus.reporters.container_markdown import ContainerMarkdownReporter
 
-    # Configure the ``argus`` logger so engine-level INFO/DEBUG output
-    # actually reaches the terminal under ``--verbose``. Without this
-    # the container engine's logger.info() calls go nowhere — same
-    # setup the source-scan handler does at the top of its body.
-    get_logger("argus", verbose=getattr(args, "verbose", False))
+    # Configure the ``argus`` logger at the user's chosen level —
+    # default INFO, --quiet WARNING, --verbose/--debug DEBUG. Without
+    # this the container engine's logger.info() calls would go
+    # nowhere even under --verbose.
+    _configure_logger(args)
 
     if container_config is None:
         try:
@@ -1812,11 +1929,18 @@ def _cmd_container_scan(
 
     # Run
     try:
-        engine = ContainerEngine(config)
+        # Build the spinner first so the engine's progress callback
+        # can update its message in place. ``_make_progress_emitter``
+        # picks the right routing (spinner vs persistent INFO line vs
+        # no-op) based on the user's --quiet/--debug/--no-spinner mix.
         with _TerminalSpinner(
             message="Running container scan",
             enabled=_spinner_enabled(args),
-        ):
+        ) as spinner:
+            engine = ContainerEngine(
+                config,
+                progress_callback=_make_progress_emitter(args, spinner),
+            )
             summary = engine.run()
     except Exception as exc:
         print(f"Error: container scan failed: {exc}", file=sys.stderr)

argus/container/engine.py

@@ -6,7 +6,9 @@
 """
 
 import logging
+import time
 from pathlib import Path
+from typing import Callable
 
 from .builder import build_image
 from .discovery import (
@@ -45,10 +47,22 @@ class ContainerEngine:
           cleanup: true          # Remove images after scanning (default)
     """
 
-    def __init__(self, config: dict):
+    def __init__(
+        self,
+        config: dict,
+        progress_callback: "Callable[[int, int, str, str, int], None] | None" = None,
+    ):
         self.config = config
         self._cleanup = config.get("cleanup", True)
         self._built_images: list[str] = []
+        # Progress callback signature: (idx, total, name, phase, elapsed_ms).
+        # Default is a no-op so engine code can call ``self._progress(...)``
+        # unconditionally without checking. The CLI installs a callback
+        # that updates the spinner message or prints a persistent INFO
+        # line based on flag combination.
+        self._progress: Callable[[int, int, str, str, int], None] = (
+            progress_callback or (lambda *a, **kw: None)
+        )
 
     def run(self) -> ContainerScanSummary:
         """Execute the full container scanning lifecycle.
@@ -77,14 +91,25 @@ def run(self) -> ContainerScanSummary:
         )
         results: list[ContainerScanResult] = []
 
+        total = len(targets)
         for i, target in enumerate(targets, 1):
             logger.info(
-                "[%d/%d] Processing %s", i, len(targets), target.name,
+                "[%d/%d] Processing %s", i, total, target.name,
             )
 
-            result = self._process_target(target)
+            # Emit phase progress at each transition. The callback
+            # decides whether to update the spinner, print a line, or
+            # ignore the event based on the user's CLI flags.
+            target_start = time.monotonic()
+            initial_phase = "build" if target.dockerfile else "pull"
+            self._progress(i, total, target.name, initial_phase, 0)
+
+            result = self._process_target(target, idx=i, total=total, target_start=target_start)
             results.append(result)
 
+            elapsed_ms = int((time.monotonic() - target_start) * 1000)
+            self._progress(i, total, target.name, "done", elapsed_ms)
+
             # Post-scan cleanup — free disk for the next container.
             # Per-target ``cleanup:`` overrides the engine-level default
             # so a long-lived base image can stay cached across runs
@@ -111,13 +136,29 @@ def run(self) -> ContainerScanSummary:
         )
         return summary
 
-    def _process_target(self, target: ContainerTarget) -> ContainerScanResult:
+    def _process_target(
+        self,
+        target: ContainerTarget,
+        idx: int = 1,
+        total: int = 1,
+        target_start: float | None = None,
+    ) -> ContainerScanResult:
         """Build (if needed) and scan a single container target.
 
         Catches disk space errors and other resource failures
         individually — a failure on one container doesn't stop
         the rest from being scanned.
+
+        ``idx`` / ``total`` / ``target_start`` are progress-tracking
+        params used to emit phase events between build and scan.
+        Default values keep backward compat for direct test callers.
         """
+        if target_start is None:
+            target_start = time.monotonic()
+
+        def _elapsed() -> int:
+            return int((time.monotonic() - target_start) * 1000)
+
         if target.dockerfile:
             success = build_image(target)
             if not success:
@@ -137,6 +178,10 @@ def _process_target(self, target: ContainerTarget) -> ContainerScanResult:
                 )
             self._built_images.append(target.image_ref)
 
+        # Build is done (or skipped for remote-pull targets); transition
+        # to the scan phase so the spinner updates.
+        self._progress(idx, total, target.name, "scan", _elapsed())
+
         try:
             # If the dispatcher set ``_raw_output_root`` in the
             # config dict, persist this target's raw scanner outputs