feat(ENG-15367): capture usage facts and send one event at the end of every run

cloudsmith_cli/cli/command.py

@@ -8,6 +8,7 @@
 from click_didyoumean import DYMGroup
 
 from ..core.certificate_errors import describe_certificate_error
+from .usage_capture import TrackedCommand, TrackingMixin, record_alias
 
 
 def _is_json_output_requested(exception):
@@ -74,7 +75,7 @@ def _echo_certificate_hint(exception):
         click.echo(f"Hint: {click.style(hint, fg='yellow')}", err=True)
 
 
-class AliasGroup(DYMGroup):
+class AliasGroup(TrackingMixin, DYMGroup):
     """A command group with DYM, alias and lazy-subcommand support.
 
     ``lazy_commands`` maps a command name to the module that registers it.
@@ -83,6 +84,8 @@ class AliasGroup(DYMGroup):
     ``lazy_aliases`` supplies the alias map for those commands up front.
     """
 
+    command_class = TrackedCommand
+
     def __init__(self, *args, **kwargs):
         lazy_commands = kwargs.pop("lazy_commands", None)
         lazy_aliases = kwargs.pop("lazy_aliases", None)
@@ -96,6 +99,8 @@ def __init__(self, *args, **kwargs):
                 self.inverse[alias] = name
 
     def resolve_command(self, ctx, args):
+        if args and args[0] in self.inverse:
+            record_alias(ctx, args[0])
         try:
             return super().resolve_command(ctx, args)
         except click.exceptions.UsageError:
@@ -106,6 +111,7 @@ def resolve_command(self, ctx, args):
                 cmds = self.list_commands(ctx)
                 matched = [cmd for cmd in cmds if cmd.startswith(cmd_name)]
                 if len(matched) == 1 and len(cmd_name) > 1:
+                    record_alias(ctx, "<prefix>")
                     args[0] = matched[0]
                     return super().resolve_command(ctx, args)
 
@@ -177,21 +183,60 @@ def format_commands(self, ctx, formatter):
         ctx.showing_help = True
         return super().format_commands(ctx, formatter)
 
+    def _main_shell_completion(self, ctx_args, prog_name, complete_var=None):
+        """Mark shell completion runs so usage telemetry ignores them.
+
+        Click handles ``*_source`` before any context exists, so the
+        ``resilient_parsing`` check in ``_finish_usage`` never sees it.
+        """
+        import os
+
+        from . import usage_capture
+
+        name = complete_var
+        if name is None:
+            # Same derivation as Click's own _main_shell_completion.
+            name = prog_name.replace("-", "_").replace(".", "_")
+            name = f"_{name}_COMPLETE".upper()
+        if os.environ.get(name):
+            usage_capture.mark_completion_run()
+        return super()._main_shell_completion(ctx_args, prog_name, complete_var)
+
     def main(self, *args, **kwargs):
         """Override main to intercept exceptions and format as JSON if requested."""
         import sys
+        import time
+
+        from . import usage_capture
 
         original_standalone_mode = kwargs.get("standalone_mode", True)
         kwargs["standalone_mode"] = False
+        usage = {"outcome": "unexpected_error", "exit_code": 1}
+        depth = usage_capture.enter_run()
+        started = time.monotonic()
 
         try:
-            return super().main(*args, **kwargs)
+            rv = super().main(*args, **kwargs)
+            usage.update(
+                outcome="success" if not rv else "error_exit",
+                exit_code=(rv if isinstance(rv, int) else 1) if rv else 0,
+            )
+            return rv
         except click.exceptions.Abort:
+            usage.update(outcome="aborted", exit_code=1)
             if not original_standalone_mode:
                 raise
             click.echo("Aborted!", err=True)
             sys.exit(1)
         except click.exceptions.ClickException as e:
+            usage.update(
+                outcome=(
+                    "usage_error"
+                    if isinstance(e, click.exceptions.UsageError)
+                    else "click_error"
+                ),
+                exit_code=e.exit_code,
+            )
             if _is_json_output_requested(e):
                 import json
 
@@ -205,15 +250,24 @@ def main(self, *args, **kwargs):
             sys.exit(e.exit_code)
         except BrokenPipeError:
             # Click does not catch a failed write in its Ctrl+C handler.
+            usage.update(outcome="aborted", exit_code=1)
             if not original_standalone_mode:
                 raise
             sys.stdout = _PacifyFlushWrapper(sys.stdout)
             sys.stderr = _PacifyFlushWrapper(sys.stderr)
             sys.exit(1)
+        except SystemExit as e:
+            code = e.code
+            usage.update(
+                exit_code=code if isinstance(code, int) else (0 if code is None else 1),
+                outcome="success" if code in (0, None) else "error_exit",
+            )
+            raise
         except Exception as e:  # pylint: disable=broad-exception-caught
             # Anything reaching here is unexpected: click's own errors and
             # handled API errors (which exit via ctx.exit) are dealt with
             # above or never get this far.
