feat: handler contract compliance scanner and models

src/onex_change_control/enums/__init__.py

@@ -7,6 +7,8 @@
     EnumAutopilotCycleStatus,
     EnumAutopilotStepStatus,
 )
+from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict
+from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation
 from onex_change_control.enums.enum_db_boundary import (
     EnumDbBoundaryExceptionStatus,
     EnumDbBoundaryReasonCategory,
@@ -27,13 +29,16 @@
 from onex_change_control.enums.enum_integration_surface import EnumIntegrationSurface
 from onex_change_control.enums.enum_interface_surface import EnumInterfaceSurface
 from onex_change_control.enums.enum_invariant_status import EnumInvariantStatus
+from onex_change_control.enums.enum_migration_status import EnumMigrationStatus
 from onex_change_control.enums.enum_pr_state import EnumPRState
 from onex_change_control.enums.enum_probe_reason import EnumProbeReason
 from onex_change_control.enums.enum_probe_status import EnumProbeStatus
 
 __all__ = [
     "EnumAutopilotCycleStatus",
     "EnumAutopilotStepStatus",
+    "EnumComplianceVerdict",
+    "EnumComplianceViolation",
     "EnumDbBoundaryExceptionStatus",
     "EnumDbBoundaryReasonCategory",
     "EnumDocReferenceType",
@@ -50,6 +55,7 @@
     "EnumIntegrationSurface",
     "EnumInterfaceSurface",
     "EnumInvariantStatus",
+    "EnumMigrationStatus",
     "EnumPRState",
     "EnumProbeReason",
     "EnumProbeStatus",

src/onex_change_control/enums/enum_compliance_verdict.py

@@ -0,0 +1,40 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""Compliance Verdict Enum.
+
+Verdict categories for handler contract compliance audits.
+"""
+
+from enum import Enum, unique
+
+
+@unique
+class EnumComplianceVerdict(str, Enum):
+    """Verdict for a handler's contract compliance status.
+
+    - COMPLIANT: Handler fully adheres to its contract declarations.
+    - IMPERATIVE: Handler bypasses the contract system (hardcoded topics, etc.).
+    - HYBRID: Handler partially uses contracts but has some imperative wiring.
+    - ALLOWLISTED: Handler has known violations tracked in the allowlist.
+    - MISSING_CONTRACT: No contract.yaml exists for the handler's node.
+    """
+
+    COMPLIANT = "compliant"
+    """Handler fully adheres to its contract declarations."""
+
+    IMPERATIVE = "imperative"
+    """Handler bypasses the contract system."""
+
+    HYBRID = "hybrid"
+    """Handler partially uses contracts but has some imperative wiring."""
+
+    ALLOWLISTED = "allowlisted"
+    """Handler has known violations tracked in the allowlist."""
+
+    MISSING_CONTRACT = "missing_contract"
+    """No contract.yaml exists for the handler's node."""
+
+    def __str__(self) -> str:
+        """Return the string value for serialization."""
+        return self.value

src/onex_change_control/enums/enum_compliance_violation.py

@@ -0,0 +1,46 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""Compliance Violation Enum.
+
+Specific violation types for handler contract compliance checks.
+"""
+
+from enum import Enum, unique
+
+
+@unique
+class EnumComplianceViolation(str, Enum):
+    """Specific contract compliance violation types.
+
+    Each value represents a distinct way a handler can bypass
+    the contract-declared dispatch system.
+    """
+
+    HARDCODED_TOPIC = "hardcoded_topic"
+    """Topic string literal in handler instead of contract declaration."""
+
+    UNDECLARED_TRANSPORT = "undeclared_transport"
+    """Handler uses a transport (DB, HTTP, Kafka) not declared in contract."""
+
+    LOGIC_IN_NODE = "logic_in_node"
+    """Business logic found in node.py instead of handler."""
+
+    MISSING_HANDLER_ROUTING = "missing_handler_routing"
+    """Handler exists in handlers/ but is not in contract.yaml handler_routing."""
+
+    UNDECLARED_PUBLISH = "undeclared_publish"
+    """Handler publishes to topic not in contract event_bus.publish_topics."""
+
+    UNDECLARED_SUBSCRIBE = "undeclared_subscribe"
+    """Handler subscribes to topic not in contract event_bus.subscribe_topics."""
+
+    DIRECT_DB_ACCESS = "direct_db_access"
+    """Handler constructs DB connections directly instead of using injected services."""
+
+    UNREGISTERED_HANDLER = "unregistered_handler"
+    """Handler file in handlers/ directory but not importable or registered."""
+
+    def __str__(self) -> str:
+        """Return the string value for serialization."""
+        return self.value

src/onex_change_control/enums/enum_migration_status.py

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""Migration Status Enum.
+
+Status tracking for imperative-to-declarative handler migrations.
+"""
+
+from enum import Enum, unique
+
+
+@unique
+class EnumMigrationStatus(str, Enum):
+    """Status of a handler's migration from imperative to declarative.
+
+    Lifecycle: PENDING -> GENERATED -> VALIDATED -> DEPLOYED -> RETIRED
+
+    - PENDING: Migration identified but not yet started.
+    - GENERATED: Migration spec generated with contract.yaml changes.
+    - VALIDATED: Migration validated via contract-dispatch equivalence.
+    - DEPLOYED: Migrated handler deployed and serving traffic.
+    - RETIRED: Old imperative wiring fully removed.
+    """
+
+    PENDING = "pending"
+    """Migration identified but not yet started."""
+
+    GENERATED = "generated"
+    """Migration spec generated with contract.yaml changes."""
+
+    VALIDATED = "validated"
+    """Migration validated via contract-dispatch equivalence."""
+
+    DEPLOYED = "deployed"
+    """Migrated handler deployed and serving traffic."""
+
+    RETIRED = "retired"
+    """Old imperative wiring fully removed."""
+
+    def __str__(self) -> str:
+        """Return the string value for serialization."""
+        return self.value

src/onex_change_control/models/__init__.py

@@ -16,6 +16,10 @@
     ModelAutopilotCycleRecord,
     ModelAutopilotStepResult,
 )
+from onex_change_control.models.model_compliance_sweep_report import (
+    ModelComplianceSweepReport,
+    ModelRepoComplianceBreakdown,
+)
 from onex_change_control.models.model_contract_drift_input import (
     ModelContractDriftInput,
 )
@@ -70,10 +74,17 @@
     ModelGoldenPathInput,
     ModelGoldenPathOutput,
 )
