SERVICES

서비스별 개발 가이드

각 서비스가 어떤 OSS를 어떻게 활용하고, 어디까지 직접 만들어야 하는지 정리. 메시지 스키마 상세는 MQTT_SCHEMA.md 참조.

분류 — 직접 개발 범위

서비스	OSS 그대로	설정 작성	코드 작성
`mosquitto`	●	○	—
`frigate`	●	●	—
`home-assistant`	●	●	—
`bridge`	(라이브러리만)	—	●
`orchestrator`	(라이브러리만)	●	●
`mcp-robot-adapter`	(라이브러리만)	—	●
`mcp-tablet-gateway`	(라이브러리만)	—	●
`tablet-app`	(프레임워크만)	—	●

1. mosquitto — MQTT broker

사용 OSS

이미지: eclipse-mosquitto:2.0
라이선스: EPL/EDL

개발 범위

그대로 사용. 설정 파일만.

설정 (`mosquitto/config/mosquitto.conf`)

listener 1883
allow_anonymous false
password_file /mosquitto/config/passwd
acl_file     /mosquitto/config/acl

persistence true
persistence_location /mosquitto/data/
log_dest stdout

# 메시지 보존 정책 (전시 운영 중 안전)
max_inflight_messages 100
max_queued_messages   1000

ACL 예시 (`acl`)

원칙: 최소 권한. 각 서비스는 자기 책임 토픽만 write, 필요한 토픽만 read. 모든 자체 서비스는 system/heartbeat/<자기이름> 발행 권한을 가진다.

# ──────────────────────────────────────────
# frigate: 자체 토픽만 발행
# ──────────────────────────────────────────
user frigate
topic write frigate/#

# ──────────────────────────────────────────
# bridge: frigate/# 구독, vision/# + session/+ + safety/proximity 발행
# ──────────────────────────────────────────
user bridge
topic read  frigate/#
topic write vision/zone/enter
topic write vision/zone/exit
topic write vision/position/update
topic write session/start
topic write safety/proximity
topic read  session/end                       # 임베딩 폐기 트리거 수신 (SCRUB)
topic write system/heartbeat/bridge

# ──────────────────────────────────────────
# orchestrator: 명령은 MCP로 전달 (MQTT 발행 X)
# ──────────────────────────────────────────
user orchestrator
topic read  vision/zone/enter
topic read  vision/zone/exit
topic read  vision/position/update
topic read  robot/+/state
topic read  robot/+/ack
topic read  robot/+/fault
topic read  tablet/+/intent
topic read  tablet/+/ack
topic read  context/+/+/update
topic read  context/environmental/update
topic read  safety/+
topic read  session/start
topic read  session/end
topic write tablet/+/show_options
topic write tablet/+/show_message
topic write context/user/+/update
topic write session/end                       # timeout 트리거
topic write system/heartbeat/orchestrator

# ──────────────────────────────────────────
# mcp-robot-adapter: 로봇 전용
# ──────────────────────────────────────────
user robot-adapter
topic write robot/+/state
topic write robot/+/ack
topic write robot/+/fault
topic read  robot/+/command                   # HA 폴백 경로 수신
topic read  safety/estop
topic read  safety/proximity
topic write safety/estop                      # 자체 감지 시 발행
topic read  vision/position/update            # proximity guard
topic write system/heartbeat/robot-adapter

# ──────────────────────────────────────────
# mcp-tablet-gateway: 태블릿 전용
# ──────────────────────────────────────────
user tablet-gateway
topic write tablet/+/intent
topic write tablet/+/ack
topic read  tablet/+/show_options
topic read  tablet/+/show_message
topic read  session/end                       # SCRUB 트리거 수신 (토큰 무효화·WSS 종료)
topic write system/heartbeat/tablet-gateway

# ──────────────────────────────────────────
# home-assistant: 라이브 모니터링·폴백 자동화·운영자 명령
# ──────────────────────────────────────────
user homeassistant
topic read  vision/#
topic read  robot/+/state
topic read  robot/+/ack
topic read  robot/+/fault
topic read  tablet/+/intent
topic read  tablet/+/ack
topic read  context/+/+/update
topic read  safety/#
topic read  session/+
topic read  system/heartbeat/+
topic write context/environmental/update
topic write safety/estop                      # 운영자 e-stop 콘솔
topic write robot/+/command                   # LLM 폴백 경로
topic write tablet/+/show_options             # 폴백 시 옵션 노출
topic write tablet/+/show_message
topic write session/end                       # 운영자 종료 트리거
topic write system/heartbeat/homeassistant

비밀번호 파일 권한: passwd 파일은 chmod 600, owner는 mosquitto 컨테이너 uid(1883). 호스트 디렉터리(./mosquitto/config)도 동일하게 제한.

운영 포인트

비밀번호는 mosquitto_passwd -c passwd <user>로 생성, .env.secret에 보관 X (파일 자체가 비밀).
TLS(Transport Layer Security, 전송 계층 보안)는 단일 사이트 LAN 내부에선 필수 아님. 외부 노출 시 8883 + cert.

2. frigate — CCTV NVR + 객체 탐지

사용 OSS

이미지: ghcr.io/blakeblackshear/frigate:0.14.1
라이선스: MIT

개발 범위

그대로 사용 + config.yml + zone polygon + 카메라 추가.
Cross-camera ReID(Re-identification, 보행자 재식별)는 부족 → bridge 서비스에서 보강.

설정 (`frigate/config.yml`)

원칙: config.yml은 git에 커밋 가능해야 하므로 자격증명을 평문으로 두지 않는다. Frigate 0.13+ 는 {ENV_VAR} 형태로 환경변수 치환을 지원하며, 컨테이너 환경변수(env_file)로 주입한다.

mqtt:
  host: mosquitto
  port: 1883
  user: frigate
  password: '{FRIGATE_MQTT_PASSWORD}'   # 환경변수 치환 (Frigate 0.13+)
  topic_prefix: frigate

detectors:
  coral:
    type: edgetpu
    device: usb
  # 또는 NVIDIA: type: tensorrt

