Architecting Autonomous AI Agent Workflows in Industrial Manufacturing: Bridging ERP, IoT, and Operational Execution

+-----------------------------------------------------------------------+
|                         ENTERPRISE IT LAYER                           |
|    +-------------------+    +-------------------+    +-----------+    |
|    | SAP S/4HANA (ERP) |    |  WMS / Inventory  |    | Cloud SCM |    |
|    +---------+---------+    +---------+---------+    +-----+-----+    |
+--------------|------------------------|--------------------|----------+
               | REST / SOAP            | gRPC               | GraphQL   
+--------------v------------------------v--------------------v----------+
|                       AI AGENT ORCHESTRATION LAYER                    |
|  +-----------------------------------------------------------------+  |
|  | State Machine Engine & Fault-Tolerant Transaction Manager       |  |
|  +-----------------------------------------------------------------+  |
+--------------^------------------------^--------------------^----------+
               | MQTT / Sparkplug B     | OPC UA             | Modbus    
+--------------|------------------------|--------------------|----------+
|                         OPERATIONAL OT LAYER                          |
|    +---------+---------+    +---------+---------+    +-----+-----+    |
|    |  MES / Dispatch   |    |  SCADA Systems    |    | PLC / Edge|    |
|    +-------------------+    +-------------------+    +-----------+    |
+-----------------------------------------------------------------------+

  [ IIoT Telemetry Stream ] ---->
              |
              v
+----------------------------+
| Anomaly Ingestion Agent    |
+--------------+-------------+
               |
               v
+----------------------------+
| Context Synthesis Agent    | <--- Queries ERP (SAP/Oracle)
+--------------+-------------+
               |
               v
+----------------------------+
| Root Cause Analysis Agent  | <--- Evaluates Causal Graphs
+--------------+-------------+
               |
               v
+----------------------------+
| Operational Planning Agent | <--- Checks Safety / Cost Bounds
+--------------+-------------+
               |
               v
+----------------------------+
| Transaction Execution Agent| ---> Executes SAP / MES Work Orders
+----------------------------+

import logging
from typing import Dict, Any, Optional, List
from enum import Enum
from pydantic import BaseModel, Field

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MachineHealthStatus(str, Enum):
    OPTIMAL = "optimal"
    DEGRADED = "degraded"
    CRITICAL = "critical"


class OperationalActionType(str, Enum):
    MAINTAIN_SCHEDULE = "maintain_schedule"
    THROTTLE_FEED_RATE = "throttle_feed_rate"
    DISPATCH_MAINTENANCE = "dispatch_maintenance"
    EMERGENCY_STOP = "emergency_stop"


class MachineTelemetry(BaseModel):
    """Real-time IIoT telemetry payload from OPC UA / MQTT edge gateway."""
    machine_id: str = Field(..., description="Unique machine identifier on factory floor")
    vibration_velocity_mm_s: float = Field(..., ge=0.0, description="ISO 10816 root-mean-square vibration velocity")
    bearing_temperature_c: float = Field(..., description="Bearing temperature in Celsius")
    spindle_load_percent: float = Field(..., ge=0.0, le=200.0, description="Active motor load percentage")
    feed_rate_actual: float = Field(..., ge=0.0, description="Actual machine feed rate in mm/min")


class ErpProductionContext(BaseModel):
    """Contextual state synchronized from SAP/Oracle ERP and MES."""
    active_batch_id: str = Field(..., description="Active production work order / batch identifier")
    sku_code: str = Field(..., description="Manufactured product stock keeping unit")
    units_completed: int = Field(..., ge=0)
    units_target: int = Field(..., gt=0)
    scheduled_finish_epoch: int = Field(..., description="Unix timestamp of target delivery shift")
    sla_penalty_tier: str = Field(default="STANDARD", description="Financial sensitivity tier")


class StateValidationResult(BaseModel):
    """Deterministic validation decision emitted by Operational Planning Agent."""
    decision_id: str
    machine_id: str
    health_status: MachineHealthStatus
    recommended_action: OperationalActionType
    requires_human_approval: bool
    adjusted_feed_rate: Optional[float] = None
    validation_reasons: List[str] = Field(default_factory=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)


