feat(cli): standardize CLI and CI outcomes around governance decisions (#30)

semapact/core/loader.py

@@ -16,7 +16,8 @@
 from open_data_contract_standard.model import OpenDataContractStandard
 
 from semapact.core.config import config_manager
-from semapact.exceptions import StorageError
+from semapact.exceptions import StorageError, ValidationError
+
 
 LOGGER = logging.getLogger(__name__)
 DEFAULT_SPARKUTILS_MAX_BYTES = int(
@@ -71,13 +72,21 @@ def load_contract(
     )
     payload = yaml.safe_load(contract_text)
     if not isinstance(payload, dict):
-        raise ValueError("Contract YAML must deserialize into a mapping object")
+        raise ValidationError("Contract YAML must deserialize into a mapping object")
 
     spec_version = payload.get("dataContractSpecification") or payload.get("apiVersion", "")
     if not (str(spec_version).startswith("3.") or str(spec_version).startswith("v3.")):
-        raise ValueError(f"SemaPact requires ODCS version 3.x, found: {spec_version}")
+        raise ValidationError(f"SemaPact requires ODCS version 3.x, found: {spec_version}")
 
-    return OpenDataContractStandard.model_validate(payload)
+    try:
+        return OpenDataContractStandard.model_validate(payload)
+    except Exception as exc:
+        from pydantic import ValidationError as PydanticValidationError
+
+        if isinstance(exc, PydanticValidationError):
+            raise ValidationError(f"ODCS contract validation failed: {exc}") from exc
+        raise
+
 
 
 def read_contract_text(

semapact/core/release.py

@@ -217,16 +217,21 @@ def apply_release_candidate(
     )
 
     if required_bump == "none":
-        raise ValueError("Contract changes do not require a release version bump")
+        from semapact.exceptions import ReleaseValidationError
+
+        raise ReleaseValidationError("Contract changes do not require a release version bump")
 
     target_version = parse_release_tag_version(release_tag)
     actual_bump = classify_version_bump(str(base_model.version or ""), target_version)
     if VERSION_RANK[actual_bump] < VERSION_RANK[required_bump]:
-        raise ValueError(
+        from semapact.exceptions import ReleaseValidationError
+
+        raise ReleaseValidationError(
             f"Release tag '{release_tag}' applies a {actual_bump} bump, but contract requires at least a "
             f"{required_bump} bump"
         )
 
+
     promoted = candidate_model.model_copy(deep=True)
     promoted.version = target_version
     return PromotionResult(

semapact/exceptions.py

@@ -20,12 +20,20 @@ class SemaPactError(Exception):
     pass
 
 
-class ValidationError(SemaPactError):
+class ValidationError(SemaPactError, ValueError):
     """Raised when a contract fails governance or structure validation."""
 
     pass
 
 
+
+class ReleaseValidationError(ValidationError):
+    """Raised when release candidate validation fails (e.g. insufficient version bump or no changes to release)."""
+
+    pass
+
+
+
 class MergeConflictError(SemaPactError):
     """Raised by the merge engine when business and technical metadata fatally conflict."""
 

semapact/interfaces/__init__.py

@@ -1,3 +1,20 @@
 from semapact.interfaces.cli import main
+from semapact.interfaces.outcomes import (
+    CliExitCode,
+    ProcessOutcome,
+    exit_code_from_exception,
+    exit_code_from_outcome,
+    outcome_from_exception,
+    outcome_from_gate_result,
+)
+
+__all__ = [
+    "CliExitCode",
+    "ProcessOutcome",
+    "exit_code_from_exception",
+    "exit_code_from_outcome",
+    "main",
+    "outcome_from_exception",
+    "outcome_from_gate_result",
+]
 
-__all__ = ["main"]

semapact/interfaces/cli.py

@@ -342,7 +342,8 @@ def main() -> int:
             except ImportError:
                 import sys
                 print("❌ TUI requires the 'tui' extra. Install it via: pip install \"semapact[tui]\"", file=sys.stderr)
-                return 1
+                from semapact.interfaces.outcomes import CliExitCode
+                return int(CliExitCode.RUNTIME_ERROR)
             from semapact.tui.app import SemaPactTUI
             app = SemaPactTUI()
             app.run()
@@ -435,25 +436,34 @@ def main() -> int:
                 print(json.dumps(payload, indent=2, sort_keys=True))
                 return 0
 
+        from semapact.interfaces.outcomes import CliExitCode
         parser.error(f"Unknown command: {args.command}")
-        return 2
+        return int(CliExitCode.VALIDATION_FAILED)
 
     except KeyboardInterrupt:
         return 130
     except SystemExit as exc:
-        return exc.code if isinstance(exc.code, int) else 1
+        from semapact.interfaces.outcomes import CliExitCode
+        return exc.code if isinstance(exc.code, int) else int(CliExitCode.RUNTIME_ERROR)
     except Exception as exc:
-        from semapact.exceptions import SemaPactError
         import logging
-
-        if isinstance(exc, SemaPactError):
+        from semapact.exceptions import (
+            GovernanceBlockedError,
+            GovernanceReviewRequiredError,
+            SemaPactError,
+        )
+        from semapact.interfaces.outcomes import exit_code_from_exception
+
+        if isinstance(exc, (GovernanceBlockedError, GovernanceReviewRequiredError)):
+            logging.getLogger("semapact").info("Governance decision: %s", exc)
+        elif isinstance(exc, SemaPactError):
             logging.getLogger("semapact").error("Fatal error: %s", exc)
         else:
             logging.getLogger("semapact").error(
                 "Fatal error: %s", exc, exc_info=True
             )
         print(f"❌ {exc}", file=__import__("sys").stderr)
-        return 1
+        return exit_code_from_exception(exc)
 
 
 if __name__ == "__main__":

semapact/interfaces/commands/export_cmd.py

@@ -109,28 +109,22 @@ def run_export(args: argparse.Namespace) -> str:
 
 
 def run_export_ge(args: argparse.Namespace) -> str:
-    import sys
-
     try:
         from semapact.quality.ge_exporter import GreatExpectationsExporter
     except ImportError as exc:
         if "great_expectations" in str(exc) or "great-expectations" in str(exc):
-            sys.exit(
-                "Error: The 'great_expectations' library is required to export GE suites.\n"
+            raise ImportError(
+                "The 'great_expectations' library is required to export GE suites.\n"
                 "Please install it using: pip install \"semapact[quality]\" or pip install great_expectations"
-            )
+            ) from exc
         raise
 
-    try:
-        output_path = GreatExpectationsExporter().export_to_path(
-            args.contract,
-            args.output,
-            schema_name=args.schema_name,
-            suite_name=args.suite_name,
-            engine=args.engine,
-        )
-        return str(output_path)
-    except (RuntimeError, ImportError) as exc:
-        if "requires pyspark to be installed" in str(exc) or "pyspark" in str(exc):
-            sys.exit(str(exc))
-        raise
+    output_path = GreatExpectationsExporter().export_to_path(
+        args.contract,
+        args.output,
+        schema_name=args.schema_name,
+        suite_name=args.suite_name,
+        engine=args.engine,
+    )
+    return str(output_path)
+

semapact/interfaces/commands/plan_cmd.py

@@ -1,4 +1,5 @@
 import argparse
+from urllib.parse import urlparse
 from typing import Any
 from semapact.governance import (
     GovernanceOperation,
@@ -19,83 +20,79 @@ def run_plan(args: argparse.Namespace) -> None:
 
     print(f"\n🔍 Contract Analysis: {args.base}")
 
-    try:
-        import_args: dict[str, Any] = {}
-        if args.type in {"delta", "delta-table"}:
-            oauth_token = None
-            if args.source.startswith("abfss://") or "dfs.core.windows.net" in args.source:
-                oauth_token = _resolve_adls_oauth_token_from_config()
-
-            table_uris = _parse_table_uris(args.tables)
-            if not table_uris:
-                from semapact.utils.storage_adapter import StorageAdapterFactory
-                adapter = StorageAdapterFactory.get_adapter(args.source)
-                try:
-                    table_uris = adapter.discover_delta_tables(args.source, credential=oauth_token)
-                except Exception as e:
-                    import logging
-                    logging.getLogger("semapact").warning(f"Failed to auto-discover delta tables: {e}")
-                    table_uris = []
-
-            if oauth_token:
-                import_args["oauth_bearer_token"] = oauth_token
-            if table_uris:
-                import_args["table_uris"] = table_uris
-        elif args.tables:
-            import_args["tables"] = args.tables
-
-        # Import temporary contract from source
-        imported = pipeline.import_schema(
-            source_type=args.type,
-            source=args.source,
-            uc_workspace_url=args.workspace_url,
-            uc_token=args.token,
-            import_args=import_args if import_args else None,
+    import_args: dict[str, Any] = {}
+    if args.type in {"delta", "delta-table"}:
+        oauth_token = None
+        source_is_abfss = args.source.startswith("abfss://")
+        parsed_source = urlparse(args.source)
+        source_host = parsed_source.hostname or ""
+        source_is_adls_host = (
+            source_host == "dfs.core.windows.net"
+            or source_host.endswith(".dfs.core.windows.net")
         )
+        if source_is_abfss or source_is_adls_host:
+            oauth_token = _resolve_adls_oauth_token_from_config()
 
-        # Load base contract (governed target)
-        base_contract = pipeline.loader.load(args.base)
+        table_uris = _parse_table_uris(args.tables)
+        if not table_uris:
+            from semapact.utils.storage_adapter import StorageAdapterFactory
+            adapter = StorageAdapterFactory.get_adapter(args.source)
+            try:
+                table_uris = adapter.discover_delta_tables(args.source, credential=oauth_token)
+            except Exception as e:
+                import logging
+                logging.getLogger("semapact").warning(f"Failed to auto-discover delta tables: {e}")
+                table_uris = []
+
+        if oauth_token:
+            import_args["oauth_bearer_token"] = oauth_token
+        if table_uris:
+            import_args["table_uris"] = table_uris
+    elif args.tables:
+        import_args["tables"] = args.tables
 
-        # Merge them (to normalize and evaluate breaks)
-        merge_result = pipeline.merge_contract_updates(
-            imported,
-            base_contract,
-            context=change_context,
-            fail_on_conflict=False,
-        )
-        merged = merge_result.contract
+    # Import temporary contract from source
+    imported = pipeline.import_schema(
+        source_type=args.type,
+        source=args.source,
+        uc_workspace_url=args.workspace_url,
+        uc_token=args.token,
+        import_args=import_args if import_args else None,
+    )
 
-        # Evaluate decision & ANALYZE operation gate (always allowed for analysis)
-        decision = evaluate_governance_decision(
-            base_contract,
-            merged,
-            context=change_context,
-            merge_conflicts=merge_result.conflicts,
-        )
-        gate_res = evaluate_governance_gate(decision, GovernanceOperation.ANALYZE)
+    # Load base contract (governed target)
+    base_contract = pipeline.loader.load(args.base)
+
+    # Merge them (to normalize and evaluate breaks)
+    merge_result = pipeline.merge_contract_updates(
+        imported,
+        base_contract,
+        context=change_context,
+        fail_on_conflict=False,
+    )
+    merged = merge_result.contract
 
-        if not decision.evidence.has_changes:
-            print("🟢 No changes detected.")
-        else:
-            print(f"📊 Governance Decision: {decision.decision.value} (Gate: {gate_res.reason})")
-            for reason in decision.reasons:
-                print(f"  • [{reason.code}] {reason.path or 'root'}: {reason.message}")
+    # Evaluate decision & ANALYZE operation gate (always allowed for analysis)
+    decision = evaluate_governance_decision(
+        base_contract,
+        merged,
+        context=change_context,
+        merge_conflicts=merge_result.conflicts,
+    )
+    gate_res = evaluate_governance_gate(decision, GovernanceOperation.ANALYZE)
 
-            bump = decision.required_version_bump.upper()
-            if bump == "NONE":
-                print("\n✅ Action Required: No version bump needed.")
-            elif bump == "MINOR":
-                print(f"\n⚠️ Action Required: Additive changes require version bump {bump}.")
-            elif bump == "MAJOR":
-                print(f"\n⚠️ Action Required: Breaking changes require version bump {bump}.")
+    if not decision.evidence.has_changes:
+        print("🟢 No changes detected.")
+    else:
+        print(f"📊 Governance Decision: {decision.decision.value} (Gate: {gate_res.reason})")
+        for reason in decision.reasons:
+            print(f"  • [{reason.code}] {reason.path or 'root'}: {reason.message}")
 
-    except Exception as e:
-        from semapact.exceptions import SemaPactError
-        import logging
+        bump = decision.required_version_bump.upper()
+        if bump == "NONE":
+            print("\n✅ Action Required: No version bump needed.")
+        elif bump == "MINOR":
+            print(f"\n⚠️ Action Required: Additive changes require version bump {bump}.")
+        elif bump == "MAJOR":
+            print(f"\n⚠️ Action Required: Breaking changes require version bump {bump}.")
 
-        if isinstance(e, SemaPactError):
-            logging.getLogger("semapact").error("Plan failed: %s", e)
-        else:
-            logging.getLogger("semapact").error("Plan failed: %s", e, exc_info=True)
-        print(f"❌ Error during plan: {e}")
-        raise SystemExit(1)