diff --git a/src/main/java/com/factoreal/backend/domain/abnormalLog/application/AbnormalLogService.java b/src/main/java/com/factoreal/backend/domain/abnormalLog/application/AbnormalLogService.java
index 08515f92..697b2f07 100644
--- a/src/main/java/com/factoreal/backend/domain/abnormalLog/application/AbnormalLogService.java
+++ b/src/main/java/com/factoreal/backend/domain/abnormalLog/application/AbnormalLogService.java
@@ -4,6 +4,8 @@
 import com.factoreal.backend.domain.abnormalLog.dto.TargetType;
 import com.factoreal.backend.domain.abnormalLog.dto.request.AbnormalPagingRequest;
 import com.factoreal.backend.domain.abnormalLog.dto.response.AbnormalLogResponse;
+import com.factoreal.backend.domain.sensor.dto.SensorKafkaDto;
+import com.factoreal.backend.domain.sensor.entity.Sensor;
 import com.factoreal.backend.domain.abnormalLog.entity.AbnormalLog;
 import com.factoreal.backend.domain.sensor.dto.SensorKafkaDto;
 import com.factoreal.backend.domain.zone.application.ZoneHistoryService;
@@ -50,6 +52,7 @@ public class AbnormalLogService {
      * @throws Exception
      */
     @Transactional(rollbackFor = Exception.class)
+
     public AbnormalLog saveAbnormalLogFromSensorKafkaDto(
             SensorKafkaDto sensorKafkaDto,
             SensorType sensorType,
@@ -62,10 +65,10 @@ public AbnormalLog saveAbnormalLogFromSensorKafkaDto(
             throw new ResponseStatusException(HttpStatus.NOT_FOUND, "존재하지 않는 공간 ID: " + sensorKafkaDto.getZoneId());
         }
 
-        log.info(">>>>>> zone : {} " ,zone);
+        log.info(">>>>>> zone : {} ", zone);
 
         // DTO의 severity (AlarmEvent.RiskLevel)를 Entity RiskLevel로 매핑
-//        RiskLevel entityRiskLevel = mapDtoSeverityToEntityRiskLevel(riskLevel);
+        // RiskLevel entityRiskLevel = mapDtoSeverityToEntityRiskLevel(riskLevel);
         // [TODO] 현재는 스프린트 1 웹 푸쉬, 대시보드 히트 맵 알림 로그만 구현되있음. worker, equip 로그용 구현 필요.
         AbnormalLog abnormalLog = AbnormalLog.builder()
                 .targetId(sensorKafkaDto.getSensorId())
@@ -97,8 +100,7 @@ public AbnormalLog saveAbnormalLogFromWearableKafkaDto(
             TargetType targetType
     ){
         // workerId에 해당되는 사람이 제일 최근에 있던 공간 조회
-        ZoneHist zonehist = zoneHistoryService.
-                findByWorker_WorkerIdAndExistFlag(wearableKafkaDto.getWorkerId(), 1);
+        ZoneHist zonehist = zoneHistoryService.getCurrentWorkerLocation(wearableKafkaDto.getWorkerId());
         Zone zone;
         if (zonehist == null){
             zone = zoneService.findByZoneId("00000000000000-000");
@@ -133,7 +135,8 @@ public Page<AbnormalLogResponse> findAllAbnormalLogsUnRead(AbnormalPagingRequest
         return abnormalLogs.map(AbnormalLog::fromEntity);
     }
 
-    public Page<AbnormalLogResponse> findAbnormalLogsByAbnormalType(AbnormalPagingRequest abnormalPagingRequest, String abnormalType){
+    public Page<AbnormalLogResponse> findAbnormalLogsByAbnormalType(AbnormalPagingRequest abnormalPagingRequest,
+            String abnormalType) {
         // 한번에 DB전체를 주는 것이 아닌 구간 나눠서 전달하기 위함
         Pageable pageable = getPageable(abnormalPagingRequest);
         Page<AbnormalLog> abnormalLogs = abnLogRepository.findAbnormalLogsByAbnormalType(abnormalType,pageable);
@@ -158,16 +161,14 @@ public Page<AbnormalLogResponse> findAbnormalLogsByTargetId(AbnormalPagingReques
                 targetId,
                 pageable);
         return abnormalLogs.map(
-                abn_log -> objectMapper.convertValue(abn_log, AbnormalLogResponse.class)
-        );
+                abn_log -> objectMapper.convertValue(abn_log, AbnormalLogResponse.class));
     }
 
-
     // FE에서 알람을 클릭한 경우 읽음으로 수정
     @Transactional
-    public boolean readCheck(Long abnormalLogId){
+    public boolean readCheck(Long abnormalLogId) {
         AbnormalLog abnormalLog = abnLogRepository.findById(abnormalLogId).orElse(null);
-        if(abnormalLog == null){
+        if (abnormalLog == null) {
             return false;
         }
 
@@ -176,18 +177,33 @@ public boolean readCheck(Long abnormalLogId){
         readRequired();
         return true;
     }
+    @Transactional
+    public AbnormalLog saveAbnormalLog(SensorKafkaDto dto, Sensor sensor, int dangerLevel) {
+        RiskLevel riskLevel = RiskLevel.fromPriority(dangerLevel);
+
+        AbnormalLog abnormalLog = AbnormalLog.builder()
+                .targetId(dto.getSensorId())
+                .targetType(TargetType.Sensor)
+                .abnormalType(riskMessageProvider.getRiskMessageBySensor(sensor.getSensorType(), riskLevel))
+                .abnVal(dto.getVal())
+                .zone(sensor.getZone())
+                .detectedAt(LocalDateTime.parse(dto.getTime()))
+                .isRead(false)
+                .build();
+
+        return abnLogRepository.save(abnormalLog);
+    }
     @Transactional(readOnly = true)
     // 읽지 않은 알람이 몇개인지 반환
-    public Long readRequired(){
-        Long count =  abnLogRepository.countByIsReadFalse();
-//        webSocketSender.sendUnreadCount(count);
+    public Long readRequired() {
+        Long count = abnLogRepository.countByIsReadFalse();
+        // webSocketSender.sendUnreadCount(count);
         return count;
     }
 
-    private Pageable getPageable(AbnormalPagingRequest abnormalPagingRequest){
+    private Pageable getPageable(AbnormalPagingRequest abnormalPagingRequest) {
         return PageRequest.of(
                 abnormalPagingRequest.getPage(),
-                abnormalPagingRequest.getSize()
-        );
+                abnormalPagingRequest.getSize());
     }
 }
diff --git a/src/main/java/com/factoreal/backend/domain/controlLog/entity/ControlLog.java b/src/main/java/com/factoreal/backend/domain/controlLog/entity/ControlLog.java
index 77bd352b..87e63bc0 100644
--- a/src/main/java/com/factoreal/backend/domain/controlLog/entity/ControlLog.java
+++ b/src/main/java/com/factoreal/backend/domain/controlLog/entity/ControlLog.java
@@ -25,7 +25,7 @@ public class ControlLog {
     private String controlType;
 
     @Column(name = "control_val")
-    private Integer controlVal;
+    private Double controlVal;
 
     @Column(name = "control_stat")
     private Integer controlStat;
@@ -44,5 +44,4 @@ public class ControlLog {
     @JoinColumn(name = "zone_id", referencedColumnName = "zone_id")
     private Zone zone;
 
-
 }
diff --git a/src/main/java/com/factoreal/backend/domain/controlLog/repository/ControlLogRepository.java b/src/main/java/com/factoreal/backend/domain/controlLog/repository/ControlLogRepository.java
new file mode 100644
index 00000000..45e75ce8
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/controlLog/repository/ControlLogRepository.java
@@ -0,0 +1,10 @@
+package com.factoreal.backend.domain.controlLog.repository;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import com.factoreal.backend.domain.controlLog.entity.ControlLog;
+
+/**
+ * 자동제어 로그 저장을 위한 레포지토리
+ */
+public interface ControlLogRepository extends JpaRepository<ControlLog, Long> {
+}
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/controlLog/service/ControlLogService.java b/src/main/java/com/factoreal/backend/domain/controlLog/service/ControlLogService.java
new file mode 100644
index 00000000..462c6e63
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/controlLog/service/ControlLogService.java
@@ -0,0 +1,68 @@
+package com.factoreal.backend.domain.controlLog.service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.factoreal.backend.domain.abnormalLog.entity.AbnormalLog;
+import com.factoreal.backend.domain.controlLog.entity.ControlLog;
+import com.factoreal.backend.domain.controlLog.repository.ControlLogRepository;
+import com.factoreal.backend.domain.zone.entity.Zone;
+import com.factoreal.backend.messaging.mqtt.MqttPublishService;
+import com.factoreal.backend.messaging.sender.WebSocketSender;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+// 제어 로그 저장 및 알림 전송 서비스
+public class ControlLogService {
+
+  private final ControlLogRepository controlLogRepository;
+  private final MqttPublishService mqttPublishService;
+  private final WebSocketSender webSocketSender;
+
+  @Transactional
+  public ControlLog saveControlLog(AbnormalLog abnormalLog, String controlType, Double controlVal, Integer controlStat,
+      Zone zone) {
+    ControlLog controlLog = ControlLog.builder()
+        .abnormalLog(abnormalLog)
+        .controlType(controlType)
+        .controlVal(controlVal)
+        .controlStat(controlStat)
+        .executedAt(LocalDateTime.now())
+        .zone(zone)
+        .build();
+
+    // 제어 로그 저장
+    ControlLog savedLog = controlLogRepository.save(controlLog);
+
+    // 발송 여부를 포함할 맵
+    Map<String, Boolean> deliveryStatus = new HashMap<>();
+
+    try {
+      // MQTT 메시지 발행
+      mqttPublishService.publishControlMessage(savedLog);
+      deliveryStatus.put("mqttDelivered", true);
+    } catch (Exception e) {
+      log.error("❌ MQTT 메시지 발행 실패: {}", e.getMessage(), e);
+      deliveryStatus.put("mqttDelivered", false);
+    }
+
+    try {
+      // WebSocket으로 제어 상태 전송 (발송 상태 포함)
+      webSocketSender.sendControlStatus(savedLog, deliveryStatus);
+
+      log.info("✅ 제어 로그 저장 및 알림 전송 완료: controlId={}, abnormalId={}, status={}",
+          savedLog.getId(), abnormalLog.getId(), deliveryStatus);
+    } catch (Exception e) {
+      log.error("❌ WebSocket 메시지 전송 실패: {}", e.getMessage(), e);
+    }
+
+    return savedLog;
+  }
+}
diff --git a/src/main/java/com/factoreal/backend/domain/sensor/api/SensorController.java b/src/main/java/com/factoreal/backend/domain/sensor/api/SensorController.java
index 8c3e654d..bd2611fb 100644
--- a/src/main/java/com/factoreal/backend/domain/sensor/api/SensorController.java
+++ b/src/main/java/com/factoreal/backend/domain/sensor/api/SensorController.java
@@ -30,6 +30,6 @@ public ResponseEntity<Void> update(
             @PathVariable("sensorId") String sensorId,
             @RequestBody SensorUpdateRequest sensorUpdateRequest) {
         service.updateSensor(sensorId, sensorUpdateRequest);
-        return ResponseEntity.ok().build();
+        return ResponseEntity.ok().build(); // 204 응답 반환
     }
 }
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/worker/api/WorkerController.java b/src/main/java/com/factoreal/backend/domain/worker/api/WorkerController.java
index 6b48d3b6..b64b32d3 100644
--- a/src/main/java/com/factoreal/backend/domain/worker/api/WorkerController.java
+++ b/src/main/java/com/factoreal/backend/domain/worker/api/WorkerController.java
@@ -1,6 +1,7 @@
 package com.factoreal.backend.domain.worker.api;
 
 import com.factoreal.backend.domain.worker.application.WorkerService;
+import com.factoreal.backend.domain.worker.dto.request.CreateWorkerRequest;
 import com.factoreal.backend.domain.worker.dto.response.WorkerDetailResponse;
 import com.factoreal.backend.domain.worker.dto.response.WorkerInfoResponse;
 import com.factoreal.backend.domain.worker.dto.response.ZoneManagerResponse;
@@ -8,10 +9,9 @@
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
 
 import java.util.List;
 
@@ -22,14 +22,23 @@
 @Tag(name = "작업자 API", description = "작업자 조회 API")
 public class WorkerController {
     private final WorkerService workerService;
-    
-    @Operation(summary = "전체 작업자 목록 조회", description = "전체 작업자 목록을 조회합니다.")
+
+    @Operation(summary = "작업자 생성", description = "새로운 작업자를 생성하고 접근 가능한 공간들을 선택합니다.")
+    @PostMapping
+    public ResponseEntity<Void> createWorker(@RequestBody CreateWorkerRequest request) {
+        log.info("작업자 생성 요청: {}", request);
+        workerService.createWorker(request);
+        return ResponseEntity.ok().build(); // 작업자 생성 성공 시 200 응답
+    }
+
+    @Operation(summary = "전체 작업자 목록 조회", description = "전체 작업자 목록과 각 작업자의 상태 및 위치 정보를 조회합니다.")
     @GetMapping
-    public List<WorkerDetailResponse> getAllWorkers() {
+    public ResponseEntity<List<WorkerDetailResponse>> getAllWorkers() {
         log.info("전체 작업자 목록 조회 요청");
-        return workerService.getAllWorkers();
+        List<WorkerDetailResponse> workers = workerService.getAllWorkers();
+        return ResponseEntity.ok(workers);
     }
-    
+
     @Operation(summary = "공간별 작업자 목록 조회", description = "공간 ID를 기반으로 현재 해당 공간에 들어가있는 작업자 리스트를 조회합니다.")
     @GetMapping("/zone/{zoneId}")
     public List<WorkerInfoResponse> getWorkersByZoneId(@PathVariable String zoneId) {
@@ -37,11 +46,10 @@ public List<WorkerInfoResponse> getWorkersByZoneId(@PathVariable String zoneId)
         return workerService.getWorkersByZoneId(zoneId);
     }
 
-    @Operation(summary = "공간 담당자 정보 조회", 
-              description = "공간 ID를 기반으로 해당 공간의 담당자와 현재 위치 정보를 조회합니다.")
+    @Operation(summary = "공간 담당자와 담당자의 현재 위치정보 조회", description = "공간 ID를 기반으로 해당 공간의 담당자와 현재 위치 정보를 조회합니다.")
     @GetMapping("/zone/{zoneId}/manager")
     public ZoneManagerResponse getZoneManager(@PathVariable String zoneId) {
         log.info("공간 ID: {}의 담당자 정보 조회 요청", zoneId);
         return workerService.getZoneManagerWithLocation(zoneId);
     }
-} 
\ No newline at end of file
+}
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/worker/api/WorkerManagerController.java b/src/main/java/com/factoreal/backend/domain/worker/api/WorkerManagerController.java
new file mode 100644
index 00000000..5aa90013
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/worker/api/WorkerManagerController.java
@@ -0,0 +1,52 @@
+package com.factoreal.backend.domain.worker.api;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+import com.factoreal.backend.domain.worker.application.WorkerManagerService;
+import com.factoreal.backend.domain.worker.dto.response.WorkerManagerResponse;
+
+import java.util.List;
+
+@Tag(name = "공간 담당자 API", description = "공간별 담당자 관리 API")
+@Slf4j
+@RestController
+@RequestMapping("/api/zone-managers")
+@RequiredArgsConstructor
+public class WorkerManagerController {
+
+  private final WorkerManagerService workerManagerService;
+
+  @Operation(summary = "공간 담당자 후보 목록 조회", description = "특정 공간의 담당자로 지정 가능한 작업자 목록을 조회합니다.")
+  @GetMapping("/candidates/{zoneId}")
+  public ResponseEntity<List<WorkerManagerResponse>> getManagerCandidates(
+      @Parameter(description = "공간 ID", required = true) @PathVariable String zoneId) {
+    log.info("공간 ID: {}의 담당자 후보 목록 조회 요청", zoneId);
+    return ResponseEntity.ok(workerManagerService.getManagerCandidates(zoneId));
+  }
+
+  @Operation(summary = "공간 담당자 지정", description = "특정 공간의 담당자를 지정합니다.")
+  @PostMapping("/{zoneId}/assign/{workerId}")
+  public ResponseEntity<Void> assignManager(
+      @Parameter(description = "공간 ID", required = true) @PathVariable String zoneId,
+      @Parameter(description = "작업자 ID", required = true) @PathVariable String workerId) {
+    log.info("공간 ID: {}의 담당자를 작업자 ID: {}로 지정 요청", zoneId, workerId);
+    workerManagerService.assignManager(zoneId, workerId);
+    return ResponseEntity.ok().build();
+  }
+
+  @Operation(summary = "현재 공간 담당자 정보 조회", description = "특정 공간의 현재 담당자 정보를 조회합니다.")
+  @GetMapping("/{zoneId}")
+  public ResponseEntity<WorkerManagerResponse> getCurrentManager(
+      @Parameter(description = "공간 ID", required = true) @PathVariable String zoneId) {
+    log.info("공간 ID: {}의 현재 담당자 조회 요청", zoneId);
+    WorkerManagerResponse manager = workerManagerService.getCurrentManager(zoneId);
+    return manager != null ? ResponseEntity.ok(manager) : ResponseEntity.noContent().build();
+    // 담당자가 있으면 담당자 정보 반환, 없으면 빈 응답
+  }
+}
diff --git a/src/main/java/com/factoreal/backend/domain/worker/application/WorkerManagerService.java b/src/main/java/com/factoreal/backend/domain/worker/application/WorkerManagerService.java
new file mode 100644
index 00000000..9e39ab9e
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/worker/application/WorkerManagerService.java
@@ -0,0 +1,106 @@
+package com.factoreal.backend.domain.worker.application;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.factoreal.backend.domain.worker.dao.WorkerRepository;
+import com.factoreal.backend.domain.worker.dao.WorkerZoneRepository;
+import com.factoreal.backend.domain.worker.dto.response.WorkerManagerResponse;
+import com.factoreal.backend.domain.worker.entity.Worker;
+import com.factoreal.backend.domain.worker.entity.WorkerZone;
+import com.factoreal.backend.domain.worker.entity.WorkerZoneId;
+import com.factoreal.backend.domain.zone.dao.ZoneRepository;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+// 공간 담당자 지정 서비스
+public class WorkerManagerService {
+
+  private final WorkerRepository workerRepository;
+  private final WorkerZoneRepository workerZoneRepository;
+  private final ZoneRepository zoneRepository;
+
+  /**
+   * 특정 공간의 담당자 후보 목록 조회
+   * - 이미 담당자가 있는 경우: 현재 담당자를 제외한 해당 공간 접근 권한이 있는 작업자 목록
+   * - 담당자가 없는 경우: 해당 공간 접근 권한이 있는 작업자 목록
+   * - 다른 공간의 담당자인 작업자는 후보 목록에서 제외
+   */
+  @Transactional(readOnly = true)
+  public List<WorkerManagerResponse> getManagerCandidates(String zoneId) {
+    log.info("공간 ID: {}의 담당자 후보 목록 조회", zoneId);
+
+    // 1. 현재 해당 공간의 담당자 조회
+    Optional<WorkerZone> currentManager = workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId);
+
+    // 2. 현재 공간을 제외한 다른 공간의 담당자 목록 조회 (workerId를 Set으로 묶어서 중복 제거)
+    Set<String> otherManagerIds = workerZoneRepository.findByZoneZoneIdNotAndManageYnIsTrue(zoneId).stream()
+        .map(wz -> wz.getWorker().getWorkerId())
+        .collect(Collectors.toSet());
+
+    // 3. 해당 공간에 접근 권한이 있는 작업자 목록 조회
+    List<WorkerZone> zoneWorkers = workerZoneRepository.findByZoneZoneId(zoneId);
+
+    // 4. 현재 담당자와 다른 공간의 담당자를 제외한 후보 목록 생성
+    List<Worker> candidates = currentManager
+        .map(manager -> zoneWorkers.stream()
+            .filter(wz -> !wz.getWorker().getWorkerId().equals(manager.getWorker().getWorkerId())) // 현재 담당자를 제외
+            .filter(wz -> !otherManagerIds.contains(wz.getWorker().getWorkerId())) // 다른 공간의 담당자인 작업자는 후보 목록에서 제외
+            .map(WorkerZone::getWorker) // 후보 목록에 포함된 작업자 객체 생성
+            .collect(Collectors.toList()))
+        .orElse(zoneWorkers.stream() // 담당자가 없는 경우, 해당 공간 접근 권한이 있는 작업자 목록
+            .filter(wz -> !otherManagerIds.contains(wz.getWorker().getWorkerId())) // 다른 공간의 담당자인 작업자는 후보 목록에서 제외
+            .map(WorkerZone::getWorker) // 후보 목록에 포함된 작업자 객체 생성
+            .collect(Collectors.toList()));
+
+    // 4. DTO 변환 (후보 목록의 작업자들은 현재 이 공간의 담당자가 아니므로 isManager = false)
+    return candidates.stream()
+        .map(worker -> WorkerManagerResponse.fromEntity(worker, false))
+        .collect(Collectors.toList());
+  }
+
+  /**
+   * 특정 공간의 담당자 지정
+   */
+  @Transactional
+  public void assignManager(String zoneId, String workerId) {
+    log.info("공간 ID: {}의 담당자를 작업자 ID: {}로 지정", zoneId, workerId);
+
+    // 1. 작업자-공간 관계 확인
+    WorkerZoneId workerZoneId = new WorkerZoneId(workerId, zoneId);
+    WorkerZone workerZone = workerZoneRepository.findById(workerZoneId)
+        .orElseThrow(() -> new IllegalArgumentException(
+            String.format("작업자 ID: %s는 공간 ID: %s에 대한 접근 권한이 없습니다.", workerId, zoneId)));
+
+    // 2. 기존 담당자가 있다면 담당자 해제
+    Optional<WorkerZone> currentManager = workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId);
+    currentManager.ifPresent(manager -> {
+      manager.setManageYn(false);
+      workerZoneRepository.save(manager);
+    });
+
+    // 3. 새로운 담당자 지정
+    workerZone.setManageYn(true);
+    workerZoneRepository.save(workerZone);
+  }
+
+  /**
+   * 특정 공간의 현재 담당자 조회
+   */
+  @Transactional(readOnly = true)
+  public WorkerManagerResponse getCurrentManager(String zoneId) {
+    log.info("공간 ID: {}의 현재 담당자 조회", zoneId);
+
+    return workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId)
+        .map(workerZone -> WorkerManagerResponse.fromEntity(workerZone.getWorker(), true))
+        .orElse(null);
+  }
+}
diff --git a/src/main/java/com/factoreal/backend/domain/worker/application/WorkerService.java b/src/main/java/com/factoreal/backend/domain/worker/application/WorkerService.java
index b368109a..35c6b342 100644
--- a/src/main/java/com/factoreal/backend/domain/worker/application/WorkerService.java
+++ b/src/main/java/com/factoreal/backend/domain/worker/application/WorkerService.java
@@ -1,14 +1,18 @@
 package com.factoreal.backend.domain.worker.application;
 
+import com.factoreal.backend.domain.worker.dto.request.CreateWorkerRequest;
+import com.factoreal.backend.domain.worker.dto.response.WorkerDetailResponse;
 import com.factoreal.backend.domain.abnormalLog.application.AbnormalLogService;
 import com.factoreal.backend.domain.abnormalLog.dto.TargetType;
 import com.factoreal.backend.domain.abnormalLog.dto.response.AbnormalLogResponse;
-import com.factoreal.backend.domain.worker.dto.response.WorkerDetailResponse;
 import com.factoreal.backend.domain.worker.dto.response.WorkerInfoResponse;
 import com.factoreal.backend.domain.worker.dto.response.ZoneManagerResponse;
+import com.factoreal.backend.domain.zone.application.ZoneHistoryService;
 import com.factoreal.backend.domain.zone.dao.ZoneHistoryRepository;
+import com.factoreal.backend.domain.zone.dao.ZoneRepository;
 import com.factoreal.backend.domain.worker.entity.Worker;
 import com.factoreal.backend.domain.worker.entity.WorkerZone;
+import com.factoreal.backend.domain.worker.entity.WorkerZoneId;
 import com.factoreal.backend.domain.zone.entity.Zone;
 import com.factoreal.backend.domain.zone.entity.ZoneHist;
 import com.factoreal.backend.domain.worker.dao.WorkerRepository;
@@ -18,6 +22,7 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -27,9 +32,11 @@
 @RequiredArgsConstructor
 public class WorkerService {
     private final WorkerRepository workerRepository;
-    private final ZoneHistoryRepository zoneHistoryRepository;
     private final WorkerZoneRepository workerZoneRepository;
+    private final ZoneRepository zoneRepository;
+    private final ZoneHistoryService zoneHistoryService;
     private final AbnormalLogService abnormalLogService;
+    private final ZoneHistoryRepository zoneHistoryRepository;
     @Transactional(readOnly = true)
     public List<WorkerDetailResponse> getAllWorkers() {
         log.info("전체 작업자 목록 조회");
@@ -52,23 +59,32 @@ public List<WorkerDetailResponse> getAllWorkers() {
             .collect(Collectors.toMap(AbnormalLogResponse::getTargetId, AbnormalLogResponse::getDangerLevel));
 
         // 위치 Map<workerId, zoneName>
-        Map<String, String> zoneMap = workerIds.stream()
+        Map<String, Map<String, String>> zoneMap = workerIds.stream()
             .collect(Collectors.toMap(
                 workerId -> workerId,
                 workerId -> {
                     ZoneHist zh = zoneHistoryRepository.findByWorker_WorkerIdAndExistFlag(workerId, 1);
                     if (zh == null || zh.getZone() == null) {
-                        return "대기실"; // 기본 ZoneId
+                        Map<String, String> defaultZone = new HashMap<>();
+                        defaultZone.put("zoneId", "00000000000000-000");
+                        defaultZone.put("zoneName", "대기실");
+                        return defaultZone;
                     }
-                    return zh.getZone().getZoneName(); // zoneName이 아니라 zoneId로 변경
+                    Map<String, String> zone = new HashMap<>();
+                    zone.put("zoneId", zh.getZone().getZoneId());
+                    zone.put("zoneName", zh.getZone().getZoneName());
+                    return zone;
                 }
             ));
         return workers.stream()
-            .map(worker -> {
-                Integer status = statusMap.getOrDefault(worker.getWorkerId(), 0); // 기본값 예: 정상
-                String zone = zoneMap.getOrDefault(worker.getWorkerId(), "대기실");
-                return WorkerDetailResponse.from(worker, false, status.toString(), zone);
-            })
+            .map(worker -> WorkerDetailResponse.fromEntity(
+                worker,
+                workerZoneRepository.findByWorkerWorkerIdAndManageYnIsTrue(worker.getWorkerId())
+                            .isPresent(),
+                statusMap.get(worker.getWorkerId()),
+                zoneMap.get(worker.getWorkerId()).get("zoneId"),
+                zoneMap.get(worker.getWorkerId()).get("zoneName")
+            ))
             .collect(Collectors.toList());
     }
 
@@ -90,12 +106,12 @@ public List<WorkerInfoResponse> getWorkersByZoneId(String zoneId) {
     @Transactional(readOnly = true)
     public ZoneManagerResponse getZoneManagerWithLocation(String zoneId) {
         log.info("공간 ID: {}의 담당자 정보 조회", zoneId);
-        
+
         WorkerZone zoneManager = workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId)
                 .orElseThrow(() -> new IllegalArgumentException("해당 공간의 담당자를 찾을 수 없습니다: " + zoneId));
-        
+
         Worker manager = zoneManager.getWorker();
-        
+
         // 2. 담당자의 현재 위치 조회 (existFlag = 1)
         ZoneHist currentLocation = zoneHistoryRepository.findByWorker_WorkerIdAndExistFlag(manager.getWorkerId(), 1);
 
@@ -104,6 +120,43 @@ public ZoneManagerResponse getZoneManagerWithLocation(String zoneId) {
 
         return ZoneManagerResponse.from(manager, currentZone);
     }
+
+    /**
+     * 작업자 생성 및 출입 가능 공간 설정
+     */
+    @Transactional
+    public void createWorker(CreateWorkerRequest request) {
+        log.info("작업자 생성 요청: {}", request);
+
+        // 1. 작업자 정보 저장
+        Worker worker = Worker.builder()
+                .workerId(request.getWorkerId())
+                .name(request.getName())
+                .phoneNumber(request.getPhoneNumber())
+                .email(request.getEmail())
+                .build();
+
+        workerRepository.save(worker); // 작업자 정보 저장
+
+        // 2. 각 공간명으로 Zone 조회 및 WorkerZone 생성
+        for (String zoneName : request.getZoneNames()) {
+            Zone zone = zoneRepository.findByZoneName(zoneName)
+                    .orElseThrow(() -> new IllegalArgumentException("존재하지 않는 공간명입니다: " + zoneName));
+
+            // WorkerZone 생성 (기본적으로 관리자 권한은 없음)
+            WorkerZone workerZone = WorkerZone.builder()
+                    .id(new WorkerZoneId(worker.getWorkerId(), zone.getZoneId())) // 복합키 생성
+                    .worker(worker)
+                    .zone(zone)
+                    .manageYn(false) // 담당자 권한은 없음이 default
+                    .build();
+
+            workerZoneRepository.save(workerZone); // WorkerZone 저장
+        }
+
+        log.info("작업자 생성 완료 - workerId: {}", worker.getWorkerId());
+    }
+
     /**
      *  workerId에 해당하는 작업자 조회
      */
diff --git a/src/main/java/com/factoreal/backend/domain/worker/dao/WorkerZoneRepository.java b/src/main/java/com/factoreal/backend/domain/worker/dao/WorkerZoneRepository.java
index ccda65b1..680171a1 100644
--- a/src/main/java/com/factoreal/backend/domain/worker/dao/WorkerZoneRepository.java
+++ b/src/main/java/com/factoreal/backend/domain/worker/dao/WorkerZoneRepository.java
@@ -8,10 +8,16 @@
 import java.util.Optional;
 
 public interface WorkerZoneRepository extends JpaRepository<WorkerZone, WorkerZoneId> {
-    
+
     // 특정 zone_id에 속한 작업자 목록 조회
     List<WorkerZone> findByZoneZoneId(String zoneId);
-    
+
     // 특정 zone_id의 담당자 조회 (manageYn = true)
     Optional<WorkerZone> findByZoneZoneIdAndManageYnIsTrue(String zoneId);
-} 
\ No newline at end of file
+
+    // 특정 공간을 제외한 다른 공간의 담당자 목록 조회
+    List<WorkerZone> findByZoneZoneIdNotAndManageYnIsTrue(String zoneId);
+
+    // 특정 작업자가 담당자로 있는 공간 조회
+    Optional<WorkerZone> findByWorkerWorkerIdAndManageYnIsTrue(String workerId);
+}
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/worker/dto/request/CreateWorkerRequest.java b/src/main/java/com/factoreal/backend/domain/worker/dto/request/CreateWorkerRequest.java
new file mode 100644
index 00000000..3a4bf8b9
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/worker/dto/request/CreateWorkerRequest.java
@@ -0,0 +1,21 @@
+package com.factoreal.backend.domain.worker.dto.request;
+
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+// 작업자 생성 요청 DTO (FE -> BE)
+public class CreateWorkerRequest {
+  private String workerId; // 작업자 ID (사원 번호)
+  private String name; // 작업자 이름
+  private String phoneNumber; // 연락처
+  private String email; // 이메일
+  private List<String> zoneNames; // 출입 가능한 공간명 리스트
+}
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerDetailResponse.java b/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerDetailResponse.java
index fb90c905..14c8e32f 100644
--- a/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerDetailResponse.java
+++ b/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerDetailResponse.java
@@ -1,6 +1,7 @@
 package com.factoreal.backend.domain.worker.dto.response;
 
 import com.factoreal.backend.domain.worker.entity.Worker;
+
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -11,19 +12,22 @@
 @NoArgsConstructor
 @SuperBuilder
 public class WorkerDetailResponse extends WorkerInfoResponse{
-    // 작업자의 위치와 상태 추가
-    private String status;
-    private String zone;
-    public static WorkerDetailResponse from(Worker worker, Boolean isManager, String status, String zone) {
-        return WorkerDetailResponse.builder()
-            .workerId(worker.getWorkerId())
-            .name(worker.getName())
-            .phoneNumber(worker.getPhoneNumber())
-            .email(worker.getEmail())
-            .fcmToken(worker.getFcmToken())
-            .status(status)
-            .zone(zone)
-            .isManager(isManager)
-            .build();
-    }
+  private String status; // 작업자 상태
+  private String currentZoneId; // 현재 위치한 공간 ID
+  private String currentZoneName; // 현재 위치한 공간 이름
+
+  // Entity -> DTO 변환
+  public static WorkerDetailResponse fromEntity(Worker worker, Boolean isManager, Integer status, String currentZoneId,
+          String currentZoneName) {
+    return WorkerDetailResponse.builder()
+        .workerId(worker.getWorkerId())
+        .name(worker.getName())
+        .phoneNumber(worker.getPhoneNumber())
+        .email(worker.getEmail())
+        .isManager(isManager)
+        .status(status != null ? status.toString() : null)
+        .currentZoneId(currentZoneId)
+        .currentZoneName(currentZoneName)
+        .build();
+  }
 }
diff --git a/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerManagerResponse.java b/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerManagerResponse.java
new file mode 100644
index 00000000..b3c01d0d
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/domain/worker/dto/response/WorkerManagerResponse.java
@@ -0,0 +1,33 @@
+package com.factoreal.backend.domain.worker.dto.response;
+
+import com.factoreal.backend.domain.worker.entity.Worker;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+// 공간 담당자 정보 응답 DTO (BE -> FE)
+// - 담당자 후보 목록 조회
+// - 담당자 정보 조회
+public class WorkerManagerResponse {
+  private String workerId; // 직원 아이디
+  private String name; // 직원 이름
+  private String email; // 이메일
+  private String phoneNumber; // 전화번호
+  private Boolean isManager; // 공간담당자 여부
+
+  public static WorkerManagerResponse fromEntity(Worker worker, Boolean isManager) {
+    return WorkerManagerResponse.builder()
+        .workerId(worker.getWorkerId())
+        .name(worker.getName())
+        .email(worker.getEmail())
+        .phoneNumber(worker.getPhoneNumber())
+        .isManager(isManager)
+        .build();
+  }
+}
diff --git a/src/main/java/com/factoreal/backend/domain/zone/application/ZoneHistoryService.java b/src/main/java/com/factoreal/backend/domain/zone/application/ZoneHistoryService.java
index c40eec68..d1823722 100644
--- a/src/main/java/com/factoreal/backend/domain/zone/application/ZoneHistoryService.java
+++ b/src/main/java/com/factoreal/backend/domain/zone/application/ZoneHistoryService.java
@@ -12,58 +12,69 @@
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
+import java.util.List;
 
 @Slf4j
 @Service
 @RequiredArgsConstructor
 public class ZoneHistoryService {
-    
-    private final ZoneHistoryRepository zoneHistRepository;
-    private final WorkerRepository workerRepository;
-    private final ZoneRepository zoneRepository;
 
-    /**
-     * 작업자의 위치 변경을 처리
-     */
-    @Transactional
-    public void updateWorkerLocation(String workerId, String zoneId, LocalDateTime timestamp) {
-        Worker worker = workerRepository.findById(workerId)
-                .orElseThrow(() -> new IllegalArgumentException("작업자를 찾을 수 없습니다: " + workerId));
-        
-        Zone zone = zoneRepository.findById(zoneId)
-                .orElseThrow(() -> new IllegalArgumentException("공간을 찾을 수 없습니다: " + zoneId));
+        private final ZoneHistoryRepository zoneHistRepository;
+        private final WorkerRepository workerRepository;
+        private final ZoneRepository zoneRepository;
 
-        // 1. workerId 기반 작업자의 이전 위치가 있으면, 새로운 기록 생성 전 해당 작업자 위치 기록에 endTime 찍어주기
-        ZoneHist currentLocation = zoneHistRepository.findByWorker_WorkerIdAndExistFlag(workerId, 1);
-        if (currentLocation != null) {
-            currentLocation.setEndTime(timestamp); // 다음 공간의 입장 시간으로 update
-            currentLocation.setExistFlag(0);
-            zoneHistRepository.save(currentLocation);
-        }
-
-        // 2. 새로운 위치 기록 생성
-        ZoneHist newLocation = ZoneHist.builder()
-                .worker(worker)
-                .zone(zone)
-                .startTime(timestamp)
-                .endTime(null)
-                .existFlag(1)
-                .build();
-        
-        zoneHistRepository.save(newLocation);
-        
         /**
-         * currentLocation이 있으면 (이전 위치가 있으면) -> 그 공간의 ID를 출력
-         * currentLocation이 없으면 (최초 입장이면) -> "없음" 출력
+         * 작업자의 위치 변경을 처리
          */
-        log.info("작업자 {} 위치 변경: {} -> {}", workerId, 
-                currentLocation != null ? currentLocation.getZone().getZoneId() : "없음", 
-                zoneId);
-    }
+        @Transactional
+        public void updateWorkerLocation(String workerId, String zoneId, LocalDateTime timestamp) {
+                Worker worker = workerRepository.findById(workerId)
+                                .orElseThrow(() -> new IllegalArgumentException("작업자를 찾을 수 없습니다: " + workerId));
+
+                Zone zone = zoneRepository.findById(zoneId)
+                                .orElseThrow(() -> new IllegalArgumentException("공간을 찾을 수 없습니다: " + zoneId));
+
+                // 1. workerId 기반 작업자의 이전 위치가 있으면, 새로운 기록 생성 전 해당 작업자 위치 기록에 endTime 찍어주기
+                ZoneHist currentLocation = zoneHistRepository.findByWorker_WorkerIdAndExistFlag(workerId, 1);
+                if (currentLocation != null) {
+                        currentLocation.setEndTime(timestamp); // 다음 공간의 입장 시간으로 update
+                        currentLocation.setExistFlag(0);
+                        zoneHistRepository.save(currentLocation);
+                }
 
-    public ZoneHist findByWorker_WorkerIdAndExistFlag(String workerid, int existFlag) {
-        return zoneHistRepository.findByWorker_WorkerIdAndExistFlag(workerid,existFlag);
-    }
+                // 2. 새로운 위치 기록 생성
+                ZoneHist newLocation = ZoneHist.builder()
+                                .worker(worker)
+                                .zone(zone)
+                                .startTime(timestamp)
+                                .endTime(null)
+                                .existFlag(1)
+                                .build();
 
+                zoneHistRepository.save(newLocation);
 
+                /**
+                 * currentLocation이 있으면 (이전 위치가 있으면) -> 그 공간의 ID를 출력
+                 * currentLocation이 없으면 (최초 입장이면) -> "없음" 출력
+                 */
+                log.info("작업자 {} 위치 변경: {} -> {}", workerId,
+                                currentLocation != null ? currentLocation.getZone().getZoneId() : "없음",
+                                zoneId);
+        }
+
+        /**
+         * 특정 공간에 현재 들어가있는 작업자 리스트 조회
+         */
+        @Transactional(readOnly = true)
+        public List<ZoneHist> getCurrentWorkersByZoneId(String zoneId) {
+                return zoneHistRepository.findByZone_ZoneIdAndExistFlag(zoneId, 1); // 해당 공간의 existFlag가 1인 모든 작업자 리스트
+        }
+
+        /**
+         * 특정 작업자의 현재 위치 조회
+         */
+        @Transactional(readOnly = true)
+        public ZoneHist getCurrentWorkerLocation(String workerId) {
+                return zoneHistRepository.findByWorker_WorkerIdAndExistFlag(workerId, 1);
+        }
 }
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/domain/zone/dto/request/ZoneHistoryRequest.java b/src/main/java/com/factoreal/backend/domain/zone/dto/request/ZoneHistoryRequest.java
index 4bbb2ad8..439411c3 100644
--- a/src/main/java/com/factoreal/backend/domain/zone/dto/request/ZoneHistoryRequest.java
+++ b/src/main/java/com/factoreal/backend/domain/zone/dto/request/ZoneHistoryRequest.java
@@ -9,7 +9,7 @@
 @Getter
 @Setter
 @ToString
-// Wearable 장치에서 받아오는 데이터 by 우영. 추후 논의 예정
+// Wearable 장치에서 받아오는 데이터 by 우영
 public class ZoneHistoryRequest {
     private String workerId;
     private String zoneId;
diff --git a/src/main/java/com/factoreal/backend/messaging/config/MqttConfig.java b/src/main/java/com/factoreal/backend/messaging/config/MqttConfig.java
index 22244ea9..7082a222 100644
--- a/src/main/java/com/factoreal/backend/messaging/config/MqttConfig.java
+++ b/src/main/java/com/factoreal/backend/messaging/config/MqttConfig.java
@@ -16,15 +16,15 @@ public class MqttConfig {
     public MqttClient mqttClient(SslUtil sslUtil) throws Exception {
         // 🟢 AWS IoT 브로커 주소 및 포트 설정
         String broker = "ssl://a2q1cmw33m6k7u-ats.iot.ap-northeast-2.amazonaws.com:8883";
-        // 🟢 고유한 MQTT 클라이언트 ID 생성
-        String clientId = "SPRING_Dain";
+        // 🟢 고유한 MQTT 클라이언트 ID 생성 (랜덤 suffix 추가)
+        String clientId = "SPRING_Dain_" + System.currentTimeMillis();
         // 🔐 SSL 인증서 경로 설정
         SSLSocketFactory sslFactory;
         try {
             // AWS Secret Manager에 정의된 secret 식별자 사용
             sslFactory = sslUtil.getSocketFactoryFromSecrets("monitory/dev/iotSecrets");
             log.info("✅Secret Manager에서 Pem키 가져오기 성공!");
-        }catch (Exception e){
+        } catch (Exception e) {
             // AWS Secret Manager에 Pem이 등록되지 않았다면 그대로 로컬의 pem키 사용.
             log.info("❌Secret Manager에서 Pem키 가져오기 실패 {}", e.getMessage());
             sslFactory = sslUtil.getSocketFactoryFromFiles(
@@ -34,18 +34,44 @@ public MqttClient mqttClient(SslUtil sslUtil) throws Exception {
             );
             log.info("✅로컬 경로에서 Pem키 가져오기 성공!");
         }
+
         MqttConnectOptions options = new MqttConnectOptions();
         options.setSocketFactory(sslFactory);
         // 영구 저장소 비활성화
         options.setCleanSession(true);
         // 자동 재연결 설정
         options.setAutomaticReconnect(true);
-        // 연결 타임아웃 설정 (5초)
-        options.setConnectionTimeout(5);
+        // 연결 타임아웃 설정 (30초)
+        options.setConnectionTimeout(30);
+        // Keep Alive 간격 설정 (60초)
+        options.setKeepAliveInterval(60);
+        // 최대 인플라이트 메시지 수 설정
+        options.setMaxInflight(100);
+        // 연결 재시도 간격 설정
+        options.setMaxReconnectDelay(5000);
 
         MqttClient client = new MqttClient(broker, clientId, null);
-        client.connect(options);
-        log.info("✅Mqtt 연결 성공!");
+        
+        try {
+            client.connect(options);
+            // 연결 성공 확인
+            int attempts = 0;
+            while (!client.isConnected() && attempts < 5) {
+                Thread.sleep(2000);
+                attempts++;
+                log.info("MQTT 연결 대기 중... (시도 {}/5)", attempts);
+            }
+            
+            if (!client.isConnected()) {
+                throw new Exception("MQTT 클라이언트 연결 실패");
+            }
+            
+            log.info("✅Mqtt 연결 성공! (clientId: {})", clientId);
+        } catch (Exception e) {
+            log.error("❌Mqtt 연결 실패: {}", e.getMessage());
+            throw e;
+        }
+        
         return client;
     }
 }
diff --git a/src/main/java/com/factoreal/backend/messaging/kafka/KafkaConsumerD.java b/src/main/java/com/factoreal/backend/messaging/kafka/KafkaConsumerD.java
deleted file mode 100644
index 3bfc078a..00000000
--- a/src/main/java/com/factoreal/backend/messaging/kafka/KafkaConsumerD.java
+++ /dev/null
@@ -1,317 +0,0 @@
-package com.factoreal.backend.messaging.kafka;
-
-import com.factoreal.backend.domain.sensor.dto.SensorKafkaDto;
-import com.factoreal.backend.domain.abnormalLog.dto.TargetType;
-import com.factoreal.backend.domain.abnormalLog.entity.AbnormalLog;
-import com.factoreal.backend.domain.zone.dao.ZoneRepository;
-import com.factoreal.backend.messaging.common.dto.SystemLogDto;
-import com.factoreal.backend.messaging.sender.WebSocketSender;
-import com.factoreal.backend.domain.zone.application.ZoneService;
-import com.factoreal.backend.domain.abnormalLog.application.AbnormalLogService;
-import com.factoreal.backend.messaging.kafka.strategy.alarmList.NotificationStrategy;
-import com.factoreal.backend.messaging.kafka.strategy.NotificationStrategyFactory;
-import com.factoreal.backend.messaging.kafka.strategy.enums.AlarmEventDto;
-import com.factoreal.backend.messaging.kafka.strategy.enums.RiskLevel;
-import com.factoreal.backend.messaging.kafka.strategy.enums.SensorType;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Service;
-import com.factoreal.backend.domain.sensor.application.SensorService;
-import com.factoreal.backend.domain.sensor.entity.Sensor;
-
-import java.time.ZonedDateTime;
-import java.time.ZoneId;
-
-import java.time.Instant;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-@Service
-@Slf4j
-@RequiredArgsConstructor
-public class KafkaConsumerD {
-
-    private final ObjectMapper objectMapper;
-    private final WebSocketSender webSocketSender;
-    private final ZoneRepository zoneRepository;
-    private final ZoneService zoneService;
-
-    // 알람 푸시 용
-    private final NotificationStrategyFactory factory;
-
-    // 공간(zone)별로 마지막 위험도 저장하기 위한 Map (초기에는 위험도 -1)
-    private static final Map<String, Integer> lastDangerLevelMap = new ConcurrentHashMap<>();
-
-    // 로그 기록용
-    private final AbnormalLogService abnormalLogService;
-
-    private final SensorService sensorService;
-
-    // @KafkaListener(topics = {"EQUIPMENT", "ENVIRONMENT"}, groupId =
-    // "monitory-consumer-group-1")
-//    @KafkaListener(topics = { "EQUIPMENT",
-//            "ENVIRONMENT" }, groupId = "${spring.kafka.consumer.group-id:danger-alert-group}")
-    public void consume(String message) {
-
-        log.info("💡수신한 Kafka 메시지 : " + message);
-        try {
-            SensorKafkaDto dto = objectMapper.readValue(message, SensorKafkaDto.class);
-
-            // 시스템 로그 (위험도 변화 감지 -> 비동기 전송)
-            // sendSystemLog(dto);
-
-            // 공간 센서일 때만 히트맵용 웹소켓 전송
-            if (dto.getEquipId() != null && dto.getZoneId() != null && dto.getEquipId().equals(dto.getZoneId())) {
-                log.info("✅ 공간 센서 로직 start");
-
-                log.info("▶︎ 위험도 감지 start");
-                int dangerLevel = getDangerLevel(dto.getSensorType(), dto.getVal());
-                log.info("⚠️ 위험도 {} 센서 타입 : {} 감지됨. Zone: {}", dangerLevel, dto.getSensorType(), dto.getZoneId());
-
-                // 자동제어 로직: threshold 및 오차범위 벗어나면 메시지 전송
-                // 중첩 try-catch 문 : Kafka 메시지 처리에서 자동제어 로직은 실패해도, 전체 처리는 멈추지 않게 하기 위해
-                performAutoControl(dto);
-
-                // #################################
-                // Abnormal 로그 기록 로직
-                // #################################
-                SensorType sensorType = SensorType.getSensorType(dto.getSensorType());
-                RiskLevel riskLevel = RiskLevel.fromPriority(dangerLevel);
-                if (sensorType == null) {
-                    log.error("SensorType not found");
-                    throw new Exception("SensorType not found");
-                }
-                AbnormalLog abnormalLog = abnormalLogService.saveAbnormalLogFromSensorKafkaDto(
-                        dto,
-                        sensorType,
-                        riskLevel,
-                        TargetType.Sensor);
-
-                // #################################
-                // 웹 앱 SMS 알람 로직
-                // #################################
-                startAlarm(dto, abnormalLog, riskLevel);
-
-                // #################################
-                // 대시보드용 히트맵 로직
-                // #################################
-                // ❗dangerLevel이 0일 때도 전송해야되면 if 문은 필요없을 것 같아 제거.
-
-                webSocketSender.sendDangerLevel(dto.getZoneId(), dto.getSensorType(), dangerLevel);
-                abnormalLogService.readRequired(); // 읽지 않은 알람 수
-            }
-
-        } catch (Exception e) {
-            log.error("❌ Kafka 메시지 파싱 실패: {}", message, e);
-        }
-
-    }
-
-    @Async
-    public void startAlarm(SensorKafkaDto sensorData, AbnormalLog abnormalLog, RiskLevel riskLevel) {
-        AlarmEventDto alarmEventDto;
-        try {
-            // 1. dangerLevel기준으로 alarmEvent 객체 생성.
-            alarmEventDto = generateAlarmDto(sensorData, abnormalLog, riskLevel);
-        } catch (Exception e) {
-            log.error("Error converting Kafka message: {}", e);
-            return;
-        }
-        // 1-1. AbnormalLog 기록.
-        try {
-            // 2. 생성된 AlarmEvent DTO 객체를 사용하여 알람 처리
-            log.info("alarmEvent: {}", alarmEventDto.toString());
-            processAlarmEvent(alarmEventDto, riskLevel);
-        } catch (Exception e) {
-            log.error("Error converting Kafka message: {}", e);
-            // TODO: 기타 처리 오류 처리
-        }
-    }
-
-    // 공간(zone)별 위험도 변경 시 시스템 로그 전송
-    @Async
-    @Deprecated
-    public void sendSystemLog(SensorKafkaDto dto) {
-        String zoneId = dto.getZoneId();
-        int newLevel = getDangerLevel(dto.getSensorType(), dto.getVal());
-        int oldLevel = lastDangerLevelMap.getOrDefault(zoneId, -1);
-
-        // 변경이 없으면 로그 전송 안함
-        if (newLevel == oldLevel) {
-            lastDangerLevelMap.put(zoneId, newLevel);
-            return;
-        }
-        lastDangerLevelMap.put(zoneId, newLevel); // 변경이 있으니 해당 공간의 마지막 위험도 업데이트
-
-        // zoneName 조회
-        String zoneName = zoneService.getAllZones().stream()
-                .filter(zone -> zone.getZoneId().equals(zoneId))
-                .findFirst()
-                .map(zone -> zone.getZoneName())
-                .orElse("");
-
-        // ISO-8601 포맷 타임스탬프 ex) 2025-05-09T16:22:45
-        // String timestamp = LocalDateTime.now()
-        //         .format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-
-        String timestamp = ZonedDateTime
-                .now(ZoneId.of("Asia/Seoul"))
-                .format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-
-        SystemLogDto logDto = new SystemLogDto(
-                zoneId, zoneName,
-                dto.getSensorType(),
-                newLevel,
-                dto.getVal(),       // 이 부분 추가
-                timestamp);
-
-        webSocketSender.sendSystemLog(logDto);
-    }
-
-    private static int getDangerLevel(String sensorType, double value) { // 위험도 계산 메서드
-        return switch (sensorType) { // 센서 타입에 따른 위험도 계산
-            case "temp" -> { // 온도 위험도 기준 (KOSHA: https://www.kosha.or.kr/)
-                if (value > 40 || value < -35) // >40℃ 또는 < -35℃ → 위험 (작업 중단 권고)
-                    yield 2;
-                else if (value > 30 || value < 25) // >30℃ 또는 < 25℃ → 주의 (작업 제한 또는 휴식 권고)
-                    yield 1;
-                else // 25℃ ≤ value ≤ 30℃ → 안전 (권장 18~21℃)
-                    yield 0;
-            }
-
-            case "humid" -> { // 상대습도 위험도 기준 (OSHA, ACGIH TLV®, NIOSH)
-                if (value >= 80) // RH ≥ 80% → 위험
-                    yield 2;
-                else if (value >= 60) // 60% ≤ RH < 80% → 주의
-                    yield 1;
-                else // RH < 60% → 안전
-                    yield 0;
-            }
-
-            case "vibration" -> { // 진동 위험도 기준 (ISO 10816-3)
-                if (value > 7.1) // >7.1 mm/s → 위험
-                    yield 2;
-                else if (value > 2.8) // >2.8 mm/s → 주의
-                    yield 1;
-                else // ≤2.8 mm/s → 안전
-                    yield 0;
-            }
-
-            case "current" -> { // 전류 위험도 기준 (KEPCO)
-                if (value >= 30) // ≥30 mA → 위험 (강한 경련, 심실세동 및 사망 위험)
-                    yield 2;
-                else if (value >= 7) // ≥7 mA → 주의 (고통 한계 전류, 불수전류)
-                    yield 1;
-                else // <7 mA → 안전 (감지전류 수준)
-                    yield 0;
-            }
-
-            case "dust" -> { // PM2.5 위험도 기준 (고용노동부)
-                if (value >= 150) // ≥ 150㎍/㎥ → 위험
-                    yield 2;
-                else if (value >= 75) // ≥ 75㎍/㎥ → 주의
-                    yield 1;
-                else // < 75㎍/㎥ → 안전
-                    yield 0;
-            }
-
-            // 그 외 센서 타입은 안전
-            default -> 0;
-        };
-    }
-
-    private AlarmEventDto generateAlarmDto(SensorKafkaDto data, AbnormalLog abnormalLog, RiskLevel riskLevel)
-            throws Exception {
-
-        String source = data.getZoneId().equals(data.getEquipId()) ? "공간 센서" : "설비 센서";
-        SensorType sensorType = SensorType.valueOf(data.getSensorType());
-        String zoneName = zoneRepository.findByZoneId(data.getZoneId()).getZoneName();
-        // 알람 이벤트 객체 반환
-        return AlarmEventDto.builder()
-                .eventId(abnormalLog.getId())
-                .sensorId(data.getSensorId())
-                .equipId(data.getEquipId())
-                .zoneId(data.getZoneId())
-                .sensorType(sensorType.name())
-                .messageBody(abnormalLog.getAbnormalType())
-                .source(source)
-                .time(data.getTime())
-                .riskLevel(riskLevel)
-                .zoneName(zoneName)
-                .build();
-    }
-
-    private void processAlarmEvent(AlarmEventDto alarmEventDto, RiskLevel riskLevel) {
-        if (alarmEventDto == null || alarmEventDto.getRiskLevel() == null) {
-            log.warn("Received null AlarmEvent DTO or DTO with null severity. Skipping notification.");
-            return;
-        }
-
-        try {
-            if (riskLevel == null) {
-                log.warn("Could not map DTO severity '{}' to Entity RiskLevel. Skipping notification.",
-                        alarmEventDto.getRiskLevel());
-
-                // TODO: 매핑 실패 시 처리 로직 추가
-                return;
-            }
-
-            log.info("Processing AlarmEvent with mapped Entity RiskLevel: {}", riskLevel);
-
-            // 3. Factory를 사용하여 매핑된 Entity RiskLevel에 해당하는 NotificationStrategy를 가져와 실행
-            List<NotificationStrategy> notificationStrategyList = factory.getStrategiesForLevel(riskLevel);
-
-            log.info("💡Notification strategy executed for AlarmEvent. \n{}", alarmEventDto.toString());
-            // 4. 알람 객체의 값으로 전략별 알람 송신.
-            notificationStrategyList.forEach(notificationStrategy -> notificationStrategy.send(alarmEventDto));
-
-        } catch (Exception e) {
-            log.error("Failed to execute notification strategy for AlarmEvent DTO: {}", alarmEventDto, e);
-            // TODO: 전략 실행 중 오류 처리
-        }
-    }
-
-    /**
-     * 측정값이 (threshold ± allowVal) 범위를 벗어나면 제어 메시지 생성
-     */
-    private void performAutoControl(SensorKafkaDto dto) {
-        try {
-            Sensor sensor = sensorService.getSensorById(dto.getSensorId());
-            String type = sensor.getSensorType().name();
-            double threshold = sensor.getSensorThres();
-            double tolerance = sensor.getAllowVal() != null ? sensor.getAllowVal() : 0.0;
-            double value = dto.getVal();
-
-            if (value < threshold - tolerance || value > threshold + tolerance) {
-                String msg = buildControlMessage(type, value, threshold, tolerance);
-                log.info("[자동제어 메시지] {}", msg);
-                // TODO: MQTT 퍼블리시 로직으로 대체
-            }
-        } catch (Exception e) {
-            log.error("자동제어 오류 처리 중 예외 발생", e);
-        }
-    }
-
-    private String buildControlMessage(  // 제어 로직
-                                         String type, double val, double thresh, double tol) {
-        return switch (type.toLowerCase()) {
-            case "temp" -> String.format("현재 온도는 %.1f℃입니다. 적정 온도 범위는 %.1f~%.1f℃입니다.",
-                    val, thresh - tol, thresh + tol);
-            case "humid" -> String.format("현재 습도는 %.1f%%입니다. 적정 습도 범위는 %.1f~%.1f%%입니다.",
-                    val, thresh - tol, thresh + tol);
-            case "vibration" -> String.format("현재 진동 값은 %.1fmm/s입니다. 허용 범위는 %.1f~%.1fmm/s입니다.",
-                    val, thresh - tol, thresh + tol);
-            case "current" -> String.format("현재 전류는 %.1fmA입니다. 허용 범위는 %.1f~%.1fmA입니다.",
-                    val, thresh - tol, thresh + tol);
-            case "dust" -> String.format("현재 미세먼지는 %.1f㎍/㎥입니다. 허용 범위는 %.1f~%.1f㎍/㎥입니다.",
-                    val, thresh - tol, thresh + tol);
-            default -> String.format("현재 값은 %.1f이고, 허용 범위는 %.1f~%.1f입니다.",
-                    val, thresh - tol, thresh + tol);
-        };
-
-    }
-}
\ No newline at end of file
diff --git a/src/main/java/com/factoreal/backend/messaging/kafka/processor/SensorEventProcessor.java b/src/main/java/com/factoreal/backend/messaging/kafka/processor/SensorEventProcessor.java
index 727e3885..915c47fc 100644
--- a/src/main/java/com/factoreal/backend/messaging/kafka/processor/SensorEventProcessor.java
+++ b/src/main/java/com/factoreal/backend/messaging/kafka/processor/SensorEventProcessor.java
@@ -28,7 +28,8 @@ public class SensorEventProcessor {
 
     /**
      * 센서 Kafka 메시지 처리
-     * @param dto 센서 데이터
+     * 
+     * @param dto   센서 데이터
      * @param topic Kafka 토픽명 (EQUIPMENT, ENVIRONMENT)
      */
     public void process(SensorKafkaDto dto, String topic) {
@@ -66,7 +67,6 @@ public void process(SensorKafkaDto dto, String topic) {
                         dto, sensorType, riskLevel, targetType
                 );
 
-
                 // 읽지 않은 알림 수 조회
                 Long count = abnormalLogService.readRequired();
 
@@ -74,13 +74,12 @@ public void process(SensorKafkaDto dto, String topic) {
                 // 1. 히트맵 전송
                 webSocketSender.sendDangerLevel(dto.getZoneId(), dto.getSensorType(), dangerLevel);
                 // 2. 위험 알림 전송 -> 위험도별 Websocket + wearable + Slack(SMS 대체)
-//                webSocketSender.sendDangerAlarm(abnLog.toAlarmEventDto());
-                alarmEventService.startAlarm(dto,abnLog,dangerLevel);
+                // Todo : (As-is) 전략 기반 startAlarm() 메서드 담당자 확인 필요
+                // webSocketSender.sendDangerAlarm(abnLog.toAlarmEventDto());
+                alarmEventService.startAlarm(dto, abnLog, dangerLevel);
                 // 3. 읽지 않은 수 전송
                 webSocketSender.sendUnreadCount(count);
 
-
-
                 log.info("✅ 센서 이벤트 처리 완료: sensorId={}, zoneId={}, level={} ({} topic)",
                         dto.getSensorId(), dto.getZoneId(), dangerLevel, topic);
             }
@@ -90,15 +89,22 @@ public void process(SensorKafkaDto dto, String topic) {
     }
 
     // 공간에 위험도 분기 로직
-    // Todo : Flink에서 적용으로 변경되어 삭제 예정
+    // Flink에서 적용으로 변경되어 사용안함
+    @Deprecated
     private int getDangerLevel(String sensorType, double val) {
         switch (sensorType.toLowerCase()) {
-            case "temp": return val > 40 || val < -35 ? 2 : (val > 30 || val < 25 ? 1 : 0);
-            case "humid": return val >= 80 ? 2 : (val >= 60 ? 1 : 0);
-            case "vibration": return val > 7.1 ? 2 : (val > 2.8 ? 1 : 0);
-            case "current": return val >= 30 ? 2 : (val >= 7 ? 1 : 0);
-            case "dust": return val >= 150 ? 2 : (val >= 75 ? 1 : 0);
-            default: return 0;
+            case "temp":
+                return val > 40 || val < -35 ? 2 : (val > 30 || val < 25 ? 1 : 0);
+            case "humid":
+                return val >= 80 ? 2 : (val >= 60 ? 1 : 0);
+            case "vibration":
+                return val > 7.1 ? 2 : (val > 2.8 ? 1 : 0);
+            case "current":
+                return val >= 30 ? 2 : (val >= 7 ? 1 : 0);
+            case "dust":
+                return val >= 150 ? 2 : (val >= 75 ? 1 : 0);
+            default:
+                return 0;
         }
     }
 
diff --git a/src/main/java/com/factoreal/backend/messaging/mqtt/MqttPublishService.java b/src/main/java/com/factoreal/backend/messaging/mqtt/MqttPublishService.java
new file mode 100644
index 00000000..55be5fcc
--- /dev/null
+++ b/src/main/java/com/factoreal/backend/messaging/mqtt/MqttPublishService.java
@@ -0,0 +1,54 @@
+package com.factoreal.backend.messaging.mqtt;
+
+import com.factoreal.backend.domain.controlLog.entity.ControlLog;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.stereotype.Service;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class MqttPublishService {
+  private final MqttClient mqttClient;
+  private final ObjectMapper objectMapper;
+
+  public void publishControlMessage(ControlLog controlLog) {
+    try {
+      // MQTT 메시지 페이로드 구성
+      Map<String, Object> payload = new HashMap<>();
+      payload.put("controlId", controlLog.getId());
+      payload.put("controlType", controlLog.getControlType());
+      payload.put("controlValue", controlLog.getControlVal());
+      payload.put("controlStatus", controlLog.getControlStat());
+      payload.put("executedAt", controlLog.getExecutedAt().toString());
+      payload.put("zoneId", controlLog.getZone().getZoneId());
+
+      // AbnormalLog 정보 추가
+      payload.put("abnormalId", controlLog.getAbnormalLog().getId());
+      payload.put("abnormalType", controlLog.getAbnormalLog().getAbnormalType());
+      payload.put("abnormalValue", controlLog.getAbnormalLog().getAbnVal());
+
+      // 토픽 구성 - control/{targetType}/{targetId} 형식
+      String targetType = controlLog.getAbnormalLog().getTargetType().name().toLowerCase();
+      String targetId = controlLog.getAbnormalLog().getTargetId();
+      String topic = String.format("control/%s/%s", targetType, targetId);
+
+      // JSON 변환 및 메시지 발행
+      String jsonPayload = objectMapper.writeValueAsString(payload);
+      MqttMessage message = new MqttMessage(jsonPayload.getBytes());
+      message.setQos(1); // QoS 레벨 설정 (최소 1회 전달 보장)
+
+      mqttClient.publish(topic, message);
+      log.info("✅ MQTT 제어 메시지 발행 완료: topic={}, payload={}", topic, jsonPayload);
+
+    } catch (Exception e) {
+      log.error("❌ MQTT 메시지 발행 실패: {}", e.getMessage(), e);
+    }
+  }
+}
diff --git a/src/main/java/com/factoreal/backend/messaging/mqtt/MqttService.java b/src/main/java/com/factoreal/backend/messaging/mqtt/MqttService.java
index fc1a6f68..55594bf1 100644
--- a/src/main/java/com/factoreal/backend/messaging/mqtt/MqttService.java
+++ b/src/main/java/com/factoreal/backend/messaging/mqtt/MqttService.java
@@ -20,6 +20,8 @@
 public class MqttService {
     private final MqttClient mqttClient;
     private final SensorService sensorService;
+    private static final int MAX_RETRY_ATTEMPTS = 5;
+    private static final long RETRY_DELAY_MS = 2000;
 
     /**
      * - 디바이스의 shadow 메타데이터 변경사항(등록/수정)을 구독
@@ -27,40 +29,76 @@ public class MqttService {
      */
     @PostConstruct
     public void SensorShadowSubscription() throws MqttException {
-        // 🟢 구독할 Thing 설정
-        String thingName = "Sensor";
-        // #는 topic의 여러 level을 대체 가능, +는 topic의 단일 level을 대체 가능
-        String topic = "$aws/things/" + thingName + "/shadow/name/+/update/documents";
-        mqttClient.subscribe(topic,1,  (t, msg) -> {
-            String payload = new String(msg.getPayload(), StandardCharsets.UTF_8);
+        int retryCount = 0;
+        while (retryCount < MAX_RETRY_ATTEMPTS) {
+            try {
+                if (!mqttClient.isConnected()) {
+                    log.warn("MQTT 클라이언트가 연결되어 있지 않습니다. 재연결 시도 중... (시도 {}/{})", 
+                            retryCount + 1, MAX_RETRY_ATTEMPTS);
+                    mqttClient.reconnect();
+                    Thread.sleep(RETRY_DELAY_MS);
+                }
 
-            // JSON 파싱 및 DB 저장은 이후 구현 예정
-            try{
-                ObjectMapper mapper = new ObjectMapper();
-                JsonNode jsonNode = mapper.readTree(payload);
-                // mqtt에서 전달되는 뎁스를 따라가야함
-                JsonNode reported  = jsonNode.at("/current/state/reported");
-                log.info("📥 MQTT 수신 (topic: {}): {}", t, jsonNode);
+                // 🟢 구독할 Thing 설정
+                String thingName = "Sensor";
+                // #는 topic의 여러 level을 대체 가능, +는 topic의 단일 level을 대체 가능
+                String topic = "$aws/things/" + thingName + "/shadow/name/+/update/documents";
+                
+                mqttClient.subscribe(topic, 1, (t, msg) -> {
+                    String payload = new String(msg.getPayload(), StandardCharsets.UTF_8);
 
-                String sensorId = reported.at("/sensorId").asText();
-                String type = reported.at("/type").asText();
-                String zoneId = reported.at("/zoneId").asText();
-                /* ---------- equipId / equipName 처리 ---------- */
-                String equipIdVal = reported.path("equipId").asText(null);   // 키가 없으면 null
-                String equipId    = (equipIdVal == null || equipIdVal.isBlank()) ? null : equipIdVal;
+                    // JSON 파싱 및 DB 저장은 이후 구현 예정
+                    try {
+                        ObjectMapper mapper = new ObjectMapper();
+                        JsonNode jsonNode = mapper.readTree(payload);
+                        // mqtt에서 전달되는 뎁스를 따라가야함
+                        JsonNode reported = jsonNode.at("/current/state/reported");
+                        log.info("📥 MQTT 수신 (topic: {}): {}", t, jsonNode);
 
-                Integer iszone = equipId.equals(zoneId) ? 1 : 0;
+                        String sensorId = reported.at("/sensorId").asText();
+                        String type = reported.at("/type").asText();
+                        String zoneId = reported.at("/zoneId").asText();
+                        /* ---------- equipId / equipName 처리 ---------- */
+                        String equipIdVal = reported.path("equipId").asText(null);   // 키가 없으면 null
+                        String equipId = (equipIdVal == null || equipIdVal.isBlank()) ? null : equipIdVal;
 
-                SensorCreateRequest dto = new SensorCreateRequest(sensorId, type , zoneId, equipId, null, null, iszone);
-                sensorService.saveSensor(dto); // 중복이면 예외 발생
-                log.info("✅ 센서 저장 완료: {}", sensorId);
-            } catch (DataIntegrityViolationException e) {
-                log.warn("⚠️ 중복 센서 저장 시도 차단됨: {}", e.getMessage());
-            } catch (Exception e) {
-                log.error("❌ JSON 파싱 또는 저장 중 오류: {}", e.getMessage());
-            }
+                        if (zoneId == null || zoneId.isBlank()) {
+                            log.error("❌ 유효하지 않은 zoneId: {}", zoneId);
+                            return;
+                        }
+
+                        Integer iszone = (equipId != null && equipId.equals(zoneId)) ? 1 : 0;
 
-        });
-        log.info("📡 MQTT subscribe 완료됨: topic = {}", topic); // ★ subscribe 후에도 로그
+                        SensorCreateRequest dto = new SensorCreateRequest(sensorId, type, zoneId, equipId, null, null, iszone);
+                        sensorService.saveSensor(dto); // 중복이면 예외 발생
+                        log.info("✅ 센서 저장 완료: {}", sensorId);
+                    } catch (DataIntegrityViolationException e) {
+                        log.warn("⚠️ 중복 센서 저장 시도 차단됨: {}", e.getMessage());
+                    } catch (Exception e) {
+                        log.error("❌ JSON 파싱 또는 저장 중 오류: {}", e.getMessage());
+                    }
+                });
+                
+                log.info("📡 MQTT subscribe 완료됨: topic = {}", topic);
+                return; // 성공적으로 구독했으면 메서드 종료
+                
+            } catch (MqttException e) {
+                log.error("MQTT 연결/구독 실패 (시도 {}/{}): {}", 
+                        retryCount + 1, MAX_RETRY_ATTEMPTS, e.getMessage());
+                retryCount++;
+                if (retryCount >= MAX_RETRY_ATTEMPTS) {
+                    throw new MqttException(e.getReasonCode());
+                }
+                try {
+                    Thread.sleep(RETRY_DELAY_MS);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new MqttException(e.getReasonCode());
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+            }
+        }
     }
 }
diff --git a/src/main/java/com/factoreal/backend/messaging/sender/WebSocketSender.java b/src/main/java/com/factoreal/backend/messaging/sender/WebSocketSender.java
index 8d5a4111..1bf6e4af 100644
--- a/src/main/java/com/factoreal/backend/messaging/sender/WebSocketSender.java
+++ b/src/main/java/com/factoreal/backend/messaging/sender/WebSocketSender.java
@@ -1,9 +1,14 @@
 package com.factoreal.backend.messaging.sender;
 
+import com.factoreal.backend.domain.controlLog.entity.ControlLog;
 import com.factoreal.backend.messaging.common.dto.SystemLogDto;
 import com.factoreal.backend.messaging.common.dto.ZoneDangerDto;
 import com.factoreal.backend.messaging.kafka.strategy.enums.AlarmEventDto;
 import lombok.RequiredArgsConstructor;
+
+import java.util.HashMap;
+import java.util.Map;
+
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Component;
 
@@ -38,7 +43,31 @@ public void sendDangerAlarm(AlarmEventDto alarmEventDto) {
     /**
      * 읽지 않은 알람수 전송
      */
-    public void sendUnreadCount(long count){
+    public void sendUnreadCount(long count) {
         messagingTemplate.convertAndSend("/topic/unread-count", count);
     }
+
+    /**
+     * 제어 상태를 WebSocket으로 전송하고 FE에서 발송 여부를 확인할 수 있도록 함
+     */
+    public void sendControlStatus(ControlLog controlLog, Map<String, Boolean> deliveryStatus) {
+        Map<String, Object> status = new HashMap<>();
+        status.put("controlId", controlLog.getId());
+        status.put("controlType", controlLog.getControlType());
+        status.put("controlValue", controlLog.getControlVal());
+        status.put("controlStatus", controlLog.getControlStat());
+        status.put("executedAt", controlLog.getExecutedAt().toString());
+        status.put("zoneId", controlLog.getZone().getZoneId());
+
+        // AbnormalLog 정보 추가
+        status.put("abnormalId", controlLog.getAbnormalLog().getId());
+        status.put("abnormalType", controlLog.getAbnormalLog().getAbnormalType());
+        status.put("targetType", controlLog.getAbnormalLog().getTargetType().name());
+        status.put("targetId", controlLog.getAbnormalLog().getTargetId());
+
+        // 발송 상태 추가
+        status.putAll(deliveryStatus);
+
+        messagingTemplate.convertAndSend("/topic/control-status", status);
+    }
 }
diff --git a/src/main/java/com/factoreal/backend/messaging/service/AutoControlService.java b/src/main/java/com/factoreal/backend/messaging/service/AutoControlService.java
index 344c792d..ff8efcce 100644
--- a/src/main/java/com/factoreal/backend/messaging/service/AutoControlService.java
+++ b/src/main/java/com/factoreal/backend/messaging/service/AutoControlService.java
@@ -2,6 +2,12 @@
 
 import com.factoreal.backend.domain.sensor.dto.SensorKafkaDto;
 import com.factoreal.backend.domain.sensor.entity.Sensor;
+
+import jakarta.transaction.Transactional;
+
+import com.factoreal.backend.domain.abnormalLog.application.AbnormalLogService;
+import com.factoreal.backend.domain.abnormalLog.entity.AbnormalLog;
+import com.factoreal.backend.domain.controlLog.service.ControlLogService;
 import com.factoreal.backend.domain.sensor.dao.SensorRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -17,12 +23,16 @@
 public class AutoControlService {
 
     private final SensorRepository sensorRepository;
+    private final ControlLogService controlLogService;
+    private final AbnormalLogService abnormalLogService;
 
     /**
      * 센서 값이 허용 범위를 벗어났을 경우 제어 메시지를 생성하거나 처리하도록 로깅
      */
+    @Transactional
     public void evaluate(SensorKafkaDto dto, int dangerLevel) {
-        if (dangerLevel == 0) return; // 정상 범위면 아무 처리 안 함
+        if (dangerLevel == 0)
+            return; // 정상 범위면 아무 처리 안 함
 
         Sensor sensor = sensorRepository.findById(dto.getSensorId())
                 .orElse(null);
@@ -36,16 +46,30 @@ public void evaluate(SensorKafkaDto dto, int dangerLevel) {
         double tolerance = sensor.getAllowVal() != null ? sensor.getAllowVal() : 0.0;
         double value = dto.getVal();
 
+        // 센서 값이 허용 범위를 벗어났을 경우
         if (value < threshold - tolerance || value > threshold + tolerance) {
             String message = buildControlMessage(sensor.getSensorType().name(), value, threshold, tolerance);
             log.info("⚙️ 자동제어 필요: {}", message);
+
+            // 1. 이상 로그 저장
+            AbnormalLog abnormalLog = abnormalLogService.saveAbnormalLog(dto, sensor, dangerLevel);
+
+            // 2. 제어 로그 저장
+            String controlType = getControlType(sensor.getSensorType().name());
+
+            controlLogService.saveControlLog(
+                    abnormalLog,
+                    controlType,
+                    threshold, // controlVal: 임계값을 목표값으로 사용
+                    1, // controlStat: 성공 상태로 설정
+                    sensor.getZone());
+
             // TODO: MQTT 퍼블리시 로직으로 대체
         } else {
             log.info("✅ 측정값은 허용 범위 내: sensorId={}, value={}", dto.getSensorId(), value);
         }
     }
 
-
     // 제어 로직
     private String buildControlMessage(String type, double val, double thresh, double tol) {
         return switch (type.toLowerCase()) {
@@ -57,4 +81,14 @@ private String buildControlMessage(String type, double val, double thresh, doubl
             default -> String.format("현재 값 %.1f, 허용 범위: %.1f~%.1f", val, thresh - tol, thresh + tol);
         };
     }
+
+    // 센서 타입에 따른 제어 타입 결정
+    private String getControlType(String sensorType) {
+        return switch (sensorType.toLowerCase()) {
+            case "temp" -> "에어컨";
+            case "humid" -> "제습기";
+            case "dust" -> "공기청정기";
+            default -> sensorType;
+        };
+    }
 }
diff --git a/src/test/java/com/factoreal/backend/service/WorkerManagerServiceTest.java b/src/test/java/com/factoreal/backend/service/WorkerManagerServiceTest.java
new file mode 100644
index 00000000..ec151eba
--- /dev/null
+++ b/src/test/java/com/factoreal/backend/service/WorkerManagerServiceTest.java
@@ -0,0 +1,213 @@
+package com.factoreal.backend.service;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import com.factoreal.backend.domain.worker.application.WorkerManagerService;
+import com.factoreal.backend.domain.worker.dao.WorkerZoneRepository;
+import com.factoreal.backend.domain.worker.dto.response.WorkerManagerResponse;
+import com.factoreal.backend.domain.worker.entity.Worker;
+import com.factoreal.backend.domain.worker.entity.WorkerZone;
+import com.factoreal.backend.domain.worker.entity.WorkerZoneId;
+import com.factoreal.backend.domain.zone.entity.Zone;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+class WorkerManagerServiceTest {
+
+    @Mock
+    private WorkerZoneRepository workerZoneRepository;
+
+    @InjectMocks
+    private WorkerManagerService workerManagerService;
+
+    private Worker worker1, worker2, worker3, worker4;
+    private Zone zone1, zone2;
+    private WorkerZone workerZone1, workerZone2, workerZone3, workerZone4, workerZone5;
+
+    @BeforeEach
+    void setUp() {
+        MockitoAnnotations.openMocks(this);
+
+        // 테스트용 Zone 데이터 생성
+        zone1 = Zone.builder()
+                .zoneId("zone1")
+                .zoneName("공간1")
+                .build();
+
+        zone2 = Zone.builder()
+                .zoneId("zone2")
+                .zoneName("공간2")
+                .build();
+
+        // 테스트용 Worker 데이터 생성
+        worker1 = Worker.builder()
+                .workerId("worker1")
+                .name("작업자1")
+                .phoneNumber("010-1111-1111")
+                .email("worker1@test.com")
+                .build();
+
+        worker2 = Worker.builder()
+                .workerId("worker2")
+                .name("작업자2")
+                .phoneNumber("010-2222-2222")
+                .email("worker2@test.com")
+                .build();
+
+        worker3 = Worker.builder()
+                .workerId("worker3")
+                .name("작업자3")
+                .phoneNumber("010-3333-3333")
+                .email("worker3@test.com")
+                .build();
+
+        worker4 = Worker.builder()
+                .workerId("worker4")
+                .name("작업자4")
+                .phoneNumber("010-4444-4444")
+                .email("worker4@test.com")
+                .build();
+
+        // 테스트용 WorkerZone 데이터 생성
+        workerZone1 = WorkerZone.builder()
+                .id(new WorkerZoneId("worker1", "zone1"))
+                .worker(worker1)
+                .zone(zone1)
+                .manageYn(true) // worker1은 zone1의 담당자
+                .build();
+
+        workerZone2 = WorkerZone.builder()
+                .id(new WorkerZoneId("worker2", "zone1"))
+                .worker(worker2)
+                .zone(zone1)
+                .manageYn(false) // worker2는 zone1에 접근 가능
+                .build();
+
+        workerZone3 = WorkerZone.builder()
+                .id(new WorkerZoneId("worker2", "zone2"))
+                .worker(worker2)
+                .zone(zone2)
+                .manageYn(true) // worker2는 zone2의 담당자
+                .build();
+
+        workerZone4 = WorkerZone.builder()
+                .id(new WorkerZoneId("worker3", "zone1"))
+                .worker(worker3)
+                .zone(zone1)
+                .manageYn(false) // worker3는 zone1에 접근 가능
+                .build();
+
+        workerZone5 = WorkerZone.builder()
+                .id(new WorkerZoneId("worker4", "zone1"))
+                .worker(worker4)
+                .zone(zone1)
+                .manageYn(false) // worker4는 zone1에 접근 가능
+                .build();
+    }
+
+    // 특정 공간의 담당자 후보 목록 조회
+    @Test
+    void getManagerCandidates_WhenZoneHasManager() {
+        // given
+        String zoneId = "zone1";
+
+        // 현재 담당자 설정
+        when(workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId))
+                .thenReturn(Optional.of(workerZone1));
+
+        // 다른 공간의 담당자 설정
+        when(workerZoneRepository.findByZoneZoneIdNotAndManageYnIsTrue(zoneId))
+                .thenReturn(Arrays.asList(workerZone3)); // worker2는 zone2의 담당자
+
+        // 해당 공간의 모든 작업자 설정
+        when(workerZoneRepository.findByZoneZoneId(zoneId))
+                .thenReturn(Arrays.asList(workerZone1, workerZone2, workerZone4, workerZone5));
+
+        // when
+        List<WorkerManagerResponse> candidates = workerManagerService.getManagerCandidates(zoneId);
+
+        // then
+        assertNotNull(candidates);
+        assertEquals(2, candidates.size()); // worker3, worker4만 후보가 되어야 함 (현재 담당자와 다른 공간 담당자 제외)
+
+        // worker2(다른 공간 담당자)가 후보 목록에 없는지 확인
+        assertTrue(candidates.stream()
+                .noneMatch(c -> c.getWorkerId().equals("worker2")));
+
+        // worker1(현재 담당자)가 후보 목록에 없는지 확인
+        assertTrue(candidates.stream()
+                .noneMatch(c -> c.getWorkerId().equals("worker1")));
+    }
+
+    // 공간 담당자 지정
+    @Test
+    void assignManager_Success() {
+        // given
+        String zoneId = "zone1";
+        String newManagerId = "worker3";
+
+        // 현재 담당자 설정
+        when(workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId))
+                .thenReturn(Optional.of(workerZone1));
+
+        // 새로운 담당자의 WorkerZone 설정
+        WorkerZoneId newManagerZoneId = new WorkerZoneId(newManagerId, zoneId);
+        when(workerZoneRepository.findById(newManagerZoneId))
+                .thenReturn(Optional.of(workerZone4));
+
+        // when
+        workerManagerService.assignManager(zoneId, newManagerId);
+
+        // then
+        // 기존 담당자의 manageYn이 false로 변경되었는지 확인
+        verify(workerZoneRepository, times(1))
+                .save(argThat(wz -> wz.getWorker().getWorkerId().equals("worker1") && !wz.getManageYn()));
+
+        // 새로운 담당자의 manageYn이 true로 변경되었는지 확인
+        verify(workerZoneRepository, times(1))
+                .save(argThat(wz -> wz.getWorker().getWorkerId().equals(newManagerId) && wz.getManageYn()));
+    }
+
+    // 현재 공간 담당자 조회 (담당자가 있는 경우)
+    @Test
+    void getCurrentManager_WhenManagerExists() {
+        // given
+        String zoneId = "zone1";
+        when(workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId))
+                .thenReturn(Optional.of(workerZone1));
+
+        // when
+        WorkerManagerResponse manager = workerManagerService.getCurrentManager(zoneId);
+
+        // then
+        assertNotNull(manager);
+        assertEquals("worker1", manager.getWorkerId());
+        assertEquals("작업자1", manager.getName());
+        assertTrue(manager.getIsManager());
+    }
+
+    // 현재 공간 담당자 조회 (담당자가 없는 경우)
+    @Test
+    void getCurrentManager_WhenNoManagerExists() {
+        // given
+        String zoneId = "zone1";
+        when(workerZoneRepository.findByZoneZoneIdAndManageYnIsTrue(zoneId))
+                .thenReturn(Optional.empty());
+
+        // when
+        WorkerManagerResponse manager = workerManagerService.getCurrentManager(zoneId);
+
+        // then
+        assertNull(manager);
+    }
+}
diff --git a/src/test/java/com/factoreal/backend/service/WorkerServiceTest.java b/src/test/java/com/factoreal/backend/service/WorkerServiceTest.java
index 0b0fcca3..8a9004c6 100644
--- a/src/test/java/com/factoreal/backend/service/WorkerServiceTest.java
+++ b/src/test/java/com/factoreal/backend/service/WorkerServiceTest.java
@@ -9,6 +9,8 @@
 import com.factoreal.backend.domain.zone.entity.Zone;
 import com.factoreal.backend.domain.zone.entity.ZoneHist;
 import com.factoreal.backend.domain.worker.dao.WorkerRepository;
+import com.factoreal.backend.domain.worker.dao.WorkerZoneRepository;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.InjectMocks;
@@ -18,13 +20,19 @@
 import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.when;
 
 public class WorkerServiceTest {
 
+
+        @Mock
+        private WorkerZoneRepository workerZoneRepository;
+
     @Mock
     private WorkerRepository workerRepository;
 
@@ -36,88 +44,93 @@ public class WorkerServiceTest {
     @InjectMocks
     private WorkerService workerService;
 
-    private Worker worker1;
-    private Worker worker2;
-    private Zone zone1;
-    private ZoneHist zoneHist1;
-
-    @BeforeEach
-    public void setup() {
-        MockitoAnnotations.openMocks(this);
-        
-        // 테스트용 작업자 데이터 생성
-        worker1 = Worker.builder()
-                .workerId("20240101-1234")
-                .name("홍길동")
-                .phoneNumber("01012345678")
-                .email("hong@example.com")
-                .build();
-        
-        worker2 = Worker.builder()
-                .workerId("20240102-5678")
-                .name("김철수")
-                .phoneNumber("01087654321")
-                .email("kim@example.com")
-                .build();
-                
-        // 테스트용 공간 데이터 생성
-        zone1 = Zone.builder()
-                .zoneId("zone1")
-                .zoneName("테스트 공간")
-                .build();
-                
-        // 테스트용 ZoneHist 데이터 생성
-        zoneHist1 = ZoneHist.builder()
-                .id(1L)
-                .worker(worker1)
-                .zone(zone1)
-                .startTime(LocalDateTime.now())
-                .endTime(null)
-                .existFlag(1)
-                .build();
-    }
-
-    @Test
-    public void testGetAllWorkers() {
-        // Mock 설정
-        when(workerRepository.findAll()).thenReturn(Arrays.asList(worker1, worker2));
-        
-        // 서비스 메소드 호출
-        List<WorkerDetailResponse> result = workerService.getAllWorkers();
-        
-        // 결과 검증
-        assertNotNull(result);
-        assertEquals(2, result.size());
-        
-        // 첫 번째 작업자 정보 확인
-        assertEquals("20240101-1234", result.get(0).getWorkerId());
-        assertEquals("홍길동", result.get(0).getName());
-        assertEquals("01012345678", result.get(0).getPhoneNumber());
-        assertEquals("hong@example.com", result.get(0).getEmail());
-        assertEquals(false, result.get(0).getIsManager());  // 기본값은 false
-        
-        // 두 번째 작업자 정보 확인
-        assertEquals("20240102-5678", result.get(1).getWorkerId());
-        assertEquals("김철수", result.get(1).getName());
-        assertEquals("01087654321", result.get(1).getPhoneNumber());
-        assertEquals("kim@example.com", result.get(1).getEmail());
-        assertEquals(false, result.get(1).getIsManager());  // 기본값은 false
-    }
-
-    @Test
-    void testGetWorkersByZoneId() {
-        // Mock 설정
-        when(zoneHistoryRepository.findByZone_ZoneIdAndExistFlag("zone1", 1)).thenReturn(Arrays.asList(zoneHist1));
-        
-        // 테스트 실행
-        List<WorkerInfoResponse> workers = workerService.getWorkersByZoneId("zone1");
-        
-        // 검증
-        assertEquals(1, workers.size());
-        assertEquals("20240101-1234", workers.get(0).getWorkerId());
-        assertEquals("홍길동", workers.get(0).getName());
-        assertEquals("01012345678", workers.get(0).getPhoneNumber());
-        assertEquals("hong@example.com", workers.get(0).getEmail());
-        assertEquals(false, workers.get(0).getIsManager());  // 관리자 여부는 더 이상 ZoneHist에서 확인할 수 없음
-    }
+        private Worker worker1;
+        private Worker worker2;
+        private Zone zone1;
+        private ZoneHist zoneHist1;
+
+        @BeforeEach
+        public void setup() {
+                MockitoAnnotations.openMocks(this);
+
+                // 테스트용 작업자 데이터 생성
+                worker1 = Worker.builder()
+                                .workerId("20240101-1234")
+                                .name("홍길동")
+                                .phoneNumber("01012345678")
+                                .email("hong@example.com")
+                                .build();
+
+                worker2 = Worker.builder()
+                                .workerId("20240102-5678")
+                                .name("김철수")
+                                .phoneNumber("01087654321")
+                                .email("kim@example.com")
+                                .build();
+
+                // 테스트용 공간 데이터 생성
+                zone1 = Zone.builder()
+                                .zoneId("zone1")
+                                .zoneName("테스트 공간")
+                                .build();
+
+                // 테스트용 ZoneHist 데이터 생성
+                zoneHist1 = ZoneHist.builder()
+                                .id(1L)
+                                .worker(worker1)
+                                .zone(zone1)
+                                .startTime(LocalDateTime.now())
+                                .endTime(null)
+                                .existFlag(1)
+                                .build();
+        }
+
+        @Test
+        public void testGetAllWorkers() {
+                // Mock 설정
+                when(workerRepository.findAll()).thenReturn(Arrays.asList(worker1, worker2));
+
+                // WorkerZoneRepository mock 설정 추가
+                when(workerZoneRepository.findByWorkerWorkerIdAndManageYnIsTrue(anyString()))
+                                .thenReturn(Optional.empty()); // 모든 작업자가 관리자가 아닌 것으로 설정
+
+                // 서비스 메소드 호출
+                List<WorkerDetailResponse> result = workerService.getAllWorkers();
+
+                // 결과 검증
+                assertNotNull(result);
+                assertEquals(2, result.size());
+
+                // 첫 번째 작업자 정보 확인
+                assertEquals("20240101-1234", result.get(0).getWorkerId());
+                assertEquals("홍길동", result.get(0).getName());
+                assertEquals("01012345678", result.get(0).getPhoneNumber());
+                assertEquals("hong@example.com", result.get(0).getEmail());
+                assertEquals(false, result.get(0).getIsManager()); // 기본값은 false
+
+                // 두 번째 작업자 정보 확인
+                assertEquals("20240102-5678", result.get(1).getWorkerId());
+                assertEquals("김철수", result.get(1).getName());
+                assertEquals("01087654321", result.get(1).getPhoneNumber());
+                assertEquals("kim@example.com", result.get(1).getEmail());
+                assertEquals(false, result.get(1).getIsManager()); // 기본값은 false
+        }
+
+        @Test
+        void testGetWorkersByZoneId() {
+                // Mock 설정
+                when(zoneHistoryRepository.findByZone_ZoneIdAndExistFlag("zone1", 1))
+                                .thenReturn(Arrays.asList(zoneHist1));
+
+                // 테스트 실행
+                List<WorkerInfoResponse> workers = workerService.getWorkersByZoneId("zone1");
+
+                // 검증
+                assertEquals(1, workers.size());
+                assertEquals("20240101-1234", workers.get(0).getWorkerId());
+                assertEquals("홍길동", workers.get(0).getName());
+                assertEquals("01012345678", workers.get(0).getPhoneNumber());
+                assertEquals("hong@example.com", workers.get(0).getEmail());
+                assertEquals(false, workers.get(0).getIsManager()); // 관리자 여부는 더 이상 ZoneHist에서 확인할 수 없음
+        }
 }