cameras:
  cafe_cam_01:
    ffmpeg:
      inputs:
        - path: rtsp://{FRIGATE_RTSP_USER}:{FRIGATE_RTSP_PASSWORD}@10.0.1.11:554/stream
          roles: [detect, record]
    detect:
      width: 1280
      height: 720
      fps: 5
    objects:
      track: [person]
    zones:
      cafe_zone:
        coordinates: 0,720,0,0,1280,0,1280,720
        objects: [person]
    record:
      enabled: false   # 영상 저장 안 함 (개인정보)
    snapshots:
      enabled: false

  park_cam_01: { ... }
  vehicle_cam_01: { ... }
  home_cam_01: { ... }

compose에서 주입:

# deploy/docker-compose.yml (frigate 서비스)
frigate:
  image: ghcr.io/blakeblackshear/frigate:0.14.1
  env_file: [./.env, ./.env.secret]   # FRIGATE_MQTT_PASSWORD, FRIGATE_RTSP_USER, FRIGATE_RTSP_PASSWORD
  ...

.env.secret 정의 (gitignore):

FRIGATE_MQTT_PASSWORD=...
FRIGATE_RTSP_USER=admin
FRIGATE_RTSP_PASSWORD=...

카메라마다 자격증명이 다르면 FRIGATE_RTSP_PASSWORD_CAFE_01 등으로 분리. Frigate는 임의 환경변수명을 모두 치환한다.

발행 토픽 (Frigate 기본)

frigate/events — 이벤트 JSON
frigate/{camera}/{label} — 카운트
frigate/{camera}/{zone} — 점유 ON/OFF

우리 도메인 토픽으로의 변환은 `bridge`가 담당.

운영 포인트

모델 정확도 향상은 labelmap·threshold 튜닝으로 시작, 커스텀 모델은 후순위.
GPU/TPU 선택: Jetson은 TensorRT(NVIDIA TensorRT, NVIDIA 딥러닝 추론 최적화 엔진), USB 환경은 Coral, x86 GPU는 NVIDIA TensorRT detector.

3. home-assistant — 상태 + 운영 대시보드

사용 OSS

이미지: ghcr.io/home-assistant/home-assistant:2026.1
라이선스: Apache 2.0

개발 범위

그대로 사용 + 다음을 우리가 작성:
- MQTT(Message Queuing Telemetry Transport, 메시지 큐 텔레메트리 전송 프로토콜) integration 활성화
- 엔티티 정의 (template sensor, MQTT discovery)
- 자동화 (rule-based 폴백 시나리오)
- Lovelace 대시보드 YAML
- Zone playbook용 input_select / input_boolean

디렉터리 (`ha-config/`)

ha-config/
├── configuration.yaml
├── automations.yaml
├── scripts.yaml
├── secrets.yaml          # gitignore
├── packages/
│   ├── sessions.yaml     # 세션 엔티티
│   ├── robots.yaml       # 로봇 엔티티
│   ├── zones.yaml        # 공간 점유 엔티티
│   └── fallback.yaml     # LLM 폴백 자동화
└── lovelace/
    └── operator.yaml     # 운영자 대시보드

`configuration.yaml` 핵심

mqtt:
  broker: mosquitto
  username: !secret mqtt_user
  password: !secret mqtt_password
  discovery: true
  discovery_prefix: homeassistant

recorder:
  purge_keep_days: 3
  exclude:
    domains: [automation, updater]

http:
  use_x_forwarded_for: true
  trusted_proxies: [127.0.0.1]

homeassistant:
  packages: !include_dir_named packages

폴백 시나리오 카탈로그 (LLM 단절 시 rule-based)

LLM(Large Language Model, 대규모 언어 모델)이 응답 못 하는 동안에도 시연이 끊기지 않도록 모든 zone에 정적 시나리오를 사전 정의한다. 트리거는 공통 — "orchestrator heartbeat 30초 이상 미수신 + zone enter 또는 tablet intent 발생".

카탈로그

Zone	트리거	폴백 액션 (로봇 + 태블릿)
`cafe`	`vision/zone/enter`	• SPOT `look_at` (사용자) • 태블릿: 옵션 `["커피 추천 (오프라인)", "조용히 둘러보기"]`
`park`	`vision/zone/enter`	• SPOT `gesture: wave` • 태블릿: 메시지 "공원 산책로를 따라 이동해보세요" + 옵션 `["로봇과 산책", "혼자 산책"]`
`vehicle`	`vision/zone/enter`	• 로봇 명령 없음 (차량 내부 협소) • 태블릿: 메시지 "운전석에 앉아보세요" + 옵션 `["오디오 데모", "디스플레이 데모"]`
`home`	`vision/zone/enter`	• SPOT `sit` (실내 안전 자세) • 태블릿: 옵션 `["거실 분위기", "주방 분위기", "침실 분위기"]`
any	`tablet/+/intent` (옵션 선택)	• 태블릿: 메시지 "선택 감사합니다. 운영자에게 안내드릴게요." (정적 응답)
any	`safety/estop active=true`	• 모든 활성 태블릿: 메시지 "잠시 시연을 멈추었습니다" 표시

자동화 코드 패턴

packages/fallback.yaml (모든 zone 공통 헬퍼):

# orchestrator 활성 여부 sensor
template:
  - binary_sensor:
      - name: orchestrator_alive
        state: >
          {{ (now() - states.sensor.orchestrator_heartbeat.last_updated).total_seconds() < 30 }}
        availability: "{{ states('sensor.orchestrator_heartbeat') not in ['unknown', 'unavailable'] }}"

# 공통 trigger·condition 변수로 묶음
input_text:
  fallback_last_session:
    name: "마지막 폴백 발동 세션"

zone별 자동화 (4개):

