← Docs hub

MQTT Task Contract

이 문서는 IoT/MQTT에서 DeviceAgent TaskManager를 호출할 때 사용할 권장 메시지 계약이다. 목표는 MQTT, System App AAR, PUI, On-device가 서로 다른 표현을 쓰지 않고 같은 task 의미 체계를 공유하는 것이다.

한 문장 결론: broker, topic, 인증, QoS와 기존 단일 명령 payload를 일괄 교체하지 않는다. TaskManager 관리가 필요한 요청에만 request.task를 추가하고, 복합 실행은 별도의 공통 submitWorkflow ingress로 확장한다.

핵심 원칙:

MQTT가 task 의도를 명시하면 그 값을 우선 반영한다.
MQTT가 task 의도를 생략하면 기존 DeviceAgent direct command로 유지한다.
request.task가 있으면 TaskManager intent로 본다.
조회, heartbeat, sensor/status update는 기본적으로 task queue에 넣지 않는다.

관련 문서:

• TaskManagerClient AAR API Specification
• TaskManager 이해 가이드: 직접 호출에서 관리형 실행까지
• DeviceAgent TaskManager 기준 사양서

1. 전체 흐름

Cloud / IoT Platform
  -> MQTT topic/message
  -> IotAgent topic parser
  -> legacy DeviceAgent Bundle 생성
  -> request.task metadata를 Bundle top-level field로 복사
  -> IotTaskManagerBridge.wrapCommand()
  -> DeviceControl.sendModuleCommand()
  -> DeviceAgent.MainApi.executeMethod()
  -> direct legacy execution 또는 TaskManager.handleControlMethod()
  -> submitTask / submitWorkflow

현재 구현 기준:

IotAgent.setCmdTopic()
  -> IotTaskManagerBridge.setCurrentCommandJson(jsonObject)
  -> topic별 기존 parser 수행
  -> IotTaskManagerBridge.wrapCommand(...)
     - request.task를 읽어 Bundle field로 복사
     - request.task가 없으면 기존 method direct 호출 유지
     - request.task가 있으면 method=submitTask, taskMethod=기존 method, source=iot로 변환
     - managed=false면 direct 유지

구현 근거:

이 방식은 기존 topic별 parser를 대량 수정하지 않기 위한 선택이다. CmdTopic 안에는 이미 많은 IotTaskManagerBridge.wrapCommand(...) 호출이 있으므로, MQTT 원문 JSON을 command 처리 scope 동안 ThreadLocal로 잡아두고 공통 wrapper가 request.task를 읽게 했다.

장점:
- 기존 setClean/comeback/moveTo/setSchedule parser를 대부분 유지한다.
- 모든 wrapCommand 호출이 같은 request.task 처리 규칙을 탄다.
- task metadata가 없는 기존 MQTT 메시지는 기존 direct command 동작을 유지한다.

주의:
- setCmdTopic 처리 thread 안에서만 현재 command JSON이 유효하다.
- finally에서 clear하지 않으면 다음 command에 metadata가 섞일 수 있으므로 clear는 필수다.

2. MQTT Envelope

모든 command topic은 아래 공통 envelope를 권장한다.

{
  "replyTo": "skmg/airbot/{serial}/{version}/response/{responseTopic}",
  "correlationId": "mqtt-request-001",
  "timestamp": "2026-07-15T10:00:00+09:00",
  "request": {
    "method": "setMoveTo",
    "payload": {},
    "task": {}
  }
}

3. Task Metadata

request.task는 optional이다. 없으면 기존 MQTT command 경로를 깨지 않기 위해 TaskManager에 등록하지 않고 direct legacy command로 보낸다. 값이 있으면 TaskManager intent로 해석하고, managed=false가 아닌 한 method=submitTask, taskMethod=<기존 method>, source=iot envelope로 변환한다.

즉 기본값은 아래처럼 나뉜다.

request.task를 생략한 기존 명령은 관리 Task와 같은 대기열에 들어가지 않는다. 두 경로 모두 기존 DeviceAgent 도메인 정책을 통과하므로 이동·청정 중복과 제품 상태 제한은 적용되지만, TaskManager가 direct 명령의 순서나 선점 결과를 기록하는 것은 아니다.