class FaultTolerantStateValidationAgent:
    """
    Deterministic operational planning agent that bridges high-frequency IIoT
    telemetry with enterprise ERP context, enforcing hard safety thresholds
    and human-in-the-loop (HITL) governance boundaries.
    """

    def __init__(
        self,
        vibration_critical_threshold: float = 7.1,  # ISO 10816 Class II threshold
        vibration_warning_threshold: float = 4.5,
        temp_critical_c: float = 95.0,
        temp_warning_c: float = 80.0,
    ):
        self.vib_critical = vibration_critical_threshold
        self.vib_warning = vibration_warning_threshold
        self.temp_critical = temp_critical_c
        self.temp_warning = temp_warning_c

    def evaluate_state(
        self,
        telemetry: MachineTelemetry,
        erp_context: ErpProductionContext,
    ) -> StateValidationResult:
        """
        Validates cross-domain operational state and computes deterministic remediation.
        """
        logger.info(
            "Evaluating machine '%s' on batch '%s' [Vib: %.2f mm/s, Temp: %.1f C, Load: %.1f%%]",
            telemetry.machine_id,
            erp_context.active_batch_id,
            telemetry.vibration_velocity_mm_s,
            telemetry.bearing_temperature_c,
            telemetry.spindle_load_percent,
        )

        reasons: List[str] = []
        action = OperationalActionType.MAINTAIN_SCHEDULE
        health = MachineHealthStatus.OPTIMAL
        requires_hitl = False
        adjusted_feed: Optional[float] = None

        # 1. HARD SAFETY CHECK: Critical physical bounds violation
        if (
            telemetry.vibration_velocity_mm_s >= self.vib_critical
            or telemetry.bearing_temperature_c >= self.temp_critical
        ):
            health = MachineHealthStatus.CRITICAL
            action = OperationalActionType.EMERGENCY_STOP
            requires_hitl = True  # Critical safety trip requires operator acknowledgment
            reasons.append(
                f"Physical operating limit exceeded: Vibration={telemetry.vibration_velocity_mm_s} mm/s "
                f"(limit {self.vib_critical}), Temp={telemetry.bearing_temperature_c} C (limit {self.temp_critical})"
            )
            logger.error("CRITICAL TRIP on %s: %s", telemetry.machine_id, reasons[-1])

        # 2. DEGRADED PERFORMANCE CHECK: Preventative throttling & maintenance trigger
        elif (
            telemetry.vibration_velocity_mm_s >= self.vib_warning
            or telemetry.bearing_temperature_c >= self.temp_warning
            or telemetry.spindle_load_percent > 105.0
        ):
            health = MachineHealthStatus.DEGRADED
            # Calculate non-invasive feed rate reduction to mitigate thermal/mechanical stress
            adjusted_feed = round(telemetry.feed_rate_actual * 0.85, 2)
            action = OperationalActionType.THROTTLE_FEED_RATE
            reasons.append(
                f"Elevated stress detected: Throttling feed rate from {telemetry.feed_rate_actual} "
                f"to {adjusted_feed} mm/min to prevent catastrophic tool wear."
            )

            # High-value or high-penalty batches require human-in-the-loop review before dispatch
            if erp_context.sla_penalty_tier in ("CRITICAL", "EXPEDITE"):
                requires_hitl = True
                reasons.append(
                    f"Batch '{erp_context.active_batch_id}' is flagged as {erp_context.sla_penalty_tier}; "
                    "requires supervisor approval to modify cycle pacing."
                )
            else:
                requires_hitl = False
                reasons.append("Autonomous feed rate dampening applied within pre-cleared envelope.")

        # 3. NOMINAL STATE: Healthy operation confirmed
        else:
            health = MachineHealthStatus.OPTIMAL
            action = OperationalActionType.MAINTAIN_SCHEDULE
            reasons.append("All mechanical and thermal telemetry within nominal operating tolerances.")

        decision = StateValidationResult(
            decision_id=f"DEC-{telemetry.machine_id}-{erp_context.active_batch_id}",
            machine_id=telemetry.machine_id,
            health_status=health,
            recommended_action=action,
            requires_human_approval=requires_hitl,
            adjusted_feed_rate=adjusted_feed,
            validation_reasons=reasons,
            metadata={
                "erp_sku": erp_context.sku_code,
                "batch_progress_percent": round((erp_context.units_completed / erp_context.units_target) * 100, 1),
            },
        )

        logger.info(
            "State validation complete: %s (Action: %s, HITL: %s)",
            decision.decision_id,
            decision.recommended_action.value,
            decision.requires_human_approval,
        )
        return decision