automation:
  - alias: "Fallback - cafe enter"
    trigger:
      - platform: mqtt
        topic: vision/zone/enter
    condition:
      - condition: state
        entity_id: binary_sensor.orchestrator_alive
        state: 'off'
      - condition: template
        value_template: "{{ trigger.payload_json.data.zone_id == 'cafe' }}"
    action:
      - service: mqtt.publish
        data:
          topic: "robot/spot-01/command"
          payload_template: >
            {
              "command_id": "{{ now().timestamp() | int | string }}",
              "robot_id": "spot-01",
              "capability": "look_at",
              "params": {"target_session_id": "{{ trigger.payload_json.data.session_id }}"},
              "priority": "operator",
              "source": {"service": "home-assistant", "trace_id": "fallback-{{ now().timestamp() | int }}"},
              "expires_at": null
            }
          qos: 1
      - service: mqtt.publish
        data:
          topic: "tablet/{{ states('input_text.session_to_tablet_' ~ trigger.payload_json.data.session_id) }}/show_options"
          payload: >
            {
              "tablet_id": "...",
              "message_id": "fallback-cafe-{{ now().timestamp() | int }}",
              "options": [
                {"id": "fallback_drink", "label": "커피 추천 (오프라인)"},
                {"id": "fallback_quiet", "label": "조용히 둘러보기"}
              ],
              "ttl_s": 60,
              "timestamp": "{{ now().isoformat() }}"
            }
          qos: 1

  - alias: "Fallback - park enter"
    # 동일 패턴, capability=gesture, params.name=wave, 옵션 변경
    ...

  - alias: "Fallback - vehicle enter"
    # 로봇 명령 없음, 태블릿 메시지만
    ...

  - alias: "Fallback - home enter"
    # capability=sit, 옵션 3개
    ...

  - alias: "Fallback - tablet intent (LLM down)"
    trigger:
      - platform: mqtt
        topic: tablet/+/intent
    condition:
      - condition: state
        entity_id: binary_sensor.orchestrator_alive
        state: 'off'
    action:
      - service: mqtt.publish
        data:
          topic: "tablet/{{ trigger.topic.split('/')[1] }}/show_message"
          payload: >
            {
              "message_id": "fallback-ack-{{ now().timestamp() | int }}",
              "content": {"text": "선택 감사합니다. 운영자에게 안내드릴게요.", "media": [], "speaker": "system"},
              "duration_s": 5,
              "timestamp": "{{ now().isoformat() }}"
            }

  - alias: "Fallback - estop broadcast"
    trigger:
      - platform: mqtt
        topic: safety/estop
    condition:
      - condition: template
        value_template: "{{ trigger.payload_json.data.active == true }}"
    action:
      # 모든 활성 태블릿에 broadcast (세션 목록은 HA entity로 보유)
      - repeat:
          for_each: "{{ state_attr('sensor.active_sessions', 'tablet_ids') | default([]) }}"
          sequence:
            - service: mqtt.publish
              data:
                topic: "tablet/{{ repeat.item }}/show_message"
                payload: >
                  {
                    "content": {"text": "잠시 시연을 멈추었습니다", "speaker": "system"},
                    "duration_s": null
                  }

검증 절차 (D-7 체크)

orchestrator 컨테이너를 docker compose stop으로 강제 종료.
30초 후 zone enter 트리거 → 위 4개 자동화 각각 발동되는지 라이브 테일로 확인:
```
docker exec mosquitto mosquitto_sub -t '#' -v
```
각 zone 시나리오의 메시지·로봇 동작 모두 검증 후 orchestrator 재기동.

충돌 정책

LLM이 복귀하면 binary_sensor.orchestrator_alive가 on이 되어 자동화 condition 차단 — 추가 폴백 발동 안 함.
진행 중이던 폴백 메시지는 자체 TTL(60s)로 자연 소멸. orchestrator의 신규 메시지가 덮어쓰면 즉시 교체.

Lovelace 운영자 대시보드 (요지)

Glance 카드: 로봇 배터리·연결·현재 동작
Picture-elements: 전시장 floor plan 위에 활성 세션 위치 오버레이
Conditional 카드: e-stop(emergency stop, 긴급 정지) 활성 시 큰 빨간 배너
History graph: 최근 1시간 zone 점유 변화

4. bridge — Frigate ↔ 도메인 변환 + ReID 보강

사용 OSS·라이브러리

베이스: python:3.12-slim
paho-mqtt: MQTT 클라이언트
pydantic: 스키마 검증
numpy, scipy: 임베딩 매칭
(선택) boxmot: cross-camera ReID 알고리즘 — AGPL 주의

개발 범위

전부 자체 개발. 외부 OSS는 라이브러리 수준만 활용.

핵심 책임

frigate/events 구독 → 우리 vision/zone/{enter|exit} 토픽으로 변환
세션 ↔ track ID 매핑: 태블릿 페어링 시 캡처한 외형 임베딩과 Frigate 추적 결과를 매칭
Cross-camera ReID: 카메라 간 동일 인물 매칭 (Frigate 미지원 영역)
Position update를 5Hz로 발행

Zone enter/exit 판정 알고리즘

Frigate frigate/events는 객체 단위로 발행되며, 동일 person이 zone 경계에서 깜빡이거나 카메라 전환 시 노이즈가 발생한다. 단순 변환 시 다음 문제가 생김:

같은 person이 cafe→park 이동 중 잠깐 둘 다 ON → 중복 enter.
person이 zone 경계에서 머무르며 ON/OFF 반복 → 깜빡거리는 enter/exit.
cross-camera 전환 시 Frigate track ID 변경 → exit 후 즉시 enter.

따라서 session_id 단위로 "현재 zone" 상태를 bridge가 보유하고 hysteresis(이력)을 두어 판정한다.

상태 모델

# bridge/zone_tracker.py
@dataclass
class SessionZoneState:
    session_id: str
    current_zone: ZoneId | None
    candidate_zone: ZoneId | None    # 전이 중 후보
    candidate_since: datetime | None # 후보 진입 시각
    last_seen: datetime

판정 규칙 (hysteresis)

