diff --git a/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md b/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md
new file mode 100644
index 0000000..74814be
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-12-p1-pilot-integration.md
@@ -0,0 +1,772 @@
+# P1 파일럿 통합 (Pilot Integration) Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 이미 구현·병합된 P1-A(엣지 감지 라이브러리) + P1-B(클라우드) 사이의 **끊긴 배선을 연결**해, 노트북 웹캠으로 실행되는 **엔드투엔드 파일럿**을 만든다: 실제 프레임 → CPU pose → 감지 → 서명 POST → 클라우드 → 관리자가 **사진 보고 알림 받고 확인**. 역루프(대시보드 구역 → 엣지 적용)와 하트비트도 살린다. (실장비 Hailo·폐쇄망 번들·드릴 게이트는 별도 후속.)
+
+**Architecture:** 순수 배선. 기존 부품(`run_once`, EventEngine, PostClient, OfflineQueue, ZoneIntrusionDetector, 클라우드 ingest/detail/snapshot)은 그대로 두고 (1) 엣지 데몬(main 루프)이 부품을 조립·구동, (2) CloudClient가 config 폴링·하트비트, (3) MoveNet ONNX 어댑터가 PosePort 구현, (4) 클라우드 detail 뷰·Web Audio가 응답루프 완성. 모든 데몬/클라이언트 로직은 주입(fake source/pose/clock/http)으로 TDD; 실 pose는 정적이미지로 게이트 테스트.
+
+**Tech Stack:** Python 3.11+(dev 3.13), pytest, numpy, opencv, requests, **onnxruntime**(CPU) + **MoveNet SinglePose Lightning ONNX**(COCO-17, Apache-2). 클라우드는 기존 바닐라 PHP + JS(Web Audio, 의존성 0). 런타임 ultralytics(AGPL) 금지 유지.
+
+## Global Constraints
+
+- Python **3.11+**, 타입힌트, `edge/` 기존 패턴 준수. PHP 8.5 `declare(strict_types=1)`, 클라우드 스위트 **0 warnings/0 deprecations** 유지.
+- **판정/루프 로직은 주입 타임스탬프·주입 의존성으로 TDD** — 데몬 루프 내부에서 `time.sleep`/`time.time` 직접 호출 금지(주입된 clock/sleeper 사용). 실 pose 추론만 게이트(모델/onnxruntime 없으면 `skip`).
+- **폐쇄망 지속:** 새 의존성(onnxruntime)·모델(.onnx)은 **빌드머신서 받아 번들**할 것 — 후속 패키징 태스크 대상. 런타임 인터넷은 Cafe24 POST + config 폴링뿐.
+- **HMAC 계약 유지:** config 폴링·하트비트도 P1-A `sign(nonce+ts+body)` 동일 서명(클라우드 T14가 이미 검증). PostClient 서명 재사용.
+- **키포인트 계약:** PosePort는 `list[Person]` (COCO-17 `Keypoint`) 반환 — MoveNet 17관절을 그 순서로 매핑.
+- YAGNI: 파일럿에 필요한 배선만. 멀티캠·감도·처리결과 등 P2/P3는 손대지 않음.
+
+## File Structure
+
+```
+edge/
+ cloudclient.py # CloudClient: fetch_config(HMAC GET), send_heartbeat(HMAC POST)
+ zones.py # zone rows(dict) -> Zone objects; thresholds
+ runtime.py # PipelineRuntime: assemble source/pose/tracker/smoother/engine/ring/snapshotter/sink; step(); apply_zones()
+ daemon.py # main loop: step every frame + periodic sync/heartbeat/queue-flush; signals; main()
+ __main__.py # `python -m edge` -> daemon.main()
+ adapters/
+ movenet_pose.py # MoveNetPoseAdapter(PosePort) via onnxruntime + COCO-17 remap
+ tools/
+ fetch_movenet.py # build-machine: download MoveNet Lightning ONNX to models/
+cloud/
+ src/View/event_detail.php # (replace stub) HTML detail shell + script
+ public/assets/detail.js # fetch GET /events/{id} -> render meta + 
+ public/assets/app.js # (edit) link rows to /events/{id}/view, desk Web Audio on new critical
+ public/index.php # (edit) add GET /events/{id}/view HTML route
+tests/ (edge) + cloud/tests/
+docs/pilot-runbook.md
+```
+
+---
+
+## Task 1: CloudClient — config 폴링 + 하트비트 (HMAC 서명)
+
+**Files:** Create `edge/cloudclient.py`, `tests/test_cloudclient.py`
+
+**Interfaces:**
+- Consumes: `sign` (edge/alert.py), injected `http_get`/`http_post`, `nonce_fn`, `now_fn`.
+- Produces:
+ - `CloudClient(base_url, device_id, secret, nonce_fn, now_fn, http_get=None, http_post=None, timeout=5.0)`
+ - `fetch_config() -> dict | None` — GET `{base}/api/config?device_id=`, headers X-Nonce/X-Timestamp/X-Signature (sign over empty body), returns parsed `{version, zones, thresholds}` on 200 else None.
+ - `send_heartbeat(metrics: dict) -> bool` — POST `{base}/api/heartbeat` body `{device_id, **metrics}` signed, True on 2xx.
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_cloudclient.py`
+
+```python
+import json
+from edge.cloudclient import CloudClient
+from edge.alert import sign
+
+class _Resp:
+ def __init__(self, code, body=""): self.status_code = code; self.text = body
+ def json(self): return json.loads(self.text)
+
+def test_fetch_config_signs_and_parses():
+ captured = {}
+ def fake_get(url, headers=None, timeout=None):
+ captured["url"] = url; captured["headers"] = headers
+ return _Resp(200, json.dumps({"version": 3, "zones": [], "thresholds": {}}))
+ c = CloudClient("https://x", "pi-1", "sek", nonce_fn=lambda: "n1", now_fn=lambda: 1000.0, http_get=fake_get)
+ cfg = c.fetch_config()
+ assert cfg["version"] == 3
+ assert "device_id=pi-1" in captured["url"]
+ assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", "")
+
+def test_fetch_config_none_on_error():
+ c = CloudClient("https://x", "pi-1", "sek", lambda: "n", lambda: 1.0, http_get=lambda *a, **k: _Resp(401))
+ assert c.fetch_config() is None
+
+def test_send_heartbeat_signs_body():
+ captured = {}
+ def fake_post(url, data=None, headers=None, timeout=None):
+ captured["data"] = data; captured["headers"] = headers; return _Resp(200)
+ c = CloudClient("https://x", "pi-1", "sek", lambda: "n1", lambda: 1000.0, http_post=fake_post)
+ assert c.send_heartbeat({"fps": 12.0}) is True
+ body = captured["data"]
+ assert json.loads(body)["device_id"] == "pi-1"
+ assert captured["headers"]["X-Signature"] == sign("sek", "n1", "1000.0", body)
+```
+
+- [ ] **Step 2: 실패 확인** — `pytest tests/test_cloudclient.py -v` → FAIL (ModuleNotFoundError)
+
+- [ ] **Step 3: 구현** — `edge/cloudclient.py`
+
+```python
+import json
+
+from edge.alert import sign
+
+
+class CloudClient:
+ def __init__(self, base_url: str, device_id: str, secret: str,
+ nonce_fn, now_fn, http_get=None, http_post=None, timeout: float = 5.0):
+ self.base_url = base_url.rstrip("/")
+ self.device_id = device_id
+ self.secret = secret
+ self.nonce_fn = nonce_fn
+ self.now_fn = now_fn
+ self.timeout = timeout
+ if http_get is None or http_post is None:
+ import requests
+ http_get = http_get or requests.get
+ http_post = http_post or requests.post
+ self._get = http_get
+ self._post = http_post
+
+ def _headers(self, body: str) -> dict:
+ nonce = self.nonce_fn()
+ ts = str(self.now_fn())
+ return {"X-Nonce": nonce, "X-Timestamp": ts,
+ "X-Signature": sign(self.secret, nonce, ts, body)}
+
+ def fetch_config(self) -> dict | None:
+ url = f"{self.base_url}/api/config?device_id={self.device_id}"
+ try:
+ resp = self._get(url, headers=self._headers(""), timeout=self.timeout)
+ except Exception:
+ return None
+ if 200 <= resp.status_code < 300:
+ try:
+ return resp.json()
+ except Exception:
+ return None
+ return None
+
+ def send_heartbeat(self, metrics: dict) -> bool:
+ payload = {"device_id": self.device_id, **metrics}
+ body = json.dumps(payload, sort_keys=True)
+ headers = {"Content-Type": "application/json", **self._headers(body)}
+ try:
+ resp = self._post(f"{self.base_url}/api/heartbeat", data=body,
+ headers=headers, timeout=self.timeout)
+ except Exception:
+ return False
+ return 200 <= resp.status_code < 300
+```
+
+- [ ] **Step 4: 통과 + 커밋** — `pytest tests/test_cloudclient.py -v` → PASS
+
+```bash
+git add edge/cloudclient.py tests/test_cloudclient.py
+git commit -m "feat(edge): CloudClient for config poll + heartbeat (HMAC-signed)"
+```
+
+---
+
+## Task 2: 클라우드 zone 행 → Zone 객체 로딩
+
+**Files:** Create `edge/zones.py`, `tests/test_zones.py`
+
+**Interfaces:**
+- Consumes: `Zone` (edge/detectors/zone.py)
+- Produces:
+ - `zones_from_config(config: dict) -> list[Zone]` — `config["zones"]` 각 행(`{zone_id, camera_id, level, polygon_json}`)을 `Zone(zone_id, level, polygon)`으로. `polygon_json`은 JSON 문자열(또는 이미 list) → `[(x,y),...]`. 파싱 실패/3점미만 행은 건너뜀.
+ - `thresholds_from_config(config: dict) -> dict` — `config["thresholds"]` 그대로 반환({}).
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_zones.py`
+
+```python
+from edge.zones import zones_from_config, thresholds_from_config
+
+def test_parses_zone_rows_json_string_polygon():
+ cfg = {"zones": [
+ {"zone_id": "z1", "camera_id": "cam-1", "level": "restricted",
+ "polygon_json": "[[0,0],[9,0],[9,9]]"},
+ ], "thresholds": {"fall_prone_s": 3.0}}
+ zones = zones_from_config(cfg)
+ assert len(zones) == 1
+ assert zones[0].zone_id == "z1"
+ assert zones[0].level == "restricted"
+ assert zones[0].polygon == [(0.0, 0.0), (9.0, 0.0), (9.0, 9.0)]
+ assert thresholds_from_config(cfg)["fall_prone_s"] == 3.0
+
+def test_skips_malformed_row():
+ cfg = {"zones": [{"zone_id": "bad"}], "thresholds": {}}
+ assert zones_from_config(cfg) == []
+
+def test_empty_config():
+ assert zones_from_config({}) == []
+ assert thresholds_from_config({}) == {}
+```
+
+- [ ] **Step 2: 실패 확인** → FAIL
+
+- [ ] **Step 3: 구현** — `edge/zones.py`
+
+```python
+import json
+
+from edge.detectors.zone import Zone
+
+
+def zones_from_config(config: dict) -> list[Zone]:
+ out: list[Zone] = []
+ for row in config.get("zones", []) or []:
+ try:
+ raw = row["polygon_json"]
+ pts = json.loads(raw) if isinstance(raw, str) else raw
+ polygon = [(float(p[0]), float(p[1])) for p in pts]
+ if len(polygon) < 3:
+ continue
+ out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon))
+ except (KeyError, ValueError, TypeError, json.JSONDecodeError):
+ continue
+ return out
+
+
+def thresholds_from_config(config: dict) -> dict:
+ t = config.get("thresholds")
+ return t if isinstance(t, dict) else {}
+```
+
+- [ ] **Step 4: 통과 + 커밋**
+
+```bash
+git add edge/zones.py tests/test_zones.py
+git commit -m "feat(edge): parse cloud config zones/thresholds into Zone objects"
+```
+
+---
+
+## Task 3: PipelineRuntime — 부품 조립 + zone 핫스왑
+
+**Files:** Create `edge/runtime.py`, `tests/test_runtime.py`
+
+**Interfaces:**
+- Consumes: `run_once` (edge/app.py), `IouTracker`, `TrackSmootherPool`, `EventEngine`+detectors, `RingBuffer`/`Snapshotter`/`FaceBlurrer`, `Config` (edge/config.py), `Zone`.
+- Produces:
+ - `PipelineRuntime(config, source, pose, sink)` — 내부에 tracker/pool/engine/ring/snapshotter 구성(ZoneIntrusionDetector는 `zone_detector` 속성으로 보관).
+ - `step(now: float) -> list` — 한 프레임 처리(`run_once` 호출), 이벤트 반환.
+ - `apply_zones(zones: list[Zone]) -> None` — 러닝 중 ZoneIntrusionDetector의 zone 목록 교체(역루프).
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_runtime.py`
+
+```python
+import numpy as np
+from edge.config import load_config
+from edge.models import Person, BBox, Keypoint
+from edge.detectors.zone import Zone
+from edge.runtime import PipelineRuntime
+
+def _lying_person():
+ kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
+ kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
+ kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
+ kps[0] = Keypoint(90, 320, 0.9)
+ return Person(BBox(90, 290, 210, 320), kps)
+
+class _Source:
+ def read(self): return True, np.full((360, 640, 3), 128, np.uint8)
+class _Pose:
+ def infer(self, frame): return [_lying_person()]
+class _Sink:
+ def __init__(self): self.events = []
+ def try_send(self, e, snaps): self.events.append(e); return True
+
+def test_step_runs_pipeline_and_emits_fall():
+ cfg = load_config({"FALL_PRONE_S": "0.5"})
+ sink = _Sink()
+ rt = PipelineRuntime(cfg, _Source(), _Pose(), sink)
+ rt.step(now=0.0)
+ rt.step(now=0.6)
+ from edge.events import EventType
+ assert any(e.event_type == EventType.FALL for e in sink.events)
+
+def test_apply_zones_hotswaps():
+ cfg = load_config({})
+ rt = PipelineRuntime(cfg, _Source(), _Pose(), _Sink())
+ rt.apply_zones([Zone("z1", "restricted", [(0, 0), (9, 0), (9, 9)])])
+ assert rt.zone_detector.zones[0].zone_id == "z1"
+```
+
+- [ ] **Step 2: 실패 확인** → FAIL
+
+- [ ] **Step 3: 구현** — `edge/runtime.py`
+
+```python
+from edge.app import run_once
+from edge.tracking import IouTracker
+from edge.smoothing import TrackSmootherPool
+from edge.detectors.fall import FallDetector
+from edge.detectors.immobility import ImmobilityWatchdog
+from edge.detectors.ghost import GhostIncidentDetector
+from edge.detectors.zone import ZoneIntrusionDetector
+from edge.engine import EventEngine
+from edge.snapshot import RingBuffer, FaceBlurrer, Snapshotter
+
+
+class PipelineRuntime:
+ def __init__(self, config, source, pose, sink):
+ self.config = config
+ self.source = source
+ self.pose = pose
+ self.sink = sink
+ self.tracker = IouTracker()
+ self.pool = TrackSmootherPool()
+ self.zone_detector = ZoneIntrusionDetector([], config.zone)
+ self.engine = EventEngine(
+ FallDetector(config.fall),
+ ImmobilityWatchdog(config.immobility),
+ GhostIncidentDetector(config.ghost),
+ self.zone_detector,
+ config.engine,
+ )
+ self.ring = RingBuffer()
+ self.snapshotter = Snapshotter(FaceBlurrer())
+
+ def step(self, now: float) -> list:
+ return run_once(self.source, self.pose, self.tracker, self.pool,
+ self.engine, self.ring, self.snapshotter, self.sink,
+ now_fn=lambda: now)
+
+ def apply_zones(self, zones: list) -> None:
+ self.zone_detector.zones = list(zones)
+```
+
+> **NOTE (implementer):** `ZoneIntrusionDetector.zones` is a public list (see edge/detectors/zone.py) — reassigning it hot-swaps the active zones without rebuilding the engine. Confirm the attribute name is `zones`. Also confirm `run_once`'s parameter order is `(source, pose, tracker, pool, engine, ring, snapshotter, sink, now_fn)` (P1-A final form after the per-track-smoother fix).
+
+- [ ] **Step 4: 통과 + 커밋**
+
+```bash
+git add edge/runtime.py tests/test_runtime.py
+git commit -m "feat(edge): PipelineRuntime assembles components + hot-swaps zones"
+```
+
+---
+
+## Task 4: 데몬 — main 루프 + 주기 동기/하트비트/큐flush
+
+**Files:** Create `edge/daemon.py`, `edge/__main__.py`, `tests/test_daemon.py`; Modify `pyproject.toml`
+
+**Interfaces:**
+- Consumes: `PipelineRuntime` (T3), `CloudClient` (T1), `zones_from_config` (T2), `OfflineQueue` (edge/queue.py), `PostClient` (edge/alert.py).
+- Produces:
+ - `run_loop(runtime, client, queue, *, clock, sleep, iterations=None, sync_every=150, heartbeat_every=900, flush_every=300, frame_interval=0.0) -> dict` — `iterations` 프레임 루프(None이면 무한, 테스트는 유한). 매 프레임 `runtime.step(clock())`(예외 삼켜 카운트); sync_every 프레임마다 `client.fetch_config()`→`runtime.apply_zones(zones_from_config(...))`; heartbeat_every마다 `send_heartbeat`; flush_every마다 `queue.flush()`. 반환 카운터 dict.
+ - `main()` — env로 구성 후 무한 루프. `edge/__main__.py`가 호출. (`# pragma: no cover` — 실장비 엔트리)
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_daemon.py`
+
+```python
+from edge.daemon import run_loop
+
+class _Runtime:
+ def __init__(self): self.steps = 0; self.applied = None
+ def step(self, now): self.steps += 1; return []
+ def apply_zones(self, zones): self.applied = zones
+class _Client:
+ def __init__(self): self.configs = 0; self.beats = 0
+ def fetch_config(self): self.configs += 1; return {"zones": [], "thresholds": {}}
+ def send_heartbeat(self, m): self.beats += 1; return True
+class _Queue:
+ def __init__(self): self.flushes = 0
+ def flush(self): self.flushes += 1; return 0
+
+def test_loop_steps_and_periodics():
+ rt, cl, q = _Runtime(), _Client(), _Queue()
+ ticks = iter(range(1000))
+ counters = run_loop(rt, cl, q, clock=lambda: float(next(ticks)),
+ sleep=lambda s: None, iterations=10,
+ sync_every=5, heartbeat_every=5, flush_every=5)
+ assert rt.steps == 10
+ assert cl.configs == 2 and cl.beats == 2 and q.flushes == 2 # at frames 5 and 10
+ assert rt.applied == [] # zones applied from fetched config
+
+def test_loop_survives_step_exception():
+ class Boom(_Runtime):
+ def step(self, now):
+ self.steps += 1
+ if self.steps == 2: raise RuntimeError("frame glitch")
+ return []
+ rt = Boom()
+ counters = run_loop(rt, _Client(), _Queue(), clock=lambda: 0.0,
+ sleep=lambda s: None, iterations=3,
+ sync_every=99, heartbeat_every=99, flush_every=99)
+ assert rt.steps == 3 # kept going past the exception
+ assert counters["errors"] == 1
+```
+
+- [ ] **Step 2: 실패 확인** → FAIL
+
+- [ ] **Step 3: 구현** — `edge/daemon.py`
+
+```python
+import os
+
+
+def run_loop(runtime, client, queue, *, clock, sleep, iterations=None,
+ sync_every=150, heartbeat_every=900, flush_every=300,
+ frame_interval=0.0):
+ counters = {"frames": 0, "syncs": 0, "beats": 0, "flushes": 0, "errors": 0}
+ i = 0
+ while iterations is None or i < iterations:
+ i += 1
+ try:
+ runtime.step(clock())
+ counters["frames"] += 1
+ except Exception:
+ counters["errors"] += 1
+ if sync_every and i % sync_every == 0:
+ cfg = client.fetch_config()
+ if cfg is not None:
+ from edge.zones import zones_from_config
+ runtime.apply_zones(zones_from_config(cfg))
+ counters["syncs"] += 1
+ if heartbeat_every and i % heartbeat_every == 0:
+ if client.send_heartbeat({"fps": 0.0, "frames": counters["frames"]}):
+ counters["beats"] += 1
+ if flush_every and i % flush_every == 0:
+ queue.flush()
+ counters["flushes"] += 1
+ if frame_interval:
+ sleep(frame_interval)
+ return counters
+
+
+def main() -> None: # pragma: no cover (integration entrypoint, exercised on device)
+ import time
+ from edge.config import load_config
+ from edge.cloudclient import CloudClient
+ from edge.alert import PostClient
+ from edge.queue import OfflineQueue
+ from edge.runtime import PipelineRuntime
+ from edge.adapters.uvc_source import UvcSource
+ from edge.adapters.movenet_pose import MoveNetPoseAdapter
+
+ cfg = load_config(dict(os.environ))
+ nonce = lambda: os.urandom(8).hex()
+ now = time.time
+ base = cfg.cafe24_url.rstrip("/")
+ client = CloudClient(base, cfg.device_id, cfg.secret, nonce, now)
+ post = PostClient(base + "/api/events", cfg.device_id, cfg.secret, nonce, now)
+ queue = OfflineQueue(os.environ.get("QUEUE_DB", "queue.db"), post)
+ source = UvcSource(int(os.environ.get("UVC_DEVICE", "0")))
+ pose = MoveNetPoseAdapter(os.environ["MOVENET_ONNX"])
+ runtime = PipelineRuntime(cfg, source, pose, queue)
+ run_loop(runtime, client, queue, clock=now, sleep=time.sleep, iterations=None,
+ frame_interval=1.0 / float(os.environ.get("TARGET_FPS", "12")))
+```
+
+`edge/__main__.py`:
+
+```python
+from edge.daemon import main
+
+if __name__ == "__main__":
+ main()
+```
+
+- [ ] **Step 4: pyproject.toml에 스크립트 엔트리 추가**
+
+```toml
+[project.scripts]
+move-detect-edge = "edge.daemon:main"
+```
+
+- [ ] **Step 5: 통과 + 커밋** — `pytest tests/test_daemon.py -v` → PASS; 전체 `pytest -q`
+
+```bash
+git add edge/daemon.py edge/__main__.py tests/test_daemon.py pyproject.toml
+git commit -m "feat(edge): runnable daemon loop (step + periodic sync/heartbeat/flush)"
+```
+
+---
+
+## Task 5: MoveNet CPU pose 어댑터 (실 pose 경로)
+
+**Files:** Create `edge/adapters/movenet_pose.py`, `edge/tools/fetch_movenet.py`, `tests/test_movenet_pose.py`
+
+**Interfaces:**
+- Consumes: `Person, BBox, Keypoint` (edge/models.py), PosePort 계약.
+- Produces: `MoveNetPoseAdapter(onnx_path, input_size=192, session=None, min_person_score=0.2)` implements PosePort `infer(frame) -> list[Person]`. MoveNet SinglePose Lightning: 출력 `[1,1,17,3]` = (y,x,score) 정규화(0..1). 프레임 크기로 역정규화 → COCO-17 순서 Keypoint(x,y,conf). bbox = 가시 관절(conf≥min_person_score) min/max; 가시 관절 0이면 빈 리스트.
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_movenet_pose.py` (session 주입 — onnxruntime 불필요)
+
+```python
+import numpy as np
+from edge.adapters.movenet_pose import MoveNetPoseAdapter
+
+class _FakeSession:
+ """MoveNet 출력 형태 [1,1,17,3] (y,x,score) 정규화."""
+ def get_inputs(self):
+ class I: name = "input"
+ return [I()]
+ def run(self, _out, feed):
+ kp = np.zeros((1, 1, 17, 3), dtype=np.float32)
+ for j in range(17):
+ kp[0, 0, j] = [0.5, 0.25, 0.9] # y=0.5, x=0.25, score=0.9
+ return [kp]
+
+def test_infer_maps_movenet_to_coco17_persons():
+ frame = np.zeros((400, 800, 3), dtype=np.uint8)
+ a = MoveNetPoseAdapter("unused.onnx", session=_FakeSession())
+ persons = a.infer(frame)
+ assert len(persons) == 1
+ p = persons[0]
+ assert len(p.keypoints) == 17
+ # x = 0.25 * 800 = 200, y = 0.5 * 400 = 200
+ assert abs(p.keypoints[0].x - 200) < 1 and abs(p.keypoints[0].y - 200) < 1
+ assert p.keypoints[0].conf == 0.9
+ assert p.bbox.x1 <= 200 <= p.bbox.x2
+
+def test_low_score_person_dropped():
+ class Low(_FakeSession):
+ def run(self, _o, feed):
+ return [np.zeros((1, 1, 17, 3), dtype=np.float32)] # all score 0
+ a = MoveNetPoseAdapter("unused.onnx", session=Low())
+ assert a.infer(np.zeros((10, 10, 3), np.uint8)) == []
+```
+
+- [ ] **Step 2: 실패 확인** → FAIL
+
+- [ ] **Step 3: 구현** — `edge/adapters/movenet_pose.py`
+
+```python
+import cv2
+import numpy as np
+
+from edge.models import Person, BBox, Keypoint
+
+
+class MoveNetPoseAdapter:
+ """MoveNet SinglePose Lightning (COCO-17) via onnxruntime. CPU dev/pilot PosePort.
+ 출력 [1,1,17,3] = (y, x, score) normalized 0..1."""
+
+ def __init__(self, onnx_path: str, input_size: int = 192, session=None,
+ min_person_score: float = 0.2):
+ self.input_size = input_size
+ self.min_person_score = min_person_score
+ if session is None:
+ import onnxruntime # gated: bundled on build machine
+ session = onnxruntime.InferenceSession(onnx_path, providers=["CPUExecutionProvider"])
+ self.session = session
+ self._input_name = session.get_inputs()[0].name
+
+ def infer(self, frame) -> list:
+ h, w = frame.shape[:2]
+ img = cv2.resize(frame, (self.input_size, self.input_size))
+ img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.int32)[None, ...]
+ out = self.session.run(None, {self._input_name: img})[0] # [1,1,17,3]
+ kp = out[0, 0] # [17,3] (y,x,score)
+ keypoints = [Keypoint(float(kp[j, 1] * w), float(kp[j, 0] * h), float(kp[j, 2]))
+ for j in range(17)]
+ vis = [k for k in keypoints if k.conf >= self.min_person_score]
+ if not vis:
+ return []
+ xs = [k.x for k in vis]; ys = [k.y for k in vis]
+ bbox = BBox(min(xs), min(ys), max(xs), max(ys))
+ return [Person(bbox, keypoints)]
+```
+
+`edge/tools/fetch_movenet.py`:
+
+```python
+"""빌드머신 전용: MoveNet SinglePose Lightning ONNX 다운로드(폐쇄망 번들). Apache-2.0."""
+import os
+import urllib.request
+
+# MoveNet Lightning ONNX (예시 소스; 조직 미러로 교체 가능)
+URL = "https://github.com/PINTO0309/PINTO_model_zoo/raw/main/115_MoveNet/resources/movenet_singlepose_lightning_4.onnx"
+DEST = os.path.join(os.path.dirname(__file__), "..", "..", "models", "movenet_lightning.onnx")
+
+def main():
+ os.makedirs(os.path.dirname(DEST), exist_ok=True)
+ print(f"downloading MoveNet ONNX -> {DEST}")
+ urllib.request.urlretrieve(URL, DEST)
+ print("done")
+
+if __name__ == "__main__":
+ main()
+```
+
+> **NOTE (implementer):** the `fetch_movenet.py` URL is a build-machine convenience; verify the model's input size/dtype and output layout `[1,1,17,3]` against the actual downloaded file before relying on it on hardware — the fake-session unit test locks the ADAPTER's remap logic, not the specific model file. If the real model expects a different input size/dtype, adjust `input_size`/dtype accordingly; the COCO-17 remap stays.
+
+- [ ] **Step 4: 통과 + 커밋** — `pytest tests/test_movenet_pose.py -v` → PASS
+
+```bash
+git add edge/adapters/movenet_pose.py edge/tools/fetch_movenet.py tests/test_movenet_pose.py
+git commit -m "feat(edge): MoveNet ONNX CPU pose adapter (COCO-17 remap)"
+```
+
+---
+
+## Task 6: 클라우드 사고 상세 뷰 — 스냅샷 표시
+
+**Files:** Modify `cloud/src/View/event_detail.php`; Create `cloud/public/assets/detail.js`; Modify `cloud/public/index.php`; Modify `cloud/tests/WiringSmokeTest.php`
+
+**Interfaces:**
+- Consumes: existing `GET /events/{id}` (JSON: event+snapshots) + `GET /snap/{uuid}` (auth image).
+- Produces: `GET /events/{id}/view` → renders `event_detail.php` (HTML shell + `
+```
+
+- [ ] **Step 3: detail.js** — `cloud/public/assets/detail.js`
+
+```javascript
+(function () {
+ var m = location.pathname.match(/\/events\/(\d+)\/view/);
+ if (!m) return;
+ var id = m[1];
+ fetch('/events/' + id, { credentials: 'same-origin' })
+ .then(function (r) { return r.ok ? r.json() : Promise.reject(r.status); })
+ .then(function (d) {
+ var ev = d.event || {};
+ var meta = document.getElementById('detail-meta');
+ meta.textContent = '';
+ [['유형', ev.event_type], ['심각도', ev.severity], ['카메라', ev.camera_id],
+ ['시각', ev.occurred_at], ['지속(초)', ev.duration_s], ['확인', ev.acknowledged_by || '미확인']]
+ .forEach(function (row) {
+ var p = document.createElement('div');
+ p.textContent = row[0] + ': ' + (row[1] == null ? '-' : row[1]);
+ meta.appendChild(p);
+ });
+ var box = document.getElementById('detail-snapshots');
+ (d.snapshots || []).forEach(function (s) {
+ var img = document.createElement('img');
+ img.src = '/snap/' + encodeURIComponent(s.file_uuid);
+ img.alt = '사고 스냅샷';
+ img.style.maxWidth = '100%';
+ box.appendChild(img);
+ });
+ })
+ .catch(function () {
+ document.getElementById('detail-meta').textContent = '불러오기 실패';
+ });
+})();
+```
+
+- [ ] **Step 4: index.php 라우트** — 뷰 라우트 섹션에 추가:
+
+```php
+$r->add('GET', '/events/{id}/view', fn() => $render('event_detail'));
+```
+
+- [ ] **Step 5: 통과 + 커밋** — 전체 `cd cloud && ./vendor/bin/phpunit --display-warnings --display-deprecations` green; `php -l cloud/public/index.php`.
+
+```bash
+git add cloud/src/View/event_detail.php cloud/public/assets/detail.js cloud/public/index.php cloud/tests/WiringSmokeTest.php
+git commit -m "feat(cloud): incident detail view renders snapshot images"
+```
+
+---
+
+## Task 7: 응답 루프 — 목록 링크 + 데스크 Web Audio
+
+**Files:** Modify `cloud/public/assets/app.js`
+
+**Interfaces:**
+- Consumes: `GET /api/events` (기존 폴링).
+- Produces: 각 목록 행이 `/events/{id}/view`로 링크; **새 미확인 critical 이벤트 감지 시 데스크 Web Audio 반복 비프**(미확인 critical 존재 동안), 첫 클릭으로 audio 언락. 색맹안전 배지 유지. 기존 buildRow/폴링/ack 동작 보존.
+
+- [ ] **Step 1: app.js 편집** — 기존 구조에 아래를 통합(구현자는 실제 함수명에 맞춤):
+
+```javascript
+// --- desk audio alert (no binary asset; WebAudio beep) ---
+var audioCtx = null;
+document.addEventListener('click', function unlock() {
+ if (!audioCtx) audioCtx = new (window.AudioContext || window.webkitAudioContext)();
+ document.removeEventListener('click', unlock);
+});
+var alertedCritical = {}; // event_id -> true (already beeped once)
+var alarmTimer = null;
+function beep() {
+ if (!audioCtx) return;
+ var o = audioCtx.createOscillator(), g = audioCtx.createGain();
+ o.frequency.value = 880; o.connect(g); g.connect(audioCtx.destination);
+ g.gain.value = 0.2; o.start(); o.stop(audioCtx.currentTime + 0.4);
+}
+function updateAlarm(events) {
+ var unackedCritical = false;
+ events.forEach(function (e) {
+ if (e.severity === 'critical' && !e.acknowledged_by) {
+ unackedCritical = true;
+ if (!alertedCritical[e.event_id]) { alertedCritical[e.event_id] = true; beep(); }
+ }
+ });
+ if (unackedCritical && !alarmTimer) alarmTimer = setInterval(beep, 3000);
+ if (!unackedCritical && alarmTimer) { clearInterval(alarmTimer); alarmTimer = null; }
+}
+```
+
+그리고: (a) 행 빌더에서 각 행을 `/events/{event_id}/view` 링크로 감싸기 — `var a = document.createElement('a'); a.href = '/events/' + e.event_id + '/view'; a.appendChild(rowEl);` 후 컨테이너에 append. (b) 폴링 성공 콜백에서 `renderList(events)` 뒤에 `updateAlarm(events)` 호출.
+
+- [ ] **Step 2: 통과 + 커밋** — `node --check cloud/public/assets/app.js`(node 있으면), 전체 phpunit green(JS만 변경).
+
+```bash
+git add cloud/public/assets/app.js
+git commit -m "feat(cloud): dashboard rows link to detail + desk Web Audio on critical"
+```
+
+---
+
+## Task 8: 파일럿 런북 + 게이트 pose 스모크
+
+**Files:** Create `docs/pilot-runbook.md`, `tests/hardware/test_movenet_smoke.py`
+
+- [ ] **Step 1: 게이트 스모크** — `tests/hardware/test_movenet_smoke.py`
+
+```python
+import os
+import numpy as np
+import pytest
+
+MODEL = os.path.join(os.path.dirname(__file__), "..", "..", "models", "movenet_lightning.onnx")
+
+@pytest.mark.skipif(not os.path.exists(MODEL), reason="MoveNet model not bundled")
+def test_real_movenet_runs():
+ pytest.importorskip("onnxruntime")
+ from edge.adapters.movenet_pose import MoveNetPoseAdapter
+ a = MoveNetPoseAdapter(MODEL)
+ persons = a.infer(np.full((480, 640, 3), 128, np.uint8))
+ assert isinstance(persons, list) # runs end-to-end; each Person has 17 keypoints
+```
+
+- [ ] **Step 2: 파일럿 런북** — `docs/pilot-runbook.md`, 노트북 파일럿 절차:
+ 1. **빌드:** `pip install onnxruntime`; `python -m edge.tools.fetch_movenet` (모델 → `models/`).
+ 2. **클라우드:** `php -S localhost:8000 -t cloud/public` (config.php에 로컬 DB) 또는 Cafe24. admin 계정 생성 + device secret을 `devices` 테이블에 등록.
+ 3. **env:** `DEVICE_ID`, `DEVICE_SECRET`(클라우드와 일치), `CAFE24_URL`(예 http://localhost:8000), `MOVENET_ONNX=models/movenet_lightning.onnx`, `UVC_DEVICE=0`, `TARGET_FPS=12`.
+ 4. **실행:** `move-detect-edge` (또는 `python -m edge`).
+ 5. **데모:** 웹캠 앞 눕기 → 3초 후 fall, 10초 후 no_motion(위급, 데스크 비프) → 목록행 클릭 → 상세서 스냅샷 확인 → 확인처리(비프 멈춤). 대시보드서 구역 그리기 → 발끝 진입 → 침입 이벤트(역루프). 네트워크 끊고 이벤트 → 복구 후 큐 flush.
+ 6. **한계(명시):** 단일캠·CPU pose(저fps)·SMS 미연동(Web Audio만)·前中後 아닌 단일프레임. 실장비 Hailo·폐쇄망 번들·드릴 게이트·F14 컴플라이언스는 후속.
+
+- [ ] **Step 3: 통과 + 커밋** — `pytest -q`(hardware skip) green.
+
+```bash
+git add docs/pilot-runbook.md tests/hardware/test_movenet_smoke.py
+git commit -m "docs: pilot runbook + gated MoveNet smoke test"
+```
+
+---
+
+## Self-Review
+
+**1. 최소경로 커버리지 (기능검토 1~7):**
+- 엣지 데몬(1) → T4 ✓ · CPU pose(2) → T5 ✓ · edge→클라우드 흐름(3) → T4 데몬+기존 PostClient ✓
+- 사고 가시화(4) → T6 ✓ · 응답 1채널(5) → T7 Web Audio ✓
+- config-sync 역루프(6) → T1+T2+T3(apply_zones)+T4(주기 sync) ✓ · 하트비트(7) → T1+T4 ✓ · 큐 flush(rank7) → T4 ✓
+- **범위 밖(명시):** 실 Hailo(하드웨어), 폐쇄망 전체 번들(T5/T8은 fetch+게이트만; 번들·차단설치는 후속 패키징), 드릴 게이트(P1-C), F14 컴플라이언스(별도), 前中後 링버퍼(rank11 — 파일럿 단일프레임 유지).
+- **의도적 미포함:** critical SMS/webhook 실연동(leak_sms 게이트 정보 필요) — T7 Web Audio로 "사람 알림" 증명, 실연동 후속.
+
+**2. Placeholder scan:** 데몬/클라이언트/런타임/어댑터/zone은 실코드+TDD. T5 실모델 스모크·T8은 게이트(모델 없으면 skip) — 의도적. fetch URL은 빌드 편의(구현자 노트로 검증 지시).
+
+**3. Type consistency:** `CloudClient.sign` = PostClient/클라우드 T14 동일(`nonce+ts+body`). `PipelineRuntime.step`가 `run_once(source,pose,tracker,pool,engine,ring,snapshotter,sink,now_fn)` — P1-A 최종형(per-track smoother 수정 후) 일치. `apply_zones`가 `ZoneIntrusionDetector.zones` 재할당. MoveNet [1,1,17,3](y,x,score)→COCO-17 Keypoint(x,y,conf) 일관.
+
+---
+
+## Execution Handoff
+
+**Plan complete. 저장: `docs/superpowers/plans/2026-07-12-p1-pilot-integration.md`.**
+
+실행: **Subagent-Driven (추천)** 또는 Inline. 어느 방식?