+from onex_change_control.models.model_handler_compliance_result import (
+    ModelHandlerComplianceResult,
+)
 from onex_change_control.models.model_integration_record import (
     ModelIntegrationProbeResult,
     ModelIntegrationRecord,
 )
+from onex_change_control.models.model_migration_spec import (
+    ModelMigrationSpec,
+    ModelMigrationValidationResult,
+)
 from onex_change_control.models.model_ticket_contract import (
     ModelDodCheck,
     ModelDodEvidenceItem,
@@ -85,6 +96,7 @@
     "EnumAutopilotStepStatus",
     "ModelAutopilotCycleRecord",
     "ModelAutopilotStepResult",
+    "ModelComplianceSweepReport",
     "ModelContractDriftInput",
     "ModelContractDriftOutput",
     "ModelDayClose",
@@ -116,8 +128,12 @@
     "ModelGoldenPathAssertion",
     "ModelGoldenPathInput",
     "ModelGoldenPathOutput",
+    "ModelHandlerComplianceResult",
     "ModelIntegrationProbeResult",
     "ModelIntegrationRecord",
+    "ModelMigrationSpec",
+    "ModelMigrationValidationResult",
+    "ModelRepoComplianceBreakdown",
     "ModelRepoDocSummary",
     "ModelTicketContract",
 ]

src/onex_change_control/models/model_compliance_sweep_report.py