입력	조건	출력
`frigate/events`의 `entered_zones`에 새 zone Z 등장 (현재 다른 zone)	`Z != current_zone`	candidate_zone=Z, candidate_since=now
candidate_zone이 `ENTER_HOLD_MS`(800ms) 동안 유지	timer 만료	`vision/zone/exit` (current_zone) → `vision/zone/enter` (Z) → current_zone=Z
candidate_zone이 hold 중 다시 사라짐	flicker	candidate 폐기 (이전 enter는 무시)
`frigate/events`에 person 사라짐 (current_zones 빔)	last_seen 갱신만	(즉시 exit 안 함)
last_seen이 `EXIT_GRACE_MS`(3000ms) 이상 stale	timer 만료	`vision/zone/exit` (current_zone) → current_zone=None
cross-camera 전환으로 Frigate track ID 변경	ReID로 같은 session_id 매칭	동일 session으로 처리 — exit/enter 발행 안 함

권장 임계값 (튜닝 가능, `config/bridge.yaml`)

zone_tracker:
  enter_hold_ms: 800       # 짧으면 깜빡거림, 길면 NFR(1.5s) 위반
  exit_grace_ms: 3000      # 짧으면 잠깐 가린 인물도 exit, 길면 zone 사이 빈 공간 길어짐
  stale_timeout_ms: 30000  # 이 시간 후 session 자체를 dropped로 처리(다른 메커니즘)

의사코드

async def on_frigate_event(event):
    person = event["after"]
    session_id = registry.match(person)         # ReID
    if not session_id:
        return  # 페어링 안 된 인물 — 무시

    zones_now = set(person["current_zones"])    # 예: {"cafe_zone"}
    domain_zone = first_match(zones_now, ZONE_MAP)  # 'cafe' 등

    state = states.get(session_id)
    state.last_seen = now()

    if domain_zone and domain_zone != state.current_zone:
        if state.candidate_zone != domain_zone:
            state.candidate_zone = domain_zone
            state.candidate_since = now()
        elif now() - state.candidate_since >= ENTER_HOLD_MS:
            await commit_transition(state, domain_zone)
    elif not domain_zone:
        # zone 없음 — exit 후보 (timer는 별도 tick에서 처리)
        state.candidate_zone = None

# 별도 tick (100ms 주기)
async def tick():
    for state in states.values():
        if not state.current_zone:
            continue
        if now() - state.last_seen >= EXIT_GRACE_MS:
            await publish_exit(state.session_id, state.current_zone)
            state.current_zone = None

멱등성·중복 방지

vision/zone/enter 발행 직후 동일 (session_id, zone_id) 중복 enter 1초 내 발행 금지.
envelope id는 항상 새 ULID(Universally Unique Lexicographically Sortable Identifier, 범용 고유 정렬 가능 식별자) — 구독자가 같은 메시지를 두 번 처리하지 않도록.

디렉터리 구조

services/bridge/
├── Dockerfile
├── requirements.txt
├── pyproject.toml
├── src/
│   └── bridge/
│       ├── __init__.py
│       ├── main.py              # 엔트리포인트
│       ├── mqtt_client.py       # paho-mqtt 래퍼
│       ├── frigate_adapter.py   # frigate/events 파서
│       ├── reid.py              # 임베딩 매칭, cross-cam
│       ├── session_registry.py  # session_id ↔ track_id
│       ├── topics.py            # 토픽 상수
│       └── schemas.py           # pydantic 모델
└── tests/
    ├── test_frigate_adapter.py
    ├── test_reid.py
    └── fixtures/
        └── frigate_events/      # 샘플 JSON

핵심 모듈 의사코드

# src/bridge/main.py
from bridge.mqtt_client import Mqtt
from bridge.frigate_adapter import handle_frigate_event
from bridge.session_registry import registry

mqtt = Mqtt(host="mosquitto", user="bridge", password=os.environ["MQTT_PWD"])

@mqtt.on("frigate/events")
def on_frigate(msg):
    domain_events = handle_frigate_event(msg, registry)
    for ev in domain_events:
        mqtt.publish(ev.topic, ev.payload, qos=1)

@mqtt.on("session/end")
def on_session_end(msg):
    registry.unbind(session_id=msg["session_id"])  # SCRUB: 임베딩 참조 해제

mqtt.run_forever()

테스트 전략

유닛: frigate_adapter에 픽스처 JSON 입력 → 도메인 이벤트 출력 검증.
통합: 임베디드 mosquitto + Frigate 이벤트 시뮬레이터로 E2E.
성능: 10명 동시 추적 시 latency p95 < 100ms 검증.

5. orchestrator — LangGraph + Claude

사용 OSS·라이브러리

베이스: python:3.12-slim
anthropic: Claude SDK(Software Development Kit, 소프트웨어 개발 키트)
langgraph: 그래프 오케스트레이션
mcp Python SDK: MCP(Model Context Protocol, 모델 컨텍스트 프로토콜) 서버 연결
paho-mqtt, pydantic, httpx(HA REST(Representational State Transfer, 표현 상태 전이)), prometheus-client

개발 범위

전부 자체 개발. LangGraph는 골격, 노드와 zone playbook은 우리가 작성.

핵심 책임

MQTT 트리거 구독 (vision/zone/enter, tablet/+/intent, context/+/+/update, context/environmental/update)
HA REST/WebSocket으로 컨텍스트 어셈블
Claude tool use 호출 (MCP 서버 노출)
가드레일 (스키마·whitelist·hard clamp)
도구 호출 결과를 MQTT publish 또는 MCP dispatch
폴백 모드 (Claude 실패 시 rule-based)

디렉터리 구조

services/orchestrator/
├── Dockerfile
├── pyproject.toml
├── src/
│   └── orchestrator/
│       ├── main.py
│       ├── graph.py             # LangGraph 정의
│       ├── nodes/
│       │   ├── assemble.py
│       │   ├── call_claude.py
│       │   ├── guardrail.py
│       │   ├── dispatch.py
│       │   └── fallback.py
│       ├── ha_client.py         # HA REST/WS
│       ├── mcp_clients.py       # robot/tablet/ha MCP 연결
│       ├── prompts/
│       │   ├── system.md
│       │   └── playbooks/
│       │       ├── cafe.md
│       │       ├── park.md
│       │       ├── vehicle.md
│       │       └── home.md
│       ├── tools.py             # Claude tool 스키마
│       ├── metrics.py           # Prometheus
│       └── topics.py
└── tests/
    ├── test_graph.py
    ├── test_guardrail.py
    └── test_playbooks.py