따라서 관리 Workflow와 경쟁할 수 있는 이동·청정·보안·외부 앱 세션 명령은 검증된 범위부터 request.task를 붙이는 것이 권장된다. 기존 direct 호출을 계속 허용해야 하는 기간에는 공통 실행권 검사와 PUI·안전 중단 결과의 Task terminal 연결이 별도로 필요하다. 상세 기준은 기존 direct 명령과 관리 Task의 동시성을 따른다.

{
  "managed": true,
  "executionMode": "async",
  "priority": "HIGH",
  "queueKey": "movement",
  "timeoutMs": 240000,
  "completionTarget": "movement.arrived",
  "preemptPolicy": "append",
  "cancellable": true
}

3.1 구현된 복사 규칙

현재 IotTaskManagerBridge는 request.task에서 아래 타입별 field를 읽어 legacy Bundle top-level로 복사한다.

String field

executionMode
priority
queueKey
completionTarget
completionAreaId
preemptPolicy
failurePolicy
cancelMethod
compensationMethod

예:

{
  "request": {
    "method": "setMoveTo",
    "task": {
      "executionMode": "async",
      "priority": "HIGH",
      "queueKey": "movement",
      "completionTarget": "movement.arrived",
      "preemptPolicy": "append"
    }
  }
}

Bundle 반영:

executionMode=async
priority=HIGH
queueKey=movement
completionTarget=movement.arrived
preemptPolicy=append

Long field

timeoutMs

예:

{
  "request": {
    "method": "setMoveTo",
    "task": {
      "timeoutMs": 240000
    }
  }
}

Bundle 반영:

timeoutMs=240000

Boolean field

managed
cancellable
stateAware
strictValidation
dryRun

예:

{
  "request": {
    "method": "setMoveTo",
    "task": {
      "managed": true,
      "cancellable": true,
      "strictValidation": false
    }
  }
}

Bundle 반영:

managed=true
cancellable=true
strictValidation=false

3.2 managed 판단 순서

managed는 TaskManager wrapping 여부를 명시하는 field다.

request.task 없음
  -> 기존 MQTT direct command로 유지한다.

request.task 있음 + managed 생략
  -> request.task 자체를 명시적 task intent로 보고 managed=true처럼 처리한다.

managed=true
  -> method가 get/is/check/request가 아니어도 명시적으로 task 등록을 시도한다.

managed=false
  -> setMoveTo처럼 task 대상처럼 보이는 method도 direct legacy command로 유지한다.

이렇게 정한 이유:

1. 기존 MQTT payload를 바꾸지 않은 상태에서 갑자기 queue/task lifecycle에 올라가는 것을 막는다.
2. TaskManager를 쓰려는 요청은 request.task 존재만으로 의도가 드러난다.
3. AAR/System App도 submitTask를 명시 호출하므로 MQTT도 명시 intent 기준이 더 일관적이다.
4. setAirQuality, setEyeLedColor 같은 상시 신호가 monitor timeline을 오염시키는 것을 줄인다.

권장 판단표:

예시:

{
  "correlationId": "mqtt-direct-move-001",
  "request": {
    "method": "setMoveTo",
    "payload": {
      "positionId": "4"
    },
    "task": {
      "managed": false,
      "executionMode": "direct"
    }
  }
}

결과:

method=setMoveTo
positionId=4
executionMode=direct
traceId=mqtt-direct-move-001
correlationId=mqtt-direct-move-001

이 경우 method=submitTask, taskMethod=setMoveTo, source=iot는 붙지 않는다.

4. Execution Mode 의미

권장:

장기 동작형: async
짧은 설정형: queued_wait
조회/상태 확인: direct 또는 task wrapping 제외

5. 단일 이동 예시

MQTT:

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/moveToResponse",
  "correlationId": "mqtt-move-001",
  "timestamp": "2026-07-15T10:00:00+09:00",
  "request": {
    "method": "setMoveTo",
    "payload": {
      "positionId": "4"
    },
    "task": {
      "managed": true,
      "executionMode": "async",
      "priority": "HIGH",
      "queueKey": "movement",
      "timeoutMs": 240000,
      "completionTarget": "movement.arrived",
      "preemptPolicy": "append"
    }
  }
}

DeviceAgent로 들어가는 Bundle:

method=submitTask
taskMethod=setMoveTo
source=iot
traceId=mqtt-move-001
correlationId=mqtt-move-001
replyTo=skmg/airbot/A1-board-005/v1/response/moveToResponse
positionId=4
executionMode=async
priority=HIGH
queueKey=movement
timeoutMs=240000
completionTarget=movement.arrived
preemptPolicy=append