@@ -0,0 +1,74 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""Compliance Sweep Report Model.
+
+Aggregated sweep report across all handlers in one or more repos.
+"""
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from onex_change_control.models.model_handler_compliance_result import (
+    ModelHandlerComplianceResult,
+)
+
+
+class ModelRepoComplianceBreakdown(BaseModel):
+    """Per-repo breakdown of compliance results."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    repo: str = Field(..., description="Repository name")
+    total_handlers: int = Field(..., description="Total handlers scanned", ge=0)
+    compliant: int = Field(..., description="Compliant handler count", ge=0)
+    imperative: int = Field(..., description="Imperative handler count", ge=0)
+    hybrid: int = Field(..., description="Hybrid handler count", ge=0)
+    top_violations: list[str] = Field(
+        default_factory=list,
+        description="Most common violation types in this repo",
+    )
+
+
+class ModelComplianceSweepReport(BaseModel):
+    """Aggregated compliance sweep report.
+
+    Contains counts, ratios, per-repo breakdown, and full audit details
+    for all handlers scanned across one or more repositories.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    timestamp: datetime = Field(
+        ...,
+        description="When the sweep was performed",
+    )
+    total_handlers: int = Field(..., description="Total handlers scanned", ge=0)
+    compliant_count: int = Field(..., description="Compliant handler count", ge=0)
+    imperative_count: int = Field(..., description="Imperative handler count", ge=0)
+    hybrid_count: int = Field(..., description="Hybrid handler count", ge=0)
+    allowlisted_count: int = Field(..., description="Allowlisted handler count", ge=0)
+    missing_contract_count: int = Field(
+        ..., description="Missing contract handler count", ge=0
+    )
+    compliant_pct: float = Field(
+        ...,
+        description="Percentage of compliant handlers (0-100)",
+    )
+    violation_histogram: dict[str, int] = Field(
+        default_factory=dict,
+        description="Count per EnumComplianceViolation value",
+    )
+    per_repo: dict[str, ModelRepoComplianceBreakdown] = Field(
+        default_factory=dict,
+        description="Breakdown by repository",
+    )
+    results: list[ModelHandlerComplianceResult] = Field(
+        default_factory=list,
+        description="Full audit details",
+    )
+    new_violations_since_last: list[str] = Field(
+        default_factory=list,
+        description="Handlers that gained violations since last sweep",
+    )

src/onex_change_control/models/model_handler_compliance_result.py

@@ -0,0 +1,84 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""Handler Compliance Result Model.
+
+Single handler audit result for contract compliance checks.
+"""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from onex_change_control.enums.enum_compliance_verdict import EnumComplianceVerdict
+from onex_change_control.enums.enum_compliance_violation import EnumComplianceViolation
+
+
+class ModelHandlerComplianceResult(BaseModel):
+    """Audit result for a single handler's contract compliance.
+
+    Captures the cross-reference between a handler's actual behavior
+    (topics used, transports imported, routing registration) and what
+    is declared in its contract.yaml.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    handler_path: str = Field(
+        ...,
+        description="Relative path to handler file",
+    )
+    node_dir: str = Field(
+        ...,
+        description="Parent node directory",
+    )
+    repo: str = Field(
+        ...,
+        description="Repository name",
+    )
+    contract_path: str | None = Field(
+        default=None,
+        description="Path to associated contract.yaml (None if missing)",
+    )
+    violations: list[EnumComplianceViolation] = Field(
+        default_factory=list,
+        description="Specific violations found",
+    )
+    violation_details: list[str] = Field(
+        default_factory=list,
+        description="Human-readable detail per violation",
+    )
+    declared_topics: list[str] = Field(
+        default_factory=list,
+        description="Topics declared in contract.yaml",
+    )
+    used_topics: list[str] = Field(
+        default_factory=list,
+        description="Topics referenced in handler source code",
+    )
+    undeclared_topics: list[str] = Field(
+        default_factory=list,
+        description="Topics used but not declared",
+    )
+    declared_transports: list[str] = Field(
+        default_factory=list,
+        description="Transports declared in contract capabilities",
+    )
+    used_transports: list[str] = Field(
+        default_factory=list,
+        description="Transports detected in handler imports/calls",
+    )
+    undeclared_transports: list[str] = Field(
+        default_factory=list,
+        description="Transports used but not declared",
+    )
+    handler_in_routing: bool = Field(
+        default=False,
+        description="Whether handler is registered in contract.yaml handler_routing",
+    )
+    verdict: EnumComplianceVerdict = Field(
+        ...,
+        description="Overall compliance verdict",
+    )
+    allowlisted: bool = Field(
+        default=False,
+        description="Whether this handler is in the allowlist",
+    )