Claude tool 정의 (`tools.py`)

TOOLS = [
    {
        "name": "robot_command",
        "description": "로봇에게 capability 명령을 발행. 안전 인터록 통과 시에만 실행됨.",
        "input_schema": {
            "type": "object",
            "properties": {
                "robot_id": {"type": "string"},  # 런타임에 ROBOT_REGISTRY로 검증 (config/robots.yaml)
                "capability": {"type": "string",
                               "enum": ["goto", "follow", "stop", "look_at", "gesture"]},
                "params": {"type": "object"},
            },
            "required": ["robot_id", "capability"],
        },
    },
    {
        "name": "show_options",
        "description": "체험자 태블릿에 선택지를 표시.",
        "input_schema": {...},
    },
    # ...
]

가드레일 (`nodes/guardrail.py`)

def validate_tool_calls(state: FlowState) -> dict:
    accepted, rejected = [], []
    for call in state["tool_calls"]:
        # 1. JSON schema
        if not validate_schema(call):
            rejected.append((call, "schema"))
            continue
        # 2. capability whitelist
        if call["input"]["capability"] not in ALLOWED_CAPS:
            rejected.append((call, "whitelist"))
            continue
        # 3. zone polygon clamp
        if "params" in call["input"] and "target_pos" in call["input"]["params"]:
            if not in_zone_polygon(call["input"]["params"]["target_pos"]):
                rejected.append((call, "zone_fence"))
                continue
        # 4. velocity clamp
        if "velocity" in call["input"].get("params", {}):
            call["input"]["params"]["velocity"] = min(
                call["input"]["params"]["velocity"], MAX_VELOCITY
            )
        accepted.append(call)
    REJECT_COUNTER.inc(len(rejected))
    return {"tool_calls": accepted, "rejected_calls": rejected}

Prompt 캐싱

시스템 프롬프트 + zone playbook은 stable → Anthropic prompt cache 활용.

테스트 전략

노드 단위: 각 LangGraph 노드 입출력 검증.
가드레일 fuzz: 의도적으로 잘못된 도구 호출 1000건 → 100% 거부 확인.
시뮬레이션: 가짜 MQTT 이벤트 stream → 그래프 종단 동작 검증.

6. mcp-robot-adapter — 로봇 SDK 추상화 + 안전 인터록

사용 OSS·라이브러리

베이스: python:3.12-slim
bosdyn-client, bosdyn-api: SPOT SDK (Boston Dynamics 별도 EULA)
(미정 로봇 결정 후) ROS2 rclpy 또는 해당 SDK
mcp Python SDK: MCP server 노출
paho-mqtt, pydantic

개발 범위

전부 자체 개발. SPOT SDK는 라이브러리 호출 수준.

핵심 책임

이종 로봇 추상화: Capability Model 인터페이스를 SPOT/미정 로봇 어댑터가 구현
MCP 도구 노출: orchestrator가 호출할 robot_command 도구
안전 인터록: Hardware e-stop · zone polygon · proximity guard · watchdog
명령 큐: priority(safety > operator > llm) FIFO
텔레메트리: 1Hz robot/{id}/state 발행
Fault 감지: robot/{id}/fault 발행

디렉터리 구조

services/mcp-robot-adapter/
├── Dockerfile
├── pyproject.toml
├── src/
│   └── adapter/
│       ├── main.py              # MCP server + MQTT bridge
│       ├── capability_model.py  # 추상 인터페이스
│       ├── adapters/
│       │   ├── spot.py          # bosdyn-client 래핑
│       │   └── generic_ros2.py  # 미정 로봇용
│       ├── safety/
│       │   ├── interlock.py     # 통합 게이트
│       │   ├── estop.py
│       │   ├── zone_fence.py
│       │   ├── proximity.py
│       │   └── watchdog.py
│       ├── command_queue.py
│       ├── telemetry.py
│       └── topics.py
└── tests/
    ├── test_safety.py
    ├── test_capability_model.py
    └── test_command_queue.py

Capability Model 인터페이스

# capability_model.py
from abc import ABC, abstractmethod
from typing import Literal

Capability = Literal["goto", "follow", "stop", "sit", "stand", "gesture", "look_at"]

class RobotAdapter(ABC):
    @abstractmethod
    async def execute(self, capability: Capability, params: dict) -> dict: ...

    @abstractmethod
    async def get_state(self) -> dict: ...

    @abstractmethod
    async def emergency_stop(self) -> None: ...

SPOT 어댑터 (스니펫)

# adapters/spot.py
from bosdyn.client import create_standard_sdk
from bosdyn.client.robot_command import RobotCommandClient, RobotCommandBuilder

class SpotAdapter(RobotAdapter):
    def __init__(self, hostname, username, password):
        sdk = create_standard_sdk("zerone-spot")
        self.robot = sdk.create_robot(hostname)
        self.robot.authenticate(username, password)
        self.cmd_client = self.robot.ensure_client(RobotCommandClient.default_service_name)

    async def execute(self, capability, params):
        if capability == "stop":
            cmd = RobotCommandBuilder.stop_command()
        elif capability == "goto":
            cmd = RobotCommandBuilder.synchro_se2_trajectory_point_command(
                goal_x=params["x"], goal_y=params["y"], goal_heading=params.get("yaw", 0),
                frame_name=ODOM_FRAME_NAME,
            )
        elif capability == "look_at":
            cmd = self._build_look_at(params["target"])
        else:
            raise ValueError(f"Unknown capability: {capability}")
        future = self.cmd_client.robot_command_async(cmd)
        return {"command_id": future.command_id}

안전 인터록 게이트

# safety/interlock.py
class SafetyGate:
    def __init__(self, estop, fence, proximity, watchdog):
        self.checks = [estop, fence, proximity, watchdog]

    async def allow(self, robot_id, capability, params) -> tuple[bool, str | None]:
        for check in self.checks:
            ok, reason = await check.evaluate(robot_id, capability, params)
            if not ok:
                return False, reason
        return True, None

테스트 전략