처리 결과:

1. CmdTopic의 moveTo topic parser가 기존 방식대로 setMoveTo용 Bundle을 만든다.
2. IotTaskManagerBridge가 request.task를 읽어 executionMode/priority/queueKey/timeoutMs/completionTarget/preemptPolicy를 복사한다.
3. managed=true이므로 method를 submitTask로 바꾸고 기존 method를 taskMethod=setMoveTo로 보존한다.
4. DeviceAgent TaskManager는 explicit metadata를 우선 반영하고 누락값은 TaskPolicyRegistry 기본값으로 채운다.

6. 스테이션 복귀 예시

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/comebackResponse",
  "correlationId": "mqtt-return-001",
  "timestamp": "2026-07-15T10:05:00+09:00",
  "request": {
    "method": "returnToStation",
    "payload": {
      "missionFinYn": "Y"
    },
    "task": {
      "managed": true,
      "executionMode": "async",
      "priority": "HIGH",
      "queueKey": "movement",
      "timeoutMs": 240000,
      "completionTarget": "movement.docked",
      "preemptPolicy": "replace_queue"
    }
  }
}

replace_queue는 이동 queue의 기존 task를 취소할 수 있으므로 IoT source에서 허용할 method를 제한해야 한다.

Bundle 핵심:

method=submitTask
taskMethod=returnToStation
source=iot
missionFinYn=Y
executionMode=async
priority=HIGH
queueKey=movement
timeoutMs=240000
completionTarget=movement.docked
preemptPolicy=replace_queue

권장 운영:

복귀 명령은 사용자가 현재 이동/청정 흐름을 중단하려는 의미일 수 있으므로 replace_queue를 허용할 수 있다.
다만 모든 IoT 명령에 replace_queue를 허용하면 기존 workflow를 예기치 않게 끊을 수 있다.

7. 청정 시작 예시

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/setCleanResponse",
  "correlationId": "mqtt-clean-001",
  "timestamp": "2026-07-15T10:10:00+09:00",
  "request": {
    "method": "setAirCleanerOperation",
    "payload": {
      "module": "app",
      "action": "1",
      "mode": "1",
      "speed": "0",
      "position": ["4"],
      "cur_destination": "4"
    },
    "task": {
      "managed": true,
      "executionMode": "async",
      "priority": "CONTROL",
      "queueKey": "cleaning",
      "timeoutMs": 1200000,
      "completionTarget": "cleaning.stopped",
      "completionAreaId": "4"
    }
  }
}

Bundle 핵심:

method=submitTask
taskMethod=setAirCleanerOperation
source=iot
module=app
action=1
mode=1
speed=0
position=[4]
cur_destination=4
executionMode=async
priority=CONTROL
queueKey=cleaning
timeoutMs=1200000
completionTarget=cleaning.stopped
completionAreaId=4

설명:

청정 동작은 command return만으로 완료 처리하면 안 된다.
completionTarget=cleaning.stopped 또는 domain callback/post-check를 통해 실제 완료 시점을 판단해야 한다.

8. Direct / Task 제외 예시

상시 상태 update나 조회성 명령은 task queue에 넣지 않는 것이 기본이다.

{
  "correlationId": "mqtt-status-001",
  "request": {
    "method": "setAirQuality",
    "payload": {
      "pm10": "12",
      "pm25": "5"
    },
    "task": {
      "managed": false
    }
  }
}

결과:

method=setAirQuality
traceId=mqtt-status-001
correlationId=mqtt-status-001
pm10=12
pm25=5

setAirQuality, heartbeat, healthCheck, AWS connection state, LED 상시 상태처럼 monitor 노이즈가 큰 값은 기본적으로 task lifecycle에 올리지 않는다.

8.1 request.task 생략 예시

기존 MQTT payload가 request.task 없이 내려오면 TaskManager에 등록하지 않고 기존 direct command로 동작한다.

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/moveToResponse",
  "correlationId": "mqtt-move-default-001",
  "timestamp": "2026-07-15T10:20:00+09:00",
  "request": {
    "method": "setMoveTo",
    "payload": {
      "positionId": "4"
    }
  }
}

Bundle 핵심:

method=setMoveTo
positionId=4
traceId=mqtt-move-default-001
correlationId=mqtt-move-default-001