+            usage.update(outcome="unexpected_error", exit_code=1)
             if not original_standalone_mode:
                 raise
             main_args = kwargs.get("args", args[0] if args else None)
@@ -228,3 +282,35 @@ def main(self, *args, **kwargs):
             click.echo(f"Error: {telemetry.format_exception_summary(e)}", err=True)
             _echo_certificate_hint(e)
             sys.exit(1)
+        finally:
+            _finish_usage(usage, started, nested=depth > 1)
+            usage_capture.exit_run()
+
+
+def _finish_usage(usage, started, *, nested):
+    """Hand the run to usage telemetry; never raises, never prints."""
+    try:
+        import time
+
+        from ..core.usage import run
+        from . import usage_capture
+
+        meta = usage_capture.current_root_meta() or {}
+        contexts = meta.get(usage_capture.META_KEY, {}).get("contexts", [])
+        # Shell completion parses with resilient_parsing (or, for the
+        # *_source script, not at all); never report it.
+        completing = usage_capture.is_completion_run() or any(
+            getattr(c, "resilient_parsing", False) for c in contexts
+        )
+        facts = usage_capture.collect_facts(
+            meta,
+            outcome=usage["outcome"],
+            exit_code=usage["exit_code"],
+            duration_ms=int((time.monotonic() - started) * 1000),
+            nested=nested or completing,
+        )
+        if facts.api_status is not None and facts.outcome == "error_exit":
+            facts.outcome = "api_error"
+        run.finish(facts)
+    except BaseException:  # pylint: disable=broad-exception-caught
+        return

cloudsmith_cli/cli/commands/credential_helper/__init__.py

@@ -8,6 +8,7 @@
 
 import click
 
+from ...usage_capture import TrackedGroup
 from ..main import main
 from .cargo import cargo as cargo_cmd
 from .docker import docker as docker_cmd
@@ -17,7 +18,7 @@
 from .terraform import terraform as terraform_cmd
 
 