유닛: 각 안전 체크가 정확한 조건에서 거부하는지.
계약 테스트: SPOT 시뮬레이터(BD에서 제공) 또는 mock 어댑터로 capability 호출 검증.
카오스: e-stop 이벤트 발생 시 큐에 남은 모든 명령 거부 확인.

7. mcp-tablet-gateway — 태블릿 WSS + MCP 도구

사용 OSS·라이브러리

베이스: node:20-alpine
ws: WebSocket 서버
mqtt: MQTT 클라이언트
@modelcontextprotocol/sdk: MCP server (TypeScript)
zod: 스키마 검증

개발 범위

전부 자체 개발.

핵심 책임

태블릿 WSS(WebSocket Secure, 보안 웹소켓) 연결 관리 (재연결·세션 페어링)
MQTT tablet/{id}/show_options·show_message 구독 → WSS push
WSS 입력 → MQTT tablet/{id}/intent 발행
MCP 도구로 show_options·show_message를 외부에 노출 (orchestrator 직접 호출용)
메시지 ID 추적 + ACK

디렉터리 구조

services/mcp-tablet-gateway/
├── Dockerfile
├── package.json
├── tsconfig.json
├── src/
│   ├── main.ts
│   ├── wss-server.ts
│   ├── mqtt-bridge.ts
│   ├── mcp-server.ts
│   ├── session-pairing.ts   # QR 토큰 발급
│   ├── schemas.ts
│   └── topics.ts
└── tests/
    ├── wss.test.ts
    └── mqtt-bridge.test.ts

WSS 서버 골격

import { WebSocketServer } from "ws";
import mqtt from "mqtt";

const wss = new WebSocketServer({ port: 8080 });
const mq = mqtt.connect("mqtt://mosquitto");

const tablets = new Map<TabletId, WebSocket>();

wss.on("connection", (ws, req) => {
  const tabletId = authenticate(req);
  tablets.set(tabletId, ws);

  mq.subscribe(`tablet/${tabletId}/+`);

  ws.on("message", (raw) => {
    const intent = parseIntent(raw);   // zod 검증
    mq.publish(`tablet/${tabletId}/intent`, JSON.stringify(intent), { qos: 1 });
  });

  ws.on("close", () => tablets.delete(tabletId));
});

mq.on("message", (topic, payload) => {
  const [, tabletId, type] = topic.split("/");
  const ws = tablets.get(tabletId);
  if (!ws) return;
  ws.send(JSON.stringify({ type, ...JSON.parse(payload.toString()) }));
});

페어링 흐름

전제: 전시장 입구에 페어링 카메라 1대(또는 입구 zone 카메라 겸용)가 항상 켜져 있고, Frigate가 person 추적 중. 태블릿은 부팅 후 운영자가 체험자에게 전달.

시퀀스 다이어그램

Tablet           tablet-gateway        Frigate         bridge          orchestrator
  │                    │                  │              │                  │
  │ POST /pair         │                  │              │                  │
  ├───────────────────►│                  │              │                  │
  │                    │ generate         │              │                  │
  │                    │   - session_id (ULID)           │                  │
  │                    │   - pairing_token (32B random)  │                  │
  │                    │   - QR(payload=session_id)      │                  │
  │ 200 { session_id, token, qr_png }    │              │                  │
  │◄───────────────────┤                  │              │                  │
  │                    │                  │              │                  │
  │ display QR + "잠시만 카메라를 봐주세요" │              │                  │
  │                    │                  │              │                  │
  │ ─── 운영자가 태블릿을 페어링 카메라 앞으로 ───        │                  │
  │                    │                  │              │                  │
  │                    │                  │ person 감지   │                  │
  │                    │                  │ + QR 디코딩   │                  │
  │                    │                  │ (zone=pairing)│                  │
  │                    │                  │              │                  │
  │                    │                  ├─MQTT frigate/events──►          │
  │                    │                  │              │                  │
  │                    │                  │              │ 매칭:             │
  │                    │                  │              │  - person bbox에서│
  │                    │                  │              │    OSNet 임베딩 추출│
  │                    │                  │              │  - QR payload에서  │
  │                    │                  │              │    session_id 추출 │
  │                    │                  │              │  - registry.bind() │
  │                    │                  │              │                  │
  │                    │                  │              │ MQTT session/start│
  │                    │                  │              ├──────────────────►│
  │                    │                  │              │                  │ assemble + welcome
  │                    │                  │              │                  │ Claude tool call:
  │                    │                  │              │                  │   show_message("환영합니다")
  │ WSS upgrade Auth: Bearer <token>      │              │                  │
  ├───────────────────►│ verify token     │              │                  │
  │                    │ session_id 매핑   │              │                  │
  │ WSS open           │                  │              │                  │
  │◄───────────────────┤                  │              │                  │
  │                    │                  │              │                  │
  │ {type:"show_message", text:"환영..."} │              │                  │
  │◄───────────────────┤◄─────────────────────────────────────────────────┤
  │                    │                  │              │                  │

단계별 책임

mcp-tablet-gateway POST /pair:
- session_id ULID 생성, pairing_token 32-byte random.
- QR 이미지 생성: payload = session_id (URL-safe base64).
- 응답: {session_id, token, qr_png_base64}. 토큰 TTL 4시간.
- 토큰은 in-memory Map에 {token: session_id} 저장.
태블릿: QR 풀스크린 + 음성 안내 "카메라를 봐주세요". 5초 카운트다운.
Frigate: 페어링 카메라(예: pairing_cam_01)에서 person 감지 + QR 디코드 (zone polygon pairing_zone). 디코드 결과를 attribute로 이벤트 발행.
bridge:
- frigate/events 구독.
- current_zones에 pairing_zone 포함 + QR attribute에서 session_id 추출 시:
  - person bbox 영역에서 OSNet(Omniscale Feature Network, 옴니스케일 특징 네트워크) 임베딩 추출 (BoxMOT 또는 자체 모델).
  - registry.bind(session_id, embedding).
  - session/start MQTT publish ({session_id, paired_tablet_id, visual_embedding_ref, started_at}).
orchestrator: session/start 수신 → 환영 시나리오 실행 (Claude show_message).
태블릿: 토큰으로 WSS 연결 → orchestrator 메시지 수신 시작.