이 경우 TaskManager task id, queue event, completion event는 생성되지 않는다. 기존 MQTT command와 동일하게 DeviceAgent method가 직접 실행된다.

8.2 빈 request.task로 TaskManager 기본 정책 사용

TaskManager에는 올리고 싶지만 세부 정책은 DeviceAgent 기본값을 쓰고 싶으면 빈 task object를 보낸다.

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/moveToResponse",
  "correlationId": "mqtt-move-policy-default-001",
  "timestamp": "2026-07-15T10:21:00+09:00",
  "request": {
    "method": "setMoveTo",
    "payload": {
      "positionId": "4"
    },
    "task": {}
  }
}

Bundle 핵심:

method=submitTask
taskMethod=setMoveTo
source=iot
positionId=4
managed=true
traceId=mqtt-move-policy-default-001
correlationId=mqtt-move-policy-default-001

TaskManager 기본 정책 예:

setMoveTo -> executionMode=ASYNC, queue=movement, priority=HIGH, timeoutMs=120000, cancellable=true

즉, 외부가 모든 정책을 매번 내려줄 필요는 없다. 다만 TaskManager를 쓰겠다는 의도는 request.task object로 명시해야 한다.

8.3 Strict Validation 예시

schema 검증을 강제로 켜고 싶으면 strictValidation=true를 내려준다.

{
  "correlationId": "mqtt-strict-001",
  "request": {
    "method": "setMoveTo",
    "payload": {},
    "task": {
      "managed": true,
      "strictValidation": true
    }
  }
}

의도:

setMoveTo는 positionId/positionName/position/positionIds 중 하나가 필요하다.
strictValidation=true이면 누락 시 TaskManager가 validation_failed로 reject해야 한다.

9. 복합 Workflow 예시

외부에서 workflow를 직접 내릴 때는 request.method=submitWorkflow를 사용한다. payload.subTasks[] 안의 각 항목은 AAR TaskWorkflowRequest의 subtask와 같은 의미다.

{
  "replyTo": "skmg/airbot/A1-board-005/v1/response/workflowResponse",
  "correlationId": "mqtt-workflow-001",
  "timestamp": "2026-07-15T10:15:00+09:00",
  "request": {
    "method": "submitWorkflow",
    "payload": {
      "workflowName": "room4_room2_clean_return",
      "subTasks": [
        {
          "taskMethod": "setMoveTo",
          "positionId": "4",
          "completionTarget": "movement.arrived",
          "timeoutMs": 240000
        },
        {
          "taskMethod": "setAirCleanerOperation",
          "module": "app",
          "action": "1",
          "mode": "1",
          "speed": "0",
          "position": ["4"],
          "cur_destination": "4",
          "completionTarget": "cleaning.stopped",
          "completionAreaId": "4",
          "timeoutMs": 1200000
        },
        {
          "taskMethod": "setMoveTo",
          "positionId": "2",
          "completionTarget": "movement.arrived",
          "timeoutMs": 240000
        },
        {
          "taskMethod": "setAirCleanerOperation",
          "module": "app",
          "action": "1",
          "mode": "1",
          "speed": "0",
          "position": ["2"],
          "cur_destination": "2",
          "completionTarget": "cleaning.stopped",
          "completionAreaId": "2",
          "timeoutMs": 1200000
        },
        {
          "taskMethod": "returnToStation",
          "completionTarget": "movement.docked",
          "timeoutMs": 240000
        }
      ]
    },
    "task": {
      "managed": true,
      "executionMode": "async",
      "priority": "HIGH",
      "queueKey": "workflow",
      "preemptPolicy": "append",
      "failurePolicy": "fail_parent"
    }
  }
}

주의:

현재 단일 command metadata 파싱은 구현되어 있다.
MQTT JSON의 submitWorkflow payload.subTasks를 ArrayList<Bundle>로 변환하는 topic handler는 별도 구현 범위다.

구현 계획:

1. request.method == submitWorkflow 감지
2. request.payload.workflowName을 Bundle.workflowName으로 복사
3. request.payload.subTasks[]를 ArrayList<Bundle>로 변환
4. request.task metadata를 workflow parent Bundle에 복사
5. method=submitWorkflow, source=iot, traceId/correlationId/replyTo 주입
6. DeviceControl.sendModuleCommand(indata, outdata) 호출