-@click.group()
+@click.group(cls=TrackedGroup)
 def credential_helper():
     """
     Credential helpers for package managers.

cloudsmith_cli/cli/commands/credential_helper/cargo.py

@@ -13,9 +13,10 @@
 
 from ....credential_helpers.cargo import execute
 from ...decorators import common_api_auth_options, resolve_credentials
+from ...usage_capture import TrackedCommand
 
 
-@click.command(context_settings={"ignore_unknown_options": True})
+@click.command(cls=TrackedCommand, context_settings={"ignore_unknown_options": True})
 @click.option(
     "--cargo-plugin",
     is_flag=True,

cloudsmith_cli/cli/commands/credential_helper/docker.py

@@ -13,9 +13,10 @@
 
 from ....credential_helpers.docker import execute
 from ...decorators import common_api_auth_options, resolve_credentials
+from ...usage_capture import TrackedCommand
 
 
-@click.command()
+@click.command(cls=TrackedCommand)
 @click.argument("operation", required=False, default="get")
 @common_api_auth_options
 @resolve_credentials

cloudsmith_cli/cli/commands/credential_helper/generic.py

@@ -15,9 +15,10 @@
     common_cli_config_options,
     resolve_credentials,
 )
+from ...usage_capture import TrackedCommand
 
 
-@click.command()
+@click.command(cls=TrackedCommand)
 @common_cli_config_options
 @common_api_auth_options
 @resolve_credentials

cloudsmith_cli/cli/commands/credential_helper/manage.py

@@ -37,6 +37,7 @@
     common_cli_output_options,
     resolve_credentials,
 )
+from ...usage_capture import TrackedCommand
 
 # ---------------------------------------------------------------------------
 # Helper registry — extend here when new helpers are added
@@ -248,7 +249,7 @@ def _terraform_workspace_next_steps(helper_args: tuple[str, ...]) -> list[str]:
 # ---------------------------------------------------------------------------
 
 
-@click.command("install")
+@click.command("install", cls=TrackedCommand)
 @click.argument("helper")
 @click.option(
     "--bin-dir",
@@ -420,7 +421,7 @@ def install_cmd(
 # ---------------------------------------------------------------------------
 
 
-@click.command("uninstall")
+@click.command("uninstall", cls=TrackedCommand)
 @click.argument("helper")
 @click.option(
     "--bin-dir",
@@ -487,7 +488,7 @@ def uninstall_cmd(ctx, opts, helper: str, bin_dir: str | None, dry_run: bool) ->
 # ---------------------------------------------------------------------------
 
 
-@click.command("list")
+@click.command("list", cls=TrackedCommand)
 @common_cli_config_options
 @common_cli_output_options
 @click.pass_context

cloudsmith_cli/cli/commands/credential_helper/pnpm.py

@@ -13,9 +13,10 @@
 from ....credential_helpers.pnpm import execute
 from ...config import Options
 from ...decorators import common_api_auth_options, resolve_credentials
+from ...usage_capture import TrackedCommand
 
 
-@click.command()
+@click.command(cls=TrackedCommand)
 @click.argument("repo", required=False, default="npm.cloudsmith.io")
 @common_api_auth_options
 @resolve_credentials
@@ -64,7 +65,7 @@
 
     exit_code, stdout, stderr = execute(
         repo,
         credential=opts.credential,
         api_host=opts.api_host,
         org=opts.org,
     )

cloudsmith_cli/cli/commands/credential_helper/terraform.py

@@ -19,9 +19,10 @@
     common_cli_config_options,
     resolve_credentials,
 )
+from ...usage_capture import TrackedCommand
 
 
-@click.command(context_settings={"ignore_unknown_options": True})
+@click.command(cls=TrackedCommand, context_settings={"ignore_unknown_options": True})
 @click.option(
     "-r",
     "--repo",

cloudsmith_cli/cli/commands/login.py

@@ -8,6 +8,7 @@
 from ...core.config import create_config_files, new_config_messaging
 from .. import decorators, deprecation, utils
 from ..exceptions import handle_api_exceptions
+from ..usage_capture import TrackedCommand
 from ..utils import maybe_spinner
 from .main import main
 
@@ -61,7 +62,7 @@
     return None
 
 
-class DeprecatedLoginCommand(click.Command):
+class DeprecatedLoginCommand(TrackedCommand):
     """Command that prints the deprecation warning before any prompt."""
 
     def parse_args(self, ctx, args):
@@ -150,7 +151,7 @@
                     totp_token=totp_token,
                     two_factor_token=e.two_factor_token,
                 )
         except cloudsmith_api.rest.ApiException:
             click.echo("\r\033[K", nl=False, err=use_stderr)
             click.secho(
                 "Authentication failed: The entered TOTP token is not valid.",
@@ -159,7 +160,7 @@
             )
             ctx.exit(1)
 
     except cloudsmith_api.rest.ApiException as e:
         click.echo("\r\033[K", nl=False, err=use_stderr)
         click.secho(f"Authentication failed: {e!s}", fg="red", err=use_stderr)
         ctx.exit(1)

cloudsmith_cli/cli/config_sources.py

@@ -0,0 +1,91 @@
+# Copyright 2026 Cloudsmith Ltd
+"""Which settings were set in config.ini / credentials.ini, by section.
+
+``ConfigReader.read_config()`` fills schema defaults, so it cannot tell an
+explicit key from a default; this reads the raw files and reports key
+presence only. A value is only checked for being empty, and never kept.
+"""
+
+import configparser
+import os
+
+from ..core.usage.event import SETTING_ALIASES
+from . import config
+
+
+def _files(reader, explicit):
+    paths = [explicit] if explicit else []
+    paths.extend(
+        name if os.path.isabs(name) else os.path.join(searchpath, name)
+        for searchpath in reader.config_searchpath
+        for name in reader.config_files
+    )
+    return [p for p in paths if p and os.path.isfile(p)]
+
+
+def _declared(schema):
+    """Key names the CLI reads for a section schema (never user-authored keys)."""
+    return {
+        param.name
+        for param in vars(schema).values()
+        if isinstance(param, config.ConfigParam)
+    }
+
+
+def _keys(paths, section, declared):
+    keys = set()
+    for path in paths:
+        parser = configparser.RawConfigParser()
+        try:
+            parser.read(path, encoding="utf-8")
+        except (configparser.Error, UnicodeDecodeError, OSError):
+            continue
+        if parser.has_section(section):
+            # An empty key (`api_host=` in the default config.ini) sets nothing.
+            keys.update(
+                key
+                for key in parser.options(section)
+                if parser.get(section, key).strip()
+            )
+    return keys & declared
+
+
+def _present(config_file, creds_file, profile):
+    """Yield ``(key, source)`` for each schema key set in the ini files."""
+    for reader, schema, explicit, base in (
+        (config.ConfigReader, config.ConfigSchema.Default, config_file, "config"),
+        (
+            config.CredentialsReader,
+            config.CredentialsSchema.Default,
+            creds_file,
+            "credentials",
+        ),
+    ):
+        paths = _files(reader, explicit)
+        declared = _declared(schema)
+        for key in _keys(paths, "default", declared):
+            yield key, f"{base}_file"
+        if profile and profile != "default":
+            for key in _keys(paths, f"profile:{profile}", declared):
+                yield key, f"{base}_profile"
+
+
+def detect(
+    *, config_file: str | None, creds_file: str | None, profile: str | None
+) -> dict[str, str]:
+    """Map each setting present in the ini files to where it was set."""
+    found: dict[str, str] = {}
+    for key, source in _present(config_file, creds_file, profile):
+        found[SETTING_ALIASES.get(key, key)] = source
+    return found
+
+
+def keys(
+    *, config_file: str | None, creds_file: str | None, profile: str | None
+) -> list[str]:
+    """The key names set in the ini files, as written: `org`, not `workspace`.
+
+    Shows which spelling of a renamed key is still in use. Only names the
+    schema declares, never values.
+    """
+    return sorted({key for key, _ in _present(config_file, creds_file, profile)})