실패 모드

상황	동작
태블릿이 QR 표시 후 60초 내 카메라 인식 못 함	태블릿 자동으로 페어링 화면 재시작, 토큰 폐기
Frigate가 person은 보지만 QR 디코드 실패	bridge가 무시 (session_id 없으면 bind 안 함)
같은 session_id로 다른 임베딩 재바인딩 시도	기존 임베딩 덮어쓰기 (재페어링 의도) + warn 로그
WSS 연결 시 토큰 만료	401 응답 → 태블릿 재페어링

운영자 가이드

페어링 카메라 위치: 입구·태블릿 수령대 기준 1.5m 거리, 머리·상반신 잘 보이게.
체험자에게 "QR이 사라질 때까지 카메라를 봐주세요" 안내.
동시 페어링 회피: 한 번에 한 명씩 카메라 앞으로.

테스트 전략

유닛: zod 스키마, 페어링 토큰 생성/만료.
통합: 가짜 mosquitto + WSS 클라이언트로 메시지 round-trip.

8. tablet-app — 체험자 인터랙션 (옵션)

사용 OSS·라이브러리

프레임워크: React Native (권장·기본). Flutter도 가능하나 본 시스템은 RN 생태계 기준으로 가이드 — 변경 시 별도 PoC 필요.
ws: WSS 연결 (mcp-tablet-gateway와 통신).
react-native-vision-camera: QR 스캔.
zustand: 상태 관리.

개발 범위

전부 자체 개발.

화면 흐름

부팅 → 페어링 대기: QR 표시 (또는 운영자 스캔 모드).
세션 진행: orchestrator가 push한 show_options/show_message를 동적으로 렌더.
인터랙션: 사용자가 옵션 선택·자유 텍스트 입력 → tablet/{id}/intent publish.
세션 종료: 운영자 트리거 또는 timeout.

권장 라이브러리 구성

UI: NativeBase / React Native Paper
애니메이션: Reanimated 3
미디어: react-native-video (LLM이 영상 응답 시)

디렉터리 (참고)

apps/tablet/
├── App.tsx
├── src/
│   ├── screens/
│   │   ├── PairingScreen.tsx
│   │   ├── SessionScreen.tsx
│   │   └── EndScreen.tsx
│   ├── components/
│   │   ├── OptionsCard.tsx
│   │   └── MessageBubble.tsx
│   ├── stores/
│   │   └── session.ts
│   ├── transport/
│   │   ├── wss.ts
│   │   └── auth.ts
│   └── schemas.ts
└── ios|android/

대안

인터랙션 단순하면 Home Assistant Lovelace 풀스크린 + 커스텀 카드로도 가능 (개발 비용 ↓, 표현 자유도 ↓).

9. 공통 개발 가이드

9.1 모노레포 구조 권장

zerone/
├── deploy/                    # 운영 docker-compose, .env, configs
│   ├── docker-compose.yml
│   ├── frigate/config.yml
│   ├── ha-config/
│   ├── mosquitto/
│   └── prometheus/
├── services/
│   ├── bridge/
│   ├── orchestrator/
│   ├── mcp-robot-adapter/
│   └── mcp-tablet-gateway/
├── apps/
│   └── tablet/
├── packages/                  # 공유 라이브러리
│   └── topics/                # 토픽 상수·스키마 (TS/Python 듀얼)
├── .github/workflows/
└── docs/                      # SYSTEM_DESIGN.md, MQTT_SCHEMA.md, ...

9.2 토픽·스키마 단일 소스

packages/topics에 TypeScript와 Python에서 모두 import 가능한 형태로 정의 (Pydantic ↔ Zod 동기화). 변경 시 양쪽 자동 반영하도록 codegen.

9.3 공통 환경변수

모든 서비스가 공유:

MQTT_HOST=mosquitto
MQTT_PORT=1883
LOG_LEVEL=info
OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318
SERVICE_NAME=<each>

9.4 헬스체크 엔드포인트

모든 자체 서비스는 GET /healthz 노출 — Docker HEALTHCHECK와 Prometheus probe에서 사용.

9.5 로깅

구조화 JSON (structlog Python, pino Node).
trace_id 필드 필수 — orchestrator가 발행, 모든 다운스트림이 전파.

9.5.1 OpenTelemetry 트레이싱 전파

MQTT envelope의 traceparent는 W3C Trace Context 형식. 모든 자체 서비스는 다음 패턴으로 추출·전파한다.

Python (bridge, orchestrator, mcp-robot-adapter)

의존성:

opentelemetry-api ~= 1.27
opentelemetry-sdk ~= 1.27
opentelemetry-exporter-otlp-proto-http ~= 1.27
opentelemetry-instrumentation-paho-mqtt ~= 0.48b0

부트스트랩:

# packages/topics/tracing.py (모든 Python 서비스 공통)
from opentelemetry import trace
from opentelemetry.propagate import extract, inject
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

def init_tracing(service_name: str):
    provider = TracerProvider(resource=Resource.create({"service.name": service_name}))
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))  # OTEL_EXPORTER_OTLP_ENDPOINT
    trace.set_tracer_provider(provider)

발행자 패턴:

tracer = trace.get_tracer(__name__)

def publish_with_trace(topic: str, data: dict, span_name: str):
    with tracer.start_as_current_span(span_name) as span:
        carrier = {}
        inject(carrier)            # carrier["traceparent"] = "00-..."
        envelope = wrap(
            type_=topic.replace("/", "."),
            data=data,
            source=f"/services/{SERVICE_NAME}",
            traceparent=carrier.get("traceparent"),
        )
        mqtt.publish(topic, json.dumps(envelope), qos=1)

구독자 패턴:

def on_message(topic, payload):
    envelope = json.loads(payload)
    ctx = extract({"traceparent": envelope.get("traceparent", "")})
    with tracer.start_as_current_span("handle_" + envelope["type"], context=ctx) as span:
        span.set_attribute("event.id", envelope["id"])
        span.set_attribute("event.type", envelope["type"])
        process(envelope["data"])

Node.js (mcp-tablet-gateway)