workflow parser가 구현되기 전에는 기존 topic별 단일 command 경로에 request.task metadata를 붙이는 방식만 보장한다.

10. AAR 계약과의 동일성

MQTT request.task는 AAR builder의 아래 호출과 같은 의미를 가져야 한다.

TaskSubmitRequest.builder(TaskMethods.SET_MOVE_TO)
        .source(TaskSources.IOT)
        .requestId("mqtt-move-001")
        .traceId("mqtt-move-001")
        .putString(TaskParams.POSITION_ID, "4")
        .executionMode(TaskExecutionModes.ASYNC)
        .priority(TaskPriorities.HIGH)
        .queue(TaskQueues.MOVEMENT)
        .timeoutMs(240_000)
        .completionTarget(TaskCompletionTargets.MOVEMENT_ARRIVED)
        .preemptPolicy(TaskPreemptPolicies.APPEND)
        .build();

동일성 기준:

10.1 System App AAR와 MQTT의 같은 요청 비교

System App AAR:

TaskSubmitRequest request = TaskSubmitRequest.builder(TaskMethods.SET_MOVE_TO)
        .source(TaskSources.IOT)
        .requestId("mqtt-move-001")
        .traceId("mqtt-move-001")
        .putString(TaskParams.POSITION_ID, "4")
        .executionMode(TaskExecutionModes.ASYNC)
        .priority(TaskPriorities.HIGH)
        .queue(TaskQueues.MOVEMENT)
        .timeoutMs(240_000)
        .completionTarget(TaskCompletionTargets.MOVEMENT_ARRIVED)
        .preemptPolicy(TaskPreemptPolicies.APPEND)
        .build();

MQTT:

{
  "correlationId": "mqtt-move-001",
  "request": {
    "method": "setMoveTo",
    "payload": {
      "positionId": "4"
    },
    "task": {
      "executionMode": "async",
      "priority": "HIGH",
      "queueKey": "movement",
      "timeoutMs": 240000,
      "completionTarget": "movement.arrived",
      "preemptPolicy": "append"
    }
  }
}

위 MQTT 예시는 managed를 생략했지만 request.task object가 존재하므로 managed=true와 같은 TaskManager intent로 처리한다.

TaskManager가 받는 의미:

둘 다 같은 taskMethod=setMoveTo 실행 요청이다.
둘 다 movement queue에 async로 등록된다.
둘 다 movement.arrived를 완료 기준으로 본다.
둘 다 traceId/correlationId로 추적 가능해야 한다.

11. 제한해야 하는 값

MQTT가 내려주는 task metadata를 그대로 신뢰하면 안 된다.

현재 구현상 request.task field를 Bundle로 복사하는 것까지는 되어 있다. 위 제한 정책 중 일부는 DeviceAgent TaskCallerPolicy, TaskPolicyRegistry, TaskResourcePolicy에서 처리되고, IoT source 전용 clamp/allowlist는 추가 보강 대상이다.

11.1 테스트 근거

현재 focused unit test는 아래를 검증한다.

검증 명령:

cd ~/work/1.A1_SoC_new/SoC/a1-packages/apps/IotAgent
ANDROID_HOME=/home/silogood/Android/Sdk ANDROID_SDK_ROOT=/home/silogood/Android/Sdk \
./gradlew testDevDebugUnitTest \
  --tests 'com.sk.airbot.iotagent.mqtt.topic.IotTaskManagerBridgeTest' \
  --tests 'com.sk.airbot.iotagent.mqtt.topic.CmdTopicTaskManagerRoutingTest'

검증 결과:

BUILD SUCCESSFUL

12. 타팀 공유용 요약

MQTT는 기존 command schema를 유지해도 된다.
TaskManager 의도가 없으면 request.task를 생략하고 기존 direct command로 둔다.
TaskManager 의도가 필요한 경우 request.task에 공통 metadata를 넣는다.
request.task가 비어 있어도 TaskManager 기본 정책을 쓰겠다는 명시 intent로 본다.
IotAgent는 request.task를 AAR/Bundle field와 같은 이름으로 복사한다.
DeviceAgent는 명시 metadata를 허용 범위 내에서 반영하고, 없는 값은 기본 정책으로 채운다.

외부 앱 AAR, MQTT, PUI, On-device는 모두 taskMethod/source/executionMode/queueKey/completionTarget이라는 같은 언어를 사용한다.