의존성:

"@opentelemetry/api": "^1.9",
"@opentelemetry/sdk-node": "^0.55",
"@opentelemetry/exporter-trace-otlp-http": "^0.55"

import { trace, context, propagation } from "@opentelemetry/api";

const tracer = trace.getTracer("mcp-tablet-gateway");

function publishWithTrace(topic: string, data: unknown, spanName: string) {
  return tracer.startActiveSpan(spanName, (span) => {
    const carrier: Record<string, string> = {};
    propagation.inject(context.active(), carrier);
    const envelope = wrap({ type: topic.replaceAll("/", "."), data, traceparent: carrier.traceparent });
    mqtt.publish(topic, JSON.stringify(envelope), { qos: 1 });
    span.end();
  });
}

외부 서비스 호출

Anthropic API: orchestrator가 호출 시 현재 span의 traceparent를 HTTP traceparent 헤더로 자동 주입 (OTLPInstrumentation 사용 시).
HA REST: httpx instrumentor 활성화 — span 자동 추가.
MCP 호출 (orchestrator → mcp-robot-adapter): MCP transport가 HTTP/SSE면 동일하게 traceparent 헤더 주입.

설정

환경변수: OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 (또는 직접 Tempo).
운영에선 Tempo 또는 Grafana Cloud Traces로 export. 미러 환경에선 Jaeger UI(localhost:16686)로 시각화.

Trace 흐름 (예)

[Frigate event] → bridge.handle_frigate_event
  └─ bridge.publish: vision/zone/enter   (traceparent 신규 생성)
      └─ orchestrator.handle_vision_zone_enter   (extract + 동일 trace 계속)
          ├─ orchestrator.assemble_context
          │   ├─ http GET /api/states/...   (HA REST)
          │   └─ http GET /api/states/...
          ├─ orchestrator.call_claude        (Anthropic API)
          ├─ orchestrator.guardrail
          └─ orchestrator.dispatch
              └─ mcp-robot-adapter.robot_command   (MCP 호출, 같은 trace)
                  └─ adapter.execute_spot          (SDK 호출)

한 줄로 이어지는 trace를 Grafana에서 펼쳐 latency 병목·오류 추적.

9.6 MCP 서버 부분 — 표준 준수

mcp-robot-adapter, mcp-tablet-gateway는 MCP spec(JSON-RPC 2.0 over stdio/SSE) 준수.
도구 정의 시 inputSchema JSON Schema 정확히 명시 → Claude가 정확히 호출.

9.7 코드 품질

Python: ruff + mypy --strict
TypeScript: eslint + tsc --strict
모든 PR에 CI에서 lint·type·test 통과 필수.

10. 마이그레이션 — 기존 시스템에서 이 구조로

기존 코드가 없으므로 N/A. 처음부터 위 구조로 시작 권장.

신규 합류자에게 보여줄 입문 경로:

SYSTEM_DESIGN.md — 전체 그림
OSS_INTEGRATION.md — 어떻게 묶이는가
이 문서 (SERVICES.md) — 어디부터 코딩하는가
MQTT_SCHEMA.md — 인터페이스 계약
DEPLOYMENT.md — 어떻게 굴리는가

설계 문서

논의 자료

팀

전체 구성도

Uh oh!

SERVICES

서비스별 개발 가이드

분류 — 직접 개발 범위

1. mosquitto — MQTT broker

사용 OSS

개발 범위

설정 (mosquitto/config/mosquitto.conf)

ACL 예시 (acl)

운영 포인트

2. frigate — CCTV NVR + 객체 탐지

사용 OSS

개발 범위

설정 (frigate/config.yml)

발행 토픽 (Frigate 기본)

우리 도메인 토픽으로의 변환은 bridge가 담당.

운영 포인트

3. home-assistant — 상태 + 운영 대시보드

사용 OSS

개발 범위

디렉터리 (ha-config/)

configuration.yaml 핵심

폴백 시나리오 카탈로그 (LLM 단절 시 rule-based)

카탈로그

자동화 코드 패턴

검증 절차 (D-7 체크)

충돌 정책

Lovelace 운영자 대시보드 (요지)

4. bridge — Frigate ↔ 도메인 변환 + ReID 보강

사용 OSS·라이브러리

개발 범위

핵심 책임

Zone enter/exit 판정 알고리즘

상태 모델

판정 규칙 (hysteresis)

권장 임계값 (튜닝 가능, config/bridge.yaml)

의사코드

멱등성·중복 방지

디렉터리 구조

핵심 모듈 의사코드

테스트 전략

5. orchestrator — LangGraph + Claude

사용 OSS·라이브러리

개발 범위

핵심 책임

디렉터리 구조

Claude tool 정의 (tools.py)

가드레일 (nodes/guardrail.py)

Prompt 캐싱

테스트 전략

6. mcp-robot-adapter — 로봇 SDK 추상화 + 안전 인터록

사용 OSS·라이브러리

개발 범위

핵심 책임

디렉터리 구조

Capability Model 인터페이스

SPOT 어댑터 (스니펫)

안전 인터록 게이트

테스트 전략

7. mcp-tablet-gateway — 태블릿 WSS + MCP 도구

사용 OSS·라이브러리

개발 범위

핵심 책임

디렉터리 구조

WSS 서버 골격

페어링 흐름

시퀀스 다이어그램

단계별 책임

실패 모드

운영자 가이드

테스트 전략

8. tablet-app — 체험자 인터랙션 (옵션)

사용 OSS·라이브러리

개발 범위

화면 흐름

권장 라이브러리 구성

디렉터리 (참고)

대안

9. 공통 개발 가이드

9.1 모노레포 구조 권장

설정 (`mosquitto/config/mosquitto.conf`)

ACL 예시 (`acl`)

설정 (`frigate/config.yml`)

우리 도메인 토픽으로의 변환은 `bridge`가 담당.

디렉터리 (`ha-config/`)

`configuration.yaml` 핵심

권장 임계값 (튜닝 가능, `config/bridge.yaml`)

Claude tool 정의 (`tools.py`)

가드레일 (`nodes/guardrail.py`)